use std::collections::BTreeSet;
use std::fs;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use runledger_postgres::{
    MIGRATOR, SchemaCompatibilityError, ensure_schema_compatible_after_idempotency_cutover,
    migrate_after_idempotency_cutover,
};
use runledger_test_support::{TestDbConnectionBudgetPermit, acquire_test_db_connection_budget};
use serde_json::{Value, json};
use sqlx::migrate::{Migrate, MigrateError};
use sqlx::{PgPool, postgres::PgPoolOptions};
use testcontainers::{
    ContainerAsync, GenericImage, ImageExt, core::ContainerPort, runners::AsyncRunner,
};

const DEFAULT_POSTGRES_IMAGE: &str = "postgres:18";
const POSTGRES_USER: &str = "runledger";
const POSTGRES_PASSWORD: &str = "runledger";
const POSTGRES_DB: &str = "postgres";
const TEST_ADMIN_DATABASE_URL_ENV: &str = "RUNLEDGER_TEST_ADMIN_DATABASE_URL";
const MAX_POSTGRES_BOOTSTRAP_ATTEMPTS: u8 = 40;
const MAX_PORT_RESOLVE_ATTEMPTS: u8 = 10;
const ENQUEUE_REQUEST_CUTOVER_VERSION: i64 = 202605220001;
const TEST_HARNESS_POOL_CONNECTIONS: u32 = 4;
const TEST_HARNESS_ADMIN_CONNECTIONS: u32 = 1;

static DATABASE_COUNTER: AtomicU64 = AtomicU64::new(1);

struct TestHarness {
    admin_url: String,
    database_name: String,
    pool: PgPool,
    _connection_budget: TestDbConnectionBudgetPermit,
    _container: Option<ContainerAsync<GenericImage>>,
}

impl TestHarness {
    async fn fresh(prefix: &str) -> Self {
        let connection_budget = acquire_test_db_connection_budget(
            TEST_HARNESS_POOL_CONNECTIONS + TEST_HARNESS_ADMIN_CONNECTIONS,
        )
        .await;
        let (admin_url, container) =
            if let Ok(admin_url) = std::env::var(TEST_ADMIN_DATABASE_URL_ENV) {
                wait_for_postgres(&admin_url).await;
                (admin_url, None)
            } else {
                let image_ref = std::env::var("RUNLEDGER_TEST_PG_IMAGE")
                    .unwrap_or_else(|_| DEFAULT_POSTGRES_IMAGE.into());
                let (repository, tag) = parse_image_ref(&image_ref);
                let container = GenericImage::new(repository, tag)
                    .with_exposed_port(ContainerPort::Tcp(5432))
                    .with_env_var("POSTGRES_USER", POSTGRES_USER)
                    .with_env_var("POSTGRES_PASSWORD", POSTGRES_PASSWORD)
                    .with_env_var("POSTGRES_DB", POSTGRES_DB)
                    .start()
                    .await
                    .expect("start postgres container");

                let port = resolve_host_port(&container, 5432).await;
                let admin_url = postgres_admin_url(port);
                wait_for_postgres(&admin_url).await;
                (admin_url, Some(container))
            };

        let database_name = build_database_name(prefix);
        let admin_pool = connect_admin_pool(&admin_url)
            .await
            .expect("connect admin postgres");
        let create_sql = format!("CREATE DATABASE {database_name}");
        sqlx::raw_sql(&create_sql)
            .execute(&admin_pool)
            .await
            .expect("create ephemeral database");
        admin_pool.close().await;

        let database_url = with_database_name(&admin_url, &database_name);
        let pool = PgPoolOptions::new()
            .max_connections(TEST_HARNESS_POOL_CONNECTIONS)
            .connect(&database_url)
            .await
            .expect("connect postgres");

        Self {
            admin_url,
            database_name,
            pool,
            _connection_budget: connection_budget,
            _container: container,
        }
    }

