evOo-local/packages/adapters/src/media-library.ts

import type { Dirent } from "node:fs";
import { lstat, readdir, realpath, stat } from "node:fs/promises";
import path from "node:path";
import type { ArtifactStore, JobPublisher } from "@rakazo/adapter-kit";
import { mediaLibrarySyncJob } from "@rakazo/adapter-kit";
import type {
  MediaLibraryKind,
  MediaLibrarySearchInput,
  MediaLibrarySource,
  MediaLibrarySourceStatus,
} from "@rakazo/contracts";
import { attachmentKindOf } from "@rakazo/contracts";
import { inferAttachmentMimeType } from "@rakazo/core";
import type { PrismaClient } from "@rakazo/db";
import { getLogger } from "@rakazo/logging";
import { storeArtifactFromFile } from "./artifact-files.js";
import { detectFfmpeg, probeMedia } from "./media/ffmpeg.js";

/*
 * Media library folders: the deployment owner points evOo at folders on the server, a sync job
 * indexes the photos and videos below them, and the user's bots search the index and import a
 * file into the conversation when they need its bytes. Files are never moved or changed.
 */

/** A folder is re-indexed this long after its last sync, so new files show up on their own. */
export const MEDIA_LIBRARY_RESYNC_MS = 6 * 60 * 60 * 1000;
const DELETE_BATCH = 500;
const DEFAULT_SEARCH_LIMIT = 20;

export class MediaLibraryError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "MediaLibraryError";
  }
}

/** Photo or video for files a conversation can hold; null for everything else. */
export function mediaLibraryFileType(
  name: string,
): { kind: MediaLibraryKind; mimeType: string } | null {
  const mimeType = inferAttachmentMimeType(name);
  if (!mimeType) return null;
  const kind = attachmentKindOf(mimeType);
  if (kind === "image") return { kind: "photo", mimeType };
  if (kind === "video") return { kind: "video", mimeType };
  return null;
}

function visible(entry: Dirent): boolean {
  return !entry.name.startsWith(".") && !entry.isSymbolicLink();
}

/**
 * Relative POSIX paths of the media files below `root`. Hidden entries and symlinks are skipped
 * and never followed, so the walk cannot leave the folder; unreadable subfolders are skipped.
 */
export async function* walkMediaLibrary(
  root: string,
  signal?: AbortSignal,
  relativeDir = "",
): AsyncGenerator<string> {
  signal?.throwIfAborted();
  let entries: Dirent[];
  try {
    entries = await readdir(path.join(root, relativeDir), { withFileTypes: true });
  } catch (error) {
    const code = (error as NodeJS.ErrnoException).code;
    if (relativeDir && (code === "ENOENT" || code === "EACCES" || code === "ENOTDIR")) return;
    throw error;
  }
  for (const entry of entries.filter(visible).sort((a, b) => (a.name < b.name ? -1 : 1))) {
    const relative = relativeDir ? `${relativeDir}/${entry.name}` : entry.name;
    if (entry.isDirectory()) yield* walkMediaLibrary(root, signal, relative);
    else if (entry.isFile() && mediaLibraryFileType(entry.name)) yield relative;
  }
}

export function mediaLibrarySourceView(row: {
  id: string;
  path: string;
  status: string;
  error: string | null;
  assetCount: number;
  lastSyncedAt: Date | null;
}): MediaLibrarySource {
  return {
    id: row.id,
    path: row.path,
    status: row.status as MediaLibrarySourceStatus,
    error: row.error,
    assetCount: row.assetCount,
    lastSyncedAt: row.lastSyncedAt?.toISOString() ?? null,
  };
}

/** An absolute path to an existing folder, or a message saying what is wrong with it. */
export async function checkMediaLibraryFolder(folder: string): Promise<string | null> {
  if (!path.isAbsolute(folder)) return "Use an absolute folder path.";
  try {
    if (!(await stat(folder)).isDirectory()) return "This is not a folder.";
  } catch {
    return "Folder not found.";
  }
  return null;
}

type IndexedFile = {
  relativePath: string;
  name: string;
  kind: MediaLibraryKind;
  mimeType: string;
  size: bigint;
  modifiedAt: Date;
  takenAt: Date;
  width: number | null;
  height: number | null;
  durationMs: number | null;
};

