tftio-kb 2.5.3

//! SQLite-backed persistence for kb.
//!
//! The schema mirrors the Haskell `KB.Storage` bootstrap. Connection
//! pragmas run on every connection; bootstrap DDL initializes a fresh
//! database. AST blobs use s-expression encoding via [`crate::sexp`]
//! matching the Haskell on-disk format.
//!
//! ## Design decisions
//!
//! [AST encoding] `nodes.ast_blob` stores the AST as an s-expression via
//! `crate::sexp`. Versioned with a `(kb-doc 1 …)` wrapper.
//!
//! [Tags] Tags are normalized into `node_tags` (one row per (node, tag)).
//!
//! [Links] One physical `links` table holds both id-links and
//! name-links, discriminated by `link_type` ('id' | 'name'). `source_id`
//! is a FOREIGN KEY into `nodes(id)` ON DELETE CASCADE. `target_id` is
//! a FOREIGN KEY into `nodes(id)` ON DELETE SET NULL — name-link rows
//! demote to broken (target_id NULL) when their target is removed; the
//! `delete_node` codepath manually removes id-link rows pointing at the
//! deleted node before the cascade so the CHECK constraint (id-link
//! rows require NOT NULL target_id) is never violated. A schema-level
//! CHECK enforces the per-link_type column contract:
//!   id  -> target_id NOT NULL, target_slug NULL
//!   name -> target_slug NOT NULL (target_id may be NULL = broken)
//!
//! [Audit log] `audit_log.node_id` is plain `TEXT` with no foreign key,
//! so a deleted node's history survives. Every mutation
//! (`insert_node`/`update_node`/`delete_node`) writes one audit row in
//! the same transaction as the data change.
//!
//! [Schema version] Tracked via `PRAGMA user_version`. Bootstrap stamps
//! it to [`CURRENT_SCHEMA_VERSION`].
//!
//! [Title/tag derivation] `insert_node` / `update_node` take a `Document`
//! and derive the row's title and tags from its content. Title comes from
//! the first heading, or first paragraph text truncated to 80 chars,
//! or "(untitled)".

use crate::error::KbError;
use std::cmp::Ordering;
use std::collections::{BTreeMap, HashSet};

use rusqlite::{Connection, OptionalExtension, params};

use crate::ast::*;
use crate::embedding;
use crate::sexp;

// ── Schema ──────────────────────────────────────────────────────────────

/// Schema version stamped into `PRAGMA user_version` at bootstrap.
///
/// v1 mirrors Haskell `KB.Storage.currentSchemaVersion`. v2 re-encodes
/// every `ast_blob` to the current sexp format (ordered lists carry a
/// start number; legacy blobs used a bare `ordered` symbol).
pub const CURRENT_SCHEMA_VERSION: u32 = 2;

const NODES_TABLE: &str = concat!(
    "CREATE TABLE IF NOT EXISTS nodes (",
    "  id         TEXT PRIMARY KEY NOT NULL,",
    "  title      TEXT NOT NULL,",
    "  ast_blob   TEXT NOT NULL,",
    "  created_at TEXT NOT NULL,",
    "  updated_at TEXT NOT NULL",
    ")"
);

const NODE_TAGS_TABLE: &str = concat!(
    "CREATE TABLE IF NOT EXISTS node_tags (",
    "  node_id TEXT NOT NULL,",
    "  tag     TEXT NOT NULL,",
    "  PRIMARY KEY (node_id, tag),",
    "  FOREIGN KEY (node_id) REFERENCES nodes(id) ON DELETE CASCADE",
    ")"
);

// Unified link graph. One physical table; `link_type` discriminates.
//
// Column contract (enforced by CHECK):
//   link_type='id'   -> target_id NOT NULL, target_slug NULL
//   link_type='name' -> target_slug NOT NULL (target_id may be NULL)
//
// `source_id` FK ON DELETE CASCADE: a node's outgoing rows go with it.
// `target_id` FK ON DELETE SET NULL: name-link rows demote to broken
// (target_id NULL, target_slug intact) when the target node is removed.
// `delete_node` removes id-link rows pointing at the deleted node
// explicitly before the cascade so the CHECK is never violated.
//
// The PK column order puts `link_type` second so a per-source lookup
// (the relinker's hot path) is index-served, and so id vs name rows
// for the same source are stored contiguously.
const LINKS_TABLE: &str = concat!(
    "CREATE TABLE IF NOT EXISTS links (",
    "  source_id   TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,",
    "  link_type   TEXT NOT NULL,",
    "  target_id   TEXT REFERENCES nodes(id) ON DELETE SET NULL,",
    "  target_slug TEXT,",
    "  PRIMARY KEY (source_id, link_type, target_id, target_slug),",
    "  CHECK (",
    "    (link_type = 'id'   AND target_id IS NOT NULL AND target_slug IS NULL)",
    "    OR (link_type = 'name' AND target_slug IS NOT NULL)",
    "  )",
    ")"
);

const AUDIT_LOG_TABLE: &str = concat!(
    "CREATE TABLE IF NOT EXISTS audit_log (",
    "  id        INTEGER PRIMARY KEY AUTOINCREMENT,",
    "  node_id   TEXT NOT NULL,",
    "  operation TEXT NOT NULL,",
    "  old_blob  TEXT,",
    "  new_blob  TEXT,",
    "  timestamp TEXT NOT NULL",
    ")"
);

const NODES_UPDATED_AT_INDEX: &str =
    "CREATE INDEX IF NOT EXISTS nodes_updated_at_idx ON nodes (updated_at DESC)";

const NODE_TAGS_TAG_INDEX: &str = "CREATE INDEX IF NOT EXISTS node_tags_tag_idx ON node_tags (tag)";

const LINKS_TARGET_INDEX: &str = "CREATE INDEX IF NOT EXISTS links_target_idx ON links (target_id)";

const LINKS_TARGET_SLUG_INDEX: &str =
    "CREATE INDEX IF NOT EXISTS links_target_slug_idx ON links (target_slug)";

const AUDIT_LOG_NODE_INDEX: &str =
    "CREATE INDEX IF NOT EXISTS audit_log_node_idx ON audit_log (node_id)";

const AUDIT_LOG_TIMESTAMP_INDEX: &str =
    "CREATE INDEX IF NOT EXISTS audit_log_ts_idx ON audit_log (timestamp)";

const NODES_FTS_TABLE: &str = concat!(
    "CREATE VIRTUAL TABLE IF NOT EXISTS nodes_fts USING fts5(",
    "  node_id UNINDEXED,",
    "  title,",
    "  body",
    ")"
);

const EMBEDDINGS_TABLE: &str = concat!(
    "CREATE TABLE IF NOT EXISTS embeddings (",
    "  node_id   TEXT NOT NULL,",
    "  model     TEXT NOT NULL,",
    "  embedding BLOB NOT NULL,",
    "  PRIMARY KEY (node_id, model),",
    "  FOREIGN KEY (node_id) REFERENCES nodes(id) ON DELETE CASCADE",
    ")"
);

const EMBEDDINGS_MODEL_INDEX: &str =
    "CREATE INDEX IF NOT EXISTS embeddings_model_idx ON embeddings (model)";

/// Open a database connection with pragmas applied.
///
/// # Errors
///
/// Returns `rusqlite::Error` on connection failure.
pub fn open_db(path: &str) -> Result<Connection, KbError> {
    // Best-effort: ensure the parent directory exists so the default
    // `$HOME/.local/share/kb/` location works on first run. A genuine
    // failure surfaces from `Connection::open` below.
    if let Some(parent) = std::path::Path::new(path).parent() {
        if !parent.as_os_str().is_empty() {
            let _ = std::fs::create_dir_all(parent);
        }
    }
    let conn = Connection::open(path)?;
    conn.execute_batch("PRAGMA foreign_keys = ON;")?;
    Ok(conn)
}

/// Default kb database path: `$HOME/.local/share/kb/kb.db`.
///
/// Falls back to `kb.db` in the current directory when `$HOME` is unset.
#[must_use]
pub fn default_db_path() -> std::path::PathBuf {
    dirs::home_dir().map_or_else(
        || std::path::PathBuf::from("kb.db"),
        |home| home.join(".local/share/kb/kb.db"),
    )
}

/// Bootstrap the database schema. Idempotent (uses IF NOT EXISTS).
///
/// Creates every table / index in the schema, runs version-gated data
/// migrations for databases below [`CURRENT_SCHEMA_VERSION`], and stamps
/// `PRAGMA user_version = CURRENT_SCHEMA_VERSION`.
///
/// # Errors
///
/// Returns [`KbError::Database`] if schema creation fails, or
/// [`KbError::CorruptAstBlob`] if a stored blob fails to decode during
/// the v2 re-encode migration.
pub fn init_db(conn: &Connection) -> Result<(), KbError> {
    conn.execute_batch("PRAGMA journal_mode = WAL;")?;
    let prior_version: u32 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
    conn.execute(NODES_TABLE, [])?;
    conn.execute(NODE_TAGS_TABLE, [])?;
    migrate_links_table(conn)?;
    conn.execute(AUDIT_LOG_TABLE, [])?;
    conn.execute(NODES_UPDATED_AT_INDEX, [])?;
    conn.execute(NODE_TAGS_TAG_INDEX, [])?;
    conn.execute(LINKS_TARGET_INDEX, [])?;
    conn.execute(LINKS_TARGET_SLUG_INDEX, [])?;
    conn.execute(AUDIT_LOG_NODE_INDEX, [])?;
    conn.execute(AUDIT_LOG_TIMESTAMP_INDEX, [])?;
    conn.execute_batch(NODES_FTS_TABLE)?;
    conn.execute(EMBEDDINGS_TABLE, [])?;
    conn.execute(EMBEDDINGS_MODEL_INDEX, [])?;
    if prior_version < 2 {
        migrate_ast_blob_encoding(conn)?;
    }
    conn.execute_batch(&format!("PRAGMA user_version = {CURRENT_SCHEMA_VERSION};"))?;
    Ok(())
}

/// Re-encode every `ast_blob` to the current sexp format (schema v2).
///
/// v1 blobs encoded ordered lists as a bare `ordered` symbol; the current
/// format is `(ordered N)`. The decoder accepts both, so this pass is
/// decode → encode → write-back-if-changed. Node timestamps, the audit
/// log, and the FTS index are untouched: the document content is
/// identical, only its serialization changes.
fn migrate_ast_blob_encoding(conn: &Connection) -> Result<(), KbError> {
    let tx = conn.unchecked_transaction()?;
    let rows: Vec<(String, String)> = {
        let mut stmt = tx.prepare("SELECT id, ast_blob FROM nodes")?;
        let mapped =
            stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
        mapped.collect::<Result<_, _>>()?
    };
    {
        let mut update = tx.prepare("UPDATE nodes SET ast_blob = ?1 WHERE id = ?2")?;
        for (id, blob) in &rows {
            let doc = decode_blob(id, blob)?;
            let encoded = sexp::encode_document(&doc);
            if encoded != *blob {
                update.execute(params![encoded, id])?;
            }
        }
    }
    tx.commit()?;
    Ok(())
}

/// Create the unified `links` table, migrating from the pre-consolidation
/// split shape if it is present.
///
/// Pre-consolidation databases (kb-link-index-v0) carried two physical
/// tables:
///   - `links(source_id, target_id, link_type='id')` — UUID id-links, FK
///     CASCADE on both columns
///   - `name_links(src_id, dst_slug, dst_id NULLABLE)` — slug name-links
///
/// The migration is idempotent and runs in one transaction:
///   1. Detect a `name_links` table or a pre-consolidation `links` shape
///      (no `target_slug` column).
///   2. If found, snapshot the rows, drop both legacy tables, recreate
///      the unified `links` table, and re-insert every row preserving
///      `target_id` (including NULL for broken name-links) and
///      `target_slug`. Id-link rows write `target_slug = NULL`.
///   3. If absent, just `CREATE TABLE IF NOT EXISTS links` — the unified
///      shape — so a fresh database lands at the new schema directly.
fn migrate_links_table(conn: &Connection) -> Result<(), rusqlite::Error> {
    let links_exists: bool = table_exists(conn, "links")?;
    let name_links_exists: bool = table_exists(conn, "name_links")?;
    let links_is_legacy: bool = links_exists && !column_exists(conn, "links", "target_slug")?;

    if !links_is_legacy && !name_links_exists {
        // Fresh DB or already at the unified shape.
        conn.execute(LINKS_TABLE, [])?;
        return Ok(());
    }

    // Snapshot rows from whichever legacy tables are present.
    let mut legacy_id_links: Vec<(String, String)> = Vec::new();
    if links_is_legacy {
        let mut stmt = conn.prepare("SELECT source_id, target_id FROM links")?;
        let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
        for row in rows {
            legacy_id_links.push(row?);
        }
    }
    let mut legacy_name_links: Vec<(String, String, Option<String>)> = Vec::new();
    if name_links_exists {
        let mut stmt = conn.prepare("SELECT src_id, dst_slug, dst_id FROM name_links")?;
        let rows = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>(0)?,
                r.get::<_, String>(1)?,
                r.get::<_, Option<String>>(2)?,
            ))
        })?;
        for row in rows {
            legacy_name_links.push(row?);
        }
    }

    let tx = conn.unchecked_transaction()?;
    if links_is_legacy {
        tx.execute("DROP TABLE links", [])?;
    }
    if name_links_exists {
        tx.execute("DROP TABLE name_links", [])?;
    }
    tx.execute(LINKS_TABLE, [])?;
    {
        let mut ins_id = tx.prepare(
            "INSERT OR IGNORE INTO links (source_id, link_type, target_id, target_slug) \
             VALUES (?1, 'id', ?2, NULL)",
        )?;
        for (src, tgt) in &legacy_id_links {
            ins_id.execute(params![src, tgt])?;
        }
        let mut ins_name = tx.prepare(
            "INSERT OR IGNORE INTO links (source_id, link_type, target_id, target_slug) \
             VALUES (?1, 'name', ?2, ?3)",
        )?;
        for (src, slug, dst) in &legacy_name_links {
            ins_name.execute(params![src, dst, slug])?;
        }
    }
    tx.commit()?;
    Ok(())
}

fn table_exists(conn: &Connection, name: &str) -> Result<bool, rusqlite::Error> {
    let n: i64 = conn.query_row(
        "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?1",
        params![name],
        |r| r.get(0),
    )?;
    Ok(n > 0)
}

fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool, rusqlite::Error> {
    let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
    let rows = stmt.query_map([], |r| r.get::<_, String>(1))?;
    for row in rows {
        if row? == column {
            return Ok(true);
        }
    }
    Ok(false)
}

// ── Row types ───────────────────────────────────────────────────────────

/// A row from the `nodes` table.
#[derive(Debug, Clone)]
pub struct NodeRow {
    pub id: NodeId,
    pub title: Title,
    pub ast_blob: String,
    pub created_at: String,
    pub updated_at: String,
}

/// A row from the unified `links` table.
///
/// `target_id` is `Some` for every resolved row (id-links by
/// construction, name-links once their slug is matched);
/// `target_slug` is `Some` for every name-link row and `None` for
/// id-link rows.
/// Discriminator for a row in the unified `links` table: an id-link
/// (UUID `target_id`) or a name-link (bracketed `target_slug`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum LinkType {
    /// A link to a node by its UUID (`target_id`).
    Id,
    /// A link to a node by bracketed slug (`target_slug`).
    Name,
}

impl LinkType {
    /// The on-disk / wire token for this link type (`"id"` or `"name"`).
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Id => "id",
            Self::Name => "name",
        }
    }
}

impl std::fmt::Display for LinkType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

impl rusqlite::types::FromSql for LinkType {
    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
        match value.as_str()? {
            "id" => Ok(Self::Id),
            "name" => Ok(Self::Name),
            // The schema CHECK restricts this column to 'id' | 'name', so any
            // other value can only come from external corruption.
            _ => Err(rusqlite::types::FromSqlError::InvalidType),
        }
    }
}

impl rusqlite::types::FromSql for NodeId {
    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
        value.as_str().map(|s| Self(s.to_owned()))
    }
}

impl rusqlite::types::FromSql for Title {
    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
        value.as_str().map(|s| Self(s.to_owned()))
    }
}

