"""
Async event bus — fan-out from providers to consumers.

Features:
- Backpressure with smart drop policy (drop quotes before trades)
- Heartbeat timer per provider (detects dead channels)
- Graceful drain on shutdown

Usage:
    bus = EventBus()
    bus.add_consumer(storage_consumer)
    bus.add_consumer(metrics_consumer)

    async for event in provider.stream():
        await bus.publish(event)
"""
from __future__ import annotations

import asyncio
import logging
import time
from typing import Protocol

from app.domain.events import Event, EventType, HeartbeatEvent

logger = logging.getLogger(__name__)


class EventConsumer(Protocol):
    """Any callable that accepts an Event."""

    async def handle(self, event: Event) -> None: ...


# Events that can be dropped under backpressure (least critical first)
_DROPPABLE_PRIORITY = {
    EventType.HEARTBEAT: 0,  # always droppable
    EventType.QUOTE: 1,  # drop quotes before trades
    EventType.BOOK_L2: 2,  # drop book snapshots before trades
    EventType.TRADE: 3,  # trades are most critical — last to drop
}


class EventBus:
    """
    Async fan-out bus with backpressure and heartbeat monitoring.

    Backpressure policy:
    - Queue 80% full → start dropping HEARTBEAT events
    - Queue 90% full → also drop QUOTE events
    - Queue 100% full → drop oldest (any type)

    Heartbeat timer:
    - Emits synthetic HeartbeatEvent if a provider sends nothing
      for `heartbeat_interval` seconds, making dead channels visible.
    """

    def __init__(
        self,
        queue_size: int = 10_000,
        heartbeat_interval: float = 10.0,
    ) -> None:
        self._consumers: list[EventConsumer] = []
        self._queue: asyncio.Queue[Event | None] = asyncio.Queue(maxsize=queue_size)
        self._max_size = queue_size
        self._running = False
        self._task: asyncio.Task | None = None
        self._heartbeat_interval = heartbeat_interval
        self._heartbeat_tasks: dict[str, asyncio.Task] = {}
        self._provider_last_seen: dict[str, float] = {}
        # Backpressure counters
        self._dropped: dict[str, int] = {}

    def add_consumer(self, consumer: EventConsumer) -> None:
        self._consumers.append(consumer)
        logger.info("bus.consumer_added", extra={"consumer": type(consumer).__name__})

    def register_provider(self, provider_name: str) -> None:
        """Register a provider for heartbeat monitoring."""
        self._provider_last_seen[provider_name] = time.monotonic()

    async def publish(self, event: Event) -> None:
        """
        Put event into internal queue with backpressure.

        Drop policy under pressure:
        - 80%+ → drop heartbeats
        - 90%+ → drop quotes/book snapshots
        - 100% → drop oldest event
        """
        current = self._queue.qsize()
        fill_pct = current / self._max_size if self._max_size > 0 else 0

        # Track provider activity for heartbeat timer
        self._provider_last_seen[event.provider] = time.monotonic()

        priority = _DROPPABLE_PRIORITY.get(event.event_type, 3)

        # Backpressure: drop low-priority events when queue is filling up
        if fill_pct >= 0.9 and priority <= 1:
            # Drop heartbeats and quotes
            self._dropped[event.event_type.value] = self._dropped.get(event.event_type.value, 0) + 1
            if self._dropped[event.event_type.value] % 1000 == 1:
                logger.warning(
                    "bus.backpressure_drop",
                    extra={
                        "type": event.event_type.value,
                        "fill_pct": f"{fill_pct:.0%}",
                        "total_dropped": self._dropped,
                    },
                )
            return

        if fill_pct >= 0.8 and priority == 0:
            # Drop heartbeats only
            return

        try:
            self._queue.put_nowait(event)
        except asyncio.QueueFull:
            # Last resort: drop oldest to make room
            try:
                dropped = self._queue.get_nowait()
                logger.warning(
                    "bus.queue_full_drop_oldest",
                    extra={"dropped_type": dropped.event_type.value if dropped else "None"},
                )
            except asyncio.QueueEmpty:
                pass
            try:
                self._queue.put_nowait(event)
            except asyncio.QueueFull:
                pass  # truly stuck

    async def _worker(self) -> None:
        """Background worker that drains the queue and fans out."""
        while self._running:
            event = await self._queue.get()
            if event is None:
                break
            tasks = [c.handle(event) for c in self._consumers]
            if tasks:
                results = await asyncio.gather(*tasks, return_exceptions=True)
                for i, result in enumerate(results):
                    if isinstance(result, Exception):
                        consumer_name = type(self._consumers[i]).__name__
                        logger.error(
                            "bus.consumer_error",
                            extra={"consumer": consumer_name, "error": str(result)},
                        )

    async def _heartbeat_monitor(self, provider_name: str) -> None:
        """Emit synthetic heartbeat if provider goes silent."""
        while self._running:
            await asyncio.sleep(self._heartbeat_interval)
            if not self._running:
                break

            last = self._provider_last_seen.get(provider_name, 0)
            elapsed = time.monotonic() - last

            if elapsed > self._heartbeat_interval:
                # Provider is silent — emit heartbeat so metrics/logs see it
                logger.warning(
                    "bus.provider_silent",
                    extra={
                        "provider": provider_name,
                        "silent_seconds": f"{elapsed:.1f}",
                    },
                )
                hb = HeartbeatEvent(provider=provider_name)
                await self.publish(hb)

    async def start(self) -> None:
        """Start the bus worker and heartbeat monitors."""
        self._running = True
        self._task = asyncio.create_task(self._worker())

        # Start heartbeat monitors for registered providers
        for pname in self._provider_last_seen:
            task = asyncio.create_task(self._heartbeat_monitor(pname))
            self._heartbeat_tasks[pname] = task

        logger.info(
            "bus.started",
            extra={
                "consumers": len(self._consumers),
                "providers_monitored": list(self._provider_last_seen.keys()),
            },
        )

    async def stop(self) -> None:
        """Graceful shutdown: stop heartbeats, drain queue, stop worker."""
        self._running = False

        # Cancel heartbeat monitors
        for task in self._heartbeat_tasks.values():
            task.cancel()
        for task in self._heartbeat_tasks.values():
            try:
                await task
            except asyncio.CancelledError:
                pass
        self._heartbeat_tasks.clear()

        # Drain remaining events
        remaining = self._queue.qsize()
        if remaining > 0:
            logger.info("bus.draining", extra={"remaining": remaining})

        # Send sentinel to stop worker
        await self._queue.put(None)
        if self._task:
            await self._task

        if self._dropped:
            logger.info("bus.drop_stats", extra={"dropped": self._dropped})

        logger.info("bus.stopped")

    @property
    def queue_size(self) -> int:
        return self._queue.qsize()

    @property
    def fill_percent(self) -> float:
        return self._queue.qsize() / self._max_size if self._max_size > 0 else 0
