feat: native S3 storage backend (#328) + presigned download follow-up
Lets PicPeak write photos, thumbnails, hero images, watermarks, and archive zips to any S3-compatible bucket (AWS S3, MinIO, Cloudflare R2, Backblaze B2, Wasabi, DigitalOcean Spaces) instead of the local filesystem. Selected via STORAGE_BACKEND=local|s3. Architecture - backend/src/services/storage/StorageBackend.js — abstract interface (put/get/exists/stat/delete/list/copy/rename/signedUrl/putFromFile/ getToFile) — typedef-only, documents the contract. - LocalFsStorage.js — wraps fs with atomic-write-via-tmp-rename, path traversal protection, list-as-walker. - S3StorageBackend.js — thin wrapper around the existing S3StorageAdapter (used by backupService) mapping it onto the canonical interface; supports optional STORAGE_S3_PREFIX namespace. - index.js — factory selected by STORAGE_BACKEND with startup ping (HEADs sentinel key on S3, fs.stat on local) so misconfig fails fast before the first request. Consumer refactors (~12 services + routes), each parametrized over the abstraction: - imageProcessor / videoProcessor — pipe Sharp/ffmpeg output through storage.put; expose withLocalCopy() helper for S3-mode regeneration paths that need a local file for sharp/ffmpeg. - archiveService / downloadZipService — finalize zip in tmp dir, then storage.putFromFile. Atomic-rename pattern preserved on local; S3 emulates via copy + delete (worker prunes orphaned .tmp.* on startup). - photoProcessor / photoReplacementService / adminPhotos upload+delete / routes/v1/events.js POST /events/:id/photos / routes/events.js — every upload path now goes storage.putFromFile(temp) → unlink temp. - gallery.js bulk-download (cached + on-the-fly + selected) — managed photos via storage.get, external-mode unchanged. - protectedImages / secureImages / photoResolver — read via storage.get; resolvePhotoStorageKey returns the canonical key. - watermarkService / watermarkGeneratorService — persistent watermarks via storage.put. - fileWatcher — bails out with a clear log warning when STORAGE_BACKEND=s3 (chokidar can't watch S3); auto-import lands via the S3 prefix walker introduced in the follow-up commit. - expirationChecker — small touch (event.expired webhook fire from #327 shipping in the next commit). Migration tooling - backend/scripts/migrate-storage.js — one-shot --dry-run capable script that walks photos.path, thumbnail_path, hero_path, watermark_path and events.archive_path/download_zip_path; streams local → S3; sha256 size-match skip for idempotent re-run; failures CSV. Presigned-URL "Download All" (#328 follow-up shipped in this commit) - routes/gallery.js — when STORAGE_BACKEND=s3 + event.allow_presigned_download + downloads enabled + watermark NOT enabled, /download-all returns a 302 redirect to a 5-minute presigned S3 URL. Per-event opt-in surface ships in the next commit's UI. Tests - backend/__tests__/integration/storageBackend.test.js — parametrized contract suite running against BOTH LocalFs AND MinIO (18 tests, both backends — 36 cases total). - backend/__tests__/integration/imageProcessor.storage.test.js — same parametrized pattern for the image processor (10 tests × 2 backends). - backend/__tests__/integration/backup-s3.test.js — bootstrap fix: drop the redundant initDb() (001_init handles it) and remove schema-drift in configureS3Backup (app_settings has no created_at anymore and the unique constraint is on setting_key alone, not composite). 0/12 → 7/12 (5 remaining are unrelated assertion drift). - backend/src/services/photoResolver.js — mixed-source events (reference mode with managed-uploaded photos) now fall back to managed when external_relpath is missing instead of throwing. - tests/e2e/s3-storage-roundtrip.spec.ts — Playwright spec that auto-skips against local backend; full upload → serve → delete round-trip when run against an S3-mode backend. Server wiring (server.js) - initStorage() called after database init, before rate limiters. - This commit's diff also includes the webhook delivery worker startup and the S3 auto-importer startup. Those features ship in the next two commits — co-located here for one bisectable diff per file. Docs + ops - README §"Storage Backends" — capability matrix, switching playbook, IAM policy snippet, MinIO/R2/B2 examples. - README §"Webhooks" — also added here (full diff bundled). - .env.example — STORAGE_BACKEND + STORAGE_S3_* + STORAGE_AUTO_IMPORT documented; WEBHOOK_* added in the same diff. - .gitignore — re-anchor the existing `storage/` rule to `/storage/` so backend/src/services/storage/ (the new abstraction code) is trackable. The runtime ./storage/ data dir stays ignored. Out of scope for v1 (per the issue): presigned URLs for individual photo display (always streamed for protection middleware), CDN integration, hybrid hot/cold tiers, S3 → local migration, multi-bucket per-event.
This commit is contained in:
Paul Nothaft
@@ -1,57 +1,47 @@
|
||||
const archiver = require('archiver');
|
||||
const fs = require('fs').promises;
|
||||
const fs = require('fs');
|
||||
const fsp = require('fs').promises;
|
||||
const path = require('path');
|
||||
const os = require('os');
|
||||
const crypto = require('crypto');
|
||||
const { db } = require('../database/db');
|
||||
const { queueEmail } = require('./emailProcessor');
|
||||
const logger = require('../utils/logger');
|
||||
const feedbackService = require('./feedbackService');
|
||||
|
||||
const getStoragePath = () => process.env.STORAGE_PATH || path.join(__dirname, '../../../storage');
|
||||
const ACTIVE_PATH = () => path.join(getStoragePath(), 'events/active');
|
||||
const ARCHIVE_PATH = () => path.join(getStoragePath(), 'events/archived');
|
||||
const { getStorage } = require('./storage');
|
||||
|
||||
async function archiveEvent(event) {
|
||||
const storage = getStorage();
|
||||
const archiveName = `${event.slug}.zip`;
|
||||
const archiveRelKey = path.posix.join('events/archived', archiveName);
|
||||
const eventPrefix = path.posix.join('events/active', event.slug);
|
||||
|
||||
const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'picpeak-archive-'));
|
||||
const tmpArchive = path.join(tmpDir, `${crypto.randomBytes(4).toString('hex')}-${archiveName}`);
|
||||
|
||||
try {
|
||||
const eventPath = path.join(ACTIVE_PATH(), event.slug);
|
||||
const archiveName = `${event.slug}.zip`;
|
||||
const archivePath = path.join(ARCHIVE_PATH(), archiveName);
|
||||
|
||||
// Ensure archive directory exists
|
||||
await fs.mkdir(ARCHIVE_PATH(), { recursive: true });
|
||||
|
||||
// Create archive
|
||||
const output = require('fs').createWriteStream(archivePath);
|
||||
const archive = archiver('zip', {
|
||||
zlib: { level: 9 } // Maximum compression
|
||||
});
|
||||
|
||||
archive.on('error', (err) => {
|
||||
throw err;
|
||||
});
|
||||
|
||||
// Export feedback data before archiving
|
||||
// Collect feedback data first so it can be included as in-memory entries.
|
||||
const feedbackEntries = [];
|
||||
const feedbackSettings = await feedbackService.getEventFeedbackSettings(event.id);
|
||||
if (feedbackSettings.feedback_enabled) {
|
||||
try {
|
||||
logger.info(`Exporting feedback data for event ${event.slug}`);
|
||||
const feedbackData = await feedbackService.exportEventFeedback(event.id);
|
||||
|
||||
|
||||
if (feedbackData && feedbackData.length > 0) {
|
||||
// Create feedback JSON file
|
||||
const feedbackJson = JSON.stringify(feedbackData, null, 2);
|
||||
const feedbackJsonPath = path.join(eventPath, 'feedback_data.json');
|
||||
await fs.writeFile(feedbackJsonPath, feedbackJson, 'utf8');
|
||||
|
||||
// Create feedback CSV file
|
||||
const feedbackCsv = convertToCSV(feedbackData);
|
||||
const feedbackCsvPath = path.join(eventPath, 'feedback_data.csv');
|
||||
await fs.writeFile(feedbackCsvPath, feedbackCsv, 'utf8');
|
||||
|
||||
// Create feedback summary
|
||||
feedbackEntries.push({
|
||||
name: 'feedback_data.json',
|
||||
buffer: Buffer.from(JSON.stringify(feedbackData, null, 2), 'utf8'),
|
||||
});
|
||||
feedbackEntries.push({
|
||||
name: 'feedback_data.csv',
|
||||
buffer: Buffer.from(convertToCSV(feedbackData), 'utf8'),
|
||||
});
|
||||
const summary = await feedbackService.getEventFeedbackSummary(event.id);
|
||||
const summaryPath = path.join(eventPath, 'feedback_summary.json');
|
||||
await fs.writeFile(summaryPath, JSON.stringify(summary, null, 2), 'utf8');
|
||||
|
||||
feedbackEntries.push({
|
||||
name: 'feedback_summary.json',
|
||||
buffer: Buffer.from(JSON.stringify(summary, null, 2), 'utf8'),
|
||||
});
|
||||
logger.info(`Feedback data exported: ${feedbackData.length} entries`);
|
||||
}
|
||||
} catch (error) {
|
||||
@@ -59,64 +49,110 @@ async function archiveEvent(event) {
|
||||
// Continue with archiving even if feedback export fails
|
||||
}
|
||||
}
|
||||
|
||||
output.on('close', async () => {
|
||||
try {
|
||||
logger.info(`Archive created: ${archiveName} (${archive.pointer()} bytes)`);
|
||||
|
||||
// Update database
|
||||
await db('events').where('id', event.id).update({
|
||||
is_archived: true,
|
||||
archive_path: path.relative(getStoragePath(), archivePath),
|
||||
archived_at: new Date()
|
||||
});
|
||||
// Stream every photo (and any other content under events/active/{slug}/) into
|
||||
// the zip directly from the storage backend.
|
||||
const photoEntries = await storage.list(eventPrefix);
|
||||
|
||||
// Delete original files
|
||||
await fs.rm(eventPath, { recursive: true });
|
||||
let totalBytes = 0;
|
||||
await new Promise((resolve, reject) => {
|
||||
const output = fs.createWriteStream(tmpArchive);
|
||||
const archive = archiver('zip', { zlib: { level: 9 } });
|
||||
|
||||
// Delete thumbnails
|
||||
const photos = await db('photos').where('event_id', event.id);
|
||||
for (const photo of photos) {
|
||||
if (photo.thumbnail_path) {
|
||||
const thumbPath = path.join(getStoragePath(), photo.thumbnail_path);
|
||||
await fs.unlink(thumbPath).catch(() => {}); // Ignore if already deleted
|
||||
}
|
||||
output.on('close', () => {
|
||||
totalBytes = archive.pointer();
|
||||
resolve();
|
||||
});
|
||||
archive.on('error', reject);
|
||||
archive.pipe(output);
|
||||
|
||||
const append = async () => {
|
||||
for (const entry of photoEntries) {
|
||||
const nameInZip = entry.key.startsWith(`${eventPrefix}/`)
|
||||
? entry.key.slice(eventPrefix.length + 1)
|
||||
: entry.key;
|
||||
const stream = await storage.get(entry.key);
|
||||
archive.append(stream, { name: nameInZip });
|
||||
}
|
||||
|
||||
// Queue completion email — admin_email is nullable on events (migration 073);
|
||||
// skip queueing rather than violating email_queue.recipient_email NOT NULL.
|
||||
if (event.admin_email) {
|
||||
await queueEmail(event.id, event.admin_email, 'archive_complete', {
|
||||
event_name: event.event_name,
|
||||
archive_size: (archive.pointer() / 1024 / 1024).toFixed(2) + ' MB'
|
||||
});
|
||||
} else {
|
||||
logger.info(`Skipping archive_complete email for event ${event.slug}: no admin_email set`);
|
||||
for (const f of feedbackEntries) {
|
||||
archive.append(f.buffer, { name: f.name });
|
||||
}
|
||||
} catch (err) {
|
||||
// Never let the close handler reject — it runs detached from the caller,
|
||||
// and an unhandled rejection here crashes the backend process.
|
||||
logger.error(`Post-archive cleanup failed for event ${event.slug}:`, err);
|
||||
}
|
||||
archive.finalize();
|
||||
};
|
||||
|
||||
append().catch(reject);
|
||||
});
|
||||
|
||||
archive.pipe(output);
|
||||
archive.directory(eventPath, false);
|
||||
await archive.finalize();
|
||||
|
||||
|
||||
// Upload the finalized zip to the storage backend.
|
||||
await storage.putFromFile(archiveRelKey, tmpArchive, { contentType: 'application/zip' });
|
||||
|
||||
logger.info(`Archive created: ${archiveName} (${totalBytes} bytes)`);
|
||||
|
||||
// Update DB BEFORE deleting originals so a crash mid-cleanup leaves the
|
||||
// archive accessible rather than orphaning the photos.
|
||||
await db('events').where('id', event.id).update({
|
||||
is_archived: true,
|
||||
archive_path: archiveRelKey,
|
||||
archived_at: new Date(),
|
||||
});
|
||||
|
||||
// Fire event.archived webhook (#327). Receivers infer per-photo loss
|
||||
// from this event — we deliberately do NOT fire photo.deleted for each
|
||||
// archived photo to avoid flooding subscribers on bulk archives.
|
||||
try {
|
||||
const webhookService = require('./webhookService');
|
||||
await webhookService.fire('event.archived', {
|
||||
event: { id: event.id, slug: event.slug, event_name: event.event_name, archive_path: archiveRelKey },
|
||||
});
|
||||
} catch (e) { /* non-fatal */ }
|
||||
|
||||
// Delete the originals from storage.
|
||||
for (const entry of photoEntries) {
|
||||
await storage.delete(entry.key).catch((err) =>
|
||||
logger.warn(`Failed to delete archived original ${entry.key}: ${err.message}`)
|
||||
);
|
||||
}
|
||||
|
||||
// Delete thumbnails for this event's photos.
|
||||
const photos = await db('photos').where('event_id', event.id);
|
||||
for (const photo of photos) {
|
||||
if (photo.thumbnail_path) {
|
||||
await storage.delete(photo.thumbnail_path).catch(() => {});
|
||||
}
|
||||
if (photo.hero_path) {
|
||||
await storage.delete(photo.hero_path).catch(() => {});
|
||||
}
|
||||
// Best effort: remove watermarked variants too if a refactor added them.
|
||||
if (photo.watermark_path) {
|
||||
await storage.delete(photo.watermark_path).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
// Queue completion email — admin_email is nullable on events (migration 073);
|
||||
// skip queueing rather than violating email_queue.recipient_email NOT NULL.
|
||||
if (event.admin_email) {
|
||||
await queueEmail(event.id, event.admin_email, 'archive_complete', {
|
||||
event_name: event.event_name,
|
||||
archive_size: (totalBytes / 1024 / 1024).toFixed(2) + ' MB',
|
||||
});
|
||||
} else {
|
||||
logger.info(`Skipping archive_complete email for event ${event.slug}: no admin_email set`);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error(`Error archiving event ${event.slug}:`, error);
|
||||
throw error;
|
||||
} finally {
|
||||
await fsp.rm(tmpDir, { recursive: true, force: true }).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
// Helper function to convert JSON to CSV
|
||||
function convertToCSV(data) {
|
||||
if (!data || data.length === 0) return '';
|
||||
|
||||
|
||||
const headers = Object.keys(data[0]);
|
||||
const csvHeaders = headers.join(',');
|
||||
|
||||
|
||||
const csvRows = data.map(row => {
|
||||
return headers.map(header => {
|
||||
const value = row[header];
|
||||
@@ -127,7 +163,7 @@ function convertToCSV(data) {
|
||||
return value || '';
|
||||
}).join(',');
|
||||
});
|
||||
|
||||
|
||||
return [csvHeaders, ...csvRows].join('\n');
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user