dbos 0.6.0

//! Applies the migration corpus to a real database.
//!
//! Everything else about the corpus is checked by reading it — file counts, slot arithmetic,
//! header numbering. Those tests are cheap, but they share a blind spot: they assert things
//! *about* SQL without ever asking a database whether it is valid. A `%s` bound to the wrong
//! value, a variant selected for the wrong dialect, or a statement that parses everywhere but
//! only executes on one backend all pass a static check and fail here.
//!
//! These run on the raw lane, because the migrations are the thing under test.

use dbos::sysdb::migrations::{Dialect, Migration, build_migrations};
use sqlx::{PgPool, Row};

use dbos_test_support::{Backend, raw_database};

fn dialect_for(backend: Backend) -> Dialect {
    match backend {
        Backend::Postgres => Dialect::Postgres,
        Backend::Cockroach => Dialect::Cockroach,
    }
}

/// Applies a migration list in order, honouring guards.
///
/// A stripped-down stand-in for `runner::run`, used where a test needs to apply an explicit
/// slice of the corpus — the runner always applies everything from the recorded version. It
/// deliberately keeps no version state.
async fn apply_all(pool: &PgPool, schema: &str, migrations: &[Migration]) {
    for m in migrations {
        if m.sql.trim().is_empty() {
            continue;
        }
        if let Some(guard) = m.guard {
            // Row presence only — never decode the value. The guard selects a literal `1`,
            // which is INT4 on PostgreSQL and INT8 on CockroachDB, so no single Rust type
            // decodes it on both.
            let already = sqlx::query(guard)
                .bind(schema)
                .fetch_optional(pool)
                .await
                .unwrap_or_else(|e| panic!("migration {} guard failed: {e}", m.version));
            if already.is_some() {
                continue;
            }
        }
        // Whole files, never split on `;` — several carry semicolons inside `$$` blocks.
        // `AssertSqlSafe` because sqlx 0.9 wants a `&'static str` otherwise; the string is
        // rendered from the embedded corpus with only the quoted schema interpolated.
        sqlx::raw_sql(sqlx::AssertSqlSafe(m.sql.clone()))
            .execute(pool)
            .await
            .unwrap_or_else(|e| panic!("migration {} failed: {e}\n---\n{}", m.version, m.sql));
    }
}

async fn table_names(pool: &PgPool, schema: &str) -> Vec<String> {
    let rows = sqlx::query("SELECT table_name FROM information_schema.tables WHERE table_schema = $1 ORDER BY table_name")
        .bind(schema)
        .fetch_all(pool)
        .await
        .expect("failed to list tables");
    rows.iter().map(|r| r.get::<String, _>(0)).collect()
}

async fn columns_of(pool: &PgPool, schema: &str, table: &str) -> Vec<String> {
    let rows = sqlx::query("SELECT column_name FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2 ORDER BY column_name")
        .bind(schema)
        .bind(table)
        .fetch_all(pool)
        .await
        .expect("failed to list columns");
    rows.iter().map(|r| r.get::<String, _>(0)).collect()
}

async fn trigger_names(pool: &PgPool, schema: &str) -> Vec<String> {
    sqlx::query_scalar(
        "SELECT DISTINCT trigger_name FROM information_schema.triggers \
         WHERE trigger_schema = $1 ORDER BY 1",
    )
    .bind(schema)
    .fetch_all(pool)
    .await
    .expect("failed to list triggers")
}

async fn function_names(pool: &PgPool, schema: &str) -> Vec<String> {
    sqlx::query_scalar(
        "SELECT p.proname FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace \
         WHERE n.nspname = $1 ORDER BY 1",
    )
    .bind(schema)
    .fetch_all(pool)
    .await
    .expect("failed to list functions")
}

/// The whole corpus applies cleanly and produces the expected tables.
#[tokio::test]
async fn the_corpus_applies_to_a_real_database() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;

    let migrations = build_migrations(schema, dialect_for(db.backend()), true);
    assert_eq!(
        migrations.len(),
        dbos::sysdb::migrations::SHARED_MIGRATIONS as usize,
        "one entry per migration, including the padding to the shared base",
    );
    // Through the real runner, so this covers the path production uses.
    dbos::sysdb::migrations::runner::run(&pool, schema, true)
        .await
        .expect("migration failed");

    let tables = table_names(&pool, schema).await;
    for expected in [
        "application_versions",
        "event_dispatch_kv",
        "notifications",
        "operation_outputs",
        "queues",
        "streams",
        "workflow_events",
        "workflow_events_history",
        "workflow_schedules",
        "workflow_status",
    ] {
        assert!(
            tables.iter().any(|t| t == expected),
            "missing table {expected}; got {tables:?}",
        );
    }
}