impl rusqlite::types::FromSql for Tag {
    fn column_result(value: rusqlite::types::ValueRef<'_>) -> rusqlite::types::FromSqlResult<Self> {
        value.as_str().map(|s| Self(s.to_owned()))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LinkRow {
    pub source_id: String,
    pub link_type: LinkType,
    pub target_id: Option<String>,
    pub target_slug: Option<String>,
}

/// A row from the `audit_log` table.
#[derive(Debug, Clone)]
pub struct AuditRow {
    pub id: i64,
    pub node_id: String,
    pub operation: String,
    pub old_blob: Option<String>,
    pub new_blob: Option<String>,
    pub timestamp: String,
}

/// Graph neighborhood of a node: outgoing and incoming edges as
/// `(other-node-id, link_type)` pairs. Order is unspecified.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Neighborhood {
    pub outgoing: Vec<(NodeId, LinkType)>,
    pub incoming: Vec<(NodeId, LinkType)>,
}

fn now_iso8601() -> String {
    chrono::Utc::now()
        .format("%Y-%m-%dT%H:%M:%S%.3fZ")
        .to_string()
}

// ── Embedding write helper ──────────────────────────────────────────────

/// Write a precomputed embedding row keyed by `(node_id, model)`. No-op
/// when `embedding` or `model` is `None`. Logs and continues on a SQL
/// failure — storage layer is otherwise ignorant of embedding HTTP.
fn write_embedding_row(
    conn: &Connection,
    node_id: &str,
    embedding: Option<Vec<f32>>,
    model: Option<&str>,
) {
    let (Some(vec), Some(model)) = (embedding, model) else {
        return;
    };
    let blob = embedding::encode_embedding(&vec);
    if let Err(e) = conn.execute(
        "INSERT OR REPLACE INTO embeddings (node_id, model, embedding) VALUES (?1, ?2, ?3)",
        params![node_id, model, blob],
    ) {
        tracing::error!(node_id, error = %e, "kb embed write failed");
    }
}

// ── Operations ──────────────────────────────────────────────────────────

/// Insert a new node. Title and tags are derived from the AST. The full
/// insert (nodes row, node_tags rows, FTS5 row, audit_log row, outgoing
/// links) is wrapped in a single transaction.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn insert_node(
    conn: &Connection,
    node_id: &str,
    document: &Document,
) -> Result<(), rusqlite::Error> {
    insert_node_with(conn, node_id, document, None, None)
}

/// [`insert_node`] with a precomputed embedding.
///
/// The caller (axum or MCP handler) computes the embedding via the
/// async [`crate::embedding::EmbeddingClient`] before invoking storage.
/// The node write commits first; the precomputed embedding row, if any,
/// is then written in a separate statement so a SQL failure on the
/// embeddings table does not roll back the data write.
///
/// When `embedding` is `Some(v)` and `model` is `Some(m)`, one row is
/// upserted into the `embeddings` table keyed by `(node_id, m)`.
/// Otherwise no embedding row is written.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure of the node write.
/// Embedding-write failures are logged and ignored.
pub fn insert_node_with(
    conn: &Connection,
    node_id: &str,
    document: &Document,
    embedding: Option<Vec<f32>>,
    model: Option<&str>,
) -> Result<(), rusqlite::Error> {
    let blob = sexp::encode_document(document);
    let title = extract_title(document);
    let tags = extract_tags(document);
    let body_text = extract_body_text(document);
    let now = now_iso8601();

    let tx = conn.unchecked_transaction()?;

    tx.execute(
        "INSERT INTO nodes (id, title, ast_blob, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5)",
        params![node_id, title, blob, now, now],
    )?;
    for tag in &tags {
        tx.execute(
            "INSERT INTO node_tags (node_id, tag) VALUES (?1, ?2)",
            params![node_id, tag.0.as_str()],
        )?;
    }
    tx.execute(
        "INSERT INTO nodes_fts (node_id, title, body) VALUES (?1, ?2, ?3)",
        params![node_id, title, body_text],
    )?;
    tx.execute(
        "INSERT INTO audit_log (node_id, operation, old_blob, new_blob, timestamp) VALUES (?1, 'insert', NULL, ?2, ?3)",
        params![node_id, blob, now],
    )?;

    relink_one_inner(&tx, node_id, document)?;
    // If this new node carries a `#+name:` slug, back-resolve any
    // previously-broken bracket references pointing at it.
    if let Some(slug) = extract_name_slug(document) {
        tx.execute(
            "UPDATE links SET target_id = ?1 \
             WHERE link_type = 'name' AND target_slug = ?2 AND target_id IS NULL",
            params![node_id, slug],
        )?;
    }

    tx.commit()?;

    write_embedding_row(conn, node_id, embedding, model);

    Ok(())
}

/// Get a node by ID. Decodes the sexp blob into a [`Document`].
///
/// # Errors
///
/// Returns [`KbError::Database`] on database failure, or
/// [`KbError::CorruptAstBlob`] when the stored blob fails to decode.
pub fn get_node(conn: &Connection, id: &str) -> Result<Option<Document>, KbError> {
    let mut stmt = conn.prepare("SELECT ast_blob FROM nodes WHERE id = ?1")?;
    let mut rows = stmt.query_map(params![id], |row| row.get::<_, String>(0))?;
    match rows.next() {
        Some(Ok(blob)) => {
            let doc = decode_blob(id, &blob)?;
            Ok(Some(doc))
        }
        Some(Err(e)) => Err(KbError::Database(e)),
        None => Ok(None),
    }
}

/// Decode a stored `ast_blob`, attributing failure to the owning node.
fn decode_blob(node_id: &str, blob: &str) -> Result<Document, KbError> {
    sexp::decode_document(blob).map_err(|e| KbError::CorruptAstBlob {
        node_id: node_id.to_string(),
        reason: e.to_string(),
    })
}

/// Get a raw node row (without decoding the blob).
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn get_node_row(conn: &Connection, id: &str) -> Result<Option<NodeRow>, KbError> {
    let mut stmt = conn
        .prepare("SELECT id, title, ast_blob, created_at, updated_at FROM nodes WHERE id = ?1")?;
    let mut rows = stmt.query_map(params![id], |row| {
        Ok(NodeRow {
            id: row.get(0)?,
            title: row.get(1)?,
            ast_blob: row.get(2)?,
            created_at: row.get(3)?,
            updated_at: row.get(4)?,
        })
    })?;
    match rows.next() {
        Some(Ok(row)) => Ok(Some(row)),
        Some(Err(e)) => Err(KbError::Database(e)),
        None => Ok(None),
    }
}

/// Update an existing node. Returns `false` if the id is unknown. The
/// full update (row replacement, node_tags refresh, FTS5 refresh, audit
/// row, outgoing links) is wrapped in a single transaction.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn update_node(
    conn: &Connection,
    id: &str,
    document: &Document,
) -> Result<bool, rusqlite::Error> {
    update_node_with(conn, id, document, None, None)
}

/// [`update_node`] with a precomputed embedding.
///
/// The caller computes the embedding via the async
/// [`crate::embedding::EmbeddingClient`] before invoking storage. The
/// node write commits first; the precomputed embedding row, if any, is
/// then upserted in a separate statement so a SQL failure on the
/// embeddings table does not roll back the data write.
///
/// When `embedding` is `Some(v)` and `model` is `Some(m)`, one row is
/// upserted into the `embeddings` table keyed by `(id, m)`. Otherwise no
/// embedding row is written.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure of the node write.
/// Embedding-write failures are logged and ignored.
pub fn update_node_with(
    conn: &Connection,
    id: &str,
    document: &Document,
    embedding: Option<Vec<f32>>,
    model: Option<&str>,
) -> Result<bool, rusqlite::Error> {
    let tx = conn.unchecked_transaction()?;

    let prior_blob: Option<String> = tx
        .query_row(
            "SELECT ast_blob FROM nodes WHERE id = ?1",
            params![id],
            |row| row.get(0),
        )
        .optional()?;
    let Some(prior_blob) = prior_blob else {
        return Ok(false);
    };

    let blob = sexp::encode_document(document);
    let title = extract_title(document);
    let tags = extract_tags(document);
    let body_text = extract_body_text(document);
    let now = now_iso8601();

    tx.execute(
        "UPDATE nodes SET title = ?1, ast_blob = ?2, updated_at = ?3 WHERE id = ?4",
        params![title, blob, now, id],
    )?;
    tx.execute("DELETE FROM node_tags WHERE node_id = ?1", params![id])?;
    for tag in &tags {
        tx.execute(
            "INSERT INTO node_tags (node_id, tag) VALUES (?1, ?2)",
            params![id, tag.0.as_str()],
        )?;
    }
    tx.execute("DELETE FROM nodes_fts WHERE node_id = ?1", params![id])?;
    tx.execute(
        "INSERT INTO nodes_fts (node_id, title, body) VALUES (?1, ?2, ?3)",
        params![id, title, body_text],
    )?;
    tx.execute(
        "INSERT INTO audit_log (node_id, operation, old_blob, new_blob, timestamp) VALUES (?1, 'update', ?2, ?3, ?4)",
        params![id, prior_blob, blob, now],
    )?;

    // Decode prior blob to see whether the node previously carried a
    // `#+name:` slug. If the slug changes (or disappears), invalidate
    // any name-link rows that were resolved at this node so they go
    // back to broken.
    let prior_slug = sexp::decode_document(&prior_blob)
        .ok()
        .as_ref()
        .and_then(extract_name_slug);
    let new_slug = extract_name_slug(document);
    if prior_slug.as_deref() != new_slug.as_deref() {
        invalidate_resolved_pointing_at(&tx, id)?;
    }
    relink_one_inner(&tx, id, document)?;
    // Back-resolve broken bracket references against this node's
    // (possibly new) `#+name:` slug.
    if let Some(slug) = new_slug {
        tx.execute(
            "UPDATE links SET target_id = ?1 \
             WHERE link_type = 'name' AND target_slug = ?2 AND target_id IS NULL",
            params![id, slug],
        )?;
    }

    tx.commit()?;

    write_embedding_row(conn, id, embedding, model);

    Ok(true)
}

/// Hard-delete a node. Cascades to `node_tags` and `links`. Audit row is
/// written **before** the delete so `old_blob` captures the prior AST.
/// Returns `false` if the id is unknown.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn delete_node(conn: &Connection, id: &str) -> Result<bool, rusqlite::Error> {
    let tx = conn.unchecked_transaction()?;

    let prior_blob: Option<String> = tx
        .query_row(
            "SELECT ast_blob FROM nodes WHERE id = ?1",
            params![id],
            |row| row.get(0),
        )
        .optional()?;
    let Some(prior_blob) = prior_blob else {
        return Ok(false);
    };

    tx.execute(
        "INSERT INTO audit_log (node_id, operation, old_blob, new_blob, timestamp) VALUES (?1, 'delete', ?2, NULL, ?3)",
        params![id, prior_blob, now_iso8601()],
    )?;

    tx.execute("DELETE FROM nodes_fts WHERE node_id = ?1", params![id])?;
    // Id-link rows pointing at this node must be removed before the
    // node delete fires the FK ON DELETE SET NULL on `target_id` —
    // otherwise the CHECK constraint (link_type='id' requires NOT NULL
    // target_id) would fail. Name-link rows demote naturally to broken
    // (target_id NULL, target_slug intact) via the same SET NULL FK.
    tx.execute(
        "DELETE FROM links WHERE link_type = 'id' AND target_id = ?1",
        params![id],
    )?;
    tx.execute("DELETE FROM nodes WHERE id = ?1", params![id])?;

    tx.commit()?;
    Ok(true)
}

/// List nodes by tag using the node_tags join.
///
/// The query tag is normalized via [`normalize_tag`] so callers need not
/// match the stored casing or separators.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn list_by_tag(conn: &Connection, tag: &str) -> Result<Vec<NodeRow>, rusqlite::Error> {
    let tag = normalize_tag(tag);
    let mut stmt = conn.prepare(
        "SELECT n.id, n.title, n.ast_blob, n.created_at, n.updated_at \
         FROM nodes n JOIN node_tags nt ON n.id = nt.node_id \
         WHERE nt.tag = ?1 ORDER BY n.updated_at DESC",
    )?;
    let rows = stmt.query_map(params![tag], |row| {
        Ok(NodeRow {
            id: row.get(0)?,
            title: row.get(1)?,
            ast_blob: row.get(2)?,
            created_at: row.get(3)?,
            updated_at: row.get(4)?,
        })
    })?;
    rows.collect()
}

/// List recent nodes, newest first.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn list_recent(conn: &Connection, limit: usize) -> Result<Vec<NodeRow>, rusqlite::Error> {
    let mut stmt = conn.prepare(
        "SELECT id, title, ast_blob, created_at, updated_at FROM nodes ORDER BY updated_at DESC LIMIT ?1",
    )?;
    let rows = stmt.query_map(params![limit as i64], |row| {
        Ok(NodeRow {
            id: row.get(0)?,
            title: row.get(1)?,
            ast_blob: row.get(2)?,
            created_at: row.get(3)?,
            updated_at: row.get(4)?,
        })
    })?;
    rows.collect()
}

/// Paginated full enumeration. Returns `(NodeId, Title)` pairs ordered
/// by `updated_at` DESC. `limit = 0` returns an empty vec.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn list_all_nodes(
    conn: &Connection,
    limit: usize,
    offset: usize,
) -> Result<Vec<(NodeId, Title)>, rusqlite::Error> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let mut stmt =
        conn.prepare("SELECT id, title FROM nodes ORDER BY updated_at DESC LIMIT ?1 OFFSET ?2")?;
    let rows = stmt.query_map(params![limit as i64, offset as i64], |row| {
        Ok((NodeId(row.get(0)?), Title(row.get(1)?)))
    })?;
    rows.collect()
}

/// Full node data: row fields plus the decoded document and tags.
#[derive(Debug, Clone)]
pub struct NodeFullData {
    pub title: String,
    pub tags: Vec<Tag>,
    pub document: Document,
    pub created_at: String,
    pub updated_at: String,
}

/// Get full node data by ID, joining node_tags and decoding the sexp blob.
///
/// # Errors
///
/// Returns [`KbError::Database`] on database failure, or
/// [`KbError::CorruptAstBlob`] when the stored blob fails to decode.
pub fn get_node_full(conn: &Connection, id: &str) -> Result<Option<NodeFullData>, KbError> {
    let mut stmt =
        conn.prepare("SELECT title, ast_blob, created_at, updated_at FROM nodes WHERE id = ?1")?;
    let mut rows = stmt.query_map(params![id], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, String>(3)?,
        ))
    })?;
    match rows.next() {
        Some(Ok((title, blob, created_at, updated_at))) => {
            let doc = decode_blob(id, &blob)?;
            let tags = get_node_tags(conn, id)?;
            Ok(Some(NodeFullData {
                title,
                tags,
                document: doc,
                created_at,
                updated_at,
            }))
        }
        Some(Err(e)) => Err(KbError::Database(e)),
        None => Ok(None),
    }
}

/// Get tags for a node from the node_tags table.
fn get_node_tags(conn: &Connection, node_id: &str) -> Result<Vec<Tag>, rusqlite::Error> {
    let mut stmt = conn.prepare("SELECT tag FROM node_tags WHERE node_id = ?1")?;
    let rows = stmt.query_map(params![node_id], |row| row.get::<_, String>(0))?;
    rows.map(|r| r.map(Tag)).collect()
}

/// Batch title lookup for a list of node IDs. Returns (id, title) pairs
/// in the same order as the input.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn fetch_titles(
    conn: &Connection,
    ids: &[String],
) -> Result<Vec<(String, String)>, rusqlite::Error> {
    if ids.is_empty() {
        return Ok(Vec::new());
    }
    let placeholders: Vec<String> = ids
        .iter()
        .enumerate()
        .map(|(i, _)| format!("?{}", i + 1))
        .collect();
    let sql = format!(
        "SELECT id, title FROM nodes WHERE id IN ({})",
        placeholders.join(",")
    );
    let mut stmt = conn.prepare(&sql)?;
    let bound: Vec<&dyn rusqlite::types::ToSql> = ids
        .iter()
        .map(|id| id as &dyn rusqlite::types::ToSql)
        .collect();
    let rows = stmt.query_map(bound.as_slice(), |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
    })?;
    let mut map: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    for row in rows {
        let (id, title) = row?;
        map.insert(id, title);
    }
    Ok(ids
        .iter()
        .map(|id| (id.clone(), map.get(id).cloned().unwrap_or_default()))
        .collect())
}

/// Full-text search via FTS5. Returns node IDs ranked by relevance.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn search_fts(conn: &Connection, query: &str) -> Result<Vec<String>, rusqlite::Error> {
    let q = query.trim();
    if q.is_empty() {
        return Ok(Vec::new());
    }
    let terms: Vec<String> = q.split_whitespace().map(|t| format!("\"{t}\"*")).collect();
    let fts_query = terms.join(" AND ");

    let mut stmt =
        conn.prepare("SELECT node_id FROM nodes_fts WHERE nodes_fts MATCH ?1 ORDER BY rank")?;
    let rows = stmt.query_map(params![fts_query], |row| row.get::<_, String>(0))?;
    let mut results = Vec::new();
    for row in rows {
        let id = row?;
        if !results.contains(&id) {
            results.push(id);
        }
    }
    Ok(results)
}

// ── Hybrid search ──────────────────────────────────────────────────────

/// Cosine similarity between two same-length vectors. Returns `0.0` if
/// the lengths differ or either vector has zero norm. Mirrors the
/// Haskell `cosineSimilarity`.
#[must_use]
pub fn cosine_similarity(a: &[f32], b: &[f32]) -> f32 {
    if a.len() != b.len() {
        return 0.0;
    }
    let na = a.iter().map(|x| x * x).sum::<f32>().sqrt();
    let nb = b.iter().map(|x| x * x).sum::<f32>().sqrt();
    if na == 0.0 || nb == 0.0 {
        return 0.0;
    }
    let dot: f32 = a.iter().zip(b.iter()).map(|(x, y)| x * y).sum();
    dot / (na * nb)
}

