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 { safeInputPath } from "./paths";
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);

  /*
    What is inside a file already on disk.

    The same answer as `/v1/assets/:id/probe`, for the same reason the ffmpeg endpoint grew a path
    mode: a caller deciding whether a conversion can be a lossless remux has to know the codecs, and
    uploading a two-hour film to ask is not a way to find out. Confined by the same `safeInputPath`.

    Read-only, so there is no output path to check — ffprobe is told a file and prints what it sees.
  */
  /*
    ── The file browser's two questions, answered for formats the platform cannot read ──

    The platform's thumbnailer uses sharp, whose bundled libheif has no HEVC decoder, so it does not
    claim `.heic` at all. This plugin does — see `fileTypes` in the manifest — and these are the routes
    the platform calls when it meets one.

    `size` and `width` are CLAMPED here rather than trusted. The caller is the platform rather than a
    browser, but a route that resizes to whatever number arrives is one bad query string from an
    allocation the size of the machine, and "the only caller is trusted" is a property that stops being
    true the first time somebody adds a second caller.

    Both stream bytes rather than returning an asset id: the platform caches the result under its own
    key, and an asset here would be a second copy of something already cached elsewhere, with no owner
    and nothing to expire it.
  */
  if ((path === "/v1/thumb" || path === "/v1/render") && request.method === "GET") {
    const target = await safeInputPath(url.searchParams.get("path"), "path");
    const thumb = path === "/v1/thumb";

    const asked = Number(url.searchParams.get(thumb ? "size" : "width"));
    const box = Number.isFinite(asked) && asked > 0 ? Math.min(Math.max(asked, 32), thumb ? 1024 : 4096) : thumb ? 256 : 2048;

    const result = await withSlot(() =>
      runSharp({
        inputPath: target,
        pipeline: [
          // EXIF orientation applied and the tag dropped, or every photo taken sideways arrives sideways.
          { rotate: {} },
          { resize: { width: box, height: box, fit: "inside", withoutEnlargement: true } },
        ],
        // JPEG for a thumbnail (small, universal); WebP for the viewer, which wants the quality back.
        output: thumb ? { format: "jpeg", options: { quality: 80, mozjpeg: true } } : { format: "webp", options: { quality: 90 } },
      }),
    );

    if (!("id" in result)) throw new Error("expected an asset");

    /*
      READ THE BYTES, THEN DELETE THE ASSET.

      `runSharp` without an `outputPath` stores its result, which is right for the API and wrong here:
      this is a derived view of a file that already exists, requested on every folder listing, and
      keeping one would fill the asset store with thumbnails nothing owns and nothing expires.

      Read fully rather than streamed from disk, because a streamed `Bun.file` would still be reading
      when the delete lands. These are a few hundred kilobytes at most — the size is bounded by the box
      clamped above.
    */
    const bytes = await Bun.file(contentPath(result.id)).arrayBuffer();
    await deleteAsset(result.id).catch(() => {});

    return new Response(bytes, {
      headers: {
        "Content-Type": thumb ? "image/jpeg" : "image/webp",
        // The platform caches under a key containing the source's mtime and size, so this answer cannot
        // go stale for a given URL. Its own cache is what matters; this is just the producer.
        "Cache-Control": "private, max-age=300",
      },
    });
  }

  if (path === "/v1/probe" && request.method === "GET") {
    const target = await safeInputPath(url.searchParams.get("path"), "path");
    return Response.json(await withSlot(() => ffprobe(target)));
  }

  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));
    if (!("id" in result)) return Response.json(result, { status: 200 });
    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}`);