//! 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) are detected via //! `legacy_column = (table, column)` — a column that exists only in //! a fully-migrated legacy DB. Present → skip all migrations. Absent → //! start from 0. See each store's `MIGRATIONS` constant for the list. 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 )"; /// 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"`). /// - `legacy_column` — `(table_name, column_name)` of a column that exists /// in a fully-migrated pre-versioning database. When there is no /// `schema_versions` row yet and this column is present, all `migrations` /// are skipped (they were previously applied via the old try-and-ignore /// approach). When the column is absent, we start at version 0. /// - `migrations` — ordered list of SQL batches; each batch runs once, at /// the index whose value is the current version. /// /// # Errors /// /// Returns an error if any migration statement fails or the version update /// fails. A failed migration leaves the DB at the last successfully /// committed version (each step is its own implicit transaction via /// `execute_batch`). pub fn apply_versioned_migrations( conn: &Connection, subsystem: &str, legacy_column: (&str, &str), migrations: &[&str], ) -> Result<()> { // Ensure the version-tracking table exists (idempotent). conn.execute_batch(SCHEMA_VERSIONS_DDL) .context("create schema_versions table")?; // Look up the stored version for this subsystem. 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 { // No entry yet. Check the legacy-column to decide where to start. // If it exists, all known migrations were already applied by the // old try-and-ignore path — skip them. If it doesn't exist, start // at 0 (fresh install or partial state). let (legacy_table, legacy_col) = legacy_column; let is_legacy: bool = conn .prepare(&format!( "SELECT 1 FROM pragma_table_info('{legacy_table}') WHERE name = ?1" )) .context("prepare legacy-column probe")? .exists([legacy_col]) .context("check legacy column")?; let initial = if is_legacy { migrations.len() } else { 0 }; conn.execute( "INSERT INTO schema_versions (store, version) VALUES (?1, ?2)", params![subsystem, i64::try_from(initial).unwrap_or(0)], ) .context("insert schema_versions row")?; initial }; if version >= migrations.len() { return Ok(()); } for (i, stmt) in migrations.iter().enumerate().skip(version) { conn.execute_batch(stmt) .with_context(|| format!("{subsystem} migration v{} failed: {stmt}", i + 1))?; 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(()) } /// 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: &[&str] = &[ "ALTER TABLE things ADD COLUMN alpha TEXT", "ALTER TABLE things ADD COLUMN beta INTEGER NOT NULL DEFAULT 0", "ALTER TABLE things ADD COLUMN gamma TEXT", ]; #[test] fn fresh_install_runs_all_migrations() { let conn = open_tmp(); conn.execute_batch(TEST_SCHEMA).unwrap(); apply_versioned_migrations(&conn, "things", ("things", "gamma"), 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", ("things", "gamma"), 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", ("things", "gamma"), 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", ("things", "gamma"), TEST_MIGRATIONS).unwrap(); // Second call is a no-op. apply_versioned_migrations(&conn, "things", ("things", "gamma"), 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", ("a", "col1"), &["ALTER TABLE a ADD COLUMN col1 TEXT"], ) .unwrap(); apply_versioned_migrations( &conn, "store_b", ("b", "colx"), &["ALTER TABLE b ADD COLUMN colx INTEGER NOT NULL DEFAULT 0"], ) .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); } }