/// Columns added by later migrations are present, so the sequence actually ran through.
///
/// Applying migration 1 alone would satisfy the table check above; this asserts that the
/// tail of the sequence took effect too.
#[tokio::test]
async fn late_migrations_take_effect() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();

    apply_all(
        &pool,
        schema,
        &build_migrations(schema, dialect_for(db.backend()), true),
    )
    .await;

    let cols = columns_of(&pool, schema, "workflow_status").await;
    for expected in [
        "queue_partition_key",        // 2
        "forked_from",                // 4
        "owner_xid",                  // 7
        "serialization",              // 11
        "delay_until_epoch_ms",       // 16
        "rate_limited",               // 33
        "completed_at",               // 36
        "attributes",                 // 40
        "schedule_name",              // 41
        "debounce_deadline_epoch_ms", // 42
    ] {
        assert!(
            cols.iter().any(|c| c == expected),
            "workflow_status is missing {expected}; got {cols:?}",
        );
    }
}

/// The shared series lands on every table it names.
///
/// Separate from the check above because these migrations are not this SDK's own: versions from
/// 100 up mean the same DDL in every SDK sharing the database, and the whole point of the padding
/// is that they arrive at those numbers. A database migrated by Rust and then opened by another SDK
/// must find the columns already there rather than try to add them again.
#[tokio::test]
async fn the_shared_series_adds_application_name_everywhere() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();

    apply_all(
        &pool,
        schema,
        &build_migrations(schema, dialect_for(db.backend()), true),
    )
    .await;

    for table in [
        "workflow_status",      // 100
        "queues",               // 101
        "workflow_schedules",   // 102
        "application_versions", // 103
        "operation_outputs",    // 104
    ] {
        let cols = columns_of(&pool, schema, table).await;
        assert!(
            cols.iter().any(|c| c == "application_name"),
            "{table} is missing application_name; got {cols:?}",
        );
    }
}

/// An enqueue can name the application it belongs to, and can still decline to.
///
/// Migration 105's parameter is last and optional, which is what lets an older SDK — or a client
/// with no application name to claim — keep calling the function it already knows. Both halves
/// are exercised: naming an application stores it, omitting the argument leaves the row
/// unclaimed rather than failing.
#[tokio::test]
async fn an_enqueue_may_claim_an_application_or_leave_it_unclaimed() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();

    apply_all(
        &pool,
        schema,
        &build_migrations(schema, dialect_for(db.backend()), true),
    )
    .await;

    // Sixteen arguments is the pre-105 call, and it must still resolve against the new function.
    let unclaimed: String = sqlx::query_scalar(
        r#"SELECT "dbos".enqueue_workflow($1::TEXT, $2::TEXT, ARRAY[]::JSON[], '{}'::JSON,
             NULL, NULL, $3::TEXT, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)"#,
    )
    .bind("a-workflow")
    .bind("a-queue")
    .bind("unclaimed-id")
    .fetch_one(&pool)
    .await
    .expect("the sixteen-argument call should still resolve");

    let claimed: String = sqlx::query_scalar(
        r#"SELECT "dbos".enqueue_workflow($1::TEXT, $2::TEXT, ARRAY[]::JSON[], '{}'::JSON,
             NULL, NULL, $3::TEXT, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, $4::TEXT)"#,
    )
    .bind("a-workflow")
    .bind("a-queue")
    .bind("claimed-id")
    .bind("alpha")
    .fetch_one(&pool)
    .await
    .expect("the seventeen-argument call should enqueue");

    for (id, expected) in [(unclaimed, None), (claimed, Some("alpha".to_string()))] {
        let stored: Option<String> = sqlx::query_scalar(
            r#"SELECT "application_name" FROM "dbos"."workflow_status" WHERE "workflow_uuid" = $1"#,
        )
        .bind(&id)
        .fetch_one(&pool)
        .await
        .expect("the enqueued workflow should exist");
        assert_eq!(stored, expected, "{id} recorded the wrong application");
    }
}

