Vault/transcoder/index.ts

// transcoder.ts

import { spawn } from "child_process";
import * as fs from "fs/promises";
import * as fssync from "fs";
import * as path from "path";
import { PrismaClient } from "@prisma/client";

const prisma = new PrismaClient();

process.umask(0o022);


// -----------------------------
// Configuration
// -----------------------------
const UPLOADS_DIR = process.env.UPLOADS_DIR || "/uploads";
const ORIGINALS_DIR = path.join(UPLOADS_DIR, "videos");
const HLS_ROOT = path.join(UPLOADS_DIR, "hls");

// Safer bitrate ladder for education content
const HLS_PRESETS = [
  { name: "1080p", width: 1920, height: 1080, bitrate: "3500k", maxrate: "4000k" },
  { name: "720p", width: 1280, height: 720, bitrate: "1800k", maxrate: "2000k" },
  { name: "480p", width: 854, height: 480, bitrate: "900k", maxrate: "1000k" },
];

// -----------------------------
// Utilities
// -----------------------------

async function fileExists(p: string): Promise<boolean> {
  try {
    await fs.access(p);
    return true;
  } catch {
    return false;
  }
}

function runFFmpeg(args: string[]): Promise<void> {
  return new Promise((resolve, reject) => {
    const ffmpeg = spawn("ffmpeg", args, { stdio: "inherit" });

    ffmpeg.on("close", (code) => {
      if (code === 0) resolve();
      else reject(new Error(`FFmpeg exited with code ${code}`));
    });
  });
}

async function probeResolution(input: string): Promise<{ width: number; height: number }> {
  return new Promise((resolve, reject) => {
    const ffprobe = spawn("ffprobe", [
      "-v", "error",
      "-select_streams", "v:0",
      "-show_entries", "stream=width,height",
      "-of", "csv=s=x:p=0",
      input,
    ]);

    let output = "";
    ffprobe.stdout.on("data", (data) => {
      output += data.toString();
    });

    ffprobe.on("close", (code) => {
      if (code !== 0) return reject(new Error("ffprobe failed"));
      const [width, height] = output.trim().split("x").map(Number);
      resolve({ width, height });
    });
  });
}

// -----------------------------
// HLS Creation
// -----------------------------

async function createVariant(
  input: string,
  outputDir: string,
  preset: typeof HLS_PRESETS[number]
) {
  const segmentPattern = path.join(outputDir, `${preset.name}_%03d.ts`);
  const playlistPath = path.join(outputDir, `${preset.name}.m3u8`);

  const args = [
    "-y",
    "-i", input,

    "-c:v", "libx264",
    "-preset", "medium",
    "-profile:v", "main",
    "-crf", "20",

    "-vf", `scale=w='min(${preset.width},iw)':h='min(${preset.height},ih)':force_original_aspect_ratio=decrease:force_divisible_by=2`,


    "-b:v", preset.bitrate,
    "-maxrate", preset.maxrate,
    "-bufsize", "4000k",

    "-c:a", "aac",
    "-b:a", "128k",

    "-f", "hls",
    "-hls_time", "6",
    "-hls_playlist_type", "vod",
    "-hls_flags", "independent_segments",
    "-hls_segment_filename", segmentPattern,

    playlistPath,
  ];

  console.log(`[HLS] Creating ${preset.name}`);
  await runFFmpeg(args);
}

async function createMasterPlaylist(outputDir: string, variants: string[]): Promise<void> {
  const lines: string[] = [
    "#EXTM3U",
    "#EXT-X-VERSION:3",
  ];

  for (const variant of variants) {
    const preset = HLS_PRESETS.find(p => p.name === variant)!;
    const bandwidth = parseInt(preset.maxrate.replace("k", "")) * 1000;

    lines.push(
      `#EXT-X-STREAM-INF:BANDWIDTH=${bandwidth},RESOLUTION=${preset.width}x${preset.height}`
    );
    lines.push(`${preset.name}.m3u8`);
  }

  await fs.writeFile(path.join(outputDir, "master.m3u8"), lines.join("\n"));
}

// -----------------------------
// Transcode One Video
// -----------------------------

async function transcodeVideo(videoId: string) {
  console.log(`\n[Transcoding] ${videoId}`);

  const inputPath = path.join(ORIGINALS_DIR, `${videoId}.mp4`);
  const finalDir = path.join(HLS_ROOT, videoId);
  const tempDir = path.join(HLS_ROOT, `${videoId}.tmp`);

  if (!(await fileExists(inputPath))) {
    throw new Error(`Original file not found: ${inputPath}`);
  }

  if (await fileExists(finalDir)) {
    console.log(`[Skip] HLS already exists`);
    return;
  }

  await fs.mkdir(tempDir, { recursive: true });

  const { width, height } = await probeResolution(inputPath);

  const allowedVariants = HLS_PRESETS.filter(
    p => p.width <= width && p.height <= height
  );

  if (allowedVariants.length === 0) {
    throw new Error("No suitable HLS variants for source resolution");
  }

  for (const preset of allowedVariants) {
    await createVariant(inputPath, tempDir, preset);
  }

  await createMasterPlaylist(tempDir, allowedVariants.map(p => p.name));

  // Atomic rename
  await fs.rename(tempDir, finalDir);

  console.log(`[Done] ${videoId}`);
}

// -----------------------------
// Main Batch Worker
// -----------------------------

async function main() {
  console.log("[Transcoder] Starting batch run");

  await fs.mkdir(HLS_ROOT, { recursive: true });

  const videos = await prisma.video.findMany({
    where: { transcodingStatus: "uploaded" },
  });

  if (videos.length === 0) {
    console.log("[Transcoder] Nothing to process");
    return;
  }

  console.log(`[Transcoder] Found ${videos.length} videos`);

  for (const video of videos) {
    try {
      // Atomically lock job
      await prisma.video.update({
        where: { id: video.id },
        data: { transcodingStatus: "processing" },
      });

      await transcodeVideo(video.id);

      await prisma.video.update({
        where: { id: video.id },
        data: { transcodingStatus: "transcoded" },
      });

    } catch (err) {
      console.error(`[Error] ${video.id}`, err);

      await prisma.video.update({
        where: { id: video.id },
        data: { transcodingStatus: "failed" },
      });
    }
  }

  await prisma.$disconnect();
  console.log("[Transcoder] Batch complete");
}

main().catch(async (err) => {
  console.error("[Fatal]", err);
  await prisma.$disconnect();
  process.exit(1);
});