Files
Lumotia/crates/storage/src/migrations.rs
2026-03-21 14:43:06 +00:00

289 lines
10 KiB
Rust

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)
"#),
(2, "phase 2 — task fields, FTS5, timer state", r#"
ALTER TABLE tasks ADD COLUMN priority TEXT NOT NULL DEFAULT 'medium';
ALTER TABLE tasks ADD COLUMN project TEXT;
ALTER TABLE tasks ADD COLUMN status TEXT NOT NULL DEFAULT 'pending';
ALTER TABLE tasks ADD COLUMN updated_at TEXT NOT NULL DEFAULT (datetime('now'));
ALTER TABLE tasks ADD COLUMN sort_order INTEGER NOT NULL DEFAULT 0;
ALTER TABLE tasks ADD COLUMN notes TEXT NOT NULL DEFAULT '';
CREATE VIRTUAL TABLE IF NOT EXISTS transcripts_fts USING fts5(
text,
title,
content='transcripts',
content_rowid='rowid'
);
CREATE TRIGGER IF NOT EXISTS transcripts_ai AFTER INSERT ON transcripts BEGIN
INSERT INTO transcripts_fts(rowid, text, title)
VALUES (new.rowid, new.text, 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, 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, old.title);
INSERT INTO transcripts_fts(rowid, text, title)
VALUES (new.rowid, new.text, new.title);
END;
CREATE TABLE IF NOT EXISTS timer_state (
id TEXT PRIMARY KEY DEFAULT 'active',
task_id TEXT NOT NULL,
total_seconds INTEGER NOT NULL,
remaining_seconds INTEGER NOT NULL,
started_at TEXT NOT NULL DEFAULT (datetime('now')),
paused INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
CREATE INDEX IF NOT EXISTS idx_tasks_priority ON tasks(priority);
CREATE INDEX IF NOT EXISTS idx_tasks_project ON tasks(project)
"#),
(3, "phase 2b — parent_task_id for micro-stepping", r#"
ALTER TABLE tasks ADD COLUMN parent_task_id TEXT;
CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_task_id)
"#),
];
/// Split SQL text into individual statements, respecting BEGIN...END blocks
/// used by CREATE TRIGGER statements. A naive split on ';' would break
/// triggers that contain semicolons inside their body.
fn split_sql_statements(sql: &str) -> Vec<String> {
let mut statements = Vec::new();
let mut current = String::new();
let mut depth = 0; // Track BEGIN...END nesting
for fragment in sql.split(';') {
let trimmed = fragment.trim();
if trimmed.is_empty() {
continue;
}
let upper = trimmed.to_uppercase();
// Count BEGIN/END keywords to track trigger body nesting
if upper.contains("BEGIN") {
depth += 1;
}
if depth > 0 {
// Inside a BEGIN...END block — accumulate fragments
if !current.is_empty() {
current.push(';');
}
current.push_str(fragment);
if upper.trim_end().ends_with("END") {
depth -= 1;
if depth == 0 {
let stmt = current.trim().to_string();
if !stmt.is_empty() {
statements.push(stmt);
}
current.clear();
}
}
} else {
// Normal statement — emit directly
statements.push(trimmed.to_string());
}
}
// Emit anything left over
let leftover = current.trim().to_string();
if !leftover.is_empty() {
statements.push(leftover);
}
statements
}
/// 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 = split_sql_statements(sql);
for statement in &statements {
sqlx::query(statement)
.execute(pool)
.await
.map_err(|e| KonError::StorageError(format!("Migration {} failed on statement: {e}\nSQL: {}", version, statement)))?;
}
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, 3);
sqlx::query("INSERT INTO settings (key, value) VALUES ('test', 'value')")
.execute(&pool)
.await
.unwrap();
}
#[test]
fn test_split_sql_handles_triggers() {
let sql = r#"
ALTER TABLE tasks ADD COLUMN priority TEXT;
CREATE TRIGGER IF NOT EXISTS test_trigger AFTER INSERT ON tasks BEGIN
INSERT INTO log(msg) VALUES (new.id);
END;
CREATE TABLE IF NOT EXISTS timer (id TEXT PRIMARY KEY)
"#;
let stmts = split_sql_statements(sql);
assert_eq!(stmts.len(), 3, "Expected 3 statements, got: {stmts:#?}");
assert!(stmts[0].contains("ALTER TABLE"));
assert!(stmts[1].contains("CREATE TRIGGER"));
assert!(stmts[1].contains("BEGIN"));
assert!(stmts[1].contains("END"));
assert!(stmts[2].contains("CREATE TABLE"));
}
#[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, 3);
}
}