import asyncio
import sqlite3
import threading
import time

import httpx
import pytest

from src import main
from src.idempotency import IdempotencyLedger


@pytest.fixture(autouse=True)
def clear_authored_operations():
    main._authored_action_operations.clear()
    main._idempotency_ledger.clear()
    yield
    main._authored_action_operations.clear()
    main._idempotency_ledger.clear()


@pytest.mark.anyio
async def test_issue_comment_replays_one_upstream_result_for_concurrent_requests(monkeypatch):
    calls = 0
    started = asyncio.Event()
    release = asyncio.Event()

    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        nonlocal calls
        calls += 1
        started.set()
        await release.wait()
        return {"id": 82, "body": body}

    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    transport = httpx.ASGITransport(app=main.app)
    headers = {"Idempotency-Key": "issue-comment-185"}
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        first = asyncio.create_task(client.post(
            "/api/v1/repos/stackchain/api/issues/7/comments",
            json={"body": "Ship it"}, headers=headers,
        ))
        await started.wait()
        second = asyncio.create_task(client.post(
            "/api/v1/repos/stackchain/api/issues/7/comments",
            json={"body": "Ship it"}, headers=headers,
        ))
        await asyncio.sleep(0)
        release.set()
        responses = await asyncio.gather(first, second)

    assert [response.status_code for response in responses] == [201, 201]
    assert calls == 1


@pytest.mark.anyio
async def test_pull_comment_rejects_changed_payload_for_same_key(monkeypatch):
    calls = []

    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        calls.append(body)
        return {"id": len(calls), "body": body}

    monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    transport = httpx.ASGITransport(app=main.app)
    headers = {"Idempotency-Key": "pull-comment-185"}
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        created = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/comments",
            json={"body": "Ship it"}, headers=headers,
        )
        conflict = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/comments",
            json={"body": "Changed"}, headers=headers,
        )

    assert created.status_code == 201
    assert conflict.status_code == 409
    assert calls == ["Ship it"]


@pytest.mark.anyio
async def test_notification_reply_retry_recovers_after_caller_timeout(monkeypatch):
    calls = 0

    async def reply(_thread_id, body):
        nonlocal calls
        calls += 1
        await asyncio.sleep(0.03)
        return {"id": 91, "body": body}

    monkeypatch.setattr(main, "NOTIFICATION_MUTATION_TIMEOUT_SECONDS", 0.01)
    monkeypatch.setattr(main.gitea_proxy, "reply_to_notification", reply)
    transport = httpx.ASGITransport(app=main.app)
    headers = {"Idempotency-Key": "notification-reply-185"}
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        timed_out = await client.post(
            "/api/v1/notifications/42/reply", json={"body": "Retry"}, headers=headers,
        )
        await asyncio.sleep(0.03)
        recovered = await client.post(
            "/api/v1/notifications/42/reply", json={"body": "Retry"}, headers=headers,
        )

    assert timed_out.status_code == 503
    assert recovered.status_code == 201
    assert recovered.json()["id"] == 91
    assert calls == 1


@pytest.mark.anyio
async def test_review_replays_same_key_and_rejects_changed_head(monkeypatch):
    calls = []

    async def requested(_repository, _number):
        return True

    async def submit(_repository, _number, head, decision, body):
        calls.append((head, decision, body))
        return {"id": 93, "state": "APPROVED"}

    monkeypatch.setattr(main, "is_requested_review", requested)
    monkeypatch.setattr(main.gitea_proxy, "submit_pull_review", submit)
    transport = httpx.ASGITransport(app=main.app)
    headers = {"Idempotency-Key": "review-185"}
    payload = {"expected_head_sha": "abc", "decision": "approve", "body": "Good"}
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        created = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review", json=payload, headers=headers,
        )
        replayed = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review", json=payload, headers=headers,
        )
        conflict = await client.post(
            "/api/v1/repos/stackchain/api/pulls/7/review",
            json={**payload, "expected_head_sha": "def"}, headers=headers,
        )

    assert [created.status_code, replayed.status_code, conflict.status_code] == [201, 201, 409]
    assert calls == [("abc", "approve", "Good")]


@pytest.mark.anyio
async def test_completed_comment_replays_after_ledger_reconstruction(monkeypatch, tmp_path):
    calls = 0

    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        nonlocal calls
        calls += 1
        return {"id": 201, "body": body}

    database = tmp_path / "actions.sqlite3"
    monkeypatch.setattr(
        main,
        "_idempotency_ledger",
        IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
        raising=False,
    )
    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    transport = httpx.ASGITransport(app=main.app)
    headers = {"Idempotency-Key": "restart-comment-201"}

    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        created = await client.post(
            "/api/v1/repos/stackchain/api/issues/7/comments",
            json={"body": "Persist me"},
            headers=headers,
        )
        main._authored_action_operations.clear()
        monkeypatch.setattr(
            main,
            "_idempotency_ledger",
            IdempotencyLedger(database, ttl_seconds=600, max_entries=256),
        )
        replayed = await client.post(
            "/api/v1/repos/stackchain/api/issues/7/comments",
            json={"body": "Persist me"},
            headers=headers,
        )

    assert [created.status_code, replayed.status_code] == [201, 201]
    assert replayed.json() == {"id": 201, "body": "Persist me"}
    assert calls == 1


