use std::sync::Arc;
use std::time::Duration;

use remora_domain::ports::MutationScope;
use remora_domain::{OpaqueId, fingerprint};
use remora_durable::{
    ADVISORY_CLASS_APP, CursorCodec, DurableError, KeysetPosition, MutationDisposition,
    MutationStatus, OutboxClaim, OutboxClaimFilter, PoolConfig, advisory_key, advisory_xact_lock,
    cas_set_text, claim_batch, claim_batch_in_tx, connect, enqueue, enqueue_after,
    execute_mutation, first_page_cursor, mark_projected, mark_projected_pool, migrate,
    next_page_cursor, release_with_error, unprojected_count, with_transaction,
};
use serde_json::json;
use sqlx::PgPool;
use time::{Duration as TimeDuration, OffsetDateTime};
use tokio::sync::Barrier;

fn database_url() -> Option<String> {
    std::env::var("DATABASE_URL")
        .ok()
        .filter(|value| !value.is_empty())
}

async fn test_pool() -> Option<PgPool> {
    let Some(url) = database_url() else {
        if std::env::var("REMORA_DURABLE_REQUIRE_DB").is_ok() {
            panic!("DATABASE_URL is required when REMORA_DURABLE_REQUIRE_DB is set");
        }
        eprintln!("skipping durable substrate integration test: DATABASE_URL is unset");
        return None;
    };
    let pool = connect(
        &url,
        PoolConfig {
            max_connections: 8,
            acquire_timeout: Duration::from_secs(10),
            idle_timeout: Some(Duration::from_secs(30)),
        },
    )
    .await
    .expect("connect to postgres");
    migrate(&pool).await.expect("run substrate migrations");
    Some(pool)
}

fn subject(label: &str) -> OpaqueId {
    OpaqueId::parse(format!("subject-{label}-aaaaaaaa")).expect("opaque id")
}

fn scope<'a>(subject: &'a OpaqueId, operation: &'a str, key: &'a str) -> MutationScope<'a> {
    MutationScope {
        subject,
        operation,
        key,
    }
}

/// Example authority binding: attempt, fence, worker incarnation.
fn eligibility(attempt: &str, fence: u64, worker: &str) -> Vec<u8> {
    format!("attempt={attempt}|fence={fence}|worker={worker}").into_bytes()
}

async fn ensure_effect_probe_table(pool: &PgPool) {
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS substrate_effect_probe (
            id TEXT PRIMARY KEY,
            hits BIGINT NOT NULL
         )",
    )
    .execute(pool)
    .await
    .expect("create effect probe table");
}

async fn clear_mutation_subject(pool: &PgPool, owner: &OpaqueId) {
    sqlx::query("DELETE FROM mutation_records WHERE scope_subject = $1")
        .bind(owner.as_str())
        .execute(pool)
        .await
        .expect("clear mutation rows for subject");
}

#[tokio::test]
async fn migration_round_trip_is_idempotent() {
    let Some(pool) = test_pool().await else {
        return;
    };
    migrate(&pool).await.expect("second migrate");
    let (table_count,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*)::bigint
         FROM information_schema.tables
         WHERE table_schema = 'public'
           AND table_name IN ('mutation_records', 'outbox_messages')",
    )
    .fetch_one(&pool)
    .await
    .expect("count substrate tables");
    assert_eq!(table_count, 2);

    let applied = remora_durable::migrator().iter().count();
    assert_eq!(applied, 1);
    let (version_count,): (i64,) = sqlx::query_as("SELECT COUNT(*)::bigint FROM _sqlx_migrations")
        .fetch_one(&pool)
        .await
        .expect("count migrations");
    assert!(version_count >= 1);
}

