diff --git a/packages/adapters/src/media-ffmpeg-renderer.ts b/packages/adapters/src/media-ffmpeg-renderer.ts new file mode 100644 index 0000000..46b1487 --- /dev/null +++ b/packages/adapters/src/media-ffmpeg-renderer.ts @@ -0,0 +1,599 @@ +import { spawn } from "node:child_process"; +import { access, mkdir, rm, stat, writeFile } from "node:fs/promises"; +import path from "node:path"; +import type { MediaRenderer, MediaRenderProgress } from "@rakazo/adapter-kit"; +import type { FfprobeReport, RenderPlan, RenderStep } from "@rakazo/contracts"; +import { getLogger } from "@rakazo/logging"; +import { assertInside } from "@rakazo/media/node"; +import { MediaToolMissingError, runMediaTool } from "./media-folder-source-probe.js"; + +/* + * Local ffmpeg executor for compiled render plans. Every process is spawned + * with an argv array and no shell. Plans come from compileRenderPlan, which + * already escapes every value; this module only adds runtime options + * (progress, hardware device) and the loudness measurement pass. + * + * Loudness: when the render step carries a single `loudnorm` with + * linear=true and no measured values, a first pass measures the program + * (print_format=json, null muxer) and the filter is rewritten with the + * measured_* options, so the second pass normalizes linearly to the target. + * When the measurement cannot be parsed the step runs single-pass (dynamic). + */ + +const HARDWARE_CANDIDATES = [ + "h264_vaapi", + "hevc_vaapi", + "h264_videotoolbox", + "hevc_videotoolbox", + "h264_nvenc", + "hevc_nvenc", +] as const; +const DEFAULT_VAAPI_DEVICE = "/dev/dri/renderD128"; +const DEFAULT_KILL_GRACE_MS = 2000; +const STDERR_TAIL_BYTES = 64 * 1024; + +export interface FfmpegMediaRendererOptions { + ffmpegPath?: string; + ffprobePath?: string; + /** When set, plan work dirs and probed or extracted paths must lie inside it. */ + root?: string; + vaapiDevice?: string; + /** Time between SIGTERM and SIGKILL when a render is aborted. */ + killGraceMs?: number; +} + +function abortError(signal: AbortSignal | undefined): Error { + const reason = signal?.reason; + if (reason instanceof Error && reason.name === "AbortError") return reason; + const error = new Error("render aborted"); + error.name = "AbortError"; + return error; +} + +/** "30000/1001" → 29.97; "0/0" → 0. */ +export function parseFrameRate(value: string | undefined): number { + if (!value) return 0; + const [num, den] = value.split("/").map(Number); + if (num === undefined || !Number.isFinite(num)) return 0; + const rate = den === undefined ? num : den > 0 ? num / den : 0; + return Number.isFinite(rate) ? Math.round(rate * 1000) / 1000 : 0; +} + +interface ProbeJson { + streams?: Array<{ + codec_type?: string; + codec_name?: string; + width?: number; + height?: number; + r_frame_rate?: string; + tags?: Record; + }>; + format?: { duration?: string; size?: string; tags?: Record }; +} + +/** ffprobe JSON → report without loudness. Tags are flattened as `format.x` / `stream.N.x`. */ +export function parseProbeReport(json: ProbeJson, fileSize: number): FfprobeReport { + const streams = json.streams ?? []; + const video = streams.find((stream) => stream.codec_type === "video"); + const audio = streams.find((stream) => stream.codec_type === "audio"); + const tags: Record = {}; + for (const [key, value] of Object.entries(json.format?.tags ?? {})) { + tags[`format.${key}`] = String(value); + } + streams.forEach((stream, index) => { + for (const [key, value] of Object.entries(stream.tags ?? {})) { + tags[`stream.${index}.${key}`] = String(value); + } + }); + const seconds = Number(json.format?.duration); + const size = Number(json.format?.size); + return { + durationMs: Number.isFinite(seconds) && seconds > 0 ? Math.round(seconds * 1000) : 0, + width: video?.width ?? 0, + height: video?.height ?? 0, + fps: parseFrameRate(video?.r_frame_rate), + videoCodec: video?.codec_name ?? "none", + audioCodec: audio?.codec_name ?? null, + sizeBytes: Number.isFinite(size) && size > 0 ? Math.round(size) : fileSize, + integratedLufs: null, + truePeakDb: null, + tags, + }; +} + +/** Integrated loudness and true peak from the ebur128 summary on stderr. */ +export function parseEbur128Summary(stderr: string): { + integratedLufs: number | null; + truePeakDb: number | null; +} { + const at = stderr.lastIndexOf("Summary:"); + if (at < 0) return { integratedLufs: null, truePeakDb: null }; + const summary = stderr.slice(at); + const integrated = /I:\s+(-?\d+(?:\.\d+)?) LUFS/.exec(summary); + const peak = /True peak:\s+Peak:\s+(-?\d+(?:\.\d+)?|-inf) dBFS/.exec(summary); + const lufs = integrated ? Number(integrated[1]) : null; + const tp = peak && peak[1] !== "-inf" ? Number(peak[1]) : null; + return { + // ebur128 reports -70 LUFS (its gate) for silence; that is "no programme". + integratedLufs: lufs !== null && Number.isFinite(lufs) && lufs > -70 ? lufs : null, + truePeakDb: tp !== null && Number.isFinite(tp) ? tp : null, + }; +} + +// --------------------------------------------------------------------------- +// Two-pass loudnorm +// --------------------------------------------------------------------------- + +const LOUDNORM = /loudnorm=((?:[^,;[\]\\]|\\.)*)/g; + +interface LoudnormSite { + /** Index of the options string inside the graph. */ + start: number; + end: number; + options: string; +} + +/** The single linear loudnorm without measured values, or null. */ +export function findTwoPassLoudnorm(graph: string): LoudnormSite | null { + const matches = [...graph.matchAll(LOUDNORM)]; + if (matches.length !== 1) return null; + const match = matches[0]!; + const options = match[1] ?? ""; + const parts = options.split(":"); + if (!parts.includes("linear=true")) return null; + if (parts.some((part) => part.startsWith("measured_"))) return null; + const start = (match.index ?? 0) + "loudnorm=".length; + return { start, end: start + options.length, options }; +} + +export interface LoudnormMeasurement { + inputI: number; + inputTp: number; + inputLra: number; + inputThresh: number; + targetOffset: number; +} + +/** The JSON block loudnorm prints after its summary line; null when absent or not finite. */ +export function parseLoudnormJson(stderr: string): LoudnormMeasurement | null { + const at = stderr.lastIndexOf("Parsed_loudnorm"); + if (at < 0) return null; + const open = stderr.indexOf("{", at); + const close = stderr.indexOf("}", open); + if (open < 0 || close < 0) return null; + let parsed: Record; + try { + parsed = JSON.parse(stderr.slice(open, close + 1)) as Record; + } catch { + return null; + } + const read = (key: string) => { + const value = Number(parsed[key]); + return Number.isFinite(value) ? value : null; + }; + const inputI = read("input_i"); + const inputTp = read("input_tp"); + const inputLra = read("input_lra"); + const inputThresh = read("input_thresh"); + const targetOffset = read("target_offset"); + if ( + inputI === null || + inputTp === null || + inputLra === null || + inputThresh === null || + targetOffset === null + ) { + return null; + } + return { inputI, inputTp, inputLra, inputThresh, targetOffset }; +} + +function withLoudnormOptions(graph: string, site: LoudnormSite, extra: string): string { + return `${graph.slice(0, site.start)}${site.options}:${extra}${graph.slice(site.end)}`; +} + +function measuredOptions(m: LoudnormMeasurement): string { + const f = (value: number) => value.toFixed(2); + return [ + `measured_I=${f(m.inputI)}`, + `measured_TP=${f(m.inputTp)}`, + `measured_LRA=${f(m.inputLra)}`, + `measured_thresh=${f(m.inputThresh)}`, + `offset=${f(m.targetOffset)}`, + ].join(":"); +} + +// --------------------------------------------------------------------------- +// Renderer +// --------------------------------------------------------------------------- + +interface RunOptions { + signal?: AbortSignal; + /** Called with the output time reached, in ms. */ + onTime?: (ms: number) => void; +} + +export class FfmpegMediaRenderer implements MediaRenderer { + readonly id = "ffmpeg"; + private readonly ffmpegPath: string; + private readonly ffprobePath: string; + private readonly vaapiDevice: string; + private readonly killGraceMs: number; + private encoders: Promise | null = null; + + constructor(private readonly options: FfmpegMediaRendererOptions = {}) { + this.ffmpegPath = options.ffmpegPath ?? "ffmpeg"; + this.ffprobePath = options.ffprobePath ?? "ffprobe"; + this.vaapiDevice = options.vaapiDevice ?? DEFAULT_VAAPI_DEVICE; + this.killGraceMs = options.killGraceMs ?? DEFAULT_KILL_GRACE_MS; + } + + hardwareEncoders(): Promise { + this.encoders ??= this.detectEncoders().catch((error) => { + getLogger().warn("media hardware encoder detection failed", { + reason: error instanceof Error ? error.message : String(error), + }); + return []; + }); + return this.encoders; + } + + private async detectEncoders(): Promise { + const { stdout } = await runMediaTool(this.ffmpegPath, ["-hide_banner", "-encoders"]); + const listed = new Set( + stdout + .toString("utf8") + .split("\n") + .map((line) => line.trim().split(/\s+/)[1]) + .filter((name): name is string => Boolean(name)), + ); + const hasVaapiDevice = await access(this.vaapiDevice).then( + () => true, + () => false, + ); + const usable: string[] = []; + for (const encoder of HARDWARE_CANDIDATES) { + if (!listed.has(encoder)) continue; + const vaapi = encoder.endsWith("_vaapi"); + if (vaapi && !hasVaapiDevice) continue; + // One-frame test encode: listed encoders often lack the device or driver. + const args = ["-hide_banner", "-v", "error", "-nostdin"]; + if (vaapi) args.push("-vaapi_device", this.vaapiDevice); + args.push("-f", "lavfi", "-i", "color=c=black:s=256x256:r=30:d=0.1", "-frames:v", "1"); + if (vaapi) args.push("-vf", "format=nv12,hwupload"); + args.push("-c:v", encoder, "-f", "null", "-"); + try { + await runMediaTool(this.ffmpegPath, args); + usable.push(encoder); + } catch { + // Not usable on this host. + } + } + return usable; + } + + async render( + plan: RenderPlan, + options: { signal?: AbortSignal; onProgress?: (progress: MediaRenderProgress) => void } = {}, + ): Promise { + const { signal, onProgress } = options; + if (signal?.aborted) throw abortError(signal); + const workDir = await this.workDir(plan.workDir); + const weight = (step: RenderStep) => + step.kind === "ffmpeg" ? Math.max(step.durationMs, 1) : 0; + const total = plan.steps.reduce((sum, step) => sum + weight(step), 0) || 1; + let done = 0; + const report = (stepId: string, fraction: number, stepWeight: number) => { + const value = (done + Math.min(1, Math.max(0, fraction)) * stepWeight) / total; + onProgress?.({ stepId, progress: Math.min(1, Math.max(0, value)) }); + }; + const written: string[] = []; + try { + for (const step of plan.steps) { + if (signal?.aborted) throw abortError(signal); + switch (step.kind) { + case "write_file": { + const file = await this.insideWorkDir(workDir, step.path); + await mkdir(path.dirname(file), { recursive: true }); + await writeFile(file, step.content, "utf8"); + written.push(file); + break; + } + case "motion": + case "inpaint": + case "synthetic_identity": { + const ready = await access(step.outPath).then( + () => true, + () => false, + ); + if (!ready) throw new Error(`${step.kind} step ${step.id} was not prepared`); + break; + } + case "ffmpeg": { + for (const produced of step.produces) await this.insideWorkDir(workDir, produced); + written.push(...step.produces); + await this.runStep(step, plan, workDir, { + signal, + report: (fraction) => report(step.id, fraction, weight(step)), + }); + done += weight(step); + report(step.id, 0, 0); + break; + } + } + } + } catch (error) { + await Promise.all(written.map((file) => rm(file, { force: true }).catch(() => undefined))); + if (signal?.aborted) throw abortError(signal); + throw error; + } + } + + private async runStep( + step: Extract, + plan: RenderPlan, + workDir: string, + options: { signal?: AbortSignal; report: (fraction: number) => void }, + ): Promise { + let argv = [...step.argv]; + const vaapi = argv.some((arg, i) => argv[i - 1] === "-c:v" && arg.endsWith("_vaapi")); + const prefix = vaapi ? ["-vaapi_device", this.vaapiDevice] : []; + const durationMs = Math.max(step.durationMs, 1); + const twoPass = await this.loudnormPlan(argv, plan, workDir); + if (twoPass) { + let measurement: LoudnormMeasurement | null = null; + try { + const { stderr } = await this.run([...prefix, ...twoPass.measureArgv], { + signal: options.signal, + onTime: (ms) => options.report((ms / durationMs) * 0.5), + }); + measurement = parseLoudnormJson(stderr); + } finally { + if (twoPass.measureScript) await rm(twoPass.measureScript, { force: true }); + } + if (measurement) { + argv = await twoPass.apply(measuredOptions(measurement)); + } else { + getLogger().warn("loudnorm measurement unreadable; rendering single pass", { + stepId: step.id, + }); + } + } + const offset = twoPass ? 0.5 : 0; + const scale = twoPass ? 0.5 : 1; + await this.run([...prefix, ...argv], { + signal: options.signal, + onTime: (ms) => options.report(offset + (ms / durationMs) * scale), + }); + options.report(1); + } + + /** + * Measurement argv and a function that writes the measured filter into the + * render argv (or its filtergraph script), or null when the step needs no + * measurement pass. + */ + private async loudnormPlan( + argv: string[], + plan: RenderPlan, + workDir: string, + ): Promise<{ + measureArgv: string[]; + measureScript: string | null; + apply: (extra: string) => Promise; + } | null> { + const inline = argv.indexOf("-filter_complex"); + const scriptAt = argv.indexOf("-filter_complex_script"); + const flagAt = inline >= 0 ? inline : scriptAt; + if (flagAt < 0 || flagAt + 1 >= argv.length) return null; + let graph: string; + let scriptPath: string | null = null; + if (inline >= 0) { + graph = argv[inline + 1]!; + } else { + scriptPath = await this.insideWorkDir(workDir, argv[scriptAt + 1]!); + let content: string | null = null; + for (const step of plan.steps) { + if (step.kind !== "write_file") continue; + if ((await this.insideWorkDir(workDir, step.path)) === scriptPath) content = step.content; + } + if (content === null) return null; + graph = content; + } + const site = findTwoPassLoudnorm(graph); + if (!site) return null; + + const tail = argv.slice(flagAt + 2); + const maps: string[] = []; + let duration: string | null = null; + for (let i = 0; i < tail.length - 1; i++) { + if (tail[i] === "-map") maps.push("-map", tail[i + 1]!); + if (tail[i] === "-t") duration = tail[i + 1]!; + } + const measureGraph = withLoudnormOptions(graph, site, "print_format=json"); + let measureScript: string | null = null; + let graphArg = measureGraph; + if (scriptPath) { + measureScript = path.join(path.dirname(scriptPath), "filtergraph.measure.txt"); + await writeFile(measureScript, measureGraph, "utf8"); + graphArg = measureScript; + } + const measureArgv = [ + ...argv.slice(0, flagAt + 1), + graphArg, + ...maps, + ...(duration ? ["-t", duration] : []), + "-f", + "null", + "-", + ]; + return { + measureArgv, + measureScript, + apply: async (extra) => { + const next = withLoudnormOptions(graph, site, extra); + const out = [...argv]; + if (scriptPath) await writeFile(scriptPath, next, "utf8"); + else out[flagAt + 1] = next; + return out; + }, + }; + } + + /** Spawn ffmpeg with progress on stdout. Rejects with the stderr tail on failure. */ + private run(args: string[], options: RunOptions): Promise<{ stderr: string }> { + const { signal } = options; + if (signal?.aborted) return Promise.reject(abortError(signal)); + return new Promise((resolve, reject) => { + const child = spawn(this.ffmpegPath, ["-progress", "pipe:1", "-nostats", ...args], { + stdio: ["ignore", "pipe", "pipe"], + shell: false, + windowsHide: true, + }); + let stderr = ""; + let pending = ""; + let killTimer: NodeJS.Timeout | null = null; + const onAbort = () => { + child.kill("SIGTERM"); + killTimer = setTimeout(() => child.kill("SIGKILL"), this.killGraceMs); + killTimer.unref(); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + child.stdout.setEncoding("utf8"); + child.stdout.on("data", (chunk: string) => { + pending += chunk; + const lines = pending.split("\n"); + pending = lines.pop() ?? ""; + for (const line of lines) { + const match = /^out_time_(us|ms)=(\d+)/.exec(line.trim()); + // Both keys are microseconds in ffmpeg 5+ (out_time_ms is misnamed). + if (match) options.onTime?.(Number(match[2]) / 1000); + } + }); + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk: string) => { + stderr += chunk; + if (stderr.length > STDERR_TAIL_BYTES * 2) stderr = stderr.slice(-STDERR_TAIL_BYTES); + }); + child.on("error", (error: NodeJS.ErrnoException) => { + signal?.removeEventListener("abort", onAbort); + reject(error.code === "ENOENT" ? new MediaToolMissingError(this.ffmpegPath) : error); + }); + child.on("close", (code) => { + signal?.removeEventListener("abort", onAbort); + if (killTimer) clearTimeout(killTimer); + if (signal?.aborted) { + reject(abortError(signal)); + return; + } + if (code !== 0) { + const tail = stderr.trim().split("\n").slice(-4).join(" "); + reject(new Error(`ffmpeg failed (${code}): ${tail}`)); + return; + } + resolve({ stderr }); + }); + }); + } + + async probe( + file: string, + options: { measureLoudness?: boolean; signal?: AbortSignal } = {}, + ): Promise { + const target = await this.checked(file); + const { stdout } = await runMediaTool( + this.ffprobePath, + ["-v", "error", "-print_format", "json", "-show_format", "-show_streams", "--", target], + { signal: options.signal }, + ); + const report = parseProbeReport( + JSON.parse(stdout.toString("utf8")) as ProbeJson, + (await stat(target)).size, + ); + if (!options.measureLoudness || report.audioCodec === null) return report; + const { stderr } = await runMediaTool( + this.ffmpegPath, + [ + "-hide_banner", + "-nostats", + "-nostdin", + "-i", + target, + "-map", + "0:a:0", + "-af", + "ebur128=peak=true", + "-f", + "null", + "-", + ], + { signal: options.signal }, + ); + return { ...report, ...parseEbur128Summary(stderr) }; + } + + async extractFrame(input: { + path: string; + tMs: number; + outPath: string; + width?: number; + signal?: AbortSignal; + }): Promise { + const source = await this.checked(input.path); + if (path.extname(input.outPath).toLowerCase() !== ".png") { + throw new Error("frames are extracted as PNG"); + } + const outPath = await this.checked(input.outPath); + if (!Number.isFinite(input.tMs) || input.tMs < 0) throw new Error("invalid frame time"); + const args = [ + "-hide_banner", + "-v", + "error", + "-nostdin", + "-y", + "-ss", + (input.tMs / 1000).toFixed(3), + "-i", + source, + "-frames:v", + "1", + ]; + if (input.width !== undefined) { + const width = Math.max(2, Math.floor(input.width / 2) * 2); + args.push("-vf", `scale=${width}:-2`); + } + args.push( + "-map_metadata", + "-1", + "-fflags", + "+bitexact", + "-flags:v", + "+bitexact", + "-f", + "image2", + "-c:v", + "png", + outPath, + ); + await runMediaTool(this.ffmpegPath, args, { signal: input.signal }); + const written = await stat(outPath).catch(() => null); + if (!written || written.size === 0) throw new Error("no frame at the requested time"); + } + + private async checked(file: string): Promise { + if (!path.isAbsolute(file)) throw new Error("media paths must be absolute"); + return this.options.root ? assertInside(this.options.root, file) : path.resolve(file); + } + + private async workDir(dir: string): Promise { + const resolved = await this.checked(dir); + await mkdir(resolved, { recursive: true }); + return resolved; + } + + /** Steps may only write inside the plan's work directory. */ + private async insideWorkDir(workDir: string, file: string): Promise { + if (!path.isAbsolute(file)) throw new Error("plan paths must be absolute"); + return assertInside(workDir, file); + } +} diff --git a/packages/adapters/src/media-render.fixtures.ts b/packages/adapters/src/media-render.fixtures.ts new file mode 100644 index 0000000..6926759 --- /dev/null +++ b/packages/adapters/src/media-render.fixtures.ts @@ -0,0 +1,676 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, readFile, realpath, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import type { + JobPublisher, + LinkShortener, + MediaInferenceProvider, + MediaPublisher, + MediaRenderer, + MotionGraphicsRenderer, +} from "@rakazo/adapter-kit"; +import type { + BrandKit, + Edl, + FfprobeReport, + GateIssue, + GateReport, + MediaAnalysisKind, + MediaEditPlan, + MediaRender, + PlatformProfile, +} from "@rakazo/contracts"; +import { BrandKitSchema, EdlSchema, PlatformProfileSchema } from "@rakazo/contracts"; +import type { MediaRenderPatch, MediaRepos, PrismaClient } from "@rakazo/db"; +import { + ACTIVE_RENDER_STATUSES, + bootstrapUserSpace, + createDb, + createMediaRepos, + IsolationError, + MediaStateConflictError, + mapMediaEditPlan, + mapMediaRender, +} from "@rakazo/db"; +import type { CompileSource } from "@rakazo/media"; +import { evaluateGate } from "@rakazo/media"; +import { LocalArtifactStore } from "./artifacts.js"; +import type { + MediaPreparedSources, + MediaPrivacyService, + MediaRealtimeEvent, + MediaServiceDeps, +} from "./media-context.js"; + +/* + * Test harness for the render service and job: the repos the render slice + * reads, either in memory or backed by PostgreSQL (VERIFY_DATABASE), plus a + * pass-through privacy service. Everything is synthetic. + */ + +export function renderTestProfile(overrides: Record = {}): PlatformProfile { + return PlatformProfileSchema.parse({ + id: "example-vertical-test", + name: "Example vertical test", + kind: "own_site", + reviewedAt: "2026-09-01", + video: { + aspectRatios: ["9:16"], + width: 1080, + height: 1920, + fps: 30, + codec: "h264", + maxBitrateKbps: 8000, + maxFileSizeMb: 500, + }, + duration: { hardMinMs: 1000, hardMaxMs: 120_000, hookWindowMs: 2000 }, + safeZones: { topPct: 8, bottomPct: 15, leftPct: 4, rightPct: 10 }, + captions: { recommended: true, maxLines: 2 }, + audio: { targetLufs: -14, truePeakDb: -1 }, + text: { captionMaxChars: 500, hashtagLimit: 5, linkPolicy: "allowed" }, + content: { thresholds: { violence: 0.5 } }, + disclosure: { aiAlteredLabelRequired: true }, + consent: { requirePerformerConsent: false, requireAgeVerification: false }, + export: {}, + ...overrides, + }); +} + +export function renderTestKit(input: { + logoArtifactId: string; + musicArtifactId: string; +}): BrandKit { + return BrandKitSchema.parse({ + name: "Studio Example", + colors: { + primary: "#224488", + secondary: "#446688", + accent: "#FFAA00", + text: "#FFFFFF", + background: "#101010", + }, + logoArtifactIds: [input.logoArtifactId], + watermark: { + artifactId: input.logoArtifactId, + position: "top_right", + opacity: 0.8, + scale: 0.2, + }, + captionPresets: [ + { + id: "clean", + name: "Clean", + template: "clean", + fontFamily: "DejaVu Sans", + sizePct: 5, + textColor: "#FFFFFF", + }, + ], + musicBeds: [ + { artifactId: input.musicArtifactId, title: "Bed", license: "own", commercialUse: true }, + ], + }); +} + +export interface HarnessAsset { + assetId: string; + source: CompileSource; +} + +export interface RenderHarness { + backend: "memory" | "postgres"; + mediaRoot: string; + spaceId: string; + userId: string; + deps: MediaServiceDeps; + events: MediaRealtimeEvent[]; + enqueued: string[]; + /** Register an asset; the caller writes the file at `assetPath(assetId, fileName)`. */ + createAsset(name: string): Promise; + assetPath(assetId: string, fileName: string): string; + setSource(assetId: string, source: CompileSource): void; + sources: Map; + addAnalysis(assetId: string, kind: MediaAnalysisKind, data: unknown): Promise; + addSegments(assetId: string, ranges: Array<{ startMs: number; endMs: number }>): Promise; + addArtifact(file: string, name: string, mimeType: string): Promise; + createProject(input: { assetIds: string[]; kit: BrandKit | null; profileId: string }): Promise; + createPlan(projectId: string, edl: unknown): Promise; + planVersions(projectId: string, profileId: string): Promise; + artifactCount(): Promise; + cleanup(): Promise; +} + +export interface HarnessOptions { + profiles: PlatformProfile[]; + renderer: MediaRenderer; + motion: MotionGraphicsRenderer; + inference?: MediaInferenceProvider | null; + now?: () => Date; +} + +const unusedPublisher: MediaPublisher = { + id: "unused", + publish: async () => { + throw new Error("not used by render tests"); + }, +}; + +/** PostgreSQL when VERIFY_DATABASE and DATABASE_URL are set, otherwise in memory. */ +export async function createRenderHarness(options: HarnessOptions): Promise { + const dir = await realpath(await mkdtemp(path.join(tmpdir(), "media-render-"))); + const mediaRoot = path.join(dir, "media"); + await mkdir(mediaRoot, { recursive: true }); + const artifacts = new LocalArtifactStore(path.join(dir, "store")); + const events: MediaRealtimeEvent[] = []; + const enqueued: string[] = []; + const jobs: JobPublisher = { + enqueue: async (job) => { + enqueued.push(`${job.name}:${JSON.stringify(job.payload)}`); + }, + cancel: async () => undefined, + close: async () => undefined, + }; + const databaseUrl = process.env.DATABASE_URL; + const backend = + process.env.VERIFY_DATABASE && databaseUrl + ? await postgresBackend(databaseUrl) + : memoryBackend(options.now ?? (() => new Date())); + const sources = new Map(); + const deps: MediaServiceDeps = { + prisma: backend.prisma, + repos: backend.repos, + jobs, + inference: options.inference ?? null, + renderer: options.renderer, + motion: options.motion, + publisher: unusedPublisher, + shortener: null as LinkShortener | null, + sources: {}, + mediaRoot, + profiles: async () => ({ + profiles: new Map( + options.profiles.map((profile) => [profile.id, { profile, origin: "builtin" as const }]), + ), + errors: [], + }), + artifacts, + model: { generate: async () => null }, + publish: async (_spaceId, event) => { + events.push(event); + }, + now: options.now ?? (() => new Date()), + }; + const context = (userId: string) => ({ + operationId: "test", + traceId: "test", + spaceId: backend.spaceId, + userId, + signal: new AbortController().signal, + }); + + return { + backend: backend.kind, + mediaRoot, + spaceId: backend.spaceId, + userId: backend.userId, + deps, + events, + enqueued, + sources, + createAsset: (name) => backend.createAsset(name), + assetPath: (assetId, fileName) => + path.join(mediaRoot, backend.spaceId, "assets", assetId, fileName), + setSource: (assetId, source) => sources.set(assetId, source), + addAnalysis: (assetId, kind, data) => backend.addAnalysis(assetId, kind, data), + addSegments: (assetId, ranges) => backend.addSegments(assetId, ranges), + async addArtifact(file, name, mimeType) { + const bytes = new Uint8Array(await readFile(file)); + const stored = await artifacts.put({ name, mimeType, bytes }, context(backend.userId)); + const row = await backend.prisma.artifact.create({ + data: { + spaceId: backend.spaceId, + userId: backend.userId, + name, + mimeType, + size: bytes.byteLength, + hash: "test", + storageKey: stored.id, + }, + select: { id: true }, + }); + return row.id; + }, + createProject: (input) => backend.createProject(input), + createPlan: (projectId, edl) => { + const parsed = EdlSchema.parse(edl); + return backend.repos.createPlanVersion(backend.spaceId, { + projectId, + profileId: parsed.output.profileId, + edl: parsed, + createdBy: "user", + rationale: null, + }); + }, + planVersions: (projectId, profileId) => + backend.repos.listPlanVersions(backend.spaceId, projectId, profileId), + artifactCount: () => backend.prisma.artifact.count({ where: { spaceId: backend.spaceId } }), + async cleanup() { + await backend.cleanup(); + await rm(dir, { recursive: true, force: true }); + }, + }; +} + +interface Backend { + kind: "memory" | "postgres"; + prisma: PrismaClient; + repos: MediaRepos; + spaceId: string; + userId: string; + createAsset(name: string): Promise; + addAnalysis(assetId: string, kind: MediaAnalysisKind, data: unknown): Promise; + addSegments(assetId: string, ranges: Array<{ startMs: number; endMs: number }>): Promise; + createProject(input: { assetIds: string[]; kit: BrandKit | null; profileId: string }): Promise; + cleanup(): Promise; +} + +async function postgresBackend(databaseUrl: string): Promise { + const db = createDb(databaseUrl, { poolMax: 4 }); + const prisma = db.prisma; + const repos = createMediaRepos(prisma); + const userId = `mr${randomUUID().replaceAll("-", "")}`; + await prisma.user.create({ data: { id: userId, name: "Test", email: `${userId}@rakazo.test` } }); + const { spaceId } = await bootstrapUserSpace( + prisma, + { id: userId }, + { signupsEnabled: "true", signupAllowlist: "" }, + { claimDeploymentOwner: false }, + ); + const source = await repos.createSource(spaceId, { + kind: "folder", + label: "Synthetic", + ownerUserId: userId, + config: {}, + }); + let analysisVersion = 0; + return { + kind: "postgres", + prisma, + repos, + spaceId, + userId, + async createAsset(name) { + const result = await repos.upsertSourceItems(spaceId, source.id, [ + { + externalId: name, + kind: "video", + capturedAt: new Date(Date.UTC(2026, 0, 1)), + durationMs: 8000, + width: 640, + height: 360, + fps: 30, + gpsLat: null, + gpsLon: null, + deviceModel: null, + albumNames: [], + isFavorite: false, + fileName: name, + sizeBytes: 1, + revision: "r1", + }, + ]); + return result.changedAssetIds[0]!; + }, + async addAnalysis(assetId, kind, data) { + analysisVersion += 1; + await repos.upsertAnalysis(spaceId, assetId, { + kind, + modelId: "emulator", + modelVersion: `v${analysisVersion}`, + data: data as never, + }); + }, + async addSegments(assetId, ranges) { + await repos.replaceSegments( + spaceId, + assetId, + 1, + ranges.map((range) => ({ kind: "shot", ...range })), + ); + }, + async createProject(input) { + const kit = input.kit ? await repos.upsertBrandKit(spaceId, { kit: input.kit }) : null; + const project = await repos.createProject(spaceId, userId, { + title: "Synthetic project", + goal: "teaser", + profileIds: [input.profileId], + assetIds: input.assetIds, + brandKitId: kit?.id ?? null, + }); + return project.id; + }, + async cleanup() { + const memberships = await prisma.member.findMany({ + where: { userId }, + select: { organizationId: true }, + }); + await prisma.organization.deleteMany({ + where: { id: { in: memberships.map((member) => member.organizationId) } }, + }); + await prisma.user.deleteMany({ where: { id: userId } }); + await prisma.$disconnect(); + await db.pool.end(); + }, + }; +} + +interface RenderRow { + id: string; + spaceId: string; + projectId: string; + planId: string; + profileId: string; + quality: string; + status: string; + progress: number; + outputs: unknown[]; + gateReport: unknown; + ffprobe: unknown; + autoFixRound: number; + error: string | null; + createdBy: string; + createdAt: Date; + updatedAt: Date; +} + +/** In-memory stand-in for the repos the render slice uses, with the same state rules. */ +function memoryBackend(now: () => Date): Backend { + const spaceId = "space-memory"; + const userId = "user-memory"; + let counter = 0; + const id = (prefix: string) => `${prefix}${(++counter).toString().padStart(6, "0")}`; + const assets = new Set(); + const analyses = new Map(); + const segments: Array<{ + id: string; + assetId: string; + startMs: number; + endMs: number; + paidOnly: boolean; + }> = []; + const kits = new Map(); + const projects = new Map< + string, + { id: string; assetIds: string[]; collectionIds: string[]; brandKitId: string | null } + >(); + const plans: Array[0]> = []; + const renders = new Map(); + const artifacts: Array & { id: string; spaceId: string }> = []; + + const requireRender = (space: string, renderId: string) => { + const row = renders.get(renderId); + if (!row || row.spaceId !== space) throw new IsolationError(); + return row; + }; + const requirePlan = (space: string, planId: string) => { + const row = plans.find((plan) => plan.id === planId && plan.spaceId === space); + if (!row) throw new IsolationError(); + return row; + }; + const toRender = (row: RenderRow): MediaRender => + mapMediaRender(structuredClone(row) as unknown as Parameters[0]); + + const repos = { + async mediaSpaceOf(kind: string, renderId: string) { + return kind === "render" ? (renders.get(renderId)?.spaceId ?? null) : null; + }, + async getPlan(space: string, planId: string) { + return mapMediaEditPlan(requirePlan(space, planId)); + }, + async createPlanVersion( + space: string, + input: { + projectId: string; + profileId: string; + edl: unknown; + createdBy: "agent" | "user"; + rationale: string | null; + }, + ) { + const edl = EdlSchema.parse(input.edl); + if (edl.output.profileId !== input.profileId) throw new Error("profile mismatch"); + if (space !== spaceId || !projects.has(input.projectId)) throw new IsolationError(); + const version = + plans + .filter((p) => p.projectId === input.projectId && p.profileId === input.profileId) + .reduce((max, p) => Math.max(max, p.version), 0) + 1; + const row = { + id: id("plan"), + spaceId: space, + projectId: input.projectId, + profileId: input.profileId, + version, + edl: structuredClone(edl), + createdBy: input.createdBy, + rationale: input.rationale, + createdAt: now(), + } as unknown as Parameters[0]; + plans.push(row); + return mapMediaEditPlan(row); + }, + async listPlanVersions(space: string, projectId: string, profileId: string) { + return plans + .filter((p) => p.spaceId === space && p.projectId === projectId && p.profileId === profileId) + .sort((a, b) => b.version - a.version) + .map(mapMediaEditPlan); + }, + async getProjectRecord(space: string, projectId: string) { + const project = projects.get(projectId); + if (!project || space !== spaceId) throw new IsolationError(); + return { ...project, createdBy: userId }; + }, + async getBrandKit(space: string, kitId: string) { + const kit = kits.get(kitId); + if (!kit || space !== spaceId) throw new IsolationError(); + return { id: kitId, kit, createdAt: now().toISOString(), updatedAt: now().toISOString() }; + }, + async collectionAssetIds() { + return []; + }, + async listSegments(space: string, assetIds: string[]) { + if (space !== spaceId) return []; + return segments.filter((segment) => assetIds.includes(segment.assetId)); + }, + async latestAnalysis(space: string, assetId: string, kind: string) { + if (space !== spaceId || !assets.has(assetId)) throw new IsolationError(); + const data = analyses.get(`${assetId}:${kind}`); + return data === undefined ? null : { data }; + }, + async createRender( + space: string, + input: { planId: string; quality: string; createdBy: string }, + ) { + const plan = requirePlan(space, input.planId); + const row: RenderRow = { + id: id("render"), + spaceId: space, + projectId: plan.projectId, + planId: plan.id, + profileId: plan.profileId, + quality: input.quality, + status: "queued", + progress: 0, + outputs: [], + gateReport: null, + ffprobe: null, + autoFixRound: 0, + error: null, + createdBy: input.createdBy, + createdAt: now(), + updatedAt: now(), + }; + renders.set(row.id, row); + return toRender(row); + }, + async getRender(space: string, renderId: string) { + return toRender(requireRender(space, renderId)); + }, + async getRenderRecord(space: string, renderId: string) { + return structuredClone(requireRender(space, renderId)); + }, + async claimRender(space: string, renderId: string) { + const row = requireRender(space, renderId); + if (row.status !== "queued") return null; + Object.assign(row, { status: "rendering", progress: 0, error: null, updatedAt: now() }); + return toRender(row); + }, + async updateRender( + space: string, + renderId: string, + patch: MediaRenderPatch, + from = ACTIVE_RENDER_STATUSES, + ) { + const row = requireRender(space, renderId); + if (!from.includes(row.status as never)) return null; + for (const [key, value] of Object.entries(patch)) { + if (value === undefined) continue; + (row as unknown as Record)[key] = + key === "progress" ? Math.min(1, Math.max(0, value as number)) : structuredClone(value); + } + row.updatedAt = now(); + return toRender(row); + }, + async cancelRender(space: string, renderId: string) { + const row = requireRender(space, renderId); + if (!ACTIVE_RENDER_STATUSES.includes(row.status as never)) { + throw new MediaStateConflictError(`Render is already ${row.status}`); + } + row.status = "canceled"; + row.updatedAt = now(); + return toRender(row); + }, + }; + + const prisma = { + artifact: { + async findFirst(args: { where: { id: string; spaceId: string } }) { + return ( + artifacts.find((a) => a.id === args.where.id && a.spaceId === args.where.spaceId) ?? null + ); + }, + async create(args: { data: Record & { spaceId: string } }) { + const row = { ...args.data, id: id("artifact") }; + artifacts.push(row); + return row; + }, + async deleteMany(args: { where: { id: string; spaceId: string } }) { + const index = artifacts.findIndex( + (a) => a.id === args.where.id && a.spaceId === args.where.spaceId, + ); + if (index >= 0) artifacts.splice(index, 1); + return { count: index >= 0 ? 1 : 0 }; + }, + async count(args: { where: { spaceId: string } }) { + return artifacts.filter((a) => a.spaceId === args.where.spaceId).length; + }, + }, + }; + + return { + kind: "memory", + prisma: prisma as unknown as PrismaClient, + repos: repos as unknown as MediaRepos, + spaceId, + userId, + async createAsset() { + const assetId = id("asset"); + assets.add(assetId); + return assetId; + }, + async addAnalysis(assetId, kind, data) { + analyses.set(`${assetId}:${kind}`, data); + }, + async addSegments(assetId, ranges) { + for (const range of ranges) { + segments.push({ id: id("segment"), assetId, ...range, paidOnly: false }); + } + }, + async createProject(input) { + let brandKitId: string | null = null; + if (input.kit) { + brandKitId = id("kit"); + kits.set(brandKitId, input.kit); + } + const projectId = id("project"); + projects.set(projectId, { + id: projectId, + assetIds: input.assetIds, + collectionIds: [], + brandKitId, + }); + return projectId; + }, + async cleanup() {}, + }; +} + +// --------------------------------------------------------------------------- +// Privacy stand-in +// --------------------------------------------------------------------------- + +export interface FakePrivacy extends MediaPrivacyService { + gateCalls: Array<{ autoFixRound: number; edl: Edl; ffprobe: FfprobeReport }>; +} + +/** + * Pass-through sources (no redactions in these plans) and the deterministic + * gate on the probed output. `failWith` adds a blocking content issue to every + * report, like a classifier hit at the start of the output. + */ +export function createFakePrivacy( + sources: Map, + options: { failWith?: GateIssue; now?: () => Date } = {}, +): FakePrivacy { + const gateCalls: FakePrivacy["gateCalls"] = []; + const unsupported = async (): Promise => { + throw new Error("not used by render tests"); + }; + return { + gateCalls, + async prepareSources(input): Promise { + const prepared = new Map(); + for (const clip of input.edl.tracks.video) { + const source = sources.get(clip.assetId); + if (!source) throw new Error(`no source for ${clip.assetId}`); + prepared.set(clip.assetId, source); + } + return { sources: prepared, aiAlterations: [], fallbacks: [] }; + }, + async gate(input): Promise { + gateCalls.push({ autoFixRound: input.autoFixRound, edl: input.edl, ffprobe: input.ffprobe }); + const report = evaluateGate({ + profile: input.profile, + edl: input.edl, + ffprobe: input.ffprobe, + content: [], + faces: [], + trackingGaps: [], + marks: [], + people: [], + copy: input.copy, + aiAlterations: [], + bundleHasDisclosure: input.bundleHasDisclosure, + watermarkBox: null, + sampledFrames: 0, + autoFixRound: input.autoFixRound, + now: (options.now ?? (() => new Date()))(), + }); + if (!options.failWith) return report; + return { ...report, passed: false, issues: [...report.issues, options.failWith] }; + }, + createMark: unsupported, + updateMark: unsupported, + removePerson: unsupported, + }; +} diff --git a/packages/adapters/src/media-render.ts b/packages/adapters/src/media-render.ts new file mode 100644 index 0000000..32fd240 --- /dev/null +++ b/packages/adapters/src/media-render.ts @@ -0,0 +1,835 @@ +import { createHash } from "node:crypto"; +import { mkdir, readFile, rename, rm, stat, writeFile } from "node:fs/promises"; +import path from "node:path"; +import type { AdapterContext, MediaJobHandlers, MediaRenderProgress } from "@rakazo/adapter-kit"; +import { mediaRenderJob, UnsupportedMediaOperationError } from "@rakazo/adapter-kit"; +import type { + AiAlteration, + BrandKit, + Edl, + GateReport, + MediaRender, + MediaRenderOutput, + MediaRenderQuality, + PlatformProfile, + RenderPlan, +} from "@rakazo/contracts"; +import { MAX_AUTO_FIX_ROUNDS } from "@rakazo/contracts"; +import type { MediaRenderPatch } from "@rakazo/db"; +import { InvalidMediaInputError } from "@rakazo/db"; +import { getLogger } from "@rakazo/logging"; +import type { CaptionWord } from "@rakazo/media"; +import { compileRenderPlan, findMetadataLeaks, planAutoFix } from "@rakazo/media"; +import { resolveInside } from "@rakazo/media/node"; +import type { + MediaActor, + MediaPrivacyService, + MediaRenderService, + MediaServiceDeps, +} from "./media-context.js"; +import { MediaToolMissingError } from "./media-folder-source-probe.js"; + +/* + * Render service (API side) and render job (worker side). + * + * A render runs: claim → prepare redacted sources (privacy slice) → pre-render + * motion clips → compile → ffmpeg → probe → gate. A blocked gate gets up to + * MAX_AUTO_FIX_ROUNDS compliance-only repairs (planAutoFix); each repair is + * stored as a new plan version by the agent and rendered again under the same + * render id. Only a passed gate moves outputs out of the work directory and + * marks the render done; every other outcome removes the render's files. + * + * Layout: `/renders//{video.mp4,poster.jpg,captions.srt, + * captions.vtt}` with intermediates in `work/` beside them. + * + * Cancel: the API marks the render canceled and aborts it when this process + * runs it. A worker in another process notices through a failed status update + * (updateRender returns null once canceled) or the periodic status check. + */ + +/** Renders running in this process, so a cancel can kill ffmpeg immediately. */ +export type RunningMediaRenders = Map; + +const runningRenders: RunningMediaRenders = new Map(); + +const PROGRESS_INTERVAL_MS = 1000; +const CANCEL_POLL_MS = 2000; +/** Share of the progress bar spent rendering; probing and gating take the rest. */ +const RENDER_SHARE = 0.85; + +const OUTPUT_FILES = { + video: "video.mp4", + poster: "poster.jpg", + subtitles_srt: "captions.srt", + subtitles_vtt: "captions.vtt", +} as const; + +const OUTPUT_MIME: Record = { + video: "video/mp4", + poster: "image/jpeg", + subtitles_srt: "application/x-subrip", + subtitles_vtt: "text/vtt", +}; + +const EXTENSION_BY_MIME: Record = { + "image/png": ".png", + "image/jpeg": ".jpg", + "image/svg+xml": ".svg", + "image/webp": ".webp", + "audio/mpeg": ".mp3", + "audio/wav": ".wav", + "audio/x-wav": ".wav", + "audio/aac": ".aac", + "audio/mp4": ".m4a", + "audio/ogg": ".ogg", + "audio/flac": ".flac", + "font/ttf": ".ttf", + "font/otf": ".otf", + "font/collection": ".ttc", +}; + +/** A failure whose message is safe to show the user (no paths, no tool output). */ +export class MediaRenderFailure extends Error { + constructor(message: string) { + super(message); + this.name = "MediaRenderFailure"; + } +} + +class RenderCanceled extends Error { + constructor() { + super("render canceled"); + this.name = "RenderCanceled"; + } +} + +function userMessage(error: unknown): string { + if (error instanceof MediaRenderFailure) return error.message; + if (error instanceof UnsupportedMediaOperationError) { + return `${error.operation} is not available on this server.`; + } + if (error instanceof MediaToolMissingError) return "ffmpeg is not installed on the media worker."; + return "Rendering failed. Check the plan and try again."; +} + +function renderEvent(render: MediaRender) { + return { + type: "media.render" as const, + renderId: render.id, + projectId: render.projectId, + status: render.status, + progress: render.progress, + }; +} + +async function publishRender(deps: MediaServiceDeps, spaceId: string, render: MediaRender) { + try { + await deps.publish(spaceId, renderEvent(render)); + } catch (error) { + getLogger().warn("media render event not published", { + renderId: render.id, + reason: error instanceof Error ? error.message : String(error), + }); + } +} + +// --------------------------------------------------------------------------- +// Service +// --------------------------------------------------------------------------- + +export function createMediaRenderService( + deps: MediaServiceDeps, + options: { running?: RunningMediaRenders } = {}, +): MediaRenderService { + const running = options.running ?? runningRenders; + return { + async start(actor, input) { + const plan = await deps.repos.getPlan(actor.spaceId, input.planId); + const { profiles } = await deps.profiles(); + if (!profiles.has(plan.profileId)) { + throw new InvalidMediaInputError("The plan's platform profile is not available"); + } + const render = await deps.repos.createRender(actor.spaceId, { + planId: plan.id, + quality: input.quality, + createdBy: actor.userId, + }); + await deps.jobs.enqueue(mediaRenderJob(render.id)); + await publishRender(deps, actor.spaceId, render); + return render; + }, + + async cancel(actor, renderId) { + const render = await deps.repos.cancelRender(actor.spaceId, renderId); + running.get(renderId)?.abort(); + await deps.jobs.cancel(mediaRenderJob(renderId).replaceKey!).catch((error: unknown) => { + getLogger().warn("media render job not dequeued", { + renderId, + reason: error instanceof Error ? error.message : String(error), + }); + }); + await publishRender(deps, actor.spaceId, render); + return render; + }, + + async outputPath(actor, renderId, fileName) { + const record = await deps.repos.getRenderRecord(actor.spaceId, renderId); + if (record.status !== "done") return null; + if (path.basename(fileName) !== fileName || fileName.startsWith(".")) return null; + const outputs = Array.isArray(record.outputs) ? record.outputs : []; + const recorded = outputs.some( + (output) => + output !== null && + typeof output === "object" && + !Array.isArray(output) && + output.fileName === fileName, + ); + if (!recorded) return null; + // Derive the location instead of trusting a stored path. + const file = await resolveInside( + deps.mediaRoot, + `${actor.spaceId}/renders/${renderId}/${fileName}`, + ); + const found = await stat(file).catch(() => null); + return found?.isFile() ? file : null; + }, + }; +} + +// --------------------------------------------------------------------------- +// Job handler +// --------------------------------------------------------------------------- + +interface BrandFiles { + /** Font artifact id → file. */ + fonts: Map; + fontList: Array<{ family: string; path: string }>; + logos: string[]; + watermark: string | null; + lut: string | null; +} + +function safeExtension(name: string, mimeType: string): string { + const fromName = path.extname(name).toLowerCase(); + if (/^\.[a-z0-9]{1,5}$/.test(fromName)) return fromName; + return EXTENSION_BY_MIME[mimeType.toLowerCase()] ?? ".bin"; +} + +const even = (value: number) => Math.max(2, Math.floor(value / 2) * 2); + +function finiteNumber(value: unknown): value is number { + return typeof value === "number" && Number.isFinite(value); +} + +/** Transcript words from a stored TranscribeResult. */ +export function transcriptWords(data: unknown): CaptionWord[] { + const words = (data as { words?: unknown } | null)?.words; + if (!Array.isArray(words)) return []; + return words.flatMap((word) => { + const w = word as { startMs?: unknown; endMs?: unknown; text?: unknown }; + if (!finiteNumber(w.startMs) || !finiteNumber(w.endMs) || typeof w.text !== "string") { + return []; + } + return [{ startMs: w.startMs, endMs: w.endMs, text: w.text }]; + }); +} + +type Subject = { tMs: number; cx: number; cy: number; strength: number }; + +/** Reframe subjects from a stored saliency result, or face centres from detections. */ +export function subjectPoints(saliency: unknown, faces: unknown): Subject[] { + const points = (saliency as { points?: unknown } | null)?.points; + if (Array.isArray(points)) { + const parsed = points.flatMap((point) => { + const p = point as Partial; + return finiteNumber(p.tMs) && + finiteNumber(p.cx) && + finiteNumber(p.cy) && + finiteNumber(p.strength) + ? [{ tMs: p.tMs, cx: p.cx, cy: p.cy, strength: p.strength }] + : []; + }); + if (parsed.length > 0) return parsed; + } + const frames = (faces as { frames?: unknown } | null)?.frames; + if (!Array.isArray(frames)) return []; + return frames.flatMap((frame) => { + const f = frame as { tMs?: unknown; faces?: unknown }; + if (!finiteNumber(f.tMs) || !Array.isArray(f.faces)) return []; + const tMs = f.tMs; + return f.faces.flatMap((face) => { + const box = (face as { box?: { x?: unknown; y?: unknown; w?: unknown; h?: unknown } }).box; + const confidence = (face as { confidence?: unknown }).confidence; + if (!box || !finiteNumber(box.x) || !finiteNumber(box.y)) return []; + if (!finiteNumber(box.w) || !finiteNumber(box.h)) return []; + return [ + { + tMs, + cx: box.x + box.w / 2, + cy: box.y + box.h / 2, + strength: finiteNumber(confidence) ? confidence : 1, + }, + ]; + }); + }); +} + +export interface MediaRenderHandlerOptions { + running?: RunningMediaRenders; + /** How often a running render checks whether another process canceled it. */ + cancelPollMs?: number; +} + +/** + * Worker handler for `media.render`. Never throws for a render failure: the + * render record carries the outcome, so a job retry has nothing to redo. + */ +export function createMediaRenderHandler( + deps: MediaServiceDeps, + privacy: MediaPrivacyService, + options: MediaRenderHandlerOptions = {}, +): MediaJobHandlers["media.render"] { + const running = options.running ?? runningRenders; + const pollMs = options.cancelPollMs ?? CANCEL_POLL_MS; + const { repos } = deps; + + return async ({ renderId }) => { + const spaceId = await repos.mediaSpaceOf("render", renderId); + if (!spaceId) return; + const claimed = await repos.claimRender(spaceId, renderId); + if (!claimed) return; + await publishRender(deps, spaceId, claimed); + + const controller = new AbortController(); + running.set(renderId, controller); + const poll = setInterval(() => { + repos + .getRender(spaceId, renderId) + .then((render) => { + if (render.status === "canceled") controller.abort(); + }) + .catch(() => undefined); + }, pollMs); + poll.unref?.(); + const createdArtifacts: Array<{ id: string; storageKey: string }> = []; + let renderDir: string | null = null; + let userId = ""; + + try { + renderDir = await resolveInside(deps.mediaRoot, `${spaceId}/renders/${renderId}`); + const record = await repos.getRenderRecord(spaceId, renderId); + userId = record.createdBy; + await runRender({ + deps, + privacy, + spaceId, + renderId, + renderDir, + record, + signal: controller.signal, + abort: () => controller.abort(), + createdArtifacts, + }); + } catch (error) { + if (renderDir) await rm(renderDir, { recursive: true, force: true }).catch(() => undefined); + await removeArtifacts(deps, spaceId, userId, renderId, createdArtifacts); + const current = await repos.getRender(spaceId, renderId).catch(() => null); + if (current?.status === "canceled") { + await publishRender(deps, spaceId, current); + return; + } + getLogger().error("media render failed", { + renderId, + reason: error instanceof Error ? error.message : String(error), + }); + const failed = await repos + .updateRender(spaceId, renderId, { status: "failed", error: userMessage(error) }) + .catch(() => null); + if (failed) await publishRender(deps, spaceId, failed); + } finally { + clearInterval(poll); + running.delete(renderId); + } + }; +} + +function artifactContext(spaceId: string, userId: string, renderId: string): AdapterContext { + return { + operationId: `media-render:${renderId}`, + traceId: `media-render:${renderId}`, + spaceId, + userId, + signal: new AbortController().signal, + }; +} + +async function removeArtifacts( + deps: MediaServiceDeps, + spaceId: string, + userId: string, + renderId: string, + created: Array<{ id: string; storageKey: string }>, +) { + for (const artifact of created) { + await deps.prisma.artifact + .deleteMany({ where: { id: artifact.id, spaceId } }) + .catch(() => undefined); + await deps.artifacts + .remove(artifact.storageKey, artifactContext(spaceId, userId, renderId)) + .catch(() => undefined); + } +} + +type RenderRecord = Awaited>; + +interface RunInput { + deps: MediaServiceDeps; + privacy: MediaPrivacyService; + spaceId: string; + renderId: string; + renderDir: string; + record: RenderRecord; + signal: AbortSignal; + abort: () => void; + createdArtifacts: Array<{ id: string; storageKey: string }>; +} + +async function runRender(input: RunInput): Promise { + const { deps, privacy, spaceId, renderId, renderDir, record, signal } = input; + const { repos } = deps; + const actor: MediaActor = { spaceId, userId: record.createdBy }; + const context = artifactContext(spaceId, actor.userId, renderId); + + /** Apply a patch while the render is active; a null result means it was canceled. */ + const update = async (patch: MediaRenderPatch) => { + const updated = await repos.updateRender(spaceId, renderId, patch); + if (!updated) { + input.abort(); + throw new RenderCanceled(); + } + return updated; + }; + const checkCanceled = () => { + if (signal.aborted) throw new RenderCanceled(); + }; + + let plan = await repos.getPlan(spaceId, record.planId); + const project = await repos.getProjectRecord(spaceId, record.projectId); + const entry = (await deps.profiles()).profiles.get(record.profileId); + if (!entry) throw new MediaRenderFailure("The platform profile is no longer available."); + const profile = entry.profile; + const kit = project.brandKitId + ? (await repos.getBrandKit(spaceId, project.brandKitId)).kit + : null; + const quality = record.quality as MediaRenderQuality; + + // A retried job starts from a clean work directory. + const workDir = path.join(renderDir, "work"); + await rm(renderDir, { recursive: true, force: true }); + await mkdir(workDir, { recursive: true }); + + const fetchArtifact = async (artifactId: string, dir: string, baseName: string) => { + const row = await deps.prisma.artifact.findFirst({ + where: { id: artifactId, spaceId }, + select: { storageKey: true, name: true, mimeType: true }, + }); + if (!row) return null; + const bytes = await deps.artifacts.get(row.storageKey, context); + await mkdir(dir, { recursive: true }); + const file = path.join(dir, `${baseName}${safeExtension(row.name, row.mimeType)}`); + await writeFile(file, bytes); + return file; + }; + const brand = await loadBrandFiles(kit, path.join(workDir, "brand"), fetchArtifact); + const audioFiles = new Map(); + + const alternatives = await alternativeSegments(deps, spaceId, project, plan.edl); + let round = record.autoFixRound; + let edl = plan.edl; + + for (;;) { + checkCanceled(); + const roundDir = path.join(workDir, `round-${round}`); + await mkdir(roundDir, { recursive: true }); + + const prepared = await privacy.prepareSources({ + actor, + projectId: project.id, + edl, + quality, + workDir: roundDir, + signal, + }); + checkCanceled(); + + for (const clip of edl.tracks.audio) { + if (audioFiles.has(clip.artifactId)) continue; + const file = await fetchArtifact( + clip.artifactId, + path.join(workDir, "audio"), + `audio-${audioFiles.size + 1}`, + ); + if (!file) throw new MediaRenderFailure("An audio file of the plan is no longer available."); + audioFiles.set(clip.artifactId, file); + } + + const motionClips = await renderMotionClips(deps, edl, kit, brand, roundDir, signal); + const { transcripts, subjects } = await analysisInputs(deps, spaceId, edl); + const videoEncoder = await pickEncoder(deps, profile, quality); + + const renderPlan = compileRenderPlan({ + edl, + profile, + brandKit: kit, + quality, + workDir: roundDir, + sources: prepared.sources, + transcripts, + subjects, + motionClips, + files: { + lutPath: brand.lut, + watermarkPath: brand.watermark, + fonts: brand.fonts, + audio: audioFiles, + logoPath: brand.logos[0] ?? null, + }, + videoEncoder, + }); + + await executePlan(input, renderPlan, update); + + const video = renderPlan.outputs.find((output) => output.kind === "video"); + if (!video) throw new Error("render plan has no video output"); + const ffprobe = await deps.renderer.probe(video.path, { measureLoudness: true, signal }); + if (findMetadataLeaks(ffprobe.tags).length > 0) { + // Spec 5: a render whose metadata could not be removed is failed, not gated. + throw new MediaRenderFailure("The export still carried source metadata."); + } + const gating = await update({ status: "gating", progress: 0.9, ffprobe }); + await publishRender(deps, spaceId, gating); + + const aiAlterations = [ + ...new Set([...edl.metadata.aiAlterations, ...prepared.aiAlterations]), + ]; + const rendered: Edl = { + ...edl, + metadata: { + ...edl.metadata, + aiAlterations, + aiAltered: edl.metadata.aiAltered || aiAlterations.length > 0, + }, + }; + const report = await privacy.gate({ + actor, + projectId: project.id, + edl: rendered, + profile, + outputPath: video.path, + ffprobe, + copy: { + title: edl.metadata.title ?? null, + caption: edl.metadata.description ?? null, + tags: edl.metadata.tags, + }, + // The publish bundle always carries the AI disclosure file and checklist + // entry when the profile requires a label and the output is altered. + bundleHasDisclosure: profile.disclosure.aiAlteredLabelRequired && aiAlterations.length > 0, + autoFixRound: round, + signal, + }); + checkCanceled(); + + if (report.passed) { + await finish(input, renderPlan, report, plan.id, update, context); + return; + } + + const fix = + round < MAX_AUTO_FIX_ROUNDS + ? planAutoFix(edl, { ...report, autoFixRound: round }, alternatives) + : null; + if (!fix) { + await rm(renderDir, { recursive: true, force: true }); + const blocked = await update({ status: "blocked", progress: 1, gateReport: report }); + await publishRender(deps, spaceId, blocked); + return; + } + const codes = [ + ...new Set(report.issues.filter((i) => i.severity === "block").map((i) => i.code)), + ]; + plan = await repos.createPlanVersion(spaceId, { + projectId: project.id, + profileId: plan.profileId, + edl: fix.edl, + createdBy: "agent", + rationale: `Auto-fix ${round + 1}/${MAX_AUTO_FIX_ROUNDS} after the gate blocked (${codes.join( + ", ", + )}): ${fix.changes.join("; ")}`, + }); + round += 1; + edl = plan.edl; + await rm(roundDir, { recursive: true, force: true }); + const retry = await update({ + status: "rendering", + progress: 0, + autoFixRound: round, + gateReport: report, + ffprobe: null, + }); + await publishRender(deps, spaceId, retry); + } +} + +async function loadBrandFiles( + kit: BrandKit | null, + dir: string, + fetchArtifact: (artifactId: string, dir: string, baseName: string) => Promise, +): Promise { + const files: BrandFiles = { + fonts: new Map(), + fontList: [], + logos: [], + watermark: null, + lut: null, + }; + if (!kit) return files; + // Missing files are skipped here; the compiler fails only if the plan needs one. + const missing = (what: string) => getLogger().warn("brand kit file unavailable", { what }); + for (const [index, font] of kit.fonts.entries()) { + const file = await fetchArtifact(font.artifactId, dir, `font-${index + 1}`); + if (!file) { + missing("font"); + continue; + } + files.fonts.set(font.artifactId, file); + files.fontList.push({ family: font.family, path: file }); + } + for (const [index, logo] of kit.logoArtifactIds.entries()) { + const file = await fetchArtifact(logo, dir, `logo-${index + 1}`); + if (file) files.logos.push(file); + else missing("logo"); + } + const watermarkId = kit.watermark?.artifactId; + if (watermarkId) { + files.watermark = await fetchArtifact(watermarkId, dir, "watermark"); + if (!files.watermark) missing("watermark"); + } else if (kit.watermark) { + files.watermark = files.logos[0] ?? null; + } + if (kit.lutArtifactId) { + files.lut = await fetchArtifact(kit.lutArtifactId, dir, "look"); + if (!files.lut) missing("lut"); + } + return files; +} + +async function renderMotionClips( + deps: MediaServiceDeps, + edl: Edl, + kit: BrandKit | null, + brand: BrandFiles, + dir: string, + signal: AbortSignal, +): Promise> { + const W = edl.output.width; + const H = edl.output.height; + const jobs: Array<{ key: string; template: NonNullable; w: number; h: number }> = + []; + if (edl.intro && edl.intro.durationMs > 0) + jobs.push({ key: "intro", template: edl.intro, w: W, h: H }); + if (edl.outro && edl.outro.durationMs > 0) + jobs.push({ key: "outro", template: edl.outro, w: W, h: H }); + for (const overlay of edl.overlays) { + if (!overlay.motion) continue; + const box = overlay.box; + jobs.push({ + key: overlay.id, + template: overlay.motion, + w: box ? even(box.w * W) : W, + h: box ? even(box.h * H) : H, + }); + } + const clips = new Map(); + if (jobs.length === 0) return clips; + if (!kit) throw new MediaRenderFailure("Motion templates need a brand kit on the project."); + for (const job of jobs) { + // Keys are SafeId values ([a-z0-9_-]), so they are safe file names. + const result = await deps.motion.render({ + template: job.template, + brandKit: kit, + assets: { fonts: brand.fontList, logoPaths: brand.logos }, + width: job.w, + height: job.h, + fps: edl.output.fps, + outPath: path.join(dir, `motion-${job.key}.mov`), + signal, + }); + clips.set(job.key, result.path); + } + return clips; +} + +async function analysisInputs(deps: MediaServiceDeps, spaceId: string, edl: Edl) { + const transcripts = new Map(); + const subjects = new Map(); + const assetIds = [...new Set(edl.tracks.video.map((clip) => clip.assetId))]; + for (const assetId of assetIds) { + const [transcript, saliency, faces] = await Promise.all([ + deps.repos.latestAnalysis(spaceId, assetId, "transcript"), + deps.repos.latestAnalysis(spaceId, assetId, "saliency"), + deps.repos.latestAnalysis(spaceId, assetId, "faces"), + ]); + const words = transcriptWords(transcript?.data ?? null); + if (words.length > 0) transcripts.set(assetId, words); + const points = subjectPoints(saliency?.data ?? null, faces?.data ?? null); + if (points.length > 0) subjects.set(assetId, points); + } + return { transcripts, subjects }; +} + +/** Final renders prefer a working hardware encoder of the profile's codec; drafts use software. */ +async function pickEncoder( + deps: MediaServiceDeps, + profile: PlatformProfile, + quality: MediaRenderQuality, +): Promise { + const family = profile.video.codec; + const software = family === "h264" ? "libx264" : "libx265"; + if (quality === "draft") return software; + const hardware = await deps.renderer.hardwareEncoders(); + return hardware.find((encoder) => encoder.startsWith(`${family}_`)) ?? software; +} + +/** Segments the auto-fix may swap in: the project's material, never paid-only. */ +async function alternativeSegments( + deps: MediaServiceDeps, + spaceId: string, + project: { assetIds: string[]; collectionIds: string[] }, + edl: Edl, +) { + const fromCollections = await deps.repos.collectionAssetIds(spaceId, project.collectionIds); + const assetIds = [ + ...new Set([ + ...project.assetIds, + ...fromCollections, + ...edl.tracks.video.map((clip) => clip.assetId), + ]), + ]; + const segments = await deps.repos.listSegments(spaceId, assetIds); + return segments + .filter((segment) => !segment.paidOnly) + .map((segment) => ({ + segmentId: segment.id, + assetId: segment.assetId, + startMs: segment.startMs, + endMs: segment.endMs, + })); +} + +async function executePlan( + input: RunInput, + plan: RenderPlan, + update: (patch: MediaRenderPatch) => Promise, +) { + const { deps, spaceId, signal } = input; + let lastWrite = Number.NEGATIVE_INFINITY; + let writing: Promise | null = null; + let failure: unknown = null; + const onProgress = (progress: MediaRenderProgress) => { + const now = deps.now().getTime(); + if (writing || now - lastWrite < PROGRESS_INTERVAL_MS) return; + lastWrite = now; + writing = update({ progress: progress.progress * RENDER_SHARE }) + .then((render) => publishRender(deps, spaceId, render)) + .catch((error: unknown) => { + failure ??= error; + }) + .finally(() => { + writing = null; + }); + }; + try { + await deps.renderer.render(plan, { signal, onProgress }); + } finally { + await writing; + } + if (failure) throw failure; + if (signal.aborted) throw new RenderCanceled(); +} + +async function finish( + input: RunInput, + plan: RenderPlan, + report: GateReport, + planId: string, + update: (patch: MediaRenderPatch) => Promise, + context: AdapterContext, +) { + const { deps, spaceId, renderId, renderDir } = input; + const outputs: Array = []; + for (const output of plan.outputs) { + const fileName = OUTPUT_FILES[output.kind]; + const relative = `${spaceId}/renders/${renderId}/${fileName}`; + const destination = await resolveInside(deps.mediaRoot, relative); + await rename(output.path, destination); + const { size } = await stat(destination); + let artifactId: string | null = null; + if (output.kind !== "video") { + artifactId = await registerArtifact(input, context, { + name: fileName, + mimeType: OUTPUT_MIME[output.kind], + file: destination, + }); + } + outputs.push({ + kind: output.kind, + artifactId, + fileName, + sizeBytes: size, + path: relative, + planId, + }); + } + const done = await update({ + status: "done", + progress: 1, + outputs, + gateReport: report, + error: null, + }); + await rm(path.join(renderDir, "work"), { recursive: true, force: true }); + await publishRender(deps, spaceId, done); +} + +async function registerArtifact( + input: RunInput, + context: AdapterContext, + file: { name: string; mimeType: string; file: string }, +): Promise { + const { deps, spaceId } = input; + const bytes = new Uint8Array(await readFile(file.file)); + const stored = await deps.artifacts.put( + { name: file.name, mimeType: file.mimeType, bytes }, + context, + ); + try { + const row = await deps.prisma.artifact.create({ + data: { + spaceId, + userId: context.userId, + name: file.name, + mimeType: file.mimeType, + size: bytes.byteLength, + hash: createHash("sha256").update(bytes).digest("hex"), + storageKey: stored.id, + }, + select: { id: true }, + }); + input.createdArtifacts.push({ id: row.id, storageKey: stored.id }); + return row.id; + } catch (error) { + await deps.artifacts.remove(stored.id, context).catch(() => undefined); + throw error; + } +}