Media-Server-Setup/bin/dumpd.py

#!/usr/bin/env python3
"""Dump-tonight endpoint. Moves one streamer's night into trash (soft delete).

POST /dump   identity from X-Authentik-Username / X-Authentik-Groups (set by Caddy after Authentik)
             body (optional JSON): {"night":"YYYY-MM-DD","user":"name"}
GET  /dump   -> lists what a dump would move (dry run)
GET  /me/stream-token -> short-lived publish token for the caller (member of streamers/admins); the dock
             and the live portal set it as the OBS/WHIP credential. No stored passwords anywhere.
GET  /me/clips  -> Syncthing self-service status for the caller: our device id, their folder/device state
POST /me/clips  -> {"device_id": "..."}: create clips/<user> (+ .stignore), add their device to our Syncthing,
             create the receive-only folder clips-<user> shared with them (min free 300 GB). Idempotent.
POST /deploy    -> Gitea push webhook (bridge port only, HMAC-verified with DEPLOY_WEBHOOK_SECRET from
             /etc/vanguard/deploy.env). Runs /usr/local/bin/vanguard-deploy pull ONLY when DEPLOY_AUTO=1;
             otherwise logs the push and answers 202 "auto-deploy disabled".
POST /heartbeat  -> LEGACY, nothing calls this now. It was the Watchdog route when Alertmanager, and then a
                   Kibana webhook connector, posted here. A basic Elastic licence has no webhook connector, so
                   Kibana's rules write into the vanguard-alerts index instead and the deadman reads the
                   heartbeat from there. Kept because it costs nothing and is a ready-made target if an
                   external monitor is ever added.
POST /deploy-bot -> Gitea Actions (job container -> host gateway) hands over a Discord-bot release bundle,
                   HMAC-SHA256 signed with BOT_DEPLOY_SECRET; installs it with vanguard-deploy-bot.
                   May carry X-Vanguard-Env (base64 JSON): the bot's own settings, kept in Gitea, written
                   into the stack .env for the keys BOT_ENV_KEYS allows. Signature covers header+body.
GET  /stuck/api  -> every unpublished recording and why it is stuck (any gated user)
POST /stuck/api  -> {"action":"retry"|"drop","src":...}: clear the blocker, or move it to trash with the
             usual 3-day undo. Admin groups only; both actions run bin/vgstuck.sh so the web page and the
             CLI share one implementation.
GET  /metrics -> Prometheus text for Metricbeat: the media scripts' counters (bin/vgmetrics.sh) plus live
             gauges for the normalise queue and MediaMTX. Loopback listener only; Caddy routes nothing here.
POST /mtx-auth -> MediaMTX authHTTP hook: allows read/playback; allows publish iff user/pass carries a valid
             token whose user owns the path prefix. Tokens are HMAC-signed with /etc/vanguard/stream-token.secret.

"Night" = everything the user recorded from 17:00 Eastern on <night> until 02:00 Eastern the next day.
Default night: today if it is 17:00 or later, else yesterday (so 00:00-02:00 still counts as last night).
Users may dump only their own prefix; members of the admins / media-admins groups may pass "user" to dump anyone, or "all".
"""
import base64, hashlib, hmac, json, os, re, shutil, subprocess, sys, threading
import time as _time
from datetime import datetime, timedelta, time
from zoneinfo import ZoneInfo
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT   = Path(os.environ.get("MEDIA_ROOT", "/srv/media"))
SRC_DIRS = [ROOT / "streams", ROOT / "clips"]
TRASH  = ROOT / "trash"
MARK   = ROOT / ".markers"
TOKEN_SECRET_FILE = Path(os.environ.get("STREAM_TOKEN_SECRET_FILE", "/etc/vanguard/stream-token.secret"))
TOKEN_TTL_H = float(os.environ.get("STREAM_TOKEN_TTL_HOURS", "12"))
PUBLISH_GROUPS = set(g.strip() for g in os.environ.get("PUBLISH_GROUPS", "streamers,admins,media-admins").split(",") if g.strip())
AUTH_USER_HEADER = os.environ.get("AUTH_USER_HEADER", "X-Authentik-Username")
AUTH_GROUPS_HEADER = os.environ.get("AUTH_GROUPS_HEADER", "X-Authentik-Groups")
ADMIN_GROUPS = set(g.strip() for g in os.environ.get("DUMP_ADMIN_GROUPS", "admins,media-admins").split(",") if g.strip())
LISTEN = ("127.0.0.1", 8080)                     # /dump, /me/* — only via Caddy (identity headers)
AUTH_LISTEN = (os.environ.get("MTX_AUTH_BIND", "0.0.0.0"), int(os.environ.get("MTX_AUTH_PORT", "8081")))   # /mtx-auth only — reachable from Docker's bridge (firewall), used by MediaMTX
TZ = ZoneInfo(os.environ.get("DUMP_TZ", "America/New_York"))
NIGHT_START = time(17, 0)   # 5 pm
NIGHT_END   = time(2, 0)    # 2 am next day

AUTH_JWT_HEADER = os.environ.get("AUTH_JWT_HEADER", "X-Authentik-Jwt")

def proxy_claim(headers, name):
    """One claim out of the token the proxy injects, or None.

    Authentik's proxy puts its signed token in X-Authentik-Jwt; a scope mapping on the provider adds the claims
    we care about (authentik/vanguard-profile-scope-mapping.py supplies discord_id). The payload is only read,
    never trusted beyond what the sibling headers already assert: all of them come from the same proxy, which is
    the only thing that can reach this port. Returned for display, never for an authorisation decision here.
    """
    token = (headers.get(AUTH_JWT_HEADER) or "").strip()
    if token.count(".") != 2:
        return None
    body = token.split(".")[1]
    try:
        claims = json.loads(base64.urlsafe_b64decode(body + "=" * (-len(body) % 4)))
    except (ValueError, TypeError):
        return None
    value = claims.get(name)
    return str(value) if value not in (None, "") else None