#[tokio::test]
async fn transaction_rollback_discards_writes() {
    let Some(pool) = test_pool().await else {
        return;
    };
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS substrate_tx_probe (
            id TEXT PRIMARY KEY,
            value TEXT NOT NULL
         )",
    )
    .execute(&pool)
    .await
    .expect("create probe table");
    sqlx::query("DELETE FROM substrate_tx_probe WHERE id = 'rollback-row'")
        .execute(&pool)
        .await
        .expect("clean probe row");

    let result = with_transaction(&pool, |tx| {
        Box::pin(async move {
            sqlx::query("INSERT INTO substrate_tx_probe (id, value) VALUES ($1, $2)")
                .bind("rollback-row")
                .bind("visible-only-if-committed")
                .execute(tx.as_mut())
                .await?;
            Err::<(), _>(DurableError::Config("force rollback".into()))
        })
    })
    .await;
    assert!(result.is_err());

    let (count,): (i64,) =
        sqlx::query_as("SELECT COUNT(*)::bigint FROM substrate_tx_probe WHERE id = 'rollback-row'")
            .fetch_one(&pool)
            .await
            .expect("count probe row");
    assert_eq!(count, 0);
}

#[tokio::test]
async fn mutation_ledger_exact_replay_conflict_and_rejection() {
    let Some(pool) = test_pool().await else {
        return;
    };
    ensure_effect_probe_table(&pool).await;
    let owner = subject("replay");
    clear_mutation_subject(&pool, &owner).await;
    let operation = "submit";
    let key = "op-replay-0001";
    let request = json!({"source": "http", "bucket": "demo"});
    let digest = fingerprint(&request).expect("fingerprint");
    let claim = eligibility("attempt-aaaaaaaa", 1, "worker-aaaaaaaa");
    let probe = format!("effect-{key}");
    sqlx::query("DELETE FROM substrate_effect_probe WHERE id = $1")
        .bind(&probe)
        .execute(&pool)
        .await
        .expect("clear probe");

    let first = execute_mutation(
        &pool,
        scope(&owner, operation, key),
        "submit",
        &claim,
        digest,
        {
            let probe = probe.clone();
            move |tx| {
                Box::pin(async move {
                    apply_effect(tx, &probe).await?;
                    Ok((MutationStatus::Accepted, json!({"job_id": "job-abcdefgh"})))
                })
            }
        },
    )
    .await
    .expect("first observation");
    assert_eq!(first.disposition, MutationDisposition::First);
    assert_eq!(first.status, MutationStatus::Accepted);
    assert_eq!(first.response, json!({"job_id": "job-abcdefgh"}));

    let replay = execute_mutation(
        &pool,
        scope(&owner, operation, key),
        "submit",
        &claim,
        digest,
        {
            let probe = probe.clone();
            move |tx| {
                Box::pin(async move {
                    apply_effect(tx, &probe).await?;
                    Ok((
                        MutationStatus::Accepted,
                        json!({"job_id": "should-not-write"}),
                    ))
                })
            }
        },
    )
    .await
    .expect("exact replay");
    assert_eq!(replay.disposition, MutationDisposition::Replay);
    assert_eq!(replay.status, MutationStatus::Accepted);
    assert_eq!(replay.response, json!({"job_id": "job-abcdefgh"}));

    let changed = fingerprint(&json!({"source": "http", "bucket": "other"})).expect("fp");
    let conflict = execute_mutation(
        &pool,
        scope(&owner, operation, key),
        "submit",
        &claim,
        changed,
        {
            let probe = probe.clone();
            move |tx| {
                Box::pin(async move {
                    apply_effect(tx, &probe).await?;
                    Ok((MutationStatus::Accepted, json!({"job_id": "conflict"})))
                })
            }
        },
    )
    .await
    .expect("conflict observation");
    assert_eq!(conflict.disposition, MutationDisposition::Conflict);
    assert_eq!(conflict.status, MutationStatus::Rejected);
    assert!(conflict.response.is_null());

    let kind_conflict = execute_mutation(
        &pool,
        scope(&owner, operation, key),
        "cancel",
        &claim,
        digest,
        {
            let probe = probe.clone();
            move |tx| {
                Box::pin(async move {
                    apply_effect(tx, &probe).await?;
                    Ok((MutationStatus::Accepted, json!({"ok": true})))
                })
            }
        },
    )
    .await
    .expect("kind conflict");
    assert_eq!(kind_conflict.disposition, MutationDisposition::Conflict);

    // Same kind and payload fingerprint with a different authority claim must
    // not replay the stored result.
    let stale_authority = eligibility("attempt-bbbbbbbb", 2, "worker-bbbbbbbb");
    let eligibility_conflict = execute_mutation(
        &pool,
        scope(&owner, operation, key),
        "submit",
        &stale_authority,
        digest,
        {
            let probe = probe.clone();
            move |tx| {
                Box::pin(async move {
                    apply_effect(tx, &probe).await?;
                    Ok((MutationStatus::Accepted, json!({"job_id": "stale"})))
                })
            }
        },
    )
    .await
    .expect("eligibility conflict");
    assert_eq!(
        eligibility_conflict.disposition,
        MutationDisposition::Conflict
    );

    let (hits,): (i64,) = sqlx::query_as("SELECT hits FROM substrate_effect_probe WHERE id = $1")
        .bind(&probe)
        .fetch_one(&pool)
        .await
        .expect("effect hits");
    assert_eq!(hits, 1, "effect must apply at most once");

    let reject_key = "op-reject-0001";
    let reject_digest = fingerprint(&json!({"op": "reject"})).expect("fp");
    let reject_claim = eligibility("attempt-cccccccc", 1, "worker-cccccccc");
    let rejected = execute_mutation(
        &pool,
        scope(&owner, operation, reject_key),
        "submit",
        &reject_claim,
        reject_digest,
        |_tx| {
            Box::pin(async move {
                Ok((
                    MutationStatus::Rejected,
                    json!({"error": "precondition_failed"}),
                ))
            })
        },
    )
    .await
    .expect("rejected first observation");
    assert_eq!(rejected.disposition, MutationDisposition::First);
    assert_eq!(rejected.status, MutationStatus::Rejected);

    let rejected_replay = execute_mutation(
        &pool,
        scope(&owner, operation, reject_key),
        "submit",
        &reject_claim,
        reject_digest,
        |_tx| {
            Box::pin(async move { Ok((MutationStatus::Accepted, json!({"should": "not apply"}))) })
        },
    )
    .await
    .expect("rejected replay");
    assert_eq!(rejected_replay.disposition, MutationDisposition::Replay);
    assert_eq!(rejected_replay.status, MutationStatus::Rejected);
    assert_eq!(
        rejected_replay.response,
        json!({"error": "precondition_failed"})
    );
}

