reliar-store-postgres 0.9.0

//! §43.A.31 — `migrate(&pool, MigrateOptions { schema })` into a **non-default** schema puts
//! both `outbox` and `_migrations` in it (nothing Reliar owns lands in `public`); a pool whose
//! URL carries that schema first on `search_path` then runs enqueue → claim with no
//! `set_config` statement at all (ADR 0047: the store never sets `search_path`), while
//! `PostgresOutboxStore::new` over a pool *without* that `search_path` constructs successfully
//! and fails at its **first call** with `NotMigrated`.

use crate::common;

use crate::common::OrderCreated;
use reliar_core::Envelope;
use reliar_outbox::{AcquireRequest, OutboxEnqueue, OutboxStore, WorkerId};
use reliar_store_postgres::{
    MigrateOptions, PostgresOutboxError, PostgresOutboxSettings, PostgresOutboxStore, migrate,
};
use sqlx::PgPool;
use sqlx::postgres::PgConnectOptions;

const CUSTOM_SCHEMA: &str = "acme_reliar";

async fn non_default_schema_end_to_end() {
    let base = common::fresh_unmigrated_db().await;

    migrate(&base, MigrateOptions::default().schema(CUSTOM_SCHEMA))
        .await
        .expect("migrate into a non-default schema");

    let outbox_in_custom_schema: bool = sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM information_schema.tables \
           WHERE table_schema = $1 AND table_name = 'outbox')",
    )
    .bind(CUSTOM_SCHEMA)
    .fetch_one(&base)
    .await
    .unwrap();
    assert!(outbox_in_custom_schema);

    let migrations_in_custom_schema: bool = sqlx::query_scalar(
        "SELECT EXISTS (SELECT 1 FROM information_schema.tables \
           WHERE table_schema = $1 AND table_name = '_migrations')",
    )
    .bind(CUSTOM_SCHEMA)
    .fetch_one(&base)
    .await
    .unwrap();
    assert!(migrations_in_custom_schema);

    let anything_in_public: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM information_schema.tables \
           WHERE table_schema = 'public' AND table_name IN ('outbox', '_migrations')",
    )
    .fetch_one(&base)
    .await
    .unwrap();
    assert_eq!(
        anything_in_public, 0,
        "nothing Reliar owns may land in public"
    );

    // A pool whose connection carries the custom schema first on search_path runs the full
    // path with no `set_config` statement at all — the store never sets `search_path`.
    let base_options: PgConnectOptions = base.connect_options().as_ref().clone();
    let scoped_pool = PgPool::connect_with(
        base_options
            .clone()
            .options([("search_path", &format!("{CUSTOM_SCHEMA},public"))]),
    )
    .await
    .unwrap();

    let store =
        PostgresOutboxStore::with_settings(scoped_pool.clone(), PostgresOutboxSettings::default());

    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    let envelope_id = envelope.id;
    let mut tx = scoped_pool.begin().await.unwrap();
    store.enqueue_envelope(&mut tx, envelope).await.unwrap();
    tx.commit().await.unwrap();

    let batch = store
        .acquire(AcquireRequest::new(WorkerId::generate()))
        .await
        .unwrap();
    assert_eq!(batch.records.len(), 1);
    assert_eq!(batch.records[0].envelope.id, envelope_id);

    // A pool without the custom schema on its search_path constructs successfully — a store
    // constructor performs no I/O (ADR 0047) — and fails at its **first** call instead.
    let unscoped_pool = PgPool::connect_with(base_options).await.unwrap();
    let unscoped_store =
        PostgresOutboxStore::with_settings(unscoped_pool, PostgresOutboxSettings::default());
    let err = unscoped_store
        .acquire(AcquireRequest::new(WorkerId::generate()))
        .await
        .unwrap_err();
    assert!(
        matches!(err, PostgresOutboxError::NotMigrated { .. }),
        "expected NotMigrated when the configured schema is not on search_path, got {err:?}"
    );
}

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