cqs 1.25.0

//! Schema migrations for cqs index database
//!
//! When the schema version changes, migrations allow upgrading existing indexes
//! without requiring a full rebuild (`cqs index --force`).
//!
//! ## Adding a new migration
//!
//! 1. Increment `CURRENT_SCHEMA_VERSION` in `helpers/mod.rs`
//! 2. Add a new migration function: `async fn migrate_vN_to_vM(pool: &SqlitePool) -> Result<()>`
//! 3. Add the case to `run_migration()`: `(N, M) => migrate_vN_to_vM(pool).await`
//! 4. Update `schema.sql` with the new schema
//!
//! ## Migration guidelines
//!
//! - Most changes are additive (new columns, new tables) - these preserve data
//! - For new columns with NOT NULL, use DEFAULT or populate from existing data
//! - Test migrations with real indexes before release
//! - Keep migrations idempotent where possible (use IF NOT EXISTS)

use sqlx::SqlitePool;

use super::helpers::StoreError;

// Used by tests and future migrations
#[allow(unused_imports)]
use super::helpers::CURRENT_SCHEMA_VERSION;

/// Run all migrations from stored version to current version.
///
/// v1.22.0 audit DS-W6: re-reads `schema_version` inside the migration
/// transaction before executing DDL. Two concurrent processes both reading
/// version=17 from the pool, then both running `ALTER TABLE ADD COLUMN`,
/// would crash the second with "duplicate column name" on a perfectly
/// healthy DB. The double-check under the transaction's implicit exclusive
/// lock prevents this: the second process sees the version has already
/// advanced and short-circuits.
pub async fn migrate(pool: &SqlitePool, from: i32, to: i32) -> Result<(), StoreError> {
    if from == to {
        return Ok(()); // Already at target version
    }
    if from > to {
        return Err(StoreError::SchemaNewerThanCq(from));
    }

    tracing::info!(
        from_version = from,
        to_version = to,
        "Starting schema migration"
    );

    let mut tx = pool.begin().await?;

    // DS-W6: re-read version under the write lock. A concurrent process may
    // have already migrated between our caller's pool-level read and our
    // transaction start. If the version is already at or past `to`, bail.
    let current_in_tx: Option<(String,)> =
        sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
            .fetch_optional(&mut *tx)
            .await?;
    let actual_from: i32 = current_in_tx
        .and_then(|(s,)| s.parse().ok())
        .unwrap_or(from);
    if actual_from >= to {
        tracing::info!(
            actual_from,
            to,
            "Schema already migrated by another process, skipping"
        );
        tx.rollback().await?;
        return Ok(());
    }

    for version in actual_from..to {
        tracing::info!(from = version, to = version + 1, "Running migration step");
        run_migration(&mut tx, version, version + 1).await?;
    }
    sqlx::query("UPDATE metadata SET value = ?1 WHERE key = 'schema_version'")
        .bind(to.to_string())
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;

    tracing::info!(new_version = to, "Schema migration complete");

    Ok(())
}

/// Run a single migration step
#[allow(clippy::match_single_binding)] // Intentional: migration arms will be added here
async fn run_migration(
    conn: &mut sqlx::SqliteConnection,
    from: i32,
    to: i32,
) -> Result<(), StoreError> {
    match (from, to) {
        (10, 11) => migrate_v10_to_v11(conn).await,
        (11, 12) => migrate_v11_to_v12(conn).await,
        (12, 13) => migrate_v12_to_v13(conn).await,
        (13, 14) => migrate_v13_to_v14(conn).await,
        (14, 15) => migrate_v14_to_v15(conn).await,
        (15, 16) => migrate_v15_to_v16(conn).await,
        (16, 17) => migrate_v16_to_v17(conn).await,
        (17, 18) => migrate_v17_to_v18(conn).await,
        (18, 19) => migrate_v18_to_v19(conn).await,
        (19, 20) => migrate_v19_to_v20(conn).await,
        _ => Err(StoreError::MigrationNotSupported(from, to)),
    }
}

// ============================================================================
// Migration functions
// ============================================================================

