Build a Video Transcoding Queue

// src/video/transcoder.ts — Video transcoding queue with multi-resolution and progress tracking
import { pool } from "../db";
import { Redis } from "ioredis";
import { randomBytes } from "node:crypto";
import { spawn } from "node:child_process";
import { stat } from "node:fs/promises";

const redis = new Redis(process.env.REDIS_URL!);

interface TranscodeJob {
  id: string;
  videoId: string;
  inputPath: string;
  outputPaths: Record<string, string>;
  status: "queued" | "processing" | "completed" | "failed";
  progress: number;            // 0-100
  renditions: Rendition[];
  thumbnails: string[];
  metadata: VideoMetadata;
  priority: number;
  workerId: string | null;
  startedAt: string | null;
  completedAt: string | null;
  error: string | null;
  createdAt: string;
}

interface Rendition {
  name: string;
  width: number;
  height: number;
  bitrate: string;             // e.g., "4000k"
  codec: string;
  status: "pending" | "processing" | "completed" | "failed";
  outputPath: string;
  fileSize: number;
  duration: number;
}

interface VideoMetadata {
  duration: number;            // seconds
  width: number;
  height: number;
  codec: string;
  fps: number;
  bitrate: number;
  fileSize: number;
  audioCodec: string;
}

const RENDITION_PRESETS: Omit<Rendition, "status" | "outputPath" | "fileSize" | "duration">[] = [
  { name: "1080p", width: 1920, height: 1080, bitrate: "4000k", codec: "libx264" },
  { name: "720p", width: 1280, height: 720, bitrate: "2500k", codec: "libx264" },
  { name: "480p", width: 854, height: 480, bitrate: "1200k", codec: "libx264" },
  { name: "360p", width: 640, height: 360, bitrate: "600k", codec: "libx264" },
  { name: "audio", width: 0, height: 0, bitrate: "128k", codec: "aac" },
];

// Submit video for transcoding
export async function submitJob(params: {
  videoId: string;
  inputPath: string;
  priority?: number;
}): Promise<TranscodeJob> {
  const id = `tj-${randomBytes(6).toString("hex")}`;

  // Probe input video metadata
  const metadata = await probeVideo(params.inputPath);

  // Determine which renditions are needed (skip upscaling)
  const applicableRenditions = RENDITION_PRESETS.filter(
    (r) => r.name === "audio" || r.height <= metadata.height
  );

  const outputDir = `/storage/transcoded/${params.videoId}`;
  const renditions: Rendition[] = applicableRenditions.map((preset) => ({
    ...preset,
    status: "pending",
    outputPath: preset.name === "audio"
      ? `${outputDir}/audio.m4a`
      : `${outputDir}/${preset.name}.mp4`,
    fileSize: 0,
    duration: metadata.duration,
  }));

  const job: TranscodeJob = {
    id, videoId: params.videoId,
    inputPath: params.inputPath,
    outputPaths: {},
    status: "queued",
    progress: 0,
    renditions,
    thumbnails: [],
    metadata,
    priority: params.priority || 5,
    workerId: null,
    startedAt: null,
    completedAt: null,
    error: null,
    createdAt: new Date().toISOString(),
  };

  await pool.query(
    `INSERT INTO transcode_jobs (id, video_id, input_path, status, renditions, metadata, priority, created_at)
     VALUES ($1, $2, $3, 'queued', $4, $5, $6, NOW())`,
    [id, params.videoId, params.inputPath, JSON.stringify(renditions), JSON.stringify(metadata), job.priority]
  );

  // Add to priority queue
  const score = job.priority * 1e13 + Date.now();
  await redis.zadd("transcode:queue", score, id);

  return job;
}

