feat(jobs): daily cleanup job for expired idempotency_keys (#120)

`withIdempotency()` only purges rows lazily on the next lookup with
the same `(userId, key, method, path)` tuple — most rows never see a
retry, so without a sweeper the table grows unbounded with stale
24h-expired entries (Security-Review HIGH-2 from PR #117).

Adds a pg-boss queue that runs daily at 03:00 Europe/Berlin (parallel
to the weekly Sunday data-backup job) and bulk-deletes expired rows.
Outcome surfaces as a Wide Event annotation
`idempotency_cleanup_deleted` so the run is queryable in Loki.

Helper extracted to `src/lib/jobs/idempotency-cleanup.ts` for clean
unit testing without mocking pg-boss internals.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Marc-André BombeckandClaude Opus 4.7 authored and GitHub committed 2026-05-05 20:30:30 +02:00
1 parent 79f0e5a06f
commit b76d579717
3 files changed
+94

No files matched your search

@@ -0,0 +1,47 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import type { PrismaClient } from "@/generated/prisma/client";
import { cleanupExpiredIdempotencyKeys } from "../idempotency-cleanup";
function makePrismaMock(deletedCount: number) {
return {
idempotencyKey: {
deleteMany: vi.fn().mockResolvedValue({ count: deletedCount }),
},
} as unknown as PrismaClient;
}
describe("cleanupExpiredIdempotencyKeys", () => {
beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(new Date("2026-05-04T03:00:00Z"));
});
it("deletes all rows whose expires_at is in the past and returns the count", async () => {
const prisma = makePrismaMock(100);
const deleted = await cleanupExpiredIdempotencyKeys(prisma);
expect(deleted).toBe(100);
expect(prisma.idempotencyKey.deleteMany).toHaveBeenCalledTimes(1);
expect(prisma.idempotencyKey.deleteMany).toHaveBeenCalledWith({
where: { expiresAt: { lt: new Date("2026-05-04T03:00:00Z") } },
});
});
it("returns 0 when no rows are expired", async () => {
const prisma = makePrismaMock(0);
const deleted = await cleanupExpiredIdempotencyKeys(prisma);
expect(deleted).toBe(0);
});
it("propagates errors from prisma so the handler can warn", async () => {
const prisma = {
idempotencyKey: {
deleteMany: vi.fn().mockRejectedValue(new Error("connection lost")),
},
} as unknown as PrismaClient;
await expect(cleanupExpiredIdempotencyKeys(prisma)).rejects.toThrow(
"connection lost",
);
});
});
+18
View File
@@ -0,0 +1,18 @@
/**
* Daily cleanup for the `idempotency_keys` table.
*
* `withIdempotency()` (src/lib/idempotency.ts) only purges a row lazily
* on the next lookup with the same `(userId, key, method, path)` tuple
* — most rows never see another retry, so without a sweeper the table
* grows unbounded with stale 24h-expired entries.
*/
import type { PrismaClient } from "@/generated/prisma/client";
export async function cleanupExpiredIdempotencyKeys(
prisma: PrismaClient,
): Promise<number> {
const { count } = await prisma.idempotencyKey.deleteMany({
where: { expiresAt: { lt: new Date() } },
});
return count;
}
+29
View File
@@ -27,6 +27,7 @@ import {
recordError,
} from "@/lib/jobs/worker-status";
import { setGlobalBoss } from "@/lib/jobs/boss-instance";
import { cleanupExpiredIdempotencyKeys } from "@/lib/jobs/idempotency-cleanup";
import { deleteMessage } from "@/lib/telegram";
import { decrypt, encrypt } from "@/lib/crypto";
import { syncMoodLogEntries } from "@/lib/moodlog/sync";
@@ -90,6 +91,8 @@ const DATA_BACKUP_QUEUE = "data-backup";
const DATA_BACKUP_CRON = "0 3 * * 0"; // weekly Sunday at 03:00
const RATE_LIMIT_CLEANUP_QUEUE = "rate-limit-cleanup";
const RATE_LIMIT_CLEANUP_CRON = "*/5 * * * *"; // every 5 minutes
const IDEMPOTENCY_CLEANUP_QUEUE = "idempotency-cleanup";
const IDEMPOTENCY_CLEANUP_CRON = "0 3 * * *"; // daily at 03:00 (Europe/Berlin)
interface ReminderCheckPayload {
triggeredAt: string;
@@ -141,6 +144,10 @@ interface RateLimitCleanupPayload {
triggeredAt: string;
}
interface IdempotencyCleanupPayload {
triggeredAt: string;
}
// Re-export timezone utilities under local names for backward compatibility
const getUserTodayBounds = getUserTodayBoundsUtil;
const getDayOfWeekInTz = getDayOfWeekInTzUtil;
@@ -818,6 +825,21 @@ async function handleRateLimitCleanup(
});
}
async function handleIdempotencyCleanup(
jobs: Job<IdempotencyCleanupPayload>[],
) {
void jobs;
await withBackgroundEvent("job.idempotency_cleanup", async (evt) => {
const p = getWorkerPrisma();
try {
const deleted = await cleanupExpiredIdempotencyKeys(p);
evt.addMeta("idempotency_cleanup_deleted", deleted);
} catch (err) {
evt.addWarning(`idempotency-cleanup failed: ${err}`);
}
});
}
async function handleDataBackup(jobs: Job<DataBackupPayload>[]) {
void jobs;
await withBackgroundEvent("job.data_backup", async (evt) => {
@@ -998,6 +1020,7 @@ export async function startReminderWorker() {
MOODLOG_SYNC_QUEUE,
DATA_BACKUP_QUEUE,
RATE_LIMIT_CLEANUP_QUEUE,
IDEMPOTENCY_CLEANUP_QUEUE,
];
for (const q of allQueues) {
@@ -1017,6 +1040,7 @@ export async function startReminderWorker() {
[MOODLOG_SYNC_QUEUE, MOODLOG_SYNC_CRON],
[DATA_BACKUP_QUEUE, DATA_BACKUP_CRON],
[RATE_LIMIT_CLEANUP_QUEUE, RATE_LIMIT_CLEANUP_CRON],
[IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON],
];
for (const [name, cron] of schedules) {
@@ -1084,6 +1108,11 @@ export async function startReminderWorker() {
{ localConcurrency: 1 },
handleRateLimitCleanup,
);
await boss.work<IdempotencyCleanupPayload>(
IDEMPOTENCY_CLEANUP_QUEUE,
{ localConcurrency: 1 },
handleIdempotencyCleanup,
);
return boss;
}