    async fn teardown(self) {
        self.pool.close().await;

        let admin_pool = connect_admin_pool(&self.admin_url)
            .await
            .expect("connect admin postgres");
        sqlx::query(
            "SELECT pg_terminate_backend(pid)
             FROM pg_stat_activity
             WHERE datname = $1
               AND pid <> pg_backend_pid()",
        )
        .bind(&self.database_name)
        .fetch_all(&admin_pool)
        .await
        .expect("terminate database backends");

        let drop_sql = format!("DROP DATABASE IF EXISTS {}", self.database_name);
        sqlx::raw_sql(&drop_sql)
            .execute(&admin_pool)
            .await
            .expect("drop ephemeral database");
        admin_pool.close().await;
    }
}

#[tokio::test]
async fn migrate_applies_bundled_schema_to_fresh_database() {
    let harness = TestHarness::fresh("runledger_pg_migrate").await;

    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("apply migrations");
    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("repeat migrate after constraints are validated");
    ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect("schema should validate after migrate");
    assert!(
        idempotency_cutover_constraints_valid(&harness.pool).await,
        "migrate should validate idempotency cutover constraints after legacy check passes"
    );

    let migrations_row_count =
        sqlx::query_scalar::<_, i64>("SELECT count(*) FROM _sqlx_migrations")
            .fetch_one(&harness.pool)
            .await
            .expect("count applied migrations");
    assert_eq!(
        migrations_row_count,
        runledger_migration_versions().len() as i64
    );

    let recorded_runledger_versions = sqlx::query_scalar::<_, i64>(
        "SELECT version FROM runledger_migration_history ORDER BY version",
    )
    .fetch_all(&harness.pool)
    .await
    .expect("list recorded runledger migrations");
    assert_eq!(recorded_runledger_versions, runledger_migration_versions());

    let metrics_view_exists = sqlx::query_scalar::<_, bool>(
        "SELECT EXISTS (
             SELECT 1
             FROM information_schema.views
             WHERE table_schema = 'public'
               AND table_name = 'job_metrics_rollup'
         )",
    )
    .fetch_one(&harness.pool)
    .await
    .expect("query metrics view");
    assert!(metrics_view_exists);

    harness.teardown().await;
}

#[tokio::test]
async fn migrate_ignores_unrelated_sqlx_history() {
    let harness = TestHarness::fresh("runledger_pg_migrate_shared").await;
    seed_unrelated_sqlx_migration(&harness.pool, 202401010001, false).await;

    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("apply runledger migrations alongside app migrations");
    ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect("schema should validate when unrelated migrations are present");

    let migration_versions =
        sqlx::query_scalar::<_, i64>("SELECT version FROM _sqlx_migrations ORDER BY version")
            .fetch_all(&harness.pool)
            .await
            .expect("list applied migrations");
    assert!(migration_versions.contains(&202401010001));
    for version in runledger_migration_versions() {
        assert!(migration_versions.contains(&version));
    }

    harness.teardown().await;
}

#[tokio::test]
async fn ensure_schema_compatible_rejects_fresh_database_without_migrations() {
    let harness = TestHarness::fresh("runledger_pg_validate").await;

    let error = ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect_err("validation should fail before migrations are applied");
    match &error {
        SchemaCompatibilityError::MissingMigrationHistory {
            required_first_migration_version,
        } => {
            assert_eq!(*required_first_migration_version, 202603280001);
        }
        other => panic!("unexpected migration validation error: {other}"),
    }
    assert!(
        error.to_string().contains("_sqlx_migrations"),
        "missing history error should explain the required table"
    );

    let migrations_table_exists =
        sqlx::query_scalar::<_, bool>("SELECT to_regclass('_sqlx_migrations') IS NOT NULL")
            .fetch_one(&harness.pool)
            .await
            .expect("check for migrations table");
    assert!(
        !migrations_table_exists,
        "schema compatibility validation must not create _sqlx_migrations"
    );

    harness.teardown().await;
}

#[tokio::test]
async fn ensure_schema_compatible_ignores_unrelated_sqlx_history() {
    let harness = TestHarness::fresh("runledger_pg_validate_shared").await;
    seed_unrelated_sqlx_migration(&harness.pool, 202401010001, false).await;
    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("apply migrations");

    ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect("validation should ignore unrelated migration versions");

    harness.teardown().await;
}

