"""Append-only, fail-open production telemetry.

This module is deliberately *not* part of the media/control path:

* Public calls only update in-memory counters or put a small record on a
  bounded queue.  They never write to disk, never wait for I/O, and never raise
  into a relay/manager thread.
* A single daemon writer appends compact JSON lines.  If storage is slow or
  unavailable, telemetry records are dropped and counted; the stream is not.
* There is intentionally no rotate/truncate/delete API.  Records live under the
  persistent data root (``<DATA_ROOT>/telemetry/telemetry.jsonl``) and are
  removed only by an explicit, manual operator action outside this code.
* No media, signed URLs, stream keys, cookies, tokens, or credentials are
  recorded.  Potentially sensitive keys/values are stripped defensively.

Set ``TELEMETRY_ENABLED=0`` to disable collection without touching streaming.
"""

import json
import math
import os
import queue
import re
import threading
import time
from datetime import datetime, timezone
from pathlib import Path


_QUEUE_MAX = 4096
_SENSITIVE_PARTS = (
    "url", "token", "cookie", "authorization", "sessionid", "password",
    "secret", "stream_key", "rtmp_key", "access_key", "refresh_token",
)
_URL_RE = re.compile(r"(?i)\b(?:rtmps?|https?)://\S+")
_SECRET_ASSIGNMENT_RE = re.compile(
    r"(?i)\b(sessionid|token|stream[_ -]?key|authorization|sign|signature)\s*[=:]\s*[^\s;&]+"
)


def _iso_now():
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def _enabled():
    return str(os.getenv("TELEMETRY_ENABLED", "1")).strip().lower() not in (
        "0", "false", "off", "no", ""
    )


def _default_path():
    explicit = str(os.getenv("TELEMETRY_PATH", "") or "").strip()
    if explicit:
        return Path(explicit).expanduser()

    events_path = str(os.getenv("STREAM_EVENTS_PATH", "") or "").strip()
    if events_path:
        events = Path(events_path).expanduser()
        root = events.parent.parent if events.parent.name == "logs" else events.parent
        return root / "telemetry" / "telemetry.jsonl"

    data_dir = str(os.getenv("DATA_DIR", "") or "").strip()
    if data_dir:
        return Path(data_dir).expanduser() / "telemetry" / "telemetry.jsonl"
    if Path("/data").is_dir():
        return Path("/data/tiktok-relay/telemetry/telemetry.jsonl")
    return Path("./storage/telemetry/telemetry.jsonl")


def _safe(value, depth=0):
    if depth > 5:
        return "…"
    if isinstance(value, dict):
        out = {}
        for key, item in list(value.items())[:80]:
            name = str(key)[:80]
            lowered = name.lower()
            if any(part in lowered for part in _SENSITIVE_PARTS):
                continue
            out[name] = _safe(item, depth + 1)
        return out
    if isinstance(value, (list, tuple)):
        return [_safe(item, depth + 1) for item in value[:30]]
    if isinstance(value, str):
        value = _URL_RE.sub("«hidden-url»", value)
        value = _SECRET_ASSIGNMENT_RE.sub(r"\1=«hidden»", value)
        return value[:300]
    if isinstance(value, bool) or value is None:
        return value
    if isinstance(value, (int, float)):
        return value if math.isfinite(float(value)) else None
    return _safe(str(value), depth + 1)


def http_status_from_exception(error):
    """Best-effort HTTP status from curl/httpx/requests-style exceptions."""
    response = getattr(error, "response", None)
    for candidate in (
        getattr(response, "status_code", None),
        getattr(error, "status_code", None),
        getattr(error, "status", None),
    ):
        try:
            status = int(candidate)
        except (TypeError, ValueError):
            continue
        if 100 <= status < 600:
            return status
    # curl_cffi sometimes exposes only a rendered message.  Accept an explicit
    # HTTP-looking code rather than persisting the exception text or any URL.
    match = re.search(r"\bHTTP(?:/[\d.]+)?\s+([1-5]\d\d)\b", str(error or ""), re.I)
    if match:
        try:
            return int(match.group(1))
        except ValueError:
            return 0
    return 0


