reliar-store-postgres 0.9.0

//! `purge`/`find`'s statements: the inbox contract's retention sweep and diagnostic read.

use super::rows::InboxRow;

/// [`find_row`]'s bind parameters.
pub(in crate::inbox) struct FindRowParams<'a> {
    pub(in crate::inbox) scope: &'a str,

    pub(in crate::inbox) message_id: uuid::Uuid,
}

/// `find`'s query — diagnostics and tests only, no Reliar code path calls it.
pub(in crate::inbox) async fn find_row<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: FindRowParams<'_>,
) -> 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"#,
        params.scope,
        params.message_id,
    )
    .fetch_optional(executor)
    .await
}

/// [`purge_completed_rows`]'s bind parameters.
pub(in crate::inbox) struct PurgeCompletedRowsParams {
    pub(in crate::inbox) retention_ms: i64,

    pub(in crate::inbox) batch_size: i64,
}

/// 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.
pub(in crate::inbox) async fn purge_completed_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgeCompletedRowsParams,
) -> 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')"#,
        params.retention_ms,
        params.batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// [`purge_incomplete_rows`]'s bind parameters.
pub(in crate::inbox) struct PurgeIncompleteRowsParams {
    pub(in crate::inbox) retention_ms: i64,

    pub(in crate::inbox) batch_size: i64,
}

/// 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.
pub(in crate::inbox) async fn purge_incomplete_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgeIncompleteRowsParams,
) -> 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')"#,
        params.retention_ms,
        params.batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// [`purge_dead_retention_rows`]'s bind parameters.
pub(in crate::inbox) struct PurgeDeadRetentionRowsParams {
    pub(in crate::inbox) retention_ms: i64,

    pub(in crate::inbox) batch_size: i64,
}

/// The dead-row retention delete, bounded by `batch_size`, keyed on `ix_inbox_dead` (ADR 0042
/// A.2.6). Renamed from `purge_dead_rows` (layout Part II §8.1) to end its name collision with
/// `dead_letters::purge_dead_rows`, a different statement (purge by reference, not by
/// retention).
pub(in crate::inbox) async fn purge_dead_retention_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgeDeadRetentionRowsParams,
) -> 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')"#,
        params.retention_ms,
        params.batch_size,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}