reliar-outbox 0.8.0

//! Storage side of the outbox: the [`OutboxStore`]/[`OutboxDeadLetters`] traits and the request
//! and result types that cross their boundary (ADR 0006, ADR 0008, ADR 0009, ADR 0016,
//! ADR 0023).

use std::time::Duration;

use reliar_core::{Classify, MessageId};

use crate::claim_token::ClaimToken;
use crate::ordering::Ordering;
use crate::record::{OutboxRecord, truncate_error};
use crate::record_id::OutboxRecordId;
use crate::settings::RetentionSettings;
use crate::worker::WorkerId;

/// What [`OutboxStore::acquire`] claims.
///
/// `#[non_exhaustive]`: build with [`Self::new`] and the builder methods, never a struct
/// literal, so a new field never breaks a caller outside this crate.
///
/// ```
/// use reliar_outbox::{AcquireRequest, WorkerId};
///
/// let request = AcquireRequest::new(WorkerId::generate()).batch_size(10);
/// assert_eq!(request.batch_size, 10);
/// ```
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct AcquireRequest {
    /// The claiming worker; every claimed row's `locked_by` is set to this value.
    pub worker: WorkerId,

    /// The maximum number of rows to claim. Default 100.
    pub batch_size: u32,

    /// How long the claim holds the lease before it may be reclaimed. Default 30 s.
    pub lease: Duration,

    /// The ordering strategy to claim under. Default [`Ordering::Unordered`].
    pub ordering: Ordering,
}

impl AcquireRequest {
    /// Starts a request for `worker` with the documented defaults: `batch_size = 100`,
    /// `lease = 30s`, `ordering = Unordered`.
    ///
    /// ```
    /// use reliar_outbox::{AcquireRequest, WorkerId};
    /// use std::time::Duration;
    ///
    /// let request = AcquireRequest::new(WorkerId::generate());
    /// assert_eq!(request.batch_size, 100);
    /// assert_eq!(request.lease, Duration::from_secs(30));
    /// ```
    #[must_use]
    pub fn new(worker: WorkerId) -> Self {
        Self {
            worker,
            batch_size: 100,
            lease: Duration::from_secs(30),
            ordering: Ordering::default(),
        }
    }

    /// Sets [`Self::batch_size`].
    ///
    /// ```
    /// use reliar_outbox::{AcquireRequest, WorkerId};
    /// let request = AcquireRequest::new(WorkerId::generate()).batch_size(25);
    /// assert_eq!(request.batch_size, 25);
    /// ```
    #[must_use]
    pub const fn batch_size(mut self, batch_size: u32) -> Self {
        self.batch_size = batch_size;

        self
    }

    /// Sets [`Self::lease`].
    ///
    /// ```
    /// use reliar_outbox::{AcquireRequest, WorkerId};
    /// use std::time::Duration;
    /// let lease = Duration::from_secs(5);
    /// let request = AcquireRequest::new(WorkerId::generate()).lease(lease);
    /// assert_eq!(request.lease, lease);
    /// ```
    #[must_use]
    pub const fn lease(mut self, lease: Duration) -> Self {
        self.lease = lease;

        self
    }

    /// Sets [`Self::ordering`].
    ///
    /// ```
    /// use reliar_outbox::{AcquireRequest, Ordering, WorkerId};
    /// let request = AcquireRequest::new(WorkerId::generate()).ordering(Ordering::Unordered);
    /// assert_eq!(request.ordering, Ordering::Unordered);
    /// ```
    #[must_use]
    pub const fn ordering(mut self, ordering: Ordering) -> Self {
        self.ordering = ordering;

        self
    }
}

/// The result of one [`OutboxStore::acquire`] call.
///
/// ```
/// use reliar_outbox::AcquiredBatch;
/// assert!(AcquiredBatch::default().is_empty());
/// ```
#[derive(Debug, Default)]
#[non_exhaustive]
pub struct AcquiredBatch {
    /// The rows claimed and successfully decoded.
    pub records: Vec<OutboxRecord>,

    /// Rows the provider could not decode **and has attempted to move to dead**
    /// ([`DeadReason::Undecodable`], ADR 0008). That follow-up is best-effort (ADR 0039): if it
    /// failed, the provider logs it and the row stays leased, re-claimed after its lease lapses
    /// — so a count derived from this field may double-count a row across two claims.
    pub poisoned: Vec<PoisonedRow>,
}

impl AcquiredBatch {
    /// Builds an acquired batch from its claimed and poisoned rows.
    ///
    /// ```
    /// use reliar_outbox::AcquiredBatch;
    /// let batch = AcquiredBatch::new(Vec::new(), Vec::new());
    /// assert!(batch.is_empty());
    /// ```
    #[must_use]
    pub fn new(records: Vec<OutboxRecord>, poisoned: Vec<PoisonedRow>) -> Self {
        Self { records, poisoned }
    }

    /// `true` when there are no records **and** no poisoned rows.
    ///
    /// ```
    /// use reliar_outbox::AcquiredBatch;
    /// assert!(AcquiredBatch::default().is_empty());
    /// ```
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.records.is_empty() && self.poisoned.is_empty()
    }
}

/// Keyset cursor for [`OutboxDeadLetters::list_dead`].
///
/// The pair is opaque so callers cannot accidentally advance only one half of the ordering key.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
pub struct DeadCursor {
    #[cfg_attr(feature = "serde", serde(with = "time::serde::rfc3339"))]
    dead_at: time::OffsetDateTime,

    id: OutboxRecordId,
}

impl DeadCursor {
    /// Reconstructs a cursor previously returned to or persisted by the caller.
    #[must_use]
    pub const fn new(dead_at: time::OffsetDateTime, id: OutboxRecordId) -> Self {
        Self { dead_at, id }
    }