#[tokio::test]
async fn ensure_schema_compatible_ignores_unrelated_sqlx_history_with_runledger_description() {
    let harness = TestHarness::fresh("runledger_pg_validate_shared_named").await;
    seed_sqlx_migration(
        &harness.pool,
        202401010001,
        "runledger host app schema",
        true,
        vec![1_u8, 2, 3, 4],
    )
    .await;
    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("apply migrations");

    ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect("validation should ignore unrelated descriptions");

    harness.teardown().await;
}

#[test]
fn vendored_migration_copies_match_root_migrations() {
    let workspace_root = Path::new(env!("CARGO_MANIFEST_DIR"))
        .parent()
        .expect("postgres crate should live under workspace root");
    let root_migrations = workspace_root.join("migrations");

    assert_migration_dir_matches(
        &root_migrations,
        &workspace_root.join("runledger-postgres/migrations"),
    );
    assert_migration_dir_matches(
        &root_migrations,
        &workspace_root.join("runledger-test-support/migrations"),
    );
}

#[tokio::test]
async fn ensure_schema_compatible_rejects_legacy_idempotency_rows() {
    let harness = TestHarness::fresh("runledger_pg_validate_legacy").await;
    apply_runledger_migrations_before_cutover(&harness.pool).await;
    seed_legacy_idempotency_rows(&harness.pool).await;
    apply_enqueue_request_cutover_migration(&harness.pool).await;
    apply_runledger_migrations_after_cutover(&harness.pool).await;

    let error = ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect_err("validation should reject legacy keyed rows without snapshots");
    assert_legacy_idempotency_snapshot_error(&error, 1, 1);

    harness.teardown().await;
}

#[tokio::test]
async fn migrate_after_idempotency_cutover_rejects_legacy_idempotency_rows() {
    let harness = TestHarness::fresh("runledger_pg_migrate_legacy").await;
    apply_runledger_migrations_before_cutover(&harness.pool).await;
    seed_legacy_idempotency_rows(&harness.pool).await;

    let error = migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect_err("migrate should reject legacy keyed rows without snapshots");
    assert_legacy_idempotency_snapshot_error(&error, 1, 1);

    harness.teardown().await;
}

#[tokio::test]
async fn workflow_results_migration_preserves_existing_enqueue_request_snapshots() {
    let harness = TestHarness::fresh("runledger_pg_workflow_results_preserve").await;
    apply_runledger_migrations_before_cutover(&harness.pool).await;
    apply_enqueue_request_cutover_migration(&harness.pool).await;
    insert_workflow_row_with_pre_result_snapshot(&harness.pool).await;

    apply_runledger_migrations_after_cutover(&harness.pool).await;

    let snapshot = sqlx::query_scalar::<_, Value>(
        "SELECT enqueue_request
         FROM workflow_runs
         WHERE idempotency_key = 'legacy-result-snapshot'
         LIMIT 1",
    )
    .fetch_one(&harness.pool)
    .await
    .expect("load preserved workflow snapshot");
    assert_eq!(
        snapshot,
        json!({
            "metadata": {},
            "steps": []
        })
    );
    assert!(snapshot.get("result_step_key").is_none());

    harness.teardown().await;
}

#[tokio::test]
async fn enqueue_request_cutover_constraints_reject_new_legacy_idempotency_rows() {
    let harness = TestHarness::fresh("runledger_pg_cutover_constraints").await;
    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("apply migrations");

    seed_legacy_job_definition(&harness.pool).await;

    let job_error = insert_legacy_job_row(&harness.pool, "legacy-job-after-cutover")
        .await
        .expect_err("job_queue constraint should reject keyed rows without enqueue_request");
    assert!(
        job_error
            .to_string()
            .contains("ck_job_queue_idempotency_enqueue_request"),
        "unexpected job constraint error: {job_error}"
    );

    let workflow_error = insert_legacy_workflow_row(&harness.pool, "legacy-workflow-after-cutover")
        .await
        .expect_err("workflow_runs constraint should reject keyed rows without enqueue_request");
    assert!(
        workflow_error
            .to_string()
            .contains("ck_workflow_runs_idempotency_enqueue_request"),
        "unexpected workflow constraint error: {workflow_error}"
    );

    harness.teardown().await;
}

