cynarAI/evOo-local
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

feat(media): media API procedures, chunked uploads, and authorization tests

Browse Source
This commit is contained in:
Claude 2026-10-03 18:37:51 +00:00
parent 4721f00546
commit e601b26292
No known key found for this signature in database
7 changed files with 989 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
apps/api/src/media-env.ts
View File

@ -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,
};
}

17
apps/api/src/media-http.ts
View File

@ -8,7 +8,7 @@ import { IsolationError, MediaStateConflictError } from "@rakazo/db";
import { assertInside, PathOutsideRootError, resolveInside } from "@rakazo/media/node"; import { assertInside, PathOutsideRootError, resolveInside } from "@rakazo/media/node";
import type { Context, Hono } from "hono"; import type { Context, Hono } from "hono";
import { cancelBody } from "./http-body.js"; 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 * 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 ID = /^[A-Za-z0-9_-]{1,64}$/;
const FILE_NAME = /^[A-Za-z0-9][A-Za-z0-9._-]{0,199}$/; const FILE_NAME = /^[A-Za-z0-9][A-Za-z0-9._-]{0,199}$/;
const CHUNK_INDEX = /^[0-9]{1,7}$/; 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<string, string> = { const CONTENT_TYPES: Record<string, string> = {
".mp4": "video/mp4", ".mp4": "video/mp4",
@ -134,7 +132,11 @@ function mediaHttpError(c: Context, error: unknown): Response {
export function mountMediaHttpRoutes( export function mountMediaHttpRoutes(
app: Hono, 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<Actor | null>, resolveActor: (c: Context) => Promise<Actor | null>,
) { ) {
async function authorize( async function authorize(
@ -169,10 +171,8 @@ export function mountMediaHttpRoutes(
} }
const lengthHeader = c.req.header("content-length"); const lengthHeader = c.req.header("content-length");
const declared = lengthHeader && /^[0-9]+$/.test(lengthHeader) ? Number(lengthHeader) : null; 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 body = c.req.raw.body;
const onTimeout = () => cancelBody(body);
timeout.addEventListener("abort", onTimeout, { once: true });
try { try {
const upload = await writeUploadChunk({ const upload = await writeUploadChunk({
services: auth.services, services: auth.services,
@ -181,14 +181,13 @@ export function mountMediaHttpRoutes(
index: Number(index), index: Number(index),
body, body,
declaredLength: declared, declaredLength: declared,
signal: AbortSignal.any([timeout, c.req.raw.signal]),
}); });
return c.json(upload); return c.json(upload);
} catch (error) { } catch (error) {
cancelBody(body); cancelBody(body);
if (timeout.aborted) return c.json({ error: "Chunk took too long" }, 408); if (timeout.aborted) return c.json({ error: "Chunk took too long" }, 408);
return mediaHttpError(c, error); return mediaHttpError(c, error);
} finally {
timeout.removeEventListener("abort", onTimeout);
} }
}); });

3
apps/api/src/media-fakes.ts → apps/api/src/media-test-fakes.ts
View File

@ -31,6 +31,7 @@ export interface FakeMediaServicesOptions {
prisma?: PrismaClient; prisma?: PrismaClient;
inference?: MediaInferenceProvider | null; inference?: MediaInferenceProvider | null;
profiles?: PlatformProfile[]; profiles?: PlatformProfile[];
now?: () => Date;
/** Render id → absolute output path the fake render service reports. */ /** Render id → absolute output path the fake render service reports. */
renderOutputs?: Map<string, string>; renderOutputs?: Map<string, string>;
onUploadCompleted?: ( onUploadCompleted?: (
@ -194,7 +195,7 @@ export function fakeMediaServices(options: FakeMediaServicesOptions): {
artifacts: {}, artifacts: {},
model: { generate: async () => null }, model: { generate: async () => null },
publish: async () => undefined, publish: async () => undefined,
now: () => new Date("2026-10-01T00:00:00.000Z"), now: options.now ?? (() => new Date()),
} as unknown as MediaServiceDeps; } as unknown as MediaServiceDeps;
return { services: { deps, ingest, privacy, render, marketing }, calls }; return { services: { deps, ingest, privacy, render, marketing }, calls };

128
apps/api/src/media-uploads.ts
View File

@ -4,6 +4,7 @@ import { mkdir, open, rename, rm, stat } from "node:fs/promises";
import path from "node:path"; import path from "node:path";
import type { MediaActor, MediaServices } from "@rakazo/adapters"; import type { MediaActor, MediaServices } from "@rakazo/adapters";
import type { MediaUpload } from "@rakazo/contracts"; import type { MediaUpload } from "@rakazo/contracts";
import type { MediaRepos } from "@rakazo/db";
import { MediaStateConflictError } from "@rakazo/db"; import { MediaStateConflictError } from "@rakazo/db";
import { resolveInside } from "@rakazo/media/node"; 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). * renamed, so a repeated or concurrent PUT of the same index is harmless).
* Completion concatenates the chunks in order into <spaceId>/uploads/<uploadId>/file * Completion concatenates the chunks in order into <spaceId>/uploads/<uploadId>/file
* while hashing, verifies size and SHA-256, and hands the file to ingest. * 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 { export class UploadChunkError extends Error {
constructor( constructor(
readonly status: 400 | 409 | 413, readonly status: 400 | 409 | 413,
@ -25,26 +35,15 @@ export class UploadChunkError extends Error {
} }
} }
interface UploadRow { type UploadRow = Awaited<ReturnType<MediaRepos["getUploadRecord"]>>;
id: string; type UploadSize = Pick<UploadRow, "sizeBytes" | "chunkBytes">;
spaceId: string;
assetId: string;
sizeBytes: bigint;
chunkBytes: number;
sha256: string;
fileName: string;
status: string;
}
export function uploadChunkCount(row: Pick<UploadRow, "sizeBytes" | "chunkBytes">): number { export function uploadChunkCount(row: UploadSize): number {
return Number((row.sizeBytes + BigInt(row.chunkBytes) - 1n) / BigInt(row.chunkBytes)); 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. */ /** Exact byte length chunk `index` must have: every chunk is full except the last. */
export function expectedChunkBytes( export function expectedChunkBytes(row: UploadSize, index: number): number {
row: Pick<UploadRow, "sizeBytes" | "chunkBytes">,
index: number,
): number {
const count = uploadChunkCount(row); const count = uploadChunkCount(row);
if (index < count - 1) return row.chunkBytes; if (index < count - 1) return row.chunkBytes;
return Number(row.sizeBytes - BigInt(row.chunkBytes) * BigInt(count - 1)); return Number(row.sizeBytes - BigInt(row.chunkBytes) * BigInt(count - 1));
@ -54,13 +53,58 @@ function uploadDir(row: Pick<UploadRow, "spaceId" | "id">): string {
return `${row.spaceId}/uploads/${row.id}`; return `${row.spaceId}/uploads/${row.id}`;
} }
async function chunksDir(mediaRoot: string, row: UploadRow): Promise<string> { async function mediaPath(mediaRoot: string, relative: string): Promise<string> {
await mkdir(mediaRoot, { recursive: true }); 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<string> {
const dir = await mediaPath(mediaRoot, `${uploadDir(row)}/chunks`);
await mkdir(dir, { recursive: true }); await mkdir(dir, { recursive: true });
return dir; return dir;
} }
async function removeUploadFiles(mediaRoot: string, row: UploadRow): Promise<void> {
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<boolean> {
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<MediaUpload> {
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. * Stream one chunk body to disk without buffering more than one network read.
* Rejects bodies longer or shorter than the chunk's exact size. * Rejects bodies longer or shorter than the chunk's exact size.
@ -72,9 +116,13 @@ export async function writeUploadChunk(input: {
index: number; index: number;
body: ReadableStream<Uint8Array> | null; body: ReadableStream<Uint8Array> | null;
declaredLength: number | null; declaredLength: number | null;
/** Aborting stops reading the body; the chunk is then discarded as incomplete. */
signal?: AbortSignal;
}): Promise<MediaUpload> { }): Promise<MediaUpload> {
const { repos, mediaRoot } = input.services.deps; const { services } = input;
const row = (await repos.getUploadRecord(input.spaceId, input.uploadId)) as UploadRow; 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"); if (row.status !== "open") throw new UploadChunkError(409, "Upload is not open");
const count = uploadChunkCount(row); const count = uploadChunkCount(row);
if (!Number.isInteger(input.index) || input.index < 0 || input.index >= count) { if (!Number.isInteger(input.index) || input.index < 0 || input.index >= count) {
@ -92,7 +140,10 @@ export async function writeUploadChunk(input: {
let written = 0; let written = 0;
let ok = false; let ok = false;
const reader = input.body.getReader(); const reader = input.body.getReader();
const stop = () => void reader.cancel().catch(() => undefined);
input.signal?.addEventListener("abort", stop, { once: true });
try { try {
if (input.signal?.aborted) stop();
for (;;) { for (;;) {
const { done, value } = await reader.read(); const { done, value } = await reader.read();
if (done) break; if (done) break;
@ -106,6 +157,7 @@ export async function writeUploadChunk(input: {
if (written !== expected) throw new UploadChunkError(400, "Chunk is incomplete"); if (written !== expected) throw new UploadChunkError(400, "Chunk is incomplete");
ok = true; ok = true;
} finally { } finally {
input.signal?.removeEventListener("abort", stop);
reader.releaseLock(); reader.releaseLock();
await handle.close(); await handle.close();
if (!ok) await rm(tempPath, { force: true }); if (!ok) await rm(tempPath, { force: true });
@ -126,11 +178,11 @@ export async function completeMediaUpload(
uploadId: string, uploadId: string,
): Promise<MediaUpload> { ): Promise<MediaUpload> {
const { repos, mediaRoot } = services.deps; 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 (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"); if (row.status !== "open") throw new MediaStateConflictError("Upload is not open");
const status = await repos.uploadStatus(actor.spaceId, uploadId); const received = new Set(row.receivedChunks);
const received = new Set(status.receivedChunks);
const count = uploadChunkCount(row); const count = uploadChunkCount(row);
for (let index = 0; index < count; index++) { for (let index = 0; index < count; index++) {
if (!received.has(index)) throw new MediaStateConflictError("Upload is missing chunks"); if (!received.has(index)) throw new MediaStateConflictError("Upload is missing chunks");
@ -140,8 +192,8 @@ export async function completeMediaUpload(
try { try {
const dir = await chunksDir(mediaRoot, row); const dir = await chunksDir(mediaRoot, row);
const relativeFile = `${uploadDir(row)}/file`; const relativeFile = `${uploadDir(row)}/file`;
const filePath = await resolveInside(mediaRoot, relativeFile); const filePath = await mediaPath(mediaRoot, relativeFile);
const tempFile = await resolveInside(mediaRoot, `${uploadDir(row)}/file.${randomUUID()}.tmp`); const tempFile = await mediaPath(mediaRoot, `${uploadDir(row)}/file.${randomUUID()}.tmp`);
const hash = createHash("sha256"); const hash = createHash("sha256");
let total = 0; let total = 0;
const out = await open(tempFile, "wx"); const out = await open(tempFile, "wx");
@ -152,10 +204,10 @@ export async function completeMediaUpload(
if (!info || info.size !== expectedChunkBytes(row, index)) { if (!info || info.size !== expectedChunkBytes(row, index)) {
throw new MediaStateConflictError("Upload is missing chunks"); throw new MediaStateConflictError("Upload is missing chunks");
} }
for await (const buffer of createReadStream(chunkPath)) { for await (const buffer of createReadStream(chunkPath) as AsyncIterable<Buffer>) {
hash.update(buffer as Buffer); hash.update(buffer);
total += (buffer as Buffer).byteLength; total += buffer.byteLength;
await out.write(buffer as Buffer); await out.write(buffer);
} }
} }
} catch (error) { } catch (error) {
@ -166,28 +218,26 @@ export async function completeMediaUpload(
await out.close(); await out.close();
const sha256 = hash.digest("hex"); const sha256 = hash.digest("hex");
if (sha256 !== row.sha256 || BigInt(total) !== row.sizeBytes) { 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); await repos.failUpload(actor.spaceId, uploadId);
throw new UploadChunkError(400, "Upload does not match its checksum"); throw new UploadChunkError(400, "Upload does not match its checksum");
} }
await rename(tempFile, filePath); 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, { await services.ingest.uploadCompleted(actor, {
assetId: row.assetId, assetId: row.assetId,
tempPath: filePath, tempPath: filePath,
sha256, sha256,
fileName: row.fileName, fileName: row.fileName,
}); });
// Ingest moves the file into the asset folder and records where; fall back // Ingest normally moves the file into the asset folder and closes the
// to the assembled file when it left the bytes in place. // upload itself; otherwise the assembled file stays as the original.
const after = (await repos.getAssetRecord(actor.spaceId, row.assetId)).storagePath; const after = await repos.getUploadRecord(actor.spaceId, uploadId);
const storagePath = after && after !== before ? after : relativeFile; if (after.status === "complete") {
const done = await repos.completeUpload(actor.spaceId, uploadId, { storagePath }); await removeUploadFiles(mediaRoot, row);
await rm(dir, { recursive: true, force: true }); return repos.uploadStatus(actor.spaceId, uploadId);
if (storagePath !== relativeFile) {
await rm(await resolveInside(mediaRoot, uploadDir(row)), { recursive: true, force: true });
} }
return done; return repos.completeUpload(actor.spaceId, uploadId, { storagePath: relativeFile });
} finally { } finally {
completing.delete(row.id); completing.delete(row.id);
} }

516
apps/api/src/media.postgres.test.ts Normal file
View File

@ -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<string, unknown>)[key];
return node as (input: unknown) => Promise<unknown>;
}
async function rpcCode(promise: Promise<unknown>): Promise<string> {
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<string, string>;
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<string> {
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<string> {
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<Uint8Array>({
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 });
}
});
});

331
apps/api/src/media.test.ts Normal file
View File

@ -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<typeof createRouterClient<ReturnType<typeof createMediaRouter>>>;
function client(router: ReturnType<typeof createMediaRouter>, 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<string, unknown>)[key];
return node as (input: unknown) => Promise<unknown>;
}
async function rpcCode(promise: Promise<unknown>): Promise<string> {
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<string, unknown>;
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<string, unknown>),
);
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<string, unknown>).code).toBe(
"NOT_FOUND",
);
expect(
(mediaRpcError(new PathOutsideRootError("../x")) as ORPCError<string, unknown>).code,
).toBe("BAD_REQUEST");
expect(
(mediaRpcError(new MediaStateConflictError("busy")) as ORPCError<string, unknown>).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");
});
});

66
apps/api/src/media.ts
View File

@ -4,7 +4,7 @@ import { fileURLToPath } from "node:url";
import { implement, ORPCError } from "@orpc/server"; import { implement, ORPCError } from "@orpc/server";
import { UnsupportedMediaOperationError } from "@rakazo/adapter-kit"; import { UnsupportedMediaOperationError } from "@rakazo/adapter-kit";
import type { MediaServices } from "@rakazo/adapters"; 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 { appContract, MediaInferenceOperationSchema } from "@rakazo/contracts";
import type { PrismaClient } from "@rakazo/db"; import type { PrismaClient } from "@rakazo/db";
import { import {
@ -18,7 +18,7 @@ import {
} from "@rakazo/db"; } from "@rakazo/db";
import { summarizeProfile } from "@rakazo/media"; import { summarizeProfile } from "@rakazo/media";
import { PathOutsideRootError, resolveInside } from "@rakazo/media/node"; 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 { export interface MediaRouterDeps {
prisma: PrismaClient; prisma: PrismaClient;
@ -122,6 +122,27 @@ export async function validateMediaFolder(
return resolved; 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<void> {
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<void> { async function removeMediaPath(mediaRoot: string, relative: string | null): Promise<void> {
if (!relative) return; if (!relative) return;
await rm(await resolveInside(mediaRoot, relative), { recursive: true, force: true }); await rm(await resolveInside(mediaRoot, relative), { recursive: true, force: true });
@ -246,20 +267,9 @@ export function createMediaRouter(deps: MediaRouterDeps) {
})), })),
}, },
uploads: { uploads: {
begin: media.uploads.begin.handler(async ({ context, input }) => { begin: media.uploads.begin.handler(({ context, input }) =>
const { repos } = context.services.deps; beginMediaUpload(context.services, context.actor, input),
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,
});
}),
status: media.uploads.status.handler(({ context, input }) => status: media.uploads.status.handler(({ context, input }) =>
context.services.deps.repos.uploadStatus(context.actor.spaceId, input.uploadId), 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. // The only consent write path: the signed-in member is the recorder.
recordConsent: media.people.recordConsent.handler(({ context, input }) => recordConsent: media.people.recordConsent.handler(async ({ context, input }) => {
context.services.deps.repos.recordConsent( 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 }, { spaceId: context.actor.spaceId, recordedByUserId: context.actor.userId },
input, input,
), );
), }),
remove: media.people.remove.handler(async ({ context, input }) => { remove: media.people.remove.handler(async ({ context, input }) => {
await context.services.deps.repos.getPerson(context.actor.spaceId, input.personId); await context.services.deps.repos.getPerson(context.actor.spaceId, input.personId);
await context.services.privacy.removePerson(context.actor, 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 }) => list: media.brandKits.list.handler(({ context }) =>
context.services.deps.repos.listBrandKits(context.actor.spaceId), context.services.deps.repos.listBrandKits(context.actor.spaceId),
), ),
upsert: media.brandKits.upsert.handler(({ context, input }) => upsert: media.brandKits.upsert.handler(async ({ context, input }) => {
context.services.deps.repos.upsertBrandKit(context.actor.spaceId, 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 }) => { remove: media.brandKits.remove.handler(async ({ context, input }) => {
await context.services.deps.repos.removeBrandKit(context.actor.spaceId, input.id); await context.services.deps.repos.removeBrandKit(context.actor.spaceId, input.id);
return { ok: true as const }; return { ok: true as const };