mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
Merge pull request #1051 from MBombeck/fix/1023-dense-stream-memory
Keep a dense heart-rate stream out of memory
This commit is contained in:
25 files changed
+1253
-103
No files matched your search
@@ -40909,7 +40909,9 @@ components:
|
||||
properties:
|
||||
id:
|
||||
type: string
|
||||
description: Measurement row id, or `sleep:<wake-day>` for a sleep night.
|
||||
description: Measurement row id; `sleep:<wake-day>` for a sleep night; `day:<local-day>` for a day bucket (`pulse` /
|
||||
`glucose` beyond 90 days); `hour:<ISO start>` 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:
|
||||
|
||||
@@ -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() },
|
||||
|
||||
@@ -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<typeof vi.fn>).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<typeof vi.fn>).mockResolvedValue(
|
||||
[],
|
||||
);
|
||||
|
||||
@@ -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<string, { sum: number; count: number }>();
|
||||
for (const s of restingPulseSeries) {
|
||||
const key = localDayKey(s.measuredAt);
|
||||
|
||||
@@ -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<Record<string, unknown>>;
|
||||
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([
|
||||
|
||||
@@ -51,6 +51,21 @@ const SLEEP_SERIES_MAX_DAYS = 365;
|
||||
const DENSE_SERIES_RAW_WINDOW_DAYS = 90;
|
||||
const DENSE_SERIES_KINDS: ReadonlySet<string> = 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(),
|
||||
|
||||
@@ -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<PrismaClient, "$queryRaw">;
|
||||
}): Promise<PulseSample[]> {
|
||||
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)),
|
||||
}));
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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",
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
|
||||
@@ -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];
|
||||
|
||||
@@ -143,6 +143,7 @@ export function summariseDenseBuckets(
|
||||
min: value.min,
|
||||
max: value.max,
|
||||
latest: value.latest,
|
||||
latestAt: value.latestAt.toISOString(),
|
||||
};
|
||||
}
|
||||
const glucoseStats: DenseMeasurementSummary["glucoseStats"] = {};
|
||||
|
||||
@@ -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];
|
||||
}
|
||||
|
||||
|
||||
@@ -228,7 +228,11 @@ export const softSkipStorage = new AsyncLocalStorage<SoftSkipTracker>();
|
||||
* 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<string, { type: MeasurementType; measuredAt: Date }>;
|
||||
}
|
||||
export const rollupDeferStorage = new AsyncLocalStorage<RollupDeferTracker>();
|
||||
|
||||
@@ -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 };
|
||||
}
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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",
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -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<typeof import("@/lib/safe-fetch")>();
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<void>,
|
||||
) => {
|
||||
const points = (await fetchDataPointsMock(
|
||||
dataType,
|
||||
token,
|
||||
verb,
|
||||
query,
|
||||
)) as unknown[];
|
||||
if (points.length > 0) await onPage(points);
|
||||
},
|
||||
),
|
||||
mapWeight: mapWeightMock,
|
||||
};
|
||||
});
|
||||
|
||||
@@ -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<GoogleHealthDataPoint[]> {
|
||||
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<void>,
|
||||
): Promise<void> {
|
||||
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<GoogleHealthDataPoint[]> => {
|
||||
const points: GoogleHealthDataPoint[] = [];
|
||||
): Promise<void> => {
|
||||
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`) ─────────────────
|
||||
|
||||
@@ -265,7 +265,12 @@ export async function runWithGoogleHealthHardFailLedger<T>(
|
||||
* 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<string, { type: MeasurementType; measuredAt: Date }>;
|
||||
}
|
||||
const rollupDeferStorage = new AsyncLocalStorage<RollupDeferTracker>();
|
||||
|
||||
@@ -286,7 +291,7 @@ export async function runWithGoogleHealthSyncCycle<T>(
|
||||
): Promise<GoogleHealthSyncCycleResult<T>> {
|
||||
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<T>(
|
||||
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 };
|
||||
|
||||
@@ -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<string, unknown>[];
|
||||
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
|
||||
|
||||
@@ -695,7 +695,9 @@ const seriesQuerySchema = z.object({
|
||||
const seriesPointSchema = z.object({
|
||||
id: z
|
||||
.string()
|
||||
.describe("Measurement row id, or `sleep:<wake-day>` for a sleep night."),
|
||||
.describe(
|
||||
"Measurement row id; `sleep:<wake-day>` for a sleep night; `day:<local-day>` for a day bucket (`pulse` / `glucose` beyond 90 days); `hour:<ISO start>` 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) ────────
|
||||
|
||||
@@ -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<void> {
|
||||
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<Response>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<typeof import("@/lib/safe-fetch")>();
|
||||
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<T>(
|
||||
fn: () => Promise<T>,
|
||||
): 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);
|
||||
});
|
||||
Reference in new issue
Block a user