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