reliar-store-postgres 0.9.0

//! `PostgresOutboxSettings`, with an opt-in environment loader (ADR 0017).
//!
//! **The library never reads the environment implicitly.** No constructor, `Default` or
//! builder method touches [`std::env`] — only [`PostgresOutboxSettings::from_env`] does, and
//! only when called (ADR 0019).

use std::env::VarError;
use std::time::Duration;

use reliar_core::SettingsError;

/// What is provider-specific about the outbox. Everything portable lives in
/// `reliar_outbox::OutboxSettings`.
///
/// Built from [`Self::default`] plus builder methods, never a struct literal (`#[non_exhaustive]`
/// so a new field never breaks a caller outside this crate):
///
/// ```
/// use reliar_store_postgres::PostgresOutboxSettings;
/// use std::time::Duration;
///
/// let settings = PostgresOutboxSettings::default().statement_timeout(Duration::from_secs(2));
///
/// assert_eq!(settings.statement_timeout, Duration::from_secs(2));
/// ```
#[derive(Clone, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(default, deny_unknown_fields))]
#[non_exhaustive]
pub struct PostgresOutboxSettings {
    /// Applied as `SET LOCAL statement_timeout` inside the short transaction Reliar opens for
    /// **every** statement it issues on its own pool — `acquire`, `complete`, `fail`, `release`,
    /// `extend_lease`, `stats`, `purge` (each of its three statements), `list_dead`,
    /// `retry_dead`, `purge_dead` — **never** the caller's `enqueue` transaction, which is the
    /// caller's own to bound. `Duration::ZERO` (the default) issues nothing and inherits the
    /// server/role setting; a non-zero value costs a `BEGIN`/`SET LOCAL`/statement(s)/`COMMIT`
    /// round trip on every call. `STATEMENT_TIMEOUT_MS`.
    #[cfg_attr(
        feature = "serde",
        serde(rename = "statement_timeout_ms", with = "crate::duration_serde")
    )]
    pub statement_timeout: Duration,
}

impl Default for PostgresOutboxSettings {
    fn default() -> Self {
        Self {
            statement_timeout: Duration::ZERO,
        }
    }
}

impl PostgresOutboxSettings {
    /// Sets [`Self::statement_timeout`].
    ///
    /// ```
    /// use reliar_store_postgres::PostgresOutboxSettings;
    /// use std::time::Duration;
    ///
    /// let settings = PostgresOutboxSettings::default()
    ///     .statement_timeout(Duration::from_millis(500));
    /// assert_eq!(settings.statement_timeout, Duration::from_millis(500));
    /// ```
    #[must_use]
    pub const fn statement_timeout(mut self, timeout: Duration) -> Self {
        self.statement_timeout = timeout;

        self
    }

    /// Opt-in. Starts from [`Self::default`], overrides **only** the variables present under
    /// `prefix`, and returns `Err` for a present-but-unparseable or out-of-range value — never
    /// a silent fallback to the default.
    ///
    /// ```
    /// use reliar_store_postgres::PostgresOutboxSettings;
    /// use std::time::Duration;
    ///
    /// // SAFETY: doctests run single-threaded per binary; no other code reads this var.
    /// unsafe { std::env::set_var("EXAMPLE_OUTBOX_STATEMENT_TIMEOUT_MS", "500") };
    /// let settings = PostgresOutboxSettings::from_env("EXAMPLE_OUTBOX_")?;
    /// assert_eq!(settings.statement_timeout, Duration::from_millis(500));
    /// # unsafe { std::env::remove_var("EXAMPLE_OUTBOX_STATEMENT_TIMEOUT_MS") };
    /// # Ok::<(), reliar_core::SettingsError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`SettingsError::Parse`] for a present variable that cannot be parsed as its
    /// declared type.
    pub fn from_env(prefix: &str) -> Result<Self, SettingsError> {
        let mut settings = Self::default();

        if let Some(v) = env_duration_ms(prefix, "STATEMENT_TIMEOUT_MS")? {
            settings.statement_timeout = v;
        }

        Ok(settings)
    }
}

/// What is provider-specific about the inbox (inbox contract §3). A **separate** settings type
/// from [`PostgresOutboxSettings`] — the inbox has no lease/ordering/retention knobs. It does
/// share the outbox's [`Self::statement_timeout`] knob, applied to the same kind of call: every
/// statement the inbox issues on its **own pool** (`fail`, `find`, `purge`) rather than the
/// caller's transaction (`claim`/`complete`, which stay the caller's to bound).
///
/// Built from [`Self::default`] plus builder methods, never a struct literal
/// (`#[non_exhaustive]` so a new field never breaks a caller outside this crate):
///
/// ```
/// use reliar_store_postgres::PostgresInboxSettings;
/// use std::time::Duration;
///
/// let settings = PostgresInboxSettings::default()
///     .statement_timeout(Duration::from_secs(2))
///     .max_attempts(5);
///
/// assert_eq!(settings.statement_timeout, Duration::from_secs(2));
/// assert_eq!(settings.max_attempts, 5);
/// ```
#[derive(Clone, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(default, deny_unknown_fields))]
#[non_exhaustive]
pub struct PostgresInboxSettings {
    /// Applied as `SET LOCAL statement_timeout` inside the short transaction Reliar opens for
    /// every statement it issues on its **own pool** — `fail`, `find`, each of `purge`'s three
    /// deletes (completed, incomplete, dead), and every `InboxDeadLetters` call
    /// (`list_dead`/`retry_dead`/`purge_dead`) — **never** the caller's `claim`/`complete`
    /// transaction, which is the caller's own to bound. Exists chiefly for `fail`'s
    /// `INSERT … ON CONFLICT DO UPDATE`, which otherwise blocks unbounded behind a concurrent
    /// claimer's still-open transaction on the same `(scope, message_id)` — a canceled statement
    /// classifies `FailureKind::Transient`, so leaving this at `Duration::ZERO` lets `fail` block
    /// for as long as the concurrent handler's own transaction runs. `Duration::ZERO` (the
    /// default) issues nothing and inherits the server/role setting; a non-zero value costs a
    /// `BEGIN`/`SET LOCAL`/statement/`COMMIT` round trip on every call. `STATEMENT_TIMEOUT_MS`.
    #[cfg_attr(
        feature = "serde",
        serde(rename = "statement_timeout_ms", with = "crate::duration_serde")
    )]
    pub statement_timeout: Duration,

    /// The bound [`crate::PostgresInboxStore`]'s `InboxStore::fail` applies to **recorded**
    /// failures before setting `dead_at` (ADR 0042 A.2.4). Lives here, in the provider, rather
    /// than in `reliar-inbox`, because the transition must be computed atomically with the
    /// increment — `fail`'s single `INSERT … ON CONFLICT DO UPDATE` decides
    /// `attempts + 1 >= max_attempts` in SQL.
    ///
    /// `0` is a configuration error, rejected by `Self::validate` and therefore by
    /// [`crate::PostgresInboxStore::with_settings`] — `0` reads as "no retries" and does the
    /// opposite. `u32::MAX` spells "unbounded" explicitly. Default 10. `MAX_ATTEMPTS`.
    pub max_attempts: u32,
}

