spacedb_crdt/
persist.rs

1//! [`CrdtStore`] — persisting convergent documents into the encrypted store.
2//!
3//! This is where Layer 1 (CRDT) meets Layer 0 (storage). Each document's CRDT
4//! state is stored as the value of a row in an **encrypted** `spacedb-store`
5//! [`Collection`] keyed by document id — so the on-disk bytes are AEAD ciphertext
6//! the engine cannot read (the zero-knowledge posture holds for CRDT data too).
7//!
8//! The write path is the mission's loop made concrete: a local mutation applies
9//! to the in-memory [`CrdtDoc`], and [`CrdtStore::save`] persists the new state in
10//! **one M1 write transaction**. The incremental update to ship to peers comes
11//! from the doc's own [`CrdtDoc::encode_update_since`] — the anti-entropy
12//! primitive — and [`CrdtStore::apply_remote`] is the receive side
13//! (load → merge → re-persist).
14//!
15//! **S2 stores a full state snapshot per save.** That is simple and correct; the
16//! op-log + convergent compaction that avoids re-encoding the whole document on
17//! every write is M2-S3.
18
19use std::sync::Arc;
20
21use spacedb_store::{Collection, Durability, KeyProvider, KvEngine, WriteTx};
22
23use crate::{CrdtDoc, CrdtResult};
24
25/// The encrypted collection that holds `doc_id → CRDT-state` rows.
26pub const CRDT_DOCS_COLLECTION: &str = "crdt_docs";
27
28/// The schema version bound into each persisted row's AEAD (see `spacedb-store`).
29const SCHEMA_VERSION: u32 = 1;
30
31/// Persists [`CrdtDoc`]s into an encrypted `spacedb-store` collection. Engine-
32/// agnostic like the underlying [`Collection`]: the methods take the engine and
33/// open their own transactions.
34pub struct CrdtStore {
35    docs: Collection<String, Vec<u8>>,
36}
37
38impl CrdtStore {
39    /// Open (or first-time provision) the CRDT document collection on `engine`.
40    pub fn open<E: KvEngine>(engine: &E, key_provider: Arc<dyn KeyProvider>) -> CrdtResult<Self> {
41        let docs =
42            Collection::open_or_create(engine, key_provider, CRDT_DOCS_COLLECTION, SCHEMA_VERSION)?;
43        Ok(Self { docs })
44    }
45
46    /// Persist `doc`'s full CRDT state under `doc_id`, encrypted, in a single
47    /// write transaction.
48    pub fn save<E: KvEngine>(&self, engine: &E, doc_id: &str, doc: &CrdtDoc) -> CrdtResult<()> {
49        let state = doc.encode_full();
50        let mut w = engine.begin_write(Durability::Immediate)?;
51        self.docs.put(&mut w, &doc_id.to_string(), &state)?;
52        w.commit()?;
53        Ok(())
54    }
55
56    /// Load the document stored under `doc_id` for replica `actor_id`. Returns a
57    /// fresh empty document if nothing has been persisted there yet (local-first:
58    /// you can always start writing).
59    pub fn load<E: KvEngine>(
60        &self,
61        engine: &E,
62        doc_id: &str,
63        actor_id: u64,
64    ) -> CrdtResult<CrdtDoc> {
65        let stored = {
66            let r = engine.begin_read()?;
67            self.docs.get(&r, &doc_id.to_string())?
68        };
69        let doc = CrdtDoc::new(actor_id);
70        if let Some(state) = stored {
71            doc.apply_update(&state)?;
72        }
73        Ok(doc)
74    }
75
76    /// Receive side of anti-entropy: merge a remote `update` into the persisted
77    /// document and re-persist it (load → merge → save). Returns the merged doc.
78    pub fn apply_remote<E: KvEngine>(
79        &self,
80        engine: &E,
81        doc_id: &str,
82        actor_id: u64,
83        update: &[u8],
84    ) -> CrdtResult<CrdtDoc> {
85        let doc = self.load(engine, doc_id, actor_id)?;
86        doc.apply_update(update)?;
87        self.save(engine, doc_id, &doc)?;
88        Ok(doc)
89    }
90
91    /// Whether a document has been persisted under `doc_id`.
92    pub fn contains<E: KvEngine>(&self, engine: &E, doc_id: &str) -> CrdtResult<bool> {
93        let r = engine.begin_read()?;
94        Ok(self.docs.get(&r, &doc_id.to_string())?.is_some())
95    }
96}
spacedb_crdt/persist.rs

spacedb_crdt/
persist.rs
