reliar-outbox 0.6.0

//! The outbox row: [`OutboxRecord`] and its builder.

use reliar_core::SerializedEnvelope;

use crate::store::{DeadReason, MessageRef};
use crate::worker::WorkerId;

/// The maximum length, in bytes, [`OutboxRecord::last_error`] and [`crate::PoisonedRow::error`]
/// are truncated to before being persisted.
const MAX_ERROR_LEN: usize = 2048;
const TRUNCATION_MARKER: &str = "…[truncated]";

/// Truncates `error` to [`MAX_ERROR_LEN`] bytes at a char boundary, appending
/// [`TRUNCATION_MARKER`] when it was cut. Shared by [`OutboxRecordBuilder::last_error`] and
/// [`crate::PoisonedRow::new`] so both truncate identically.
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;
    }

    tracing::debug!(
        original_len = error.len(),
        truncated_len = end,
        "outbox error truncated before persisting"
    );

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

    truncated
}

/// An envelope **plus** its outbound delivery state. Distinct from [`reliar_core::Envelope`]:
/// nothing here reaches the wire.
///
/// `Clone` and `PartialEq` are required: the `test-support` fakes hand records out by value and
/// the acceptance tests compare them. `Debug` is derived and is payload-safe — it delegates to
/// `Envelope`'s manual `Debug`, which elides the body for every `T`; `last_error` is already
/// truncated and redacted, so it is safe to print in full.
///
/// ```
/// use bytes::Bytes;
/// use reliar_core::{Envelope, Message};
/// use reliar_outbox::OutboxRecord;
/// use time::OffsetDateTime;
///
/// # #[derive(serde::Serialize, serde::Deserialize)]
/// # struct Ping;
/// # impl Message for Ping {
/// #     const TYPE: &'static str = "ping";
/// #     const VERSION: u16 = 1;
/// # }
/// let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
/// let now = OffsetDateTime::now_utc();
/// let record = OutboxRecord::builder(envelope, 1, now).build();
/// assert_eq!(record.attempts, 0);
/// assert_eq!(record.available_at, now);
/// assert!(record.locked_by.is_none());
/// ```
#[derive(Clone, Debug, PartialEq)]
#[non_exhaustive]
pub struct OutboxRecord {
    /// The envelope this row carries.
    pub envelope: SerializedEnvelope,

    /// Monotonic, store-assigned. Not gap-free.
    pub sequence: i64,

    /// Immutable once written (ADR 0016).
    pub created_at: time::OffsetDateTime,

    /// `None` means unordered.
    pub ordering_key: Option<String>,

    /// Publish outcomes observed, **not** claims (ADR 0009). A claimed-then-crashed row
    /// reports 0 here even though it was claimed once.
    pub attempts: u32,

    /// The time before which this row is not claimable. On a record just returned by
    /// [`crate::OutboxStore::acquire`] this is the **lease end**, not the enqueue time — a claim
    /// (and a lease renewal) moves it forward to `now() + lease` so the row is invisible to the
    /// claim scan for exactly its lease (ADR 0040). Use [`Self::created_at`] for the enqueue
    /// time.
    pub available_at: time::OffsetDateTime,

    /// The worker currently holding this row's lease, if any.
    pub locked_by: Option<WorkerId>,

    /// When the current lease expires, if any.
    pub locked_until: Option<time::OffsetDateTime>,

    /// When this row was marked published, if it was.
    pub published_at: Option<time::OffsetDateTime>,

    /// When this row was marked dead, if it was.
    pub dead_at: Option<time::OffsetDateTime>,

    /// Why this row is dead, if it is.
    pub dead_reason: Option<DeadReason>,

    /// The last failure's `Display` output, truncated to 2 KiB at a char boundary with a
    /// `"…[truncated]"` marker. Never payload bytes, header values, or credentials.
    pub last_error: Option<String>,
}