#[tokio::test]
async fn migrate_rejects_conflicting_sqlx_version_namespace() {
    let harness = TestHarness::fresh("runledger_pg_migrate_conflict").await;
    let conflicting_version = runledger_migration_versions()
        .into_iter()
        .next()
        .expect("runledger should include at least one up migration");
    seed_unrelated_sqlx_migration(&harness.pool, conflicting_version, true).await;

    let error = migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect_err("migrate should reject conflicting version namespace");
    assert!(
        matches!(
            &error,
            SchemaCompatibilityError::Incompatible(MigrateError::VersionMismatch(version))
                if *version == conflicting_version
        ),
        "unexpected migration error: {error}"
    );

    harness.teardown().await;
}

#[tokio::test]
async fn migrate_rejects_newer_runledger_migration_history() {
    let harness = TestHarness::fresh("runledger_pg_migrate_newer").await;
    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("apply current migrations");

    let newer_version = runledger_migration_versions()
        .into_iter()
        .max()
        .expect("runledger should include at least one up migration")
        + 1;
    seed_runledger_migration_history(&harness.pool, newer_version).await;

    let error = migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect_err("migrate should reject newer runledger history");
    assert!(
        matches!(
            &error,
            SchemaCompatibilityError::Incompatible(MigrateError::VersionMissing(version))
                if *version == newer_version
        ),
        "unexpected migration error: {error}"
    );

    harness.teardown().await;
}

#[tokio::test]
async fn ensure_schema_compatible_rejects_conflicting_sqlx_version_namespace() {
    let harness = TestHarness::fresh("runledger_pg_validate_conflict").await;
    let conflicting_version = runledger_migration_versions()
        .into_iter()
        .next()
        .expect("runledger should include at least one up migration");
    seed_unrelated_sqlx_migration(&harness.pool, conflicting_version, true).await;

    let error = ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect_err("validation should reject conflicting version namespace");
    assert!(
        matches!(
            &error,
            SchemaCompatibilityError::Incompatible(MigrateError::VersionMismatch(version))
                if *version == conflicting_version
        ),
        "unexpected schema compatibility error: {error}"
    );

    harness.teardown().await;
}

#[tokio::test]
async fn ensure_schema_compatible_rejects_newer_runledger_migration_history() {
    let harness = TestHarness::fresh("runledger_pg_validate_newer").await;
    migrate_after_idempotency_cutover(&harness.pool)
        .await
        .expect("apply current migrations");

    let newer_version = runledger_migration_versions()
        .into_iter()
        .max()
        .expect("runledger should include at least one up migration")
        + 1;
    seed_runledger_migration_history(&harness.pool, newer_version).await;

    let error = ensure_schema_compatible_after_idempotency_cutover(&harness.pool)
        .await
        .expect_err("validation should reject newer runledger history");
    assert!(
        matches!(
            &error,
            SchemaCompatibilityError::Incompatible(MigrateError::VersionMissing(version))
                if *version == newer_version
        ),
        "unexpected schema compatibility error: {error}"
    );

    harness.teardown().await;
}

async fn connect_admin_pool(admin_url: &str) -> Result<PgPool, sqlx::Error> {
    PgPoolOptions::new()
        .max_connections(1)
        .connect(admin_url)
        .await
}

async fn resolve_host_port(container: &ContainerAsync<GenericImage>, internal_port: u16) -> u16 {
    for attempt in 1..=MAX_PORT_RESOLVE_ATTEMPTS {
        match container.get_host_port_ipv4(internal_port).await {
            Ok(port) => return port,
            Err(err) => {
                if attempt == MAX_PORT_RESOLVE_ATTEMPTS {
                    panic!(
                        "resolve mapped postgres port after {MAX_PORT_RESOLVE_ATTEMPTS} attempts: {err}"
                    );
                }
                tokio::time::sleep(Duration::from_millis(250)).await;
            }
        }
    }

    unreachable!()
}

