reliar-store-postgres 0.6.0

#![allow(dead_code)]
//! Shared test fixtures for the single-binary Postgres harness (RELIAR-27): a minimal `Message`
//! body, two `Serializer` fixtures, and the real-Postgres harness itself (§8.2).
//!
//! **The one thing that changed for RELIAR-27**: the shared `ContainerAsync` is never stored in
//! a `static`. A `static` is never dropped at process exit, so `Drop` — the *only* container
//! removal path in `testcontainers` 0.27 (there is no Ryuk/reaper) — never ran; combined with
//! one container-starting binary per scenario file, that leaked one Postgres container (with
//! its volumes) per file, per run. [`start_shared_container`] instead returns the container to
//! its caller (`tests/postgres/main.rs`), which keeps it as a **local** for the process's whole
//! lifetime and drops it explicitly before exiting — see that file's module docs.

use std::sync::LazyLock;

use reliar_core::{ContentType, Message, Serializer};
use reliar_outbox::OutboxEnqueue;
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use sqlx::postgres::PgConnectOptions;
use testcontainers::ContainerAsync;
use testcontainers::ImageExt;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::postgres::Postgres;
use tokio::sync::OnceCell;

/// A minimal message body used across scenario files.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct OrderCreated {
    pub order_id: u64,
}

impl Message for OrderCreated {
    const TYPE: &'static str = "orders.created";
    const VERSION: u16 = 1;
}

/// A second `Serializer` with a **non-JSON** `ContentType`, so §43.A.4's round-trip equality
/// (`acquired.content_type == store.content_type()`) is proven for a store that is not `JSON`,
/// not just by coincidence with the default. Encodes as JSON under the hood — only the declared
/// `ContentType` differs — so the test body stays a plain `Message`.
#[derive(Clone, Debug, Default)]
pub(crate) struct TestVndSerializer;

#[derive(Debug)]
pub(crate) struct TestVndSerializerError(String);

impl std::fmt::Display for TestVndSerializerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "test serializer error: {}", self.0)
    }
}

impl std::error::Error for TestVndSerializerError {}

impl Serializer for TestVndSerializer {
    type Error = TestVndSerializerError;

    fn content_type(&self) -> &ContentType {
        static CONTENT_TYPE: LazyLock<ContentType> =
            LazyLock::new(|| ContentType::parse("application/vnd.reliar-test+json").unwrap());

        &CONTENT_TYPE
    }

    fn serialize<T: Message>(&self, body: &T) -> Result<bytes::Bytes, Self::Error> {
        serde_json::to_vec(body)
            .map(bytes::Bytes::from)
            .map_err(|err| TestVndSerializerError(err.to_string()))
    }

    fn deserialize<T: Message>(&self, bytes: &[u8]) -> Result<T, Self::Error> {
        serde_json::from_slice(bytes).map_err(|err| TestVndSerializerError(err.to_string()))
    }
}

/// A `Serializer` whose `serialize` always fails — proves `enqueue`'s
/// `EnqueueError::Serialize` path never reaches the database at all (no partial `INSERT`, no
/// SQL round trip for a body the serializer itself rejected).
#[derive(Clone, Debug, Default)]
pub(crate) struct AlwaysFailingSerializer;

#[derive(Debug)]
pub(crate) struct AlwaysFailingSerializerError;

impl std::fmt::Display for AlwaysFailingSerializerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "this serializer always fails, by design")
    }
}

impl std::error::Error for AlwaysFailingSerializerError {}

impl Serializer for AlwaysFailingSerializer {
    type Error = AlwaysFailingSerializerError;

    fn content_type(&self) -> &ContentType {
        &ContentType::JSON
    }

    fn serialize<T: Message>(&self, _body: &T) -> Result<bytes::Bytes, Self::Error> {
        Err(AlwaysFailingSerializerError)
    }

    fn deserialize<T: Message>(&self, _bytes: &[u8]) -> Result<T, Self::Error> {
        Err(AlwaysFailingSerializerError)
    }
}

