"""BroadcastGroup: fan-out a single provider stream to N simultaneous clients.

Recovery model
--------------
The pump loop runs indefinitely while self._running is True.  Each iteration
spawns an ffmpeg process in copy (remux) mode.  ffmpeg owns the upstream
HTTP connection and uses its built-in -reconnect_streamed flag to handle
short provider interruptions (e.g. expiring session tokens) transparently,
normalising PTS/DTS in the output so clients never see an A/V discontinuity.

If ffmpeg itself exits (unrecoverable error or stall beyond its own timeout)
the outer loop drains all client queues, backs off briefly, and spawns a
fresh ffmpeg process — again without disconnecting clients.

SENTINEL is only pushed when stop() is called or the pump truly gives up.

Stall detection
---------------
ffmpeg's -timeout option (socket-level read timeout) handles provider stalls.
A safety-net asyncio.wait_for wraps each stdout read in case ffmpeg hangs
without producing output or exiting.

A/V sync
--------
ffmpeg re-muxes the incoming MPEG-TS into a fresh output stream.  On every
reconnect the output PTS/DTS sequence is continuous regardless of provider-
side timestamp resets, eliminating audio desync entirely.  No video or audio
is re-encoded — quality is bit-for-bit identical to the source.
"""
import asyncio
import logging
import os
import shutil
import time
import uuid
from dataclasses import dataclass, field
from typing import AsyncIterator

from ..config import settings

logger = logging.getLogger(__name__)

CHUNK_SIZE    = settings.stream_chunk_size
QUEUE_MAX     = settings.stream_queue_maxsize
STALL_TIMEOUT = 10.0   # seconds — passed to ffmpeg -timeout (in µs below)
MAX_RETRIES   = 999    # effectively infinite; pool stops the group when 0 clients
_BACKOFF_CAP  = 5.0    # max seconds between outer retries

_SENTINEL = object()   # signals end-of-stream to client queues

_FFMPEG = shutil.which("ffmpeg") or "ffmpeg"


@dataclass
class ClientHandle:
    client_id: str
    queue: asyncio.Queue = field(default_factory=lambda: asyncio.Queue(maxsize=QUEUE_MAX))
    connected_at: float  = field(default_factory=time.time)
    is_priority: bool    = False
    username: str        = ""
    _overflow_count: int = 0
    _overflow_since: float | None = None

    async def read(self) -> AsyncIterator[bytes]:
        while True:
            chunk = await self.queue.get()
            if chunk is _SENTINEL:
                break
            yield chunk


