glass beadian
b24ebc5862
From the logic audit (HEAD 04f5f7e):
- P0 CORS: enableCors methods lacked PUT — browsers could not chunk-upload
at all (cross-origin preflight refused)
- P0 chunked: validateUpload enforced the 2 GiB whole-body cap on chunked
completes, so every >2 GiB chunked upload failed after full upload
- P1 presignPut: the SDK's flexible-checksum default added an
x-amz-checksum-crc32 header browsers cannot reproduce; S3Client now uses
requestChecksumCalculation WHEN_REQUIRED
- P1 refresh rotation race: concurrent refreshes both succeeded (consume
ignored rowCount) forking the token family; consume now returns whether
it won and the loser revokes the chain
- P2 logout: awaits the revocation and ignores missing/forged bearers
honestly
776 lines
32 KiB
TypeScript
776 lines
32 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
||
import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises";
|
||
import { tmpdir } from "node:os";
|
||
import { join } from "node:path";
|
||
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,
|
||
directEnabled, directObjectKey,
|
||
} from "../storage/storage.js";
|
||
import { type ProduceRequest } from "./produce-request.js";
|
||
import { MAX_CHUNKED_UPLOAD_BYTES, MAX_VIDEO_UPLOAD_BYTES } from "./upload-limits.js";
|
||
|
||
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 | null;
|
||
}
|
||
|
||
export type VideoUploadResult = VideoStatus;
|
||
|
||
/** GET /video/direct/config — the browser's capability check before it seals. */
|
||
export interface DirectUploadConfig {
|
||
enabled: boolean;
|
||
publicKeyJwk: { kty: "EC"; crv: "P-256"; x: string; y: string } | null;
|
||
maxBytes: number;
|
||
}
|
||
|
||
/** POST /video/direct/init response. Nothing is persisted until complete. */
|
||
export interface DirectUploadInit {
|
||
uploadId: string;
|
||
objectKey: string;
|
||
putUrl: string;
|
||
}
|
||
|
||
/** POST /video/direct/complete request (HEAD-checked, then registered + queued). */
|
||
export interface DirectUploadCompleteInput {
|
||
uploadId: string;
|
||
objectKey: string;
|
||
sizeBytes: number;
|
||
mimeType: string;
|
||
filename: string;
|
||
}
|
||
|
||
/** Presigned PUT lifetime for a browser upload (spec: 900 s). */
|
||
export const DIRECT_PUT_TTL_SECONDS = 900;
|
||
const DIRECT_KEY_PREFIX = "direct/";
|
||
const DIRECT_OBJECT_SUFFIX = ".enc";
|
||
|
||
type TenantTransaction = <T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>) => Promise<T>;
|
||
type AuditAppender = (workspaceId: string, event: AuditEvent) => Promise<string>;
|
||
|
||
export interface VideoServiceDependencies {
|
||
tenantTransaction: TenantTransaction;
|
||
appendAudit: AuditAppender;
|
||
probe: (inputPath: string) => Promise<VideoProbeResult>;
|
||
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<VideoServiceDependencies>,
|
||
) {
|
||
this.dependencies = { ...DEFAULT_DEPENDENCIES, ...dependencies };
|
||
this.storage = dependencies?.storage;
|
||
}
|
||
|
||
async uploadVideoBytes(workspaceId: string, bytes: Buffer, mimeType: string): Promise<VideoUploadResult> {
|
||
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<VideoUploadResult> {
|
||
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;
|
||
}
|
||
|
||
/**
|
||
* Chunked-ingest tail end: seal + store + probe an already-assembled file on
|
||
* local disk (see chunked-upload.ts). The bytes never sat in a request body,
|
||
* so the 2 GiB whole-body cap does not apply — MAX_CHUNKED_UPLOAD_BYTES does.
|
||
*
|
||
* MEMORY: reading the assembled file into one Buffer peaks around 2× file
|
||
* size of RAM (seal takes a Buffer and S3 takes the sealed copy). A 20 GB
|
||
* upload therefore needs a host with ≥ 48 GB RAM; /tmp must also hold the
|
||
* file itself.
|
||
*/
|
||
async uploadVideoFromPath(
|
||
workspaceId: string,
|
||
inputPath: string,
|
||
mimeType: string,
|
||
sizeBytes: number,
|
||
): Promise<VideoUploadResult> {
|
||
validateUpload("pending", mimeType, sizeBytes);
|
||
if (sizeBytes > MAX_CHUNKED_UPLOAD_BYTES) {
|
||
throw new BadRequestException("video exceeds the 100 GiB chunked upload limit");
|
||
}
|
||
const onDisk = await stat(inputPath).catch(() => {
|
||
throw new BadRequestException("assembled upload file is missing on disk");
|
||
});
|
||
if (onDisk.size !== sizeBytes) {
|
||
throw new BadRequestException(
|
||
`assembled upload is ${onDisk.size} bytes on disk but declared ${sizeBytes}`,
|
||
);
|
||
}
|
||
const storage = this.requireStorage();
|
||
const bytes = await readFile(inputPath).catch(() => {
|
||
throw new ServiceUnavailableException("could not read the assembled upload from disk");
|
||
});
|
||
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);
|
||
}
|
||
|
||
// -------------------------------------------------------------------------
|
||
// Direct-to-R2 (browser-sealed SPENC2): the api mints the presign and the
|
||
// rows; the payload goes browser → R2 → media-worker. No video bytes ever
|
||
// touch this process, so the chunked/whole-body memory ceilings do not apply.
|
||
// -------------------------------------------------------------------------
|
||
|
||
/** Whether this deployment can unwrap browser-sealed frames (DIRECT_EC_PRIVATE_KEY). */
|
||
async directUploadConfig(): Promise<DirectUploadConfig> {
|
||
const storage = this.requireStorage();
|
||
const publicKeyJwk = await storage.directPublicJwk();
|
||
return {
|
||
enabled: directEnabled(),
|
||
publicKeyJwk,
|
||
maxBytes: MAX_CHUNKED_UPLOAD_BYTES,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Mint the presigned PUT for a browser upload. The size is bounded up front
|
||
* so an oversized pick fails before any bytes are sealed; nothing is
|
||
* persisted — the asset row only appears on complete, so an abandoned PUT
|
||
* leaves an orphan object (no row), never a dangling row without an object.
|
||
*/
|
||
async initDirectUpload(
|
||
workspaceId: string,
|
||
input: { filename: string; mimeType: string; sizeBytes: number },
|
||
): Promise<DirectUploadInit> {
|
||
validateDirectUpload(input);
|
||
if ((await this.requireStorage().directPublicJwk()) === null) {
|
||
throw new ServiceUnavailableException("DIRECT_EC_PRIVATE_KEY is not configured");
|
||
}
|
||
// ONE uuid: the object key binds to the upload id so /complete can pin it.
|
||
const uploadId = randomUUID();
|
||
const objectKey = `direct/${uploadId}.enc`;
|
||
const putUrl = await this.requireStorage()
|
||
.presignPut(objectKey, input.mimeType, DIRECT_PUT_TTL_SECONDS)
|
||
.catch(() => { throw new ServiceUnavailableException("Video storage is unavailable: could not presign the direct PUT"); });
|
||
return { uploadId, objectKey, putUrl };
|
||
}
|
||
|
||
/**
|
||
* Register a completed direct upload. The object is HEAD-checked (exists,
|
||
* exact declared size) BEFORE any row is written, so metadata can never point
|
||
* at a missing or partial PUT — the same gate /files/direct/complete uses.
|
||
*
|
||
* The probe deliberately does NOT run here: the insert queues kind 'probe'
|
||
* for the media-worker, which downloads the sealed object off R2, decrypts it
|
||
* and runs ffprobe itself. This box never touches video bytes on this path.
|
||
*/
|
||
async completeDirectUpload(workspaceId: string, input: DirectUploadCompleteInput): Promise<VideoStatus> {
|
||
if (!isUuid(input.uploadId)) throw new BadRequestException("uploadId must be a UUID");
|
||
if (input.objectKey !== `${DIRECT_KEY_PREFIX}${input.uploadId}${DIRECT_OBJECT_SUFFIX}`) {
|
||
throw new BadRequestException("objectKey must be direct/<uploadId>.enc");
|
||
}
|
||
validateDirectUpload(input);
|
||
|
||
const head = await this.requireStorage().headObject(input.objectKey).catch(() => {
|
||
throw new ServiceUnavailableException("Video storage is unavailable: could not verify the uploaded object");
|
||
});
|
||
if (!head.exists || head.contentLength !== input.sizeBytes) {
|
||
throw new BadRequestException("uploaded object is missing or its size does not match sizeBytes");
|
||
}
|
||
|
||
const registered = await this.dependencies.tenantTransaction(workspaceId, async (client) => {
|
||
const assetResult = await client.query(
|
||
`INSERT INTO video_assets (workspace_id, storage_key, mime_type, size_bytes)
|
||
VALUES ($1, $2, $3, $4) RETURNING *`,
|
||
[workspaceId, input.objectKey, input.mimeType, input.sizeBytes],
|
||
);
|
||
const videoAsset = assetFromRow(requiredRow(assetResult.rows[0], "created video asset"));
|
||
// kind 'probe' is 039's default: the media-worker claims it and runs ffprobe.
|
||
const jobResult = await client.query(
|
||
`INSERT INTO video_processing_jobs (workspace_id, video_asset_id, state, kind)
|
||
VALUES ($1, $2, 'queued', 'probe') RETURNING *`,
|
||
[workspaceId, videoAsset.id],
|
||
);
|
||
return {
|
||
videoAsset,
|
||
processingJob: jobFromRow(requiredRow(jobResult.rows[0], "created video processing job")),
|
||
};
|
||
});
|
||
|
||
await this.audit(workspaceId, "video.uploaded", registered.videoAsset.id, {
|
||
mimeType: input.mimeType,
|
||
sizeBytes: input.sizeBytes,
|
||
uploadPath: "direct",
|
||
});
|
||
// The worker probes; status reports the queued job the drawer polls on.
|
||
return this.getVideoStatus(registered.videoAsset.id, workspaceId);
|
||
}
|
||
|
||
async uploadVideo(
|
||
workspaceId: string,
|
||
storageKey: string,
|
||
mimeType: string,
|
||
sizeBytes: number,
|
||
): Promise<VideoUploadResult> {
|
||
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<VideoUploadResult> {
|
||
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 === null || 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<VideoStatus> {
|
||
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");
|
||
// A jobless asset is a legitimate state (pre-probe direct registrations,
|
||
// legacy rows): surface it instead of 500ing the library click.
|
||
if (row.job_id === null || row.job_id === undefined) {
|
||
return { videoAsset: assetFromRow(row), processingJob: null };
|
||
}
|
||
return { videoAsset: assetFromRow(row), processingJob: joinedJobFromRow(row) };
|
||
});
|
||
}
|
||
|
||
async probeVideo(assetId: string, workspaceId: string): Promise<VideoAssetRecord> {
|
||
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<VideoLibraryPage> {
|
||
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<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.
|
||
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<string, unknown>,
|
||
): Promise<void> {
|
||
await this.dependencies.appendAudit(workspaceId, {
|
||
actorUserId: null,
|
||
actorRole: "api",
|
||
action,
|
||
targetType: "video_asset",
|
||
targetId,
|
||
diff,
|
||
});
|
||
}
|
||
}
|
||
|
||
/** Shared validation for the direct init/complete payloads (no sniffing: the declared type stands). */
|
||
function validateDirectUpload(input: { filename?: string; mimeType: string; sizeBytes: number }): void {
|
||
if (input.filename !== undefined && (input.filename.trim() === "" || input.filename.length > 255)) {
|
||
throw new BadRequestException("filename must be 1-255 characters");
|
||
}
|
||
if (!input.mimeType.toLowerCase().startsWith("video/")) {
|
||
throw new BadRequestException("mimeType must be a video media type");
|
||
}
|
||
if (!Number.isSafeInteger(input.sizeBytes) || input.sizeBytes < 1) {
|
||
throw new BadRequestException("sizeBytes must be a positive safe integer");
|
||
}
|
||
if (input.sizeBytes > MAX_CHUNKED_UPLOAD_BYTES) {
|
||
throw new BadRequestException("video exceeds the 100 GiB chunked upload limit");
|
||
}
|
||
}
|
||
|
||
function validateUpload(storageKey: string, mimeType: string, sizeBytes: number): void {
|
||
if (storageKey.trim() === "") throw new BadRequestException("storageKey must be a non-empty string");
|
||
if (!mimeType.toLowerCase().startsWith("video/")) {
|
||
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_CHUNKED_UPLOAD_BYTES) throw new BadRequestException("video exceeds the chunked upload limit");
|
||
}
|
||
|
||
function requiredRow(row: unknown, description: string): Record<string, unknown> {
|
||
if (typeof row !== "object" || row === null || Array.isArray(row)) {
|
||
throw new Error(`database did not return ${description}`);
|
||
}
|
||
return row as Record<string, unknown>;
|
||
}
|
||
|
||
function assetFromRow(row: Record<string, unknown>): 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<string, unknown>): 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<string, unknown>): 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<string, unknown>): 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<string, unknown>): 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<K extends string>(key: K, value: unknown): Partial<Record<K, number>> {
|
||
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<K, number>;
|
||
}
|
||
|
||
function optionalString<K extends string>(key: K, value: unknown): Partial<Record<K, string>> {
|
||
return value === null || value === undefined ? {} : { [key]: String(value) } as Record<K, string>;
|
||
}
|
||
|
||
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);
|
||
}
|