Vault/app/api/transcoder/upload/[videoId]/route.ts

// app/api/transcoder/upload/[videoId]/route.ts
// Receives a ZIP archive of the completed HLS package from the remote worker,
// extracts it to disk, validates it, renames the temp dir to its final name,
// and marks the video as transcoded.
//
// The request body must be raw application/zip (no multipart wrapper).
// The file is streamed to disk before extraction – never fully buffered in RAM.

import { NextResponse } from "next/server";
import * as fssync from "fs";
import * as fs from "fs/promises";
import * as path from "path";
import { Readable } from "stream";
import { pipeline } from "stream/promises";
import * as unzipper from "unzipper";
import { prisma } from "@/lib/prisma";
import { verifyTranscoderToken } from "@/lib/transcoder-auth";

// Allow up to 45 minutes for very large uploads.
export const maxDuration = 2700;

const UPLOADS_DIR = process.env.UPLOADS_DIR ?? "/uploads";
const HLS_ROOT = path.join(UPLOADS_DIR, "hls");

const SAFE_ID = /^[a-zA-Z0-9_-]{1,64}$/;

export async function POST(
  request: Request,
  context: { params: Promise<{ videoId: string }> }
) {
  if (!verifyTranscoderToken(request)) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  const { videoId } = await context.params;

  if (!SAFE_ID.test(videoId)) {
    return NextResponse.json({ error: "Invalid video ID" }, { status: 400 });
  }

  const video = await prisma.video.findUnique({ where: { id: videoId } });
  if (!video) {
    return NextResponse.json({ error: "Not found" }, { status: 404 });
  }
  if (video.transcodingStatus !== "processing") {
    return NextResponse.json(
      { error: "Video is not in processing state" },
      { status: 409 }
    );
  }

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

  const tempZipPath = path.join(HLS_ROOT, `${videoId}.incoming.zip`);
  const tempDir = path.join(HLS_ROOT, `${videoId}.tmp`);
  const finalDir = path.join(HLS_ROOT, videoId);

  // Security: path traversal guard.
  for (const p of [tempZipPath, tempDir, finalDir]) {
    if (!path.resolve(p).startsWith(path.resolve(HLS_ROOT))) {
      return NextResponse.json({ error: "Forbidden" }, { status: 403 });
    }
  }

  try {
    if (!request.body) {
      return NextResponse.json({ error: "Empty request body" }, { status: 400 });
    }

    // 1. Stream upload body to a temp zip file on disk.
    const nodeReadable = Readable.fromWeb(
      // DOM ReadableStream and node:stream/web ReadableStream are structurally
      // different types; fromWeb expects the latter
      request.body as unknown as import("node:stream/web").ReadableStream<Uint8Array>
    );
    const writeStream = fssync.createWriteStream(tempZipPath);
    await pipeline(nodeReadable, writeStream);

    // 2. Prepare extraction directory.
    if (fssync.existsSync(tempDir)) {
      await fs.rm(tempDir, { recursive: true, force: true });
    }
    await fs.mkdir(tempDir, { recursive: true });

    // 3. Stream-extract the zip.
    await fssync
      .createReadStream(tempZipPath)
      .pipe(unzipper.Extract({ path: tempDir }))
      .promise();

    // 4. Remove temp zip.
    await fs.unlink(tempZipPath).catch(() => {});

    // 5. Validate contents.
    const masterPath = path.join(tempDir, "master.m3u8");
    if (!fssync.existsSync(masterPath)) {
      await fs.rm(tempDir, { recursive: true, force: true }).catch(() => {});
      return NextResponse.json(
        { error: "master.m3u8 not found in archive" },
        { status: 422 }
      );
    }

    const files = await fs.readdir(tempDir);
    const hasVariant = files.some(
      (f) => f.endsWith(".m3u8") && f !== "master.m3u8"
    );
    if (!hasVariant) {
      await fs.rm(tempDir, { recursive: true, force: true }).catch(() => {});
      return NextResponse.json(
        { error: "No variant playlist found in archive" },
        { status: 422 }
      );
    }

    // 6. Remove any stale final directory and atomically rename.
    if (fssync.existsSync(finalDir)) {
      await fs.rm(finalDir, { recursive: true, force: true });
    }
    await fs.rename(tempDir, finalDir);

    // 7. Mark as transcoded in the database.
    await prisma.video.update({
      where: { id: videoId },
      data: { transcodingStatus: "transcoded" },
    });

    console.log(`[Transcoder Upload] ${videoId} – success (${files.length} files)`);
    return NextResponse.json({ success: true });
  } catch (err) {
    console.error(`[Transcoder Upload] Error for ${videoId}:`, err);

    // Best-effort cleanup.
    await fs.unlink(tempZipPath).catch(() => {});
    await fs.rm(tempDir, { recursive: true, force: true }).catch(() => {});

    return NextResponse.json(
      { error: "Upload processing failed" },
      { status: 500 }
    );
  }
}