"""Run-Ereignisse per SSE und Abbruch (→ Konzept 4.4, 11)."""

import asyncio
import uuid
from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import APIRouter, Depends, Header, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import Response
from fastapi.sse import EventSourceResponse, ServerSentEvent

from app import web
from app.tasks import engine, events, states
from app.tasks.models import RunEvent, TaskRun

# Wartezeit auf ein NOTIFY; ohne LISTEN-Verbindung kürzer (Fallback-Polling)
WAIT_SECONDS = 15.0
POLL_SECONDS = 2.0


def visible_run(request: Request, principal: web.User, run_id: uuid.UUID) -> uuid.UUID:
    """Als Dependency, damit Fremde ein 404 bekommen, bevor der Stream beginnt."""
    with web.db(request).session() as session:
        engine.get(session, principal, run_id)
    return run_id


def start_after(last_event_id: str | None, after: int | None) -> int:
    for value in (last_event_id, after):
        try:
            return max(int(value), 0)  # type: ignore[arg-type]
        except (TypeError, ValueError):
            continue
    return 0


def router() -> APIRouter:
    r = APIRouter()

    @r.get("/runs/{run_id}/events", response_class=EventSourceResponse)
    async def run_events(request: Request, run_id: Annotated[uuid.UUID, Depends(visible_run)],
                         last_event_id: Annotated[str | None, Header()] = None,
                         after: int | None = None) -> AsyncIterable[ServerSentEvent]:
        db, hub = web.db(request), request.app.state.hub
        position = start_after(last_event_id, after)

        def load(since: int) -> tuple[list[RunEvent], str]:
            # Erst den Status, dann die Ereignisse: Statuswechsel und Ereignis werden gemeinsam
            # committet. Andersherum könnte ein Abschluss zwischen beiden Abfragen liegen; der
            # Stream endete dann, ohne das letzte Ereignis gesendet zu haben.
            with db.session() as session:
                status = session.get(TaskRun, run_id).status
                rows = events.read(session, run_id, since)
            return rows, status

        wakeup = hub.subscribe(run_id)
        try:
            while not await request.is_disconnected():
                # Erst zurücksetzen, dann lesen: so geht kein NOTIFY zwischen Lesen und Warten verloren
                wakeup.clear()
                rows, status = await run_in_threadpool(load, position)
                for row in rows:
                    position = row.seq
                    yield ServerSentEvent(data={"seq": row.seq, **row.data}, event=row.type, id=str(row.seq))
                if rows:
                    continue
                if status in states.RESTING:
                    yield ServerSentEvent(data={"status": status}, event="end")
                    return
                try:
                    await asyncio.wait_for(wakeup.wait(), WAIT_SECONDS if hub.connected else POLL_SECONDS)
                except TimeoutError:
                    yield ServerSentEvent(comment="ping")
        finally:
            hub.unsubscribe(run_id, wakeup)

    @r.post("/runs/{run_id}/cancel")
    def cancel(request: Request, principal: web.Checked, run_id: uuid.UUID) -> Response:
        with web.db(request).session() as session:
            engine.request_cancel(session, principal, run_id)
        return Response(status_code=204)

    return r