paul/picpeak
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

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

Browse Source 
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.
This commit is contained in:
Paul Nothaft
2026-05-02 00:25:51 +02:00
parent 86dfcc4f11
commit 851744c3c4
6 changed files with 815 additions and 277 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/core/085_async_photo_processing.js
+72
View File
@@ -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'));
}
};
backend/server.js
+6
View File
@@ -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');
@@ -664,6 +665,11 @@ async function startServer() {
// 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'}`);
backend/src/routes/adminPhotos.js
+273 -233
View File
@@ -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,40 +283,37 @@ 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();
try {
// Get initial counter for this batch based on photo type
const existingCount = await trx('photos')
// 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 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
let counter = (parseInt(existingCount.count) || 0) + 1;
const storage = getStorage();
for (const file of filesToUpload) {
try {
// Verify file is complete before processing
const tempStats = await fs.stat(tempPath);
const tempStats = await fs.stat(file.path);
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,
@@ -324,221 +321,59 @@ router.post('/:eventId/upload', adminAuth, requirePermission('photos.upload'), r
counter,
extension
);
counter += 1;
// 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
// 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,
});
} catch (error) {
console.error(`Error preparing file ${file.originalname}:`, error);
errors.push({ filename: file.originalname, error: error.message });
}
}
await fs.unlink(file.path).catch(() => {});
// 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')}`
// 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'}`
);
const result = await processUploadedVideo(operation.tempPath, videoThumbnailKey);
thumbnailPath = result.thumbnailKey;
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);
// Update the database with thumbnail path and image dimensions
if (photoId) {
const updateData = {};
if (thumbnailPath) updateData.thumbnail_path = thumbnailPath;
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);
}
// 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];
// 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
id: photoId,
filename: newFilename,
size: tempStats.size,
category_id: parsedCategoryId,
});
} 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();
}
} 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);
} 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,13 +417,13 @@ 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
@@ -588,7 +431,8 @@ router.post('/:eventId/upload', adminAuth, requirePermission('photos.upload'), r
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 {
@@ -1064,6 +1078,16 @@ router.get('/:eventId/photo/:photoId', adminAuth, requirePermission('photos.view
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);
@@ -1112,6 +1136,22 @@ router.get('/:eventId/thumbnail/:photoId', adminAuth, requirePermission('photos.
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);
backend/src/routes/gallery.js
+30 -8
View File
@@ -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
@@ -1381,17 +1390,30 @@ router.post('/:eventId/upload', verifyGalleryAccess, async (req, res) => {
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);
// 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.json({
message: 'Photos uploaded successfully',
count: results.length,
photos: results
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);
backend/src/services/backgroundProcessor.js
+176
View File
@@ -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
+224 -2
View File
@@ -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
};