reliar-store-postgres 0.7.0

//! The [`PostgresOutboxStore`] type itself: fields, construction (`connect`/`new`/
//! `with_settings`), the small shared helpers every other concern module calls back into
//! (`map_err`, `set_local_timeout`), and the `OutboxStore` trait impl — which delegates each
//! method's body to its concern module (`claim`, `outcomes`, `purge`) so this file states *what*
//! the public surface is without also carrying every query.

use std::sync::Arc;

use reliar_core::{ContentType, Serializer};
use reliar_outbox::{
    AcquireRequest, AcquiredBatch, CompletedRecord, FailedRecord, OutboxStats, OutboxStore,
    PurgeReport, PurgeRequest, RecordRef, WorkerId,
};
use sqlx::{PgPool, Postgres, Transaction};

use crate::connection::schema;
use crate::settings::PostgresOutboxSettings;

#[cfg(feature = "json")]
use reliar_core::JsonSerializer;

use super::error::{self, PostgresOutboxError};
use super::{claim, outcomes, purge};

/// Columns [`PostgresOutboxStore::connect`] requires on the resolved `outbox` relation to be
/// **present and `NOT NULL`** (ADR 0044 Amendment A.4, corrected by Amendment A.5) — this is a
/// **completion marker for migrations `0005`–`0010`**, not a column inventory: `message_id` is
/// absent before `0005`; `id` exists from `0005` onward but stays nullable until `0010`'s `SET
/// NOT NULL`, so checking `id`'s mere existence would pass in the `0005`–`0009` window where rows
/// can still have a `NULL` `id` and every `acquire` fails decoding. Order is load-bearing — the
/// first unsatisfied entry is what `connect` reports. Closed and known at compile time, hence
/// `&'static str`.
const REQUIRED_OUTBOX_COLUMNS: &[&str] = &["message_id", "id"];

/// Reliar's PostgreSQL outbox provider. Cheap to clone into an `AppState` — it wraps a
/// [`PgPool`]; no outer `Arc` required. The connection pool stays the host's: Reliar never owns
/// or reads a `DATABASE_URL`.
///
/// The default type parameter only exists behind the crate's default `json` feature: under
/// `--no-default-features` there is no default, so [`Self::connect`] is the only
/// constructor and `cargo hack --feature-powerset` compiles every combination. This block's
/// `PostgresOutboxStore::new` leans on that default, so it only compiles under `json`; without
/// it this block still shows the shape but is not compiled.
#[cfg_attr(not(feature = "json"), doc = "```ignore")]
#[cfg_attr(feature = "json", doc = "```no_run")]
/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
/// use reliar_store_postgres::{PostgresOutboxStore, migrate};
/// use sqlx::postgres::PgPoolOptions;
///
/// let pool = PgPoolOptions::new()
///     .connect(&std::env::var("DATABASE_URL")?)
///     .await?;
/// migrate(&pool, Default::default()).await?;
///
/// let store = PostgresOutboxStore::new(pool).await?;
/// // `store` now implements `OutboxEnqueue`, `OutboxStore` and `OutboxDeadLetters` —
/// // hand it to an application's write path and to an `OutboxDispatcher`.
/// # Ok(())
/// # }
/// ```
#[non_exhaustive]
pub struct PostgresOutboxStore<
    #[cfg(feature = "json")] Ser = JsonSerializer,
    #[cfg(not(feature = "json"))] Ser,
> {
    // `pub(super)`: every concern module under `outbox/` reads these directly rather than through
    // an accessor — they are `outbox`-private, never part of this crate's public surface.
    pub(super) pool: PgPool,

    pub(super) settings: PostgresOutboxSettings,

    pub(super) serializer: Arc<Ser>,
}

/// **Manual impl, never derived**: a derived `Clone` would condition on `Ser: Clone`. The
/// serializer is held as `Arc<Ser>` — stateless and cheap to share — so cloning the store never
/// requires the serializer itself to be `Clone`.
impl<Ser> Clone for PostgresOutboxStore<Ser> {
    fn clone(&self) -> Self {
        Self {
            pool: self.pool.clone(),
            settings: self.settings.clone(),
            serializer: Arc::clone(&self.serializer),
        }
    }
}

