diff --git a/apps/api/src/media-http.ts b/apps/api/src/media-http.ts new file mode 100644 index 0000000..2dd7369 --- /dev/null +++ b/apps/api/src/media-http.ts @@ -0,0 +1,265 @@ +import { createReadStream } from "node:fs"; +import { mkdir, stat } from "node:fs/promises"; +import path from "node:path"; +import { Readable } from "node:stream"; +import type { MediaServices } from "@rakazo/adapters"; +import type { Actor } from "@rakazo/contracts"; +import { IsolationError, MediaStateConflictError } from "@rakazo/db"; +import { assertInside, PathOutsideRootError, resolveInside } from "@rakazo/media/node"; +import type { Context, Hono } from "hono"; +import { cancelBody } from "./http-body.js"; +import { MEDIA_CHUNK_TIMEOUT_MS, UploadChunkError, writeUploadChunk } from "./media-uploads.js"; + +/* + * Byte routes Media Studio cannot express as JSON RPC: chunk uploads and + * streaming of proxies, thumbnails, renders, and bundles. Every id is checked + * against the actor's space before a path is derived, and every path is + * resolved inside the media root. + */ + +const ID = /^[A-Za-z0-9_-]{1,64}$/; +const FILE_NAME = /^[A-Za-z0-9][A-Za-z0-9._-]{0,199}$/; +const CHUNK_INDEX = /^[0-9]{1,7}$/; + +const CONTENT_TYPES: Record = { + ".mp4": "video/mp4", + ".m4v": "video/mp4", + ".mov": "video/quicktime", + ".webm": "video/webm", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".webp": "image/webp", + ".gif": "image/gif", + ".srt": "application/x-subrip", + ".vtt": "text/vtt; charset=utf-8", + ".json": "application/json", + ".md": "text/markdown; charset=utf-8", + ".txt": "text/plain; charset=utf-8", + ".zip": "application/zip", + ".m4a": "audio/mp4", + ".wav": "audio/wav", +}; + +export function mediaContentType(fileName: string): string { + return CONTENT_TYPES[path.extname(fileName).toLowerCase()] ?? "application/octet-stream"; +} + +export type ByteRange = { start: number; end: number }; + +/** + * Parse a single `bytes=` range. Returns null for no (or an ignorable + * multi-part) range and "unsatisfiable" when it cannot be served. + */ +export function parseByteRange( + header: string | undefined, + size: number, +): ByteRange | null | "unsatisfiable" { + if (!header) return null; + const match = /^bytes=(\d*)-(\d*)$/.exec(header.trim()); + if (!match) return header.includes(",") ? null : "unsatisfiable"; + const [, first = "", last = ""] = match; + if (first === "" && last === "") return "unsatisfiable"; + if (first === "") { + const suffix = Number(last); + if (suffix === 0 || size === 0) return "unsatisfiable"; + return { start: Math.max(0, size - suffix), end: size - 1 }; + } + const start = Number(first); + const end = last === "" ? size - 1 : Math.min(Number(last), size - 1); + if (start >= size || end < start) return "unsatisfiable"; + return { start, end }; +} + +async function sendFile( + c: Context, + absolutePath: string, + options: { fileName: string; download?: boolean }, +): Promise { + const info = await stat(absolutePath).catch(() => null); + if (!info?.isFile()) return c.json({ error: "Not found" }, 404); + const size = info.size; + const headers: Record = { + "content-type": mediaContentType(options.fileName), + "accept-ranges": "bytes", + "cache-control": "private, max-age=300", + }; + if (options.download) { + headers["content-disposition"] = `attachment; filename="${options.fileName}"`; + } + const range = parseByteRange(c.req.header("range"), size); + if (range === "unsatisfiable") { + return new Response(null, { + status: 416, + headers: { ...headers, "content-range": `bytes */${size}` }, + }); + } + if (c.req.method === "HEAD") { + return new Response(null, { + status: 200, + headers: { ...headers, "content-length": String(size) }, + }); + } + const start = range?.start ?? 0; + const end = range?.end ?? size - 1; + const body = + size === 0 + ? null + : (Readable.toWeb( + createReadStream(absolutePath, { start, end }), + ) as ReadableStream); + return new Response(body, { + status: range ? 206 : 200, + headers: { + ...headers, + "content-length": String(size === 0 ? 0 : end - start + 1), + ...(range ? { "content-range": `bytes ${start}-${end}/${size}` } : {}), + }, + }); +} + +function mediaHttpError(c: Context, error: unknown): Response { + if (error instanceof IsolationError || error instanceof PathOutsideRootError) { + return c.json({ error: "Not found" }, 404); + } + if (error instanceof UploadChunkError) return c.json({ error: error.message }, error.status); + if (error instanceof MediaStateConflictError) return c.json({ error: error.message }, 409); + if ((error as NodeJS.ErrnoException | null)?.code === "ENOENT") { + return c.json({ error: "Not found" }, 404); + } + throw error; +} + +export function mountMediaHttpRoutes( + app: Hono, + deps: { + /** Null when Media Studio is not configured: every route answers 404. */ + services: MediaServices | null; + chunkTimeoutMs?: number; + }, + resolveActor: (c: Context) => Promise, +) { + async function authorize( + c: Context, + ): Promise< + { actor: Actor; services: MediaServices; response?: undefined } | { response: Response } + > { + const actor = await resolveActor(c); + if (!actor) return { response: c.json({ error: "Unauthorized" }, 401) }; + if (!deps.services) + return { response: c.json({ error: "Media Studio is not configured" }, 404) }; + return { actor, services: deps.services }; + } + + async function mediaFile(services: MediaServices, relative: string | null) { + if (!relative) return null; + await mkdir(services.deps.mediaRoot, { recursive: true }); + return resolveInside(services.deps.mediaRoot, relative); + } + + app.put("/api/media/uploads/:uploadId/chunks/:index", async (c) => { + const auth = await authorize(c); + if (auth.response) { + cancelBody(c.req.raw.body); + return auth.response; + } + const uploadId = c.req.param("uploadId"); + const index = c.req.param("index"); + if (!ID.test(uploadId) || !CHUNK_INDEX.test(index)) { + cancelBody(c.req.raw.body); + return c.json({ error: "Not found" }, 404); + } + const lengthHeader = c.req.header("content-length"); + const declared = lengthHeader && /^[0-9]+$/.test(lengthHeader) ? Number(lengthHeader) : null; + const timeout = AbortSignal.timeout(deps.chunkTimeoutMs ?? MEDIA_CHUNK_TIMEOUT_MS); + const body = c.req.raw.body; + try { + const upload = await writeUploadChunk({ + services: auth.services, + spaceId: auth.actor.spaceId, + uploadId, + index: Number(index), + body, + declaredLength: declared, + signal: AbortSignal.any([timeout, c.req.raw.signal]), + }); + return c.json(upload); + } catch (error) { + cancelBody(body); + if (timeout.aborted) return c.json({ error: "Chunk took too long" }, 408); + return mediaHttpError(c, error); + } + }); + + app.get("/api/media/renders/:renderId/files/:fileName", async (c) => { + const auth = await authorize(c); + if (auth.response) return auth.response; + const renderId = c.req.param("renderId"); + const fileName = c.req.param("fileName"); + if (!ID.test(renderId) || !FILE_NAME.test(fileName) || fileName.includes("..")) { + return c.json({ error: "Not found" }, 404); + } + try { + const render = await auth.services.deps.repos.getRenderRecord(auth.actor.spaceId, renderId); + const download = c.req.query("download") === "1"; + // Only a render that passed its gate leaves the app as a download. + if (download && render.status !== "done") return c.json({ error: "Not found" }, 404); + const absolute = await auth.services.render.outputPath(auth.actor, renderId, fileName); + if (!absolute) return c.json({ error: "Not found" }, 404); + await mkdir(auth.services.deps.mediaRoot, { recursive: true }); + const inside = await assertInside( + await resolveInside( + auth.services.deps.mediaRoot, + `${auth.actor.spaceId}/renders/${renderId}`, + ), + absolute, + ); + return await sendFile(c, inside, { fileName, download }); + } catch (error) { + return mediaHttpError(c, error); + } + }); + + for (const kind of ["thumb", "proxy"] as const) { + app.get(`/api/media/assets/:assetId/${kind}`, async (c) => { + const auth = await authorize(c); + if (auth.response) return auth.response; + const assetId = c.req.param("assetId"); + if (!ID.test(assetId)) return c.json({ error: "Not found" }, 404); + try { + const asset = await auth.services.deps.repos.getAssetRecord(auth.actor.spaceId, assetId); + const relative = kind === "thumb" ? asset.thumbPath : asset.proxyPath; + const absolute = await mediaFile(auth.services, relative); + if (!absolute) return c.json({ error: "Not found" }, 404); + return await sendFile(c, absolute, { fileName: path.basename(absolute) }); + } catch (error) { + return mediaHttpError(c, error); + } + }); + } + + app.get("/api/media/bundles/:publicationId", async (c) => { + const auth = await authorize(c); + if (auth.response) return auth.response; + const publicationId = c.req.param("publicationId"); + if (!ID.test(publicationId)) return c.json({ error: "Not found" }, 404); + try { + const publication = await auth.services.deps.repos.getPublicationRecord( + auth.actor.spaceId, + publicationId, + ); + if (publication.bundleArtifactId) return c.json({ error: "Not found" }, 404); + const absolute = await mediaFile( + auth.services, + `${auth.actor.spaceId}/bundles/${publication.id}/bundle.zip`, + ); + if (!absolute) return c.json({ error: "Not found" }, 404); + return await sendFile(c, absolute, { + fileName: `bundle-${publication.id}.zip`, + download: true, + }); + } catch (error) { + return mediaHttpError(c, error); + } + }); +} diff --git a/apps/api/src/media-test-fakes.ts b/apps/api/src/media-test-fakes.ts new file mode 100644 index 0000000..69c8872 --- /dev/null +++ b/apps/api/src/media-test-fakes.ts @@ -0,0 +1,387 @@ +import type { MediaInferenceProvider } from "@rakazo/adapter-kit"; +import type { + MediaActor, + MediaIngestService, + MediaMarketingService, + MediaPrivacyService, + MediaRenderService, + MediaServiceDeps, + MediaServices, +} from "@rakazo/adapters"; +import type { PlatformProfile } from "@rakazo/contracts"; +import { mediaRpcShapes } from "@rakazo/contracts"; +import type { MediaRepos, PrismaClient } from "@rakazo/db"; + +/* + * In-memory stand-ins for the media service slices, for API tests only. They + * record what the router hands them and do the minimum through the real + * repositories; nothing here renders, detects, or publishes. + */ + +export interface MediaServiceCall { + service: "ingest" | "privacy" | "render" | "marketing"; + method: string; + actor: MediaActor; + input: unknown; +} + +export interface FakeMediaServicesOptions { + repos: MediaRepos; + mediaRoot: string; + prisma?: PrismaClient; + inference?: MediaInferenceProvider | null; + profiles?: PlatformProfile[]; + now?: () => Date; + /** Render id → absolute output path the fake render service reports. */ + renderOutputs?: Map; + onUploadCompleted?: ( + actor: MediaActor, + input: { assetId: string; tempPath: string; sha256: string; fileName: string }, + ) => Promise; +} + +export function fakeMediaServices(options: FakeMediaServicesOptions): { + services: MediaServices; + calls: MediaServiceCall[]; +} { + const calls: MediaServiceCall[] = []; + const { repos } = options; + const record = ( + service: MediaServiceCall["service"], + method: string, + actor: MediaActor, + input: unknown, + ) => { + calls.push({ service, method, actor: { spaceId: actor.spaceId, userId: actor.userId }, input }); + }; + + const ingest: MediaIngestService = { + async pushItems(actor, input) { + record("ingest", "pushItems", actor, input); + return { created: 0, updated: 0, unchanged: 0, deleted: 0 }; + }, + async requestSync(actor, sourceId) { + record("ingest", "requestSync", actor, sourceId); + }, + async uploadCompleted(actor, input) { + record("ingest", "uploadCompleted", actor, input); + await options.onUploadCompleted?.(actor, input); + }, + async recluster(actor) { + record("ingest", "recluster", actor, null); + }, + async removeSource(actor, sourceId) { + record("ingest", "removeSource", actor, sourceId); + await repos.removeSource(actor.spaceId, sourceId); + }, + }; + + const privacy: MediaPrivacyService = { + async prepareSources() { + throw new Error("prepareSources is not part of the API fakes"); + }, + async gate() { + throw new Error("gate is not part of the API fakes"); + }, + async createMark(actor, input) { + record("privacy", "createMark", actor, input); + return repos.createMark(actor.spaceId, input); + }, + async updateMark(actor, input) { + record("privacy", "updateMark", actor, input); + const { markId, ...patch } = input; + return repos.updateMark(actor.spaceId, markId, patch); + }, + async removePerson(actor, personId) { + record("privacy", "removePerson", actor, personId); + await repos.removePersonCompletely(actor.spaceId, personId); + }, + }; + + const render: MediaRenderService = { + async start(actor, input) { + record("render", "start", actor, input); + return repos.createRender(actor.spaceId, { ...input, createdBy: actor.userId }); + }, + async cancel(actor, renderId) { + record("render", "cancel", actor, renderId); + return repos.cancelRender(actor.spaceId, renderId); + }, + async outputPath(actor, renderId, fileName) { + record("render", "outputPath", actor, { renderId, fileName }); + const file = options.renderOutputs?.get(renderId); + return file?.endsWith(`/${fileName}`) ? file : null; + }, + }; + + const marketing: MediaMarketingService = { + async autoPlan(actor, input) { + record("marketing", "autoPlan", actor, input); + return []; + }, + async submitPlan(actor, input) { + record("marketing", "submitPlan", actor, input); + return { ok: false, errors: [] }; + }, + async introVariants(actor, input) { + record("marketing", "introVariants", actor, input); + return []; + }, + async thumbnails(actor, input) { + record("marketing", "thumbnails", actor, input); + return []; + }, + async copy(actor, input) { + record("marketing", "copy", actor, input); + const render = await repos.getRender(actor.spaceId, input.renderId); + return { + profileId: render.profileId, + language: input.language, + title: "Title", + hook: "Hook", + caption: "Caption", + hashtags: [], + issues: [], + }; + }, + async createPublication(actor, input) { + record("marketing", "createPublication", actor, input); + return repos.createPublication(actor.spaceId, actor.userId, { + ...input, + trackingUrl: null, + scheduledAt: input.scheduledAt ? new Date(input.scheduledAt) : null, + }); + }, + async bundle(actor, publicationId) { + record("marketing", "bundle", actor, publicationId); + return repos.markPublicationBundled(actor.spaceId, publicationId, null); + }, + async cancelPublication(actor, publicationId) { + record("marketing", "cancelPublication", actor, publicationId); + return repos.cancelPublication(actor.spaceId, publicationId); + }, + async importMetricsCsv(actor, input) { + record("marketing", "importMetricsCsv", actor, input); + return { imported: 0, skipped: [] }; + }, + }; + + const deps = { + prisma: options.prisma, + repos, + jobs: { publish: async () => undefined }, + inference: options.inference ?? null, + renderer: { + id: "fake", + hardwareEncoders: async () => ["h264_fake"], + render: async () => { + throw new Error("render is not part of the API fakes"); + }, + }, + motion: { id: "fake", templates: () => [], render: async () => ({ path: "", hasAlpha: true }) }, + publisher: { id: "fake", publish: async () => ({ status: "bundled", externalUrl: null }) }, + shortener: null, + sources: {}, + mediaRoot: options.mediaRoot, + profiles: async () => ({ + profiles: new Map( + (options.profiles ?? []).map((profile) => [ + profile.id, + { profile, origin: "builtin" as const }, + ]), + ), + errors: [], + }), + artifacts: {}, + model: { generate: async () => null }, + publish: async () => undefined, + now: options.now ?? (() => new Date()), + } as unknown as MediaServiceDeps; + + return { services: { deps, ingest, privacy, render, marketing }, calls }; +} + +/** Every RPC path of the media contract, e.g. "sources/list". */ +export function mediaProcedurePaths(): string[] { + const paths: string[] = []; + const walk = (node: Record, prefix: string[]) => { + for (const [key, value] of Object.entries(node)) { + const child = value as Record; + if ("input" in child && "output" in child) paths.push([...prefix, key].join("/")); + else walk(child, [...prefix, key]); + } + }; + walk(mediaRpcShapes, []); + return paths; +} + +export interface MediaSampleIds { + sourceId: string; + assetId: string; + secondAssetId: string; + segmentId: string; + trackId: string; + personId: string; + markId: string; + collectionId: string; + brandKitId: string; + projectId: string; + planId: string; + renderId: string; + publicationId: string; + mappingId: string; + uploadId: string; +} + +/** + * How a procedure must treat another space's ids: refuse them as not found, + * answer only from the caller's own space, or be covered by a dedicated test. + */ +export type ForeignExpectation = "not_found" | "scoped" | "dedicated"; + +/** A valid input for every media procedure, built from one space's ids. */ +export function sampleMediaInputs( + ids: MediaSampleIds, + kit: unknown, +): Record { + const at = "2026-09-01T10:00:00.000Z"; + return { + capabilities: { input: {}, foreign: "scoped" }, + "sources/list": { input: {}, foreign: "scoped" }, + "sources/createFolder": { input: { label: "Footage", path: "/" }, foreign: "dedicated" }, + "sources/registerClient": { + input: { kind: "ios_library", label: "Phone", deviceId: "device-sample" }, + foreign: "scoped", + }, + "sources/pushItems": { + input: { sourceId: ids.sourceId, items: [], syncToken: null, complete: true }, + foreign: "not_found", + }, + "sources/sync": { input: { sourceId: ids.sourceId }, foreign: "not_found" }, + "sources/remove": { input: { sourceId: ids.sourceId }, foreign: "not_found" }, + "assets/search": { input: {}, foreign: "scoped" }, + "assets/get": { input: { assetId: ids.assetId }, foreign: "not_found" }, + "assets/setPaidOnly": { + input: { segmentIds: [ids.segmentId], paidOnly: true }, + foreign: "not_found", + }, + "assets/requestUpload": { input: { assetIds: [ids.assetId] }, foreign: "not_found" }, + "uploads/begin": { + input: { + assetId: ids.secondAssetId, + sizeBytes: 10, + sha256: "a".repeat(64), + fileName: "clip.mp4", + }, + foreign: "not_found", + }, + "uploads/status": { input: { uploadId: ids.uploadId }, foreign: "not_found" }, + "uploads/complete": { input: { uploadId: ids.uploadId }, foreign: "not_found" }, + "collections/list": { input: {}, foreign: "scoped" }, + "collections/get": { input: { collectionId: ids.collectionId }, foreign: "not_found" }, + "collections/rename": { + input: { collectionId: ids.collectionId, title: "Renamed" }, + foreign: "not_found", + }, + "collections/merge": { + input: { collectionIds: [ids.collectionId, `${ids.collectionId}x`] }, + foreign: "not_found", + }, + "collections/split": { + input: { collectionId: ids.collectionId, atAssetId: ids.assetId }, + foreign: "not_found", + }, + "collections/createManual": { + input: { title: "Picks", assetIds: [ids.assetId] }, + foreign: "not_found", + }, + "collections/recluster": { input: {}, foreign: "scoped" }, + "people/list": { input: {}, foreign: "scoped" }, + "people/update": { input: { personId: ids.personId, displayName: "X" }, foreign: "not_found" }, + "people/merge": { + input: { personIds: [ids.personId, `${ids.personId}x`] }, + foreign: "not_found", + }, + "people/assignTrack": { input: { trackId: ids.trackId, personId: null }, foreign: "not_found" }, + "people/recordConsent": { + input: { + personId: ids.personId, + consentArtifactId: null, + consentRecordedAt: at, + ageVerificationMethod: "platform_verified", + ageVerifiedAt: at, + }, + foreign: "not_found", + }, + "people/remove": { input: { personId: ids.personId }, foreign: "not_found" }, + "marks/list": { input: { assetId: ids.assetId }, foreign: "scoped" }, + "marks/create": { + input: { kind: "tattoo", assetId: ids.assetId, prompt: { tMs: 0 }, treatment: "blur" }, + foreign: "not_found", + }, + "marks/update": { input: { markId: ids.markId, label: "x" }, foreign: "not_found" }, + "marks/remove": { input: { markId: ids.markId }, foreign: "not_found" }, + "brandKits/list": { input: {}, foreign: "scoped" }, + "brandKits/upsert": { input: { id: ids.brandKitId, kit }, foreign: "not_found" }, + "brandKits/remove": { input: { id: ids.brandKitId }, foreign: "not_found" }, + "brandKits/templates": { input: {}, foreign: "scoped" }, + "brandKits/introVariants": { input: { id: ids.brandKitId }, foreign: "not_found" }, + "profiles/list": { input: {}, foreign: "scoped" }, + "profiles/get": { input: { profileId: "example-short-vertical" }, foreign: "scoped" }, + "projects/list": { input: {}, foreign: "scoped" }, + "projects/get": { input: { projectId: ids.projectId }, foreign: "not_found" }, + "projects/create": { + input: { + title: "Teaser", + goal: "teaser", + collectionIds: [ids.collectionId], + profileIds: ["example-short-vertical"], + }, + foreign: "not_found", + }, + "projects/update": { input: { projectId: ids.projectId, title: "X" }, foreign: "not_found" }, + "projects/remove": { input: { projectId: ids.projectId }, foreign: "not_found" }, + "projects/autoPlan": { input: { projectId: ids.projectId }, foreign: "not_found" }, + "plans/submit": { input: { projectId: ids.projectId, edl: {} }, foreign: "not_found" }, + "renders/start": { input: { planId: ids.planId, quality: "draft" }, foreign: "not_found" }, + "renders/get": { input: { renderId: ids.renderId }, foreign: "not_found" }, + "renders/cancel": { input: { renderId: ids.renderId }, foreign: "not_found" }, + "renders/thumbnails": { input: { renderId: ids.renderId }, foreign: "not_found" }, + "marketing/copy": { input: { renderId: ids.renderId }, foreign: "not_found" }, + "publications/list": { input: { projectId: ids.projectId }, foreign: "scoped" }, + "publications/create": { input: { renderId: ids.renderId }, foreign: "not_found" }, + "publications/bundle": { input: { publicationId: ids.publicationId }, foreign: "not_found" }, + "publications/markPublished": { + input: { publicationId: ids.publicationId, externalUrl: null }, + foreign: "not_found", + }, + "publications/cancel": { input: { publicationId: ids.publicationId }, foreign: "not_found" }, + "metrics/mappings": { input: {}, foreign: "scoped" }, + "metrics/saveMapping": { + input: { + id: ids.mappingId, + name: "Imported", + columns: { publication: "id", observedAt: "date" }, + publicationKey: "id", + }, + foreign: "not_found", + }, + "metrics/importCsv": { + input: { mappingId: ids.mappingId, csv: "id,date" }, + foreign: "not_found", + }, + "metrics/record": { + input: { + publicationId: ids.publicationId, + observedAt: at, + views: 1, + clicks: 0, + conversions: 0, + revenueCents: 0, + currency: "EUR", + }, + foreign: "not_found", + }, + "metrics/funnel": { input: { projectId: ids.projectId }, foreign: "not_found" }, + }; +} diff --git a/apps/api/src/media-uploads.ts b/apps/api/src/media-uploads.ts new file mode 100644 index 0000000..084404c --- /dev/null +++ b/apps/api/src/media-uploads.ts @@ -0,0 +1,244 @@ +import { createHash, randomUUID } from "node:crypto"; +import { createReadStream } from "node:fs"; +import { mkdir, open, rename, rm, stat } from "node:fs/promises"; +import path from "node:path"; +import type { MediaActor, MediaServices } from "@rakazo/adapters"; +import type { MediaUpload } from "@rakazo/contracts"; +import type { MediaRepos } from "@rakazo/db"; +import { MediaStateConflictError } from "@rakazo/db"; +import { resolveInside } from "@rakazo/media/node"; + +/* + * Resumable chunked uploads. Each chunk lands in its own file + * (/uploads//chunks/.part, written to a temp name and + * renamed, so a repeated or concurrent PUT of the same index is harmless). + * Completion concatenates the chunks in order into /uploads//file + * while hashing, verifies size and SHA-256, and hands the file to ingest. + * + * Limits: the contract caps the total size, every chunk must have its exact + * length, one chunk must arrive within MEDIA_CHUNK_TIMEOUT_MS, and an upload + * that saw no progress for MEDIA_UPLOAD_IDLE_MS expires and must start over. + */ + +/** A single chunk must arrive within this window; slow links resume per chunk. */ +export const MEDIA_CHUNK_TIMEOUT_MS = 10 * 60_000; +/** An open upload without a new chunk for this long is failed and its bytes removed. */ +export const MEDIA_UPLOAD_IDLE_MS = 24 * 60 * 60_000; + +export class UploadChunkError extends Error { + constructor( + readonly status: 400 | 409 | 413, + message: string, + ) { + super(message); + this.name = "UploadChunkError"; + } +} + +type UploadRow = Awaited>; +type UploadSize = Pick; + +export function uploadChunkCount(row: UploadSize): number { + return Number((row.sizeBytes + BigInt(row.chunkBytes) - 1n) / BigInt(row.chunkBytes)); +} + +/** Exact byte length chunk `index` must have: every chunk is full except the last. */ +export function expectedChunkBytes(row: UploadSize, index: number): number { + const count = uploadChunkCount(row); + if (index < count - 1) return row.chunkBytes; + return Number(row.sizeBytes - BigInt(row.chunkBytes) * BigInt(count - 1)); +} + +function uploadDir(row: Pick): string { + return `${row.spaceId}/uploads/${row.id}`; +} + +async function mediaPath(mediaRoot: string, relative: string): Promise { + await mkdir(mediaRoot, { recursive: true }); + return resolveInside(mediaRoot, relative); +} + +async function chunksDir(mediaRoot: string, row: UploadRow): Promise { + const dir = await mediaPath(mediaRoot, `${uploadDir(row)}/chunks`); + await mkdir(dir, { recursive: true }); + return dir; +} + +async function removeUploadFiles(mediaRoot: string, row: UploadRow): Promise { + await rm(await mediaPath(mediaRoot, uploadDir(row)), { recursive: true, force: true }); +} + +function isIdle(services: MediaServices, row: UploadRow): boolean { + return services.deps.now().getTime() - row.updatedAt.getTime() > MEDIA_UPLOAD_IDLE_MS; +} + +/** Fail an idle open upload and drop its bytes. Returns true when it expired. */ +async function expireIfIdle(services: MediaServices, row: UploadRow): Promise { + if (row.status !== "open" || !isIdle(services, row)) return false; + await services.deps.repos.failUpload(row.spaceId, row.id).catch((error: unknown) => { + // A concurrent request already closed it; the files still go. + if (!(error instanceof MediaStateConflictError)) throw error; + }); + await removeUploadFiles(services.deps.mediaRoot, row); + return true; +} + +/** + * Open an upload, or resume the open one for the same asset and content. An + * idle one expires first so the client starts over with a fresh upload. + */ +export async function beginMediaUpload( + services: MediaServices, + actor: MediaActor, + input: { assetId: string; sizeBytes: number; sha256: string; fileName: string }, +): Promise { + const { repos } = services.deps; + const asset = await repos.getAssetRecord(actor.spaceId, input.assetId); + // Device items wait for their original; files read from a source never upload. + if (asset.storage !== "remote" && asset.ingestState !== "awaiting_upload") { + throw new MediaStateConflictError("This item does not need an upload"); + } + const begin = () => + repos.beginUpload(actor.spaceId, { ...input, assetId: asset.id, userId: actor.userId }); + const upload = await begin(); + const row = await repos.getUploadRecord(actor.spaceId, upload.id); + return (await expireIfIdle(services, row)) ? begin() : upload; +} + +/** + * Stream one chunk body to disk without buffering more than one network read. + * Rejects bodies longer or shorter than the chunk's exact size. + */ +export async function writeUploadChunk(input: { + services: MediaServices; + spaceId: string; + uploadId: string; + index: number; + body: ReadableStream | null; + declaredLength: number | null; + /** Aborting stops reading the body; the chunk is then discarded as incomplete. */ + signal?: AbortSignal; +}): Promise { + const { services } = input; + const { repos, mediaRoot } = services.deps; + const row = await repos.getUploadRecord(input.spaceId, input.uploadId); + if (await expireIfIdle(services, row)) throw new UploadChunkError(409, "Upload expired"); + if (row.status !== "open") throw new UploadChunkError(409, "Upload is not open"); + const count = uploadChunkCount(row); + if (!Number.isInteger(input.index) || input.index < 0 || input.index >= count) { + throw new UploadChunkError(400, "Chunk index out of range"); + } + const expected = expectedChunkBytes(row, input.index); + if (input.declaredLength !== null && input.declaredLength > expected) { + throw new UploadChunkError(413, "Chunk is too large"); + } + if (!input.body) throw new UploadChunkError(400, "Chunk is empty"); + const dir = await chunksDir(mediaRoot, row); + const finalPath = path.join(dir, `${input.index}.part`); + const tempPath = path.join(dir, `${input.index}.${randomUUID()}.tmp`); + const handle = await open(tempPath, "wx"); + let written = 0; + let ok = false; + const reader = input.body.getReader(); + const stop = () => void reader.cancel().catch(() => undefined); + input.signal?.addEventListener("abort", stop, { once: true }); + try { + if (input.signal?.aborted) stop(); + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + written += value.byteLength; + if (written > expected) { + void reader.cancel().catch(() => undefined); + throw new UploadChunkError(413, "Chunk is too large"); + } + await handle.write(value); + } + if (written !== expected) throw new UploadChunkError(400, "Chunk is incomplete"); + ok = true; + } finally { + input.signal?.removeEventListener("abort", stop); + reader.releaseLock(); + await handle.close(); + if (!ok) await rm(tempPath, { force: true }); + } + await rename(tempPath, finalPath); + return repos.markChunkReceived(input.spaceId, input.uploadId, input.index); +} + +const completing = new Set(); + +/** + * Assemble a fully received upload, verify it, hand it to ingest, and close it. + * A checksum mismatch fails the upload so the client starts a fresh one. + */ +export async function completeMediaUpload( + services: MediaServices, + actor: MediaActor, + uploadId: string, +): Promise { + const { repos, mediaRoot } = services.deps; + const row = await repos.getUploadRecord(actor.spaceId, uploadId); + if (row.status === "complete") return repos.uploadStatus(actor.spaceId, uploadId); + if (await expireIfIdle(services, row)) throw new UploadChunkError(409, "Upload expired"); + if (row.status !== "open") throw new MediaStateConflictError("Upload is not open"); + const received = new Set(row.receivedChunks); + const count = uploadChunkCount(row); + for (let index = 0; index < count; index++) { + if (!received.has(index)) throw new MediaStateConflictError("Upload is missing chunks"); + } + if (completing.has(row.id)) throw new MediaStateConflictError("Upload is already completing"); + completing.add(row.id); + try { + const dir = await chunksDir(mediaRoot, row); + const relativeFile = `${uploadDir(row)}/file`; + const filePath = await mediaPath(mediaRoot, relativeFile); + const tempFile = await mediaPath(mediaRoot, `${uploadDir(row)}/file.${randomUUID()}.tmp`); + const hash = createHash("sha256"); + let total = 0; + const out = await open(tempFile, "wx"); + try { + for (let index = 0; index < count; index++) { + const chunkPath = path.join(dir, `${index}.part`); + const info = await stat(chunkPath).catch(() => null); + if (!info || info.size !== expectedChunkBytes(row, index)) { + throw new MediaStateConflictError("Upload is missing chunks"); + } + for await (const buffer of createReadStream(chunkPath) as AsyncIterable) { + hash.update(buffer); + total += buffer.byteLength; + await out.write(buffer); + } + } + } catch (error) { + await out.close(); + await rm(tempFile, { force: true }); + throw error; + } + await out.close(); + const sha256 = hash.digest("hex"); + if (sha256 !== row.sha256 || BigInt(total) !== row.sizeBytes) { + await removeUploadFiles(mediaRoot, row); + await repos.failUpload(actor.spaceId, uploadId); + throw new UploadChunkError(400, "Upload does not match its checksum"); + } + await rename(tempFile, filePath); + await rm(dir, { recursive: true, force: true }); + await services.ingest.uploadCompleted(actor, { + assetId: row.assetId, + tempPath: filePath, + sha256, + fileName: row.fileName, + }); + // Ingest normally moves the file into the asset folder and closes the + // upload itself; otherwise the assembled file stays as the original. + const after = await repos.getUploadRecord(actor.spaceId, uploadId); + if (after.status === "complete") { + await removeUploadFiles(mediaRoot, row); + return repos.uploadStatus(actor.spaceId, uploadId); + } + return repos.completeUpload(actor.spaceId, uploadId, { storagePath: relativeFile }); + } finally { + completing.delete(row.id); + } +} diff --git a/apps/api/src/media.postgres.test.ts b/apps/api/src/media.postgres.test.ts new file mode 100644 index 0000000..6f7aae1 --- /dev/null +++ b/apps/api/src/media.postgres.test.ts @@ -0,0 +1,516 @@ +import { createHash } from "node:crypto"; +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { createRouterClient, ORPCError } from "@orpc/server"; +import type { Actor } from "@rakazo/contracts"; +import type { MediaRepos } from "@rakazo/db"; +import { createMediaRepos } from "@rakazo/db"; +import { Hono } from "hono"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import type { MediaTestDb, SeededMediaSpace } from "../../../packages/db/src/media-fixtures.js"; +import { + openMediaTestDb, + seedMediaSpace, + sourceItem, + testBrandKit, +} from "../../../packages/db/src/media-fixtures.js"; +import { testProfile } from "../../../packages/media/src/test-fixtures.js"; +import { createMediaRouter } from "./media.js"; +import { mountMediaHttpRoutes } from "./media-http.js"; +import type { MediaServiceCall } from "./media-test-fakes.js"; +import { fakeMediaServices, sampleMediaInputs } from "./media-test-fakes.js"; +import { MEDIA_UPLOAD_IDLE_MS } from "./media-uploads.js"; + +const databaseUrl = process.env.DATABASE_URL; +const describePostgres = + process.env.VERIFY_DATABASE && databaseUrl ? describe.sequential : describe.skip; + +type Space = { userId: string; spaceId: string }; + +function actorOf(space: Space, isDeploymentOwner = false): Actor { + return { + userId: space.userId, + spaceId: space.spaceId, + email: `${space.userId}@rakazo.test`, + isDeploymentOwner, + }; +} + +function procedure(client: unknown, rpcPath: string) { + let node = client; + for (const key of rpcPath.split("/")) node = (node as Record)[key]; + return node as (input: unknown) => Promise; +} + +async function rpcCode(promise: Promise): Promise { + try { + await promise; + } catch (error) { + if (error instanceof ORPCError) return error.code; + throw error; + } + return "OK"; +} + +function sha256(bytes: Uint8Array): string { + return createHash("sha256").update(bytes).digest("hex"); +} + +describePostgres("media API authorization and uploads (PostgreSQL)", () => { + let db: MediaTestDb; + let repos: MediaRepos; + let mediaRoot: string; + let a: Space; + let b: Space; + let seededA: SeededMediaSpace; + let seededB: SeededMediaSpace; + let calls: MediaServiceCall[]; + let renderOutputs: Map; + let rpc: (actor: Actor) => unknown; + let app: Hono; + let currentActor: Actor | null; + + function setup(options: { now?: () => Date; chunkTimeoutMs?: number } = {}) { + const fake = fakeMediaServices({ + repos, + mediaRoot, + prisma: db.prisma, + profiles: [testProfile()], + renderOutputs, + ...(options.now ? { now: options.now } : {}), + }); + calls = fake.calls; + const router = createMediaRouter({ + prisma: db.prisma, + services: fake.services, + dataDir: path.join(mediaRoot, ".."), + }); + rpc = (actor) => createRouterClient(router, { context: { actor } }); + app = new Hono(); + mountMediaHttpRoutes( + app, + { + services: fake.services, + ...(options.chunkTimeoutMs ? { chunkTimeoutMs: options.chunkTimeoutMs } : {}), + }, + async () => currentActor, + ); + } + + async function snapshot(spaceId: string) { + const p = db.prisma; + const where = { spaceId }; + const orderBy = { id: "asc" as const }; + return JSON.stringify( + [ + await p.mediaSource.findMany({ where, orderBy }), + await p.mediaAsset.findMany({ where, orderBy }), + await p.mediaUpload.findMany({ where, orderBy }), + await p.mediaSegment.findMany({ where, orderBy }), + await p.mediaAnalysis.findMany({ where, orderBy }), + await p.mediaCollection.findMany({ where, orderBy }), + await p.mediaCollectionItem.findMany({ + where: { collection: where }, + orderBy: [{ collectionId: "asc" }, { assetId: "asc" }], + }), + await p.mediaPerson.findMany({ where, orderBy }), + await p.mediaFaceTrack.findMany({ where, orderBy }), + await p.mediaMark.findMany({ where, orderBy }), + await p.mediaBrandKit.findMany({ where, orderBy }), + await p.mediaProject.findMany({ where, orderBy }), + await p.mediaEditPlan.findMany({ where, orderBy }), + await p.mediaRender.findMany({ where, orderBy }), + await p.mediaPublication.findMany({ where, orderBy }), + await p.mediaMetric.findMany({ where, orderBy }), + await p.mediaMetricMapping.findMany({ where, orderBy }), + ], + (_key, value) => (typeof value === "bigint" ? value.toString() : value), + ); + } + + async function artifactIn(space: Space): Promise { + const artifact = await db.prisma.artifact.create({ + data: { + spaceId: space.spaceId, + userId: space.userId, + name: "consent.pdf", + mimeType: "application/pdf", + size: 1, + hash: "0".repeat(64), + storageKey: `test/${space.spaceId}/consent.pdf`, + }, + }); + return artifact.id; + } + + /** A device item in `space` that waits for its original. */ + async function remoteAsset(space: Space, index: number): Promise { + const source = await procedure( + rpc(actorOf(space)), + "sources/registerClient", + )({ kind: "ios_library", label: "Phone", deviceId: `device-${index}` }); + const synced = await repos.upsertSourceItems(space.spaceId, (source as { id: string }).id, [ + sourceItem(index), + ]); + const assetId = synced.changedAssetIds[0] as string; + await repos.requestUpload(space.spaceId, [assetId]); + return assetId; + } + + function putChunk(uploadId: string, index: number, body: BodyInit, actor: Actor | null) { + currentActor = actor; + return app.request(`/api/media/uploads/${uploadId}/chunks/${index}`, { method: "PUT", body }); + } + + beforeAll(async () => { + db = openMediaTestDb(databaseUrl!); + repos = createMediaRepos(db.prisma); + const root = await mkdtemp(path.join(tmpdir(), "media-api-pg-")); + mediaRoot = path.join(root, "media"); + await mkdir(mediaRoot); + renderOutputs = new Map(); + a = await db.newUser(); + b = await db.newUser(); + seededA = await seedMediaSpace(repos, a.spaceId, a.userId); + seededB = await seedMediaSpace(repos, b.spaceId, b.userId); + setup(); + }); + + afterAll(async () => { + await db?.cleanup(); + if (mediaRoot) await rm(path.dirname(mediaRoot), { recursive: true, force: true }); + }); + + it("refuses another space's ids in every media procedure and leaves that space untouched", async () => { + const before = await snapshot(a.spaceId); + const foreignIds = Object.values(seededA); + const intruder = rpc(actorOf(b)); + const inputs = sampleMediaInputs(seededA, testBrandKit); + for (const [rpcPath, sample] of Object.entries(inputs)) { + if (sample.foreign === "dedicated") continue; + const call = procedure(intruder, rpcPath)(sample.input); + if (sample.foreign === "not_found") { + expect(await rpcCode(call), rpcPath).toBe("NOT_FOUND"); + continue; + } + const result = JSON.stringify(await call); + for (const id of foreignIds) expect(result, rpcPath).not.toContain(id); + } + expect(await snapshot(a.spaceId)).toBe(before); + // No service ever received the other space's ids. + const leaked = calls.filter((call) => + foreignIds.some((id) => JSON.stringify(call.input ?? null).includes(id)), + ); + expect(leaked).toEqual([]); + expect(calls.every((call) => call.actor.spaceId === b.spaceId)).toBe(true); + }); + + it("serves the same inputs inside the caller's own space", async () => { + // Positive control: NOT_FOUND above comes from isolation, not from bad samples. + const fabricated = new Set(["collections/merge", "people/merge"]); + for (const [rpcPath, sample] of Object.entries(sampleMediaInputs(seededA, testBrandKit))) { + if (sample.foreign !== "not_found" || fabricated.has(rpcPath)) continue; + const space = await db.newUser(); + const seeded = await seedMediaSpace(repos, space.spaceId, space.userId); + const input = sampleMediaInputs(seeded, testBrandKit)[rpcPath]?.input; + const code = await rpcCode(procedure(rpc(actorOf(space)), rpcPath)(input)); + expect(code, rpcPath).not.toBe("NOT_FOUND"); + } + }, 120_000); + + it("refuses another space's records referenced from the caller's own ones", async () => { + const own = rpc(actorOf(b)); + const foreignArtifact = await artifactIn(a); + const attempts: Array<[string, unknown]> = [ + ["projects/update", { projectId: seededB.projectId, collectionIds: [seededA.collectionId] }], + ["projects/update", { projectId: seededB.projectId, brandKitId: seededA.brandKitId }], + [ + "projects/create", + { + title: "Mixed", + goal: "teaser", + assetIds: [seededA.assetId], + profileIds: ["example-short-vertical"], + }, + ], + ["people/assignTrack", { trackId: seededB.trackId, personId: seededA.personId }], + [ + "marks/create", + { + kind: "tattoo", + assetId: seededB.assetId, + personId: seededA.personId, + prompt: { tMs: 0 }, + treatment: "blur", + }, + ], + ["marks/update", { markId: seededB.markId, personId: seededA.personId }], + [ + "assets/setPaidOnly", + { segmentIds: [seededB.segmentId, seededA.segmentId], paidOnly: true }, + ], + ["collections/merge", { collectionIds: [seededB.collectionId, seededA.collectionId] }], + ["people/merge", { personIds: [seededB.personId, seededA.personId] }], + ["brandKits/upsert", { kit: { ...testBrandKit, logoArtifactIds: [foreignArtifact] } }], + [ + "people/recordConsent", + { + personId: seededB.personId, + consentArtifactId: foreignArtifact, + consentRecordedAt: "2026-09-01T10:00:00.000Z", + ageVerificationMethod: "document_on_file", + ageVerifiedAt: "2026-09-01T10:00:00.000Z", + }, + ], + ]; + const before = await snapshot(a.spaceId); + for (const [rpcPath, input] of attempts) { + expect(await rpcCode(procedure(own, rpcPath)(input)), rpcPath).toBe("NOT_FOUND"); + } + expect(await snapshot(a.spaceId)).toBe(before); + const person = await db.prisma.mediaPerson.findUniqueOrThrow({ + where: { id: seededB.personId }, + }); + expect(person.consentRecordedAt).toBeNull(); + }); + + it("records consent only as the signed-in member and ignores it on updates", async () => { + const space = await db.newUser(); + const seeded = await seedMediaSpace(repos, space.spaceId, space.userId); + const client = rpc(actorOf(space)); + await procedure( + client, + "people/update", + )({ + personId: seeded.personId, + displayName: "Person 1", + consentRecordedAt: "2026-09-01T10:00:00.000Z", + ageVerificationMethod: "platform_verified", + ageVerifiedAt: "2026-09-01T10:00:00.000Z", + }); + let person = await db.prisma.mediaPerson.findUniqueOrThrow({ where: { id: seeded.personId } }); + expect(person.consentRecordedAt).toBeNull(); + expect(person.ageVerifiedAt).toBeNull(); + + const artifactId = await artifactIn(space); + await procedure( + client, + "people/recordConsent", + )({ + personId: seeded.personId, + consentArtifactId: artifactId, + consentRecordedAt: "2026-09-01T10:00:00.000Z", + ageVerificationMethod: "platform_verified", + ageVerifiedAt: "2026-09-01T10:00:00.000Z", + }); + person = await db.prisma.mediaPerson.findUniqueOrThrow({ where: { id: seeded.personId } }); + expect(person.consentRecordedBy).toBe(space.userId); + expect(person.consentArtifactId).toBe(artifactId); + expect(person.ageVerificationMethod).toBe("platform_verified"); + }); + + it("assembles a resumable chunked upload and hands it to ingest", async () => { + const assetId = await remoteAsset(b, 101); + const bytes = new TextEncoder().encode("0123456789"); + const upload = await repos.beginUpload(b.spaceId, { + assetId, + userId: b.userId, + sizeBytes: bytes.length, + sha256: sha256(bytes), + fileName: "clip.mp4", + chunkBytes: 4, + }); + const owner = actorOf(b); + + expect((await putChunk(upload.id, 0, bytes.slice(0, 4), actorOf(a))).status).toBe(404); + expect((await putChunk(upload.id, 0, bytes.slice(0, 4), null)).status).toBe(401); + expect((await putChunk(upload.id, 3, bytes.slice(0, 2), owner)).status).toBe(400); + expect((await putChunk(upload.id, 0, bytes.slice(0, 5), owner)).status).toBe(413); + expect((await putChunk(upload.id, 0, bytes.slice(0, 3), owner)).status).toBe(400); + + // Out of order and repeated chunks are fine; the client resumes from status. + expect((await putChunk(upload.id, 2, bytes.slice(8), owner)).status).toBe(200); + expect((await putChunk(upload.id, 0, bytes.slice(0, 4), owner)).status).toBe(200); + expect((await putChunk(upload.id, 0, bytes.slice(0, 4), owner)).status).toBe(200); + const client = rpc(owner); + const partial = (await procedure(client, "uploads/status")({ uploadId: upload.id })) as { + receivedChunks: number[]; + }; + expect([...partial.receivedChunks].sort()).toEqual([0, 2]); + expect(await rpcCode(procedure(client, "uploads/complete")({ uploadId: upload.id }))).toBe( + "CONFLICT", + ); + + expect((await putChunk(upload.id, 1, bytes.slice(4, 8), owner)).status).toBe(200); + calls.length = 0; + const done = (await procedure(client, "uploads/complete")({ uploadId: upload.id })) as { + status: string; + }; + expect(done.status).toBe("complete"); + const handed = calls.filter((call) => call.method === "uploadCompleted"); + expect(handed).toHaveLength(1); + const input = handed[0]?.input as { tempPath: string; sha256: string; assetId: string }; + expect(input.assetId).toBe(assetId); + expect(input.sha256).toBe(sha256(bytes)); + expect(input.tempPath.startsWith(path.join(mediaRoot, b.spaceId, "uploads"))).toBe(true); + expect(await readFile(input.tempPath, "utf8")).toBe("0123456789"); + const asset = await repos.getAssetRecord(b.spaceId, assetId); + expect(asset.storage).toBe("uploaded"); + expect(asset.storagePath).toBe(`${b.spaceId}/uploads/${upload.id}/file`); + // Completing again is idempotent and does not hand the file over twice. + await procedure(client, "uploads/complete")({ uploadId: upload.id }); + expect(calls.filter((call) => call.method === "uploadCompleted")).toHaveLength(1); + expect((await putChunk(upload.id, 0, bytes.slice(0, 4), owner)).status).toBe(409); + }); + + it("opens uploads through the RPC only for items that wait for their original", async () => { + const client = rpc(actorOf(b)); + const assetId = await remoteAsset(b, 102); + const input = { assetId, sizeBytes: 10, sha256: "a".repeat(64), fileName: "clip.mp4" }; + const first = (await procedure(client, "uploads/begin")(input)) as { id: string }; + const resumed = (await procedure(client, "uploads/begin")(input)) as { id: string }; + expect(resumed.id).toBe(first.id); + const referenced = { ...input, assetId: seededB.assetId }; + expect(await rpcCode(procedure(client, "uploads/begin")(referenced))).toBe("CONFLICT"); + const tooLarge = { ...input, sizeBytes: 21 * 1024 * 1024 * 1024 }; + expect(await rpcCode(procedure(client, "uploads/begin")(tooLarge))).toBe("BAD_REQUEST"); + }); + + it("fails an upload whose bytes do not match the announced checksum", async () => { + const assetId = await remoteAsset(b, 103); + const bytes = new TextEncoder().encode("abcdef"); + const upload = await repos.beginUpload(b.spaceId, { + assetId, + userId: b.userId, + sizeBytes: bytes.length, + sha256: "f".repeat(64), + fileName: "clip.mp4", + chunkBytes: 4, + }); + const owner = actorOf(b); + await putChunk(upload.id, 0, bytes.slice(0, 4), owner); + await putChunk(upload.id, 1, bytes.slice(4), owner); + const client = rpc(owner); + expect(await rpcCode(procedure(client, "uploads/complete")({ uploadId: upload.id }))).toBe( + "BAD_REQUEST", + ); + const status = (await procedure(client, "uploads/status")({ uploadId: upload.id })) as { + status: string; + }; + expect(status.status).toBe("failed"); + const dir = path.join(mediaRoot, b.spaceId, "uploads", upload.id); + await expect(stat(dir)).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("expires an idle upload and starts a fresh one", async () => { + const assetId = await remoteAsset(b, 104); + const input = { assetId, sizeBytes: 10, sha256: "b".repeat(64), fileName: "clip.mp4" }; + const first = (await procedure(rpc(actorOf(b)), "uploads/begin")(input)) as { id: string }; + setup({ now: () => new Date(Date.now() + MEDIA_UPLOAD_IDLE_MS + 60_000) }); + try { + const response = await putChunk(first.id, 0, new Uint8Array(10), actorOf(b)); + expect(response.status).toBe(409); + const status = (await procedure( + rpc(actorOf(b)), + "uploads/status", + )({ + uploadId: first.id, + })) as { status: string }; + expect(status.status).toBe("failed"); + const fresh = (await procedure(rpc(actorOf(b)), "uploads/begin")(input)) as { id: string }; + expect(fresh.id).not.toBe(first.id); + } finally { + setup(); + } + }); + + it("drops a chunk that does not arrive in time", async () => { + const assetId = await remoteAsset(b, 105); + const upload = await repos.beginUpload(b.spaceId, { + assetId, + userId: b.userId, + sizeBytes: 10, + sha256: "c".repeat(64), + fileName: "clip.mp4", + }); + setup({ chunkTimeoutMs: 50 }); + try { + const stalled = new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(2)); + }, + }); + currentActor = actorOf(b); + const response = await app.request(`/api/media/uploads/${upload.id}/chunks/0`, { + method: "PUT", + body: stalled, + duplex: "half", + } as RequestInit); + expect(response.status).toBe(408); + const status = await repos.uploadStatus(b.spaceId, upload.id); + expect(status.receivedChunks).toEqual([]); + } finally { + setup(); + } + }); + + it("streams render files only inside the caller's space and render folder", async () => { + const renderDir = path.join(mediaRoot, a.spaceId, "renders", seededA.renderId); + await mkdir(renderDir, { recursive: true }); + const video = path.join(renderDir, "video.mp4"); + await writeFile(video, "0123456789"); + renderOutputs.set(seededA.renderId, video); + const url = `/api/media/renders/${seededA.renderId}/files/video.mp4`; + + currentActor = actorOf(a); + const full = await app.request(url); + expect(full.status).toBe(200); + expect(await full.text()).toBe("0123456789"); + const ranged = await app.request(url, { headers: { range: "bytes=2-4" } }); + expect(ranged.status).toBe(206); + expect(ranged.headers.get("content-range")).toBe("bytes 2-4/10"); + expect(await ranged.text()).toBe("234"); + const download = await app.request(`${url}?download=1`); + expect(download.headers.get("content-disposition")).toContain("attachment"); + + // A render that has not passed its gate never leaves as a download. + const queuedVideo = path.join(mediaRoot, a.spaceId, "renders", seededA.queuedRenderId, "v.mp4"); + await mkdir(path.dirname(queuedVideo), { recursive: true }); + await writeFile(queuedVideo, "x"); + renderOutputs.set(seededA.queuedRenderId, queuedVideo); + const queuedUrl = `/api/media/renders/${seededA.queuedRenderId}/files/v.mp4`; + expect((await app.request(queuedUrl)).status).toBe(200); + expect((await app.request(`${queuedUrl}?download=1`)).status).toBe(404); + + // A path the service reports outside the render folder is refused. + const outside = path.join(mediaRoot, a.spaceId, "video.mp4"); + await writeFile(outside, "secret"); + renderOutputs.set(seededA.renderId, outside); + expect((await app.request(url)).status).toBe(404); + renderOutputs.set(seededA.renderId, video); + + currentActor = actorOf(b); + expect((await app.request(url)).status).toBe(404); + expect((await app.request(`/api/media/assets/${seededA.assetId}/thumb`)).status).toBe(404); + expect((await app.request(`/api/media/bundles/${seededA.publicationId}`)).status).toBe(404); + }); + + it("refuses a folder source to members who do not own the deployment", async () => { + const folder = await mkdtemp(path.join(tmpdir(), "media-footage-")); + try { + const input = { label: "Footage", path: folder }; + const member = procedure(rpc(actorOf(b)), "sources/createFolder")(input); + expect(await rpcCode(member)).toBe("FORBIDDEN"); + const created = (await procedure(rpc(actorOf(b, true)), "sources/createFolder")(input)) as { + id: string; + kind: string; + }; + expect(created.kind).toBe("folder"); + const row = await db.prisma.mediaSource.findUniqueOrThrow({ where: { id: created.id } }); + expect(row.ownerUserId).toBe(b.userId); + } finally { + await rm(folder, { recursive: true, force: true }); + } + }); +}); diff --git a/apps/api/src/media.test.ts b/apps/api/src/media.test.ts new file mode 100644 index 0000000..aabd2be --- /dev/null +++ b/apps/api/src/media.test.ts @@ -0,0 +1,333 @@ +import { mkdir, mkdtemp, rm, symlink } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { createRouterClient, ORPCError } from "@orpc/server"; +import type { Actor } from "@rakazo/contracts"; +import { MediaToolInputSchemas, mediaRpcShapes } from "@rakazo/contracts"; +import type { MediaRepos, PrismaClient } from "@rakazo/db"; +import { InvalidMediaInputError, IsolationError, MediaStateConflictError } from "@rakazo/db"; +import { PathOutsideRootError } from "@rakazo/media/node"; +import { Hono } from "hono"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + createMediaRouter, + MEDIA_NOT_CONFIGURED, + mediaRpcError, + unconfiguredCapabilities, + validateMediaFolder, +} from "./media.js"; +import { mountMediaHttpRoutes, parseByteRange } from "./media-http.js"; +import type { MediaSampleIds } from "./media-test-fakes.js"; +import { fakeMediaServices, mediaProcedurePaths, sampleMediaInputs } from "./media-test-fakes.js"; +import { expectedChunkBytes, uploadChunkCount } from "./media-uploads.js"; + +const member: Actor = { + userId: "user-member", + spaceId: "space-a", + email: "member@rakazo.test", + isDeploymentOwner: false, +}; +const owner: Actor = { ...member, userId: "user-owner", isDeploymentOwner: true }; + +const ids = Object.fromEntries( + [ + "sourceId", + "assetId", + "secondAssetId", + "segmentId", + "trackId", + "personId", + "markId", + "collectionId", + "brandKitId", + "projectId", + "planId", + "renderId", + "publicationId", + "mappingId", + "uploadId", + ].map((key) => [key, `id-${key.toLowerCase()}`]), +) as unknown as MediaSampleIds; +const kit = { + name: "Kit", + colors: { + primary: "#101010", + secondary: "#202020", + accent: "#303030", + text: "#f0f0f0", + background: "#000000", + }, +}; + +function client(router: ReturnType, actor: Actor | null) { + return createRouterClient(router, { context: { actor } }); +} + +function procedure(target: unknown, rpcPath: string) { + let node: unknown = target; + for (const key of rpcPath.split("/")) node = (node as Record)[key]; + return node as (input: unknown) => Promise; +} + +async function rpcCode(promise: Promise): Promise { + try { + await promise; + } catch (error) { + if (error instanceof ORPCError) return error.code; + throw error; + } + return "OK"; +} + +/** Repositories that fail the test when reached. */ +function untouchedRepos(): MediaRepos { + return new Proxy({} as MediaRepos, { + get: (_target, key) => () => { + throw new Error(`repository ${String(key)} must not be called`); + }, + }); +} + +const prismaStub = {} as PrismaClient; + +describe("media router without a configured sidecar", () => { + const router = createMediaRouter({ prisma: prismaStub, services: null, dataDir: "/data" }); + const inputs = sampleMediaInputs(ids, kit); + + it("has a sample input for every media procedure", () => { + expect(Object.keys(inputs).sort()).toEqual(mediaProcedurePaths().sort()); + }); + + it("reports media as not configured so every surface keeps it hidden", async () => { + const capabilities = await client(router, member).capabilities({}); + expect(capabilities).toEqual(unconfiguredCapabilities()); + expect(capabilities.configured).toBe(false); + expect( + Object.values(capabilities.operations).every( + (op) => !(op as { available: boolean }).available, + ), + ).toBe(true); + }); + + it("refuses every other procedure", async () => { + for (const [rpcPath, sample] of Object.entries(inputs)) { + if (rpcPath === "capabilities") continue; + const promise = procedure(client(router, member), rpcPath)(sample.input); + await expect(promise, rpcPath).rejects.toMatchObject({ + code: "PRECONDITION_FAILED", + message: MEDIA_NOT_CONFIGURED, + }); + } + }); + + it("requires a signed-in member for every procedure", async () => { + for (const [rpcPath, sample] of Object.entries(inputs)) { + const code = await rpcCode(procedure(client(router, null), rpcPath)(sample.input)); + expect(code, rpcPath).toBe("UNAUTHORIZED"); + } + }); + + it("answers the byte routes with 404 and anonymous callers with 401", async () => { + const app = new Hono(); + let actor: Actor | null = member; + mountMediaHttpRoutes(app, { services: null }, async () => actor); + const requests = [ + new Request("http://api/api/media/uploads/u1/chunks/0", { method: "PUT", body: "abcd" }), + new Request("http://api/api/media/renders/r1/files/video.mp4"), + new Request("http://api/api/media/assets/a1/thumb"), + new Request("http://api/api/media/assets/a1/proxy"), + new Request("http://api/api/media/bundles/p1"), + ]; + for (const request of requests) { + expect((await app.request(request.clone())).status, request.url).toBe(404); + } + actor = null; + for (const request of requests) { + expect((await app.request(request.clone())).status, request.url).toBe(401); + } + }); +}); + +describe("media router permissions", () => { + let root: string; + + beforeAll(async () => { + root = await mkdtemp(path.join(tmpdir(), "media-api-")); + }); + + afterAll(async () => { + await rm(root, { recursive: true, force: true }); + }); + + it("lets only the deployment owner add a folder source", async () => { + const created: unknown[] = []; + const repos = new Proxy({} as MediaRepos, { + get: (_target, key) => { + if (key === "createSource") { + return async (_spaceId: string, input: unknown) => { + created.push(input); + return { + id: "src-1", + kind: "folder", + label: "Footage", + status: "idle", + assetCount: 0, + lastSyncedAt: null, + error: null, + createdAt: "2026-10-01T00:00:00.000Z", + }; + }; + } + return () => { + throw new Error(`repository ${String(key)} must not be called`); + }; + }, + }); + const { services, calls } = fakeMediaServices({ repos, mediaRoot: path.join(root, "media") }); + const footage = path.join(root, "footage"); + await mkdir(footage); + const router = createMediaRouter({ + prisma: prismaStub, + services, + dataDir: path.join(root, "data"), + appRoot: path.join(root, "app"), + }); + + const denied = client(router, member).sources.createFolder({ label: "Footage", path: footage }); + await expect(denied).rejects.toMatchObject({ code: "FORBIDDEN" }); + expect(created).toEqual([]); + expect(calls).toEqual([]); + + await client(router, owner).sources.createFolder({ label: "Footage", path: footage }); + expect(created).toEqual([ + { kind: "folder", label: "Footage", ownerUserId: owner.userId, config: { path: footage } }, + ]); + expect(calls.map((call) => `${call.service}.${call.method}`)).toEqual(["ingest.requestSync"]); + }); + + it("accepts only a real, specific folder apart from evOo's own directories", async () => { + const dataDir = path.join(root, "data"); + const appRoot = path.join(root, "app"); + const folder = path.join(root, "clips"); + await mkdir(path.join(dataDir, "media"), { recursive: true }); + await mkdir(path.join(appRoot, "packages"), { recursive: true }); + await mkdir(folder, { recursive: true }); + await symlink(folder, path.join(root, "clips-link")); + const forbidden = { dataDir, appRoot }; + + await expect(validateMediaFolder(folder, forbidden)).resolves.toBe(folder); + const rejected = [ + "clips", + path.join(root, "missing"), + path.join(root, "clips-link"), + "/", + dataDir, + path.join(dataDir, "media"), + root, + path.join(appRoot, "packages"), + ]; + for (const candidate of rejected) { + await expect(validateMediaFolder(candidate, forbidden), candidate).rejects.toBeInstanceOf( + InvalidMediaInputError, + ); + } + }); +}); + +describe("consent stays with the signed-in user", () => { + it("never accepts consent or age fields from the agent's people tool", () => { + const schema = MediaToolInputSchemas.media_people; + expect(schema.shape.action.options).toEqual(["list", "update", "merge"]); + const parsed = schema.parse({ + action: "update", + personId: "person-1", + consentArtifactId: "artifact-1", + consentRecordedAt: "2026-09-01T00:00:00.000Z", + ageVerificationMethod: "platform_verified", + ageVerifiedAt: "2026-09-01T00:00:00.000Z", + }); + expect(Object.keys(parsed).sort()).toEqual(["action", "personId"]); + }); + + it("drops consent fields sent to people.update", async () => { + const patches: unknown[] = []; + const repos = new Proxy({} as MediaRepos, { + get: (_target, key) => { + if (key === "updatePerson") { + return async (_spaceId: string, _personId: string, patch: unknown) => { + patches.push(patch); + return {}; + }; + } + return () => { + throw new Error(`repository ${String(key)} must not be called`); + }; + }, + }); + const { services } = fakeMediaServices({ repos, mediaRoot: "/unused" }); + const router = createMediaRouter({ prisma: prismaStub, services, dataDir: "/data" }); + await procedure( + client(router, member), + "people/update", + )({ + personId: "person-1", + displayName: "Person 1", + consentRecordedAt: "2026-09-01T00:00:00.000Z", + ageVerificationMethod: "platform_verified", + ageVerifiedAt: "2026-09-01T00:00:00.000Z", + }).catch(() => undefined); + expect(patches).toHaveLength(1); + expect(Object.entries(patches[0] as object).filter(([, value]) => value !== undefined)).toEqual( + [["displayName", "Person 1"]], + ); + }); + + it("marks recordConsent as the only user-only write and keeps it out of the services", () => { + const people = mediaRpcShapes.people as Record; + expect(Object.keys(people)).toContain("recordConsent"); + const { services } = fakeMediaServices({ repos: untouchedRepos(), mediaRoot: "/unused" }); + const serviceMethods = (["ingest", "privacy", "render", "marketing"] as const).flatMap((name) => + Object.keys(services[name] as unknown as Record), + ); + expect(serviceMethods.filter((method) => /consent|age/i.test(method))).toEqual([]); + }); +}); + +describe("media error mapping", () => { + it("never reveals whether a foreign id exists", () => { + expect((mediaRpcError(new IsolationError()) as ORPCError).code).toBe( + "NOT_FOUND", + ); + expect( + (mediaRpcError(new PathOutsideRootError("../x")) as ORPCError).code, + ).toBe("BAD_REQUEST"); + expect( + (mediaRpcError(new MediaStateConflictError("busy")) as ORPCError).code, + ).toBe("CONFLICT"); + const other = new Error("boom"); + expect(mediaRpcError(other)).toBe(other); + }); +}); + +describe("upload and streaming helpers", () => { + it("splits uploads into full chunks plus a shorter last one", () => { + const row = { sizeBytes: 10n, chunkBytes: 4 }; + expect(uploadChunkCount(row)).toBe(3); + expect([0, 1, 2].map((index) => expectedChunkBytes(row, index))).toEqual([4, 4, 2]); + expect(uploadChunkCount({ sizeBytes: 8n, chunkBytes: 4 })).toBe(2); + expect(expectedChunkBytes({ sizeBytes: 8n, chunkBytes: 4 }, 1)).toBe(4); + }); + + it("parses single byte ranges and refuses unsatisfiable ones", () => { + expect(parseByteRange(undefined, 100)).toBeNull(); + expect(parseByteRange("bytes=0-9", 100)).toEqual({ start: 0, end: 9 }); + expect(parseByteRange("bytes=90-", 100)).toEqual({ start: 90, end: 99 }); + expect(parseByteRange("bytes=-10", 100)).toEqual({ start: 90, end: 99 }); + expect(parseByteRange("bytes=50-500", 100)).toEqual({ start: 50, end: 99 }); + expect(parseByteRange("bytes=0-1,5-6", 100)).toBeNull(); + expect(parseByteRange("bytes=100-", 100)).toBe("unsatisfiable"); + expect(parseByteRange("bytes=9-3", 100)).toBe("unsatisfiable"); + expect(parseByteRange("items=0-1", 100)).toBe("unsatisfiable"); + expect(parseByteRange("bytes=-0", 100)).toBe("unsatisfiable"); + }); +}); diff --git a/apps/api/src/media.ts b/apps/api/src/media.ts index 6b4ec4a..038b421 100644 --- a/apps/api/src/media.ts +++ b/apps/api/src/media.ts @@ -1,113 +1,547 @@ +import { lstat, realpath, rm } from "node:fs/promises"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; import { implement, ORPCError } from "@orpc/server"; -import type { Actor } from "@rakazo/contracts"; -import { appContract } from "@rakazo/contracts"; +import { UnsupportedMediaOperationError } from "@rakazo/adapter-kit"; +import type { MediaServices } from "@rakazo/adapters"; +import type { Actor, BrandKit, MediaCapabilities, MediaSourceKind } from "@rakazo/contracts"; +import { appContract, MediaInferenceOperationSchema } from "@rakazo/contracts"; import type { PrismaClient } from "@rakazo/db"; +import { + InvalidMediaInputError, + InvalidProjectWorkspacePathError, + IsolationError, + lockMediaKey, + MediaStateConflictError, + mapMediaSource, + resolveProjectWorkspaceDirectory, +} from "@rakazo/db"; +import { summarizeProfile } from "@rakazo/media"; +import { PathOutsideRootError, resolveInside } from "@rakazo/media/node"; +import { beginMediaUpload, completeMediaUpload, UploadChunkError } from "./media-uploads.js"; export interface MediaRouterDeps { prisma: PrismaClient; + /** Null when Media Studio is not configured: everything but capabilities refuses. */ + services?: MediaServices | null; + /** DATA_DIR; a folder source may neither sit inside it nor contain it. */ dataDir: string; + /** Application install directory a folder source may not point into. */ + appRoot?: string; } -// Phase A scaffold: every procedure is replaced by its real implementation. -function notImplemented(): never { - throw new ORPCError("NOT_IMPLEMENTED", { message: "Media Studio is not available yet" }); +export const MEDIA_NOT_CONFIGURED = "Media Studio is not configured"; + +const DEFAULT_APP_ROOT = fileURLToPath(new URL("../../../", import.meta.url)); + +/** Translate domain errors into RPC errors; everything else passes through. */ +export function mediaRpcError(error: unknown): unknown { + if (error instanceof ORPCError) return error; + if (error instanceof IsolationError) return new ORPCError("NOT_FOUND"); + if ( + error instanceof InvalidMediaInputError || + error instanceof InvalidProjectWorkspacePathError + ) { + return new ORPCError("BAD_REQUEST", { message: error.message }); + } + if (error instanceof PathOutsideRootError) { + return new ORPCError("BAD_REQUEST", { message: "That path is not allowed." }); + } + if (error instanceof MediaStateConflictError) { + return new ORPCError("CONFLICT", { message: error.message }); + } + if (error instanceof UploadChunkError) { + return new ORPCError(error.status === 409 ? "CONFLICT" : "BAD_REQUEST", { + message: error.message, + }); + } + if (error instanceof UnsupportedMediaOperationError) { + return new ORPCError("PRECONDITION_FAILED", { message: error.message }); + } + return error; } -export function createMediaRouter(_deps: MediaRouterDeps) { +export function unconfiguredCapabilities(): MediaCapabilities { + const unavailable = { + available: false, + experimental: false, + model: null, + license: null, + commercialUse: null, + }; + return { + configured: false, + operations: Object.fromEntries( + MediaInferenceOperationSchema.options.map((operation) => [operation, unavailable]), + ) as MediaCapabilities["operations"], + accelerator: "none", + hardwareEncoders: [], + }; +} + +function overlaps(a: string, b: string): boolean { + const inside = (root: string, candidate: string) => { + const relative = path.relative(root, candidate); + return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative)); + }; + return inside(a, b) || inside(b, a); +} + +async function realOrResolved(target: string): Promise { + return realpath(target).catch(() => path.resolve(target)); +} + +/** + * A host folder for a folder source: absolute, an existing directory, not a + * symlink, and apart from DATA_DIR and the application itself. Returns the real path. + */ +export async function validateMediaFolder( + input: string, + forbidden: { dataDir: string; appRoot: string }, +): Promise { + const trimmed = input.trim(); + if (!path.isAbsolute(trimmed)) { + throw new InvalidMediaInputError("Use an absolute folder path."); + } + const info = await lstat(trimmed).catch(() => null); + if (!info) throw new InvalidMediaInputError("That folder does not exist or is not readable."); + if (info.isSymbolicLink()) + throw new InvalidMediaInputError("Choose the folder, not a link to it."); + const resolved = await resolveProjectWorkspaceDirectory(trimmed); + if (path.parse(resolved).root === resolved) { + throw new InvalidMediaInputError("Choose a specific folder."); + } + const dataDir = await realOrResolved(forbidden.dataDir); + const appRoot = await realOrResolved(forbidden.appRoot); + const relativeToApp = path.relative(appRoot, resolved); + const insideApp = + relativeToApp === "" || (!relativeToApp.startsWith("..") && !path.isAbsolute(relativeToApp)); + if (overlaps(dataDir, resolved) || insideApp) { + throw new InvalidMediaInputError("That folder belongs to evOo itself."); + } + return resolved; +} + +/** Artifacts referenced from media records must belong to the caller's space. */ +export async function assertSpaceArtifacts( + prisma: PrismaClient, + spaceId: string, + artifactIds: string[], +): Promise { + const ids = [...new Set(artifactIds)]; + if (ids.length === 0) return; + const owned = await prisma.artifact.count({ where: { spaceId, id: { in: ids } } }); + if (owned !== ids.length) throw new IsolationError(); +} + +function brandKitArtifactIds(kit: BrandKit): string[] { + return [ + ...kit.fonts.map((font) => font.artifactId), + ...kit.logoArtifactIds, + ...(kit.lutArtifactId ? [kit.lutArtifactId] : []), + ...kit.musicBeds.map((bed) => bed.artifactId), + ]; +} + +async function removeMediaPath(mediaRoot: string, relative: string | null): Promise { + if (!relative) return; + await rm(await resolveInside(mediaRoot, relative), { recursive: true, force: true }); +} + +export function createMediaRouter(deps: MediaRouterDeps) { const os = implement(appContract.media).$context<{ actor: Actor | null; signal?: AbortSignal }>(); const authed = os.use(async ({ context, next }) => { if (!context.actor) throw new ORPCError("UNAUTHORIZED"); - return next({ context: { ...context, actor: context.actor } }); + try { + return await next({ context: { ...context, actor: context.actor } }); + } catch (error) { + throw mediaRpcError(error); + } }); + const media = authed.use(async ({ next }) => { + if (!deps.services) { + throw new ORPCError("PRECONDITION_FAILED", { message: MEDIA_NOT_CONFIGURED }); + } + return next({ context: { services: deps.services } }); + }); + const forbiddenRoots = { dataDir: deps.dataDir, appRoot: deps.appRoot ?? DEFAULT_APP_ROOT }; + + async function knownProfiles(services: MediaServices, profileIds: string[] | undefined) { + if (!profileIds) return; + const { profiles } = await services.deps.profiles(); + const unknown = profileIds.filter((id) => !profiles.has(id)); + if (unknown.length > 0) { + throw new InvalidMediaInputError(`Unknown profile: ${unknown.join(", ")}`); + } + } return os.router({ - capabilities: authed.capabilities.handler(notImplemented), + capabilities: authed.capabilities.handler(async ({ context }) => { + const services = deps.services; + if (!services) return unconfiguredCapabilities(); + const hardwareEncoders = await services.deps.renderer.hardwareEncoders().catch(() => []); + const inference = services.deps.inference; + if (!inference) return { ...unconfiguredCapabilities(), hardwareEncoders }; + const capabilities = await inference + .capabilities(context.signal) + .catch(() => ({ ...unconfiguredCapabilities(), configured: true })); + return { ...capabilities, configured: true, hardwareEncoders }; + }), sources: { - list: authed.sources.list.handler(notImplemented), - createFolder: authed.sources.createFolder.handler(notImplemented), - registerClient: authed.sources.registerClient.handler(notImplemented), - pushItems: authed.sources.pushItems.handler(notImplemented), - sync: authed.sources.sync.handler(notImplemented), - remove: authed.sources.remove.handler(notImplemented), + list: media.sources.list.handler(({ context }) => + context.services.deps.repos.listSources(context.actor.spaceId), + ), + createFolder: media.sources.createFolder.handler(async ({ context, input }) => { + // A folder lives on the host, and the host is the deployment owner's machine. + if (!context.actor.isDeploymentOwner) throw new ORPCError("FORBIDDEN"); + const folder = await validateMediaFolder(input.path, forbiddenRoots); + const { repos } = context.services.deps; + const source = await repos.createSource(context.actor.spaceId, { + kind: "folder", + label: input.label, + ownerUserId: context.actor.userId, + config: { path: folder }, + }); + await context.services.ingest.requestSync(context.actor, source.id); + return source; + }), + registerClient: media.sources.registerClient.handler(async ({ context, input }) => { + const { spaceId, userId } = context.actor; + const kind: MediaSourceKind = input.kind; + // One source per device: re-registering returns the existing one. + return deps.prisma.$transaction(async (tx) => { + await lockMediaKey(tx, `source:${spaceId}:${kind}:${input.deviceId}`); + const existing = await tx.mediaSource.findFirst({ + where: { spaceId, kind, config: { path: ["deviceId"], equals: input.deviceId } }, + orderBy: [{ createdAt: "asc" }, { id: "asc" }], + include: { _count: { select: { assets: true } } }, + }); + if (existing) return mapMediaSource(existing, existing._count.assets); + const created = await tx.mediaSource.create({ + data: { + spaceId, + kind, + label: input.label, + ownerUserId: userId, + config: { deviceId: input.deviceId }, + }, + }); + return mapMediaSource(created, 0); + }); + }), + pushItems: media.sources.pushItems.handler(async ({ context, input }) => { + await context.services.deps.repos.getSource(context.actor.spaceId, input.sourceId); + return context.services.ingest.pushItems(context.actor, input); + }), + sync: media.sources.sync.handler(async ({ context, input }) => { + await context.services.deps.repos.getSource(context.actor.spaceId, input.sourceId); + await context.services.ingest.requestSync(context.actor, input.sourceId); + return { ok: true as const }; + }), + remove: media.sources.remove.handler(async ({ context, input }) => { + await context.services.deps.repos.getSource(context.actor.spaceId, input.sourceId); + await context.services.ingest.removeSource(context.actor, input.sourceId); + return { ok: true as const }; + }), }, assets: { - search: authed.assets.search.handler(notImplemented), - get: authed.assets.get.handler(notImplemented), - setPaidOnly: authed.assets.setPaidOnly.handler(notImplemented), - requestUpload: authed.assets.requestUpload.handler(notImplemented), + search: media.assets.search.handler(({ context, input }) => + context.services.deps.repos.searchAssets(context.actor.spaceId, input), + ), + get: media.assets.get.handler(({ context, input }) => + context.services.deps.repos.getAsset(context.actor.spaceId, input.assetId), + ), + setPaidOnly: media.assets.setPaidOnly.handler(async ({ context, input }) => { + await context.services.deps.repos.setSegmentsPaidOnly( + context.actor.spaceId, + input.segmentIds, + input.paidOnly, + ); + return { ok: true as const }; + }), + requestUpload: media.assets.requestUpload.handler(async ({ context, input }) => ({ + awaiting: await context.services.deps.repos.requestUpload( + context.actor.spaceId, + input.assetIds, + ), + })), }, uploads: { - begin: authed.uploads.begin.handler(notImplemented), - status: authed.uploads.status.handler(notImplemented), - complete: authed.uploads.complete.handler(notImplemented), + begin: media.uploads.begin.handler(({ context, input }) => + beginMediaUpload(context.services, context.actor, input), + ), + status: media.uploads.status.handler(({ context, input }) => + context.services.deps.repos.uploadStatus(context.actor.spaceId, input.uploadId), + ), + complete: media.uploads.complete.handler(({ context, input }) => + completeMediaUpload(context.services, context.actor, input.uploadId), + ), }, collections: { - list: authed.collections.list.handler(notImplemented), - get: authed.collections.get.handler(notImplemented), - rename: authed.collections.rename.handler(notImplemented), - merge: authed.collections.merge.handler(notImplemented), - split: authed.collections.split.handler(notImplemented), - createManual: authed.collections.createManual.handler(notImplemented), - recluster: authed.collections.recluster.handler(notImplemented), + list: media.collections.list.handler(({ context, input }) => + context.services.deps.repos.listCollections(context.actor.spaceId, input), + ), + get: media.collections.get.handler(({ context, input }) => + context.services.deps.repos.getCollection(context.actor.spaceId, input.collectionId), + ), + rename: media.collections.rename.handler(({ context, input }) => + context.services.deps.repos.renameCollection( + context.actor.spaceId, + input.collectionId, + input.title, + ), + ), + merge: media.collections.merge.handler(({ context, input }) => + context.services.deps.repos.mergeCollections(context.actor.spaceId, input.collectionIds), + ), + split: media.collections.split.handler(({ context, input }) => + context.services.deps.repos.splitCollection( + context.actor.spaceId, + input.collectionId, + input.atAssetId, + ), + ), + createManual: media.collections.createManual.handler(({ context, input }) => + context.services.deps.repos.createManualCollection( + context.actor.spaceId, + input.title, + input.assetIds, + ), + ), + recluster: media.collections.recluster.handler(async ({ context }) => { + await context.services.ingest.recluster(context.actor); + return { ok: true as const }; + }), }, people: { - list: authed.people.list.handler(notImplemented), - update: authed.people.update.handler(notImplemented), - merge: authed.people.merge.handler(notImplemented), - assignTrack: authed.people.assignTrack.handler(notImplemented), - recordConsent: authed.people.recordConsent.handler(notImplemented), - remove: authed.people.remove.handler(notImplemented), + list: media.people.list.handler(({ context }) => + context.services.deps.repos.listPeople(context.actor.spaceId), + ), + update: media.people.update.handler(({ context, input }) => + context.services.deps.repos.updatePerson(context.actor.spaceId, input.personId, { + displayName: input.displayName, + isSelf: input.isSelf, + confirmed: input.confirmed, + anonymizeByDefault: input.anonymizeByDefault, + }), + ), + merge: media.people.merge.handler(({ context, input }) => + context.services.deps.repos.mergePeople(context.actor.spaceId, input.personIds), + ), + assignTrack: media.people.assignTrack.handler(({ context, input }) => + context.services.deps.repos.assignTrack( + context.actor.spaceId, + input.trackId, + input.personId, + ), + ), + // The only consent write path: the signed-in member is the recorder. + recordConsent: media.people.recordConsent.handler(async ({ context, input }) => { + await assertSpaceArtifacts( + deps.prisma, + context.actor.spaceId, + input.consentArtifactId ? [input.consentArtifactId] : [], + ); + return context.services.deps.repos.recordConsent( + { spaceId: context.actor.spaceId, recordedByUserId: context.actor.userId }, + input, + ); + }), + remove: media.people.remove.handler(async ({ context, input }) => { + await context.services.deps.repos.getPerson(context.actor.spaceId, input.personId); + await context.services.privacy.removePerson(context.actor, input.personId); + return { ok: true as const }; + }), }, marks: { - list: authed.marks.list.handler(notImplemented), - create: authed.marks.create.handler(notImplemented), - update: authed.marks.update.handler(notImplemented), - remove: authed.marks.remove.handler(notImplemented), + list: media.marks.list.handler(({ context, input }) => + context.services.deps.repos.listMarks(context.actor.spaceId, { + ...(input.personId ? { personId: input.personId } : {}), + ...(input.assetId ? { assetId: input.assetId } : {}), + }), + ), + create: media.marks.create.handler(async ({ context, input }) => { + const { repos } = context.services.deps; + await repos.getAssetRecord(context.actor.spaceId, input.assetId); + if (input.personId) await repos.getPerson(context.actor.spaceId, input.personId); + return context.services.privacy.createMark(context.actor, input); + }), + update: media.marks.update.handler(async ({ context, input }) => { + const { repos } = context.services.deps; + await repos.getMark(context.actor.spaceId, input.markId); + if (input.personId) await repos.getPerson(context.actor.spaceId, input.personId); + return context.services.privacy.updateMark(context.actor, input); + }), + remove: media.marks.remove.handler(async ({ context, input }) => { + const { repos, mediaRoot } = context.services.deps; + const { maskDir } = await repos.removeMark(context.actor.spaceId, input.markId); + await removeMediaPath(mediaRoot, maskDir); + return { ok: true as const }; + }), }, brandKits: { - list: authed.brandKits.list.handler(notImplemented), - upsert: authed.brandKits.upsert.handler(notImplemented), - remove: authed.brandKits.remove.handler(notImplemented), - templates: authed.brandKits.templates.handler(notImplemented), - introVariants: authed.brandKits.introVariants.handler(notImplemented), + list: media.brandKits.list.handler(({ context }) => + context.services.deps.repos.listBrandKits(context.actor.spaceId), + ), + upsert: media.brandKits.upsert.handler(async ({ context, input }) => { + await assertSpaceArtifacts( + deps.prisma, + context.actor.spaceId, + brandKitArtifactIds(input.kit), + ); + return context.services.deps.repos.upsertBrandKit(context.actor.spaceId, input); + }), + remove: media.brandKits.remove.handler(async ({ context, input }) => { + await context.services.deps.repos.removeBrandKit(context.actor.spaceId, input.id); + return { ok: true as const }; + }), + templates: media.brandKits.templates.handler(({ context }) => + context.services.deps.motion.templates(), + ), + introVariants: media.brandKits.introVariants.handler(async ({ context, input }) => { + await context.services.deps.repos.getBrandKit(context.actor.spaceId, input.id); + return context.services.marketing.introVariants(context.actor, { + brandKitId: input.id, + count: input.count, + }); + }), }, profiles: { - list: authed.profiles.list.handler(notImplemented), - get: authed.profiles.get.handler(notImplemented), + list: media.profiles.list.handler(async ({ context }) => { + const { profiles } = await context.services.deps.profiles(); + const now = context.services.deps.now(); + return [...profiles.values()].map(({ profile, origin }) => + summarizeProfile(profile, origin, now), + ); + }), + get: media.profiles.get.handler(async ({ context, input }) => { + const { profiles } = await context.services.deps.profiles(); + const entry = profiles.get(input.profileId); + if (!entry) throw new ORPCError("NOT_FOUND"); + return entry.profile; + }), }, projects: { - list: authed.projects.list.handler(notImplemented), - get: authed.projects.get.handler(notImplemented), - create: authed.projects.create.handler(notImplemented), - update: authed.projects.update.handler(notImplemented), - remove: authed.projects.remove.handler(notImplemented), - autoPlan: authed.projects.autoPlan.handler(notImplemented), + list: media.projects.list.handler(({ context, input }) => + context.services.deps.repos.listProjects(context.actor.spaceId, input), + ), + get: media.projects.get.handler(({ context, input }) => + context.services.deps.repos.getProject(context.actor.spaceId, input.projectId), + ), + create: media.projects.create.handler(async ({ context, input }) => { + await knownProfiles(context.services, input.profileIds); + return context.services.deps.repos.createProject( + context.actor.spaceId, + context.actor.userId, + input, + ); + }), + update: media.projects.update.handler(async ({ context, input }) => { + await knownProfiles(context.services, input.profileIds); + const { projectId, ...patch } = input; + return context.services.deps.repos.updateProject(context.actor.spaceId, projectId, patch); + }), + remove: media.projects.remove.handler(async ({ context, input }) => { + const { repos, mediaRoot } = context.services.deps; + const { spaceId } = context.actor; + await repos.getProjectRecord(spaceId, input.projectId); + const renders = await repos.listRenders(spaceId, input.projectId); + await repos.removeProject(spaceId, input.projectId); + for (const render of renders) + await removeMediaPath(mediaRoot, `${spaceId}/renders/${render.id}`); + return { ok: true as const }; + }), + autoPlan: media.projects.autoPlan.handler(async ({ context, input }) => { + await context.services.deps.repos.getProjectRecord(context.actor.spaceId, input.projectId); + await knownProfiles(context.services, input.profileIds); + return context.services.marketing.autoPlan(context.actor, input); + }), }, plans: { - submit: authed.plans.submit.handler(notImplemented), + submit: media.plans.submit.handler(async ({ context, input }) => { + await context.services.deps.repos.getProjectRecord(context.actor.spaceId, input.projectId); + return context.services.marketing.submitPlan(context.actor, { + projectId: input.projectId, + edl: input.edl, + rationale: input.rationale, + createdBy: "user", + }); + }), }, renders: { - start: authed.renders.start.handler(notImplemented), - get: authed.renders.get.handler(notImplemented), - cancel: authed.renders.cancel.handler(notImplemented), - thumbnails: authed.renders.thumbnails.handler(notImplemented), + start: media.renders.start.handler(async ({ context, input }) => { + await context.services.deps.repos.getPlan(context.actor.spaceId, input.planId); + return context.services.render.start(context.actor, input); + }), + get: media.renders.get.handler(({ context, input }) => + context.services.deps.repos.getRender(context.actor.spaceId, input.renderId), + ), + cancel: media.renders.cancel.handler(async ({ context, input }) => { + await context.services.deps.repos.getRender(context.actor.spaceId, input.renderId); + return context.services.render.cancel(context.actor, input.renderId); + }), + thumbnails: media.renders.thumbnails.handler(async ({ context, input }) => { + await context.services.deps.repos.getRender(context.actor.spaceId, input.renderId); + return context.services.marketing.thumbnails(context.actor, input); + }), }, marketing: { - copy: authed.marketing.copy.handler(notImplemented), + copy: media.marketing.copy.handler(async ({ context, input }) => { + await context.services.deps.repos.getRender(context.actor.spaceId, input.renderId); + return context.services.marketing.copy(context.actor, input); + }), }, publications: { - list: authed.publications.list.handler(notImplemented), - create: authed.publications.create.handler(notImplemented), - bundle: authed.publications.bundle.handler(notImplemented), - markPublished: authed.publications.markPublished.handler(notImplemented), - cancel: authed.publications.cancel.handler(notImplemented), + list: media.publications.list.handler(({ context, input }) => + context.services.deps.repos.listPublications(context.actor.spaceId, input), + ), + create: media.publications.create.handler(async ({ context, input }) => { + await context.services.deps.repos.getRender(context.actor.spaceId, input.renderId); + return context.services.marketing.createPublication(context.actor, input); + }), + bundle: media.publications.bundle.handler(async ({ context, input }) => { + await context.services.deps.repos.getPublication( + context.actor.spaceId, + input.publicationId, + ); + return context.services.marketing.bundle(context.actor, input.publicationId); + }), + markPublished: media.publications.markPublished.handler(({ context, input }) => + context.services.deps.repos.markPublicationPublished( + context.actor.spaceId, + input.publicationId, + input.externalUrl, + context.services.deps.now(), + ), + ), + cancel: media.publications.cancel.handler(async ({ context, input }) => { + await context.services.deps.repos.getPublication( + context.actor.spaceId, + input.publicationId, + ); + return context.services.marketing.cancelPublication(context.actor, input.publicationId); + }), }, metrics: { - mappings: authed.metrics.mappings.handler(notImplemented), - saveMapping: authed.metrics.saveMapping.handler(notImplemented), - importCsv: authed.metrics.importCsv.handler(notImplemented), - record: authed.metrics.record.handler(notImplemented), - funnel: authed.metrics.funnel.handler(notImplemented), + mappings: media.metrics.mappings.handler(({ context }) => + context.services.deps.repos.listMetricMappings(context.actor.spaceId), + ), + saveMapping: media.metrics.saveMapping.handler(({ context, input }) => + context.services.deps.repos.saveMetricMapping(context.actor.spaceId, input), + ), + importCsv: media.metrics.importCsv.handler(async ({ context, input }) => { + await context.services.deps.repos.getMetricMapping(context.actor.spaceId, input.mappingId); + return context.services.marketing.importMetricsCsv(context.actor, input); + }), + record: media.metrics.record.handler(({ context, input }) => + context.services.deps.repos.recordMetric(context.actor.spaceId, { + ...input, + source: "manual", + }), + ), + funnel: media.metrics.funnel.handler(({ context, input }) => + context.services.deps.repos.funnel(context.actor.spaceId, input.projectId), + ), }, }); }