refactor(jobs): decompose reminder-worker into domain queue registrars

Split the 2143-LOC startReminderWorker boot file into five domain
registrars under src/lib/jobs/reminder/ (integration-sync, status,
rollup, reminders, maintenance), each owning its queue-name constants,
its allQueues membership, its schedule tuples, and its boss.work
bindings. A shared createAndSchedule helper provisions every queue
before scheduling it. The boot file is now a 115-LOC registration table
that composes the registrars and runs the boot one-shots + discovery
enqueues.

Every queue declaration, schedule, boss.work binding, and boot-discovery
enqueue is preserved verbatim — verified by diffing the old monolith
against the union of the registrars (66/66 queues, 66/66 bindings,
identical schedule tuples and boot enqueues). The diff caught one
withings-oauth-state-cleanup binding that was dropped mid-refactor and
restored it.

Update the 18 queue-registration guard tests to follow the wiring into
its registrar module so the dead-queue protection moves with the code
rather than weakening: each grep guard now reads its queue's registrar,
the stress-strain-retention guard unions the allQueues blocks across the
status + rollup registrars, and the intake-slot-dedup schedules regex
tracks the new module-level indentation.

src/lib/jobs/__tests__/coach-memory-refresh-queue.test.ts

@@ -13,8 +13,11 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
-const source = readFileSync(REMINDER_WORKER_PATH, "utf8");
+// v1.18.1 — the coach-memory-refresh wiring moved out of the 2143-LOC
+// reminder-worker boot file into the status registrar. The dead-queue guard
+// follows the wiring there.
+const REGISTRAR_PATH = join(__dirname, "..", "reminder", "register-status.ts");
+const source = readFileSync(REGISTRAR_PATH, "utf8");
 
 function allQueuesBlock(): string {
   const m = source.match(/const allQueues\s*=\s*\[([\s\S]*?)\];/);

src/lib/jobs/__tests__/cycle-reminder-queue.test.ts

@@ -13,9 +13,18 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
+// v1.18.1 — the reminder-dispatch wiring moved out of the 2143-LOC
+// reminder-worker boot file into the reminders registrar. The dead-queue guard
+// follows the wiring there (the handler module is still concatenated for the
+// handler-symbol assertions).
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-reminders.ts",
+);
 const source =