class BroadcastGroup:
    """Streams one provider URL and distributes chunks to all registered clients."""

    def __init__(
        self,
        channel_id: int,
        stream_url: str,
        provider_account_id: int,
        channel_name: str = "",
        provider_name: str = "",
    ):
        self.channel_id          = channel_id
        self.channel_name        = channel_name
        self.stream_url          = stream_url
        self.provider_account_id = provider_account_id
        self.provider_name       = provider_name
        self.started_at          = time.time()

        self._clients: dict[str, ClientHandle] = {}
        self._lock    = asyncio.Lock()
        self._task: asyncio.Task | None = None
        self._running = False

        # Throughput tracking
        self._bytes_pumped: int   = 0
        self._bytes_prev:   int   = 0
        self._bps_time:     float = time.time()
        self._bps_down:     float = 0.0

        # Health / metadata
        self.stream_info:      dict  = {}
        self._reconnect_count: int   = 0
        self._status:          str   = "connecting"
        self._last_data_at:    float = time.time()

        # ffmpeg process tracking
        self._ffmpeg_pid:        int | None   = None
        self._ffmpeg_start_time: float | None = None   # wall time when ffmpeg started
        self._ffmpeg_cpu:        float        = 0.0
        self._cpu_task:          asyncio.Task | None = None

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    @property
    def client_count(self) -> int:
        return len(self._clients)

    @property
    def client_ids(self) -> list[str]:
        return list(self._clients.keys())

    @property
    def all_non_priority(self) -> bool:
        return all(not h.is_priority for h in self._clients.values())

    async def start(self) -> None:
        self._running = True
        self._task = asyncio.create_task(
            self._pump_with_retry(), name=f"pump-ch{self.channel_id}"
        )
        self._cpu_task = asyncio.create_task(
            self._monitor_cpu(), name=f"cpu-ch{self.channel_id}"
        )
        asyncio.create_task(self._probe_info(), name=f"probe-ch{self.channel_id}")

    async def add_client(self, user_id: str, username: str = "", is_priority: bool = False) -> ClientHandle:
        client_id = f"{user_id}-{uuid.uuid4().hex[:8]}"
        handle = ClientHandle(client_id=client_id, is_priority=is_priority, username=username)
        async with self._lock:
            self._clients[client_id] = handle
        logger.info(
            f"Client {client_id} ({username}) joined channel {self.channel_id} "
            f"(total={len(self._clients)})"
        )
        return handle

    async def force_disconnect_client(self, client_id: str) -> bool:
        async with self._lock:
            handle = self._clients.get(client_id)
            if handle is None:
                return False
            _drain_queue(handle.queue)
            try:
                handle.queue.put_nowait(_SENTINEL)
            except asyncio.QueueFull:
                pass
        return True

    async def remove_client(self, client_id: str) -> int:
        async with self._lock:
            self._clients.pop(client_id, None)
        remaining = len(self._clients)
        logger.info(f"Client {client_id} left channel {self.channel_id} (remaining={remaining})")
        return remaining

    async def stop(self) -> None:
        self._running = False
        self._status  = "stopped"
        if self._cpu_task and not self._cpu_task.done():
            self._cpu_task.cancel()
        if self._task and not self._task.done():
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
        async with self._lock:
            for handle in self._clients.values():
                try:
                    handle.queue.put_nowait(_SENTINEL)
                except asyncio.QueueFull:
                    pass

    def stats(self) -> dict:
        n = self.client_count or 1
        return {
            "stream_type":         "live",
            "channel_id":          self.channel_id,
            "channel_name":        self.channel_name or f"Canal #{self.channel_id}",
            "provider_account_id": self.provider_account_id,
            "provider_name":       self.provider_name,
            "client_count":        self.client_count,
            "clients": [
                {
                    "client_id": h.client_id,
                    "username":  h.username or h.client_id.rsplit("-", 1)[0],
                }
                for h in self._clients.values()
            ],
            "started_at":       self.started_at,
            "bytes_pumped":     self._bytes_pumped,
            "bps_down":         self._bps_down,
            "bps_up":           self._bps_down * n,
            "running":          self._running,
            "status":           self._status,
            "reconnect_count":  self._reconnect_count,
            "last_data_ago":    round(time.time() - self._last_data_at, 1),
            "stream_info":      self.stream_info,
            # ffmpeg process info
            "ffmpeg_pid":       self._ffmpeg_pid,
            "ffmpeg_cpu_pct":   round(self._ffmpeg_cpu, 2),
        }

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    async def _monitor_cpu(self) -> None:
        """Update self._ffmpeg_cpu every 3 s using cumulative average since start.

        Delta-based sampling fails for low-CPU processes (copy-mode ffmpeg uses
        <1% CPU) because the tick delta in 2 s is 0-1 integer ticks, making the
        result always 0.0%.  Cumulative average ((total_cpu_ticks / clk_tck) /
        elapsed_wall) gives a stable, accurate reading — the same method ps uses.
        """
        clk_tck = float(os.sysconf(os.sysconf_names.get("SC_CLK_TCK", 2)) or 100)

        while self._running:
            await asyncio.sleep(3)
            pid        = self._ffmpeg_pid
            start_time = self._ffmpeg_start_time
            if pid is None or start_time is None:
                self._ffmpeg_cpu = 0.0
                continue
            try:
                with open(f"/proc/{pid}/stat") as f:
                    parts = f.read().split()
                cpu_ticks = int(parts[13]) + int(parts[14])  # utime + stime
                elapsed   = time.time() - start_time
                if elapsed > 0:
                    self._ffmpeg_cpu = cpu_ticks / clk_tck / elapsed * 100
            except (FileNotFoundError, IndexError, ValueError, OSError):
                self._ffmpeg_cpu = 0.0

    async def _probe_info(self) -> None:
        from .probe import probe_stream
        try:
            info = await probe_stream(self.stream_url)
            self.stream_info.update(info)
        except Exception as e:
            logger.debug(f"probe_info failed for ch{self.channel_id}: {e}")

    async def _pump_with_retry(self) -> None:
        consecutive = 0
        base_wait   = 0.5

        try:
            while self._running:
                try:
                    await self._pump_once()
                    break
                except asyncio.CancelledError:
                    raise
                except Exception as e:
                    if not self._running:
                        break
                    consecutive += 1
                    self._reconnect_count += 1
                    wait = min(base_wait * (2 ** (consecutive - 1)), _BACKOFF_CAP)
                    self._status = "reconnecting"
                    logger.warning(
                        f"[ch{self.channel_id}] ffmpeg exited (attempt {consecutive}), "
                        f"restarting in {wait:.1f}s: {e}"
                    )
                    # Drain queues so stale buffered chunks are not delivered
                    # after the fresh ffmpeg process starts.
                    async with self._lock:
                        for handle in self._clients.values():
                            _drain_queue(handle.queue)
                    await asyncio.sleep(wait)
        finally:
            self._status  = "stopped"
            self._running = False
            async with self._lock:
                for handle in self._clients.values():
                    _drain_queue(handle.queue)
                    try:
                        handle.queue.put_nowait(_SENTINEL)
                    except asyncio.QueueFull:
                        pass

    async def _pump_once(self) -> None:
        """
        Spawn ffmpeg in copy (remux) mode and pump its stdout to all client
        queues until ffmpeg exits or self._running goes False.

        ffmpeg handles HTTP reconnects internally (-reconnect_streamed), which
        covers provider session-token refreshes without any visible interruption
        to clients.  PTS/DTS values are normalised by ffmpeg across reconnects,
        eliminating A/V desync entirely.
        """
        cmd = [
            _FFMPEG,
            "-hide_banner", "-loglevel", "warning",
            # Follow HTTP redirects and reconnect when the connection drops.
            # -reconnect_streamed is the key flag: it retries even on non-
            # seekable (live) streams, re-issuing the original GET so that
            # providers using short-lived 302 tokens get a fresh token.
            "-reconnect",           "1",
            "-reconnect_streamed",  "1",
            "-reconnect_delay_max", "4",
            # Socket-level read timeout in microseconds.  If the provider
            # stops sending bytes for STALL_TIMEOUT seconds ffmpeg closes
            # the connection and tries to reconnect (or exits if it gives up).
            "-timeout", str(int(STALL_TIMEOUT * 1_000_000)),
            "-i", self.stream_url,
            "-c", "copy",   # remux only — zero transcoding, bit-for-bit quality
            "-f", "mpegts",
            "pipe:1",
        ]

        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        self._status             = "ok"
        self._last_data_at       = time.time()
        self._ffmpeg_pid         = proc.pid
        self._ffmpeg_start_time  = time.time()
        logger.info(
            f"[ch{self.channel_id}] ffmpeg started pid={proc.pid} "
            f"(reconnects so far: {self._reconnect_count})"
        )

        # Drain stderr in background so the pipe never fills and blocks ffmpeg.
        stderr_task = asyncio.create_task(self._log_ffmpeg_stderr(proc))

        try:
            while self._running:
                try:
                    chunk = await asyncio.wait_for(
                        proc.stdout.read(CHUNK_SIZE),
                        timeout=STALL_TIMEOUT + 15,  # safety net beyond ffmpeg's own timeout
                    )
                except asyncio.TimeoutError:
                    raise RuntimeError(
                        f"ffmpeg produced no output for {STALL_TIMEOUT + 15:.0f}s"
                    )

                if not chunk:
                    # ffmpeg exited cleanly or with error
                    rc = proc.returncode
                    raise RuntimeError(f"ffmpeg exited (rc={rc})")

                # Throughput accounting
                self._bytes_pumped += len(chunk)
                self._last_data_at  = time.time()
                now = self._last_data_at
                if now - self._bps_time >= 2.0:
                    self._bps_down   = (self._bytes_pumped - self._bytes_prev) / (now - self._bps_time)
                    self._bytes_prev = self._bytes_pumped
                    self._bps_time   = now

                await self._dispatch(chunk, now)

        finally:
            self._ffmpeg_pid        = None
            self._ffmpeg_start_time = None
            self._ffmpeg_cpu        = 0.0
            stderr_task.cancel()
            try:
                proc.kill()
            except ProcessLookupError:
                pass
            try:
                await asyncio.wait_for(proc.wait(), timeout=5)
            except asyncio.TimeoutError:
                pass

    async def _log_ffmpeg_stderr(self, proc: asyncio.subprocess.Process) -> None:
        """Read ffmpeg stderr lines and forward them to the logger."""
        try:
            async for line in proc.stderr:
                text = line.decode(errors="replace").rstrip()
                if text:
                    logger.warning(f"[ch{self.channel_id}] ffmpeg: {text}")
        except asyncio.CancelledError:
            pass
        except Exception:
            pass

    async def _dispatch(self, chunk: bytes, now: float) -> None:
        """Push chunk to every client queue; evict clients that have been full for >30 s."""
        async with self._lock:
            dead: list[str] = []
            for cid, handle in self._clients.items():
                try:
                    handle.queue.put_nowait(chunk)
                    handle._overflow_count = 0
                    handle._overflow_since = None
                except asyncio.QueueFull:
                    handle._overflow_count += 1
                    if handle._overflow_since is None:
                        handle._overflow_since = now
                    elif now - handle._overflow_since > 30:
                        dead.append(cid)

            for cid in dead:
                logger.warning(
                    f"[ch{self.channel_id}] Evicting unresponsive client {cid} "
                    f"after 30 s of sustained queue overflow"
                )
                dropped = self._clients.pop(cid, None)
                if dropped is not None:
                    _drain_queue(dropped.queue)
                    try:
                        dropped.queue.put_nowait(_SENTINEL)
                    except asyncio.QueueFull:
                        pass


def _drain_queue(q: asyncio.Queue) -> None:
    """Empty a queue non-blockingly."""
    while not q.empty():
        try:
            q.get_nowait()
        except asyncio.QueueEmpty:
            break