diff --git a/apps/api/src/media-env.ts b/apps/api/src/media-env.ts deleted file mode 100644 index 701f9ee..0000000 --- a/apps/api/src/media-env.ts +++ /dev/null @@ -1,19 +0,0 @@ -import { mediaConfigFromEnv } from "@rakazo/adapters"; - -export interface MediaEnv { - /** Generic sidecar connection; unset keeps Media Studio hidden (except the scripted emulator). */ - mediaWorkerUrl: string | undefined; - mediaWorkerToken: string | undefined; - /** Absolute directory of private platform profiles that override the builtin ones. */ - mediaProfilesDir: string; -} - -/** Media Studio settings, read through the adapters' single media config parser. */ -export function readMediaEnv(source: NodeJS.ProcessEnv = process.env): MediaEnv { - const config = mediaConfigFromEnv(source); - return { - mediaWorkerUrl: config.workerUrl ?? undefined, - mediaWorkerToken: config.workerToken ?? undefined, - mediaProfilesDir: config.profilesDir, - }; -} diff --git a/apps/api/src/media-http.ts b/apps/api/src/media-http.ts index 62fbdc2..2dd7369 100644 --- a/apps/api/src/media-http.ts +++ b/apps/api/src/media-http.ts @@ -8,7 +8,7 @@ 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 { UploadChunkError, writeUploadChunk } from "./media-uploads.js"; +import { MEDIA_CHUNK_TIMEOUT_MS, UploadChunkError, writeUploadChunk } from "./media-uploads.js"; /* * Byte routes Media Studio cannot express as JSON RPC: chunk uploads and @@ -20,8 +20,6 @@ import { UploadChunkError, writeUploadChunk } from "./media-uploads.js"; 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}$/; -/** A single chunk must arrive within this window; slow links resume per chunk. */ -const CHUNK_TIMEOUT_MS = 10 * 60_000; const CONTENT_TYPES: Record = { ".mp4": "video/mp4", @@ -134,7 +132,11 @@ function mediaHttpError(c: Context, error: unknown): Response { export function mountMediaHttpRoutes( app: Hono, - deps: { services: MediaServices | null }, + deps: { + /** Null when Media Studio is not configured: every route answers 404. */ + services: MediaServices | null; + chunkTimeoutMs?: number; + }, resolveActor: (c: Context) => Promise, ) { async function authorize( @@ -169,10 +171,8 @@ export function mountMediaHttpRoutes( } const lengthHeader = c.req.header("content-length"); const declared = lengthHeader && /^[0-9]+$/.test(lengthHeader) ? Number(lengthHeader) : null; - const timeout = AbortSignal.timeout(CHUNK_TIMEOUT_MS); + const timeout = AbortSignal.timeout(deps.chunkTimeoutMs ?? MEDIA_CHUNK_TIMEOUT_MS); const body = c.req.raw.body; - const onTimeout = () => cancelBody(body); - timeout.addEventListener("abort", onTimeout, { once: true }); try { const upload = await writeUploadChunk({ services: auth.services, @@ -181,14 +181,13 @@ export function mountMediaHttpRoutes( 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); - } finally { - timeout.removeEventListener("abort", onTimeout); } }); diff --git a/apps/api/src/media-fakes.ts b/apps/api/src/media-test-fakes.ts similarity index 99% rename from apps/api/src/media-fakes.ts rename to apps/api/src/media-test-fakes.ts index ac7ec5d..69c8872 100644 --- a/apps/api/src/media-fakes.ts +++ b/apps/api/src/media-test-fakes.ts @@ -31,6 +31,7 @@ export interface FakeMediaServicesOptions { prisma?: PrismaClient; inference?: MediaInferenceProvider | null; profiles?: PlatformProfile[]; + now?: () => Date; /** Render id → absolute output path the fake render service reports. */ renderOutputs?: Map; onUploadCompleted?: ( @@ -194,7 +195,7 @@ export function fakeMediaServices(options: FakeMediaServicesOptions): { artifacts: {}, model: { generate: async () => null }, publish: async () => undefined, - now: () => new Date("2026-10-01T00:00:00.000Z"), + now: options.now ?? (() => new Date()), } as unknown as MediaServiceDeps; return { services: { deps, ingest, privacy, render, marketing }, calls }; diff --git a/apps/api/src/media-uploads.ts b/apps/api/src/media-uploads.ts index 2d52365..084404c 100644 --- a/apps/api/src/media-uploads.ts +++ b/apps/api/src/media-uploads.ts @@ -4,6 +4,7 @@ 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"; @@ -13,8 +14,17 @@ import { resolveInside } from "@rakazo/media/node"; * 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, @@ -25,26 +35,15 @@ export class UploadChunkError extends Error { } } -interface UploadRow { - id: string; - spaceId: string; - assetId: string; - sizeBytes: bigint; - chunkBytes: number; - sha256: string; - fileName: string; - status: string; -} +type UploadRow = Awaited>; +type UploadSize = Pick; -export function uploadChunkCount(row: Pick): number { +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: Pick, - index: number, -): number { +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)); @@ -54,13 +53,58 @@ function uploadDir(row: Pick): string { return `${row.spaceId}/uploads/${row.id}`; } -async function chunksDir(mediaRoot: string, row: UploadRow): Promise { +async function mediaPath(mediaRoot: string, relative: string): Promise { await mkdir(mediaRoot, { recursive: true }); - const dir = await resolveInside(mediaRoot, `${uploadDir(row)}/chunks`); + 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. @@ -72,9 +116,13 @@ export async function writeUploadChunk(input: { index: number; body: ReadableStream | null; declaredLength: number | null; + /** Aborting stops reading the body; the chunk is then discarded as incomplete. */ + signal?: AbortSignal; }): Promise { - const { repos, mediaRoot } = input.services.deps; - const row = (await repos.getUploadRecord(input.spaceId, input.uploadId)) as UploadRow; + 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) { @@ -92,7 +140,10 @@ export async function writeUploadChunk(input: { 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; @@ -106,6 +157,7 @@ export async function writeUploadChunk(input: { 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 }); @@ -126,11 +178,11 @@ export async function completeMediaUpload( uploadId: string, ): Promise { const { repos, mediaRoot } = services.deps; - const row = (await repos.getUploadRecord(actor.spaceId, uploadId)) as UploadRow; + 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 status = await repos.uploadStatus(actor.spaceId, uploadId); - const received = new Set(status.receivedChunks); + 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"); @@ -140,8 +192,8 @@ export async function completeMediaUpload( try { const dir = await chunksDir(mediaRoot, row); const relativeFile = `${uploadDir(row)}/file`; - const filePath = await resolveInside(mediaRoot, relativeFile); - const tempFile = await resolveInside(mediaRoot, `${uploadDir(row)}/file.${randomUUID()}.tmp`); + 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"); @@ -152,10 +204,10 @@ export async function completeMediaUpload( if (!info || info.size !== expectedChunkBytes(row, index)) { throw new MediaStateConflictError("Upload is missing chunks"); } - for await (const buffer of createReadStream(chunkPath)) { - hash.update(buffer as Buffer); - total += (buffer as Buffer).byteLength; - await out.write(buffer as Buffer); + for await (const buffer of createReadStream(chunkPath) as AsyncIterable) { + hash.update(buffer); + total += buffer.byteLength; + await out.write(buffer); } } } catch (error) { @@ -166,28 +218,26 @@ export async function completeMediaUpload( await out.close(); const sha256 = hash.digest("hex"); if (sha256 !== row.sha256 || BigInt(total) !== row.sizeBytes) { - await rm(await resolveInside(mediaRoot, uploadDir(row)), { recursive: true, force: true }); + await removeUploadFiles(mediaRoot, row); await repos.failUpload(actor.spaceId, uploadId); throw new UploadChunkError(400, "Upload does not match its checksum"); } await rename(tempFile, filePath); - const before = (await repos.getAssetRecord(actor.spaceId, row.assetId)).storagePath; + await rm(dir, { recursive: true, force: true }); await services.ingest.uploadCompleted(actor, { assetId: row.assetId, tempPath: filePath, sha256, fileName: row.fileName, }); - // Ingest moves the file into the asset folder and records where; fall back - // to the assembled file when it left the bytes in place. - const after = (await repos.getAssetRecord(actor.spaceId, row.assetId)).storagePath; - const storagePath = after && after !== before ? after : relativeFile; - const done = await repos.completeUpload(actor.spaceId, uploadId, { storagePath }); - await rm(dir, { recursive: true, force: true }); - if (storagePath !== relativeFile) { - await rm(await resolveInside(mediaRoot, uploadDir(row)), { recursive: true, force: true }); + // 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 done; + 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..623326f --- /dev/null +++ b/apps/api/src/media.test.ts @@ -0,0 +1,331 @@ +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", + }, +}; + +type Client = ReturnType>>; + +function client(router: ReturnType, actor: Actor | null): Client { + return createRouterClient(router, { context: { actor } }); +} + +function procedure(target: Client, 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.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 7361a12..038b421 100644 --- a/apps/api/src/media.ts +++ b/apps/api/src/media.ts @@ -4,7 +4,7 @@ import { fileURLToPath } from "node:url"; import { implement, ORPCError } from "@orpc/server"; import { UnsupportedMediaOperationError } from "@rakazo/adapter-kit"; import type { MediaServices } from "@rakazo/adapters"; -import type { Actor, MediaCapabilities, MediaSourceKind } from "@rakazo/contracts"; +import type { Actor, BrandKit, MediaCapabilities, MediaSourceKind } from "@rakazo/contracts"; import { appContract, MediaInferenceOperationSchema } from "@rakazo/contracts"; import type { PrismaClient } from "@rakazo/db"; import { @@ -18,7 +18,7 @@ import { } from "@rakazo/db"; import { summarizeProfile } from "@rakazo/media"; import { PathOutsideRootError, resolveInside } from "@rakazo/media/node"; -import { completeMediaUpload, UploadChunkError } from "./media-uploads.js"; +import { beginMediaUpload, completeMediaUpload, UploadChunkError } from "./media-uploads.js"; export interface MediaRouterDeps { prisma: PrismaClient; @@ -122,6 +122,27 @@ export async function validateMediaFolder( 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 }); @@ -246,20 +267,9 @@ export function createMediaRouter(deps: MediaRouterDeps) { })), }, uploads: { - begin: media.uploads.begin.handler(async ({ context, input }) => { - const { repos } = context.services.deps; - const asset = await repos.getAssetRecord(context.actor.spaceId, input.assetId); - if (asset.storage !== "remote" && asset.ingestState !== "awaiting_upload") { - throw new MediaStateConflictError("This item does not need an upload"); - } - return repos.beginUpload(context.actor.spaceId, { - assetId: input.assetId, - userId: context.actor.userId, - sizeBytes: input.sizeBytes, - sha256: input.sha256, - fileName: input.fileName, - }); - }), + 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), ), @@ -326,12 +336,17 @@ export function createMediaRouter(deps: MediaRouterDeps) { ), ), // The only consent write path: the signed-in member is the recorder. - recordConsent: media.people.recordConsent.handler(({ context, input }) => - context.services.deps.repos.recordConsent( + 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); @@ -368,9 +383,14 @@ export function createMediaRouter(deps: MediaRouterDeps) { list: media.brandKits.list.handler(({ context }) => context.services.deps.repos.listBrandKits(context.actor.spaceId), ), - upsert: media.brandKits.upsert.handler(({ context, input }) => - context.services.deps.repos.upsertBrandKit(context.actor.spaceId, input), - ), + 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 };