reliar-store-postgres 0.6.0

//! [`PostgresOutboxStore`]'s [`OutboxDeadLetters`] implementation: `list_dead`/`retry_dead`/
//! `purge_dead`.

use reliar_core::Serializer;
use reliar_outbox::{DeadLetterPage, DeadQuery, MessageRef, OutboxDeadLetters, PoisonedRow};

use crate::error::PostgresStoreError;
use crate::records::{RawRow, decode_row};

use super::PostgresOutboxStore;

/// The largest `DeadQuery::limit` [`OutboxDeadLetters::list_dead`] honours — a caller-supplied
/// value above this is silently capped, never sent to the database: this store is
/// provider-capped, with a default of 100.
const MAX_LIST_DEAD_LIMIT: u32 = 1000;

impl<Ser: Serializer + Send + Sync + 'static> OutboxDeadLetters for PostgresOutboxStore<Ser> {
    type Error = PostgresStoreError;

    /// **`ORDER BY sequence ASC` is normative**: `after_sequence` is a keyset
    /// cursor over `sequence`, the column `ix_outbox_dead` orders by; `message_type`/
    /// `tenant_id`/`dead_before` are filters only, expressed as `($n::type IS NULL OR ...)` so
    /// one static statement serves every combination. The cursor returned is the largest
    /// `sequence` **scanned**, poisoned rows included, so a poisoned tail cannot loop the
    /// caller forever.
    async fn list_dead(&self, query: DeadQuery) -> Result<DeadLetterPage, Self::Error> {
        // Provider-capped, default 100: a caller-supplied
        // limit above this never reaches the database, regardless of what `DeadQuery` carries.
        let capped_limit = query.limit.min(MAX_LIST_DEAD_LIMIT);
        let limit = i64::from(capped_limit);

        let rows = if self.settings.statement_timeout.is_zero() {
            list_dead_rows(&self.pool, &query, limit)
                .await
                .map_err(|e| self.map_err(e))?
        } else {
            let mut tx = self.pool.begin().await.map_err(|e| self.map_err(e))?;

            self.set_local_timeout(&mut tx).await?;
            let rows = list_dead_rows(&mut *tx, &query, limit)
                .await
                .map_err(|e| self.map_err(e))?;
            tx.commit().await.map_err(|e| self.map_err(e))?;

            rows
        };

        let scanned = rows.len();
        let mut records = Vec::with_capacity(scanned);
        let mut poisoned = Vec::new();
        let mut max_sequence: Option<i64> = None;

        for raw in rows {
            max_sequence = Some(max_sequence.map_or(raw.sequence, |m| m.max(raw.sequence)));

            match decode_row(raw) {
                Ok(record) => records.push(record),
                Err(err) => poisoned.push(PoisonedRow::new(err.id, err.sequence, err.detail)),
            }
        }

        // "Full" is scanned == limit, poisoned rows included — they occupy a row in the scan,
        // so counting only decoded records would stop pagination early on a poisoned tail.
        let next_after_sequence = if scanned == capped_limit as usize {
            max_sequence
        } else {
            None
        };

        Ok(DeadLetterPage::new(records, poisoned, next_after_sequence))
    }

    /// Returns dead rows to pending: clears the lease that already isn't there, resets
    /// `attempts` to 0 (the **only** operation that does), keeps `last_error` for audit. Not
    /// worker-guarded — a dead row holds no lease, so there is no owner to check against.
    async fn retry_dead(&self, refs: &[MessageRef]) -> Result<u64, Self::Error> {
        if refs.is_empty() {
            return Ok(0);
        }

        let ids: Vec<uuid::Uuid> = refs.iter().map(|r| r.id.as_uuid()).collect();
        let affected = if self.settings.statement_timeout.is_zero() {
            retry_dead_rows(&self.pool, &ids)
                .await
                .map_err(|e| self.map_err(e))?
        } else {
            let mut tx = self.pool.begin().await.map_err(|e| self.map_err(e))?;

            self.set_local_timeout(&mut tx).await?;
            let affected = retry_dead_rows(&mut *tx, &ids)
                .await
                .map_err(|e| self.map_err(e))?;
            tx.commit().await.map_err(|e| self.map_err(e))?;

            affected
        };

        Ok(affected)
    }

    /// Deletes dead rows by reference, regardless of
    /// [`PurgeRequest::dead_retention`](reliar_outbox::PurgeRequest::dead_retention).
    async fn purge_dead(&self, refs: &[MessageRef]) -> Result<u64, Self::Error> {
        if refs.is_empty() {
            return Ok(0);
        }

        let ids: Vec<uuid::Uuid> = refs.iter().map(|r| r.id.as_uuid()).collect();
        let affected = if self.settings.statement_timeout.is_zero() {
            purge_dead_rows(&self.pool, &ids)
                .await
                .map_err(|e| self.map_err(e))?
        } else {
            let mut tx = self.pool.begin().await.map_err(|e| self.map_err(e))?;

            self.set_local_timeout(&mut tx).await?;
            let affected = purge_dead_rows(&mut *tx, &ids)
                .await
                .map_err(|e| self.map_err(e))?;
            tx.commit().await.map_err(|e| self.map_err(e))?;

            affected
        };

        Ok(affected)
    }
}

/// `list_dead`'s query, shared by the plain-pool and `statement_timeout`-wrapped-transaction
/// call sites. Named against [`RawRow`] via `query_as!`, same as the claim — the `SELECT`
/// list matches its field order exactly.
async fn list_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    query: &DeadQuery,
    limit: i64,
) -> Result<Vec<RawRow>, sqlx::Error> {
    sqlx::query_as!(
        RawRow,
        r#"SELECT id, sequence, message_type, message_version,
                  correlation_id, conversation_id, causation_id, request_id,
                  content_type, payload, tenant_id, expires_at, ordering_key,
                  metadata, headers, metadata_version,
                  created_at, available_at,
                  attempts, locked_by, locked_until,
                  published_at, dead_at, dead_reason, last_error
             FROM outbox
            WHERE dead_at IS NOT NULL
              AND ($1::text IS NULL OR message_type = $1)
              AND ($2::text IS NULL OR tenant_id = $2)
              AND ($3::timestamptz IS NULL OR dead_at < $3)
              AND ($4::bigint IS NULL OR sequence > $4)
            ORDER BY sequence ASC
            LIMIT $5"#,
        query.message_type,
        query.tenant_id,
        query.dead_before,
        query.after_sequence,
        limit,
    )
    .fetch_all(executor)
    .await
}

/// `retry_dead`'s query, shared by the plain-pool and `statement_timeout`-wrapped-transaction
/// call sites. Not worker-guarded — a dead row holds no lease, so there is no owner to check
/// against.
async fn retry_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    ids: &[uuid::Uuid],
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"UPDATE outbox
              SET dead_at      = NULL,
                  dead_reason  = NULL,
                  available_at = now(),
                  attempts     = 0,
                  locked_by    = NULL,
                  locked_until = NULL,
                  updated_at   = now()
            WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
        ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// `purge_dead`'s query, shared by the plain-pool and `statement_timeout`-wrapped-transaction
/// call sites.
async fn purge_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    ids: &[uuid::Uuid],
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        "DELETE FROM outbox WHERE id = ANY($1) AND dead_at IS NOT NULL",
        ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}