#[tokio::test]
async fn concurrent_first_observation_serializes_to_one_effect() {
    let Some(pool) = test_pool().await else {
        return;
    };
    ensure_effect_probe_table(&pool).await;
    let owner = subject("race");
    clear_mutation_subject(&pool, &owner).await;
    let operation = "submit";
    let key = "op-race-0001";
    let digest = fingerprint(&json!({"race": true})).expect("fingerprint");
    let claim = eligibility("attempt-raceaaaa", 1, "worker-raceaaaa");
    let probe = format!("effect-{key}");
    sqlx::query("DELETE FROM substrate_effect_probe WHERE id = $1")
        .bind(&probe)
        .execute(&pool)
        .await
        .expect("clear probe");
    let workers = 8_usize;
    let barrier = Arc::new(Barrier::new(workers));
    let mut handles = Vec::with_capacity(workers);

    for _ in 0..workers {
        let pool = pool.clone();
        let barrier = Arc::clone(&barrier);
        let owner = owner.clone();
        let probe = probe.clone();
        let claim = claim.clone();
        handles.push(tokio::spawn(async move {
            barrier.wait().await;
            execute_mutation(
                &pool,
                scope(&owner, operation, key),
                "submit",
                &claim,
                digest,
                {
                    let probe = probe.clone();
                    move |tx| {
                        Box::pin(async move {
                            apply_effect(tx, &probe).await?;
                            Ok((MutationStatus::Accepted, json!({"job_id": "job-race-01"})))
                        })
                    }
                },
            )
            .await
        }));
    }

    let mut first = 0_u32;
    let mut replay = 0_u32;
    for handle in handles {
        let outcome = handle.await.expect("join").expect("mutation observation");
        match outcome.disposition {
            MutationDisposition::First => {
                first += 1;
                assert_eq!(outcome.status, MutationStatus::Accepted);
                assert_eq!(outcome.response, json!({"job_id": "job-race-01"}));
            }
            MutationDisposition::Replay => {
                replay += 1;
                assert_eq!(outcome.status, MutationStatus::Accepted);
                assert_eq!(outcome.response, json!({"job_id": "job-race-01"}));
            }
            MutationDisposition::Conflict => {
                panic!("exact concurrent request must not conflict")
            }
        }
    }
    assert_eq!(first, 1);
    assert_eq!(replay, u32::try_from(workers - 1).expect("workers fit u32"));

    let (hits,): (i64,) = sqlx::query_as("SELECT hits FROM substrate_effect_probe WHERE id = $1")
        .bind(&probe)
        .fetch_one(&pool)
        .await
        .expect("effect hits");
    assert_eq!(hits, 1, "effect must apply at most once under contention");
}

