evOo-local/packages/adapters/src/media/video-operations.ts

import { copyFile, mkdtemp, open, stat } from "node:fs/promises";
import { join } from "node:path";
import type { MediaJobContext, MediaJobInputFile, MediaOperationResult } from "@rakazo/adapter-kit";
import type { EditVideoInput, ExportVideoInput, MediaOutput } from "@rakazo/contracts";
import {
  ARTIFACT_MAX_BYTES_BY_KIND,
  EditVideoInputSchema,
  ExportVideoInputSchema,
  MEDIA_PRESETS,
} from "@rakazo/contracts";
import type { MediaOperationRegistration } from "../media-runtime.js";
import type { FfmpegInfo } from "./ffmpeg.js";
import {
  detectFfmpeg,
  outputTooLargeMessage,
  probeMedia,
  renderPoster,
  runFfmpeg,
} from "./ffmpeg.js";
import { detectSpeech } from "./silence.js";
import type { PlanInput, VideoEditPlan } from "./video-edit-plan.js";
import { planVideoEdit } from "./video-edit-plan.js";

export interface VideoOperationsDeps {
  /** Finds ffmpeg; defaults to PATH lookup. */
  detect?: () => Promise<FfmpegInfo | null>;
}

/** Share of each output's progress spent encoding; the rest is probing and the poster. */
const ENCODE_SHARE = 0.95;
const SUBTITLE_MAX_BYTES = 5 * 1024 * 1024;

/** `edit_video` and `export_video`, available wherever ffmpeg and ffprobe are installed. */
export function createVideoOperations(
  deps: VideoOperationsDeps = {},
): MediaOperationRegistration[] {
  const detect = deps.detect ?? detectFfmpeg;
  const available = async () => (await detect()) !== null;
  const requireTools = async () => {
    if (!(await available()))
      throw new Error("ffmpeg and ffprobe are not installed on this server.");
  };
  const editVideo: MediaOperationRegistration<EditVideoInput> = {
    operation: "edit_video",
    inputSchema: EditVideoInputSchema,
    title: (input) => input.title,
    available,
    artifactIds: (input) => [
      input.artifact_id,
      ...input.steps.flatMap((step) =>
        step.op === "concat"
          ? step.artifact_ids
          : step.op === "subtitles"
            ? [step.artifact_id]
            : [],
      ),
    ],
    handler: {
      operation: "edit_video",
      run: async (context) => {
        await requireTools();
        return runEditVideo(context);
      },
    },
  };
  const exportVideo: MediaOperationRegistration<ExportVideoInput> = {
    operation: "export_video",
    inputSchema: ExportVideoInputSchema,
    title: (input) =>
      input.title ?? input.presets.map((preset) => MEDIA_PRESETS[preset].aspect).join(" · "),
    available,
    artifactIds: (input) => [input.artifact_id],
    handler: {
      operation: "export_video",
      run: async (context) => {
        await requireTools();
        return runExportVideo(context);
      },
    },
  };
  return [editVideo, exportVideo];
}

