"""Single-fetch hub for HLS segments + LRU ring buffer.

Why this exists
───────────────
Live IPTV via HLS is bursty: every ~6 s, every viewer of the same
channel asks for the same .ts segment within a 1-2 s window. If the
proxy fetches each segment N times (once per viewer) the upstream
provider sees N parallel connections and our outbound bandwidth grows
linearly with viewer count.

`SegmentHub.get_or_fetch` deduplicates: the FIRST request kicks off
the upstream fetch, while subsequent requests for the SAME segment
key wait on an `asyncio.Event` and reuse the cached bytes. Once a
segment has been fetched it stays in the ring buffer for `ttl_seconds`
so the second wave of viewers — those whose players are 5-10 s behind
live — also gets a memory hit.

Memory budget (May 2026 brief)
──────────────────────────────
  • Ring buffer cap: 200 MB total (well within the 3 GB host RAM).
  • TTL: 60 s — typical live HLS segments are 4-10 s long, so a 60 s
    window covers most "lagging viewer" cases.
  • Eviction: oldest-entry first (LRU on insertion order). When the
    buffer would exceed `max_bytes`, we drop entries one-by-one until
    the new segment fits.

The hub is NOT a general HTTP cache — it intentionally only stores
segment BODIES (bytes) keyed by upstream URL. Playlist refreshes are
handled separately because they have very different freshness needs.
"""
from __future__ import annotations

import asyncio
import logging
import time
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Optional

import httpx

logger = logging.getLogger(__name__)


@dataclass
class _Entry:
    key: str
    upstream_url: str
    data: Optional[bytes] = None
    content_type: str = "video/mp2t"
    error: Optional[BaseException] = None
    ready: asyncio.Event = field(default_factory=asyncio.Event)
    inserted_at: float = field(default_factory=time.monotonic)
    last_access: float = field(default_factory=time.monotonic)
    # True only for the entry that triggered the upstream fetch. All
    # subsequent requests waiting on the same key see this as False
    # and just `await ready` instead of re-running the fetch.
    fetch_started: bool = False


class SegmentHub:
    """Async-safe single-fetch + ring buffer.

    Usage::

        hub = SegmentHub(max_bytes=200 * 1024 * 1024, ttl_seconds=60)
        body, ct = await hub.get_or_fetch(
            key="live:abc:123:seg18.ts",
            upstream_url="http://provider.com/.../seg18.ts",
            client=async_http_client,
            request_timeout=15.0,
        )
    """

    def __init__(self, max_bytes: int = 200 * 1024 * 1024, ttl_seconds: int = 60):
        self.max_bytes = int(max_bytes)
        self.ttl_seconds = float(ttl_seconds)
        self._entries: "OrderedDict[str, _Entry]" = OrderedDict()
        self._total_bytes = 0
        self._lock = asyncio.Lock()
        # Telemetry — exposed via /api/proxy/stats so the admin widget
        # can show single-fetch effectiveness (hits / misses) and how
        # many upstream connections are currently in flight (one per
        # unique key being fetched).
        self._hits = 0
        self._misses = 0
        self._inflight = 0

    # ── Stats (tiny, for /api/proxy/stats debugging if ever exposed) ──
    @property
    def cached_count(self) -> int:
        return len(self._entries)

    @property
    def total_bytes(self) -> int:
        return self._total_bytes

    @property
    def hits(self) -> int:
        return self._hits

    @property
    def misses(self) -> int:
        return self._misses

    @property
    def inflight(self) -> int:
        return self._inflight

    # ── Internal: drop expired entries (called under the lock) ────────
    def _evict_expired_locked(self) -> None:
        now = time.monotonic()
        dead = [k for k, e in self._entries.items()
                if e.data is not None and (now - e.inserted_at) > self.ttl_seconds]
        for k in dead:
            entry = self._entries.pop(k)
            if entry.data is not None:
                self._total_bytes -= len(entry.data)

    def _evict_until_fits_locked(self, incoming: int) -> None:
        """Drop the oldest fully-loaded entries until room exists."""
        while self._total_bytes + incoming > self.max_bytes and self._entries:
            # Find the OLDEST loaded entry (skip in-flight entries).
            for key in list(self._entries.keys()):
                entry = self._entries[key]
                if entry.data is not None:
                    self._entries.pop(key)
                    self._total_bytes -= len(entry.data)
                    break
            else:
                # All entries are in-flight — can't evict; let the new
                # one go through anyway (we'll exceed max briefly).
                break

    # ── Public API ────────────────────────────────────────────────────
    async def get_or_fetch(
        self,
        *,
        key: str,
        upstream_url: str,
        client: httpx.AsyncClient,
        request_timeout: float = 15.0,
        upstream_headers: Optional[dict] = None,
    ) -> tuple[bytes, str]:
        """Return the cached or freshly-fetched segment bytes."""
        async with self._lock:
            self._evict_expired_locked()
            entry = self._entries.get(key)
            if entry is not None:
                # Already known. Move to LRU end + bump last_access.
                self._entries.move_to_end(key, last=True)
                entry.last_access = time.monotonic()
                is_creator = False
                # If the entry already has data we count this as a hit
                # right away. If the fetch is still in-flight we leave
                # the hit/miss count alone — the eventual result will
                # be the same bytes everyone is waiting for, so it
                # would be misleading to count waiters as hits AND the
                # creator as a miss.
                if entry.ready.is_set() and entry.error is None:
                    self._hits += 1
            else:
                entry = _Entry(key=key, upstream_url=upstream_url)
                entry.fetch_started = True  # SET INSIDE THE LOCK
                self._entries[key] = entry
                is_creator = True
                self._misses += 1
                self._inflight += 1

        # Already cached → return immediately.
        if entry.ready.is_set():
            if entry.error is not None:
                raise entry.error
            assert entry.data is not None
            return entry.data, entry.content_type

        if not is_creator:
            # Wait for the creator to finish the upstream fetch.
            await entry.ready.wait()
            if entry.error is not None:
                raise entry.error
            assert entry.data is not None
            return entry.data, entry.content_type

        # We are the creator → run the fetch.
        try:
            headers = dict(upstream_headers or {})
            headers.setdefault("User-Agent", "Nadiplayer/1.3 (+proxy)")
            r = await client.get(upstream_url, headers=headers, timeout=request_timeout)
            r.raise_for_status()
            body = r.content
            ct = r.headers.get("content-type", "video/mp2t")
            async with self._lock:
                self._evict_until_fits_locked(len(body))
                entry.data = body
                entry.content_type = ct
                entry.inserted_at = time.monotonic()
                entry.last_access = entry.inserted_at
                self._total_bytes += len(body)
                self._entries.move_to_end(key, last=True)
                self._inflight = max(0, self._inflight - 1)
            entry.ready.set()
            return body, ct
        except BaseException as exc:
            entry.error = exc
            entry.ready.set()
            async with self._lock:
                if self._entries.get(key) is entry:
                    self._entries.pop(key, None)
                self._inflight = max(0, self._inflight - 1)
            logger.warning("[SegmentHub] fetch %s failed: %s", upstream_url, exc)
            raise


# Singleton — reused across requests in the FastAPI process. The hub
# is purely in-memory so no extra setup / teardown is required.
SEGMENT_HUB = SegmentHub(max_bytes=10 * 1024 * 1024, ttl_seconds=30)