impl<Ser> std::fmt::Debug for PostgresOutboxStore<Ser> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PostgresOutboxStore")
            .field("settings", &self.settings)
            .finish_non_exhaustive()
    }
}

impl<Ser: Serializer + Send + Sync + 'static> PostgresOutboxStore<Ser> {
    /// Wraps `pool` with `settings` and `serializer`. **Verifies once at construction**, in
    /// order: that the connected server's `server_version_num` meets
    /// [`crate::MIN_SERVER_VERSION_NUM`] (ADR 0041 — a wrong server version
    /// explains a missing relation, and the reverse is never true), then that the unqualified
    /// name `outbox` resolves to `settings.schema`, then that the resolved relation has finished
    /// the row-identity split — `message_id` and `id` both present and `NOT NULL` (ADR 0044 §1,
    /// Amendment A.5 — a schema migrated only through `0004` is missing `message_id` entirely,
    /// and one stopped anywhere in `0005`–`0009` has `id` but it is still nullable): fails fast
    /// with
    /// [`PostgresOutboxError::UnsupportedServerVersion`],
    /// [`PostgresOutboxError::SchemaNotOnSearchPath`] (`search_path` problem),
    /// [`PostgresOutboxError::NotMigrated`] (the relation is missing entirely), or
    /// [`PostgresOutboxError::SchemaOutOfDate`] (the relation exists but is not yet on
    /// `0.7.0`'s schema) rather than surprising the first `acquire`. Logs a `tracing::warn!` when
    /// a same-named table also exists in another schema on the path.
    ///
    /// # Errors
    ///
    /// Returns [`PostgresOutboxError::UnsupportedServerVersion`],
    /// [`PostgresOutboxError::NotMigrated`], [`PostgresOutboxError::SchemaNotOnSearchPath`],
    /// [`PostgresOutboxError::SchemaOutOfDate`], or [`PostgresOutboxError::Database`] for a
    /// connection failure during verification.
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_core::JsonSerializer;
    /// use reliar_store_postgres::{PostgresOutboxSettings, PostgresOutboxStore};
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresOutboxStore::connect(
    ///     pool,
    ///     PostgresOutboxSettings::default(),
    ///     JsonSerializer,
    /// )
    /// .await?;
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect(
        pool: PgPool,
        settings: PostgresOutboxSettings,
        serializer: Ser,
    ) -> Result<Self, PostgresOutboxError> {
        if !schema::is_valid_schema_name(&settings.schema) {
            return Err(PostgresOutboxError::InvalidSchema {
                schema: settings.schema,
            });
        }

        let detected = crate::connection::version::detected_server_version_num(&pool).await?;

        if detected < crate::MIN_SERVER_VERSION_NUM {
            return Err(PostgresOutboxError::UnsupportedServerVersion {
                required: crate::MIN_SERVER_VERSION_NUM,
                detected,
            });
        }

        let check =
            schema::verify_table_schema(&pool, &settings.schema, "outbox", REQUIRED_OUTBOX_COLUMNS)
                .await
                .map_err(|err| error::map_operational_error(&settings.schema, err))?;

        let resolved_here = check.resolved_schema.as_deref() == Some(settings.schema.as_str());

        if !resolved_here {
            if !check.configured_exists {
                return Err(PostgresOutboxError::NotMigrated {
                    schema: settings.schema,
                });
            }

            return Err(PostgresOutboxError::SchemaNotOnSearchPath {
                configured: settings.schema,
                observed: check.search_path,
            });
        }

        // ADR 0044 Amendment A.4 (marker corrected by Amendment A.5): `outbox` exists and
        // resolves correctly, but a pre-0.7.0 schema is missing — or has not yet finished
        // migrating — a column a later migration completes; refuse `connect` here rather than
        // surprising the first `enqueue`/`acquire` with a bare `42703` or a decode failure on a
        // still-nullable `id`.
        if let Some(&missing) = REQUIRED_OUTBOX_COLUMNS
            .iter()
            .find(|col| !check.satisfied_required_columns.iter().any(|c| c == *col))
        {
            return Err(PostgresOutboxError::SchemaOutOfDate {
                schema: settings.schema,
                missing,
            });
        }

        let others = schema::other_table_schemas(&pool, &settings.schema, "outbox")
            .await
            .map_err(PostgresOutboxError::from)?;

        if !others.is_empty() {
            tracing::warn!(
                configured_schema = %settings.schema,
                other_schemas = ?others,
                "a table named `outbox` also exists outside the configured schema; \
                 an unqualified reference from another session could resolve to it"
            );
        }

        Ok(Self {
            pool,
            settings,
            serializer: Arc::new(serializer),
        })
    }

    /// The `ContentType` this store writes to every row — `Serializer::content_type()`. The
    /// only way a caller can predict the `content_type` of an envelope it will later acquire:
    /// `enqueue` writes this value, ignoring whatever `envelope.metadata.delivery.content_type`
    /// held. `PostgresOutboxStore::new` here leans on the default type parameter, gated on the
    /// default `json` feature; without it this block still shows the shape but is not compiled.
    #[cfg_attr(not(feature = "json"), doc = "```ignore")]
    #[cfg_attr(feature = "json", doc = "```no_run")]
    /// # async fn run(pool: sqlx::PgPool) -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::PostgresOutboxStore;
    ///
    /// let store = PostgresOutboxStore::new(pool).await?;
    /// assert_eq!(store.content_type().as_str(), "application/json");
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn content_type(&self) -> &ContentType {
        self.serializer.content_type()
    }

    /// Maps a `sqlx::Error` from one of this store's own operations to a typed
    /// [`PostgresOutboxError`], catching SQLSTATE `42P01` on **every** call, not just startup
    /// verification.
    pub(super) fn map_err(&self, err: sqlx::Error) -> PostgresOutboxError {
        error::map_operational_error(&self.settings.schema, err)
    }

    /// Issues `SET LOCAL statement_timeout` on an already-open transaction — the shared half of
    /// every `Duration::ZERO`-vs-non-zero split in the concern modules below.
    pub(super) async fn set_local_timeout(
        &self,
        tx: &mut Transaction<'_, Postgres>,
    ) -> Result<(), PostgresOutboxError> {
        self.set_local_timeout_raw(tx)
            .await
            .map_err(|e| self.map_err(e))
    }

    /// [`Self::set_local_timeout`] without the `PostgresOutboxError` mapping — for the one caller
    /// (`claim::acquire`'s best-effort poison sweep, ADR 0039 §4) that folds this into a larger
    /// `sqlx::Error`-returning block rather than propagating a typed error immediately.
    pub(super) async fn set_local_timeout_raw(
        &self,
        tx: &mut Transaction<'_, Postgres>,
    ) -> Result<(), sqlx::Error> {
        let timeout_ms = i64::try_from(self.settings.statement_timeout.as_millis())
            .unwrap_or(i64::MAX)
            .to_string();

        sqlx::query_scalar!(
            "SELECT set_config('statement_timeout', $1, true)",
            timeout_ms
        )
        .fetch_one(&mut **tx)
        .await?;

        Ok(())
    }
}

