Merge pull request #772 from heygen-com/feat/producer-audio-pad-trim

feat(producer): audio post-pad/trim helper for assemble

packages/engine/src/index.ts

@@ -175,7 +175,12 @@ export {
 } from "./utils/ffprobe.js";
 
 export { downloadToTemp, isHttpUrl } from "./utils/urlDownloader.js";
-export { runFfmpeg, type RunFfmpegOptions, type RunFfmpegResult } from "./utils/runFfmpeg.js";
+export {
+  runFfmpeg,
+  formatFfmpegError,
+  type RunFfmpegOptions,
+  type RunFfmpegResult,
+} from "./utils/runFfmpeg.js";
 
 export {
   decodePng,

packages/producer/src/services/render/audioPadTrim.test.ts

@@ -0,0 +1,223 @@
+/**
+ * Tests for the `audioPadTrim` helper that backs the distributed
+ * `assemble()` step's audio duration correction.
+ *
+ * The helper is split into:
+ *   - `buildPadTrimAudioArgs(...)` — pure function, builds the ffmpeg argv
+ *     and labels the operation. Fully unit-testable.
+ *   - `padOrTrimAudioToVideoFrameCount(...)` — wrapper that probes the
+ *     video for frame count + fps, probes the audio for current duration,
+ *     and runs ffmpeg with the args from the pure helper. Tests inject
+ *     stubs for the probes and the ffmpeg runner.
+ *
+ * No real ffmpeg/ffprobe runs in these tests.
+ */
+
+import { describe, expect, it } from "bun:test";
+import {
+  buildPadTrimAudioArgs,
+  padOrTrimAudioToVideoFrameCount,
+  type AudioProbeInfo,
+  type PadTrimAudioInput,
+  type ProbeVideoFrameInfo,
+} from "./audioPadTrim.js";
+
+describe("buildPadTrimAudioArgs", () => {
+  it("emits an apad filter when audio is shorter than target", () => {
+    const { args, operation } = buildPadTrimAudioArgs("/tmp/in.aac", "/tmp/out.aac", 4.0, 5.0);
+    expect(operation).toBe("pad");
+    const afIdx = args.indexOf("-af");
+    expect(afIdx).toBeGreaterThan(-1);
+    expect(args[afIdx + 1]).toContain("apad=pad_dur=");
+    expect(args[afIdx + 1]).toMatch(/pad_dur=1\.0+/);
+    // Pad must re-encode — apad is a filter and filters can't combine with copy.
+    const codecIdx = args.indexOf("-c:a");
+    expect(args[codecIdx + 1]).toBe("aac");
+    expect(args[args.length - 1]).toBe("/tmp/out.aac");
+    expect(args.includes("-y")).toBe(true);
+  });
+
+  it("emits -t when audio is longer than target", () => {
+    const { args, operation } = buildPadTrimAudioArgs("/tmp/in.aac", "/tmp/out.aac", 6.123, 5.0);
+    expect(operation).toBe("trim");
+    const tIdx = args.indexOf("-t");
+    expect(tIdx).toBeGreaterThan(-1);
+    expect(args[tIdx + 1]).toBe("5.000000");
+    // Trim preserves AAC stream copy.
+    const codecIdx = args.indexOf("-c:a");
+    expect(args[codecIdx + 1]).toBe("copy");
+  });
+
+  it("emits a plain copy when source duration matches target within ~1ms", () => {
+    const { args, operation } = buildPadTrimAudioArgs("/tmp/in.aac", "/tmp/out.aac", 5.0, 5.0);
+    expect(operation).toBe("copy");
+    expect(args.indexOf("-af")).toBe(-1);
+    expect(args.indexOf("-t")).toBe(-1);
+    const codecIdx = args.indexOf("-c:a");
+    expect(args[codecIdx + 1]).toBe("copy");
+  });
+
+  it("emits 6-decimal-place pad_dur (no scientific notation)", () => {
+    // 1.23ms — just over the AUDIO_DURATION_TOLERANCE_SECONDS=1ms threshold,
+    // so we exercise the pad path with a tiny duration that would round to
+    // exponent notation if we used `toString()` instead of `toFixed(6)`.
+    const { args, operation } = buildPadTrimAudioArgs("/tmp/in.aac", "/tmp/out.aac", 0.0, 0.00123);
+    expect(operation).toBe("pad");
+    const afIdx = args.indexOf("-af");
+    expect(args[afIdx + 1]).toBe("apad=pad_dur=0.001230");
+  });
+
+  it("flags ~1ms drift as a copy (below the tolerance threshold)", () => {
+    const close = buildPadTrimAudioArgs("/tmp/a.aac", "/tmp/o.aac", 5.0, 5.0005);
+    expect(close.operation).toBe("copy");
+  });
+
+  it("flags >1ms drift in either direction as pad/trim", () => {
+    const padNeeded = buildPadTrimAudioArgs("/tmp/a.aac", "/tmp/o.aac", 5.0, 5.002);
+    expect(padNeeded.operation).toBe("pad");
+    const trimNeeded = buildPadTrimAudioArgs("/tmp/a.aac", "/tmp/o.aac", 5.002, 5.0);
+    expect(trimNeeded.operation).toBe("trim");
+  });
+});
+
+describe("padOrTrimAudioToVideoFrameCount", () => {
+  // Build a minimal harness that stubs out the three injectables.
+  function harness(opts: {
+    video: ProbeVideoFrameInfo | "throw";
+    audio: AudioProbeInfo | "throw";
+    ffmpeg?: (args: string[]) => Promise<{ success: boolean; error?: string }>;
+  }): { input: PadTrimAudioInput; captured: { args: string[][] } } {
+    const captured = { args: [] as string[][] };
+    const input: PadTrimAudioInput = {
+      videoPath: "/tmp/v.mp4",
+      audioPath: "/tmp/a.aac",
+      outputPath: "/tmp/o.aac",
+      probeVideoFrameInfo: async () => {
+        if (opts.video === "throw") throw new Error("video probe boom");
+        return opts.video;
+      },
+      probeAudioInfo: async () => {
+        if (opts.audio === "throw") throw new Error("audio probe boom");
+        return opts.audio;
+      },
+      runFfmpeg:
+        opts.ffmpeg ??
+        (async (args) => {
+          captured.args.push(args);
+          return { success: true };
+        }),
+    };
+    return { input, captured };
+  }
+
+  it("pads a video of N=180 frames at 30/1 fps with shorter audio", async () => {
+    const { input, captured } = harness({
+      video: { frameCount: 180, fpsNum: 30, fpsDen: 1 },
+      audio: { durationSeconds: 5.5 },
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(true);
+    expect(result.operation).toBe("pad");
+    expect(result.targetDurationSeconds).toBe(6);
+    expect(result.sourceDurationSeconds).toBe(5.5);
+    expect(captured.args).toHaveLength(1);
+    const afIdx = captured.args[0]!.indexOf("-af");
+    expect(captured.args[0]![afIdx + 1]).toBe("apad=pad_dur=0.500000");
+  });
+
+  it("trims a video of N=120 frames at 30/1 fps with longer audio", async () => {
+    const { input, captured } = harness({
+      video: { frameCount: 120, fpsNum: 30, fpsDen: 1 },
+      audio: { durationSeconds: 4.5 },
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(true);
+    expect(result.operation).toBe("trim");
+    expect(result.targetDurationSeconds).toBe(4);
+    expect(captured.args).toHaveLength(1);
+    const tIdx = captured.args[0]!.indexOf("-t");
+    expect(captured.args[0]![tIdx + 1]).toBe("4.000000");
+  });
+
+  it("emits a copy when audio duration already equals frameCount/fps", async () => {
+    const { input, captured } = harness({
+      video: { frameCount: 90, fpsNum: 30, fpsDen: 1 },
+      audio: { durationSeconds: 3.0 },
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(true);
+    expect(result.operation).toBe("copy");
+    expect(result.targetDurationSeconds).toBe(3);
+    expect(captured.args[0]!.includes("-af")).toBe(false);
+    expect(captured.args[0]!.includes("-t")).toBe(false);
+  });
+
+  it("handles NTSC fps (30000/1001) exactly", async () => {
+    // 120 frames at 30000/1001 ≈ 4.004s. Audio 4.0s → pad 0.004s.
+    const { input, captured } = harness({
+      video: { frameCount: 120, fpsNum: 30000, fpsDen: 1001 },
+      audio: { durationSeconds: 4.0 },
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(true);
+    expect(result.operation).toBe("pad");
+    expect(result.targetDurationSeconds).toBeCloseTo((120 * 1001) / 30000, 9);
+    const afIdx = captured.args[0]!.indexOf("-af");
+    expect(captured.args[0]![afIdx + 1]).toMatch(/^apad=pad_dur=0\.004\d+$/);
+  });
+
+  it("propagates video probe failure as success=false", async () => {
+    const { input } = harness({
+      video: "throw",
+      audio: { durationSeconds: 4.0 },
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(false);
+    expect(result.error).toContain("failed to probe video");
+    expect(result.error).toContain("video probe boom");
+  });
+
+  it("propagates audio probe failure as success=false", async () => {
+    const { input } = harness({
+      video: { frameCount: 90, fpsNum: 30, fpsDen: 1 },
+      audio: "throw",
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(false);
+    expect(result.error).toContain("failed to probe audio");
+    expect(result.error).toContain("audio probe boom");
+  });
+
+  it("propagates invalid video info (frameCount=0) as success=false", async () => {
+    const { input } = harness({
+      video: { frameCount: 0, fpsNum: 30, fpsDen: 1 },
+      audio: { durationSeconds: 4.0 },
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(false);
+    expect(result.error).toContain("invalid video frame info");
+  });
+
+  it("propagates invalid video info (fpsDen=0)", async () => {
+    const { input } = harness({
+      video: { frameCount: 100, fpsNum: 30, fpsDen: 0 },
+      audio: { durationSeconds: 4.0 },
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(false);
+    expect(result.error).toContain("invalid video frame info");
+  });
+
+  it("propagates ffmpeg failure as success=false with the ffmpeg error preserved", async () => {
+    const { input } = harness({
+      video: { frameCount: 180, fpsNum: 30, fpsDen: 1 },
+      audio: { durationSeconds: 5.0 },
+      ffmpeg: async () => ({ success: false, error: "synthetic ffmpeg failure" }),
+    });
+    const result = await padOrTrimAudioToVideoFrameCount(input);
+    expect(result.success).toBe(false);
+    expect(result.error).toBe("synthetic ffmpeg failure");
+    expect(result.operation).toBe("pad");
+    expect(result.targetDurationSeconds).toBe(6);
+  });
+});

packages/producer/src/services/render/audioPadTrim.ts

@@ -0,0 +1,352 @@
+/**
+ * audioPadTrim — pad-or-trim an `audio.aac` file so its exact duration
+ * matches the assembled video's frame count divided by fps.
+ *
+ * Distributed render assemble step needs this because:
+ *   - The plan's audio is mixed once against the composition's *declared*
+ *     duration.
+ *   - The actual video produced by concatenating per-chunk encodes is the
+ *     sum of per-chunk frame counts. With closed-GOP concat-copy this is
+ *     deterministic and equals the planned frame count, BUT downstream
+ *     muxers (ffmpeg `-shortest` plus Apple's mov demuxer in particular)
+ *     are sensitive to ±1ms audio/video duration drift and produce silent
+ *     "audio cuts off early" or "video shows a frozen final frame" bugs.
+ *
+ * The fix: post-pad/trim audio to *exactly* `frameCount / fps` seconds at
+ * assemble time. Pad with `apad=pad_dur=…` (silence fill), trim with `-t`.
+ */
+
+import { spawn } from "node:child_process";
+import {
+  extractAudioMetadata,
+  formatFfmpegError,
+  runFfmpeg,
+  type AudioMetadata,
+} from "@hyperframes/engine";
+
+/**
+ * Tolerance used to decide whether an audio file is already short enough to
+ * skip the pad/trim operation entirely. ~1ms is well below the perceptual
+ * threshold and well below any frame interval at 24/30/60fps.
+ */
+const AUDIO_DURATION_TOLERANCE_SECONDS = 0.001;
+
+export interface ProbeVideoFrameInfo {
+  /** Number of video frames in the stream. */
+  frameCount: number;
+  /** Numerator of the frame rate fraction (e.g. 30 or 30000). */
+  fpsNum: number;
+  /** Denominator of the frame rate fraction (e.g. 1 or 1001). */
+  fpsDen: number;
+}
+
+export interface AudioProbeInfo {
+  /** Decoded duration in seconds. */
+  durationSeconds: number;
+}
+
+export interface PadTrimAudioInput {
+  /** Path to the assembled video. Used to derive `frameCount / fps`. */
+  videoPath: string;
+  /** Path to the pre-mixed audio (typically `<planDir>/audio.aac`). */
+  audioPath: string;
+  /** Path the helper writes the duration-corrected audio to. */
+  outputPath: string;
+  /**
+   * Optional injectables for unit tests. Production callers omit them and
+   * get the real `ffprobe`/`ffmpeg`-backed implementations.
+   */
+  probeVideoFrameInfo?: (videoPath: string) => Promise<ProbeVideoFrameInfo>;
+  probeAudioInfo?: (audioPath: string) => Promise<AudioProbeInfo>;
+  runFfmpeg?: (args: string[]) => Promise<{ success: boolean; error?: string }>;
+}
+
+export type PadTrimOperation = "pad" | "trim" | "copy";
+
+export interface PadTrimAudioResult {
+  success: boolean;
+  outputPath: string;
+  /** `frameCount / fps` to ~nanosecond precision. */
+  targetDurationSeconds: number;
+  /** Probed duration of the input audio. */
+  sourceDurationSeconds: number;
+  /** How the duration was corrected. */
+  operation: PadTrimOperation;
+  /** Populated only when `success === false`. */
+  error?: string;
+}
+
+/**
+ * Pure helper: decide the pad/trim operation and build the ffmpeg argv list
+ * that materializes it. Exported separately so unit tests can pin both
+ * branches without spawning ffmpeg.
+ *
+ *   - `sourceDuration < targetDuration` → pad with `apad=pad_dur=Δ`.
+ *     Re-encode is required: `apad` is a filter and filters can't combine
+ *     with `-c:a copy`.
+ *   - `sourceDuration > targetDuration` → trim with `-t target`. `-c:a copy`
+ *     is preserved when the input is already AAC.
+ *   - `|Δ| < AUDIO_DURATION_TOLERANCE_SECONDS` → no-op `copy`, but we still
+ *     run ffmpeg with `-c:a copy` to materialize the output path.
+ */
+export function buildPadTrimAudioArgs(
+  audioPath: string,
+  outputPath: string,
+  sourceDurationSeconds: number,
+  targetDurationSeconds: number,
+): { args: string[]; operation: PadTrimOperation } {
+  const delta = targetDurationSeconds - sourceDurationSeconds;
+  const targetSec = formatSeconds(targetDurationSeconds);
+  if (Math.abs(delta) < AUDIO_DURATION_TOLERANCE_SECONDS) {
+    return {
+      operation: "copy",
+      args: ["-i", audioPath, "-c:a", "copy", "-y", outputPath],
+    };
+  }
+  if (delta > 0) {
+    const padDur = formatSeconds(delta);
+    return {
+      operation: "pad",
+      args: [
+        "-i",
+        audioPath,
+        "-af",
+        `apad=pad_dur=${padDur}`,
+        "-c:a",
+        "aac",
+        "-b:a",
+        "192k",
+        "-y",
+        outputPath,
+      ],
+    };
+  }
+  // Trim. `-t` truncates AAC without re-encoding because AAC frames are
+  // independently decodable; ffmpeg snaps the cut point to the nearest
+  // packet boundary, fine for the ±1ms tolerance we care about here.
+  return {
+    operation: "trim",
+    args: ["-i", audioPath, "-t", targetSec, "-c:a", "copy", "-y", outputPath],
+  };
+}
+
+/**
+ * Format a duration as a fixed-precision decimal string. ffmpeg parses
+ * scientific notation inconsistently across versions (some treat `1e-3` as
+ * a literal time arg, some don't), so we explicitly avoid it.
+ */
+function formatSeconds(sec: number): string {
+  // 6 decimal places = ~microseconds, well under one AAC frame at 48 kHz.
+  return sec.toFixed(6);
+}
+
+/**
+ * Pad or trim `audio.aac` so its exact duration matches `frameCount / fps`
+ * for the assembled video.
+ */
+export async function padOrTrimAudioToVideoFrameCount(
+  input: PadTrimAudioInput,
+): Promise<PadTrimAudioResult> {
+  const probeVideo = input.probeVideoFrameInfo ?? defaultProbeVideoFrameInfo;
+  const probeAudio = input.probeAudioInfo ?? defaultProbeAudioInfo;
+  const runner = input.runFfmpeg ?? defaultRunFfmpeg;
+
+  // Probe video and audio in parallel — the two ffprobe invocations are
+  // independent and account for most of this function's wall-clock time.
+  const [videoResult, audioResult] = await Promise.allSettled([
+    probeVideo(input.videoPath),
+    probeAudio(input.audioPath),
+  ]);
+
+  if (videoResult.status === "rejected") {
+    return failResult(
+      input.outputPath,
+      0,
+      audioResult.status === "fulfilled" ? audioResult.value.durationSeconds : 0,
+      `audioPadTrim: failed to probe video: ${(videoResult.reason as Error).message}`,
+    );
+  }
+  if (audioResult.status === "rejected") {
+    return failResult(
+      input.outputPath,
+      0,
+      0,
+      `audioPadTrim: failed to probe audio: ${(audioResult.reason as Error).message}`,
+    );
+  }
+
+  const videoInfo = videoResult.value;
+  const audioInfo = audioResult.value;
+
+  if (
+    !Number.isFinite(videoInfo.frameCount) ||
+    videoInfo.frameCount <= 0 ||
+    !Number.isFinite(videoInfo.fpsNum) ||
+    videoInfo.fpsNum <= 0 ||
+    !Number.isFinite(videoInfo.fpsDen) ||
+    videoInfo.fpsDen <= 0
+  ) {
+    return failResult(
+      input.outputPath,
+      0,
+      audioInfo.durationSeconds,
+      `audioPadTrim: invalid video frame info: ${JSON.stringify(videoInfo)}`,
+    );
+  }
+
+  const targetDurationSeconds = (videoInfo.frameCount * videoInfo.fpsDen) / videoInfo.fpsNum;
+  const { args, operation } = buildPadTrimAudioArgs(
+    input.audioPath,
+    input.outputPath,
+    audioInfo.durationSeconds,
+    targetDurationSeconds,
+  );
+
+  const ffmpegResult = await runner(args);
+  if (!ffmpegResult.success) {
+    return {
+      success: false,
+      outputPath: input.outputPath,
+      targetDurationSeconds,
+      sourceDurationSeconds: audioInfo.durationSeconds,
+      operation,
+      error: ffmpegResult.error,
+    };
+  }
+  return {
+    success: true,
+    outputPath: input.outputPath,
+    targetDurationSeconds,
+    sourceDurationSeconds: audioInfo.durationSeconds,
+    operation,
+  };
+}
+
+function failResult(
+  outputPath: string,
+  target: number,
+  source: number,
+  error: string,
+): PadTrimAudioResult {
+  return {
+    success: false,
+    outputPath,
+    targetDurationSeconds: target,
+    sourceDurationSeconds: source,
+    operation: "copy",
+    error,
+  };
+}
+
+// ── default probe/run implementations ─────────────────────────────────────
+
+interface FfprobeStreamInfo {
+  nb_read_packets?: string;
+  nb_frames?: string;
+  r_frame_rate?: string;
+}
+
+interface FfprobeOutput {
+  streams?: FfprobeStreamInfo[];
+}
+
+async function defaultProbeVideoFrameInfo(videoPath: string): Promise<ProbeVideoFrameInfo> {
+  // Try the container header (`nb_frames`) first — single moov atom read,
+  // no decode. Closed-GOP, B-frame-free streams (the only ones we'll ever
+  // ask to pad/trim) reliably set it. Fall back to `-count_packets` which
+  // walks the packet stream when the header doesn't carry the count.
+  const fastInfo = await runFfprobeJson<FfprobeOutput>([
+    "-v",
+    "error",
+    "-select_streams",
+    "v:0",
+    "-show_entries",
+    "stream=nb_frames,r_frame_rate",
+    "-of",
+    "json",
+    videoPath,
+  ]);
+  let stream = fastInfo.streams?.[0];
+  const fastCount = Number(stream?.nb_frames);
+  if (stream && Number.isFinite(fastCount) && fastCount > 0) {
+    return { frameCount: fastCount, ...parseFrameRate(stream.r_frame_rate ?? "") };
+  }
+  const slowInfo = await runFfprobeJson<FfprobeOutput>([
+    "-v",
+    "error",
+    "-select_streams",
+    "v:0",
+    "-count_packets",
+    "-show_entries",
+    "stream=nb_read_packets,r_frame_rate",
+    "-of",
+    "json",
+    videoPath,
+  ]);
+  stream = slowInfo.streams?.[0];
+  if (!stream) throw new Error(`ffprobe found no video stream in ${videoPath}`);
+  const slowCount = Number(stream.nb_read_packets);
+  if (!Number.isFinite(slowCount) || slowCount <= 0) {
+    throw new Error(`ffprobe returned no frame count: ${JSON.stringify(stream)}`);
+  }
+  return { frameCount: slowCount, ...parseFrameRate(stream.r_frame_rate ?? "") };
+}
+
+function parseFrameRate(rate: string): { fpsNum: number; fpsDen: number } {
+  const [n, d] = rate.split("/");
+  const fpsNum = Number(n);
+  const fpsDen = d === undefined ? 1 : Number(d);
+  if (!Number.isFinite(fpsNum) || !Number.isFinite(fpsDen) || fpsNum <= 0 || fpsDen <= 0) {
+    throw new Error(`Invalid r_frame_rate: ${JSON.stringify(rate)}`);
+  }
+  return { fpsNum, fpsDen };
+}
+
+async function defaultProbeAudioInfo(audioPath: string): Promise<AudioProbeInfo> {
+  // extractAudioMetadata is the shared ffprobe wrapper (caches results).
+  const metadata: AudioMetadata = await extractAudioMetadata(audioPath);
+  return { durationSeconds: metadata.durationSeconds };
+}
+
+async function defaultRunFfmpeg(args: string[]): Promise<{ success: boolean; error?: string }> {
+  const result = await runFfmpeg(args);
+  if (result.success) return { success: true };
+  return {
+    success: false,
+    error: `[audioPadTrim] ${formatFfmpegError(result.exitCode, result.stderr)}`,
+  };
+}
+
+// ── ffprobe JSON runner (shared between fast/slow video probe paths) ─────
+
+function runFfprobeJson<T>(args: string[]): Promise<T> {
+  return new Promise((resolve, reject) => {
+    const proc = spawn("ffprobe", args);
+    let stdout = "";
+    let stderr = "";
+    proc.stdout.on("data", (data: Buffer) => {
+      stdout += data.toString();
+    });
+    proc.stderr.on("data", (data: Buffer) => {
+      stderr += data.toString();
+    });
+    proc.on("error", (err) => {
+      if ((err as NodeJS.ErrnoException).code === "ENOENT") {
+        reject(new Error("[audioPadTrim] ffprobe not found. Please install FFmpeg."));
+      } else {
+        reject(err);
+      }
+    });
+    proc.on("close", (code) => {
+      if (code !== 0) {
+        reject(new Error(`ffprobe exited ${code}: ${stderr}`));
+        return;
+      }
+      try {
+        resolve(JSON.parse(stdout) as T);
+      } catch (err) {
+        reject(new Error(`Failed to parse ffprobe output: ${(err as Error).message}`));
+      }
+    });
+  });
+}