    /// Returns the database time at which the cursor's row became dead.
    #[must_use]
    pub const fn dead_at(self) -> time::OffsetDateTime {
        self.dead_at
    }

    /// Returns the row id used to break ties between equal death times.
    #[must_use]
    pub const fn id(self) -> OutboxRecordId {
        self.id
    }
}

/// One page of [`OutboxDeadLetters::list_dead`].
///
/// Poisoned rows here are **already dead**, so unlike [`AcquiredBatch`] there is no transition to
/// make — they are reported so an operator can see them (ADR 0023).
///
/// ```
/// use reliar_outbox::DeadLetterPage;
/// assert!(DeadLetterPage::default().is_empty());
/// ```
#[derive(Debug, Default)]
#[non_exhaustive]
pub struct DeadLetterPage {
    /// The dead rows decoded successfully.
    pub records: Vec<OutboxRecord>,

    /// Dead rows the store could not decode, reported rather than silently skipped.
    pub poisoned: Vec<PoisonedRow>,

    /// Feeds the next [`DeadQuery::after`]. `None` when the page was not full — the cursor is
    /// computed over every row scanned, including poisoned ones, so a poisoned tail never loops
    /// the caller forever.
    pub next_after: Option<DeadCursor>,
}

impl DeadLetterPage {
    /// Builds a dead-letter page from its rows and keyset cursor.
    ///
    /// ```
    /// use reliar_outbox::DeadLetterPage;
    /// let page = DeadLetterPage::new(Vec::new(), Vec::new(), None);
    /// assert!(page.is_empty());
    /// assert!(page.next_after.is_none());
    /// ```
    #[must_use]
    pub fn new(
        records: Vec<OutboxRecord>,
        poisoned: Vec<PoisonedRow>,
        next_after: Option<DeadCursor>,
    ) -> Self {
        Self {
            records,
            poisoned,
            next_after,
        }
    }

    /// `true` when there are no records **and** no poisoned rows.
    ///
    /// ```
    /// use reliar_outbox::DeadLetterPage;
    /// assert!(DeadLetterPage::default().is_empty());
    /// ```
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.records.is_empty() && self.poisoned.is_empty()
    }
}

/// A row the store could not decode (a corrupt envelope, an unknown `dead_reason`, an
/// unsupported metadata version). Reported, never silently dropped.
///
/// A poisoned row is undecodable in its *envelope* columns; `id` and `message_id` are plain `uuid`
/// columns and are always readable regardless, so both are reported (ADR 0044 §3).
///
/// ```
/// use reliar_core::MessageId;
/// use reliar_outbox::{OutboxRecordId, PoisonedRow};
/// use reliar_core::uuid::Uuid;
///
/// let id = OutboxRecordId::from_uuid(Uuid::now_v7());
/// let row = PoisonedRow::new(id, MessageId::new(), "corrupt envelope");
/// assert_eq!(row.error, "corrupt envelope");
/// ```
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct PoisonedRow {
    /// The row's own identity.
    pub id: OutboxRecordId,

    /// The row's message id.
    pub message_id: MessageId,

    /// The decode failure. Truncated to 2 KiB at a char boundary.
    pub error: String,
}

impl PoisonedRow {
    /// Builds a poisoned-row report, truncating `error` to 2 KiB at a char boundary with a
    /// `"…[truncated]"` marker.
    ///
    /// ```
    /// use reliar_core::MessageId;
    /// use reliar_outbox::{OutboxRecordId, PoisonedRow};
    /// use reliar_core::uuid::Uuid;
    ///
    /// let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// let row = PoisonedRow::new(id, MessageId::new(), "unknown dead_reason column value");
    /// assert_eq!(row.id, id);
    /// ```
    #[must_use]
    pub fn new(id: OutboxRecordId, message_id: MessageId, error: impl Into<String>) -> Self {
        Self {
            id,
            message_id,
            error: truncate_error(error.into()),
        }
    }
}

/// Identifies one row for a by-id [`OutboxStore`] operation. Carries `created_at` alongside `id`
/// because a partitioned table needs it to prune to the right partition (ADR 0016). Built from
/// [`OutboxRecordId`] — the row's own identity, never the envelope's `MessageId` (ADR 0044 §3).
///
/// ```
/// use reliar_outbox::{OutboxRecordId, RecordRef};
/// use time::OffsetDateTime;
/// use reliar_core::uuid::Uuid;
/// let record_ref = RecordRef::new(OutboxRecordId::from_uuid(Uuid::now_v7()), OffsetDateTime::now_utc());
/// assert_eq!(record_ref, record_ref.clone());
/// ```
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct RecordRef {
    /// The row's own identity.
    pub id: OutboxRecordId,

    /// The row's immutable creation timestamp.
    pub created_at: time::OffsetDateTime,

    /// The claim this reference belongs to, when it came from one. `None` on a reference built
    /// for a dead row ([`OutboxDeadLetters::retry_dead`]/[`OutboxDeadLetters::purge_dead`]), which
    /// holds no lease, or on one built by [`Self::new`].
    ///
    /// **A `None` token is fenced** (ADR 0046 Amendment A): `complete`/`fail`/`release`/
    /// `extend_lease` affect no row for it. Obtain a claim-scoped reference from
    /// [`crate::OutboxRecord::record_ref`], never [`Self::new`].
    pub claim_token: Option<ClaimToken>,
}

