feat(measurements): fold Google Health and Oura heart rate like the exports

The dense hourly fold only took Apple Health and Health Connect samples, so heart rate from the provider syncs that store one row per sample stayed raw forever. One self-hosted account held about 25 million Google Health pulse rows after a full-history backfill, and every reader and fold that falls back to the raw table paid for it (#1297). The fold now walks a map of sources to the dense types each stores sample by sample: every dense type for Apple Health and Health Connect, heart rate for Google Health and Oura. Their daily SpO2 values are no intraday stream and stay as they are; Fitbit does not sync intraday heart rate, and WHOOP, Withings and Polar write dense types only once per recovery, night or spot check. Each source folds into its own hourly stats rows, and the DAY rollup is still taken from the raw samples before they go. The fold deletes the samples it averages, and the provider syncs write first-write-wins on the externalId, so a backfill or full sync would have put every folded sample back. Google Health would even have re-keyed an hourly row into a sample that fell on its anchor through the natural-key rescue. The folded_window rule now has one hour rule per fold source, and both syncs drop the samples of an hour the fold already took before they write. Recent samples cost no lookup. The boot discovery of the fold probes each account and source for the first raw sample it would take instead of grouping every such row, and the one-time fold repair stays on the two sources it was written for.
2026-11-08 20:42:43 +00:00 · 2026-10-29 09:06:17 +01:00
1 parent 1d8e2e0ee3
commit 9eed407ccc
@@ -1,7 +1,7 @@
 {
  "$comment": "Production files over 800 lines and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.",
  "count": 84,
-  "total": 94733,
+  "total": 94686,
  "entries": {
    "app/medications/page-client.tsx": 839,
    "components/charts/mood-chart.tsx": 1166,
@@ -71,7 +71,7 @@
    "lib/managed-profiles/handover.ts": 986,
    "lib/mcp/record-search.ts": 1467,
    "lib/mcp/rich-reads.ts": 1467,
-    "lib/measurements/dense-intraday-retention.ts": 1057,
+    "lib/measurements/dense-intraday-retention.ts": 1010,
    "lib/measurements/import-apple-health-export.ts": 1722,
    "lib/medications/scheduling/cadence.ts": 976,
    "lib/medications/scheduling/recurrence.ts": 1100,
@@ -89,7 +89,7 @@ Scope: `googlehealth.health_metrics_and_measurements.readonly`.
 | `daily-heart-rate-variability`        | `dailyHeartRateVariability.averageHeartRateVariabilityMilliseconds`   | `HRV_RMSSD`          | ms          | `hrv`        | daily  | Google documents the daily average as RMSSD, so it lands in `HRV_RMSSD`, apart from the SDNN `HEART_RATE_VARIABILITY`.                                                           |
 | `daily-resting-heart-rate`            | `dailyRestingHeartRate.beatsPerMinute` (int64 string)                 | `RESTING_HEART_RATE` | bpm         | `rhr`        | daily  |                                                                                                                                                                                  |
 | `daily-respiratory-rate`              | `dailyRespiratoryRate.breathsPerMinute` (number)                      | `RESPIRATORY_RATE`   | breaths/min | `resp_rate`  | daily  | `respiratory-rate` does NOT exist in the catalogue. Schema is `{date, breathsPerMinute}` — the earlier `dailyRespiratoryRateBpm` leaf never existed.                             |
-| `heart-rate`                          | `heartRate.beatsPerMinute` (int64 string)                             | `PULSE`              | bpm         | `hr`         | sample | Intraday spot HR; per-minute volume — consider the 14-day-max rollup if PULSE should stay coarse.                                                                                |
+| `heart-rate`                          | `heartRate.beatsPerMinute` (int64 string)                             | `PULSE`              | bpm         | `hr`         | sample | Intraday spot HR, one row per sample; older than 90 days it folds to hourly means (`dense-intraday-sources.ts`).                                                                 |
 | `blood-glucose`                       | `bloodGlucose.bloodGlucoseMilligramsPerDeciliter` (number)            | `BLOOD_GLUCOSE`      | mg/dL       | `glucose`    | sample | Already in the canonical mg/dL storage unit. Meal/timing/specimen enums not stored.                                                                                              |
 | `core-body-temperature`               | `coreBodyTemperature.temperatureCelsius` (number)                     | `BODY_TEMPERATURE`   | celsius     | `core_temp`  | sample | Core slot — distinct from SKIN_TEMPERATURE / WRIST_TEMPERATURE.                                                                                                                  |
 | `daily-sleep-temperature-derivations` | `dailySleepTemperatureDerivations.nightlyTemperatureCelsius` (number) | `WRIST_TEMPERATURE`  | celsius     | `wrist_temp` | daily  | ABSOLUTE nightly skin temperature (mean over the night's samples) — the schema refutes the earlier "deviation-only" premise. Baseline/stddev derivations not stored.             |
@@ -36,6 +36,7 @@ import {
  type MeasurementReconciliationVerdict,
 } from "@/lib/measurements/reconcile-external-measurement";
 import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate";
+import { withoutFoldedSamples } from "@/lib/measurements/folded-window";
 import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit";
 import { recordSyncFailure, type FailureKind } from "@/lib/integrations/status";
 import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups";
@@ -1128,7 +1129,7 @@ export async function upsertGoogleHealthMeasurements(
  // before they can become rows. A refused reading is dropped and tallied on
  // the ambient event, never clamped and never written for a later reader to
  // discover.
-  const { kept: readings, dropped: naturalKeyTwins } = collapseNaturalKeyTwins(
+  const { kept: collapsed, dropped: naturalKeyTwins } = collapseNaturalKeyTwins(
    dropImplausibleMeasurements("googleHealth", incoming, (r) => ({
      type: r.type,
      value: r.value,
@@ -1137,14 +1138,18 @@ export async function upsertGoogleHealthMeasurements(
  if (naturalKeyTwins > 0) {
    annotate({ meta: { "googleHealth.naturalKeyTwins": naturalKeyTwins } });
  }
+  // A sample of an hour the dense fold already took is not written again.
+  const { kept: readings } = await withoutFoldedSamples(
+    prisma,
+    userId,
+    "GOOGLE_HEALTH",
+    collapsed,
+  );
  if (readings.length === 0) return { imported: 0, touched: [], inserted: [] };

  // Probe EVERY existing row (live AND tombstoned) for the batch's externalIds
-  // in a single query. The full unique index guarantees at most one row per
-  // `(type, externalId)` key, and a tombstoned match must take the UPDATE
-  // branch (which clears `deletedAt`) — treating it as absent would plan an
-  // insert that `skipDuplicates` silently drops against the tombstone's key,
-  // permanently wedging the key (see TOMBSTONES RESURRECT above).
+  // in one query: a tombstoned match must take the UPDATE branch, or the
+  // insert is silently dropped against its key (see TOMBSTONES RESURRECT).
  const externalIds = readings.map((r) => r.externalId);
  interface ProbedRow {
    id: string;
@@ -1162,10 +1167,8 @@ export async function upsertGoogleHealthMeasurements(
        source: "GOOGLE_HEALTH",
        externalId: { in: externalIds },
      },
-      // The payload fields ride along so the update branch can skip a no-op
-      // overwrite: the 24 h overlap re-fetches every recent row hourly, and an
-      // unconditional write would bump `syncVersion` (iOS delta churn) on rows
-      // whose values did not change.
+      // The payload rides along so the update branch can skip a no-op
+      // overwrite (the hourly 24 h overlap would bump `syncVersion` for nothing).
      select: {
        id: true,
        type: true,
@@ -1193,9 +1196,8 @@ export async function upsertGoogleHealthMeasurements(
        ]),
    );
  } catch (err) {
-    // A probe failure must not strand the whole batch; fall back to treating
-    // every row as fresh (`skipDuplicates` absorbs the collision on the full
-    // unique index — those rows simply retry on the next sync tick).
+    // A probe failure must not strand the batch: every row is treated as
+    // fresh, and `skipDuplicates` absorbs the collisions until the next tick.
    getEvent()?.addWarning(`google-health: row probe failed: ${err}`);
  }

@@ -1225,11 +1227,9 @@ export async function upsertGoogleHealthMeasurements(
    const key = `${type} ${r.externalId}`;
    const existingRow = rowByKey.get(key);
    if (existingRow) {
-      // Skip the no-op overwrite: a LIVE row whose payload matches the
-      // re-fetched reading exactly gains nothing from a write, and the
-      // unconditional `syncVersion` bump churned the DB + every iOS delta pull
-      // hourly for the whole 24 h overlap window. A TOMBSTONED row always
-      // updates — the write is what resurrects it (`deletedAt: null`).
+      // Skip the no-op overwrite of a LIVE row whose payload matches exactly
+      // (it would churn `syncVersion` and every iOS delta pull). A TOMBSTONED
+      // row always updates: the write is what resurrects it.
      const unchanged =
        existingRow.deletedAt === null &&
        existingRow.value === r.value &&
@@ -30,10 +30,8 @@ import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import { DENSE_INTRADAY_RETENTION_ENABLED } from "@/lib/jobs/dense-intraday-retention";
-import {
-  DENSE_INTRADAY_RETENTION_DAYS,
-  DENSE_INTRADAY_RETENTION_TYPES,
-} from "@/lib/measurements/dense-intraday-retention";
+import { DENSE_INTRADAY_RETENTION_DAYS } from "@/lib/measurements/dense-intraday-retention";
+import { DENSE_INTRADAY_RETENTION_TYPES } from "@/lib/measurements/dense-intraday-sources";
 import { runDenseIntradayHourlyRebuild } from "@/lib/measurements/dense-intraday-hourly-rebuild";
 import { DENSE_INTRADAY_HOURLY_REBUILD_QUEUE } from "./dense-intraday-hourly-rebuild-queue";

@@ -18,16 +18,18 @@
 * `src/lib/jobs/reminder-worker.ts` so pg-boss provisions it at boot; an
 * unregistered queue silently never drains.
 */
-import { dropPendingDeletion } from "@/lib/account-deletion/active-account";
+import { ACTIVE_ACCOUNT } from "@/lib/account-deletion/active-account";
 import { prisma } from "@/lib/db";
 import { annotate } from "@/lib/logging/context";
 import { getGlobalBoss } from "@/lib/jobs/boss-instance";
 import {
  runDenseIntradayRetention,
-  DENSE_INTRADAY_RETENTION_TYPES,
  DENSE_INTRADAY_RETENTION_DAYS,
-  DENSE_INTRADAY_FOLD_SOURCES,
 } from "@/lib/measurements/dense-intraday-retention";
+import {
+  DENSE_INTRADAY_FOLD_SOURCES,
+  DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
+} from "@/lib/measurements/dense-intraday-sources";
 import { DENSE_INTRADAY_RETENTION_QUEUE } from "./dense-intraday-retention-queue";

 /**
@@ -133,8 +135,8 @@ export async function runDenseIntradayRetentionForUser(

 /**
 * Boot-time discovery. Finds every user holding at least one LIVE
- * per-sample dense-tier row (a row of a `DENSE_INTRADAY_FOLD_SOURCES` source
- * and a `DENSE_INTRADAY_RETENTION_TYPES` type whose externalId does NOT start
+ * per-sample dense-tier row (a row of a source and type the fold takes,
+ * `DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE`, whose externalId does NOT start
 * with the daily-stats prefix, that is not tombstoned, AND whose
 * `measuredAt` is older than the retention window) and enqueues one
 * retention job per account.
@@ -161,36 +163,40 @@ export async function enqueueBootTimeDenseIntradayRetention(): Promise<{
  }

  try {
-    // The dense-tier type set is a closed compile-time list of enum
-    // members; splice-free — Prisma binds the `type IN (...)` array as
-    // parameters.
-    const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES);
-    if (types.length === 0) {
-      return { enqueued: 0, skipped: 0, error: null };
-    }
-
    const windowStart = new Date(
      Date.now() - DENSE_INTRADAY_RETENTION_DAYS * 24 * 60 * 60 * 1000,
    );

-    // `groupBy` compiles to a server-side `GROUP BY` — Prisma's
-    // `distinct` dedups in the client AFTER pulling every matching row,
-    // an instance-wide scan of every out-of-window dense-tier row just
-    // to list user ids.
-    const users = await dropPendingDeletion(
-      prisma,
-      await prisma.measurement.groupBy({
-        by: ["userId"],
-        where: {
-          source: { in: [...DENSE_INTRADAY_FOLD_SOURCES] },
-          type: { in: types },
-          deletedAt: null,
-          NOT: { externalId: { startsWith: "stats:" } },
-          measuredAt: { lt: windowStart },
-        },
-      }),
-      (u) => u.userId,
-    );
+    // One probe per account and source for the first raw sample the fold
+    // would take, which stops at that row. A grouped count over every such
+    // row read the whole out-of-window history of every account to list
+    // their ids, and on an account with tens of millions of heart-rate rows
+    // that is most of the table (#1297).
+    const accounts = await prisma.user.findMany({
+      where: ACTIVE_ACCOUNT,
+      select: { id: true },
+      orderBy: { createdAt: "asc" },
+    });
+    const users: Array<{ userId: string }> = [];
+    for (const { id } of accounts) {
+      for (const source of DENSE_INTRADAY_FOLD_SOURCES) {
+        const raw = await prisma.measurement.findFirst({
+          where: {
+            userId: id,
+            source,
+            type: { in: [...DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE[source]] },
+            deletedAt: null,
+            NOT: { externalId: { startsWith: "stats:" } },
+            measuredAt: { lt: windowStart },
+          },
+          select: { id: true },
+        });
+        if (raw) {
+          users.push({ userId: id });
+          break;
+        }
+      }
+    }

    if (users.length === 0) {
      return { enqueued: 0, skipped: 0, error: null };
@@ -116,12 +116,11 @@ import {
  hourOfDayForUserTz,
 } from "@/lib/measurements/consolidation-tz";
 import {
-  DENSE_INTRADAY_FOLD_SOURCES,
  DENSE_INTRADAY_RETENTION_DAYS,
-  DENSE_INTRADAY_RETENTION_TYPES,
  deriveDailyRestingFromPulse,
  hourlyStatsExternalId,
 } from "@/lib/measurements/dense-intraday-retention";
+import { DENSE_INTRADAY_RETENTION_TYPES } from "@/lib/measurements/dense-intraday-sources";
 import {
  absorbIntoFold,
  foldConstituentHorizon,
@@ -160,6 +159,9 @@ interface RepairStep {
  foldedAfterMs: number;
 }

+/** Sources the dense fold took when this repair was written; later ones never split a window. */
+const REPAIRED_DENSE_SOURCES = ["APPLE_HEALTH", "HEALTH_CONNECT"] as const;
+
 /** Fixed order, so a resume position (`step`) means the same on every run. */
 export const FOLD_REPAIR_STEPS: readonly RepairStep[] = [
  ...[...HIGH_FREQUENCY_MEAN_TYPES].map((type): RepairStep => ({
@@ -168,7 +170,7 @@ export const FOLD_REPAIR_STEPS: readonly RepairStep[] = [
    grain: "day",
    foldedAfterMs: CONSOLIDATION_GRACE_CUTOFF_HOURS * HOUR_MS,
  })),
-  ...DENSE_INTRADAY_FOLD_SOURCES.flatMap((source) =>
+  ...REPAIRED_DENSE_SOURCES.flatMap((source) =>
    [...DENSE_INTRADAY_RETENTION_TYPES].map((type): RepairStep => ({
      type,
      source,
@@ -26,7 +26,7 @@ import {
  CUMULATIVE_HK_TYPES,
  HIGH_FREQUENCY_MEAN_TYPES,
 } from "../apple-health-mapping";
-import { DENSE_INTRADAY_RETENTION_TYPES } from "../dense-intraday-retention";
+import { DENSE_INTRADAY_RETENTION_TYPES } from "../dense-intraday-sources";

 describe("dense intra-day retention — drain exemption (WX-E)", () => {
  it("EXEMPTS HEART_RATE_VARIABILITY from the destructive daily-mean drain", () => {
@@ -21,8 +21,8 @@ vi.mock("@/lib/rollups/measurement-rollups", () => ({
 import {
  runDenseIntradayRetention,
  DENSE_INTRADAY_RETENTION_DAYS,
-  DENSE_INTRADAY_RETENTION_TYPES,
 } from "../dense-intraday-retention";
+import { DENSE_INTRADAY_RETENTION_TYPES } from "../dense-intraday-sources";
 import type { PerSampleRow } from "../drain-per-sample-cumulative";
 import { foldBoundary } from "../consolidation-tz";
 import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
@@ -161,7 +161,7 @@ describe("runDenseIntradayRetention — retention bound", () => {
    }
  });

-  it("scans exactly the dense-tier types, once per fold source, Apple Health first", async () => {
+  it("scans each fold source's dense types, once per source, Apple Health first", async () => {
    const { mock, findManyMeasurement } = buildPrismaMock({});
    await runDenseIntradayRetention(mock, { log: () => {} });

@@ -172,9 +172,13 @@ describe("runDenseIntradayRetention — retention bound", () => {
    const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES);
    // One walk per source, each scoped to its one source: two sources
    // never share a mean, and manual or Withings rows are never scanned.
+    // The provider syncs fold only what they store sample by sample: heart
+    // rate, never their daily SpO2 (#1297).
    expect(scans).toEqual([
      ...types.map((type) => `APPLE_HEALTH:${type}`),
      ...types.map((type) => `HEALTH_CONNECT:${type}`),
+      "GOOGLE_HEALTH:PULSE",
+      "OURA:PULSE",
    ]);
  });
 });
@@ -14,6 +14,7 @@ import {
  findCompactionTombstones,
  findFoldedWindowDuplicates,
  isFoldedWindowCandidate,
+  withoutFoldedSamples,
 } from "../folded-window";

 const DAY = 86_400_000;
@@ -332,3 +333,95 @@ describe("Health Connect", () => {
    ]);
  });
 });
+
+describe("provider syncs (#1297)", () => {
+  const old = new Date("2026-03-01T10:15:00.000Z");
+  const hour = "stats:HKQuantityTypeIdentifierHeartRate:2026-03-01T10";
+
+  it("takes the heart rate Google Health and Oura store per sample, and nothing else of theirs", () => {
+    for (const source of ["GOOGLE_HEALTH", "OURA"]) {
+      expect(isFoldedWindowCandidate(raw("PULSE", old, { source }), NOW)).toBe(
+        true,
+      );
+      // A nightly or daily SpO2 value is no intraday stream: never folded.
+      expect(
+        isFoldedWindowCandidate(raw("OXYGEN_SATURATION", old, { source }), NOW),
+      ).toBe(false);
+    }
+    // Neither stores intraday heart rate as samples.
+    for (const source of ["FITBIT", "WHOOP", "WITHINGS"]) {
+      expect(isFoldedWindowCandidate(raw("PULSE", old, { source }), NOW)).toBe(
+        false,
+      );
+    }
+  });
+
+  it("drops the samples of an hour the fold took, under the write's own source", async () => {
+    // The externalId read finds the hour's row; the reach read finds nothing.
+    const findMany = vi.fn(
+      async (args: {
+        where: { source: string; externalId: { in?: string[] } };
+      }) =>
+        args.where.source === "GOOGLE_HEALTH" &&
+        args.where.externalId.in?.includes(hour)
+          ? [{ type: "PULSE", externalId: hour }]
+          : [],
+    );
+    const findUnique = vi.fn(async () => ({ timezone: "UTC" }));
+    const rows = [
+      { type: "PULSE", externalId: "hr:a", measuredAt: old, value: 61 },
+      {
+        type: "PULSE",
+        externalId: "hr:b",
+        measuredAt: new Date("2026-03-01T12:15:00.000Z"),
+        value: 62,
+      },
+      {
+        type: "PULSE",
+        externalId: "hr:c",
+        measuredAt: new Date(NOW.getTime() - DAY),
+        value: 63,
+      },
+    ];
+    const { kept, dropped } = await withoutFoldedSamples(
+      {
+        measurement: { findMany },
+        user: { findUnique },
+      } as unknown as Parameters<typeof withoutFoldedSamples>[0],
+      "u1",
+      "GOOGLE_HEALTH",
+      rows,
+      { now: NOW },
+    );
+    expect(dropped).toBe(1);
+    expect(kept.map((row) => row.externalId)).toEqual(["hr:b", "hr:c"]);
+    for (const call of findMany.mock.calls) {
+      expect(call[0].where.source).toBe("GOOGLE_HEALTH");
+    }
+  });
+
+  it("asks nothing for a write of recent samples", async () => {
+    const findMany = vi.fn();
+    const findUnique = vi.fn();
+    const rows = [
+      {
+        type: "PULSE",
+        externalId: "heartrate:x",
+        measuredAt: new Date(NOW.getTime() - 2 * 3_600_000),
+      },
+    ];
+    const { kept } = await withoutFoldedSamples(
+      {
+        measurement: { findMany },
+        user: { findUnique },
+      } as unknown as Parameters<typeof withoutFoldedSamples>[0],
+      "u1",
+      "OURA",
+      rows,
+      { now: NOW },
+    );
+    expect(kept).toEqual(rows);
+    expect(findMany).not.toHaveBeenCalled();
+    expect(findUnique).not.toHaveBeenCalled();
+  });
+});
@@ -72,11 +72,11 @@ import { localDayWindow } from "@/lib/tz/local-day";
 import { meanBucketValue } from "./consolidate-daily-mean";
 import {
  DENSE_INTRADAY_RETENTION_DAYS,
-  DENSE_INTRADAY_RETENTION_TYPES,
  adoptOrMintHourlyRows,
  bucketRowsByLocalHour,
  hourlyStatsExternalId,
 } from "./dense-intraday-retention";
+import { DENSE_INTRADAY_RETENTION_TYPES } from "./dense-intraday-sources";
 import {
  bucketSpan,
  enqueueRollupRecompute,
@@ -56,7 +56,6 @@ import {
  FOLD_TRANSACTION_OPTIONS,
 } from "@/lib/export/restore-lock";
 import type {
-  MeasurementSource,
  MeasurementType,
  Prisma,
  PrismaClient,
@@ -78,6 +77,11 @@ import { percentile } from "@/lib/insights/strain-score";
 import type { PerSampleRow } from "./drain-per-sample-cumulative";
 import { DEFAULT_TIMEZONE } from "@/lib/tz/format";
 import { localDayWindow } from "@/lib/tz/local-day";
+import {
+  DENSE_INTRADAY_FOLD_SOURCES,
+  DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
+  type DenseIntradayFoldSource,
+} from "./dense-intraday-sources";
 import {
  absorbIntoFold,
  foldConstituentHorizon,
@@ -121,40 +125,6 @@ import {
 *      they did against the raw stream's successor rows before.
 */

-/**
- * The dense-tier types. These are the high-frequency intra-day signals that
- * accumulate raw per-sample rows the destructive daily-mean drain never
- * touches:
- *
- *   - `HEART_RATE_VARIABILITY` + `PULSE` — the daytime shape the Stress
- *     engine reads (v1.10.0 / v1.18.11).
- *   - `OXYGEN_SATURATION` — Apple Watch's Blood-Oxygen app samples SpO2
- *     periodically through the day and overnight; the readings map to
- *     `aggregation: "latest"` and sit in NEITHER `CUMULATIVE_HK_TYPES` nor
- *     `HIGH_FREQUENCY_MEAN_TYPES`, so every sample piled up raw forever.
- *     Folded here to hourly-mean rows with the daily min/max preserved on
- *     the rollup tier (the lowest sample is the overnight-desaturation
- *     nadir — clinically the meaningful figure, not the mean).
- *
- * Every member is deliberately EXEMPT from the destructive daily-mean drain
- * (`HIGH_FREQUENCY_MEAN_TYPES`); the disjointness from that set is the
- * load-bearing invariant a regression test pins — a type in both would be
- * folded by two drains at once and corrupt the value.
- *
- * `RESPIRATORY_RATE` is deliberately NOT here: it is already a member of
- * `HIGH_FREQUENCY_MEAN_TYPES`, so the nightly mean-consolidation already
- * collapses it to a daily-mean row. Adding it here would double-fold it.
- *
- * One fold grain per tier: every member folds HOURLY. A per-type grain
- * split would fork the write path and the rebuild for no reader benefit.
- */
-export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
-  new Set<MeasurementType>([
-    "HEART_RATE_VARIABILITY",
-    "PULSE",
-    "OXYGEN_SATURATION",
-  ]);
-
 /**
 * Retention bound (days). Per-sample dense-tier rows older than this window
 * fold to hourly-mean rows and the raw rows are deleted; rows inside the
@@ -173,23 +143,6 @@ export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
 */
 export const DENSE_INTRADAY_RETENTION_DAYS = 90;

-/**
- * The sources whose raw dense samples fold (v1.42). Apple Health from the
- * start; Health Connect because its importer writes the samples of the last
- * 90 days raw, exactly like an Apple Health sync, and without a fold every
- * new export would add another quarter of minute-by-minute heart rate that
- * never ages into hourly means. Each source folds into its OWN `stats:` rows,
- * so an hour two sources both covered keeps one mean per source, the same
- * way the raw samples kept their source.
- */
-export const DENSE_INTRADAY_FOLD_SOURCES = [
-  "APPLE_HEALTH",
-  "HEALTH_CONNECT",
-] as const satisfies readonly MeasurementSource[];
-
-export type DenseIntradayFoldSource =
-  (typeof DENSE_INTRADAY_FOLD_SOURCES)[number];
-
 /** The stats externalId prefix marks an already-collapsed row. */
 const DAILY_STATS_PREFIX = "stats:";

@@ -590,7 +543,7 @@ export async function runDenseIntradayRetention(
    await runConsolidation<MeasurementType>({
      prismaClient,
      options: { ...options, cutoffHours },
-      types: DENSE_INTRADAY_RETENTION_TYPES,
+      types: DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE[source],
      hkIdentifierForType,
      dailyStatsExternalId,
      statsPrefix: DAILY_STATS_PREFIX,
@@ -1034,8 +987,8 @@ export async function runDenseIntradayRetention(
      },
    });

-  // One walk per source, Apple Health first so its behaviour and its order
-  // are exactly what they were before Health Connect folded too. A walk that
+  // One walk per source, in the map's order (Apple Health first, so its
+  // behaviour is what it was before other sources folded too). A walk that
  // stopped early stops the pass: the next run resumes where it left off.
  for (const source of DENSE_INTRADAY_FOLD_SOURCES) {
    const walk = await walkSource(source);
@@ -0,0 +1,99 @@
+/**
+ * What the dense hourly fold (`dense-intraday-retention.ts`) takes: the
+ * dense-tier types, and for each source the types it stores one row per
+ * intraday sample of. The `folded_window` ingest guard (`folded-window.ts`)
+ * reads the same map, so a source's raw samples are refused at ingest for
+ * exactly the hours the fold has taken from it.
+ */
+import type {
+  MeasurementSource,
+  MeasurementType,
+} from "@/generated/prisma/client";
+
+/**
+ * The dense-tier types. These are the high-frequency intra-day signals that
+ * accumulate raw per-sample rows the destructive daily-mean drain never
+ * touches:
+ *
+ *   - `HEART_RATE_VARIABILITY` + `PULSE` — the daytime shape the Stress
+ *     engine reads.
+ *   - `OXYGEN_SATURATION` — Apple Watch's Blood-Oxygen app samples SpO2
+ *     periodically through the day and overnight; the readings map to
+ *     `aggregation: "latest"` and sit in NEITHER `CUMULATIVE_HK_TYPES` nor
+ *     `HIGH_FREQUENCY_MEAN_TYPES`, so every sample piled up raw forever.
+ *     Folded here to hourly-mean rows with the daily min/max preserved on
+ *     the rollup tier (the lowest sample is the overnight-desaturation
+ *     nadir — clinically the meaningful figure, not the mean).
+ *
+ * Every member is deliberately EXEMPT from the destructive daily-mean drain
+ * (`HIGH_FREQUENCY_MEAN_TYPES`); the disjointness from that set is the
+ * load-bearing invariant a regression test pins — a type in both would be
+ * folded by two drains at once and corrupt the value.
+ *
+ * `RESPIRATORY_RATE` is deliberately NOT here: it is already a member of
+ * `HIGH_FREQUENCY_MEAN_TYPES`, so the nightly mean-consolidation already
+ * collapses it to a daily-mean row. Adding it here would double-fold it.
+ *
+ * One fold grain per tier: every member folds HOURLY. A per-type grain
+ * split would fork the write path and the rebuild for no reader benefit.
+ */
+export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
+  new Set<MeasurementType>([
+    "HEART_RATE_VARIABILITY",
+    "PULSE",
+    "OXYGEN_SATURATION",
+  ]);
+
+/**
+ * The sources whose raw dense samples fold, and which dense types of each.
+ * A source belongs here only for a type it stores sample by sample through
+ * the day; a type it stores as one value a day (a nightly SpO2 average, a
+ * recovery's HRV) is no intraday stream, and folding that row would only
+ * move it to another hour.
+ *
+ *   - Apple Health and Health Connect: every dense type. Both write the
+ *     samples of an export raw, and without the fold every new export would
+ *     add another quarter of minute-by-minute heart rate.
+ *   - Google Health: heart rate, one row per sample from the `heart-rate`
+ *     collection (SpO2 comes from the daily collection). A full-history
+ *     backfill brought tens of millions of these onto one account (#1297).
+ *   - Oura: heart rate, a reading every few minutes from the `heartrate`
+ *     collection (SpO2 is the nightly average).
+ *
+ * Fitbit does not sync intraday heart rate, and WHOOP, Withings and Polar
+ * write dense types only as one value per recovery, night or spot check, so
+ * none of them is here.
+ *
+ * Each source folds into its OWN `stats:` rows, so an hour two sources both
+ * covered keeps one mean per source, the same way the raw samples kept their
+ * source. The walk runs in this order.
+ */
+export const DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE = {
+  APPLE_HEALTH: DENSE_INTRADAY_RETENTION_TYPES,
+  HEALTH_CONNECT: DENSE_INTRADAY_RETENTION_TYPES,
+  GOOGLE_HEALTH: new Set<MeasurementType>(["PULSE"]),
+  OURA: new Set<MeasurementType>(["PULSE"]),
+} as const satisfies Partial<
+  Record<MeasurementSource, ReadonlySet<MeasurementType>>
+>;
+
+export type DenseIntradayFoldSource =
+  keyof typeof DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE;
+
+/** Every source the fold walks, in walk order. */
+export const DENSE_INTRADAY_FOLD_SOURCES = Object.keys(
+  DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
+) as readonly DenseIntradayFoldSource[];
+
+/** Whether the fold takes `type` from `source`. */
+export function isDenseFoldedSample(
+  source: MeasurementSource,
+  type: MeasurementType,
+): boolean {
+  const types = (
+    DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE as Partial<
+      Record<MeasurementSource, ReadonlySet<MeasurementType>>
+    >
+  )[source];
+  return types?.has(type) ?? false;
+}
@@ -47,12 +47,14 @@
 * of tombstones does not use that reach; a tombstone it misses stays class B,
 * which keeps a row longer, never loses one.
 *
- * Health Connect folds its raw heart-rate history the same way under its own
- * source (the importer writes samples of the last 90 days raw, and the
- * nightly pass folds them once they age out), so the hour rule covers both
- * sources and a covering row is always looked up under the sample's own
- * source. The day rule stays Apple Health only: the nightly mean
- * consolidation reads no other source.
+ * Every source the dense fold walks folds under its own source, for the
+ * types it stores sample by sample (`DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE`):
+ * Health Connect from its importer, and Google Health and Oura from their
+ * syncs, whose backfills walk years of heart rate (#1297). So there is one
+ * hour rule per source, and a covering row is always looked up under the
+ * sample's own source. The provider syncs ask the same question before they
+ * write (`withoutFoldedSamples`). The day rule stays Apple Health only: the
+ * nightly mean consolidation reads no other source.
 */
 import type {
  MeasurementSource,
@@ -71,11 +73,14 @@ import {
  hourOfDayForUserTz,
 } from "@/lib/measurements/consolidation-tz";
 import {
-  DENSE_INTRADAY_FOLD_SOURCES,
  DENSE_INTRADAY_RETENTION_DAYS,
-  DENSE_INTRADAY_RETENTION_TYPES,
  hourlyStatsExternalId,
 } from "@/lib/measurements/dense-intraday-retention";
+import {
+  DENSE_INTRADAY_FOLD_SOURCES,
+  DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE,
+  DENSE_INTRADAY_RETENTION_TYPES,
+} from "@/lib/measurements/dense-intraday-sources";

 /** The per-entry status reason the batch result carries for such a row. */
 export type FoldedWindowReason = "folded_window";
@@ -88,33 +93,38 @@ const HOUR_MS = 3_600_000;
 * `reachMs` is how far from a sample the anchor of its window can sit in any
 * zone (see the module comment), for the timezone-change check.
 */
-export const FOLDED_WINDOW_RULES = [
-  {
+export const FOLDED_WINDOW_RULES: readonly FoldedWindowRule[] = [
+  // One hour rule per source the dense fold walks, with that source's types.
+  ...DENSE_INTRADAY_FOLD_SOURCES.map((source): FoldedWindowRule => ({
    window: "hour",
-    sources: DENSE_INTRADAY_FOLD_SOURCES,
-    types: DENSE_INTRADAY_RETENTION_TYPES,
+    sources: [source],
+    types: DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE[source],
    foldedAfterMs: DENSE_INTRADAY_RETENTION_DAYS * 24 * HOUR_MS,
    reachMs: HOUR_MS / 2,
-  },
+  })),
  {
    window: "day",
-    sources: ["APPLE_HEALTH"] as const,
+    sources: ["APPLE_HEALTH"],
    types: HIGH_FREQUENCY_MEAN_TYPES,
    foldedAfterMs: CONSOLIDATION_GRACE_CUTOFF_HOURS * HOUR_MS,
    reachMs: 14 * HOUR_MS,
  },
-] as const;
+];

-type FoldedWindowRule = (typeof FOLDED_WINDOW_RULES)[number];
+interface FoldedWindowRule {
+  window: "hour" | "day";
+  sources: readonly MeasurementSource[];
+  types: ReadonlySet<MeasurementType>;
+  foldedAfterMs: number;
+  reachMs: number;
+}

 function ruleFor(
  row: Pick<FoldedWindowCandidateRow, "type" | "source">,
 ): FoldedWindowRule | null {
  return (
    FOLDED_WINDOW_RULES.find(
-      (rule) =>
-        (rule.sources as readonly MeasurementSource[]).includes(row.source) &&
-        rule.types.has(row.type),
+      (rule) => rule.sources.includes(row.source) && rule.types.has(row.type),
    ) ?? null
  );
 }
@@ -409,6 +419,46 @@ export async function findFoldedWindowDuplicates(
  );
 }

+/**
+ * The rows of one provider write without the raw samples whose hour the
+ * dense fold has already taken from that source (#1297).
+ *
+ * The provider syncs write first-write-wins on `(user, type, source,
+ * externalId)`, and the fold deletes the samples it averages, so a backfill
+ * or a full sync that walks old history would find no row for a folded
+ * sample and insert it again beside the hourly mean that already accounts
+ * for it. Google Health would even re-key the hourly row: a sample that fell
+ * on the hour's anchor shares its instant, and the sync's natural-key rescue
+ * then overwrote the mean with the sample. Rows too young to have been folded
+ * cost no query, so the hourly syncs never ask.
+ */
+export async function withoutFoldedSamples<
+  T extends { type: string; externalId?: string | null; measuredAt: Date },
+>(
+  client: LookupClient,
+  userId: string,
+  source: MeasurementSource,
+  rows: readonly T[],
+  options: { now?: Date } = {},
+): Promise<{ kept: T[]; dropped: number }> {
+  const folded = await findFoldedWindowDuplicates(
+    client,
+    userId,
+    rows.map((row) => ({
+      type: row.type as MeasurementType,
+      source,
+      externalId: row.externalId ?? null,
+      measuredAt: row.measuredAt,
+    })),
+    options,
+  );
+  if (folded.size === 0) return { kept: [...rows], dropped: 0 };
+  return {
+    kept: rows.filter((_, index) => !folded.has(index)),
+    dropped: folded.size,
+  };
+}
+
 /** A tombstone as the classifier reads it. */
 export interface TombstoneCandidateRow extends FoldedWindowCandidateRow {
  deletedAt: Date | null;
@@ -11,14 +11,16 @@
 * Writes are first-write-wins: a sample is final once it is in Oura's cloud,
 * and it is keyed on its own instant (`heartrate:<ISO>`), so a re-read of the
 * same range is an insert that `skipDuplicates` drops. A row the person
- * deleted keeps its tombstone; the re-read does not bring it back. Oura's
- * samples go raw, like Google Health's and Fitbit's minute stream: the dense
- * hourly fold covers only the export sources (`DENSE_INTRADAY_FOLD_SOURCES`).
+ * deleted keeps its tombstone; the re-read does not bring it back. Samples
+ * older than the retention window are folded into hourly means like Google
+ * Health's (`DENSE_INTRADAY_FOLD_TYPES_BY_SOURCE`), and a re-read of a folded
+ * hour writes nothing (`withoutFoldedSamples`).
 */
 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 { withoutFoldedSamples } from "@/lib/measurements/folded-window";
 import { collapseToTypeDayKeys } from "@/lib/rollups/measurement-rollups";
 import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation";
 import { forEachHeartRatePage } from "./client";
@@ -76,10 +78,17 @@ export async function writeOuraHeartRatePage(
    const m = mapHeartRateSample(s);
    if (m) mapped.push(m);
  }
-  const kept = dropImplausibleMeasurements("oura", mapped, (m) => ({
-    type: m.type,
-    value: m.value,
-  }));
+  // A sample of an hour the dense fold already took is not written again: it
+  // was deleted with the fold, so `skipDuplicates` would not stop it.
+  const { kept } = await withoutFoldedSamples(
+    prisma,
+    userId,
+    "OURA",
+    dropImplausibleMeasurements("oura", mapped, (m) => ({
+      type: m.type,
+      value: m.value,
+    })),
+  );
  let inserted = 0;
  for (let i = 0; i < kept.length; i += OURA_HEART_RATE_WRITE_CHUNK) {
    const chunk = kept.slice(i, i + OURA_HEART_RATE_WRITE_CHUNK);
@@ -0,0 +1,185 @@
+/**
+ * The dense hourly fold over the provider syncs that store heart rate one row
+ * per sample, and their ingest guard, against Postgres (#1297).
+ *
+ *   - Google Health and Oura heart rate older than the retention window folds
+ *     to hourly `stats:` rows under its own source, the raw rows go, and the
+ *     DAY rollup keeps the figures it had from the raw samples.
+ *   - A daily SpO2 value of the same sources is no intraday stream and stays.
+ *   - Writing the same old samples again, as a backfill or a full sync does,
+ *     inserts nothing and leaves every hourly mean as it was, including the
+ *     hour whose anchor a sample fell on (Google Health's natural-key rescue
+ *     used to re-key that row back into the sample).
+ *   - Samples inside the window still write as before.
+ */
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { upsertGoogleHealthMeasurements } from "@/lib/google-health/sync-core";
+import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention";
+import {
+  TouchedDays,
+  writeOuraHeartRatePage,
+} from "@/lib/oura/sync-heart-rate";
+
+import { makeUser } from "../kit/factories";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const prisma = getPrismaClient();
+const DAY_MS = 86_400_000;
+const MIN_MS = 60_000;
+
+beforeEach(async () => {
+  await truncateAllTables(prisma);
+});
+
+/** Noon UTC, `days` ago: every sample below stays inside that UTC day. */
+function dayStart(days: number): number {
+  const now = Date.now();
+  return Math.floor(now / DAY_MS) * DAY_MS - days * DAY_MS;
+}
+
+/** Samples of 10:00-11:59 UTC, one every 10 minutes, 10:30 included. */
+function samples(base: number): Array<{ at: Date; bpm: number }> {
+  const out: Array<{ at: Date; bpm: number }> = [];
+  for (let m = 0; m < 120; m += 10) {
+    out.push({
+      at: new Date(base + 10 * 60 * MIN_MS + m * MIN_MS),
+      bpm: 50 + ((m * 7) % 45),
+    });
+  }
+  return out;
+}
+
+const gh = (s: { at: Date; bpm: number }) => ({
+  type: "PULSE",
+  value: s.bpm,
+  unit: "bpm",
+  measuredAt: s.at,
+  externalId: `${s.at.getTime()}:hr`,
+});
+
+async function liveRows(userId: string, source: "GOOGLE_HEALTH" | "OURA") {
+  return prisma.measurement.findMany({
+    where: { userId, source, deletedAt: null },
+    select: { type: true, value: true, externalId: true, measuredAt: true },
+    orderBy: [{ type: "asc" }, { measuredAt: "asc" }],
+  });
+}
+
+describe("dense fold of provider heart rate (#1297)", () => {
+  it("folds old Google Health heart rate to hourly means, and a re-sync writes none of it back", async () => {
+    const user = await makeUser({ timezone: "UTC" });
+    const old = samples(dayStart(120));
+    const recent = samples(dayStart(5));
+    await upsertGoogleHealthMeasurements(user.id, old.map(gh));
+    // A nightly SpO2 average: one value a day, never folded.
+    await prisma.measurement.create({
+      data: {
+        userId: user.id,
+        type: "OXYGEN_SATURATION",
+        value: 96,
+        unit: "%",
+        source: "GOOGLE_HEALTH",
+        measuredAt: new Date(dayStart(120) + 7 * 3_600_000),
+        externalId: "2026-01-01:spo2",
+      },
+    });
+    const dayBefore = await prisma.measurementRollup.findFirstOrThrow({
+      where: { userId: user.id, type: "PULSE", granularity: "DAY" },
+    });
+
+    const summary = await runDenseIntradayRetention(prisma, {
+      userId: user.id,
+      log: () => {},
+    });
+    expect(summary.totals.perSampleRowsSoftDeleted).toBeGreaterThan(0);
+
+    const folded = await liveRows(user.id, "GOOGLE_HEALTH");
+    const pulse = folded.filter((r) => r.type === "PULSE");
+    // Two hours, two hourly means; no raw sample of them is left.
+    expect(pulse).toHaveLength(2);
+    for (const row of pulse) {
+      expect(row.externalId).toMatch(
+        /^stats:HKQuantityTypeIdentifierHeartRate:\d{4}-\d{2}-\d{2}T1[01]$/,
+      );
+    }
+    const firstHour = old.slice(0, 6).map((s) => s.bpm);
+    expect(pulse[0].value).toBeCloseTo(
+      firstHour.reduce((a, b) => a + b, 0) / firstHour.length,
+      6,
+    );
+    expect(folded.filter((r) => r.type === "OXYGEN_SATURATION")).toHaveLength(
+      1,
+    );
+    // The DAY bucket keeps the raw figures.
+    const dayAfter = await prisma.measurementRollup.findFirstOrThrow({
+      where: { userId: user.id, type: "PULSE", granularity: "DAY" },
+    });
+    expect(dayAfter.count).toBe(dayBefore.count);
+    expect(dayAfter.minValue).toBe(dayBefore.minValue);
+    expect(dayAfter.maxValue).toBe(dayBefore.maxValue);
+    expect(dayAfter.mean).toBeCloseTo(dayBefore.mean, 9);
+
+    // A backfill walks the same history again.
+    const again = await upsertGoogleHealthMeasurements(
+      user.id,
+      [...old, ...recent].map(gh),
+    );
+    expect(again.imported).toBe(recent.length);
+    const afterResync = await liveRows(user.id, "GOOGLE_HEALTH");
+    expect(afterResync.filter((r) => r.type === "PULSE")).toHaveLength(
+      2 + recent.length,
+    );
+    // The hourly means are untouched, the 10:30 one included.
+    expect(
+      afterResync.filter((r) => r.externalId?.startsWith("stats:")),
+    ).toEqual(pulse);
+  });
+
+  it("folds old Oura heart rate the same way, and a re-read writes none of it back", async () => {
+    const user = await makeUser({ timezone: "UTC" });
+    const old = samples(dayStart(150)).map((s) => ({
+      bpm: s.bpm,
+      timestamp: s.at.toISOString(),
+      source: "awake",
+    }));
+    expect(
+      await writeOuraHeartRatePage(user.id, old, new TouchedDays(), {
+        emitArrivals: false,
+      }),
+    ).toBe(old.length);
+
+    await runDenseIntradayRetention(prisma, {
+      userId: user.id,
+      log: () => {},
+    });
+    const folded = await liveRows(user.id, "OURA");
+    expect(folded).toHaveLength(2);
+    expect(folded.every((r) => r.externalId?.startsWith("stats:"))).toBe(true);
+
+    expect(
+      await writeOuraHeartRatePage(user.id, old, new TouchedDays(), {
+        emitArrivals: false,
+      }),
+    ).toBe(0);
+    expect(await liveRows(user.id, "OURA")).toEqual(folded);
+  });
+
+  it("leaves heart rate inside the window raw", async () => {
+    const user = await makeUser({ timezone: "UTC" });
+    const recent = samples(dayStart(10));
+    await upsertGoogleHealthMeasurements(user.id, recent.map(gh));
+    await runDenseIntradayRetention(prisma, {
+      userId: user.id,
+      log: () => {},
+    });
+    expect(await liveRows(user.id, "GOOGLE_HEALTH")).toHaveLength(
+      recent.length,
+    );
+  });
+});