PiCloud/crates/manager-core/tests/migration_queue_messages.rs

//! v1.1.9 migration smoke test: applies all manager-core migrations to
//! `DATABASE_URL` and verifies the `queue_messages` + `queue_trigger_details`
//! shape (table existence, columns, partial indexes). Skips cleanly when
//! `DATABASE_URL` is unset.
//!
//! Heavier schema assertions live in the schema_snapshot golden — this
//! test is a quick gate for the migration itself + an easy place to
//! exercise the partial-index conditions.

#![allow(clippy::needless_pass_by_value)]

use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;

async fn pool_or_skip() -> Option<PgPool> {
    let Ok(url) = std::env::var("DATABASE_URL") else {
        picloud_test_support::abort_if_db_required("migration_queue_messages");
        eprintln!("migration_queue_messages: DATABASE_URL unset — skipping");
        return None;
    };
    let pool = PgPoolOptions::new()
        .max_connections(2)
        .connect(&url)
        .await
        .expect("connect to DATABASE_URL");
    sqlx::migrate!("./migrations")
        .run(&pool)
        .await
        .expect("apply migrations");
    Some(pool)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queue_messages_table_exists_with_expected_columns() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let rows: Vec<(String, String, String)> = sqlx::query_as(
        "SELECT column_name, data_type, is_nullable \
         FROM information_schema.columns \
         WHERE table_name = 'queue_messages' \
         ORDER BY ordinal_position",
    )
    .fetch_all(&pool)
    .await
    .expect("column read");

    let names: Vec<&str> = rows.iter().map(|(n, _, _)| n.as_str()).collect();
    for required in [
        "id",
        "app_id",
        "queue_name",
        "payload",
        "enqueued_at",
        "deliver_after",
        "claim_token",
        "claimed_at",
        "attempt",
        "max_attempts",
        "enqueued_by_principal",
    ] {
        assert!(
            names.contains(&required),
            "queue_messages should have column {required}, columns are {names:?}"
        );
    }

    // Verify nullability on the right columns.
    let nullable: std::collections::HashMap<String, String> =
        rows.into_iter().map(|(n, _, nl)| (n, nl)).collect();
    assert_eq!(nullable["app_id"], "NO");
    assert_eq!(nullable["queue_name"], "NO");
    assert_eq!(nullable["payload"], "NO");
    assert_eq!(nullable["claim_token"], "YES");
    assert_eq!(nullable["deliver_after"], "YES");
    assert_eq!(nullable["enqueued_by_principal"], "YES");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queue_trigger_details_table_exists() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let rows: Vec<(String,)> = sqlx::query_as(
        "SELECT column_name FROM information_schema.columns \
         WHERE table_name = 'queue_trigger_details' \
         ORDER BY ordinal_position",
    )
    .fetch_all(&pool)
    .await
    .expect("read");
    let names: Vec<&str> = rows.iter().map(|(n,)| n.as_str()).collect();
    for required in [
        "trigger_id",
        "queue_name",
        "visibility_timeout_secs",
        "last_fired_at",
    ] {
        assert!(
            names.contains(&required),
            "queue_trigger_details should have {required}, columns are {names:?}"
        );
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queue_widens_trigger_kind_and_outbox_source_kind() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    // The triggers.kind constraint must accept 'queue'.
    sqlx::query(
        "DO $$ BEGIN PERFORM 1 FROM pg_constraint WHERE conname = 'triggers_kind_check'; END $$;",
    )
    .execute(&pool)
    .await
    .expect("constraint exists");

    let (def,): (String,) = sqlx::query_as(
        "SELECT pg_get_constraintdef(oid) FROM pg_constraint \
         WHERE conname = 'triggers_kind_check'",
    )
    .fetch_one(&pool)
    .await
    .expect("read constraint");
    assert!(
        def.contains("'queue'"),
        "triggers_kind_check should admit 'queue', got: {def}"
    );

    let (outbox_def,): (String,) = sqlx::query_as(
        "SELECT pg_get_constraintdef(oid) FROM pg_constraint \
         WHERE conname = 'outbox_source_kind_check'",
    )
    .fetch_one(&pool)
    .await
    .expect("read constraint");
    assert!(
        outbox_def.contains("'invoke'"),
        "outbox_source_kind_check should admit 'invoke', got: {outbox_def}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn queue_messages_dispatch_index_is_partial() {
    let Some(pool) = pool_or_skip().await else {
        return;
    };
    let (defn,): (String,) = sqlx::query_as(
        "SELECT indexdef FROM pg_indexes \
         WHERE indexname = 'idx_queue_messages_dispatch'",
    )
    .fetch_one(&pool)
    .await
    .expect("read index");
    // Partial WHERE (claim_token IS NULL) is the key feature; assert
    // it's there so a future migration can't quietly drop it.
    assert!(
        defn.contains("claim_token IS NULL"),
        "idx_queue_messages_dispatch should be partial on claim_token IS NULL, got: {defn}"
    );
}