use std::path::Path; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use sqlx::{Row, SqlitePool}; use kon_core::error::{KonError, Result}; /// Initialise the SQLite database with connection pool and run migrations. pub async fn init(db_path: &Path) -> Result { if let Some(parent) = db_path.parent() { std::fs::create_dir_all(parent)?; } let options = SqliteConnectOptions::new() .filename(db_path) .create_if_missing(true); let pool = SqlitePoolOptions::new() .max_connections(5) .connect_with(options) .await .map_err(|e| KonError::StorageError(format!("Database connect failed: {e}")))?; run_migrations(&pool).await?; Ok(pool) } /// Run schema migrations. 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(()) } // --- Transcript CRUD --- /// Parameters for inserting a transcript with full provenance. pub struct InsertTranscriptParams<'a> { pub id: &'a str, pub text: &'a str, pub source: &'a str, pub title: Option<&'a str>, pub audio_path: Option<&'a str>, pub duration: f64, pub engine: Option<&'a str>, pub model_id: Option<&'a str>, pub inference_ms: Option, pub sample_rate: Option, pub audio_channels: Option, pub format_mode: Option<&'a str>, pub remove_fillers: bool, pub british_english: bool, pub anti_hallucination: bool, } pub async fn insert_transcript( pool: &SqlitePool, params: &InsertTranscriptParams<'_>, ) -> Result<()> { 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) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(params.id) .bind(params.text) .bind(params.source) .bind(params.title) .bind(params.audio_path) .bind(params.duration) .bind(params.engine) .bind(params.model_id) .bind(params.inference_ms) .bind(params.sample_rate) .bind(params.audio_channels) .bind(params.format_mode) .bind(params.remove_fillers) .bind(params.british_english) .bind(params.anti_hallucination) .execute(pool) .await .map_err(|e| KonError::StorageError(format!("Insert transcript failed: {e}")))?; Ok(()) } pub async fn get_transcript( pool: &SqlitePool, id: &str, ) -> Result> { let row = sqlx::query( "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 FROM transcripts WHERE id = ?", ) .bind(id) .fetch_optional(pool) .await .map_err(|e| KonError::StorageError(format!("Get transcript failed: {e}")))?; Ok(row.map(|r| transcript_row_from(&r))) } pub async fn list_transcripts(pool: &SqlitePool, limit: i64) -> Result> { let rows = sqlx::query( "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 FROM transcripts ORDER BY created_at DESC LIMIT ?", ) .bind(limit) .fetch_all(pool) .await .map_err(|e| KonError::StorageError(format!("List transcripts failed: {e}")))?; Ok(rows.iter().map(transcript_row_from).collect()) } pub async fn delete_transcript(pool: &SqlitePool, id: &str) -> Result<()> { sqlx::query("DELETE FROM transcripts WHERE id = ?") .bind(id) .execute(pool) .await .map_err(|e| KonError::StorageError(format!("Delete transcript failed: {e}")))?; Ok(()) } // --- Task CRUD --- pub async fn insert_task( pool: &SqlitePool, id: &str, text: &str, bucket: &str, source_transcript_id: Option<&str>, ) -> Result<()> { sqlx::query( "INSERT INTO tasks (id, text, bucket, source_transcript_id) VALUES (?, ?, ?, ?)", ) .bind(id) .bind(text) .bind(bucket) .bind(source_transcript_id) .execute(pool) .await .map_err(|e| KonError::StorageError(format!("Insert task failed: {e}")))?; Ok(()) } pub async fn list_tasks(pool: &SqlitePool) -> Result> { let rows = sqlx::query("SELECT id, text, bucket, list_id, effort, done, done_at, created_at, source_transcript_id FROM tasks ORDER BY created_at DESC") .fetch_all(pool) .await .map_err(|e| KonError::StorageError(format!("List tasks failed: {e}")))?; Ok(rows .into_iter() .map(|r| TaskRow { id: r.get("id"), text: r.get("text"), bucket: r.get("bucket"), list_id: r.get("list_id"), effort: r.get("effort"), done: r.get("done"), done_at: r.get("done_at"), created_at: r.get("created_at"), source_transcript_id: r.get("source_transcript_id"), }) .collect()) } pub async fn complete_task(pool: &SqlitePool, id: &str) -> Result<()> { sqlx::query("UPDATE tasks SET done = 1, done_at = datetime('now') WHERE id = ?") .bind(id) .execute(pool) .await .map_err(|e| KonError::StorageError(format!("Complete task failed: {e}")))?; Ok(()) } pub async fn delete_task(pool: &SqlitePool, id: &str) -> Result<()> { sqlx::query("DELETE FROM tasks WHERE id = ?") .bind(id) .execute(pool) .await .map_err(|e| KonError::StorageError(format!("Delete task failed: {e}")))?; Ok(()) } // --- Settings CRUD --- pub async fn set_setting(pool: &SqlitePool, key: &str, value: &str) -> Result<()> { sqlx::query("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)") .bind(key) .bind(value) .execute(pool) .await .map_err(|e| KonError::StorageError(format!("Set setting failed: {e}")))?; Ok(()) } pub async fn get_setting(pool: &SqlitePool, key: &str) -> Result> { let row = sqlx::query("SELECT value FROM settings WHERE key = ?") .bind(key) .fetch_optional(pool) .await .map_err(|e| KonError::StorageError(format!("Get setting failed: {e}")))?; Ok(row.map(|r| r.get("value"))) } // --- Row types --- #[derive(Debug, Clone)] pub struct TranscriptRow { pub id: String, pub text: String, pub source: String, pub title: Option, pub audio_path: Option, pub duration: f64, pub engine: Option, pub model_id: Option, pub inference_ms: Option, pub sample_rate: Option, pub audio_channels: Option, pub format_mode: Option, pub remove_fillers: bool, pub british_english: bool, pub anti_hallucination: bool, pub created_at: String, } #[derive(Debug, Clone)] pub struct TaskRow { pub id: String, pub text: String, pub bucket: String, pub list_id: Option, pub effort: Option, pub done: bool, pub done_at: Option, pub created_at: String, pub source_transcript_id: Option, } fn transcript_row_from(r: &sqlx::sqlite::SqliteRow) -> TranscriptRow { TranscriptRow { id: r.get("id"), text: r.get("text"), source: r.get("source"), title: r.get("title"), audio_path: r.get("audio_path"), duration: r.get("duration"), engine: r.get("engine"), model_id: r.get("model_id"), inference_ms: r.get("inference_ms"), sample_rate: r.get("sample_rate"), audio_channels: r.get("audio_channels"), format_mode: r.get("format_mode"), remove_fillers: r.get("remove_fillers"), british_english: r.get("british_english"), anti_hallucination: r.get("anti_hallucination"), created_at: r.get("created_at"), } } // --- Error Logging --- /// Log a structured error to the error_log table. pub async fn log_error( pool: &SqlitePool, context: &str, error_code: Option<&str>, message: &str, metadata: Option<&str>, ) -> Result<()> { sqlx::query( "INSERT INTO error_log (context, error_code, message, metadata) VALUES (?, ?, ?, ?)", ) .bind(context) .bind(error_code) .bind(message) .bind(metadata) .execute(pool) .await .map_err(|e| KonError::StorageError(format!("Error log failed: {e}")))?; Ok(()) } #[cfg(test)] mod tests { use super::*; async fn test_pool() -> SqlitePool { let pool = SqlitePoolOptions::new() .connect("sqlite::memory:") .await .unwrap(); run_migrations(&pool).await.unwrap(); pool } #[tokio::test] async fn transcript_crud_roundtrip() { let pool = test_pool().await; insert_transcript( &pool, &InsertTranscriptParams { id: "t1", text: "Hello world", source: "microphone", title: Some("Test"), audio_path: None, duration: 1.5, engine: Some("whisper"), model_id: Some("whisper-tiny-en"), inference_ms: Some(250), sample_rate: Some(48000), audio_channels: Some(1), format_mode: Some("Clean"), remove_fillers: true, british_english: true, anti_hallucination: false, }, ) .await .unwrap(); let t = get_transcript(&pool, "t1").await.unwrap().unwrap(); assert_eq!(t.text, "Hello world"); assert_eq!(t.source, "microphone"); assert_eq!(t.duration, 1.5); assert_eq!(t.engine.as_deref(), Some("whisper")); assert_eq!(t.model_id.as_deref(), Some("whisper-tiny-en")); assert_eq!(t.inference_ms, Some(250)); assert!(t.remove_fillers); assert!(t.british_english); let list = list_transcripts(&pool, 10).await.unwrap(); assert_eq!(list.len(), 1); delete_transcript(&pool, "t1").await.unwrap(); let deleted = get_transcript(&pool, "t1").await.unwrap(); assert!(deleted.is_none()); } #[tokio::test] async fn task_crud_roundtrip() { let pool = test_pool().await; insert_task(&pool, "task1", "Buy groceries", "today", None) .await .unwrap(); let tasks = list_tasks(&pool).await.unwrap(); assert_eq!(tasks.len(), 1); assert_eq!(tasks[0].text, "Buy groceries"); assert!(!tasks[0].done); complete_task(&pool, "task1").await.unwrap(); let tasks = list_tasks(&pool).await.unwrap(); assert!(tasks[0].done); delete_task(&pool, "task1").await.unwrap(); let tasks = list_tasks(&pool).await.unwrap(); assert!(tasks.is_empty()); } #[tokio::test] async fn settings_crud_roundtrip() { let pool = test_pool().await; set_setting(&pool, "theme", "dark").await.unwrap(); let val = get_setting(&pool, "theme").await.unwrap(); assert_eq!(val.as_deref(), Some("dark")); set_setting(&pool, "theme", "light").await.unwrap(); let val = get_setting(&pool, "theme").await.unwrap(); assert_eq!(val.as_deref(), Some("light")); let missing = get_setting(&pool, "nonexistent").await.unwrap(); assert!(missing.is_none()); } }