impl OutboxRecord {
    /// The value the dead-letter API and by-id operations take.
    ///
    /// ```
    /// use bytes::Bytes;
    /// use reliar_core::{Envelope, Message};
    /// use reliar_outbox::OutboxRecord;
    /// use time::OffsetDateTime;
    ///
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping {
    /// #     const TYPE: &'static str = "ping";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let id = envelope.id;
    /// let now = OffsetDateTime::now_utc();
    /// let record = OutboxRecord::builder(envelope, 1, now).build();
    /// let message_ref = record.message_ref();
    /// assert_eq!(message_ref.id, id);
    /// assert_eq!(message_ref.created_at, now);
    /// ```
    #[must_use]
    pub fn message_ref(&self) -> MessageRef {
        MessageRef::new(self.envelope.id, self.created_at)
    }

    /// **Provider entry point.** `OutboxRecord` is `#[non_exhaustive]`, so a crate other than
    /// `reliar-outbox` cannot build one with struct-literal syntax — every provider rehydrating
    /// a row goes through this builder.
    ///
    /// Defaults: `attempts = 0`, `available_at = created_at`, no lease, not published, not dead,
    /// no error, no ordering key.
    ///
    /// ```
    /// use bytes::Bytes;
    /// use reliar_core::{Envelope, Message};
    /// use reliar_outbox::OutboxRecord;
    /// use time::OffsetDateTime;
    ///
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping {
    /// #     const TYPE: &'static str = "ping";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(envelope, 1, OffsetDateTime::now_utc()).build();
    /// assert_eq!(record.sequence, 1);
    /// assert_eq!(record.attempts, 0);
    /// ```
    pub fn builder(
        envelope: SerializedEnvelope,
        sequence: i64,
        created_at: time::OffsetDateTime,
    ) -> OutboxRecordBuilder {
        OutboxRecordBuilder::new(envelope, sequence, created_at)
    }
}

/// Builds an [`OutboxRecord`]. Obtained from [`OutboxRecord::builder`].
///
/// ```
/// use bytes::Bytes;
/// use reliar_core::{Envelope, Message};
/// use reliar_outbox::OutboxRecord;
/// # #[derive(serde::Serialize, serde::Deserialize)]
/// # struct Ping;
/// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
/// let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
/// let builder = OutboxRecord::builder(envelope, 1, time::OffsetDateTime::now_utc());
/// assert_eq!(builder.build().sequence, 1);
/// ```
#[must_use]
#[derive(Debug)]
pub struct OutboxRecordBuilder {
    envelope: SerializedEnvelope,

    sequence: i64,

    created_at: time::OffsetDateTime,

    ordering_key: Option<String>,

    attempts: u32,

    available_at: time::OffsetDateTime,

    locked_by: Option<WorkerId>,

    locked_until: Option<time::OffsetDateTime>,

    published_at: Option<time::OffsetDateTime>,

    dead_at: Option<time::OffsetDateTime>,

    dead_reason: Option<DeadReason>,

    last_error: Option<String>,
}

impl OutboxRecordBuilder {
    fn new(envelope: SerializedEnvelope, sequence: i64, created_at: time::OffsetDateTime) -> Self {
        Self {
            envelope,
            sequence,
            available_at: created_at,
            created_at,
            ordering_key: None,
            attempts: 0,
            locked_by: None,
            locked_until: None,
            published_at: None,
            dead_at: None,
            dead_reason: None,
            last_error: None,
        }
    }

    /// Sets [`OutboxRecord::ordering_key`].
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::OutboxRecord;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(envelope, 1, time::OffsetDateTime::now_utc())
    ///     .ordering_key(Some("order-1".to_string()))
    ///     .build();
    /// assert_eq!(record.ordering_key.as_deref(), Some("order-1"));
    /// ```
    pub fn ordering_key(mut self, key: Option<String>) -> Self {
        self.ordering_key = key;

        self
    }

    /// Sets [`OutboxRecord::attempts`].
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::OutboxRecord;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(envelope, 1, 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 [`OutboxRecord::available_at`].
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::OutboxRecord;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let created_at = time::OffsetDateTime::now_utc();
    /// let available_at = created_at + time::Duration::seconds(30);
    /// let record = OutboxRecord::builder(envelope, 1, created_at)
    ///     .available_at(available_at)
    ///     .build();
    /// assert_eq!(record.available_at, available_at);
    /// ```
    pub const fn available_at(mut self, at: time::OffsetDateTime) -> Self {
        self.available_at = at;

        self
    }

    /// Sets [`OutboxRecord::locked_by`] and [`OutboxRecord::locked_until`] together — a lease is
    /// either held by a worker until a time, or not held at all.
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::{OutboxRecord, WorkerId};
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let worker = WorkerId::parse("worker-1").unwrap();
    /// let until = time::OffsetDateTime::now_utc() + time::Duration::seconds(30);
    /// let record = OutboxRecord::builder(envelope, 1, time::OffsetDateTime::now_utc())
    ///     .lease(Some(worker.clone()), Some(until))
    ///     .build();
    /// assert_eq!(record.locked_by, Some(worker));
    /// assert_eq!(record.locked_until, Some(until));
    /// ```
    pub fn lease(mut self, by: Option<WorkerId>, until: Option<time::OffsetDateTime>) -> Self {
        self.locked_by = by;
        self.locked_until = until;

        self
    }

    /// Sets [`OutboxRecord::published_at`].
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::OutboxRecord;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let now = time::OffsetDateTime::now_utc();
    /// let record = OutboxRecord::builder(envelope, 1, now)
    ///     .published_at(Some(now))
    ///     .build();
    /// assert_eq!(record.published_at, Some(now));
    /// ```
    pub const fn published_at(mut self, at: Option<time::OffsetDateTime>) -> Self {
        self.published_at = at;

        self
    }

    /// Sets [`OutboxRecord::dead_at`] and [`OutboxRecord::dead_reason`] together.
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::{DeadReason, OutboxRecord};
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let now = time::OffsetDateTime::now_utc();
    /// let record = OutboxRecord::builder(envelope, 1, now)
    ///     .dead(Some(now), Some(DeadReason::AttemptsExhausted))
    ///     .build();
    /// assert_eq!(record.dead_reason, Some(DeadReason::AttemptsExhausted));
    /// ```
    pub const fn dead(
        mut self,
        at: Option<time::OffsetDateTime>,
        reason: Option<DeadReason>,
    ) -> Self {
        self.dead_at = at;
        self.dead_reason = reason;

        self
    }

    /// Sets [`OutboxRecord::last_error`], truncating to 2 KiB at a char boundary with a
    /// `"…[truncated]"` marker.
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::OutboxRecord;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(envelope, 1, 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 bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::OutboxRecord;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(envelope, 1, time::OffsetDateTime::now_utc()).build();
    /// assert!(record.published_at.is_none());
    /// ```
    #[must_use]
    pub fn build(self) -> OutboxRecord {
        OutboxRecord {
            envelope: self.envelope,
            sequence: self.sequence,
            created_at: self.created_at,
            ordering_key: self.ordering_key,
            attempts: self.attempts,
            available_at: self.available_at,
            locked_by: self.locked_by,
            locked_until: self.locked_until,
            published_at: self.published_at,
            dead_at: self.dead_at,
            dead_reason: self.dead_reason,
            last_error: self.last_error,
        }
    }
}