diff --git a/apps/api/migrations/038_media_assets.sql b/apps/api/migrations/038_media_assets.sql new file mode 100644 index 0000000..02b7c3b --- /dev/null +++ b/apps/api/migrations/038_media_assets.sql @@ -0,0 +1,28 @@ +-- 038: media assets for publishing (media attachments backend). +-- +-- Owner need: posts need images/video, not text alone. This is the asset +-- spine the draft's mediaAssetIds point at — one row per uploaded file, the +-- bytes themselves in object storage (envelope-encrypted, content-addressed) +-- keyed by storage_key. Everything is per-workspace; the sha256 is the +-- plaintext content hash the storage service computes, so a re-upload +-- deduplicates the OBJECT but still gets its own row (different alt text, +-- different post usage). created_by is SET NULL on user deletion: the asset +-- outlives the account that uploaded it. + +CREATE TABLE media_assets ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + workspace_id uuid NOT NULL REFERENCES workspaces (id) ON DELETE CASCADE, + storage_key text NOT NULL, + mime text NOT NULL, + size_bytes bigint NOT NULL CHECK (size_bytes > 0), + sha256 text NOT NULL, + alt_text text, + created_by uuid REFERENCES users (id) ON DELETE SET NULL, + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (id, workspace_id) +); + +CREATE POLICY tenant_isolation ON media_assets + USING (workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid); + +CREATE INDEX media_assets_workspace_created ON media_assets (workspace_id, created_at); diff --git a/apps/api/migrations/039_producer.sql b/apps/api/migrations/039_producer.sql new file mode 100644 index 0000000..09ce798 --- /dev/null +++ b/apps/api/migrations/039_producer.sql @@ -0,0 +1,43 @@ +-- 039: producer job intake — a producer can ask for a NEW asset derived from an +-- existing one (burned-in lower thirds/logos/clock, an optional voiceover mix). +-- +-- WHY COLUMNS AND NOT A NEW TABLE: video_processing_jobs (009) already models the +-- per-asset work queue (state machine, progress, output_key, RLS, tenant-immutability +-- triggers, workspace-scoped grants). Intake only needs to distinguish WHAT a queued +-- row asks the media-worker to do and carry the request payload; that is two columns +-- on the existing queue plus a nullable pointer to the row it produces. A second +-- table would duplicate the state machine and the isolation policy for no gain. +-- +-- kind 'probe' — the 009 behaviour: ingest + ffprobe + no-op transcode. +-- 'compose' — burn overlays (and optionally a logo PNG) over the source. +-- 'mix' — compose AND replace/duck the audio bed with a voiceover. +-- Default 'probe' keeps every existing row and INSERT (009 code path) +-- meaning exactly what it meant before this migration. +-- input_json the validated producer request ({...request, requestedBy}), only +-- ever written by the API after strict allowlist parsing; the worker +-- re-validates. NULL for probe rows. +-- produced_asset_id the video_assets row the job writes when it finishes, so a +-- finished job points forward to its output. +-- video_assets.derived_from the SOURCE asset id of a produced row; NULL for ingested +-- uploads. Produced rows are ordinary video_assets rows, so they appear +-- in the library listing and inherit its tenant isolation unchanged. +-- +-- The kind enum is CHECK-constrained so an undecodable row cannot exist. + +BEGIN; + +ALTER TABLE video_processing_jobs + ADD COLUMN kind text NOT NULL DEFAULT 'probe' + CHECK (kind IN ('probe', 'compose', 'mix')), + ADD COLUMN input_json jsonb, + ADD COLUMN produced_asset_id uuid; + +ALTER TABLE video_assets + ADD COLUMN derived_from uuid REFERENCES video_assets(id); + +-- Jobs that produce a derived asset are claimed by the media-worker by kind. +CREATE INDEX video_processing_jobs_kind_queued + ON video_processing_jobs (kind, created_at DESC) + WHERE state = 'queued'; + +COMMIT; diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index fdd953c..ee1dd29 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -42,9 +42,10 @@ import { CollectivesModule } from "./collectives/collectives.module.js"; import { MatrixSsoModule } from "./matrix-sso/matrix-sso.module.js"; import { OidcModule } from "./oidc/oidc.module.js"; import { SubmissionsModule } from "./submissions/submissions.module.js"; +import { MediaModule } from "./media/media.module.js"; @Module({ - imports: [HealthModule, LiveMonitorModule, MediaControlModule, ProductionModule, RealtimeModule, LiveChatModule, LiveModule, VirtualChannelsModule, ChatgovModule, IntakeModule, GroupFilesModule, SettingsModule, CollectivesModule, MatrixSsoModule, OidcModule, SubmissionsModule], + imports: [HealthModule, LiveMonitorModule, MediaControlModule, ProductionModule, RealtimeModule, LiveChatModule, LiveModule, VirtualChannelsModule, ChatgovModule, IntakeModule, GroupFilesModule, SettingsModule, CollectivesModule, MatrixSsoModule, OidcModule, SubmissionsModule, MediaModule], controllers: [ DemoController, PublishingController, diff --git a/apps/api/src/media/media.controller.ts b/apps/api/src/media/media.controller.ts new file mode 100644 index 0000000..072c045 --- /dev/null +++ b/apps/api/src/media/media.controller.ts @@ -0,0 +1,134 @@ +/** + * Media assets (posting attachments): the asset spine that + * `drafts.post.mediaAssetIds` points at. + * + * POST /media draft.create — multipart `file` (optional + * `altText` text field), magic-byte sniffed + * (jpeg/png/webp, mp4/webm), capped 10 MiB image / + * 64 MiB video, stored through the shared + * envelope-encrypted object storage, one + * media_assets row + `media.uploaded` audit + * GET /media/:id draft.view — same-origin proxy of the bytes + * (never a redirect to S3), ETag + immutable cache + * + * The upload reuses the submissions/branding conventions: the declared + * content type is never trusted, only the magic bytes decide; the storage + * service content-addresses and envelope-encrypts; the row keeps the + * plaintext sha256 the storage layer computed. + */ + +import { + BadRequestException, + Body, + Controller, + ForbiddenException, + Get, + HttpCode, + Inject, + Param, + Post, + Req, + Res, + UploadedFile, + UseGuards, + UseInterceptors, +} from "@nestjs/common"; +import { FileInterceptor } from "@nestjs/platform-express"; +import type { Request, Response } from "express"; +import { requireSession } from "../http/validation.js"; +import { RbacGuard, RequirePermission, type AuthenticatedRequest } from "../rbac/rbac.guard.js"; +import { MediaService, MEDIA_MAX_VIDEO_BYTES, mediaAssetPath } from "./media.service.js"; + +interface UploadFile { + buffer: Buffer; + mimetype?: string; +} + +/** Bounds the buffered multipart request; MediaService re-checks after sniffing. */ +const FILE_LIMITS = { + fileSize: MEDIA_MAX_VIDEO_BYTES, + files: 1, + fields: 2, + fieldSize: 2048, + parts: 4, +}; + +export interface MediaUploadResult { + id: string; + url: string; +} + +@Controller("media") +@UseGuards(RbacGuard) +export class MediaController { + constructor(@Inject(MediaService) private readonly media: MediaService) {} + + @Post() + @HttpCode(200) + @RequirePermission("draft.create") + @UseInterceptors(FileInterceptor("file", { limits: FILE_LIMITS })) + async upload( + @Req() request: Request & AuthenticatedRequest, + @UploadedFile() file?: UploadFile, + @Body() body?: Record, + ): Promise { + const session = requireSession(request); + if (file?.buffer === undefined || file.buffer.length === 0) { + throw new BadRequestException("a file is required (field: file)"); + } + const altText = optionalAltText(body?.altText); + const asset = await this.media.upload(session.workspaceId, { + bytes: file.buffer, + declaredMime: typeof file.mimetype === "string" ? file.mimetype : "", + ...(altText === undefined ? {} : { altText }), + actor: { userId: session.userId, role: session.role }, + }); + return { id: asset.id, url: mediaAssetPath(asset.id) }; + } + + /** + * Same-origin asset read (the branding/video-thumbnail pattern): the API + * proxies the decrypted bytes from its own origin — an encrypted S3 object + * is never redirected to — with a bounded immutable cache, nosniff and an + * ETag for conditional revalidation. + */ + @Get(":id") + @RequirePermission("draft.view") + async asset( + @Param("id") id: string, + @Req() request: Request & AuthenticatedRequest, + @Res({ passthrough: true }) response: Response, + ): Promise { + const session = requireSession(request); + if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(id)) { + throw new BadRequestException("id must be a UUID"); + } + const origin = request.headers.origin; + const ownOrigin = process.env.PUBLIC_ORIGIN ?? `${request.protocol}://${request.get("host")}`; + if (origin !== undefined && origin !== ownOrigin) { + throw new ForbiddenException("cross-origin media asset denied"); + } + const asset = await this.media.readAsset(session.workspaceId, id); + const etag = `"${asset.sha256}"`; + response.setHeader("ETag", etag); + response.setHeader("Cache-Control", "private, max-age=86400, immutable"); + response.setHeader("Cross-Origin-Resource-Policy", "same-origin"); + response.setHeader("X-Content-Type-Options", "nosniff"); + const match = request.headers["if-none-match"]; + if (match === etag || match === "*" || (typeof match === "string" && match.split(/,\s*/).includes(etag))) { + response.status(304).end(); + return; + } + response.type(asset.mime).send(asset.bytes); + } +} + +/** Optional free-text alt; empty/whitespace means "no alt text". */ +function optionalAltText(value: unknown): string | undefined { + if (value === undefined || value === null) return undefined; + if (typeof value !== "string") throw new BadRequestException("altText must be a string"); + const trimmed = value.trim(); + if (trimmed === "") return undefined; + if (trimmed.length > 1000) throw new BadRequestException("altText must be at most 1000 characters"); + return trimmed; +} diff --git a/apps/api/src/media/media.module.ts b/apps/api/src/media/media.module.ts new file mode 100644 index 0000000..4cbacc4 --- /dev/null +++ b/apps/api/src/media/media.module.ts @@ -0,0 +1,17 @@ +import { Module } from "@nestjs/common"; +import { MediaController } from "./media.controller.js"; +import { MediaService } from "./media.service.js"; + +/** + * Media attachments for publishing: POST /media (draft.create) uploads a + * sniffed, capped attachment into the shared encrypted object storage and + * records a media_assets row; GET /media/:id (draft.view) proxies the bytes + * same-origin. Everything mutates through MediaService, which keeps the row + * write and the `media.uploaded` tenant audit event together. + */ +@Module({ + controllers: [MediaController], + providers: [MediaService], + exports: [MediaService], +}) +export class MediaModule {} diff --git a/apps/api/src/media/media.service.ts b/apps/api/src/media/media.service.ts new file mode 100644 index 0000000..871d00d --- /dev/null +++ b/apps/api/src/media/media.service.ts @@ -0,0 +1,203 @@ +/** + * Media assets for publishing: upload (magic-byte sniffed, capped), row + * insertion with a tenant audit event, and the auth-scoped read the + * `/media/:id` route proxies. + * + * Deliberately mirrors the submissions/branding seams: + * * the declared content type is never trusted — sniffMedia decides; + * * the size cap is re-checked AFTER sniffing, because the cap depends on + * the sniffed kind (image 10 MiB, video 64 MiB); + * * storage is the shared envelope-encrypted, content-addressed S3 path; + * * the row keeps the plaintext sha256 storage computed, so a later read + * can serve a strong ETag and the draft can cross-check the asset. + */ + +import { BadRequestException, Inject, Injectable, NotFoundException, Optional, ServiceUnavailableException } from "@nestjs/common"; +import { appendTenantAudit } from "../audit/audit.service.js"; +import { withTenant } from "../db/client.js"; +import { S3Storage, s3ConfigFromEnv, sha256, type StorageService } from "../storage/storage.js"; + +/** Images are small; video costs real bytes. Both are hard caps. */ +export const MEDIA_MAX_IMAGE_BYTES = 10 * 1024 * 1024; +export const MEDIA_MAX_VIDEO_BYTES = 64 * 1024 * 1024; + +/** Provider-compatible ceiling: a post carries at most 4 attachments. */ +export const MAX_MEDIA_PER_POST = 4; + +export type MediaKind = "image" | "video"; + +export interface SniffedMedia { + kind: MediaKind; + mime: string; +} + +/** URL path of the auth-scoped read route for a media asset. */ +export function mediaAssetPath(id: string): string { + return `/media/${id}`; +} + +/** + * Magic-byte sniffing for the five posting attachment types. Order matters + * only for readability — every signature is disjoint. + */ +export function sniffMediaAsset(bytes: Buffer): SniffedMedia | null { + if (bytes.length < 16) return null; + // png: 89 50 4E 47 0D 0A 1A 0A + if (bytes.subarray(0, 8).toString("hex") === "89504e470d0a1a0a") return { kind: "image", mime: "image/png" }; + // jpeg: FF D8 FF + if (bytes[0] === 0xff && bytes[1] === 0xd8 && bytes[2] === 0xff) return { kind: "image", mime: "image/jpeg" }; + // webp: "RIFF"...."WEBP" + if (bytes.subarray(0, 4).toString("latin1") === "RIFF" && bytes.subarray(8, 12).toString("latin1") === "WEBP") { + return { kind: "image", mime: "image/webp" }; + } + // mp4: "ftyp" at offset 4 + if (bytes.subarray(4, 8).toString("latin1") === "ftyp") return { kind: "video", mime: "video/mp4" }; + // webm/mkv: EBML magic 1A 45 DF A3 + if (bytes.subarray(0, 4).toString("hex") === "1a45dfa3") return { kind: "video", mime: "video/webm" }; + return null; +} + +export function mediaCapFor(kind: MediaKind): number { + return kind === "video" ? MEDIA_MAX_VIDEO_BYTES : MEDIA_MAX_IMAGE_BYTES; +} + +export interface MediaAssetRecord { + id: string; + workspaceId: string; + storageKey: string; + mime: string; + kind: MediaKind; + sizeBytes: number; + sha256: string; + altText: string | null; + createdAt: Date; +} + +export interface MediaActor { + userId: string; + role: string; +} + +export interface MediaDependencies { + storage: StorageService; +} + +export const MEDIA_DEPENDENCIES = Symbol("MEDIA_DEPENDENCIES"); + +interface MediaRow { + id: string; + workspace_id: string; + storage_key: string; + mime: string; + size_bytes: string | number; + sha256: string; + alt_text: string | null; + created_by: string | null; + created_at: Date; +} + +function kindForMime(mime: string): MediaKind { + return mime.startsWith("video/") ? "video" : "image"; +} + +function fromRow(row: MediaRow): MediaAssetRecord { + return { + id: row.id, + workspaceId: row.workspace_id, + storageKey: row.storage_key, + mime: row.mime, + kind: kindForMime(row.mime), + sizeBytes: Number(row.size_bytes), + sha256: row.sha256, + altText: row.alt_text, + createdAt: new Date(row.created_at), + }; +} + +@Injectable() +export class MediaService { + private readonly storage: StorageService; + + constructor(@Optional() @Inject(MEDIA_DEPENDENCIES) deps?: Partial) { + // Same lazy-construction decision as SubmissionsService.requireStorage: + // tests inject a stub, production constructs from env once. + this.storage = deps?.storage ?? new S3Storage(s3ConfigFromEnv()); + } + + /** Upload one attachment: sniff → cap → store → row + audit. */ + async upload(workspaceId: string, input: { + bytes: Buffer; + declaredMime: string; + altText?: string; + actor: MediaActor; + }): Promise { + const sniffed = sniffMediaAsset(input.bytes); + if (sniffed === null) { + throw new BadRequestException("unrecognised file type — expected jpeg/png/webp image or mp4/webm video"); + } + const max = mediaCapFor(sniffed.kind); + if (input.bytes.length === 0 || input.bytes.length > max) { + throw new BadRequestException(`file must be 1 byte to ${max} bytes for a ${sniffed.kind}`); + } + if (input.declaredMime !== "" && input.declaredMime !== sniffed.mime) { + throw new BadRequestException("file content does not match the declared type"); + } + const stored = await this.storage.putContent(input.bytes, sniffed.mime).catch(() => { + throw new ServiceUnavailableException("media storage unavailable — try again later"); + }); + const digest = sha256(input.bytes); + const record = await withTenant(workspaceId, async (client) => { + const inserted = await client.query( + `INSERT INTO media_assets (workspace_id, storage_key, mime, size_bytes, sha256, alt_text, created_by) + VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING *`, + [workspaceId, stored.key, sniffed.mime, stored.sizeBytes, digest, input.altText ?? null, input.actor.userId], + ); + const row = inserted.rows[0]; + if (row === undefined) throw new Error("media asset insert returned no row"); + return fromRow(row); + }); + await appendTenantAudit(workspaceId, { + actorUserId: input.actor.userId, + actorRole: input.actor.role, + action: "media.uploaded", + targetType: "media_asset", + targetId: record.id, + diff: { kind: sniffed.kind, mime: sniffed.mime, sizeBytes: input.bytes.length, ...(input.altText !== undefined ? { altText: input.altText } : {}) }, + }); + return record; + } + + /** Auth-scoped read: RLS already fences this to the caller's workspace. */ + async readAsset(workspaceId: string, id: string): Promise { + const asset = await withTenant(workspaceId, async (client) => { + const result = await client.query("SELECT * FROM media_assets WHERE id = $1", [id]); + const row = result.rows[0]; + if (row === undefined) throw new NotFoundException(`media asset ${id} not found`); + return fromRow(row); + }); + const bytes = await this.storage.getObject(asset.storageKey).catch(() => { + throw new ServiceUnavailableException("media asset unavailable"); + }); + return { ...asset, bytes }; + } + + /** + * Draft-side existence check: every mediaAssetId the caller sent must be a + * row in THIS workspace. Runs inside the draft's tenant transaction, so the + * RLS policy is the fence — a foreign tenant's id simply counts as absent. + */ + static async assertAssetsExist( + client: { query: (sql: string, values: unknown[]) => Promise<{ rows: Array<{ count: string | number }> }> }, + mediaAssetIds: string[], + ): Promise { + if (mediaAssetIds.length === 0) return; + const result = await client.query( + "SELECT count(*)::integer AS count FROM media_assets WHERE id = ANY($1::uuid[])", + [mediaAssetIds], + ); + const count = Number(result.rows[0]?.count ?? 0); + if (count !== mediaAssetIds.length) { + throw new BadRequestException("one or more mediaAssetIds do not exist in this workspace"); + } + } +} diff --git a/apps/api/src/media/media.test.ts b/apps/api/src/media/media.test.ts new file mode 100644 index 0000000..98226c3 --- /dev/null +++ b/apps/api/src/media/media.test.ts @@ -0,0 +1,312 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { BadRequestException, ServiceUnavailableException } from "@nestjs/common"; +import type { PoolClient } from "pg"; +import { + MEDIA_MAX_IMAGE_BYTES, + MEDIA_MAX_VIDEO_BYTES, + MAX_MEDIA_PER_POST, + MediaService, + mediaAssetPath, + sniffMediaAsset, +} from "./media.service.js"; +import { appendTenantAudit } from "../audit/audit.service.js"; +import { sha256, type StorageService } from "../storage/storage.js"; +import { MediaController } from "./media.controller.js"; +import type { AuthenticatedRequest } from "../rbac/rbac.guard.js"; +import type { Request, Response } from "express"; + +/** + * Unit level: sniffing, per-kind caps, declared-mime cross-check, the + * storage->row->audit contract — with the DB and S3 faked. Live-DB coverage + * (RLS tenant isolation, migration shape) is the integration suite's job. + */ + +const dbMock = vi.hoisted(() => ({ withTenant: vi.fn() })); +vi.mock("../db/client.js", () => ({ withTenant: dbMock.withTenant })); +vi.mock("../audit/audit.service.js", () => ({ appendTenantAudit: vi.fn(async () => "audit-id") })); + +const WS = "0d9a1b40-6f0f-4b1e-8f0e-3f3c9e6b4a10"; +const USER = "8f7f6e50-7b0f-4a2e-9d3e-2f2d1a0c5b21"; +const ACTOR = { userId: USER, role: "Owner" }; + +const PNG = Buffer.concat([ + Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]), + Buffer.alloc(24, 0x11), +]); +const JPEG = Buffer.concat([Buffer.from([0xff, 0xd8, 0xff]), Buffer.alloc(24, 0x22)]); +const WEBP = Buffer.concat([ + Buffer.from("RIFF", "latin1"), + Buffer.from([0x24, 0x00, 0x00, 0x00]), + Buffer.from("WEBP", "latin1"), + Buffer.alloc(16, 0x33), +]); +const MP4 = Buffer.concat([ + Buffer.alloc(4, 0x00), + Buffer.from("ftyp", "latin1"), + Buffer.from("isom", "latin1"), + Buffer.alloc(24, 0x44), +]); +const WEBM = Buffer.concat([Buffer.from([0x1a, 0x45, 0xdf, 0xa3]), Buffer.alloc(24, 0x55)]); + +function fakeStorage(): StorageService & { + putContent: ReturnType; + getObject: ReturnType; +} { + return { + putContent: vi.fn(async (bytes: Buffer, contentType: string) => { + const hash = sha256(bytes); + return { + key: `sha256/${hash}.${contentType === "image/jpeg" ? "jpg" : contentType.split("/")[1]}`, + hash, + sizeBytes: bytes.length, + contentType, + deduplicated: false, + }; + }), + getObject: vi.fn(async () => { + throw new Error("missing"); + }), + hasContent: vi.fn(async () => false), + presign: vi.fn(async () => ""), + }; +} + +/** withTenant fake: records the insert and hands back the row it would store. */ +function tenantInsertReturningRow() { + const inserts: unknown[][] = []; + dbMock.withTenant.mockImplementation(async (_ws: string, fn: (client: PoolClient) => Promise) => + fn({ + query: async (sql: string, values?: unknown[]) => { + if (sql.startsWith("INSERT INTO media_assets")) { + inserts.push(values ?? []); + const [workspaceId, storageKey, mime, sizeBytes, digest, altText, createdBy] = values as unknown[]; + return { + rows: [{ + id: "11111111-2222-4333-8444-555555555555", + workspace_id: workspaceId, + storage_key: storageKey, + mime, + size_bytes: sizeBytes, + sha256: digest, + alt_text: altText, + created_by: createdBy, + created_at: new Date("2026-01-02T03:04:05Z"), + }], + }; + } + throw new Error(`unexpected query: ${sql}`); + }, + } as unknown as PoolClient), + ); + return inserts; +} + +beforeEach(() => { + dbMock.withTenant.mockReset(); + vi.mocked(appendTenantAudit).mockClear(); +}); + +describe("sniffMediaAsset", () => { + it("recognises the two video and three image types by magic bytes", () => { + expect(sniffMediaAsset(PNG)).toEqual({ kind: "image", mime: "image/png" }); + expect(sniffMediaAsset(JPEG)).toEqual({ kind: "image", mime: "image/jpeg" }); + expect(sniffMediaAsset(WEBP)).toEqual({ kind: "image", mime: "image/webp" }); + expect(sniffMediaAsset(MP4)).toEqual({ kind: "video", mime: "video/mp4" }); + expect(sniffMediaAsset(WEBM)).toEqual({ kind: "video", mime: "video/webm" }); + }); + + it("rejects anything else — including a lying content type", () => { + expect(sniffMediaAsset(Buffer.from("definitely not media at all, sorry", "utf8"))).toBeNull(); + expect(sniffMediaAsset(Buffer.alloc(8, 0x00))).toBeNull(); + // too short to sniff + expect(sniffMediaAsset(Buffer.from([0x89, 0x50]))).toBeNull(); + }); +}); + +describe("caps", () => { + it("image cap is 10 MiB and video cap is 64 MiB", () => { + expect(MEDIA_MAX_IMAGE_BYTES).toBe(10 * 1024 * 1024); + expect(MEDIA_MAX_VIDEO_BYTES).toBe(64 * 1024 * 1024); + expect(MAX_MEDIA_PER_POST).toBe(4); + }); + + it("refuses an image over the image cap even though it fits the video cap", async () => { + const storage = fakeStorage(); + const media = new MediaService({ storage }); + const oversized = Buffer.concat([PNG.subarray(0, 8), Buffer.alloc(MEDIA_MAX_IMAGE_BYTES + 1, 0x00)]); + await expect(media.upload(WS, { bytes: oversized, declaredMime: "", actor: ACTOR })) + .rejects.toThrow(BadRequestException); + expect(storage.putContent).not.toHaveBeenCalled(); + }); + + it("refuses a video over the video cap", async () => { + const storage = fakeStorage(); + const media = new MediaService({ storage }); + const oversized = Buffer.concat([MP4, Buffer.alloc(MEDIA_MAX_VIDEO_BYTES + 1, 0x00)]); + await expect(media.upload(WS, { bytes: oversized, declaredMime: "", actor: ACTOR })) + .rejects.toThrow(BadRequestException); + expect(storage.putContent).not.toHaveBeenCalled(); + }); + + it("accepts a video larger than the image cap (kind-dependent limit)", async () => { + const storage = fakeStorage(); + const media = new MediaService({ storage }); + tenantInsertReturningRow(); + const bytes = Buffer.concat([MP4, Buffer.alloc(MEDIA_MAX_IMAGE_BYTES + 1024, 0x00)]); + const asset = await media.upload(WS, { bytes, declaredMime: "", actor: ACTOR }); + expect(asset.kind).toBe("video"); + expect(asset.sizeBytes).toBe(bytes.length); + expect(storage.putContent).toHaveBeenCalledWith(bytes, "video/mp4"); + }); +}); + +describe("MediaService.upload", () => { + it("sniffs, stores through putContent, inserts the row and audits media.uploaded", async () => { + const storage = fakeStorage(); + const media = new MediaService({ storage }); + const inserts = tenantInsertReturningRow(); + const asset = await media.upload(WS, { bytes: PNG, declaredMime: "image/png", altText: "a crowd", actor: ACTOR }); + + expect(asset.mime).toBe("image/png"); + expect(asset.kind).toBe("image"); + expect(asset.sha256).toBe(sha256(PNG)); + expect(asset.altText).toBe("a crowd"); + expect(mediaAssetPath(asset.id)).toBe(`/media/${asset.id}`); + expect(storage.putContent).toHaveBeenCalledTimes(1); + expect(storage.putContent).toHaveBeenCalledWith(PNG, "image/png"); + expect(inserts).toHaveLength(1); + expect(inserts[0]).toEqual([WS, `sha256/${sha256(PNG)}.png`, "image/png", PNG.length, sha256(PNG), "a crowd", USER]); + expect(appendTenantAudit).toHaveBeenCalledWith(WS, expect.objectContaining({ + actorUserId: USER, + action: "media.uploaded", + targetType: "media_asset", + targetId: asset.id, + })); + }); + + it("rejects a declared type that contradicts the sniffed bytes", async () => { + const storage = fakeStorage(); + const media = new MediaService({ storage }); + await expect(media.upload(WS, { bytes: PNG, declaredMime: "video/mp4", actor: ACTOR })) + .rejects.toThrow("file content does not match the declared type"); + expect(storage.putContent).not.toHaveBeenCalled(); + expect(appendTenantAudit).not.toHaveBeenCalled(); + }); + + it("rejects an unrecognised file before touching storage", async () => { + const storage = fakeStorage(); + const media = new MediaService({ storage }); + await expect(media.upload(WS, { bytes: Buffer.alloc(64, 0x41), declaredMime: "image/png", actor: ACTOR })) + .rejects.toThrow(BadRequestException); + expect(storage.putContent).not.toHaveBeenCalled(); + }); + + it("surfaces a storage failure as 503 without leaving a row", async () => { + const storage = fakeStorage(); + storage.putContent.mockRejectedValueOnce(new Error("ECONNREFUSED")); + const media = new MediaService({ storage }); + await expect(media.upload(WS, { bytes: JPEG, declaredMime: "", actor: ACTOR })) + .rejects.toThrow(ServiceUnavailableException); + expect(dbMock.withTenant).not.toHaveBeenCalled(); + }); +}); + +describe("MediaController", () => { + const session = { workspaceId: WS, userId: USER, role: "Owner" }; + + function requestWith(overrides: Partial = {}): Request & AuthenticatedRequest { + return { + headers: {} as Record, + protocol: "http", + get: (name: string) => (name === "host" ? "api.test" : undefined), + session, + ...overrides, + } as unknown as Request & AuthenticatedRequest; + } + + function responseRecording(): { response: Response; headers: Record; body: unknown[]; status: number[] } { + const headers: Record = {}; + const body: unknown[] = []; + const status: number[] = []; + const response = { + setHeader: (k: string, v: unknown) => { headers[k.toLowerCase()] = v; }, + status: (code: number) => { status.push(code); return response; }, + type: () => response, + send: (payload: unknown) => { body.push(payload); return response; }, + end: () => { body.push(undefined); return response; }, + } as unknown as Response; + return { response, headers, body, status }; + } + + it("rejects a missing upload file", async () => { + const controller = new MediaController(new MediaService({ storage: fakeStorage() })); + await expect(controller.upload(requestWith(), undefined, {})).rejects.toThrow("a file is required"); + }); + + it("returns the asset id and its /media/ url", async () => { + const storage = fakeStorage(); + tenantInsertReturningRow(); + const controller = new MediaController(new MediaService({ storage })); + const result = await controller.upload(requestWith(), { buffer: WEBP, mimetype: "image/webp" }, {}); + expect(result.url).toBe(`/media/${result.id}`); + expect(result.id).toBe("11111111-2222-4333-8444-555555555555"); + }); + + it("serves bytes with an ETag, cache and nosniff headers", async () => { + const storage = fakeStorage(); + storage.getObject.mockResolvedValue(PNG); + dbMock.withTenant.mockImplementation(async (_ws: string, fn: (client: PoolClient) => Promise) => + fn({ + query: async () => ({ + rows: [{ + id: "11111111-2222-4333-8444-555555555555", workspace_id: WS, + storage_key: `sha256/${sha256(PNG)}.png`, mime: "image/png", size_bytes: PNG.length, + sha256: sha256(PNG), alt_text: null, created_by: USER, created_at: new Date(), + }], + }), + } as unknown as PoolClient)); + const controller = new MediaController(new MediaService({ storage })); + const { response, headers, body } = responseRecording(); + await controller.asset("11111111-2222-4333-8444-555555555555", requestWith(), response); + expect(headers.etag).toBe(`"${sha256(PNG)}"`); + expect(String(headers["cache-control"])).toContain("immutable"); + expect(headers["x-content-type-options"]).toBe("nosniff"); + expect(body[0]).toBe(PNG); + }); + + it("answers a matching If-None-Match with 304 and no body", async () => { + const storage = fakeStorage(); + storage.getObject.mockResolvedValue(JPEG); + dbMock.withTenant.mockImplementation(async (_ws: string, fn: (client: PoolClient) => Promise) => + fn({ query: async () => ({ rows: [{ id: "x", workspace_id: WS, storage_key: "k", mime: "image/jpeg", size_bytes: 1, sha256: sha256(JPEG), alt_text: null, created_by: null, created_at: new Date() }] }) } as unknown as PoolClient)); + const controller = new MediaController(new MediaService({ storage })); + const { response, status, body } = responseRecording(); + await controller.asset("11111111-2222-4333-8444-555555555555", + requestWith({ headers: { "if-none-match": `"${sha256(JPEG)}"` } }), response); + expect(status).toContain(304); + // 304 carries no body (end() pushes a single undefined entry). + expect(body).toEqual([undefined]); + // A mismatching ETag still sends the bytes. + const second = responseRecording(); + await controller.asset("11111111-2222-4333-8444-555555555555", + requestWith({ headers: { "if-none-match": '"stale"' } }), second.response); + expect(second.status).not.toContain(304); + expect(second.body[0]).toBe(JPEG); + }); + + it("refuses a cross-origin read", async () => { + const storage = fakeStorage(); + const controller = new MediaController(new MediaService({ storage })); + const { response } = responseRecording(); + await expect(controller.asset("11111111-2222-4333-8444-555555555555", + requestWith({ headers: { origin: "https://elsewhere.example" } }), response)) + .rejects.toThrow("cross-origin media asset denied"); + expect(storage.getObject).not.toHaveBeenCalled(); + }); + + it("refuses a non-UUID asset id", async () => { + const controller = new MediaController(new MediaService({ storage: fakeStorage() })); + const { response } = responseRecording(); + await expect(controller.asset("not-a-uuid", requestWith(), response)).rejects.toThrow("id must be a UUID"); + }); +}); diff --git a/apps/api/src/publishing/publishing.controller.ts b/apps/api/src/publishing/publishing.controller.ts index ca66e14..2d99603 100644 --- a/apps/api/src/publishing/publishing.controller.ts +++ b/apps/api/src/publishing/publishing.controller.ts @@ -17,6 +17,7 @@ import { import type { Request, Response } from "express"; import { RbacGuard, RequireGroupAccess, RequirePermission, type AuthenticatedRequest } from "../rbac/rbac.guard.js"; import { ApprovalGateService } from "./approval-gate.service.js"; +import { MAX_MEDIA_PER_POST } from "../media/media.service.js"; import type { GateDecisionRecord } from "../chatgov/gate.service.js"; import { PublishingService, @@ -238,20 +239,19 @@ function parseCreateDraft(value: unknown): CreateDraftDto { const post = requireObject(body.post, "post"); const text = requireString(post.text, "post.text"); const visibility = requireVisibility(post.visibility); - if (!Array.isArray(post.mediaAssetIds) || !post.mediaAssetIds.every((id) => typeof id === "string")) { - throw new BadRequestException("post.mediaAssetIds must be an array of strings"); - } + const mediaAssetIds = parseMediaAssetIds(post.mediaAssetIds); + const altTexts = parseAltTexts(post.altTexts, mediaAssetIds.length); const providerMetadata = requireObject(post.providerMetadata, "post.providerMetadata"); - if (post.mediaAssetIds.length > 0) throw new BadRequestException("publishing currently supports text only"); if (["idempotencyKey", "draftId", "mediaIds"].some((key) => key in providerMetadata)) throw new BadRequestException("post metadata contains reserved execution keys"); if (Buffer.byteLength(JSON.stringify(providerMetadata)) > 16_384) throw new BadRequestException("post metadata is too large"); if (post.scheduledAt !== undefined) requireTimestamp(post.scheduledAt, "post.scheduledAt"); return { post: { text, - mediaAssetIds: post.mediaAssetIds, + mediaAssetIds, visibility, providerMetadata, + ...(altTexts === undefined ? {} : { altTexts }), ...(typeof post.scheduledAt === "string" ? { scheduledAt: post.scheduledAt } : {}), }, }; @@ -322,6 +322,40 @@ function parseListLimit(value: string | undefined): number { return Math.min(n, 200); } +/** + * 0..4 attachments, each a media_assets id. Existence in the caller's + * workspace is checked service-side (RLS-scoped) — here we only bound the + * shape, the count and the id syntax. + */ +function parseMediaAssetIds(value: unknown): string[] { + if (value === undefined) return []; + if (!Array.isArray(value) || !value.every((id) => typeof id === "string")) { + throw new BadRequestException("post.mediaAssetIds must be an array of strings"); + } + if (value.length > MAX_MEDIA_PER_POST) { + throw new BadRequestException(`post.mediaAssetIds supports at most ${MAX_MEDIA_PER_POST} attachments`); + } + return value.map((id) => parseId(id)); +} + +/** Optional alt texts, positionally aligned with mediaAssetIds (null = none). */ +function parseAltTexts(value: unknown, mediaCount: number): (string | null)[] | undefined { + if (value === undefined) return undefined; + if (!Array.isArray(value) || !value.every((entry) => entry === null || typeof entry === "string")) { + throw new BadRequestException("post.altTexts must be an array of strings or nulls"); + } + if (value.length !== mediaCount) { + throw new BadRequestException("post.altTexts must align with post.mediaAssetIds (same length)"); + } + return value.map((entry) => { + if (entry === null || entry === undefined) return null; + const trimmed = entry.trim(); + if (trimmed === "") return null; + if (trimmed.length > 1000) throw new BadRequestException("post.altTexts entries must be at most 1000 characters"); + return trimmed; + }); +} + function requireVisibility(value: unknown): DraftPost["visibility"] { if ( value !== "public" && diff --git a/apps/api/src/publishing/publishing.service.ts b/apps/api/src/publishing/publishing.service.ts index 1a2af6f..f32d8c1 100644 --- a/apps/api/src/publishing/publishing.service.ts +++ b/apps/api/src/publishing/publishing.service.ts @@ -22,7 +22,10 @@ export type DraftState = "editing" | "in_review" | "approved" | "rejected" | "sc export interface DraftPost { text: string; + /** media_assets ids (0..4); existence is checked against this workspace on write. */ mediaAssetIds: string[]; + /** Optional alt texts, positionally aligned with mediaAssetIds. */ + altTexts?: (string | null)[]; visibility: "public" | "unlisted" | "private" | "followers" | "provider_default"; scheduledAt?: string; providerMetadata: Record; @@ -180,6 +183,9 @@ export class PublishingService { await this.dependencies.assertReady(); const draft = await this.dependencies.tenantTransaction(workspaceId, async (client) => { await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]); + // FK checks bypass RLS, so a foreign tenant's asset id would satisfy the + // constraint; this RLS-scoped count is the actual tenancy check. + await assertMediaAssetsExist(client, post.mediaAssetIds); const result = await client.query( `INSERT INTO drafts (workspace_id, author_user_id, post) VALUES ($1, $2, $3) RETURNING *`, @@ -365,6 +371,7 @@ export class PublishingService { await this.dependencies.assertReady(); const draft = await this.dependencies.tenantTransaction(workspaceId, async (client) => { await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]); + await assertMediaAssetsExist(client, post.mediaAssetIds); const result = await client.query( `UPDATE drafts SET post = $2, updated_at = now() WHERE id = $1 AND (state = 'editing' OR (state = 'in_review' AND EXISTS (SELECT 1 FROM polls WHERE draft_id = $1 AND state = 'open'))) RETURNING *`, @@ -639,6 +646,22 @@ export class PublishingService { } } +/** + * Every mediaAssetId must be a media_assets row in THIS workspace. Runs inside + * the caller's tenant transaction so the RLS tenant_isolation policy is the + * fence -- a foreign tenant's id simply counts as absent. + */ +async function assertMediaAssetsExist(client: PoolClient, mediaAssetIds: string[]): Promise { + if (mediaAssetIds.length === 0) return; + const result = await client.query<{ count: string | number }>( + "SELECT count(*)::integer AS count FROM media_assets WHERE id = ANY($1::uuid[])", + [mediaAssetIds], + ); + if (Number(result.rows[0]?.count ?? 0) !== mediaAssetIds.length) { + throw new BadRequestException("one or more post.mediaAssetIds do not exist in this workspace"); + } +} + function requiredRow(row: unknown, description: string): Record { if (typeof row !== "object" || row === null) throw new Error(`database did not return ${description}`); return row as Record; diff --git a/apps/api/src/video/produce-request.test.ts b/apps/api/src/video/produce-request.test.ts new file mode 100644 index 0000000..cea4ec1 --- /dev/null +++ b/apps/api/src/video/produce-request.test.ts @@ -0,0 +1,143 @@ +import { describe, expect, it } from "vitest"; +import { + MAX_PRODUCE_OVERLAY_TEXT, + MAX_PRODUCE_POSITION, + parseProduceRequest, +} from "./produce-request.js"; + +const voiceoverAssetId = "20000000-0000-4000-8000-000000000002"; + +describe("produce request validation", () => { + it("accepts a full request and canonicalizes it (fixed key order, defaults dropped)", () => { + const parsed = parseProduceRequest({ + overlays: [ + { kind: "lower_third", text: "Live from the occupation", position: { x: 0, y: 100 } }, + { kind: "logo", logoStorageKey: "sha256/9f2b1c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d.png", position: { x: 100, y: 0 } }, + { kind: "clock" }, + ], + voiceover: { assetId: voiceoverAssetId, gainDb: -6, duckOriginal: true }, + outputLabel: "Riot recap v2", + lane: "remote", + }); + expect(parsed).toEqual({ + overlays: [ + { kind: "lower_third", text: "Live from the occupation", position: { x: 0, y: 100 } }, + { kind: "logo", logoStorageKey: "sha256/9f2b1c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d.png", position: { x: 100, y: 0 } }, + { kind: "clock", position: undefined }, + ], + voiceover: { assetId: voiceoverAssetId, gainDb: -6, duckOriginal: true }, + outputLabel: "Riot recap v2", + lane: "remote", + }); + // Same input always serializes to the same input_json bytes. + expect(JSON.stringify(parsed)).toBe(JSON.stringify(parseProduceRequest(JSON.parse(JSON.stringify({ + lane: "remote", + outputLabel: "Riot recap v2", + voiceover: { duckOriginal: true, gainDb: -6, assetId: voiceoverAssetId }, + overlays: [ + { position: { y: 100, x: 0 }, text: "Live from the occupation", kind: "lower_third" }, + { position: { y: 0, x: 100 }, logoStorageKey: "sha256/9f2b1c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d.png", kind: "logo" }, + { kind: "clock" }, + ], + }))))); + }); + + it("accepts an empty body and treats every field as optional", () => { + expect(parseProduceRequest({})).toEqual({}); + expect(parseProduceRequest({ overlays: null, voiceover: null, outputLabel: null, lane: null })).toEqual({}); + }); + + it("rejects an unknown key on the body, an overlay and a voiceover (never ignores one)", () => { + expect(() => parseProduceRequest({ outputLabel: "x", filter: "drawtext" })).toThrow(/body\.filter is not allowed/); + expect(() => parseProduceRequest({ apiKey: "x" })).toThrow(/secrets are never accepted/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", layer: 0 }] })).toThrow(/overlays\[0\]\.layer is not allowed/); + expect(() => parseProduceRequest({ voiceover: { assetId: voiceoverAssetId, streamKey: "x" } })).toThrow(/secrets are never accepted/); + }); + + it("rejects filtergraph and shell metacharacters in overlay text", () => { + for (const hostile of [ + "text';drawtext=1", + "text[0]", + "text,volume=9", + "text'quote", + 'text"quote', + "text`cmd`", + "text$HOME", + "text\\n", + "text{a}", + "text=x", + ]) { + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: hostile }] }), hostile).toThrow(/metacharacter|control characters|must be a string/); + } + // Newlines are control characters too. + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: "a\nb" }] })).toThrow(/control characters/); + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: "\u0000" }] })).toThrow(/control characters/); + }); + + it("bounds overlay text length and rejects empty text", () => { + expect(parseProduceRequest({ overlays: [{ kind: "lower_third", text: "x".repeat(MAX_PRODUCE_OVERLAY_TEXT) }] }).overlays?.[0]?.text).toHaveLength(200); + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: "x".repeat(201) }] })).toThrow(/at most 200 characters/); + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: "" }] })).toThrow(/must not be empty/); + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: " " }] })).toThrow(/must not be empty/); + }); + + it("restricts position to finite numbers 0-100 on both axes", () => { + expect(parseProduceRequest({ overlays: [{ kind: "clock", position: { x: 0, y: MAX_PRODUCE_POSITION } }] }).overlays?.[0]?.position).toEqual({ x: 0, y: 100 }); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", position: { x: -1, y: 0 } }] })).toThrow(/position\.x must be between 0 and 100/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", position: { x: 101, y: 0 } }] })).toThrow(/position\.y|position\.x must be between/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", position: { x: Number.NaN, y: 0 } }] })).toThrow(/position\.x must be a number/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", position: { x: "50", y: 0 } }] })).toThrow(/position\.x must be a number/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", position: { x: 50 } }] })).toThrow(/position\.y/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", position: { x: 50, y: 0, w: 10 } }] })).toThrow(/position\.w is not allowed/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", position: { x: Number.POSITIVE_INFINITY, y: 0 } }] })).toThrow(/position\.x must be a number|between 0 and 100/); + }); + + it("keeps overlay kinds closed and per-kind fields exclusive", () => { + expect(() => parseProduceRequest({ overlays: [{ kind: "chyron" }] })).toThrow(/overlays\[0\]\.kind must be one of lower_third, logo, clock/); + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third" }] })).toThrow(/overlays\[0\]\.text must be a string/); + expect(() => parseProduceRequest({ overlays: [{ kind: "logo" }] })).toThrow(/overlays\[0\]\.logoStorageKey must be a content-addressed storage key/); + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", text: "nope" }] })).toThrow(/overlays\[0\]\.text is not allowed/); + expect(() => parseProduceRequest({ overlays: [{ kind: "logo", text: "nope" }] })).toThrow(/overlays\[0\]\.text is not allowed/); + // A storage key riding on a clock is rejected; the "key"-shaped name trips the + // secrets branch first (same precedence as production.ts) — either way, refused. + expect(() => parseProduceRequest({ overlays: [{ kind: "clock", logoStorageKey: "sha256/9f2b1c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d.png" }] })).toThrow(/logoStorageKey/); + expect(() => parseProduceRequest({ overlays: [{ kind: "lower_third", text: "ok", logoStorageKey: "sha256/9f2b1c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d0e0f0a0b0c0d.png" }] })).toThrow(/logoStorageKey/); + expect(() => parseProduceRequest({ overlays: [{}] })).toThrow(/overlays\[0\]\.kind must be one of/); + }); + + it("accepts only content-addressed storage keys for logos (no traversal, no separators)", () => { + expect(() => parseProduceRequest({ overlays: [{ kind: "logo", logoStorageKey: "../../etc/passwd" }] })).toThrow(/content-addressed storage key/); + expect(() => parseProduceRequest({ overlays: [{ kind: "logo", logoStorageKey: "sha256/../x.png" }] })).toThrow(/content-addressed storage key/); + expect(() => parseProduceRequest({ overlays: [{ kind: "logo", logoStorageKey: "C:\\evil.png" }] })).toThrow(/content-addressed storage key/); + expect(() => parseProduceRequest({ overlays: [{ kind: "logo", logoStorageKey: "file:///tmp/x.png" }] })).toThrow(/content-addressed storage key/); + expect(() => parseProduceRequest({ overlays: [{ kind: "logo", logoStorageKey: "png" }] })).toThrow(/content-addressed storage key/); + }); + + it("validates the voiceover: UUID asset, bounded gain, boolean duck", () => { + expect(() => parseProduceRequest({ voiceover: { assetId: "not-a-uuid" } })).toThrow(/voiceover\.assetId must be a UUID/); + expect(() => parseProduceRequest({ voiceover: { assetId: voiceoverAssetId, gainDb: -49 } })).toThrow(/gainDb must be a number between -48 and 12 dB/); + expect(() => parseProduceRequest({ voiceover: { assetId: voiceoverAssetId, gainDb: 13 } })).toThrow(/gainDb must be a number between/); + expect(() => parseProduceRequest({ voiceover: { assetId: voiceoverAssetId, gainDb: "loud" } })).toThrow(/gainDb/); + expect(() => parseProduceRequest({ voiceover: { assetId: voiceoverAssetId, duckOriginal: "yes" } })).toThrow(/duckOriginal must be a boolean/); + expect(parseProduceRequest({ voiceover: { assetId: voiceoverAssetId } }).voiceover).toEqual({ assetId: voiceoverAssetId }); + }); + + it("bounds outputLabel and the overlay list, and closes the lane enum", () => { + expect(() => parseProduceRequest({ outputLabel: "x".repeat(65) })).toThrow(/outputLabel must be 1-64/); + expect(() => parseProduceRequest({ outputLabel: "has;semicolon" })).toThrow(/outputLabel must be 1-64/); + expect(() => parseProduceRequest({ outputLabel: "=cmd" })).toThrow(/outputLabel must be 1-64/); + expect(() => parseProduceRequest({ lane: "cloud" })).toThrow(/lane must be one of local, remote/); + expect(() => parseProduceRequest({ overlays: new Array(65).fill({ kind: "clock" }) })).toThrow(/at most 64 entries/); + expect(() => parseProduceRequest({ overlays: "clock" })).toThrow(/overlays must be an array/); + expect(() => parseProduceRequest("nope")).toThrow(/body must be an object/); + expect(() => parseProduceRequest([{ kind: "clock" }])).toThrow(/body must be an object/); + }); + + it("returns a serializable, JSON-round-trippable request for input_json", () => { + const parsed = parseProduceRequest({ overlays: [{ kind: "lower_third", text: "ok", position: { x: 1, y: 2 } }], voiceover: { assetId: voiceoverAssetId, gainDb: 0 } }); + expect(JSON.parse(JSON.stringify(parsed))).toEqual({ + overlays: [{ kind: "lower_third", text: "ok", position: { x: 1, y: 2 } }], + voiceover: { assetId: voiceoverAssetId, gainDb: 0 }, + }); + }); +}); diff --git a/apps/api/src/video/produce-request.ts b/apps/api/src/video/produce-request.ts new file mode 100644 index 0000000..9019065 --- /dev/null +++ b/apps/api/src/video/produce-request.ts @@ -0,0 +1,178 @@ +import { BadRequestException } from "@nestjs/common"; +import { MAX_OVERLAYS } from "../media-control/production.js"; +import { isUuid } from "../http/validation.js"; + +/** + * Producer job intake: the request shape for POST /video/:id/produce. + * + * Everything accepted here is persisted verbatim into + * video_processing_jobs.input_json and later reaches an ffmpeg filtergraph in + * the media-worker compositor, so the rules follow + * media-control/production.ts: closed enums, exhaustive key allowlists (an + * unknown key is rejected, never ignored — that is what keeps a stray + * `movie=`/`enable=` field from riding into a filtergraph), bounded geometry, + * bounded text, and no filtergraph or shell metacharacter inside operator text. + * + * The parse is total: it either returns a canonicalized request (fixed key + * order, so the stored input_json bytes are deterministic) or throws + * BadRequestException with a field path. + */ + +export type ProduceOverlayKind = "lower_third" | "logo" | "clock"; +export type ProduceLane = "local" | "remote"; + +export const PRODUCE_OVERLAY_KINDS: readonly ProduceOverlayKind[] = ["lower_third", "logo", "clock"]; +export const PRODUCE_LANES: readonly ProduceLane[] = ["local", "remote"]; + +export const MAX_PRODUCE_OVERLAY_TEXT = 200; +export const MAX_PRODUCE_LABEL = 64; +export const PRODUCE_POSITION_MIN = 0; +export const MAX_PRODUCE_POSITION = 100; +export const PRODUCE_GAIN_DB_MIN = -48; +export const PRODUCE_GAIN_DB_MAX = 12; + +export interface ProduceOverlay { + kind: ProduceOverlayKind; + /** lower_third only: the burned-in caption. */ + text?: string; + /** Percent of the output frame, 0-100 on each axis. */ + position?: { x: number; y: number }; + /** logo only: storage key of the rendered PNG. */ + logoStorageKey?: string; +} + +export interface ProduceVoiceover { + assetId: string; + gainDb?: number; + duckOriginal?: boolean; +} + +export interface ProduceRequest { + overlays?: ProduceOverlay[]; + voiceover?: ProduceVoiceover; + outputLabel?: string; + lane?: ProduceLane; +} + +/** Keys nobody may post, mirroring production.ts: secrets are never accepted. */ +const SECRETISH = /key|secret|token|password|passwd|credential(?!HandleId)|auth/i; +/** + * Characters that change an ffmpeg filtergraph (`;` splits, `[`/`]` label pads, + * `,` separates args, `'` quotes, `=` sets an option) or a shell string. A lower + * third is operator prose, so the check is a blocklist of these rather than + * production.ts's tight label allowlist — prose keeps its punctuation and no + * metacharacter survives. + */ +const TEXT_FORBIDDEN = new Set([';', "[", "]", "'", '"', "`", "$", "\\", "{", "}", "=", ","]); +/** Storage keys are content-addressed, flat and extension-suffixed; no traversal, no separators. */ +const STORAGE_KEY = /^[a-z0-9][a-z0-9/._-]{7,255}$/; +const LABEL = /^[\p{L}\p{N}][\p{L}\p{N} _.-]{0,63}$/u; + +function fail(message: string): never { + throw new BadRequestException(message); +} + +function record(value: unknown, field: string, allowed: readonly string[]): Record { + if (typeof value !== "object" || value === null || Array.isArray(value)) fail(`${field} must be an object`); + for (const key of Object.keys(value)) { + if (allowed.includes(key)) continue; + if (SECRETISH.test(key)) fail(`${field}.${key}: secrets are never accepted; reference a vault credentialHandleId`); + fail(`${field}.${key} is not allowed`); + } + return value as Record; +} + +function optional(value: unknown, field: string, parse: (v: unknown, f: string) => T): T | undefined { + return value === undefined || value === null ? undefined : parse(value, field); +} + +function text(value: unknown, field: string): string { + if (typeof value !== "string") fail(`${field} must be a string`); + if (value.trim() === "") fail(`${field} must not be empty`); + if (value.length > MAX_PRODUCE_OVERLAY_TEXT) fail(`${field} must be at most ${MAX_PRODUCE_OVERLAY_TEXT} characters`); + for (const character of value) { + const code = character.codePointAt(0) ?? 0; + if (code < 0x20 || code === 0x7f) fail(`${field} must not contain control characters`); + if (TEXT_FORBIDDEN.has(character)) fail(`${field} must not contain the filtergraph metacharacter ${JSON.stringify(character)}`); + } + return value; +} + +function position(value: unknown, field: string): { x: number; y: number } { + const p = record(value, field, ["x", "y"]); + return { x: coordinate(p.x, `${field}.x`), y: coordinate(p.y, `${field}.y`) }; +} + +function coordinate(value: unknown, field: string): number { + if (typeof value !== "number" || !Number.isFinite(value)) fail(`${field} must be a number`); + if (value < PRODUCE_POSITION_MIN || value > MAX_PRODUCE_POSITION) { + fail(`${field} must be between ${PRODUCE_POSITION_MIN} and ${MAX_PRODUCE_POSITION} (percent of frame)`); + } + return value; +} + +function storageKey(value: unknown, field: string): string { + if (typeof value !== "string" || !STORAGE_KEY.test(value) || value.includes("..")) { + fail(`${field} must be a content-addressed storage key`); + } + return value; +} + +function overlay(value: unknown, field: string): ProduceOverlay { + if (typeof value !== "object" || value === null || Array.isArray(value)) fail(`${field} must be an object`); + const kind = (value as Record).kind; + if (kind !== "lower_third" && kind !== "logo" && kind !== "clock") { + fail(`${field}.kind must be one of ${PRODUCE_OVERLAY_KINDS.join(", ")}`); + } + // Per-kind allowlist: a clock has no text and no asset, a logo has no caption. + const allowed = kind === "lower_third" ? ["kind", "text", "position"] : kind === "logo" ? ["kind", "logoStorageKey", "position"] : ["kind", "position"]; + const o = record(value, field, allowed); + return { + kind, + ...(kind === "lower_third" + ? { text: text(o.text, `${field}.text`) } + : kind === "logo" + ? { logoStorageKey: storageKey(o.logoStorageKey, `${field}.logoStorageKey`) } + : {}), + ...(o.position === undefined ? {} : { position: position(o.position, `${field}.position`) }), + }; +} + +function voiceover(value: unknown, field: string): ProduceVoiceover { + const v = record(value, field, ["assetId", "gainDb", "duckOriginal"]); + if (typeof v.assetId !== "string" || !isUuid(v.assetId)) fail(`${field}.assetId must be a UUID`); + const gainDb = optional(v.gainDb, `${field}.gainDb`, (g, f) => { + if (typeof g !== "number" || !Number.isFinite(g) || g < PRODUCE_GAIN_DB_MIN || g > PRODUCE_GAIN_DB_MAX) { + fail(`${f} must be a number between ${PRODUCE_GAIN_DB_MIN} and ${PRODUCE_GAIN_DB_MAX} dB`); + } + return g; + }); + const duckOriginal = optional(v.duckOriginal, `${field}.duckOriginal`, (d, f) => { + if (typeof d !== "boolean") fail(`${f} must be a boolean`); + return d; + }); + return { assetId: v.assetId, ...(gainDb === undefined ? {} : { gainDb }), ...(duckOriginal === undefined ? {} : { duckOriginal }) }; +} + +/** Parse and canonicalize a produce request; throws BadRequestException on the first problem. */ +export function parseProduceRequest(value: unknown, field = "body"): ProduceRequest { + const b = record(value, field, ["overlays", "voiceover", "outputLabel", "lane"]); + if (b.overlays !== undefined && b.overlays !== null) { + if (!Array.isArray(b.overlays)) fail(`${field}.overlays must be an array`); + if (b.overlays.length > MAX_OVERLAYS) fail(`${field}.overlays allows at most ${MAX_OVERLAYS} entries`); + } + if (b.lane !== undefined && b.lane !== null && b.lane !== "local" && b.lane !== "remote") { + fail(`${field}.lane must be one of ${PRODUCE_LANES.join(", ")}`); + } + const outputLabel = optional(b.outputLabel, `${field}.outputLabel`, (l, f) => { + if (typeof l !== "string" || !LABEL.test(l)) fail(`${f} must be 1-${MAX_PRODUCE_LABEL} letters, digits, spaces, '_', '.', '-'`); + return l; + }); + const overlays = (b.overlays ?? []).map((item, index) => overlay(item, `${field}.overlays[${index}]`)); + return { + ...(overlays.length === 0 ? {} : { overlays }), + ...(b.voiceover === undefined || b.voiceover === null ? {} : { voiceover: voiceover(b.voiceover, `${field}.voiceover`) }), + ...(outputLabel === undefined ? {} : { outputLabel }), + ...(b.lane === undefined || b.lane === null ? {} : { lane: b.lane }), + }; +} diff --git a/apps/api/src/video/test-fixtures.ts b/apps/api/src/video/test-fixtures.ts index 63bdc2a..8e61d08 100644 --- a/apps/api/src/video/test-fixtures.ts +++ b/apps/api/src/video/test-fixtures.ts @@ -14,6 +14,8 @@ export interface FakeVideoDatabase { asset?: Record; job?: Record; transitions?: string[]; + /** Raw INSERT statements seen, for asserting what intake wrote. */ + statements?: Array<{ sql: string; values: readonly unknown[] }>; } function result(rows: Array>): QueryResult> { @@ -24,7 +26,7 @@ export function serviceWithDatabase( database: FakeVideoDatabase, probe: VideoServiceDependencies["probe"], storageOverride?: StorageService, -): { service: VideoService; appendAudit: VideoServiceDependencies["appendAudit"]; storage: StorageService; objects: Map } { +): { service: VideoService; appendAudit: VideoServiceDependencies["appendAudit"] & { mockClear(): void }; storage: StorageService; objects: Map } { const objects = new Map(); const storage: StorageService = storageOverride ?? { putContent: vi.fn(async (bytes: Buffer, contentType: string) => { @@ -43,6 +45,7 @@ export function serviceWithDatabase( presign: async () => { throw new Error("Encrypted source must be decrypted before probing"); }, }; database.transitions = []; + database.statements = []; const client = { async query(sql: string, values: readonly unknown[] = []): Promise>> { if (sql.includes("INSERT INTO video_assets")) { @@ -64,11 +67,17 @@ export function serviceWithDatabase( return result([database.asset]); } if (sql.includes("INSERT INTO video_processing_jobs")) { + database.statements?.push({ sql, values }); + const produce = sql.includes("input_json"); database.job = { id: jobId, workspace_id: values[0], video_asset_id: values[1], state: "queued", + // kind/input_json exist only since migration 039; produce rows carry them. + // pg returns jsonb already parsed, not as the JSON text we sent. + ...(produce ? { kind: values[2], input_json: JSON.parse(String(values[3])) } : {}), + produced_asset_id: null, progress_percent: 0, output_key: null, error: null, diff --git a/apps/api/src/video/video.controller.ts b/apps/api/src/video/video.controller.ts index f4612b6..5fd8233 100644 --- a/apps/api/src/video/video.controller.ts +++ b/apps/api/src/video/video.controller.ts @@ -19,10 +19,12 @@ import { parseId, parseLimit, requireObject, requireSession, requireString } fro import { RbacGuard, RequirePermission, type AuthenticatedRequest } from "../rbac/rbac.guard.js"; import { VideoService, + type ProduceRequestResult, type VideoLibraryPage, type VideoStatus, type VideoUploadResult, } from "./video.service.js"; +import { parseProduceRequest } from "./produce-request.js"; import { MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; import { loadDriveBytes, loadMatrixBytes, parseMxcUri } from "./import-sources.js"; @@ -91,6 +93,20 @@ export class VideoController { return this.video.listVideos(requireSession(request).workspaceId, cursor, parseLimit(limit, 50, 200)); } + @Post(":id/produce") + @RequirePermission("media.manage") + async produce( + @Req() request: Request & AuthenticatedRequest, + @Param("id") id: string, + @Body() body: unknown, + ): Promise { + const session = requireSession(request); + // Strict allowlist parse: unknown keys are rejected, never ignored, so no + // field can ride into input_json (and on into a worker filtergraph). + const produce = parseProduceRequest(body); + return this.video.produceVideo(parseId(id), session.workspaceId, produce, session.userId); + } + @Get(":id/status") @RequirePermission("media.view") async status( diff --git a/apps/api/src/video/video.service.test.ts b/apps/api/src/video/video.service.test.ts index c76c637..e08c310 100644 --- a/apps/api/src/video/video.service.test.ts +++ b/apps/api/src/video/video.service.test.ts @@ -3,6 +3,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { VideoService } from "./video.service.js"; import { assetId, serviceWithDatabase, workspaceId, type FakeVideoDatabase } from "./test-fixtures.js"; import { MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; +import type { ProduceRequest } from "./produce-request.js"; afterEach(() => vi.unstubAllEnvs()); @@ -26,6 +27,7 @@ describe("video byte ingest and processing-job state machine", () => { expect(uploaded.videoAsset).toMatchObject({ sizeBytes: bytes.length, width: 1280, height: 720, durationMs: 42_000, codec: "h264" }); expect(uploaded.processingJob).toMatchObject({ state: "complete", progressPercent: 100, outputKey: uploaded.videoAsset.storageKey }); expect(uploaded.processingJob.processingNote).toContain("Transcode skipped"); + // ...and the produce INSERT is the only thing that followed it. expect(database.transitions).toEqual(["queued", "probing", "transcoding", "complete"]); await expect(stat(inputPath)).rejects.toMatchObject({ code: "ENOENT" }); expect(appendAudit).toHaveBeenCalledWith(workspaceId, expect.objectContaining({ action: "video.uploaded", targetId: assetId })); @@ -98,6 +100,59 @@ describe("video byte ingest and processing-job state machine", () => { expect(objects.get(uploaded.videoAsset.storageKey)).toEqual(bytes); }); + it("queues a compose job with the canonicalized request in input_json and audits it", async () => { + const database: FakeVideoDatabase = {}; + const { service, appendAudit } = serviceWithDatabase(database, async () => ({})); + await service.uploadVideoBytes(workspaceId, Buffer.from("source"), "video/mp4"); + appendAudit.mockClear(); + const request: ProduceRequest = { + overlays: [{ kind: "lower_third", text: "Live now", position: { x: 0, y: 100 } }], + outputLabel: "Recap", + }; + const produced = await service.produceVideo(assetId, workspaceId, request, "10000000-0000-4000-8000-000000000009"); + // kind is compose unless a voiceover is present, then mix. + expect(produced.job).toMatchObject({ state: "queued", kind: "compose", videoAssetId: assetId }); + // Nothing produced yet: the job points at no output asset until the worker finishes. + expect("producedAssetId" in produced.job).toBe(false); + expect(produced.videoAsset.id).toBe(assetId); + expect(database.statements?.at(-1)).toMatchObject({ + sql: expect.stringContaining("INSERT INTO video_processing_jobs (workspace_id, video_asset_id, state, kind, input_json)"), + values: [workspaceId, assetId, "compose", JSON.stringify({ ...request, requestedBy: "10000000-0000-4000-8000-000000000009" })], + }); + // input_json is passed as jsonb text and never leaves the tenant transaction. + expect(JSON.parse(String(database.statements?.at(-1)?.values[3]))).toEqual({ overlays: [{ kind: "lower_third", text: "Live now", position: { x: 0, y: 100 } }], outputLabel: "Recap", requestedBy: "10000000-0000-4000-8000-000000000009" }); + expect(appendAudit).toHaveBeenCalledWith(workspaceId, expect.objectContaining({ + action: "video.produce_requested", + targetId: assetId, + diff: expect.objectContaining({ kind: "compose", overlayCount: 1, voiceover: null, outputLabel: "Recap" }), + })); + // The produce row never disturbs the probe state machine of the source: + // the fake holds one job slot, so the probe transitions are all still there. + // ...and the produce INSERT is the only thing that followed it. + expect(database.transitions).toEqual(["queued", "probing", "transcoding", "complete", "queued"]); + }); + + it("queues a mix job when a voiceover is present and does not touch the source asset", async () => { + const database: FakeVideoDatabase = {}; + const { service } = serviceWithDatabase(database, async () => ({})); + await service.uploadVideoBytes(workspaceId, Buffer.from("source"), "video/mp4"); + await service.produceVideo(assetId, workspaceId, { + voiceover: { assetId: "20000000-0000-4000-8000-000000000004", gainDb: -6, duckOriginal: true }, + lane: "remote", + }, "10000000-0000-4000-8000-000000000009"); + expect(database.statements?.at(-1)?.values[2]).toBe("mix"); + expect(JSON.parse(String(database.statements?.at(-1)?.values[3]))).toEqual({ + voiceover: { assetId: "20000000-0000-4000-8000-000000000004", gainDb: -6, duckOriginal: true }, + lane: "remote", + requestedBy: "10000000-0000-4000-8000-000000000009", + }); + }); + + it("refuses to produce from an unknown asset", async () => { + const { service } = serviceWithDatabase({}, async () => ({})); + await expect(service.produceVideo("30000000-0000-4000-8000-000000000005", workspaceId, {}, "u")).rejects.toMatchObject({ status: 404 }); + }); + it("rejects empty or non-video uploads before storage", async () => { const { service, storage } = serviceWithDatabase({}, async () => ({})); await expect(service.uploadVideoBytes(workspaceId, Buffer.alloc(0), "video/mp4")).rejects.toMatchObject({ status: 400 }); diff --git a/apps/api/src/video/video.service.ts b/apps/api/src/video/video.service.ts index 4147579..72cfd68 100644 --- a/apps/api/src/video/video.service.ts +++ b/apps/api/src/video/video.service.ts @@ -7,6 +7,7 @@ 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 { type ProduceRequest } from "./produce-request.js"; import { MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js"; const NO_OP_TRANSCODE_NOTE = "Transcode skipped: original stored object retained; rendering belongs to the media-worker compositor."; @@ -28,17 +29,35 @@ export interface VideoAssetRecord { thumbnailKey?: string; } +export type VideoProcessingJobKind = "probe" | "compose" | "mix"; + export interface VideoProcessingJobRecord { id: string; workspaceId: string; videoAssetId: string; state: VideoProcessingJobState; + /** What the media-worker should do; 'probe' is the ingest default. */ + kind: VideoProcessingJobKind; + /** Producer request (kind compose/mix): { ...request, requestedBy }. */ + input?: ProduceJobInput; + /** The video_assets row this job writes when it finishes. */ + producedAssetId?: string; progressPercent?: number; outputKey?: string; error?: string; processingNote?: string; } +export interface ProduceJobInput extends ProduceRequest { + requestedBy: string; +} + +export interface ProduceRequestResult { + /** The newly queued compose/mix job. */ + job: VideoProcessingJobRecord; + videoAsset: VideoAssetRecord; +} + export interface VideoPublicationSummary { connectionId: string; remoteId: string; @@ -176,7 +195,8 @@ export class VideoService { `SELECT va.*, j.id AS job_id, j.video_asset_id AS job_video_asset_id, j.state AS job_state, j.progress_percent AS job_progress_percent, - j.output_key AS job_output_key, j.error AS job_error + j.output_key AS job_output_key, j.error AS job_error, + j.kind AS job_kind, j.input_json AS job_input_json, j.produced_asset_id AS job_produced_asset_id FROM video_assets va LEFT JOIN LATERAL ( SELECT * FROM video_processing_jobs @@ -345,6 +365,43 @@ export class VideoService { }); } + /** + * Queue a derived asset: burn overlays (and optionally mix a voiceover) over an + * existing asset. Validates nothing itself — the controller parses the request + * with parseProduceRequest before calling, and the media-worker re-validates. + */ + async produceVideo( + assetId: string, + workspaceId: string, + request: ProduceRequest, + requestedBy: string, + ): Promise { + const input: ProduceJobInput = { ...request, requestedBy }; + const kind: VideoProcessingJobKind = request.voiceover === undefined ? "compose" : "mix"; + const created = await this.dependencies.tenantTransaction(workspaceId, async (client) => { + const source = await client.query("SELECT * FROM video_assets WHERE id = $1", [assetId]); + if (source.rowCount !== 1) throw new NotFoundException(`video ${assetId} not found`); + const jobResult = await client.query( + `INSERT INTO video_processing_jobs (workspace_id, video_asset_id, state, kind, input_json) + VALUES ($1, $2, 'queued', $3, $4::jsonb) RETURNING *`, + [workspaceId, assetId, kind, JSON.stringify(input)], + ); + return { + videoAsset: assetFromRow(requiredRow(source.rows[0], "source video asset")), + job: jobFromRow(requiredRow(jobResult.rows[0], "created produce job")), + }; + }); + await this.audit(workspaceId, "video.produce_requested", assetId, { + kind, + jobId: created.job.id, + overlayCount: request.overlays?.length ?? 0, + voiceover: request.voiceover === undefined ? null : { assetId: request.voiceover.assetId, duckOriginal: request.voiceover.duckOriginal ?? false }, + outputLabel: request.outputLabel ?? null, + lane: request.lane ?? null, + }); + return created; + } + async deleteVideo(assetId: string, workspaceId: string): Promise { // Objects are shared by content hash across assets/workspaces. Retain for future // reference-aware GC; deleting here could destroy another asset's source. @@ -373,7 +430,7 @@ export class VideoService { private async audit( workspaceId: string, - action: "video.uploaded" | "video.probed" | "video.deleted" | "video.imported", + action: "video.uploaded" | "video.probed" | "video.deleted" | "video.imported" | "video.produce_requested", targetId: string, diff: Record, ): Promise { @@ -420,6 +477,7 @@ function assetFromRow(row: Record): VideoAssetRecord { ...optionalString("codec", row.codec), ...optionalString("audioCodec", row.audio_codec), ...optionalString("thumbnailKey", row.thumbnail_key), + ...optionalString("derivedFrom", row.derived_from), }; } @@ -429,6 +487,9 @@ function jobFromRow(row: Record): VideoProcessingJobRecord { workspaceId: String(row.workspace_id), videoAssetId: String(row.video_asset_id), state: processingState(row.state), + kind: processingKind(row.kind), + ...(row.input_json === null || row.input_json === undefined ? {} : { input: requiredRow(row.input_json, "job input") as unknown as ProduceJobInput }), + ...optionalString("producedAssetId", row.produced_asset_id), ...optionalNumber("progressPercent", row.progress_percent), ...optionalString("outputKey", row.output_key), ...optionalString("error", row.error), @@ -444,6 +505,9 @@ function joinedJobFromRow(row: Record): VideoProcessingJobRecor progress_percent: row.job_progress_percent, output_key: row.job_output_key, error: row.job_error, + kind: row.job_kind, + input_json: row.job_input_json, + produced_asset_id: row.job_produced_asset_id, }); return { ...job, @@ -469,6 +533,13 @@ function publicationFromJson(row: Record): VideoPublicationSumm }; } +function processingKind(value: unknown): VideoProcessingJobKind { + if (value === "probe" || value === "compose" || value === "mix") return value; + // Rows created before migration 039 have no kind and are probe jobs. + if (value === null || value === undefined) return "probe"; + throw new Error(`invalid video processing job kind: ${String(value)}`); +} + function processingState(value: unknown): VideoProcessingJobState { if ( value === "queued" ||