reliar_store_postgres/inbox/

inbox_store.rs

//! The [`PostgresInboxStore`] type itself: fields, construction (`new`/`with_settings`), the
//! whole [`reliar_inbox::InboxStore`] implementation (one file per public trait, mirroring
//! `outbox::outbox_store`'s shape — `docs/architecture/store-postgres-layout.md` Part II §6), and the private helpers its method bodies
//! call into or share: `claim_locked` (and its own `claim_row_params`/`claim_from_state`/
//! `claimed_attempt`/`claimed_attempts_recorded`/`ADVISORY_LOCK_CLASS`) for `claim`,
//! `format_error_chain` for `fail`, and `build_record` — shared with `InboxDeadLetters::list_dead`
//! in `inbox_store_dead_letters.rs`, so it stays `pub(super)`.

use reliar_core::{ConversationId, CorrelationId, MessageId, MessageType};
use reliar_inbox::InboxStore;
use reliar_inbox::{
    InboxClaim, InboxFailure, InboxMessage, InboxPurgeReport, InboxPurgeRequest, InboxRecord,
    InboxRecordId, InboxScope,
};
use sqlx::{PgConnection, Postgres, Transaction};
use tracing::Instrument as _;

use crate::connection::session::Session;
use crate::records::truncate_last_error;
use crate::settings::PostgresInboxSettings;

use super::claim as claim_repo;
use super::claim::ClaimStateRow;
use super::error::PostgresInboxError;
use super::outcomes as outcomes_repo;
use super::purge as purge_repo;
use super::rows::InboxRow;

/// Reliar's PostgreSQL inbox provider (inbox contract §3). A **separate type** from
/// [`crate::PostgresOutboxStore`]: the inbox stores no payload, so it needs no `Serializer` type
/// parameter and none of the outbox's lease/ordering/retention settings. Same crate, same schema,
/// same [`crate::migrate`]. Cheap to clone — wraps a [`sqlx::PgPool`]; no outer `Arc` required.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct PostgresInboxStore {
    // `pub(super)`: `inbox_store_dead_letters.rs` reads this directly — its whole
    // `InboxDeadLetters` impl (signature and body) lives there rather than delegating from this
    // file. `claim`/`fail`/`find`/`purge` above read it as `&self.session` from the trait methods
    // instead. `complete` never touches it: it runs inside the caller's own transaction, never
    // through `Session::run`.
    pub(super) session: Session,

    settings: PostgresInboxSettings,
}

impl PostgresInboxStore {
    /// Wraps `pool` with [`PostgresInboxSettings::default`]. Performs **no I/O**: it issues no
    /// query, opens no connection and verifies nothing about the database. The pool stays the
    /// host's.
    ///
    /// Call [`crate::migrate`] (or apply the published SQL through your own pipeline) **before**
    /// the first store call, and make sure the connection's `search_path` resolves the
    /// unqualified name `inbox` to the migrated schema — see the crate docs. An un-migrated or
    /// unreachable table surfaces at the first statement as
    /// [`PostgresInboxError::NotMigrated`], never here.
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::PostgresInboxStore;
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresInboxStore::new(pool);
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn new(pool: sqlx::PgPool) -> Self {
        let settings = PostgresInboxSettings::default();
        let session = Session::new(pool, settings.statement_timeout);

        Self { session, settings }
    }

    /// As [`Self::new`], with explicit `settings`. Performs no I/O beyond the settings' own
    /// validation, which never touches the database.
    ///
    /// # Errors
    ///
    /// [`PostgresInboxError::InvalidSettings`] when `settings.max_attempts == 0` — the one
    /// rejection this crate can make without asking the database (ADR 0042 A.2.4).
    ///
    /// ```no_run
    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
    /// use reliar_store_postgres::{PostgresInboxSettings, PostgresInboxStore};
    /// use sqlx::postgres::PgPoolOptions;
    ///
    /// let pool = PgPoolOptions::new()
    ///     .connect(&std::env::var("DATABASE_URL")?)
    ///     .await?;
    /// let store = PostgresInboxStore::with_settings(
    ///     pool,
    ///     PostgresInboxSettings::default().max_attempts(5),
    /// )?;
    /// # let _ = store;
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_settings(
        pool: sqlx::PgPool,
        settings: PostgresInboxSettings,
    ) -> Result<Self, PostgresInboxError> {
        settings.validate()?;

        let session = Session::new(pool, settings.statement_timeout);

        Ok(Self { session, settings })
    }