async fn wait_for_postgres(admin_url: &str) {
    for attempt in 1..=MAX_POSTGRES_BOOTSTRAP_ATTEMPTS {
        match connect_admin_pool(admin_url).await {
            Ok(pool) => {
                if sqlx::query_scalar::<_, i64>("SELECT 1")
                    .fetch_one(&pool)
                    .await
                    .is_ok()
                {
                    pool.close().await;
                    return;
                }
                pool.close().await;
            }
            Err(err) => {
                if attempt == MAX_POSTGRES_BOOTSTRAP_ATTEMPTS {
                    panic!(
                        "connect postgres after {MAX_POSTGRES_BOOTSTRAP_ATTEMPTS} attempts: {err}"
                    );
                }
            }
        }

        tokio::time::sleep(Duration::from_millis(250)).await;
    }
}

fn postgres_admin_url(port: u16) -> String {
    format!("postgres://{POSTGRES_USER}:{POSTGRES_PASSWORD}@127.0.0.1:{port}/{POSTGRES_DB}")
}

fn parse_image_ref(image_ref: &str) -> (String, String) {
    let (name_and_tag, digest) = image_ref
        .split_once('@')
        .map_or((image_ref, None), |(name_and_tag, digest)| {
            (name_and_tag, Some(digest))
        });

    let last_slash = name_and_tag.rfind('/');
    let split_tag = name_and_tag
        .rfind(':')
        .filter(|index| last_slash.is_none_or(|slash| *index > slash));

    let (repository, mut tag) = split_tag.map_or_else(
        || (name_and_tag.to_owned(), String::from("latest")),
        |index| {
            (
                name_and_tag[..index].to_owned(),
                name_and_tag[index + 1..].to_owned(),
            )
        },
    );

    if let Some(digest) = digest {
        tag.push('@');
        tag.push_str(digest);
    }

    (repository, tag)
}

fn build_database_name(prefix: &str) -> String {
    let sanitized_prefix = sanitize_identifier(prefix);
    let compact_prefix = if sanitized_prefix.len() > 24 {
        sanitized_prefix[..24].to_string()
    } else {
        sanitized_prefix
    };
    let index = DATABASE_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("{}_{}_{}", compact_prefix, std::process::id(), index)
}

fn with_database_name(admin_url: &str, database_name: &str) -> String {
    let (base, _) = admin_url
        .rsplit_once('/')
        .expect("DATABASE_URL must include database name");
    format!("{base}/{database_name}")
}

fn sanitize_identifier(input: &str) -> String {
    let mut normalized = String::with_capacity(input.len() + 3);
    let mut previous_was_underscore = false;

    for ch in input.chars() {
        let mapped = if ch.is_ascii_alphanumeric() || ch == '_' {
            ch.to_ascii_lowercase()
        } else {
            '_'
        };

        if mapped == '_' {
            if !previous_was_underscore {
                normalized.push(mapped);
            }
            previous_was_underscore = true;
        } else {
            normalized.push(mapped);
            previous_was_underscore = false;
        }
    }

    if normalized.is_empty() {
        normalized.push_str("db");
    }
    if normalized
        .chars()
        .next()
        .is_some_and(|first| first.is_ascii_digit())
    {
        normalized.insert_str(0, "db_");
    }

    normalized
}

fn runledger_migration_versions() -> Vec<i64> {
    MIGRATOR
        .iter()
        .filter(|migration| migration.migration_type.is_up_migration())
        .map(|migration| migration.version)
        .collect()
}

fn assert_migration_dir_matches(expected_dir: &Path, actual_dir: &Path) {
    let expected_names = migration_file_names(expected_dir);
    let actual_names = migration_file_names(actual_dir);
    assert_eq!(
        actual_names,
        expected_names,
        "migration filenames in {} must match {}",
        actual_dir.display(),
        expected_dir.display()
    );

    for name in expected_names {
        let expected = fs::read(expected_dir.join(&name)).unwrap_or_else(|error| {
            panic!(
                "read expected migration {} from {}: {error}",
                name,
                expected_dir.display()
            )
        });
        let actual = fs::read(actual_dir.join(&name)).unwrap_or_else(|error| {
            panic!(
                "read actual migration {} from {}: {error}",
                name,
                actual_dir.display()
            )
        });
        assert_eq!(
            actual,
            expected,
            "migration {} in {} must match {}",
            name,
            actual_dir.display(),
            expected_dir.display()
        );
    }
}

