feat(wasm,runtime): complete WASM runtime stubs and add crate-level documentation

- Wire `wasm_bindgen_futures::spawn_local` into `runtime::spawn` (was
  a compile-gate placeholder).
- Wire `gloo_timers::future::sleep` into `runtime::sleep` (was a
  placeholder).
- Add `gloo-timers`, `wasm-bindgen-futures`, and `js-sys` as
  unconditional dependencies in `nodedb-lite/Cargo.toml` so the WASM
  target compiles without feature flags.
- Add crate-level documentation to `lib.rs` with a quick-start snippet,
  durability contract, and `Encryption` re-export.
- Minor cleanups in columnar store, CRDT engine, FTS checkpoint/manager,
  timeseries sync, and vector lazy-load (comment accuracy, import hygiene).

Author: farhan-syah

nodedb-lite/Cargo.toml

@@ -51,6 +51,9 @@ getrandom = { workspace = true }
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 pagedb = { workspace = true, features = ["opfs"] }
 tokio = { workspace = true, features = ["sync", "rt", "macros"] }
+gloo-timers = { version = "0.3", features = ["futures"] }
+wasm-bindgen-futures = "0.4"
+js-sys = { workspace = true }
 
 # Async runtime (native only — WASM uses wasm-bindgen-futures)
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]

nodedb-lite/src/engine/columnar/store.rs

@@ -27,7 +27,7 @@ use crate::error::LiteError;
 use crate::runtime::now_millis_i64;
 use crate::storage::engine::{StorageEngine, WriteOp};
 #[cfg(not(target_arch = "wasm32"))]
-use crate::sync::ColumnarOutbound;
+use crate::sync::outbound::columnar::ColumnarOutbound;
 #[cfg(not(target_arch = "wasm32"))]
 use crate::sync::outbound::timeseries::TimeseriesOutbound;
 
