"""Server-sent events for the live feed: poll `events` every `POLL_S` with a `(detected_at, id)` cursor, heartbeat every `HEARTBEAT_S`.

The first message is always a heartbeat carrying the cursor so clients (and proxies) see bytes immediately; reconnecting clients
pass the last cursor back as `?since=`. `max_s` bounds the connection so upstream proxies with idle timeouts can reconnect cleanly.
"""
from __future__ import annotations

import asyncio
import logging
from collections.abc import AsyncIterator
from datetime import datetime
from typing import Any

import orjson
from fastapi import Request
from sse_starlette.sse import EventSourceResponse

from companyatlas.api import queries as q
from companyatlas.api import serializers as ser
from companyatlas.db import connection

log = logging.getLogger("companyatlas.api.sse")

POLL_S = 5.0
HEARTBEAT_S = 20.0
BATCH = 50
MAX_STREAM_S = 3600
SSE_HEADERS = {"cache-control": "no-store, no-transform", "x-accel-buffering": "no", "connection": "keep-alive"}


def _dump(payload: Any) -> str:
    return orjson.dumps(payload, option=orjson.OPT_UTC_Z, default=str).decode()


async def poll_new_events(since: datetime | None, last_id: str | None, *, event_type: str | None = None, min_importance: float | None = None,
                          limit: int = BATCH) -> list[dict[str, Any]]:
    where, params = q.event_filters(event_type=event_type, min_importance=min_importance)
    if since is not None:
        if last_id:
            where.append("(e.detected_at, e.id) > (:cur_at, :cur_id)")
            params.update(cur_at=since, cur_id=last_id)
        else:
            where.append("e.detected_at > :cur_at")
            params["cur_at"] = since
    async with connection() as conn:
        rows = await q.fetch_events(conn, where, params, sort="recent", limit=limit)
    return [ser.event(r) for r in reversed(rows)]      # oldest first so clients append in order


async def live_event_stream(request: Request | None, *, since: datetime | None, event_type: str | None = None,
                            min_importance: float | None = None, max_s: float = MAX_STREAM_S) -> AsyncIterator[dict[str, Any]]:
    loop = asyncio.get_running_loop()
    started = loop.time()
    cursor_at: datetime | None = since or q.now_utc()
    cursor_id: str | None = None
    last_sent = loop.time()
    yield {"event": "heartbeat", "data": _dump({"at": q.now_utc(), "cursor": cursor_at})}
    while True:
        if request is not None and await request.is_disconnected():
            return
        try:
            events = await poll_new_events(cursor_at, cursor_id, event_type=event_type, min_importance=min_importance)
        except Exception:
            log.warning("live stream poll failed", exc_info=True)
            events = []
        for ev in events:
            cursor_at, cursor_id = ev["detected_at"], ev["id"]
            last_sent = loop.time()
            yield {"event": "event", "id": ev["id"], "data": _dump(ev)}
        now = loop.time()
        if now - last_sent >= HEARTBEAT_S:
            last_sent = now
            yield {"event": "heartbeat", "data": _dump({"at": q.now_utc(), "cursor": cursor_at})}
        remaining = max_s - (now - started)
        if remaining <= 0:
            yield {"event": "end", "data": _dump({"cursor": cursor_at, "reason": "max_s reached"})}
            return
        await asyncio.sleep(min(POLL_S, remaining))


def sse_response(generator: AsyncIterator[dict[str, Any]]) -> EventSourceResponse:
    return EventSourceResponse(generator, headers=SSE_HEADERS, ping=HEARTBEAT_S * 3)


__all__ = ["HEARTBEAT_S", "MAX_STREAM_S", "POLL_S", "live_event_stream", "poll_new_events", "sse_response"]