//! Schema migrations and pool construction. Migrations are an ordered list of //! (version, description, SQL) applied in one transaction each; pre-migration //! databases are detected and stamped as version 1. use super::{FromStr, Path, Result, SqliteConnectOptions, SqlitePool, SqlitePoolOptions}; use tracing::{info, instrument}; /// Each migration is a (version, description, SQL) tuple. Versions start at 1. /// The SQL may contain multiple statements separated by semicolons. const MIGRATIONS: &[(i64, &str, &str)] = &[ ( 1, "initial schema", r" CREATE TABLE IF NOT EXISTS health_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, status TEXT NOT NULL, checked_at TEXT NOT NULL, response_time_ms INTEGER NOT NULL, details_json TEXT, error TEXT ); CREATE TABLE IF NOT EXISTS test_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, started_at TEXT NOT NULL, finished_at TEXT, duration_secs INTEGER, exit_code INTEGER, passed INTEGER NOT NULL, summary_json TEXT NOT NULL, raw_output TEXT NOT NULL, filter TEXT ); CREATE TABLE IF NOT EXISTS peer_identities ( peer_name TEXT PRIMARY KEY, instance_id TEXT NOT NULL, first_seen TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS peer_heartbeats ( id INTEGER PRIMARY KEY AUTOINCREMENT, peer_name TEXT NOT NULL, status TEXT NOT NULL, latency_ms INTEGER NOT NULL, checked_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_health_checks_target_id ON health_checks(target, id DESC); CREATE INDEX IF NOT EXISTS idx_health_checks_target_checked ON health_checks(target, checked_at); CREATE INDEX IF NOT EXISTS idx_test_runs_target_id ON test_runs(target, id DESC); CREATE INDEX IF NOT EXISTS idx_peer_heartbeats_peer_id ON peer_heartbeats(peer_name, id DESC); ", ), ( 2, "add alerts table", r" CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, alert_type TEXT NOT NULL, from_status TEXT, to_status TEXT, sent_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_alerts_target_sent ON alerts(target, sent_at); ", ), ( 3, "add tls_checks table", r" CREATE TABLE IF NOT EXISTS tls_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, host TEXT NOT NULL, valid INTEGER NOT NULL, days_remaining INTEGER NOT NULL, not_before TEXT NOT NULL, not_after TEXT NOT NULL, subject TEXT NOT NULL, issuer TEXT NOT NULL, checked_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_tls_checks_target_id ON tls_checks(target, id DESC); ", ), ( 4, "add incidents table", r" CREATE TABLE IF NOT EXISTS incidents ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, started_at TEXT NOT NULL, ended_at TEXT, duration_secs INTEGER, from_status TEXT NOT NULL, to_status TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_incidents_target_id ON incidents(target, id DESC); ", ), ( 5, "add route_checks table", r" CREATE TABLE IF NOT EXISTS route_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, path TEXT NOT NULL, status_code INTEGER NOT NULL, ok INTEGER NOT NULL, response_time_ms INTEGER NOT NULL, checked_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_route_checks_target_path ON route_checks(target, path, id DESC); CREATE INDEX IF NOT EXISTS idx_route_checks_target ON route_checks(target, checked_at DESC); ", ), ( 6, "add dns_checks and whois_checks tables", r" CREATE TABLE IF NOT EXISTS dns_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, name TEXT NOT NULL, record_type TEXT NOT NULL, expected TEXT NOT NULL, actual TEXT NOT NULL, matches INTEGER NOT NULL, checked_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_dns_checks_target ON dns_checks(target, name, id DESC); CREATE TABLE IF NOT EXISTS whois_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, domain TEXT NOT NULL, registrar TEXT, expiry_date TEXT, days_remaining INTEGER, nameservers TEXT, checked_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_whois_checks_target ON whois_checks(target, id DESC); ", ), ( 7, "add test_details table", r" CREATE TABLE IF NOT EXISTS test_details ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id INTEGER NOT NULL REFERENCES test_runs(id) ON DELETE CASCADE, test_name TEXT NOT NULL, passed INTEGER NOT NULL, duration_ms INTEGER ); CREATE INDEX IF NOT EXISTS idx_test_details_run_id ON test_details(run_id); CREATE INDEX IF NOT EXISTS idx_test_details_name ON test_details(test_name, run_id DESC); ", ), ( 8, "add cors_checks table", r" CREATE TABLE IF NOT EXISTS cors_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, url TEXT NOT NULL, origin TEXT NOT NULL, method TEXT NOT NULL, passes INTEGER NOT NULL, checked_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_cors_target_id ON cors_checks(target, id DESC); ", ), ( 9, "add backup_checks table", r" CREATE TABLE IF NOT EXISTS backup_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, database_name TEXT NOT NULL, status TEXT NOT NULL, last_backup_at TEXT, size_bytes INTEGER, age_hours INTEGER, checked_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_backup_checks_target ON backup_checks(target, id DESC); ", ), ( 10, "add pending_alerts retry queue", r" CREATE TABLE IF NOT EXISTS pending_alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, alert_key TEXT NOT NULL, category TEXT NOT NULL, channel TEXT NOT NULL, -- 'wam' or 'email' subject TEXT NOT NULL, body TEXT NOT NULL, priority TEXT, -- WAM only source TEXT, -- WAM only source_ref TEXT, -- WAM only from_status TEXT, to_status TEXT, error TEXT, attempts INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, next_retry_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_pending_alerts_due ON pending_alerts(next_retry_at, id); ", ), ( 11, "add scan_pipeline_checks table", r" CREATE TABLE IF NOT EXISTS scan_pipeline_checks ( id INTEGER PRIMARY KEY AUTOINCREMENT, target TEXT NOT NULL, status TEXT NOT NULL, issues TEXT NOT NULL, -- JSON array of fired-threshold lines queue_pending INTEGER NOT NULL, queue_running INTEGER NOT NULL, queue_stuck INTEGER NOT NULL, held_total INTEGER NOT NULL, checked_at TEXT NOT NULL, error TEXT ); CREATE INDEX IF NOT EXISTS idx_scan_pipeline_checks_target ON scan_pipeline_checks(target, id DESC); ", ), ]; #[instrument(skip_all)] pub async fn connect(path: &Path) -> Result { let opts = SqliteConnectOptions::from_str(&format!("sqlite:{}", path.display()))? .create_if_missing(true) .foreign_keys(true) .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal); let pool = SqlitePoolOptions::new() .max_connections(5) .connect_with(opts) .await?; run_migrations(&pool).await?; Ok(pool) } #[instrument(skip_all)] pub async fn connect_in_memory() -> Result { let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.foreign_keys(true); let pool = SqlitePoolOptions::new() .max_connections(1) .connect_with(opts) .await?; run_migrations(&pool).await?; Ok(pool) } /// Run pending schema migrations. Detects pre-migration databases by checking /// for existing tables and stamps them as version 1 without re-running. #[instrument(skip_all)] pub async fn run_migrations(pool: &SqlitePool) -> Result<()> { // Ensure the schema_version table exists sqlx::query( "CREATE TABLE IF NOT EXISTS schema_version ( version INTEGER NOT NULL, description TEXT NOT NULL, applied_at TEXT NOT NULL )", ) .execute(pool) .await?; let current_version = get_schema_version(pool).await?; // Detect pre-migration databases: if schema_version is empty but tables exist, // this is an existing database that predates the migration system. if current_version == 0 && has_existing_tables(pool).await? { info!("detected pre-migration database, stamping as version 1"); stamp_version(pool, 1, "initial schema (pre-existing)").await?; // Run remaining migrations (2+) if any for &(version, description, sql) in MIGRATIONS { if version > 1 { run_one_migration(pool, version, description, sql).await?; } } return Ok(()); } // Run all migrations newer than current version for &(version, description, sql) in MIGRATIONS { if version > current_version { run_one_migration(pool, version, description, sql).await?; } } Ok(()) } /// Get the current schema version (0 if no migrations have been applied). #[instrument(skip_all)] pub async fn get_schema_version(pool: &SqlitePool) -> Result { let row = sqlx::query_as::<_, (i64,)>("SELECT COALESCE(MAX(version), 0) FROM schema_version") .fetch_one(pool) .await?; Ok(row.0) } /// Check whether the database has existing tables from before the migration system. async fn has_existing_tables(pool: &SqlitePool) -> Result { let row = sqlx::query_as::<_, (i64,)>( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'health_checks'", ) .fetch_one(pool) .await?; Ok(row.0 > 0) } /// Execute a single migration's SQL and record it in schema_version. /// Wrapped in an explicit transaction so partial failures roll back cleanly. async fn run_one_migration( pool: &SqlitePool, version: i64, description: &str, sql: &str, ) -> Result<()> { info!(version, description, "running migration"); let mut tx = pool.begin().await?; // Execute the whole migration as a batch via `raw_sql`, which hands the // string to SQLite's own parser to split on statement boundaries. A naive // `split(';')` breaks on the first `;` inside a trigger body or string // literal (fuzz-2026-07-06 migration split trap); today's migrations have // none, but this makes the first such migration safe rather than a startup // failure. sqlx::raw_sql(sqlx::AssertSqlSafe(sql)) .execute(&mut *tx) .await?; // Record the version inside the same transaction let now = chrono::Utc::now().to_rfc3339(); sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)") .bind(version) .bind(description) .bind(&now) .execute(&mut *tx) .await?; tx.commit().await?; Ok(()) } /// Record a version in the schema_version table. async fn stamp_version(pool: &SqlitePool, version: i64, description: &str) -> Result<()> { let now = chrono::Utc::now().to_rfc3339(); sqlx::query("INSERT INTO schema_version (version, description, applied_at) VALUES (?, ?, ?)") .bind(version) .bind(description) .bind(&now) .execute(pool) .await?; Ok(()) }