Sharp-based image processing worker with pipeline support

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 = `
+        <svg width="1000" height="200">
+          <style>
+            text {
+              font-family: Arial, Helvetica, sans-serif;
+              font-size: ${fontSize}px;
+              font-weight: bold;
+              fill: ${color};
+              opacity: ${opacity};
+            }
+          </style>
+          <text x="50%" y="50%" text-anchor="middle" dominant-baseline="middle">${text}</text>
+        </svg>
+      `;
+
+      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 };

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();