#[cfg(feature = "json")]
#[cfg_attr(docsrs, doc(cfg(feature = "json")))]
impl PostgresOutboxStore<JsonSerializer> {
    /// Convenience over [`Self::connect`], behind the crate's default `json` feature.
    ///
    /// # Errors
    ///
    /// Same as [`Self::connect`].
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::PostgresOutboxStore;
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresOutboxStore::new(pool).await?;
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn new(pool: PgPool) -> Result<Self, PostgresOutboxError> {
        Self::connect(pool, PostgresOutboxSettings::default(), JsonSerializer).await
    }

    /// Convenience over [`Self::connect`] with explicit settings, behind the crate's default
    /// `json` feature.
    ///
    /// # Errors
    ///
    /// Same as [`Self::connect`].
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::{PostgresOutboxSettings, PostgresOutboxStore};
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresOutboxStore::with_settings(
    ///     pool,
    ///     PostgresOutboxSettings::default().schema("orders"),
    /// )
    /// .await?;
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn with_settings(
        pool: PgPool,
        settings: PostgresOutboxSettings,
    ) -> Result<Self, PostgresOutboxError> {
        Self::connect(pool, settings, JsonSerializer).await
    }
}

impl<Ser: Serializer + Send + Sync + 'static> OutboxStore for PostgresOutboxStore<Ser> {
    type Error = PostgresOutboxError;