/// Migration 113's `enqueue_workflow` writes the inputs to `workflow_input` and nothing to the
/// legacy `workflow_status.inputs`, the same as every SDK's own enqueue.
#[tokio::test]
async fn an_sql_enqueue_writes_its_inputs_to_the_payload_table() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();
    apply_all(
        &pool,
        schema,
        &build_migrations(schema, dialect_for(db.backend()), true),
    )
    .await;

    let id: String = sqlx::query_scalar(
        r#"SELECT "dbos".enqueue_workflow($1::TEXT, $2::TEXT, ARRAY['7'::JSON], '{}'::JSON,
             NULL, NULL, $3::TEXT)"#,
    )
    .bind("a-workflow")
    .bind("a-queue")
    .bind("sql-enqueued")
    .fetch_one(&pool)
    .await
    .expect("the enqueue should succeed");

    let legacy: Option<String> = sqlx::query_scalar(
        r#"SELECT "inputs" FROM "dbos"."workflow_status" WHERE "workflow_uuid" = $1"#,
    )
    .bind(&id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert_eq!(legacy, None, "the legacy column must not be written");

    let inputs: Option<String> = sqlx::query_scalar(
        r#"SELECT "inputs" FROM "dbos"."workflow_input" WHERE "workflow_uuid" = $1"#,
    )
    .bind(&id)
    .fetch_one(&pool)
    .await
    .expect("the inputs should be in workflow_input");
    let inputs: serde_json::Value =
        serde_json::from_str(&inputs.expect("inputs were written")).unwrap();
    assert_eq!(
        inputs,
        serde_json::json!({"positionalArgs": [7], "namedArgs": {}})
    );
}

/// Migrations 106 and 107 both land, splitting version-name uniqueness by owner.
///
/// The two are separate migrations because they build differently: 106's predicate matches no
/// existing row and goes in with the version write, while 107's matches every one of them and has
/// to build online. Applying the corpus is what proves the second half actually completed —
/// PostgreSQL leaves a failed concurrent build in place, marked invalid.
#[tokio::test]
async fn version_uniqueness_is_split_by_owner() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();

    apply_all(
        &pool,
        schema,
        &build_migrations(schema, dialect_for(db.backend()), true),
    )
    .await;

    let indexes: Vec<String> = sqlx::query_scalar(
        "SELECT i.relname FROM pg_index ix \
         JOIN pg_class i ON i.oid = ix.indexrelid \
         JOIN pg_class t ON t.oid = ix.indrelid \
         JOIN pg_namespace n ON n.oid = t.relnamespace \
         WHERE n.nspname = $1 AND t.relname = 'application_versions' AND ix.indisvalid",
    )
    .bind(schema)
    .fetch_all(&pool)
    .await
    .expect("failed to list indexes");

    for expected in [
        "uq_application_versions_owner_version",     // 106
        "uq_application_versions_unclaimed_version", // 107
    ] {
        assert!(
            indexes.iter().any(|i| i == expected),
            "{expected} is missing or invalid; got {indexes:?}",
        );
    }
}

/// Splits an index definition, as `pg_indexes.indexdef` reports it, into its key columns and its
/// included columns.
///
/// PostgreSQL prints `(a, b) INCLUDE (c)` and CockroachDB `(a ASC, b ASC) STORING (c)`, so this
/// reads past both spellings to the column lists the schema has to agree on.
fn index_columns(definition: &str) -> (Vec<String>, Vec<String>) {
    fn list(after: &str) -> Vec<String> {
        let inner = &after[..after.find(')').expect("unterminated column list")];
        inner
            .split(',')
            .map(|c| c.trim().trim_end_matches(" ASC").to_owned())
            .collect()
    }
    let keys = definition
        .split_once("USING btree (")
        .map(|(_, rest)| list(rest))
        .expect("a btree column list");
    let included = ["INCLUDE (", "STORING ("]
        .iter()
        .find_map(|marker| definition.split_once(marker))
        .map(|(_, rest)| list(rest))
        .unwrap_or_default();
    (keys, included)
}

