reliar-store-postgres 0.7.0

//! N6/N7/N9 — business data and an outbox row written in one `sqlx` transaction are both present
//! after commit and both absent after rollback; a reused `MessageId` aborts the transaction and
//! discards an earlier business write in it; the future is `Send` through `tokio::spawn` against a
//! real `Transaction<'_, Postgres>` (ADR 0036 amendment B). Replaces
//! `outbox_enqueue_atomic.rs` and `outbox_publisher_enqueue.rs` — after RELIAR-61 there is no
//! facade type and `OutboxEnqueue` has one method, so the two files' scenarios collapsed into one.

use crate::common;

use crate::common::OrderCreated;
use crate::common::transport::StubTransport;
use reliar_core::Serializer as _;
use reliar_core::{Classify, Envelope, FailureKind};
use reliar_outbox::OutboxEnqueue;
use reliar_store_postgres::PostgresOutboxStore;

async fn count_business_rows(pool: &sqlx::PgPool) -> i64 {
    sqlx::query_scalar("SELECT count(*) FROM widgets")
        .fetch_one(pool)
        .await
        .unwrap()
}

async fn count_outbox_rows(pool: &sqlx::PgPool) -> i64 {
    sqlx::query_scalar("SELECT count(*) FROM outbox")
        .fetch_one(pool)
        .await
        .unwrap()
}

async fn create_widgets_table(pool: &sqlx::PgPool) {
    sqlx::query("CREATE TABLE widgets (id bigserial PRIMARY KEY)")
        .execute(pool)
        .await
        .unwrap();
}

/// N6 — a business row and an outbox row written in one transaction are both visible after
/// commit, and the committed row carries the store's serializer's `content_type` and its
/// serialized bytes.
async fn commit_makes_both_rows_visible() {
    let pool = common::fresh_db().await;

    create_widgets_table(&pool).await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();

    let mut tx = pool.begin().await.unwrap();
    sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await
        .unwrap();
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    let envelope_id = envelope.id;
    let expected_payload = reliar_core::JsonSerializer
        .serialize(&envelope.body)
        .unwrap();
    store.enqueue_envelope(&mut tx, envelope).await.unwrap();
    tx.commit().await.unwrap();

    assert_eq!(count_business_rows(&pool).await, 1);
    assert_eq!(count_outbox_rows(&pool).await, 1);

    let content_type: String =
        sqlx::query_scalar("SELECT content_type FROM outbox WHERE message_id = $1")
            .bind(envelope_id.as_uuid())
            .fetch_one(&pool)
            .await
            .unwrap();
    let payload: Vec<u8> = sqlx::query_scalar("SELECT payload FROM outbox WHERE message_id = $1")
        .bind(envelope_id.as_uuid())
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(content_type, store.content_type().to_string());
    assert_eq!(payload, expected_payload.as_ref());
}

/// N6 — rollback leaves neither row visible.
async fn rollback_makes_neither_row_visible() {
    let pool = common::fresh_db().await;

    create_widgets_table(&pool).await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();

    let mut tx = pool.begin().await.unwrap();
    sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await
        .unwrap();
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    store.enqueue_envelope(&mut tx, envelope).await.unwrap();
    tx.rollback().await.unwrap();

    assert_eq!(count_business_rows(&pool).await, 0);
    assert_eq!(count_outbox_rows(&pool).await, 0);
}

/// P-7 (ADR 0044 §7) — after `enqueue_envelope`, the row's `message_id` equals the envelope's id,
/// its `id` differs from it, is a v7 UUID, and is not null.
async fn enqueue_writes_both_ids() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone()).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).await.unwrap();
    tx.commit().await.unwrap();

    let (id, message_id): (uuid::Uuid, uuid::Uuid) =
        sqlx::query_as("SELECT id, message_id FROM outbox WHERE message_id = $1")
            .bind(envelope_id.as_uuid())
            .fetch_one(&pool)
            .await
            .unwrap();

    assert_eq!(
        message_id,
        envelope_id.as_uuid(),
        "message_id must equal the envelope's own id"
    );
    assert_ne!(id, message_id, "id must differ from message_id");
    assert_eq!(
        id.get_version_num(),
        7,
        "id must be a database-assigned v7 UUID"
    );
}

/// N7 — a reused `MessageId` maps to [`reliar_store_postgres::EnqueueError::Duplicate`], leaves
/// the transaction aborted (the next statement on it fails too), and an earlier business write
/// made in that same transaction is gone once the caller gives up on it — "an earlier `Ok` is not
/// durable"; "treat any enqueue error as *abort this transaction*".
async fn duplicate_message_id_aborts_the_transaction_and_discards_the_earlier_business_write() {
    let pool = common::fresh_db().await;

    create_widgets_table(&pool).await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();
    let envelope = Envelope::builder(OrderCreated { order_id: 1 }).build();
    let envelope_id = envelope.id;

    // Committed ahead of time, in its own transaction — its id is what the second `enqueue` below
    // reuses, forcing that `INSERT` (and therefore the whole transaction) to abort. The envelope
    // is cloned explicitly for the reuse — by-value `enqueue` never clones on the caller's behalf.
    let mut seed_tx = pool.begin().await.unwrap();
    store
        .enqueue_envelope(&mut seed_tx, envelope.clone())
        .await
        .unwrap();
    seed_tx.commit().await.unwrap();

    let mut tx = pool.begin().await.unwrap();
    sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await
        .unwrap();

    let result = store.enqueue_envelope(&mut tx, envelope).await;

    match result.expect_err("a reused MessageId must be rejected") {
        reliar_store_postgres::EnqueueError::Duplicate { id } => {
            assert_eq!(id, envelope_id);
        }
        other => panic!("expected EnqueueError::Duplicate, got {other:?}"),
    }

    // The transaction is already aborted server-side: the next statement on it fails too.
    let next_statement = sqlx::query("INSERT INTO widgets DEFAULT VALUES")
        .execute(&mut *tx)
        .await;
    assert!(
        next_statement.is_err(),
        "a statement on an aborted transaction must fail"
    );

    tx.rollback().await.unwrap();

    // The earlier business write never survives — the whole transaction was all-or-nothing.
    assert_eq!(
        count_business_rows(&pool).await,
        0,
        "an earlier write in the aborted transaction must not be durable"
    );
}

