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.

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),
           ),

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(

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,

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);

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({