reliar-store-postgres 0.9.0

//! §43.A.24 — `migrate()` creates schema `reliar`, is idempotent, serializes concurrent
//! callers, records versions in Reliar's own `_migrations` bookkeeping table (never
//! `_sqlx_migrations`), and is never invoked implicitly; an unmigrated pool has no `outbox`.

use crate::common;

use reliar_outbox::{AcquireRequest, OutboxStore, WorkerId};
use reliar_store_postgres::{MigrateOptions, PostgresOutboxStore, migrate};
use sqlx::migrate::Migrator;
use sqlx::postgres::PgConnection;
use sqlx::{Connection, Executor};

/// The crate's own migration set, embedded a second time from test code (test code cannot see
/// `src/migrate.rs`'s private `MIGRATOR`). `CARGO_MANIFEST_DIR` is the same for this package's
/// `lib` and `test` targets, so `"./migrations"` resolves identically to the production copy.
/// Used only to apply a **prefix** of the set (`migrations[..1]`), simulating a database a
/// pre-0.5.0 `migrate()` created before `0002`/`0003` existed (P-N23).
static ALL_MIGRATIONS: Migrator = sqlx::migrate!("./migrations");

/// Applies only `0001_outbox.sql` against a fresh database, on a dedicated connection with
/// `search_path` set the way [`migrate`] itself sets it — `0001`'s SQL is unqualified.
async fn apply_only_migration_0001(pool: &sqlx::PgPool) {
    let mut conn = PgConnection::connect_with(&pool.connect_options())
        .await
        .expect("dedicated connection for the partial migration");

    conn.execute("SET search_path = reliar, public")
        .await
        .expect("set search_path for the unqualified 0001 migration");

    let mut only_0001 = Migrator {
        migrations: std::borrow::Cow::Owned(ALL_MIGRATIONS.migrations[..1].to_vec()),
        ignore_missing: ALL_MIGRATIONS.ignore_missing,
        locking: ALL_MIGRATIONS.locking,
        no_tx: ALL_MIGRATIONS.no_tx,
        table_name: ALL_MIGRATIONS.table_name.clone(),
        create_schemas: ALL_MIGRATIONS.create_schemas.clone(),
    };
    only_0001.create_schema("reliar".to_owned());
    only_0001.dangerous_set_table_name("reliar._migrations");
    only_0001.set_locking(false);
    only_0001
        .run(&mut conn)
        .await
        .expect("apply migration 0001 alone");

    conn.close().await.expect("close the dedicated connection");
}

