"""Stream Proxy / Restreaming router — restricted to Shared Profiles.

Architecture (May 2026 brief)
─────────────────────────────
This module exposes endpoints under `/api/proxy/...` that the React
player calls INSTEAD of the upstream Xtream server, but only when the
active profile is one of the admin's Shared Profiles. Per-user (own)
playlists and user-added private playlists keep going direct — the
proxy never sees their credentials.

Three groups of endpoints:

  1. POST /api/proxy/sign  (JWT-authenticated)
     → returns short-lived URLs that the client can hand to its
       <video> element / hls.js / ExoPlayer.

  2. GET  /api/proxy/live/{token}.m3u8       (token-only)
     GET  /api/proxy/seg/{token}             (token-only)
     → live HLS playlist + segments. Segments piggy-back on the
       SegmentHub for single-fetch + 200 MB ring buffer.

  3. GET  /api/proxy/vod/{token}.{ext}       (token-only)
     → movies + series episodes. Range-aware passthrough — single
       fetch is NOT applied here (each viewer has their own seek
       position so dedup buys nothing meaningful).

Token format / TTL: see core.proxy_signer (HMAC-SHA256, 30 min TTL,
keyed with the JWT_SECRET). All tokens carry the `pid` of a Shared
Profile (`shared:*`) — the sign endpoint refuses to issue tokens for
any other profile id, so the proxy is structurally locked to the
admin's curated catalogue.
"""
from __future__ import annotations

import asyncio
import base64
import logging
import re
import time
from typing import Any, Dict, Optional
from urllib.parse import quote, urljoin, urlparse

import httpx
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import PlainTextResponse, Response, StreamingResponse
from pydantic import BaseModel, Field
from starlette.status import HTTP_403_FORBIDDEN

from core.auth_deps import require_admin, require_user
from core.proxy_signer import DEFAULT_TTL_SECONDS, sign_token, verify_token
from core.stream_hub import SEGMENT_HUB

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/proxy", tags=["proxy"])

# ── Shared httpx client (one TCP connection pool for the whole hub) ──
_HTTP_CLIENT: Optional[httpx.AsyncClient] = None


def _get_client() -> httpx.AsyncClient:
    global _HTTP_CLIENT
    if _HTTP_CLIENT is None:
        # Generous limits — IPTV upstreams routinely keep connections
        # open for the duration of the live stream. The proxy itself
        # may hold dozens to hundreds of in-flight upstream sockets.
        _HTTP_CLIENT = httpx.AsyncClient(
            timeout=httpx.Timeout(connect=10.0, read=30.0, write=30.0, pool=10.0),
            limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
            follow_redirects=True,
            http2=False,  # IPTV servers are pretty old, http2 is risky
        )
    return _HTTP_CLIENT


# ── DB hookup (the server.py wires `db` here at startup) ──────────────
_db = None


def set_db(db) -> None:
    global _db
    _db = db


async def _get_settings() -> Dict[str, Any]:
    if _db is None:
        raise HTTPException(503, "Proxy not configured (db not set)")
    doc = await _db.settings.find_one({"id": "global"}, {"_id": 0})
    if not doc:
        return {}
    return doc


async def _resolve_shared_profile(profile_id: str) -> Dict[str, Any]:
    """Look up a SHARED Profile by its public id. Raises 403 if not found
    or if the id doesn't belong to the admin's Shared Profile list."""
    if not profile_id or not profile_id.startswith("shared:"):
        raise HTTPException(HTTP_403_FORBIDDEN, "proxy is restricted to shared profiles")
    settings = await _get_settings()
    shared = settings.get("shared_profiles") or []
    bare_id = profile_id.split(":", 1)[1]
    for sp in shared:
        # Try a few common id shapes (str id, dict with id, dict with _id)
        sp_id = str(sp.get("id") or sp.get("_id") or "")
        if not sp_id:
            continue
        if sp_id == bare_id or f"shared:{sp_id}" == profile_id:
            return sp
    raise HTTPException(HTTP_403_FORBIDDEN, "shared profile not found")


def _xtream_base(server: str) -> str:
    return (server or "").rstrip("/")


