reliar-inbox 0.1.0

//! The inbox row: [`InboxRecord`] and its builder, plus the derived [`InboxState`] (ADR 0042
//! Amendment B).

use core::fmt;

use reliar_core::{ConversationId, CorrelationId, MessageId, MessageType};
use time::OffsetDateTime;

use crate::message::InboxMessage;
use crate::record_id::InboxRecordId;
use crate::scope::InboxScope;

/// The maximum length, in bytes, [`InboxRecord::last_error`] is truncated to before being
/// persisted — the same bound `reliar-outbox`'s `OutboxRecord::last_error` uses.
pub(crate) const MAX_ERROR_LEN: usize = 2048;
pub(crate) const TRUNCATION_MARKER: &str = "…[truncated]";

/// Truncates `error` to [`MAX_ERROR_LEN`] bytes at a char boundary, appending
/// [`TRUNCATION_MARKER`] when it was cut.
pub(crate) fn truncate_error(error: impl Into<String>) -> String {
    let error = error.into();

    if error.len() <= MAX_ERROR_LEN {
        return error;
    }

    let budget = MAX_ERROR_LEN.saturating_sub(TRUNCATION_MARKER.len());
    let mut end = budget.min(error.len());

    while end > 0 && !error.is_char_boundary(end) {
        end -= 1;
    }

    let mut truncated = String::with_capacity(end + TRUNCATION_MARKER.len());
    truncated.push_str(&error[..end]);
    truncated.push_str(TRUNCATION_MARKER);

    truncated
}

/// Where a row stands, derived from its timestamps. There is deliberately **no status column**:
/// a derived state cannot disagree with the columns the claim path actually reads. Read it
/// through [`InboxRecord::state`]; the operator-facing SQL that computes the same thing is
/// documented once in `docs/guides/inbox.md`.
///
/// ```
/// use reliar_core::MessageId;
/// use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope, InboxState};
/// use reliar_core::MessageType;
///
/// let message_type = MessageType::new("orders.created", 1);
/// let message = InboxMessage::new(MessageId::new(), &message_type);
/// let scope = InboxScope::new("orders-projection").unwrap();
/// let now = time::OffsetDateTime::now_utc();
///
/// let claimed = InboxRecord::builder(InboxRecordId::new(), scope, message, now).build();
/// assert_eq!(claimed.state(), InboxState::Claimed);
/// ```
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum InboxState {
    /// A claim row committed, but `complete` was never called and no failure was recorded —
    /// the commit-without-`complete` anomaly, or a row [`crate::InboxDeadLetters::retry_dead`]
    /// just un-deaded — either way the next redelivery answers `Claimed { attempt: 1 }` and
    /// re-runs the handler. `incomplete_retention` collects this state exactly as it collects
    /// [`Self::Retrying`] (ADR 0042 Amendment C.2): both age by `InboxRecord::updated_at` alone.
    Claimed,

    /// Uncompleted, not dead, with at least one recorded failure — a redelivery will run the
    /// handler again.
    Retrying,

    /// `completed_at` is set. Terminal.
    Completed,

    /// `dead_at` is set. Terminal until [`crate::InboxDeadLetters::retry_dead`].
    Dead,
}

impl fmt::Display for InboxState {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::Claimed => "claimed",
            Self::Retrying => "retrying",
            Self::Completed => "completed",
            Self::Dead => "dead",
        })
    }
}

/// A row as stored, keyed `(scope, message_id)` and identified by [`Self::id`]. Diagnostics and
/// tests only — no Reliar code path reads it.
///
/// There is deliberately **no payload** — the inbox deduplicates, it does not archive (ADR 0042
/// §6). `request_id` is deliberately **not** copied (ADR 0042 Amendment A named four fields), and no
/// `dead_reason` column exists because the inbox has exactly one way to die.
///
/// `#[non_exhaustive]`: build one with [`Self::builder`] — a provider's `find` uses it too, so a
/// struct literal is never needed outside this crate either.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct InboxRecord {
    /// The surrogate row id — `pk_inbox`, and the handle every operator call takes.
    pub id: InboxRecordId,

    /// The consumer whose progress this row records.
    pub scope: InboxScope,

    /// The message's own id, as produced by the sender.
    pub message_id: MessageId,

    /// Rehydrated from the `message_type` + `message_version` columns.
    pub message_type: MessageType,

    /// `NOT NULL` in the row; [`ConversationId::UNSET`] when the sender rooted none.
    pub conversation_id: ConversationId,

    /// The caller's business correlation, if the sender set one.
    pub correlation_id: Option<CorrelationId>,

    /// The inbound envelope's own causation, **not** this row's `message_id`.
    pub causation_id: Option<MessageId>,

    /// When the row was first claimed.
    pub received_at: OffsetDateTime,

    /// The row's last state transition, in **database** time: written by `claim`'s insert, moved
    /// by `complete`, `fail` and [`crate::InboxDeadLetters::retry_dead`]. This is the column
    /// [`crate::InboxPurgeRequest::incomplete_retention`] ages a row by (ADR 0042 Amendment C.3).
    pub updated_at: OffsetDateTime,

    /// When the row was completed, if it was.
    pub completed_at: Option<OffsetDateTime>,

    /// Set once `attempts` reached the provider's `max_attempts`. Mutually exclusive with
    /// `completed_at` (`ck_inbox_terminal`).
    pub dead_at: Option<OffsetDateTime>,

    /// Recorded failed attempts (see [`crate::InboxClaim::Claimed::attempt`]). A lower bound,
    /// not a guarantee — see [`crate::InboxStore::fail`].
    pub attempts: u32,

    /// `Display` of the last failure's error chain, truncated to 2 KiB at a char boundary with
    /// a `"…[truncated]"` marker. Never payload bytes, header values or credentials. Survives
    /// both `complete` and [`crate::InboxDeadLetters::retry_dead`], for audit.
    pub last_error: Option<String>,
}

