feat: add worker threads, persistent Python sidecar, graceful shutdown, and architectural improvements

- Graceful shutdown: SIGTERM/SIGINT handlers drain HTTP, stop workers, close DB
- Thumbnail caching: disk-cached thumbnails with immutable Cache-Control headers
- Worker thread pool: Piscina offloads Sharp processing off the main event loop
- Persistent Python dispatcher: pre-imports ML libraries, eliminates cold-start latency
- Tool page registry: declarative tool-to-component mapping replaces 750-line switch
- File store cleanup: remove dead derived fields, stable files array reference
- Job persistence: progress written to SQLite jobs table, stale jobs recovered on startup

apps/api/package.json

@@ -30,6 +30,7 @@
     "jsqr": "^1.4.0",
     "p-queue": "^9.1.0",
     "pdfkit": "^0.18.0",
+    "piscina": "^5.1.4",
     "potrace": "^2.1.8",
     "qrcode": "^1.5.4",
     "sharp": "^0.33.0",

apps/api/src/index.ts

@@ -6,6 +6,7 @@ import { env } from "./config.js";
 import { db, schema } from "./db/index.js";
 import { runMigrations } from "./db/migrate.js";
 import { startCleanupCron } from "./lib/cleanup.js";
+import { shutdownWorkerPool } from "./lib/worker-pool.js";
 import { authMiddleware, authRoutes, ensureDefaultAdmin, requireAdmin } from "./plugins/auth.js";
 import { registerStatic } from "./plugins/static.js";
 import { registerUpload } from "./plugins/upload.js";
@@ -15,7 +16,7 @@ import { brandingRoutes } from "./routes/branding.js";
 import { docsRoutes } from "./routes/docs.js";
 import { fileRoutes } from "./routes/files.js";
 import { registerPipelineRoutes } from "./routes/pipeline.js";
-import { registerProgressRoutes } from "./routes/progress.js";
+import { recoverStaleJobs, registerProgressRoutes } from "./routes/progress.js";
 import { settingsRoutes } from "./routes/settings.js";
 import { teamsRoutes } from "./routes/teams.js";
 import { registerToolRoutes } from "./routes/tools/index.js";
@@ -28,6 +29,9 @@ console.log("Database initialized");
 // Create default admin user if no users exist
 await ensureDefaultAdmin();
 
