reliar-store-postgres 0.7.0

//! `purge`/`find`: the inbox contract's retention sweep and diagnostic read, both on the
//! provider's own pool, [`crate::PostgresInboxSettings::statement_timeout`]-bounded exactly like
//! `outcomes::fail`.

use reliar_core::{ConversationId, CorrelationId, MessageId, MessageType};
use reliar_inbox::{
    InboxMessage, InboxPurgeReport, InboxPurgeRequest, InboxRecord, InboxRecordId, InboxScope,
};

use super::error::PostgresInboxError;

use super::PostgresInboxStore;

/// [`reliar_inbox::InboxStore::find`]'s body. Diagnostics and tests only — no Reliar code path
/// calls it.
pub(super) async fn find(
    store: &PostgresInboxStore,
    scope: &InboxScope,
    message_id: MessageId,
) -> Result<Option<InboxRecord>, PostgresInboxError> {
    let scope_str = scope.as_str();
    let id = message_id.as_uuid();
    let row = if store.settings.statement_timeout.is_zero() {
        find_row(&store.pool, scope_str, id).await?
    } else {
        let mut tx = store.pool.begin().await?;

        store.set_local_timeout(&mut tx).await?;
        let row = find_row(&mut *tx, scope_str, id).await?;
        tx.commit().await?;

        row
    };

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

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

/// A row's full column set, named via `query_as!` (never `FromRow`) so `find` and
/// [`super::dead_letters::list_dead`] share one shape — every [`InboxDeadLetters`] listing reads
/// the same columns [`find`] does.
///
/// [`InboxDeadLetters`]: reliar_inbox::InboxDeadLetters
pub(super) struct InboxRow {
    pub(super) id: uuid::Uuid,

    pub(super) scope: String,

    pub(super) message_id: uuid::Uuid,

    pub(super) message_type: String,

    pub(super) message_version: i32,

    pub(super) conversation_id: uuid::Uuid,

    pub(super) correlation_id: Option<String>,

    pub(super) causation_id: Option<uuid::Uuid>,

    pub(super) received_at: time::OffsetDateTime,

    pub(super) updated_at: time::OffsetDateTime,

    pub(super) completed_at: Option<time::OffsetDateTime>,

    pub(super) dead_at: Option<time::OffsetDateTime>,

    pub(super) attempts: i32,

    pub(super) last_error: Option<String>,
}

/// Rehydrates an [`InboxRecord`] from a raw [`InboxRow`] — shared by [`find`] and
/// `dead_letters::list_dead`.
///
/// # 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())
}

async fn find_row<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    scope: &str,
    message_id: uuid::Uuid,
) -> Result<Option<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 scope = $1 AND message_id = $2"#,
        scope,
        message_id,
    )
    .fetch_optional(executor)
    .await
}

/// [`reliar_inbox::InboxStore::purge`]'s body.
pub(super) async fn purge(
    store: &PostgresInboxStore,
    request: InboxPurgeRequest,
) -> Result<InboxPurgeReport, PostgresInboxError> {
    let batch_size = i64::from(request.batch_size);

    let (completed_deleted, incomplete_deleted, dead_deleted) =
        if store.settings.statement_timeout.is_zero() {
            let completed_deleted = if let Some(retention) = request.completed_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_completed_rows(&store.pool, retention_ms, batch_size).await?
            } else {
                0
            };

            let incomplete_deleted = if let Some(retention) = request.incomplete_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_incomplete_rows(&store.pool, retention_ms, batch_size).await?
            } else {
                0
            };

            let dead_deleted = if let Some(retention) = request.dead_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_dead_rows(&store.pool, retention_ms, batch_size).await?
            } else {
                0
            };

            (completed_deleted, incomplete_deleted, dead_deleted)
        } else {
            let mut tx = store.pool.begin().await?;

            store.set_local_timeout(&mut tx).await?;

            let completed_deleted = if let Some(retention) = request.completed_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_completed_rows(&mut *tx, retention_ms, batch_size).await?
            } else {
                0
            };

            let incomplete_deleted = if let Some(retention) = request.incomplete_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_incomplete_rows(&mut *tx, retention_ms, batch_size).await?
            } else {
                0
            };

            let dead_deleted = if let Some(retention) = request.dead_retention {
                let retention_ms = i64::try_from(retention.as_millis()).unwrap_or(i64::MAX);

                purge_dead_rows(&mut *tx, retention_ms, batch_size).await?
            } else {
                0
            };

            tx.commit().await?;

            (completed_deleted, incomplete_deleted, dead_deleted)
        };

    Ok(InboxPurgeReport::new(
        completed_deleted,
        incomplete_deleted,
        dead_deleted,
    ))
}

/// The completed-row delete, bounded by `batch_size`. The outer `WHERE` repeats the subselect's
/// own predicate **in full**, the same `EvalPlanQual` guard `reliar-store-postgres`'s outbox
/// `purge_published_rows` uses — see that function's doc for why the age comparison is repeated,
/// not only nullness. Keys on `id` now that `id` is the primary key.
async fn purge_completed_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    retention_ms: i64,
    batch_size: i64,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"DELETE FROM inbox WHERE id IN (
               SELECT id FROM inbox
                WHERE completed_at IS NOT NULL
                  AND completed_at < now() - ($1::bigint * interval '1 millisecond')
                LIMIT $2)
             AND completed_at IS NOT NULL
             AND completed_at < now() - ($1::bigint * interval '1 millisecond')"#,
        retention_ms,
        batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// The incomplete-row delete (never completed, not dead — the third, disjoint retention category,
/// ADR 0042 Amendment C.2), bounded by `batch_size`. `dead_at IS NULL` keeps the incomplete and
/// dead categories disjoint; there is deliberately **no** `attempts > 0` clause — the three
/// retention categories partition the table, so an uncompleted, non-dead row with `attempts = 0`
/// (a bare `claim` committed without `complete`, or one `InboxDeadLetters::retry_dead` just
/// un-deaded) ages by `updated_at` exactly like one that recorded failures, rather than being
/// left uncollectable by every category. No index backs this sweep — see the migration's own
/// comment for why an opt-in, `None`-by-default, small-by-construction sweep does not earn one.
async fn purge_incomplete_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    retention_ms: i64,
    batch_size: i64,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"DELETE FROM inbox WHERE id IN (
               SELECT id FROM inbox
                WHERE completed_at IS NULL AND dead_at IS NULL
                  AND updated_at < now() - ($1::bigint * interval '1 millisecond')
                LIMIT $2)
             AND completed_at IS NULL AND dead_at IS NULL
             AND updated_at < now() - ($1::bigint * interval '1 millisecond')"#,
        retention_ms,
        batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// The dead-row delete, bounded by `batch_size`, keyed on `ix_inbox_dead` (ADR 0042 A.2.6).
async fn purge_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    retention_ms: i64,
    batch_size: i64,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"DELETE FROM inbox WHERE id IN (
               SELECT id FROM inbox
                WHERE dead_at IS NOT NULL
                  AND dead_at < now() - ($1::bigint * interval '1 millisecond')
                LIMIT $2)
             AND dead_at IS NOT NULL
             AND dead_at < now() - ($1::bigint * interval '1 millisecond')"#,
        retention_ms,
        batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}