diff --git a/apps/api/migrations/042_probe_jobs_claimable.sql b/apps/api/migrations/042_probe_jobs_claimable.sql new file mode 100644 index 0000000..33614be --- /dev/null +++ b/apps/api/migrations/042_probe_jobs_claimable.sql @@ -0,0 +1,48 @@ +-- 042: the produce worker claims PROBE jobs too. +-- +-- Direct-to-R2 video uploads register an unprobed asset plus a queued probe +-- job and return immediately — the api never touches the bytes, so it cannot +-- probe inline the way 009's ingest path does. The worker (which can decrypt +-- SPENC2 frames via DIRECT_EC_PRIVATE_KEY) owns the probe from here on. +-- +-- Same body as 040 with the kind filter widened; 'transcoding' remains the +-- lease state (the UI's in-flight set already covers queued/probing/transcoding). + +CREATE OR REPLACE FUNCTION produce_job_claim(p_lane text, p_stale_after interval) +RETURNS TABLE ( + id uuid, + workspace_id uuid, + kind text, + input_json jsonb, + video_asset_id uuid +) +LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ +BEGIN + UPDATE video_processing_jobs stale + SET state = 'queued', progress_percent = 0, error = NULL, updated_at = now() + WHERE stale.state = 'transcoding' + AND stale.updated_at < now() - p_stale_after; + + RETURN QUERY + UPDATE video_processing_jobs j + SET state = 'transcoding', progress_percent = 5, error = NULL, updated_at = now() + WHERE j.id IN ( + SELECT c.id + FROM video_processing_jobs c + WHERE c.kind IN ('probe', 'compose', 'mix') + AND c.state = 'queued' + AND ( + (p_lane = 'remote' AND c.input_json->>'lane' = 'remote') + OR (p_lane <> 'remote' AND c.input_json->>'lane' IS DISTINCT FROM 'remote') + ) + ORDER BY c.created_at, c.id + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + RETURNING j.id, j.workspace_id, j.kind, j.input_json, j.video_asset_id; +END +$$; + +-- Ownership unchanged from 040. +REVOKE ALL ON FUNCTION produce_job_claim(text, interval) FROM PUBLIC; +GRANT EXECUTE ON FUNCTION produce_job_claim(text, interval) TO sp_app; diff --git a/apps/api/src/video/video.controller.ts b/apps/api/src/video/video.controller.ts index 0b35dc6..c34b3f4 100644 --- a/apps/api/src/video/video.controller.ts +++ b/apps/api/src/video/video.controller.ts @@ -21,6 +21,9 @@ import { parseId, parseLimit, requireObject, requireSession, requireString } fro import { RbacGuard, RequirePermission, type AuthenticatedRequest } from "../rbac/rbac.guard.js"; import { VideoService, + type DirectUploadCompleteInput, + type DirectUploadConfig, + type DirectUploadInit, type ProduceRequestResult, type VideoLibraryPage, type VideoStatus, @@ -39,6 +42,7 @@ import { type InitChunkedUploadRequest, } from "./chunked-upload.js"; import { loadMatrixBytes, parseMxcUri } from "./import-sources.js"; +import { isUuid } from "../http/validation.js"; interface VideoFile { buffer: Buffer; @@ -68,6 +72,42 @@ export class VideoController { * MEMORY: reading the assembled file for the seal peaks around 2× file size of * RAM (see chunked-upload.ts) — a 20 GB upload needs a ≥ 48 GB host. */ + /** + * Direct-to-R2 ingest: the BROWSER seals (SPENC2) and PUTs straight to object + * storage, so this box never touches video bytes. The api only mints the + * presigned PUT (init), HEAD-checks + registers + queues a probe job for the + * media-worker (complete). Same ceiling as the chunked path, without the RAM. + * + * GET /video/direct/config capability + the JWK the browser seals against + * POST /video/direct/init presigned PUT for a browser-sealed frame + * POST /video/direct/complete HEAD-checked register + queued probe job + */ + @Get("direct/config") + @RequirePermission("media.view") + async directConfig(): Promise { + return this.video.directUploadConfig(); + } + + @Post("direct/init") + @RequirePermission("media.manage") + async initDirectUpload( + @Req() request: Request & AuthenticatedRequest, + @Body() body: unknown, + ): Promise { + const session = requireSession(request); + return this.video.initDirectUpload(session.workspaceId, requireDirectInit(body)); + } + + @Post("direct/complete") + @RequirePermission("media.manage") + async completeDirectUpload( + @Req() request: Request & AuthenticatedRequest, + @Body() body: unknown, + ): Promise { + const session = requireSession(request); + return this.video.completeDirectUpload(session.workspaceId, parseDirectComplete(body)); + } + @Post("upload/init") @RequirePermission("media.manage") async initChunkedUpload( @@ -201,6 +241,41 @@ export class VideoController { } } +/** POST /video/direct/init body: the browser's pick, bounded before any bytes are sealed. */ +function requireDirectInit(value: unknown): { filename: string; mimeType: string; sizeBytes: number } { + const body = requireObject(value, "body"); + const sizeBytes = body.sizeBytes; + if (typeof sizeBytes !== "number" || !Number.isSafeInteger(sizeBytes) || sizeBytes < 1) { + throw new BadRequestException("sizeBytes must be a positive safe integer"); + } + if (sizeBytes > MAX_CHUNKED_UPLOAD_BYTES) { + throw new BadRequestException("sizeBytes exceeds the 100 GiB chunked upload limit"); + } + return { + filename: requireString(body.filename, "filename"), + mimeType: requireString(body.mimeType, "mimeType"), + sizeBytes, + }; +} + +/** POST /video/direct/complete body: the commit half of the direct handshake. */ +function parseDirectComplete(value: unknown): DirectUploadCompleteInput { + const body = requireObject(value, "body"); + const uploadId = requireString(body.uploadId, "uploadId"); + if (!isUuid(uploadId)) throw new BadRequestException("uploadId must be a UUID"); + const sizeBytes = body.sizeBytes; + if (typeof sizeBytes !== "number" || !Number.isSafeInteger(sizeBytes) || sizeBytes < 1) { + throw new BadRequestException("sizeBytes must be a positive safe integer"); + } + return { + uploadId, + objectKey: requireString(body.objectKey, "objectKey"), + sizeBytes, + mimeType: requireString(body.mimeType, "mimeType"), + filename: requireString(body.filename, "filename"), + }; +} + function parseUploadVideo(value: unknown): UploadVideoDto { const body = requireObject(value, "body"); const sizeBytes = body.sizeBytes; diff --git a/apps/api/src/video/video.service.ts b/apps/api/src/video/video.service.ts index 251c09b..688a1b3 100644 --- a/apps/api/src/video/video.service.ts +++ b/apps/api/src/video/video.service.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -6,7 +7,10 @@ import type { PoolClient } from "pg"; import { appendTenantAudit, type AuditEvent } from "../audit/audit.service.js"; import { withTenant } from "../db/client.js"; import { probeVideoFile, type VideoProbeResult } from "./ffprobe.js"; -import { S3Storage, s3ConfigFromEnv, type StorageService } from "../storage/storage.js"; +import { + S3Storage, s3ConfigFromEnv, type StorageService, + directEnabled, directObjectKey, +} from "../storage/storage.js"; import { type ProduceRequest } from "./produce-request.js"; import { MAX_CHUNKED_UPLOAD_BYTES, MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; @@ -83,6 +87,34 @@ export interface VideoStatus { export type VideoUploadResult = VideoStatus; +/** GET /video/direct/config — the browser's capability check before it seals. */ +export interface DirectUploadConfig { + enabled: boolean; + publicKeyJwk: { kty: "EC"; crv: "P-256"; x: string; y: string } | null; + maxBytes: number; +} + +/** POST /video/direct/init response. Nothing is persisted until complete. */ +export interface DirectUploadInit { + uploadId: string; + objectKey: string; + putUrl: string; +} + +/** POST /video/direct/complete request (HEAD-checked, then registered + queued). */ +export interface DirectUploadCompleteInput { + uploadId: string; + objectKey: string; + sizeBytes: number; + mimeType: string; + filename: string; +} + +/** Presigned PUT lifetime for a browser upload (spec: 900 s). */ +export const DIRECT_PUT_TTL_SECONDS = 900; +const DIRECT_KEY_PREFIX = "direct/"; +const DIRECT_OBJECT_SUFFIX = ".enc"; + type TenantTransaction = (workspaceId: string, fn: (client: PoolClient) => Promise) => Promise; type AuditAppender = (workspaceId: string, event: AuditEvent) => Promise; @@ -177,6 +209,96 @@ export class VideoService { return this.registerAndProbe(workspaceId, stored.key, stored.contentType, stored.sizeBytes); } + // ------------------------------------------------------------------------- + // Direct-to-R2 (browser-sealed SPENC2): the api mints the presign and the + // rows; the payload goes browser → R2 → media-worker. No video bytes ever + // touch this process, so the chunked/whole-body memory ceilings do not apply. + // ------------------------------------------------------------------------- + + /** Whether this deployment can unwrap browser-sealed frames (DIRECT_EC_PRIVATE_KEY). */ + async directUploadConfig(): Promise { + const storage = this.requireStorage(); + const publicKeyJwk = await storage.directPublicJwk(); + return { + enabled: directEnabled(), + publicKeyJwk, + maxBytes: MAX_CHUNKED_UPLOAD_BYTES, + }; + } + + /** + * Mint the presigned PUT for a browser upload. The size is bounded up front + * so an oversized pick fails before any bytes are sealed; nothing is + * persisted — the asset row only appears on complete, so an abandoned PUT + * leaves an orphan object (no row), never a dangling row without an object. + */ + async initDirectUpload( + workspaceId: string, + input: { filename: string; mimeType: string; sizeBytes: number }, + ): Promise { + validateDirectUpload(input); + if ((await this.requireStorage().directPublicJwk()) === null) { + throw new ServiceUnavailableException("DIRECT_EC_PRIVATE_KEY is not configured"); + } + const uploadId = randomUUID(); + const objectKey = directObjectKey(); + const putUrl = await this.requireStorage() + .presignPut(objectKey, input.mimeType, DIRECT_PUT_TTL_SECONDS) + .catch(() => { throw new ServiceUnavailableException("Video storage is unavailable: could not presign the direct PUT"); }); + return { uploadId, objectKey, putUrl }; + } + + /** + * Register a completed direct upload. The object is HEAD-checked (exists, + * exact declared size) BEFORE any row is written, so metadata can never point + * at a missing or partial PUT — the same gate /files/direct/complete uses. + * + * The probe deliberately does NOT run here: the insert queues kind 'probe' + * for the media-worker, which downloads the sealed object off R2, decrypts it + * and runs ffprobe itself. This box never touches video bytes on this path. + */ + async completeDirectUpload(workspaceId: string, input: DirectUploadCompleteInput): Promise { + if (!isUuid(input.uploadId)) throw new BadRequestException("uploadId must be a UUID"); + if (input.objectKey !== `${DIRECT_KEY_PREFIX}${input.uploadId}${DIRECT_OBJECT_SUFFIX}`) { + throw new BadRequestException("objectKey must be direct/.enc"); + } + validateDirectUpload(input); + + const head = await this.requireStorage().headObject(input.objectKey).catch(() => { + throw new ServiceUnavailableException("Video storage is unavailable: could not verify the uploaded object"); + }); + if (!head.exists || head.contentLength !== input.sizeBytes) { + throw new BadRequestException("uploaded object is missing or its size does not match sizeBytes"); + } + + const registered = await this.dependencies.tenantTransaction(workspaceId, async (client) => { + const assetResult = await client.query( + `INSERT INTO video_assets (workspace_id, storage_key, mime_type, size_bytes) + VALUES ($1, $2, $3, $4) RETURNING *`, + [workspaceId, input.objectKey, input.mimeType, input.sizeBytes], + ); + const videoAsset = assetFromRow(requiredRow(assetResult.rows[0], "created video asset")); + // kind 'probe' is 039's default: the media-worker claims it and runs ffprobe. + const jobResult = await client.query( + `INSERT INTO video_processing_jobs (workspace_id, video_asset_id, state, kind) + VALUES ($1, $2, 'queued', 'probe') RETURNING *`, + [workspaceId, videoAsset.id], + ); + return { + videoAsset, + processingJob: jobFromRow(requiredRow(jobResult.rows[0], "created video processing job")), + }; + }); + + await this.audit(workspaceId, "video.uploaded", registered.videoAsset.id, { + mimeType: input.mimeType, + sizeBytes: input.sizeBytes, + uploadPath: "direct", + }); + // The worker probes; status reports the queued job the drawer polls on. + return this.getVideoStatus(registered.videoAsset.id, workspaceId); + } + async uploadVideo( workspaceId: string, storageKey: string, @@ -483,6 +605,22 @@ export class VideoService { } } +/** Shared validation for the direct init/complete payloads (no sniffing: the declared type stands). */ +function validateDirectUpload(input: { filename?: string; mimeType: string; sizeBytes: number }): void { + if (input.filename !== undefined && (input.filename.trim() === "" || input.filename.length > 255)) { + throw new BadRequestException("filename must be 1-255 characters"); + } + if (!input.mimeType.toLowerCase().startsWith("video/")) { + throw new BadRequestException("mimeType must be a video media type"); + } + if (!Number.isSafeInteger(input.sizeBytes) || input.sizeBytes < 1) { + throw new BadRequestException("sizeBytes must be a positive safe integer"); + } + if (input.sizeBytes > MAX_CHUNKED_UPLOAD_BYTES) { + throw new BadRequestException("video exceeds the 100 GiB chunked upload limit"); + } +} + function validateUpload(storageKey: string, mimeType: string, sizeBytes: number): void { if (storageKey.trim() === "") throw new BadRequestException("storageKey must be a non-empty string"); if (!mimeType.toLowerCase().startsWith("video/")) { diff --git a/apps/media-worker/src/producer/command-builder.ts b/apps/media-worker/src/producer/command-builder.ts index 60d8682..3e1f889 100644 --- a/apps/media-worker/src/producer/command-builder.ts +++ b/apps/media-worker/src/producer/command-builder.ts @@ -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): 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; diff --git a/apps/media-worker/src/producer/ffprobe.ts b/apps/media-worker/src/producer/ffprobe.ts new file mode 100644 index 0000000..f25d38c --- /dev/null +++ b/apps/media-worker/src/producer/ffprobe.ts @@ -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 { + if (inputPath.length === 0) throw new Error("ffprobe input path must not be empty"); + + const output = await new Promise((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 => 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 | undefined { + if (typeof value !== "object" || value === null || Array.isArray(value)) return undefined; + return value as Record; +} + +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(key: K, value: number | undefined): Partial> { + return value === undefined ? {} : { [key]: value } as Record; +} + +function optionalString(key: K, value: unknown): Partial> { + return typeof value !== "string" || value === "" ? {} : { [key]: value } as Record; +} diff --git a/apps/media-worker/src/producer/pg-store.test.ts b/apps/media-worker/src/producer/pg-store.test.ts index 65ad864..4d76e90 100644 --- a/apps/media-worker/src/producer/pg-store.test.ts +++ b/apps/media-worker/src/producer/pg-store.test.ts @@ -107,6 +107,15 @@ function claimRow(overrides: Record = {}): Record = {}): Record { + return { + id: JOB, workspace_id: WS, kind: "probe", input_json: null, + video_asset_id: ASSET, + ...overrides, + }; +} + function assetRow(): Record { 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/, + ); }); diff --git a/apps/media-worker/src/producer/pg-store.ts b/apps/media-worker/src/producer/pg-store.ts index ab414b2..01e41c8 100644 --- a/apps/media-worker/src/producer/pg-store.ts +++ b/apps/media-worker/src/producer/pg-store.ts @@ -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 { + 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 { await this.tenant(await this.workspaceOf(jobId), async (client) => { await client.query( diff --git a/apps/media-worker/src/producer/runner.test.ts b/apps/media-worker/src/producer/runner.test.ts index ffbda83..9750365 100644 --- a/apps/media-worker/src/producer/runner.test.ts +++ b/apps/media-worker/src/producer/runner.test.ts @@ -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"); diff --git a/apps/media-worker/src/producer/runner.ts b/apps/media-worker/src/producer/runner.ts index 2f0c4ba..1f86f43 100644 --- a/apps/media-worker/src/producer/runner.ts +++ b/apps/media-worker/src/producer/runner.ts @@ -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; } -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; private running: Promise | 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 { 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 { + 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 { + 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 { + const probe = this.deps.probeInput ?? ((path: string) => probeVideoFile(path, this.deps.ffprobeBinary)); + return await new Promise((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 { + 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(promise: Promise, ms: number, name: ProduceFailure): Promise { return new Promise((resolve, reject) => { const timer = setTimeout(() => reject(produceError(name)), ms); diff --git a/apps/media-worker/src/producer/s3-storage.ts b/apps/media-worker/src/producer/s3-storage.ts index 525e9d2..16851af 100644 --- a/apps/media-worker/src/producer/s3-storage.ts +++ b/apps/media-worker/src/producer/s3-storage.ts @@ -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; } 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; + /** 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 | 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): 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, ephemeralPoint: Buffer): Buffer { + const x = ephemeralPoint.subarray(1, 33).toString("base64url"); + const y = ephemeralPoint.subarray(33, 65).toString("base64url"); + let peer: ReturnType; + 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): 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): 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 { + 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 { 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"); diff --git a/apps/media-worker/src/producer/store.ts b/apps/media-worker/src/producer/store.ts index df8ada9..34b5cf6 100644 --- a/apps/media-worker/src/producer/store.ts +++ b/apps/media-worker/src/producer/store.ts @@ -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; /** Insert the derived asset and complete the job pointing at it. */ complete(jobId: string, output: ProduceOutput, derivedFrom: string): Promise; + /** Write a probe job's ffprobe metadata onto its own video_assets row, then complete. */ + completeProbe(jobId: string, sourceAssetId: string, result: ProbeResult): Promise; /** Fail a claimed job; `reason` must already be redacted/bounded. */ fail(jobId: string, reason: string): Promise; /** Resolve a voiceover asset id to its storage key. */ @@ -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; + /** Raw still-sealed bytes (the api's getSealedObject twin): the direct frame verbatim. */ + getSealedObject(key: string): Promise; putContent(body: Buffer, contentType: "video/mp4"): Promise<{ key: string }>; } diff --git a/apps/web/src/components/video/upload-dropzone.tsx b/apps/web/src/components/video/upload-dropzone.tsx index f257d65..49c0865 100644 --- a/apps/web/src/components/video/upload-dropzone.tsx +++ b/apps/web/src/components/video/upload-dropzone.tsx @@ -5,19 +5,73 @@ import { Button } from "../ui/primitives"; import { Loading } from "../brand/loading"; import { apiUploadFile, errorMessage } from "../../lib/api-client"; import { CHUNK_SIZE_BYTES, uploadVideoChunked } from "../../lib/chunked-upload"; -import type { VideoStatusDto } from "../../lib/video-types"; +import { + directEligible, + fetchVideoDirectConfig, + uploadVideoDirect, +} from "../../lib/video-direct"; +import type { VideoDirectConfigDto, VideoStatusDto } from "../../lib/video-types"; import { formatBytes } from "./format"; /** - * Drag-drop + file-picker target for video ingest. Files at or under one chunk - * keep the single multipart POST /video/upload; larger ones go through the - * chunked API (init → PUT slices → complete) so no single request approaches - * Cloudflare's ~100 MB proxied-body cap. The API stores encrypted bytes, mints - * the content-addressed key, and probes the stored object before responding. - * Progress measures the actual transfer. + * Drag-drop + file-picker target for video ingest — MULTIPLE files at once, + * one upload in flight at a time (sequential queue, per-file status rows). + * + * Per-file transport, first match wins: + * direct — browser seals (SPENC2) and PUTs straight to storage through a + * presigned URL, then POST /video/direct/complete. Chosen when + * the deployment enables it, a public key is served, and the file + * fits both the server cap and the browser memory cap (sealing + * holds plaintext + frame at once). Any failure here falls back + * to the paths below, surfacing the reason as a warning. + * chunked — over one chunk (10 MiB): the existing init → PUT slices → + * complete API path, so no single request approaches the ~100 MB + * proxied-body cap. + * whole — at or under one chunk: the single multipart POST /video/upload. + * + * One file's failure never aborts the queue. */ const MAX_UPLOAD_BYTES = 2 * 1024 * 1024 * 1024; +const MAX_LABEL = "2 GiB"; +/** formatBytes would render the chunk as decimal "10.0 MB"; the slicing is binary. */ +const CHUNK_LABEL = "10 MiB"; + +/** + * EN/ES strings. The Files and Wiki sections each own a provider-backed + * `t({en, es})`; the Video section has none, so this reads the same + * per-browser localStorage preference directly instead of importing the + * Files context (whose provider is not mounted on this route). + */ +const LANG_KEY = "tmbh.files.lang"; + +interface LocalizedText { + readonly en: string; + readonly es: string; +} + +function readLang(): "en" | "es" { + if (typeof window === "undefined") return "en"; + try { + return window.localStorage.getItem(LANG_KEY) === "es" ? "es" : "en"; + } catch { + return "en"; + } +} + +function useLang(): (text: LocalizedText) => string { + const [lang, setLang] = useState<"en" | "es">("en"); + useEffect(() => { + setLang(readLang()); + // Storage events keep two open tabs of the same browser in agreement. + const onStorage = (e: StorageEvent): void => { + if (e.key === LANG_KEY) setLang(readLang()); + }; + window.addEventListener("storage", onStorage); + return () => window.removeEventListener("storage", onStorage); + }, []); + return useCallback((text: LocalizedText) => text[lang], [lang]); +} export interface UploadDraftInfo { key: string; @@ -33,66 +87,185 @@ interface Props { onUploaded: (status: VideoStatusDto) => void; } -export function UploadDropzone({ live, canUpload, onStarted, onSettled, onUploaded }: Props): React.ReactElement { - const inputRef = useRef(null); - const busyRef = useRef(false); - const [dragging, setDragging] = useState(false); - const [uploading, setUploading] = useState(null); - const [error, setError] = useState(null); - const [progress, setProgress] = useState(0); +type QueueState = "queued" | "uploading" | "done" | "failed"; - const register = useCallback( - async (file: File): Promise => { - if (!live || !canUpload || busyRef.current) return; - const draft: UploadDraftInfo = { key: `${file.name}:${file.size}:${file.lastModified}`, name: file.name, detail: formatBytes(file.size) }; - if (!file.type.startsWith("video/")) { - setError(`${file.name}: not a video file (${file.type === "" ? "unknown type" : file.type})`); - return; - } - if (file.size === 0 || file.size > MAX_UPLOAD_BYTES) { - setError("Choose a non-empty video up to 2 GiB."); - return; - } - setError(null); - busyRef.current = true; - setProgress(0); - setUploading(`${file.name} (${formatBytes(file.size)})`); - onStarted(draft); - try { - // Above one chunk the whole-body POST would trip Cloudflare's body cap. - const result = file.size > CHUNK_SIZE_BYTES - ? await uploadVideoChunked(file, setProgress) - : await apiUploadFile("/video/upload", file, setProgress); - onUploaded(result); - if (result.processingJob.state === "failed") setError(result.processingJob.error ?? "Video probing failed."); - } catch (err) { - setError(errorMessage(err)); - } finally { - onSettled(draft.key); - setUploading(null); - busyRef.current = false; +interface QueueItem { + id: string; + name: string; + size: number; + state: QueueState; + progress: number; + error: string | null; + /** Set when the direct path failed and a fallback transport carried the file. */ + warning: string | null; + file: File; +} + +export function UploadDropzone({ live, canUpload, onStarted, onSettled, onUploaded }: Props): React.ReactElement { + const t = useLang(); + const inputRef = useRef(null); + /** Synchronous in-flight guard: a drop can land before `uploading` re-renders. */ + const busyRef = useRef(false); + /** Direct config is cached for the lifetime of the mount: one fetch. */ + const configRef = useRef(undefined); + const [dragging, setDragging] = useState(false); + const [queue, setQueue] = useState([]); + const [warning, setWarning] = useState(null); + + const uploading = queue.some((item) => item.state === "uploading" || item.state === "queued"); + const disabled = !live || !canUpload || uploading || busyRef.current; + + const patch = useCallback((id: string, changes: Partial): void => { + setQueue((current) => current.map((item) => (item.id === id ? { ...item, ...changes } : item))); + }, []); + + /** Direct config, fetched once per mount (undefined = not asked yet). */ + const configFor = useCallback(async (): Promise => { + if (configRef.current === undefined) { + configRef.current = await fetchVideoDirectConfig().catch(() => null); + } + return configRef.current; + }, []); + + /** + * One file through whichever transport applies, first match wins. Returns + * the created video; throws only when every applicable path has failed. + */ + const uploadOne = useCallback( + async (item: QueueItem, onProgress: (percent: number) => void): Promise => { + const config = await configFor(); + if (directEligible(config, item.size) && config !== null && config.publicKeyJwk !== null) { + try { + return await uploadVideoDirect(item.file, config, onProgress); + } catch (directError) { + // Direct is best-effort: fall through to the API paths, but say why. + const reason = errorMessage(directError); + setWarning( + t({ + en: `${item.name}: direct upload failed (${reason}) — retrying through the API.`, + es: `${item.name}: la subida directa falló (${reason}) — reintentando por la API.`, + }), + ); + onProgress(0); + patch(item.id, { + warning: t({ en: `direct upload failed (${reason}) — sent via the API`, es: `la subida directa falló (${reason}) — enviado por la API` }), + }); + } } + // Above one chunk the whole-body POST would trip the proxied-body cap. + return item.size > CHUNK_SIZE_BYTES + ? uploadVideoChunked(item.file, onProgress) + : apiUploadFile("/video/upload", item.file, onProgress); }, - [live, canUpload, onStarted, onSettled, onUploaded], + [configFor, patch, t], ); - // Paste a video file straight into the library. + const runQueue = useCallback( + async (items: QueueItem[]): Promise => { + busyRef.current = true; + for (const item of items) { + // A file can already be terminal (empty/oversize, marked at register time). + if (item.state !== "queued") continue; + const draft: UploadDraftInfo = { key: item.id, name: item.name, detail: formatBytes(item.size) }; + patch(item.id, { state: "uploading", progress: 0 }); + onStarted(draft); + try { + const status = await uploadOne(item, (progress) => patch(item.id, { progress })); + // The bytes landed; a failed probe is a processing result, not an + // upload failure — the library row carries it, so surface it inline. + const processingError = status.processingJob.state === "failed" ? status.processingJob.error ?? null : null; + patch(item.id, { + state: "done", + progress: 100, + warning: processingError === null ? null : t({ en: `probe failed: ${processingError}`, es: `falló el análisis: ${processingError}` }), + }); + onUploaded(status); + } catch (err) { + patch(item.id, { state: "failed", error: errorMessage(err) }); + } finally { + onSettled(draft.key); + } + } + busyRef.current = false; + // Let the results stay visible briefly, then clear finished entries. + window.setTimeout(() => { + setQueue((current) => current.filter((entry) => entry.state === "uploading" || entry.state === "queued")); + }, 8000); + }, + [onStarted, onSettled, onUploaded, patch, t, uploadOne], + ); + + const register = useCallback( + (files: File[]): void => { + if (disabled || files.length === 0) return; + setWarning(null); + const accepted: QueueItem[] = []; + for (const file of files) { + const id = `${file.name}:${file.size}:${file.lastModified}:${Math.random().toString(36).slice(2, 8)}`; + const base = { id, name: file.name, size: file.size, progress: 0, file }; + if (!file.type.startsWith("video/")) { + accepted.push({ + ...base, + state: "failed", + error: t({ + en: `not a video file (${file.type === "" ? "unknown type" : file.type}).`, + es: `no es un archivo de vídeo (${file.type === "" ? "tipo desconocido" : file.type}).`, + }), + warning: null, + }); + continue; + } + if (file.size === 0) { + accepted.push({ + ...base, + state: "failed", + error: t({ en: "the file is empty.", es: "el archivo está vacío." }), + warning: null, + }); + continue; + } + if (file.size > MAX_UPLOAD_BYTES) { + accepted.push({ + ...base, + state: "failed", + error: t({ + en: `is ${formatBytes(file.size)} — the limit is ${MAX_LABEL}.`, + es: `pesa ${formatBytes(file.size)} — el límite es ${MAX_LABEL}.`, + }), + warning: null, + }); + continue; + } + accepted.push({ ...base, state: "queued", progress: 0, error: null, warning: null }); + } + setQueue((current) => [...current, ...accepted]); + if (accepted.some((entry) => entry.state === "queued")) void runQueue(accepted); + }, + [disabled, runQueue, t], + ); + + // Paste a video straight into the library. useEffect(() => { if (!live) return; const onPaste = (e: ClipboardEvent): void => { const file = Array.from(e.clipboardData?.files ?? []).find((f) => f.type.startsWith("video/")); - if (file !== undefined) void register(file); + if (file !== undefined) register([file]); }; window.addEventListener("paste", onPaste); return () => window.removeEventListener("paste", onPaste); }, [live, register]); - const disabled = !live || !canUpload || uploading !== null; const hint = !live - ? "Offline — connect to the API to upload." + ? t({ en: "Offline — connect to the API to upload.", es: "Sin conexión — conéctate a la API para subir." }) : !canUpload - ? "Uploading needs the media.manage permission." - : "Drop a video file, paste one, or browse. Up to 2 GiB — larger videos upload in 10 MiB chunks; stored encrypted and checked before it is ready."; + ? t({ + en: "Uploading needs the media.manage permission.", + es: "Subir requiere el permiso media.manage.", + }) + : t({ + en: `Drop videos, paste one, or browse — several at a time, up to ${MAX_LABEL} each. Large videos upload in ${CHUNK_LABEL} chunks or straight to storage, encrypted end to end.`, + es: `Arrastra vídeos, pega uno o búscalos — varios a la vez, hasta ${MAX_LABEL} cada uno. Los vídeos grandes suben en trozos de ${CHUNK_LABEL} o directo al almacenamiento, cifrados de extremo a extremo.`, + }); return (
@@ -100,11 +273,11 @@ export function UploadDropzone({ live, canUpload, onStarted, onSettled, onUpload role="button" tabIndex={disabled ? -1 : 0} aria-disabled={disabled} - aria-label="Upload a video" + aria-label={t({ en: "Upload videos", es: "Subir vídeos" })} onClick={() => { if (!disabled) inputRef.current?.click(); }} - onKeyDown={(e) => { + onKeyDown={(e: React.KeyboardEvent) => { if (disabled) return; if (e.key === "Enter" || e.key === " ") { e.preventDefault(); @@ -120,8 +293,8 @@ export function UploadDropzone({ live, canUpload, onStarted, onSettled, onUpload e.preventDefault(); setDragging(false); if (disabled) return; - const file = e.dataTransfer.files[0]; - if (file !== undefined) void register(file); + const files = Array.from(e.dataTransfer.files); + if (files.length > 0) register(files); }} className={`flex flex-col items-center justify-center gap-2 rounded-lg border border-dashed px-4 py-6 text-center transition-colors duration-150 ${ disabled @@ -131,17 +304,47 @@ export function UploadDropzone({ live, canUpload, onStarted, onSettled, onUpload : "cursor-pointer border-border hover:border-accent hover:bg-surface-2" }`} > - {uploading !== null ? ( -
- - - {progress < 100 ? `Uploading ${progress}%` : "Upload sent — storing and probing…"} {uploading} - - + {queue.length > 0 ? ( +
+ {queue.map((item) => ( +
+ {item.state === "uploading" ? ( + + ) : item.state === "done" ? ( + ✓ + ) : item.state === "failed" ? ( + ✗ + ) : ( + … + )} + + {item.name} ({formatBytes(item.size)}) + {item.state === "uploading" + ? ` — ${item.progress < 100 ? `${item.progress}%` : t({ en: "probing…", es: "analizando…" })}` + : ""} + {item.state === "failed" && item.error !== null ? ` — ${item.error}` : ""} + {item.warning !== null ? ` — ${item.warning}` : ""} + + {item.state === "uploading" && ( + + )} +
+ ))}
) : ( <> - Drop a video here + {t({ en: "Drop videos here", es: "Suelta vídeos aquí" })} {hint} {!disabled && ( )} @@ -160,16 +363,21 @@ export function UploadDropzone({ live, canUpload, onStarted, onSettled, onUpload { - const file = e.target.files?.[0]; + const files = Array.from(e.target.files ?? []); e.target.value = ""; - if (file !== undefined) void register(file); + if (files.length > 0) register(files); }} />
- {error !== null &&

{error}

} + {warning !== null && ( +

+ {warning} +

+ )}
); } diff --git a/apps/web/src/lib/direct-storage.test.ts b/apps/web/src/lib/direct-storage.test.ts index 88bae82..9a43bbd 100644 --- a/apps/web/src/lib/direct-storage.test.ts +++ b/apps/web/src/lib/direct-storage.test.ts @@ -34,7 +34,8 @@ test("tampered segment ciphertext throws on open", async () => { const plaintext = new Uint8Array(1024); const { frame } = await sealDirect(plaintext, jwk); const copy = Buffer.from(frame); - copy[copy.length - 20] ^= 0xff; + const tamperAt = copy.length - 20; + copy[tamperAt] = (copy[tamperAt] ?? 0) ^ 0xff; assert.throws(() => openDirectFrame(copy, privateKey)); }); diff --git a/apps/web/src/lib/video-direct.ts b/apps/web/src/lib/video-direct.ts new file mode 100644 index 0000000..30f5f39 --- /dev/null +++ b/apps/web/src/lib/video-direct.ts @@ -0,0 +1,96 @@ +/** + * Direct-to-R2 ingest for video — the /video/direct routes. The API mints a + * presigned PUT and records metadata; payload bytes go browser → storage → + * browser sealed client-side (SPENC2 — see ./direct-storage.ts), so the + * server never sees plaintext and no single request is proxied through the + * API at all. + * + * Kept as a pure lib module (no React), same standing as ./chunked-upload.ts + * and ./files-direct.ts, so the transport is testable on its own. + */ +import { api } from "./api-client"; +import { sealDirect } from "./direct-storage"; +import type { VideoDirectConfigDto, VideoDirectInitDto, VideoStatusDto } from "./video-types"; + +/** + * Browser-side ceiling for the direct path: sealing holds the plaintext AND + * the sealed frame in memory at once, so above this the queue falls back to + * the chunked API path (which streams 10 MiB slices instead). + */ +export const DIRECT_MEMORY_CAP_BYTES = 512 * 1024 * 1024; + +export async function fetchVideoDirectConfig(): Promise { + return api.get("/video/direct/config"); +} + +/** True when the direct path applies to this file: enabled, keyed, and fits both ceilings. */ +export function directEligible(config: VideoDirectConfigDto | null, sizeBytes: number): boolean { + return ( + config !== null && + config.enabled && + config.publicKeyJwk !== null && + sizeBytes > 0 && + sizeBytes <= config.maxBytes && + sizeBytes <= DIRECT_MEMORY_CAP_BYTES + ); +} + +/** Empty mimeTypes (drag-drop of unknown extensions) would break metadata downstream. */ +function normalizeMime(file: File): string { + return file.type === "" ? "application/octet-stream" : file.type; +} + +/** + * Seal in the browser, PUT the frame straight to storage, then tell the API + * the object exists. `sizeBytes` attests the SEALED frame (what was actually + * stored); the processing job starts queued and the caller's existing status + * polling takes over from there. + */ +export async function uploadVideoDirect( + file: File, + config: VideoDirectConfigDto, + onProgress: (percent: number) => void, +): Promise { + if (config.publicKeyJwk === null) throw new Error("direct video upload is not enabled"); + const bytes = new Uint8Array(await file.arrayBuffer()); + const { frame } = await sealDirect(bytes, config.publicKeyJwk); + const mimeType = normalizeMime(file); + onProgress(0); + const init = await api.post("/video/direct/init", { + filename: file.name, + mimeType, + sizeBytes: file.size, + }); + await putFrameToStorage(init.putUrl, frame, onProgress); + return api.post("/video/direct/complete", { + uploadId: init.uploadId, + objectKey: init.objectKey, + sizeBytes: frame.byteLength, + mimeType, + filename: file.name, + }); +} + +/** + * Presigned storage PUT. Deliberately NOT apiPutBytes: the signature lives in + * the query string and covers ONLY content-type, so any extra header — the + * bearer token, credentials, withCredentials — breaks it. XHR (not fetch) + * because fetch has no upload progress. + */ +function putFrameToStorage(url: string, frame: Uint8Array, onProgress: (percent: number) => void): Promise { + return new Promise((resolve, reject) => { + const xhr = new XMLHttpRequest(); + xhr.open("PUT", url); + xhr.setRequestHeader("Content-Type", "application/octet-stream"); + xhr.upload.onprogress = (event) => { + if (event.lengthComputable) onProgress(Math.floor((event.loaded / event.total) * 100)); + }; + xhr.onload = () => { + if (xhr.status >= 200 && xhr.status < 300) resolve(); + else reject(new Error(`direct video upload failed (HTTP ${xhr.status})`)); + }; + xhr.onerror = () => reject(new Error("direct video upload failed — network error")); + xhr.ontimeout = () => reject(new Error("direct video upload timed out")); + xhr.send(new Blob([frame.slice()], { type: "application/octet-stream" })); + }); +} diff --git a/apps/web/src/lib/video-types.ts b/apps/web/src/lib/video-types.ts index 4d318fc..53ae62f 100644 --- a/apps/web/src/lib/video-types.ts +++ b/apps/web/src/lib/video-types.ts @@ -3,6 +3,7 @@ * apps/api VideoController/VideoService (same standing as api-types.ts; * declared here because api-client.ts is owned by another worker right now). */ +import type { DirectPublicJwk } from "./direct-storage"; /** video_processing_jobs.state — the processing pipeline a video walks. */ export type VideoProcessingState = "queued" | "probing" | "transcoding" | "complete" | "failed"; @@ -56,6 +57,25 @@ export interface VideoStatusDto { processingJob: VideoProcessingJobDto; } +/** + * Direct-to-R2 ingest for video — same shape as /files/direct/config. When + * `enabled` and a public key is present, the browser seals the file itself + * (SPENC2, ./direct-storage.ts) and PUTs it straight to storage; the API only + * mints URLs and records metadata. + */ +export interface VideoDirectConfigDto { + enabled: boolean; + publicKeyJwk: DirectPublicJwk | null; + maxBytes: number; +} + +/** Response of POST /video/direct/init — a presigned PUT plus the object it targets. */ +export interface VideoDirectInitDto { + uploadId: string; + objectKey: string; + putUrl: string; +} + /** * Producer (POST /video/:id/produce) request/response shapes — same * hand-synced standing as the routes above.