add analytics emitter

analytics/__init__.py

@@ -0,0 +1,17 @@
+"""Analytics event emission. Optional and best-effort: it must never affect playback."""
+from .emitter import (
+    AnalyticsEmitter,
+    configure_from_env,
+    get_emitter,
+    shutdown_emitter,
+)
+from .integration import annotate, tracked
+
+__all__ = [
+    "AnalyticsEmitter",
+    "configure_from_env",
+    "get_emitter",
+    "shutdown_emitter",
+    "annotate",
+    "tracked",
+]

analytics/emitter.py

@@ -0,0 +1,292 @@
+"""
+Asynchronous, best-effort event emitter.
+
+Design goal: the request thread pays for ONE deque.append and nothing else.
+
+  - emit() is a length check + deque.append. No lock, no condition-variable
+    notify, no I/O, no logging, no allocation beyond the dict the caller built.
+    It cannot block and does not wake the consumer thread.
+  - The consumer thread polls every `flush_interval` seconds (Event.wait), so
+    there is no per-event cross-thread signalling. Anonymisation, JSON encoding
+    and HTTP all happen there.
+  - Overflow drops the NEWEST event and counts it; the consumer logs the drop
+    count (rate-limited). Nothing is ever logged from a request thread.
+  - A failing sink drops the batch (no retry queue) and backs off exponentially
+    up to `max_backoff`, so a dead endpoint is not hammered.
+  - Counters are intentionally lock-free. `dropped` is incremented from request
+    threads and may under-count under heavy contention; it is a health signal,
+    not an audit log.
+"""
+from __future__ import annotations
+
+import atexit
+import collections
+import os
+import threading
+import time
+from typing import Any, Callable, Deque, Dict, List, Mapping, Optional
+
+import requests
+
+from streaming_providers.base.utils import logger
+from .events import finalize
+from .privacy import make_anonymizer
+
+Sink = Callable[[List[Dict[str, Any]]], None]
+Transform = Callable[[Dict[str, Any]], Dict[str, Any]]
+
+
+class HttpSink:
+    """Batched JSON POST: {"events": [...]}. Raises on non-2xx."""
+
+    def __init__(self, url: str, *, timeout: float = 5.0, auth_token: Optional[str] = None):
+        self._url = url
+        self._timeout = (min(3.05, timeout), timeout)  # (connect, read)
+        self._session = requests.Session()  # only ever used from the consumer thread
+        self._session.headers["User-Agent"] = "ultimate-backend-analytics/1"
+        if auth_token:
+            self._session.headers["Authorization"] = f"Bearer {auth_token}"
+
+    def __call__(self, batch: List[Dict[str, Any]]) -> None:
+        resp = self._session.post(self._url, json={"events": batch}, timeout=self._timeout)
+        resp.raise_for_status()
+
+    def close(self) -> None:
+        self._session.close()
+
+
+class AnalyticsEmitter:
+    def __init__(
+        self,
+        sink: Sink,
+        *,
+        transform: Optional[Transform] = None,
+        max_queue: int = 10_000,
+        batch_size: int = 100,
+        flush_interval: float = 1.0,
+        max_backoff: float = 30.0,
+        warn_interval: float = 60.0,
+        report_interval: float = 300.0,
+    ):
+        self._sink = sink
+        self._transform = transform or finalize
+        self._max_queue = max_queue
+        self._batch_size = batch_size
+        self._flush_interval = flush_interval
+        self._max_backoff = max_backoff
+        self._warn_interval = warn_interval
+        self._report_interval = report_interval
+
+        self._q: Deque[Dict[str, Any]] = collections.deque()
+        self._dropped = 0          # queue overflow (approximate, see module docstring)
+        self._dropped_logged = 0
+        self._shipped = 0
+        self._failed_batches = 0
+        self._failed_events = 0    # lost to sink failure or a malformed event
+
+        self._last_warn: Dict[str, float] = {}
+        self._last_report = time.monotonic()
+        self._report_snapshot = (0, 0, 0)
+
+        self._drain_deadline = 0.0
+        self._stop = threading.Event()
+        self._thread = threading.Thread(target=self._run, name="analytics-emitter", daemon=True)
+        self._thread.start()
+
+    # -- public API (request threads) ------------------------------------
+
+    def emit(self, raw: Dict[str, Any]) -> None:
+        """Non-blocking, lock-free, never raises. Drops the event if the queue is full."""
+        if len(self._q) >= self._max_queue:
+            self._dropped += 1
+            return
+        self._q.append(raw)
+
+    # -- public API (anywhere) --------------------------------------------
+
+    def stats(self) -> Dict[str, int]:
+        return {
+            "queue_depth": len(self._q),
+            "dropped": self._dropped,
+            "shipped": self._shipped,
+            "failed_batches": self._failed_batches,
+            "failed_events": self._failed_events,
+        }
+
+    def shutdown(self, timeout: float = 5.0) -> None:
+        """Flush what is queued (bounded by `timeout`), then stop. Idempotent."""
+        if self._stop.is_set():
+            return
+        self._drain_deadline = time.monotonic() + timeout
+        self._stop.set()
+        self._thread.join(timeout=timeout + 1.0)
+        close = getattr(self._sink, "close", None)
+        if callable(close):
+            try:
+                close()
+            except Exception:  # noqa: BLE001
+                pass
+
+    # -- consumer thread ---------------------------------------------------
+
+    def _warn(self, key: str, message: str) -> None:
+        now = time.monotonic()
+        if now - self._last_warn.get(key, -1e9) >= self._warn_interval:
+            self._last_warn[key] = now
+            logger.warning(message)
+
+    def _take(self) -> List[Dict[str, Any]]:
+        batch: List[Dict[str, Any]] = []
+        pop = self._q.popleft
+        while len(batch) < self._batch_size:
+            try:
+                batch.append(pop())
+            except IndexError:
+                break
+        return batch
+
+    def _ship(self, raw_batch: List[Dict[str, Any]]) -> bool:
+        wire: List[Dict[str, Any]] = []
+        for raw in raw_batch:
+            try:
+                wire.append(self._transform(raw))
+            except Exception:  # noqa: BLE001 — one malformed event must not sink the batch
+                self._failed_events += 1
+        if not wire:
+            return True
+        try:
+            self._sink(wire)
+        except Exception as err:  # noqa: BLE001 — sink is pluggable
+            self._failed_batches += 1
+            self._failed_events += len(wire)
+            self._warn("ship", f"Analytics ship failed, dropped {len(wire)} events: {err}")
+            return False
+        self._shipped += len(wire)
+        return True
+
+    def _report(self) -> None:
+        dropped = self._dropped
+        if dropped > self._dropped_logged:
+            self._warn(
+                "drop",
+                f"Analytics queue full: {dropped - self._dropped_logged} events dropped "
+                f"(total {dropped}). Analytics is best-effort and does not affect playback.",
+            )
+            self._dropped_logged = dropped
+        now = time.monotonic()
+        if now - self._last_report >= self._report_interval:
+            self._last_report = now
+            snapshot = (self._shipped, self._dropped, self._failed_events)
+            if snapshot != self._report_snapshot:  # stay quiet when idle
+                self._report_snapshot = snapshot
+                logger.info(f"Analytics stats: {self.stats()}")
+
+    def _run(self) -> None:
+        backoff = 0.0
+        while not self._stop.is_set():
+            wait = self._flush_interval
+            batch = self._take()
+            if batch:
+                if self._ship(batch):
+                    backoff = 0.0
+                    if len(self._q) >= self._batch_size:
+                        wait = 0.0  # backlog: keep draining without sleeping
+                else:
+                    backoff = min(max(backoff * 2, 1.0), self._max_backoff)
+                    wait = backoff
+            self._report()
+            self._stop.wait(wait)
+        self._final_flush()
+
+    def _final_flush(self) -> None:
+        while self._q and time.monotonic() < self._drain_deadline:
+            if not self._ship(self._take()):
+                break
+        if self._q:
+            logger.warning(f"Analytics shutdown: {len(self._q)} queued events discarded")
+
+
+# -- process-wide singleton -------------------------------------------------
+
+_emitter: Optional[AnalyticsEmitter] = None
+_lock = threading.Lock()
+
+
+def _env_num(env: Mapping[str, str], name: str, default, cast, minimum):
+    raw = (env.get(name) or "").strip()
+    if not raw:
+        return default
+    try:
+        value = cast(raw)
+    except ValueError:
+        logger.warning(f"Ignoring invalid {name}={raw!r}; using {default}")
+        return default
+    return value if value >= minimum else default
+
+
+def configure_from_env(environ: Optional[Mapping[str, str]] = None) -> Optional[AnalyticsEmitter]:
+    """
+    Create the process-wide emitter from ANALYTICS_* variables. Idempotent.
+    Never raises. Returns None (analytics fully off, zero overhead) when
+    ANALYTICS_ENDPOINT_URL is unset or invalid.
+
+      ANALYTICS_ENDPOINT_URL      full URL batches are POSTed to (unset = disabled)
+      ANALYTICS_AUTH_TOKEN        optional bearer token
+      ANALYTICS_MAX_QUEUE         default 10000
+      ANALYTICS_BATCH_SIZE        default 100
+      ANALYTICS_FLUSH_INTERVAL    seconds, default 1.0
+      ANALYTICS_TIMEOUT           HTTP read timeout in seconds, default 5.0
+      ANALYTICS_ANONYMIZE         off | truncate | hash   (default off)
+      ANALYTICS_HASH_SALT         required for hash mode
+    """
+    global _emitter
+    env = os.environ if environ is None else environ
+    try:
+        with _lock:
+            if _emitter is not None:
+                return _emitter
+            url = (env.get("ANALYTICS_ENDPOINT_URL") or "").strip()
+            if not url:
+                logger.info("Analytics disabled (ANALYTICS_ENDPOINT_URL not set)")
+                return None
+            if not url.lower().startswith(("http://", "https://")):
+                logger.warning(f"Analytics disabled: ANALYTICS_ENDPOINT_URL must be http(s), got {url!r}")
+                return None
+
+            token = (env.get("ANALYTICS_AUTH_TOKEN") or "").strip() or None
+            if token and url.lower().startswith("http://"):
+                logger.warning("Analytics auth token is sent over plain HTTP; use HTTPS")
+            timeout = _env_num(env, "ANALYTICS_TIMEOUT", 5.0, float, 0.1)
+
+            anonymize = make_anonymizer(
+                env.get("ANALYTICS_ANONYMIZE", "off"), (env.get("ANALYTICS_HASH_SALT") or "")
+            )
+            _emitter = AnalyticsEmitter(
+                HttpSink(url, timeout=timeout, auth_token=token),
+                transform=lambda raw: finalize(anonymize(raw)),
+                max_queue=_env_num(env, "ANALYTICS_MAX_QUEUE", 10_000, int, 1),
+                batch_size=_env_num(env, "ANALYTICS_BATCH_SIZE", 100, int, 1),
+                flush_interval=_env_num(env, "ANALYTICS_FLUSH_INTERVAL", 1.0, float, 0.05),
+            )
+            atexit.register(shutdown_emitter)
+            logger.info(f"Analytics enabled -> {url.split('?')[0]}")
+            return _emitter
+    except Exception as err:  # noqa: BLE001 — analytics must never stop the service
+        logger.warning(f"Analytics could not be initialised: {err}")
+        return None
+
+
+def get_emitter() -> Optional[AnalyticsEmitter]:
+    return _emitter
+
+
+def shutdown_emitter(timeout: float = 5.0) -> None:
+    """Flush and stop the emitter. Safe to call repeatedly or when disabled."""
+    global _emitter
+    with _lock:
+        emitter, _emitter = _emitter, None
+    if emitter is not None:
+        try:
+            emitter.shutdown(timeout)
+        except Exception:  # noqa: BLE001
+            pass

