"""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, and
can ask whether it did (``dropped``) to catch up another way.
"""

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
#: How many of a message's last artifact digests are remembered.
RECENT_DIGESTS = 8


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] = {}
        #: The last few artifacts each message pointed at, newest last. A
        #: screen fetching a camera frame by frame asks for one the message
        #: may already have moved past, and a device's credential reaches
        #: these as well as the current one.
        self.digests: dict[str, deque[str]] = {}
        #: Subscribers that lost an event since they last asked. Touched on
        #: the event loop only, by the flush and the subscribers.
        self._dropped: set[asyncio.Queue[dict[str, Any]]] = set()

    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
        elif kind == "message_value":
            value = event.get("value")
            if isinstance(value, dict) and isinstance(value.get("digest"), str):
                name = str(event.get("name") or "")
                recent = self.digests.get(name)
                if recent is None:
                    recent = self.digests[name] = deque(maxlen=RECENT_DIGESTS)
                recent.append(value["digest"])
        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
                    self._dropped.add(queue)
                queue.put_nowait(event)

    def dropped(self, queue: asyncio.Queue[dict[str, Any]]) -> bool:
        """Whether this subscriber lost events since it last asked.

        A device's socket that did catches up with a snapshot: what it lost
        may have been its demand, or the event that tells it it was unpaired.
        """
        if queue in self._dropped:
            self._dropped.discard(queue)
            return True
        return False

    @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)
            self._dropped.discard(queue)


event_bus = EventBus()