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

feat(media): space-scoped Media Studio repositories

Browse Source 
Sources, assets with keyset search, analyses, segments, uploads,
collections with re-clustering, people with consent and complete
biometric removal, marks, brand kits, projects with versioned plans,
renders with atomic claims, publications, metrics and the funnel.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj
This commit is contained in:
Claude 2026-10-03 17:55:49 +00:00
parent 7ccc96ac11
commit c1c39b6e46
No known key found for this signature in database
9 changed files with 3457 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
packages/db/src/index.ts
View File

@ -7,6 +7,7 @@ export * from "./events.js";
export * from "./external-conversations.js"; export * from "./external-conversations.js";
export * from "./groups.js"; export * from "./groups.js";
export * from "./main-agent.js"; export * from "./main-agent.js";
export * from "./media.js";
export * from "./memory-config.js"; export * from "./memory-config.js";
export * from "./messages.js"; export * from "./messages.js";
export * from "./messaging.js"; export * from "./messaging.js";

395
packages/db/src/media-collections.ts Normal file
View File

@ -0,0 +1,395 @@
import type { MediaCollection, MediaCollectionKind, MediaLabelSource } from "@rakazo/contracts";
import { roundCoordinate } from "@rakazo/media";
import type { MediaCollection as MediaCollectionRow, PrismaClient } from "./client.js";
import { Prisma } from "./client.js";
import type { CollectionStats, MediaDb, MediaPage, MediaPageInput } from "./media-shared.js";
import {
assertAssetsInSpace,
decodeKeysetCursor,
EMPTY_COLLECTION_STATS,
InvalidMediaInputError,
keysetAfterSql,
lockMediaKey,
mapMediaAsset,
mapMediaCollection,
pageLimit,
toPage,
} from "./media-shared.js";
import { IsolationError } from "./scope.js";
/** One session or theme cluster computed by the worker. */
export interface MediaClusterInput {
/** Stable key so a re-run updates the same collection. */
key: string;
title: string;
labelSource: Exclude<MediaLabelSource, "user">;
startAt: Date | null;
endAt: Date | null;
centroid: { lat: number; lon: number } | null;
coverAssetId: string | null;
/** In display order. */
assetIds: string[];
}
export interface ReplaceClustersResult {
created: number;
updated: number;
deleted: number;
}
async function collectionStats(db: MediaDb, ids: string[]): Promise<Map<string, CollectionStats>> {
if (ids.length === 0) return new Map();
const rows = await db.$queryRaw<
Array<{ id: string; assetCount: number; videoCount: number; totalDurationMs: bigint }>
>(Prisma.sql`
SELECT ci."collectionId" AS id,
count(*)::int AS "assetCount",
(count(*) FILTER (WHERE a.kind = 'video'))::int AS "videoCount",
coalesce(sum(a."durationMs"), 0)::bigint AS "totalDurationMs"
FROM media_collection_items ci
JOIN media_assets a ON a.id = ci."assetId"
WHERE ci."collectionId" IN (${Prisma.join(ids)})
GROUP BY ci."collectionId"
`);
return new Map(
rows.map((row) => [
row.id,
{
assetCount: row.assetCount,
videoCount: row.videoCount,
totalDurationMs: Number(row.totalDurationMs),
},
]),
);
}
async function collectionDto(db: MediaDb, row: MediaCollectionRow): Promise<MediaCollection> {
const stats = await collectionStats(db, [row.id]);
return mapMediaCollection(row, stats.get(row.id) ?? EMPTY_COLLECTION_STATS);
}
/** Assets ordered by capture time, undated last, with the span they cover. */
async function orderedAssets(db: MediaDb, spaceId: string, assetIds: string[]) {
const assets = await db.mediaAsset.findMany({
where: { spaceId, id: { in: [...new Set(assetIds)] } },
orderBy: [{ capturedAt: { sort: "asc", nulls: "last" } }, { id: "asc" }],
select: { id: true, capturedAt: true },
});
const dated = assets.flatMap((asset) => (asset.capturedAt ? [asset.capturedAt] : []));
return {
ids: assets.map((asset) => asset.id),
startAt: dated[0] ?? null,
endAt: dated.at(-1) ?? null,
};
}
async function writeItems(db: MediaDb, collectionId: string, assetIds: string[]): Promise<void> {
await db.mediaCollectionItem.deleteMany({ where: { collectionId } });
if (assetIds.length === 0) return;
await db.mediaCollectionItem.createMany({
data: [...new Set(assetIds)].map((assetId, position) => ({ collectionId, assetId, position })),
});
}
/** Sessions, themes and manual collections. */
export function createMediaCollectionRepos(prisma: PrismaClient) {
// -------------------------------------------------------------------------
// Collections
// -------------------------------------------------------------------------
async function requireCollection(db: MediaDb, spaceId: string, collectionId: string) {
const row = await db.mediaCollection.findFirst({ where: { id: collectionId, spaceId } });
if (!row) throw new IsolationError();
return row;
}
/**
* Apply a full clustering run for one kind. Collections are matched by cluster key;
* a title the user set survives, stale rule/model clusters are removed, and manual
* collections are never touched.
*/
async function replaceClusters(
spaceId: string,
kind: Exclude<MediaCollectionKind, "manual">,
clusters: MediaClusterInput[],
): Promise<ReplaceClustersResult> {
const keys = new Set<string>();
for (const cluster of clusters) {
if (keys.has(cluster.key)) throw new InvalidMediaInputError("Duplicate cluster key");
keys.add(cluster.key);
if (cluster.coverAssetId && !cluster.assetIds.includes(cluster.coverAssetId)) {
throw new InvalidMediaInputError("Cover asset is not in its cluster");
}
}
return prisma.$transaction(
async (tx) => {
await lockMediaKey(tx, `clusters:${spaceId}:${kind}`);
await assertAssetsInSpace(
tx,
spaceId,
clusters.flatMap((cluster) => cluster.assetIds),
);
const existing = await tx.mediaCollection.findMany({
where: { spaceId, kind, clusterKey: { not: null } },
});
const byKey = new Map(existing.map((row) => [row.clusterKey as string, row]));
const result: ReplaceClustersResult = { created: 0, updated: 0, deleted: 0 };
for (const cluster of clusters) {
const fields = {
startAt: cluster.startAt,
endAt: cluster.endAt,
centroidLat: cluster.centroid ? roundCoordinate(cluster.centroid.lat) : null,
centroidLon: cluster.centroid ? roundCoordinate(cluster.centroid.lon) : null,
coverAssetId: cluster.coverAssetId,
};
const current = byKey.get(cluster.key);
let collectionId: string;
if (current) {
const keepTitle = current.labelSource === "user";
await tx.mediaCollection.update({
where: { id: current.id },
data: {
...fields,
...(keepTitle ? {} : { title: cluster.title, labelSource: cluster.labelSource }),
},
});
collectionId = current.id;
result.updated += 1;
} else {
const created = await tx.mediaCollection.create({
data: {
...fields,
spaceId,
kind,
clusterKey: cluster.key,
title: cluster.title,
labelSource: cluster.labelSource,
},
select: { id: true },
});
collectionId = created.id;
result.created += 1;
}
await writeItems(tx, collectionId, cluster.assetIds);
}
const stale = existing.filter(
(row) => !keys.has(row.clusterKey as string) && row.labelSource !== "user",
);
if (stale.length > 0) {
const deleted = await tx.mediaCollection.deleteMany({
where: { spaceId, id: { in: stale.map((row) => row.id) } },
});
result.deleted = deleted.count;
}
return result;
},
{ timeout: 120_000 },
);
}
async function listCollections(
spaceId: string,
input: MediaPageInput & { kind?: MediaCollectionKind },
): Promise<MediaPage<MediaCollection>> {
const limit = pageLimit(input.limit);
const cursor = input.cursor ? decodeKeysetCursor(input.cursor) : null;
const keys = await prisma.$queryRaw<Array<{ id: string }>>(Prisma.sql`
SELECT c.id FROM media_collections c
WHERE c."spaceId" = ${spaceId}
${input.kind ? Prisma.sql`AND c.kind = ${input.kind}` : Prisma.empty}
AND ${keysetAfterSql(cursor, Prisma.sql`c."startAt"`, Prisma.sql`c.id`)}
ORDER BY c."startAt" DESC NULLS LAST, c.id DESC
LIMIT ${limit + 1}
`);
const ids = keys.map((key) => key.id);
const [rows, stats] = await Promise.all([
prisma.mediaCollection.findMany({ where: { spaceId, id: { in: ids } } }),
collectionStats(prisma, ids),
]);
const byId = new Map(rows.map((row) => [row.id, row]));
const ordered = ids.flatMap((id) => {
const row = byId.get(id);
return row ? [row] : [];
});
return toPage(
ordered,
limit,
(row) => ({ at: row.startAt, id: row.id }),
(row) => mapMediaCollection(row, stats.get(row.id) ?? EMPTY_COLLECTION_STATS),
);
}
async function getCollection(spaceId: string, collectionId: string) {
const row = await requireCollection(prisma, spaceId, collectionId);
const items = await prisma.mediaCollectionItem.findMany({
where: { collectionId, asset: { spaceId } },
orderBy: [{ position: "asc" }, { assetId: "asc" }],
include: { asset: true },
});
return {
collection: await collectionDto(prisma, row),
assets: items.map((item) => mapMediaAsset(item.asset)),
};
}
/** Asset ids of a collection in display order. */
async function collectionAssetIds(spaceId: string, collectionIds: string[]): Promise<string[]> {
const ids = [...new Set(collectionIds)];
if (ids.length === 0) return [];
const owned = await prisma.mediaCollection.count({ where: { spaceId, id: { in: ids } } });
if (owned !== ids.length) throw new IsolationError();
const items = await prisma.mediaCollectionItem.findMany({
where: { collectionId: { in: ids } },
orderBy: [{ position: "asc" }, { assetId: "asc" }],
select: { assetId: true, collectionId: true },
});
const order = new Map(ids.map((id, index) => [id, index]));
items.sort((a, b) => (order.get(a.collectionId) ?? 0) - (order.get(b.collectionId) ?? 0));
return [...new Set(items.map((item) => item.assetId))];
}
/** The user's title wins over every later clustering label. */
async function renameCollection(
spaceId: string,
collectionId: string,
title: string,
): Promise<MediaCollection> {
const result = await prisma.mediaCollection.updateMany({
where: { id: collectionId, spaceId },
data: { title, labelSource: "user" },
});
if (result.count !== 1) throw new IsolationError();
return collectionDto(prisma, await requireCollection(prisma, spaceId, collectionId));
}
/** Merge into the first collection; the others are deleted. */
async function mergeCollections(
spaceId: string,
collectionIds: string[],
): Promise<MediaCollection> {
const ids = [...new Set(collectionIds)];
if (ids.length < 2) throw new InvalidMediaInputError("Merge needs two collections");
return prisma.$transaction(async (tx) => {
const rows = await tx.mediaCollection.findMany({ where: { spaceId, id: { in: ids } } });
if (rows.length !== ids.length) throw new IsolationError();
const target = rows.find((row) => row.id === ids[0]) as MediaCollectionRow;
const items = await tx.mediaCollectionItem.findMany({
where: { collectionId: { in: ids } },
select: { assetId: true },
});
const assets = await orderedAssets(
tx,
spaceId,
items.map((item) => item.assetId),
);
await writeItems(tx, target.id, assets.ids);
await tx.mediaCollection.deleteMany({
where: { spaceId, id: { in: ids.filter((id) => id !== target.id) } },
});
const updated = await tx.mediaCollection.update({
where: { id: target.id },
data: {
startAt: assets.startAt,
endAt: assets.endAt,
labelSource: "user",
coverAssetId: target.coverAssetId ?? assets.ids[0] ?? null,
},
});
return collectionDto(tx, updated);
});
}
/** Split before `atAssetId`: it and every later asset move to a new collection. */
async function splitCollection(
spaceId: string,
collectionId: string,
atAssetId: string,
): Promise<MediaCollection[]> {
return prisma.$transaction(async (tx) => {
const row = await requireCollection(tx, spaceId, collectionId);
const items = await tx.mediaCollectionItem.findMany({
where: { collectionId },
orderBy: [{ position: "asc" }, { assetId: "asc" }],
select: { assetId: true },
});
const index = items.findIndex((item) => item.assetId === atAssetId);
if (index <= 0) throw new InvalidMediaInputError("Split asset must be inside the collection");
const first = await orderedAssets(
tx,
spaceId,
items.slice(0, index).map((item) => item.assetId),
);
const second = await orderedAssets(
tx,
spaceId,
items.slice(index).map((item) => item.assetId),
);
await writeItems(tx, collectionId, first.ids);
const kept = await tx.mediaCollection.update({
where: { id: collectionId },
data: {
startAt: first.startAt,
endAt: first.endAt,
labelSource: "user",
coverAssetId:
row.coverAssetId && first.ids.includes(row.coverAssetId)
? row.coverAssetId
: (first.ids[0] ?? null),
},
});
const created = await tx.mediaCollection.create({
data: {
spaceId,
kind: row.kind,
title: `${row.title} (2)`,
labelSource: "user",
startAt: second.startAt,
endAt: second.endAt,
centroidLat: row.centroidLat,
centroidLon: row.centroidLon,
coverAssetId:
row.coverAssetId && second.ids.includes(row.coverAssetId)
? row.coverAssetId
: (second.ids[0] ?? null),
},
});
await writeItems(tx, created.id, second.ids);
return [await collectionDto(tx, kept), await collectionDto(tx, created)];
});
}
async function createManualCollection(
spaceId: string,
title: string,
assetIds: string[],
): Promise<MediaCollection> {
return prisma.$transaction(async (tx) => {
await assertAssetsInSpace(tx, spaceId, assetIds);
const assets = await orderedAssets(tx, spaceId, assetIds);
const created = await tx.mediaCollection.create({
data: {
spaceId,
kind: "manual",
title,
labelSource: "user",
startAt: assets.startAt,
endAt: assets.endAt,
coverAssetId: assets.ids[0] ?? null,
},
});
await writeItems(tx, created.id, assets.ids);
return collectionDto(tx, created);
});
}
return {
replaceClusters,
listCollections,
getCollection,
collectionAssetIds,
renameCollection,
mergeCollections,
splitCollection,
createManualCollection,
};
}

