hold-slayer/core/event_bus.py

"""
Event Bus — Async pub/sub for real-time gateway events.

WebSocket connections, MCP server, and internal services
all subscribe to events here. Pure asyncio — no external deps.
"""

import asyncio
import logging
from typing import Optional

from models.events import EventType, GatewayEvent

logger = logging.getLogger(__name__)


class EventBus:
    """
    Async pub/sub event bus using asyncio.Queue per subscriber.

    Features:
    - Non-blocking publish (put_nowait)
    - Slow subscribers lose their oldest event, never their subscription
    - Event history (last N events, replayable to late joiners)
    - Typed event filtering on subscriptions
    - Async iteration via EventSubscription
    """

    def __init__(self, max_history: int = 1000):
        self._subscribers: list[EventSubscription] = []
        self._history: list[GatewayEvent] = []
        self._max_history = max_history

    async def publish(self, event: GatewayEvent) -> None:
        """Publish an event to all subscribers.

        A full subscriber queue drops its oldest event (counted on the
        subscription) — a slow dashboard must never be silently
        unsubscribed while its socket stays open.
        """
        self._history.append(event)
        if len(self._history) > self._max_history:
            self._history = self._history[-self._max_history :]

        logger.info(f"📡 Event: {event.type.value} | {event.message or ''}")

        for sub in self._subscribers:
            if sub.type_filter and event.type not in sub.type_filter:
                continue
            sub.deliver(event)

    def subscribe(
        self,
        max_size: int = 100,
        event_types: Optional[set[EventType]] = None,
        replay_last: int = 0,
    ) -> "EventSubscription":
        """
        Create a new subscription.

        Args:
            max_size: Queue depth; overflow drops the oldest event.
            event_types: Optional filter — only receive these event types.
                         None means receive everything.
            replay_last: Seed the queue with up to N most recent
                         history events (post-filter) before live ones.

        Returns:
            An async iterator of GatewayEvents.
        """
        queue: asyncio.Queue[GatewayEvent] = asyncio.Queue(maxsize=max_size)
        sub = EventSubscription(queue, self, event_types)
        if replay_last > 0:
            replayable = [
                e for e in self._history
                if not event_types or e.type in event_types
            ]
            for event in replayable[-replay_last:]:
                sub.deliver(event)
        self._subscribers.append(sub)
        return sub

    def unsubscribe(self, sub: "EventSubscription") -> None:
        """Remove a subscriber."""
        if sub in self._subscribers:
            self._subscribers.remove(sub)

    @property
    def recent_events(self) -> list[GatewayEvent]:
        """Get recent event history."""
        return list(self._history)

    @property
    def subscriber_count(self) -> int:
        return len(self._subscribers)


class EventSubscription:
    """An async iterator that yields events from the bus."""

    def __init__(
        self,
        queue: asyncio.Queue[GatewayEvent],
        bus: EventBus,
        type_filter: Optional[set[EventType]] = None,
    ):
        self._queue = queue
        self._bus = bus
        self.type_filter = type_filter
        self.dropped = 0  # events lost to queue overflow

    def deliver(self, event: GatewayEvent) -> None:
        """Enqueue an event, dropping the oldest on overflow."""
        try:
            self._queue.put_nowait(event)
        except asyncio.QueueFull:
            try:
                self._queue.get_nowait()
                self._queue.put_nowait(event)
            except (asyncio.QueueEmpty, asyncio.QueueFull):
                pass
            self.dropped += 1
            if self.dropped in (1, 10, 100) or self.dropped % 1000 == 0:
                logger.warning(
                    f"📡 Slow subscriber: {self.dropped} events dropped"
                )

    def __aiter__(self):
        return self

    async def __anext__(self) -> GatewayEvent:
        try:
            return await self._queue.get()
        except asyncio.CancelledError:
            self._bus.unsubscribe(self)
            raise

    async def get(self, timeout: Optional[float] = None) -> GatewayEvent:
        """Get next event with optional timeout."""
        return await asyncio.wait_for(self._queue.get(), timeout=timeout)

    def close(self):
        """Unsubscribe from the event bus."""
        self._bus.unsubscribe(self)