analytics/events.py

@@ -0,0 +1,54 @@
+"""
+Analytics event schema. Pure data, no I/O.
+
+The request thread builds a plain dict of raw values (cheapest possible).
+Everything else (anonymisation, None-stripping, schema_version) happens on the
+consumer thread via finalize(), so none of it costs the request path anything.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict
+
+try:  # TypedDict is documentation only; keep import safe on old interpreters
+    from typing import TypedDict
+except ImportError:  # pragma: no cover
+    TypedDict = dict  # type: ignore
+
+SCHEMA_VERSION = 1
+
+# Event vocabulary: the backend only names what happened.
+EVENT_MANIFEST_REQUEST = "manifest_request"
+
+# How the request was served.
+DELIVERY_REDIRECT = "redirect"   # 302 to the upstream manifest; the client talks to upstream directly afterwards
+DELIVERY_PROXIED = "proxied"     # manifest body returned (rewritten/decrypted); the media proxy sees the segments
+DELIVERY_ERROR = "error"         # status >= 400 or an unhandled exception
+
+
+class AnalyticsEvent(TypedDict, total=False):
+    """Wire format (after finalize). Keys with None values are omitted."""
+    event_type: str
+    ts: float                    # Unix epoch seconds, taken on the request thread
+    client_ip: str               # raw, truncated or hashed depending on ANALYTICS_ANONYMIZE
+    provider: str
+    channel_id: str              # id of the content item (channel/event/vod/recording)
+    content_type: str            # "channel" | "event" | "vod" | "recording"
+    http_status: int
+    delivery: str                # DELIVERY_*
+    user_agent: str
+    country: str
+    is_catchup: bool
+    drm_variant: str
+    client_drm: bool
+    no_proxy: bool
+    highest_quality_only: bool
+    proxy: str                   # only if a route calls annotate(proxy=...); userinfo is always stripped
+    manifest_refresh_hint: int   # only if a route calls annotate(manifest_refresh_hint=...)
+    schema_version: int
+
+
+def finalize(raw: Dict[str, Any]) -> Dict[str, Any]:
+    """Drop None values (so consumers can key-presence-check) and stamp the schema version."""
+    out = {k: v for k, v in raw.items() if v is not None}
+    out["schema_version"] = SCHEMA_VERSION
+    return out

analytics/integration.py

@@ -0,0 +1,99 @@
+"""
+The only module route code touches.
+
+    @tracked("channel")          wrap a stream handler: one event per call,
+    def _handle_channel_stream(provider, channel_id): ...
+
+    annotate(is_catchup=..., proxy=...)   optional extra fields for the current request
+
+Cost on the request thread when analytics is enabled: a handful of attribute
+reads, one dict literal and one deque.append. When disabled: a single
+`get_emitter() is None` check. Everything is wrapped so that no analytics
+failure can change a response.
+"""
+from __future__ import annotations
+
+import functools
+import time
+from typing import Any, Callable
+
+from bottle import HTTPResponse, request, response
+
+from .emitter import get_emitter
+from .events import (
+    DELIVERY_ERROR,
+    DELIVERY_PROXIED,
+    DELIVERY_REDIRECT,
+    EVENT_MANIFEST_REQUEST,
+)
+
+_ENV_KEY = "analytics.extra"
+_UA_MAX = 256
+
+
+def annotate(**fields: Any) -> None:
+    """Attach extra fields to the event of the current request. Never raises."""
+    try:
+        request.environ.setdefault(_ENV_KEY, {}).update(fields)
+    except Exception:  # noqa: BLE001
+        pass
+
+
+def _classify(status: int, redirected: bool) -> str:
+    if status >= 400:
+        return DELIVERY_ERROR
+    return DELIVERY_REDIRECT if redirected else DELIVERY_PROXIED
+
+
+def _emit(emitter, content_type: str, provider: str, content_id: str,
+          status: int, delivery: str) -> None:
+    try:
+        ua = request.headers.get("User-Agent")
+        raw = {
+            "event_type": EVENT_MANIFEST_REQUEST,
+            "ts": time.time(),
+            "client_ip": request.remote_addr,
+            "provider": provider,
+            "channel_id": content_id,
+            "content_type": content_type,
+            "http_status": status,
+            "delivery": delivery,
+            "user_agent": ua[:_UA_MAX] if ua else None,
+            "country": request.query.get("country") or None,
+        }
+        extra = request.environ.get(_ENV_KEY)
+        if extra:
+            raw.update(extra)
+        emitter.emit(raw)
+    except Exception:  # noqa: BLE001
+        pass
+
+
+def tracked(content_type: str) -> Callable:
+    """
+    Decorator for `handler(provider, content_id, ...)`. Emits exactly one event
+    per call, including for redirects (which Bottle raises as HTTPResponse),
+    handler-set error statuses and unexpected exceptions. Never alters the
+    handler's return value, raised exception, or response.
+    """
+    def decorator(fn: Callable) -> Callable:
+        @functools.wraps(fn)
+        def wrapper(provider, content_id, *args, **kwargs):
+            emitter = get_emitter()
+            if emitter is None:
+                return fn(provider, content_id, *args, **kwargs)
+
+            status, delivery = 500, DELIVERY_ERROR
+            try:
+                result = fn(provider, content_id, *args, **kwargs)
+                status = response.status_code
+                delivery = _classify(status, redirected=False)
+                return result
+            except HTTPResponse as exc:
+                status = exc.status_code
+                delivery = _classify(status, redirected=300 <= status < 400)
+                raise
+            finally:
+                _emit(emitter, content_type, provider, content_id, status, delivery)
+        return wrapper
+    return decorator

analytics/privacy.py

@@ -0,0 +1,73 @@
+"""
+Optional anonymisation. Runs on the consumer thread only, never on the request path.
+
+ANALYTICS_ANONYMIZE:
+  off       client_ip is sent as-is (default)
+  truncate  IPv4 -> /24, IPv6 -> /48 (203.0.113.7 -> 203.0.113.0)
+  hash      salted SHA-256 (first 16 hex chars). Keeps "same client" clustering
+            without sending the address. Requires ANALYTICS_HASH_SALT; rotate the
+            salt on a schedule. Note: a salted hash of an IP is pseudonymous, not
+            anonymous, under GDPR.
+
+Proxy URLs always lose their userinfo (user:pass@). Unless mode is "off" the
+host is also replaced by a short hash, so events still group by proxy.
+"""
+from __future__ import annotations
+
+import hashlib
+import ipaddress
+from typing import Any, Callable, Dict, Optional
+from urllib.parse import urlsplit
+
+from streaming_providers.base.utils import logger
+
+Anonymizer = Callable[[Dict[str, Any]], Dict[str, Any]]
+
+_MODES = ("off", "truncate", "hash")
+
+
+def _hash(value: str, salt: str, length: int = 16) -> str:
+    return hashlib.sha256(f"{salt}|{value}".encode("utf-8")).hexdigest()[:length]
+
+
+def truncate_ip(ip: str) -> Optional[str]:
+    """IPv4 -> /24 network address, IPv6 -> /48. Returns None if unparsable."""
+    try:
+        addr = ipaddress.ip_address(ip.strip())
+    except ValueError:
+        return None
+    prefix = 24 if addr.version == 4 else 48
+    return str(ipaddress.ip_network(f"{addr}/{prefix}", strict=False).network_address)
+
+
+def sanitize_proxy(proxy: str, *, hide_host: bool, salt: str = "") -> Optional[str]:
+    """Strip credentials; optionally replace the host with a short hash."""
+    parts = urlsplit(proxy)
+    host = parts.netloc.rpartition("@")[2]
+    if not parts.scheme or not host:
+        return None
+    if hide_host:
+        return f"{parts.scheme}://{_hash(host, salt, 8)}"
+    return f"{parts.scheme}://{host}"
+
+
+def make_anonymizer(mode: str = "off", salt: str = "") -> Anonymizer:
+    mode = (mode or "off").strip().lower()
+    if mode not in _MODES:
+        logger.warning(f"Unknown ANALYTICS_ANONYMIZE={mode!r}; falling back to 'truncate'")
+        mode = "truncate"
+    if mode == "hash" and not salt:
+        logger.warning("ANALYTICS_ANONYMIZE=hash needs ANALYTICS_HASH_SALT; falling back to 'truncate'")
+        mode = "truncate"
+
+    def apply(raw: Dict[str, Any]) -> Dict[str, Any]:
+        out = dict(raw)
+        ip = out.get("client_ip")
+        if ip and mode != "off":
+            out["client_ip"] = _hash(ip, salt) if mode == "hash" else truncate_ip(ip)
+        proxy = out.get("proxy")
+        if proxy:
+            out["proxy"] = sanitize_proxy(proxy, hide_host=(mode != "off"), salt=salt)
+        return out
+
+    return apply

routes/streams/channels.py

@@ -8,7 +8,9 @@ channel-specific catchup handling logic.
 """
 
 import time
