From 851744c3c4df7deba8d946b6592fdb5042c52a26 Mon Sep 17 00:00:00 2001 From: Paul Nothaft Date: Sat, 2 May 2026 00:25:51 +0200 Subject: [PATCH] =?UTF-8?q?feat(upload):=20async=20photo=20processing=20?= =?UTF-8?q?=E2=80=94=20backend=20(PR-B=20part=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../core/085_async_photo_processing.js | 72 +++ backend/server.js | 10 +- backend/src/routes/adminPhotos.js | 564 ++++++++++-------- backend/src/routes/gallery.js | 44 +- backend/src/services/backgroundProcessor.js | 176 ++++++ backend/src/services/photoProcessor.js | 226 ++++++- 6 files changed, 815 insertions(+), 277 deletions(-) create mode 100644 backend/migrations/core/085_async_photo_processing.js create mode 100644 backend/src/services/backgroundProcessor.js diff --git a/backend/migrations/core/085_async_photo_processing.js b/backend/migrations/core/085_async_photo_processing.js new file mode 100644 index 00000000..55d2d75f --- /dev/null +++ b/backend/migrations/core/085_async_photo_processing.js @@ -0,0 +1,72 @@ +/** + * Async photo-processing infrastructure. + * + * Adds: + * - photos.processing_status — enum: pending | processing | complete | failed + * - photos.processing_error — text, populated on 'failed' + * - photos.processing_started_at — claim timestamp for janitor recovery + * - photos.upload_id — groups all photos from one upload request + * so the frontend can poll/stream by group + * + * All existing rows default to 'complete' (they were processed synchronously + * before this migration and there's nothing pending). New uploads insert + * with 'pending' and a background worker (services/backgroundProcessor.js) + * picks them up. + * + * Partial-style indexes keep lookups fast as the queue drains. We use plain + * indexes here instead of postgres-specific WHERE clauses so the migration + * works on SQLite too; the workload (only-pending rows) keeps the index small. + */ + +exports.up = async function up(knex) { + if (!(await knex.schema.hasTable('photos'))) return; + + const hasStatus = await knex.schema.hasColumn('photos', 'processing_status'); + if (!hasStatus) { + await knex.schema.alterTable('photos', (table) => { + table.string('processing_status', 16).notNullable().defaultTo('complete'); + table.text('processing_error').nullable(); + table.timestamp('processing_started_at').nullable(); + table.string('upload_id', 64).nullable(); + }); + } + + // Indexes — wrap in try/catch so re-running the migration on a partially + // applied schema is a no-op rather than an error. + try { + await knex.schema.alterTable('photos', (table) => { + table.index(['processing_status'], 'idx_photos_processing_status'); + }); + } catch (_) { /* already exists */ } + + try { + await knex.schema.alterTable('photos', (table) => { + table.index(['upload_id'], 'idx_photos_upload_id'); + }); + } catch (_) { /* already exists */ } +}; + +exports.down = async function down(knex) { + if (!(await knex.schema.hasTable('photos'))) return; + + // Drop indexes first (best-effort) + try { + await knex.schema.alterTable('photos', (t) => t.dropIndex([], 'idx_photos_upload_id')); + } catch (_) { /* not present */ } + try { + await knex.schema.alterTable('photos', (t) => t.dropIndex([], 'idx_photos_processing_status')); + } catch (_) { /* not present */ } + + if (await knex.schema.hasColumn('photos', 'upload_id')) { + await knex.schema.alterTable('photos', (t) => t.dropColumn('upload_id')); + } + if (await knex.schema.hasColumn('photos', 'processing_started_at')) { + await knex.schema.alterTable('photos', (t) => t.dropColumn('processing_started_at')); + } + if (await knex.schema.hasColumn('photos', 'processing_error')) { + await knex.schema.alterTable('photos', (t) => t.dropColumn('processing_error')); + } + if (await knex.schema.hasColumn('photos', 'processing_status')) { + await knex.schema.alterTable('photos', (t) => t.dropColumn('processing_status')); + } +}; diff --git a/backend/server.js b/backend/server.js index 69879c89..77b9c32d 100644 --- a/backend/server.js +++ b/backend/server.js @@ -23,6 +23,7 @@ const { startExpirationChecker } = require('./src/services/expirationChecker'); const { initializeTransporter, startEmailQueueProcessor } = require('./src/services/emailProcessor'); const { startBackupService } = require('./src/services/backupService'); const { startScheduledBackups } = require('./src/services/databaseBackup'); +const backgroundProcessor = require('./src/services/backgroundProcessor'); const { maintenanceMiddleware } = require('./src/middleware/maintenance'); const { sessionTimeoutMiddleware } = require('./src/middleware/sessionTimeout'); const { errorHandler, notFoundHandler } = require('./src/middleware/errorHandler'); @@ -660,10 +661,15 @@ async function startServer() { // Start backup service await startBackupService(); - + // Start database backup service await startScheduledBackups(); - + + // Start the async photo-processing worker pool. Picks up + // photos in 'pending' state (from POST /upload) and runs the + // sharp/ffmpeg/EXIF pipeline off the request thread. + backgroundProcessor.start(); + app.listen(PORT, () => { logger.info(`Server running on port ${PORT}`); logger.info(`Admin interface: ${process.env.ADMIN_URL || 'http://localhost:3000'}`); diff --git a/backend/src/routes/adminPhotos.js b/backend/src/routes/adminPhotos.js index 1a577260..791ee012 100644 --- a/backend/src/routes/adminPhotos.js +++ b/backend/src/routes/adminPhotos.js @@ -5,8 +5,8 @@ const fs = require('fs').promises; const { db, logActivity } = require('../database/db'); const { adminAuth } = require('../middleware/auth'); const { requirePermission } = require('../middleware/permissions'); -const { generateThumbnail, ensureThumbnail, extractCaptureDate } = require('../services/imageProcessor'); -const { processUploadedVideo, isVideoMimeType } = require('../services/videoProcessor'); +const { ensureThumbnail } = require('../services/imageProcessor'); +const { isVideoMimeType } = require('../services/videoProcessor'); const { generatePhotoFilename } = require('../utils/filenameSanitizer'); const { escapeLikePattern } = require('../utils/sqlSecurity'); const { validateUploadedFiles } = require('../middleware/uploadValidation'); @@ -283,262 +283,97 @@ router.post('/:eventId/upload', adminAuth, requirePermission('photos.upload'), r filesToUpload = newFiles; } - // Process remaining new files in batches - const BATCH_SIZE = 25; // Increased batch size for better performance with large uploads + // Async-processing flow: + // 1. Move each file to its final storage location. + // 2. Insert a photo row with processing_status='pending' and a + // shared upload_id. EXIF / sharp / thumbnails / ffmpeg / + // watermark / webhook all happen in the background worker + // (services/backgroundProcessor.js) so the request returns in + // seconds even on NFS-backed storage. + // + // The previous code processed thumbnails+EXIF synchronously in + // batches of 25 inside this handler, which is why large uploads on + // slow storage looked frozen — see #357 review. + const crypto = require('crypto'); + const uploadId = crypto.randomBytes(16).toString('hex'); - for (let i = 0; i < filesToUpload.length; i += BATCH_SIZE) { - const batch = filesToUpload.slice(i, i + BATCH_SIZE); - - // Start a single transaction for the batch - const trx = await db.transaction(); - + // Counter base — same approximation as before. Strict uniqueness is + // already enforced by the filename template + DB unique index, so a + // small race here just retries a counter on conflict (rare). + 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(); + + for (const file of filesToUpload) { try { - // Get initial counter for this batch based on photo type - const existingCount = await trx('photos') - .where({ event_id: eventId, type: photoType }) - .count('id as count') - .first(); - let batchCounter = (parseInt(existingCount.count) || 0) + 1; - - const batchPhotos = []; - const fileRenameOperations = []; // Store rename operations to do after commit - - // First pass: prepare data and move files from temp to final location - for (let fileIndex = 0; fileIndex < batch.length; fileIndex++) { - const file = batch[fileIndex]; - const counter = batchCounter + fileIndex; - const tempPath = file.path; // Original temp path - - try { - // Verify file is complete before processing - const tempStats = await fs.stat(tempPath); - if (tempStats.size === 0) { - throw new Error('File is empty - upload may have been interrupted'); - } - - // Generate new filename - const extension = path.extname(file.originalname); - const newFilename = generatePhotoFilename( - event.event_name, - categoryName, - counter, - extension - ); - - // Storage key: events/active/{slug}/{newFilename} - const finalKey = path.posix.join(finalDestPathRel, newFilename); - // photo.path is stored relative to events/active so resolvePhotoStorageKey - // can rebuild the full key on read. - const relativePath = path.posix.join(event.slug, newFilename); - - // Extract capture date from EXIF metadata - let capturedAt = null; - try { - capturedAt = await extractCaptureDate(tempPath); - } catch (exifError) { - // Non-fatal - just log and continue without capture date - console.log(`Could not extract EXIF date for ${file.originalname}`); - } - - // Determine media type - const isVideo = isVideoMimeType(file.mimetype); - const mediaType = isVideo ? 'video' : 'image'; - - // Prepare photo data for batch insert - const photoData = { - event_id: parseInt(eventId), - filename: newFilename, - original_filename: file.originalname, // Preserve original filename for Lightroom export - path: relativePath, - thumbnail_path: null, // Will generate after successful commit - type: photoType, - category_id: parsedCategoryId, // Save the selected category - size_bytes: tempStats.size, // Use actual file size from stat - captured_at: capturedAt, // EXIF capture date (if available) - media_type: mediaType, - mime_type: file.mimetype - }; - - batchPhotos.push(photoData); - - // Store upload operation for later (after DB commit) - fileRenameOperations.push({ - tempPath: tempPath, - finalKey: finalKey, - filename: newFilename, - photoData: photoData - }); - } catch (error) { - console.error(`Error preparing file ${file.originalname}:`, error); - errors.push({ filename: file.originalname, error: error.message }); - } + const tempStats = await fs.stat(file.path); + if (tempStats.size === 0) { + throw new Error('File is empty - upload may have been interrupted'); } - - // Insert all photos in this batch - if (batchPhotos.length > 0) { - console.log(`Inserting batch of ${batchPhotos.length} photos with type: ${photoType}`); - - const insertedIds = await trx('photos').insert(batchPhotos).returning('id'); - - // No need to update counter as we calculate it dynamically - - // Commit the transaction first - await trx.commit(); - console.log(`Successfully committed batch of ${batchPhotos.length} photos`); - - // Now upload files from temp into the storage backend after successful commit - const storage = getStorage(); - for (let idx = 0; idx < fileRenameOperations.length; idx++) { - const operation = fileRenameOperations[idx]; - try { - // Process source-dependent steps (sharp/ffmpeg) FIRST while the - // tmp file is still on local disk, then upload the original and - // unlink the tmp. - const photoId = insertedIds[idx]?.id || insertedIds[idx]; - const isVideoFile = isVideoMimeType(operation.photoData.mime_type); - let thumbnailPath = null; - try { - if (isVideoFile) { - const videoThumbnailKey = path.posix.join( - 'thumbnails', - `thumb_${operation.filename.replace(/\.[^.]+$/, '.jpg')}` - ); - const result = await processUploadedVideo(operation.tempPath, videoThumbnailKey); - thumbnailPath = result.thumbnailKey; + const extension = path.extname(file.originalname); + const newFilename = generatePhotoFilename( + event.event_name, + categoryName, + counter, + extension + ); + counter += 1; - if (photoId && result.metadata) { - await db('photos') - .where({ id: photoId }) - .update({ - thumbnail_path: thumbnailPath, - duration: result.metadata.duration, - video_codec: result.metadata.videoCodec, - audio_codec: result.metadata.audioCodec, - width: result.metadata.width, - height: result.metadata.height - }); - } - } else { - thumbnailPath = await generateThumbnail(operation.tempPath); + const finalKey = path.posix.join(finalDestPathRel, newFilename); + const relativePath = path.posix.join(event.slug, newFilename); + const isVideo = isVideoMimeType(file.mimetype); - // Update the database with thumbnail path and image dimensions - if (photoId) { - const updateData = {}; - if (thumbnailPath) updateData.thumbnail_path = thumbnailPath; + // 1. Move file to its final storage key first. If the worker + // later picks up the photo row, the file is guaranteed to + // exist at the recorded path. + await storage.putFromFile(finalKey, file.path, { + contentType: file.mimetype, + }); + await fs.unlink(file.path).catch(() => {}); - try { - const sharp = require('sharp'); - const metadata = await sharp(operation.tempPath).metadata(); - if (metadata.width && metadata.height) { - updateData.width = metadata.width; - updateData.height = metadata.height; - } - } catch (metadataError) { - console.warn(`Could not extract image dimensions for ${operation.filename}:`, metadataError.message); - } - - if (Object.keys(updateData).length > 0) { - await db('photos') - .where({ id: photoId }) - .update(updateData); - } - } - } - } catch (thumbError) { - console.error(`Thumbnail/metadata processing failed for ${operation.filename}:`, thumbError.message); - } - - // Upload the original through the storage backend, then drop the - // local tmp file. We do this AFTER thumbnail/metadata processing - // so sharp/ffmpeg still have a local source to work from. - await storage.putFromFile(operation.finalKey, operation.tempPath, { - contentType: operation.photoData.mime_type, - }); - await fs.unlink(operation.tempPath).catch(() => {}); - - // Sanity check: round-trip the size we just wrote. - const stat = await storage.stat(operation.finalKey); - if (!stat || stat.size !== operation.photoData.size_bytes) { - throw new Error(`Size mismatch after upload: expected ${operation.photoData.size_bytes}, got ${stat ? stat.size : 'null'}`); - } - - // Queue watermark generation in background (non-blocking, images only) - if (photoId && !isVideoFile) { - watermarkGeneratorService.generateForPhoto(photoId) - .catch(err => console.warn(`Watermark generation queued failed for photo ${photoId}:`, err.message)); - } - - // Webhook (#327): per-photo upload event. - try { - const webhookService = require('../services/webhookService'); - await webhookService.fire('photo.uploaded', { - event: { id: parseInt(eventId, 10), slug: event.slug, event_name: event.event_name }, - photo: { - id: insertedIds[idx]?.id || insertedIds[idx], - filename: operation.filename, - original_filename: operation.photoData.original_filename, - size_bytes: operation.photoData.size_bytes, - }, - }); - } catch (e) { /* non-fatal */ } - - // Add to successful uploads - uploadedPhotos.push({ - id: insertedIds[idx]?.id || insertedIds[idx], - filename: operation.filename, - size: operation.photoData.size_bytes, - category_id: operation.photoData.category_id - }); - } catch (moveError) { - console.error(`Failed to upload ${operation.tempPath} → ${operation.finalKey}:`, moveError); - errors.push({ - filename: operation.filename, - error: `File upload failed: ${moveError.message}` - }); - - // Try to clean up the database entry if file move failed - if (insertedIds[idx]) { - const photoId = insertedIds[idx]?.id || insertedIds[idx]; - try { - await db('photos').where({ id: photoId }).delete(); - console.log(`Cleaned up database entry for failed photo ${photoId}`); - } catch (cleanupError) { - console.error(`Failed to clean up database entry:`, cleanupError); - } - } - } - } - } else { - // No photos to insert, just rollback - await trx.rollback(); + // Sanity check the round-tripped size — same guard as before. + 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'}` + ); } - } catch (error) { - console.error(`Error processing batch starting at index ${i}:`, error); - console.error('Stack trace:', error.stack); - - // Rollback if not already committed - if (!trx.isCompleted()) { - await trx.rollback(); - } - - // Add all files in this batch to errors - for (const file of batch) { - errors.push({ - filename: file.originalname, - error: `Batch processing failed: ${error.message}` - }); - } - } - } - - // Clean up temp upload directory - if (req.tempUploadPath) { - try { - await fs.rm(req.tempUploadPath, { recursive: true, force: true }); - console.log(`Cleaned up temp upload directory: ${req.tempUploadPath}`); - } catch (e) { - console.error('Failed to clean up temp upload directory:', e); + + // 2. Insert a pending photo row. The background processor + // will pick it up, generate thumbnail/dimensions/EXIF, and + // flip status to 'complete' (or 'failed' with the error). + 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: parsedCategoryId, + 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]; + + uploadedPhotos.push({ + id: photoId, + filename: newFilename, + size: tempStats.size, + category_id: parsedCategoryId, + }); + } catch (err) { + console.error(`Error queuing file ${file.originalname}:`, err); + errors.push({ filename: file.originalname, error: err.message }); } } @@ -561,12 +396,20 @@ router.post('/:eventId/upload', adminAuth, requirePermission('photos.upload'), r // Include any files that were invalid from the validation middleware const totalInvalidFiles = (req.invalidFiles || []).concat(errors); - // Prepare response + // Prepare response. The new fields (upload_id, count, photo_ids) + // are what the new frontend uses to poll for processing status; the + // existing fields (successCount, replacedCount, ...) are kept for + // back-compat with older clients that haven't upgraded yet. const totalAttempted = req.files.length + (req.invalidFiles ? req.invalidFiles.length : 0); - const uploadMsg = uploadedPhotos.length > 0 ? `${uploadedPhotos.length} uploaded` : ''; + const uploadMsg = uploadedPhotos.length > 0 ? `${uploadedPhotos.length} queued` : ''; const replaceMsg = replacedPhotos.length > 0 ? `${replacedPhotos.length} replaced` : ''; const parts = [uploadMsg, replaceMsg].filter(Boolean).join(', '); const response = { + // New async-processing fields + upload_id: uploadId, + count: uploadedPhotos.length, + photo_ids: uploadedPhotos.map((p) => p.id), + // Existing back-compat fields message: parts ? `Successfully ${parts}` : 'No photos processed', photos: uploadedPhotos, replaced: replacedPhotos, @@ -574,21 +417,22 @@ router.post('/:eventId/upload', adminAuth, requirePermission('photos.upload'), r skippedReplacements, totalFiles: totalAttempted, successCount: uploadedPhotos.length + replacedPhotos.length, - failureCount: totalInvalidFiles.length + failureCount: totalInvalidFiles.length, }; - + // Include error details if any files failed if (totalInvalidFiles.length > 0) { response.errors = totalInvalidFiles; - response.message = `Uploaded ${uploadedPhotos.length} of ${totalAttempted} photos. ${totalInvalidFiles.length} failed.`; + response.message = `Queued ${uploadedPhotos.length} of ${totalAttempted} photos. ${totalInvalidFiles.length} failed.`; } - + // Invalidate download zip cache after successful upload or replacement if (uploadedPhotos.length > 0 || replacedPhotos.length > 0) { downloadZipService.invalidate(parseInt(eventId)); } - res.json(response); + // 202 Accepted — files stored, processing happens in background. + res.status(202).json(response); } catch (error) { console.error('Error uploading photos:', error); // Temp directory cleanup is handled by the response finish/close @@ -597,6 +441,176 @@ router.post('/:eventId/upload', adminAuth, requirePermission('photos.upload'), r } }); +// Helper — load the upload group + verify the requesting admin owns the +// underlying event. Returns { event, photos } or sends a 4xx response. +async function loadUploadGroup(req, res) { + const { upload_id: uploadId } = req.params; + if (!uploadId || typeof uploadId !== 'string' || uploadId.length > 64) { + res.status(400).json({ error: 'Invalid upload_id' }); + return null; + } + + const photos = await db('photos').where({ upload_id: uploadId }); + if (photos.length === 0) { + res.status(404).json({ error: 'Upload group not found' }); + return null; + } + + const eventId = photos[0].event_id; + let eventQuery = db('events').where('id', eventId); + if (req.admin.roleName === 'editor') { + eventQuery = eventQuery.where('created_by', req.admin.id); + } + const event = await eventQuery.first(); + if (!event) { + res.status(404).json({ error: 'Event not found' }); + return null; + } + return { event, photos, uploadId }; +} + +function summariseUpload(photos) { + const summary = { + total: photos.length, + pending: 0, + processing: 0, + complete: 0, + failed: 0, + photos: photos.map((p) => ({ + id: p.id, + filename: p.filename, + original_filename: p.original_filename, + status: p.processing_status, + error: p.processing_error || null, + })), + }; + for (const p of photos) { + summary[p.processing_status] = (summary[p.processing_status] || 0) + 1; + } + return summary; +} + +// JSON snapshot of upload status — frontends poll this every 1.5s while +// any photo in the group is still pending or processing. +router.get( + '/uploads/:upload_id/status', + adminAuth, + requirePermission('photos.view'), + async (req, res) => { + try { + const group = await loadUploadGroup(req, res); + if (!group) return; + res.json({ + upload_id: group.uploadId, + event_id: group.event.id, + ...summariseUpload(group.photos), + }); + } catch (error) { + console.error('Error reading upload status:', error); + res.status(500).json({ error: 'Failed to read upload status' }); + } + } +); + +// Server-Sent Events stream for upload progress. Optional upgrade over +// the polling endpoint above. Streams the current snapshot on connect, +// then re-emits whenever the snapshot changes (debounced) until all +// photos in the group reach a terminal state. +router.get( + '/uploads/:upload_id/stream', + adminAuth, + requirePermission('photos.view'), + async (req, res) => { + const group = await loadUploadGroup(req, res); + if (!group) return; + + res.set({ + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache, no-transform', + 'Connection': 'keep-alive', + 'X-Accel-Buffering': 'no', + }); + res.flushHeaders(); + + let lastJson = ''; + let closed = false; + let timer = null; + + const send = async () => { + if (closed) return; + try { + const photos = await db('photos').where({ upload_id: group.uploadId }); + const summary = summariseUpload(photos); + const payload = JSON.stringify({ + upload_id: group.uploadId, + event_id: group.event.id, + ...summary, + }); + if (payload !== lastJson) { + lastJson = payload; + res.write(`data: ${payload}\n\n`); + } + // Stop streaming once everything has reached a terminal state. + if (summary.pending === 0 && summary.processing === 0) { + closed = true; + clearInterval(timer); + res.end(); + return; + } + } catch (e) { + console.error('Upload stream poll error:', e); + } + }; + + await send(); + timer = setInterval(send, 1500); + + req.on('close', () => { + closed = true; + if (timer) clearInterval(timer); + }); + } +); + +// Retry a failed photo — flip back to 'pending' so the worker picks it +// up again. Used by the admin grid's "Retry" button when a previous run +// hit a transient sharp/ffmpeg error. +router.post( + '/photos/:photoId/retry', + adminAuth, + requirePermission('photos.edit'), + async (req, res) => { + try { + const photo = await db('photos').where({ id: req.params.photoId }).first(); + if (!photo) return res.status(404).json({ error: 'Photo not found' }); + + // Editor role: only allow retry on photos in events they own. + if (req.admin.roleName === 'editor') { + const event = await db('events') + .where({ id: photo.event_id, created_by: req.admin.id }) + .first(); + if (!event) return res.status(404).json({ error: 'Photo not found' }); + } + + if (photo.processing_status !== 'failed') { + return res.status(409).json({ + error: `Photo is in '${photo.processing_status}' state and cannot be retried`, + }); + } + + await db('photos').where({ id: photo.id }).update({ + processing_status: 'pending', + processing_error: null, + processing_started_at: null, + }); + res.json({ id: photo.id, status: 'pending' }); + } catch (error) { + console.error('Error retrying photo processing:', error); + res.status(500).json({ error: 'Failed to retry photo processing' }); + } + } +); + // Delete a photo router.delete('/:eventId/photos/:photoId', adminAuth, requirePermission('photos.delete'), requireEventOwnership, async (req, res) => { try { @@ -1055,15 +1069,25 @@ router.get('/:eventId/photos', adminAuth, requirePermission('photos.view'), requ router.get('/:eventId/photo/:photoId', adminAuth, requirePermission('photos.view'), requireEventOwnership, async (req, res) => { try { const { eventId, photoId } = req.params; - + const photo = await db('photos') .where({ id: photoId, event_id: eventId }) .first(); - + if (!photo) { return res.status(404).json({ error: 'Photo not found' }); } - + + // Photos still in async processing don't have all metadata in the DB + // yet; serving the original is fine, but downstream consumers (admin + // grid lightbox) read width/height which won't be set until processing + // completes. We let the original through here — the file is on disk — + // but tell the caller it's not done yet via a header so they can + // poll /uploads/:upload_id/status if they care. + if (photo.processing_status && photo.processing_status !== 'complete') { + res.setHeader('X-PicPeak-Photo-Status', photo.processing_status); + } + const { resolvePhotoFilePath, resolvePhotoStorageKey } = require('../services/photoResolver'); const event = await db('events').where('id', eventId).first(); const storageKey = resolvePhotoStorageKey(event, photo); @@ -1106,12 +1130,28 @@ router.get('/:eventId/thumbnail/:photoId', adminAuth, requirePermission('photos. const photo = await db('photos') .where({ id: photoId, event_id: eventId }) .first(); - + if (!photo) { console.error(`Photo not found: ${photoId}, event ${eventId}`); return res.status(404).json({ error: 'Photo not found' }); } - + + // Async processing is still working on this one — no thumbnail yet. + // Return 503 with Retry-After so the admin grid (which auto-refreshes + // every 2s while any photo is non-complete) keeps the placeholder + // until the worker catches up. + if (photo.processing_status === 'pending' || photo.processing_status === 'processing') { + res.setHeader('Retry-After', '2'); + return res.status(503).json({ error: 'Thumbnail not ready', status: photo.processing_status }); + } + if (photo.processing_status === 'failed') { + return res.status(422).json({ + error: 'Photo processing failed', + status: 'failed', + details: photo.processing_error || null, + }); + } + // Ensure thumbnail exists and is valid, regenerate if needed const thumbnailPath = await ensureThumbnail(photo); diff --git a/backend/src/routes/gallery.js b/backend/src/routes/gallery.js index 5e48d9dd..31cea158 100644 --- a/backend/src/routes/gallery.js +++ b/backend/src/routes/gallery.js @@ -212,6 +212,15 @@ router.get('/:slug/photos', verifyGalleryAccess, async (req, res) => { const isClient = req.accessLevel === 'client'; let photosQuery = db('photos') .where('photos.event_id', req.event.id) + // Guests/clients never see photos still being processed by the + // background worker — the original is on disk but the thumbnail + // / dimensions / EXIF haven't landed yet. Photos with a NULL + // processing_status are pre-async-migration rows and are treated + // as complete (the migration's column default is 'complete' so + // this is just defensive against partial migration states). + .where(function() { + this.where('photos.processing_status', 'complete').orWhereNull('photos.processing_status'); + }) .select('photos.*'); // Guests only see visible photos; clients see all @@ -1380,18 +1389,31 @@ router.post('/:eventId/upload', verifyGalleryAccess, async (req, res) => { if (!req.files || req.files.length === 0) { return res.status(400).json({ error: 'No files uploaded' }); } - - const { processUploadedPhotos } = require('../services/photoProcessor'); - const categoryId = req.body.category_id || req.event.upload_category_id || null; - + + const { queueFilesForProcessing } = require('../services/photoProcessor'); + const rawCategory = req.body.category_id || req.event.upload_category_id || null; + const numericCategoryId = (() => { + if (rawCategory === null || rawCategory === undefined) return null; + const n = parseInt(rawCategory, 10); + return Number.isFinite(n) ? n : null; + })(); + try { - // Process uploaded photos - const results = await processUploadedPhotos(req.files, eventId, 'user', categoryId); - - res.json({ - message: 'Photos uploaded successfully', - count: results.length, - photos: results + // Queue files as 'pending' — the background worker will process + // thumbnails / EXIF / dimensions off the request thread (#357). + const result = await queueFilesForProcessing(req.files, { + eventId, + photoType: 'individual', + categoryId: numericCategoryId, + }); + + res.status(202).json({ + message: 'Photos queued for processing', + upload_id: result.uploadId, + count: result.photos.length, + photo_ids: result.photos.map((p) => p.id), + photos: result.photos, + errors: result.errors.length > 0 ? result.errors : undefined, }); } catch (processError) { console.error('Photo processing error:', processError); diff --git a/backend/src/services/backgroundProcessor.js b/backend/src/services/backgroundProcessor.js new file mode 100644 index 00000000..32148405 --- /dev/null +++ b/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 }; diff --git a/backend/src/services/photoProcessor.js b/backend/src/services/photoProcessor.js index 5d793600..9b64d2e9 100644 --- a/backend/src/services/photoProcessor.js +++ b/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 };