impl InboxRecord {
    /// Starts a builder for an inbox row, seeded from the same [`InboxMessage`] view
    /// [`crate::InboxStore::claim`]/[`crate::InboxStore::fail`] take, so the record's identity
    /// and trace columns are one vocabulary rather than positional arguments.
    ///
    /// Defaults: `completed_at = None`, `dead_at = None`, `attempts = 0`, `last_error = None` —
    /// a freshly claimed, not-yet-completed row.
    ///
    /// ```
    /// use reliar_core::{MessageId, MessageType};
    /// use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope};
    ///
    /// let scope = InboxScope::new("orders-projection").unwrap();
    /// let message_type = MessageType::new("orders.created", 1);
    /// let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let record = InboxRecord::builder(
    ///     InboxRecordId::new(),
    ///     scope,
    ///     message,
    ///     time::OffsetDateTime::now_utc(),
    /// )
    /// .build();
    ///
    /// assert_eq!(record.attempts, 0);
    /// assert!(record.completed_at.is_none());
    /// ```
    pub fn builder(
        id: InboxRecordId,
        scope: InboxScope,
        message: InboxMessage<'_>,
        received_at: OffsetDateTime,
    ) -> InboxRecordBuilder {
        InboxRecordBuilder::new(id, scope, message, received_at)
    }

    /// This row's derived [`InboxState`] (ADR 0042 Amendment B). Precedence is `claim`'s step 3:
    /// completed, then dead, then the attempt count; `ck_inbox_terminal` makes the first two
    /// mutually exclusive in a real row, so the order documents rather than arbitrates.
    ///
    /// **No `now` parameter, unlike `OutboxRecord::state(now)`.** Nothing on an inbox row is
    /// time-dependent — no lease, no `available_at`, no expiry — so a clock argument would be
    /// symmetry for its own sake. `const fn`, for the same reason.
    ///
    /// ```
    /// use reliar_core::{MessageId, MessageType};
    /// use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope, InboxState};
    ///
    /// let scope = InboxScope::new("orders-projection").unwrap();
    /// let message_type = MessageType::new("orders.created", 1);
    /// let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let now = time::OffsetDateTime::now_utc();
    ///
    /// let retrying = InboxRecord::builder(InboxRecordId::new(), scope, message, now)
    ///     .attempts(1)
    ///     .build();
    /// assert_eq!(retrying.state(), InboxState::Retrying);
    /// ```
    #[must_use]
    pub const fn state(&self) -> InboxState {
        if self.completed_at.is_some() {
            InboxState::Completed
        } else if self.dead_at.is_some() {
            InboxState::Dead
        } else if self.attempts > 0 {
            InboxState::Retrying
        } else {
            InboxState::Claimed
        }
    }
}

/// Builds an [`InboxRecord`]. Obtained from [`InboxRecord::builder`].
#[must_use]
#[derive(Debug)]
pub struct InboxRecordBuilder {
    id: InboxRecordId,

    scope: InboxScope,

    message_id: MessageId,

    message_type: MessageType,

    conversation_id: ConversationId,

    correlation_id: Option<CorrelationId>,

    causation_id: Option<MessageId>,

    received_at: OffsetDateTime,

    updated_at: OffsetDateTime,

    completed_at: Option<OffsetDateTime>,

    dead_at: Option<OffsetDateTime>,

    attempts: u32,

    last_error: Option<String>,
}

impl InboxRecordBuilder {
    fn new(
        id: InboxRecordId,
        scope: InboxScope,
        message: InboxMessage<'_>,
        received_at: OffsetDateTime,
    ) -> Self {
        Self {
            id,
            scope,
            message_id: message.id,
            message_type: message.message_type.clone(),
            conversation_id: message.conversation_id,
            correlation_id: message.correlation_id.cloned(),
            causation_id: message.causation_id,
            received_at,
            // A row that has not transitioned since its insert — the truthful default for the
            // fresh-claim shape this builder's four required arguments describe (ADR 0042
            // Amendment C.3; not a fifth required argument, matching A.2.1's fixed builder head).
            updated_at: received_at,
            completed_at: None,
            dead_at: None,
            attempts: 0,
            last_error: None,
        }
    }

