import asyncio
import time

import httpx
import pytest

from src import gitea_proxy, main
from src.live_snapshot_store import LiveSnapshotStore


@pytest.fixture(autouse=True)
def isolated_live_snapshot_store(tmp_path, monkeypatch):
    monkeypatch.setattr(
        main,
        "_live_snapshot_store",
        LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: main._live_snapshot_clock()),
    )
    monkeypatch.setattr(main, "_read_notification_ids", set())


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

    class Response:
        def raise_for_status(self):
            return None

    class Client:
        async def patch(self, path, headers):
            calls.append((path, headers))
            return Response()

    monkeypatch.setattr(gitea_proxy, "_get_client", lambda: Client())
    monkeypatch.setattr(gitea_proxy, "_auth", lambda: {"Authorization": "token test"})

    await gitea_proxy.mark_notification_read(42)

    assert calls == [
        ("/api/v1/notifications/threads/42?to-status=read", {"Authorization": "token test"})
    ]


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

    async def fake_fetch(path):
        calls.append(("fetch", path))
        if path == "notifications/threads/42":
            return {
                "repository": {"full_name": "stackchain/api"},
                "subject": {
                    "type": "Issue",
                    "url": "https://forge.example/api/v1/repos/stackchain/api/issues/7",
                    "latest_comment_url": (
                        "https://forge.example/api/v1/repos/stackchain/api/issues/comments/91"
                    ),
                },
            }
        if path == "user":
            return {"login": "timmy"}
        raise AssertionError(path)

    class Response:
        def __init__(self, payload=None):
            self._payload = payload

        def raise_for_status(self):
            return None

        def json(self):
            return self._payload

    class Client:
        async def get(self, path, headers):
            calls.append(("get", path))
            return Response([])

        async def post(self, path, headers, json):
            calls.append(("post", path, json))
            return Response({"content": "+1"})

        async def patch(self, path, headers):
            calls.append(("patch", path))
            return Response()

    monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example")
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
    monkeypatch.setattr(gitea_proxy, "_get_client", lambda: Client())
    monkeypatch.setattr(gitea_proxy, "_auth", lambda: {"Authorization": "token test"})

    result = await gitea_proxy.acknowledge_notification(42)

    reaction_path = "/api/v1/repos/stackchain/api/issues/comments/91/reactions"
    assert result == {"id": 42, "reaction": "created", "status": "read"}
    assert calls == [
        ("fetch", "notifications/threads/42"),
        ("fetch", "user"),
        ("get", reaction_path),
        ("post", reaction_path, {"content": "+1"}),
        ("patch", "/api/v1/notifications/threads/42?to-status=read"),
    ]


@pytest.mark.anyio
async def test_acknowledge_notification_reuses_existing_operator_reaction_on_retry(monkeypatch):
    posts = []

    async def fake_fetch(path):
        if path == "user":
            return {"login": "timmy"}
        return {
            "repository": {"full_name": "stackchain/api"},
            "subject": {
                "type": "Pull",
                "url": "https://forge.example/api/v1/repos/stackchain/api/pulls/7",
                "latest_comment_url": (
                    "https://forge.example/api/v1/repos/stackchain/api/issues/comments/91"
                ),
            },
        }

    class Response:
        def __init__(self, payload=None):
            self.payload = payload

        def raise_for_status(self):
            return None

        def json(self):
            return self.payload

    class Client:
        async def get(self, path, headers):
            return Response([{"content": "+1", "user": {"login": "timmy"}}])

        async def post(self, path, headers, json):
            posts.append((path, json))
            return Response()

        async def patch(self, path, headers):
            return Response()

    monkeypatch.setattr(gitea_proxy, "GITEA_URL", "https://forge.example")
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
    monkeypatch.setattr(gitea_proxy, "_get_client", lambda: Client())

    result = await gitea_proxy.acknowledge_notification(42)

    assert result == {"id": 42, "reaction": "existing", "status": "read"}
    assert posts == []