async fn apply_effect(
    tx: &mut remora_durable::PgTx<'_>,
    probe_id: &str,
) -> Result<(), DurableError> {
    sqlx::query(
        "INSERT INTO substrate_effect_probe (id, hits) VALUES ($1, 1)
         ON CONFLICT (id) DO UPDATE SET hits = substrate_effect_probe.hits + 1",
    )
    .bind(probe_id)
    .execute(tx.as_mut())
    .await?;
    Ok(())
}

#[tokio::test]
async fn concurrent_cas_admits_exactly_one_writer() {
    let Some(pool) = test_pool().await else {
        return;
    };
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS substrate_cas_rows (
            id TEXT PRIMARY KEY,
            value TEXT NOT NULL,
            revision BIGINT NOT NULL
         )",
    )
    .execute(&pool)
    .await
    .expect("create cas table");
    sqlx::query(
        "INSERT INTO substrate_cas_rows (id, value, revision) VALUES ($1, $2, 0)
         ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value, revision = 0",
    )
    .bind("cas-row")
    .bind("initial")
    .execute(&pool)
    .await
    .expect("seed cas row");

    let barrier = Arc::new(Barrier::new(8));
    let mut handles = Vec::new();
    for index in 0..8 {
        let pool = pool.clone();
        let barrier = Arc::clone(&barrier);
        handles.push(tokio::spawn(async move {
            barrier.wait().await;
            with_transaction(&pool, |tx| {
                Box::pin(async move {
                    cas_set_text(
                        tx,
                        "substrate_cas_rows",
                        "id",
                        "cas-row",
                        "value",
                        &format!("writer-{index}"),
                        0,
                    )
                    .await
                })
            })
            .await
        }));
    }

    let mut successes = 0_u32;
    let mut conflicts = 0_u32;
    for handle in handles {
        match handle.await.expect("join") {
            Ok(revision) => {
                assert_eq!(revision, 1);
                successes += 1;
            }
            Err(DurableError::CasConflict) => conflicts += 1,
            Err(error) => panic!("unexpected CAS error: {error}"),
        }
    }
    assert_eq!(successes, 1);
    assert_eq!(conflicts, 7);

    let (value, revision): (String, i64) =
        sqlx::query_as("SELECT value, revision FROM substrate_cas_rows WHERE id = 'cas-row'")
            .fetch_one(&pool)
            .await
            .expect("read cas row");
    assert_eq!(revision, 1);
    assert!(value.starts_with("writer-"));
}

/// Namespace topics so concurrent cargo-nextest cases never claim each other.
fn outbox_topic(suite: &str, name: &str) -> String {
    format!("substrate.outbox.{suite}.{name}")
}

async fn clear_outbox_suite(pool: &PgPool, suite: &str) {
    let pattern = format!("substrate.outbox.{suite}.%");
    sqlx::query("DELETE FROM outbox_messages WHERE topic LIKE $1")
        .bind(pattern)
        .execute(pool)
        .await
        .expect("clear suite outbox rows");
}

