reliar-store-postgres 0.8.0

use sqlx::PgPool;

/// Reads the current `locked_until` of the row whose `message_id` is `id` — 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`, 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 locked_until(pool: &PgPool, id: uuid::Uuid) -> Option<time::OffsetDateTime> {
    sqlx::query_scalar("SELECT locked_until 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). Moves **both** `locked_until` and
/// `available_at` (ADR 0040 §1: a claim now writes them to the same instant, so simulating "the
/// lease has lapsed" means moving both — leaving `available_at` at its claimed value would leave
/// the row invisible to the claim's own `available_at <= now()` boundary even though
/// `locked_until` has passed). Filters on `message_id`, see [`locked_until`].
pub(crate) async fn expire_lease(pool: &PgPool, id: uuid::Uuid) {
    sqlx::query(
        "UPDATE outbox SET locked_until = now() - interval '1 second', \
                            available_at = now() - interval '1 second' WHERE message_id = $1",
    )
    .bind(id)
    .execute(pool)
    .await
    .unwrap();
}

/// 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 [`locked_until`].
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)"
    );
}