solo-storage 0.11.5

// SPDX-License-Identifier: Apache-2.0

//! SQL schema migrations. Runs once at startup against the SQLCipher database
//! after `PRAGMA key` has been bound.
//!
//! Migrations are append-only — once a version has shipped to a user, never
//! change its SQL. Bug fixes go in subsequent migrations.
//!
//! The runner advances the version tracker row-by-row inside a single
//! `BEGIN IMMEDIATE` transaction per migration, so a crash mid-migration
//! either applies the whole thing or none of it.
//!
//! ## Two schema chains
//!
//! v0.8.0 introduced a second SQLCipher database at
//! `<data_dir>/tenants_index.db` for the multi-tenant registry, with its
//! own independent migration chain tracked in
//! `schema_migrations_tenants_index`. The per-tenant DBs at
//! `<data_dir>/tenants/<id>.db` still use the original `schema_migrations`
//! table.
//!
//! * Per-tenant chain  →  [`run_migrations`] (still 1-3 today; what every
//!   existing callsite means when it says "run migrations").
//! * tenants_index chain → [`run_tenants_index_migrations`] (just 0004 today).
//!
//! Both share the same `BEGIN IMMEDIATE` / step + row insert mechanics via
//! the private [`apply_one`] helper.

use rusqlite::{Connection, TransactionBehavior, params};
use solo_core::{Error, Result};

/// One migration step. The `up` SQL may contain multiple statements (it's
/// passed to `execute_batch`).
#[derive(Debug)]
struct Migration {
    version: u32,
    description: &'static str,
    up: &'static str,
}

/// Per-tenant DB migrations, in order. Append new entries; never modify
/// existing ones.
///
/// As of v0.8.1 P1 the highest per-tenant version is **7** — migration 4
/// only lives in tenants_index.db (the multi-tenant registry, P1); P4
/// added migration 5 (per-tenant audit_events table); P5+P6 added
/// migration 6 (principal-attribution columns on episodes and
/// document_chunks for redaction telemetry + GDPR forget_user filtering);
/// v0.8.1 P1 added migration 7 (`triples.source_episode_id` FK + best-
/// effort backfill from `provenance_json` so GDPR `forget` cascades
/// through derived triples).
/// Numbering is monotonic across both chains for easy git blame.
const MIGRATIONS: &[Migration] = &[
    Migration {
        version: 1,
        description: "initial schema (v0): episodes + triples + steward outputs + pending_index + FTS",
        up: include_str!("migrations/0001_initial.sql"),
    },
    Migration {
        version: 2,
        description: "triples.cluster_id FK + index for absorb→regen cascade",
        up: include_str!("migrations/0002_triples_cluster_id.sql"),
    },
    Migration {
        version: 3,
        description: "documents + document_chunks + chunk_embeddings + pending_index.kind discriminator",
        up: include_str!("migrations/0003_documents.sql"),
    },
    Migration {
        version: 5,
        description: "per-tenant audit_events table + indices (v0.8.0 P4)",
        up: include_str!("migrations/0005_audit.sql"),
    },
    Migration {
        version: 6,
        description: "principal_subject on episodes + ingested_by_principal on document_chunks (v0.8.0 P5+P6)",
        up: include_str!("migrations/0006_redaction_audit.sql"),
    },
    Migration {
        version: 7,
        description: "triples.source_episode_id FK + backfill from provenance_json (v0.8.1 P1)",
        up: include_str!("migrations/0007_triples_source.sql"),
    },
    Migration {
        version: 10,
        description: "contradiction lifecycle fields for resolve/reopen flows",
        up: include_str!("migrations/0010_contradiction_lifecycle.sql"),
    },
];

