feat: claim durable image derivative jobs

apps/api/src/modules/inventory/repositories/image-derivative-job-repository.js

@@ -1,6 +1,9 @@
-import { asc, eq } from 'drizzle-orm';
+import { and, asc, eq } from 'drizzle-orm';
 
-import { imageDerivativeJobs } from '../../../infrastructure/database/schema.js';
+import {
+  imageDerivativeJobs,
+  inventoryImages,
+} from '../../../infrastructure/database/schema.js';
 
 export function createImageDerivativeJobRepository(db) {
   return {
@@ -16,5 +19,71 @@ export function createImageDerivativeJobRepository(db) {
         .where(eq(imageDerivativeJobs.inventoryImageId, inventoryImageId))
         .orderBy(asc(imageDerivativeJobs.createdAt));
     },
+
+    async claimNext() {
+      return db.transaction(async (tx) => {
+        const [job] = await tx
+          .select()
+          .from(imageDerivativeJobs)
+          .where(eq(imageDerivativeJobs.status, 'PENDING'))
+          .orderBy(asc(imageDerivativeJobs.createdAt))
+          .limit(1)
+          .for('update', { skipLocked: true });
+
+        if (!job) return null;
+
+        const [claimed] = await tx
+          .update(imageDerivativeJobs)
+          .set({ status: 'PROCESSING', updatedAt: new Date(), error: null })
+          .where(and(
+            eq(imageDerivativeJobs.id, job.id),
+            eq(imageDerivativeJobs.status, 'PENDING'),
+          ))
+          .returning();
+
+        if (claimed) {
+          await tx
+            .update(inventoryImages)
+            .set({ processingStatus: 'PROCESSING', updatedAt: new Date() })
+            .where(eq(inventoryImages.id, claimed.inventoryImageId));
+        }
+
+        return claimed ?? null;
+      });
+    },
+
+    async complete(id, derivatives) {
+      const [completed] = await db
+        .update(imageDerivativeJobs)
+        .set({ status: 'COMPLETED', derivatives, updatedAt: new Date(), error: null })
+        .where(and(eq(imageDerivativeJobs.id, id), eq(imageDerivativeJobs.status, 'PROCESSING')))
+        .returning();
+
+      if (completed) {
+        await db
+          .update(inventoryImages)
+          .set({ processingStatus: 'COMPLETED', updatedAt: new Date() })
+          .where(eq(inventoryImages.id, completed.inventoryImageId));
+      }
+
+      return completed ?? null;
+    },
+
+    async fail(id, message) {
+      const [failed] = await db
+        .update(imageDerivativeJobs)
+        .set({ status: 'FAILED', error: message, updatedAt: new Date() })
+        .where(and(eq(imageDerivativeJobs.id, id), eq(imageDerivativeJobs.status, 'PROCESSING')))
+        .returning();
+
+      if (failed) {
+        await db
+          .update(inventoryImages)
+          .set({ processingStatus: 'FAILED', updatedAt: new Date() })
+          .where(eq(inventoryImages.id, failed.inventoryImageId));
+      }
+
+      return failed ?? null;
+    },
   };
 }

apps/api/tests/integration/catalog-repositories.test.js