/// Migrations 115 to 120 replace three indexes, and only the replacements are left.
///
/// Each pair builds the new index before dropping the one it supersedes, both online. The
/// replacements must have exactly the agreed key columns, `application_name` as an included
/// column, and the predicate of the index they replace, because every implementation sharing the
/// database builds them under the same names. Only valid indexes count as built, since PostgreSQL
/// leaves a failed concurrent build in place, marked invalid; a superseded index must be gone
/// whether it is valid or not, since an interrupted concurrent drop leaves it invalid.
#[tokio::test]
async fn the_index_rebuilds_replace_their_predecessors() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();

    apply_all(
        &pool,
        schema,
        &build_migrations(schema, dialect_for(db.backend()), true),
    )
    .await;

    let definitions: Vec<(String, String)> = sqlx::query_as(
        "SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = $1 \
         AND tablename IN ('workflow_status', 'operation_outputs')",
    )
    .bind(schema)
    .fetch_all(&pool)
    .await
    .expect("failed to list index definitions");
    let valid: Vec<String> = sqlx::query_scalar(
        "SELECT i.relname FROM pg_index ix \
         JOIN pg_class i ON i.oid = ix.indexrelid \
         JOIN pg_class t ON t.oid = ix.indrelid \
         JOIN pg_namespace n ON n.oid = t.relnamespace \
         WHERE n.nspname = $1 AND t.relname IN ('workflow_status', 'operation_outputs') \
         AND ix.indisvalid",
    )
    .bind(schema)
    .fetch_all(&pool)
    .await
    .expect("failed to list valid indexes");

    let in_flight = ["ENQUEUED", "PENDING"].as_slice();
    for (built, keys, predicate, dropped) in [
        (
            "idx_workflow_status_in_flight_v2", // 115
            ["queue_name", "status", "priority", "created_at"].as_slice(),
            in_flight,
            "idx_workflow_status_in_flight", // 116
        ),
        (
            "idx_workflow_status_partition_dequeue_v3", // 117
            [
                "queue_name",
                "status",
                "queue_partition_key",
                "priority",
                "created_at",
                "workflow_uuid",
            ]
            .as_slice(),
            ["ENQUEUED", "PENDING", "queue_partition_key IS NOT NULL"].as_slice(),
            "idx_workflow_status_partition_dequeue_v2", // 118
        ),
        (
            "idx_operation_outputs_completed_at_function_name_v2", // 119
            ["completed_at_epoch_ms", "function_name"].as_slice(),
            [].as_slice(),
            "idx_operation_outputs_completed_at_function_name", // 120
        ),
    ] {
        assert!(
            valid.iter().any(|i| i == built),
            "{built} is missing or invalid; got {valid:?}",
        );
        let definition = &definitions
            .iter()
            .find(|(name, _)| name == built)
            .expect("a valid index has a definition")
            .1;
        let (actual_keys, included) = index_columns(definition);
        assert_eq!(actual_keys, keys, "{built} key columns: {definition}");
        assert_eq!(
            included,
            ["application_name"],
            "{built} included columns: {definition}"
        );
        match predicate {
            [] => assert!(
                !definition.contains(" WHERE "),
                "{built} should be a full index: {definition}"
            ),
            fragments => {
                let (_, condition) = definition
                    .split_once(" WHERE ")
                    .unwrap_or_else(|| panic!("{built} should be partial: {definition}"));
                for fragment in fragments {
                    assert!(
                        condition.contains(fragment),
                        "{built} predicate lacks {fragment}: {definition}"
                    );
                }
            }
        }

        assert!(
            !definitions.iter().any(|(name, _)| name == dropped),
            "{dropped} should have been dropped; got {definitions:?}",
        );
    }
}

/// Migration 121 adds `notifications.consumed_by_function_id` as a 32-bit integer on both
/// backends, matching every other step-id column.
///
/// `INTEGER` would be 64-bit on CockroachDB, and a 32-bit read of the column would then fail
/// there and only there, so the type is asserted rather than the column's presence alone.
#[tokio::test]
async fn migration_121_adds_a_32_bit_consumer_step_column() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    dbos::sysdb::migrations::runner::run(&pool, schema, true)
        .await
        .expect("migration failed");

    let data_type: String = sqlx::query_scalar(
        "SELECT data_type FROM information_schema.columns \
         WHERE table_schema = $1 AND table_name = 'notifications' \
           AND column_name = 'consumed_by_function_id'",
    )
    .bind(schema)
    .fetch_one(&pool)
    .await
    .expect("consumed_by_function_id is missing");
    assert_eq!(data_type, "integer");
}

