manyangles/apps/worker/src/index.ts

import { sql } from "drizzle-orm";
import { and, eq } from "drizzle-orm";
import sharp from "sharp";
import { bannerVariant, photoVariants, VARIANT_CONTENT_TYPE } from "./image-variants";
import type { BannerCrop } from "@album/contracts";
import { processPhotoExport } from "./exports";
import { processEmailDelivery } from "@album/email/queue";
import convert from "heic-convert";
import { eventBanners, 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 photoVariants(image);
  await Promise.all([
    putObject({ key: displayKey, body: display, contentType: VARIANT_CONTENT_TYPE }),
    putObject({ key: thumbKey, body: thumb, contentType: VARIANT_CONTENT_TYPE }),
  ]);
  await db.transaction(async (tx) => {
    await tx
      .update(photos)
      .set({
        processingStatus: "ready",
        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) {
      if (await processBanner()) continue;
      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);
    }
  }
}

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, crop
  `);
  const banner = (rows as unknown as { id: string; event_id: string; original_key: string; content_type: string; crop: BannerCrop | null }[])[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 bannerVariant(image, banner.crop);
    const displayKey = `events/${banner.event_id}/banners/${banner.id}/display.webp`;
    await putObject({ key: displayKey, body: display, contentType: VARIANT_CONTENT_TYPE });
    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)`,
);

async function emailLoop() {
  while (true) {
    try { await processEmailDelivery(); } catch { console.error("Email queue check failed"); }
    await Bun.sleep(1000);
  }
}

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)),
]);