//! Postgres-runner Chunk 3: verify the saga migration creates the
//! expected tables + indexes, is idempotent, and concurrent runs
//! don't race the schema.
//!
//! Gated on `DATABASE_URL` per the mire integration convention.

use mire::EventStore;
use sqlx::{PgPool, Row};

async fn maybe_store() -> Option<EventStore> {
    let url = std::env::var("DATABASE_URL")
        .ok()
        .filter(|s| !s.is_empty())?;
    let pool = PgPool::connect(&url).await.expect("connect to postgres");
    let store = EventStore::new(pool);
    store.migrate().await.expect("mire migrate");
    Some(store)
}

async fn table_exists(pool: &PgPool, name: &str) -> bool {
    let row = sqlx::query(
        "SELECT EXISTS (
            SELECT 1 FROM information_schema.tables
            WHERE table_schema = 'public' AND table_name = $1
        ) AS exists",
    )
    .bind(name)
    .fetch_one(pool)
    .await
    .expect("query");
    row.get::<bool, _>("exists")
}

async fn index_exists(pool: &PgPool, name: &str) -> bool {
    let row = sqlx::query(
        "SELECT EXISTS (
            SELECT 1 FROM pg_indexes
            WHERE schemaname = 'public' AND indexname = $1
        ) AS exists",
    )
    .bind(name)
    .fetch_one(pool)
    .await
    .expect("query");
    row.get::<bool, _>("exists")
}

#[tokio::test]
async fn migrate_creates_saga_tables() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    mire_sagas::migrate(&store).await.expect("saga migrate");

    assert!(table_exists(store.pool(), "es_saga_awaits").await);
    assert!(table_exists(store.pool(), "es_saga_signals").await);
}

#[tokio::test]
async fn migrate_creates_expected_indexes() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    mire_sagas::migrate(&store).await.expect("saga migrate");

    // Routing index — single seek per arriving response event.
    assert!(index_exists(store.pool(), "idx_saga_awaits_route").await);
    // Deadline index — drives the timer poller (partial).
    assert!(index_exists(store.pool(), "idx_saga_awaits_deadline").await);
    // Signal queue drain index.
    assert!(index_exists(store.pool(), "idx_saga_signals_pending").await);
}

#[tokio::test]
async fn migrate_is_idempotent() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    // Run twice — second run must not error (CREATE … IF NOT EXISTS).
    mire_sagas::migrate(&store).await.expect("first migrate");
    mire_sagas::migrate(&store).await.expect("second migrate");

    // Both tables still queryable after the repeated migration (the
    // CREATE … IF NOT EXISTS didn't drop or corrupt them). We do NOT
    // assert the tables are empty — other tests share this database and
    // may have in-flight rows; a global row-count assertion would be a
    // false-sharing flake.
    sqlx::query("SELECT 1 FROM es_saga_awaits LIMIT 1")
        .fetch_optional(store.pool())
        .await
        .expect("es_saga_awaits still queryable");
    sqlx::query("SELECT 1 FROM es_saga_signals LIMIT 1")
        .fetch_optional(store.pool())
        .await
        .expect("es_saga_signals still queryable");
}

#[tokio::test]
async fn migrate_concurrent_calls_serialise() {
    // The advisory lock should make two simultaneous migrates produce
    // the same end state as one. If the lock isn't held, we'd see
    // duplicate-key errors from `CREATE INDEX` races on pg_class.
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    let s1 = store.clone();
    let s2 = store.clone();
    let h1 = tokio::spawn(async move { mire_sagas::migrate(&s1).await });
    let h2 = tokio::spawn(async move { mire_sagas::migrate(&s2).await });

    h1.await.unwrap().expect("first migrate");
    h2.await.unwrap().expect("second migrate");

    assert!(table_exists(store.pool(), "es_saga_awaits").await);
    assert!(table_exists(store.pool(), "es_saga_signals").await);
}

#[tokio::test]
async fn awaits_table_has_expected_columns() {
    // Lock the column shape — the runner relies on it for INSERT/DELETE
    // SQL. Schema drift here would break the durable runner silently.
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    mire_sagas::migrate(&store).await.expect("migrate");

    let rows = sqlx::query(
        "SELECT column_name, data_type, is_nullable
         FROM information_schema.columns
         WHERE table_name = 'es_saga_awaits'
         ORDER BY ordinal_position",
    )
    .fetch_all(store.pool())
    .await
    .expect("describe");

    let cols: Vec<(String, String, String)> = rows
        .into_iter()
        .map(|r| {
            (
                r.get::<String, _>("column_name"),
                r.get::<String, _>("data_type"),
                r.get::<String, _>("is_nullable"),
            )
        })
        .collect();

    // Spot-check critical columns. Full structural check would be
    // brittle to add-only changes; this asserts the runner-relied
    // surface.
    let names: Vec<&str> = cols.iter().map(|(n, _, _)| n.as_str()).collect();
    for required in [
        "saga_name",
        "instance_id",
        "step",
        "event_type",
        "is_compensation",
        "shard_id",
    ] {
        assert!(
            names.contains(&required),
            "missing required column `{required}` in es_saga_awaits: got {names:?}"
        );
    }
}

#[tokio::test]
async fn signals_table_has_expected_columns() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    mire_sagas::migrate(&store).await.expect("migrate");

    let rows = sqlx::query(
        "SELECT column_name FROM information_schema.columns
         WHERE table_name = 'es_saga_signals' ORDER BY ordinal_position",
    )
    .fetch_all(store.pool())
    .await
    .expect("describe");

    let names: Vec<String> = rows
        .into_iter()
        .map(|r| r.get::<String, _>("column_name"))
        .collect();
    for required in [
        "id",
        "saga_name",
        "instance_id",
        "signal_type",
        "payload",
        "shard_id",
    ] {
        assert!(
            names.contains(&required.to_string()),
            "missing column `{required}`: got {names:?}"
        );
    }
}