impl RecordRef {
    /// Builds a row reference carrying no claim token — for the dead-letter operations, which
    /// hold no lease. A reference built this way is **fenced** for every claim-scoped operation
    /// (`complete`/`fail`/`release`/`extend_lease`): use [`crate::OutboxRecord::record_ref`]
    /// instead when the row is claimed.
    ///
    /// ```
    /// use reliar_outbox::{OutboxRecordId, RecordRef};
    /// use time::OffsetDateTime;
    /// use reliar_core::uuid::Uuid;
    ///
    /// let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// let created_at = OffsetDateTime::now_utc();
    /// let record_ref = RecordRef::new(id, created_at);
    /// assert_eq!(record_ref.id, id);
    /// assert!(record_ref.claim_token.is_none());
    /// ```
    #[must_use]
    pub const fn new(id: OutboxRecordId, created_at: time::OffsetDateTime) -> Self {
        Self {
            id,
            created_at,
            claim_token: None,
        }
    }

    /// Builds a row reference scoped to the claim that produced `token` — what
    /// [`crate::OutboxRecord::record_ref`] returns for a row [`OutboxStore::acquire`] claimed.
    ///
    /// ```
    /// use reliar_outbox::{ClaimToken, OutboxRecordId, RecordRef};
    /// use time::OffsetDateTime;
    /// use reliar_core::uuid::Uuid;
    ///
    /// let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// let created_at = OffsetDateTime::now_utc();
    /// let token = ClaimToken::from_uuid(Uuid::now_v7());
    /// let record_ref = RecordRef::claimed(id, created_at, token);
    /// assert_eq!(record_ref.claim_token, Some(token));
    /// ```
    #[must_use]
    pub const fn claimed(
        id: OutboxRecordId,
        created_at: time::OffsetDateTime,
        token: ClaimToken,
    ) -> Self {
        Self {
            id,
            created_at,
            claim_token: Some(token),
        }
    }
}

/// One row to mark published in [`OutboxStore::complete`].
///
/// ```
/// use reliar_outbox::{CompletedRecord, OutboxRecordId, RecordRef};
/// use time::OffsetDateTime;
/// use reliar_core::uuid::Uuid;
/// let record_ref = RecordRef::new(OutboxRecordId::from_uuid(Uuid::now_v7()), OffsetDateTime::now_utc());
/// assert_eq!(CompletedRecord::new(record_ref).record, record_ref);
/// ```
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct CompletedRecord {
    /// The row that published successfully.
    pub record: RecordRef,
}

impl CompletedRecord {
    /// Builds a completed-record report.
    ///
    /// ```
    /// use reliar_outbox::{CompletedRecord, OutboxRecordId, RecordRef};
    /// use time::OffsetDateTime;
    /// use reliar_core::uuid::Uuid;
    ///
    /// let record_ref = RecordRef::new(OutboxRecordId::from_uuid(Uuid::now_v7()), OffsetDateTime::now_utc());
    /// let completed = CompletedRecord::new(record_ref);
    /// assert_eq!(completed.record, record_ref);
    /// ```
    #[must_use]
    pub const fn new(record: RecordRef) -> Self {
        Self { record }
    }
}

/// One row to apply a [`FailureOutcome`] to in [`OutboxStore::fail`].
///
/// ```
/// use reliar_outbox::{DeadReason, FailedRecord, FailureOutcome, OutboxRecordId, RecordRef};
/// use time::OffsetDateTime;
/// use reliar_core::uuid::Uuid;
/// let record_ref = RecordRef::new(OutboxRecordId::from_uuid(Uuid::now_v7()), OffsetDateTime::now_utc());
/// let outcome = FailureOutcome::Dead { reason: DeadReason::PermanentError };
/// let failed = FailedRecord::new(record_ref, "connection refused", outcome);
/// assert_eq!(failed.record, record_ref);
/// ```
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct FailedRecord {
    /// The row whose publish failed.
    pub record: RecordRef,

    /// The failure, already truncated and redacted.
    pub error: String,

    /// The decided outcome. The store never re-derives retry policy — it only applies this.
    pub outcome: FailureOutcome,
}

impl FailedRecord {
    /// Builds a failed-record report, truncating `error` to 2 KiB at a char boundary with a
    /// `"…[truncated]"` marker.
    ///
    /// ```
    /// use reliar_outbox::{DeadReason, FailedRecord, FailureOutcome, OutboxRecordId, RecordRef};
    /// use time::OffsetDateTime;
    /// use reliar_core::uuid::Uuid;
    ///
    /// let record_ref = RecordRef::new(OutboxRecordId::from_uuid(Uuid::now_v7()), OffsetDateTime::now_utc());
    /// let failed = FailedRecord::new(
    ///     record_ref,
    ///     "connection refused",
    ///     FailureOutcome::Dead { reason: DeadReason::PermanentError },
    /// );
    /// assert_eq!(failed.error, "connection refused");
    /// ```
    #[must_use]
    pub fn new(record: RecordRef, error: impl Into<String>, outcome: FailureOutcome) -> Self {
        Self {
            record,
            error: truncate_error(error),
            outcome,
        }
    }
}

/// What [`OutboxStore::fail`] should do with one row, as decided by a [`crate::RetryPolicy`].
///
/// ```
/// use reliar_outbox::{DeadReason, FailureOutcome};
/// use std::time::Duration;
///
/// let retry = FailureOutcome::Retry { delay: Duration::from_secs(1) };
/// let dead = FailureOutcome::Dead { reason: DeadReason::AttemptsExhausted };
/// assert_ne!(retry, dead);
/// ```
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum FailureOutcome {
    /// Retry later. The store applies it as `available_at = now() + delay` **in SQL**
    /// (ADR 0009).
    Retry {
        /// The delay before the row becomes claimable again.
        delay: Duration,
    },

    /// Terminal. The store sets `dead_at = now()` and records `reason`.
    Dead {
        /// Why the row is dead.
        reason: DeadReason,
    },
}