def _build_upstream_live(sp: Dict[str, Any], stream_id: str) -> str:
    base = _xtream_base(sp.get("server", ""))
    user = sp.get("username", "")
    pwd = sp.get("password", "")
    return f"{base}/live/{quote(user, safe='')}/{quote(pwd, safe='')}/{quote(stream_id, safe='')}.m3u8"


def _build_upstream_vod(
    sp: Dict[str, Any], stype: str, stream_id: str, ext: str
) -> str:
    base = _xtream_base(sp.get("server", ""))
    user = sp.get("username", "")
    pwd = sp.get("password", "")
    # Xtream URL shapes:
    #   movies: {base}/movie/{user}/{pwd}/{id}.{ext}
    #   series: {base}/series/{user}/{pwd}/{id}.{ext}
    bucket = "movie" if stype == "movie" else "series"
    return (
        f"{base}/{bucket}/{quote(user, safe='')}/{quote(pwd, safe='')}/"
        f"{quote(stream_id, safe='')}.{quote(ext, safe='')}"
    )


# ─────────────────────────────────────────────────────────────────────
# 1) Sign endpoint — clients call this to get a short-lived proxy URL
# ─────────────────────────────────────────────────────────────────────
class SignRequest(BaseModel):
    profile_id: str = Field(..., description="MUST start with 'shared:'")
    kind: str = Field(..., description="'live' or 'vod'")
    stream_id: str
    # vod-only:
    stream_type: Optional[str] = Field(None, description="'movie' or 'series'")
    ext: Optional[str] = Field("mp4")


class SignResponse(BaseModel):
    proxy_url: str
    expires_in: int


# ─────────────────────────────────────────────────────────────────────
# Module-level sign-request counter — exposed via /api/proxy/stats so
# the admin dashboard can answer "is any device hitting the sign
# endpoint at all?". Stays at 0 ⇒ the installed APK predates the
# proxy substitution code (rebuild needed).
# ─────────────────────────────────────────────────────────────────────
_SIGN_REQUEST_COUNT = 0

# Most-recent upstream error stash — surfaced in /api/proxy/stats so
# the admin widget can show "why the last playback failed" without
# grepping FastAPI logs.
_LAST_UPSTREAM_ERROR: Optional[Dict[str, Any]] = None


def _record_upstream_error(where: str, url: str, status: int, message: str) -> None:
    """Stash the most-recent upstream failure for /api/proxy/stats."""
    global _LAST_UPSTREAM_ERROR
    _LAST_UPSTREAM_ERROR = {
        "where": where,
        "url": url,
        "http": int(status),
        "msg": str(message)[:512],
        "ts": time.strftime("%Y-%m-%d %H:%M:%S"),
    }
    logging.warning("[proxy:%s] http=%s url=%s err=%s", where, status, url, message)


@router.post("/sign", response_model=SignResponse)
async def sign(req: SignRequest, _user: str = Depends(require_user)):
    """Return a signed proxy URL for the given Shared Profile stream.

    Refuses to sign for non-shared profiles — the structural guarantee
    that the proxy ONLY restreams admin-managed catalogues.
    """
    global _SIGN_REQUEST_COUNT
    _SIGN_REQUEST_COUNT += 1
    sp = await _resolve_shared_profile(req.profile_id)
    if req.kind == "live":
        upstream_base = _build_upstream_live(sp, req.stream_id).rsplit("/", 1)[0]
        token = sign_token(
            {
                "v": 1,
                "k": "live",
                "pid": req.profile_id,
                "sid": str(req.stream_id),
                "ub": upstream_base,
            }
        )
        return SignResponse(
            proxy_url=f"/api/proxy/live/{token}.m3u8", expires_in=DEFAULT_TTL_SECONDS
        )
    if req.kind == "vod":
        if req.stream_type not in {"movie", "series"}:
            raise HTTPException(400, "stream_type must be 'movie' or 'series'")
        ext = (req.ext or "mp4").strip(".") or "mp4"
        token = sign_token(
            {
                "v": 1,
                "k": "vod",
                "pid": req.profile_id,
                "sid": str(req.stream_id),
                "st": req.stream_type,
                "ext": ext,
            }
        )
        return SignResponse(
            proxy_url=f"/api/proxy/vod/{token}.{ext}", expires_in=DEFAULT_TTL_SECONDS
        )
    raise HTTPException(400, "kind must be 'live' or 'vod'")


