feat(upload): async photo processing — backend (PR-B part 1)

Move thumbnail / EXIF / dimensions / watermark / webhook work off the
upload request thread and into a background worker pool. Upload
requests now return 202 in seconds even on NFS-backed storage; the
worker(s) drain the pending queue independently and update each
photo's processing_status to 'complete' or 'failed' on its own.

Schema (migration 085_async_photo_processing.js):
  - photos.processing_status     enum default 'complete' (existing
                                 rows are already done)
  - photos.processing_error      populated on 'failed'
  - photos.processing_started_at timestamp for janitor recovery
  - photos.upload_id             groups all photos from one upload
                                 request so the frontend can poll
                                 status by group
  - indexes on processing_status and upload_id for queue lookups

services/photoProcessor.js
  - queueFilesForProcessing(files, options) — shared helper used by
    the admin and gallery upload routes. Moves files to final storage
    + inserts pending rows; returns { uploadId, photos, errors }.
  - processPhoto(photoId) — worker-mode: reads original from storage
    via withLocalCopy (transparent local/S3), generates thumbnail and
    EXIF/dimensions or video metadata, queues watermark, fires
    photo.uploaded webhook, marks 'complete'. Throws => caller marks
    'failed' with the error message.
  - processUploadedPhotos kept untouched — chunkedUploadService still
    uses the synchronous path.

services/backgroundProcessor.js (new)
  - N independent worker loops per backend instance (default 2,
    UPLOAD_PROCESSOR_CONCURRENCY env override).
  - Multi-pod safe: postgres SELECT FOR UPDATE SKIP LOCKED, sqlite
    UPDATE-with-status-guard. Pods race on rows, exactly one wins.
  - Janitor every minute resets photos stuck in 'processing' for >10
    minutes (worker died, pod restarted) back to 'pending'.
  - UPLOAD_PROCESSOR_DISABLED=true opt-out for CI/test.
  - Started from server.js after the other long-running workers.

routes/adminPhotos.js — POST /:eventId/upload
  - Replaced batch-of-25 sync processing loop with per-file
    move-to-storage + insert-pending. Response is now 202 with
    upload_id, count, photo_ids in addition to the legacy
    successCount / replacedCount fields the existing frontend reads.
  - Per-request temp directory cleanup is now a single idempotent
    handler on res.finish/res.close (was three inline blocks for
    error paths only, leaking dirs on success — original bug from
    contributor analysis).
  - GET /uploads/:upload_id/status — JSON snapshot of pending /
    processing / complete / failed counts plus per-photo state.
  - GET /uploads/:upload_id/stream — SSE upgrade. Polls internally
    every 1.5s, emits on snapshot change, ends when all photos
    reach a terminal state.
  - POST /photos/:photoId/retry — flips a 'failed' photo back to
    'pending' so the worker picks it up again.
  - GET /:eventId/thumbnail/:photoId now returns 503 with Retry-After
    while the photo is still pending/processing, and 422 on 'failed'.
    The admin grid renders placeholders accordingly.

routes/gallery.js — POST /:eventId/upload (guest)
  - Refactored to use queueFilesForProcessing instead of the synchronous
    processUploadedPhotos. Same 202 + upload_id shape.
  - GET /:slug/photos now filters processing_status to 'complete' (or
    NULL for pre-migration rows) so guests never see in-flight photos.

Side-effect timing change:
  - photo.uploaded webhook now fires from the worker after the photo
    is actually processed (thumbnail + dimensions populated) instead
    of from inside the upload request. Same payload fields. Worth a
    one-line note in the changelog.

backend/src/services/backgroundProcessor.js