#[tokio::test]
async fn outbox_claim_projection_recovery_and_claim_fencing() {
    let Some(pool) = test_pool().await else {
        return;
    };
    const SUITE: &str = "fencing";
    clear_outbox_suite(&pool, SUITE).await;
    let batch_topic = outbox_topic(SUITE, "batch");
    let recover_topic = outbox_topic(SUITE, "recover");
    let retry_topic = outbox_topic(SUITE, "retry");
    let batch_filter = OutboxClaimFilter::Topic(&batch_topic);
    let recover_filter = OutboxClaimFilter::Topic(&recover_topic);
    let retry_filter = OutboxClaimFilter::Topic(&retry_topic);

    with_transaction(&pool, {
        let batch_topic = batch_topic.clone();
        move |tx| {
            Box::pin(async move {
                enqueue(tx, &batch_topic, json!({"job_id": "job-outbox-1"})).await?;
                enqueue(tx, &batch_topic, json!({"job_id": "job-outbox-2"})).await?;
                enqueue(tx, &batch_topic, json!({"job_id": "job-outbox-3"})).await?;
                Ok(())
            })
        }
    })
    .await
    .expect("enqueue");

    let claimed = claim_batch(
        &pool,
        "worker-a",
        batch_filter,
        2,
        TimeDuration::seconds(30),
    )
    .await
    .expect("claim batch");
    assert_eq!(claimed.len(), 2);
    assert_eq!(claimed[0].attempt_count, 1);
    assert_eq!(claimed[0].claim.worker_id, "worker-a");
    assert_eq!(claimed[0].claim.generation, 1);

    let overlapping = claim_batch(
        &pool,
        "worker-b",
        batch_filter,
        10,
        TimeDuration::seconds(30),
    )
    .await
    .expect("second claim");
    assert_eq!(
        overlapping.len(),
        1,
        "lease exclusion leaves only the unclaimed remainder"
    );

    // Unclaimed or foreign claims cannot complete projection.
    let foreign = OutboxClaim {
        message_id: claimed[0].id,
        worker_id: "worker-intruder".into(),
        generation: claimed[0].claim.generation,
    };
    assert!(matches!(
        mark_projected_pool(&pool, &foreign).await,
        Err(DurableError::OutboxClaimMismatch { .. })
    ));
    let unclaimed = OutboxClaim {
        message_id: claimed[0].id,
        worker_id: "worker-a".into(),
        generation: 0,
    };
    assert!(matches!(
        mark_projected_pool(&pool, &unclaimed).await,
        Err(DurableError::OutboxClaimMismatch { .. })
    ));

    for message in &claimed {
        mark_projected_pool(&pool, &message.claim)
            .await
            .expect("mark projected");
        // Retry after success is idempotent for ambiguous-commit recovery.
        mark_projected_pool(&pool, &message.claim)
            .await
            .expect("idempotent mark projected");
    }
    mark_projected_pool(&pool, &overlapping[0].claim)
        .await
        .expect("mark remaining");

    assert_eq!(
        unprojected_count(&pool, batch_filter).await.expect("count"),
        0
    );

    let recovered_id = with_transaction(&pool, {
        let recover_topic = recover_topic.clone();
        move |tx| {
            Box::pin(
                async move { enqueue(tx, &recover_topic, json!({"job_id": "job-recover"})).await },
            )
        }
    })
    .await
    .expect("enqueue recovery");

    let first_claim = claim_batch(
        &pool,
        "worker-crash",
        recover_filter,
        1,
        TimeDuration::milliseconds(1),
    )
    .await
    .expect("claim before crash");
    assert_eq!(first_claim.len(), 1);
    assert_eq!(first_claim[0].id, recovered_id);
    let stale = first_claim[0].claim.clone();
    tokio::time::sleep(Duration::from_millis(20)).await;

    let reclaimed = claim_batch(
        &pool,
        "worker-recover",
        recover_filter,
        1,
        TimeDuration::seconds(30),
    )
    .await
    .expect("reclaim after lease expiry");
    assert_eq!(reclaimed.len(), 1);
    assert_eq!(reclaimed[0].id, recovered_id);
    assert_eq!(reclaimed[0].attempt_count, 2);
    assert_eq!(reclaimed[0].claim.generation, 2);
    assert_eq!(reclaimed[0].claim.worker_id, "worker-recover");

    // Stale claimant must not project or release after reclaim.
    assert!(matches!(
        mark_projected_pool(&pool, &stale).await,
        Err(DurableError::OutboxClaimMismatch { .. })
    ));
    with_transaction(&pool, |tx| {
        let stale = stale.clone();
        Box::pin(
            async move { release_with_error(tx, &stale, "stale", TimeDuration::seconds(1)).await },
        )
    })
    .await
    .expect_err("stale release must fail");

    mark_projected_pool(&pool, &reclaimed[0].claim)
        .await
        .expect("project recovered");

    let retry_id = with_transaction(&pool, {
        let retry_topic = retry_topic.clone();
        move |tx| {
            Box::pin(async move {
                enqueue_after(
                    tx,
                    &retry_topic,
                    json!({"job_id": "job-retry-error"}),
                    TimeDuration::ZERO,
                )
                .await
            })
        }
    })
    .await
    .expect("enqueue retry");

    let failed_claim = claim_batch(
        &pool,
        "worker-fail",
        retry_filter,
        1,
        TimeDuration::seconds(30),
    )
    .await
    .expect("claim for release_with_error");
    assert_eq!(failed_claim.len(), 1);
    assert_eq!(failed_claim[0].id, retry_id);

    with_transaction(&pool, |tx| {
        let claim = failed_claim[0].claim.clone();
        Box::pin(async move {
            release_with_error(tx, &claim, "projection failed", TimeDuration::seconds(60)).await
        })
    })
    .await
    .expect("release with error");

    // Ambiguous-commit: same claim token may retry release successfully.
    with_transaction(&pool, |tx| {
        let claim = failed_claim[0].claim.clone();
        Box::pin(async move {
            release_with_error(tx, &claim, "projection failed", TimeDuration::seconds(60)).await
        })
    })
    .await
    .expect("idempotent release_with_error");

    // Missing message id is distinct from claim mismatch.
    let missing = with_transaction(&pool, |tx| {
        Box::pin(async move {
            release_with_error(
                tx,
                &OutboxClaim {
                    message_id: i64::MAX,
                    worker_id: "worker-fail".into(),
                    generation: 1,
                },
                "missing",
                TimeDuration::seconds(1),
            )
            .await
        })
    })
    .await;
    assert!(
        matches!(missing, Err(DurableError::OutboxMissing { .. })),
        "expected OutboxMissing, got {missing:?}"
    );

    let early = claim_batch(
        &pool,
        "worker-early",
        retry_filter,
        1,
        TimeDuration::seconds(30),
    )
    .await
    .expect("early claim");
    assert!(
        early.is_empty(),
        "release_with_error must delay availability on the database clock"
    );

    // Advance availability using the database clock relative update.
    sqlx::query("UPDATE outbox_messages SET available_at = now() WHERE id = $1")
        .bind(retry_id)
        .execute(&pool)
        .await
        .expect("advance available_at for recovery");

    let after_error = claim_batch(
        &pool,
        "worker-retry",
        retry_filter,
        1,
        TimeDuration::seconds(30),
    )
    .await
    .expect("reclaim after release_with_error");
    assert_eq!(after_error.len(), 1);
    assert_eq!(after_error[0].id, retry_id);
    assert_eq!(after_error[0].attempt_count, 2);

    let (last_error,): (Option<String>,) =
        sqlx::query_as("SELECT last_error FROM outbox_messages WHERE id = $1")
            .bind(retry_id)
            .fetch_one(&pool)
            .await
            .expect("read last_error");
    assert_eq!(last_error.as_deref(), Some("projection failed"));

    let concurrent = claim_batch(
        &pool,
        "worker-other",
        retry_filter,
        1,
        TimeDuration::seconds(30),
    )
    .await
    .expect("concurrent claim while leased");
    assert!(concurrent.is_empty());

    mark_projected_pool(&pool, &after_error[0].claim)
        .await
        .expect("project after release recovery");
    assert_eq!(
        unprojected_count(&pool, retry_filter).await.expect("count"),
        0
    );

    let bad_limit = claim_batch(
        &pool,
        "worker-bad-limit",
        batch_filter,
        0,
        TimeDuration::seconds(30),
    )
    .await;
    assert!(
        matches!(bad_limit, Err(DurableError::Config(_))),
        "non-positive claim limit must fail closed: {bad_limit:?}"
    );
}