756
packages/db/src/media-library.ts Normal file
View File

@ -0,0 +1,756 @@
import type { MediaSourceItem } from "@rakazo/adapter-kit";
import type {
MediaAnalysisKind,
MediaAsset,
MediaAssetQuery,
MediaFaceTrack,
MediaIngestState,
MediaSegment,
MediaSegmentKind,
MediaSource,
MediaSourceKind,
MediaSourceStatus,
TeaserFeature,
} from "@rakazo/contracts";
import { CLIENT_PUSHED_MEDIA_SOURCES } from "@rakazo/contracts";
import { roundCoordinate } from "@rakazo/media";
import type {
MediaAnalysis as MediaAnalysisRow,
MediaAsset as MediaAssetRow,
MediaSource as MediaSourceRow,
PrismaClient,
} from "./client.js";
import { Prisma } from "./client.js";
import type { MediaPage, MediaPageInput } from "./media-shared.js";
import {
decodeKeysetCursor,
InvalidMediaInputError,
keysetAfterSql,
likeContains,
lockMediaKey,
mapMediaAsset,
mapMediaFaceTrack,
mapMediaSegment,
mapMediaSource,
pageLimit,
toBytes,
toPage,
} from "./media-shared.js";
import { IsolationError } from "./scope.js";
export interface NewMediaSource {
kind: MediaSourceKind;
label: string;
ownerUserId: string;
/** Folder: { path }. Client-pushed: { deviceId }. Never secrets. */
config: Record<string, unknown>;
}
/** Source row for the sync worker, including its config and cursor. */
export interface MediaSourceRecord extends MediaSource {
spaceId: string;
ownerUserId: string;
config: Record<string, unknown>;
lastSyncCursor: string | null;
}
export interface MediaSourceSyncPatch {
status?: MediaSourceStatus;
lastSyncCursor?: string | null;
lastSyncedAt?: Date | null;
error?: string | null;
}
export interface UpsertSourceItemsResult {
created: number;
updated: number;
unchanged: number;
/** New assets and assets whose revision changed, in input order. */
changedAssetIds: string[];
}
export type MediaAssetSearch = Omit<MediaAssetQuery, "cursor" | "limit"> & MediaPageInput;
export interface MediaAssetDetail {
asset: MediaAsset;
segments: MediaSegment[];
faceTracks: MediaFaceTrack[];
transcript: string | null;
}
export interface MediaProxyResult {
contentHash: string;
proxyPath: string | null;
thumbPath: string | null;
spritePath?: string | null;
durationMs?: number | null;
width?: number | null;
height?: number | null;
fps?: number | null;
codec?: string | null;
orientation?: number | null;
}
export interface NewMediaAnalysis {
kind: MediaAnalysisKind;
modelId: string;
modelVersion: string;
data?: Prisma.InputJsonValue | null;
dataPath?: string | null;
/** Float32 little-endian vectors (see encodeVectors in @rakazo/media). */
vectors?: Uint8Array | null;
}
export interface NewMediaSegment {
kind: MediaSegmentKind;
startMs: number;
endMs: number;
scores?: Partial<Record<TeaserFeature, number>>;
teaserScore?: number | null;
transcriptExcerpt?: string | null;
labels?: string[];
paidOnly?: boolean;
}
export type MediaSpaceLookup = "source" | "asset" | "render" | "publication";
const EARTH_RADIUS_KM = 6371.0088;
const KM_PER_DEGREE = 111.32;
function sourceRecord(row: MediaSourceRow, assetCount: number): MediaSourceRecord {
const config =
row.config && typeof row.config === "object" && !Array.isArray(row.config)
? (row.config as Record<string, unknown>)
: {};
return {
...mapMediaSource(row, assetCount),
spaceId: row.spaceId,
ownerUserId: row.ownerUserId,
config,
lastSyncCursor: row.lastSyncCursor,
};
}
function rounded(value: number | null): number | null {
return value === null ? null : roundCoordinate(value);
}
function itemData(item: MediaSourceItem) {
return {
kind: item.kind,
revision: item.revision,
capturedAt: item.capturedAt,
durationMs: item.durationMs,
width: item.width,
height: item.height,
fps: item.fps,
gpsLat: rounded(item.gpsLat),
gpsLon: rounded(item.gpsLon),
deviceModel: item.deviceModel,
albumNames: item.albumNames,
isFavorite: item.isFavorite,
fileName: item.fileName,
sizeBytes: item.sizeBytes === null ? null : BigInt(item.sizeBytes),
};
}
/** Plain text of a stored transcript analysis (TranscribeResult words or a text field). */
function transcriptText(data: Prisma.JsonValue | null): string | null {
if (!data || typeof data !== "object" || Array.isArray(data)) return null;
if (typeof data.text === "string") return data.text;
if (!Array.isArray(data.words)) return null;
const words = data.words
.map((word) =>
word && typeof word === "object" && !Array.isArray(word) && typeof word.text === "string"
? word.text.trim()
: "",
)
.filter(Boolean);
return words.length > 0 ? words.join(" ") : null;
}
function nearSql(near: NonNullable<MediaAssetQuery["near"]>): Prisma.Sql {
const { lat, lon, radiusKm } = near;
const dLat = radiusKm / KM_PER_DEGREE;
const conditions: Prisma.Sql[] = [
Prisma.sql`a."gpsLat" IS NOT NULL AND a."gpsLon" IS NOT NULL`,
Prisma.sql`a."gpsLat" BETWEEN ${lat - dLat}::float8 AND ${lat + dLat}::float8`,
];
// Longitude prefilter only where it is a simple band; near the poles the haversine alone decides.
const maxAbsLat = Math.min(90, Math.abs(lat) + dLat);
const cos = Math.cos((maxAbsLat * Math.PI) / 180);
if (cos > 0.01) {
const dLon = radiusKm / (KM_PER_DEGREE * cos);
if (dLon < 180) {
const west = lon - dLon;
const east = lon + dLon;
if (west < -180) {
conditions.push(
Prisma.sql`(a."gpsLon" >= ${west + 360}::float8 OR a."gpsLon" <= ${east}::float8)`,
);
} else if (east > 180) {
conditions.push(
Prisma.sql`(a."gpsLon" >= ${west}::float8 OR a."gpsLon" <= ${east - 360}::float8)`,
);
} else {
conditions.push(Prisma.sql`a."gpsLon" BETWEEN ${west}::float8 AND ${east}::float8`);
}
}
}
conditions.push(Prisma.sql`
2 * ${EARTH_RADIUS_KM}::float8 * asin(least(1, sqrt(
power(sin(radians(a."gpsLat" - ${lat}::float8) / 2), 2)
+ cos(radians(${lat}::float8)) * cos(radians(a."gpsLat"))
* power(sin(radians(a."gpsLon" - ${lon}::float8) / 2), 2)
))) <= ${radiusKm}::float8`);
return Prisma.join(conditions, " AND ");
}
function textSql(spaceId: string, text: string): Prisma.Sql {
const pattern = likeContains(text);
return Prisma.sql`(
a."fileName" ILIKE ${pattern}
OR EXISTS (SELECT 1 FROM unnest(a."albumNames") AS album(name) WHERE album.name ILIKE ${pattern})
OR EXISTS (
SELECT 1 FROM media_collection_items ci
JOIN media_collections c ON c.id = ci."collectionId"
WHERE ci."assetId" = a.id AND c."spaceId" = ${spaceId} AND c.title ILIKE ${pattern}
)
OR EXISTS (
SELECT 1 FROM media_segments s
WHERE s."assetId" = a.id AND s."spaceId" = ${spaceId}
AND (
s."transcriptExcerpt" ILIKE ${pattern}
OR EXISTS (SELECT 1 FROM unnest(s.labels) AS label(name) WHERE label.name ILIKE ${pattern})
)
)
)`;
}
export function createMediaLibraryRepos(prisma: PrismaClient) {
async function requireAsset(
db: Prisma.TransactionClient | PrismaClient,
spaceId: string,
assetId: string,
): Promise<MediaAssetRow> {
const asset = await db.mediaAsset.findFirst({ where: { id: assetId, spaceId } });
if (!asset) throw new IsolationError();
return asset;
}
async function updateAsset(
spaceId: string,
assetId: string,
data: Prisma.MediaAssetUpdateManyMutationInput,
): Promise<MediaAsset> {
const result = await prisma.mediaAsset.updateMany({ where: { id: assetId, spaceId }, data });
if (result.count !== 1) throw new IsolationError();
return mapMediaAsset(await requireAsset(prisma, spaceId, assetId));
}
// -------------------------------------------------------------------------
// Sources
// -------------------------------------------------------------------------
async function createSource(spaceId: string, input: NewMediaSource): Promise<MediaSource> {
const row = await prisma.mediaSource.create({
data: {
spaceId,
kind: input.kind,
label: input.label,
ownerUserId: input.ownerUserId,
config: input.config as Prisma.InputJsonObject,
},
});
return mapMediaSource(row, 0);
}
async function listSources(spaceId: string): Promise<MediaSource[]> {
const rows = await prisma.mediaSource.findMany({
where: { spaceId },
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
include: { _count: { select: { assets: true } } },
});
return rows.map((row) => mapMediaSource(row, row._count.assets));
}
async function getSource(spaceId: string, sourceId: string): Promise<MediaSourceRecord> {
const row = await prisma.mediaSource.findFirst({
where: { id: sourceId, spaceId },
include: { _count: { select: { assets: true } } },
});
if (!row) throw new IsolationError();
return sourceRecord(row, row._count.assets);
}
async function updateSourceSync(
spaceId: string,
sourceId: string,
patch: MediaSourceSyncPatch,
): Promise<MediaSourceRecord> {
const result = await prisma.mediaSource.updateMany({
where: { id: sourceId, spaceId },
data: {
...(patch.status !== undefined ? { status: patch.status } : {}),
...(patch.lastSyncCursor !== undefined ? { lastSyncCursor: patch.lastSyncCursor } : {}),
...(patch.lastSyncedAt !== undefined ? { lastSyncedAt: patch.lastSyncedAt } : {}),
...(patch.error !== undefined ? { error: patch.error } : {}),
},
});
if (result.count !== 1) throw new IsolationError();
return getSource(spaceId, sourceId);
}
/** Deletes the source and, by cascade, its assets and everything derived from them. */
async function removeSource(spaceId: string, sourceId: string): Promise<void> {
const result = await prisma.mediaSource.deleteMany({ where: { id: sourceId, spaceId } });
if (result.count !== 1) throw new IsolationError();
}
/**
* Worker entry point: jobs carry only a record id. Returns the owning space so every
* further call stays scoped, or null when the record no longer exists.
*/
async function mediaSpaceOf(kind: MediaSpaceLookup, id: string): Promise<string | null> {
const select = { spaceId: true } as const;
const row =
kind === "source"
? await prisma.mediaSource.findUnique({ where: { id }, select })
: kind === "asset"
? await prisma.mediaAsset.findUnique({ where: { id }, select })
: kind === "render"
? await prisma.mediaRender.findUnique({ where: { id }, select })
: await prisma.mediaPublication.findUnique({ where: { id }, select });
return row?.spaceId ?? null;
}
// -------------------------------------------------------------------------
// Assets
// -------------------------------------------------------------------------
/**
* Index one page of source items keyed by (sourceId, externalId). An item whose
* revision matches the stored one is unchanged; syncs of the same source are serialized.
*/
async function upsertSourceItems(
spaceId: string,
sourceId: string,
items: MediaSourceItem[],
): Promise<UpsertSourceItemsResult> {
const unique = new Map<string, MediaSourceItem>();
for (const item of items) unique.set(item.externalId, item);
return prisma.$transaction(
async (tx) => {
const source = await tx.mediaSource.findFirst({
where: { id: sourceId, spaceId },
select: { kind: true },
});
if (!source) throw new IsolationError();
await lockMediaKey(tx, `source:${sourceId}`);
const storage = (CLIENT_PUSHED_MEDIA_SOURCES as readonly string[]).includes(source.kind)
? "remote"
: "referenced";
const existing = await tx.mediaAsset.findMany({
where: { sourceId, externalId: { in: [...unique.keys()] } },
select: { id: true, externalId: true, revision: true },
});
const byExternalId = new Map(existing.map((row) => [row.externalId, row]));
const toCreate: MediaSourceItem[] = [];
const changed = new Map<string, string>();
let unchanged = 0;
for (const item of unique.values()) {
const current = byExternalId.get(item.externalId);
if (!current) {
toCreate.push(item);
} else if (current.revision === item.revision) {
unchanged += 1;
} else {
await tx.mediaAsset.update({ where: { id: current.id }, data: itemData(item) });
changed.set(item.externalId, current.id);
}
}
const updated = changed.size;
if (toCreate.length > 0) {
const created = await tx.mediaAsset.createManyAndReturn({
data: toCreate.map((item) => ({
...itemData(item),
spaceId,
sourceId,
externalId: item.externalId,
storage,
ingestState: storage === "remote" ? "awaiting_upload" : "indexed",
})),
select: { id: true, externalId: true },
});
for (const row of created) changed.set(row.externalId, row.id);
}
return {
created: toCreate.length,
updated,
unchanged,
changedAssetIds: [...unique.keys()].flatMap((externalId) => {
const id = changed.get(externalId);
return id ? [id] : [];
}),
};
},
{ timeout: 60_000 },
);
}
/**
* Keyset-paginated search, newest capture first, undated assets last. The cursor is
* the (capturedAt, id) of the previous page's last row, so inserts never shift pages.
*/
async function searchAssets(
spaceId: string,
query: MediaAssetSearch,
): Promise<MediaPage<MediaAsset>> {
const limit = pageLimit(query.limit);
const cursor = query.cursor ? decodeKeysetCursor(query.cursor) : null;
const conditions: Prisma.Sql[] = [Prisma.sql`a."spaceId" = ${spaceId}`];
if (query.sourceId) conditions.push(Prisma.sql`a."sourceId" = ${query.sourceId}`);
if (query.kinds && query.kinds.length > 0) {
conditions.push(Prisma.sql`a.kind IN (${Prisma.join(query.kinds)})`);
}
if (query.from) conditions.push(Prisma.sql`a."capturedAt" >= ${new Date(query.from)}`);
if (query.to) conditions.push(Prisma.sql`a."capturedAt" <= ${new Date(query.to)}`);
if (query.favoritesOnly) conditions.push(Prisma.sql`a."isFavorite" = TRUE`);
if (query.minQuality !== undefined) {
conditions.push(Prisma.sql`a."qualityScore" >= ${query.minQuality}::float8`);
}
if (query.near) conditions.push(nearSql(query.near));
if (query.personId) {
conditions.push(Prisma.sql`EXISTS (
SELECT 1 FROM media_face_tracks ft
WHERE ft."assetId" = a.id AND ft."spaceId" = ${spaceId} AND ft."personId" = ${query.personId}
)`);
}
if (query.collectionId) {
conditions.push(Prisma.sql`EXISTS (
SELECT 1 FROM media_collection_items ci
JOIN media_collections c ON c.id = ci."collectionId"
WHERE ci."assetId" = a.id AND c.id = ${query.collectionId} AND c."spaceId" = ${spaceId}
)`);
}
const text = query.text?.trim();
if (text) conditions.push(textSql(spaceId, text));
conditions.push(keysetAfterSql(cursor, Prisma.sql`a."capturedAt"`, Prisma.sql`a.id`));
const keys = await prisma.$queryRaw<Array<{ id: string }>>(Prisma.sql`
SELECT a.id FROM media_assets a
WHERE ${Prisma.join(conditions, " AND ")}
ORDER BY a."capturedAt" DESC NULLS LAST, a.id DESC
LIMIT ${limit + 1}
`);
const rows = await prisma.mediaAsset.findMany({
where: { spaceId, id: { in: keys.map((key) => key.id) } },
});
const byId = new Map(rows.map((row) => [row.id, row]));
const ordered = keys.flatMap((key) => {
const row = byId.get(key.id);
return row ? [row] : [];
});
return toPage(ordered, limit, (row) => ({ at: row.capturedAt, id: row.id }), mapMediaAsset);
}
async function getAsset(spaceId: string, assetId: string): Promise<MediaAssetDetail> {
const asset = await requireAsset(prisma, spaceId, assetId);
const [segments, faceTracks, transcript] = await Promise.all([
prisma.mediaSegment.findMany({
where: { assetId, spaceId },
orderBy: [{ startMs: "asc" }, { id: "asc" }],
}),
prisma.mediaFaceTrack.findMany({
where: { assetId, spaceId },
orderBy: [{ startMs: "asc" }, { id: "asc" }],
}),
prisma.mediaAnalysis.findFirst({
where: { assetId, spaceId, kind: "transcript" },
orderBy: [{ createdAt: "desc" }, { id: "desc" }],
select: { data: true },
}),
]);
return {
asset: mapMediaAsset(asset),
segments: segments.map(mapMediaSegment),
faceTracks: faceTracks.map(mapMediaFaceTrack),
transcript: transcriptText(transcript?.data ?? null),
};
}
/** Full row for workers: storage path, hashes, derived file paths. */
async function getAssetRecord(spaceId: string, assetId: string): Promise<MediaAssetRow> {
return requireAsset(prisma, spaceId, assetId);
}
async function listAssetRecords(spaceId: string, assetIds: string[]): Promise<MediaAssetRow[]> {
if (assetIds.length === 0) return [];
return prisma.mediaAsset.findMany({
where: { spaceId, id: { in: [...new Set(assetIds)] } },
orderBy: [{ capturedAt: { sort: "asc", nulls: "last" } }, { id: "asc" }],
});
}
/** Every asset of the space in capture order, with the fields clustering reads. */
async function listAssetsForClustering(spaceId: string) {
return prisma.mediaAsset.findMany({
where: { spaceId },
orderBy: [{ capturedAt: { sort: "asc", nulls: "last" } }, { id: "asc" }],
select: {
id: true,
kind: true,
capturedAt: true,
durationMs: true,
gpsLat: true,
gpsLon: true,
deviceModel: true,
albumNames: true,
contentHash: true,
},
});
}
/**
* Atomic ingest state change: only applies when the asset is currently in one of
* `from`, so two workers cannot both take the same step. Returns whether it applied.
*/
async function transitionIngestState(
spaceId: string,
assetId: string,
transition: { from: MediaIngestState[]; to: MediaIngestState; error?: string | null },
): Promise<boolean> {
const result = await prisma.mediaAsset.updateMany({
where: { id: assetId, spaceId, ingestState: { in: transition.from } },
data: {
ingestState: transition.to,
ingestError: transition.to === "failed" ? (transition.error ?? null) : null,
},
});
if (result.count === 1) return true;
await requireAsset(prisma, spaceId, assetId);
return false;
}
async function recordProxy(
spaceId: string,
assetId: string,
proxy: MediaProxyResult,
): Promise<MediaAsset> {
return updateAsset(spaceId, assetId, {
contentHash: proxy.contentHash,
proxyPath: proxy.proxyPath,
thumbPath: proxy.thumbPath,
...(proxy.spritePath !== undefined ? { spritePath: proxy.spritePath } : {}),
...(proxy.durationMs !== undefined ? { durationMs: proxy.durationMs } : {}),
...(proxy.width !== undefined ? { width: proxy.width } : {}),
...(proxy.height !== undefined ? { height: proxy.height } : {}),
...(proxy.fps !== undefined ? { fps: proxy.fps } : {}),
...(proxy.codec !== undefined ? { codec: proxy.codec } : {}),
...(proxy.orientation !== undefined ? { orientation: proxy.orientation } : {}),
ingestState: "proxied",
ingestError: null,
});
}
async function recordAnalysisVersion(
spaceId: string,
assetId: string,
analysisVersion: number,
qualityScore?: number | null,
): Promise<MediaAsset> {
return updateAsset(spaceId, assetId, {
analysisVersion,
...(qualityScore !== undefined ? { qualityScore } : {}),
ingestState: "analyzed",
ingestError: null,
});
}
/** Remote assets move to awaiting_upload; returns the ids that now wait for bytes. */
async function requestUpload(spaceId: string, assetIds: string[]): Promise<string[]> {
const ids = [...new Set(assetIds)];
const owned = await prisma.mediaAsset.count({ where: { spaceId, id: { in: ids } } });
if (owned !== ids.length) throw new IsolationError();
await prisma.mediaAsset.updateMany({
where: {
spaceId,
id: { in: ids },
storage: "remote",
ingestState: { in: ["indexed", "failed"] },
},
data: { ingestState: "awaiting_upload", ingestError: null },
});
const awaiting = await prisma.mediaAsset.findMany({
where: { spaceId, id: { in: ids }, ingestState: "awaiting_upload" },
select: { id: true },
});
const set = new Set(awaiting.map((row) => row.id));
return ids.filter((id) => set.has(id));
}
async function assetsNeedingWork(
spaceId: string,
state: MediaIngestState,
limit: number,
): Promise<MediaAssetRow[]> {
return prisma.mediaAsset.findMany({
where: { spaceId, ingestState: state },
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
take: pageLimit(limit),
});
}
// -------------------------------------------------------------------------
// Analyses
// -------------------------------------------------------------------------
/** Insert or replace the result for (asset, kind, modelVersion); other versions stay. */
async function upsertAnalysis(
spaceId: string,
assetId: string,
analysis: NewMediaAnalysis,
): Promise<MediaAnalysisRow> {
await requireAsset(prisma, spaceId, assetId);
const data = {
modelId: analysis.modelId,
data: analysis.data ?? Prisma.DbNull,
dataPath: analysis.dataPath ?? null,
vectors: analysis.vectors ? toBytes(analysis.vectors) : null,
};
return prisma.mediaAnalysis.upsert({
where: {
assetId_kind_modelVersion: {
assetId,
kind: analysis.kind,
modelVersion: analysis.modelVersion,
},
},
create: {
...data,
spaceId,
assetId,
kind: analysis.kind,
modelVersion: analysis.modelVersion,
},
update: data,
});
}
async function latestAnalysis(
spaceId: string,
assetId: string,
kind: MediaAnalysisKind,
): Promise<MediaAnalysisRow | null> {
await requireAsset(prisma, spaceId, assetId);
return prisma.mediaAnalysis.findFirst({
where: { spaceId, assetId, kind },
orderBy: [{ createdAt: "desc" }, { id: "desc" }],
});
}
/** Latest analysis of one kind per asset across the space (e.g. embeddings for clustering). */
async function latestAnalysesByKind(
spaceId: string,
kind: MediaAnalysisKind,
): Promise<Map<string, MediaAnalysisRow>> {
const rows = await prisma.mediaAnalysis.findMany({
where: { spaceId, kind },
orderBy: [{ assetId: "asc" }, { createdAt: "desc" }, { id: "desc" }],
});
const latest = new Map<string, MediaAnalysisRow>();
for (const row of rows) if (!latest.has(row.assetId)) latest.set(row.assetId, row);
return latest;
}
// -------------------------------------------------------------------------
// Segments
// -------------------------------------------------------------------------
/** Replace every segment of the asset with the result of one analysis version. */
async function replaceSegments(
spaceId: string,
assetId: string,
analysisVersion: number,
segments: NewMediaSegment[],
): Promise<number> {
for (const segment of segments) {
if (segment.endMs <= segment.startMs || segment.startMs < 0) {
throw new InvalidMediaInputError("Segment range is empty");
}
}
return prisma.$transaction(async (tx) => {
await requireAsset(tx, spaceId, assetId);
await lockMediaKey(tx, `segments:${assetId}`);
await tx.mediaSegment.deleteMany({ where: { assetId, spaceId } });
const created = await tx.mediaSegment.createMany({
data: segments.map((segment) => ({
spaceId,
assetId,
analysisVersion,
kind: segment.kind,
startMs: segment.startMs,
endMs: segment.endMs,
scores: segment.scores ?? {},
teaserScore: segment.teaserScore ?? null,
transcriptExcerpt: segment.transcriptExcerpt ?? null,
labels: segment.labels ?? [],
paidOnly: segment.paidOnly ?? false,
})),
});
return created.count;
});
}
async function listSegments(spaceId: string, assetIds: string[]): Promise<MediaSegment[]> {
if (assetIds.length === 0) return [];
const rows = await prisma.mediaSegment.findMany({
where: { spaceId, assetId: { in: [...new Set(assetIds)] } },
orderBy: [{ assetId: "asc" }, { startMs: "asc" }, { id: "asc" }],
});
return rows.map(mapMediaSegment);
}
/** All or nothing: any id outside the space fails the whole call. */
async function setSegmentsPaidOnly(
spaceId: string,
segmentIds: string[],
paidOnly: boolean,
): Promise<number> {
const ids = [...new Set(segmentIds)];
return prisma.$transaction(async (tx) => {
const result = await tx.mediaSegment.updateMany({
where: { spaceId, id: { in: ids } },
data: { paidOnly },
});
if (result.count !== ids.length) throw new IsolationError();
return result.count;
});
}
return {
createSource,
listSources,
getSource,
updateSourceSync,
removeSource,
mediaSpaceOf,
upsertSourceItems,
searchAssets,
getAsset,
getAssetRecord,
listAssetRecords,
listAssetsForClustering,
transitionIngestState,
recordProxy,
recordAnalysisVersion,
requestUpload,
assetsNeedingWork,
upsertAnalysis,
latestAnalysis,
latestAnalysesByKind,
replaceSegments,
listSegments,
setSegmentsPaidOnly,
};
}

