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, title: 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, }, }); // Quota was reserved at job creation; do not increment again on success. 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( 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...`);