@@ -138,7 +138,7 @@ pub struct ColumnarEngine<S: StorageEngine> {
     /// Optional outbound queue for plain columnar insert sync.
     /// `None` when sync is disabled or not yet configured.
     #[cfg(not(target_arch = "wasm32"))]
-    outbound: Option<Arc<ColumnarOutbound>>,
+    outbound: Option<Arc<ColumnarOutbound<S>>>,
     /// Optional outbound queue for timeseries-profile insert sync.
     ///
     /// Timeseries collections must use `TimeseriesPush` frames on Origin
@@ -147,7 +147,7 @@ pub struct ColumnarEngine<S: StorageEngine> {
     /// collections with `ColumnarProfile::Timeseries` are enqueued here
     /// instead of `outbound`.
     #[cfg(not(target_arch = "wasm32"))]
-    timeseries_outbound: Option<Arc<TimeseriesOutbound>>,
+    timeseries_outbound: Option<Arc<TimeseriesOutbound<S>>>,
 }
 
 impl<S: StorageEngine> ColumnarEngine<S> {
@@ -167,7 +167,7 @@ impl<S: StorageEngine> ColumnarEngine<S> {
     ///
     /// Must be called before any inserts if columnar sync is desired.
     #[cfg(not(target_arch = "wasm32"))]
-    pub fn set_outbound(&mut self, outbound: Arc<ColumnarOutbound>) {
+    pub fn set_outbound(&mut self, outbound: Arc<ColumnarOutbound<S>>) {
         self.outbound = Some(outbound);
     }
 
@@ -177,7 +177,7 @@ impl<S: StorageEngine> ColumnarEngine<S> {
     /// are routed here instead of `outbound`, so the transport can send them
     /// as `TimeseriesPush` frames to Origin's timeseries engine.
     #[cfg(not(target_arch = "wasm32"))]
-    pub fn set_timeseries_outbound(&mut self, outbound: Arc<TimeseriesOutbound>) {
+    pub fn set_timeseries_outbound(&mut self, outbound: Arc<TimeseriesOutbound<S>>) {
         self.timeseries_outbound = Some(outbound);
     }
 
@@ -518,32 +518,103 @@ impl<S: StorageEngine> ColumnarEngine<S> {
 
     /// Insert a row into a columnar collection's memtable.
     ///
-    /// When a sync outbound queue is attached the row is also enqueued for
-    /// replication to Origin.  Timeseries-profile collections use the
-    /// `timeseries_outbound` queue (→ `TimeseriesPush` frames); all other
-    /// columnar collections use `outbound` (→ `ColumnarInsert` frames).
+    /// This is a pure in-memory operation. Call [`enqueue_outbound`] from
+    /// an async context after inserting to durably enqueue the row for
+    /// replication to Origin.
     pub fn insert(&self, collection: &str, values: &[Value]) -> Result<(), LiteError> {
         let state_arc = self.lookup(collection)?;
         let mut s = Self::lock_state(&state_arc)?;
         s.mutation.insert(values).map_err(columnar_err_to_lite)?;
+        Ok(())
+    }
 
-        #[cfg(not(target_arch = "wasm32"))]
-        if matches!(s.profile, ColumnarProfile::Timeseries { .. }) {
-            // Timeseries rows must replicate via TimeseriesPush so that Origin
-            // stores them in its timeseries engine, not the columnar MutationEngine.
-            if let Some(ts_out) = &self.timeseries_outbound {
-                let column_names: Vec<String> = s
-                    .mutation
-                    .schema()
-                    .columns
-                    .iter()
-                    .map(|c| c.name.clone())
-                    .collect();
-                ts_out.enqueue_row(collection, column_names, values.to_vec());
+    /// Durably enqueue a batch of inserted rows for replication to Origin.
+    ///
+    /// Must be called from an async context after one or more successful
+    /// [`insert`] calls. Timeseries-profile collections are routed to the
+    /// `timeseries_outbound` queue; all other columnar collections use the
+    /// plain `outbound` queue.
+    ///
+    /// Returns [`LiteError::Backpressure`] when the queue is at cap so the
+    /// caller can propagate back-pressure. Other enqueue errors are logged as
+    /// warnings (the local insert already succeeded) and `Ok(())` is returned.
+    #[cfg(not(target_arch = "wasm32"))]
+    pub async fn enqueue_outbound(
+        &self,
+        collection: &str,
+        rows: &[Vec<Value>],
+    ) -> Result<(), LiteError> {
+        if rows.is_empty() {
+            return Ok(());
+        }
+
+        // Read the profile and schema metadata under the lock, then drop it
+        // before any await point to satisfy the no-lock-across-await rule.
+        enum OutboundRoute {
+            Timeseries { column_names: Vec<String> },
+            Columnar { schema_bytes: Vec<u8> },
+            None,
+        }
+
+        let route: OutboundRoute = {
+            let state_arc = self.lookup(collection)?;
+            let s = Self::lock_state(&state_arc)?;
+            if matches!(s.profile, ColumnarProfile::Timeseries { .. }) {
+                if self.timeseries_outbound.is_some() {
+                    let column_names: Vec<String> = s
+                        .mutation
+                        .schema()
+                        .columns
+                        .iter()
+                        .map(|c| c.name.clone())
+                        .collect();
+                    OutboundRoute::Timeseries { column_names }
+                } else {
+                    OutboundRoute::None
+                }
+            } else if self.outbound.is_some() {
+                let schema_bytes = zerompk::to_msgpack_vec(s.mutation.schema()).unwrap_or_default();
+                OutboundRoute::Columnar { schema_bytes }
+            } else {
+                OutboundRoute::None
+            }
+            // lock `s` is dropped here at end of block
+        };
+
+        match route {
+            OutboundRoute::Timeseries { column_names } => {
+                let queue = match &self.timeseries_outbound {
+                    Some(q) => Arc::clone(q),
+                    None => return Ok(()),
+                };
+                for row in rows {
+                    crate::sync::reconcile_outbound_enqueue(
+                        queue
+                            .enqueue_row(collection, column_names.clone(), row.clone())
+                            .await,
+                        "timeseries insert",
+                        collection,
+                        "",
+                    )?;
+                }
+            }
+            OutboundRoute::Columnar { schema_bytes } => {
+                let queue = match &self.outbound {
+                    Some(q) => Arc::clone(q),
+                    None => return Ok(()),
+                };
+                for row in rows {
+                    crate::sync::reconcile_outbound_enqueue(
+                        queue
+                            .enqueue_row(collection, row.clone(), schema_bytes.clone())
+                            .await,
+                        "columnar insert",
+                        collection,
+                        "",
+                    )?;
+                }
             }
-        } else if let Some(outbound) = &self.outbound {
-            let schema_bytes = zerompk::to_msgpack_vec(s.mutation.schema()).unwrap_or_default();
-            outbound.enqueue_row(collection, values.to_vec(), schema_bytes);
+            OutboundRoute::None => {}
         }
 
         Ok(())

nodedb-lite/src/engine/crdt/engine.rs

@@ -80,6 +80,11 @@ pub struct PendingDelta {
     pub document_id: String,
     /// Loro delta bytes (compact binary).
     pub delta_bytes: Vec<u8>,
+    /// Stable idempotent-producer seq for this delta. 0 = unassigned;
+    /// assigned at first send and reused on reconnect re-send so Origin
+    /// dedups instead of double-applying.
+    #[serde(default)]
+    pub seq: u64,
 }
 
 impl CrdtEngine {
@@ -162,6 +167,7 @@ impl CrdtEngine {
             collection: collection.to_string(),
             document_id: doc_id.to_string(),
             delta_bytes,
+            seq: 0,
         });
 
         Ok(mutation_id)
@@ -191,6 +197,7 @@ impl CrdtEngine {
             collection: collection.to_string(),
             document_id: doc_id.to_string(),
             delta_bytes,
+            seq: 0,
         });
 
         Ok(mutation_id)
@@ -240,6 +247,7 @@ impl CrdtEngine {
             collection: collection_name,
             document_id: format!("{}_ops", ops.len()),
             delta_bytes,
+            seq: 0,
         });
 
         Ok(mutation_id)
@@ -317,6 +325,7 @@ impl CrdtEngine {
             collection: "deferred".to_string(),
             document_id: format!("{count}_ops"),
             delta_bytes,
+            seq: 0,
         });
 
         self.deferred_count = 0;
@@ -375,6 +384,7 @@ impl CrdtEngine {
                 collection: collection.to_string(),
                 document_id: "*".to_string(),
                 delta_bytes,
+                seq: 0,
             });
         }
 
@@ -444,6 +454,22 @@ impl CrdtEngine {
         self.pending_deltas.retain(|d| d.mutation_id != mutation_id);
     }
 
+    /// Assign a stable stream seq to a pending delta the first time it is sent.
+    ///
+    /// If the delta already has a non-zero seq (assigned on a previous send)
+    /// the call is a no-op — the existing seq is reused on reconnect re-sends
+    /// so Origin can deduplicate rather than double-apply.
+    pub fn set_pending_delta_seq(&mut self, mutation_id: u64, seq: u64) {
+        if let Some(d) = self
+            .pending_deltas
+            .iter_mut()
+            .find(|d| d.mutation_id == mutation_id)
+            && d.seq == 0
+        {
+            d.seq = seq;
+        }
+    }
+
     /// Mark deltas as acknowledged by Origin (after DeltaAck received).
     ///
     /// Removes all pending deltas with `mutation_id <= acked_id`.
@@ -685,6 +711,7 @@ impl CrdtEngine {
             collection: collection.to_string(),
             document_id: document_id.to_string(),
             delta_bytes,
+            seq: 0,
         });
         Ok(())
     }

nodedb-lite/src/engine/fts/checkpoint.rs

@@ -338,6 +338,8 @@ where
         let idx = FtsIndex::new(backend);
 
         // ── Posting data: try pagedb segment path first, fall back to KV ─────
+        // Flipped only by the native segment path, compiled out on wasm32.
+        #[cfg_attr(target_arch = "wasm32", allow(unused_mut))]
         let mut restored_from_segment = false;
 
         #[cfg(not(target_arch = "wasm32"))]

nodedb-lite/src/engine/fts/manager.rs

@@ -362,17 +362,17 @@ impl FtsCollectionManager {
         &mut self,
         collection: &str,
         origin_surrogate: Surrogate,
-    ) -> bool {
+    ) -> Option<String> {
         let Some(doc_id) = self.origin_surrogate_to_doc_id.remove(&origin_surrogate.0) else {
             tracing::debug!(
                 collection,
                 sur = origin_surrogate.0,
                 "FtsDeleteDoc: no Lite mapping for Origin surrogate — document was never indexed here"
             );
-            return false;
+            return None;
         };
         self.remove_document(collection, &doc_id);
-        true
+        Some(doc_id)
     }
 
     // ── Per-field indexing (used by strict collections via index_integration) ─
@@ -562,7 +562,7 @@ mod tests {
 
         // Delete via origin surrogate.
         let removed = mgr.remove_by_origin_surrogate("col", Surrogate(42));
-        assert!(removed, "doc2 must be found and removed");
+        assert!(removed.is_some(), "doc2 must be found and removed");
 
         // doc1 and doc3 still searchable, doc2 not.
         let results = mgr.search("col", "rust", 10, &default_params());
@@ -612,7 +612,7 @@ mod tests {
         mgr.index_document("col", "doc1", "hello world");
 
         let removed = mgr.remove_by_origin_surrogate("col", Surrogate(99));
-        assert!(!removed, "unknown surrogate must return false");
+        assert!(removed.is_none(), "unknown surrogate must return None");
 
         // doc1 unaffected.
         let results = mgr.search("col", "hello", 10, &default_params());

nodedb-lite/src/engine/timeseries/engine/sync.rs

@@ -128,6 +128,9 @@ impl TimeseriesEngine {
             min_ts,
             max_ts,
             watermarks,
+            producer_id: 0,
+            epoch: 0,
+            seq: 0,
         })
     }
 

nodedb-lite/src/engine/vector/search/lazy_load.rs

@@ -49,6 +49,8 @@ pub(super) async fn ensure_index_loaded<S: StorageEngine>(
         return Ok(());
     };
 
+    // `index` is mutated only by the native segment-backing path (wasm32: none).
+    #[cfg_attr(target_arch = "wasm32", allow(unused_mut))]
     let Ok(Some(mut index)) = HnswIndex::from_checkpoint(&checkpoint) else {
         return Ok(());
     };

nodedb-lite/src/lib.rs

@@ -1,3 +1,33 @@
+//! # NodeDB-Lite
+//!
+//! Embedded, offline-first build of NodeDB for phones, browsers (WASM), and
+//! desktops. A single in-process library exposing the same [`NodeDb`] trait as
+//! the Origin server — document, key-value, vector, graph, full-text, spatial,
+//! columnar, timeseries, and array engines over one storage core — with CRDT
+//! sync to an Origin server over WebSocket.
+//!
+//! ## Quick start
+//!
+//! ```no_run
+//! use nodedb_lite::{NodeDbLite, PagedbStorageMem};
+//! use nodedb_client::NodeDb;
+//!
+//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
+//! let storage = PagedbStorageMem::open_in_memory().await?;
+//! let db = NodeDbLite::open(storage, 1u64).await?;
+//! db.execute_sql("CREATE COLLECTION notes", &[]).await?;
+//! # Ok(())
+//! # }
+//! ```
+//!
+//! ## Durability
+//!
+//! Writes are buffered for batching; `await` returning `Ok` does **not** by
+//! itself guarantee on-disk durability. Durability is bounded by the
+//! [`config::LiteConfig::auto_flush_ms`] background flush interval, or forced
+//! by an explicit [`NodeDbLite::flush`]. For at-rest encryption see
+//! [`Encryption`]. [`NodeDb`]: nodedb_client::NodeDb
+
 pub mod config;
 pub mod engine;
 pub mod error;
@@ -17,6 +47,7 @@ pub use memory::MemoryGovernor;
 pub use nodedb::{BatchItem, NodeDbLite, SyncGate};
 pub use nodedb_query;
 pub use nodedb_types::id_gen;
+pub use storage::encryption::Encryption;
 pub use storage::engine::{StorageEngine, WriteOp};
 #[cfg(not(target_arch = "wasm32"))]
 pub use storage::pagedb_storage::PagedbStorageDefault;