import type { MediaSourceItem } from "@rakazo/adapter-kit";
import type {
  MediaAnalysisKind,
  MediaAsset,
  MediaAssetQuery,
  MediaFaceTrack,
  MediaIngestState,
  MediaSegment,
  MediaSegmentKind,
  MediaSource,
  MediaSourceKind,
  MediaSourceStatus,
  TeaserFeature,
} from "@rakazo/contracts";
import { CLIENT_PUSHED_MEDIA_SOURCES } from "@rakazo/contracts";
import { roundCoordinate } from "@rakazo/media";
import type {
  MediaAnalysis as MediaAnalysisRow,
  MediaAsset as MediaAssetRow,
  MediaSource as MediaSourceRow,
  PrismaClient,
} from "./client.js";
import { Prisma } from "./client.js";
import type { MediaPage, MediaPageInput } from "./media-shared.js";
import {
  decodeKeysetCursor,
  InvalidMediaInputError,
  keysetAfterSql,
  likeContains,
  lockMediaKey,
  mapMediaAsset,
  mapMediaFaceTrack,
  mapMediaSegment,
  mapMediaSource,
  pageLimit,
  toBytes,
  toPage,
} from "./media-shared.js";
import { IsolationError } from "./scope.js";

export interface NewMediaSource {
  kind: MediaSourceKind;
  label: string;
  ownerUserId: string;
  /** Folder: { path }. Client-pushed: { deviceId }. Never secrets. */
  config: Record<string, unknown>;
}

/** Source row for the sync worker, including its config and cursor. */
export interface MediaSourceRecord extends MediaSource {
  spaceId: string;
  ownerUserId: string;
  config: Record<string, unknown>;
  lastSyncCursor: string | null;
}

export interface MediaSourceSyncPatch {
  status?: MediaSourceStatus;
  lastSyncCursor?: string | null;
  lastSyncedAt?: Date | null;
  error?: string | null;
}

export interface UpsertSourceItemsResult {
  created: number;
  updated: number;
  unchanged: number;
  /** New assets and assets whose revision changed, in input order. */
  changedAssetIds: string[];
}

export type MediaAssetSearch = Omit<MediaAssetQuery, "cursor" | "limit"> & MediaPageInput;

export interface MediaAssetDetail {
  asset: MediaAsset;
  segments: MediaSegment[];
  faceTracks: MediaFaceTrack[];
  transcript: string | null;
}

export interface MediaProxyResult {
  contentHash: string;
  proxyPath: string | null;
  thumbPath: string | null;
  spritePath?: string | null;
  durationMs?: number | null;
  width?: number | null;
  height?: number | null;
  fps?: number | null;
  codec?: string | null;
  orientation?: number | null;
}

export interface NewMediaAnalysis {
  kind: MediaAnalysisKind;
  modelId: string;
  modelVersion: string;
  data?: Prisma.InputJsonValue | null;
  dataPath?: string | null;
  /** Float32 little-endian vectors (see encodeVectors in @rakazo/media). */
  vectors?: Uint8Array | null;
}

export interface NewMediaSegment {
  kind: MediaSegmentKind;
  startMs: number;
  endMs: number;
  scores?: Partial<Record<TeaserFeature, number>>;
  teaserScore?: number | null;
  transcriptExcerpt?: string | null;
  labels?: string[];
  paidOnly?: boolean;
}

export type MediaSpaceLookup = "source" | "asset" | "render" | "publication";

const EARTH_RADIUS_KM = 6371.0088;
const KM_PER_DEGREE = 111.32;

function sourceRecord(row: MediaSourceRow, assetCount: number): MediaSourceRecord {
  const config =
    row.config && typeof row.config === "object" && !Array.isArray(row.config)
      ? (row.config as Record<string, unknown>)
      : {};
  return {
    ...mapMediaSource(row, assetCount),
    spaceId: row.spaceId,
    ownerUserId: row.ownerUserId,
    config,
    lastSyncCursor: row.lastSyncCursor,
  };
}

function rounded(value: number | null): number | null {
  return value === null ? null : roundCoordinate(value);
}

