feat: add queued image derivative lifecycle

apps/worker/src/image-derivative-worker.js

@@ -1,3 +1,90 @@
+export function createInMemoryImageDerivativeQueue() {
+  const jobs = new Map();
+  const pendingIds = [];
+  let activeJobId = null;
+
+  return {
+    enqueue(job) {
+      if (!job || !job.id) {
+        throw new Error('Image derivative job requires an id.');
+      }
+
+      const persisted = {
+        ...job,
+        status: job.status ?? 'PENDING',
+        error: null,
+        derivatives: job.derivatives ?? [],
+      };
+
+      jobs.set(job.id, persisted);
+      pendingIds.push(job.id);
+      return persisted;
+    },
+
+    claimNext() {
+      const nextId = pendingIds.shift();
+      if (!nextId) return null;
+
+      const current = jobs.get(nextId);
+      if (!current) return null;
+
+      activeJobId = nextId;
+      const claimed = {
+        ...current,
+        status: 'PROCESSING',
+        error: null,
+        startedAt: new Date(),
+      };
+
+      jobs.set(nextId, claimed);
+      return claimed;
+    },
+
+    getCurrentJob() {
+      if (!activeJobId) return null;
+      return jobs.get(activeJobId) ?? null;
+    },
+
+    getJob(id) {
+      return jobs.get(id) ?? null;
+    },
+
+    updateJob(id, updates) {
+      const current = jobs.get(id);
+      if (!current) return null;
+
+      const next = {
+        ...current,
+        ...updates,
+        status: updates.status ?? current.status,
+      };
+
+      jobs.set(id, next);
+      if (next.status === 'COMPLETED' || next.status === 'FAILED') {
+        if (activeJobId === id) {
+          activeJobId = null;
+        }
+      }
+      return next;
+    },
+  };
+}
+
+export async function processNextImageDerivativeJob({ queue, loadSource, processor }) {
+  if (!queue || typeof queue.claimNext !== 'function') {
+    throw new TypeError('Image derivative queue is required.');
+  }
+
+  const claimed = queue.getCurrentJob() ?? queue.claimNext();
+  if (!claimed) {
+    return null;
+  }
+
+  const result = await runImageDerivativeJob({ job: claimed, loadSource, processor });
+  queue.updateJob(claimed.id, result);
+  return result;
+}
+
 export async function runImageDerivativeJob({ job, loadSource, processor }) {
   if (!job || job.type !== 'IMAGE_DERIVATIVES') {
     throw new Error('Unsupported image derivative job.');

apps/worker/tests/image-derivative-worker.test.js

@@ -1,7 +1,11 @@
 import assert from 'node:assert/strict';
 import test from 'node:test';
 
-import { runImageDerivativeJob } from '../src/image-derivative-worker.js';
+import {
+  createInMemoryImageDerivativeQueue,
+  processNextImageDerivativeJob,
+  runImageDerivativeJob,
+} from '../src/image-derivative-worker.js';
 
 test('completes an image derivative job after loading and processing the source', async () => {
   const calls = [];
@@ -37,4 +41,29 @@ test('records a failed image derivative job without throwing worker errors', asy
   assert.equal(result.status, 'FAILED');
   assert.equal(result.error, 'source unavailable');
   assert.ok(result.failedAt instanceof Date);
+});
+
+test('claims queued jobs and persists their processing lifecycle', async () => {
+  const queue = createInMemoryImageDerivativeQueue();
+  await queue.enqueue({ id: 'job-3', type: 'IMAGE_DERIVATIVES', sourceKey: 'queued.jpg', status: 'PENDING' });
+
+  const claimed = await queue.claimNext();
+  assert.equal(claimed.status, 'PROCESSING');
+  assert.equal(queue.getJob('job-3').status, 'PROCESSING');
+
+  const result = await processNextImageDerivativeJob({
+    queue,
+    async loadSource(sourceKey) {
+      return { sourceKey, data: Buffer.from('queued-image') };
+    },
+    processor: {
+      async process(source) {
+        return [{ name: 'thumbnail', key: `${source.sourceKey}-thumbnail.webp` }];
+      },
+    },
+  });
+
+  assert.equal(result.status, 'COMPLETED');
+  assert.equal(queue.getJob('job-3').status, 'COMPLETED');
+  assert.equal(queue.getJob('job-3').derivatives[0].key, 'queued.jpg-thumbnail.webp');
 });