diff --git a/CHANGELOG.md b/CHANGELOG.md index ed0f4e5607..c4d0f8828e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,15 @@ See [STATUS.md](server/STATUS.md) to learn more about which features will remain a store seeded by an older one. Existing values, including user edits to default resources, are never overwritten. `--repopulate-defaults` remains as a forced re-run with the same add-only semantics. +- `atomic_lib`: `atomic_lib::runtime::AtomicNode` — a named node surface over + `Db` (`open`, `get`, `query`, `apply_commit(json, IngestPolicy)`, `mutate`, + `subscribe`, `sync_with_peer`). Thin delegation, no behaviour change; the + WASM `ClientDb` is the first adapter on it. `IngestPolicy::{Hub, Peer, + Replica, LocalCache}` names the four commit-validation profiles. + `sync::engine::ingest_commit` returns the `CommitResponse` that + `ingest_commit_json` serializes; `sync::ws_apply::apply_commit_json` now + returns the `CommitResponse` instead of `()`. See + `planning/runtime-boundary-decision.md`. ## [v0.41.0-beta.2] - 2026-08-01 diff --git a/TESTING_COVERAGE.md b/TESTING_COVERAGE.md index fe8eacbc38..b206d66533 100644 --- a/TESTING_COVERAGE.md +++ b/TESTING_COVERAGE.md @@ -121,6 +121,7 @@ Both matter because `iroh_transport` holds the router and node identity in | **`POST /iroh-sync` request shape, both sides** | `testdata/pairing-request.json` + `pairing.test.ts` + `iroh_pairing.rs` | shared fixture binds them | | Dart pairing-code parser, peer-sync result formatting | `flutter/test/atomic/` | pure parsers | | Rotation does not treat a metrics-change pop as "back to gallery" | `flutter/test/canvas/rotation_pop_test.dart` | | +| `AtomicNode`: `mutate` on one node, `apply_commit(IngestPolicy::Peer)` on another, query + `DbEvent` reflect it | `lib/src/runtime/node.rs` | in-process, no transport; `LocalCache` skips signature check, `Peer` does not | ### Flow — the thin layer diff --git a/lib/src/lib.rs b/lib/src/lib.rs index 783488eead..f313584d78 100644 --- a/lib/src/lib.rs +++ b/lib/src/lib.rs @@ -95,6 +95,9 @@ pub mod plugins; pub mod populate; pub mod resources; +/// The node runtime boundary (`AtomicNode`). Wraps `Db`, so it needs `db`. +#[cfg(feature = "db")] +pub mod runtime; pub mod schema; pub mod serialize; pub mod store; diff --git a/lib/src/runtime/mod.rs b/lib/src/runtime/mod.rs new file mode 100644 index 0000000000..d5d88174f4 --- /dev/null +++ b/lib/src/runtime/mod.rs @@ -0,0 +1,11 @@ +//! The node runtime boundary: [`AtomicNode`] is the one surface adapters +//! (HTTP, WebSocket, Iroh, WASM, FFI, Flutter) bind to instead of wrapping +//! [`crate::Db`] themselves. +//! +//! Slice 1 (`planning/atomic-lib-runtime.md`, `planning/runtime-boundary-decision.md`) +//! is a thin wrapper: every method delegates to code that already existed, so +//! there is no behaviour change — only a named place for it. + +mod node; + +pub use node::{AtomicNode, IngestPolicy, NodeConfig, NodeStorage, ResourceEdit}; diff --git a/lib/src/runtime/node.rs b/lib/src/runtime/node.rs new file mode 100644 index 0000000000..cbd06ec289 --- /dev/null +++ b/lib/src/runtime/node.rs @@ -0,0 +1,477 @@ +//! [`AtomicNode`]: a thin, named surface over [`Db`]. +//! +//! Every method here delegates to a function that already existed before the +//! runtime module did (the doc comment on each one names it). The point of +//! this slice is not new behaviour but one place for adapters to bind, so +//! that `wasm/src/lib.rs`, `flutter/rust/src/api/simple.rs`, `ffi/`, +//! `python/` and the Actix handlers stop each re-wrapping `Db` with their own +//! copy of the commit-validation knobs. + +use tokio::sync::broadcast; + +use crate::{ + agents::{Agent, ForAgent}, + commit::CommitResponse, + db::{Db, DbEvent}, + errors::AtomicResult, + storelike::{Query, QueryResult, ResourceResponse}, + sync::engine::{ingest_commit, CommitIngestOpts}, + Resource, Storelike, Subject, +}; + +/// Where a node keeps its data. Each variant maps to exactly one existing +/// `Db` constructor. +#[derive(Debug, Clone)] +pub enum NodeStorage { + /// `Db::init_memory` — BTreeMap store, no persistence. Tests and small + /// embedded runtimes. + Memory, + /// `Db::init_redb` — redb with an in-memory backend. What the WASM + /// `ClientDb.newInMemory` uses. + #[cfg(feature = "db-redb")] + RedbMemory, + /// `Db::init_redb_file` — redb on disk. Native servers and apps. + #[cfg(all(feature = "db-redb", not(target_arch = "wasm32")))] + RedbFile { + path: std::path::PathBuf, + uploads_path: std::path::PathBuf, + }, + /// `Db::init_redb_opfs` — redb in the browser's OPFS, optionally + /// encrypted at rest. What the WASM `ClientDb` constructor uses. + #[cfg(all(feature = "db-redb", target_arch = "wasm32"))] + Opfs { + filename: String, + encryption_key: Option<[u8; 32]>, + }, +} + +/// How to open an [`AtomicNode`]. +#[derive(Debug, Clone)] +pub struct NodeConfig { + pub storage: NodeStorage, + /// The node's own origin (`https://example.com`). `None` for a pure + /// client cache that owns no subjects. + pub base_domain: Option, + /// The local agent that signs [`AtomicNode::mutate`] commits. Set as the + /// store's default agent; `None` leaves the node read-only for local + /// edits until one is set via [`AtomicNode::set_agent`]. + pub agent: Option, +} + +impl NodeConfig { + /// In-memory node with no owned domain and no agent. + pub fn memory() -> Self { + Self { + storage: NodeStorage::Memory, + base_domain: None, + agent: None, + } + } + + pub fn with_base_domain(mut self, base_domain: impl Into) -> Self { + self.base_domain = Some(base_domain.into()); + self + } + + pub fn with_agent(mut self, agent: Agent) -> Self { + self.agent = Some(agent); + self + } +} + +/// The trust role under which a signed commit is ingested. Names follow +/// `CommitIngestOpts::{hub, peer, replica}` (PR #1274); `LocalCache` is the +/// "fourth policy" that PR left out — the browser's WASM cache applying a +/// commit the server already accepted. +/// +/// | policy | signature | rights | timestamp | ownership | loro causality | live echo | +/// | --- | --- | --- | --- | --- | --- | --- | +/// | `Hub` | yes | yes | yes | yes | yes | fan out | +/// | `Peer` | yes | yes | yes | no | no | suppressed | +/// | `Replica` | yes | no | no | no | no | suppressed | +/// | `LocalCache` | no | no | no | no | no | fan out | +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub enum IngestPolicy { + /// This node owns the subject and is the authority: HTTP `/commit` and + /// hub WS `COMMIT`. `source_id` is the connection the commit came from, + /// so the commit monitor does not echo it back; `response_origin` + /// resolves `internal:/` subjects in the response. + Hub { + source_id: Option, + response_origin: Option, + }, + /// A signed commit from another full node (Iroh / peer `COMMIT` frame). + /// Fully validated, but this node may host subjects it does not own and + /// concurrent writes are expected. + #[default] + Peer, + /// Catch-up from a hub that already validated the commit (Flutter WS + /// session): only the signature is re-checked. + Replica, + /// A trusted local cache (browser WASM/OPFS) mirroring what its hub + /// already accepted. Nothing is validated; only the index is updated. + LocalCache, +} + +impl IngestPolicy { + /// Hub policy with no source id and the store's own base domain as origin. + pub fn hub() -> Self { + Self::Hub { + source_id: None, + response_origin: None, + } + } +} + +/// A local edit for [`AtomicNode::mutate`]. Both arms sign with the node's +/// agent and apply locally, without validating the signature or rights +/// again (the node trusts its own agent) and without any network I/O. +pub enum ResourceEdit<'a> { + /// Sign and apply the pending changes on an existing resource + /// (`Resource::save_locally`). + Update(&'a mut Resource), + /// Mint a new DID resource from this draft; its subject becomes + /// `did:ad:{signature}` (`Resource::save_as_genesis`). + Genesis(&'a mut Resource), +} + +/// A running Atomic node: durable store plus the local agent that signs for +/// it. Cheap to clone (`Db` is a bundle of `Arc`s); clones share the store, +/// its event channel and its default agent. +#[derive(Clone)] +pub struct AtomicNode { + db: Db, +} + +impl AtomicNode { + /// Open a node. Delegates to `Db::init_memory` / `init_redb` / + /// `init_redb_file` / `init_redb_opfs` depending on + /// [`NodeConfig::storage`], then installs the agent as the store's + /// default agent. + pub async fn open(config: NodeConfig) -> AtomicResult { + let NodeConfig { + storage, + base_domain, + agent, + } = config; + let db = match storage { + NodeStorage::Memory => Db::init_memory(base_domain).await?, + #[cfg(feature = "db-redb")] + NodeStorage::RedbMemory => Db::init_redb(base_domain).await?, + #[cfg(all(feature = "db-redb", not(target_arch = "wasm32")))] + NodeStorage::RedbFile { path, uploads_path } => { + Db::init_redb_file(&path, base_domain, &uploads_path).await? + } + #[cfg(all(feature = "db-redb", target_arch = "wasm32"))] + NodeStorage::Opfs { + filename, + encryption_key, + } => Db::init_redb_opfs(base_domain, &filename, encryption_key.as_ref()).await?, + }; + let node = Self::from_db(db); + if let Some(agent) = agent { + node.set_agent(agent); + } + Ok(node) + } + + /// Wrap an already-opened store. For adapters that still construct `Db` + /// themselves (the WASM `ClientDb`, the server's `AppState`) while they + /// migrate to [`AtomicNode::open`]. + pub fn from_db(db: Db) -> Self { + Self { db } + } + + /// The underlying store, for operations this slice does not name yet + /// (blobs, version vectors, import/export, drive setup). + pub fn db(&self) -> &Db { + &self.db + } + + /// The agent that signs local edits (`Storelike::get_default_agent`). + pub fn agent(&self) -> Option { + self.db.get_default_agent().ok() + } + + /// Install (or replace) the agent that signs local edits + /// (`Storelike::set_default_agent`). + pub fn set_agent(&self, agent: Agent) { + self.db.set_default_agent(agent); + } + + /// Read a resource as `for_agent` would see it, including dynamic + /// (endpoint / class-extender) properties and the read-rights check + /// (`Storelike::get_resource_extended` with `skip_dynamic = false`). + pub async fn get( + &self, + subject: &Subject, + for_agent: &ForAgent, + ) -> AtomicResult { + self.db + .get_resource_extended(subject, false, for_agent) + .await + } + + /// Run an indexed query (`Storelike::query`). Rights are checked per hit + /// via `Query::for_agent`. + pub async fn query(&self, q: &Query) -> AtomicResult { + self.db.query(q).await + } + + /// Ingest a signed JSON-AD commit under `policy`. + /// + /// - `Hub` / `Peer`: `sync::engine::ingest_commit` with the matching + /// `CommitIngestOpts` — the path `server/src/handlers/commit.rs` and + /// the peer `COMMIT` frame handler already use. + /// - `Replica`: `sync::ws_apply::apply_commit_json` — the Flutter WS + /// catch-up path. + /// - `LocalCache`: `Db::apply_commit` with every `validate_*` off — the + /// options the WASM `ClientDb.applyCommit` used to carry inline. + pub async fn apply_commit( + &self, + commit_json: &str, + policy: IngestPolicy, + ) -> AtomicResult { + match policy { + IngestPolicy::Hub { + source_id, + response_origin, + } => { + ingest_commit( + &self.db, + commit_json, + &CommitIngestOpts { + source_id, + validate_loro_causality: true, + enforce_subject_ownership: true, + suppress_live_echo: false, + response_origin, + }, + ) + .await + } + IngestPolicy::Peer => { + ingest_commit( + &self.db, + commit_json, + &CommitIngestOpts { + source_id: None, + validate_loro_causality: false, + enforce_subject_ownership: false, + suppress_live_echo: true, + response_origin: None, + }, + ) + .await + } + IngestPolicy::Replica => { + crate::sync::ws_apply::apply_commit_json(&self.db, commit_json).await + } + IngestPolicy::LocalCache => { + // `DontSave`: the default would persist the parsed Commit + // resource (validating required props) before `apply_commit` + // runs; `apply_commit` is the persistence step here. + let commit_resource = crate::parse::parse_json_ad_resource( + commit_json, + &self.db, + &crate::parse::ParseOpts { + save: crate::parse::SaveOpts::DontSave, + ..Default::default() + }, + ) + .await?; + let commit = crate::Commit::from_resource(commit_resource)?; + let opts = crate::commit::CommitOpts { + update_index: true, + ..crate::commit::CommitOpts::no_validations_no_index() + }; + self.db.apply_commit(commit, &opts).await + } + } + } + + /// Sign a local edit with the node's agent and apply it to this store + /// (`Resource::save_locally` / `Resource::save_as_genesis`). Nothing is + /// sent anywhere: hand `CommitResponse::commit` to a transport (or to + /// another node's [`apply_commit`](Self::apply_commit)) to propagate it. + /// + /// Fails with the store's "No agent set" error when the node has no agent. + pub async fn mutate(&self, edit: ResourceEdit<'_>) -> AtomicResult { + match edit { + ResourceEdit::Update(resource) => resource.save_locally(&self.db).await, + ResourceEdit::Genesis(resource) => resource.save_as_genesis(&self.db).await, + } + } + + /// Change notifications for every write to this store, whatever path it + /// came in on (`Db::subscribe_events`). Lagging receivers drop the + /// oldest events, as with any `tokio::sync::broadcast` channel. + pub fn subscribe(&self) -> broadcast::Receiver { + self.db.subscribe_events() + } + + /// Bulk-sync `drive` with an Iroh peer + /// (`sync::peer::sync_drive_with_peer_outcome`). Requires the global Iroh + /// endpoint to be running (`sync::peer::start`); that lifecycle is not + /// owned by the node yet. + #[cfg(feature = "iroh")] + pub async fn sync_with_peer( + &self, + node_id: &str, + drive: &Subject, + ) -> AtomicResult { + crate::sync::peer::sync_drive_with_peer_outcome(node_id, drive.as_str(), &self.db).await + } +} + +#[cfg(all(test, feature = "db-redb"))] +mod tests { + use super::*; + use crate::{client::commit_to_wire_json, urls, Value}; + + async fn open_test_node(label: &str) -> AtomicNode { + let node = AtomicNode::open(NodeConfig { + storage: NodeStorage::RedbMemory, + ..NodeConfig::memory().with_base_domain("https://localhost") + }) + .await + .unwrap_or_else(|e| panic!("{label}: open failed: {e}")); + node.db().populate().await.unwrap(); + node + } + + /// Two nodes in one process, no Actix: a genesis commit minted on one via + /// `mutate` is ingested on the other via `apply_commit(Peer)`, after which + /// `query` and `get` on the second node reflect it and `subscribe` on the + /// second node saw exactly one change. + #[tokio::test] + async fn two_nodes_mutate_then_peer_ingest() { + let alice_node = open_test_node("alice").await; + let (alice, drive) = alice_node.db().setup("Alice").await.unwrap(); + let drive = Subject::from(drive); + assert_eq!( + alice_node.agent().map(|a| a.subject), + Some(alice.subject.clone()) + ); + + let bob_node = open_test_node("bob").await; + bob_node.db().setup("Bob").await.unwrap(); + let mut bob_events = bob_node.subscribe(); + + // Alice mints a classless DID document under her drive. + let mut draft = Resource::new("did:ad:placeholder".into()); + draft + .set_unsafe(urls::NAME.into(), Value::String("Peer Doc".into())) + .unwrap(); + draft + .set_unsafe(urls::PARENT.into(), Value::AtomicUrl(drive.clone())) + .unwrap(); + let response = alice_node + .mutate(ResourceEdit::Genesis(&mut draft)) + .await + .unwrap(); + let subject = response.commit.subject.clone(); + assert!(subject.as_str().starts_with("did:ad:"), "got {subject}"); + assert!( + bob_node.db().get_resource(&subject).await.is_err(), + "bob must not see alice's write before ingesting it" + ); + + // The commit crosses the (in-process) wire as JSON-AD. + let wire = commit_to_wire_json(&response.commit, alice_node.db()) + .await + .unwrap(); + let ingested = bob_node + .apply_commit(&wire, IngestPolicy::Peer) + .await + .expect("bob must accept alice's signed genesis commit under Peer policy"); + assert_eq!(ingested.commit.signer, alice.subject); + + // `get` sees it (as sudo: bob has no rights on alice's doc). + let got = bob_node + .get(&subject, &ForAgent::Sudo) + .await + .unwrap() + .to_single(); + assert_eq!(got.get(urls::NAME).unwrap().to_string(), "Peer Doc"); + + // `query` sees it. + let result = bob_node + .query(&Query { + property: Some(urls::PARENT.into()), + value: Some(Value::AtomicUrl(drive.clone())), + for_agent: ForAgent::Sudo, + ..Query::new() + }) + .await + .unwrap(); + assert_eq!(result.subjects, vec![subject.clone()]); + + // `subscribe` saw the ingest: the ingest stores the commit resource + // and the document, each emitting a `Changed` event. + let mut changed = Vec::new(); + while let Ok(event) = bob_events.try_recv() { + if let DbEvent::Changed { subject, .. } = event { + changed.push(subject.pure_id()); + } + } + assert!( + changed.contains(&subject.pure_id()), + "expected a Changed event for {subject}, got {changed:?}" + ); + } + + /// `LocalCache` is today's WASM `applyCommit`: it applies an unsigned, + /// unauthorized commit without complaint, because the cache trusts its + /// hub. `Peer` must reject the very same bytes. + #[tokio::test] + async fn local_cache_skips_validation_peer_does_not() { + let hub = open_test_node("hub").await; + let (_alice, drive) = hub.db().setup("Alice").await.unwrap(); + let drive = Subject::from(drive); + let mut draft = Resource::new("did:ad:placeholder".into()); + draft + .set_unsafe(urls::NAME.into(), Value::String("Cached".into())) + .unwrap(); + draft + .set_unsafe(urls::PARENT.into(), Value::AtomicUrl(drive)) + .unwrap(); + let response = hub.mutate(ResourceEdit::Genesis(&mut draft)).await.unwrap(); + // What the hub pushes to its caches over WS: the stored commit + // resource, `@id` included (`ingest_commit_json`'s return value). + let mut wire: serde_json::Value = + serde_json::from_str(&response.commit_resource.to_json_ad(None).unwrap()).unwrap(); + // Corrupt the signature: a peer must notice, a local cache does not check. + wire[urls::SIGNATURE] = serde_json::Value::String("AAAA".into()); + let tampered = wire.to_string(); + + let cache = open_test_node("cache").await; + cache + .apply_commit(&tampered, IngestPolicy::LocalCache) + .await + .expect("LocalCache applies without validating the signature"); + cache + .get(&response.commit.subject, &ForAgent::Sudo) + .await + .expect("cached resource is readable"); + + let peer = open_test_node("peer").await; + peer.apply_commit(&tampered, IngestPolicy::Peer) + .await + .expect_err("Peer validates the signature and must reject the tampered commit"); + } + + /// Without an agent, `mutate` fails with the store's own error instead + /// of panicking or silently signing with nothing. + #[tokio::test] + async fn mutate_without_agent_is_an_error() { + let node = AtomicNode::open(NodeConfig::memory()).await.unwrap(); + assert!(node.agent().is_none()); + let mut draft = Resource::new("did:ad:placeholder".into()); + let err = node + .mutate(ResourceEdit::Genesis(&mut draft)) + .await + .expect_err("no agent, no signature"); + assert!(err.to_string().contains("No agent set"), "got: {err}"); + } +} diff --git a/lib/src/sync/engine.rs b/lib/src/sync/engine.rs index 8369a319f4..de57764e54 100644 --- a/lib/src/sync/engine.rs +++ b/lib/src/sync/engine.rs @@ -336,6 +336,22 @@ pub async fn ingest_commit_json( commit_json: &str, opts: &CommitIngestOpts, ) -> crate::errors::AtomicResult { + let response = ingest_commit(store, commit_json, opts).await?; + let base_domain = store.get_base_domain(); + let origin = opts.response_origin.as_deref().or(base_domain.as_deref()); + let json = response.commit_resource.to_json_ad(origin)?; + Ok(json) +} + +/// [`ingest_commit_json`] minus the final JSON-AD serialization: the same +/// validation and application, returning the full [`CommitResponse`] so an +/// in-process caller (`crate::runtime::AtomicNode`) can use the changed +/// resource and atoms without re-parsing its own output. +pub async fn ingest_commit( + store: &Db, + commit_json: &str, + opts: &CommitIngestOpts, +) -> crate::errors::AtomicResult { // Reject commits with deprecated set/push/remove fields — use loroUpdate instead. if commit_json.contains("\"https://atomicdata.dev/properties/set\"") || commit_json.contains("\"https://atomicdata.dev/properties/push\"") @@ -445,23 +461,17 @@ pub async fn ingest_commit_json( source_id: opts.source_id.clone(), }; - let base_domain = store.get_base_domain(); - - let response = if opts.suppress_live_echo { + if opts.suppress_live_echo { // Applying a remote peer's commit must not rebroadcast to live peers // (the sender included) — mirrors `ws_apply::apply_commit_json`'s // suppression of the same echo via the live push loop. super::ws_apply::set_importing(true); let result = store.apply_commit(incoming_commit, &commit_opts).await; super::ws_apply::set_importing(false); - result? + result } else { - store.apply_commit(incoming_commit, &commit_opts).await? - }; - - let origin = opts.response_origin.as_deref().or(base_domain.as_deref()); - let json = response.commit_resource.to_json_ad(origin)?; - Ok(json) + store.apply_commit(incoming_commit, &commit_opts).await + } } /// Apply a JSON-AD `COMMIT` received over a peer transport, returning the diff --git a/lib/src/sync/ws_apply.rs b/lib/src/sync/ws_apply.rs index 6a0c51d1dd..319d700b87 100644 --- a/lib/src/sync/ws_apply.rs +++ b/lib/src/sync/ws_apply.rs @@ -6,7 +6,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use crate::{ - commit::{Commit, CommitOpts}, + commit::{Commit, CommitOpts, CommitResponse}, db::Db, errors::AtomicResult, parse::parse_json_ad_commit_resource, @@ -335,7 +335,11 @@ pub async fn apply_destroy_checked(store: &Db, subject: &str) -> AtomicResult<() } /// Apply a JSON-AD commit received over WS (legacy text `COMMIT` or after fetch). -pub async fn apply_commit_json(store: &Db, body: &str) -> AtomicResult<()> { +/// +/// Replica policy: the hub already accepted this commit, so only the +/// signature is re-checked — rights and timestamp are not. This is +/// `crate::runtime::IngestPolicy::Replica`. +pub async fn apply_commit_json(store: &Db, body: &str) -> AtomicResult { set_importing(true); let result = async { let resource = parse_json_ad_commit_resource(body, store).await?; @@ -348,8 +352,7 @@ pub async fn apply_commit_json(store: &Db, body: &str) -> AtomicResult<()> { update_index: true, ..CommitOpts::no_validations_no_index() }; - store.apply_commit(commit, &opts).await?; - Ok::<(), crate::AtomicError>(()) + store.apply_commit(commit, &opts).await } .await; set_importing(false); diff --git a/planning/atomic-lib-runtime.md b/planning/atomic-lib-runtime.md index 03c78b8c34..9c58d1c3ee 100644 --- a/planning/atomic-lib-runtime.md +++ b/planning/atomic-lib-runtime.md @@ -2,7 +2,50 @@ ## Status -Proposal. This document describes the target architecture for making +**Slice 1 shipped:** `lib/src/runtime/` exists. `AtomicNode` wraps `Db` with +`open` / `from_db`, `get`, `query`, `apply_commit(json, IngestPolicy)`, +`mutate(ResourceEdit)`, `subscribe`, and (behind `iroh`) `sync_with_peer`. +Every method delegates to a function that already existed — see the doc +comment on each one in `lib/src/runtime/node.rs`. `IngestPolicy::{Hub, Peer, +Replica, LocalCache}` names the four commit-validation profiles that used to be +inline `CommitOpts` / `CommitIngestOpts` literals in `server/src/handlers/commit.rs`, +`sync/engine.rs`, `sync/ws_apply.rs`, and `wasm/src/lib.rs`. The WASM +`ClientDb` is the first adapter on it: `applyCommit` is +`node.apply_commit(_, LocalCache)` and `query` is `node.query`. Tests: +`runtime::node::tests` — two in-process nodes, `mutate` on one, `apply_commit(Peer)` +on the other, `query`/`get`/`subscribe` reflect it; `LocalCache` accepts what `Peer` +rejects; `mutate` without an agent errors. + +What slice 1 did **not** do (deliberately — no behaviour change): + +- `mutate` still applies through `Resource::save_locally` / `save_as_genesis` + (local opts: signature and rights not re-checked), not through + `apply_commit(Hub)`. Routing it through `Hub` would add a JSON round-trip and + ownership/causality checks that today's local save does not run. +- `Replica` and `LocalCache` delegate to `ws_apply::apply_commit_json` and the + former WASM `CommitOpts` respectively, because `CommitIngestOpts` on `develop` + has no `validate_rights` / `validate_timestamp` knobs yet. Once #1274 lands, + fold both into `ingest_commit` with `CommitIngestOpts::{replica, local_cache}`. +- `sync_with_peer` wraps `sync::peer::sync_drive_with_peer_outcome` and so needs + the global Iroh endpoint (`peer::start`) — the node does not own that lifecycle. +- No `NodeEvent`, blob service, outbox, or `AtomicTransport`; `subscribe` is + `Db::subscribe_events` (`DbEvent`) as-is. +- WASM `getResource` keeps the raw `Db::get_resource` (no rights check, no + dynamic properties) — that is what the cache semantics require; the node's + `get` is the rights-checked extended read. + +Next: + +1. #1274 merges → `IngestPolicy` maps 1:1 onto `CommitIngestOpts` constructors + and the `Replica` / `LocalCache` arms lose their separate code paths. +2. `server/src/handlers/commit.rs` → `node.apply_commit(_, Hub { source_id, + response_origin })` (Phase 2 below); `AppState` holds an `AtomicNode`. +3. `ffi/` (#1277) and `python/` bind `AtomicNode`; `flutter/rust/src/api/simple.rs` + store group calls it (#1241 follow-up). +4. Shared fixtures for the remaining pure twins + (`planning/runtime-boundary-decision.md`, sequencing step 3). + +This document describes the target architecture for making `atomic_lib` able to run a complete Atomic node by itself. HTTP remains a supported adapter for hosted servers and interoperability, but local app runtime behavior must not depend on HTTP endpoints or a loopback server. @@ -588,18 +631,18 @@ needing an HTTP server in front of it. ### Phase 1: Introduce `AtomicNode` Without Behavior Change -- Add `lib/src/runtime/`. -- Add `AtomicNode`, `NodeConfig`, and simple constructors around existing `Db` +- [x] Add `lib/src/runtime/`. +- [x] Add `AtomicNode`, `NodeConfig`, and simple constructors around existing `Db` initialization. -- Add `get`, `query`, `apply_commit`, `put_blob`, and `get_blob` by delegating to - existing code. -- Keep server using `AppState`, but allow `AppState` to hold an `AtomicNode`. +- [x] Add `get`, `query`, `apply_commit` by delegating to existing code. +- [ ] Add `put_blob` and `get_blob`. +- [ ] Keep server using `AppState`, but allow `AppState` to hold an `AtomicNode`. Tests: -- Existing `atomic_lib` tests still pass. -- New smoke test opens an in-memory `AtomicNode`, creates an agent, mutates a - resource, queries it, and reads it back. +- [x] Existing `atomic_lib` tests still pass. +- [x] New smoke test opens in-memory `AtomicNode`s, creates an agent, mutates a + resource, ingests it on a second node, queries it, and reads it back. ### Phase 2: Thin Server Handlers diff --git a/wasm/src/lib.rs b/wasm/src/lib.rs index 8afd541a47..4ba0cb698c 100644 --- a/wasm/src/lib.rs +++ b/wasm/src/lib.rs @@ -7,15 +7,15 @@ #![cfg(target_arch = "wasm32")] use atomic_lib::{ - commit::CommitOpts, parse::ParseOpts, + runtime::{AtomicNode, IngestPolicy}, storelike::{Query, QueryResult, Storelike}, vault::dek::DriveVaultKey, vault::keys::{argon2id_derive_key, Argon2Params}, vault::secret_envelope::{NewWrapper, SecretEnvelope, Unlock}, vault::store::{MemoryVaultStore, VaultObjectStore}, vault::sync::{commit_lane_state, drive_prefix, export_vault_delta, import_vault_batch}, - Commit, Db, Resource, Subject, Value, + Db, Resource, Subject, Value, }; use wasm_bindgen::prelude::*; @@ -43,9 +43,20 @@ const STORAGE_BLOCKED_MARKER: &str = "ATOMIC_DB_STORAGE_BLOCKED"; /// A client-side Atomic Data database backed by redb (in-memory, future OPFS). /// Provides indexed queries, resource storage, and commit application. +/// +/// A JS binding over [`AtomicNode`]: query and commit application go through +/// the node; the cache-shaped operations (raw get/put, blobs, version +/// vectors, import/export, vault) still reach the store directly until the +/// node names them. #[wasm_bindgen] pub struct ClientDb { - db: Db, + node: AtomicNode, +} + +impl ClientDb { + fn db(&self) -> &Db { + self.node.db() + } } #[wasm_bindgen] @@ -100,7 +111,9 @@ impl ClientDb { ) .into(), ); - Ok(ClientDb { db }) + Ok(ClientDb { + node: AtomicNode::from_db(db), + }) } /// Create a non-persistent in-memory ClientDb. Used in environments @@ -109,7 +122,9 @@ impl ClientDb { #[wasm_bindgen(js_name = "newInMemory")] pub async fn new_in_memory(base_url: Option) -> Result { let db = Db::init_redb(base_url).await.map_err(to_js_err)?; - Ok(ClientDb { db }) + Ok(ClientDb { + node: AtomicNode::from_db(db), + }) } /// Persist buffered writes to durable OPFS storage. @@ -122,7 +137,7 @@ impl ClientDb { /// server re-fetches) but data loss the moment you're disconnected. The /// worker calls this on a short periodic tick, mirroring the native server. pub fn flush(&self) -> Result<(), JsError> { - self.db.flush().map_err(to_js_err) + self.db().flush().map_err(to_js_err) } /// Get a resource by its subject URL. Returns JSON-AD string or null. @@ -132,8 +147,8 @@ impl ClientDb { // URL it was served, while the store is keyed by `internal:`. // `Subject::from` drops the base domain and would look up an // `External` subject that does not exist here. - let subject = Subject::from_raw(subject, self.db.get_base_domain().as_deref()); - match self.db.get_resource(&subject).await { + let subject = Subject::from_raw(subject, self.db().get_base_domain().as_deref()); + match self.db().get_resource(&subject).await { Ok(resource) => { let json = resource_to_json_ad(&resource, &self.origin())?; Ok(JsValue::from_str(&json)) @@ -154,7 +169,7 @@ impl ClientDb { // the intended persistence step — it skips validation deliberately. let resource = atomic_lib::parse::parse_json_ad_resource( json_ad, - &self.db, + self.db(), &ParseOpts { skip_unknown_props: true, save: atomic_lib::parse::SaveOpts::DontSave, @@ -163,7 +178,7 @@ impl ClientDb { ) .await .map_err(to_js_err)?; - self.db + self.db() .add_resource_opts(&resource, false, true, true) .await .map_err(to_js_err)?; @@ -174,36 +189,14 @@ impl ClientDb { /// This is the efficient incremental update path: the Loro diff /// determines exactly which atoms changed, so only affected index /// entries are updated. Use this for real-time updates (COMMIT messages). + /// + /// The server already validated the commit, so this is + /// [`IngestPolicy::LocalCache`]: no signature, rights, timestamp or + /// schema checks — index update only. #[wasm_bindgen(js_name = "applyCommit")] pub async fn apply_commit(&self, commit_json_ad: &str) -> Result<(), JsError> { - // `DontSave` is required: the default would store the parsed Commit - // resource via `add_resource()` (which validates required props) - // before `apply_commit` runs. `apply_commit` is the proper persistence - // path here. - let commit_resource = atomic_lib::parse::parse_json_ad_resource( - commit_json_ad, - &self.db, - &ParseOpts { - save: atomic_lib::parse::SaveOpts::DontSave, - ..Default::default() - }, - ) - .await - .map_err(to_js_err)?; - let commit = Commit::from_resource(commit_resource).map_err(to_js_err)?; - let opts = CommitOpts { - validate_schema: false, - validate_signature: false, - validate_timestamp: false, - validate_rights: false, - validate_previous_commit: false, - validate_loro_causality: false, - validate_for_agent: None, - update_index: true, - source_id: None, - }; - self.db - .apply_commit(commit, &opts) + self.node + .apply_commit(commit_json_ad, IngestPolicy::LocalCache) .await .map_err(to_js_err)?; Ok(()) @@ -213,7 +206,10 @@ impl ClientDb { #[wasm_bindgen(js_name = "removeResource")] pub async fn remove_resource(&self, subject: &str) -> Result<(), JsError> { let subject = Subject::from(subject); - self.db.remove_resource(&subject).await.map_err(to_js_err)?; + self.db() + .remove_resource(&subject) + .await + .map_err(to_js_err)?; Ok(()) } @@ -244,7 +240,7 @@ impl ClientDb { operator: Option, } - let base_domain = self.db.get_base_domain(); + let base_domain = self.db().get_base_domain(); let mut extra: Vec = if filters.is_null() || filters.is_undefined() { @@ -276,7 +272,7 @@ impl ClientDb { }; let raw = raw.clone(); filter.value = Some( - atomic_lib::collections::delocalize_filter_value(&self.db, Some(property), &raw) + atomic_lib::collections::delocalize_filter_value(self.db(), Some(property), &raw) .await, ); } @@ -284,7 +280,7 @@ impl ClientDb { let value = match value { Some(raw) => Some( atomic_lib::collections::delocalize_filter_value( - &self.db, + self.db(), property.as_deref(), &raw, ) @@ -333,7 +329,7 @@ impl ClientDb { drive: drive.map(|d| Subject::from_raw(&d, base_domain.as_deref())), }; - let result = self.db.query(&q).await.map_err(to_js_err)?; + let result = self.node.query(&q).await.map_err(to_js_err)?; let response = QueryResponse::from_result(&result, &self.origin())?; serde_wasm_bindgen::to_value(&response).map_err(|e| JsError::new(&e.to_string())) } @@ -343,7 +339,7 @@ impl ClientDb { /// the wasm boundary must be a URL (or a DID) the client can actually /// fetch, matching what the server sends over HTTP. fn origin(&self) -> String { - self.db + self.db() .get_base_domain() .unwrap_or_else(|| "http://localhost".to_string()) } @@ -352,7 +348,7 @@ impl ClientDb { #[wasm_bindgen(js_name = "putLoroSnapshot")] pub fn put_loro_snapshot(&self, subject: &str, data: &[u8]) -> Result<(), JsError> { use atomic_lib::db::trees::Tree; - self.db + self.db() .kv .insert(Tree::LoroSnapshots, subject.as_bytes(), data) .map_err(to_js_err) @@ -361,13 +357,13 @@ impl ClientDb { /// Opaque versioned state bytes for a resource. Returns null if not found. #[wasm_bindgen(js_name = "getStateSnapshot")] pub fn get_state_snapshot(&self, subject: &str) -> Result { - Self::state_snapshot_js(&self.db, subject) + Self::state_snapshot_js(self.db(), subject) } /// Back-compat alias for browser client-db (`getLoroSnapshot`). #[wasm_bindgen(js_name = "getLoroSnapshot")] pub fn get_loro_snapshot(&self, subject: &str) -> Result { - Self::state_snapshot_js(&self.db, subject) + Self::state_snapshot_js(self.db(), subject) } fn state_snapshot_js(db: &atomic_lib::Db, subject: &str) -> Result { @@ -386,7 +382,7 @@ impl ClientDb { if hash.len() != 32 { return Err(to_js_err("Hash must be 32 bytes")); } - self.db + self.db() .kv .insert(Tree::Blobs, hash, data) .map_err(to_js_err) @@ -399,7 +395,7 @@ impl ClientDb { if hash.len() != 32 { return Err(to_js_err("Hash must be 32 bytes")); } - match self.db.kv.get(Tree::Blobs, hash) { + match self.db().kv.get(Tree::Blobs, hash) { Ok(Some(data)) => Ok(js_sys::Uint8Array::from(data.as_slice()).into()), Ok(None) => Ok(JsValue::NULL), Err(e) => Err(to_js_err(e)), @@ -422,7 +418,7 @@ impl ClientDb { let mut result: HashMap> = HashMap::new(); - for item in self.db.kv.iter_tree(Tree::LoroSnapshots) { + for item in self.db().kv.iter_tree(Tree::LoroSnapshots) { let (key_bytes, snapshot_bytes) = item.map_err(to_js_err)?; let subject = String::from_utf8(key_bytes).map_err(|e| JsError::new(&e.to_string()))?; @@ -458,16 +454,16 @@ impl ClientDb { use std::collections::HashMap; let drive_subject = - atomic_lib::Subject::from_raw(&drive, self.db.get_base_domain().as_deref()); + atomic_lib::Subject::from_raw(&drive, self.db().get_base_domain().as_deref()); let subjects = - atomic_lib::sync::engine::collect_drive_subjects(&self.db, &drive_subject).await; + atomic_lib::sync::engine::collect_drive_subjects(self.db(), &drive_subject).await; let mut result: HashMap> = HashMap::new(); for subject in subjects { // `collect_drive_subjects` yields `pure_id()` strings, which are // exactly the `LoroSnapshots` keys. - match self.db.kv.get(Tree::LoroSnapshots, subject.as_bytes()) { + match self.db().kv.get(Tree::LoroSnapshots, subject.as_bytes()) { Ok(Some(snapshot_bytes)) => { match AtomicLoroDoc::vv_map_from_snapshot(&snapshot_bytes) { Ok(vv) => { @@ -499,7 +495,7 @@ impl ClientDb { #[wasm_bindgen(js_name = "allSubjects")] pub fn all_subjects(&self) -> Result { let subjects: Vec = self - .db + .db() .all_resources(true) .map(|r| r.get_subject().to_string()) .collect(); @@ -509,7 +505,7 @@ impl ClientDb { /// Populate the database with default Atomic Data vocabulary /// (classes, properties, datatypes). pub async fn populate(&self) -> Result<(), JsError> { - self.db.populate().await.map_err(to_js_err) + self.db().populate().await.map_err(to_js_err) } /// Export all resources as a JSON array of JSON-AD objects. @@ -518,7 +514,7 @@ impl ClientDb { pub fn export_all_resources(&self) -> Result { let mut resources = Vec::new(); - for resource in self.db.all_resources(true) { + for resource in self.db().all_resources(true) { if let Ok(json_ad) = resource.to_json_ad(None) { resources.push(json_ad); } @@ -541,7 +537,7 @@ impl ClientDb { if let Ok(resource) = atomic_lib::parse::parse_json_ad_resource( &json_str, - &self.db, + self.db(), &ParseOpts { skip_unknown_props: true, save: atomic_lib::parse::SaveOpts::DontSave, @@ -552,7 +548,7 @@ impl ClientDb { { // Store without indexing — we build the index once at the end if self - .db + .db() .add_resource_opts(&resource, false, false, true) .await .is_ok() @@ -563,7 +559,7 @@ impl ClientDb { } // Build the full index once - self.db.build_index(true).map_err(to_js_err)?; + self.db().build_index(true).map_err(to_js_err)?; Ok(count) } @@ -899,11 +895,11 @@ impl ClientDb { segment: u32, ) -> Result { let key = drive_key(key_bytes, key_epoch)?; - let subject = Subject::from_raw(drive_subject, self.db.get_base_domain().as_deref()); + let subject = Subject::from_raw(drive_subject, self.db().get_base_domain().as_deref()); let staging = MemoryVaultStore::new(); let summary = export_vault_delta( - &self.db, + self.db(), &subject, &key, &staging, @@ -943,7 +939,7 @@ impl ClientDb { device_pubkey: &str, segment: u32, ) -> Result<(), JsError> { - commit_lane_state(&self.db, drive_pseudonym, device_pubkey, segment).map_err(to_js_err) + commit_lane_state(self.db(), drive_pseudonym, device_pubkey, segment).map_err(to_js_err) } /// Merge downloaded vault objects into this store. @@ -978,7 +974,7 @@ impl ClientDb { .map_err(to_js_err)?; } - let summary = import_vault_batch(&self.db, &key, &staging, &drive_prefix(drive_pseudonym)) + let summary = import_vault_batch(self.db(), &key, &staging, &drive_prefix(drive_pseudonym)) .await .map_err(to_js_err)?;