import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { BadRequestException, Inject, Injectable, NotFoundException, Optional, ServiceUnavailableException } from "@nestjs/common"; 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 { 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."; export type VideoProcessingJobState = "queued" | "probing" | "transcoding" | "complete" | "failed"; export interface VideoAssetRecord { id: string; workspaceId: string; storageKey: string; mimeType: string; sizeBytes: number; width?: number; height?: number; durationMs?: number; frameRate?: number; codec?: string; audioCodec?: string; 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; platform: string; publishedAt: Date; } export interface VideoLibraryEntryRecord { videoAsset: VideoAssetRecord; processingState: VideoProcessingJobState; publications: VideoPublicationSummary[]; } export interface VideoLibraryPage { items: VideoLibraryEntryRecord[]; nextCursor?: string; } export interface VideoStatus { videoAsset: VideoAssetRecord; processingJob: VideoProcessingJobRecord; } export type VideoUploadResult = VideoStatus; type TenantTransaction = (workspaceId: string, fn: (client: PoolClient) => Promise) => Promise; type AuditAppender = (workspaceId: string, event: AuditEvent) => Promise; export interface VideoServiceDependencies { tenantTransaction: TenantTransaction; appendAudit: AuditAppender; probe: (inputPath: string) => Promise; storage?: StorageService; } export const VIDEO_SERVICE_DEPENDENCIES = Symbol("VIDEO_SERVICE_DEPENDENCIES"); const DEFAULT_DEPENDENCIES: VideoServiceDependencies = { tenantTransaction: withTenant, appendAudit: appendTenantAudit, probe: probeVideoFile, }; @Injectable() export class VideoService { private readonly dependencies: VideoServiceDependencies; private storage?: StorageService; constructor( @Optional() @Inject(VIDEO_SERVICE_DEPENDENCIES) dependencies?: Partial, ) { this.dependencies = { ...DEFAULT_DEPENDENCIES, ...dependencies }; this.storage = dependencies?.storage; } async uploadVideoBytes(workspaceId: string, bytes: Buffer, mimeType: string): Promise { validateUpload("pending", mimeType, bytes.length); const storage = this.requireStorage(); const stored = await storage.putContent(bytes, mimeType).catch(() => { throw new ServiceUnavailableException("Video storage unavailable: could not store encrypted upload in S3/MinIO."); }); return this.registerAndProbe(workspaceId, stored.key, stored.contentType, stored.sizeBytes); } async importVideoBytes( workspaceId: string, bytes: Buffer, mimeType: string, source: "drive" | "chat", sourceRef: string, ): Promise { validateUpload("pending", mimeType, bytes.length); const storage = this.requireStorage(); const stored = await storage.putContent(bytes, mimeType).catch(() => { throw new ServiceUnavailableException("Video storage unavailable: could not store imported video."); }); const result = await this.registerAndProbe(workspaceId, stored.key, stored.contentType, stored.sizeBytes); await this.audit(workspaceId, "video.imported", result.videoAsset.id, { source, sourceRef, sizeBytes: bytes.length }); return result; } async uploadVideo( workspaceId: string, storageKey: string, mimeType: string, sizeBytes: number, ): Promise { validateUpload(storageKey, mimeType, sizeBytes); // Legacy metadata registration is accepted only for an object that really exists. const bytes = await this.requireStorage().getObject(storageKey).catch(() => { throw new BadRequestException("storageKey must identify an existing readable encrypted object; upload a file first"); }); if (bytes.length !== sizeBytes) throw new BadRequestException("sizeBytes does not match the stored object"); return this.registerAndProbe(workspaceId, storageKey, mimeType, sizeBytes); } private async registerAndProbe( workspaceId: string, storageKey: string, mimeType: string, sizeBytes: number, ): Promise { const uploaded = 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, storageKey, mimeType, sizeBytes], ); const videoAsset = assetFromRow(requiredRow(assetResult.rows[0], "created video asset")); const jobResult = await client.query( `INSERT INTO video_processing_jobs (workspace_id, video_asset_id, state) VALUES ($1, $2, 'queued') RETURNING *`, [workspaceId, videoAsset.id], ); return { videoAsset, processingJob: jobFromRow(requiredRow(jobResult.rows[0], "created video processing job")), }; }); try { await this.probeVideo(uploaded.videoAsset.id, workspaceId); } catch (error) { // Probe errors are recorded on the job; return its ID and failure in status. const status = await this.getVideoStatus(uploaded.videoAsset.id, workspaceId); if (status.processingJob.state !== "failed") throw error; } await this.audit(workspaceId, "video.uploaded", uploaded.videoAsset.id, { mimeType, sizeBytes, }); return this.getVideoStatus(uploaded.videoAsset.id, workspaceId); } async getVideoStatus(assetId: string, workspaceId: string): Promise { return this.dependencies.tenantTransaction(workspaceId, async (client) => { const result = await client.query( `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.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 WHERE video_asset_id = va.id ORDER BY created_at DESC, id DESC LIMIT 1 ) j ON true WHERE va.id = $1`, [assetId], ); if (result.rowCount !== 1) throw new NotFoundException(`video ${assetId} not found`); const row = requiredRow(result.rows[0], "video status"); if (row.job_id === null || row.job_id === undefined) { throw new Error(`video ${assetId} has no processing job`); } return { videoAsset: assetFromRow(row), processingJob: joinedJobFromRow(row) }; }); } async probeVideo(assetId: string, workspaceId: string): Promise { const claimed = await this.dependencies.tenantTransaction(workspaceId, async (client) => { const asset = await client.query("SELECT * FROM video_assets WHERE id = $1", [assetId]); if (asset.rowCount !== 1) throw new NotFoundException(`video ${assetId} not found`); const changed = await client.query( `UPDATE video_processing_jobs SET state = 'probing', progress_percent = 10, error = NULL, updated_at = now() WHERE id = ( SELECT id FROM video_processing_jobs WHERE video_asset_id = $1 ORDER BY created_at DESC, id DESC LIMIT 1 ) AND state = 'queued' RETURNING id`, [assetId], ); if (changed.rowCount !== 1) { const latest = await client.query( `SELECT state FROM video_processing_jobs WHERE video_asset_id = $1 ORDER BY created_at DESC, id DESC LIMIT 1`, [assetId], ); const state = latest.rows[0]?.state; throw new BadRequestException( state === undefined ? `video ${assetId} has no processing job` : `cannot probe video in processing state ${String(state)}`, ); } return { storageKey: String(asset.rows[0]?.storage_key), jobId: String(changed.rows[0]?.id), }; }); let metadata: VideoProbeResult; let videoAsset: VideoAssetRecord; try { const bytes = await this.requireStorage().getObject(claimed.storageKey); const directory = await mkdtemp(join(tmpdir(), "tmbh-video-")); try { const inputPath = join(directory, "input"); await writeFile(inputPath, bytes, { mode: 0o600 }); metadata = await this.dependencies.probe(inputPath); } finally { await rm(directory, { recursive: true, force: true }); } videoAsset = await this.dependencies.tenantTransaction(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 *`, [ assetId, metadata.width ?? null, metadata.height ?? null, metadata.durationMs ?? null, metadata.frameRate ?? null, metadata.codec ?? null, metadata.audioCodec ?? null, ], ); if (updated.rowCount !== 1) throw new NotFoundException(`video ${assetId} not found`); const transcoding = await client.query( `UPDATE video_processing_jobs SET state = 'transcoding', progress_percent = 90, error = NULL, updated_at = now() WHERE id = $1 AND state = 'probing' RETURNING id`, [claimed.jobId], ); if (transcoding.rowCount !== 1) throw new Error(`video processing job ${claimed.jobId} lost its claim`); return assetFromRow(requiredRow(updated.rows[0], "probed video asset")); }); // This slice deliberately retains the source. The compositor owns real transcoding. await this.dependencies.tenantTransaction(workspaceId, async (client) => { 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`, [claimed.jobId, claimed.storageKey], ); if (completed.rowCount !== 1) throw new Error(`video processing job ${claimed.jobId} lost its claim`); }); } catch (error) { const message = error instanceof Error ? error.message : String(error); await this.dependencies.tenantTransaction(workspaceId, async (client) => { await client.query( `UPDATE video_processing_jobs SET state = 'failed', error = $2, updated_at = now() WHERE id = $1 AND state IN ('probing', 'transcoding')`, [claimed.jobId, message], ); }); throw error; } await this.audit(workspaceId, "video.probed", assetId, { ...metadata, processingNote: NO_OP_TRANSCODE_NOTE }); return videoAsset; } async listVideos( workspaceId: string, cursor: string | undefined, limit: number, ): Promise { if (!Number.isInteger(limit) || limit < 1 || limit > 200) { throw new BadRequestException("limit must be an integer between 1 and 200"); } const decoded = cursor === undefined || cursor === "" ? undefined : decodeCursor(cursor); return this.dependencies.tenantTransaction(workspaceId, async (client) => { const result = await client.query( `SELECT va.*, COALESCE(j.state, 'queued') AS processing_state, COALESCE(( SELECT jsonb_agg(jsonb_build_object( 'connectionId', p.connection_id, 'remoteId', p.remote_id, 'platform', c.provider, 'publishedAt', p.published_at ) ORDER BY p.published_at DESC) FROM publications p JOIN drafts d ON d.id = p.draft_id JOIN connections c ON c.id = p.connection_id WHERE d.post ->> 'videoAssetId' = va.id::text OR d.post -> 'mediaAssetIds' ? va.id::text ), '[]'::jsonb) AS publications FROM video_assets va LEFT JOIN LATERAL ( SELECT state FROM video_processing_jobs WHERE video_asset_id = va.id ORDER BY created_at DESC, id DESC LIMIT 1 ) j ON true WHERE ($1::timestamptz IS NULL OR (va.created_at, va.id) < ($1::timestamptz, $2::uuid)) ORDER BY va.created_at DESC, va.id DESC LIMIT $3`, [decoded?.createdAt ?? null, decoded?.id ?? null, limit + 1], ); const rows = result.rows.map((row: unknown) => requiredRow(row, "video library entry")); const hasMore = rows.length > limit; const pageRows = rows.slice(0, limit); const items = pageRows.map(libraryEntryFromRow); const last = pageRows[pageRows.length - 1]; return { items, ...(hasMore && last !== undefined ? { nextCursor: encodeCursor({ createdAt: new Date(String(last.created_at)).toISOString(), id: String(last.id) }) } : {}), }; }); } /** * 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`); await assertLogoKeysOwned(client, workspaceId, request); 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. await this.dependencies.tenantTransaction(workspaceId, async (client) => { const removed = await client.query("DELETE FROM video_assets WHERE id = $1 RETURNING id", [assetId]); if (removed.rowCount !== 1) throw new NotFoundException(`video ${assetId} not found`); }); await this.audit(workspaceId, "video.deleted", assetId, {}); } private requireStorage(): StorageService { if (this.storage !== undefined) return this.storage; const config = s3ConfigFromEnv(); if (!config.accessKeyId?.trim() || !config.secretAccessKey?.trim()) { throw new ServiceUnavailableException( "Video storage is not configured: set S3_ACCESS_KEY and S3_SECRET_KEY, plus S3_ENDPOINT for MinIO.", ); } try { this.storage = new S3Storage(config); } catch { throw new ServiceUnavailableException("Video storage configuration is invalid; check S3 settings and S3_KEK."); } return this.storage; } private async audit( workspaceId: string, action: "video.uploaded" | "video.probed" | "video.deleted" | "video.imported" | "video.produce_requested", targetId: string, diff: Record, ): Promise { await this.dependencies.appendAudit(workspaceId, { actorUserId: null, actorRole: "api", action, targetType: "video_asset", targetId, diff, }); } } 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/")) { throw new BadRequestException("mimeType must be a video media type"); } if (!Number.isSafeInteger(sizeBytes) || sizeBytes < 1) { throw new BadRequestException("sizeBytes must be a positive safe integer"); } if (sizeBytes > MAX_VIDEO_UPLOAD_BYTES) throw new BadRequestException("video exceeds the 100 MiB upload limit"); } function requiredRow(row: unknown, description: string): Record { if (typeof row !== "object" || row === null || Array.isArray(row)) { throw new Error(`database did not return ${description}`); } return row as Record; } function assetFromRow(row: Record): VideoAssetRecord { return { id: String(row.id), workspaceId: String(row.workspace_id), storageKey: String(row.storage_key), mimeType: String(row.mime_type), sizeBytes: safeDatabaseNumber(row.size_bytes, "size_bytes"), ...optionalNumber("width", row.width), ...optionalNumber("height", row.height), ...optionalNumber("durationMs", row.duration_ms), ...optionalNumber("frameRate", row.frame_rate), ...optionalString("codec", row.codec), ...optionalString("audioCodec", row.audio_codec), ...optionalString("thumbnailKey", row.thumbnail_key), ...optionalString("derivedFrom", row.derived_from), }; } function jobFromRow(row: Record): VideoProcessingJobRecord { return { id: String(row.id), 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), }; } function joinedJobFromRow(row: Record): VideoProcessingJobRecord { const job = jobFromRow({ id: row.job_id, workspace_id: row.workspace_id, video_asset_id: row.job_video_asset_id, state: row.job_state, 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, ...(job.state === "complete" && job.outputKey === row.storage_key ? { processingNote: NO_OP_TRANSCODE_NOTE } : {}), }; } function libraryEntryFromRow(row: Record): VideoLibraryEntryRecord { const rawPublications = Array.isArray(row.publications) ? row.publications : []; return { videoAsset: assetFromRow(row), processingState: processingState(row.processing_state), publications: rawPublications.map((publication) => publicationFromJson(requiredRow(publication, "publication"))), }; } function publicationFromJson(row: Record): VideoPublicationSummary { return { connectionId: String(row.connectionId), remoteId: String(row.remoteId), platform: String(row.platform), publishedAt: new Date(String(row.publishedAt)), }; } 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" || value === "probing" || value === "transcoding" || value === "complete" || value === "failed" ) { return value; } throw new Error(`invalid video processing state: ${String(value)}`); } function safeDatabaseNumber(value: unknown, field: string): number { const number = typeof value === "number" ? value : typeof value === "string" ? Number(value) : NaN; if (!Number.isSafeInteger(number)) throw new Error(`${field} is outside JavaScript's safe integer range`); return number; } function optionalNumber(key: K, value: unknown): Partial> { if (value === null || value === undefined) return {}; const number = typeof value === "number" ? value : typeof value === "string" ? Number(value) : NaN; if (!Number.isFinite(number)) throw new Error(`${key} is not numeric`); return { [key]: number } as Record; } function optionalString(key: K, value: unknown): Partial> { return value === null || value === undefined ? {} : { [key]: String(value) } as Record; } interface VideoCursor { createdAt: string; id: string; } function encodeCursor(cursor: VideoCursor): string { return Buffer.from(JSON.stringify(cursor), "utf8").toString("base64url"); } function decodeCursor(cursor: string): VideoCursor { try { const decoded = JSON.parse(Buffer.from(cursor, "base64url").toString("utf8")) as unknown; const row = requiredRow(decoded, "video cursor"); const createdAt = String(row.createdAt); const id = String(row.id); if (!Number.isFinite(Date.parse(createdAt)) || !isUuid(id)) throw new Error("invalid fields"); return { createdAt, id }; } catch { throw new BadRequestException("cursor is invalid"); } } function isUuid(value: string): boolean { return /^[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(value); } /** * Storage keys are content-addressed in one shared bucket, so a well-formed key * alone proves nothing about who owns the object. A logo overlay may only use * an image this workspace holds: an attachment, a group asset or its branding. */ async function assertLogoKeysOwned( client: { query: (sql: string, values: unknown[]) => Promise<{ rowCount: number | null }> }, workspaceId: string, request: ProduceRequest, ): Promise { const keys = [...new Set((request.overlays ?? []).flatMap((overlay) => (overlay.logoStorageKey === undefined ? [] : [overlay.logoStorageKey])))]; for (const key of keys) { const owned = await client.query( `SELECT 1 WHERE EXISTS (SELECT 1 FROM media_assets WHERE storage_key = $1 AND workspace_id = $2) OR EXISTS (SELECT 1 FROM group_assets WHERE asset_id = $1 AND workspace_id = $2) OR EXISTS (SELECT 1 FROM workspaces WHERE id = $2 AND branding->>'logoAssetKey' = $1)`, [key, workspaceId], ); if (owned.rowCount !== 1) throw new BadRequestException("logoStorageKey is not an image of this collective"); } }