reliar-store-postgres 0.6.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, CompletedMessage, FailedMessage, MessageRef, OutboxStats,
    OutboxStore, PurgeReport, PurgeRequest, WorkerId,
};
use sqlx::{PgPool, Postgres, Transaction};

use crate::error::PostgresStoreError;
use crate::settings::PostgresOutboxSettings;

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

use super::{claim, outcomes, purge, schema};

/// 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 `store/` reads these directly rather than through
    // an accessor — they are `store`-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`: fails fast with
    /// [`PostgresStoreError::UnsupportedServerVersion`], [`PostgresStoreError::SchemaResolution`]
    /// (`search_path` problem) or [`PostgresStoreError::NotMigrated`] (the relation is missing
    /// entirely) 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 [`PostgresStoreError::UnsupportedServerVersion`],
    /// [`PostgresStoreError::NotMigrated`], [`PostgresStoreError::SchemaResolution`], or
    /// [`PostgresStoreError::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, PostgresStoreError> {
        if !crate::error::is_valid_schema_name(&settings.schema) {
            return Err(PostgresStoreError::InvalidSchema {
                schema: settings.schema,
            });
        }

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

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

        let check = schema::verify_schema(&pool, &settings.schema).await?;

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

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

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

        let others = schema::other_outbox_schemas(&pool, &settings.schema).await?;

        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
    /// [`PostgresStoreError`], catching SQLSTATE `42P01` on **every** call, not just startup
    /// verification.
    pub(super) fn map_err(&self, err: sqlx::Error) -> PostgresStoreError {
        crate::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<(), PostgresStoreError> {
        self.set_local_timeout_raw(tx)
            .await
            .map_err(|e| self.map_err(e))
    }

    /// [`Self::set_local_timeout`] without the `PostgresStoreError` 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, PostgresStoreError> {
        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, PostgresStoreError> {
        Self::connect(pool, settings, JsonSerializer).await
    }
}

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

    /// 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: &[CompletedMessage],
    ) -> 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: &[FailedMessage]) -> 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: &[MessageRef]) -> 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: &[MessageRef],
        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_at`, `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
    }
}