From 409d2cd0ca59a77b9bce0c679bc7521835404bd6 Mon Sep 17 00:00:00 2001 From: FragginWagon Date: Wed, 9 Sep 2026 11:53:14 -0400 Subject: [PATCH] feat: persist pricing refresh jobs --- .../api/src/infrastructure/database/schema.js | 14 ++++++ .../pricing-refresh-job-repository.js | 44 +++++++++++++++++++ .../transactions-pricing-sales.test.js | 23 +++++++++- apps/worker/src/pricing-refresh-worker.js | 13 +++++- .../tests/pricing-refresh-worker.test.js | 18 +++++++- .../migrations/0017_pricing_refresh_jobs.sql | 16 +++++++ database/migrations/meta/_journal.json | 7 +++ docs/IMPLEMENTATION-STATUS.md | 2 +- docs/PRICING.md | 4 +- 9 files changed, 136 insertions(+), 5 deletions(-) create mode 100644 apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js create mode 100644 database/migrations/0017_pricing_refresh_jobs.sql diff --git a/apps/api/src/infrastructure/database/schema.js b/apps/api/src/infrastructure/database/schema.js index dd973b2..b90327c 100644 --- a/apps/api/src/infrastructure/database/schema.js +++ b/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', { diff --git a/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js b/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js new file mode 100644 index 0000000..5f2c268 --- /dev/null +++ b/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; + }, + }; +} \ No newline at end of file diff --git a/apps/api/tests/integration/transactions-pricing-sales.test.js b/apps/api/tests/integration/transactions-pricing-sales.test.js index 1f353bb..a95c03d 100644 --- a/apps/api/tests/integration/transactions-pricing-sales.test.js +++ b/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'); +}); diff --git a/apps/worker/src/pricing-refresh-worker.js b/apps/worker/src/pricing-refresh-worker.js index 9a81626..b3fd1e8 100644 --- a/apps/worker/src/pricing-refresh-worker.js +++ b/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.'); } diff --git a/apps/worker/tests/pricing-refresh-worker.test.js b/apps/worker/tests/pricing-refresh-worker.test.js index 8afee7a..b13023d 100644 --- a/apps/worker/tests/pricing-refresh-worker.test.js +++ b/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 }]]); +}); diff --git a/database/migrations/0017_pricing_refresh_jobs.sql b/database/migrations/0017_pricing_refresh_jobs.sql new file mode 100644 index 0000000..759d426 --- /dev/null +++ b/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"); \ No newline at end of file diff --git a/database/migrations/meta/_journal.json b/database/migrations/meta/_journal.json index b811253..dae838e 100644 --- a/database/migrations/meta/_journal.json +++ b/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 } ] } \ No newline at end of file diff --git a/docs/IMPLEMENTATION-STATUS.md b/docs/IMPLEMENTATION-STATUS.md index d02c95c..cf04434 100644 --- a/docs/IMPLEMENTATION-STATUS.md +++ b/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 diff --git a/docs/PRICING.md b/docs/PRICING.md index c35be2b..ac3ed65 100644 --- a/docs/PRICING.md +++ b/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.