reliar-inbox 0.2.0

//! [`InboxDeadLetters`]: the operator surface over dead rows (ADR 0042 A.2.5).

use time::OffsetDateTime;

use crate::record::InboxRecord;
use crate::record_id::InboxRecordId;
use crate::scope::InboxScope;

/// Keyset cursor for [`InboxDeadLetters::list_dead`].
///
/// The pair is opaque so callers cannot accidentally advance only one half of the ordering key.
/// Build it from the last row in a page with [`Self::from_record`], or reconstruct a persisted
/// cursor with [`Self::new`].
#[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 InboxDeadCursor {
    #[cfg_attr(feature = "serde", serde(with = "time::serde::rfc3339"))]
    dead_at: OffsetDateTime,

    id: InboxRecordId,
}

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

    /// Builds the cursor that follows `record`, or `None` when the record is not dead.
    #[must_use]
    pub fn from_record(record: &InboxRecord) -> Option<Self> {
        record.dead_at.map(|dead_at| Self::new(dead_at, record.id))
    }

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

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

/// Reads and rewinds the dead rows [`crate::InboxStore::fail`] created when a message exhausted
/// `max_attempts`. On the provider's own pool; no method here touches the caller's transaction.
///
/// Separate from [`crate::InboxStore`] for the same reason `reliar-outbox`'s
/// `OutboxDeadLetters` is separate from `OutboxStore`: no Reliar code path calls it, and a host
/// that only consumes messages should not have to implement it.
pub trait InboxDeadLetters: 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.**
    /// [`InboxDeadQuery::after`] is a keyset cursor over both columns. Database-authored
    /// A `dead_at` later than the cursor puts a row after it even when the row's client-minted id
    /// predates the current walk; unique `id` breaks ties. `scope`, `message_type` and
    /// `dead_before` are filters, never part of the order.
    ///
    /// Returns a bare `Vec`, not a page type: the inbox stores **no payload** and constrains
    /// every column, so no row can be undecodable. Build the next cursor with
    /// `records.last().and_then(InboxDeadCursor::from_record)`.
    ///
    /// # Errors
    ///
    /// Provider-defined.
    fn list_dead(
        &self,
        query: InboxDeadQuery,
    ) -> impl Future<Output = Result<Vec<InboxRecord>, Self::Error>> + Send;

    /// Clears `dead_at`, resets `attempts` to `0`, keeps `last_error` for audit, sets
    /// `updated_at = now()`. Affects only rows with `dead_at IS NOT NULL`. Returns the number of
    /// rows affected.
    ///
    /// # This does not redeliver anything
    ///
    /// The inbox has no `available_at` and no worker: **nothing re-runs**. This makes the row
    /// *claimable again*, so that a redelivery the operator arranges separately — replaying the
    /// stream, re-publishing from the producer — is handled instead of terminated. An operator
    /// who expects the handler to run will watch nothing happen.
    ///
    /// `attempts` is reset for the same reason the outbox resets it on its own `retry_dead`:
    /// otherwise the next failure immediately re-deads the row.
    ///
    /// # Errors
    ///
    /// Provider-defined.
    fn retry_dead(
        &self,
        ids: &[InboxRecordId],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;

    /// Deletes dead rows by id, regardless of retention. Returns the number of rows affected.
    ///
    /// # Errors
    ///
    /// Provider-defined.
    fn purge_dead(
        &self,
        ids: &[InboxRecordId],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send;
}

/// A filtered, paginated query over dead rows.
///
/// ```
/// use reliar_inbox::InboxDeadQuery;
///
/// let query = InboxDeadQuery::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 InboxDeadQuery {
    /// Restricts to one consumer scope, if set.
    pub scope: Option<InboxScope>,

    /// Restricts to one message type **name** (every version), if set.
    pub message_type: Option<String>,

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

    /// Maximum rows to return. Provider-capped. Default 100.
    pub limit: u32,

    /// Keyset cursor: only rows ordered after this `(dead_at, id)` pair.
    pub after: Option<InboxDeadCursor>,
}

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

impl InboxDeadQuery {
    /// Sets [`Self::scope`].
    ///
    /// ```
    /// use reliar_inbox::{InboxDeadQuery, InboxScope};
    /// let scope = InboxScope::new("orders-projection").unwrap();
    /// let query = InboxDeadQuery::default().scope(scope.clone());
    /// assert_eq!(query.scope, Some(scope));
    /// ```
    #[must_use]
    pub fn scope(mut self, scope: InboxScope) -> Self {
        self.scope = Some(scope);

        self
    }

    /// Sets [`Self::message_type`].
    ///
    /// ```
    /// use reliar_inbox::InboxDeadQuery;
    /// let query = InboxDeadQuery::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::dead_before`].
    ///
    /// ```
    /// use reliar_inbox::InboxDeadQuery;
    /// use time::OffsetDateTime;
    /// let before = OffsetDateTime::now_utc();
    /// let query = InboxDeadQuery::default().dead_before(before);
    /// assert_eq!(query.dead_before, Some(before));
    /// ```
    #[must_use]
    pub const fn dead_before(mut self, dead_before: OffsetDateTime) -> Self {
        self.dead_before = Some(dead_before);

        self
    }

    /// Sets [`Self::limit`].
    ///
    /// ```
    /// use reliar_inbox::InboxDeadQuery;
    /// assert_eq!(InboxDeadQuery::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_inbox::{InboxDeadCursor, InboxDeadQuery, InboxRecordId};
    /// use time::OffsetDateTime;
    /// let cursor = InboxDeadCursor::new(OffsetDateTime::now_utc(), InboxRecordId::new());
    /// let query = InboxDeadQuery::default().after(cursor);
    /// assert_eq!(query.after, Some(cursor));
    /// ```
    #[must_use]
    pub const fn after(mut self, after: InboxDeadCursor) -> Self {
        self.after = Some(after);

        self
    }
}