Prepare production and Coolify deployment with original exports and datetime pickers

apps/worker/src/exports.integration.test.ts

@@ -0,0 +1,42 @@
+import {test,expect} from "bun:test";
+import {unzipSync} from "fflate";
+import {and,eq} from "drizzle-orm";
+import {events,getDb,photos,photoExports,user,guests,submissions} from "@album/database";
+import {getObjectBuffer,deleteObject,putObject,headObject} from "@album/storage";
+import {processPhotoExport} from "./exports";
+
+test.skipIf(process.env.EXPORT_INTEGRATION!=="1")("original ZIP bytes, approved filtering, duplicate requests and expiry",async()=>{
+  if (!["localhost","127.0.0.1"].includes(new URL(process.env.DATABASE_URL!).hostname) || !["localhost","127.0.0.1"].includes(new URL(process.env.S3_ENDPOINT!).hostname)) throw new Error("Local storage/database required");
+  const db=getDb();
+  const [demo]=await db.select().from(events).where(eq(events.slug,"demo"));
+  const [owner]=await db.select({id:user.id}).from(user).limit(1);
+  const [event]=await db.insert(events).values({groupId:demo!.groupId,title:"Export test",slug:`export-${crypto.randomUUID()}`}).returning();
+  const [guest]=await db.insert(guests).values({eventId:event!.id,tokenHash:crypto.randomUUID()}).returning();
+  const [submission]=await db.insert(submissions).values({eventId:event!.id,guestId:guest!.id}).returning();
+  const bytes=Buffer.from("original bytes preserved exactly");
+  const sourceKey=`events/${event!.id}/export-test/original`;
+  const jobIds:string[]=[];
+  try {
+    await putObject({key:sourceKey,body:bytes,contentType:"image/jpeg"});
+    const [photo]=await db.insert(photos).values({eventId:event!.id,submissionId:submission!.id,originalKey:sourceKey,contentType:"image/jpeg",byteSize:bytes.length,processingStatus:"ready",visibility:"public"}).returning();
+    await db.insert(photos).values({eventId:event!.id,submissionId:submission!.id,originalKey:sourceKey,contentType:"image/jpeg",byteSize:bytes.length,processingStatus:"ready",visibility:"private"});
+    for(const filter of ["approved","all"]) {
+      const [job]=await db.insert(photoExports).values({eventId:event!.id,requestedBy:owner!.id,filter,expiresAt:new Date(Date.now()+86400000)}).returning();
+      jobIds.push(job!.id);
+      expect(await db.insert(photoExports).values({eventId:event!.id,requestedBy:owner!.id,filter,expiresAt:new Date(Date.now()+86400000)}).onConflictDoNothing().returning()).toHaveLength(0);
+      await processPhotoExport();
+      const [finished]=await db.select().from(photoExports).where(and(eq(photoExports.id,job!.id),eq(photoExports.eventId,event!.id)));
+      expect(finished?.status).toBe("ready");
+      const zip=unzipSync(await getObjectBuffer(`exports/${event!.id}/${job!.id}.zip`));
+      expect(Object.keys(zip)).toHaveLength(filter==="approved"?1:2);
+      expect(Buffer.from(zip[`${photo!.id}.jpg`]!)).toEqual(bytes);
+      await db.update(photoExports).set({expiresAt:new Date(0)}).where(and(eq(photoExports.id,job!.id),eq(photoExports.eventId,event!.id)));
+      await processPhotoExport();
+      expect(await headObject(`exports/${event!.id}/${job!.id}.zip`)).toBeNull();
+    }
+  } finally {
+    for(const id of jobIds) await deleteObject(`exports/${event!.id}/${id}.zip`);
+    await deleteObject(sourceKey);
+    await db.delete(events).where(eq(events.id,event!.id));
+  }
+},30000);

apps/worker/src/exports.ts

