cameras/src/lib/video.ts

import "server-only";
import { randomBytes } from "node:crypto";
import { rmSync } from "node:fs";
import { chmod, mkdir, rename, writeFile } from "node:fs/promises";
import path from "node:path";
import { rtspSourceWithLogin, type CameraTarget } from "./camera";
import {
  mediamtxApiUrl,
  mediamtxConfig,
  pathConfig,
  pathName,
  recordingPathConfig,
  recordPathName,
  type ApiLogin,
  type StreamKind,
} from "./mediamtx-config";
import { recordingsDir } from "./recordings";
import { mediamtxBinary as mediamtxBinaryPath, videoBridgeDir } from "./paths";
import { forgetPid, PID_FILE, recordPid, stopLeftoverMediamtx } from "./mediamtx-leftover";
import { MediamtxSupervisor, type BridgeStatus } from "./mediamtx-supervisor";

/**
 * The local video bridge (vrek gol-sxakryh, dec-xkn4z0e): starts MediaMTX with a
 * localhost-only config and a per-boot API password, and adds camera streams to it on
 * demand. Kept on globalThis because instrumentation.ts and the app are bundled
 * separately; both must see the same bridge.
 */

export class VideoBridgeError extends Error {
  name = "VideoBridgeError";
}

interface Bridge {
  supervisor: MediamtxSupervisor;
  login: ApiLogin;
  /** Path name → source it was last configured with, to skip no-op updates. */
  paths: Map<string, string>;
  /** The restart count those paths belong to; a restarted MediaMTX has none. */
  pathsEpoch: number;
}

const KEY = Symbol.for("cameras.videoBridge");
const g = globalThis as { [KEY]?: Bridge };

export function mediamtxBinary(): string {
  return mediamtxBinaryPath();
}

function bridgeDir(): string {
  return videoBridgeDir();
}

function authHeader(login: ApiLogin): string {
  return `Basic ${Buffer.from(`${login.user}:${login.pass}`).toString("base64")}`;
}

/** Basic auth for MediaMTX's localhost endpoints (API and WebRTC signaling). */
export function mediamtxAuthHeader(): string {
  const bridge = g[KEY];
  if (!bridge) throw new VideoBridgeError("Video bridge not started");
  return authHeader(bridge.login);
}

async function api(login: ApiLogin, route: string, init?: RequestInit): Promise<Response> {
  return fetch(mediamtxApiUrl(route), {
    ...init,
    headers: { ...init?.headers, Authorization: authHeader(login) },
    signal: AbortSignal.timeout(5_000),
  });
}

/** Polls MediaMTX's API until it answers, or rejects after about `attempts × intervalMs`. */
export async function waitForApi(login: ApiLogin, attempts = 40, intervalMs = 250): Promise<void> {
  for (let i = 0; i < attempts; i++) {
    const res = await api(login, "/v3/info").catch(() => null);
    if (res?.ok) return;
    await new Promise((r) => setTimeout(r, intervalMs));
  }
  throw new VideoBridgeError("MediaMTX API did not answer");
}

/** Writes the config owner-only, atomically. */
async function writeConfig(dir: string, text: string): Promise<string> {
  await mkdir(dir, { recursive: true, mode: 0o700 });
  await chmod(dir, 0o700);
  const file = path.join(dir, "mediamtx.yml");
  const tmp = `${file}.tmp`;
  await writeFile(tmp, text, { mode: 0o600 });
  await rename(tmp, file);
  return file;
}

/**
 * Starts the bridge once per server process. Doesn't wait for MediaMTX to be ready, so
 * server startup isn't held up; status() reports progress.
 */
