From abd574293b02db0e6a0e3bcaaf0ecef5d7e8bdf1 Mon Sep 17 00:00:00 2001 From: FragginWagon Date: Tue, 8 Sep 2026 22:04:02 -0400 Subject: [PATCH] feat: claim durable image derivative jobs --- .../image-derivative-job-repository.js | 73 ++++++++++++++++++- .../integration/catalog-repositories.test.js | 44 ++++++++++- .../0011_ensure_image_derivative_jobs.sql | 13 ++++ database/migrations/meta/_journal.json | 7 ++ 4 files changed, 134 insertions(+), 3 deletions(-) create mode 100644 database/migrations/0011_ensure_image_derivative_jobs.sql diff --git a/apps/api/src/modules/inventory/repositories/image-derivative-job-repository.js b/apps/api/src/modules/inventory/repositories/image-derivative-job-repository.js index 074f3f4..5b4a664 100644 --- a/apps/api/src/modules/inventory/repositories/image-derivative-job-repository.js +++ b/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; + }, }; } \ No newline at end of file diff --git a/apps/api/tests/integration/catalog-repositories.test.js b/apps/api/tests/integration/catalog-repositories.test.js index 3ce7f57..bc445ed 100644 --- a/apps/api/tests/integration/catalog-repositories.test.js +++ b/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'; diff --git a/database/migrations/0011_ensure_image_derivative_jobs.sql b/database/migrations/0011_ensure_image_derivative_jobs.sql new file mode 100644 index 0000000..753d4e0 --- /dev/null +++ b/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"); diff --git a/database/migrations/meta/_journal.json b/database/migrations/meta/_journal.json index 49d5f25..8f8a3df 100644 --- a/database/migrations/meta/_journal.json +++ b/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 } ] } \ No newline at end of file