feat: bound pricing job retries

apps/api/src/infrastructure/database/schema.js

@@ -321,6 +321,8 @@ export const pricingRefreshJobs = pgTable('pricing_refresh_jobs', {
   conditionId: uuid('condition_id').notNull().references(() => conditions.id),
   providerId: text('provider_id').notNull(),
   status: text('status').default('PENDING').notNull(),
+  attemptCount: integer('attempt_count').default(0).notNull(),
+  nextAttemptAt: timestamp('next_attempt_at', { withTimezone: true }),
   observations: jsonb('observations').default([]).notNull(),
   valuation: jsonb('valuation'),
   error: text('error'),

apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js

@@ -1,8 +1,9 @@
-import { and, asc, desc, eq } from 'drizzle-orm';
+import { and, asc, desc, eq, isNull, lte, or } from 'drizzle-orm';
 
 import { pricingRefreshJobs } from '../../../infrastructure/database/schema.js';
 
 export function createPricingRefreshJobRepository(db) {
+  const maxAttempts = 3;
   return {
     async create(job) {
       const [created] = await db.insert(pricingRefreshJobs).values(job).returning();
@@ -12,13 +13,16 @@ export function createPricingRefreshJobRepository(db) {
     async claimNext() {
       return db.transaction(async (transaction) => {
         const [job] = await transaction.select().from(pricingRefreshJobs)
-          .where(eq(pricingRefreshJobs.status, 'PENDING'))
+          .where(and(
+            eq(pricingRefreshJobs.status, 'PENDING'),
+            or(isNull(pricingRefreshJobs.nextAttemptAt), lte(pricingRefreshJobs.nextAttemptAt, new Date())),
+          ))
           .orderBy(asc(pricingRefreshJobs.createdAt))
           .limit(1)
           .for('update', { skipLocked: true });
         if (!job) return null;
         const [claimed] = await transaction.update(pricingRefreshJobs)
-          .set({ status: 'PROCESSING', updatedAt: new Date(), error: null })
+          .set({ status: 'PROCESSING', attemptCount: pricingRefreshJobs.attemptCount, updatedAt: new Date(), error: null, nextAttemptAt: null })
           .where(and(eq(pricingRefreshJobs.id, job.id), eq(pricingRefreshJobs.status, 'PENDING')))
           .returning();
         return claimed ?? null;
@@ -46,8 +50,8 @@ export function createPricingRefreshJobRepository(db) {
 
     async retryForSeller(id, sellerId) {
       const [job] = await db.update(pricingRefreshJobs)
-        .set({ status: 'PENDING', error: null, observations: [], valuation: null, updatedAt: new Date() })
-        .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.sellerId, sellerId), eq(pricingRefreshJobs.status, 'FAILED')))
+        .set({ status: 'PENDING', error: null, observations: [], valuation: null, nextAttemptAt: new Date(), updatedAt: new Date() })
+        .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.sellerId, sellerId), eq(pricingRefreshJobs.status, 'FAILED'), lte(pricingRefreshJobs.attemptCount, maxAttempts - 1)))
         .returning();
       return job ?? null;
     },
@@ -61,11 +65,31 @@ export function createPricingRefreshJobRepository(db) {
     },
 
     async fail(id, message) {
-      const [failed] = await db.update(pricingRefreshJobs)
-        .set({ status: 'FAILED', error: message, updatedAt: new Date() })
-        .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.status, 'PROCESSING')))
-        .returning();
-      return failed ?? null;
+      return db.transaction(async (transaction) => {
+        const [current] = await transaction.select().from(pricingRefreshJobs)
+          .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.status, 'PROCESSING')));
+        if (!current) return null;
+        const attemptCount = current.attemptCount + 1;
+        const retryable = attemptCount < maxAttempts;
+        const [updated] = await transaction.update(pricingRefreshJobs)
+          .set({
+            status: retryable ? 'PENDING' : 'FAILED',
+            error: message,
+            attemptCount,
+            updatedAt: new Date(),
+            nextAttemptAt: retryable ? new Date(Date.now() + (2 ** attemptCount) * 1000) : null,
+          })
+          .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.status, 'PROCESSING')))
+          .returning();
+        return updated ?? null;
+      });
+    },
+
+    async purgeCompletedBefore(sellerId, cutoff) {
+      const deleted = await db.delete(pricingRefreshJobs)
+        .where(and(eq(pricingRefreshJobs.sellerId, sellerId), eq(pricingRefreshJobs.status, 'COMPLETED'), lte(pricingRefreshJobs.updatedAt, cutoff)))
+        .returning({ id: pricingRefreshJobs.id });
+      return deleted.length;
     },
   };
 }

apps/worker/tests/pricing-refresh-worker.test.js

@@ -62,3 +62,17 @@ test('persists completed and failed outcomes through the job repository', async
 
   assert.deepEqual(calls, [['complete', 'job-3', [{ id: 'observation-1' }], { valueMinor: 2500 }]]);
 });
+
+test('persists failed pricing refresh outcomes through the repository', async () => {
+  const calls = [];
+  await processNextPersistentPricingRefreshJob({
+    jobRepository: {
+      async claimNext() { return { id: 'job-4' }; },
+      async complete() { throw new Error('not expected'); },
+      async fail(id, error) { calls.push([id, error]); },
+    },
+    async pricingRefreshJob() { return { status: 'FAILED', error: 'Provider unavailable.' }; },
+  });
+
+  assert.deepEqual(calls, [['job-4', 'Provider unavailable.']]);
+});