//! Shared sqlite connection setup and versioned schema migrations for //! hive-c0re's host-side stores. //! //! All stores share one sqlite file (`db/broker.sqlite`) but each owns //! its own `Mutex`. [`open`] handles the connection open //! dance; [`apply_versioned_migrations`] handles schema evolution. //! //! ## Migration strategy //! //! Each store calls [`apply_versioned_migrations`] from its `open`. It //! tracks the applied count in a `schema_versions` table (one row per //! `subsystem` key). Only migrations past the stored version run. //! //! Legacy databases (written before versioning) need no special marker: //! each column-adding migration carries the `(table, column)` it //! introduces, so re-walking a pre-versioning DB from version 0 simply //! skips the migrations whose column already exists — converging a fully- //! or partially-migrated legacy DB without ever re-running an `ADD COLUMN` //! against an existing column. See each store's `MIGRATIONS` constant. use std::path::Path; use std::time::Duration; use anyhow::{Context, Result}; use rusqlite::{Connection, OptionalExtension, params}; /// How long a write waits on another connection's lock before erroring. /// Generous relative to the stores' tiny transactions — a timeout here /// means something is genuinely wedged, not ordinary contention. const BUSY_TIMEOUT: Duration = Duration::from_secs(5); const SCHEMA_VERSIONS_DDL: &str = " CREATE TABLE IF NOT EXISTS schema_versions ( store TEXT PRIMARY KEY, version INTEGER NOT NULL DEFAULT 0 )"; /// One numbered schema migration. /// /// `sql` runs once, at the index whose value is the current stored version. /// `adds_column`, when set, is the `(table, column)` this migration /// introduces: if that column already exists — a pre-versioning DB migrated /// by the old try-and-ignore path, or a partially-migrated one — `sql` is /// skipped and only the version counter advances. This makes column-adding /// migrations idempotent *without* `SQLite`'s unsupported `ALTER TABLE … ADD /// COLUMN IF NOT EXISTS`, so re-walking a legacy DB from version 0 never /// re-runs an `ADD COLUMN` against a column that is already there (which /// would fail with "duplicate column name" and wedge startup permanently). /// /// Multi-statement migrations wrap their `ALTER` + backfill / index rebuild /// in an explicit `BEGIN; … COMMIT;` batch, so "the added column exists" is a /// faithful proxy for "this whole migration committed". pub struct Migration { /// SQL for this version — a single statement or a `BEGIN; … COMMIT;` batch. pub sql: &'static str, /// `(table, column)` this migration adds, if any. Present → the migration /// is skipped when the column already exists (idempotent replay). pub adds_column: Option<(&'static str, &'static str)>, } /// Apply numbered schema migrations for `subsystem`, tracking progress in /// the shared `schema_versions` table. /// /// - `subsystem` — unique key for this store in `schema_versions` /// (e.g. `"approvals"`, `"broker"`). /// - `migrations` — ordered list of [`Migration`]s; migration `i` runs when /// the stored version is `i`. A pre-versioning DB (no `schema_versions` /// row) starts at version 0 and re-walks every migration, but each /// column-adding migration is skipped when its `adds_column` is already /// present — so a fully- or partially-migrated legacy DB converges without /// re-running an `ADD COLUMN` against an existing column. /// /// # Errors /// /// Returns an error if any migration statement fails or the version update /// fails. Each migration + its version bump is a single `execute_batch` /// (multi-statement migrations wrap themselves in `BEGIN; … COMMIT;`), so a /// failure leaves the DB at the last successfully committed version. pub fn apply_versioned_migrations( conn: &Connection, subsystem: &str, migrations: &[Migration], ) -> Result<()> { // Ensure the version-tracking table exists (idempotent). conn.execute_batch(SCHEMA_VERSIONS_DDL) .context("create schema_versions table")?; // Current version for this subsystem — 0 (and a fresh row) when absent. let stored: Option = conn .query_row( "SELECT version FROM schema_versions WHERE store = ?1", [subsystem], |row| row.get(0), ) .optional() .context("read schema_versions")?; let version = if let Some(v) = stored { usize::try_from(v.max(0)).unwrap_or(0) } else { conn.execute( "INSERT INTO schema_versions (store, version) VALUES (?1, 0)", [subsystem], ) .context("insert schema_versions row")?; 0 }; if version >= migrations.len() { return Ok(()); } for (i, mig) in migrations.iter().enumerate().skip(version) { // Skip a column-adding migration whose column is already present: // a legacy DB migrated by the old path, or a resumed partial one. let already_applied = match mig.adds_column { Some((table, column)) => column_exists(conn, table, column)?, None => false, }; if !already_applied { conn.execute_batch(mig.sql) .with_context(|| format!("{subsystem} migration v{} failed: {}", i + 1, mig.sql))?; } let next = i64::try_from(i + 1).unwrap_or(i64::MAX); conn.execute( "UPDATE schema_versions SET version = ?1 WHERE store = ?2", params![next, subsystem], ) .with_context(|| format!("{subsystem} update schema_versions to v{}", i + 1))?; } Ok(()) } /// Whether `column` exists on `table` in this connection's schema. Both /// arguments are bound parameters — `pragma_table_info(?1)` accepts the table /// name as a bound value — so this is not a SQL-injection surface even if a /// caller ever passes a non-static name. fn column_exists(conn: &Connection, table: &str, column: &str) -> Result { conn.prepare("SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2") .context("prepare column probe")? .exists(params![table, column]) .context("check column existence") } /// Open a connection to the sqlite file at `path`, creating the parent /// directory if needed. `subsystem` labels error contexts (`"broker"`, /// `"approvals"`, …). pub fn open(path: &Path, subsystem: &str) -> Result { if let Some(parent) = path.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("create {subsystem} db parent {}", parent.display()))?; } let conn = Connection::open(path) .with_context(|| format!("open {subsystem} db {}", path.display()))?; conn.busy_timeout(BUSY_TIMEOUT) .with_context(|| format!("set {subsystem} busy_timeout"))?; Ok(conn) } #[cfg(test)] mod tests { use super::*; use std::sync::atomic::{AtomicU64, Ordering}; static TEST_COUNTER: AtomicU64 = AtomicU64::new(0); fn open_tmp() -> Connection { let n = TEST_COUNTER.fetch_add(1, Ordering::Relaxed); let path = format!("/tmp/hive-db-test-{}-{}.sqlite", std::process::id(), n); Connection::open(&path).expect("open tmp db") } const TEST_SCHEMA: &str = "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY)"; const TEST_MIGRATIONS: &[Migration] = &[ Migration { sql: "ALTER TABLE things ADD COLUMN alpha TEXT", adds_column: Some(("things", "alpha")), }, Migration { sql: "ALTER TABLE things ADD COLUMN beta INTEGER NOT NULL DEFAULT 0", adds_column: Some(("things", "beta")), }, Migration { sql: "ALTER TABLE things ADD COLUMN gamma TEXT", adds_column: Some(("things", "gamma")), }, ]; #[test] fn fresh_install_runs_all_migrations() { let conn = open_tmp(); conn.execute_batch(TEST_SCHEMA).unwrap(); apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap(); // All three columns should exist. for col in ["alpha", "beta", "gamma"] { let exists: bool = conn .prepare(&format!( "SELECT 1 FROM pragma_table_info('things') WHERE name = '{col}'" )) .unwrap() .exists([]) .unwrap(); assert!(exists, "column {col} missing after fresh migration"); } // Version should be 3. let v: i64 = conn .query_row( "SELECT version FROM schema_versions WHERE store = 'things'", [], |r| r.get(0), ) .unwrap(); assert_eq!(v, 3); } #[test] fn legacy_install_skips_all_migrations() { let conn = open_tmp(); // Simulate a fully-migrated legacy DB: all columns exist, // but schema_versions does not yet. conn.execute_batch( "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY, alpha TEXT, beta INTEGER NOT NULL DEFAULT 0, gamma TEXT)", ) .unwrap(); apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap(); // Version should be set to migrations.len() (3), not 0. let v: i64 = conn .query_row( "SELECT version FROM schema_versions WHERE store = 'things'", [], |r| r.get(0), ) .unwrap(); assert_eq!(v, 3, "legacy install should skip to latest version"); } #[test] fn partial_migration_resumes_from_version() { let conn = open_tmp(); // Simulate a DB that has only the first migration applied. conn.execute_batch( "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY, alpha TEXT); CREATE TABLE IF NOT EXISTS schema_versions (store TEXT PRIMARY KEY, version INTEGER NOT NULL DEFAULT 0); INSERT INTO schema_versions (store, version) VALUES ('things', 1)", ) .unwrap(); apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap(); // Only beta and gamma should have been added (alpha already existed). for col in ["alpha", "beta", "gamma"] { let exists: bool = conn .prepare(&format!( "SELECT 1 FROM pragma_table_info('things') WHERE name = '{col}'" )) .unwrap() .exists([]) .unwrap(); assert!( exists, "column {col} missing after partial migration resume" ); } let v: i64 = conn .query_row( "SELECT version FROM schema_versions WHERE store = 'things'", [], |r| r.get(0), ) .unwrap(); assert_eq!(v, 3); } #[test] fn already_at_latest_is_noop() { let conn = open_tmp(); conn.execute_batch(TEST_SCHEMA).unwrap(); apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap(); // Second call is a no-op. apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap(); let v: i64 = conn .query_row( "SELECT version FROM schema_versions WHERE store = 'things'", [], |r| r.get(0), ) .unwrap(); assert_eq!(v, 3); } #[test] fn multiple_stores_in_same_db() { let conn = open_tmp(); conn.execute_batch( "CREATE TABLE IF NOT EXISTS a (id INTEGER PRIMARY KEY); CREATE TABLE IF NOT EXISTS b (id INTEGER PRIMARY KEY)", ) .unwrap(); apply_versioned_migrations( &conn, "store_a", &[Migration { sql: "ALTER TABLE a ADD COLUMN col1 TEXT", adds_column: Some(("a", "col1")), }], ) .unwrap(); apply_versioned_migrations( &conn, "store_b", &[Migration { sql: "ALTER TABLE b ADD COLUMN colx INTEGER NOT NULL DEFAULT 0", adds_column: Some(("b", "colx")), }], ) .unwrap(); // Each store tracks independently. let va: i64 = conn .query_row( "SELECT version FROM schema_versions WHERE store = 'store_a'", [], |r| r.get(0), ) .unwrap(); let vb: i64 = conn .query_row( "SELECT version FROM schema_versions WHERE store = 'store_b'", [], |r| r.get(0), ) .unwrap(); assert_eq!(va, 1); assert_eq!(vb, 1); } /// A partially-migrated legacy database: some columns from the old /// try-and-ignore path exist, with no `schema_versions` row yet. The /// per-migration `adds_column` guard must skip the already-present /// columns and add only the missing ones — no "duplicate column name". #[test] fn partial_legacy_completes_without_error() { let conn = open_tmp(); // Simulate a DB that has `alpha` (v1) but not `beta` (v2) or // `gamma` (v3). No schema_versions row exists → start from v0 and // guard-skip alpha while adding beta + gamma. conn.execute_batch( "CREATE TABLE IF NOT EXISTS things (id INTEGER PRIMARY KEY, alpha TEXT)", ) .unwrap(); apply_versioned_migrations(&conn, "things", TEST_MIGRATIONS).unwrap(); // All three columns must now exist. for col in ["alpha", "beta", "gamma"] { let exists: bool = conn .prepare(&format!( "SELECT 1 FROM pragma_table_info('things') WHERE name = '{col}'" )) .unwrap() .exists([]) .unwrap(); assert!( exists, "column {col} missing after partial-legacy migration" ); } // Version must be at the latest. let v: i64 = conn .query_row( "SELECT version FROM schema_versions WHERE store = 'things'", [], |r| r.get(0), ) .unwrap(); assert_eq!( v, 3, "must reach latest version after completing partial legacy" ); } }