    /// [`PostgresInboxSettings::max_attempts`] — the one setting a runtime call reads outside
    /// construction, read directly by `fail`'s body below rather than a hidden dependency on
    /// `self` further down the call chain.
    pub(super) fn max_attempts(&self) -> u32 {
        self.settings.max_attempts
    }
}

impl<'c> InboxStore<Transaction<'c, Postgres>> for PostgresInboxStore {
    type Error = PostgresInboxError;

    /// The three-statement claim (inbox contract §3.1): the in-flight advisory-lock guard, the
    /// `INSERT … ON CONFLICT DO NOTHING` claim, and — only when nothing was inserted — the state
    /// read that decides `AlreadyCompleted`/`Dead` vs. `Claimed`. The three statements themselves
    /// live in `claim_locked`, below.
    // Block form — reason (a): `inbox.scope`/`message.id`/`message.type` must be recorded on the
    // `reliar.inbox.claim` span before the first statement runs (inbox contract §4), so the span
    // itself has to exist before the async block that runs those statements does.
    fn claim(
        &self,
        tx: &mut Transaction<'c, Postgres>,
        scope: &InboxScope,
        message: InboxMessage<'_>,
    ) -> impl Future<Output = Result<InboxClaim, Self::Error>> + Send {
        let span = tracing::debug_span!(
            "reliar.inbox.claim",
            inbox.scope = %scope,
            message.id = %message.id,
            // `r#type`, not `type` — `type` is a Rust keyword; tracing strips the `r#` prefix
            // from the field name, so this still renders as `message.type` (inbox contract §4).
            message.r#type = %message.message_type,
            inbox.outcome = tracing::field::Empty,
            inbox.attempt = tracing::field::Empty,
            inbox.attempts = tracing::field::Empty,
            inbox.record_id = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let result = claim_locked(tx, scope.as_str(), message).await;

            if let Ok(claim) = &result {
                record_claim_outcome(&recording_span, claim);
            }

            result
        }
        .instrument(span)
    }

    /// Marks the row completed in the caller's transaction. Zero rows affected ⇒ `NotClaimed`
    /// (including a row that has since gone dead — the guard is `completed_at IS NULL AND
    /// dead_at IS NULL`).
    // Block form — reason (a): the span's `inbox.scope`/`message.id` fields (inbox contract §4)
    // must be created before the statement they describe runs.
    fn complete(
        &self,
        tx: &mut Transaction<'c, Postgres>,
        scope: &InboxScope,
        id: MessageId,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        let span = tracing::debug_span!(
            "reliar.inbox.complete",
            inbox.scope = %scope,
            message.id = %id,
        );

        async move {
            let scope_str = scope.as_str();
            let message_id = id.as_uuid();
            let affected = outcomes_repo::complete_row(
                tx,
                outcomes_repo::CompleteRowParams {
                    scope: scope_str,
                    message_id,
                },
            )
            .await?;

            if affected == 0 {
                return Err(PostgresInboxError::NotClaimed {
                    scope: scope_str.to_owned(),
                    message_id: id,
                });
            }

            Ok(())
        }
        .instrument(span)
    }