#[tokio::test]
async fn outbox_skip_locked_excludes_concurrent_open_claim() {
    let Some(pool) = test_pool().await else {
        return;
    };
    const SUITE: &str = "skip";
    clear_outbox_suite(&pool, SUITE).await;
    let topic = outbox_topic(SUITE, "dispatch");
    let filter = OutboxClaimFilter::Topic(&topic);

    with_transaction(&pool, {
        let topic = topic.clone();
        move |tx| {
            Box::pin(async move {
                enqueue(tx, &topic, json!({"job_id": "job-skip-locked"})).await?;
                Ok(())
            })
        }
    })
    .await
    .expect("enqueue");

    // Hold the claim open in an uncommitted transaction so another connection
    // must observe SKIP LOCKED rather than only claimed_until exclusion.
    let mut holder = pool.begin().await.expect("holder begin");
    let held = claim_batch_in_tx(
        &mut holder,
        "worker-hold",
        filter,
        1,
        TimeDuration::seconds(30),
    )
    .await
    .expect("hold claim");
    assert_eq!(held.len(), 1);

    let concurrent = claim_batch(&pool, "worker-other", filter, 1, TimeDuration::seconds(30))
        .await
        .expect("concurrent claim");
    assert!(
        concurrent.is_empty(),
        "SKIP LOCKED must hide a row locked by another transaction"
    );

    mark_projected(&mut holder, &held[0].claim)
        .await
        .expect("project under holder");
    holder.commit().await.expect("holder commit");
    assert_eq!(unprojected_count(&pool, filter).await.expect("count"), 0);
}