/// tenants_index.db migrations, in order.
///
/// Numbering shares space with per-tenant migrations (4 follows 3) to keep
/// the on-disk migration file numbering monotonic across the whole
/// codebase — easier to git blame, easier to scan. The runner uses a
/// SEPARATE tracker table so the chains evolve independently.
///
/// As of v0.9.0 P1 the tenants_index chain is **4 → 8 → 9** — migration
/// 0008 (v0.8.1 P3) added `tenants.quota_bytes`; migration 0009 (v0.9.0
/// P1) adds `tenants.last_accessed` and closes the v0.8.0 doc-vs-code
/// gap (lesson #39 second incident — `last_accessed` was referenced in
/// v0.8.0 release notes but never landed in 0004).
const TENANTS_INDEX_MIGRATIONS: &[Migration] = &[
    Migration {
        version: 4,
        description: "tenants_index registry (per-tenant DB layout)",
        up: include_str!("migrations/0004_tenants.sql"),
    },
    Migration {
        version: 8,
        description: "tenants.quota_bytes column for v0.8.1 P3 enforcement",
        up: include_str!("migrations/0008_tenants_quota.sql"),
    },
    Migration {
        version: 9,
        description: "tenants.last_accessed column for v0.9.0 P1 (closes v0.8.0 doc-vs-code gap)",
        up: include_str!("migrations/0009_tenants_last_accessed.sql"),
    },
];

/// Name of the version tracker table inside per-tenant DBs.
const PER_TENANT_TRACKER: &str = "schema_migrations";

/// Name of the version tracker table inside tenants_index.db.
const TENANTS_INDEX_TRACKER: &str = "schema_migrations_tenants_index";

/// Run every per-tenant migration that hasn't been applied yet.
///
/// Idempotent — calling on an up-to-date per-tenant DB is a no-op + ~1ms read
/// of `schema_migrations`. Returns the highest version applied (after the run).
pub fn run_migrations(conn: &mut Connection) -> Result<u32> {
    run_migrations_for_connection(conn, MIGRATIONS, PER_TENANT_TRACKER)
}

/// Run every tenants_index.db migration that hasn't been applied yet.
///
/// Same mechanics as [`run_migrations`] but targets the parallel tracker
/// table. Idempotent. Returns the highest version applied.
pub fn run_tenants_index_migrations(conn: &mut Connection) -> Result<u32> {
    run_migrations_for_connection(conn, TENANTS_INDEX_MIGRATIONS, TENANTS_INDEX_TRACKER)
}

/// Generic migration runner.
///
/// Takes a connection, the migration list to apply, and the name of the
/// tracker table to record progress in. The tracker table is created
/// on-demand with the canonical (version, description, applied_at) schema
/// if it doesn't exist yet — this is how a brand-new DB picks up its
/// first migration without requiring out-of-band table creation.
///
/// Each migration runs inside its own `BEGIN IMMEDIATE` transaction, so a
/// crash mid-migration either applies the whole thing (SQL + tracker row)
/// or none of it.
fn run_migrations_for_connection(
    conn: &mut Connection,
    list: &[Migration],
    tracker_table: &str,
) -> Result<u32> {
    // tracker table is created out-of-band so the first migration doesn't
    // have to bootstrap its own tracking row. CREATE IF NOT EXISTS makes
    // this safe to call before checking existing state.
    let create_sql = format!(
        "CREATE TABLE IF NOT EXISTS {tracker_table} (
             version     INTEGER PRIMARY KEY,
             description TEXT    NOT NULL,
             applied_at  INTEGER NOT NULL
         );"
    );
    conn.execute_batch(&create_sql)
        .map_err(|e| Error::storage(format!("create {tracker_table}: {e}")))?;

    let current = current_version_in(conn, tracker_table)?;
    let mut highest = current;

    for m in list {
        if m.version <= current {
            continue;
        }
        apply_one(conn, m, tracker_table)?;
        highest = m.version;
        tracing::info!(
            version = m.version,
            description = m.description,
            tracker = tracker_table,
            "applied migration"
        );
    }

    Ok(highest)
}

/// Highest applied version in the per-tenant `schema_migrations` tracker,
/// or 0 if nothing has been applied yet.
///
/// Kept for back-compat with existing callsites that probe the per-tenant
/// chain (e.g. `startup.rs`). The tenants_index equivalent uses
/// [`current_tenants_index_version`].
pub fn current_version(conn: &Connection) -> Result<u32> {
    current_version_in(conn, PER_TENANT_TRACKER)
}