598
packages/db/src/media-people.ts Normal file
View File

@ -0,0 +1,598 @@
import type {
AnonymizeMode,
ConsentInput,
MarkTreatment,
MediaFaceTrack,
MediaMark,
MediaMarkKind,
MediaMarkPrompt,
MediaPerson,
} from "@rakazo/contracts";
import type { MediaMark as MediaMarkRow, PrismaClient } from "./client.js";
import type { MediaDb } from "./media-shared.js";
import {
assertAssetsInSpace,
InvalidMediaInputError,
lockMediaKey,
mapMediaFaceTrack,
mapMediaMark,
mapMediaPerson,
toBytes,
} from "./media-shared.js";
import { IsolationError } from "./scope.js";
export interface MediaPersonSuggestion {
/** Float32 mean identity embedding. Biometric: stays in this database. */
centroid: Uint8Array;
trackIds: string[];
}
export interface MediaPersonPatch {
displayName?: string;
isSelf?: boolean;
confirmed?: boolean;
anonymizeByDefault?: AnonymizeMode;
syntheticSeed?: number | null;
}
/**
* Who records consent. A user id is mandatory by type and checked against the space,
* so a caller without an authenticated user (an agent tool) cannot produce one.
*/
export interface ConsentRecorder {
spaceId: string;
recordedByUserId: string;
}
export interface RemovedPerson {
deletedFaceTracks: number;
deletedMarks: number;
/** Files the caller must delete too: box keyframes and mark masks. */
keyframesPaths: string[];
maskDirs: string[];
}
export interface NewMediaFaceTrack {
startMs: number;
endMs: number;
confidenceMin: number;
keyframesPath: string;
/** Float32 identity embedding. Biometric: stays in this database. */
embedding: Uint8Array | null;
personId?: string | null;
}
export interface MediaFaceTrackRecord extends MediaFaceTrack {
keyframesPath: string;
modelVersion: string;
}
export interface MediaFaceEmbedding {
trackId: string;
assetId: string;
personId: string | null;
embedding: Uint8Array;
}
export interface NewMediaMark {
kind: MediaMarkKind;
label: string | null;
personId: string | null;
assetId: string;
prompt: MediaMarkPrompt;
treatment: MarkTreatment;
}
export interface MediaMarkPatch {
treatment?: MarkTreatment;
label?: string | null;
personId?: string | null;
}
export interface MediaMarkResult {
qualityScore: number | null;
suggestedFallback: MarkTreatment | null;
maskDir: string | null;
}
export interface MediaMarkRecord extends MediaMark {
maskDir: string | null;
}
function markRecord(row: MediaMarkRow): MediaMarkRecord {
return { ...mapMediaMark(row), maskDir: row.maskDir };
}
/** People, face tracks and identifying marks. Biometric data stays in this database. */
export function createMediaPeopleRepos(prisma: PrismaClient) {
// -------------------------------------------------------------------------
// People
// -------------------------------------------------------------------------
const personInclude = {
_count: { select: { faceTracks: true } },
faceTracks: {
orderBy: [{ confidenceMin: "desc" as const }, { id: "asc" as const }],
take: 1,
select: { assetId: true },
},
};
async function personDto(db: MediaDb, spaceId: string, personId: string): Promise<MediaPerson> {
const row = await db.mediaPerson.findFirst({
where: { id: personId, spaceId },
include: personInclude,
});
if (!row) throw new IsolationError();
return mapMediaPerson(row, row._count.faceTracks, row.faceTracks[0]?.assetId ?? null);
}
async function listPeople(spaceId: string): Promise<MediaPerson[]> {
const rows = await prisma.mediaPerson.findMany({
where: { spaceId },
orderBy: [{ isSelf: "desc" }, { createdAt: "asc" }, { id: "asc" }],
include: personInclude,
});
return rows.map((row) =>
mapMediaPerson(row, row._count.faceTracks, row.faceTracks[0]?.assetId ?? null),
);
}
async function getPerson(spaceId: string, personId: string): Promise<MediaPerson> {
return personDto(prisma, spaceId, personId);
}
/** People with consent fields and seed, for the render gate. */
async function listPersonRecords(spaceId: string, personIds?: string[]) {
return prisma.mediaPerson.findMany({
where: { spaceId, ...(personIds ? { id: { in: personIds } } : {}) },
omit: { centroid: true },
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
});
}
/** Identity centroids for matching new face tracks to known people. Local only. */
async function personCentroids(
spaceId: string,
): Promise<Array<{ personId: string; centroid: Uint8Array }>> {
const rows = await prisma.mediaPerson.findMany({
where: { spaceId, centroid: { not: null } },
select: { id: true, centroid: true },
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
});
return rows.map((row) => ({ personId: row.id, centroid: row.centroid as Uint8Array }));
}
/**
* Turn face clusters into unconfirmed people ("Person N"). Only tracks that have no
* person yet are assigned, so a user's earlier assignment is never overwritten.
*/
async function createPersonSuggestions(
spaceId: string,
suggestions: MediaPersonSuggestion[],
): Promise<MediaPerson[]> {
if (suggestions.length === 0) return [];
const ids = await prisma.$transaction(async (tx) => {
await lockMediaKey(tx, `people:${spaceId}`);
const trackIds = [...new Set(suggestions.flatMap((suggestion) => suggestion.trackIds))];
const owned = await tx.mediaFaceTrack.count({ where: { spaceId, id: { in: trackIds } } });
if (owned !== trackIds.length) throw new IsolationError();
let number = await tx.mediaPerson.count({ where: { spaceId } });
const created: string[] = [];
for (const suggestion of suggestions) {
number += 1;
const person = await tx.mediaPerson.create({
data: {
spaceId,
displayName: `Person ${number}`,
centroid: toBytes(suggestion.centroid),
},
select: { id: true },
});
await tx.mediaFaceTrack.updateMany({
where: { spaceId, id: { in: suggestion.trackIds }, personId: null },
data: { personId: person.id },
});
created.push(person.id);
}
return created;
});
return Promise.all(ids.map((id) => personDto(prisma, spaceId, id)));
}
async function updatePerson(
spaceId: string,
personId: string,
patch: MediaPersonPatch,
): Promise<MediaPerson> {
await prisma.$transaction(async (tx) => {
const result = await tx.mediaPerson.updateMany({
where: { id: personId, spaceId },
data: {
...(patch.displayName !== undefined ? { displayName: patch.displayName } : {}),
...(patch.isSelf !== undefined ? { isSelf: patch.isSelf } : {}),
...(patch.confirmed !== undefined ? { confirmed: patch.confirmed } : {}),
...(patch.anonymizeByDefault !== undefined
? { anonymizeByDefault: patch.anonymizeByDefault }
: {}),
...(patch.syntheticSeed !== undefined ? { syntheticSeed: patch.syntheticSeed } : {}),
},
});
if (result.count !== 1) throw new IsolationError();
// One person per space is the user themself.
if (patch.isSelf) {
await tx.mediaPerson.updateMany({
where: { spaceId, isSelf: true, id: { not: personId } },
data: { isSelf: false },
});
}
});
return personDto(prisma, spaceId, personId);
}
/**
* Merge into the first person: tracks and marks move, the others are deleted along
* with their centroids. The target keeps its own consent record.
*/
async function mergePeople(spaceId: string, personIds: string[]): Promise<MediaPerson> {
const ids = [...new Set(personIds)];
if (ids.length < 2) throw new InvalidMediaInputError("Merge needs two people");
const [targetId, ...others] = ids as [string, ...string[]];
await prisma.$transaction(async (tx) => {
const owned = await tx.mediaPerson.count({ where: { spaceId, id: { in: ids } } });
if (owned !== ids.length) throw new IsolationError();
await tx.mediaFaceTrack.updateMany({
where: { spaceId, personId: { in: others } },
data: { personId: targetId },
});
await tx.mediaMark.updateMany({
where: { spaceId, personId: { in: others } },
data: { personId: targetId },
});
await tx.mediaPerson.updateMany({
where: { spaceId, id: { in: others } },
data: { centroid: null },
});
await tx.mediaPerson.deleteMany({ where: { spaceId, id: { in: others } } });
await tx.mediaPerson.update({ where: { id: targetId }, data: { confirmed: true } });
});
return personDto(prisma, spaceId, targetId);
}
async function assignTrack(
spaceId: string,
trackId: string,
personId: string | null,
): Promise<MediaFaceTrack> {
return prisma.$transaction(async (tx) => {
if (personId) {
const person = await tx.mediaPerson.findFirst({
where: { id: personId, spaceId },
select: { id: true },
});
if (!person) throw new IsolationError();
}
const result = await tx.mediaFaceTrack.updateMany({
where: { id: trackId, spaceId },
data: { personId },
});
if (result.count !== 1) throw new IsolationError();
const row = await tx.mediaFaceTrack.findUniqueOrThrow({ where: { id: trackId } });
return mapMediaFaceTrack(row);
});
}
/**
* Record consent and age verification for a person. Only a signed-in member of the
* space can do this; the recorder is stored with the record.
*/
async function recordConsent(
recorder: ConsentRecorder,
input: ConsentInput,
): Promise<MediaPerson> {
const { spaceId, recordedByUserId } = recorder;
if (typeof recordedByUserId !== "string" || recordedByUserId.trim() === "") {
throw new InvalidMediaInputError("Consent needs the recording user");
}
if (input.ageVerificationMethod !== "none" && !input.ageVerifiedAt) {
throw new InvalidMediaInputError("Age verification needs a date");
}
const member = await prisma.spaceMember.findFirst({
where: { spaceId, userId: recordedByUserId },
select: { id: true },
});
if (!member) throw new IsolationError();
const result = await prisma.mediaPerson.updateMany({
where: { id: input.personId, spaceId },
data: {
consentArtifactId: input.consentArtifactId,
consentRecordedAt: new Date(input.consentRecordedAt),
consentRecordedBy: recordedByUserId,
ageVerificationMethod: input.ageVerificationMethod,
ageVerifiedAt:
input.ageVerificationMethod === "none" || !input.ageVerifiedAt
? null
: new Date(input.ageVerifiedAt),
},
});
if (result.count !== 1) throw new IsolationError();
return personDto(prisma, spaceId, input.personId);
}
/**
* Delete a person and every biometric record tied to them: the centroid, the
* embeddings of their face tracks, the tracks themselves, and person-bound marks.
* Embeddings are blanked before the rows go so no copy survives in the same
* transaction's row versions; the caller deletes the returned files.
*/
async function removePersonCompletely(spaceId: string, personId: string): Promise<RemovedPerson> {
return prisma.$transaction(async (tx) => {
const person = await tx.mediaPerson.findFirst({
where: { id: personId, spaceId },
select: { id: true },
});
if (!person) throw new IsolationError();
const tracks = await tx.mediaFaceTrack.findMany({
where: { spaceId, personId },
select: { keyframesPath: true },
});
const marks = await tx.mediaMark.findMany({
where: { spaceId, personId },
select: { maskDir: true },
});
await tx.mediaFaceTrack.updateMany({
where: { spaceId, personId },
data: { embedding: null },
});
const deletedTracks = await tx.mediaFaceTrack.deleteMany({ where: { spaceId, personId } });
const deletedMarks = await tx.mediaMark.deleteMany({ where: { spaceId, personId } });
await tx.mediaPerson.update({ where: { id: personId }, data: { centroid: null } });
await tx.mediaPerson.delete({ where: { id: personId } });
return {
deletedFaceTracks: deletedTracks.count,
deletedMarks: deletedMarks.count,
keyframesPaths: tracks.map((track) => track.keyframesPath),
maskDirs: marks.flatMap((mark) => (mark.maskDir ? [mark.maskDir] : [])),
};
});
}
// -------------------------------------------------------------------------
// Face tracks
// -------------------------------------------------------------------------
/** Replace the face tracks of one asset with a fresh tracking run. */
async function replaceFaceTracks(
spaceId: string,
assetId: string,
modelVersion: string,
tracks: NewMediaFaceTrack[],
): Promise<MediaFaceTrack[]> {
return prisma.$transaction(async (tx) => {
const asset = await tx.mediaAsset.findFirst({
where: { id: assetId, spaceId },
select: { id: true },
});
if (!asset) throw new IsolationError();
const personIds = [
...new Set(tracks.flatMap((track) => (track.personId ? [track.personId] : []))),
];
if (personIds.length > 0) {
const owned = await tx.mediaPerson.count({ where: { spaceId, id: { in: personIds } } });
if (owned !== personIds.length) throw new IsolationError();
}
await lockMediaKey(tx, `faces:${assetId}`);
await tx.mediaFaceTrack.updateMany({
where: { spaceId, assetId },
data: { embedding: null },
});
await tx.mediaFaceTrack.deleteMany({ where: { spaceId, assetId } });
const rows = await tx.mediaFaceTrack.createManyAndReturn({
data: tracks.map((track) => ({
spaceId,
assetId,
modelVersion,
startMs: track.startMs,
endMs: track.endMs,
confidenceMin: track.confidenceMin,
keyframesPath: track.keyframesPath,
embedding: track.embedding ? toBytes(track.embedding) : null,
personId: track.personId ?? null,
})),
omit: { embedding: true },
});
return rows
.sort((a, b) => a.startMs - b.startMs || a.id.localeCompare(b.id))
.map(mapMediaFaceTrack);
});
}
async function listFaceTracks(
spaceId: string,
filter: { assetId?: string; personId?: string } = {},
): Promise<MediaFaceTrack[]> {
const rows = await prisma.mediaFaceTrack.findMany({
where: { spaceId, ...filter },
omit: { embedding: true },
orderBy: [{ assetId: "asc" }, { startMs: "asc" }, { id: "asc" }],
});
return rows.map(mapMediaFaceTrack);
}
/** Tracks with their keyframe files, for anonymization and the gate. */
async function faceTrackRecords(
spaceId: string,
assetIds: string[],
): Promise<MediaFaceTrackRecord[]> {
if (assetIds.length === 0) return [];
const rows = await prisma.mediaFaceTrack.findMany({
where: { spaceId, assetId: { in: [...new Set(assetIds)] } },
omit: { embedding: true },
orderBy: [{ assetId: "asc" }, { startMs: "asc" }, { id: "asc" }],
});
return rows.map((row) => ({
...mapMediaFaceTrack(row),
keyframesPath: row.keyframesPath,
modelVersion: row.modelVersion,
}));
}
/** Identity embeddings for local person clustering. Never leaves the host. */
async function faceTrackEmbeddings(
spaceId: string,
options: { unassignedOnly?: boolean } = {},
): Promise<MediaFaceEmbedding[]> {
const rows = await prisma.mediaFaceTrack.findMany({
where: {
spaceId,
embedding: { not: null },
...(options.unassignedOnly ? { personId: null } : {}),
},
select: { id: true, assetId: true, personId: true, embedding: true },
orderBy: [{ assetId: "asc" }, { startMs: "asc" }, { id: "asc" }],
});
return rows.map((row) => ({
trackId: row.id,
assetId: row.assetId,
personId: row.personId,
embedding: row.embedding as Uint8Array,
}));
}
// -------------------------------------------------------------------------
// Marks
// -------------------------------------------------------------------------
async function assertPerson(
db: MediaDb,
spaceId: string,
personId: string | null,
): Promise<void> {
if (!personId) return;
const person = await db.mediaPerson.findFirst({
where: { id: personId, spaceId },
select: { id: true },
});
if (!person) throw new IsolationError();
}
async function requireMark(spaceId: string, markId: string): Promise<MediaMarkRow> {
const row = await prisma.mediaMark.findFirst({ where: { id: markId, spaceId } });
if (!row) throw new IsolationError();
return row;
}
async function createMark(spaceId: string, input: NewMediaMark): Promise<MediaMark> {
await assertAssetsInSpace(prisma, spaceId, [input.assetId]);
await assertPerson(prisma, spaceId, input.personId);
const row = await prisma.mediaMark.create({
data: {
spaceId,
kind: input.kind,
label: input.label,
personId: input.personId,
assetId: input.assetId,
prompt: input.prompt,
treatment: input.treatment,
},
});
return mapMediaMark(row);
}
async function listMarks(
spaceId: string,
filter: { personId?: string; assetId?: string } = {},
): Promise<MediaMark[]> {
const rows = await prisma.mediaMark.findMany({
where: { spaceId, ...filter },
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
});
return rows.map(mapMediaMark);
}
/**
* Marks that apply to an asset: those drawn on it plus the standing marks of every
* person seen in it (person rules hold across all projects).
*/
async function marksForAsset(spaceId: string, assetId: string): Promise<MediaMarkRecord[]> {
const rows = await prisma.mediaMark.findMany({
where: {
spaceId,
OR: [{ assetId }, { person: { faceTracks: { some: { assetId, spaceId } } } }],
},
orderBy: [{ createdAt: "asc" }, { id: "asc" }],
});
return rows.map(markRecord);
}
async function getMark(spaceId: string, markId: string): Promise<MediaMarkRecord> {
return markRecord(await requireMark(spaceId, markId));
}
async function updateMark(
spaceId: string,
markId: string,
patch: MediaMarkPatch,
): Promise<MediaMark> {
if (patch.personId !== undefined) await assertPerson(prisma, spaceId, patch.personId);
const result = await prisma.mediaMark.updateMany({
where: { id: markId, spaceId },
data: {
...(patch.treatment !== undefined ? { treatment: patch.treatment } : {}),
...(patch.label !== undefined ? { label: patch.label } : {}),
...(patch.personId !== undefined ? { personId: patch.personId } : {}),
},
});
if (result.count !== 1) throw new IsolationError();
return mapMediaMark(await requireMark(spaceId, markId));
}
/** Store the segmentation/treatment outcome: mask, quality, and any fallback proposal. */
async function recordMarkResult(
spaceId: string,
markId: string,
outcome: MediaMarkResult,
): Promise<MediaMarkRecord> {
const result = await prisma.mediaMark.updateMany({
where: { id: markId, spaceId },
data: {
qualityScore: outcome.qualityScore,
suggestedFallback: outcome.suggestedFallback,
maskDir: outcome.maskDir,
},
});
if (result.count !== 1) throw new IsolationError();
return getMark(spaceId, markId);
}
async function removeMark(spaceId: string, markId: string): Promise<{ maskDir: string | null }> {
const row = await requireMark(spaceId, markId);
const result = await prisma.mediaMark.deleteMany({ where: { id: markId, spaceId } });
if (result.count !== 1) throw new IsolationError();
return { maskDir: row.maskDir };
}
return {
listPeople,
getPerson,
listPersonRecords,
personCentroids,
createPersonSuggestions,
updatePerson,
mergePeople,
assignTrack,
recordConsent,
removePersonCompletely,
replaceFaceTracks,
listFaceTracks,
faceTrackRecords,
faceTrackEmbeddings,
createMark,
listMarks,
marksForAsset,
getMark,
updateMark,
recordMarkResult,
removeMark,
};
}

