reliar-store-postgres 0.7.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.

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

use super::error::PostgresInboxError;

use super::PostgresInboxStore;
use super::purge::{InboxRow, build_record};

/// 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 = if self.settings.statement_timeout.is_zero() {
                list_dead_rows(
                    &self.pool,
                    scope,
                    message_type.as_deref(),
                    dead_before,
                    after_dead_at,
                    after_id,
                    limit,
                )
                .await?
            } else {
                let mut tx = self.pool.begin().await?;

                self.set_local_timeout(&mut tx).await?;
                let rows = list_dead_rows(
                    &mut *tx,
                    scope,
                    message_type.as_deref(),
                    dead_before,
                    after_dead_at,
                    after_id,
                    limit,
                )
                .await?;
                tx.commit().await?;

                rows
            };

            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 = if self.settings.statement_timeout.is_zero() {
                retry_dead_rows(&self.pool, &ids).await?
            } else {
                let mut tx = self.pool.begin().await?;

                self.set_local_timeout(&mut tx).await?;
                let result = retry_dead_rows(&mut *tx, &ids).await?;
                tx.commit().await?;

                result
            };

            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 = if self.settings.statement_timeout.is_zero() {
                purge_dead_rows(&self.pool, &ids).await?
            } else {
                let mut tx = self.pool.begin().await?;

                self.set_local_timeout(&mut tx).await?;
                let result = purge_dead_rows(&mut *tx, &ids).await?;
                tx.commit().await?;

                result
            };

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

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

#[allow(
    clippy::too_many_arguments,
    reason = "each argument is one InboxDeadQuery filter; a struct wrapper would just move the \
              same six names one level down"
)]
async fn list_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    scope: Option<&str>,
    message_type: Option<&str>,
    dead_before: Option<time::OffsetDateTime>,
    after_dead_at: Option<time::OffsetDateTime>,
    after_id: Option<uuid::Uuid>,
    limit: i64,
) -> Result<Vec<InboxRow>, sqlx::Error> {
    sqlx::query_as!(
        InboxRow,
        r#"SELECT id, scope, message_id, message_type, message_version, conversation_id,
                  correlation_id, causation_id, received_at, updated_at, completed_at, dead_at,
                  attempts, last_error
             FROM inbox
            WHERE dead_at IS NOT NULL
              AND ($1::text IS NULL OR scope = $1)
              AND ($2::text IS NULL OR message_type = $2)
              AND ($3::timestamptz IS NULL OR dead_at < $3)
              AND ($4::timestamptz IS NULL OR (dead_at, id) > ($4, $5::uuid))
            ORDER BY dead_at, id
            LIMIT $6"#,
        scope,
        message_type,
        dead_before,
        after_dead_at,
        after_id,
        limit,
    )
    .fetch_all(executor)
    .await
}

async fn retry_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    ids: &[uuid::Uuid],
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"UPDATE inbox SET dead_at = NULL, attempts = 0, updated_at = now()
            WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
        ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

async fn purge_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    ids: &[uuid::Uuid],
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"DELETE FROM inbox WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
        ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}