+from typing import Optional
 
+from analytics.integration import annotate, tracked
 from bottle import HTTPResponse, request, response
 from streaming_providers.base.stream_format import StreamFormat
 from streaming_providers.base.utils import logger
@@ -23,6 +25,7 @@ def setup_channel_routes(app, manager, service, helpers):
     _resolve_stream_unified = helpers["_resolve_stream_unified"]
     _get_drm_configs = helpers["_get_drm_configs"]
 
+    @tracked("channel")
     def _handle_channel_stream(provider, channel_id):
         """Single implementation backing /stream/index.mpd — the only channel
         stream route. Replaces the former /stream/sw-drm/, /stream/noproxy/,
@@ -58,8 +61,8 @@ def setup_channel_routes(app, manager, service, helpers):
             # Always defined so the _resolve_stream_unified call below is
             # unconditionally safe, even though the ternary guards already
             # prevent None from being passed when is_catchup is False.
-            start_time_int: int | None = None
-            end_time_int: int | None = None
+            start_time_int: Optional[int] = None
+            end_time_int: Optional[int] = None
 
             if is_catchup:
                 try:
@@ -87,6 +90,16 @@ def setup_channel_routes(app, manager, service, helpers):
                 # Window validation (catchup_hours lookup, age check) lives
                 # inside _resolve_stream_unified via _validate_catchup_window.
 
+            # Analytics context for @tracked — a dict update on request.environ,
+            # nothing else. Emission itself happens in the decorator's finally.
+            annotate(
+                is_catchup=is_catchup,
+                drm_variant=drm_variant,
+                client_drm=client_drm,
+                no_proxy=no_proxy,
+                highest_quality_only=highest_quality_only,
+            )
+
             return _resolve_stream_unified(
                 CONTENT_TYPE_CHANNEL, provider, channel_id,
                 country=country,

routes/streams/events.py

@@ -13,6 +13,7 @@ no_proxy/highest_quality_only query params on the one route, resolved
 directly through _resolve_stream_unified.
 """
 
