nodedb-client 0.3.0

// SPDX-License-Identifier: Apache-2.0

//! The `NodeDb` trait: unified query interface for both Origin and Lite.
//!
//! Application code writes against this trait once. The runtime determines
//! whether queries execute locally (in-memory engines on Lite) or remotely
//! (pgwire to Origin).
//!
//! All methods are `async` — on native this runs on Tokio, on WASM this
//! runs on `wasm-bindgen-futures`.
//!
//! This file must remain a single block for object-safety: Rust does not
//! permit a `trait` body to be split across files. Splitting `NodeDb` into
//! supertraits would break the `Arc<dyn NodeDb>` pattern all callers depend
//! on and is therefore out of scope for any mechanical refactor.

use std::collections::HashSet;

use async_trait::async_trait;

use nodedb_types::document::Document;
use nodedb_types::dropped_collection::DroppedCollection;
use nodedb_types::error::{NodeDbError, NodeDbResult};
use nodedb_types::filter::{EdgeFilter, MetadataFilter};
use nodedb_types::graph::GraphStats;
use nodedb_types::id::{EdgeId, NodeId};
use nodedb_types::protocol::Limits;
use nodedb_types::result::{QueryResult, SearchResult, SubGraph};
use nodedb_types::text_search::TextSearchParams;
use nodedb_types::value::Value;

use super::marker::NodeDbMarker;
use super::quote::quote_ident;
use crate::traits::document::CollectionPurgedHandler;

