feat: media attachments backend + producer job intake (t1+t3)

Rich media (t1): media_assets (migration 038) + /media module — multipart
upload with magic-byte sniffing, size caps, auth-scoped serve with
ETag/same-origin; the publishing 'text only' gate lifted to accept
0..4 validated mediaAssetIds with optional positional altTexts.

Producer intake (t3): video_processing_jobs gains kind compose|mix +
input_json + produced_asset_id (migration 039), video_assets.derived_from;
POST /video/:id/produce validates overlays/voiceover strictly and queues
the job for the media-worker runner (landing separately).
This commit is contained in:
glass beadian 2026-10-14 20:00:31 +00:00
parent abb343284a
commit 71db0a08b2
15 changed files with 1276 additions and 9 deletions

View file

@ -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<ProduceRequestResult> {
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<void> {
// 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<string, unknown>,
): Promise<void> {
@ -420,6 +477,7 @@ function assetFromRow(row: Record<string, unknown>): 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<string, unknown>): 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<string, unknown>): 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<string, unknown>): 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" ||