diff --git a/src/__tests__/baselines/source-file-lines.json b/src/__tests__/baselines/source-file-lines.json index d86b28d50..31b681f6f 100644 --- a/src/__tests__/baselines/source-file-lines.json +++ b/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": 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, diff --git a/src/lib/google-health/mapping.md b/src/lib/google-health/mapping.md index eb958bf6b..0d92bef67 100644 --- a/src/lib/google-health/mapping.md +++ b/src/lib/google-health/mapping.md @@ -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. | diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index 44f200e28..7553f1268 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -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 && diff --git a/src/lib/jobs/dense-intraday-hourly-rebuild.ts b/src/lib/jobs/dense-intraday-hourly-rebuild.ts index e5a8c0f8a..b58219802 100644 --- a/src/lib/jobs/dense-intraday-hourly-rebuild.ts +++ b/src/lib/jobs/dense-intraday-hourly-rebuild.ts @@ -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"; diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index 27b7322ae..2ced5d3f5 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/src/lib/jobs/dense-intraday-retention.ts @@ -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 }; diff --git a/src/lib/jobs/measurement-fold-repair.ts b/src/lib/jobs/measurement-fold-repair.ts index c9c42e3b0..8a8738e41 100644 --- a/src/lib/jobs/measurement-fold-repair.ts +++ b/src/lib/jobs/measurement-fold-repair.ts @@ -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, diff --git a/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts b/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts index fb0311aa7..5878092f0 100644 --- a/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-drain-exemption.test.ts @@ -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", () => { diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts index 2d76f13ae..6c7c88ba3 100644 --- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts @@ -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", ]); }); }); diff --git a/src/lib/measurements/__tests__/folded-window.test.ts b/src/lib/measurements/__tests__/folded-window.test.ts index d83a3d464..4b36a5b98 100644 --- a/src/lib/measurements/__tests__/folded-window.test.ts +++ b/src/lib/measurements/__tests__/folded-window.test.ts @@ -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[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[0], + "u1", + "OURA", + rows, + { now: NOW }, + ); + expect(kept).toEqual(rows); + expect(findMany).not.toHaveBeenCalled(); + expect(findUnique).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/measurements/dense-intraday-hourly-rebuild.ts b/src/lib/measurements/dense-intraday-hourly-rebuild.ts index 0c49f4c61..6f34f3d0a 100644 --- a/src/lib/measurements/dense-intraday-hourly-rebuild.ts +++ b/src/lib/measurements/dense-intraday-hourly-rebuild.ts @@ -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, diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index fd7a1babc..89abd6362 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -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 = - new Set([ - "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 = */ 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({ 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); diff --git a/src/lib/measurements/dense-intraday-sources.ts b/src/lib/measurements/dense-intraday-sources.ts new file mode 100644 index 000000000..2db64daa9 --- /dev/null +++ b/src/lib/measurements/dense-intraday-sources.ts @@ -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 = + new Set([ + "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(["PULSE"]), + OURA: new Set(["PULSE"]), +} as const satisfies Partial< + Record> +>; + +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> + > + )[source]; + return types?.has(type) ?? false; +} diff --git a/src/lib/measurements/folded-window.ts b/src/lib/measurements/folded-window.ts index 9fa41a6e1..eb84dfb2c 100644 --- a/src/lib/measurements/folded-window.ts +++ b/src/lib/measurements/folded-window.ts @@ -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; + foldedAfterMs: number; + reachMs: number; +} function ruleFor( row: Pick, ): 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; diff --git a/src/lib/oura/sync-heart-rate.ts b/src/lib/oura/sync-heart-rate.ts index 9c0a9a1f4..f2b8ff4b0 100644 --- a/src/lib/oura/sync-heart-rate.ts +++ b/src/lib/oura/sync-heart-rate.ts @@ -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:`), 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); diff --git a/tests/integration/provider-dense-fold.test.ts b/tests/integration/provider-dense-fold.test.ts new file mode 100644 index 000000000..d36c21a52 --- /dev/null +++ b/tests/integration/provider-dense-fold.test.ts @@ -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, + ); + }); +});