diff --git a/README.md b/README.md index a0a0743..e4ab75c 100644 --- a/README.md +++ b/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. diff --git a/apps/worker/package.json b/apps/worker/package.json new file mode 100644 index 0000000..b5e82e6 --- /dev/null +++ b/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" + } +} \ No newline at end of file diff --git a/apps/worker/src/image-derivative-worker.js b/apps/worker/src/image-derivative-worker.js index 9add04c..2199886 100644 --- a/apps/worker/src/image-derivative-worker.js +++ b/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; } \ No newline at end of file diff --git a/apps/worker/src/worker-runtime.js b/apps/worker/src/worker-runtime.js new file mode 100644 index 0000000..5249caf --- /dev/null +++ b/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; +}); \ No newline at end of file diff --git a/apps/worker/tests/image-derivative-worker.test.js b/apps/worker/tests/image-derivative-worker.test.js index fcd6ae0..49e8b36 100644 --- a/apps/worker/tests/image-derivative-worker.test.js +++ b/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' }], + }]); }); \ No newline at end of file diff --git a/package.json b/package.json index b6afc61..b353fed 100644 --- a/package.json +++ b/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\"",