import { lstat, mkdir, rename } from "node:fs/promises";
import path from "node:path";
import type { BackgroundJob, MediaJobHandlers, MediaSourceProvider } from "@rakazo/adapter-kit";
import { mediaAnalyzeJob, mediaClusterJob, mediaProxyJob, mediaSyncJob } from "@rakazo/adapter-kit";
import type { MediaIngestState, MediaSourceKind } from "@rakazo/contracts";
import { CLIENT_PUSHED_MEDIA_SOURCES } from "@rakazo/contracts";
import type { MediaSourceRecord } from "@rakazo/db";
import { clientMediaItemToSourceItem, InvalidMediaInputError, IsolationError } from "@rakazo/db";
import { getLogger } from "@rakazo/logging";
import type { SessionOptions } from "@rakazo/media";
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,
  isTransientInferenceError,
  MEDIA_ANALYSIS_VERSION,
} from "./media-ingest-analyze.js";
import { clusterSpace } from "./media-ingest-cluster.js";
import {
  assetDir,
  deleteMediaPaths,
  errorMessage,
  fileExists,
  moveFile,
  originalExtension,
  publishSafe,
  sha256File,
} from "./media-ingest-files.js";
import { buildProxy } from "./media-ingest-proxy.js";
import type { MediaJobLimits } from "./media-jobs.js";
import {
  assertDiskSpace,
  defaultMediaJobLimits,
  isDiskFull,
  MEDIA_MIN_FREE_BYTES,
  MediaDiskFullError,
  MediaJobLimiter,
} from "./media-jobs.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.
 *
 * Sync walks every page of a server-side source from the start: positional
 * cursors (folder paths) cannot express "changed since", and per-item
 * revisions make unchanged pages cheap. Unchanged items enqueue nothing.
 *
 * Proxy and analysis work is pumped, not fanned out: the queue holds at most a
 * small window of proxy and analyze jobs per space (the oldest assets waiting
 * in `indexed` or `proxied`), and every finished job refills the window. A
 * 50k-asset library therefore never floods the shared job queue that chat runs
 * also use. Job keys are per asset, so refills replace instead of duplicating.
 *
 * Idempotency: a proxy is rebuilt only when the original's content hash
 * differs from the one it was built from; analysis runs only the kinds that
 * have no result at the provider's current model version (explicit kinds
 * always re-run). A crashed claim (proxying/analyzing) is taken over after
 * MEDIA_STALE_CLAIM_MS.
 */

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;
/** A job that found no free permit runs again after this delay. */
export const MEDIA_BUSY_RETRY_MS = 15_000;
/** A job paused by a full disk or an unreachable sidecar retries after this delay. */
export const MEDIA_PAUSE_RETRY_MS = 5 * 60_000;
/** An in-flight claim older than this belongs to a crashed worker. */
export const MEDIA_STALE_CLAIM_MS = 30 * 60_000;

const DISK_PAUSED = "Media processing is paused because the disk is full";
const SERVER_SOURCES = ["folder", "emulator"] as const;
const REMOVE_BATCH = 500;

type ServerSourceKind = (typeof SERVER_SOURCES)[number];
type AssetRecord = Awaited<ReturnType<MediaServiceDeps["repos"]["getAssetRecord"]>>;

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

/**
 * Content changed under an asset: analyses, segments, and face tracks describe
 * the old bytes and are dropped so the next analysis starts clean.
 */
async function dropStaleAnalysis(
  deps: MediaServiceDeps,
  spaceId: string,
  assetId: string,
): Promise<void> {
  await deps.repos.replaceFaceTracks(spaceId, assetId, "stale", []);
  await deps.repos.replaceSegments(spaceId, assetId, 0, []);
  await deps.prisma.mediaAnalysis.deleteMany({ where: { spaceId, assetId } });
  await deleteMediaPaths(deps.mediaRoot, [`${assetDir(spaceId, assetId)}/faces`]);
}