490
packages/db/src/media-projects.ts Normal file
View File

@ -0,0 +1,490 @@
import type {
BrandKitRecord,
FfprobeReport,
GateReport,
MediaEditPlan,
MediaProject,
MediaProjectGoal,
MediaProjectStatus,
MediaRender,
MediaRenderOutput,
MediaRenderQuality,
MediaRenderStatus,
} from "@rakazo/contracts";
import { BrandKitSchema, EdlSchema, MAX_AUTO_FIX_ROUNDS } from "@rakazo/contracts";
import type {
MediaProject as MediaProjectRow,
MediaRender as MediaRenderRow,
PrismaClient,
} from "./client.js";
import { Prisma } from "./client.js";
import type { MediaDb, MediaPage, MediaPageInput } from "./media-shared.js";
import {
decodeKeysetCursor,
InvalidMediaInputError,
isUniqueViolation,
MediaStateConflictError,
mapBrandKit,
mapMediaEditPlan,
mapMediaProject,
mapMediaRender,
pageLimit,
toPage,
} from "./media-shared.js";
import { IsolationError } from "./scope.js";
export interface NewMediaProject {
title: string;
goal: MediaProjectGoal;
profileIds: string[];
collectionIds?: string[];
assetIds?: string[];
brandKitId?: string | null;
}
export interface MediaProjectPatch {
title?: string;
goal?: MediaProjectGoal;
collectionIds?: string[];
assetIds?: string[];
profileIds?: string[];
brandKitId?: string | null;
status?: MediaProjectStatus;
}
export interface MediaProjectRecord extends MediaProject {
assetIds: string[];
createdBy: string;
}
export interface NewMediaEditPlan {
projectId: string;
profileId: string;
/** Validated against EdlSchema here; semantic validation happens before (validateEdl). */
edl: unknown;
createdBy: "agent" | "user";
rationale: string | null;
}
/** Fields a render worker reports; outputs may carry internal paths next to the public fields. */
export interface MediaRenderPatch {
status?: MediaRenderStatus;
progress?: number;
outputs?: Array<MediaRenderOutput & Record<string, unknown>>;
gateReport?: GateReport | null;
ffprobe?: FfprobeReport | null;
autoFixRound?: number;
error?: string | null;
}
/** Statuses a render can still be changed or canceled from. */
export const ACTIVE_RENDER_STATUSES: MediaRenderStatus[] = ["queued", "rendering", "gating"];
const PLAN_VERSION_ATTEMPTS = 50;
function projectRecord(row: MediaProjectRow): MediaProjectRecord {
return { ...mapMediaProject(row), assetIds: row.assetIds, createdBy: row.createdBy };
}
/** Brand kits, projects, versioned edit plans and renders. */
export function createMediaProjectRepos(prisma: PrismaClient) {
async function assertOwned(
db: MediaDb,
spaceId: string,
model: "mediaBrandKit" | "mediaCollection" | "mediaAsset",
ids: string[],
): Promise<void> {
const unique = [...new Set(ids)];
if (unique.length === 0) return;
const where = { spaceId, id: { in: unique } };
const owned =
model === "mediaBrandKit"
? await db.mediaBrandKit.count({ where })
: model === "mediaCollection"
? await db.mediaCollection.count({ where })
: await db.mediaAsset.count({ where });
if (owned !== unique.length) throw new IsolationError();
}
// -------------------------------------------------------------------------
// Brand kits
// -------------------------------------------------------------------------
async function upsertBrandKit(
spaceId: string,
input: { id?: string; kit: unknown },
): Promise<BrandKitRecord> {
const parsed = BrandKitSchema.safeParse(input.kit);
if (!parsed.success) throw new InvalidMediaInputError("Invalid brand kit");
const kit = parsed.data as Prisma.InputJsonObject;
if (!input.id) {
return mapBrandKit(await prisma.mediaBrandKit.create({ data: { spaceId, kit } }));
}
const result = await prisma.mediaBrandKit.updateMany({
where: { id: input.id, spaceId },
data: { kit },
});
if (result.count !== 1) throw new IsolationError();
return getBrandKit(spaceId, input.id);
}
async function listBrandKits(spaceId: string): Promise<BrandKitRecord[]> {
const rows = await prisma.mediaBrandKit.findMany({
where: { spaceId },
orderBy: [{ updatedAt: "desc" }, { id: "desc" }],
});
return rows.map(mapBrandKit);
}
async function getBrandKit(spaceId: string, id: string): Promise<BrandKitRecord> {
const row = await prisma.mediaBrandKit.findFirst({ where: { id, spaceId } });
if (!row) throw new IsolationError();
return mapBrandKit(row);
}
async function removeBrandKit(spaceId: string, id: string): Promise<void> {
const result = await prisma.mediaBrandKit.deleteMany({ where: { id, spaceId } });
if (result.count !== 1) throw new IsolationError();
}
// -------------------------------------------------------------------------
// Projects and plans
// -------------------------------------------------------------------------
async function requireProject(db: MediaDb, spaceId: string, projectId: string) {
const row = await db.mediaProject.findFirst({ where: { id: projectId, spaceId } });
if (!row) throw new IsolationError();
return row;
}
async function createProject(
spaceId: string,
createdBy: string,
input: NewMediaProject,
): Promise<MediaProject> {
return prisma.$transaction(async (tx) => {
await assertOwned(tx, spaceId, "mediaBrandKit", input.brandKitId ? [input.brandKitId] : []);
await assertOwned(tx, spaceId, "mediaCollection", input.collectionIds ?? []);
await assertOwned(tx, spaceId, "mediaAsset", input.assetIds ?? []);
const row = await tx.mediaProject.create({
data: {
spaceId,
createdBy,
title: input.title,
goal: input.goal,
profileIds: [...new Set(input.profileIds)],
collectionIds: [...new Set(input.collectionIds ?? [])],
assetIds: [...new Set(input.assetIds ?? [])],
brandKitId: input.brandKitId ?? null,
},
});
return mapMediaProject(row);
});
}
async function updateProject(
spaceId: string,
projectId: string,
patch: MediaProjectPatch,
): Promise<MediaProject> {
return prisma.$transaction(async (tx) => {
await assertOwned(tx, spaceId, "mediaBrandKit", patch.brandKitId ? [patch.brandKitId] : []);
await assertOwned(tx, spaceId, "mediaCollection", patch.collectionIds ?? []);
await assertOwned(tx, spaceId, "mediaAsset", patch.assetIds ?? []);
const result = await tx.mediaProject.updateMany({
where: { id: projectId, spaceId },
data: {
...(patch.title !== undefined ? { title: patch.title } : {}),
...(patch.goal !== undefined ? { goal: patch.goal } : {}),
...(patch.status !== undefined ? { status: patch.status } : {}),
...(patch.brandKitId !== undefined ? { brandKitId: patch.brandKitId } : {}),
...(patch.profileIds !== undefined ? { profileIds: [...new Set(patch.profileIds)] } : {}),
...(patch.collectionIds !== undefined
? { collectionIds: [...new Set(patch.collectionIds)] }
: {}),
...(patch.assetIds !== undefined ? { assetIds: [...new Set(patch.assetIds)] } : {}),
},
});
if (result.count !== 1) throw new IsolationError();
return mapMediaProject(await requireProject(tx, spaceId, projectId));
});
}
/** Most recently changed first. */
async function listProjects(
spaceId: string,
input: MediaPageInput = {},
): Promise<MediaPage<MediaProject>> {
const limit = pageLimit(input.limit);
const cursor = input.cursor ? decodeKeysetCursor(input.cursor) : null;
if (cursor && !cursor.at) throw new InvalidMediaInputError("Invalid cursor");
const rows = await prisma.mediaProject.findMany({
where: {
spaceId,
...(cursor?.at
? {
OR: [
{ updatedAt: { lt: cursor.at } },
{ updatedAt: cursor.at, id: { lt: cursor.id } },
],
}
: {}),
},
orderBy: [{ updatedAt: "desc" }, { id: "desc" }],
take: limit + 1,
});
return toPage(rows, limit, (row) => ({ at: row.updatedAt, id: row.id }), mapMediaProject);
}
async function getProjectRecord(spaceId: string, projectId: string): Promise<MediaProjectRecord> {
return projectRecord(await requireProject(prisma, spaceId, projectId));
}
/** Project with the latest plan per profile and its renders, newest first. */
async function getProject(spaceId: string, projectId: string) {
const row = await requireProject(prisma, spaceId, projectId);
const [plans, renders] = await Promise.all([
latestPlans(spaceId, projectId),
listRenders(spaceId, projectId),
]);
return { project: mapMediaProject(row), plans, renders };
}
async function removeProject(spaceId: string, projectId: string): Promise<void> {
const result = await prisma.mediaProject.deleteMany({ where: { id: projectId, spaceId } });
if (result.count !== 1) throw new IsolationError();
}
/**
* Store the next plan version for (project, profile). Concurrent writers race on the
* unique (projectId, profileId, version); the loser re-reads and takes the next number.
*/
async function createPlanVersion(
spaceId: string,
input: NewMediaEditPlan,
): Promise<MediaEditPlan> {
const parsed = EdlSchema.safeParse(input.edl);
if (!parsed.success) throw new InvalidMediaInputError("Invalid edit decision list");
if (parsed.data.output.profileId !== input.profileId) {
throw new InvalidMediaInputError("Plan profile does not match its EDL");
}
await requireProject(prisma, spaceId, input.projectId);
for (let attempt = 1; ; attempt += 1) {
const latest = await prisma.mediaEditPlan.findFirst({
where: { projectId: input.projectId, profileId: input.profileId },
orderBy: { version: "desc" },
select: { version: true },
});
try {
const row = await prisma.mediaEditPlan.create({
data: {
spaceId,
projectId: input.projectId,
profileId: input.profileId,
version: (latest?.version ?? 0) + 1,
edl: parsed.data as Prisma.InputJsonObject,
createdBy: input.createdBy,
rationale: input.rationale,
},
});
return mapMediaEditPlan(row);
} catch (error) {
if (!isUniqueViolation(error) || attempt >= PLAN_VERSION_ATTEMPTS) throw error;
}
}
}
async function getPlan(spaceId: string, planId: string): Promise<MediaEditPlan> {
const row = await prisma.mediaEditPlan.findFirst({ where: { id: planId, spaceId } });
if (!row) throw new IsolationError();
return mapMediaEditPlan(row);
}
async function latestPlan(
spaceId: string,
projectId: string,
profileId: string,
): Promise<MediaEditPlan | null> {
await requireProject(prisma, spaceId, projectId);
const row = await prisma.mediaEditPlan.findFirst({
where: { spaceId, projectId, profileId },
orderBy: { version: "desc" },
});
return row ? mapMediaEditPlan(row) : null;
}
/** The newest plan of every profile in the project, ordered by profile id. */
async function latestPlans(spaceId: string, projectId: string): Promise<MediaEditPlan[]> {
const rows = await prisma.mediaEditPlan.findMany({
where: { spaceId, projectId },
orderBy: [{ profileId: "asc" }, { version: "desc" }],
distinct: ["profileId"],
});
return rows.map(mapMediaEditPlan);
}
async function listPlanVersions(
spaceId: string,
projectId: string,
profileId: string,
): Promise<MediaEditPlan[]> {
await requireProject(prisma, spaceId, projectId);
const rows = await prisma.mediaEditPlan.findMany({
where: { spaceId, projectId, profileId },
orderBy: { version: "desc" },
});
return rows.map(mapMediaEditPlan);
}
// -------------------------------------------------------------------------
// Renders
// -------------------------------------------------------------------------
async function requireRender(db: MediaDb, spaceId: string, renderId: string) {
const row = await db.mediaRender.findFirst({ where: { id: renderId, spaceId } });
if (!row) throw new IsolationError();
return row;
}
async function createRender(
spaceId: string,
input: { planId: string; quality: MediaRenderQuality; createdBy: string },
): Promise<MediaRender> {
const plan = await prisma.mediaEditPlan.findFirst({
where: { id: input.planId, spaceId },
select: { id: true, projectId: true, profileId: true },
});
if (!plan) throw new IsolationError();
const row = await prisma.mediaRender.create({
data: {
spaceId,
projectId: plan.projectId,
planId: plan.id,
profileId: plan.profileId,
quality: input.quality,
createdBy: input.createdBy,
},
});
return mapMediaRender(row);
}
async function getRender(spaceId: string, renderId: string): Promise<MediaRender> {
return mapMediaRender(await requireRender(prisma, spaceId, renderId));
}
/** Full row, including outputs with internal paths. */
async function getRenderRecord(spaceId: string, renderId: string): Promise<MediaRenderRow> {
return requireRender(prisma, spaceId, renderId);
}
async function listRenders(spaceId: string, projectId: string): Promise<MediaRender[]> {
const rows = await prisma.mediaRender.findMany({
where: { spaceId, projectId },
orderBy: [{ createdAt: "desc" }, { id: "desc" }],
});
return rows.map(mapMediaRender);
}
/**
* Take a queued render for processing. The conditional update makes exactly one
* concurrent caller win; the others get null.
*/
async function claimRender(spaceId: string, renderId: string): Promise<MediaRender | null> {
const result = await prisma.mediaRender.updateMany({
where: { id: renderId, spaceId, status: "queued" },
data: { status: "rendering", progress: 0, error: null },
});
const row = await requireRender(prisma, spaceId, renderId);
return result.count === 1 ? mapMediaRender(row) : null;
}
/**
* Apply worker progress or results while the render is in one of `from` (default:
* still active). Returns null when the render already left those states, e.g. after
* a cancel, so a late worker never overwrites it.
*/
async function updateRender(
spaceId: string,
renderId: string,
patch: MediaRenderPatch,
from: MediaRenderStatus[] = ACTIVE_RENDER_STATUSES,
): Promise<MediaRender | null> {
if (
patch.autoFixRound !== undefined &&
(!Number.isInteger(patch.autoFixRound) ||
patch.autoFixRound < 0 ||
patch.autoFixRound > MAX_AUTO_FIX_ROUNDS)
) {
throw new InvalidMediaInputError("Auto-fix round out of range");
}
const result = await prisma.mediaRender.updateMany({
where: { id: renderId, spaceId, status: { in: from } },
data: {
...(patch.status !== undefined ? { status: patch.status } : {}),
...(patch.progress !== undefined
? { progress: Math.min(1, Math.max(0, patch.progress)) }
: {}),
...(patch.outputs !== undefined
? { outputs: patch.outputs as unknown as Prisma.InputJsonArray }
: {}),
...(patch.gateReport !== undefined
? {
gateReport:
patch.gateReport === null
? Prisma.DbNull
: (patch.gateReport as unknown as Prisma.InputJsonObject),
}
: {}),
...(patch.ffprobe !== undefined
? {
ffprobe:
patch.ffprobe === null
? Prisma.DbNull
: (patch.ffprobe as unknown as Prisma.InputJsonObject),
}
: {}),
...(patch.autoFixRound !== undefined ? { autoFixRound: patch.autoFixRound } : {}),
...(patch.error !== undefined ? { error: patch.error } : {}),
},
});
const row = await requireRender(prisma, spaceId, renderId);
return result.count === 1 ? mapMediaRender(row) : null;
}
async function cancelRender(spaceId: string, renderId: string): Promise<MediaRender> {
const result = await prisma.mediaRender.updateMany({
where: { id: renderId, spaceId, status: { in: ACTIVE_RENDER_STATUSES } },
data: { status: "canceled" },
});
const row = await requireRender(prisma, spaceId, renderId);
if (result.count !== 1) {
throw new MediaStateConflictError(`Render is already ${row.status}`);
}
return mapMediaRender(row);
}
return {
upsertBrandKit,
listBrandKits,
getBrandKit,
removeBrandKit,
createProject,
updateProject,
listProjects,
getProject,
getProjectRecord,
removeProject,
createPlanVersion,
getPlan,
latestPlan,
latestPlans,
listPlanVersions,
createRender,
getRender,
getRenderRecord,
listRenders,
claimRender,
updateRender,
cancelRender,
};
}

