Direct-to-bucket video upload + transcode pipeline

- server/videoUpload.ts: S3 multipart presign helpers, bucket inventory listing - routes: /api/admin/uploads/{init,part-url,complete,abort}, /api/admin/inventory, /api/internal/transcode-done (worker callback, bearer INTERNAL_TOKEN) - admin UI: browser-to-bucket uploader with progress, bucket-vs-database panel Originals land in folx-tv/_incoming/{guid}/, the worker on the remotion box packages them to HLS and reports back.
2026-08-05 10:25:18 +00:00
parent f6c73f543c
commit 49d9c83201
5 changed files with 747 additions and 0 deletions
@@ -19,6 +19,14 @@ import fetch from "node-fetch";
 import { setupAuth, isAuthenticated, isAdmin } from "./replitAuth";
 import { ObjectStorageService, ObjectNotFoundError } from "./objectStorage";
 import { generateVideoDescription, generateBulkDescriptions } from "./aiService";
+import { db } from "./db";
+import { videos } from "@shared/schema";
+import { eq } from "drizzle-orm";
+import {
+  initUpload, partUrl, completeUpload, abortUpload, writeJobFile,
+  listPackagedGuids, listPendingJobs, publicUrlFor,
+  FOLX_PREFIX, INCOMING_PREFIX,
+} from "./videoUpload";

 // Extract unique artist names from video titles
 function extractArtists(videos: any[]): { name: string; videoCount: number; videos: any[] }[] {
@@ -2134,6 +2142,174 @@ Sitemap: ${baseUrl}/sitemap.xml
    }
  });