/// P-N23 — every other `migrate()` test in this file migrates an
/// **empty** `outbox`, which is not the upgrade path a real 0.3.0-vintage deployment takes:
/// `CREATE INDEX CONCURRENTLY` (`0002`) building over live rows, and `0003`'s `ACCESS EXCLUSIVE`
/// `DROP INDEX` under `lock_timeout` running against a populated table, are both untested without
/// this. Seeds 1000 rows (900 pending, 50 leased, 50 dead) onto a database migrated only through
/// `0001`, then calls the crate's real `migrate()` to bring it up to `0003`, and asserts both the
/// schema shape *and* that a claim still succeeds afterwards.
async fn migrate_upgrades_a_populated_0001_only_database_and_claim_still_works() {
    let pool = common::fresh_unmigrated_db().await;

    apply_only_migration_0001(&pool).await;

    sqlx::query(
        "INSERT INTO reliar.outbox (id, message_type, message_version, conversation_id, content_type, \
                              payload, available_at, created_at) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now(), now() \
           FROM generate_series(1, 900)",
    )
    .execute(&pool)
    .await
    .expect("seed 900 pending rows");
    sqlx::query(
        "INSERT INTO reliar.outbox (id, message_type, message_version, conversation_id, content_type, \
                              payload, available_at, created_at, locked_by, locked_until) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now() + interval '30 seconds', now(), 'legacy-worker', now() + interval '30 seconds' \
           FROM generate_series(1, 50)",
    )
    .execute(&pool)
    .await
    .expect("seed 50 leased rows");
    sqlx::query(
        "INSERT INTO reliar.outbox (id, message_type, message_version, conversation_id, content_type, \
                              payload, available_at, created_at, dead_at, dead_reason) \
         SELECT uuidv7(), 'orders.created', 1, uuidv7(), 'application/json', '{}'::bytea, \
                now() - interval '1 day', now() - interval '1 day', now() - interval '12 hours', \
                'permanent_error' \
           FROM generate_series(1, 50)",
    )
    .execute(&pool)
    .await
    .expect("seed 50 dead rows");

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("migrate() upgrades a populated 0001-only database to 0002/0003");

    let applied_count: i64 = sqlx::query_scalar("SELECT count(*) FROM reliar._migrations")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(
        usize::try_from(applied_count).expect("applied_count is never negative"),
        ALL_MIGRATIONS.migrations.len(),
        "every migration after 0001 must now be recorded too"
    );

    let claimable_valid: Option<bool> = sqlx::query_scalar(
        "SELECT indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_outbox_claimable'",
    )
    .fetch_optional(&pool)
    .await
    .unwrap();
    assert_eq!(
        claimable_valid,
        Some(true),
        "ix_outbox_claimable must exist and be valid after upgrading a populated database"
    );

    let pending_exists: bool =
        sqlx::query_scalar("SELECT to_regclass('reliar.ix_outbox_pending') IS NOT NULL")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(
        !pending_exists,
        "ix_outbox_pending must be dropped once ix_outbox_claimable is confirmed valid"
    );

    // The design's other P-N23 requirement: a claim must still succeed against the upgraded
    // schema, not just the schema shape looking right. `pool` itself has no `search_path` set
    // (`fresh_unmigrated_db` deliberately leaves it unconfigured), so reconnect to the same
    // database the way a host normally would — `reliar` first on `search_path` — for the store's
    // own unqualified SQL and its startup verification to resolve `outbox` correctly.
    let store_pool = sqlx::PgPool::connect_with(
        (*pool.connect_options())
            .clone()
            .options([("search_path", "reliar,public")]),
    )
    .await
    .expect("reconnect with search_path set");
    let store = PostgresOutboxStore::new(store_pool);
    let batch = store
        .acquire(AcquireRequest::new(WorkerId::generate()).batch_size(50))
        .await
        .expect("claim still works after the upgrade");
    assert_eq!(
        batch.records.len(),
        50,
        "acquire must return exactly batch_size claimable rows after the upgrade \
         (weaker than !is_empty(): this also catches an index the claim can no longer use)"
    );
}

/// ADR 0040 §2 — `0003`'s `DO $$ … RAISE EXCEPTION` guard refuses to drop
/// `ix_outbox_pending` when `ix_outbox_claimable` is missing or invalid, and the documented
/// recovery (rebuild the index, re-run `migrate()`) actually works. Simulated by migrating fully,
/// then dropping the new index and rolling `_migrations` back one version — exactly what a
/// connection loss mid-`0002` on a real deployment would leave behind.
async fn migrate_refuses_0003_when_claimable_index_is_missing_and_recovers() {
    let pool = common::fresh_db().await; // already migrated through 0003 by the shared template

    sqlx::query("DROP INDEX CONCURRENTLY ix_outbox_claimable")
        .execute(&pool)
        .await
        .expect("drop ix_outbox_claimable to simulate an interrupted 0002");
    sqlx::query("DELETE FROM reliar._migrations WHERE version = 3")
        .execute(&pool)
        .await
        .expect("roll back 0003's bookkeeping row");

    let err = migrate(&pool, MigrateOptions::default())
        .await
        .expect_err("0003 must refuse to run without a valid ix_outbox_claimable");
    assert!(
        err.to_string().contains("ix_outbox_claimable"),
        "the error must name the missing index, got: {err}"
    );

    // Recovery: rebuild the index in its current, `id`-keyed shape (ADR 0049 Amendment A —
    // `fresh_db()` is already migrated all the way to `0016`, so `sequence` and `locked_until` no
    // longer exist), then re-run migrate() from the start.
    sqlx::query(
        "CREATE INDEX CONCURRENTLY ix_outbox_claimable ON outbox (available_at, id) \
           INCLUDE (expires_at) \
          WHERE published_at IS NULL AND dead_at IS NULL",
    )
    .execute(&pool)
    .await
    .expect("rebuild ix_outbox_claimable");

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("migrate() recovers once ix_outbox_claimable is valid again");

    let pending_exists: bool =
        sqlx::query_scalar("SELECT to_regclass('reliar.ix_outbox_pending') IS NOT NULL")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(!pending_exists, "recovery must still finish 0003");
}

