"""Серверные события (ТЗ 3.14): шина публикации и SSE-генератор.

TestClient буферизует весь ответ (portal.call ждёт завершения приложения),
поэтому бесконечный SSE-стрим через client.stream нечитаем в принципе —
генератор `_event_source` тестируем напрямую, HTTP-публикации — перехватом
publish.
"""

import asyncio
import threading
from collections.abc import AsyncIterator, Iterator
from typing import Any

import pytest

from app.api import events as events_api
from app.realtime import Event, bus, publish


@pytest.fixture(autouse=True)
def _fresh_bus() -> Iterator[None]:
    """Шина между тестами пуста: asyncio.run юнит-тестов оставляет мёртвый loop."""
    bus._subscribers.clear()
    bus._loop = None
    yield
    bus._subscribers.clear()
    bus._loop = None


class _FakeRequest:
    """Минимальный двойник Request для _event_source (is_disconnected → False)."""

    async def is_disconnected(self) -> bool:
        return False


async def _anext_timeout(gen: AsyncIterator[str], timeout: float = 2.0) -> str:
    return await asyncio.wait_for(gen.__anext__(), timeout=timeout)


# --- юниты шины ---


def test_dispatch_same_loop() -> None:
    """Подписка и публикация в текущем loop: событие попадает в очередь."""

    async def scenario() -> Any:
        queue = bus.subscribe("u1")
        try:
            publish("u1", "task.changed", {"id": 1})
            return await asyncio.wait_for(queue.get(), timeout=2)
        finally:
            bus.unsubscribe("u1", queue)

    event = asyncio.run(scenario())
    assert event.kind == "task.changed"
    assert event.data == {"id": 1}


def test_publish_from_thread() -> None:
    """publish из другого потока (как MCP/detail_task) доезжает через call_soon_threadsafe."""

    async def scenario() -> Any:
        queue = bus.subscribe("u1")
        try:
            thread = threading.Thread(
                target=publish, args=(None, "garden.changed"), kwargs={"data": {"reason": "shop"}}
            )
            thread.start()
            return await asyncio.wait_for(queue.get(), timeout=2)
        finally:
            bus.unsubscribe("u1", queue)

    event = asyncio.run(scenario())
    assert event.kind == "garden.changed"


def test_queue_overflow_skips() -> None:
    """Переполнение очереди не бросает: событие пропускается (coarse-модель)."""

    async def scenario() -> None:
        queue = bus.subscribe("u1")
        try:
            for i in range(100):
                publish("u1", "task.changed", {"i": i})
            # очередь забита до maxsize — следующий publish не бросает
            publish("u1", "task.changed", {"overflow": True})
        finally:
            bus.unsubscribe("u1", queue)

    asyncio.run(scenario())


def test_no_subscribers_is_noop() -> None:
    """Publish без подписчиков — тихий no-op (клиент догонит refetch'ем)."""
    publish(None, "task.changed", {"id": 1})


# --- SSE-генератор ---


def test_event_source_ready_then_publish() -> None:
    """Первый кадр ready; событие шины доезжает в стрим; отсоединение завершает."""

    async def scenario() -> list[str]:
        gen = events_api._event_source(_FakeRequest(), "u1")  # type: ignore[arg-type]
        frames: list[str] = []
        frames.append(await _anext_timeout(gen))  # subscribe внутри → loop забинден
        publish("u1", "project.changed", {"id": 5})
        frames.append(await _anext_timeout(gen))
        await gen.aclose()
        return frames

    frames = asyncio.run(scenario())
    assert frames[0] == "data: " + events_api._sse_json("ready") + "\n\n"
    assert '"project.changed"' in frames[1]
    assert '"id": 5' in frames[1]
    # после aclose подписчик отписан
    assert bus._subscribers == {}


def test_event_source_unsubscribe_on_disconnect() -> None:
    """is_disconnected → генератор завершается, подписка снята (finally)."""

    class _Disconnecting:
        def __init__(self) -> None:
            self.calls = 0

        async def is_disconnected(self) -> bool:
            self.calls += 1
            return True  # ready уходит без проверки, затем сразу disconnect

    async def scenario() -> None:
        gen = events_api._event_source(_Disconnecting(), "u1")  # type: ignore[arg-type]
        await _anext_timeout(gen)
        with pytest.raises(StopAsyncIteration):
            await _anext_timeout(gen)

    asyncio.run(scenario())
    assert bus._subscribers == {}


def test_event_source_ping_on_idle() -> None:
    """Тишина в шине дольше PING_SECONDS → кадр ping (keep-alive для прокси)."""

    async def scenario() -> str:
        gen = events_api._event_source(_FakeRequest(), "u1")  # type: ignore[arg-type]
        await _anext_timeout(gen)  # ready
        # подменить таймаут на минимум, чтобы не ждать 20 секунд
        saved = events_api.PING_SECONDS
        events_api.PING_SECONDS = 0.01
        try:
            return await _anext_timeout(gen)
        finally:
            events_api.PING_SECONDS = saved
            await gen.aclose()

    frame = asyncio.run(scenario())
    assert '"ping"' in frame


# --- HTTP-публикации (перехват publish: TestClient стрим не читает) ---


@pytest.fixture
def published(monkeypatch: pytest.MonkeyPatch) -> list[Event]:
    """Перехват publish в модуле api.tasks: события собираются в список."""
    seen: list[Event] = []
    import app.api.tasks as tasks_api

    def _spy(user_id: str | None, kind: str, data: dict[str, Any] | None = None) -> None:
        seen.append(Event(kind=kind, data=data))
        publish(user_id, kind, data)

    monkeypatch.setattr(tasks_api, "publish", _spy)
    return seen


def test_create_task_publishes(client: Any, published: list[Event]) -> None:
    """Создание задачи → task.changed + xp.changed (без конфетти)."""
    created = client.post("/api/tasks", json={"title": "Тест SSE"})
    assert created.status_code == 200
    kinds = [e.kind for e in published]
    assert "task.changed" in kinds
    assert "xp.changed" in kinds
    xp = next(e for e in published if e.kind == "xp.changed")
    assert xp.data == {"celebrate": False}


def test_close_task_publishes_xp(client: Any, published: list[Event]) -> None:
    """Закрытие задачи → task.changed + xp.changed {celebrate: True} + garden.changed."""
    task_id = client.post("/api/tasks", json={"title": "Закрыть"}).json()["id"]
    published.clear()
    approved = client.post(f"/api/tasks/{task_id}/approve")
    assert approved.status_code == 200
    published.clear()
    closed = client.patch(f"/api/tasks/{task_id}", json={"status": "done"})
    assert closed.status_code == 200
    kinds = [e.kind for e in published]
    assert "task.changed" in kinds
    assert "xp.changed" in kinds
    assert "garden.changed" in kinds
    xp = next(e for e in published if e.kind == "xp.changed")
    assert xp.data is not None and xp.data["celebrate"] is True


def test_delete_task_publishes_deleted(client: Any, published: list[Event]) -> None:
    """Удаление задачи → task.deleted {id}."""
    task_id = client.post("/api/tasks", json={"title": "Удалить"}).json()["id"]
    published.clear()
    deleted = client.delete(f"/api/tasks/{task_id}")
    assert deleted.status_code == 200
    assert [e for e in published if e.kind == "task.deleted" and e.data == {"id": task_id}]