Lumotia/crates/storage/src/migrations.rs

use sqlx::SqlitePool;
use kon_core::error::{KonError, Result};

/// 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)
    "#),
];

/// Ensure the schema_version table exists and run any pending migrations.
pub async fn run_migrations(pool: &SqlitePool) -> 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(|e| KonError::StorageError(format!("Schema version table creation failed: {e}")))?;

    let current: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(version), 0) FROM schema_version")
        .fetch_one(pool)
        .await
        .map_err(|e| KonError::StorageError(format!("Schema version query failed: {e}")))?;

    for (version, description, sql) in MIGRATIONS {
        if *version > current {
            log::info!("Running migration {}: {}", version, description);

            let statements: Vec<&str> = sql.split(';')
                .map(|s| s.trim())
                .filter(|s| !s.is_empty())
                .collect();

            for statement in statements {
                sqlx::query(statement)
                    .execute(pool)
                    .await
                    .map_err(|e| KonError::StorageError(format!("Migration {} failed: {e}", version)))?;
            }

            sqlx::query("INSERT INTO schema_version (version, description) VALUES (?, ?)")
                .bind(version)
                .bind(description)
                .execute(pool)
                .await
                .map_err(|e| KonError::StorageError(format!("Migration version record failed: {e}")))?;

            log::info!("Migration {} complete", version);
        }
    }

    Ok(())
}

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

    #[tokio::test]
    async fn test_migrations_run_on_empty_db() {
        let pool = SqlitePoolOptions::new()
            .connect("sqlite::memory:")
            .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, 1);

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

    #[tokio::test]
    async fn test_migrations_idempotent() {
        let pool = SqlitePoolOptions::new()
            .connect("sqlite::memory:")
            .await
            .unwrap();

        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, 1);
    }
}