+from analytics.integration import annotate, tracked
 from bottle import HTTPResponse, request, response
 from streaming_providers.base.stream_format import StreamFormat
 from streaming_providers.base.utils import logger
@@ -28,6 +29,7 @@ def setup_event_routes(app, manager, service, helpers):
     _resolve_stream_unified = helpers["_resolve_stream_unified"]
     _get_drm_configs = helpers["_get_drm_configs"]
 
+    @tracked("event")
     def _handle_event_stream(provider, event_id):
         """Single implementation backing /stream/index.mpd — the only event
         stream route. Replaces the former /stream/sw-drm/, /stream/proxied/,
@@ -58,6 +60,14 @@ def setup_event_routes(app, manager, service, helpers):
                 f"highest_quality_only={highest_quality_only}"
             )
 
+            # Analytics context for @tracked: a dict update on request.environ.
+            annotate(
+                drm_variant=drm_variant,
+                client_drm=client_drm,
+                no_proxy=no_proxy,
+                highest_quality_only=highest_quality_only,
+            )
+
             return _resolve_stream_unified(
                 CONTENT_TYPE_EVENT, provider, event_id,
                 country=country,

routes/streams/recordings.py

@@ -13,6 +13,7 @@ client_drm on the one route, resolved directly through
 _resolve_stream_unified.
 """
 
