reliar-store-postgres 0.7.0

//! §43.A.36 — `PostgresOutboxStore::new` resolves `outbox` against the configured schema
//! exactly once at construction: succeeds when they match, fails naming the configured schema
//! and the observed `search_path` when unresolvable, and warns (asserted with a recording
//! `tracing` subscriber) when a same-named table also exists elsewhere on the path. With
//! `enqueue_sets_search_path = true`, `enqueue` sets and restores the path inside the caller's
//! transaction and leaves it unchanged afterward.

use crate::common;

use std::sync::{Arc, Mutex};

use crate::common::OrderCreated;
use reliar_core::Envelope;
use reliar_outbox::OutboxEnqueue;
use reliar_store_postgres::{
    MigrateOptions, PostgresOutboxError, PostgresOutboxSettings, PostgresOutboxStore, migrate,
};
use sqlx::PgPool;
use sqlx::postgres::PgConnectOptions;
use tracing::field::{Field, Visit};
use tracing_subscriber::layer::{Context, SubscriberExt};

#[derive(Default, Clone)]
struct Recorded(Arc<Mutex<Vec<String>>>);

struct Recorder(Recorded);

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Recorder {
    fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
        struct MessageVisitor(String);
        impl Visit for MessageVisitor {
            fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
                if field.name() == "message" {
                    self.0 = format!("{value:?}");
                }
            }
        }
        let mut visitor = MessageVisitor(String::new());
        event.record(&mut visitor);

        self.0.0.lock().unwrap().push(visitor.0);
    }
}

/// Async-friendly: `set_default` returns a guard that stays active across `.await` points on
/// the *current* OS thread, which is exactly what a default (single-threaded) `#[tokio::test]`
/// runs on — unlike `tracing::subscriber::with_default`, this needs no nested runtime. Goes
/// through [`common::install_recording_subscriber`] to rebuild the process-wide
/// callsite interest cache right after installing, so a concurrent trial cannot leave this
/// recorder's callsites cached as `never` before it gets a chance to see them.
async fn with_recording_subscriber<Fut: std::future::Future<Output = ()>>(f: Fut) -> Vec<String> {
    let recorded = Recorded::default();
    let subscriber = tracing_subscriber::registry().with(Recorder(recorded.clone()));
    let _guard = common::install_recording_subscriber(subscriber);

    f.await;

    recorded.0.lock().unwrap().clone()
}

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();
    // A pool whose `search_path` explicitly excludes `reliar` — deliberately set to just
    // `public` rather than left at the server default, since 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 construction_fails_fast_without_search_path() {
    let pool = pool_without_search_path().await;
    let err = PostgresOutboxStore::new(pool).await.unwrap_err();

    match err {
        PostgresOutboxError::SchemaNotOnSearchPath { configured, .. } => {
            assert_eq!(configured, "reliar");
        }
        other => panic!("expected SchemaNotOnSearchPath, got {other:?}"),
    }
}

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

    PostgresOutboxStore::new(pool)
        .await
        .expect("construction succeeds once outbox resolves to the configured schema");
}

/// P-13 (ADR 0044 Amendment A.4) — `connect` refuses a schema that resolves correctly (the
/// `search_path` check above passes) but predates the `message_id` split: migrated only through
/// `0004`, the same shape `reliar-store-postgres` 0.6.0 shipped.
async fn connect_refuses_a_schema_missing_the_message_id_column() {
    let pool = common::fresh_unmigrated_db().await;

    common::apply_migration_prefix(&pool, 4).await;

    let options: PgConnectOptions = pool
        .connect_options()
        .as_ref()
        .clone()
        .options([("search_path", "reliar,public")]);
    let store_pool = PgPool::connect_with(options).await.unwrap();

    let err = PostgresOutboxStore::new(store_pool).await.unwrap_err();

    match &err {
        PostgresOutboxError::SchemaOutOfDate { schema, missing } => {
            assert_eq!(schema, "reliar");
            assert_eq!(*missing, "message_id");
        }
        other => panic!("expected SchemaOutOfDate, got {other:?}"),
    }

    // Formats the error `connect` actually returned, not a reconstructed literal (review round 1
    // nit) — a `Display` regression that only showed up on the real variant would otherwise slip
    // past this trial.
    let message = err.to_string();
    assert!(
        message.contains("migrate"),
        "the message must name migrate() as the remedy: {message:?}"
    );
}

