Refine workspaces and event publishing; harden uploads and email delivery

apps/worker/package.json

@@ -11,6 +11,7 @@
     "lint": "tsc --noEmit"
   },
   "dependencies": {
+    "@album/email": "workspace:*",
     "@album/database": "workspace:*",
     "@album/storage": "workspace:*",
     "drizzle-orm": "^0.45.2",

apps/worker/src/index.ts

@@ -1,8 +1,9 @@
 import { sql } from "drizzle-orm";
-import { eq } from "drizzle-orm";
+import { and, eq } from "drizzle-orm";
 import sharp from "sharp";
+import { processEmailDelivery } from "@album/email/queue";
 import convert from "heic-convert";
-import { getDb, photoJobs, photos } from "@album/database";
+import { eventBanners, getDb, photoJobs, photos } from "@album/database";
 import {
   displayObjectKey,
   getObjectBuffer,
@@ -113,7 +114,6 @@ async function processJob(job: ClaimedJob) {
       .update(photos)
       .set({
         processingStatus: "ready",
-        visibility: photo.visibility === "pending" ? "pending" : photo.visibility,
         displayKey,
         thumbKey,
         width: metadata.width ?? null,
@@ -150,6 +150,7 @@ async function workerLoop(workerId: number) {
     const started = Date.now();
     const job = await claimJob();
     if (!job) {
+      if (await processBanner()) continue;
       await Bun.sleep(POLL_MS);
       continue;
     }
@@ -167,10 +168,46 @@ async function workerLoop(workerId: number) {
   }
 }
 
+async function processBanner() {
+  const rows = await getDb().execute(sql`
+    UPDATE event_banners SET status = 'processing', updated_at = now()
+    WHERE id = (
+      SELECT id FROM event_banners
+      WHERE status = 'pending' OR (status = 'processing' AND updated_at < now() - interval '10 minutes')
+      ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1
+    ) RETURNING id, event_id, original_key, content_type
+  `);
+  const banner = (rows as unknown as { id: string; event_id: string; original_key: string; content_type: string }[])[0];
+  if (!banner) return false;
+  const scope = and(eq(eventBanners.id, banner.id), eq(eventBanners.eventId, banner.event_id));
+  try {
+    const original = await getObjectBuffer(banner.original_key);
+    const image = await decodeImage(original, banner.content_type);
+    const display = await image.resize({ width: 2400, height: 2400, fit: "inside", withoutEnlargement: true })
+      .jpeg({ quality: 82, mozjpeg: true }).toBuffer();
+    const displayKey = `events/${banner.event_id}/banners/${banner.id}/display.jpg`;
+    await putObject({ key: displayKey, body: display, contentType: "image/jpeg" });
+    await getDb().update(eventBanners).set({ status: "ready", displayKey, updatedAt: new Date() }).where(scope);
+    console.info(`Banner processed for event ${banner.event_id}`);
+  } catch {
+    await getDb().update(eventBanners).set({ status: "failed", updatedAt: new Date() }).where(scope);
+    console.error(`Banner processing failed for event ${banner.event_id}`);
+  }
+  return true;
+}
+
 console.info(
   `Manyangles worker listening for photo jobs (concurrency ${CONCURRENCY}, min interval ${MIN_INTERVAL_MS}ms)`,
 );
 
-await Promise.all(
-  Array.from({ length: CONCURRENCY }, (_, index) => workerLoop(index + 1)),
-);
+async function emailLoop() {
+  while (true) {
+    try { await processEmailDelivery(); } catch { console.error("Email queue check failed"); }
+    await Bun.sleep(1000);
+  }
+}
+
+await Promise.all([
+  emailLoop(),
+  ...Array.from({ length: CONCURRENCY }, (_, index) => workerLoop(index + 1)),
+]);