+
+  // ────────────────────────────────────────────────────────────────────
+  //  DIRECT-TO-BUCKET VIDEO UPLOAD  (browser → Hetzner S3, multipart)
+  //  Flow: init → part-url (per chunk) → complete → worker transcodes
+  // ────────────────────────────────────────────────────────────────────
+
+  app.post('/api/admin/uploads/init', isAdmin, async (req, res) => {
+    try {
+      const { fileName, fileSize, contentType } = req.body || {};
+      if (!fileName || !fileSize) {
+        return res.status(400).json({ message: "fileName and fileSize are required" });
+      }
+      const size = parseInt(String(fileSize), 10);
+      if (!Number.isFinite(size) || size <= 0) {
+        return res.status(400).json({ message: "fileSize must be a positive number" });
+      }
+      const guid = randomUUID();
+      const result = await initUpload(guid, fileName, size, contentType);
+      res.json(result);
+    } catch (error) {
+      console.error("upload init failed:", error);
+      res.status(500).json({ message: "Failed to start upload" });
+    }
+  });
+
+  app.post('/api/admin/uploads/part-url', isAdmin, async (req, res) => {
+    try {
+      const { key, uploadId, partNumber } = req.body || {};
+      if (!key || !uploadId || !partNumber) {
+        return res.status(400).json({ message: "key, uploadId and partNumber are required" });
+      }
+      if (!String(key).startsWith(INCOMING_PREFIX)) {
+        return res.status(400).json({ message: "Invalid key" });
+      }
+      const url = await partUrl(String(key), String(uploadId), parseInt(String(partNumber), 10));
+      res.json({ url });
+    } catch (error) {
+      console.error("part url failed:", error);
+      res.status(500).json({ message: "Failed to sign part" });
+    }
+  });
+
+  app.post('/api/admin/uploads/complete', isAdmin, async (req, res) => {
+    try {
+      const { guid, key, uploadId, parts, title, artist, description, category, contentType, genre, isPublic, fileSize, originalFileName } = req.body || {};
+      if (!guid || !key || !uploadId || !Array.isArray(parts) || parts.length === 0) {
+        return res.status(400).json({ message: "guid, key, uploadId and parts are required" });
+      }
+      if (!String(key).startsWith(INCOMING_PREFIX)) {
+        return res.status(400).json({ message: "Invalid key" });
+      }
+
+      await completeUpload(String(key), String(uploadId), parts);
+
+      const cleanTitle = (title && String(title).trim()) || String(originalFileName || "Neues Video").replace(/\.[^.]+$/, "");
+
+      await writeJobFile(String(guid), {
+        guid,
+        sourceKey: key,
+        title: cleanTitle,
+        originalFileName: originalFileName || null,
+        createdAt: new Date().toISOString(),
+      });
+
+      // Row appears in admin immediately with status "processing"
+      await db.insert(videos).values({
+        id: String(guid),
+        title: cleanTitle,
+        artist: artist || null,
+        description: description || "",
+        thumbnailUrl: `${process.env.S3_PUBLIC_BASE || "https://folxvideos.b-cdn.net"}/${FOLX_PREFIX}${guid}/thumbnail.jpg`,
+        videoUrl: publicUrlFor(String(guid)),
+        duration: 0,
+        category: category || "",
+        contentType: contentType || 'video',
+        genre: genre || 'other',
+        isPublic: isPublic === undefined ? false : !!isPublic,
+        uploadStatus: 'processing',
+        originalFileName: originalFileName || null,
+        fileSize: fileSize ? parseInt(String(fileSize), 10) : null,
+      } as any);
+
+      res.json({ ok: true, guid, status: 'processing', videoUrl: publicUrlFor(String(guid)) });
+    } catch (error) {
+      console.error("upload complete failed:", error);
+      res.status(500).json({ message: "Failed to finalise upload" });
+    }
+  });
+
+  app.post('/api/admin/uploads/abort', isAdmin, async (req, res) => {
+    try {
+      const { key, uploadId } = req.body || {};
+      if (key && uploadId) await abortUpload(String(key), String(uploadId));
+      res.json({ ok: true });
+    } catch {
+      res.json({ ok: true });
+    }
+  });
+
+  // Called by the transcode worker on the remotion box when a package is ready.
+  app.post('/api/internal/transcode-done', async (req, res) => {
+    const expected = process.env.INTERNAL_TOKEN;
+    const given = (req.headers.authorization || "").replace(/^Bearer\s+/i, "");
+    if (!expected || given !== expected) {
+      return res.status(401).json({ message: "Unauthorized" });
+    }
+    try {
+      const { guid, status, duration, resolution, bitrate, format, encoding, errorMessage } = req.body || {};
+      if (!guid) return res.status(400).json({ message: "guid required" });
+
+      if (status === 'failed') {
+        await db.update(videos)
+          .set({ uploadStatus: 'failed', description: errorMessage ? String(errorMessage).slice(0, 500) : undefined, updatedAt: new Date() } as any)
+          .where(eq(videos.id, String(guid)));
+        console.error(`transcode failed for ${guid}: ${errorMessage}`);
+        return res.json({ ok: true });
+      }
+
+      await db.update(videos)
+        .set({
+          uploadStatus: 'completed',
+          duration: duration ? Math.round(Number(duration)) : 0,
+          resolution: resolution || null,
+          bitrate: bitrate ? Math.round(Number(bitrate)) : null,
+          format: format || 'hls',
+          encoding: encoding || 'h264',
+          updatedAt: new Date(),
+        } as any)
+        .where(eq(videos.id, String(guid)));
+
+      console.log(`✅ transcode done: ${guid}`);
+      res.json({ ok: true });
+    } catch (error) {
+      console.error("transcode-done failed:", error);
+      res.status(500).json({ message: "Failed" });
+    }
+  });
+
+  // Inventory: what is in the bucket vs. what is in the database
+  app.get('/api/admin/inventory', isAdmin, async (_req, res) => {
+    try {
+      const [packaged, pending, rows] = await Promise.all([
+        listPackagedGuids(),
+        listPendingJobs(),
+        db.select({ id: videos.id, title: videos.title, uploadStatus: videos.uploadStatus, createdAt: videos.createdAt }).from(videos),
+      ]);
+
+      const dbIds = new Set(rows.map(r => r.id));
+      const missingInBucket = rows
+        .filter(r => !packaged.has(r.id))
+        .map(r => ({ id: r.id, title: r.title, uploadStatus: r.uploadStatus }));
+      const orphansInBucket = Array.from(packaged).filter(g => !dbIds.has(g));
+
+      res.json({
+        databaseCount: rows.length,
+        bucketCount: packaged.size,
+        playable: rows.length - missingInBucket.length,
+        processingCount: pending.length,
+        missingInBucket,
+        orphansInBucket,
+        pendingJobs: pending,
+      });
+    } catch (error) {
+      console.error("inventory failed:", error);
+      res.status(500).json({ message: "Failed to build inventory" });
+    }
+  });
+
  const httpServer = createServer(app);
  return httpServer;
 }
