How I Built a Bulk Video Processing Pipeline with Next.js, BullMQ, and FFmpeg

for (const video of videos) {
  await processVideo(video);
}

Browser → Next.js API → BullMQ Queue 
                              ↓
                     Worker Pool (N workers)
                     ├── Worker 1 → FFmpeg
                     ├── Worker 2 → FFmpeg  
                     ├── Worker 3 → Whisper
                     └── Worker 4 → FFmpeg
                              ↓
                     Cloudflare R2 (output)
                              ↓
                     Webhook → User notified

Browser → Next.js → R2

// Generate presigned URL server-side
const url = await getSignedUrl(
  s3Client,
  new PutObjectCommand({
    Bucket: process.env.R2_BUCKET_NAME,
    Key: `uploads/${uuid}-${filename}`,
    ContentType: file.type,
  }),
  { expiresIn: 3600 }
);
// Return URL to browser
return { url, key };

// Browser uploads directly to R2
// Using XHR for progress tracking
function uploadWithProgress(
  url: string,
  file: File,
  onProgress: (pct: number) => void
): Promise<void> {
  return new Promise((resolve, reject) => {
    const xhr = new XMLHttpRequest();
    xhr.open("PUT", url);
    xhr.setRequestHeader("Content-Type", file.type);
    xhr.upload.onprogress = (e) => {
      if (e.lengthComputable) {
        onProgress(Math.round(e.loaded / e.total * 100));
      }
    };
    xhr.onload = () => 
      xhr.status === 200 ? resolve() : reject();
    xhr.send(file);
  });
}

// p-limit prevents overwhelming 
// the connection on slow networks
import pLimit from "p-limit";
const limit = pLimit(5);
await Promise.all(
  files.map((file, i) =>
    limit(() => uploadWithProgress(
      presignedUrls[i].url,
      file,
      (pct) => updateProgress(i, pct)
    ))
  )
);

// Add all jobs to queue at once
await Promise.all(
  uploadedFiles.map((file) =>
    videoQueue.add("process-video", {
      userId,
      videoId: file.id,
      key: file.r2Key,
      settings: presetSettings,
    })
  )
);

// workers/videoProcessor.ts
const worker = new Worker(
  "video-processing",
  async (job) => {
    const { videoId, key, settings } = job.data;
    const startTime = Date.now();
    // Download from R2
    const videoBuffer = await downloadFromR2(key);
    // Transcribe with Whisper
    const transcription = await transcribeAudio(
      videoBuffer
    );
    // Remove silence using FFmpeg
    const silenceRemoved = await removeSilence(
      videoBuffer,
      transcription.sentences,
      settings.silenceLevel
    );
    // Burn captions
    const withCaptions = await burnCaptions(
      silenceRemoved,
      transcription.sentences,
      settings.captionStyle
    );
    // Apply zoom
    const withZoom = await applyZoom(
      withCaptions,
      transcription.sentences,
      settings.zoomLevel
    );
    // Crop for platform
    const output = await cropVideo(
      withZoom,
      settings.cropFormat
    );
    // Upload output to R2
    const outputKey = await uploadToR2(output);
    // Store processing time for analytics
    const processingTime = 
      (Date.now() - startTime) / 1000;
    await db.update(processedVideos)
      .set({
        status: "completed",
        outputKey,
        totalProcessingTimeSeconds: 
          Math.round(processingTime),
      })
      .where(eq(processedVideos.id, videoId));
  },
  {
    connection: redisConnection,
    concurrency: 10, // 10 videos simultaneously
  }
);

async function detectSilence(
  inputPath: string,
  threshold: number = -30, // dB
  minDuration: number = 1.5 // seconds
): Promise<{ start: number; end: number }[]> {
  return new Promise((resolve, reject) => {
    const silences: { start: number; end: number }[] = [];
    let currentStart: number | null = null;
    const ffmpeg = spawn("ffmpeg", [
      "-i", inputPath,
      "-af", `silencedetect=noise=${threshold}dB:d=${minDuration}`,
      "-f", "null",
      "-",
    ]);
    ffmpeg.stderr.on("data", (data: Buffer) => {
      const output = data.toString();
      const startMatch = output.match(
        /silence_start: ([\d.]+)/
      );
      const endMatch = output.match(
        /silence_end: ([\d.]+)/
      );
      if (startMatch) {
        currentStart = parseFloat(startMatch[1]);
      }
      if (endMatch && currentStart !== null) {
        silences.push({
          start: currentStart,
          end: parseFloat(endMatch[1]),
        });
        currentStart = null;
      }
    });
    ffmpeg.on("close", () => resolve(silences));
    ffmpeg.on("error", reject);
  });
}

async function removeSilences(
  inputPath: string,
  silences: { start: number; end: number }[],
  outputPath: string
): Promise<void> {
  // Build FFmpeg filter to keep 
  // everything except silence segments
  const keepSegments = getKeepSegments(
    silences, 
    videoDuration
  );
  const filterParts = keepSegments.map(
    (seg, i) => 
      `[0:v]trim=${seg.start}:${seg.end},` +
      `setpts=PTS-STARTPTS[v${i}];` +
      `[0:a]atrim=${seg.start}:${seg.end},` +
      `asetpts=PTS-STARTPTS[a${i}]`
  );
  const concatInputs = keepSegments
    .map((_, i) => `[v${i}][a${i}]`)
    .join("");
  const filter = [
    ...filterParts,
    `${concatInputs}concat=n=${keepSegments.length}` +
    `:v=1:a=1[outv][outa]`
  ].join(";");
  await runFFmpeg([
    "-i", inputPath,
    "-filter_complex", filter,
    "-map", "[outv]",
    "-map", "[outa]",
    outputPath,
  ]);
}

const transcription = await openai.audio
  .transcriptions.create({
    file: audioFile,
    model: "whisper-1",
    response_format: "verbose_json",
    timestamp_granularities: ["word"],
  });
// transcription.words contains:
// [{ word: "Hello", start: 0.0, end: 0.3 }]

function buildCaptionFilter(
  words: { word: string; start: number; end: number }[],
  style: CaptionStyle
): string {
  return words.map((word) => {
    const escaped = word.word
      .replace(/'/g, "\\'")
      .replace(/:/g, "\\:");
    return (
      `drawtext=text='${escaped}':` +
      `fontsize=${style.fontSize}:` +
      `fontcolor=${style.color}:` +
      `x=(w-text_w)/2:` +
      `y=(h-text_h)/2:` +
      `enable='between(t,${word.start},${word.end})'`
    );
  }).join(",");
}