/// P-13's sibling (ADR 0044 Amendment A.5, M1 review round 1) — a schema stuck anywhere in
/// `0005`-`0009` has `message_id`, but `id` exists and is still nullable (no migration through
/// `0010` has run `SET NOT NULL` on it yet). `connect` must report `missing: "id"`, not pass
/// silently the way a bare "does `message_id` exist" check would — that gap is exactly what let
/// `acquire` fail decoding a `NULL` `id` instead of `connect` refusing cleanly. Once `0010`
/// completes, `connect` succeeds.
async fn connect_refuses_a_schema_stuck_between_0005_and_0009() {
    let pool = common::fresh_unmigrated_db().await;

    common::apply_migration_prefix(&pool, 9).await;

    let options: PgConnectOptions = pool
        .connect_options()
        .as_ref()
        .clone()
        .options([("search_path", "reliar,public")]);
    let store_pool = PgPool::connect_with(options.clone()).await.unwrap();

    let err = PostgresOutboxStore::new(store_pool).await.unwrap_err();

    match &err {
        PostgresOutboxError::SchemaOutOfDate { schema, missing } => {
            assert_eq!(schema, "reliar");
            assert_eq!(*missing, "id");
        }
        other => panic!("expected SchemaOutOfDate {{ missing: \"id\" }}, got {other:?}"),
    }

    migrate(&pool, MigrateOptions::default())
        .await
        .expect("migrate() completes 0010");

    let store_pool = PgPool::connect_with(options).await.unwrap();
    PostgresOutboxStore::new(store_pool)
        .await
        .expect("connect succeeds once 0010 has completed");
}

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

    reliar_store_postgres::migrate(&base, reliar_store_postgres::MigrateOptions::default())
        .await
        .unwrap();
    sqlx::query("CREATE TABLE public.outbox (id uuid)")
        .execute(&base)
        .await
        .unwrap();

    let options: PgConnectOptions = base
        .connect_options()
        .as_ref()
        .clone()
        .options([("search_path", "reliar,public")]);
    let pool = PgPool::connect_with(options).await.unwrap();

    let messages = with_recording_subscriber(async {
        PostgresOutboxStore::new(pool).await.unwrap();
    })
    .await;

    assert!(
        messages.iter().any(|m| m.contains("outbox")),
        "expected a warning naming the duplicate `outbox` table; got {messages:?}"
    );
}

async fn enqueue_sets_search_path_restores_the_callers_value() {
    let pool = common::fresh_db().await;
    let settings = PostgresOutboxSettings::default().enqueue_sets_search_path(true);
    let store = PostgresOutboxStore::with_settings(pool.clone(), settings)
        .await
        .unwrap();

    let mut tx = pool.begin().await.unwrap();
    let before: String = sqlx::query_scalar("SELECT current_setting('search_path')")
        .fetch_one(&mut *tx)
        .await
        .unwrap();

    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();

    store.enqueue_envelope(&mut tx, envelope).await.unwrap();

    let after: String = sqlx::query_scalar("SELECT current_setting('search_path')")
        .fetch_one(&mut *tx)
        .await
        .unwrap();
    assert_eq!(
        before, after,
        "enqueue must restore the caller's search_path"
    );

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

/// `enqueue_sets_search_path(true)`'s `set_config`/restore wrap must not interfere with the
/// ordinary `pk_outbox`-violation mapping: a duplicate id still surfaces as
/// `EnqueueError::Duplicate`, not masked by the (correctly skipped-on-failure) restore.
async fn duplicate_id_surfaces_correctly_with_enqueue_sets_search_path() {
    let pool = common::fresh_db().await;
    let settings = PostgresOutboxSettings::default().enqueue_sets_search_path(true);
    let store = PostgresOutboxStore::with_settings(pool.clone(), settings)
        .await
        .unwrap();
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    let envelope_id = envelope.id;

    let mut tx = pool.begin().await.unwrap();

    store
        .enqueue_envelope(&mut tx, envelope.clone())
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let mut tx = pool.begin().await.unwrap();

    match store.enqueue_envelope(&mut tx, envelope).await {
        Err(reliar_store_postgres::EnqueueError::Duplicate { id }) => assert_eq!(id, envelope_id),
        other => panic!("expected EnqueueError::Duplicate, got {other:?}"),
    }
}

/// Every trial in this file except `warns_when_a_same_named_table_exists_elsewhere` — that one
/// installs a thread-local recording subscriber and must run in `main.rs`'s serialised recorder
/// phase instead (see [`recorder_trials`], RELIAR-66).
pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "outbox_schema_verification::construction_fails_fast_without_search_path",
            move || {
                rt.block_on(construction_fails_fast_without_search_path());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_schema_verification::construction_succeeds_with_search_path_set",
            move || {
                rt.block_on(construction_succeeds_with_search_path_set());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_schema_verification::connect_refuses_a_schema_missing_the_message_id_column",
            move || {
                rt.block_on(connect_refuses_a_schema_missing_the_message_id_column());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_schema_verification::connect_refuses_a_schema_stuck_between_0005_and_0009",
            move || {
                rt.block_on(connect_refuses_a_schema_stuck_between_0005_and_0009());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_schema_verification::enqueue_sets_search_path_restores_the_callers_value",
            move || {
                rt.block_on(enqueue_sets_search_path_restores_the_callers_value());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_schema_verification::duplicate_id_surfaces_correctly_with_enqueue_sets_search_path",
            move || {
                rt.block_on(duplicate_id_surfaces_correctly_with_enqueue_sets_search_path());
                Ok(())
            },
        ),
    ]
}

/// `warns_when_a_same_named_table_exists_elsewhere` installs a thread-local recording subscriber
/// (`with_recording_subscriber`) — must run in `main.rs`'s serialised recorder phase, never in
/// the parallel batch (RELIAR-66; see `common::install_recording_subscriber`'s doc for why).
pub(crate) fn recorder_trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![libtest_mimic::Trial::test(
        "outbox_schema_verification::warns_when_a_same_named_table_exists_elsewhere",
        move || {
            rt.block_on(warns_when_a_same_named_table_exists_elsewhere());
            Ok(())
        },
    )]
}