omgbase_store/
graph.rs

1//! The graph tables (`spec/graph`; `spec/store` §5.4 after the blocks,
2//! sections and properties): `nodes` + `nodes_fts` (§2), `external_nodes`
3//! and the edge resolution (§3.2), the `edges` intervals (§3.3), the
4//! `doc_edges` rollup (§3.4) and phantom adoption (§3.5). Extraction itself
5//! is `omgbase-graph`, pure; everything here reads or writes the database
6//! inside the commit transaction. Deletion (§3.6) touches none of it.
7
8use std::collections::{HashMap, HashSet};
9
10use omgbase_graph::{
11    DstKind, EdgeDescriptor, ProjectedNode, Provenance, canonical_path, doc_dir, node_rows,
12    resolve_relative,
13};
14use rusqlite::{Connection, OptionalExtension, params};
15
16use crate::error::Result;
17use crate::mint::Mint;
18
19// ---- nodes (§2) -----------------------------------------------------------------
20
21/// §2.2: one `md:section` node per `sections` row of the document, in
22/// `first_ordinal` order, named by the heading block's text.
23pub fn project_section_nodes(conn: &Connection, doc_id: &str) -> Result<Vec<ProjectedNode>> {
24    let mut stmt = conn.prepare_cached(
25        "SELECT s.heading_block, s.level, s.first_ordinal, s.last_ordinal, hb.text
26           FROM sections s
27           JOIN blocks hb ON hb.block_id = s.heading_block AND hb.doc_id = s.doc_id
28          WHERE s.doc_id = ?1 AND hb.deleted_commit IS NULL
29          ORDER BY s.first_ordinal",
30    )?;
31    let rows = stmt.query_map(params![doc_id], |r| {
32        Ok(ProjectedNode::section(
33            &r.get::<_, String>(0)?,
34            &r.get::<_, String>(4)?,
35            r.get(1)?,
36            r.get(2)?,
37            r.get(3)?,
38        ))
39    })?;
40    Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
41}
42
43/// Delete a document's `nodes` rows and their `nodes_fts` entries
44/// (external-content index: a `'delete'` with the indexed `name`/`value`,
45/// `""` for null, as they were inserted).
46pub fn delete_doc_nodes(conn: &Connection, doc_id: &str) -> Result<()> {
47    let rows: Vec<(i64, Option<String>, Option<String>)> = {
48        let mut stmt =
49            conn.prepare_cached("SELECT rowid, name, value FROM nodes WHERE doc_id = ?1")?;
50        let it = stmt.query_map(params![doc_id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
51        it.collect::<std::result::Result<Vec<_>, _>>()?
52    };
53    let mut del = conn.prepare_cached(
54        "INSERT INTO nodes_fts(nodes_fts, rowid, name, value) VALUES('delete', ?1, ?2, ?3)",
55    )?;
56    for (rowid, name, value) in rows {
57        del.execute(params![
58            rowid,
59            name.unwrap_or_default(),
60            value.unwrap_or_default()
61        ])?;
62    }
63    conn.execute("DELETE FROM nodes WHERE doc_id = ?1", params![doc_id])?;
64    Ok(())
65}
66
67/// §2.3: replace the document's `nodes` (and FTS) rows with `nodes` — the
68/// adapter nodes followed by the section nodes — assigning ordinals and ids
69/// in list order. Returns how many rows were written.
70pub fn write_doc_nodes(
71    conn: &Connection,
72    repo_id: &str,
73    doc_id: &str,
74    nodes: &[ProjectedNode],
75) -> Result<usize> {
76    delete_doc_nodes(conn, doc_id)?;
77    let mut insert = conn.prepare_cached(
78        "INSERT INTO nodes (node_id, repo_id, doc_id, block_id, kind, name, value, span_start, span_end, attrs)
79         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
80    )?;
81    let mut fts =
82        conn.prepare_cached("INSERT INTO nodes_fts(rowid, name, value) VALUES (?1, ?2, ?3)")?;
83    for row in node_rows(doc_id, nodes) {
84        let n = row.node;
85        insert.execute(params![
86            row.node_id,
87            repo_id,
88            doc_id,
89            n.block_id,
90            n.kind.as_str(),
91            n.name,
92            n.value,
93            n.span.map(|(s, _)| s as i64),
94            n.span.map(|(_, e)| e as i64),
95            serde_json::Value::Object(n.attrs.clone()).to_string(),
96        ])?;
97        let rowid = conn.last_insert_rowid();
98        fts.execute(params![
99            rowid,
100            n.name.as_deref().unwrap_or(""),
101            n.value.as_deref().unwrap_or("")
102        ])?;
103    }
104    Ok(nodes.len())
105}
106
107// ---- resolution (§3.2) ---------------------------------------------------------------
108
109/// An edge with its target resolved to a node id.
110#[derive(Clone, Debug, PartialEq, Eq)]
111pub struct ResolvedEdge {
112    pub src_block: Option<String>,
113    pub src_field: Option<String>,
114    pub predicate: String,
115    pub dst_kind: DstKind,
116    /// A doc id, an `external_nodes` id, or `phantom:<path>`.
117    pub dst_node: String,
118    pub anchor: Option<String>,
119    pub provenance: Provenance,
120}
121
122impl ResolvedEdge {
123    /// §3.3 key: `src_block|src_field|predicate|dst_node|anchor`, nulls empty.
124    #[must_use]
125    pub fn key(&self) -> String {
126        edge_key(
127            self.src_block.as_deref(),
128            self.src_field.as_deref(),
129            &self.predicate,
130            &self.dst_node,
131            self.anchor.as_deref(),
132        )
133    }
134}
135
136fn edge_key(
137    src_block: Option<&str>,
138    src_field: Option<&str>,
139    predicate: &str,
140    dst_node: &str,
141    anchor: Option<&str>,
142) -> String {
143    format!(
144        "{}|{}|{predicate}|{dst_node}|{}",
145        src_block.unwrap_or(""),
146        src_field.unwrap_or(""),
147        anchor.unwrap_or("")
148    )
149}
150
151/// The repo's `external_nodes` row for `uri`, **minting `x`** and inserting
152/// `(node_id, repo_id, uri, title = null)` when absent.
153pub fn resolve_external(
154    conn: &Connection,
155    minter: &mut Mint<'_>,
156    repo_id: &str,
157    uri: &str,
158) -> Result<String> {
159    let existing: Option<String> = conn
160        .query_row(
161            "SELECT node_id FROM external_nodes WHERE repo_id = ?1 AND uri = ?2",
162            params![repo_id, uri],
163            |r| r.get(0),
164        )
165        .optional()?;
166    if let Some(id) = existing {
167        return Ok(id);
168    }
169    let id = minter.mint("x")?;
170    conn.execute(
171        "INSERT INTO external_nodes (node_id, repo_id, uri) VALUES (?1, ?2, ?3)",
172        params![id, repo_id, uri],
173    )?;
174    Ok(id)
175}
176
177/// A repo path (one leading `/` stripped) to the live doc's id at that path,
178/// else `phantom:<path>`.
179pub fn resolve_doc_path(conn: &Connection, repo_id: &str, path: &str) -> Result<String> {
180    let canonical = canonical_path(path);
181    let doc: Option<String> = conn
182        .query_row(
183            "SELECT doc_id FROM docs WHERE repo_id = ?1 AND path = ?2 AND deleted_commit IS NULL",
184            params![repo_id, canonical],
185            |r| r.get(0),
186        )
187        .optional()?;
188    Ok(doc.unwrap_or_else(|| format!("phantom:{canonical}")))
189}
190
191/// §3.2: resolve each descriptor's target, in order — external URIs to
192/// `external_nodes` (minting `x`), a pure fragment to the source document,
193/// anything else through relative resolution against the document's
194/// directory to a live doc or a phantom, `dst_kind` forced to `document`.
195pub fn resolve_edges(
196    conn: &Connection,
197    minter: &mut Mint<'_>,
198    repo_id: &str,
199    src_doc: &str,
200    doc_path: &str,
201    descriptors: &[EdgeDescriptor],
202) -> Result<Vec<ResolvedEdge>> {
203    let dir = doc_dir(doc_path);
204    let mut out = Vec::with_capacity(descriptors.len());
205    for e in descriptors {
206        let (dst_kind, dst_node) = if e.dst_kind == DstKind::External {
207            (
208                DstKind::External,
209                resolve_external(conn, minter, repo_id, &e.target)?,
210            )
211        } else if e.target.is_empty() {
212            (DstKind::Document, src_doc.to_owned())
213        } else {
214            (
215                DstKind::Document,
216                resolve_doc_path(conn, repo_id, &resolve_relative(&e.target, dir))?,
217            )
218        };
219        out.push(ResolvedEdge {
220            src_block: e.src_block.clone(),
221            src_field: e.src_field.clone(),
222            predicate: e.predicate.clone(),
223            dst_kind,
224            dst_node,
225            anchor: e.anchor.clone(),
226            provenance: e.provenance,
227        });
228    }
229    Ok(out)
230}
231
232// ---- intervals (§3.3) and rollup (§3.4) -----------------------------------------------
233
234/// §3.3: reconcile the wanted edge set against the document's open rows —
235/// **mint `e`** and insert for each wanted key not open (wanted order,
236/// `from_commit = commit_id`), then close (`to_commit = commit_id`) every open
237/// row whose key is not wanted or that duplicates an earlier kept row, in
238/// `rowid` order. Then rebuild the rollup (§3.4).
239pub fn maintain_edges(
240    conn: &Connection,
241    minter: &mut Mint<'_>,
242    repo_id: &str,
243    src_doc: &str,
244    commit_id: &str,
245    edges: &[ResolvedEdge],
246) -> Result<()> {
247    let open: Vec<(String, String)> = {
248        let mut stmt = conn.prepare_cached(
249            "SELECT edge_id, src_block, src_field, predicate, dst_node, anchor FROM edges
250              WHERE src_doc = ?1 AND to_commit IS NULL ORDER BY rowid",
251        )?;
252        let it = stmt.query_map(params![src_doc], |r| {
253            let src_block: Option<String> = r.get(1)?;
254            let src_field: Option<String> = r.get(2)?;
255            let predicate: String = r.get(3)?;
256            let dst_node: String = r.get(4)?;
257            let anchor: Option<String> = r.get(5)?;
258            Ok((
259                r.get::<_, String>(0)?,
260                edge_key(
261                    src_block.as_deref(),
262                    src_field.as_deref(),
263                    &predicate,
264                    &dst_node,
265                    anchor.as_deref(),
266                ),
267            ))
268        })?;
269        it.collect::<std::result::Result<Vec<_>, _>>()?
270    };
271    let open_keys: HashSet<&str> = open.iter().map(|(_, k)| k.as_str()).collect();
272
273    let mut wanted: HashSet<String> = HashSet::new();
274    {
275        let mut insert = conn.prepare_cached(
276            "INSERT INTO edges (edge_id, repo_id, src_doc, src_block, src_field, predicate, dst_kind, dst_node, anchor, provenance, from_commit, to_commit)
277             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, NULL)",
278        )?;
279        for e in edges {
280            let key = e.key();
281            if !wanted.insert(key.clone()) {
282                continue;
283            }
284            if open_keys.contains(key.as_str()) {
285                continue;
286            }
287            insert.execute(params![
288                minter.mint("e")?,
289                repo_id,
290                src_doc,
291                e.src_block,
292                e.src_field,
293                e.predicate,
294                e.dst_kind.as_str(),
295                e.dst_node,
296                e.anchor,
297                e.provenance.as_str(),
298                commit_id,
299            ])?;
300        }
301    }
302    {
303        let mut close =
304            conn.prepare_cached("UPDATE edges SET to_commit = ?1 WHERE edge_id = ?2")?;
305        let mut kept: HashSet<&str> = HashSet::new();
306        for (edge_id, key) in &open {
307            if wanted.contains(key) && kept.insert(key.as_str()) {
308                continue;
309            }
310            close.execute(params![commit_id, edge_id])?;
311        }
312    }
313    rebuild_doc_edges(conn, src_doc)
314}
315
316/// §3.4: replace the document's `doc_edges` with one row per group of its
317/// open edges by `(predicate, dst_node, dst_kind)`: `count` = the group
318/// size, `samples` = up to three non-null `src_block`s in edge `rowid` order.
319pub fn rebuild_doc_edges(conn: &Connection, src_doc: &str) -> Result<()> {
320    conn.execute("DELETE FROM doc_edges WHERE src_doc = ?1", params![src_doc])?;
321    let rows: Vec<(String, String, String, Option<String>)> = {
322        let mut stmt = conn.prepare_cached(
323            "SELECT predicate, dst_node, dst_kind, src_block FROM edges
324              WHERE src_doc = ?1 AND to_commit IS NULL ORDER BY rowid",
325        )?;
326        let it = stmt.query_map(params![src_doc], |r| {
327            Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))
328        })?;
329        it.collect::<std::result::Result<Vec<_>, _>>()?
330    };
331    let mut order: Vec<(String, String, String)> = Vec::new();
332    let mut groups: HashMap<(String, String, String), (i64, Vec<String>)> = HashMap::new();
333    for (predicate, dst_node, dst_kind, src_block) in rows {
334        let key = (predicate, dst_node, dst_kind);
335        let entry = groups.entry(key.clone()).or_insert_with(|| {
336            order.push(key);
337            (0, Vec::new())
338        });
339        entry.0 += 1;
340        if let Some(b) = src_block {
341            entry.1.push(b);
342        }
343    }
344    let mut insert = conn.prepare_cached(
345        "INSERT INTO doc_edges (src_doc, predicate, dst_node, dst_kind, count, samples) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
346    )?;
347    for key in order {
348        let (count, samples) = &groups[&key];
349        let samples: Vec<&str> = samples.iter().take(3).map(String::as_str).collect();
350        insert.execute(params![
351            src_doc,
352            key.0,
353            key.1,
354            key.2,
355            count,
356            serde_json::to_string(&samples).expect("strings serialize"),
357        ])?;
358    }
359    Ok(())
360}
361
362// ---- phantom adoption (§3.5) ---------------------------------------------------------
363
364/// §3.5: when a document row is created at `path`, rewrite every open edge
365/// with `dst_node = "phantom:" + path` to `doc_id` in place and recompute
366/// the rollup of each affected source document.
367pub fn adopt_phantoms(conn: &Connection, path: &str, doc_id: &str) -> Result<()> {
368    let phantom = format!("phantom:{}", canonical_path(path));
369    let affected: Vec<String> = {
370        let mut stmt = conn.prepare_cached(
371            "SELECT DISTINCT src_doc FROM edges WHERE dst_node = ?1 AND to_commit IS NULL",
372        )?;
373        let it = stmt.query_map(params![phantom], |r| r.get(0))?;
374        it.collect::<std::result::Result<Vec<_>, _>>()?
375    };
376    if affected.is_empty() {
377        return Ok(());
378    }
379    conn.execute(
380        "UPDATE edges SET dst_node = ?1 WHERE dst_node = ?2 AND to_commit IS NULL",
381        params![doc_id, phantom],
382    )?;
383    for src in &affected {
384        rebuild_doc_edges(conn, src)?;
385    }
386    Ok(())
387}
388
389#[cfg(test)]
390mod tests {
391    use omgbase_graph::node_id;
392    use omgbase_reconcile::Config;
393    use rusqlite::types::Value as Sql;
394
395    use crate::{SequentialMinter, Store};
396
397    const T0: &str = "2026-09-26T10:00:00.000Z";
398    const T1: &str = "2026-09-26T10:01:00.000Z";
399    const T2: &str = "2026-09-26T10:02:00.000Z";
400
401    fn fixture() -> (Store, String) {
402        let mut store =
403            Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
404        let repo = store.create_repo("fixture").unwrap();
405        (store, repo)
406    }
407
408    fn observe(store: &mut Store, repo: &str, path: &str, source: &str, ts: &str) -> String {
409        store
410            .observe_one(repo, path, source, ts, &Config::default())
411            .unwrap()
412            .doc_id
413    }
414
415    fn rows(store: &Store, sql: &str) -> Vec<Vec<Sql>> {
416        let mut stmt = store.conn().prepare(sql).unwrap();
417        let n = stmt.column_count();
418        stmt.query_map([], |r| (0..n).map(|i| r.get::<_, Sql>(i)).collect())
419            .unwrap()
420            .map(|r| r.unwrap())
421            .collect()
422    }
423
424    fn texts(store: &Store, sql: &str) -> Vec<String> {
425        rows(store, sql)
426            .into_iter()
427            .map(|r| match &r[0] {
428                Sql::Text(t) => t.clone(),
429                Sql::Null => "null".to_owned(),
430                other => format!("{other:?}"),
431            })
432            .collect()
433    }
434
435    #[test]
436    fn nodes_are_written_with_ids_spans_and_fts() {
437        let (mut store, repo) = fixture();
438        let doc = observe(
439            &mut store,
440            &repo,
441            "a.md",
442            "# Tïtle\n\nSee [x](y.md) and [[w]] ^a\n\n- [ ] todo\n\n## Sub\n\nk:: v\n",
443            T0,
444        );
445        let kinds = texts(&store, "SELECT kind FROM nodes ORDER BY rowid");
446        assert_eq!(
447            kinds,
448            [
449                "md:link",
450                "md:wikilink",
451                "md:anchor",
452                "md:task",
453                "md:inline_field",
454                "md:section",
455                "md:section"
456            ]
457        );
458        let link = rows(
459            &store,
460            "SELECT node_id, block_id, name, value, span_start, span_end, attrs FROM nodes WHERE kind = 'md:link'",
461        );
462        assert_eq!(link[0][0], Sql::Text(node_id(&doc, "b_1", "md:link", 0)));
463        assert_eq!(link[0][1], Sql::Text("b_1".into()));
464        assert_eq!(link[0][2], Sql::Text("x".into()));
465        assert_eq!(link[0][3], Sql::Text("y.md".into()));
466        assert_eq!(link[0][4], Sql::Integer(4));
467        assert_eq!(link[0][5], Sql::Integer(13));
468        assert_eq!(link[0][6], Sql::Text("{}".into()));
469        let sections = rows(
470            &store,
471            "SELECT block_id, name, value, span_start, attrs FROM nodes WHERE kind = 'md:section' ORDER BY rowid",
472        );
473        assert_eq!(sections[0][0], Sql::Text("b_0".into()));
474        assert_eq!(sections[0][1], Sql::Text("Tïtle".into()));
475        assert_eq!(sections[0][2], Sql::Null);
476        assert_eq!(sections[0][3], Sql::Null);
477        assert_eq!(
478            sections[0][4],
479            Sql::Text(r#"{"level":1,"first_ordinal":0,"last_ordinal":4}"#.into())
480        );
481        let task = rows(
482            &store,
483            "SELECT attrs, value FROM nodes WHERE kind = 'md:task'",
484        );
485        assert_eq!(task[0][0], Sql::Text(r#"{"checked":false}"#.into()));
486        assert_eq!(task[0][1], Sql::Text("todo".into()));
487        // FTS rows exist and follow the nodes.
488        let hits = texts(
489            &store,
490            "SELECT n.kind FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid WHERE nodes_fts MATCH 'todo'",
491        );
492        assert_eq!(hits, ["md:task"]);
493        // A re-ingest replaces the rows (and the FTS entries) wholesale.
494        observe(&mut store, &repo, "a.md", "# Title\n\nplain\n", T1);
495        assert_eq!(
496            texts(&store, "SELECT kind FROM nodes ORDER BY rowid"),
497            ["md:section"]
498        );
499        assert!(texts(
500            &store,
501            "SELECT n.kind FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid WHERE nodes_fts MATCH 'todo'"
502        )
503        .is_empty());
504        let stale: i64 = store
505            .conn()
506            .query_row(
507                "SELECT count(*) FROM nodes_fts WHERE nodes_fts MATCH 'todo'",
508                [],
509                |r| r.get(0),
510            )
511            .unwrap();
512        assert_eq!(
513            stale, 0,
514            "the external-content delete removed the index entry"
515        );
516    }
517
518    #[test]
519    fn edges_open_close_and_roll_up() {
520        let (mut store, repo) = fixture();
521        let a = observe(
522            &mut store,
523            &repo,
524            "a.md",
525            "See [b](b.md) and [b again](./b.md)\n\n<https://x.com/> and https://x.com\n",
526            T0,
527        );
528        // One phantom edge per (block, key) — two spellings of b.md resolve to
529        // one key; two spellings of the URI to one external node.
530        let edges = rows(
531            &store,
532            "SELECT edge_id, src_block, predicate, dst_kind, dst_node, provenance, from_commit, to_commit FROM edges ORDER BY rowid",
533        );
534        assert_eq!(edges.len(), 2);
535        assert_eq!(edges[0][0], Sql::Text("e_0".into()));
536        assert_eq!(edges[0][1], Sql::Text("b_0".into()));
537        assert_eq!(edges[0][4], Sql::Text("phantom:b.md".into()));
538        assert_eq!(edges[0][6], Sql::Text("c_0".into()));
539        assert_eq!(edges[0][7], Sql::Null);
540        assert_eq!(edges[1][1], Sql::Text("b_1".into()));
541        assert_eq!(edges[1][3], Sql::Text("external".into()));
542        assert_eq!(edges[1][4], Sql::Text("x_0".into()));
543        assert_eq!(
544            texts(&store, "SELECT uri FROM external_nodes"),
545            ["https://x.com"]
546        );
547        let roll = rows(
548            &store,
549            "SELECT predicate, dst_node, dst_kind, count, samples FROM doc_edges ORDER BY dst_node",
550        );
551        assert_eq!(roll.len(), 2);
552        assert_eq!(roll[0][1], Sql::Text("phantom:b.md".into()));
553        assert_eq!(roll[0][3], Sql::Integer(1));
554        assert_eq!(roll[0][4], Sql::Text(r#"["b_0"]"#.into()));
555
556        // Creating b.md adopts the phantom in place (same edge id, same from_commit).
557        let b = observe(&mut store, &repo, "b.md", "# B\n", T1);
558        let adopted = rows(
559            &store,
560            "SELECT edge_id, dst_node, from_commit, to_commit FROM edges WHERE edge_id = 'e_0'",
561        );
562        assert_eq!(adopted[0][1], Sql::Text(b.clone()));
563        assert_eq!(adopted[0][2], Sql::Text("c_0".into()));
564        assert_eq!(adopted[0][3], Sql::Null);
565        assert_eq!(
566            texts(
567                &store,
568                "SELECT dst_node FROM doc_edges WHERE src_doc = 'd_0' ORDER BY dst_node"
569            ),
570            [b.as_str(), "x_0"]
571        );
572
573        // Editing a.md: the first paragraph is unchanged, so its edge to b stays
574        // open untouched; the URL paragraph is replaced (its edge closes) by a
575        // self-fragment link; a new paragraph carries an inline relation to the
576        // same URI (the external node is reused, and the URL value is also a
577        // bare URL — two edges, different keys).
578        observe(
579            &mut store,
580            &repo,
581            "a.md",
582            "See [b](b.md) and [b again](./b.md)\n\nNow [here](#Top) only\n\nrel:: https://x.com/\n",
583            T2,
584        );
585        let edges = rows(
586            &store,
587            "SELECT edge_id, src_block, src_field, predicate, dst_kind, dst_node, anchor, from_commit, to_commit FROM edges ORDER BY rowid",
588        );
589        assert_eq!(edges.len(), 5);
590        assert_eq!(
591            (&edges[0][7], &edges[0][8]),
592            (&Sql::Text("c_0".into()), &Sql::Null)
593        );
594        assert_eq!(
595            edges[1][8],
596            Sql::Text("c_2".into()),
597            "the bare-URL edge closed"
598        );
599        assert_eq!(edges[2][0], Sql::Text("e_2".into()));
600        assert_eq!(
601            edges[2][5],
602            Sql::Text(a.clone()),
603            "a pure fragment targets the doc itself"
604        );
605        assert_eq!(edges[2][6], Sql::Text("Top".into()));
606        assert_eq!(edges[2][7], Sql::Text("c_2".into()));
607        assert_eq!(edges[3][2], Sql::Text("rel".into()));
608        assert_eq!(edges[3][3], Sql::Text("rel".into()));
609        assert_eq!(
610            edges[3][5],
611            Sql::Text("x_0".into()),
612            "external node reused, none minted"
613        );
614        assert_eq!(edges[4][3], Sql::Text("references".into()));
615        assert_eq!(edges[4][5], Sql::Text("x_0".into()));
616        assert_eq!(edges[4][0], Sql::Text("e_4".into()));
617        assert_eq!(texts(&store, "SELECT node_id FROM external_nodes"), ["x_0"]);
618        let roll = rows(
619            &store,
620            "SELECT predicate, dst_node, count FROM doc_edges WHERE src_doc = 'd_0' ORDER BY predicate, dst_node",
621        );
622        assert_eq!(roll.len(), 4);
623        assert_eq!(roll[0][0], Sql::Text("references".into()));
624        assert_eq!(roll[3][0], Sql::Text("rel".into()));
625    }
626
627    #[test]
628    fn containers_count_once_and_ref_targets_are_documents() {
629        let (mut store, repo) = fixture();
630        observe(&mut store, &repo, "a.md", "- see [[n^blk]]\n", T0);
631        let edges = rows(
632            &store,
633            "SELECT src_block, dst_kind, dst_node, anchor FROM edges ORDER BY rowid",
634        );
635        // spec/graph 1.1: the item's edge only (the list's own text is blank).
636        assert_eq!(edges.len(), 1);
637        assert_eq!(edges[0][0], Sql::Text("b_1".into()));
638        assert_eq!(edges[0][1], Sql::Text("document".into()));
639        assert_eq!(edges[0][2], Sql::Text("phantom:n".into()));
640        assert_eq!(edges[0][3], Sql::Text("blk".into()));
641        let roll = rows(&store, "SELECT count, samples FROM doc_edges");
642        assert_eq!(roll[0][0], Sql::Integer(1));
643        assert_eq!(roll[0][1], Sql::Text(r#"["b_1"]"#.into()));
644    }
645
646    #[test]
647    fn frontmatter_relations_resolve_relative_to_the_root_and_carry_no_block() {
648        let (mut store, repo) = fixture();
649        observe(
650            &mut store,
651            &repo,
652            "dir/a.md",
653            "---\nOwner: /people/x.md\nrels: [\"[[n]]\", \"[[m]]\"]\n---\n\nSee [s](./s.md) and [u](../u.md)\n",
654            T0,
655        );
656        let edges = rows(
657            &store,
658            "SELECT src_block, src_field, predicate, dst_node, provenance FROM edges ORDER BY rowid",
659        );
660        let view: Vec<(String, String)> = edges
661            .iter()
662            .map(|r| {
663                let f = |v: &Sql| match v {
664                    Sql::Text(t) => t.clone(),
665                    _ => "null".to_owned(),
666                };
667                (f(&r[2]), f(&r[3]))
668            })
669            .collect();
670        assert_eq!(
671            view,
672            [
673                ("references".to_owned(), "phantom:dir/s.md".to_owned()),
674                ("references".to_owned(), "phantom:u.md".to_owned()),
675                ("Owner".to_owned(), "phantom:people/x.md".to_owned()),
676                ("rels".to_owned(), "phantom:n".to_owned()),
677                ("rels".to_owned(), "phantom:m".to_owned()),
678            ]
679        );
680        assert_eq!(edges[2][0], Sql::Null);
681        assert_eq!(edges[2][4], Sql::Text("frontmatter".into()));
682        let samples = texts(
683            &store,
684            "SELECT samples FROM doc_edges WHERE predicate = 'Owner'",
685        );
686        assert_eq!(samples, ["[]"]);
687    }
688
689    #[test]
690    fn deletion_leaves_the_graph_rows() {
691        let (mut store, repo) = fixture();
692        observe(&mut store, &repo, "a.md", "[b](b.md)\n", T0);
693        observe(&mut store, &repo, "b.md", "# B\n\n[a](a.md)\n", T1);
694        assert_eq!(
695            texts(&store, "SELECT dst_node FROM edges ORDER BY rowid"),
696            ["d_1", "d_0"]
697        );
698        store.observe_delete(&repo, "b.md", T2).unwrap();
699        assert_eq!(
700            texts(
701                &store,
702                "SELECT dst_node FROM edges WHERE to_commit IS NULL ORDER BY rowid"
703            ),
704            ["d_1", "d_0"],
705            "edges stay open and keep the tombstoned doc's id"
706        );
707        let counts: Vec<i64> = ["nodes", "doc_edges"]
708            .iter()
709            .map(|t| {
710                store
711                    .conn()
712                    .query_row(&format!("SELECT count(*) FROM {t} WHERE 1"), [], |r| {
713                        r.get(0)
714                    })
715                    .unwrap()
716            })
717            .collect();
718        assert_eq!(
719            counts,
720            [3, 2],
721            "a's link; b's section and link; both rollups"
722        );
723        // While b.md is tombstoned, a new link to it resolves to a phantom.
724        observe(&mut store, &repo, "c.md", "[b](b.md)\n", T2);
725        assert_eq!(
726            texts(&store, "SELECT dst_node FROM edges WHERE src_doc = 'd_2'"),
727            ["phantom:b.md"]
728        );
729        // Re-creation revives the row: it becomes live, so it adopts that
730        // phantom (§3.5); the edge that kept d_1 all along is untouched.
731        observe(
732            &mut store,
733            &repo,
734            "b.md",
735            "# B2\n",
736            "2026-09-26T10:03:00.000Z",
737        );
738        assert_eq!(
739            texts(
740                &store,
741                "SELECT dst_node FROM edges WHERE to_commit IS NULL AND dst_node LIKE 'd_%' AND src_doc != 'd_1' ORDER BY src_doc"
742            ),
743            ["d_1", "d_1"]
744        );
745        assert_eq!(
746            texts(
747                &store,
748                "SELECT dst_node FROM doc_edges WHERE src_doc = 'd_2'"
749            ),
750            ["d_1"],
751            "the affected rollup is recomputed"
752        );
753    }
754
755    #[test]
756    fn mint_order_is_x_during_resolution_then_e() {
757        let (mut store, repo) = fixture();
758        observe(
759            &mut store,
760            &repo,
761            "a.md",
762            "<https://b.com> then <https://a.com>\n",
763            T0,
764        );
765        let ext = rows(
766            &store,
767            "SELECT node_id, uri FROM external_nodes ORDER BY node_id",
768        );
769        assert_eq!(
770            ext[0],
771            vec![Sql::Text("x_0".into()), Sql::Text("https://b.com".into())]
772        );
773        assert_eq!(
774            ext[1],
775            vec![Sql::Text("x_1".into()), Sql::Text("https://a.com".into())]
776        );
777        assert_eq!(
778            texts(&store, "SELECT edge_id FROM edges ORDER BY rowid"),
779            ["e_0", "e_1"]
780        );
781    }
782}
omgbase_store/graph.rs

omgbase_store/
graph.rs
