feat: produce worker deployment + web producer/collective/branding UI + mobile CSS (wave-2c)
Deployment (t4g2): media-worker gains font-dejavu (drawtext lower-thirds) and a dedicated produce-worker compose service running src/produce.ts — claims video_processing_jobs compose|mix by lane, renders assembled/ overlaid/voiced cuts, registers derived library assets. Deliberately its own process so a remote-lane box needs only Postgres + S3 + ffmpeg. Web: compose media attachments UI (t2), Producer panel in the video drawer (t5), collective creation + branding settings + shell branding (brandweb), mobile CSS (mcss: touch targets, icon sizing, safe areas).
This commit is contained in:
parent
1183d69849
commit
d9e8cdf789
18 changed files with 2384 additions and 20 deletions
32
apps/media-worker/src/env.ts
Normal file
32
apps/media-worker/src/env.ts
Normal file
|
|
@ -0,0 +1,32 @@
|
|||
/**
|
||||
* Produce worker env parsing: the two settings a produce box needs that are not
|
||||
* plain pass-throughs — the lane this worker claims, and the lower-third font.
|
||||
*
|
||||
* Both are validated loudly at boot (a silently-defaulted lane would claim the
|
||||
* wrong rows; a missing font fails every compose job at encode time, far too late).
|
||||
*/
|
||||
import { accessSync, constants } from "node:fs";
|
||||
|
||||
/** `MEDIA_WORKER_LANE`: default worker claims local-or-NULL, a remote worker only remote. */
|
||||
export function laneFromEnv(value: string | undefined): "local" | "remote" {
|
||||
if (value === undefined || value === "") return "local";
|
||||
if (value === "local" || value === "remote") return value;
|
||||
throw new Error(`MEDIA_WORKER_LANE must be 'local' or 'remote' (got ${JSON.stringify(value)})`);
|
||||
}
|
||||
|
||||
/**
|
||||
* `MEDIA_PRODUCE_FONT`: absolute path to a readable .ttf. Defaults to the
|
||||
* DejaVu face present on the worker's base image; validated now (ENOENT at
|
||||
* encode time is a 15-minute-late failure for the first job that needs it).
|
||||
*/
|
||||
export function resolveProduceFont(value: string | undefined): string {
|
||||
const font = value ?? "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf";
|
||||
if (!font.startsWith("/")) throw new Error("MEDIA_PRODUCE_FONT must be an absolute path");
|
||||
if (!font.endsWith(".ttf")) throw new Error("MEDIA_PRODUCE_FONT must point to a .ttf");
|
||||
try {
|
||||
accessSync(font, constants.R_OK);
|
||||
} catch {
|
||||
throw new Error(`MEDIA_PRODUCE_FONT is not readable: ${font}`);
|
||||
}
|
||||
return font;
|
||||
}
|
||||
73
apps/media-worker/src/produce.ts
Normal file
73
apps/media-worker/src/produce.ts
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
/**
|
||||
* Produce worker entrypoint: the compose/mix job runner as its own process.
|
||||
*
|
||||
* Kept separate from src/main.ts (the live-channel supervisor) so a produce
|
||||
* worker runs where its lane needs it without dragging the supervisor's
|
||||
* dependencies (Redis, the fence, signed control) along: a remote-lane box needs
|
||||
* only Postgres + object storage + ffmpeg.
|
||||
*
|
||||
* MEDIA_WORKER_LANE=remote → claim only `input_json->>'lane' = 'remote'` rows
|
||||
* MEDIA_WORKER_LANE unset / 'local' (the default worker) → claim lane local or NULL
|
||||
*
|
||||
* Everything else is env parity with the API's storage (S3_ENDPOINT/S3_ACCESS_KEY/
|
||||
* S3_SECRET_KEY/S3_BUCKET/S3_KEK) and DATABASE_URL for the job tables.
|
||||
*/
|
||||
import pino from "pino";
|
||||
import { Pool as PgPool } from "pg";
|
||||
import { laneFromEnv, resolveProduceFont } from "./env";
|
||||
import { PostgresProduceJobStore } from "./producer/pg-store";
|
||||
import { ProduceRunner } from "./producer/runner";
|
||||
import { WorkerS3Storage, workerStorageConfigFromEnv } from "./producer/s3-storage";
|
||||
|
||||
const logger = pino({ level: process.env.LOG_LEVEL ?? "info" });
|
||||
|
||||
export async function startProduceWorker(): Promise<{ runner: ProduceRunner; pool: PgPool }> {
|
||||
const lane = laneFromEnv(process.env.MEDIA_WORKER_LANE);
|
||||
const fontFile = resolveProduceFont(process.env.MEDIA_PRODUCE_FONT);
|
||||
const pool = new PgPool({ connectionString: process.env.DATABASE_URL });
|
||||
const runner = new ProduceRunner({
|
||||
jobs: new PostgresProduceJobStore(pool),
|
||||
storage: new WorkerS3Storage(workerStorageConfigFromEnv()),
|
||||
fontFile,
|
||||
scratchRoot: process.env.MEDIA_SCRATCH_DIR ?? "/var/tmp/media-worker/produce",
|
||||
lane,
|
||||
redactor: { redact: () => "[produce stderr redacted]" },
|
||||
onStderr: (message) => logger.warn({ message }, "ffmpeg stderr"),
|
||||
onEvent: (event) => {
|
||||
switch (event.type) {
|
||||
case "idle": return;
|
||||
case "claimed": return logger.info({ jobId: event.jobId, lane }, "produce job claimed");
|
||||
case "complete": return logger.info({ jobId: event.jobId, detail: event.detail }, "produce job complete");
|
||||
case "failed": return logger.warn({ jobId: event.jobId, detail: event.detail }, "produce job failed");
|
||||
case "store-error": return logger.warn({ detail: event.detail }, "produce store error");
|
||||
}
|
||||
},
|
||||
});
|
||||
logger.info({ lane, fontFile }, "produce worker starting");
|
||||
runner.start();
|
||||
return { runner, pool };
|
||||
}
|
||||
|
||||
/**
|
||||
* Entry when run directly (`tsx src/produce.ts` / `node dist/produce.js`). No
|
||||
* top-level await: this package compiles CommonJS, so the start promise is
|
||||
* driven by main() and a boot failure still sets the process exit code.
|
||||
*/
|
||||
async function main(): Promise<void> {
|
||||
const entry = await startProduceWorker();
|
||||
for (const signal of ["SIGINT", "SIGTERM"] as const) {
|
||||
process.on(signal, () => {
|
||||
logger.info({ signal }, "produce worker stopping");
|
||||
// Stop polling first; an in-flight job finishes (its DB row is its lease),
|
||||
// then the pool closes and the process exits cleanly.
|
||||
void entry.runner.stop().finally(() => {
|
||||
void entry.pool.end().finally(() => process.exit(0));
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void main().catch((error: unknown) => {
|
||||
logger.error({ err: String(error) }, "produce worker startup failed");
|
||||
process.exitCode = 1;
|
||||
});
|
||||
273
apps/media-worker/src/producer/command-builder.test.ts
Normal file
273
apps/media-worker/src/producer/command-builder.test.ts
Normal file
|
|
@ -0,0 +1,273 @@
|
|||
/**
|
||||
* Produce command builder: pure-function tests for the filter chain and argv —
|
||||
* no ffmpeg run (that is the soak/e2e layer's job), exactly like the argv-shape
|
||||
* half of compositor/command-builder.test.ts.
|
||||
*
|
||||
* The contract under test is the one the API persists into input_json
|
||||
* (apps/api/src/video/produce-request.ts): overlay kinds lower_third / logo /
|
||||
* clock, positions as {x,y} percent or a named bottom preset, logos by
|
||||
* content-addressed storage key, an optional voiceover with optional gain, and
|
||||
* a `requestedBy` audit field the worker accepts but never renders.
|
||||
*
|
||||
* Covers: base scale chain, position resolution, drawtext quoting/escaping, logo
|
||||
* input ordering, clock expansion, amix vs sidechain ducking, and the rejection
|
||||
* paths that keep operator text out of the graph (§4.3).
|
||||
*/
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
import {
|
||||
buildProduceArgs, buildProduceFilter, drawtextQuoted, xExpression, yExpression, ProduceBuildError,
|
||||
parseProduceRequest, scratchAssetName, CLOCK_TEXT, resolvePosition,
|
||||
type OverlayPosition, type ProducePlan,
|
||||
} from "./command-builder";
|
||||
|
||||
const SCRATCH = "/var/tmp/media-worker/scratch";
|
||||
const HASH = "a".repeat(64);
|
||||
const LOGO_HASH = "b".repeat(64);
|
||||
const VO_HASH = "c".repeat(64);
|
||||
const FONT = `${SCRATCH}/fonts/Inter-SemiBold.ttf`;
|
||||
|
||||
function plan(overrides: Partial<ProducePlan> = {}): ProducePlan {
|
||||
return {
|
||||
input: { path: `${SCRATCH}/${HASH}.mp4`, durationSec: 12.5, audioBed: "source" },
|
||||
logos: [],
|
||||
voiceover: null,
|
||||
overlays: [],
|
||||
fontFile: FONT,
|
||||
outputPath: `${SCRATCH}/output.mp4`,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function graph(overrides: Partial<ProducePlan> = {}): string {
|
||||
const p = plan(overrides);
|
||||
return buildProduceFilter(p, {
|
||||
logos: p.logos.map((_, i) => i + 1),
|
||||
voiceover: p.voiceover === null ? null : p.logos.length + 1,
|
||||
});
|
||||
}
|
||||
|
||||
test("base chain scales and pads to 720p30 with no overlays", () => {
|
||||
const g = graph();
|
||||
assert.equal(g, `[0:v]scale=1280:720:force_original_aspect_ratio=decrease:force_divisible_by=2,pad=1280:720:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1,fps=30[base];[base]format=yuv420p[vout];[0:a]aformat=sample_fmts=fltp:sample_rates=48000:channel_layouts=stereo[aout]`);
|
||||
});
|
||||
|
||||
test("percent positions become clamped px expressions, single-quoted", () => {
|
||||
assert.equal(xExpression({ x: 5, y: 88 }), "min(64,W-tw)");
|
||||
assert.equal(yExpression({ x: 5, y: 88 }), "min(634,H-th)");
|
||||
assert.equal(xExpression({ x: 50, y: 0 }), "min(640,W-tw)");
|
||||
const g = graph({ overlays: [{ kind: "lower_third", text: "Ada Lovelace", logoStorageKey: null, position: { x: 5, y: 88 } }] });
|
||||
assert.ok(g.includes(`x='min(64,W-tw)':y='min(634,H-th)'`));
|
||||
});
|
||||
|
||||
test("named presets resolve to the same percent grid as the web's defaults", () => {
|
||||
const bottomLeft = resolvePosition("lower_third", "bottom-left", "p");
|
||||
assert.equal(bottomLeft.x, 5);
|
||||
assert.equal(bottomLeft.y, 88);
|
||||
const bottomRight = resolvePosition("lower_third", "bottom-right", "p");
|
||||
assert.equal(bottomRight.x, 95);
|
||||
const omitted = resolvePosition("clock", undefined, "p");
|
||||
assert.deepEqual(omitted, { x: 95, y: 5 });
|
||||
assert.throws(() => resolvePosition("clock", "top-left", "p"), ProduceBuildError);
|
||||
});
|
||||
|
||||
test("lower thirds draw in request order, after the base chain", () => {
|
||||
const g = graph({ overlays: [
|
||||
{ kind: "lower_third", text: "Ada Lovelace", logoStorageKey: null, position: { x: 5, y: 88 } },
|
||||
{ kind: "lower_third", text: "Counting Engine", logoStorageKey: null, position: { x: 50, y: 88 } },
|
||||
] });
|
||||
const chains = g.split(";");
|
||||
assert.match(chains[1]!, /^\[base\]drawtext=/);
|
||||
assert.match(chains[2]!, /^\[ov0\]drawtext=/);
|
||||
assert.match(chains[3]!, /^\[ov1\]format=yuv420p\[vout\]$/);
|
||||
assert.ok(chains[1]!.includes(`text='Ada Lovelace'`));
|
||||
// The second text is drawn on the first's output, not on the base again.
|
||||
assert.ok(!chains[2]!.includes("[base]"));
|
||||
});
|
||||
|
||||
test("drawtext quoting escapes backslash and quote, and expansion is disabled", () => {
|
||||
assert.equal(drawtextQuoted("plain"), "'plain'");
|
||||
assert.equal(drawtextQuoted("it's"), "'it'\\''s'");
|
||||
assert.equal(drawtextQuoted("a\\b"), "'a\\\\b'");
|
||||
// A caption containing filtergraph metacharacters is rejected outright (the
|
||||
// API blocks the same set on intake); the ones that survive are inert literals.
|
||||
const g = graph({ overlays: [{ kind: "lower_third", text: "50% louder", logoStorageKey: null, position: { x: 5, y: 88 } }] });
|
||||
assert.ok(g.includes("expansion=none"));
|
||||
assert.ok(g.includes(`text='50% louder'`));
|
||||
});
|
||||
|
||||
test("a clock overlay renders the builder's own constant with ffmpeg expansion", () => {
|
||||
const g = graph({ overlays: [{ kind: "clock", text: null, logoStorageKey: null, position: { x: 95, y: 5 } }] });
|
||||
assert.ok(g.includes(`text='${CLOCK_TEXT}'`));
|
||||
assert.ok(g.includes("expansion=normal"));
|
||||
// The clock payload carries no operator text at all.
|
||||
assert.ok(!g.includes("expansion=none"));
|
||||
});
|
||||
|
||||
test("logo overlays consume one downloaded input each, in request order", () => {
|
||||
const g = graph({
|
||||
overlays: [
|
||||
{ kind: "lower_third", text: "x", logoStorageKey: null, position: { x: 5, y: 88 } },
|
||||
{ kind: "logo", text: null, logoStorageKey: `sha256/${LOGO_HASH}.png`, position: { x: 95, y: 5 } },
|
||||
{ kind: "logo", text: null, logoStorageKey: `sha256/${"d".repeat(64)}.png`, position: { x: 5, y: 5 } },
|
||||
],
|
||||
logos: [{ path: `${SCRATCH}/${LOGO_HASH}.png` }, { path: `${SCRATCH}/${"d".repeat(64)}.png` }],
|
||||
});
|
||||
const chains = g.split(";");
|
||||
// chains: [base], drawtext(lt), overlay(logo1), overlay(logo2), format
|
||||
assert.match(chains[2]!, /^\[ov0\]\[1:v\]overlay=x='min\(1216,W-w\)':y='min\(36,H-h\)':eof_action=repeat\[ov1\]$/);
|
||||
assert.match(chains[3]!, /^\[ov1\]\[2:v\]overlay=x='min\(64,W-w\)':y='min\(36,H-h\)':eof_action=repeat\[ov2\]$/);
|
||||
assert.match(chains[4]!, /^\[ov2\]format=yuv420p\[vout\]$/);
|
||||
// No -loop 1: the overlay is one decoded frame held by eof_action=repeat.
|
||||
assert.ok(!buildProduceArgs(plan({
|
||||
overlays: [{ kind: "logo", text: null, logoStorageKey: `sha256/${LOGO_HASH}.png`, position: { x: 95, y: 5 } }],
|
||||
logos: [{ path: `${SCRATCH}/${LOGO_HASH}.png` }],
|
||||
})).includes("-loop"));
|
||||
});
|
||||
|
||||
test("a logo overlay without its downloaded logo is refused", () => {
|
||||
assert.throws(() => graph({
|
||||
overlays: [{ kind: "logo", text: null, logoStorageKey: `sha256/${LOGO_HASH}.png`, position: { x: 5, y: 5 } }],
|
||||
logos: [],
|
||||
}), ProduceBuildError);
|
||||
});
|
||||
|
||||
test("a voiceover without ducking is amix'd, unnormalized, gain applied to the voiceover", () => {
|
||||
const g = graph({ voiceover: { path: `${SCRATCH}/${VO_HASH}.m4a`, gainDb: -6, duck: false } });
|
||||
const chains = g.split(";");
|
||||
// chains: [base], format, bedFmt, voGain, amix
|
||||
assert.match(chains[1]!, /^\[base\]format=yuv420p\[vout\]$/);
|
||||
assert.match(chains[2]!, /^\[0:a\]aformat=[^;]*\[bedFmt\]$/);
|
||||
assert.match(chains[3]!, /^\[1:a\]aformat=[^;]*,volume=-6dB\[voGain\]$/);
|
||||
assert.match(chains[4]!, /^\[bedFmt\]\[voGain\]amix=inputs=2:duration=first:dropout_transition=2:normalize=0\[aout\]$/);
|
||||
assert.ok(!g.includes("sidechaincompress"));
|
||||
});
|
||||
|
||||
test("an unspecified gain emits no volume filter at all", () => {
|
||||
const g = graph({ voiceover: { path: `${SCRATCH}/${VO_HASH}.m4a`, gainDb: null, duck: false } });
|
||||
assert.ok(!g.includes("volume="));
|
||||
// chains: [base], format, bedFmt, voGain(no volume), amix
|
||||
assert.match(g.split(";")[3]!, /^\[1:a\]aformat=[^;]*\[voGain\]$/);
|
||||
});
|
||||
|
||||
test("ducking runs the voiceover as a sidechain key on the source bed", () => {
|
||||
const g = graph({ voiceover: { path: `${SCRATCH}/${VO_HASH}.m4a`, gainDb: -9, duck: true } });
|
||||
const chains = g.split(";");
|
||||
// chains: [base], format, vo asplit, bedRaw, sidechaincompress, voGain, amix
|
||||
assert.match(chains[2]!, /^\[1:a\]aformat=[^;]*,asplit=2\[voMix\]\[voKey\]$/);
|
||||
assert.match(chains[3]!, /^\[0:a\]aformat=[^;]*\[bedRaw\]$/);
|
||||
assert.match(chains[4]!, /^\[bedRaw\]\[voKey\]sidechaincompress=threshold=0\.03:ratio=8:attack=20:release=300:makeup=1:link=average\[duck\]$/);
|
||||
assert.match(chains[5]!, /^\[voMix\]volume=-9dB\[voGain\]$/);
|
||||
assert.match(chains[6]!, /^\[duck\]\[voGain\]amix=inputs=2:duration=first:dropout_transition=2:normalize=0\[aout\]$/);
|
||||
});
|
||||
|
||||
test("a source without audio mixes the voiceover over a synthetic silence bed", () => {
|
||||
const g = graph({ input: { path: `${SCRATCH}/${HASH}.mp4`, durationSec: 10, audioBed: "none" },
|
||||
voiceover: { path: `${SCRATCH}/${VO_HASH}.m4a`, gainDb: 0, duck: false } });
|
||||
assert.ok(g.includes("anullsrc=r=48000:cl=stereo"));
|
||||
// The synthetic bed is unbounded, so the mix runs to longest and -t bounds it.
|
||||
assert.ok(g.includes("duration=longest"));
|
||||
});
|
||||
|
||||
test("a synthetic bed must be bounded by the source duration", () => {
|
||||
assert.throws(() => graph({
|
||||
input: { path: `${SCRATCH}/${HASH}.mp4`, durationSec: null, audioBed: "none" },
|
||||
voiceover: { path: `${SCRATCH}/${VO_HASH}.m4a`, gainDb: 0, duck: false },
|
||||
}), ProduceBuildError);
|
||||
});
|
||||
|
||||
test("argv: inputs in filter order, one filter_complex, h264+aac mp4 out", () => {
|
||||
const args = buildProduceArgs(plan({
|
||||
overlays: [
|
||||
{ kind: "lower_third", text: "Hello", logoStorageKey: null, position: { x: 50, y: 88 } },
|
||||
{ kind: "logo", text: null, logoStorageKey: `sha256/${LOGO_HASH}.png`, position: { x: 5, y: 5 } },
|
||||
],
|
||||
logos: [{ path: `${SCRATCH}/${LOGO_HASH}.png` }],
|
||||
voiceover: { path: `${SCRATCH}/${VO_HASH}.m4a`, gainDb: -3, duck: true },
|
||||
}));
|
||||
const inputIndexes = args.flatMap((a, i) => (a === "-i" ? [args[i + 1]] : []));
|
||||
assert.deepEqual(inputIndexes, [
|
||||
`file:${SCRATCH}/${HASH}.mp4`,
|
||||
`file:${SCRATCH}/${LOGO_HASH}.png`,
|
||||
`file:${SCRATCH}/${VO_HASH}.m4a`,
|
||||
]);
|
||||
const fc = args.indexOf("-filter_complex");
|
||||
assert.ok(fc > 0 && args[fc + 1]!.includes("[2:a]"));
|
||||
assert.ok(args.includes("-t") && args[args.indexOf("-t") + 1] === "12.500");
|
||||
for (const token of ["-map", "[vout]", "[aout]", "-c:v", "libx264", "-pix_fmt", "yuv420p",
|
||||
"-b:v", "3000k", "-c:a", "aac", "-movflags", "+faststart", "-f", "mp4"]) {
|
||||
assert.ok(args.includes(token), `argv must include ${token}`);
|
||||
}
|
||||
const out = args[args.length - 1]!;
|
||||
assert.ok(out.startsWith("file:") && out.endsWith("output.mp4"));
|
||||
assert.ok(!args.some((a) => a.includes("$(") || a.startsWith(";")));
|
||||
});
|
||||
|
||||
test("validation: paths, fonts, and plan shape are rejected before any spawn", () => {
|
||||
assert.throws(() => buildProduceArgs(plan({ input: { ...plan().input, path: `${SCRATCH}/../../etc/passwd` } })), ProduceBuildError);
|
||||
assert.throws(() => buildProduceArgs(plan({ outputPath: plan().input.path })), ProduceBuildError);
|
||||
assert.throws(() => buildProduceArgs(plan({ fontFile: "/font.ttf" })), ProduceBuildError);
|
||||
assert.throws(() => buildProduceArgs(plan({ fontFile: "DejaVu Sans" })), ProduceBuildError);
|
||||
assert.throws(() => buildProduceArgs(plan({ input: { ...plan().input, durationSec: -1 } })), ProduceBuildError);
|
||||
assert.throws(() => buildProduceArgs(plan({
|
||||
overlays: [{ kind: "logo", text: null, logoStorageKey: `sha256/${LOGO_HASH}.png`, position: { x: 5, y: 5 } }],
|
||||
})), ProduceBuildError);
|
||||
assert.throws(() => buildProduceFilter(plan({ voiceover: { path: `${SCRATCH}/${VO_HASH}.m4a`, gainDb: 0, duck: false } }), { logos: [], voiceover: null }), ProduceBuildError);
|
||||
});
|
||||
|
||||
test("storage keys name their scratch copy by content hash", () => {
|
||||
assert.equal(scratchAssetName(`sha256/${HASH}.mp4`), `${HASH}.mp4`);
|
||||
assert.equal(scratchAssetName(`sha256/${LOGO_HASH}.png`), `${LOGO_HASH}.png`);
|
||||
assert.match(scratchAssetName("branding/logos/acme.png"), /^[0-9a-f]{64}\.bin$/);
|
||||
assert.throws(() => scratchAssetName("../../etc/passwd"), ProduceBuildError);
|
||||
assert.throws(() => scratchAssetName("sha256/nope.mp4"), ProduceBuildError);
|
||||
});
|
||||
|
||||
test("input_json is re-validated against the API's persisted shape", () => {
|
||||
const request = parseProduceRequest({
|
||||
overlays: [{ kind: "lower_third", text: "Ada", position: { x: 0, y: 100 } }],
|
||||
lane: "remote",
|
||||
outputLabel: "promo v2",
|
||||
requestedBy: "00000000-0000-4000-8000-000000000009",
|
||||
}, "compose");
|
||||
assert.equal(request.overlays.length, 1);
|
||||
assert.equal(request.overlays[0]!.kind, "lower_third");
|
||||
assert.deepEqual(request.overlays[0]!.position, { x: 0, y: 100 });
|
||||
assert.equal(request.lane, "remote");
|
||||
assert.equal(request.voiceover, null);
|
||||
assert.equal(request.requestedBy, "00000000-0000-4000-8000-000000000009");
|
||||
|
||||
// The exact payload the API persists for a mix request: optional gain/duck
|
||||
// omitted, voiceover present → kind mix, requestedBy appended by the API.
|
||||
const mix = parseProduceRequest({
|
||||
voiceover: { assetId: "00000000-0000-4000-8000-000000000002" },
|
||||
requestedBy: "00000000-0000-4000-8000-000000000009",
|
||||
}, "mix");
|
||||
assert.deepEqual(mix.voiceover, { assetId: "00000000-0000-4000-8000-000000000002", gainDb: null, duckOriginal: false });
|
||||
assert.equal(mix.lane, null);
|
||||
assert.equal(mix.overlays.length, 0);
|
||||
|
||||
// A logo overlay names a pre-rendered PNG by storage key; a clock has no payload.
|
||||
const branded = parseProduceRequest({
|
||||
overlays: [
|
||||
{ kind: "logo", logoStorageKey: `sha256/${LOGO_HASH}.png`, position: { x: 5, y: 5 } },
|
||||
{ kind: "clock", position: { x: 95, y: 5 } },
|
||||
],
|
||||
}, "compose");
|
||||
assert.equal(branded.overlays[0]!.logoStorageKey, `sha256/${LOGO_HASH}.png`);
|
||||
assert.equal(branded.overlays[1]!.kind, "clock");
|
||||
|
||||
// A voiceover on a compose row is refused (the API derives mix from the voiceover).
|
||||
assert.throws(() => parseProduceRequest({ voiceover: { assetId: "00000000-0000-4000-8000-000000000002", gainDb: -6, duckOriginal: true } }, "compose"), ProduceBuildError);
|
||||
// So are unknown fields, bad lanes, out-of-range positions and free-form text.
|
||||
assert.throws(() => parseProduceRequest({ extra: 1 }, "compose"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ lane: "gpu" }, "compose"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: "x", position: { x: 101, y: 0 } }] }, "compose"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ overlays: [{ kind: "chyron", text: "x", position: { x: 0, y: 0 } }] }, "compose"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ voiceover: { assetId: "not-a-uuid", gainDb: -6, duckOriginal: true } }, "mix"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ voiceover: { assetId: "00000000-0000-4000-8000-000000000002", gainDb: -49, duckOriginal: true } }, "mix"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ voiceover: { assetId: "00000000-0000-4000-8000-000000000002", gainDb: 13, duckOriginal: true } }, "mix"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ outputLabel: "bad label!" }, "compose"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: "a;b", position: { x: 0, y: 0 } }] }, "compose"), ProduceBuildError);
|
||||
assert.throws(() => parseProduceRequest({ overlays: [{ kind: "logo", logoStorageKey: "../../etc/passwd", position: { x: 0, y: 0 } }] }, "compose"), ProduceBuildError);
|
||||
});
|
||||
415
apps/media-worker/src/producer/command-builder.ts
Normal file
415
apps/media-worker/src/producer/command-builder.ts
Normal file
|
|
@ -0,0 +1,415 @@
|
|||
/**
|
||||
* Produce command builder (migrations/039_producer.sql; streaming-architecture.md §4.3).
|
||||
*
|
||||
* Typed params in, argv out — the same discipline as compositor/command-builder.ts:
|
||||
* nothing here takes a free-form filter string. Lower-third text is the one piece of
|
||||
* operator input that must reach ffmpeg, so it is charset-checked (the same blocklist
|
||||
* the API applies), quoted with `expansion=none`, and escaped for the filtergraph
|
||||
* parser; every other token (positions are percent → clamped px expressions, gains
|
||||
* are dB numbers, paths are scratch paths named by content hash) is produced from
|
||||
* validated values.
|
||||
*
|
||||
* THE CONTRACT this parses is the one the API persists, verbatim, into
|
||||
* video_processing_jobs.input_json (apps/api/src/video/produce-request.ts —
|
||||
* `{ ...request, requestedBy }`): overlay kinds `lower_third` / `logo` / `clock`,
|
||||
* positions as `{x,y}` PERCENT of frame (or one of three bottom presets), logos by
|
||||
* content-addressed storage key, an optional voiceover mix with optional gain. The
|
||||
* worker re-validates because input_json is a queued row anyone with DB write
|
||||
* access could have touched since the API parsed it.
|
||||
*
|
||||
* One process per job: input → scale/pad to 720p30 → overlays in request order
|
||||
* (lower thirds, logos, clock) → optional voiceover mix (amix, or sidechain
|
||||
* ducking) → h264 + aac in an mp4 written to a scratch path.
|
||||
*/
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
/** The one output preset a produce job renders (the 009 profile table's 720p30). */
|
||||
export const PRODUCE_PRESET = { width: 1280, height: 720, fps: 30, videoBitrateKbps: 3000, audioBitrateKbps: 128, sampleRate: 48000 } as const;
|
||||
|
||||
export type Lane = "local" | "remote";
|
||||
export type ProduceJobKind = "compose" | "mix";
|
||||
|
||||
/** Named bottom-row presets the web posts; the API also accepts raw `{x,y}` percent. */
|
||||
export type OverlayPositionPreset = "bottom-left" | "bottom-center" | "bottom-right";
|
||||
export const OVERLAY_POSITION_PRESETS: readonly OverlayPositionPreset[] = ["bottom-left", "bottom-center", "bottom-right"];
|
||||
|
||||
/** Percent of the output frame on each axis (produce-request.ts: 0..100 inclusive). */
|
||||
export interface OverlayPosition { x: number; y: number }
|
||||
export const POSITION_MIN = 0;
|
||||
export const POSITION_MAX = 100;
|
||||
|
||||
/** Per-kind default placement, as percent: thirds sit low, the clock sits top-right, logos top-left. */
|
||||
export const DEFAULT_POSITIONS: Readonly<Record<ParsedOverlay["kind"], OverlayPosition>> = {
|
||||
lower_third: { x: 5, y: 88 },
|
||||
logo: { x: 5, y: 5 },
|
||||
clock: { x: 95, y: 5 },
|
||||
};
|
||||
|
||||
export const MAX_OVERLAYS = 64; // mirrors production.ts / the API's cap
|
||||
export const MAX_TEXT_CHARS = 200; // mirrors produce-request.ts
|
||||
export const GAIN_DB_MIN = -48; // mirrors produce-request.ts (continuous, not stepped)
|
||||
export const GAIN_DB_MAX = 12;
|
||||
|
||||
/**
|
||||
* Characters that change an ffmpeg filtergraph (`;` splits a chain, `[`/`]` pad
|
||||
* labels, `,` separates filter args, `'` quotes, `=` sets an option) plus the shell
|
||||
* metacharacters. The API blocks this set on intake; the worker re-blocks it.
|
||||
*/
|
||||
const TEXT_FORBIDDEN = new Set([";", "[", "]", "'", '"', "`", "$", "\\", "{", "}", "=", ",", ":"]);
|
||||
|
||||
/** Storage keys are content-addressed, flat, lowercase (produce-request.ts's shape). */
|
||||
const STORAGE_KEY = /^[a-z0-9][a-z0-9/._-]{7,255}$/;
|
||||
/** A file the worker wrote into the job's scratch dir: a 64-hex name, or the encode. */
|
||||
const SCRATCH_FILE = /^\/(?:[A-Za-z0-9._-]+\/)*(?:[0-9a-f]{64}|output)\.(?:mp4|png|webp|jpg|jpeg|m4a|aac|mp3|wav|webm)$/;
|
||||
const FONT_FILE = /^\/(?:[A-Za-z0-9._+-]+\/)+[A-Za-z0-9._+-]+\.ttf$/;
|
||||
const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
|
||||
/** Mirrors the API's label rule: letters/digits first, then letters, digits, space, `_ . -`. */
|
||||
const LABEL = /^[\p{L}\p{N}][\p{L}\p{N} _.-]{0,63}$/u;
|
||||
|
||||
export class ProduceBuildError extends Error {
|
||||
constructor(message: string) { super(message); this.name = "ProduceBuildError"; }
|
||||
}
|
||||
function fail(message: string): never { throw new ProduceBuildError(message); }
|
||||
|
||||
/** The clock overlay's text: a builder constant (never operator input), evaluated by ffmpeg. */
|
||||
export const CLOCK_TEXT = "%{pts\\:hms}";
|
||||
|
||||
/**
|
||||
* drawtext text value: single-quoted for the filtergraph parser (`\` and `'` are the
|
||||
* only two characters that can escape a quote), with `%` expansion disabled at the
|
||||
* filter, so no operator text can become an expression or end the argument.
|
||||
*/
|
||||
export function drawtextQuoted(text: string): string {
|
||||
return `'${text.replace(/\\/g, "\\\\").replace(/'/g, "'\\''")}'`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Percent -> a clamped px x expression. `extent` is the width expression of the
|
||||
* thing being placed: `tw` for drawtext (rendered text width) but `w` for overlay
|
||||
* (the overlaid input's width - in `overlay`, `t` is the timestamp, so a stray
|
||||
* `tw` there would drift the logo across the frame over time).
|
||||
*/
|
||||
export function xExpression(position: OverlayPosition, extent: "tw" | "w" = "tw"): string {
|
||||
return `min(${Math.round((position.x / 100) * PRODUCE_PRESET.width)},W-${extent})`;
|
||||
}
|
||||
/** Percent -> a clamped px y expression (`th` for drawtext, `h` for overlay). */
|
||||
export function yExpression(position: OverlayPosition, extent: "th" | "h" = "th"): string {
|
||||
return `min(${Math.round((position.y / 100) * PRODUCE_PRESET.height)},H-${extent})`;
|
||||
}
|
||||
|
||||
/** Resolve a posted position (percent object or a named preset) to percent coordinates. */
|
||||
export function resolvePosition(kind: ParsedOverlay["kind"], posted: unknown, field: string): OverlayPosition {
|
||||
if (posted === undefined || posted === null) return DEFAULT_POSITIONS[kind];
|
||||
if (typeof posted === "string") {
|
||||
if (!OVERLAY_POSITION_PRESETS.includes(posted as OverlayPositionPreset)) fail(`${field} must be a position object or one of ${OVERLAY_POSITION_PRESETS.join("|")}`);
|
||||
switch (posted as OverlayPositionPreset) {
|
||||
case "bottom-left": return { x: 5, y: 88 };
|
||||
case "bottom-center": return { x: 50, y: 88 };
|
||||
case "bottom-right": return { x: 95, y: 88 };
|
||||
}
|
||||
}
|
||||
if (typeof posted !== "object" || Array.isArray(posted)) fail(`${field} must be a position object or one of ${OVERLAY_POSITION_PRESETS.join("|")}`);
|
||||
const p = posted as Record<string, unknown>;
|
||||
for (const key of Object.keys(p)) if (key !== "x" && key !== "y") fail(`${field}.${key} is not allowed`);
|
||||
return { x: percent(p.x, `${field}.x`), y: percent(p.y, `${field}.y`) };
|
||||
}
|
||||
|
||||
function percent(value: unknown, field: string): number {
|
||||
if (typeof value !== "number" || !Number.isFinite(value)) fail(`${field} must be a number (percent of frame)`);
|
||||
if (value < POSITION_MIN || value > POSITION_MAX) fail(`${field} must be between ${POSITION_MIN} and ${POSITION_MAX}`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function operatorText(value: unknown, field: string): string {
|
||||
if (typeof value !== "string") fail(`${field} must be a string`);
|
||||
if (value.trim() === "") fail(`${field} must not be empty`);
|
||||
if (value.length > MAX_TEXT_CHARS) fail(`${field} must be at most ${MAX_TEXT_CHARS} characters`);
|
||||
for (const character of value) {
|
||||
const code = character.codePointAt(0) ?? 0;
|
||||
if (code < 0x20 || code === 0x7f) fail(`${field} must not contain control characters`);
|
||||
if (TEXT_FORBIDDEN.has(character)) fail(`${field} must not contain the filtergraph metacharacter ${JSON.stringify(character)}`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function contentKey(value: unknown, field: string): string {
|
||||
if (typeof value !== "string" || !STORAGE_KEY.test(value) || value.includes("..")) fail(`${field} must be a content-addressed storage key`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function uuid(value: unknown, field: string): string {
|
||||
if (typeof value !== "string" || !UUID.test(value)) fail(`${field} must be a UUID`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function bool(value: unknown, field: string): boolean {
|
||||
if (typeof value !== "boolean") fail(`${field} must be a boolean`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function optional<T>(value: unknown, field: string, parse: (v: unknown, f: string) => T): T | null {
|
||||
return value === undefined || value === null ? null : parse(value, field);
|
||||
}
|
||||
|
||||
function onlyKeys(value: object, allowed: readonly string[], field: string): void {
|
||||
for (const key of Object.keys(value)) if (!allowed.includes(key)) fail(`${field}.${key} is not allowed`);
|
||||
}
|
||||
|
||||
/** One overlay after worker-side parsing: kind, its payload, and a resolved position. */
|
||||
export interface ParsedOverlay {
|
||||
kind: "lower_third" | "logo" | "clock";
|
||||
/** lower_third only: the burned-in caption. */
|
||||
text: string | null;
|
||||
/** logo only: the pre-rendered PNG's storage key (downloaded like the source bytes). */
|
||||
logoStorageKey: string | null;
|
||||
/** Resolved percent placement (default per kind when the request omitted it). */
|
||||
position: OverlayPosition;
|
||||
}
|
||||
|
||||
export interface ParsedVoiceover { assetId: string; gainDb: number | null; duckOriginal: boolean }
|
||||
|
||||
/** The validated `video_processing_jobs.input_json` of a compose/mix row. */
|
||||
export interface ProduceRequest {
|
||||
overlays: ParsedOverlay[];
|
||||
/** Present only on mix rows (the API derives the kind from the voiceover). */
|
||||
voiceover: ParsedVoiceover | null;
|
||||
lane: Lane | null;
|
||||
outputLabel: string | null;
|
||||
requestedBy: string | null;
|
||||
}
|
||||
|
||||
function gainDbChecked(value: unknown): number {
|
||||
if (typeof value !== "number" || !Number.isFinite(value) || value < GAIN_DB_MIN || value > GAIN_DB_MAX) {
|
||||
fail(`voiceover.gainDb must be a number between ${GAIN_DB_MIN} and ${GAIN_DB_MAX} dB`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
/** Strict allowlist parse of the API's persisted `input_json`; the worker re-validates it. */
|
||||
export function parseProduceRequest(value: unknown, kind: ProduceJobKind): ProduceRequest {
|
||||
if (typeof value !== "object" || value === null || Array.isArray(value)) fail("input_json must be an object");
|
||||
onlyKeys(value, ["overlays", "voiceover", "outputLabel", "lane", "requestedBy"], "input_json");
|
||||
const raw = value as Record<string, unknown>;
|
||||
|
||||
const overlaysRaw = raw.overlays === undefined || raw.overlays === null ? [] : raw.overlays;
|
||||
if (!Array.isArray(overlaysRaw) || overlaysRaw.length > MAX_OVERLAYS) fail(`overlays must be an array of at most ${MAX_OVERLAYS} entries`);
|
||||
const overlays = overlaysRaw.map((item, i) => {
|
||||
const field = `overlays[${i}]`;
|
||||
if (typeof item !== "object" || item === null || Array.isArray(item)) fail(`${field} must be an object`);
|
||||
const o = item as Record<string, unknown>;
|
||||
if (o.kind !== "lower_third" && o.kind !== "logo" && o.kind !== "clock") fail(`${field}.kind must be one of lower_third, logo, clock`);
|
||||
// Per-kind allowlist: a clock has neither caption nor asset, a logo has no caption.
|
||||
onlyKeys(item, o.kind === "lower_third" ? ["kind", "text", "position"] : o.kind === "logo" ? ["kind", "logoStorageKey", "position"] : ["kind", "position"], field);
|
||||
return {
|
||||
kind: o.kind,
|
||||
text: o.kind === "lower_third" ? operatorText(o.text, `${field}.text`) : null,
|
||||
logoStorageKey: o.kind === "logo" ? contentKey(o.logoStorageKey, `${field}.logoStorageKey`) : null,
|
||||
position: resolvePosition(o.kind, o.position, `${field}.position`),
|
||||
} satisfies ParsedOverlay;
|
||||
});
|
||||
|
||||
let voiceover: ParsedVoiceover | null = null;
|
||||
if (raw.voiceover !== undefined && raw.voiceover !== null) {
|
||||
// A voiceover only belongs to a mix row; a compose row keeps its source audio.
|
||||
if (kind !== "mix") fail("voiceover requires a 'mix' job");
|
||||
if (typeof raw.voiceover !== "object" || Array.isArray(raw.voiceover)) fail("voiceover must be an object");
|
||||
onlyKeys(raw.voiceover, ["assetId", "gainDb", "duckOriginal"], "voiceover");
|
||||
const v = raw.voiceover as Record<string, unknown>;
|
||||
voiceover = {
|
||||
assetId: uuid(v.assetId, "voiceover.assetId"),
|
||||
gainDb: optional(v.gainDb, "voiceover.gainDb", gainDbChecked),
|
||||
duckOriginal: optional(v.duckOriginal, "voiceover.duckOriginal", bool) ?? false,
|
||||
};
|
||||
}
|
||||
|
||||
const lane = optional(raw.lane, "lane", (l) => (l === "local" || l === "remote" ? l : fail("lane must be 'local' or 'remote'")));
|
||||
const outputLabel = optional(raw.outputLabel, "outputLabel", (l) => {
|
||||
if (typeof l !== "string" || !LABEL.test(l)) fail("outputLabel must be 1-64 letters, digits, spaces, '_', '.', '-'");
|
||||
return l;
|
||||
});
|
||||
return { overlays, voiceover, lane, outputLabel, requestedBy: optional(raw.requestedBy, "requestedBy", (r) => uuid(r, "requestedBy")) };
|
||||
}
|
||||
|
||||
/**
|
||||
* Name for a downloaded object's scratch copy. A `sha256/<64 hex>.<ext>` key names
|
||||
* the file by its content hash; any other valid key is hashed itself — either way
|
||||
* the path is a bare 64-hex filename the request cannot influence.
|
||||
*/
|
||||
export function scratchAssetName(storageKey: string, fallbackExt = "bin"): string {
|
||||
const match = /^sha256\/([0-9a-f]{64})\.([a-z0-9]+)$/.exec(storageKey);
|
||||
if (match) return `${match[1]}.${match[2]}`;
|
||||
// `sha256/` is the content-addressed namespace: a malformed key there is a bug
|
||||
// (or tampering), never a validly-placed asset, so it is refused rather than
|
||||
// silently renamed. Other well-formed keys (e.g. branding renders) are hashed
|
||||
// so the scratch path is still a bare 64-hex filename.
|
||||
if (storageKey.startsWith("sha256/") || !STORAGE_KEY.test(storageKey) || storageKey.includes("..")) {
|
||||
fail("storage key must be content-addressed");
|
||||
}
|
||||
return `${createHash("sha256").update(storageKey).digest("hex")}.${fallbackExt}`;
|
||||
}
|
||||
|
||||
export type AudioBed = "source" | "none";
|
||||
|
||||
export interface ProducePlan {
|
||||
/** The source asset's downloaded bytes, at a scratch path named by content hash. */
|
||||
input: { path: string; durationSec: number | null; audioBed: AudioBed };
|
||||
/** Downloaded logo PNGs, in request order (one input each, consumed by the logo overlays). */
|
||||
logos: { path: string }[];
|
||||
voiceover: { path: string; gainDb: number | null; duck: boolean } | null;
|
||||
overlays: ParsedOverlay[];
|
||||
fontFile: string;
|
||||
/** Scratch path the mp4 is written to. */
|
||||
outputPath: string;
|
||||
}
|
||||
|
||||
/** ffmpeg input indexes: source 0, then each logo, then the voiceover — argv order. */
|
||||
export interface ProduceInputIndex { logos: number[]; voiceover: number | null }
|
||||
|
||||
function audioFormat(): string {
|
||||
return `aformat=sample_fmts=fltp:sample_rates=${PRODUCE_PRESET.sampleRate}:channel_layouts=stereo`;
|
||||
}
|
||||
|
||||
function gain(gainDb: number): string { return `volume=${gainDb}dB`; }
|
||||
|
||||
/**
|
||||
* The `-filter_complex` string: video chain (scale → overlays in request order) and
|
||||
* the audio mix. `min(px,W-tw)`-style expressions are single-quoted because `min`
|
||||
* contains the filter-arg separator `,` — the quotes are on the value, so the
|
||||
* operator never gets to inject one.
|
||||
*/
|
||||
export function buildProduceFilter(plan: ProducePlan, index: ProduceInputIndex): string {
|
||||
const { width: W, height: H, fps } = PRODUCE_PRESET;
|
||||
const chains: string[] = [];
|
||||
chains.push(`[0:v]scale=${W}:${H}:force_original_aspect_ratio=decrease:force_divisible_by=2,pad=${W}:${H}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1,fps=${fps}[base]`);
|
||||
|
||||
// One chain step per overlay, in request order, so the request decides the z-order.
|
||||
let cur = "base";
|
||||
let logoInput = 0;
|
||||
plan.overlays.forEach((overlay, i) => {
|
||||
const next = `ov${i}`;
|
||||
if (overlay.kind === "logo") {
|
||||
const at = index.logos[logoInput];
|
||||
if (at === undefined) fail(`logo overlay ${i} has no downloaded logo`);
|
||||
// One decoded frame held by `eof_action=repeat` (the compositor's asset rule):
|
||||
// no `-loop 1`, so nothing is framesync'd against the main input on a bounded job.
|
||||
chains.push(`[${cur}][${at}:v]overlay=x='${xExpression(overlay.position, "w")}':y='${yExpression(overlay.position, "h")}':eof_action=repeat[${next}]`);
|
||||
cur = next;
|
||||
logoInput += 1;
|
||||
return;
|
||||
}
|
||||
// The clock's text is the builder's own constant and needs ffmpeg's expansion;
|
||||
// operator captions are inert literals.
|
||||
const text = overlay.kind === "clock" ? `'${CLOCK_TEXT}'` : drawtextQuoted(overlay.text ?? fail("lower_third overlay has no text"));
|
||||
const expansion = overlay.kind === "clock" ? "normal" : "none";
|
||||
chains.push(`[${cur}]drawtext=fontfile=${plan.fontFile}:text=${text}:fontcolor=white:fontsize=34:x='${xExpression(overlay.position)}':y='${yExpression(overlay.position)}':box=1:boxcolor=black@0.55:boxborderw=16:expansion=${expansion}[${next}]`);
|
||||
cur = next;
|
||||
});
|
||||
if (logoInput !== plan.logos.length) fail("downloaded logos and logo overlays disagree");
|
||||
chains.push(`[${cur}]format=yuv420p[vout]`);
|
||||
|
||||
// Audio: the source bed (or silence) plus the voiceover, amix'd without normalisation
|
||||
// so the requested gain is the gain; ducking runs the voiceover as a sidechain key
|
||||
// on the bed. `duration=first` ends with the source bed; a synthetic bed is
|
||||
// unbounded, so it mixes to `longest` and only the source's `-t` can end the encode.
|
||||
const vo = plan.voiceover;
|
||||
const bed = plan.input.audioBed === "source" ? "[0:a]" : `anullsrc=r=${PRODUCE_PRESET.sampleRate}:cl=stereo`;
|
||||
|
||||
if (vo === null) {
|
||||
chains.push(`${bed}${audioFormat()}[aout]`);
|
||||
return chains.join(";");
|
||||
}
|
||||
const voIndex = index.voiceover ?? fail("voiceover input index required");
|
||||
const voIn = `[${voIndex}:a]`;
|
||||
const amixDuration = plan.input.audioBed === "source" ? "first" : "longest";
|
||||
if (vo.duck) {
|
||||
// asplit feeds the same voiceover to the compressor key and to the mix.
|
||||
chains.push(`${voIn}${audioFormat()},asplit=2[voMix][voKey]`);
|
||||
chains.push(`${bed}${audioFormat()}[bedRaw]`);
|
||||
chains.push(`[bedRaw][voKey]sidechaincompress=threshold=0.03:ratio=8:attack=20:release=300:makeup=1:link=average[duck]`);
|
||||
if (vo.gainDb !== null && vo.gainDb !== 0) chains.push(`[voMix]${gain(vo.gainDb)}[voGain]`);
|
||||
const mixIn = vo.gainDb !== null && vo.gainDb !== 0 ? "voGain" : "voMix";
|
||||
chains.push(`[duck][${mixIn}]amix=inputs=2:duration=${amixDuration}:dropout_transition=2:normalize=0[aout]`);
|
||||
} else {
|
||||
chains.push(`${bed}${audioFormat()}[bedFmt]`);
|
||||
chains.push(`${voIn}${audioFormat()}${vo.gainDb === null || vo.gainDb === 0 ? "" : `,${gain(vo.gainDb)}`}[voGain]`);
|
||||
chains.push(`[bedFmt][voGain]amix=inputs=2:duration=${amixDuration}:dropout_transition=2:normalize=0[aout]`);
|
||||
}
|
||||
// A synthetic bed is unbounded, so only the source's `-t` can end the encode: it must be known.
|
||||
if (plan.input.audioBed === "none" && plan.input.durationSec === null) fail("a synthetic audio bed needs the source duration to bound the mix");
|
||||
return chains.join(";");
|
||||
}
|
||||
|
||||
function scratchPath(path: string, field: string): string {
|
||||
if (!SCRATCH_FILE.test(path) || path.split("/").includes("..")) fail(`${field} must be an absolute scratch path to a downloaded <sha256>.<ext> file`);
|
||||
return path;
|
||||
}
|
||||
|
||||
function validatePlan(plan: ProducePlan): void {
|
||||
scratchPath(plan.input.path, "input.path");
|
||||
plan.logos.forEach((logo, i) => scratchPath(logo.path, `logos[${i}].path`));
|
||||
if (plan.voiceover) scratchPath(plan.voiceover.path, "voiceover.path");
|
||||
if (plan.outputPath === plan.input.path) fail("outputPath must differ from the input path");
|
||||
scratchPath(plan.outputPath, "outputPath");
|
||||
if (!FONT_FILE.test(plan.fontFile) || plan.fontFile.split("/").includes("..")) fail("fontFile must be an absolute path to a .ttf");
|
||||
if (plan.input.durationSec !== null && !(plan.input.durationSec > 0 && plan.input.durationSec <= 86_400)) fail("input.durationSec must be in (0..86400]");
|
||||
if (plan.input.audioBed !== "source" && plan.input.audioBed !== "none") fail("input.audioBed");
|
||||
if (plan.overlays.length > MAX_OVERLAYS) fail(`at most ${MAX_OVERLAYS} overlays`);
|
||||
const logos = plan.overlays.filter((o) => o.kind === "logo").length;
|
||||
if (logos !== plan.logos.length) fail("every logo overlay needs exactly one downloaded logo");
|
||||
plan.overlays.forEach((o, i) => {
|
||||
if (o.kind === "lower_third" && (o.text === null || o.text.length === 0 || o.text.length > MAX_TEXT_CHARS)) fail(`overlays[${i}].text length`);
|
||||
if (o.kind === "logo" && o.logoStorageKey === null) fail(`overlays[${i}].logoStorageKey`);
|
||||
});
|
||||
if (plan.voiceover && plan.voiceover.gainDb !== null && (plan.voiceover.gainDb < GAIN_DB_MIN || plan.voiceover.gainDb > GAIN_DB_MAX)) fail("voiceover.gainDb out of range");
|
||||
}
|
||||
|
||||
/** Full produce argv: one source input (plus logos and voiceover), one graph, one mp4. */
|
||||
export function buildProduceArgs(plan: ProducePlan): string[] {
|
||||
validatePlan(plan);
|
||||
const { width, height, fps, videoBitrateKbps, audioBitrateKbps } = PRODUCE_PRESET;
|
||||
const args = ["-hide_banner", "-nostdin", "-loglevel", "error", "-y"];
|
||||
args.push("-i", `file:${plan.input.path}`);
|
||||
const index: ProduceInputIndex = { logos: [], voiceover: null };
|
||||
let next = 1;
|
||||
for (const logo of plan.logos) {
|
||||
args.push("-framerate", String(fps), "-i", `file:${logo.path}`);
|
||||
index.logos.push(next++);
|
||||
}
|
||||
if (plan.voiceover) {
|
||||
args.push("-i", `file:${plan.voiceover.path}`);
|
||||
index.voiceover = next++;
|
||||
}
|
||||
args.push("-filter_complex", buildProduceFilter(plan, index));
|
||||
|
||||
// A known source duration bounds the whole encode (and a synthetic audio bed).
|
||||
if (plan.input.durationSec !== null) args.push("-t", plan.input.durationSec.toFixed(3));
|
||||
|
||||
args.push(
|
||||
"-map", "[vout]", "-map", "[aout]",
|
||||
"-c:v", "libx264", "-preset", "veryfast", "-profile:v", "high", "-pix_fmt", "yuv420p",
|
||||
"-b:v", `${videoBitrateKbps}k`, "-maxrate", `${videoBitrateKbps}k`, "-bufsize", `${videoBitrateKbps * 2}k`,
|
||||
"-g", String(fps * 2), "-keyint_min", String(fps * 2), "-sc_threshold", "0",
|
||||
"-c:a", "aac", "-b:a", `${audioBitrateKbps}k`, "-ar", String(PRODUCE_PRESET.sampleRate),
|
||||
"-movflags", "+faststart",
|
||||
"-metadata", `width=${width}`, "-metadata", `height=${height}`,
|
||||
"-f", "mp4", `file:${plan.outputPath}`,
|
||||
);
|
||||
return args;
|
||||
}
|
||||
|
||||
/** Failure reasons the runner reports; none of them carries operator text or a path. */
|
||||
export type ProduceFailure =
|
||||
| "invalid request"
|
||||
| "asset unavailable"
|
||||
| "download failed"
|
||||
| "encode timed out"
|
||||
| "encode failed"
|
||||
| "publish failed"
|
||||
| "job store failed";
|
||||
|
||||
export function produceError(failure: ProduceFailure, detail?: string): Error {
|
||||
const error = new Error(`produce: ${failure}`);
|
||||
error.name = failure;
|
||||
if (detail !== undefined) (error as Error & { detail?: string }).detail = detail;
|
||||
return error;
|
||||
}
|
||||
245
apps/media-worker/src/producer/pg-store.ts
Normal file
245
apps/media-worker/src/producer/pg-store.ts
Normal file
|
|
@ -0,0 +1,245 @@
|
|||
/**
|
||||
* Postgres produce job store: the poll/claim/complete half of the produce worker.
|
||||
*
|
||||
* - Claim: `UPDATE ... SET state='transcoding' WHERE id IN (SELECT id ... FOR UPDATE
|
||||
* SKIP LOCKED) RETURNING` — one row, one statement, lane-filtered, so concurrent
|
||||
* workers never race and a claimed row is invisible to everyone else until its
|
||||
* owner finishes it. The `queued → transcoding` bump inside SKIP LOCKED is the
|
||||
* lease: there is no separate lease column to renew, because a claimed row is
|
||||
* owned until its owner writes complete/failed.
|
||||
* - Every write happens inside the job's tenant transaction (`app.workspace_id`),
|
||||
* because both tables are FORCE ROW LEVEL SECURITY with a `tenant_isolation`
|
||||
* policy (009). The claim reads the workspace off the row first, then re-claims
|
||||
* inside that tenant's scope.
|
||||
* - A row whose `input_json` or source asset fails re-validation is FAILED by THE
|
||||
* CLAIM ITSELF (never thrown out of `claim`), so an undecodable row can never sit
|
||||
* in `transcoding` forever holding its lease.
|
||||
*/
|
||||
import type { Pool, PoolClient } from "pg";
|
||||
import type { Lane, ProduceJobKind } from "./command-builder";
|
||||
import { parseProduceRequest } from "./command-builder";
|
||||
import type { ClaimedProduceJob, ProduceJobStore, ProduceOutput, ProduceSourceAsset } from "./store";
|
||||
|
||||
const KINDS: readonly ProduceJobKind[] = ["compose", "mix"];
|
||||
|
||||
/**
|
||||
* Lane → which rows this worker may claim (039 lanes: default worker = local or NULL,
|
||||
* i.e. anything but an explicit `lane:'remote'`; a remote worker takes only remote).
|
||||
*/
|
||||
export function lanePredicate(lane: Lane): { laneSql: string; params: [boolean] } {
|
||||
// $1 binds the boolean flag: a local/default worker skips remote rows, a remote
|
||||
// worker claims ONLY remote rows. The predicate never matches a local row to a
|
||||
// remote worker, so nothing is locked and then released.
|
||||
return {
|
||||
laneSql: lane === "remote"
|
||||
? "(j.input_json->>'lane' = 'remote')"
|
||||
: "(j.input_json->>'lane' IS DISTINCT FROM 'remote')",
|
||||
params: [lane === "remote"],
|
||||
};
|
||||
}
|
||||
|
||||
function uuid(value: unknown, field: string): string {
|
||||
const text = typeof value === "string" ? value : undefined;
|
||||
if (text === undefined || !/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(text)) {
|
||||
throw new Error(`produce job store: row has an invalid ${field}`);
|
||||
}
|
||||
return text;
|
||||
}
|
||||
|
||||
function nullableNumber(value: unknown): number | null {
|
||||
if (value === null || value === undefined) return null;
|
||||
const n = typeof value === "number" ? value : Number(value);
|
||||
return Number.isFinite(n) ? n : null;
|
||||
}
|
||||
|
||||
function sourceFromRow(row: Record<string, unknown>): ProduceSourceAsset {
|
||||
return {
|
||||
id: uuid(row.id, "video_asset id"),
|
||||
storageKey: String(row.storage_key ?? ""),
|
||||
mimeType: String(row.mime_type ?? ""),
|
||||
sizeBytes: Number(row.size_bytes ?? 0),
|
||||
width: nullableNumber(row.width),
|
||||
height: nullableNumber(row.height),
|
||||
durationMs: row.duration_ms === null || row.duration_ms === undefined ? null : Math.round(Number(row.duration_ms)),
|
||||
frameRate: nullableNumber(row.frame_rate),
|
||||
// Probed "no audio track" (009 stores an empty codec) → a silent bed, so the
|
||||
// output always has an audio stream.
|
||||
hasAudio: row.audio_codec === null || row.audio_codec === undefined ? null : String(row.audio_codec) !== "",
|
||||
};
|
||||
}
|
||||
|
||||
/** The store's own reads are RLS-scoped to the job's workspace (009 tenant_isolation). */
|
||||
export class PostgresProduceJobStore implements ProduceJobStore {
|
||||
constructor(private readonly pool: Pool, private readonly now: () => Date = () => new Date()) {}
|
||||
|
||||
private async tenant<T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>): Promise<T> {
|
||||
const client = await this.pool.connect();
|
||||
try {
|
||||
await client.query("BEGIN");
|
||||
await client.query("SELECT set_config('app.workspace_id', $1, true)", [workspaceId]);
|
||||
const result = await fn(client);
|
||||
await client.query("COMMIT");
|
||||
return result;
|
||||
} catch (error) {
|
||||
await client.query("ROLLBACK").catch(() => undefined);
|
||||
throw error;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
}
|
||||
|
||||
async claim(lane: Lane): Promise<ClaimedProduceJob | null> {
|
||||
// Lane filter in the sub-select, so a worker never locks a row it would have to
|
||||
// release: a remote worker only ever locks remote rows, the default worker only
|
||||
// ever locks non-remote rows.
|
||||
const { laneSql, params } = lanePredicate(lane);
|
||||
const claimed = await this.pool.query<{ id: string; workspace_id: string }>(
|
||||
`UPDATE video_processing_jobs j
|
||||
SET state = 'transcoding', progress_percent = 5, error = NULL, updated_at = now()
|
||||
WHERE j.id IN (
|
||||
SELECT id FROM video_processing_jobs
|
||||
WHERE kind IN ('compose', 'mix')
|
||||
AND state = 'queued'
|
||||
AND ${laneSql}
|
||||
ORDER BY created_at, id
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
)
|
||||
RETURNING j.id, j.workspace_id`,
|
||||
params,
|
||||
);
|
||||
if (claimed.rowCount === 0) return null;
|
||||
const row = claimed.rows[0]!;
|
||||
const workspaceId = uuid(row.workspace_id, "workspace id");
|
||||
|
||||
return this.tenant<ClaimedProduceJob | null>(workspaceId, async (client) => {
|
||||
// Re-claim inside the tenant's RLS scope (the pool-level row above is outside
|
||||
// it; claiming again is the cheap way to stay correct under FORCE RLS).
|
||||
const mine = await client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'transcoding', progress_percent = 5, error = NULL, updated_at = now()
|
||||
WHERE id = $1 AND state IN ('queued', 'transcoding')
|
||||
RETURNING id, kind, input_json, video_asset_id`,
|
||||
[row.id],
|
||||
);
|
||||
if (mine.rowCount === 0) return null;
|
||||
const job = mine.rows[0]! as Record<string, unknown>;
|
||||
const kind = job.kind === "mix" ? "mix" : job.kind === "compose" ? "compose" : null;
|
||||
let request = null;
|
||||
if (kind !== null) {
|
||||
try {
|
||||
request = parseProduceRequest(job.input_json, kind);
|
||||
} catch {
|
||||
request = null;
|
||||
}
|
||||
}
|
||||
if (kind === null || request === null) {
|
||||
// A row the worker cannot honour must FAIL here — never strand in `transcoding`.
|
||||
// Written inside this transaction and COMMITTED by returning null (not thrown:
|
||||
// a throw would roll the failure back and re-strand the row).
|
||||
await this.failInvalid(client, job, kind === null ? "produce job store: unknown kind" : "produce invalid request");
|
||||
return null;
|
||||
}
|
||||
|
||||
const assets = await client.query(
|
||||
`SELECT id, storage_key, mime_type, size_bytes, width, height, duration_ms, frame_rate, audio_codec
|
||||
FROM video_assets WHERE id = $1`,
|
||||
[job.video_asset_id],
|
||||
);
|
||||
if (assets.rowCount === 0) {
|
||||
await this.failInvalid(client, job, "produce asset unavailable");
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
jobId: uuid(job.id, "job id"),
|
||||
workspaceId,
|
||||
kind,
|
||||
request,
|
||||
source: sourceFromRow(assets.rows[0] as Record<string, unknown>),
|
||||
claimedAt: this.now().toISOString(),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Fail a just-claimed row whose payload cannot be honoured, inside the claim's
|
||||
* own transaction so the failure commits with it. Callers then return `null`:
|
||||
* the row is done (failed), there is simply nothing for this tick to encode.
|
||||
*/
|
||||
private async failInvalid(client: PoolClient, job: Record<string, unknown>, reason: string): Promise<void> {
|
||||
await client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'failed', progress_percent = 100, error = $2, updated_at = now()
|
||||
WHERE id = $1 AND state = 'transcoding'`,
|
||||
[job.id, reason],
|
||||
);
|
||||
}
|
||||
|
||||
async progress(jobId: string, percent: number): Promise<void> {
|
||||
await this.tenant(await this.workspaceOf(jobId), async (client) => {
|
||||
await client.query(
|
||||
`UPDATE video_processing_jobs SET progress_percent = $2, updated_at = now()
|
||||
WHERE id = $1 AND state = 'transcoding'`,
|
||||
[jobId, percent],
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async complete(jobId: string, output: ProduceOutput, derivedFrom: string): Promise<string> {
|
||||
const workspaceId = await this.workspaceOf(jobId);
|
||||
return this.tenant(workspaceId, async (client) => {
|
||||
const inserted = await client.query<{ id: string }>(
|
||||
`INSERT INTO video_assets
|
||||
(workspace_id, storage_key, mime_type, size_bytes, width, height, duration_ms, frame_rate, codec, audio_codec, derived_from)
|
||||
VALUES ($1, $2, 'video/mp4', $3, $4, $5, $6, $7, 'h264', 'aac', $8)
|
||||
RETURNING id`,
|
||||
[workspaceId, output.storageKey, output.sizeBytes, output.width, output.height, output.durationMs, output.frameRate, derivedFrom],
|
||||
);
|
||||
const producedAssetId = inserted.rows[0]!.id;
|
||||
const updated = await client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'complete', progress_percent = 100, output_key = $2,
|
||||
produced_asset_id = $3, error = NULL, updated_at = now()
|
||||
WHERE id = $1 AND state = 'transcoding'
|
||||
RETURNING id`,
|
||||
[jobId, output.storageKey, producedAssetId],
|
||||
);
|
||||
if (updated.rowCount === 0) throw new Error(`produce job store: job ${jobId} lost its claim`);
|
||||
return producedAssetId;
|
||||
});
|
||||
}
|
||||
|
||||
async fail(jobId: string, reason: string): Promise<void> {
|
||||
await this.tenant(await this.workspaceOf(jobId), async (client) => {
|
||||
await client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'failed', error = $2, updated_at = now()
|
||||
WHERE id = $1 AND state IN ('queued', 'transcoding')`,
|
||||
[jobId, reason],
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async asset(workspaceId: string, assetId: string): Promise<{ storageKey: string; mimeType: string } | null> {
|
||||
return this.tenant(workspaceId, async (client) => {
|
||||
const rows = await client.query<{ storage_key: string; mime_type: string }>(
|
||||
`SELECT storage_key, mime_type FROM video_assets WHERE id = $1`,
|
||||
[assetId],
|
||||
);
|
||||
if (rows.rowCount === 0) return null;
|
||||
return { storageKey: rows.rows[0]!.storage_key, mimeType: rows.rows[0]!.mime_type };
|
||||
});
|
||||
}
|
||||
|
||||
/** A claimed job's workspace, for the tenant-scoped follow-up writes. */
|
||||
private async workspaceOf(jobId: string): Promise<string> {
|
||||
const rows = await this.pool.query<{ workspace_id: string }>(
|
||||
`SELECT workspace_id FROM video_processing_jobs WHERE id = $1`,
|
||||
[jobId],
|
||||
);
|
||||
if (rows.rowCount === 0) throw new Error(`produce job store: job ${jobId} not found`);
|
||||
return uuid(rows.rows[0]!.workspace_id, "workspace id");
|
||||
}
|
||||
}
|
||||
|
||||
export const PRODUCE_JOB_KINDS = KINDS;
|
||||
196
apps/media-worker/src/producer/runner.test.ts
Normal file
196
apps/media-worker/src/producer/runner.test.ts
Normal file
|
|
@ -0,0 +1,196 @@
|
|||
/**
|
||||
* 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`; }
|
||||
|
||||
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" };
|
||||
},
|
||||
};
|
||||
const storage: ProduceObjectStorage = {
|
||||
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),
|
||||
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("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.after(() => { rmSync(SCRATCH, { recursive: true, force: true }); });
|
||||
276
apps/media-worker/src/producer/runner.ts
Normal file
276
apps/media-worker/src/producer/runner.ts
Normal file
|
|
@ -0,0 +1,276 @@
|
|||
/**
|
||||
* Produce runner: the compose/mix job executor.
|
||||
*
|
||||
* Poll loop in the supervisor's shape (one claim per tick, retry with backoff on
|
||||
* store errors, stop flag honoured between ticks), one ffmpeg per job through
|
||||
* `spawnManaged` (argv only, no shell, minimal env, redacted truncated stderr),
|
||||
* scratch per job (mkdtemp, removed in `finally`), and a hard timeout that
|
||||
* terminates the child SIGTERM → SIGKILL before it fails the job.
|
||||
*
|
||||
* Per job: download the source bytes (plus each logo overlay's PNG and the
|
||||
* voiceover) to scratch → build one argv from the validated request → encode
|
||||
* 720p30 h264+aac mp4 → putContent → insert the derived video_assets row →
|
||||
* complete the job. Any failure marks the job failed with a bounded, redacted
|
||||
* reason.
|
||||
*/
|
||||
import { spawn } from "node:child_process";
|
||||
import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { spawnManaged, type StderrRedactor } from "../relay/ffmpeg";
|
||||
import {
|
||||
buildProduceArgs, produceError, PRODUCE_PRESET, scratchAssetName,
|
||||
type Lane, type ProducePlan, type ProduceFailure,
|
||||
} from "./command-builder";
|
||||
import type { ClaimedProduceJob, ProduceJobStore, ProduceObjectStorage } from "./store";
|
||||
|
||||
export interface ProduceRunnerDeps {
|
||||
jobs: ProduceJobStore;
|
||||
storage: ProduceObjectStorage;
|
||||
/** Font for lower thirds; an absolute path to a .ttf the sandbox can read. */
|
||||
fontFile: string;
|
||||
scratchRoot: string;
|
||||
lane: Lane;
|
||||
redactor: StderrRedactor;
|
||||
onStderr(message: string): void;
|
||||
onEvent?(event: { type: "claimed" | "complete" | "failed" | "idle" | "store-error"; jobId?: string; detail?: string }): void;
|
||||
/** Milliseconds between empty polls. */
|
||||
pollIntervalMs?: number;
|
||||
/** Hard cap on one ffmpeg run. */
|
||||
encodeTimeoutMs?: number;
|
||||
/** Cap on one download. */
|
||||
downloadTimeoutMs?: number;
|
||||
/** Injection for tests; `which ffmpeg`. */
|
||||
ffmpegBinary?: string;
|
||||
ffprobeBinary?: string;
|
||||
scheduler?: { setTimeout(ms: number, action: () => void): { cancel(): void } };
|
||||
}
|
||||
|
||||
const DEFAULTS = { pollIntervalMs: 2_000, encodeTimeoutMs: 15 * 60_000, downloadTimeoutMs: 120_000 };
|
||||
const defaultScheduler = {
|
||||
setTimeout(ms: number, action: () => void): { cancel(): void } {
|
||||
const timer = setTimeout(action, ms);
|
||||
return { cancel: () => clearTimeout(timer) };
|
||||
},
|
||||
};
|
||||
|
||||
/** ffmpeg's own progress line is not parsed; these are the checkpoints a job walks through. */
|
||||
const STAGES: ReadonlyArray<[label: string, percent: number]> = [["claimed", 5], ["downloaded", 20], ["encoded", 80], ["published", 95]];
|
||||
|
||||
export class ProduceRunner {
|
||||
private stopped = false;
|
||||
private timer: { cancel(): void } | null = null;
|
||||
private readonly pollIntervalMs: number;
|
||||
private readonly encodeTimeoutMs: number;
|
||||
private readonly downloadTimeoutMs: number;
|
||||
private readonly scheduler: NonNullable<ProduceRunnerDeps["scheduler"]>;
|
||||
private running: Promise<unknown> | null = null;
|
||||
|
||||
constructor(private readonly deps: ProduceRunnerDeps) {
|
||||
this.pollIntervalMs = deps.pollIntervalMs ?? DEFAULTS.pollIntervalMs;
|
||||
this.encodeTimeoutMs = deps.encodeTimeoutMs ?? DEFAULTS.encodeTimeoutMs;
|
||||
this.downloadTimeoutMs = deps.downloadTimeoutMs ?? DEFAULTS.downloadTimeoutMs;
|
||||
this.scheduler = deps.scheduler ?? defaultScheduler;
|
||||
}
|
||||
|
||||
/** Poll until `stop()`. Each tick claims at most one job and runs it to completion. */
|
||||
start(): void {
|
||||
this.stopped = false;
|
||||
void this.tick();
|
||||
}
|
||||
|
||||
private scheduleNext(): void {
|
||||
if (this.stopped) return;
|
||||
this.timer = this.scheduler.setTimeout(this.pollIntervalMs, () => { this.timer = null; void this.tick(); });
|
||||
}
|
||||
|
||||
private async tick(): Promise<void> {
|
||||
if (this.stopped) return;
|
||||
try {
|
||||
const job = await this.deps.jobs.claim(this.deps.lane);
|
||||
if (job === null) {
|
||||
this.deps.onEvent?.({ type: "idle" });
|
||||
} else {
|
||||
const run = this.run(job).finally(() => { if (this.running === run) this.running = null; });
|
||||
this.running = run;
|
||||
await run;
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
// A store/poll error must never kill the loop (the supervisor's DEGRADED path).
|
||||
this.deps.onEvent?.({ type: "store-error", detail: error instanceof Error ? error.name : "unknown error" });
|
||||
}
|
||||
this.scheduleNext();
|
||||
}
|
||||
|
||||
/** Claim one job now and run it; the loop uses it, tests call it directly. */
|
||||
async pollOnce(): Promise<boolean> {
|
||||
const job = await this.deps.jobs.claim(this.deps.lane);
|
||||
if (job === null) return false;
|
||||
await this.run(job);
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Stop polling; an in-flight job finishes first (its DB row is its lease). */
|
||||
async stop(): Promise<void> {
|
||||
this.stopped = true;
|
||||
this.timer?.cancel();
|
||||
this.timer = null;
|
||||
while (this.running !== null) await Promise.resolve(this.running).catch(() => undefined);
|
||||
}
|
||||
|
||||
private async run(job: ClaimedProduceJob): Promise<void> {
|
||||
this.deps.onEvent?.({ type: "claimed", jobId: job.jobId });
|
||||
const directory = await mkdtemp(join(this.deps.scratchRoot, "produce-"));
|
||||
try {
|
||||
const plan = await this.prepare(job, directory);
|
||||
await this.deps.jobs.progress(job.jobId, STAGES[1]![1]);
|
||||
await this.encode(job, plan);
|
||||
await this.deps.jobs.progress(job.jobId, STAGES[2]![1]);
|
||||
const output = await this.publish(job, plan);
|
||||
await this.deps.jobs.progress(job.jobId, STAGES[3]![1]);
|
||||
const producedAssetId = await this.deps.jobs.complete(job.jobId, output, job.source.id);
|
||||
this.deps.onEvent?.({ type: "complete", jobId: job.jobId, detail: `${output.storageKey} → asset ${producedAssetId}` });
|
||||
} catch (error: unknown) {
|
||||
const reason = this.reason(error);
|
||||
this.deps.onStderr(`produce job ${job.jobId}: ${reason}`);
|
||||
await this.deps.jobs.fail(job.jobId, reason).catch((failError: unknown) => {
|
||||
this.deps.onEvent?.({ type: "store-error", jobId: job.jobId, detail: failError instanceof Error ? failError.name : "unknown error" });
|
||||
});
|
||||
this.deps.onEvent?.({ type: "failed", jobId: job.jobId, detail: reason });
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Download every input the validated request names, and build the plan: the
|
||||
* source bytes plus one file per logo overlay (keyed by content hash) and the
|
||||
* voiceover, all straight into this job's scratch dir.
|
||||
*/
|
||||
private async prepare(job: ClaimedProduceJob, directory: string): Promise<ProducePlan> {
|
||||
const source = await this.download(job.source.storageKey, directory);
|
||||
const logos: { path: string }[] = [];
|
||||
for (const overlay of job.request.overlays) {
|
||||
if (overlay.kind !== "logo") continue;
|
||||
// A logo overlay names a pre-rendered PNG by its content key (039), so it is
|
||||
// downloaded like the source bytes.
|
||||
logos.push({ path: await this.download(overlay.logoStorageKey ?? "", directory) });
|
||||
}
|
||||
let voiceover: ProducePlan["voiceover"] = null;
|
||||
if (job.request.voiceover !== null) {
|
||||
const asset = await this.deps.jobs.asset(job.workspaceId, job.request.voiceover.assetId);
|
||||
if (asset === null) throw produceError("asset unavailable", "voiceover");
|
||||
voiceover = {
|
||||
path: await this.download(asset.storageKey, directory),
|
||||
gainDb: job.request.voiceover.gainDb,
|
||||
duck: job.request.voiceover.duckOriginal,
|
||||
};
|
||||
}
|
||||
const durationSec = job.source.durationMs === null ? null : job.source.durationMs / 1000;
|
||||
// A probed asset without an audio track gets a silent bed, so the output always has audio.
|
||||
const audioBed = job.source.hasAudio === false ? "none" : "source";
|
||||
return {
|
||||
input: { path: source, durationSec, audioBed },
|
||||
logos,
|
||||
voiceover,
|
||||
overlays: job.request.overlays,
|
||||
fontFile: this.deps.fontFile,
|
||||
outputPath: join(directory, "output.mp4"),
|
||||
};
|
||||
}
|
||||
|
||||
/** One object → one scratch file named by its content hash (paths never carry ids). */
|
||||
private async download(storageKey: string, directory: string): Promise<string> {
|
||||
const name = scratchAssetName(storageKey);
|
||||
let bytes: Buffer;
|
||||
try {
|
||||
bytes = await withTimeout(this.deps.storage.getObject(storageKey), this.downloadTimeoutMs, "download failed");
|
||||
} catch (error: unknown) {
|
||||
// A timeout already carries the failure name; anything else is a storage failure.
|
||||
if (error instanceof Error && error.name === "download failed") throw error;
|
||||
throw produceError("download failed");
|
||||
}
|
||||
const path = join(directory, name);
|
||||
await writeFile(path, bytes, { mode: 0o600 });
|
||||
return path;
|
||||
}
|
||||
|
||||
private async encode(job: ClaimedProduceJob, plan: ProducePlan): Promise<void> {
|
||||
const args = buildProduceArgs(plan);
|
||||
const redactor = this.deps.redactor;
|
||||
const managed = spawnManaged(this.deps.ffmpegBinary ?? "ffmpeg", args, ["ignore", "ignore", "pipe"], redactor,
|
||||
(message) => this.deps.onStderr(`job ${job.jobId}: ${message}`));
|
||||
const timer = this.scheduler.setTimeout(this.encodeTimeoutMs, () => {
|
||||
void managed.terminate(2_000).then(() => undefined, () => undefined);
|
||||
});
|
||||
try {
|
||||
const exited = await managed.exited;
|
||||
if (this.stopped) throw produceError("encode failed", "worker stopped");
|
||||
if (exited.code !== 0) throw produceError(exited.code === null ? "encode timed out" : "encode failed", `code=${exited.code} signal=${exited.signal ?? "-"}`);
|
||||
const info = await stat(plan.outputPath).catch(() => null);
|
||||
if (info === null || info.size === 0) throw produceError("encode failed", "no output written");
|
||||
} finally {
|
||||
timer.cancel();
|
||||
// Belt and braces: a still-live child (the timeout raced the exit path) is terminated.
|
||||
await managed.terminate(0).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
private async publish(job: ClaimedProduceJob, plan: ProducePlan): Promise<{ storageKey: string; sizeBytes: number; durationMs: number | null; width: number; height: number; frameRate: number }> {
|
||||
let body: Buffer;
|
||||
try {
|
||||
body = await readFile(plan.outputPath);
|
||||
} catch {
|
||||
throw produceError("encode failed", "output unreadable");
|
||||
}
|
||||
let key: string;
|
||||
try {
|
||||
const stored = await withTimeout(this.deps.storage.putContent(body, "video/mp4"), this.downloadTimeoutMs, "publish failed");
|
||||
key = stored.key;
|
||||
} catch {
|
||||
throw produceError("publish failed");
|
||||
}
|
||||
const durationMs = await this.probeDurationMs(plan.outputPath);
|
||||
return {
|
||||
storageKey: key,
|
||||
sizeBytes: body.byteLength,
|
||||
durationMs,
|
||||
width: PRODUCE_PRESET.width,
|
||||
height: PRODUCE_PRESET.height,
|
||||
frameRate: PRODUCE_PRESET.fps,
|
||||
};
|
||||
}
|
||||
|
||||
/** Output duration from ffprobe (-show_entries), so the derived row is a real library entry. */
|
||||
private async probeDurationMs(path: string): Promise<number | null> {
|
||||
const result = await new Promise<{ status: number | null; stdout: string }>((resolve) => {
|
||||
const child = spawn(this.deps.ffprobeBinary ?? "ffprobe",
|
||||
["-hide_banner", "-loglevel", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1", `file:${path}`],
|
||||
{ shell: false });
|
||||
let stdout = "";
|
||||
child.stdout?.on("data", (chunk: Buffer) => { stdout += chunk.toString("utf8").slice(0, 1024); });
|
||||
child.once("error", () => resolve({ status: null, stdout }));
|
||||
child.once("exit", (code) => resolve({ status: code, stdout }));
|
||||
});
|
||||
const seconds = /duration=([0-9.]+)/.exec(result.stdout)?.[1];
|
||||
const parsed = seconds === undefined ? NaN : Number(seconds);
|
||||
if (!Number.isFinite(parsed) || parsed <= 0) return null;
|
||||
return Math.round(parsed * 1000);
|
||||
}
|
||||
|
||||
/** Bounded, secret-free failure reason: a known failure name, never stderr or a path. */
|
||||
private reason(error: unknown): string {
|
||||
const name = error instanceof Error ? error.name : "";
|
||||
const known = ["invalid request", "asset unavailable", "download failed", "encode timed out", "encode failed", "publish failed", "job store failed"] as const;
|
||||
if ((known as readonly string[]).includes(name)) return `produce ${name}`;
|
||||
return `produce failed: ${this.deps.redactor.redact(String(error)).slice(0, 200)}`;
|
||||
}
|
||||
}
|
||||
|
||||
function withTimeout<T>(promise: Promise<T>, ms: number, name: ProduceFailure): Promise<T> {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timer = setTimeout(() => reject(produceError(name)), ms);
|
||||
promise.then((value) => { clearTimeout(timer); resolve(value); },
|
||||
(error: unknown) => { clearTimeout(timer); reject(error); });
|
||||
});
|
||||
}
|
||||
149
apps/media-worker/src/producer/s3-storage.ts
Normal file
149
apps/media-worker/src/producer/s3-storage.ts
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
/**
|
||||
* Worker-side storage adapter: the same envelope encryption the API's
|
||||
* `apps/api/src/storage/storage.ts` uses (SPENC1 frames, KEK from S3_KEK), so a
|
||||
* `getObject` here opens bytes a `putContent` there sealed — and vice versa.
|
||||
*
|
||||
* The worker must never reimplement the API's policy (dedup HEAD, presign, bucket
|
||||
* provisioning); it needs only the two operations a produce job uses: download
|
||||
* source/logo/voiceover bytes, publish the produced mp4. Reusing the seal/open
|
||||
* frame format (rather than routing through the API) keeps the worker off the
|
||||
* API's request path, which is the point of a queue.
|
||||
*/
|
||||
import { createCipheriv, createDecipheriv, createHash, randomBytes } from "node:crypto";
|
||||
import { GetObjectCommand, PutObjectCommand, S3Client } from "@aws-sdk/client-s3";
|
||||
import type { ProduceObjectStorage } from "./store";
|
||||
|
||||
const FRAME_MAGIC = Buffer.from("SPENC1");
|
||||
|
||||
export interface WorkerStorageConfig {
|
||||
endpoint?: string;
|
||||
region?: string;
|
||||
accessKeyId?: string;
|
||||
secretAccessKey?: string;
|
||||
bucket?: string;
|
||||
forcePathStyle?: boolean;
|
||||
/** 32-byte KEK; when omitted, the dev mirror of the API's loadMediaKek(). */
|
||||
mediaKek?: Buffer;
|
||||
}
|
||||
|
||||
function loadMediaKek(): Buffer {
|
||||
const env = process.env.S3_KEK;
|
||||
if (env !== undefined && env !== "") {
|
||||
const kek = Buffer.from(env, "base64");
|
||||
if (kek.length !== 32) throw new Error("S3_KEK must be 32 bytes base64");
|
||||
return kek;
|
||||
}
|
||||
if (process.env.NODE_ENV === "production") {
|
||||
throw new Error("S3_KEK is required in production (media encryption is mandatory)");
|
||||
}
|
||||
// Same derivation as the API's dev fallback, so a dev worker reads dev writes.
|
||||
return createHash("sha256").update(`dev-media-kek:${process.env.HOSTNAME ?? "local"}`).digest();
|
||||
}
|
||||
|
||||
export function workerStorageConfigFromEnv(): WorkerStorageConfig {
|
||||
return {
|
||||
endpoint: process.env.S3_ENDPOINT,
|
||||
region: process.env.S3_REGION ?? "us-east-1",
|
||||
accessKeyId: process.env.S3_ACCESS_KEY,
|
||||
secretAccessKey: process.env.S3_SECRET_KEY,
|
||||
bucket: process.env.S3_BUCKET ?? "tactical-media-broadcast-hub-media",
|
||||
forcePathStyle: process.env.S3_FORCE_PATH_STYLE !== "false",
|
||||
};
|
||||
}
|
||||
|
||||
function aesGcmSeal(key: Buffer, iv: Buffer, plaintext: Buffer): Buffer {
|
||||
const cipher = createCipheriv("aes-256-gcm", key, iv);
|
||||
return Buffer.concat([cipher.update(plaintext), cipher.final(), cipher.getAuthTag()]);
|
||||
}
|
||||
|
||||
function aesGcmOpen(key: Buffer, iv: Buffer, sealed: Buffer): Buffer {
|
||||
const tag = sealed.subarray(sealed.length - 16);
|
||||
const decipher = createDecipheriv("aes-256-gcm", key, iv);
|
||||
decipher.setAuthTag(tag);
|
||||
return Buffer.concat([decipher.update(sealed.subarray(0, sealed.length - 16)), decipher.final()]);
|
||||
}
|
||||
|
||||
/** Seal plaintext under the KEK: [magic | iv(12) | wrappedCEK | ciphertext+tag] — the API's exact frame. */
|
||||
function sealFrame(plaintext: Buffer, kek: Buffer): Buffer {
|
||||
const cek = randomBytes(32);
|
||||
const iv = randomBytes(12);
|
||||
const sealedCek = aesGcmSeal(kek, iv, cek);
|
||||
const cipher = createCipheriv("aes-256-gcm", cek, iv);
|
||||
const body = Buffer.concat([cipher.update(plaintext), cipher.final(), cipher.getAuthTag()]);
|
||||
return Buffer.concat([FRAME_MAGIC, iv, sealedCek, body]);
|
||||
}
|
||||
|
||||
/** Open a sealed frame; verifies GCM tags (CEK wrap + body) and returns plaintext. */
|
||||
function openFrame(sealed: Buffer, kek: Buffer): Buffer {
|
||||
if (sealed.subarray(0, FRAME_MAGIC.length).equals(FRAME_MAGIC) === false) {
|
||||
throw new Error("storage: not an encrypted frame (magic mismatch)");
|
||||
}
|
||||
let offset = FRAME_MAGIC.length;
|
||||
const iv = sealed.subarray(offset, offset + 12);
|
||||
offset += 12;
|
||||
const cekLen = 32 + 16;
|
||||
const sealedCek = sealed.subarray(offset, offset + cekLen);
|
||||
offset += cekLen;
|
||||
const body = sealed.subarray(offset);
|
||||
const cek = aesGcmOpen(kek, iv, sealedCek);
|
||||
const tag = body.subarray(body.length - 16);
|
||||
const decipher = createDecipheriv("aes-256-gcm", cek, iv);
|
||||
decipher.setAuthTag(tag);
|
||||
return Buffer.concat([decipher.update(body.subarray(0, body.length - 16)), decipher.final()]);
|
||||
}
|
||||
|
||||
function keyForHash(hash: string): string { return `sha256/${hash}.mp4`; }
|
||||
|
||||
/**
|
||||
* Envelope-encrypted S3 storage for the produce worker. Object bodies are sealed
|
||||
* exactly like the API's `S3Storage` (magic | iv | wrapped CEK | body+tag), so
|
||||
* the worker and the API interoperate on one bucket.
|
||||
*/
|
||||
export class WorkerS3Storage implements ProduceObjectStorage {
|
||||
private readonly client: S3Client;
|
||||
private readonly bucket: string;
|
||||
private readonly kek: Buffer;
|
||||
|
||||
constructor(config: WorkerStorageConfig) {
|
||||
this.bucket = config.bucket ?? "tactical-media-broadcast-hub-media";
|
||||
this.kek = config.mediaKek ?? loadMediaKek();
|
||||
this.client = new S3Client({
|
||||
endpoint: config.endpoint,
|
||||
region: config.region ?? "us-east-1",
|
||||
credentials:
|
||||
config.accessKeyId !== undefined && config.secretAccessKey !== undefined
|
||||
? { accessKeyId: config.accessKeyId, secretAccessKey: config.secretAccessKey }
|
||||
: undefined,
|
||||
forcePathStyle: config.forcePathStyle ?? true,
|
||||
});
|
||||
}
|
||||
|
||||
async getObject(key: string): Promise<Buffer> {
|
||||
const res = await this.client.send(new GetObjectCommand({ Bucket: this.bucket, Key: key }));
|
||||
if (res.Body === undefined) throw new Error(`empty body for ${key}`);
|
||||
const stored = Buffer.from(await res.Body.transformToByteArray());
|
||||
const plaintext = openFrame(stored, this.kek);
|
||||
// Content addressing: the key names the plaintext hash, so tampering is detected.
|
||||
const hashFromKey = key.split("/")[1]?.split(".")[0];
|
||||
if (hashFromKey !== undefined && /^[0-9a-f]{64}$/.test(hashFromKey)) {
|
||||
if (createHash("sha256").update(plaintext).digest("hex") !== hashFromKey) {
|
||||
throw new Error("storage: integrity failure (plaintext hash mismatch)");
|
||||
}
|
||||
}
|
||||
return plaintext;
|
||||
}
|
||||
|
||||
async putContent(body: Buffer, contentType: "video/mp4"): Promise<{ key: string }> {
|
||||
if (contentType !== "video/mp4") throw new Error("produce storage publishes video/mp4 only");
|
||||
const hash = createHash("sha256").update(body).digest("hex");
|
||||
const key = keyForHash(hash);
|
||||
await this.client.send(new PutObjectCommand({
|
||||
Bucket: this.bucket,
|
||||
Key: key,
|
||||
Body: sealFrame(body, this.kek),
|
||||
ContentType: contentType,
|
||||
ServerSideEncryption: "AES256",
|
||||
}));
|
||||
return { key };
|
||||
}
|
||||
}
|
||||
79
apps/media-worker/src/producer/store.ts
Normal file
79
apps/media-worker/src/producer/store.ts
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
/**
|
||||
* Produce job store ports (migrations/009_video.sql + 039_producer.sql).
|
||||
*
|
||||
* The worker owns the same two tables the API's video service owns: it claims
|
||||
* `video_processing_jobs` rows by kind, writes the derived `video_assets` row a
|
||||
* job produces, and drives the state machine to `complete`/`failed`. Ports, not
|
||||
* a concrete client, so tests and the (single-process) e2e path can double them –
|
||||
* the same standing as `ObservedStore`, `AssetStore` and `PreviewObjectStore`.
|
||||
*/
|
||||
import type { Lane, ProduceJobKind, ProduceRequest } from "./command-builder";
|
||||
|
||||
export type { Lane, ProduceJobKind, ProduceRequest } from "./command-builder";
|
||||
|
||||
/** The source `video_assets` row a claimed job renders from. */
|
||||
export interface ProduceSourceAsset {
|
||||
id: string;
|
||||
storageKey: string;
|
||||
mimeType: string;
|
||||
sizeBytes: number;
|
||||
width: number | null;
|
||||
height: number | null;
|
||||
durationMs: number | null;
|
||||
frameRate: number | null;
|
||||
/** True when the asset is known to carry an audio track (probed by 009's probe job). */
|
||||
hasAudio: boolean | null;
|
||||
}
|
||||
|
||||
export interface ClaimedProduceJob {
|
||||
jobId: string;
|
||||
workspaceId: string;
|
||||
kind: ProduceJobKind;
|
||||
request: ProduceRequest;
|
||||
source: ProduceSourceAsset;
|
||||
claimedAt: string;
|
||||
}
|
||||
|
||||
export interface ProduceOutput {
|
||||
/** Storage key of the produced mp4. */
|
||||
storageKey: string;
|
||||
sizeBytes: number;
|
||||
durationMs: number | null;
|
||||
width: number;
|
||||
height: number;
|
||||
frameRate: number;
|
||||
}
|
||||
|
||||
export interface ProduceJobStore {
|
||||
/**
|
||||
* Claim exactly one queued compose/mix row for this worker's lane, with lease
|
||||
* semantics: the `queued → transcoding` state bump inside
|
||||
* `FOR UPDATE SKIP LOCKED` is the lease, so two workers can never take the
|
||||
* same row and a row stays claimed until this worker finishes it.
|
||||
*
|
||||
* Returns `null` when nothing is claimable — including when a row was claimed
|
||||
* but could not be honoured (an undecodable `input_json`, a missing source
|
||||
* asset): such a row is failed inside the claim, never thrown, so no row can
|
||||
* strand in `transcoding` holding its lease. Store/transport errors still throw
|
||||
* (the poll loop's store-error path).
|
||||
*/
|
||||
claim(lane: Lane): Promise<ClaimedProduceJob | null>;
|
||||
/** Report progress on a claimed job (never changes state). */
|
||||
progress(jobId: string, percent: number): Promise<void>;
|
||||
/** Insert the derived asset and complete the job pointing at it. */
|
||||
complete(jobId: string, output: ProduceOutput, derivedFrom: string): Promise<string>;
|
||||
/** Fail a claimed job; `reason` must already be redacted/bounded. */
|
||||
fail(jobId: string, reason: string): Promise<void>;
|
||||
/** Resolve a voiceover asset id to its storage key. */
|
||||
asset(workspaceId: string, assetId: string): Promise<{ storageKey: string; mimeType: string } | null>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Object storage, i.e. the API's `StorageService` (content-addressed, envelope
|
||||
* encrypted): the worker downloads source bytes and publishes the produced mp4
|
||||
* through exactly this pair. The adapter lives in `s3-storage.ts`.
|
||||
*/
|
||||
export interface ProduceObjectStorage {
|
||||
getObject(key: string): Promise<Buffer>;
|
||||
putContent(body: Buffer, contentType: "video/mp4"): Promise<{ key: string }>;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue