From a1f61bb29b7787586a1fef3f4efbbd9a6078ab5f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= <mbombeck@gmail.com>
Date: Sun, 17 May 2026 14:29:23 +0200
Subject: [PATCH] chore(queue): schedule nightly drainPerSampleCumulative with
 36h cutoff

Registers a new drain-per-sample-cumulative pg-boss queue + cron
(03:45 Europe/Berlin, slotting between the audit-log cleanup at
03:15 and the feedback aggregator at 04:00). The handler invokes
drainPerSampleCumulative across every user with cutoffHours=36 so
today + the trailing watch-sync window stay per-sample while
completed-and-stable days collapse into one stats: row per
calendar day per cumulative HK type.

Replaces the manual CLI + admin-endpoint pattern from v1.4.30 for
the recurring case (the operator-driven paths remain for one-shot
backfills) and avoids the prod-image tsx + Prisma symlink hack
documented in feedback_prod_image_tsx_prisma_gap.md.
---
 src/lib/jobs/reminder-worker.ts | 49 +++++++++++++++++++++++++++++++++
 1 file changed, 49 insertions(+)

diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts
index 0d0b76fd7..9ba0661ad 100644
--- a/src/lib/jobs/reminder-worker.ts
+++ b/src/lib/jobs/reminder-worker.ts
@@ -62,6 +62,7 @@ import {
   type RollupFullBackfillPayload,
   type RollupRecomputePayload,
 } from "@/lib/measurements/rollups";
+import { drainPerSampleCumulative } from "@/lib/measurements/drain-per-sample-cumulative";
 import { expireStaleInUseItems } from "@/lib/medications/inventory/service";
 import { rotateLegacyMoodLogSecrets } from "@/lib/moodlog-secret";
 import { deleteMessage } from "@/lib/telegram";
@@ -157,6 +158,20 @@ const HOST_METRIC_CRON = "* * * * *";
 // 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 * * *";
+// 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 * * *";
+const DRAIN_CUMULATIVE_CUTOFF_HOURS = 36;
+interface DrainCumulativePayload {
+  triggeredAt: string;
+}
 
 interface ReminderCheckPayload {
   triggeredAt: string;
@@ -1522,6 +1537,10 @@ export async function startReminderWorker() {
     // 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.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],
   ];
 
   for (const [name, cron] of schedules) {
@@ -1691,6 +1710,36 @@ export async function startReminderWorker() {
     },
   );
 
+  // 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 Marc'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();
+          workerLog("error", "[drain-cumulative] run failed", err);
+        }
+      }
+    },
+  );
+
   // v1.4.35.1 — fire-and-forget boot-time backfill discovery. Finds
   // every user with measurements but no rollup coverage and enqueues
   // a full-fold per account. Idempotent across reboots: the discovery