/// Reciprocal Rank Fusion. Each input list ranks an item by 1-indexed
/// position; a node's score is `sum(1 / (k + rank))` across all lists
/// in which it appears. Returns nodes sorted by descending score.
///
/// Tie-breaking is deterministic: equal scores are returned in
/// ascending lexicographic order of the underlying node id.
#[must_use]
pub fn reciprocal_rank_fusion(k: usize, lists: &[Vec<NodeId>]) -> Vec<NodeId> {
    let mut scores: BTreeMap<String, f64> = BTreeMap::new();
    for list in lists {
        for (rank0, NodeId(id)) in list.iter().enumerate() {
            let rank = rank0 + 1;
            #[allow(
                clippy::cast_precision_loss,
                reason = "(k + rank) is a small positive rank index; the f64 cast for RRF scoring loses no significant precision"
            )]
            let contribution = 1.0_f64 / (k + rank) as f64;
            *scores.entry(id.clone()).or_insert(0.0) += contribution;
        }
    }
    let mut entries: Vec<(String, f64)> = scores.into_iter().collect();
    // BTreeMap iter is ascending by key; stable sort by score descending
    // preserves ascending key order on ties.
    entries.sort_by(|a, b| {
        b.1.partial_cmp(&a.1)
            .unwrap_or(Ordering::Equal)
            .then_with(|| a.0.cmp(&b.0))
    });
    entries.into_iter().map(|(s, _)| NodeId(s)).collect()
}

/// Score every stored embedding (for `model`) by cosine similarity to
/// the query vector and return the node ids in descending order.
/// Embeddings whose blob fails to decode are skipped silently.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn rank_by_embedding(
    conn: &Connection,
    query_vec: &[f32],
    model: &str,
) -> Result<Vec<NodeId>, rusqlite::Error> {
    let mut stmt = conn.prepare("SELECT node_id, embedding FROM embeddings WHERE model = ?1")?;
    let rows = stmt.query_map(params![model], |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
    })?;
    let mut scored: Vec<(String, f32)> = Vec::new();
    for row in rows {
        let (nid, blob) = row?;
        if let Ok(ev) = embedding::decode_embedding(&blob) {
            let s = cosine_similarity(query_vec, &ev);
            scored.push((nid, s));
        }
    }
    scored.sort_by(|a, b| {
        b.1.partial_cmp(&a.1)
            .unwrap_or(Ordering::Equal)
            .then_with(|| a.0.cmp(&b.0))
    });
    Ok(scored.into_iter().map(|(id, _)| NodeId(id)).collect())
}

/// Hybrid keyword + vector search ported from Haskell `searchHybrid`.
///
/// Algorithm:
///
/// 1. Run [`search_fts`] for the keyword ranking.
/// 2. If `query_embedding` is `Some` and the query is **not**
///    (whitespace-only AND keyword list empty), rank stored embeddings
///    for the configured model by cosine similarity to the supplied
///    query vector.
/// 3. Fuse the two lists via [`reciprocal_rank_fusion`] with `k = 60`.
///
/// With no query embedding, the result equals the FTS ranking. With
/// both lists empty the result is `[]`.
///
/// The caller (axum or MCP handler) is responsible for computing the
/// query embedding asynchronously via the [`crate::embedding::EmbeddingClient`]
/// trait and passing the result here. Storage no longer talks HTTP.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure of the keyword or
/// vector queries.
pub fn search_hybrid(
    conn: &Connection,
    query: &str,
    query_embedding: Option<(Vec<f32>, &str)>,
) -> Result<Vec<NodeId>, rusqlite::Error> {
    let keyword_ranked: Vec<NodeId> = search_fts(conn, query)?.into_iter().map(NodeId).collect();
    let vector_ranked: Vec<NodeId> = match query_embedding {
        None => Vec::new(),
        Some((qv, model)) => {
            if query.trim().is_empty() && keyword_ranked.is_empty() {
                Vec::new()
            } else {
                rank_by_embedding(conn, &qv, model)?
            }
        }
    };
    Ok(reciprocal_rank_fusion(60, &[keyword_ranked, vector_ranked]))
}

// ── sqlite-vec extension loader ────────────────────────────────────────

/// Why loading the `sqlite-vec` dynamic extension failed.
///
/// Both variants leave the connection usable for normal queries; a
/// caller can branch on whether enabling extension loading failed
/// versus the extension file itself failing to load.
#[derive(Debug, thiserror::Error)]
pub enum VecExtensionError {
    /// Enabling extension loading (the `LoadExtensionGuard`) failed.
    #[error("enable_load_extension failed: {0}")]
    EnableLoad(rusqlite::Error),

    /// `load_extension` failed (file not found, ABI mismatch, &c.).
    #[error("load_extension failed: {0}")]
    LoadExtension(rusqlite::Error),
}

/// Attempt to load the `sqlite-vec` dynamic extension at `path`.
///
/// Mirrors the Haskell `tryLoadVecExtension`: on any failure (file not
/// found, ABI mismatch, extension already loaded, &c.) returns an
/// [`VecExtensionError`] and leaves the connection usable. Success
/// returns `Ok(())`.
///
/// # Errors
///
/// Returns [`VecExtensionError`] on extension-loader failure. The
/// connection remains valid for normal queries either way.
pub fn try_load_vec_extension(conn: &Connection, path: &str) -> Result<(), VecExtensionError> {
    // SAFETY: rusqlite gates extension loading behind `unsafe` because a
    // malicious shared object could violate memory safety. `path` is a
    // trusted, operator-supplied sqlite-vec extension, and the
    // `LoadExtensionGuard` re-disables extension loading on scope exit
    // (including the error paths), so no load-extension capability leaks.
    #[allow(
        unsafe_code,
        reason = "FFI: loads the sqlite-vec dynamic extension; soundness argued in the SAFETY comment above"
    )]
    let result: Result<(), VecExtensionError> = unsafe {
        let _guard =
            rusqlite::LoadExtensionGuard::new(conn).map_err(VecExtensionError::EnableLoad)?;
        conn.load_extension(path, None::<&str>)
            .map_err(VecExtensionError::LoadExtension)?;
        Ok(())
    };
    result
}

// ── Links (unified id + name graph) ────────────────────────────────────

/// A single link reference extracted from a document body.
///
/// The relinker walks the document once and emits one `LinkRef` per
/// outgoing edge — both id-link UUID targets and name-link bracket
/// slugs come out of the same walk so the body is never traversed
/// twice.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LinkRef {
    /// `[[id:UUID]]` — the target is a node id.
    Id(String),
    /// `[[slug]]` — the target is a `#+name:` slug.
    Name(String),
}

/// Walk a [`Document`] once and extract every outgoing link reference,
/// both id-links and name-links, in first-seen order with per-type
/// deduplication. This is the single body walker the relinker uses.
///
/// Walks every nested `Inline` position — paragraphs, headings, table
/// cells, list items, quote blocks. Skips literal regions:
/// `Block::SrcBlock` / `Block::ExampleBlock` and `Inline::InlineCode` /
/// `Inline::Verbatim` never contribute link rows.
///
/// Classification at each link inline:
///   - target starts with `id:` -> `LinkRef::Id` (UUID after the scheme)
///   - bare slug (no `:` and no `/`, description absent) -> `LinkRef::Name`
///   - anything else (https links, paths, links with descriptions other
///     than id-links) -> skipped
#[must_use]
pub fn extract_links(doc: &Document) -> Vec<LinkRef> {
    let mut seen_id = HashSet::<String>::new();
    let mut seen_name = HashSet::<String>::new();
    let mut out = Vec::new();
    collect_links_blocks(&doc.blocks, &mut seen_id, &mut seen_name, &mut out);
    out
}

/// Is `target` a bare bracket-style name slug (not a scheme-qualified
/// URI or a path)?
fn is_bracket_name_ref(target: &str) -> bool {
    !target.is_empty() && !target.contains(':') && !target.contains('/')
}

fn collect_links_blocks(
    blocks: &[Block],
    seen_id: &mut HashSet<String>,
    seen_name: &mut HashSet<String>,
    out: &mut Vec<LinkRef>,
) {
    for block in blocks {
        match block {
            Block::Heading { children, .. } => {
                collect_links_blocks(children, seen_id, seen_name, out);
            }
            Block::Paragraph { inlines } => {
                for inline in inlines {
                    collect_links_inline(inline, seen_id, seen_name, out);
                }
            }
            Block::QuoteBlock { children } => {
                collect_links_blocks(children, seen_id, seen_name, out);
            }
            Block::List { items, .. } => {
                for item in items {
                    collect_links_blocks(&item.content, seen_id, seen_name, out);
                }
            }
            Block::Table { rows } => {
                for row in rows {
                    for cell in row {
                        for inline in &cell.inlines {
                            collect_links_inline(inline, seen_id, seen_name, out);
                        }
                    }
                }
            }
            // Literal text regions (src/example blocks) never contribute.
            Block::SrcBlock { .. } | Block::ExampleBlock { .. } => {}
            _ => {}
        }
    }
}

fn collect_links_inline(
    inline: &Inline,
    seen_id: &mut HashSet<String>,
    seen_name: &mut HashSet<String>,
    out: &mut Vec<LinkRef>,
) {
    match inline {
        Inline::Link {
            target,
            description,
        } => {
            if let Some(rest) = target.strip_prefix("id:") {
                // `[[id:UUID]]` with any description shape.
                let id = rest.to_string();
                if !id.is_empty() && seen_id.insert(id.clone()) {
                    out.push(LinkRef::Id(id));
                }
            } else if description.is_none() && is_bracket_name_ref(target) {
                let slug = target.clone();
                if seen_name.insert(slug.clone()) {
                    out.push(LinkRef::Name(slug));
                }
            }
        }
        Inline::Bold(is) | Inline::Italic(is) | Inline::Strikethrough(is) => {
            for i in is {
                collect_links_inline(i, seen_id, seen_name, out);
            }
        }
        // Literal inline spans never contribute.
        Inline::InlineCode(_) | Inline::Verbatim(_) => {}
        _ => {}
    }
}

/// Read the node's `#+name:` slug, if any. Whitespace-trimmed; empty
/// values yield `None`.
#[must_use]
pub fn extract_name_slug(doc: &Document) -> Option<String> {
    for block in &doc.blocks {
        if let Block::Keyword { name, value } = block {
            if name.eq_ignore_ascii_case("name") {
                let trimmed = value.trim();
                if !trimmed.is_empty() {
                    return Some(trimmed.to_string());
                }
            }
        }
    }
    None
}

/// Replace this node's outgoing links (both id-links and name-links)
/// with whatever its [`Document`] references.
///
/// Forward references for id-links (UUIDs not yet present in `nodes`)
/// are silently dropped — a subsequent [`relink_all`] resolves them.
/// Name-links with no matching `#+name:` slug are stored explicitly
/// with `target_id = NULL` (broken) and can be reverse-resolved when a
/// matching node later appears. Runs in a single transaction.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn relink_one(
    conn: &Connection,
    source_id: &str,
    document: &Document,
) -> Result<(), rusqlite::Error> {
    let tx = conn.unchecked_transaction()?;
    relink_one_inner(&tx, source_id, document)?;
    tx.commit()?;
    Ok(())
}

/// Inner relink implementation. Caller owns the transaction. Used by
/// [`insert_node`] / [`update_node`] (which already have a transaction
/// open) and by the public [`relink_one`] (which opens one).
///
/// One walk of the document body produces both id-link and name-link
/// rows; the unified `links` table is the single sink for both.
fn relink_one_inner(
    conn: &Connection,
    source_id: &str,
    document: &Document,
) -> Result<(), rusqlite::Error> {
    let refs = extract_links(document);
    conn.execute("DELETE FROM links WHERE source_id = ?1", params![source_id])?;
    if refs.is_empty() {
        return Ok(());
    }
    let mut ins_id = conn.prepare(
        "INSERT OR IGNORE INTO links (source_id, link_type, target_id, target_slug) \
         VALUES (?1, 'id', ?2, NULL)",
    )?;
    let mut ins_name = conn.prepare(
        "INSERT OR IGNORE INTO links (source_id, link_type, target_id, target_slug) \
         VALUES (?1, 'name', ?2, ?3)",
    )?;
    for r in &refs {
        match r {
            LinkRef::Id(tgt) => {
                if node_exists(conn, tgt)? {
                    ins_id.execute(params![source_id, tgt])?;
                }
            }
            LinkRef::Name(slug) => {
                let dst_id = resolve_slug_to_id(conn, slug)?;
                ins_name.execute(params![source_id, dst_id, slug])?;
            }
        }
    }
    Ok(())
}

fn node_exists(conn: &Connection, id: &str) -> Result<bool, rusqlite::Error> {
    let n: Option<i64> = conn
        .query_row(
            "SELECT 1 FROM nodes WHERE id = ?1 LIMIT 1",
            params![id],
            |r| r.get(0),
        )
        .optional()?;
    Ok(n.is_some())
}

/// Walk every node and rebuild its outgoing links (both id-links and
/// name-links) under the unified schema.
///
/// Used after a bulk import to resolve forward references that were
/// skipped on first write and to back-resolve broken name-links once
/// their target slug appears. Returns `(nodes_processed, links_written)`.
/// Corrupt blobs increment `nodes_processed` but contribute zero links
/// and never panic.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn relink_all(conn: &Connection) -> Result<(usize, usize), rusqlite::Error> {
    let rows: Vec<(String, String)> = {
        let mut stmt = conn.prepare("SELECT id, ast_blob FROM nodes")?;
        let mapped = stmt.query_map([], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        })?;
        mapped.collect::<Result<_, _>>()?
    };

    let tx = conn.unchecked_transaction()?;
    tx.execute("DELETE FROM links", [])?;

    let mut nodes_processed: usize = 0;
    let mut links_written: usize = 0;
    {
        let mut ins_id = tx.prepare(
            "INSERT OR IGNORE INTO links (source_id, link_type, target_id, target_slug) \
             VALUES (?1, 'id', ?2, NULL)",
        )?;
        let mut ins_name = tx.prepare(
            "INSERT OR IGNORE INTO links (source_id, link_type, target_id, target_slug) \
             VALUES (?1, 'name', ?2, ?3)",
        )?;
        for (nid, blob) in &rows {
            nodes_processed += 1;
            let Ok(doc) = sexp::decode_document(blob) else {
                continue;
            };
            for r in extract_links(&doc) {
                match r {
                    LinkRef::Id(tgt) => {
                        if node_exists(&tx, &tgt)? {
                            ins_id.execute(params![nid, &tgt])?;
                            links_written += 1;
                        }
                    }
                    LinkRef::Name(slug) => {
                        let dst_id = resolve_slug_to_id(&tx, &slug)?;
                        ins_name.execute(params![nid, dst_id, &slug])?;
                        links_written += 1;
                    }
                }
            }
        }
    }

    tx.commit()?;
    Ok((nodes_processed, links_written))
}

/// Find a node id whose `#+name:` slug equals `slug`. Scans the nodes
/// table, decoding each ast_blob just enough to read the name keyword.
/// Returns the first match (lexicographically smallest id on ties).
fn resolve_slug_to_id(conn: &Connection, slug: &str) -> Result<Option<String>, rusqlite::Error> {
    let mut stmt = conn.prepare("SELECT id, ast_blob FROM nodes ORDER BY id")?;
    let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
    for row in rows {
        let (id, blob) = row?;
        let Ok(doc) = sexp::decode_document(&blob) else {
            continue;
        };
        if let Some(node_slug) = extract_name_slug(&doc) {
            if node_slug == slug {
                return Ok(Some(id));
            }
        }
    }
    Ok(None)
}

/// Demote every resolved name-link row currently pointing at `node_id`
/// back to broken (target_id = NULL, target_slug intact). Used when a
/// target node's `#+name:` slug changes; the FK ON DELETE SET NULL
/// handles the delete case automatically.
fn invalidate_resolved_pointing_at(
    conn: &Connection,
    node_id: &str,
) -> Result<(), rusqlite::Error> {
    conn.execute(
        "UPDATE links SET target_id = NULL \
         WHERE link_type = 'name' AND target_id = ?1",
        params![node_id],
    )?;
    Ok(())
}

/// Look up a node's [`Neighborhood`] (incoming + outgoing edges across
/// both link types). Unknown source ids return an empty neighborhood.
///
/// Edges are projected `(other_node_id, link_type)`. For outgoing
/// name-links that are broken (no resolved target_id), the row is
/// omitted from this projection — call [`get_links`] for the full
/// shape including broken name-links.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn get_neighborhood(conn: &Connection, node_id: &str) -> Result<Neighborhood, rusqlite::Error> {
    let outgoing: Vec<(NodeId, LinkType)> = {
        let mut stmt = conn.prepare(
            "SELECT target_id, link_type FROM links \
             WHERE source_id = ?1 AND target_id IS NOT NULL",
        )?;
        let rows = stmt.query_map(params![node_id], |row| {
            Ok((NodeId(row.get(0)?), row.get::<_, LinkType>(1)?))
        })?;
        rows.collect::<Result<_, _>>()?
    };
    let incoming: Vec<(NodeId, LinkType)> = {
        let mut stmt =
            conn.prepare("SELECT source_id, link_type FROM links WHERE target_id = ?1")?;
        let rows = stmt.query_map(params![node_id], |row| {
            Ok((NodeId(row.get(0)?), row.get::<_, LinkType>(1)?))
        })?;
        rows.collect::<Result<_, _>>()?
    };
    Ok(Neighborhood { outgoing, incoming })
}