static ADMIN_URL: OnceCell<String> = OnceCell::const_new();

/// Starts the shared Postgres container this whole binary's scenarios use (unless `DATABASE_URL`
/// is set, e.g. CI's service container), and records its admin connection URL for every
/// [`fresh_db`]/[`fresh_unmigrated_db`] call. Returns the container itself — **the caller
/// (`main`) owns it as a local for the rest of the process's life and must drop it explicitly
/// before exiting** (RELIAR-27); this function never stashes it anywhere longer-lived than that.
///
/// Must run to completion exactly once, before any trial touches Postgres.
pub(crate) async fn start_shared_container() -> Option<ContainerAsync<Postgres>> {
    if let Ok(url) = std::env::var("DATABASE_URL") {
        ADMIN_URL
            .set(url)
            .expect("start_shared_container must run exactly once");

        return None;
    }

    // `reliar-` name prefix + `reliar.test=true` label (RELIAR-27): the
    // built-in `org.testcontainers.managed-by=testcontainers` label alone is too broad for
    // the manual sweep in `CONTRIBUTING.md` to key on — it would also match a different project's
    // testcontainers-managed containers on the same Docker host. Both together are what let the
    // sweep (and a human skimming `docker ps`) tell "this crate's leftovers" apart from
    // anything else testcontainers-rs is managing.
    let container = Postgres::default()
        .with_tag("18-alpine")
        .with_container_name(format!("reliar-pg-{}", uuid::Uuid::now_v7().simple()))
        .with_label("reliar.test", "true")
        .start()
        .await
        .expect("start postgres container");
    let port = container
        .get_host_port_ipv4(5432)
        .await
        .expect("mapped port");
    let url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
    ADMIN_URL
        .set(url)
        .expect("start_shared_container must run exactly once");

    Some(container)
}

fn admin_url() -> &'static str {
    ADMIN_URL
        .get()
        .expect("start_shared_container must run before any scenario touches Postgres")
}

