diff --git a/packages/adapters/src/media-ingest-analyze.ts b/packages/adapters/src/media-ingest-analyze.ts new file mode 100644 index 0000000..a2fb404 --- /dev/null +++ b/packages/adapters/src/media-ingest-analyze.ts @@ -0,0 +1,633 @@ +import { readFile, rm } from "node:fs/promises"; +import type { + AudioFeaturesResult, + DetectScenesResult, + MediaInferenceProvider, + SaliencyResult, + TranscribeResult, + TranscriptWord, +} from "@rakazo/adapter-kit"; +import { UnsupportedMediaOperationError } from "@rakazo/adapter-kit"; +import type { + MediaAnalysisKind, + MediaCapabilities, + MediaInferenceOperation, + TeaserFeature, + TeaserWeights, +} from "@rakazo/contracts"; +import { + BoxKeyframeSchema, + MediaAnalysisKindSchema, + TEASER_LEARNING_MIN_SAMPLES, +} from "@rakazo/contracts"; +import type { NewMediaFaceTrack, NewMediaSegment, Prisma } from "@rakazo/db"; +import { + cosineSimilarity, + decodeVectors, + defaultTeaserWeights, + encodeVectors, + meanVector, + scoreSegment, + speechHookScore, +} from "@rakazo/media"; +import { assertInside, resolveInside } from "@rakazo/media/node"; +import { z } from "zod"; +import type { MediaServiceDeps } from "./media-context.js"; +import { assetDir, writeFileAtomic } from "./media-ingest-files.js"; + +/* + * Sidecar analyses for one asset and the scored segments built from them. + * + * Biometric rule: face identity embeddings are stored only in + * MediaFaceTrack.embedding. The "faces" analysis row holds track spans and + * counts, never an embedding; box keyframes live in a JSON file per track. + * + * Each analysis is stored under modelVersion ":", so a + * model change adds rows instead of overwriting earlier results. An operation + * the provider reports as unsupported is skipped and records nothing. + */ + +/** Version of the segment pipeline written to MediaAsset.analysisVersion. */ +export const MEDIA_ANALYSIS_VERSION = 1; + +const KEYFRAME_LIMIT = 60; +const WINDOW_MS = 3000; +const STILL_SEGMENT_MS = 3000; +const FACE_SAMPLE_FPS = 2; +const CLASSIFIER_SAMPLE_FPS = 1; +const SAME_FACE_SIMILARITY = 0.6; +const EXCERPT_LIMIT = 500; + +const ANALYSIS_OPERATION: Record = { + scenes: "detectScenes", + transcript: "transcribe", + faces: "trackFaces", + embeddings: "embedFrames", + classifier: "classifyContent", + audio: "audioFeatures", + saliency: "saliency", +}; + +/** Kinds that need a time axis and audio; stills skip them. */ +const VIDEO_ONLY: ReadonlySet = new Set(["scenes", "transcript", "audio"]); + +/** Keyframe times: one per second, at most 60 spread evenly; a still is sampled once. */ +export function keyframeTimes(durationMs: number | null, still: boolean): number[] { + if (still || !durationMs || durationMs <= 0) return [0]; + const seconds = Math.floor(durationMs / 1000); + if (seconds <= 1) return [0]; + const count = Math.min(KEYFRAME_LIMIT, seconds); + const step = count === seconds ? 1000 : durationMs / count; + return Array.from({ length: count }, (_, index) => Math.round(index * step)); +} + +// --------------------------------------------------------------------------- +// Stored analysis shapes (written here, read back for partial re-runs) +// --------------------------------------------------------------------------- + +const Span = z.object({ startMs: z.number(), endMs: z.number() }); +const ScenesData = z.object({ + shots: z.array(Span.extend({ score: z.number() })), + scenes: z.array(Span), +}); +const TranscriptData = z.object({ + language: z.string().nullable(), + words: z.array(Span.extend({ text: z.string(), confidence: z.number() })), +}); +const FacesData = z.object({ + sampleFps: z.number(), + trackCount: z.number(), + tracks: z.array(Span.extend({ keyframeCount: z.number() })), +}); +const EmbeddingsData = z.object({ dim: z.number().int().positive(), timesMs: z.array(z.number()) }); +const AudioData = z.object({ + beatsMs: z.array(z.number()), + energy: z.array(z.object({ tMs: z.number(), value: z.number() })), +}); +const SaliencyData = z.object({ + points: z.array( + z.object({ tMs: z.number(), cx: z.number(), cy: z.number(), strength: z.number() }), + ), +}); + +/** Box keyframes of one face track as written to `faces/.json`. */ +export const FaceKeyframesFileSchema = z.object({ + sampleFps: z.number().positive(), + keyframes: z.array(BoxKeyframeSchema.extend({ confidence: z.number() })), +}); +export type FaceKeyframesFile = z.infer; + +/** Read a face track's keyframes file (path relative to the media root). */ +export async function readFaceKeyframes( + mediaRoot: string, + keyframesPath: string, +): Promise { + const file = await resolveInside(mediaRoot, keyframesPath); + return FaceKeyframesFileSchema.parse(JSON.parse(await readFile(file, "utf8"))); +} + +/** Everything segment scoring reads; null where the analysis is missing or unsupported. */ +export interface AnalysisBundle { + scenes: DetectScenesResult | null; + transcript: Pick | null; + faceSpans: Array<{ startMs: number; endMs: number }> | null; + embeddings: { timesMs: number[]; vectors: Float32Array[] } | null; + audio: Pick | null; + saliency: SaliencyResult | null; +} + +const EMPTY_BUNDLE: AnalysisBundle = { + scenes: null, + transcript: null, + faceSpans: null, + embeddings: null, + audio: null, + saliency: null, +}; + +// --------------------------------------------------------------------------- +// Segments +// --------------------------------------------------------------------------- + +const round3 = (value: number) => Math.round(Math.min(1, Math.max(0, value)) * 1000) / 1000; + +function inSpan(tMs: number, span: { startMs: number; endMs: number }): boolean { + return tMs >= span.startMs && tMs < span.endMs; +} + +function overlapMs(a: { startMs: number; endMs: number }, b: { startMs: number; endMs: number }) { + return Math.max(0, Math.min(a.endMs, b.endMs) - Math.max(a.startMs, b.startMs)); +} + +/** Union length of `spans` clipped to `window`. */ +function coveredMs( + spans: Array<{ startMs: number; endMs: number }>, + window: { startMs: number; endMs: number }, +): number { + const clipped = spans + .map((span) => ({ + startMs: Math.max(span.startMs, window.startMs), + endMs: Math.min(span.endMs, window.endMs), + })) + .filter((span) => span.endMs > span.startMs) + .sort((a, b) => a.startMs - b.startMs); + let total = 0; + let end = Number.NEGATIVE_INFINITY; + for (const span of clipped) { + const start = Math.max(span.startMs, end); + if (span.endMs > start) total += span.endMs - start; + end = Math.max(end, span.endMs); + } + return total; +} + +function segmentSpans( + durationMs: number | null, + still: boolean, + scenes: DetectScenesResult | null, +): Array<{ startMs: number; endMs: number; kind: "shot" | "scene" }> { + if (still) return [{ startMs: 0, endMs: STILL_SEGMENT_MS, kind: "shot" }]; + const total = durationMs && durationMs > 0 ? durationMs : WINDOW_MS; + const shots = (scenes?.shots ?? []) + .map((shot) => ({ + startMs: Math.max(0, Math.round(shot.startMs)), + endMs: Math.min(total, Math.round(shot.endMs)), + })) + .filter((shot) => shot.endMs > shot.startMs) + .sort((a, b) => a.startMs - b.startMs); + if (shots.length > 0) return shots.map((shot) => ({ ...shot, kind: "shot" as const })); + const windows: Array<{ startMs: number; endMs: number; kind: "scene" }> = []; + for (let start = 0; start < total; start += WINDOW_MS) { + windows.push({ startMs: start, endMs: Math.min(total, start + WINDOW_MS), kind: "scene" }); + } + return windows; +} + +/** + * Scored segments: detected shots, else fixed 3 s windows (one span for a still). + * Features are 0..1; a feature without its analysis is omitted (scores as 0). + * motion mean visual change between consecutive keyframe embeddings + * sharpness resolution proxy (shorter side / 1080); no blur measure yet + * faces share of the span covered by face tracks + * audioPeak loudest energy sample in the span relative to the asset maximum + * speechHook speechHookScore of the words inside the span + * novelty embedding distance to the previous span (first span 0.5) + * beatProximity closeness of the span start to a beat (within 500 ms) + * aesthetics saliency strength proxy + * gaze stays unset: no landmarks are computed during ingest. + */ +export function buildSegments(input: { + durationMs: number | null; + still: boolean; + width: number | null; + height: number | null; + bundle: AnalysisBundle; + weights: TeaserWeights; +}): NewMediaSegment[] { + const { bundle } = input; + const spans = segmentSpans(input.durationMs, input.still, bundle.scenes); + const shortSide = Math.min(input.width ?? 0, input.height ?? 0); + const energyMax = Math.max(0, ...(bundle.audio?.energy.map((sample) => sample.value) ?? [])); + const language = bundle.transcript?.language ?? undefined; + let previousMean: Float32Array | null = null; + + return spans.map((span, index) => { + const scores: Partial> = {}; + const embeddings = bundle.embeddings; + const frames = embeddings + ? embeddings.timesMs.flatMap((tMs, i) => { + const vector = embeddings.vectors[i]; + return vector && (input.still || inSpan(tMs, span)) ? [vector] : []; + }) + : []; + if (frames.length >= 2) { + let change = 0; + for (let i = 1; i < frames.length; i++) { + change += 1 - cosineSimilarity(frames[i - 1] as Float32Array, frames[i] as Float32Array); + } + scores.motion = round3((change / (frames.length - 1)) * 4); + } else if (bundle.scenes && !input.still) { + const cuts = bundle.scenes.shots.filter( + (shot) => shot.startMs > span.startMs && shot.startMs < span.endMs, + ).length; + scores.motion = round3(cuts / Math.max(1, (span.endMs - span.startMs) / 1000)); + } + if (shortSide > 0) scores.sharpness = round3(shortSide / 1080); + if (bundle.faceSpans) { + scores.faces = input.still + ? bundle.faceSpans.length > 0 + ? 1 + : 0 + : round3(coveredMs(bundle.faceSpans, span) / (span.endMs - span.startMs)); + } + if (bundle.audio && energyMax > 0) { + const inside = bundle.audio.energy.filter((sample) => inSpan(sample.tMs, span)); + scores.audioPeak = round3(Math.max(0, ...inside.map((sample) => sample.value)) / energyMax); + } + let excerpt: string | null = null; + if (bundle.transcript) { + const words = bundle.transcript.words.filter((word) => + inSpan((word.startMs + word.endMs) / 2, span), + ); + excerpt = words.map((word) => word.text.trim()).filter(Boolean).join(" ") || null; + scores.speechHook = excerpt ? round3(speechHookScore(excerpt, language)) : 0; + } + const mean = frames.length > 0 ? meanVector(frames) : null; + if (mean) { + scores.novelty = + index === 0 || !previousMean || previousMean.length !== mean.length + ? 0.5 + : round3(1 - cosineSimilarity(previousMean, mean)); + previousMean = mean; + } + if (bundle.audio && bundle.audio.beatsMs.length > 0 && !input.still) { + const distance = Math.min(...bundle.audio.beatsMs.map((beat) => Math.abs(beat - span.startMs))); + scores.beatProximity = round3(1 - distance / 500); + } + if (bundle.saliency && bundle.saliency.points.length > 0) { + const inside = bundle.saliency.points.filter((point) => input.still || inSpan(point.tMs, span)); + const points = inside.length > 0 ? inside : bundle.saliency.points; + scores.aesthetics = round3( + points.reduce((sum, point) => sum + point.strength, 0) / points.length, + ); + } + return { + kind: span.kind, + startMs: span.startMs, + endMs: span.endMs, + scores, + teaserScore: Math.round(scoreSegment(scores, input.weights) * 1000) / 1000, + transcriptExcerpt: excerpt ? excerpt.slice(0, EXCERPT_LIMIT) : null, + labels: [], + }; + }); +} + +// --------------------------------------------------------------------------- +// Running analyses +// --------------------------------------------------------------------------- + +type AssetRow = Awaited>; + +function modelOf(capabilities: MediaCapabilities | null, operation: MediaInferenceOperation) { + const entry = capabilities?.operations[operation]; + return { available: entry?.available !== false, model: entry?.model ?? null }; +} + +/** Run `fn`, or return null when the provider does not support the operation. */ +async function unlessUnsupported(fn: () => Promise): Promise { + try { + return await fn(); + } catch (error) { + if (error instanceof UnsupportedMediaOperationError) return null; + throw error; + } +} + +function safeName(value: string): string { + return value.replace(/[^A-Za-z0-9_-]/g, "_").slice(0, 64) || "track"; +} + +/** + * Re-analysis keeps the person a track was assigned to when the new track of the + * same asset carries the same identity; tracks of other assets are never matched here. + */ +async function previousAssignments(deps: MediaServiceDeps, spaceId: string, assetId: string) { + const rows = await deps.prisma.mediaFaceTrack.findMany({ + where: { spaceId, assetId, personId: { not: null }, embedding: { not: null } }, + select: { personId: true, embedding: true }, + }); + return rows.flatMap((row) => { + const bytes = row.embedding as Uint8Array; + if (!row.personId || bytes.byteLength === 0 || bytes.byteLength % 4 !== 0) return []; + const [vector] = decodeVectors(bytes, bytes.byteLength / 4); + return vector ? [{ personId: row.personId, vector }] : []; + }); +} + +async function storedBundle( + deps: MediaServiceDeps, + spaceId: string, + assetId: string, + kinds: MediaAnalysisKind[], +): Promise { + const bundle: AnalysisBundle = { ...EMPTY_BUNDLE }; + for (const kind of kinds) { + const row = await deps.repos.latestAnalysis(spaceId, assetId, kind); + if (!row) continue; + if (kind === "scenes") { + const parsed = ScenesData.safeParse(row.data); + if (parsed.success) bundle.scenes = parsed.data; + } else if (kind === "transcript") { + const parsed = TranscriptData.safeParse(row.data); + if (parsed.success) bundle.transcript = parsed.data; + } else if (kind === "faces") { + const parsed = FacesData.safeParse(row.data); + if (parsed.success) bundle.faceSpans = parsed.data.tracks; + } else if (kind === "embeddings") { + const parsed = EmbeddingsData.safeParse(row.data); + if (parsed.success && row.vectors) { + bundle.embeddings = { + timesMs: parsed.data.timesMs, + vectors: decodeVectors(row.vectors, parsed.data.dim), + }; + } + } else if (kind === "audio") { + const parsed = AudioData.safeParse(row.data); + if (parsed.success) bundle.audio = parsed.data; + } else if (kind === "saliency") { + const parsed = SaliencyData.safeParse(row.data); + if (parsed.success) bundle.saliency = parsed.data; + } + } + return bundle; +} + +export interface AnalyzeOutcome { + ran: MediaAnalysisKind[]; + skipped: MediaAnalysisKind[]; + segments: number; + qualityScore: number | null; +} + +/** + * Run the requested analyses (all when omitted) on `inputPath`, store them, and + * rebuild the asset's segments. The caller owns ingest state transitions. + */ +export async function analyzeAsset(input: { + deps: MediaServiceDeps; + inference: MediaInferenceProvider; + spaceId: string; + asset: AssetRow; + inputPath: string; + kinds?: MediaAnalysisKind[]; +}): Promise { + const { deps, inference, spaceId, asset, inputPath } = input; + const still = asset.kind === "photo" || !asset.durationMs; + const requested = new Set(input.kinds ?? MediaAnalysisKindSchema.options); + const capabilities = await inference.capabilities().catch(() => null); + const times = keyframeTimes(asset.durationMs, still); + const bundle: AnalysisBundle = { ...EMPTY_BUNDLE }; + const ran: MediaAnalysisKind[] = []; + const skipped: MediaAnalysisKind[] = []; + + const version = (kind: MediaAnalysisKind, resultModel?: string) => { + const model = + modelOf(capabilities, ANALYSIS_OPERATION[kind]).model ?? resultModel ?? "default"; + return { modelId: model, modelVersion: `${inference.id}:${model}` }; + }; + const store = async ( + kind: MediaAnalysisKind, + data: Prisma.InputJsonValue, + extra: { vectors?: Uint8Array; resultModel?: string } = {}, + ) => { + await deps.repos.upsertAnalysis(spaceId, asset.id, { + kind, + ...version(kind, extra.resultModel), + data, + vectors: extra.vectors ?? null, + }); + ran.push(kind); + }; + const runnable = (kind: MediaAnalysisKind) => { + if (!requested.has(kind) || (still && VIDEO_ONLY.has(kind))) return false; + if (!modelOf(capabilities, ANALYSIS_OPERATION[kind]).available) { + skipped.push(kind); + return false; + } + return true; + }; + const attempt = async (kind: MediaAnalysisKind, fn: () => Promise) => { + const result = await unlessUnsupported(fn); + if (result === null) skipped.push(kind); + return result; + }; + + if (runnable("scenes")) { + const scenes = await attempt("scenes", () => inference.detectScenes({ path: inputPath })); + if (scenes) { + bundle.scenes = scenes; + await store("scenes", { shots: scenes.shots, scenes: scenes.scenes }); + } + } + + if (runnable("transcript")) { + const transcript = await attempt("transcript", () => inference.transcribe({ path: inputPath })); + if (transcript) { + bundle.transcript = transcript; + const words: TranscriptWord[] = transcript.words; + await store("transcript", { + language: transcript.language, + words: words.map((word) => ({ ...word })), + text: words + .map((word) => word.text.trim()) + .filter(Boolean) + .join(" "), + }); + } + } + + if (runnable("faces")) { + const sampleFps = still ? 1 : FACE_SAMPLE_FPS; + const tracked = await attempt("faces", () => + inference.trackFaces({ path: inputPath, sampleFps }), + ); + if (tracked) { + const facesDir = `${assetDir(spaceId, asset.id)}/faces`; + await rm(await resolveInside(deps.mediaRoot, facesDir), { recursive: true, force: true }); + const previous = await previousAssignments(deps, spaceId, asset.id); + const stepMs = Math.round(1000 / tracked.sampleFps); + const tracks: NewMediaFaceTrack[] = []; + const spans: Array<{ startMs: number; endMs: number; keyframeCount: number }> = []; + for (const [index, track] of tracked.tracks.entries()) { + const keyframes = [...track.keyframes].sort((a, b) => a.tMs - b.tMs); + const first = keyframes[0]; + const last = keyframes.at(-1); + if (!first || !last) continue; + const keyframesPath = `${facesDir}/${index}-${safeName(track.trackId)}.json`; + const file: FaceKeyframesFile = { sampleFps: tracked.sampleFps, keyframes }; + await writeFileAtomic( + await resolveInside(deps.mediaRoot, keyframesPath), + JSON.stringify(file), + ); + const embedding = track.embedding && track.embedding.length > 0 ? track.embedding : null; + let personId: string | null = null; + if (embedding) { + let best = SAME_FACE_SIMILARITY; + for (const candidate of previous) { + if (candidate.vector.length !== embedding.length) continue; + const similarity = cosineSimilarity(candidate.vector, embedding); + if (similarity >= best) { + best = similarity; + personId = candidate.personId; + } + } + } + const startMs = Math.max(0, Math.round(first.tMs)); + const endMs = Math.max(startMs + 1, Math.round(last.tMs) + stepMs); + tracks.push({ + startMs, + endMs, + confidenceMin: Math.min(...keyframes.map((keyframe) => keyframe.confidence)), + keyframesPath, + embedding: embedding ? encodeVectors([embedding]) : null, + personId, + }); + spans.push({ startMs, endMs, keyframeCount: keyframes.length }); + } + const { modelVersion } = version("faces"); + await deps.repos.replaceFaceTracks(spaceId, asset.id, modelVersion, tracks); + bundle.faceSpans = spans; + // Spans and counts only: identity embeddings never enter an analysis row. + await store("faces", { sampleFps: tracked.sampleFps, trackCount: spans.length, tracks: spans }); + } + } + + if (runnable("embeddings")) { + const embedded = await attempt("embeddings", () => + inference.embedFrames({ path: inputPath, timesMs: times }), + ); + if (embedded && embedded.vectors.length > 0) { + const timesMs = times.slice(0, embedded.vectors.length); + bundle.embeddings = { + timesMs, + vectors: embedded.vectors.map((vector) => Float32Array.from(vector)), + }; + await store( + "embeddings", + { model: embedded.model, dim: embedded.dim, timesMs }, + { vectors: encodeVectors(embedded.vectors), resultModel: embedded.model }, + ); + } + } + + if (runnable("classifier")) { + const boundaries = (bundle.scenes?.shots ?? []).map((shot) => shot.startMs).filter((t) => t > 0); + const classified = await attempt("classifier", () => + inference.classifyContent({ + path: inputPath, + sampleFps: CLASSIFIER_SAMPLE_FPS, + ...(boundaries.length > 0 ? { extraTimesMs: boundaries } : {}), + }), + ); + if (classified) { + await store( + "classifier", + { model: classified.model, frames: classified.frames }, + { resultModel: classified.model }, + ); + } + } + + if (runnable("audio")) { + const audio = await attempt("audio", () => inference.audioFeatures({ path: inputPath })); + if (audio) { + bundle.audio = audio; + await store("audio", { ...audio }); + } + } + + if (runnable("saliency")) { + const saliency = await attempt("saliency", () => + inference.saliency({ path: inputPath, timesMs: times }), + ); + if (saliency) { + bundle.saliency = saliency; + await store("saliency", { points: saliency.points }); + } + } + + // Kinds not run this time come from their latest stored result. + const missing = MediaAnalysisKindSchema.options.filter( + (kind) => !ran.includes(kind) && kind !== "classifier", + ); + if (input.kinds && missing.length > 0) { + const stored = await storedBundle(deps, spaceId, asset.id, missing); + bundle.scenes ??= stored.scenes; + bundle.transcript ??= stored.transcript; + bundle.faceSpans ??= stored.faceSpans; + bundle.embeddings ??= stored.embeddings; + bundle.audio ??= stored.audio; + bundle.saliency ??= stored.saliency; + } + + const learned = await deps.repos.getTeaserModel(spaceId); + const weights = + learned && learned.sampleCount >= TEASER_LEARNING_MIN_SAMPLES + ? learned.weights + : defaultTeaserWeights("teaser"); + const segments = buildSegments({ + durationMs: asset.durationMs, + still, + width: asset.width, + height: asset.height, + bundle, + weights, + }); + await deps.repos.replaceSegments(spaceId, asset.id, MEDIA_ANALYSIS_VERSION, segments); + const teaserScores = segments.flatMap((segment) => + typeof segment.teaserScore === "number" ? [segment.teaserScore] : [], + ); + const qualityScore = teaserScores.length > 0 ? Math.max(...teaserScores) : null; + return { ran, skipped, segments: segments.length, qualityScore }; +} + +/** + * The file the sidecar analyses. A referenced original inside the media root + * (the sidecar's only mount) is analysed directly; everything else uses the + * proxy, which shares the original's timeline and applies its rotation. + */ +export async function analysisInputPath(input: { + mediaRoot: string; + proxyPath: string; + referencedOriginal: string | null; +}): Promise { + if (input.referencedOriginal) { + try { + return await assertInside(input.mediaRoot, input.referencedOriginal); + } catch { + // Outside the media root: the sidecar cannot read it. + } + } + return resolveInside(input.mediaRoot, input.proxyPath); +} diff --git a/packages/adapters/src/media-ingest-cluster.ts b/packages/adapters/src/media-ingest-cluster.ts new file mode 100644 index 0000000..3bd141b --- /dev/null +++ b/packages/adapters/src/media-ingest-cluster.ts @@ -0,0 +1,415 @@ +import { readFile } from "node:fs/promises"; +import type { AgentInputImage } from "@rakazo/adapter-kit"; +import type { MediaClusterInput } from "@rakazo/db"; +import type { ClusterAsset, VectorItem } from "@rakazo/media"; +import { + buildSessions, + clusterFaces, + clusterThemes, + cosineSimilarity, + decodeVectors, + encodeVectors, + meanVector, + ruleLabel, +} from "@rakazo/media"; +import { resolveInside } from "@rakazo/media/node"; +import { getLogger } from "@rakazo/logging"; +import { z } from "zod"; +import type { MediaServiceDeps } from "./media-context.js"; +import { errorMessage, publishSafe } from "./media-ingest-files.js"; + +/* + * Sessions, themes, and person suggestions for one space. + * + * Bounds: themes use average-linkage clustering (O(n²) time and memory), so + * each run clusters at most THEME_BATCH_LIMIT of the most recently captured + * analysed assets; older assets keep no theme until they fall into the window. + * Face clustering is bounded the same way per run (FACE_BATCH_LIMIT unassigned + * tracks); the rest are picked up by the next run. + * + * Theme titles come from the model only when the media model client returns + * one (it checks AI consent for the space's library owner); it receives cover + * thumbnails and nothing else. Face embeddings and person data never leave + * this process. + * + * Collections the user edited (labelSource "user") keep their membership when + * it no longer matches the computed cluster (merged or split): their assets are + * withheld from every other computed cluster so re-clustering never re-adds + * them elsewhere. A collection whose stored assets are still a subset of the + * computed cluster (only renamed) keeps updating. + */ + +export const THEME_BATCH_LIMIT = 2000; +export const FACE_BATCH_LIMIT = 2000; +const THEME_MIN_SIZE = 3; +const THEME_SIMILARITY = 0.8; +const FACE_SIMILARITY = 0.6; +const COVER_IMAGES = 9; +const MODEL_LABELS_PER_RUN = 12; +const EMBEDDING_BATCH = 500; + +const ThemeTitleSchema = z.object({ title: z.string().trim().min(1).max(80) }); + +const THEME_INSTRUCTIONS = + 'You name groups of photos and videos. Reply with one JSON object {"title": string}: ' + + "a short, neutral title of at most five words for what the images have in common " + + "(place type, activity, or subject). Never name or describe people's identities."; + +interface UserCollection { + clusterKey: string | null; + assetIds: string[]; +} + +/** + * Apply the user-edit rules above to computed clusters of one kind. Clusters + * that lose assets are re-keyed by their first remaining asset (the same rule + * the clustering uses), and dropped when empty or colliding with a kept key. + */ +export function reconcileUserEdits( + clusters: T[], + userCollections: UserCollection[], + prefix: "session" | "theme", +): T[] { + const splitOff = new Set( + userCollections.flatMap((collection) => (collection.clusterKey ? [] : collection.assetIds)), + ); + const computed = new Map( + clusters.map((cluster) => [ + cluster.key, + new Set(cluster.assetIds.filter((id) => !splitOff.has(id))), + ]), + ); + const frozenKeys = new Set(); + const frozenAssets = new Set(); + for (const collection of userCollections) { + if (!collection.clusterKey) continue; + const members = computed.get(collection.clusterKey); + if (!members || collection.assetIds.some((id) => !members.has(id))) { + frozenKeys.add(collection.clusterKey); + for (const id of collection.assetIds) frozenAssets.add(id); + } + } + const seen = new Set(); + const result: T[] = []; + for (const cluster of clusters) { + const assetIds = cluster.assetIds.filter((id) => !splitOff.has(id) && !frozenAssets.has(id)); + const first = assetIds[0]; + if (!first) continue; + const keep = !frozenKeys.has(cluster.key) && first === cluster.assetIds[0]; + const key = keep ? cluster.key : `${prefix}:${first}`; + if (frozenKeys.has(key) || seen.has(key)) continue; + seen.add(key); + result.push({ ...cluster, key, assetIds }); + } + return result; +} + +type ClusterRow = Awaited>[number]; + +/** Mean frame embedding per asset from the latest embeddings analysis, read in batches. */ +async function meanEmbeddings( + deps: MediaServiceDeps, + spaceId: string, + assetIds: string[], +): Promise> { + const means = new Map(); + for (let offset = 0; offset < assetIds.length; offset += EMBEDDING_BATCH) { + const batch = assetIds.slice(offset, offset + EMBEDDING_BATCH); + const rows = await deps.prisma.mediaAnalysis.findMany({ + where: { spaceId, kind: "embeddings", assetId: { in: batch }, vectors: { not: null } }, + select: { assetId: true, data: true, vectors: true }, + orderBy: [{ createdAt: "desc" }, { id: "desc" }], + }); + for (const row of rows) { + if (means.has(row.assetId) || !row.vectors) continue; + const data = row.data as { dim?: unknown } | null; + const dim = typeof data?.dim === "number" ? data.dim : 0; + if (!Number.isInteger(dim) || dim <= 0 || row.vectors.byteLength % (dim * 4) !== 0) continue; + const mean = meanVector(decodeVectors(row.vectors, dim)); + if (mean) means.set(row.assetId, mean); + } + } + return means; +} + +/** The vectors of the most common dimension (a model change mixes dimensions). */ +function sameDimension(items: T[]): T[] { + const counts = new Map(); + for (const item of items) counts.set(item.vector.length, (counts.get(item.vector.length) ?? 0) + 1); + const dim = [...counts.entries()].sort((a, b) => b[1] - a[1] || b[0] - a[0])[0]?.[0]; + return items.filter((item) => item.vector.length === dim); +} + +async function userCollections( + deps: MediaServiceDeps, + spaceId: string, + kind: "session" | "theme", +): Promise { + const rows = await deps.prisma.mediaCollection.findMany({ + where: { spaceId, kind, labelSource: "user" }, + select: { clusterKey: true, items: { select: { assetId: true }, orderBy: { position: "asc" } } }, + }); + return rows.map((row) => ({ + clusterKey: row.clusterKey, + assetIds: row.items.map((item) => item.assetId), + })); +} + +/** Whose AI consent governs model labels: the owner of the space's first media source. */ +async function libraryOwner(deps: MediaServiceDeps, spaceId: string): Promise { + const source = await deps.prisma.mediaSource.findFirst({ + where: { spaceId }, + orderBy: [{ createdAt: "asc" }, { id: "asc" }], + select: { ownerUserId: true }, + }); + if (source) return source.ownerUserId; + const space = await deps.prisma.space.findUnique({ + where: { id: spaceId }, + select: { createdByUserId: true }, + }); + return space?.createdByUserId ?? null; +} + +export interface ClusterOutcome { + sessions: number; + themes: number; + modelLabels: number; + assignedTracks: number; + suggestedPeople: number; +} + +export async function clusterSpace( + deps: MediaServiceDeps, + spaceId: string, + options: { signal?: AbortSignal } = {}, +): Promise { + const rows = await deps.repos.listAssetsForClustering(spaceId); + const extras = await deps.prisma.mediaAsset.findMany({ + where: { spaceId }, + select: { id: true, thumbPath: true, qualityScore: true }, + }); + const extra = new Map(extras.map((row) => [row.id, row])); + const embeddings = await meanEmbeddings( + deps, + spaceId, + rows.map((row) => row.id), + ); + const byId = new Map(rows.map((row) => [row.id, row])); + + const cover = (assetIds: string[]): string | null => { + const ranked = assetIds + .map((id) => extra.get(id)) + .filter((row) => row?.thumbPath) + .sort((a, b) => (b?.qualityScore ?? 0) - (a?.qualityScore ?? 0)); + return ranked[0]?.id ?? assetIds[0] ?? null; + }; + const span = (assetIds: string[]) => { + const dates = assetIds + .flatMap((id) => { + const at = byId.get(id)?.capturedAt; + return at ? [at.getTime()] : []; + }) + .sort((a, b) => a - b); + return { + startAt: dates.length > 0 ? new Date(dates[0] as number) : null, + endAt: dates.length > 0 ? new Date(dates.at(-1) as number) : null, + }; + }; + const albums = (assetIds: string[]) => assetIds.flatMap((id) => byId.get(id)?.albumNames ?? []); + + // Sessions ----------------------------------------------------------------- + const clusterAssets: ClusterAsset[] = rows.map((row: ClusterRow) => ({ + id: row.id, + kind: row.kind as ClusterAsset["kind"], + capturedAt: row.capturedAt, + gpsLat: row.gpsLat, + gpsLon: row.gpsLon, + deviceModel: row.deviceModel, + albumNames: row.albumNames, + embedding: embeddings.get(row.id) ?? null, + })); + const computedSessions = buildSessions(clusterAssets); + const centroids = new Map(computedSessions.map((session) => [session.key, session.centroid])); + const sessions: MediaClusterInput[] = reconcileUserEdits( + computedSessions.map((session) => ({ key: session.key, assetIds: session.assetIds })), + await userCollections(deps, spaceId, "session"), + "session", + ).map((session) => { + const range = span(session.assetIds); + return { + key: session.key, + title: ruleLabel({ ...range, albumNames: albums(session.assetIds) }), + labelSource: "rule", + ...range, + centroid: centroids.get(session.key) ?? null, + coverAssetId: cover(session.assetIds), + assetIds: session.assetIds, + }; + }); + options.signal?.throwIfAborted(); + await deps.repos.replaceClusters(spaceId, "session", sessions); + await publishSafe(deps, spaceId, { type: "media.collections", kind: "session" }); + + // Themes ------------------------------------------------------------------- + const recent = rows + .flatMap((row) => { + const vector = embeddings.get(row.id); + return vector ? [{ id: row.id, vector, at: row.capturedAt?.getTime() ?? 0 }] : []; + }) + .sort((a, b) => b.at - a.at || a.id.localeCompare(b.id)) + .slice(0, THEME_BATCH_LIMIT); + const items: VectorItem[] = sameDimension(recent).map(({ id, vector }) => ({ id, vector })); + const computedThemes = reconcileUserEdits( + clusterThemes(items, { similarity: THEME_SIMILARITY, minSize: THEME_MIN_SIZE }).map( + (theme) => ({ key: theme.key, assetIds: theme.ids }), + ), + await userCollections(deps, spaceId, "theme"), + "theme", + ); + const existing = await deps.prisma.mediaCollection.findMany({ + where: { spaceId, kind: "theme", clusterKey: { in: computedThemes.map((theme) => theme.key) } }, + select: { clusterKey: true, title: true, labelSource: true }, + }); + const modelTitles = new Map( + existing.flatMap((row) => + row.labelSource === "model" && row.clusterKey ? [[row.clusterKey, row.title] as const] : [], + ), + ); + const userId = await libraryOwner(deps, spaceId); + let modelCalls = 0; + let modelLabels = 0; + const themes: MediaClusterInput[] = []; + for (const theme of computedThemes) { + options.signal?.throwIfAborted(); + const range = span(theme.assetIds); + const coverAssetId = cover(theme.assetIds); + let title = modelTitles.get(theme.key) ?? null; + if (!title && userId && modelCalls < MODEL_LABELS_PER_RUN) { + modelCalls += 1; + title = await themeTitle(deps, spaceId, userId, theme.assetIds, extra, options.signal); + } + if (title) modelLabels += 1; + themes.push({ + key: theme.key, + title: title ?? ruleLabel({ ...range, albumNames: albums(theme.assetIds) }), + labelSource: title ? "model" : "rule", + ...range, + centroid: null, + coverAssetId, + assetIds: theme.assetIds, + }); + } + await deps.repos.replaceClusters(spaceId, "theme", themes); + await publishSafe(deps, spaceId, { type: "media.collections", kind: "theme" }); + + // People ------------------------------------------------------------------- + const people = await suggestPeople(deps, spaceId); + if (people.assignedTracks + people.suggestedPeople > 0) { + await publishSafe(deps, spaceId, { type: "media.people" }); + } + return { sessions: sessions.length, themes: themes.length, modelLabels, ...people }; +} + +/** Ask the model for a theme title from cover thumbnails only; null falls back to rules. */ +async function themeTitle( + deps: MediaServiceDeps, + spaceId: string, + userId: string, + assetIds: string[], + extra: Map, + signal?: AbortSignal, +): Promise { + const covers = assetIds + .flatMap((id) => { + const row = extra.get(id); + return row?.thumbPath ? [row] : []; + }) + .sort((a, b) => (b.qualityScore ?? 0) - (a.qualityScore ?? 0) || a.id.localeCompare(b.id)) + .slice(0, COVER_IMAGES); + const images: AgentInputImage[] = []; + for (const [index, row] of covers.entries()) { + try { + const file = await resolveInside(deps.mediaRoot, row.thumbPath as string); + images.push({ name: `cover-${index + 1}.jpg`, mimeType: "image/jpeg", data: await readFile(file) }); + } catch { + // A missing thumbnail only shrinks the set. + } + } + if (images.length === 0) return null; + try { + const reply = await deps.model.generate({ + spaceId, + userId, + instructions: THEME_INSTRUCTIONS, + prompt: `Name what these ${images.length} images have in common.`, + images, + schema: ThemeTitleSchema, + tier: "fast", + ...(signal ? { signal } : {}), + }); + return reply?.title ?? null; + } catch (error) { + getLogger().warn("media theme label failed", { reason: errorMessage(error) }); + return null; + } +} + +/** + * Unassigned face tracks join an unconfirmed suggestion with the same identity; + * the rest are clustered into new suggestions. Confirmed people only ever gain + * tracks through the user. + */ +async function suggestPeople( + deps: MediaServiceDeps, + spaceId: string, +): Promise<{ assignedTracks: number; suggestedPeople: number }> { + const unassigned = (await deps.repos.faceTrackEmbeddings(spaceId, { unassignedOnly: true })) + .slice(0, FACE_BATCH_LIMIT) + .flatMap((track) => { + const bytes = track.embedding; + if (bytes.byteLength === 0 || bytes.byteLength % 4 !== 0) return []; + const [vector] = decodeVectors(bytes, bytes.byteLength / 4); + return vector ? [{ id: track.trackId, vector }] : []; + }); + if (unassigned.length === 0) return { assignedTracks: 0, suggestedPeople: 0 }; + + const confirmed = new Set( + (await deps.repos.listPersonRecords(spaceId)) + .filter((person) => person.confirmed) + .map((person) => person.id), + ); + const suggestions = (await deps.repos.personCentroids(spaceId)).flatMap((person) => { + if (confirmed.has(person.personId)) return []; + const bytes = person.centroid; + if (bytes.byteLength === 0 || bytes.byteLength % 4 !== 0) return []; + const [vector] = decodeVectors(bytes, bytes.byteLength / 4); + return vector ? [{ personId: person.personId, vector }] : []; + }); + + let assignedTracks = 0; + const remaining: VectorItem[] = []; + for (const track of unassigned) { + let best: string | null = null; + let bestSimilarity = FACE_SIMILARITY; + for (const suggestion of suggestions) { + if (suggestion.vector.length !== track.vector.length) continue; + const similarity = cosineSimilarity(suggestion.vector, track.vector); + if (similarity >= bestSimilarity) { + best = suggestion.personId; + bestSimilarity = similarity; + } + } + if (best) { + await deps.repos.assignTrack(spaceId, track.id, best); + assignedTracks += 1; + } else { + remaining.push(track); + } + } + const groups = clusterFaces(sameDimension(remaining), { similarity: FACE_SIMILARITY }); + const created = await deps.repos.createPersonSuggestions( + spaceId, + groups.map((group) => ({ centroid: encodeVectors([group.centroid]), trackIds: group.ids })), + ); + return { assignedTracks, suggestedPeople: created.length }; +} diff --git a/packages/adapters/src/media-ingest-files.ts b/packages/adapters/src/media-ingest-files.ts new file mode 100644 index 0000000..de6938e --- /dev/null +++ b/packages/adapters/src/media-ingest-files.ts @@ -0,0 +1,107 @@ +import { createHash, randomUUID } from "node:crypto"; +import { createReadStream } from "node:fs"; +import { copyFile, mkdir, rename, rm, unlink, writeFile } from "node:fs/promises"; +import path from "node:path"; +import { pipeline } from "node:stream/promises"; +import type { MediaRealtimeEvent, MediaServiceDeps } from "./media-context.js"; +import { getLogger } from "@rakazo/logging"; +import { resolveInside } from "@rakazo/media/node"; +import { FOLDER_MEDIA_EXTENSIONS } from "./media-folder-source.js"; + +/* + * File helpers for media ingest. Every stored path is relative to the media + * root and resolved through resolveInside; writes go to a temp name in the + * target directory and are renamed into place. + */ + +/** Relative directory of one asset's derived files. */ +export function assetDir(spaceId: string, assetId: string): string { + return `${spaceId}/assets/${assetId}`; +} + +/** Lower-case original extension for an uploaded file name, or null when it is not a media type. */ +export function originalExtension(fileName: string): string | null { + const ext = path.extname(fileName).slice(1).toLowerCase(); + return Object.hasOwn(FOLDER_MEDIA_EXTENSIONS, ext) ? ext : null; +} + +/** Streamed SHA-256 (hex) of a file. */ +export async function sha256File(file: string, signal?: AbortSignal): Promise { + const hash = createHash("sha256"); + await pipeline(createReadStream(file), hash, { signal }); + return hash.digest("hex"); +} + +/** Delete media-root-relative files or directories; missing ones are fine. */ +export async function deleteMediaPaths(mediaRoot: string, paths: Iterable): Promise { + for (const relative of new Set(paths)) { + try { + await rm(await resolveInside(mediaRoot, relative), { recursive: true, force: true }); + } catch (error) { + getLogger().warn("media file cleanup failed", { + reason: error instanceof Error ? error.message : String(error), + }); + } + } +} + +/** A temp path next to `target` that a later rename moves into place atomically. */ +export function tempSibling(target: string, ext: string): string { + return path.join(path.dirname(target), `.tmp-${randomUUID()}${ext}`); +} + +/** Write bytes atomically: temp file in the same directory, then rename. */ +export async function writeFileAtomic(target: string, data: string | Uint8Array): Promise { + await mkdir(path.dirname(target), { recursive: true }); + const temp = tempSibling(target, path.extname(target)); + try { + await writeFile(temp, data); + await rename(temp, target); + } finally { + await rm(temp, { force: true }); + } +} + +/** Rename, falling back to copy + unlink across devices. */ +export async function moveFile(from: string, to: string): Promise { + await mkdir(path.dirname(to), { recursive: true }); + try { + await rename(from, to); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EXDEV") throw error; + const temp = tempSibling(to, path.extname(to)); + try { + await copyFile(from, temp); + await rename(temp, to); + } finally { + await rm(temp, { force: true }); + } + await unlink(from); + } +} + +/** Realtime is best effort: a failed notification never fails ingest work. */ +export async function publishSafe( + deps: Pick, + spaceId: string, + event: MediaRealtimeEvent, +): Promise { + try { + await deps.publish(spaceId, event); + } catch (error) { + getLogger().warn("media realtime publish failed", { + type: event.type, + reason: error instanceof Error ? error.message : String(error), + }); + } +} + +export function errorMessage(error: unknown): string { + return (error instanceof Error ? error.message : String(error)).slice(0, 500); +} + +/** A full disk is retried later instead of failing the asset for good. */ +export function isDiskFull(error: unknown): boolean { + if ((error as NodeJS.ErrnoException | null)?.code === "ENOSPC") return true; + return error instanceof Error && /No space left on device/i.test(error.message); +} diff --git a/packages/adapters/src/media-ingest-proxy.ts b/packages/adapters/src/media-ingest-proxy.ts new file mode 100644 index 0000000..6726ce7 --- /dev/null +++ b/packages/adapters/src/media-ingest-proxy.ts @@ -0,0 +1,214 @@ +import { mkdir, rename, rm, stat } from "node:fs/promises"; +import path from "node:path"; +import { getLogger } from "@rakazo/logging"; +import { resolveInside } from "@rakazo/media/node"; +import type { MediaToolPaths, ProbedMedia } from "./media-folder-source-probe.js"; +import { parseFfprobe, runMediaTool } from "./media-folder-source-probe.js"; +import { assetDir, tempSibling } from "./media-ingest-files.js"; + +/* + * Proxy, poster and sprite generation with ffmpeg (argv only, never a shell). + * + * Videos: 720p (shorter side) H.264 with a keyframe every second, constant + * frame rate (variable-rate sources are normalized for frame-accurate cuts), + * stereo AAC, display rotation applied by ffmpeg's autorotate, HDR (PQ/HLG) + * tone-mapped to SDR BT.709 when zscale and tonemap are available. Photos get + * a resized JPEG proxy. Posters and sprites are JPEGs written with + * `-map_metadata -1`, so no capture date, GPS, or device tags survive. + * Every output is written to a temp name and renamed into place. + */ + +const PROXY_SHORT_SIDE = 720; +const THUMB_WIDTH = 480; +const SPRITE_TILES = 10; +const SPRITE_TILE_WIDTH = 160; +const MAX_PROXY_FPS = 60; +const HDR_TRANSFERS = new Set(["smpte2084", "arib-std-b67"]); + +export interface ProxyOutputs { + proxyPath: string; + thumbPath: string; + spritePath: string | null; + probe: ProbedMedia; + hdr: boolean; + tonemapped: boolean; +} + +interface ProbeStream { + codec_type?: string; + color_transfer?: string; +} + +/** Even-dimension scale so the shorter side is at most 720 px, never upscaled. */ +const SCALE_720 = + `scale=w='if(gte(iw,ih),-2,trunc(min(${PROXY_SHORT_SIDE},iw)/2)*2)'` + + `:h='if(gte(iw,ih),trunc(min(${PROXY_SHORT_SIDE},ih)/2)*2,-2)'`; + +const TONEMAP = + "zscale=t=linear:npl=100,format=gbrpf32le,zscale=p=bt709," + + "tonemap=tonemap=hable:desat=0,zscale=t=bt709:m=bt709:r=tv"; + +let tonemapSupport: Promise | null = null; + +/** Whether this ffmpeg build has zscale and tonemap (cached per process). */ +function hasTonemapFilters(ffmpeg: string): Promise { + tonemapSupport ??= runMediaTool(ffmpeg, ["-hide_banner", "-filters"]) + .then(({ stdout }) => { + const text = stdout.toString("utf8"); + return /\szscale\s/.test(text) && /\stonemap\s/.test(text); + }) + .catch(() => false); + return tonemapSupport; +} + +export async function probeOriginal( + file: string, + tools: MediaToolPaths, + signal?: AbortSignal, +): Promise<{ probe: ProbedMedia; hdr: boolean }> { + const { stdout } = await runMediaTool( + tools.ffprobePath ?? "ffprobe", + ["-v", "error", "-print_format", "json", "-show_format", "-show_streams", "--", file], + { signal }, + ); + const json = JSON.parse(stdout.toString("utf8")) as { streams?: ProbeStream[] }; + const video = json.streams?.find((stream) => stream.codec_type === "video"); + return { + probe: parseFfprobe(json as Parameters[0]), + hdr: HDR_TRANSFERS.has(video?.color_transfer ?? ""), + }; +} + +async function ffmpegTo( + target: string, + ext: string, + build: (out: string) => string[], + tools: MediaToolPaths, + signal?: AbortSignal, +): Promise { + await mkdir(path.dirname(target), { recursive: true }); + const temp = tempSibling(target, ext); + try { + await runMediaTool(tools.ffmpegPath ?? "ffmpeg", build(temp), { signal }); + if ((await stat(temp)).size === 0) throw new Error("ffmpeg produced an empty file"); + await rename(temp, target); + } finally { + await rm(temp, { force: true }); + } +} + +function proxyFps(probe: ProbedMedia): number { + const fps = probe.fps && probe.fps > 0 ? probe.fps : 30; + return Math.min(MAX_PROXY_FPS, Math.max(1, Math.round(fps))); +} + +/** Build proxy, poster, and sprite for one original. Returns media-root-relative paths. */ +export async function buildProxy(input: { + mediaRoot: string; + spaceId: string; + assetId: string; + original: string; + still: boolean; + tools?: MediaToolPaths; + signal?: AbortSignal; +}): Promise { + const tools = input.tools ?? {}; + const signal = input.signal; + const dir = assetDir(input.spaceId, input.assetId); + const { probe, hdr } = await probeOriginal(input.original, tools, signal); + const still = input.still || !probe.hasVideo || !probe.durationMs; + const ffmpeg = tools.ffmpegPath ?? "ffmpeg"; + + let tonemapped = false; + const filters: string[] = []; + if (hdr) { + if (await hasTonemapFilters(ffmpeg)) { + filters.push(TONEMAP); + tonemapped = true; + } else { + getLogger().warn("media proxy: HDR source kept without tone mapping", { + reason: "ffmpeg lacks zscale or tonemap", + }); + } + } + filters.push(SCALE_720); + + const thumbRel = `${dir}/thumb.jpg`; + const thumb = await resolveInside(input.mediaRoot, thumbRel); + + if (still) { + const proxyRel = `${dir}/proxy.jpg`; + const proxy = await resolveInside(input.mediaRoot, proxyRel); + await ffmpegTo( + proxy, + ".jpg", + (out) => [ + ...["-v", "error", "-y", "-i", input.original, "-map", "0:v:0"], + ...["-map_metadata", "-1", "-frames:v", "1", "-update", "1"], + ...["-vf", [...filters, "format=yuvj420p"].join(","), "-q:v", "3", out], + ], + tools, + signal, + ); + await writeThumb(proxy, thumb, 0, tools, signal); + return { proxyPath: proxyRel, thumbPath: thumbRel, spritePath: null, probe, hdr, tonemapped }; + } + + const proxyRel = `${dir}/proxy.mp4`; + const proxy = await resolveInside(input.mediaRoot, proxyRel); + filters.push(`fps=${proxyFps(probe)}`, "format=yuv420p"); + await ffmpegTo( + proxy, + ".mp4", + (out) => [ + ...["-v", "error", "-y", "-i", input.original], + ...["-map", "0:v:0", "-map", "0:a:0?", "-map_metadata", "-1", "-map_chapters", "-1"], + ...["-vf", filters.join(",")], + ...["-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-profile:v", "high"], + ...["-force_key_frames", "expr:gte(t,n_forced*1)", "-sc_threshold", "0"], + ...["-c:a", "aac", "-b:a", "128k", "-ac", "2", "-ar", "48000"], + ...["-movflags", "+faststart", "-f", "mp4", out], + ], + tools, + signal, + ); + + const durationMs = probe.durationMs ?? 1000; + await writeThumb(proxy, thumb, Math.min(1000, Math.floor(durationMs / 2)), tools, signal); + + const spriteRel = `${dir}/sprite.jpg`; + const sprite = await resolveInside(input.mediaRoot, spriteRel); + const sampleFps = (SPRITE_TILES / Math.max(0.1, durationMs / 1000)).toFixed(6); + await ffmpegTo( + sprite, + ".jpg", + (out) => [ + ...["-v", "error", "-y", "-i", proxy, "-map", "0:v:0", "-map_metadata", "-1"], + ...["-vf", `fps=${sampleFps},scale=${SPRITE_TILE_WIDTH}:-2,tile=${SPRITE_TILES}x1`], + ...["-frames:v", "1", "-update", "1", "-q:v", "4", out], + ], + tools, + signal, + ); + return { proxyPath: proxyRel, thumbPath: thumbRel, spritePath: spriteRel, probe, hdr, tonemapped }; +} + +async function writeThumb( + source: string, + target: string, + atMs: number, + tools: MediaToolPaths, + signal?: AbortSignal, +): Promise { + await ffmpegTo( + target, + ".jpg", + (out) => [ + ...["-v", "error", "-y", "-ss", (atMs / 1000).toFixed(3), "-i", source], + ...["-map", "0:v:0", "-map_metadata", "-1", "-frames:v", "1", "-update", "1"], + ...["-vf", `scale='min(${THUMB_WIDTH},iw)':-2`, "-q:v", "3", out], + ], + tools, + signal, + ); +} diff --git a/packages/adapters/src/media-ingest.ts b/packages/adapters/src/media-ingest.ts new file mode 100644 index 0000000..69311c6 --- /dev/null +++ b/packages/adapters/src/media-ingest.ts @@ -0,0 +1,452 @@ +import { rename } from "node:fs/promises"; +import path from "node:path"; +import type { MediaJobHandlers, MediaSourceProvider } from "@rakazo/adapter-kit"; +import { + mediaAnalyzeJob, + mediaClusterJob, + mediaProxyJob, + mediaSyncJob, +} from "@rakazo/adapter-kit"; +import type { MediaSourceKind } from "@rakazo/contracts"; +import { CLIENT_PUSHED_MEDIA_SOURCES } from "@rakazo/contracts"; +import type { MediaSourceRecord } from "@rakazo/db"; +import { clientMediaItemToSourceItem, InvalidMediaInputError } from "@rakazo/db"; +import { getLogger } from "@rakazo/logging"; +import { assertInside, resolveInside } from "@rakazo/media/node"; +import type { MediaIngestService, MediaServiceDeps } from "./media-context.js"; +import type { MediaToolPaths } from "./media-folder-source-probe.js"; +import { analysisInputPath, analyzeAsset, MEDIA_ANALYSIS_VERSION } from "./media-ingest-analyze.js"; +import { clusterSpace } from "./media-ingest-cluster.js"; +import { + assetDir, + deleteMediaPaths, + errorMessage, + isDiskFull, + moveFile, + originalExtension, + publishSafe, + sha256File, +} from "./media-ingest-files.js"; +import { buildProxy } from "./media-ingest-proxy.js"; + +/* + * Media ingest: client pushes, uploads, and the sync → proxy → analyze → + * cluster job chain. Jobs carry only record ids; every step re-reads its + * record, scopes by the owning space, and is safe to repeat. + * + * A full sync walks every page from the start on each run: positional cursors + * (folder paths) cannot express "changed since", and per-item revisions make + * repeated pages cheap. Unchanged items enqueue nothing. Memory stays bounded + * by the page size plus the set of external ids seen. + */ + +export const MEDIA_SYNC_PAGE = 500; +/** Clustering waits for a quiet period so a burst of ingest work clusters once. */ +export const MEDIA_CLUSTER_DEBOUNCE_MS = 30_000; + +const SERVER_SOURCES = ["folder", "emulator"] as const; +const REMOVE_BATCH = 500; + +type ServerSourceKind = (typeof SERVER_SOURCES)[number]; + +function isServerSource(kind: MediaSourceKind): kind is ServerSourceKind { + return (SERVER_SOURCES as readonly string[]).includes(kind); +} + +function isClientPushed(kind: MediaSourceKind): boolean { + return (CLIENT_PUSHED_MEDIA_SOURCES as readonly string[]).includes(kind); +} + +async function scheduleCluster(deps: MediaServiceDeps, spaceId: string): Promise { + await deps.jobs.enqueue( + mediaClusterJob(spaceId, new Date(deps.now().getTime() + MEDIA_CLUSTER_DEBOUNCE_MS)), + ); +} + +/** Every derived file of a source's assets lives in their asset directories. */ +async function sourceAssetDirs( + deps: MediaServiceDeps, + spaceId: string, + sourceId: string, + externalIds?: string[], +): Promise { + const rows = await deps.prisma.mediaAsset.findMany({ + where: { spaceId, sourceId, ...(externalIds ? { externalId: { in: externalIds } } : {}) }, + select: { id: true }, + }); + return rows.map((row) => assetDir(spaceId, row.id)); +} + +/** Delete assets by external id along with every file derived from them. */ +async function removeItems( + deps: MediaServiceDeps, + spaceId: string, + sourceId: string, + externalIds: string[], +): Promise { + let deleted = 0; + for (let offset = 0; offset < externalIds.length; offset += REMOVE_BATCH) { + const batch = externalIds.slice(offset, offset + REMOVE_BATCH); + const dirs = await sourceAssetDirs(deps, spaceId, sourceId, batch); + const removed = await deps.repos.removeSourceItems(spaceId, sourceId, batch); + await deleteMediaPaths(deps.mediaRoot, [...removed.paths, ...dirs]); + deleted += removed.deleted; + } + return deleted; +} + +export function createMediaIngestService(deps: MediaServiceDeps): MediaIngestService { + return { + async pushItems(actor, input) { + const source = await deps.repos.getSource(actor.spaceId, input.sourceId); + if (!isClientPushed(source.kind)) { + throw new InvalidMediaInputError("Only device sources accept pushed items"); + } + const upserted = await deps.repos.upsertSourceItems( + actor.spaceId, + source.id, + input.items.map(clientMediaItemToSourceItem), + ); + const deleted = await removeItems(deps, actor.spaceId, source.id, [ + ...new Set(input.deletedExternalIds), + ]); + const updated = await deps.repos.updateSourceSync( + actor.spaceId, + source.id, + input.complete + ? { + status: "idle", + lastSyncCursor: input.syncToken, + lastSyncedAt: deps.now(), + error: null, + } + : { status: "syncing", error: null }, + ); + await publishSafe(deps, actor.spaceId, { + type: "media.source", + sourceId: source.id, + status: updated.status, + error: null, + }); + // Pushed assets wait for an upload; proxy work starts in uploadCompleted. + if (input.complete) await scheduleCluster(deps, actor.spaceId); + return { + created: upserted.created, + updated: upserted.updated, + unchanged: upserted.unchanged, + deleted, + }; + }, + + async requestSync(actor, sourceId) { + const source = await deps.repos.getSource(actor.spaceId, sourceId); + if (!isServerSource(source.kind)) { + throw new InvalidMediaInputError("This source syncs from its device"); + } + await deps.jobs.enqueue(mediaSyncJob(source.id)); + }, + + async uploadCompleted(actor, input) { + const ext = originalExtension(input.fileName); + if (!ext) throw new InvalidMediaInputError("Unsupported media file type"); + if (!/^[0-9a-f]{64}$/.test(input.sha256)) { + throw new InvalidMediaInputError("Invalid content hash"); + } + const asset = await deps.repos.getAssetRecord(actor.spaceId, input.assetId); + if (asset.storage === "referenced") { + throw new InvalidMediaInputError("This asset is read from its source"); + } + const temp = await assertInside(deps.mediaRoot, input.tempPath); + const uploads = await resolveInside(deps.mediaRoot, `${actor.spaceId}/uploads`); + const relative = path.relative(uploads, temp); + if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) { + throw new InvalidMediaInputError("Upload is outside this space's upload area"); + } + const actual = await sha256File(temp); + if (actual !== input.sha256) throw new InvalidMediaInputError("Content hash mismatch"); + + const storagePath = `${assetDir(actor.spaceId, asset.id)}/original.${ext}`; + const target = await resolveInside(deps.mediaRoot, storagePath); + await moveFile(temp, target); + try { + const upload = await deps.prisma.mediaUpload.findFirst({ + where: { spaceId: actor.spaceId, assetId: asset.id, status: "open", sha256: actual }, + orderBy: [{ createdAt: "desc" }, { id: "desc" }], + select: { id: true }, + }); + if (upload) { + // Closes the upload and points the asset at the stored original in one transaction. + await deps.repos.completeUpload(actor.spaceId, upload.id, { storagePath }); + } else { + const result = await deps.prisma.mediaAsset.updateMany({ + where: { id: asset.id, spaceId: actor.spaceId }, + data: { + storage: "uploaded", + storagePath, + contentHash: actual, + ingestState: "indexed", + ingestError: null, + }, + }); + if (result.count !== 1) throw new InvalidMediaInputError("Asset no longer exists"); + } + } catch (error) { + await rename(target, temp).catch(() => undefined); + throw error; + } + if (asset.storagePath && asset.storagePath !== storagePath) { + await deleteMediaPaths(deps.mediaRoot, [asset.storagePath]); + } + await publishSafe(deps, actor.spaceId, { + type: "media.asset", + assetId: asset.id, + ingestState: "indexed", + }); + await deps.jobs.enqueue(mediaProxyJob(asset.id)); + }, + + async recluster(actor) { + await deps.jobs.enqueue(mediaClusterJob(actor.spaceId)); + }, + + async removeSource(actor, sourceId) { + const source = await deps.repos.getSource(actor.spaceId, sourceId); + await deps.jobs.cancel(mediaSyncJob(source.id).replaceKey as string); + await deleteMediaPaths(deps.mediaRoot, await sourceAssetDirs(deps, actor.spaceId, source.id)); + await deps.repos.removeSource(actor.spaceId, source.id); + await scheduleCluster(deps, actor.spaceId); + }, + }; +} + +// --------------------------------------------------------------------------- +// Job handlers +// --------------------------------------------------------------------------- + +export interface MediaIngestHandlerOptions extends MediaToolPaths { + /** Aborts long-running ingest work when the worker shuts down. */ + signal?: AbortSignal; +} + +async function syncSource( + deps: MediaServiceDeps, + source: MediaSourceRecord, + provider: MediaSourceProvider, + signal?: AbortSignal, +): Promise<{ changed: number; deleted: number }> { + const seen = new Set(); + let cursor: string | null = null; + let changed = 0; + do { + signal?.throwIfAborted(); + const page = await provider.list({ + config: source.config, + cursor, + limit: MEDIA_SYNC_PAGE, + ...(signal ? { signal } : {}), + }); + for (const item of page.items) seen.add(item.externalId); + const result = await deps.repos.upsertSourceItems(source.spaceId, source.id, page.items); + for (const assetId of result.changedAssetIds) await deps.jobs.enqueue(mediaProxyJob(assetId)); + changed += result.changedAssetIds.length; + cursor = page.nextCursor; + } while (cursor); + + // Items gone from the source. An empty listing of a non-empty library is + // treated as an unavailable source (unmounted disk), never as mass deletion. + const known = await deps.prisma.mediaAsset.findMany({ + where: { spaceId: source.spaceId, sourceId: source.id }, + select: { externalId: true }, + }); + if (seen.size === 0 && known.length > 0) { + throw new Error("The source listed no files; check that it is reachable"); + } + const gone = known.map((row) => row.externalId).filter((id) => !seen.has(id)); + const deleted = await removeItems(deps, source.spaceId, source.id, gone); + return { changed, deleted }; +} + +export function createMediaIngestHandlers( + deps: MediaServiceDeps, + options: MediaIngestHandlerOptions = {}, +): Pick { + const { signal } = options; + const tools: MediaToolPaths = { + ...(options.ffmpegPath ? { ffmpegPath: options.ffmpegPath } : {}), + ...(options.ffprobePath ? { ffprobePath: options.ffprobePath } : {}), + }; + + const setAssetState = async ( + spaceId: string, + assetId: string, + from: Parameters[2]["from"], + to: Parameters[2]["to"], + error?: string, + ) => { + const applied = await deps.repos.transitionIngestState(spaceId, assetId, { + from, + to, + ...(error !== undefined ? { error } : {}), + }); + if (applied) await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: to }); + return applied; + }; + + /** Absolute path of the original, or null for a device asset that has not been uploaded. */ + const originalOf = async ( + spaceId: string, + asset: Awaited>, + ): Promise<{ path: string; referenced: boolean } | null> => { + if (asset.storage === "uploaded" && asset.storagePath) { + return { path: await resolveInside(deps.mediaRoot, asset.storagePath), referenced: false }; + } + if (asset.storage !== "referenced") return null; + const source = await deps.repos.getSource(spaceId, asset.sourceId); + const provider = isServerSource(source.kind) ? deps.sources[source.kind] : undefined; + if (!provider) throw new Error(`No ${source.kind} source provider is configured`); + return { path: await provider.resolve(source.config, asset.externalId), referenced: true }; + }; + + return { + async "media.sync"({ sourceId }) { + const spaceId = await deps.repos.mediaSpaceOf("source", sourceId); + if (!spaceId) return; + const source = await deps.repos.getSource(spaceId, sourceId); + const provider = isServerSource(source.kind) ? deps.sources[source.kind] : undefined; + if (!provider) { + const error = `No ${source.kind} source provider is configured`; + await deps.repos.updateSourceSync(spaceId, sourceId, { status: "unavailable", error }); + await publishSafe(deps, spaceId, { + type: "media.source", + sourceId, + status: "unavailable", + error, + }); + return; + } + await deps.repos.updateSourceSync(spaceId, sourceId, { status: "syncing", error: null }); + await publishSafe(deps, spaceId, { type: "media.source", sourceId, status: "syncing" }); + try { + const { changed, deleted } = await syncSource(deps, source, provider, signal); + await deps.repos.updateSourceSync(spaceId, sourceId, { + status: "idle", + lastSyncedAt: deps.now(), + error: null, + }); + await publishSafe(deps, spaceId, { type: "media.source", sourceId, status: "idle" }); + if (changed + deleted > 0) await scheduleCluster(deps, spaceId); + } catch (error) { + if (signal?.aborted) throw error; + const message = errorMessage(error); + getLogger().error("media sync failed", error, { sourceId }); + await deps.repos.updateSourceSync(spaceId, sourceId, { status: "error", error: message }); + await publishSafe(deps, spaceId, { + type: "media.source", + sourceId, + status: "error", + error: message, + }); + } + }, + + async "media.proxy"({ assetId }) { + const spaceId = await deps.repos.mediaSpaceOf("asset", assetId); + if (!spaceId) return; + const asset = await deps.repos.getAssetRecord(spaceId, assetId); + if (asset.storage === "remote") return; + const started = await setAssetState( + spaceId, + assetId, + ["indexed", "proxying", "proxied", "analyzing", "analyzed", "failed"], + "proxying", + ); + if (!started) return; + try { + const original = await originalOf(spaceId, asset); + if (!original) throw new Error("The original has not been uploaded"); + const contentHash = await sha256File(original.path, signal); + const outputs = await buildProxy({ + mediaRoot: deps.mediaRoot, + spaceId, + assetId, + original: original.path, + still: asset.kind === "photo", + tools, + ...(signal ? { signal } : {}), + }); + const still = outputs.spritePath === null; + await deps.repos.recordProxy(spaceId, assetId, { + contentHash, + proxyPath: outputs.proxyPath, + thumbPath: outputs.thumbPath, + spritePath: outputs.spritePath, + durationMs: still ? asset.durationMs : outputs.probe.durationMs, + width: outputs.probe.width, + height: outputs.probe.height, + fps: still ? asset.fps : outputs.probe.fps, + codec: outputs.probe.codec, + orientation: outputs.probe.rotation, + }); + await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: "proxied" }); + await deps.jobs.enqueue(mediaAnalyzeJob(assetId)); + } catch (error) { + if (signal?.aborted) throw error; + getLogger().error("media proxy failed", error, { assetId }); + if (isDiskFull(error)) { + await setAssetState(spaceId, assetId, ["proxying"], "failed", "Disk is full; retrying"); + throw error; + } + await setAssetState(spaceId, assetId, ["proxying"], "failed", errorMessage(error)); + } + }, + + async "media.analyze"({ assetId, kinds }) { + const inference = deps.inference; + if (!inference) return; + const spaceId = await deps.repos.mediaSpaceOf("asset", assetId); + if (!spaceId) return; + const asset = await deps.repos.getAssetRecord(spaceId, assetId); + if (!asset.proxyPath) return; + const started = await setAssetState( + spaceId, + assetId, + ["proxied", "analyzing", "analyzed", "failed"], + "analyzing", + ); + if (!started) return; + try { + const original = asset.storage === "referenced" ? await originalOf(spaceId, asset) : null; + const inputPath = await analysisInputPath({ + mediaRoot: deps.mediaRoot, + proxyPath: asset.proxyPath, + referencedOriginal: original?.path ?? null, + }); + const outcome = await analyzeAsset({ + deps, + inference, + spaceId, + asset, + inputPath, + ...(kinds ? { kinds } : {}), + }); + await deps.repos.recordAnalysisVersion( + spaceId, + assetId, + MEDIA_ANALYSIS_VERSION, + outcome.qualityScore, + ); + await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: "analyzed" }); + await scheduleCluster(deps, spaceId); + } catch (error) { + if (signal?.aborted) throw error; + getLogger().error("media analysis failed", error, { assetId }); + await setAssetState(spaceId, assetId, ["analyzing"], "failed", errorMessage(error)); + if (isDiskFull(error)) throw error; + } + }, + + async "media.cluster"({ spaceId }) { + await clusterSpace(deps, spaceId, signal ? { signal } : {}); + }, + }; +}