497
packages/db/src/media-publishing.ts Normal file
View File

@ -0,0 +1,497 @@
import type {
FunnelRow,
MediaMetric,
MediaPublication,
MediaPublicationStatus,
MetricCsvMapping,
TeaserWeights,
Utm,
} from "@rakazo/contracts";
import { MetricCsvMappingSchema, TeaserWeightsSchema } from "@rakazo/contracts";
import type { MediaPublication as MediaPublicationRow, PrismaClient } from "./client.js";
import { Prisma } from "./client.js";
import type { MediaDb, MediaPage, MediaPageInput } from "./media-shared.js";
import {
decodeKeysetCursor,
InvalidMediaInputError,
isUniqueViolation,
MediaStateConflictError,
mapMediaMetric,
mapMediaPublication,
mapMetricMapping,
pageLimit,
toPage,
} from "./media-shared.js";
import { IsolationError } from "./scope.js";
export interface NewMediaPublication {
renderId: string;
destination: string;
targetUrl: string | null;
trackingUrl: string | null;
utm: Utm | null;
title: string | null;
caption: string | null;
tags: string[];
thumbnailVariant: string | null;
scheduledAt: Date | null;
}
export type PublicationKey = MetricCsvMapping["publicationKey"];
export type NewMediaMetric = Omit<MediaMetric, "id">;
export interface TeaserModelRecord {
weights: TeaserWeights;
sampleCount: number;
updatedAt: string;
}
function assertMetric(metric: NewMediaMetric): void {
for (const value of [metric.views, metric.clicks, metric.conversions, metric.revenueCents]) {
if (!Number.isInteger(value) || value < 0) {
throw new InvalidMediaInputError("Metric values must be non-negative integers");
}
}
if (!/^[A-Z]{3}$/.test(metric.currency)) throw new InvalidMediaInputError("Invalid currency");
}
/** Publications, their metrics and the learned teaser weights. */
export function createMediaPublishingRepos(prisma: PrismaClient) {
async function requireRender(spaceId: string, renderId: string) {
const row = await prisma.mediaRender.findFirst({ where: { id: renderId, spaceId } });
if (!row) throw new IsolationError();
return row;
}
async function requireProject(spaceId: string, projectId: string): Promise<void> {
const row = await prisma.mediaProject.findFirst({
where: { id: projectId, spaceId },
select: { id: true },
});
if (!row) throw new IsolationError();
}
async function requirePublication(spaceId: string, publicationId: string) {
const row = await prisma.mediaPublication.findFirst({ where: { id: publicationId, spaceId } });
if (!row) throw new IsolationError();
return row;
}
/** Only a finished render that passed its gate can be published. */
async function createPublication(
spaceId: string,
createdBy: string,
input: NewMediaPublication,
): Promise<MediaPublication> {
const render = await requireRender(spaceId, input.renderId);
if (render.status !== "done") {
throw new MediaStateConflictError("Only a finished render can be published");
}
const row = await prisma.mediaPublication.create({
data: {
spaceId,
createdBy,
renderId: render.id,
profileId: render.profileId,
destination: input.destination,
status: input.scheduledAt ? "scheduled" : "draft",
scheduledAt: input.scheduledAt,
targetUrl: input.targetUrl,
trackingUrl: input.trackingUrl,
utm: input.utm === null ? Prisma.DbNull : (input.utm as Prisma.InputJsonObject),
title: input.title,
caption: input.caption,
tags: input.tags,
thumbnailVariant: input.thumbnailVariant,
},
});
return mapMediaPublication(row);
}
async function getPublication(spaceId: string, publicationId: string): Promise<MediaPublication> {
return mapMediaPublication(await requirePublication(spaceId, publicationId));
}
async function getPublicationRecord(
spaceId: string,
publicationId: string,
): Promise<MediaPublicationRow> {
return requirePublication(spaceId, publicationId);
}
async function listPublications(
spaceId: string,
input: MediaPageInput & { projectId?: string } = {},
): Promise<MediaPage<MediaPublication>> {
const limit = pageLimit(input.limit);
const cursor = input.cursor ? decodeKeysetCursor(input.cursor) : null;
if (cursor && !cursor.at) throw new InvalidMediaInputError("Invalid cursor");
const rows = await prisma.mediaPublication.findMany({
where: {
spaceId,
...(input.projectId ? { render: { projectId: input.projectId, spaceId } } : {}),
...(cursor?.at
? {
OR: [
{ createdAt: { lt: cursor.at } },
{ createdAt: cursor.at, id: { lt: cursor.id } },
],
}
: {}),
},
orderBy: [{ createdAt: "desc" }, { id: "desc" }],
take: limit + 1,
});
return toPage(rows, limit, (row) => ({ at: row.createdAt, id: row.id }), mapMediaPublication);
}
async function transitionPublication(
spaceId: string,
publicationId: string,
from: MediaPublicationStatus[],
data: Prisma.MediaPublicationUpdateManyMutationInput,
): Promise<MediaPublication> {
const result = await prisma.mediaPublication.updateMany({
where: { id: publicationId, spaceId, status: { in: from } },
data,
});
const row = await requirePublication(spaceId, publicationId);
if (result.count !== 1) {
throw new MediaStateConflictError(`Publication is already ${row.status}`);
}
return mapMediaPublication(row);
}
async function schedulePublication(
spaceId: string,
publicationId: string,
scheduledAt: Date,
): Promise<MediaPublication> {
return transitionPublication(spaceId, publicationId, ["draft", "scheduled", "failed"], {
status: "scheduled",
scheduledAt,
});
}
async function cancelPublication(
spaceId: string,
publicationId: string,
): Promise<MediaPublication> {
return transitionPublication(
spaceId,
publicationId,
["draft", "scheduled", "bundled", "failed"],
{ status: "canceled" },
);
}
async function markPublicationBundled(
spaceId: string,
publicationId: string,
bundleArtifactId: string | null,
): Promise<MediaPublication> {
return transitionPublication(
spaceId,
publicationId,
["draft", "scheduled", "bundled", "failed"],
{ status: "bundled", bundleArtifactId },
);
}
async function markPublicationPublished(
spaceId: string,
publicationId: string,
externalUrl: string | null,
publishedAt: Date = new Date(),
): Promise<MediaPublication> {
return transitionPublication(spaceId, publicationId, ["draft", "scheduled", "bundled"], {
status: "published",
externalUrl,
publishedAt,
});
}
async function markPublicationFailed(
spaceId: string,
publicationId: string,
): Promise<MediaPublication> {
return transitionPublication(spaceId, publicationId, ["draft", "scheduled", "bundled"], {
status: "failed",
});
}
/**
* Map CSV keys to publication ids of this space. Values that match nothing (or a
* publication in another space) are absent from the result.
*/
async function resolvePublicationKeys(
spaceId: string,
key: PublicationKey,
values: string[],
): Promise<Map<string, string>> {
const wanted = new Set(values.map((value) => value.trim()).filter(Boolean));
if (wanted.size === 0) return new Map();
const rows = await prisma.mediaPublication.findMany({
where: {
spaceId,
...(key === "id" ? { id: { in: [...wanted] } } : {}),
...(key === "tracking_url" ? { trackingUrl: { in: [...wanted] } } : {}),
...(key === "external_url" ? { externalUrl: { in: [...wanted] } } : {}),
},
select: { id: true, trackingUrl: true, externalUrl: true, utm: true },
});
const resolved = new Map<string, string>();
for (const row of rows) {
const utm = row.utm as { content?: unknown } | null;
const value =
key === "id"
? row.id
: key === "tracking_url"
? row.trackingUrl
: key === "external_url"
? row.externalUrl
: typeof utm?.content === "string"
? utm.content
: null;
if (value && wanted.has(value) && !resolved.has(value)) resolved.set(value, row.id);
}
return resolved;
}
// -------------------------------------------------------------------------
// Metrics
// -------------------------------------------------------------------------
async function upsertMetric(db: MediaDb, spaceId: string, metric: NewMediaMetric) {
assertMetric(metric);
const observedAt = new Date(metric.observedAt);
if (Number.isNaN(observedAt.getTime())) throw new InvalidMediaInputError("Invalid date");
const values = {
views: metric.views,
clicks: metric.clicks,
conversions: metric.conversions,
revenueCents: metric.revenueCents,
currency: metric.currency,
};
return db.mediaMetric.upsert({
where: {
publicationId_observedAt_source: {
publicationId: metric.publicationId,
observedAt,
source: metric.source,
},
},
create: {
...values,
spaceId,
publicationId: metric.publicationId,
observedAt,
source: metric.source,
},
update: values,
});
}
/** One observation; re-recording the same (publication, time, source) replaces it. */
async function recordMetric(spaceId: string, metric: NewMediaMetric): Promise<MediaMetric> {
await requirePublication(spaceId, metric.publicationId);
return mapMediaMetric(await upsertMetric(prisma, spaceId, metric));
}
/** Batch import, all or nothing. */
async function recordMetrics(spaceId: string, metrics: NewMediaMetric[]): Promise<number> {
if (metrics.length === 0) return 0;
return prisma.$transaction(
async (tx) => {
const ids = [...new Set(metrics.map((metric) => metric.publicationId))];
const owned = await tx.mediaPublication.count({ where: { spaceId, id: { in: ids } } });
if (owned !== ids.length) throw new IsolationError();
for (const metric of metrics) await upsertMetric(tx, spaceId, metric);
return metrics.length;
},
{ timeout: 60_000 },
);
}
async function listMetrics(spaceId: string, publicationId: string): Promise<MediaMetric[]> {
await requirePublication(spaceId, publicationId);
const rows = await prisma.mediaMetric.findMany({
where: { spaceId, publicationId },
orderBy: [{ observedAt: "asc" }, { source: "asc" }],
});
return rows.map(mapMediaMetric);
}
async function saveMetricMapping(
spaceId: string,
input: MetricCsvMapping,
): Promise<MetricCsvMapping> {
const parsed = MetricCsvMappingSchema.safeParse(input);
if (!parsed.success) throw new InvalidMediaInputError("Invalid CSV mapping");
const { id, ...mapping } = parsed.data;
const json = mapping as Prisma.InputJsonObject;
try {
if (id) {
const result = await prisma.mediaMetricMapping.updateMany({
where: { id, spaceId },
data: { name: mapping.name, mapping: json },
});
if (result.count !== 1) throw new IsolationError();
return getMetricMapping(spaceId, id);
}
const row = await prisma.mediaMetricMapping.upsert({
where: { spaceId_name: { spaceId, name: mapping.name } },
create: { spaceId, name: mapping.name, mapping: json },
update: { mapping: json },
});
return mapMetricMapping(row);
} catch (error) {
if (isUniqueViolation(error)) throw new InvalidMediaInputError("Mapping name already used");
throw error;
}
}
async function listMetricMappings(spaceId: string): Promise<MetricCsvMapping[]> {
const rows = await prisma.mediaMetricMapping.findMany({
where: { spaceId },
orderBy: [{ name: "asc" }, { id: "asc" }],
});
return rows.map(mapMetricMapping);
}
async function getMetricMapping(spaceId: string, id: string): Promise<MetricCsvMapping> {
const row = await prisma.mediaMetricMapping.findFirst({ where: { id, spaceId } });
if (!row) throw new IsolationError();
return mapMetricMapping(row);
}
/**
* Views → clicks → conversions → revenue per publication. Observations are
* cumulative snapshots, so only the latest one per (publication, source) counts;
* importing the same report twice never doubles the numbers.
*/
async function funnel(spaceId: string, projectId?: string): Promise<FunnelRow[]> {
if (projectId) await requireProject(spaceId, projectId);
const rows = await prisma.$queryRaw<
Array<{
publicationId: string;
projectId: string;
profileId: string;
variant: string | null;
thumbnailVariant: string | null;
views: bigint;
clicks: bigint;
conversions: bigint;
revenueCents: bigint;
currency: string | null;
}>
>(Prisma.sql`
WITH latest AS (
SELECT DISTINCT ON (m."publicationId", m.source)
m."publicationId", m.views, m.clicks, m.conversions, m."revenueCents", m.currency
FROM media_metrics m
WHERE m."spaceId" = ${spaceId}
ORDER BY m."publicationId", m.source, m."observedAt" DESC, m.id DESC
)
SELECT p.id AS "publicationId", r."projectId", p."profileId",
p.utm ->> 'content' AS variant, p."thumbnailVariant",
coalesce(sum(l.views), 0)::bigint AS views,
coalesce(sum(l.clicks), 0)::bigint AS clicks,
coalesce(sum(l.conversions), 0)::bigint AS conversions,
coalesce(sum(l."revenueCents"), 0)::bigint AS "revenueCents",
min(l.currency) AS currency
FROM media_publications p
JOIN media_renders r ON r.id = p."renderId" AND r."spaceId" = p."spaceId"
LEFT JOIN latest l ON l."publicationId" = p.id
WHERE p."spaceId" = ${spaceId}
${projectId ? Prisma.sql`AND r."projectId" = ${projectId}` : Prisma.empty}
GROUP BY p.id, r."projectId"
ORDER BY p."createdAt" DESC, p.id DESC
`);
return rows.map((row) => ({
publicationId: row.publicationId,
projectId: row.projectId,
profileId: row.profileId,
variant: row.variant,
thumbnailVariant: row.thumbnailVariant,
views: Number(row.views),
clicks: Number(row.clicks),
conversions: Number(row.conversions),
revenueCents: Number(row.revenueCents),
currency: row.currency ?? "EUR",
}));
}
/** Publications with at least one metric: the sample size for teaser learning. */
async function publicationsWithMetricsCount(spaceId: string): Promise<number> {
return prisma.mediaPublication.count({ where: { spaceId, metrics: { some: {} } } });
}
// -------------------------------------------------------------------------
// Teaser model
// -------------------------------------------------------------------------
async function getTeaserModel(spaceId: string): Promise<TeaserModelRecord | null> {
const row = await prisma.mediaTeaserModel.findUnique({ where: { spaceId } });
if (!row) return null;
const weights = TeaserWeightsSchema.safeParse(row.weights);
if (!weights.success) return null;
return {
weights: weights.data,
sampleCount: row.sampleCount,
updatedAt: row.updatedAt.toISOString(),
};
}
async function putTeaserModel(
spaceId: string,
weights: TeaserWeights,
sampleCount: number,
): Promise<TeaserModelRecord> {
const parsed = TeaserWeightsSchema.safeParse(weights);
if (!parsed.success || !Number.isInteger(sampleCount) || sampleCount < 0) {
throw new InvalidMediaInputError("Invalid teaser model");
}
const json = parsed.data as Prisma.InputJsonObject;
const row = await prisma.mediaTeaserModel.upsert({
where: { spaceId },
create: { spaceId, weights: json, sampleCount },
update: { weights: json, sampleCount },
});
return {
weights: parsed.data,
sampleCount: row.sampleCount,
updatedAt: row.updatedAt.toISOString(),
};
}
/** Back to default weights. */
async function resetTeaserModel(spaceId: string): Promise<void> {
await prisma.mediaTeaserModel.deleteMany({ where: { spaceId } });
}
return {
createPublication,
getPublication,
getPublicationRecord,
listPublications,
schedulePublication,
cancelPublication,
markPublicationBundled,
markPublicationPublished,
markPublicationFailed,
resolvePublicationKeys,
recordMetric,
recordMetrics,
listMetrics,
saveMetricMapping,
listMetricMappings,
getMetricMapping,
funnel,
publicationsWithMetricsCount,
getTeaserModel,
putTeaserModel,
resetTeaserModel,
};
}

