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/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/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-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/__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/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/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..7eee64efc --- /dev/null +++ b/src/lib/google-health/__tests__/sync-metrics-dense-pages.test.ts @@ -0,0 +1,133 @@ +/** + * #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, 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) => { + 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, + runWithGoogleHealthClientOutcome, +} 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 < shape.perPage; i++) { + const g = page * shape.perPage + i; + dataPoints.push({ + heartRate: { + beatsPerMinute: String(55 + (g % 50)), + sampleTime: { physicalTime: new Date(NOW - g * 60_000).toISOString() }, + }, + }); + } + return { + dataPoints, + ...(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}`); + 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}`, + ]); + }); + + 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); + }); +}); 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/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..886507ace --- /dev/null +++ b/tests/integration/dense-pulse-raw-window-reads.test.ts @@ -0,0 +1,251 @@ +/** + * #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 { 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"; + +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"); + // 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); + 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); + // 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 () => { + const data = await report(SPARSE, 30); + expect(data.measurements.PULSE).toHaveLength(sparseReadings().length); + expect(data.stats.PULSE?.count).toBe(sparseReadings().length); + }); +}); 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); +});