perf(rollups): refold the buckets of every provider sync in one pass

Every provider sync and the shared mutation tail refolded its touched
(type, day) buckets one at a time: an aggregate, a transaction and three
queue sends per day. They now hand the whole set to
recomputeBucketsForKeys, which leaves the DAY tier exactly as the per-key
hook did (pinned by a parity test against it) at the cost of one
aggregate per chunk of keys and one coarse job per run of buckets.

src/__tests__/measurement-write-tail-guard.test.ts

@@ -176,7 +176,9 @@ describe("measurement write tail", () => {
     const missing = handRolled.filter((file) => {
       const src = readFileSync(join(SRC, file), "utf8");
       return !(
-        /recomputeBucketsForMeasurement|afterMeasurementMutation/.test(src) &&
+        /recomputeBucketsFor(Measurement|Keys)|afterMeasurementMutation/.test(
+          src,
+        ) &&
         /invalidateStatusInsightsForTypes|afterMeasurementMutation/.test(src)
       );
     });

src/app/api/import/__tests__/route.test.ts

@@ -54,9 +54,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/rollups/measurement-rollups")
   >("@/lib/rollups/measurement-rollups");
+  const perKey = vi.fn().mockResolvedValue(undefined);
   return {
     ...actual,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForMeasurement: perKey,
+    // The keyed refold, observed as the per-key recomputes it stands for.
+    recomputeBucketsForKeys: vi.fn(
+      async (
+        userId: string,
+        keys: Array<{ type: string; measuredAt: Date }>,
+      ) => {
+        for (const k of keys) await perKey(userId, k.type, k.measuredAt);
+      },
+    ),
   };
 });
 

src/app/api/import/csv/__tests__/route.test.ts

@@ -58,9 +58,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/rollups/measurement-rollups")
   >("@/lib/rollups/measurement-rollups");
+  const perKey = vi.fn().mockResolvedValue(undefined);
   return {
     ...actual,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForMeasurement: perKey,
+    // The keyed refold, observed as the per-key recomputes it stands for.
+    recomputeBucketsForKeys: vi.fn(
+      async (
+        userId: string,
+        keys: Array<{ type: string; measuredAt: Date }>,
+      ) => {
+        for (const k of keys) await perKey(userId, k.type, k.measuredAt);
+      },
+    ),
   };
 });
 

src/app/api/measurements/bulk-delete/__tests__/route.test.ts

@@ -47,10 +47,20 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/rollups/measurement-rollups")
   >("@/lib/rollups/measurement-rollups");
+  const perKey = vi.fn().mockResolvedValue(undefined);
   return {
     // keep the real collapse so the test exercises the actual fan-out logic
     collapseToTypeDayKeys: actual.collapseToTypeDayKeys,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForMeasurement: perKey,
+    // The keyed refold, observed as the per-key recomputes it stands for.
+    recomputeBucketsForKeys: vi.fn(
+      async (
+        userId: string,
+        keys: Array<{ type: string; measuredAt: Date }>,
+      ) => {
+        for (const k of keys) await perKey(userId, k.type, k.measuredAt);
+      },
+    ),
   };
 });
 vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));

src/app/api/measurements/restore/__tests__/route.test.ts

@@ -47,10 +47,20 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/rollups/measurement-rollups")
   >("@/lib/rollups/measurement-rollups");
+  const perKey = vi.fn().mockResolvedValue(undefined);
   return {
     // keep the real collapse so the test exercises the actual fan-out logic
     collapseToTypeDayKeys: actual.collapseToTypeDayKeys,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForMeasurement: perKey,
+    // The keyed refold, observed as the per-key recomputes it stands for.
+    recomputeBucketsForKeys: vi.fn(
+      async (
+        userId: string,
+        keys: Array<{ type: string; measuredAt: Date }>,
+      ) => {
+        for (const k of keys) await perKey(userId, k.type, k.measuredAt);
+      },
+    ),
   };
 });
 vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() }));

src/lib/fitbit/sync-core.ts

@@ -18,7 +18,7 @@ import { getEvent } from "@/lib/logging/context";
 import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { refreshAccessToken } from "./client";
 import { getUserFitbitCredentials } from "./credentials";