@@ -0,0 +1,195 @@
+import {
+  S3Client,
+  CreateMultipartUploadCommand,
+  UploadPartCommand,
+  CompleteMultipartUploadCommand,
+  AbortMultipartUploadCommand,
+  ListObjectsV2Command,
+  HeadObjectCommand,
+  DeleteObjectCommand,
+  PutObjectCommand,
+} from "@aws-sdk/client-s3";
+import { getSignedUrl } from "@aws-sdk/s3-request-presigner";
+
+const S3_ENDPOINT = process.env.S3_ENDPOINT || "https://fsn1.your-objectstorage.com";
+const S3_BUCKET = process.env.S3_BUCKET || "folxvideos-pub";
+const S3_PUBLIC_BASE = process.env.S3_PUBLIC_BASE || "https://folxvideos.b-cdn.net";
+
+// Prefix where finished HLS packages live: folx-tv/{guid}/master.m3u8
+export const FOLX_PREFIX = "folx-tv/";
+// Staging prefix the transcode worker polls: folx-tv/_incoming/{guid}/original.<ext>
+export const INCOMING_PREFIX = "folx-tv/_incoming/";
+
+// 64 MB parts -> 10 000 parts max = 640 GB ceiling, plenty for our masters.
+export const PART_SIZE = 64 * 1024 * 1024;
+
+const s3 = new S3Client({
+  region: process.env.S3_REGION || "fsn1",
+  endpoint: S3_ENDPOINT,
+  forcePathStyle: true,
+  credentials: {
+    accessKeyId: process.env.S3_ACCESS_KEY_ID || "",
+    secretAccessKey: process.env.S3_SECRET_ACCESS_KEY || "",
+  },
+});
+
+export interface InitResult {
+  guid: string;
+  key: string;
+  uploadId: string;
+  partSize: number;
+  partCount: number;
+}
+
+function safeExt(fileName: string): string {
+  const m = /\.([a-zA-Z0-9]{1,5})$/.exec(fileName || "");
+  const ext = (m ? m[1] : "mp4").toLowerCase();
+  const allowed = ["mp4", "mov", "mkv", "m4v", "avi", "mxf", "mpg", "mpeg", "ts", "webm"];
+  return allowed.includes(ext) ? ext : "mp4";
+}
+
+export function publicUrlFor(guid: string): string {
+  return `${S3_PUBLIC_BASE}/${FOLX_PREFIX}${guid}/master.m3u8`;
+}
+
+/** Start a multipart upload for the original master file. */
+export async function initUpload(
+  guid: string,
+  fileName: string,
+  fileSize: number,
+  contentType?: string,
+): Promise<InitResult> {
+  const ext = safeExt(fileName);
+  const key = `${INCOMING_PREFIX}${guid}/original.${ext}`;
+  const cmd = new CreateMultipartUploadCommand({
+    Bucket: S3_BUCKET,
+    Key: key,
+    ContentType: contentType || "video/mp4",
+    ACL: "public-read",
+  });
+  const res = await s3.send(cmd);
+  if (!res.UploadId) throw new Error("S3 did not return an UploadId");
+  return {
+    guid,
+    key,
+    uploadId: res.UploadId,
+    partSize: PART_SIZE,
+    partCount: Math.max(1, Math.ceil(fileSize / PART_SIZE)),
+  };
+}
+
+/** Presigned PUT URL for one part. Valid 6 h so slow lines can finish. */
+export async function partUrl(key: string, uploadId: string, partNumber: number): Promise<string> {
+  const cmd = new UploadPartCommand({
+    Bucket: S3_BUCKET,
+    Key: key,
+    UploadId: uploadId,
+    PartNumber: partNumber,
+  });
+  return getSignedUrl(s3, cmd, { expiresIn: 6 * 3600 });
+}
+
+export async function completeUpload(
+  key: string,
+  uploadId: string,
+  parts: Array<{ PartNumber: number; ETag: string }>,
+): Promise<void> {
+  const ordered = [...parts].sort((a, b) => a.PartNumber - b.PartNumber);
+  await s3.send(
+    new CompleteMultipartUploadCommand({
+      Bucket: S3_BUCKET,
+      Key: key,
+      UploadId: uploadId,
+      MultipartUpload: { Parts: ordered },
+    }),
+  );
+}
+
+export async function abortUpload(key: string, uploadId: string): Promise<void> {
+  try {
+    await s3.send(new AbortMultipartUploadCommand({ Bucket: S3_BUCKET, Key: key, UploadId: uploadId }));
+  } catch {
+    /* best effort */
+  }
+}
+
+/** Drop a small JSON sidecar next to the original so the worker knows the metadata. */
+export async function writeJobFile(guid: string, job: Record<string, any>): Promise<void> {
+  await s3.send(
+    new PutObjectCommand({
+      Bucket: S3_BUCKET,
+      Key: `${INCOMING_PREFIX}${guid}/job.json`,
+      Body: JSON.stringify(job, null, 2),
+      ContentType: "application/json",
+      ACL: "public-read",
+    }),
+  );
+}
+
+export async function objectExists(key: string): Promise<boolean> {
+  try {
+    await s3.send(new HeadObjectCommand({ Bucket: S3_BUCKET, Key: key }));
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+export async function deleteObject(key: string): Promise<void> {
+  try {
+    await s3.send(new DeleteObjectCommand({ Bucket: S3_BUCKET, Key: key }));
+  } catch {
+    /* best effort */
+  }
+}
+
+/**
+ * Every GUID that has a finished master.m3u8 in the bucket.
+ * Used by the admin inventory page to diff bucket against database.
+ */
+export async function listPackagedGuids(): Promise<Set<string>> {
+  const found = new Set<string>();
+  let token: string | undefined = undefined;
+  do {
+    const res: any = await s3.send(
+      new ListObjectsV2Command({
+        Bucket: S3_BUCKET,
+        Prefix: FOLX_PREFIX,
+        MaxKeys: 1000,
+        ContinuationToken: token,
+      }),
+    );
+    for (const obj of res.Contents || []) {
+      const k: string = obj.Key || "";
+      if (k.endsWith("/master.m3u8") && !k.startsWith(INCOMING_PREFIX)) {
+        const guid = k.slice(FOLX_PREFIX.length, -"/master.m3u8".length);
+        if (guid && !guid.includes("/")) found.add(guid);
+      }
+    }
+    token = res.IsTruncated ? res.NextContinuationToken : undefined;
+  } while (token);
+  return found;
+}
+
+/** Jobs still sitting in the staging prefix (uploaded but not yet packaged). */
+export async function listPendingJobs(): Promise<string[]> {
+  const guids = new Set<string>();
+  let token: string | undefined = undefined;
+  do {
+    const res: any = await s3.send(
+      new ListObjectsV2Command({
+        Bucket: S3_BUCKET,
+        Prefix: INCOMING_PREFIX,
+        MaxKeys: 1000,
+        ContinuationToken: token,
+      }),
+    );
+    for (const obj of res.Contents || []) {
+      const rest = (obj.Key || "").slice(INCOMING_PREFIX.length);
+      const guid = rest.split("/")[0];
+      if (guid) guids.add(guid);
+    }
+    token = res.IsTruncated ? res.NextContinuationToken : undefined;
+  } while (token);
+  return Array.from(guids);
+}