nodedb_types/sync/wire/
shape.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Shape subscribe/snapshot/delta/unsubscribe + vector clock sync messages.
4
5use std::collections::HashMap;
6
7use serde::{Deserialize, Serialize};
8
9use crate::sync::shape::ShapeDefinition;
10
11/// Shape subscribe request (client → server, 0x20).
12#[derive(
13    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
14)]
15pub struct ShapeSubscribeMsg {
16    /// Shape definition to subscribe to.
17    pub shape: ShapeDefinition,
18}
19
20/// Shape snapshot response (server → client, 0x21).
21#[derive(
22    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
23)]
24pub struct ShapeSnapshotMsg {
25    /// Shape ID this snapshot belongs to.
26    pub shape_id: String,
27    /// Initial dataset: serialized document rows matching the shape.
28    pub data: Vec<u8>,
29    /// LSN at snapshot time — deltas after this LSN will follow.
30    pub snapshot_lsn: u64,
31    /// Number of documents in the snapshot.
32    pub doc_count: usize,
33}
34
35/// Shape delta message (server → client, 0x22).
36#[derive(
37    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
38)]
39pub struct ShapeDeltaMsg {
40    /// Shape ID this delta applies to.
41    pub shape_id: String,
42    /// Collection affected.
43    pub collection: String,
44    /// Document ID affected.
45    pub document_id: String,
46    /// Operation type: "INSERT", "UPDATE", "DELETE".
47    pub operation: String,
48    /// Delta payload (CRDT delta bytes or document value).
49    pub delta: Vec<u8>,
50    /// WAL LSN of this mutation.
51    pub lsn: u64,
52}
53
54/// Shape unsubscribe request (client → server, 0x23).
55#[derive(
56    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
57)]
58pub struct ShapeUnsubscribeMsg {
59    pub shape_id: String,
60}
61
62/// Vector clock sync message (bidirectional, 0x30).
63#[derive(
64    Debug, Clone, Serialize, Deserialize, zerompk::ToMessagePack, zerompk::FromMessagePack,
65)]
66pub struct VectorClockSyncMsg {
67    /// Per-collection clock: `{ collection: max_lsn }`.
68    pub clocks: HashMap<String, u64>,
69    /// Sender's node/peer ID.
70    pub sender_id: u64,
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use crate::sync::shape::ShapeType;
77    use crate::sync::wire::{SyncFrame, SyncMessageType};
78
79    #[test]
80    fn shape_subscribe_roundtrip() {
81        let msg = ShapeSubscribeMsg {
82            shape: ShapeDefinition {
83                shape_id: "s1".into(),
84                tenant_id: 1,
85                shape_type: ShapeType::Vector {
86                    collection: "embeddings".into(),
87                    field_name: None,
88                },
89                description: "all embeddings".into(),
90                field_filter: vec![],
91            },
92        };
93        let frame = SyncFrame::new_msgpack(SyncMessageType::ShapeSubscribe, &msg).unwrap();
94        let decoded: ShapeSubscribeMsg = SyncFrame::from_bytes(&frame.to_bytes())
95            .unwrap()
96            .decode_body()
97            .unwrap();
98        assert_eq!(decoded.shape.shape_id, "s1");
99    }
100}
nodedb_types/sync/wire/shape.rs

nodedb_types/sync/wire/
shape.rs
