From e34741c077ca9f824139c597f027b98fa65976f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 27 Sep 2026 20:42:54 +0200 Subject: [PATCH 1/5] fix(google-health): write a dense collection page by page during a backfill A full-history backfill read every heart-rate point of an account into one array, mapped it into a second, and wrote the lot in one call; the cycle's deferred-rollup ledger then kept one entry per written reading. On a watch that records a reading a minute that is over a million points, and the backfill ran a 1 GB heap out of memory within seconds, restarting the instance it shares with the app (#1023). The metric walk now hands each page to its write before requesting the next, and both the Google Health and Fitbit ledgers collapse to one key per touched type-day as they record. The heart-rate walk's page ceiling rises from 1000 to 10000: at one point a minute the old ceiling is under two years, a longer history always ended truncated, and the backfill retried the whole walk on every attempt and every boot without completing. --- src/lib/fitbit/sync-core.ts | 19 +- src/lib/fitbit/sync.ts | 11 +- .../sync-deferred-rollup-ledger.test.ts | 97 +++++++ .../sync-metrics-dense-pages.test.ts | 111 ++++++++ .../__tests__/sync-metrics-failsoft.test.ts | 19 ++ src/lib/google-health/client.ts | 60 ++++- src/lib/google-health/sync-core.ts | 26 +- src/lib/google-health/sync-metrics.ts | 75 +++--- ...oogle-health-dense-backfill-memory.test.ts | 247 ++++++++++++++++++ 9 files changed, 598 insertions(+), 67 deletions(-) create mode 100644 src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts create mode 100644 src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts create mode 100644 tests/integration/google-health-dense-backfill-memory.test.ts diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index 133c7fdf2..d25ac2487 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/src/lib/fitbit/sync-core.ts @@ -228,7 +228,11 @@ export const softSkipStorage = new AsyncLocalStorage(); * the soft-skip tracker. */ export interface RollupDeferTracker { - keys: Array<{ type: MeasurementType; measuredAt: Date }>; + /** + * One entry per touched `(type, UTC day)`, never one per reading, so the + * ledger stays the size of the history in days rather than in samples. + */ + keys: Map; } export const rollupDeferStorage = new AsyncLocalStorage(); @@ -877,14 +881,13 @@ export async function upsertFitbitMeasurements( // path keeps the inline per-day hook here (small touched set, warm next read). if (opts.deferRollup) { const tracker = rollupDeferStorage.getStore(); - // Plain loop, never `push(...touched)`: `touched` carries ONE entry PER - // READING, and a full-history batch on a sample-dense account (a - // multi-year heart-rate series is six figures of rows) blows the call - // stack when spread into a single call — the engine caps argument counts. - // The RangeError aborted the whole metrics collection on exactly the - // accounts with the most data. + // Collapsed to type-days on the way in: `touched` carries ONE entry PER + // READING and the tracker lives for the whole cycle, so keeping the raw + // entries made the ledger grow with the sample count of the history. if (tracker) { - for (const t of touched) tracker.keys.push(t); + for (const k of collapseToTypeDayKeys(touched)) { + tracker.keys.set(`${k.type}|${k.measuredAt.getTime()}`, k); + } } return { imported, touched, inserted: insertedRows }; } diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index 9195f7bcb..449efbff2 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -5,10 +5,7 @@ import { prisma } from "@/lib/db"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status"; import { annotate, getEvent } from "@/lib/logging/context"; -import { - collapseToTypeDayKeys, - recomputeUserRollups, -} from "@/lib/rollups/measurement-rollups"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; import { syncUserActivity } from "./sync-activity"; import { hardFailStorage, @@ -90,7 +87,7 @@ export async function syncUserFitbit( ]; const tracker: SoftSkipTracker = { count: 0 }; - const deferTracker: RollupDeferTracker = { keys: [] }; + const deferTracker: RollupDeferTracker = { keys: new Map() }; const hardFailTracker: HardFailTracker = { failures: [] }; let total = 0; let anyFailed = false; @@ -120,9 +117,9 @@ export async function syncUserFitbit( // touched days. One pass replaces the thousands of per-(type,day) round-trips a // deep backfill would otherwise pay (each = an aggregate SELECT + rollup upsert // + 3 queue sends). Best-effort: a populator hiccup never fails the backfill. - if (opts.fullSync && deferTracker.keys.length > 0) { + if (opts.fullSync && deferTracker.keys.size > 0) { try { - const days = collapseToTypeDayKeys(deferTracker.keys); + const days = Array.from(deferTracker.keys.values()); const types = Array.from(new Set(days.map((k) => k.type))); const sorted = days .map((k) => k.measuredAt.getTime()) diff --git a/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts b/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts new file mode 100644 index 000000000..cea735f92 --- /dev/null +++ b/src/lib/google-health/__tests__/sync-deferred-rollup-ledger.test.ts @@ -0,0 +1,97 @@ +/** + * #1023 — the deferred-rollup ledger of a backfill cycle holds type-days, not + * readings. + * + * On a full-history backfill every write defers its rollup fold to one pass at + * the end of the cycle, and the cycle-wide ledger used to keep one entry per + * written reading. A per-minute heart-rate history of two and a half years is + * 1.3 M readings over about 900 days, so the ledger alone grew with the stream + * for the whole walk. It now keeps one entry per touched `(type, UTC day)`. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { findManyMock, createManyMock } = vi.hoisted(() => ({ + findManyMock: vi.fn(async () => [] as unknown[]), + createManyMock: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + measurement: { + findMany: findManyMock, + createManyAndReturn: createManyMock, + update: vi.fn(), + }, + }, +})); +vi.mock("@/lib/crypto", () => ({ encrypt: vi.fn(), decrypt: vi.fn() })); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: vi.fn(async () => false), + recordSyncFailure: vi.fn(async () => {}), + recordSyncSuccess: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); +vi.mock("@/lib/arrivals/measurement-emit", () => ({ + emitInsertedMeasurementArrivals: vi.fn(async () => {}), +})); +vi.mock("../credentials", () => ({ + getUserGoogleHealthCredentials: vi.fn(async () => null), +})); + +import { + runWithGoogleHealthSyncCycle, + upsertGoogleHealthMeasurements, +} from "../sync-core"; + +const DAYS = 3; +const PER_DAY = 1440; +const START = Date.parse("2026-09-01T00:00:00.000Z"); + +beforeEach(() => { + findManyMock.mockReset().mockResolvedValue([]); + createManyMock + .mockReset() + .mockImplementation( + async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((d, i) => ({ + id: `row-${i}`, + type: d.type, + measuredAt: d.measuredAt, + })), + ); +}); + +describe("deferred rollup ledger on a dense backfill", () => { + it("keeps one key per touched type-day however many readings land", async () => { + const cycle = await runWithGoogleHealthSyncCycle(async () => { + for (let day = 0; day < DAYS; day++) { + const readings = []; + for (let m = 0; m < PER_DAY; m++) { + const at = new Date(START + day * 86_400_000 + m * 60_000); + readings.push({ + type: "PULSE", + value: 60, + unit: "bpm", + measuredAt: at, + externalId: `${at.toISOString()}:hr`, + }); + } + await upsertGoogleHealthMeasurements("dense-user", readings, { + deferRollup: true, + }); + } + return null; + }); + + expect(cycle.deferredRollupKeys).toHaveLength(DAYS); + expect( + cycle.deferredRollupKeys.map((k) => k.measuredAt.toISOString()).sort(), + ).toEqual([ + "2026-09-01T00:00:00.000Z", + "2026-09-02T00:00:00.000Z", + "2026-09-03T00:00:00.000Z", + ]); + }); +}); diff --git a/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts new file mode 100644 index 000000000..22ea1ad96 --- /dev/null +++ b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts @@ -0,0 +1,111 @@ +/** + * #1023 — a dense heart-rate collection is written page by page. + * + * Intraday heart rate is one point a minute. `syncUserMetrics` used to collect + * the whole collection, map all of it, and hand every reading to one upsert, + * so a full-history backfill held a million raw points and a million readings + * at once and ran a 1 GB heap out of memory (the reproduction on 1.3 M + * readings dies within seconds of the walk). The contract pinned here: every + * upsert carries at most one page, and a page is written before the next one + * is requested, so nothing but the current page is ever resident. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { safeFetchMock, upsertMock, events } = vi.hoisted(() => ({ + safeFetchMock: vi.fn(), + upsertMock: vi.fn(), + events: [] as string[], +})); + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, safeFetch: safeFetchMock }; +}); + +vi.mock("../sync-core", () => ({ + getValidToken: vi.fn(async () => ({ accessToken: "token" })), + handleCollectionFetchError: vi.fn(async () => 0), + upsertGoogleHealthMeasurements: upsertMock, +})); + +vi.mock("@/lib/db", () => ({ + prisma: { user: { findUnique: vi.fn(), update: vi.fn() } }, +})); + +vi.mock("@/lib/tz/resolver", () => ({ + resolveUserTimezone: vi.fn(async () => "UTC"), +})); + +import { GOOGLE_HEALTH_PAGE_SIZE } from "../client"; +import { syncUserMetrics } from "../sync-metrics"; + +const PAGES = 3; +const NOW = Date.parse("2026-09-27T17:00:00.000Z"); + +function heartRatePage(page: number): unknown { + const dataPoints = []; + for (let i = 0; i < GOOGLE_HEALTH_PAGE_SIZE; i++) { + const g = page * GOOGLE_HEALTH_PAGE_SIZE + i; + dataPoints.push({ + heartRate: { + beatsPerMinute: String(55 + (g % 50)), + sampleTime: { physicalTime: new Date(NOW - g * 60_000).toISOString() }, + }, + }); + } + return { + dataPoints, + ...(page + 1 < PAGES ? { nextPageToken: String(page + 1) } : {}), + }; +} + +beforeEach(() => { + events.length = 0; + upsertMock.mockReset().mockImplementation(async (_u, readings: unknown[]) => { + events.push(`upsert:${readings.length}`); + return { imported: readings.length, touched: [], inserted: [] }; + }); + safeFetchMock.mockReset().mockImplementation(async (url: string) => { + const u = new URL(url); + let body: unknown = {}; + if (u.pathname.endsWith("/dataTypes/heart-rate/dataPoints")) { + const page = Number(u.searchParams.get("pageToken") ?? 0); + events.push(`fetch:${page}`); + body = heartRatePage(page); + } + return { + status: 200, + ok: true, + json: async () => body, + } as unknown as Response; + }); +}); + +describe("syncUserMetrics over a dense heart-rate collection", () => { + it("writes each page before the next one is requested", async () => { + const imported = await syncUserMetrics("dense-user", { + deferRollup: true, + }); + + expect(imported).toBe(PAGES * GOOGLE_HEALTH_PAGE_SIZE); + const pulseWrites = upsertMock.mock.calls + .map((c) => c[1] as Array<{ type: string }>) + .filter((readings) => readings.some((r) => r.type === "PULSE")); + expect(pulseWrites).toHaveLength(PAGES); + for (const readings of pulseWrites) { + expect(readings.length).toBeLessThanOrEqual(GOOGLE_HEALTH_PAGE_SIZE); + } + const hr = events.filter( + (e) => + e.startsWith("fetch:") || e === `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + ); + expect(hr).toEqual([ + "fetch:0", + `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + "fetch:1", + `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + "fetch:2", + `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, + ]); + }); +}); diff --git a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts index c9cfe6a94..8d48929df 100644 --- a/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts +++ b/src/lib/google-health/__tests__/sync-metrics-failsoft.test.ts @@ -12,6 +12,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const { fetchDataPointsMock, handleErrorMock, mapWeightMock } = vi.hoisted( () => ({ + // Stands in for the page walk: every metric type is streamed through + // `forEachDataPointPage`, which hands its pages to the callback. fetchDataPointsMock: vi.fn(), handleErrorMock: vi.fn(async () => 0), mapWeightMock: vi.fn(() => { @@ -25,6 +27,23 @@ vi.mock("../client", async (importOriginal) => { return { ...actual, fetchDataPoints: fetchDataPointsMock, + forEachDataPointPage: vi.fn( + async ( + dataType: unknown, + token: string, + verb: string, + query: unknown, + onPage: (points: unknown[]) => Promise, + ) => { + const points = (await fetchDataPointsMock( + dataType, + token, + verb, + query, + )) as unknown[]; + if (points.length > 0) await onPage(points); + }, + ), mapWeight: mapWeightMock, }; }); diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index 0a21e75fc..b38e95a1b 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -512,6 +512,20 @@ export const GOOGLE_HEALTH_PAGE_SIZE = 1000; /** Sleep/exercise read cap — matches the Google Health 25-cap for those types. */ export const GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE = 25; +/** + * Page ceiling for the intraday heart-rate walk. + * + * The default of 1000 pages is a million points, which a watch that records + * a reading a minute passes in under two years. A walk that hits the ceiling + * reports `truncated`, the backfill treats that as incomplete and throws, and + * pg-boss retries it — so an account with a longer history re-ran the whole + * walk on every retry and every boot and never completed. Ten thousand pages + * is nineteen years at that rate: still a bound on a runaway cursor, no longer + * a bound a real history reaches. The walk is paged (`forEachDataPointPage`), + * so the ceiling costs time, not memory. + */ +export const GOOGLE_HEALTH_DENSE_MAX_PAGES = 10_000; + /** `dataPoints.list` envelope: `{ dataPoints, nextPageToken }`. */ interface GoogleHealthDataPointPage { dataPoints?: GoogleHealthDataPoint[]; @@ -715,6 +729,35 @@ export async function fetchDataPoints( verb: string, query: DataPointQuery = {}, ): Promise { + const points: GoogleHealthDataPoint[] = []; + await forEachDataPointPage(dataType, accessToken, verb, query, (page) => { + for (const p of page) points.push(p); + }); + return points; +} + +/** + * The same walk as `fetchDataPoints`, handing each page to `onPage` as it + * arrives instead of collecting the whole collection first. + * + * A dense collection has to be read this way. Intraday heart rate is one + * point a minute, so a full-history backfill of two years is over a million + * points; held as parsed JSON and then again as mapped readings, that is more + * than a 1 GB heap, and the process that runs the backfill also serves the + * app. Page by page, only one page is ever resident. + * + * `onPage` is awaited before the next page is requested, so a slow write + * applies back-pressure to the fetch. The date-filter fallback still works: + * it only fires when the first request is rejected, which is before any page + * reaches `onPage`. + */ +export async function forEachDataPointPage( + dataType: GoogleHealthDataType, + accessToken: string, + verb: string, + query: DataPointQuery, + onPage: (points: GoogleHealthDataPoint[]) => void | Promise, +): Promise { const maxPages = query.maxPages ?? 1000; const pageSize = query.pageSize ?? GOOGLE_HEALTH_PAGE_SIZE; @@ -722,8 +765,8 @@ export async function fetchDataPoints( const walk = async ( dateStyle: GoogleHealthDateFilterStyle, - ): Promise => { - const points: GoogleHealthDataPoint[] = []; + ): Promise => { + let fetched = 0; let pageToken: string | null | undefined; let pageCount = 0; @@ -780,13 +823,14 @@ export async function fetchDataPoints( }); } - for (const p of json?.dataPoints ?? []) points.push(p); + const page = json?.dataPoints ?? []; pageToken = json?.nextPageToken ?? null; pageCount += 1; + fetched += page.length; + await onPage(page); } while (pageToken && pageCount < maxPages); - notePagination(pageCount, points.length, Boolean(pageToken)); - return points; + notePagination(pageCount, fetched, Boolean(pageToken)); }; // The fallback only exists for filtered daily-summary reads — every other @@ -795,9 +839,8 @@ export async function fetchDataPoints( dataType.timeField === "date" && query.start !== undefined; let style: GoogleHealthDateFilterStyle = "camel"; - let points: GoogleHealthDataPoint[]; try { - points = await walk("camel"); + await walk("camel"); } catch (err) { const firstRequestRejected = canFallBack && @@ -806,13 +849,12 @@ export async function fetchDataPoints( err.httpStatus === 400; if (!firstRequestRejected) throw err; style = "snake"; - points = await walk("snake"); + await walk("snake"); } if (canFallBack) { annotate({ meta: { "googleHealth.dateFilter.style": style } }); query.onDateFilterStyle?.(style); } - return points; } // ─── Daily roll-up reads (`POST …/dataPoints:dailyRollUp`) ───────────────── diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index bef34d971..3b478e8d4 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -265,7 +265,12 @@ export async function runWithGoogleHealthHardFailLedger( * AsyncLocalStorage so concurrent per-user jobs never cross-pollinate. */ interface RollupDeferTracker { - keys: Array<{ type: MeasurementType; measuredAt: Date }>; + /** + * One entry per touched `(type, UTC day)`, never one per reading. A + * full-history heart-rate backfill touches over a million readings but only + * a few hundred days; a per-reading list grew with the stream. + */ + keys: Map; } const rollupDeferStorage = new AsyncLocalStorage(); @@ -286,7 +291,7 @@ export async function runWithGoogleHealthSyncCycle( ): Promise> { const softSkips: SoftSkipTracker = { count: 0 }; const hardFailures: HardFailTracker = { failures: [] }; - const deferredRollups: RollupDeferTracker = { keys: [] }; + const deferredRollups: RollupDeferTracker = { keys: new Map() }; const result = await softSkipStorage.run(softSkips, () => hardFailStorage.run(hardFailures, () => rollupDeferStorage.run(deferredRollups, fn), @@ -296,7 +301,7 @@ export async function runWithGoogleHealthSyncCycle( result, softSkipCount: softSkips.count, hardFailures: hardFailures.failures, - deferredRollupKeys: deferredRollups.keys, + deferredRollupKeys: Array.from(deferredRollups.keys.values()), }; } @@ -831,14 +836,15 @@ export async function upsertGoogleHealthMeasurements( // of the cycle. The incremental path keeps the inline per-day hook here. if (opts.deferRollup) { const tracker = rollupDeferStorage.getStore(); - // Plain loop, never `push(...touched)`: `touched` carries ONE entry PER - // READING, and a full-history batch on a sample-dense account (a - // multi-year heart-rate series is six figures of rows) blows the call - // stack when spread into a single call — the engine caps argument counts. - // The RangeError aborted the whole metrics collection on exactly the - // accounts with the most data. + // Collapsed to type-days on the way in: `touched` carries ONE entry PER + // READING, and the tracker lives for the whole cycle, so keeping the raw + // entries made the ledger grow with the stream (a million-reading + // heart-rate backfill held a million of them to describe a few hundred + // days). if (tracker) { - for (const t of touched) tracker.keys.push(t); + for (const k of collapseToTypeDayKeys(touched)) { + tracker.keys.set(`${k.type}|${k.measuredAt.getTime()}`, k); + } } noteGoogleHealthWritten(imported); return { imported, touched, inserted: insertedRows }; diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts index 6c204c7ab..d253ed437 100644 --- a/src/lib/google-health/sync-metrics.ts +++ b/src/lib/google-health/sync-metrics.ts @@ -27,9 +27,11 @@ */ import { GOOGLE_HEALTH_DATA_TYPES, + GOOGLE_HEALTH_DENSE_MAX_PAGES, type GoogleHealthDataType, type GoogleHealthMappedMeasurement, fetchDataPoints, + forEachDataPointPage, mapBloodGlucose, mapBodyFat, mapCoreBodyTemperature, @@ -61,6 +63,8 @@ interface MetricResource { tz: string, ) => GoogleHealthMappedMeasurement[]; verb: string; + /** Page ceiling for the walk; the client default when omitted. */ + maxPages?: number; } /** The launch metric resources (Measurement-producing). Height handled separately. */ @@ -99,6 +103,7 @@ const METRIC_RESOURCES: MetricResource[] = [ dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate, map: mapHeartRate, verb: "fetchHeartRate", + maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES, }, { dataType: GOOGLE_HEALTH_DATA_TYPES.bloodGlucose, @@ -139,48 +144,52 @@ export async function syncUserMetrics( // Each metric data type is fetched + mapped independently so a per-class 403 // soft-skips only that class. + // + // Every type is written page by page as it is read. Intraday heart rate is + // one point a minute, so a full-history backfill is over a million points; + // collecting the whole collection first and mapping it second held the raw + // points and the readings at once and ran a 1 GB heap out of memory on + // exactly the accounts with the most history. A page is at most + // `GOOGLE_HEALTH_PAGE_SIZE` points, and nothing from it outlives its write. + // + // The mapper runs inside the same per-type catch: a single malformed point + // whose `resource.map(point)` throws must not escape the METRIC_RESOURCES + // loop and skip every metric type ordered after it (which also blocked the + // watermark, so the bad point refetched hourly and those types stayed dead). + // A map throw goes through the same ledger as a fetch failure — record it, + // fail the cycle, and move on to the next type. Pages already written before + // the failure stay written: every write is keyed, so the retry that the + // failed verdict causes overwrites them in place. for (const resource of METRIC_RESOURCES) { - let points: Record[]; try { - points = await fetchDataPoints( + await forEachDataPointPage( resource.dataType, tokenInfo.accessToken, resource.verb, - { start }, + { start, maxPages: resource.maxPages }, + async (points) => { + const readings: GoogleHealthMeasurementUpsert[] = []; + for (const point of points) { + for (const m of resource.map(point, tz)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.fieldTag, + }); + } + } + imported += ( + await upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + }, ); } catch (err) { imported += await handleCollectionFetchError(resource.verb, userId, err); - continue; } - - // The mapper runs INSIDE a per-type catch too: a single malformed point - // whose `resource.map(point)` throws must not escape the METRIC_RESOURCES - // loop and skip every metric type ordered after it (which also blocked the - // watermark, so the bad point refetched hourly and those types stayed dead). - // Route a map throw through the same ledger as a fetch failure — record it, - // fail the cycle, and move on to the next type. - const readings: GoogleHealthMeasurementUpsert[] = []; - try { - for (const point of points) { - for (const m of resource.map(point, tz)) { - readings.push({ - type: m.type, - value: m.value, - unit: m.unit, - measuredAt: m.measuredAt, - externalId: m.fieldTag, - }); - } - } - } catch (err) { - imported += await handleCollectionFetchError(resource.verb, userId, err); - continue; - } - imported += ( - await upsertGoogleHealthMeasurements(userId, readings, { - deferRollup: opts.deferRollup, - }) - ).imported; } // Height → User.heightCm, only when the user has no height yet. Never mint a diff --git a/tests/integration/google-health-dense-backfill-memory.test.ts b/tests/integration/google-health-dense-backfill-memory.test.ts new file mode 100644 index 000000000..31414108e --- /dev/null +++ b/tests/integration/google-health-dense-backfill-memory.test.ts @@ -0,0 +1,247 @@ +/** + * #1023 — a full-history Google Health backfill over a dense heart-rate + * stream, with its live memory measured rather than assumed. + * + * What this pins. The backfill read every heart-rate point of the account's + * history into one array, mapped all of it into a second array, and wrote the + * lot in one call, so the raw points, the readings and the write's working + * sets were resident together and all of them scaled with the history. A + * watch that records a reading a minute passes a million points in two years: + * reproduced on 1.3 M readings under `--max-old-space-size=1024`, the backfill + * reaches `FATAL ERROR: Reached heap limit` within seconds, and because the + * backfill lane shares the app process, the whole instance restarts. The walk + * now writes each page as it arrives, and the cycle's deferred-rollup ledger + * keeps one key per touched day instead of one per reading. + * + * How it measures. Every heap reading is taken after a forced collection, so + * what is compared is what the code HOLDS, not how much garbage V8 chose to + * leave lying about. And what is compared is GROWTH: the heap once a quarter + * of the pages have been read, against the most it reaches afterwards. The + * first pages carry a fixed cost that has nothing to do with the stream + * (modules loaded on first use, the ORM's compiled queries), and that cost is + * about the size of this fixture, so an absolute figure could not tell a + * streaming walk from one that holds everything. Growth can: a walk that + * holds one page stays flat, a walk that holds what it has read climbs with + * every page. + * + * The two halves are the point: the backfill stays inside the budget, and + * collecting the same collection into one array (what the backfill used to do + * before anything else) does not. Without the second half the budget could + * be any number and the first would still pass. + * + * Google's API is replaced at the `safeFetch` seam by a generator that builds + * each page on request, so the fixture itself never sits in memory. + */ +import v8 from "node:v8"; +import vm from "node:vm"; + +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const USER = "gh-dense-backfill-owner"; +const POINTS = 300_000; +const NOW = Date.parse("2026-09-27T17:00:00.000Z"); + +/** + * How much the heap may grow between the first quarter of the walk and its + * end. Measured, not chosen: the paged walk grows by a few MB on this fixture + * (noise around a flat line), and collecting the same pages into one array + * grows by about 80 MB over the same stretch. + */ +const GROWTH_BUDGET_BYTES = 24 * 1024 * 1024; + +const fake = vi.hoisted(() => ({ + total: 0, + now: 0, + heartRatePages: 0, + /** Called as each heart-rate page is requested (the sampler's hook). */ + onPage: null as (() => void) | null, +})); + +vi.mock("@/lib/safe-fetch", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + // A plain function, not `vi.fn`: a mock records every call's result, and + // here each result is a whole page's response, so the recorder alone + // would hold the collection this file exists to keep out of memory. + safeFetch: async (url: string) => { + const u = new URL(url); + let body: unknown = {}; + if (u.pathname.endsWith("/dataTypes/heart-rate/dataPoints")) { + const pageSize = Number(u.searchParams.get("pageSize") ?? 1000); + const page = Number(u.searchParams.get("pageToken") ?? 0); + const end = Math.min(fake.total, (page + 1) * pageSize); + const dataPoints = []; + for (let g = page * pageSize; g < end; g++) { + dataPoints.push({ + name: `users/me/dataTypes/heart-rate/dataPoints/${9_000_000_000_000 + g}`, + dataSource: { + recordingMethod: "PASSIVE_MEASUREMENT", + device: { manufacturer: "Google", displayName: "Watch" }, + platform: "FITBIT", + }, + heartRate: { + beatsPerMinute: String(55 + (g % 50)), + sampleTime: { + physicalTime: new Date(fake.now - g * 60_000).toISOString(), + }, + }, + }); + } + fake.heartRatePages++; + fake.onPage?.(); + body = { + dataPoints, + ...(end < fake.total ? { nextPageToken: String(page + 1) } : {}), + }; + } + return new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }, + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const forceGc = ((): (() => void) => { + v8.setFlagsFromString("--expose-gc"); + const gc = vm.runInNewContext("gc") as () => void; + v8.setFlagsFromString("--no-expose-gc"); + return gc; +})(); + +function liveHeapBytes(): number { + forceGc(); + forceGc(); + return process.memoryUsage().heapUsed; +} + +const mb = (bytes: number): string => (bytes / 1024 / 1024).toFixed(1); + +function report(line: string): void { + process.stderr.write(`[gh-dense-backfill] ${line}\n`); +} + +/** + * Run `fn` while sampling the collected heap every few pages, from inside the + * page request, so each sample lands between two pages of the work being + * measured: exactly where anything it holds across a page is visible. (A + * timer is not enough; a walk whose pages resolve without touching the event + * loop never lets it fire.) Returns how far the heap rose after the walk + * passed a quarter of its pages. + */ +async function growthPastFirstQuarter( + fn: () => Promise, +): Promise<{ result: T; growth: number; samples: number }> { + const quarter = Math.ceil(POINTS / 1000 / 4); + const startPages = fake.heartRatePages; + let anchor: number | null = null; + let peak = 0; + let samples = 0; + const sample = (): void => { + const live = liveHeapBytes(); + if (anchor === null) { + if (fake.heartRatePages - startPages >= quarter) anchor = peak = live; + return; + } + samples++; + peak = Math.max(peak, live); + }; + fake.onPage = () => { + if ((fake.heartRatePages - startPages) % 5 === 0) sample(); + }; + try { + const result = await fn(); + sample(); + return { + result, + growth: anchor === null ? Number.NaN : peak - anchor, + samples, + }; + } finally { + fake.onPage = null; + } +} + +beforeAll(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + const { encrypt } = await import("@/lib/crypto"); + await prisma.user.create({ + data: { id: USER, username: USER, timezone: "Europe/Berlin" }, + }); + await prisma.googleHealthConnection.create({ + data: { + userId: USER, + googleUserId: `google-${USER}`, + accessToken: encrypt("access"), + refreshToken: encrypt("refresh"), + tokenExpiresAt: new Date(Date.now() + 86_400_000), + }, + }); + fake.total = POINTS; + fake.now = NOW; +}, 120_000); + +afterAll(async () => { + await truncateAllTables(getPrismaClient()); +}); + +describe("Google Health backfill over a dense heart-rate stream (#1023)", () => { + it("writes the whole history without growing with it", async () => { + const { runGoogleHealthBackfillForUser } = + await import("@/lib/jobs/google-health-backfill"); + + const { result, growth, samples } = await growthPastFirstQuarter(() => + runGoogleHealthBackfillForUser(USER), + ); + report( + `backfill: imported=${result.imported} growth=${mb(growth)} MB ` + + `over ${samples} samples (budget ${mb(GROWTH_BUDGET_BYTES)} MB)`, + ); + + const prisma = getPrismaClient(); + expect( + await prisma.measurement.count({ + where: { userId: USER, type: "PULSE", source: "GOOGLE_HEALTH" }, + }), + ).toBe(POINTS); + const connection = await prisma.googleHealthConnection.findUniqueOrThrow({ + where: { userId: USER }, + select: { backfillCompletedAt: true }, + }); + expect(connection.backfillCompletedAt).not.toBeNull(); + // The deferred fold reached the rollup tier for the whole span. + const days = await prisma.measurementRollup.count({ + where: { userId: USER, type: "PULSE", granularity: "DAY" }, + }); + expect(days).toBeGreaterThanOrEqual(Math.floor(POINTS / 1440)); + + expect(samples).toBeGreaterThan(5); + expect(growth).toBeLessThan(GROWTH_BUDGET_BYTES); + }, 900_000); + + it("the same pages collected into one array outgrow the budget", async () => { + const { fetchDataPoints, GOOGLE_HEALTH_DATA_TYPES } = + await import("@/lib/google-health/client"); + let held: unknown[] = []; + const { growth, samples } = await growthPastFirstQuarter(async () => { + held = await fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.heartRate, + "access", + "fetchHeartRate", + ); + }); + report( + `collected: points=${held.length} growth=${mb(growth)} MB over ${samples} samples`, + ); + expect(held).toHaveLength(POINTS); + expect(growth).toBeGreaterThan(GROWTH_BUDGET_BYTES); + }, 900_000); +}); From 821c2eb88409bf50ba1cac14accf487e9d92867c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 27 Sep 2026 20:43:15 +0200 Subject: [PATCH 2/5] test(google-health): pin a heart-rate walk past a thousand pages --- .../sync-metrics-dense-pages.test.ts | 32 ++++++++++++++++--- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts index 22ea1ad96..7eee64efc 100644 --- a/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts +++ b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts @@ -11,10 +11,12 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { safeFetchMock, upsertMock, events } = vi.hoisted(() => ({ +const { safeFetchMock, upsertMock, events, shape } = vi.hoisted(() => ({ safeFetchMock: vi.fn(), upsertMock: vi.fn(), events: [] as string[], + /** How the fake heart-rate collection is cut into pages. */ + shape: { pages: 3, perPage: 0 }, })); vi.mock("@/lib/safe-fetch", async (importOriginal) => { @@ -36,7 +38,10 @@ vi.mock("@/lib/tz/resolver", () => ({ resolveUserTimezone: vi.fn(async () => "UTC"), })); -import { GOOGLE_HEALTH_PAGE_SIZE } from "../client"; +import { + GOOGLE_HEALTH_PAGE_SIZE, + runWithGoogleHealthClientOutcome, +} from "../client"; import { syncUserMetrics } from "../sync-metrics"; const PAGES = 3; @@ -44,8 +49,8 @@ const NOW = Date.parse("2026-09-27T17:00:00.000Z"); function heartRatePage(page: number): unknown { const dataPoints = []; - for (let i = 0; i < GOOGLE_HEALTH_PAGE_SIZE; i++) { - const g = page * GOOGLE_HEALTH_PAGE_SIZE + i; + for (let i = 0; i < shape.perPage; i++) { + const g = page * shape.perPage + i; dataPoints.push({ heartRate: { beatsPerMinute: String(55 + (g % 50)), @@ -55,11 +60,13 @@ function heartRatePage(page: number): unknown { } return { dataPoints, - ...(page + 1 < PAGES ? { nextPageToken: String(page + 1) } : {}), + ...(page + 1 < shape.pages ? { nextPageToken: String(page + 1) } : {}), }; } beforeEach(() => { + shape.pages = PAGES; + shape.perPage = GOOGLE_HEALTH_PAGE_SIZE; events.length = 0; upsertMock.mockReset().mockImplementation(async (_u, readings: unknown[]) => { events.push(`upsert:${readings.length}`); @@ -108,4 +115,19 @@ describe("syncUserMetrics over a dense heart-rate collection", () => { `upsert:${GOOGLE_HEALTH_PAGE_SIZE}`, ]); }); + + it("walks a heart-rate history longer than a thousand pages to its end", async () => { + // A thousand pages is a million points: under two years of one reading a + // minute. The old ceiling ended every longer history truncated, which the + // backfill reads as incomplete, so it retried the whole walk forever. + shape.pages = 1_001; + shape.perPage = 1; + + const { outcome } = await runWithGoogleHealthClientOutcome(() => + syncUserMetrics("dense-user", { deferRollup: true }), + ); + + expect(events.filter((e) => e.startsWith("fetch:"))).toHaveLength(1_001); + expect(outcome.truncated).toBe(false); + }); }); From 0e991c3640d2b0da6121f69d1e89de028c306ee1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 27 Sep 2026 21:07:29 +0200 Subject: [PATCH 3/5] fix(measurements): fold a dense pulse stream inside the 90-day raw window Past 90 days every pulse reader already folded in SQL. Inside it, three still read every raw heart-rate row, which on a watch that records a reading a minute is 130 000 rows per read (#1023): - The insights page's resting-pulse proxy, used when an account has no resting heart rate, walked the whole window to take one percentile per day and held the event loop for over a second per view. It now folds in Postgres (`readRestingPulseProxy`) with the same percentile, rounding, minimum-sample rule and day anchor. - The per-kind series sent every reading: 14.7 MB for 90 days. A pulse window with more than 10 000 readings is now bucketed per local hour, mean plus low/high band, with stats still over the readings. - The doctor report drew every reading into one sparkline. A pulse window over the same cap now takes the per-day path longer reports already use. Sparse pulse (a cuff, hourly imports, a watch sampling every few minutes) stays under the cap and keeps its raw shape. --- docs/api/openapi.yaml | 9 +- .../comprehensive/__tests__/route.test.ts | 12 +- src/app/api/insights/comprehensive/route.ts | 35 +-- .../series/__tests__/route.test.ts | 69 ++++- src/app/api/measurements/series/route.ts | 44 +++- src/lib/analytics/resting-pulse-read.ts | 66 +++++ .../__tests__/leaf-gating-sweep.test.ts | 1 + src/lib/doctor-report/collect.ts | 26 +- src/lib/openapi/routes/measurements.ts | 6 +- .../dense-pulse-raw-window-reads.test.ts | 238 ++++++++++++++++++ 10 files changed, 474 insertions(+), 32 deletions(-) create mode 100644 src/lib/analytics/resting-pulse-read.ts create mode 100644 tests/integration/dense-pulse-raw-window-reads.test.ts diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 26f29e710..bde3f1cbb 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -40909,7 +40909,9 @@ components: properties: id: type: string - description: Measurement row id, or `sleep:` for a sleep night. + description: Measurement row id; `sleep:` for a sleep night; `day:` for a day bucket (`pulse` / + `glucose` beyond 90 days); `hour:` for an hour bucket (`pulse` within 90 days when the + window holds more than 10 000 readings). at: type: string format: date-time @@ -40982,7 +40984,10 @@ components: - stats additionalProperties: false description: "iOS-friendly per-kind time series: one point per reading (or per reconstructed night for `sleep`), an - explicit `unit` token, and a summary `stats` block." + explicit `unit` token, and a summary `stats` block. Sample-dense kinds are bucketed server-side: `pulse` and + `glucose` beyond 90 days per local day, and `pulse` within 90 days per local hour when the window holds more + than 10 000 readings (a bucket carries the mean as `value` and its range as `valueMin` / `valueMax`). `stats` is + always computed over the underlying readings, never over the buckets." ListMeasurementsResponse: type: object properties: diff --git a/src/app/api/insights/comprehensive/__tests__/route.test.ts b/src/app/api/insights/comprehensive/__tests__/route.test.ts index d6673001e..3ffce0401 100644 --- a/src/app/api/insights/comprehensive/__tests__/route.test.ts +++ b/src/app/api/insights/comprehensive/__tests__/route.test.ts @@ -48,6 +48,12 @@ vi.mock("@/lib/rollups/mood-rollups", async () => { vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); +// The raw-PULSE resting proxy folds in SQL; against the mocked client it is an +// account with no pulse. +vi.mock("@/lib/analytics/resting-pulse-read", () => ({ + readRestingPulseProxy: vi.fn(), +})); + vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn().mockResolvedValue(undefined), })); @@ -107,6 +113,7 @@ import { getSession } from "@/lib/auth/session"; import { requireModuleEnabled } from "@/lib/modules/gate"; import { aiCapabilityToServe } from "@/lib/ai/capabilities/gate"; import { probeProviderPresence } from "@/lib/ai/provider"; +import { readRestingPulseProxy } from "@/lib/analytics/resting-pulse-read"; import { prisma } from "@/lib/db"; import { buildComprehensiveAggregate } from "@/lib/insights/comprehensive-aggregator"; import { checkAnalyticsReadRateLimit } from "@/lib/rate-limit"; @@ -155,8 +162,9 @@ beforeEach(() => { dateOfBirth: new Date("1985-01-01"), }); (prisma.moodEntry.findMany as ReturnType).mockResolvedValue([]); - // The resting-pulse pairing (M-CS2) reads RESTING_HEART_RATE rows (and the - // raw-PULSE proxy only when none exist) straight from the measurement table. + vi.mocked(readRestingPulseProxy).mockResolvedValue([]); + // The resting-pulse pairing (M-CS2) reads RESTING_HEART_RATE rows straight + // from the measurement table (the raw-PULSE proxy is mocked above). (prisma.measurement.findMany as ReturnType).mockResolvedValue( [], ); diff --git a/src/app/api/insights/comprehensive/route.ts b/src/app/api/insights/comprehensive/route.ts index cc55cd882..cfe291dd4 100644 --- a/src/app/api/insights/comprehensive/route.ts +++ b/src/app/api/insights/comprehensive/route.ts @@ -24,6 +24,7 @@ import { } from "@/lib/analytics/compliance"; import { getMedicationCategories } from "@/lib/medication-category"; import { resolveRestingPulseSeries } from "@/lib/analytics/resting-pulse"; +import { readRestingPulseProxy } from "@/lib/analytics/resting-pulse-read"; import { apiHandler, requireRecordAuth, @@ -341,24 +342,24 @@ export async function buildComprehensiveResponse(user: AuthedUser) { }, select: { measuredAt: true, value: true }, }); - const rawPulseRowsForResting = - restingPulseRows.length === 0 - ? await prisma.measurement.findMany({ - where: { - userId, - type: "PULSE", - deletedAt: null, - measuredAt: { gte: ninetyDaysAgo }, - }, - select: { measuredAt: true, value: true }, - }) - : []; + // + // #1023 — the proxy folds in Postgres. It needs one percentile per day, and + // reading the raw samples to get it walked 130 000 rows per page view on a + // watch that records once a minute, blocking the event loop for over a + // second. `readRestingPulseProxy` returns the same series one row per day. const localDayKey = (d: Date) => userDayKey(d, userTz); - const { series: restingPulseSeries } = resolveRestingPulseSeries({ - restingSamples: restingPulseRows, - pulseSamples: rawPulseRowsForResting, - dayKeyOf: localDayKey, - }); + const restingPulseSeries = + restingPulseRows.length > 0 + ? resolveRestingPulseSeries({ + restingSamples: restingPulseRows, + pulseSamples: [], + dayKeyOf: localDayKey, + }).series + : await readRestingPulseProxy({ + userId, + since: ninetyDaysAgo, + timeZone: userTz, + }); const restingByDay = new Map(); for (const s of restingPulseSeries) { const key = localDayKey(s.measuredAt); diff --git a/src/app/api/measurements/series/__tests__/route.test.ts b/src/app/api/measurements/series/__tests__/route.test.ts index fbeaa46a8..de6f8dced 100644 --- a/src/app/api/measurements/series/__tests__/route.test.ts +++ b/src/app/api/measurements/series/__tests__/route.test.ts @@ -3,7 +3,7 @@ import { NextRequest } from "next/server"; vi.mock("@/lib/db", () => ({ prisma: { - measurement: { findMany: vi.fn() }, + measurement: { findMany: vi.fn(), count: vi.fn() }, user: { findUnique: vi.fn() }, auditLog: { create: vi.fn() }, // v1.28.25 — dense kinds (glucose / pulse) day-bucket in SQL for @@ -68,6 +68,7 @@ function req(query: string): NextRequest { beforeEach(() => { vi.resetAllMocks(); vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.count).mockResolvedValue(0 as never); vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); vi.mocked(prisma.auditLog.create).mockResolvedValue({} as never); vi.mocked(prisma.user.findUnique).mockResolvedValue({ @@ -586,6 +587,72 @@ describe("GET /api/measurements/series — dense-kind day-bucketing (v1.28.25)", expect(body.data.points[0].id).toBe("g1"); }); + it("hour-buckets a pulse window too dense to send raw (#1023)", async () => { + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + // One reading a minute for 30 days. + vi.mocked(prisma.measurement.count).mockResolvedValue(43_200 as never); + vi.mocked(prisma.$queryRaw) + .mockResolvedValueOnce([ + { + bucket_start: new Date("2026-06-01T08:00:00Z"), + mean: 71.666, + min_value: 58, + max_value: 96, + }, + ] as never) + .mockResolvedValueOnce([ + { n: 43_200, mean: 72.1, min: 50, max: 150, sd: 11.5 }, + ] as never); + + const res = await GET(req("kind=pulse&days=30")); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { + points: Array>; + stats: { count: number; mean: number }; + }; + }; + expect(prisma.measurement.findMany).not.toHaveBeenCalled(); + expect(body.data.points).toEqual([ + { + id: "hour:2026-06-01T08:00:00.000Z", + at: "2026-06-01T08:00:00.000Z", + value: 71.67, + secondary: null, + valueMin: 58, + valueMax: 96, + }, + ]); + // Stats describe the readings, not the buckets. + expect(body.data.stats.count).toBe(43_200); + expect(body.data.stats.mean).toBe(72.1); + // The bucket read folds per hour, not per day. + const bucketSql = vi.mocked(prisma.$queryRaw).mock.calls[0] as unknown[]; + expect(bucketSql.slice(1)).toContain("hour"); + }); + + it("keeps a pulse window at or under the row cap raw", async () => { + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + vi.mocked(prisma.measurement.count).mockResolvedValue(10_000 as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + id: "p1", + value: 64, + measuredAt: new Date("2026-06-01T08:03:00Z"), + valueMin: null, + valueMax: null, + }, + ] as never); + const res = await GET(req("kind=pulse&days=90")); + const body = (await res.json()) as { + data: { points: Array<{ id: string; at: string }> }; + }; + expect(prisma.$queryRaw).not.toHaveBeenCalled(); + expect(body.data.points).toEqual([ + expect.objectContaining({ id: "p1", at: "2026-06-01T08:03:00.000Z" }), + ]); + }); + it("keeps sparse kinds raw even on the ten-year window", async () => { vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); vi.mocked(prisma.measurement.findMany).mockResolvedValue([ diff --git a/src/app/api/measurements/series/route.ts b/src/app/api/measurements/series/route.ts index 925d778a4..80d5ed035 100644 --- a/src/app/api/measurements/series/route.ts +++ b/src/app/api/measurements/series/route.ts @@ -51,6 +51,21 @@ const SLEEP_SERIES_MAX_DAYS = 365; const DENSE_SERIES_RAW_WINDOW_DAYS = 90; const DENSE_SERIES_KINDS: ReadonlySet = new Set(["glucose", "pulse"]); +/** + * #1023 — inside the raw window, a pulse series with more rows than this is + * bucketed per local hour in SQL instead of sent raw. + * + * A watch that records heart rate once a minute puts 43 000 rows in the + * default 30-day window and 130 000 in 90 days: a 15 MB response the client + * could not draw at that resolution anyway, built from as many objects on the + * server. Hourly buckets carry the hour's mean with its low/high band, the + * shape an hourly-bucket import already has, and cap the answer at 24 points + * a day. The cap sits far above what sparse sources produce (a cuff, hourly + * buckets over 90 days, a watch sampling every five minutes over 30 days), so + * those series stay raw and unchanged. + */ +const DENSE_SERIES_RAW_ROW_CAP = 10_000; + const kindEnum = z.enum([ "weight", "bloodPressure", @@ -352,9 +367,20 @@ export const GET = apiHandler(async (request: NextRequest) => { }; }); } else if ( - DENSE_SERIES_KINDS.has(kind) && - days > DENSE_SERIES_RAW_WINDOW_DAYS + (DENSE_SERIES_KINDS.has(kind) && days > DENSE_SERIES_RAW_WINDOW_DAYS) || + (kind === "pulse" && + (await prisma.measurement.count({ + where: { + userId: user.id, + type: "PULSE", + measuredAt: { gte: since }, + deletedAt: null, + }, + })) > DENSE_SERIES_RAW_ROW_CAP) ) { + // Day buckets past the raw window; hour buckets for a pulse stream too + // dense to send raw inside it. + const grain = days > DENSE_SERIES_RAW_WINDOW_DAYS ? "day" : "hour"; // v1.28.25 — long-window read of a sample-dense kind (CGM glucose, // per-sample / hourly pulse). Day-bucket in SQL instead of walking // every raw row into JS: one aggregate pass in Postgres returns at @@ -377,7 +403,7 @@ export const GET = apiHandler(async (request: NextRequest) => { m."value_min", m."value_max", date_trunc( - 'day', + ${grain}, (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${userTz} ) AS local_day FROM measurements m @@ -422,7 +448,10 @@ export const GET = apiHandler(async (request: NextRequest) => { AND m."deleted_at" IS NULL `; const round2 = (v: number) => Math.round(v * 100) / 100; - const dayId = (d: Date) => `day:${userDayKey(d, userTz)}`; + const dayId = (d: Date) => + grain === "day" + ? `day:${userDayKey(d, userTz)}` + : `hour:${d.toISOString()}`; if (kind === "glucose") { // Same unit engine as the raw glucose branch: canonical mg/dL in // the DB, converted ONCE at serialization to the user's preference. @@ -457,9 +486,10 @@ export const GET = apiHandler(async (request: NextRequest) => { }; } } else { - // pulse — `value` is the day's average; valueMin/valueMax carry the - // day's low/high band (folding each hourly bucket's own spread via - // the COALESCE above), same band semantics as the per-hour shape. + // pulse — `value` is the bucket's average; valueMin/valueMax carry the + // bucket's low/high band (folding each hourly import bucket's own + // spread via the COALESCE above), same band semantics as the per-hour + // shape. points = bucketRows.map((r) => ({ id: dayId(r.bucket_start), at: r.bucket_start.toISOString(), diff --git a/src/lib/analytics/resting-pulse-read.ts b/src/lib/analytics/resting-pulse-read.ts new file mode 100644 index 000000000..58996b42e --- /dev/null +++ b/src/lib/analytics/resting-pulse-read.ts @@ -0,0 +1,66 @@ +/** + * The low-percentile resting-pulse proxy, folded in Postgres. + * + * `deriveRestingProxyFromPulse` takes every raw `PULSE` sample of a window + * and reduces each local day to its 20th percentile. On a watch that records + * heart rate once a minute, a 90-day window is 130 000 samples: reading them + * as objects, keying each one to its local day and sorting each day held the + * event loop for over a second per request, and the route that did it runs on + * every insights page view. The reduction needs only the day's percentile, + * its sample count and its first instant, so it runs in SQL and one row per + * day crosses the wire. + * + * Output matches `deriveRestingProxyFromPulse` exactly: `percentile_cont` + * interpolates between the two neighbouring order statistics the same way + * `percentile` in `strain-score.ts` does, the value is rounded the same way + * in JavaScript, days under `RESTING_PROXY_MIN_DAILY_SAMPLES` are skipped, + * and each day is anchored on its earliest sample. Days are cut in the zone + * the caller passes, with the same expression `readDayAggregates` uses. + */ +import { Prisma, type PrismaClient } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { isValidTimezone } from "@/lib/tz/format"; + +import { + RESTING_PROXY_DAILY_PERCENTILE, + RESTING_PROXY_MIN_DAILY_SAMPLES, + type PulseSample, +} from "./resting-pulse"; + +export async function readRestingPulseProxy(opts: { + userId: string; + /** Inclusive lower bound on `measuredAt`. */ + since: Date; + /** IANA zone the day boundary is cut in. */ + timeZone: string; + db?: Pick; +}): Promise { + const timeZone = isValidTimezone(opts.timeZone) ? opts.timeZone : "UTC"; + const fraction = RESTING_PROXY_DAILY_PERCENTILE / 100; + const rows = await (opts.db ?? prisma).$queryRaw< + Array<{ first_at: Date; p: number }> + >(Prisma.sql` + WITH src AS ( + SELECT + to_char((m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}, 'YYYY-MM-DD') AS day, + m."measured_at" AS measured_at, + m."value" AS value + FROM measurements m + WHERE m."user_id" = ${opts.userId} + AND m."type" = 'PULSE'::measurement_type + AND m."deleted_at" IS NULL + AND m."measured_at" >= ${opts.since} + ) + SELECT + MIN(measured_at) AS first_at, + (percentile_cont(${fraction}::double precision) WITHIN GROUP (ORDER BY value))::double precision AS p + FROM src + GROUP BY day + HAVING COUNT(*) >= ${RESTING_PROXY_MIN_DAILY_SAMPLES} + ORDER BY first_at ASC + `); + return rows.map((r) => ({ + measuredAt: r.first_at, + value: Math.round(Number(r.p)), + })); +} diff --git a/src/lib/doctor-report/__tests__/leaf-gating-sweep.test.ts b/src/lib/doctor-report/__tests__/leaf-gating-sweep.test.ts index 7517335c4..c279ed081 100644 --- a/src/lib/doctor-report/__tests__/leaf-gating-sweep.test.ts +++ b/src/lib/doctor-report/__tests__/leaf-gating-sweep.test.ts @@ -48,6 +48,7 @@ const { calls, rows, prismaMock } = vi.hoisted(() => { } return Promise.resolve(rows[name] ?? []); }, + count: () => Promise.resolve(0), findUnique: () => { calls.push(name); return Promise.resolve(rows[`${name}Unique`]?.[0] ?? null); diff --git a/src/lib/doctor-report/collect.ts b/src/lib/doctor-report/collect.ts index 05fa74ec1..fe2154b0a 100644 --- a/src/lib/doctor-report/collect.ts +++ b/src/lib/doctor-report/collect.ts @@ -76,6 +76,20 @@ import { const DENSE_REPORT_RAW_WINDOW_DAYS = 90; +/** + * #1023 — inside the raw window, pulse takes the per-day path anyway once the + * window holds more readings than this. + * + * A watch that records heart rate once a minute puts 130 000 readings in a + * 90-day report. Read raw, every one of them became an object, went through + * the canonical-source collapse, and landed in the report's series, where the + * trend line and the sparkline draw a few hundred points at most. The day + * path already serves every longer report: statistics over the raw readings, + * one point per local day for the series. A cuff, an hourly-bucket import or + * a watch sampling every few minutes stays under the cap and stays raw. + */ +const DENSE_REPORT_RAW_ROW_CAP = 10_000; + /** No medication row is read unless at least one medication leaf is admitted. */ const MEDICATION_LEAVES = [ "MEDICATION_LIST", @@ -156,7 +170,17 @@ export async function collectDoctorReportData( const lastDay = new Date(`${userDayKey(end, reportTz)}T00:00:00.000Z`); const aggregateDenseTypes = days > DENSE_REPORT_RAW_WINDOW_DAYS; - const densePulse = aggregateDenseTypes && !excluded.includes("PULSE"); + const densePulse = + !excluded.includes("PULSE") && + (aggregateDenseTypes || + (await prisma.measurement.count({ + where: { + userId, + type: "PULSE", + measuredAt: { gte: start, lte: end }, + deletedAt: null, + }, + })) > DENSE_REPORT_RAW_ROW_CAP); const denseGlucose = aggregateDenseTypes && !excluded.includes("BLOOD_GLUCOSE"); const rawExcluded = [...excluded]; diff --git a/src/lib/openapi/routes/measurements.ts b/src/lib/openapi/routes/measurements.ts index 13158a3e4..e36506223 100644 --- a/src/lib/openapi/routes/measurements.ts +++ b/src/lib/openapi/routes/measurements.ts @@ -695,7 +695,9 @@ const seriesQuerySchema = z.object({ const seriesPointSchema = z.object({ id: z .string() - .describe("Measurement row id, or `sleep:` for a sleep night."), + .describe( + "Measurement row id; `sleep:` for a sleep night; `day:` for a day bucket (`pulse` / `glucose` beyond 90 days); `hour:` for an hour bucket (`pulse` within 90 days when the window holds more than 10 000 readings).", + ), at: z.iso.datetime({ offset: true }).describe("Point timestamp (ISO-8601)."), value: z.number().describe("Primary value in the top-level `unit`."), secondary: z @@ -749,7 +751,7 @@ const seriesResponse = z .meta({ id: "MeasurementsSeriesResponse", description: - "iOS-friendly per-kind time series: one point per reading (or per reconstructed night for `sleep`), an explicit `unit` token, and a summary `stats` block.", + "iOS-friendly per-kind time series: one point per reading (or per reconstructed night for `sleep`), an explicit `unit` token, and a summary `stats` block. Sample-dense kinds are bucketed server-side: `pulse` and `glucose` beyond 90 days per local day, and `pulse` within 90 days per local hour when the window holds more than 10 000 readings (a bucket carries the mean as `value` and its range as `valueMin` / `valueMax`). `stats` is always computed over the underlying readings, never over the buckets.", }); // ── Batched daily series (v1.18.6 dashboard fetch coalescing) ──────── diff --git a/tests/integration/dense-pulse-raw-window-reads.test.ts b/tests/integration/dense-pulse-raw-window-reads.test.ts new file mode 100644 index 000000000..1ff3d0398 --- /dev/null +++ b/tests/integration/dense-pulse-raw-window-reads.test.ts @@ -0,0 +1,238 @@ +/** + * #1023 — three reads inside the 90-day raw window, on a dense pulse stream. + * + * Past 90 days every pulse reader already folded in SQL. Inside it, three + * still read every raw heart-rate row: the insights page's resting-pulse + * proxy, the per-kind series the iOS chart draws, and the doctor report. On + * a watch that records a reading a minute that is 130 000 rows per read; on + * the reproduction fixture (1.3 M rows) the insights page blocked the event + * loop for 1.3 s, the 90-day series answered with 14.7 MB of JSON and the + * 90-day report held 6.9 MB of pulse points it would draw as one sparkline. + * + * Each case here runs on real Postgres against two accounts: a dense one (one + * reading a minute) and a sparse one (a cuff and a few readings a day), so + * the tests pin both that the dense stream folds and that sparse data keeps + * exactly its old shape. + */ +import { NextRequest } from "next/server"; +import { beforeAll, describe, expect, it, vi } from "vitest"; + +import { cookieJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; +import { deriveRestingProxyFromPulse } from "@/lib/analytics/resting-pulse"; +import { readRestingPulseProxy } from "@/lib/analytics/resting-pulse-read"; +import { collectDoctorReportData } from "@/lib/doctor-report-data"; +import { selectionFromLeaves } from "@/lib/report-selection/selection"; +import { userDayKey } from "@/lib/tz/format"; +import { invalidateUserTimezone } from "@/lib/tz/resolver"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const prisma = getPrismaClient(); +const TZ = "America/New_York"; +const DENSE = "dense-raw-window-owner"; +const SPARSE = "sparse-raw-window-owner"; +/** Ten days of one reading a minute: over every raw-window row cap. */ +const DENSE_ROWS = 10 * 24 * 60; +const NOW = new Date(Math.floor(Date.now() / 60_000) * 60_000 - 60_000); + +/** + * A cuff and a watch that samples a few times a day: irregular times, values + * with fractional percentiles, one day with too few readings for the proxy, + * and days that straddle the New York midnight in UTC. + */ +function sparseReadings(): Array<{ at: Date; value: number }> { + const out: Array<{ at: Date; value: number }> = []; + for (let day = 1; day <= 20; day++) { + const perDay = day === 7 ? 2 : 3 + (day % 5); + for (let i = 0; i < perDay; i++) { + out.push({ + at: new Date( + NOW.getTime() - day * 86_400_000 + (i * 197 + day * 13) * 60_000, + ), + value: 52 + ((day * 7 + i * 11) % 41) + (i % 2 ? 0.5 : 0), + }); + } + } + return out; +} + +async function session(userId: string): Promise { + const s = await prisma.session.create({ + data: { userId, expiresAt: new Date(Date.now() + 3_600_000) }, + }); + cookieJar.clear(); + cookieJar.set("healthlog_session", s.id); +} + +beforeAll(async () => { + await truncateAllTables(prisma); + for (const id of [DENSE, SPARSE]) { + await prisma.user.create({ data: { id, username: id, timezone: TZ } }); + invalidateUserTimezone(id); + } + await prisma.$executeRawUnsafe( + `INSERT INTO measurements ( + id, user_id, type, value, unit, source, measured_at, + created_at, updated_at, sync_version) + SELECT 'dr' || lpad(g::text, 7, '0'), $1, 'PULSE'::measurement_type, + 55 + (g % 50), 'bpm', 'GOOGLE_HEALTH'::measurement_source, + ($2::timestamptz AT TIME ZONE 'UTC') - (g * interval '1 minute'), + now(), now(), 1 + FROM generate_series(0, $3::int - 1) g`, + DENSE, + NOW.toISOString(), + DENSE_ROWS, + ); + await prisma.measurement.createMany({ + data: sparseReadings().map((r, i) => ({ + id: `sp${String(i).padStart(4, "0")}`, + userId: SPARSE, + type: "PULSE" as const, + value: r.value, + unit: "bpm", + source: "MANUAL" as const, + measuredAt: r.at, + })), + }); + await prisma.$executeRawUnsafe(`ANALYZE measurements`); +}, 120_000); + +describe("resting-pulse proxy folded in Postgres", () => { + it("returns the in-memory derivation's series exactly", async () => { + const since = new Date(NOW.getTime() - 90 * 86_400_000); + for (const userId of [SPARSE, DENSE]) { + const raw = await prisma.measurement.findMany({ + where: { userId, type: "PULSE", measuredAt: { gte: since } }, + select: { measuredAt: true, value: true }, + }); + const expected = deriveRestingProxyFromPulse(raw, (d) => + userDayKey(d, TZ), + ); + const actual = await readRestingPulseProxy({ + userId, + since, + timeZone: TZ, + }); + expect(actual.length).toBeGreaterThan(0); + expect(actual).toEqual(expected); + } + }); + + it("is what the insights page reads, not the raw stream", async () => { + await session(DENSE); + const spy = vi.spyOn(prisma.measurement, "findMany"); + try { + const { GET } = await import("@/app/api/insights/comprehensive/route"); + const res = await GET( + new NextRequest("http://localhost/api/insights/comprehensive"), + ); + expect(res.status).toBe(200); + const pulseReads = spy.mock.calls.filter( + ([args]) => + (args as { where?: { type?: unknown } } | undefined)?.where?.type === + "PULSE", + ); + expect(pulseReads).toEqual([]); + } finally { + spy.mockRestore(); + } + }); +}); + +describe("per-kind pulse series inside the raw window", () => { + async function series(userId: string, days: number) { + await session(userId); + const { GET } = await import("@/app/api/measurements/series/route"); + const res = await GET( + new NextRequest( + `http://localhost/api/measurements/series?kind=pulse&days=${days}`, + ), + ); + expect(res.status).toBe(200); + return ( + (await res.json()) as { + data: { + points: Array<{ + id: string; + at: string; + value: number; + valueMin?: number | null; + valueMax?: number | null; + }>; + stats: { count: number; min: number; max: number }; + }; + } + ).data; + } + + it("sends a dense stream as hour buckets with its full range", async () => { + const data = await series(DENSE, 30); + // Ten days of hours, give or take the partial hours at either end. + expect(data.points.length).toBeLessThanOrEqual(10 * 24 + 2); + expect(data.points.length).toBeGreaterThanOrEqual(10 * 24 - 2); + expect(data.points.every((p) => p.id.startsWith("hour:"))).toBe(true); + expect(data.stats.count).toBe(DENSE_ROWS); + expect(Math.min(...data.points.map((p) => p.valueMin ?? Infinity))).toBe( + data.stats.min, + ); + expect(Math.max(...data.points.map((p) => p.valueMax ?? -Infinity))).toBe( + data.stats.max, + ); + }); + + it("keeps sparse readings raw, one point per reading", async () => { + const data = await series(SPARSE, 30); + expect(data.points).toHaveLength(sparseReadings().length); + expect(data.points.every((p) => p.id.startsWith("sp"))).toBe(true); + }); +}); + +describe("doctor report inside the raw window", () => { + async function report(userId: string, days: number) { + const end = new Date(); + const start = new Date(end.getTime() - days * 86_400_000); + return collectDoctorReportData( + userId, + { start, end, days }, + selectionFromLeaves(["PULSE"]), + ); + } + + it("folds a dense pulse stream to one point per day, statistics exact", async () => { + const data = await report(DENSE, 30); + const points = data.measurements.PULSE ?? []; + expect(points.length).toBeLessThanOrEqual(12); + expect(points.length).toBeGreaterThanOrEqual(10); + expect(data.stats.PULSE?.count).toBe(DENSE_ROWS); + }); + + it("keeps sparse readings raw", async () => { + const data = await report(SPARSE, 30); + expect(data.measurements.PULSE).toHaveLength(sparseReadings().length); + expect(data.stats.PULSE?.count).toBe(sparseReadings().length); + }); +}); From 3458100ac4d3e6848391504c15461652ed2c9921 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 27 Sep 2026 21:16:20 +0200 Subject: [PATCH 4/5] test: give the report's pulse-window count a mock and a typed route call --- src/app/api/export/health-record/__tests__/route.test.ts | 3 ++- tests/integration/dense-pulse-raw-window-reads.test.ts | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/src/app/api/export/health-record/__tests__/route.test.ts b/src/app/api/export/health-record/__tests__/route.test.ts index 664e6fc4f..29d5fe949 100644 --- a/src/app/api/export/health-record/__tests__/route.test.ts +++ b/src/app/api/export/health-record/__tests__/route.test.ts @@ -19,7 +19,8 @@ import { unzipSync } from "fflate"; vi.mock("@/lib/db", () => ({ prisma: { - measurement: { findMany: vi.fn() }, + // `count` sizes the report's pulse window (#1023): a sparse account. + measurement: { findMany: vi.fn(), count: vi.fn(async () => 0) }, medication: { findMany: vi.fn() }, medicationIntakeEvent: { findMany: vi.fn() }, moodEntry: { findMany: vi.fn() }, diff --git a/tests/integration/dense-pulse-raw-window-reads.test.ts b/tests/integration/dense-pulse-raw-window-reads.test.ts index 1ff3d0398..b613840db 100644 --- a/tests/integration/dense-pulse-raw-window-reads.test.ts +++ b/tests/integration/dense-pulse-raw-window-reads.test.ts @@ -147,7 +147,9 @@ describe("resting-pulse proxy folded in Postgres", () => { const spy = vi.spyOn(prisma.measurement, "findMany"); try { const { GET } = await import("@/app/api/insights/comprehensive/route"); - const res = await GET( + // The route reads no request field, so its handler is typed without one. + const callGet = GET as unknown as (req: NextRequest) => Promise; + const res = await callGet( new NextRequest("http://localhost/api/insights/comprehensive"), ); expect(res.status).toBe(200); From 68368300389002fa550ebcedff6f5b1cd9b755db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 27 Sep 2026 21:20:24 +0200 Subject: [PATCH 5/5] fix(doctor-report): report readings, not day buckets, for a dense type A type served from day buckets (pulse or glucose past 90 days, and since the previous change a dense pulse stream on any window) is drawn one point per day, each the day's mean at the day's start. The summary line counted those points as measurements, and the FHIR export stamped the last day's mean as the latest observation: a value and a time nobody measured. The day-bucket stats now carry when their newest reading was taken (`latestAt`). `latestReading` returns that reading when it is there, and the summary line counts each type's readings from its stats. Raw series are unchanged: their stats count equals their points and their last point is the newest reading. --- src/lib/doctor-report-pdf/clinical-summary.ts | 6 +- src/lib/doctor-report-types.ts | 8 ++ .../day-bucket-latest-and-count.test.ts | 121 ++++++++++++++++++ src/lib/doctor-report/measurement-series.ts | 1 + src/lib/fhir/resources/common.ts | 11 +- .../dense-pulse-raw-window-reads.test.ts | 11 ++ 6 files changed, 155 insertions(+), 3 deletions(-) create mode 100644 src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts diff --git a/src/lib/doctor-report-pdf/clinical-summary.ts b/src/lib/doctor-report-pdf/clinical-summary.ts index 9543f5e45..013847f6a 100644 --- a/src/lib/doctor-report-pdf/clinical-summary.ts +++ b/src/lib/doctor-report-pdf/clinical-summary.ts @@ -55,8 +55,10 @@ export function buildClinicalSummaryLines( num: DoctorReportNumberFormatter, ): string[] { const lines: string[] = []; - const totalReadings = Object.values(data.measurements).reduce( - (sum, arr) => sum + arr.length, + // The reading count, not the point count: a dense type served from day + // buckets draws one point per day, but its stats count every reading. + const totalReadings = Object.entries(data.measurements).reduce( + (sum, [type, arr]) => sum + (data.stats[type]?.count ?? arr.length), 0, ); const paramCount = Object.keys(data.stats).length; diff --git a/src/lib/doctor-report-types.ts b/src/lib/doctor-report-types.ts index c37bab2c4..cdacd85b1 100644 --- a/src/lib/doctor-report-types.ts +++ b/src/lib/doctor-report-types.ts @@ -36,8 +36,16 @@ export interface DoctorReportStats { avg: number; min: number; max: number; + /** Readings behind the figures, however the series was drawn. */ count: number; latest: number; + /** + * When the newest reading behind `latest` was taken (ISO). Set for a type + * served from day buckets, whose series points are day means: the series' + * last point is then not a reading, and this is the only place the newest + * one survives. Absent on the raw path, where the series' last point is it. + */ + latestAt?: string; } /** diff --git a/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts new file mode 100644 index 000000000..feb000741 --- /dev/null +++ b/src/lib/doctor-report/__tests__/day-bucket-latest-and-count.test.ts @@ -0,0 +1,121 @@ +/** + * A type served from day buckets (pulse or glucose on a long window, or a + * dense stream on any window) is drawn one point per day, and each point is + * that day's mean stamped at the day's start. Two readers used to treat those + * points as readings: the report's summary line counted them as measurements, + * and the FHIR export stamped the last day's mean as the latest observation. + * Both now read the stats, which describe the readings themselves. + */ +import { describe, expect, it } from "vitest"; + +import type { DoctorReportData } from "@/lib/doctor-report-data"; +import { buildClinicalSummaryLines } from "@/lib/doctor-report-pdf/clinical-summary"; +import { latestReading } from "@/lib/fhir/resources/common"; +import type { DenseMeasurementBucket } from "../dense-buckets"; +import { summariseDenseBuckets } from "../measurement-series"; + +function bucket( + day: string, + count: number, + sum: number, + latestAt: string, + latestValue: number, +): DenseMeasurementBucket { + return { + bucketStart: new Date(`${day}T00:00:00.000Z`), + type: "PULSE", + source: "GOOGLE_HEALTH", + deviceType: null, + glucoseContext: null, + count, + sumValue: sum, + minValue: 50, + maxValue: 120, + latestAt: new Date(latestAt), + latestValue, + clinicalCount: count, + clinicalSum: sum, + clinicalSumSquares: null, + clinicalFirstAt: null, + clinicalLastAt: null, + clinicalTirCount: 0, + clinicalTbr1Count: 0, + clinicalTbr2Count: 0, + clinicalTar1Count: 0, + clinicalTar2Count: 0, + clinicalLowRiskSum: null, + clinicalHighRiskSum: null, + }; +} + +/** Two days of one reading a minute; the newest reading is 61 bpm at 23:59. */ +const summary = summariseDenseBuckets( + [ + bucket("2026-09-01", 1440, 1440 * 70, "2026-09-01T23:59:00.000Z", 66), + bucket("2026-09-02", 1440, 1440 * 75, "2026-09-02T23:59:00.000Z", 61), + ], + "UTC", + null, + 120, +); + +function reportData(): DoctorReportData { + return { + measurements: summary.byType, + stats: summary.stats, + } as unknown as DoctorReportData; +} + +describe("a pulse series served from day buckets", () => { + it("is drawn as day means", () => { + expect(summary.byType.PULSE).toEqual([ + { value: 70, measuredAt: "2026-09-01T00:00:00.000Z" }, + { value: 75, measuredAt: "2026-09-02T00:00:00.000Z" }, + ]); + }); + + it("reports its newest actual reading to FHIR, not the last day's mean", () => { + expect(latestReading(reportData(), "PULSE")).toEqual({ + value: 61, + measuredAt: "2026-09-02T23:59:00.000Z", + }); + }); + + it("counts its readings in the summary line, not its points", () => { + const counts: unknown[] = []; + const lines = buildClinicalSummaryLines( + { + ...reportData(), + period: { days: 120 }, + compliance: {}, + } as unknown as DoctorReportData, + (key, vars) => { + if (key === "doctorReport.summaryReadings") counts.push(vars?.count); + return key; + }, + (n) => String(n), + ); + expect(lines).toContain("doctorReport.summaryReadings"); + expect(counts).toEqual([2880]); + }); +}); + +describe("a raw series", () => { + it("still reports its last point as the latest reading", () => { + const data = { + measurements: { + WEIGHT: [ + { value: 80, measuredAt: "2026-09-01T07:00:00.000Z" }, + { value: 79.5, measuredAt: "2026-09-02T07:00:00.000Z" }, + ], + }, + stats: { + WEIGHT: { avg: 79.75, min: 79.5, max: 80, count: 2, latest: 79.5 }, + }, + } as unknown as DoctorReportData; + expect(latestReading(data, "WEIGHT")).toEqual({ + value: 79.5, + measuredAt: "2026-09-02T07:00:00.000Z", + }); + }); +}); diff --git a/src/lib/doctor-report/measurement-series.ts b/src/lib/doctor-report/measurement-series.ts index f6d6e1d07..1f0d0a6ef 100644 --- a/src/lib/doctor-report/measurement-series.ts +++ b/src/lib/doctor-report/measurement-series.ts @@ -143,6 +143,7 @@ export function summariseDenseBuckets( min: value.min, max: value.max, latest: value.latest, + latestAt: value.latestAt.toISOString(), }; } const glucoseStats: DenseMeasurementSummary["glucoseStats"] = {}; diff --git a/src/lib/fhir/resources/common.ts b/src/lib/fhir/resources/common.ts index 920ef0592..13c0751b7 100644 --- a/src/lib/fhir/resources/common.ts +++ b/src/lib/fhir/resources/common.ts @@ -127,13 +127,22 @@ export function codeableFromMapping(m: LoincMapping): FhirCodeableConcept { return { text: m.display }; } -/** Latest `{ value, measuredAt }` for a type, or null when no rows. */ +/** + * Latest `{ value, measuredAt }` for a type, or null when no rows. + * + * A dense type (pulse, glucose) can be served from day buckets, and then the + * series' last point is the last day's MEAN stamped at that day's start: a + * value nobody measured, at a time nobody measured it. Its stats carry the + * newest actual reading, so that wins whenever it is there. + */ export function latestReading( data: DoctorReportData, type: string, ): { value: number; measuredAt: string } | null { const series = data.measurements[type]; if (!series || series.length === 0) return null; + const stat = data.stats[type]; + if (stat?.latestAt) return { value: stat.latest, measuredAt: stat.latestAt }; return series[series.length - 1]; } diff --git a/tests/integration/dense-pulse-raw-window-reads.test.ts b/tests/integration/dense-pulse-raw-window-reads.test.ts index b613840db..886507ace 100644 --- a/tests/integration/dense-pulse-raw-window-reads.test.ts +++ b/tests/integration/dense-pulse-raw-window-reads.test.ts @@ -22,6 +22,7 @@ import { getPrismaClient, truncateAllTables } from "./setup"; import { deriveRestingProxyFromPulse } from "@/lib/analytics/resting-pulse"; import { readRestingPulseProxy } from "@/lib/analytics/resting-pulse-read"; import { collectDoctorReportData } from "@/lib/doctor-report-data"; +import { latestReading } from "@/lib/fhir/resources/common"; import { selectionFromLeaves } from "@/lib/report-selection/selection"; import { userDayKey } from "@/lib/tz/format"; import { invalidateUserTimezone } from "@/lib/tz/resolver"; @@ -230,6 +231,16 @@ describe("doctor report inside the raw window", () => { expect(points.length).toBeLessThanOrEqual(12); expect(points.length).toBeGreaterThanOrEqual(10); expect(data.stats.PULSE?.count).toBe(DENSE_ROWS); + // The export's latest observation is the newest reading itself. + const newest = await prisma.measurement.findFirstOrThrow({ + where: { userId: DENSE, type: "PULSE" }, + orderBy: { measuredAt: "desc" }, + select: { value: true, measuredAt: true }, + }); + expect(latestReading(data, "PULSE")).toEqual({ + value: newest.value, + measuredAt: newest.measuredAt.toISOString(), + }); }); it("keeps sparse readings raw", async () => {