qualia_client_core/wellfair/api/
sync.rs

1//! Sync-operation protocol + transport
2
3use super::super::journal::JournalEntry;
4use super::super::sync_outbox::{SyncOutbox, SyncOutboxEntry, SyncOutboxState};
5use super::super::sync_protocol::{AdmitOutcome, InboxRecord, SyncInbox, SyncOperation};
6use super::super::sync_transport::SyncTransport;
7use ed25519_dalek::Signer;
8
9use super::*;
10
11impl WebizenHostApi {
12    // --- Phase 5 sync-operation protocol (SyncService, §4.2 / §9.5 / §17) ---
13
14    /// Build a signed outbound sync operation from a committed journal entry.
15    /// Returns `None` for Classified/Sanctuary records — they never enter the ordinary sync
16    /// lane (§5.2). The signature is a real ed25519 signature over the operation's bound payload.
17    pub fn build_outbound_operation(
18        &self,
19        entry: &JournalEntry,
20        lamport: u64,
21    ) -> Option<SyncOperation> {
22        if entry.sensitivity == "Classified" {
23            return None;
24        }
25        let op = SyncOperation::new(
26            uuid::Uuid::new_v4().to_string(),
27            entry.id.clone(),
28            entry.kind.clone(),
29            self.author_did.clone(),
30            entry.sensitivity.clone(),
31            entry.summary.clone().unwrap_or_default(),
32            lamport,
33            entry.committed_unix,
34        );
35        let signature = self.signing_key.sign(&op.signing_payload());
36        Some(op.with_signature(hex::encode(signature.to_bytes())))
37    }
38
39    /// Admit an inbound sync operation into the durable quarantined inbox. Idempotent: a
40    /// replayed operation id is recorded as `Duplicate` and never applied twice.
41    pub fn admit_sync_operation(&self, op: &SyncOperation) -> Result<AdmitOutcome, String> {
42        let inbox = SyncInbox::open(&self.storage_root).map_err(|e| e.to_string())?;
43        inbox
44            .admit(op, Self::now_unix() as u32)
45            .map_err(|e| e.to_string())
46    }
47
48    /// Validated operations currently held in the inbox, in Lamport order.
49    pub fn validated_sync_operations(&self) -> Result<Vec<SyncOperation>, String> {
50        SyncInbox::open(&self.storage_root)
51            .map_err(|e| e.to_string())?
52            .validated_operations()
53            .map_err(|e| e.to_string())
54    }
55
56    pub fn list_sync_inbox(&self, limit: usize) -> Result<Vec<InboxRecord>, String> {
57        SyncInbox::open(&self.storage_root)
58            .map_err(|e| e.to_string())?
59            .list_recent(limit)
60            .map_err(|e| e.to_string())
61    }
62
63    // --- Sync transport orchestration (T3.1: drain outbox → transport → peer inbox) ---
64
65    /// The next Lamport value to stamp on outbound operations: one past the greatest observed among
66    /// the locally-validated inbox operations. Keeps outbound clocks causally ahead of what we've seen.
67    fn next_sync_lamport(&self) -> Result<u64, String> {
68        let max = self
69            .validated_sync_operations()?
70            .iter()
71            .map(|o| o.lamport)
72            .max()
73            .unwrap_or(0);
74        Ok(max + 1)
75    }
76
77    /// **Drain the outbox through a transport.** For each `Queued` outbox entry, build a signed
78    /// [`SyncOperation`] from its committed journal entry and publish it; on success the entry is
79    /// marked `Sent`. Classified/Sanctuary records never enter the ordinary lane — they are marked
80    /// `Rejected` so they stop being retried. Returns the number of operations published.
81    ///
82    /// The transport is a dumb pipe; correctness (dedup, convergence) is enforced by the peer's
83    /// fail-closed inbox on the other side.
84    pub fn sync_push_via<T: SyncTransport>(
85        &self,
86        transport: &T,
87        limit: usize,
88    ) -> Result<usize, String> {
89        let outbox = SyncOutbox::open(&self.storage_root).map_err(|e| e.to_string())?;
90        let queued: Vec<SyncOutboxEntry> = outbox
91            .list_all()
92            .map_err(|e| e.to_string())?
93            .into_iter()
94            .filter(|e| e.state == SyncOutboxState::Queued)
95            .take(limit)
96            .collect();
97        if queued.is_empty() {
98            return Ok(0);
99        }
100        let journal = self.list_health_records(512)?;
101        let mut lamport = self.next_sync_lamport()?;
102        let mut ops = Vec::new();
103        let mut sent_ids = Vec::new();
104        for entry in &queued {
105            let Some(journal_entry) = journal.iter().find(|j| j.id == entry.record_id) else {
106                continue; // the record is outside the recent window; leave it queued for a later drain
107            };
108            match self.build_outbound_operation(journal_entry, lamport) {
109                Some(op) => {
110                    lamport += 1;
111                    ops.push(op);
112                    sent_ids.push(entry.operation_id.clone());
113                }
114                None => {
115                    // Classified/Sanctuary — never syncs; stop retrying it.
116                    let _ = outbox.update_state(&entry.operation_id, SyncOutboxState::Rejected);
117                }
118            }
119        }
120        if ops.is_empty() {
121            return Ok(0);
122        }
123        transport.publish(&ops)?;
124        for id in &sent_ids {
125            let _ = outbox.update_state(id, SyncOutboxState::Sent);
126        }
127        Ok(ops.len())
128    }
129
130    /// **Pull from a transport and admit into the quarantined inbox.** Every op is validated
131    /// fail-closed on admission (bad signature/hash/version/oversize/Classified → `Rejected`;
132    /// replays → `Duplicate`), so a hostile peer can only cause rejections. Returns the admission
133    /// tally.
134    pub fn sync_pull_via<T: SyncTransport>(
135        &self,
136        transport: &T,
137        since: u64,
138    ) -> Result<SyncPullReport, String> {
139        let ops = transport.pull(since)?;
140        let mut report = SyncPullReport {
141            pulled: ops.len(),
142            validated: 0,
143            duplicate: 0,
144            rejected: 0,
145        };
146        for op in &ops {
147            match self.admit_sync_operation(op)? {
148                AdmitOutcome::Validated => report.validated += 1,
149                AdmitOutcome::Duplicate => report.duplicate += 1,
150                AdmitOutcome::Rejected(_) => report.rejected += 1,
151            }
152        }
153        Ok(report)
154    }
155
156    /// One-shot sync against an HTTP relay (the production wire): drain the outbox to the relay,
157    /// then pull + admit from it. Returns `(pushed, pull_report)`. Native-only (`reqwest`).
158    #[cfg(not(target_arch = "wasm32"))]
159    pub fn sync_with_http_relay(
160        &self,
161        base_url: &str,
162        since: u64,
163    ) -> Result<(usize, SyncPullReport), String> {
164        let transport = super::super::sync_transport::HttpRelayTransport::new(base_url);
165        let pushed = self.sync_push_via(&transport, 256)?;
166        let report = self.sync_pull_via(&transport, since)?;
167        Ok((pushed, report))
168    }
169
170    /// One-shot sync against a **libp2p** peer/relay (noise-encrypted request-response — the peer-to-peer
171    /// wire): drain the outbox to the peer, then pull + admit from it. `peer_id` is the base58 peer id,
172    /// `peer_addr` a libp2p multiaddr (e.g. `/ip4/1.2.3.4/tcp/4001`). Returns `(pushed, pull_report)`.
173    /// Native-only (libp2p). Same dumb-pipe contract as [`Self::sync_with_http_relay`]: correctness is
174    /// enforced by the fail-closed inbox, not the transport.
175    #[cfg(not(target_arch = "wasm32"))]
176    pub fn sync_with_libp2p_peer(
177        &self,
178        peer_id: &str,
179        peer_addr: &str,
180        since: u64,
181    ) -> Result<(usize, SyncPullReport), String> {
182        let transport =
183            super::super::sync_transport::Libp2pSyncTransport::connect(peer_id, peer_addr)?;
184        let pushed = self.sync_push_via(&transport, 256)?;
185        let report = self.sync_pull_via(&transport, since)?;
186        Ok((pushed, report))
187    }
188}
qualia_client_core/wellfair/api/sync.rs

qualia_client_core/wellfair/api/
sync.rs
