Build a Media Transcoding Pipeline

import { pool } from "../db";
import { Redis } from "ioredis";
import { randomBytes } from "node:crypto";
import { execSync, exec } from "node:child_process";
const redis = new Redis(process.env.REDIS_URL!);

interface TranscodeJob {
  id: string;
  inputPath: string;
  outputPaths: Record<string, string>;
  status: "queued" | "processing" | "completed" | "failed";
  progress: number;
  profiles: TranscodeProfile[];
  thumbnails: string[];
  metadata: { duration: number; width: number; height: number; codec: string; size: number };
  webhookUrl: string | null;
  createdAt: string;
}

interface TranscodeProfile {
  name: string;
  width: number;
  height: number;
  bitrate: string;
  codec: string;
  format: string;
}

const PROFILES: TranscodeProfile[] = [
  { name: "1080p", width: 1920, height: 1080, bitrate: "5000k", codec: "libx264", format: "mp4" },
  { name: "720p", width: 1280, height: 720, bitrate: "2500k", codec: "libx264", format: "mp4" },
  { name: "480p", width: 854, height: 480, bitrate: "1000k", codec: "libx264", format: "mp4" },
  { name: "360p", width: 640, height: 360, bitrate: "500k", codec: "libx264", format: "mp4" },
];

export async function submitTranscode(params: { inputPath: string; profiles?: string[]; webhookUrl?: string }): Promise<TranscodeJob> {
  const id = `transcode-${randomBytes(8).toString("hex")}`;
  const selectedProfiles = params.profiles ? PROFILES.filter((p) => params.profiles!.includes(p.name)) : PROFILES;

  // Extract metadata
  const probeResult = execSync(`ffprobe -v quiet -print_format json -show_format -show_streams ${params.inputPath}`, { encoding: "utf-8" });
  const probe = JSON.parse(probeResult);
  const videoStream = probe.streams.find((s: any) => s.codec_type === "video");

  const metadata = {
    duration: parseFloat(probe.format.duration || "0"),
    width: videoStream?.width || 0,
    height: videoStream?.height || 0,
    codec: videoStream?.codec_name || "unknown",
    size: parseInt(probe.format.size || "0"),
  };

  // Filter profiles that are smaller than input
  const applicableProfiles = selectedProfiles.filter((p) => p.width <= metadata.width);

  const job: TranscodeJob = {
    id, inputPath: params.inputPath,
    outputPaths: {}, status: "queued", progress: 0,
    profiles: applicableProfiles, thumbnails: [],
    metadata, webhookUrl: params.webhookUrl || null,
    createdAt: new Date().toISOString(),
  };

  await pool.query(
    `INSERT INTO transcode_jobs (id, input_path, status, profiles, metadata, webhook_url, created_at) VALUES ($1, $2, 'queued', $3, $4, $5, NOW())`,
    [id, params.inputPath, JSON.stringify(applicableProfiles), JSON.stringify(metadata), params.webhookUrl]
  );

  await redis.rpush("transcode:queue", id);
  return job;
}

export async function processJob(jobId: string): Promise<void> {
  await pool.query("UPDATE transcode_jobs SET status = 'processing' WHERE id = $1", [jobId]);
  const { rows: [job] } = await pool.query("SELECT * FROM transcode_jobs WHERE id = $1", [jobId]);
  const profiles: TranscodeProfile[] = JSON.parse(job.profiles);
  const outputPaths: Record<string, string> = {};
  const totalSteps = profiles.length + 1; // profiles + thumbnails
  let completed = 0;

  try {
    // Transcode each profile
    for (const profile of profiles) {
      const outputPath = `/tmp/output/${jobId}_${profile.name}.${profile.format}`;
      execSync(
        `ffmpeg -i ${job.input_path} -vf scale=${profile.width}:${profile.height} -b:v ${profile.bitrate} -c:v ${profile.codec} -c:a aac -b:a 128k -movflags +faststart -y ${outputPath}`,
        { timeout: 600000 }
      );
      outputPaths[profile.name] = outputPath;
      completed++;
      const progress = Math.round((completed / totalSteps) * 100);
      await redis.set(`transcode:progress:${jobId}`, progress);
    }

    // Generate thumbnails
    const metadata = JSON.parse(job.metadata);
    const thumbnails: string[] = [];
    const intervals = [0, 25, 50, 75];
    for (const pct of intervals) {
      const time = Math.floor(metadata.duration * pct / 100);
      const thumbPath = `/tmp/output/${jobId}_thumb_${pct}.jpg`;
      execSync(`ffmpeg -i ${job.input_path} -ss ${time} -vframes 1 -vf scale=320:-1 -y ${thumbPath}`, { timeout: 30000 });
      thumbnails.push(thumbPath);
    }

    // Generate HLS playlist for adaptive streaming
    // In production: create .m3u8 master playlist pointing to each quality level

    await pool.query(
      "UPDATE transcode_jobs SET status = 'completed', output_paths = $2, thumbnails = $3, progress = 100 WHERE id = $1",
      [jobId, JSON.stringify(outputPaths), JSON.stringify(thumbnails)]
    );

    if (job.webhook_url) {
      await fetch(job.webhook_url, {
        method: "POST", headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ jobId, status: "completed", outputPaths, thumbnails }),
      }).catch(() => {});
    }
  } catch (error: any) {
    await pool.query("UPDATE transcode_jobs SET status = 'failed', error = $2 WHERE id = $1", [jobId, error.message]);
    if (job.webhook_url) {
      await fetch(job.webhook_url, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ jobId, status: "failed", error: error.message }) }).catch(() => {});
    }
  }
}

export async function getProgress(jobId: string): Promise<number> {
  return parseInt(await redis.get(`transcode:progress:${jobId}`) || "0");
}