use std::collections::{HashMap, HashSet};
use omgbase_graph::{
DstKind, EdgeDescriptor, ProjectedNode, Provenance, canonical_path, doc_dir, node_rows,
resolve_relative,
};
use rusqlite::{Connection, OptionalExtension, params};
use crate::error::Result;
use crate::mint::Mint;
pub fn project_section_nodes(conn: &Connection, doc_id: &str) -> Result<Vec<ProjectedNode>> {
let mut stmt = conn.prepare_cached(
"SELECT s.heading_block, s.level, s.first_ordinal, s.last_ordinal, hb.text
FROM sections s
JOIN blocks hb ON hb.block_id = s.heading_block AND hb.doc_id = s.doc_id
WHERE s.doc_id = ?1 AND hb.deleted_commit IS NULL
ORDER BY s.first_ordinal",
)?;
let rows = stmt.query_map(params![doc_id], |r| {
Ok(ProjectedNode::section(
&r.get::<_, String>(0)?,
&r.get::<_, String>(4)?,
r.get(1)?,
r.get(2)?,
r.get(3)?,
))
})?;
Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
}
pub fn delete_doc_nodes(conn: &Connection, doc_id: &str) -> Result<()> {
let rows: Vec<(i64, Option<String>, Option<String>)> = {
let mut stmt =
conn.prepare_cached("SELECT rowid, name, value FROM nodes WHERE doc_id = ?1")?;
let it = stmt.query_map(params![doc_id], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
it.collect::<std::result::Result<Vec<_>, _>>()?
};
let mut del = conn.prepare_cached(
"INSERT INTO nodes_fts(nodes_fts, rowid, name, value) VALUES('delete', ?1, ?2, ?3)",
)?;
for (rowid, name, value) in rows {
del.execute(params![
rowid,
name.unwrap_or_default(),
value.unwrap_or_default()
])?;
}
conn.execute("DELETE FROM nodes WHERE doc_id = ?1", params![doc_id])?;
Ok(())
}
pub fn write_doc_nodes(
conn: &Connection,
repo_id: &str,
doc_id: &str,
nodes: &[ProjectedNode],
) -> Result<usize> {
delete_doc_nodes(conn, doc_id)?;
let mut insert = conn.prepare_cached(
"INSERT INTO nodes (node_id, repo_id, doc_id, block_id, kind, name, value, span_start, span_end, attrs)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
)?;
let mut fts =
conn.prepare_cached("INSERT INTO nodes_fts(rowid, name, value) VALUES (?1, ?2, ?3)")?;
for row in node_rows(doc_id, nodes) {
let n = row.node;
insert.execute(params![
row.node_id,
repo_id,
doc_id,
n.block_id,
n.kind.as_str(),
n.name,
n.value,
n.span.map(|(s, _)| s as i64),
n.span.map(|(_, e)| e as i64),
serde_json::Value::Object(n.attrs.clone()).to_string(),
])?;
let rowid = conn.last_insert_rowid();
fts.execute(params![
rowid,
n.name.as_deref().unwrap_or(""),
n.value.as_deref().unwrap_or("")
])?;
}
Ok(nodes.len())
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ResolvedEdge {
pub src_block: Option<String>,
pub src_field: Option<String>,
pub predicate: String,
pub dst_kind: DstKind,
pub dst_node: String,
pub anchor: Option<String>,
pub provenance: Provenance,
}
impl ResolvedEdge {
#[must_use]
pub fn key(&self) -> String {
edge_key(
self.src_block.as_deref(),
self.src_field.as_deref(),
&self.predicate,
&self.dst_node,
self.anchor.as_deref(),
)
}
}
fn edge_key(
src_block: Option<&str>,
src_field: Option<&str>,
predicate: &str,
dst_node: &str,
anchor: Option<&str>,
) -> String {
format!(
"{}|{}|{predicate}|{dst_node}|{}",
src_block.unwrap_or(""),
src_field.unwrap_or(""),
anchor.unwrap_or("")
)
}
pub fn resolve_external(
conn: &Connection,
minter: &mut Mint<'_>,
repo_id: &str,
uri: &str,
) -> Result<String> {
let existing: Option<String> = conn
.query_row(
"SELECT node_id FROM external_nodes WHERE repo_id = ?1 AND uri = ?2",
params![repo_id, uri],
|r| r.get(0),
)
.optional()?;
if let Some(id) = existing {
return Ok(id);
}
let id = minter.mint("x")?;
conn.execute(
"INSERT INTO external_nodes (node_id, repo_id, uri) VALUES (?1, ?2, ?3)",
params![id, repo_id, uri],
)?;
Ok(id)
}
pub fn resolve_doc_path(conn: &Connection, repo_id: &str, path: &str) -> Result<String> {
let canonical = canonical_path(path);
let doc: Option<String> = conn
.query_row(
"SELECT doc_id FROM docs WHERE repo_id = ?1 AND path = ?2 AND deleted_commit IS NULL",
params![repo_id, canonical],
|r| r.get(0),
)
.optional()?;
Ok(doc.unwrap_or_else(|| format!("phantom:{canonical}")))
}
pub fn resolve_edges(
conn: &Connection,
minter: &mut Mint<'_>,
repo_id: &str,
src_doc: &str,
doc_path: &str,
descriptors: &[EdgeDescriptor],
) -> Result<Vec<ResolvedEdge>> {
let dir = doc_dir(doc_path);
let mut out = Vec::with_capacity(descriptors.len());
for e in descriptors {
let (dst_kind, dst_node) = if e.dst_kind == DstKind::External {
(
DstKind::External,
resolve_external(conn, minter, repo_id, &e.target)?,
)
} else if e.target.is_empty() {
(DstKind::Document, src_doc.to_owned())
} else {
(
DstKind::Document,
resolve_doc_path(conn, repo_id, &resolve_relative(&e.target, dir))?,
)
};
out.push(ResolvedEdge {
src_block: e.src_block.clone(),
src_field: e.src_field.clone(),
predicate: e.predicate.clone(),
dst_kind,
dst_node,
anchor: e.anchor.clone(),
provenance: e.provenance,
});
}
Ok(out)
}
pub fn maintain_edges(
conn: &Connection,
minter: &mut Mint<'_>,
repo_id: &str,
src_doc: &str,
commit_id: &str,
edges: &[ResolvedEdge],
) -> Result<()> {
let open: Vec<(String, String)> = {
let mut stmt = conn.prepare_cached(
"SELECT edge_id, src_block, src_field, predicate, dst_node, anchor FROM edges
WHERE src_doc = ?1 AND to_commit IS NULL ORDER BY rowid",
)?;
let it = stmt.query_map(params![src_doc], |r| {
let src_block: Option<String> = r.get(1)?;
let src_field: Option<String> = r.get(2)?;
let predicate: String = r.get(3)?;
let dst_node: String = r.get(4)?;
let anchor: Option<String> = r.get(5)?;
Ok((
r.get::<_, String>(0)?,
edge_key(
src_block.as_deref(),
src_field.as_deref(),
&predicate,
&dst_node,
anchor.as_deref(),
),
))
})?;
it.collect::<std::result::Result<Vec<_>, _>>()?
};
let open_keys: HashSet<&str> = open.iter().map(|(_, k)| k.as_str()).collect();
let mut wanted: HashSet<String> = HashSet::new();
{
let mut insert = conn.prepare_cached(
"INSERT INTO edges (edge_id, repo_id, src_doc, src_block, src_field, predicate, dst_kind, dst_node, anchor, provenance, from_commit, to_commit)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, NULL)",
)?;
for e in edges {
let key = e.key();
if !wanted.insert(key.clone()) {
continue;
}
if open_keys.contains(key.as_str()) {
continue;
}
insert.execute(params![
minter.mint("e")?,
repo_id,
src_doc,
e.src_block,
e.src_field,
e.predicate,
e.dst_kind.as_str(),
e.dst_node,
e.anchor,
e.provenance.as_str(),
commit_id,
])?;
}
}
{
let mut close =
conn.prepare_cached("UPDATE edges SET to_commit = ?1 WHERE edge_id = ?2")?;
let mut kept: HashSet<&str> = HashSet::new();
for (edge_id, key) in &open {
if wanted.contains(key) && kept.insert(key.as_str()) {
continue;
}
close.execute(params![commit_id, edge_id])?;
}
}
rebuild_doc_edges(conn, src_doc)
}
pub fn rebuild_doc_edges(conn: &Connection, src_doc: &str) -> Result<()> {
conn.execute("DELETE FROM doc_edges WHERE src_doc = ?1", params![src_doc])?;
let rows: Vec<(String, String, String, Option<String>)> = {
let mut stmt = conn.prepare_cached(
"SELECT predicate, dst_node, dst_kind, src_block FROM edges
WHERE src_doc = ?1 AND to_commit IS NULL ORDER BY rowid",
)?;
let it = stmt.query_map(params![src_doc], |r| {
Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?))
})?;
it.collect::<std::result::Result<Vec<_>, _>>()?
};
let mut order: Vec<(String, String, String)> = Vec::new();
let mut groups: HashMap<(String, String, String), (i64, Vec<String>)> = HashMap::new();
for (predicate, dst_node, dst_kind, src_block) in rows {
let key = (predicate, dst_node, dst_kind);
let entry = groups.entry(key.clone()).or_insert_with(|| {
order.push(key);
(0, Vec::new())
});
entry.0 += 1;
if let Some(b) = src_block {
entry.1.push(b);
}
}
let mut insert = conn.prepare_cached(
"INSERT INTO doc_edges (src_doc, predicate, dst_node, dst_kind, count, samples) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
)?;
for key in order {
let (count, samples) = &groups[&key];
let samples: Vec<&str> = samples.iter().take(3).map(String::as_str).collect();
insert.execute(params![
src_doc,
key.0,
key.1,
key.2,
count,
serde_json::to_string(&samples).expect("strings serialize"),
])?;
}
Ok(())
}
pub fn adopt_phantoms(conn: &Connection, path: &str, doc_id: &str) -> Result<()> {
let phantom = format!("phantom:{}", canonical_path(path));
let affected: Vec<String> = {
let mut stmt = conn.prepare_cached(
"SELECT DISTINCT src_doc FROM edges WHERE dst_node = ?1 AND to_commit IS NULL",
)?;
let it = stmt.query_map(params![phantom], |r| r.get(0))?;
it.collect::<std::result::Result<Vec<_>, _>>()?
};
if affected.is_empty() {
return Ok(());
}
conn.execute(
"UPDATE edges SET dst_node = ?1 WHERE dst_node = ?2 AND to_commit IS NULL",
params![doc_id, phantom],
)?;
for src in &affected {
rebuild_doc_edges(conn, src)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use omgbase_graph::node_id;
use omgbase_reconcile::Config;
use rusqlite::types::Value as Sql;
use crate::{SequentialMinter, Store};
const T0: &str = "2026-09-26T10:00:00.000Z";
const T1: &str = "2026-09-26T10:01:00.000Z";
const T2: &str = "2026-09-26T10:02:00.000Z";
fn fixture() -> (Store, String) {
let mut store =
Store::open_in_memory_with_minter(Box::new(SequentialMinter::new())).unwrap();
let repo = store.create_repo("fixture").unwrap();
(store, repo)
}
fn observe(store: &mut Store, repo: &str, path: &str, source: &str, ts: &str) -> String {
store
.observe_one(repo, path, source, ts, &Config::default())
.unwrap()
.doc_id
}
fn rows(store: &Store, sql: &str) -> Vec<Vec<Sql>> {
let mut stmt = store.conn().prepare(sql).unwrap();
let n = stmt.column_count();
stmt.query_map([], |r| (0..n).map(|i| r.get::<_, Sql>(i)).collect())
.unwrap()
.map(|r| r.unwrap())
.collect()
}
fn texts(store: &Store, sql: &str) -> Vec<String> {
rows(store, sql)
.into_iter()
.map(|r| match &r[0] {
Sql::Text(t) => t.clone(),
Sql::Null => "null".to_owned(),
other => format!("{other:?}"),
})
.collect()
}
#[test]
fn nodes_are_written_with_ids_spans_and_fts() {
let (mut store, repo) = fixture();
let doc = observe(
&mut store,
&repo,
"a.md",
"# Tïtle\n\nSee [x](y.md) and [[w]] ^a\n\n- [ ] todo\n\n## Sub\n\nk:: v\n",
T0,
);
let kinds = texts(&store, "SELECT kind FROM nodes ORDER BY rowid");
assert_eq!(
kinds,
[
"md:link",
"md:wikilink",
"md:anchor",
"md:task",
"md:inline_field",
"md:section",
"md:section"
]
);
let link = rows(
&store,
"SELECT node_id, block_id, name, value, span_start, span_end, attrs FROM nodes WHERE kind = 'md:link'",
);
assert_eq!(link[0][0], Sql::Text(node_id(&doc, "b_1", "md:link", 0)));
assert_eq!(link[0][1], Sql::Text("b_1".into()));
assert_eq!(link[0][2], Sql::Text("x".into()));
assert_eq!(link[0][3], Sql::Text("y.md".into()));
assert_eq!(link[0][4], Sql::Integer(4));
assert_eq!(link[0][5], Sql::Integer(13));
assert_eq!(link[0][6], Sql::Text("{}".into()));
let sections = rows(
&store,
"SELECT block_id, name, value, span_start, attrs FROM nodes WHERE kind = 'md:section' ORDER BY rowid",
);
assert_eq!(sections[0][0], Sql::Text("b_0".into()));
assert_eq!(sections[0][1], Sql::Text("Tïtle".into()));
assert_eq!(sections[0][2], Sql::Null);
assert_eq!(sections[0][3], Sql::Null);
assert_eq!(
sections[0][4],
Sql::Text(r#"{"level":1,"first_ordinal":0,"last_ordinal":4}"#.into())
);
let task = rows(
&store,
"SELECT attrs, value FROM nodes WHERE kind = 'md:task'",
);
assert_eq!(task[0][0], Sql::Text(r#"{"checked":false}"#.into()));
assert_eq!(task[0][1], Sql::Text("todo".into()));
let hits = texts(
&store,
"SELECT n.kind FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid WHERE nodes_fts MATCH 'todo'",
);
assert_eq!(hits, ["md:task"]);
observe(&mut store, &repo, "a.md", "# Title\n\nplain\n", T1);
assert_eq!(
texts(&store, "SELECT kind FROM nodes ORDER BY rowid"),
["md:section"]
);
assert!(texts(
&store,
"SELECT n.kind FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid WHERE nodes_fts MATCH 'todo'"
)
.is_empty());
let stale: i64 = store
.conn()
.query_row(
"SELECT count(*) FROM nodes_fts WHERE nodes_fts MATCH 'todo'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(
stale, 0,
"the external-content delete removed the index entry"
);
}
#[test]
fn edges_open_close_and_roll_up() {
let (mut store, repo) = fixture();
let a = observe(
&mut store,
&repo,
"a.md",
"See [b](b.md) and [b again](./b.md)\n\n<https://x.com/> and https://x.com\n",
T0,
);
let edges = rows(
&store,
"SELECT edge_id, src_block, predicate, dst_kind, dst_node, provenance, from_commit, to_commit FROM edges ORDER BY rowid",
);
assert_eq!(edges.len(), 2);
assert_eq!(edges[0][0], Sql::Text("e_0".into()));
assert_eq!(edges[0][1], Sql::Text("b_0".into()));
assert_eq!(edges[0][4], Sql::Text("phantom:b.md".into()));
assert_eq!(edges[0][6], Sql::Text("c_0".into()));
assert_eq!(edges[0][7], Sql::Null);
assert_eq!(edges[1][1], Sql::Text("b_1".into()));
assert_eq!(edges[1][3], Sql::Text("external".into()));
assert_eq!(edges[1][4], Sql::Text("x_0".into()));
assert_eq!(
texts(&store, "SELECT uri FROM external_nodes"),
["https://x.com"]
);
let roll = rows(
&store,
"SELECT predicate, dst_node, dst_kind, count, samples FROM doc_edges ORDER BY dst_node",
);
assert_eq!(roll.len(), 2);
assert_eq!(roll[0][1], Sql::Text("phantom:b.md".into()));
assert_eq!(roll[0][3], Sql::Integer(1));
assert_eq!(roll[0][4], Sql::Text(r#"["b_0"]"#.into()));
let b = observe(&mut store, &repo, "b.md", "# B\n", T1);
let adopted = rows(
&store,
"SELECT edge_id, dst_node, from_commit, to_commit FROM edges WHERE edge_id = 'e_0'",
);
assert_eq!(adopted[0][1], Sql::Text(b.clone()));
assert_eq!(adopted[0][2], Sql::Text("c_0".into()));
assert_eq!(adopted[0][3], Sql::Null);
assert_eq!(
texts(
&store,
"SELECT dst_node FROM doc_edges WHERE src_doc = 'd_0' ORDER BY dst_node"
),
[b.as_str(), "x_0"]
);
observe(
&mut store,
&repo,
"a.md",
"See [b](b.md) and [b again](./b.md)\n\nNow [here](#Top) only\n\nrel:: https://x.com/\n",
T2,
);
let edges = rows(
&store,
"SELECT edge_id, src_block, src_field, predicate, dst_kind, dst_node, anchor, from_commit, to_commit FROM edges ORDER BY rowid",
);
assert_eq!(edges.len(), 5);
assert_eq!(
(&edges[0][7], &edges[0][8]),
(&Sql::Text("c_0".into()), &Sql::Null)
);
assert_eq!(
edges[1][8],
Sql::Text("c_2".into()),
"the bare-URL edge closed"
);
assert_eq!(edges[2][0], Sql::Text("e_2".into()));
assert_eq!(
edges[2][5],
Sql::Text(a.clone()),
"a pure fragment targets the doc itself"
);
assert_eq!(edges[2][6], Sql::Text("Top".into()));
assert_eq!(edges[2][7], Sql::Text("c_2".into()));
assert_eq!(edges[3][2], Sql::Text("rel".into()));
assert_eq!(edges[3][3], Sql::Text("rel".into()));
assert_eq!(
edges[3][5],
Sql::Text("x_0".into()),
"external node reused, none minted"
);
assert_eq!(edges[4][3], Sql::Text("references".into()));
assert_eq!(edges[4][5], Sql::Text("x_0".into()));
assert_eq!(edges[4][0], Sql::Text("e_4".into()));
assert_eq!(texts(&store, "SELECT node_id FROM external_nodes"), ["x_0"]);
let roll = rows(
&store,
"SELECT predicate, dst_node, count FROM doc_edges WHERE src_doc = 'd_0' ORDER BY predicate, dst_node",
);
assert_eq!(roll.len(), 4);
assert_eq!(roll[0][0], Sql::Text("references".into()));
assert_eq!(roll[3][0], Sql::Text("rel".into()));
}
#[test]
fn containers_count_once_and_ref_targets_are_documents() {
let (mut store, repo) = fixture();
observe(&mut store, &repo, "a.md", "- see [[n^blk]]\n", T0);
let edges = rows(
&store,
"SELECT src_block, dst_kind, dst_node, anchor FROM edges ORDER BY rowid",
);
assert_eq!(edges.len(), 1);
assert_eq!(edges[0][0], Sql::Text("b_1".into()));
assert_eq!(edges[0][1], Sql::Text("document".into()));
assert_eq!(edges[0][2], Sql::Text("phantom:n".into()));
assert_eq!(edges[0][3], Sql::Text("blk".into()));
let roll = rows(&store, "SELECT count, samples FROM doc_edges");
assert_eq!(roll[0][0], Sql::Integer(1));
assert_eq!(roll[0][1], Sql::Text(r#"["b_1"]"#.into()));
}
#[test]
fn frontmatter_relations_resolve_relative_to_the_root_and_carry_no_block() {
let (mut store, repo) = fixture();
observe(
&mut store,
&repo,
"dir/a.md",
"---\nOwner: /people/x.md\nrels: [\"[[n]]\", \"[[m]]\"]\n---\n\nSee [s](./s.md) and [u](../u.md)\n",
T0,
);
let edges = rows(
&store,
"SELECT src_block, src_field, predicate, dst_node, provenance FROM edges ORDER BY rowid",
);
let view: Vec<(String, String)> = edges
.iter()
.map(|r| {
let f = |v: &Sql| match v {
Sql::Text(t) => t.clone(),
_ => "null".to_owned(),
};
(f(&r[2]), f(&r[3]))
})
.collect();
assert_eq!(
view,
[
("references".to_owned(), "phantom:dir/s.md".to_owned()),
("references".to_owned(), "phantom:u.md".to_owned()),
("Owner".to_owned(), "phantom:people/x.md".to_owned()),
("rels".to_owned(), "phantom:n".to_owned()),
("rels".to_owned(), "phantom:m".to_owned()),
]
);
assert_eq!(edges[2][0], Sql::Null);
assert_eq!(edges[2][4], Sql::Text("frontmatter".into()));
let samples = texts(
&store,
"SELECT samples FROM doc_edges WHERE predicate = 'Owner'",
);
assert_eq!(samples, ["[]"]);
}
#[test]
fn deletion_leaves_the_graph_rows() {
let (mut store, repo) = fixture();
observe(&mut store, &repo, "a.md", "[b](b.md)\n", T0);
observe(&mut store, &repo, "b.md", "# B\n\n[a](a.md)\n", T1);
assert_eq!(
texts(&store, "SELECT dst_node FROM edges ORDER BY rowid"),
["d_1", "d_0"]
);
store.observe_delete(&repo, "b.md", T2).unwrap();
assert_eq!(
texts(
&store,
"SELECT dst_node FROM edges WHERE to_commit IS NULL ORDER BY rowid"
),
["d_1", "d_0"],
"edges stay open and keep the tombstoned doc's id"
);
let counts: Vec<i64> = ["nodes", "doc_edges"]
.iter()
.map(|t| {
store
.conn()
.query_row(&format!("SELECT count(*) FROM {t} WHERE 1"), [], |r| {
r.get(0)
})
.unwrap()
})
.collect();
assert_eq!(
counts,
[3, 2],
"a's link; b's section and link; both rollups"
);
observe(&mut store, &repo, "c.md", "[b](b.md)\n", T2);
assert_eq!(
texts(&store, "SELECT dst_node FROM edges WHERE src_doc = 'd_2'"),
["phantom:b.md"]
);
observe(
&mut store,
&repo,
"b.md",
"# B2\n",
"2026-09-26T10:03:00.000Z",
);
assert_eq!(
texts(
&store,
"SELECT dst_node FROM edges WHERE to_commit IS NULL AND dst_node LIKE 'd_%' AND src_doc != 'd_1' ORDER BY src_doc"
),
["d_1", "d_1"]
);
assert_eq!(
texts(
&store,
"SELECT dst_node FROM doc_edges WHERE src_doc = 'd_2'"
),
["d_1"],
"the affected rollup is recomputed"
);
}
#[test]
fn mint_order_is_x_during_resolution_then_e() {
let (mut store, repo) = fixture();
observe(
&mut store,
&repo,
"a.md",
"<https://b.com> then <https://a.com>\n",
T0,
);
let ext = rows(
&store,
"SELECT node_id, uri FROM external_nodes ORDER BY node_id",
);
assert_eq!(
ext[0],
vec![Sql::Text("x_0".into()), Sql::Text("https://b.com".into())]
);
assert_eq!(
ext[1],
vec![Sql::Text("x_1".into()), Sql::Text("https://a.com".into())]
);
assert_eq!(
texts(&store, "SELECT edge_id FROM edges ORDER BY rowid"),
["e_0", "e_1"]
);
}
}