async function runEditVideo(context: MediaJobContext): Promise<MediaOperationResult> {
  const input = EditVideoInputSchema.parse(context.input);
  const workDir = await mkdtemp(join(context.tempDir, "edit-"));
  const progress = createProgressReporter(context);
  const source = await loadVideo(context, input.artifact_id);
  const name = cleanFileName(input.title, cleanFileName(withoutExtension(source.name), "video"));

  const [first] = input.steps;
  if (input.steps.length === 1 && first?.op === "poster") {
    const at = first.at;
    if (at * 1000 >= source.probe.durationMs) {
      throw new Error(
        `poster at ${at} s is past the end of the ${source.probe.durationMs / 1000} s video.`,
      );
    }
    const posterPath = join(workDir, "poster.jpg");
    await renderPoster(source.path, posterPath, {
      signal: context.signal,
      mimeType: source.mimeType,
      atMs: at * 1000,
    });
    const output = await context.addOutput({
      path: posterPath,
      name: `${name}.jpg`,
      mimeType: "image/jpeg",
      width: source.probe.video?.width,
      height: source.probe.video?.height,
      parentArtifactId: source.artifactId,
    });
    await progress.done();
    return { summary: `Saved the frame at ${at} s of ${source.name} as ${describe(output)}.` };
  }

  const clips = new Map<string, PlanInput>();
  const subtitleFiles = new Map<string, string>();
  for (const step of input.steps) {
    if (step.op === "concat") {
      for (const id of step.artifact_ids) {
        if (!clips.has(id)) clips.set(id, await loadVideo(context, id));
      }
    } else if (step.op === "subtitles" && !subtitleFiles.has(step.artifact_id)) {
      subtitleFiles.set(
        step.artifact_id,
        await prepareSubtitles(context, step.artifact_id, workDir, subtitleFiles.size),
      );
    }
  }
  const audioOnly = input.steps.at(-1)?.op === "extract_audio";
  const outputPath = join(workDir, audioOnly ? "output.m4a" : "output.mp4");
  const speech = await detectSpeech(
    { source, steps: input.steps, clips, subtitleFiles },
    { workDir, signal: context.signal },
  );
  const plan = planVideoEdit({
    source,
    steps: input.steps,
    clips,
    subtitleFiles,
    speech,
    outputPath,
  });
  const output = await encode(context, plan, outputPath, workDir, progress.range(0, 1), {
    name: `${name}${audioOnly ? ".m4a" : ".mp4"}`,
    parentArtifactId: source.artifactId,
  });
  await progress.done();
  return {
    summary: `Edited ${source.name} into ${describe(output)}.`,
  };
}

async function runExportVideo(context: MediaJobContext): Promise<MediaOperationResult> {
  const input = ExportVideoInputSchema.parse(context.input);
  const presets = [...new Set(input.presets)];
  const workDir = await mkdtemp(join(context.tempDir, "export-"));
  const progress = createProgressReporter(context);
  const source = await loadVideo(context, input.artifact_id);
  const base = cleanFileName(withoutExtension(source.name), "video");
  const outputs: MediaOutput[] = [];
  for (const [index, preset] of presets.entries()) {
    const outputPath = join(workDir, `${preset}.mp4`);
    const plan = planVideoEdit({
      source,
      steps: [{ op: "reframe", preset, fit: input.fit }],
      outputPath,
    });
    const share = 1 / presets.length;
    outputs.push(
      await encode(
        context,
        plan,
        outputPath,
        workDir,
        progress.range(index * share, (index + 1) * share),
        {
          name: `${base}-${MEDIA_PRESETS[preset].aspect.replace(":", "x")}.mp4`,
          parentArtifactId: source.artifactId,
          preset,
        },
      ),
    );
  }
  await progress.done();
  return { summary: `Exported ${source.name} as ${outputs.map(describe).join(", ")}.` };
}

/** Runs a plan, checks the result, renders its poster and stores it as an output. */
async function encode(
  context: MediaJobContext,
  plan: VideoEditPlan,
  outputPath: string,
  workDir: string,
  report: (fraction: number) => void,
  file: Pick<Parameters<MediaJobContext["addOutput"]>[0], "name" | "parentArtifactId" | "preset">,
): Promise<MediaOutput> {
  const maxBytes = ARTIFACT_MAX_BYTES_BY_KIND[plan.kind];
  await runFfmpeg(plan.args, {
    signal: context.signal,
    cwd: workDir,
    durationMs: plan.durationMs,
    onProgress: (fraction) => report(fraction * ENCODE_SHARE),
    maxOutputBytes: maxBytes,
    outputs: [outputPath],
  });
  if ((await stat(outputPath)).size > maxBytes) throw new Error(outputTooLargeMessage(maxBytes));
  const probe = await probeMedia(outputPath, {
    signal: context.signal,
    mimeType: plan.kind === "audio" ? "audio/mp4" : "video/mp4",
  });
  if (plan.kind === "audio") {
    const output = await context.addOutput({
      ...file,
      path: outputPath,
      mimeType: "audio/mp4",
      durationMs: probe.durationMs,
    });
    report(1);
    return output;
  }
  const posterPath = outputPath.replace(/\.mp4$/, ".jpg");
  await renderPoster(outputPath, posterPath, {
    signal: context.signal,
    mimeType: "video/mp4",
    atMs: plan.posterAtMs ?? Math.min(1_000, probe.durationMs / 4),
  });
  const output = await context.addOutput({
    ...file,
    path: outputPath,
    mimeType: "video/mp4",
    width: probe.video?.width ?? plan.width,
    height: probe.video?.height ?? plan.height,
    durationMs: probe.durationMs,
    posterPath,
  });
  report(1);
  return output;
}

