feat: persist pricing refresh jobs

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

@@ -299,6 +299,20 @@ export const marketValueOverrides = pgTable('market_value_overrides', {
   createdAt: timestamp('created_at', { withTimezone: true }).defaultNow().notNull(),
 });
 
+export const pricingRefreshJobs = pgTable('pricing_refresh_jobs', {
+  id: uuid('id').defaultRandom().primaryKey(),
+  sellerId: uuid('seller_id').notNull().references(() => sellers.id),
+  inventoryItemId: uuid('inventory_item_id').notNull().references(() => inventoryItems.id),
+  productId: uuid('product_id').notNull().references(() => products.id),
+  conditionId: uuid('condition_id').notNull().references(() => conditions.id),
+  providerId: text('provider_id').notNull(),
+  status: text('status').default('PENDING').notNull(),
+  observations: jsonb('observations').default([]).notNull(),
+  valuation: jsonb('valuation'),
+  error: text('error'),
+  ...timestamps,
+}, (table) => [index('pricing_refresh_jobs_status_created_at_index').on(table.status, table.createdAt)]);
+
 export const inventoryImages = pgTable(
   'inventory_images',
   {

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

@@ -0,0 +1,44 @@
+import { and, asc, eq } from 'drizzle-orm';
+
+import { pricingRefreshJobs } from '../../../infrastructure/database/schema.js';
+
+export function createPricingRefreshJobRepository(db) {
+  return {
+    async create(job) {
+      const [created] = await db.insert(pricingRefreshJobs).values(job).returning();
+      return created;
+    },
+
+    async claimNext() {
+      return db.transaction(async (transaction) => {
+        const [job] = await transaction.select().from(pricingRefreshJobs)
+          .where(eq(pricingRefreshJobs.status, 'PENDING'))
+          .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 })
+          .where(and(eq(pricingRefreshJobs.id, job.id), eq(pricingRefreshJobs.status, 'PENDING')))
+          .returning();
+        return claimed ?? null;
+      });
+    },
+
+    async complete(id, observations, valuation) {
+      const [completed] = await db.update(pricingRefreshJobs)
+        .set({ status: 'COMPLETED', observations, valuation, error: null, updatedAt: new Date() })
+        .where(and(eq(pricingRefreshJobs.id, id), eq(pricingRefreshJobs.status, 'PROCESSING')))
+        .returning();
+      return completed ?? null;
+    },
+
+    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;
+    },
+  };
+}

apps/api/tests/integration/transactions-pricing-sales.test.js

@@ -16,6 +16,7 @@ import {
 } from '../../src/infrastructure/database/schema.js';
 import { createAcquisitionRepository } from '../../src/modules/acquisitions/repositories/acquisition-repository.js';
 import { createPricingRepository } from '../../src/modules/pricing/repositories/pricing-repository.js';
+import { createPricingRefreshJobRepository } from '../../src/modules/pricing/repositories/pricing-refresh-job-repository.js';
 import { createSaleRepository } from '../../src/modules/sales/repositories/sale-repository.js';
 
 const connectionString = process.env.TEST_DATABASE_URL;