@pytest.mark.anyio
async def test_orphaned_pending_comment_fails_closed_without_upstream_retry(
    monkeypatch, tmp_path
):
    calls = 0

    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        nonlocal calls
        calls += 1
        return {"id": 202, "body": body}

    ledger = IdempotencyLedger(
        tmp_path / "orphan.sqlite3", ttl_seconds=600, max_entries=256
    )
    fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first")
    assert ledger.reserve("orphan-comment-201", fingerprint).state == "reserved"
    monkeypatch.setattr(main, "_idempotency_ledger", ledger)
    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    transport = httpx.ASGITransport(app=main.app)

    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/repos/stackchain/api/issues/7/comments",
            json={"body": "Verify first"},
            headers={"Idempotency-Key": "orphan-comment-201"},
        )

    assert response.status_code == 503
    assert response.json()["detail"] == (
        "This action may still be completing; verify its result before retrying"
    )
    assert calls == 0


@pytest.mark.anyio
async def test_slow_ledger_reservation_does_not_block_health_requests(monkeypatch):
    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        return {"id": 248, "body": body}

    started = threading.Event()
    original_reserve = main._idempotency_ledger.reserve

    def slow_reserve(key, fingerprint):
        started.set()
        time.sleep(0.2)
        return original_reserve(key, fingerprint)

    monkeypatch.setattr(main._idempotency_ledger, "reserve", slow_reserve)
    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    transport = httpx.ASGITransport(app=main.app)

    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        started_at = asyncio.get_running_loop().time()
        authored = asyncio.create_task(client.post(
            "/api/v1/repos/stackchain/api/issues/248/comments",
            json={"body": "Keep the dashboard responsive"},
            headers={"Idempotency-Key": "contention-248"},
        ))
        assert await asyncio.to_thread(started.wait, 1)
        health = await client.get("/healthz")
        health_elapsed = asyncio.get_running_loop().time() - started_at
        authored_response = await authored

    assert health.status_code == 200
    assert health_elapsed < 0.15
    assert authored_response.status_code == 201


@pytest.mark.anyio
async def test_locked_ledger_returns_bounded_retryable_response(monkeypatch, tmp_path):
    upstream_calls = 0

    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        nonlocal upstream_calls
        upstream_calls += 1
        return {"id": 248, "body": body}

    database = tmp_path / "contended.sqlite3"
    ledger = IdempotencyLedger(
        database,
        ttl_seconds=600,
        max_entries=256,
        lock_timeout_seconds=0.05,
    )
    monkeypatch.setattr(main, "_idempotency_ledger", ledger)
    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    blocker = sqlite3.connect(database)
    blocker.execute("BEGIN IMMEDIATE")
    transport = httpx.ASGITransport(app=main.app)

    try:
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            started_at = asyncio.get_running_loop().time()
            response = await client.post(
                "/api/v1/repos/stackchain/api/issues/248/comments",
                json={"body": "Retry without freezing"},
                headers={"Idempotency-Key": "locked-ledger-248"},
            )
            elapsed = asyncio.get_running_loop().time() - started_at
    finally:
        blocker.rollback()
        blocker.close()

    assert response.status_code == 503
    assert response.headers["Retry-After"] == "1"
    assert response.json()["detail"] == "Authored action queue is busy; please retry"
    assert elapsed < 0.5
    assert upstream_calls == 0


@pytest.mark.anyio
async def test_slow_ledger_completion_does_not_block_health_requests(monkeypatch):
    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        return {"id": 249, "body": body}

    started = threading.Event()
    original_complete = main._idempotency_ledger.complete

    def slow_complete(key, response):
        started.set()
        time.sleep(0.2)
        return original_complete(key, response)

    monkeypatch.setattr(main._idempotency_ledger, "complete", slow_complete)
    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    transport = httpx.ASGITransport(app=main.app)

    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        started_at = asyncio.get_running_loop().time()
        authored = asyncio.create_task(client.post(
            "/api/v1/repos/stackchain/api/issues/248/comments",
            json={"body": "Persist without freezing"},
            headers={"Idempotency-Key": "slow-completion-248"},
        ))
        assert await asyncio.to_thread(started.wait, 1)
        health = await client.get("/healthz")
        health_elapsed = asyncio.get_running_loop().time() - started_at
        authored_response = await authored

    assert health.status_code == 200
    assert health_elapsed < 0.15
    assert authored_response.status_code == 201


@pytest.mark.anyio
async def test_locked_completion_fails_closed_with_retry_guidance(monkeypatch, tmp_path):
    database = tmp_path / "completion-contention.sqlite3"
    ledger = IdempotencyLedger(
        database, ttl_seconds=600, max_entries=256, lock_timeout_seconds=0.05
    )
    blocker = None

    async def assigned(_repository, _number):
        return True

    async def comment(_repository, _number, body):
        nonlocal blocker
        blocker = sqlite3.connect(database)
        blocker.execute("BEGIN IMMEDIATE")
        return {"id": 250, "body": body}

    monkeypatch.setattr(main, "_idempotency_ledger", ledger)
    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", assigned)
    monkeypatch.setattr(main.gitea_proxy, "comment_on_issue", comment)
    transport = httpx.ASGITransport(app=main.app)

    try:
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.post(
                "/api/v1/repos/stackchain/api/issues/248/comments",
                json={"body": "The upstream mutation may already exist"},
                headers={"Idempotency-Key": "locked-completion-248"},
            )
    finally:
        if blocker is not None:
            blocker.rollback()
            blocker.close()

    assert response.status_code == 503
    assert response.headers["Retry-After"] == "5"
    assert response.json()["detail"] == (
        "This action may have completed; verify its result before retrying"
    )
    assert ledger.reserve(
        "locked-completion-248",
        ("issue-comment", "stackchain/api", 248, "The upstream mutation may already exist"),
    ).state == "pending"