+// Mark any jobs left in processing/queued from a previous unclean shutdown
+recoverStaleJobs();
+
 const app = Fastify({
   logger: true,
   bodyLimit: env.MAX_UPLOAD_SIZE_MB * 1024 * 1024,
@@ -143,7 +147,7 @@ if (process.env.NODE_ENV === "production") {
 }
 
 // Start workspace cleanup cron
-startCleanupCron();
+const cleanupCron = startCleanupCron();
 
 // Start
 try {
@@ -153,3 +157,48 @@ try {
   app.log.error(err);
   process.exit(1);
 }
+
+// Graceful shutdown
+let shuttingDown = false;
+async function shutdown(signal: string) {
+  if (shuttingDown) return;
+  shuttingDown = true;
+  console.log(`\n${signal} received, shutting down gracefully...`);
+
+  cleanupCron.stop();
+
+  try {
+    await app.close();
+    console.log("HTTP server closed");
+  } catch (err) {
+    console.error("Error closing HTTP server:", err);
+  }
+
+  try {
+    await shutdownWorkerPool();
+    console.log("Worker pool shut down");
+  } catch (err) {
+    console.error("Error shutting down worker pool:", err);
+  }
+
+  try {
+    const { shutdownDispatcher } = await import("@stirling-image/ai");
+    shutdownDispatcher();
+    console.log("Python dispatcher shut down");
+  } catch {
+    // AI package may not be available
+  }
+
+  try {
+    const { sqlite: sqliteConn } = await import("./db/index.js");
+    sqliteConn.close();
+    console.log("Database connection closed");
+  } catch (err) {
+    console.error("Error closing database:", err);
+  }
+
+  process.exit(0);
+}
+
+process.on("SIGTERM", () => shutdown("SIGTERM"));
+process.on("SIGINT", () => shutdown("SIGINT"));

apps/api/src/lib/cleanup.ts

@@ -43,7 +43,7 @@ export function shouldRunStartupCleanup(): boolean {
   }
 }
 
-export function startCleanupCron() {
+export function startCleanupCron(): { stop: () => void } {
   // Ensure workspace directory exists
   mkdirSync(env.WORKSPACE_PATH, { recursive: true });
 
@@ -97,9 +97,16 @@ export function startCleanupCron() {
   }
 
   // Schedule recurring cleanup
-  setInterval(cleanup, intervalMs);
-  setInterval(purgeExpiredSessions, 60 * 60 * 1000); // Hourly
+  const cleanupTimer = setInterval(cleanup, intervalMs);
+  const sessionTimer = setInterval(purgeExpiredSessions, 60 * 60 * 1000); // Hourly
   console.log(
     `Cleanup scheduled: every ${env.CLEANUP_INTERVAL_MINUTES}m, max age configurable (env default: ${env.FILE_MAX_AGE_HOURS}h)`,
   );
+
+  return {
+    stop: () => {
+      clearInterval(cleanupTimer);
+      clearInterval(sessionTimer);
+    },
+  };
 }

apps/api/src/lib/file-storage.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { mkdir, unlink, writeFile } from "node:fs/promises";
+import { mkdir, readFile, unlink, writeFile } from "node:fs/promises";
 import { extname, join } from "node:path";
 import { env } from "../config.js";
 
@@ -49,3 +49,39 @@ export async function deleteStoredFile(storedName: string): Promise<void> {
 export function getStoredFilePath(storedName: string): string {
   return join(env.FILES_STORAGE_PATH, storedName);
 }
+
+// ── Thumbnail cache ──────────────────────────────────────────────────
+
+const THUMB_DIR = ".thumbs";
+let thumbDirReady = false;
+
+async function ensureThumbDir(): Promise<void> {
+  if (thumbDirReady) return;
+  await mkdir(join(env.FILES_STORAGE_PATH, THUMB_DIR), { recursive: true });
+  thumbDirReady = true;
+}
+
+function thumbPath(storedName: string): string {
+  return join(env.FILES_STORAGE_PATH, THUMB_DIR, `${storedName}.thumb.jpg`);
+}
+
+export async function getCachedThumbnail(storedName: string): Promise<Buffer | null> {
+  try {
+    return await readFile(thumbPath(storedName));
+  } catch {
+    return null;
+  }
+}
+
+export async function saveThumbnail(storedName: string, buffer: Buffer): Promise<void> {
+  await ensureThumbDir();
+  await writeFile(thumbPath(storedName), buffer);
+}
+
+export async function deleteThumbnail(storedName: string): Promise<void> {
+  try {
+    await unlink(thumbPath(storedName));
+  } catch {
+    // Thumbnail may not exist
+  }
+}

apps/api/src/lib/image-worker.ts

@@ -0,0 +1,61 @@
+/**
+ * Piscina worker that executes image tool processing in a worker thread.
+ *
+ * On first call, it imports all tool registration modules using a mock
+ * Fastify instance (only the registry-populating side effects are needed,
+ * not the HTTP route registrations). Subsequent calls reuse the populated
+ * registry for O(1) lookup.
+ */
+import { autoOrient } from "./auto-orient.js";
+
+export interface WorkerInput {
+  toolId: string;
+  inputBuffer: Buffer;
+  settings: unknown;
+  filename: string;
+}
+
+export interface WorkerOutput {
+  buffer: Buffer;
+  filename: string;
+  contentType: string;
+}
+
+let registryReady = false;
+
+async function ensureRegistry(): Promise<void> {
+  if (registryReady) return;
+
+  // Create a minimal mock that satisfies the register functions.
+  // createToolRoute calls app.post() (no-op here) and toolRegistry.set() (the part we want).
+  // AI tools also call app.post() and registerToolProcessFn() (also populates the registry).
+  const mockApp = {
+    post: () => {},
+    get: () => {},
+    log: { info: () => {}, warn: () => {}, error: () => {} },
+  };
+
+  const { registerToolRoutes } = await import("../routes/tools/index.js");
+  await registerToolRoutes(mockApp as never);
+  registryReady = true;
+}
+
+export default async function processInWorker(input: WorkerInput): Promise<WorkerOutput> {
+  await ensureRegistry();
+
+  const { getToolConfig } = await import("../routes/tool-factory.js");
+  const config = getToolConfig(input.toolId);
+
+  if (!config) {
+    throw new Error(`Tool "${input.toolId}" not found in worker registry`);
+  }
+
+  const oriented = await autoOrient(Buffer.from(input.inputBuffer));
+  const result = await config.process(oriented, input.settings, input.filename);
+
+  return {
+    buffer: result.buffer,
+    filename: result.filename,
+    contentType: result.contentType,
+  };
+}

apps/api/src/lib/worker-pool.ts

@@ -0,0 +1,37 @@
+/**
+ * Worker pool for offloading CPU-bound image processing from the main event loop.
+ *
+ * Uses Piscina (backed by worker_threads) so Sharp operations don't block
+ * HTTP request handling, SSE streams, or health checks.
+ */
+import { availableParallelism } from "node:os";
+import { dirname, resolve } from "node:path";
+import { fileURLToPath } from "node:url";
+import Piscina from "piscina";
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+
+// Size the pool: leave 1 thread for the event loop, min 1 worker
+const maxThreads = Math.max(1, Math.min(availableParallelism() - 1, 4));
+
+let pool: Piscina | null = null;
+
+export function getWorkerPool(): Piscina {
+  if (!pool) {
+    pool = new Piscina({
+      filename: resolve(__dirname, "image-worker.ts"),
+      // Inherit tsx loader flags from the main process so .ts files work in workers
+      execArgv: [...process.execArgv],
+      maxThreads,
+      idleTimeout: 30000,
+    });
+  }
+  return pool;
+}
+
+export async function shutdownWorkerPool(): Promise<void> {
+  if (pool) {
+    await pool.destroy();
+    pool = null;
+  }
+}

apps/api/src/routes/progress.ts

@@ -4,8 +4,13 @@
  * GET /api/v1/jobs/:jobId/progress
  *
  * Sends Server-Sent Events with progress data until the job finishes.
+ *
+ * Progress is held in-memory for real-time SSE delivery and also
+ * persisted to the `jobs` table so that state survives container restarts.
  */
+import { eq } from "drizzle-orm";
 import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
+import { db, schema } from "../db/index.js";
 
 export interface JobProgress {
   jobId: string;
@@ -34,11 +39,128 @@ const jobProgressStore = new Map<string, JobProgress>();
 /** SSE listeners waiting for updates, keyed by jobId. */
 const listeners = new Map<string, Set<(data: JobProgress | SingleFileProgress) => void>>();
 
+// ── DB persistence helpers ──────────────────────────────────────────
+
+function persistJobProgress(progress: JobProgress): void {
+  try {
+    const completionRatio =
+      progress.totalFiles > 0 ? progress.completedFiles / progress.totalFiles : 0;
+    const existing = db
+      .select({ id: schema.jobs.id })
+      .from(schema.jobs)
+      .where(eq(schema.jobs.id, progress.jobId))
+      .get();
+
+    if (existing) {
+      db.update(schema.jobs)
+        .set({
+          status: progress.status,
+          progress: completionRatio,
+          error: progress.errors.length > 0 ? JSON.stringify(progress.errors) : null,
+          completedAt:
+            progress.status === "completed" || progress.status === "failed" ? new Date() : null,
+        })
+        .where(eq(schema.jobs.id, progress.jobId))
+        .run();
+    } else {
+      db.insert(schema.jobs)
+        .values({
+          id: progress.jobId,
+          type: "batch",
+          status: progress.status,
+          progress: completionRatio,
+          inputFiles: JSON.stringify({ totalFiles: progress.totalFiles }),
+          error: progress.errors.length > 0 ? JSON.stringify(progress.errors) : null,
+        })
+        .run();
+    }
+  } catch {
+    // DB persistence is best-effort; don't break real-time SSE
+  }
+}
+
+function persistSingleFileProgress(progress: Omit<SingleFileProgress, "type">): void {
+  try {
+    const status =
+      progress.phase === "complete"
+        ? "completed"
+        : progress.phase === "failed"
+          ? "failed"
+          : "processing";
+    const existing = db
+      .select({ id: schema.jobs.id })
+      .from(schema.jobs)
+      .where(eq(schema.jobs.id, progress.jobId))
+      .get();
+
+    if (existing) {
+      db.update(schema.jobs)
+        .set({
+          status,
+          progress: progress.percent / 100,
+          error: progress.error ?? null,
+          completedAt: status === "completed" || status === "failed" ? new Date() : null,
+        })
+        .where(eq(schema.jobs.id, progress.jobId))
+        .run();
+    } else {
+      db.insert(schema.jobs)
+        .values({
+          id: progress.jobId,
+          type: "single",
+          status,
+          progress: progress.percent / 100,
+          inputFiles: "[]",
+          error: progress.error ?? null,
+        })
+        .run();
+    }
+  } catch {
+    // Best-effort
+  }
+}
+
+/**
+ * Mark any jobs left in "processing" or "queued" state as failed.
+ * Called once at startup to recover from unclean shutdown.
+ */
+export function recoverStaleJobs(): void {
+  try {
+    const result = db
+      .update(schema.jobs)
+      .set({
+        status: "failed",
+        error: "Server restarted while job was in progress",
+        completedAt: new Date(),
+      })
+      .where(eq(schema.jobs.status, "processing"))
+      .run();
+    const result2 = db
+      .update(schema.jobs)
+      .set({
+        status: "failed",
+        error: "Server restarted while job was queued",
+        completedAt: new Date(),
+      })
+      .where(eq(schema.jobs.status, "queued"))
+      .run();
+    const total = result.changes + result2.changes;
+    if (total > 0) {
+      console.log(`Recovered ${total} stale jobs from previous run`);
+    }
+  } catch {
+    // DB not ready
+  }
+}
+
+// ── Public API (unchanged signatures) ───────────────────────────────
+
 /**
  * Create or update progress for a job.
  */
 export function updateJobProgress(progress: JobProgress): void {
   jobProgressStore.set(progress.jobId, progress);
+  persistJobProgress(progress);
   // Notify all SSE listeners
   const subs = listeners.get(progress.jobId);
   if (subs) {
@@ -57,6 +179,7 @@ export function updateJobProgress(progress: JobProgress): void {
 
 export function updateSingleFileProgress(progress: Omit<SingleFileProgress, "type">): void {
   const event: SingleFileProgress = { ...progress, type: "single" };
+  persistSingleFileProgress(progress);
   const subs = listeners.get(progress.jobId);
   if (subs) {
     for (const cb of subs) {

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

@@ -9,6 +9,8 @@ import { db, schema } from "../db/index.js";
 import { autoOrient } from "../lib/auto-orient.js";
 import { validateImageBuffer } from "../lib/file-validation.js";
 import { sanitizeFilename } from "../lib/filename.js";
+import type { WorkerInput, WorkerOutput } from "../lib/image-worker.js";
+import { getWorkerPool } from "../lib/worker-pool.js";
 import { createWorkspace } from "../lib/workspace.js";
 
 export interface ToolRouteConfig<T> {
@@ -41,6 +43,16 @@ export interface AnyToolRouteConfig {
  */
 const toolRegistry = new Map<string, AnyToolRouteConfig>();
 
+/** Tools that use the Python bridge and should NOT be offloaded to workers. */
+const SKIP_WORKER_TOOLS = new Set([
+  "remove-background",
+  "upscale",
+  "ocr",
+  "blur-faces",
+  "erase-object",
+  "smart-crop",
+]);
+
 /**
  * Retrieve a registered tool config by its ID.
  */
@@ -151,12 +163,43 @@ export function createToolRoute<T>(app: FastifyInstance, config: ToolRouteConfig
         return reply.status(400).send({ error: "Settings must be valid JSON" });
       }
 
-      // Auto-orient based on EXIF metadata before processing.
-      const processBuffer = await autoOrient(fileBuffer);
-
-      // Process the image
+      // Process the image (worker thread or main thread)
       try {
-        const result = await config.process(processBuffer, settings, filename);
+        let result: { buffer: Buffer; filename: string; contentType: string };
+
+        // Offload to worker thread for non-AI tools.
+        // Falls back to main-thread processing on any worker error.
+        // Disabled in test environments where worker_threads can't load .ts files.
+        const useWorker = !SKIP_WORKER_TOOLS.has(config.toolId) && process.env.NODE_ENV !== "test";
+        if (useWorker) {
+          try {
+            const pool = getWorkerPool();
+            const workerInput: WorkerInput = {
+              toolId: config.toolId,
+              inputBuffer: fileBuffer,
+              settings,
+              filename,
+            };
+            const workerResult: WorkerOutput = await pool.run(workerInput);
+            result = {
+              buffer: Buffer.from(workerResult.buffer),
+              filename: workerResult.filename,
+              contentType: workerResult.contentType,
+            };
+          } catch (workerErr) {
+            // Worker failed - fall back to main-thread processing
+            request.log.warn(
+              { workerErr, toolId: config.toolId },
+              "Worker processing failed, falling back to main thread",
+            );
+            const processBuffer = await autoOrient(fileBuffer);
+            result = await config.process(processBuffer, settings, filename);
+          }
+        } else {
+          // AI tools: always main thread (they use Python bridge)
+          const processBuffer = await autoOrient(fileBuffer);
+          result = await config.process(processBuffer, settings, filename);
+        }
 
         // Create workspace and save output
         const jobId = randomUUID();

apps/api/src/routes/user-files.ts

@@ -17,7 +17,14 @@ import type { FastifyInstance, FastifyReply, FastifyRequest } from "fastify";
 import sharp from "sharp";
 import { db, schema, sqlite } from "../db/index.js";
 import { auditLog } from "../lib/audit.js";
-import { deleteStoredFile, getStoredFilePath, saveFile } from "../lib/file-storage.js";
+import {
+  deleteStoredFile,
+  deleteThumbnail,
+  getCachedThumbnail,
+  getStoredFilePath,
+  saveFile,
+  saveThumbnail,
+} from "../lib/file-storage.js";
 import { validateImageBuffer } from "../lib/file-validation.js";
 import { sanitizeFilename } from "../lib/filename.js";
 import { getAuthUser } from "../plugins/auth.js";
@@ -342,6 +349,15 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
         return reply.status(404).send({ error: "File not found" });
       }
 
+      // Serve from disk cache if available
+      const cached = await getCachedThumbnail(file.storedName);
+      if (cached) {
+        return reply
+          .header("Content-Type", "image/jpeg")
+          .header("Cache-Control", "public, max-age=86400, immutable")
+          .send(cached);
+      }
+
       const filePath = getStoredFilePath(file.storedName);
 
       try {
@@ -350,9 +366,12 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
           .jpeg({ quality: 80 })
           .toBuffer();
 
+        // Cache to disk (non-blocking, don't fail the request)
+        saveThumbnail(file.storedName, thumbnail).catch(() => {});
+
         return reply
           .header("Content-Type", "image/jpeg")
-          .header("Cache-Control", "public, max-age=86400")
+          .header("Cache-Control", "public, max-age=86400, immutable")
           .send(thumbnail);
       } catch {
         return reply.status(422).send({ error: "Could not generate thumbnail" });
@@ -412,6 +431,7 @@ export async function userFileRoutes(app: FastifyInstance): Promise<void> {
 
       for (const row of chainRows) {
         await deleteStoredFile(row.stored_name);
+        await deleteThumbnail(row.stored_name);
         db.delete(schema.userFiles).where(eq(schema.userFiles.id, row.id)).run();
         deletedCount++;
       }