feat: video direct-to-R2 ingest — browser seals, worker probes off R2
POST /video/direct/init|complete: the browser seals (SPENC2) and PUTs straight to object storage; the api HEAD-checks, registers the unprobed asset and queues a probe job — video bytes never cross the api box. Worker side: produce_job_claim widened to probe jobs (042), the runner gains a probe lane (sealed fetch → SPENC1/SPENC2 open by magic → scratch → ffprobe → metadata write-back → complete 100), worker storage reads SPENC2 frames via DIRECT_EC_PRIVATE_KEY (inherited through compose appenv), and a ported ffprobe util. Probe rows carrying input_json fail instead of stranding; the claim guard no longer strands probe rows as invalid.
This commit is contained in:
parent
a7c3f26a4f
commit
1429e4e9af
15 changed files with 1325 additions and 90 deletions
|
|
@ -27,7 +27,13 @@ import { createHash } from "node:crypto";
|
|||
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";
|
||||
/**
|
||||
* Ingest probe jobs (039's default kind) are claimed by the worker too: the
|
||||
* direct-to-R2 path queues one and the worker ffprobes the sealed object off
|
||||
* R2, so the api box never touches video bytes. parseProduceRequest still only
|
||||
* accepts compose/mix payloads — a probe row carries no input_json at all.
|
||||
*/
|
||||
export type ProduceJobKind = "probe" | "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";
|
||||
|
|
@ -186,7 +192,7 @@ function gainDbChecked(value: unknown): number {
|
|||
}
|
||||
|
||||
/** Strict allowlist parse of the API's persisted `input_json`; the worker re-validates it. */
|
||||
export function parseProduceRequest(value: unknown, kind: ProduceJobKind): ProduceRequest {
|
||||
export function parseProduceRequest(value: unknown, kind: Exclude<ProduceJobKind, "probe">): ProduceRequest {
|
||||
if (typeof value !== "object" || value === null || Array.isArray(value)) fail("input_json must be an object");
|
||||
onlyKeys(value, ["overlays", "voiceover", "outputLabel", "lane", "requestedBy"], "input_json");
|
||||
const raw = value as Record<string, unknown>;
|
||||
|
|
|
|||
166
apps/media-worker/src/producer/ffprobe.ts
Normal file
166
apps/media-worker/src/producer/ffprobe.ts
Normal file
|
|
@ -0,0 +1,166 @@
|
|||
/**
|
||||
* ffprobe for the worker probe lane — PORTED VERBATIM from
|
||||
* apps/api/src/video/ffprobe.ts (buildFfprobeArgs / probeVideoFile /
|
||||
* parseFfprobeOutput and the helpers beneath). The api keeps its own copy: the
|
||||
* worker package must not depend on the api module graph, and the two parsers
|
||||
* must agree bit for bit on the metadata written to video_assets.
|
||||
*/
|
||||
import { spawn } from "node:child_process";
|
||||
|
||||
const FFPROBE_TIMEOUT_MS = 30_000;
|
||||
const MAX_OUTPUT_BYTES = 10 * 1024 * 1024;
|
||||
|
||||
export interface VideoProbeResult {
|
||||
width?: number;
|
||||
height?: number;
|
||||
durationMs?: number;
|
||||
frameRate?: number;
|
||||
codec?: string;
|
||||
audioCodec?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Keep the input as a discrete argv value. `-i` ensures even a path beginning
|
||||
* with a dash is treated as input rather than another ffprobe option.
|
||||
*/
|
||||
export function buildFfprobeArgs(inputPath: string): string[] {
|
||||
return [
|
||||
"-v",
|
||||
"error",
|
||||
"-print_format",
|
||||
"json",
|
||||
"-show_format",
|
||||
"-show_streams",
|
||||
"-i",
|
||||
inputPath,
|
||||
];
|
||||
}
|
||||
|
||||
export async function probeVideoFile(
|
||||
inputPath: string,
|
||||
executable = process.env.FFPROBE_PATH ?? "ffprobe",
|
||||
): Promise<VideoProbeResult> {
|
||||
if (inputPath.length === 0) throw new Error("ffprobe input path must not be empty");
|
||||
|
||||
const output = await new Promise<string>((resolve, reject) => {
|
||||
const child = spawn(executable, buildFfprobeArgs(inputPath), {
|
||||
shell: false,
|
||||
windowsHide: true,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
const stdout: Buffer[] = [];
|
||||
const stderr: Buffer[] = [];
|
||||
let outputBytes = 0;
|
||||
let settled = false;
|
||||
|
||||
const finish = (error: Error | undefined, value?: string): void => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
if (error === undefined) resolve(value ?? "");
|
||||
else reject(error);
|
||||
};
|
||||
|
||||
const timeout = setTimeout(() => {
|
||||
child.kill("SIGKILL");
|
||||
finish(new Error(`ffprobe timed out after ${FFPROBE_TIMEOUT_MS}ms`));
|
||||
}, FFPROBE_TIMEOUT_MS);
|
||||
|
||||
child.stdout.on("data", (chunk: Buffer) => {
|
||||
outputBytes += chunk.length;
|
||||
if (outputBytes > MAX_OUTPUT_BYTES) {
|
||||
child.kill("SIGKILL");
|
||||
finish(new Error("ffprobe output exceeded the 10 MiB limit"));
|
||||
return;
|
||||
}
|
||||
stdout.push(chunk);
|
||||
});
|
||||
child.stderr.on("data", (chunk: Buffer) => {
|
||||
if (Buffer.concat(stderr).length < 16_384) stderr.push(chunk);
|
||||
});
|
||||
child.once("error", (error) => finish(error));
|
||||
child.once("close", (code, signal) => {
|
||||
if (code !== 0) {
|
||||
const detail = Buffer.concat(stderr).toString("utf8").trim();
|
||||
finish(
|
||||
new Error(
|
||||
`ffprobe failed (${signal === null ? `exit ${String(code)}` : `signal ${signal}`})${
|
||||
detail === "" ? "" : `: ${detail}`
|
||||
}`,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
finish(undefined, Buffer.concat(stdout).toString("utf8"));
|
||||
});
|
||||
});
|
||||
|
||||
return parseFfprobeOutput(output);
|
||||
}
|
||||
|
||||
export function parseFfprobeOutput(output: string): VideoProbeResult {
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(output) as unknown;
|
||||
} catch {
|
||||
throw new Error("ffprobe returned invalid JSON");
|
||||
}
|
||||
const root = asRecord(parsed);
|
||||
if (root === undefined) throw new Error("ffprobe returned an invalid result");
|
||||
const streams = Array.isArray(root.streams)
|
||||
? root.streams.map(asRecord).filter((stream): stream is Record<string, unknown> => stream !== undefined)
|
||||
: [];
|
||||
const format = asRecord(root.format);
|
||||
const video = streams.find((stream) => stream.codec_type === "video");
|
||||
if (video === undefined) throw new Error("ffprobe found no video stream");
|
||||
const audio = streams.find((stream) => stream.codec_type === "audio");
|
||||
const durationSeconds = finiteNumber(format?.duration) ?? finiteNumber(video?.duration);
|
||||
const frameRate = rationalNumber(video?.avg_frame_rate) ?? rationalNumber(video?.r_frame_rate);
|
||||
|
||||
return {
|
||||
...optionalNumber("width", finiteInteger(video?.width)),
|
||||
...optionalNumber("height", finiteInteger(video?.height)),
|
||||
...optionalNumber(
|
||||
"durationMs",
|
||||
durationSeconds === undefined ? undefined : Math.max(0, Math.round(durationSeconds * 1000)),
|
||||
),
|
||||
...optionalNumber("frameRate", frameRate),
|
||||
...optionalString("codec", video?.codec_name),
|
||||
...optionalString("audioCodec", audio?.codec_name),
|
||||
};
|
||||
}
|
||||
|
||||
function asRecord(value: unknown): Record<string, unknown> | undefined {
|
||||
if (typeof value !== "object" || value === null || Array.isArray(value)) return undefined;
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
function finiteNumber(value: unknown): number | undefined {
|
||||
const number = typeof value === "number" ? value : typeof value === "string" ? Number(value) : NaN;
|
||||
return Number.isFinite(number) ? number : undefined;
|
||||
}
|
||||
|
||||
function finiteInteger(value: unknown): number | undefined {
|
||||
const number = finiteNumber(value);
|
||||
return number !== undefined && Number.isInteger(number) ? number : undefined;
|
||||
}
|
||||
|
||||
function rationalNumber(value: unknown): number | undefined {
|
||||
if (typeof value !== "string") return finiteNumber(value);
|
||||
const [numeratorText, denominatorText, extra] = value.split("/");
|
||||
if (extra !== undefined || numeratorText === undefined || denominatorText === undefined) {
|
||||
return finiteNumber(value);
|
||||
}
|
||||
const numerator = Number(numeratorText);
|
||||
const denominator = Number(denominatorText);
|
||||
if (!Number.isFinite(numerator) || !Number.isFinite(denominator) || denominator === 0) return undefined;
|
||||
return numerator / denominator;
|
||||
}
|
||||
|
||||
function optionalNumber<K extends string>(key: K, value: number | undefined): Partial<Record<K, number>> {
|
||||
return value === undefined ? {} : { [key]: value } as Record<K, number>;
|
||||
}
|
||||
|
||||
function optionalString<K extends string>(key: K, value: unknown): Partial<Record<K, string>> {
|
||||
return typeof value !== "string" || value === "" ? {} : { [key]: value } as Record<K, string>;
|
||||
}
|
||||
|
|
@ -107,6 +107,15 @@ function claimRow(overrides: Record<string, unknown> = {}): Record<string, unkno
|
|||
};
|
||||
}
|
||||
|
||||
/** 039's ingest row: kind 'probe', input_json NULL — the direct-to-R2 queue. */
|
||||
function probeRow(overrides: Record<string, unknown> = {}): Record<string, unknown> {
|
||||
return {
|
||||
id: JOB, workspace_id: WS, kind: "probe", input_json: null,
|
||||
video_asset_id: ASSET,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function assetRow(): Record<string, unknown> {
|
||||
return {
|
||||
id: ASSET, storage_key: "sha256/" + "a".repeat(64) + ".mp4", mime_type: "video/mp4",
|
||||
|
|
@ -137,7 +146,7 @@ test("claim goes through the definer function with typed lane + lease params", a
|
|||
assert.equal(job!.workspaceId, WS);
|
||||
assert.equal(job!.source.storageKey.startsWith("sha256/"), true);
|
||||
assert.equal(job!.source.hasAudio, true);
|
||||
assert.equal(job!.request.overlays[0]!.kind, "lower_third");
|
||||
assert.equal(job!.request!.overlays[0]!.kind, "lower_third");
|
||||
});
|
||||
|
||||
test("a probed audio-less asset reports hasAudio:false so the runner uses a silent bed", async () => {
|
||||
|
|
@ -306,6 +315,109 @@ test("a mix row's voiceover request survives the claim", async () => {
|
|||
})]);
|
||||
const job: ClaimedProduceJob | null = await storeOf(pool).claim("local");
|
||||
assert.equal(job!.kind, "mix");
|
||||
assert.equal(job!.request.voiceover!.assetId, VOICEOVER);
|
||||
assert.equal(job!.request.voiceover!.duckOriginal, true);
|
||||
assert.equal(job!.request!.voiceover!.assetId, VOICEOVER);
|
||||
assert.equal(job!.request!.voiceover!.duckOriginal, true);
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Probe lane (direct-to-R2 ingest): the worker claims kind 'probe' rows and
|
||||
// ffprobes the sealed source itself, so the api box never touches video bytes.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
test("claim returns a queued probe row with request null and its source asset", async () => {
|
||||
const pool = new FakePool();
|
||||
pool.expectClient({ rows: [assetRow()] });
|
||||
pool.reply([probeRow()]);
|
||||
|
||||
const job = await storeOf(pool).claim("local");
|
||||
|
||||
assert.ok(job !== null, "a probe row is claimable, not skipped");
|
||||
assert.equal(job.kind, "probe");
|
||||
assert.equal(job.request, null, "a probe row carries no input_json to parse");
|
||||
assert.equal(job.jobId, JOB);
|
||||
assert.equal(job.workspaceId, WS);
|
||||
assert.equal(job.source.storageKey, assetRow().storage_key);
|
||||
});
|
||||
|
||||
test("a probe row that somehow carries input_json is failed, never stranded", async () => {
|
||||
const pool = new FakePool();
|
||||
pool.reply([probeRow({ input_json: { overlays: [] } })]);
|
||||
const client = pool.expectClient();
|
||||
|
||||
const job = await storeOf(pool).claim("local");
|
||||
|
||||
assert.equal(job, null);
|
||||
const failing = client.calls.find((c) => /state = 'failed'/.test(c.sql))!;
|
||||
assert.ok(failing !== undefined);
|
||||
assert.match(failing.params[1] as string, /probe row carries input_json/);
|
||||
});
|
||||
|
||||
test("a truly unknown kind is still failed as unknown", async () => {
|
||||
const pool = new FakePool();
|
||||
pool.reply([claimRow({ kind: "transmogrify" })]);
|
||||
const client = pool.expectClient();
|
||||
|
||||
const job = await storeOf(pool).claim("local");
|
||||
|
||||
assert.equal(job, null);
|
||||
const failing = client.calls.find((c) => /state = 'failed'/.test(c.sql))!;
|
||||
assert.match(failing.params[1] as string, /unknown kind/);
|
||||
});
|
||||
|
||||
test("completeProbe writes the ffprobe columns on the source's own row, then completes the job", async () => {
|
||||
const pool = new FakePool();
|
||||
pool.reply([{ workspace_id: WS }]); // produce_job_workspace
|
||||
const client = pool.expectClient(
|
||||
{ rows: [{ storage_key: "direct/" + JOB + ".enc" }] }, // UPDATE video_assets
|
||||
{ rows: [{ id: JOB }] }, // UPDATE video_processing_jobs
|
||||
);
|
||||
|
||||
await storeOf(pool).completeProbe(JOB, ASSET, {
|
||||
width: 1920, height: 1080, durationMs: 61_000, frameRate: 29.97, codec: "h264", audioCodec: "aac",
|
||||
});
|
||||
|
||||
const setLocal = client.calls.find((c) => /set_config\('app\.workspace_id'/.test(c.sql))!;
|
||||
assert.equal(setLocal.params[0], WS, "the metadata write is workspace-scoped");
|
||||
const assetUpdate = client.calls.find((c) => /UPDATE video_assets/.test(c.sql))!;
|
||||
assert.deepEqual(
|
||||
assetUpdate.params.slice(0, 7),
|
||||
[ASSET, 1920, 1080, 61_000, 29.97, "h264", "aac"],
|
||||
"039's column set, in the api's probeVideo order",
|
||||
);
|
||||
assert.match(assetUpdate.sql, /SET width = \$2, height = \$3, duration_ms = \$4/);
|
||||
const jobUpdate = client.calls.find((c) => /UPDATE video_processing_jobs/.test(c.sql))!;
|
||||
assert.equal(jobUpdate.params[0], JOB);
|
||||
assert.equal(jobUpdate.params[1], "direct/" + JOB + ".enc", "output_key points at the retained object");
|
||||
assert.match(jobUpdate.sql, /state = 'complete', progress_percent = 100/);
|
||||
assert.match(client.calls.at(-1)!.sql, /COMMIT/);
|
||||
assert.equal(client.released(), 1);
|
||||
});
|
||||
|
||||
test("completeProbe throws when the source asset is gone (tenant-deleted mid-probe)", async () => {
|
||||
const pool = new FakePool();
|
||||
pool.reply([{ workspace_id: WS }]);
|
||||
pool.expectClient({ rows: [] }); // UPDATE video_assets matches nothing
|
||||
|
||||
await assert.rejects(
|
||||
() => storeOf(pool).completeProbe(JOB, ASSET, {
|
||||
width: null, height: null, durationMs: null, frameRate: null, codec: null, audioCodec: null,
|
||||
}),
|
||||
/source asset .* not found/,
|
||||
);
|
||||
});
|
||||
|
||||
test("completeProbe throws when the job lost its claim", async () => {
|
||||
const pool = new FakePool();
|
||||
pool.reply([{ workspace_id: WS }]);
|
||||
pool.expectClient(
|
||||
{ rows: [{ storage_key: "direct/" + JOB + ".enc" }] },
|
||||
{ rows: [] }, // the job UPDATE matches nothing: the lease was reclaimed
|
||||
);
|
||||
|
||||
await assert.rejects(
|
||||
() => storeOf(pool).completeProbe(JOB, ASSET, {
|
||||
width: null, height: null, durationMs: null, frameRate: null, codec: null, audioCodec: null,
|
||||
}),
|
||||
/lost its claim/,
|
||||
);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -18,9 +18,9 @@
|
|||
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";
|
||||
import type { ClaimedProduceJob, ProduceJobStore, ProduceOutput, ProduceSourceAsset, ProbeResult } from "./store";
|
||||
|
||||
const KINDS: readonly ProduceJobKind[] = ["compose", "mix"];
|
||||
const KINDS: readonly ProduceJobKind[] = ["probe", "compose", "mix"];
|
||||
|
||||
/**
|
||||
* A claimed `transcoding` row older than this is assumed ownerless (the worker
|
||||
|
|
@ -109,24 +109,45 @@ export class PostgresProduceJobStore implements ProduceJobStore {
|
|||
if (claimed.rowCount === 0) return null;
|
||||
const row = claimed.rows[0]!;
|
||||
const workspaceId = uuid(row.workspace_id, "workspace id");
|
||||
const kind = row.kind === "mix" ? "mix" : row.kind === "compose" ? "compose" : null;
|
||||
const kind = row.kind === "mix" ? "mix" : row.kind === "compose" ? "compose" : row.kind === "probe" ? "probe" : null;
|
||||
let request = null;
|
||||
if (kind !== null) {
|
||||
if (kind === "compose" || kind === "mix") {
|
||||
try {
|
||||
request = parseProduceRequest(row.input_json, kind);
|
||||
} catch {
|
||||
request = null;
|
||||
}
|
||||
if (request === null) {
|
||||
// A row the worker cannot honour must FAIL here — never strand in `transcoding`.
|
||||
// A null return (not a throw) commits the failure instead of rolling it back.
|
||||
await this.tenant(workspaceId, (client) =>
|
||||
client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'failed', progress_percent = 100, error = $2, updated_at = now()
|
||||
WHERE id = $1 AND state = 'transcoding'`,
|
||||
[row.id, "produce invalid request"],
|
||||
));
|
||||
return null;
|
||||
}
|
||||
}
|
||||
if (kind === null || request === null) {
|
||||
// A row the worker cannot honour must FAIL here — never strand in `transcoding`.
|
||||
// A null return (not a throw) commits the failure instead of rolling it back.
|
||||
if (kind === null) {
|
||||
await this.tenant(workspaceId, (client) =>
|
||||
client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'failed', progress_percent = 100, error = $2, updated_at = now()
|
||||
WHERE id = $1 AND state = 'transcoding'`,
|
||||
[row.id, kind === null ? "produce job store: unknown kind" : "produce invalid request"],
|
||||
[row.id, "produce job store: unknown kind"],
|
||||
));
|
||||
return null;
|
||||
}
|
||||
if (kind === "probe" && row.input_json !== null) {
|
||||
// Probe rows carry no request payload; input_json means upstream wrote it wrong.
|
||||
await this.tenant(workspaceId, (client) =>
|
||||
client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'failed', progress_percent = 100, error = $2, updated_at = now()
|
||||
WHERE id = $1 AND state = 'transcoding'`,
|
||||
[row.id, "produce job store: probe row carries input_json"],
|
||||
));
|
||||
return null;
|
||||
}
|
||||
|
|
@ -192,6 +213,36 @@ export class PostgresProduceJobStore implements ProduceJobStore {
|
|||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* A probe job's terminal write: the ffprobe metadata onto the source's OWN
|
||||
* video_assets row (039's column set — the api's probeVideo writes exactly
|
||||
* these), then the job to complete 100 with the retained object as its
|
||||
* output. Runs inside the job's tenant transaction like every post-claim
|
||||
* write, so FORCE RLS still binds it as sp_app.
|
||||
*/
|
||||
async completeProbe(jobId: string, sourceAssetId: string, result: ProbeResult): Promise<void> {
|
||||
const workspaceId = await this.workspaceOf(jobId);
|
||||
await this.tenant(workspaceId, async (client) => {
|
||||
const updated = await client.query(
|
||||
`UPDATE video_assets
|
||||
SET width = $2, height = $3, duration_ms = $4, frame_rate = $5,
|
||||
codec = $6, audio_codec = $7
|
||||
WHERE id = $1 RETURNING storage_key`,
|
||||
[sourceAssetId, result.width, result.height, result.durationMs, result.frameRate, result.codec, result.audioCodec],
|
||||
);
|
||||
if (updated.rowCount === 0) throw new Error(`produce job store: source asset ${sourceAssetId} not found`);
|
||||
const completed = await client.query(
|
||||
`UPDATE video_processing_jobs
|
||||
SET state = 'complete', progress_percent = 100,
|
||||
output_key = $2, error = NULL, updated_at = now()
|
||||
WHERE id = $1 AND state = 'transcoding'
|
||||
RETURNING id`,
|
||||
[jobId, updated.rows[0]?.storage_key],
|
||||
);
|
||||
if (completed.rowCount === 0) throw new Error(`produce job store: job ${jobId} lost its claim`);
|
||||
});
|
||||
}
|
||||
|
||||
async fail(jobId: string, reason: string): Promise<void> {
|
||||
await this.tenant(await this.workspaceOf(jobId), async (client) => {
|
||||
await client.query(
|
||||
|
|
|
|||
|
|
@ -87,8 +87,14 @@ function doubles() {
|
|||
const storageKey = assetKeys.get(assetId);
|
||||
return storageKey === undefined ? null : { storageKey, mimeType: "audio/mp4" };
|
||||
},
|
||||
async completeProbe() { /* probe completions are asserted in the pg-store suite */ },
|
||||
};
|
||||
const storage: ProduceObjectStorage = {
|
||||
async getSealedObject(k) {
|
||||
const body = objects.get(k);
|
||||
if (body === undefined) throw new Error("not found");
|
||||
return body;
|
||||
},
|
||||
async getObject(k) {
|
||||
const body = objects.get(k);
|
||||
if (body === undefined) throw new Error("not found");
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
/**
|
||||
* Produce runner: the compose/mix job executor.
|
||||
* Produce runner: the compose/mix job executor, plus the ingest probe lane.
|
||||
*
|
||||
* 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
|
||||
|
|
@ -21,7 +21,8 @@ import {
|
|||
buildProduceArgs, produceError, PRODUCE_PRESET, scratchAssetName,
|
||||
type Lane, type ProducePlan, type ProduceFailure,
|
||||
} from "./command-builder";
|
||||
import type { ClaimedProduceJob, ProduceJobStore, ProduceObjectStorage } from "./store";
|
||||
import type { ClaimedProduceJob, ProduceJobStore, ProduceObjectStorage, ProbeResult } from "./store";
|
||||
import { probeVideoFile, type VideoProbeResult } from "./ffprobe";
|
||||
|
||||
export interface ProduceRunnerDeps {
|
||||
jobs: ProduceJobStore;
|
||||
|
|
@ -39,13 +40,17 @@ export interface ProduceRunnerDeps {
|
|||
encodeTimeoutMs?: number;
|
||||
/** Cap on one download. */
|
||||
downloadTimeoutMs?: number;
|
||||
/** Hard cap on one probe-lane ffprobe run. */
|
||||
probeTimeoutMs?: number;
|
||||
/** Injection for tests; `which ffmpeg`. */
|
||||
ffmpegBinary?: string;
|
||||
ffprobeBinary?: string;
|
||||
scheduler?: { setTimeout(ms: number, action: () => void): { cancel(): void } };
|
||||
/** The probe lane's ffprobe result; defaults to a real ffprobe run. */
|
||||
probeInput?: (inputPath: string) => Promise<VideoProbeResult>;
|
||||
}
|
||||
|
||||
const DEFAULTS = { pollIntervalMs: 2_000, encodeTimeoutMs: 15 * 60_000, downloadTimeoutMs: 120_000 };
|
||||
const DEFAULTS = { pollIntervalMs: 2_000, encodeTimeoutMs: 15 * 60_000, downloadTimeoutMs: 120_000, probeTimeoutMs: 30_000 };
|
||||
const defaultScheduler = {
|
||||
setTimeout(ms: number, action: () => void): { cancel(): void } {
|
||||
const timer = setTimeout(action, ms);
|
||||
|
|
@ -62,6 +67,7 @@ export class ProduceRunner {
|
|||
private readonly pollIntervalMs: number;
|
||||
private readonly encodeTimeoutMs: number;
|
||||
private readonly downloadTimeoutMs: number;
|
||||
private readonly probeTimeoutMs: number;
|
||||
private readonly scheduler: NonNullable<ProduceRunnerDeps["scheduler"]>;
|
||||
private running: Promise<unknown> | null = null;
|
||||
|
||||
|
|
@ -69,6 +75,7 @@ export class ProduceRunner {
|
|||
this.pollIntervalMs = deps.pollIntervalMs ?? DEFAULTS.pollIntervalMs;
|
||||
this.encodeTimeoutMs = deps.encodeTimeoutMs ?? DEFAULTS.encodeTimeoutMs;
|
||||
this.downloadTimeoutMs = deps.downloadTimeoutMs ?? DEFAULTS.downloadTimeoutMs;
|
||||
this.probeTimeoutMs = deps.probeTimeoutMs ?? DEFAULTS.probeTimeoutMs;
|
||||
this.scheduler = deps.scheduler ?? defaultScheduler;
|
||||
}
|
||||
|
||||
|
|
@ -122,6 +129,11 @@ export class ProduceRunner {
|
|||
|
||||
private async run(job: ClaimedProduceJob): Promise<void> {
|
||||
this.deps.onEvent?.({ type: "claimed", jobId: job.jobId });
|
||||
// The probe lane never touches ffmpeg's encoder: it downloads the sealed
|
||||
// source, opens it (SPENC1 or SPENC2 by magic), ffprobes a scratch copy and
|
||||
// writes the metadata back — the api's probeVideo, relocated off the api box
|
||||
// so the direct-to-R2 path never moves video bytes through a request.
|
||||
if (job.kind === "probe") return this.runProbe(job);
|
||||
const directory = await mkdtemp(join(this.deps.scratchRoot, "produce-"));
|
||||
try {
|
||||
const plan = await this.prepare(job, directory);
|
||||
|
|
@ -146,12 +158,86 @@ export class ProduceRunner {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The probe lane: sealed source → scratch file → ffprobe → metadata onto the
|
||||
* source's own video_assets row → job complete 100. Failures mark the job
|
||||
* failed with the ffprobe error, mirroring the api's probeVideo failure path.
|
||||
*/
|
||||
private async runProbe(job: ClaimedProduceJob): Promise<void> {
|
||||
const directory = await mkdtemp(join(this.deps.scratchRoot, "probe-"));
|
||||
try {
|
||||
const bytes = await this.downloadSealed(job.source.storageKey);
|
||||
const inputPath = join(directory, scratchAssetName(job.source.storageKey, "bin"));
|
||||
await writeFile(inputPath, bytes, { mode: 0o600 });
|
||||
await this.deps.jobs.progress(job.jobId, 50);
|
||||
const metadata = await this.probeInput(inputPath);
|
||||
await this.deps.jobs.completeProbe(job.jobId, job.source.id, {
|
||||
width: metadata.width ?? null,
|
||||
height: metadata.height ?? null,
|
||||
durationMs: metadata.durationMs ?? null,
|
||||
frameRate: metadata.frameRate ?? null,
|
||||
codec: metadata.codec ?? null,
|
||||
audioCodec: metadata.audioCodec ?? null,
|
||||
});
|
||||
this.deps.onEvent?.({ type: "complete", jobId: job.jobId, detail: `probe → ${job.source.id}` });
|
||||
} catch (error: unknown) {
|
||||
const reason = this.probeReason(error);
|
||||
this.deps.onStderr(`probe job ${job.jobId}: ${reason}`);
|
||||
await this.deps.jobs.fail(job.jobId, reason).catch((failError: unknown) => {
|
||||
// A failed `fail()` leaves the row in `transcoding`; the claim's
|
||||
// stale-lease sweep requeues it, so say what actually went wrong.
|
||||
this.deps.onEvent?.({ type: "store-error", jobId: job.jobId, detail: errorMessage(failError) });
|
||||
});
|
||||
this.deps.onEvent?.({ type: "failed", jobId: job.jobId, detail: reason });
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch the sealed source. getSealedObject is the raw read (the api's twin);
|
||||
* getObject then picks the frame branch by magic, so both SPENC1 (api-sealed)
|
||||
* and SPENC2 (browser-sealed direct upload) sources probe identically here.
|
||||
*/
|
||||
private async downloadSealed(storageKey: string): Promise<Buffer> {
|
||||
try {
|
||||
return await withTimeout(this.deps.storage.getSealedObject(storageKey), this.downloadTimeoutMs, "download failed");
|
||||
} catch (error: unknown) {
|
||||
if (error instanceof Error && error.name === "download failed") throw error;
|
||||
throw produceError("download failed");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The probe lane's ffprobe: a real ffprobe run unless the deps inject one.
|
||||
* Reuses the api's arg list + parser (apps/api/src/video/ffprobe.ts ported
|
||||
* verbatim into ./ffprobe.ts) so the metadata matches the api's bit for bit.
|
||||
*/
|
||||
private async probeInput(inputPath: string): Promise<VideoProbeResult> {
|
||||
const probe = this.deps.probeInput ?? ((path: string) => probeVideoFile(path, this.deps.ffprobeBinary));
|
||||
return await new Promise<VideoProbeResult>((resolve, reject) => {
|
||||
const timer = setTimeout(() => reject(probeTimeoutError()), this.probeTimeoutMs);
|
||||
probe(inputPath).then(
|
||||
(value) => { clearTimeout(timer); resolve(value); },
|
||||
(error: unknown) => { clearTimeout(timer); reject(error); },
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** Bounded, secret-free probe failure reason: a known name, never stderr or a path. */
|
||||
private probeReason(error: unknown): string {
|
||||
const name = error instanceof Error ? error.name : "";
|
||||
if ((["download failed", "probe timed out"] as readonly string[]).includes(name)) return `probe ${name}`;
|
||||
return `probe failed: ${this.deps.redactor.redact(String(error)).slice(0, 200)}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Download every input the validated request names, and build the plan: the
|
||||
* source bytes plus one file per logo overlay (keyed by content hash) and the
|
||||
* voiceover, all straight into this job's scratch dir.
|
||||
*/
|
||||
private async prepare(job: ClaimedProduceJob, directory: string): Promise<ProducePlan> {
|
||||
if (job.request === null) throw produceError("invalid request", "probe row on the compose lane");
|
||||
const source = await this.download(job.source.storageKey, directory);
|
||||
const logos: { path: string }[] = [];
|
||||
for (const overlay of job.request.overlays) {
|
||||
|
|
@ -282,6 +368,13 @@ function errorMessage(error: unknown): string {
|
|||
return error instanceof Error && error.name ? error.name : "unknown error";
|
||||
}
|
||||
|
||||
/** The probe lane's own timeout error (name is what probeReason matches on). */
|
||||
function probeTimeoutError(): Error {
|
||||
const error = new Error("probe: probe timed out");
|
||||
error.name = "probe timed out";
|
||||
return error;
|
||||
}
|
||||
|
||||
function withTimeout<T>(promise: Promise<T>, ms: number, name: ProduceFailure): Promise<T> {
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const timer = setTimeout(() => reject(produceError(name)), ms);
|
||||
|
|
|
|||
|
|
@ -3,18 +3,31 @@
|
|||
* `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.
|
||||
*
|
||||
* SPENC2 (direct-to-R2): objects under `direct/` are sealed IN THE BROWSER and
|
||||
* PUT straight to R2. The worker opens them with the ECDH share it derives from
|
||||
* DIRECT_EC_PRIVATE_KEY, so the API box never has to touch video bytes at all.
|
||||
* loadDirectPrivateJwk / directWrapKey / unwrapDirectCek / openDirectFrame below
|
||||
* are PORTED VERBATIM from apps/api/src/storage/storage.ts (same env var, same
|
||||
* PKCS8 base64 DER, same HKDF info, same frame offsets) — the two readers must
|
||||
* stay byte-compatible; a divergence breaks the whole direct path silently.
|
||||
*
|
||||
* 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 { createCipheriv, createDecipheriv, createHash, createPrivateKey, createPublicKey, diffieHellman, hkdfSync, randomBytes } from "node:crypto";
|
||||
import { GetObjectCommand, PutObjectCommand, S3Client } from "@aws-sdk/client-s3";
|
||||
import type { ProduceObjectStorage } from "./store";
|
||||
|
||||
const FRAME_MAGIC = Buffer.from("SPENC1");
|
||||
|
||||
/** True when DIRECT_EC_PRIVATE_KEY is present (the produce worker reports it). */
|
||||
export function workerDirectEnabled(env: NodeJS.ProcessEnv = process.env): boolean {
|
||||
return directKeyPairFromEnv(env) !== null;
|
||||
}
|
||||
|
||||
export interface WorkerStorageConfig {
|
||||
endpoint?: string;
|
||||
region?: string;
|
||||
|
|
@ -24,6 +37,8 @@ export interface WorkerStorageConfig {
|
|||
forcePathStyle?: boolean;
|
||||
/** 32-byte KEK; when omitted, the dev mirror of the API's loadMediaKek(). */
|
||||
mediaKek?: Buffer;
|
||||
/** Overrides DIRECT_EC_PRIVATE_KEY (tests). */
|
||||
directPrivateKey?: ReturnType<typeof createPrivateKey>;
|
||||
}
|
||||
|
||||
function loadMediaKek(): Buffer {
|
||||
|
|
@ -94,6 +109,164 @@ function openFrame(sealed: Buffer, kek: Buffer): Buffer {
|
|||
|
||||
function keyForHash(hash: string): string { return `sha256/${hash}.mp4`; }
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// SPENC2 — browser-sealed direct-to-R2 frames
|
||||
// (ported verbatim from apps/api/src/storage/storage.ts; see the file header)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Frame magic for browser-sealed objects (spec: "SPENC2" exactly). */
|
||||
export const DIRECT_FRAME_MAGIC = Buffer.from("SPENC2", "ascii");
|
||||
|
||||
/**
|
||||
* Fixed frame layout, all integers big-endian. Apps/web/src/lib/direct-storage.ts
|
||||
* seals the same shape; the api's reader is the cross-implementation proof target.
|
||||
*
|
||||
* [0..6) magic "SPENC2"
|
||||
* [6..71) ephemeral P-256 public key, uncompressed: 0x04 || X(32) || Y(32)
|
||||
* [71..83) wrapNonce, 12 random bytes
|
||||
* [83..131) wrappedCek: AES-256-GCM(CEK), key = HKDF-SHA256(ECDH, salt=∅,
|
||||
* info="tmbh-direct-cek-wrap", L=32), nonce = wrapNonce → 48 bytes
|
||||
* [131..135) segmentCount, u32
|
||||
* [135..] segments: ctLen u32 | iv 12 | AES-256-GCM ciphertext+tag, no AAD
|
||||
*/
|
||||
const EPHEMERAL_PUBLIC_OFFSET = 6;
|
||||
const WRAP_NONCE_OFFSET = 71;
|
||||
const WRAPPED_CEK_OFFSET = 83;
|
||||
const SEGMENT_COUNT_OFFSET = 131;
|
||||
const SEGMENTS_OFFSET = 135;
|
||||
const WRAPPED_CEK_BYTES = 32 + 16;
|
||||
|
||||
/** HKDF info binding the wrap key to this purpose. ASCII, exact. */
|
||||
const DIRECT_WRAP_INFO = Buffer.from("tmbh-direct-cek-wrap", "ascii");
|
||||
|
||||
/** Segment plaintext ceiling. Mirrors the web side, which seals 10 MiB chunks. */
|
||||
export const DIRECT_SEGMENT_BYTES = 10 * 1024 * 1024;
|
||||
|
||||
export class DirectKeyError extends Error {}
|
||||
|
||||
/** Server direct keypair: P-256 private key from DIRECT_EC_PRIVATE_KEY (base64 DER). */
|
||||
export interface DirectKeyPair {
|
||||
privateKey: ReturnType<typeof createPrivateKey>;
|
||||
/** Compact JWK {kty,crv,x,y} — exactly what the browser's importKey wants. */
|
||||
publicJwk: { kty: "EC"; crv: "P-256"; x: string; y: string };
|
||||
}
|
||||
|
||||
function loadDirectPrivateJwk(env: NodeJS.ProcessEnv): ReturnType<typeof createPrivateKey> | null {
|
||||
const raw = env.DIRECT_EC_PRIVATE_KEY;
|
||||
if (raw === undefined || raw.trim() === "") return null;
|
||||
let der: Buffer;
|
||||
try {
|
||||
der = Buffer.from(raw.trim(), "base64");
|
||||
} catch {
|
||||
throw new DirectKeyError("DIRECT_EC_PRIVATE_KEY must be base64 DER");
|
||||
}
|
||||
try {
|
||||
return createPrivateKey({ key: der, format: "der", type: "pkcs8" });
|
||||
} catch {
|
||||
throw new DirectKeyError("DIRECT_EC_PRIVATE_KEY is not a PKCS8 P-256 private key");
|
||||
}
|
||||
}
|
||||
|
||||
/** Build a DirectKeyPair from an already-loaded private key (config + tests). */
|
||||
function directKeyPairFromPrivateKey(privateKey: ReturnType<typeof createPrivateKey>): DirectKeyPair {
|
||||
const jwk = createPublicKey(privateKey).export({ format: "jwk" });
|
||||
if (jwk.kty !== "EC" || jwk.crv !== "P-256" || typeof jwk.x !== "string" || typeof jwk.y !== "string") {
|
||||
throw new DirectKeyError("direct key is not a P-256 key");
|
||||
}
|
||||
return { privateKey, publicJwk: { kty: "EC", crv: "P-256", x: jwk.x, y: jwk.y } };
|
||||
}
|
||||
|
||||
/** The direct keypair from env, deriving the public key from the private one. */
|
||||
export function directKeyPairFromEnv(env: NodeJS.ProcessEnv = process.env): DirectKeyPair | null {
|
||||
const privateKey = loadDirectPrivateJwk(env);
|
||||
if (privateKey === null) return null;
|
||||
return directKeyPairFromPrivateKey(privateKey);
|
||||
}
|
||||
|
||||
/** ECDH → HKDF-SHA256(salt empty, info "tmbh-direct-cek-wrap") → 32-byte wrap key. */
|
||||
function directWrapKey(privateKey: ReturnType<typeof createPrivateKey>, ephemeralPoint: Buffer): Buffer {
|
||||
const x = ephemeralPoint.subarray(1, 33).toString("base64url");
|
||||
const y = ephemeralPoint.subarray(33, 65).toString("base64url");
|
||||
let peer: ReturnType<typeof createPublicKey>;
|
||||
try {
|
||||
peer = createPublicKey({ key: { kty: "EC", crv: "P-256", x, y }, format: "jwk" });
|
||||
} catch {
|
||||
throw new Error("storage: SPENC2 frame holds a malformed ephemeral key");
|
||||
}
|
||||
const ikm = Buffer.from(diffieHellman({ privateKey, publicKey: peer }));
|
||||
// RFC 5869: an absent salt defaults to HashLen zero bytes; WebCrypto's empty
|
||||
// salt and Node's zero-filled salt produce the identical PRK (verified).
|
||||
return Buffer.from(hkdfSync("sha256", ikm, Buffer.alloc(0), DIRECT_WRAP_INFO, 32));
|
||||
}
|
||||
|
||||
interface DirectHeader {
|
||||
ephemeralPoint: Buffer;
|
||||
wrapNonce: Buffer;
|
||||
wrappedCek: Buffer;
|
||||
segmentCount: number;
|
||||
}
|
||||
|
||||
/** Validate + slice the fixed header. Throws on any truncation or bad shape. */
|
||||
function parseDirectHeader(stored: Buffer): DirectHeader {
|
||||
if (stored.length < SEGMENTS_OFFSET) {
|
||||
throw new Error("storage: SPENC2 frame is truncated");
|
||||
}
|
||||
const ephemeralPoint = stored.subarray(EPHEMERAL_PUBLIC_OFFSET, WRAP_NONCE_OFFSET);
|
||||
if (ephemeralPoint[0] !== 0x04) throw new Error("storage: SPENC2 ephemeral key is not an uncompressed point");
|
||||
const wrapNonce = stored.subarray(WRAP_NONCE_OFFSET, WRAPPED_CEK_OFFSET);
|
||||
const wrappedCek = stored.subarray(WRAPPED_CEK_OFFSET, WRAPPED_CEK_OFFSET + WRAPPED_CEK_BYTES);
|
||||
const segmentCount = stored.readUInt32BE(SEGMENT_COUNT_OFFSET);
|
||||
if (segmentCount === 0 || segmentCount > 4_294_967_295 / DIRECT_SEGMENT_BYTES + 1) {
|
||||
throw new Error("storage: SPENC2 segment count is out of range");
|
||||
}
|
||||
return { ephemeralPoint, wrapNonce, wrappedCek, segmentCount };
|
||||
}
|
||||
|
||||
/** Walk the segment table; yields {ciphertext, iv} in order. */
|
||||
function* directSegments(stored: Buffer, segmentCount: number): Generator<{ iv: Buffer; ciphertext: Buffer }> {
|
||||
let offset = SEGMENTS_OFFSET;
|
||||
for (let index = 0; index < segmentCount; index += 1) {
|
||||
if (offset + 4 + 12 > stored.length) throw new Error("storage: SPENC2 segment table is truncated");
|
||||
const ctLen = stored.readUInt32BE(offset);
|
||||
offset += 4;
|
||||
if (ctLen < 16 || offset + 12 + ctLen > stored.length) throw new Error("storage: SPENC2 segment length is out of range");
|
||||
const iv = stored.subarray(offset, offset + 12);
|
||||
offset += 12;
|
||||
const ciphertext = stored.subarray(offset, offset + ctLen);
|
||||
offset += ctLen;
|
||||
yield { iv, ciphertext };
|
||||
}
|
||||
if (offset !== stored.length) throw new Error("storage: SPENC2 frame has trailing bytes");
|
||||
}
|
||||
|
||||
/** Unwrap the CEK (verifies the wrap GCM tag). */
|
||||
function unwrapDirectCek(stored: Buffer, privateKey: ReturnType<typeof createPrivateKey>): Buffer {
|
||||
const header = parseDirectHeader(stored);
|
||||
return aesGcmOpen(directWrapKey(privateKey, header.ephemeralPoint), header.wrapNonce, header.wrappedCek);
|
||||
}
|
||||
|
||||
/**
|
||||
* Open a SPENC2 frame: unwrap the CEK, then decrypt every segment. Segment
|
||||
* GCM tags are verified individually, so any flipped byte throws. Empty
|
||||
* plaintexts are rejected (the upload floor is 1 byte).
|
||||
*/
|
||||
export function openDirectFrame(stored: Buffer, privateKey: ReturnType<typeof createPrivateKey>): Buffer {
|
||||
const { segmentCount } = parseDirectHeader(stored);
|
||||
const cek = unwrapDirectCek(stored, privateKey);
|
||||
const parts: Buffer[] = [];
|
||||
for (const segment of directSegments(stored, segmentCount)) {
|
||||
parts.push(aesGcmOpen(cek, segment.iv, segment.ciphertext));
|
||||
}
|
||||
const plaintext = Buffer.concat(parts);
|
||||
if (plaintext.length === 0) throw new Error("storage: SPENC2 frame decrypts to zero bytes");
|
||||
return plaintext;
|
||||
}
|
||||
|
||||
/** True when the leading magic says the object is browser-sealed. */
|
||||
function isDirectFrame(stored: Buffer): boolean {
|
||||
return stored.subarray(0, DIRECT_FRAME_MAGIC.length).equals(DIRECT_FRAME_MAGIC);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
|
|
@ -103,10 +276,12 @@ export class WorkerS3Storage implements ProduceObjectStorage {
|
|||
private readonly client: S3Client;
|
||||
private readonly bucket: string;
|
||||
private readonly kek: Buffer;
|
||||
private readonly directKey: DirectKeyPair | null;
|
||||
|
||||
constructor(config: WorkerStorageConfig) {
|
||||
this.bucket = config.bucket ?? "tactical-media-broadcast-hub-media";
|
||||
this.kek = config.mediaKek ?? loadMediaKek();
|
||||
this.directKey = config.directPrivateKey !== undefined ? directKeyPairFromPrivateKey(config.directPrivateKey) : directKeyPairFromEnv();
|
||||
this.client = new S3Client({
|
||||
endpoint: config.endpoint,
|
||||
region: config.region ?? "us-east-1",
|
||||
|
|
@ -118,11 +293,29 @@ export class WorkerS3Storage implements ProduceObjectStorage {
|
|||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Raw still-sealed object bytes — no decrypt, no hash check. The direct path
|
||||
* keeps the CEK header only in the sealed form (the api's getSealedObject
|
||||
* twin), so a caller that needs the frame verbatim reads it here.
|
||||
*/
|
||||
async getSealedObject(key: string): Promise<Buffer> {
|
||||
const res = await this.client.send(new GetObjectCommand({ Bucket: this.bucket, Key: key }));
|
||||
if (res.Body === undefined) throw new Error(`empty body for ${key}`);
|
||||
return Buffer.from(await res.Body.transformToByteArray());
|
||||
}
|
||||
|
||||
/**
|
||||
* Open an object by its leading magic, mirroring the api's getObject: SPENC1
|
||||
* opens under the KEK, SPENC2 (browser-sealed direct upload) opens under the
|
||||
* ECDH share from DIRECT_EC_PRIVATE_KEY. One reader serves the whole bucket.
|
||||
*/
|
||||
async getObject(key: string): Promise<Buffer> {
|
||||
const res = await this.client.send(new GetObjectCommand({ Bucket: this.bucket, Key: key }));
|
||||
if (res.Body === undefined) throw new Error(`empty body for ${key}`);
|
||||
const stored = Buffer.from(await res.Body.transformToByteArray());
|
||||
const plaintext = openFrame(stored, this.kek);
|
||||
const plaintext = isDirectFrame(stored)
|
||||
? this.openDirectStored(stored)
|
||||
: openFrame(stored, this.kek); // raises the magic-mismatch error for junk
|
||||
// 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)) {
|
||||
|
|
@ -133,6 +326,12 @@ export class WorkerS3Storage implements ProduceObjectStorage {
|
|||
return plaintext;
|
||||
}
|
||||
|
||||
/** SPENC2 branch: only a process holding the direct private key can unwrap. */
|
||||
private openDirectStored(stored: Buffer): Buffer {
|
||||
if (this.directKey === null) throw new Error("DIRECT_EC_PRIVATE_KEY is not configured");
|
||||
return openDirectFrame(stored, this.directKey.privateKey);
|
||||
}
|
||||
|
||||
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");
|
||||
|
|
|
|||
|
|
@ -29,7 +29,8 @@ export interface ClaimedProduceJob {
|
|||
jobId: string;
|
||||
workspaceId: string;
|
||||
kind: ProduceJobKind;
|
||||
request: ProduceRequest;
|
||||
/** The validated producer request; null for a probe row (no input_json). */
|
||||
request: ProduceRequest | null;
|
||||
source: ProduceSourceAsset;
|
||||
claimedAt: string;
|
||||
}
|
||||
|
|
@ -44,10 +45,20 @@ export interface ProduceOutput {
|
|||
frameRate: number;
|
||||
}
|
||||
|
||||
/** ffprobe's write-back for a probe job (039's column set, all nullable). */
|
||||
export interface ProbeResult {
|
||||
width: number | null;
|
||||
height: number | null;
|
||||
durationMs: number | null;
|
||||
frameRate: number | null;
|
||||
codec: string | null;
|
||||
audioCodec: string | null;
|
||||
}
|
||||
|
||||
export interface ProduceJobStore {
|
||||
/**
|
||||
* Claim exactly one queued compose/mix row for this worker's lane, with lease
|
||||
* semantics: the `queued → transcoding` state bump inside
|
||||
* Claim exactly one queued probe/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.
|
||||
*
|
||||
|
|
@ -62,6 +73,8 @@ export interface ProduceJobStore {
|
|||
progress(jobId: string, percent: number): Promise<void>;
|
||||
/** Insert the derived asset and complete the job pointing at it. */
|
||||
complete(jobId: string, output: ProduceOutput, derivedFrom: string): Promise<string>;
|
||||
/** Write a probe job's ffprobe metadata onto its own video_assets row, then complete. */
|
||||
completeProbe(jobId: string, sourceAssetId: string, result: ProbeResult): Promise<void>;
|
||||
/** Fail a claimed job; `reason` must already be redacted/bounded. */
|
||||
fail(jobId: string, reason: string): Promise<void>;
|
||||
/** Resolve a voiceover asset id to its storage key. */
|
||||
|
|
@ -74,6 +87,9 @@ export interface ProduceJobStore {
|
|||
* through exactly this pair. The adapter lives in `s3-storage.ts`.
|
||||
*/
|
||||
export interface ProduceObjectStorage {
|
||||
/** Opens SPENC1 (KEK) and SPENC2 (direct ECDH) frames by leading magic. */
|
||||
getObject(key: string): Promise<Buffer>;
|
||||
/** Raw still-sealed bytes (the api's getSealedObject twin): the direct frame verbatim. */
|
||||
getSealedObject(key: string): Promise<Buffer>;
|
||||
putContent(body: Buffer, contentType: "video/mp4"): Promise<{ key: string }>;
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue