fix(cache): mark measurement caches stale after every bulk write

Closed analytics blocks survive a write that says it reached no further
back, so a writer that never invalidated left them on their pre-write
values. The Apple Health and Health Connect imports, the CSV and JSON
imports, the score engines, the WHOOP webhook, the Withings activity, sleep
and ECG legs, the Oura history backfill and the sleep-segment sweep wrote
measurements without touching the read caches; a long-history import
showed the old all-time figures until the entries aged out.

The shared post-mutation tail now marks the caches stale back to the
oldest row it is handed, and the Withings and Oura legs that hand-rolled
the same tail run through it. The two export imports share one tail that
refolds the rollups, marks the caches stale back to the oldest sample
written (all of them when none was) and re-warms the status assessments.
A sweep that tombstones a sleep segment drops every cached block.

src/__tests__/baselines/source-file-lines.json

@@ -1,7 +1,7 @@
 {
   "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
   "count": 123,
-  "total": 145651,
+  "total": 145555,
   "entries": {
     "app/api/dashboard/summary/route.ts": 1032,
     "app/api/documents/inbound/route.ts": 811,
@@ -91,7 +91,7 @@
     "lib/insights/mood-status.ts": 814,
     "lib/insights/status-cache.ts": 812,
     "lib/integrations/status.ts": 1445,
-    "lib/jobs/apple-health-import-worker.ts": 863,
+    "lib/jobs/apple-health-import-worker.ts": 801,
     "lib/jobs/backup-restore.ts": 888,
     "lib/jobs/coach-nudge.ts": 1089,
     "lib/jobs/insight-pregenerate.ts": 1266,
@@ -114,7 +114,7 @@
     "lib/medications/scheduling/slot-upsert.ts": 926,
     "lib/notifications/senders/apns.ts": 1100,
     "lib/oura/client.ts": 1360,
-    "lib/oura/sync.ts": 917,
+    "lib/oura/sync.ts": 902,
     "lib/polar/client.ts": 1010,
     "lib/rollups/measurement-rollups.ts": 1496,
     "lib/search/search-records.ts": 996,
@@ -123,7 +123,7 @@
     "lib/telegram-webhook-handlers.ts": 1341,
     "lib/validations/inbound-documents.ts": 1130,
     "lib/whoop/client.ts": 917,
-    "lib/withings/sync-sleep.ts": 903,
+    "lib/withings/sync-sleep.ts": 884,
     "lib/withings/sync.ts": 978,
     "proxy.ts": 870
   }

src/__tests__/baselines/source-function-lines.json

@@ -1,7 +1,7 @@
 {
   "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
   "count": 411,
-  "total": 148324,
+  "total": 148227,
   "entries": {
     "app/achievements/page.tsx :: AchievementsPage": 206,
     "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314,
@@ -350,7 +350,7 @@
     "lib/insights/weight-status.ts :: prepareWeightStatusForUser": 598,
     "lib/integrations/connection-health-load.ts :: loadConnectionFacts": 227,
     "lib/integrations/status.ts :: recordSyncFailure": 292,
-    "lib/jobs/apple-health-import-worker.ts :: handleAppleHealthImport": 357,
+    "lib/jobs/apple-health-import-worker.ts :: handleAppleHealthImport": 300,
     "lib/jobs/coach-nudge.ts :: findTriggerForUser": 279,
     "lib/jobs/coach-nudge.ts :: runCoachNudgeTick": 292,
     "lib/jobs/coach-reminder-sweep.ts :: runCoachReminderSweep": 222,
@@ -409,9 +409,9 @@
     "lib/targets/vitals-builder.ts :: buildVitalTargets": 324,
     "lib/telegram-webhook-handlers.ts :: handleCallback": 599,
     "lib/telegram-webhook-handlers.ts :: handleTextMessage": 337,
-    "lib/withings/sync-activity.ts :: syncUserActivity": 391,
-    "lib/withings/sync-ecg.ts :: syncUserEcg": 213,
-    "lib/withings/sync-sleep.ts :: syncUserSleep": 429,
+    "lib/withings/sync-activity.ts :: syncUserActivity": 377,
+    "lib/withings/sync-ecg.ts :: syncUserEcg": 202,
+    "lib/withings/sync-sleep.ts :: syncUserSleep": 414,
     "lib/withings/sync.ts :: syncUserMeasurements": 373,
     "proxy.ts :: proxy": 499
   }

src/__tests__/baselines/version-stamp-comments.json

@@ -1,7 +1,7 @@
 {
   "$comment": "Comment lines that open with a release version, per file under src/. Maintained by src/__tests__/version-stamp-ratchet.test.ts.",
   "count": 2959,
-  "total": 10520,
+  "total": 10517,
   "entries": {
     "__tests__/account-payload-consumer-guard.test.ts": 2,
     "__tests__/acting-account-boundary-guard.test.ts": 7,
@@ -2346,7 +2346,7 @@
     "lib/jobs/__tests__/whoop-resource-sync.test.ts": 1,
     "lib/jobs/__tests__/withings-queues.test.ts": 2,
     "lib/jobs/__tests__/workout-insight-queue.test.ts": 1,
-    "lib/jobs/apple-health-import-worker.ts": 5,
+    "lib/jobs/apple-health-import-worker.ts": 4,
     "lib/jobs/coach-message-cleanup.ts": 1,
     "lib/jobs/coach-nudge-ai.ts": 3,
     "lib/jobs/coach-nudge.ts": 17,
@@ -2953,9 +2953,9 @@
     "lib/withings/client.ts": 5,
     "lib/withings/oauth-state.ts": 1,
     "lib/withings/response-classifier.ts": 1,
-    "lib/withings/sync-activity.ts": 8,
+    "lib/withings/sync-activity.ts": 7,
     "lib/withings/sync-ecg.ts": 4,
-    "lib/withings/sync-sleep.ts": 10,
+    "lib/withings/sync-sleep.ts": 9,
     "lib/withings/sync.ts": 9,
     "lib/withings/webhook-handler.ts": 3,
     "lib/workouts/list-read.ts": 1,

src/lib/cache/closed-block.ts

@@ -21,3 +21,20 @@ export function closedBlockUntouched(key: string, touchedFrom: Date): boolean {
   const match = CLOSED_BLOCK_RE.exec(key);
   return match !== null && Number(match[1]) <= touchedFrom.getTime();
 }
+
+/**
+ * The earliest `measuredAt` among rows a write touched: the `touchedFrom` a
+ * writer that knows every row it wrote hands to `invalidateUserMeasurements`.
+ * `undefined` for no rows, which makes the invalidation drop closed blocks too.
+ */
+export function earliestMeasuredAt(
+  rows: Iterable<{ measuredAt: Date }>,
+): Date | undefined {
+  let earliest: Date | undefined;
+  for (const row of rows) {
+    if (earliest === undefined || row.measuredAt < earliest) {
+      earliest = row.measuredAt;
+    }
+  }
+  return earliest;
+}

src/lib/jobs/apple-health-import-worker.ts

@@ -26,7 +26,7 @@
  */
 import { unlinkSync } from "node:fs";
 import { getHeapStatistics } from "node:v8";
-import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
+import type { PrismaClient } from "@/generated/prisma/client";
 import { prisma, toJson } from "@/lib/db";
 import type { Job } from "pg-boss";
 
@@ -40,18 +40,13 @@ import {
   type WorkoutRouteReference,
 } from "@/lib/apple-health/route-import";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
-import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation";
-import { trackBackgroundTask } from "@/lib/logging/background-tasks";
+import { afterBulkImport } from "@/lib/jobs/import-tail";
 import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
 import {
   streamParseExportXml,
   type ImportJobProgress,
   type ImportJobResult,
 } from "@/lib/measurements/import-apple-health-export";
-import {
-  recomputeUserRollups,
-  ROLLUP_FOLD_WINDOW_MS,
-} from "@/lib/rollups/measurement-rollups";
 import { withBackgroundEvent } from "@/lib/logging/background";
 import { DEFAULT_TIMEZONE } from "@/lib/tz/format";
 import {
@@ -589,70 +584,13 @@ export async function handleAppleHealthImport(
       },
     });
 
-    // v1.5.0 — fold the persistent rollup table for the imported
-    // user. The per-row write hooks are intentionally skipped on the
-    // streaming-ingest path (a 100k-row import would otherwise pay
-    // 100k DAY-recompute round-trips); we run the rollup once at
-    // the end, scoped to the user's full measurement span, so
-    // post-import reads of the analytics + comprehensive surfaces
-    // hit the warm rollup table on first paint.
-    //
-    // The miss rides out as a fact rather than as a failed job: the
-    // import is committed and its mirror row already reads `done`, so
-    // failing here would report a completed import as failed and
-    // contradict the row the status poll surfaces to the operator.
-    let rollupFailed = false;
-    try {
-      // Folded over what this import wrote, not over the account's whole
-      // history, and never further back than the fold window every other
-      // path keeps to: buckets older than `ROLLUP_FOLD_WINDOW_MS` are by
-      // design never written, so reads there fall back to live SQL. The old
-      // span ran from the account's first measurement, whatever the import
-      // carried, and folded any history older than the window as well.
-      const span = result.measuredSpan;
-      if (span) {
-        const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS);
-        const spanFrom = new Date(span.from);
-        const from = spanFrom > windowStart ? spanFrom : windowStart;
-        // `to` is exclusive in the aggregator, so one millisecond past the
-        // newest row.
-        const to = new Date(new Date(span.to).getTime() + 1);
-        if (to > from) {
-          await recomputeUserRollups(userId, { from, to });
-        }
-      }
-    } catch (rollupErr) {
-      // Rollup failure is non-fatal — the next read falls through to
-      // live aggregation. Log but don't poison the import.
-      rollupFailed = true;
-      console.warn(
-        `[apple-health-import] Rollup recompute failed for user ${userId}`,
-        rollupErr,
-      );
-    }
-
-    // The imported types' cached status assessments describe data that just
-    // changed; re-warm them, the second leg every other measurement write
-    // runs (`afterMeasurementMutation`). The rollups above are refolded over
-    // the import's span rather than per day, so the leg is called directly.
-    const importedTypes = Object.entries(result.perType)
-      .filter(([, stat]) => stat.inserted + stat.updated > 0)
-      .map(([type]) => type as MeasurementType);
-    if (importedTypes.length > 0) {
-      // Registered so the test suite can await the detached handle.
-      trackBackgroundTask(
-        invalidateStatusInsightsForTypes(userId, importedTypes, {
-          // An export file the person chose to import, narrated like any
-          // other import rather than held to the sync gap.
-          source: "manual",
-        }).catch((err) => {
-          console.warn(
-            `[apple-health-import] status-insight invalidate failed for user ${userId}`,
-            err,
-          );
-        }),
-      );
-    }
+    // Rollups, measurement caches and status assessments over what the
+    // import wrote (`afterBulkImport`).
+    const { rollupFailed } = await afterBulkImport(
+      userId,
+      result,
+      "apple-health-import",
+    );
 
     // Best-effort cleanup. A failed unlink is not fatal — `/tmp` is
     // periodically swept on the host.

src/lib/jobs/health-connect-import-worker.ts

@@ -19,7 +19,6 @@ import type { Job } from "pg-boss";
 
 import { caughtAs, logCaught } from "@/lib/logging/signal";
 
-import type { MeasurementType } from "@/generated/prisma/client";
 import { prisma, toJson } from "@/lib/db";
 import { extractHealthConnectDb } from "@/lib/import/unzip-export-xml";
 import {
@@ -28,15 +27,10 @@ import {
 } from "@/lib/import/health-connect/import";
 import { HealthConnectImportError } from "@/lib/import/health-connect/reader";
 import { connectedDirectIntegrations } from "@/lib/import/health-connect/skip-packages";
-import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation";
 import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
-import { trackBackgroundTask } from "@/lib/logging/background-tasks";
+import { afterBulkImport } from "@/lib/jobs/import-tail";
 import { annotate } from "@/lib/logging/context";
 import { withBackgroundEvent } from "@/lib/logging/background";
-import {
-  recomputeUserRollups,
-  ROLLUP_FOLD_WINDOW_MS,
-} from "@/lib/rollups/measurement-rollups";
 import { DEFAULT_TIMEZONE } from "@/lib/tz/format";
 
 export const HEALTH_CONNECT_IMPORT_QUEUE = "health-connect-import";
@@ -203,43 +197,13 @@ export async function runHealthConnectImport(
       },
     });
 
-    // The per-row tails are skipped on a bulk import; refold the rollups over
-    // what this import wrote, inside the window every other path keeps to,
-    // and re-warm the cached assessments of the types it touched.
-    let rollupFailed = false;
-    const span = result.measuredSpan;
-    if (span) {
-      try {
-        const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS);
-        const spanFrom = new Date(span.from);
-        const from = spanFrom > windowStart ? spanFrom : windowStart;
-        const to = new Date(new Date(span.to).getTime() + 1);
-        if (to > from) await recomputeUserRollups(userId, { from, to });
-      } catch (err) {
-        rollupFailed = true;
-        console.warn(
-          `[health-connect-import] Rollup recompute failed for user ${userId}`,
-          err,
-        );
-      }
-    }
-    const touched = Object.entries(result.perType)
-      .filter(([, stat]) => stat.inserted + stat.updated > 0)
-      .map(([type]) => type as MeasurementType);
-    if (touched.length > 0) {
-      trackBackgroundTask(
-        invalidateStatusInsightsForTypes(userId, touched, {
-          // An export file the person chose to import, like the Apple Health
-          // import: narrated like a hand entry, not held to the sync gap.
-          source: "manual",
-        }).catch((err) => {
-          console.warn(
-            `[health-connect-import] status-insight invalidate failed for user ${userId}`,
-            err,
-          );
-        }),
-      );
-    }
+    // Rollups, measurement caches and status assessments over what the
+    // import wrote, as the Apple Health import does (`afterBulkImport`).
+    const { rollupFailed } = await afterBulkImport(
+      userId,
+      result,
+      "health-connect-import",
+    );
 
     annotate({
       action: { name: "import.health-connect.done" },

src/lib/jobs/import-tail.ts

@@ -0,0 +1,82 @@
+/**
+ * The tail of a bulk export import (Apple Health, Health Connect), run once
+ * the import is committed. The per-row write hooks are skipped while the
+ * export streams in (a 100k-row import would otherwise pay 100k DAY
+ * recomputes), so the three legs every other measurement write runs through
+ * `afterMeasurementMutation` run here once, over what the import wrote:
+ *
+ *   1. the rollup table, refolded over the span of the rows written and never
+ *      further back than the fold window every other path keeps to (buckets
+ *      older than `ROLLUP_FOLD_WINDOW_MS` are by design never written, so
+ *      reads there fall back to live SQL);
+ *   2. the measurement read caches, marked stale back to the oldest sample
+ *      written. An export carries years of history, and a closed block that a
+ *      later sync leaves in place would otherwise keep serving the pre-import
+ *      all-time values. An import that wrote no measurement (only workouts or
+ *      ECGs) drops them all;
+ *   3. the cached status assessments of the types written, narrated like a
+ *      hand entry: an export is a file the person chose to import.
+ *
+ * A rollup miss rides out as a fact rather than a failed job: the import is
+ * committed and its row already reads `done`, so failing here would report a
+ * completed import as failed.
+ */
+import type { MeasurementType } from "@/generated/prisma/client";
+import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
+import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalidation";
+import { trackBackgroundTask } from "@/lib/logging/background-tasks";
+import {
+  recomputeUserRollups,
+  ROLLUP_FOLD_WINDOW_MS,
+} from "@/lib/rollups/measurement-rollups";
+
+export async function afterBulkImport(
+  userId: string,
+  result: {
+    measuredSpan: { from: string; to: string } | null;
+    perType: Record<string, { inserted: number; updated: number }>;
+  },
+  label: string,
+): Promise<{ rollupFailed: boolean }> {
+  const span = result.measuredSpan;
+  let rollupFailed = false;
+  if (span) {
+    try {
+      const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS);
+      const spanFrom = new Date(span.from);
+      const from = spanFrom > windowStart ? spanFrom : windowStart;
+      // `to` is exclusive in the aggregator: one millisecond past the newest.
+      const to = new Date(new Date(span.to).getTime() + 1);
+      if (to > from) await recomputeUserRollups(userId, { from, to });
+    } catch (err) {
+      rollupFailed = true;
+      console.warn(
+        `[${label}] Rollup recompute failed for user ${userId}`,
+        err,
+      );
+    }
+  }
+
+  invalidateUserMeasurements(
+    userId,
+    span ? { touchedFrom: new Date(span.from) } : undefined,
+  );
+
+  const types = Object.entries(result.perType)
+    .filter(([, stat]) => stat.inserted + stat.updated > 0)
+    .map(([type]) => type as MeasurementType);
+  if (types.length > 0) {
+    // Registered so the test suite can await the detached handle.
+    trackBackgroundTask(
+      invalidateStatusInsightsForTypes(userId, types, {
+        source: "manual",
+      }).catch((err) => {
+        console.warn(
+          `[${label}] status-insight invalidate failed for user ${userId}`,
+          err,
+        );
+      }),
+    );
+  }
+  return { rollupFailed };
+}

src/lib/oura/sync-heart-rate.ts

@@ -19,12 +19,8 @@ import { prisma } from "@/lib/db";
 import type { MeasurementType } from "@/generated/prisma/client";
 import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
 import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
-import {
-  collapseToTypeDayKeys,
-  recomputeBucketsForKeys,
-} from "@/lib/rollups/measurement-rollups";
-import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
-import { getEvent } from "@/lib/logging/context";
+import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups";
+import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
 import {
   forEachHeartRatePage,
   mapHeartRateSample,
@@ -120,23 +116,13 @@ export async function refoldOuraTouchedDays(
   touched: TouchedDays,
 ): Promise<void> {
   if (touched.size === 0) return;
-  const keys = touched.list();
-  try {
-    await recomputeBucketsForKeys(userId, keys);
-    invalidateStatusInsightsForTypes(
-      userId,
-      [...new Set(keys.map((k) => k.type))],
-      { source: "sync" },
-    ).catch((err) => {
-      getEvent()?.addWarning(
-        `oura: status-insight invalidate failed for ${userId}: ${err}`,
-      );
-    });
-  } catch (err) {
-    getEvent()?.addWarning(
-      `oura: heart-rate rollup recompute failed for ${userId}: ${err}`,
-    );
-  }
+  // The shared tail: the rollups, the measurement read caches back to the
+  // first touched day (a backfill window reaches years back), and the cached
+  // assessments of the types.
+  await afterMeasurementMutation(userId, touched.list(), {
+    source: "sync",
+    label: "oura heart rate",
+  });
 }
 
 /**

src/lib/oura/sync.ts

@@ -45,17 +45,12 @@ import {
   toFailureKind,
   type FailureKind,
 } from "@/lib/integrations/status";
-import {
-  collapseToTypeDayKeys,
-  recomputeBucketsForKeys,
-} from "@/lib/rollups/measurement-rollups";
 import {
   emitInsertedMeasurementArrivals,
   type InsertedMeasurementArrivalRow,
 } from "@/lib/arrivals/measurement-emit";
 import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
 import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
-import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
 import type { SyncWriteResult } from "@/lib/outcome/written-outcome";
 import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
 import {
@@ -895,23 +890,13 @@ export async function upsertOuraMeasurements(
       () => {},
     );
   }
-  try {
-    const keys = collapseToTypeDayKeys(touched);
-    await recomputeBucketsForKeys(userId, keys);
-    invalidateStatusInsightsForTypes(
-      userId,
-      keys.map((k) => k.type),
-      { source: "sync" },
-    ).catch((err) => {
-      getEvent()?.addWarning(
-        `oura: status-insight invalidate failed for ${userId}: ${err}`,
-      );
-    });
-  } catch (err) {
-    getEvent()?.addWarning(
-      `oura: rollup recompute failed for ${userId}: ${err}`,
-    );
-  }
+  // The shared tail: the rollups, the measurement read caches back to the
+  // oldest reading written (a history backfill writes through here and has
+  // no other invalidation), and the cached assessments of the types.
+  await afterMeasurementMutation(userId, touched, {
+    source: "sync",
+    label: "oura",
+  });
 
   return imported;
 }

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

@@ -12,6 +12,16 @@
  *      inherently: a day left empty drops its rollup row.
  *   2. the cached per-metric status assessments — re-warmed via
  *      `invalidateStatusInsightsForTypes` (fire-and-forget).
+ *   3. the measurement read caches (analytics, targets, derived, workouts),
+ *      marked stale back to the earliest touched reading once the rollups
+ *      are refolded. Without it a writer that relied on the tail alone (the
+ *      CSV and JSON imports, the score engines, the WHOOP webhook, the Withings
+ *      activity, sleep and ECG legs, the Oura history backfill) left
+ *      every cached read, the closed history blocks above all, on the
+ *      pre-write values until their TTL. A caller that invalidated already
+ *      (an interactive write evicts) loses nothing: marking stale after an
+ *      eviction only catches an entry rebuilt from the old rollups between
+ *      the two.
  *
  * Both legs are best-effort: a populator hiccup never fails the user's
  * mutation. Callers collect the affected `(type, measuredAt)` pairs
@@ -27,6 +37,8 @@ import {
   type StatusIngestSource,
 } from "@/lib/insights/status-invalidation";
 import { trackBackgroundTask } from "@/lib/logging/background-tasks";
+import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
+import { earliestMeasuredAt } from "@/lib/cache/closed-block";
 import type { MeasurementType } from "@/generated/prisma/client";
 
 export async function afterMeasurementMutation(
@@ -54,6 +66,7 @@ export async function afterMeasurementMutation(
   } catch (err) {
     console.warn(`[${label}] rollup recompute failed`, err);
   }
+  invalidateUserMeasurements(userId, { touchedFrom: earliestMeasuredAt(rows) });
 
   // Status-insight re-warm is fire-and-forget — it must run even when
   // the rollup leg above failed, so the two legs never gate each other.

src/lib/sleep/sweep-stale-segments.ts

@@ -30,6 +30,7 @@
 import type { MeasurementSource } from "@/generated/prisma/client";
 
 import { prisma } from "@/lib/db";
+import { invalidateUserMeasurements } from "@/lib/cache/invalidate";
 import { getEvent } from "@/lib/logging/context";
 
 /** One just-fetched sleep session: its externalId prefix + the fresh ids. */
@@ -79,5 +80,9 @@ export async function sweepStaleSleepSegments(
       );
     }
   }
