Initial OSS scaffold from Songs2VID (pre-strip)

2026-08-03 06:15:05 +02:00
commit 7d7043b150
195 changed files with 25023 additions and 0 deletions
@@ -0,0 +1,171 @@
+import { Worker } from "bullmq";
+import path from "path";
+import { JobItemStatus, JobStatus } from "@prisma/client";
+import { QUEUE_NAME } from "../lib/constants";
+import { prisma } from "../lib/db";
+import { cleanupFiles, encodeVideo, getFfmpegPath } from "../lib/ffmpeg/encode";
+import { getRedisConnection } from "../lib/queue/client";
+import { releaseReservation } from "../lib/quota";
+import { getJobDir } from "../lib/storage";
+import type { VideoJobData } from "../lib/types";
+import { formatYouTubeErrorForUser } from "../lib/youtube/errors";
+import { uploadToYouTube } from "../lib/youtube/upload";
+
+async function updateJobStatus(jobId: string) {
+  const items = await prisma.jobItem.findMany({ where: { jobId } });
+  const completed = items.filter((i) => i.status === JobItemStatus.COMPLETED).length;
+  const failed = items.filter((i) => i.status === JobItemStatus.FAILED).length;
+  const total = items.length;
+
+  let status: JobStatus = JobStatus.PROCESSING;
+  if (completed + failed === total) {
+    if (completed === total) status = JobStatus.COMPLETED;
+    else if (failed === total) status = JobStatus.FAILED;
+    else status = JobStatus.PARTIAL;
+  }
+
+  await prisma.job.update({
+    where: { id: jobId },
+    data: {
+      status,
+      completedAt: completed + failed === total ? new Date() : null,
+    },
+  });
+}
+
+async function processJobItem(data: VideoJobData) {
+  const item = await prisma.jobItem.findUniqueOrThrow({
+    where: { id: data.jobItemId },
+    include: { job: true },
+  });
+
+  if (item.job.userId !== data.userId || item.jobId !== data.jobId) {
+    const message = "Job ownership mismatch";
+    console.error(`[security] ${message} for item ${item.id}`);
+    await prisma.jobItem.update({
+      where: { id: item.id },
+      data: { status: JobItemStatus.FAILED, error: message },
+    });
+    await releaseReservation(item.job.userId, item.billingSource, 1).catch(() => {});
+    await updateJobStatus(item.jobId);
+    throw new Error(message);
+  }
+
+  const jobDir = getJobDir(data.userId, data.jobId);
+  const outputPath = path.join(jobDir, `${item.id}.mp4`);
+
+  try {
+    await prisma.jobItem.update({
+      where: { id: item.id },
+      data: { status: JobItemStatus.ENCODING },
+    });
+    await prisma.job.update({
+      where: { id: data.jobId },
+      data: { status: JobStatus.PROCESSING },
+    });
+
+    await encodeVideo({
+      imagePath: item.itemImagePath || item.job.imagePath,
+      audioPath: item.audioPath,
+      outputPath,
+      resolution: item.resolution,
+      includeWatermark: item.includeWatermark,
+      songTitle: item.songTitle || item.title,
+      artist: item.artist,
+      layout: item.layoutTemplate
+        ? {
+            template: item.layoutTemplate as
+              | "COVER_LEFT_TEXT_RIGHT"
+              | "COVER_TOP_TEXT_BOTTOM"
+              | "COVER_RIGHT_TEXT_LEFT"
+              | "CENTERED_COMPACT",
+            blurAmount: item.blurAmount ?? 55,
+            blurOpacity: item.blurOpacity ?? 100,
+            textPadding: item.textPadding ?? 48,
+            titleArtistGap: item.titleArtistGap ?? 10,
+            textOffsetX: item.textOffsetX ?? 0,
+            textOffsetY: item.textOffsetY ?? 0,
+          }
+        : null,
+      watermark: {
+        mode: (item.watermarkMode as "none" | "default" | "text" | "logo") || "default",
+        text: item.watermarkText,
+        logoPath: item.watermarkLogoPath,
+        fontKey: (item.watermarkFontKey as
+          | "system"
+          | "custom"
+          | "inter"
+          | "montserrat"
+          | "roboto"
+          | "oswald"
+          | "playfair") || "system",
+        fontPath: item.watermarkFontPath,
+        position:
+          (item.watermarkPosition as
+            | "top-left"
+            | "top-right"
+            | "bottom-left"
+            | "bottom-right"
+            | "center") || "bottom-right",
+        offsetX: item.watermarkOffsetX ?? 20,
+        offsetY: item.watermarkOffsetY ?? 20,
+      },
+    });
+
+    await prisma.jobItem.update({
+      where: { id: item.id },
+      data: { status: JobItemStatus.UPLOADING, outputPath },
+    });
+
+    const youtubeVideoId = await uploadToYouTube(data.userId, outputPath, item);
+
+    await prisma.jobItem.update({
+      where: { id: item.id },
+      data: {
+        status: JobItemStatus.COMPLETED,
+        youtubeVideoId,
+        error: null,
+      },
+    });
+
+    // Lifetime successful renders only (not monthly quota — that was reserved at create).
+    await prisma.user.update({
+      where: { id: data.userId },
+      data: { createdVideoCount: { increment: 1 } },
+    });
+
+    await cleanupFiles([outputPath]);
+  } catch (err) {
+    const message = formatYouTubeErrorForUser(err);
+    console.error(`[youtube] Job item ${item.id} ("${item.title}") failed: ${message}`);
+    await prisma.jobItem.update({
+      where: { id: item.id },
+      data: { status: JobItemStatus.FAILED, error: message },
+    });
+    await releaseReservation(data.userId, item.billingSource, 1).catch(() => {});
+    throw new Error(message);
+  } finally {
+    await updateJobStatus(data.jobId);
+  }
+}
+
+const worker = new Worker<VideoJobData>(
+  QUEUE_NAME,
+  async (job) => {
+    await processJobItem(job.data);
+  },
+  {
+    connection: getRedisConnection(),
+    concurrency: 2,
+  },
+);
+
+worker.on("completed", (job) => {
+  console.log(`Job item ${job.data.jobItemId} completed`);
+});
+
+worker.on("failed", (job, err) => {
+  console.error(`Job item ${job?.data.jobItemId} failed:`, err.message);
+});
+
+console.log(`Songs2VID worker started (ffmpeg: ${getFfmpegPath()}), waiting for jobs...`);