diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 6690025e..63dcfb21 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -1,7 +1,8 @@ -use anyhow::Result; +use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use sqlx::{PgPool, Row}; +use tracing::info; use uuid::Uuid; // ── Public data types ───────────────────────────────────────────────────────── @@ -197,8 +198,174 @@ impl Db { Ok(db) } + /// Run all pending versioned migrations in order, inside a single + /// transaction per migration. Idempotent — migrations whose version is + /// already recorded in `schema_migrations` are skipped. + /// + /// Concurrency: the whole routine is guarded by a Postgres advisory lock so + /// two node instances pointed at the same database (e.g. during a + /// blue/green or rolling deploy) cannot race to apply the same migration + /// and trip the `schema_migrations` primary key. + /// + /// Legacy installs: v1 bundles the entire pre-versioning schema, and every + /// statement in it is idempotent (`CREATE TABLE IF NOT EXISTS`, + /// `CREATE INDEX IF NOT EXISTS`, `ADD COLUMN IF NOT EXISTS`). So an existing + /// node that predates this system just runs v1 once: existing objects are + /// no-ops, and any objects it was missing are created. We deliberately do + /// *not* short-circuit on the presence of a single canonical table — a node + /// that was behind on schema would then be marked complete while still + /// missing newer objects. async fn migrate(&self) -> Result<()> { - let stmts = [ + // Bootstrap: ensure the `schema_migrations` table itself exists. + sqlx::query( + r#"CREATE TABLE IF NOT EXISTS schema_migrations ( + version BIGINT NOT NULL PRIMARY KEY, + name TEXT NOT NULL, + applied_at TEXT NOT NULL + )"#, + ) + .execute(&self.pool) + .await + .context("creating schema_migrations table")?; + + // Serialize migrations across processes: hold a session-level advisory + // lock on a dedicated connection for the whole run. Another instance + // starting up blocks here until we finish. The lock is released when we + // explicitly unlock below, or automatically if the connection is + // dropped (e.g. on panic), so a crash can't wedge future restarts. + let mut lock_conn = self + .pool + .acquire() + .await + .context("acquiring connection for migration advisory lock")?; + sqlx::query("SELECT pg_advisory_lock($1)") + .bind(MIGRATION_ADVISORY_LOCK) + .execute(&mut *lock_conn) + .await + .context("acquiring migration advisory lock")?; + + let result = self.run_pending_migrations().await; + + let _ = sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(MIGRATION_ADVISORY_LOCK) + .execute(&mut *lock_conn) + .await; + + result + } + + /// Apply every migration whose version isn't yet recorded, in order. + /// Must be called while holding the migration advisory lock. + async fn run_pending_migrations(&self) -> Result<()> { + for m in MIGRATIONS { + let already: bool = sqlx::query( + "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1) AS applied", + ) + .bind(m.version) + .fetch_one(&self.pool) + .await? + .get::("applied"); + + if already { + continue; + } + + let started = std::time::Instant::now(); + info!( + version = m.version, + name = m.name, + statements = m.stmts.len(), + "applying migration" + ); + + // Run the migration body in a single transaction so a failure + // mid-way leaves the database in its prior state rather than + // partially mutated. + let mut tx = self.pool.begin().await?; + for stmt in m.stmts { + sqlx::query(stmt).execute(&mut *tx).await.with_context(|| { + format!( + "migration v{} ({}) failed on statement: {}", + m.version, m.name, stmt + ) + })?; + } + sqlx::query( + "INSERT INTO schema_migrations (version, name, applied_at) + VALUES ($1, $2, $3)", + ) + .bind(m.version) + .bind(m.name) + .bind(Utc::now().to_rfc3339()) + .execute(&mut *tx) + .await + .context("recording migration as applied")?; + tx.commit() + .await + .with_context(|| format!("committing migration v{}", m.version))?; + + info!( + version = m.version, + name = m.name, + elapsed_ms = started.elapsed().as_millis() as u64, + "migration applied" + ); + } + + Ok(()) + } + + /// Returns `(version, name, applied_at)` for every applied migration, + /// oldest first. Useful for ops/observability — surface via `gl status` + /// or `/api/v1/stats` in a follow-up. + #[allow(dead_code)] + pub async fn migration_status(&self) -> Result> { + let rows = sqlx::query( + "SELECT version, name, applied_at FROM schema_migrations ORDER BY version ASC", + ) + .fetch_all(&self.pool) + .await?; + Ok(rows + .into_iter() + .map(|r| { + ( + r.get::("version"), + r.get("name"), + r.get("applied_at"), + ) + }) + .collect()) + } +} + +// ── Migration catalogue ────────────────────────────────────────────────────── +// +// All schema statements are bundled into a single v1 migration so we can ship +// versioned migrations on a live network without breaking the existing +// install base. Future schema changes MUST be added as v2, v3, … — never +// appended to v1. Operators can read `schema_migrations` to confirm a node +// is at the expected version. +// +// Each migration runs in a single transaction, so statements that Postgres +// forbids inside a transaction (notably `CREATE INDEX CONCURRENTLY`) cannot be +// used here. Build such indexes the ordinary, transaction-safe way, or stage +// them as a dedicated out-of-band operational step. + +// Arbitrary but stable key for the migration advisory lock ("gitlawb_" bytes). +const MIGRATION_ADVISORY_LOCK: i64 = 0x6769_746C_6177_625F; + +const MIGRATION_V1_NAME: &str = "initial_schema"; + +struct Migration { + version: i64, + name: &'static str, + stmts: &'static [&'static str], +} + +const MIGRATIONS: &[Migration] = &[Migration { + version: 1, + name: MIGRATION_V1_NAME, + stmts: &[ r#"CREATE TABLE IF NOT EXISTS repos ( id TEXT NOT NULL PRIMARY KEY, name TEXT NOT NULL, @@ -453,14 +620,8 @@ impl Db { "CREATE INDEX IF NOT EXISTS idx_bounties_status ON bounties(status)", "CREATE INDEX IF NOT EXISTS idx_bounties_repo ON bounties(repo_owner, repo_name)", "CREATE INDEX IF NOT EXISTS idx_bounties_claimant ON bounties(claimant_did)", - ]; - - for stmt in &stmts { - sqlx::query(stmt).execute(&self.pool).await?; - } - Ok(()) - } -} + ], +}]; // ── Repos ───────────────────────────────────────────────────────────────────── @@ -2189,3 +2350,93 @@ impl Db { } } } + +// ── Tests ───────────────────────────────────────────────────────────────────── +// +// These tests don't require a live Postgres connection. They validate the +// static migration catalogue is well-formed so a future maintainer can't +// ship a regression like duplicate versions, negative versions, or empty +// migration bodies. The actual SQL execution is exercised by integration +// tests / first-run on a real node. + +#[cfg(test)] +mod migration_tests { + use super::{MIGRATIONS, MIGRATION_V1_NAME}; + + #[test] + fn migrations_are_non_empty() { + assert!( + !MIGRATIONS.is_empty(), + "MIGRATIONS must contain at least the initial v1 schema" + ); + } + + #[test] + fn migration_versions_are_strictly_increasing() { + let mut last = i64::MIN; + for m in MIGRATIONS { + assert!( + m.version > last, + "migration versions must be strictly increasing; \ + found {} after {}", + m.version, + last + ); + last = m.version; + } + } + + #[test] + fn migration_versions_start_at_one() { + // A version of 0 (or negative) would be a footgun: any future + // `WHERE version > current_max` style query would skip it. + assert_eq!( + MIGRATIONS.first().map(|m| m.version), + Some(1), + "the first migration must have version 1" + ); + } + + #[test] + fn migration_names_are_non_empty_and_distinct() { + let mut seen = std::collections::HashSet::new(); + for m in MIGRATIONS { + assert!( + !m.name.is_empty(), + "migration v{} has empty name", + m.version + ); + assert!( + !m.name.contains(char::is_whitespace), + "migration v{} name {:?} contains whitespace", + m.version, + m.name + ); + assert!( + seen.insert(m.name), + "duplicate migration name: {:?}", + m.name + ); + } + } + + #[test] + fn migration_bodies_are_non_empty() { + for m in MIGRATIONS { + assert!( + !m.stmts.is_empty(), + "migration v{} ({}) has no SQL statements", + m.version, + m.name + ); + } + } + + #[test] + fn v1_name_is_the_initial_schema() { + // This is what the legacy-install backfill writes to + // `schema_migrations` when an existing node upgrades. If you rename + // it, you must also update the backfill. + assert_eq!(MIGRATIONS[0].name, MIGRATION_V1_NAME); + } +}