av-club/sidecar/index.ts

import { timingSafeEqual } from "node:crypto";
import { ASSETS_DIR, MAX_CONCURRENT_JOBS, MAX_UPLOAD_BYTES, PORT, requireToken } from "./config";
import { ApiError, invalid, notFound, toErrorResponse } from "./errors";
import { contentPath, deleteAsset, putAsset, readAsset } from "./assets/store";
import { ffprobe } from "./probe/ffprobe";
import { runSharp, type SharpRequest } from "./sharp/run";
import { runFfmpeg, type FfmpegRequest } from "./ffmpeg/run";
import sharp from "sharp";
import { createSidecarConnector } from "@@/sidecar/connect";
import { API_URL } from "@@/officer-url.mjs";

/**
 * av-club — ffmpeg, sharp and ffprobe over HTTP.
 *
 * The security model is in docs/SECURITY.md and is not optional reading: the caller never names a
 * path, and that single invariant is what makes it safe to expose the rest of these tools' surface.
 */

const encoder = new TextEncoder();

// Read once, at boot, so a missing token stops the process here rather than at the first request.
const TOKEN = requireToken();

/**
 * Constant-time bearer check.
 *
 * `===` on secrets leaks their length and their common prefix through timing. The cost of doing it
 * properly is three lines.
 */
function authorised(request: Request): boolean {
  /*
    The platform, having already authenticated somebody.

    `createSidecarProxy` forwards `X-Officer-User` after the account gate and the `av-club` permission
    gate have both passed, and this server binds LOOPBACK — so that header is only settable by something
    already on this machine, which is the same trust boundary caldav's DAV door relies on. Requiring the
    bearer token as well would mean shipping av-club's secret to every caller for no gain: the platform
    is the thing deciding who may encode, and it has already decided.
  */
  if (request.headers.get("x-officer-user")) return true;

  const header = request.headers.get("authorization");
  if (!header?.startsWith("Bearer ")) return false;

  const provided = encoder.encode(header.slice(7));
  const expected = encoder.encode(TOKEN);
  if (provided.byteLength !== expected.byteLength) return false;

  return timingSafeEqual(provided, expected);
}

/**
 * Bound concurrent processing.
 *
 * sharp and ffmpeg will each take every core available. Without this, two large requests can make
 * the box unresponsive to everything including the health check.
 */
let running = 0;
const queue: (() => void)[] = [];

async function withSlot<T>(work: () => Promise<T>): Promise<T> {
  if (running >= MAX_CONCURRENT_JOBS) {
    await new Promise<void>((resolve) => queue.push(resolve));
  }
  running++;
  try {
    return await work();
  } finally {
    running--;
    queue.shift()?.();
  }
}

/** Serve stored bytes, honouring Range — browsers will not scrub a <video> served without it. */
async function serveContent(id: string, request: Request): Promise<Response> {
  const asset = await readAsset(id);
  const file = Bun.file(contentPath(id));

  const type =
    asset.kind === "image"
      ? `image/${asset.format === "jpeg" ? "jpeg" : asset.format}`
      : file.type || "application/octet-stream";

  const range = request.headers.get("range");
  const match = range?.match(/^bytes=(\d*)-(\d*)$/);

  if (match) {
    const size = asset.bytes;
    const start = match[1] ? Number(match[1]) : 0;
    const end = match[2] ? Math.min(Number(match[2]), size - 1) : size - 1;

    if (!Number.isFinite(start) || start > end || start >= size) {
      return new Response(null, { status: 416, headers: { "Content-Range": `bytes */${size}` } });
    }

    return new Response(file.slice(start, end + 1), {
      status: 206,
      headers: {
        "Content-Type": type,
        "Content-Range": `bytes ${start}-${end}/${size}`,
        "Accept-Ranges": "bytes",
        "Content-Length": String(end - start + 1),
      },
    });
  }

  return new Response(file, {
    headers: {
      "Content-Type": type,
      "Accept-Ranges": "bytes",
      // Every id is unique and content never changes under one, so this is honest.
      "Cache-Control": "public, max-age=31536000, immutable",
      "X-Content-Type-Options": "nosniff",
    },
  });
}