/// A node's full link neighborhood under the unified graph.
///
/// Carries every column of every row — including broken name-links
/// that have no resolved target_id. Used by `kb links` so the verb
/// can surface `[[slug]] -> (broken)` rows alongside resolved edges.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct LinkNeighborhood {
    /// Outgoing edges from this node. Each row carries `source_id =
    /// node_id`. Broken name-links (target_id NULL) ARE included.
    pub outgoing: Vec<LinkRow>,
    /// Incoming edges into this node. Each row has `target_id =
    /// node_id`; broken rows by definition have no incoming side.
    pub incoming: Vec<LinkRow>,
}

/// Return the full outgoing and incoming link rows for a node under
/// the unified graph. Outgoing rows include broken name-links;
/// incoming rows by definition all have non-NULL target_id.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn get_links(conn: &Connection, node_id: &str) -> Result<LinkNeighborhood, rusqlite::Error> {
    let outgoing: Vec<LinkRow> = {
        let mut stmt = conn.prepare(
            "SELECT source_id, link_type, target_id, target_slug FROM links \
             WHERE source_id = ?1 \
             ORDER BY link_type, target_slug, target_id",
        )?;
        let rows = stmt.query_map(params![node_id], |r| {
            Ok(LinkRow {
                source_id: r.get(0)?,
                link_type: r.get(1)?,
                target_id: r.get(2)?,
                target_slug: r.get(3)?,
            })
        })?;
        rows.collect::<Result<_, _>>()?
    };
    let incoming: Vec<LinkRow> = {
        let mut stmt = conn.prepare(
            "SELECT source_id, link_type, target_id, target_slug FROM links \
             WHERE target_id = ?1 \
             ORDER BY link_type, source_id",
        )?;
        let rows = stmt.query_map(params![node_id], |r| {
            Ok(LinkRow {
                source_id: r.get(0)?,
                link_type: r.get(1)?,
                target_id: r.get(2)?,
                target_slug: r.get(3)?,
            })
        })?;
        rows.collect::<Result<_, _>>()?
    };
    Ok(LinkNeighborhood { outgoing, incoming })
}

/// List orphan nodes — nodes nothing else points at, under either link type.
///
/// A node is an orphan when no `links` row of any `link_type` has a
/// resolved `target_id` equal to its id. Returns rows ordered by
/// `updated_at DESC` so the freshest orphans surface first.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn list_orphans(conn: &Connection) -> Result<Vec<NodeRow>, rusqlite::Error> {
    let mut stmt = conn.prepare(
        "SELECT n.id, n.title, n.ast_blob, n.created_at, n.updated_at \
         FROM nodes n \
         WHERE NOT EXISTS ( \
           SELECT 1 FROM links l \
           WHERE l.target_id = n.id \
         ) \
         ORDER BY n.updated_at DESC",
    )?;
    let rows = stmt.query_map([], |row| {
        Ok(NodeRow {
            id: row.get(0)?,
            title: row.get(1)?,
            ast_blob: row.get(2)?,
            created_at: row.get(3)?,
            updated_at: row.get(4)?,
        })
    })?;
    rows.collect()
}

/// One hub entry: a node id, its title, and its total in-degree
/// across both link types in the unified graph.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HubEntry {
    pub id: String,
    pub title: String,
    pub in_degree: i64,
}

/// Hubs: nodes ranked by total resolved in-degree across both link
/// types (id-links plus resolved name-links). Limit caps the result;
/// `0` returns an empty vec. Ties break lexicographically on node id.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn list_hubs(conn: &Connection, limit: usize) -> Result<Vec<HubEntry>, rusqlite::Error> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let mut stmt = conn.prepare(
        "SELECT n.id, n.title, COUNT(l.source_id) AS in_degree \
         FROM nodes n \
         JOIN links l ON l.target_id = n.id \
         GROUP BY n.id, n.title \
         ORDER BY in_degree DESC, n.id ASC \
         LIMIT ?1",
    )?;
    let rows = stmt.query_map(params![limit as i64], |r| {
        Ok(HubEntry {
            id: r.get(0)?,
            title: r.get(1)?,
            in_degree: r.get(2)?,
        })
    })?;
    rows.collect()
}

/// Broken bracket references in the unified graph.
///
/// Rows with `link_type = 'name'` and `target_id` still NULL —
/// `[[name]]` mentions of a slug no node has claimed via `#+name:`.
/// Id-links cannot be broken (CHECK enforces NOT NULL target_id for
/// them). Ordered by `(target_slug, source_id)` for deterministic
/// output.
///
/// # Errors
///
/// Returns `rusqlite::Error` on database failure.
pub fn list_broken_links(conn: &Connection) -> Result<Vec<LinkRow>, rusqlite::Error> {
    let mut stmt = conn.prepare(
        "SELECT source_id, link_type, target_id, target_slug FROM links \
         WHERE link_type = 'name' AND target_id IS NULL \
         ORDER BY target_slug ASC, source_id ASC",
    )?;
    let rows = stmt.query_map([], |r| {
        Ok(LinkRow {
            source_id: r.get(0)?,
            link_type: r.get(1)?,
            target_id: r.get(2)?,
            target_slug: r.get(3)?,
        })
    })?;
    rows.collect()
}

// ── AST → row-field projection ─────────────────────────────────────────

/// Derive the `nodes.title` from an AST: a `#+title:` keyword line wins;
/// else the first heading title; else the first paragraph's plain text;
/// truncated to 80 chars; else "(untitled)".
#[must_use]
pub fn extract_title(doc: &Document) -> String {
    title_keyword(&doc.blocks)
        .or_else(|| find_first_title(&doc.blocks))
        .map(|t| truncate80(t.trim()))
        .unwrap_or_else(|| "(untitled)".into())
}

/// The value of the first non-empty `#+title:` keyword line, if any.
fn title_keyword(blocks: &[Block]) -> Option<String> {
    blocks.iter().find_map(|block| match block {
        Block::Keyword { name, value } if name.eq_ignore_ascii_case("title") => {
            let trimmed = value.trim();
            (!trimmed.is_empty()).then(|| trimmed.to_string())
        }
        _ => None,
    })
}

fn find_first_title(blocks: &[Block]) -> Option<String> {
    for block in blocks {
        match block {
            Block::Heading {
                title, children, ..
            } => {
                if !title.0.trim().is_empty() {
                    return Some(title.0.clone());
                }
                if let Some(t) = find_first_title(children) {
                    return Some(t);
                }
            }
            Block::Paragraph { inlines } => {
                let text = inline_text(inlines);
                if !text.trim().is_empty() {
                    return Some(text);
                }
            }
            Block::QuoteBlock { children } => {
                if let Some(t) = find_first_title(children) {
                    return Some(t);
                }
            }
            _ => {}
        }
    }
    None
}

fn truncate80(s: &str) -> String {
    if s.chars().count() <= 80 {
        s.to_string()
    } else {
        s.chars().take(80).collect()
    }
}

/// Union of every tag in the document — heading tags plus any
/// `#+filetags:` keyword lines — normalized, deduplicated, in
/// first-seen order. See [`normalize_tag`] for the normal form.
#[must_use]
pub fn extract_tags(doc: &Document) -> Vec<Tag> {
    let mut seen = HashSet::new();
    let mut result = Vec::new();
    collect_tags(&doc.blocks, &mut seen, &mut result);
    result
}

/// Normalize a tag to lowercase kebab-case.
///
/// Alphanumeric characters are lowercased; every run of other characters
/// (spaces, `_`, punctuation) collapses to a single `-`, with leading and
/// trailing separators trimmed. camelCase is not split (`silentCritic` ->
/// `silentcritic`). Applied symmetrically on the write path
/// ([`extract_tags`]) and the query path ([`list_by_tag`]) so lookups
/// match regardless of how the caller cased or spaced the tag.
#[must_use]
pub fn normalize_tag(raw: &str) -> String {
    let mut out = String::with_capacity(raw.len());
    let mut pending_sep = false;
    for ch in raw.chars() {
        if ch.is_alphanumeric() {
            if pending_sep && !out.is_empty() {
                out.push('-');
            }
            pending_sep = false;
            out.extend(ch.to_lowercase());
        } else {
            pending_sep = true;
        }
    }
    out
}

/// Parse an org `#+filetags:` value into bare tag names.
///
/// org-roam writes file tags colon-delimited (`:a:b:c:`); the keyword
/// value is verbatim, so leading/trailing padding and the surrounding
/// colons are stripped here.
fn parse_filetags(value: &str) -> impl Iterator<Item = String> + '_ {
    value
        .split(':')
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(str::to_string)
}

/// Normalize `raw`, then push it onto `out` if non-empty and unseen.
fn push_tag(raw: &str, seen: &mut HashSet<String>, out: &mut Vec<Tag>) {
    let tag = normalize_tag(raw);
    if !tag.is_empty() && seen.insert(tag.clone()) {
        out.push(Tag(tag));
    }
}

fn collect_tags(blocks: &[Block], seen: &mut HashSet<String>, out: &mut Vec<Tag>) {
    for block in blocks {
        match block {
            Block::Heading { tags, children, .. } => {
                for tag in tags {
                    push_tag(&tag.0, seen, out);
                }
                collect_tags(children, seen, out);
            }
            Block::QuoteBlock { children } => {
                collect_tags(children, seen, out);
            }
            Block::Keyword { name, value } if name.eq_ignore_ascii_case("filetags") => {
                for tag in parse_filetags(value) {
                    push_tag(&tag, seen, out);
                }
            }
            _ => {}
        }
    }
}

/// Extract all plain text from a document for FTS5 body indexing.
pub fn extract_body_text(doc: &Document) -> String {
    let mut texts = Vec::new();
    collect_texts(&doc.blocks, &mut texts);
    texts
        .into_iter()
        .filter(|s| !s.is_empty())
        .collect::<Vec<_>>()
        .join(" ")
}

fn collect_texts(blocks: &[Block], out: &mut Vec<String>) {
    for block in blocks {
        match block {
            Block::Heading { children, .. } => collect_texts(children, out),
            Block::Paragraph { inlines } => {
                out.push(inline_text(inlines));
            }
            Block::SrcBlock { content, .. } | Block::ExampleBlock { content } => {
                out.push(content.clone());
            }
            Block::QuoteBlock { children } => collect_texts(children, out),
            Block::List { items, .. } => {
                for item in items {
                    collect_texts(&item.content, out);
                }
            }
            Block::Table { rows } => {
                for row in rows {
                    for cell in row {
                        out.push(inline_text(&cell.inlines));
                    }
                }
            }
            Block::PropertyDrawer { entries } => {
                for (_, v) in entries {
                    out.push(v.clone());
                }
            }
            Block::LogbookDrawer { entries } => {
                for entry in entries {
                    out.push(entry.note.clone());
                }
            }
            Block::Comment { text } => {
                out.push(text.clone());
            }
            Block::Keyword { value, .. } => {
                out.push(value.clone());
            }
            Block::Planning { .. } | Block::BlankLine | Block::HorizontalRule => {}
        }
    }
}

fn inline_text(inlines: &[Inline]) -> String {
    let mut parts = Vec::new();
    for inline in inlines {
        match inline {
            Inline::Plain(t) => parts.push(t.clone()),
            Inline::Bold(is) | Inline::Italic(is) | Inline::Strikethrough(is) => {
                parts.push(inline_text(is));
            }
            Inline::InlineCode(t) | Inline::Verbatim(t) => parts.push(t.clone()),
            Inline::Link {
                description: Some(d),
                ..
            } => parts.push(d.clone()),
            Inline::Link { target, .. } => parts.push(target.clone()),
            Inline::LineBreak => parts.push(" ".to_string()),
        }
    }
    parts.join("")
}

#[cfg(test)]
mod tests {
    use super::*;