async function indexFile(
  root: string,
  relativePath: string,
  canProbe: boolean,
  signal?: AbortSignal,
): Promise<IndexedFile | null> {
  const name = path.posix.basename(relativePath);
  const type = mediaLibraryFileType(name);
  if (!type) return null;
  const full = path.join(root, ...relativePath.split("/"));
  const info = await lstat(full).catch(() => null);
  if (!info?.isFile()) return null;
  const probe = canProbe
    ? await probeMedia(full, { signal, mimeType: type.mimeType }).catch(() => null)
    : null;
  return {
    relativePath,
    name,
    ...type,
    size: BigInt(info.size),
    modifiedAt: info.mtime,
    takenAt: probe?.createdAt ?? info.mtime,
    width: probe?.video?.width ?? null,
    height: probe?.video?.height ?? null,
    durationMs: type.kind === "video" && probe?.durationMs ? probe.durationMs : null,
  };
}

/**
 * Re-indexes one folder: new and changed files are probed and stored, files that are gone are
 * dropped, unchanged files cost one stat. A failure is recorded on the source instead of
 * retried; the next sync is scheduled either way.
 */
export async function syncMediaLibrarySource(
  deps: { prisma: PrismaClient; jobs: JobPublisher },
  sourceId: string,
  options: { signal?: AbortSignal; now?: () => Date } = {},
): Promise<void> {
  const now = options.now ?? (() => new Date());
  const source = await deps.prisma.mediaLibrarySource.findUnique({ where: { id: sourceId } });
  if (!source) return;
  await deps.prisma.mediaLibrarySource.update({
    where: { id: sourceId },
    data: { status: "syncing", error: null },
  });
  let result: { status: MediaLibrarySourceStatus; error: string | null; synced: boolean };
  try {
    const problem = await checkMediaLibraryFolder(source.path);
    if (problem) throw new MediaLibraryError(problem);
    await indexFolder(deps.prisma, source, options.signal);
    result = { status: "idle", error: null, synced: true };
  } catch (error) {
    const message =
      error instanceof MediaLibraryError ? error.message : "Could not read this folder.";
    if (!(error instanceof MediaLibraryError)) {
      getLogger().error(`media library sync failed for source ${sourceId}`, error);
    }
    result = { status: "failed", error: message, synced: false };
  }
  const assetCount = await deps.prisma.mediaLibraryAsset.count({ where: { sourceId } });
  const updated = await deps.prisma.mediaLibrarySource.updateMany({
    where: { id: sourceId },
    data: {
      status: result.status,
      error: result.error,
      assetCount,
      ...(result.synced ? { lastSyncedAt: now() } : {}),
    },
  });
  // Removed while syncing: nothing left to schedule.
  if (updated.count === 0) return;
  await deps.jobs.enqueue(
    mediaLibrarySyncJob(sourceId, new Date(now().getTime() + MEDIA_LIBRARY_RESYNC_MS)),
  );
}

async function indexFolder(
  prisma: PrismaClient,
  source: { id: string; userId: string; path: string },
  signal?: AbortSignal,
) {
  const canProbe = (await detectFfmpeg()) !== null;
  const known = new Map(
    (
      await prisma.mediaLibraryAsset.findMany({
        where: { sourceId: source.id },
        select: { id: true, relativePath: true, size: true, modifiedAt: true },
      })
    ).map((row) => [row.relativePath, row]),
  );
  const seen = new Set<string>();
  for await (const relativePath of walkMediaLibrary(source.path, signal)) {
    seen.add(relativePath);
    const row = known.get(relativePath);
    if (row) {
      const info = await lstat(path.join(source.path, ...relativePath.split("/"))).catch(
        () => null,
      );
      if (
        info &&
        BigInt(info.size) === row.size &&
        info.mtime.getTime() === row.modifiedAt.getTime()
      ) {
        continue;
      }
    }
    const file = await indexFile(source.path, relativePath, canProbe, signal);
    if (!file) continue;
    await prisma.mediaLibraryAsset.upsert({
      where: { sourceId_relativePath: { sourceId: source.id, relativePath } },
      create: { ...file, sourceId: source.id, userId: source.userId },
      update: file,
    });
  }
  const gone = [...known.values()].filter((row) => !seen.has(row.relativePath));
  for (let index = 0; index < gone.length; index += DELETE_BATCH) {
    await prisma.mediaLibraryAsset.deleteMany({
      where: { id: { in: gone.slice(index, index + DELETE_BATCH).map((row) => row.id) } },
    });
  }
}