/// Why a row is dead. Persisted so an operator inspecting [`OutboxDeadLetters::list_dead`] can
/// tell a broker rejection from an expired message without reading `last_error`.
///
/// ```
/// use reliar_outbox::DeadReason;
/// assert_ne!(DeadReason::PermanentError, DeadReason::AttemptsExhausted);
/// ```
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum DeadReason {
    /// The publisher classified the failure as [`crate::FailureKind::Permanent`].
    PermanentError,

    /// The retry policy's `max_attempts` was reached.
    AttemptsExhausted,

    /// The row's `expires_at` passed before it was published.
    Expired,

    /// The store could not decode the row (a corrupt envelope, an unknown column value).
    Undecodable,
}

/// What [`OutboxStore::purge`] should delete or sweep in one bounded pass.
///
/// # Warning
///
/// [`Self::default`] leaves [`Self::dead_retention`] at `None`, so it deletes **zero** dead rows
/// and still reports success. A host that wants dead rows collected must set a value.
///
/// ```
/// use reliar_outbox::PurgeRequest;
/// use std::time::Duration;
///
/// let request = PurgeRequest::default().dead_retention(Some(Duration::from_secs(30 * 24 * 60 * 60)));
/// assert!(request.dead_retention.is_some());
/// ```
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct PurgeRequest {
    /// How long a published row is kept before it is deleted. `None` disables published
    /// purging. Default `Some(7 days)`.
    pub published_retention: Option<Duration>,

    /// How long a dead row is kept before it is deleted. `None` (the default) keeps dead rows
    /// until an explicit purge — deleting one is always a deliberate act.
    ///
    /// # Warning
    ///
    /// [`PurgeRequest::default`] leaves this at `None`: a default purge deletes zero dead rows
    /// and still reports success.
    pub dead_retention: Option<Duration>,

    /// The maximum number of rows deleted per call, for each of the published and dead passes.
    /// Default 1000.
    pub batch_size: u32,
}

/// **Hand-written, never derived**: a derived `Default` would give `None`/`None`/`0` — a
/// `purge` that deletes nothing and reports success.
impl Default for PurgeRequest {
    fn default() -> Self {
        Self {
            published_retention: Some(Duration::from_secs(7 * 24 * 60 * 60)),
            dead_retention: None,
            batch_size: 1_000,
        }
    }
}

impl PurgeRequest {
    /// Sets [`Self::published_retention`].
    ///
    /// ```
    /// use reliar_outbox::PurgeRequest;
    /// use std::time::Duration;
    /// let request = PurgeRequest::default().published_retention(None);
    /// assert!(request.published_retention.is_none());
    /// ```
    #[must_use]
    pub const fn published_retention(mut self, retention: Option<Duration>) -> Self {
        self.published_retention = retention;

        self
    }

    /// Sets [`Self::dead_retention`].
    ///
    /// ```
    /// use reliar_outbox::PurgeRequest;
    /// use std::time::Duration;
    /// let request = PurgeRequest::default().dead_retention(Some(Duration::from_secs(60)));
    /// assert_eq!(request.dead_retention, Some(Duration::from_secs(60)));
    /// ```
    #[must_use]
    pub const fn dead_retention(mut self, retention: Option<Duration>) -> Self {
        self.dead_retention = retention;

        self
    }

    /// Sets [`Self::batch_size`].
    ///
    /// ```
    /// use reliar_outbox::PurgeRequest;
    /// let request = PurgeRequest::default().batch_size(200);
    /// assert_eq!(request.batch_size, 200);
    /// ```
    #[must_use]
    pub const fn batch_size(mut self, batch_size: u32) -> Self {
        self.batch_size = batch_size;

        self
    }
}

/// Maps [`RetentionSettings`] onto the request shape [`OutboxStore::purge`] takes:
/// `published_retention` (already a `Duration`) becomes `Some(..)`, `dead_retention` carries
/// over as-is (`None` by default — see its `# Warning`), and `purge_batch_size` becomes
/// [`PurgeRequest::batch_size`]. Gives [`RetentionSettings`] its first consumer.
///
/// ```
/// use reliar_outbox::{PurgeRequest, RetentionSettings};
///
/// let settings = RetentionSettings::default();
/// let request = PurgeRequest::from(&settings);
/// assert_eq!(request.batch_size, settings.purge_batch_size);
/// ```
impl From<&RetentionSettings> for PurgeRequest {
    fn from(settings: &RetentionSettings) -> Self {
        Self {
            published_retention: Some(settings.published_retention),
            dead_retention: settings.dead_retention,
            batch_size: settings.purge_batch_size,
        }
    }
}

/// What one [`OutboxStore::purge`] call did.
///
/// ```
/// use reliar_outbox::PurgeReport;
/// assert_eq!(PurgeReport::default(), PurgeReport::new(0, 0, 0));
/// ```
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct PurgeReport {
    /// Published rows deleted.
    pub published_deleted: u64,

    /// Dead rows deleted.
    pub dead_deleted: u64,

    /// Pending rows swept to dead because their `expires_at` had passed.
    pub expired_to_dead: u64,
}

impl PurgeReport {
    /// Builds a purge report.
    ///
    /// ```
    /// use reliar_outbox::PurgeReport;
    /// let report = PurgeReport::new(3, 1, 0);
    /// assert_eq!(report.published_deleted, 3);
    /// ```
    #[must_use]
    pub const fn new(published_deleted: u64, dead_deleted: u64, expired_to_dead: u64) -> Self {
        Self {
            published_deleted,
            dead_deleted,
            expired_to_dead,
        }
    }

