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

import pytest

from src.idempotency import IdempotencyLedger
from src.state_encryption import PrivateStateEncryptionError


PRIVATE_KEY = b"i" * 32


def test_authored_payloads_are_encrypted_at_rest_and_replay_after_restart(tmp_path):
    database = tmp_path / "idempotency.sqlite3"
    fingerprint_canary = "private-issue-body-canary"
    response_canary = "private-gitea-response-canary"
    ledger = IdempotencyLedger(
        database,
        ttl_seconds=600,
        max_entries=256,
        encryption_key=PRIVATE_KEY,
    )

    assert ledger.reserve("create-201", ("issue-create", fingerprint_canary)).state == "reserved"
    ledger.complete("create-201", {"title": response_canary})

    persisted = database.read_bytes()
    assert fingerprint_canary.encode() not in persisted
    assert response_canary.encode() not in persisted
    with sqlite3.connect(database) as connection:
        fingerprint, response = connection.execute(
            "SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
            ("create-201",),
        ).fetchone()
    assert fingerprint.startswith("v1:")
    assert response.startswith("v1:")

    reopened = IdempotencyLedger(
        database,
        ttl_seconds=600,
        max_entries=256,
        encryption_key=PRIVATE_KEY,
    )
    replay = reopened.reserve("create-201", ("issue-create", fingerprint_canary))
    assert replay.state == "completed"
    assert replay.response == {"title": response_canary}


def test_legacy_plaintext_completed_row_is_migrated_without_losing_replay(tmp_path):
    database = tmp_path / "legacy.sqlite3"
    ledger = IdempotencyLedger(
        database,
        ttl_seconds=600,
        max_entries=256,
        encryption_key=PRIVATE_KEY,
        clock=lambda: 100.0,
    )
    with sqlite3.connect(database) as connection:
        connection.execute(
            "INSERT INTO idempotency_operations "
            "(key, fingerprint, status, response_json, created_at, completed_at) "
            "VALUES (?, ?, 'completed', ?, ?, ?)",
            (
                "legacy-comment",
                '["issue-comment","stackchain/api",7,"legacy secret"]',
                '{"id":42,"body":"legacy response"}',
                90.0,
                95.0,
            ),
        )

    replay = ledger.reserve(
        "legacy-comment",
        ("issue-comment", "stackchain/api", 7, "legacy secret"),
    )

    assert replay.state == "completed"
    assert replay.response == {"id": 42, "body": "legacy response"}
    with sqlite3.connect(database) as connection:
        fingerprint, response = connection.execute(
            "SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
            ("legacy-comment",),
        ).fetchone()
    assert fingerprint.startswith("v1:")
    assert response.startswith("v1:")


def test_wrong_key_fails_closed_before_a_completed_result_can_replay(tmp_path):
    database = tmp_path / "wrong-key.sqlite3"
    ledger = IdempotencyLedger(
        database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
    )
    fingerprint = ("issue-comment", "stackchain/api", 7, "secret")
    assert ledger.reserve("comment-7", fingerprint).state == "reserved"
    ledger.complete("comment-7", {"id": 42})

    wrong_key = IdempotencyLedger(
        database, ttl_seconds=600, max_entries=256, encryption_key=b"x" * 32
    )
    with pytest.raises(
        PrivateStateEncryptionError, match="private state could not be decrypted"
    ):
        wrong_key.reserve("comment-7", fingerprint)


def test_completed_ciphertext_cannot_be_substituted_between_operation_keys(tmp_path):
    database = tmp_path / "substitution.sqlite3"
    ledger = IdempotencyLedger(
        database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
    )
    fingerprint = ("issue-create", "stackchain/api", "same authored request")
    for key, number in (("create-a", 41), ("create-b", 42)):
        assert ledger.reserve(key, fingerprint).state == "reserved"
        ledger.complete(key, {"number": number})
    with sqlite3.connect(database) as connection:
        responses = dict(
            connection.execute(
                "SELECT key, response_json FROM idempotency_operations"
            ).fetchall()
        )
        connection.execute(
            "UPDATE idempotency_operations SET response_json = ? WHERE key = ?",
            (responses["create-b"], "create-a"),
        )

    with pytest.raises(
        PrivateStateEncryptionError, match="private state could not be decrypted"
    ):
        ledger.reserve("create-a", fingerprint)


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"