+  // A tombstoned segment changes the nights the cached reads hold. The sweep
+  // does not know how far back the swept rows reach, so every cached block
+  // goes; it removes rows only when a session was re-scored, which is rare.
+  if (removed > 0) invalidateUserMeasurements(userId);
   return removed;
 }

src/lib/withings/sync-activity.ts

@@ -44,11 +44,7 @@ import { getEvent } from "@/lib/logging/context";
 import { safeFetch } from "@/lib/safe-fetch";
 import { getUnitForType } from "@/lib/validations/measurement";
 import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
-import {
-  collapseToTypeDayKeys,
-  recomputeBucketsForKeys,
-} from "@/lib/rollups/measurement-rollups";
-import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
+import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
 
 import {
   hasActivityScope,
@@ -578,26 +574,12 @@ export async function syncUserActivity(
   // v1.4.39.1 — refresh the persistent rollup table for every distinct
   // (type, day) the sync touched. Collapsed so a multi-day Withings
   // catch-up costs at most ~N (type, day) recomputes.
-  try {
-    const keys = collapseToTypeDayKeys(touched);
-    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.
-    invalidateStatusInsightsForTypes(
-      userId,
-      keys.map((k) => k.type),
-      { source: "sync" },
-    ).catch((err) => {
-      getEvent()?.addWarning(
-        `withings activity: status-insight invalidate failed for ${userId}: ${err}`,
-      );
-    });
-  } catch (err) {
-    getEvent()?.addWarning(
-      `withings activity: rollup recompute failed for ${userId}: ${err}`,
-    );
-  }
+  // The shared tail: the rollups, the measurement read caches back to the
+  // oldest touched reading, and the cached assessments of the types.
+  await afterMeasurementMutation(userId, touched, {
+    source: "sync",
+    label: "withings activity",
+  });
 
   // Only a clean write stamps success. A swallowed batch-write left the
   // ledger honest via the transient failure recorded above.

src/lib/withings/sync-ecg.ts

@@ -54,11 +54,7 @@ import { prisma } from "@/lib/db";
 import { annotate, getEvent } from "@/lib/logging/context";
 import { safeFetch } from "@/lib/safe-fetch";
 import { persistEcgRecording } from "@/lib/ecg/persist-recording";
-import {
-  collapseToTypeDayKeys,
-  recomputeBucketsForKeys,
-} from "@/lib/rollups/measurement-rollups";
-import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
+import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
 
 import {
   WithingsApiError,
@@ -517,23 +513,12 @@ export async function syncUserEcg(
 
   // Refresh the persistent rollup tier for every distinct (type, day) touched,
   // then drop the per-metric assessment caches the new events dirty.
-  try {
-    const keys = collapseToTypeDayKeys(touched);
-    await recomputeBucketsForKeys(userId, keys);
-    invalidateStatusInsightsForTypes(
-      userId,
-      keys.map((k) => k.type),
-      { source: "sync" },
-    ).catch((err) => {
-      getEvent()?.addWarning(
-        `withings ecg: status-insight invalidate failed for ${userId}: ${err}`,
-      );
-    });
-  } catch (err) {
-    getEvent()?.addWarning(
-      `withings ecg: rollup recompute failed for ${userId}: ${err}`,
-    );
-  }
+  // The shared tail: the rollups, the measurement read caches back to the
+  // oldest touched reading, and the cached assessments of the types.
+  await afterMeasurementMutation(userId, touched, {
+    source: "sync",
+    label: "withings ecg",
+  });
 
   if (sourceWriteError) {
     await recordWithingsSyncFailure(

src/lib/withings/sync-sleep.ts

@@ -43,11 +43,7 @@ import { prisma } from "@/lib/db";
 import { annotate, getEvent } from "@/lib/logging/context";
 import { safeFetch } from "@/lib/safe-fetch";
 import { getUnitForType } from "@/lib/validations/measurement";
-import {
-  collapseToTypeDayKeys,
-  recomputeBucketsForKeys,
-} from "@/lib/rollups/measurement-rollups";
-import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
+import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
 import { maybeEnqueueMorningRefresh } from "@/lib/daily/morning-refresh-trigger";
 
 import {
@@ -693,27 +689,12 @@ export async function syncUserSleep(
   // v1.4.39.1 — refresh the persistent rollup table for every distinct
   // (type, day) the sync touched. Sleep segments collapse heavily —
   // ~10 stage rows per night land in one DAY recompute.
-  try {
-    const keys = collapseToTypeDayKeys(touched);
-    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).
-    // Fire-and-forget.
-    invalidateStatusInsightsForTypes(
-      userId,
-      keys.map((k) => k.type),
-      { source: "sync" },
-    ).catch((err) => {
-      getEvent()?.addWarning(
-        `withings sleep: status-insight invalidate failed for ${userId}: ${err}`,
-      );
-    });
-  } catch (err) {
-    getEvent()?.addWarning(
-      `withings sleep: rollup recompute failed for ${userId}: ${err}`,
-    );
-  }
+  // The shared tail: the rollups, the measurement read caches back to the
+  // oldest touched reading, and the cached assessments of the types.
+  await afterMeasurementMutation(userId, touched, {
+    source: "sync",
+    label: "withings sleep",
+  });
 
   // If a stage segment for last night just landed, kick the debounced
   // morning refresh so the digest/score finalise with the current sleep. The

tests/integration/bulk-writer-cache-invalidation.test.ts

@@ -0,0 +1,176 @@
+/**
+ * Every bulk writer of measurements marks the cached reads stale back to the
+ * oldest reading it wrote.
+ *
+ * A closed analytics block (a value over readings before a fixed instant)
+ * survives every write that says it reached no further back, so a writer that
+ * did not invalidate at all left the block on its pre-write value: an Apple
+ * Health export carrying years of history showed the old all-time figures
+ * until the entry aged out, and the next sync, reaching back only a day,
+ * spared it. Pinned here for the writers that had no invalidation: the Apple
+ * Health import worker, the shared post-mutation tail the CSV and JSON
+ * imports and the score engines rely on, the Oura upsert a history backfill
+ * writes through, and the sleep-segment sweep.
+ */
+import { createHash } from "node:crypto";
+import { mkdtempSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { strToU8, zipSync } from "fflate";
+import type { Job } from "pg-boss";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { closedBlockKeyPart } from "@/lib/cache/closed-block";
+import { caches, type ServerCache } from "@/lib/cache/server-cache";
+import {
+  APPLE_HEALTH_IMPORT_PARSER_REVISION,
+  handleAppleHealthImport,
+  _setWorkerPrismaForTests,
+  type AppleHealthImportPayload,
+} from "@/lib/jobs/apple-health-import-worker";
+import { upsertOuraMeasurements } from "@/lib/oura/sync";
+import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
+import { sweepStaleSleepSegments } from "@/lib/sleep/sweep-stale-segments";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+vi.mock("@/lib/jobs/boss-instance", () => ({
+  getGlobalBoss: vi.fn(() => null),
+}));
+
+const prisma = getPrismaClient();
+const analytics = caches.analytics as ServerCache<{ v: number }>;
+const USER = "user-bulk-writer-cache";
+const DAY_MS = 86_400_000;
+const NOW = Date.now();
+const yearsAgo = (years: number) => new Date(NOW - years * 365 * DAY_MS);
+
+/** A closed block over the readings before `before`, and an open entry. */
+function seedCache(before: Date) {
+  const closed = `${USER}|pre-fold-remainder|${closedBlockKeyPart(before)}|x`;
+  const open = `${USER}|dashboard-snapshot|en`;
+  analytics.deleteByPrefix(`${USER}|`);
+  analytics.set(closed, { v: 1 });
+  analytics.set(open, { v: 2 });
+  return { closed, open };
+}
+
+beforeEach(async () => {
+  await truncateAllTables(prisma);
+  _setWorkerPrismaForTests(prisma);
+  await prisma.user.create({
+    data: {
+      id: USER,
+      username: USER,
+      email: `${USER}@example.test`,
+      timezone: "UTC",
+    },
+  });
+});
+
+function record(at: Date, value: number): string {
+  const stamp = at.toISOString().replace("T", " ").slice(0, 19) + " +0000";
+  return `<Record type="HKQuantityTypeIdentifierBodyMass" unit="kg" startDate="${stamp}" endDate="${stamp}" value="${value}" sourceName="Scale"/>`;
+}
+
+async function runImport(xmlRecords: string[]) {
+  const xml = `<?xml version="1.0" encoding="UTF-8"?>
+<HealthData locale="en_US">
+  ${xmlRecords.join("\n  ")}
+</HealthData>`;
+  const dir = mkdtempSync(join(tmpdir(), "hl-import-cache-"));
+  const path = join(dir, "export.zip");
+  const bytes = Buffer.from(
+    zipSync({ "apple_health_export/export.xml": strToU8(xml) }),
+  );
+  writeFileSync(path, bytes);
+  await prisma.importJob.create({
+    data: {
+      userId: USER,
+      pgBossJobId: "cache-job",
+      status: "queued",
+      uploadBytes: bytes.length,
+      uploadSha256: createHash("sha256").update(bytes).digest("hex"),
+      parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION,
+    },
+  });
+  const outcome = await handleAppleHealthImport({
+    id: "cache-job",
+    data: {
+      userId: USER,
+      uploadPath: path,
+      uploadBytes: bytes.length,
+      enqueuedAt: new Date().toISOString(),
+    },
+  } as unknown as Job<AppleHealthImportPayload>);
+  expect(outcome.ok).toBe(true);
+}
+
+describe("bulk writers invalidate the measurement read caches", () => {
+  it("the Apple Health import marks stale every block its oldest sample reaches", async () => {
+    const reached = seedCache(yearsAgo(1));
+    const older = `${USER}|pre-fold-remainder|${closedBlockKeyPart(yearsAgo(9))}|x`;
+    analytics.set(older, { v: 3 });
+
+    await runImport([record(yearsAgo(7), 88), record(yearsAgo(0.1), 84)]);
+
+    expect(analytics.get(reached.closed)).toBeNull();
+    expect(analytics.get(reached.open)).toBeNull();
+    // A block of readings older than anything imported stays.
+    expect(analytics.get(older)).toEqual({ v: 3 });
+  });
+
+  it("the shared post-mutation tail invalidates back to its oldest row", async () => {
+    const cache = seedCache(yearsAgo(1));
+    await afterMeasurementMutation(
+      USER,
+      [{ type: "WEIGHT", measuredAt: yearsAgo(3) }],
+      { source: "manual" },
+    );
+    expect(analytics.get(cache.closed)).toBeNull();
+    expect(analytics.get(cache.open)).toBeNull();
+  });
+
+  it("the Oura upsert a history backfill writes through invalidates", async () => {
+    const cache = seedCache(yearsAgo(1));
+    const written = await upsertOuraMeasurements(
+      USER,
+      [
+        {
+          type: "RESTING_HEART_RATE",
+          value: 52,
+          unit: "bpm",
+          measuredAt: yearsAgo(2),
+          externalId: "oura:test:resting:1",
+        },
+      ],
+      { emitArrivals: false },
+    );
+    expect(written).toBe(1);
+    expect(analytics.get(cache.closed)).toBeNull();
+  });
+
+  it("a sleep-segment sweep that tombstones a row invalidates", async () => {
+    await prisma.measurement.create({
+      data: {
+        userId: USER,
+        type: "SLEEP_DURATION",
+        value: 60,
+        unit: "min",
+        source: "OURA",
+        measuredAt: yearsAgo(2),
+        externalId: "night-1:stale",
+      },
+    });
+    const cache = seedCache(yearsAgo(1));
+    const removed = await sweepStaleSleepSegments(USER, "OURA", [
+      { prefix: "night-1:", keepIds: ["night-1:fresh"] },
+    ]);
+    expect(removed).toBe(1);
+    expect(analytics.get(cache.closed)).toBeNull();
+  });
+});