fn migration_file_names(dir: &Path) -> BTreeSet<String> {
    fs::read_dir(dir)
        .unwrap_or_else(|error| panic!("read migration directory {}: {error}", dir.display()))
        .map(|entry| {
            let entry = entry.unwrap_or_else(|error| {
                panic!("read migration entry in {}: {error}", dir.display())
            });
            entry
                .file_name()
                .to_str()
                .unwrap_or_else(|| {
                    panic!(
                        "migration file name is not valid UTF-8 in {}",
                        dir.display()
                    )
                })
                .to_owned()
        })
        .filter(|name| name.ends_with(".sql"))
        .collect()
}

async fn apply_runledger_migrations_before_cutover(pool: &PgPool) {
    let mut conn = pool.acquire().await.expect("acquire migration connection");
    (*conn)
        .ensure_migrations_table()
        .await
        .expect("create sqlx migrations table");

    for migration in MIGRATOR
        .iter()
        .filter(|migration| migration.migration_type.is_up_migration())
        .filter(|migration| migration.version < ENQUEUE_REQUEST_CUTOVER_VERSION)
    {
        (*conn).apply(migration).await.unwrap_or_else(|error| {
            panic!(
                "apply pre-cutover Runledger migration {}: {error}",
                migration.version
            )
        });
    }
}

async fn apply_enqueue_request_cutover_migration(pool: &PgPool) {
    let mut conn = pool.acquire().await.expect("acquire migration connection");
    let migration = MIGRATOR
        .iter()
        .find(|migration| {
            migration.migration_type.is_up_migration()
                && migration.version == ENQUEUE_REQUEST_CUTOVER_VERSION
        })
        .expect("enqueue request cutover migration should exist");

    (*conn)
        .apply(migration)
        .await
        .expect("apply enqueue request cutover migration");
}

async fn apply_runledger_migrations_after_cutover(pool: &PgPool) {
    let mut conn = pool.acquire().await.expect("acquire migration connection");

    for migration in MIGRATOR
        .iter()
        .filter(|migration| migration.migration_type.is_up_migration())
        .filter(|migration| migration.version > ENQUEUE_REQUEST_CUTOVER_VERSION)
    {
        (*conn).apply(migration).await.unwrap_or_else(|error| {
            panic!(
                "apply post-cutover Runledger migration {}: {error}",
                migration.version
            )
        });
    }
}

fn assert_legacy_idempotency_snapshot_error(
    error: &SchemaCompatibilityError,
    job_count: i64,
    workflow_count: i64,
) {
    assert!(
        matches!(
            error,
            SchemaCompatibilityError::LegacyIdempotencySnapshotsMissing {
                job_count: actual_job_count,
                workflow_count: actual_workflow_count,
            } if *actual_job_count == job_count && *actual_workflow_count == workflow_count
        ),
        "unexpected schema compatibility error: {error}"
    );
}