def identity(headers):
    """(username, is_admin) from the headers Caddy sets after Authentik; (None, False) if absent.
    dumpd binds to loopback only, so these headers can only come from Caddy."""
    who = (headers.get(AUTH_USER_HEADER) or "").strip().lower()
    if not who:
        return None, False
    groups = {g.strip() for g in re.split(r"[|,]", headers.get(AUTH_GROUPS_HEADER) or "") if g.strip()}
    return who, bool(groups & ADMIN_GROUPS)

def default_night(now=None):
    now = now or datetime.now(TZ)
    d = now.date() if now.time() >= NIGHT_START else (now - timedelta(days=1)).date()
    return d.isoformat()

def night_window(night):
    d = datetime.strptime(night, "%Y-%m-%d").date()
    return (datetime.combine(d, NIGHT_START, tzinfo=TZ),
            datetime.combine(d + timedelta(days=1), NIGHT_END, tzinfo=TZ))

def candidates(user, night):
    start, end = night_window(night)
    out = []
    for base in SRC_DIRS:
        udirs = [p for p in base.iterdir() if p.is_dir()] if user == "all" else [base / user]
        for ud in udirs:
            if not ud.is_dir():
                continue
            for f in ud.rglob("*"):
                if not f.is_file():
                    continue
                mt = datetime.fromtimestamp(f.stat().st_mtime, TZ)
                if start <= mt <= end:
                    out.append(f)
    return out

def do_move(files, night):
    moved, size = 0, 0
    for f in files:
        rel = f.relative_to(ROOT)
        dst = TRASH / night / rel
        dst.parent.mkdir(parents=True, exist_ok=True)
        try:
            size += f.stat().st_size
            shutil.move(str(f), str(dst))
            moved += 1
        except FileNotFoundError:
            pass
    # drop now-empty title folders
    for base in SRC_DIRS:
        for d in sorted((p for p in base.rglob("*") if p.is_dir()), key=lambda p: -len(p.parts)):
            if d.parent != base and not any(d.iterdir()):
                d.rmdir()
    return moved, size

# ------------------------------------------------------------------ Syncthing (self-service clip intake)
import urllib.request, urllib.error
ST_URL = os.environ.get("SYNCTHING_URL", "http://127.0.0.1:8384")
ST_CONFIG = Path(os.environ.get("SYNCTHING_CONFIG", "/srv/media/.syncthing/config.xml"))
ST_CLIPS_PREFIX = os.environ.get("SYNCTHING_CLIPS_PREFIX", "/var/syncthing/clips")      # path INSIDE the syncthing container
STIGNORE_SRC = Path(os.environ.get("STIGNORE_SRC", "/etc/vanguard/stignore"))
MIN_FREE_GB = int(os.environ.get("CLIPS_MIN_FREE_GB", "300"))
CLIPS_QUOTA_GB = float(os.environ.get("CLIPS_QUOTA_GB", "100"))          # default per-user quota for clips/<user>
CLIPS_QUOTA_FILE = Path(os.environ.get("CLIPS_QUOTA_FILE", "/etc/vanguard/clip-quotas.json"))   # {"user": GB, ...} overrides
QUOTA_CHECK_S = int(os.environ.get("CLIPS_QUOTA_CHECK_S", "300"))
QUOTA_STATE = Path(os.environ.get("CLIPS_QUOTA_STATE", "/srv/media/.markers/clip-quota-paused.json"))

def quota_for(user):
    try:
        q = json.loads(CLIPS_QUOTA_FILE.read_text()).get(user)
        if q is not None: return float(q)
    except (OSError, json.JSONDecodeError, ValueError):
        pass
    return CLIPS_QUOTA_GB

def folder_bytes(path):
    total = 0
    for root, dirs, files in os.walk(path):
        for f in files:
            try: total += os.lstat(os.path.join(root, f)).st_size
            except OSError: pass
    return total

def quota_state():
    try: return json.loads(QUOTA_STATE.read_text())
    except (OSError, json.JSONDecodeError): return {}

def quota_enforce_once():
    """Pause a user's Syncthing folder when clips/<user> exceeds their quota; resume when back under 90 %."""
    paused = quota_state(); changed = False
    try:
        folders = st("GET", "/rest/config/folders")
    except Exception as e:
        log("quota_unreachable", level="warn", error=e); return
    for f in folders:
        if not f["id"].startswith("clips-"): continue
        user = f["id"][len("clips-"):]
        path = Path(ROOT) / "clips" / user
        if not path.is_dir(): continue
        used = folder_bytes(path); q = quota_for(user) * 1e9
        over = used > q; under = used < q * 0.9
        if over and not f.get("paused"):
            try: st("POST", f"/rest/db/pause?folder={f['id']}"); paused[user] = {"at": datetime.now(TZ).isoformat(), "used_gb": round(used / 1e9, 1), "quota_gb": q / 1e9}; changed = True
            except Exception as e: log("quota_pause_failed", level="error", user=user, error=e); continue
            log("quota_paused", level="warn", user=user, folder=f"clips-{user}", used_gb=f"{used/1e9:.1f}", quota_gb=f"{q/1e9:.0f}")
        elif under and f.get("paused") and user in paused:
            try: st("POST", f"/rest/db/resume?folder={f['id']}"); paused.pop(user, None); changed = True
            except Exception as e: log("quota_resume_failed", level="error", user=user, error=e); continue
            log("quota_resumed", user=user, folder=f"clips-{user}", used_gb=f"{used/1e9:.1f}")
    if changed:
        try: QUOTA_STATE.parent.mkdir(exist_ok=True); QUOTA_STATE.write_text(json.dumps(paused, indent=2))
        except OSError: pass

def quota_loop():
    while True:
        try: quota_enforce_once()
        except Exception as e: log("quota_loop_error", level="error", error=e)
        _time.sleep(QUOTA_CHECK_S)

