reliar-store-postgres 0.9.0

//! `complete`/`fail`/`release`/`extend_lease`'s statements — every one fenced by the claimed
//! row's `claim_token` (ADR 0046 Amendment A).

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

    pub(in crate::outbox) tokens: &'a [Option<uuid::Uuid>],
}

/// Fenced `complete`: clears the lease and sets `published_at`, only for rows whose stored
/// `claim_token` still matches the one this reference carries (ADR 0046 Amendment A) — a row
/// already reclaimed under a fresh token, or a reference built with no token at all
/// (`RecordRef::new`), contributes nothing (fail-closed by construction: `NULL = token` is
/// `NULL`, never `true`).
pub(in crate::outbox) async fn complete_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: CompleteRowsParams<'_>,
) -> Result<Vec<uuid::Uuid>, sqlx::Error> {
    let rows = sqlx::query!(
        r#"UPDATE outbox o
              SET published_at = now(),
                  attempts     = o.attempts + 1,
                  locked_by    = NULL,
                  claim_token  = NULL,
                  updated_at   = now()
             FROM UNNEST($1::uuid[], $2::uuid[]) AS f(id, token)
            WHERE o.id = f.id AND o.claim_token = f.token
        RETURNING o.id"#,
        params.ids,
        params.tokens as &[Option<uuid::Uuid>],
    )
    .fetch_all(executor)
    .await?;

    Ok(rows.into_iter().map(|r| r.id).collect())
}

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

    pub(in crate::outbox) tokens: &'a [Option<uuid::Uuid>],
}

/// Fenced release: clears the lease and resets `available_at = now()` (ADR 0040 §1) — a released
/// row must be claimable at once, not stalled behind the lease it was just released from.
pub(in crate::outbox) async fn release_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: ReleaseRowsParams<'_>,
) -> Result<Vec<uuid::Uuid>, sqlx::Error> {
    let rows = sqlx::query!(
        r#"UPDATE outbox o
              SET locked_by    = NULL,
                  claim_token  = NULL,
                  available_at = now(),
                  updated_at   = now()
             FROM UNNEST($1::uuid[], $2::uuid[]) AS f(id, token)
            WHERE o.id = f.id AND o.claim_token = f.token
        RETURNING o.id"#,
        params.ids,
        params.tokens as &[Option<uuid::Uuid>],
    )
    .fetch_all(executor)
    .await?;

    Ok(rows.into_iter().map(|r| r.id).collect())
}

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

    pub(in crate::outbox) tokens: &'a [Option<uuid::Uuid>],

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

/// Advances `available_at` — the lease clock, and the only one (ADR 0050 §1) — without rotating
/// `claim_token` — a renewal extends the same claim, it is not a new one (ADR 0046 Amendment A).
/// Touches neither `locked_by` nor `claim_token`: this is the one statement that does not.
pub(in crate::outbox) async fn extend_lease_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: ExtendLeaseRowsParams<'_>,
) -> Result<Vec<uuid::Uuid>, sqlx::Error> {
    let rows = sqlx::query!(
        r#"UPDATE outbox o
              SET available_at = now() + ($3::bigint * interval '1 millisecond'),
                  updated_at   = now()
             FROM UNNEST($1::uuid[], $2::uuid[]) AS f(id, token)
            WHERE o.id = f.id AND o.claim_token = f.token
        RETURNING o.id"#,
        params.ids,
        params.tokens as &[Option<uuid::Uuid>],
        params.lease_ms,
    )
    .fetch_all(executor)
    .await?;

    Ok(rows.into_iter().map(|r| r.id).collect())
}

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

    pub(in crate::outbox) tokens: &'a [Option<uuid::Uuid>],

    pub(in crate::outbox) errors: &'a [String],

    pub(in crate::outbox) delays_ms: &'a [i64],
}

pub(in crate::outbox) async fn fail_retry_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: FailRetryRowsParams<'_>,
) -> Result<Vec<uuid::Uuid>, sqlx::Error> {
    let rows = sqlx::query!(
        r#"UPDATE outbox o
              SET attempts     = o.attempts + 1,
                  last_error   = f.err,
                  locked_by    = NULL,
                  claim_token  = NULL,
                  available_at = now() + (f.delay_ms * interval '1 millisecond'),
                  updated_at   = now()
             FROM UNNEST($1::uuid[], $2::uuid[], $3::text[], $4::bigint[]) AS f(id, token, err, delay_ms)
            WHERE o.id = f.id AND o.claim_token = f.token
        RETURNING o.id"#,
        params.ids,
        params.tokens as &[Option<uuid::Uuid>],
        params.errors,
        params.delays_ms,
    )
    .fetch_all(executor)
    .await?;

    Ok(rows.into_iter().map(|r| r.id).collect())
}

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

    pub(in crate::outbox) tokens: &'a [Option<uuid::Uuid>],

    pub(in crate::outbox) errors: &'a [String],

    pub(in crate::outbox) reasons: &'a [&'a str],
}

pub(in crate::outbox) async fn fail_dead_rows<'e>(
    executor: impl sqlx::PgExecutor<'e>,
    params: FailDeadRowsParams<'_>,
) -> Result<Vec<uuid::Uuid>, sqlx::Error> {
    let rows = sqlx::query!(
        r#"UPDATE outbox o
              SET attempts     = o.attempts + 1,
                  last_error   = f.err,
                  dead_at      = now(),
                  dead_reason  = f.reason,
                  locked_by    = NULL,
                  claim_token  = NULL,
                  updated_at   = now()
             FROM UNNEST($1::uuid[], $2::uuid[], $3::text[], $4::text[]) AS f(id, token, err, reason)
            WHERE o.id = f.id AND o.claim_token = f.token
        RETURNING o.id"#,
        params.ids,
        params.tokens as &[Option<uuid::Uuid>],
        params.errors,
        params.reasons as &[&str],
    )
    .fetch_all(executor)
    .await?;

    Ok(rows.into_iter().map(|r| r.id).collect())
}