reliar-store-postgres 0.9.0

#![allow(dead_code)]
//! Shared fixtures for the inbox scenario files (inbox contract §5.2): a throwaway "business"
//! table a handler writes into (proving same-transaction atomicity with the inbox row), and
//! several [`InboxHandler`] implementations — one that always succeeds, one that writes then
//! fails, one that completes its own row mid-`handle`, and a no-op.

use std::sync::OnceLock;

use reliar_core::{MessageId, MessageType};
use reliar_inbox::{InboxHandler, InboxMessage, InboxScope, InboxStore};
use reliar_store_postgres::PostgresInboxStore;
use sqlx::{Postgres, Transaction};

/// A shared `orders.created.v1` message type — the trace-field view every scenario builds its
/// [`InboxMessage`] from when the test does not care what the type actually is.
pub(crate) fn message_type() -> &'static MessageType {
    static MESSAGE_TYPE: OnceLock<MessageType> = OnceLock::new();

    MESSAGE_TYPE.get_or_init(|| MessageType::new("orders.created", 1))
}

/// The minimal [`InboxMessage`] view `claim`/`fail`/`process` take, built fresh from `id` each
/// call (mirroring `InboxMessage::from_envelope` at a real call site).
pub(crate) fn message(id: MessageId) -> InboxMessage<'static> {
    InboxMessage::new(id, message_type())
}

/// Creates the per-test "business" table a handler writes into, so a test can assert the
/// business write and the inbox row commit or roll back **together** — the whole of the inbox's
/// guarantee (inbox contract §1). Each test's database is already isolated (`common::fresh_db`
/// clones a fresh copy per test), so `CREATE TABLE` needs no `IF NOT EXISTS`/cleanup.
///
/// Runtime string API, not `query!`: `business_events` does not exist when `cargo sqlx prepare`
/// introspects the schema (it is created by this very function, at test run time), so a
/// compile-time macro has nothing to check against — the same reason `outbox_enqueue.rs`'s
/// `widgets` fixture stays unmacroed.
pub(crate) async fn create_business_table(pool: &sqlx::PgPool) {
    sqlx::query("CREATE TABLE business_events (id bigserial PRIMARY KEY, value bigint NOT NULL)")
        .execute(pool)
        .await
        .unwrap();
}

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

/// A handler that inserts one row into `business_events` and succeeds.
pub(crate) struct InsertBusinessRow {
    pub(crate) value: i64,
}

impl InboxHandler<Transaction<'_, Postgres>> for InsertBusinessRow {
    type Output = i64;

    type Error = sqlx::Error;

    async fn handle(&self, tx: &mut Transaction<'_, Postgres>) -> Result<i64, sqlx::Error> {
        sqlx::query("INSERT INTO business_events (value) VALUES ($1)")
            .bind(self.value)
            .execute(&mut **tx)
            .await?;

        Ok(self.value)
    }
}

/// A handler that writes a business row and then fails — proves a handler failure rolls the
/// business write back along with the claim (inbox contract §2.4).
pub(crate) struct InsertThenFail {
    pub(crate) value: i64,
}

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

impl std::fmt::Display for HandlerFailed {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "handler failed, by design")
    }
}

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

impl InboxHandler<Transaction<'_, Postgres>> for InsertThenFail {
    type Output = ();

    type Error = HandlerFailed;

    async fn handle(&self, tx: &mut Transaction<'_, Postgres>) -> Result<(), HandlerFailed> {
        sqlx::query("INSERT INTO business_events (value) VALUES ($1)")
            .bind(self.value)
            .execute(&mut **tx)
            .await
            .unwrap();

        Err(HandlerFailed)
    }
}

/// A handler that completes its own row **during** `handle`, on the same `tx` — so `process`'s
/// own `complete` call afterward finds the row already completed and fails with the store's
/// "no claimed row" error, forcing `process` to return `Err(InboxProcessError::Store(_))`
/// deterministically (inbox contract I-N6), without needing real concurrency. Mirrors
/// `reliar-inbox`'s own (removed) fake-backed `SelfCompletingHandler` fixture, ADR 0043.
pub(crate) struct SelfCompletingHandler<'a> {
    pub(crate) store: &'a PostgresInboxStore,

    pub(crate) scope: InboxScope,

    pub(crate) id: MessageId,
}

impl InboxHandler<Transaction<'_, Postgres>> for SelfCompletingHandler<'_> {
    type Output = ();

    type Error = std::convert::Infallible;

    #[allow(
        clippy::expect_used,
        reason = "a fixture helper's own precondition, not a #[test] body"
    )]
    async fn handle(
        &self,
        tx: &mut Transaction<'_, Postgres>,
    ) -> Result<(), std::convert::Infallible> {
        self.store
            .complete(tx, &self.scope, self.id)
            .await
            .expect("the row is freshly claimed, so this store-side complete must succeed");

        Ok(())
    }
}

/// A handler that never runs any I/O and always succeeds — for scenarios that only need to
/// observe the claim/complete path, not a business write.
pub(crate) struct NoopHandler;

impl InboxHandler<Transaction<'_, Postgres>> for NoopHandler {
    type Output = ();

    type Error = std::convert::Infallible;

    fn handle(
        &self,
        _tx: &mut Transaction<'_, Postgres>,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        std::future::ready(Ok(()))
    }
}