/** Assets waiting in `waiting`, or stuck in `inFlight` by a worker that died. */
function waitingWhere(
  deps: MediaServiceDeps,
  spaceId: string,
  waiting: MediaIngestState,
  inFlight: MediaIngestState,
) {
  const staleBefore = new Date(deps.now().getTime() - MEDIA_STALE_CLAIM_MS);
  const storage = { in: ["referenced", "uploaded"] };
  return {
    spaceId,
    storage,
    OR: [{ ingestState: waiting }, { ingestState: inFlight, updatedAt: { lt: staleBefore } }],
  };
}

/** Enqueue proxy jobs for the oldest assets waiting for one. */
async function refillProxy(deps: MediaServiceDeps, spaceId: string, window: number) {
  const rows = await deps.prisma.mediaAsset.findMany({
    where: waitingWhere(deps, spaceId, "indexed", "proxying"),
    orderBy: [{ createdAt: "asc" }, { id: "asc" }],
    take: window,
    select: { id: true },
  });
  for (const row of rows) await deps.jobs.enqueue(mediaProxyJob(row.id));
}

/** Enqueue analyze jobs for the oldest assets waiting for one. */
async function refillAnalyze(deps: MediaServiceDeps, spaceId: string, window: number) {
  if (!deps.inference) return;
  const rows = await deps.prisma.mediaAsset.findMany({
    where: waitingWhere(deps, spaceId, "proxied", "analyzing"),
    orderBy: [{ updatedAt: "asc" }, { id: "asc" }],
    take: window,
    select: { id: true },
  });
  for (const row of rows) await deps.jobs.enqueue(mediaAnalyzeJob(row.id));
}

/**
 * Restart the proxy and analysis pumps of every space with waiting or stranded
 * work. The worker calls this on start (after a crash or a sidecar was added)
 * and periodically; it only enqueues jobs, so repeating it is harmless.
 */
export async function resumeMediaIngest(
  deps: MediaServiceDeps,
  limits: MediaJobLimits = defaultMediaJobLimits(),
): Promise<number> {
  const states: MediaIngestState[] = deps.inference
    ? ["indexed", "proxying", "proxied", "analyzing"]
    : ["indexed", "proxying"];
  const spaces = await deps.prisma.mediaAsset.findMany({
    where: { ingestState: { in: states }, storage: { in: ["referenced", "uploaded"] } },
    distinct: ["spaceId"],
    select: { spaceId: true },
  });
  for (const { spaceId } of spaces) {
    await refillProxy(deps, spaceId, limits.cpu * 2);
    await refillAnalyze(deps, spaceId, limits.gpu * 2);
  }
  return spaces.length;
}

/** Schedule clustering once the space has no proxy or analysis work left. */
async function clusterWhenDrained(deps: MediaServiceDeps, spaceId: string): Promise<void> {
  const pending: MediaIngestState[] = deps.inference
    ? ["proxying", "proxied", "analyzing"]
    : ["proxying"];
  const busy = await deps.prisma.mediaAsset.count({
    where: {
      spaceId,
      OR: [
        { ingestState: { in: pending } },
        { ingestState: "indexed", storage: { in: ["referenced", "uploaded"] } },
      ],
    },
  });
  if (busy === 0) await scheduleCluster(deps, spaceId);
}

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");
      }
      if (!(await lstat(temp)).isFile()) throw new InvalidMediaInputError("Upload is not a file");
      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]);
      }
      if (asset.contentHash && asset.contentHash !== actual) {
        await dropStaleAnalysis(deps, actor.spaceId, asset.id);
      }
      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;
  /** Per-process job limits; defaults to defaultMediaJobLimits(). */
  limits?: MediaJobLimits;
  /** Free bytes the media root keeps in reserve; jobs pause below it. 0 disables the check. */
  minFreeBytes?: number;
  /** Session rules (gap minutes and so on); the time zone defaults to the library owner's. */
  sessionOptions?: Partial<SessionOptions>;
}

