Initial commit

2026-06-11 10:46:09 +02:00
commit 81ad7e4ea9
223 changed files with 39530 additions and 0 deletions
@@ -0,0 +1,60 @@
+// app/api/transcoder/claim/route.ts
+// Atomically claims the next available transcoding job.
+// Uses SELECT … FOR UPDATE SKIP LOCKED so multiple workers never race on the
+// same video.
+
+import { NextResponse } from "next/server";
+import { prisma } from "@/lib/prisma";
+import { Prisma } from "@prisma/client";
+import { verifyTranscoderToken } from "@/lib/transcoder-auth";
+
+export const dynamic = "force-dynamic";
+
+export async function POST(request: Request) {
+  if (!verifyTranscoderToken(request)) {
+    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+  }
+
+  try {
+    const video = await prisma.$transaction(async (tx) => {
+      // Atomically lock the oldest uploaded video, skipping rows already
+      // locked by concurrent workers.
+      const rows = await tx.$queryRaw<{ id: string }[]>(
+        Prisma.sql`
+          SELECT id
+          FROM   "Video"
+          WHERE  "transcodingStatus" = 'uploaded'
+          ORDER  BY "createdAt" ASC
+          LIMIT  1
+          FOR    UPDATE SKIP LOCKED
+        `
+      );
+
+      if (rows.length === 0) return null;
+
+      return tx.video.update({
+        where: { id: rows[0].id },
+        data: { transcodingStatus: "processing" },
+      });
+    });
+
+    if (!video) {
+      // No work available – return null body so the worker knows to stop.
+      return NextResponse.json(null, { status: 200 });
+    }
+
+    // Build the download URL from the public CMS base URL.
+    const cmsBase =
+      process.env.NEXTAUTH_URL?.replace(/\/$/, "") ??
+      process.env.CMS_PUBLIC_URL?.replace(/\/$/, "") ??
+      "";
+
+    return NextResponse.json({
+      videoId: video.id,
+      downloadUrl: `${cmsBase}/api/transcoder/download/${video.id}`,
+    });
+  } catch (err) {
+    console.error("[Transcoder Claim] Error:", err);
+    return NextResponse.json({ error: "Internal server error" }, { status: 500 });
+  }
+}
@@ -0,0 +1,72 @@
+// app/api/transcoder/download/[videoId]/route.ts
+// Streams the original MP4 directly to the remote transcoder worker.
+// Never buffers the file in memory.
+
+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 { prisma } from "@/lib/prisma";
+import { verifyTranscoderToken } from "@/lib/transcoder-auth";
+
+// Allow up to 45 minutes for large file transfers.
+export const maxDuration = 2700;
+
+const UPLOADS_DIR = process.env.UPLOADS_DIR ?? "/uploads";
+const ORIGINALS_DIR = path.join(UPLOADS_DIR, "videos");
+
+// Narrow character set – CUIDs are alphanumeric plus underscore/dash.
+const SAFE_ID = /^[a-zA-Z0-9_-]{1,64}$/;
+
+export async function GET(
+  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 });
+  }
+
+  // Only allow download while the job is actively claimed.
+  if (video.transcodingStatus !== "processing") {
+    return NextResponse.json(
+      { error: "Video is not in processing state" },
+      { status: 409 }
+    );
+  }
+
+  const filePath = path.join(ORIGINALS_DIR, `${videoId}.mp4`);
+
+  // Security: ensure the resolved path stays within ORIGINALS_DIR.
+  const resolved = path.resolve(filePath);
+  if (!resolved.startsWith(path.resolve(ORIGINALS_DIR))) {
+    return NextResponse.json({ error: "Forbidden" }, { status: 403 });
+  }
+
+  try {
+    const stat = await fs.stat(resolved);
+    const nodeStream = fssync.createReadStream(resolved);
+    const webStream = Readable.toWeb(nodeStream) as ReadableStream;
+
+    return new Response(webStream, {
+      headers: {
+        "Content-Type": "video/mp4",
+        "Content-Length": stat.size.toString(),
+        "Content-Disposition": `attachment; filename="${videoId}.mp4"`,
+      },
+    });
+  } catch {
+    return NextResponse.json({ error: "File not found on disk" }, { status: 404 });
+  }
+}
@@ -0,0 +1,54 @@
+// app/api/transcoder/fail/[videoId]/route.ts
+// Marks a video job as failed. Called by the remote worker when transcoding
+// or upload encounters an unrecoverable error.
+
+import { NextResponse } from "next/server";
+import { prisma } from "@/lib/prisma";
+import { verifyTranscoderToken } from "@/lib/transcoder-auth";
+
+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 });
+  }
+
+  try {
+    await prisma.video.update({
+      where: { id: videoId },
+      data: { transcodingStatus: "failed" },
+    });
+
+    // Log the error message from the worker if provided.
+    let workerError = "";
+    try {
+      const body = await request.json();
+      workerError = typeof body?.error === "string" ? body.error : "";
+    } catch {
+      // Body may be empty – that's fine.
+    }
+
+    console.error(
+      `[Transcoder Fail] ${videoId}${workerError ? ` – ${workerError}` : ""}`
+    );
+
+    return NextResponse.json({ success: true });
+  } catch (err) {
+    console.error(`[Transcoder Fail] DB error for ${videoId}:`, err);
+    return NextResponse.json({ error: "Internal server error" }, { status: 500 });
+  }
+}
@@ -0,0 +1,140 @@
+// 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(
+      request.body as 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 }
+    );
+  }
+}