@@ -9,6 +9,8 @@ import {
   conditionProfiles,
   conditions,
   inventoryItems,
+  inventoryImages,
+  imageDerivativeJobs,
   listings,
   mfaMethods,
   passwordResetTokens,
@@ -25,6 +27,7 @@ import { createPasswordResetRepository } from '../../src/modules/auth/repositori
 import { createMfaMethodRepository } from '../../src/modules/auth/repositories/mfa-method-repository.js';
 import { createInventoryItemRepository } from '../../src/modules/inventory/repositories/inventory-item-repository.js';
 import { createProductRepository } from '../../src/modules/products/repositories/product-repository.js';
+import { createImageDerivativeJobRepository } from '../../src/modules/inventory/repositories/image-derivative-job-repository.js';
 import { createSellerRepository } from '../../src/modules/sellers/repositories/seller-repository.js';
 
 const connectionString = process.env.TEST_DATABASE_URL;
@@ -39,6 +42,7 @@ if (!new URL(connectionString).pathname.endsWith('_test')) {
 
 const database = createDatabase(connectionString);
 const productRepository = createProductRepository(database.db);
+const imageDerivativeJobRepository = createImageDerivativeJobRepository(database.db);
 const sellerRepository = createSellerRepository(database.db);
 const inventoryItemRepository = createInventoryItemRepository(database.db);
 const sessionRepository = createSessionRepository(database.db);
@@ -79,7 +83,7 @@ async function createCatalogPrerequisites() {
 
 beforeEach(async () => {
   await database.db.execute(
-    sql`TRUNCATE security_audit_events, password_reset_tokens, mfa_methods, sessions, inventory_items, products, conditions, condition_profiles, categories, sellers, users CASCADE`,
+    sql`TRUNCATE image_derivative_jobs, inventory_images, security_audit_events, password_reset_tokens, mfa_methods, sessions, inventory_items, products, conditions, condition_profiles, categories, sellers, users CASCADE`,
   );
 });
 
@@ -221,6 +225,44 @@ test('lists only sellers with publicly visible inventory', async () => {
   assert.deepEqual(await sellerRepository.listPublic(), []);
 });
 
+test('claims and completes image derivative jobs while updating image processing status', async () => {
+  const { seller, category, condition } = await createCatalogPrerequisites();
+  const [product] = await database.db.insert(products).values({
+    sellerId: seller.id,
+    categoryId: category.id,
+    name: 'Derivative Job Product',
+    slug: 'derivative-job-product',
+  }).returning();
+  const [inventoryItem] = await database.db.insert(inventoryItems).values({
+    productId: product.id,
+    conditionId: condition.id,
+  }).returning();
+  const [image] = await database.db.insert(inventoryImages).values({
+    inventoryItemId: inventoryItem.id,
+    storageKey: 'inventory/derivative-source.jpg',
+    originalFilename: 'source.jpg',
+    mimeType: 'image/jpeg',
+    fileSize: 4,
+    width: 100,
+    height: 100,
+  }).returning();
+  const [job] = await database.db.insert(imageDerivativeJobs).values({
+    inventoryImageId: image.id,
+    sourceKey: image.storageKey,
+  }).returning();
+
+  const claimed = await imageDerivativeJobRepository.claimNext();
+  assert.equal(claimed.id, job.id);
+  assert.equal(claimed.status, 'PROCESSING');
+  const [processingImage] = await database.db.select().from(inventoryImages).where(sql`id = ${image.id}`);
+  assert.equal(processingImage.processingStatus, 'PROCESSING');
+
+  const completed = await imageDerivativeJobRepository.complete(job.id, [{ name: 'thumbnail', key: 'thumb.webp' }]);
+  assert.equal(completed.status, 'COMPLETED');
+  const [completedImage] = await database.db.select().from(inventoryImages).where(sql`id = ${image.id}`);
+  assert.equal(completedImage.processingStatus, 'COMPLETED');
+});
+
 test('retrieves only non-revoked, non-expired sessions', async () => {
   const { user, seller } = await createCatalogPrerequisites();
   const tokenHash = 'a hashed session token';

database/migrations/0011_ensure_image_derivative_jobs.sql

@@ -0,0 +1,13 @@
+CREATE TABLE IF NOT EXISTS "image_derivative_jobs" (
+  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+  "inventory_image_id" uuid NOT NULL REFERENCES "inventory_images"("id"),
+  "source_key" text NOT NULL,
+  "status" text DEFAULT 'PENDING' NOT NULL,
+  "derivatives" jsonb DEFAULT '[]'::jsonb NOT NULL,
+  "error" text,
+  "created_at" timestamp with time zone DEFAULT now() NOT NULL,
+  "updated_at" timestamp with time zone DEFAULT now() NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS "image_derivative_jobs_image_id_index"
+  ON "image_derivative_jobs" ("inventory_image_id");

database/migrations/meta/_journal.json

@@ -78,6 +78,13 @@
       "when": 1788919152853,
       "tag": "0010_plain_charles_xavier",
       "breakpoints": true
+    },
+    {
+      "idx": 11,
+      "version": "7",
+      "when": 1789545600000,
+      "tag": "0011_ensure_image_derivative_jobs",
+      "breakpoints": true
     }
   ]
 }