#[tokio::test]
async fn stable_keyset_pages_respect_high_water_and_cursor_binding() {
    let Some(pool) = test_pool().await else {
        return;
    };
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS substrate_keyset_jobs (
            id TEXT PRIMARY KEY,
            subject TEXT NOT NULL,
            created_at TIMESTAMPTZ NOT NULL
         )",
    )
    .execute(&pool)
    .await
    .expect("create keyset table");
    sqlx::query("DELETE FROM substrate_keyset_jobs")
        .execute(&pool)
        .await
        .expect("clean keyset table");

    let owner = "subject-keyset-aaaaaaaa";
    let base = OffsetDateTime::now_utc() - TimeDuration::seconds(100);
    for index in 0..10 {
        sqlx::query(
            "INSERT INTO substrate_keyset_jobs (id, subject, created_at)
             VALUES ($1, $2, $3)",
        )
        .bind(format!("job-keyset-{index:04}"))
        .bind(owner)
        .bind(base + TimeDuration::seconds(i64::from(index)))
        .execute(&pool)
        .await
        .expect("insert job");
    }

    let codec = CursorCodec::new(b"integration-cursor-secret").expect("codec");
    let filters = json!({"lifecycle": "visible"});
    let filter_fp = CursorCodec::filter_fingerprint(&filters).expect("filter fp");

    let first_page: Vec<(String, OffsetDateTime)> = sqlx::query_as(
        "SELECT id, created_at
         FROM substrate_keyset_jobs
         WHERE subject = $1
         ORDER BY created_at DESC, id DESC
         LIMIT 3",
    )
    .bind(owner)
    .fetch_all(&pool)
    .await
    .expect("first page");
    assert_eq!(first_page.len(), 3);
    assert_eq!(first_page[0].0, "job-keyset-0009");

    let cursor = first_page_cursor(
        owner,
        filter_fp,
        KeysetPosition {
            created_at: first_page[0].1,
            id: first_page[0].0.clone(),
        },
        KeysetPosition {
            created_at: first_page[2].1,
            id: first_page[2].0.clone(),
        },
    );
    let token = codec.encode(&cursor).expect("encode");

    sqlx::query(
        "INSERT INTO substrate_keyset_jobs (id, subject, created_at)
         VALUES ($1, $2, $3)",
    )
    .bind("job-keyset-9999")
    .bind(owner)
    .bind(OffsetDateTime::now_utc())
    .execute(&pool)
    .await
    .expect("insert newer job");

    let bound = codec.bind(&token, owner, filter_fp).expect("bind cursor");
    let after = bound.after.clone().expect("after position");
    let second_page: Vec<(String, OffsetDateTime)> = sqlx::query_as(
        "SELECT id, created_at
         FROM substrate_keyset_jobs
         WHERE subject = $1
           AND (created_at, id) <= ($2, $3)
           AND (created_at, id) < ($4, $5)
         ORDER BY created_at DESC, id DESC
         LIMIT 3",
    )
    .bind(owner)
    .bind(bound.high_water.created_at)
    .bind(&bound.high_water.id)
    .bind(after.created_at)
    .bind(&after.id)
    .fetch_all(&pool)
    .await
    .expect("second page");

    assert_eq!(
        second_page
            .iter()
            .map(|row| row.0.as_str())
            .collect::<Vec<_>>(),
        vec!["job-keyset-0006", "job-keyset-0005", "job-keyset-0004"]
    );
    assert!(
        second_page.iter().all(|row| row.0 != "job-keyset-9999"),
        "jobs newer than the high-water mark must not enter a stable traversal"
    );

    let advanced = next_page_cursor(
        &bound,
        KeysetPosition {
            created_at: second_page[2].1,
            id: second_page[2].0.clone(),
        },
    );
    let advanced_token = codec.encode(&advanced).expect("encode advanced");
    assert!(matches!(
        codec.bind(&advanced_token, "other-subject-bbbbbbbb", filter_fp),
        Err(DurableError::CursorBindingMismatch)
    ));
}