/// P-N16 (ADR 0040 §2) — the sibling above proves 0003's guard against a **missing**
/// `ix_outbox_claimable`; this proves the other branch the same `DO $$` checks: an index that is
/// still *present* but `indisvalid = false`, exactly what a `CREATE INDEX CONCURRENTLY` that fails
/// partway through leaves behind (connection dropped mid-build, `0002` never rolls back). The
/// documented recovery (`DROP INDEX CONCURRENTLY`, rebuild, re-run `migrate()`) must also work from
/// this branch, not only from the missing-index one.
async fn migrate_refuses_0003_when_claimable_index_is_invalid_and_recovers() {
    let pool = common::fresh_db().await; // already migrated through 0003 by the shared template

    sqlx::query(
        "UPDATE pg_index SET indisvalid = false \
          WHERE indexrelid = 'ix_outbox_claimable'::regclass",
    )
    .execute(&pool)
    .await
    .expect(
        "mark ix_outbox_claimable invalid, simulating a CONCURRENTLY build that failed partway",
    );
    sqlx::query("DELETE FROM reliar._migrations WHERE version = 3")
        .execute(&pool)
        .await
        .expect("roll back 0003's bookkeeping row");

    let err = migrate(&pool, MigrateOptions::default())
        .await
        .expect_err("0003 must refuse to run over an invalid ix_outbox_claimable");
    assert!(
        err.to_string().contains("ix_outbox_claimable"),
        "the error must name the invalid index, got: {err}"
    );

    // Recovery: drop the invalid index, rebuild it in its current, `id`-keyed shape (ADR 0049
    // Amendment A — `sequence` and `locked_until` no longer exist on a database migrated all the
    // way to `0016`), then re-run migrate().
    sqlx::query("DROP INDEX CONCURRENTLY ix_outbox_claimable")
        .execute(&pool)
        .await
        .expect("drop the invalid index");
    sqlx::query(
        "CREATE INDEX CONCURRENTLY ix_outbox_claimable ON outbox (available_at, id) \
           INCLUDE (expires_at) \
          WHERE published_at IS NULL AND dead_at IS NULL",
    )
    .execute(&pool)
    .await
    .expect("rebuild ix_outbox_claimable");

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("migrate() recovers once ix_outbox_claimable is valid again");

    let pending_exists: bool =
        sqlx::query_scalar("SELECT to_regclass('reliar.ix_outbox_pending') IS NOT NULL")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(!pending_exists, "recovery must still finish 0003");
}

async fn sequence_column_exists(pool: &sqlx::PgPool) -> bool {
    sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM information_schema.columns \
          WHERE table_schema = 'reliar' AND table_name = 'outbox' AND column_name = 'sequence')",
    )
    .fetch_one(pool)
    .await
    .unwrap()
}

/// ADR 0049 Amendment A — `0014`'s `DO $$ … RAISE EXCEPTION` guard refuses to rename
/// `ix_outbox_claimable_id`/`ix_outbox_ordering_key_id` into place and drop the `sequence`
/// column when either transient index is **missing**, and the documented recovery (rebuild it
/// concurrently, re-run `migrate()`) actually works — the same shape as the `0003` pair above,
/// adapted for the extra column-drop `0014` performs. Built on `common::apply_migration_prefix`
/// (stopping at `0013`) rather than `fresh_db()` plus a bookkeeping rollback, the way `0003`'s
/// pair is: `0014`, unlike `0003`, also drops a column, and reversing a *fully* migrated
/// database cannot bring `sequence` back.
async fn migrate_refuses_0014_when_a_transient_index_is_missing_and_recovers() {
    let pool = common::fresh_unmigrated_db().await;

    common::apply_migration_prefix(&pool, 13).await;

    sqlx::query("DROP INDEX CONCURRENTLY reliar.ix_outbox_claimable_id")
        .execute(&pool)
        .await
        .expect("drop ix_outbox_claimable_id to simulate it never having been (re)built");

    let err = migrate(&pool, MigrateOptions::default())
        .await
        .expect_err("0014 must refuse to run without a valid ix_outbox_claimable_id");
    assert!(
        err.to_string().contains("ix_outbox_claimable_id")
            && err.to_string().contains("ix_outbox_ordering_key_id"),
        "the error must name both transient indexes, got: {err}"
    );
    assert!(
        sequence_column_exists(&pool).await,
        "0014 must refuse before dropping the sequence column, not after"
    );

    // Recovery: rebuild the missing index in its documented shape, then re-run migrate().
    sqlx::query(
        "CREATE INDEX CONCURRENTLY ix_outbox_claimable_id ON reliar.outbox (available_at, id) \
           INCLUDE (locked_until, expires_at) \
          WHERE published_at IS NULL AND dead_at IS NULL",
    )
    .execute(&pool)
    .await
    .expect("rebuild ix_outbox_claimable_id");

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("migrate() recovers once both transient indexes are valid again");

    assert!(
        !sequence_column_exists(&pool).await,
        "recovery must still finish 0014 and drop the sequence column"
    );

    let claimable_valid: Option<bool> = sqlx::query_scalar(
        "SELECT indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_outbox_claimable'",
    )
    .fetch_optional(&pool)
    .await
    .unwrap();
    assert_eq!(
        claimable_valid,
        Some(true),
        "ix_outbox_claimable must exist under its permanent name and be valid"
    );

    let ordering_key_valid: Option<bool> = sqlx::query_scalar(
        "SELECT indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_outbox_ordering_key'",
    )
    .fetch_optional(&pool)
    .await
    .unwrap();
    assert_eq!(
        ordering_key_valid,
        Some(true),
        "ix_outbox_ordering_key must exist under its permanent name and be valid"
    );
}