class _CpuSampler:
    """Cached cgroup CPU sampler; reads tiny proc files at most every 10s."""

    def __init__(self):
        self._lock = threading.Lock()
        self._last_at = 0.0
        self._last_usage_seconds = None
        self._cached = {}

    @staticmethod
    def _read_text(path):
        try:
            return Path(path).read_text(encoding="utf-8", errors="ignore")
        except OSError:
            return ""

    def _usage_seconds(self):
        text = self._read_text("/sys/fs/cgroup/cpu.stat")
        for line in text.splitlines():
            if line.startswith("usage_usec"):
                try:
                    return int(line.split()[1]) / 1_000_000.0
                except (IndexError, ValueError):
                    pass
        for path in (
            "/sys/fs/cgroup/cpuacct/cpuacct.usage",
            "/sys/fs/cgroup/cpu,cpuacct/cpuacct.usage",
        ):
            text = self._read_text(path).strip()
            if text:
                try:
                    return int(text) / 1_000_000_000.0
                except ValueError:
                    pass
        return None

    def _quota_cpus(self):
        text = self._read_text("/sys/fs/cgroup/cpu.max").split()
        if len(text) >= 2 and text[0] != "max":
            try:
                return max(0.01, int(text[0]) / int(text[1]))
            except (ValueError, ZeroDivisionError):
                pass
        try:
            quota = int(self._read_text("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").strip())
            period = int(self._read_text("/sys/fs/cgroup/cpu/cpu.cfs_period_us").strip())
            if quota > 0 and period > 0:
                return quota / period
        except ValueError:
            pass
        return float(os.cpu_count() or 1)

    def fields(self):
        now = time.monotonic()
        with self._lock:
            if self._cached and now - self._last_at < 10.0:
                return dict(self._cached)
            out = {}
            try:
                out["load1"] = round(float(os.getloadavg()[0]), 3)
            except (AttributeError, OSError, ValueError):
                pass
            usage = self._usage_seconds()
            if usage is not None and self._last_usage_seconds is not None:
                elapsed = now - self._last_at
                if elapsed > 0:
                    cores = max(0.0, (usage - self._last_usage_seconds) / elapsed)
                    quota = self._quota_cpus()
                    out["cpu_cores_used"] = round(cores, 4)
                    out["cpu_pct_of_quota"] = round(min(999.0, 100.0 * cores / quota), 2) if quota else None
            if usage is not None:
                self._last_usage_seconds = usage
            # Only advance the timestamp when we actually sampled usage; this
            # also makes the first real interval available about 10s later.
            self._last_at = now
            self._cached = dict(out)
            return out


