"""SSE realtime endpoint. One long-lived stream per connected client,
PERSPECTIVE-FILTERED per subscriber (M6, plan §6.3: "a Tingang touch never
even reaches Rian's event stream").

The bus (app.events) broadcasts raw truth; THIS boundary projects each
event for each subscriber through the same rep() machinery every read
uses (services.projection.project_event): ball events get rep()'d
assignments and shed invisible actors (or are dropped when the projected
movement is invisible), and notification.created is delivered ONLY to its
recipient — the M5 bell used to broadcast to every connected queue.

The subscriber's viewer is resolved ONCE at connect (session player, or
the View As / dev-header target — the SPA reconnects the stream when
view-as changes). Topology is loaded per event from the event's match_id:
the tables are four tiny reads and events are human-paced, and a fresh
load means a team-tree restructure re-projects the live stream instantly.
Both DB touches run in a worker thread (asyncio.to_thread) so the shared
event loop never blocks on the database (react.md: sync for DB, async
only for SSE).
"""
from __future__ import annotations

import asyncio
import json
from collections.abc import AsyncIterator

from fastapi import APIRouter, Request
from sse_starlette.sse import EventSourceResponse

from .. import events
from ..db import SessionLocal
from ..services import projection as projection_service
from ..services import viewer as viewer_service

router = APIRouter()

_HEARTBEAT_SECONDS = 20


def _resolve_viewer_id(request: Request):
    """The subscriber's effective player id (None = outside observer),
    resolved with a short-lived session — never one held for the stream's
    lifetime."""
    db = SessionLocal()
    try:
        viewer = viewer_service.resolve_viewer(request, db)
        return viewer.id if viewer is not None else None
    finally:
        db.close()


def _project_for(evt: dict, viewer_id) -> dict | None:
    """Load the event's topology (when it names a match) and project.
    Runs in a worker thread per event."""
    db = SessionLocal()
    try:
        topology = projection_service.topology_for_event(db, evt)
        return projection_service.project_event(evt, topology, viewer_id)
    finally:
        db.close()


@router.get("/api/events")
async def stream(request: Request) -> EventSourceResponse:
    viewer_id = await asyncio.to_thread(_resolve_viewer_id, request)
    q = events.subscribe()

    async def gen() -> AsyncIterator[dict]:
        # Announce readiness so the client knows the stream is live, with
        # the current on-court set so a fresh tab paints presence at once
        # (M3; subsequent changes arrive as presence.updated events).
        events.presence_connect(viewer_id)
        yield {
            "event": "hello",
            "data": json.dumps(
                {
                    "viewer": str(viewer_id) if viewer_id else None,
                    "online": events.online_ids(),
                }
            ),
        }
        try:
            while True:
                if await request.is_disconnected():
                    break
                try:
                    evt = await asyncio.wait_for(q.get(), timeout=_HEARTBEAT_SECONDS)
                except asyncio.TimeoutError:
                    # Comment heartbeat keeps proxies (Caddy) from idling us out.
                    yield {"event": "ping", "data": "{}"}
                    continue
                projected = await asyncio.to_thread(_project_for, evt, viewer_id)
                if projected is None:
                    continue  # no visible change for THIS viewer — dropped
                yield {
                    "event": projected.get("type", "message"),
                    "data": json.dumps(projected),
                }
        finally:
            events.unsubscribe(q)
            events.presence_disconnect(viewer_id)

    return EventSourceResponse(gen())