/// N1 — `enqueue_envelope` writes exactly one row and never calls a transport `Publisher`: the
/// trait signature itself carries no `Publisher` parameter, so a stub publisher handed to nothing
/// but held alive alongside the store stays untouched — a mutation guard, not just a type
/// argument.
async fn enqueue_never_calls_a_publisher() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();
    let publisher = StubTransport::ok();

    let envelope = Envelope::builder(OrderCreated { order_id: 9 }).build();
    let mut tx = pool.begin().await.unwrap();

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

    assert!(
        publisher.published().is_empty(),
        "enqueue must never call Publisher::publish"
    );
}

/// N2/N8 — a reused `MessageId`'s `EnqueueError::Duplicate` is transparent and classified
/// (`Classify::kind`), and its `Display` never mentions the payload or a header value even when
/// the offending envelope carries them.
async fn a_duplicate_error_is_classified_and_never_leaks_the_payload_or_a_header_value() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();

    let envelope = Envelope::builder(OrderCreated { order_id: 1 })
        .header("x-secret", "SUPER_SECRET_HEADER_VALUE")
        .unwrap()
        .build();

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

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

    let mut tx = pool.begin().await.unwrap();
    let err = store
        .enqueue_envelope(&mut tx, envelope)
        .await
        .expect_err("a reused MessageId must be rejected");
    tx.rollback().await.unwrap();

    assert_eq!(
        err.kind(),
        FailureKind::Permanent,
        "a reused id is never worth retrying as-is"
    );

    let text = err.to_string();
    assert!(
        !text.contains("order_id"),
        "must never mention the payload: {text}"
    );
    assert!(
        !text.contains("SUPER_SECRET_HEADER_VALUE"),
        "must never mention a header value: {text}"
    );
}

/// N9 — the R23 regression guard: the future `enqueue` returns is `Send` even though it borrows a
/// non-`'static` `Transaction<'_, Postgres>` scope — `tokio::spawn` requires exactly that, and
/// this is the one place a change to `T: Message + Sync` on the typed method (ADR 0036 amendment B) would
/// show up as a compile failure, not just a runtime one.
async fn enqueue_is_send_through_tokio_spawn() {
    let pool = common::fresh_db().await;
    let store = PostgresOutboxStore::new(pool.clone()).await.unwrap();

    let envelope = Envelope::builder(OrderCreated { order_id: 7 }).build();
    let envelope_id = envelope.id;
    let mut tx = pool.begin().await.unwrap();

    tokio::spawn(async move {
        store.enqueue_envelope(&mut tx, envelope).await.unwrap();

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

    let count: i64 = sqlx::query_scalar("SELECT count(*) FROM outbox WHERE message_id = $1")
        .bind(envelope_id.as_uuid())
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(count, 1);
}

pub(crate) fn trials(rt: &'static tokio::runtime::Runtime) -> Vec<libtest_mimic::Trial> {
    vec![
        libtest_mimic::Trial::test(
            "outbox_enqueue::commit_makes_both_rows_visible",
            move || {
                rt.block_on(commit_makes_both_rows_visible());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue::rollback_makes_neither_row_visible",
            move || {
                rt.block_on(rollback_makes_neither_row_visible());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test("outbox_enqueue::enqueue_writes_both_ids", move || {
            rt.block_on(enqueue_writes_both_ids());
            Ok(())
        }),
        libtest_mimic::Trial::test(
            "outbox_enqueue::duplicate_message_id_aborts_the_transaction_and_discards_the_earlier_business_write",
            move || {
                rt.block_on(
                    duplicate_message_id_aborts_the_transaction_and_discards_the_earlier_business_write(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue::enqueue_never_calls_a_publisher",
            move || {
                rt.block_on(enqueue_never_calls_a_publisher());
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue::a_duplicate_error_is_classified_and_never_leaks_the_payload_or_a_header_value",
            move || {
                rt.block_on(
                    a_duplicate_error_is_classified_and_never_leaks_the_payload_or_a_header_value(),
                );
                Ok(())
            },
        ),
        libtest_mimic::Trial::test(
            "outbox_enqueue::enqueue_is_send_through_tokio_spawn",
            move || {
                rt.block_on(enqueue_is_send_through_tokio_spawn());
                Ok(())
            },
        ),
    ]
}