@@ -894,9 +894,7 @@ export async function upsertFitbitMeasurements(
 
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
     invalidateStatusInsightsForTypes(
       userId,
       keys.map((k) => k.type),

src/lib/google-health/sync-core.ts

@@ -41,7 +41,7 @@ import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit
 import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import {
@@ -1587,9 +1587,7 @@ export async function upsertGoogleHealthMeasurements(
 
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
     invalidateStatusInsightsForTypes(
       userId,
       keys.map((k) => k.type),

src/lib/nightscout/__tests__/sync.test.ts

@@ -43,7 +43,12 @@ vi.mock("@/lib/integrations/status", async (importOriginal) => ({
 vi.mock("@/lib/rollups/measurement-rollups", () => ({
   collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) =>
     rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })),
-  recomputeBucketsForMeasurement: recomputeMock,
+  recomputeBucketsForKeys: async (
+    userId: string,
+    keys: Array<{ type: string; measuredAt: Date }>,
+  ) => {
+    for (const k of keys) await recomputeMock(userId, k.type, k.measuredAt);
+  },
 }));
 
 vi.mock("@/lib/insights/comprehensive-generate", () => ({

src/lib/nightscout/sync.ts

@@ -34,7 +34,7 @@ import type { IntegrationClassification } from "@/lib/integrations/http-status-c
 import type { SyncWriteResult } from "@/lib/outcome/written-outcome";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import {
   emitInsertedMeasurementArrivals,
@@ -327,9 +327,7 @@ export async function upsertNightscoutEntries(
 
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
     invalidateStatusInsightsForTypes(
       userId,
       keys.map((k) => k.type),

src/lib/oura/__tests__/sync.test.ts

@@ -98,7 +98,12 @@ vi.mock("@/lib/integrations/status", async (importOriginal) => ({
 vi.mock("@/lib/rollups/measurement-rollups", () => ({
   collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) =>
     rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })),
-  recomputeBucketsForMeasurement: recomputeMock,
+  recomputeBucketsForKeys: async (
+    userId: string,
+    keys: Array<{ type: string; measuredAt: Date }>,
+  ) => {
+    for (const k of keys) await recomputeMock(userId, k.type, k.measuredAt);
+  },
 }));
 
 vi.mock("@/lib/insights/comprehensive-generate", () => ({

src/lib/oura/sync.ts

@@ -45,7 +45,7 @@ import {
 } from "@/lib/integrations/status";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import {
   emitInsertedMeasurementArrivals,
@@ -805,9 +805,7 @@ export async function upsertOuraMeasurements(
   );
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
     invalidateStatusInsightsForTypes(
       userId,
       keys.map((k) => k.type),

src/lib/polar/sync.ts

@@ -48,7 +48,7 @@ import {
 } from "@/lib/integrations/status";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
@@ -398,9 +398,7 @@ export async function upsertPolarMeasurements(
   );
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
     invalidateStatusInsightsForTypes(
       userId,
       keys.map((k) => k.type),

src/lib/rollups/__tests__/after-measurement-mutation.test.ts

@@ -4,7 +4,7 @@
  * invalidation).
  *
  * Watched red: with the recompute loop removed from the helper the
- * first test fails naming the missing `recomputeBucketsForMeasurement`
+ * first test fails naming the missing `recomputeBucketsForKeys`
  * call; with the `.catch` isolation removed the failure-isolation test
  * throws. Both verified red against a deliberately broken helper before
  * the real implementation landed.
@@ -27,7 +27,12 @@ vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => {
     await importOriginal<typeof import("@/lib/rollups/measurement-rollups")>();
   return {
     ...actual,
-    recomputeBucketsForMeasurement: mocks.recompute,
+    recomputeBucketsForKeys: async (
+      userId: string,
+      keys: Array<{ type: string; measuredAt: Date }>,
+    ) => {
+      for (const k of keys) await mocks.recompute(userId, k.type, k.measuredAt);
+    },
   };
 });
 

src/lib/rollups/after-measurement-mutation.ts

@@ -8,7 +8,7 @@
  *
  *   1. the `(type, day)` rollup buckets the mutation dirtied — recomputed
  *      inline (DAY) + enqueued (WEEK/MONTH/YEAR) via
- *      `recomputeBucketsForMeasurement`. The recompute handles deletions
+ *      `recomputeBucketsForKeys`. The recompute handles deletions
  *      inherently: a day left empty drops its rollup row.
  *   2. the cached per-metric status assessments — re-warmed via
  *      `invalidateStatusInsightsForTypes` (fire-and-forget).
@@ -20,7 +20,7 @@
  */
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import {
   invalidateStatusInsightsForTypes,
@@ -50,9 +50,7 @@ export async function afterMeasurementMutation(
   // 500-row batch fires at most one DAY recompute per touched bucket.
   const keys = collapseToTypeDayKeys(rows);
   try {
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
   } catch (err) {
     console.warn(`[${label}] rollup recompute failed`, err);
   }

src/lib/whoop/__tests__/sync.test.ts

@@ -90,8 +90,13 @@ vi.mock("@/lib/integrations/status", () => ({
 vi.mock("@/lib/rollups/measurement-rollups", () => ({
   collapseToTypeDayKeys: (rows: Array<{ type: string; measuredAt: Date }>) =>
     rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })),
-  recomputeBucketsForMeasurement: (...a: unknown[]) =>
-    recomputeBucketsMock(...a),
+  recomputeBucketsForKeys: async (
+    userId: string,
+    keys: Array<{ type: string; measuredAt: Date }>,
+  ) => {
+    for (const k of keys)
+      await recomputeBucketsMock(userId, k.type, k.measuredAt);
+  },
 }));
 
 vi.mock("@/lib/insights/comprehensive-generate", () => ({

src/lib/whoop/sync-core.ts

@@ -14,7 +14,7 @@
  *     `(userId, type, source = WHOOP, externalId)` so a re-post (WHOOP
  *     re-scores recovery/sleep after the fact) overwrites in place rather than
  *     minting a duplicate. After the upserts the rollup tier is re-folded
- *     (`recomputeBucketsForMeasurement`) and the status-insight caches are
+ *     (`recomputeBucketsForKeys`) and the status-insight caches are
  *     invalidated, identical to the Withings tail.
  *
  * The incremental window starts from `lastSyncedAt - overlap`. WHOOP re-scores
@@ -37,7 +37,7 @@ import {
 } from "@/lib/integrations/status";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import {
@@ -503,9 +503,7 @@ export async function upsertWhoopMeasurements(
     );
     try {
       const keys = collapseToTypeDayKeys(touched);
-      for (const k of keys) {
-        await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-      }
+      await recomputeBucketsForKeys(userId, keys);
     } catch (err) {
       getEvent()?.addWarning(
         `whoop: rollup recompute failed for ${userId}: ${err}`,

src/lib/withings/__tests__/sync-activity.test.ts

@@ -69,9 +69,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/rollups/measurement-rollups")
   >("@/lib/rollups/measurement-rollups");
+  const perKey = vi.fn().mockResolvedValue(undefined);
   return {
     ...actual,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForMeasurement: perKey,
+    // The keyed refold, observed as the per-key recomputes it stands for.
+    recomputeBucketsForKeys: vi.fn(
+      async (
+        userId: string,
+        keys: Array<{ type: string; measuredAt: Date }>,
+      ) => {
+        for (const k of keys) await perKey(userId, k.type, k.measuredAt);
+      },
+    ),
   };
 });
 

src/lib/withings/__tests__/sync-ecg.test.ts

@@ -68,9 +68,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/rollups/measurement-rollups")
   >("@/lib/rollups/measurement-rollups");
+  const perKey = vi.fn().mockResolvedValue(undefined);
   return {
     ...actual,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForMeasurement: perKey,
+    // The keyed refold, observed as the per-key recomputes it stands for.
+    recomputeBucketsForKeys: vi.fn(
+      async (
+        userId: string,
+        keys: Array<{ type: string; measuredAt: Date }>,
+      ) => {
+        for (const k of keys) await perKey(userId, k.type, k.measuredAt);
+      },
+    ),
   };
 });
 

src/lib/withings/__tests__/sync-sleep.test.ts

@@ -67,9 +67,19 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => {
   const actual = await vi.importActual<
     typeof import("@/lib/rollups/measurement-rollups")
   >("@/lib/rollups/measurement-rollups");
+  const perKey = vi.fn().mockResolvedValue(undefined);
   return {
     ...actual,
-    recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined),
+    recomputeBucketsForMeasurement: perKey,
+    // The keyed refold, observed as the per-key recomputes it stands for.
+    recomputeBucketsForKeys: vi.fn(
+      async (
+        userId: string,
+        keys: Array<{ type: string; measuredAt: Date }>,
+      ) => {
+        for (const k of keys) await perKey(userId, k.type, k.measuredAt);
+      },
+    ),
   };
 });
 

src/lib/withings/sync-activity.ts

@@ -45,7 +45,7 @@ import { getUnitForType } from "@/lib/validations/measurement";
 import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 
@@ -540,9 +540,7 @@ export async function syncUserActivity(
   // catch-up costs at most ~N (type, day) recomputes.
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
 
     // v1.8.0 — drop the per-metric assessment caches the synced types
     // dirty (steps feed the general overview). Fire-and-forget.

src/lib/withings/sync-ecg.ts

@@ -56,7 +56,7 @@ import { safeFetch } from "@/lib/safe-fetch";
 import { persistEcgRecording } from "@/lib/ecg/persist-recording";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 
@@ -519,9 +519,7 @@ export async function syncUserEcg(
   // then drop the per-metric assessment caches the new events dirty.
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
     invalidateStatusInsightsForTypes(
       userId,
       keys.map((k) => k.type),

src/lib/withings/sync-sleep.ts

@@ -45,7 +45,7 @@ import { safeFetch } from "@/lib/safe-fetch";
 import { getUnitForType } from "@/lib/validations/measurement";
 import {
   collapseToTypeDayKeys,
-  recomputeBucketsForMeasurement,
+  recomputeBucketsForKeys,
 } from "@/lib/rollups/measurement-rollups";
 import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
@@ -628,9 +628,7 @@ export async function syncUserSleep(
   // ~10 stage rows per night land in one DAY recompute.
   try {
     const keys = collapseToTypeDayKeys(touched);
-    for (const k of keys) {
-      await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt);
-    }
+    await recomputeBucketsForKeys(userId, keys);
 
     // v1.8.0 — drop the per-metric assessment caches the synced types
     // dirty (sleep + resting-HR feed the pulse / general cards).