diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 3a251fa..5e45283 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -7,6 +7,7 @@ export * from "./events.js"; export * from "./external-conversations.js"; export * from "./groups.js"; export * from "./main-agent.js"; +export * from "./media.js"; export * from "./memory-config.js"; export * from "./messages.js"; export * from "./messaging.js"; diff --git a/packages/db/src/media-collections.ts b/packages/db/src/media-collections.ts new file mode 100644 index 0000000..c679019 --- /dev/null +++ b/packages/db/src/media-collections.ts @@ -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; + 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> { + 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 { + 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 { + 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, + clusters: MediaClusterInput[], + ): Promise { + const keys = new Set(); + 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> { + const limit = pageLimit(input.limit); + const cursor = input.cursor ? decodeKeysetCursor(input.cursor) : null; + const keys = await prisma.$queryRaw>(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 { + 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 { + 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 { + 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 { + 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 { + 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, + }; +} diff --git a/packages/db/src/media-library.ts b/packages/db/src/media-library.ts new file mode 100644 index 0000000..3bee8a5 --- /dev/null +++ b/packages/db/src/media-library.ts @@ -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; +} + +/** Source row for the sync worker, including its config and cursor. */ +export interface MediaSourceRecord extends MediaSource { + spaceId: string; + ownerUserId: string; + config: Record; + 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 & 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>; + 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) + : {}; + 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): 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + const unique = new Map(); + 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(); + 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> { + 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>(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 { + 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 { + return requireAsset(prisma, spaceId, assetId); + } + + async function listAssetRecords(spaceId: string, assetIds: string[]): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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> { + const rows = await prisma.mediaAnalysis.findMany({ + where: { spaceId, kind }, + orderBy: [{ assetId: "asc" }, { createdAt: "desc" }, { id: "desc" }], + }); + const latest = new Map(); + 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 { + 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 { + 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 { + 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, + }; +} diff --git a/packages/db/src/media-people.ts b/packages/db/src/media-people.ts new file mode 100644 index 0000000..05261b4 --- /dev/null +++ b/packages/db/src/media-people.ts @@ -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 { + 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 { + 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 { + 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> { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + return markRecord(await requireMark(spaceId, markId)); + } + + async function updateMark( + spaceId: string, + markId: string, + patch: MediaMarkPatch, + ): Promise { + 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 { + 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, + }; +} diff --git a/packages/db/src/media-projects.ts b/packages/db/src/media-projects.ts new file mode 100644 index 0000000..9b113ee --- /dev/null +++ b/packages/db/src/media-projects.ts @@ -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>; + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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> { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + return mapMediaRender(await requireRender(prisma, spaceId, renderId)); + } + + /** Full row, including outputs with internal paths. */ + async function getRenderRecord(spaceId: string, renderId: string): Promise { + return requireRender(prisma, spaceId, renderId); + } + + async function listRenders(spaceId: string, projectId: string): Promise { + 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 { + 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 { + 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 { + 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, + }; +} diff --git a/packages/db/src/media-publishing.ts b/packages/db/src/media-publishing.ts new file mode 100644 index 0000000..2da941f --- /dev/null +++ b/packages/db/src/media-publishing.ts @@ -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; + +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 { + 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 { + 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 { + return mapMediaPublication(await requirePublication(spaceId, publicationId)); + } + + async function getPublicationRecord( + spaceId: string, + publicationId: string, + ): Promise { + return requirePublication(spaceId, publicationId); + } + + async function listPublications( + spaceId: string, + input: MediaPageInput & { projectId?: string } = {}, + ): Promise> { + 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 { + 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 { + return transitionPublication(spaceId, publicationId, ["draft", "scheduled", "failed"], { + status: "scheduled", + scheduledAt, + }); + } + + async function cancelPublication( + spaceId: string, + publicationId: string, + ): Promise { + return transitionPublication( + spaceId, + publicationId, + ["draft", "scheduled", "bundled", "failed"], + { status: "canceled" }, + ); + } + + async function markPublicationBundled( + spaceId: string, + publicationId: string, + bundleArtifactId: string | null, + ): Promise { + 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 { + return transitionPublication(spaceId, publicationId, ["draft", "scheduled", "bundled"], { + status: "published", + externalUrl, + publishedAt, + }); + } + + async function markPublicationFailed( + spaceId: string, + publicationId: string, + ): Promise { + 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> { + 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(); + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + return prisma.mediaPublication.count({ where: { spaceId, metrics: { some: {} } } }); + } + + // ------------------------------------------------------------------------- + // Teaser model + // ------------------------------------------------------------------------- + + async function getTeaserModel(spaceId: string): Promise { + 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 { + 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 { + 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, + }; +} diff --git a/packages/db/src/media-shared.ts b/packages/db/src/media-shared.ts new file mode 100644 index 0000000..589160e --- /dev/null +++ b/packages/db/src/media-shared.ts @@ -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 { + 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 { + 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( + rows: Row[], + limit: number, + keyOf: (row: Row) => KeysetCursor, + map: (row: Row) => Item, +): MediaPage { + 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 { + 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 { + 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> { + if (!value || typeof value !== "object" || Array.isArray(value)) return {}; + const out: Partial> = {}; + 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 }; +} diff --git a/packages/db/src/media-uploads.ts b/packages/db/src/media-uploads.ts new file mode 100644 index 0000000..d58eb74 --- /dev/null +++ b/packages/db/src/media-uploads.ts @@ -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 { + 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 { + 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 { + 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 { + 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 { + 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, + }; +} diff --git a/packages/db/src/media.ts b/packages/db/src/media.ts new file mode 100644 index 0000000..bfc3fd2 --- /dev/null +++ b/packages/db/src/media.ts @@ -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;