Skip to main content

rto_graph/
store.rs

1//! SQLite-backed graph store.
2
3use std::path::Path;
4
5use rusqlite::{Connection, OptionalExtension, params};
6
7use crate::migrations;
8use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
9use crate::provenance::Provenance;
10
11/// Errors raised by the store.
12#[derive(Debug, thiserror::Error)]
13pub enum StoreError {
14    /// Underlying `SQLite` failure.
15    #[error("sqlite error: {0}")]
16    Sqlite(#[from] rusqlite::Error),
17    /// A node's `meta` could not be (de)serialized as JSON.
18    #[error("json error: {0}")]
19    Json(#[from] serde_json::Error),
20    /// An edge referenced a node key that does not exist in the store.
21    #[error("unknown node key: {0}")]
22    UnknownNode(String),
23    /// An edge violated the provenance/confidence invariant.
24    #[error("invalid edge: {0}")]
25    InvalidEdge(String),
26    /// A stored value could not be interpreted (database corruption).
27    #[error("corrupt store: {0}")]
28    Corrupt(String),
29}
30
31/// A summary of applying/re-applying import layers (see
32/// [`Store::apply_import_layer`] and [`Store::reapply_imports`]).
33#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
34pub struct ImportApplied {
35    /// Number of import layers processed.
36    pub layers: usize,
37    /// Import nodes upserted (across all layers).
38    pub nodes: usize,
39    /// Import edges applied — both endpoints resolved. A duplicate of an
40    /// already-present edge is a harmless no-op but still counted as applied.
41    pub edges_applied: usize,
42    /// Import edges **pruned**: an endpoint was absent (a cross-reference to code
43    /// that no longer exists), so the edge was dropped from the persisted layer
44    /// rather than kept as stale data.
45    pub edges_pruned: usize,
46}
47
48/// Qualified node columns for `SELECT`s that alias the `nodes` table as `n`.
49const NODE_COLS: &str = "n.key, n.kind, n.name, n.path, n.lang, n.blob_hash, n.span_start, n.span_end, n.provenance, n.meta";
50
51/// `SELECT` prefix that yields an [`Edge`] row (endpoints resolved back to keys).
52const EDGE_SELECT: &str = "SELECT ns.key AS src, nd.key AS dst, e.kind, e.provenance, \
53     e.confidence, e.src_ref \
54     FROM edges e JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst";
55
56/// A Roteiro graph store backed by a single `SQLite` database.
57pub struct Store {
58    conn: Connection,
59}
60
61impl Store {
62    /// Open (creating if absent) a store at `path` and apply pending migrations.
63    ///
64    /// # Errors
65    /// Returns [`StoreError::Sqlite`] if the database cannot be opened or a
66    /// migration fails.
67    pub fn open(path: &Path) -> Result<Self, StoreError> {
68        let conn = Connection::open(path)?;
69        Self::from_conn(conn)
70    }
71
72    /// Open an in-memory store (tests, previews).
73    ///
74    /// # Errors
75    /// Returns [`StoreError::Sqlite`] if a migration fails.
76    pub fn open_in_memory() -> Result<Self, StoreError> {
77        let conn = Connection::open_in_memory()?;
78        Self::from_conn(conn)
79    }
80
81    fn from_conn(mut conn: Connection) -> Result<Self, StoreError> {
82        conn.execute_batch("PRAGMA foreign_keys = ON;")?;
83        // Wait briefly for a concurrent writer instead of failing a read with
84        // `database is locked`. Matters for workspace `serve` (ADR-0008), where a
85        // long-lived server reads a project's graph while that repo's own
86        // `roteiro sync` commits an update to the same file. Syncs are
87        // sub-second, so this only ever costs a short wait, never a lost query.
88        conn.busy_timeout(std::time::Duration::from_secs(5))?;
89        migrations::apply(&mut conn)?;
90        Ok(Self { conn })
91    }
92
93    /// The schema version this store has been migrated to.
94    ///
95    /// # Errors
96    /// Returns [`StoreError::Sqlite`] on query failure.
97    pub fn schema_version(&self) -> Result<u32, StoreError> {
98        let v: i64 = self.conn.query_row(
99            "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
100            [],
101            |r| r.get(0),
102        )?;
103        Ok(u32::try_from(v).unwrap_or(0))
104    }
105
106    /// Number of nodes currently in the store.
107    ///
108    /// # Errors
109    /// Returns [`StoreError::Sqlite`] on query failure.
110    pub fn node_count(&self) -> Result<u64, StoreError> {
111        let n: i64 = self
112            .conn
113            .query_row("SELECT COUNT(*) FROM nodes", [], |r| r.get(0))?;
114        Ok(u64::try_from(n).unwrap_or(0))
115    }
116
117    /// Number of edges currently in the store.
118    ///
119    /// # Errors
120    /// Returns [`StoreError::Sqlite`] on query failure.
121    pub fn edge_count(&self) -> Result<u64, StoreError> {
122        let n: i64 = self
123            .conn
124            .query_row("SELECT COUNT(*) FROM edges", [], |r| r.get(0))?;
125        Ok(u64::try_from(n).unwrap_or(0))
126    }
127
128    /// Insert or update a node, keyed by its natural [`Node::key`].
129    ///
130    /// # Errors
131    /// Returns [`StoreError::Json`] if `meta` cannot be serialized, or
132    /// [`StoreError::Sqlite`] on write failure.
133    pub fn upsert_node(&self, node: &Node) -> Result<(), StoreError> {
134        upsert_node(&self.conn, node)
135    }
136
137    /// Insert an edge. Both endpoints must already resolve to nodes.
138    ///
139    /// # Errors
140    /// Returns [`StoreError::InvalidEdge`] if the provenance/confidence
141    /// invariant is violated, [`StoreError::UnknownNode`] if an endpoint key is
142    /// absent, or [`StoreError::Sqlite`] on write failure.
143    pub fn insert_edge(&self, edge: &Edge) -> Result<(), StoreError> {
144        insert_edge(&self.conn, edge)
145    }
146
147    /// Apply a fact set atomically: all nodes are upserted, then all edges are
148    /// inserted, in a single transaction. On any error nothing is committed.
149    ///
150    /// # Errors
151    /// Returns the first error encountered (see [`Store::upsert_node`] and
152    /// [`Store::insert_edge`]); the transaction is rolled back.
153    pub fn apply_factset(&mut self, facts: &FactSet) -> Result<(), StoreError> {
154        let tx = self.conn.transaction()?;
155        for node in &facts.nodes {
156            upsert_node(&tx, node)?;
157        }
158        for edge in &facts.edges {
159            insert_edge(&tx, edge)?;
160        }
161        tx.commit()?;
162        Ok(())
163    }
164
165    /// The `HEAD` tree id recorded at the last successful [`Store::rebuild`], if
166    /// any. Used by the sync engine to detect an unchanged tree.
167    ///
168    /// # Errors
169    /// Returns [`StoreError::Sqlite`] on query failure.
170    pub fn sync_state(&self) -> Result<Option<String>, StoreError> {
171        Ok(self
172            .conn
173            .query_row("SELECT tree FROM sync_state WHERE id = 0", [], |r| r.get(0))
174            .optional()?)
175    }
176
177    /// The extractor environment recorded with the last committed [`sync`],
178    /// `None` if unset (a legacy row, or the last sync was a worktree/index
179    /// preview). The incremental committed `sync` compares this to the current
180    /// env and falls back to a full re-extraction when they differ.
181    ///
182    /// # Errors
183    /// Returns [`StoreError::Sqlite`] on query failure.
184    pub fn sync_env(&self) -> Result<Option<String>, StoreError> {
185        Ok(self
186            .conn
187            .query_row("SELECT env FROM sync_state WHERE id = 0", [], |r| r.get(0))
188            .optional()?
189            .flatten())
190    }
191
192    /// Record the extractor environment for the current synced tree. Called by a
193    /// committed `sync` right after it writes the tree, so a later sync can decide
194    /// whether the incremental fast path is sound. A no-op if no tree is recorded.
195    ///
196    /// # Errors
197    /// Returns [`StoreError::Sqlite`] on write failure.
198    pub fn set_sync_env(&self, env: &str) -> Result<(), StoreError> {
199        self.conn
200            .execute("UPDATE sync_state SET env = ?1 WHERE id = 0", [env])?;
201        Ok(())
202    }
203
204    /// Atomically replace the entire graph with `facts`, recording `tree` as the
205    /// synced state (or clearing it when `tree` is `None`). All existing nodes
206    /// and edges are deleted first, so the store reflects exactly the given fact
207    /// set.
208    ///
209    /// Passing `None` records *no* synced tree — distinct from an empty string —
210    /// so [`Store::sync_state`] returns `None` and a later `sync` will not
211    /// spuriously short-circuit.
212    ///
213    /// # Errors
214    /// Returns the first error encountered (see [`Store::apply_factset`]); on any
215    /// error nothing is committed.
216    pub fn rebuild(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
217        let tx = self.conn.transaction()?;
218        tx.execute("DELETE FROM edges", [])?;
219        tx.execute("DELETE FROM nodes", [])?;
220        for node in &facts.nodes {
221            upsert_node(&tx, node)?;
222        }
223        for edge in &facts.edges {
224            insert_edge(&tx, edge)?;
225        }
226        write_sync_state(&tx, tree)?;
227        tx.commit()?;
228        Ok(())
229    }
230
231    /// Bring the store to exactly `facts` (as [`Store::rebuild`] does) but writing
232    /// only what **differs** instead of wiping and reinserting the whole graph —
233    /// the git-style "write only the delta". Unchanged node rows (which carry the
234    /// heavy JSON `meta`) and unchanged edge rows are left untouched; only removed
235    /// rows are deleted and new/changed rows written. The final state — nodes,
236    /// edges, and `sync_state` — is identical to `rebuild(facts, tree)`.
237    ///
238    /// Leaving unchanged edges in place means their row ids do not match a cold
239    /// rebuild's — which is safe *because* every edge query is content-ordered
240    /// (`(src, dst, kind, provenance)`, see [`Store::all_edges`]), never by row
241    /// id. So an incrementally reconciled store and a fresh rebuild return every
242    /// query identically; the delta is invisible above the storage layer.
243    ///
244    /// # Errors
245    /// Returns [`StoreError`] on a query failure; the transaction is rolled back.
246    pub fn reconcile(&mut self, facts: &FactSet, tree: Option<&str>) -> Result<(), StoreError> {
247        let current_nodes = self.all_nodes()?;
248        let current_edges = self.all_edges()?;
249        let cur_by_key: std::collections::HashMap<&str, &Node> =
250            current_nodes.iter().map(|n| (n.key.as_str(), n)).collect();
251        let new_keys: std::collections::HashSet<&str> =
252            facts.nodes.iter().map(|n| n.key.as_str()).collect();
253
254        // Edge identity is the full tuple, so a changed `confidence`/`src_ref`
255        // counts as remove-old + add-new — keeping the result identical to a
256        // wholesale rebuild, not the store's insert-time `DO NOTHING` semantics.
257        let new_edge_ids: std::collections::HashSet<EdgeId> =
258            facts.edges.iter().map(edge_identity).collect();
259        let cur_edge_ids: std::collections::HashSet<EdgeId> =
260            current_edges.iter().map(edge_identity).collect();
261
262        let tx = self.conn.transaction()?;
263        // 1. Delete removed edges first, so any node they reference can then be
264        //    dropped (edges are FK-constrained on node ids, with no cascade). A
265        //    removed node's edges are all removals, so they are gone before step 2.
266        for edge in &current_edges {
267            if !new_edge_ids.contains(&edge_identity(edge)) {
268                delete_edge(&tx, edge)?;
269            }
270        }
271        // 2. Drop nodes that no longer exist.
272        for old in &current_nodes {
273            if !new_keys.contains(old.key.as_str()) {
274                tx.execute("DELETE FROM nodes WHERE key = ?1", [&old.key])?;
275            }
276        }
277        // 3. Upsert only the nodes that are new or whose content changed (an upsert
278        //    keeps the row id, so unchanged edges stay valid).
279        for node in &facts.nodes {
280            if cur_by_key
281                .get(node.key.as_str())
282                .is_none_or(|cur| *cur != node)
283            {
284                upsert_node(&tx, node)?;
285            }
286        }
287        // 4. Insert only the added edges (their endpoints now all exist).
288        for edge in &facts.edges {
289            if !cur_edge_ids.contains(&edge_identity(edge)) {
290                insert_edge(&tx, edge)?;
291            }
292        }
293        write_sync_state(&tx, tree)?;
294        tx.commit()?;
295        Ok(())
296    }
297
298    /// Fetch a node by its natural key.
299    ///
300    /// # Errors
301    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
302    /// [`StoreError::Corrupt`] if a stored value cannot be decoded.
303    pub fn get_node(&self, key: &str) -> Result<Option<Node>, StoreError> {
304        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.key = ?1");
305        let mut stmt = self.conn.prepare(&sql)?;
306        let mut rows = stmt.query([key])?;
307        match rows.next()? {
308            Some(row) => Ok(Some(row_to_node(row)?)),
309            None => Ok(None),
310        }
311    }
312
313    /// Every node key in the store, ordered. Useful for whole-graph exports.
314    ///
315    /// # Errors
316    /// Returns [`StoreError::Sqlite`] on query failure.
317    pub fn all_keys(&self) -> Result<Vec<String>, StoreError> {
318        let mut stmt = self.conn.prepare("SELECT key FROM nodes ORDER BY key")?;
319        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
320        let mut out = Vec::new();
321        for row in rows {
322            out.push(row?);
323        }
324        Ok(out)
325    }
326
327    /// Dump the entire graph as a single [`FactSet`], with nodes and edges in a
328    /// deterministic order — suitable for a portable, content-stable artifact.
329    ///
330    /// # Errors
331    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
332    /// [`StoreError::Corrupt`] on decode failure.
333    pub fn export_factset(&self) -> Result<FactSet, StoreError> {
334        let node_sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
335        let mut node_stmt = self.conn.prepare(&node_sql)?;
336        let mut node_rows = node_stmt.query([])?;
337        let nodes = collect_nodes(&mut node_rows)?;
338
339        // Order edges by their resolved endpoint keys (not row id) so the dump is
340        // stable regardless of insertion order.
341        let edge_sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
342        let mut edge_stmt = self.conn.prepare(&edge_sql)?;
343        let mut edge_rows = edge_stmt.query([])?;
344        let edges = collect_edges(&mut edge_rows)?;
345
346        Ok(FactSet { nodes, edges })
347    }
348
349    /// All nodes of a given kind.
350    ///
351    /// # Errors
352    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
353    /// [`StoreError::Corrupt`] on decode failure.
354    pub fn nodes_by_kind(&self, kind: &NodeKind) -> Result<Vec<Node>, StoreError> {
355        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.kind = ?1 ORDER BY n.key");
356        let mut stmt = self.conn.prepare(&sql)?;
357        let mut rows = stmt.query([kind.as_str()])?;
358        collect_nodes(&mut rows)
359    }
360
361    /// Nodes of a given kind whose `name` equals `name_lower` **case-insensitively**,
362    /// ordered by key. Narrows a lookup at the SQL layer — using the `kind` index and
363    /// filtering `name` in-query — so only matching rows are decoded, never every
364    /// node of that kind. Used by the cross-repo follow bridge to fetch just the
365    /// candidate struct(s) for a config section rather than scanning all structs.
366    ///
367    /// # Errors
368    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
369    /// [`StoreError::Corrupt`] on decode failure.
370    pub fn nodes_by_kind_named(
371        &self,
372        kind: &NodeKind,
373        name_lower: &str,
374    ) -> Result<Vec<Node>, StoreError> {
375        let sql = format!(
376            "SELECT {NODE_COLS} FROM nodes n \
377             WHERE n.kind = ?1 AND lower(n.name) = ?2 ORDER BY n.key"
378        );
379        let mut stmt = self.conn.prepare(&sql)?;
380        let mut rows = stmt.query([kind.as_str(), name_lower])?;
381        collect_nodes(&mut rows)
382    }
383
384    /// Every `config_key` node's flattened setting (ADR-0009), read back out of
385    /// the graph as [`crate::ConfigKey`]s — the graph-native source the cross-repo
386    /// link matcher (`roteiro links --infer`) consumes, so it never re-parses
387    /// config files. Ordered by node key (deterministic). A node missing the
388    /// `key`/`path` a well-formed `config_key` carries is skipped defensively.
389    ///
390    /// # Errors
391    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
392    /// [`StoreError::Corrupt`] on decode failure.
393    pub fn config_keys(&self) -> Result<Vec<crate::ConfigKey>, StoreError> {
394        let nodes = self.nodes_by_kind(&NodeKind::Other(crate::config_keys::KIND.to_owned()))?;
395        let mut out = Vec::with_capacity(nodes.len());
396        for n in &nodes {
397            let key = n.meta.get("key").and_then(serde_json::Value::as_str);
398            let value = n.meta.get("value").and_then(serde_json::Value::as_str);
399            if let (Some(key), Some(file)) = (key, n.path.as_deref()) {
400                out.push(crate::ConfigKey {
401                    file: file.to_owned(),
402                    key: key.to_owned(),
403                    value: value.unwrap_or_default().to_owned(),
404                });
405            }
406        }
407        Ok(out)
408    }
409
410    /// Every node whose source `path` is `path`, ordered by key — the file node
411    /// plus the symbols and markers defined in it. Used to scope a change to the
412    /// graph (e.g. `roteiro review`).
413    ///
414    /// # Errors
415    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
416    /// [`StoreError::Corrupt`] on decode failure.
417    pub fn nodes_by_path(&self, path: &str) -> Result<Vec<Node>, StoreError> {
418        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.path = ?1 ORDER BY n.key");
419        let mut stmt = self.conn.prepare(&sql)?;
420        let mut rows = stmt.query([path])?;
421        collect_nodes(&mut rows)
422    }
423
424    /// Every node produced by a given layer, ordered by key. The incremental
425    /// `sync` loads the `Derived` layer to reconstruct the extraction graph
426    /// without re-reading every blob.
427    ///
428    /// # Errors
429    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
430    /// [`StoreError::Corrupt`] on decode failure.
431    pub fn nodes_by_provenance(&self, provenance: Provenance) -> Result<Vec<Node>, StoreError> {
432        let sql = format!("SELECT {NODE_COLS} FROM nodes n WHERE n.provenance = ?1 ORDER BY n.key");
433        let mut stmt = self.conn.prepare(&sql)?;
434        let mut rows = stmt.query([provenance.as_str()])?;
435        collect_nodes(&mut rows)
436    }
437
438    /// Every node in the store, ordered by key. Unlike [`Store::export_factset`]
439    /// this decodes no edges, so it is cheap for node-only scans (e.g. search).
440    ///
441    /// # Errors
442    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
443    /// [`StoreError::Corrupt`] on decode failure.
444    pub fn all_nodes(&self) -> Result<Vec<Node>, StoreError> {
445        let sql = format!("SELECT {NODE_COLS} FROM nodes n ORDER BY n.key");
446        let mut stmt = self.conn.prepare(&sql)?;
447        let mut rows = stmt.query([])?;
448        collect_nodes(&mut rows)
449    }
450
451    /// Every edge in the store, with endpoints resolved to their node keys. Used
452    /// by [`Store::reconcile`] to diff the edge set.
453    ///
454    /// Ordered by the edge's **content** — `(src key, dst key, kind, provenance)`,
455    /// the table's unique tuple — not by row id. This makes the order a function
456    /// of the *graph*, not of insertion history, so an incrementally
457    /// [`reconcile`](Store::reconcile)d store and a cold [`rebuild`](Store::rebuild)
458    /// return edges identically. (The same reason node scans order by `key`.)
459    ///
460    /// # Errors
461    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
462    pub fn all_edges(&self) -> Result<Vec<Edge>, StoreError> {
463        let sql = format!("{EDGE_SELECT} ORDER BY ns.key, nd.key, e.kind, e.provenance");
464        let mut stmt = self.conn.prepare(&sql)?;
465        let mut rows = stmt.query([])?;
466        collect_edges(&mut rows)
467    }
468
469    /// Edges whose source is the node with the given key, in content order
470    /// (`(dst key, kind, provenance)` — `src` is fixed). Content-ordered rather
471    /// than by row id so the result is history-independent; see [`Store::all_edges`].
472    ///
473    /// # Errors
474    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
475    pub fn edges_from(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
476        let sql = format!("{EDGE_SELECT} WHERE ns.key = ?1 ORDER BY nd.key, e.kind, e.provenance");
477        let mut stmt = self.conn.prepare(&sql)?;
478        let mut rows = stmt.query([key])?;
479        collect_edges(&mut rows)
480    }
481
482    /// Edges whose destination is the node with the given key, in content order
483    /// (`(src key, kind, provenance)` — `dst` is fixed). See [`Store::all_edges`].
484    ///
485    /// # Errors
486    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
487    pub fn edges_to(&self, key: &str) -> Result<Vec<Edge>, StoreError> {
488        let sql = format!("{EDGE_SELECT} WHERE nd.key = ?1 ORDER BY ns.key, e.kind, e.provenance");
489        let mut stmt = self.conn.prepare(&sql)?;
490        let mut rows = stmt.query([key])?;
491        collect_edges(&mut rows)
492    }
493
494    /// All edges with the given provenance, in content order
495    /// (`(src key, dst key, kind)` — `provenance` is fixed). See [`Store::all_edges`].
496    ///
497    /// # Errors
498    /// Returns [`StoreError::Sqlite`] or [`StoreError::Corrupt`] on failure.
499    pub fn edges_by_provenance(&self, provenance: Provenance) -> Result<Vec<Edge>, StoreError> {
500        let sql = format!("{EDGE_SELECT} WHERE e.provenance = ?1 ORDER BY ns.key, nd.key, e.kind");
501        let mut stmt = self.conn.prepare(&sql)?;
502        let mut rows = stmt.query([provenance.as_str()])?;
503        collect_edges(&mut rows)
504    }
505
506    /// Delete all edges with the given provenance, returning how many were
507    /// removed. Used to re-derive a whole provenance class authoritatively (e.g.
508    /// `inferred` edges when re-running inference with different parameters).
509    ///
510    /// # Errors
511    /// Returns [`StoreError::Sqlite`] on write failure.
512    pub fn delete_edges_by_provenance(&self, provenance: Provenance) -> Result<u64, StoreError> {
513        let n = self.conn.execute(
514            "DELETE FROM edges WHERE provenance = ?1",
515            [provenance.as_str()],
516        )?;
517        Ok(u64::try_from(n).unwrap_or(0))
518    }
519
520    /// Delete all edges carrying the given `src_ref`, returning how many were
521    /// removed. Lets one producer of `inferred` edges (e.g. the embedding layer,
522    /// or a Graphify import) re-derive its own edges authoritatively without
523    /// touching edges another producer contributed.
524    ///
525    /// # Errors
526    /// Returns [`StoreError::Sqlite`] on write failure.
527    pub fn delete_edges_by_src_ref(&self, src_ref: &str) -> Result<u64, StoreError> {
528        let n = self
529            .conn
530            .execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
531        Ok(u64::try_from(n).unwrap_or(0))
532    }
533
534    /// Apply an import layer to the live graph **and** persist it durably under
535    /// `src_ref`, validating as it goes: this ref's prior edges are cleared
536    /// (an authoritative re-import), the layer's nodes are upserted, and each
537    /// edge is applied only if both endpoints resolve. Dangling edges — cross-
538    /// references to code that is not present — are dropped, and only the
539    /// validated (trimmed) layer is persisted, so stale data is never stored.
540    ///
541    /// This is the "validate on import" half; [`Store::reapply_imports`] is the
542    /// "validate on sync" half, re-checking layers against the rebuilt graph.
543    ///
544    /// # Errors
545    /// Returns [`StoreError::Json`] if `facts` cannot be (de)serialized,
546    /// [`StoreError::InvalidEdge`] on a malformed edge, or [`StoreError::Sqlite`]
547    /// on write failure.
548    pub fn apply_import_layer(
549        &mut self,
550        src_ref: &str,
551        facts: &FactSet,
552    ) -> Result<ImportApplied, StoreError> {
553        let tx = self.conn.transaction()?;
554        // Authoritative re-import: drop this ref's prior edges from the live graph.
555        tx.execute("DELETE FROM edges WHERE src_ref = ?1", [src_ref])?;
556        for node in &facts.nodes {
557            upsert_node(&tx, node)?;
558        }
559        let (kept, applied) = apply_edges_pruning(&tx, &facts.edges)?;
560        let trimmed = FactSet {
561            nodes: facts.nodes.clone(),
562            edges: kept,
563        };
564        put_import_row(&tx, src_ref, &trimmed)?;
565        tx.commit()?;
566        Ok(ImportApplied {
567            layers: 1,
568            nodes: facts.nodes.len(),
569            ..applied
570        })
571    }
572
573    /// Remove the persisted import layer for `src_ref`, returning whether one
574    /// existed. Does not remove edges already in the live graph (use
575    /// [`Store::delete_edges_by_src_ref`] for that).
576    ///
577    /// # Errors
578    /// Returns [`StoreError::Sqlite`] on write failure.
579    pub fn delete_import(&self, src_ref: &str) -> Result<bool, StoreError> {
580        let n = self
581            .conn
582            .execute("DELETE FROM imports WHERE src_ref = ?1", [src_ref])?;
583        Ok(n > 0)
584    }
585
586    /// The `src_ref`s of all persisted import layers, ordered.
587    ///
588    /// # Errors
589    /// Returns [`StoreError::Sqlite`] on query failure.
590    pub fn import_refs(&self) -> Result<Vec<String>, StoreError> {
591        let mut stmt = self
592            .conn
593            .prepare("SELECT src_ref FROM imports ORDER BY src_ref")?;
594        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
595        let mut out = Vec::new();
596        for row in rows {
597            out.push(row?);
598        }
599        Ok(out)
600    }
601
602    /// Re-apply every persisted import layer on top of the current graph and
603    /// **re-validate** it: all import nodes are upserted first (so cross-layer
604    /// and self references resolve), then each edge is applied; an edge whose
605    /// endpoint is now absent — a cross-reference to code a sync removed — is
606    /// pruned from the persisted layer, not merely skipped. So the durable store
607    /// keeps only still-correct data. Idempotent; safe to run after each rebuild.
608    ///
609    /// # Errors
610    /// Returns [`StoreError::Json`] if a stored layer cannot be (de)serialized,
611    /// or [`StoreError::Sqlite`] on write failure.
612    pub fn reapply_imports(&mut self) -> Result<ImportApplied, StoreError> {
613        let layers = self.load_import_layers()?;
614        let tx = self.conn.transaction()?;
615        // Pass 1: upsert every layer's nodes so intra-import edges resolve
616        // regardless of which layer defines the endpoint.
617        for (_, facts) in &layers {
618            for node in &facts.nodes {
619                upsert_node(&tx, node)?;
620            }
621        }
622        // Pass 2: apply edges, pruning (and rewriting) any that dangle.
623        let mut applied = ImportApplied {
624            layers: layers.len(),
625            ..ImportApplied::default()
626        };
627        for (src_ref, facts) in &layers {
628            applied.nodes += facts.nodes.len();
629            let (kept, counts) = apply_edges_pruning(&tx, &facts.edges)?;
630            applied.edges_applied += counts.edges_applied;
631            applied.edges_pruned += counts.edges_pruned;
632            if kept.len() != facts.edges.len() {
633                let trimmed = FactSet {
634                    nodes: facts.nodes.clone(),
635                    edges: kept,
636                };
637                put_import_row(&tx, src_ref, &trimmed)?;
638            }
639        }
640        tx.commit()?;
641        Ok(applied)
642    }
643
644    /// Load and decode every persisted import layer as `(src_ref, FactSet)`, in
645    /// `src_ref` order.
646    fn load_import_layers(&self) -> Result<Vec<(String, FactSet)>, StoreError> {
647        let mut stmt = self
648            .conn
649            .prepare("SELECT src_ref, facts FROM imports ORDER BY src_ref")?;
650        let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
651        let mut out = Vec::new();
652        for row in rows {
653            let (src_ref, json) = row?;
654            let mut facts: FactSet = serde_json::from_str(&json)?;
655            // An import-layer node is *never* derived (derivation is `sync`'s job),
656            // so a `Derived` tag here is always wrong. It arises two ways, both
657            // repaired the same: a legacy layer persisted before nodes carried
658            // provenance (the field is absent → serde defaults `Derived`), or —
659            // anomalously — a layer that stored an explicit `"provenance":"derived"`
660            // (a producer/data bug). We deliberately repair *both* rather than only
661            // the absent case: leaving an explicit-`derived` import node in place
662            // would let a layer-scoped `sync` treat it as derived and delete it —
663            // the exact corruption this guards against — so repair is the safe
664            // recovery, not silent masking. Idempotent, runs on every reapply
665            // (old stores self-heal), and a no-op for correctly-tagged fresh imports.
666            for node in &mut facts.nodes {
667                if node.provenance == Provenance::Derived {
668                    node.provenance = import_node_provenance(&node.key);
669                }
670            }
671            out.push((src_ref, facts));
672        }
673        Ok(out)
674    }
675
676    /// Neighbouring nodes reachable from `key` in the given direction. Returns
677    /// an empty vector if the node does not exist.
678    ///
679    /// # Errors
680    /// Returns [`StoreError::Sqlite`], [`StoreError::Json`], or
681    /// [`StoreError::Corrupt`] on failure.
682    pub fn neighbors(&self, key: &str, dir: Direction) -> Result<Vec<Node>, StoreError> {
683        let out = format!(
684            "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.dst \
685             JOIN nodes s ON s.id = e.src WHERE s.key = ?1"
686        );
687        let inc = format!(
688            "SELECT {NODE_COLS} FROM nodes n JOIN edges e ON n.id = e.src \
689             JOIN nodes d ON d.id = e.dst WHERE d.key = ?1"
690        );
691        // Order by output column 1 (the node key) so results are deterministic
692        // across SQLite versions/plans. Positional ordering avoids both the
693        // ambiguity of a bare `key` (present in every joined table) and the fact
694        // that a table-qualified name cannot be used after the `Both` UNION.
695        let sql = match dir {
696            Direction::Outgoing => format!("{out} ORDER BY 1"),
697            Direction::Incoming => format!("{inc} ORDER BY 1"),
698            Direction::Both => format!("{out} UNION {inc} ORDER BY 1"),
699        };
700        let mut stmt = self.conn.prepare(&sql)?;
701        let mut rows = stmt.query([key])?;
702        collect_nodes(&mut rows)
703    }
704
705    /// Fetch the cached context bundle for `key` as `(fingerprint, json)`, if
706    /// present. The caller compares the fingerprint to the node's current one to
707    /// decide whether the entry is fresh (see [`crate::context`]).
708    ///
709    /// # Errors
710    /// Returns [`StoreError::Sqlite`] on query failure.
711    pub fn context_cache_get(&self, key: &str) -> Result<Option<(String, String)>, StoreError> {
712        let row = self
713            .conn
714            .query_row(
715                "SELECT fingerprint, json FROM node_context WHERE key = ?1",
716                [key],
717                |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)),
718            )
719            .optional()?;
720        Ok(row)
721    }
722
723    /// Fetch just the cached fingerprint for `key`, without reading the (larger)
724    /// JSON payload — for a cheap freshness check.
725    ///
726    /// # Errors
727    /// Returns [`StoreError::Sqlite`] on query failure.
728    pub fn context_cache_fingerprint(&self, key: &str) -> Result<Option<String>, StoreError> {
729        let fp = self
730            .conn
731            .query_row(
732                "SELECT fingerprint FROM node_context WHERE key = ?1",
733                [key],
734                |r| r.get::<_, String>(0),
735            )
736            .optional()?;
737        Ok(fp)
738    }
739
740    /// Store (or replace) the cached context bundle for `key`.
741    ///
742    /// # Errors
743    /// Returns [`StoreError::Sqlite`] on write failure.
744    pub fn context_cache_put(
745        &self,
746        key: &str,
747        fingerprint: &str,
748        json: &str,
749    ) -> Result<(), StoreError> {
750        self.conn.execute(
751            "INSERT INTO node_context (key, fingerprint, json) VALUES (?1, ?2, ?3)
752             ON CONFLICT(key) DO UPDATE SET
753                 fingerprint = excluded.fingerprint, json = excluded.json",
754            [key, fingerprint, json],
755        )?;
756        Ok(())
757    }
758
759    /// Delete the cached context entry for `key`, returning whether one existed.
760    ///
761    /// # Errors
762    /// Returns [`StoreError::Sqlite`] on write failure.
763    pub fn context_cache_delete(&self, key: &str) -> Result<bool, StoreError> {
764        let n = self
765            .conn
766            .execute("DELETE FROM node_context WHERE key = ?1", [key])?;
767        Ok(n > 0)
768    }
769
770    /// Every key with a cached context entry, ordered. Used to prune entries for
771    /// nodes that no longer exist.
772    ///
773    /// # Errors
774    /// Returns [`StoreError::Sqlite`] on query failure.
775    pub fn context_cache_keys(&self) -> Result<Vec<String>, StoreError> {
776        let mut stmt = self
777            .conn
778            .prepare("SELECT key FROM node_context ORDER BY key")?;
779        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
780        let mut out = Vec::new();
781        for row in rows {
782            out.push(row?);
783        }
784        Ok(out)
785    }
786}
787
788// --- Free helpers operating on a `Connection` (a `Transaction` derefs to one) ---
789
790fn node_row_id(conn: &Connection, key: &str) -> rusqlite::Result<Option<i64>> {
791    conn.query_row("SELECT id FROM nodes WHERE key = ?1", [key], |r| r.get(0))
792        .optional()
793}
794
795/// Record (or clear) the last-synced `HEAD` tree id. Shared by `rebuild` and
796/// `reconcile` so both leave identical `sync_state`.
797fn write_sync_state(conn: &Connection, tree: Option<&str>) -> Result<(), StoreError> {
798    match tree {
799        // Clear `env` on every tree write: it is only valid for the tree a
800        // committed `sync` set it against, and that sync re-records it (via
801        // `set_sync_env`) immediately after. So a worktree/index sync, or any
802        // path that does not re-set it, leaves `env` NULL — reading as "unknown"
803        // and forcing the safe full re-extraction next time.
804        Some(tree) => conn.execute(
805            "INSERT INTO sync_state (id, tree) VALUES (0, ?1)
806             ON CONFLICT(id) DO UPDATE SET tree = excluded.tree, env = NULL",
807            [tree],
808        )?,
809        None => conn.execute("DELETE FROM sync_state WHERE id = 0", [])?,
810    };
811    Ok(())
812}
813
814fn upsert_node(conn: &Connection, node: &Node) -> Result<(), StoreError> {
815    let meta = serde_json::to_string(&node.meta)?;
816    let (span_start, span_end) = match node.span {
817        Some(s) => (Some(i64::from(s.start)), Some(i64::from(s.end))),
818        None => (None, None),
819    };
820    conn.execute(
821        "INSERT INTO nodes (key, kind, name, path, lang, blob_hash, span_start, span_end, provenance, meta)
822         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
823         ON CONFLICT(key) DO UPDATE SET
824             kind = excluded.kind, name = excluded.name, path = excluded.path,
825             lang = excluded.lang, blob_hash = excluded.blob_hash,
826             span_start = excluded.span_start, span_end = excluded.span_end,
827             provenance = excluded.provenance, meta = excluded.meta",
828        params![
829            node.key,
830            node.kind.as_str(),
831            node.name,
832            node.path,
833            node.lang,
834            node.blob_hash,
835            span_start,
836            span_end,
837            node.provenance.as_str(),
838            meta,
839        ],
840    )?;
841    Ok(())
842}
843
844fn insert_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
845    validate_edge(edge)?;
846    let src_id =
847        node_row_id(conn, &edge.src)?.ok_or_else(|| StoreError::UnknownNode(edge.src.clone()))?;
848    let dst_id =
849        node_row_id(conn, &edge.dst)?.ok_or_else(|| StoreError::UnknownNode(edge.dst.clone()))?;
850    insert_edge_row(conn, edge, src_id, dst_id)
851}
852
853/// Apply `edge` only if both endpoints already resolve to nodes, returning
854/// whether it was **applied** (both endpoints resolved; a duplicate of an
855/// existing edge is a harmless no-op via `ON CONFLICT DO NOTHING` but still
856/// reports `true`). A missing endpoint returns `false` rather than erroring —
857/// the caller prunes such dangling cross-references from the import layer.
858fn insert_edge_if_present(conn: &Connection, edge: &Edge) -> Result<bool, StoreError> {
859    validate_edge(edge)?;
860    let (Some(src_id), Some(dst_id)) =
861        (node_row_id(conn, &edge.src)?, node_row_id(conn, &edge.dst)?)
862    else {
863        return Ok(false);
864    };
865    insert_edge_row(conn, edge, src_id, dst_id)?;
866    Ok(true)
867}
868
869/// Apply `edges`, keeping those whose endpoints resolve and pruning the rest.
870/// Returns the kept edges plus the applied/pruned counts (in an [`ImportApplied`]
871/// whose `layers`/`nodes` are left zero for the caller to fill).
872fn apply_edges_pruning(
873    conn: &Connection,
874    edges: &[Edge],
875) -> Result<(Vec<Edge>, ImportApplied), StoreError> {
876    let mut kept = Vec::with_capacity(edges.len());
877    let mut counts = ImportApplied::default();
878    for edge in edges {
879        if insert_edge_if_present(conn, edge)? {
880            kept.push(edge.clone());
881            counts.edges_applied += 1;
882        } else {
883            counts.edges_pruned += 1;
884        }
885    }
886    Ok((kept, counts))
887}
888
889/// Upsert a persisted import layer row. Free helper so it can run inside the same
890/// transaction as an apply/prune pass.
891fn put_import_row(conn: &Connection, src_ref: &str, facts: &FactSet) -> Result<(), StoreError> {
892    let json = serde_json::to_string(facts)?;
893    conn.execute(
894        "INSERT INTO imports (src_ref, facts) VALUES (?1, ?2)
895         ON CONFLICT(src_ref) DO UPDATE SET facts = excluded.facts, imported_at = datetime('now')",
896        params![src_ref, json],
897    )?;
898    Ok(())
899}
900
901/// The provenance/confidence invariant guard shared by the strict and tolerant
902/// edge inserts.
903fn validate_edge(edge: &Edge) -> Result<(), StoreError> {
904    if edge.is_valid() {
905        Ok(())
906    } else {
907        Err(StoreError::InvalidEdge(format!(
908            "confidence must be present iff provenance is inferred (src={}, dst={})",
909            edge.src, edge.dst
910        )))
911    }
912}
913
914/// Insert an edge row given already-resolved endpoint ids. Edges are a set: a
915/// duplicate `(src, dst, kind, provenance)` is a no-op via `ON CONFLICT … DO
916/// NOTHING`, so re-applying a fact set never accumulates duplicates. Other
917/// constraint violations (guarded in Rust above) still surface.
918fn insert_edge_row(
919    conn: &Connection,
920    edge: &Edge,
921    src_id: i64,
922    dst_id: i64,
923) -> Result<(), StoreError> {
924    conn.execute(
925        "INSERT INTO edges (src, dst, kind, provenance, confidence, src_ref)
926         VALUES (?1, ?2, ?3, ?4, ?5, ?6)
927         ON CONFLICT(src, dst, kind, provenance) DO NOTHING",
928        params![
929            src_id,
930            dst_id,
931            edge.kind.as_str(),
932            edge.provenance.as_str(),
933            edge.confidence,
934            edge.src_ref,
935        ],
936    )?;
937    Ok(())
938}
939
940/// A hashable identity for an edge over **all** its fields — used by
941/// [`Store::reconcile`] to diff the edge set. A tuple (not a delimiter-joined
942/// string) so no field value can be confused with a separator: node keys embed
943/// git paths, which may legally contain any byte (including control characters),
944/// so a joined string could collapse distinct edges to one identity and drop an
945/// edge. Confidence is compared by its exact bit pattern (`f64::to_bits`, wrapped
946/// in `Option` so `None` and `Some(_)` stay distinct), the only non-`Eq` field.
947fn edge_identity(edge: &Edge) -> EdgeId {
948    (
949        edge.src.clone(),
950        edge.dst.clone(),
951        edge.kind.as_str().to_owned(),
952        edge.provenance.as_str().to_owned(),
953        edge.confidence.map(f64::to_bits),
954        edge.src_ref.clone(),
955    )
956}
957
958/// The tuple form of an edge's full-field identity (see [`edge_identity`]):
959/// `(src, dst, kind, provenance, confidence-bits, src_ref)`.
960type EdgeId = (String, String, String, String, Option<u64>, Option<String>);
961
962/// Delete the edge row identified by `(src, dst, kind, provenance)` — the table's
963/// unique key — resolving the endpoint node keys to ids. A no-op if absent.
964fn delete_edge(conn: &Connection, edge: &Edge) -> Result<(), StoreError> {
965    conn.execute(
966        "DELETE FROM edges
967         WHERE src = (SELECT id FROM nodes WHERE key = ?1)
968           AND dst = (SELECT id FROM nodes WHERE key = ?2)
969           AND kind = ?3 AND provenance = ?4",
970        params![
971            edge.src,
972            edge.dst,
973            edge.kind.as_str(),
974            edge.provenance.as_str()
975        ],
976    )?;
977    Ok(())
978}
979
980fn collect_nodes(rows: &mut rusqlite::Rows) -> Result<Vec<Node>, StoreError> {
981    let mut out = Vec::new();
982    while let Some(row) = rows.next()? {
983        out.push(row_to_node(row)?);
984    }
985    Ok(out)
986}
987
988fn collect_edges(rows: &mut rusqlite::Rows) -> Result<Vec<Edge>, StoreError> {
989    let mut out = Vec::new();
990    while let Some(row) = rows.next()? {
991        out.push(row_to_edge(row)?);
992    }
993    Ok(out)
994}
995
996fn row_to_node(row: &rusqlite::Row) -> Result<Node, StoreError> {
997    let kind: String = row.get("kind")?;
998    let span_start: Option<i64> = row.get("span_start")?;
999    let span_end: Option<i64> = row.get("span_end")?;
1000    let span = match (span_start, span_end) {
1001        (Some(s), Some(e)) => Some(Span::new(to_u32(s)?, to_u32(e)?)),
1002        _ => None,
1003    };
1004    let meta: String = row.get("meta")?;
1005    let provenance: String = row.get("provenance")?;
1006    let provenance = Provenance::from_token(&provenance)
1007        .ok_or_else(|| StoreError::Corrupt(format!("unknown node provenance: {provenance}")))?;
1008    Ok(Node {
1009        key: row.get("key")?,
1010        kind: NodeKind::from_token(&kind),
1011        name: row.get("name")?,
1012        path: row.get("path")?,
1013        lang: row.get("lang")?,
1014        blob_hash: row.get("blob_hash")?,
1015        span,
1016        provenance,
1017        meta: serde_json::from_str(&meta)?,
1018    })
1019}
1020
1021fn row_to_edge(row: &rusqlite::Row) -> Result<Edge, StoreError> {
1022    let kind: String = row.get("kind")?;
1023    let provenance: String = row.get("provenance")?;
1024    let provenance = Provenance::from_token(&provenance)
1025        .ok_or_else(|| StoreError::Corrupt(format!("unknown provenance: {provenance}")))?;
1026    Ok(Edge {
1027        src: row.get("src")?,
1028        dst: row.get("dst")?,
1029        kind: EdgeKind::from_token(&kind),
1030        provenance,
1031        confidence: row.get("confidence")?,
1032        src_ref: row.get("src_ref")?,
1033    })
1034}
1035
1036fn to_u32(v: i64) -> Result<u32, StoreError> {
1037    u32::try_from(v).map_err(|_| StoreError::Corrupt(format!("span offset out of range: {v}")))
1038}
1039
1040/// The true provenance of an import-layer node, from its key namespace: Graphify
1041/// nodes (`graphify:`) are [`Provenance::Inferred`]; every other import node (lat,
1042/// …) is [`Provenance::Authored`]. Import-layer nodes are never derived, so this
1043/// is used to repair a legacy `Derived` tag on load (see `load_import_layers`).
1044fn import_node_provenance(key: &str) -> Provenance {
1045    if key.starts_with("graphify:") {
1046        Provenance::Inferred
1047    } else {
1048        Provenance::Authored
1049    }
1050}
1051
1052#[cfg(test)]
1053mod tests {
1054    use super::Store;
1055    use crate::model::{Direction, Edge, EdgeKind, FactSet, Node, NodeKind, Span};
1056    use crate::provenance::Provenance;
1057
1058    fn sample_node(key: &str) -> Node {
1059        Node {
1060            key: key.to_owned(),
1061            kind: NodeKind::Fn,
1062            name: "sample".to_owned(),
1063            path: Some("src/lib.rs".to_owned()),
1064            lang: Some("rust".to_owned()),
1065            blob_hash: Some("deadbeef".to_owned()),
1066            span: Some(Span::new(10, 42)),
1067            provenance: Provenance::Derived,
1068            meta: serde_json::json!({"vis": "pub"}),
1069        }
1070    }
1071
1072    #[test]
1073    fn reconcile_matches_a_full_rebuild() {
1074        // reconcile must leave the store identical to a fresh rebuild, across an
1075        // add, a remove, a content change, and edge churn.
1076        let node = |k: &str, name: &str| {
1077            let mut n = sample_node(k);
1078            n.name = name.to_owned();
1079            n
1080        };
1081        let edge =
1082            |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
1083        // An inferred edge carries confidence and a src_ref — exercise both so the
1084        // equivalence claim covers every edge field, not just derived calls.
1085        let inferred = |src: &str, dst: &str, conf: f64| {
1086            let mut e = Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, conf);
1087            e.src_ref = Some("import:demo".to_owned());
1088            e
1089        };
1090
1091        let facts1 = FactSet {
1092            nodes: vec![node("a", "A"), node("b", "B"), node("c", "C")],
1093            edges: vec![edge("a", "b"), edge("b", "c"), inferred("a", "c", 0.7)],
1094        };
1095        // b changes (name), c is removed, d is added; edge b->c drops, a->d added,
1096        // and the inferred edge's confidence changes.
1097        let facts2 = FactSet {
1098            nodes: vec![node("a", "A"), node("b", "B2"), node("d", "D")],
1099            edges: vec![edge("a", "b"), edge("a", "d"), inferred("a", "d", 0.9)],
1100        };
1101
1102        // Path 1: rebuild facts1, then reconcile to facts2.
1103        let mut reconciled = Store::open_in_memory().expect("open");
1104        reconciled.rebuild(&facts1, Some("t1")).expect("rebuild");
1105        reconciled
1106            .reconcile(&facts2, Some("t2"))
1107            .expect("reconcile");
1108
1109        // Path 2: a fresh full rebuild of facts2.
1110        let mut rebuilt = Store::open_in_memory().expect("open");
1111        rebuilt.rebuild(&facts2, Some("t2")).expect("rebuild");
1112
1113        let canon = |fs: FactSet| {
1114            let mut nodes = fs.nodes;
1115            nodes.sort_by(|a, b| a.key.cmp(&b.key));
1116            let mut edges: Vec<String> = fs
1117                .edges
1118                .iter()
1119                .map(|e| {
1120                    format!(
1121                        "{}\0{}\0{}\0{}\0{:?}\0{:?}",
1122                        e.kind.as_str(),
1123                        e.src,
1124                        e.dst,
1125                        e.provenance.as_str(),
1126                        e.confidence,
1127                        e.src_ref
1128                    )
1129                })
1130                .collect();
1131            edges.sort();
1132            (nodes, edges)
1133        };
1134        assert_eq!(
1135            canon(reconciled.export_factset().expect("export")),
1136            canon(rebuilt.export_factset().expect("export")),
1137            "reconcile must match a full rebuild",
1138        );
1139        assert_eq!(
1140            reconciled.sync_state().expect("state").as_deref(),
1141            Some("t2")
1142        );
1143    }
1144
1145    #[test]
1146    fn reconcile_writes_only_the_edge_delta() {
1147        // An unchanged edge must keep its row (proving reconcile does not wipe and
1148        // reinsert the whole edge set); a removed edge's row goes; a new edge's row
1149        // appears. Row identity is the SQLite `rowid` — stable unless deleted.
1150        let n = |k: &str| sample_node(k);
1151        let e =
1152            |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
1153
1154        let mut store = Store::open_in_memory().expect("open");
1155        store
1156            .rebuild(
1157                &FactSet {
1158                    nodes: vec![n("a"), n("b"), n("c")],
1159                    edges: vec![e("a", "b"), e("b", "c")],
1160                },
1161                None,
1162            )
1163            .expect("rebuild");
1164
1165        // Map (src_key, dst_key) → rowid via the private connection.
1166        let rowids = |store: &Store| -> std::collections::HashMap<(String, String), i64> {
1167            let mut stmt = store
1168                .conn
1169                .prepare(
1170                    "SELECT ns.key, nd.key, e.rowid FROM edges e \
1171                     JOIN nodes ns ON ns.id = e.src JOIN nodes nd ON nd.id = e.dst",
1172                )
1173                .expect("prepare");
1174            stmt.query_map([], |r| {
1175                Ok((
1176                    (r.get::<_, String>(0)?, r.get::<_, String>(1)?),
1177                    r.get::<_, i64>(2)?,
1178                ))
1179            })
1180            .expect("query")
1181            .map(Result::unwrap)
1182            .collect()
1183        };
1184
1185        let before = rowids(&store);
1186        let ab_rowid = before[&("a".to_owned(), "b".to_owned())];
1187
1188        // Keep a->b, drop b->c, add a->c.
1189        store
1190            .reconcile(
1191                &FactSet {
1192                    nodes: vec![n("a"), n("b"), n("c")],
1193                    edges: vec![e("a", "b"), e("a", "c")],
1194                },
1195                None,
1196            )
1197            .expect("reconcile");
1198
1199        let after = rowids(&store);
1200        assert_eq!(
1201            after.get(&("a".to_owned(), "b".to_owned())),
1202            Some(&ab_rowid),
1203            "the unchanged edge keeps its row (not rewritten)"
1204        );
1205        assert!(
1206            !after.contains_key(&("b".to_owned(), "c".to_owned())),
1207            "the removed edge's row is gone"
1208        );
1209        assert!(
1210            after.contains_key(&("a".to_owned(), "c".to_owned())),
1211            "the added edge has a new row"
1212        );
1213    }
1214
1215    #[test]
1216    fn reconcile_is_history_independent_for_edge_queries() {
1217        // The whole point of the edge delta: it must be invisible above storage.
1218        // A store reached by rebuild(f1)+reconcile(f2) has different edge row ids
1219        // than a cold rebuild at f2, yet every edge query must return byte-for-byte
1220        // the same result — order included — because the queries are content-ordered.
1221        let n = |k: &str| sample_node(k);
1222        let d =
1223            |src: &str, dst: &str| Edge::derived(src.to_owned(), dst.to_owned(), EdgeKind::Calls);
1224        let inf = |src: &str, dst: &str, c: f64| {
1225            Edge::inferred(src.to_owned(), dst.to_owned(), EdgeKind::Related, c)
1226        };
1227        let f1 = FactSet {
1228            nodes: vec![n("a"), n("b"), n("c")],
1229            edges: vec![d("a", "b"), d("b", "c"), inf("a", "c", 0.7)],
1230        };
1231        let f2 = FactSet {
1232            nodes: vec![n("a"), n("b"), n("d")],
1233            edges: vec![d("a", "b"), d("a", "d"), inf("a", "d", 0.9)],
1234        };
1235
1236        let mut incremental = Store::open_in_memory().expect("open");
1237        incremental.rebuild(&f1, None).expect("rebuild");
1238        incremental.reconcile(&f2, None).expect("reconcile");
1239        let mut cold = Store::open_in_memory().expect("open");
1240        cold.rebuild(&f2, None).expect("rebuild");
1241
1242        // Project to all fields so the comparison covers order *and* content.
1243        let proj = |es: Vec<Edge>| -> Vec<String> {
1244            es.into_iter()
1245                .map(|e| {
1246                    format!(
1247                        "{}|{}|{}|{}|{:?}|{:?}",
1248                        e.src,
1249                        e.dst,
1250                        e.kind.as_str(),
1251                        e.provenance.as_str(),
1252                        e.confidence,
1253                        e.src_ref
1254                    )
1255                })
1256                .collect()
1257        };
1258
1259        for key in ["a", "b", "d"] {
1260            assert_eq!(
1261                proj(incremental.edges_from(key).expect("from")),
1262                proj(cold.edges_from(key).expect("from")),
1263                "edges_from({key}) must match a cold rebuild"
1264            );
1265            assert_eq!(
1266                proj(incremental.edges_to(key).expect("to")),
1267                proj(cold.edges_to(key).expect("to")),
1268                "edges_to({key}) must match a cold rebuild"
1269            );
1270        }
1271        assert_eq!(
1272            proj(incremental.all_edges().expect("all")),
1273            proj(cold.all_edges().expect("all")),
1274            "all_edges must match a cold rebuild"
1275        );
1276        for p in [Provenance::Derived, Provenance::Inferred] {
1277            assert_eq!(
1278                proj(incremental.edges_by_provenance(p).expect("prov")),
1279                proj(cold.edges_by_provenance(p).expect("prov")),
1280                "edges_by_provenance({}) must match a cold rebuild",
1281                p.as_str()
1282            );
1283        }
1284    }
1285
1286    #[test]
1287    fn reconcile_updates_confidence_on_an_unchanged_tuple() {
1288        // A change to *only* an edge's confidence — same (src, dst, kind,
1289        // provenance) — must still be applied. Edge identity includes confidence,
1290        // so it is delete+add (matching a full rebuild), not the insert-time
1291        // `DO NOTHING` that would leave the stale confidence in place.
1292        let n = |k: &str| sample_node(k);
1293        let inf = |c: f64| {
1294            let mut e = Edge::inferred("a".to_owned(), "b".to_owned(), EdgeKind::Related, c);
1295            e.src_ref = Some("import:demo".to_owned());
1296            e
1297        };
1298
1299        let mut store = Store::open_in_memory().expect("open");
1300        store
1301            .rebuild(
1302                &FactSet {
1303                    nodes: vec![n("a"), n("b")],
1304                    edges: vec![inf(0.5)],
1305                },
1306                None,
1307            )
1308            .expect("rebuild");
1309        store
1310            .reconcile(
1311                &FactSet {
1312                    nodes: vec![n("a"), n("b")],
1313                    edges: vec![inf(0.9)],
1314                },
1315                None,
1316            )
1317            .expect("reconcile");
1318
1319        let edges = store.edges_from("a").expect("edges");
1320        assert_eq!(edges.len(), 1);
1321        assert_eq!(
1322            edges[0].confidence,
1323            Some(0.9),
1324            "confidence updated, not left stale"
1325        );
1326    }
1327
1328    #[test]
1329    fn edge_identity_does_not_collide_across_field_boundaries() {
1330        // Node keys embed git paths, which may contain any byte — including the
1331        // unit separator (`\x1f`). A delimiter-joined identity would map these two
1332        // distinct edges to the same string (`a\x1fb\x1fc\x1f…`); the tuple identity
1333        // must keep them apart, or reconcile would drop one edge as a "duplicate".
1334        let e1 = super::edge_identity(&Edge::derived(
1335            "a\u{1f}b".to_owned(),
1336            "c".to_owned(),
1337            EdgeKind::Calls,
1338        ));
1339        let e2 = super::edge_identity(&Edge::derived(
1340            "a".to_owned(),
1341            "b\u{1f}c".to_owned(),
1342            EdgeKind::Calls,
1343        ));
1344        assert_ne!(
1345            e1, e2,
1346            "control chars in a key must not collapse identities"
1347        );
1348
1349        // A confidence-only difference (same tuple otherwise) also stays distinct,
1350        // and `None` (derived) never equals `Some(0.0)`.
1351        let derived = super::edge_identity(&Edge::derived(
1352            "a".to_owned(),
1353            "b".to_owned(),
1354            EdgeKind::Related,
1355        ));
1356        let inferred0 = super::edge_identity(&Edge::inferred(
1357            "a".to_owned(),
1358            "b".to_owned(),
1359            EdgeKind::Related,
1360            0.0,
1361        ));
1362        assert_ne!(derived, inferred0, "None vs Some(0.0) confidence differ");
1363    }
1364
1365    #[test]
1366    fn open_in_memory_applies_schema() {
1367        let store = Store::open_in_memory().expect("open");
1368        assert_eq!(store.node_count().expect("count"), 0);
1369        assert_eq!(store.schema_version().expect("version"), 7);
1370    }
1371
1372    #[test]
1373    fn upsert_and_get_round_trips_all_fields() {
1374        let store = Store::open_in_memory().expect("open");
1375        let node = sample_node("sym:rust:src/lib.rs#sample");
1376        store.upsert_node(&node).expect("upsert");
1377        let got = store.get_node(&node.key).expect("get").expect("present");
1378        assert_eq!(got, node);
1379    }
1380
1381    #[test]
1382    fn upsert_updates_in_place() {
1383        let store = Store::open_in_memory().expect("open");
1384        let mut node = sample_node("k");
1385        store.upsert_node(&node).expect("insert");
1386        node.name = "renamed".to_owned();
1387        node.kind = NodeKind::Struct;
1388        store.upsert_node(&node).expect("update");
1389        assert_eq!(store.node_count().expect("count"), 1);
1390        let got = store.get_node("k").expect("get").expect("present");
1391        assert_eq!(got.name, "renamed");
1392        assert_eq!(got.kind, NodeKind::Struct);
1393    }
1394
1395    #[test]
1396    fn edge_with_unknown_endpoint_is_rejected() {
1397        let store = Store::open_in_memory().expect("open");
1398        store
1399            .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
1400            .expect("a");
1401        let edge = Edge::derived("a", "missing", EdgeKind::Calls);
1402        let err = store.insert_edge(&edge).expect_err("should reject");
1403        assert!(matches!(err, super::StoreError::UnknownNode(k) if k == "missing"));
1404    }
1405
1406    #[test]
1407    fn inferred_edge_requires_confidence() {
1408        let store = Store::open_in_memory().expect("open");
1409        store
1410            .upsert_node(&Node::new("a", NodeKind::Fn, "a"))
1411            .expect("a");
1412        store
1413            .upsert_node(&Node::new("b", NodeKind::Fn, "b"))
1414            .expect("b");
1415        // Hand-build an inferred edge with no confidence to violate the invariant.
1416        let bad = Edge {
1417            src: "a".to_owned(),
1418            dst: "b".to_owned(),
1419            kind: EdgeKind::References,
1420            provenance: Provenance::Inferred,
1421            confidence: None,
1422            src_ref: None,
1423        };
1424        assert!(matches!(
1425            store.insert_edge(&bad).expect_err("reject"),
1426            super::StoreError::InvalidEdge(_)
1427        ));
1428    }
1429
1430    #[test]
1431    fn apply_factset_is_atomic() {
1432        let mut store = Store::open_in_memory().expect("open");
1433        // Second edge references a missing node, so the whole set must roll back.
1434        let facts = FactSet::new()
1435            .with_node(Node::new("a", NodeKind::Fn, "a"))
1436            .with_node(Node::new("b", NodeKind::Fn, "b"))
1437            .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
1438            .with_edge(Edge::derived("a", "ghost", EdgeKind::Calls));
1439        assert!(store.apply_factset(&facts).is_err());
1440        assert_eq!(store.node_count().expect("count"), 0, "rolled back");
1441        assert_eq!(store.edge_count().expect("count"), 0, "rolled back");
1442    }
1443
1444    #[test]
1445    fn neighbors_and_provenance_queries() {
1446        let mut store = Store::open_in_memory().expect("open");
1447        let facts = FactSet::new()
1448            .with_node(Node::new("a", NodeKind::Fn, "a"))
1449            .with_node(Node::new("b", NodeKind::Fn, "b"))
1450            .with_node(Node::new("c", NodeKind::Fn, "c"))
1451            .with_edge(Edge::derived("a", "b", EdgeKind::Calls))
1452            .with_edge(Edge::inferred("a", "c", EdgeKind::References, 0.5));
1453        store.apply_factset(&facts).expect("apply");
1454
1455        let out = store.neighbors("a", Direction::Outgoing).expect("out");
1456        let mut keys: Vec<_> = out.iter().map(|n| n.key.clone()).collect();
1457        keys.sort();
1458        assert_eq!(keys, ["b", "c"]);
1459
1460        assert!(
1461            store
1462                .neighbors("b", Direction::Outgoing)
1463                .expect("b out")
1464                .is_empty()
1465        );
1466        assert_eq!(
1467            store
1468                .neighbors("b", Direction::Incoming)
1469                .expect("b in")
1470                .len(),
1471            1
1472        );
1473
1474        let inferred = store
1475            .edges_by_provenance(Provenance::Inferred)
1476            .expect("inf");
1477        assert_eq!(inferred.len(), 1);
1478        assert_eq!(inferred[0].confidence, Some(0.5));
1479    }
1480
1481    #[test]
1482    fn neighbors_of_absent_node_is_empty() {
1483        let store = Store::open_in_memory().expect("open");
1484        assert!(
1485            store
1486                .neighbors("nope", Direction::Both)
1487                .expect("q")
1488                .is_empty()
1489        );
1490    }
1491
1492    #[test]
1493    fn get_missing_node_is_none() {
1494        let store = Store::open_in_memory().expect("open");
1495        assert!(store.get_node("absent").expect("get").is_none());
1496    }
1497
1498    #[test]
1499    fn nodes_by_kind_and_edges_to() {
1500        let mut store = Store::open_in_memory().expect("open");
1501        let facts = FactSet::new()
1502            .with_node(Node::new("f1", NodeKind::Fn, "f1"))
1503            .with_node(Node::new("f2", NodeKind::Fn, "f2"))
1504            .with_node(Node::new("s1", NodeKind::Struct, "s1"))
1505            .with_edge(Edge::derived("f1", "s1", EdgeKind::References))
1506            .with_edge(Edge::derived("f2", "s1", EdgeKind::References));
1507        store.apply_factset(&facts).expect("apply");
1508
1509        let fns = store.nodes_by_kind(&NodeKind::Fn).expect("fns");
1510        assert_eq!(
1511            fns.iter().map(|n| n.key.as_str()).collect::<Vec<_>>(),
1512            ["f1", "f2"]
1513        );
1514        assert!(
1515            store
1516                .nodes_by_kind(&NodeKind::Enum)
1517                .expect("enums")
1518                .is_empty()
1519        );
1520
1521        let into_s1 = store.edges_to("s1").expect("edges_to");
1522        assert_eq!(into_s1.len(), 2);
1523        assert!(into_s1.iter().all(|e| e.dst == "s1"));
1524    }
1525
1526    #[test]
1527    fn open_persists_across_reopen() {
1528        let path =
1529            std::env::temp_dir().join(format!("roteiro-open-test-{}.db", std::process::id()));
1530        std::fs::remove_file(&path).ok();
1531        {
1532            let store = Store::open(&path).expect("open");
1533            store
1534                .upsert_node(&sample_node("persisted"))
1535                .expect("upsert");
1536        }
1537        {
1538            let store = Store::open(&path).expect("reopen");
1539            assert_eq!(store.node_count().expect("count"), 1);
1540            assert_eq!(store.schema_version().expect("version"), 7);
1541            assert!(store.get_node("persisted").expect("get").is_some());
1542        }
1543        std::fs::remove_file(&path).expect("cleanup");
1544    }
1545
1546    fn graphify_layer(dst: &str) -> FactSet {
1547        FactSet::new()
1548            .with_node(Node::new("graphify:doc1", NodeKind::Doc, "Doc 1"))
1549            .with_edge({
1550                let mut e = Edge::inferred("graphify:doc1", dst, EdgeKind::References, 0.9);
1551                e.src_ref = Some("import:graphify".to_owned());
1552                e
1553            })
1554    }
1555
1556    /// A persisted import layer is re-applied after a `rebuild` wipes the graph,
1557    /// so imported facts survive a code-changing sync.
1558    #[test]
1559    fn imports_survive_rebuild() {
1560        let mut store = Store::open_in_memory().expect("open");
1561        let derived = FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a.rs"));
1562        store.rebuild(&derived, Some("tree1")).expect("rebuild");
1563
1564        // apply_import_layer applies to the live graph and persists in one step.
1565        let applied = store
1566            .apply_import_layer("import:graphify", &graphify_layer("file:a.rs"))
1567            .expect("apply import");
1568        assert_eq!(applied.edges_applied, 1);
1569        assert_eq!(applied.edges_pruned, 0);
1570        assert_eq!(store.import_refs().expect("refs"), vec!["import:graphify"]);
1571
1572        // Simulate a code-changing sync: the derived graph is rebuilt (still has
1573        // file:a.rs), which drops the imported doc + edge from the live graph.
1574        store.rebuild(&derived, Some("tree2")).expect("rebuild2");
1575        assert!(store.get_node("graphify:doc1").expect("get").is_none());
1576
1577        // Re-applying imports restores them; nothing is pruned (target present).
1578        let applied = store.reapply_imports().expect("reapply");
1579        assert_eq!(applied.layers, 1);
1580        assert_eq!(applied.nodes, 1);
1581        assert_eq!(applied.edges_applied, 1);
1582        assert_eq!(applied.edges_pruned, 0);
1583        assert!(store.get_node("graphify:doc1").expect("get").is_some());
1584        assert_eq!(store.edges_from("graphify:doc1").expect("edges").len(), 1);
1585    }
1586
1587    /// When a sync removes an edge's target (e.g. a deleted file), re-applying
1588    /// **prunes** that stale cross-reference from the persisted layer — it is not
1589    /// kept and retried forever. The import node itself is preserved.
1590    #[test]
1591    fn reapply_prunes_stale_cross_references() {
1592        let mut store = Store::open_in_memory().expect("open");
1593        let derived = FactSet::new().with_node(Node::new("file:gone.rs", NodeKind::File, "g"));
1594        store.rebuild(&derived, Some("t1")).expect("rebuild");
1595        store
1596            .apply_import_layer("import:graphify", &graphify_layer("file:gone.rs"))
1597            .expect("import");
1598
1599        // Code-changing sync: file:gone.rs is deleted from the derived graph.
1600        store
1601            .rebuild(&FactSet::new(), Some("t2"))
1602            .expect("rebuild2");
1603        let applied = store.reapply_imports().expect("reapply");
1604        assert_eq!(applied.nodes, 1);
1605        assert_eq!(applied.edges_applied, 0);
1606        assert_eq!(
1607            applied.edges_pruned, 1,
1608            "the edge to the deleted file is pruned"
1609        );
1610        assert!(store.get_node("graphify:doc1").expect("get").is_some());
1611
1612        // The prune is durable: a second reapply finds nothing left to prune,
1613        // proving the stale edge was removed from the persisted layer.
1614        let again = store.reapply_imports().expect("reapply2");
1615        assert_eq!(again.edges_applied, 0);
1616        assert_eq!(again.edges_pruned, 0, "already pruned; not retried");
1617    }
1618
1619    /// `apply_import_layer` validates on import: a dangling edge in the incoming
1620    /// layer is dropped and never persisted.
1621    #[test]
1622    fn apply_import_layer_prunes_on_import() {
1623        let mut store = Store::open_in_memory().expect("open");
1624        let present = || FactSet::new().with_node(Node::new("file:a.rs", NodeKind::File, "a"));
1625        store.rebuild(&present(), Some("t")).expect("rebuild");
1626
1627        let layer = graphify_layer("file:a.rs").with_edge({
1628            // Points at a file that does not exist → pruned on import.
1629            let mut e = Edge::inferred("graphify:doc1", "file:ghost.rs", EdgeKind::References, 0.9);
1630            e.src_ref = Some("import:graphify".to_owned());
1631            e
1632        });
1633        let applied = store
1634            .apply_import_layer("import:graphify", &layer)
1635            .expect("import");
1636        assert_eq!(applied.edges_applied, 1);
1637        assert_eq!(applied.edges_pruned, 1);
1638
1639        // A rebuild + reapply confirms only the valid edge was persisted.
1640        store.rebuild(&present(), Some("t2")).expect("rebuild2");
1641        let re = store.reapply_imports().expect("reapply");
1642        assert_eq!(re.edges_applied, 1);
1643        assert_eq!(re.edges_pruned, 0, "ghost edge was not persisted");
1644    }
1645
1646    #[test]
1647    fn legacy_import_layer_nodes_are_retagged_non_derived() {
1648        // A layer persisted before nodes carried provenance: its node objects have
1649        // no `provenance` field, so serde defaults them to Derived. On reapply the
1650        // store must repair them — a Graphify node to Inferred, a lat node to
1651        // Authored — so a later derived-only sync never mistakes them for derived.
1652        let mut store = Store::open_in_memory().expect("open");
1653        let legacy = r#"{"nodes":[
1654            {"key":"graphify:doc1","kind":"doc","name":"d","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null},
1655            {"key":"lat:lat.md/a.md","kind":"doc","name":"a","path":null,"lang":null,"blob_hash":null,"span":null,"meta":null}
1656        ],"edges":[]}"#;
1657        store
1658            .conn
1659            .execute(
1660                "INSERT INTO imports (src_ref, facts) VALUES ('import:legacy', ?1)",
1661                [legacy],
1662            )
1663            .expect("seed legacy import row");
1664
1665        store.reapply_imports().expect("reapply");
1666
1667        let g = store
1668            .get_node("graphify:doc1")
1669            .expect("get")
1670            .expect("graphify node");
1671        assert_eq!(
1672            g.provenance,
1673            Provenance::Inferred,
1674            "graphify import node repaired to inferred"
1675        );
1676        let l = store
1677            .get_node("lat:lat.md/a.md")
1678            .expect("get")
1679            .expect("lat node");
1680        assert_eq!(
1681            l.provenance,
1682            Provenance::Authored,
1683            "lat import node repaired to authored"
1684        );
1685    }
1686
1687    /// `apply_import_layer` replaces the layer for a ref; `delete_import` removes.
1688    #[test]
1689    fn apply_import_replaces_and_delete_removes() {
1690        let mut store = Store::open_in_memory().expect("open");
1691        let a = FactSet::new().with_node(Node::new("graphify:x", NodeKind::Doc, "x"));
1692        let b = FactSet::new().with_node(Node::new("graphify:y", NodeKind::Doc, "y"));
1693        store.apply_import_layer("import:graphify", &a).expect("a");
1694        store.apply_import_layer("import:graphify", &b).expect("b");
1695        assert_eq!(
1696            store.import_refs().expect("refs").len(),
1697            1,
1698            "same ref replaced"
1699        );
1700        assert!(store.delete_import("import:graphify").expect("del"));
1701        assert!(store.import_refs().expect("refs").is_empty());
1702        assert!(!store.delete_import("import:graphify").expect("del again"));
1703    }
1704}