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
31impl SyncEnvelopeV1 {
32    /// Creates a v1 envelope after checking the message/source node binding.
33    pub fn new(source_identity: CoreIdentity, message: SyncMessage) -> SyncResult<Self> {
34        if source_identity.runtime.node_id != message.source_node_id {
35            return Err(SyncError::InvalidSyncMessage(
36                "source identity does not match message node",
37            ));
38        }
39        Ok(Self {
40            schema: SYNC_WIRE_SCHEMA_V1.to_string(),
41            source_identity,
42            message,
43        })
44    }
45
46    /// Verifies schema, source binding and compatibility with the receiver.
47    pub fn validate_for(&self, local_identity: &CoreIdentity) -> SyncResult<()> {
48        if self.schema != SYNC_WIRE_SCHEMA_V1 {
49            return Err(SyncError::InvalidSyncMessage(
50                "unsupported sync wire schema",
51            ));
52        }
53        if self.source_identity.runtime.node_id != self.message.source_node_id {
54            return Err(SyncError::InvalidSyncMessage(
55                "source identity does not match message node",
56            ));
57        }
58        let policy = CoreCompatibilityPolicy {
59            require_same_cluster: true,
60            required_capability: None,
61        };
62        local_identity
63            .ensure_compatible(&self.source_identity, &policy, &[])
64            .map_err(|_| SyncError::IncompatiblePeer)
65    }
66}
67
68/// Encodes an identity-aware v1 envelope as JSON.
69pub fn encode_sync_envelope_v1(
70    source_identity: &CoreIdentity,
71    message: &SyncMessage,
72) -> SyncResult<String> {
73    let envelope = SyncEnvelopeV1::new(source_identity.clone(), message.clone())?;
74    serde_json::to_string(&envelope)
75        .map_err(|_| SyncError::InvalidSyncMessage("sync wire serialization failed"))
76}
77
78/// Decodes the identity-aware v1 JSON envelope.
79pub fn decode_sync_envelope(input: &str) -> SyncResult<SyncEnvelopeV1> {
80    if input.is_empty() {
81        return Err(SyncError::EmptyRequestBody);
82    }
83    if !input.trim_start().starts_with('{') {
84        return Err(SyncError::InvalidSyncMessage(UPDATE_REQUIRED_MESSAGE));
85    }
86    let envelope = serde_json::from_str::<SyncEnvelopeV1>(input)
87        .map_err(|_| SyncError::InvalidSyncMessage("invalid sync wire envelope"))?;
88    if envelope.schema != SYNC_WIRE_SCHEMA_V1 {
89        return Err(SyncError::InvalidSyncMessage(UPDATE_REQUIRED_MESSAGE));
90    }
91    if envelope.source_identity.runtime.node_id != envelope.message.source_node_id {
92        return Err(SyncError::InvalidSyncMessage(
93            "source identity does not match message node",
94        ));
95    }
96    Ok(envelope)
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102    use appcore_core::{
103        AppFamily, AppId, ClusterId, CoreId, CoreKind, InstanceId, NodeId, ProtocolVersion,
104        RuntimeContractVersion, RuntimeIdentity, SyncGroup, TenantId,
105    };
106
107    fn identity(tenant: &str, node: &str) -> CoreIdentity {
108        CoreIdentity {
109            tenant_id: TenantId::new(tenant).unwrap(),
110            cluster_id: ClusterId::new("cluster-a").unwrap(),
111            core_id: CoreId::new(format!("core-{node}")).unwrap(),
112            instance_id: InstanceId::new(format!("instance-{node}")).unwrap(),
113            kind: CoreKind::new("replica").unwrap(),
114            protocol_version: ProtocolVersion::new(1),
115            runtime: RuntimeIdentity {
116                app_id: AppId::new("app-a").unwrap(),
117                app_family: AppFamily::new("family-a").unwrap(),
118                sync_group: SyncGroup::new("cluster-a").unwrap(),
119                runtime_contract: RuntimeContractVersion::new(1),
120                node_id: NodeId::new(node).unwrap(),
121            },
122        }
123    }
124
125    fn message() -> SyncMessage {
126        SyncMessage {
127            batch_id: "batch-1".to_string(),
128            source_node_id: NodeId::new("node-a").unwrap(),
129            sequence_start: 1,
130            sequence_end: 1,
131            event_count: 1,
132            events_hash: "hash".to_string(),
133            created_at_ms: 10,
134            previous_batch_hash: None,
135            events: vec![b"event".to_vec()],
136        }
137    }
138
139    #[test]
140    fn v1_encoding_matches_golden_fixture() {
141        let encoded = encode_sync_envelope_v1(&identity("tenant-a", "node-a"), &message())
142            .expect("v1 envelope");
143
144        assert_eq!(encoded, include_str!("fixtures/sync-wire-v1.json").trim());
145        assert!(decode_sync_envelope(&encoded).is_ok());
146    }
147
148    #[test]
149    fn v1_rejects_source_node_mismatch() {
150        assert!(matches!(
151            SyncEnvelopeV1::new(identity("tenant-a", "node-b"), message()),
152            Err(SyncError::InvalidSyncMessage(_))
153        ));
154    }
155
156    #[test]
157    fn v1_rejects_incompatible_tenant() {
158        let envelope = SyncEnvelopeV1::new(identity("tenant-a", "node-a"), message()).unwrap();
159
160        assert_eq!(
161            envelope.validate_for(&identity("tenant-b", "node-b")),
162            Err(SyncError::IncompatiblePeer)
163        );
164    }
165
166    #[test]
167    fn decoder_rejects_unversioned_wire_with_update_wall() {
168        assert_eq!(
169            decode_sync_envelope("batch-1\nnode-a\n1\n1\n"),
170            Err(SyncError::InvalidSyncMessage(UPDATE_REQUIRED_MESSAGE))
171        );
172    }
173}
appcore_sync/sync/wire.rs

appcore_sync/sync/
wire.rs