/// The sibling of the test above: `0014`'s guard against an index that is still *present* but
/// `indisvalid = false`, exactly what a `CREATE INDEX CONCURRENTLY` that fails partway leaves
/// behind. Targets `ix_outbox_ordering_key_id` rather than `ix_outbox_claimable_id` so the two
/// tests together demonstrate the guard actually checks *both* names, not just one.
async fn migrate_refuses_0014_when_a_transient_index_is_invalid_and_recovers() {
    let pool = common::fresh_unmigrated_db().await;

    common::apply_migration_prefix(&pool, 13).await;

    sqlx::query(
        "UPDATE pg_index SET indisvalid = false \
          WHERE indexrelid = 'reliar.ix_outbox_ordering_key_id'::regclass",
    )
    .execute(&pool)
    .await
    .expect(
        "mark ix_outbox_ordering_key_id invalid, simulating a CONCURRENTLY build that failed \
         partway",
    );

    let err = migrate(&pool, MigrateOptions::default())
        .await
        .expect_err("0014 must refuse to run over an invalid ix_outbox_ordering_key_id");
    assert!(
        err.to_string().contains("ix_outbox_claimable_id")
            && err.to_string().contains("ix_outbox_ordering_key_id"),
        "the error must name both transient indexes, got: {err}"
    );
    assert!(
        sequence_column_exists(&pool).await,
        "0014 must refuse before dropping the sequence column, not after"
    );

    // Recovery: drop the invalid index, rebuild it in its documented shape, then re-run
    // migrate().
    sqlx::query("DROP INDEX CONCURRENTLY reliar.ix_outbox_ordering_key_id")
        .execute(&pool)
        .await
        .expect("drop the invalid index");
    sqlx::query(
        "CREATE INDEX CONCURRENTLY ix_outbox_ordering_key_id ON reliar.outbox (ordering_key, id) \
          WHERE ordering_key IS NOT NULL AND published_at IS NULL AND dead_at IS NULL",
    )
    .execute(&pool)
    .await
    .expect("rebuild ix_outbox_ordering_key_id");

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("migrate() recovers once both transient indexes are valid again");

    assert!(
        !sequence_column_exists(&pool).await,
        "recovery must still finish 0014 and drop the sequence column"
    );

    let ordering_key_valid: Option<bool> = sqlx::query_scalar(
        "SELECT indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_outbox_ordering_key'",
    )
    .fetch_optional(&pool)
    .await
    .unwrap();
    assert_eq!(
        ordering_key_valid,
        Some(true),
        "ix_outbox_ordering_key must exist under its permanent name and be valid"
    );
}