@@ -31,6 +32,7 @@ if (!new URL(connectionString).pathname.endsWith('_test')) {
 const database = createDatabase(connectionString);
 const acquisitionRepository = createAcquisitionRepository(database.db);
 const pricingRepository = createPricingRepository(database.db);
+const pricingRefreshJobRepository = createPricingRefreshJobRepository(database.db);
 const saleRepository = createSaleRepository(database.db);
 
 async function createFixture() {
@@ -71,7 +73,7 @@ async function createFixture() {
 }
 
 beforeEach(async () => {
-  await database.db.execute(sql`TRUNCATE market_value_overrides, market_valuations, pricing_observations, sale_items, sales, acquisitions, security_audit_events, inventory_items, products, conditions, condition_profiles, categories, sellers, users CASCADE`);
+  await database.db.execute(sql`TRUNCATE pricing_refresh_jobs, market_value_overrides, market_valuations, pricing_observations, sale_items, sales, acquisitions, security_audit_events, inventory_items, products, conditions, condition_profiles, categories, sellers, users CASCADE`);
 });
 
 after(async () => {
@@ -155,3 +157,22 @@ test('persists acquisitions, valuations, multi-item sales, voids, and audit even
   const [auditEvent] = await database.db.select().from(securityAuditEvents).where(sql`event = 'SALE_VOIDED'`);
   assert.equal(auditEvent.userId, user.id);
 });
+
+test('claims and completes a durable pricing refresh job', async () => {
+  const { user, seller, product, condition, inventory } = await createFixture();
+  const [job] = await pricingRefreshJobRepository.create({
+    sellerId: seller.id,
+    inventoryItemId: inventory[0].id,
+    productId: product.id,
+    conditionId: condition.id,
+    providerId: 'integration',
+  }).then((created) => [created]);
+
+  const claimed = await pricingRefreshJobRepository.claimNext();
+  assert.equal(claimed.id, job.id);
+  assert.equal(claimed.status, 'PROCESSING');
+  const completed = await pricingRefreshJobRepository.complete(job.id, [{ priceMinor: 2500 }], { valueMinor: 2500 });
+  assert.equal(completed.status, 'COMPLETED');
+  assert.deepEqual(completed.valuation, { valueMinor: 2500 });
+  assert.equal(user.email, 'transactions@example.test');
+});

apps/worker/src/pricing-refresh-worker.js

@@ -1,4 +1,15 @@
-export async function runPricingRefreshJob({ job, providerRegistry, pricingService }) {
+export async function processNextPersistentPricingRefreshJob({ jobRepository, pricingRefreshJob }) {
+  const claimed = await jobRepository.claimNext();
+  if (!claimed) return null;
+
+  const result = await pricingRefreshJob({ ...claimed, type: 'PRICING_REFRESH' });
+  if (result.status === 'COMPLETED') {
+    await jobRepository.complete(claimed.id, result.observations, result.valuation);
+  } else {
+    await jobRepository.fail(claimed.id, result.error);
+  }
+  return result;
+}export async function runPricingRefreshJob({ job, providerRegistry, pricingService }) {
   if (!job || job.type !== 'PRICING_REFRESH') {
     throw new Error('Unsupported pricing refresh job.');
   }

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

@@ -2,7 +2,7 @@ import assert from 'node:assert/strict';
 import test from 'node:test';
 
 import { createPricingProvider, createPricingProviderRegistry } from '../../api/src/modules/pricing/providers/pricing-provider.js';
-import { runPricingRefreshJob } from '../src/pricing-refresh-worker.js';
+import { processNextPersistentPricingRefreshJob, runPricingRefreshJob } from '../src/pricing-refresh-worker.js';
 
 test('completes a pricing refresh through the provider and pricing service boundaries', async () => {
   const providerRegistry = createPricingProviderRegistry([
@@ -46,3 +46,19 @@ test('fails a refresh clearly when a provider is unavailable', async () => {
   assert.equal(result.status, 'FAILED');
   assert.match(result.error, /not configured/);
 });
+
+test('persists completed and failed outcomes through the job repository', async () => {
+  const calls = [];
+  const jobRepository = {
+    async claimNext() { return { id: 'job-3', type: 'PRICING_REFRESH', providerId: 'fixture' }; },
+    async complete(id, observations, valuation) { calls.push(['complete', id, observations, valuation]); },
+    async fail(id, error) { calls.push(['fail', id, error]); },
+  };
+
+  await processNextPersistentPricingRefreshJob({
+    jobRepository,
+    async pricingRefreshJob() { return { status: 'COMPLETED', observations: [{ id: 'observation-1' }], valuation: { valueMinor: 2500 } }; },
+  });
+
+  assert.deepEqual(calls, [['complete', 'job-3', [{ id: 'observation-1' }], { valueMinor: 2500 }]]);
+});

database/migrations/0017_pricing_refresh_jobs.sql

@@ -0,0 +1,16 @@
+CREATE TABLE "pricing_refresh_jobs" (
+  "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+  "seller_id" uuid NOT NULL REFERENCES "sellers"("id"),
+  "inventory_item_id" uuid NOT NULL REFERENCES "inventory_items"("id"),
+  "product_id" uuid NOT NULL REFERENCES "products"("id"),
+  "condition_id" uuid NOT NULL REFERENCES "conditions"("id"),
+  "provider_id" text NOT NULL,
+  "status" text DEFAULT 'PENDING' NOT NULL,
+  "observations" jsonb DEFAULT '[]'::jsonb NOT NULL,
+  "valuation" jsonb,
+  "error" text,
+  "created_at" timestamp with time zone DEFAULT now() NOT NULL,
+  "updated_at" timestamp with time zone DEFAULT now() NOT NULL
+);
+--> statement-breakpoint
+CREATE INDEX "pricing_refresh_jobs_status_created_at_index" ON "pricing_refresh_jobs" USING btree ("status", "created_at");

database/migrations/meta/_journal.json

@@ -120,6 +120,13 @@
       "when": 1789977600000,
       "tag": "0016_pricing_quality",
       "breakpoints": true
+    },
+    {
+      "idx": 17,
+      "version": "7",
+      "when": 1790064000000,
+      "tag": "0017_pricing_refresh_jobs",
+      "breakpoints": true
     }
   ]
 }

docs/IMPLEMENTATION-STATUS.md

@@ -149,7 +149,7 @@ Work through one bounded vertical slice at a time. The linked documentation rema
 - No initial User/Seller is provisioned in the development database yet. Run `npm run provision:admin` after setting its required local `.env` variables.
 - Listings are partially implemented: draft creation, seller-scoped reads, publication gating, lifecycle writes, archive behavior, sold/archived inventory checks, and item-owned image metadata are complete; storage-backed processing remains.
 - Concrete provider adapters, asynchronous pricing refresh transport, richer exports/settings, full integration CI services, and deployment/backup automation remain incomplete to varying degrees.
-- The pricing refresh worker boundary now has explicit completed/failed outcomes and routes provider observations through the pricing service; persistent refresh-job storage and a configured external provider remain deployment work.
+- The pricing refresh worker now has durable PostgreSQL refresh jobs with row-locked claim, completed/failed persistence, and provider observations routed through the pricing service; configured external providers remain deployment work.
 - `npm install` currently reports dependency audit findings. Review and remediate them before production deployment; do not use `npm audit fix --force` without reviewing its changes.
 
 ## Local Commands

docs/PRICING.md

@@ -108,7 +108,9 @@ The worker-side refresh boundary is implemented with explicit `COMPLETED` and
 `FAILED` outcomes. It resolves a registered provider, persists normalized
 observations through the pricing service, and recalculates the valuation. A
 provider that is missing or fails does not produce a fabricated value. Durable
-refresh-job storage and external provider configuration remain deployment work.
+refresh-job storage and row-locked worker claim/complete/fail lifecycle are
+implemented in PostgreSQL. External provider configuration remains deployment
+work.
 
 Provider failure should not invalidate other valid observations. The failed provider is recorded for admin diagnostics.