@@ -0,0 +1,176 @@
+/**
+ * Background photo-processing worker pool.
+ *
+ * Polls `photos.processing_status = 'pending'`, atomically claims one
+ * row per worker, hands it to `photoProcessor.processPhoto(photoId)`,
+ * and marks the row 'complete' or 'failed' depending on outcome. A
+ * janitor loop resets rows stuck in 'processing' for too long (worker
+ * died, pod restarted, etc.).
+ *
+ * Concurrency model: N independent worker loops per backend instance.
+ * Multi-pod safe via:
+ *   - Postgres: SELECT ... FOR UPDATE SKIP LOCKED — pods race for rows,
+ *     only one wins, the others move on.
+ *   - SQLite:   SELECT then UPDATE-with-status-guard — second writer
+ *     loses the guard and tries again (single-pod typical; the guard
+ *     is enough for the rare two-process case during dev).
+ *
+ * Tunables (env, all optional):
+ *   UPLOAD_PROCESSOR_CONCURRENCY        default 2
+ *   UPLOAD_PROCESSOR_POLL_MS            default 1000
+ *   UPLOAD_PROCESSOR_STUCK_TIMEOUT_MS   default 600000  (10 minutes)
+ *   UPLOAD_PROCESSOR_DISABLED           default false   (set 'true' to opt out, e.g. in CI)
+ */
+
+const { db } = require('../database/db');
+const logger = require('../utils/logger');
+const { processPhoto } = require('./photoProcessor');
+
+const POLL_INTERVAL_MS = parseInt(process.env.UPLOAD_PROCESSOR_POLL_MS || '1000', 10);
+const CONCURRENCY = Math.max(1, parseInt(process.env.UPLOAD_PROCESSOR_CONCURRENCY || '2', 10));
+const STUCK_TIMEOUT_MS = parseInt(process.env.UPLOAD_PROCESSOR_STUCK_TIMEOUT_MS || '600000', 10);
+const JANITOR_INTERVAL_MS = 60 * 1000;
+
+let running = false;
+let workerHandles = [];
+let janitorHandle = null;
+
+const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+
+function isPostgres() {
+  const c = db.client.config.client;
+  return c === 'pg' || (typeof c === 'string' && c.includes('postgres'));
+}
+
+/**
+ * Atomically claim the oldest pending photo. Returns the row or null.
+ * The claimed row's processing_status is now 'processing' and
+ * processing_started_at is set so the janitor can recover it.
+ */
+async function claimNextPhoto() {
+  if (isPostgres()) {
+    return db.transaction(async (trx) => {
+      const row = await trx('photos')
+        .where('processing_status', 'pending')
+        .orderBy('id', 'asc')
+        .forUpdate()
+        .skipLocked()
+        .first();
+      if (!row) return null;
+      await trx('photos').where('id', row.id).update({
+        processing_status: 'processing',
+        processing_started_at: new Date(),
+      });
+      return row;
+    });
+  }
+
+  // SQLite path — no SKIP LOCKED, but the UPDATE-with-guard ensures
+  // exactly one worker wins per row.
+  return db.transaction(async (trx) => {
+    const row = await trx('photos')
+      .where('processing_status', 'pending')
+      .orderBy('id', 'asc')
+      .first();
+    if (!row) return null;
+    const updated = await trx('photos')
+      .where({ id: row.id, processing_status: 'pending' })
+      .update({
+        processing_status: 'processing',
+        processing_started_at: new Date(),
+      });
+    return updated > 0 ? row : null;
+  });
+}
+
+async function workerLoop(workerIdx) {
+  while (running) {
+    let claimed;
+    try {
+      claimed = await claimNextPhoto();
+    } catch (e) {
+      logger.warn(`backgroundProcessor[${workerIdx}]: claim error`, { error: e.message });
+      await sleep(POLL_INTERVAL_MS);
+      continue;
+    }
+
+    if (!claimed) {
+      await sleep(POLL_INTERVAL_MS);
+      continue;
+    }
+
+    try {
+      await processPhoto(claimed.id);
+    } catch (err) {
+      logger.error(`backgroundProcessor[${workerIdx}]: photo ${claimed.id} failed`, {
+        error: err.message,
+        stack: err.stack,
+      });
+      try {
+        await db('photos').where({ id: claimed.id }).update({
+          processing_status: 'failed',
+          processing_error: String(err.message || err).slice(0, 1000),
+        });
+      } catch (updateErr) {
+        logger.error(`backgroundProcessor[${workerIdx}]: failed to mark photo ${claimed.id} as failed`, {
+          error: updateErr.message,
+        });
+      }
+    }
+  }
+}
+
+async function janitorLoop() {
+  while (running) {
+    try {
+      const cutoff = new Date(Date.now() - STUCK_TIMEOUT_MS);
+      const reset = await db('photos')
+        .where('processing_status', 'processing')
+        .where('processing_started_at', '<', cutoff)
+        .update({ processing_status: 'pending', processing_started_at: null });
+      if (reset > 0) {
+        logger.warn(
+          `backgroundProcessor: janitor reset ${reset} stuck photo(s) from 'processing' to 'pending'`
+        );
+      }
+    } catch (e) {
+      logger.warn('backgroundProcessor: janitor error', { error: e.message });
+    }
+    await sleep(JANITOR_INTERVAL_MS);
+  }
+}
+
+function start() {
+  if (running) return;
+  if (process.env.UPLOAD_PROCESSOR_DISABLED === 'true') {
+    logger.info('backgroundProcessor: disabled via UPLOAD_PROCESSOR_DISABLED');
+    return;
+  }
+
+  running = true;
+  workerHandles = [];
+  for (let i = 0; i < CONCURRENCY; i++) {
+    workerHandles.push(
+      workerLoop(i).catch((e) =>
+        logger.error(`backgroundProcessor[${i}]: crashed`, { error: e.message, stack: e.stack })
+      )
+    );
+  }
+  janitorHandle = janitorLoop().catch((e) =>
+    logger.error('backgroundProcessor: janitor crashed', { error: e.message, stack: e.stack })
+  );
+
+  logger.info(
+    `backgroundProcessor: started ${CONCURRENCY} worker(s), poll=${POLL_INTERVAL_MS}ms, stuck=${STUCK_TIMEOUT_MS}ms`
+  );
+}
+
+async function stop() {
+  if (!running) return;
+  running = false;
+  await Promise.all([...workerHandles, janitorHandle].filter(Boolean));
+  workerHandles = [];
+  janitorHandle = null;
+}
+
+module.exports = { start, stop, claimNextPhoto };

