"""Event bus bridging the engine's worker threads to async subscribers.

Nodes execute in a thread pool; websocket clients live on the event loop.
Publishers are therefore thread-safe and never block: a subscriber that cannot
keep up loses its oldest queued events rather than stalling the engine.
"""

from __future__ import annotations

import asyncio
import logging
from collections import deque
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any

logger = logging.getLogger(__name__)

QUEUE_SIZE = 256
LOG_HISTORY = 400


class EventBus:
    """Fan-out of engine events to any number of async subscribers."""

    def __init__(self) -> None:
        self._loop: asyncio.AbstractEventLoop | None = None
        self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
        # Kept whether or not anyone is listening, so opening the log panel
        # shows what just happened rather than an empty box.
        self.recent_logs: deque[dict[str, Any]] = deque(maxlen=LOG_HISTORY)
        # How often each qualified node has emitted, for the same reason: a
        # client that reconnects between two pages would otherwise start from
        # zero and draw a busy graph as idle. A session count, not a metric —
        # the rollups in `metrics.py` are what survives a restart.
        self.emits: dict[str, int] = {}

    def bind(self, loop: asyncio.AbstractEventLoop) -> None:
        """Attach the bus to the running event loop (called once at startup)."""
        self._loop = loop

    def publish(self, event: dict[str, Any]) -> None:
        """Publish an event from any thread."""
        kind = event.get("type")
        if kind == "node_log":
            # deque.append is atomic, so worker threads need no lock here.
            self.recent_logs.append(event)
        elif kind == "node_executed" and event.get("outputs"):
            # The condition the canvas animates on, so both ends agree on what
            # counts as an emission. Two worker threads racing here can lose an
            # increment, which is a count nobody is going to miss.
            node = str(event.get("node") or "")
            self.emits[node] = self.emits.get(node, 0) + 1
        loop = self._loop
        if loop is None or not self._subscribers:
            return
        try:
            loop.call_soon_threadsafe(self._dispatch, event)
        except RuntimeError:
            # Loop already closed — shutting down.
            pass

    def _dispatch(self, event: dict[str, Any]) -> None:
        for queue in self._subscribers:
            if queue.full():
                # Drop the oldest so a slow client never blocks the engine.
                try:
                    queue.get_nowait()
                except asyncio.QueueEmpty:
                    pass
            queue.put_nowait(event)

    @asynccontextmanager
    async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]:
        """Yield a queue receiving every event published while subscribed."""
        queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_SIZE)
        self._subscribers.add(queue)
        try:
            yield queue
        finally:
            self._subscribers.discard(queue)


event_bus = EventBus()