/// T5 (ADR 0050 §6) — `0016`'s `DO $$ … RAISE EXCEPTION` guard refuses to drop `locked_until`
/// when `0015`'s concurrent build, `ix_outbox_claimable_v2`, is **missing** — the same shape as
/// the `0003`/`0014` pairs above, adapted for `0016`'s own transient name. Built on
/// `common::apply_migration_prefix` (stopping at `0015`) so the column still exists to be
/// dropped once recovery lets `0016` proceed.
async fn migrate_refuses_0016_when_the_transient_index_is_missing_and_recovers() {
    let pool = common::fresh_unmigrated_db().await;

    common::apply_migration_prefix(&pool, 15).await;

    sqlx::query("DROP INDEX CONCURRENTLY reliar.ix_outbox_claimable_v2")
        .execute(&pool)
        .await
        .expect("drop ix_outbox_claimable_v2 to simulate it never having been (re)built");

    let err = migrate(&pool, MigrateOptions::default())
        .await
        .expect_err("0016 must refuse to run without a valid ix_outbox_claimable_v2");
    assert!(
        err.to_string().contains("ix_outbox_claimable_v2"),
        "the error must name the missing transient index, got: {err}"
    );
    assert!(
        locked_until_column_exists(&pool).await,
        "0016 must refuse before dropping the locked_until column, not after"
    );

    // Recovery: rebuild the missing index in its documented shape, then re-run migrate().
    sqlx::query(
        "CREATE INDEX CONCURRENTLY ix_outbox_claimable_v2 ON reliar.outbox (available_at, id) \
           INCLUDE (expires_at) \
          WHERE published_at IS NULL AND dead_at IS NULL",
    )
    .execute(&pool)
    .await
    .expect("rebuild ix_outbox_claimable_v2");

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("migrate() recovers once ix_outbox_claimable_v2 is valid again");

    assert!(
        !locked_until_column_exists(&pool).await,
        "recovery must still finish 0016 and drop the locked_until column"
    );

    let claimable_valid: Option<bool> = sqlx::query_scalar(
        "SELECT indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_outbox_claimable'",
    )
    .fetch_optional(&pool)
    .await
    .unwrap();
    assert_eq!(
        claimable_valid,
        Some(true),
        "ix_outbox_claimable must exist under its permanent name and be valid"
    );
}

async fn locked_until_column_exists(pool: &sqlx::PgPool) -> bool {
    sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM information_schema.columns \
          WHERE table_schema = 'reliar' AND table_name = 'outbox' \
            AND column_name = 'locked_until')",
    )
    .fetch_one(pool)
    .await
    .unwrap()
}

/// T6 (ADR 0050 §6, §8) — after `0016` drops `locked_until`, `ix_outbox_claimable` still exists
/// under its permanent name, is valid, keeps `INCLUDE (expires_at)` only, and its predicate is
/// unchanged. `fresh_db()` is already migrated all the way to `0016`.
async fn ix_outbox_claimable_survives_dropping_locked_until() {
    let pool = common::fresh_db().await;

    let (claimable_valid, claimable_def): (Option<bool>, Option<String>) = {
        let valid: Option<bool> = sqlx::query_scalar(
            "SELECT indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \
              WHERE c.relname = 'ix_outbox_claimable'",
        )
        .fetch_optional(&pool)
        .await
        .unwrap();
        let def: Option<String> =
            sqlx::query_scalar("SELECT pg_get_indexdef('ix_outbox_claimable'::regclass)")
                .fetch_optional(&pool)
                .await
                .unwrap();

        (valid, def)
    };

    assert_eq!(
        claimable_valid,
        Some(true),
        "ix_outbox_claimable must exist under its permanent name and be valid"
    );
    let claimable_def = claimable_def.expect("ix_outbox_claimable must exist");
    assert!(
        claimable_def.contains("INCLUDE (expires_at)"),
        "expected INCLUDE (expires_at) only, got: {claimable_def}"
    );
    assert!(
        !claimable_def.contains("locked_until"),
        "ix_outbox_claimable must no longer INCLUDE locked_until, got: {claimable_def}"
    );
    assert!(
        claimable_def.contains("published_at IS NULL") && claimable_def.contains("dead_at IS NULL"),
        "the predicate must be unchanged, got: {claimable_def}"
    );
}

