diff --git a/src/lib/jobs/__tests__/pr-detection.test.ts b/src/lib/jobs/__tests__/pr-detection.test.ts new file mode 100644 index 000000000..e00b849ed --- /dev/null +++ b/src/lib/jobs/__tests__/pr-detection.test.ts @@ -0,0 +1,53 @@ +/** + * `enqueuePrDetection` is a thin wrapper around the global boss + * singleton. We assert two behaviours: + * 1. No boss attached → the call is a silent no-op (route paths in + * test environments don't run a worker process). + * 2. Boss attached → `boss.send` receives the expected payload with + * the `silent` flag propagated through. + */ +import { describe, expect, it, vi, beforeEach } from "vitest"; + +import { setGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + PR_DETECTION_QUEUE, + enqueuePrDetection, +} from "@/lib/jobs/pr-detection"; + +describe("enqueuePrDetection", () => { + beforeEach(() => { + // Reset the global singleton between tests. + (globalThis as Record)["__healthlog_pgboss__"] = undefined; + }); + + it("is a no-op when no boss instance is attached", async () => { + await expect( + enqueuePrDetection("user-1", { silent: false }), + ).resolves.toBeUndefined(); + }); + + it("forwards the payload to boss.send when a boss instance is attached", async () => { + const send = vi.fn().mockResolvedValue(undefined); + setGlobalBoss({ send } as never); + + await enqueuePrDetection("user-1", { silent: true }); + + expect(send).toHaveBeenCalledTimes(1); + const [queue, payload] = send.mock.calls[0]; + expect(queue).toBe(PR_DETECTION_QUEUE); + expect(payload).toMatchObject({ userId: "user-1", silent: true }); + expect(typeof (payload as { triggeredAt: string }).triggeredAt).toBe( + "string", + ); + }); + + it("defaults silent to false when not provided", async () => { + const send = vi.fn().mockResolvedValue(undefined); + setGlobalBoss({ send } as never); + + await enqueuePrDetection("user-2"); + + const [, payload] = send.mock.calls[0]; + expect(payload).toMatchObject({ userId: "user-2", silent: false }); + }); +}); diff --git a/src/lib/jobs/pr-detection.ts b/src/lib/jobs/pr-detection.ts new file mode 100644 index 000000000..e373e29c4 --- /dev/null +++ b/src/lib/jobs/pr-detection.ts @@ -0,0 +1,72 @@ +/** + * v1.4.25 W16c — Enqueue helper + handler glue for the pg-boss + * `pr-detection` queue. Lives in `src/lib/jobs/` next to the other + * job-side helpers; the worker process binds the handler in + * `reminder-worker.ts` via `boss.work(PR_DETECTION_QUEUE, ...)`. + * + * The Next.js side (ingest routes) imports `enqueuePrDetection` and + * never touches pg-boss directly — `getGlobalBoss()` returns the + * worker-side singleton when one is attached, and a no-op when the + * route runs in a context without a worker (tests, scripts). The + * 30-minute fallback cron in `reminder-worker.ts` is the safety net + * for any ingest path that forgets to enqueue. + */ +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; + +export const PR_DETECTION_QUEUE = "pr-detection"; + +/** Concurrency budget for the worker process — five jobs in flight + * is enough to drain a multi-user backfill spike without crowding + * the reminder check or the daily insights workload that runs on the + * same node. */ +export const PR_DETECTION_CONCURRENCY = 5; + +/** + * Cron schedule (Europe/Berlin) for the fallback rescan. Every 30 + * minutes the worker re-runs detection for every user — protects + * against ingest paths that ship rows without enqueueing a job (a + * future Withings sync that lands without the hook wired, a manual + * row imported via SQL, a backfill from a future migration). + */ +export const PR_DETECTION_FALLBACK_CRON = "*/30 * * * *"; + +export interface PrDetectionPayload { + userId: string; + /** + * When true, the worker writes records but suppresses the push + * notification for any PR found in this run. Set by batch ingest + * paths once they cross the historical-backfill threshold + * (`silent = entries.length > 50`) so multi-year Apple Health + * imports don't fire hundreds of notifications. + */ + silent: boolean; + /** ISO timestamp — handy for debugging, never used for logic. */ + triggeredAt: string; +} + +/** + * Submit a PR detection job for one user. Best-effort: when no boss + * instance is attached (the route runs in a context without the + * worker process — typically tests), the call is a silent no-op and + * the fallback cron will pick the user up within 30 minutes. + */ +export async function enqueuePrDetection( + userId: string, + options: { silent?: boolean } = {}, +): Promise { + const boss = getGlobalBoss(); + if (!boss) return; + const payload: PrDetectionPayload = { + userId, + silent: options.silent ?? false, + triggeredAt: new Date().toISOString(), + }; + await boss.send(PR_DETECTION_QUEUE, payload, { + // Stagger retries gently — the detector is purely read-heavy on + // the user's own data, but a stampede of failed jobs against a + // briefly-flaky DB would compound the problem. + retryLimit: 3, + retryDelay: 30, + retryBackoff: true, + }); +} diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 0f111d277..3652930b3 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -33,6 +33,13 @@ import { cleanupExpiredIdempotencyKeys } from "@/lib/jobs/idempotency-cleanup"; import { cleanupOldAuditLogs } from "@/lib/jobs/audit-log-cleanup"; import { runHostMetricTick } from "@/lib/jobs/host-metric-sampler"; import { aggregateRecommendationFeedback } from "@/lib/jobs/feedback-aggregator"; +import { + PR_DETECTION_QUEUE, + PR_DETECTION_CONCURRENCY, + PR_DETECTION_FALLBACK_CRON, + type PrDetectionPayload, +} from "@/lib/jobs/pr-detection"; +import { detectPersonalRecordsForUser } from "@/lib/personal-records/pr-detection-worker"; import { rotateLegacyMoodLogSecrets } from "@/lib/moodlog-secret"; import { deleteMessage } from "@/lib/telegram"; import { decrypt, encrypt } from "@/lib/crypto"; @@ -1125,6 +1132,45 @@ async function handleFeedbackAggregator( }); } +async function handlePrDetection( + jobs: Job[], +) { + for (const job of jobs) { + await withBackgroundEvent("job.pr_detection", async (evt) => { + const p = getWorkerPrisma(); + // The cron-fired job carries an empty payload — iterate all + // users in that case. The push-suppression flag is irrelevant + // for the cron path (the dispatcher's per-user opt-in handles + // the loud/quiet decision once the row is written). + const payloadUserId = (job.data as PrDetectionPayload | undefined)?.userId; + const silent = (job.data as PrDetectionPayload | undefined)?.silent ?? false; + const userIds: string[] = payloadUserId + ? [payloadUserId] + : (await p.user.findMany({ select: { id: true } })).map((u) => u.id); + + let insertedTotal = 0; + let tiesTotal = 0; + for (const userId of userIds) { + try { + const result = await detectPersonalRecordsForUser(userId, { + silent, + prisma: p, + }); + insertedTotal += result.inserted; + tiesTotal += result.ties; + } catch (err) { + evt.addWarning(`pr-detection failed for user ${userId}: ${err}`); + } + } + evt.addMeta("pr_detection_users", userIds.length); + evt.addMeta("pr_detection_inserted", insertedTotal); + evt.addMeta("pr_detection_ties", tiesTotal); + evt.addMeta("pr_detection_silent", silent); + evt.addMeta("pr_detection_mode", payloadUserId ? "ingest" : "cron"); + }); + } +} + async function handleOffhostBackup(jobs: Job[]) { void jobs; await withBackgroundEvent("job.offhost_backup", async (evt) => { @@ -1374,6 +1420,7 @@ export async function startReminderWorker() { OFFHOST_BACKUP_QUEUE, HOST_METRIC_QUEUE, FEEDBACK_AGGREGATOR_QUEUE, + PR_DETECTION_QUEUE, ]; for (const q of allQueues) { @@ -1400,6 +1447,12 @@ export async function startReminderWorker() { [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON], [HOST_METRIC_QUEUE, HOST_METRIC_CRON], [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON], + // Fallback rescan every 30 minutes — protects against ingest paths + // that ship measurements without enqueueing a per-user job. The + // cron payload deliberately omits a `userId` so the handler iterates + // every user; per-user push-suppression cannot apply on the cron + // path (the silent flag is set by the ingest hooks). + [PR_DETECTION_QUEUE, PR_DETECTION_FALLBACK_CRON], ]; for (const [name, cron] of schedules) { @@ -1502,6 +1555,11 @@ export async function startReminderWorker() { { localConcurrency: 1 }, handleFeedbackAggregator, ); + await boss.work( + PR_DETECTION_QUEUE, + { localConcurrency: PR_DETECTION_CONCURRENCY }, + handlePrDetection, + ); return boss; }