wip(media): w2-ingest partial work (handover, unverified)
Stopped mid-implementation for the handover to the project thread. Not typechecked or tested; do not merge as is. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0138YqDj3ttfayRwuMKKrqVj
This commit is contained in:
Claude
parent
2556ed197e
commit
55e129cb7e
633
packages/adapters/src/media-ingest-analyze.ts
Normal file
633
packages/adapters/src/media-ingest-analyze.ts
Normal file
@ -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 "<provider id>:<model>", 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<MediaAnalysisKind, MediaInferenceOperation> = {
|
||||
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<MediaAnalysisKind> = 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/<track>.json`. */
|
||||
export const FaceKeyframesFileSchema = z.object({
|
||||
sampleFps: z.number().positive(),
|
||||
keyframes: z.array(BoxKeyframeSchema.extend({ confidence: z.number() })),
|
||||
});
|
||||
export type FaceKeyframesFile = z.infer<typeof FaceKeyframesFileSchema>;
|
||||
|
||||
/** Read a face track's keyframes file (path relative to the media root). */
|
||||
export async function readFaceKeyframes(
|
||||
mediaRoot: string,
|
||||
keyframesPath: string,
|
||||
): Promise<FaceKeyframesFile> {
|
||||
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<TranscribeResult, "language" | "words"> | null;
|
||||
faceSpans: Array<{ startMs: number; endMs: number }> | null;
|
||||
embeddings: { timesMs: number[]; vectors: Float32Array[] } | null;
|
||||
audio: Pick<AudioFeaturesResult, "beatsMs" | "energy"> | 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<Record<TeaserFeature, number>> = {};
|
||||
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<ReturnType<MediaServiceDeps["repos"]["getAssetRecord"]>>;
|
||||
|
||||
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<T>(fn: () => Promise<T>): Promise<T | null> {
|
||||
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<AnalysisBundle> {
|
||||
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<AnalyzeOutcome> {
|
||||
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 <T>(kind: MediaAnalysisKind, fn: () => Promise<T>) => {
|
||||
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<string> {
|
||||
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);
|
||||
}
|
||||
415
packages/adapters/src/media-ingest-cluster.ts
Normal file
415
packages/adapters/src/media-ingest-cluster.ts
Normal file
@ -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<T extends { key: string; assetIds: string[] }>(
|
||||
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<string>();
|
||||
const frozenAssets = new Set<string>();
|
||||
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<string>();
|
||||
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<ReturnType<MediaServiceDeps["repos"]["listAssetsForClustering"]>>[number];
|
||||
|
||||
/** Mean frame embedding per asset from the latest embeddings analysis, read in batches. */
|
||||
async function meanEmbeddings(
|
||||
deps: MediaServiceDeps,
|
||||
spaceId: string,
|
||||
assetIds: string[],
|
||||
): Promise<Map<string, Float32Array>> {
|
||||
const means = new Map<string, Float32Array>();
|
||||
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<T extends { vector: Float32Array }>(items: T[]): T[] {
|
||||
const counts = new Map<number, number>();
|
||||
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<UserCollection[]> {
|
||||
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<string | null> {
|
||||
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<ClusterOutcome> {
|
||||
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<string, { id: string; thumbPath: string | null; qualityScore: number | null }>,
|
||||
signal?: AbortSignal,
|
||||
): Promise<string | null> {
|
||||
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 };
|
||||
}
|
||||
107
packages/adapters/src/media-ingest-files.ts
Normal file
107
packages/adapters/src/media-ingest-files.ts
Normal file
@ -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<string> {
|
||||
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<string>): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<MediaServiceDeps, "publish">,
|
||||
spaceId: string,
|
||||
event: MediaRealtimeEvent,
|
||||
): Promise<void> {
|
||||
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);
|
||||
}
|
||||
214
packages/adapters/src/media-ingest-proxy.ts
Normal file
214
packages/adapters/src/media-ingest-proxy.ts
Normal file
@ -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<boolean> | null = null;
|
||||
|
||||
/** Whether this ffmpeg build has zscale and tonemap (cached per process). */
|
||||
function hasTonemapFilters(ffmpeg: string): Promise<boolean> {
|
||||
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<typeof parseFfprobe>[0]),
|
||||
hdr: HDR_TRANSFERS.has(video?.color_transfer ?? ""),
|
||||
};
|
||||
}
|
||||
|
||||
async function ffmpegTo(
|
||||
target: string,
|
||||
ext: string,
|
||||
build: (out: string) => string[],
|
||||
tools: MediaToolPaths,
|
||||
signal?: AbortSignal,
|
||||
): Promise<void> {
|
||||
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<ProxyOutputs> {
|
||||
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<void> {
|
||||
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,
|
||||
);
|
||||
}
|
||||
452
packages/adapters/src/media-ingest.ts
Normal file
452
packages/adapters/src/media-ingest.ts
Normal file
@ -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<void> {
|
||||
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<string[]> {
|
||||
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<number> {
|
||||
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<string>();
|
||||
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<MediaJobHandlers, "media.sync" | "media.proxy" | "media.analyze" | "media.cluster"> {
|
||||
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<MediaServiceDeps["repos"]["transitionIngestState"]>[2]["from"],
|
||||
to: Parameters<MediaServiceDeps["repos"]["transitionIngestState"]>[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<ReturnType<MediaServiceDeps["repos"]["getAssetRecord"]>>,
|
||||
): 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 } : {});
|
||||
},
|
||||
};
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user