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 `text_len u32 + text bytes`.
18//! SpatialPut additionally carries `field_len u32 + field bytes + geometry_len u32 + geometry bytes`.
19//! SpatialDelete additionally carries `field_len u32 + field bytes`.
20//!
21//! These structs are net-new (no legacy records to stay compatible with).
22//! Field set can be extended when the handler is wired; the length-prefixed
23//! layout is forward-compatible.
24
25use nodedb_types::sync::wire::SyncProvenance;
26
27use crate::error::{Result, WalError};
28
29// ── helpers ──────────────────────────────────────────────────────────────────
30
31fn read_u32_le(buf: &[u8], offset: usize) -> Result<u32> {
32    buf.get(offset..offset + 4)
33        .map(|b| u32::from_le_bytes([b[0], b[1], b[2], b[3]]))
34        .ok_or_else(|| WalError::InvalidPayload {
35            detail: format!("truncated at offset {offset}, need 4 bytes"),
36        })
37}
38
39fn read_u64_le(buf: &[u8], offset: usize) -> Result<u64> {
40    buf.get(offset..offset + 8)
41        .map(|b| u64::from_le_bytes([b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7]]))
42        .ok_or_else(|| WalError::InvalidPayload {
43            detail: format!("truncated at offset {offset}, need 8 bytes"),
44        })
45}
46
47fn read_utf8_field(buf: &[u8], offset: usize) -> Result<(String, usize)> {
48    let len = read_u32_le(buf, offset)? as usize;
49    let start = offset + 4;
50    buf.get(start..start + len)
51        .and_then(|b| std::str::from_utf8(b).ok())
52        .map(|s| (s.to_string(), start + len))
53        .ok_or_else(|| WalError::InvalidPayload {
54            detail: format!("invalid utf8 field at offset {offset}"),
55        })
56}
57
58fn read_bytes_field(buf: &[u8], offset: usize) -> Result<(Vec<u8>, usize)> {
59    let len = read_u32_le(buf, offset)? as usize;
60    let start = offset + 4;
61    buf.get(start..start + len)
62        .map(|b| (b.to_vec(), start + len))
63        .ok_or_else(|| WalError::InvalidPayload {
64            detail: format!("truncated bytes field at offset {offset}"),
65        })
66}
67
68fn push_u32_le(buf: &mut Vec<u8>, v: u32) {
69    buf.extend_from_slice(&v.to_le_bytes());
70}
71
72fn push_u64_le(buf: &mut Vec<u8>, v: u64) {
73    buf.extend_from_slice(&v.to_le_bytes());
74}
75
76fn push_str_field(buf: &mut Vec<u8>, s: &str) -> Result<()> {
77    let bytes = s.as_bytes();
78    if bytes.len() > u32::MAX as usize {
79        return Err(WalError::InvalidPayload {
80            detail: format!("string field too long: {} bytes", bytes.len()),
81        });
82    }
83    push_u32_le(buf, bytes.len() as u32);
84    buf.extend_from_slice(bytes);
85    Ok(())
86}
87
88fn push_bytes_field(buf: &mut Vec<u8>, data: &[u8]) -> Result<()> {
89    if data.len() > u32::MAX as usize {
90        return Err(WalError::InvalidPayload {
91            detail: format!("bytes field too long: {} bytes", data.len()),
92        });
93    }
94    push_u32_le(buf, data.len() as u32);
95    buf.extend_from_slice(data);
96    Ok(())
97}
98
99// ── common provenance decode/encode helpers ───────────────────────────────────
100
101fn read_provenance(buf: &[u8]) -> Result<(SyncProvenance, usize)> {
102    let producer_id = read_u64_le(buf, 0)?;
103    let epoch = read_u64_le(buf, 8)?;
104    let stream_id = read_u64_le(buf, 16)?;
105    let seq = read_u64_le(buf, 24)?;
106    Ok((
107        SyncProvenance {
108            producer_id,
109            epoch,
110            stream_id,
111            seq,
112        },
113        32,
114    ))
115}
116
117fn push_provenance(buf: &mut Vec<u8>, prov: &SyncProvenance) {
118    push_u64_le(buf, prov.producer_id);
119    push_u64_le(buf, prov.epoch);
120    push_u64_le(buf, prov.stream_id);
121    push_u64_le(buf, prov.seq);
122}
123
124// ── FtsIndexPayload ────────────────────────────────────────────────────────────
125
126/// WAL payload for `RecordType::FtsIndex`.
127///
128/// Carries the minimum fields needed for Data-Plane replay: the collection
129/// name, document identifier, the text to index, and producer provenance for
130/// idempotency checks. Additional fields (field weights, analyzer config) can
131/// be appended when the handler is wired — the length-prefixed layout is
132/// forward-compatible.
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct FtsIndexPayload {
135    /// Producer provenance for idempotency checks.
136    pub provenance: SyncProvenance,
137    /// Target collection name.
138    pub collection: String,
139    /// External document identifier.
140    pub doc_id: String,
141    /// Concatenated text to index.
142    pub text: String,
143}
144
145impl FtsIndexPayload {
146    pub fn new(
147        provenance: SyncProvenance,
148        collection: impl Into<String>,
149        doc_id: impl Into<String>,
150        text: impl Into<String>,
151    ) -> Self {
152        Self {
153            provenance,
154            collection: collection.into(),
155            doc_id: doc_id.into(),
156            text: text.into(),
157        }
158    }
159
160    pub fn to_bytes(&self) -> Result<Vec<u8>> {
161        let mut buf = Vec::new();
162        push_provenance(&mut buf, &self.provenance);
163        push_str_field(&mut buf, &self.collection)?;
164        push_str_field(&mut buf, &self.doc_id)?;
165        push_str_field(&mut buf, &self.text)?;
166        Ok(buf)
167    }
168
169    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
170        let (provenance, mut off) = read_provenance(buf)?;
171        let (collection, next) = read_utf8_field(buf, off)?;
172        off = next;
173        let (doc_id, next) = read_utf8_field(buf, off)?;
174        off = next;
175        let (text, _) = read_utf8_field(buf, off)?;
176        Ok(Self {
177            provenance,
178            collection,
179            doc_id,
180            text,
181        })
182    }
183}
184
185// ── FtsDeletePayload ───────────────────────────────────────────────────────────
186
187/// WAL payload for `RecordType::FtsDelete`.
188#[derive(Debug, Clone, PartialEq, Eq)]
189pub struct FtsDeletePayload {
190    pub provenance: SyncProvenance,
191    pub collection: String,
192    pub doc_id: String,
193}
194
195impl FtsDeletePayload {
196    pub fn new(
197        provenance: SyncProvenance,
198        collection: impl Into<String>,
199        doc_id: impl Into<String>,
200    ) -> Self {
201        Self {
202            provenance,
203            collection: collection.into(),
204            doc_id: doc_id.into(),
205        }
206    }
207
208    pub fn to_bytes(&self) -> Result<Vec<u8>> {
209        let mut buf = Vec::new();
210        push_provenance(&mut buf, &self.provenance);
211        push_str_field(&mut buf, &self.collection)?;
212        push_str_field(&mut buf, &self.doc_id)?;
213        Ok(buf)
214    }
215
216    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
217        let (provenance, mut off) = read_provenance(buf)?;
218        let (collection, next) = read_utf8_field(buf, off)?;
219        off = next;
220        let (doc_id, _) = read_utf8_field(buf, off)?;
221        Ok(Self {
222            provenance,
223            collection,
224            doc_id,
225        })
226    }
227}
228
229// ── SpatialPutPayload ──────────────────────────────────────────────────────────
230
231/// WAL payload for `RecordType::SpatialPut`.
232///
233/// `geometry_bytes` is a MessagePack-serialised `nodedb_types::geometry::Geometry`
234/// value (the same encoding used in `SpatialInsertMsg`).
235#[derive(Debug, Clone, PartialEq, Eq)]
236pub struct SpatialPutPayload {
237    pub provenance: SyncProvenance,
238    pub collection: String,
239    pub field: String,
240    pub doc_id: String,
241    pub geometry_bytes: Vec<u8>,
242}
243
244impl SpatialPutPayload {
245    pub fn new(
246        provenance: SyncProvenance,
247        collection: impl Into<String>,
248        field: impl Into<String>,
249        doc_id: impl Into<String>,
250        geometry_bytes: impl Into<Vec<u8>>,
251    ) -> Self {
252        Self {
253            provenance,
254            collection: collection.into(),
255            field: field.into(),
256            doc_id: doc_id.into(),
257            geometry_bytes: geometry_bytes.into(),
258        }
259    }
260
261    pub fn to_bytes(&self) -> Result<Vec<u8>> {
262        let mut buf = Vec::new();
263        push_provenance(&mut buf, &self.provenance);
264        push_str_field(&mut buf, &self.collection)?;
265        push_str_field(&mut buf, &self.field)?;
266        push_str_field(&mut buf, &self.doc_id)?;
267        push_bytes_field(&mut buf, &self.geometry_bytes)?;
268        Ok(buf)
269    }
270
271    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
272        let (provenance, mut off) = read_provenance(buf)?;
273        let (collection, next) = read_utf8_field(buf, off)?;
274        off = next;
275        let (field, next) = read_utf8_field(buf, off)?;
276        off = next;
277        let (doc_id, next) = read_utf8_field(buf, off)?;
278        off = next;
279        let (geometry_bytes, _) = read_bytes_field(buf, off)?;
280        Ok(Self {
281            provenance,
282            collection,
283            field,
284            doc_id,
285            geometry_bytes,
286        })
287    }
288}
289
290// ── SpatialDeletePayload ───────────────────────────────────────────────────────
291
292/// WAL payload for `RecordType::SpatialDelete`.
293#[derive(Debug, Clone, PartialEq, Eq)]
294pub struct SpatialDeletePayload {
295    pub provenance: SyncProvenance,
296    pub collection: String,
297    pub field: String,
298    pub doc_id: String,
299}
300
301impl SpatialDeletePayload {
302    pub fn new(
303        provenance: SyncProvenance,
304        collection: impl Into<String>,
305        field: impl Into<String>,
306        doc_id: impl Into<String>,
307    ) -> Self {
308        Self {
309            provenance,
310            collection: collection.into(),
311            field: field.into(),
312            doc_id: doc_id.into(),
313        }
314    }
315
316    pub fn to_bytes(&self) -> Result<Vec<u8>> {
317        let mut buf = Vec::new();
318        push_provenance(&mut buf, &self.provenance);
319        push_str_field(&mut buf, &self.collection)?;
320        push_str_field(&mut buf, &self.field)?;
321        push_str_field(&mut buf, &self.doc_id)?;
322        Ok(buf)
323    }
324
325    pub fn from_bytes(buf: &[u8]) -> Result<Self> {
326        let (provenance, mut off) = read_provenance(buf)?;
327        let (collection, next) = read_utf8_field(buf, off)?;
328        off = next;
329        let (field, next) = read_utf8_field(buf, off)?;
330        off = next;
331        let (doc_id, _) = read_utf8_field(buf, off)?;
332        Ok(Self {
333            provenance,
334            collection,
335            field,
336            doc_id,
337        })
338    }
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344
345    fn prov(producer_id: u64, epoch: u64, stream_id: u64, seq: u64) -> SyncProvenance {
346        SyncProvenance {
347            producer_id,
348            epoch,
349            stream_id,
350            seq,
351        }
352    }
353
354    #[test]
355    fn fts_index_roundtrip() {
356        let p = FtsIndexPayload::new(
357            prov(0xCAFE_BABE, 3, 7, 42),
358            "articles",
359            "doc-1",
360            "hello world",
361        );
362        let bytes = p.to_bytes().unwrap();
363        assert_eq!(FtsIndexPayload::from_bytes(&bytes).unwrap(), p);
364    }
365
366    #[test]
367    fn fts_index_empty_text_roundtrip() {
368        let p = FtsIndexPayload::new(prov(0, 0, 0, 0), "c", "d", "");
369        assert_eq!(
370            FtsIndexPayload::from_bytes(&p.to_bytes().unwrap()).unwrap(),
371            p
372        );
373    }
374
375    #[test]
376    fn fts_delete_roundtrip() {
377        let p = FtsDeletePayload::new(prov(1, 2, 3, 4), "articles", "doc-99");
378        let bytes = p.to_bytes().unwrap();
379        assert_eq!(FtsDeletePayload::from_bytes(&bytes).unwrap(), p);
380    }
381
382    #[test]
383    fn spatial_put_roundtrip() {
384        let p = SpatialPutPayload::new(
385            prov(5, 6, 7, 8),
386            "places",
387            "loc",
388            "poi-1",
389            vec![0xDE, 0xAD, 0xBE, 0xEF],
390        );
391        let bytes = p.to_bytes().unwrap();
392        assert_eq!(SpatialPutPayload::from_bytes(&bytes).unwrap(), p);
393    }
394
395    #[test]
396    fn spatial_put_empty_geometry_roundtrip() {
397        let p = SpatialPutPayload::new(prov(0, 0, 0, 0), "c", "f", "d", vec![]);
398        assert_eq!(
399            SpatialPutPayload::from_bytes(&p.to_bytes().unwrap()).unwrap(),
400            p
401        );
402    }
403
404    #[test]
405    fn spatial_delete_roundtrip() {
406        let p = SpatialDeletePayload::new(prov(9, 10, 11, 12), "places", "loc", "poi-1");
407        let bytes = p.to_bytes().unwrap();
408        assert_eq!(SpatialDeletePayload::from_bytes(&bytes).unwrap(), p);
409    }
410
411    #[test]
412    fn truncated_buf_rejected() {
413        let p = FtsIndexPayload::new(prov(1, 2, 3, 4), "col", "id", "text");
414        let bytes = p.to_bytes().unwrap();
415        // Truncated to just provenance — should fail on collection field.
416        assert!(FtsIndexPayload::from_bytes(&bytes[..32]).is_err());
417    }
418}
nodedb_wal/record/fts_spatial.rs

nodedb_wal/record/
fts_spatial.rs
