Initial commit of Vellum, an event photo product for guest uploads, host moderation, and original-quality galleries.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-07 19:36:14 -04:00
commit 27e2f196eb
149 changed files with 13847 additions and 0 deletions
@@ -0,0 +1,176 @@
+import { sql } from "drizzle-orm";
+import { eq } from "drizzle-orm";
+import sharp from "sharp";
+import convert from "heic-convert";
+import { getDb, photoJobs, photos } from "@album/database";
+import {
+  displayObjectKey,
+  getObjectBuffer,
+  putObject,
+  thumbObjectKey,
+} from "@album/storage";
+
+const POLL_MS = 1000;
+const CONCURRENCY = Math.max(1, Number(process.env.WORKER_CONCURRENCY ?? 1));
+const MIN_INTERVAL_MS = Math.max(
+  0,
+  Number(process.env.WORKER_MIN_INTERVAL_MS ?? 250),
+);
+
+type ClaimedJob = {
+  id: string;
+  photo_id: string;
+  attempts: number;
+};
+
+async function decodeImage(buffer: Buffer, contentType: string) {
+  const heic = contentType === "image/heic" || contentType === "image/heif";
+  if (heic) {
+    try {
+      await sharp(buffer).metadata();
+      return sharp(buffer).rotate();
+    } catch {
+      const jpeg = Buffer.from(
+        await convert({
+          buffer,
+          format: "JPEG",
+          quality: 0.9,
+        }),
+      );
+      return sharp(jpeg).rotate();
+    }
+  }
+  return sharp(buffer).rotate();
+}
+
+async function claimJob(): Promise<ClaimedJob | null> {
+  const rows = await getDb().execute(sql`
+    UPDATE photo_jobs
+    SET status = 'processing',
+        attempts = attempts + 1,
+        updated_at = now()
+    WHERE id = (
+      SELECT id FROM photo_jobs
+      WHERE status = 'pending'
+      ORDER BY created_at
+      FOR UPDATE SKIP LOCKED
+      LIMIT 1
+    )
+    RETURNING id, photo_id, attempts
+  `);
+  const row = (rows as unknown as ClaimedJob[])[0];
+  return row ?? null;
+}
+
+async function processJob(job: ClaimedJob) {
+  const db = getDb();
+  const [photo] = await db
+    .select()
+    .from(photos)
+    .where(eq(photos.id, job.photo_id))
+    .limit(1);
+  if (!photo) {
+    await db
+      .update(photoJobs)
+      .set({ status: "failed", lastError: "Photo missing", updatedAt: new Date() })
+      .where(eq(photoJobs.id, job.id));
+    return;
+  }
+
+  const original = await getObjectBuffer(photo.originalKey);
+  const image = await decodeImage(original, photo.contentType);
+  const metadata = await image.metadata();
+  const displayKey = displayObjectKey(photo.eventId, photo.id);
+  const thumbKey = thumbObjectKey(photo.eventId, photo.id);
+  const [display, thumb] = await Promise.all([
+    image
+      .clone()
+      .resize({
+        width: 2048,
+        height: 2048,
+        fit: "inside",
+        withoutEnlargement: true,
+      })
+      .jpeg({ quality: 82, mozjpeg: true })
+      .toBuffer(),
+    image
+      .clone()
+      .resize({
+        width: 400,
+        height: 400,
+        fit: "inside",
+        withoutEnlargement: true,
+      })
+      .jpeg({ quality: 75, mozjpeg: true })
+      .toBuffer(),
+  ]);
+  await Promise.all([
+    putObject({ key: displayKey, body: display, contentType: "image/jpeg" }),
+    putObject({ key: thumbKey, body: thumb, contentType: "image/jpeg" }),
+  ]);
+  await db.transaction(async (tx) => {
+    await tx
+      .update(photos)
+      .set({
+        processingStatus: "ready",
+        visibility: photo.visibility === "pending" ? "pending" : photo.visibility,
+        displayKey,
+        thumbKey,
+        width: metadata.width ?? null,
+        height: metadata.height ?? null,
+        updatedAt: new Date(),
+      })
+      .where(eq(photos.id, photo.id));
+    await tx
+      .update(photoJobs)
+      .set({ status: "completed", lastError: null, updatedAt: new Date() })
+      .where(eq(photoJobs.id, job.id));
+  });
+}
+
+async function failJob(job: ClaimedJob, error: unknown) {
+  const message = error instanceof Error ? error.message : "Transcode failed";
+  const db = getDb();
+  await db
+    .update(photoJobs)
+    .set({
+      status: "failed",
+      lastError: message.slice(0, 500),
+      updatedAt: new Date(),
+    })
+    .where(eq(photoJobs.id, job.id));
+  await db
+    .update(photos)
+    .set({ processingStatus: "failed", updatedAt: new Date() })
+    .where(eq(photos.id, job.photo_id));
+}
+
+async function workerLoop(workerId: number) {
+  while (true) {
+    const started = Date.now();
+    const job = await claimJob();
+    if (!job) {
+      await Bun.sleep(POLL_MS);
+      continue;
+    }
+    try {
+      await processJob(job);
+      console.info(`worker ${workerId} transcoded photo ${job.photo_id}`);
+    } catch (error) {
+      console.error(`photo job ${job.id} failed`, error);
+      await failJob(job, error);
+    }
+    const elapsed = Date.now() - started;
+    if (elapsed < MIN_INTERVAL_MS) {
+      await Bun.sleep(MIN_INTERVAL_MS - elapsed);
+    }
+  }
+}
+
+console.info(
+  `Vellum worker listening for photo jobs (concurrency ${CONCURRENCY}, min interval ${MIN_INTERVAL_MS}ms)`,
+);
+
+await Promise.all(
+  Array.from({ length: CONCURRENCY }, (_, index) => workerLoop(index + 1)),
+);