    /// `false` when any of the three counts hit `batch_size` — the caller should call
    /// [`OutboxStore::purge`] again; one call is one bounded pass, never an internal loop
    /// (ADR 0009). The expired-to-dead sweep is bounded by the same `batch_size` as the two
    /// deletes, so it is checked here too.
    ///
    /// ```
    /// use reliar_outbox::PurgeReport;
    /// assert!(PurgeReport::new(3, 1, 0).is_complete(1_000));
    /// assert!(!PurgeReport::new(1_000, 0, 0).is_complete(1_000));
    /// ```
    #[must_use]
    #[allow(
        clippy::cast_lossless,
        reason = "widening u32 -> u64; `u64::from` is not callable from a const fn on stable"
    )]
    pub const fn is_complete(&self, batch_size: u32) -> bool {
        self.published_deleted < batch_size as u64
            && self.dead_deleted < batch_size as u64
            && self.expired_to_dead < batch_size as u64
    }
}

/// A snapshot of the outbox's backlog, for the outbox-lag and dead-count gauges.
///
/// ```
/// use reliar_outbox::OutboxStats;
/// use time::OffsetDateTime;
/// let now = OffsetDateTime::now_utc();
/// let stats = OutboxStats::new(0, 0, 0, None, now);
/// assert_eq!(stats.pending, 0);
/// ```
#[derive(Clone, Copy, Debug)]
#[non_exhaustive]
pub struct OutboxStats {
    /// Claimable rows only — the same predicate `acquire` uses, so an expired row is excluded.
    /// Narrower than [`crate::OutboxRecord::state`]'s [`crate::OutboxState::Pending`]: a row
    /// backed off to a future `available_at` is `OutboxState::Pending` (not leased, published,
    /// or dead) but is not counted here until it becomes due.
    pub pending: u64,

    /// Dead rows.
    pub dead: u64,

    /// Pending rows whose `expires_at` has passed but have not yet been swept to dead by
    /// `purge`. Unclaimable; counted separately so they can be alerted on without pinning
    /// [`Self::lag`].
    pub expired_pending: u64,

    /// The oldest claimable row's `available_at`, over claimable rows only, so it cannot be
    /// pinned by an expired row.
    pub oldest_pending_available_at: Option<time::OffsetDateTime>,

    /// The database's `now()` at the moment of the query, so [`Self::lag`] never compares an
    /// application clock against a database one.
    pub as_of: time::OffsetDateTime,
}

impl OutboxStats {
    /// Builds a stats snapshot.
    ///
    /// ```
    /// use reliar_outbox::OutboxStats;
    /// use time::OffsetDateTime;
    ///
    /// let now = OffsetDateTime::now_utc();
    /// let stats = OutboxStats::new(3, 1, 0, Some(now), now);
    /// assert_eq!(stats.pending, 3);
    /// assert_eq!(stats.lag(), Some(std::time::Duration::ZERO));
    /// ```
    #[must_use]
    pub const fn new(
        pending: u64,
        dead: u64,
        expired_pending: u64,
        oldest_pending_available_at: Option<time::OffsetDateTime>,
        as_of: time::OffsetDateTime,
    ) -> Self {
        Self {
            pending,
            dead,
            expired_pending,
            oldest_pending_available_at,
            as_of,
        }
    }

    /// `as_of - oldest_pending_available_at`, clamped at zero. The "outbox lag" gauge. `None`
    /// when there is no claimable pending row.
    ///
    /// This is **scheduling lag** — how long the oldest claimable row has been *due* — not
    /// end-to-end age since it was enqueued; a row retried several times can be due "now" while
    /// having existed for hours. Use [`OutboxRecord::created_at`] for end-to-end age instead.
    ///
    /// ```
    /// use reliar_outbox::OutboxStats;
    /// use time::OffsetDateTime;
    ///
    /// let now = OffsetDateTime::now_utc();
    /// let stats = OutboxStats::new(0, 0, 0, None, now);
    /// assert!(stats.lag().is_none());
    ///
    /// let due = now - time::Duration::seconds(5);
    /// let stats = OutboxStats::new(1, 0, 0, Some(due), now);
    /// assert_eq!(stats.lag(), Some(std::time::Duration::from_secs(5)));
    /// ```
    #[must_use]
    pub fn lag(&self) -> Option<Duration> {
        let oldest = self.oldest_pending_available_at?;
        let diff = self.as_of - oldest;

        Some(if diff.is_negative() {
            Duration::ZERO
        } else {
            diff.unsigned_abs()
        })
    }
}

/// A filtered, paginated query over dead rows.
///
/// ```
/// use reliar_outbox::DeadQuery;
///
/// let query = DeadQuery::default().message_type("orders.created").limit(20);
/// assert_eq!(query.message_type.as_deref(), Some("orders.created"));
/// assert_eq!(query.limit, 20);
/// ```
#[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 DeadQuery {
    /// Restricts to one message type name (every version), if set.
    pub message_type: Option<String>,

    /// Restricts to one tenant, if set.
    pub tenant_id: Option<String>,

    /// Restricts to rows that went dead before this time, if set.
    #[cfg_attr(feature = "serde", serde(with = "time::serde::rfc3339::option"))]
    pub dead_before: Option<time::OffsetDateTime>,

    /// The maximum number of rows to return. Provider-capped. Default 100.
    pub limit: u32,

    /// Keyset pagination cursor: only rows ordered after this `(dead_at, id)` pair. Feed with
    /// [`DeadLetterPage::next_after`].
    pub after: Option<DeadCursor>,
}

/// **Hand-written, never derived**: a derived `Default` would set `limit = 0` and return
/// nothing.
impl Default for DeadQuery {
    fn default() -> Self {
        Self {
            message_type: None,
            tenant_id: None,
            dead_before: None,
            limit: 100,
            after: None,
        }
    }
}

