reliar-outbox 0.9.0

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

use core::fmt;

use reliar_core::SerializedEnvelope;

use crate::claim_token::ClaimToken;
use crate::record_id::OutboxRecordId;
use crate::store::{DeadReason, RecordRef};
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
}

/// A row's lifecycle position, **derived** from [`OutboxRecord`]'s timestamps — there is no
/// physical `status` column to drift out of sync with them (ADR 0042 Amendment B). Read it through
/// [`OutboxRecord::state`]; nothing constructs it directly, since a `state` field on the row
/// itself is exactly what this type replaces. The operator-facing SQL that computes the same
/// thing is documented once in `docs/guides/postgres.md` under "Reading state".
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum OutboxState {
    /// Not leased, not published, not dead. **Wider than claimable**: a `Pending` row is
    /// claimable only once `available_at` has passed **and** `expires_at` has not — an expired
    /// pending row reports `Pending` here, is never claimed, and is swept to [`Self::Dead`] by
    /// `purge` (ADR 0050 §3).
    Pending,

    /// Currently held by a worker's lease — `locked_by` is set and `available_at` (the lease
    /// clock, ADR 0040 §1, ADR 0050 §1) has not yet passed.
    Leased,

    /// `published_at` is set. Terminal.
    Published,

    /// `dead_at` is set. Terminal.
    Dead,
}

impl fmt::Display for OutboxState {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(match self {
            Self::Pending => "pending",
            Self::Leased => "leased",
            Self::Published => "published",
            Self::Dead => "dead",
        })
    }
}

/// An envelope **plus** its outbound delivery state. Distinct from [`reliar_core::Envelope`]:
/// nothing here reaches the wire.
///
/// `Clone` and `PartialEq` are required: a provider store hands records out by value and the
/// acceptance tests compare them against real Postgres rows (ADR 0043). `Debug` is derived and is
/// payload-safe — it delegates to
/// `Envelope`'s manual `Debug`, which elides the body for every `T`. `last_error` is not
/// redacted: it is the classified [`reliar_core::Publisher::Error`]'s `Display` text, truncated
/// to 2 KiB. This crate does not scrub it — every publisher's error `Display` must never carry
/// payload bytes or header values, so there is nothing left to redact once that convention is
/// followed.
///
/// ```
/// use bytes::Bytes;
/// use reliar_core::{Envelope, Message};
/// use reliar_outbox::{OutboxRecord, OutboxRecordId};
/// use time::OffsetDateTime;
/// use reliar_core::uuid::Uuid;
///
/// # #[derive(serde::Serialize, serde::Deserialize)]
/// # struct Ping;
/// # impl Message for Ping {
/// #     const TYPE: &'static str = "ping";
/// #     const VERSION: u16 = 1;
/// # }
/// let id = OutboxRecordId::from_uuid(Uuid::now_v7());
/// let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
/// let now = OffsetDateTime::now_utc();
/// let record = OutboxRecord::builder(id, envelope, now).build();
/// assert_eq!(record.id, id);
/// 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 {
    /// This row's own identity — `pk_outbox`. Distinct from [`Self::envelope`]'s
    /// [`reliar_core::MessageId`] (ADR 0044 §1).
    pub id: OutboxRecordId,

    /// The envelope this row carries.
    pub envelope: SerializedEnvelope,

    /// 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 — a diagnostic for outcome writes
    /// (never a fence, [`Self::claim_token`] is, ADR 0046 Amendment A); read by the expiry sweep
    /// (ADR 0050 §2.3) and by [`Self::state`].
    pub locked_by: Option<WorkerId>,

    /// The token of the claim currently holding this row's lease, if any — what
    /// [`Self::record_ref`] carries forward for a claim-scoped outcome write to quote back
    /// (ADR 0046 Amendment A). `None` on a row with no live claim (e.g. read through
    /// [`crate::OutboxDeadLetters::list_dead`]).
    pub claim_token: Option<ClaimToken>,

    /// 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. Not scrubbed by this crate: every publisher's error `Display`
    /// must never carry payload bytes, header values, or credentials, so a compliant
    /// [`reliar_core::Publisher::Error`] means this field never does either.
    pub last_error: Option<String>,
}

