import asyncio

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