/// P-N24c (ADR 0040 §2) — 0003's guard must be scoped to the schema it is migrating, not to any
/// schema that happens to have an index of the same unqualified name. Both tenants fully migrate
/// first (so `tenant_b`'s own `ix_outbox_pending` is already gone, same as `tenant_a`'s); then
/// `tenant_b`'s `ix_outbox_claimable` is invalidated and its 0003 bookkeeping row rolled back to
/// force a re-run. Without the `pg_namespace`/`current_schema()` join, `tenant_b`'s guard would
/// pass by matching `tenant_a`'s unrelated valid index of the same name instead of refusing — the
/// `expect_err` below is what an unqualified guard fails.
async fn migrate_0003_guard_is_schema_scoped_and_ignores_another_schemas_valid_index() {
    let pool = common::fresh_unmigrated_db().await;

    migrate(&pool, MigrateOptions::default().schema("tenant_a"))
        .await
        .expect("tenant_a migrates cleanly");
    migrate(&pool, MigrateOptions::default().schema("tenant_b"))
        .await
        .expect("tenant_b migrates cleanly");

    sqlx::query(
        "UPDATE pg_index SET indisvalid = false \
          WHERE indexrelid = 'tenant_b.ix_outbox_claimable'::regclass",
    )
    .execute(&pool)
    .await
    .expect("invalidate tenant_b's claimable index only");
    sqlx::query("DELETE FROM tenant_b._migrations WHERE version = 3")
        .execute(&pool)
        .await
        .expect("roll back tenant_b's 0003 bookkeeping row");

    let err = migrate(&pool, MigrateOptions::default().schema("tenant_b"))
        .await
        .expect_err("tenant_b's 0003 must refuse despite tenant_a's valid index of the same name");
    assert!(
        err.to_string().contains("ix_outbox_claimable"),
        "the error must name the missing/invalid index, got: {err}"
    );

    let tenant_a_valid: Option<bool> = sqlx::query_scalar(
        "SELECT indisvalid FROM pg_index i \
           JOIN pg_class c ON c.oid = i.indexrelid \
           JOIN pg_namespace n ON n.oid = c.relnamespace \
          WHERE n.nspname = 'tenant_a' AND c.relname = 'ix_outbox_claimable'",
    )
    .fetch_optional(&pool)
    .await
    .unwrap();
    assert_eq!(
        tenant_a_valid,
        Some(true),
        "tenant_a's own index must be left untouched throughout"
    );
}

async fn migrate_creates_schema_and_bookkeeping_table() {
    let pool = common::fresh_unmigrated_db().await;

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("first migrate succeeds");

    let outbox_exists: bool = sqlx::query_scalar("SELECT to_regclass('reliar.outbox') IS NOT NULL")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert!(outbox_exists, "migrate() must create reliar.outbox");

    let bookkeeping_exists: bool =
        sqlx::query_scalar("SELECT to_regclass('reliar._migrations') IS NOT NULL")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(
        bookkeeping_exists,
        "migrate() must record versions in reliar._migrations, never _sqlx_migrations"
    );

    let host_migrations_exists: bool =
        sqlx::query_scalar("SELECT to_regclass('public._sqlx_migrations') IS NOT NULL")
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(
        !host_migrations_exists,
        "migrate() must never write to the shared, one-per-database _sqlx_migrations table"
    );

    // P-32 (ADR 0049 Amendment A) — a migrated database has no `sequence` column on `outbox` at
    // all, catalog-checked rather than assumed from any dropped index's absence.
    let sequence_column_exists: bool = sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM information_schema.columns \
          WHERE table_schema = 'reliar' AND table_name = 'outbox' AND column_name = 'sequence')",
    )
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(
        !sequence_column_exists,
        "outbox must have no sequence column after migrate()"
    );
}

async fn migrate_is_idempotent() {
    let pool = common::fresh_unmigrated_db().await;
    let options = MigrateOptions::default();

    migrate(&pool, options).await.expect("first call");

    migrate(&pool, options)
        .await
        .expect("second call observes Ok(())");
}

/// P-N24 (ADR 0040 amendment A) — two concurrent `migrate()` calls on the same schema both
/// return `Ok`, migrations are applied exactly once (`reliar._migrations` has one row per
/// migration file), and — the part a deadlock disguised as success would
/// still get wrong —
/// `ix_outbox_claimable` exists **and** `indisvalid`, since Reliar's own poll-based lock replaces
/// sqlx's blocking one specifically because that lock used to deadlock against `0002`'s
/// `CREATE INDEX CONCURRENTLY` the moment two callers' runs overlapped.
async fn migrate_serializes_concurrent_callers() {
    let pool = common::fresh_unmigrated_db().await;
    let options = MigrateOptions::default();

    let (first, second) = tokio::join!(migrate(&pool, options), migrate(&pool, options));

    first.expect("first concurrent caller succeeds");
    second.expect("second concurrent caller succeeds, serialized by Reliar's own advisory lock");

    let outbox_exists: bool = sqlx::query_scalar("SELECT to_regclass('reliar.outbox') IS NOT NULL")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert!(outbox_exists);

    let applied_count: i64 = sqlx::query_scalar("SELECT count(*) FROM reliar._migrations")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(
        usize::try_from(applied_count).expect("applied_count is never negative"),
        ALL_MIGRATIONS.migrations.len(),
        "concurrent callers must not double-apply or partially apply the migration set"
    );

    let claimable_valid: Option<bool> = sqlx::query_scalar(
        "SELECT indisvalid FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid \
          WHERE c.relname = 'ix_outbox_claimable'",
    )
    .fetch_optional(&pool)
    .await
    .unwrap();
    assert_eq!(
        claimable_valid,
        Some(true),
        "ix_outbox_claimable must exist and be valid after two concurrent migrate() calls"
    );
}