function itemData(item: MediaSourceItem) {
  return {
    kind: item.kind,
    revision: item.revision,
    capturedAt: item.capturedAt,
    durationMs: item.durationMs,
    width: item.width,
    height: item.height,
    fps: item.fps,
    gpsLat: rounded(item.gpsLat),
    gpsLon: rounded(item.gpsLon),
    deviceModel: item.deviceModel,
    albumNames: item.albumNames,
    isFavorite: item.isFavorite,
    fileName: item.fileName,
    sizeBytes: item.sizeBytes === null ? null : BigInt(item.sizeBytes),
  };
}

/** Plain text of a stored transcript analysis (TranscribeResult words or a text field). */
function transcriptText(data: Prisma.JsonValue | null): string | null {
  if (!data || typeof data !== "object" || Array.isArray(data)) return null;
  if (typeof data.text === "string") return data.text;
  if (!Array.isArray(data.words)) return null;
  const words = data.words
    .map((word) =>
      word && typeof word === "object" && !Array.isArray(word) && typeof word.text === "string"
        ? word.text.trim()
        : "",
    )
    .filter(Boolean);
  return words.length > 0 ? words.join(" ") : null;
}

function nearSql(near: NonNullable<MediaAssetQuery["near"]>): Prisma.Sql {
  const { lat, lon, radiusKm } = near;
  const dLat = radiusKm / KM_PER_DEGREE;
  const conditions: Prisma.Sql[] = [
    Prisma.sql`a."gpsLat" IS NOT NULL AND a."gpsLon" IS NOT NULL`,
    Prisma.sql`a."gpsLat" BETWEEN ${lat - dLat}::float8 AND ${lat + dLat}::float8`,
  ];
  // Longitude prefilter only where it is a simple band; near the poles the haversine alone decides.
  const maxAbsLat = Math.min(90, Math.abs(lat) + dLat);
  const cos = Math.cos((maxAbsLat * Math.PI) / 180);
  if (cos > 0.01) {
    const dLon = radiusKm / (KM_PER_DEGREE * cos);
    if (dLon < 180) {
      const west = lon - dLon;
      const east = lon + dLon;
      if (west < -180) {
        conditions.push(
          Prisma.sql`(a."gpsLon" >= ${west + 360}::float8 OR a."gpsLon" <= ${east}::float8)`,
        );
      } else if (east > 180) {
        conditions.push(
          Prisma.sql`(a."gpsLon" >= ${west}::float8 OR a."gpsLon" <= ${east - 360}::float8)`,
        );
      } else {
        conditions.push(Prisma.sql`a."gpsLon" BETWEEN ${west}::float8 AND ${east}::float8`);
      }
    }
  }
  conditions.push(Prisma.sql`
    2 * ${EARTH_RADIUS_KM}::float8 * asin(least(1, sqrt(
      power(sin(radians(a."gpsLat" - ${lat}::float8) / 2), 2)
      + cos(radians(${lat}::float8)) * cos(radians(a."gpsLat"))
        * power(sin(radians(a."gpsLon" - ${lon}::float8) / 2), 2)
    ))) <= ${radiusKm}::float8`);
  return Prisma.join(conditions, " AND ");
}

function textSql(spaceId: string, text: string): Prisma.Sql {
  const pattern = likeContains(text);
  return Prisma.sql`(
    a."fileName" ILIKE ${pattern}
    OR EXISTS (SELECT 1 FROM unnest(a."albumNames") AS album(name) WHERE album.name ILIKE ${pattern})
    OR EXISTS (
      SELECT 1 FROM media_collection_items ci
      JOIN media_collections c ON c.id = ci."collectionId"
      WHERE ci."assetId" = a.id AND c."spaceId" = ${spaceId} AND c.title ILIKE ${pattern}
    )
    OR EXISTS (
      SELECT 1 FROM media_segments s
      WHERE s."assetId" = a.id AND s."spaceId" = ${spaceId}
        AND (
          s."transcriptExcerpt" ILIKE ${pattern}
          OR EXISTS (SELECT 1 FROM unnest(s.labels) AS label(name) WHERE label.name ILIKE ${pattern})
        )
    )
  )`;
}

