feat(jobs): pr-detection pg-boss queue (concurrency=5, 30min fallback cron)

Adds the pr-detection queue and its handler to the reminder worker.
Ingest paths enqueue per-user jobs via enqueuePrDetection(); a
30-minute fallback cron iterates every user as a safety net for
ingest paths that skip the hook. Local concurrency is 5 so a
backfill spike drains without crowding the reminder check or the
daily insights workload running on the same node.

The payload carries a silent flag — ingest hooks set it on bulk
historical batches so a multi-year Apple Health backfill writes
the PR rows without firing hundreds of push notifications.

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<string, unknown>)["__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 });
+  });
+});

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<void> {
+  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,
+  });
+}

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<PrDetectionPayload | { userId?: undefined }>[],
+) {
+  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<OffhostBackupPayload>[]) {
   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<PrDetectionPayload>(
+    PR_DETECTION_QUEUE,
+    { localConcurrency: PR_DETECTION_CONCURRENCY },
+    handlePrDetection,
+  );
 
   return boss;
 }