reliar-store-postgres 0.9.0

//! Contract §7 J1/J2/J4 — per-variant `Classify`, SQLSTATE-class-based classification of
//! `Database` errors on **every** path, and `migrate()`'s schema-identifier validation before it
//! ever reaches `dangerous_set_table_name`/`SET search_path` — the one remaining validation after
//! ADR 0047 (the store side never validated a schema name; it never sees one).

use crate::common;

use reliar_core::{Classify, FailureKind};
use reliar_outbox::{OutboxStore, WorkerId};
use reliar_store_postgres::{
    MigrateError, MigrateOptions, PostgresOutboxError, PostgresOutboxSettings, PostgresOutboxStore,
};

async fn undefined_table_maps_to_not_migrated_on_the_operational_path_and_is_permanent() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone());

    // `outbox` existed at construction; drop it out from under an already-constructed store —
    // construction issues no query (ADR 0047), so this is the first `42P01` the store ever sees.
    sqlx::query("DROP TABLE outbox")
        .execute(&pool)
        .await
        .unwrap();

    let err = store.stats().await.unwrap_err();

    match &err {
        PostgresOutboxError::NotMigrated { source } => {
            assert!(
                matches!(source, sqlx::Error::Database(db) if db.code().as_deref() == Some("42P01")),
                "expected SQLSTATE 42P01, got {source:?}"
            );
        }
        other => panic!("expected NotMigrated, got {other:?}"),
    }

    assert_eq!(err.kind(), FailureKind::Permanent);
}

async fn a_closed_pool_classifies_transient() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone());

    pool.close().await;

    let err = store.stats().await.unwrap_err();
    assert!(matches!(err, PostgresOutboxError::Database { .. }));
    assert_eq!(err.kind(), FailureKind::Transient);
}

async fn a_data_exception_sqlstate_classifies_permanent() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone());
    let envelopes = common::seed(&store, &pool, 1).await;

    let worker = WorkerId::generate();
    let batch = store
        .acquire(reliar_outbox::AcquireRequest::new(worker.clone()))
        .await
        .unwrap();
    let record = &batch.records[0];

    // `now() + (i64::MAX milliseconds)` overflows `timestamptz`'s representable range —
    // PostgreSQL raises SQLSTATE 22008 (datetime_field_overflow, class 22 = data exception),
    // which the table classifies **permanent**.
    let err = store
        .extend_lease(&worker, &[record.record_ref()], std::time::Duration::MAX)
        .await
        .unwrap_err();

    assert!(matches!(err, PostgresOutboxError::Database { .. }));
    assert_eq!(err.kind(), FailureKind::Permanent);
    let _ = envelopes;
}

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

    // "Foo" (P-N25, ADR 0040 §5): `migrate()` rejects an uppercase schema rather than silently
    // folding it — the only place this crate still validates a schema name (ADR 0047: the store
    // never sees one).
    for invalid in ["1leading_digit", "has-a-dash", "", "has space", "Foo"] {
        let result =
            reliar_store_postgres::migrate(&pool, MigrateOptions::default().schema(invalid)).await;

        match result {
            Err(MigrateError::InvalidSchema { schema }) => assert_eq!(schema, invalid),
            other => panic!("expected InvalidSchema for {invalid:?}, got {other:?}"),
        }
    }

    // Nothing was created by the rejected attempts.
    let outbox_exists: bool = sqlx::query_scalar("SELECT to_regclass('reliar.outbox') IS NOT NULL")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert!(!outbox_exists);
}

/// P-N25 (ADR 0040 §5) — the positive half: a lowercase schema using every character class the
/// grammar allows (`_`, digits, `$`) still migrates, and a store built over a pool scoped to that
/// schema's `search_path` constructs (synchronously, ADR 0047) with no schema of its own to
/// disagree with `migrate()`'s.
async fn a_lowercase_schema_using_every_allowed_character_class_still_works() {
    const SCHEMA: &str = "reliar_x$1";
    let pool = common::fresh_unmigrated_db().await;
    reliar_store_postgres::migrate(&pool, MigrateOptions::default().schema(SCHEMA))
        .await
        .expect("a lowercase name with '_', digits and '$' must still be accepted");

    let options: sqlx::postgres::PgConnectOptions = pool
        .connect_options()
        .as_ref()
        .clone()
        .options([("search_path", format!("{SCHEMA},public"))]);
    let scoped_pool = sqlx::PgPool::connect_with(options).await.unwrap();

    let store = PostgresOutboxStore::with_settings(scoped_pool, PostgresOutboxSettings::default());
    store
        .stats()
        .await
        .expect("the store must actually work against the scoped schema, not merely construct");
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "outbox_error_classification::undefined_table_maps_to_not_migrated_on_the_operational_path_and_is_permanent",
            move || {
                rt.block_on(
                    undefined_table_maps_to_not_migrated_on_the_operational_path_and_is_permanent(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_error_classification::a_closed_pool_classifies_transient",
            move || {
                rt.block_on(a_closed_pool_classifies_transient());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_error_classification::a_data_exception_sqlstate_classifies_permanent",
            move || {
                rt.block_on(a_data_exception_sqlstate_classifies_permanent());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_error_classification::migrate_rejects_an_invalid_schema_name_before_touching_the_database",
            move || {
                rt.block_on(migrate_rejects_an_invalid_schema_name_before_touching_the_database());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_error_classification::a_lowercase_schema_using_every_allowed_character_class_still_works",
            move || {
                rt.block_on(a_lowercase_schema_using_every_allowed_character_class_still_works());
                Ok(())
            },
        ),
    ]
}