impl OutboxRecord {
    /// The value the dead-letter API and by-id operations take. Built from [`Self::id`], **not**
    /// [`Self::envelope`]'s `MessageId` — a by-row operation identifies the row, not the message
    /// (ADR 0044 §3).
    ///
    /// ```
    /// use bytes::Bytes;
    /// use reliar_core::{Envelope, Message};
    /// use reliar_outbox::{OutboxRecord, OutboxRecordId};
    /// use time::OffsetDateTime;
    /// use reliar_core::uuid::Uuid;
    ///
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping {
    /// #     const TYPE: &'static str = "ping";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let now = OffsetDateTime::now_utc();
    /// let record = OutboxRecord::builder(id, envelope, now).build();
    /// let record_ref = record.record_ref();
    /// assert_eq!(record_ref.id, id);
    /// assert_eq!(record_ref.created_at, now);
    /// ```
    ///
    /// Propagates [`Self::claim_token`] (ADR 0046 Amendment A): a record freshly claimed by
    /// [`crate::OutboxStore::acquire`] carries `Some`, so the reference is claim-scoped and a
    /// claim-scoped outcome write (`complete`/`fail`/`release`/`extend_lease`) applies; a record
    /// with no live claim (e.g. from [`crate::OutboxDeadLetters::list_dead`]) carries `None`, and
    /// such a reference is fenced for those same operations.
    #[must_use]
    pub fn record_ref(&self) -> RecordRef {
        match self.claim_token {
            Some(token) => RecordRef::claimed(self.id, self.created_at, token),
            None => RecordRef::new(self.id, self.created_at),
        }
    }

    /// This row's derived [`OutboxState`] (ADR 0042 Amendment B) — computed only from the timestamps
    /// above, never stored.
    ///
    /// `now` is a parameter rather than [`time::OffsetDateTime::now_utc`] because leases are
    /// **DB-authoritative** (§6): a caller passes the store's own clock — `OutboxStats::as_of`,
    /// or a fresh `SELECT now()` — never its own, or a worker with clock skew could see a row as
    /// `Pending` (or `Leased`) that the database itself would place on the other side of the
    /// claim boundary.
    ///
    /// Precedence mirrors the store's claim predicate exactly (`WHERE published_at IS NULL AND
    /// dead_at IS NULL AND available_at <= now()`, `crates/reliar-store-postgres/src/outbox/claim.rs`):
    /// a row is `Dead` first, `Published` second, `Leased` when `locked_by` is set **and**
    /// `available_at` has **not yet** passed `now` — i.e. the claim query's own
    /// `available_at <= now()` claimable test failed — and `Pending` otherwise. **Derived from
    /// `locked_by`, not `claim_token`**: `claim_token IS NOT NULL AND available_at > now()` would
    /// misread a token-less lease (a pre-`0011` binary, or a rolled-back one, that writes
    /// `locked_by` without a token) as `Pending`, contradicting ADR 0046 §7's protection for that
    /// state (ADR 0050 §3). **The boundary is `available_at == now` ⇒ `Pending`**, matching the
    /// claim's own `available_at <= now()` test, which *succeeds* at that instant — the tick ADR
    /// 0050 corrected.
    ///
    /// ```
    /// use bytes::Bytes;
    /// use reliar_core::{Envelope, Message};
    /// use reliar_outbox::{DeadReason, OutboxRecord, OutboxRecordId, OutboxState, WorkerId};
    /// use time::{Duration, OffsetDateTime};
    /// use reliar_core::uuid::Uuid;
    ///
    /// # #[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 = || OutboxRecordId::from_uuid(Uuid::now_v7());
    /// let now = OffsetDateTime::now_utc();
    ///
    /// let pending = OutboxRecord::builder(id(), envelope(), now).build();
    /// assert_eq!(pending.state(now), OutboxState::Pending);
    ///
    /// let leased = OutboxRecord::builder(id(), envelope(), now)
    ///     .locked_by(Some(WorkerId::generate()))
    ///     .available_at(now + Duration::seconds(30))
    ///     .build();
    /// assert_eq!(leased.state(now), OutboxState::Leased);
    ///
    /// let published = OutboxRecord::builder(id(), envelope(), now).published_at(Some(now)).build();
    /// assert_eq!(published.state(now), OutboxState::Published);
    ///
    /// let dead = OutboxRecord::builder(id(), envelope(), now)
    ///     .dead(Some(now), Some(DeadReason::PermanentError))
    ///     .build();
    /// assert_eq!(dead.state(now), OutboxState::Dead);
    /// ```
    #[must_use]
    pub fn state(&self, now: time::OffsetDateTime) -> OutboxState {
        if self.dead_at.is_some() {
            OutboxState::Dead
        } else if self.published_at.is_some() {
            OutboxState::Published
        } else if self.locked_by.is_some() && self.available_at > now {
            OutboxState::Leased
        } else {
            OutboxState::Pending
        }
    }

    /// **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, OutboxRecordId};
    /// use time::OffsetDateTime;
    /// use reliar_core::uuid::Uuid;
    ///
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping {
    /// #     const TYPE: &'static str = "ping";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(id, envelope, OffsetDateTime::now_utc()).build();
    /// assert_eq!(record.id, id);
    /// assert_eq!(record.attempts, 0);
    /// ```
    pub fn builder(
        id: OutboxRecordId,
        envelope: SerializedEnvelope,
        created_at: time::OffsetDateTime,
    ) -> OutboxRecordBuilder {
        OutboxRecordBuilder::new(id, envelope, created_at)
    }
}

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

    envelope: SerializedEnvelope,

    created_at: time::OffsetDateTime,

    ordering_key: Option<String>,

    attempts: u32,

    available_at: time::OffsetDateTime,

    locked_by: Option<WorkerId>,

    claim_token: Option<ClaimToken>,

    published_at: Option<time::OffsetDateTime>,

    dead_at: Option<time::OffsetDateTime>,

    dead_reason: Option<DeadReason>,

    last_error: Option<String>,
}

impl OutboxRecordBuilder {
    fn new(
        id: OutboxRecordId,
        envelope: SerializedEnvelope,
        created_at: time::OffsetDateTime,
    ) -> Self {
        Self {
            id,
            envelope,
            available_at: created_at,
            created_at,
            ordering_key: None,
            attempts: 0,
            locked_by: None,
            claim_token: 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, OutboxRecordId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(id, envelope, 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, OutboxRecordId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(id, envelope, 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, OutboxRecordId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # 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(id, envelope, 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`]. A lease's end is [`Self::available_at`] — since ADR 0050
    /// there is no second lease column, so a provider rehydrating a leased row calls both:
    /// `.locked_by(Some(worker)).available_at(lease_end)`.
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::{OutboxRecord, OutboxRecordId, WorkerId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # 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(id, envelope, time::OffsetDateTime::now_utc())
    ///     .locked_by(Some(worker.clone()))
    ///     .available_at(until)
    ///     .build();
    /// assert_eq!(record.locked_by, Some(worker));
    /// assert_eq!(record.available_at, until);
    /// ```
    pub fn locked_by(mut self, by: Option<WorkerId>) -> Self {
        self.locked_by = by;

        self
    }

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

        self
    }

    /// Sets [`OutboxRecord::published_at`].
    ///
    /// ```
    /// # use bytes::Bytes;
    /// # use reliar_core::{Envelope, Message};
    /// # use reliar_outbox::{OutboxRecord, OutboxRecordId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let now = time::OffsetDateTime::now_utc();
    /// let record = OutboxRecord::builder(id, envelope, 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, OutboxRecordId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let now = time::OffsetDateTime::now_utc();
    /// let record = OutboxRecord::builder(id, envelope, 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, OutboxRecordId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(id, envelope, 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, OutboxRecordId};
    /// # use reliar_core::uuid::Uuid;
    /// # #[derive(serde::Serialize, serde::Deserialize)]
    /// # struct Ping;
    /// # impl Message for Ping { const TYPE: &'static str = "ping"; const VERSION: u16 = 1; }
    /// # let id = OutboxRecordId::from_uuid(Uuid::now_v7());
    /// # let envelope = Envelope::builder(Ping).build().map_body(|_| Bytes::from_static(b"{}"));
    /// let record = OutboxRecord::builder(id, envelope, time::OffsetDateTime::now_utc()).build();
    /// assert!(record.published_at.is_none());
    /// ```
    #[must_use]
    pub fn build(self) -> OutboxRecord {
        OutboxRecord {
            id: self.id,
            envelope: self.envelope,
            created_at: self.created_at,
            ordering_key: self.ordering_key,
            attempts: self.attempts,
            available_at: self.available_at,
            locked_by: self.locked_by,
            claim_token: self.claim_token,
            published_at: self.published_at,
            dead_at: self.dead_at,
            dead_reason: self.dead_reason,
            last_error: self.last_error,
        }
    }
}