/// Migration 122 builds the workflow-timeout sweep's index: valid, keyed on the deadline alone,
/// and partial over the unsettled statuses with a deadline, on both backends.
///
/// The sweep's plan depends on the predicate: its status literals and its `IS NOT NULL` have to be
/// ones the planner can prove from the sweep's own `WHERE`.
#[tokio::test]
async fn migration_122_builds_the_deadline_index() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    dbos::sysdb::migrations::runner::run(&pool, schema, true)
        .await
        .expect("migration failed");

    let definition: String = sqlx::query_scalar(
        "SELECT indexdef FROM pg_indexes WHERE schemaname = $1 \
         AND tablename = 'workflow_status' AND indexname = 'idx_workflow_status_deadline'",
    )
    .bind(schema)
    .fetch_one(&pool)
    .await
    .expect("idx_workflow_status_deadline is missing");
    let valid: bool = sqlx::query_scalar(
        "SELECT ix.indisvalid FROM pg_index ix \
         JOIN pg_class i ON i.oid = ix.indexrelid \
         JOIN pg_namespace n ON n.oid = i.relnamespace \
         WHERE n.nspname = $1 AND i.relname = 'idx_workflow_status_deadline'",
    )
    .bind(schema)
    .fetch_one(&pool)
    .await
    .expect("failed to read the index's validity");
    assert!(valid, "the index is invalid: {definition}");

    let (keys, included) = index_columns(&definition);
    assert_eq!(keys, ["workflow_deadline_epoch_ms"], "{definition}");
    assert!(included.is_empty(), "{definition}");
    let (_, condition) = definition
        .split_once(" WHERE ")
        .unwrap_or_else(|| panic!("the index should be partial: {definition}"));
    for fragment in [
        "ENQUEUED",
        "PENDING",
        "DELAYED",
        "workflow_deadline_epoch_ms IS NOT NULL",
    ] {
        assert!(
            condition.contains(fragment),
            "the predicate lacks {fragment}: {definition}"
        );
    }
}

/// Migrations 43 and 44 leave only the notifications trigger behind.
///
/// Migration 1 installs notification and workflow-events triggers and 39 adds the streams
/// one; 43 and 44 then drop two of the three, because those writes are coalesced by the
/// notifier off the write path. Only `notifications` still fires per row.
#[tokio::test]
async fn only_the_notifications_trigger_survives() {
    let db = raw_database().await;
    if db.backend() == Backend::Cockroach {
        return; // CockroachDB installs none of them.
    }
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();
    apply_all(
        &pool,
        schema,
        &build_migrations(schema, Dialect::Postgres, true),
    )
    .await;

    assert_eq!(
        trigger_names(&pool, schema).await,
        ["dbos_notifications_trigger"]
    );

    // The v1 partition index is created by 45 and dropped by 47, v2 is created by 46 and dropped
    // by 118; only 117's v3 survives.
    let indexes: Vec<String> = sqlx::query_scalar(
        "SELECT indexname FROM pg_indexes WHERE schemaname = $1 \
         AND indexname LIKE 'idx_workflow_status_partition%' ORDER BY 1",
    )
    .bind(schema)
    .fetch_all(&pool)
    .await
    .expect("failed to list indexes");
    assert_eq!(
        indexes,
        vec!["idx_workflow_status_partition_dequeue_v3".to_owned()],
    );
}

/// Migrations 43 and 44 drop their triggers and functions even when the process running them
/// doesn't listen.
///
/// A database migrated up to 42 with notifications on has the workflow-events and streams
/// triggers. A process that later takes it past 44 with notifications off must still drop them;
/// otherwise 43 and 44 are recorded as applied and the triggers stay forever.
#[tokio::test]
async fn the_trigger_drops_run_without_listen_notify() {
    let db = raw_database().await;
    if db.backend() == Backend::Cockroach {
        return; // CockroachDB installs none of them.
    }
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();
    let with_notify = build_migrations(schema, Dialect::Postgres, true);
    let without = build_migrations(schema, Dialect::Postgres, false);
    assert_eq!(with_notify[41].version, 42);
    apply_all(&pool, schema, &with_notify[..42]).await;
    apply_all(&pool, schema, &without[42..]).await;

    assert_eq!(
        trigger_names(&pool, schema).await,
        ["dbos_notifications_trigger"]
    );
    let functions = function_names(&pool, schema).await;
    for dropped in ["workflow_events_function", "streams_function"] {
        assert!(
            !functions.iter().any(|f| f == dropped),
            "{dropped} should have been dropped, got {functions:?}",
        );
    }
}