# ─────────────────────────────────────────────────────────────────────
# 2) Live HLS — playlist + segments
# ─────────────────────────────────────────────────────────────────────
_M3U8_LINE_RE = re.compile(r"^[^#\s].*$", re.MULTILINE)


def _rewrite_m3u8(text: str, upstream_base: str, master_token: str) -> str:
    """Rewrite every URL line of an .m3u8 playlist to flow through us.

    Each URL line (i.e. not starting with `#` or whitespace) is replaced
    with a `/api/proxy/seg/{seg_token}` URL. The seg_token is a fresh
    HMAC-signed token whose payload contains the absolute upstream URL —
    no server-side state needed.
    """
    def repl(match: "re.Match[str]") -> str:
        url = match.group(0).strip()
        if not url:
            return url
        # Resolve relative URLs against upstream base.
        absolute = url if "://" in url else urljoin(upstream_base + "/", url)
        seg_token = sign_token(
            {"v": 1, "k": "seg", "u": absolute, "pid_token": master_token[:24]}
        )
        return f"/api/proxy/seg/{seg_token}"

    return _M3U8_LINE_RE.sub(repl, text)


@router.get("/live/{token}.m3u8")
async def live_playlist(token: str, request: Request):
    payload = verify_token(token)
    if not payload or payload.get("k") != "live":
        raise HTTPException(HTTP_403_FORBIDDEN, "invalid or expired token")

    sp = await _resolve_shared_profile(str(payload.get("pid", "")))
    upstream = _build_upstream_live(sp, str(payload.get("sid", "")))
    upstream_base = upstream.rsplit("/", 1)[0]

    # The upstream playlist refreshes every few seconds. We DON'T cache
    # it server-side beyond the duration of this request — the segment
    # cache is what matters for resource savings, and stale playlists
    # cause player stalls.
    try:
        client = _get_client()
        # Forward Range header (some Xtream variants require it for HLS)
        headers = {"User-Agent": "Nadiplayer/1.3 (+proxy)"}
        r = await client.get(upstream, headers=headers, timeout=15.0)
        r.raise_for_status()
    except httpx.HTTPStatusError as exc:
        _record_upstream_error(
            "live-playlist", upstream, exc.response.status_code,
            f"{type(exc).__name__}: {exc}",
        )
        raise HTTPException(502, f"upstream playlist HTTP {exc.response.status_code}")
    except httpx.HTTPError as exc:
        _record_upstream_error(
            "live-playlist", upstream, 0,
            f"{type(exc).__name__}: {exc}",
        )
        raise HTTPException(502, f"upstream playlist unavailable: {type(exc).__name__}")

    body = r.text
    if "#EXTM3U" not in body[:64]:
        # Not actually a playlist (some upstreams 302 → MPEG-TS). Pass
        # the raw bytes through with a hint for the player.
        return Response(
            content=r.content,
            media_type=r.headers.get("content-type", "application/octet-stream"),
            headers={"x-proxy-kind": "passthrough-not-m3u8"},
        )

    rewritten = _rewrite_m3u8(body, upstream_base, master_token=token)
    return PlainTextResponse(
        rewritten,
        media_type="application/vnd.apple.mpegurl",
        headers={
            "Cache-Control": "no-store",
            "x-proxy-kind": "m3u8",
        },
    )


@router.get("/seg/{token}")
async def serve_segment(token: str):
    payload = verify_token(token)
    if not payload or payload.get("k") != "seg":
        raise HTTPException(HTTP_403_FORBIDDEN, "invalid or expired segment token")

    upstream = str(payload.get("u", ""))
    if not upstream.startswith(("http://", "https://")):
        raise HTTPException(HTTP_403_FORBIDDEN, "bad segment url")

    # Single-fetch: collisions on the same upstream URL share one fetch.
    try:
        body, ct = await SEGMENT_HUB.get_or_fetch(
            key=upstream,
            upstream_url=upstream,
            client=_get_client(),
            request_timeout=20.0,
        )
    except httpx.HTTPStatusError as exc:
        _record_upstream_error(
            "segment", upstream, exc.response.status_code,
            f"{type(exc).__name__}: {exc}",
        )
        # Pass-through 404 / 5xx so the player can decide whether to
        # retry or give up.
        return Response(
            content=b"", status_code=exc.response.status_code,
            headers={"x-proxy-kind": "seg-error"},
        )
    except httpx.HTTPError as exc:
        _record_upstream_error(
            "segment", upstream, 0,
            f"{type(exc).__name__}: {exc}",
        )
        raise HTTPException(502, f"segment unavailable: {type(exc).__name__}")
    return Response(
        content=body,
        media_type=ct,
        headers={
            "Cache-Control": "public, max-age=30",
            "x-proxy-kind": "seg",
        },
    )


