reliar-store-postgres 0.9.0

//! `complete`/`fail`'s statements.

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

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

/// `complete`'s `UPDATE`, in the caller's own transaction. `AND dead_at IS NULL` is
/// load-bearing, not defensive: `ck_inbox_terminal` would otherwise turn "complete a dead row"
/// into a raw check-constraint error instead of a clean `NotClaimed` (ADR 0042 A.2.4). Returns
/// rows affected — the caller decides `NotClaimed` from `0`.
pub(in crate::inbox) async fn complete_row(
    executor: &mut sqlx::PgConnection,
    params: CompleteRowParams<'_>,
) -> Result<u64, sqlx::Error> {
    let result = sqlx::query!(
        r#"UPDATE inbox SET completed_at = now(), updated_at = now()
            WHERE scope = $1 AND message_id = $2 AND completed_at IS NULL AND dead_at IS NULL"#,
        params.scope,
        params.message_id,
    )
    .execute(executor)
    .await?;

    Ok(result.rows_affected())
}

/// [`fail_row`]'s bind parameters. `id` is client-minted by the caller (a fresh row's id, used
/// only on the insert branch of the statement's `ON CONFLICT`) and `max_attempts` is
/// [`crate::PostgresInboxSettings::max_attempts`], read by the store layer and passed through as
/// data — the connection concern's own knob never reaches this concern function as anything but a
/// bind parameter.
pub(in crate::inbox) struct FailRowParams<'a> {
    pub(in crate::inbox) id: uuid::Uuid,

    pub(in crate::inbox) scope: &'a str,

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

    pub(in crate::inbox) message_type: &'a str,

    pub(in crate::inbox) message_version: i32,

    pub(in crate::inbox) conversation_id: uuid::Uuid,

    pub(in crate::inbox) correlation_id: Option<&'a str>,

    pub(in crate::inbox) causation_id: Option<uuid::Uuid>,

    pub(in crate::inbox) last_error: &'a str,

    pub(in crate::inbox) max_attempts: i32,
}

/// [`fail_row`]'s row shape, named via `query_as!` (never `FromRow`).
pub(in crate::inbox) struct FailedRow {
    pub(in crate::inbox) id: uuid::Uuid,

    pub(in crate::inbox) attempts: i32,

    pub(in crate::inbox) dead_at: Option<time::OffsetDateTime>,
}

/// The `INSERT … ON CONFLICT DO UPDATE … RETURNING` (inbox contract §3.1): the dead transition
/// is computed in SQL (`attempts + 1 >= max_attempts`) so it is atomic with the increment — two
/// concurrent `fail`s cannot race the transition. `$9` binds `max_attempts` for both branches:
/// the insert branch must apply the bound too, or `max_attempts = 1` would never die on its
/// first `fail`. `dead_at`'s `COALESCE(inbox.dead_at, CASE …)` keeps the row's **original**
/// `dead_at` once it is set: without it, a later `fail` on an already-dead row (`attempts + 1 >=
/// max_attempts` still holds) would re-stamp `dead_at = now()` every time, moving the recorded
/// death time forward on every subsequent failed attempt.
pub(in crate::inbox) async fn fail_row<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: FailRowParams<'_>,
) -> Result<Option<FailedRow>, sqlx::Error> {
    sqlx::query_as!(
        FailedRow,
        r#"INSERT INTO inbox (id, scope, message_id, message_type, message_version,
                               conversation_id, correlation_id, causation_id,
                               attempts, last_error, dead_at)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8,
                   1, $9, CASE WHEN $10::integer <= 1 THEN now() END)
           ON CONFLICT (scope, message_id) DO UPDATE
              SET attempts   = inbox.attempts + 1,
                  last_error = excluded.last_error,
                  dead_at    = COALESCE(inbox.dead_at,
                                 CASE WHEN inbox.attempts + 1 >= $10::integer THEN now() END),
                  updated_at = now()
            WHERE inbox.completed_at IS NULL
           RETURNING id, attempts, dead_at"#,
        params.id,
        params.scope,
        params.message_id,
        params.message_type,
        params.message_version,
        params.conversation_id,
        params.correlation_id,
        params.causation_id,
        params.last_error,
        params.max_attempts,
    )
    .fetch_optional(executor)
    .await
}