reddb_wire/redwire/
typed_insert.rs

1//! RedWire typed single-row insert payload.
2//!
3//! V1 layout:
4//! `[version:u8][collection_len:u16][collection][field_count:u16]`
5//! followed by `field_count` entries of
6//! `[name_len:u16][name][query_with_params::ParamValue]`.
7
8use std::collections::BTreeSet;
9use std::fmt;
10
11use crate::query_with_params::{decode_value, encode_value, ParamCodecError, ParamValue};
12
13pub const FEATURE_TYPED_MUTATIONS: u32 = 0x0000_0004;
14pub const TYPED_INSERT_VERSION: u8 = 1;
15
16#[derive(Debug, Clone, PartialEq)]
17pub struct TypedInsertField {
18    pub name: String,
19    pub value: ParamValue,
20}
21
22#[derive(Debug, Clone, PartialEq)]
23pub struct TypedInsertRequest {
24    pub collection: String,
25    pub fields: Vec<TypedInsertField>,
26}
27
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub enum TypedInsertError {
30    UnsupportedVersion(u8),
31    LengthOverflow(&'static str),
32    Truncated(&'static str),
33    InvalidUtf8(&'static str),
34    EmptyCollection,
35    EmptyFieldName,
36    DuplicateField(String),
37    Value(ParamCodecError),
38    TrailingBytes(usize),
39}
40
41impl fmt::Display for TypedInsertError {
42    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
43        match self {
44            Self::UnsupportedVersion(version) => {
45                write!(f, "unsupported typed insert version {version}")
46            }
47            Self::LengthOverflow(field) => write!(f, "{field} is too large for typed insert v1"),
48            Self::Truncated(field) => write!(f, "truncated {field}"),
49            Self::InvalidUtf8(field) => write!(f, "{field} must be valid UTF-8"),
50            Self::EmptyCollection => write!(f, "collection cannot be empty"),
51            Self::EmptyFieldName => write!(f, "field name cannot be empty"),
52            Self::DuplicateField(name) => write!(f, "duplicate field '{name}'"),
53            Self::Value(error) => write!(f, "invalid field value: {error}"),
54            Self::TrailingBytes(count) => write!(f, "{count} trailing bytes after typed insert"),
55        }
56    }
57}
58
59impl std::error::Error for TypedInsertError {}
60
61pub fn encode_typed_insert(request: &TypedInsertRequest) -> Result<Vec<u8>, TypedInsertError> {
62    validate_name(&request.collection, "collection")?;
63    let collection_len = u16::try_from(request.collection.len())
64        .map_err(|_| TypedInsertError::LengthOverflow("collection"))?;
65    let field_count = u16::try_from(request.fields.len())
66        .map_err(|_| TypedInsertError::LengthOverflow("fields"))?;
67
68    let mut names = BTreeSet::new();
69    let mut out = Vec::new();
70    out.push(TYPED_INSERT_VERSION);
71    out.extend_from_slice(&collection_len.to_le_bytes());
72    out.extend_from_slice(request.collection.as_bytes());
73    out.extend_from_slice(&field_count.to_le_bytes());
74    for field in &request.fields {
75        validate_name(&field.name, "field name")?;
76        if !names.insert(field.name.as_str()) {
77            return Err(TypedInsertError::DuplicateField(field.name.clone()));
78        }
79        let name_len = u16::try_from(field.name.len())
80            .map_err(|_| TypedInsertError::LengthOverflow("field name"))?;
81        out.extend_from_slice(&name_len.to_le_bytes());
82        out.extend_from_slice(field.name.as_bytes());
83        encode_value(&field.value, &mut out).map_err(TypedInsertError::Value)?;
84    }
85    Ok(out)
86}
87
88pub fn decode_typed_insert(payload: &[u8]) -> Result<TypedInsertRequest, TypedInsertError> {
89    let mut pos = 0;
90    let version = read_bytes(payload, &mut pos, 1, "version")?[0];
91    if version != TYPED_INSERT_VERSION {
92        return Err(TypedInsertError::UnsupportedVersion(version));
93    }
94
95    let collection_len = read_u16(payload, &mut pos, "collection_len")? as usize;
96    let collection = read_string(payload, &mut pos, collection_len, "collection")?;
97    validate_name(&collection, "collection")?;
98
99    let field_count = read_u16(payload, &mut pos, "field_count")? as usize;
100    let minimum_field_bytes = field_count
101        .checked_mul(3)
102        .ok_or(TypedInsertError::Truncated("fields"))?;
103    if payload.len().saturating_sub(pos) < minimum_field_bytes {
104        return Err(TypedInsertError::Truncated("fields"));
105    }
106
107    let mut names = BTreeSet::new();
108    let mut fields = Vec::with_capacity(field_count);
109    for _ in 0..field_count {
110        let name_len = read_u16(payload, &mut pos, "field_name_len")? as usize;
111        let name = read_string(payload, &mut pos, name_len, "field name")?;
112        validate_name(&name, "field name")?;
113        if !names.insert(name.clone()) {
114            return Err(TypedInsertError::DuplicateField(name));
115        }
116        let value = decode_value(payload, &mut pos).map_err(TypedInsertError::Value)?;
117        fields.push(TypedInsertField { name, value });
118    }
119    if pos != payload.len() {
120        return Err(TypedInsertError::TrailingBytes(payload.len() - pos));
121    }
122    Ok(TypedInsertRequest { collection, fields })
123}
124
125fn validate_name(value: &str, field: &'static str) -> Result<(), TypedInsertError> {
126    if value.trim().is_empty() {
127        return Err(if field == "collection" {
128            TypedInsertError::EmptyCollection
129        } else {
130            TypedInsertError::EmptyFieldName
131        });
132    }
133    Ok(())
134}
135
136fn read_u16(payload: &[u8], pos: &mut usize, field: &'static str) -> Result<u16, TypedInsertError> {
137    let bytes = read_bytes(payload, pos, 2, field)?;
138    Ok(u16::from_le_bytes([bytes[0], bytes[1]]))
139}
140
141fn read_string(
142    payload: &[u8],
143    pos: &mut usize,
144    len: usize,
145    field: &'static str,
146) -> Result<String, TypedInsertError> {
147    let bytes = read_bytes(payload, pos, len, field)?;
148    std::str::from_utf8(bytes)
149        .map(str::to_owned)
150        .map_err(|_| TypedInsertError::InvalidUtf8(field))
151}
152
153fn read_bytes<'a>(
154    payload: &'a [u8],
155    pos: &mut usize,
156    len: usize,
157    field: &'static str,
158) -> Result<&'a [u8], TypedInsertError> {
159    let end = pos
160        .checked_add(len)
161        .ok_or(TypedInsertError::Truncated(field))?;
162    if end > payload.len() {
163        return Err(TypedInsertError::Truncated(field));
164    }
165    let bytes = &payload[*pos..end];
166    *pos = end;
167    Ok(bytes)
168}
169
170#[cfg(test)]
171mod tests {
172    use super::*;
173
174    fn all_values_request() -> TypedInsertRequest {
175        TypedInsertRequest {
176            collection: "events".to_string(),
177            fields: vec![
178                TypedInsertField {
179                    name: "null".into(),
180                    value: ParamValue::Null,
181                },
182                TypedInsertField {
183                    name: "bool".into(),
184                    value: ParamValue::Bool(true),
185                },
186                TypedInsertField {
187                    name: "int".into(),
188                    value: ParamValue::Int(-7),
189                },
190                TypedInsertField {
191                    name: "float".into(),
192                    value: ParamValue::Float(1.5),
193                },
194                TypedInsertField {
195                    name: "text".into(),
196                    value: ParamValue::Text("olá".into()),
197                },
198                TypedInsertField {
199                    name: "bytes".into(),
200                    value: ParamValue::Bytes(vec![0, 255]),
201                },
202                TypedInsertField {
203                    name: "vector".into(),
204                    value: ParamValue::Vector(vec![1.0, -2.0]),
205                },
206                TypedInsertField {
207                    name: "json".into(),
208                    value: ParamValue::Json(br#"{"ok":true}"#.to_vec()),
209                },
210                TypedInsertField {
211                    name: "timestamp".into(),
212                    value: ParamValue::Timestamp(123),
213                },
214                TypedInsertField {
215                    name: "uuid".into(),
216                    value: ParamValue::Uuid([0xAB; 16]),
217                },
218            ],
219        }
220    }
221
222    #[test]
223    fn round_trips_all_parameter_value_variants() {
224        let request = all_values_request();
225        let encoded = encode_typed_insert(&request).expect("encode");
226        assert_eq!(decode_typed_insert(&encoded).expect("decode"), request);
227    }
228
229    #[test]
230    fn simple_request_bytes_are_pinned_for_other_drivers() {
231        let request = TypedInsertRequest {
232            collection: "users".into(),
233            fields: vec![
234                TypedInsertField {
235                    name: "name".into(),
236                    value: ParamValue::Text("Ada".into()),
237                },
238                TypedInsertField {
239                    name: "age".into(),
240                    value: ParamValue::Int(37),
241                },
242            ],
243        };
244        assert_eq!(
245            encode_typed_insert(&request).expect("encode"),
246            vec![
247                0x01, 0x05, 0x00, b'u', b's', b'e', b'r', b's', 0x02, 0x00, 0x04, 0x00, b'n', b'a',
248                b'm', b'e', 0x04, 0x03, 0x00, 0x00, 0x00, b'A', b'd', b'a', 0x03, 0x00, b'a', b'g',
249                b'e', 0x02, 0x25, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
250            ]
251        );
252    }
253
254    #[test]
255    fn preserves_empty_field_set_for_default_only_rows() {
256        let request = TypedInsertRequest {
257            collection: "defaults".into(),
258            fields: Vec::new(),
259        };
260        let encoded = encode_typed_insert(&request).expect("encode");
261        assert_eq!(decode_typed_insert(&encoded).expect("decode"), request);
262    }
263
264    #[test]
265    fn rejects_duplicate_empty_and_trailing_fields() {
266        let duplicate = TypedInsertRequest {
267            collection: "events".into(),
268            fields: vec![
269                TypedInsertField {
270                    name: "id".into(),
271                    value: ParamValue::Int(1),
272                },
273                TypedInsertField {
274                    name: "id".into(),
275                    value: ParamValue::Int(2),
276                },
277            ],
278        };
279        assert!(
280            matches!(encode_typed_insert(&duplicate), Err(TypedInsertError::DuplicateField(name)) if name == "id")
281        );
282
283        let empty = TypedInsertRequest {
284            collection: "events".into(),
285            fields: vec![TypedInsertField {
286                name: " ".into(),
287                value: ParamValue::Null,
288            }],
289        };
290        assert_eq!(
291            encode_typed_insert(&empty),
292            Err(TypedInsertError::EmptyFieldName)
293        );
294
295        let mut encoded = encode_typed_insert(&all_values_request()).expect("encode");
296        encoded.push(0);
297        assert_eq!(
298            decode_typed_insert(&encoded),
299            Err(TypedInsertError::TrailingBytes(1))
300        );
301    }
302
303    #[test]
304    fn rejects_bad_version_utf8_and_truncation() {
305        assert_eq!(
306            decode_typed_insert(&[2]),
307            Err(TypedInsertError::UnsupportedVersion(2))
308        );
309        assert_eq!(
310            decode_typed_insert(&[1, 1, 0, 0xFF, 0, 0]),
311            Err(TypedInsertError::InvalidUtf8("collection"))
312        );
313
314        let encoded = encode_typed_insert(&all_values_request()).expect("encode");
315        for len in 0..encoded.len() {
316            assert!(
317                decode_typed_insert(&encoded[..len]).is_err(),
318                "prefix {len} unexpectedly decoded"
319            );
320        }
321    }
322}
reddb_wire/redwire/typed_insert.rs

reddb_wire/redwire/
typed_insert.rs
