feat(db): versioned schema migrations with idempotent backfill

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::<bool, _>("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<Vec<(i64, String, String)>> {
+        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::<i64, _>("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);
+    }
+}