From 79794f4fad1913de2c1dcf2aaac63a38432d9c73 Mon Sep 17 00:00:00 2001 From: Walusimbi Silver Date: Tue, 17 Mar 2026 13:02:31 +0300 Subject: [PATCH] Sharp-based image processing worker with pipeline support --- src/workers/image.worker.js | 232 ++++++++++++++++++++++++++++++++++++ src/workers/standalone.js | 13 ++ 2 files changed, 245 insertions(+) create mode 100644 src/workers/image.worker.js create mode 100644 src/workers/standalone.js diff --git a/src/workers/image.worker.js b/src/workers/image.worker.js new file mode 100644 index 0000000..44f4977 --- /dev/null +++ b/src/workers/image.worker.js @@ -0,0 +1,232 @@ +const sharp = require("sharp"); +const path = require("path"); +const fs = require("fs"); +const { nanoid } = require("nanoid"); +const imageQueue = require("../config/queue"); +const wsManager = require("../utils/ws"); +const config = require("../config"); + +// Ensure output dir exists +fs.mkdirSync(config.storage.outputDir, { recursive: true }); + +/** + * Apply a single operation to a sharp pipeline. + * Returns the modified pipeline (sharp is chainable). + */ +function applyOperation(pipeline, op, params) { + switch (op) { + case "resize": + return pipeline.resize({ + width: params.width || null, + height: params.height || null, + fit: params.fit || "cover", + withoutEnlargement: params.withoutEnlargement !== false, + }); + + case "convert": + return pipeline.toFormat(params.format, { + quality: params.quality || 80, + }); + + case "compress": { + // Auto-detect format and compress + const q = params.quality || 75; + return pipeline + .jpeg({ quality: q, mozjpeg: true }) + .png({ quality: q, compressionLevel: 9 }) + .webp({ quality: q }) + .avif({ quality: q }); + } + + case "watermark": { + const text = params.text; + const fontSize = params.fontSize || 48; + const opacity = params.opacity || 0.3; + const color = params.color || "white"; + + // SVG text overlay + const svg = ` + + + ${text} + + `; + + return pipeline.composite([ + { + input: Buffer.from(svg), + gravity: params.position || "southeast", + tile: !!params.tile, + }, + ]); + } + + case "strip-meta": + return pipeline.withMetadata(false); + + case "blur": + return pipeline.blur(params.sigma || 3); + + case "grayscale": + return pipeline.grayscale(); + + case "rotate": + return pipeline.rotate(params.angle, { + background: params.background || { r: 0, g: 0, b: 0, alpha: 0 }, + }); + + default: + throw new Error(`Unknown operation: ${op}`); + } +} + +/** + * Determine output file extension based on operations. + * If a 'convert' op exists, use that format. Otherwise keep original. + */ +function resolveOutputExt(operations, originalName) { + const convertOp = [...operations].reverse().find((o) => o.op === "convert"); + if (convertOp) { + const fmt = convertOp.params.format; + return fmt === "jpeg" ? ".jpg" : `.${fmt}`; + } + return path.extname(originalName).toLowerCase() || ".jpg"; +} + +/** + * Process a job: run all operations sequentially through a sharp pipeline. + */ +async function processJob(job) { + const { inputPath, originalName, operations } = job.data; + const totalOps = operations.length; + + const notify = (type, data) => { + wsManager.notify(String(job.id), { type, ...data }); + }; + + notify("started", { operations: operations.map((o) => o.op) }); + + try { + // Get input file info + const inputMeta = await sharp(inputPath).metadata(); + notify("info", { + input: { + width: inputMeta.width, + height: inputMeta.height, + format: inputMeta.format, + size: fs.statSync(inputPath).size, + }, + }); + + // Build pipeline + let pipeline = sharp(inputPath); + let hasConvert = false; + + for (let i = 0; i < operations.length; i++) { + const { op, params = {} } = operations[i]; + const stage = `${op} (${i + 1}/${totalOps})`; + + notify("progress", { + stage, + progress: Math.round(((i) / totalOps) * 100), + }); + + // Special handling: compress needs format context + if (op === "compress" && !hasConvert) { + // Apply compression matching input format + const q = params.quality || 75; + const fmtMap = { + jpeg: () => pipeline.jpeg({ quality: q, mozjpeg: true }), + png: () => pipeline.png({ quality: q, compressionLevel: 9 }), + webp: () => pipeline.webp({ quality: q }), + avif: () => pipeline.avif({ quality: q }), + tiff: () => pipeline.tiff({ quality: q }), + }; + const fn = fmtMap[inputMeta.format]; + if (fn) pipeline = fn(); + } else { + pipeline = applyOperation(pipeline, op, params); + if (op === "convert") hasConvert = true; + } + + // Report progress + await job.progress(Math.round(((i + 1) / totalOps) * 100)); + } + + // Write output + const ext = resolveOutputExt(operations, originalName); + const baseName = path.basename(originalName, path.extname(originalName)); + const outputFilename = `${baseName}_processed_${nanoid(6)}${ext}`; + const outputPath = path.join(config.storage.outputDir, outputFilename); + + await pipeline.toFile(outputPath); + + // Get output stats + const outputStats = fs.statSync(outputPath); + const outputMeta = await sharp(outputPath).metadata(); + + const result = { + outputPath, + outputFilename, + output: { + width: outputMeta.width, + height: outputMeta.height, + format: outputMeta.format, + size: outputStats.size, + }, + savings: { + bytes: fs.statSync(inputPath).size - outputStats.size, + percent: Math.round( + ((fs.statSync(inputPath).size - outputStats.size) / + fs.statSync(inputPath).size) * + 100 + ), + }, + processingTime: Date.now() - job.timestamp, + }; + + notify("completed", result); + + // Clean up input file + fs.unlinkSync(inputPath); + + return result; + } catch (err) { + notify("failed", { error: err.message }); + // Clean up input on failure too + if (fs.existsSync(inputPath)) fs.unlinkSync(inputPath); + throw err; + } +} + +/** + * Start the worker — called from either the standalone worker + * process or from the main API process. + */ +function startWorker() { + imageQueue.process("process", config.worker.concurrency, processJob); + + imageQueue.on("completed", (job) => { + console.log(`[Worker] Job ${job.id} completed in ${Date.now() - job.timestamp}ms`); + }); + + imageQueue.on("failed", (job, err) => { + console.error(`[Worker] Job ${job.id} failed (attempt ${job.attemptsMade}):`, err.message); + }); + + imageQueue.on("stalled", (jobId) => { + console.warn(`[Worker] Job ${jobId} stalled — will be retried`); + }); + + console.log(`[Worker] Processing with concurrency=${config.worker.concurrency}`); +} + +module.exports = { startWorker }; diff --git a/src/workers/standalone.js b/src/workers/standalone.js new file mode 100644 index 0000000..6516281 --- /dev/null +++ b/src/workers/standalone.js @@ -0,0 +1,13 @@ +/** + * Standalone worker process. + * Run separately with: pm2 start ecosystem.config.js --only imgforge-worker + * + * This connects to the same Redis queue and processes jobs independently + * of the API server. Use this when you want to scale workers separately. + */ + +require("../config"); // load env +const { startWorker } = require("./image.worker"); + +console.log("[Worker] Starting standalone worker process..."); +startWorker();