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

import { copyFile, mkdtemp, open, rename, stat } from "node:fs/promises";
import { join } from "node:path";
import type {
  MediaJobContext,
  MediaJobInputFile,
  MediaJobOutputFile,
  MediaOperationResult,
} from "@rakazo/adapter-kit";
import type {
  EditVideoInput,
  ExportVideoInput,
  MediaFit,
  MediaOutput,
  MediaPlatformId,
  MediaSkippedTarget,
  PlatformProfile,
} from "@rakazo/contracts";
import {
  ARTIFACT_MAX_BYTES_BY_KIND,
  EditVideoInputSchema,
  ExportVideoInputSchema,
  MEDIA_PLATFORM_PROFILES,
  MEDIA_PRESETS,
} from "@rakazo/contracts";
import type { MediaOperationRegistration } from "../media-runtime.js";
import type { FfmpegInfo, MediaProbe } from "./ffmpeg.js";
import {
  detectFfmpeg,
  measureLoudness,
  outputTooLargeMessage,
  probeMedia,
  renderPoster,
  runFfmpeg,
} from "./ffmpeg.js";
import type { ExportMeasurement } from "./platform-export.js";
import {
  checkExport,
  describeIssue,
  describeSkipped,
  durationIssues,
  needsLoudnessFix,
  platformTarget,
  sizeFixKbps,
} from "./platform-export.js";
import type { PlanInput, VideoEditPlan } from "./video-edit-plan.js";
import { planLoudnessFix, 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.platforms ?? []).map((platform) => MEDIA_PLATFORM_PROFILES[platform].name),
        ...(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 plan = planVideoEdit({ source, steps: input.steps, clips, subtitleFiles, 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 platforms = [...new Set(input.platforms ?? [])];
  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");
  // A platform the video is too long or too short for is not rendered; the bot trims first.
  const skipped: MediaSkippedTarget[] = [];
  const rendered: MediaPlatformId[] = [];
  for (const platform of platforms) {
    const issues = durationIssues(MEDIA_PLATFORM_PROFILES[platform], source.probe.durationMs);
    if (issues.length) skipped.push({ platform, issues });
    else rendered.push(platform);
  }
  for (const target of skipped) await context.addSkipped(target);
  const notExported = skipped.map(({ platform, issues }) =>
    describeSkipped(MEDIA_PLATFORM_PROFILES[platform], issues),
  );
  const count = rendered.length + presets.length;
  if (count === 0) throw new Error(notExported.join(" "));
  const share = 1 / count;
  const report = (index: number) => progress.range(index * share, (index + 1) * share);
  const outputs: MediaOutput[] = [];
  for (const platform of rendered) {
    outputs.push(
      await exportForPlatform(context, source, MEDIA_PLATFORM_PROFILES[platform], {
        fit: input.fit,
        name: `${base}-${platform}.mp4`,
        workDir,
        report: report(outputs.length),
      }),
    );
  }
  for (const preset of presets) {
    const outputPath = join(workDir, `${preset}.mp4`);
    const plan = planVideoEdit({
      source,
      steps: [{ op: "reframe", preset, fit: input.fit }],
      outputPath,
    });
    outputs.push(
      await encode(context, plan, outputPath, workDir, report(outputs.length), {
        name: `${base}-${MEDIA_PRESETS[preset].aspect.replace(":", "x")}.mp4`,
        parentArtifactId: source.artifactId,
        preset,
      }),
    );
  }
  await progress.done();
  const checks = outputs.flatMap((output) =>
    output.platform && output.check && output.check.status !== "pass"
      ? [
          `${MEDIA_PLATFORM_PROFILES[output.platform].name} check ${output.check.status}: ${output.check.issues.map(describeIssue).join("; ")}.`,
        ]
      : [],
  );
  return {
    summary: [
      `Exported ${source.name} as ${outputs.map(describe).join(", ")}.`,
      ...checks,
      ...notExported,
    ].join(" "),
  };
}

/**
 * Renders a video to a platform's frame, frame rate, bitrate and loudness, measures the file
 * and stores it with the check. A file over the size limit is encoded once more at a lower
 * bitrate, and audio off the loudness target gets one more, linear loudness pass.
 */
async function exportForPlatform(
  context: MediaJobContext,
  source: PlanInput & MediaJobInputFile,
  profile: PlatformProfile,
  options: { fit: MediaFit; name: string; workDir: string; report: (fraction: number) => void },
): Promise<MediaOutput> {
  const { workDir, report } = options;
  const outputPath = join(workDir, `${profile.id}.mp4`);
  let target = platformTarget(profile, options.fit, source.probe.durationMs);
  const renderTarget = async () => {
    const plan = planVideoEdit({ source, steps: [], target, outputPath });
    await render(context, plan, outputPath, workDir, report);
    return plan;
  };
  let plan = await renderTarget();
  let measured = await measureExport(context, outputPath, profile);
  const lowerKbps = sizeFixKbps(profile, measured.result, target.maxBitrateKbps);
  if (lowerKbps !== null) {
    target = { ...target, maxBitrateKbps: lowerKbps };
    plan = await renderTarget();
    measured = await measureExport(context, outputPath, profile);
  }
  if (needsLoudnessFix(profile, measured.result) && measured.result.loudness) {
    const fixedPath = join(workDir, `${profile.id}-loudness.mp4`);
    await runFfmpeg(
      planLoudnessFix({
        inputPath: outputPath,
        outputPath: fixedPath,
        durationMs: measured.result.durationMs,
        target,
        measured: measured.result.loudness,
      }),
      {
        signal: context.signal,
        cwd: workDir,
        maxOutputBytes: ARTIFACT_MAX_BYTES_BY_KIND.video,
        outputs: [fixedPath],
      },
    );
    await rename(fixedPath, outputPath);
    measured = await measureExport(context, outputPath, profile);
  }
  return store(context, plan, outputPath, measured.probe, report, {
    name: options.name,
    parentArtifactId: source.artifactId,
    platform: profile.id,
    check: checkExport(profile, measured.result),
  });
}

async function measureExport(
  context: MediaJobContext,
  path: string,
  profile: PlatformProfile,
): Promise<{ probe: MediaProbe; result: ExportMeasurement }> {
  const options = { signal: context.signal, mimeType: "video/mp4" };
  const probe = await probeMedia(path, options);
  const loudness = probe.hasAudio
    ? await measureLoudness(
        path,
        { integratedLufs: profile.audio.targetLufs, truePeakDb: profile.audio.truePeakDb },
        options,
      )
    : null;
  return {
    probe,
    result: {
      width: probe.video?.width ?? 0,
      height: probe.video?.height ?? 0,
      fps: probe.video?.fps ?? 0,
      codec: probe.video?.codec ?? "",
      durationMs: probe.durationMs,
      sizeBytes: (await stat(path)).size,
      ...(probe.video?.bitRateKbps !== undefined ? { videoKbps: probe.video.bitRateKbps } : {}),
      loudness,
    },
  };
}

type OutputDetails = Pick<
  MediaJobOutputFile,
  "name" | "parentArtifactId" | "preset" | "platform" | "check"
>;

/** 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: OutputDetails,
): Promise<MediaOutput> {
  await render(context, plan, outputPath, workDir, report);
  const probe = await probeMedia(outputPath, {
    signal: context.signal,
    mimeType: plan.kind === "audio" ? "audio/mp4" : "video/mp4",
  });
  return store(context, plan, outputPath, probe, report, file);
}

/** Runs a plan's ffmpeg, refusing a result larger than an artifact may be. */
async function render(
  context: MediaJobContext,
  plan: VideoEditPlan,
  outputPath: string,
  workDir: string,
  report: (fraction: number) => void,
): Promise<void> {
  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));
}

/** Renders a video's poster and stores the file as an output. */
async function store(
  context: MediaJobContext,
  plan: VideoEditPlan,
  outputPath: string,
  probe: MediaProbe,
  report: (fraction: number) => void,
  file: OutputDetails,
): Promise<MediaOutput> {
  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 for TikTok (1080×1920, 12.5 s)" for the bot's summary; the runtime adds the ids. */
function describe(output: MediaOutput): string {
  const name = output.platform
    ? `${output.name} for ${MEDIA_PLATFORM_PROFILES[output.platform].name}`
    : output.name;
  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 ? `${name} (${details.join(", ")})` : name;
}

/** Maps ffmpeg progress onto the card: monotonic, written in order, never failing the job. */
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;
    },
  };
}