From d9e8cdf7892aeeef647a5a9b2b3cb6dae3c15323 Mon Sep 17 00:00:00 2001 From: glass beadian Date: Wed, 14 Oct 2026 20:40:49 +0000 Subject: [PATCH] feat: produce worker deployment + web producer/collective/branding UI + mobile CSS (wave-2c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- apps/media-worker/src/env.ts | 32 ++ apps/media-worker/src/produce.ts | 73 +++ .../src/producer/command-builder.test.ts | 273 ++++++++++++ .../src/producer/command-builder.ts | 415 ++++++++++++++++++ apps/media-worker/src/producer/pg-store.ts | 245 +++++++++++ apps/media-worker/src/producer/runner.test.ts | 196 +++++++++ apps/media-worker/src/producer/runner.ts | 276 ++++++++++++ apps/media-worker/src/producer/s3-storage.ts | 149 +++++++ apps/media-worker/src/producer/store.ts | 79 ++++ apps/web/src/app/(main)/layout.tsx | 6 +- apps/web/src/app/(main)/settings/page.tsx | 331 +++++++++++++- apps/web/src/app/globals.css | 88 ++++ apps/web/src/components/mobile/top-bar.tsx | 4 + apps/web/src/components/shell/sidebar.tsx | 39 +- apps/web/src/lib/api-client.ts | 2 +- apps/web/src/lib/branding.ts | 174 ++++++++ infra/media-worker.Dockerfile | 4 +- infra/occupymedia/docker-compose.app.yml | 18 + 18 files changed, 2384 insertions(+), 20 deletions(-) create mode 100644 apps/media-worker/src/env.ts create mode 100644 apps/media-worker/src/produce.ts create mode 100644 apps/media-worker/src/producer/command-builder.test.ts create mode 100644 apps/media-worker/src/producer/command-builder.ts create mode 100644 apps/media-worker/src/producer/pg-store.ts create mode 100644 apps/media-worker/src/producer/runner.test.ts create mode 100644 apps/media-worker/src/producer/runner.ts create mode 100644 apps/media-worker/src/producer/s3-storage.ts create mode 100644 apps/media-worker/src/producer/store.ts create mode 100644 apps/web/src/lib/branding.ts diff --git a/apps/media-worker/src/env.ts b/apps/media-worker/src/env.ts new file mode 100644 index 0000000..2c1d3f3 --- /dev/null +++ b/apps/media-worker/src/env.ts @@ -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; +} diff --git a/apps/media-worker/src/produce.ts b/apps/media-worker/src/produce.ts new file mode 100644 index 0000000..aa3b053 --- /dev/null +++ b/apps/media-worker/src/produce.ts @@ -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 { + 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; +}); diff --git a/apps/media-worker/src/producer/command-builder.test.ts b/apps/media-worker/src/producer/command-builder.test.ts new file mode 100644 index 0000000..a091018 --- /dev/null +++ b/apps/media-worker/src/producer/command-builder.test.ts @@ -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 { + 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 = {}): 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); +}); diff --git a/apps/media-worker/src/producer/command-builder.ts b/apps/media-worker/src/producer/command-builder.ts new file mode 100644 index 0000000..81f3a64 --- /dev/null +++ b/apps/media-worker/src/producer/command-builder.ts @@ -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> = { + 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; + 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(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; + + 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; + 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; + 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>.` 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 . 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; +} diff --git a/apps/media-worker/src/producer/pg-store.ts b/apps/media-worker/src/producer/pg-store.ts new file mode 100644 index 0000000..117a585 --- /dev/null +++ b/apps/media-worker/src/producer/pg-store.ts @@ -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): 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(workspaceId: string, fn: (client: PoolClient) => Promise): Promise { + 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 { + // 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(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; + 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), + 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, reason: string): Promise { + 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 { + 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 { + 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 { + 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 { + 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; diff --git a/apps/media-worker/src/producer/runner.test.ts b/apps/media-worker/src/producer/runner.test.ts new file mode 100644 index 0000000..eb38332 --- /dev/null +++ b/apps/media-worker/src/producer/runner.test.ts @@ -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(); + let published: { key: string } | null = null; + let queue: ClaimedProduceJob[] = []; + let lane: string | null = null; + const assetKeys = new Map(); + 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 { + 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, overrides: Partial[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 }); }); diff --git a/apps/media-worker/src/producer/runner.ts b/apps/media-worker/src/producer/runner.ts new file mode 100644 index 0000000..7b748f7 --- /dev/null +++ b/apps/media-worker/src/producer/runner.ts @@ -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; + private running: Promise | 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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(promise: Promise, ms: number, name: ProduceFailure): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(produceError(name)), ms); + promise.then((value) => { clearTimeout(timer); resolve(value); }, + (error: unknown) => { clearTimeout(timer); reject(error); }); + }); +} diff --git a/apps/media-worker/src/producer/s3-storage.ts b/apps/media-worker/src/producer/s3-storage.ts new file mode 100644 index 0000000..525e9d2 --- /dev/null +++ b/apps/media-worker/src/producer/s3-storage.ts @@ -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 { + 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 }; + } +} diff --git a/apps/media-worker/src/producer/store.ts b/apps/media-worker/src/producer/store.ts new file mode 100644 index 0000000..df8ada9 --- /dev/null +++ b/apps/media-worker/src/producer/store.ts @@ -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; + /** Report progress on a claimed job (never changes state). */ + progress(jobId: string, percent: number): Promise; + /** Insert the derived asset and complete the job pointing at it. */ + complete(jobId: string, output: ProduceOutput, derivedFrom: string): Promise; + /** Fail a claimed job; `reason` must already be redacted/bounded. */ + fail(jobId: string, reason: string): Promise; + /** 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; + putContent(body: Buffer, contentType: "video/mp4"): Promise<{ key: string }>; +} diff --git a/apps/web/src/app/(main)/layout.tsx b/apps/web/src/app/(main)/layout.tsx index f581d0a..2ab0dfb 100644 --- a/apps/web/src/app/(main)/layout.tsx +++ b/apps/web/src/app/(main)/layout.tsx @@ -7,6 +7,7 @@ import { CommandPalette } from "../../components/shell/command-palette"; import { BottomNav } from "../../components/mobile/bottom-nav"; import { MobileTopBar } from "../../components/mobile/top-bar"; import { AuthProvider, useAuth } from "../../lib/auth-store"; +import { BrandingProvider } from "../../lib/branding"; import { AuthGate, SessionStatus } from "../../components/auth/auth-gate"; import { useApiResource } from "../../lib/use-api"; import type { DraftRecord } from "../../lib/api-types"; @@ -69,5 +70,8 @@ function Shell({ children }: { children: ReactNode }): React.ReactElement { } export default function MainLayout({ children }: { children: ReactNode }): React.ReactNode { - return {children}; + // BrandingProvider sits inside AuthProvider (it reads the session to know + // which collective to brand) and outside the shell, so both the sidebar + // and Settings share one boot read of /workspace/branding. + return {children}; } diff --git a/apps/web/src/app/(main)/settings/page.tsx b/apps/web/src/app/(main)/settings/page.tsx index 13f5537..4138c5b 100644 --- a/apps/web/src/app/(main)/settings/page.tsx +++ b/apps/web/src/app/(main)/settings/page.tsx @@ -1,16 +1,28 @@ "use client"; -import { useState } from "react"; +import { useEffect, useRef, useState } from "react"; import { useRouter } from "next/navigation"; -import { Avatar, Button } from "../../../components/ui/primitives"; -import { errorMessage } from "../../../lib/api-client"; +import { Avatar, Badge, Button, Card, Input } from "../../../components/ui/primitives"; +import { api, errorMessage } from "../../../lib/api-client"; import { useAuth } from "../../../lib/auth-store"; +import type { AuthResult } from "../../../lib/api-types"; +import { + BRANDING_IMAGE_TYPES, + MAX_BRANDING_IMAGE_BYTES, + MAX_TAGLINE_LENGTH, + brandingAssetUrl, + isHexAccent, + saveBranding, + uploadBrandingLogo, + useBranding, + type BrandingPatch, +} from "../../../lib/branding"; /** - * Deliberately small: this page states exactly what the API knows about the - * current session and how to end it. There is no notifications/preferences - * endpoint in the API, so none of that is faked here — no toggles that - * pretend to persist. + * Settings stays honest about what the API knows: the session block below is + * read-only truth, and the two mutation surfaces the API actually offers sit + * beside it — creating another collective (POST /auth/workspaces switches you + * into it) and the collective's branding (PUT /workspace/branding, Admin). */ function Field({ label, value, mono = false }: { label: string; value: string; mono?: boolean }): React.ReactElement { @@ -22,6 +34,304 @@ function Field({ label, value, mono = false }: { label: string; value: string; m ); } +function SectionHeading({ title, note }: { title: string; note: string }): React.ReactElement { + return ( +
+

{title}

+

{note}

+
+ ); +} + +/** Create a collective and switch this device into it (Owner of the new one). */ +function CollectivesSection(): React.ReactElement { + const { adoptToken, refresh } = useAuth(); + const [name, setName] = useState(""); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(null); + const [created, setCreated] = useState(null); + + const ready = name.trim().length >= 2; + + async function create(): Promise { + setBusy(true); + setError(null); + setCreated(null); + try { + const result = await api.post("/auth/workspaces", { workspaceName: name.trim() }, { redirectOn401: false }); + // The response is a session for the NEW collective: install its token + // (the old one stays valid on the server for the old workspace), then + // re-verify so /me, the shell and branding all follow the switch. + adoptToken(result.token); + await refresh(); + setCreated(result.me.workspace.name); + setName(""); + } catch (err) { + setError(errorMessage(err)); + } finally { + setBusy(false); + } + } + + return ( +
+ + +
+ setName(event.target.value)} + placeholder="New collective name" + aria-label="New collective name" + maxLength={100} + className="sm:flex-1" + /> + +
+
+ {error !== null &&

{error}

} + {error === null && created !== null && ( +

+ Created {created} — you are now working inside it. +

+ )} +
+ +
+ ); +} + +/** + * Branding: tagline, accent, logo. Admin-only in practice because the API + * gates PUT /workspace/branding (and the logo upload) on workspace.members.manage. + */ +function BrandingSection(): React.ReactElement { + const { can } = useAuth(); + const branding = useBranding(); + const allowed = can("workspace.members.manage"); + + const [tagline, setTagline] = useState(""); + const [accent, setAccent] = useState(""); + const [logoKey, setLogoKey] = useState(null); + const [logoPreview, setLogoPreview] = useState(null); + const [uploading, setUploading] = useState(false); + const [progress, setProgress] = useState(0); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(null); + const [saved, setSaved] = useState(false); + + // Seed the form once per collective from the boot read; after a save the + // local state already matches the server, so no reseed is needed. + const seeded = useRef(null); + useEffect(() => { + if (branding.source !== "live" || branding.workspaceId === null || seeded.current === branding.workspaceId) return; + seeded.current = branding.workspaceId; + setTagline(branding.tagline ?? ""); + setAccent(branding.accent ?? ""); + setLogoKey(branding.logoAssetKey); + setLogoPreview(branding.logoUrl); + setError(null); + setSaved(false); + }, [branding]); + + if (!allowed) { + return ( +
+ + +
+ read-only +

+ Your role ({can("workspace.view") ? "member" : "guest"}) cannot edit branding — the API requires the + workspace.members.manage permission. +

+
+ +
+ ); + } + + const trimmedTagline = tagline.trim(); + const trimmedAccent = accent.trim(); + const taglineProblem = + trimmedTagline.length > MAX_TAGLINE_LENGTH ? `At most ${MAX_TAGLINE_LENGTH} characters.` : null; + const accentProblem = + trimmedAccent === "" || isHexAccent(trimmedAccent) ? null : "Use a hex color like #2E7D32 or #E33."; + const ready = taglineProblem === null && accentProblem === null && !busy && !uploading; + const swatch = isHexAccent(trimmedAccent) ? trimmedAccent : null; + + async function upload(file: File): Promise { + setError(null); + setSaved(false); + if (!(BRANDING_IMAGE_TYPES as readonly string[]).includes(file.type)) { + setError(`Logo must be one of ${BRANDING_IMAGE_TYPES.join(", ")}.`); + return; + } + if (file.size > MAX_BRANDING_IMAGE_BYTES) { + setError(`Logo must be under ${Math.round(MAX_BRANDING_IMAGE_BYTES / 1024 / 1024)} MB.`); + return; + } + setUploading(true); + setProgress(0); + try { + // Upload stores the image and hands back its key; it goes live as the + // collective's mark only when Save PUTs it below. + const result = await uploadBrandingLogo(file, setProgress); + setLogoKey(result.key); + setLogoPreview(brandingAssetUrl(result.assetUrl)); + } catch (err) { + setError(errorMessage(err)); + } finally { + setUploading(false); + } + } + + async function save(): Promise { + setError(null); + setSaved(false); + setBusy(true); + try { + const patch: BrandingPatch = { + // Empty means "back to the Hub default" — the API takes null to clear. + tagline: trimmedTagline === "" ? null : trimmedTagline, + accent: trimmedAccent === "" ? null : trimmedAccent, + logoAssetKey: logoKey, + }; + await saveBranding(patch); + branding.reload(); + setSaved(true); + } catch (err) { + setError(errorMessage(err)); + } finally { + setBusy(false); + } + } + + return ( +
+ + +
+
+ + { setTagline(event.target.value); setSaved(false); }} + placeholder="Toma los medios" + maxLength={MAX_TAGLINE_LENGTH + 20} + /> +
+ Shown under the collective name in the sidebar. + + {trimmedTagline.length}/{MAX_TAGLINE_LENGTH} + +
+
+ +
+ +
+ { setAccent(event.target.value); setSaved(false); }} + placeholder="#FF1610" + className="font-mono" + aria-invalid={accentProblem !== null} + /> +
+

Hex only (#RGB or #RRGGBB). Empty uses the Hub red.

+
+ +
+ Logo +
+
+ {logoPreview !== null ? ( + Collective logo preview + ) : ( + none + )} +
+
+ { + const file = event.target.files?.[0]; + event.target.value = ""; + if (file !== undefined) void upload(file); + }} + className="w-full cursor-pointer text-2xs text-text-dim file:mr-2 file:cursor-pointer file:rounded-md file:border-0 file:bg-accent file:px-2 file:py-1 file:text-2xs file:font-medium file:text-white hover:file:bg-accent-hover" + /> +
+ {uploading ? ( + Uploading… {progress}% + ) : logoKey !== null ? ( + <> + {logoKey} + + + ) : ( + PNG, JPEG, WebP or GIF up to {Math.round(MAX_BRANDING_IMAGE_BYTES / 1024 / 1024)} MB. + )} +
+
+
+ {logoPreview !== null && saved === false && ( +

Saved only when you press Save branding.

+ )} +
+ + {error !== null &&

{error}

} + {error === null && saved &&

Branding saved.

} +
+ + {(taglineProblem !== null || accentProblem !== null) && ( + {taglineProblem ?? accentProblem} + )} +
+
+ +
+ ); +} + export default function SettingsPage(): React.ReactElement { const router = useRouter(); const { session, status, offline, logout } = useAuth(); @@ -50,10 +360,13 @@ export default function SettingsPage(): React.ReactElement {

Settings

- What the API knows about this session. Workspace-level configuration lives in Team and Connections; there is no - per-user preference store in the API, so none is shown here. + Your session, the collectives you can spin up, and this collective's branding. There is still no + per-user preference store in the API, so nothing here pretends to save one.

+ + +
diff --git a/apps/web/src/app/globals.css b/apps/web/src/app/globals.css index c6d1267..c3e8a36 100644 --- a/apps/web/src/app/globals.css +++ b/apps/web/src/app/globals.css @@ -74,6 +74,94 @@ html, body { max-width: 100%; } [data-mobile-main] .mobile-stack-table td > * { max-width: 100%; } } +/* --------------------------------------------------------------------------- + Phone tier (≤640px) — native feel, touch targets, safe areas, overflow. + Scoped globally (body-wide) rather than to [data-mobile-main] because the + auth screens (/login, /signup, /join, /submit, /oidc) and the body-portal + sheets render outside the main shell and were getting none of this. + These rules are unlayered, so they outrank Tailwind's `@layer utilities` + without needing !important. + --------------------------------------------------------------------------- */ +@media (max-width: 640px) { + /* -- Native feel: no tap highlight flash, no double-tap-zoom delay on + controls (pinch zoom stays available — `manipulation` only disables + the double-tap gesture), no iOS landscape text inflation, no document + rubber-band behind the app shell. -- */ + html { + -webkit-text-size-adjust: 100%; + text-size-adjust: 100%; + overscroll-behavior: none; + overflow-x: hidden; + } + body { + max-width: 100%; + overflow-x: hidden; + -webkit-tap-highlight-color: transparent; + } + body :where(button, a[href], [role="tab"], [role="button"], [role="switch"]) { + touch-action: manipulation; + } + + /* -- ≥44px touch targets for buttons / icon-buttons (glyph buttons here are + plain + ) : branding.logoUrl !== null ? ( + Collective logo ) : }
{title}
{actions} diff --git a/apps/web/src/components/shell/sidebar.tsx b/apps/web/src/components/shell/sidebar.tsx index 9b9750c..0734399 100644 --- a/apps/web/src/components/shell/sidebar.tsx +++ b/apps/web/src/components/shell/sidebar.tsx @@ -3,6 +3,7 @@ import Link from "next/link"; import { usePathname } from "next/navigation"; import { useAuth } from "../../lib/auth-store"; +import { useBranding } from "../../lib/branding"; import { useApiResource } from "../../lib/use-api"; import type { DraftRecord } from "../../lib/api-types"; import { LogoMark } from "../brand/logo"; @@ -22,7 +23,7 @@ const NAV: { href: string; label: string; icon: string; external?: boolean }[] = { href: "/analytics", label: "Analytics", icon: "analytics" }, { href: "/connections", label: "Channels", icon: "connections" }, { href: "/team", label: "Team", icon: "team" }, - { href: "/audit", label: "Audit", icon: "audit" }, + { href: "/audit", label: "Audit log", icon: "audit" }, { href: "/onboarding", label: "Onboarding", icon: "home" }, { href: "/settings", label: "Settings", icon: "settings" }, ]; @@ -33,19 +34,41 @@ const emptyDrafts: DraftRecord[] = []; export function Sidebar(): React.ReactElement { const pathname = usePathname(); const { session, can } = useAuth(); + const branding = useBranding(); const drafts = useApiResource("/publishing/drafts", identity, emptyDrafts, { state: "in_review", limit: 100 }); // Real in-review count from the API; hidden while loading/offline and for non-approvers. const approvals = can("approval.grant") && drafts.source === "live" ? drafts.data.length : 0; - const workspaceName = session?.workspace.name ?? "Studio"; + // Branding is per collective; the session's workspace is the source of truth + // for the name, branding only decorates it (and may be loading/unavailable). + const workspaceName = session?.workspace.name ?? branding.workspaceName ?? "Studio"; + const showTagline = branding.source === "live" && branding.tagline !== null; return (