    fn setup() -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch("PRAGMA foreign_keys = ON;").unwrap();
        init_db(&conn).unwrap();
        conn
    }

    fn sample_doc() -> Document {
        Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("Test".into()),
                tags: vec![Tag("rust".into())],
                children: vec![Block::Paragraph {
                    inlines: vec![Inline::Plain("hello world".into())],
                }],
            }],
        }
    }

    fn link_doc(targets: &[&str]) -> Document {
        let inlines: Vec<Inline> = targets
            .iter()
            .map(|t| Inline::Link {
                target: format!("id:{t}"),
                description: None,
            })
            .collect();
        Document {
            blocks: vec![Block::Paragraph { inlines }],
        }
    }

    // ── Basic ops (unchanged behaviour) ────────────────────────────────

    #[test]
    fn insert_and_get_node() {
        let conn = setup();
        let id = "test-1";
        insert_node(&conn, id, &sample_doc()).unwrap();
        let doc = get_node(&conn, id).unwrap().unwrap();
        assert_eq!(doc, sample_doc());
    }

    #[test]
    fn get_nonexistent_node() {
        let conn = setup();
        let doc = get_node(&conn, "no-such-id").unwrap();
        assert!(doc.is_none());
    }

    #[test]
    fn update_node_works() {
        let conn = setup();
        let id = "test-2";
        insert_node(&conn, id, &sample_doc()).unwrap();
        let mut updated_doc = sample_doc();
        updated_doc.blocks.push(Block::Paragraph {
            inlines: vec![Inline::Plain("extra".into())],
        });
        let ok = update_node(&conn, id, &updated_doc).unwrap();
        assert!(ok);
        let doc = get_node(&conn, id).unwrap().unwrap();
        assert_eq!(doc, updated_doc);
    }

    #[test]
    fn update_nonexistent_returns_false() {
        let conn = setup();
        let ok = update_node(&conn, "no-such", &sample_doc()).unwrap();
        assert!(!ok);
    }

    #[test]
    fn delete_node_works() {
        let conn = setup();
        let id = "test-3";
        insert_node(&conn, id, &sample_doc()).unwrap();
        let ok = delete_node(&conn, id).unwrap();
        assert!(ok);
        assert!(get_node(&conn, id).unwrap().is_none());
    }

    #[test]
    fn delete_nonexistent_returns_false() {
        let conn = setup();
        let ok = delete_node(&conn, "no-such").unwrap();
        assert!(!ok);
    }

    #[test]
    fn list_by_tag_finds_match() {
        let conn = setup();
        insert_node(&conn, "a", &sample_doc()).unwrap();

        let doc_b = Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("Other".into()),
                tags: vec![Tag("python".into())],
                children: vec![],
            }],
        };
        insert_node(&conn, "b", &doc_b).unwrap();

        let results = list_by_tag(&conn, "rust").unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].id.as_str(), "a");
    }

    #[test]
    fn list_recent_newest_first() {
        let conn = setup();
        insert_node(&conn, "first", &sample_doc()).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(10));
        insert_node(&conn, "second", &sample_doc()).unwrap();
        let recent = list_recent(&conn, 10).unwrap();
        assert_eq!(recent.first().unwrap().id.as_str(), "second");
    }

    #[test]
    fn title_from_first_heading() {
        let doc = Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("My Note".into()),
                tags: vec![],
                children: vec![],
            }],
        };
        assert_eq!(extract_title(&doc), "My Note");
    }

    #[test]
    fn title_from_paragraph_fallback() {
        let doc = Document {
            blocks: vec![Block::Paragraph {
                inlines: vec![Inline::Plain("First paragraph text".into())],
            }],
        };
        assert_eq!(extract_title(&doc), "First paragraph text");
    }

    #[test]
    fn title_untitled_when_empty() {
        let doc = Document { blocks: vec![] };
        assert_eq!(extract_title(&doc), "(untitled)");
    }

    #[test]
    fn title_truncated_at_80() {
        let long = "a".repeat(100);
        let doc = Document {
            blocks: vec![Block::Paragraph {
                inlines: vec![Inline::Plain(long)],
            }],
        };
        let title = extract_title(&doc);
        assert_eq!(title.chars().count(), 80);
    }

    #[test]
    fn extract_tags_collects_all() {
        let doc = Document {
            blocks: vec![
                Block::Heading {
                    level: 1,
                    title: Title("A".into()),
                    tags: vec![Tag("rust".into()), Tag("kb".into())],
                    children: vec![Block::Heading {
                        level: 2,
                        title: Title("B".into()),
                        tags: vec![Tag("testing".into())],
                        children: vec![],
                    }],
                },
                Block::Heading {
                    level: 1,
                    title: Title("C".into()),
                    tags: vec![Tag("rust".into())], // duplicate
                    children: vec![],
                },
            ],
        };
        let tags = extract_tags(&doc);
        assert_eq!(tags.len(), 3);
        assert_eq!(tags[0].0, "rust");
        assert_eq!(tags[1].0, "kb");
        assert_eq!(tags[2].0, "testing");
    }

    #[test]
    fn extract_tags_includes_filetags_keyword() {
        // org-roam document shape: a `#+filetags:` line, no heading tags.
        let doc = Document {
            blocks: vec![
                Block::Keyword {
                    name: "filetags".into(),
                    value: " :design:claude-memory:project:adr:".into(),
                },
                Block::Heading {
                    level: 1,
                    title: Title("Decision".into()),
                    tags: vec![Tag("design".into())], // duplicate of a filetag
                    children: vec![],
                },
            ],
        };
        let extracted = extract_tags(&doc);
        let tags: Vec<&str> = extracted.iter().map(|t| t.0.as_str()).collect();
        assert_eq!(tags, ["design", "claude-memory", "project", "adr"]);
    }

    #[test]
    fn normalize_tag_lowercase_kebab() {
        assert_eq!(normalize_tag("Rust"), "rust");
        assert_eq!(normalize_tag("Claude Memory"), "claude-memory");
        assert_eq!(normalize_tag("claude_memory"), "claude-memory");
        assert_eq!(normalize_tag("silent-critic"), "silent-critic");
        assert_eq!(normalize_tag("silentCritic"), "silentcritic");
        assert_eq!(normalize_tag("  spaced  tag  "), "spaced-tag");
        assert_eq!(normalize_tag("+++"), "");
    }

    #[test]
    fn extract_tags_normalizes_heading_tags() {
        let doc = Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("H".into()),
                tags: vec![Tag("Rust".into()), Tag("rust".into())],
                children: vec![],
            }],
        };
        let tags = extract_tags(&doc);
        // Both collapse to one normalized tag.
        assert_eq!(tags.len(), 1);
        assert_eq!(tags[0].0, "rust");
    }

    #[test]
    fn extract_title_prefers_title_keyword() {
        let doc = Document {
            blocks: vec![
                Block::Keyword {
                    name: "title".into(),
                    value: " The Real Title".into(),
                },
                Block::Heading {
                    level: 1,
                    title: Title("First Heading".into()),
                    tags: vec![],
                    children: vec![],
                },
            ],
        };
        assert_eq!(extract_title(&doc), "The Real Title");
    }

    #[test]
    fn extract_title_falls_back_to_heading_without_keyword() {
        let doc = Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("First Heading".into()),
                tags: vec![],
                children: vec![],
            }],
        };
        assert_eq!(extract_title(&doc), "First Heading");
    }

    #[test]
    fn sexp_blob_roundtrip() {
        let conn = setup();
        let id = "sexp-test";
        insert_node(&conn, id, &sample_doc()).unwrap();
        let row = get_node_row(&conn, id).unwrap().unwrap();
        assert!(row.ast_blob.starts_with("(kb-doc 1"));
        let decoded = sexp::decode_document(&row.ast_blob).unwrap();
        assert_eq!(decoded, sample_doc());
    }

    // ── Schema parity ──────────────────────────────────────────────────

    #[test]
    fn schema_version_constant_is_two() {
        assert_eq!(CURRENT_SCHEMA_VERSION, 2);
    }

    #[test]
    fn migration_reencodes_legacy_ordered_list_blob() {
        let conn = setup();
        let id = "11111111-1111-1111-1111-111111111111";
        insert_node(&conn, id, &sample_doc()).unwrap();
        // Rewrite the row to the v1 blob format (bare `ordered` symbol)
        // and roll user_version back so the v2 migration re-runs.
        let legacy = "(kb-doc 1 (list ordered (item no-checkbox (paragraph (plain \"x\")))))";
        conn.execute(
            "UPDATE nodes SET ast_blob = ?1 WHERE id = ?2",
            params![legacy, id],
        )
        .unwrap();
        conn.execute_batch("PRAGMA user_version = 1;").unwrap();

        init_db(&conn).unwrap();

        let row = get_node_row(&conn, id).unwrap().unwrap();
        assert!(row.ast_blob.contains("(list (ordered 1)"));
        let full = get_node_full(&conn, id).unwrap().unwrap();
        assert_eq!(
            full.document.blocks,
            vec![Block::List {
                list_type: ListType::Ordered(1),
                items: vec![ListItem {
                    content: vec![Block::Paragraph {
                        inlines: vec![Inline::Plain("x".into())],
                    }],
                    checkbox: Checkbox::NoCheckbox,
                }],
            }]
        );
    }

    #[test]
    fn migration_skipped_at_current_version() {
        let conn = setup();
        let id = "22222222-2222-2222-2222-222222222222";
        insert_node(&conn, id, &sample_doc()).unwrap();
        // At user_version 2 the blob pass must not run: a legacy-format
        // blob planted now survives a second init_db untouched.
        let legacy = "(kb-doc 1 (list ordered (item no-checkbox (paragraph (plain \"x\")))))";
        conn.execute(
            "UPDATE nodes SET ast_blob = ?1 WHERE id = ?2",
            params![legacy, id],
        )
        .unwrap();

        init_db(&conn).unwrap();

        let row = get_node_row(&conn, id).unwrap().unwrap();
        assert_eq!(row.ast_blob, legacy);
    }

    #[test]
    fn get_node_reports_corrupt_blob() {
        let conn = setup();
        let id = "44444444-4444-4444-4444-444444444444";
        insert_node(&conn, id, &sample_doc()).unwrap();
        conn.execute(
            "UPDATE nodes SET ast_blob = '(not a kb-doc' WHERE id = ?1",
            params![id],
        )
        .unwrap();
        match get_node(&conn, id).unwrap_err() {
            KbError::CorruptAstBlob { node_id, .. } => assert_eq!(node_id, id),
            other => panic!("expected CorruptAstBlob, got {other}"),
        }
    }

    #[test]
    fn get_node_full_reports_corrupt_blob() {
        let conn = setup();
        let id = "55555555-5555-5555-5555-555555555555";
        insert_node(&conn, id, &sample_doc()).unwrap();
        conn.execute(
            "UPDATE nodes SET ast_blob = '(not a kb-doc' WHERE id = ?1",
            params![id],
        )
        .unwrap();
        match get_node_full(&conn, id).unwrap_err() {
            KbError::CorruptAstBlob { node_id, .. } => assert_eq!(node_id, id),
            other => panic!("expected CorruptAstBlob, got {other}"),
        }
    }

    #[test]
    fn migration_reports_corrupt_blob() {
        let conn = setup();
        let id = "33333333-3333-3333-3333-333333333333";
        insert_node(&conn, id, &sample_doc()).unwrap();
        conn.execute(
            "UPDATE nodes SET ast_blob = '(not a kb-doc' WHERE id = ?1",
            params![id],
        )
        .unwrap();
        conn.execute_batch("PRAGMA user_version = 1;").unwrap();

        let err = init_db(&conn).unwrap_err();
        match err {
            KbError::CorruptAstBlob { node_id, .. } => assert_eq!(node_id, id),
            other => panic!("expected CorruptAstBlob, got {other}"),
        }
    }

    #[test]
    fn schema_version_pragma_is_stamped() {
        let conn = setup();
        let v: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(v, i64::from(CURRENT_SCHEMA_VERSION));
    }

    #[test]
    fn schema_links_table_columns_and_pk() {
        let conn = setup();
        let cols: Vec<(String, String, i64, i64)> = conn
            .prepare("PRAGMA table_info(links)")
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(1)?, // name
                    r.get::<_, String>(2)?, // type
                    r.get::<_, i64>(3)?,    // notnull
                    r.get::<_, i64>(5)?,    // pk
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        // Unified schema: source_id NOT NULL, link_type NOT NULL,
        // target_id NULLABLE (set NULL on target delete for name-links),
        // target_slug NULLABLE (NULL for id-links, NOT NULL for
        // name-links — enforced by CHECK). PK composite over all four.
        assert_eq!(
            cols,
            vec![
                ("source_id".into(), "TEXT".into(), 1, 1),
                ("link_type".into(), "TEXT".into(), 1, 2),
                ("target_id".into(), "TEXT".into(), 0, 3),
                ("target_slug".into(), "TEXT".into(), 0, 4),
            ]
        );
    }

    #[test]
    fn schema_links_table_foreign_keys() {
        let conn = setup();
        let fks: Vec<(String, String, String, String)> = conn
            .prepare("PRAGMA foreign_key_list(links)")
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(2)?, // table
                    r.get::<_, String>(3)?, // from
                    r.get::<_, String>(4)?, // to
                    r.get::<_, String>(6)?, // on_delete
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(fks.len(), 2);
        let mut by_from: std::collections::HashMap<String, (String, String, String)> =
            std::collections::HashMap::new();
        for (table, from, to, on_delete) in fks {
            by_from.insert(from, (table, to, on_delete));
        }
        let src = by_from.get("source_id").expect("source_id FK present");
        assert_eq!(src.0, "nodes");
        assert_eq!(src.1, "id");
        assert_eq!(src.2, "CASCADE");
        let tgt = by_from.get("target_id").expect("target_id FK present");
        assert_eq!(tgt.0, "nodes");
        assert_eq!(tgt.1, "id");
        // Name-link rows demote to broken when the target node is
        // deleted; id-link rows pointing at the deleted node are
        // removed manually in `delete_node` before the cascade fires.
        assert_eq!(tgt.2, "SET NULL");
    }

    #[test]
    fn links_table_check_constraint_rejects_invalid_shapes() {
        let conn = setup();
        // Need a source node so the FK on source_id holds.
        insert_node(&conn, "src", &empty_doc()).unwrap();
        // id-link with NULL target_id -> CHECK violation.
        let e = conn.execute(
            "INSERT INTO links (source_id, link_type, target_id, target_slug) \
             VALUES ('src', 'id', NULL, NULL)",
            [],
        );
        assert!(e.is_err());
        // id-link with target_slug set -> CHECK violation.
        let e = conn.execute(
            "INSERT INTO links (source_id, link_type, target_id, target_slug) \
             VALUES ('src', 'id', 'src', 'slug')",
            [],
        );
        assert!(e.is_err());
        // name-link with NULL target_slug -> CHECK violation.
        let e = conn.execute(
            "INSERT INTO links (source_id, link_type, target_id, target_slug) \
             VALUES ('src', 'name', NULL, NULL)",
            [],
        );
        assert!(e.is_err());
    }

    #[test]
    fn schema_name_links_table_does_not_exist() {
        let conn = setup();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='name_links'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 0, "pre-consolidation name_links table must not exist");
    }

    #[test]
    fn schema_links_target_index_present() {
        let conn = setup();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name='links_target_idx' AND tbl_name='links'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 1);
    }

    #[test]
    fn schema_audit_log_columns() {
        let conn = setup();
        let cols: Vec<(String, String, i64, i64)> = conn
            .prepare("PRAGMA table_info(audit_log)")
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(1)?, // name
                    r.get::<_, String>(2)?, // type
                    r.get::<_, i64>(3)?,    // notnull
                    r.get::<_, i64>(5)?,    // pk
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(
            cols,
            vec![
                ("id".into(), "INTEGER".into(), 0, 1),
                ("node_id".into(), "TEXT".into(), 1, 0),
                ("operation".into(), "TEXT".into(), 1, 0),
                ("old_blob".into(), "TEXT".into(), 0, 0),
                ("new_blob".into(), "TEXT".into(), 0, 0),
                ("timestamp".into(), "TEXT".into(), 1, 0),
            ]
        );
    }

    #[test]
    fn schema_audit_log_has_no_foreign_keys() {
        let conn = setup();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM pragma_foreign_key_list('audit_log')",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 0);
    }

    #[test]
    fn schema_audit_log_autoincrement_works() {
        let conn = setup();
        conn.execute(
            "INSERT INTO audit_log (node_id, operation, timestamp) VALUES ('x', 'insert', 'now')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO audit_log (node_id, operation, timestamp) VALUES ('y', 'insert', 'now')",
            [],
        )
        .unwrap();
        let ids: Vec<i64> = conn
            .prepare("SELECT id FROM audit_log ORDER BY id")
            .unwrap()
            .query_map([], |r| r.get::<_, i64>(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(ids, vec![1, 2]);
        // sqlite_sequence is created only when AUTOINCREMENT is in effect.
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='sqlite_sequence'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 1);
    }

    #[test]
    fn schema_audit_log_indices_present() {
        let conn = setup();
        for name in ["audit_log_node_idx", "audit_log_ts_idx"] {
            let n: i64 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name=?1 AND tbl_name='audit_log'",
                    params![name],
                    |r| r.get(0),
                )
                .unwrap();
            assert_eq!(n, 1, "missing index {name}");
        }
    }

    // ── Audit log writers ──────────────────────────────────────────────

    fn fetch_audit_rows(conn: &Connection, node_id: &str) -> Vec<AuditRow> {
        conn.prepare(
            "SELECT id, node_id, operation, old_blob, new_blob, timestamp FROM audit_log WHERE node_id = ?1 ORDER BY id",
        )
        .unwrap()
        .query_map(params![node_id], |r| {
            Ok(AuditRow {
                id: r.get(0)?,
                node_id: r.get(1)?,
                operation: r.get(2)?,
                old_blob: r.get(3)?,
                new_blob: r.get(4)?,
                timestamp: r.get(5)?,
            })
        })
        .unwrap()
        .collect::<Result<_, _>>()
        .unwrap()
    }

    #[test]
    fn audit_writers_insert_writes_one_row() {
        let conn = setup();
        insert_node(&conn, "n", &sample_doc()).unwrap();
        let rows = fetch_audit_rows(&conn, "n");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].operation, "insert");
        assert!(rows[0].old_blob.is_none());
        assert!(
            rows[0]
                .new_blob
                .as_deref()
                .unwrap()
                .starts_with("(kb-doc 1")
        );
        // ISO8601 with millisecond precision and Z suffix.
        let ts = &rows[0].timestamp;
        assert!(
            ts.ends_with('Z') && ts.contains('.'),
            "timestamp not ISO8601-ms: {ts}"
        );
    }

    #[test]
    fn audit_writers_update_records_old_and_new() {
        let conn = setup();
        insert_node(&conn, "n", &sample_doc()).unwrap();
        let mut next = sample_doc();
        next.blocks.push(Block::Paragraph {
            inlines: vec![Inline::Plain("more".into())],
        });
        update_node(&conn, "n", &next).unwrap();
        let rows = fetch_audit_rows(&conn, "n");
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[1].operation, "update");
        let old = rows[1].old_blob.as_deref().unwrap();
        let new = rows[1].new_blob.as_deref().unwrap();
        assert_ne!(old, new);
        assert!(new.contains("more"));
    }

    #[test]
    fn audit_writers_delete_records_old_only() {
        let conn = setup();
        insert_node(&conn, "n", &sample_doc()).unwrap();
        delete_node(&conn, "n").unwrap();
        let rows = fetch_audit_rows(&conn, "n");
        // History survives the delete (audit_log has no FK).
        assert_eq!(rows.len(), 2);
        assert_eq!(rows[1].operation, "delete");
        assert!(rows[1].old_blob.is_some());
        assert!(rows[1].new_blob.is_none());
    }

    #[test]
    fn audit_writers_timestamp_format_matches_node_created_at() {
        let conn = setup();
        insert_node(&conn, "n", &sample_doc()).unwrap();
        let node_ts: String = conn
            .query_row("SELECT created_at FROM nodes WHERE id = 'n'", [], |r| {
                r.get(0)
            })
            .unwrap();
        let audit_ts: String = conn
            .query_row(
                "SELECT timestamp FROM audit_log WHERE node_id = 'n'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(is_iso8601_ms_z(&node_ts), "node ts: {node_ts}");
        assert!(is_iso8601_ms_z(&audit_ts), "audit ts: {audit_ts}");
    }

    /// Match `YYYY-MM-DDTHH:MM:SS.fffZ` (millisecond precision, UTC).
    fn is_iso8601_ms_z(s: &str) -> bool {
        let bytes = s.as_bytes();
        if bytes.len() != 24 {
            return false;
        }
        // Expected layout (1-indexed positions):
        //  YYYY = 0..4    digits
        //   '-' = 4
        //   MM  = 5..7    digits
        //   '-' = 7
        //   DD  = 8..10   digits
        //   'T' = 10
        //   HH  = 11..13  digits
        //   ':' = 13
        //   MM  = 14..16  digits
        //   ':' = 16
        //   SS  = 17..19  digits
        //   '.' = 19
        //   fff = 20..23  digits
        //   'Z' = 23
        let digits =
            |range: std::ops::Range<usize>| range.into_iter().all(|i| bytes[i].is_ascii_digit());
        digits(0..4)
            && bytes[4] == b'-'
            && digits(5..7)
            && bytes[7] == b'-'
            && digits(8..10)
            && bytes[10] == b'T'
            && digits(11..13)
            && bytes[13] == b':'
            && digits(14..16)
            && bytes[16] == b':'
            && digits(17..19)
            && bytes[19] == b'.'
            && digits(20..23)
            && bytes[23] == b'Z'
    }

    /// Id-link extraction walks every nested Block position
    /// (heading children, quote blocks, list items, table cells).
    /// Preserved from the dropped `extract_id_links_walks_nested_positions`
    /// shim test since the `link_parser_extracts_walks_nested_inline_positions`
    /// test only exercises nested `Inline` positions (Bold/Italic), not
    /// nested `Block` positions.
    #[test]
    fn extract_links_walks_nested_block_positions() {
        let inner_link = Inline::Link {
            target: "id:nested".into(),
            description: Some("nested".into()),
        };
        let doc = Document {
            blocks: vec![
                Block::Heading {
                    level: 1,
                    title: Title("h".into()),
                    tags: vec![],
                    children: vec![Block::Paragraph {
                        inlines: vec![Inline::Bold(vec![inner_link])],
                    }],
                },
                Block::QuoteBlock {
                    children: vec![Block::Paragraph {
                        inlines: vec![Inline::Italic(vec![Inline::Link {
                            target: "id:in_quote".into(),
                            description: None,
                        }])],
                    }],
                },
                Block::List {
                    list_type: ListType::Unordered,
                    items: vec![ListItem {
                        content: vec![Block::Paragraph {
                            inlines: vec![Inline::Link {
                                target: "id:in_list".into(),
                                description: None,
                            }],
                        }],
                        checkbox: Checkbox::NoCheckbox,
                    }],
                },
                Block::Table {
                    rows: vec![vec![TableCell {
                        inlines: vec![Inline::Link {
                            target: "id:in_table".into(),
                            description: None,
                        }],
                    }]],
                },
            ],
        };
        let ids: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Id(s) => Some(s),
                LinkRef::Name(_) => None,
            })
            .collect();
        assert_eq!(
            ids,
            vec![
                "nested".to_string(),
                "in_quote".to_string(),
                "in_list".to_string(),
                "in_table".to_string(),
            ]
        );
    }

    // ── relink_one ─────────────────────────────────────────────────────

    fn empty_doc() -> Document {
        Document { blocks: vec![] }
    }

    #[test]
    fn relink_one_inserts_existing_targets() {
        let conn = setup();
        insert_node(&conn, "src", &empty_doc()).unwrap();
        insert_node(&conn, "tgt1", &empty_doc()).unwrap();
        insert_node(&conn, "tgt2", &empty_doc()).unwrap();
        relink_one(&conn, "src", &link_doc(&["tgt1", "tgt2"])).unwrap();

        let mut rows: Vec<(String, String)> = conn
            .prepare("SELECT target_id, link_type FROM links WHERE source_id = 'src'")
            .unwrap()
            .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        rows.sort();
        assert_eq!(
            rows,
            vec![
                ("tgt1".to_string(), "id".to_string()),
                ("tgt2".to_string(), "id".to_string()),
            ]
        );
    }

    #[test]
    fn relink_one_drops_forward_refs() {
        let conn = setup();
        insert_node(&conn, "src", &empty_doc()).unwrap();
        // tgt-missing does not exist as a node yet
        relink_one(&conn, "src", &link_doc(&["tgt-missing"])).unwrap();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM links WHERE source_id = 'src'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 0);
    }

    #[test]
    fn relink_one_replaces_prior_outgoing() {
        let conn = setup();
        insert_node(&conn, "src", &empty_doc()).unwrap();
        insert_node(&conn, "old", &empty_doc()).unwrap();
        insert_node(&conn, "new", &empty_doc()).unwrap();
        relink_one(&conn, "src", &link_doc(&["old"])).unwrap();
        relink_one(&conn, "src", &link_doc(&["new"])).unwrap();
        let rows: Vec<String> = conn
            .prepare("SELECT target_id FROM links WHERE source_id = 'src'")
            .unwrap()
            .query_map([], |r| r.get::<_, String>(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(rows, vec!["new".to_string()]);
    }

    // ── relink_all ─────────────────────────────────────────────────────

    #[test]
    fn relink_all_resolves_forward_refs() {
        let conn = setup();
        // a links to b before b exists
        insert_node(&conn, "a", &link_doc(&["b"])).unwrap();
        insert_node(&conn, "b", &empty_doc()).unwrap();
        // a's outgoing link should be empty (forward ref dropped)
        let n_before: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM links WHERE source_id = 'a'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_before, 0);

        let (nodes, links) = relink_all(&conn).unwrap();
        assert_eq!(nodes, 2);
        assert_eq!(links, 1);
        let rows: Vec<String> = conn
            .prepare("SELECT target_id FROM links WHERE source_id = 'a'")
            .unwrap()
            .query_map([], |r| r.get::<_, String>(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(rows, vec!["b".to_string()]);
    }

    #[test]
    fn relink_all_skips_corrupt_blob() {
        let conn = setup();
        insert_node(&conn, "good", &empty_doc()).unwrap();
        // Inject a corrupt blob bypassing insert_node.
        conn.execute(
            "INSERT INTO nodes (id, title, ast_blob, created_at, updated_at) VALUES ('bad', 'x', 'not-a-sexp', '0', '0')",
            [],
        )
        .unwrap();
        let (nodes, links) = relink_all(&conn).unwrap();
        assert_eq!(nodes, 2);
        assert_eq!(links, 0);
    }

    #[test]
    fn relink_all_clears_stale_links_first() {
        let conn = setup();
        insert_node(&conn, "src", &empty_doc()).unwrap();
        insert_node(&conn, "tgt", &empty_doc()).unwrap();
        // Manually insert a stale link to a (still-existing) target.
        conn.execute(
            "INSERT INTO links (source_id, target_id, link_type) VALUES ('src', 'tgt', 'id')",
            [],
        )
        .unwrap();
        // No nodes reference anything in their blobs, so relink_all
        // should drop the stale link.
        let (_, links) = relink_all(&conn).unwrap();
        assert_eq!(links, 0);
        let n: i64 = conn
            .query_row("SELECT COUNT(*) FROM links", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 0);
    }

    // ── Neighborhood ───────────────────────────────────────────────────

    #[test]
    fn neighborhood_unknown_id_is_empty() {
        let conn = setup();
        let n = get_neighborhood(&conn, "nope").unwrap();
        assert!(n.outgoing.is_empty());
        assert!(n.incoming.is_empty());
    }

    #[test]
    fn neighborhood_returns_outgoing_and_incoming() {
        let conn = setup();
        insert_node(&conn, "a", &empty_doc()).unwrap();
        insert_node(&conn, "b", &empty_doc()).unwrap();
        insert_node(&conn, "c", &empty_doc()).unwrap();
        // a -> b, c -> a
        relink_one(&conn, "a", &link_doc(&["b"])).unwrap();
        relink_one(&conn, "c", &link_doc(&["a"])).unwrap();

        let n = get_neighborhood(&conn, "a").unwrap();
        assert_eq!(n.outgoing.len(), 1);
        assert_eq!(n.outgoing[0].0.0, "b");
        assert_eq!(n.outgoing[0].1, LinkType::Id);
        assert_eq!(n.incoming.len(), 1);
        assert_eq!(n.incoming[0].0.0, "c");
        assert_eq!(n.incoming[0].1, LinkType::Id);
    }

    // ── list_all_nodes ─────────────────────────────────────────────────

    #[test]
    fn list_all_nodes_zero_limit_empty() {
        let conn = setup();
        insert_node(&conn, "a", &sample_doc()).unwrap();
        let v = list_all_nodes(&conn, 0, 0).unwrap();
        assert!(v.is_empty());
    }

    #[test]
    fn list_all_nodes_orders_by_updated_at_desc() {
        let conn = setup();
        insert_node(&conn, "first", &sample_doc()).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(10));
        insert_node(&conn, "second", &sample_doc()).unwrap();
        std::thread::sleep(std::time::Duration::from_millis(10));
        insert_node(&conn, "third", &sample_doc()).unwrap();

        let v = list_all_nodes(&conn, 10, 0).unwrap();
        let ids: Vec<String> = v.iter().map(|(n, _)| n.0.clone()).collect();
        assert_eq!(ids, vec!["third", "second", "first"]);
    }

    #[test]
    fn list_all_nodes_paginates() {
        let conn = setup();
        for i in 0..5 {
            insert_node(&conn, &format!("n{i}"), &sample_doc()).unwrap();
            std::thread::sleep(std::time::Duration::from_millis(2));
        }
        let page1 = list_all_nodes(&conn, 2, 0).unwrap();
        let page2 = list_all_nodes(&conn, 2, 2).unwrap();
        assert_eq!(page1.len(), 2);
        assert_eq!(page2.len(), 2);
        // No overlap.
        for (id1, _) in &page1 {
            for (id2, _) in &page2 {
                assert_ne!(id1.0, id2.0);
            }
        }
    }

    // ── Write-path relinks ────────────────────────────────────────────

    #[test]
    fn write_path_relinks_on_insert() {
        let conn = setup();
        insert_node(&conn, "tgt", &empty_doc()).unwrap();
        insert_node(&conn, "src", &link_doc(&["tgt"])).unwrap();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM links WHERE source_id = 'src' AND target_id = 'tgt'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 1);
    }

    #[test]
    fn write_path_relinks_on_update() {
        let conn = setup();
        insert_node(&conn, "tgt1", &empty_doc()).unwrap();
        insert_node(&conn, "tgt2", &empty_doc()).unwrap();
        insert_node(&conn, "src", &link_doc(&["tgt1"])).unwrap();
        update_node(&conn, "src", &link_doc(&["tgt2"])).unwrap();
        let rows: Vec<String> = conn
            .prepare("SELECT target_id FROM links WHERE source_id = 'src'")
            .unwrap()
            .query_map([], |r| r.get::<_, String>(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(rows, vec!["tgt2".to_string()]);
    }

    #[test]
    fn write_path_relinks_drop_forward_refs_silently() {
        let conn = setup();
        // Source linking to a not-yet-existing target.
        insert_node(&conn, "src", &link_doc(&["future"])).unwrap();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM links WHERE source_id = 'src'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 0);
        // After the target shows up, relink_all picks it up.
        insert_node(&conn, "future", &empty_doc()).unwrap();
        let (_, links) = relink_all(&conn).unwrap();
        assert_eq!(links, 1);
    }

    // ── Embeddings: schema parity ─────────────────────────────────────

    #[test]
    fn schema_embeddings_table_columns_and_pk() {
        let conn = setup();
        let cols: Vec<(String, String, i64, i64)> = conn
            .prepare("PRAGMA table_info(embeddings)")
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(1)?, // name
                    r.get::<_, String>(2)?, // type
                    r.get::<_, i64>(3)?,    // notnull
                    r.get::<_, i64>(5)?,    // pk
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(
            cols,
            vec![
                ("node_id".into(), "TEXT".into(), 1, 1),
                ("model".into(), "TEXT".into(), 1, 2),
                ("embedding".into(), "BLOB".into(), 1, 0),
            ]
        );
    }

    #[test]
    fn schema_embeddings_table_foreign_keys() {
        let conn = setup();
        let fks: Vec<(String, String, String, String)> = conn
            .prepare("PRAGMA foreign_key_list(embeddings)")
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(2)?, // table
                    r.get::<_, String>(3)?, // from
                    r.get::<_, String>(4)?, // to
                    r.get::<_, String>(6)?, // on_delete
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(fks.len(), 1);
        let (table, from, to, on_delete) = &fks[0];
        assert_eq!(table, "nodes");
        assert_eq!(from, "node_id");
        assert_eq!(to, "id");
        assert_eq!(on_delete, "CASCADE");
    }

    #[test]
    fn schema_embeddings_model_index_present() {
        let conn = setup();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name='embeddings_model_idx' AND tbl_name='embeddings'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 1);
    }

    #[test]
    fn schema_embeddings_version_unchanged() {
        // The embeddings feature must NOT bump the schema version on its
        // own; the stamped version tracks CURRENT_SCHEMA_VERSION.
        let conn = setup();
        let v: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(v, i64::from(CURRENT_SCHEMA_VERSION));
    }

    #[test]
    fn schema_embeddings_table_cascades_on_node_delete() {
        let conn = setup();
        insert_node(&conn, "n", &sample_doc()).unwrap();
        conn.execute(
            "INSERT INTO embeddings (node_id, model, embedding) VALUES ('n', 'm', X'00000000')",
            [],
        )
        .unwrap();
        delete_node(&conn, "n").unwrap();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM embeddings WHERE node_id = 'n'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 0, "FOREIGN KEY ON DELETE CASCADE must clear embeddings");
    }

    // ── cosine_similarity ────────────────────────────────────────────

    #[test]
    fn cosine_similarity_identical_vectors_is_one() {
        let v = vec![1.0_f32, 2.0, 3.0];
        let s = cosine_similarity(&v, &v);
        assert!((s - 1.0).abs() < 1e-6, "got {s}");
    }

    #[test]
    fn cosine_similarity_orthogonal_vectors_is_zero() {
        let s = cosine_similarity(&[1.0_f32, 0.0], &[0.0, 1.0]);
        assert!(s.abs() < 1e-6, "got {s}");
    }

    #[test]
    fn cosine_similarity_opposite_vectors_is_negative_one() {
        let s = cosine_similarity(&[1.0_f32, 0.0], &[-1.0, 0.0]);
        assert!((s + 1.0).abs() < 1e-6, "got {s}");
    }

    #[test]
    fn cosine_similarity_length_mismatch_returns_zero() {
        let s = cosine_similarity(&[1.0_f32, 2.0], &[1.0, 2.0, 3.0]);
        assert!(s.to_bits() == 0.0_f32.to_bits(), "got {s}");
    }

    #[test]
    fn cosine_similarity_zero_norm_returns_zero() {
        let s = cosine_similarity(&[0.0_f32, 0.0, 0.0], &[1.0, 2.0, 3.0]);
        assert!(s.to_bits() == 0.0_f32.to_bits(), "got {s}");
        let s = cosine_similarity(&[1.0_f32, 2.0], &[0.0, 0.0]);
        assert!(s.to_bits() == 0.0_f32.to_bits(), "got {s}");
    }

    #[test]
    fn cosine_similarity_empty_vectors_is_zero() {
        // Empty vectors have norm 0 -> 0.
        let s = cosine_similarity(&[], &[]);
        assert!(s.to_bits() == 0.0_f32.to_bits(), "got {s}");
    }

    // ── reciprocal_rank_fusion ───────────────────────────────────────

    #[test]
    fn rrf_empty_lists_yields_empty() {
        let out = reciprocal_rank_fusion(60, &[]);
        assert!(out.is_empty());
        let out = reciprocal_rank_fusion(60, &[Vec::<NodeId>::new(), Vec::<NodeId>::new()]);
        assert!(out.is_empty());
    }

    #[test]
    fn rrf_single_list_preserves_order() {
        let l = vec![NodeId("a".into()), NodeId("b".into()), NodeId("c".into())];
        let out = reciprocal_rank_fusion(60, std::slice::from_ref(&l));
        assert_eq!(out, l);
    }

    #[test]
    fn rrf_two_lists_combine_scores() {
        let l1 = vec![NodeId("a".into()), NodeId("b".into())];
        let l2 = vec![NodeId("b".into()), NodeId("a".into())];
        // Both nodes appear in both lists. With k=60:
        //   a: 1/61 + 1/62 = 0.0163934 + 0.0161290 = 0.0325224
        //   b: 1/61 + 1/62 = same
        // Tie-broken by ascending id -> a, b.
        let out = reciprocal_rank_fusion(60, &[l1, l2]);
        assert_eq!(out, vec![NodeId("a".into()), NodeId("b".into())]);
    }

    #[test]
    fn rrf_node_only_in_one_list_ranks_lower() {
        // a in list 1 only; b in both lists.
        let l1 = vec![NodeId("a".into()), NodeId("b".into())];
        let l2 = vec![NodeId("b".into())];
        // a: 1/61 = 0.01639
        // b: 1/62 + 1/61 = 0.01613 + 0.01639 = 0.03252
        // -> b first, a second.
        let out = reciprocal_rank_fusion(60, &[l1, l2]);
        assert_eq!(out, vec![NodeId("b".into()), NodeId("a".into())]);
    }

    #[test]
    fn rrf_ties_break_by_ascending_id() {
        let l1 = vec![NodeId("z".into()), NodeId("a".into()), NodeId("m".into())];
        let l2 = vec![NodeId("z".into()), NodeId("a".into()), NodeId("m".into())];
        // All three appear at identical ranks in both lists -> equal scores.
        // Tie-break by ascending id -> a, m, z (within each rank tier).
        // But ranks differ: z@1, a@2, m@3 in both. Scores:
        //   z: 2/61
        //   a: 2/62
        //   m: 2/63
        // Distinct, so order is z, a, m.
        let out = reciprocal_rank_fusion(60, &[l1, l2]);
        assert_eq!(
            out,
            vec![NodeId("z".into()), NodeId("a".into()), NodeId("m".into())]
        );
    }

    #[test]
    fn rrf_truly_tied_scores_break_by_ascending_id() {
        // Force a real tie: same node at rank 1 in two lists has same score.
        // To engineer two distinct nodes with truly equal scores, give each
        // one the same rank in some list.
        let l1 = vec![NodeId("z".into())]; // z@1 -> 1/61
        let l2 = vec![NodeId("a".into())]; // a@1 -> 1/61
        let out = reciprocal_rank_fusion(60, &[l1, l2]);
        // Equal score -> ascending id -> a, z.
        assert_eq!(out, vec![NodeId("a".into()), NodeId("z".into())]);
    }

    // ── try_load_vec_extension ───────────────────────────────────────

    #[test]
    fn vec_extension_missing_path_returns_err() {
        let conn = setup();
        let err = try_load_vec_extension(&conn, "/nonexistent/path/to/sqlite-vec.dylib")
            .expect_err("loading a missing path must fail");
        assert!(matches!(err, VecExtensionError::LoadExtension(_)));
    }

    #[test]
    fn vec_extension_failure_leaves_connection_usable() {
        let conn = setup();
        let _ = try_load_vec_extension(&conn, "/nonexistent/path/to/sqlite-vec.dylib");
        // The connection must remain usable for normal queries.
        insert_node(&conn, "after-fail", &sample_doc()).unwrap();
        let doc = get_node(&conn, "after-fail").unwrap().unwrap();
        assert_eq!(doc, sample_doc());
    }

    // ── Embedding write path (precomputed vector) ───────────────────

    fn count_embeddings(conn: &Connection, model: &str) -> i64 {
        conn.query_row(
            "SELECT COUNT(*) FROM embeddings WHERE model = ?1",
            params![model],
            |r| r.get(0),
        )
        .unwrap()
    }

    /// Criterion: storage-takes-precomputed-embedding.
    /// Verifies `insert_node_with` writes one embeddings row when both a
    /// vector and a model are supplied, and skips the row when either is
    /// absent.
    #[test]
    fn storage_precomputed_embedding_inserts_one_row_on_insert() {
        let conn = setup();
        insert_node_with(
            &conn,
            "n",
            &sample_doc(),
            Some(vec![1.0, 2.0, 3.0]),
            Some("test-model"),
        )
        .unwrap();
        assert_eq!(count_embeddings(&conn, "test-model"), 1);
        let blob: Vec<u8> = conn
            .query_row(
                "SELECT embedding FROM embeddings WHERE node_id = 'n' AND model = 'test-model'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        let decoded = embedding::decode_embedding(&blob).unwrap();
        assert_eq!(decoded, vec![1.0_f32, 2.0, 3.0]);
    }

    #[test]
    fn storage_precomputed_embedding_replaces_on_update() {
        let conn = setup();
        insert_node_with(
            &conn,
            "n",
            &sample_doc(),
            Some(vec![1.0, 0.0]),
            Some("test-model"),
        )
        .unwrap();
        let mut doc = sample_doc();
        doc.blocks.push(Block::Paragraph {
            inlines: vec![Inline::Plain("more".into())],
        });
        update_node_with(&conn, "n", &doc, Some(vec![0.0, 1.0]), Some("test-model")).unwrap();
        assert_eq!(count_embeddings(&conn, "test-model"), 1);
        let blob: Vec<u8> = conn
            .query_row(
                "SELECT embedding FROM embeddings WHERE node_id = 'n'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(
            embedding::decode_embedding(&blob).unwrap(),
            vec![0.0_f32, 1.0]
        );
    }

    #[test]
    fn storage_precomputed_embedding_skipped_without_vector() {
        let conn = setup();
        insert_node_with(&conn, "n", &sample_doc(), None, None).unwrap();
        update_node_with(&conn, "n", &sample_doc(), None, None).unwrap();
        assert_eq!(
            conn.query_row::<i64, _, _>("SELECT COUNT(*) FROM embeddings", [], |r| r.get(0))
                .unwrap(),
            0
        );
    }

    #[test]
    fn storage_precomputed_embedding_skipped_without_model() {
        // A vector with no model is treated as "no embedding" — both must
        // be Some for a row to be written.
        let conn = setup();
        insert_node_with(&conn, "n", &sample_doc(), Some(vec![1.0]), None).unwrap();
        assert_eq!(
            conn.query_row::<i64, _, _>("SELECT COUNT(*) FROM embeddings", [], |r| r.get(0))
                .unwrap(),
            0
        );
    }

    #[test]
    fn storage_precomputed_embedding_keys_by_node_and_model() {
        let conn = setup();
        insert_node_with(
            &conn,
            "n",
            &sample_doc(),
            Some(vec![1.0, 0.0]),
            Some("model-a"),
        )
        .unwrap();
        update_node_with(
            &conn,
            "n",
            &sample_doc(),
            Some(vec![0.0, 1.0]),
            Some("model-b"),
        )
        .unwrap();
        assert_eq!(count_embeddings(&conn, "model-a"), 1);
        assert_eq!(count_embeddings(&conn, "model-b"), 1);
    }

    // ── search_hybrid ────────────────────────────────────────────────

    #[test]
    fn search_hybrid_no_query_embedding_matches_search_fts() {
        let conn = setup();
        let doc1 = Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("Rust".into()),
                tags: vec![],
                children: vec![Block::Paragraph {
                    inlines: vec![Inline::Plain("systems language".into())],
                }],
            }],
        };
        let doc2 = Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("Python".into()),
                tags: vec![],
                children: vec![Block::Paragraph {
                    inlines: vec![Inline::Plain("scripting language".into())],
                }],
            }],
        };
        insert_node(&conn, "a", &doc1).unwrap();
        insert_node(&conn, "b", &doc2).unwrap();
        let fts: Vec<NodeId> = search_fts(&conn, "language")
            .unwrap()
            .into_iter()
            .map(NodeId)
            .collect();
        let hybrid = search_hybrid(&conn, "language", None).unwrap();
        assert_eq!(hybrid, fts);
    }

    #[test]
    fn search_hybrid_empty_query_yields_empty() {
        let conn = setup();
        insert_node(&conn, "a", &sample_doc()).unwrap();
        // No query embedding.
        assert!(search_hybrid(&conn, "", None).unwrap().is_empty());
        // With a query embedding and whitespace-only query AND empty
        // keyword list -> still empty.
        assert!(
            search_hybrid(&conn, "   ", Some((vec![1.0], "stub")))
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn search_hybrid_with_query_embedding_combines_keyword_and_vector_rankings() {
        let conn = setup();
        insert_node(
            &conn,
            "alpha",
            &Document {
                blocks: vec![Block::Heading {
                    level: 1,
                    title: Title("alpha note".into()),
                    tags: vec![],
                    children: vec![Block::Paragraph {
                        inlines: vec![Inline::Plain("the quick brown fox".into())],
                    }],
                }],
            },
        )
        .unwrap();
        insert_node(
            &conn,
            "beta",
            &Document {
                blocks: vec![Block::Heading {
                    level: 1,
                    title: Title("beta note".into()),
                    tags: vec![],
                    children: vec![Block::Paragraph {
                        inlines: vec![Inline::Plain("a different idea entirely".into())],
                    }],
                }],
            },
        )
        .unwrap();
        let alpha_vec: Vec<f32> = vec![1.0, 0.0];
        let beta_vec: Vec<f32> = vec![0.0, 1.0];
        for (id, v) in [("alpha", &alpha_vec), ("beta", &beta_vec)] {
            conn.execute(
                "INSERT INTO embeddings (node_id, model, embedding) VALUES (?1, 'hybrid-model', ?2)",
                params![id, embedding::encode_embedding(v)],
            )
            .unwrap();
        }
        // Caller-supplied query vector aligned with beta -> vector
        // ranking puts beta first.
        let out =
            search_hybrid(&conn, "alpha", Some((vec![0.0_f32, 1.0], "hybrid-model"))).unwrap();
        // FTS for "alpha" -> [alpha]; vector -> [beta, alpha].
        // RRF k=60:
        //   alpha: 1/61 + 1/62 ~ 0.0325
        //   beta:  1/61      ~ 0.0164
        // -> alpha first, beta second.
        assert_eq!(out, vec![NodeId("alpha".into()), NodeId("beta".into())]);
    }

    #[test]
    fn search_hybrid_with_query_embedding_returns_vector_only_when_fts_empty() {
        let conn = setup();
        insert_node(&conn, "a", &sample_doc()).unwrap();
        conn.execute(
            "INSERT INTO embeddings (node_id, model, embedding) VALUES ('a', 'm', ?1)",
            params![embedding::encode_embedding(&[1.0_f32])],
        )
        .unwrap();
        // Query "nomatch" — FTS returns nothing, but query is non-whitespace,
        // so vector ranking still runs.
        let out = search_hybrid(&conn, "nomatch", Some((vec![1.0_f32], "m"))).unwrap();
        assert_eq!(out, vec![NodeId("a".into())]);
    }

    // ── embedding_disabled (default behaviour with no precomputed embedding) ──

    #[test]
    fn embedding_disabled_writes_succeed() {
        let conn = setup();
        insert_node_with(&conn, "n", &sample_doc(), None, None).unwrap();
        update_node_with(&conn, "n", &sample_doc(), None, None).unwrap();
        delete_node(&conn, "n").unwrap();
        assert!(get_node(&conn, "n").unwrap().is_none());
    }

    #[test]
    fn embedding_disabled_keeps_embeddings_table_empty() {
        let conn = setup();
        insert_node(&conn, "a", &sample_doc()).unwrap();
        insert_node(&conn, "b", &sample_doc()).unwrap();
        let n: i64 = conn
            .query_row("SELECT COUNT(*) FROM embeddings", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 0);
    }

    #[test]
    fn embedding_disabled_search_hybrid_degrades_to_search_fts() {
        let conn = setup();
        let doc = Document {
            blocks: vec![Block::Heading {
                level: 1,
                title: Title("hello".into()),
                tags: vec![],
                children: vec![Block::Paragraph {
                    inlines: vec![Inline::Plain("world".into())],
                }],
            }],
        };
        insert_node(&conn, "a", &doc).unwrap();
        let fts: Vec<NodeId> = search_fts(&conn, "hello")
            .unwrap()
            .into_iter()
            .map(NodeId)
            .collect();
        let hybrid = search_hybrid(&conn, "hello", None).unwrap();
        assert_eq!(hybrid, fts);
    }

    // ── Unified links table: id-links + name-links coexist ────────────

    /// Build a document that has a `#+name:` slug and a body paragraph.
    fn named_doc(slug: &str, body: &str) -> Document {
        Document {
            blocks: vec![
                Block::Keyword {
                    name: "name".into(),
                    value: format!(" {slug}"),
                },
                Block::Paragraph {
                    inlines: vec![Inline::Plain(body.into())],
                },
            ],
        }
    }

    /// Build a document that references `[[slug]]` bracket names.
    fn referencing_doc(slugs: &[&str]) -> Document {
        let inlines: Vec<Inline> = slugs
            .iter()
            .map(|s| Inline::Link {
                target: (*s).to_string(),
                description: None,
            })
            .collect();
        Document {
            blocks: vec![Block::Paragraph { inlines }],
        }
    }

    fn count_links_total(conn: &Connection) -> i64 {
        conn.query_row("SELECT COUNT(*) FROM links", [], |r| r.get(0))
            .unwrap()
    }

    /// Fetch a single name-link row from the unified table.
    fn fetch_name_link(conn: &Connection, src_id: &str, slug: &str) -> Option<LinkRow> {
        conn.query_row(
            "SELECT source_id, link_type, target_id, target_slug FROM links \
             WHERE source_id = ?1 AND link_type = 'name' AND target_slug = ?2",
            params![src_id, slug],
            |r| {
                Ok(LinkRow {
                    source_id: r.get(0)?,
                    link_type: r.get(1)?,
                    target_id: r.get(2)?,
                    target_slug: r.get(3)?,
                })
            },
        )
        .optional()
        .unwrap()
    }

    /// Sentinel: the consolidated graph passes id+name semantics through
    /// the unified table. Aggregated under one criterion so the contract
    /// can pin a single filter per behavior.
    #[test]
    fn links_table_unified_schema() {
        let conn = setup();
        // Single physical link table, no name_links shadow.
        let n_links: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='links'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_links, 1);
        let n_name_links: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='name_links'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_name_links, 0, "name_links table must not exist");

        // No view named name_links either.
        let n_view: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='view' AND name='name_links'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_view, 0, "name_links view must not exist");

        // Columns and PK shape.
        let cols: Vec<(String, String, i64, i64)> = conn
            .prepare("PRAGMA table_info(links)")
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(1)?,
                    r.get::<_, String>(2)?,
                    r.get::<_, i64>(3)?,
                    r.get::<_, i64>(5)?,
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(
            cols,
            vec![
                ("source_id".into(), "TEXT".into(), 1, 1),
                ("link_type".into(), "TEXT".into(), 1, 2),
                ("target_id".into(), "TEXT".into(), 0, 3),
                ("target_slug".into(), "TEXT".into(), 0, 4),
            ]
        );

        // CHECK enforces the per-link_type column contract.
        insert_node(&conn, "src", &empty_doc()).unwrap();
        for sql in [
            // id-link with NULL target_id
            "INSERT INTO links (source_id, link_type, target_id, target_slug) \
             VALUES ('src', 'id', NULL, NULL)",
            // id-link with target_slug
            "INSERT INTO links (source_id, link_type, target_id, target_slug) \
             VALUES ('src', 'id', 'src', 'foo')",
            // name-link with NULL target_slug
            "INSERT INTO links (source_id, link_type, target_id, target_slug) \
             VALUES ('src', 'name', NULL, NULL)",
            // unknown link_type
            "INSERT INTO links (source_id, link_type, target_id, target_slug) \
             VALUES ('src', 'other', 'src', NULL)",
        ] {
            assert!(
                conn.execute(sql, []).is_err(),
                "expected CHECK rejection: {sql}"
            );
        }

        // Real writes coexist in the same table under different link_types.
        insert_node(&conn, "named", &named_doc("alpha", "x")).unwrap();
        let mut doc_with_both = referencing_doc(&["alpha"]);
        doc_with_both.blocks.push(Block::Paragraph {
            inlines: vec![Inline::Link {
                target: "id:src".into(),
                description: None,
            }],
        });
        insert_node(&conn, "mixed", &doc_with_both).unwrap();
        let rows: Vec<(String, String, Option<String>, Option<String>)> = conn
            .prepare(
                "SELECT source_id, link_type, target_id, target_slug FROM links \
                 WHERE source_id = 'mixed' ORDER BY link_type",
            )
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(0)?,
                    r.get::<_, String>(1)?,
                    r.get::<_, Option<String>>(2)?,
                    r.get::<_, Option<String>>(3)?,
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(
            rows,
            vec![
                ("mixed".into(), "id".into(), Some("src".into()), None),
                (
                    "mixed".into(),
                    "name".into(),
                    Some("named".into()),
                    Some("alpha".into())
                ),
            ]
        );
    }

    /// Aggregated id-link semantics under the unified schema.
    #[test]
    fn id_link_semantics_preserved() {
        // Extraction: `[[id:UUID]]` extraction matches pre-consolidation.
        let doc = link_doc(&["aaa", "bbb", "aaa"]);
        let ids: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Id(s) => Some(s),
                LinkRef::Name(_) => None,
            })
            .collect();
        assert_eq!(ids, vec!["aaa".to_string(), "bbb".to_string()]);

        // Storage: insert + neighborhood + relinker behavior on a small graph.
        let conn = setup();
        insert_node(&conn, "tgt", &empty_doc()).unwrap();
        insert_node(&conn, "src", &link_doc(&["tgt"])).unwrap();
        let row: (Option<String>, String, Option<String>) = conn
            .query_row(
                "SELECT target_id, link_type, target_slug FROM links WHERE source_id = 'src'",
                [],
                |r| {
                    Ok((
                        r.get::<_, Option<String>>(0)?,
                        r.get::<_, String>(1)?,
                        r.get::<_, Option<String>>(2)?,
                    ))
                },
            )
            .unwrap();
        assert_eq!(row, (Some("tgt".into()), "id".into(), None));

        // Forward refs to nonexistent targets are silently dropped.
        insert_node(&conn, "fwd", &link_doc(&["missing-tgt"])).unwrap();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM links WHERE source_id = 'fwd'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 0);

        // Neighborhood projects (NodeId, link_type).
        let n = get_neighborhood(&conn, "src").unwrap();
        assert_eq!(n.outgoing, vec![(NodeId("tgt".into()), LinkType::Id)]);
        let n = get_neighborhood(&conn, "tgt").unwrap();
        assert_eq!(n.incoming, vec![(NodeId("src".into()), LinkType::Id)]);

        // Update replaces outgoing.
        insert_node(&conn, "tgt2", &empty_doc()).unwrap();
        update_node(&conn, "src", &link_doc(&["tgt2"])).unwrap();
        let n = get_neighborhood(&conn, "src").unwrap();
        assert_eq!(n.outgoing, vec![(NodeId("tgt2".into()), LinkType::Id)]);

        // relink_all picks up forward references after the target appears.
        insert_node(&conn, "future-src", &link_doc(&["future-tgt"])).unwrap();
        insert_node(&conn, "future-tgt", &empty_doc()).unwrap();
        let (_, links) = relink_all(&conn).unwrap();
        assert!(links >= 1);
        let row: Option<String> = conn
            .query_row(
                "SELECT target_id FROM links \
                 WHERE source_id = 'future-src' AND link_type = 'id'",
                [],
                |r| r.get(0),
            )
            .optional()
            .unwrap()
            .flatten();
        assert_eq!(row.as_deref(), Some("future-tgt"));

        // Id-link rows pointing at a deleted target are removed, NOT
        // demoted to broken (only name-link rows demote).
        delete_node(&conn, "future-tgt").unwrap();
        let n: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM links \
                 WHERE source_id = 'future-src' AND link_type = 'id'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n, 0);
    }

    /// Name-link broken + reverse-resolve behavior under the unified
    /// schema. Aggregated under one criterion.
    #[test]
    fn name_link_broken_and_reverse_resolve_preserved() {
        let conn = setup();

        // Broken on write when no slug match exists.
        insert_node(&conn, "src1", &referencing_doc(&["future"])).unwrap();
        let row = fetch_name_link(&conn, "src1", "future").unwrap();
        assert_eq!(row.link_type, LinkType::Name);
        assert_eq!(row.target_slug.as_deref(), Some("future"));
        assert!(row.target_id.is_none());

        // Resolved on write when a matching slug already exists.
        insert_node(&conn, "named", &named_doc("alpha", "I am alpha")).unwrap();
        insert_node(&conn, "src2", &referencing_doc(&["alpha"])).unwrap();
        let row = fetch_name_link(&conn, "src2", "alpha").unwrap();
        assert_eq!(row.target_id.as_deref(), Some("named"));

        // Back-resolve when the matching target is created later.
        insert_node(&conn, "src3", &referencing_doc(&["beta"])).unwrap();
        assert!(
            fetch_name_link(&conn, "src3", "beta")
                .unwrap()
                .target_id
                .is_none()
        );
        insert_node(&conn, "beta-target", &named_doc("beta", "x")).unwrap();
        let row = fetch_name_link(&conn, "src3", "beta").unwrap();
        assert_eq!(row.target_id.as_deref(), Some("beta-target"));

        // Back-resolve when an existing target is updated to carry the slug.
        insert_node(&conn, "src4", &referencing_doc(&["gamma"])).unwrap();
        insert_node(
            &conn,
            "gamma-target",
            &Document {
                blocks: vec![Block::Paragraph {
                    inlines: vec![Inline::Plain("unnamed".into())],
                }],
            },
        )
        .unwrap();
        assert!(
            fetch_name_link(&conn, "src4", "gamma")
                .unwrap()
                .target_id
                .is_none()
        );
        update_node(&conn, "gamma-target", &named_doc("gamma", "x")).unwrap();
        let row = fetch_name_link(&conn, "src4", "gamma").unwrap();
        assert_eq!(row.target_id.as_deref(), Some("gamma-target"));

        // Update replaces prior outgoing name-link rows.
        update_node(&conn, "src2", &referencing_doc(&["beta"])).unwrap();
        assert!(fetch_name_link(&conn, "src2", "alpha").is_none());
        let row = fetch_name_link(&conn, "src2", "beta").unwrap();
        assert_eq!(row.target_id.as_deref(), Some("beta-target"));

        // list_broken_links surfaces only unresolved name-link rows.
        insert_node(&conn, "src5", &referencing_doc(&["missing-1", "missing-2"])).unwrap();
        let broken = list_broken_links(&conn).unwrap();
        let slugs: Vec<&str> = broken
            .iter()
            .map(|r| r.target_slug.as_deref().unwrap_or(""))
            .collect();
        assert!(slugs.contains(&"missing-1"));
        assert!(slugs.contains(&"missing-2"));
        for row in &broken {
            assert_eq!(row.link_type, LinkType::Name);
            assert!(row.target_id.is_none());
        }
    }

    /// Demote-on-delete and demote-on-slug-change semantics under the
    /// unified schema.
    #[test]
    fn name_link_demote_preserved() {
        let conn = setup();
        insert_node(&conn, "target", &named_doc("alpha", "x")).unwrap();
        insert_node(&conn, "source", &referencing_doc(&["alpha"])).unwrap();
        assert_eq!(
            fetch_name_link(&conn, "source", "alpha")
                .unwrap()
                .target_id
                .as_deref(),
            Some("target")
        );

        // Slug change demotes resolved rows back to broken.
        update_node(&conn, "target", &named_doc("renamed", "x")).unwrap();
        let row = fetch_name_link(&conn, "source", "alpha").unwrap();
        assert!(row.target_id.is_none(), "slug change must demote to broken");
        assert_eq!(row.target_slug.as_deref(), Some("alpha"));

        // Restore + delete: the row resolves back, then the delete
        // demotes it once more via the ON DELETE SET NULL FK.
        update_node(&conn, "target", &named_doc("alpha", "x")).unwrap();
        assert_eq!(
            fetch_name_link(&conn, "source", "alpha")
                .unwrap()
                .target_id
                .as_deref(),
            Some("target")
        );
        delete_node(&conn, "target").unwrap();
        let row = fetch_name_link(&conn, "source", "alpha").unwrap();
        assert!(row.target_id.is_none(), "delete must demote to broken");
        assert_eq!(row.target_slug.as_deref(), Some("alpha"));

        // Source delete cascades the entire source's outgoing rows.
        delete_node(&conn, "source").unwrap();
        assert_eq!(count_links_total(&conn), 0);
    }

    /// Migration from the pre-consolidation split shape (kb-link-index-v0).
    #[test]
    fn migration_from_split_tables() {
        // Build a fresh database, then *replace* the unified schema
        // with the legacy split shape and populate it with rows of all
        // three shapes (id-link, resolved name-link, broken name-link).
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch("PRAGMA foreign_keys = ON;").unwrap();
        conn.execute(NODES_TABLE, []).unwrap();
        conn.execute(NODE_TAGS_TABLE, []).unwrap();
        // Legacy split shape (verbatim from kb-link-index-v0).
        conn.execute(
            "CREATE TABLE links ( \
               source_id TEXT NOT NULL, \
               target_id TEXT NOT NULL, \
               link_type TEXT NOT NULL, \
               PRIMARY KEY (source_id, target_id, link_type), \
               FOREIGN KEY (source_id) REFERENCES nodes(id) ON DELETE CASCADE, \
               FOREIGN KEY (target_id) REFERENCES nodes(id) ON DELETE CASCADE)",
            [],
        )
        .unwrap();
        conn.execute(
            "CREATE TABLE name_links ( \
               src_id   TEXT NOT NULL, \
               dst_slug TEXT NOT NULL, \
               dst_id   TEXT, \
               PRIMARY KEY (src_id, dst_slug), \
               FOREIGN KEY (src_id) REFERENCES nodes(id) ON DELETE CASCADE)",
            [],
        )
        .unwrap();
        // Insert nodes.
        for nid in ["a", "b", "c", "d"] {
            conn.execute(
                "INSERT INTO nodes (id, title, ast_blob, created_at, updated_at) \
                 VALUES (?1, 't', ?2, '0', '0')",
                params![nid, sexp::encode_document(&Document { blocks: vec![] })],
            )
            .unwrap();
        }
        // Legacy id-link row.
        conn.execute(
            "INSERT INTO links (source_id, target_id, link_type) VALUES ('a', 'b', 'id')",
            [],
        )
        .unwrap();
        // Legacy resolved name-link row.
        conn.execute(
            "INSERT INTO name_links (src_id, dst_slug, dst_id) VALUES ('c', 'beta', 'b')",
            [],
        )
        .unwrap();
        // Legacy broken name-link row.
        conn.execute(
            "INSERT INTO name_links (src_id, dst_slug, dst_id) VALUES ('d', 'ghost', NULL)",
            [],
        )
        .unwrap();

        // Run the migration via init_db (the supported entry point).
        init_db(&conn).unwrap();

        // The legacy tables are gone; only the unified `links` table remains.
        let n_name_links: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='name_links'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_name_links, 0);

        // All three rows survive in the unified shape.
        let mut rows: Vec<(String, String, Option<String>, Option<String>)> = conn
            .prepare(
                "SELECT source_id, link_type, target_id, target_slug FROM links \
                 ORDER BY source_id",
            )
            .unwrap()
            .query_map([], |r| {
                Ok((
                    r.get::<_, String>(0)?,
                    r.get::<_, String>(1)?,
                    r.get::<_, Option<String>>(2)?,
                    r.get::<_, Option<String>>(3)?,
                ))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        rows.sort();
        assert_eq!(
            rows,
            vec![
                ("a".into(), "id".into(), Some("b".into()), None),
                (
                    "c".into(),
                    "name".into(),
                    Some("b".into()),
                    Some("beta".into())
                ),
                ("d".into(), "name".into(), None, Some("ghost".into())),
            ]
        );

        // Migration is idempotent — a second init_db is a no-op.
        init_db(&conn).unwrap();
        let n: i64 = count_links_total(&conn);
        assert_eq!(n, 3);
    }

    #[test]
    fn link_parser_extracts_bracket_name_refs() {
        let doc = Document {
            blocks: vec![Block::Paragraph {
                inlines: vec![
                    Inline::Plain("see ".into()),
                    Inline::Link {
                        target: "alpha".into(),
                        description: None,
                    },
                    Inline::Plain(" and ".into()),
                    Inline::Link {
                        target: "beta".into(),
                        description: None,
                    },
                ],
            }],
        };
        let slugs: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Name(s) => Some(s),
                LinkRef::Id(_) => None,
            })
            .collect();
        assert_eq!(slugs, vec!["alpha".to_string(), "beta".to_string()]);
    }

    #[test]
    fn link_parser_extracts_dedupes_preserving_first_seen_order() {
        let doc = referencing_doc(&["alpha", "beta", "alpha", "gamma"]);
        let slugs: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Name(s) => Some(s),
                LinkRef::Id(_) => None,
            })
            .collect();
        assert_eq!(slugs, vec!["alpha", "beta", "gamma"]);
    }

    #[test]
    fn link_parser_extracts_walks_nested_inline_positions() {
        let doc = Document {
            blocks: vec![Block::Paragraph {
                inlines: vec![Inline::Bold(vec![Inline::Italic(vec![Inline::Link {
                    target: "nested".into(),
                    description: None,
                }])])],
            }],
        };
        let slugs: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Name(s) => Some(s),
                LinkRef::Id(_) => None,
            })
            .collect();
        assert_eq!(slugs, vec!["nested".to_string()]);
    }

    #[test]
    fn link_parser_extracts_ignores_scheme_links() {
        let doc = Document {
            blocks: vec![Block::Paragraph {
                inlines: vec![
                    Inline::Link {
                        target: "id:abc-123".into(),
                        description: None,
                    },
                    Inline::Link {
                        target: "https://example.com".into(),
                        description: Some("ex".into()),
                    },
                    Inline::Link {
                        target: "keep".into(),
                        description: None,
                    },
                    Inline::Link {
                        target: "alpha".into(),
                        description: Some("Alpha".into()),
                    },
                ],
            }],
        };
        let slugs: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Name(s) => Some(s),
                LinkRef::Id(_) => None,
            })
            .collect();
        assert_eq!(slugs, vec!["keep".to_string()]);
    }

    #[test]
    fn parser_respects_literal_regions_for_bracket_refs() {
        let doc = Document {
            blocks: vec![
                Block::Paragraph {
                    inlines: vec![
                        Inline::Plain("Use ".into()),
                        Inline::Verbatim("[[verbatim-target]]".into()),
                        Inline::Plain(" for verbatim, ".into()),
                        Inline::InlineCode("[[code-target]]".into()),
                        Inline::Plain(" for inline code.".into()),
                    ],
                },
                Block::SrcBlock {
                    language: "org".into(),
                    content: "Example: [[src-block-target]]\n".into(),
                },
                Block::ExampleBlock {
                    content: "Example: [[example-block-target]]\n".into(),
                },
                Block::Paragraph {
                    inlines: vec![Inline::Link {
                        target: "real-target".into(),
                        description: None,
                    }],
                },
            ],
        };
        let slugs: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Name(s) => Some(s),
                LinkRef::Id(_) => None,
            })
            .collect();
        assert_eq!(slugs, vec!["real-target".to_string()]);
    }

    #[test]
    fn parser_respects_literal_regions_round_trip_via_org_parser() {
        let org = "Use =[[verbatim-target]]= for verbatim, \
                   ~[[code-target]]~ for inline code.\n\
                   #+begin_src org\n\
                   Example: [[src-block-target]]\n\
                   #+end_src\n\
                   #+begin_example\n\
                   Example: [[example-block-target]]\n\
                   #+end_example\n\
                   [[real-target]]\n";
        let doc = crate::parser::parse_document(org).unwrap();
        let slugs: Vec<String> = extract_links(&doc)
            .into_iter()
            .filter_map(|l| match l {
                LinkRef::Name(s) => Some(s),
                LinkRef::Id(_) => None,
            })
            .collect();
        assert_eq!(slugs, vec!["real-target".to_string()]);
    }

    /// The single body walker emits both link types from one pass.
    #[test]
    fn unified_walker_emits_both_link_types_in_one_pass() {
        let doc = Document {
            blocks: vec![Block::Paragraph {
                inlines: vec![
                    Inline::Link {
                        target: "id:abc".into(),
                        description: None,
                    },
                    Inline::Link {
                        target: "alpha".into(),
                        description: None,
                    },
                    Inline::Link {
                        target: "id:def".into(),
                        description: None,
                    },
                    Inline::Link {
                        target: "beta".into(),
                        description: None,
                    },
                ],
            }],
        };
        let refs = extract_links(&doc);
        assert_eq!(
            refs,
            vec![
                LinkRef::Id("abc".into()),
                LinkRef::Name("alpha".into()),
                LinkRef::Id("def".into()),
                LinkRef::Name("beta".into()),
            ]
        );
    }

    #[test]
    fn list_orphans_counts_across_both_link_types() {
        let conn = setup();
        // `hub` named, referenced by `src-name`.
        insert_node(&conn, "hub", &named_doc("hub", "x")).unwrap();
        insert_node(&conn, "src-name", &referencing_doc(&["hub"])).unwrap();
        // `id-hub` referenced by an id-link from `src-id`.
        insert_node(&conn, "id-hub", &empty_doc()).unwrap();
        insert_node(&conn, "src-id", &link_doc(&["id-hub"])).unwrap();
        // `lonely` carries no incoming refs.
        insert_node(
            &conn,
            "lonely",
            &Document {
                blocks: vec![Block::Paragraph {
                    inlines: vec![Inline::Plain("no one links to me".into())],
                }],
            },
        )
        .unwrap();
        let orphans = list_orphans(&conn).unwrap();
        let ids: Vec<&str> = orphans.iter().map(|r| r.id.as_str()).collect();
        // Targets are NOT orphans regardless of link_type; everyone
        // else is.
        assert!(!ids.contains(&"hub"));
        assert!(!ids.contains(&"id-hub"));
        assert!(ids.contains(&"src-name"));
        assert!(ids.contains(&"src-id"));
        assert!(ids.contains(&"lonely"));
    }

    #[test]
    fn list_hubs_ranks_by_total_in_degree_across_link_types() {
        let conn = setup();
        insert_node(&conn, "hot", &named_doc("hot", "x")).unwrap();
        insert_node(&conn, "warm", &empty_doc()).unwrap();
        // hot picks up two name-link incoming.
        insert_node(&conn, "a", &referencing_doc(&["hot"])).unwrap();
        insert_node(&conn, "b", &referencing_doc(&["hot"])).unwrap();
        // hot also picks up one id-link incoming -> total in-degree 3.
        insert_node(&conn, "c", &link_doc(&["hot"])).unwrap();
        // warm picks up one id-link incoming.
        insert_node(&conn, "d", &link_doc(&["warm"])).unwrap();
        let hubs = list_hubs(&conn, 10).unwrap();
        assert_eq!(hubs[0].id, "hot");
        assert_eq!(hubs[0].in_degree, 3);
        assert_eq!(hubs[1].id, "warm");
        assert_eq!(hubs[1].in_degree, 1);
    }

    #[test]
    fn list_hubs_zero_limit_returns_empty() {
        let conn = setup();
        insert_node(&conn, "hot", &named_doc("hot", "x")).unwrap();
        insert_node(&conn, "a", &referencing_doc(&["hot"])).unwrap();
        assert!(list_hubs(&conn, 0).unwrap().is_empty());
    }

    #[test]
    fn list_broken_links_returns_only_unresolved_name_rows() {
        let conn = setup();
        insert_node(&conn, "named", &named_doc("alpha", "x")).unwrap();
        insert_node(&conn, "src1", &referencing_doc(&["alpha", "missing"])).unwrap();
        insert_node(&conn, "src2", &referencing_doc(&["also-missing"])).unwrap();
        let broken = list_broken_links(&conn).unwrap();
        let slugs: Vec<&str> = broken
            .iter()
            .map(|r| r.target_slug.as_deref().unwrap_or(""))
            .collect();
        assert_eq!(slugs, vec!["also-missing", "missing"]);
        for row in &broken {
            assert_eq!(row.link_type, LinkType::Name);
            assert!(row.target_id.is_none());
        }
    }

    #[test]
    fn get_links_returns_full_outgoing_including_broken() {
        let conn = setup();
        insert_node(&conn, "hub", &named_doc("hub", "x")).unwrap();
        insert_node(&conn, "src", &referencing_doc(&["hub", "ghost"])).unwrap();
        let nb = get_links(&conn, "src").unwrap();
        // Two outgoing rows; one resolved, one broken.
        assert_eq!(nb.outgoing.len(), 2);
        let resolved = nb
            .outgoing
            .iter()
            .find(|r| r.target_slug.as_deref() == Some("hub"))
            .unwrap();
        assert_eq!(resolved.target_id.as_deref(), Some("hub"));
        let broken = nb
            .outgoing
            .iter()
            .find(|r| r.target_slug.as_deref() == Some("ghost"))
            .unwrap();
        assert!(broken.target_id.is_none());
    }

    #[test]
    fn relink_all_skips_corrupt_blob_in_unified_path() {
        let conn = setup();
        insert_node(&conn, "good", &named_doc("g", "x")).unwrap();
        conn.execute(
            "INSERT INTO nodes (id, title, ast_blob, created_at, updated_at) \
             VALUES ('bad', 'x', 'not-a-sexp', '0', '0')",
            [],
        )
        .unwrap();
        let (nodes, _) = relink_all(&conn).unwrap();
        assert_eq!(nodes, 2);
    }
}