/// Migration 10 is skipped, because migration 1 already created the primary key.
///
/// Its guard is the only conditional in the corpus, and the branch it protects is one this
/// implementation should never take — migration 1 creates `message_uuid ... PRIMARY KEY`
/// inline. If this ever fails, either migration 1 changed or the guard is wrong.
#[tokio::test]
async fn migration_ten_is_a_no_op_on_a_schema_we_created() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();

    let migrations = build_migrations(schema, dialect_for(db.backend()), true);
    let ten = migrations.iter().find(|m| m.version == 10).unwrap();
    let guard = ten.guard.expect("migration 10 carries a guard");

    // Apply everything up to and including 9, then ask the guard.
    apply_all(&pool, schema, &migrations[..9]).await;
    // Presence, not value: the literal `1` is INT4 on PostgreSQL and INT8 on CockroachDB.
    let already = sqlx::query(guard)
        .bind(schema)
        .fetch_optional(&pool)
        .await
        .expect("the guard query should execute on both backends");
    assert!(
        already.is_some(),
        "migration 1 should already have created the notifications primary key",
    );
}

/// A dialect's list is the same length as any other's, empties included.
///
/// Versions are positional: a migration that does not apply on this backend has to render
/// empty rather than be dropped, or every migration after it shifts by one.
#[tokio::test]
async fn dialects_agree_on_the_number_of_versions() {
    let pg = build_migrations("dbos", Dialect::Postgres, true);
    let crdb = build_migrations("dbos", Dialect::Cockroach, true);
    let no_notify = build_migrations("dbos", Dialect::Postgres, false);

    assert_eq!(pg.len(), crdb.len());
    assert_eq!(pg.len(), no_notify.len());
    for (a, b) in pg.iter().zip(crdb.iter()) {
        assert_eq!(a.version, b.version);
    }

    // CockroachDB has no ALTER FUNCTION ... SET and no triggers here, so 20 and 39 are
    // no-ops — but they still occupy their version numbers.
    assert!(
        crdb.iter()
            .find(|m| m.version == 20)
            .unwrap()
            .sql
            .is_empty()
    );
    assert!(
        crdb.iter()
            .find(|m| m.version == 39)
            .unwrap()
            .sql
            .is_empty()
    );
    assert!(!pg.iter().find(|m| m.version == 20).unwrap().sql.is_empty());

    // No migration is marked online on CockroachDB, where DDL is online anyway.
    assert!(crdb.iter().all(|m| !m.online));
    assert_eq!(
        pg.iter().filter(|m| m.online).count(),
        pg.iter()
            .filter(|m| dbos::sysdb::migrations::is_online(m.version))
            .count(),
    );
}

/// The corpus applies with notifications turned off, which is a supported configuration.
///
/// This is the path where migration 1 sheds its trigger half, 20 hardens only the functions
/// that exist, and 39 becomes a no-op. Getting that wrong fails here and nowhere else:
/// hardening a function that was never installed is an error the static tests cannot see.
#[tokio::test]
async fn the_corpus_applies_without_listen_notify() {
    let db = raw_database().await;
    if db.backend() == Backend::Cockroach {
        return; // CockroachDB never installs the triggers; the Postgres run covers this.
    }
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    sqlx::raw_sql(r#"CREATE SCHEMA IF NOT EXISTS "dbos""#)
        .execute(&pool)
        .await
        .unwrap();

    apply_all(
        &pool,
        schema,
        &build_migrations(schema, Dialect::Postgres, false),
    )
    .await;

    let triggers: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM information_schema.triggers WHERE trigger_schema = $1",
    )
    .bind(schema)
    .fetch_one(&pool)
    .await
    .expect("failed to count triggers");
    assert_eq!(
        triggers, 0,
        "no triggers should exist with notifications off"
    );

    // The *functions* go too — migration 20 must not try to harden what migration 1 never
    // installed — while the SQL-client functions, which have nothing to do with notifications,
    // must still be there.
    let functions = function_names(&pool, schema).await;
    for absent in [
        "notifications_function",
        "workflow_events_function",
        "streams_function",
    ] {
        assert!(
            !functions.iter().any(|f| f == absent),
            "{absent} should not exist with notifications off, got {functions:?}",
        );
    }
    for present in ["enqueue_workflow", "send_message"] {
        assert!(
            functions.iter().any(|f| f == present),
            "{present} is unrelated to notifications and should exist, got {functions:?}",
        );
    }

    // The tables still arrive; only the notification plumbing is absent.
    assert!(
        table_names(&pool, schema)
            .await
            .iter()
            .any(|t| t == "workflow_status")
    );
}

