reliar-store-postgres 0.7.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()
///     .schema("orders")
///     .enqueue_sets_search_path(true)
///     .statement_timeout(Duration::from_secs(2));
///
/// assert_eq!(settings.schema, "orders");
/// assert!(settings.enqueue_sets_search_path);
/// 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 {
    /// The schema `PostgresOutboxStore::new`/`connect` verifies `outbox` resolves to, and the
    /// same default [`crate::MigrateOptions::schema`] uses. The two SHALL agree — if `migrate()`
    /// used a different schema, `outbox` is absent here and construction fails with
    /// [`crate::PostgresOutboxError::NotMigrated`] or, if a same-named table exists elsewhere on
    /// the path, [`crate::PostgresOutboxError::SchemaNotOnSearchPath`]. `SCHEMA`. Default `"reliar"`.
    pub schema: String,

    /// When `true`, `enqueue` wraps its `INSERT` in a transaction-local
    /// `set_config('search_path', …, true)` and restores the caller's previous value
    /// afterward — for hosts that can change neither the connection URL nor the role. Costs
    /// three extra statements per `enqueue`, which is why it defaults to `false`.
    /// `ENQUEUE_SETS_SEARCH_PATH`.
    pub enqueue_sets_search_path: bool,

    /// 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 {
            schema: "reliar".to_owned(),
            enqueue_sets_search_path: false,
            statement_timeout: Duration::ZERO,
        }
    }
}

impl PostgresOutboxSettings {
    /// Sets [`Self::schema`].
    ///
    /// ```
    /// use reliar_store_postgres::PostgresOutboxSettings;
    /// let settings = PostgresOutboxSettings::default().schema("orders");
    /// assert_eq!(settings.schema, "orders");
    /// ```
    #[must_use]
    pub fn schema(mut self, schema: impl Into<String>) -> Self {
        self.schema = schema.into();

        self
    }

    /// Sets [`Self::enqueue_sets_search_path`].
    ///
    /// ```
    /// use reliar_store_postgres::PostgresOutboxSettings;
    /// let settings = PostgresOutboxSettings::default().enqueue_sets_search_path(true);
    /// assert!(settings.enqueue_sets_search_path);
    /// ```
    #[must_use]
    pub const fn enqueue_sets_search_path(mut self, enabled: bool) -> Self {
        self.enqueue_sets_search_path = enabled;

        self
    }

    /// 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;
    ///
    /// // SAFETY: doctests run single-threaded per binary; no other code reads this var.
    /// unsafe { std::env::set_var("EXAMPLE_OUTBOX_SCHEMA", "orders") };
    /// let settings = PostgresOutboxSettings::from_env("EXAMPLE_OUTBOX_")?;
    /// assert_eq!(settings.schema, "orders");
    /// # unsafe { std::env::remove_var("EXAMPLE_OUTBOX_SCHEMA") };
    /// # 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_raw(prefix, "SCHEMA")? {
            settings.schema = v;
        }

        if let Some(v) = env_bool(prefix, "ENQUEUE_SETS_SEARCH_PATH")? {
            settings.enqueue_sets_search_path = v;
        }

        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()
///     .schema("orders")
///     .claim_sets_search_path(true)
///     .statement_timeout(Duration::from_secs(2))
///     .max_attempts(5);
///
/// assert_eq!(settings.schema, "orders");
/// assert!(settings.claim_sets_search_path);
/// 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 {
    /// The schema [`crate::PostgresInboxStore::connect`] verifies `inbox` resolves to. SHALL
    /// agree with whatever schema `migrate()` was given — the inbox shares the outbox's single
    /// `migrate()`, schema and `_migrations` table (inbox contract §1). `SCHEMA`. Default
    /// `"reliar"`.
    pub schema: String,

    /// When `true`, [`crate::PostgresInboxStore`]'s `claim` and `complete` wrap their statements
    /// in a transaction-local `set_config('search_path', …, true)` and restore the caller's
    /// previous value afterward — for a caller that overrides `search_path` inside its own
    /// transaction (a host that could change neither the connection URL nor the role would
    /// already fail `connect`'s own `search_path` verification, so this is not that case).
    /// Mirrors [`PostgresOutboxSettings::enqueue_sets_search_path`]; defaults to `false` for the
    /// same reason (it costs extra statements on the caller's own transaction).
    /// `CLAIM_SETS_SEARCH_PATH`.
    pub claim_sets_search_path: bool,

    /// 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::connect`] — `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 {
            schema: "reliar".to_owned(),
            claim_sets_search_path: false,
            statement_timeout: Duration::ZERO,
            max_attempts: DEFAULT_MAX_ATTEMPTS,
        }
    }
}

impl PostgresInboxSettings {
    /// Sets [`Self::schema`].
    ///
    /// ```
    /// use reliar_store_postgres::PostgresInboxSettings;
    /// let settings = PostgresInboxSettings::default().schema("orders");
    /// assert_eq!(settings.schema, "orders");
    /// ```
    #[must_use]
    pub fn schema(mut self, schema: impl Into<String>) -> Self {
        self.schema = schema.into();

        self
    }

    /// Sets [`Self::claim_sets_search_path`].
    ///
    /// ```
    /// use reliar_store_postgres::PostgresInboxSettings;
    /// let settings = PostgresInboxSettings::default().claim_sets_search_path(true);
    /// assert!(settings.claim_sets_search_path);
    /// ```
    #[must_use]
    pub const fn claim_sets_search_path(mut self, enabled: bool) -> Self {
        self.claim_sets_search_path = enabled;

        self
    }

    /// 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_raw(prefix, "SCHEMA")? {
            settings.schema = v;
        }

        if let Some(v) = env_bool(prefix, "CLAIM_SETS_SEARCH_PATH")? {
            settings.claim_sets_search_path = v;
        }

        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::connect`], 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_raw(prefix: &str, suffix: &str) -> Result<Option<String>, SettingsError> {
    let key = format!("{prefix}{suffix}");

    match std::env::var(&key) {
        Ok(value) => Ok(Some(value)),
        Err(VarError::NotPresent) => Ok(None),
        Err(VarError::NotUnicode(_)) => Err(SettingsError::parse(key, "a UTF-8 string")),
    }
}

fn env_bool(prefix: &str, suffix: &str) -> Result<Option<bool>, SettingsError> {
    let Some(raw) = env_raw(prefix, suffix)? else {
        return Ok(None);
    };

    match raw.trim().to_ascii_lowercase().as_str() {
        "true" | "1" => Ok(Some(true)),
        "false" | "0" => Ok(Some(false)),
        _ => Err(SettingsError::parse(
            format!("{prefix}{suffix}"),
            "bool (\"true\"/\"false\"/\"1\"/\"0\")",
        )),
    }
}

fn env_duration_ms(prefix: &str, suffix: &str) -> Result<Option<Duration>, SettingsError> {
    let Some(raw) = env_raw(prefix, suffix)? else {
        return Ok(None);
    };
    let ms = raw
        .trim()
        .parse::<u64>()
        .map_err(|_| SettingsError::parse(format!("{prefix}{suffix}"), "milliseconds"))?;

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

fn env_u32(prefix: &str, suffix: &str) -> Result<Option<u32>, SettingsError> {
    let Some(raw) = env_raw(prefix, suffix)? else {
        return Ok(None);
    };

    raw.trim()
        .parse::<u32>()
        .map(Some)
        .map_err(|_| SettingsError::parse(format!("{prefix}{suffix}"), "u32"))
}