@@ -0,0 +1,66 @@
+import { ZipArchive } from "archiver";
+import { Upload } from "@aws-sdk/lib-storage";
+import { once } from "node:events";
+import { PassThrough } from "node:stream";
+import { and, eq, sql } from "drizzle-orm";
+import { getDb, photoExports, photos } from "@album/database";
+import { deleteObject, getObjectBuffer, getSigningClient, storageConfig } from "@album/storage";
+
+export async function processPhotoExport() {
+  const db = getDb();
+  // Interrupted jobs are failed, never reclaimed while an old worker may write.
+  await db.execute(sql`UPDATE photo_exports SET status = 'failed', updated_at = now()
+    WHERE status = 'processing' AND updated_at < now() - interval '1 hour'`);
+  const expired = await db.execute<{id:string;event_id:string}>(sql`SELECT id,event_id FROM photo_exports WHERE expires_at < now() AND status <> 'expired' LIMIT 1`);
+  if (expired[0]) {
+    const item = expired[0];
+    await deleteObject(`exports/${item.event_id}/${item.id}.zip`);
+    await db.update(photoExports).set({status:"expired",updatedAt:new Date()}).where(and(eq(photoExports.id,item.id),eq(photoExports.eventId,item.event_id)));
+  }
+  const rows = await db.execute<{id:string;event_id:string;filter:string}>(sql`UPDATE photo_exports SET status = 'processing', updated_at = now()
+    WHERE id = (SELECT id FROM photo_exports WHERE status = 'pending' AND expires_at > now() ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1)
+    RETURNING id,event_id,filter`);
+  const job = rows[0];
+  if (!job) return false;
+  const scope = and(eq(photoExports.id,job.id),eq(photoExports.eventId,job.event_id),eq(photoExports.status,"processing"));
+  const key = `exports/${job.event_id}/${job.id}.zip`;
+  const archive = new ZipArchive({ store:true, forceZip64:true });
+  const body = new PassThrough();
+  archive.pipe(body);
+  const upload = new Upload({ client:getSigningClient(), params:{Bucket:storageConfig().bucket,Key:key,Body:body,ContentType:"application/zip",ContentDisposition:'attachment; filename="manyangles-originals.zip"'}, queueSize:2, partSize:8*1024*1024 });
+  const uploaded = upload.done();
+  // Attach immediately: a storage failure must not become an unhandled rejection.
+  void uploaded.catch(() => archive.destroy(new Error("Export storage failed")));
+  archive.on("error", error => body.destroy(error));
+  body.on("error", () => {});
+  try {
+    const items = await db.select({id:photos.id,key:photos.originalKey,type:photos.contentType,bytes:photos.byteSize}).from(photos)
+      .where(and(eq(photos.eventId,job.event_id),eq(photos.processingStatus,"ready"),job.filter === "approved" ? eq(photos.visibility,"public") : undefined)).orderBy(photos.id).limit(10001);
+    if (!items.length || items.length > 10000 || items.reduce((sum,p)=>sum+p.bytes,0)>20*1024**3) throw new Error("Export size limit");
+    await db.update(photoExports).set({total:items.length,updatedAt:new Date()}).where(scope);
+    let processed = 0;
+    for (const item of items) {
+      const buffer = await getObjectBuffer(item.key);
+      if (archive.destroyed) { await uploaded; throw new Error("Export stream closed"); }
+      const extension = ({"image/jpeg":"jpg","image/png":"png","image/webp":"webp","image/heic":"heic","image/heif":"heif"} as Record<string,string>)[item.type] ?? "bin";
+      const consumed = once(archive,"entry");
+      archive.append(buffer,{name:`${item.id}.${extension}`});
+      await consumed;
+      const changed = await db.update(photoExports).set({processed:++processed,updatedAt:new Date()}).where(scope).returning({id:photoExports.id});
+      if (!changed.length) throw new Error("Export no longer active");
+    }
+    await archive.finalize();
+    await uploaded;
+    const changed = await db.update(photoExports).set({status:"ready",updatedAt:new Date()}).where(scope).returning({id:photoExports.id});
+    if (!changed.length) await deleteObject(key);
+  } catch {
+    archive.abort();
+    body.destroy();
+    await upload.abort().catch(()=>{});
+    await uploaded.catch(()=>{});
+    await deleteObject(key).catch(()=>{});
+    await db.update(photoExports).set({status:"failed",updatedAt:new Date()}).where(scope);
+    console.error(`Export failed for event ${job.event_id}`);
+  }
+  return true;
+}

apps/worker/src/index.ts

@@ -1,6 +1,7 @@
 import { sql } from "drizzle-orm";
 import { and, eq } from "drizzle-orm";
 import sharp from "sharp";
+import { processPhotoExport } from "./exports";
 import { processEmailDelivery } from "@album/email/queue";
 import convert from "heic-convert";
 import { eventBanners, getDb, photoJobs, photos } from "@album/database";
@@ -208,6 +209,7 @@ async function emailLoop() {
 }
 
 await Promise.all([
+  (async () => { while (true) { try { await processPhotoExport(); } catch { console.error("Export queue check failed"); } await Bun.sleep(1000); } })(),
   emailLoop(),
   ...Array.from({ length: CONCURRENCY }, (_, index) => workerLoop(index + 1)),
 ]);