"""Fan-out of pipeline result frames and status to connected web clients.

A single broadcaster task polls the :class:`WebController` off the event loop and
pushes the freshest frame (latest-wins) plus a slower status heartbeat to every
registered client. Each client is a bounded ``asyncio.Queue`` with a drop-oldest
policy, so a slow socket can never stall the producer or the loop — if a client
falls behind it simply skips intermediate frames and always gets the newest.
"""

from __future__ import annotations

import asyncio
import contextlib
import logging

from python_app.webui.controller import WebController

logger = logging.getLogger(__name__)

# Poll the controller this often; the C++ pipeline publishes well below this rate,
# so this is a comfortable latest-wins cadence without busy-spinning the loop.
_FRAME_INTERVAL_S = 0.05
# Status is cheap but rarely changes; emit it about once a second.
_STATUS_INTERVAL_S = 1.0


class RingBroadcaster:
    """Polls the controller and fans frames/status out to all WebSocket clients."""

    def __init__(self, controller: WebController) -> None:
        self._controller = controller
        self._clients: set[asyncio.Queue[dict]] = set()
        self._task: asyncio.Task[None] | None = None
        self._last_frame_seq: int | None = None
        self._last_settings: dict | None = None

    def register(self, queue: asyncio.Queue[dict]) -> None:
        """Add a client queue to receive subsequent frames and status."""
        self._clients.add(queue)
        logger.debug("Registered web client queue (clients=%d)", len(self._clients))

    def unregister(self, queue: asyncio.Queue[dict]) -> None:
        """Remove a client queue; safe to call more than once."""
        self._clients.discard(queue)
        logger.debug("Unregistered web client queue (clients=%d)", len(self._clients))

    def start(self) -> None:
        """Launch the single polling task (idempotent)."""
        if self._task is None or self._task.done():
            self._task = asyncio.create_task(self._run(), name="ring-broadcaster")
            self._task.add_done_callback(self._on_task_done)
            logger.info("Ring broadcaster started")

    @staticmethod
    def _on_task_done(task: "asyncio.Task[None]") -> None:
        """Surface an unexpected broadcaster death (the loop should never exit)."""
        if not task.cancelled() and task.exception() is not None:
            logger.error("ring broadcaster task exited unexpectedly: %r", task.exception())

    async def stop(self) -> None:
        """Cancel the polling task and wait for it to unwind."""
        if self._task is None:
            return
        self._task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._task
        self._task = None
        logger.info("Ring broadcaster stopped")

    def _publish(self, message: dict) -> None:
        """Push a message to every client, dropping the oldest on a full queue."""
        for queue in self._clients:
            if queue.full():
                with contextlib.suppress(asyncio.QueueEmpty):
                    queue.get_nowait()
            with contextlib.suppress(asyncio.QueueFull):
                queue.put_nowait(message)

    def _status_message(self) -> dict:
        """Build a status broadcast from the controller's current state."""
        return {"type": "status", **self._controller.status()}

    async def _run(self) -> None:
        """Poll on a fixed cadence; never block the loop or die on a bad frame."""
        loop = asyncio.get_running_loop()
        next_status = loop.time()
        while True:
            try:
                # peek_frame may touch shared memory / NumPy, so keep it off the loop.
                frame = await loop.run_in_executor(None, self._controller.peek_frame)
                if frame is not None and frame["seq"] != self._last_frame_seq:
                    self._last_frame_seq = frame["seq"]
                    self._publish(frame)

                now = loop.time()
                if now >= next_status:
                    self._publish(self._status_message())
                    # Push live settings (Qt -> web) only when they change, so the
                    # browser form mirrors desktop edits in real time without churn.
                    settings = self._controller.current_live_settings()
                    if settings != self._last_settings:
                        self._last_settings = settings
                        self._publish({"type": "settings", "schema": settings})
                    next_status = now + _STATUS_INTERVAL_S
            except asyncio.CancelledError:
                raise
            except Exception:  # noqa: BLE001 - one bad frame must not stop streaming
                logger.warning("ring broadcaster iteration failed; continuing", exc_info=True)
            await asyncio.sleep(_FRAME_INTERVAL_S)