/// Migrate from v10 to v11: add type_edges table
///
/// Adds type-level dependency tracking. Each edge records which chunk references
/// which type, with an edge_kind classification (Param, Return, Field, Impl, Bound, Alias).
/// Catch-all types (inside generics, etc.) use empty string '' for edge_kind instead of NULL
/// to simplify WHERE clause filtering.
///
/// The table will be empty after migration — run `cqs index --force` to populate.
async fn migrate_v10_to_v11(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS type_edges (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_chunk_id TEXT NOT NULL,
            target_type_name TEXT NOT NULL,
            edge_kind TEXT NOT NULL DEFAULT '',
            line_number INTEGER NOT NULL,
            FOREIGN KEY (source_chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query("CREATE INDEX IF NOT EXISTS idx_type_edges_source ON type_edges(source_chunk_id)")
        .execute(&mut *conn)
        .await?;
    sqlx::query("CREATE INDEX IF NOT EXISTS idx_type_edges_target ON type_edges(target_type_name)")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Created type_edges table. Run 'cqs index --force' to populate type edges.");
    Ok(())
}

/// Migrate from v11 to v12: add parent_type_name column to chunks
///
/// Stores the enclosing class/struct/impl name for method chunks.
/// The column will be NULL after migration — run `cqs index --force` to populate.
async fn migrate_v11_to_v12(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query("ALTER TABLE chunks ADD COLUMN parent_type_name TEXT")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Added parent_type_name column. Run 'cqs index --force' to populate method→class links."
    );
    Ok(())
}

/// Migrate from v12 to v13: enrichment idempotency + HNSW dirty flag
///
/// - `enrichment_hash` column on chunks: blake3 hash of call context used during
///   enrichment. NULL means not yet enriched. Allows skipping already-enriched
///   chunks on re-index and detecting partial enrichment after crash.
/// - `hnsw_dirty` metadata key: set to "1" before SQLite chunk writes, cleared
///   to "0" after successful HNSW save. Detects crash between the two writes.
async fn migrate_v12_to_v13(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query("ALTER TABLE chunks ADD COLUMN enrichment_hash TEXT")
        .execute(&mut *conn)
        .await?;

    sqlx::query("INSERT OR IGNORE INTO metadata (key, value) VALUES ('hnsw_dirty', '0')")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Added enrichment_hash column and hnsw_dirty flag. \
         Run 'cqs index --force' to populate enrichment hashes."
    );
    Ok(())
}

/// Migrate from v13 to v14: LLM summaries cache table (SQ-6)
///
/// Stores one-sentence LLM-generated summaries keyed by content_hash.
/// Summaries survive chunk deletion and --force rebuilds.
async fn migrate_v13_to_v14(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS llm_summaries (
            content_hash TEXT PRIMARY KEY,
            summary TEXT NOT NULL,
            model TEXT NOT NULL,
            created_at TEXT NOT NULL
        )",
    )
    .execute(&mut *conn)
    .await?;

    tracing::info!("Created llm_summaries table for LLM-generated function summaries.");
    Ok(())
}

/// Migrate from v14 to v15: 768-dim embeddings (SQ-9)
///
/// Dropped the sentiment dimension — embeddings are now pure model-native output.
/// - Updates dimensions metadata from 769 to model dim (was 768 for E5-base-v2)
/// - Sets hnsw_dirty to trigger HNSW rebuild (old index has 769-dim vectors)
/// - Notes embedding column is left as-is (we write empty blobs now, old data is harmless)
async fn migrate_v14_to_v15(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    // DS-4: Only update dimensions from 769→768 (the old sentiment-augmented size).
    // Databases already using a different model dim (e.g. 1024 for BGE-large) must
    // not be overwritten to 768.
    sqlx::query("UPDATE metadata SET value = '768' WHERE key = 'dimensions' AND value = '769'")
        .execute(&mut *conn)
        .await?;

    sqlx::query("UPDATE metadata SET value = '1' WHERE key = 'hnsw_dirty'")
        .execute(&mut *conn)
        .await?;

    tracing::info!(
        "Updated dimensions and marked HNSW dirty. \
         Run 'cqs index --force' to rebuild embeddings."
    );
    Ok(())
}