    /// Records a failed attempt on this store's own pool, guarded by `completed_at IS NULL`, and
    /// bounds it at `settings.max_attempts` atomically with the increment (ADR 0042 A.2.4).
    // Block form: `error: &(dyn Error + 'static)` is not `Send`, so its `Display` chain must be
    // extracted into an owned `String` before the async block is built, never inside a plain
    // `async fn` (conventions §3(b); the trait's own rustdoc calls this out) — also reason (a):
    // the `reliar.inbox.fail` span's entry fields must be recorded before the statement runs.
    fn fail(
        &self,
        scope: &InboxScope,
        message: InboxMessage<'_>,
        error: &(dyn std::error::Error + 'static),
    ) -> impl Future<Output = Result<InboxFailure, Self::Error>> + Send {
        let last_error = format_error_chain(error);
        let span = tracing::debug_span!(
            "reliar.inbox.fail",
            inbox.scope = %scope,
            message.id = %message.id,
            // `r#type`, not `type` — `type` is a Rust keyword; tracing strips the `r#` prefix
            // from the field name, so this still renders as `message.type` (inbox contract §4).
            message.r#type = %message.message_type,
            inbox.outcome = tracing::field::Empty,
            inbox.attempts = tracing::field::Empty,
            inbox.record_id = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let session = &self.session;
            let scope_str = scope.as_str();
            let last_error = truncate_last_error(last_error);
            let max_attempts = i32::try_from(self.max_attempts()).unwrap_or(i32::MAX);
            let id = InboxRecordId::new().as_uuid();
            let message_id = message.id.as_uuid();
            let message_type = message.message_type.name();
            let message_version = i32::from(message.message_type.version());
            let conversation_id = message.conversation_id.as_uuid();
            let correlation_id = message.correlation_id.map(CorrelationId::as_str);
            let causation_id = message.causation_id.map(|c| c.as_uuid());

            let row = session
                .run(async |conn: &mut PgConnection| {
                    outcomes_repo::fail_row(
                        &mut *conn,
                        outcomes_repo::FailRowParams {
                            id,
                            scope: scope_str,
                            message_id,
                            message_type,
                            message_version,
                            conversation_id,
                            correlation_id,
                            causation_id,
                            last_error: &last_error,
                            max_attempts,
                        },
                    )
                    .await
                })
                .await
                .map_err(|e| session.map_err::<PostgresInboxError>(e))?;

            let failure = match row {
                None => InboxFailure::AlreadyCompleted,
                Some(row) => {
                    let attempts = u32::try_from(row.attempts).unwrap_or(u32::MAX);

                    match row.dead_at {
                        Some(dead_at) => InboxFailure::Dead {
                            id: InboxRecordId::from_uuid(row.id),
                            attempts,
                            dead_at,
                        },
                        None => InboxFailure::Recorded { attempts },
                    }
                }
            };

            record_fail_outcome(&recording_span, &failure);

            Ok(failure)
        }
        .instrument(span)
    }

    /// Reads a row for diagnostics. No Reliar code path calls it. No span — the inbox contract's
    /// observability table (§4) does not list `find`, since no Reliar code path calls it.
    async fn find(
        &self,
        scope: &InboxScope,
        id: MessageId,
    ) -> Result<Option<InboxRecord>, PostgresInboxError> {
        let session = &self.session;
        let scope_str = scope.as_str();
        let message_id = id.as_uuid();
        let row = session
            .run(async |conn: &mut PgConnection| {
                purge_repo::find_row(
                    &mut *conn,
                    purge_repo::FindRowParams {
                        scope: scope_str,
                        message_id,
                    },
                )
                .await
            })
            .await
            .map_err(|e| session.map_err::<PostgresInboxError>(e))?;

        let Some(row) = row else {
            return Ok(None);
        };

        Ok(Some(build_record(row)?))
    }

    /// One bounded pass, three statements, each capped at `request.batch_size`: the
    /// completed-row, incomplete-row and dead-row deletes.
    // Block form — reason (a): the `reliar.inbox.purge` span must exist before the three
    // statements it will report on run.
    fn purge(
        &self,
        request: InboxPurgeRequest,
    ) -> impl Future<Output = Result<InboxPurgeReport, Self::Error>> + Send {
        let span = tracing::debug_span!(
            "reliar.inbox.purge",
            inbox.completed_deleted = tracing::field::Empty,
            inbox.incomplete_deleted = tracing::field::Empty,
            inbox.dead_deleted = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let session = &self.session;
            let batch_size = i64::from(request.batch_size);
            let completed_ms = request.completed_retention.map(to_millis);
            let incomplete_ms = request.incomplete_retention.map(to_millis);
            let dead_ms = request.dead_retention.map(to_millis);

            let (completed_deleted, incomplete_deleted, dead_deleted) = session
                .run(async |conn: &mut PgConnection| {
                    let completed_deleted = match completed_ms {
                        Some(retention_ms) => {
                            purge_repo::purge_completed_rows(
                                &mut *conn,
                                purge_repo::PurgeCompletedRowsParams {
                                    retention_ms,
                                    batch_size,
                                },
                            )
                            .await?
                        }
                        None => 0,
                    };

                    let incomplete_deleted = match incomplete_ms {
                        Some(retention_ms) => {
                            purge_repo::purge_incomplete_rows(
                                &mut *conn,
                                purge_repo::PurgeIncompleteRowsParams {
                                    retention_ms,
                                    batch_size,
                                },
                            )
                            .await?
                        }
                        None => 0,
                    };

                    let dead_deleted = match dead_ms {
                        Some(retention_ms) => {
                            purge_repo::purge_dead_retention_rows(
                                &mut *conn,
                                purge_repo::PurgeDeadRetentionRowsParams {
                                    retention_ms,
                                    batch_size,
                                },
                            )
                            .await?
                        }
                        None => 0,
                    };

                    Ok((completed_deleted, incomplete_deleted, dead_deleted))
                })
                .await
                .map_err(|e| session.map_err::<PostgresInboxError>(e))?;

            let report = InboxPurgeReport::new(completed_deleted, incomplete_deleted, dead_deleted);

            recording_span.record("inbox.completed_deleted", report.completed_deleted);
            recording_span.record("inbox.incomplete_deleted", report.incomplete_deleted);
            recording_span.record("inbox.dead_deleted", report.dead_deleted);

            Ok(report)
        }
        .instrument(span)
    }
}

/// The `reliar.inbox.claim` span's outcome fields (ADR 0042 Amendment C.5): `inbox.outcome`
/// always, `inbox.attempt` on `Claimed`, `inbox.record_id` + `inbox.attempts` on `Dead`. Never
/// recorded on an `Err` path — the caller's own log carries the failure.
fn record_claim_outcome(span: &tracing::Span, claim: &InboxClaim) {
    match claim {
        InboxClaim::Claimed { attempt } => {
            span.record("inbox.outcome", "claimed");
            span.record("inbox.attempt", attempt);
        }
        InboxClaim::AlreadyCompleted { .. } => {
            span.record("inbox.outcome", "already_completed");
        }
        InboxClaim::InProgress => {
            span.record("inbox.outcome", "in_progress");
        }
        InboxClaim::Dead { id, attempts, .. } => {
            span.record("inbox.outcome", "dead");
            span.record("inbox.record_id", tracing::field::display(id));
            span.record("inbox.attempts", attempts);
        }
        // `InboxClaim` is `#[non_exhaustive]`; every variant this crate's contract defines is
        // matched above.
        _ => {}
    }
}

/// The `reliar.inbox.fail` span's outcome fields (inbox contract §4): `inbox.outcome` always,
/// `inbox.attempts`/`inbox.record_id` where the variant carries them. Never recorded on an `Err`
/// path.
fn record_fail_outcome(span: &tracing::Span, failure: &InboxFailure) {
    match failure {
        InboxFailure::Recorded { attempts } => {
            span.record("inbox.outcome", "recorded");
            span.record("inbox.attempts", attempts);
        }
        InboxFailure::Dead {
            id,
            attempts,
            dead_at: _,
        } => {
            span.record("inbox.outcome", "dead");
            span.record("inbox.attempts", attempts);
            span.record("inbox.record_id", tracing::field::display(id));
        }
        InboxFailure::AlreadyCompleted => {
            span.record("inbox.outcome", "already_completed");
        }
        // `InboxFailure` is `#[non_exhaustive]`; every variant this crate's contract defines is
        // matched above.
        _ => {}
    }
}

/// The fixed Reliar advisory-lock "class" for the inbox's in-flight guard (inbox contract §3.1):
/// `i32::from_be_bytes(*b"RELI")`, the first argument to the two-argument
/// `pg_try_advisory_xact_lock`, which PostgreSQL documents as a distinct key space from the
/// one-argument `bigint` form — so this can never collide with a key any other Reliar or host
/// code chooses in that space. Released by the caller's commit or rollback; nothing to clean up.
const ADVISORY_LOCK_CLASS: i32 = i32::from_be_bytes(*b"RELI");

/// [`InboxStore::claim`]'s body — see that trait method's rustdoc for the full contract.
async fn claim_locked(
    tx: &mut Transaction<'_, Postgres>,
    scope: &str,
    message: InboxMessage<'_>,
) -> Result<InboxClaim, PostgresInboxError> {
    let message_id = message.id.as_uuid();

    // 1. the in-flight guard (inbox contract §3.1/ADR 0042 §3). `false` ⇒ `InProgress`, return
    // now, no write, tx still usable. A collision between two *concurrently claimed* keys can
    // report this spuriously — a redelivery, never a lost or doubled effect.
    let acquired = claim_repo::try_advisory_lock(
        &mut **tx,
        claim_repo::TryAdvisoryLockParams {
            class: ADVISORY_LOCK_CLASS,
            scope,
            message_id,
        },
    )
    .await?;

    if !acquired {
        return Ok(InboxClaim::InProgress);
    }

    // 2. the claim. DO NOTHING, never DO UPDATE: the AlreadyCompleted path is the hot path of a
    // redelivery storm and must not write, WAL or bloat a row it only reads. A returned row means
    // we inserted ⇒ read its own `attempts` back rather than assume 0, so this stays correct even
    // if a future migration ever gives the row a non-zero starting value.
    let id = InboxRecordId::new();
    let claim_params = claim_row_params(id, scope, &message);

    if let Some(attempts) = claim_repo::insert_claim_row(&mut **tx, claim_params).await? {
        return Ok(InboxClaim::Claimed {
            attempt: claimed_attempt(attempts),
        });
    }

    // 3. only when 2 returned nothing: the key was committed a moment ago, so decide from its
    // state. A concurrent `purge` can delete the row between step 2's conflict and this read's
    // own READ COMMITTED snapshot — the advisory lock held since step 1 serializes *claims*
    // only, so a deleted-then-recreated key is a real possibility here, handled below.
    let row = claim_repo::select_claim_state(
        &mut **tx,
        claim_repo::SelectClaimStateParams { scope, message_id },
    )
    .await?;

    let Some(row) = row else {
        // The advisory lock held since step 1 serializes *claims* only — `fail` inserts this
        // same `(scope, message_id)` (ADR 0042 §4's `ON CONFLICT … DO UPDATE`) as a plain pool
        // statement, without ever taking it, precisely for the case where the claiming
        // transaction has rolled back. So a concurrent `purge` deleting the row this session's
        // `SELECT` just missed, followed by a concurrent `fail` recreating it before this
        // session's own re-insert runs, is a real race — not a corrupt-row scenario. Rather than
        // a plain `INSERT … DO NOTHING` that could still return `None` here, the re-insert
        // upserts and reads in one statement (ADR 0042 Amendment C.8): PostgreSQL guarantees an
        // atomic insert-or-update outcome for `ON CONFLICT DO UPDATE` with no `WHERE` clause, so
        // the statement can never return zero rows.
        let upsert_params = claim_row_params(InboxRecordId::new(), scope, &message);
        let row = claim_repo::upsert_claim_row(&mut **tx, upsert_params).await?;

        return Ok(claim_from_state(&row));
    };

    Ok(claim_from_state(&row))
}

fn claim_row_params<'a>(
    id: InboxRecordId,
    scope: &'a str,
    message: &InboxMessage<'a>,
) -> claim_repo::ClaimRowParams<'a> {
    claim_repo::ClaimRowParams {
        id: id.as_uuid(),
        scope,
        message_id: message.id.as_uuid(),
        message_type: message.message_type.name(),
        message_version: i32::from(message.message_type.version()),
        conversation_id: message.conversation_id.as_uuid(),
        correlation_id: message.correlation_id.map(CorrelationId::as_str),
        causation_id: message.causation_id.map(|c| c.as_uuid()),
    }
}