backend/src/services/photoProcessor.js

@@ -1,10 +1,12 @@
 const path = require('path');
 const fs = require('fs').promises;
 const { db } = require('../database/db');
-const { generateThumbnail } = require('./imageProcessor');
+const { generateThumbnail, extractCaptureDate, withLocalCopy } = require('./imageProcessor');
 const { generatePhotoFilename } = require('../utils/filenameSanitizer');
 const { processUploadedVideo, isVideoMimeType } = require('./videoProcessor');
 const { getStorage } = require('./storage');
+const { resolvePhotoStorageKey } = require('./photoResolver');
+const logger = require('../utils/logger');
 
 function normalizeFiles(files) {
   // Handle null, undefined, or falsy values
@@ -287,6 +289,226 @@ async function processUploadedPhotos(files, eventId, uploadedBy = 'admin', categ
   return uploadedPhotos;
 }
 
+/**
+ * Queue uploaded files for async processing.
+ *
+ * Moves each file from its multer temp path to the final storage key
+ * and inserts a `photos` row with `processing_status = 'pending'` and
+ * a shared `upload_id`. The background worker
+ * (services/backgroundProcessor.js) picks up pending rows, generates
+ * thumbnails / EXIF / dimensions, then flips status to 'complete'
+ * (or 'failed' with the error).
+ *
+ * Used by both the admin upload route and the gallery (guest) upload
+ * route so they share the same fast-return semantics.
+ *
+ * Options:
+ *   - eventId           required
+ *   - photoType         'individual' | 'collage' (default 'individual')
+ *   - categoryId        numeric category id or null
+ *   - uploadId          optional pre-generated upload id (caller can
+ *                       provide it for chunked uploads that span
+ *                       multiple HTTP requests)
+ *
+ * Returns: { uploadId, photos: [{id, filename, size, category_id}], errors: [{filename, error}] }
+ */
+async function queueFilesForProcessing(files, options = {}) {
+  const crypto = require('crypto');
+  const { eventId, photoType = 'individual', categoryId = null, uploadId: providedUploadId } = options;
+  const uploadId = providedUploadId || crypto.randomBytes(16).toString('hex');
+
+  const event = await db('events').where({ id: eventId }).first();
+  if (!event) throw new Error(`Event ${eventId} not found`);
+
+  const fileList = normalizeFiles(files);
+  const queued = [];
+  const errors = [];
+
+  if (fileList.length === 0) return { uploadId, photos: queued, errors };
+
+  // Counter base — same approximation the upload route used pre-async.
+  // Strict uniqueness is still enforced by the filename template; on a
+  // collision the worker would just fail one photo.
+  const existingCount = await db('photos')
+    .where({ event_id: eventId, type: photoType })
+    .count('id as count')
+    .first();
+  let counter = (parseInt(existingCount?.count) || 0) + 1;
+
+  const storage = getStorage();
+  const finalDestPathRel = path.posix.join('events/active', event.slug);
+  const categoryName = photoType === 'collage' ? 'collages' : 'individual';
+
+  for (const file of fileList) {
+    const tempPath = file?.path || file?.filepath || file?.tempFilePath;
+    try {
+      if (!tempPath) {
+        throw new Error('Uploaded file is missing a temporary path');
+      }
+      const tempStats = await fs.stat(tempPath);
+      if (tempStats.size === 0) {
+        throw new Error('File is empty - upload may have been interrupted');
+      }
+
+      const extension = path.extname(file.originalname);
+      const newFilename = generatePhotoFilename(event.event_name, categoryName, counter, extension);
+      counter += 1;
+
+      const finalKey = path.posix.join(finalDestPathRel, newFilename);
+      const relativePath = path.posix.join(event.slug, newFilename);
+      const isVideo = isVideoMimeType(file.mimetype);
+
+      // Move to storage first so the file is at its recorded path by the
+      // time the worker picks up the row.
+      await storage.putFromFile(finalKey, tempPath, { contentType: file.mimetype });
+      await fs.unlink(tempPath).catch(() => {});
+
+      const stat = await storage.stat(finalKey);
+      if (!stat || stat.size !== tempStats.size) {
+        throw new Error(`Size mismatch after upload: expected ${tempStats.size}, got ${stat ? stat.size : 'null'}`);
+      }
+
+      const inserted = await db('photos')
+        .insert({
+          event_id: parseInt(eventId, 10),
+          filename: newFilename,
+          original_filename: file.originalname,
+          path: relativePath,
+          thumbnail_path: null,
+          type: photoType,
+          category_id: categoryId,
+          size_bytes: tempStats.size,
+          captured_at: null,
+          media_type: isVideo ? 'video' : 'image',
+          mime_type: file.mimetype,
+          processing_status: 'pending',
+          upload_id: uploadId,
+        })
+        .returning('id');
+      const photoId = inserted[0]?.id || inserted[0];
+
+      queued.push({
+        id: photoId,
+        filename: newFilename,
+        size: tempStats.size,
+        category_id: categoryId,
+      });
+    } catch (err) {
+      errors.push({ filename: file?.originalname || 'unknown', error: err.message });
+    }
+  }
+
+  return { uploadId, photos: queued, errors };
+}
+
+/**
+ * Worker-mode processing for a single already-stored photo.
+ *
+ * Called by the background processor after a row has been claimed
+ * (`processing_status` == 'processing'). The photo file already exists
+ * at its final storage key — this function reads it back, generates a
+ * thumbnail, extracts EXIF + dimensions (or video metadata), then
+ * updates the photo row to `complete` and fires the queued side
+ * effects (watermark, webhook).
+ *
+ * Throwing causes the background processor to mark the row as
+ * 'failed' with the error message; partial successes (e.g. thumbnail
+ * fails but dimensions succeed) are persisted up to the failure point.
+ */
+async function processPhoto(photoId) {
+  const photo = await db('photos').where({ id: photoId }).first();
+  if (!photo) throw new Error(`Photo ${photoId} not found`);
+
+  const event = await db('events').where({ id: photo.event_id }).first();
+  if (!event) throw new Error(`Event ${photo.event_id} not found for photo ${photoId}`);
+
+  const sourceKey = resolvePhotoStorageKey(event, photo);
+  const isVideo =
+    photo.media_type === 'video' ||
+    (typeof photo.mime_type === 'string' && photo.mime_type.startsWith('video/'));
+
+  const updateData = {};
+
+  // withLocalCopy materialises the original from the storage backend so
+  // sharp/ffmpeg can read it. For local storage this is a free O(1) path
+  // resolution; for S3 it downloads to a tmpdir that's auto-cleaned.
+  await withLocalCopy(sourceKey, async (localPath) => {
+    if (!photo.captured_at && !isVideo) {
+      try {
+        const captured = await extractCaptureDate(localPath);
+        if (captured) updateData.captured_at = captured;
+      } catch (e) {
+        logger.warn(`processPhoto: EXIF extraction failed for ${photoId}`, { error: e.message });
+      }
+    }
+
+    if (isVideo) {
+      const videoThumbnailKey = path.posix.join(
+        'thumbnails',
+        `thumb_${photo.filename.replace(/\.[^.]+$/, '.jpg')}`
+      );
+      const result = await processUploadedVideo(localPath, videoThumbnailKey);
+      updateData.thumbnail_path = result.thumbnailKey;
+      if (result.metadata) {
+        if (result.metadata.duration != null) updateData.duration = result.metadata.duration;
+        if (result.metadata.videoCodec) updateData.video_codec = result.metadata.videoCodec;
+        if (result.metadata.audioCodec) updateData.audio_codec = result.metadata.audioCodec;
+        if (result.metadata.width) updateData.width = result.metadata.width;
+        if (result.metadata.height) updateData.height = result.metadata.height;
+      }
+    } else {
+      try {
+        const thumbnailPath = await generateThumbnail(localPath);
+        if (thumbnailPath) updateData.thumbnail_path = thumbnailPath;
+      } catch (e) {
+        logger.warn(`processPhoto: thumbnail generation failed for ${photoId}`, { error: e.message });
+      }
+      try {
+        const sharp = require('sharp');
+        const metadata = await sharp(localPath).metadata();
+        if (metadata.width && metadata.height) {
+          updateData.width = metadata.width;
+          updateData.height = metadata.height;
+        }
+      } catch (e) {
+        logger.warn(`processPhoto: dimensions extraction failed for ${photoId}`, { error: e.message });
+      }
+    }
+  });
+
+  // Mark complete
+  updateData.processing_status = 'complete';
+  updateData.processing_error = null;
+  await db('photos').where({ id: photoId }).update(updateData);
+
+  // Side effects (best-effort, never fail the photo if these break)
+  if (!isVideo) {
+    const watermarkGeneratorService = require('./watermarkGeneratorService');
+    watermarkGeneratorService
+      .generateForPhoto(photoId)
+      .catch((err) => logger.warn(`processPhoto: watermark queue failed for ${photoId}`, { error: err.message }));
+  }
+
+  try {
+    const webhookService = require('./webhookService');
+    await webhookService.fire('photo.uploaded', {
+      event: { id: event.id, slug: event.slug, event_name: event.event_name },
+      photo: {
+        id: photo.id,
+        filename: photo.filename,
+        original_filename: photo.original_filename,
+        size_bytes: photo.size_bytes,
+      },
+    });
+  } catch (e) {
+    logger.warn(`processPhoto: webhook fire failed for ${photoId}`, { error: e.message });
+  }
+
+  return updateData;
+}
+
 module.exports = {
-  processUploadedPhotos
+  processUploadedPhotos,
+  queueFilesForProcessing,
+  processPhoto
 };