+from analytics.integration import annotate, tracked
 from bottle import HTTPResponse, request, response
 from streaming_providers.base.stream_format import StreamFormat
 from streaming_providers.base.utils import logger
@@ -28,6 +29,7 @@ def setup_recording_routes(app, manager, service, helpers):
     _resolve_stream_unified = helpers["_resolve_stream_unified"]
     _get_drm_configs = helpers["_get_drm_configs"]
 
+    @tracked("recording")
     def _handle_recording_stream(provider, recording_id):
         """Single implementation backing /stream/index.mpd — the only
         recording stream route. Replaces the former /stream/proxied/ route,
@@ -56,6 +58,14 @@ def setup_recording_routes(app, manager, service, helpers):
                 f"highest_quality_only={highest_quality_only}"
             )
 
+            # Analytics context for @tracked: a dict update on request.environ.
+            annotate(
+                drm_variant=drm_variant,
+                client_drm=client_drm,
+                no_proxy=no_proxy,
+                highest_quality_only=highest_quality_only,
+            )
+
             return _resolve_stream_unified(
                 CONTENT_TYPE_RECORDING, provider, recording_id,
                 country=country,

routes/streams/vod.py

@@ -13,6 +13,7 @@ query params on the one route, resolved directly through
 _resolve_stream_unified.
 """
 