def st_apikey():
    try:
        m = re.search(r"<apikey>([^<]+)</apikey>", ST_CONFIG.read_text())
        return m.group(1) if m else None
    except OSError:
        return None

def st(method, path, body=None):
    key = st_apikey()
    if not key: raise RuntimeError("syncthing api key not found")
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(ST_URL + path, data=data, method=method, headers={"X-API-Key": key, "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=10) as r:
        raw = r.read()
        return json.loads(raw) if raw else None

DEVICE_ID_RE = re.compile(r"^[A-Z2-7]{7}(-[A-Z2-7]{7}){7}$")

def clips_status(user):
    out = {"user": user, "folder_id": f"clips-{user}", "server_path": f"{ROOT}/clips/{user}"}
    try:
        out["server_device_id"] = st("GET", "/rest/system/status")["myID"]
        cfg_devices = st("GET", "/rest/config/devices")
        cfg_folders = st("GET", "/rest/config/folders")
        folder = next((f for f in cfg_folders if f["id"] == out["folder_id"]), None)
        out["folder_configured"] = folder is not None
        theirs = [d["deviceID"] for d in (folder or {}).get("devices", []) if d["deviceID"] != out["server_device_id"]]
        out["device_id"] = theirs[0] if theirs else None
        if out["device_id"]:
            conns = st("GET", "/rest/system/connections")["connections"]
            out["device_connected"] = bool(conns.get(out["device_id"], {}).get("connected"))
            out["device_name"] = next((d["name"] for d in cfg_devices if d["deviceID"] == out["device_id"]), None)
            try:
                comp = st("GET", f"/rest/db/completion?folder={out['folder_id']}&device={out['device_id']}")
                out["remote_completion"] = comp.get("completion")
            except Exception:
                pass
            try:
                stt = st("GET", f"/rest/db/status?folder={out['folder_id']}")
                out["files_here"] = stt.get("localFiles"); out["state"] = stt.get("state")
            except Exception:
                pass
    except Exception as e:
        out["error"] = f"syncthing: {e}"
    out["folder_exists"] = Path(out["server_path"]).is_dir()
    out["quota_gb"] = quota_for(user)
    out["used_gb"] = round(folder_bytes(out["server_path"]) / 1e9, 2) if out["folder_exists"] else 0.0
    ps = quota_state().get(user)
    out["paused_for_quota"] = bool(ps)
    if ps: out["paused_since"] = ps.get("at")
    return out

def clips_setup(user, device_id, device_name=None):
    device_id = device_id.strip().upper()
    if not DEVICE_ID_RE.match(device_id):
        raise ValueError("that does not look like a Syncthing Device ID (8 groups of 7 characters, e.g. ABCDEFG-HIJKLMN-...)")
    # 1. server folder with the video-only ignore list
    d = Path(ROOT) / "clips" / user
    d.mkdir(parents=True, exist_ok=True); shutil.chown(d, 1000, 1000)
    ign = d / ".stignore"
    if not ign.exists() and STIGNORE_SRC.exists():
        ign.write_text(STIGNORE_SRC.read_text()); shutil.chown(ign, 1000, 1000); ign.chmod(0o644)
    loc = d / ".stlocal"                     # #include'd by .stignore; Syncthing refuses to scan while it is missing
    if not loc.exists():
        loc.write_text(""); shutil.chown(loc, 1000, 1000); loc.chmod(0o644)
    # 2. their device
    my_id = st("GET", "/rest/system/status")["myID"]
    if device_id == my_id: raise ValueError("that is the server's own Device ID — paste yours (your Syncthing → Actions → Show ID)")
    devices = st("GET", "/rest/config/devices")
    if not any(x["deviceID"] == device_id for x in devices):
        st("POST", "/rest/config/devices", {"deviceID": device_id, "name": device_name or user, "addresses": ["dynamic"],
                                             "autoAcceptFolders": False, "introducer": False, "paused": False})
    # 3. receive-only folder shared with them
    fid = f"clips-{user}"
    folders = st("GET", "/rest/config/folders")
    cur = next((f for f in folders if f["id"] == fid), None)
    spec = {"id": fid, "label": f"Clips — {user}", "path": f"{ST_CLIPS_PREFIX}/{user}", "type": "receiveonly",
            "devices": [{"deviceID": my_id}, {"deviceID": device_id}],
            "minDiskFree": {"value": MIN_FREE_GB, "unit": "GB"}, "fsWatcherEnabled": True, "rescanIntervalS": 3600,
            "ignorePerms": True, "ignoreDelete": False}
    if cur is None:
        st("POST", "/rest/config/folders", spec)
    else:
        if not any(x["deviceID"] == device_id for x in cur.get("devices", [])):
            cur["devices"].append({"deviceID": device_id})
        cur["type"] = "receiveonly"; cur["minDiskFree"] = spec["minDiskFree"]
        st("PUT", f"/rest/config/folders/{fid}", cur)
    log("clips_setup", user=user, device=device_id[:7], folder=fid)
    return clips_status(user)

DEPLOY_ENV = Path(os.environ.get("DEPLOY_ENV_FILE", "/etc/vanguard/deploy.env"))
def deploy_env():
    d = {}
    try:
        for line in DEPLOY_ENV.read_text().splitlines():
            if "=" in line and not line.startswith("#"): k, v = line.split("=", 1); d[k.strip()] = v.strip()
    except OSError:
        pass
    return d

BOT_ENV_DEFAULT_KEYS = "DISCORD_BOT_TOKEN,DISCORD_GUILD_ID"
BOT_ENV_MAX_HEADER = 8192
_env_write_lock = threading.Lock()

def stack_env_path():
    """The stack's .env — docker compose interpolates ${...} from it, so the bot's settings live there."""
    repo = deploy_env().get("DEPLOY_REPO_DIR", "")
    return (Path(repo) / ".env") if repo else None

def apply_bot_env(values):
    """Merge allowed keys into the stack .env. Returns (applied, refused); never logs a value."""
    allowed = {k.strip() for k in deploy_env().get("BOT_ENV_KEYS", BOT_ENV_DEFAULT_KEYS).split(",") if k.strip()}
    refused = sorted(k for k in values if k not in allowed)
    wanted = {k: str(v) for k, v in values.items() if k in allowed}
    if not wanted:
        return [], refused
    path = stack_env_path()
    if path is None or not path.is_file():
        raise OSError("DEPLOY_REPO_DIR is not set or holds no .env")
    bad = [k for k, v in wanted.items() if not re.fullmatch(r"[A-Z][A-Z0-9_]{1,48}", k) or "\n" in v or "\r" in v]
    if bad:
        raise ValueError(f"not a usable env assignment: {bad}")
    with _env_write_lock:
        lines = path.read_text().splitlines()
        seen = set()
        for i, line in enumerate(lines):
            if "=" in line and not line.lstrip().startswith("#"):
                key = line.split("=", 1)[0].strip()
                if key in wanted:
                    lines[i] = f"{key}={wanted[key]}"; seen.add(key)
        for key in wanted:
            if key not in seen: lines.append(f"{key}={wanted[key]}")
        stat = path.stat()
        tmp = path.with_suffix(".env.incoming")
        tmp.write_text("\n".join(lines) + "\n")
        os.chmod(tmp, stat.st_mode & 0o777); os.chown(tmp, stat.st_uid, stat.st_gid)
        os.replace(tmp, path)
    applied = sorted(wanted)
    log("bot_env_applied", keys=",".join(applied), refused=",".join(refused) or None)
    return applied, refused

HEARTBEAT_FILE = Path(os.environ.get("ALERT_HEARTBEAT_FILE", "/srv/media/.markers/alert-heartbeat"))

_bot_deploy_lock = threading.Lock()
BOT_DEPLOY_MAX_BYTES = 300 * 1024 * 1024
BOT_DEPLOY_INSTALLER = os.environ.get("BOT_DEPLOY_INSTALLER", "/usr/local/bin/vanguard-deploy-bot")

def install_bot_bundle(raw, reason):
    """Write the uploaded bundle to a temp file and let vanguard-deploy-bot verify, swap and restart it."""
    import tempfile
    if not _bot_deploy_lock.acquire(blocking=False):
        return 409, {"error": "a bot deploy is already running"}
    tmp = None
    try:
        with tempfile.NamedTemporaryFile(prefix="vanguard-bot-bundle.", suffix=".bin", delete=False) as handle:
            handle.write(raw); tmp = handle.name
        log("bot_deploy_start", bytes=len(raw), reason=reason)
        done = subprocess.run([BOT_DEPLOY_INSTALLER, tmp], capture_output=True, text=True, timeout=600)
        output = (done.stdout + done.stderr).strip().splitlines()
        log("bot_deploy_exit", level=("info" if done.returncode == 0 else "error"), rc=done.returncode, reason=reason)
        return (200 if done.returncode == 0 else 500), {"installed": done.returncode == 0, "rc": done.returncode,
                                                        "output": output[-40:]}
    except subprocess.TimeoutExpired:
        log("bot_deploy_timeout", level="error", reason=reason)
        return 504, {"error": "the installer did not finish within 10 minutes"}
    except OSError as error:
        log("bot_deploy_error", level="error", error=error)
        return 500, {"error": str(error)}
    finally:
        if tmp:
            try: os.unlink(tmp)
            except OSError: pass
        _bot_deploy_lock.release()

_deploy_lock = {"running": False}
def run_deploy(reason):
    if _deploy_lock["running"]: return False
    _deploy_lock["running"] = True
    def go():
        try:
            log("deploy_start", reason=reason)
            r = subprocess.run(["/usr/local/bin/vanguard-deploy", "pull"], capture_output=True, text=True, timeout=1800)
            for line in (r.stdout + r.stderr).splitlines()[-40:]: print(f"deploy| {line}", flush=True)
            log("deploy_exit", level=("info" if r.returncode == 0 else "error"), rc=r.returncode)
        except Exception as e:
            log("deploy_error", level="error", error=e)
        finally:
            _deploy_lock["running"] = False
    threading.Thread(target=go, daemon=True).start()
    return True

def log(event, level="info", **kv):
    """One logfmt line per event (see LOGGING.md): ts= level= component=dumpd event= key=value ..."""
    def q(v):
        v = str(v)
        return v if v and not re.search(r'[\s"=]', v) else '"' + v.replace("\\", "\\\\").replace('"', '\\"') + '"'
    print(f"ts={datetime.now(TZ).isoformat(timespec='seconds')} level={level} component=dumpd event={event} "
          + " ".join(f"{k}={q(v)}" for k, v in kv.items() if v is not None), flush=True)

def token_secret():
    try:
        sec = TOKEN_SECRET_FILE.read_bytes().strip()
        if len(sec) >= 16: return sec
    except OSError:
        pass
    return None

def mint_token(user):
    """user|exp signed: <b64url(user|exp)>.<hmac hex[:40]> — URL/RTMP/SRT-safe characters only."""
    sec = token_secret()
    if not sec: raise RuntimeError("no stream token secret")
    exp = int(_time.time() + TOKEN_TTL_H * 3600)
    payload = f"{user}|{exp}".encode()
    sig = hmac.new(sec, payload, hashlib.sha256).hexdigest()[:40]
    return base64.urlsafe_b64encode(payload).decode().rstrip("=") + "." + sig, exp

def verify_token(token):
    """-> username or None"""
    sec = token_secret()
    if not sec or not token or "." not in token: return None
    b64, sig = token.rsplit(".", 1)
    try:
        payload = base64.urlsafe_b64decode(b64 + "=" * (-len(b64) % 4))
    except Exception:
        return None
    if not hmac.compare_digest(hmac.new(sec, payload, hashlib.sha256).hexdigest()[:40], sig): return None
    try:
        user, exp = payload.decode().split("|", 1)
        if int(exp) < _time.time(): return None
    except Exception:
        return None
    return user.lower()

# ---- stuck recordings -----------------------------------------------------------------------------
# GET  /stuck/api   every unpublished recording and why, for anyone the gate let through
# POST /stuck/api   {"action":"retry"|"drop","src":"<path>"} — admins only
# The classification and the file moves both live in bin/vgstuck.sh, so the web page and the CLI can never
# disagree about why something is stuck, and "delete" means the same three-day trash in both.
STUCK_TOOL = os.environ.get("VGSTUCK_TOOL", "/usr/local/bin/vgstuck.sh")

def stuck_list():
    done = subprocess.run([STUCK_TOOL, "json"], capture_output=True, text=True, timeout=180)
    try:
        return json.loads(done.stdout or "[]")
    except ValueError:
        log("stuck_list_failed", level="error", rc=done.returncode, err=(done.stderr or "")[:200])
        return []

def stuck_target(src):
    """Resolve what the browser asked us to act on, or raise. This is the only thing between a request body
    and a file move, so it is deliberately strict: inside the streams tree, an unpublished .live recording,
    and really there."""
    path = Path(src).resolve()
    root = (ROOT / "streams").resolve()
    if root != path and root not in path.parents:
        raise ValueError("that path is not inside the streams tree")
    if f"{os.sep}.live{os.sep}" not in str(path) or path.suffix.lower() != ".mp4":
        raise ValueError("that is not an unpublished recording")
    if not path.is_file():
        raise ValueError("no such recording (already published or already binned?)")
    return path

def stuck_action(action, src, who):
    path = stuck_target(src)
    done = subprocess.run([STUCK_TOOL, action, str(path)], capture_output=True, text=True, timeout=600)
    lines = [l for l in (done.stdout + done.stderr).splitlines() if l.strip()]
    log(f"stuck_{action}", level="info" if done.returncode == 0 else "error",
        file=str(path), by=who, rc=done.returncode)
    return {"ok": done.returncode == 0, "rc": done.returncode, "output": lines[-8:]}

# ---- GET /metrics -----------------------------------------------------------------------------------
# Prometheus text for Metricbeat's prometheus module. Two sources are merged:
#   1. counters the media scripts keep (bin/vgmetrics.sh writes MARK/metrics/<component>.counters)
#   2. gauges computed here at scrape time: unpublished recordings, encodes in flight, failures, MediaMTX
# Only the loopback listener serves it, and Caddy routes no host to /metrics (every dumpd route over there is
# an explicit path matcher), so this is reachable from the host and from nowhere else. Nothing here is
# authenticated and nothing here is allowed to fail: a scrape must never be able to disturb the stack.
METRICS_DIR = Path(os.environ.get("VG_METRICS_DIR", str(MARK / "metrics")))
MTX_API = os.environ.get("MTX_API_URL", "http://127.0.0.1:9997")
IDLE_S = int(os.environ.get("MINS_IDLE", "3")) * 60      # the sweep's own "finished being written" rule

# counters key head -> (metric, type, help, the labels the remaining key fields fill, in order)
COUNTER_MAP = {
    "jobs":          ("vanguard_media_jobs_total", "counter", "normalise/deband jobs that finished", ("verdict", "outcome", "kind")),
    "seconds":       ("vanguard_media_job_seconds_total", "counter", "wall-clock seconds spent in ffmpeg", ("verdict", "kind")),
    "runs":          ("vanguard_media_runs_total", "counter", "times the script has started work", ()),
    "deband":        ("vanguard_media_deband_total", "counter", "timecode bands cropped off", ()),
    "last_finish":   ("vanguard_media_last_finish_timestamp_seconds", "gauge", "when the script last finished", ()),
    "last_duration": ("vanguard_media_last_run_seconds", "gauge", "how long the last finished run took", ()),
}

def _num(value):
    """Prometheus wants a plain number: %g would turn a unix timestamp into 1.76e+09."""
    return "%d" % value if float(value).is_integer() else "%.3f" % value

def _glob(*patterns):
    found = []
    for pattern in patterns:
        try:
            found.extend(ROOT.glob(pattern))
        except OSError:
            pass
    return found

def mtx_paths():
    """MediaMTX's own view of what is live. None if it did not answer."""
    try:
        with urllib.request.urlopen(f"{MTX_API}/v3/paths/list?itemsPerPage=1000", timeout=1.5) as response:
            return json.load(response).get("items") or []
    except Exception:
        return None

def metrics_text():
    order, grouped = [], {}
    def add(metric, typ, helptext, labels, value):
        if metric not in grouped:
            grouped[metric] = (typ, helptext, [])
            order.append(metric)
        grouped[metric][2].append((labels, value))

    # 1. whatever the scripts have counted
    for counters in sorted(METRICS_DIR.glob("*.counters")):
        component = counters.name[: -len(".counters")]
        try:
            lines = counters.read_text().splitlines()
        except OSError:
            continue
        for line in lines:
            field = line.split()
            if len(field) != 2:
                continue
            try:
                value = float(field[1])
            except ValueError:
                continue
            head, *rest = field[0].split(":")
            if head not in COUNTER_MAP:
                continue
            metric, typ, helptext, names = COUNTER_MAP[head]
            labels = {"component": component}
            labels.update(dict(zip(names, rest)))
            add(metric, typ, helptext, labels, value)

    # 2. is each script executing right now? (its pid file exists only while it runs; a crash leaves a stale
    #    one, so the pid is checked against /proc rather than trusted)
    for pid_file in sorted(METRICS_DIR.glob("*.pid")):
        try:
            pid = int(pid_file.read_text().strip() or 0)
        except (OSError, ValueError):
            pid = 0
        add("vanguard_media_running", "gauge", "1 while the script is executing",
            {"component": pid_file.name[: -len(".pid")]}, 1 if pid and Path(f"/proc/{pid}").is_dir() else 0)

    # 3. the queue. Depth-limited globs on purpose: never a walk of the whole media tree.
    now = _time.time()
    recording, pending, oldest = 0, 0, 0.0
    for f in _glob("streams/*/*/.live/*"):
        if f.name.startswith(".") or not f.is_file():
            continue
        try:
            age = now - f.stat().st_mtime
        except OSError:
            continue
        # the sweep ignores anything touched in the last MINS_IDLE, so younger files are still being written
        if age <= IDLE_S:
            recording += 1
        else:
            pending += 1
            oldest = max(oldest, age)
    add("vanguard_media_recording", "gauge", "recordings MediaMTX is still writing into .live/", {}, recording)
    add("vanguard_media_pending", "gauge", "finished recordings waiting to be normalised", {"kind": "stream"}, pending)
    add("vanguard_media_pending_oldest_seconds", "gauge", "age of the oldest waiting recording", {}, round(oldest))
    add("vanguard_media_encoding_in_flight", "gauge", "ffmpeg output files open right now", {},
        len(_glob("streams/*/*.part", "streams/*/*/*.part", "streams/*/*/.live/*.part",
                  "clips/*/*.part", "clips/*/*/*.part", "clips/*/*/*/*.part")))
    add("vanguard_media_failed_files", "gauge", "files parked with a normalize-failed log", {},
        len(_glob("streams/*/*.normalize-failed.log", "streams/*/*/*.normalize-failed.log",
                  "clips/*/*.normalize-failed.log", "clips/*/*/*.normalize-failed.log",
                  "clips/*/*/*/*.normalize-failed.log")))

    # 4. MediaMTX. The Prometheus scrape of :9998 already has per-path detail; this is the count the portal
    #    and the dock actually mean by "live", with the derived .webrtc audio mirrors separated out.
    paths = mtx_paths()
    add("vanguard_mediamtx_up", "gauge", "did the MediaMTX API answer this scrape", {}, 0 if paths is None else 1)
    if paths is not None:
        ready = [p for p in paths if p.get("ready")]
        add("vanguard_streams_active", "gauge", "streams publishing right now, mirrors excluded", {},
            sum(1 for p in ready if not str(p.get("name", "")).endswith(".webrtc")))
        add("vanguard_streams_mirrors", "gauge", "derived .webrtc audio mirrors running", {},
            sum(1 for p in ready if str(p.get("name", "")).endswith(".webrtc")))
        add("vanguard_stream_readers", "gauge", "viewers attached across all paths", {},
            sum(len(p.get("readers") or []) for p in ready))

    out = []
    for metric in order:
        typ, helptext, series = grouped[metric]
        if helptext:
            out.append(f"# HELP {metric} {helptext}")
        out.append(f"# TYPE {metric} {typ}")
        for labels, value in series:
            tags = ",".join(f'{k}="{v}"' for k, v in sorted(labels.items()))
            out.append(f"{metric}{{{tags}}} {_num(value)}" if tags else f"{metric} {_num(value)}")
    return "\n".join(out) + "\n"

class H(BaseHTTPRequestHandler):
    server_version = "dumpd/1"
    def _json(self, code, obj):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def _text(self, code, body):
        raw = body.encode()
        self.send_response(code)
        self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)
    def _deny(self):
        self._json(401, {"error": "not authenticated (no identity header from the proxy)"})
    def _parse(self):
        n = int(self.headers.get("Content-Length") or 0)
        body = json.loads(self.rfile.read(n) or b"{}") if n else {}
        night = body.get("night")
        if night:
            datetime.strptime(night, "%Y-%m-%d")  # validate
        return night, body.get("user")
    def _resolve(self, night, tgt):
        """Explicit night wins. Otherwise: tonight, falling back to last night if tonight is empty."""
        if night:
            return night, candidates(tgt, night)
        n = default_night()
        files = candidates(tgt, n)
        if not files:
            prev = (datetime.strptime(n, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d")
            pf = candidates(tgt, prev)
            if pf:
                return prev, pf
        return n, files
    def _target(self, auth_user, is_admin, req_user):
        if is_admin:
            return req_user or auth_user          # admins default to their own night; pass "all" or a name for others
        return auth_user
    def do_GET(self):
        if self.server.server_port == AUTH_LISTEN[1]:
            return self._json(404, {"error": "not found"})          # the auth port serves POST /mtx-auth only
        if self.path == "/metrics":
            try:
                return self._text(200, metrics_text())
            except Exception as error:                  # a scrape must never take dumpd down
                log("metrics_failed", level="error", error=str(error))
                return self._text(500, f"# metrics unavailable: {error}\n")
        if self.path.startswith("/time"):
            # server clock for the timecode strip (the /clock/ page and Go Live measure their offset against this)
            return self._json(200, {"now_ms": int(_time.time() * 1000)})
        if self.path == "/me" or self.path == "/me/":
            who, is_admin = identity(self.headers)
            if not who: return self._json(401, {"error": "not authenticated"})
            groups = sorted({g.strip() for g in re.split(r"[|,]", self.headers.get(AUTH_GROUPS_HEADER) or "") if g.strip()})
            return self._json(200, {"user": who, "groups": groups, "admin": is_admin,
                                    "name": self.headers.get("X-Authentik-Name") or who, "email": self.headers.get("X-Authentik-Email") or "",
                                    "discord_id": proxy_claim(self.headers, "discord_id")})
        if self.path == "/stuck/api":
            who, is_admin = identity(self.headers)
            if not who: return self._json(401, {"error": "not authenticated"})
            return self._json(200, {"admin": is_admin, "user": who, "items": stuck_list()})
        if self.path.startswith("/me/clips"):
            who, _ = identity(self.headers)
            if not who: return self._json(401, {"error": "not authenticated"})
            return self._json(200, clips_status(who))
        if self.path.startswith("/me/stream-token") or self.path.startswith("/me/stream-key"):
            # Only reachable through Caddy behind Authentik forward-auth, which injects the identity headers.
            who, _ = identity(self.headers)
            if not who:
                return self._json(401, {"error": "not authenticated"})
            groups = {g.strip() for g in re.split(r"[|,]", self.headers.get(AUTH_GROUPS_HEADER) or "") if g.strip()}
            if not (groups & PUBLISH_GROUPS):
                return self._json(403, {"error": f"'{who}' is not in a publishing group ({', '.join(sorted(PUBLISH_GROUPS))}) — ask an admin for the Discord role"})
            try:
                tok, exp = mint_token(who)
            except RuntimeError as e:
                return self._json(500, {"error": str(e)})
            # "pass" kept for the dock/portal: the token IS the password for MediaMTX
            return self._json(200, {"user": who, "pass": tok, "token": tok, "expires": exp, "ttl_hours": TOKEN_TTL_H})
        if not self.path.startswith("/dump"):
            return self._json(404, {"error": "not found"})
        u, is_admin = identity(self.headers)
        if not u:
            return self._deny()
        night, req_user = self._parse()
        tgt = self._target(u, is_admin, req_user)
        night, files = self._resolve(night, tgt)
        self._json(200, {"night": night, "user": tgt, "dry_run": True,
                         "files": [str(f.relative_to(ROOT)) for f in files],
                         "gb": round(sum(f.stat().st_size for f in files) / 1e9, 2)})
    def do_POST(self):
        if self.server.server_port == AUTH_LISTEN[1] and self.path.startswith("/heartbeat"):
            # Alertmanager posts here every minute because the Watchdog alert always fires. Nothing is read from
            # the body and nothing is authenticated: the port only accepts the Docker bridge, and the only thing
            # this grants is keeping a timestamp fresh. The deadman treats a STALE file as the alarm.
            try:
                self.rfile.read(int(self.headers.get("Content-Length") or 0))
                HEARTBEAT_FILE.parent.mkdir(parents=True, exist_ok=True)
                HEARTBEAT_FILE.write_text(f"{int(_time.time())}\n")
            except OSError as error:
                return self._json(500, {"error": str(error)})
            return self._json(200, {"ok": True})
        if self.server.server_port == AUTH_LISTEN[1] and self.path.startswith("/deploy-bot"):
            # The bot's CI posts its release bundle here (bridge-only port; the firewall allows docker0/br-* alone).
            # The body is the artifact; the signature proves it came from a runner holding BOT_DEPLOY_SECRET.
            env = deploy_env()
            if env.get("BOT_DEPLOY_AUTO", "1") != "1":
                log("bot_deploy_refused", level="warn", reason="BOT_DEPLOY_AUTO=0")
                return self._json(403, {"error": "automatic bot deploys are turned off (BOT_DEPLOY_AUTO=0)"})
            secret = env.get("BOT_DEPLOY_SECRET", "")
            size = int(self.headers.get("Content-Length") or 0)
            if not secret:
                return self._json(503, {"error": "BOT_DEPLOY_SECRET is not set on the server"})
            if not 0 < size <= BOT_DEPLOY_MAX_BYTES:
                return self._json(413, {"error": f"body must be 1..{BOT_DEPLOY_MAX_BYTES} bytes, got {size}"})
            raw = self.rfile.read(size)
            env_b64 = (self.headers.get("X-Vanguard-Env") or "").strip()
            if len(env_b64) > BOT_ENV_MAX_HEADER:
                return self._json(413, {"error": "X-Vanguard-Env is too large"})
            sig = (self.headers.get("X-Vanguard-Signature") or "").replace("sha256=", "").strip()
            signed = env_b64.encode() + raw          # the settings are signed with the bundle, not beside it
            if not sig or not hmac.compare_digest(hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest(), sig):
                log("bot_deploy_bad_signature", level="warn", ip=self.client_address[0], bytes=len(raw))
                return self._json(401, {"error": "bad signature"})
            applied, refused = [], []
            if env_b64:
                try:
                    values = json.loads(base64.b64decode(env_b64, validate=True))
                    if not isinstance(values, dict): raise ValueError("X-Vanguard-Env must be a JSON object")
                    applied, refused = apply_bot_env(values)
                except (ValueError, OSError, TypeError) as error:
                    log("bot_env_failed", level="error", error=error)
                    return self._json(400, {"error": f"settings rejected: {error}"})
            code, body = install_bot_bundle(raw, self.headers.get("X-Vanguard-Reason") or "ci")
            body["settings_applied"] = applied
            if refused: body["settings_refused"] = refused
            return self._json(code, body)
        if self.server.server_port == AUTH_LISTEN[1] and self.path.startswith("/deploy"):
            # Gitea (container) -> host gateway -> here. Verify HMAC-SHA256 of the raw body with the webhook secret.
            n = int(self.headers.get("Content-Length") or 0); raw = self.rfile.read(n)
            env = deploy_env(); secret = env.get("DEPLOY_WEBHOOK_SECRET", "")
            sig = self.headers.get("X-Gitea-Signature") or self.headers.get("X-Hub-Signature-256", "").replace("sha256=", "")
            if not secret or not sig or not hmac.compare_digest(hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest(), sig):
                log("deploy_webhook_bad_signature", level="warn", ip=self.client_address[0])
                return self._json(401, {"error": "bad signature"})
            try: payload = json.loads(raw or b"{}")
            except json.JSONDecodeError: return self._json(400, {"error": "bad json"})
            ref = payload.get("ref", ""); branch = env.get("DEPLOY_BRANCH", "main")
            after = (payload.get("after") or "")[:10]; pusher = (payload.get("pusher") or {}).get("username", "?")
            if ref != f"refs/heads/{branch}":
                return self._json(200, {"ignored": f"ref {ref} is not {branch}"})
            if env.get("DEPLOY_AUTO", "0") != "1":
                log("deploy_webhook_push", rev=after, pusher=pusher, branch=branch, auto="0")
                return self._json(202, {"queued": False, "note": "auto-deploy disabled (DEPLOY_AUTO=0); run vanguard-deploy pull by hand"})
            started = run_deploy(f"push {after} by {pusher}")
            return self._json(202, {"queued": started, "note": "deploy started" if started else "a deploy is already running"})
        if self.server.server_port == AUTH_LISTEN[1] and not self.path.startswith("/mtx-auth"):
            return self._json(404, {"error": "not found"})
        if self.path.startswith("/mtx-auth"):
            # MediaMTX authHTTP: 200 = allow, anything else = deny. Body: user, password, token, ip, action, path, protocol, id, query
            n = int(self.headers.get("Content-Length") or 0)
            try:
                req = json.loads(self.rfile.read(n) or b"{}")
            except json.JSONDecodeError:
                return self._json(400, {"error": "bad json"})
            action = req.get("action", "")
            if action in ("read", "playback"):
                return self._json(200, {"ok": True})            # viewers are gated by Caddy + Authentik
            if action != "publish":
                return self._json(401, {"error": f"action {action} not allowed"})
            path = (req.get("path") or "").lower()
            if path.endswith(".webrtc"):
                # audio mirrors (<user>/<Title>.webrtc) are published by the hook inside the mediamtx container only
                if req.get("ip") in ("127.0.0.1", "::1") and re.fullmatch(r"[a-z0-9_.-]+/[a-z0-9_.-]+\.webrtc", path):
                    return self._json(200, {"ok": True})
                log("publish_deny", level="warn", stream=path, ip=req.get("ip"), reason="mirror paths are internal")
                return self._json(401, {"error": ".webrtc paths are reserved"})
            cred = req.get("password") or req.get("token") or ""
            if cred.lower().startswith("bearer "): cred = cred[7:]
            if ":" in cred and "." in cred.split(":", 1)[1]: cred = cred.split(":", 1)[1]   # "user:token" form
            user = verify_token(cred)
            if not user:
                log("publish_deny", level="warn", stream=path, ip=req.get("ip"), reason="bad or expired token")
                return self._json(401, {"error": "invalid or expired stream token"})
            if not re.fullmatch(r"[a-z0-9_.-]+/[a-z0-9_.-]+", path) or not path.startswith(user + "/"):
                log("publish_deny", level="warn", user=user, stream=path, reason="not their prefix")
                return self._json(401, {"error": "path must be <your-username>/<Title>"})
            log("publish_allow", user=user, stream=path, ip=req.get("ip"))
            return self._json(200, {"ok": True})
        if self.path == "/stuck/api":
            who, is_admin = identity(self.headers)
            if not who: return self._deny()
            if not is_admin:
                return self._json(403, {"error": f"retrying and binning recordings is for {', '.join(sorted(ADMIN_GROUPS))}"})
            try:
                body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}")
            except ValueError as error:
                return self._json(400, {"error": str(error)})
            action = body.get("action")
            if action not in ("retry", "drop"):
                return self._json(400, {"error": "action must be retry or drop"})
            try:
                return self._json(200, stuck_action(action, str(body.get("src", "")), who))
            except ValueError as error:
                return self._json(400, {"error": str(error)})
            except Exception as error:
                return self._json(500, {"error": str(error)})
        if self.path.startswith("/me/clips"):
            who, _ = identity(self.headers)
            if not who: return self._json(401, {"error": "not authenticated"})
            n = int(self.headers.get("Content-Length") or 0)
            try:
                body = json.loads(self.rfile.read(n) or b"{}")
                return self._json(200, clips_setup(who, str(body.get("device_id", "")), body.get("device_name")))
            except ValueError as e:
                return self._json(400, {"error": str(e)})
            except Exception as e:
                return self._json(502, {"error": f"syncthing: {e}"})
        if not self.path.startswith("/dump"):
            return self._json(404, {"error": "not found"})
        u, is_admin = identity(self.headers)
        if not u:
            return self._deny()
        try:
            night, req_user = self._parse()
        except Exception as e:
            return self._json(400, {"error": str(e)})
        tgt = self._target(u, is_admin, req_user)
        night, files = self._resolve(night, tgt)
        moved, size = do_move(files, night)
        MARK.mkdir(exist_ok=True)
        (MARK / f"{night}-{tgt}.dumped").write_text(f"{datetime.now(TZ).isoformat()} by {u}\n")
        log("dump", night=night, user=tgt, by=u, files=moved, gb=f"{size/1e9:.2f}")
        self._json(200, {"night": night, "user": tgt, "moved": moved, "gb": round(size / 1e9, 2),
                         "undo_until": (datetime.now(TZ) + timedelta(days=3)).strftime("%F")})
    def log_message(self, *a):  # quiet
        pass

if __name__ == "__main__":
    TRASH.mkdir(exist_ok=True); MARK.mkdir(exist_ok=True)
    print(f"dumpd listening on {LISTEN[0]}:{LISTEN[1]} (dump, me) and {AUTH_LISTEN[0]}:{AUTH_LISTEN[1]} (mtx-auth)  tz={TZ.key}  night={NIGHT_START:%H:%M}-{NIGHT_END:%H:%M}  token-secret={'ok' if token_secret() else 'MISSING'}", flush=True)
    auth_srv = ThreadingHTTPServer(AUTH_LISTEN, H)
    threading.Thread(target=auth_srv.serve_forever, daemon=True).start()
    threading.Thread(target=quota_loop, daemon=True).start()        # per-user clip quotas (pause/resume Syncthing folders)
    ThreadingHTTPServer(LISTEN, H).serve_forever()