feat: adopt sqlx::migrate for PostgreSQL catalog and data migrations (ADR-0003)

.gitattributes

@@ -0,0 +1,3 @@
+# sqlx checksums migration file bytes. Pin LF so a contributor's line-ending
+# rewrite does not trip a false checksum mismatch (ADR-0003).
+*.sql text eol=lf

Cargo.toml

@@ -57,7 +57,7 @@ inventory = "0.3"
 moka = { version = "0.12", features = ["future"] }
 
 # Database
-sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "json", "time", "uuid", "bigdecimal"] }
+sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "json", "time", "uuid", "bigdecimal", "migrate"] }
 
 # Crypto & checksums
 crc32fast = "1"

crates/bin/src/cmd_init.rs

@@ -179,6 +179,8 @@ pub async fn run(args: InitArgs) -> anyhow::Result<u8> {
     if initialized {
         println!("--- Catalog already initialized. Use 'extenddb migrate' for pending migrations.");
     } else {
+        // run_catalog_migrations applies the schema and writes catalog_version
+        // (the version write moved here from the migration SQL; ADR-0003).
         bootstrapper
             .run_catalog_migrations()
             .await

crates/bin/src/cmd_migrate.rs

@@ -3,7 +3,8 @@
 
 //! `extenddb migrate` — apply catalog schema migrations (REQ-CAT-014).
 //!
-//! Reads current catalog version, runs pending migrations, and reports the result.
+//! Reads the current catalog version, runs both migrators (which validate the
+//! checksums of already-applied migrations), and reports the result.
 
 use clap::Args;
 
@@ -52,6 +53,22 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> {
             .await
             .map_err(|e| anyhow::anyhow!("{e:?}"))?;
 
+    // ADR-0003: a catalog created by the pre-sqlx runner cannot be upgraded in
+    // place. Refuse with the re-init directive instead of failing later on a
+    // non-idempotent DDL re-run.
+    if bootstrap
+        .catalog_predates_sqlx()
+        .await
+        .map_err(|e| anyhow::anyhow!("{e:?}"))?
+    {
+        anyhow::bail!(
+            "This catalog predates the sqlx migration system (ADR-0003). In-place \
+             upgrade is not supported. Run 'extenddb destroy' then 'extenddb init' \
+             to recreate both databases (this drops all data). See \
+             docs/manuals/07-upgrade-manual.md."
+        );
+    }
+
     // Show current catalog version.
     println!("--- Checking current catalog version...");
     let current = bootstrap
@@ -79,13 +96,11 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> {
         println!("  Pending: {}", data_pending.join(", "));
     }
 
-    if !catalog_pending && data_pending.is_empty() {
-        println!();
-        println!("Everything is up to date (catalog version {expected}). No migrations needed.");
-        return Ok(());
-    }
+    let has_pending = catalog_pending || !data_pending.is_empty();
 
-    if !args.yes {
+    // Applying migrations mutates the schema, so require explicit confirmation
+    // when there is pending work.
+    if has_pending && !args.yes {
         let mut what = Vec::new();
         if catalog_pending {
             what.push(format!("catalog {current_display} -> {expected}"));
@@ -99,16 +114,15 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> {
         );
     }
 
-    if catalog_pending {
-        bootstrap
-            .run_catalog_migrations()
-            .await
-            .map_err(|e| anyhow::anyhow!("{e:?}"))?;
-    }
-
-    // Always run data migrations: the data-database ledger is idempotent
-    // (already-applied migrations are skipped) and independent of the catalog
-    // version, so an upgrade that only touched the data schema is still applied.
+    // Run both migrators unconditionally, even when nothing is pending. sqlx
+    // validates the checksum of every already-applied migration on each run, so
+    // a migration file edited after it shipped is caught loudly here instead of
+    // drifting silently. Applying is idempotent; an up-to-date run applies
+    // nothing.
+    bootstrap
+        .run_catalog_migrations()
+        .await
+        .map_err(|e| anyhow::anyhow!("{e:?}"))?;
     bootstrap
         .run_data_migrations()
         .await
@@ -122,8 +136,12 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> {
     let new_display = new.as_deref().unwrap_or("none");
 
     println!();
-    println!("=== extenddb migrate complete ===");
-    println!("Catalog version: {current_display} -> {new_display}");
+    if has_pending {
+        println!("=== extenddb migrate complete ===");
+        println!("Catalog version: {current_display} -> {new_display}");
+    } else {
+        println!("Everything is up to date (catalog version {expected}). No migrations applied.");
+    }
 
     Ok(())
 }

crates/storage-postgres/data_migrations/001_data_schema.sql

@@ -5,8 +5,6 @@
 -- stream records and idempotency tokens can be written atomically with item
 -- data within a single PostgreSQL transaction (P54 Bug 1).
 
-BEGIN;
-
 -- Stream shards — fixed shards per table, assigned by partition key hash.
 -- No FK to catalog tables (cross-database FKs are not possible).
 -- Application-level integrity ensures table_id validity.
@@ -37,9 +35,6 @@ CREATE INDEX IF NOT EXISTS idx_stream_records_created
     ON stream_records (created_at);
 
 -- Monotonic sequence for stream record ordering (CB-21).
--- Note: the idempotency check in run_data_migrations uses stream_shards
--- existence to decide whether to skip this entire migration. If stream_shards
--- is created manually without the sequence, stream_seq will not exist.
 CREATE SEQUENCE IF NOT EXISTS stream_seq START 1;
 SELECT setval('stream_seq', GREATEST(
     (EXTRACT(EPOCH FROM now()) * 1000000)::BIGINT,
@@ -55,5 +50,3 @@ CREATE TABLE IF NOT EXISTS idempotency_tokens (
 
 CREATE INDEX IF NOT EXISTS idx_idempotency_tokens_created
     ON idempotency_tokens (created_at);
-
-COMMIT;

crates/storage-postgres/data_migrations/003_idempotency_account_scope.sql

@@ -15,8 +15,6 @@
 -- acceptable for a short-lived cache. Apply data migrations during the
 -- stop / migrate / restart upgrade sequence to bound that window.
 
-BEGIN;
-
 DROP TABLE IF EXISTS idempotency_tokens;
 
 CREATE TABLE idempotency_tokens (
@@ -29,5 +27,3 @@ CREATE TABLE idempotency_tokens (
 
 CREATE INDEX idx_idempotency_tokens_created
     ON idempotency_tokens (created_at);
-
-COMMIT;

crates/storage-postgres/migrations/001_schema.sql

@@ -1,9 +1,9 @@
 -- Copyright 2026 ExtendDB contributors
 -- SPDX-License-Identifier: Apache-2.0
--- Consolidated catalog schema for extenddb (catalog version 0.0.2).
--- This is the complete schema applied on fresh installs.
-
-BEGIN;
+-- Consolidated catalog schema for extenddb.
+-- This is the complete schema applied on fresh installs. The catalog version
+-- is written by the migration runner after this file applies (sqlx has no
+-- knowledge of our semver), so it is not seeded here.
 
 -- Accounts — multi-account support (REQ-AUTH-005).
 CREATE TABLE IF NOT EXISTS accounts (
@@ -69,12 +69,6 @@ CREATE TABLE IF NOT EXISTS tags (
     PRIMARY KEY (resource_arn, tag_key)
 );
 
--- Migration tracking.
-CREATE TABLE IF NOT EXISTS schema_history (
-    filename TEXT PRIMARY KEY,
-    applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
-);
-
 -- Settings (catalog version, data database connection, runtime config).
 CREATE TABLE IF NOT EXISTS settings (
     key TEXT PRIMARY KEY,
@@ -310,12 +304,9 @@ SELECT setval('stream_seq', GREATEST(
     1
 ));
 
--- Seed settings.
-INSERT INTO settings (key, value) VALUES ('catalog_version', '0.0.2')
-ON CONFLICT (key) DO NOTHING;
+-- Seed settings. The catalog version is written by the migration runner after
+-- this file applies, not seeded here.
 INSERT INTO settings (key, value) VALUES ('control_plane_delay_seconds', '0.25')
 ON CONFLICT (key) DO NOTHING;
 INSERT INTO settings (key, value) VALUES ('gsi_propagation_delay_ms', '10')
 ON CONFLICT (key) DO NOTHING;
-
-COMMIT;

crates/storage-postgres/src/bootstrapper.rs

@@ -417,6 +417,23 @@ impl Bootstrapper for PostgresBootstrapper {
         Ok(row.map(|(v,)| v))
     }
 
+    async fn catalog_predates_sqlx(&self) -> OpResult<bool> {
+        let pool = self.app_pool(&self.config.catalog_db).await?;
+        // Uninitialized catalog: nothing to guard (init creates the sqlx ledger).
+        if !migrations::table_exists(&pool, "settings").await? {
+            return Ok(false);
+        }
+        // The old runner tracked migrations in `schema_history`; sqlx uses
+        // `_sqlx_migrations`. An initialized catalog carrying the legacy table,
+        // or lacking the sqlx ledger, predates the sqlx migrator. Checking the
+        // catalog is sufficient: init creates both databases together (or
+        // aborts), so a pre-sqlx data database never appears without a pre-sqlx
+        // catalog, which this refuses first.
+        let has_legacy = migrations::table_exists(&pool, "schema_history").await?;
+        let has_sqlx = migrations::table_exists(&pool, "_sqlx_migrations").await?;
+        Ok(has_legacy || !has_sqlx)
+    }
+
     fn expected_catalog_version(&self) -> String {
         CATALOG_VERSION.to_string()
     }

crates/storage-postgres/src/lib.rs

@@ -114,7 +114,7 @@ use sqlx::postgres::PgPoolOptions;
 ///
 /// The tuple is the single source of truth. Use `CATALOG_VERSION.to_string()`
 /// wherever a string representation is needed.
-pub const CATALOG_VERSION: CatalogVersion = CatalogVersion::new(0, 0, 2);
+pub const CATALOG_VERSION: CatalogVersion = CatalogVersion::new(0, 1, 0);
 
 /// Minimum number of connections allowed per pool.
 ///