525
packages/db/src/media-shared.ts Normal file
View File

@ -0,0 +1,525 @@
import { createHash } from "node:crypto";
import type { MediaSourceItem } from "@rakazo/adapter-kit";
import type {
AgeVerificationMethod,
AnonymizeMode,
BrandKitRecord,
ClientMediaItem,
MarkTreatment,
MediaAsset,
MediaAssetKind,
MediaAssetStorage,
MediaCollection,
MediaCollectionKind,
MediaEditPlan,
MediaFaceTrack,
MediaIngestState,
MediaLabelSource,
MediaMark,
MediaMarkKind,
MediaMetric,
MediaMetricSource,
MediaPerson,
MediaProject,
MediaProjectGoal,
MediaProjectStatus,
MediaPublication,
MediaPublicationStatus,
MediaRender,
MediaRenderOutput,
MediaRenderQuality,
MediaRenderStatus,
MediaSegment,
MediaSegmentKind,
MediaSource,
MediaSourceKind,
MediaSourceStatus,
MediaUpload,
MetricCsvMapping,
TeaserFeature,
} from "@rakazo/contracts";
import {
BrandKitSchema,
EdlSchema,
FfprobeReportSchema,
GateReportSchema,
MediaMarkPromptSchema,
MediaRenderOutputSchema,
MetricCsvMappingSchema,
UtmSchema,
} from "@rakazo/contracts";
import type {
MediaAsset as MediaAssetRow,
MediaBrandKit as MediaBrandKitRow,
MediaCollection as MediaCollectionRow,
MediaEditPlan as MediaEditPlanRow,
MediaFaceTrack as MediaFaceTrackRow,
MediaMark as MediaMarkRow,
MediaMetricMapping as MediaMetricMappingRow,
MediaMetric as MediaMetricRow,
MediaPerson as MediaPersonRow,
MediaProject as MediaProjectRow,
MediaPublication as MediaPublicationRow,
MediaRender as MediaRenderRow,
MediaSegment as MediaSegmentRow,
MediaSource as MediaSourceRow,
MediaUpload as MediaUploadRow,
PrismaClient,
} from "./client.js";
import { Prisma } from "./client.js";
import { IsolationError } from "./scope.js";
/*
* Shared pieces of the Media Studio repositories: errors, keyset cursors, and
* the row → DTO mappers. Not-found and foreign-space access both surface as
* IsolationError (from scope.ts), like every other repository.
*/
/** Input the repository refuses: bad cursor, invalid EDL or brand kit, foreign ids. */
export class InvalidMediaInputError extends Error {
constructor(message: string) {
super(message);
this.name = "InvalidMediaInputError";
}
}
/** The record exists but its state does not allow the requested change. */
export class MediaStateConflictError extends Error {
constructor(message: string) {
super(message);
this.name = "MediaStateConflictError";
}
}
export type MediaDb = Prisma.TransactionClient | PrismaClient;
/** Throws IsolationError unless every id is an asset of the space. */
export async function assertAssetsInSpace(
db: MediaDb,
spaceId: string,
assetIds: string[],
): Promise<void> {
const ids = [...new Set(assetIds)];
if (ids.length === 0) return;
const owned = await db.mediaAsset.count({ where: { spaceId, id: { in: ids } } });
if (owned !== ids.length) throw new IsolationError();
}
export const MEDIA_PAGE_MAX = 500;
export const MEDIA_PAGE_DEFAULT = 100;
export interface MediaPage<T> {
items: T[];
nextCursor: string | null;
}
export interface MediaPageInput {
cursor?: string | null;
limit?: number;
}
export function pageLimit(limit: number | undefined): number {
if (limit === undefined || !Number.isFinite(limit)) return MEDIA_PAGE_DEFAULT;
return Math.min(MEDIA_PAGE_MAX, Math.max(1, Math.trunc(limit)));
}
/**
* Position after the last row of a page ordered by `(at DESC NULLS LAST, id DESC)`.
* `at` is null for rows without a timestamp, which sort after every dated row.
*/
export interface KeysetCursor {
at: Date | null;
id: string;
}
/** Opaque base64url token for a keyset position. */
export function encodeKeysetCursor(cursor: KeysetCursor): string {
return Buffer.from(JSON.stringify([cursor.at?.toISOString() ?? null, cursor.id])).toString(
"base64url",
);
}
export function decodeKeysetCursor(token: string): KeysetCursor {
let parsed: unknown;
try {
parsed = JSON.parse(Buffer.from(token, "base64url").toString("utf8"));
} catch {
throw new InvalidMediaInputError("Invalid cursor");
}
if (!Array.isArray(parsed) || parsed.length !== 2) {
throw new InvalidMediaInputError("Invalid cursor");
}
const [at, id] = parsed as unknown[];
if (typeof id !== "string" || id.length === 0 || id.length > 100) {
throw new InvalidMediaInputError("Invalid cursor");
}
if (at === null) return { at: null, id };
if (typeof at !== "string") throw new InvalidMediaInputError("Invalid cursor");
const date = new Date(at);
if (Number.isNaN(date.getTime())) throw new InvalidMediaInputError("Invalid cursor");
return { at: date, id };
}
/**
* SQL predicate for "after the cursor" under `ORDER BY at DESC NULLS LAST, id DESC`.
* `atColumn` and `idColumn` are trusted column references.
*/
export function keysetAfterSql(
cursor: KeysetCursor | null,
atColumn: Prisma.Sql,
idColumn: Prisma.Sql,
): Prisma.Sql {
if (!cursor) return Prisma.sql`TRUE`;
if (cursor.at === null) {
return Prisma.sql`(${atColumn} IS NULL AND ${idColumn} < ${cursor.id})`;
}
return Prisma.sql`(${atColumn} < ${cursor.at} OR (${atColumn} = ${cursor.at} AND ${idColumn} < ${cursor.id}) OR ${atColumn} IS NULL)`;
}
/** Slice a `limit + 1` result into a page with its next cursor. */
export function toPage<Row, Item>(
rows: Row[],
limit: number,
keyOf: (row: Row) => KeysetCursor,
map: (row: Row) => Item,
): MediaPage<Item> {
const hasMore = rows.length > limit;
const pageRows = hasMore ? rows.slice(0, limit) : rows;
const last = pageRows.at(-1);
return {
items: pageRows.map(map),
nextCursor: hasMore && last ? encodeKeysetCursor(keyOf(last)) : null,
};
}
/** Escape LIKE wildcards so user text matches literally. */
export function likeContains(text: string): string {
return `%${text.replace(/[\\%_]/g, (char) => `\\${char}`)}%`;
}
export function isUniqueViolation(error: unknown): boolean {
return Boolean(error && typeof error === "object" && "code" in error && error.code === "P2002");
}
/** Advisory lock held until the surrounding transaction ends. */
export async function lockMediaKey(tx: Prisma.TransactionClient, key: string): Promise<void> {
await tx.$queryRaw(Prisma.sql`
SELECT pg_advisory_xact_lock(hashtextextended(${`media:${key}`}, 0))::text AS "lock"
`);
}
/** Prisma Bytes columns want an ArrayBuffer-backed view; copy when the input is a slice. */
export function toBytes(value: Uint8Array): Uint8Array<ArrayBuffer> {
return new Uint8Array(value);
}
/**
* Change detector for a client-pushed item, which has no revision of its own:
* any metadata change produces a different value.
*/
export function clientMediaItemToSourceItem(item: ClientMediaItem): MediaSourceItem {
const canonical = JSON.stringify([
item.kind,
item.capturedAt,
item.durationMs,
item.width,
item.height,
item.fps,
item.gpsLat,
item.gpsLon,
item.deviceModel,
item.albumNames,
item.isFavorite,
item.fileName,
]);
return {
externalId: item.externalId,
kind: item.kind,
capturedAt: item.capturedAt ? new Date(item.capturedAt) : null,
durationMs: item.durationMs,
width: item.width,
height: item.height,
fps: item.fps,
gpsLat: item.gpsLat,
gpsLon: item.gpsLon,
deviceModel: item.deviceModel,
albumNames: item.albumNames,
isFavorite: item.isFavorite,
fileName: item.fileName,
sizeBytes: null,
revision: createHash("sha256").update(canonical).digest("hex").slice(0, 32),
};
}
// ---------------------------------------------------------------------------
// Mappers
// ---------------------------------------------------------------------------
function iso(value: Date | null): string | null {
return value ? value.toISOString() : null;
}
export function mapMediaSource(row: MediaSourceRow, assetCount: number): MediaSource {
return {
id: row.id,
kind: row.kind as MediaSourceKind,
label: row.label,
status: row.status as MediaSourceStatus,
assetCount,
lastSyncedAt: iso(row.lastSyncedAt),
error: row.error,
createdAt: row.createdAt.toISOString(),
};
}
export function mapMediaAsset(row: MediaAssetRow): MediaAsset {
return {
id: row.id,
sourceId: row.sourceId,
kind: row.kind as MediaAssetKind,
storage: row.storage as MediaAssetStorage,
capturedAt: iso(row.capturedAt),
durationMs: row.durationMs,
width: row.width,
height: row.height,
fps: row.fps,
gpsLat: row.gpsLat,
gpsLon: row.gpsLon,
deviceModel: row.deviceModel,
albumNames: row.albumNames,
isFavorite: row.isFavorite,
ingestState: row.ingestState as MediaIngestState,
hasProxy: row.proxyPath !== null,
hasThumbnail: row.thumbPath !== null,
qualityScore: row.qualityScore,
createdAt: row.createdAt.toISOString(),
};
}
function numberRecord(value: Prisma.JsonValue): Partial<Record<TeaserFeature, number>> {
if (!value || typeof value !== "object" || Array.isArray(value)) return {};
const out: Partial<Record<TeaserFeature, number>> = {};
for (const [key, score] of Object.entries(value)) {
if (typeof score === "number" && Number.isFinite(score)) {
out[key as TeaserFeature] = score;
}
}
return out;
}
export function mapMediaSegment(row: MediaSegmentRow): MediaSegment {
return {
id: row.id,
assetId: row.assetId,
kind: row.kind as MediaSegmentKind,
startMs: row.startMs,
endMs: row.endMs,
scores: numberRecord(row.scores),
teaserScore: row.teaserScore,
transcriptExcerpt: row.transcriptExcerpt,
labels: row.labels,
paidOnly: row.paidOnly,
};
}
export interface CollectionStats {
assetCount: number;
videoCount: number;
totalDurationMs: number;
}
export const EMPTY_COLLECTION_STATS: CollectionStats = {
assetCount: 0,
videoCount: 0,
totalDurationMs: 0,
};
export function mapMediaCollection(
row: MediaCollectionRow,
stats: CollectionStats,
): MediaCollection {
return {
id: row.id,
kind: row.kind as MediaCollectionKind,
title: row.title,
labelSource: row.labelSource as MediaLabelSource,
startAt: iso(row.startAt),
endAt: iso(row.endAt),
coverAssetId: row.coverAssetId,
assetCount: stats.assetCount,
videoCount: stats.videoCount,
totalDurationMs: stats.totalDurationMs,
};
}
export function mapMediaPerson(
row: MediaPersonRow,
faceTrackCount: number,
coverThumbAssetId: string | null,
): MediaPerson {
return {
id: row.id,
displayName: row.displayName,
isSelf: row.isSelf,
confirmed: row.confirmed,
anonymizeByDefault: row.anonymizeByDefault as AnonymizeMode,
consentRecordedAt: iso(row.consentRecordedAt),
consentArtifactId: row.consentArtifactId,
ageVerifiedAt: iso(row.ageVerifiedAt),
ageVerificationMethod: row.ageVerificationMethod as AgeVerificationMethod,
faceTrackCount,
coverThumbAssetId,
};
}
export function mapMediaFaceTrack(
row: Pick<
MediaFaceTrackRow,
"id" | "assetId" | "personId" | "startMs" | "endMs" | "confidenceMin"
>,
): MediaFaceTrack {
return {
id: row.id,
assetId: row.assetId,
personId: row.personId,
startMs: row.startMs,
endMs: row.endMs,
confidenceMin: row.confidenceMin,
};
}
export function mapMediaMark(row: MediaMarkRow): MediaMark {
const prompt = row.prompt === null ? null : MediaMarkPromptSchema.safeParse(row.prompt);
return {
id: row.id,
kind: row.kind as MediaMarkKind,
label: row.label,
personId: row.personId,
assetId: row.assetId,
prompt: prompt?.success ? prompt.data : null,
treatment: row.treatment as MarkTreatment,
qualityScore: row.qualityScore,
suggestedFallback: row.suggestedFallback as MarkTreatment | null,
createdAt: row.createdAt.toISOString(),
};
}
export function mapBrandKit(row: MediaBrandKitRow): BrandKitRecord {
return {
id: row.id,
kit: BrandKitSchema.parse(row.kit),
createdAt: row.createdAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
};
}
export function mapMediaProject(row: MediaProjectRow): MediaProject {
return {
id: row.id,
title: row.title,
goal: row.goal as MediaProjectGoal,
status: row.status as MediaProjectStatus,
brandKitId: row.brandKitId,
collectionIds: row.collectionIds,
profileIds: row.profileIds,
createdAt: row.createdAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
};
}
export function mapMediaEditPlan(row: MediaEditPlanRow): MediaEditPlan {
return {
id: row.id,
projectId: row.projectId,
profileId: row.profileId,
version: row.version,
edl: EdlSchema.parse(row.edl),
createdBy: row.createdBy === "agent" ? "agent" : "user",
rationale: row.rationale,
createdAt: row.createdAt.toISOString(),
};
}
/** Public render outputs; internal paths stored next to them are dropped. */
export function renderOutputsOf(value: Prisma.JsonValue): MediaRenderOutput[] {
if (!Array.isArray(value)) return [];
const outputs: MediaRenderOutput[] = [];
for (const entry of value) {
const parsed = MediaRenderOutputSchema.safeParse(entry);
if (parsed.success) outputs.push(parsed.data);
}
return outputs;
}
export function mapMediaRender(row: MediaRenderRow): MediaRender {
const gate = row.gateReport === null ? null : GateReportSchema.safeParse(row.gateReport);
const probe = row.ffprobe === null ? null : FfprobeReportSchema.safeParse(row.ffprobe);
return {
id: row.id,
planId: row.planId,
projectId: row.projectId,
profileId: row.profileId,
quality: row.quality as MediaRenderQuality,
status: row.status as MediaRenderStatus,
progress: Math.min(1, Math.max(0, row.progress)),
outputs: renderOutputsOf(row.outputs),
gateReport: gate?.success ? gate.data : null,
ffprobe: probe?.success ? probe.data : null,
error: row.error,
createdAt: row.createdAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
};
}
export function mapMediaUpload(row: MediaUploadRow): MediaUpload {
return {
id: row.id,
assetId: row.assetId,
sizeBytes: Number(row.sizeBytes),
chunkBytes: row.chunkBytes,
receivedChunks: [...row.receivedChunks].sort((a, b) => a - b),
status: row.status as MediaUpload["status"],
};
}
export function mapMediaPublication(row: MediaPublicationRow): MediaPublication {
const utm = row.utm === null ? null : UtmSchema.safeParse(row.utm);
return {
id: row.id,
renderId: row.renderId,
profileId: row.profileId,
destination: row.destination,
status: row.status as MediaPublicationStatus,
scheduledAt: iso(row.scheduledAt),
publishedAt: iso(row.publishedAt),
externalUrl: row.externalUrl,
targetUrl: row.targetUrl,
trackingUrl: row.trackingUrl,
utm: utm?.success ? utm.data : null,
title: row.title,
caption: row.caption,
tags: row.tags,
thumbnailVariant: row.thumbnailVariant,
bundleArtifactId: row.bundleArtifactId,
createdAt: row.createdAt.toISOString(),
};
}
export function mapMediaMetric(row: MediaMetricRow): MediaMetric {
return {
id: row.id,
publicationId: row.publicationId,
observedAt: row.observedAt.toISOString(),
views: row.views,
clicks: row.clicks,
conversions: row.conversions,
revenueCents: row.revenueCents,
currency: row.currency,
source: row.source as MediaMetricSource,
};
}
export function mapMetricMapping(row: MediaMetricMappingRow): MetricCsvMapping {
const mapping = MetricCsvMappingSchema.parse(row.mapping);
return { ...mapping, id: row.id, name: row.name };
}