    /// The canonical single-statement claim (ADR 0006): a CTE
    /// `SELECT … FOR UPDATE SKIP LOCKED` feeding an `UPDATE … RETURNING`, so the row lock is
    /// released before this future resolves and no network I/O to a publisher can ever happen
    /// while it is held.
    ///
    /// A row this call cannot decode is **excluded from `records`**, reported in `poisoned`,
    /// and **moved to dead** with `DeadReason::Undecodable` by a follow-up statement guarded by
    /// `locked_by` — the batch continues rather than failing outright (ADR 0008).
    async fn acquire(&self, request: AcquireRequest) -> Result<AcquiredBatch, Self::Error> {
        claim::acquire(self, request).await
    }

    /// Marks rows published, worker-guarded (`locked_by = $2`). A row already completed or
    /// reclaimed by another worker contributes nothing to the count — a shortfall is logged at
    /// `debug`, never an error (ADR 0008).
    async fn complete(
        &self,
        worker: &WorkerId,
        items: &[CompletedRecord],
    ) -> Result<u64, Self::Error> {
        outcomes::complete(self, worker, items).await
    }

    /// Applies each item's [`FailureOutcome`](reliar_outbox::FailureOutcome), worker-guarded. Retry rows get
    /// `available_at = now() + delay` computed in SQL (ADR 0009); dead rows get `dead_at`/
    /// `dead_reason` set together (`ck_outbox_dead_reason`). Both increment `attempts` — on
    /// outcome, never on claim.
    async fn fail(&self, worker: &WorkerId, items: &[FailedRecord]) -> Result<u64, Self::Error> {
        outcomes::fail(self, worker, items).await
    }

    /// Clears the lease for rows this worker still owns. `available_at` and `attempts` are
    /// untouched — a release is not a failure.
    async fn release(&self, worker: &WorkerId, items: &[RecordRef]) -> Result<u64, Self::Error> {
        outcomes::release(self, worker, items).await
    }

    /// Renews `locked_until = now() + lease` for rows this worker still owns. Best-effort: a
    /// shortfall means the lease already expired.
    async fn extend_lease(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
        lease: std::time::Duration,
    ) -> Result<u64, Self::Error> {
        outcomes::extend_lease(self, worker, items, lease).await
    }

    /// **One bounded pass, three statements, each capped at `request.batch_size`**:
    /// published-row delete, dead-row delete, and the expired→dead sweep — none of the
    /// three is ever an unbounded `DELETE`/`UPDATE`. The sweep's predicate carries the claim's
    /// lease clause (`locked_until IS NULL OR locked_until < now()`), so it never transitions a
    /// row a live worker still owns — that worker's own `complete`/`fail`
    /// wins, and the row becomes sweepable only once its lease lapses.
    async fn purge(&self, request: PurgeRequest) -> Result<PurgeReport, Self::Error> {
        purge::purge(self, request).await
    }

    /// One statement, **four independently planned scalar subqueries** (ADR 0040 §3; supersedes
    /// the earlier single-scan `FILTER`-aggregate form, which was `O(table)`). Each subquery is
    /// aimed at its own partial index — `pending` and `oldest_pending_available_at` at
    /// `ix_outbox_claimable` (an index-only scan can evaluate a filter on its `INCLUDE`d
    /// `locked_until`/`expires_at`), `dead` at `ix_outbox_dead_cursor`, `expired_pending` at
    /// `ix_outbox_expires` — so the cost is `O(claimable backlog)`/`O(dead rows)`/`O(expired
    /// rows)`, never `O(table)`, and `oldest_pending_available_at` is a single-row `LIMIT`. One
    /// round trip, one transaction snapshot (`now()` evaluated once), so `as_of` and the four
    /// values are consistent with each other even though each is planned separately. Measured at
    /// 100k rows (mixed pending/leased/published/dead/expired) on a vacuumed table, every
    /// subquery plans as an index-only scan with zero heap fetches.
    async fn stats(&self) -> Result<OutboxStats, Self::Error> {
        purge::stats(self).await
    }
}