async function syncSource(
  deps: MediaServiceDeps,
  source: MediaSourceRecord,
  provider: MediaSourceProvider,
  onChanged: () => Promise<void>,
  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);
    if (result.changedAssetIds.length > 0) {
      // A changed revision means new bytes: the asset goes back through the pipeline.
      await deps.prisma.mediaAsset.updateMany({
        where: {
          spaceId: source.spaceId,
          id: { in: result.changedAssetIds },
          storage: "referenced",
          ingestState: { not: "indexed" },
        },
        data: { ingestState: "indexed", ingestError: null },
      });
      changed += result.changedAssetIds.length;
      await onChanged();
    }
    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 limiter = new MediaJobLimiter(options.limits);
  const proxyWindow = limiter.limits.cpu * 2;
  const analyzeWindow = limiter.limits.gpu * 2;
  const minFreeBytes = options.minFreeBytes ?? MEDIA_MIN_FREE_BYTES;
  const tools: MediaToolPaths = {
    ...(options.ffmpegPath ? { ffmpegPath: options.ffmpegPath } : {}),
    ...(options.ffprobePath ? { ffprobePath: options.ffprobePath } : {}),
  };

  const later = async (job: BackgroundJob, delayMs: number) => {
    await deps.jobs.enqueue({ ...job, availableAt: new Date(deps.now().getTime() + delayMs) });
  };

  const checkDisk = async () => {
    await mkdir(deps.mediaRoot, { recursive: true });
    await assertDiskSpace(deps.mediaRoot, minFreeBytes);
  };

  /** Pause on a full disk: tell the source's owner why and try again later. */
  const pauseForDisk = async (spaceId: string, sourceId: string, job: BackgroundJob) => {
    getLogger().warn("media job paused: disk is full", { job: job.name });
    const source = await deps.repos
      .updateSourceSync(spaceId, sourceId, { error: DISK_PAUSED })
      .catch(() => null);
    if (source) {
      await publishSafe(deps, spaceId, {
        type: "media.source",
        sourceId,
        status: source.status,
        error: DISK_PAUSED,
      });
    }
    await later(job, MEDIA_PAUSE_RETRY_MS);
  };

  const resumeAfterDisk = async (spaceId: string, sourceId: string) => {
    const result = await deps.prisma.mediaSource.updateMany({
      where: { id: sourceId, spaceId, error: DISK_PAUSED },
      data: { error: null },
    });
    if (result.count === 0) return;
    const source = await deps.repos.getSource(spaceId, sourceId).catch(() => null);
    if (source) {
      await publishSafe(deps, spaceId, {
        type: "media.source",
        sourceId,
        status: source.status,
        error: null,
      });
    }
  };

  /**
   * Claim an asset for a step: from one of `from`, or from `inFlight` when that
   * claim is stale (its worker died). Only one worker wins.
   */
  const claim = async (
    spaceId: string,
    assetId: string,
    from: MediaIngestState[],
    inFlight: MediaIngestState,
  ): Promise<boolean> => {
    const staleBefore = new Date(deps.now().getTime() - MEDIA_STALE_CLAIM_MS);
    const result = await deps.prisma.mediaAsset.updateMany({
      where: {
        id: assetId,
        spaceId,
        OR: [
          { ingestState: { in: from } },
          { ingestState: inFlight, updatedAt: { lt: staleBefore } },
        ],
      },
      data: { ingestState: inFlight, ingestError: null },
    });
    if (result.count !== 1) return false;
    await publishSafe(deps, spaceId, { type: "media.asset", assetId, ingestState: inFlight });
    return true;
  };

  const setState = async (
    spaceId: string,
    assetId: string,
    from: MediaIngestState,
    to: MediaIngestState,
    error?: string,
  ) => {
    const applied = await deps.repos
      .transitionIngestState(spaceId, assetId, {
        from: [from],
        to,
        ...(error !== undefined ? { error } : {}),
      })
      .catch((cause) => {
        if (cause instanceof IsolationError) return false;
        throw cause;
      });
    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: AssetRecord): Promise<string | null> => {
    if (asset.storage === "uploaded" && asset.storagePath) {
      return resolveInside(deps.mediaRoot, asset.storagePath);
    }
    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 provider.resolve(source.config, asset.externalId);
  };

  /** Load the asset a job names; null when it (or its space) is gone. */
  const loadAsset = async (assetId: string) => {
    const spaceId = await deps.repos.mediaSpaceOf("asset", assetId);
    if (!spaceId) return null;
    try {
      return { spaceId, asset: await deps.repos.getAssetRecord(spaceId, assetId) };
    } catch (error) {
      if (error instanceof IsolationError) return null;
      throw error;
    }
  };

  const proxyAsset = async (spaceId: string, asset: AssetRecord): Promise<void> => {
    const original = await originalOf(spaceId, asset);
    if (!original) throw new Error("The original has not been uploaded");
    const contentHash = await sha256File(original, signal);
    const unchanged =
      contentHash === asset.contentHash &&
      asset.proxyPath !== null &&
      asset.thumbPath !== null &&
      (await fileExists(deps.mediaRoot, asset.proxyPath)) &&
      (await fileExists(deps.mediaRoot, asset.thumbPath));
    if (unchanged) {
      await deps.repos.recordProxy(spaceId, asset.id, {
        contentHash,
        proxyPath: asset.proxyPath,
        thumbPath: asset.thumbPath,
      });
      return;
    }
    if (asset.contentHash && asset.contentHash !== contentHash) {
      await dropStaleAnalysis(deps, spaceId, asset.id);
    }
    const outputs = await buildProxy({
      mediaRoot: deps.mediaRoot,
      spaceId,
      assetId: asset.id,
      original,
      still: asset.kind === "photo",
      tools,
      ...(signal ? { signal } : {}),
    });
    const still = outputs.spritePath === null;
    await deps.repos.recordProxy(spaceId, asset.id, {
      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,
    });
  };

  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",
        error: null,
      });
      try {
        const { changed, deleted } = await syncSource(
          deps,
          source,
          provider,
          () => refillProxy(deps, spaceId, proxyWindow),
          signal,
        );
        await deps.repos.updateSourceSync(spaceId, sourceId, {
          status: "idle",
          lastSyncedAt: deps.now(),
          error: null,
        });
        await publishSafe(deps, spaceId, {
          type: "media.source",
          sourceId,
          status: "idle",
          error: null,
        });
        if (changed + deleted > 0) await scheduleCluster(deps, spaceId);
      } catch (error) {
        if (signal?.aborted) {
          await deps.repos
            .updateSourceSync(spaceId, sourceId, { status: "idle" })
            .catch(() => undefined);
          throw error;
        }
        // The source was removed while it synced.
        if (error instanceof IsolationError) return;
        const message = errorMessage(error);
        getLogger().error("media sync failed", error, { sourceId });
        await deps.repos
          .updateSourceSync(spaceId, sourceId, { status: "error", error: message })
          .catch(() => undefined);
        await publishSafe(deps, spaceId, {
          type: "media.source",
          sourceId,
          status: "error",
          error: message,
        });
      }
    },

    async "media.proxy"({ assetId }) {
      const loaded = await loadAsset(assetId);
      if (!loaded || loaded.asset.storage === "remote") return;
      const { spaceId, asset } = loaded;
      const job = mediaProxyJob(assetId);
      const release = limiter.tryAcquire("cpu");
      if (!release) {
        await later(job, MEDIA_BUSY_RETRY_MS);
        return;
      }
      try {
        try {
          await checkDisk();
        } catch (error) {
          if (!(error instanceof MediaDiskFullError)) throw error;
          await pauseForDisk(spaceId, asset.sourceId, job);
          return;
        }
        if (!(await claim(spaceId, assetId, ["indexed", "failed"], "proxying"))) return;
        try {
          await proxyAsset(spaceId, asset);
          await publishSafe(deps, spaceId, {
            type: "media.asset",
            assetId,
            ingestState: "proxied",
          });
          await resumeAfterDisk(spaceId, asset.sourceId);
        } catch (error) {
          if (signal?.aborted) {
            await setState(spaceId, assetId, "proxying", "indexed");
            throw error;
          }
          if (error instanceof IsolationError) {
            // Deleted while it was processed: drop whatever was written for it.
            await deleteMediaPaths(deps.mediaRoot, [assetDir(spaceId, assetId)]);
            return;
          }
          if (isDiskFull(error)) {
            await setState(spaceId, assetId, "proxying", "indexed");
            await pauseForDisk(spaceId, asset.sourceId, job);
            return;
          }
          getLogger().error("media proxy failed", error, { assetId });
          await setState(spaceId, assetId, "proxying", "failed", errorMessage(error));
        }
      } finally {
        release();
      }
      await refillProxy(deps, spaceId, proxyWindow);
      await refillAnalyze(deps, spaceId, analyzeWindow);
      await clusterWhenDrained(deps, spaceId);
    },

    async "media.analyze"({ assetId, kinds }) {
      const inference = deps.inference;
      if (!inference) return;
      const loaded = await loadAsset(assetId);
      if (!loaded?.asset.proxyPath) return;
      const { spaceId, asset } = loaded;
      // A repeated default request for an asset already at the current pipeline version.
      if (!kinds && asset.ingestState === "analyzed") {
        if (asset.analysisVersion >= MEDIA_ANALYSIS_VERSION) return;
      }
      const job = mediaAnalyzeJob(assetId, kinds);
      const release = limiter.tryAcquire("gpu");
      if (!release) {
        await later(job, MEDIA_BUSY_RETRY_MS);
        return;
      }
      try {
        try {
          await checkDisk();
        } catch (error) {
          if (!(error instanceof MediaDiskFullError)) throw error;
          await pauseForDisk(spaceId, asset.sourceId, job);
          return;
        }
        if (!(await claim(spaceId, assetId, ["proxied", "analyzed", "failed"], "analyzing"))) {
          return;
        }
        try {
          const original = asset.storage === "referenced" ? await originalOf(spaceId, asset) : null;
          const inputPath = await analysisInputPath({
            mediaRoot: deps.mediaRoot,
            proxyPath: asset.proxyPath as string,
            referencedOriginal: original,
          });
          const outcome = await analyzeAsset({
            deps,
            inference,
            spaceId,
            asset,
            inputPath,
            ...(kinds ? { kinds } : {}),
            ...(signal ? { signal } : {}),
          });
          await deps.repos.recordAnalysisVersion(
            spaceId,
            assetId,
            MEDIA_ANALYSIS_VERSION,
            outcome.qualityScore,
          );
          await publishSafe(deps, spaceId, {
            type: "media.asset",
            assetId,
            ingestState: "analyzed",
          });
          await resumeAfterDisk(spaceId, asset.sourceId);
        } catch (error) {
          if (signal?.aborted) {
            await setState(spaceId, assetId, "analyzing", "proxied");
            throw error;
          }
          if (error instanceof IsolationError) return;
          if (isDiskFull(error)) {
            await setState(spaceId, assetId, "analyzing", "proxied");
            await pauseForDisk(spaceId, asset.sourceId, job);
            return;
          }
          if (isTransientInferenceError(error)) {
            // Sidecar down or saturated: not the asset's fault, try again later.
            getLogger().warn("media analysis deferred", { assetId, reason: errorMessage(error) });
            await setState(spaceId, assetId, "analyzing", "proxied");
            await later(job, MEDIA_PAUSE_RETRY_MS);
            return;
          }
          getLogger().error("media analysis failed", error, { assetId });
          await setState(spaceId, assetId, "analyzing", "failed", errorMessage(error));
        }
      } finally {
        release();
      }
      await refillAnalyze(deps, spaceId, analyzeWindow);
      await clusterWhenDrained(deps, spaceId);
    },

    async "media.cluster"({ spaceId }) {
      const release = limiter.tryAcquire("cpu");
      if (!release) {
        await later(mediaClusterJob(spaceId), MEDIA_BUSY_RETRY_MS);
        return;
      }
      try {
        await clusterSpace(deps, spaceId, {
          ...(signal ? { signal } : {}),
          ...(options.sessionOptions ? { sessionOptions: options.sessionOptions } : {}),
        });
      } finally {
        release();
      }
    },
  };
}