import sqlite3
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier

from src.idempotency import IdempotencyLedger


def test_completed_result_replays_after_ledger_reconstruction(tmp_path):
    database = tmp_path / "idempotency.sqlite3"
    first = IdempotencyLedger(database, ttl_seconds=600, max_entries=256)

    reservation = first.reserve("create-201", ("issue-create", "stackchain/api", "Ship it"))
    assert reservation.state == "reserved"
    first.complete("create-201", {"number": 42, "title": "Ship it"})

    reconstructed = IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
    replay = reconstructed.reserve(
        "create-201", ("issue-create", "stackchain/api", "Ship it")
    )

    assert replay.state == "completed"
    assert replay.response == {"number": 42, "title": "Ship it"}
    assert sqlite3.connect(database).execute(
        "SELECT status FROM idempotency_operations WHERE key = ?", ("create-201",)
    ).fetchone() == ("completed",)


def test_reservation_expires_completed_rows_and_enforces_capacity(tmp_path):
    now = [1_000.0]
    ledger = IdempotencyLedger(
        tmp_path / "bounded.sqlite3",
        ttl_seconds=10,
        max_entries=2,
        clock=lambda: now[0],
    )
    assert ledger.reserve("old", ("comment", "old")).state == "reserved"
    ledger.complete("old", {"id": 1})
    now[0] += 11

    assert ledger.reserve("new-1", ("comment", "new-1")).state == "reserved"
    ledger.complete("new-1", {"id": 2})
    assert ledger.reserve("new-2", ("comment", "new-2")).state == "reserved"
    ledger.complete("new-2", {"id": 3})
    replacement = ledger.reserve("new-3", ("comment", "new-3"))

    assert replacement.state == "reserved"
    with sqlite3.connect(ledger.path) as connection:
        keys = connection.execute(
            "SELECT key FROM idempotency_operations ORDER BY key"
        ).fetchall()
        indexes = connection.execute(
            "PRAGMA index_list(idempotency_operations)"
        ).fetchall()
    assert keys == [("new-2",), ("new-3",)]
    assert any(index[1] == "idempotency_completed_at_idx" for index in indexes)


def test_independent_connections_atomically_reserve_one_operation(tmp_path):
    database = tmp_path / "workers.sqlite3"
    ledgers = [
        IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
        for _ in range(2)
    ]
    barrier = Barrier(2)

    def reserve(ledger):
        barrier.wait()
        return ledger.reserve("worker-key-201", ("review", "stackchain/api", 7)).state

    with ThreadPoolExecutor(max_workers=2) as executor:
        states = list(executor.map(reserve, ledgers))

    assert sorted(states) == ["pending", "reserved"]


def test_stale_pending_reservation_becomes_uncertain_without_consuming_capacity(tmp_path):
    now = [1_000.0]
    ledger = IdempotencyLedger(
        tmp_path / "orphaned.sqlite3",
        ttl_seconds=10,
        max_entries=1,
        clock=lambda: now[0],
    )
    fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first")

    assert ledger.reserve("orphan", fingerprint).state == "reserved"
    now[0] += 11

    assert ledger.reserve("orphan", fingerprint).state == "uncertain"
    assert ledger.reserve("unrelated", ("issue-comment", "stackchain/api", 8, "Continue")).state == "reserved"
    assert ledger.reserve("orphan", ("issue-comment", "stackchain/api", 7, "Changed")).state == "conflict"