import { Job, Queue, QueueEvents, Worker } from "bullmq"; import dotenv from "dotenv"; import { fileTypeFromFile } from "file-type"; import { FileTypeResult } from "file-type/core"; import fs from "fs-extra"; import gm from "gm"; import path from "path"; import { fileURLToPath } from "url"; import { logger } from "../server.js"; import { generateUniqueHeicFilename } from "./generateUniqueFilename.js"; import { FolderPaths } from "./serverInit.js"; dotenv.config({ path: path.resolve(process.cwd(), `.env.${process.env.NODE_ENV || "development"}`) }); const QUEUE_NAME = "heicQueue"; const connectionOpts = { host: "localhost", port: 6379, enableReadyCheck: true, reconnectOnError: (err: Error) => err.message.includes("READONLY") }; const heicQueue = new Queue(QUEUE_NAME, { connection: connectionOpts, defaultJobOptions: { removeOnComplete: true, removeOnFail: true, attempts: 3, backoff: { type: "exponential", delay: 1000 } } }); // Re-added QueueEvents for waiting on job completion const heicQueueEvents = new QueueEvents(QUEUE_NAME, { connection: connectionOpts }); const CLEANUP_INTERVAL_MS = 10 * 60 * 1000; // 10 minutes setInterval(cleanupQueue, CLEANUP_INTERVAL_MS); async function cleanupQueue() { try { const ONE_HOUR = 60 * 60 * 1000; const SIX_HOURS = 6 * ONE_HOUR; await heicQueue.clean(ONE_HOUR, 500, "completed"); await heicQueue.clean(SIX_HOURS, 500, "failed"); const counts = await heicQueue.getJobCounts(); logger.debug(`HEIC Queue status: ${JSON.stringify(counts)}`); } catch (error) { logger.error("HEIC Queue cleanup error:", error); } } /** * Filter files to include only valid HEIC images. */ async function filterHeicFiles(files: Express.Multer.File[]) { const valid: Express.Multer.File[] = []; for (const file of files) { const type: FileTypeResult | undefined = await fileTypeFromFile(file.path); if (type?.mime === "image/heic") valid.push(file); } return valid; } /** * Handle original file based on environment variable. */ async function handleOriginalFile(fileInfo: { path: string; destination: string; originalFilename: string }) { try { if (process.env.KEEP_CONVERTED_ORIGINALS) { const destDir = path.join(fileInfo.destination, FolderPaths.ConvertedOriginalSubDir); await fs.ensureDir(destDir); await fs.move(fileInfo.path, path.join(destDir, fileInfo.originalFilename), { overwrite: true }); } else { await fs.unlink(fileInfo.path); } } catch (error) { logger.error("Error handling original file:", error); throw error; } } /** * Convert HEIC to JPEG using GraphicsMagick stream. */ async function convertToJpeg(inputPath: string, outputPath: string): Promise { return new Promise((resolve, reject) => { const readStream = fs.createReadStream(inputPath); const writeStream = fs.createWriteStream(outputPath); gm(readStream) .setFormat("jpg") .stream() .pipe(writeStream) .on("finish", () => resolve(outputPath)) .on("error", reject); }); } /** * Add HEIC files to the conversion queue and wait for job completion. */ export async function convertHeicFiles(files: Express.Multer.File[]) { const heicFiles = await filterHeicFiles(files); if (heicFiles.length === 0) { logger.debug("No HEIC files found to convert."); return; } const jobs = heicFiles.map((file) => ({ name: file.filename, data: { convertedFileName: generateUniqueHeicFilename(file), fileInfo: { path: file.path, destination: file.destination, originalFilename: file.filename } } })); // Add jobs and wait for completion of each before proceeding for (const jobData of jobs) { try { const job = await heicQueue.add(jobData.name, jobData.data); await job.waitUntilFinished(heicQueueEvents); logger.debug(`Job ${job.id} finished successfully.`); } catch (error) { logger.error(`Job for ${jobData.data.fileInfo.originalFilename} failed:`, error); // Depending on your error handling strategy you might rethrow or continue } } // Update original files list with new names, mimetype, and path const filenameToIndex = new Map(files.map((f, i) => [f.filename, i])); for (const { data } of jobs) { const idx = filenameToIndex.get(data.fileInfo.originalFilename); if (idx !== undefined) { const oldPath = files[idx].path; files[idx].filename = data.convertedFileName; files[idx].mimetype = "image/jpeg"; files[idx].path = path.join(data.fileInfo.destination, data.convertedFileName); logger.debug(`Updated file entry: ${data.fileInfo.originalFilename} -> ${data.convertedFileName}`, { oldPath, newPath: files[idx].path, newMimetype: files[idx].mimetype }); } } } // Worker processing HEIC conversion jobs const heicWorker = new Worker( QUEUE_NAME, async ( job: Job<{ fileInfo: { path: string; destination: string; originalFilename: string }; convertedFileName: string; }> ) => { const { fileInfo, convertedFileName } = job.data; try { logger.debug(`Converting ${fileInfo.originalFilename} from HEIC to JPEG.`); await job.updateProgress(10); const outputPath = path.join(fileInfo.destination, convertedFileName); await convertToJpeg(fileInfo.path, outputPath); await job.updateProgress(50); await handleOriginalFile(fileInfo); logger.debug(`Successfully converted ${fileInfo.originalFilename} to JPEG.`); await job.updateProgress(100); return true; } catch (error) { logger.error(`Error converting ${fileInfo.originalFilename}:`, error); throw error; } }, { connection: connectionOpts, concurrency: 1 } ); // Event listeners for queue and worker heicQueue.on("waiting", (job) => { logger.debug(`[heicQueue] Job waiting in queue: ${job.data.convertedFileName}`); }); heicQueue.on("error", (error) => { logger.error(`[heicQueue] Queue error:`, error); }); heicWorker.on("ready", () => { logger.debug("[heicWorker] Worker ready"); }); heicWorker.on("active", (job, prev) => { logger.debug(`[heicWorker] Job ${job.id} active (previous: ${prev})`); }); heicWorker.on("completed", async (job) => { logger.debug(`[heicWorker] Job ${job.id} completed`); await job.remove(); logger.debug(`Job ${job.id} removed from Redis`); }); heicWorker.on("failed", (jobId, reason) => { logger.error(`[heicWorker] Job ${jobId} failed: ${reason}`); }); heicWorker.on("error", (error) => { logger.error(`[heicWorker] Worker error:`, error); }); heicWorker.on("stalled", (job) => { logger.error(`[heicWorker] Worker stalled: ${job}`); }); heicWorker.on("ioredis:close", () => { logger.error("[heicWorker] Redis connection closed"); });