async function handleUpload(request: Request): Promise<Response> {
  const form = await request.formData().catch(() => {
    throw invalid("Expected a multipart/form-data body with a `file` field");
  });

  const file = form.get("file");
  if (!(file instanceof File)) throw invalid("Expected a `file` field containing a file", "file");

  if (file.size > MAX_UPLOAD_BYTES) {
    throw new ApiError(
      "payload_too_large",
      `That file is ${(file.size / 1_048_576).toFixed(1)} MB. The limit is ${(MAX_UPLOAD_BYTES / 1_048_576).toFixed(0)} MB.`,
    );
  }

  const asset = await withSlot(async () => putAsset(await file.arrayBuffer()));
  return Response.json(asset, { status: 201 });
}

async function route(request: Request): Promise<Response> {
  const url = new URL(request.url);
  const path = url.pathname;

  if (path === "/health") {
    return Response.json({ ok: true, service: "av-club", running, queued: queue.length });
  }

  if (!authorised(request)) {
    return new ApiError("unauthorized", "Missing or invalid bearer token").toResponse();
  }

  if (path === "/v1/assets" && request.method === "POST") return handleUpload(request);

  const asset = path.match(/^\/v1\/assets\/([^/]+)(\/content|\/probe)?$/);
  if (asset) {
    const [, id, suffix] = asset;

    if (suffix === "/content" && request.method === "GET") return serveContent(id!, request);

    if (suffix === "/probe" && request.method === "GET") {
      const stored = await readAsset(id!);
      if (stored.kind === "image") {
        const metadata = await sharp(contentPath(id!)).metadata();
        return Response.json({ kind: "image", metadata });
      }
      const probe = await withSlot(() => ffprobe(contentPath(id!)));
      return Response.json({ kind: stored.kind, ...probe });
    }

    if (!suffix && request.method === "GET") return Response.json(await readAsset(id!));

    if (!suffix && request.method === "DELETE") {
      if (!(await readAsset(id!).then(() => true).catch(() => false))) throw notFound();
      await deleteAsset(id!);
      return new Response(null, { status: 204 });
    }
  }

  if (path === "/v1/sharp" && request.method === "POST") {
    const body = (await request.json().catch(() => {
      throw invalid("Expected a JSON body");
    })) as SharpRequest;
    const result = await withSlot(() => runSharp(body));
    return Response.json(result, { status: 201 });
  }

  if (path === "/v1/ffmpeg" && request.method === "POST") {
    const body = (await request.json().catch(() => {
      throw invalid("Expected a JSON body");
    })) as FfmpegRequest;
    const result = await withSlot(() => runFfmpeg(body));
    // A path result is not an asset and has no id, so it cannot carry a Location header.
    if (!("id" in result)) return Response.json(result, { status: 200 });
    return Response.json(result, { status: 201 });
  }

  throw notFound(`No route for ${request.method} ${path}`);
}

const server = Bun.serve({
  port: PORT,
  // Loopback, because `X-Officer-User` above is trusted and a header is only as good as who can set it.
  hostname: "127.0.0.1",
  // Bun's default is 10 seconds, which silently kills a long encode mid-flight.
  idleTimeout: 255,
  maxRequestBodySize: MAX_UPLOAD_BYTES,
  fetch: (request) => route(request).catch(toErrorResponse),
});

/*
  Register with the platform, and report the port it can be reached on.

  The port is still `AV_CLUB_PORT` rather than an ephemeral one: this service is also usable directly —
  by a script on the box, or by fofos/platform, which is its stated first consumer and is not Officer.
  Being a plugin is an additional way in, not a replacement for the one it already had.
*/
const connection = createSidecarConnector({
  apiUrl: `${API_URL}/api/sidecar/register`,
  name: "av-club",
  handles: ["av-club"],
  onCommand(_cmd, reply) {
    reply({ type: "error", id: (_cmd as { id: string }).id, error: "av-club takes no commands; use its HTTP API" });
  },
  onConnected() {
    connection.send({ type: "av-club:server", port: server.port! });
    console.log(`[av-club] reported port ${server.port} to the API`);
  },
});

function shutdown(signal: string) {
  console.log(`[av-club] ${signal} received, shutting down...`);
  try {
    server.stop(true);
  } catch {
    /* already stopped */
  }
  connection.destroy();
  process.exit(0);
}
process.on("SIGTERM", () => shutdown("SIGTERM"));
process.on("SIGINT", () => shutdown("SIGINT"));

console.log(`av-club on ${server.url}`);
console.log(`  assets      ${ASSETS_DIR}`);
console.log(`  max upload  ${(MAX_UPLOAD_BYTES / 1_048_576).toFixed(0)} MB`);
console.log(`  concurrency ${MAX_CONCURRENT_JOBS}`);