/// Decides `AlreadyCompleted` → `Dead` → `Claimed` from a [`ClaimStateRow`] — the one place this
/// precedence is written, shared by step 3's `SELECT` path and its upsert fallback (layout Part II
/// §8.1; previously duplicated in both).
fn claim_from_state(row: &ClaimStateRow) -> InboxClaim {
    if let Some(completed_at) = row.completed_at {
        return InboxClaim::AlreadyCompleted { completed_at };
    }

    if let Some(dead_at) = row.dead_at {
        return InboxClaim::Dead {
            id: InboxRecordId::from_uuid(row.id),
            attempts: claimed_attempts_recorded(row.attempts),
            dead_at,
        };
    }

    InboxClaim::Claimed {
        attempt: claimed_attempt(row.attempts),
    }
}

/// `u32::try_from`/`saturating_add`, never `as` (inbox contract §3.1): an out-of-range
/// `attempts` is a corrupt row, not a panic — it saturates instead.
fn claimed_attempt(attempts: i32) -> u32 {
    u32::try_from(attempts)
        .unwrap_or(u32::MAX)
        .saturating_add(1)
}

/// Same conversion as [`claimed_attempt`], without the `+ 1`: [`InboxClaim::Dead`] reports the
/// recorded count as-is, not the next attempt ordinal.
fn claimed_attempts_recorded(attempts: i32) -> u32 {
    u32::try_from(attempts).unwrap_or(u32::MAX)
}

