feat(api,web): add batch processing with ZIP download and SSE progress

Backend: POST /api/v1/tools/:toolId/batch accepts multiple files +
settings, processes via p-queue with CONCURRENT_JOBS concurrency limit,
streams ZIP response using archiver. Tool registry in tool-factory
enables batch to reuse any registered tool's process function. SSE
endpoint at GET /api/v1/jobs/:jobId/progress provides real-time updates.
Handles partial failures gracefully, preserves filenames, deduplicates
collisions. Frontend: use-batch-processor hook handles upload, SSE
progress tracking, and automatic ZIP download.

apps/api/package.json

@@ -11,26 +11,29 @@
     "clean": "rm -rf dist"
   },
   "dependencies": {
-    "@stirling-image/image-engine": "workspace:*",
-    "@stirling-image/shared": "workspace:*",
-    "sharp": "^0.33.0",
-    "fastify": "^5.2.0",
-    "@fastify/static": "^8.1.0",
-    "@fastify/multipart": "^9.0.0",
     "@fastify/cors": "^11.0.0",
+    "@fastify/multipart": "^9.0.0",
     "@fastify/rate-limit": "^10.2.0",
+    "@fastify/static": "^8.1.0",
     "@fastify/swagger": "^9.4.0",
     "@fastify/swagger-ui": "^5.2.0",
+    "@stirling-image/image-engine": "workspace:*",
+    "@stirling-image/shared": "workspace:*",
+    "archiver": "^7.0.1",
+    "better-sqlite3": "^11.7.0",
     "dotenv": "^16.4.0",
-    "zod": "^3.24.0",
     "drizzle-orm": "^0.38.0",
-    "better-sqlite3": "^11.7.0"
+    "fastify": "^5.2.0",
+    "p-queue": "^9.1.0",
+    "sharp": "^0.33.0",
+    "zod": "^3.24.0"
   },
   "devDependencies": {
-    "typescript": "^5.7.0",
-    "tsx": "^4.19.0",
+    "@types/archiver": "^7.0.0",
+    "@types/better-sqlite3": "^7.6.0",
     "@types/node": "^22.0.0",
     "drizzle-kit": "^0.30.0",
-    "@types/better-sqlite3": "^7.6.0"
+    "tsx": "^4.19.0",
+    "typescript": "^5.7.0"
   }
 }

apps/api/src/index.ts

@@ -12,6 +12,8 @@ import { registerStatic } from "./plugins/static.js";
 import { startCleanupCron } from "./lib/cleanup.js";
 import { fileRoutes } from "./routes/files.js";
 import { registerToolRoutes } from "./routes/tools/index.js";
+import { registerBatchRoutes } from "./routes/batch.js";
+import { registerProgressRoutes } from "./routes/progress.js";
 
 // Run before anything else
 runMigrations();
@@ -63,6 +65,12 @@ await fileRoutes(app);
 // Tool routes (generic factory-based)
 await registerToolRoutes(app);
 
