reliar-store-postgres 0.9.0

use sqlx::PgPool;

/// Whether the row whose `message_id` is `id` is currently held by a claim (`locked_by IS NOT
/// NULL`) — the "has it been claimed" half of what a single `locked_until` read used to answer
/// before ADR 0050 dropped that column; `available_at` is `NOT NULL`, so there is no longer an
/// `Option` to test for that question.
///
/// Filters on `message_id`, not `id` (ADR 0044 §1): every call site holds the envelope's
/// `MessageId` (from `common::seed`'s returned envelopes or a captured publish), never the row's
/// own database-assigned surrogate id.
pub(crate) async fn claimed(pool: &PgPool, id: uuid::Uuid) -> bool {
    sqlx::query_scalar::<_, Option<String>>("SELECT locked_by FROM outbox WHERE message_id = $1")
        .bind(id)
        .fetch_one(pool)
        .await
        .unwrap()
        .is_some()
}

/// Reads the current `available_at` of the row whose `message_id` is `id` — since ADR 0050 this
/// **is** the lease end for a leased row, the lease clock and the only one. The read half of the
/// lease-renewal probes (ADR 0043 §5): a real-time *eventually*/*never* assertion reads this twice
/// around a renewal tick rather than inferring renewal from whether the row is claimable, since
/// claimability also changes the instant a row completes (§43 review round 1, B2/B3).
///
/// Filters on `message_id`, see [`claimed`].
pub(crate) async fn lease_end(pool: &PgPool, id: uuid::Uuid) -> time::OffsetDateTime {
    sqlx::query_scalar("SELECT available_at FROM outbox WHERE message_id = $1")
        .bind(id)
        .fetch_one(pool)
        .await
        .unwrap()
}

/// Moves the row whose `message_id` is `id`'s lease into the past — SQL time-travel for
/// lease-expiry tests, never a wall-clock sleep (§8.2). Since ADR 0050 there is no second lease
/// column: `available_at` **is** the lease clock, so this is now the identical statement as
/// [`make_available_now`] — kept as its own name because the two describe different intents at
/// the call site ("this lease has lapsed" vs. "this backoff is due"), even though both now move
/// the one clock they share.
pub(crate) async fn expire_lease(pool: &PgPool, id: uuid::Uuid) {
    make_available_now(pool, id).await;
}

/// Moves the `available_at` of the row whose `message_id` is `id` into the past — makes a
/// retry-delayed row due without waiting. Filters on `message_id`, see [`claimed`].
pub(crate) async fn make_available_now(pool: &PgPool, id: uuid::Uuid) {
    sqlx::query(
        "UPDATE outbox SET available_at = now() - interval '1 second' WHERE message_id = $1",
    )
    .bind(id)
    .execute(pool)
    .await
    .unwrap();
}

/// Ages an already-committed inbox row's `updated_at` by `age` (a Postgres interval literal, e.g.
/// `"30 days"`) — SQL time-travel for retention-ageing and in-flight-row scenarios (ADR 0043 §5),
/// promoted from the identical inline `UPDATE` three inbox scenario files each wrote by hand.
/// Never used to move `completed_at`/`dead_at` themselves — those are set at seed time by the
/// caller's own `INSERT` when the seeded state (not the ageing) is what a test is proving.
pub(crate) async fn age_inbox_updated_at(pool: &PgPool, id: uuid::Uuid, age: &str) {
    sqlx::query("UPDATE inbox SET updated_at = now() - $2::text::interval WHERE id = $1")
        .bind(id)
        .bind(age)
        .execute(pool)
        .await
        .unwrap();
}

/// ADR 0046 Correction B.3's shared post-condition: no row may carry a `claim_token` once its
/// `locked_by` has been cleared. Every statement that ends a claim (B.1's closed list — the four
/// outcome writes, the poison sweep, the expiry sweep, `retry_dead`) clears both columns in the
/// same statement, so a row failing this check means one of those writers forgot to clear the
/// token, not that the token is merely stale (a stale-but-still-owned token is exactly what the
/// fence is for and is not what this asserts). Call at the end of any trial that exercises
/// `purge`, `retry_dead` or an outcome write.
pub(crate) async fn assert_no_orphan_claim_tokens(pool: &PgPool) {
    let orphans: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM outbox WHERE claim_token IS NOT NULL AND locked_by IS NULL",
    )
    .fetch_one(pool)
    .await
    .unwrap();

    assert_eq!(
        orphans, 0,
        "found {orphans} outbox row(s) with a claim_token but no locked_by — a claim-ending \
         write left the token behind (ADR 0046 Correction B.1)"
    );
}