+from analytics.integration import annotate, tracked
 from bottle import HTTPResponse, request, response
 from streaming_providers.base.stream_format import StreamFormat
 from streaming_providers.base.utils import logger
@@ -28,6 +29,7 @@ def setup_vod_routes(app, manager, service, helpers):
     _resolve_stream_unified = helpers["_resolve_stream_unified"]
     _get_drm_configs = helpers["_get_drm_configs"]
 
+    @tracked("vod")
     def _handle_vod_stream(provider, vod_id):
         """Single implementation backing /stream/index.mpd — the only VOD
         stream route. Replaces the former /stream/proxied/ and
@@ -60,6 +62,14 @@ def setup_vod_routes(app, manager, service, helpers):
                 f"highest_quality_only={highest_quality_only}"
             )
 
+            # Analytics context for @tracked: a dict update on request.environ.
+            annotate(
+                drm_variant=drm_variant,
+                client_drm=client_drm,
+                no_proxy=no_proxy,
+                highest_quality_only=highest_quality_only,
+            )
+
             return _resolve_stream_unified(
                 CONTENT_TYPE_VOD, provider, vod_id,
                 country=country,

service.py

@@ -3,6 +3,7 @@ import hashlib
 import json
 import math
 import os
+import signal
 import sys
 import threading
 import time
@@ -48,6 +49,7 @@ try:
         get_vfs_instance,
         is_kodi_environment,
     )
+    from analytics.emitter import configure_from_env, shutdown_emitter
 except ImportError as import_err:
     print(
         f"Ultimate Backend: Critical import failed - {str(import_err)}", file=sys.stderr
@@ -114,6 +116,10 @@ class UltimateService:
         self.vfs = get_vfs_instance(subdir="m3u_cache")
         logger.info(f"VFS initialized for M3U caching: {self.vfs.base_path}")
 
+        # Analytics: optional, best-effort, disabled unless ANALYTICS_ENDPOINT_URL
+        # is set. configure_from_env() never raises.
+        configure_from_env()
+
         self.mpd_cache = MPDCacheManager()
         logger.info(f"MPD cache initialized: {self.mpd_cache.vfs.base_path}")
 
@@ -2223,6 +2229,24 @@ class ThreadedWSGIRefServer(ServerAdapter):
             server.server_close()
 
 
+def _install_sigterm_handler():
+    """
+    Turn SIGTERM (docker stop, systemd) into a normal exit so start_service's
+    finally block runs. Python's default SIGTERM action kills the process
+    without running it. No-op outside the main thread (e.g. Kodi mode).
+    """
+    if threading.current_thread() is not threading.main_thread():
+        return
+
+    def _handler(signum, frame):
+        raise SystemExit(0)
+
+    try:
+        signal.signal(signal.SIGTERM, _handler)
+    except (ValueError, OSError):
+        pass
+
+
 def start_service(service_instance):
     """Start the Bottle server"""
     port = service_instance.server_port
@@ -2238,6 +2262,8 @@ def start_service(service_instance):
     except Exception as e:
         logger.warning(f"Channel warm-up could not be started: {e}")
 
+    _install_sigterm_handler()
+
     try:
         run(
             service_instance.app,
@@ -2249,6 +2275,7 @@ def start_service(service_instance):
         )
     finally:
         service_instance.manager.stop_auto_refresh()
+        shutdown_emitter()  # flush queued analytics events (bounded, never raises)
 
 
 def run_kodi_service():