reliar-store-postgres 0.9.0

//! [`reliar_inbox::InboxDeadLetters`] for [`PostgresInboxStore`] (inbox contract §3.1, ADR 0042
//! A.2.5). Three statements, all on the provider's own pool; `Tx` is inert for every one of them.
//! The whole store-layer policy lives in this trait impl, reading `self.session` directly and
//! calling `dead_letters`'s query functions for their statements.

use sqlx::PgConnection;

use reliar_inbox::{InboxDeadLetters, InboxDeadQuery, InboxRecord, InboxRecordId};
use tracing::Instrument as _;

use super::dead_letters as repo;
use super::error::PostgresInboxError;
use super::inbox_store::build_record;

use super::PostgresInboxStore;

/// The largest [`InboxDeadQuery::limit`] [`InboxDeadLetters::list_dead`] honours — a
/// caller-supplied value above this is silently capped, never sent to the database, mirroring
/// [`crate::PostgresOutboxStore`]'s own `MAX_LIST_DEAD_LIMIT` (outbox contract; inbox contract
/// §5.2 I-P23).
const MAX_LIST_DEAD_LIMIT: u32 = 1000;

impl InboxDeadLetters for PostgresInboxStore {
    type Error = PostgresInboxError;

    /// `ORDER BY dead_at, id` is normative: database-authored death time is the keyset's leading
    /// column, and the unique row id breaks ties.
    // Block form — reason (a): the `reliar.inbox.list_dead` span's entry fields must be recorded
    // before the statement runs.
    fn list_dead(
        &self,
        query: InboxDeadQuery,
    ) -> impl Future<Output = Result<Vec<InboxRecord>, Self::Error>> + Send {
        // Provider-capped at `MAX_LIST_DEAD_LIMIT` (1000) — distinct from `InboxDeadQuery`'s own
        // `limit` default of 100: a caller-supplied `limit` above the cap never reaches the
        // database, whatever value `InboxDeadQuery` carries.
        let capped_limit = query.limit.min(MAX_LIST_DEAD_LIMIT);
        let limit = i64::from(capped_limit);
        let scope_owned = query.scope.clone();
        let dead_before = query.dead_before;
        let (after_dead_at, after_id) = query.after.map_or((None, None), |cursor| {
            (Some(cursor.dead_at()), Some(cursor.id().as_uuid()))
        });
        let message_type = query.message_type.clone();

        let span = tracing::debug_span!(
            "reliar.inbox.list_dead",
            inbox.scope = scope_owned.as_ref().map(reliar_inbox::InboxScope::as_str),
            inbox.limit = capped_limit,
            inbox.returned = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            let scope = scope_owned.as_ref().map(reliar_inbox::InboxScope::as_str);

            let rows = self
                .session
                .run(async |conn: &mut PgConnection| {
                    repo::list_dead_rows(
                        &mut *conn,
                        repo::ListDeadRowsParams {
                            scope,
                            message_type: message_type.as_deref(),
                            dead_before,
                            after_dead_at,
                            after_id,
                            limit,
                        },
                    )
                    .await
                })
                .await
                .map_err(|e| self.session.map_err::<PostgresInboxError>(e))?;

            recording_span.record("inbox.returned", rows.len());

            rows.into_iter().map(build_record).collect()
        }
        .instrument(span)
    }

    // Block form — reason (a): the `reliar.inbox.retry_dead` span's `inbox.requested` field must
    // be recorded before the statement runs.
    fn retry_dead(
        &self,
        ids: &[InboxRecordId],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        let requested = ids.len();
        let ids: Vec<uuid::Uuid> = ids
            .iter()
            .map(reliar_inbox::InboxRecordId::as_uuid)
            .collect();

        let span = tracing::debug_span!(
            "reliar.inbox.retry_dead",
            inbox.requested = requested,
            inbox.affected = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            if ids.is_empty() {
                recording_span.record("inbox.affected", 0_u64);

                return Ok(0);
            }

            let result = self
                .session
                .run(async |conn: &mut PgConnection| {
                    repo::retry_dead_rows(&mut *conn, repo::RetryDeadRowsParams { ids: &ids }).await
                })
                .await
                .map_err(|e| self.session.map_err::<PostgresInboxError>(e))?;

            recording_span.record("inbox.affected", result);

            Ok(result)
        }
        .instrument(span)
    }

    // Block form — reason (a): the `reliar.inbox.purge_dead` span's `inbox.requested` field must
    // be recorded before the statement runs.
    fn purge_dead(
        &self,
        ids: &[InboxRecordId],
    ) -> impl Future<Output = Result<u64, Self::Error>> + Send {
        let requested = ids.len();
        let ids: Vec<uuid::Uuid> = ids
            .iter()
            .map(reliar_inbox::InboxRecordId::as_uuid)
            .collect();

        let span = tracing::debug_span!(
            "reliar.inbox.purge_dead",
            inbox.requested = requested,
            inbox.affected = tracing::field::Empty,
        );
        let recording_span = span.clone();

        async move {
            if ids.is_empty() {
                recording_span.record("inbox.affected", 0_u64);

                return Ok(0);
            }

            let result = self
                .session
                .run(async |conn: &mut PgConnection| {
                    repo::purge_dead_rows(&mut *conn, repo::PurgeDeadRowsParams { ids: &ids }).await
                })
                .await
                .map_err(|e| self.session.map_err::<PostgresInboxError>(e))?;

            recording_span.record("inbox.affected", result);

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