feat(core-shared): add retention-purge background job

Adds retention-purge.job.ts to packages/core-shared/src/payload/retention-purge/.
Walks every Payload collection's custom.retention.purgeSchedule, registers one
scheduled job per collection via IJobQueue, queries rows past their activeRetention
period (by createdAt for from-creation, updatedAt for from-last-access), and applies
pseudonymize (null PII fields) or hard-delete per postDeletion.action. Emits one
IAuditLog.record entry per processed row; gracefully skips when auditLog is absent.

Unit tests cover: schedule registration, trigger-type routing, hard-delete branch,
pseudonymize branch, audit emission, graceful auditLog skip, no-activeRetention
short-circuit, and input validation error cases.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

coverage/summary.json

@@ -1,18 +1,18 @@
 {
-  "generatedAt": "2026-05-18T18:33:27.459Z",
+  "generatedAt": "2026-05-18T18:55:54.185Z",
-  "commit": "fa1a10c",
+  "commit": "d326270",
   "repo": {
-    "statements": 96.34,
+    "statements": 96.43,
-    "branches": 91.41,
+    "branches": 91.71,
-    "functions": 96.76,
+    "functions": 96.81,
-    "lines": 96.34,
+    "lines": 96.43,
     "counts": {
-      "lf": 4100,
+      "lf": 4205,
-      "lh": 3950,
+      "lh": 4055,
-      "brf": 768,
+      "brf": 808,
-      "brh": 702,
+      "brh": 741,
-      "fnf": 247,
+      "fnf": 251,
-      "fnh": 239
+      "fnh": 243
     }
   },
   "byPackage": {
@@ -59,17 +59,17 @@
       }
     },
     "@repo/core-shared": {
-      "statements": 97.75,
+      "statements": 97.98,
-      "branches": 95.54,
+      "branches": 95.79,
-      "functions": 91.58,
+      "functions": 91.92,
-      "lines": 97.75,
+      "lines": 97.98,
       "counts": {
-        "lf": 935,
+        "lf": 1040,
-        "lh": 914,
+        "lh": 1019,
-        "brf": 269,
+        "brf": 309,
-        "brh": 257,
+        "brh": 296,
-        "fnf": 95,
+        "fnf": 99,
-        "fnh": 87
+        "fnh": 91
       }
     },
     "@repo/marketing-pages": {

packages/core-shared/src/payload/index.ts

@@ -14,3 +14,14 @@ export type {
 } from "./pii-types";
 export { PAYLOAD_AUTH_PII_DEFAULTS } from "./pii-types";
 export type { PurgeSchedule, CollectionRetention } from "./retention-types";
+export {
+  parseDurationMs,
+  scheduleDelayMs,
+  buildPurgeHandler,
+  registerRetentionPurgeJobs,
+} from "./retention-purge/retention-purge.job";
+export type {
+  PayloadPurgeApi,
+  GetPayloadFn,
+  RetentionPurgeJobDeps,
+} from "./retention-purge/retention-purge.job";

packages/core-shared/src/payload/retention-purge/retention-purge.job.test.ts

@@ -0,0 +1,608 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import type { SanitizedConfig } from "payload";
+import type { IJobQueue } from "@/jobs/job-queue.interface";
+import type { AuditLogProtocol } from "@/di/bind-protocols";
+import {
+  parseDurationMs,
+  scheduleDelayMs,
+  buildPurgeHandler,
+  registerRetentionPurgeJobs,
+  type PayloadPurgeApi,
+  type RetentionPurgeJobDeps,
+} from "./retention-purge.job";
+
+// ---- test helpers ----
+
+type MockCollection = {
+  slug: string;
+  custom?: { retention?: Record<string, unknown> };
+  fields?: Array<{ name?: string; custom?: { pii?: unknown } }>;
+};
+
+function makeConfig(collections: MockCollection[]): SanitizedConfig {
+  return { collections } as unknown as SanitizedConfig;
+}
+
+function makeQueue() {
+  const enqueue = vi.fn().mockResolvedValue({ jobId: "job-1" });
+  const queue = { enqueue } as unknown as IJobQueue;
+  return { queue, enqueue };
+}
+
+function makePayloadApi(
+  docs: Array<Record<string, unknown>> = [],
+): PayloadPurgeApi {
+  return {
+    find: vi.fn().mockResolvedValue({ docs }),
+    update: vi.fn().mockResolvedValue({}),
+    delete: vi.fn().mockResolvedValue({}),
+  };
+}
+
+function makeAuditLog(): {
+  auditLog: AuditLogProtocol;
+  record: ReturnType<typeof vi.fn>;
+} {
+  const record = vi.fn().mockResolvedValue(undefined);
+  return { auditLog: { record } as AuditLogProtocol, record };
+}
+
+// ---- parseDurationMs ----
+
+describe("parseDurationMs", () => {
+  it("parses years: P2Y → 2 × 365 days", () => {
+    expect(parseDurationMs("P2Y")).toBe(2 * 365 * 86_400_000);
+  });
+
+  it("parses months: P1M → 30 days", () => {
+    expect(parseDurationMs("P1M")).toBe(30 * 86_400_000);
+  });
+
+  it("parses weeks: P1W → 7 days", () => {
+    expect(parseDurationMs("P1W")).toBe(7 * 86_400_000);
+  });
+
+  it("parses days: P30D → 30 days", () => {
+    expect(parseDurationMs("P30D")).toBe(30 * 86_400_000);
+  });
+
+  it("combines components: P1Y2M3D", () => {
+    expect(parseDurationMs("P1Y2M3D")).toBe(
+      365 * 86_400_000 + 60 * 86_400_000 + 3 * 86_400_000,
+    );
+  });
+
+  it("returns 0 for P0D", () => {
+    expect(parseDurationMs("P0D")).toBe(0);
+  });
+
+  it("returns 0 for unrecognised strings", () => {
+    expect(parseDurationMs("invalid")).toBe(0);
+    expect(parseDurationMs("")).toBe(0);
+    expect(parseDurationMs("PT2H")).toBe(0);
+  });
+});
+
+// ---- scheduleDelayMs ----
+
+describe("scheduleDelayMs", () => {
+  it("returns 1 day for 'daily'", () => {
+    expect(scheduleDelayMs("daily")).toBe(86_400_000);
+  });
+
+  it("returns 7 days for 'weekly'", () => {
+    expect(scheduleDelayMs("weekly")).toBe(7 * 86_400_000);
+  });
+
+  it("returns 30 days for 'monthly'", () => {
+    expect(scheduleDelayMs("monthly")).toBe(30 * 86_400_000);
+  });
+
+  it("falls back to 1 day for cron-style strings", () => {
+    expect(scheduleDelayMs("0 3 * * 0")).toBe(86_400_000);
+  });
+});
+
+// ---- registerRetentionPurgeJobs ----
+
+describe("registerRetentionPurgeJobs", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("enqueues one job per collection with a purgeSchedule", async () => {
+    const { queue, enqueue } = makeQueue();
+    const config = makeConfig([
+      { slug: "users", custom: { retention: { purgeSchedule: "daily" } } },
+      { slug: "articles", custom: { retention: { purgeSchedule: "weekly" } } },
+    ]);
+    await registerRetentionPurgeJobs({ queue, config, getPayload: vi.fn() });
+    expect(enqueue).toHaveBeenCalledTimes(2);
+  });
+
+  it("skips collections without a retention config", async () => {
+    const { queue, enqueue } = makeQueue();
+    const config = makeConfig([{ slug: "media" }]);
+    await registerRetentionPurgeJobs({ queue, config, getPayload: vi.fn() });
+    expect(enqueue).not.toHaveBeenCalled();
+  });
+
+  it("uses the correct taskSlug and runAt for each schedule type", async () => {
+    const { queue, enqueue } = makeQueue();
+    const config = makeConfig([
+      { slug: "users", custom: { retention: { purgeSchedule: "weekly" } } },
+    ]);
+    await registerRetentionPurgeJobs({ queue, config, getPayload: vi.fn() });
+    expect(enqueue).toHaveBeenCalledWith(
+      "retention-purge--users",
+      {},
+      { runAt: new Date("2026-01-08T00:00:00.000Z") },
+    );
+  });
+
+  it("schedules daily purge 1 day from now", async () => {
+    const { queue, enqueue } = makeQueue();
+    const config = makeConfig([
+      { slug: "sessions", custom: { retention: { purgeSchedule: "daily" } } },
+    ]);
+    await registerRetentionPurgeJobs({ queue, config, getPayload: vi.fn() });
+    expect(enqueue).toHaveBeenCalledWith(
+      "retention-purge--sessions",
+      {},
+      { runAt: new Date("2026-01-02T00:00:00.000Z") },
+    );
+  });
+});
+
+// ---- buildPurgeHandler — input validation ----
+
+describe("buildPurgeHandler — input validation", () => {
+  it("throws when the collection slug is not found in the config", () => {
+    const { queue } = makeQueue();
+    const config = makeConfig([]);
+    expect(() =>
+      buildPurgeHandler("missing", { queue, config, getPayload: vi.fn() }),
+    ).toThrow("collection not found: missing");
+  });
+
+  it("throws when the collection has no retention config", () => {
+    const { queue } = makeQueue();
+    const config = makeConfig([{ slug: "media" }]);
+    expect(() =>
+      buildPurgeHandler("media", { queue, config, getPayload: vi.fn() }),
+    ).toThrow("no retention config on collection: media");
+  });
+});
+
+// ---- buildPurgeHandler — hard-delete branch ----
+
+describe("buildPurgeHandler — hard-delete", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("queries by createdAt for from-creation trigger", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([]);
+    const config = makeConfig([
+      {
+        slug: "users",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P2Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "hard-delete",
+              duration: "P30D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await buildPurgeHandler("users", deps)();
+
+    expect(payload.find).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: {
+          createdAt: {
+            less_than: new Date(
+              Date.now() - parseDurationMs("P2Y"),
+            ).toISOString(),
+          },
+        },
+      }),
+    );
+  });
+
+  it("queries by updatedAt for from-last-access trigger", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([]);
+    const config = makeConfig([
+      {
+        slug: "sessions",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P30D", trigger: "from-last-access" },
+            postDeletion: {
+              action: "hard-delete",
+              duration: "P0D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await buildPurgeHandler("sessions", deps)();
+
+    expect(payload.find).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { updatedAt: { less_than: expect.any(String) } },
+      }),
+    );
+    expect(payload.find).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({ createdAt: expect.anything() }),
+      }),
+    );
+  });
+
+  it("deletes each returned row", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([{ id: "row-1" }, { id: "row-2" }]);
+    const config = makeConfig([
+      {
+        slug: "users",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "hard-delete",
+              duration: "P0D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await buildPurgeHandler("users", deps)();
+
+    expect(payload.delete).toHaveBeenCalledTimes(2);
+    expect(payload.delete).toHaveBeenCalledWith({
+      collection: "users",
+      id: "row-1",
+      overrideAccess: true,
+    });
+    expect(payload.delete).toHaveBeenCalledWith({
+      collection: "users",
+      id: "row-2",
+      overrideAccess: true,
+    });
+    expect(payload.update).not.toHaveBeenCalled();
+  });
+
+  it("re-enqueues itself for the next purge cycle", async () => {
+    const { queue, enqueue } = makeQueue();
+    const payload = makePayloadApi([]);
+    const config = makeConfig([
+      {
+        slug: "users",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "hard-delete",
+              duration: "P0D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await buildPurgeHandler("users", deps)();
+
+    expect(enqueue).toHaveBeenCalledOnce();
+    expect(enqueue).toHaveBeenCalledWith(
+      "retention-purge--users",
+      {},
+      { runAt: new Date("2026-01-02T00:00:00.000Z") },
+    );
+  });
+
+  it("defaults to hard-delete when postDeletion is not declared", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([{ id: "row-x" }]);
+    const config = makeConfig([
+      {
+        slug: "logs",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await buildPurgeHandler("logs", deps)();
+
+    expect(payload.delete).toHaveBeenCalledWith({
+      collection: "logs",
+      id: "row-x",
+      overrideAccess: true,
+    });
+    expect(payload.update).not.toHaveBeenCalled();
+  });
+});
+
+// ---- buildPurgeHandler — pseudonymize branch ----
+
+describe("buildPurgeHandler — pseudonymize", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("nulls only PII-annotated fields for each matched row", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([{ id: "row-2" }]);
+    const config = makeConfig([
+      {
+        slug: "contacts",
+        custom: {
+          retention: {
+            purgeSchedule: "monthly",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "pseudonymize",
+              duration: "P30D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [
+          { name: "email", custom: { pii: { category: "contact-email" } } },
+          { name: "phone", custom: { pii: { category: "contact-phone" } } },
+          { name: "status" }, // no pii — must NOT be nulled
+        ],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await buildPurgeHandler("contacts", deps)();
+
+    expect(payload.update).toHaveBeenCalledWith({
+      collection: "contacts",
+      id: "row-2",
+      data: { email: null, phone: null },
+      overrideAccess: true,
+    });
+    expect(payload.delete).not.toHaveBeenCalled();
+  });
+
+  it("skips update when no PII fields are declared on the collection", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([{ id: "row-3" }]);
+    const config = makeConfig([
+      {
+        slug: "tags",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "pseudonymize",
+              duration: "P0D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [
+          { name: "label" }, // no pii
+        ],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await buildPurgeHandler("tags", deps)();
+
+    expect(payload.update).not.toHaveBeenCalled();
+    expect(payload.delete).not.toHaveBeenCalled();
+  });
+});
+
+// ---- buildPurgeHandler — audit emission ----
+
+describe("buildPurgeHandler — audit emission", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-01-01T00:00:00.000Z"));
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("emits one audit record per processed row", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([{ id: "a" }, { id: "b" }]);
+    const { auditLog, record } = makeAuditLog();
+    const config = makeConfig([
+      {
+        slug: "users",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "hard-delete",
+              duration: "P0D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+      auditLog,
+    };
+    await buildPurgeHandler("users", deps)();
+
+    expect(record).toHaveBeenCalledTimes(2);
+    expect(record).toHaveBeenCalledWith(
+      expect.objectContaining({
+        actorId: "system",
+        actorType: "system",
+        action: "DELETE",
+        reason: "retention-policy",
+        outcome: "success",
+        from: { ipTruncated: "system", userAgent: "background-job" },
+      }),
+    );
+  });
+
+  it("includes resource type and id in the audit entry", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([{ id: "row-42" }]);
+    const { auditLog, record } = makeAuditLog();
+    const config = makeConfig([
+      {
+        slug: "orders",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "hard-delete",
+              duration: "P0D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+      auditLog,
+    };
+    await buildPurgeHandler("orders", deps)();
+
+    expect(record).toHaveBeenCalledWith(
+      expect.objectContaining({
+        resource: { type: "orders", id: "row-42" },
+      }),
+    );
+  });
+
+  it("gracefully skips audit emission when auditLog is not provided", async () => {
+    const { queue } = makeQueue();
+    const payload = makePayloadApi([{ id: "x" }]);
+    const config = makeConfig([
+      {
+        slug: "users",
+        custom: {
+          retention: {
+            purgeSchedule: "daily",
+            activeRetention: { duration: "P1Y", trigger: "from-creation" },
+            postDeletion: {
+              action: "hard-delete",
+              duration: "P0D",
+              trigger: "after-deletion",
+            },
+          },
+        },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+    };
+    await expect(buildPurgeHandler("users", deps)()).resolves.toBeUndefined();
+  });
+
+  it("skips all processing and audit when activeRetention is not declared", async () => {
+    const { queue, enqueue } = makeQueue();
+    const payload = makePayloadApi([{ id: "y" }]);
+    const { auditLog, record } = makeAuditLog();
+    const config = makeConfig([
+      {
+        slug: "logs",
+        custom: { retention: { purgeSchedule: "daily" } },
+        fields: [],
+      },
+    ]);
+    const deps: RetentionPurgeJobDeps = {
+      queue,
+      config,
+      getPayload: vi.fn().mockResolvedValue(payload),
+      auditLog,
+    };
+    await buildPurgeHandler("logs", deps)();
+
+    expect(payload.find).not.toHaveBeenCalled();
+    expect(record).not.toHaveBeenCalled();
+    // Still re-enqueues for the next cycle
+    expect(enqueue).toHaveBeenCalledOnce();
+  });
+});

packages/core-shared/src/payload/retention-purge/retention-purge.job.ts

@@ -0,0 +1,200 @@
+import type { SanitizedConfig } from "payload";
+import type { IJobQueue } from "../../jobs/job-queue.interface";
+import type { AuditLogProtocol } from "../../di/bind-protocols";
+
+/**
+ * Minimal Payload API surface needed by the retention purge job.
+ * Injected via getPayload for testability.
+ */
+export type PayloadPurgeApi = {
+  find(args: {
+    collection: string;
+    where: Record<string, unknown>;
+    limit: number;
+    overrideAccess: true;
+  }): Promise<{ docs: Array<Record<string, unknown>> }>;
+  update(args: {
+    collection: string;
+    id: string | number;
+    data: Record<string, unknown>;
+    overrideAccess: true;
+  }): Promise<unknown>;
+  delete(args: {
+    collection: string;
+    id: string | number;
+    overrideAccess: true;
+  }): Promise<unknown>;
+};
+
+export type GetPayloadFn = (args: {
+  config: SanitizedConfig;
+}) => Promise<PayloadPurgeApi>;
+
+export type RetentionPurgeJobDeps = {
+  queue: IJobQueue;
+  config: SanitizedConfig;
+  getPayload: GetPayloadFn;
+  auditLog?: AuditLogProtocol;
+};
+
+const MS_PER_DAY = 86_400_000;
+
+/**
+ * Parse a subset of ISO 8601 duration notation (date components: Y, M, W, D)
+ * to milliseconds. Returns 0 for unrecognised patterns.
+ * Approximations: 1 year = 365 days, 1 month = 30 days.
+ */
+export function parseDurationMs(iso: string): number {
+  const match = /^P(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)W)?(?:(\d+)D)?$/.exec(iso);
+  if (!match) return 0;
+  const years = parseInt(match[1] ?? "0", 10);
+  const months = parseInt(match[2] ?? "0", 10);
+  const weeks = parseInt(match[3] ?? "0", 10);
+  const days = parseInt(match[4] ?? "0", 10);
+  return (
+    years * 365 * MS_PER_DAY +
+    months * 30 * MS_PER_DAY +
+    weeks * 7 * MS_PER_DAY +
+    days * MS_PER_DAY
+  );
+}
+
+/**
+ * Convert a PurgeSchedule value to the delay in ms before the next run.
+ * Cron-style strings fall back to daily cadence; the Payload scheduler handles
+ * actual cron-aligned firing.
+ */
+export function scheduleDelayMs(schedule: string): number {
+  if (schedule === "weekly") return 7 * MS_PER_DAY;
+  if (schedule === "monthly") return 30 * MS_PER_DAY;
+  return MS_PER_DAY; // "daily" and cron fallback
+}
+
+/**
+ * Build the purge handler for a single collection. The returned async function
+ * is intended to be registered as a Payload job task handler.
+ *
+ * Per run:
+ *  1. Query rows past their activeRetention period.
+ *  2. Apply postDeletion.action (pseudonymize | hard-delete).
+ *  3. Emit one audit entry per processed row (skipped when auditLog is absent).
+ *  4. Re-enqueue itself for the next purge cycle.
+ *
+ * `from-last-access` uses updatedAt as a proxy; a dedicated lastAccessedAt hook
+ * is deferred to Q2 per the PRD.
+ */
+export function buildPurgeHandler(
+  collectionSlug: string,
+  deps: RetentionPurgeJobDeps,
+): () => Promise<void> {
+  const { queue, config, getPayload, auditLog } = deps;
+
+  const collection = config.collections.find((c) => c.slug === collectionSlug);
+  if (!collection) {
+    throw new Error(`retention-purge: collection not found: ${collectionSlug}`);
+  }
+
+  const retention = collection.custom?.retention;
+  if (!retention) {
+    throw new Error(
+      `retention-purge: no retention config on collection: ${collectionSlug}`,
+    );
+  }
+
+  const taskSlug = `retention-purge--${collectionSlug}`;
+
+  return async () => {
+    const payload = await getPayload({ config });
+    const now = Date.now();
+
+    if (retention.activeRetention) {
+      const { duration, trigger } = retention.activeRetention;
+      const retentionMs = parseDurationMs(duration);
+      const cutoff = new Date(now - retentionMs).toISOString();
+      const dateField = trigger === "from-creation" ? "createdAt" : "updatedAt";
+
+      const { docs } = await payload.find({
+        collection: collectionSlug,
+        where: { [dateField]: { less_than: cutoff } },
+        limit: 1000,
+        overrideAccess: true,
+      });
+
+      const action = retention.postDeletion?.action ?? "hard-delete";
+
+      for (const doc of docs) {
+        const id = doc["id"] as string | number;
+
+        if (action === "pseudonymize") {
+          const piiFields: Record<string, null> = {};
+          for (const field of collection.fields) {
+            const f = field as { name?: string; custom?: { pii?: unknown } };
+            if (f.name && f.custom?.pii) {
+              piiFields[f.name] = null;
+            }
+          }
+          if (Object.keys(piiFields).length > 0) {
+            await payload.update({
+              collection: collectionSlug,
+              id,
+              data: piiFields,
+              overrideAccess: true,
+            });
+          }
+        } else {
+          await payload.delete({
+            collection: collectionSlug,
+            id,
+            overrideAccess: true,
+          });
+        }
+
+        if (auditLog) {
+          await auditLog.record({
+            actorId: "system",
+            actorType: "system",
+            actorRoles: [],
+            action: "DELETE",
+            resource: { type: collectionSlug, id: String(id) },
+            at: new Date(),
+            scope: {
+              feature: "core-shared",
+              environment: process.env["NODE_ENV"] ?? "production",
+              tenant: "default",
+            },
+            reason: "retention-policy",
+            from: { ipTruncated: "system", userAgent: "background-job" },
+            containsPii: false,
+            outcome: "success",
+          });
+        }
+      }
+    }
+
+    const delay = scheduleDelayMs(retention.purgeSchedule);
+    await queue.enqueue(taskSlug, {}, { runAt: new Date(now + delay) });
+  };
+}
+
+/**
+ * Walk all Payload collections that declare `custom.retention.purgeSchedule`
+ * and schedule the first purge run for each via the provided IJobQueue.
+ *
+ * Call once at app startup (inside bindAll or equivalent). Idempotent per
+ * queue implementation — duplicate enqueues are the queue's responsibility.
+ */
+export async function registerRetentionPurgeJobs(
+  deps: RetentionPurgeJobDeps,
+): Promise<void> {
+  const { queue, config } = deps;
+  const now = Date.now();
+
+  for (const collection of config.collections) {
+    const retention = collection.custom?.retention;
+    if (!retention?.purgeSchedule) continue;
+
+    const taskSlug = `retention-purge--${collection.slug}`;
+    const delay = scheduleDelayMs(retention.purgeSchedule);
+    await queue.enqueue(taskSlug, {}, { runAt: new Date(now + delay) });
+  }
+}