async function loadVideo(
  context: MediaJobContext,
  artifactId: string,
): Promise<PlanInput & MediaJobInputFile> {
  const file = await context.fetchArtifact(artifactId);
  if (file.kind !== "video") throw new Error(`${file.name} is not a video.`);
  const probe = await probeMedia(file.path, {
    signal: context.signal,
    mimeType: file.mimeType,
  });
  if (!probe.video) throw new Error(`${file.name} has no video.`);
  if (probe.durationMs <= 0) throw new Error(`Could not read the length of ${file.name}.`);
  return { ...file, probe };
}

/** Copies an SRT/VTT artifact to a generated name, so no user text reaches the filter graph. */
async function prepareSubtitles(
  context: MediaJobContext,
  artifactId: string,
  workDir: string,
  index: number,
): Promise<string> {
  const file = await context.fetchArtifact(artifactId);
  const format = file.size <= SUBTITLE_MAX_BYTES ? subtitleFormat(await readHead(file.path)) : null;
  if (!format) throw new Error(`${file.name} is not an SRT or VTT file.`);
  const name = `subtitles-${index}.${format}`;
  await copyFile(file.path, join(workDir, name));
  return name;
}

async function readHead(path: string): Promise<string> {
  const handle = await open(path, "r");
  try {
    const buffer = Buffer.alloc(1024);
    const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0);
    return buffer.subarray(0, bytesRead).toString("utf8");
  } finally {
    await handle.close();
  }
}

/**
 * Recognizes WebVTT and SubRip by content. Anything else is refused, so ffmpeg never probes
 * an arbitrary file (a playlist, say) handed in as subtitles.
 */
export function subtitleFormat(head: string): "srt" | "vtt" | null {
  const text = head.replace(/^/, "").trimStart();
  if (/^WEBVTT(?:[ \t\r\n]|$)/.test(text)) return "vtt";
  if (/^(?:\d+[ \t]*\r?\n)?\d{1,2}:\d{2}:\d{2}[,.]\d{1,3}[ \t]*-->/.test(text)) return "srt";
  return null;
}

/** A name for produced files: no path separators or control characters, at most 80 chars. */
export function cleanFileName(text: string, fallback: string): string {
  const name = text
    .replaceAll(/[\p{Cc}\\/:*?"<>|]+/gu, " ")
    .replaceAll(/\s+/g, " ")
    .replace(/^[\s.]+/, "")
    .slice(0, 80)
    .trim();
  return name || fallback;
}

function withoutExtension(name: string): string {
  return name.replace(/\.[A-Za-z0-9]{1,5}$/, "");
}

/** "name (1080×1920, 12.5 s)" for the bot's summary; the runtime adds the artifact ids. */
function describe(output: MediaOutput): string {
  const details = [
    output.width && output.height ? `${output.width}×${output.height}` : "",
    output.durationMs !== undefined ? `${Number((output.durationMs / 1000).toFixed(1))} s` : "",
  ].filter(Boolean);
  return details.length ? `${output.name} (${details.join(", ")})` : output.name;
}

/** Maps ffmpeg progress onto the card: monotonic, written in order, never failing the job. */
export function createProgressReporter(context: MediaJobContext) {
  let reported = 0;
  let writes = Promise.resolve();
  const write = (fraction: number) => {
    reported = fraction;
    writes = writes.then(() => context.progress(fraction)).catch(() => undefined);
  };
  return {
    range: (from: number, to: number) => (fraction: number) => {
      const value = Math.min(1, from + (to - from) * Math.min(1, Math.max(0, fraction)));
      if (value > reported) write(value);
    },
    done: async () => {
      if (reported < 1) write(1);
      await writes;
    },
  };
}