-  readFileSync(REMINDER_WORKER_PATH, "utf8") +
+  readFileSync(REGISTRAR_PATH, "utf8") +
   readFileSync(
     join(__dirname, "..", "reminder", "mood-cycle-checks.ts"),
     "utf8",

src/lib/jobs/__tests__/drain-cumulative-queue.test.ts

@@ -14,8 +14,11 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
-const source = readFileSync(REMINDER_WORKER_PATH, "utf8");
+// v1.18.1 — the drain-cumulative wiring moved out of the 2143-LOC
+// reminder-worker boot file into the rollup registrar. The dead-queue guard
+// follows the wiring there.
+const REGISTRAR_PATH = join(__dirname, "..", "reminder", "register-rollup.ts");
+const source = readFileSync(REGISTRAR_PATH, "utf8");
 
 describe("reminder-worker — drainPerSampleCumulative nightly schedule", () => {
   it("declares the drain queue at the documented Berlin cadence", () => {

src/lib/jobs/__tests__/fitbit-queues.test.ts

@@ -17,9 +17,18 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
+// v1.18.1 — the queue wiring + boot discovery moved out of the 2143-LOC
+// reminder-worker boot file into domain registrars; the Fitbit queues live in
+// the integration-sync registrar. The dead-queue guard follows the wiring there
+// (the handler module is still concatenated for the import-routine assertions).
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-integration-sync.ts",
+);
 const source =
-  readFileSync(REMINDER_WORKER_PATH, "utf8") +
+  readFileSync(REGISTRAR_PATH, "utf8") +
   readFileSync(join(__dirname, "..", "reminder", "fitbit-sync.ts"), "utf8");
 
 const FITBIT_QUEUE_CONSTS = [

src/lib/jobs/__tests__/geo-backfill.test.ts

@@ -242,8 +242,10 @@ describe("geo-backfill scheduling contract (v1.4.37)", () => {
     // shape end-to-end.
     const fs = await import("node:fs/promises");
     const path = await import("node:path");
+    // v1.18.1 — the geo-backfill wiring moved out of the 2143-LOC
+    // reminder-worker boot file into the maintenance registrar.
     const workerSrc = await fs.readFile(
-      path.resolve(__dirname, "..", "reminder-worker.ts"),
+      path.resolve(__dirname, "..", "reminder", "register-maintenance.ts"),
       "utf-8",
     );
     const handlerSrc = await fs.readFile(

src/lib/jobs/__tests__/insight-pregenerate.test.ts

@@ -928,8 +928,11 @@ describe("forceWarmUser — on-demand single-user warm (v1.8.7.1)", () => {
 });
 
 describe("queue registration", () => {
+  // v1.18.1 — the insight-pregenerate wiring moved out of the 2143-LOC
+  // reminder-worker boot file into the status registrar. The dead-queue guard
+  // follows the wiring there.
   const workerSrc = fs.readFileSync(
-    path.resolve(__dirname, "../reminder-worker.ts"),
+    path.resolve(__dirname, "../reminder/register-status.ts"),
     "utf8",
   );
 

src/lib/jobs/__tests__/insight-status-generate.test.ts

@@ -133,8 +133,11 @@ describe("enqueueStatusGeneration", () => {
 });
 
 describe("queue registration", () => {
+  // v1.18.1 — the insight-status-generate wiring moved out of the 2143-LOC
+  // reminder-worker boot file into the status registrar. The dead-queue guard
+  // follows the wiring there.
   const workerSrc = fs.readFileSync(
-    path.resolve(__dirname, "../reminder-worker.ts"),
+    path.resolve(__dirname, "../reminder/register-status.ts"),
     "utf8",
   );
 

src/lib/jobs/__tests__/intake-slot-dedup-queue.test.ts

@@ -13,8 +13,16 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
-const source = readFileSync(REMINDER_WORKER_PATH, "utf8");
+// v1.18.1 — the intake-slot-dedup wiring + boot discovery moved out of the
+// 2143-LOC reminder-worker boot file into the maintenance registrar. The
+// dead-queue guard follows the wiring there.
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-maintenance.ts",
+);
+const source = readFileSync(REGISTRAR_PATH, "utf8");
 
 describe("reminder-worker — intake-slot-dedup wiring", () => {
   it("imports the queue symbols from the intake-slot-dedup module", () => {
@@ -65,8 +73,11 @@ describe("reminder-worker — intake-slot-dedup wiring", () => {
     // v1.15.20 — the schedules tuple grew an optional send-options slot
     // (retry policy for the insight queues); match on the assignment, not
     // the exact tuple type.
+    // v1.18.1 — `schedules` is now a module-level const in the registrar, so it
+    // closes at column 0 (`\n];`) rather than the monolith's function-body
+    // two-space indent.
     const schedulesMatch = source.match(
-      /const schedules:[^=]*=\s*\[([\s\S]*?)\n  \];/,
+      /const schedules:[^=]*=\s*\[([\s\S]*?)\n\];/,
     );
     expect(schedulesMatch).not.toBeNull();
     expect(schedulesMatch![1]).toMatch(

src/lib/jobs/__tests__/mean-consolidation-queue.test.ts

@@ -12,8 +12,11 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
-const source = readFileSync(REMINDER_WORKER_PATH, "utf8");
+// v1.18.1 — the mean-consolidation wiring + boot discovery moved out of the
+// 2143-LOC reminder-worker boot file into the rollup registrar. The dead-queue
+// guard follows the wiring there.
+const REGISTRAR_PATH = join(__dirname, "..", "reminder", "register-rollup.ts");
+const source = readFileSync(REGISTRAR_PATH, "utf8");
 
 describe("reminder-worker — mean-consolidation wiring", () => {
   it("imports the queue symbols from the mean-consolidation module", () => {

src/lib/jobs/__tests__/measurement-reminder-queue.test.ts

@@ -13,8 +13,15 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
+// v1.18.1 — the reminder-dispatch wiring (measurement reminder +
+// reminder-satisfy) moved out of the 2143-LOC reminder-worker boot file into
+// the reminders registrar. The dead-queue guard follows the wiring there (the
+// handler module is still concatenated for the handler-symbol assertions).
 const source =
-  readFileSync(join(__dirname, "..", "reminder-worker.ts"), "utf8") +
+  readFileSync(
+    join(__dirname, "..", "reminder", "register-reminders.ts"),
+    "utf8",
+  ) +
   readFileSync(
     join(__dirname, "..", "reminder", "mood-cycle-checks.ts"),
     "utf8",

src/lib/jobs/__tests__/measurement-tombstone-cleanup.test.ts

@@ -48,8 +48,10 @@ describe("cleanupExpiredMeasurementTombstones", () => {
 });
 
 describe("reminder-worker — measurement-tombstone-cleanup schedule", () => {
+  // v1.18.1 — the cleanup wiring moved out of the 2143-LOC reminder-worker
+  // boot file into the maintenance registrar. The dead-queue guard follows it.
   const source = readFileSync(
-    join(__dirname, "..", "reminder-worker.ts"),
+    join(__dirname, "..", "reminder", "register-maintenance.ts"),
     "utf8",
   );
 

src/lib/jobs/__tests__/nightscout-queues.test.ts

@@ -14,8 +14,15 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
+// v1.18.1 — the queue wiring moved out of the 2143-LOC reminder-worker boot
+// file into domain registrars; the Nightscout / Polar / Oura queues live in the
+// integration-sync registrar. The dead-queue guard follows the wiring there
+// (the handler module is still concatenated for the handler-symbol assertions).
 const source =
-  readFileSync(join(__dirname, "..", "reminder-worker.ts"), "utf8") +
+  readFileSync(
+    join(__dirname, "..", "reminder", "register-integration-sync.ts"),
+    "utf8",
+  ) +
   readFileSync(join(__dirname, "..", "reminder", "nightscout-sync.ts"), "utf8");
 
 describe("reminder-worker — Nightscout sync queue", () => {

src/lib/jobs/__tests__/period-narrative-warm.test.ts

@@ -176,8 +176,11 @@ describe("runPeriodNarrativeWarm", () => {
 });
 
 describe("queue registration", () => {
+  // v1.18.1 — the period-narrative wiring moved out of the 2143-LOC
+  // reminder-worker boot file into the status registrar. The dead-queue guard
+  // follows the wiring there.
   const workerSrc = fs.readFileSync(
-    path.resolve(__dirname, "../reminder-worker.ts"),
+    path.resolve(__dirname, "../reminder/register-status.ts"),
     "utf8",
   );
 

src/lib/jobs/__tests__/recovery-score-queue.test.ts

@@ -16,8 +16,16 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
-const workerSource = readFileSync(REMINDER_WORKER_PATH, "utf8");
+// v1.18.1 — the queue wiring moved out of the 2143-LOC reminder-worker boot
+// file into domain registrars; the recovery-score queue lives in the status
+// registrar. The dead-queue guard follows the wiring there.
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-status.ts",
+);
+const workerSource = readFileSync(REGISTRAR_PATH, "utf8");
 
 const BATCH_ROUTE_PATH = join(
   __dirname,

src/lib/jobs/__tests__/sleep-timeline-queues.test.ts

@@ -12,8 +12,16 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
-const source = readFileSync(REMINDER_WORKER_PATH, "utf8");
+// v1.18.1 — the queue wiring + boot discovery moved out of the 2143-LOC
+// reminder-worker boot file into domain registrars; the sleep-timeline backfill
+// lives in the integration-sync registrar. The dead-queue guard follows it.
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-integration-sync.ts",
+);
+const source = readFileSync(REGISTRAR_PATH, "utf8");
 
 describe("reminder-worker — sleep-timeline backfill queue", () => {
   it("registers the queue in allQueues (v1.4.37 dead-queue guard)", () => {

src/lib/jobs/__tests__/status-cron-candidates.test.ts

@@ -175,8 +175,11 @@ describe("findStatusCronCandidates — gates", () => {
 });
 
 describe("mood-status queue registration (dead-queue guard)", () => {
+  // v1.18.1 — the nightly status-ladder wiring moved out of the 2143-LOC
+  // reminder-worker boot file into the status registrar. The dead-queue guard
+  // follows the wiring there.
   const workerSrc = fs.readFileSync(
-    path.resolve(__dirname, "../reminder-worker.ts"),
+    path.resolve(__dirname, "../reminder/register-status.ts"),
     "utf8",
   );
 

src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts

@@ -14,16 +14,37 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
-const source = readFileSync(REMINDER_WORKER_PATH, "utf8");
+// v1.18.1 — the queue wiring moved out of the 2143-LOC reminder-worker boot
+// file into domain registrars. Stress + Strain live in the status registrar;
+// the dense intra-day retention queue (and the nightly fold onto the
+// drain-cumulative tick) lives in the rollup registrar. Concatenate both so the
+// dead-queue guard follows the wiring into each.
+const STATUS_REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-status.ts",
+);
+const ROLLUP_REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-rollup.ts",
+);
+const source =
+  readFileSync(STATUS_REGISTRAR_PATH, "utf8") +
+  readFileSync(ROLLUP_REGISTRAR_PATH, "utf8");
 
 const DENSE_RETENTION_PATH = join(__dirname, "..", "dense-intraday-retention.ts");
 const denseRetentionSource = readFileSync(DENSE_RETENTION_PATH, "utf8");
 
 function allQueuesBlock(): string {
-  const m = source.match(/const allQueues\s*=\s*\[([\s\S]*?)\];/);
-  expect(m).not.toBeNull();
-  return m![1];
+  // Each registrar owns its own `const allQueues = [...]`; union every block so
+  // a queue registered in either the status or rollup registrar satisfies the
+  // membership assertion.
+  const blocks = [...source.matchAll(/const allQueues\s*=\s*\[([\s\S]*?)\];/g)];
+  expect(blocks.length).toBeGreaterThan(0);
+  return blocks.map((m) => m[1]).join("\n");
 }
 
 describe("reminder-worker — stress-score wiring", () => {

src/lib/jobs/__tests__/tls-pin-monitor-queue.test.ts

@@ -12,9 +12,18 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
+// v1.18.1 — the TLS-pin-monitor wiring moved out of the 2143-LOC
+// reminder-worker boot file into the maintenance registrar. The dead-queue
+// guard follows the wiring there (the ops-handlers module is still concatenated
+// for the handler-body `runTlsPinMonitor` assertion).
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-maintenance.ts",
+);
 const workerSource =
-  readFileSync(REMINDER_WORKER_PATH, "utf8") +
+  readFileSync(REGISTRAR_PATH, "utf8") +
   readFileSync(join(__dirname, "..", "reminder", "ops-handlers.ts"), "utf8");
 
 describe("reminder-worker — tls-pin-monitor wiring", () => {

src/lib/jobs/__tests__/whoop-queues.test.ts

@@ -13,9 +13,18 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
+// v1.18.1 — the queue wiring + boot discovery moved out of the 2143-LOC
+// reminder-worker boot file into domain registrars; the WHOOP queues live in
+// the integration-sync registrar. The dead-queue guard follows the wiring there
+// (the handler module is still concatenated for the import-routine assertions).
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-integration-sync.ts",
+);
 const source =
-  readFileSync(REMINDER_WORKER_PATH, "utf8") +
+  readFileSync(REGISTRAR_PATH, "utf8") +
   readFileSync(join(__dirname, "..", "reminder", "whoop-sync.ts"), "utf8");
 
 const WHOOP_QUEUE_CONSTS = [

src/lib/jobs/__tests__/withings-queues.test.ts

@@ -14,10 +14,19 @@ import { join } from "node:path";
 
 import { describe, expect, it } from "vitest";
 
-const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts");
+// v1.18.1 — the queue wiring moved out of the 2143-LOC reminder-worker boot
+// file into domain registrars; the Withings queues live in the integration-sync
+// registrar. The dead-queue guard follows the wiring there (the handler module
+// is still concatenated for the import-routine assertions below).
+const REGISTRAR_PATH = join(
+  __dirname,
+  "..",
+  "reminder",
+  "register-integration-sync.ts",
+);
 
 const source =
-  readFileSync(REMINDER_WORKER_PATH, "utf8") +
+  readFileSync(REGISTRAR_PATH, "utf8") +
   readFileSync(join(__dirname, "..", "reminder", "withings-sync.ts"), "utf8");
 
 describe("reminder-worker — Withings activity + sleep v2 queues", () => {

src/lib/jobs/reminder/register-integration-sync.ts

@@ -0,0 +1,509 @@
+/**
+ * Integration-sync queue registrar.
+ *
+ * Owns the third-party-provider sync queues — Withings (fallback / activity /
+ * sleep), WHOOP (recovery / sleep / workout / cycle + backfill + OAuth-state
+ * cleanup), Fitbit (sync + backfill + OAuth-state cleanup), Nightscout, Polar,
+ * Oura, MoodLog, and the two one-shot backfills (sleep-timeline, lab-biomarker)
+ * — plus their boot-time discovery enqueues.
+ *
+ * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
+ * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
+ * and a `boss.work(QUEUE, …, handler)` binding drains it. The queue-wiring
+ * guards (`withings-queues`, `whoop-queues`, `fitbit-queues`,
+ * `nightscout-queues`, `sleep-timeline-queues`) read THIS module.
+ */
+import { PgBoss } from "pg-boss";
+import {
+  WHOOP_BACKFILL_QUEUE,
+  WHOOP_BACKFILL_CONCURRENCY,
+  runWhoopBackfillForUser,
+  enqueueBootTimeWhoopBackfill,
+  type WhoopBackfillPayload,
+} from "@/lib/jobs/whoop-backfill";
+import {
+  FITBIT_BACKFILL_QUEUE,
+  FITBIT_BACKFILL_CONCURRENCY,
+  runFitbitBackfillForUser,
+  enqueueBootTimeFitbitBackfill,
+  type FitbitBackfillPayload,
+} from "@/lib/jobs/fitbit-backfill";
+import {
+  SLEEP_TIMELINE_BACKFILL_QUEUE,
+  SLEEP_TIMELINE_BACKFILL_CONCURRENCY,
+  runSleepTimelineBackfillForUser,
+  enqueueBootTimeSleepTimelineBackfill,
+  type SleepTimelineBackfillPayload,
+} from "@/lib/jobs/sleep-timeline-backfill";
+import {
+  LAB_BIOMARKER_BACKFILL_QUEUE,
+  LAB_BIOMARKER_BACKFILL_CONCURRENCY,
+  runLabBiomarkerBackfillForUser,
+  enqueueBootTimeLabBiomarkerBackfill,
+  type LabBiomarkerBackfillPayload,
+} from "@/lib/jobs/lab-biomarker-backfill";
+import { workerLog } from "./shared";
+import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+import {
+  WithingsSyncPayload,
+  WithingsActivitySyncPayload,
+  WithingsSleepSyncPayload,
+  handleWithingsFallbackSync,
+  handleWithingsActivitySync,
+  handleWithingsSleepSync,
+} from "./withings-sync";
+import {
+  WhoopSyncPayload,
+  handleWhoopRecoverySync,
+  handleWhoopSleepSync,
+  handleWhoopWorkoutSync,
+  handleWhoopCycleSync,
+} from "./whoop-sync";
+import { MoodLogSyncPayload, handleMoodLogSync } from "./moodlog-sync";
+import {
+  WithingsOAuthStateCleanupPayload,
+  handleWithingsOAuthStateCleanup,
+  WhoopOAuthStateCleanupPayload,
+  handleWhoopOAuthStateCleanup,
+} from "./cleanup-handlers";
+import { NightscoutSyncPayload, handleNightscoutSync } from "./nightscout-sync";
+import { PolarSyncPayload, handlePolarSync } from "./polar-sync";
+import { OuraSyncPayload, handleOuraSync } from "./oura-sync";
+import {
+  FitbitSyncPayload,
+  handleFitbitSync,
+  FitbitOAuthStateCleanupPayload,
+  handleFitbitOAuthStateCleanup,
+} from "./fitbit-sync";
+
+const WITHINGS_SYNC_QUEUE = "withings-fallback-sync";
+
+const WITHINGS_SYNC_CRON = "0 * * * *"; // every 60 minutes
+// v1.4.25 W17b/c — webhook-primary + cron-safety-net for activity and
+// sleep v2. The webhook handler enqueues per-user jobs on appli=16 / 44
+// notifications; the crons below are the catch-net for the 1 % of
+// notifications Withings drops. Offset 15-minute cadence per the
+// research recommendation so the two queues don't lockstep against the
+// existing measure cron at :00.
+
+const WITHINGS_ACTIVITY_QUEUE = "withings-activity-sync";
+
+const WITHINGS_ACTIVITY_CRON = "0 * * * *"; // every hour at :00
+
+const WITHINGS_SLEEP_QUEUE = "withings-sleep-sync";
+
+const WITHINGS_SLEEP_CRON = "15 * * * *"; // every hour at :15
+
+const MOODLOG_SYNC_QUEUE = "moodlog-sync";
+
+const MOODLOG_SYNC_CRON = "30 * * * *"; // every hour at :30
+// v1.4.47 W6 — daily sweep for the Withings OAuth state ledger. Slots
+// at :20 between the audit-log cleanup (:15) and the mood-reminder
+// cleanup (:25), inside the existing 02:xx-03:xx maintenance window.
+// Rows are normally consumed by the callback handler in single-use
+// fashion; this sweep picks up the long tail where a user closed the
+// Withings approval tab without bouncing back to the callback URL.
+
+const WITHINGS_OAUTH_STATE_CLEANUP_QUEUE = "withings-oauth-state-cleanup";
+
+const WITHINGS_OAUTH_STATE_CLEANUP_CRON = "20 3 * * *";
+// v1.11.0 — WHOOP sync queues. Webhook-primary + cron-safety-net, mirroring
+// the Withings activity/sleep crons. Recovery / sleep / workout each have a
+// WHOOP webhook (`*.updated`) that enqueues the matching per-resource job; the
+// crons below are the catch-net for dropped deliveries. Cycle has NO webhook,
+// so its cron is the only driver. Minutes are staggered off the Withings crons
+// (:00/:15) to spread DB load.
+
+const WHOOP_RECOVERY_SYNC_QUEUE = "whoop-recovery-sync";
+
+const WHOOP_RECOVERY_SYNC_CRON = "5 * * * *"; // every hour at :05
+
+const WHOOP_SLEEP_SYNC_QUEUE = "whoop-sleep-sync";
+
+const WHOOP_SLEEP_SYNC_CRON = "20 * * * *"; // every hour at :20
+
+const WHOOP_WORKOUT_SYNC_QUEUE = "whoop-workout-sync";
+
+const WHOOP_WORKOUT_SYNC_CRON = "35 * * * *"; // every hour at :35
+
+const WHOOP_CYCLE_SYNC_QUEUE = "whoop-cycle-sync";
+
+const WHOOP_CYCLE_SYNC_CRON = "50 * * * *"; // every hour at :50 (poll-only)
+// v1.11.0 — daily sweep for the WHOOP OAuth state ledger. Slots at 03:22,
+// next to the Withings sweep (03:20), inside the maintenance window.
+
+const WHOOP_OAUTH_STATE_CLEANUP_QUEUE = "whoop-oauth-state-cleanup";
+
+const WHOOP_OAUTH_STATE_CLEANUP_CRON = "22 3 * * *";
+// v1.12.0 — Fitbit / Google Health poll-only sync. There is no Fitbit webhook
+// at launch (Pub/Sub deferred), so a single hourly cron drives the per-user
+// `syncUserFitbit` driver across every connection. Minute staggered off the
+// WHOOP slots (:05/:20/:35/:50) and the Withings slots (:00/:15) so the hourly
+// ticks don't pile up on one boss poll.
+
+const FITBIT_SYNC_QUEUE = "fitbit-sync";
+
+const FITBIT_SYNC_CRON = "8 * * * *"; // every hour at :08
+// v1.12.0 — daily sweep for the Fitbit OAuth state ledger. Slots at 03:24, next
+// to the WHOOP sweep (03:22), inside the maintenance window.
+
+const FITBIT_OAUTH_STATE_CLEANUP_QUEUE = "fitbit-oauth-state-cleanup";
+
+const FITBIT_OAUTH_STATE_CLEANUP_CRON = "24 3 * * *";
+// v1.17.0 — Nightscout CGM poll sync. Poll-only (no webhook): one hourly tick
+// pulls the recent SGV window per configured instance. :11 staggers off the
+// WHOOP (:05), Fitbit (:08), and Withings sync ticks so the hourly polls don't
+// pile up on one boss poll.
+const NIGHTSCOUT_SYNC_QUEUE = "nightscout-sync";
+
+const NIGHTSCOUT_SYNC_CRON = "11 * * * *"; // every hour at :11
+// v1.17.0 (F4) — Polar + Oura OAuth poll sync. Poll-only (one hourly tick per
+// provider re-walks every connected user). :13 / :15 stagger off the other
+// hourly sync ticks (WHOOP :05, Fitbit :08, Nightscout :11) so the polls don't
+// pile up on one boss poll. The queues MUST be registered in `allQueues` below
+// or pg-boss never provisions them and the schedule silently no-ops (the
+// v1.4.37 dead-queue class).
+const POLAR_SYNC_QUEUE = "polar-sync";
+const POLAR_SYNC_CRON = "13 * * * *"; // every hour at :13
+const OURA_SYNC_QUEUE = "oura-sync";
+const OURA_SYNC_CRON = "15 * * * *"; // every hour at :15
+
+// pg-boss v12 requires explicit queue creation before scheduling. Every queue
+// MUST be registered here or pg-boss never provisions it and both the webhook
+// enqueue AND the cron schedule below silently no-op (the v1.4.37 dead-queue
+// class).
+const allQueues = [
+  WITHINGS_SYNC_QUEUE,
+  WITHINGS_ACTIVITY_QUEUE,
+  WITHINGS_SLEEP_QUEUE,
+  MOODLOG_SYNC_QUEUE,
+  WITHINGS_OAUTH_STATE_CLEANUP_QUEUE,
+  // v1.11.0 — WHOOP sync queues. Webhook-primary + cron-safety-net for
+  // recovery / sleep / workout; cycle is poll-only (no WHOOP webhook).
+  WHOOP_RECOVERY_SYNC_QUEUE,
+  WHOOP_SLEEP_SYNC_QUEUE,
+  WHOOP_WORKOUT_SYNC_QUEUE,
+  WHOOP_CYCLE_SYNC_QUEUE,
+  // v1.11.0 — self-converging boot backfill for newly connected WHOOP
+  // accounts. Discovery enqueues one full-history sync per un-backfilled
+  // connection; idempotent across reboots.
+  WHOOP_BACKFILL_QUEUE,
+  // v1.11.0 — daily sweep for the WHOOP OAuth state ledger.
+  WHOOP_OAUTH_STATE_CLEANUP_QUEUE,
+  // v1.12.0 — Fitbit / Google Health poll-only sync (no webhook at launch),
+  // self-converging boot backfill, and the daily OAuth-state ledger sweep.
+  FITBIT_SYNC_QUEUE,
+  FITBIT_BACKFILL_QUEUE,
+  FITBIT_OAUTH_STATE_CLEANUP_QUEUE,
+  // v1.17.1 — one-shot sleep-timeline backfill for WHOOP + Withings.
+  // Discovery enqueues one job per connection whose sleep rows predate the
+  // stamp/shape fix; the pass deletes the affected SLEEP_DURATION rows and
+  // re-syncs. Idempotent across reboots.
+  SLEEP_TIMELINE_BACKFILL_QUEUE,
+  // v1.18.1 — one-shot backfill that links legacy free-text lab readings to
+  // a user-scoped Biomarker catalog entry (group by `lower(analyte)`).
+  LAB_BIOMARKER_BACKFILL_QUEUE,
+  // v1.17.0 — Nightscout CGM poll sync. Poll-only (no webhook, no OAuth, no
+  // backfill queue — the hourly window walks the recent SGV set).
+  NIGHTSCOUT_SYNC_QUEUE,
+  // v1.17.0 (F4) — Polar + Oura OAuth poll sync.
+  POLAR_SYNC_QUEUE,
+  OURA_SYNC_QUEUE,
+];
+
+const schedules: ScheduleEntry[] = [
+  [WITHINGS_SYNC_QUEUE, WITHINGS_SYNC_CRON],
+  [WITHINGS_ACTIVITY_QUEUE, WITHINGS_ACTIVITY_CRON],
+  [WITHINGS_SLEEP_QUEUE, WITHINGS_SLEEP_CRON],
+  [MOODLOG_SYNC_QUEUE, MOODLOG_SYNC_CRON],
+  [WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, WITHINGS_OAUTH_STATE_CLEANUP_CRON],
+  // v1.11.0 — WHOOP poll-fallback crons. Recovery/sleep/workout catch
+  // dropped webhooks; cycle is the sole driver (no webhook). Staggered off
+  // the Withings crons so the hourly ticks don't pile up on one boss poll.
+  [WHOOP_RECOVERY_SYNC_QUEUE, WHOOP_RECOVERY_SYNC_CRON],
+  [WHOOP_SLEEP_SYNC_QUEUE, WHOOP_SLEEP_SYNC_CRON],
+  [WHOOP_WORKOUT_SYNC_QUEUE, WHOOP_WORKOUT_SYNC_CRON],
+  [WHOOP_CYCLE_SYNC_QUEUE, WHOOP_CYCLE_SYNC_CRON],
+  // v1.11.0 — daily 03:22 Europe/Berlin prune for expired WHOOP OAuth states.
+  [WHOOP_OAUTH_STATE_CLEANUP_QUEUE, WHOOP_OAUTH_STATE_CLEANUP_CRON],
+  // v1.12.0 — hourly Fitbit poll (:08, staggered off WHOOP/Withings) + the
+  // daily 03:24 Europe/Berlin prune for expired Fitbit OAuth states.
+  [FITBIT_SYNC_QUEUE, FITBIT_SYNC_CRON],
+  [FITBIT_OAUTH_STATE_CLEANUP_QUEUE, FITBIT_OAUTH_STATE_CLEANUP_CRON],
+  // v1.17.0 — hourly Nightscout CGM poll (:11, staggered off the other sync
+  // ticks).
+  [NIGHTSCOUT_SYNC_QUEUE, NIGHTSCOUT_SYNC_CRON],
+  // v1.17.0 (F4) — hourly Polar (:13) + Oura (:15) OAuth polls.
+  [POLAR_SYNC_QUEUE, POLAR_SYNC_CRON],
+  [OURA_SYNC_QUEUE, OURA_SYNC_CRON],
+];
+
+/**
+ * Register every integration-sync queue: create, schedule, and bind handlers.
+ * Returns the queue names created (for the boot-level aggregate assertion).
+ */
+export async function registerIntegrationSyncQueues(
+  boss: PgBoss,
+): Promise<readonly string[]> {
+  await createAndSchedule(boss, allQueues, schedules);
+
+  await boss.work<WithingsSyncPayload>(
+    WITHINGS_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleWithingsFallbackSync,
+  );
+  await boss.work<WithingsActivitySyncPayload>(
+    WITHINGS_ACTIVITY_QUEUE,
+    { localConcurrency: 1 },
+    handleWithingsActivitySync,
+  );
+  await boss.work<WithingsSleepSyncPayload>(
+    WITHINGS_SLEEP_QUEUE,
+    { localConcurrency: 1 },
+    handleWithingsSleepSync,
+  );
+  await boss.work<WithingsOAuthStateCleanupPayload>(
+    WITHINGS_OAUTH_STATE_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleWithingsOAuthStateCleanup,
+  );
+  // v1.11.0 — WHOOP per-resource sync handlers. Webhook-driven per-user +
+  // cron full-iteration. Serial concurrency so a backfill-heavy tick never
+  // crowds the request pool and stays inside WHOOP's 100 req/min app cap.
+  await boss.work<WhoopSyncPayload>(
+    WHOOP_RECOVERY_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleWhoopRecoverySync,
+  );
+  await boss.work<WhoopSyncPayload>(
+    WHOOP_SLEEP_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleWhoopSleepSync,
+  );
+  await boss.work<WhoopSyncPayload>(
+    WHOOP_WORKOUT_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleWhoopWorkoutSync,
+  );
+  await boss.work<WhoopSyncPayload>(
+    WHOOP_CYCLE_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleWhoopCycleSync,
+  );
+  // v1.11.0 — self-converging WHOOP backfill. The boot enqueue below sends one
+  // full-history sync per un-backfilled connection; this handler runs it and
+  // stamps `backfillCompletedAt` so the discovery query drops the account.
+  await boss.work<WhoopBackfillPayload>(
+    WHOOP_BACKFILL_QUEUE,
+    { localConcurrency: WHOOP_BACKFILL_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        const { imported } = await runWhoopBackfillForUser(userId);
+        workerLog(
+          "info",
+          `[whoop-backfill] user=${userId} imported=${imported}`,
+        );
+      }
+    },
+  );
+  await boss.work<WhoopOAuthStateCleanupPayload>(
+    WHOOP_OAUTH_STATE_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleWhoopOAuthStateCleanup,
+  );
+  // v1.12.0 — Fitbit poll-sync (cron full-iteration; no webhook). Serial
+  // concurrency so a backfill-heavy tick never crowds the request pool.
+  await boss.work<FitbitSyncPayload>(
+    FITBIT_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleFitbitSync,
+  );
+  // v1.12.0 — self-converging Fitbit backfill. The boot enqueue below sends one
+  // full-history sync per un-backfilled connection; this handler runs it and
+  // stamps `backfillCompletedAt` so the discovery query drops the account.
+  await boss.work<FitbitBackfillPayload>(
+    FITBIT_BACKFILL_QUEUE,
+    { localConcurrency: FITBIT_BACKFILL_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        const { imported } = await runFitbitBackfillForUser(userId);
+        workerLog(
+          "info",
+          `[fitbit-backfill] user=${userId} imported=${imported}`,
+        );
+      }
+    },
+  );
+  await boss.work<FitbitOAuthStateCleanupPayload>(
+    FITBIT_OAUTH_STATE_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleFitbitOAuthStateCleanup,
+  );
+  // v1.17.1 — one-shot sleep-timeline backfill. The boot enqueue below sends
+  // one job per (user, provider) whose sleep rows predate the stamp/shape fix;
+  // this handler deletes the affected rows, re-syncs, and stamps the marker so
+  // the discovery query drops the connection.
+  await boss.work<SleepTimelineBackfillPayload>(
+    SLEEP_TIMELINE_BACKFILL_QUEUE,
+    { localConcurrency: SLEEP_TIMELINE_BACKFILL_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId, provider } = job.data;
+        const { deleted, imported } = await runSleepTimelineBackfillForUser(
+          userId,
+          provider,
+        );
+        workerLog(
+          "info",
+          `[sleep-timeline-backfill] user=${userId} provider=${provider} deleted=${deleted} imported=${imported}`,
+        );
+      }
+    },
+  );
+  // v1.18.1 — one-shot lab-biomarker backfill. The boot enqueue below sends
+  // one job per user holding un-linked free-text lab readings; this handler
+  // groups them by `lower(analyte)`, mints/reuses a Biomarker per group, and
+  // links the rows. Idempotent — a re-run links only what is still un-linked.
+  await boss.work<LabBiomarkerBackfillPayload>(
+    LAB_BIOMARKER_BACKFILL_QUEUE,
+    { localConcurrency: LAB_BIOMARKER_BACKFILL_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        const { markers, linked } = await runLabBiomarkerBackfillForUser(userId);
+        workerLog(
+          "info",
+          `[lab-biomarker-backfill] user=${userId} markers=${markers} linked=${linked}`,
+        );
+      }
+    },
+  );
+  // v1.17.0 — Nightscout CGM poll-cohort sync. The hourly cron tick (no
+  // `userId`) walks every configured instance; one user's unreachable host is
+  // warned, not fatal.
+  await boss.work<NightscoutSyncPayload>(
+    NIGHTSCOUT_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleNightscoutSync,
+  );
+  // v1.17.0 (F4) — Polar + Oura OAuth poll-cohort sync. The hourly cron tick
+  // (no `userId`) re-walks every connected user; one user's revoked grant is
+  // warned, not fatal.
+  await boss.work<PolarSyncPayload>(
+    POLAR_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handlePolarSync,
+  );
+  await boss.work<OuraSyncPayload>(
+    OURA_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleOuraSync,
+  );
+  await boss.work<MoodLogSyncPayload>(
+    MOODLOG_SYNC_QUEUE,
+    { localConcurrency: 1 },
+    handleMoodLogSync,
+  );
+
+  return allQueues;
+}
+
+/**
+ * Fire-and-forget boot discovery for the self-converging integration backfills.
+ * Each pass is idempotent across reboots and never fails worker boot on a miss
+ * (errors come back through the helper's result value).
+ */
+export async function enqueueIntegrationSyncBootDiscovery(): Promise<void> {
+  // v1.11.0 — WHOOP backfill. Finds every WHOOP connection not yet backfilled
+  // and enqueues one full-history sync per account. A completed backfill stamps
+  // `backfillCompletedAt`, dropping the connection from the discovery set.
+  try {
+    const { enqueued, skipped, error } = await enqueueBootTimeWhoopBackfill();
+    if (error) {
+      workerLog("error", `[whoop-backfill] boot discovery failed: ${error}`);
+    } else {
+      workerLog(
+        "info",
+        `[whoop-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[whoop-backfill] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.12.0 — Fitbit backfill. Finds every Fitbit connection not yet
+  // backfilled and enqueues one full-history sync per account.
+  try {
+    const { enqueued, skipped, error } = await enqueueBootTimeFitbitBackfill();
+    if (error) {
+      workerLog("error", `[fitbit-backfill] boot discovery failed: ${error}`);
+    } else {
+      workerLog(
+        "info",
+        `[fitbit-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[fitbit-backfill] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.17.1 — one-shot sleep-timeline backfill. Finds every WHOOP + Withings
+  // connection whose sleep rows predate the stamp/shape fix and enqueues one
+  // job per (user, provider). A completed pass stamps `sleepTimelineBackfillAt`.
+  try {
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeSleepTimelineBackfill();
+    if (error) {
+      workerLog(
+        "error",
+        `[sleep-timeline-backfill] boot discovery failed: ${error}`,
+      );
+    } else {
+      workerLog(
+        "info",
+        `[sleep-timeline-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[sleep-timeline-backfill] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.18.1 — one-shot lab-biomarker backfill. Finds every user holding an
+  // un-linked live lab reading and enqueues one job each. A completed pass
+  // links every reading, dropping the user from the discovery set.
+  try {
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeLabBiomarkerBackfill();
+    if (error) {
+      workerLog(
+        "error",
+        `[lab-biomarker-backfill] boot discovery failed: ${error}`,
+      );
+    } else {
+      workerLog(
+        "info",
+        `[lab-biomarker-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[lab-biomarker-backfill] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+}

src/lib/jobs/reminder/register-maintenance.ts

@@ -0,0 +1,433 @@
+/**
+ * Maintenance / ops / cleanup queue registrar.
+ *
+ * Owns the housekeeping tier: data backup (in-DB weekly + off-host nightly),
+ * the restore drill, every ledger cleanup (rate-limit, idempotency, audit-log,
+ * mood-reminder, push-attempt, measurement-tombstone), medication inventory
+ * expiry, intake auto-skip, the duplicate dose-slot dedup (boot + daily-cron),
+ * host-metric sampling, the rec-feedback aggregator, the geo backfill, the TLS
+ * pin monitor, PR detection, and the Apple Health export ingest.
+ *
+ * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
+ * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
+ * and a `boss.work(QUEUE, …, handler)` binding drains it. The
+ * `measurement-tombstone-cleanup`, `withings-oauth-state-cleanup`,
+ * `intake-slot-dedup-queue`, `tls-pin-monitor-queue`, and `geo-backfill`
+ * guards read THIS module.
+ */
+import { PgBoss } from "pg-boss";
+import { GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON } from "@/lib/jobs/geo-backfill";
+import {
+  TLS_PIN_MONITOR_QUEUE,
+  TLS_PIN_MONITOR_CRON,
+} from "@/lib/jobs/tls-pin-monitor";
+import {
+  PR_DETECTION_QUEUE,
+  PR_DETECTION_CONCURRENCY,
+  PR_DETECTION_FALLBACK_CRON,
+  type PrDetectionPayload,
+} from "@/lib/jobs/pr-detection";
+import {
+  MEDICATION_INVENTORY_EXPIRE_QUEUE,
+  MEDICATION_INVENTORY_EXPIRE_CRON,
+  type MedicationInventoryExpirePayload,
+} from "@/lib/jobs/medication-inventory-expire";
+import {
+  INTAKE_AUTO_SKIP_QUEUE,
+  INTAKE_AUTO_SKIP_CRON,
+  type IntakeAutoSkipPayload,
+} from "@/lib/jobs/intake-auto-skip";
+import {
+  APPLE_HEALTH_IMPORT_QUEUE,
+  APPLE_HEALTH_IMPORT_CONCURRENCY,
+  handleAppleHealthImport,
+  type AppleHealthImportPayload,
+} from "@/lib/jobs/apple-health-import-worker";
+import {
+  INTAKE_SLOT_DEDUP_QUEUE,
+  INTAKE_SLOT_DEDUP_CONCURRENCY,
+  dedupeUserIntakeSlots,
+  enqueueBootTimeIntakeSlotDedup,
+  type IntakeSlotDedupPayload,
+} from "@/lib/medications/intake-slot-dedup";
+import {
+  handleRestoreDrill,
+  RESTORE_DRILL_CRON,
+  RESTORE_DRILL_QUEUE,
+} from "@/lib/jobs/restore-drill";
+import { recordError } from "@/lib/jobs/worker-status";
+import { workerLog } from "./shared";
+import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+import {
+  DataBackupPayload,
+  OffhostBackupPayload,
+  handleOffhostBackup,
+  handleDataBackup,
+} from "./backup-handlers";
+import {
+  RateLimitCleanupPayload,
+  IdempotencyCleanupPayload,
+  AuditLogCleanupPayload,
+  MoodReminderCleanupPayload,
+  handleMoodReminderCleanup,
+  PushAttemptCleanupPayload,
+  handlePushAttemptCleanup,
+  MeasurementTombstoneCleanupPayload,
+  handleMeasurementTombstoneCleanup,
+  handleRateLimitCleanup,
+  handleIdempotencyCleanup,
+  handleAuditLogCleanup,
+} from "./cleanup-handlers";
+import {
+  HostMetricSamplePayload,
+  FeedbackAggregatorPayload,
+  GeoBackfillPayload,
+  TlsPinMonitorPayload,
+  handleHostMetricSample,
+  handleFeedbackAggregator,
+  handleGeoBackfill,
+  handleTlsPinMonitor,
+  handlePrDetection,
+} from "./ops-handlers";
+import {
+  handleMedicationInventoryExpire,
+  handleIntakeAutoSkip,
+} from "./medication-maintenance";
+
+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)
+
+const AUDIT_LOG_CLEANUP_QUEUE = "audit-log-cleanup";
+
+const AUDIT_LOG_CLEANUP_CRON = "15 3 * * *"; // daily at 03:15 (Europe/Berlin)
+// v1.15.19 — daily duplicate dose-slot dedup discovery tick. The boot-time
+// pass only ran on worker restart, so a cross-source duplicate slot created
+// between deploys (a pending REMINDER row plus a standalone API/WEB row on
+// the same instant) survived until the next reboot. The cron payload omits
+// `userId`; the handler treats that as the discovery tick and fans out one
+// per-user job (singletonKey-coalesced) exactly like the boot pass. Slots at
+// 03:28 between the mood-reminder cleanup (03:25) and the inventory expire
+// (03:30), inside the maintenance window.
+
+const INTAKE_SLOT_DEDUP_CRON = "28 3 * * *";
+
+const OFFHOST_BACKUP_QUEUE = "data-backup-offhost";
+// 02:30 Europe/Berlin — runs after audit-log/idempotency cleanups so old
+// rows are gone before they're snapshotted, but before the existing
+// in-DB DATA_BACKUP at 03:00 (Sundays only) so the off-host copy is
+// always at-or-ahead of the local one.
+
+const OFFHOST_BACKUP_CRON = "30 2 * * *";
+
+const HOST_METRIC_QUEUE = "host-metric-sample";
+// Per-minute cadence — matches the chart's 60s polling refetchInterval.
+
+const HOST_METRIC_CRON = "* * * * *";
+// v1.4.16 phase B5e — daily rec-feedback aggregator. 04:00 Europe/Berlin
+// runs the slot AFTER all the cleanup jobs (rate-limit, idempotency,
+// audit-log) so the previous-day's noise is gone before we aggregate.
+
+const FEEDBACK_AGGREGATOR_QUEUE = "feedback-aggregator";
+
+const FEEDBACK_AGGREGATOR_CRON = "0 4 * * *";
+// v0.5.4 ios-coord — daily mood-reminder dispatch-ledger retention sweep.
+// Rows older than 90 days are behavioural footprints of mood-log gaps; we
+// keep them long enough to debug a duplicate-push report (~one billing
+// cycle) but no longer. Slots between the audit-log cleanup (03:15) and the
+// drain (03:45).
+
+const MOOD_REMINDER_CLEANUP_QUEUE = "mood-reminder-cleanup";
+
+const MOOD_REMINDER_CLEANUP_CRON = "25 3 * * *";
+
+const PUSH_ATTEMPT_CLEANUP_QUEUE = "push-attempt-cleanup";
+
+const PUSH_ATTEMPT_CLEANUP_CRON = "35 3 * * *";
+
+const MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE = "measurement-tombstone-cleanup";
+
+const MEASUREMENT_TOMBSTONE_CLEANUP_CRON = "40 3 * * *";
+
+const allQueues = [
+  DATA_BACKUP_QUEUE,
+  RATE_LIMIT_CLEANUP_QUEUE,
+  IDEMPOTENCY_CLEANUP_QUEUE,
+  AUDIT_LOG_CLEANUP_QUEUE,
+  OFFHOST_BACKUP_QUEUE,
+  RESTORE_DRILL_QUEUE,
+  HOST_METRIC_QUEUE,
+  FEEDBACK_AGGREGATOR_QUEUE,
+  GEO_BACKFILL_QUEUE,
+  // v1.12.2 ios-coord — TLS leaf SPKI-change monitor. The iOS client
+  // pins the served leaf certificate; this queue probes it every 6 h and
+  // alarms when the served SPKI leaves the operator's known-good set.
+  TLS_PIN_MONITOR_QUEUE,
+  PR_DETECTION_QUEUE,
+  MEDICATION_INVENTORY_EXPIRE_QUEUE,
+  // v1.4.46 — hourly auto-skip pass for stale unmarked intakes. Without
+  // this entry the schedule silently no-ops and pending rows older than
+  // 24 h pile up unflipped.
+  INTAKE_AUTO_SKIP_QUEUE,
+  APPLE_HEALTH_IMPORT_QUEUE,
+  // v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery enqueues
+  // one job per user holding two live intake rows that snap to the same
+  // canonical slot (the pre-fix REMINDER-pending + API-taken pair). Also
+  // carries a daily discovery cron (03:28). Without this entry pg-boss never
+  // provisions it and both the boot enqueue and the cron silently no-op.
+  INTAKE_SLOT_DEDUP_QUEUE,
+  // v0.5.4 ios-coord — mood-reminder dispatch-ledger retention sweep.
+  MOOD_REMINDER_CLEANUP_QUEUE,
+  // v1.4.49 — push-attempt ledger cleanup. The daily schedule below would
+  // silently no-op without this entry.
+  PUSH_ATTEMPT_CLEANUP_QUEUE,
+  // v1.7.0 — soft-deleted measurement tombstone prune. Without this entry
+  // the daily schedule silently no-ops and pruned-past-retention tombstones
+  // pile up forever.
+  MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE,
+];
+
+const schedules: ScheduleEntry[] = [
+  [DATA_BACKUP_QUEUE, DATA_BACKUP_CRON],
+  [RATE_LIMIT_CLEANUP_QUEUE, RATE_LIMIT_CLEANUP_CRON],
+  [IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON],
+  [AUDIT_LOG_CLEANUP_QUEUE, AUDIT_LOG_CLEANUP_CRON],
+  [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON],
+  [RESTORE_DRILL_QUEUE, RESTORE_DRILL_CRON],
+  [HOST_METRIC_QUEUE, HOST_METRIC_CRON],
+  [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON],
+  // v1.4.37 — hourly geo backfill. The helper is idempotent + capped
+  // at 5 000 rows per pass; running it at :40 every hour catches the
+  // long tail of audit rows that landed with the offline MMDB
+  // missing or the online provider unreachable.
+  [GEO_BACKFILL_QUEUE, GEO_BACKFILL_CRON],
+  // v1.12.2 ios-coord — every-6-hour TLS leaf SPKI probe (:07 off the
+  // hourly sync crons). Surfaces a pinned-leaf rotation well inside the
+  // ≥11-day re-pin window the iOS release owner needs.
+  [TLS_PIN_MONITOR_QUEUE, TLS_PIN_MONITOR_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],
+  // v1.4.25 W19b — daily expire-stale pass for the per-pen inventory
+  // entities. Flips IN_USE rows whose 30-day clock has blown to
+  // EXPIRED at 03:30 Europe/Berlin (in the existing 02:xx–03:xx
+  // maintenance window, right after idempotency-cleanup).
+  [MEDICATION_INVENTORY_EXPIRE_QUEUE, MEDICATION_INVENTORY_EXPIRE_CRON],
+  // v1.4.46 — hourly auto-skip for medication intakes the user never
+  // marked. Cron `5 * * * *` slots off the top-of-the-hour
+  // reminder-check (:00) and the moodlog-sync (:30) so the three
+  // hourly ticks don't pile up on the same boss poll.
+  [INTAKE_AUTO_SKIP_QUEUE, INTAKE_AUTO_SKIP_CRON],
+  // v0.5.4 ios-coord — daily mood-reminder dispatch-ledger retention sweep.
+  [MOOD_REMINDER_CLEANUP_QUEUE, MOOD_REMINDER_CLEANUP_CRON],
+  // v1.15.19 — daily 03:28 Europe/Berlin duplicate dose-slot dedup
+  // discovery. The empty cron payload (no `userId`) is the handler's
+  // signal to run the discovery fan-out instead of a per-user pass, so
+  // cross-source duplicate slots created between deploys collapse within
+  // a day instead of waiting for the next worker reboot.
+  [INTAKE_SLOT_DEDUP_QUEUE, INTAKE_SLOT_DEDUP_CRON],
+  // v1.4.49 — daily 03:35 Europe/Berlin prune for push_attempts.
+  [PUSH_ATTEMPT_CLEANUP_QUEUE, PUSH_ATTEMPT_CLEANUP_CRON],
+  // v1.7.0 — daily 03:40 Europe/Berlin prune for expired measurement
+  // tombstones.
+  [MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE, MEASUREMENT_TOMBSTONE_CLEANUP_CRON],
+];
+
+/**
+ * Register every maintenance / ops / cleanup queue. Returns the queue names
+ * created (for the boot-level aggregate assertion).
+ */
+export async function registerMaintenanceQueues(
+  boss: PgBoss,
+): Promise<readonly string[]> {
+  await createAndSchedule(boss, allQueues, schedules);
+
+  await boss.work<DataBackupPayload>(
+    DATA_BACKUP_QUEUE,
+    { localConcurrency: 1 },
+    handleDataBackup,
+  );
+  await boss.work<RateLimitCleanupPayload>(
+    RATE_LIMIT_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleRateLimitCleanup,
+  );
+  await boss.work<IdempotencyCleanupPayload>(
+    IDEMPOTENCY_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleIdempotencyCleanup,
+  );
+  await boss.work<AuditLogCleanupPayload>(
+    AUDIT_LOG_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleAuditLogCleanup,
+  );
+  await boss.work<OffhostBackupPayload>(
+    OFFHOST_BACKUP_QUEUE,
+    { localConcurrency: 1 },
+    handleOffhostBackup,
+  );
+  // prettier-ignore
+  await boss.work(RESTORE_DRILL_QUEUE, { localConcurrency: 1 }, handleRestoreDrill);
+  await boss.work<HostMetricSamplePayload>(
+    HOST_METRIC_QUEUE,
+    { localConcurrency: 1 },
+    handleHostMetricSample,
+  );
+  await boss.work<FeedbackAggregatorPayload>(
+    FEEDBACK_AGGREGATOR_QUEUE,
+    { localConcurrency: 1 },
+    handleFeedbackAggregator,
+  );
+  await boss.work<GeoBackfillPayload>(
+    GEO_BACKFILL_QUEUE,
+    { localConcurrency: 1 },
+    handleGeoBackfill,
+  );
+  // v1.12.2 ios-coord — TLS leaf SPKI-change monitor. Single-flight: one
+  // short outbound TLS handshake per tick, no benefit to overlapping ticks.
+  await boss.work<TlsPinMonitorPayload>(
+    TLS_PIN_MONITOR_QUEUE,
+    { localConcurrency: 1 },
+    handleTlsPinMonitor,
+  );
+  await boss.work<MoodReminderCleanupPayload>(
+    MOOD_REMINDER_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleMoodReminderCleanup,
+  );
+  // v1.4.49 — daily prune of the push-attempt ledger. Single-flight
+  // matches every other cleanup queue; two ticks racing on the same
+  // DELETE statement is wasted work and the second tick's payload
+  // would be a no-op anyway.
+  await boss.work<PushAttemptCleanupPayload>(
+    PUSH_ATTEMPT_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handlePushAttemptCleanup,
+  );
+  // v1.7.0 — daily prune of expired measurement tombstones. Single-flight
+  // like every other cleanup queue.
+  await boss.work<MeasurementTombstoneCleanupPayload>(
+    MEASUREMENT_TOMBSTONE_CLEANUP_QUEUE,
+    { localConcurrency: 1 },
+    handleMeasurementTombstoneCleanup,
+  );
+  await boss.work<PrDetectionPayload>(
+    PR_DETECTION_QUEUE,
+    { localConcurrency: PR_DETECTION_CONCURRENCY },
+    handlePrDetection,
+  );
+  await boss.work<MedicationInventoryExpirePayload>(
+    MEDICATION_INVENTORY_EXPIRE_QUEUE,
+    { localConcurrency: 1 },
+    handleMedicationInventoryExpire,
+  );
+  // v1.4.46 — hourly auto-skip for stale unmarked intakes.
+  // Single-flight: two ticks racing against the same row pile is wasted
+  // work, and the underlying `updateMany` is the canonical idempotent
+  // shape anyway.
+  await boss.work<IntakeAutoSkipPayload>(
+    INTAKE_AUTO_SKIP_QUEUE,
+    { localConcurrency: 1 },
+    handleIntakeAutoSkip,
+  );
+  // v1.4.34 — Apple Health export.zip ingest worker. localConcurrency
+  // caps at 1 because the parse loop is CPU-bound and a concurrent
+  // second 1 GB import would race the first for RSS.
+  await boss.work<AppleHealthImportPayload>(
+    APPLE_HEALTH_IMPORT_QUEUE,
+    { localConcurrency: APPLE_HEALTH_IMPORT_CONCURRENCY },
+    async (jobs) => {
+      // pg-boss v12 work callbacks always receive an array (batched
+      // worker mode); for our concurrency-1 + batchSize-1 case we just
+      // process each job sequentially.
+      for (const job of jobs) {
+        await handleAppleHealthImport(job);
+      }
+    },
+  );
+
+  // v1.8.2 — duplicate dose-slot cleanup worker. The boot enqueue helper
+  // below sends one job per user holding two live intake rows that snap
+  // to the same canonical slot; this handler keeps the winner, soft-
+  // deletes the losers, normalises the winner's scheduledFor, and
+  // recomputes the affected compliance rollups. Serial concurrency so the
+  // pass never crowds the request pool.
+  //
+  // v1.15.19 — the queue also carries a daily cron tick (03:28, scheduled
+  // above) whose payload omits `userId`. That tick runs the SAME discovery
+  // fan-out the boot pass uses, so duplicate slots created between deploys
+  // collapse within a day instead of waiting for the next worker reboot.
+  await boss.work<Partial<IntakeSlotDedupPayload>>(
+    INTAKE_SLOT_DEDUP_QUEUE,
+    { localConcurrency: INTAKE_SLOT_DEDUP_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        if (!userId) {
+          // Daily discovery tick — fan out one per-user job per account
+          // still holding duplicate-slot candidates (singletonKey-coalesced,
+          // identical to the boot pass).
+          const result = await enqueueBootTimeIntakeSlotDedup();
+          workerLog(
+            "info",
+            `[intake-slot-dedup] daily discovery enqueued=${result.enqueued} skipped=${result.skipped}${result.error ? ` error=${result.error}` : ""}`,
+          );
+          continue;
+        }
+        try {
+          const summary = await dedupeUserIntakeSlots(userId);
+          workerLog(
+            "info",
+            `[intake-slot-dedup] user=${userId} slotsCollapsed=${summary.slotsCollapsed} rowsSoftDeleted=${summary.rowsSoftDeleted} rowsNormalised=${summary.rowsNormalised} daysRecomputed=${summary.daysRecomputed}`,
+          );
+        } catch (err) {
+          recordError();
+          workerLog("error", `[intake-slot-dedup] user=${userId} failed`, err);
+          throw err;
+        }
+      }
+    },
+  );
+
+  return allQueues;
+}
+
+/**
+ * Fire-and-forget boot discovery for the duplicate dose-slot cleanup. Finds
+ * every user holding two live intake rows within the drift window on the same
+ * medication and enqueues one dedup job per account. Idempotent across reboots:
+ * collapsed losers are soft-deleted so the `deleted_at IS NULL` discovery
+ * predicate drops them. Errors are returned through the helper's result value —
+ * the worker boot never fails because of a dedup miss.
+ */
+export async function enqueueMaintenanceBootDiscovery(): Promise<void> {
+  try {
+    const { enqueued, skipped, error } = await enqueueBootTimeIntakeSlotDedup();
+    if (error) {
+      workerLog("error", `[intake-slot-dedup] boot discovery failed: ${error}`);
+    } else {
+      workerLog(
+        "info",
+        `[intake-slot-dedup] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[intake-slot-dedup] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+}

src/lib/jobs/reminder/register-reminders.ts

@@ -0,0 +1,165 @@
+/**
+ * Reminder-dispatch queue registrar.
+ *
+ * Owns the user-facing reminder ticks: the medication overdue check, the daily
+ * mood reminder, the cycle reminder (period-soon + period-start-confirm), the
+ * Vorsorge (measurement) reminder, and the eventful reminder-satisfy queue.
+ *
+ * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
+ * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
+ * and a `boss.work(QUEUE, …, handler)` binding drains it. The
+ * `cycle-reminder-queue` and `measurement-reminder-queue` guards read THIS
+ * module.
+ */
+import { PgBoss } from "pg-boss";
+import {
+  REMINDER_SATISFY_QUEUE,
+  REMINDER_SATISFY_CONCURRENCY,
+  type ReminderSatisfyPayload,
+} from "@/lib/jobs/reminder-satisfy";
+import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+import {
+  ReminderCheckPayload,
+  handleReminderCheck,
+} from "./medication-reminder-check";
+import {
+  MoodReminderPayload,
+  CycleReminderPayload,
+  MeasurementReminderPayload,
+  handleMoodReminderCheck,
+  handleCycleReminderCheck,
+  handleMeasurementReminderCheck,
+  handleReminderSatisfy,
+} from "./mood-cycle-checks";
+
+const QUEUE_NAME = "medication-reminder-check";
+
+const CHECK_INTERVAL_CRON = "*/15 * * * *"; // every 15 minutes
+// v0.5.4 ios-coord — daily mood-reminder cron.
+//
+// Runs every 15 minutes so the handler can pick up users whose local
+// time has just crossed 22:00 across any IANA timezone without having
+// to schedule one cron entry per zone. The handler short-circuits when
+// the user's local hour isn't 22, so the 15-min cadence translates to
+// ~4 ticks-per-hour × 1 actual-dispatch-window-per-user = at most one
+// push per user per day. Idempotency is enforced by the
+// `MoodReminderDispatch` ledger inside the handler.
+
+const MOOD_REMINDER_QUEUE = "mood-reminder-check";
+
+const MOOD_REMINDER_CRON = "*/15 * * * *";
+// v1.15 — daily cycle reminder cron (period-soon + period-start-confirm).
+//
+// Runs every 15 minutes for the same reason as the mood reminder: the
+// handler short-circuits unless the candidate user's local time is the
+// cycle-reminder hour (09:00), so the 15-min cadence picks up every IANA
+// timezone crossing that hour without one cron entry per zone. At most one
+// push per event per user per local day — the `push_attempts` ledger is the
+// idempotency anchor inside the handler.
+
+const CYCLE_REMINDER_QUEUE = "cycle-reminder-check";
+
+const CYCLE_REMINDER_CRON = "*/15 * * * *";
+
+// v1.17.1 — every-15-min tick for Vorsorge (measurement) reminders.
+// Same cadence + short-circuit shape as the mood / cycle reminders: the
+// handler only fires a reminder whose `nextDueAt` is past AND whose local
+// time is the reminder's `notifyHour`, so the 15-min cadence picks up
+// every IANA timezone crossing that hour without one cron entry per zone.
+// Dedup is the reminder's own `nextDueAt` advance — no ledger table.
+
+const MEASUREMENT_REMINDER_QUEUE = "measurement-reminder-check";
+
+const MEASUREMENT_REMINDER_CRON = "*/15 * * * *";
+
+const allQueues = [
+  QUEUE_NAME,
+  // v0.5.4 ios-coord — mood-reminder cron tick. Same pg-boss v12
+  // createQueue contract as the drain queue; without this entry the
+  // every-15-min schedule silently no-ops and the dispatcher never fires.
+  MOOD_REMINDER_QUEUE,
+  // v1.15 — cycle-reminder cron tick (period-soon + period-start-confirm).
+  // Without this entry the every-15-min schedule silently no-ops and the
+  // cycle dispatcher never fires (the v1.4.37 dead-queue class).
+  CYCLE_REMINDER_QUEUE,
+  // v1.17.1 — Vorsorge (measurement) reminder cron tick. Without this entry
+  // pg-boss never provisions the queue and the every-15-min schedule
+  // silently no-ops (the v1.4.37 dead-queue class).
+  MEASUREMENT_REMINDER_QUEUE,
+  // v1.18.1 — eventful Vorsorge satisfaction. No cron of its own (the
+  // 15-min measurement-reminder check is the safety-net); enqueued by the
+  // ingest paths. Must still be registered here or the worker binding
+  // below never provisions the queue (v1.4.37 dead-queue lesson).
+  REMINDER_SATISFY_QUEUE,
+];
+
+const schedules: ScheduleEntry[] = [
+  [QUEUE_NAME, CHECK_INTERVAL_CRON],
+  // v0.5.4 ios-coord — every-15-min tick for the daily mood reminder.
+  // The handler short-circuits unless the candidate user's local
+  // time is the 22:00 hour, so the cron costs ~one user-row scan
+  // per tick for the entire opted-in cohort.
+  [MOOD_REMINDER_QUEUE, MOOD_REMINDER_CRON],
+  // v1.15 — every-15-min tick for the daily cycle reminder. The handler
+  // short-circuits unless the candidate user's local time is the 09:00
+  // hour, so the cron costs ~one prediction-row scan per tick for the
+  // opted-in cohort.
+  [CYCLE_REMINDER_QUEUE, CYCLE_REMINDER_CRON],
+  // v1.17.1 — every-15-min tick for the Vorsorge (measurement) reminder.
+  // The handler short-circuits unless a reminder is past-due AND the
+  // user's local time matches the reminder's notifyHour, so the cron
+  // costs ~one reminder-row scan per tick for the active cohort.
+  [MEASUREMENT_REMINDER_QUEUE, MEASUREMENT_REMINDER_CRON],
+];
+
+/**
+ * Register every reminder-dispatch queue. Returns the queue names created (for
+ * the boot-level aggregate assertion).
+ */
+export async function registerReminderQueues(
+  boss: PgBoss,
+): Promise<readonly string[]> {
+  await createAndSchedule(boss, allQueues, schedules);
+
+  await boss.work<ReminderCheckPayload>(
+    QUEUE_NAME,
+    { localConcurrency: 1 },
+    handleReminderCheck,
+  );
+  // v0.5.4 ios-coord — single-flight worker. localConcurrency=1 keeps
+  // two reminder ticks from interleaving against the same user row;
+  // the dedup ledger would still save us, but skipping the race here
+  // avoids spurious P2002 errors in the wide-event log.
+  await boss.work<MoodReminderPayload>(
+    MOOD_REMINDER_QUEUE,
+    { localConcurrency: 1 },
+    handleMoodReminderCheck,
+  );
+  // v1.15 — single-flight cycle-reminder worker. localConcurrency=1 keeps
+  // two ticks from racing the fire-and-forget `push_attempts` ledger that
+  // anchors the per-day idempotency, exactly like the mood-reminder worker.
+  await boss.work<CycleReminderPayload>(
+    CYCLE_REMINDER_QUEUE,
+    { localConcurrency: 1 },
+    handleCycleReminderCheck,
+  );
+  // v1.17.1 — single-flight Vorsorge (measurement) reminder worker.
+  // localConcurrency=1 keeps two ticks from racing the `nextDueAt`
+  // advance that anchors the per-cycle idempotency, exactly like the
+  // mood / cycle reminder workers.
+  await boss.work<MeasurementReminderPayload>(
+    MEASUREMENT_REMINDER_QUEUE,
+    { localConcurrency: 1 },
+    handleMeasurementReminderCheck,
+  );
+  // v1.18.1 — eventful Vorsorge satisfaction. Resolves a user's reminders
+  // against their just-landed measurement / lab. Read-heavy on the user's
+  // own data; the same small concurrency budget as PR detection.
+  await boss.work<ReminderSatisfyPayload>(
+    REMINDER_SATISFY_QUEUE,
+    { localConcurrency: REMINDER_SATISFY_CONCURRENCY },
+    handleReminderSatisfy,
+  );
+
+  return allQueues;
+}

src/lib/jobs/reminder/register-rollup.ts

@@ -0,0 +1,599 @@
+/**
+ * Rollup / consolidation queue registrar.
+ *
+ * Owns the pre-aggregation + consolidation tier: measurement-rollup recompute
+ * + boot-backfill, mood-rollup recompute + boot-backfill, medication-compliance
+ * boot-backfill, legacy step consolidation, daily-mean consolidation, dense
+ * intra-day retention, and the nightly per-sample cumulative drain (which also
+ * folds the daily-mean + dense-retention passes onto the same tick). Includes
+ * the boot-time discovery enqueues for the self-converging backfills.
+ *
+ * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
+ * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
+ * and a `boss.work(QUEUE, …, handler)` binding drains it. The
+ * `drain-cumulative-queue`, `mean-consolidation-queue`, and
+ * `stress-strain-retention-queue` (dense-retention facet) guards read THIS
+ * module.
+ */
+import { PgBoss } from "pg-boss";
+import { reportWorkerError } from "@/lib/jobs/report-worker-error";
+import { recordError } from "@/lib/jobs/worker-status";
+import {
+  ROLLUP_FULL_BACKFILL_QUEUE,
+  ROLLUP_FULL_BACKFILL_CONCURRENCY,
+  ROLLUP_RECOMPUTE_QUEUE,
+  ROLLUP_RECOMPUTE_CONCURRENCY,
+  enqueueBootTimeRollupBackfill,
+  recomputeUserRollups,
+  type RollupFullBackfillPayload,
+  type RollupRecomputePayload,
+} from "@/lib/rollups/measurement-rollups";
+import {
+  MOOD_ROLLUP_FULL_BACKFILL_QUEUE,
+  MOOD_ROLLUP_FULL_BACKFILL_CONCURRENCY,
+  MOOD_ROLLUP_RECOMPUTE_QUEUE,
+  MOOD_ROLLUP_RECOMPUTE_CONCURRENCY,
+  enqueueBootTimeMoodRollupBackfill,
+  recomputeUserMoodRollups,
+  type MoodRollupFullBackfillPayload,
+  type MoodRollupRecomputePayload,
+} from "@/lib/rollups/mood-rollups";
+import {
+  MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
+  MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY,
+  recomputeUserMedicationCompliance,
+  enqueueBootTimeMedicationComplianceBackfill,
+  type MedicationComplianceBackfillPayload,
+} from "@/lib/rollups/medication-compliance-rollups";
+import {
+  drainPerSampleCumulative,
+  DRAIN_CUMULATIVE_CUTOFF_HOURS,
+} from "@/lib/measurements/drain-per-sample-cumulative";
+import {
+  consolidateDailyMean,
+  MEAN_CONSOLIDATION_CUTOFF_HOURS,
+} from "@/lib/measurements/consolidate-daily-mean";
+import {
+  STEP_CONSOLIDATION_QUEUE,
+  STEP_CONSOLIDATION_CONCURRENCY,
+  runStepConsolidationForUser,
+  enqueueBootTimeStepConsolidation,
+  type StepConsolidationPayload,
+} from "@/lib/jobs/step-consolidation";
+import {
+  MEAN_CONSOLIDATION_QUEUE,
+  MEAN_CONSOLIDATION_CONCURRENCY,
+  runMeanConsolidationForUser,
+  enqueueBootTimeMeanConsolidation,
+  type MeanConsolidationPayload,
+} from "@/lib/jobs/mean-consolidation";
+import {
+  DENSE_INTRADAY_RETENTION_QUEUE,
+  DENSE_INTRADAY_RETENTION_CONCURRENCY,
+  runDenseIntradayRetentionForUser,
+  enqueueBootTimeDenseIntradayRetention,
+  DENSE_INTRADAY_RETENTION_ENABLED,
+  type DenseIntradayRetentionPayload,
+} from "@/lib/jobs/dense-intraday-retention";
+import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention";
+import { getWorkerPrisma, workerLog } from "./shared";
+import { createAndSchedule, type ScheduleEntry } from "./registrar-shared";
+// v1.4.37 W7c — nightly drain of per-sample APPLE_HEALTH cumulative rows.
+// Collapses each user × cumulative-type × calendar-day bucket into one
+// `stats:…` row so the list view stops painting hundreds of step chunks per
+// day. 03:45 Europe/Berlin slots in between the 03:15 audit-log cleanup and
+// the 04:00 feedback aggregator. The 36-hour grace window keeps today + the
+// trailing watch-sync window intact for real-time visibility; only
+// completed-and-stable days fall to the drain.
+
+const DRAIN_CUMULATIVE_QUEUE = "drain-per-sample-cumulative";
+
+const DRAIN_CUMULATIVE_CRON = "45 3 * * *";
+
+interface DrainCumulativePayload {
+  triggeredAt: string;
+}
+
+const allQueues = [
+  ROLLUP_RECOMPUTE_QUEUE,
+  ROLLUP_FULL_BACKFILL_QUEUE,
+  // v1.4.39 W-MOOD — per-bucket WEEK/MONTH/YEAR fold queue for the
+  // mood rollup tier. The DAY pass runs inline in the write hook;
+  // these enqueue paths cover the cross-granularity buckets the
+  // worker materialises off the request path.
+  MOOD_ROLLUP_RECOMPUTE_QUEUE,
+  // v1.4.39 W-MOOD — boot-time fold queue for the mood rollup tier.
+  MOOD_ROLLUP_FULL_BACKFILL_QUEUE,
+  // v1.4.39 W-MED — boot-time fold for the medication-compliance
+  // rollup tier. Discovery enqueues one job per user with intake
+  // events but no rollup coverage; idempotent across reboots.
+  MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
+  // v1.5.6 — boot-time legacy step consolidation. Discovery enqueues
+  // one job per user still holding live pre-v1.5.0 granular step rows.
+  STEP_CONSOLIDATION_QUEUE,
+  // v1.7.0 — daily-mean consolidation for high-frequency spot HealthKit
+  // metrics (walking speed/step length, respiratory rate, audio exposure).
+  MEAN_CONSOLIDATION_QUEUE,
+  // v1.10.0 — computed scores (WX-E). Dense intra-day retention drain of
+  // daytime HRV / HR samples (per-user backfill queue, boot-discovery
+  // driven like mean-consolidation; the steady-state nightly walk folds
+  // onto the drain-cumulative tick).
+  DENSE_INTRADAY_RETENTION_QUEUE,
+  // v1.4.37 W7c — explicit createQueue is required before the nightly
+  // schedule below registers (pg-boss v12 contract). Without this entry the
+  // drain schedule silently no-ops and the per-sample APPLE_HEALTH rows
+  // never collapse.
+  DRAIN_CUMULATIVE_QUEUE,
+];
+
+const schedules: ScheduleEntry[] = [
+  // v1.4.37 W7c — nightly fold of per-sample APPLE_HEALTH cumulative
+  // rows into one row per day per type. Slots between the
+  // audit-log cleanup (03:15) and the feedback aggregator (04:00).
+  [DRAIN_CUMULATIVE_QUEUE, DRAIN_CUMULATIVE_CRON],
+];
+
+/**
+ * Register every rollup / consolidation queue. Returns the queue names created
+ * (for the boot-level aggregate assertion).
+ */
+export async function registerRollupQueues(
+  boss: PgBoss,
+): Promise<readonly string[]> {
+  await createAndSchedule(boss, allQueues, schedules);
+
+  // v1.5.0 — persistent measurement rollup worker. Folds the
+  // WEEK / MONTH / YEAR buckets that the write-path hooks enqueue;
+  // the DAY bucket is already recomputed synchronously by the hook
+  // itself. Concurrency-2 keeps two recomputes in flight without
+  // crowding the dashboard request pool.
+  await boss.work<RollupRecomputePayload>(
+    ROLLUP_RECOMPUTE_QUEUE,
+    { localConcurrency: ROLLUP_RECOMPUTE_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const payload = job.data;
+        await recomputeUserRollups(payload.userId, {
+          types: [payload.type],
+          granularities: [payload.granularity],
+          from: new Date(payload.from),
+          to: new Date(payload.to),
+        });
+      }
+    },
+  );
+
+  // v1.4.35.1 — boot-time full-fold worker. The boot enqueue helper
+  // below sends one job per uncovered user; this handler runs the
+  // full `recomputeUserRollups` against the default 5-year window
+  // across every granularity. Serial concurrency so the populator
+  // never crowds the dashboard request pool.
+  await boss.work<RollupFullBackfillPayload>(
+    ROLLUP_FULL_BACKFILL_QUEUE,
+    { localConcurrency: ROLLUP_FULL_BACKFILL_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        const { rowsUpserted, durationMs } = await recomputeUserRollups(userId);
+        workerLog(
+          "info",
+          `[rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
+        );
+      }
+    },
+  );
+
+  // v1.5.6 — legacy step consolidation worker. The boot enqueue helper
+  // below sends one job per user still holding live pre-v1.5.0 granular
+  // step rows; this handler collapses them into one daily-total row per
+  // calendar day and soft-deletes the originals. Serial concurrency so
+  // the populator never crowds the dashboard request pool.
+  await boss.work<StepConsolidationPayload>(
+    STEP_CONSOLIDATION_QUEUE,
+    { localConcurrency: STEP_CONSOLIDATION_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        const { daysConsolidated, legacyRowsSoftDeleted } =
+          await runStepConsolidationForUser(userId);
+        workerLog(
+          "info",
+          `[step-consolidation] user=${userId} days=${daysConsolidated} legacyRowsSoftDeleted=${legacyRowsSoftDeleted}`,
+        );
+      }
+    },
+  );
+
+  // v1.7.0 — daily-mean consolidation worker. The boot enqueue helper
+  // below sends one job per user holding live per-sample high-frequency
+  // mean-type rows; this handler collapses each completed day to its
+  // mean and soft-deletes the originals. Serial concurrency so the
+  // populator never crowds the dashboard request pool.
+  await boss.work<MeanConsolidationPayload>(
+    MEAN_CONSOLIDATION_QUEUE,
+    { localConcurrency: MEAN_CONSOLIDATION_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        const { daysConsolidated, perSampleRowsSoftDeleted } =
+          await runMeanConsolidationForUser(userId);
+        workerLog(
+          "info",
+          `[mean-consolidation] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted}`,
+        );
+      }
+    },
+  );
+
+  // v1.10.0 WX-E — dense intra-day retention per-user backfill worker. The
+  // boot enqueue helper below sends one job per user holding live per-sample
+  // dense-tier (HRV / HR) rows older than the retention window; this handler
+  // folds those out-of-window samples to a daily mean and soft-deletes the
+  // originals, keeping the in-window intra-day shape intact for the Stress
+  // engine. Serial concurrency so the backfill never crowds the request pool.
+  await boss.work<DenseIntradayRetentionPayload>(
+    DENSE_INTRADAY_RETENTION_QUEUE,
+    { localConcurrency: DENSE_INTRADAY_RETENTION_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        try {
+          const { daysConsolidated, perSampleRowsSoftDeleted } =
+            await runDenseIntradayRetentionForUser(userId);
+          workerLog(
+            "info",
+            `[dense-intraday-retention] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted}`,
+          );
+        } catch (err) {
+          recordError();
+          workerLog(
+            "error",
+            `[dense-intraday-retention] user=${userId} failed`,
+            err,
+          );
+          throw err;
+        }
+      }
+    },
+  );
+
+  // v1.4.39 W-MOOD — mood-rollup per-bucket worker. Folds the
+  // WEEK / MONTH / YEAR buckets that the mood-entry write hooks
+  // enqueue; the DAY bucket runs synchronously in the hook itself.
+  // No current read path consumes these buckets — they exist so a
+  // future cross-granularity reader can ship without a backfill
+  // step. Concurrency-2 mirrors the measurement-rollup worker.
+  await boss.work<MoodRollupRecomputePayload>(
+    MOOD_ROLLUP_RECOMPUTE_QUEUE,
+    { localConcurrency: MOOD_ROLLUP_RECOMPUTE_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const payload = job.data;
+        await recomputeUserMoodRollups(payload.userId, {
+          granularities: [payload.granularity],
+          from: new Date(payload.from),
+          to: new Date(payload.to),
+        });
+      }
+    },
+  );
+
+  // v1.4.39 W-MOOD — mood-rollup boot-time fold worker. The boot
+  // discovery helper below sends one job per user with mood entries
+  // but zero rollup rows; this handler folds the full 5-year window
+  // across every granularity. Concurrency-1 so the populator never
+  // crowds the request pool.
+  await boss.work<MoodRollupFullBackfillPayload>(
+    MOOD_ROLLUP_FULL_BACKFILL_QUEUE,
+    { localConcurrency: MOOD_ROLLUP_FULL_BACKFILL_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        try {
+          const { rowsUpserted, durationMs } =
+            await recomputeUserMoodRollups(userId);
+          workerLog(
+            "info",
+            `[mood-rollup-full-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
+          );
+        } catch (err) {
+          recordError();
+          workerLog(
+            "error",
+            `[mood-rollup-full-backfill] user=${userId} failed`,
+            err,
+          );
+          throw err;
+        }
+      }
+    },
+  );
+
+  // v1.4.39 W-MED — medication-compliance boot-backfill worker. The
+  // discovery helper below sends one job per user with intake events
+  // but zero rollup coverage; this handler folds the trailing 90-day
+  // window per account. Concurrency-1 so the populator never crowds
+  // the request pool.
+  await boss.work<MedicationComplianceBackfillPayload>(
+    MEDICATION_COMPLIANCE_BACKFILL_QUEUE,
+    { localConcurrency: MEDICATION_COMPLIANCE_BACKFILL_CONCURRENCY },
+    async (jobs) => {
+      for (const job of jobs) {
+        const { userId } = job.data;
+        try {
+          const { rowsUpserted, durationMs } =
+            await recomputeUserMedicationCompliance(userId);
+          workerLog(
+            "info",
+            `[medication-compliance-backfill] user=${userId} rows=${rowsUpserted} duration=${durationMs}ms`,
+          );
+        } catch (err) {
+          recordError();
+          workerLog(
+            "error",
+            `[medication-compliance-backfill] user=${userId} failed`,
+            err,
+          );
+          throw err;
+        }
+      }
+    },
+  );
+
+  // v1.4.37 W7c — nightly drain worker. Walks every user × cumulative
+  // type and folds per-sample APPLE_HEALTH rows older than the cutoff
+  // into one `stats:…` row per calendar day. Idempotent — a second run
+  // collapses zero buckets once every day is in the `stats:` shape.
+  // Concurrency-1 so the drain never crowds the dashboard request pool
+  // and a long backfill on the maintainer's account (300 k+ measurement rows)
+  // stays a single sequential walk.
+  await boss.work<DrainCumulativePayload>(
+    DRAIN_CUMULATIVE_QUEUE,
+    { localConcurrency: 1 },
+    async (jobs) => {
+      for (const job of jobs) {
+        try {
+          const summary = await drainPerSampleCumulative(getWorkerPrisma(), {
+            dryRun: false,
+            cutoffHours: DRAIN_CUMULATIVE_CUTOFF_HOURS,
+            log: (line) => workerLog("info", line),
+          });
+          workerLog(
+            "info",
+            `[drain-cumulative] triggeredAt=${job.data.triggeredAt} usersScanned=${summary.totals.usersScanned} bucketsCollapsed=${summary.totals.bucketsCollapsed} perSampleRowsDeleted=${summary.totals.perSampleRowsDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted}`,
+          );
+        } catch (err) {
+          recordError();
+          await reportWorkerError("drain-cumulative", err);
+        }
+
+        // v1.8.5 — fold the daily-MEAN drain onto the same nightly tick as
+        // the cumulative (SUM) drain. The mean drain was previously
+        // boot-discovery only, so between worker reboots new high-frequency
+        // spot samples (walking speed, respiratory rate, gait/mobility,
+        // audio exposure) accumulated raw. Running both passes on one
+        // concurrency-1 cron keeps the maintenance window a single
+        // sequential walk and never crowds the request pool. The global
+        // (no `userId`) signature drains every user; the 36-hour grace
+        // cutoff keeps today's in-flight watch syncs raw. Boot discovery
+        // stays as the back-fill for accounts that accumulated raw rows
+        // before this cron shipped.
+        try {
+          const meanSummary = await consolidateDailyMean(getWorkerPrisma(), {
+            dryRun: false,
+            cutoffHours: MEAN_CONSOLIDATION_CUTOFF_HOURS,
+            log: (line) => workerLog("info", line),
+          });
+          workerLog(
+            "info",
+            `[mean-consolidation] triggeredAt=${job.data.triggeredAt} usersScanned=${meanSummary.totals.usersScanned} daysConsolidated=${meanSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${meanSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${meanSummary.totals.dailyRowsUpserted} daysFailed=${meanSummary.totals.daysFailed}`,
+          );
+        } catch (err) {
+          recordError();
+          await reportWorkerError("mean-consolidation", err, {
+            tick: "nightly",
+          });
+        }
+
+        // v1.10.0 WX-E — fold the dense intra-day retention drain onto the
+        // same nightly tick. Unlike the daily-mean drain, this scopes to the
+        // dense-tier types (HEART_RATE_VARIABILITY, PULSE) and keeps the last
+        // DENSE_INTRADAY_RETENTION_DAYS of raw per-sample rows so the Stress
+        // engine still sees the intra-day SDNN shape; only out-of-window
+        // samples fold to a daily mean. These two types are NEVER in the
+        // destructive HIGH_FREQUENCY_MEAN_TYPES allowlist — the drain
+        // exemption the intra-day shape depends on. The global (no `userId`)
+        // signature drains every user. Boot discovery (below) back-fills
+        // accounts that accumulated out-of-window raw rows before this
+        // shipped.
+        try {
+          if (!DENSE_INTRADAY_RETENTION_ENABLED) {
+            workerLog(
+              "info",
+              "[dense-intraday-retention] disabled via DENSE_INTRADAY_RETENTION_ENABLED — skipping the nightly walk (operator kill-switch)",
+            );
+          } else {
+            const denseSummary = await runDenseIntradayRetention(
+              getWorkerPrisma(),
+              {
+                dryRun: false,
+                log: (line) => workerLog("info", line),
+              },
+            );
+            workerLog(
+              "info",
+              `[dense-intraday-retention] triggeredAt=${job.data.triggeredAt} usersScanned=${denseSummary.totals.usersScanned} daysConsolidated=${denseSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${denseSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${denseSummary.totals.dailyRowsUpserted}`,
+            );
+          }
+        } catch (err) {
+          recordError();
+          await reportWorkerError("dense-intraday-retention", err, {
+            tick: "nightly",
+          });
+        }
+      }
+    },
+  );
+
+  return allQueues;
+}
+
+/**
+ * Fire-and-forget boot discovery for the self-converging rollup / consolidation
+ * backfills. Each pass is idempotent across reboots and never fails worker boot
+ * on a miss (errors come back through the helper's result value).
+ */
+export async function enqueueRollupBootDiscovery(): Promise<void> {
+  // v1.4.35.1 — measurement-rollup backfill. Finds every user with
+  // measurements but no rollup coverage and enqueues a full-fold per account.
+  // The discovery query only matches accounts with zero rollup rows, so once
+  // a fold completes the user drops off the list.
+  try {
+    const { enqueued, skipped, error } = await enqueueBootTimeRollupBackfill();
+    if (error) {
+      workerLog(
+        "error",
+        `[rollup-full-backfill] boot discovery failed: ${error}`,
+      );
+    } else {
+      // v1.4.38.7 — log the discovery result on every boot, including
+      // the silent `enqueued=0 skipped=0` case. Without this, an
+      // operator chasing "analytics is slow" cannot tell whether the
+      // discovery query ran successfully (and found nothing to fold)
+      // vs. silently no-op'd. The line is one row per worker boot, so
+      // the log cost is negligible.
+      workerLog(
+        "info",
+        `[rollup-full-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[rollup-full-backfill] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.5.6 — legacy step consolidation. Finds every user still holding live
+  // pre-v1.5.0 granular step rows and enqueues one consolidation job per
+  // account. Consolidated legacy rows are soft-deleted, so the
+  // `deleted_at IS NULL` discovery predicate drops them.
+  try {
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeStepConsolidation();
+    if (error) {
+      workerLog("error", `[step-consolidation] boot discovery failed: ${error}`);
+    } else {
+      workerLog(
+        "info",
+        `[step-consolidation] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[step-consolidation] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.7.0 — daily-mean consolidation. Finds every user holding live
+  // per-sample high-frequency mean-type rows and enqueues one job per account.
+  // Consolidated rows are soft-deleted, so the discovery predicate drops them.
+  try {
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeMeanConsolidation();
+    if (error) {
+      workerLog("error", `[mean-consolidation] boot discovery failed: ${error}`);
+    } else {
+      workerLog(
+        "info",
+        `[mean-consolidation] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[mean-consolidation] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.10.0 WX-E — dense intra-day retention drain. Finds every user holding
+  // live per-sample dense-tier (HRV / HR) rows OLDER than the retention window
+  // and enqueues one job per account. Folded rows are soft-deleted, so the
+  // discovery predicate drops them.
+  try {
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeDenseIntradayRetention();
+    if (error) {
+      workerLog(
+        "error",
+        `[dense-intraday-retention] boot discovery failed: ${error}`,
+      );
+    } else {
+      workerLog(
+        "info",
+        `[dense-intraday-retention] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[dense-intraday-retention] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.4.39 W-MOOD — mood rollup tier. Mirrors the v1.4.35.1
+  // measurement-rollup pattern: one job per user with mood entries but no
+  // rollup coverage. Idempotent across reboots and singleton-keyed inside
+  // pg-boss so a fast restart while a backfill is queued doesn't double up.
+  try {
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeMoodRollupBackfill();
+    if (error) {
+      workerLog(
+        "error",
+        `[mood-rollup-full-backfill] boot discovery failed: ${error}`,
+      );
+    } else {
+      workerLog(
+        "info",
+        `[mood-rollup-full-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[mood-rollup-full-backfill] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+
+  // v1.4.39 W-MED — medication compliance rollup tier. Mirrors the v1.4.35.1
+  // pattern: one job per user with intake events but no rollup coverage.
+  // Idempotent across reboots and singleton-keyed inside pg-boss.
+  try {
+    const { enqueued, skipped, error } =
+      await enqueueBootTimeMedicationComplianceBackfill();
+    if (error) {
+      workerLog(
+        "error",
+        `[medication-compliance-backfill] boot discovery failed: ${error}`,
+      );
+    } else {
+      workerLog(
+        "info",
+        `[medication-compliance-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`,
+      );
+    }
+  } catch (err) {
+    workerLog(
+      "error",
+      "[medication-compliance-backfill] boot discovery threw an unexpected error",
+      err,
+    );
+  }
+}

src/lib/jobs/reminder/register-status.ts

@@ -0,0 +1,446 @@
+/**
+ * Status / insight / computed-score queue registrar.
+ *
+ * Owns the nightly per-metric insight-status ladder (general / BP / weight /
+ * pulse / BMI / mood / medication-compliance), the comprehensive-insight
+ * pre-generation, the on-demand per-metric status generation, the nightly
+ * computed scores (recovery / stress / strain), the proactive Coach nudge, the
+ * Coach memory refresh, the period-narrative warm, and the medication
+ * low-stock pass.
+ *
+ * v1.4.37 dead-queue contract: every queue name appears in `allQueues`, its
+ * cron (where it has one) appears as a `[QUEUE, CRON]` tuple in `schedules`,
+ * and a `boss.work(QUEUE, …, handler)` binding drains it. The status-wiring
+ * guards (`status-cron-candidates`, `insight-pregenerate`,
+ * `insight-status-generate`, `recovery-score-queue`,
+ * `stress-strain-retention-queue`, `period-narrative-warm`,
+ * `coach-memory-refresh-queue`) read THIS module.
+ */
+import { PgBoss } from "pg-boss";
+import { reportWorkerError } from "@/lib/jobs/report-worker-error";
+import { recordError } from "@/lib/jobs/worker-status";
+import {
+  INSIGHT_PREGENERATE_QUEUE,
+  INSIGHT_PREGENERATE_CRON,
+  type InsightPregeneratePayload,
+} from "@/lib/jobs/insight-pregenerate";
+import {
+  RECOVERY_SCORE_QUEUE,
+  RECOVERY_SCORE_CRON,
+  runRecoveryScore,
+} from "@/lib/jobs/recovery-score";
+import {
+  COACH_NUDGE_QUEUE,
+  COACH_NUDGE_CRON,
+  runCoachNudgeTick,
+} from "@/lib/jobs/coach-nudge";
+import {
+  MEDICATION_LOW_STOCK_QUEUE,
+  MEDICATION_LOW_STOCK_CRON,
+  runMedicationLowStockTick,
+} from "@/lib/jobs/medication-low-stock";
+import {
+  STRESS_SCORE_QUEUE,
+  STRESS_SCORE_CRON,
+  runStressScore,
+} from "@/lib/jobs/stress-score";
+import {
+  STRAIN_SCORE_QUEUE,
+  STRAIN_SCORE_CRON,
+  runStrainScore,
+} from "@/lib/jobs/strain-score";
+import {
+  PERIOD_NARRATIVE_QUEUE,
+  PERIOD_NARRATIVE_CRON,
+  runPeriodNarrativeWarm,
+  warmOneNarrative,
+  type PeriodNarrativePayload,
+} from "@/lib/jobs/period-narrative-warm";
+import {
+  COACH_MEMORY_REFRESH_QUEUE,
+  type CoachMemoryRefreshPayload,
+} from "@/lib/ai/coach/coach-memory-shared";
+import { runCoachMemoryRefresh } from "@/lib/ai/coach/coach-memory-refresh-worker";
+import {
+  INSIGHT_STATUS_GENERATE_QUEUE,
+  INSIGHT_STATUS_GENERATE_CONCURRENCY,
+  type InsightStatusGeneratePayload,
+} from "@/lib/jobs/insight-status-generate";
+import { withBackgroundEvent } from "@/lib/logging/background";
+import { getWorkerPrisma, workerLog } from "./shared";
+import {
+  createAndSchedule,
+  insightRetryOptions,
+  type ScheduleEntry,
+} from "./registrar-shared";
+import {
+  GeneralStatusPayload,
+  BloodPressureStatusPayload,
+  WeightStatusPayload,
+  PulseStatusPayload,
+  BmiStatusPayload,
+  MoodStatusPayload,
+  MedicationComplianceStatusPayload,
+  handleGeneralStatusGenerate,
+  handleBloodPressureStatusGenerate,
+  handleWeightStatusGenerate,
+  handlePulseStatusGenerate,
+  handleBmiStatusGenerate,
+  handleMoodStatusGenerate,
+  handleMedicationComplianceStatusGenerate,
+  handleInsightPregenerateJob,
+  handleInsightStatusGenerate,
+} from "./insights-handlers";
+
+const GENERAL_STATUS_QUEUE = "insights-general-status";
+
+const GENERAL_STATUS_CRON = "0 2 * * *"; // daily at 02:00
+
+const BLOOD_PRESSURE_STATUS_QUEUE = "insights-blood-pressure-status";
+
+const BLOOD_PRESSURE_STATUS_CRON = "5 2 * * *"; // daily at 02:05
+
+const WEIGHT_STATUS_QUEUE = "insights-weight-status";
+
+const WEIGHT_STATUS_CRON = "10 2 * * *"; // daily at 02:10
+
+const PULSE_STATUS_QUEUE = "insights-pulse-status";
+
+const PULSE_STATUS_CRON = "15 2 * * *"; // daily at 02:15
+
+const BMI_STATUS_QUEUE = "insights-bmi-status";
+
+const BMI_STATUS_CRON = "20 2 * * *"; // daily at 02:20
+// v1.15.20 — mood joins the nightly per-metric status ladder. Same gate +
+// discovery as the six older crons (see status-cron-candidates.ts); 02:30
+// continues the 5-minute stagger after BMI (02:20) and compliance (02:25).
+
+const MOOD_STATUS_QUEUE = "insights-mood-status";
+
+const MOOD_STATUS_CRON = "30 2 * * *"; // daily at 02:30
+
+const MEDICATION_COMPLIANCE_STATUS_QUEUE =
+  "insights-medication-compliance-status";
+
+const MEDICATION_COMPLIANCE_STATUS_CRON = "25 2 * * *"; // daily at 02:25
+
+const allQueues = [
+  GENERAL_STATUS_QUEUE,
+  BLOOD_PRESSURE_STATUS_QUEUE,
+  WEIGHT_STATUS_QUEUE,
+  PULSE_STATUS_QUEUE,
+  BMI_STATUS_QUEUE,
+  // v1.15.20 — mood joins the nightly status ladder. The queue MUST be
+  // registered here or pg-boss never provisions it and the 02:30 schedule
+  // silently no-ops (the v1.4.37 dead-queue class).
+  MOOD_STATUS_QUEUE,
+  MEDICATION_COMPLIANCE_STATUS_QUEUE,
+  // v1.7.0 — nightly comprehensive-insight pre-generation so the daily
+  // briefing is warm before the user opens /insights or the dashboard
+  // snapshot. Without this entry the 04:30 schedule silently no-ops.
+  INSIGHT_PREGENERATE_QUEUE,
+  // v1.8.3 — on-demand per-metric status generation. The read-only status
+  // route enqueues here on a cold card; without this entry pg-boss never
+  // provisions the queue and the enqueue silently drops. No cron schedule —
+  // a send-only queue driven by navigation.
+  INSIGHT_STATUS_GENERATE_QUEUE,
+  // v1.10.0 — computed scores (WX-C / WX-E). Nightly Recovery / Stress /
+  // Strain compute + store. The queue MUST be registered here or pg-boss
+  // never provisions it and the nightly schedule silently never fires.
+  RECOVERY_SCORE_QUEUE,
+  STRESS_SCORE_QUEUE,
+  STRAIN_SCORE_QUEUE,
+  // v1.11.0 — nightly period-narrative warm + single-user warm enqueued by
+  // the read-only narrative GET.
+  PERIOD_NARRATIVE_QUEUE,
+  // v1.11.1 — combined Coach memory-refresh (rolling conversation summary +
+  // durable fact extraction), enqueued fire-and-forget from a long chat turn.
+  COACH_MEMORY_REFRESH_QUEUE,
+  // v1.15.20 — proactive Coach nudge. Without this entry the daily 05:15
+  // schedule silently no-ops and no nudge ever fires.
+  COACH_NUDGE_QUEUE,
+  // v1.16.11 — daily medication low-stock pass. Without this entry the 09:00
+  // schedule silently no-ops and no low-stock alert ever fires.
+  MEDICATION_LOW_STOCK_QUEUE,
+];
+
+const schedules: ScheduleEntry[] = [
+  [GENERAL_STATUS_QUEUE, GENERAL_STATUS_CRON, insightRetryOptions],
+  [BLOOD_PRESSURE_STATUS_QUEUE, BLOOD_PRESSURE_STATUS_CRON, insightRetryOptions],
+  [WEIGHT_STATUS_QUEUE, WEIGHT_STATUS_CRON, insightRetryOptions],
+  [PULSE_STATUS_QUEUE, PULSE_STATUS_CRON, insightRetryOptions],
+  [BMI_STATUS_QUEUE, BMI_STATUS_CRON, insightRetryOptions],
+  // v1.15.20 — mood status nightly, continuing the 02:xx ladder.
+  [MOOD_STATUS_QUEUE, MOOD_STATUS_CRON, insightRetryOptions],
+  [
+    MEDICATION_COMPLIANCE_STATUS_QUEUE,
+    MEDICATION_COMPLIANCE_STATUS_CRON,
+    insightRetryOptions,
+  ],
+  // v1.7.0 — nightly 04:30 Europe/Berlin comprehensive-insight
+  // pre-generation. Budget-gated per user inside the handler.
+  [INSIGHT_PREGENERATE_QUEUE, INSIGHT_PREGENERATE_CRON, insightRetryOptions],
+  // v1.10.0 — computed scores (WX-C). Nightly 04:45 Europe/Berlin
+  // Recovery-score compute + store, after the rollup-feeding consolidation
+  // + drain so the signals it reads are already folded.
+  [RECOVERY_SCORE_QUEUE, RECOVERY_SCORE_CRON],
+  // v1.10.0 — computed scores (WX-E). Nightly 04:50 Europe/Berlin
+  // Stress-score compute + store, after the dense intra-day retention
+  // drain so the HRV inputs it reads are settled.
+  [STRESS_SCORE_QUEUE, STRESS_SCORE_CRON],
+  // v1.10.0 — computed scores (WX-E). Nightly 04:55 Europe/Berlin
+  // Strain-score compute + store, after the recovery + stress passes so
+  // the nightly score writes stay ordered.
+  [STRAIN_SCORE_QUEUE, STRAIN_SCORE_CRON],
+  // v1.11.0 — nightly 05:05 Europe/Berlin period-narrative warm. The
+  // handler only fans out on a week (Mon) / month (1st) boundary; every
+  // other night is a cheap no-op. Budget-gated per user inside the runner.
+  [PERIOD_NARRATIVE_QUEUE, PERIOD_NARRATIVE_CRON, insightRetryOptions],
+  // v1.15.20 — daily 05:15 Europe/Berlin proactive Coach nudge, after
+  // the 04:45–04:55 score crons so the recovery-score trigger reads
+  // settled rows. Deterministic triggers only — no AI call on this path.
+  [COACH_NUDGE_QUEUE, COACH_NUDGE_CRON],
+  // v1.16.11 — medication low-stock pass at 09:00 Europe/Berlin: a
+  // supply alert is an errand prompt, so it fires at a time the user
+  // can act on it. Once daily; the per-medication stamp keeps it at
+  // one push per threshold crossing.
+  [MEDICATION_LOW_STOCK_QUEUE, MEDICATION_LOW_STOCK_CRON],
+];
+
+/**
+ * Register every status / insight / computed-score queue. Returns the queue
+ * names created (for the boot-level aggregate assertion).
+ */
+export async function registerStatusQueues(
+  boss: PgBoss,
+): Promise<readonly string[]> {
+  await createAndSchedule(boss, allQueues, schedules);
+
+  await boss.work<GeneralStatusPayload>(
+    GENERAL_STATUS_QUEUE,
+    { localConcurrency: 1 },
+    handleGeneralStatusGenerate,
+  );
+  await boss.work<BloodPressureStatusPayload>(
+    BLOOD_PRESSURE_STATUS_QUEUE,
+    { localConcurrency: 1 },
+    handleBloodPressureStatusGenerate,
+  );
+  await boss.work<WeightStatusPayload>(
+    WEIGHT_STATUS_QUEUE,
+    { localConcurrency: 1 },
+    handleWeightStatusGenerate,
+  );
+  await boss.work<PulseStatusPayload>(
+    PULSE_STATUS_QUEUE,
+    { localConcurrency: 1 },
+    handlePulseStatusGenerate,
+  );
+  await boss.work<BmiStatusPayload>(
+    BMI_STATUS_QUEUE,
+    { localConcurrency: 1 },
+    handleBmiStatusGenerate,
+  );
+  await boss.work<MoodStatusPayload>(
+    MOOD_STATUS_QUEUE,
+    { localConcurrency: 1 },
+    handleMoodStatusGenerate,
+  );
+  await boss.work<MedicationComplianceStatusPayload>(
+    MEDICATION_COMPLIANCE_STATUS_QUEUE,
+    { localConcurrency: 1 },
+    handleMedicationComplianceStatusGenerate,
+  );
+  // v1.7.0 — nightly comprehensive-insight pre-generation.
+  //
+  // v1.16.8 — localConcurrency 2 (was 1). The queue carries BOTH the
+  // scheduled 04:30 cohort walk AND the visit-triggered per-user force
+  // warms; with a single slot a force warm enqueued during the nightly
+  // batch sat behind the entire cohort and the visiting user stared at
+  // a cold dashboard for the duration. Two slots let one force warm run
+  // alongside the batch while still bounding provider-level concurrency
+  // (the cohort walk is itself sequential per user, and the content-hash
+  // gate + per-user budget gate inside the handler cover the rare
+  // double-tick overlap the old single slot serialised away).
+  await boss.work<InsightPregeneratePayload>(
+    INSIGHT_PREGENERATE_QUEUE,
+    { localConcurrency: 2 },
+    handleInsightPregenerateJob,
+  );
+  // v1.8.3 — on-demand per-metric status generation enqueued by the
+  // read-only status route on a cold card. Low concurrency so a first
+  // visit that cold-misses several cards can't saturate the Prisma pool
+  // or fan out an unbounded number of concurrent provider calls.
+  await boss.work<InsightStatusGeneratePayload>(
+    INSIGHT_STATUS_GENERATE_QUEUE,
+    { localConcurrency: INSIGHT_STATUS_GENERATE_CONCURRENCY },
+    handleInsightStatusGenerate,
+  );
+  // v1.10.0 — computed scores (WX-C). Nightly Recovery-score compute +
+  // store. The cron tick carries an empty payload; the runner iterates every
+  // eligible user and upserts one `COMPUTED RECOVERY_SCORE` row per scored
+  // day (idempotent — a re-fire overwrites in place). Single-flight so two
+  // ticks never double-walk the cohort.
+  await boss.work(RECOVERY_SCORE_QUEUE, { localConcurrency: 1 }, async () => {
+    try {
+      const summary = await runRecoveryScore(getWorkerPrisma());
+      workerLog(
+        "info",
+        `[recovery-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
+      );
+    } catch (err) {
+      recordError();
+      workerLog("error", "[recovery-score] pass failed", err);
+      throw err;
+    }
+  });
+  // v1.15.20 — proactive Coach nudge. Single-flight; the push-attempts
+  // ledger caps a user at one nudge per rolling week, so an overlapping
+  // tick would only waste reads. Deterministic triggers, no AI call.
+  await boss.work(COACH_NUDGE_QUEUE, { localConcurrency: 1 }, async () => {
+    await withBackgroundEvent("job.coach_nudge", async (evt) => {
+      try {
+        const summary = await runCoachNudgeTick(getWorkerPrisma(), new Date());
+        evt.setBackground({
+          task_name: "job.coach_nudge",
+          result: {
+            candidates_scanned: summary.candidatesScanned,
+            dispatched: summary.dispatched,
+            skipped_opted_out: summary.skippedOptedOut,
+            skipped_no_provider: summary.skippedNoProvider,
+            skipped_recent_nudge: summary.skippedRecentNudge,
+            skipped_no_trigger: summary.skippedNoTrigger,
+            skipped_no_channel: summary.skippedNoChannel,
+            failed: summary.failed,
+          },
+        });
+      } catch (err) {
+        evt.setError(err);
+        recordError();
+        throw err;
+      }
+    });
+  });
+  // v1.16.11 — medication low-stock pass. Single-flight; the
+  // per-medication stamp makes a re-fire idempotent (already-notified
+  // crossings skip), so an overlapping tick would only waste reads.
+  await boss.work(
+    MEDICATION_LOW_STOCK_QUEUE,
+    { localConcurrency: 1 },
+    async () => {
+      await withBackgroundEvent("job.medication_low_stock", async (evt) => {
+        try {
+          const summary = await runMedicationLowStockTick(
+            getWorkerPrisma(),
+            new Date(),
+          );
+          evt.setBackground({
+            task_name: "job.medication_low_stock",
+            result: {
+              users_scanned: summary.usersScanned,
+              skipped_threshold_off: summary.skippedThresholdOff,
+              medications_evaluated: summary.medicationsEvaluated,
+              notified: summary.notified,
+              rearmed: summary.rearmed,
+              skipped_already_notified: summary.skippedAlreadyNotified,
+              skipped_above_threshold: summary.skippedAboveThreshold,
+              skipped_no_runway: summary.skippedNoRunway,
+              skipped_no_channel: summary.skippedNoChannel,
+              failed: summary.failed,
+            },
+          });
+        } catch (err) {
+          evt.setError(err);
+          recordError();
+          throw err;
+        }
+      });
+    },
+  );
+  // v1.10.0 — computed scores (WX-E). Nightly Stress-score (HRV-derived
+  // proxy) compute + store. Single-flight so two ticks never double-walk
+  // the cohort. The runner iterates every eligible user and upserts one
+  // `COMPUTED STRESS_SCORE` row per scored day (idempotent — a re-fire
+  // overwrites in place).
+  await boss.work(STRESS_SCORE_QUEUE, { localConcurrency: 1 }, async () => {
+    try {
+      const summary = await runStressScore(getWorkerPrisma());
+      workerLog(
+        "info",
+        `[stress-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
+      );
+    } catch (err) {
+      recordError();
+      workerLog("error", "[stress-score] pass failed", err);
+      throw err;
+    }
+  });
+  // v1.10.0 — computed scores (WX-E). Nightly Strain-score (Banister TRIMP
+  // cardio-load) compute + store. Single-flight; upserts one `COMPUTED
+  // STRAIN_SCORE` row per scored day (idempotent).
+  await boss.work(STRAIN_SCORE_QUEUE, { localConcurrency: 1 }, async () => {
+    try {
+      const summary = await runStrainScore(getWorkerPrisma());
+      workerLog(
+        "info",
+        `[strain-score] considered=${summary.considered} stored=${summary.stored} insufficient=${summary.insufficient} errored=${summary.errored}`,
+      );
+    } catch (err) {
+      recordError();
+      workerLog("error", "[strain-score] pass failed", err);
+      throw err;
+    }
+  });
+  // v1.11.0 — period-narrative warm. A scheduled tick (no `userId`) runs the
+  // boundary-gated nightly fan-out; a `userId` payload runs a single-user warm
+  // enqueued by the read-only GET on a cold/stale read. Single-flight so two
+  // ticks never double-walk the cohort; the per-user budget gate covers the
+  // fan-out and the enqueue `singletonKey` covers the single-user path.
+  await boss.work<PeriodNarrativePayload>(
+    PERIOD_NARRATIVE_QUEUE,
+    { localConcurrency: 1 },
+    async (jobs) => {
+      for (const job of jobs) {
+        try {
+          if (job.data?.userId) {
+            await warmOneNarrative(job.data);
+          } else {
+            const summary = await runPeriodNarrativeWarm(getWorkerPrisma());
+            workerLog(
+              "info",
+              `[period-narrative] periods=${summary.periods.join(",") || "none"} total=${summary.total} generated=${summary.generated} cached=${summary.cached} skipped=${summary.skipped} insufficient=${summary.insufficient} failed=${summary.failed} budget=${summary.budgetBlocked}`,
+            );
+          }
+        } catch (err) {
+          recordError();
+          await reportWorkerError(PERIOD_NARRATIVE_QUEUE, err, {
+            mode: job.data?.userId ? "single-user" : "scheduled",
+          });
+          throw err;
+        }
+      }
+    },
+  );
+  // v1.11.1 — combined Coach memory refresh: rolling conversation summary +
+  // durable fact extraction for one long conversation. localConcurrency 1 so a
+  // burst of long-conversation turns can't fan out concurrent provider calls;
+  // each step is budget-gated inside runStatusCompletion and fault-isolated.
+  await boss.work<CoachMemoryRefreshPayload>(
+    COACH_MEMORY_REFRESH_QUEUE,
+    { localConcurrency: 1 },
+    async (jobs) => {
+      for (const job of jobs) {
+        if (!job.data?.conversationId || !job.data?.userId) continue;
+        try {
+          await runCoachMemoryRefresh(job.data);
+        } catch (err) {
+          recordError();
+          workerLog("error", "[coach-memory-refresh] failed", err);
+          throw err;
+        }
+      }
+    },
+  );
+
+  return allQueues;
+}

src/lib/jobs/reminder/registrar-shared.ts

@@ -0,0 +1,65 @@
+/**
+ * Shared plumbing for the domain-grouped queue registrars.
+ *
+ * v1.18.1 — `startReminderWorker()` was a 2143-LOC monolith that declared
+ * every pg-boss queue name, listed them in one `allQueues` array, scheduled
+ * every cron, and bound every `boss.work` handler inline. It is now decomposed
+ * into domain registrars (integration-sync, status, rollup, reminders,
+ * maintenance) under `src/lib/jobs/reminder/`. Each registrar owns the four
+ * facts the v1.4.37 dead-queue guards pin — the queue-name constant, its
+ * `allQueues` membership, its `[QUEUE, CRON]` schedule tuple, and its
+ * `boss.work(QUEUE, …, handler)` binding — so a queue can never be declared
+ * without being provisioned, scheduled, and drained. The boot file composes
+ * the registrars; the guards follow the wiring into each registrar module.
+ *
+ * Every registrar returns the queue names it created so the boot file can
+ * assert a single aggregate `allQueues` (defence in depth: the per-registrar
+ * arrays AND the boot-level union both have to agree).
+ */
+import { PgBoss } from "pg-boss";
+
+/**
+ * Cron schedule tuple: `[queueName, cronExpression, sendOptions?]`. The
+ * optional third element carries per-queue send options (e.g. the LLM-bound
+ * insight retry policy) merged into the `boss.schedule` call.
+ */
+export type ScheduleEntry = [string, string, Record<string, unknown>?];
+
+/**
+ * Shared retry policy for the LLM-bound insight queues. A transient failure
+ * (provider hiccup, pool exhaustion) used to fail the nightly tick silently
+ * until the NEXT night; three backed-off retries match the backfill queues'
+ * established shape (see e.g. whoop-backfill.ts).
+ */
+export const insightRetryOptions = {
+  retryLimit: 3,
+  retryDelay: 60,
+  retryBackoff: true,
+} as const;
+
+/**
+ * Create every queue in `queues`, then schedule every cron in `schedules`.
+ * Centralised so each registrar provisions before it schedules in the exact
+ * order the monolith did, and the `Europe/Berlin` tz default stays in one
+ * place.
+ */
+export async function createAndSchedule(
+  boss: PgBoss,
+  queues: readonly string[],
+  schedules: readonly ScheduleEntry[],
+): Promise<void> {
+  for (const q of queues) {
+    await boss.createQueue(q);
+  }
+  for (const [name, cron, sendOptions] of schedules) {
+    await boss.schedule(
+      name,
+      cron,
+      {},
+      {
+        tz: "Europe/Berlin",
+        ...(sendOptions ?? {}),
+      },
+    );
+  }
+}