/// P-N24b (ADR 0040 amendment A) — four concurrent callers on the **same** schema all serialize
/// cleanly through Reliar's own poll-based lock (the lock id is schema-scoped, but same-schema
/// callers still race on `CREATE SCHEMA IF NOT EXISTS` and one `_migrations` table, so they must
/// still fully serialize, not merely avoid deadlock).
async fn four_concurrent_callers_on_the_same_schema_all_succeed() {
    let pool = common::fresh_unmigrated_db().await;
    let options = MigrateOptions::default();

    let (a, b, c, d) = tokio::join!(
        migrate(&pool, options),
        migrate(&pool, options),
        migrate(&pool, options),
        migrate(&pool, options),
    );

    a.expect("caller a");
    b.expect("caller b");
    c.expect("caller c");
    d.expect("caller d");

    let applied_count: i64 = sqlx::query_scalar("SELECT count(*) FROM reliar._migrations")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(
        usize::try_from(applied_count).expect("applied_count is never negative"),
        ALL_MIGRATIONS.migrations.len()
    );
}

/// P-N24b (ADR 0040 amendment A) — two concurrent callers into **different** schemas both
/// succeed without waiting on each other: the lock id is schema-scoped (not database-scoped), and
/// `CREATE INDEX CONCURRENTLY` builds on different tables do not wait on each other's snapshots
/// (`PROC_IN_SAFE_IC`, PostgreSQL 14+).
async fn two_concurrent_callers_into_different_schemas_both_succeed() {
    let pool = common::fresh_unmigrated_db().await;

    let (a, b) = tokio::join!(
        migrate(&pool, MigrateOptions::default().schema("tenant_a")),
        migrate(&pool, MigrateOptions::default().schema("tenant_b")),
    );

    a.expect("tenant_a's migrate() succeeds");
    b.expect("tenant_b's migrate() succeeds");

    for schema in ["tenant_a", "tenant_b"] {
        let outbox_exists: bool = sqlx::query_scalar(
            "SELECT EXISTS (SELECT 1 FROM information_schema.tables \
               WHERE table_schema = $1 AND table_name = 'outbox')",
        )
        .bind(schema)
        .fetch_one(&pool)
        .await
        .unwrap();

        assert!(outbox_exists, "{schema} must have its own outbox table");

        // P-N24b — what actually proves two concurrent `CONCURRENTLY` builds
        // coexisted without waiting on each other (`PROC_IN_SAFE_IC`), not merely that each
        // schema has *an* index of that name: both must be `indisvalid`.
        let claimable_valid: Option<bool> = sqlx::query_scalar(
            "SELECT indisvalid FROM pg_index i \
               JOIN pg_class c ON c.oid = i.indexrelid \
               JOIN pg_namespace n ON n.oid = c.relnamespace \
              WHERE n.nspname = $1 AND c.relname = 'ix_outbox_claimable'",
        )
        .bind(schema)
        .fetch_optional(&pool)
        .await
        .unwrap();
        assert_eq!(
            claimable_valid,
            Some(true),
            "{schema}'s ix_outbox_claimable must exist and be valid"
        );
    }
}