/// Highest applied version in the tenants_index.db tracker, or 0 if the
/// table doesn't exist / has no rows yet. Used by callers that need to
/// decide whether migration 0004 is already done.
pub fn current_tenants_index_version(conn: &Connection) -> Result<u32> {
    // If the tracker table doesn't exist yet (fresh tenants_index.db
    // before its first migration), treat that as version 0 rather than
    // surfacing the SQL error.
    let table_present: bool = conn
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?",
            params![TENANTS_INDEX_TRACKER],
            |row| row.get::<_, i64>(0).map(|n| n > 0),
        )
        .map_err(|e| Error::storage(format!("check {TENANTS_INDEX_TRACKER} presence: {e}")))?;
    if !table_present {
        return Ok(0);
    }
    current_version_in(conn, TENANTS_INDEX_TRACKER)
}

fn current_version_in(conn: &Connection, tracker_table: &str) -> Result<u32> {
    let sql = format!("SELECT MAX(version) FROM {tracker_table}");
    let v: Option<u32> = conn
        .query_row(&sql, [], |row| row.get::<_, Option<u32>>(0))
        .map_err(|e| Error::storage(format!("query current version from {tracker_table}: {e}")))?;
    Ok(v.unwrap_or(0))
}

fn apply_one(conn: &mut Connection, m: &Migration, tracker_table: &str) -> Result<()> {
    let tx = conn
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(|e| Error::storage(format!("BEGIN IMMEDIATE for migration {}: {e}", m.version)))?;
    tx.execute_batch(m.up)
        .map_err(|e| Error::storage(format!("apply migration {}: {e}", m.version)))?;
    let now_ms: i64 = chrono::Utc::now().timestamp_millis();
    let insert_sql =
        format!("INSERT INTO {tracker_table} (version, description, applied_at) VALUES (?, ?, ?)");
    tx.execute(&insert_sql, params![m.version, m.description, now_ms])
        .map_err(|e| Error::storage(format!("insert {tracker_table} row {}: {e}", m.version)))?;
    tx.commit()
        .map_err(|e| Error::storage(format!("commit migration {}: {e}", m.version)))?;
    Ok(())
}

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

    fn open_in_memory() -> Connection {
        Connection::open_in_memory().expect("open in-memory DB")
    }

    #[test]
    fn empty_db_runs_all_migrations() {
        let mut conn = open_in_memory();
        let v = run_migrations(&mut conn).unwrap();
        // Per-tenant chain ends at 7 in v0.8.1 P1 — migration 0004
        // applies to tenants_index.db, NOT to per-tenant DBs (the
        // numbering skip is intentional, see MIGRATIONS comment).
        assert_eq!(v, 10);
        assert_eq!(current_version(&conn).unwrap(), 10);
    }

    #[test]
    fn tenants_index_migrations_apply_independently() {
        // run_tenants_index_migrations targets its own tracker table.
        // Per-tenant chain on the SAME connection must remain at 0
        // (we never ran run_migrations).
        let mut conn = open_in_memory();
        let v = run_tenants_index_migrations(&mut conn).unwrap();
        // v0.9.0 P1 added migration 0009 (last_accessed). Previously:
        // v0.8.1 P3 added 0008 (quota_bytes); v0.8.0 P1 added 0004
        // (initial registry).
        assert_eq!(v, 9);
        // tenants_index tracker reflects 9.
        assert_eq!(current_tenants_index_version(&conn).unwrap(), 9);
        // Per-tenant tracker was never created — current_version returns
        // an error here because schema_migrations doesn't exist, but
        // current_tenants_index_version returning 0 when its table is
        // absent is the contract we care about. Verify the inverse:
        // current_tenants_index_version on a fresh conn returns 0.
        let conn2 = open_in_memory();
        assert_eq!(current_tenants_index_version(&conn2).unwrap(), 0);
    }

    #[test]
    fn tenants_index_migrations_idempotent() {
        let mut conn = open_in_memory();
        let v1 = run_tenants_index_migrations(&mut conn).unwrap();
        let v2 = run_tenants_index_migrations(&mut conn).unwrap();
        assert_eq!(v1, v2);
        let count: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM schema_migrations_tenants_index WHERE version = 4",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1, "tracker row must not be inserted twice");
    }

    /// v0.9.0 P1: migration 0009 adds `tenants.last_accessed` (INTEGER
    /// ms-since-epoch, nullable). Closes the v0.8.0 doc-vs-code gap
    /// (lesson #39 second incident).
    #[test]
    fn migration_0009_adds_tenants_last_accessed_column() {
        let mut conn = open_in_memory();
        run_tenants_index_migrations(&mut conn).unwrap();
        // PRAGMA table_info gives (cid, name, type, notnull, dflt_value, pk)
        let cols: Vec<(String, String, i64)> = conn
            .prepare("PRAGMA table_info('tenants')")
            .unwrap()
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, i64>(3)?,
                ))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        let last_accessed = cols.iter().find(|(n, _, _)| n == "last_accessed");
        let (_, ty, notnull) =
            last_accessed.expect("last_accessed column missing after migration 0009");
        assert_eq!(ty, "INTEGER", "last_accessed must be INTEGER ms-epoch");
        assert_eq!(*notnull, 0, "last_accessed must be nullable");
    }

    /// v0.9.0 P1: migration 0009 is idempotent under re-runs. The runner
    /// guards via `schema_migrations_tenants_index` — calling
    /// `run_tenants_index_migrations` twice in a row must not error
    /// (and must not double-insert the tracker row).
    #[test]
    fn migration_0009_idempotent_on_rerun() {
        let mut conn = open_in_memory();
        let v1 = run_tenants_index_migrations(&mut conn).unwrap();
        assert_eq!(v1, 9);
        let v2 = run_tenants_index_migrations(&mut conn).unwrap();
        assert_eq!(v2, 9);
        let count: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM schema_migrations_tenants_index WHERE version = 9",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1, "0009 tracker row must not be inserted twice");
    }

    #[test]
    fn migration_0004_creates_tenants_registry() {
        let mut conn = open_in_memory();
        run_tenants_index_migrations(&mut conn).unwrap();
        for table in [
            "tenants",
            "audit_events_admin",
            "schema_migrations_tenants_index",
        ] {
            let exists: u32 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?",
                    params![table],
                    |row| row.get(0),
                )
                .unwrap();
            assert_eq!(exists, 1, "missing table after 0004: {table}");
        }
        // Status CHECK rejects illegal values.
        let bad = conn.execute(
            "INSERT INTO tenants (tenant_id, db_filename, created_at_ms, status)
             VALUES ('x', 'x.db', 0, 'bogus')",
            [],
        );
        assert!(bad.is_err(), "status CHECK must reject 'bogus'");
    }

    #[test]
    fn migration_0002_adds_triples_cluster_id_column() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        // PRAGMA table_info gives (cid, name, type, notnull, dflt_value, pk)
        let cols: Vec<(String, String)> = conn
            .prepare("PRAGMA table_info('triples')")
            .unwrap()
            .query_map([], |row| {
                Ok((row.get::<_, String>(1)?, row.get::<_, String>(2)?))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        let names: Vec<&str> = cols.iter().map(|(n, _)| n.as_str()).collect();
        assert!(
            names.contains(&"cluster_id"),
            "triples missing cluster_id after 0002; got {names:?}"
        );
        // Index exists.
        let idx_exists: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type='index' AND name='idx_triples_cluster'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(idx_exists, 1, "idx_triples_cluster missing after 0002");
    }

    #[test]
    fn migration_0002_cluster_delete_cascades_to_triples() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        conn.execute("PRAGMA foreign_keys = ON", []).unwrap();
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        // Seed minimal cluster + triple.
        let cid = "00000000-0000-0000-0000-000000000077";
        let tid = "00000000-0000-0000-0000-000000000099";
        conn.execute(
            "INSERT INTO clusters (cluster_id, coherence, created_at_ms) VALUES (?, ?, ?)",
            params![cid, 0.9, now_ms],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO triples (
                triple_id, subject_id, predicate, object_id, object_kind,
                valid_from_ms, valid_to_ms, confidence, provenance_json,
                created_at_ms, updated_at_ms, cluster_id
             ) VALUES (?, 'subj', 'pred', 'obj', 'literal', ?, NULL, 0.9, '{}', ?, ?, ?)",
            params![tid, now_ms, now_ms, now_ms, cid],
        )
        .unwrap();
        // Pre-condition: triple exists.
        let n_before: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM triples WHERE triple_id = ?",
                params![tid],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_before, 1);
        // Drop the cluster — CASCADE should remove the triple.
        conn.execute("DELETE FROM clusters WHERE cluster_id = ?", params![cid])
            .unwrap();
        let n_after: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM triples WHERE triple_id = ?",
                params![tid],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_after, 0, "CASCADE on clusters should drop the triple");
    }

    #[test]
    fn second_run_is_a_noop() {
        let mut conn = open_in_memory();
        let v1 = run_migrations(&mut conn).unwrap();
        let v2 = run_migrations(&mut conn).unwrap();
        assert_eq!(v1, v2);
        let count: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM schema_migrations WHERE version = 1",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1, "schema_migrations row must not be inserted twice");
    }

    #[test]
    fn all_canonical_tables_present() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let want = [
            "schema_migrations",
            "embedders",
            "episodes",
            "embeddings",
            "pending_index",
            "triples",
            "clusters",
            "cluster_episodes",
            "semantic_abstractions",
            "contradictions",
        ];
        for table in want {
            let exists: u32 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?",
                    params![table],
                    |row| row.get(0),
                )
                .unwrap();
            assert_eq!(exists, 1, "missing canonical table: {table}");
        }
    }

    #[test]
    fn fts_virtual_table_present() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let exists: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='episodes_fts'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(exists, 1, "episodes_fts virtual table missing");
    }

    #[test]
    fn pending_index_schema_matches_adr() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        // The pending_index schema is canonical per ADR-0003 §pending_index.
        // memory_id PK, embedding BLOB, embedding_dim INTEGER, enqueued_at INTEGER.
        let cols: Vec<(String, String)> = conn
            .prepare("PRAGMA table_info('pending_index')")
            .unwrap()
            .query_map([], |row| {
                Ok((row.get::<_, String>(1)?, row.get::<_, String>(2)?))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        let names: Vec<&str> = cols.iter().map(|(n, _)| n.as_str()).collect();
        for required in ["memory_id", "embedding", "embedding_dim", "enqueued_at"] {
            assert!(
                names.contains(&required),
                "pending_index missing column {required}"
            );
        }
    }

    #[test]
    fn fts_trigger_keeps_episodes_content_indexed() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        // Insert a minimal episode row.
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        conn.execute(
            "INSERT INTO episodes (
                memory_id, ts_ms, source_type, content,
                encoding_context_json, confidence, strength, salience,
                tier, created_at_ms, updated_at_ms
             ) VALUES (?, ?, 'user_message', 'the rain in spain falls mainly on the plain',
                       '{}', 0.9, 0.5, 0.5, 'hot', ?, ?)",
            params![
                "00000000-0000-0000-0000-000000000001",
                now_ms,
                now_ms,
                now_ms
            ],
        )
        .unwrap();
        // FTS table should now have a row matching 'spain'.
        let hit: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM episodes_fts WHERE episodes_fts MATCH 'spain'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(hit, 1);
    }

    #[test]
    fn migration_0010_adds_contradiction_lifecycle_columns() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let cols: Vec<String> = conn
            .prepare("PRAGMA table_info('contradictions')")
            .unwrap()
            .query_map([], |row| row.get::<_, String>(1))
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        for required in [
            "status",
            "resolved_at_ms",
            "resolution_note",
            "winning_triple_id",
        ] {
            assert!(
                cols.iter().any(|name| name == required),
                "contradictions missing lifecycle column {required}"
            );
        }
    }

    #[test]
    fn cascade_delete_removes_pending_index_row() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        conn.execute("PRAGMA foreign_keys = ON", []).unwrap();
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        let mid = "00000000-0000-0000-0000-000000000042";
        conn.execute(
            "INSERT INTO episodes (
                memory_id, ts_ms, source_type, content,
                encoding_context_json, confidence, strength, salience,
                tier, created_at_ms, updated_at_ms
             ) VALUES (?, ?, 'user_message', 'hello', '{}', 1.0, 0.5, 0.5, 'hot', ?, ?)",
            params![mid, now_ms, now_ms, now_ms],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO pending_index (memory_id, embedding, embedding_dim, enqueued_at)
             VALUES (?, x'00', 1, ?)",
            params![mid, now_ms],
        )
        .unwrap();
        conn.execute("DELETE FROM episodes WHERE memory_id = ?", params![mid])
            .unwrap();
        let remaining: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM pending_index WHERE memory_id = ?",
                params![mid],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(remaining, 0, "CASCADE should have removed the pending row");
    }

    // -------- 0003 documents + document_chunks + pending_index.kind --------

    /// Helper: insert a minimal document row (only NOT NULL cols + reasonable defaults).
    fn insert_test_document(conn: &Connection, doc_id: &str) {
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        conn.execute(
            "INSERT INTO documents (doc_id, source, mime_type, ingested_at_ms)
             VALUES (?, ?, ?, ?)",
            params![doc_id, "/tmp/test.md", "text/markdown", now_ms],
        )
        .unwrap();
    }

    /// Helper: insert a minimal chunk row tied to `doc_id` at `idx`.
    /// Returns (chunk_id, rowid).
    fn insert_test_chunk(
        conn: &Connection,
        doc_id: &str,
        idx: i64,
        content: &str,
    ) -> (String, i64) {
        let chunk_id = format!("00000000-0000-0000-0000-{:012x}", idx + 0x100);
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        conn.execute(
            "INSERT INTO document_chunks (
                chunk_id, doc_id, chunk_index, content, token_count,
                start_offset, end_offset, created_at_ms
             ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            params![
                chunk_id,
                doc_id,
                idx,
                content,
                content.split_whitespace().count() as i64,
                0i64,
                content.len() as i64,
                now_ms,
            ],
        )
        .unwrap();
        let rowid = conn.last_insert_rowid();
        (chunk_id, rowid)
    }

    #[test]
    fn migration_0003_creates_documents_and_chunks_tables() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        for table in [
            "documents",
            "document_chunks",
            "chunk_embeddings",
            "document_chunks_fts",
        ] {
            let exists: u32 = conn
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type IN ('table','virtual','vtable') AND name=?",
                    params![table],
                    |row| row.get(0),
                )
                .unwrap();
            assert_eq!(exists, 1, "missing table after 0003: {table}");
        }
    }

    #[test]
    fn migration_0003_pending_index_has_kind_column() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let cols: Vec<(String, String)> = conn
            .prepare("PRAGMA table_info('pending_index')")
            .unwrap()
            .query_map([], |row| {
                Ok((row.get::<_, String>(1)?, row.get::<_, String>(2)?))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        let names: Vec<&str> = cols.iter().map(|(n, _)| n.as_str()).collect();
        for required in [
            "kind",
            "memory_id",
            "chunk_id",
            "embedding",
            "embedding_dim",
            "enqueued_at",
        ] {
            assert!(
                names.contains(&required),
                "pending_index missing column {required} after 0003"
            );
        }
    }

    #[test]
    fn migration_0003_backfills_existing_pending_rows_as_episode_kind() {
        // Simulate a DB that was at v0.6.x (migrations 1+2 applied, with a
        // pending_index row pre-existing) and then runs 0003. Pre-0003 rows
        // should land in the rebuilt table with kind='episode'.
        let mut conn = open_in_memory();
        // Apply just the first two migrations manually by slicing MIGRATIONS.
        conn.execute_batch(
            "CREATE TABLE IF NOT EXISTS schema_migrations (
                 version     INTEGER PRIMARY KEY,
                 description TEXT    NOT NULL,
                 applied_at  INTEGER NOT NULL
             );",
        )
        .unwrap();
        for m in &MIGRATIONS[..2] {
            apply_one(&mut conn, m, PER_TENANT_TRACKER).unwrap();
        }
        // Seed an episode + pending_index row using the 0001 schema.
        conn.execute("PRAGMA foreign_keys = ON", []).unwrap();
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        let mid = "00000000-0000-0000-0000-0000000000aa";
        conn.execute(
            "INSERT INTO episodes (
                memory_id, ts_ms, source_type, content,
                encoding_context_json, confidence, strength, salience,
                tier, created_at_ms, updated_at_ms
             ) VALUES (?, ?, 'user_message', 'pre-0003 row', '{}', 1.0, 0.5, 0.5, 'hot', ?, ?)",
            params![mid, now_ms, now_ms, now_ms],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO pending_index (memory_id, embedding, embedding_dim, enqueued_at)
             VALUES (?, x'00', 1, ?)",
            params![mid, now_ms],
        )
        .unwrap();
        // Now run all migrations — 0003 should rebuild the table and preserve the row.
        run_migrations(&mut conn).unwrap();
        let (kind, mem_id, chunk_id): (String, Option<String>, Option<String>) = conn
            .query_row(
                "SELECT kind, memory_id, chunk_id FROM pending_index",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
        assert_eq!(kind, "episode");
        assert_eq!(mem_id.as_deref(), Some(mid));
        assert!(
            chunk_id.is_none(),
            "back-filled row must have NULL chunk_id"
        );
    }

    #[test]
    fn migration_0003_documents_cascade_drops_chunks() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        conn.execute("PRAGMA foreign_keys = ON", []).unwrap();
        let doc = "00000000-0000-0000-0000-0000000000d1";
        insert_test_document(&conn, doc);
        for i in 0..3 {
            insert_test_chunk(&conn, doc, i, &format!("chunk {i}"));
        }
        let n_before: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks WHERE doc_id = ?",
                params![doc],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_before, 3);
        conn.execute("DELETE FROM documents WHERE doc_id = ?", params![doc])
            .unwrap();
        let n_after: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks WHERE doc_id = ?",
                params![doc],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_after, 0, "CASCADE on documents must drop chunks");
    }

    #[test]
    fn migration_0003_pending_index_kind_check_constraint_refuses_bogus_kind() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        let res = conn.execute(
            "INSERT INTO pending_index (kind, memory_id, embedding, embedding_dim, enqueued_at)
             VALUES ('bogus', '00000000-0000-0000-0000-000000000001', x'00', 1, ?)",
            params![now_ms],
        );
        assert!(res.is_err(), "kind='bogus' must violate CHECK constraint");
    }

    #[test]
    fn migration_0003_pending_index_xor_refuses_both_episode_and_chunk_set() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        // Both memory_id AND chunk_id set → violates XOR check (also kind has to disagree)
        let res = conn.execute(
            "INSERT INTO pending_index (kind, memory_id, chunk_id, embedding, embedding_dim, enqueued_at)
             VALUES ('episode', '00000000-0000-0000-0000-000000000001', '00000000-0000-0000-0000-000000000002', x'00', 1, ?)",
            params![now_ms],
        );
        assert!(
            res.is_err(),
            "memory_id AND chunk_id both NOT NULL must violate XOR"
        );
    }

    #[test]
    fn migration_0003_chunk_fts_keeps_in_sync_on_insert() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let doc = "00000000-0000-0000-0000-0000000000d2";
        insert_test_document(&conn, doc);
        insert_test_chunk(&conn, doc, 0, "the rain in spain falls mainly on the plain");
        let hit: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks_fts WHERE document_chunks_fts MATCH 'spain'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            hit, 1,
            "FTS trigger must index the inserted chunk's content"
        );
    }

    #[test]
    fn migration_0003_chunk_fts_keeps_in_sync_on_delete() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let doc = "00000000-0000-0000-0000-0000000000d3";
        insert_test_document(&conn, doc);
        let (chunk_id, _) =
            insert_test_chunk(&conn, doc, 0, "blackbirds singing in the dead of night");
        conn.execute(
            "DELETE FROM document_chunks WHERE chunk_id = ?",
            params![chunk_id],
        )
        .unwrap();
        let hit: u32 = conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks_fts WHERE document_chunks_fts MATCH 'blackbirds'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(
            hit, 0,
            "FTS trigger must remove the chunk from the index after DELETE"
        );
    }

    #[test]
    fn migration_0003_unique_doc_id_chunk_index_enforced() {
        let mut conn = open_in_memory();
        run_migrations(&mut conn).unwrap();
        let doc = "00000000-0000-0000-0000-0000000000d4";
        insert_test_document(&conn, doc);
        insert_test_chunk(&conn, doc, 0, "first");
        // Second chunk with the same chunk_index for the same doc.
        let now_ms: i64 = chrono::Utc::now().timestamp_millis();
        let res = conn.execute(
            "INSERT INTO document_chunks (
                chunk_id, doc_id, chunk_index, content, token_count,
                start_offset, end_offset, created_at_ms
             ) VALUES (?, ?, 0, 'duplicate', 1, 0, 9, ?)",
            params!["00000000-0000-0000-0000-00000000aaaa", doc, now_ms],
        );
        assert!(res.is_err(), "(doc_id, chunk_index) must be UNIQUE");
    }
}