feat: run durable image worker locally

README.md

@@ -79,7 +79,7 @@ After creating `.env` and setting the required local SMTP and MFA values, start
 npm run dev
 ```
 
-This command starts healthy PostgreSQL and Redis containers, applies migrations, and runs the Fastify API and Vue/Vite web app together. Open `http://localhost:5173/` in a browser. The Vite server proxies `/api` requests to the API on port `3000`.
+This command starts healthy PostgreSQL and Redis containers, applies migrations, and runs the Fastify API, durable image worker, and Vue/Vite web app together. Open `http://localhost:5173/` in a browser. The Vite server proxies `/api` requests to the API on port `3000`.
 
 The command does not provision an administrator automatically. Run `npm run provision:admin` once after setting the initial administrator variables in `.env` when admin access is needed. Stop the application processes with `Ctrl+C`; stop the containers separately with `npm run db:down` when they are no longer needed.
 

apps/worker/package.json

@@ -0,0 +1,9 @@
+{
+  "name": "@pokemon-inventory/worker",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "start": "node src/worker-runtime.js",
+    "test": "node --test tests/*.test.js"
+  }
+}

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

@@ -110,4 +110,21 @@ export async function runImageDerivativeJob({ job, loadSource, processor }) {
       failedAt: new Date(),
     };
   }
+}
+
+export async function processNextPersistentImageDerivativeJob({ jobRepository, loadSource, processor }) {
+  const claimed = await jobRepository.claimNext();
+
+  if (!claimed) return null;
+
+  const job = { ...claimed, type: 'IMAGE_DERIVATIVES' };
+  const result = await runImageDerivativeJob({ job, loadSource, processor });
+
+  if (result.status === 'COMPLETED') {
+    await jobRepository.complete(claimed.id, result.derivatives);
+  } else {
+    await jobRepository.fail(claimed.id, result.error);
+  }
+
+  return result;
 }

apps/worker/src/worker-runtime.js

@@ -0,0 +1,50 @@
+import dotenv from 'dotenv';
+import { fileURLToPath } from 'node:url';
+
+dotenv.config({ path: fileURLToPath(new URL('../../../.env', import.meta.url)) });
+
+import { createDatabase } from '../../api/src/infrastructure/database/client.js';
+import { createLocalImageStorage } from '../../api/src/infrastructure/images/local-image-storage.js';
+import { createImageDerivativeProcessor } from '../../api/src/infrastructure/images/image-derivative-processor.js';
+import { createSharpImageRenderer } from '../../api/src/infrastructure/images/sharp-image-renderer.js';
+import { createImageDerivativeJobRepository } from '../../api/src/modules/inventory/repositories/image-derivative-job-repository.js';
+import { processNextPersistentImageDerivativeJob } from './image-derivative-worker.js';
+
+const pollIntervalMs = Number.parseInt(process.env.WORKER_POLL_INTERVAL_MS ?? '2000', 10);
+
+async function runWorker() {
+  const database = createDatabase();
+  const imageStorage = createLocalImageStorage(process.env.IMAGE_STORAGE_ROOT ?? '.local/image-storage');
+  const jobRepository = createImageDerivativeJobRepository(database.db);
+  const processor = createImageDerivativeProcessor({
+    imageStorage,
+    renderer: createSharpImageRenderer(),
+  });
+
+  async function processOne() {
+    const result = await processNextPersistentImageDerivativeJob({
+      jobRepository,
+      async loadSource(sourceKey) {
+        return { sourceKey, data: await imageStorage.get(sourceKey) };
+      },
+      processor,
+    });
+
+    if (result) console.log(`Image derivative job ${result.id} ${result.status.toLowerCase()}.`);
+  }
+
+  await processOne();
+  const timer = setInterval(() => processOne().catch((error) => console.error(error)), pollIntervalMs);
+
+  const shutdown = async () => {
+    clearInterval(timer);
+    await database.close();
+  };
+  process.once('SIGINT', shutdown);
+  process.once('SIGTERM', shutdown);
+}
+
+runWorker().catch((error) => {
+  console.error(error);
+  process.exitCode = 1;
+});

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

@@ -4,6 +4,7 @@ import test from 'node:test';
 import {
   createInMemoryImageDerivativeQueue,
   processNextImageDerivativeJob,
+  processNextPersistentImageDerivativeJob,
   runImageDerivativeJob,
 } from '../src/image-derivative-worker.js';
 
@@ -66,4 +67,32 @@ test('claims queued jobs and persists their processing lifecycle', async () => {
   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');
+});
+
+test('claims and persists a durable image derivative job lifecycle', async () => {
+  const calls = [];
+  const jobRepository = {
+    async claimNext() {
+      return { id: 'job-4', sourceKey: 'durable.jpg', status: 'PROCESSING' };
+    },
+    async complete(id, derivatives) {
+      calls.push({ action: 'complete', id, derivatives });
+    },
+    async fail() {
+      throw new Error('A successful durable job must not fail.');
+    },
+  };
+
+  const result = await processNextPersistentImageDerivativeJob({
+    jobRepository,
+    async loadSource(sourceKey) { return { sourceKey, data: Buffer.from('image') }; },
+    processor: { async process() { return [{ name: 'thumbnail', key: 'durable-thumbnail.webp' }]; } },
+  });
+
+  assert.equal(result.status, 'COMPLETED');
+  assert.deepEqual(calls, [{
+    action: 'complete',
+    id: 'job-4',
+    derivatives: [{ name: 'thumbnail', key: 'durable-thumbnail.webp' }],
+  }]);
 });

package.json

@@ -11,10 +11,11 @@
     "db:generate": "drizzle-kit generate",
     "db:migrate": "drizzle-kit migrate",
     "db:up": "docker compose up -d --wait postgres redis",
-    "dev": "npm run dev:prepare && concurrently --kill-others --names api,web --prefix-colors blue,green \"npm run dev:api\" \"npm run dev:web\"",
+    "dev": "npm run dev:prepare && concurrently --kill-others --names api,web,worker --prefix-colors blue,green,yellow \"npm run dev:api\" \"npm run dev:web\" \"npm run dev:worker\"",
     "dev:api": "npm run start --workspace=@pokemon-inventory/api",
     "dev:prepare": "npm run db:up && npm run db:migrate",
     "dev:web": "npm run dev --workspace=@pokemon-inventory/web",
+    "dev:worker": "npm run start --workspace=@pokemon-inventory/worker",
     "storybook": "npm run storybook --workspace=@pokemon-inventory/web",
     "build-storybook": "npm run build-storybook --workspace=@pokemon-inventory/web",
     "test-storybook": "concurrently --kill-others --success first \"npm run storybook\" \"wait-on http://127.0.0.1:6006 && npm run test-storybook --workspace=@pokemon-inventory/web\"",