/// Migrations 124 and 125 build the latest-version lookup's indexes: valid, not unique, keyed on
/// `version_timestamp` newest first and on `(application_name, version_timestamp DESC)`, with no
/// included columns, on both backends; and applying them again is harmless.
#[tokio::test]
async fn migrations_124_and_125_build_the_version_indexes() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    dbos::sysdb::migrations::runner::run(&pool, schema, true)
        .await
        .expect("migration failed");
    let again: Vec<Migration> = build_migrations(schema, dialect_for(db.backend()), true)
        .into_iter()
        .filter(|m| matches!(m.version, 124 | 125))
        .collect();
    assert_eq!(again.len(), 2);
    apply_all(&pool, schema, &again).await;

    for (index, expected) in [
        (
            "idx_application_versions_timestamp",
            &["version_timestamp DESC"][..],
        ),
        (
            "idx_application_versions_owner_timestamp",
            &["application_name", "version_timestamp DESC"][..],
        ),
    ] {
        let definition: String = sqlx::query_scalar(
            "SELECT indexdef FROM pg_indexes WHERE schemaname = $1 \
             AND tablename = 'application_versions' AND indexname = $2",
        )
        .bind(schema)
        .bind(index)
        .fetch_one(&pool)
        .await
        .unwrap_or_else(|e| panic!("{index} is missing: {e}"));
        let valid: bool = sqlx::query_scalar(
            "SELECT ix.indisvalid FROM pg_index ix \
             JOIN pg_class i ON i.oid = ix.indexrelid \
             JOIN pg_namespace n ON n.oid = i.relnamespace \
             WHERE n.nspname = $1 AND i.relname = $2",
        )
        .bind(schema)
        .bind(index)
        .fetch_one(&pool)
        .await
        .expect("failed to read the index's validity");
        assert!(valid, "the index is invalid: {definition}");
        assert!(
            definition.starts_with("CREATE INDEX"),
            "the index should not be unique: {definition}"
        );

        let (keys, included) = index_columns(&definition);
        assert_eq!(keys, expected, "{definition}");
        assert!(included.is_empty(), "{definition}");
    }
}

/// Migration 123 adds `creator_xid`, a nullable text column with no default value, and applying
/// it twice is harmless.
#[tokio::test]
async fn migration_123_adds_the_creator_token_column() {
    let db = raw_database().await;
    let pool = db.pool().await;
    let schema = dbos::sysdb::DEFAULT_SCHEMA;
    dbos::sysdb::migrations::runner::run(&pool, schema, true)
        .await
        .expect("migration failed");

    let (data_type, nullable, default): (String, String, Option<String>) = sqlx::query_as(
        "SELECT data_type, is_nullable, column_default FROM information_schema.columns \
         WHERE table_schema = $1 AND table_name = 'workflow_status' \
           AND column_name = 'creator_xid'",
    )
    .bind(schema)
    .fetch_one(&pool)
    .await
    .expect("creator_xid is missing");
    assert_eq!(data_type, "text");
    assert_eq!(nullable, "YES");
    assert!(
        default
            .as_deref()
            .is_none_or(|default| default.eq_ignore_ascii_case("NULL")),
        "creator_xid defaults to {default:?}"
    );

    let statement = dbos::sysdb::migrations::SOURCES
        .iter()
        .find(|source| source.version == 123)
        .expect("migration 123 is registered")
        .sql
        .replace(
            "{{schema}}",
            &dbos::sysdb::migrations::quote_identifier(schema),
        );
    sqlx::raw_sql(sqlx::AssertSqlSafe(statement))
        .execute(&pool)
        .await
        .expect("applying migration 123 again failed");
}