/// The default [`PostgresInboxSettings::max_attempts`] — restated here rather than imported from
/// `reliar-inbox`, which names no such constant (it depends on no storage engine and defines no
/// default retry bound of its own).
const DEFAULT_MAX_ATTEMPTS: u32 = 10;

impl Default for PostgresInboxSettings {
    fn default() -> Self {
        Self {
            statement_timeout: Duration::ZERO,
            max_attempts: DEFAULT_MAX_ATTEMPTS,
        }
    }
}

impl PostgresInboxSettings {
    /// Sets [`Self::statement_timeout`].
    ///
    /// ```
    /// use reliar_store_postgres::PostgresInboxSettings;
    /// use std::time::Duration;
    ///
    /// let settings = PostgresInboxSettings::default()
    ///     .statement_timeout(Duration::from_millis(500));
    /// assert_eq!(settings.statement_timeout, Duration::from_millis(500));
    /// ```
    #[must_use]
    pub const fn statement_timeout(mut self, timeout: Duration) -> Self {
        self.statement_timeout = timeout;

        self
    }

    /// Sets [`Self::max_attempts`].
    ///
    /// ```
    /// use reliar_store_postgres::PostgresInboxSettings;
    /// let settings = PostgresInboxSettings::default().max_attempts(3);
    /// assert_eq!(settings.max_attempts, 3);
    /// ```
    #[must_use]
    pub const fn max_attempts(mut self, max_attempts: u32) -> Self {
        self.max_attempts = max_attempts;

        self
    }

    /// Opt-in, mirroring [`PostgresOutboxSettings::from_env`]. Starts from [`Self::default`],
    /// overrides **only** the variables present under `prefix`.
    ///
    /// # Errors
    ///
    /// Returns [`SettingsError::Parse`] for a present-but-unparseable value.
    pub fn from_env(prefix: &str) -> Result<Self, SettingsError> {
        let mut settings = Self::default();

        if let Some(v) = env_duration_ms(prefix, "STATEMENT_TIMEOUT_MS")? {
            settings.statement_timeout = v;
        }

        if let Some(v) = env_u32(prefix, "MAX_ATTEMPTS")? {
            settings.max_attempts = v;
        }

        Ok(settings)
    }

    /// Rejects a configuration this crate can never honour: `max_attempts == 0` (see
    /// [`Self::max_attempts`]). Called by [`crate::PostgresInboxStore::with_settings`], not
    /// implicitly.
    ///
    /// # Errors
    ///
    /// [`crate::PostgresInboxError::InvalidSettings`].
    pub(crate) fn validate(&self) -> Result<(), crate::PostgresInboxError> {
        if self.max_attempts == 0 {
            return Err(crate::PostgresInboxError::InvalidSettings {
                message: "max_attempts must not be 0 (reads as \"no retries\"; use u32::MAX for \
                          unbounded)"
                    .to_owned(),
            });
        }

        Ok(())
    }
}

fn env_duration_ms(prefix: &str, suffix: &str) -> Result<Option<Duration>, SettingsError> {
    let key = format!("{prefix}{suffix}");
    let raw = match std::env::var(&key) {
        Ok(value) => value,
        Err(VarError::NotPresent) => return Ok(None),
        Err(VarError::NotUnicode(_)) => return Err(SettingsError::parse(key, "a UTF-8 string")),
    };
    let ms = raw
        .trim()
        .parse::<u64>()
        .map_err(|_| SettingsError::parse(key, "milliseconds"))?;

    Ok(Some(Duration::from_millis(ms)))
}

fn env_u32(prefix: &str, suffix: &str) -> Result<Option<u32>, SettingsError> {
    let key = format!("{prefix}{suffix}");
    let raw = match std::env::var(&key) {
        Ok(value) => value,
        Err(VarError::NotPresent) => return Ok(None),
        Err(VarError::NotUnicode(_)) => return Err(SettingsError::parse(key, "a UTF-8 string")),
    };

    raw.trim()
        .parse::<u32>()
        .map(Some)
        .map_err(|_| SettingsError::parse(key, "u32"))
}