From 76e40d7289d19e259cec996e9d3f2f1c75dab8b3 Mon Sep 17 00:00:00 2001 From: FragginWagon Date: Wed, 9 Sep 2026 22:32:04 -0400 Subject: [PATCH] feat: schedule pricing job retention cleanup Schedule configurable completed pricing-job cleanup in the worker, broaden repository cleanup across sellers, add unit and integration coverage, and update pricing operational documentation. --- .env.example | 2 + .../pricing-refresh-job-repository.js | 3 +- .../transactions-pricing-sales.test.js | 53 ++++++++++++++++--- apps/worker/src/pricing-refresh-worker.js | 17 ++++++ apps/worker/src/worker-runtime.js | 24 +++++++++ .../tests/pricing-refresh-worker.test.js | 30 +++++++++++ docs/IMPLEMENTATION-STATUS.md | 6 +-- docs/PRICING.md | 5 +- 8 files changed, 126 insertions(+), 14 deletions(-) diff --git a/.env.example b/.env.example index 5a5732f..9d98b8f 100644 --- a/.env.example +++ b/.env.example @@ -9,6 +9,8 @@ PRICING_FIXTURE_PROVIDER_ENABLED=false PRICING_HTTP_PROVIDER_ID=http-provider PRICING_HTTP_PROVIDER_ENDPOINT= PRICING_HTTP_PROVIDER_TOKEN= +PRICING_JOB_RETENTION_DAYS=30 +PRICING_JOB_CLEANUP_INTERVAL_MS=3600000 PASSWORD_RESET_URL=http://localhost:5173/reset-password SMTP_FROM=no-reply@example.test SMTP_URL=smtp://localhost:1025 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 index 7881fd9..9cbdc90 100644 --- a/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js +++ b/apps/api/src/modules/pricing/repositories/pricing-refresh-job-repository.js @@ -164,12 +164,11 @@ export function createPricingRefreshJobRepository(db) { }); }, - async purgeCompletedBefore(sellerId, cutoff) { + async purgeCompletedBefore(cutoff) { const deleted = await db .delete(pricingRefreshJobs) .where( and( - eq(pricingRefreshJobs.sellerId, sellerId), eq(pricingRefreshJobs.status, "COMPLETED"), lte(pricingRefreshJobs.updatedAt, cutoff), ), diff --git a/apps/api/tests/integration/transactions-pricing-sales.test.js b/apps/api/tests/integration/transactions-pricing-sales.test.js index 4612992..d958376 100644 --- a/apps/api/tests/integration/transactions-pricing-sales.test.js +++ b/apps/api/tests/integration/transactions-pricing-sales.test.js @@ -41,11 +41,12 @@ const pricingRefreshJobRepository = createPricingRefreshJobRepository( const providerMappingRepository = createProviderMappingRepository(database.db); const saleRepository = createSaleRepository(database.db); -async function createFixture() { +async function createFixture(suffix = "") { + const fixtureName = suffix ? `transactions-${suffix}` : "transactions"; const [user] = await database.db .insert(users) .values({ - email: "transactions@example.test", + email: `${fixtureName}@example.test`, passwordHash: "not-a-real-password", }) .returning(); @@ -53,14 +54,14 @@ async function createFixture() { .insert(sellers) .values({ userId: user.id, - displayName: "Transactions Owner", + displayName: `${fixtureName} Owner`, }) .returning(); const [category] = await database.db .insert(categories) .values({ - name: "Transactions Cards", - slug: "transactions-cards", + name: `${fixtureName} Cards`, + slug: `${fixtureName}-cards`, }) .returning(); const [profile] = await database.db @@ -84,8 +85,8 @@ async function createFixture() { .values({ sellerId: seller.id, categoryId: category.id, - name: "Integration Card", - slug: "integration-card", + name: `${fixtureName} Card`, + slug: `${fixtureName}-card`, }) .returning(); const inventory = await database.db @@ -229,6 +230,44 @@ test("claims and completes a durable pricing refresh job", async () => { assert.equal(user.email, "transactions@example.test"); }); +test("purges completed pricing jobs across sellers before the cutoff", async () => { + const first = await createFixture(); + const second = await createFixture("second"); + const jobs = await Promise.all( + [first, second].map(({ seller, product, condition, inventory }) => + pricingRefreshJobRepository.create({ + sellerId: seller.id, + inventoryItemId: inventory[0].id, + productId: product.id, + conditionId: condition.id, + providerId: "integration", + }), + ), + ); + + for (const job of jobs) { + await pricingRefreshJobRepository.claimNext(); + await pricingRefreshJobRepository.complete( + job.id, + [{ priceMinor: 2500 }], + { valueMinor: 2500 }, + ); + } + + const deleted = await pricingRefreshJobRepository.purgeCompletedBefore( + new Date(Date.now() + 1000), + ); + assert.equal(deleted, 2); + assert.equal( + (await pricingRefreshJobRepository.listForSeller(first.seller.id)).length, + 0, + ); + assert.equal( + (await pricingRefreshJobRepository.listForSeller(second.seller.id)).length, + 0, + ); +}); + test("persists and disables seller-owned provider mappings", async () => { const { seller, product } = await createFixture(); const created = await providerMappingRepository.createForSeller( diff --git a/apps/worker/src/pricing-refresh-worker.js b/apps/worker/src/pricing-refresh-worker.js index fa7fe5e..af0c7d3 100644 --- a/apps/worker/src/pricing-refresh-worker.js +++ b/apps/worker/src/pricing-refresh-worker.js @@ -1,3 +1,20 @@ +export function getPricingJobRetentionCutoff({ now = new Date(), retentionDays }) { + if (!Number.isInteger(retentionDays) || retentionDays < 1) + throw new Error("Pricing job retention days must be a positive integer."); + const cutoff = new Date(now); + cutoff.setUTCDate(cutoff.getUTCDate() - retentionDays); + return cutoff; +} + +export async function purgeCompletedPricingRefreshJobs({ + jobRepository, + now, + retentionDays, +}) { + const cutoff = getPricingJobRetentionCutoff({ now, retentionDays }); + return jobRepository.purgeCompletedBefore(cutoff); +} + export async function processNextPersistentPricingRefreshJob({ jobRepository, pricingRefreshJob, diff --git a/apps/worker/src/worker-runtime.js b/apps/worker/src/worker-runtime.js index 122bc1d..01d4eeb 100644 --- a/apps/worker/src/worker-runtime.js +++ b/apps/worker/src/worker-runtime.js @@ -22,6 +22,7 @@ import { createHttpPricingProvider } from "../../api/src/modules/pricing/provide import { processNextPersistentImageDerivativeJob } from "./image-derivative-worker.js"; import { processNextPersistentPricingRefreshJob, + purgeCompletedPricingRefreshJobs, runPricingRefreshJob, } from "./pricing-refresh-worker.js"; @@ -29,6 +30,14 @@ const pollIntervalMs = Number.parseInt( process.env.WORKER_POLL_INTERVAL_MS ?? "2000", 10, ); +const pricingJobRetentionDays = Number.parseInt( + process.env.PRICING_JOB_RETENTION_DAYS ?? "30", + 10, +); +const pricingJobCleanupIntervalMs = Number.parseInt( + process.env.PRICING_JOB_CLEANUP_INTERVAL_MS ?? "3600000", + 10, +); async function runWorker() { const imageStorageRoot = resolveProjectImageStorageRoot( @@ -63,6 +72,15 @@ async function runWorker() { renderer: createSharpImageRenderer(), }); + async function cleanupPricingJobs() { + const deletedCount = await purgeCompletedPricingRefreshJobs({ + jobRepository: pricingRefreshJobRepository, + retentionDays: pricingJobRetentionDays, + }); + if (deletedCount > 0) + console.log(`Purged ${deletedCount} completed pricing refresh job(s).`); + } + async function processOne() { const result = await processNextPersistentImageDerivativeJob({ jobRepository, @@ -94,13 +112,19 @@ async function runWorker() { } await processOne(); + await cleanupPricingJobs(); const timer = setInterval( () => processOne().catch((error) => console.error(error)), pollIntervalMs, ); + const cleanupTimer = setInterval( + () => cleanupPricingJobs().catch((error) => console.error(error)), + pricingJobCleanupIntervalMs, + ); const shutdown = async () => { clearInterval(timer); + clearInterval(cleanupTimer); await database.close(); }; process.once("SIGINT", shutdown); diff --git a/apps/worker/tests/pricing-refresh-worker.test.js b/apps/worker/tests/pricing-refresh-worker.test.js index 0baf4a4..4f4a835 100644 --- a/apps/worker/tests/pricing-refresh-worker.test.js +++ b/apps/worker/tests/pricing-refresh-worker.test.js @@ -6,10 +6,40 @@ import { createPricingProviderRegistry, } from "../../api/src/modules/pricing/providers/pricing-provider.js"; import { + getPricingJobRetentionCutoff, processNextPersistentPricingRefreshJob, + purgeCompletedPricingRefreshJobs, runPricingRefreshJob, } from "../src/pricing-refresh-worker.js"; +test("calculates a UTC retention cutoff and purges completed jobs before it", async () => { + const now = new Date("2026-09-09T12:00:00.000Z"); + const cutoff = getPricingJobRetentionCutoff({ now, retentionDays: 30 }); + assert.equal(cutoff.toISOString(), "2026-08-10T12:00:00.000Z"); + + let receivedCutoff; + const deleted = await purgeCompletedPricingRefreshJobs({ + jobRepository: { + async purgeCompletedBefore(value) { + receivedCutoff = value; + return 3; + }, + }, + now, + retentionDays: 30, + }); + + assert.equal(deleted, 3); + assert.equal(receivedCutoff.toISOString(), cutoff.toISOString()); +}); + +test("rejects invalid pricing job retention days", () => { + assert.throws( + () => getPricingJobRetentionCutoff({ retentionDays: 0 }), + /positive integer/, + ); +}); + test("completes a pricing refresh through the provider and pricing service boundaries", async () => { const providerRegistry = createPricingProviderRegistry([ createPricingProvider({ diff --git a/docs/IMPLEMENTATION-STATUS.md b/docs/IMPLEMENTATION-STATUS.md index 216cca9..3ef82d2 100644 --- a/docs/IMPLEMENTATION-STATUS.md +++ b/docs/IMPLEMENTATION-STATUS.md @@ -125,7 +125,7 @@ Verified locally: `npm test` passes 116 tests with no failures; worker tests, we Complete production configuration and operational maturity around the implemented vertical slices. 1. Configure and validate an approved external pricing provider and provider-specific mapping data. -2. Schedule pricing-job retention cleanup and establish restore drills for PostgreSQL and image storage. +2. Establish restore drills for PostgreSQL and image storage. 3. Add richer exports/settings, broader accessibility/manual coverage, and any remaining integration edge cases. Acceptance criteria: production provider credentials are secret-managed, migrations and backups are exercised in the target environment, and CI runs integration, authenticated E2E, web, build, Storybook build, and Storybook interaction/accessibility gates. @@ -136,7 +136,7 @@ Work through one bounded vertical slice at a time. The linked documentation rema Completed vertical slices include authentication, listings and image prerequisites, acquisitions, sales and audited corrections, dashboard metrics/alerts, pricing observations and quality handling, provider mappings, durable pricing refresh jobs, public search/detail/image states, admin workflows, CI integration/E2E gates, production Docker packaging, rollback/health checks, PostgreSQL backup verification, and deployment smoke checks. -Remaining roadmap areas are richer exports/settings, approved external provider credentials/adapters, scheduled operational cleanup, broader accessibility/manual coverage, and deployment-specific NAS/TLS validation. +Remaining roadmap areas are richer exports/settings, approved external provider credentials/adapters, broader accessibility/manual coverage, and deployment-specific NAS/TLS validation. ## Known Gaps @@ -147,7 +147,7 @@ Remaining roadmap areas are richer exports/settings, approved external provider - Product-level provider mappings now persist seller-scoped `PRODUCT`/`SEARCH` relationships and support active listing and controlled disabling. - Pricing refresh enqueue now resolves its provider from an active Product mapping and rejects unmapped Products explicitly. - Local development and CI can enable a deterministic fixture provider for end-to-end refresh coverage; production provider adapters remain explicitly configured deployment boundaries. -- Pricing refresh jobs have a bounded three-attempt retry policy with exponential backoff and an explicit completed-job purge operation for retention scheduling. +- Pricing refresh jobs have a bounded three-attempt retry policy with exponential backoff. The worker schedules completed-job cleanup using configurable retention days and cleanup interval settings. - `npm audit --omit=dev` currently reports zero production vulnerabilities. Re-run the audit after dependency changes and review development-dependency findings separately before production releases. ## Local Commands diff --git a/docs/PRICING.md b/docs/PRICING.md index e825511..02a71fb 100644 --- a/docs/PRICING.md +++ b/docs/PRICING.md @@ -119,8 +119,9 @@ work. Refresh jobs allow three attempts. Failed attempts requeue with short exponential backoff; the third failure becomes terminal `FAILED`. Completed-job -retention is handled explicitly through the repository cleanup operation so a -deployment can schedule cleanup according to its retention policy. +retention is handled explicitly through the repository cleanup operation. The +worker schedules cleanup across all sellers using `PRICING_JOB_RETENTION_DAYS` +(default `30`) and `PRICING_JOB_CLEANUP_INTERVAL_MS` (default one hour). Provider failure should not invalidate other valid observations. The failed provider is recorded for admin diagnostics.