appcore_sync/sync/
wire.rs

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

appcore_sync/sync/
wire.rs
