qualia_client_core/wellfair/
vault.rs

1use std::path::{Path, PathBuf};
2use std::time::{SystemTime, UNIX_EPOCH};
3
4use ed25519_dalek::SigningKey;
5use qualia_core_db::crdt::SuspendedTransactionQueue;
6use qualia_core_db::git_bridge::DagStore;
7use qualia_core_db::wal::{commit_semantic_mutation, WalHandoffResult, WriteAheadLog};
8use qualia_core_db::NQuin;
9use wellfare_core::record::{RecordEnvelope, SensitivityClass};
10
11use super::checkpoint_store::{self, load_dag, load_meta};
12use super::consent_store::ConsentStore;
13use super::graph_store::GraphStore;
14use super::journal::{JournalEntry, WellfairJournal};
15use super::receipt::{ReceiptLog, ReceiptRecord};
16use super::sync_outbox::{SyncOutbox, SyncOutboxEntry};
17
18/// Coordinates transactions, graph materialization, and crash recovery.
19pub struct VaultService {
20    wal: WriteAheadLog,
21    dag: DagStore,
22    graph: GraphStore,
23    suspended: SuspendedTransactionQueue,
24    author_did: u64,
25    storage_root: PathBuf,
26    journal: WellfairJournal,
27    receipts: ReceiptLog,
28    consents: ConsentStore,
29    sync_outbox: SyncOutbox,
30    last_checkpoint_hash: Option<[u8; 32]>,
31}
32
33impl VaultService {
34    pub fn open<W: AsRef<Path>, S: AsRef<Path>>(
35        wal_path: W,
36        storage_root: S,
37        author_did: u64,
38    ) -> std::io::Result<Self> {
39        let storage_root = storage_root.as_ref().to_path_buf();
40        let wal = WriteAheadLog::open(wal_path)?;
41        let dag = load_dag(&storage_root);
42        let graph = GraphStore::open(&storage_root)?;
43        let suspended = SuspendedTransactionQueue::new();
44        let journal = WellfairJournal::open(&storage_root)?;
45        let receipts = ReceiptLog::open(&storage_root)?;
46        let consents = ConsentStore::open(&storage_root)?;
47        let sync_outbox = SyncOutbox::open(&storage_root)?;
48
49        let last_checkpoint_hash = if wal.prev_dag_hash != [0u8; 32] {
50            Some(wal.prev_dag_hash)
51        } else if let Some(meta) = load_meta(&storage_root) {
52            hex::decode(&meta.last_hash_hex)
53                .ok()
54                .and_then(|b| b.try_into().ok())
55        } else {
56            None
57        };
58
59        Ok(Self {
60            wal,
61            dag,
62            graph,
63            suspended,
64            author_did,
65            storage_root,
66            journal,
67            receipts,
68            consents,
69            sync_outbox,
70            last_checkpoint_hash,
71        })
72    }
73
74    pub fn graph_quin_count(&self) -> usize {
75        self.graph.count()
76    }
77
78    pub fn list_graph_quins(&self, limit: usize) -> std::io::Result<Vec<NQuin>> {
79        self.graph.list_recent(limit)
80    }
81
82    pub fn graph_coverage(
83        &self,
84        journal_limit: usize,
85    ) -> std::io::Result<Vec<super::graph_query::GraphCoverageRow>> {
86        let journal = self.journal.list_recent(journal_limit)?;
87        let quin_limit = journal.len().saturating_mul(8).max(64);
88        let quins = self.graph.list_recent(quin_limit)?;
89        Ok(super::graph_query::coverage_for_journal(&journal, &quins))
90    }
91
92    pub fn journal_count(&self) -> std::io::Result<usize> {
93        self.journal.count()
94    }
95
96    pub fn list_health_records(&self, limit: usize) -> std::io::Result<Vec<JournalEntry>> {
97        self.journal.list_recent(limit)
98    }
99
100    pub fn list_receipts(&self, limit: usize) -> std::io::Result<Vec<ReceiptRecord>> {
101        self.receipts.list_recent(limit)
102    }
103
104    pub fn list_outbox(&self, limit: usize) -> std::io::Result<Vec<SyncOutboxEntry>> {
105        self.sync_outbox.list_recent(limit)
106    }
107
108    pub fn outbox_queued_count(&self) -> std::io::Result<usize> {
109        self.sync_outbox.count_queued()
110    }
111
112    pub fn list_active_consents(
113        &self,
114        now_unix: u64,
115    ) -> std::io::Result<Vec<super::consent_store::ConsentGrantRecord>> {
116        self.consents.list_active(now_unix)
117    }
118
119    pub fn append_consent(
120        &self,
121        grant: &super::consent_store::ConsentGrantRecord,
122    ) -> std::io::Result<()> {
123        self.consents.append(grant)
124    }
125
126    pub fn revoke_consent(&self, grant_id: &str) -> std::io::Result<bool> {
127        self.consents.revoke(grant_id)
128    }
129
130    pub fn wal_buffered_quins(&mut self) -> std::io::Result<usize> {
131        self.wal.buffered_count()
132    }
133
134    /// Appends a semantic mutation to the durable WAL.
135    pub fn commit_quin(
136        &mut self,
137        mut quin: NQuin,
138        signing_key: &SigningKey,
139        principal_did_hash: u64,
140    ) -> std::io::Result<WalHandoffResult> {
141        commit_semantic_mutation(
142            &mut self.wal,
143            &mut quin,
144            principal_did_hash,
145            self.author_did,
146            signing_key,
147            &mut self.suspended,
148        )
149    }
150
151    /// Commits buffered WAL events into the content-addressed DAG, materializes graph, persists checkpoint.
152    pub fn checkpoint(&mut self) -> std::io::Result<[u8; 32]> {
153        let timestamp_ms = SystemTime::now()
154            .duration_since(UNIX_EPOCH)
155            .map(|d| d.as_millis() as u64)
156            .unwrap_or(0);
157        let hash = self
158            .wal
159            .checkpoint_to_dag(&mut self.dag, self.author_did, timestamp_ms)?;
160        let committed = self.wal.recover()?;
161        if !committed.is_empty() {
162            self.graph.append_quins(&committed)?;
163            checkpoint_store::persist_checkpoint(
164                &self.storage_root,
165                &self.dag,
166                hash,
167                self.graph.count(),
168                &committed,
169                self.author_did,
170            )?;
171        } else if self.dag.nodes().len() > 0 {
172            checkpoint_store::save_dag(&self.storage_root, &self.dag)?;
173        }
174        self.wal.truncate()?;
175        self.last_checkpoint_hash = Some(hash);
176        Ok(hash)
177    }
178
179    pub fn last_checkpoint_hash(&self) -> Option<[u8; 32]> {
180        self.last_checkpoint_hash
181    }
182
183    pub fn checkpoint_meta(&self) -> Option<checkpoint_store::CheckpointMeta> {
184        load_meta(&self.storage_root)
185    }
186
187    /// Converts a wellfare RecordEnvelope into NQuins and commits them.
188    pub fn commit_envelope(
189        &mut self,
190        envelope: &RecordEnvelope,
191        signing_key: &SigningKey,
192        principal_did_hash: u64,
193        source: &str,
194        summary: Option<String>,
195    ) -> std::io::Result<usize> {
196        let mut buffer = [wellfare_core::record::NQuin::default(); 8];
197        let count = envelope.compile_to_quins(&mut buffer);
198        let mut committed = 0;
199        for i in 0..count {
200            let src = &buffer[i];
201            let q = NQuin {
202                subject: src.subject,
203                predicate: src.predicate,
204                object: src.object,
205                context: src.context,
206                metadata: src.metadata,
207                parity: src.parity,
208            };
209            self.commit_quin(q, signing_key, principal_did_hash)?;
210            committed += 1;
211        }
212
213        let committed_unix = SystemTime::now()
214            .duration_since(UNIX_EPOCH)
215            .map(|d| d.as_secs() as u32)
216            .unwrap_or(0);
217        let entry = JournalEntry::from_envelope(envelope, source, committed_unix, summary);
218        self.journal.append(&entry)?;
219        // Sanctuary/Classified records are excluded from the ordinary sync outbox
220        // before an operation can enter it (master plan §5.2). Their existence,
221        // record id, and kind never leak onto the ordinary routing lane; a
222        // dedicated Sanctuary transport (future ADR) is the only permitted path.
223        if envelope.sensitivity != SensitivityClass::Classified {
224            let outbox_entry = SyncOutboxEntry::from_envelope(envelope, committed_unix);
225            self.sync_outbox.enqueue(&outbox_entry)?;
226        }
227        Ok(committed)
228    }
229
230    pub fn append_receipt(&self, receipt: &ReceiptRecord) -> std::io::Result<()> {
231        self.receipts.append(receipt)
232    }
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use ed25519_dalek::SigningKey;
239    use wellfare_core::record::{EpistemicStatus, EvidenceType, RecordEnvelope, SensitivityClass};
240
241    #[test]
242    fn checkpoint_persists_graph_and_meta() {
243        let dir = tempfile::tempdir().unwrap();
244        let wal_path = dir.path().join("test.wal");
245        let mut vault = VaultService::open(&wal_path, dir.path(), 0xBEEF).unwrap();
246        let signing_key = SigningKey::from_bytes(&[7u8; 32]);
247        let envelope = RecordEnvelope {
248            id: "urn:wellfair:weight:chk".into(),
249            owner_did: "did:wf:owner".into(),
250            author_did: "did:wf:owner".into(),
251            proxy_did: None,
252            epistemic_status: EpistemicStatus::Asserted,
253            evidence_type: EvidenceType::DeviceMeasured,
254            sensitivity: SensitivityClass::Restricted,
255            asserted_time_unix: 1_700_000_000,
256            valid_time_start_unix: None,
257            valid_time_end_unix: None,
258            predecessor_id: None,
259            blob_hash: Some("abc".into()),
260            tombstone: false,
261        };
262        vault
263            .commit_envelope(&envelope, &signing_key, 1, "test", None)
264            .unwrap();
265        let hash = vault.checkpoint().unwrap();
266        assert_ne!(hash, [0u8; 32]);
267        assert!(vault.graph_quin_count() > 0);
268        assert!(dir.path().join(checkpoint_store::META_FILE).exists());
269        assert!(dir.path().join(checkpoint_store::DAG_FILE).exists());
270
271        let wal_path2 = dir.path().join("test2.wal");
272        std::fs::copy(&wal_path, &wal_path2).unwrap();
273        let graph_before = vault.graph_quin_count();
274        let meta_before = vault.checkpoint_meta().unwrap();
275        let reopened = VaultService::open(&wal_path2, dir.path(), 0xBEEF).unwrap();
276        assert_eq!(reopened.graph_quin_count(), graph_before);
277        let meta_after = reopened.checkpoint_meta().unwrap();
278        assert_eq!(meta_after.graph_quin_count, meta_before.graph_quin_count);
279        assert_eq!(meta_after.last_hash_hex, meta_before.last_hash_hex);
280        #[cfg(not(target_arch = "wasm32"))]
281        assert!(dir.path().join(checkpoint_store::Q42_FILE).exists());
282    }
283
284    #[test]
285    fn classified_records_excluded_from_sync_outbox() {
286        let dir = tempfile::tempdir().unwrap();
287        let wal_path = dir.path().join("sanct.wal");
288        let mut vault = VaultService::open(&wal_path, dir.path(), 0xBEEF).unwrap();
289        let signing_key = SigningKey::from_bytes(&[3u8; 32]);
290
291        let restricted = RecordEnvelope {
292            id: "urn:wellfair:weight:r1".into(),
293            owner_did: "did:wf:owner".into(),
294            author_did: "did:wf:owner".into(),
295            proxy_did: None,
296            epistemic_status: EpistemicStatus::Asserted,
297            evidence_type: EvidenceType::DeviceMeasured,
298            sensitivity: SensitivityClass::Restricted,
299            asserted_time_unix: 1_700_000_000,
300            valid_time_start_unix: None,
301            valid_time_end_unix: None,
302            predecessor_id: None,
303            blob_hash: None,
304            tombstone: false,
305        };
306        let classified = RecordEnvelope {
307            id: "urn:wellfair:sanctuary_note:s1".into(),
308            owner_did: "did:wf:owner".into(),
309            author_did: "did:wf:owner".into(),
310            proxy_did: None,
311            epistemic_status: EpistemicStatus::Asserted,
312            evidence_type: EvidenceType::SelfReported,
313            sensitivity: SensitivityClass::Classified,
314            asserted_time_unix: 1_700_000_100,
315            valid_time_start_unix: None,
316            valid_time_end_unix: None,
317            predecessor_id: None,
318            blob_hash: None,
319            tombstone: false,
320        };
321
322        vault
323            .commit_envelope(&restricted, &signing_key, 1, "test", None)
324            .unwrap();
325        vault
326            .commit_envelope(&classified, &signing_key, 1, "test", None)
327            .unwrap();
328
329        // Both records are durably journaled...
330        assert_eq!(vault.journal_count().unwrap(), 2);
331        // ...but only the Restricted record reaches the ordinary sync outbox (§5.2).
332        let outbox = vault.list_outbox(16).unwrap();
333        assert_eq!(outbox.len(), 1);
334        assert_eq!(outbox[0].record_id, "urn:wellfair:weight:r1");
335        assert!(outbox
336            .iter()
337            .all(|e| e.record_id != "urn:wellfair:sanctuary_note:s1"));
338    }
339}
qualia_client_core/wellfair/vault.rs

qualia_client_core/wellfair/
vault.rs
