appcore_sync/sync/
wire.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: wire.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/21 19:22:40 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/24 16:07:49 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Versioned wire envelopes for replication messages.
12
13use crate::sync::error::{SyncError, SyncResult, UPDATE_REQUIRED_MESSAGE};
14use crate::sync::types::SyncMessage;
15use appcore_core::{CoreCompatibilityPolicy, CoreIdentity};
16
17/// Schema identifier for the first identity-aware sync wire envelope.
18pub const SYNC_WIRE_SCHEMA_V1: &str = "appcore.sync.v1";
19
20/// Versioned sync envelope carrying the complete distributed source identity.
21#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
22pub struct SyncEnvelopeV1 {
23    /// Stable wire schema identifier.
24    pub schema: String,
25    /// Identity of the Core that produced the replication batch.
26    pub source_identity: CoreIdentity,
27    /// Replication message protected by its existing event hash chain.
28    pub message: SyncMessage,
29}
30
31#[derive(serde::Serialize)]
32struct SyncEnvelopeV1Ref<'a> {
33    schema: &'static str,
34    source_identity: &'a CoreIdentity,
35    message: &'a SyncMessage,
36}
37
38impl SyncEnvelopeV1 {
39    /// Creates a v1 envelope after checking the message/source node binding.
40    pub fn new(source_identity: CoreIdentity, message: SyncMessage) -> SyncResult<Self> {
41        validate_source_binding(&source_identity, &message)?;
42        Ok(Self {
43            schema: SYNC_WIRE_SCHEMA_V1.to_string(),
44            source_identity,
45            message,
46        })
47    }
48
49    /// Verifies schema, source binding and compatibility with the receiver.
50    pub fn validate_for(&self, local_identity: &CoreIdentity) -> SyncResult<()> {
51        if self.schema != SYNC_WIRE_SCHEMA_V1 {
52            return Err(SyncError::InvalidSyncMessage(
53                "unsupported sync wire schema",
54            ));
55        }
56        if self.source_identity.runtime.node_id != self.message.source_node_id {
57            return Err(SyncError::InvalidSyncMessage(
58                "source identity does not match message node",
59            ));
60        }
61        let policy = CoreCompatibilityPolicy {
62            require_same_cluster: true,
63            required_capability: None,
64        };
65        local_identity
66            .ensure_compatible(&self.source_identity, &policy, &[])
67            .map_err(|_| SyncError::IncompatiblePeer)
68    }
69}
70
71/// Encodes an identity-aware v1 envelope as JSON.
72pub fn encode_sync_envelope_v1(
73    source_identity: &CoreIdentity,
74    message: &SyncMessage,
75) -> SyncResult<String> {
76    validate_source_binding(source_identity, message)?;
77    let envelope = SyncEnvelopeV1Ref {
78        schema: SYNC_WIRE_SCHEMA_V1,
79        source_identity,
80        message,
81    };
82    serde_json::to_string(&envelope)
83        .map_err(|_| SyncError::InvalidSyncMessage("sync wire serialization failed"))
84}
85
86fn validate_source_binding(
87    source_identity: &CoreIdentity,
88    message: &SyncMessage,
89) -> SyncResult<()> {
90    if source_identity.runtime.node_id != message.source_node_id {
91        return Err(SyncError::InvalidSyncMessage(
92            "source identity does not match message node",
93        ));
94    }
95    Ok(())
96}
97
98/// Decodes the identity-aware v1 JSON envelope.
99pub fn decode_sync_envelope(input: &str) -> SyncResult<SyncEnvelopeV1> {
100    if input.is_empty() {
101        return Err(SyncError::EmptyRequestBody);
102    }
103    if !input.trim_start().starts_with('{') {
104        return Err(SyncError::InvalidSyncMessage(UPDATE_REQUIRED_MESSAGE));
105    }
106    let envelope = serde_json::from_str::<SyncEnvelopeV1>(input)
107        .map_err(|_| SyncError::InvalidSyncMessage("invalid sync wire envelope"))?;
108    if envelope.schema != SYNC_WIRE_SCHEMA_V1 {
109        return Err(SyncError::InvalidSyncMessage(UPDATE_REQUIRED_MESSAGE));
110    }
111    if envelope.source_identity.runtime.node_id != envelope.message.source_node_id {
112        return Err(SyncError::InvalidSyncMessage(
113            "source identity does not match message node",
114        ));
115    }
116    Ok(envelope)
117}
118
119#[cfg(test)]
120mod tests {
121    use super::*;
122    use crate::sync::{MAX_SYNC_BATCH_PAYLOAD_BYTES, MAX_SYNC_REQUEST_BODY_BYTES};
123    use appcore_core::{
124        AppFamily, AppId, ClusterId, CoreId, CoreKind, InstanceId, NodeId, ProtocolVersion,
125        RuntimeContractVersion, RuntimeIdentity, SyncGroup, TenantId,
126    };
127
128    fn identity(tenant: &str, node: &str) -> CoreIdentity {
129        CoreIdentity {
130            tenant_id: TenantId::new(tenant).unwrap(),
131            cluster_id: ClusterId::new("cluster-a").unwrap(),
132            core_id: CoreId::new(format!("core-{node}")).unwrap(),
133            instance_id: InstanceId::new(format!("instance-{node}")).unwrap(),
134            kind: CoreKind::new("replica").unwrap(),
135            protocol_version: ProtocolVersion::new(1),
136            runtime: RuntimeIdentity {
137                app_id: AppId::new("app-a").unwrap(),
138                app_family: AppFamily::new("family-a").unwrap(),
139                sync_group: SyncGroup::new("cluster-a").unwrap(),
140                runtime_contract: RuntimeContractVersion::new(1),
141                node_id: NodeId::new(node).unwrap(),
142            },
143        }
144    }
145
146    fn message() -> SyncMessage {
147        SyncMessage {
148            batch_id: "batch-1".to_string(),
149            source_node_id: NodeId::new("node-a").unwrap(),
150            sequence_start: 1,
151            sequence_end: 1,
152            event_count: 1,
153            events_hash: "hash".to_string(),
154            created_at_ms: 10,
155            previous_batch_hash: None,
156            events: vec![b"event".to_vec()],
157        }
158    }
159
160    #[test]
161    fn v1_encoding_matches_golden_fixture() {
162        let encoded = encode_sync_envelope_v1(&identity("tenant-a", "node-a"), &message())
163            .expect("v1 envelope");
164
165        assert_eq!(encoded, include_str!("fixtures/sync-wire-v1.json").trim());
166        assert!(decode_sync_envelope(&encoded).is_ok());
167    }
168
169    #[test]
170    fn maximum_raw_batch_fits_the_bounded_http_envelope() {
171        let message = SyncMessage::new(
172            "batch-max".to_string(),
173            NodeId::new("node-a").unwrap(),
174            1,
175            1,
176            10,
177            None,
178            vec![vec![u8::MAX; MAX_SYNC_BATCH_PAYLOAD_BYTES]],
179        );
180        let encoded = encode_sync_envelope_v1(&identity("tenant-a", "node-a"), &message).unwrap();
181
182        assert!(encoded.len() <= MAX_SYNC_REQUEST_BODY_BYTES);
183    }
184
185    #[test]
186    fn borrowed_v1_encoding_matches_the_owned_contract() {
187        let source_identity = identity("tenant-a", "node-a");
188        let mut message = message();
189        message.events = vec!["Olá 日本語 العربية".as_bytes().to_vec(), vec![0, 1, 255]];
190        message.event_count = 2;
191        message.sequence_end = 2;
192        message.previous_batch_hash = Some("previous-hash".to_string());
193        let owned = SyncEnvelopeV1::new(source_identity.clone(), message.clone()).unwrap();
194
195        assert_eq!(
196            encode_sync_envelope_v1(&source_identity, &message).unwrap(),
197            serde_json::to_string(&owned).unwrap()
198        );
199    }
200
201    #[test]
202    fn v1_rejects_source_node_mismatch() {
203        assert!(matches!(
204            SyncEnvelopeV1::new(identity("tenant-a", "node-b"), message()),
205            Err(SyncError::InvalidSyncMessage(_))
206        ));
207        assert!(matches!(
208            encode_sync_envelope_v1(&identity("tenant-a", "node-b"), &message()),
209            Err(SyncError::InvalidSyncMessage(_))
210        ));
211    }
212
213    #[test]
214    fn v1_rejects_incompatible_tenant() {
215        let envelope = SyncEnvelopeV1::new(identity("tenant-a", "node-a"), message()).unwrap();
216
217        assert_eq!(
218            envelope.validate_for(&identity("tenant-b", "node-b")),
219            Err(SyncError::IncompatiblePeer)
220        );
221    }
222
223    #[test]
224    fn decoder_rejects_unversioned_wire_with_update_wall() {
225        assert_eq!(
226            decode_sync_envelope("batch-1\nnode-a\n1\n1\n"),
227            Err(SyncError::InvalidSyncMessage(UPDATE_REQUIRED_MESSAGE))
228        );
229    }
230}
appcore_sync/sync/wire.rs

appcore_sync/sync/
wire.rs
