diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index a386ca8..143ed57 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -12,3 +12,6 @@ sqlx = { version = "0.8", features = ["sqlite", "runtime-tokio"] } # Async runtime tokio = { version = "1", features = ["rt", "sync", "macros"] } + +# Logging +log = "0.4" diff --git a/crates/storage/src/database.rs b/crates/storage/src/database.rs index ef23166..2bb5f87 100644 --- a/crates/storage/src/database.rs +++ b/crates/storage/src/database.rs @@ -26,119 +26,9 @@ pub async fn init(db_path: &Path) -> Result { Ok(pool) } -/// Run schema migrations. -/// -/// TODO(pre-release): Implement schema versioning for safe upgrades. -/// Currently uses `CREATE TABLE IF NOT EXISTS` which silently ignores column -/// additions/removals. Before shipping, add a `schema_version` table and -/// sequential migration steps so existing user databases upgrade cleanly. +/// Run schema migrations via the versioned migration system. async fn run_migrations(pool: &SqlitePool) -> Result<()> { - sqlx::query( - "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')) - )", - ) - .execute(pool) - .await - .map_err(|e| KonError::StorageError(format!("Migration failed: {e}")))?; - - sqlx::query( - "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 '' - )", - ) - .execute(pool) - .await - .map_err(|e| KonError::StorageError(format!("Migration failed: {e}")))?; - - sqlx::query( - "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 - )", - ) - .execute(pool) - .await - .map_err(|e| KonError::StorageError(format!("Migration failed: {e}")))?; - - sqlx::query( - "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')) - )", - ) - .execute(pool) - .await - .map_err(|e| KonError::StorageError(format!("Migration failed: {e}")))?; - - sqlx::query( - "CREATE TABLE IF NOT EXISTS settings ( - key TEXT PRIMARY KEY, - value TEXT NOT NULL - )", - ) - .execute(pool) - .await - .map_err(|e| KonError::StorageError(format!("Migration failed: {e}")))?; - - sqlx::query( - "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 - )", - ) - .execute(pool) - .await - .map_err(|e| KonError::StorageError(format!("Migration failed: {e}")))?; - - // Indices for frequently queried columns - for ddl in [ - "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)", - ] { - sqlx::query(ddl) - .execute(pool) - .await - .map_err(|e| KonError::StorageError(format!("Index creation failed: {e}")))?; - } - - Ok(()) + crate::migrations::run_migrations(pool).await } // --- Transcript CRUD --- diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 5703f39..b445049 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -1,5 +1,6 @@ pub mod database; pub mod file_storage; +pub mod migrations; pub use database::{ complete_task, delete_task, delete_transcript, get_setting, get_transcript, init, insert_task, diff --git a/crates/storage/src/migrations.rs b/crates/storage/src/migrations.rs new file mode 100644 index 0000000..2af4269 --- /dev/null +++ b/crates/storage/src/migrations.rs @@ -0,0 +1,168 @@ +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); + } +}