export async function userHasMediaLibrary(prisma: PrismaClient, userId: string) {
  return (await prisma.mediaLibrarySource.count({ where: { userId } })) > 0;
}

/** The `search_media` tool: the user's library, newest capture first. */
export async function searchMediaLibrary(
  prisma: PrismaClient,
  userId: string,
  input: MediaLibrarySearchInput,
) {
  const words = input.query?.split(/\s+/).filter(Boolean) ?? [];
  const where = {
    userId,
    ...(input.kind ? { kind: input.kind } : {}),
    ...(input.taken_after || input.taken_before
      ? {
          takenAt: {
            ...(input.taken_after ? { gte: new Date(input.taken_after) } : {}),
            ...(input.taken_before ? { lt: new Date(input.taken_before) } : {}),
          },
        }
      : {}),
    // Folder and file names carry most of what people know about their media.
    AND: words.map((word) => ({
      relativePath: { contains: word, mode: "insensitive" as const },
    })),
  };
  const [total, rows] = await Promise.all([
    prisma.mediaLibraryAsset.count({ where }),
    prisma.mediaLibraryAsset.findMany({
      where,
      orderBy: [{ takenAt: "desc" }, { id: "asc" }],
      take: input.limit ?? DEFAULT_SEARCH_LIMIT,
    }),
  ]);
  return {
    total,
    items: rows.map((row) => ({
      asset_id: row.id,
      path: row.relativePath,
      kind: row.kind,
      taken_at: row.takenAt.toISOString(),
      size_mb: Math.round((Number(row.size) / (1024 * 1024)) * 10) / 10,
      ...(row.width && row.height ? { width: row.width, height: row.height } : {}),
      ...(row.durationMs ? { duration_s: Math.round(row.durationMs / 100) / 10 } : {}),
    })),
  };
}

/**
 * The asset's file, or a MediaLibraryError. The resolved path must still lie inside the folder:
 * a folder swapped for a symlink after indexing does not lead anywhere else.
 */
export async function resolveMediaLibraryFile(
  prisma: PrismaClient,
  userId: string,
  assetId: string,
) {
  const asset = await prisma.mediaLibraryAsset.findFirst({
    where: { id: assetId, userId },
    include: { source: { select: { path: true } } },
  });
  if (!asset) throw new MediaLibraryError("No media with this id in the library.");
  const root = await realpath(asset.source.path).catch(() => null);
  const file = await realpath(path.join(asset.source.path, ...asset.relativePath.split("/"))).catch(
    () => null,
  );
  if (!root || !file?.startsWith(`${root}${path.sep}`)) {
    throw new MediaLibraryError("This file is no longer in the library folder.");
  }
  return { asset, file };
}

/** The `import_media` tool: copies a library file into the conversation as an artifact. */
export async function importMediaLibraryAsset(
  deps: { prisma: PrismaClient; artifacts: ArtifactStore },
  scope: {
    spaceId: string;
    userId: string;
    botId: string | null;
    groupId: string | null;
    runId: string | null;
    signal?: AbortSignal;
  },
  assetId: string,
) {
  const { asset, file } = await resolveMediaLibraryFile(deps.prisma, scope.userId, assetId);
  return storeArtifactFromFile(deps, {
    spaceId: scope.spaceId,
    userId: scope.userId,
    botId: scope.botId,
    groupId: scope.groupId,
    runId: scope.runId,
    name: asset.name,
    mimeType: asset.mimeType,
    path: file,
    signal: scope.signal,
    ...(asset.width ? { width: asset.width } : {}),
    ...(asset.height ? { height: asset.height } : {}),
    ...(asset.durationMs ? { durationMs: asset.durationMs } : {}),
  });
}