export function createMediaLibraryRepos(prisma: PrismaClient) {
  async function requireAsset(
    db: Prisma.TransactionClient | PrismaClient,
    spaceId: string,
    assetId: string,
  ): Promise<MediaAssetRow> {
    const asset = await db.mediaAsset.findFirst({ where: { id: assetId, spaceId } });
    if (!asset) throw new IsolationError();
    return asset;
  }

  async function updateAsset(
    spaceId: string,
    assetId: string,
    data: Prisma.MediaAssetUpdateManyMutationInput,
  ): Promise<MediaAsset> {
    const result = await prisma.mediaAsset.updateMany({ where: { id: assetId, spaceId }, data });
    if (result.count !== 1) throw new IsolationError();
    return mapMediaAsset(await requireAsset(prisma, spaceId, assetId));
  }

  // -------------------------------------------------------------------------
  // Sources
  // -------------------------------------------------------------------------

  async function createSource(spaceId: string, input: NewMediaSource): Promise<MediaSource> {
    const row = await prisma.mediaSource.create({
      data: {
        spaceId,
        kind: input.kind,
        label: input.label,
        ownerUserId: input.ownerUserId,
        config: input.config as Prisma.InputJsonObject,
      },
    });
    return mapMediaSource(row, 0);
  }

  async function listSources(spaceId: string): Promise<MediaSource[]> {
    const rows = await prisma.mediaSource.findMany({
      where: { spaceId },
      orderBy: [{ createdAt: "asc" }, { id: "asc" }],
      include: { _count: { select: { assets: true } } },
    });
    return rows.map((row) => mapMediaSource(row, row._count.assets));
  }

  async function getSource(spaceId: string, sourceId: string): Promise<MediaSourceRecord> {
    const row = await prisma.mediaSource.findFirst({
      where: { id: sourceId, spaceId },
      include: { _count: { select: { assets: true } } },
    });
    if (!row) throw new IsolationError();
    return sourceRecord(row, row._count.assets);
  }

  async function updateSourceSync(
    spaceId: string,
    sourceId: string,
    patch: MediaSourceSyncPatch,
  ): Promise<MediaSourceRecord> {
    const result = await prisma.mediaSource.updateMany({
      where: { id: sourceId, spaceId },
      data: {
        ...(patch.status !== undefined ? { status: patch.status } : {}),
        ...(patch.lastSyncCursor !== undefined ? { lastSyncCursor: patch.lastSyncCursor } : {}),
        ...(patch.lastSyncedAt !== undefined ? { lastSyncedAt: patch.lastSyncedAt } : {}),
        ...(patch.error !== undefined ? { error: patch.error } : {}),
      },
    });
    if (result.count !== 1) throw new IsolationError();
    return getSource(spaceId, sourceId);
  }

  /** Deletes the source and, by cascade, its assets and everything derived from them. */
  async function removeSource(spaceId: string, sourceId: string): Promise<void> {
    const result = await prisma.mediaSource.deleteMany({ where: { id: sourceId, spaceId } });
    if (result.count !== 1) throw new IsolationError();
  }

  /**
   * Delete assets a device reported as removed. Returns the media-root-relative files
   * the caller must delete (originals, proxies, thumbnails, face keyframes, masks).
   */
  async function removeSourceItems(
    spaceId: string,
    sourceId: string,
    externalIds: string[],
  ): Promise<{ deleted: number; paths: string[] }> {
    if (externalIds.length === 0) return { deleted: 0, paths: [] };
    return prisma.$transaction(async (tx) => {
      const source = await tx.mediaSource.findFirst({ where: { id: sourceId, spaceId } });
      if (!source) throw new IsolationError();
      const assets = await tx.mediaAsset.findMany({
        where: { spaceId, sourceId, externalId: { in: externalIds } },
        select: {
          id: true,
          storagePath: true,
          proxyPath: true,
          thumbPath: true,
          spritePath: true,
          faceTracks: { select: { keyframesPath: true } },
          marks: { select: { maskDir: true } },
        },
      });
      const paths = assets.flatMap((asset) => [
        ...[asset.storagePath, asset.proxyPath, asset.thumbPath, asset.spritePath],
        ...asset.faceTracks.map((track) => track.keyframesPath),
        ...asset.marks.map((mark) => mark.maskDir),
      ]);
      const result = await tx.mediaAsset.deleteMany({
        where: { spaceId, id: { in: assets.map((asset) => asset.id) } },
      });
      return {
        deleted: result.count,
        paths: paths.filter((path): path is string => typeof path === "string" && path !== ""),
      };
    });
  }

  /**
   * Worker entry point: jobs carry only a record id. Returns the owning space so every
   * further call stays scoped, or null when the record no longer exists.
   */
  async function mediaSpaceOf(kind: MediaSpaceLookup, id: string): Promise<string | null> {
    const select = { spaceId: true } as const;
    const row =
      kind === "source"
        ? await prisma.mediaSource.findUnique({ where: { id }, select })
        : kind === "asset"
          ? await prisma.mediaAsset.findUnique({ where: { id }, select })
          : kind === "render"
            ? await prisma.mediaRender.findUnique({ where: { id }, select })
            : await prisma.mediaPublication.findUnique({ where: { id }, select });
    return row?.spaceId ?? null;
  }

  // -------------------------------------------------------------------------
  // Assets
  // -------------------------------------------------------------------------

  /**
   * Index one page of source items keyed by (sourceId, externalId). An item whose
   * revision matches the stored one is unchanged; syncs of the same source are serialized.
   */
  async function upsertSourceItems(
    spaceId: string,
    sourceId: string,
    items: MediaSourceItem[],
  ): Promise<UpsertSourceItemsResult> {
    const unique = new Map<string, MediaSourceItem>();
    for (const item of items) unique.set(item.externalId, item);
    return prisma.$transaction(
      async (tx) => {
        const source = await tx.mediaSource.findFirst({
          where: { id: sourceId, spaceId },
          select: { kind: true },
        });
        if (!source) throw new IsolationError();
        await lockMediaKey(tx, `source:${sourceId}`);
        const storage = (CLIENT_PUSHED_MEDIA_SOURCES as readonly string[]).includes(source.kind)
          ? "remote"
          : "referenced";
        const existing = await tx.mediaAsset.findMany({
          where: { sourceId, externalId: { in: [...unique.keys()] } },
          select: { id: true, externalId: true, revision: true },
        });
        const byExternalId = new Map(existing.map((row) => [row.externalId, row]));
        const toCreate: MediaSourceItem[] = [];
        const changed = new Map<string, string>();
        let unchanged = 0;
        for (const item of unique.values()) {
          const current = byExternalId.get(item.externalId);
          if (!current) {
            toCreate.push(item);
          } else if (current.revision === item.revision) {
            unchanged += 1;
          } else {
            await tx.mediaAsset.update({ where: { id: current.id }, data: itemData(item) });
            changed.set(item.externalId, current.id);
          }
        }
        const updated = changed.size;
        if (toCreate.length > 0) {
          const created = await tx.mediaAsset.createManyAndReturn({
            data: toCreate.map((item) => ({
              ...itemData(item),
              spaceId,
              sourceId,
              externalId: item.externalId,
              storage,
              ingestState: storage === "remote" ? "awaiting_upload" : "indexed",
            })),
            select: { id: true, externalId: true },
          });
          for (const row of created) changed.set(row.externalId, row.id);
        }
        return {
          created: toCreate.length,
          updated,
          unchanged,
          changedAssetIds: [...unique.keys()].flatMap((externalId) => {
            const id = changed.get(externalId);
            return id ? [id] : [];
          }),
        };
      },
      { timeout: 60_000 },
    );
  }

  /**
   * Keyset-paginated search, newest capture first, undated assets last. The cursor is
   * the (capturedAt, id) of the previous page's last row, so inserts never shift pages.
   */
  async function searchAssets(
    spaceId: string,
    query: MediaAssetSearch,
  ): Promise<MediaPage<MediaAsset>> {
    const limit = pageLimit(query.limit);
    const cursor = query.cursor ? decodeKeysetCursor(query.cursor) : null;
    const conditions: Prisma.Sql[] = [Prisma.sql`a."spaceId" = ${spaceId}`];
    if (query.sourceId) conditions.push(Prisma.sql`a."sourceId" = ${query.sourceId}`);
    if (query.kinds && query.kinds.length > 0) {
      conditions.push(Prisma.sql`a.kind IN (${Prisma.join(query.kinds)})`);
    }
    if (query.ingestStates && query.ingestStates.length > 0) {
      conditions.push(Prisma.sql`a."ingestState" IN (${Prisma.join(query.ingestStates)})`);
    }
    if (query.from) conditions.push(Prisma.sql`a."capturedAt" >= ${new Date(query.from)}`);
    if (query.to) conditions.push(Prisma.sql`a."capturedAt" <= ${new Date(query.to)}`);
    if (query.favoritesOnly) conditions.push(Prisma.sql`a."isFavorite" = TRUE`);
    if (query.minQuality !== undefined) {
      conditions.push(Prisma.sql`a."qualityScore" >= ${query.minQuality}::float8`);
    }
    if (query.near) conditions.push(nearSql(query.near));
    if (query.personId) {
      conditions.push(Prisma.sql`EXISTS (
        SELECT 1 FROM media_face_tracks ft
        WHERE ft."assetId" = a.id AND ft."spaceId" = ${spaceId} AND ft."personId" = ${query.personId}
      )`);
    }
    if (query.collectionId) {
      conditions.push(Prisma.sql`EXISTS (
        SELECT 1 FROM media_collection_items ci
        JOIN media_collections c ON c.id = ci."collectionId"
        WHERE ci."assetId" = a.id AND c.id = ${query.collectionId} AND c."spaceId" = ${spaceId}
      )`);
    }
    const text = query.text?.trim();
    if (text) conditions.push(textSql(spaceId, text));
    conditions.push(keysetAfterSql(cursor, Prisma.sql`a."capturedAt"`, Prisma.sql`a.id`));

    const keys = await prisma.$queryRaw<Array<{ id: string }>>(Prisma.sql`
      SELECT a.id FROM media_assets a
      WHERE ${Prisma.join(conditions, " AND ")}
      ORDER BY a."capturedAt" DESC NULLS LAST, a.id DESC
      LIMIT ${limit + 1}
    `);
    const rows = await prisma.mediaAsset.findMany({
      where: { spaceId, id: { in: keys.map((key) => key.id) } },
    });
    const byId = new Map(rows.map((row) => [row.id, row]));
    const ordered = keys.flatMap((key) => {
      const row = byId.get(key.id);
      return row ? [row] : [];
    });
    return toPage(ordered, limit, (row) => ({ at: row.capturedAt, id: row.id }), mapMediaAsset);
  }

  async function getAsset(spaceId: string, assetId: string): Promise<MediaAssetDetail> {
    const asset = await requireAsset(prisma, spaceId, assetId);
    const [segments, faceTracks, transcript] = await Promise.all([
      prisma.mediaSegment.findMany({
        where: { assetId, spaceId },
        orderBy: [{ startMs: "asc" }, { id: "asc" }],
      }),
      prisma.mediaFaceTrack.findMany({
        where: { assetId, spaceId },
        orderBy: [{ startMs: "asc" }, { id: "asc" }],
      }),
      prisma.mediaAnalysis.findFirst({
        where: { assetId, spaceId, kind: "transcript" },
        orderBy: [{ createdAt: "desc" }, { id: "desc" }],
        select: { data: true },
      }),
    ]);
    return {
      asset: mapMediaAsset(asset),
      segments: segments.map(mapMediaSegment),
      faceTracks: faceTracks.map(mapMediaFaceTrack),
      transcript: transcriptText(transcript?.data ?? null),
    };
  }

  /** Full row for workers: storage path, hashes, derived file paths. */
  async function getAssetRecord(spaceId: string, assetId: string): Promise<MediaAssetRow> {
    return requireAsset(prisma, spaceId, assetId);
  }

  async function listAssetRecords(spaceId: string, assetIds: string[]): Promise<MediaAssetRow[]> {
    if (assetIds.length === 0) return [];
    return prisma.mediaAsset.findMany({
      where: { spaceId, id: { in: [...new Set(assetIds)] } },
      orderBy: [{ capturedAt: { sort: "asc", nulls: "last" } }, { id: "asc" }],
    });
  }

  /** Every asset of the space in capture order, with the fields clustering reads. */
  async function listAssetsForClustering(spaceId: string) {
    return prisma.mediaAsset.findMany({
      where: { spaceId },
      orderBy: [{ capturedAt: { sort: "asc", nulls: "last" } }, { id: "asc" }],
      select: {
        id: true,
        kind: true,
        capturedAt: true,
        durationMs: true,
        gpsLat: true,
        gpsLon: true,
        deviceModel: true,
        albumNames: true,
        contentHash: true,
      },
    });
  }

  /**
   * Atomic ingest state change: only applies when the asset is currently in one of
   * `from`, so two workers cannot both take the same step. Returns whether it applied.
   */
  async function transitionIngestState(
    spaceId: string,
    assetId: string,
    transition: { from: MediaIngestState[]; to: MediaIngestState; error?: string | null },
  ): Promise<boolean> {
    const result = await prisma.mediaAsset.updateMany({
      where: { id: assetId, spaceId, ingestState: { in: transition.from } },
      data: {
        ingestState: transition.to,
        ingestError: transition.to === "failed" ? (transition.error ?? null) : null,
      },
    });
    if (result.count === 1) return true;
    await requireAsset(prisma, spaceId, assetId);
    return false;
  }

  async function recordProxy(
    spaceId: string,
    assetId: string,
    proxy: MediaProxyResult,
  ): Promise<MediaAsset> {
    return updateAsset(spaceId, assetId, {
      contentHash: proxy.contentHash,
      proxyPath: proxy.proxyPath,
      thumbPath: proxy.thumbPath,
      ...(proxy.spritePath !== undefined ? { spritePath: proxy.spritePath } : {}),
      ...(proxy.durationMs !== undefined ? { durationMs: proxy.durationMs } : {}),
      ...(proxy.width !== undefined ? { width: proxy.width } : {}),
      ...(proxy.height !== undefined ? { height: proxy.height } : {}),
      ...(proxy.fps !== undefined ? { fps: proxy.fps } : {}),
      ...(proxy.codec !== undefined ? { codec: proxy.codec } : {}),
      ...(proxy.orientation !== undefined ? { orientation: proxy.orientation } : {}),
      ingestState: "proxied",
      ingestError: null,
    });
  }

  async function recordAnalysisVersion(
    spaceId: string,
    assetId: string,
    analysisVersion: number,
    qualityScore?: number | null,
  ): Promise<MediaAsset> {
    return updateAsset(spaceId, assetId, {
      analysisVersion,
      ...(qualityScore !== undefined ? { qualityScore } : {}),
      ingestState: "analyzed",
      ingestError: null,
    });
  }

  /** Remote assets move to awaiting_upload; returns the ids that now wait for bytes. */
  async function requestUpload(spaceId: string, assetIds: string[]): Promise<string[]> {
    const ids = [...new Set(assetIds)];
    const owned = await prisma.mediaAsset.count({ where: { spaceId, id: { in: ids } } });
    if (owned !== ids.length) throw new IsolationError();
    await prisma.mediaAsset.updateMany({
      where: {
        spaceId,
        id: { in: ids },
        storage: "remote",
        ingestState: { in: ["indexed", "failed"] },
      },
      data: { ingestState: "awaiting_upload", ingestError: null },
    });
    const awaiting = await prisma.mediaAsset.findMany({
      where: { spaceId, id: { in: ids }, ingestState: "awaiting_upload" },
      select: { id: true },
    });
    const set = new Set(awaiting.map((row) => row.id));
    return ids.filter((id) => set.has(id));
  }

  async function assetsNeedingWork(
    spaceId: string,
    state: MediaIngestState,
    limit: number,
  ): Promise<MediaAssetRow[]> {
    return prisma.mediaAsset.findMany({
      where: { spaceId, ingestState: state },
      orderBy: [{ createdAt: "asc" }, { id: "asc" }],
      take: pageLimit(limit),
    });
  }

  // -------------------------------------------------------------------------
  // Analyses
  // -------------------------------------------------------------------------

  /** Insert or replace the result for (asset, kind, modelVersion); other versions stay. */
  async function upsertAnalysis(
    spaceId: string,
    assetId: string,
    analysis: NewMediaAnalysis,
  ): Promise<MediaAnalysisRow> {
    await requireAsset(prisma, spaceId, assetId);
    const data = {
      modelId: analysis.modelId,
      data: analysis.data ?? Prisma.DbNull,
      dataPath: analysis.dataPath ?? null,
      vectors: analysis.vectors ? toBytes(analysis.vectors) : null,
    };
    return prisma.mediaAnalysis.upsert({
      where: {
        assetId_kind_modelVersion: {
          assetId,
          kind: analysis.kind,
          modelVersion: analysis.modelVersion,
        },
      },
      create: {
        ...data,
        spaceId,
        assetId,
        kind: analysis.kind,
        modelVersion: analysis.modelVersion,
      },
      update: data,
    });
  }

  async function latestAnalysis(
    spaceId: string,
    assetId: string,
    kind: MediaAnalysisKind,
  ): Promise<MediaAnalysisRow | null> {
    await requireAsset(prisma, spaceId, assetId);
    return prisma.mediaAnalysis.findFirst({
      where: { spaceId, assetId, kind },
      orderBy: [{ createdAt: "desc" }, { id: "desc" }],
    });
  }

  /** Latest analysis of one kind per asset across the space (e.g. embeddings for clustering). */
  async function latestAnalysesByKind(
    spaceId: string,
    kind: MediaAnalysisKind,
  ): Promise<Map<string, MediaAnalysisRow>> {
    const rows = await prisma.mediaAnalysis.findMany({
      where: { spaceId, kind },
      orderBy: [{ assetId: "asc" }, { createdAt: "desc" }, { id: "desc" }],
    });
    const latest = new Map<string, MediaAnalysisRow>();
    for (const row of rows) if (!latest.has(row.assetId)) latest.set(row.assetId, row);
    return latest;
  }

  // -------------------------------------------------------------------------
  // Segments
  // -------------------------------------------------------------------------

  /** Replace every segment of the asset with the result of one analysis version. */
  async function replaceSegments(
    spaceId: string,
    assetId: string,
    analysisVersion: number,
    segments: NewMediaSegment[],
  ): Promise<number> {
    for (const segment of segments) {
      if (segment.endMs <= segment.startMs || segment.startMs < 0) {
        throw new InvalidMediaInputError("Segment range is empty");
      }
    }
    return prisma.$transaction(async (tx) => {
      await requireAsset(tx, spaceId, assetId);
      await lockMediaKey(tx, `segments:${assetId}`);
      await tx.mediaSegment.deleteMany({ where: { assetId, spaceId } });
      const created = await tx.mediaSegment.createMany({
        data: segments.map((segment) => ({
          spaceId,
          assetId,
          analysisVersion,
          kind: segment.kind,
          startMs: segment.startMs,
          endMs: segment.endMs,
          scores: segment.scores ?? {},
          teaserScore: segment.teaserScore ?? null,
          transcriptExcerpt: segment.transcriptExcerpt ?? null,
          labels: segment.labels ?? [],
          paidOnly: segment.paidOnly ?? false,
        })),
      });
      return created.count;
    });
  }

  async function listSegments(spaceId: string, assetIds: string[]): Promise<MediaSegment[]> {
    if (assetIds.length === 0) return [];
    const rows = await prisma.mediaSegment.findMany({
      where: { spaceId, assetId: { in: [...new Set(assetIds)] } },
      orderBy: [{ assetId: "asc" }, { startMs: "asc" }, { id: "asc" }],
    });
    return rows.map(mapMediaSegment);
  }

  /** All or nothing: any id outside the space fails the whole call. */
  async function setSegmentsPaidOnly(
    spaceId: string,
    segmentIds: string[],
    paidOnly: boolean,
  ): Promise<number> {
    const ids = [...new Set(segmentIds)];
    return prisma.$transaction(async (tx) => {
      const result = await tx.mediaSegment.updateMany({
        where: { spaceId, id: { in: ids } },
        data: { paidOnly },
      });
      if (result.count !== ids.length) throw new IsolationError();
      return result.count;
    });
  }

  return {
    createSource,
    listSources,
    getSource,
    updateSourceSync,
    removeSource,
    removeSourceItems,
    mediaSpaceOf,
    upsertSourceItems,
    searchAssets,
    getAsset,
    getAssetRecord,
    listAssetRecords,
    listAssetsForClustering,
    transitionIngestState,
    recordProxy,
    recordAnalysisVersion,
    requestUpload,
    assetsNeedingWork,
    upsertAnalysis,
    latestAnalysis,
    latestAnalysesByKind,
    replaceSegments,
    listSegments,
    setSegmentsPaidOnly,
  };
}