grit-core 0.2.4

//! Forward-only schema migrations via `PRAGMA user_version`
//! (Design Invariant 8).

use rusqlite::Connection;

use crate::error::{Error, Result};

/// The schema version this build reads and writes.
pub const SCHEMA_VERSION: i64 = 5;

/// Creates the LATEST schema from scratch (schema.sql tracks head).
const SCHEMA_LATEST: &str = include_str!("schema.sql");

/// v1 → v2 (grit 0.2.0): per-field LWW records backing `UpdateNode`.
/// Frozen — never edit; schema.sql carries the same DDL for fresh DBs.
const MIGRATE_1_TO_2: &str = "
CREATE TABLE node_updates (
    node_id TEXT NOT NULL,
    field   TEXT NOT NULL,
    value   TEXT NOT NULL,
    hlc     TEXT NOT NULL,
    PRIMARY KEY (node_id, field)
) STRICT, WITHOUT ROWID;
";

/// v2 → v3 (grit 0.2.2): `episodes.kind` — a free-form source-kind tag
/// (`"message"` / `"text"` / `"json"`), opaque to grit, empty for
/// pre-existing rows. Frozen — never edit.
const MIGRATE_2_TO_3: &str = "
ALTER TABLE episodes ADD COLUMN kind TEXT NOT NULL DEFAULT '';
";

/// v3 → v4 (grit 0.2.3): trigram FTS mirrors for CJK substring recall —
/// tables + triggers identical to schema.sql, then a rebuild populates
/// them from existing rows (external-content FTS reads the base tables).
/// Frozen — never edit.
const MIGRATE_3_TO_4: &str = "
CREATE VIRTUAL TABLE nodes_fts_tri USING fts5(
    name, summary,
    content='nodes', content_rowid='rowid',
    tokenize='trigram'
);
CREATE TRIGGER nodes_fts_tri_ai AFTER INSERT ON nodes BEGIN
    INSERT INTO nodes_fts_tri(rowid, name, summary)
    VALUES (new.rowid, new.name, new.summary);
END;
CREATE TRIGGER nodes_fts_tri_ad AFTER DELETE ON nodes BEGIN
    INSERT INTO nodes_fts_tri(nodes_fts_tri, rowid, name, summary)
    VALUES ('delete', old.rowid, old.name, old.summary);
END;
CREATE TRIGGER nodes_fts_tri_au AFTER UPDATE OF name, summary ON nodes BEGIN
    INSERT INTO nodes_fts_tri(nodes_fts_tri, rowid, name, summary)
    VALUES ('delete', old.rowid, old.name, old.summary);
    INSERT INTO nodes_fts_tri(rowid, name, summary)
    VALUES (new.rowid, new.name, new.summary);
END;
CREATE VIRTUAL TABLE edges_fts_tri USING fts5(
    fact,
    content='edges', content_rowid='rowid',
    tokenize='trigram'
);
CREATE TRIGGER edges_fts_tri_ai AFTER INSERT ON edges BEGIN
    INSERT INTO edges_fts_tri(rowid, fact) VALUES (new.rowid, new.fact);
END;
CREATE TRIGGER edges_fts_tri_ad AFTER DELETE ON edges BEGIN
    INSERT INTO edges_fts_tri(edges_fts_tri, rowid, fact)
    VALUES ('delete', old.rowid, old.fact);
END;
CREATE TRIGGER edges_fts_tri_au AFTER UPDATE OF fact ON edges BEGIN
    INSERT INTO edges_fts_tri(edges_fts_tri, rowid, fact)
    VALUES ('delete', old.rowid, old.fact);
    INSERT INTO edges_fts_tri(rowid, fact) VALUES (new.rowid, new.fact);
END;
CREATE VIRTUAL TABLE episodes_fts_tri USING fts5(
    content,
    content='episodes', content_rowid='rowid',
    tokenize='trigram'
);
CREATE TRIGGER episodes_fts_tri_ai AFTER INSERT ON episodes BEGIN
    INSERT INTO episodes_fts_tri(rowid, content) VALUES (new.rowid, new.content);
END;
CREATE TRIGGER episodes_fts_tri_ad AFTER DELETE ON episodes BEGIN
    INSERT INTO episodes_fts_tri(episodes_fts_tri, rowid, content)
    VALUES ('delete', old.rowid, old.content);
