reliar-store-postgres 0.9.0

//! `OutboxDeadLetters`'s three statements: `list_dead`/`retry_dead`/`purge_dead`.

/// [`list_dead_rows`]'s bind parameters.
pub(in crate::outbox) struct ListDeadRowsParams<'a> {
    pub(in crate::outbox) message_type: Option<&'a str>,

    pub(in crate::outbox) tenant_id: Option<&'a str>,

    pub(in crate::outbox) dead_before: Option<time::OffsetDateTime>,

    pub(in crate::outbox) after_dead_at: Option<time::OffsetDateTime>,

    pub(in crate::outbox) after_id: Option<uuid::Uuid>,

    pub(in crate::outbox) limit: i64,
}

/// `list_dead`'s query. Named against [`crate::records::RawRow`] via `query_as!`, same as the
/// claim — the `SELECT` list matches its field order exactly. **`ORDER BY dead_at ASC, id ASC` is
/// normative**: `after` is a composite keyset cursor over the columns `ix_outbox_dead_cursor`
/// orders by.
pub(in crate::outbox) async fn list_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: ListDeadRowsParams<'_>,
) -> Result<Vec<crate::records::RawRow>, sqlx::Error> {
    sqlx::query_as!(
        crate::records::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, claim_token,
                  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"#,
        params.message_type,
        params.tenant_id,
        params.dead_before,
        params.after_dead_at,
        params.after_id,
        params.limit,
    )
    .fetch_all(executor)
    .await
}

/// [`retry_dead_rows`]'s bind parameters.
pub(in crate::outbox) struct RetryDeadRowsParams<'a> {
    pub(in crate::outbox) ids: &'a [uuid::Uuid],
}

/// `retry_dead`'s query. Not fenced by `claim_token` — a dead row holds no lease, so there is no
/// owner to check against. Clears `claim_token` alongside `locked_by` so the stale token a
/// pre-death claim left behind cannot fence a stale `complete`/`fail` into the resurrected row.
pub(in crate::outbox) async fn retry_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: RetryDeadRowsParams<'_>,
) -> 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,
                  claim_token  = NULL,
                  updated_at   = now()
            WHERE id = ANY($1) AND dead_at IS NOT NULL"#,
        params.ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// [`purge_dead_rows`]'s bind parameters.
pub(in crate::outbox) struct PurgeDeadRowsParams<'a> {
    pub(in crate::outbox) ids: &'a [uuid::Uuid],
}

/// `purge_dead`'s query: deletes dead rows by reference, regardless of retention.
pub(in crate::outbox) async fn purge_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: PurgeDeadRowsParams<'_>,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        "DELETE FROM outbox WHERE id = ANY($1) AND dead_at IS NOT NULL",
        params.ids,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}