Medformatik · Oct 31, 2026 · Oct 31, 2026
@@ -232,6 +232,7 @@ PASSKEY_ORIGIN=http://127.0.0.1:3000,openmapx://
 # apps/api/src/server.ts. Set lower for tight-disk deployments, higher to
 # keep more history. In-flight jobs are never pruned.
 # AUDIT_LOG_RETENTION_DAYS=90
+# Finished admin jobs and data-manager pipeline runs older than this are pruned.
 # ADMIN_JOB_RETENTION_DAYS=30
 
 # Isochrone provider override (default: valhalla). Use "otp" to route

@@ -71,6 +71,31 @@ describe("privacy preservation", () => {
     expect(query.params).toEqual(["admin-audit-attribution"]);
   });
 
+  it("also keeps rows through the statuses a caller names", () => {
+    const query = dialect.sqlToQuery(
+      preservationAllowsDeletion({
+        registrationId: "data-manager-trigger-attribution",
+        candidateTimestamp: sql.identifier("started_at"),
+        subjectIds: [sql.identifier("subject")],
+        blockingStatuses: ["held", "action_required"],
+      }),
+    );
+
+    expect(query.sql).toContain("preservation.status in ('held', 'action_required')");
+    expect(query.params).toEqual(["data-manager-trigger-attribution"]);
+  });
+
+  it("refuses a guard that no hold status could block", () => {
+    expect(() =>
+      preservationAllowsDeletion({
+        registrationId: "data-manager-trigger-attribution",
+        candidateTimestamp: sql.identifier("started_at"),
+        subjectIds: [sql.identifier("subject")],
+        blockingStatuses: [],
+      }),
+    ).toThrow("requires a blocking hold status");
+  });
+
   it("takes the transaction lock in a separate statement before retention work", async () => {
     const order: string[] = [];
     const transaction = {

@@ -41,6 +41,19 @@ export function preservationReleaseTransition(
     : { status: "released", released: true };
 }
 
+export type PreservationBlockingStatus = "held" | "action_required";
+
+const BLOCKING_STATUS_LITERALS: Record<PreservationBlockingStatus, string> = {
+  held: "'held'",
+  action_required: "'action_required'",
+};
+
+/** The closed set of statuses renders as fixed literals, never caller text. */
+function blockingStatusCondition(statuses: readonly PreservationBlockingStatus[]): SQL {
+  const literals = [...new Set(statuses)].map((status) => BLOCKING_STATUS_LITERALS[status]);
+  return sql.raw(literals.length === 1 ? `= ${literals[0]}` : `in (${literals.join(", ")})`);
+}
+
 /**
  * SQL guard for a source retention DELETE. Keeping the hold check in the same
  * statement as the delete prevents a capture/release worker from creating a
@@ -54,9 +67,18 @@ export function preservationAllowsDeletion(input: {
   registrationId: string;
   candidateTimestamp: SQLWrapper;
   subjectIds: readonly SQLWrapper[];
+  /**
+   * Hold statuses that keep the rows. An `operator_task` registration's hold
+   * becomes `action_required` while the operator still has to preserve the
+   * rows by hand, so its source must also stay through that status.
+   */
+  blockingStatuses?: readonly PreservationBlockingStatus[];
 }): SQL {
   if (input.subjectIds.length === 0)
     throw new Error("A preservation deletion guard requires an exact subject locator");
+  const blockingStatuses = input.blockingStatuses ?? ["held"];
+  if (blockingStatuses.length === 0)
+    throw new Error("A preservation deletion guard requires a blocking hold status");
   const subjectMatch = sql.join(
     input.subjectIds.map((subjectId) => sql`preservation_request.user_id = ${subjectId}`),
     sql` or `,
@@ -67,7 +89,7 @@ export function preservationAllowsDeletion(input: {
     inner join data_subject_request preservation_request
       on preservation_request.id = preservation.request_id
     where preservation.registration_id = ${input.registrationId}
-      and preservation.status = 'held'
+      and preservation.status ${blockingStatusCondition(blockingStatuses)}
       and preservation.released_at is null
       and preservation.source_cutoff_at >= ${input.candidateTimestamp}
       and preservation_request.user_id is not null

@@ -79,6 +79,7 @@ import {
   pruneAppLogs,
   pruneAuditLog,
   pruneCompletedJobs,
+  pruneDataManagerJobs,
   pruneExpiredVerifications,
 } from "./services/activity-retention";
 import {
@@ -644,6 +645,9 @@ const runRetention = () => {
   void pruneCompletedJobs(JOB_RETENTION_DAYS).catch((err) =>
     server.log.warn(err, "Admin job prune failed"),
   );
+  void pruneDataManagerJobs(JOB_RETENTION_DAYS).catch((err) =>
+    server.log.warn(err, "Data-manager job prune failed"),
+  );
   void pruneAppLogs(APP_LOG_RETENTION_DAYS).catch((err) =>
     server.log.warn(err, "Application log prune failed"),
   );

@@ -12,6 +12,13 @@ vi.mock("../../db/schema", () => ({
   },
   adminJob: { id: "jobId", createdBy: "jobCreatedBy", finishedAt: "jobFinishedAt" },
   appLog: { id: "appLogId", metadata: "appLogMetadata", createdAt: "appLogCreatedAt" },
+  dataManagerJobs: {
+    id: "dataManagerJobId",
+    kind: "dataManagerJobKind",
+    startedAt: "dataManagerJobStartedAt",
+    finishedAt: "dataManagerJobFinishedAt",
+    triggeredBy: "dataManagerJobTriggeredBy",
+  },
   verification: { id: "verificationId", expiresAt: "verificationExpiresAt" },
 }));
 vi.mock("drizzle-orm", async (importOriginal) => {
@@ -25,6 +32,7 @@ const {
   pruneAppLogs,
   pruneAuditLog,
   pruneCompletedJobs,
+  pruneDataManagerJobs,
   pruneExpiredVerifications,
 } = await import("../activity-retention.js");
 
@@ -100,6 +108,24 @@ describe("pruneAuditLog / pruneCompletedJobs retention guard", () => {
   });
 });
 
+// Its deletion against PostgreSQL is data-manager-job-retention-postgres.test.ts's.
+describe("pruneDataManagerJobs retention guard", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.spyOn(console, "warn").mockImplementation(() => {});
+  });
+  afterEach(() => vi.restoreAllMocks());
+
+  it.each([0, -5, Number.NaN, Number.POSITIVE_INFINITY])(
+    "days = %s is a no-op guard",
+    async (days) => {
+      expect(await pruneDataManagerJobs(days)).toBe(0);
+      expect(dbMock.db.delete).not.toHaveBeenCalled();
+      expect(console.warn).toHaveBeenCalled();
+    },
+  );
+});
+
 describe("parseRequiredRetentionDays", () => {
   it("uses the documented fallback when the setting is absent", () => {
     expect(parseRequiredRetentionDays("BACKUP_RETENTION_DAYS", 30, {})).toBe(30);

@@ -0,0 +1,157 @@
+import { randomUUID } from "node:crypto";
+import { and, eq, inArray, like } from "drizzle-orm";
+import { afterEach, describe, expect, it } from "vitest";
+import { db } from "../../db/index.js";
+import {
+  dataManagerJobStages,
+  dataManagerJobs,
+  dataSubjectRequest,
+  dataSubjectRequestPreservation,
+  user,
+} from "../../db/schema.js";
+import { loadMasterKeyRing } from "../../privacy/crypto.js";
+import { releasePreservation } from "../../privacy/preservation.js";
+import { PrivacyRequestService } from "../../privacy/request-service.js";
+import { pruneDataManagerJobs } from "../activity-retention.js";
+
+const DAY = 86_400_000;
+const users: string[] = [];
+const requests: string[] = [];
+
+afterEach(async () => {
+  await db.delete(dataManagerJobs).where(like(dataManagerJobs.kind, "retention-fixture-%"));
+  for (const id of requests.splice(0))
+    await db.delete(dataSubjectRequest).where(eq(dataSubjectRequest.id, id));
+  for (const id of users.splice(0)) await db.delete(user).where(eq(user.id, id));
+});
+
+interface Fixture {
+  status?: string;
+  daysAgo: number;
+  triggeredBy?: string;
+  finished?: boolean;
+}
+
+async function insertJob(kind: string, fixture: Fixture): Promise<string> {
+  const startedAt = new Date(Date.now() - fixture.daysAgo * DAY);
+  const [row] = await db
+    .insert(dataManagerJobs)
+    .values({
+      kind,
+      status: fixture.status ?? "ok",
+      startedAt,
+      finishedAt: fixture.finished === false ? null : new Date(startedAt.getTime() + 60_000),
+      triggeredBy: fixture.triggeredBy ?? "cron:data-manager-cron",
+    })
+    .returning({ id: dataManagerJobs.id });
+  await db.insert(dataManagerJobStages).values({
+    jobId: row.id,
+    stage: "fetch",
+    status: "ok",
+    startedAt,
+    finishedAt: startedAt,
+    durationMs: 1,
+  });
+  return row.id;
+}
+
+async function remaining(ids: string[]): Promise<string[]> {
+  const rows = await db
+    .select({ id: dataManagerJobs.id })
+    .from(dataManagerJobs)
+    .where(inArray(dataManagerJobs.id, ids));
+  return rows.map((row) => row.id).sort();
+}
+
+describe.skipIf(process.env.OPENMAPX_RUN_DATABASE_TESTS !== "1")(
+  "data-manager job retention on PostgreSQL",
+  () => {
+    it("deletes finished jobs past retention with their stages and keeps the rest", async () => {
+      const kind = `retention-fixture-${randomUUID()}`;
+      const otherKind = `retention-fixture-${randomUUID()}`;
+      const expired = await insertJob(kind, { daysAgo: 40 });
+      const interrupted = await insertJob(kind, { daysAgo: 35, status: "interrupted" });
+      const running = await insertJob(kind, { daysAgo: 45, status: "running", finished: false });
+      const newest = await insertJob(kind, { daysAgo: 2 });
+      // A rarely run kind keeps its only run, however old.
+      const onlyRunOfOtherKind = await insertJob(otherKind, { daysAgo: 90 });
+
+      expect(await pruneDataManagerJobs(30)).toBeGreaterThanOrEqual(2);
+
+      expect(await remaining([expired, interrupted, running, newest, onlyRunOfOtherKind])).toEqual(
+        [running, newest, onlyRunOfOtherKind].sort(),
+      );
+      const stages = await db
+        .select({ jobId: dataManagerJobStages.jobId })
+        .from(dataManagerJobStages)
+        .where(inArray(dataManagerJobStages.jobId, [expired, interrupted]));
+      expect(stages).toEqual([]);
+    });
+
+    it("keeps the jobs an account triggered while its data-subject request needs them", async () => {
+      const userId = randomUUID();
+      users.push(userId);
+      await db.insert(user).values({
+        id: userId,
+        name: "Retention fixture",
+        email: `${userId}@example.test`,
+        updatedAt: new Date(),
+      });
+      const ring = loadMasterKeyRing({
+        env: {
+          NODE_ENV: "development",
+          OPENMAPX_EXPORTS_KEY: Buffer.alloc(32, 9).toString("base64url"),
+        },
+      });
+      const service = new PrivacyRequestService({ database: db, keyRing: ring });
+      const request = await service.create({
+        userId,
+        kind: "access",
+        channel: "self_service",
+        locale: "en",
+        timeZone: "UTC",
+      });
+      requests.push(request.id);
+
+      const kind = `retention-fixture-${randomUUID()}`;
+      const heldApi = await insertJob(kind, { daysAgo: 40, triggeredBy: `api:user:${userId}` });
+      const heldManual = await insertJob(kind, {
+        daysAgo: 39,
+        triggeredBy: `manual:admin:${userId}`,
+      });
+      const otherUser = await insertJob(kind, {
+        daysAgo: 38,
+        triggeredBy: `api:user:${randomUUID()}`,
+      });
+      const scheduled = await insertJob(kind, { daysAgo: 37 });
+      const newest = await insertJob(kind, { daysAgo: 1 });
+
+      await pruneDataManagerJobs(30);
+      expect(await remaining([heldApi, heldManual, otherUser, scheduled, newest])).toEqual(
+        [heldApi, heldManual, newest].sort(),
+      );
+
+      const holds = await db
+        .select({
+          id: dataSubjectRequestPreservation.id,
+          status: dataSubjectRequestPreservation.status,
+        })
+        .from(dataSubjectRequestPreservation)
+        .where(
+          and(
+            eq(dataSubjectRequestPreservation.requestId, request.id),
+            eq(dataSubjectRequestPreservation.registrationId, "data-manager-trigger-attribution"),
+          ),
+        );
+      // The operator still has to preserve this registration's rows by hand.
+      expect(holds).toEqual([{ id: holds[0]?.id, status: "action_required" }]);
+      await pruneDataManagerJobs(30);
+      expect(await remaining([heldApi, heldManual, newest])).toHaveLength(3);
+
+      await releasePreservation({ id: holds[0].id, database: db });
+
+      await pruneDataManagerJobs(30);
+      expect(await remaining([heldApi, heldManual, newest])).toEqual([newest]);
+    });
+  },
+);
@@ -17,6 +17,7 @@ describe("activity job normalization", () => {
     ["manual:admin:user-2", "user-2", "Administrator"],
     ["cron:data-manager-cron", null, "Scheduled"],
     ["bootstrap", null, "Bootstrap"],
+    ["cron:data-manager-auto-bump", null, "Automatic version bump"],
     ["data-manager-auto-bump", null, "Automatic version bump"],
     [null, null, "System"],
   ])("parses data-manager actor %s", (triggeredBy, userId, label) => {

@@ -74,7 +74,8 @@ export function parseDataManagerActor(triggeredBy: string | null): {
     return { userId: null, label: "Scheduled" };
   }
   if (triggeredBy === "bootstrap") return { userId: null, label: "Bootstrap" };
-  if (triggeredBy === "data-manager-auto-bump") {
+  // The data-manager stores its scheduled runs as `cron:<scheduler>`.
+  if (triggeredBy === "data-manager-auto-bump" || triggeredBy === "cron:data-manager-auto-bump") {
     return { userId: null, label: "Automatic version bump" };
   }
 

@@ -1,6 +1,6 @@
-import { and, lt, sql } from "drizzle-orm";
+import { and, desc, isNotNull, lt, notInArray, sql } from "drizzle-orm";
 import { db } from "../db";
-import { adminAuditLog, adminJob, appLog, verification } from "../db/schema";
+import { adminAuditLog, adminJob, appLog, dataManagerJobs, verification } from "../db/schema";
 import {
   preservationAllowsDeletion,
   withPreservationRetentionLock,
@@ -84,6 +84,52 @@ export async function pruneCompletedJobs(days: number): Promise<number> {
   return result.length;
 }
 
+/**
+ * The account a data-manager job's `triggered_by` names, in the formats
+ * `parseDataManagerActor` reads; NULL for scheduled and other system runs.
+ */
+const dataManagerJobSubject = sql`substring(${dataManagerJobs.triggeredBy} from '^(?:api:user:|manual:admin:|manual:user:)(.+)$')`;
+
+/**
+ * Delete finished data-manager pipeline jobs, and by cascade their stages,
+ * that finished more than `days` ago. A running job has no finish time and is
+ * never deleted. The newest finished job of each kind is kept, so the transit
+ * state still reports the last sync when the schedule runs rarely or is off.
+ */
+export async function pruneDataManagerJobs(days: number): Promise<number> {
+  if (!Number.isFinite(days) || days <= 0) {
+    console.warn(
+      `Skipping data-manager job prune: retention days must be a positive finite number, got ${days}`,
+    );
+    return 0;
+  }
+  const cutoff = new Date(Date.now() - days * MS_PER_DAY);
+  const result = await withPreservationRetentionLock(db, (tx) => {
+    const newestFinishedOfEachKind = tx
+      .selectDistinctOn([dataManagerJobs.kind], { id: dataManagerJobs.id })
+      .from(dataManagerJobs)
+      .where(isNotNull(dataManagerJobs.finishedAt))
+      .orderBy(dataManagerJobs.kind, desc(dataManagerJobs.startedAt));
+    return tx
+      .delete(dataManagerJobs)
+      .where(
+        and(
+          lt(dataManagerJobs.finishedAt, cutoff),
+          notInArray(dataManagerJobs.id, newestFinishedOfEachKind),
+          preservationAllowsDeletion({
+            registrationId: "data-manager-trigger-attribution",
+            candidateTimestamp: dataManagerJobs.startedAt,
+            subjectIds: [dataManagerJobSubject],
+            // An operator preserves this registration's rows by hand.
+            blockingStatuses: ["held", "action_required"],
+          }),
+        ),
+      )
+      .returning({ id: dataManagerJobs.id });
+  });
+  return result.length;
+}
+
 /** Delete persisted operational log records after the disclosed retention period. */
 export async function pruneAppLogs(days: number): Promise<number> {
   if (!Number.isFinite(days) || days <= 0) {