# ─────────────────────────────────────────────────────────────────────
# 3) VOD pass-through — movies + series episodes
# ─────────────────────────────────────────────────────────────────────
@router.get("/vod/{filename}")
async def vod_passthrough(filename: str, request: Request):
    """Range-aware pass-through for movies / series episodes.

    `filename` is `{token}.{ext}` — we split off the last dot. We do NOT
    apply single-fetch here because each viewer has their own seek
    position and the dedup hit-rate is essentially zero.
    """
    if "." not in filename:
        raise HTTPException(400, "bad vod filename")
    token, _ext = filename.rsplit(".", 1)
    payload = verify_token(token)
    if not payload or payload.get("k") != "vod":
        raise HTTPException(HTTP_403_FORBIDDEN, "invalid or expired token")

    sp = await _resolve_shared_profile(str(payload.get("pid", "")))
    upstream = _build_upstream_vod(
        sp,
        stype=str(payload.get("st", "movie")),
        stream_id=str(payload.get("sid", "")),
        ext=str(payload.get("ext", "mp4")),
    )

    # Forward the client's Range header so the player can seek.
    headers = {"User-Agent": "Nadiplayer/1.3 (+proxy)"}
    rng = request.headers.get("range")
    if rng:
        headers["Range"] = rng

    client = _get_client()
    try:
        req = client.build_request("GET", upstream, headers=headers)
        upstream_resp = await client.send(req, stream=True)
    except httpx.HTTPError as exc:
        logger.warning("[proxy] vod open failed: %s", exc)
        raise HTTPException(502, "vod unavailable")

    # Filter the headers we relay back. Hop-by-hop headers and gzip
    # encoding (we're streaming raw video) must be stripped.
    safe_keys = {
        "content-type", "content-length", "content-range",
        "accept-ranges", "etag", "last-modified",
    }
    out_headers = {
        k: v for k, v in upstream_resp.headers.items() if k.lower() in safe_keys
    }
    out_headers["x-proxy-kind"] = "vod-passthrough"
    out_headers["Cache-Control"] = "public, max-age=300"

    async def streamer():
        try:
            async for chunk in upstream_resp.aiter_raw(64 * 1024):
                yield chunk
        finally:
            await upstream_resp.aclose()

    return StreamingResponse(
        streamer(),
        status_code=upstream_resp.status_code,
        media_type=out_headers.get("content-type", "video/mp4"),
        headers=out_headers,
    )


# ─────────────────────────────────────────────────────────────────────
# 4) Tiny health / stats endpoint (admin only)
# ─────────────────────────────────────────────────────────────────────
@router.get("/stats")
async def stats(_admin: str = Depends(require_admin)):
    """Expose hub stats so the admin dashboard can show RAM usage."""
    hits = SEGMENT_HUB.hits
    misses = SEGMENT_HUB.misses
    total = hits + misses
    return {
        "backend": "ram",  # Python hub is always in-process RAM
        "cached_segments": SEGMENT_HUB.cached_count,
        "buffer_bytes": SEGMENT_HUB.total_bytes,
        "buffer_max_bytes": SEGMENT_HUB.max_bytes,
        "buffer_pct": (
            round(100.0 * SEGMENT_HUB.total_bytes / SEGMENT_HUB.max_bytes, 2)
            if SEGMENT_HUB.max_bytes > 0 else 0
        ),
        "ttl_seconds": SEGMENT_HUB.ttl_seconds,
        "hits": hits,
        "misses": misses,
        "hit_rate_pct": round(100.0 * hits / total, 2) if total > 0 else 0,
        "active_streams": SEGMENT_HUB.inflight,
        # Mirrors the PHP backend's "sign_requests" counter.
        "sign_requests": _SIGN_REQUEST_COUNT,
        "last_upstream_error": _LAST_UPSTREAM_ERROR,
    }