impl DeadQuery {
    /// Sets [`Self::message_type`].
    ///
    /// ```
    /// use reliar_outbox::DeadQuery;
    /// let query = DeadQuery::default().message_type("orders.created");
    /// assert_eq!(query.message_type.as_deref(), Some("orders.created"));
    /// ```
    #[must_use]
    pub fn message_type(mut self, message_type: impl Into<String>) -> Self {
        self.message_type = Some(message_type.into());

        self
    }

    /// Sets [`Self::tenant_id`].
    ///
    /// ```
    /// use reliar_outbox::DeadQuery;
    /// let query = DeadQuery::default().tenant_id("tenant-1");
    /// assert_eq!(query.tenant_id.as_deref(), Some("tenant-1"));
    /// ```
    #[must_use]
    pub fn tenant_id(mut self, tenant_id: impl Into<String>) -> Self {
        self.tenant_id = Some(tenant_id.into());

        self
    }

    /// Sets [`Self::dead_before`].
    ///
    /// ```
    /// use reliar_outbox::DeadQuery;
    /// use time::OffsetDateTime;
    /// let before = OffsetDateTime::now_utc();
    /// let query = DeadQuery::default().dead_before(before);
    /// assert_eq!(query.dead_before, Some(before));
    /// ```
    #[must_use]
    pub const fn dead_before(mut self, dead_before: time::OffsetDateTime) -> Self {
        self.dead_before = Some(dead_before);

        self
    }

    /// Sets [`Self::limit`].
    ///
    /// ```
    /// use reliar_outbox::DeadQuery;
    /// assert_eq!(DeadQuery::default().limit(20).limit, 20);
    /// ```
    #[must_use]
    pub const fn limit(mut self, limit: u32) -> Self {
        self.limit = limit;

        self
    }

    /// Sets [`Self::after`].
    ///
    /// ```
    /// use reliar_outbox::{DeadCursor, DeadQuery, OutboxRecordId};
    /// use time::OffsetDateTime;
    /// use reliar_core::uuid::Uuid;
    /// let cursor = DeadCursor::new(OffsetDateTime::now_utc(), OutboxRecordId::from_uuid(Uuid::now_v7()));
    /// assert_eq!(DeadQuery::default().after(cursor).after, Some(cursor));
    /// ```
    #[must_use]
    pub const fn after(mut self, after: DeadCursor) -> Self {
        self.after = Some(after);

        self
    }
}

/// The dispatcher's portable side of the outbox. One provider implements this per database.
///
/// **`enqueue` is deliberately not here** — it must join the application's own transaction and
/// stays provider-inherent (ADR 0008).
///
/// Every row [`Self::acquire`] returns carries a [`ClaimToken`] identifying that claim of that
/// row. Every state-changing operation — `complete`, `fail`, `release`, `extend_lease`, and any
/// bookkeeping the provider does on a claimed row — **SHALL** match the row's stored token and
/// **SHALL** affect no row whose token differs or is absent. The `worker` argument identifies the
/// caller for diagnostics; **it is not the guard** — a store that matches only the worker does
/// not implement this trait (ADR 0046 Amendment A). Each returns the count of rows affected; a
/// shortfall means the row's claim was superseded (a fresher claim of the same row, by any
/// worker) and is **benign, never an error**.
///
/// This crate constructs no store of its own (ADR 0043) — a compiled generic call shape, never
/// invoked; `reliar-store-postgres`'s own rustdoc has the runnable version over
/// `PgPool::connect`.
///
/// ```
/// use reliar_outbox::OutboxStore;
///
/// async fn report_pending<S: OutboxStore>(store: &S) -> Result<(), S::Error> {
///     let stats = store.stats().await?;
///     println!("{} rows pending", stats.pending);
///     Ok(())
/// }
/// ```
pub trait OutboxStore: Send + Sync {
    /// A failure of the *call* — never a property of one row's content. Must self-classify via
    /// [`crate::Classify`] so the dispatcher's `run()` can tell a transient outage from a
    /// permanent one (ADR 0014).
    type Error: std::error::Error + Send + Sync + 'static + Classify;

    /// Claims up to `request.batch_size` due, unlocked, unexpired rows. **Must have committed
    /// before the future resolves** — the caller publishes outside any transaction (ADR 0006).
    ///
    /// **SHALL** stamp every row it claims with a [`ClaimToken`] minted by the **database** in
    /// the claim statement itself, and return it on each record. The token identifies the *claim
    /// of that row* — not the worker, not the lease interval, and not the batch: rows claimed by
    /// one call carry distinct tokens. Lease renewal **SHALL NOT** change it, and any write that
    /// ends the claim **SHALL** clear it (ADR 0046 Amendment A).
    ///
    /// Once the claim has committed, the rows it leased **SHALL** be returned to the caller. Any
    /// follow-up bookkeeping the provider performs on the same batch — moving an undecodable row
    /// to dead, for example — is **best-effort** (ADR 0039): its failure **SHALL NOT** turn a
    /// committed claim into an `Err`. The provider logs the failure; the affected rows keep
    /// their lease and are re-claimed after it lapses, when the bookkeeping is attempted again.
    /// `acquire` still returns `Err` when the **claim itself** fails, in which case nothing was
    /// committed and no row is leased.
    ///
    /// ```
    /// use reliar_outbox::{AcquireRequest, OutboxStore, WorkerId};
    ///
    /// async fn claim_a_batch<S: OutboxStore>(store: &S) -> Result<(), S::Error> {
    ///     let batch = store.acquire(AcquireRequest::new(WorkerId::generate())).await?;
    ///     println!("claimed {} rows", batch.records.len());
    ///     Ok(())
    /// }
    /// ```
    fn acquire(
        &self,
        request: AcquireRequest,
    ) -> impl Future<Output = Result<AcquiredBatch, Self::Error>> + Send;