/// Joins `error`'s `Display` with every `source()` in its chain, `": "`-separated — the inbox
/// contract's "last failure's error chain". Never touches payload bytes or header values: it
/// only ever sees what the caller's own error type chose to put in its `Display`.
///
/// **Must run before `fail`'s future is constructed.** `&(dyn Error + 'static)` is not `Send`
/// (`&T: Send` requires `T: Sync`, and `dyn Error` is not `Sync`), so a plain `async fn fail`
/// holding it across the generator's state — even before any `.await` — fails
/// [`InboxStore::fail`]'s `+ Send` bound. This is the synchronous extraction the trait's own
/// rustdoc points implementors to.
fn format_error_chain(error: &(dyn std::error::Error + 'static)) -> String {
    let mut out = error.to_string();
    let mut source = error.source();

    while let Some(err) = source {
        out.push_str(": ");
        out.push_str(&err.to_string());
        source = err.source();
    }

    out
}

/// Rehydrates an [`InboxRecord`] from a raw [`InboxRow`] — shared by `find` above and
/// `inbox_store_dead_letters::list_dead`, so it stays `pub(super)`. Rehydration is store-layer
/// policy, unlike the row shape itself, which belongs to the concern layer (`rows::InboxRow`).
///
/// # Errors
///
/// [`PostgresInboxError::Database`] wrapping a decode failure if `row.scope`/`row.correlation_id`
/// no longer satisfy the type's own validation — only reachable if the schema and the type's
/// invariant have drifted apart, never in ordinary operation.
pub(super) fn build_record(row: InboxRow) -> Result<InboxRecord, PostgresInboxError> {
    // `InboxScope::new` re-validates a value this row's own `ck_inbox_scope_len` constraint
    // already guarantees is 1..=128 bytes, so this can only fail if the schema and the type's
    // invariant have drifted apart — treated as a corrupt row (`Database`), never a panic.
    let scope = InboxScope::new(row.scope).map_err(|err| PostgresInboxError::Database {
        source: sqlx::Error::Decode(err.into()),
    })?;

    let correlation_id = row
        .correlation_id
        .map(CorrelationId::parse)
        .transpose()
        .map_err(|err| PostgresInboxError::Database {
            source: sqlx::Error::Decode(err.into()),
        })?;

    let message_type = MessageType::from_parts(
        row.message_type,
        u16::try_from(row.message_version).unwrap_or(u16::MAX),
    );
    let message_id = MessageId::from_uuid(row.message_id);

    let mut message = InboxMessage::new(message_id, &message_type)
        .conversation(ConversationId::from_uuid(row.conversation_id));

    if let Some(correlation_id) = correlation_id.as_ref() {
        message = message.correlation(correlation_id);
    }

    if let Some(causation_id) = row.causation_id {
        message = message.causation(MessageId::from_uuid(causation_id));
    }

    Ok(InboxRecord::builder(
        InboxRecordId::from_uuid(row.id),
        scope,
        message,
        row.received_at,
    )
    .updated_at(row.updated_at)
    .completed_at(row.completed_at)
    .dead_at(row.dead_at)
    .attempts(u32::try_from(row.attempts).unwrap_or(u32::MAX))
    .last_error(row.last_error)
    .build())
}

fn to_millis(duration: std::time::Duration) -> i64 {
    i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
}