// Process next job from queue (called by workers)
export async function processNextJob(workerId: string): Promise<TranscodeJob | null> {
  const result = await redis.zpopmin("transcode:queue");
  if (!result || result.length === 0) return null;

  const jobId = result[0];
  const { rows: [row] } = await pool.query("SELECT * FROM transcode_jobs WHERE id = $1", [jobId]);
  if (!row) return null;

  const job: TranscodeJob = { ...row, renditions: JSON.parse(row.renditions), metadata: JSON.parse(row.metadata) };
  job.status = "processing";
  job.workerId = workerId;
  job.startedAt = new Date().toISOString();

  await pool.query(
    "UPDATE transcode_jobs SET status='processing', worker_id=$2, started_at=NOW() WHERE id=$1",
    [jobId, workerId]
  );

  // Process each rendition
  const totalRenditions = job.renditions.length;
  let completed = 0;

  for (const rendition of job.renditions) {
    rendition.status = "processing";
    await updateProgress(job, (completed / totalRenditions) * 100);

    try {
      if (rendition.name === "audio") {
        await transcodeAudio(job.inputPath, rendition);
      } else {
        await transcodeVideo(job.inputPath, rendition, job.metadata);
      }
      rendition.status = "completed";

      const stats = await stat(rendition.outputPath).catch(() => null);
      rendition.fileSize = stats?.size || 0;
    } catch (err: any) {
      rendition.status = "failed";
      job.error = `Rendition ${rendition.name} failed: ${err.message}`;
    }

    completed++;
    await updateProgress(job, (completed / totalRenditions) * 100);
  }

  // Generate thumbnails
  job.thumbnails = await generateThumbnails(job.inputPath, job.metadata.duration);

  // Finalize
  const allCompleted = job.renditions.every((r) => r.status === "completed");
  job.status = allCompleted ? "completed" : "failed";
  job.completedAt = new Date().toISOString();
  job.progress = 100;

  await pool.query(
    "UPDATE transcode_jobs SET status=$2, renditions=$3, thumbnails=$4, progress=100, completed_at=NOW() WHERE id=$1",
    [job.id, job.status, JSON.stringify(job.renditions), JSON.stringify(job.thumbnails)]
  );

  return job;
}

async function transcodeVideo(inputPath: string, rendition: Rendition, metadata: VideoMetadata): Promise<void> {
  return new Promise((resolve, reject) => {
    const args = [
      "-i", inputPath,
      "-vf", `scale=${rendition.width}:${rendition.height}`,
      "-c:v", rendition.codec,
      "-b:v", rendition.bitrate,
      "-c:a", "aac",
      "-b:a", "128k",
      "-movflags", "+faststart",  // enable progressive download
      "-preset", "medium",
      "-y",
      rendition.outputPath,
    ];

    const proc = spawn("ffmpeg", args);
    proc.on("close", (code) => code === 0 ? resolve() : reject(new Error(`FFmpeg exited ${code}`)));
    proc.on("error", reject);
  });
}

async function transcodeAudio(inputPath: string, rendition: Rendition): Promise<void> {
  return new Promise((resolve, reject) => {
    const proc = spawn("ffmpeg", [
      "-i", inputPath, "-vn", "-c:a", "aac", "-b:a", rendition.bitrate, "-y", rendition.outputPath,
    ]);
    proc.on("close", (code) => code === 0 ? resolve() : reject(new Error(`FFmpeg exited ${code}`)));
    proc.on("error", reject);
  });
}

async function generateThumbnails(inputPath: string, duration: number): Promise<string[]> {
  const timestamps = [1, duration * 0.25, duration * 0.5, duration * 0.75].filter((t) => t < duration);
  const thumbnails: string[] = [];

  for (const ts of timestamps) {
    const outPath = inputPath.replace(/\.[^.]+$/, `_thumb_${Math.floor(ts)}.jpg`);
    await new Promise<void>((resolve) => {
      const proc = spawn("ffmpeg", ["-i", inputPath, "-ss", String(ts), "-frames:v", "1", "-q:v", "3", "-y", outPath]);
      proc.on("close", () => resolve());
    });
    thumbnails.push(outPath);
  }

  return thumbnails;
}

async function probeVideo(inputPath: string): Promise<VideoMetadata> {
  // In production: calls ffprobe
  return { duration: 120, width: 1920, height: 1080, codec: "h264", fps: 30, bitrate: 8000000, fileSize: 0, audioCodec: "aac" };
}

async function updateProgress(job: TranscodeJob, progress: number): Promise<void> {
  job.progress = Math.round(progress);
  await redis.setex(`transcode:progress:${job.id}`, 3600, String(job.progress));
  await redis.publish("transcode:progress", JSON.stringify({ jobId: job.id, progress: job.progress }));
}

// Get job progress (for client polling)
export async function getJobProgress(jobId: string): Promise<{ progress: number; status: string }> {
  const progress = await redis.get(`transcode:progress:${jobId}`);
  const { rows: [row] } = await pool.query("SELECT status FROM transcode_jobs WHERE id = $1", [jobId]);
  return { progress: parseInt(progress || "0"), status: row?.status || "unknown" };
}