163
packages/db/src/media-uploads.ts Normal file
View File

@ -0,0 +1,163 @@
import type { MediaUpload } from "@rakazo/contracts";
import { MEDIA_UPLOAD_CHUNK_BYTES } from "@rakazo/contracts";
import type { PrismaClient } from "./client.js";
import { Prisma } from "./client.js";
import {
InvalidMediaInputError,
lockMediaKey,
MediaStateConflictError,
mapMediaUpload,
} from "./media-shared.js";
import { IsolationError } from "./scope.js";
export interface BeginMediaUpload {
assetId: string;
userId: string;
sizeBytes: number;
sha256: string;
fileName: string;
chunkBytes?: number;
}
function chunkCountOf(row: { sizeBytes: bigint; chunkBytes: number }): number {
return Number((row.sizeBytes + BigInt(row.chunkBytes) - 1n) / BigInt(row.chunkBytes));
}
/** Resumable chunked uploads of originals from client-pushed sources. */
export function createMediaUploadRepos(prisma: PrismaClient) {
/** Opens an upload, or resumes the open one for the same asset and content. */
async function beginUpload(spaceId: string, input: BeginMediaUpload): Promise<MediaUpload> {
const chunkBytes = input.chunkBytes ?? MEDIA_UPLOAD_CHUNK_BYTES;
if (!Number.isInteger(input.sizeBytes) || input.sizeBytes <= 0) {
throw new InvalidMediaInputError("Upload size must be positive");
}
if (!Number.isInteger(chunkBytes) || chunkBytes <= 0) {
throw new InvalidMediaInputError("Chunk size must be positive");
}
return prisma.$transaction(async (tx) => {
const asset = await tx.mediaAsset.findFirst({
where: { id: input.assetId, spaceId },
select: { id: true },
});
if (!asset) throw new IsolationError();
await lockMediaKey(tx, `upload:${input.assetId}`);
const open = await tx.mediaUpload.findFirst({
where: {
spaceId,
assetId: input.assetId,
status: "open",
sha256: input.sha256,
sizeBytes: BigInt(input.sizeBytes),
chunkBytes,
},
orderBy: [{ createdAt: "desc" }, { id: "desc" }],
});
if (open) return mapMediaUpload(open);
const row = await tx.mediaUpload.create({
data: {
spaceId,
assetId: input.assetId,
userId: input.userId,
sizeBytes: BigInt(input.sizeBytes),
chunkBytes,
sha256: input.sha256,
fileName: input.fileName,
},
});
return mapMediaUpload(row);
});
}
async function requireUpload(spaceId: string, uploadId: string) {
const row = await prisma.mediaUpload.findFirst({ where: { id: uploadId, spaceId } });
if (!row) throw new IsolationError();
return row;
}
async function uploadStatus(spaceId: string, uploadId: string): Promise<MediaUpload> {
return mapMediaUpload(await requireUpload(spaceId, uploadId));
}
/** Full row (sha256, file name, owner) for the upload route. */
async function getUploadRecord(spaceId: string, uploadId: string) {
return requireUpload(spaceId, uploadId);
}
/** Idempotent: the index is appended once, atomically, however often it is reported. */
async function markChunkReceived(
spaceId: string,
uploadId: string,
index: number,
): Promise<MediaUpload> {
const row = await requireUpload(spaceId, uploadId);
if (!Number.isInteger(index) || index < 0 || index >= chunkCountOf(row)) {
throw new InvalidMediaInputError("Chunk index out of range");
}
if (row.status !== "open") throw new MediaStateConflictError("Upload is not open");
await prisma.$executeRaw(Prisma.sql`
UPDATE media_uploads
SET "receivedChunks" = array_append("receivedChunks", ${index}::int), "updatedAt" = now()
WHERE id = ${uploadId} AND "spaceId" = ${spaceId} AND status = 'open'
AND NOT (${index}::int = ANY("receivedChunks"))
`);
return uploadStatus(spaceId, uploadId);
}
/**
* Closes a fully received upload and points the asset at the stored original. The
* caller has verified the assembled file against `sha256` before calling.
*/
async function completeUpload(
spaceId: string,
uploadId: string,
stored: { storagePath: string },
): Promise<MediaUpload> {
return prisma.$transaction(async (tx) => {
const row = await tx.mediaUpload.findFirst({ where: { id: uploadId, spaceId } });
if (!row) throw new IsolationError();
if (row.status === "complete") return mapMediaUpload(row);
if (row.status !== "open") throw new MediaStateConflictError("Upload is not open");
const received = new Set(row.receivedChunks);
for (let index = 0; index < chunkCountOf(row); index++) {
if (!received.has(index)) throw new MediaStateConflictError("Upload is missing chunks");
}
const updated = await tx.mediaUpload.update({
where: { id: row.id },
data: { status: "complete" },
});
await tx.mediaAsset.update({
where: { id: row.assetId },
data: {
storage: "uploaded",
storagePath: stored.storagePath,
contentHash: row.sha256,
sizeBytes: row.sizeBytes,
ingestState: "indexed",
ingestError: null,
},
});
return mapMediaUpload(updated);
});
}
async function failUpload(spaceId: string, uploadId: string): Promise<MediaUpload> {
const result = await prisma.mediaUpload.updateMany({
where: { id: uploadId, spaceId, status: "open" },
data: { status: "failed" },
});
if (result.count !== 1) {
await requireUpload(spaceId, uploadId);
throw new MediaStateConflictError("Upload is not open");
}
return uploadStatus(spaceId, uploadId);
}
return {
beginUpload,
uploadStatus,
getUploadRecord,
markChunkReceived,
completeUpload,
failUpload,
};
}

