T-HUB/apps/media-worker/src/producer/runner.test.ts

/**
 * Produce runner: one job end to end against a real ffmpeg, plus the loop's
 * lane/claim/failure behaviour, using store/storage doubles — no Postgres, no S3.
 *
 * The encode itself is exercised only when the system ffmpeg exists (the same
 * guard cpu-smoke/golden-frame tests lean on); everything else is double-driven.
 */
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { mkdtempSync, rmSync, statSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import test from "node:test";
import { createHash } from "node:crypto";
import { parseProduceRequest } from "./command-builder";
import { ProduceRunner } from "./runner";
import type { ClaimedProduceJob, ProduceJobStore, ProduceObjectStorage } from "./store";

const UUID = (n: number): string => `00000000-0000-4000-8000-${String(n).padStart(12, "0")}`;
const WS = "8f0c5a8e-3b1d-4c3e-9a5b-0c1d2e3f4a5b";
const SCRATCH = mkdtempSync(join(tmpdir(), "produce-runner-"));
const FONT = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf";

function hasFfmpeg(): boolean {
  return spawnSync("ffmpeg", ["-version"], { stdio: "ignore" }).status === 0;
}
/** ffmpeg 5.1 actually loads the font (drawtext) — a plain file probe is not enough. */
function hasFont(): boolean {
  const out = join(SCRATCH, "font-probe.png");
  return spawnSync("ffmpeg", ["-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "color=c=black:s=64x32:d=1",
    "-vf", `drawtext=fontfile=${FONT}:text='ok':fontsize=12`, "-frames:v", "1", out], { stdio: "ignore" }).status === 0;
}

/** A tiny 2 s 320x180 clip with tone audio — the produce preset rescales it. */
function makeClip(path: string): void {
  const r = spawnSync("ffmpeg", ["-hide_banner", "-loglevel", "error", "-y",
    "-f", "lavfi", "-i", "testsrc2=s=320x180:r=30:d=2",
    "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:d=2",
    "-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p", "-c:a", "aac", "-shortest",
    "-movflags", "+faststart", path]);
  if (r.status !== 0) throw new Error(`fixture encode failed: ${r.stderr?.toString() ?? ""}`);
}

function key(bytes: Buffer): string { return `sha256/${createHash("sha256").update(bytes).digest("hex")}.mp4`; }

/**
 * The audio-less class from the bug report: video only, no lavfi sine, no audio
 * track at all (muted screen recording / GIF conversion / synthetic test clip).
 */
function makeSilentClip(path: string): void {
  const r = spawnSync("ffmpeg", ["-hide_banner", "-loglevel", "error", "-y",
    "-f", "lavfi", "-i", "testsrc2=s=320x180:r=30:d=2",
    "-c:v", "libx264", "-preset", "ultrafast", "-pix_fmt", "yuv420p", "-movflags", "+faststart", path]);
  if (r.status !== 0) throw new Error(`silent fixture encode failed: ${r.stderr?.toString() ?? ""}`);
}

function doubles() {
  const events: Array<{ type: string; jobId?: string; detail?: string }> = [];
  const jobs: Array<{ jobId: string; state: string; reason?: string; output?: { storageKey: string; producedAssetId: string } }> = [];
  const objects = new Map<string, Buffer>();
  let published: { key: string } | null = null;
  let queue: ClaimedProduceJob[] = [];
  let lane: string | null = null;
  const assetKeys = new Map<string, string>();
  const store: ProduceJobStore = {
    async claim(requestedLane) {
      lane = requestedLane;
      const next = queue.shift();
      if (!next) return null;
      jobs.push({ jobId: next.jobId, state: "transcoding" });
      return next;
    },
    async progress() { /* the runner reports checkpoints; not asserted here */ },
    async complete(jobId, output) {
      const producedAssetId = UUID(99);
      const row = jobs.find((j) => j.jobId === jobId);
      if (!row) throw new Error("job not claimed");
      row.state = "complete";
      row.output = { storageKey: output.storageKey, producedAssetId };
      return producedAssetId;
    },
    async fail(jobId, reason) {
      const row = jobs.find((j) => j.jobId === jobId);
      if (row) { row.state = "failed"; row.reason = reason; }
    },
    async asset(_workspaceId, assetId) {
      const storageKey = assetKeys.get(assetId);
      return storageKey === undefined ? null : { storageKey, mimeType: "audio/mp4" };
    },
    async completeProbe() { /* probe completions are asserted in the pg-store suite */ },
    async assembleAsset(_workspaceId, assetId) {
      const storageKey = assetKeys.get(assetId);
      return storageKey === undefined ? null
        : { id: assetId, storageKey, mimeType: "video/mp4", sizeBytes: 1, width: 1280, height: 720, durationMs: 30_000, frameRate: 30, hasAudio: true };
    },
  };
  const storage: ProduceObjectStorage = {
    async getSealedObject(k) {
      const body = objects.get(k);
      if (body === undefined) throw new Error("not found");
      return body;
    },
    async getObject(k) {
      const body = objects.get(k);
      if (body === undefined) throw new Error("not found");
      return body;
    },
    async putContent(body) { published = { key: key(body) }; objects.set(published.key, body); return published; },
  };
  return { events, jobs, objects, assetKeys, published: () => published, setQueue: (items: ClaimedProduceJob[]) => { queue = items; }, laneOf: () => lane, store, storage };
}

function job(input: string, kind: "compose" | "mix", request: unknown, source: Partial<ClaimedProduceJob["source"]> = {}): ClaimedProduceJob {
  return {
    jobId: UUID(7),
    workspaceId: WS,
    kind,
    request: parseProduceRequest(request, kind),
    assemble: null,
    source: {
      id: UUID(1), storageKey: input, mimeType: "video/mp4", sizeBytes: 1,
      width: 320, height: 180, durationMs: 2000, frameRate: 30, hasAudio: true, ...source,
    },
    claimedAt: new Date().toISOString(),
  };
}

function runner(d: ReturnType<typeof doubles>, overrides: Partial<ConstructorParameters<typeof ProduceRunner>[0]> = {}): ProduceRunner {
  return new ProduceRunner({
    jobs: d.store, storage: d.storage, fontFile: FONT, scratchRoot: SCRATCH, lane: "local",
    redactor: { redact: (s) => s.replace(/\/[A-Za-z0-9._/-]+/g, "[path]") },
    onStderr: () => undefined,
    onEvent: (e) => d.events.push(e),
    ...overrides,
  });
}

test("compose job: encode, publish, derived asset, complete", { skip: !hasFfmpeg() || !hasFont() ? "needs ffmpeg + a ttf font" : false }, async () => {
  const d = doubles();
  const clip = join(SCRATCH, "source.mp4");
  makeClip(clip);
  const bytes = await import("node:fs/promises").then((fs) => fs.readFile(clip));
  const inputKey = key(bytes);
  d.objects.set(inputKey, bytes);
  d.setQueue([job(inputKey, "compose", { overlays: [{ kind: "lower_third", text: "Ada Lovelace", position: { x: 8, y: 85 } }], lane: "local" })]);

  await runner(d).pollOnce();

  assert.equal(d.jobs[0]?.state, "complete");
  assert.ok(d.jobs[0]?.output?.storageKey.startsWith("sha256/"));
  assert.ok(d.published()?.key.startsWith("sha256/"));
  assert.ok(d.events.some((e) => e.type === "complete"));
  // The publish is a real mp4 body (faststart box at the front).
  const body = d.objects.get(d.published()!.key);
  assert.ok(body !== undefined && body.subarray(4, 8).toString("latin1") === "ftyp");
});

test("an audio-less source produces output WITH audio (synthetic bed), not a [0:a] crash", { skip: !hasFfmpeg() || !hasFont() ? "needs ffmpeg + a ttf font" : false }, async () => {
  const d = doubles();
  const clip = join(SCRATCH, "source-silent.mp4");
  makeSilentClip(clip);
  const bytes = await import("node:fs/promises").then((fs) => fs.readFile(clip));
  const inputKey = key(bytes);
  d.objects.set(inputKey, bytes);
  // hasAudio:false is what the pg store reports for a probed audio-less asset.
  d.setQueue([job(inputKey, "compose",
    { overlays: [{ kind: "lower_third", text: "Ada Lovelace", position: { x: 8, y: 85 } }], lane: "local" },
    { hasAudio: false, durationMs: 2000 })]);

  await runner(d).pollOnce();

  assert.equal(d.jobs[0]?.state, "complete", d.jobs[0]?.reason);
  assert.ok(d.jobs[0]?.output?.storageKey.startsWith("sha256/"));
  // The produced mp4 must actually carry an audio stream — the whole point of the bed.
  const body = d.objects.get(d.published()!.key);
  assert.ok(body !== undefined && body.subarray(4, 8).toString("latin1") === "ftyp");
  const probe = spawnSync("ffprobe", ["-v", "error", "-print_format", "json", "-show_streams", "-i", `pipe:0`],
    { input: body, encoding: "buffer" });
  const streams = JSON.parse(probe.stdout.toString("utf8")).streams as Array<{ codec_type: string }>;
  assert.ok(streams.some((s) => s.codec_type === "audio"), "output must have an audio stream");
  assert.ok(streams.some((s) => s.codec_type === "video"), "output must keep its video stream");
});

test("a mix job with a voiceover runs the ducking path end to end", { skip: !hasFfmpeg() || !hasFont() ? "needs ffmpeg + a ttf font" : false }, async () => {
  const d = doubles();
  const clip = join(SCRATCH, "source-mix.mp4");
  makeClip(clip);
  const bytes = await import("node:fs/promises").then((fs) => fs.readFile(clip));
  const inputKey = key(bytes);
  d.objects.set(inputKey, bytes);
  const voPath = join(SCRATCH, "vo.m4a");
  const vo = spawnSync("ffmpeg", ["-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "sine=frequency=880:sample_rate=48000:d=2", "-c:a", "aac", voPath]);
  if (vo.status !== 0) throw new Error("voiceover fixture failed");
  const voBytes = await import("node:fs/promises").then((fs) => fs.readFile(voPath));
  const voKey = `sha256/${createHash("sha256").update(voBytes).digest("hex")}.m4a`;
  d.objects.set(voKey, voBytes);
  d.setQueue([job(inputKey, "mix", { voiceover: { assetId: UUID(5), gainDb: -6, duckOriginal: true } })]);
  d.assetKeys.set(UUID(5), voKey);

  await runner(d).pollOnce();
  assert.equal(d.jobs[0]?.state, "complete", d.jobs[0]?.reason);
});

test("an unresolvable voiceover fails the job with a bounded reason", async () => {
  const d = doubles();
  d.setQueue([job("sha256/" + "0".repeat(64) + ".mp4", "mix", { voiceover: { assetId: UUID(5), gainDb: -6, duckOriginal: false } })]);
  await runner(d).pollOnce();
  assert.equal(d.jobs[0]?.state, "failed");
  assert.equal(d.jobs[0]?.reason, "produce download failed");
  assert.ok(!d.events.some((e) => e.type === "complete"));
});

test("the loop claims for its lane only, idles when empty, and survives store errors", async () => {
  const d = doubles();
  let failures = 0;
  const flaky: ProduceJobStore = {
    ...d.store,
    claim: async (lane) => { if (failures++ < 2) throw new Error("connection refused"); return d.store.claim(lane); },
  };
  const r = new ProduceRunner({
    jobs: flaky, storage: d.storage, fontFile: FONT, scratchRoot: SCRATCH, lane: "remote",
    redactor: { redact: (s) => s }, onStderr: () => undefined, onEvent: (e) => d.events.push(e),
    pollIntervalMs: 5, scheduler: { setTimeout: (_ms, action) => { const t = setTimeout(action, 0); return { cancel: () => clearTimeout(t) }; } },
  });
  r.start();
  await new Promise((resolve) => setTimeout(resolve, 30));
  await r.stop();
  assert.equal(d.laneOf(), "remote");
  assert.ok(d.events.some((e) => e.type === "store-error"));
  assert.ok(d.events.some((e) => e.type === "idle"));
});

test("scratch is removed after a job, success or failure", async () => {
  const d = doubles();
  const before = join(SCRATCH, "before.txt");
  writeFileSync(before, "x");
  d.setQueue([job("sha256/" + "0".repeat(64) + ".mp4", "compose", {})]);
  await runner(d).pollOnce();
  assert.equal(d.jobs[0]?.state, "failed"); // the input object does not exist
  assert.ok(statSync(before).isFile(), "the scratch root itself must survive");
  const leftovers = spawnSync("find", [SCRATCH, "-mindepth", "1", "-maxdepth", "1", "-name", "produce-*"], { encoding: "utf8" });
  assert.equal(leftovers.stdout.trim(), "");
});

test("a store error is reported with the driver's real message, not just its name", async () => {
  const d = doubles();
  // node-postgres errors carry name:"error" and the cause in the message — the
  // production symptom was detail:"error" with the cause swallowed.
  const boom = new Error("bind message supplies 1 parameters, but prepared statement \"\" requires 0");
  boom.name = "error";
  const broken: ProduceJobStore = { ...d.store, claim: async () => { throw boom; } };
  // pollOnce is the direct path (the loop swallows into the same event), so the
  // rejection itself proves the error escaped the store call.
  await assert.rejects(() => runner(d, { jobs: broken }).pollOnce(), /bind message supplies/);
  const looping = runner(d, { jobs: broken, pollIntervalMs: 5, scheduler: { setTimeout: (_ms, action) => { const t = setTimeout(action, 0); return { cancel: () => clearTimeout(t) }; } } });
  looping.start();
  await new Promise((resolve) => setTimeout(resolve, 30));
  await looping.stop();
  const event = d.events.find((e) => e.type === "store-error");
  assert.ok(event !== undefined);
  assert.match(event.detail ?? "", /bind message supplies 1 parameters/);
  assert.notEqual(event.detail, "error");
});

test("a store error while failing a job is also reported with its real message", async () => {
  const d = doubles();
  d.setQueue([job("sha256/" + "0".repeat(64) + ".mp4", "compose", {})]); // the input object does not exist
  const boom = new Error("connection terminated unexpectedly");
  boom.name = "error";
  const failingStore: ProduceJobStore = { ...d.store, fail: async () => { throw boom; } };
  const r = runner(d, { jobs: failingStore });
  await r.pollOnce(); // run() swallows the fail() error into an event
  const event = d.events.find((e) => e.type === "store-error");
  assert.ok(event !== undefined);
  assert.match(event.detail ?? "", /connection terminated/);
});

test("the poll loop survives a store error and idles on the next tick", async () => {
  const d = doubles();
  let failures = 0;
  const flaky: ProduceJobStore = {
    ...d.store,
    claim: async (lane) => { if (failures++ < 2) throw new Error("connection refused"); return d.store.claim(lane); },
  };
  const r = new ProduceRunner({
    jobs: flaky, storage: d.storage, fontFile: FONT, scratchRoot: SCRATCH, lane: "remote",
    redactor: { redact: (s) => s }, onStderr: () => undefined, onEvent: (e) => d.events.push(e),
    pollIntervalMs: 5, scheduler: { setTimeout: (_ms, action) => { const t = setTimeout(action, 0); return { cancel: () => clearTimeout(t) }; } },
  });
  r.start();
  await new Promise((resolve) => setTimeout(resolve, 30));
  await r.stop();
  assert.equal(d.laneOf(), "remote");
  const storeErrors = d.events.filter((e) => e.type === "store-error");
  assert.ok(storeErrors.length >= 2);
  for (const event of storeErrors) assert.match(event.detail ?? "", /connection refused/);
  assert.ok(d.events.some((e) => e.type === "idle"));
});

test.after(() => { rmSync(SCRATCH, { recursive: true, force: true }); });