/// Migrate from v15 to v16: composite PK on llm_summaries (content_hash, purpose)
///
/// Recreates llm_summaries with a composite primary key so the same content_hash
/// can have multiple summaries for different purposes (e.g., 'summary', 'doc-comment').
/// Existing rows get purpose='summary' as the default.
///
/// Safety: CREATE TABLE, INSERT INTO ... SELECT, DROP TABLE, and ALTER TABLE RENAME
/// are all transactional in SQLite (they write to sqlite_master within the same
/// transaction). If any step fails, the entire migration rolls back and the original
/// llm_summaries table remains intact. The caller (`migrate`) wraps all steps in a
/// single BEGIN/COMMIT via `pool.begin()`.
async fn migrate_v15_to_v16(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS llm_summaries_v2 (
            content_hash TEXT NOT NULL,
            purpose TEXT NOT NULL DEFAULT 'summary',
            summary TEXT NOT NULL,
            model TEXT NOT NULL,
            created_at TEXT NOT NULL,
            PRIMARY KEY (content_hash, purpose)
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query(
        "INSERT OR IGNORE INTO llm_summaries_v2 (content_hash, purpose, summary, model, created_at) \
         SELECT content_hash, 'summary', summary, model, created_at FROM llm_summaries",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query("DROP TABLE IF EXISTS llm_summaries")
        .execute(&mut *conn)
        .await?;

    sqlx::query("ALTER TABLE llm_summaries_v2 RENAME TO llm_summaries")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Recreated llm_summaries with composite PK (content_hash, purpose).");
    Ok(())
}

/// Migrate from v16 to v17: sparse_vectors table + enrichment_version column
///
/// - `sparse_vectors`: stores SPLADE sparse vectors for hybrid search.
///   Each chunk gets a set of (token_id, weight) pairs from the learned sparse encoder.
/// - `enrichment_version`: RT-DATA-2 idempotency marker. Tracks which enrichment pass
///   last processed each chunk, preventing double-application of call graph context.
async fn migrate_v16_to_v17(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v16_to_v17").entered();

    sqlx::query(
        "CREATE TABLE IF NOT EXISTS sparse_vectors (
            chunk_id TEXT NOT NULL,
            token_id INTEGER NOT NULL,
            weight REAL NOT NULL,
            PRIMARY KEY (chunk_id, token_id)
        )",
    )
    .execute(&mut *conn)
    .await?;

    sqlx::query("CREATE INDEX IF NOT EXISTS idx_sparse_token ON sparse_vectors(token_id)")
        .execute(&mut *conn)
        .await?;

    // RT-DATA-2: enrichment idempotency marker
    sqlx::query("ALTER TABLE chunks ADD COLUMN enrichment_version INTEGER NOT NULL DEFAULT 0")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Migrated to v17: sparse_vectors table + enrichment_version column");
    Ok(())
}

/// Migrate from v17 to v18: add embedding_base column to chunks
///
/// Phase 5 of adaptive retrieval: dual embeddings. Each chunk gets a second
/// embedding built from the raw NL description (without LLM summary or call-graph
/// enrichment). Conceptual/behavioral/negation queries route to the base index,
/// structural/multi-step queries keep the enriched index.
///
/// NULL is a valid state post-migration — chunks haven't been re-embedded yet.
/// The base HNSW index is only built once the column is populated; until then
/// the router silently falls back to the enriched index.
async fn migrate_v17_to_v18(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v17_to_v18").entered();

    sqlx::query("ALTER TABLE chunks ADD COLUMN embedding_base BLOB")
        .execute(&mut *conn)
        .await?;

    tracing::info!("Migrated to v18: embedding_base column (NULL until next index pass)");
    Ok(())
}

/// Migrate from v18 to v19: add FK(chunk_id) ON DELETE CASCADE to sparse_vectors
///
/// v1.22.0 audit finding DS-W3: the v17 `sparse_vectors` table was declared
/// without a foreign key to `chunks`, so three code paths in
/// `src/store/chunks/crud.rs` (`delete_by_origin`, `delete_phantom_chunks`,
/// `upsert_chunks_and_calls`) leaked orphan sparse rows — every chunks-delete
/// produced sparse_vectors rows that no query could reach, and `prune_missing`
/// / `prune_all` had to clean them up manually. Worse, the cleanup paths
/// forgot to bump `splade_generation` (DS-W1), so the persisted
/// `splade.index.bin` kept serving stale chunk_ids after a GC.
///
/// This migration makes the invariant structural: any delete from `chunks`
/// now cascades to `sparse_vectors` automatically, the same way
/// `calls.source_chunk_id` and `type_edges.source_chunk_id` already cascade
/// since v10/v11. Memory rule: invalidation counters attached to mutable
/// state must be enforced at the schema layer, not instrumented at specific
/// call sites.
///
/// SQLite does not support `ALTER TABLE ADD FOREIGN KEY`, so we rebuild the
/// table in place. Orphan rows (sparse_vectors whose chunk_id no longer
/// exists in `chunks`) are dropped during the copy — they were leaked data
/// by definition. Row count before and after is logged for transparency.
///
/// After the table swap, the SPLADE generation counter is bumped
/// unconditionally so any persisted `splade.index.bin` from a pre-v19 schema
/// is invalidated on the next load (the on-disk file's embedded generation
/// won't match, forcing a clean rebuild from the new FK-protected table).
async fn migrate_v18_to_v19(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v18_to_v19").entered();

    // Count rows before the rebuild so the migration log makes any silent
    // orphan purge visible instead of invisible.
    let (before_rows,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM sparse_vectors")
        .fetch_one(&mut *conn)
        .await?;

    // Create the new table with the FK constraint. Same PRIMARY KEY shape as
    // v17 so the idx_sparse_token rebuild below is a drop-in replacement.
    sqlx::query(
        "CREATE TABLE sparse_vectors_v19 (
            chunk_id TEXT NOT NULL,
            token_id INTEGER NOT NULL,
            weight REAL NOT NULL,
            PRIMARY KEY (chunk_id, token_id),
            FOREIGN KEY (chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
        )",
    )
    .execute(&mut *conn)
    .await?;

    // Copy only rows whose chunk_id exists in chunks — the INNER JOIN is the
    // orphan filter. Any row that doesn't match was already unreachable; the
    // migration is the right place to drop them.
    sqlx::query(
        "INSERT INTO sparse_vectors_v19 (chunk_id, token_id, weight)
         SELECT s.chunk_id, s.token_id, s.weight
         FROM sparse_vectors s
         INNER JOIN chunks c ON c.id = s.chunk_id",
    )
    .execute(&mut *conn)
    .await?;

    let (after_rows,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM sparse_vectors_v19")
        .fetch_one(&mut *conn)
        .await?;
    let dropped = before_rows - after_rows;
    if dropped > 0 {
        tracing::warn!(
            before = before_rows,
            after = after_rows,
            dropped_orphans = dropped,
            "v18→v19 migration dropped orphan sparse_vectors rows (chunks no longer exist). \
             These were leaks from pre-v19 delete paths."
        );
    } else {
        tracing::info!(
            rows = before_rows,
            "v18→v19 sparse_vectors row count unchanged after FK filter"
        );
    }

    // Drop the old idx_sparse_token first (it's tied to the old table) so the
    // swap doesn't trip the UNIQUE-index-name constraint.
    sqlx::query("DROP INDEX IF EXISTS idx_sparse_token")
        .execute(&mut *conn)
        .await?;
    sqlx::query("DROP TABLE sparse_vectors")
        .execute(&mut *conn)
        .await?;
    sqlx::query("ALTER TABLE sparse_vectors_v19 RENAME TO sparse_vectors")
        .execute(&mut *conn)
        .await?;
    sqlx::query("CREATE INDEX idx_sparse_token ON sparse_vectors(token_id)")
        .execute(&mut *conn)
        .await?;

    // Bump splade_generation so any on-disk splade.index.bin from pre-v19
    // is invalidated — its header generation won't match the new value.
    // The UPSERT seeds the row if it wasn't present (older DBs predating
    // the PR #895 counter never had this metadata key).
    sqlx::query(
        "INSERT INTO metadata (key, value) VALUES ('splade_generation', '1')
         ON CONFLICT(key) DO UPDATE SET
             value = CAST((CAST(value AS INTEGER) + 1) AS TEXT)",
    )
    .execute(&mut *conn)
    .await?;

    tracing::info!(
        "Migrated to v19: sparse_vectors has FK(chunk_id) → chunks(id) ON DELETE CASCADE"
    );
    Ok(())
}

/// Migrate from v19 to v20: AFTER DELETE trigger on chunks bumps splade_generation
///
/// v1.22.0 audit DS-W2 / OB-22 / PB-NEW-6 (triple-confirmed by three
/// independent auditors): `cqs watch` never touched SPLADE. When watch
/// detected a file edit and called `delete_phantom_chunks` or
/// `delete_by_origin`, the v19 FK CASCADE correctly removed the orphan
/// sparse rows, but nothing bumped `splade_generation`. The on-disk
/// `splade.index.bin` still matched the unchanged counter, so readers
/// trusted the stale file and served chunk_ids that no longer existed.
///
/// This trigger fires on every `DELETE FROM chunks` statement, once per
/// deleted row, and bumps the generation via a single metadata UPDATE.
/// For a watch cycle that touches 1-200 chunks, that's 1-200 metadata
/// updates — negligible. For `cqs index --force`, the new DB is fresh
/// and receives no DELETE statements at all, so the trigger cost is
/// zero on the bulk-reindex path. The only concern is `delete_by_origin`
/// during normal reindex when many chunks are displaced; even then the
/// write amplification is ~1-5s per 10k deletions, vs. the minutes the
/// actual rebuild takes.
///
/// The trigger is scoped to `chunks` deletions specifically. sparse_vectors
/// writes are still bumped explicitly by `bump_splade_generation_tx` (one
/// call per upsert transaction, not per row) because that's the only site
/// the trigger-on-sparse_vectors alternative would have caught and it
/// would have fired millions of times during a bulk upsert — row-level
/// triggers on the high-cardinality table were the wrong design.
async fn migrate_v19_to_v20(conn: &mut sqlx::SqliteConnection) -> Result<(), StoreError> {
    let _span = tracing::info_span!("migrate_v19_to_v20").entered();

    sqlx::query(
        "CREATE TRIGGER IF NOT EXISTS bump_splade_on_chunks_delete \
         AFTER DELETE ON chunks \
         BEGIN \
             INSERT INTO metadata (key, value) VALUES ('splade_generation', '1') \
             ON CONFLICT(key) DO UPDATE SET \
                 value = CAST((CAST(value AS INTEGER) + 1) AS TEXT); \
         END",
    )
    .execute(&mut *conn)
    .await?;

    // Bump generation immediately so any pre-v20 persisted splade.index.bin
    // (possibly already out of sync with sparse_vectors because watch-mode
    // deletes since v19 landed never bumped the counter) gets invalidated
    // on the next load.
    sqlx::query(
        "INSERT INTO metadata (key, value) VALUES ('splade_generation', '1')
         ON CONFLICT(key) DO UPDATE SET
             value = CAST((CAST(value AS INTEGER) + 1) AS TEXT)",
    )
    .execute(&mut *conn)
    .await?;

    tracing::info!(
        "Migrated to v20: AFTER DELETE trigger on chunks bumps splade_generation (DS-W2/OB-22 fix)"
    );
    Ok(())
}

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

    #[test]
    fn test_migration_not_supported_error() {
        // Verify unknown migrations produce clear errors
        let err = StoreError::MigrationNotSupported(5, 6);
        let msg = err.to_string();
        assert!(msg.contains("5"));
        assert!(msg.contains("6"));
    }

    #[test]
    fn test_current_schema_version_documented() {
        // Ensure the current version matches what we document
        assert_eq!(CURRENT_SCHEMA_VERSION, 20);
    }

    #[test]
    fn test_migrate_noop_same_version() {
        // Migration from N to N should be a no-op
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            let result = migrate(&pool, 15, 15).await;
            assert!(result.is_ok(), "same-version migration should be no-op");
        });
    }

    #[test]
    fn test_migrate_rejects_downgrade() {
        // from > to should error with SchemaNewerThanCq
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            let result = migrate(&pool, 15, 14).await;
            assert!(result.is_err(), "downgrade should fail");
            match result.unwrap_err() {
                StoreError::SchemaNewerThanCq(v) => assert_eq!(v, 15),
                other => panic!("Expected SchemaNewerThanCq, got: {:?}", other),
            }
        });
    }

    #[test]
    fn test_migrate_v10_to_v11_creates_type_edges() {
        // Full migration test: set up a v10 schema, run migration, verify type_edges exists
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create the minimal schema that a v10 store would have
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    language TEXT NOT NULL DEFAULT '',
                    chunk_type TEXT NOT NULL DEFAULT '',
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL DEFAULT '',
                    content TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL DEFAULT 0,
                    line_end INTEGER NOT NULL DEFAULT 0,
                    parent_id TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Set schema_version to 10
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '10')")
                .execute(&pool)
                .await
                .unwrap();

            // Verify type_edges does NOT exist before migration
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='type_edges'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(table_check.is_none(), "type_edges should not exist yet");

            // Run migration from v10 to v11
            migrate(&pool, 10, 11).await.unwrap();

            // Verify type_edges now exists
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='type_edges'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(
                table_check.is_some(),
                "type_edges should exist after migration"
            );

            // Verify schema_version was updated to 11
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "11");

            // Verify the indexes were created
            let idx_source: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_type_edges_source'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(idx_source.is_some(), "source index should exist");

            let idx_target: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_type_edges_target'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(idx_target.is_some(), "target index should exist");
        });
    }

    #[test]
    fn test_migrate_v12_to_v13() {
        // Full migration test: set up a v12 schema, run migration, verify enrichment_hash + hnsw_dirty
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v12 schema: chunks WITHOUT enrichment_hash, metadata WITHOUT hnsw_dirty
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '12')")
                .execute(&pool)
                .await
                .unwrap();

            // Run migration from v12 to v13
            migrate(&pool, 12, 13).await.unwrap();

            // Verify enrichment_hash column exists by inserting a row that uses it
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at, enrichment_hash) \
                 VALUES ('test', 'file:test.rs', 'file', 'rust', 'function', 'test_fn', \
                 '', 'fn test() {}', 'abc123', 0, 1, X'00', '2026-01-01', '2026-01-01', 'hash123')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Verify hnsw_dirty metadata key exists with value '0'
            let dirty: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'hnsw_dirty'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(dirty.0, "0");

            // Verify schema_version was updated to 13
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "13");
        });
    }

    #[test]
    fn test_migrate_v13_to_v14() {
        // Full migration test: set up a v13 schema, run migration, verify llm_summaries table
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v13 schema: chunks WITH enrichment_hash, metadata WITH hnsw_dirty
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT,
                    enrichment_hash TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '13')")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('hnsw_dirty', '0')")
                .execute(&pool)
                .await
                .unwrap();

            // Verify llm_summaries does NOT exist before migration
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='llm_summaries'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(table_check.is_none(), "llm_summaries should not exist yet");

            // Run migration from v13 to v14
            migrate(&pool, 13, 14).await.unwrap();

            // Verify llm_summaries table exists
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='llm_summaries'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(
                table_check.is_some(),
                "llm_summaries should exist after migration"
            );

            // Verify we can insert into llm_summaries
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, summary, model, created_at) \
                 VALUES ('abc123', 'Test summary', 'claude-4', '2026-01-01')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Verify schema_version was updated to 14
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "14");
        });
    }

    #[test]
    fn test_migrate_v14_to_v15() {
        // Full migration test: set up a v14 schema, run migration, verify dimensions + hnsw_dirty
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v14 schema: chunks WITH enrichment_hash, llm_summaries table
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT,
                    enrichment_hash TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS llm_summaries (
                    content_hash TEXT PRIMARY KEY,
                    summary TEXT NOT NULL,
                    model TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '14')")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('dimensions', '769')")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('hnsw_dirty', '0')")
                .execute(&pool)
                .await
                .unwrap();

            // Run migration from v14 to v15
            migrate(&pool, 14, 15).await.unwrap();

            // Verify dimensions updated to 768
            let dims: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'dimensions'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(dims.0, "768", "dimensions should be updated to 768");

            // Verify hnsw_dirty set to 1 (triggers rebuild)
            let dirty: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'hnsw_dirty'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(dirty.0, "1", "hnsw_dirty should be set to 1");

            // Verify schema_version was updated to 15
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "15");
        });
    }

    #[test]
    fn test_migrate_v15_to_v16() {
        // Full migration test: set up a v15 schema, run migration, verify composite PK
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v15 schema with llm_summaries (single PK on content_hash)
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS llm_summaries (
                    content_hash TEXT PRIMARY KEY,
                    summary TEXT NOT NULL,
                    model TEXT NOT NULL,
                    created_at TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '15')")
                .execute(&pool)
                .await
                .unwrap();

            // Insert two test summaries
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, summary, model, created_at) \
                 VALUES ('hash_a', 'Summary A', 'claude-4', '2026-01-01')",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, summary, model, created_at) \
                 VALUES ('hash_b', 'Summary B', 'claude-4', '2026-01-02')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run migration from v15 to v16
            migrate(&pool, 15, 16).await.unwrap();

            // Verify existing rows have purpose='summary'
            let count: (i64,) =
                sqlx::query_as("SELECT COUNT(*) FROM llm_summaries WHERE purpose = 'summary'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                count.0, 2,
                "both existing rows should have purpose='summary'"
            );

            // Verify composite PK: same content_hash with different purpose should succeed
            sqlx::query(
                "INSERT INTO llm_summaries (content_hash, purpose, summary, model, created_at) \
                 VALUES ('hash_a', 'doc-comment', 'Doc comment A', 'claude-4', '2026-01-03')",
            )
            .execute(&pool)
            .await
            .expect("inserting same content_hash with different purpose should succeed");

            // Verify we now have 3 rows total
            let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM llm_summaries")
                .fetch_one(&pool)
                .await
                .unwrap();
            assert_eq!(count.0, 3, "should have 3 rows after inserting doc-comment");

            // Verify schema_version was updated to 16
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "16");
        });
    }

    #[test]
    fn test_migrate_v12_to_v14_full_chain() {
        // Full chain migration: v12 → v13 → v14 in one call
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create v12 schema: chunks WITHOUT enrichment_hash, no hnsw_dirty
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query(
                "CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    doc TEXT,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    source_mtime INTEGER,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    parent_id TEXT,
                    window_idx INTEGER,
                    parent_type_name TEXT
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '12')")
                .execute(&pool)
                .await
                .unwrap();

            // Run full chain migration from v12 to v14
            migrate(&pool, 12, 14).await.unwrap();

            // Verify enrichment_hash column exists (from v12→v13)
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at, enrichment_hash) \
                 VALUES ('test', 'file:test.rs', 'file', 'rust', 'function', 'test_fn', \
                 '', 'fn test() {}', 'abc123', 0, 1, X'00', '2026-01-01', '2026-01-01', 'hash123')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Verify llm_summaries table exists (from v13→v14)
            let table_check: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='llm_summaries'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(
                table_check.is_some(),
                "llm_summaries should exist after full chain migration"
            );

            // Verify schema_version was updated to 14
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "14");
        });
    }

    #[test]
    fn test_migrate_unsupported_version_range() {
        // Migration from an unsupported range should fail with MigrationNotSupported
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Create metadata table so the SQL doesn't fail on table-not-found
            sqlx::query(
                "CREATE TABLE IF NOT EXISTS metadata (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '8')")
                .execute(&pool)
                .await
                .unwrap();

            let result = migrate(&pool, 8, 11).await;
            assert!(result.is_err(), "unsupported range should fail");
            match result.unwrap_err() {
                StoreError::MigrationNotSupported(from, to) => {
                    assert_eq!(from, 8);
                    assert_eq!(to, 9);
                }
                other => panic!("Expected MigrationNotSupported, got: {:?}", other),
            }
        });
    }

    /// Phase 5 regression: v17→v18 adds embedding_base column without touching
    /// existing rows, and the migration is idempotent-ish in the sense that a
    /// follow-up attempt errors on the duplicate ALTER (caller must not re-run).
    #[test]
    fn test_migrate_v17_to_v18_adds_embedding_base_column() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            // Minimal v17 schema (chunks + metadata); only the columns
            // touched by the v17→v18 migration matter here.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();

            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '17')")
                .execute(&pool)
                .await
                .unwrap();

            // Insert a row with a non-trivial embedding so we can verify it
            // survives the migration untouched.
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('chunk-1', 'file:lib.rs', 'file', 'rust', 'function', 'foo', \
                 'fn foo()', 'fn foo() {}', 'hash1', 10, 20, X'deadbeef', \
                 '2026-04-10', '2026-04-10')",
            )
            .execute(&pool)
            .await
            .unwrap();

            migrate(&pool, 17, 18).await.unwrap();

            // Column exists and defaults to NULL for pre-existing rows.
            let (embedding_existing, embedding_base): (Vec<u8>, Option<Vec<u8>>) =
                sqlx::query_as("SELECT embedding, embedding_base FROM chunks WHERE id = 'chunk-1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(
                embedding_existing,
                vec![0xde, 0xad, 0xbe, 0xef],
                "existing embedding must survive migration untouched"
            );
            assert!(
                embedding_base.is_none(),
                "embedding_base must be NULL for pre-existing rows (base pass hasn't run yet)"
            );

            // Schema version bumped.
            let version: (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(version.0, "18");

            // NULL is a writeable state — caller can populate it later.
            sqlx::query("UPDATE chunks SET embedding_base = X'cafef00d' WHERE id = 'chunk-1'")
                .execute(&pool)
                .await
                .unwrap();
            let (base_after,): (Option<Vec<u8>>,) =
                sqlx::query_as("SELECT embedding_base FROM chunks WHERE id = 'chunk-1'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(base_after, Some(vec![0xca, 0xfe, 0xf0, 0x0d]));
        });
    }

    /// Phase 5: full migrate() chain is idempotent at the dispatcher level.
    /// Calling `migrate(pool, 18, 18)` after a successful upgrade must be a
    /// no-op — the schema_version metadata gates re-execution of the ALTER.
    /// (The raw migration function itself is NOT idempotent — `ALTER TABLE
    /// ADD COLUMN` errors on duplicate column. This test exercises the
    /// dispatcher contract that protects users from the underlying limitation.)
    #[test]
    fn test_migrate_v17_to_v18_dispatcher_is_idempotent() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true),
                )
                .await
                .unwrap();

            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO metadata (key, value) VALUES ('schema_version', '17')")
                .execute(&pool)
                .await
                .unwrap();

            // First call: 17→18 succeeds.
            migrate(&pool, 17, 18).await.unwrap();

            // Second call at the same target version: should be a no-op.
            // This is the property users actually depend on — re-running
            // `cqs index` should not fail just because the schema is current.
            migrate(&pool, 18, 18).await.unwrap();
        });
    }

    /// v1.22.0 audit DS-W3: v18→v19 migration adds FK(chunk_id) ON DELETE
    /// CASCADE on sparse_vectors. Test covers:
    /// - Orphan sparse rows (chunks no longer exist) are dropped during
    ///   the rebuild
    /// - Non-orphan sparse rows survive the rebuild
    /// - The idx_sparse_token index is recreated
    /// - splade_generation is bumped so any pre-v19 persisted index file
    ///   is invalidated
    /// - FK CASCADE works after the migration: deleting a chunk auto-deletes
    ///   its sparse rows
    #[test]
    fn test_migrate_v18_to_v19_adds_fk_cascade_and_purges_orphans() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        // PRAGMA foreign_keys must be ON for ON DELETE CASCADE to fire.
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v18 schema — chunks with embedding_base, sparse_vectors
            // WITHOUT the FK constraint (as shipped in v17).
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "CREATE TABLE sparse_vectors (
                    chunk_id TEXT NOT NULL,
                    token_id INTEGER NOT NULL,
                    weight REAL NOT NULL,
                    PRIMARY KEY (chunk_id, token_id)
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("CREATE INDEX idx_sparse_token ON sparse_vectors(token_id)")
                .execute(&pool)
                .await
                .unwrap();

            sqlx::query(
                "INSERT INTO metadata (key, value) VALUES ('schema_version', '18'),
                                                          ('splade_generation', '5')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Insert two chunks and sparse rows for both of them.
            for id in ["chunk-live", "chunk-also-live"] {
                sqlx::query(
                    "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                     signature, content, content_hash, line_start, line_end, embedding, \
                     created_at, updated_at) \
                     VALUES (?1, 'file:lib.rs', 'file', 'rust', 'function', ?1, \
                     '', '', 'hash', 1, 10, X'00', '2026-04-11', '2026-04-11')",
                )
                .bind(id)
                .execute(&pool)
                .await
                .unwrap();
            }
            sqlx::query(
                "INSERT INTO sparse_vectors (chunk_id, token_id, weight) VALUES
                    ('chunk-live', 1, 0.5),
                    ('chunk-live', 2, 0.3),
                    ('chunk-also-live', 3, 0.8),
                    ('chunk-orphan', 4, 0.9),
                    ('chunk-orphan', 5, 0.1)",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Run the migration.
            migrate(&pool, 18, 19).await.unwrap();

            // Orphan rows dropped, live rows survive.
            let (count,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sparse_vectors WHERE chunk_id = 'chunk-orphan'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(count, 0, "orphan rows should have been dropped");
            let (count_live,): (i64,) =
                sqlx::query_as("SELECT COUNT(*) FROM sparse_vectors WHERE chunk_id = 'chunk-live'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(count_live, 2, "chunk-live rows should survive");

            // idx_sparse_token exists after the rebuild.
            let idx: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_sparse_token'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(idx.is_some(), "idx_sparse_token must be recreated");

            // splade_generation bumped.
            let (gen_val,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_parsed: u64 = gen_val.parse().unwrap();
            assert!(gen_parsed > 5, "splade_generation must be bumped past 5");

            // schema_version updated.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "19");

            // FK CASCADE contract: deleting a chunk removes its sparse rows.
            sqlx::query("DELETE FROM chunks WHERE id = 'chunk-also-live'")
                .execute(&pool)
                .await
                .unwrap();
            let (remaining,): (i64,) = sqlx::query_as(
                "SELECT COUNT(*) FROM sparse_vectors WHERE chunk_id = 'chunk-also-live'",
            )
            .fetch_one(&pool)
            .await
            .unwrap();
            assert_eq!(
                remaining, 0,
                "CASCADE should have removed chunk-also-live's sparse rows"
            );
        });
    }

    /// v1.22.0 audit DS-W2 / OB-22 / PB-NEW-6: v19→v20 adds a trigger on
    /// chunks that bumps splade_generation on every DELETE. Test covers:
    /// - Migration bumps the generation immediately
    /// - The trigger fires on subsequent DELETE FROM chunks
    /// - Multiple deletes produce multiple bumps (cardinality check)
    /// - A no-op reindex (all INSERT no DELETE) does NOT bump via the trigger
    #[test]
    fn test_migrate_v19_to_v20_adds_trigger_that_bumps_on_chunks_delete() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        rt.block_on(async {
            let pool = SqlitePoolOptions::new()
                .max_connections(1)
                .connect_with(
                    sqlx::sqlite::SqliteConnectOptions::new()
                        .filename(&db_path)
                        .create_if_missing(true)
                        .foreign_keys(true),
                )
                .await
                .unwrap();

            // Minimal v19 schema: chunks, sparse_vectors with v19 FK, metadata.
            sqlx::query("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "CREATE TABLE chunks (
                    id TEXT PRIMARY KEY,
                    origin TEXT NOT NULL,
                    source_type TEXT NOT NULL,
                    language TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    signature TEXT NOT NULL,
                    content TEXT NOT NULL,
                    content_hash TEXT NOT NULL,
                    line_start INTEGER NOT NULL,
                    line_end INTEGER NOT NULL,
                    embedding BLOB NOT NULL,
                    embedding_base BLOB,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    enrichment_version INTEGER NOT NULL DEFAULT 0
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query(
                "CREATE TABLE sparse_vectors (
                    chunk_id TEXT NOT NULL,
                    token_id INTEGER NOT NULL,
                    weight REAL NOT NULL,
                    PRIMARY KEY (chunk_id, token_id),
                    FOREIGN KEY (chunk_id) REFERENCES chunks(id) ON DELETE CASCADE
                )",
            )
            .execute(&pool)
            .await
            .unwrap();
            sqlx::query("CREATE INDEX idx_sparse_token ON sparse_vectors(token_id)")
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "INSERT INTO metadata (key, value) VALUES ('schema_version', '19'),
                                                          ('splade_generation', '10')",
            )
            .execute(&pool)
            .await
            .unwrap();

            // Seed a chunk so we have something to delete.
            for id in ["c1", "c2", "c3"] {
                sqlx::query(
                    "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                     signature, content, content_hash, line_start, line_end, embedding, \
                     created_at, updated_at) \
                     VALUES (?1, 'file:lib.rs', 'file', 'rust', 'function', ?1, \
                     '', '', 'h', 1, 10, X'00', '2026-04-12', '2026-04-12')",
                )
                .bind(id)
                .execute(&pool)
                .await
                .unwrap();
            }

            // Run v19 → v20 migration.
            migrate(&pool, 19, 20).await.unwrap();

            // Migration itself bumps the generation once.
            let (gen_after_migration,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_migration: u64 = gen_after_migration.parse().unwrap();
            assert!(
                gen_after_migration > 10,
                "migration should bump generation past starting value 10"
            );

            // Trigger exists.
            let trigger: Option<(String,)> = sqlx::query_as(
                "SELECT name FROM sqlite_master WHERE type='trigger' \
                 AND name='bump_splade_on_chunks_delete'",
            )
            .fetch_optional(&pool)
            .await
            .unwrap();
            assert!(trigger.is_some(), "v20 trigger must exist after migration");

            // Schema version.
            let (v,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'schema_version'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            assert_eq!(v, "20");

            // Deleting a chunk fires the trigger once and bumps the
            // generation.
            let before_one_delete = gen_after_migration;
            sqlx::query("DELETE FROM chunks WHERE id = 'c1'")
                .execute(&pool)
                .await
                .unwrap();
            let (gen_after_one_delete,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_one_delete: u64 = gen_after_one_delete.parse().unwrap();
            assert_eq!(
                gen_after_one_delete,
                before_one_delete + 1,
                "one chunk delete should bump generation by exactly one"
            );

            // Deleting two chunks bumps by two (trigger is row-level).
            sqlx::query("DELETE FROM chunks WHERE id IN ('c2', 'c3')")
                .execute(&pool)
                .await
                .unwrap();
            let (gen_after_two_delete,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_two_delete: u64 = gen_after_two_delete.parse().unwrap();
            assert_eq!(
                gen_after_two_delete,
                gen_after_one_delete + 2,
                "two chunk deletes should bump generation by exactly two"
            );

            // INSERTs do NOT bump via this trigger (new chunks don't
            // invalidate existing sparse data).
            sqlx::query(
                "INSERT INTO chunks (id, origin, source_type, language, chunk_type, name, \
                 signature, content, content_hash, line_start, line_end, embedding, \
                 created_at, updated_at) \
                 VALUES ('c4', 'file:lib.rs', 'file', 'rust', 'function', 'c4', \
                 '', '', 'h', 1, 10, X'00', '2026-04-12', '2026-04-12')",
            )
            .execute(&pool)
            .await
            .unwrap();
            let (gen_after_insert,): (String,) =
                sqlx::query_as("SELECT value FROM metadata WHERE key = 'splade_generation'")
                    .fetch_one(&pool)
                    .await
                    .unwrap();
            let gen_after_insert: u64 = gen_after_insert.parse().unwrap();
            assert_eq!(
                gen_after_insert, gen_after_two_delete,
                "INSERT should NOT bump the generation (no DELETE happened)"
            );
        });
    }
}