Lumotia/crates/storage/src/migrations.rs

use crate::error::{Error, MigrationStep, Result};
use sqlx::SqlitePool;

/// Each migration is a (version, description, sql) tuple.
/// Migrations MUST be append-only — never modify an existing migration.
/// Column defaults and NOT NULL constraints must exactly match the existing schema.
const MIGRATIONS: &[(i64, &str, &str)] = &[
    (
        1,
        "initial schema",
        r#"
        CREATE TABLE IF NOT EXISTS transcripts (
            id TEXT PRIMARY KEY,
            text TEXT NOT NULL DEFAULT '',
            source TEXT NOT NULL DEFAULT 'microphone',
            title TEXT,
            audio_path TEXT,
            duration REAL NOT NULL DEFAULT 0.0,
            engine TEXT,
            model_id TEXT,
            inference_ms INTEGER,
            sample_rate INTEGER,
            audio_channels INTEGER,
            format_mode TEXT,
            remove_fillers INTEGER NOT NULL DEFAULT 0,
            british_english INTEGER NOT NULL DEFAULT 0,
            anti_hallucination INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE TABLE IF NOT EXISTS segments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            transcript_id TEXT NOT NULL REFERENCES transcripts(id) ON DELETE CASCADE,
            start_time REAL NOT NULL,
            end_time REAL NOT NULL,
            text TEXT NOT NULL DEFAULT ''
        );

        CREATE TABLE IF NOT EXISTS tasks (
            id TEXT PRIMARY KEY,
            text TEXT NOT NULL,
            bucket TEXT NOT NULL DEFAULT 'inbox',
            list_id TEXT,
            effort TEXT,
            done INTEGER NOT NULL DEFAULT 0,
            done_at TEXT,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            source_transcript_id TEXT
        );

        CREATE TABLE IF NOT EXISTS task_lists (
            id TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            built_in INTEGER NOT NULL DEFAULT 0,
            profile_id TEXT,
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE TABLE IF NOT EXISTS settings (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS error_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL DEFAULT (datetime('now')),
            context TEXT NOT NULL,
            error_code TEXT,
            message TEXT NOT NULL,
            metadata TEXT
        );

        CREATE INDEX IF NOT EXISTS idx_transcripts_created ON transcripts(created_at);
        CREATE INDEX IF NOT EXISTS idx_segments_transcript ON segments(transcript_id);
        CREATE INDEX IF NOT EXISTS idx_tasks_bucket ON tasks(bucket);
        CREATE INDEX IF NOT EXISTS idx_tasks_transcript ON tasks(source_transcript_id);
        CREATE INDEX IF NOT EXISTS idx_error_log_context ON error_log(context)
    "#,
    ),
    (
        2,
        "transcripts FTS5 + dictionary table",
        r#"
        CREATE VIRTUAL TABLE IF NOT EXISTS transcripts_fts USING fts5(
            text,
            title,
            content='transcripts',
            content_rowid='rowid',
            tokenize='porter unicode61 remove_diacritics 2'
        );

        CREATE TRIGGER IF NOT EXISTS transcripts_ai AFTER INSERT ON transcripts BEGIN
            INSERT INTO transcripts_fts(rowid, text, title)
            VALUES (new.rowid, new.text, COALESCE(new.title, ''));
        END;

        CREATE TRIGGER IF NOT EXISTS transcripts_ad AFTER DELETE ON transcripts BEGIN
            INSERT INTO transcripts_fts(transcripts_fts, rowid, text, title)
            VALUES ('delete', old.rowid, old.text, COALESCE(old.title, ''));
        END;

        CREATE TRIGGER IF NOT EXISTS transcripts_au AFTER UPDATE ON transcripts BEGIN
            INSERT INTO transcripts_fts(transcripts_fts, rowid, text, title)
            VALUES ('delete', old.rowid, old.text, COALESCE(old.title, ''));
            INSERT INTO transcripts_fts(rowid, text, title)
            VALUES (new.rowid, new.text, COALESCE(new.title, ''));
        END;

        CREATE TABLE IF NOT EXISTS dictionary (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            term TEXT NOT NULL UNIQUE,
            note TEXT,
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE INDEX IF NOT EXISTS idx_dictionary_term ON dictionary(term)
    "#,
    ),
    (
        3,
        "micro-stepping: parent_task_id on tasks",
        r#"
        ALTER TABLE tasks ADD COLUMN parent_task_id TEXT REFERENCES tasks(id) ON DELETE CASCADE;
        CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_task_id)
    "#,
    ),
    (
        4,
        "tasks_meta: notes column for persisted free-text",
        r#"
        ALTER TABLE tasks ADD COLUMN notes TEXT NOT NULL DEFAULT ''
    "#,
    ),
    (
        5,
        "transcripts_meta",
        r#"
        ALTER TABLE transcripts ADD COLUMN starred INTEGER NOT NULL DEFAULT 0;
        ALTER TABLE transcripts ADD COLUMN manual_tags TEXT NOT NULL DEFAULT '';
        ALTER TABLE transcripts ADD COLUMN template TEXT NOT NULL DEFAULT '';
        ALTER TABLE transcripts ADD COLUMN language TEXT NOT NULL DEFAULT '';
        ALTER TABLE transcripts ADD COLUMN segments_json TEXT NOT NULL DEFAULT '';
        "#,
    ),
    (
        6,
        "profiles",
        r#"
        CREATE TABLE IF NOT EXISTS profiles (
            id TEXT PRIMARY KEY,
            name TEXT NOT NULL UNIQUE,
            initial_prompt TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS profile_terms (
            id TEXT PRIMARY KEY,
            profile_id TEXT NOT NULL REFERENCES profiles(id) ON DELETE CASCADE,
            term TEXT NOT NULL,
            note TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            UNIQUE(profile_id, term)
        );

        CREATE INDEX IF NOT EXISTS idx_profile_terms_profile_id
            ON profile_terms(profile_id);

        INSERT OR IGNORE INTO profiles (id, name, initial_prompt, created_at)
        VALUES ('00000000-0000-0000-0000-000000000001', 'Default', '', datetime('now'));

        INSERT OR IGNORE INTO profile_terms (id, profile_id, term, note, created_at)
        SELECT
            d.id,
            '00000000-0000-0000-0000-000000000001',
            d.term,
            COALESCE(d.note, ''),
            d.created_at
        FROM dictionary d;

        CREATE TRIGGER IF NOT EXISTS trg_protect_default_profile_delete
        BEFORE DELETE ON profiles
        FOR EACH ROW
        WHEN OLD.id = '00000000-0000-0000-0000-000000000001'
        BEGIN
            SELECT RAISE(ABORT, 'Default profile cannot be deleted');
        END;

        CREATE TRIGGER IF NOT EXISTS trg_protect_default_profile_rename
        BEFORE UPDATE OF name ON profiles
        FOR EACH ROW
        WHEN OLD.id = '00000000-0000-0000-0000-000000000001'
          AND NEW.name != OLD.name
        BEGIN
            SELECT RAISE(ABORT, 'Default profile cannot be renamed');
        END;
        "#,
    ),
    (
        7,
        "drop_dictionary",
        r#"
        DROP INDEX IF EXISTS idx_dictionary_term;
        DROP TABLE IF EXISTS dictionary;
        "#,
    ),
    (
        8,
        "transcript_profile_provenance",
        r#"
        ALTER TABLE transcripts
            ADD COLUMN profile_id TEXT NOT NULL
            DEFAULT '00000000-0000-0000-0000-000000000001';

        CREATE INDEX IF NOT EXISTS idx_transcripts_profile_id
            ON transcripts(profile_id);
        "#,
    ),
    (
        9,
        "transcript_profile_fk",
        r#"
        INSERT OR IGNORE INTO profiles (id, name, initial_prompt, created_at)
        VALUES ('00000000-0000-0000-0000-000000000001', 'Default', '', datetime('now'));

        DROP TRIGGER IF EXISTS transcripts_ai;
        DROP TRIGGER IF EXISTS transcripts_ad;
        DROP TRIGGER IF EXISTS transcripts_au;
        DROP TABLE IF EXISTS transcripts_fts;
        DROP INDEX IF EXISTS idx_segments_transcript;
        DROP INDEX IF EXISTS idx_transcripts_created;
        DROP INDEX IF EXISTS idx_transcripts_profile_id;

        ALTER TABLE segments RENAME TO segments_old;
        ALTER TABLE transcripts RENAME TO transcripts_old;

        CREATE TABLE transcripts (
            id TEXT PRIMARY KEY,
            text TEXT NOT NULL DEFAULT '',
            source TEXT NOT NULL DEFAULT 'microphone',
            title TEXT,
            audio_path TEXT,
            duration REAL NOT NULL DEFAULT 0.0,
            engine TEXT,
            model_id TEXT,
            inference_ms INTEGER,
            sample_rate INTEGER,
            audio_channels INTEGER,
            format_mode TEXT,
            remove_fillers INTEGER NOT NULL DEFAULT 0,
            british_english INTEGER NOT NULL DEFAULT 0,
            anti_hallucination INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            starred INTEGER NOT NULL DEFAULT 0,
            manual_tags TEXT NOT NULL DEFAULT '',
            template TEXT NOT NULL DEFAULT '',
            language TEXT NOT NULL DEFAULT '',
            segments_json TEXT NOT NULL DEFAULT '',
            profile_id TEXT NOT NULL DEFAULT '00000000-0000-0000-0000-000000000001'
                REFERENCES profiles(id) ON DELETE RESTRICT
        );

        CREATE INDEX idx_transcripts_created
            ON transcripts(created_at);
        CREATE INDEX idx_transcripts_profile_id
            ON transcripts(profile_id);

        INSERT INTO transcripts (
            id, text, source, title, audio_path, duration, engine, model_id,
            inference_ms, sample_rate, audio_channels, format_mode,
            remove_fillers, british_english, anti_hallucination, created_at,
            starred, manual_tags, template, language, segments_json, profile_id
        )
        SELECT
            id, text, source, title, audio_path, duration, engine, model_id,
            inference_ms, sample_rate, audio_channels, format_mode,
            remove_fillers, british_english, anti_hallucination, created_at,
            starred, manual_tags, template, language, segments_json,
            CASE
                WHEN profile_id IS NOT NULL
                 AND EXISTS (
                    SELECT 1 FROM profiles
                    WHERE id = transcripts_old.profile_id
                 )
                    THEN profile_id
                ELSE '00000000-0000-0000-0000-000000000001'
            END
        FROM transcripts_old;

        CREATE TABLE segments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            transcript_id TEXT NOT NULL REFERENCES transcripts(id) ON DELETE CASCADE,
            start_time REAL NOT NULL,
            end_time REAL NOT NULL,
            text TEXT NOT NULL DEFAULT ''
        );

        CREATE INDEX idx_segments_transcript
            ON segments(transcript_id);

        INSERT INTO segments (id, transcript_id, start_time, end_time, text)
        SELECT id, transcript_id, start_time, end_time, text
        FROM segments_old;

        DROP TABLE segments_old;
        DROP TABLE transcripts_old;

        CREATE VIRTUAL TABLE transcripts_fts USING fts5(
            text,
            title,
            content='transcripts',
            content_rowid='rowid',
            tokenize='porter unicode61 remove_diacritics 2'
        );

        CREATE TRIGGER transcripts_ai AFTER INSERT ON transcripts BEGIN
            INSERT INTO transcripts_fts(rowid, text, title)
            VALUES (new.rowid, new.text, COALESCE(new.title, ''));
        END;

        CREATE TRIGGER transcripts_ad AFTER DELETE ON transcripts BEGIN
            INSERT INTO transcripts_fts(transcripts_fts, rowid, text, title)
            VALUES ('delete', old.rowid, old.text, COALESCE(old.title, ''));
        END;

        CREATE TRIGGER transcripts_au AFTER UPDATE ON transcripts BEGIN
            INSERT INTO transcripts_fts(transcripts_fts, rowid, text, title)
            VALUES ('delete', old.rowid, old.text, COALESCE(old.title, ''));
            INSERT INTO transcripts_fts(rowid, text, title)
            VALUES (new.rowid, new.text, COALESCE(new.title, ''));
        END;

        INSERT INTO transcripts_fts(rowid, text, title)
        SELECT rowid, text, COALESCE(title, '')
        FROM transcripts;
        "#,
    ),
    (
        10,
        "feedback: HITL thumbs + correction capture",
        r#"
        -- Feedback rows capture human-in-the-loop signal on AI-generated
        -- output. Two flavours bundled into one table:
        --   - thumbs (rating = -1 | +1, original_text optional, corrected_text NULL)
        --   - correction (rating defaults to +1, original_text + corrected_text present)
        --
        -- `target_type` names the producing surface:
        --   'microstep'  — subtask decomposition from DECOMPOSE_TASK_SYSTEM
        --   'task_extraction' — tasks lifted from a transcript (EXTRACT_TASKS_SYSTEM)
        --   'cleanup'    — transcript cleanup output
        --
        -- `target_id` is the surface-specific identifier where one exists
        -- (subtask id, task id, transcript id). NULL is allowed because
        -- not every feedback event has a stable target id yet.
        --
        -- `context_json` carries the input the AI was conditioned on
        -- (parent task text, transcript chunk, etc.) so future prompt
        -- builders can reconstruct the original I/O pair for few-shot
        -- injection or semantic retrieval.
        CREATE TABLE feedback (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            target_type TEXT NOT NULL
                CHECK (target_type IN ('microstep', 'task_extraction', 'cleanup')),
            target_id TEXT,
            rating INTEGER NOT NULL
                CHECK (rating IN (-1, 0, 1)),
            original_text TEXT,
            corrected_text TEXT,
            context_json TEXT,
            profile_id TEXT NOT NULL DEFAULT '00000000-0000-0000-0000-000000000001'
                REFERENCES profiles(id) ON DELETE RESTRICT,
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE INDEX idx_feedback_target_type_rating
            ON feedback(target_type, rating, created_at DESC);
        CREATE INDEX idx_feedback_profile
            ON feedback(profile_id, target_type, created_at DESC);
        "#,
    ),
    (
        11,
        "tasks: energy tagging for match-my-energy sort",
        r#"
        -- Phase 3 of the feature-complete roadmap: replaces the cut
        -- temptation-bundling feature with a deterministic client-side
        -- sort that matches tasks to the user's current energy state.
        -- NULL is the expected normal case — users who never tag get
        -- Medium-equivalent treatment at sort time (see Match-my-energy
        -- logic in src/lib/pages/TasksPage.svelte).
        --
        -- profile_id is deliberately absent from the index: tasks
        -- currently carry no profile_id column, so a per-profile index
        -- is out of scope until the broader task → profile migration
        -- lands. See HANDOVER deferred list.
        ALTER TABLE tasks
            ADD COLUMN energy TEXT
            CHECK (energy IS NULL OR energy IN ('high', 'medium', 'brain_dead'));

        CREATE INDEX idx_tasks_energy_created
            ON tasks(energy, created_at DESC);
        "#,
    ),
    (
        12,
        "implementation intentions: if-then automation rules",
        r#"
        -- Phase 7 of the feature-complete roadmap. Rules are local-only,
        -- user-authored implementation intentions: "if this happens, then
        -- do this small thing". Execution stays in the frontend event bus;
        -- SQLite owns the durable definition and the once-per-day marker
        -- for time-of-day rules.
        CREATE TABLE implementation_rules (
            id TEXT PRIMARY KEY,
            enabled INTEGER NOT NULL DEFAULT 1
                CHECK (enabled IN (0, 1)),
            trigger_kind TEXT NOT NULL
                CHECK (trigger_kind IN (
                    'time_of_day',
                    'task_completed',
                    'morning_triage_finished'
                )),
            trigger_value TEXT NOT NULL DEFAULT '',
            actions_json TEXT NOT NULL DEFAULT '[]',
            last_fired_key TEXT,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            updated_at TEXT NOT NULL DEFAULT (datetime('now'))
        );

        CREATE INDEX idx_implementation_rules_enabled_trigger
            ON implementation_rules(enabled, trigger_kind);
        "#,
    ),
    (
        13,
        "gamification: auto_completed flag for cascade-completed parents",
        r#"
        -- Phase 8 of the feature-complete roadmap. Parents that close via
        -- the complete_subtask_and_check_parent cascade must not count
        -- towards daily completion totals. The user already got credit
        -- for ticking the subtask. This column distinguishes manual
        -- completions (0) from cascade completions (1). The daily-count
        -- query then excludes auto_completed = 1.
        --
        -- Partial index keeps the index small: only completed rows occupy
        -- it, since uncompleted rows have done_at IS NULL.
        ALTER TABLE tasks ADD COLUMN auto_completed INTEGER NOT NULL DEFAULT 0
            CHECK (auto_completed IN (0, 1));

        CREATE INDEX idx_tasks_done_at_auto_completed
            ON tasks(done_at, auto_completed)
            WHERE done_at IS NOT NULL;
        "#,
    ),
    (
        14,
        "transcripts: llm_tags column for Phase 9 LLM content tags",
        r#"
        -- Phase 9 of the feature-complete roadmap. AI-generated content
        -- tags (topic:* and intent:*) are stored alongside manual_tags as
        -- a comma-joined string, mirroring how manual_tags persists. Pre-
        -- existing rows default to empty string. The frontend chips loop
        -- handles "" as "no tags".
        ALTER TABLE transcripts ADD COLUMN llm_tags TEXT NOT NULL DEFAULT '';
        "#,
    ),
    (
        15,
        "transcripts: composite (profile_id, created_at DESC) index",
        r#"
        -- Performance index for the dominant transcripts query path:
        --   WHERE profile_id = ? ORDER BY created_at DESC LIMIT ?
        -- The standalone idx_transcripts_profile_id and idx_transcripts_created
        -- forced SQLite to either filter by profile then sort, or scan the date
        -- index and filter — fine at hundreds of rows, painful past a few thousand.
        -- A composite index covers both predicates in one ordered seek.
        CREATE INDEX IF NOT EXISTS idx_transcripts_profile_created
            ON transcripts(profile_id, created_at DESC);
        "#,
    ),
    (
        16,
        "transcripts: soft-delete column for trash + audio cleanup (Rev-2, Rev-3)",
        r#"
        -- Atomiser fix 2026-05-12 — Rev-2 (hard-DELETE w/ no trash) and
        -- Rev-3 (orphan WAV files on transcript delete) both fanned out
        -- from the same DELETE-based delete path. The new contract:
        --
        --   * `delete_transcript` UPDATEs deleted_at = datetime('now')
        --     and best-effort removes the audio file at audio_path.
        --     The DB row is preserved so the user can restore.
        --   * `list_transcripts*`, `search_transcripts`, `count_transcripts`
        --     filter `deleted_at IS NULL`.
        --   * `purge_deleted_transcripts(older_than_days)` does the
        --     actual hard DELETE on rows past the retention window,
        --     scheduled once at startup. FK CASCADE on segments still
        --     fires at purge time.
        --
        -- Pre-existing rows default to NULL (not deleted). The column
        -- is plain TEXT (ISO-8601 datetime), nullable, no CHECK — we
        -- only ever compare against datetime('now', '-N days') and
        -- IS NULL / IS NOT NULL.
        ALTER TABLE transcripts ADD COLUMN deleted_at TEXT;

        -- Partial index over the trash so the purge query stays cheap
        -- even on databases with months of soft-deleted rows.
        CREATE INDEX IF NOT EXISTS idx_transcripts_deleted_at
            ON transcripts(deleted_at) WHERE deleted_at IS NOT NULL;
        "#,
    ),
    (
        17,
        "onboarding_events + lumotia_events tables",
        r#"
        -- onboarding_events: gates first-run, supplies time-to-first-capture metric
        CREATE TABLE IF NOT EXISTS onboarding_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            event TEXT NOT NULL,
            completed_at INTEGER NOT NULL,
            version TEXT NOT NULL,
            skipped INTEGER NOT NULL DEFAULT 0,
            notes TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_onboarding_events_event ON onboarding_events(event);

        -- lumotia_events: opt-in local activation log
        CREATE TABLE IF NOT EXISTS lumotia_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            kind TEXT NOT NULL,
            occurred_at INTEGER NOT NULL,
            payload TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_lumotia_events_kind ON lumotia_events(kind);
        CREATE INDEX IF NOT EXISTS idx_lumotia_events_occurred ON lumotia_events(occurred_at);
        "#,
    ),
];

/// Split SQL into individual statements, respecting BEGIN...END trigger blocks.
fn split_statements(sql: &str) -> Vec<String> {
    let mut result = Vec::new();
    let mut current = String::new();
    let mut depth: i32 = 0;
    let upper = sql.to_ascii_uppercase();
    let ub = upper.as_bytes();
    let ob = sql.as_bytes();
    let n = ob.len();
    let mut i = 0;
    while i < n {
        let prev_alpha = i > 0 && ob[i - 1].is_ascii_alphanumeric();
        if !prev_alpha && ub[i..].starts_with(b"BEGIN") {
            let next_alpha = i + 5 < n && ob[i + 5].is_ascii_alphanumeric();
            if !next_alpha {
                depth += 1;
            }
        }
        if !prev_alpha && ub[i..].starts_with(b"END") {
            let next_alpha = i + 3 < n && ob[i + 3].is_ascii_alphanumeric();
            if !next_alpha && depth > 0 {
                depth -= 1;
            }
        }
        if ob[i] == b';' && depth == 0 {
            let stmt = current.trim().to_string();
            if !stmt.is_empty() {
                result.push(stmt);
            }
            current = String::new();
        } else {
            current.push(ob[i] as char);
        }
        i += 1;
    }
    let stmt = current.trim().to_string();
    if !stmt.is_empty() {
        result.push(stmt);
    }
    result
}

/// Ensure the schema_version table exists and run any pending migrations.
pub async fn run_migrations(pool: &SqlitePool) -> Result<()> {
    run_migrations_slice(pool, MIGRATIONS).await
}

/// Apply the pending prefix of `migrations`, each inside its own
/// transaction along with the matching `schema_version` row insert.
///
/// Atomicity was added in response to the 2026-04-22 review (RB-02):
/// the previous implementation executed statements individually against
/// the pool and only recorded the new version after all statements had
/// succeeded. A multi-statement migration that failed midway therefore
/// left the schema partially changed but still appearing unapplied —
/// the next startup would replay the migration against a mutated DB
/// and fail in surprising ways.
///
/// Wrapping both the statements and the version record in a single
/// `Transaction` is sufficient for SQLite (DDL participates in
/// transactions there). If a future migration needs an operation that
/// implicitly commits (`VACUUM`, `REINDEX`, `ATTACH`), it must be split
/// out into its own non-transactional migration — reviewer's job to
/// flag.
async fn run_migrations_slice(pool: &SqlitePool, migrations: &[(i64, &str, &str)]) -> Result<()> {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS schema_version (
            version INTEGER PRIMARY KEY,
            description TEXT NOT NULL,
            applied_at TEXT NOT NULL DEFAULT (datetime('now'))
        )",
    )
    .execute(pool)
    .await
    .map_err(|source| Error::Migration {
        version: None,
        step: MigrationStep::SchemaVersionTableCreate,
        source,
    })?;

    let current: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(version), 0) FROM schema_version")
        .fetch_one(pool)
        .await
        .map_err(|source| Error::Migration {
            version: None,
            step: MigrationStep::SchemaVersionQuery,
            source,
        })?;

    for (version, description, sql) in migrations {
        if *version > current {
            tracing::info!(
                target: "lumotia_storage",
                version,
                description,
                "running migration",
            );

            let mut tx = pool.begin().await.map_err(|source| Error::Migration {
                version: Some(*version),
                step: MigrationStep::TxBegin,
                source,
            })?;

            for statement in split_statements(sql) {
                sqlx::query(&statement)
                    .execute(&mut *tx)
                    .await
                    .map_err(|source| Error::Migration {
                        version: Some(*version),
                        step: MigrationStep::Apply,
                        source,
                    })?;
            }

            sqlx::query("INSERT INTO schema_version (version, description) VALUES (?, ?)")
                .bind(version)
                .bind(description)
                .execute(&mut *tx)
                .await
                .map_err(|source| Error::Migration {
                    version: Some(*version),
                    step: MigrationStep::RecordVersion,
                    source,
                })?;

            tx.commit().await.map_err(|source| Error::Migration {
                version: Some(*version),
                step: MigrationStep::Commit,
                source,
            })?;

            tracing::info!(
                target: "lumotia_storage",
                version,
                "migration complete",
            );
        }
    }

    Ok(())
}

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

    async fn fk_test_pool() -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        sqlx::query("PRAGMA foreign_keys = ON")
            .execute(&pool)
            .await
            .expect("enable foreign keys");
        pool
    }

    #[tokio::test]
    async fn test_migrations_run_on_empty_db() {
        let pool = fk_test_pool().await;

        run_migrations(&pool).await.unwrap();

        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM schema_version")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count, 17);

        sqlx::query("INSERT INTO settings (key, value) VALUES ('test', 'value')")
            .execute(&pool)
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_migrations_idempotent() {
        let pool = fk_test_pool().await;

        run_migrations(&pool).await.unwrap();
        run_migrations(&pool).await.unwrap();

        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM schema_version")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count, 17);
    }

    #[tokio::test]
    async fn migration_tasks_meta_adds_columns() {
        // Task 2.6 — verify list_id / effort / notes are all present on the
        // tasks table after migrations run. list_id and effort have been
        // present since v1 (nullable); notes is added by v4.
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        let info = sqlx::query("PRAGMA table_info(tasks)")
            .fetch_all(&pool)
            .await
            .unwrap();
        let names: Vec<String> = info.iter().map(|r| r.get::<String, _>("name")).collect();
        for col in ["list_id", "effort", "notes"] {
            assert!(
                names.contains(&col.to_string()),
                "tasks must have {col}; got {names:?}"
            );
        }
    }

    #[tokio::test]
    async fn migration_implementation_rules_adds_rule_table() {
        let pool = fk_test_pool().await;
        run_migrations(&pool).await.expect("migrate");

        let info = sqlx::query("PRAGMA table_info(implementation_rules)")
            .fetch_all(&pool)
            .await
            .unwrap();
        let names: Vec<String> = info.iter().map(|r| r.get::<String, _>("name")).collect();
        for col in [
            "id",
            "enabled",
            "trigger_kind",
            "trigger_value",
            "actions_json",
            "last_fired_key",
            "created_at",
            "updated_at",
        ] {
            assert!(
                names.contains(&col.to_string()),
                "implementation_rules must have {col}; got {names:?}"
            );
        }

        let rejected = sqlx::query(
            "INSERT INTO implementation_rules (id, trigger_kind, actions_json)
             VALUES ('bad', 'calendar_event', '[]')",
        )
        .execute(&pool)
        .await;
        assert!(
            rejected.is_err(),
            "trigger_kind CHECK constraint must reject unknown triggers"
        );
    }

    #[tokio::test]
    async fn migration_transcripts_meta_adds_columns() {
        // Task 2.5 — verify starred / manual_tags / template / language /
        // segments_json are all present on the transcripts table after
        // migrations run. All five are added by v5.
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        let info = sqlx::query("PRAGMA table_info(transcripts)")
            .fetch_all(&pool)
            .await
            .unwrap();
        let names: Vec<String> = info.iter().map(|r| r.get::<String, _>("name")).collect();
        for col in [
            "starred",
            "manual_tags",
            "template",
            "language",
            "segments_json",
        ] {
            assert!(
                names.contains(&col.to_string()),
                "transcripts must have {col}; got {names:?}"
            );
        }
    }

    #[tokio::test]
    async fn migration_transcript_profile_provenance_adds_profile_id() {
        let pool = fk_test_pool().await;
        run_migrations(&pool).await.expect("migrate");

        let info = sqlx::query("PRAGMA table_info(transcripts)")
            .fetch_all(&pool)
            .await
            .unwrap();
        let names: Vec<String> = info.iter().map(|r| r.get::<String, _>("name")).collect();
        assert!(
            names.contains(&"profile_id".to_string()),
            "transcripts must have profile_id; got {names:?}"
        );
    }

    #[tokio::test]
    async fn migration_v9_reconciles_orphaned_transcript_profiles_and_adds_fk() {
        let pool = fk_test_pool().await;
        run_migrations_up_to(&pool, 8).await.expect("migrate to v8");

        sqlx::query(
            "INSERT INTO profiles (id, name, initial_prompt, created_at)
             VALUES ('profile-valid', 'Valid', '', datetime('now'))",
        )
        .execute(&pool)
        .await
        .expect("seed valid profile");

        sqlx::query(
            "INSERT INTO transcripts (
                id, text, source, title, audio_path, duration, engine, model_id,
                inference_ms, sample_rate, audio_channels, format_mode,
                remove_fillers, british_english, anti_hallucination, created_at,
                starred, manual_tags, template, language, segments_json, profile_id
            ) VALUES (
                't-orphan', 'orphan body', 'microphone', 'Orphan', NULL, 0.0, NULL, NULL,
                NULL, NULL, NULL, NULL, 0, 0, 0, datetime('now'),
                0, '', '', '', '', 'profile-missing'
            )",
        )
        .execute(&pool)
        .await
        .expect("seed orphan transcript");
        sqlx::query(
            "INSERT INTO transcripts (
                id, text, source, title, audio_path, duration, engine, model_id,
                inference_ms, sample_rate, audio_channels, format_mode,
                remove_fillers, british_english, anti_hallucination, created_at,
                starred, manual_tags, template, language, segments_json, profile_id
            ) VALUES (
                't-valid', 'valid body', 'microphone', 'Valid', NULL, 0.0, NULL, NULL,
                NULL, NULL, NULL, NULL, 0, 0, 0, datetime('now'),
                0, '', '', '', '', 'profile-valid'
            )",
        )
        .execute(&pool)
        .await
        .expect("seed valid transcript");
        sqlx::query(
            "INSERT INTO segments (transcript_id, start_time, end_time, text)
             VALUES ('t-orphan', 0.0, 1.0, 'segment')",
        )
        .execute(&pool)
        .await
        .expect("seed segment");

        run_migrations(&pool).await.expect("migrate to v9");

        let orphan_profile: String =
            sqlx::query_scalar("SELECT profile_id FROM transcripts WHERE id = 't-orphan'")
                .fetch_one(&pool)
                .await
                .expect("read healed orphan");
        assert_eq!(orphan_profile, crate::DEFAULT_PROFILE_ID);

        let valid_profile: String =
            sqlx::query_scalar("SELECT profile_id FROM transcripts WHERE id = 't-valid'")
                .fetch_one(&pool)
                .await
                .expect("read preserved profile");
        assert_eq!(valid_profile, "profile-valid");

        let segment_count: i64 =
            sqlx::query_scalar("SELECT COUNT(*) FROM segments WHERE transcript_id = 't-orphan'")
                .fetch_one(&pool)
                .await
                .expect("read migrated segments");
        assert_eq!(segment_count, 1, "segments must survive transcript rebuild");

        let fk_rows = sqlx::query("PRAGMA foreign_key_list(transcripts)")
            .fetch_all(&pool)
            .await
            .expect("read transcript foreign keys");
        assert!(
            fk_rows.iter().any(|row| {
                row.get::<String, _>("table") == "profiles"
                    && row.get::<String, _>("from") == "profile_id"
                    && row.get::<String, _>("to") == "id"
                    && row.get::<String, _>("on_delete") == "RESTRICT"
            }),
            "transcripts.profile_id must reference profiles(id) with ON DELETE RESTRICT"
        );

        let fts_hits: i64 = sqlx::query_scalar(
            "SELECT COUNT(*) FROM transcripts_fts WHERE transcripts_fts MATCH 'orphan'",
        )
        .fetch_one(&pool)
        .await
        .expect("query rebuilt fts");
        assert_eq!(
            fts_hits, 1,
            "fts index must be rebuilt for existing transcripts"
        );
    }

    #[tokio::test]
    async fn test_parent_task_id_cascade_delete() {
        let pool = fk_test_pool().await;

        run_migrations(&pool).await.unwrap();

        // Insert parent task
        sqlx::query("INSERT INTO tasks (id, text) VALUES ('parent-1', 'Parent task')")
            .execute(&pool)
            .await
            .unwrap();

        // Insert child task with parent_task_id
        sqlx::query("INSERT INTO tasks (id, text, parent_task_id) VALUES ('child-1', 'Child task', 'parent-1')")
            .execute(&pool)
            .await
            .unwrap();

        // Delete parent — child should cascade
        sqlx::query("DELETE FROM tasks WHERE id = 'parent-1'")
            .execute(&pool)
            .await
            .unwrap();

        let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tasks")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(
            remaining, 0,
            "cascade delete should remove child when parent is deleted"
        );
    }

    /// Test-only helper: run migrations only up to (and including) `target_version`.
    /// Used by the v6 upgrade-path test to seed a v5 schema with
    /// dictionary rows before applying v6.
    async fn run_migrations_up_to(pool: &SqlitePool, target_version: i64) -> Result<()> {
        let filtered: Vec<(i64, &str, &str)> = MIGRATIONS
            .iter()
            .filter(|(v, _, _)| *v <= target_version)
            .copied()
            .collect();
        run_migrations_slice(pool, &filtered).await
    }

    #[tokio::test]
    async fn migration_v6_seeds_default_profile_on_fresh_db() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM profiles")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count, 1, "Default profile must be seeded on fresh install");

        let name: String = sqlx::query_scalar("SELECT name FROM profiles WHERE id = ?")
            .bind(crate::DEFAULT_PROFILE_ID)
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(name, "Default");

        let terms_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM profile_terms")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(terms_count, 0, "Fresh DB has no dictionary rows to copy");
    }

    #[tokio::test]
    async fn migration_v6_copies_dictionary_rows_to_default_profile_terms() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");

        // Bring schema up to v5 (dictionary + transcripts_meta, no profile_terms).
        run_migrations_up_to(&pool, 5).await.expect("migrate to v5");

        // dictionary.id is INTEGER PK AUTOINCREMENT (see v2); let SQLite assign rowids.
        sqlx::query(
            "INSERT INTO dictionary (term, note, created_at) VALUES \
             ('Lumotia', '', datetime('now')), \
             ('CORBEL', 'brand', datetime('now')), \
             ('Wren', '', datetime('now'))",
        )
        .execute(&pool)
        .await
        .expect("seed dictionary");

        run_migrations(&pool).await.expect("migrate to v6");

        let copied: i64 =
            sqlx::query_scalar("SELECT COUNT(*) FROM profile_terms WHERE profile_id = ?")
                .bind(crate::DEFAULT_PROFILE_ID)
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(copied, 3);

        let corbel_note: String =
            sqlx::query_scalar("SELECT note FROM profile_terms WHERE term = 'CORBEL'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(corbel_note, "brand");
    }

    #[tokio::test]
    async fn migration_v6_trigger_rejects_default_profile_delete() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        let result = sqlx::query("DELETE FROM profiles WHERE id = ?")
            .bind(crate::DEFAULT_PROFILE_ID)
            .execute(&pool)
            .await;

        assert!(result.is_err(), "trigger must block Default deletion");
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("Default profile cannot be deleted"),
            "got: {msg}"
        );
    }

    #[tokio::test]
    async fn migration_v6_trigger_rejects_default_profile_rename() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        let result = sqlx::query("UPDATE profiles SET name = 'NotDefault' WHERE id = ?")
            .bind(crate::DEFAULT_PROFILE_ID)
            .execute(&pool)
            .await;

        assert!(result.is_err(), "trigger must block Default rename");
    }

    #[tokio::test]
    async fn migration_v7_drops_dictionary_table() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        // dictionary table should not exist after v7.
        let result: std::result::Result<i64, sqlx::Error> =
            sqlx::query_scalar("SELECT COUNT(*) FROM dictionary")
                .fetch_one(&pool)
                .await;
        assert!(result.is_err(), "dictionary table must be gone after v7");
        let err = result.unwrap_err().to_string().to_lowercase();
        assert!(err.contains("no such table"), "got: {err}");
    }

    // RB-02 regression: a multi-statement migration that fails part-way
    // through must leave no trace on disk — the transaction rolls back
    // both the partial schema change and (implicitly) the `schema_version`
    // row that the pre-fix implementation would have recorded after
    // statement-level success.
    //
    // The poisoned migration below first creates `poison_marker`
    // (syntactically valid, would succeed against any SQLite) and then
    // runs a guaranteed-invalid function call. Under the new atomic
    // implementation, neither `poison_marker` nor the poison row should
    // survive the failed call.
    //
    // Version number must sit above the real MIGRATIONS max so the
    // baseline migrate cleanly finishes first.
    #[tokio::test]
    async fn multi_statement_migration_rolls_back_on_failure() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");

        run_migrations(&pool).await.expect("baseline migrate");

        // Discover the real max version so the poison migration is
        // always exactly one past the end of MIGRATIONS, regardless of
        // how many real migrations we add in future.
        let real_max: i64 =
            sqlx::query_scalar("SELECT COALESCE(MAX(version), 0) FROM schema_version")
                .fetch_one(&pool)
                .await
                .expect("read schema_version");
        let poison_version = real_max + 1;

        let poison: &[(i64, &str, &str)] = &[(
            poison_version,
            "rb-02 atomicity poison",
            r#"
            CREATE TABLE poison_marker (id INTEGER PRIMARY KEY);
            SELECT this_function_does_not_exist();
            "#,
        )];

        let result = run_migrations_slice(&pool, poison).await;
        assert!(
            result.is_err(),
            "poisoned migration must return Err, got: {result:?}"
        );

        // `poison_marker` must not be on disk — transaction rolled back.
        let marker: std::result::Result<i64, sqlx::Error> =
            sqlx::query_scalar("SELECT COUNT(*) FROM poison_marker")
                .fetch_one(&pool)
                .await;
        assert!(
            marker.is_err(),
            "poison_marker must not exist; got: {marker:?}"
        );

        // `schema_version` must not include the poison version — version
        // insert is part of the same transaction that rolled back.
        let max: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(version), 0) FROM schema_version")
            .fetch_one(&pool)
            .await
            .expect("read schema_version");
        assert_eq!(
            max, real_max,
            "schema_version must not advance past the failed migration"
        );
    }

    #[tokio::test]
    async fn migration_v13_adds_auto_completed_column() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        // Column exists.
        let info = sqlx::query("PRAGMA table_info(tasks)")
            .fetch_all(&pool)
            .await
            .expect("pragma");
        let names: Vec<String> = info.iter().map(|r| r.get::<String, _>("name")).collect();
        assert!(
            names.iter().any(|n| n == "auto_completed"),
            "expected auto_completed column, got {names:?}"
        );

        // Existing completed rows default to 0. Insert a pre-existing-looking
        // task via raw SQL to simulate a row from before the migration.
        sqlx::query(
            "INSERT INTO tasks (id, text, bucket, done, done_at) \
             VALUES ('t1', 'pre-existing', 'inbox', 1, '2026-04-20 12:00:00')",
        )
        .execute(&pool)
        .await
        .expect("insert");

        let auto: i64 = sqlx::query_scalar("SELECT auto_completed FROM tasks WHERE id = 't1'")
            .fetch_one(&pool)
            .await
            .expect("query");
        assert_eq!(auto, 0, "pre-existing completed rows must default to 0");
    }

    #[tokio::test]
    async fn migration_v15_creates_profile_created_index() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .expect("pool");
        run_migrations(&pool).await.expect("migrate");

        // Index exists by name.
        let names: Vec<String> = sqlx::query_scalar(
            "SELECT name FROM sqlite_master \
             WHERE type = 'index' AND tbl_name = 'transcripts'",
        )
        .fetch_all(&pool)
        .await
        .expect("read indexes");
        assert!(
            names.iter().any(|n| n == "idx_transcripts_profile_created"),
            "expected composite (profile_id, created_at) index, got {names:?}",
        );

        // Query planner picks the composite index for the dominant
        // profile-scoped, date-ordered list query. EXPLAIN QUERY PLAN
        // returns (id, parent, notused, detail) — we want detail.
        let plan_rows = sqlx::query(
            "EXPLAIN QUERY PLAN \
             SELECT id FROM transcripts \
             WHERE profile_id = ? ORDER BY created_at DESC LIMIT 50",
        )
        .bind(crate::DEFAULT_PROFILE_ID)
        .fetch_all(&pool)
        .await
        .expect("explain");
        let plan: Vec<String> = plan_rows
            .iter()
            .map(|r| r.get::<String, _>("detail"))
            .collect();
        assert!(
            plan.iter()
                .any(|row| row.contains("idx_transcripts_profile_created")),
            "planner should use the composite index, got plan: {plan:?}",
        );
    }

    #[tokio::test]
    async fn migration_v17_creates_onboarding_and_lumotia_events_tables() {
        // v0.1 onboarding-event tracking and opt-in local activation log.
        // Verify both tables exist with the expected columns after running
        // all migrations. Uses PRAGMA table_info to inspect column presence.
        let pool = fk_test_pool().await;
        run_migrations(&pool).await.expect("migrate");

        // --- onboarding_events ---
        let info = sqlx::query("PRAGMA table_info(onboarding_events)")
            .fetch_all(&pool)
            .await
            .expect("pragma onboarding_events");
        assert!(
            !info.is_empty(),
            "onboarding_events table must exist after v17"
        );
        let names: Vec<String> = info.iter().map(|r| r.get::<String, _>("name")).collect();
        for col in ["id", "event", "completed_at", "version", "skipped", "notes"] {
            assert!(
                names.contains(&col.to_string()),
                "onboarding_events must have column {col}; got {names:?}"
            );
        }

        // Index on onboarding_events(event) must exist.
        let idx_names: Vec<String> = sqlx::query_scalar(
            "SELECT name FROM sqlite_master \
             WHERE type = 'index' AND tbl_name = 'onboarding_events'",
        )
        .fetch_all(&pool)
        .await
        .expect("read onboarding_events indexes");
        assert!(
            idx_names.iter().any(|n| n == "idx_onboarding_events_event"),
            "expected idx_onboarding_events_event, got {idx_names:?}",
        );

        // --- lumotia_events ---
        let info2 = sqlx::query("PRAGMA table_info(lumotia_events)")
            .fetch_all(&pool)
            .await
            .expect("pragma lumotia_events");
        assert!(
            !info2.is_empty(),
            "lumotia_events table must exist after v17"
        );
        let names2: Vec<String> = info2.iter().map(|r| r.get::<String, _>("name")).collect();
        for col in ["id", "kind", "occurred_at", "payload"] {
            assert!(
                names2.contains(&col.to_string()),
                "lumotia_events must have column {col}; got {names2:?}"
            );
        }

        // Both indexes on lumotia_events must exist.
        let idx_names2: Vec<String> = sqlx::query_scalar(
            "SELECT name FROM sqlite_master \
             WHERE type = 'index' AND tbl_name = 'lumotia_events'",
        )
        .fetch_all(&pool)
        .await
        .expect("read lumotia_events indexes");
        for idx in ["idx_lumotia_events_kind", "idx_lumotia_events_occurred"] {
            assert!(
                idx_names2.iter().any(|n| n == idx),
                "expected index {idx}, got {idx_names2:?}",
            );
        }
    }

    #[tokio::test]
    async fn migration_v16_adds_deleted_at_column_and_index() {
        // Rev-2 / Rev-3 atomiser fix (2026-05-12). Verify the soft-delete
        // column is present, nullable, defaulting to NULL on pre-existing
        // rows so the migration doesn't accidentally mark old transcripts
        // as deleted.
        let pool = fk_test_pool().await;
        run_migrations_up_to(&pool, 15)
            .await
            .expect("migrate to v15");

        // Seed a pre-v16 row to verify backfill preserves NULL.
        sqlx::query(
            "INSERT INTO transcripts (
                id, text, source, profile_id, title, audio_path, duration,
                engine, model_id, inference_ms, sample_rate, audio_channels,
                format_mode, remove_fillers, british_english, anti_hallucination,
                created_at, starred, manual_tags, template, language,
                segments_json, llm_tags
            ) VALUES (
                'pre-v16', 'pre-existing body', 'microphone', ?, 'Pre-V16',
                NULL, 1.0, NULL, NULL, NULL, NULL, NULL, NULL, 0, 0, 0,
                datetime('now'), 0, '', '', '', '', ''
            )",
        )
        .bind(crate::DEFAULT_PROFILE_ID)
        .execute(&pool)
        .await
        .expect("seed pre-v16 row");

        run_migrations(&pool).await.expect("migrate to v16");

        let info = sqlx::query("PRAGMA table_info(transcripts)")
            .fetch_all(&pool)
            .await
            .expect("pragma");
        let names: Vec<String> = info.iter().map(|r| r.get::<String, _>("name")).collect();
        assert!(
            names.contains(&"deleted_at".to_string()),
            "transcripts must have deleted_at after v16; got {names:?}",
        );

        // Pre-existing rows must default to NULL — emphatically NOT
        // pre-soft-deleted.
        let deleted_at: Option<String> =
            sqlx::query_scalar("SELECT deleted_at FROM transcripts WHERE id = 'pre-v16'")
                .fetch_one(&pool)
                .await
                .expect("read deleted_at");
        assert!(
            deleted_at.is_none(),
            "pre-v16 rows must have deleted_at = NULL after migration, got {deleted_at:?}",
        );

        // Partial index exists.
        let index_names: Vec<String> = sqlx::query_scalar(
            "SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'transcripts'",
        )
        .fetch_all(&pool)
        .await
        .expect("read indexes");
        assert!(
            index_names
                .iter()
                .any(|n| n == "idx_transcripts_deleted_at"),
            "expected idx_transcripts_deleted_at, got {index_names:?}",
        );
    }
}