qualia_client_core/wellfair/
sync_outbox.rs

1//! Durable sync outbox: queued record commits awaiting upstream acknowledgement.
2
3use std::fs::{self, OpenOptions};
4use std::io::{BufRead, BufReader, Write};
5use std::path::{Path, PathBuf};
6
7use serde::{Deserialize, Serialize};
8use wellfare_core::conditions::journal_kind_for_record_id;
9use wellfare_core::record::RecordEnvelope;
10
11pub const SYNC_OUTBOX_FILE: &str = "wellfair/sync_outbox.jsonl";
12pub const MAX_LIST: usize = 256;
13
14#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
15#[serde(rename_all = "PascalCase")]
16pub enum SyncOutboxState {
17    Queued,
18    Sent,
19    Acknowledged,
20    Rejected,
21}
22
23#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
24pub struct SyncOutboxEntry {
25    pub operation_id: String,
26    pub record_id: String,
27    pub kind: String,
28    pub committed_unix: u32,
29    pub state: SyncOutboxState,
30}
31
32impl SyncOutboxEntry {
33    pub fn from_envelope(envelope: &RecordEnvelope, committed_unix: u32) -> Self {
34        Self {
35            operation_id: uuid::Uuid::new_v4().to_string(),
36            record_id: envelope.id.clone(),
37            kind: journal_kind_for_record_id(&envelope.id).to_string(),
38            committed_unix,
39            state: SyncOutboxState::Queued,
40        }
41    }
42}
43
44pub struct SyncOutbox {
45    path: PathBuf,
46}
47
48impl SyncOutbox {
49    pub fn open(storage_root: impl AsRef<Path>) -> std::io::Result<Self> {
50        let path = storage_root.as_ref().join(SYNC_OUTBOX_FILE);
51        if let Some(parent) = path.parent() {
52            fs::create_dir_all(parent)?;
53        }
54        if !path.exists() {
55            OpenOptions::new().create(true).write(true).open(&path)?;
56        }
57        Ok(Self { path })
58    }
59
60    pub fn enqueue(&self, entry: &SyncOutboxEntry) -> std::io::Result<()> {
61        let line =
62            serde_json::to_string(entry).map_err(|e| std::io::Error::other(e.to_string()))?;
63        let mut file = OpenOptions::new().append(true).open(&self.path)?;
64        writeln!(file, "{line}")?;
65        file.sync_all()?;
66        Ok(())
67    }
68
69    pub fn list_all(&self) -> std::io::Result<Vec<SyncOutboxEntry>> {
70        let file = fs::File::open(&self.path)?;
71        let reader = BufReader::new(file);
72        let mut entries = Vec::new();
73        for line in reader.lines() {
74            let line = line?;
75            if line.trim().is_empty() {
76                continue;
77            }
78            if let Ok(entry) = serde_json::from_str::<SyncOutboxEntry>(&line) {
79                entries.push(entry);
80            }
81        }
82        Ok(entries)
83    }
84
85    pub fn list_recent(&self, limit: usize) -> std::io::Result<Vec<SyncOutboxEntry>> {
86        let mut entries = self.list_all()?;
87        let keep = limit.min(MAX_LIST);
88        if entries.len() > keep {
89            entries.drain(0..entries.len() - keep);
90        }
91        entries.reverse();
92        Ok(entries)
93    }
94
95    pub fn count_queued(&self) -> std::io::Result<usize> {
96        Ok(self
97            .list_all()?
98            .into_iter()
99            .filter(|e| e.state == SyncOutboxState::Queued)
100            .count())
101    }
102
103    pub fn update_state(
104        &self,
105        operation_id: &str,
106        state: SyncOutboxState,
107    ) -> std::io::Result<bool> {
108        let all = self.list_all()?;
109        let mut found = false;
110        let mut rewritten = Vec::with_capacity(all.len());
111        for mut entry in all {
112            if entry.operation_id == operation_id {
113                entry.state = state;
114                found = true;
115            }
116            rewritten.push(entry);
117        }
118        if !found {
119            return Ok(false);
120        }
121        let tmp = self.path.with_extension("jsonl.tmp");
122        {
123            let mut file = OpenOptions::new()
124                .create(true)
125                .write(true)
126                .truncate(true)
127                .open(&tmp)?;
128            for entry in &rewritten {
129                let line = serde_json::to_string(entry)
130                    .map_err(|e| std::io::Error::other(e.to_string()))?;
131                writeln!(file, "{line}")?;
132            }
133            file.sync_all()?;
134        }
135        fs::rename(&tmp, &self.path)?;
136        Ok(true)
137    }
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143    use wellfare_core::record::{EpistemicStatus, EvidenceType, RecordEnvelope, SensitivityClass};
144
145    fn sample_envelope(id: &str) -> RecordEnvelope {
146        RecordEnvelope {
147            id: id.into(),
148            owner_did: "did:wf:owner".into(),
149            author_did: "did:wf:owner".into(),
150            proxy_did: None,
151            epistemic_status: EpistemicStatus::Asserted,
152            evidence_type: EvidenceType::DeviceMeasured,
153            sensitivity: SensitivityClass::Restricted,
154            asserted_time_unix: 1_700_000_000,
155            valid_time_start_unix: None,
156            valid_time_end_unix: None,
157            predecessor_id: None,
158            blob_hash: Some("deadbeef".into()),
159            tombstone: false,
160        }
161    }
162
163    #[test]
164    fn sync_outbox_enqueue_and_list() {
165        let dir = tempfile::tempdir().unwrap();
166        let outbox = SyncOutbox::open(dir.path()).unwrap();
167        let envelope = sample_envelope("urn:wellfair:weight:abc");
168        let entry = SyncOutboxEntry::from_envelope(&envelope, 1_700_000_100);
169        outbox.enqueue(&entry).unwrap();
170
171        let listed = outbox.list_recent(10).unwrap();
172        assert_eq!(listed.len(), 1);
173        assert_eq!(listed[0].record_id, "urn:wellfair:weight:abc");
174        assert_eq!(listed[0].kind, "weight");
175        assert_eq!(listed[0].committed_unix, 1_700_000_100);
176        assert_eq!(listed[0].state, SyncOutboxState::Queued);
177        assert!(!listed[0].operation_id.is_empty());
178    }
179
180    #[test]
181    fn sync_outbox_survives_reopen() {
182        let dir = tempfile::tempdir().unwrap();
183        let outbox = SyncOutbox::open(dir.path()).unwrap();
184        let envelope = sample_envelope("urn:wellfair:sleep:xyz");
185        let entry = SyncOutboxEntry::from_envelope(&envelope, 42);
186        outbox.enqueue(&entry).unwrap();
187
188        let reopened = SyncOutbox::open(dir.path()).unwrap();
189        let listed = reopened.list_recent(5).unwrap();
190        assert_eq!(listed.len(), 1);
191        assert_eq!(listed[0].kind, "sleep");
192        assert_eq!(reopened.count_queued().unwrap(), 1);
193    }
194
195    #[test]
196    fn sync_outbox_state_transition() {
197        let dir = tempfile::tempdir().unwrap();
198        let outbox = SyncOutbox::open(dir.path()).unwrap();
199        let envelope = sample_envelope("urn:wellfair:steps:1");
200        let entry = SyncOutboxEntry::from_envelope(&envelope, 99);
201        outbox.enqueue(&entry).unwrap();
202
203        assert!(outbox
204            .update_state(&entry.operation_id, SyncOutboxState::Sent)
205            .unwrap());
206        let listed = outbox.list_recent(1).unwrap();
207        assert_eq!(listed[0].state, SyncOutboxState::Sent);
208        assert_eq!(outbox.count_queued().unwrap(), 0);
209
210        assert!(outbox
211            .update_state(&entry.operation_id, SyncOutboxState::Acknowledged)
212            .unwrap());
213        assert_eq!(
214            outbox.list_recent(1).unwrap()[0].state,
215            SyncOutboxState::Acknowledged
216        );
217    }
218
219    #[test]
220    fn sync_outbox_serializes_pascal_case_state() {
221        let dir = tempfile::tempdir().unwrap();
222        let outbox = SyncOutbox::open(dir.path()).unwrap();
223        let entry =
224            SyncOutboxEntry::from_envelope(&sample_envelope("urn:wellfair:heart_rate:hr1"), 1);
225        outbox.enqueue(&entry).unwrap();
226
227        let raw = fs::read_to_string(dir.path().join(SYNC_OUTBOX_FILE)).unwrap();
228        assert!(raw.contains("\"state\":\"Queued\""));
229        assert!(raw.contains("\"kind\":\"heart_rate\""));
230    }
231}
qualia_client_core/wellfair/sync_outbox.rs

qualia_client_core/wellfair/
sync_outbox.rs