@pytest.mark.anyio
async def test_mark_notification_read_api_is_bounded_and_never_cacheable(monkeypatch):
    marked = []

    async def mark(thread_id):
        marked.append(thread_id)

    monkeypatch.setattr(main, "mark_notification_read", mark)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.patch("/api/v1/notifications/42/read")
        invalid = await client.patch("/api/v1/notifications/0/read")

    assert response.status_code == 200
    assert response.json() == {"id": 42, "status": "read"}
    assert response.headers["cache-control"] == "no-store"
    assert invalid.status_code == 422
    assert marked == [42]


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

    async def acknowledge(thread_id):
        calls.append(thread_id)
        return {"id": thread_id, "reaction": "created", "status": "read"}

    monkeypatch.setattr(main.gitea_proxy, "acknowledge_notification", acknowledge, raising=False)
    monkeypatch.setattr(
        main,
        "_live_snapshot_value",
        {"notifications": [{"id": 42}, {"id": 43}]},
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/api/v1/notifications/42/acknowledge")
        invalid = await client.post("/api/v1/notifications/0/acknowledge")

    assert response.status_code == 200
    assert response.json() == {"id": 42, "reaction": "created", "status": "read"}
    assert response.headers["cache-control"] == "no-store"
    assert invalid.status_code == 422
    assert calls == [42]
    assert main._live_snapshot_value == {"notifications": [{"id": 43}]}


@pytest.mark.anyio
async def test_snapshot_maintenance_does_not_block_the_event_loop(monkeypatch):
    async def mark(_thread_id):
        return None

    calls = []

    class BlockingStore:
        def remove_notification(self, thread_id):
            calls.append([thread_id])
            time.sleep(0.08)

        def remove_notifications(self, thread_ids):
            calls.append(list(thread_ids))
            time.sleep(0.08)

    monkeypatch.setattr(main, "mark_notification_read", mark)
    monkeypatch.setattr(main, "_live_snapshot_store", BlockingStore())
    monkeypatch.setattr(main, "_live_snapshot_value", None)
    started_at = asyncio.get_running_loop().time()

    request = asyncio.create_task(main.read_notification(42))
    await asyncio.sleep(0.01)
    heartbeat_delay = asyncio.get_running_loop().time() - started_at
    response = await request

    assert heartbeat_delay < 0.05
    assert response.status_code == 200
    assert calls == [[42]]


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

    async def mark(thread_id):
        calls.append(thread_id)
        if thread_id == 43:
            raise httpx.HTTPError("upstream unavailable")

    monkeypatch.setattr(main, "mark_notification_read", mark)
    monkeypatch.setattr(
        main,
        "_live_snapshot_value",
        {"notifications": [{"id": 42}, {"id": 43}, {"id": 44}]},
    )
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.patch(
            "/api/v1/notifications/read", json={"ids": [42, 43, 42, 44]}
        )

    assert response.status_code == 200
    assert response.json() == {"marked": [42, 44], "failed": [43]}
    assert response.headers["cache-control"] == "no-store"
    assert sorted(calls) == [42, 43, 44]
    assert main._live_snapshot_value == {"notifications": [{"id": 43}]}


@pytest.mark.anyio
async def test_bulk_mark_read_updates_shared_snapshot_once_for_successes(monkeypatch):
    async def mark(thread_id):
        if thread_id == 43:
            raise httpx.HTTPError("upstream unavailable")

    calls = []

    class RecordingStore:
        def remove_notifications(self, thread_ids):
            calls.append(tuple(sorted(thread_ids)))

    monkeypatch.setattr(main, "mark_notification_read", mark)
    monkeypatch.setattr(main, "_live_snapshot_store", RecordingStore())
    monkeypatch.setattr(
        main,
        "_live_snapshot_value",
        {"notifications": [{"id": 42}, {"id": 43}, {"id": 44}]},
    )

    response = await main.read_notifications(main.NotificationReadBatch(ids=[42, 43, 44]))

    assert response.status_code == 200
    assert calls == [(42, 44)]
    assert main._live_snapshot_value == {"notifications": [{"id": 43}]}


@pytest.mark.anyio
async def test_bulk_mark_read_limits_upstream_concurrency(monkeypatch):
    active = 0
    peak_active = 0
    started = asyncio.Event()
    release = asyncio.Event()
    calls = []

    monkeypatch.setattr(main, "_read_notification_ids", set())
    monkeypatch.setattr(main, "_live_snapshot_value", None)

    async def mark(thread_id):
        nonlocal active, peak_active
        calls.append(thread_id)
        active += 1
        peak_active = max(peak_active, active)
        if active == 5:
            started.set()
        try:
            await release.wait()
        finally:
            active -= 1

    monkeypatch.setattr(main, "mark_notification_read", mark)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        request = asyncio.create_task(
            client.patch("/api/v1/notifications/read", json={"ids": list(range(1, 51))})
        )
        await asyncio.wait_for(started.wait(), timeout=1)
        await asyncio.sleep(0)
        assert peak_active == 5
        assert len(calls) == 5
        release.set()
        response = await request

    assert response.status_code == 200
    assert response.json() == {"marked": list(range(1, 51)), "failed": []}
    assert sorted(calls) == list(range(1, 51))


@pytest.mark.anyio
async def test_bulk_mark_read_cancels_unfinished_work_at_batch_deadline(monkeypatch):
    active = 0
    cancelled = []

    async def mark(thread_id):
        nonlocal active
        active += 1
        try:
            await asyncio.sleep(0.08)
        except asyncio.CancelledError:
            cancelled.append(thread_id)
            raise
        finally:
            active -= 1

    monkeypatch.setattr(main, "mark_notification_read", mark)
    monkeypatch.setattr(main, "BULK_NOTIFICATION_DEADLINE_SECONDS", 0.01, raising=False)
    monkeypatch.setattr(main, "_read_notification_ids", set())
    monkeypatch.setattr(main, "_live_snapshot_value", None)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        response = await asyncio.wait_for(
            client.patch(
                "/api/v1/notifications/read", json={"ids": list(range(1, 11))}
            ),
            timeout=0.5,
        )

    assert response.status_code == 200
    assert response.json() == {"marked": [], "failed": list(range(1, 11))}
    assert response.headers["retry-after"] == "1"
    assert active == 0
    assert sorted(cancelled) == list(range(1, 6))


@pytest.mark.anyio
async def test_bulk_mark_read_rejects_empty_invalid_and_oversized_batches(monkeypatch):
    marked = []

    async def mark(thread_id):
        marked.append(thread_id)

    monkeypatch.setattr(main, "mark_notification_read", mark)
    transport = httpx.ASGITransport(app=main.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        empty = await client.patch("/api/v1/notifications/read", json={"ids": []})
        invalid = await client.patch(
            "/api/v1/notifications/read", json={"ids": [42, 0]}
        )
        oversized = await client.patch(
            "/api/v1/notifications/read", json={"ids": list(range(1, 52))}
        )

    assert [empty.status_code, invalid.status_code, oversized.status_code] == [
        422,
        422,
        422,
    ]
    assert all(response.headers["cache-control"] == "no-store" for response in [empty, invalid, oversized])
    assert marked == []


@pytest.mark.anyio
async def test_mark_notification_read_removes_thread_from_retained_live_snapshot(monkeypatch):
    async def mark(_thread_id):
        return None

    monkeypatch.setattr(main, "mark_notification_read", mark)
    monkeypatch.setattr(
        main,
        "_live_snapshot_value",
        {
            "context": {},
            "events": [],
            "notifications": [{"id": 42}, {"id": 43}],
            "sections": {"notifications": "fresh"},
        },
    )

    response = await main.read_notification(42)

    assert response.status_code == 200
    assert main._live_snapshot_value is not None
    assert main._live_snapshot_value["notifications"] == [{"id": 43}]


@pytest.mark.anyio
async def test_inflight_refresh_cannot_restore_a_notification_marked_read(monkeypatch):
    refresh_started = asyncio.Event()
    release_refresh = asyncio.Event()

    async def mark(_thread_id):
        return None

    async def stale_upstream_snapshot(_sections):
        refresh_started.set()
        await release_refresh.wait()
        return {
            "context": {},
            "events": [],
            "notifications": [{"id": 42}, {"id": 43}],
            "sections": {"notifications": "fresh"},
        }

    monkeypatch.setattr(main, "mark_notification_read", mark)
    monkeypatch.setattr(
        main, "_build_live_snapshot_before_deadline", stale_upstream_snapshot
    )
    monkeypatch.setattr(main, "_live_snapshot_task", None)
    monkeypatch.setattr(main, "_live_snapshot_value", {"notifications": [{"id": 42}]})

    refresh = main._start_live_snapshot_refresh(set(main.LIVE_SNAPSHOT_SECTIONS))
    await refresh_started.wait()
    await main.read_notification(42)
    release_refresh.set()
    await refresh

    assert main._live_snapshot_value is not None
    assert main._live_snapshot_value["notifications"] == [{"id": 43}]