32
packages/db/src/media.ts Normal file
View File

@ -0,0 +1,32 @@
import type { PrismaClient } from "./client.js";
import { createMediaCollectionRepos } from "./media-collections.js";
import { createMediaLibraryRepos } from "./media-library.js";
import { createMediaPeopleRepos } from "./media-people.js";
import { createMediaProjectRepos } from "./media-projects.js";
import { createMediaPublishingRepos } from "./media-publishing.js";
import { createMediaUploadRepos } from "./media-uploads.js";
export * from "./media-collections.js";
export * from "./media-library.js";
export * from "./media-people.js";
export * from "./media-projects.js";
export * from "./media-publishing.js";
export * from "./media-shared.js";
export * from "./media-uploads.js";
/*
* Media Studio persistence. Every function takes the space explicitly and never reads
* or writes another space's rows; a foreign or missing id is an IsolationError.
*/
export function createMediaRepos(prisma: PrismaClient) {
return {
...createMediaLibraryRepos(prisma),
...createMediaUploadRepos(prisma),
...createMediaCollectionRepos(prisma),
...createMediaPeopleRepos(prisma),
...createMediaProjectRepos(prisma),
...createMediaPublishingRepos(prisma),
};
}
export type MediaRepos = ReturnType<typeof createMediaRepos>;