spacedb_sdk/
db.rs

1//! The [`Database`] — the offline-first local replica a developer holds.
2//!
3//! It owns the local CRDT documents (one per collection), a strong-tier quorum, an
4//! mID key directory + revocation set, and a rate card. Every mutating op runs the
5//! same spine: **authorize** (mID), **charge** (budget), **route by tier**, and
6//! **return the honest [`Outcome`]** of what was actually achieved. Reads are local
7//! and always available; sync is explicit CRDT merge, so the whole thing works with
8//! no network at all.
9
10use std::collections::hash_map::DefaultHasher;
11use std::collections::HashMap;
12use std::hash::{Hash, Hasher};
13
14use spacedb_access::{
15    authorize, AccessRequest, Decision, Identity, MemKeyDirectory, Ops, RevocationSet, Scope,
16    SignedCapability,
17};
18use spacedb_consistency::{Outcome, QuorumGroup, StrongResult, Tier};
19use spacedb_crdt::{CrdtDoc, CrdtError, Watcher};
20use spacedb_meter::{MeterError, RateCard, Usage};
21
22use crate::error::{SdkError, SdkResult};
23use crate::schema::{CrdtType, FieldSpec, Schema};
24use crate::session::Session;
25
26/// Fuel charged per mutating op (a flat, deterministic write cost for the demo;
27/// a real deployment would meter actual work).
28const WRITE_FUEL: u64 = 1_000;
29
30/// An offline-first SpaceDB replica.
31pub struct Database {
32    node: Identity,
33    actor_id: u64,
34    directory: MemKeyDirectory,
35    revocations: RevocationSet,
36    rate_card: RateCard,
37    clock: u64,
38    schemas: HashMap<String, Schema>,
39    docs: HashMap<String, CrdtDoc>,
40    quorum: QuorumGroup,
41}
42
43impl Database {
44    /// Open a local replica for `node`. In-memory and offline by construction;
45    /// durability via `spacedb-store` and a real placed quorum are wiring the host
46    /// supplies.
47    pub fn open(node: Identity) -> Self {
48        let actor_id = actor_id_for(&node.did().0);
49        Self {
50            node,
51            actor_id,
52            directory: MemKeyDirectory::new(),
53            revocations: RevocationSet::new(),
54            rate_card: default_rate_card(),
55            clock: 0,
56            schemas: HashMap::new(),
57            docs: HashMap::new(),
58            // a self-hosted 3-member strong-tier quorum (placement is the M4 seam)
59            quorum: QuorumGroup::new(["q0", "q1", "q2"]),
60        }
61    }
62
63    /// This replica's mID.
64    pub fn node(&self) -> &Identity {
65        &self.node
66    }
67
68    /// Publish an identity's key so capabilities it issues can be verified.
69    pub fn register_identity(&self, identity: &Identity) -> SdkResult<()> {
70        self.directory
71            .publish(identity)
72            .map_err(|e| SdkError::Auth(e.to_string()))
73    }
74
75    /// Register a collection's schema.
76    pub fn define(&mut self, schema: Schema) {
77        self.schemas.insert(schema.collection().to_string(), schema);
78    }
79
80    /// Set the logical clock used for capability expiry checks.
81    pub fn set_clock(&mut self, now_unix: u64) {
82        self.clock = now_unix;
83    }
84
85    /// The micro-`$MATA` cost of one mutating op under the current rate card.
86    pub fn write_cost(&self) -> u64 {
87        self.rate_card.price(&Usage::compute(WRITE_FUEL, 1))
88    }
89
90    /// Begin a session for the holder of `capability`.
91    pub fn session(&self, capability: SignedCapability) -> Session {
92        Session::from_capability(capability)
93    }
94
95    /// Revoke a capability by id (mID kill-switch).
96    pub fn revoke(&mut self, capability_id: [u8; 16]) {
97        self.revocations.revoke(capability_id);
98    }
99
100    // ── convergent / causal writes ──────────────────────────────────────────
101
102    /// Set a `Register` field. Returns the honest [`Outcome`] (`Local` for
103    /// convergent/causal — durable here, converging outward).
104    pub fn put_register(
105        &mut self,
106        session: &mut Session,
107        collection: &str,
108        field: &str,
109        value: &str,
110    ) -> SdkResult<Outcome> {
111        let spec = self.require_field(collection, field, CrdtType::Register)?;
112        let tier = self.begin_write(session, collection, spec)?;
113        let doc = self.doc_mut(collection);
114        doc.set_register(field, &value.to_string()).map_err(crdt_err)?;
115        Ok(local_outcome(tier, session, doc))
116    }
117
118    /// Add `delta` to a `Counter` field.
119    pub fn increment(
120        &mut self,
121        session: &mut Session,
122        collection: &str,
123        field: &str,
124        delta: i64,
125    ) -> SdkResult<Outcome> {
126        let spec = self.require_field(collection, field, CrdtType::Counter)?;
127        let tier = self.begin_write(session, collection, spec)?;
128        let doc = self.doc_mut(collection);
129        doc.increment(field, delta);
130        Ok(local_outcome(tier, session, doc))
131    }
132
133    /// Append to a `Text` field.
134    pub fn append_text(
135        &mut self,
136        session: &mut Session,
137        collection: &str,
138        field: &str,
139        text: &str,
140    ) -> SdkResult<Outcome> {
141        let spec = self.require_field(collection, field, CrdtType::Text)?;
142        let tier = self.begin_write(session, collection, spec)?;
143        let doc = self.doc_mut(collection);
144        doc.text_push(field, text);
145        Ok(local_outcome(tier, session, doc))
146    }
147
148    /// Add an element to a `Set` field.
149    pub fn add_to_set(
150        &mut self,
151        session: &mut Session,
152        collection: &str,
153        field: &str,
154        element: &str,
155    ) -> SdkResult<Outcome> {
156        let spec = self.require_field(collection, field, CrdtType::Set)?;
157        let tier = self.begin_write(session, collection, spec)?;
158        let doc = self.doc_mut(collection);
159        doc.set_add(field, element);
160        Ok(local_outcome(tier, session, doc))
161    }
162
163    // ── strong writes ───────────────────────────────────────────────────────
164
165    /// Claim a globally-unique `value` for a strong-tier field (e.g. a username).
166    /// Returns the quorum's honest [`StrongResult`]: `Committed`, `Rejected`
167    /// (already taken), or `Unavailable` (no quorum — fails safe, commits nothing).
168    pub fn claim_unique(
169        &mut self,
170        session: &mut Session,
171        collection: &str,
172        field: &str,
173        value: &str,
174    ) -> SdkResult<StrongResult> {
175        let spec = self.require_field(collection, field, CrdtType::Register)?;
176        if spec.tier != Tier::Strong {
177            return Err(SdkError::WrongType {
178                field: field.to_string(),
179                expected: CrdtType::Register,
180                found: spec.crdt,
181            });
182        }
183        self.authorize_op(session, collection, Ops::WRITE)?;
184        self.charge(session)?;
185        let key = format!("{collection}/{field}/{value}");
186        let owner = session.actor.0.as_bytes().to_vec();
187        Ok(self.quorum.claim_unique(&key, &owner))
188    }
189
190    /// Who owns a claimed unique `value`, if anyone (a quorum read).
191    pub fn unique_owner(
192        &self,
193        collection: &str,
194        field: &str,
195        value: &str,
196    ) -> Option<String> {
197        let key = format!("{collection}/{field}/{value}");
198        match self.quorum.read(&key) {
199            Ok((Some(bytes), _)) => Some(String::from_utf8_lossy(&bytes).into_owned()),
200            _ => None,
201        }
202    }
203
204    /// Partition a strong-tier quorum member (testing / chaos).
205    pub fn quorum_partition(&mut self, member: &str) -> bool {
206        self.quorum.partition(member)
207    }
208
209    /// Heal a strong-tier quorum member.
210    pub fn quorum_heal(&mut self, member: &str) -> bool {
211        self.quorum.heal(member)
212    }
213
214    // ── reads ───────────────────────────────────────────────────────────────
215
216    /// Read a `Register`, with the honest consistency [`Outcome`] of the read.
217    /// Requires read authorization.
218    pub fn read_register(
219        &self,
220        session: &mut Session,
221        collection: &str,
222        field: &str,
223    ) -> SdkResult<(Option<String>, Outcome)> {
224        let spec = self.require_field(collection, field, CrdtType::Register)?;
225        self.authorize_op(session, collection, Ops::READ)?;
226        let doc = self.docs.get(collection);
227        let value = match doc {
228            Some(d) => d.get_register::<String>(field).map_err(crdt_err)?,
229            None => None,
230        };
231        let outcome = match (spec.tier, doc) {
232            (Tier::Causal, Some(d)) => session.causal.read(d),
233            (Tier::Causal, None) => Outcome::Committed(Tier::Causal),
234            (tier, _) => Outcome::Committed(tier),
235        };
236        Ok((value, outcome))
237    }
238
239    /// Local (always-available) reads — the convergent view on this replica.
240    pub fn counter(&self, collection: &str, field: &str) -> i64 {
241        self.docs.get(collection).map_or(0, |d| d.counter(field))
242    }
243
244    pub fn text(&self, collection: &str, field: &str) -> String {
245        self.docs.get(collection).map_or_else(String::new, |d| d.text(field))
246    }
247
248    pub fn set_members(&self, collection: &str, field: &str) -> Vec<String> {
249        self.docs
250            .get(collection)
251            .map_or_else(Vec::new, |d| d.set_members(field))
252    }
253
254    // ── reactive + sync ─────────────────────────────────────────────────────
255
256    /// A reactive watcher for a collection: `drain_changed()` returns true after
257    /// any local or merged change.
258    pub fn watch(&mut self, collection: &str) -> Watcher {
259        self.doc_mut(collection).watch()
260    }
261
262    /// Export this replica's state for a collection (offline sync).
263    pub fn export(&self, collection: &str) -> Vec<u8> {
264        self.docs
265            .get(collection)
266            .map_or_else(Vec::new, |d| d.encode_full())
267    }
268
269    /// Merge another replica's exported state into this one (convergent).
270    pub fn import(&mut self, collection: &str, update: &[u8]) -> SdkResult<()> {
271        self.doc_mut(collection).apply_update(update).map_err(crdt_err)
272    }
273
274    // ── internals ───────────────────────────────────────────────────────────
275
276    fn require_field(
277        &self,
278        collection: &str,
279        field: &str,
280        expected: CrdtType,
281    ) -> SdkResult<FieldSpec> {
282        let schema = self
283            .schemas
284            .get(collection)
285            .ok_or_else(|| SdkError::UnknownCollection(collection.to_string()))?;
286        let spec = schema.spec(field).ok_or_else(|| SdkError::UnknownField {
287            collection: collection.to_string(),
288            field: field.to_string(),
289        })?;
290        if spec.crdt != expected {
291            return Err(SdkError::WrongType {
292                field: field.to_string(),
293                expected,
294                found: spec.crdt,
295            });
296        }
297        Ok(spec)
298    }
299
300    /// The shared write prologue: reject strong puts, authorize, then charge.
301    /// Returns the field's tier for outcome routing.
302    fn begin_write(
303        &self,
304        session: &mut Session,
305        collection: &str,
306        spec: FieldSpec,
307    ) -> SdkResult<Tier> {
308        if spec.tier == Tier::Strong {
309            return Err(SdkError::StrongFieldNeedsClaim(collection.to_string()));
310        }
311        self.authorize_op(session, collection, Ops::WRITE)?;
312        self.charge(session)?;
313        Ok(spec.tier)
314    }
315
316    fn authorize_op(&self, session: &Session, collection: &str, op: Ops) -> SdkResult<()> {
317        let scope = Scope::Collection(collection.to_string());
318        let request = AccessRequest {
319            bearer: &session.actor,
320            scope: &scope,
321            op,
322        };
323        match authorize(
324            &session.capability,
325            &request,
326            &self.directory,
327            self.clock,
328            &self.revocations,
329        ) {
330            Ok(Decision::Allow) => Ok(()),
331            Ok(Decision::Deny(reason)) => Err(SdkError::Denied(reason)),
332            Err(e) => Err(SdkError::Auth(e.to_string())),
333        }
334    }
335
336    fn charge(&self, session: &mut Session) -> SdkResult<()> {
337        let cost = self.write_cost();
338        session.budget.charge(cost).map_err(|e| match e {
339            MeterError::OverBudget { cost, remaining } => SdkError::OverBudget { cost, remaining },
340            other => SdkError::Auth(other.to_string()),
341        })
342    }
343
344    fn doc_mut(&mut self, collection: &str) -> &CrdtDoc {
345        let actor = self.actor_id;
346        self.docs
347            .entry(collection.to_string())
348            .or_insert_with(|| CrdtDoc::new(actor))
349    }
350}
351
352/// Convergent/causal writes are locally durable (`Local`); a causal field also
353/// advances the session's causal token so it reads its own write.
354fn local_outcome(tier: Tier, session: &mut Session, doc: &CrdtDoc) -> Outcome {
355    match tier {
356        Tier::Causal => session.causal.record_write(doc),
357        _ => Outcome::Local,
358    }
359}
360
361fn actor_id_for(did: &str) -> u64 {
362    let mut hasher = DefaultHasher::new();
363    did.hash(&mut hasher);
364    hasher.finish() | 1 // never 0
365}
366
367fn default_rate_card() -> RateCard {
368    RateCard {
369        storage_per_gib_month: 5_000_000,
370        compute_per_megafuel: 1_000_000,
371        compute_per_invocation: 1_000,
372        transit_per_gib: 1_000_000,
373    }
374}
375
376fn crdt_err(e: CrdtError) -> SdkError {
377    SdkError::Crdt(e.to_string())
378}
spacedb_sdk/db.rs

spacedb_sdk/
db.rs
