feat(jobs): add retention prune for stored Coach history

coach_messages is append-only and grows forever — a daily Coach user
accrues encrypted conversation rows with no lifecycle, while every
sibling ledger (push_attempts, audit_logs, host_metrics) already prunes.
Add a daily pg-boss cleanup that hard-deletes rows past a configurable
window (COACH_MESSAGE_RETENTION_DAYS, default 365, 30-day floor so a
misconfig cannot drop live in-context history). Wire it into the
maintenance registrar with the dead-queue contract guard and a focused
helper test.
This commit is contained in:
Marc-André Bombeck committed 2026-06-19 20:54:56 +02:00
1 parent b3d303217c
commit 0278e22ed1
5 files changed
+242

No files matched your search

@@ -0,0 +1,56 @@
/**
* v1.18.7 — coach-message-cleanup queue registration guard.
*
* Same source-text-grep approach as the other queue-wiring guards: assert the
* queue is registered in `allQueues`, scheduled, and wired to a `boss.work`
* handler — without booting pg-boss + Prisma. An unregistered queue silently
* never drains (the recurring v1.4.37 dead-queue bug), which would leave the
* encrypted Coach history growing unbounded with the retention job dark.
*/
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { describe, expect, it } from "vitest";
const REGISTRAR_PATH = join(
__dirname,
"..",
"reminder",
"register-maintenance.ts",
);
const workerSource =
readFileSync(REGISTRAR_PATH, "utf8") +
readFileSync(join(__dirname, "..", "reminder", "cleanup-handlers.ts"), "utf8");
describe("reminder-worker — coach-message-cleanup wiring", () => {
it("imports the cleanup handler from the cleanup-handlers module", () => {
expect(workerSource).toMatch(/\bhandleCoachMessageCleanup\b/);
expect(workerSource).toMatch(/\bCoachMessageCleanupPayload\b/);
});
it("registers the coach-message-cleanup queue in the allQueues loop", () => {
const allQueuesMatch = workerSource.match(
/const allQueues\s*=\s*\[([\s\S]*?)\];/,
);
expect(allQueuesMatch).not.toBeNull();
expect(allQueuesMatch![1]).toMatch(/\bCOACH_MESSAGE_CLEANUP_QUEUE\b/);
});
it("schedules the coach-message-cleanup cron", () => {
expect(workerSource).toMatch(
/\[COACH_MESSAGE_CLEANUP_QUEUE,\s*COACH_MESSAGE_CLEANUP_CRON\]/,
);
});
it("registers a boss.work handler for the queue", () => {
expect(workerSource).toMatch(
/boss\.work[\s\S]{0,200}COACH_MESSAGE_CLEANUP_QUEUE[\s\S]{0,200}handleCoachMessageCleanup/,
);
});
it("prunes coach messages inside the handler", () => {
expect(workerSource).toMatch(
/handleCoachMessageCleanup[\s\S]{0,400}cleanupOldCoachMessages/,
);
});
});
@@ -0,0 +1,90 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import type { PrismaClient } from "@/generated/prisma/client";
import {
cleanupOldCoachMessages,
getCoachMessageRetentionDays,
DEFAULT_COACH_MESSAGE_RETENTION_DAYS,
} from "../coach-message-cleanup";
function makePrismaMock(deletedCount: number) {
return {
coachMessage: {
deleteMany: vi.fn().mockResolvedValue({ count: deletedCount }),
},
} as unknown as PrismaClient;
}
const ORIGINAL_ENV = { ...process.env };
beforeEach(() => {
process.env = { ...ORIGINAL_ENV };
delete process.env.COACH_MESSAGE_RETENTION_DAYS;
});
afterEach(() => {
process.env = ORIGINAL_ENV;
vi.restoreAllMocks();
vi.useRealTimers();
});
describe("getCoachMessageRetentionDays", () => {
it("returns 365 when env is unset", () => {
expect(getCoachMessageRetentionDays()).toBe(
DEFAULT_COACH_MESSAGE_RETENTION_DAYS,
);
});
it("respects a valid override", () => {
process.env.COACH_MESSAGE_RETENTION_DAYS = "180";
expect(getCoachMessageRetentionDays()).toBe(180);
});
it("ignores nonsensical values (NaN, 0, negative)", () => {
process.env.COACH_MESSAGE_RETENTION_DAYS = "0";
expect(getCoachMessageRetentionDays()).toBe(
DEFAULT_COACH_MESSAGE_RETENTION_DAYS,
);
process.env.COACH_MESSAGE_RETENTION_DAYS = "-100";
expect(getCoachMessageRetentionDays()).toBe(
DEFAULT_COACH_MESSAGE_RETENTION_DAYS,
);
process.env.COACH_MESSAGE_RETENTION_DAYS = "not a number";
expect(getCoachMessageRetentionDays()).toBe(
DEFAULT_COACH_MESSAGE_RETENTION_DAYS,
);
});
it("rejects too-short retention (< 30 days) to protect live history", () => {
process.env.COACH_MESSAGE_RETENTION_DAYS = "7";
expect(getCoachMessageRetentionDays()).toBe(
DEFAULT_COACH_MESSAGE_RETENTION_DAYS,
);
});
});
describe("cleanupOldCoachMessages", () => {
it("deletes coach messages older than 365 days by default", async () => {
const prisma = makePrismaMock(42);
const now = new Date("2026-05-04T00:00:00Z");
const deleted = await cleanupOldCoachMessages(prisma, now);
expect(deleted).toBe(42);
expect(prisma.coachMessage.deleteMany).toHaveBeenCalledTimes(1);
const cutoff = new Date(now.getTime() - 365 * 86_400_000);
expect(prisma.coachMessage.deleteMany).toHaveBeenCalledWith({
where: { createdAt: { lt: cutoff } },
});
});
it("uses the configured retention", async () => {
process.env.COACH_MESSAGE_RETENTION_DAYS = "90";
const prisma = makePrismaMock(7);
const now = new Date("2026-05-04T00:00:00Z");
await cleanupOldCoachMessages(prisma, now);
const cutoff = new Date(now.getTime() - 90 * 86_400_000);
expect(prisma.coachMessage.deleteMany).toHaveBeenCalledWith({
where: { createdAt: { lt: cutoff } },
});
});
});
+50
View File
@@ -0,0 +1,50 @@
/**
* Daily cleanup for the `coach_messages` table.
*
* v1.18.7 retention (GDPR Art. 5(1)(e) "storage limitation"): the Coach
* persists every turn's prose as an encrypted `encryptedContent` Bytes
* column. The table is append-only and grows forever — a daily Coach user
* accrues rows indefinitely with no lifecycle. The conversation history is
* useful context, but stale beyond a year it is dead weight inflating the
* table and the `(conversation_id, created_at)` indexes.
*
* Default retention is 365 days (configurable via COACH_MESSAGE_RETENTION_DAYS
* env). Rows older than the cutoff are deleted in a single bulk `deleteMany`;
* runs daily via pg-boss. The DELETE is keyed on `created_at` directly — the
* pre-existing `(conversation_id, created_at)` index is not a perfect cover
* for an unqualified `created_at` scan, but the trailing-edge delete touches
* only the oldest sliver of the table.
*
* The retention floor is 30 days: a misconfigured tiny value never nukes
* recent, in-context conversation history. Content stays encrypted at rest
* until the moment it is hard-deleted; there is no plaintext path.
*/
import type { PrismaClient } from "@/generated/prisma/client";
export const DEFAULT_COACH_MESSAGE_RETENTION_DAYS = 365;
export function getCoachMessageRetentionDays(): number {
const raw = process.env.COACH_MESSAGE_RETENTION_DAYS;
if (raw === undefined) return DEFAULT_COACH_MESSAGE_RETENTION_DAYS;
const parsed = parseInt(raw, 10);
if (!Number.isFinite(parsed) || parsed <= 0) {
return DEFAULT_COACH_MESSAGE_RETENTION_DAYS;
}
// Refuse a too-short retention window. Coach history is in-context for the
// proactive nudge frequency gate and the conversation rail; 30 days is the
// safe floor so a misconfig measured in seconds cannot prune live history.
if (parsed < 30) return DEFAULT_COACH_MESSAGE_RETENTION_DAYS;
return parsed;
}
export async function cleanupOldCoachMessages(
prisma: PrismaClient,
now: Date = new Date(),
): Promise<number> {
const days = getCoachMessageRetentionDays();
const cutoff = new Date(now.getTime() - days * 86_400_000);
const { count } = await prisma.coachMessage.deleteMany({
where: { createdAt: { lt: cutoff } },
});
return count;
}
+20
View File
@@ -12,6 +12,7 @@ import {
} from "@/lib/jobs/whoop-oauth-state-cleanup";
import { cleanupExpiredIdempotencyKeys } from "@/lib/jobs/idempotency-cleanup";
import { cleanupOldAuditLogs } from "@/lib/jobs/audit-log-cleanup";
import { cleanupOldCoachMessages } from "@/lib/jobs/coach-message-cleanup";
import { cleanupExpiredWithingsOAuthStates } from "@/lib/jobs/withings-oauth-state-cleanup";
import {
cleanupExpiredMeasurementTombstones,
@@ -49,6 +50,10 @@ export interface AuditLogCleanupPayload {
triggeredAt: string;
}
export interface CoachMessageCleanupPayload {
triggeredAt: string;
}
export interface WithingsOAuthStateCleanupPayload {
triggeredAt: string;
}
@@ -201,6 +206,21 @@ export async function handleAuditLogCleanup(
});
}
export async function handleCoachMessageCleanup(
jobs: Job<CoachMessageCleanupPayload>[],
) {
void jobs;
await withBackgroundEvent("job.coach_message_cleanup", async (evt) => {
const p = getWorkerPrisma();
try {
const deleted = await cleanupOldCoachMessages(p);
evt.addMeta("coach_message_cleanup_deleted", deleted);
} catch (err) {
evt.addWarning(`coach-message-cleanup failed: ${err}`);
}
});
}
export async function handleWithingsOAuthStateCleanup(
jobs: Job<WithingsOAuthStateCleanupPayload>[],
) {
@@ -77,6 +77,8 @@ import {
handleRateLimitCleanup,
handleIdempotencyCleanup,
handleAuditLogCleanup,
CoachMessageCleanupPayload,
handleCoachMessageCleanup,
} from "./cleanup-handlers";
import {
HostMetricSamplePayload,
@@ -156,6 +158,16 @@ const PUSH_ATTEMPT_CLEANUP_CRON = "35 3 * * *";
const MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE = "measurement-tombstone-cleanup";
const MEASUREMENT_TOMBSTONE_CLEANUP_CRON = "40 3 * * *";
// v1.18.7 — daily prune for the encrypted Coach conversation history.
// coach_messages is append-only and grows forever; rows older than the
// configurable retention window (default 365 days) are hard-deleted.
// Slots at 03:50 Europe/Berlin after the measurement-tombstone cleanup
// (03:40) and the cumulative drain (03:45), staying inside the existing
// 03:xx maintenance window.
const COACH_MESSAGE_CLEANUP_QUEUE = "coach-message-cleanup";
const COACH_MESSAGE_CLEANUP_CRON = "50 3 * * *";
const allQueues = [
DATA_BACKUP_QUEUE,
@@ -193,6 +205,10 @@ const allQueues = [
// the daily schedule silently no-ops and pruned-past-retention tombstones
// pile up forever.
MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE,
// v1.18.7 — Coach conversation-history retention prune. Without this entry
// the daily schedule silently no-ops and the encrypted coach_messages
// table grows unbounded.
COACH_MESSAGE_CLEANUP_QUEUE,
];
const schedules: ScheduleEntry[] = [
@@ -242,6 +258,8 @@ const schedules: ScheduleEntry[] = [
// v1.7.0 — daily 03:40 Europe/Berlin prune for expired measurement
// tombstones.
[MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE, MEASUREMENT_TOMBSTONE_CLEANUP_CRON],
// v1.18.7 — daily 03:50 Europe/Berlin prune for stale Coach history.
[COACH_MESSAGE_CLEANUP_QUEUE, COACH_MESSAGE_CLEANUP_CRON],
];
/**
@@ -323,6 +341,14 @@ export async function registerMaintenanceQueues(
{ localConcurrency: 1 },
handleMeasurementTombstoneCleanup,
);
// v1.18.7 — daily prune of the encrypted Coach conversation history.
// Single-flight like every other cleanup queue; two ticks racing on the
// same DELETE is wasted work and the second is a no-op.
await boss.work<CoachMessageCleanupPayload>(
COACH_MESSAGE_CLEANUP_QUEUE,
{ localConcurrency: 1 },
handleCoachMessageCleanup,
);
await boss.work<PrDetectionPayload>(
PR_DETECTION_QUEUE,
{ localConcurrency: PR_DETECTION_CONCURRENCY },