// 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 ); 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 } ); } }