#[tokio::test]
async fn advisory_lock_serializes_critical_section() {
    let Some(pool) = test_pool().await else {
        return;
    };
    sqlx::query(
        "CREATE TABLE IF NOT EXISTS substrate_lock_probe (
            id TEXT PRIMARY KEY,
            value BIGINT NOT NULL
         )",
    )
    .execute(&pool)
    .await
    .expect("create lock probe");
    sqlx::query(
        "INSERT INTO substrate_lock_probe (id, value) VALUES ('lock-row', 0)
         ON CONFLICT (id) DO UPDATE SET value = 0",
    )
    .execute(&pool)
    .await
    .expect("seed lock probe");

    let barrier = Arc::new(Barrier::new(4));
    let mut handles = Vec::new();
    for _ in 0..4 {
        let pool = pool.clone();
        let barrier = Arc::clone(&barrier);
        handles.push(tokio::spawn(async move {
            barrier.wait().await;
            with_transaction(&pool, |tx| {
                Box::pin(async move {
                    advisory_xact_lock(tx, advisory_key(ADVISORY_CLASS_APP, 42)).await?;
                    let (value,): (i64,) = sqlx::query_as(
                        "SELECT value FROM substrate_lock_probe WHERE id = 'lock-row'",
                    )
                    .fetch_one(tx.as_mut())
                    .await?;
                    tokio::time::sleep(Duration::from_millis(20)).await;
                    sqlx::query("UPDATE substrate_lock_probe SET value = $1 WHERE id = 'lock-row'")
                        .bind(value + 1)
                        .execute(tx.as_mut())
                        .await?;
                    Ok::<_, DurableError>(())
                })
            })
            .await
        }));
    }
    for handle in handles {
        handle.await.expect("join").expect("lock section");
    }
    let (value,): (i64,) =
        sqlx::query_as("SELECT value FROM substrate_lock_probe WHERE id = 'lock-row'")
            .fetch_one(&pool)
            .await
            .expect("read lock probe");
    assert_eq!(value, 4);
}