    /// Marks rows published and increments `attempts`. Guarded by each item's
    /// [`RecordRef::claim_token`] (ADR 0046 Amendment A) — idempotent under it: a row already
    /// completed or reclaimed (by any worker, including this one, after the lease lapsed)
    /// contributes nothing to the count. `worker` is diagnostic only; it is not the guard.
    ///
    /// ```
    /// use reliar_outbox::{CompletedRecord, OutboxStore, RecordRef, WorkerId};
    ///
    /// async fn mark_complete<S: OutboxStore>(
    ///     store: &S,
    ///     worker: &WorkerId,
    ///     record: RecordRef,
    /// ) -> Result<(), S::Error> {
    ///     let affected = store.complete(worker, &[CompletedRecord::new(record)]).await?;
    ///     println!("{affected} row(s) marked complete");
    ///     Ok(())
    /// }
    /// ```
    fn complete(
        &self,
        worker: &WorkerId,
        items: &[CompletedRecord],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;

    /// Applies each item's [`FailureOutcome`] and increments `attempts`. Guarded by each item's
    /// [`RecordRef::claim_token`] (ADR 0046 Amendment A); `worker` is diagnostic only.
    ///
    /// ```
    /// use reliar_outbox::{FailedRecord, FailureOutcome, OutboxStore, RecordRef, WorkerId};
    ///
    /// async fn retry_after_a_second<S: OutboxStore>(
    ///     store: &S,
    ///     worker: &WorkerId,
    ///     record: RecordRef,
    /// ) -> Result<(), S::Error> {
    ///     let outcome = FailureOutcome::Retry { delay: std::time::Duration::from_secs(1) };
    ///     let failed = FailedRecord::new(record, "connection refused", outcome);
    ///     let affected = store.fail(worker, &[failed]).await?;
    ///     println!("{affected} row(s) scheduled for retry");
    ///     Ok(())
    /// }
    /// ```
    fn fail(
        &self,
        worker: &WorkerId,
        items: &[FailedRecord],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;

    /// Hands rows back at once: clears the lease, **resets `available_at` to now** so any
    /// worker may claim the row immediately, and leaves `attempts` unchanged. Used on graceful
    /// shutdown. A provider that leaves `available_at` at its leased value turns every graceful
    /// shutdown into a lease-length stall (ADR 0040). Guarded by each item's
    /// [`RecordRef::claim_token`] (ADR 0046 Amendment A); `worker` is diagnostic only.
    ///
    /// ```
    /// use reliar_outbox::{OutboxStore, RecordRef, WorkerId};
    ///
    /// async fn hand_back<S: OutboxStore>(
    ///     store: &S,
    ///     worker: &WorkerId,
    ///     record: RecordRef,
    /// ) -> Result<(), S::Error> {
    ///     let affected = store.release(worker, &[record]).await?;
    ///     println!("{affected} row(s) released");
    ///     Ok(())
    /// }
    /// ```
    fn release(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;

    /// Renews `locked_until = now() + lease` for rows this claim still owns. Best-effort: a
    /// shortfall means the claim was superseded — the lease already expired and lapsed to another
    /// claim, by any worker. Guarded by each item's [`RecordRef::claim_token`], which renewal
    /// **never rotates** — a renewal extends the same claim, it is not a new one (ADR 0046
    /// Amendment A). `worker` is diagnostic only.
    ///
    /// ```
    /// use reliar_outbox::{OutboxStore, RecordRef, WorkerId};
    /// use std::time::Duration;
    ///
    /// async fn renew<S: OutboxStore>(
    ///     store: &S,
    ///     worker: &WorkerId,
    ///     record: RecordRef,
    /// ) -> Result<(), S::Error> {
    ///     let affected = store
    ///         .extend_lease(worker, &[record], Duration::from_secs(30))
    ///         .await?;
    ///     println!("{affected} row(s) renewed");
    ///     Ok(())
    /// }
    /// ```
    fn extend_lease(
        &self,
        worker: &WorkerId,
        items: &[RecordRef],
        lease: Duration,
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;

    /// **One bounded pass**: deletes at most `request.batch_size` published rows and at most
    /// `request.batch_size` dead rows, and sweeps expired pending rows to dead
    /// ([`DeadReason::Expired`]). It **does not loop internally** — unbounded work inside a
    /// trait method has no cancellation point and no progress reporting. The **caller** repeats
    /// while `!report.is_complete(request.batch_size)`. Reliar starts no maintenance timer; the
    /// host calls this from its own periodic task.
    ///
    /// # Example
    ///
    /// The host-owned purge loop: run on an interval, respect cancellation, and keep calling
    /// `purge` while one pass is not yet complete.
    /// ```
    /// use reliar_outbox::{OutboxStore, PurgeRequest};
    /// use std::time::Duration;
    /// use tokio_util::sync::CancellationToken;
    ///
    /// // The host-owned purge loop: run on an interval, respect cancellation, and keep calling
    /// // `purge` while one pass is not yet complete. Storage-agnostic (`S: OutboxStore`) since
    /// // this crate constructs no store of its own (ADR 0043).
    /// async fn purge_loop<S: OutboxStore>(store: &S, cancel: CancellationToken) {
    ///     // A host that wants dead rows collected sets `dead_retention` explicitly — the
    ///     // default leaves it `None` and deletes zero dead rows (see `PurgeRequest`'s `# Warning`).
    ///     let request =
    ///         PurgeRequest::default().dead_retention(Some(Duration::from_secs(30 * 24 * 60 * 60)));
    ///     let mut interval = tokio::time::interval(Duration::from_secs(60));
    ///
    ///     loop {
    ///         tokio::select! {
    ///             () = cancel.cancelled() => return,
    ///             _ = interval.tick() => {
    ///                 // One bounded pass may not clear a whole backlog — keep calling until it does.
    ///                 loop {
    ///                     let Ok(report) = store.purge(request.clone()).await else { return };
    ///                     if report.is_complete(request.batch_size) {
    ///                         break;
    ///                     }
    ///                 }
    ///             }
    ///         }
    ///     }
    /// }
    /// ```
    fn purge(
        &self,
        request: PurgeRequest,
    ) -> impl Future<Output = Result<PurgeReport, Self::Error>> + Send;

    /// Feeds the outbox-lag and dead-count gauges. Polled by the dispatcher's own `run` loop
    /// every `DispatcherSettings::stats_interval`, never per batch; `Duration::ZERO` there
    /// disables the tick entirely, so a host that never calls this directly gets no gauges at
    /// all rather than a zero-valued one. Also `pub`, so a host may call it directly (e.g. for an
    /// admin endpoint) independently of the dispatcher.
    ///
    /// ```
    /// use reliar_outbox::OutboxStore;
    ///
    /// async fn report_pending<S: OutboxStore>(store: &S) -> Result<(), S::Error> {
    ///     let stats = store.stats().await?;
    ///     println!("{} rows pending", stats.pending);
    ///     Ok(())
    /// }
    /// ```
    fn stats(&self) -> impl Future<Output = Result<OutboxStats, Self::Error>> + Send;
}

/// The operator surface over dead rows. A separate small capability — the dispatcher never
/// calls it.
///
/// ```
/// use reliar_outbox::{DeadQuery, OutboxDeadLetters};
///
/// async fn list_first_page<S: OutboxDeadLetters>(store: &S) -> Result<(), S::Error> {
///     let page = store.list_dead(DeadQuery::default()).await?;
///     println!("{} dead row(s)", page.records.len());
///     Ok(())
/// }
/// ```
pub trait OutboxDeadLetters: Send + Sync {
    /// A failure of the *call*.
    type Error: std::error::Error + Send + Sync + 'static;

    /// **`ORDER BY dead_at ASC, id ASC` is normative, not an implementation detail.**
    /// [`DeadQuery::after`] is a keyset cursor over both columns. A `dead_at` later than the
    /// cursor puts a row after it even when the row's own id
    /// predates the current walk; unique `id` breaks ties. `message_type`, `tenant_id` and
    /// `dead_before` are filters, never part of the order.
    ///
    /// Returns a page, not a bare `Vec`: a dead row can itself be undecodable, and such a row
    /// is reported, never silently skipped (ADR 0023). The cursor is computed over every row
    /// scanned, including the poisoned ones — deriving it only from decoded rows would loop
    /// forever on a poisoned tail.
    ///
    /// ```
    /// use reliar_outbox::{DeadQuery, OutboxDeadLetters};
    ///
    /// async fn list_first_page<S: OutboxDeadLetters>(store: &S) -> Result<(), S::Error> {
    ///     let page = store.list_dead(DeadQuery::default()).await?;
    ///     println!("{} dead row(s)", page.records.len());
    ///     Ok(())
    /// }
    /// ```
    fn list_dead(
        &self,
        query: DeadQuery,
    ) -> impl Future<Output = Result<DeadLetterPage, Self::Error>> + Send;

    /// Returns dead rows to pending: clears `dead_at`/`dead_reason`, sets `available_at =
    /// now()`, resets `attempts` to 0, keeps `last_error` for audit. Affects only rows with
    /// `dead_at IS NOT NULL`.
    ///
    /// The **only** operation in the system that resets `attempts`, and always an explicit
    /// operator action. Not guarded by [`RecordRef::claim_token`] at all — a dead row holds no
    /// lease and no claim, so this operation ignores whatever the field carries (ADR 0046
    /// Amendment A). A conforming provider clears the row's own stored claim token
    /// unconditionally, even though this call never quotes one: a token surviving the
    /// resurrection would otherwise let a stale write from before the row died land on the row
    /// after `retry_dead` returned it to pending (ADR 0046 Correction B.1, row 7).
    ///
    /// ```
    /// use reliar_outbox::{OutboxDeadLetters, RecordRef};
    ///
    /// async fn un_deadletter<S: OutboxDeadLetters>(
    ///     store: &S,
    ///     dead_ref: RecordRef,
    /// ) -> Result<(), S::Error> {
    ///     let affected = store.retry_dead(&[dead_ref]).await?;
    ///     println!("{affected} row(s) returned to pending");
    ///     Ok(())
    /// }
    /// ```
    fn retry_dead(
        &self,
        refs: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;

    /// Deletes dead rows by reference, regardless of [`PurgeRequest::dead_retention`]. Not
    /// guarded by [`RecordRef::claim_token`] — a dead row holds no claim (ADR 0046 Amendment A).
    ///
    /// ```
    /// use reliar_outbox::{OutboxDeadLetters, RecordRef};
    ///
    /// async fn delete_dead<S: OutboxDeadLetters>(
    ///     store: &S,
    ///     dead_ref: RecordRef,
    /// ) -> Result<(), S::Error> {
    ///     let affected = store.purge_dead(&[dead_ref]).await?;
    ///     println!("{affected} dead row(s) deleted");
    ///     Ok(())
    /// }
    /// ```
    fn purge_dead(
        &self,
        refs: &[RecordRef],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;
}