reliar-store-postgres 0.7.0

//! Inbox contract §5.2 I-P12 (the `search_path` half — the server-version half extends
//! `server_version_floor`'s existing sub-18 container instead of duplicating it) —
//! `PostgresInboxStore::connect` fails fast, naming the configured schema and the observed
//! `search_path`, when `inbox` does not resolve there; it never carries a DSN, host or
//! credential.

use crate::common;

use reliar_core::{Classify, FailureKind, MessageId};
use reliar_inbox::{InboxClaim, InboxScope, InboxStore};
use reliar_store_postgres::{PostgresInboxError, PostgresInboxSettings, PostgresInboxStore};
use sqlx::PgPool;
use sqlx::postgres::PgConnectOptions;

async fn pool_without_search_path() -> PgPool {
    let base = common::fresh_unmigrated_db().await;

    reliar_store_postgres::migrate(&base, reliar_store_postgres::MigrateOptions::default())
        .await
        .unwrap();

    // Deliberately `public` only, not the server default: a local role happening to share the
    // schema's name would otherwise resolve it via Postgres's own `"$user", public` default.
    let options: PgConnectOptions = base
        .connect_options()
        .as_ref()
        .clone()
        .options([("search_path", "public")]);

    PgPool::connect_with(options).await.unwrap()
}

async fn connect_fails_fast_without_search_path() {
    let pool = pool_without_search_path().await;
    let err = PostgresInboxStore::connect(pool, PostgresInboxSettings::default())
        .await
        .unwrap_err();
    let text = err.to_string();

    assert!(
        !text.contains("postgres://") && !text.contains("127.0.0.1"),
        "PostgresInboxError::Display must carry no DSN/host, got: {text}"
    );
    assert_eq!(err.kind(), FailureKind::Permanent);

    match err {
        PostgresInboxError::SchemaNotOnSearchPath {
            configured,
            observed,
        } => {
            assert_eq!(configured, "reliar");
            assert!(
                text.contains("reliar") && text.contains(&observed),
                "Display must name both the configured schema and the observed search_path"
            );
        }
        other => panic!("expected PostgresInboxError::SchemaNotOnSearchPath, got {other:?}"),
    }
}

/// m16: `configured_exists = false` (the relation itself is missing — `migrate()` never ran) must
/// report the sharper [`PostgresInboxError::NotMigrated`], not `SchemaNotOnSearchPath` — mirrors
/// `PostgresOutboxStore::connect`'s own ordering.
async fn connect_reports_not_migrated_when_the_relation_is_entirely_missing() {
    let pool = common::fresh_unmigrated_db().await;

    let err = PostgresInboxStore::connect(pool, PostgresInboxSettings::default())
        .await
        .unwrap_err();

    match err {
        PostgresInboxError::NotMigrated { schema } => assert_eq!(schema, "reliar"),
        other => panic!("expected PostgresInboxError::NotMigrated, got {other:?}"),
    }
}

async fn connect_succeeds_once_search_path_resolves_inbox() {
    let pool = common::fresh_db().await;

    PostgresInboxStore::connect(pool, PostgresInboxSettings::default())
        .await
        .expect("inbox resolves under the default reliar,public search_path");
}

/// B2 regression: `claim_sets_search_path(true)` must fix **both** `claim` and `complete`, not
/// just `claim`. Simulates a caller whose own transaction-local `search_path` never includes
/// `reliar` (e.g. a multi-tenant host that sets its own schema first) — `claim`'s wrap wasn't the
/// bug (it already restored the caller's value before returning); `complete`'s unwrapped
/// `UPDATE inbox` was, since by the time it ran, `claim` had already put the caller's non-`reliar`
/// value back. Proves claim → complete → commit succeeds end to end, and that the caller's
/// (non-`reliar`) `search_path` is exactly what is left in place afterward.
///
/// The business write is schema-qualified into `public` explicitly (never sharing
/// `common::inbox::create_business_table`'s unqualified helper) — it must stay resolvable under
/// the caller's own `public`-only `search_path`, which is exactly the scenario under test.
async fn claim_and_complete_work_when_the_callers_search_path_lacks_the_schema() {
    let pool = common::fresh_db().await;

    sqlx::query(
        "CREATE TABLE public.business_events (id bigserial PRIMARY KEY, value bigint NOT NULL)",
    )
    .execute(&pool)
    .await
    .unwrap();
    let settings = PostgresInboxSettings::default().claim_sets_search_path(true);
    let store = PostgresInboxStore::connect(pool.clone(), settings)
        .await
        .unwrap();
    let scope = InboxScope::new("orders-projection").unwrap();
    let id = MessageId::new();

    let mut tx = pool.begin().await.unwrap();
    sqlx::query("SELECT set_config('search_path', 'public', true)")
        .execute(&mut *tx)
        .await
        .unwrap();

    let claim = store
        .claim(&mut tx, &scope, crate::common::inbox::message(id))
        .await
        .unwrap();
    assert_eq!(claim, InboxClaim::Claimed { attempt: 1 });

    sqlx::query("INSERT INTO business_events (value) VALUES ($1)")
        .bind(1_i64)
        .execute(&mut *tx)
        .await
        .unwrap();

    store.complete(&mut tx, &scope, id).await.unwrap();

    let after: String = sqlx::query_scalar("SELECT current_setting('search_path')")
        .fetch_one(&mut *tx)
        .await
        .unwrap();
    assert_eq!(
        after, "public",
        "complete must restore the caller's own search_path, not leave `reliar` on it"
    );

    tx.commit().await.unwrap();

    let business_rows: i64 = sqlx::query_scalar("SELECT count(*) FROM public.business_events")
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(business_rows, 1);
    let record = store.find(&scope, id).await.unwrap().unwrap();
    assert!(record.completed_at.is_some());
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "inbox_schema_verification::connect_fails_fast_without_search_path",
            move || {
                rt.block_on(connect_fails_fast_without_search_path());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_schema_verification::connect_succeeds_once_search_path_resolves_inbox",
            move || {
                rt.block_on(connect_succeeds_once_search_path_resolves_inbox());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_schema_verification::connect_reports_not_migrated_when_the_relation_is_entirely_missing",
            move || {
                rt.block_on(connect_reports_not_migrated_when_the_relation_is_entirely_missing());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "inbox_schema_verification::claim_and_complete_work_when_the_callers_search_path_lacks_the_schema",
            move || {
                rt.block_on(
                    claim_and_complete_work_when_the_callers_search_path_lacks_the_schema(),
                );
                Ok(())
            },
        ),
    ]
}