/// Unified database interface for NodeDB.
///
/// Two implementations:
/// - `NodeDbLite`: executes queries against in-memory HNSW/CSR/Loro engines
///   on the edge device. Writes produce CRDT deltas synced to Origin in background.
/// - `NodeDbRemote`: translates trait calls into parameterized SQL and sends
///   them over pgwire to the Origin cluster.
///
/// The developer writes agent logic once. Switching between local and cloud
/// is a one-line configuration change.
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait NodeDb: NodeDbMarker {
    // ─── Vector Operations ───────────────────────────────────────────

    /// Search for the `k` nearest vectors to `query` in `collection`.
    ///
    /// Returns results ordered by ascending distance. Optional metadata
    /// filter constrains which vectors are considered. When `allowed_ids`
    /// is `Some`, only documents whose string ID appears in the set are
    /// eligible — the filter is pushed into HNSW traversal on Lite so
    /// the returned top-k is drawn exclusively from the allowed set.
    ///
    /// On Lite: direct in-memory HNSW search with optional ID prefilter. Sub-millisecond.
    /// On Remote: translated to `SELECT ... ORDER BY embedding <-> $1 LIMIT $2`
    /// (allowed_ids is ignored on the remote path — pass `None`).
    async fn vector_search(
        &self,
        collection: &str,
        query: &[f32],
        k: usize,
        filter: Option<&MetadataFilter>,
        allowed_ids: Option<&HashSet<String>>,
    ) -> NodeDbResult<Vec<SearchResult>>;

    /// Insert a vector with optional metadata into `collection`.
    ///
    /// On Lite: inserts into in-memory HNSW + emits CRDT delta + persists to SQLite.
    /// On Remote: translated to `INSERT INTO collection (id, embedding, metadata) VALUES (...)`.
    async fn vector_insert(
        &self,
        collection: &str,
        id: &str,
        embedding: &[f32],
        metadata: Option<Document>,
    ) -> NodeDbResult<()>;

    /// Delete a vector by ID from `collection`.
    ///
    /// On Lite: marks deleted in HNSW + emits CRDT tombstone.
    /// On Remote: `DELETE FROM collection WHERE id = $1`.
    async fn vector_delete(&self, collection: &str, id: &str) -> NodeDbResult<()>;

    // ─── Graph Operations ────────────────────────────────────────────

    /// Traverse the graph from `start` up to `depth` hops within
    /// `collection`.
    ///
    /// `collection` names the graph collection holding the adjacency
    /// data. NodeDB's graph overlay scopes edges per collection, so the
    /// caller picks which graph to walk. Returns the discovered subgraph
    /// (nodes + edges). Optional edge filter constrains which edges are
    /// followed.
    ///
    /// On Lite: direct CSR pointer-chasing in contiguous memory. Microseconds.
    /// On Remote: `GRAPH TRAVERSE FROM '<start>' DEPTH <n> [LABEL '<l>']`.
    async fn graph_traverse(
        &self,
        collection: &str,
        start: &NodeId,
        depth: u8,
        edge_filter: Option<&EdgeFilter>,
    ) -> NodeDbResult<SubGraph>;

    /// Insert a directed edge from `from` to `to` with the given label
    /// into `collection`.
    ///
    /// Returns the generated edge ID.
    ///
    /// On Lite: appends to mutable adjacency buffer + CRDT delta + SQLite.
    /// On Remote: `GRAPH INSERT EDGE IN '<collection>' FROM '<from>' TO '<to>' TYPE '<label>'`.
    async fn graph_insert_edge(
        &self,
        collection: &str,
        from: &NodeId,
        to: &NodeId,
        edge_type: &str,
        properties: Option<Document>,
    ) -> NodeDbResult<EdgeId>;

    /// Delete a graph edge by ID from `collection`.
    ///
    /// On Lite: marks deleted + CRDT tombstone.
    /// On Remote: `GRAPH DELETE EDGE IN '<collection>' FROM '<src>' TO '<dst>' TYPE '<label>'`.
    async fn graph_delete_edge(&self, collection: &str, edge_id: &EdgeId) -> NodeDbResult<()>;

    /// Read aggregated graph statistics.
    ///
    /// When `collection` is `Some(name)`, returns statistics for that one
    /// collection — a vec of length 0 (no edges recorded) or 1. When
    /// `collection` is `None`, returns one `GraphStats` per collection that
    /// has edges (tenant-wide). Each entry contains the global edge count,
    /// distinct node count, distinct label count, and per-label edge counts
    /// (sorted ascending by label name). Reads what was *persisted* in the
    /// edge store, bypassing any in-memory CSR view.
    ///
    /// `as_of` pins the read to a past system-time epoch (milliseconds
    /// since Unix epoch). When `None`, the live (current) state is
    /// returned. Bitemporal reads are supported on Origin; the Lite backend
    /// returns an error when `as_of` is `Some`.
    ///
    /// On Lite: direct read of the local edge store.
    /// On Remote: `SHOW GRAPH STATS [<'collection'>] [AS OF SYSTEM TIME <ms>]`.
    async fn graph_stats(
        &self,
        collection: Option<&str>,
        as_of: Option<i64>,
    ) -> NodeDbResult<Vec<GraphStats>>;

    /// Run PageRank on a graph collection's edges.
    ///
    /// When `personalization` is `Some`, computes Personalized PageRank — initial
    /// rank is biased toward the seed nodes in the map (values are normalized).
    /// When `None`, runs standard uniform-init PageRank.
    ///
    /// `damping` defaults to 0.85 when `None`. `max_iterations` defaults to 20.
    ///
    /// Returns `(node_id, rank)` pairs sorted by rank descending. Ranks sum to 1.0.
    async fn graph_pagerank(
        &self,
        collection: &str,
        personalization: Option<std::collections::HashMap<String, f64>>,
        damping: Option<f64>,
        max_iterations: Option<u32>,
    ) -> NodeDbResult<Vec<(String, f64)>> {
        let _ = (collection, personalization, damping, max_iterations);
        Err(NodeDbError::storage(
            "graph_pagerank is not implemented for this NodeDb backend",
        ))
    }

    // ─── Document Operations ─────────────────────────────────────────

    /// Get a document by ID from `collection`.
    ///
    /// On Lite: direct Loro state read. Sub-millisecond.
    /// On Remote: `SELECT * FROM collection WHERE id = $1`.
    async fn document_get(&self, collection: &str, id: &str) -> NodeDbResult<Option<Document>>;

    /// Put (insert or update) a document into `collection`.
    ///
    /// The document's `id` field determines the key. If a document with that
    /// ID already exists, it is overwritten (last-writer-wins locally; CRDT
    /// merge on sync).
    ///
    /// On Lite: Loro apply + CRDT delta + SQLite persist.
    /// On Remote: `INSERT ... ON CONFLICT (id) DO UPDATE SET ...`.
    async fn document_put(&self, collection: &str, doc: Document) -> NodeDbResult<()>;

    /// Delete a document by ID from `collection`.
    ///
    /// On Lite: Loro delete + CRDT tombstone.
    /// On Remote: `DELETE FROM collection WHERE id = $1`.
    async fn document_delete(&self, collection: &str, id: &str) -> NodeDbResult<()>;

    /// Put a document and insert its embedding vector in a single CRDT lock acquisition.
    ///
    /// Equivalent to calling `document_put` then `vector_insert`, but acquires the
    /// CRDT lock only once — halving the per-insert oplog-walk cost on Lite. When
    /// `embedding` is empty this behaves identically to `document_put`.
    ///
    /// `vector_collection` names the vector index to insert into (may differ from
    /// the document `collection`).
    ///
    /// The default implementation falls back to two separate calls. Implementations
    /// that can batch the two operations under one lock should override this method.
    async fn document_put_with_vector(
        &self,
        doc_collection: &str,
        doc: Document,
        vector_collection: &str,
        id: &str,
        embedding: &[f32],
    ) -> NodeDbResult<()> {
        self.document_put(doc_collection, doc).await?;
        if !embedding.is_empty() {
            self.vector_insert(vector_collection, id, embedding, None)
                .await?;
        }
        Ok(())
    }

    /// Read a document as-of a system time, optionally filtered by valid_time.
    ///
    /// Only valid on collections created `WITH (bitemporal=true)` — returns an
    /// error on plain (non-bitemporal) collections.
    ///
    /// When `as_of_ms` is `None`, returns the current LIVE version (equivalent
    /// to `document_get`). When `as_of_ms` is `Some(t)`, returns the version
    /// visible at system time `t`. If `valid_time_ms` is `Some(vt)`, the
    /// returned version must additionally satisfy
    /// `valid_from_ms <= vt < valid_until_ms`.
    ///
    /// Returns `Err` on implementations that do not support bitemporal reads.
    async fn document_get_as_of(
        &self,
        collection: &str,
        id: &str,
        as_of_ms: Option<i64>,
        valid_time_ms: Option<i64>,
    ) -> NodeDbResult<Option<Document>> {
        let _ = (collection, id, as_of_ms, valid_time_ms);
        Err(NodeDbError::storage(
            "document_get_as_of is not implemented on this client",
        ))
    }

    /// Put a document with explicit valid-time bounds.
    ///
    /// Only valid on collections created `WITH (bitemporal=true)`.
    /// `valid_from_ms` and `valid_until_ms` specify the application-time
    /// interval for which the version is considered current.  Both default
    /// to system time / open-ended when `None`.
    ///
    /// Returns `Err` on implementations that do not support bitemporal writes.
    async fn document_put_with_valid_time(
        &self,
        collection: &str,
        doc: Document,
        valid_from_ms: Option<i64>,
        valid_until_ms: Option<i64>,
    ) -> NodeDbResult<()> {
        let _ = (collection, doc, valid_from_ms, valid_until_ms);
        Err(NodeDbError::storage(
            "document_put_with_valid_time is not implemented on this client",
        ))
    }

    // ─── Named Vector Fields ──────────────────────────────────────────

    /// Insert a vector into a named field within a collection.
    ///
    /// Enables multiple embeddings per collection (e.g., "title_embedding",
    /// "body_embedding") with independent HNSW indexes.
    ///
    /// Default returns `Err` — silently delegating to `vector_insert` and
    /// dropping `field_name` would land the vector in the wrong field.
    /// Implementations that route through to a server with field-aware
    /// support must override.
    async fn vector_insert_field(
        &self,
        collection: &str,
        field_name: &str,
        id: &str,
        embedding: &[f32],
        metadata: Option<Document>,
    ) -> NodeDbResult<()> {
        let _ = (collection, id, embedding, metadata);
        Err(NodeDbError::storage(format!(
            "vector_insert_field is not implemented on this client; \
             field_name={field_name} would have been silently dropped"
        )))
    }

    /// Search a named vector field.
    ///
    /// Default returns `Err` — silently delegating to `vector_search`
    /// and dropping `field_name` would search the wrong field.
    /// Implementations that route through to a server with field-aware
    /// support must override.
    async fn vector_search_field(
        &self,
        collection: &str,
        field_name: &str,
        query: &[f32],
        k: usize,
        filter: Option<&MetadataFilter>,
    ) -> NodeDbResult<Vec<SearchResult>> {
        let _ = (collection, query, k, filter);
        Err(NodeDbError::storage(format!(
            "vector_search_field is not implemented on this client; \
             field_name={field_name} would have been silently dropped"
        )))
    }

    // ─── Graph Shortest Path ────────────────────────────────────────

    /// Find the shortest path between two nodes.
    ///
    /// Returns the path as a list of node IDs (`from` first, `to` last),
    /// or `None` if no path exists within `max_depth` hops.
    ///
    /// Default: forward breadth-first search built on `graph_traverse`.
    /// Each frontier expansion calls `graph_traverse(node, 1,
    /// edge_filter)` to discover outgoing neighbors. Inherits the
    /// underlying impl's edge direction semantics. Implementations with
    /// a server-side shortest-path operator (e.g. NodeDB's
    /// `GRAPH PATH FROM <src> TO <dst>` DSL) should override for
    /// performance — round-tripping per-hop is O(path_length) wire
    /// hops.
    async fn graph_shortest_path(
        &self,
        collection: &str,
        from: &NodeId,
        to: &NodeId,
        max_depth: u8,
        edge_filter: Option<&EdgeFilter>,
    ) -> NodeDbResult<Option<Vec<NodeId>>> {
        if from == to {
            return Ok(Some(vec![from.clone()]));
        }
        if max_depth == 0 {
            return Ok(None);
        }

        // Map of `node -> parent` used to reconstruct the path once the
        // target is reached. The source has no parent entry.
        let mut parent: std::collections::HashMap<NodeId, NodeId> =
            std::collections::HashMap::new();
        let mut frontier: Vec<NodeId> = vec![from.clone()];

        for _ in 0..max_depth {
            let mut next_frontier: Vec<NodeId> = Vec::new();
            for node in &frontier {
                let sg = self
                    .graph_traverse(collection, node, 1, edge_filter)
                    .await?;
                for edge in &sg.edges {
                    // Only follow edges originating from the current
                    // node — `graph_traverse` may include adjacent
                    // edges that don't extend the BFS frontier.
                    if &edge.from != node {
                        continue;
                    }
                    let dst = &edge.to;
                    if dst == from || parent.contains_key(dst) {
                        continue;
                    }
                    parent.insert(dst.clone(), node.clone());
                    if dst == to {
                        let mut path = vec![to.clone()];
                        let mut cur = to.clone();
                        while &cur != from {
                            let p = parent
                                .get(&cur)
                                .expect("BFS reached `to` so all ancestors are tracked")
                                .clone();
                            path.push(p.clone());
                            cur = p;
                        }
                        path.reverse();
                        return Ok(Some(path));
                    }
                    next_frontier.push(dst.clone());
                }
            }
            if next_frontier.is_empty() {
                return Ok(None);
            }
            frontier = next_frontier;
        }
        Ok(None)
    }

    // ─── Text Search ────────────────────────────────────────────────

    /// Full-text search with BM25 scoring against the FTS-indexed
    /// `field` on `collection`.
    ///
    /// NodeDB's FTS is per-field — every BM25 index is scoped to one
    /// declared field, so the caller names which field to search.
    /// Returns document IDs with relevance scores, ordered by
    /// descending score. Pass [`TextSearchParams::default()`] for
    /// standard OR-mode non-fuzzy search.
    ///
    /// Default returns `Err` — `Ok(Vec::new())` is indistinguishable
    /// from a real "no matches" answer and would silently mask the
    /// missing implementation. Implementations must override (e.g., a
    /// `SEARCH IN '<collection>' FIELD '<field>' QUERY '<q>'` round-trip
    /// via `execute_sql`).
    /// Full-text BM25 search. When `allowed_ids` is `Some`, only documents
    /// whose ID is in the set are returned. On Lite, the filter is applied
    /// after an over-fetch; on Remote, `allowed_ids` is ignored (pass `None`).
    async fn text_search(
        &self,
        collection: &str,
        field: &str,
        query: &str,
        top_k: usize,
        params: TextSearchParams,
        allowed_ids: Option<&HashSet<String>>,
    ) -> NodeDbResult<Vec<SearchResult>> {
        let _ = (collection, field, query, top_k, params, allowed_ids);
        Err(NodeDbError::storage(
            "text_search is not implemented on this client",
        ))
    }

    // ─── Batch Operations ───────────────────────────────────────────

    /// Batch insert vectors — amortizes CRDT delta export to O(1) per batch.
    async fn batch_vector_insert(
        &self,
        collection: &str,
        vectors: &[(&str, &[f32])],
    ) -> NodeDbResult<()> {
        for &(id, embedding) in vectors {
            self.vector_insert(collection, id, embedding, None).await?;
        }
        Ok(())
    }

    /// Batch insert graph edges into `collection` — amortizes CRDT
    /// delta export to O(1) per batch.
    async fn batch_graph_insert_edges(
        &self,
        collection: &str,
        edges: &[(&str, &str, &str)],
    ) -> NodeDbResult<()> {
        for &(from, to, label) in edges {
            let src = NodeId::try_new(from)
                .map_err(|e| NodeDbError::storage(format!("invalid node id: {e}")))?;
            let dst = NodeId::try_new(to)
                .map_err(|e| NodeDbError::storage(format!("invalid node id: {e}")))?;
            self.graph_insert_edge(collection, &src, &dst, label, None)
                .await?;
        }
        Ok(())
    }

    // ─── Connection Metadata ─────────────────────────────────────────────

    /// The protocol version negotiated during the connection handshake.
    ///
    /// Returns `0` for implementations that do not maintain a persistent
    /// connection and therefore never perform a handshake.
    fn proto_version(&self) -> u16 {
        0
    }

    /// The raw capability bitfield advertised by the server.
    ///
    /// Returns `0` when no handshake was performed. Use
    /// `Capabilities::from_raw(self.capabilities())` for named predicates.
    fn capabilities(&self) -> u64 {
        0
    }

    /// The server version string from `HelloAckFrame` (e.g. `"0.1.0-dev"`).
    ///
    /// Returns an empty string when no handshake was performed.
    fn server_version(&self) -> String {
        String::new()
    }

    /// Per-operation limits announced by the server.
    ///
    /// All fields are `None` when no handshake was performed — the caller
    /// should treat `None` as "no server-side cap" for that dimension.
    fn limits(&self) -> Limits {
        Limits::default()
    }

    // ─── SQL Escape Hatch ────────────────────────────────────────────

    /// Execute a raw SQL query with parameters.
    ///
    /// On Lite: requires the `sql` feature flag (compiles in DataFusion parser).
    ///   Returns `NodeDbError::SqlNotEnabled` if the feature is not compiled in.
    /// On Remote: pass-through to Origin via pgwire.
    ///
    /// For most AI agent workloads, the typed methods above are sufficient
    /// and faster. Use this for BI tools, existing ORMs, or ad-hoc queries.
    async fn execute_sql(&self, query: &str, params: &[Value]) -> NodeDbResult<QueryResult>;

    // ─── Collection Lifecycle (soft-delete / undrop / hard-delete) ───

    /// Restore a soft-deleted collection within its retention window.
    ///
    /// Equivalent to `UNDROP COLLECTION <name>`. Fails with 42P01 if
    /// the retention window has elapsed and the row is gone, or with
    /// 42501 if the caller is neither preserved owner nor admin.
    ///
    /// Default impl routes through `execute_sql` so any implementation
    /// that can execute SQL inherits the correct behavior for free.
    async fn undrop_collection(&self, name: &str) -> NodeDbResult<()> {
        let sql = format!("UNDROP COLLECTION {}", quote_ident(name));
        self.execute_sql(&sql, &[]).await?;
        Ok(())
    }

    /// Hard-delete a collection, skipping soft-delete and retention.
    ///
    /// Equivalent to `DROP COLLECTION <name> PURGE`. Admin-only on the
    /// server; the server rejects non-admin callers with 42501.
    /// Bypasses the retention safety net — data is unrecoverable.
    async fn drop_collection_purge(&self, name: &str) -> NodeDbResult<()> {
        let sql = format!("DROP COLLECTION {} PURGE", quote_ident(name));
        self.execute_sql(&sql, &[]).await?;
        Ok(())
    }

    /// List every soft-deleted collection in the current tenant that
    /// is still within its retention window.
    ///
    /// Equivalent to `SELECT tenant_id, name, owner, deactivated_at_ns,
    /// retention_expires_at_ns FROM _system.dropped_collections`.
    /// Returns `Vec<DroppedCollection>` — empty if no soft-deleted rows
    /// exist for the caller's tenant.
    async fn list_dropped_collections(&self) -> NodeDbResult<Vec<DroppedCollection>> {
        let sql = "SELECT tenant_id, name, owner, engine_type, \
                   deactivated_at_ns, retention_expires_at_ns \
                   FROM _system.dropped_collections";
        let result = self.execute_sql(sql, &[]).await?;
        crate::row_decode::parse_dropped_collection_rows(&result.rows)
    }

    /// Register a handler fired when a collection the caller has
    /// synced is purged on Origin and the local copy is removed.
    ///
    /// Default impl returns `NodeDbError::storage` with a
    /// `"not supported"` detail — implementations that maintain a
    /// sync client (Lite, any future push-capable remote client)
    /// override with registration into their internal handler list.
    /// Stateless clients (pgwire-only `NodeDbRemote`) have nothing
    /// to push, so the default rejection is the correct behavior.
    async fn on_collection_purged(&self, _handler: CollectionPurgedHandler) -> NodeDbResult<()> {
        Err(NodeDbError::storage(
            "on_collection_purged is not supported on this client — \
             requires a push-capable sync connection (NodeDbLite or a \
             sync-enabled remote client)",
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::capabilities::Capabilities;
    use async_trait::async_trait;
    use nodedb_types::document::Document;
    use nodedb_types::error::{NodeDbError, NodeDbResult};
    use nodedb_types::filter::{EdgeFilter, MetadataFilter};
    use nodedb_types::graph::GraphStats;
    use nodedb_types::id::{EdgeId, NodeId};
    use nodedb_types::result::{QueryResult, SearchResult, SubGraph};
    use nodedb_types::value::Value;
    use std::collections::HashMap;

    /// Mock implementation to verify the trait is object-safe and
    /// can be used as `Arc<dyn NodeDb>`.
    struct MockDb;

    #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
    #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
    impl NodeDb for MockDb {
        async fn vector_search(
            &self,
            _collection: &str,
            _query: &[f32],
            _k: usize,
            _filter: Option<&MetadataFilter>,
            _allowed_ids: Option<&HashSet<String>>,
        ) -> NodeDbResult<Vec<SearchResult>> {
            Ok(vec![SearchResult {
                id: "vec-1".into(),
                node_id: None,
                distance: 0.1,
                metadata: HashMap::new(),
            }])
        }

        async fn vector_insert(
            &self,
            _collection: &str,
            _id: &str,
            _embedding: &[f32],
            _metadata: Option<Document>,
        ) -> NodeDbResult<()> {
            Ok(())
        }

        async fn vector_delete(&self, _collection: &str, _id: &str) -> NodeDbResult<()> {
            Ok(())
        }

        async fn graph_traverse(
            &self,
            _collection: &str,
            _start: &NodeId,
            _depth: u8,
            _edge_filter: Option<&EdgeFilter>,
        ) -> NodeDbResult<SubGraph> {
            Ok(SubGraph::empty())
        }

        async fn graph_insert_edge(
            &self,
            _collection: &str,
            from: &NodeId,
            to: &NodeId,
            edge_type: &str,
            _properties: Option<Document>,
        ) -> NodeDbResult<EdgeId> {
            EdgeId::try_first(from.clone(), to.clone(), edge_type)
                .map_err(|e| NodeDbError::storage(format!("invalid edge label: {e}")))
        }

        async fn graph_delete_edge(
            &self,
            _collection: &str,
            _edge_id: &EdgeId,
        ) -> NodeDbResult<()> {
            Ok(())
        }

        async fn graph_stats(
            &self,
            collection: Option<&str>,
            _as_of: Option<i64>,
        ) -> NodeDbResult<Vec<GraphStats>> {
            Ok(vec![GraphStats::zero(collection.unwrap_or("mock"))])
        }

        async fn document_get(
            &self,
            _collection: &str,
            id: &str,
        ) -> NodeDbResult<Option<Document>> {
            let mut doc = Document::new(id);
            doc.set("title", Value::String("test".into()));
            Ok(Some(doc))
        }

        async fn document_put(&self, _collection: &str, _doc: Document) -> NodeDbResult<()> {
            Ok(())
        }

        async fn document_delete(&self, _collection: &str, _id: &str) -> NodeDbResult<()> {
            Ok(())
        }

        async fn execute_sql(&self, _query: &str, _params: &[Value]) -> NodeDbResult<QueryResult> {
            Ok(QueryResult::empty())
        }
    }

    #[test]
    fn trait_is_object_safe() {
        fn _accepts_dyn(_db: &dyn NodeDb) {}
        let db = MockDb;
        _accepts_dyn(&db);
    }

    #[test]
    fn trait_works_with_arc() {
        use std::sync::Arc;
        let db: Arc<dyn NodeDb> = Arc::new(MockDb);
        let _ = db;
    }

    #[tokio::test]
    async fn mock_vector_search() {
        let db = MockDb;
        let results = db
            .vector_search("embeddings", &[0.1, 0.2, 0.3], 5, None, None)
            .await
            .unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].id, "vec-1");
        assert!(results[0].distance < 1.0);
    }

    #[tokio::test]
    async fn mock_vector_insert_and_delete() {
        let db = MockDb;
        db.vector_insert("coll", "v1", &[1.0, 2.0], None)
            .await
            .unwrap();
        db.vector_delete("coll", "v1").await.unwrap();
    }

    #[tokio::test]
    async fn mock_graph_stats_returns_zero() {
        let db = MockDb;
        let result = db.graph_stats(Some("social"), None).await.unwrap();
        assert_eq!(result.len(), 1);
        let stats = &result[0];
        assert_eq!(stats.collection, "social");
        assert_eq!(stats.node_count, 0);
        assert_eq!(stats.edge_count, 0);
        assert_eq!(stats.distinct_label_count, 0);
        assert!(stats.labels.is_empty());
    }

    #[tokio::test]
    async fn mock_graph_stats_tenant_wide_uses_mock_key() {
        let db = MockDb;
        let result = db.graph_stats(None, None).await.unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].collection, "mock");
    }

    #[tokio::test]
    async fn mock_graph_operations() {
        let db = MockDb;
        let start = NodeId::try_new("alice").expect("test fixture");
        let subgraph = db.graph_traverse("social", &start, 2, None).await.unwrap();
        assert_eq!(subgraph.node_count(), 0);

        let from = NodeId::try_new("alice").expect("test fixture");
        let to = NodeId::try_new("bob").expect("test fixture");
        let edge_id = db
            .graph_insert_edge("social", &from, &to, "KNOWS", None)
            .await
            .unwrap();
        assert_eq!(edge_id.src.as_str(), "alice");
        assert_eq!(edge_id.dst.as_str(), "bob");
        assert_eq!(edge_id.label, "KNOWS");
        assert_eq!(edge_id.seq, 0);

        db.graph_delete_edge("social", &edge_id).await.unwrap();
    }

    #[tokio::test]
    async fn mock_document_operations() {
        let db = MockDb;
        let doc = db.document_get("notes", "n1").await.unwrap().unwrap();
        assert_eq!(doc.id, "n1");
        assert_eq!(doc.get_str("title"), Some("test"));

        let mut new_doc = Document::new("n2");
        new_doc.set("body", Value::String("hello".into()));
        db.document_put("notes", new_doc).await.unwrap();

        db.document_delete("notes", "n1").await.unwrap();
    }

    #[tokio::test]
    async fn mock_execute_sql() {
        let db = MockDb;
        let result = db.execute_sql("SELECT 1", &[]).await.unwrap();
        assert_eq!(result.row_count(), 0);
    }

    /// Verify the full "one API, any runtime" pattern: application
    /// code switches between `NodeDbLite` and `NodeDbRemote` only at
    /// the construction site.
    #[tokio::test]
    async fn unified_api_pattern() {
        use std::sync::Arc;

        let db: Arc<dyn NodeDb> = Arc::new(MockDb);

        let results = db
            .vector_search("knowledge_base", &[0.1, 0.2], 5, None, None)
            .await
            .unwrap();
        assert!(!results.is_empty());

        let start = NodeId::from_validated(results[0].id.clone());
        let _subgraph = db
            .graph_traverse("knowledge_base", &start, 2, None)
            .await
            .unwrap();

        let doc = Document::new("note-1");
        db.document_put("notes", doc).await.unwrap();
    }

    #[test]
    fn default_proto_version_is_zero() {
        let db = MockDb;
        assert_eq!(db.proto_version(), 0);
    }

    #[test]
    fn default_capabilities_is_zero() {
        let db = MockDb;
        assert_eq!(db.capabilities(), 0);
        let caps = Capabilities::from_raw(db.capabilities());
        assert!(!caps.supports_streaming());
        assert!(!caps.supports_graphrag());
    }

    #[test]
    fn default_server_version_is_empty() {
        let db = MockDb;
        assert!(db.server_version().is_empty());
    }

    #[test]
    fn default_limits_all_none() {
        let db = MockDb;
        let limits = db.limits();
        assert!(limits.max_vector_dim.is_none());
        assert!(limits.max_top_k.is_none());
        assert!(limits.max_scan_limit.is_none());
        assert!(limits.max_batch_size.is_none());
        assert!(limits.max_crdt_delta_bytes.is_none());
        assert!(limits.max_query_text_bytes.is_none());
        assert!(limits.max_graph_depth.is_none());
    }

    #[test]
    fn capabilities_newtype_smoke() {
        use nodedb_types::protocol::{CAP_FTS, CAP_STREAMING};
        let caps = Capabilities::from_raw(CAP_STREAMING | CAP_FTS);
        assert!(caps.supports_streaming());
        assert!(caps.supports_fts());
        assert!(!caps.supports_graphrag());
        assert!(!caps.supports_crdt());
    }
}