reliar-store-postgres 0.9.0

//! `claim`'s three statements (inbox contract §3.1): the in-flight advisory-lock guard, the
//! `INSERT … ON CONFLICT DO NOTHING` claim, and the state read/upsert fallback.

/// [`try_advisory_lock`]'s bind parameters.
pub(in crate::inbox) struct TryAdvisoryLockParams<'a> {
    pub(in crate::inbox) class: i32,

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

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

/// The in-flight guard (inbox contract §3.1/ADR 0042 §3): a two-argument
/// `pg_try_advisory_xact_lock`, released by the caller's own commit or rollback. `hashtext` is a
/// built-in PostgreSQL function (a 32-bit hash of its `text` argument) used here only to fold
/// `scope || '/' || message_id` into the single `int4` the two-argument lock takes as its key —
/// not a cryptographic hash, and not indexed.
pub(in crate::inbox) async fn try_advisory_lock<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: TryAdvisoryLockParams<'_>,
) -> Result<bool, sqlx::Error> {
    sqlx::query_scalar!(
        r#"SELECT pg_try_advisory_xact_lock($1, hashtext($2 || '/' || $3::uuid::text)) AS "acquired!""#,
        params.class,
        params.scope,
        params.message_id,
    )
    .fetch_one(executor)
    .await
}

/// The bind parameters shared by [`insert_claim_row`] and [`upsert_claim_row`] — both bind the
/// same eight columns (layout Part II §8.1, carve-out 2).
pub(in crate::inbox) struct ClaimRowParams<'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>,
}

/// One row of [`select_claim_state`]/[`upsert_claim_row`]'s statements.
pub(in crate::inbox) struct ClaimStateRow {
    pub(in crate::inbox) id: uuid::Uuid,

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

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

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

/// Step 2's `INSERT … ON CONFLICT DO NOTHING` — the redelivery hot path, never reused by step 3
/// (ADR 0042 Amendment C.8): a redelivery storm conflicts here and answers `AlreadyCompleted` with
/// no further write, so paying `DO NOTHING`'s conflict cost on every redelivery is the trade §3
/// makes deliberately. Returns the inserted row's `attempts` on success, `None` on a conflict.
/// `id` is client-minted (ADR 0042 A.2.1); `ON CONFLICT (scope, message_id)` infers from
/// `ix_inbox_scope_message_id`.
pub(in crate::inbox) async fn insert_claim_row<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: ClaimRowParams<'_>,
) -> Result<Option<i32>, sqlx::Error> {
    sqlx::query_scalar!(
        r#"INSERT INTO inbox (id, scope, message_id, message_type, message_version,
                               conversation_id, correlation_id, causation_id)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
           ON CONFLICT (scope, message_id) DO NOTHING
           RETURNING attempts"#,
        params.id,
        params.scope,
        params.message_id,
        params.message_type,
        params.message_version,
        params.conversation_id,
        params.correlation_id,
        params.causation_id,
    )
    .fetch_optional(executor)
    .await
}

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

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

/// Step 3's state read, reached only when step 2 conflicted: the key was committed a moment ago,
/// so decide from its state. `fetch_optional`, not `fetch_one`: a concurrent `purge` can delete
/// the row between step 2's conflict and this read's own READ COMMITTED snapshot.
pub(in crate::inbox) async fn select_claim_state<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: SelectClaimStateParams<'_>,
) -> Result<Option<ClaimStateRow>, sqlx::Error> {
    sqlx::query_as!(
        ClaimStateRow,
        r#"SELECT id, attempts, completed_at, dead_at FROM inbox WHERE scope = $1 AND message_id = $2"#,
        params.scope,
        params.message_id,
    )
    .fetch_optional(executor)
    .await
}

/// Step 3's fallback re-insert (ADR 0042 Amendment C.8), reached only when step 2 conflicted and
/// step 3's own [`select_claim_state`] then found the key gone — a concurrent `purge` deleted it
/// after the conflict but before this session's read. Upserts and reads in one statement, `DO
/// UPDATE` with no `WHERE` clause so PostgreSQL's documented atomic insert-or-update guarantee
/// applies: exactly one row is always returned. `SET updated_at = inbox.updated_at` is an
/// identity assignment — this path transitions nothing, but `DO UPDATE` requires a `SET` — so it
/// costs a HOT update on a row already in the buffer pool, on a path that has already lost a race
/// to `purge`. On the conflict branch `RETURNING id` is the **existing** row's id, never the `id`
/// this call minted.
pub(in crate::inbox) async fn upsert_claim_row<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: ClaimRowParams<'_>,
) -> Result<ClaimStateRow, sqlx::Error> {
    sqlx::query_as!(
        ClaimStateRow,
        r#"INSERT INTO inbox (id, scope, message_id, message_type, message_version,
                               conversation_id, correlation_id, causation_id)
           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
           ON CONFLICT (scope, message_id) DO UPDATE
              SET updated_at = inbox.updated_at
           RETURNING id, attempts, completed_at, dead_at"#,
        params.id,
        params.scope,
        params.message_id,
        params.message_type,
        params.message_version,
        params.conversation_id,
        params.correlation_id,
        params.causation_id,
    )
    .fetch_one(executor)
    .await
}