"""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
import threading
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()
        # Events published since the last flush was scheduled. A cascade
        # publishes 13-16 of them and each used to cost its own
        # `call_soon_threadsafe`; they now cost one between two turns of the
        # loop, which is what a wave is.
        self._pending: list[dict[str, Any]] = []
        self._pending_lock = threading.Lock()
        self._flush_scheduled = False
        # 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
        with self._pending_lock:
            self._pending.append(event)
            if self._flush_scheduled:
                # A flush is already on its way and has not run yet, so it
                # will find this event too.
                return
            self._flush_scheduled = True
        try:
            loop.call_soon_threadsafe(self._dispatch)
        except RuntimeError:
            # Loop already closed — shutting down.
            with self._pending_lock:
                self._pending.clear()
                self._flush_scheduled = False

    def _dispatch(self) -> None:
        """Hand everything published since the last turn to every subscriber."""
        with self._pending_lock:
            batch = self._pending
            self._pending = []
            self._flush_scheduled = False
        if not batch:
            return
        for queue in self._subscribers:
            for event in batch:
                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()