export async function startVideoBridge(
  makeSupervisor: (opts: ConstructorParameters<typeof MediamtxSupervisor>[0]) => MediamtxSupervisor = (o) =>
    new MediamtxSupervisor(o),
  stopLeftover: typeof stopLeftoverMediamtx = stopLeftoverMediamtx,
): Promise<BridgeStatus> {
  if (g[KEY]) return g[KEY].supervisor.status();
  const login: ApiLogin = { user: "app", pass: randomBytes(24).toString("hex") };
  const dir = bridgeDir();
  const binary = mediamtxBinary();
  const configPath = path.join(dir, "mediamtx.yml");

  // A MediaMTX from a server that didn't exit cleanly would hold the ports (iss-yd2sq2q).
  const leftover = await stopLeftover(dir, binary, configPath);
  if (leftover === "stopped" || leftover === "killed") {
    console.log(`[video] stopped a MediaMTX left running by an earlier server (${leftover})`);
  }

  await writeConfig(dir, mediamtxConfig(login));
  const supervisor = makeSupervisor({
    binary,
    configPath,
    cwd: dir,
    waitReady: () => waitForApi(login),
    onSpawn: (pid) => void recordPid(dir, pid).catch(() => {}),
    onExit: () => void forgetPid(dir).catch(() => {}),
  });
  g[KEY] = { supervisor, login, paths: new Map(), pathsEpoch: 0 };
  process.once("exit", () => {
    supervisor.stop();
    // Synchronous: nothing async runs once the process is exiting.
    rmSync(path.join(dir, PID_FILE), { force: true });
  });
  supervisor.start();
  return supervisor.status();
}

export function videoBridgeStatus(): BridgeStatus {
  return g[KEY]?.supervisor.status() ?? { state: "stopped", restarts: 0 };
}

/** Test hook: forget the bridge (does not stop a real process). */
export function resetVideoBridgeForTests() {
  delete g[KEY];
}

/** Sends one path configuration to MediaMTX, creating the path if it doesn't exist yet. */
async function putPath(bridge: Bridge, name: string, config: object): Promise<void> {
  const init = {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(config),
  };
  let res = await api(bridge.login, `/v3/config/paths/replace/${name}`, init);
  if (res.status === 404) res = await api(bridge.login, `/v3/config/paths/add/${name}`, init);
  // MediaMTX's error text can echo the source URL, which holds the camera password.
  if (!res.ok) throw new VideoBridgeError(`MediaMTX refused the stream configuration (HTTP ${res.status})`);
}

function runningBridge(): Bridge {
  const bridge = g[KEY];
  const status = bridge?.supervisor.status();
  if (!bridge || status?.state !== "running") throw new VideoBridgeError("Video bridge is not running");
  if (status.restarts !== bridge.pathsEpoch) {
    bridge.paths.clear();
    bridge.pathsEpoch = status.restarts;
  }
  return bridge;
}

/**
 * Starts or stops recording a camera's stream (vrek iss-ws9nb88). While recording, the
 * camera is pulled continuously through a path of its own, so a viewer is never
 * interrupted; stopping puts that path back to on-demand, which drops the connection.
 */
export async function setRecording(target: CameraTarget, stream: StreamKind, on: boolean): Promise<void> {
  const bridge = runningBridge();
  const name = recordPathName(target.id, stream);
  const source = await rtspSourceWithLogin(target, stream);
  await putPath(bridge, name, on ? recordingPathConfig(source, recordingsDir()) : pathConfig(source));
  // The recording path's source is re-sent each time; don't let the cache skip a change.
  bridge.paths.delete(name);
}

/**
 * Makes sure MediaMTX has a path for this camera stream, pointing at the camera with its
 * current login, and returns the path name. The camera is only pulled while someone
 * watches (sourceOnDemand).
 */
export async function ensureStreamPath(target: CameraTarget, stream: StreamKind): Promise<string> {
  const bridge = runningBridge();
  const name = pathName(target.id, stream);
  const source = await rtspSourceWithLogin(target, stream);
  if (bridge.paths.get(name) === source) return name;
  await putPath(bridge, name, pathConfig(source));
  bridge.paths.set(name, source);
  return name;
}