reliar-store-postgres 0.7.0

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

use reliar_core::Serializer;
use reliar_outbox::{
    DeadCursor, DeadLetterPage, DeadQuery, OutboxDeadLetters, OutboxRecordId, PoisonedRow,
    RecordRef,
};

use super::error::PostgresOutboxError;
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 = PostgresOutboxError;

    /// **`ORDER BY dead_at ASC, id ASC` is normative**: `after` is a composite keyset
    /// cursor over the columns `ix_outbox_dead_cursor` orders by; `message_type`/`tenant_id`/
    /// `dead_before` are filters only. The cursor returned comes from the last row **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 last_cursor: Option<DeadCursor> = None;

        for raw in rows {
            debug_assert!(raw.dead_at.is_some(), "list_dead selects only dead rows");

            if let Some(dead_at) = raw.dead_at {
                last_cursor = Some(DeadCursor::new(dead_at, OutboxRecordId::from_uuid(raw.id)));
            }

            match decode_row(raw) {
                Ok(record) => records.push(record),
                Err(err) => poisoned.push(PoisonedRow::new(err.id, err.message_id, 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 = if scanned == capped_limit as usize {
            last_cursor
        } else {
            None
        };

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

    /// 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: &[RecordRef]) -> 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: &[RecordRef]) -> 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> {
    let (after_dead_at, after_id) = query.after.map_or((None, None), |cursor| {
        (Some(cursor.dead_at()), Some(cursor.id().as_uuid()))
    });

    sqlx::query_as!(
        RawRow,
        r#"SELECT id, message_id, 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::timestamptz IS NULL OR (dead_at, id) > ($4, $5::uuid))
            ORDER BY dead_at ASC, id ASC
            LIMIT $6"#,
        query.message_type,
        query.tenant_id,
        query.dead_before,
        after_dead_at,
        after_id,
        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())
}