/// ADR 0040 amendment A — `migrate()` used to run its `SET search_path` on a connection borrowed
/// from the *caller's* pool via `pool.acquire()`; sqlx 0.9 does not reset session-level GUCs
/// when a connection is released back to a pool, so that connection would silently keep
/// resolving unqualified names against the migrated schema for the rest of its life, for any
/// unrelated query the host later happened to run on it. Proven here by holding several
/// connections from the same pool open *simultaneously* (forcing the pool to open that many
/// distinct physical connections, not just reuse one) both before and after `migrate()`, and
/// asserting every one of them still reports the pool's original `search_path`.
async fn migrate_does_not_leak_search_path_into_the_callers_pool_connections() {
    let pool = common::fresh_unmigrated_db().await;

    let original: String = sqlx::query_scalar("SELECT current_setting('search_path')")
        .fetch_one(&pool)
        .await
        .unwrap();

    migrate(&pool, MigrateOptions::default().schema("tenant_a"))
        .await
        .expect("migrate into a non-default schema");

    let mut held = Vec::new();

    for _ in 0..5 {
        held.push(pool.acquire().await.unwrap());
    }

    for mut conn in held {
        let observed: String = sqlx::query_scalar("SELECT current_setting('search_path')")
            .fetch_one(&mut *conn)
            .await
            .unwrap();

        assert_eq!(
            observed, original,
            "a pool connection's search_path must be untouched by migrate()"
        );
    }
}

async fn unmigrated_pool_has_no_outbox_table() {
    let pool = common::fresh_unmigrated_db().await;

    let outbox_exists: bool = sqlx::query_scalar("SELECT to_regclass('reliar.outbox') IS NOT NULL")
        .fetch_one(&pool)
        .await
        .unwrap();

    assert!(
        !outbox_exists,
        "a pool nobody has called migrate() on must have no outbox table"
    );
}

#[allow(
    clippy::too_many_lines,
    reason = "one flat Trial-registration list, one line per scenario — splitting it would scatter \
              the file's own table of contents across several functions with no reuse to justify it"
)]
pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "migrate::migrate_creates_schema_and_bookkeeping_table",
            move || {
                rt.block_on(migrate_creates_schema_and_bookkeeping_table());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test("migrate::migrate_is_idempotent", move || {
            rt.block_on(migrate_is_idempotent());
            Ok(())
        }),
        libtest_mimic::Trial::test(
            "migrate::migrate_serializes_concurrent_callers",
            move || {
                rt.block_on(migrate_serializes_concurrent_callers());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::four_concurrent_callers_on_the_same_schema_all_succeed",
            move || {
                rt.block_on(four_concurrent_callers_on_the_same_schema_all_succeed());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::two_concurrent_callers_into_different_schemas_both_succeed",
            move || {
                rt.block_on(two_concurrent_callers_into_different_schemas_both_succeed());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test("migrate::unmigrated_pool_has_no_outbox_table", move || {
            rt.block_on(unmigrated_pool_has_no_outbox_table());
            Ok(())
        }),
        libtest_mimic::Trial::test(
            "migrate::migrate_does_not_leak_search_path_into_the_callers_pool_connections",
            move || {
                rt.block_on(migrate_does_not_leak_search_path_into_the_callers_pool_connections());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::migrate_upgrades_a_populated_0001_only_database_and_claim_still_works",
            move || {
                rt.block_on(
                    migrate_upgrades_a_populated_0001_only_database_and_claim_still_works(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::migrate_refuses_0003_when_claimable_index_is_missing_and_recovers",
            move || {
                rt.block_on(migrate_refuses_0003_when_claimable_index_is_missing_and_recovers());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::migrate_refuses_0003_when_claimable_index_is_invalid_and_recovers",
            move || {
                rt.block_on(migrate_refuses_0003_when_claimable_index_is_invalid_and_recovers());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::migrate_0003_guard_is_schema_scoped_and_ignores_another_schemas_valid_index",
            move || {
                rt.block_on(
                    migrate_0003_guard_is_schema_scoped_and_ignores_another_schemas_valid_index(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::migrate_refuses_0014_when_a_transient_index_is_missing_and_recovers",
            move || {
                rt.block_on(migrate_refuses_0014_when_a_transient_index_is_missing_and_recovers());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::migrate_refuses_0014_when_a_transient_index_is_invalid_and_recovers",
            move || {
                rt.block_on(migrate_refuses_0014_when_a_transient_index_is_invalid_and_recovers());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::migrate_refuses_0016_when_the_transient_index_is_missing_and_recovers",
            move || {
                rt.block_on(
                    migrate_refuses_0016_when_the_transient_index_is_missing_and_recovers(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "migrate::ix_outbox_claimable_survives_dropping_locked_until",
            move || {
                rt.block_on(ix_outbox_claimable_survives_dropping_locked_until());
                Ok(())
            },
        ),
    ]
}