END;
CREATE TRIGGER episodes_fts_tri_au AFTER UPDATE OF content ON episodes BEGIN
    INSERT INTO episodes_fts_tri(episodes_fts_tri, rowid, content)
    VALUES ('delete', old.rowid, old.content);
    INSERT INTO episodes_fts_tri(rowid, content) VALUES (new.rowid, new.content);
END;
INSERT INTO nodes_fts_tri(nodes_fts_tri) VALUES ('rebuild');
INSERT INTO edges_fts_tri(edges_fts_tri) VALUES ('rebuild');
INSERT INTO episodes_fts_tri(episodes_fts_tri) VALUES ('rebuild');
";

/// v4 → v5 (grit 0.2.4): `vec_nodes`/`vec_edges` gain a `group_id`
/// PARTITION KEY so the KNN legs filter by group inside the scan instead of
/// post-filtering a global top-k (a recall bug for small groups). The vec
/// tables are dimension-dynamic and lazily created, so this step is Rust
/// driving clearly-delimited SQL rather than a frozen constant: each
/// existing table is rebuilt through a plain stash table, with `group_id`
/// joined in from the base graph table. Vector rows whose base row is gone
/// are dropped by the join — they were already unreachable (every read
/// resolves through the base row). Frozen — never edit.
fn migrate_4_to_5(conn: &mut Connection) -> Result<()> {
    let tx = conn.transaction()?;
    let dim: Option<i64> = tx
        .query_row("SELECT dim FROM embedding_meta WHERE id = 1", [], |r| {
            r.get(0)
        })
        .map(Some)
        .or_else(|e| match e {
            rusqlite::Error::QueryReturnedNoRows => Ok(None),
            other => Err(other),
        })?;
    for (vec_table, base) in [("vec_nodes", "nodes"), ("vec_edges", "edges")] {
        let exists: i64 = tx.query_row(
            "SELECT COUNT(*) FROM sqlite_master WHERE name = ?1",
            rusqlite::params![vec_table],
            |r| r.get(0),
        )?;
        if exists == 0 {
            continue;
        }
        // A vec table only ever exists alongside its embedding_meta row
        // (register_model writes both in one transaction); a file violating
        // that is left untouched rather than guessed at.
        let Some(dim) = dim else { continue };
        tx.execute_batch(&format!(
            "CREATE TABLE vec_migrate_stash (
                 id       TEXT PRIMARY KEY,
                 group_id TEXT NOT NULL,
                 embedding BLOB NOT NULL
             ) STRICT;
             INSERT INTO vec_migrate_stash (id, group_id, embedding)
                 SELECT v.id, b.group_id, v.embedding
                 FROM {vec_table} AS v JOIN {base} AS b ON b.id = v.id;
             DROP TABLE {vec_table};
             CREATE VIRTUAL TABLE {vec_table} USING vec0(
                 id TEXT PRIMARY KEY, group_id TEXT PARTITION KEY,
                 embedding FLOAT[{dim}] distance_metric=cosine);
             INSERT INTO {vec_table} (id, group_id, embedding)
                 SELECT id, group_id, embedding FROM vec_migrate_stash;
             DROP TABLE vec_migrate_stash;"
        ))?;
    }
    tx.pragma_update(None, "user_version", 5)?;
    tx.commit()?;
    Ok(())
}

/// Bring `conn`'s schema up to [`SCHEMA_VERSION`]. A brand-new database gets
/// the full latest schema; older versions are migrated step by step; a newer
/// version is refused (migrations are forward-only, never down).
pub(crate) fn migrate(conn: &mut Connection) -> Result<()> {
    loop {
        let version: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
        match version {
            SCHEMA_VERSION => return Ok(()),
            v if v > SCHEMA_VERSION => {
                return Err(Error::SchemaTooNew {
                    found: v,
                    supported: SCHEMA_VERSION,
                });
            }
            0 => apply_step(conn, SCHEMA_LATEST, SCHEMA_VERSION)?,
            1 => apply_step(conn, MIGRATE_1_TO_2, 2)?,
            2 => apply_step(conn, MIGRATE_2_TO_3, 3)?,
            3 => apply_step(conn, MIGRATE_3_TO_4, 4)?,
            4 => migrate_4_to_5(conn)?,
            v => unreachable!("no migration path from version {v}"),
        }
    }
}

fn apply_step(conn: &mut Connection, sql: &str, target: i64) -> Result<()> {
    let tx = conn.transaction()?;
    tx.execute_batch(sql)?;
    // PRAGMA can't take bound parameters; target is a compile-time constant.
    tx.pragma_update(None, "user_version", target)?;
    tx.commit()?;
    Ok(())
}