nodedb_wal/record/
fts_spatial.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! WAL payload structs for FTS and Spatial sync records.
4//!
5//! Payload layout: variable-length LE encoding (no serde/msgpack dep).
6//! Each payload carries the engine-specific fields plus the four provenance
7//! fields (`producer_id`, `epoch`, `stream_id`, `seq`) inline so deduplication
8//! never requires a separate decode step.
9//!
10//! Wire layout for all four payload types:
11//!
12//! ```text
13//! ┌──────────────┬──────────┬───────────┬──────────┬──────────────┬──────────────┬────────────────────┐
14//! │producer_id u64│ epoch u64│stream_id u64│  seq u64 │ name_len u32 │ collection   │ id_len u32 + id... │
15//! └──────────────┴──────────┴───────────┴──────────┴──────────────┴──────────────┴────────────────────┘
16//! ```
17//! FtsIndex additionally carries `field_count u32` followed by
18//! `field_len u32 + field bytes + text_len u32 + text bytes` per field.
19//! SpatialPut additionally carries `field_len u32 + field bytes + geometry_len u32 + geometry bytes`.
20//! SpatialDelete additionally carries `field_len u32 + field bytes`.
21//! FtsDelete and SpatialDelete prefix the id with a presence tag `u8`:
22//! `0` means the key's home binds no row and no id follows, `1` means the id
23//! follows.
24//!
25//! These structs are net-new (no legacy records to stay compatible with).
26//! Field set can be extended when the handler is wired; the length-prefixed
27//! layout is forward-compatible.
28
29use nodedb_types::decode_bounds::checked_decode_capacity;
30use nodedb_types::sync::wire::SyncProvenance;
31
32use crate::error::{Result, WalError};
33
34// ── helpers ──────────────────────────────────────────────────────────────────
35
36fn read_u32_le(buf: &[u8], offset: usize) -> Result<u32> {
37    buf.get(offset..offset + 4)
38        .map(|b| u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
39        .ok_or_else(|| WalError::InvalidPayload {
40            detail: format!("truncated at offset {offset}, need 4 bytes"),
41        })
42}
43
44fn read_u64_le(buf: &[u8], offset: usize) -> Result<u64> {
45    buf.get(offset..offset + 8)
46        .map(|b| u64::from_le_bytes([b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7]]))
47        .ok_or_else(|| WalError::InvalidPayload {
48            detail: format!("truncated at offset {offset}, need 8 bytes"),
49        })
50}
51
52fn read_utf8_field(buf: &[u8], offset: usize) -> Result<(String, usize)> {
53    let len = read_u32_le(buf, offset)? as usize;
54    let start = offset + 4;
55    buf.get(start..start + len)
56        .and_then(|b| std::str::from_utf8(b).ok())
57        .map(|s| (s.to_string(), start + len))
58        .ok_or_else(|| WalError::InvalidPayload {
59            detail: format!("invalid utf8 field at offset {offset}"),
60        })
61}
62
63fn read_bytes_field(buf: &[u8], offset: usize) -> Result<(Vec<u8>, usize)> {
64    let len = read_u32_le(buf, offset)? as usize;
65    let start = offset + 4;
66    buf.get(start..start + len)
67        .map(|b| (b.to_vec(), start + len))
68        .ok_or_else(|| WalError::InvalidPayload {
69            detail: format!("truncated bytes field at offset {offset}"),
70        })
71}
72
73fn push_u32_le(buf: &mut Vec<u8>, v: u32) {
74    buf.extend_from_slice(&v.to_le_bytes());
75}
76
77fn push_u64_le(buf: &mut Vec<u8>, v: u64) {
78    buf.extend_from_slice(&v.to_le_bytes());
79}
80
81fn push_str_field(buf: &mut Vec<u8>, s: &str) -> Result<()> {
82    let bytes = s.as_bytes();
83    if bytes.len() > u32::MAX as usize {
84        return Err(WalError::InvalidPayload {
85            detail: format!("string field too long: {} bytes", bytes.len()),
86        });
87    }
88    push_u32_le(buf, bytes.len() as u32);
89    buf.extend_from_slice(bytes);
90    Ok(())
91}
92
93/// Presence tag of an optional string field: absent.
94const FIELD_ABSENT: u8 = 0;
95/// Presence tag of an optional string field: present, followed by the field.
96const FIELD_PRESENT: u8 = 1;
97
98fn push_opt_str_field(buf: &mut Vec<u8>, s: Option<&str>) -> Result<()> {
99    match s {
100        None => {
101            buf.push(FIELD_ABSENT);
102            Ok(())
103        }
104        Some(s) => {
105            buf.push(FIELD_PRESENT);
106            push_str_field(buf, s)
107        }
108    }
109}
110
111fn read_opt_utf8_field(buf: &[u8], offset: usize) -> Result<(Option<String>, usize)> {
112    match buf.get(offset).copied() {
113        Some(FIELD_ABSENT) => Ok((None, offset + 1)),
114        Some(FIELD_PRESENT) => {
115            let (s, next) = read_utf8_field(buf, offset + 1)?;
116            Ok((Some(s), next))
117        }
118        Some(tag) => Err(WalError::InvalidPayload {
119            detail: format!("optional field at offset {offset} has unknown presence tag {tag}"),
120        }),
121        None => Err(WalError::InvalidPayload {
122            detail: format!("truncated at offset {offset}, need a presence tag"),
123        }),
124    }
125}
126
127fn push_bytes_field(buf: &mut Vec<u8>, data: &[u8]) -> Result<()> {
128    if data.len() > u32::MAX as usize {
129        return Err(WalError::InvalidPayload {
130            detail: format!("bytes field too long: {} bytes", data.len()),
131        });
132    }
133    push_u32_le(buf, data.len() as u32);
134    buf.extend_from_slice(data);
135    Ok(())
136}
137
138// ── common provenance decode/encode helpers ───────────────────────────────────
139
140fn read_provenance(buf: &[u8]) -> Result<(SyncProvenance, usize)> {
141    let producer_id = read_u64_le(buf, 0)?;
142    let epoch = read_u64_le(buf, 8)?;
143    let stream_id = read_u64_le(buf, 16)?;
144    let seq = read_u64_le(buf, 24)?;
145    Ok((
146        SyncProvenance {
147            producer_id,
148            epoch,
149            stream_id,
150            seq,
151        },
152        32,
153    ))
154}
155
156fn push_provenance(buf: &mut Vec<u8>, prov: &SyncProvenance) {
157    push_u64_le(buf, prov.producer_id);
158    push_u64_le(buf, prov.epoch);
159    push_u64_le(buf, prov.stream_id);
160    push_u64_le(buf, prov.seq);
161}
162
163// ── FtsIndexPayload ────────────────────────────────────────────────────────────
164
165/// WAL payload for `RecordType::FtsIndex`.
166///
167/// Carries the minimum fields needed for Data-Plane replay: the collection
168/// name, document identifier, the document's `(field, text)` pairs, and
169/// producer provenance for idempotency checks. Empty `fields` remove the
170/// document from every index.
171#[derive(Debug, Clone, PartialEq, Eq)]
172pub struct FtsIndexPayload {
173    /// Producer provenance for idempotency checks.
174    pub provenance: SyncProvenance,
175    /// Target collection name.
176    pub collection: String,
177    /// External document identifier.
178    pub doc_id: String,
179    /// `(field, text)` per top-level string field.
180    pub fields: Vec<(String, String)>,
181}
182
183/// Smallest encoding of one `(field, text)` pair: two empty length-prefixed
184/// strings.
185const MIN_FIELD_PAIR_BYTES: usize = 8;
186
187impl FtsIndexPayload {
188    pub fn new(
189        provenance: SyncProvenance,
190        collection: impl Into<String>,
191        doc_id: impl Into<String>,
192        fields: Vec<(String, String)>,
193    ) -> Self {
194        Self {
195            provenance,
196            collection: collection.into(),
197            doc_id: doc_id.into(),
198            fields,
199        }
200    }
201
202    pub fn to_bytes(&self) -> Result<Vec<u8>> {
203        let mut buf = Vec::new();
204        push_provenance(&mut buf, &self.provenance);
205        push_str_field(&mut buf, &self.collection)?;
206        push_str_field(&mut buf, &self.doc_id)?;
207        let count = u32::try_from(self.fields.len()).map_err(|_| WalError::InvalidPayload {
208            detail: format!("too many FTS fields: {}", self.fields.len()),
209        })?;
210        push_u32_le(&mut buf, count);
211        for (field, text) in &self.fields {
212            push_str_field(&mut buf, field)?;
213            push_str_field(&mut buf, text)?;
214        }
215        Ok(buf)
216    }
217
218    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
219        let (provenance, mut off) = read_provenance(buf)?;
220        let (collection, next) = read_utf8_field(buf, off)?;
221        off = next;
222        let (doc_id, next) = read_utf8_field(buf, off)?;
223        off = next;
224        let count = read_u32_le(buf, off)? as usize;
225        off += 4;
226        let remaining = buf.len().saturating_sub(off);
227        let Some(capacity) = checked_decode_capacity(
228            count,
229            std::mem::size_of::<(String, String)>(),
230            remaining,
231            MIN_FIELD_PAIR_BYTES,
232            u32::MAX as usize,
233            isize::MAX as usize,
234        ) else {
235            return Err(WalError::InvalidPayload {
236                detail: format!(
237                    "FTS field count {count} exceeds what {remaining} remaining bytes can hold"
238                ),
239            });
240        };
241        let mut fields = Vec::with_capacity(capacity);
242        for _ in 0..count {
243            let (field, next) = read_utf8_field(buf, off)?;
244            let (text, next) = read_utf8_field(buf, next)?;
245            off = next;
246            fields.push((field, text));
247        }
248        Ok(Self {
249            provenance,
250            collection,
251            doc_id,
252            fields,
253        })
254    }
255}
256
257// ── FtsDeletePayload ───────────────────────────────────────────────────────────
258
259/// WAL payload for `RecordType::FtsDelete`.
260#[derive(Debug, Clone, PartialEq, Eq)]
261pub struct FtsDeletePayload {
262    pub provenance: SyncProvenance,
263    pub collection: String,
264    /// The deleted document's identifier. `None` when the key's home binds
265    /// no row: the delete removes nothing and still commits its provenance.
266    pub doc_id: Option<String>,
267}
268
269impl FtsDeletePayload {
270    pub fn new(
271        provenance: SyncProvenance,
272        collection: impl Into<String>,
273        doc_id: Option<String>,
274    ) -> Self {
275        Self {
276            provenance,
277            collection: collection.into(),
278            doc_id,
279        }
280    }
281
282    pub fn to_bytes(&self) -> Result<Vec<u8>> {
283        let mut buf = Vec::new();
284        push_provenance(&mut buf, &self.provenance);
285        push_str_field(&mut buf, &self.collection)?;
286        push_opt_str_field(&mut buf, self.doc_id.as_deref())?;
287        Ok(buf)
288    }
289
290    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
291        let (provenance, mut off) = read_provenance(buf)?;
292        let (collection, next) = read_utf8_field(buf, off)?;
293        off = next;
294        let (doc_id, _) = read_opt_utf8_field(buf, off)?;
295        Ok(Self {
296            provenance,
297            collection,
298            doc_id,
299        })
300    }
301}
302
303// ── SpatialPutPayload ──────────────────────────────────────────────────────────
304
305/// WAL payload for `RecordType::SpatialPut`.
306///
307/// `geometry_bytes` is a MessagePack-serialised `nodedb_types::geometry::Geometry`
308/// value (the same encoding used in `SpatialInsertMsg`).
309#[derive(Debug, Clone, PartialEq, Eq)]
310pub struct SpatialPutPayload {
311    pub provenance: SyncProvenance,
312    pub collection: String,
313    pub field: String,
314    pub doc_id: String,
315    pub geometry_bytes: Vec<u8>,
316}
317
318impl SpatialPutPayload {
319    pub fn new(
320        provenance: SyncProvenance,
321        collection: impl Into<String>,
322        field: impl Into<String>,
323        doc_id: impl Into<String>,
324        geometry_bytes: impl Into<Vec<u8>>,
325    ) -> Self {
326        Self {
327            provenance,
328            collection: collection.into(),
329            field: field.into(),
330            doc_id: doc_id.into(),
331            geometry_bytes: geometry_bytes.into(),
332        }
333    }
334
335    pub fn to_bytes(&self) -> Result<Vec<u8>> {
336        let mut buf = Vec::new();
337        push_provenance(&mut buf, &self.provenance);
338        push_str_field(&mut buf, &self.collection)?;
339        push_str_field(&mut buf, &self.field)?;
340        push_str_field(&mut buf, &self.doc_id)?;
341        push_bytes_field(&mut buf, &self.geometry_bytes)?;
342        Ok(buf)
343    }
344
345    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
346        let (provenance, mut off) = read_provenance(buf)?;
347        let (collection, next) = read_utf8_field(buf, off)?;
348        off = next;
349        let (field, next) = read_utf8_field(buf, off)?;
350        off = next;
351        let (doc_id, next) = read_utf8_field(buf, off)?;
352        off = next;
353        let (geometry_bytes, _) = read_bytes_field(buf, off)?;
354        Ok(Self {
355            provenance,
356            collection,
357            field,
358            doc_id,
359            geometry_bytes,
360        })
361    }
362}
363
364// ── SpatialDeletePayload ───────────────────────────────────────────────────────
365
366/// WAL payload for `RecordType::SpatialDelete`.
367#[derive(Debug, Clone, PartialEq, Eq)]
368pub struct SpatialDeletePayload {
369    pub provenance: SyncProvenance,
370    pub collection: String,
371    pub field: String,
372    /// The deleted row's identifier. `None` when the key's home binds no
373    /// row: the delete removes nothing and still commits its provenance.
374    pub doc_id: Option<String>,
375}
376
377impl SpatialDeletePayload {
378    pub fn new(
379        provenance: SyncProvenance,
380        collection: impl Into<String>,
381        field: impl Into<String>,
382        doc_id: Option<String>,
383    ) -> Self {
384        Self {
385            provenance,
386            collection: collection.into(),
387            field: field.into(),
388            doc_id,
389        }
390    }
391
392    pub fn to_bytes(&self) -> Result<Vec<u8>> {
393        let mut buf = Vec::new();
394        push_provenance(&mut buf, &self.provenance);
395        push_str_field(&mut buf, &self.collection)?;
396        push_str_field(&mut buf, &self.field)?;
397        push_opt_str_field(&mut buf, self.doc_id.as_deref())?;
398        Ok(buf)
399    }
400
401    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
402        let (provenance, mut off) = read_provenance(buf)?;
403        let (collection, next) = read_utf8_field(buf, off)?;
404        off = next;
405        let (field, next) = read_utf8_field(buf, off)?;
406        off = next;
407        let (doc_id, _) = read_opt_utf8_field(buf, off)?;
408        Ok(Self {
409            provenance,
410            collection,
411            field,
412            doc_id,
413        })
414    }
415}
416
417#[cfg(test)]
418mod tests {
419    use super::*;
420
421    fn prov(producer_id: u64, epoch: u64, stream_id: u64, seq: u64) -> SyncProvenance {
422        SyncProvenance {
423            producer_id,
424            epoch,
425            stream_id,
426            seq,
427        }
428    }
429
430    fn pairs(fields: &[(&str, &str)]) -> Vec<(String, String)> {
431        fields
432            .iter()
433            .map(|(f, t)| ((*f).to_string(), (*t).to_string()))
434            .collect()
435    }
436
437    #[test]
438    fn fts_index_roundtrip() {
439        let p = FtsIndexPayload::new(
440            prov(0xCAFE_BABE, 3, 7, 42),
441            "articles",
442            "doc-1",
443            pairs(&[("body", "hello world"), ("title", "Rust"), ("empty", "")]),
444        );
445        let bytes = p.to_bytes().unwrap();
446        assert_eq!(FtsIndexPayload::from_bytes(&bytes).unwrap(), p);
447    }
448
449    #[test]
450    fn fts_index_no_fields_roundtrip() {
451        let p = FtsIndexPayload::new(prov(0, 0, 0, 0), "c", "d", Vec::new());
452        assert_eq!(
453            FtsIndexPayload::from_bytes(&p.to_bytes().unwrap()).unwrap(),
454            p
455        );
456    }
457
458    #[test]
459    fn fts_delete_roundtrip() {
460        let p = FtsDeletePayload::new(prov(1, 2, 3, 4), "articles", Some("doc-99".to_string()));
461        let bytes = p.to_bytes().unwrap();
462        assert_eq!(FtsDeletePayload::from_bytes(&bytes).unwrap(), p);
463    }
464
465    #[test]
466    fn fts_delete_of_an_unbound_key_roundtrip() {
467        let p = FtsDeletePayload::new(prov(1, 2, 3, 4), "articles", None);
468        let bytes = p.to_bytes().unwrap();
469        assert_eq!(FtsDeletePayload::from_bytes(&bytes).unwrap(), p);
470    }
471
472    #[test]
473    fn unknown_presence_tag_rejected() {
474        let p = FtsDeletePayload::new(prov(1, 2, 3, 4), "articles", None);
475        let mut bytes = p.to_bytes().unwrap();
476        if let Some(tag) = bytes.last_mut() {
477            *tag = 7;
478        }
479        assert!(FtsDeletePayload::from_bytes(&bytes).is_err());
480    }
481
482    #[test]
483    fn spatial_put_roundtrip() {
484        let p = SpatialPutPayload::new(
485            prov(5, 6, 7, 8),
486            "places",
487            "loc",
488            "poi-1",
489            vec![0xDE, 0xAD, 0xBE, 0xEF],
490        );
491        let bytes = p.to_bytes().unwrap();
492        assert_eq!(SpatialPutPayload::from_bytes(&bytes).unwrap(), p);
493    }
494
495    #[test]
496    fn spatial_put_empty_geometry_roundtrip() {
497        let p = SpatialPutPayload::new(prov(0, 0, 0, 0), "c", "f", "d", vec![]);
498        assert_eq!(
499            SpatialPutPayload::from_bytes(&p.to_bytes().unwrap()).unwrap(),
500            p
501        );
502    }
503
504    #[test]
505    fn spatial_delete_roundtrip() {
506        let p = SpatialDeletePayload::new(
507            prov(9, 10, 11, 12),
508            "places",
509            "loc",
510            Some("poi-1".to_string()),
511        );
512        let bytes = p.to_bytes().unwrap();
513        assert_eq!(SpatialDeletePayload::from_bytes(&bytes).unwrap(), p);
514    }
515
516    #[test]
517    fn spatial_delete_of_an_unbound_key_roundtrip() {
518        let p = SpatialDeletePayload::new(prov(9, 10, 11, 12), "places", "loc", None);
519        let bytes = p.to_bytes().unwrap();
520        assert_eq!(SpatialDeletePayload::from_bytes(&bytes).unwrap(), p);
521    }
522
523    #[test]
524    fn truncated_buf_rejected() {
525        let p = FtsIndexPayload::new(prov(1, 2, 3, 4), "col", "id", pairs(&[("body", "text")]));
526        let bytes = p.to_bytes().unwrap();
527        // Truncated to just provenance — should fail on collection field.
528        assert!(FtsIndexPayload::from_bytes(&bytes[..32]).is_err());
529        // Truncated inside the last field's text.
530        assert!(FtsIndexPayload::from_bytes(&bytes[..bytes.len() - 1]).is_err());
531    }
532
533    #[test]
534    fn fts_index_field_count_past_the_buffer_is_rejected() {
535        let p = FtsIndexPayload::new(prov(1, 2, 3, 4), "col", "id", Vec::new());
536        let mut bytes = p.to_bytes().unwrap();
537        let count_at = bytes.len() - 4;
538        bytes[count_at..].copy_from_slice(&u32::MAX.to_le_bytes());
539        assert!(FtsIndexPayload::from_bytes(&bytes).is_err());
540    }
541}
nodedb_wal/record/fts_spatial.rs

nodedb_wal/record/
fts_spatial.rs