class Telemetry:
    """In-memory session aggregation plus a non-blocking append-only writer."""

    def __init__(self, path=None):
        self.path = Path(path).expanduser() if path else _default_path()
        self.enabled = _enabled()
        self._queue = queue.Queue(maxsize=_QUEUE_MAX) if self.enabled else None
        self._lock = threading.RLock()
        self._writer_lock = threading.Lock()
        self._writer = None
        self._sessions = {}
        self._sequence = 0
        self.dropped_records = 0
        self.write_errors = 0
        self._cpu = _CpuSampler()

    # ---------- writer ----------
    def _ensure_writer(self):
        if not self.enabled:
            return False
        writer = self._writer
        if writer is not None and writer.is_alive():
            return True
        with self._writer_lock:
            writer = self._writer
            if writer is not None and writer.is_alive():
                return True
            self._writer = threading.Thread(
                target=self._writer_loop, name="telemetry-writer", daemon=True
            )
            self._writer.start()
            return True

    def _writer_loop(self):
        while True:
            record = self._queue.get()
            try:
                line = json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n"
                self.path.parent.mkdir(parents=True, exist_ok=True)
                with self.path.open("a", encoding="utf-8") as handle:
                    handle.write(line)
                    handle.flush()
                    if record.get("type") in ("session_finish", "session_start"):
                        try:
                            os.fsync(handle.fileno())
                        except OSError:
                            pass
                try:
                    os.chmod(self.path, 0o600)
                except OSError:
                    pass
            except Exception:
                with self._lock:
                    self.write_errors += 1
            finally:
                self._queue.task_done()

    def _enqueue(self, record):
        if not self.enabled:
            return False
        try:
            if not self._ensure_writer():
                return False
            self._queue.put_nowait(_safe(record))
            return True
        except queue.Full:
            with self._lock:
                self.dropped_records += 1
            return False
        except Exception:
            return False

    # ---------- aggregation ----------
    @staticmethod
    def _new_stat(value):
        return {"count": 1, "sum": float(value), "min": float(value), "max": float(value), "last": float(value)}

    def _update_session_locked(self, session, fields):
        now = time.monotonic()
        stats = session["stats"]
        last = session["last"]
        for name, value in (fields or {}).items():
            name = str(name)[:80]
            if isinstance(value, bool) or value is None:
                if value is not None:
                    last[name] = value
                continue
            if isinstance(value, (int, float)) and math.isfinite(float(value)):
                number = float(value)
                stat = stats.get(name)
                if stat is None:
                    stats[name] = self._new_stat(number)
                else:
                    stat["count"] += 1
                    stat["sum"] += number
                    stat["min"] = min(stat["min"], number)
                    stat["max"] = max(stat["max"], number)
                    stat["last"] = number
                last[name] = number
                if name == "ffmpeg_speed":
                    if number < 0.95:
                        session["counters"]["ffmpeg_speed_below_0_95_samples"] = (
                            session["counters"].get("ffmpeg_speed_below_0_95_samples", 0) + 1
                        )
                    if number < 1.0:
                        session["counters"]["ffmpeg_speed_below_1_samples"] = (
                            session["counters"].get("ffmpeg_speed_below_1_samples", 0) + 1
                        )
            elif isinstance(value, (str, int, float)):
                last[name] = _safe(value)

        minute = int(max(0.0, now - session["started_monotonic"]) // 60)
        bucket = session["minutes"].setdefault(
            str(minute),
            {"samples": 0, "ffmpeg_speed_sum": 0.0, "ffmpeg_speed_samples": 0,
             "ffmpeg_speed_below_0_95": 0, "source_bps_sum": 0.0, "source_bps_samples": 0},
        )
        bucket["samples"] += 1
        if isinstance((fields or {}).get("ffmpeg_speed"), (int, float)):
            speed = float(fields["ffmpeg_speed"])
            bucket["ffmpeg_speed_sum"] += speed
            bucket["ffmpeg_speed_samples"] += 1
            if speed < 0.95:
                bucket["ffmpeg_speed_below_0_95"] += 1
        if isinstance((fields or {}).get("source_bps"), (int, float)):
            bucket["source_bps_sum"] += float(fields["source_bps"])
            bucket["source_bps_samples"] += 1
        session["last_observed_at"] = _iso_now()
        session["last_observed_monotonic"] = now

    def _summary_locked(self, session, *, finished=False, reason="", extra=None, include_minutes=True):
        now = time.monotonic()
        stats = {}
        for name, stat in session["stats"].items():
            count = int(stat["count"])
            stats[name] = {
                "count": count,
                "avg": round(stat["sum"] / count, 4) if count else None,
                "min": round(stat["min"], 4),
                "max": round(stat["max"], 4),
                "last": round(stat["last"], 4),
            }
        minutes = []
        if include_minutes:
            for minute, bucket in sorted(session["minutes"].items(), key=lambda item: int(item[0])):
                speed_samples = int(bucket["ffmpeg_speed_samples"])
                source_samples = int(bucket["source_bps_samples"])
                minutes.append({
                    "minute": int(minute),
                    "samples": int(bucket["samples"]),
                    "ffmpeg_speed_avg": round(bucket["ffmpeg_speed_sum"] / speed_samples, 4) if speed_samples else None,
                    "ffmpeg_speed_below_0_95": int(bucket["ffmpeg_speed_below_0_95"]),
                    "source_bps_avg": round(bucket["source_bps_sum"] / source_samples, 1) if source_samples else None,
                })
        return _safe({
            "session_key": session["key"],
            "broadcaster": session["broadcaster"],
            "started_at": session["started_at"],
            "ended_at": _iso_now() if finished else None,
            "duration_seconds": round(max(0.0, now - session["started_monotonic"]), 3),
            "meta": session["meta"],
            "counters": dict(session["counters"]),
            "stats": stats,
            "last": dict(session["last"]),
            "minutes": minutes,
            "minutes_omitted": not include_minutes,
            "reason": str(reason or "")[:200],
            "extra": extra or {},
            "host": self._cpu.fields(),
            "telemetry_dropped_records": self.dropped_records,
            "telemetry_write_errors": self.write_errors,
        })

    # ---------- public fail-open API ----------
    def start_session(self, broadcaster, meta=None):
        if not self.enabled:
            return None
        try:
            user = str(broadcaster or "system")[:100]
            with self._lock:
                self._sequence += 1
                key = "%s:%d:%d" % (user, int(time.time() * 1000), self._sequence)
                session = {
                    "key": key,
                    "broadcaster": user,
                    "started_at": _iso_now(),
                    "started_monotonic": time.monotonic(),
                    "meta": _safe(meta or {}),
                    "counters": {},
                    "stats": {},
                    "last": {},
                    "minutes": {},
                    "dedupe": {},
                    "last_observed_at": None,
                    "last_observed_monotonic": time.monotonic(),
                }
                self._sessions[key] = session
                summary = self._summary_locked(session)
            self._enqueue({
                "type": "session_start", "recorded_at": _iso_now(),
                "session_key": key, "broadcaster": user, "summary": summary,
            })
            return key
        except Exception:
            return None

    def observe(self, session_key, fields=None):
        if not self.enabled or not session_key:
            return False
        try:
            with self._lock:
                session = self._sessions.get(session_key)
                if session is None:
                    return False
                self._update_session_locked(session, fields or {})
            return True
        except Exception:
            return False

    def event(self, session_key, broadcaster, event, details=None, dedupe_key=None):
        if not self.enabled:
            return False
        try:
            name = str(event or "UNKNOWN")[:80]
            safe_details = _safe(details or {})
            elapsed = None
            with self._lock:
                session = self._sessions.get(session_key) if session_key else None
                if session is not None:
                    counters = session["counters"]
                    counters[name] = counters.get(name, 0) + 1
                    if name == "SOURCE_OPEN_ERROR":
                        status = (safe_details or {}).get("http_status")
                        if status:
                            label = "source_open_http_%s" % status
                            counters[label] = counters.get(label, 0) + 1
                    elapsed = round(max(0.0, time.monotonic() - session["started_monotonic"]), 3)
                    if dedupe_key:
                        previous = session["dedupe"].get(str(dedupe_key))
                        if previous == safe_details:
                            label = name + "_suppressed"
                            counters[label] = counters.get(label, 0) + 1
                            return True
                        session["dedupe"][str(dedupe_key)] = safe_details
            return self._enqueue({
                "type": "event", "recorded_at": _iso_now(), "session_key": session_key,
                "broadcaster": str(broadcaster or "system")[:100], "event": name,
                "elapsed_seconds": elapsed, "details": safe_details,
            })
        except Exception:
            return False

    def checkpoint(self, session_key, reason="periodic"):
        if not self.enabled or not session_key:
            return False
        try:
            with self._lock:
                session = self._sessions.get(session_key)
                if session is None:
                    return False
                summary = self._summary_locked(session, reason=reason, include_minutes=False)
                user = session["broadcaster"]
            return self._enqueue({
                "type": "checkpoint", "recorded_at": _iso_now(), "session_key": session_key,
                "broadcaster": user, "summary": summary,
            })
        except Exception:
            return False

    def finish(self, session_key, reason="", extra=None):
        if not self.enabled or not session_key:
            return False
        try:
            with self._lock:
                session = self._sessions.pop(session_key, None)
                if session is None:
                    return False
                summary = self._summary_locked(
                    session, finished=True, reason=reason, extra=_safe(extra or {})
                )
                user = session["broadcaster"]
            return self._enqueue({
                "type": "session_finish", "recorded_at": _iso_now(), "session_key": session_key,
                "broadcaster": user, "summary": summary,
            })
        except Exception:
            return False

    def read_records(self, limit=200, broadcaster=None):
        """Read recent records for the authenticated diagnostics API.

        This is an operator/read path only; it is never called by media code.
        It reads a bounded tail so a long-lived append-only file cannot make
        the request consume unbounded memory.
        """
        try:
            limit = max(1, min(int(limit), 5000))
        except (TypeError, ValueError):
            limit = 200
        try:
            if not self.path.exists():
                return []
            size = self.path.stat().st_size
            max_bytes = 16 * 1024 * 1024
            offset = max(0, size - max_bytes)
            with self.path.open("rb") as handle:
                handle.seek(offset)
                raw = handle.read()
            if offset:
                newline = raw.find(b"\n")
                raw = raw[newline + 1:] if newline >= 0 else b""
            records = []
            wanted = str(broadcaster or "").strip().lstrip("@")
            for line in raw.decode("utf-8", "ignore").splitlines():
                try:
                    item = json.loads(line)
                except ValueError:
                    continue
                if not isinstance(item, dict):
                    continue
                if wanted:
                    summary = item.get("summary") or {}
                    if item.get("broadcaster") != wanted and summary.get("broadcaster") != wanted:
                        continue
                records.append(item)
            return records[-limit:]
        except OSError:
            return []

    def status(self):
        with self._lock:
            return {
                "enabled": bool(self.enabled),
                "file": "telemetry/telemetry.jsonl",
                "active_sessions": len(self._sessions),
                "dropped_records": self.dropped_records,
                "write_errors": self.write_errors,
                "queued_records": self._queue.qsize() if self._queue is not None else 0,
            }


telemetry = Telemetry()


def start_session(broadcaster, meta=None):
    try:
        return telemetry.start_session(broadcaster, meta)
    except Exception:
        return None


def observe(session_key, fields=None):
    try:
        return telemetry.observe(session_key, fields)
    except Exception:
        return False


def record_event(session_key, broadcaster, event, details=None, dedupe_key=None):
    try:
        return telemetry.event(session_key, broadcaster, event, details, dedupe_key=dedupe_key)
    except Exception:
        return False


def checkpoint(session_key, reason="periodic"):
    try:
        return telemetry.checkpoint(session_key, reason)
    except Exception:
        return False


def finish(session_key, reason="", extra=None):
    try:
        return telemetry.finish(session_key, reason, extra)
    except Exception:
        return False


def read_records(limit=200, broadcaster=None):
    try:
        return telemetry.read_records(limit, broadcaster)
    except Exception:
        return []


def status():
    try:
        return telemetry.status()
    except Exception:
        return {"enabled": False, "file": "telemetry/telemetry.jsonl", "active_sessions": 0}