jake/Lumotia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kon): add schema versioning — sequential migration system replaces CREATE TABLE IF NOT EXISTS

Browse Source 
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jake
2026-03-17 02:13:19 +00:00
parent e70cfca63d
commit b9178f8117
4 changed files with 174 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
crates/storage/Cargo.toml
View File

@@ -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"

114
crates/storage/src/database.rs
View File

@@ -26,119 +26,9 @@ pub async fn init(db_path: &Path) -> Result<SqlitePool> {
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 ---

1
crates/storage/src/lib.rs
View File

@@ -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,

168
crates/storage/src/migrations.rs Normal file
View File

@@ -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);
}
}