async fn seed_runledger_migration_history(pool: &PgPool, version: i64) {
    sqlx::query(
        r#"
CREATE TABLE IF NOT EXISTS runledger_migration_history (
    version BIGINT PRIMARY KEY,
    recorded_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
        "#,
    )
    .execute(pool)
    .await
    .expect("create runledger migration history table");

    sqlx::query(
        r#"
INSERT INTO runledger_migration_history (version)
VALUES ($1)
        "#,
    )
    .bind(version)
    .execute(pool)
    .await
    .expect("insert runledger migration history");
}

async fn idempotency_cutover_constraints_valid(pool: &PgPool) -> bool {
    sqlx::query_scalar::<_, bool>(
        "SELECT COUNT(*) FILTER (WHERE c.convalidated) = 2
         FROM pg_constraint c
         JOIN pg_class t ON t.oid = c.conrelid
         WHERE (t.relname, c.conname) IN (
             ('job_queue', 'ck_job_queue_idempotency_enqueue_request'),
             ('workflow_runs', 'ck_workflow_runs_idempotency_enqueue_request')
         )",
    )
    .fetch_one(pool)
    .await
    .expect("check idempotency cutover constraint validation")
}

async fn seed_legacy_idempotency_rows(pool: &PgPool) {
    seed_legacy_job_definition(pool).await;
    insert_legacy_job_row(pool, "legacy-job")
        .await
        .expect("insert legacy job row");
    insert_legacy_workflow_row(pool, "legacy-workflow")
        .await
        .expect("insert legacy workflow row");
}

async fn seed_legacy_job_definition(pool: &PgPool) {
    sqlx::query(
        "INSERT INTO job_definitions (
            job_type,
            version,
            max_attempts,
            default_timeout_seconds,
            default_priority,
            is_enabled
         )
         VALUES ('jobs.test.legacy_cutover', 1, 3, 30, 100, true)",
    )
    .execute(pool)
    .await
    .expect("insert job definition");
}

async fn insert_legacy_job_row(
    pool: &PgPool,
    idempotency_key: &str,
) -> Result<sqlx::postgres::PgQueryResult, sqlx::Error> {
    sqlx::query(
        "INSERT INTO job_queue (
            job_type,
            payload,
            priority,
            max_attempts,
            timeout_seconds,
            idempotency_key
         )
         VALUES (
            'jobs.test.legacy_cutover',
            '{}'::jsonb,
            100,
            3,
            30,
            $1
         )",
    )
    .bind(idempotency_key)
    .execute(pool)
    .await
}

async fn insert_legacy_workflow_row(
    pool: &PgPool,
    idempotency_key: &str,
) -> Result<sqlx::postgres::PgQueryResult, sqlx::Error> {
    sqlx::query(
        "INSERT INTO workflow_runs (
            workflow_type,
            idempotency_key,
            metadata
         )
         VALUES (
            'workflow.test.legacy_cutover',
            $1,
            '{}'::jsonb
         )",
    )
    .bind(idempotency_key)
    .execute(pool)
    .await
}

async fn insert_workflow_row_with_pre_result_snapshot(pool: &PgPool) {
    sqlx::query(
        "INSERT INTO workflow_runs (
            workflow_type,
            idempotency_key,
            metadata,
            enqueue_request
         )
         VALUES (
            'workflow.test.pre_result_snapshot',
            'legacy-result-snapshot',
            '{}'::jsonb,
            $1::jsonb
         )",
    )
    .bind(json!({
        "metadata": {},
        "steps": []
    }))
    .execute(pool)
    .await
    .expect("insert pre-result workflow snapshot");
}

async fn seed_unrelated_sqlx_migration(pool: &PgPool, version: i64, success: bool) {
    seed_sqlx_migration(
        pool,
        version,
        "host app schema",
        success,
        vec![1_u8, 2, 3, 4],
    )
    .await;
}

async fn seed_sqlx_migration(
    pool: &PgPool,
    version: i64,
    description: &str,
    success: bool,
    checksum: Vec<u8>,
) {
    sqlx::query(
        r#"
CREATE TABLE IF NOT EXISTS _sqlx_migrations (
    version BIGINT PRIMARY KEY,
    description TEXT NOT NULL,
    installed_on TIMESTAMPTZ NOT NULL DEFAULT now(),
    success BOOLEAN NOT NULL,
    checksum BYTEA NOT NULL,
    execution_time BIGINT NOT NULL
)
        "#,
    )
    .execute(pool)
    .await
    .expect("create shared sqlx migrations table");

    sqlx::query(
        r#"
INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time)
VALUES ($1, $2, $3, $4, 0)
        "#,
    )
    .bind(version)
    .bind(description)
    .bind(success)
    .bind(checksum)
    .execute(pool)
    .await
    .expect("insert sqlx migration history");
}