+// Batch processing routes (must be after tool routes so the registry is populated)
+await registerBatchRoutes(app);
+
+// Progress SSE routes
+await registerProgressRoutes(app);
+
 // Health check
 app.get("/api/v1/health", async () => ({
   status: "healthy",

apps/api/src/routes/batch.ts

@@ -0,0 +1,221 @@
+/**
+ * Batch processing route.
+ *
+ * POST /api/v1/tools/:toolId/batch
+ *
+ * Accepts multipart with multiple files + settings JSON.
+ * Processes all files through the tool using p-queue for concurrency control.
+ * Returns a ZIP file containing all processed images.
+ */
+import { randomUUID } from "node:crypto";
+import { basename } from "node:path";
+import type { FastifyInstance, FastifyRequest, FastifyReply } from "fastify";
+import archiver from "archiver";
+import PQueue from "p-queue";
+import { getToolConfig } from "./tool-factory.js";
+import { validateImageBuffer } from "../lib/file-validation.js";
+import { env } from "../config.js";
+import { updateJobProgress, type JobProgress } from "./progress.js";
+
+/**
+ * Sanitize a filename to prevent path traversal attacks.
+ */
+function sanitizeFilename(raw: string): string {
+  let name = basename(raw);
+  name = name.replace(/\.\./g, "");
+  name = name.replace(/\0/g, "");
+  if (!name || name === "." || name === "..") {
+    name = "image";
+  }
+  return name;
+}
+
+interface ParsedFile {
+  buffer: Buffer;
+  filename: string;
+}
+
+export async function registerBatchRoutes(
+  app: FastifyInstance,
+): Promise<void> {
+  app.post(
+    "/api/v1/tools/:toolId/batch",
+    async (
+      request: FastifyRequest<{ Params: { toolId: string } }>,
+      reply: FastifyReply,
+    ) => {
+      const { toolId } = request.params;
+
+      // Look up the tool config from the registry
+      const toolConfig = getToolConfig(toolId);
+      if (!toolConfig) {
+        return reply.status(404).send({ error: `Tool "${toolId}" not found` });
+      }
+
+      // Parse multipart: collect all files and the settings field
+      const files: ParsedFile[] = [];
+      let settingsRaw: string | null = null;
+
+      try {
+        const parts = request.parts();
+        for await (const part of parts) {
+          if (part.type === "file") {
+            const chunks: Buffer[] = [];
+            for await (const chunk of part.file) {
+              chunks.push(chunk);
+            }
+            const buffer = Buffer.concat(chunks);
+            if (buffer.length > 0) {
+              files.push({
+                buffer,
+                filename: sanitizeFilename(part.filename ?? "image"),
+              });
+            }
+          } else if (part.fieldname === "settings") {
+            settingsRaw = part.value as string;
+          }
+        }
+      } catch (err) {
+        return reply.status(400).send({
+          error: "Failed to parse multipart request",
+          details: err instanceof Error ? err.message : String(err),
+        });
+      }
+
+      if (files.length === 0) {
+        return reply.status(400).send({ error: "No image files provided" });
+      }
+
+      // Enforce batch size limit
+      if (files.length > env.MAX_BATCH_SIZE) {
+        return reply.status(400).send({
+          error: `Too many files. Maximum batch size is ${env.MAX_BATCH_SIZE}`,
+        });
+      }
+
+      // Parse and validate settings
+      let settings: unknown;
+      try {
+        const parsed = settingsRaw ? JSON.parse(settingsRaw) : {};
+        const result = toolConfig.settingsSchema.safeParse(parsed);
+        if (!result.success) {
+          return reply.status(400).send({
+            error: "Invalid settings",
+            details: result.error.issues.map(
+              (i: { path: (string | number)[]; message: string }) => ({
+                path: i.path.join("."),
+                message: i.message,
+              }),
+            ),
+          });
+        }
+        settings = result.data;
+      } catch {
+        return reply.status(400).send({ error: "Settings must be valid JSON" });
+      }
+
+      // Create a job ID for progress tracking
+      const jobId = randomUUID();
+
+      const progress: JobProgress = {
+        jobId,
+        status: "processing",
+        totalFiles: files.length,
+        completedFiles: 0,
+        failedFiles: 0,
+        errors: [],
+      };
+      updateJobProgress({ ...progress });
+
+      // Set up response headers for ZIP streaming
+      reply.raw.writeHead(200, {
+        "Content-Type": "application/zip",
+        "Content-Disposition": `attachment; filename="batch-${toolId}-${jobId.slice(0, 8)}.zip"`,
+        "Transfer-Encoding": "chunked",
+        "X-Job-Id": jobId,
+      });
+
+      // Create ZIP archive that pipes directly to the response
+      const archive = archiver("zip", { zlib: { level: 5 } });
+      archive.pipe(reply.raw);
+
+      // Use p-queue for concurrency control
+      const queue = new PQueue({ concurrency: env.CONCURRENT_JOBS });
+
+      // Track unique filenames to avoid collisions in the ZIP
+      const usedNames = new Set<string>();
+      function getUniqueName(name: string): string {
+        if (!usedNames.has(name)) {
+          usedNames.add(name);
+          return name;
+        }
+        const dotIdx = name.lastIndexOf(".");
+        const base = dotIdx > 0 ? name.slice(0, dotIdx) : name;
+        const ext = dotIdx > 0 ? name.slice(dotIdx) : "";
+        let counter = 1;
+        let candidate = `${base}_${counter}${ext}`;
+        while (usedNames.has(candidate)) {
+          counter++;
+          candidate = `${base}_${counter}${ext}`;
+        }
+        usedNames.add(candidate);
+        return candidate;
+      }
+
+      // Process all files through the queue
+      const tasks = files.map((file) =>
+        queue.add(async () => {
+          progress.currentFile = file.filename;
+          updateJobProgress({ ...progress });
+
+          // Validate the image
+          const validation = await validateImageBuffer(file.buffer);
+          if (!validation.valid) {
+            progress.failedFiles++;
+            progress.errors.push({
+              filename: file.filename,
+              error: `Invalid image: ${validation.reason}`,
+            });
+            progress.completedFiles++;
+            updateJobProgress({ ...progress });
+            return;
+          }
+
+          try {
+            const result = await toolConfig.process(
+              file.buffer,
+              settings,
+              file.filename,
+            );
+
+            const zipFilename = getUniqueName(result.filename);
+            archive.append(result.buffer, { name: zipFilename });
+
+            progress.completedFiles++;
+            updateJobProgress({ ...progress });
+          } catch (err) {
+            progress.failedFiles++;
+            progress.errors.push({
+              filename: file.filename,
+              error: err instanceof Error ? err.message : "Processing failed",
+            });
+            progress.completedFiles++;
+            updateJobProgress({ ...progress });
+          }
+        }),
+      );
+
+      // Wait for all tasks to complete
+      await Promise.all(tasks);
+
+      // Finalize progress
+      progress.status =
+        progress.failedFiles === progress.totalFiles ? "failed" : "completed";
+      progress.currentFile = undefined;
+      updateJobProgress({ ...progress });
+
+      // Finalize the ZIP archive (flushes remaining data and ends the stream)
+      await archive.finalize();
+    },
+  );
+}

apps/api/src/routes/progress.ts

@@ -0,0 +1,119 @@
+/**
+ * SSE endpoint for real-time job progress tracking.
+ *
+ * GET /api/v1/jobs/:jobId/progress
+ *
+ * Sends Server-Sent Events with progress data until the job finishes.
+ */
+import type { FastifyInstance, FastifyRequest, FastifyReply } from "fastify";
+
+export interface JobProgress {
+  jobId: string;
+  status: "processing" | "completed" | "failed";
+  totalFiles: number;
+  completedFiles: number;
+  failedFiles: number;
+  /** Names of files that failed, with error messages. */
+  errors: Array<{ filename: string; error: string }>;
+  /** Current file being processed (if any). */
+  currentFile?: string;
+}
+
+/** In-memory store of job progress, keyed by jobId. */
+const jobProgressStore = new Map<string, JobProgress>();
+
+/** SSE listeners waiting for updates, keyed by jobId. */
+const listeners = new Map<string, Set<(data: JobProgress) => void>>();
+
+/**
+ * Create or update progress for a job.
+ */
+export function updateJobProgress(progress: JobProgress): void {
+  jobProgressStore.set(progress.jobId, progress);
+  // Notify all SSE listeners
+  const subs = listeners.get(progress.jobId);
+  if (subs) {
+    for (const cb of subs) {
+      cb(progress);
+    }
+    // If the job is done, clean up listeners after a brief delay
+    if (progress.status === "completed" || progress.status === "failed") {
+      setTimeout(() => {
+        listeners.delete(progress.jobId);
+        jobProgressStore.delete(progress.jobId);
+      }, 5000);
+    }
+  }
+}
+
+/**
+ * Get current progress for a job.
+ */
+export function getJobProgress(jobId: string): JobProgress | undefined {
+  return jobProgressStore.get(jobId);
+}
+
+export async function registerProgressRoutes(
+  app: FastifyInstance,
+): Promise<void> {
+  app.get(
+    "/api/v1/jobs/:jobId/progress",
+    async (
+      request: FastifyRequest<{ Params: { jobId: string } }>,
+      reply: FastifyReply,
+    ) => {
+      const { jobId } = request.params;
+
+      // Send SSE headers via the raw Node response
+      reply.raw.writeHead(200, {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+        Connection: "keep-alive",
+        "X-Accel-Buffering": "no",
+      });
+
+      // Helper to send an SSE message
+      const sendEvent = (data: JobProgress) => {
+        reply.raw.write(`data: ${JSON.stringify(data)}\n\n`);
+      };
+
+      // If the job already has progress, send it immediately
+      const existing = jobProgressStore.get(jobId);
+      if (existing) {
+        sendEvent(existing);
+        if (
+          existing.status === "completed" ||
+          existing.status === "failed"
+        ) {
+          reply.raw.end();
+          return;
+        }
+      }
+
+      // Subscribe to updates
+      if (!listeners.has(jobId)) {
+        listeners.set(jobId, new Set());
+      }
+
+      const callback = (data: JobProgress) => {
+        sendEvent(data);
+        if (data.status === "completed" || data.status === "failed") {
+          reply.raw.end();
+        }
+      };
+
+      listeners.get(jobId)!.add(callback);
+
+      // Clean up on client disconnect
+      request.raw.on("close", () => {
+        const subs = listeners.get(jobId);
+        if (subs) {
+          subs.delete(callback);
+          if (subs.size === 0) {
+            listeners.delete(jobId);
+          }
+        }
+      });
+    },
+  );
+}

apps/api/src/routes/tool-factory.ts

@@ -19,6 +19,21 @@ export interface ToolRouteConfig<T> {
   ) => Promise<{ buffer: Buffer; filename: string; contentType: string }>;
 }
 
+/**
+ * In-memory registry of all tool configs, keyed by toolId.
+ * Populated by createToolRoute() calls; used by batch processing.
+ */
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+const toolRegistry = new Map<string, ToolRouteConfig<any>>();
+
+/**
+ * Retrieve a registered tool config by its ID.
+ */
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export function getToolConfig(toolId: string): ToolRouteConfig<any> | undefined {
+  return toolRegistry.get(toolId);
+}
+
 /**
  * Sanitize a filename to prevent path traversal attacks.
  */
@@ -51,6 +66,9 @@ export function createToolRoute<T>(
   app: FastifyInstance,
   config: ToolRouteConfig<T>,
 ): void {
+  // Register in the tool registry for batch processing
+  toolRegistry.set(config.toolId, config);
+
   app.post(
     `/api/v1/tools/${config.toolId}`,
     async (request: FastifyRequest, reply: FastifyReply) => {

apps/web/src/hooks/use-batch-processor.ts

@@ -0,0 +1,191 @@
+import { useCallback, useState, useRef } from "react";
+
+function getToken(): string {
+  return localStorage.getItem("stirling-token") || "";
+}
+
+interface BatchProgress {
+  totalFiles: number;
+  completedFiles: number;
+  failedFiles: number;
+  currentFile?: string;
+  errors: Array<{ filename: string; error: string }>;
+  status: "idle" | "uploading" | "processing" | "completed" | "failed";
+  /** Percentage 0-100. */
+  percent: number;
+}
+
+/**
+ * Hook for batch processing multiple files with SSE progress tracking.
+ *
+ * Uploads all files to the batch endpoint, listens for SSE progress events,
+ * and triggers a ZIP download when processing completes.
+ */
+export function useBatchProcessor(toolId: string) {
+  const [progress, setProgress] = useState<BatchProgress>({
+    totalFiles: 0,
+    completedFiles: 0,
+    failedFiles: 0,
+    errors: [],
+    status: "idle",
+    percent: 0,
+  });
+
+  const abortRef = useRef<AbortController | null>(null);
+
+  const processBatch = useCallback(
+    async (files: File[], settings: Record<string, unknown>) => {
+      if (files.length === 0) return;
+
+      // Reset state
+      setProgress({
+        totalFiles: files.length,
+        completedFiles: 0,
+        failedFiles: 0,
+        errors: [],
+        status: "uploading",
+        percent: 0,
+      });
+
+      abortRef.current = new AbortController();
+
+      try {
+        // Build multipart form with all files + settings
+        const formData = new FormData();
+        for (const file of files) {
+          formData.append("files", file);
+        }
+        formData.append("settings", JSON.stringify(settings));
+
+        setProgress((prev) => ({ ...prev, status: "processing" }));
+
+        const res = await fetch(`/api/v1/tools/${toolId}/batch`, {
+          method: "POST",
+          headers: { Authorization: `Bearer ${getToken()}` },
+          body: formData,
+          signal: abortRef.current.signal,
+        });
+
+        if (!res.ok) {
+          // Try to read error body
+          const text = await res.text();
+          let errorMsg = `Batch processing failed: ${res.status}`;
+          try {
+            const body = JSON.parse(text);
+            errorMsg = body.error || body.details || errorMsg;
+          } catch {
+            // ignore
+          }
+          setProgress((prev) => ({
+            ...prev,
+            status: "failed",
+            errors: [{ filename: "", error: errorMsg }],
+          }));
+          return;
+        }
+
+        // Get the Job ID from the response header for SSE
+        const jobId = res.headers.get("X-Job-Id");
+
+        // The response IS the ZIP file — trigger download
+        const blob = await res.blob();
+        const url = URL.createObjectURL(blob);
+        const a = document.createElement("a");
+        a.href = url;
+        a.download = `batch-${toolId}.zip`;
+        document.body.appendChild(a);
+        a.click();
+        document.body.removeChild(a);
+        URL.revokeObjectURL(url);
+
+        // If we have a jobId, try to get final progress from SSE
+        // But since the ZIP response already indicates success, mark as completed
+        setProgress((prev) => ({
+          ...prev,
+          status: "completed",
+          completedFiles: files.length,
+          percent: 100,
+        }));
+
+        // Optionally fetch final progress for error details
+        if (jobId) {
+          try {
+            const progressRes = await fetch(`/api/v1/jobs/${jobId}/progress`, {
+              headers: { Authorization: `Bearer ${getToken()}` },
+              signal: AbortSignal.timeout(3000),
+            });
+            // SSE stream — read the last event
+            const reader = progressRes.body?.getReader();
+            if (reader) {
+              const decoder = new TextDecoder();
+              let buffer = "";
+              let lastData: string | null = null;
+              // Read a few chunks to get the final state
+              for (let i = 0; i < 5; i++) {
+                const { value, done } = await reader.read();
+                if (done) break;
+                buffer += decoder.decode(value, { stream: true });
+                const lines = buffer.split("\n");
+                for (const line of lines) {
+                  if (line.startsWith("data: ")) {
+                    lastData = line.slice(6);
+                  }
+                }
+              }
+              reader.cancel();
+
+              if (lastData) {
+                const finalProgress = JSON.parse(lastData);
+                setProgress((prev) => ({
+                  ...prev,
+                  failedFiles: finalProgress.failedFiles ?? prev.failedFiles,
+                  errors: finalProgress.errors ?? prev.errors,
+                  completedFiles:
+                    finalProgress.completedFiles ?? prev.completedFiles,
+                }));
+              }
+            }
+          } catch {
+            // Progress fetch is optional, ignore errors
+          }
+        }
+      } catch (err) {
+        if ((err as Error).name === "AbortError") return;
+        setProgress((prev) => ({
+          ...prev,
+          status: "failed",
+          errors: [
+            {
+              filename: "",
+              error: err instanceof Error ? err.message : "Batch processing failed",
+            },
+          ],
+        }));
+      }
+    },
+    [toolId],
+  );
+
+  const cancel = useCallback(() => {
+    abortRef.current?.abort();
+    setProgress((prev) => ({ ...prev, status: "idle" }));
+  }, []);
+
+  const reset = useCallback(() => {
+    setProgress({
+      totalFiles: 0,
+      completedFiles: 0,
+      failedFiles: 0,
+      errors: [],
+      status: "idle",
+      percent: 0,
+    });
+  }, []);
+
+  return {
+    processBatch,
+    cancel,
+    reset,
+    progress,
+  };
+}