    /// Sets [`InboxRecord::updated_at`], overriding the default of `received_at`.
    ///
    /// ```
    /// # use reliar_core::{MessageId, MessageType};
    /// # use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope};
    /// # let scope = InboxScope::new("orders-projection").unwrap();
    /// # let message_type = MessageType::new("orders.created", 1);
    /// # let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let received_at = time::OffsetDateTime::now_utc();
    /// let updated_at = received_at + time::Duration::seconds(5);
    /// let record = InboxRecord::builder(InboxRecordId::new(), scope, message, received_at)
    ///     .updated_at(updated_at)
    ///     .build();
    /// assert_eq!(record.updated_at, updated_at);
    /// ```
    pub const fn updated_at(mut self, updated_at: OffsetDateTime) -> Self {
        self.updated_at = updated_at;

        self
    }

    /// Sets [`InboxRecord::completed_at`].
    ///
    /// ```
    /// # use reliar_core::{MessageId, MessageType};
    /// # use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope};
    /// # let scope = InboxScope::new("orders-projection").unwrap();
    /// # let message_type = MessageType::new("orders.created", 1);
    /// # let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let now = time::OffsetDateTime::now_utc();
    /// let record = InboxRecord::builder(InboxRecordId::new(), scope, message, now)
    ///     .completed_at(Some(now))
    ///     .build();
    /// assert_eq!(record.completed_at, Some(now));
    /// ```
    pub const fn completed_at(mut self, completed_at: Option<OffsetDateTime>) -> Self {
        self.completed_at = completed_at;

        self
    }

    /// Sets [`InboxRecord::dead_at`].
    ///
    /// ```
    /// # use reliar_core::{MessageId, MessageType};
    /// # use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope};
    /// # let scope = InboxScope::new("orders-projection").unwrap();
    /// # let message_type = MessageType::new("orders.created", 1);
    /// # let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let now = time::OffsetDateTime::now_utc();
    /// let record = InboxRecord::builder(InboxRecordId::new(), scope, message, now)
    ///     .dead_at(Some(now))
    ///     .build();
    /// assert_eq!(record.dead_at, Some(now));
    /// ```
    pub const fn dead_at(mut self, dead_at: Option<OffsetDateTime>) -> Self {
        self.dead_at = dead_at;

        self
    }

    /// Sets [`InboxRecord::attempts`].
    ///
    /// ```
    /// # use reliar_core::{MessageId, MessageType};
    /// # use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope};
    /// # let scope = InboxScope::new("orders-projection").unwrap();
    /// # let message_type = MessageType::new("orders.created", 1);
    /// # let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let record = InboxRecord::builder(InboxRecordId::new(), scope, message, time::OffsetDateTime::now_utc())
    ///     .attempts(2)
    ///     .build();
    /// assert_eq!(record.attempts, 2);
    /// ```
    pub const fn attempts(mut self, attempts: u32) -> Self {
        self.attempts = attempts;

        self
    }

    /// Sets [`InboxRecord::last_error`], truncating to 2 KiB at a char boundary with a
    /// `"…[truncated]"` marker — so a value read back from a provider that stored it unbounded
    /// is still bounded here.
    ///
    /// ```
    /// # use reliar_core::{MessageId, MessageType};
    /// # use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope};
    /// # let scope = InboxScope::new("orders-projection").unwrap();
    /// # let message_type = MessageType::new("orders.created", 1);
    /// # let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let record = InboxRecord::builder(InboxRecordId::new(), scope, message, time::OffsetDateTime::now_utc())
    ///     .last_error(Some("connection refused".to_string()))
    ///     .build();
    /// assert_eq!(record.last_error.as_deref(), Some("connection refused"));
    /// ```
    pub fn last_error(mut self, error: Option<String>) -> Self {
        self.last_error = error.map(truncate_error);

        self
    }

    /// Builds the record.
    ///
    /// ```
    /// # use reliar_core::{MessageId, MessageType};
    /// # use reliar_inbox::{InboxMessage, InboxRecord, InboxRecordId, InboxScope};
    /// # let scope = InboxScope::new("orders-projection").unwrap();
    /// # let message_type = MessageType::new("orders.created", 1);
    /// # let message = InboxMessage::new(MessageId::new(), &message_type);
    /// let record = InboxRecord::builder(InboxRecordId::new(), scope, message, time::OffsetDateTime::now_utc())
    ///     .build();
    /// assert!(record.last_error.is_none());
    /// ```
    #[must_use]
    pub fn build(self) -> InboxRecord {
        InboxRecord {
            id: self.id,
            scope: self.scope,
            message_id: self.message_id,
            message_type: self.message_type,
            conversation_id: self.conversation_id,
            correlation_id: self.correlation_id,
            causation_id: self.causation_id,
            received_at: self.received_at,
            updated_at: self.updated_at,
            completed_at: self.completed_at,
            dead_at: self.dead_at,
            attempts: self.attempts,
            last_error: self.last_error,
        }
    }
}