/// Creates a fresh, empty, uniquely named database on the admin connection and returns
/// [`PgConnectOptions`] pointing at it (no `search_path` set yet).
async fn create_fresh_database() -> PgConnectOptions {
    let admin = PgPool::connect(admin_url())
        .await
        .expect("connect to admin database");
    let name = format!("t_{}", uuid::Uuid::now_v7().simple());

    // `CREATE DATABASE` cannot take a bind parameter; `name` is a freshly generated UUIDv7, never
    // user input, so this is the one sanctioned exception to "macros only" (test code, not the
    // crate) — asserted safe explicitly via `AssertSqlSafe` (sqlx 0.9's SQL-injection audit gate).
    sqlx::query(sqlx::AssertSqlSafe(format!(r#"CREATE DATABASE "{name}""#)))
        .execute(&admin)
        .await
        .expect("create test database");

    let options: PgConnectOptions = admin_url()
        .parse()
        .expect("admin url parses as PgConnectOptions");

    options.database(&name)
}

/// A fresh, empty database — **not yet migrated**. For tests that exercise
/// construction/migration failure paths before `migrate()` has run.
pub(crate) async fn fresh_unmigrated_db() -> PgPool {
    PgPool::connect_with(create_fresh_database().await)
        .await
        .expect("connect to fresh database")
}

/// Migrated once per process; every [`fresh_db`] call clones it via `CREATE DATABASE …
/// TEMPLATE …` instead of re-running `migrate()` — a storage-level file copy is materially
/// faster than replaying DDL for every test (§8.2, RELIAR-16).
static TEMPLATE_NAME: OnceCell<String> = OnceCell::const_new();

async fn template_name() -> &'static str {
    TEMPLATE_NAME
        .get_or_init(|| async {
            let options = create_fresh_database().await;
            let name = options.get_database().unwrap().to_owned();
            let pool = PgPool::connect_with(options)
                .await
                .expect("connect to template database");

            reliar_store_postgres::migrate(&pool, reliar_store_postgres::MigrateOptions::default())
                .await
                .expect("migrate the template database");
            // `CREATE DATABASE … TEMPLATE` refuses a source with open connections; closing the
            // pool here is what makes every later clone safe.
            pool.close().await;

            name
        })
        .await
}

/// A fresh database, cloned from the migrated [`template_name`], with `search_path` set on the
/// returned pool itself (§24) so the store's startup verification and every query resolve
/// `outbox` in the default `reliar` schema with **no** URL/role configuration — the equivalent
/// of a host putting `reliar` first on its connection URL.
pub(crate) async fn fresh_db() -> PgPool {
    let admin = PgPool::connect(admin_url())
        .await
        .expect("connect to admin database");
    let name = format!("t_{}", uuid::Uuid::now_v7().simple());
    let template = template_name().await;

    // Same sanctioned `AssertSqlSafe` exception as `create_fresh_database`: both names are
    // freshly generated UUIDv7s, never user input.
    sqlx::query(sqlx::AssertSqlSafe(format!(
        r#"CREATE DATABASE "{name}" TEMPLATE "{template}""#
    )))
    .execute(&admin)
    .await
    .expect("clone the migrated template database");

    let options: PgConnectOptions = admin_url()
        .parse()
        .expect("admin url parses as PgConnectOptions");

    PgPool::connect_with(
        options
            .database(&name)
            .options([("search_path", "reliar,public")]),
    )
    .await
    .expect("connect with search_path set")
}

/// Enqueues `n` plain [`OrderCreated`] envelopes and returns their [`reliar_core::MessageId`]s
/// in enqueue order.
pub(crate) async fn seed<Ser: reliar_core::Serializer + Send + Sync + 'static>(
    store: &reliar_store_postgres::PostgresOutboxStore<Ser>,
    pool: &PgPool,
    n: u64,
) -> Vec<reliar_core::Envelope<OrderCreated>> {
    let mut envelopes = Vec::with_capacity(n as usize);

    for i in 0..n {
        let envelope = reliar_core::Envelope::builder(OrderCreated { order_id: i }).build();
        let mut tx = pool.begin().await.unwrap();

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

        envelopes.push(envelope);
    }

    envelopes
}

/// Moves `id`'s lease into the past — SQL time-travel for lease-expiry tests, never a
/// wall-clock sleep (§8.2). Moves **both** `locked_until` and `available_at` (ADR 0040 §1: a
/// claim now writes them to the same instant, so simulating "the lease has lapsed" means moving
/// both — leaving `available_at` at its claimed value would leave the row invisible to the
/// claim's own `available_at <= now()` boundary even though `locked_until` has passed).
pub(crate) async fn expire_lease(pool: &PgPool, id: uuid::Uuid) {
    sqlx::query(
        "UPDATE outbox SET locked_until = now() - interval '1 second', \
                            available_at = now() - interval '1 second' WHERE id = $1",
    )
    .bind(id)
    .execute(pool)
    .await
    .unwrap();
}

/// Moves `id`'s `available_at` into the past — makes a retry-delayed row due without waiting.
pub(crate) async fn make_available_now(pool: &PgPool, id: uuid::Uuid) {
    sqlx::query("UPDATE outbox SET available_at = now() - interval '1 second' WHERE id = $1")
        .bind(id)
        .execute(pool)
        .await
        .unwrap();
}

/// Installs `subscriber` as the thread-local default and immediately rebuilds `tracing`'s
/// process-global per-callsite interest cache (RELIAR-66).
///
/// **The hazard this exists for.** `tracing::subscriber::set_default` only changes the default
/// for the calling OS thread, but a callsite's computed `Interest` (whether *any* currently
/// registered subscriber cares about it) is cached **process-wide**, not per thread. Four
/// scenarios in this binary (`outbox_enqueue_spans`'s two trials, one trial each in
/// `outbox_schema_verification` and `outbox_poison_sweep_failure`, plus
/// `recorder_spawn_capture`) each install their own thread-local recording subscriber this way
/// to capture spans/events for assertion. If a *different* trial, running concurrently on
/// another OS thread with no subscriber installed, is the first to hit one of these callsites,
/// `tracing` can cache its interest as `Interest::never` before this subscriber ever gets a
/// chance to register — this subscriber is then the active default for its own thread but never
/// sees the span/event at all (observed as an empty transcript roughly 1 run in 12 before this
/// was understood). Rebuilding right after `set_default` re-evaluates every registered
/// callsite's interest against the subscriber that is *now* the default and closes that specific
/// window, but it does not prove no other thread can still race a callsite this subscriber has
/// not yet rebuilt interest for — mid-test, another trial's callsite hit is still a race with
/// whatever this subscriber has or hasn't seen so far.
///
/// **The actual, deterministic guarantee** is not this rebuild — it is that `tests/postgres/main.rs`
/// runs every recorder trial in a second, `--test-threads 1` phase that only starts after the
/// first (ordinary) phase's `libtest_mimic::run` call has returned. That call runs its trials
/// inside `std::thread::scope`, which blocks until every worker thread it spawned has been
/// joined — so by construction, no other **trial's own** thread is still alive for the entire
/// duration of the recorder phase (the shared multi-thread `rt` and its sqlx-pool background
/// tasks from phase one *are* still running by then, but every callsite they touch was already
/// registered with interest during phase one, before this phase's subscriber ever took over, so
/// they have nothing left to race). Recorder trials also cannot race *each other* there, since
/// `--test-threads 1` runs them one at a time on the phase's own thread. This function's rebuild
/// call is kept anyway as cheap defense in depth (in case a
/// recorder trial is ever run in isolation via a test filter, without the surrounding phase
/// structure), not as the mechanism the acceptance tests actually rely on. The guard is
/// returned so the default reverts on drop, exactly like a plain `set_default`.
pub(crate) fn install_recording_subscriber<S>(subscriber: S) -> tracing::subscriber::DefaultGuard
where
    S: tracing::Subscriber + Send + Sync + 'static,
{
    let guard = tracing::subscriber::set_default(subscriber);

    tracing::callsite::rebuild_interest_cache();

    guard
}

/// Runs `fut` to completion on a **dedicated, fresh current-thread Tokio runtime** with
/// `subscriber` installed as the `tracing` default for the entire call — including for any task
/// `fut` itself hands to `tokio::spawn`.
///
/// **The hazard this exists for.** [`install_recording_subscriber`] (and a plain
/// `tracing::subscriber::set_default`) only sets the default for the calling OS thread. A
/// `#[tokio::test]`-style trial that runs on this crate's *shared, multi-thread* runtime
/// (`main.rs`'s `rt`) and then `tokio::spawn`s a task hands that task to one of the runtime's
/// worker threads — which never had the recorder installed, so a span or event the spawned task
/// emits is invisible to the recorder even though the spawning code ran under it a moment
/// earlier. No store method spawns today (RELIAR-66), so nothing in this crate's test suite has
/// hit this yet, but a recorder test written against one that does *would* silently see an
/// incomplete transcript rather than fail loudly — exactly the class of bug this card exists to
/// prevent.
///
/// Building a **current-thread** runtime here (never the shared multi-thread `rt`) is what fixes
/// it: a current-thread runtime multiplexes every task it drives — the root future and anything
/// it spawns — onto the single OS thread that calls `block_on`, so a thread-local default set
/// around that call stays in scope for all of them. See `recorder_spawn_capture.rs` for a test
/// that proves a spawned task's event is captured through this helper.
pub(crate) fn with_recorder_dispatch<S, Fut>(subscriber: S, fut: Fut) -> Fut::Output
where
    S: tracing::Subscriber + Send + Sync + 'static,
    Fut: std::future::Future,
{
    let dispatch = tracing::Dispatch::new(subscriber);

    tracing::dispatcher::with_default(&dispatch, || {
        tracing::callsite::rebuild_interest_cache();
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("build a dedicated current-thread runtime for the recorder");

        runtime.block_on(fut)
    })
}