diff --git a/src/__tests__/dense-measurement-read-guard.test.ts b/src/__tests__/dense-measurement-read-guard.test.ts new file mode 100644 index 000000000..d669c99d4 --- /dev/null +++ b/src/__tests__/dense-measurement-read-guard.test.ts @@ -0,0 +1,177 @@ +/** + * Structural guard: background and insight readers must not materialise a + * dense measurement stream. + * + * #1023: a single account syncing heart rate from a watch took the worker + * down within half an hour. The status-card read was bounded in time + * (`measuredAt >= now − 91 days`) but not in rows, and a stream sampled every + * minute or faster turns 91 days into six or seven figures of objects. The + * read also ran into the statement timeout. The cure was to fold per day in + * SQL (`readDayAggregates`, `readLiveBuckets`), so the reader's size tracks + * the number of days, not the sampling rate. + * + * What this guard pins, in the directories that run on the worker or feed an + * insight: every `measurement.findMany` without a `take` reads a type that is + * named as a literal and is not one a device streams. A read whose type is a + * variable could be handed a dense type by the next caller, so it must be + * listed below with the reason it is safe. The list is frozen both ways: a new + * unlisted read fails, and an entry whose read is gone fails too. + * + * It is a tripwire, not a proof: the matcher reads source text, and a read + * written some other way (a raw SELECT of rows, a helper in another + * directory) passes it. The floor below fails the guard if the matcher stops + * finding reads at all. + */ +import { describe, expect, it } from "vitest"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { walkSourceFiles } from "./helpers/source-files"; +import { + CUMULATIVE_HK_TYPES, + HIGH_FREQUENCY_MEAN_TYPES, +} from "@/lib/measurements/apple-health-mapping"; + +const SRC = join(process.cwd(), "src"); + +const SCOPED_DIRS = [ + "lib/insights/", + "lib/jobs/", + "lib/rollups/", + "lib/ai/", + "lib/analytics/", + "lib/dashboard/", + "lib/daily/", +]; + +/** Types a device writes at sampling rate rather than a few times a day. */ +const DENSE_TYPES: ReadonlySet = new Set([ + "PULSE", + "BLOOD_GLUCOSE", + "HEART_RATE_VARIABILITY", + "OXYGEN_SATURATION", + ...CUMULATIVE_HK_TYPES, + ...HIGH_FREQUENCY_MEAN_TYPES, +]); + +/** + * Untaken reads whose type is not a sparse literal, keyed + * `::`, with why each stays bounded. + */ +const ALLOWED: Record = { + "lib/insights/comprehensive-generate.ts::(shorthand)": + 'Only reached for SLEEP_DURATION (guarded by `type === "SLEEP_DURATION"`); a few stage rows per night.', + "lib/insights/derived/vascular-age.ts::VASCULAR_AGE_TYPE": + "A constant for a type written at most a few times a day.", + "lib/insights/derived/fitness-age.ts::VO2_MAX_TYPE": + "A constant for VO2 max, written at most daily.", + "lib/insights/derived/six-minute-walk.ts::SIX_MINUTE_WALK_TYPE": + "A constant for six-minute-walk distance, written at most daily.", + "lib/insights/derived/wellness-scores.ts::measurementType": + "Daily computed / provider scores (recovery, strain, readiness), one row per day.", + "lib/jobs/reaction-line.ts::{ in: [...types]": + "Exact-instant match (`measuredAt: row.occurredAt`), not a window.", + "lib/jobs/step-consolidation-repair.ts::STEP_TYPE": + "Repair job over one user's step rows for a bounded day range; runs once per repair, not per status refresh.", + 'lib/ai/coach/snapshot.ts::"BLOOD_GLUCOSE" as never': + "Clinical CGM metrics (time in range, GMI, CV) need individual readings; 30 days at a sensor's fixed rate (at most one a minute) stays below 45 000 rows.", + 'lib/dashboard/snapshot.ts::"BLOOD_GLUCOSE"': + "Same 30-day clinical glucose panel as the coach snapshot, grouped by meal context.", + "lib/ai/coach/cycle-snapshot.ts::{ in: PHASE_CROSSTAB_METRIC_TYPES": + "Cycle-phase crosstab over 365 days. Its dense members (steps, CGM glucose) are the known remaining case; steps are drained to one row per day nightly, CGM is not. Listed so it cannot grow unseen.", + "lib/analytics/score/reader.ts::{ in: types": + "Health-score inputs: steps (drained to one row per day nightly), sleep, waist, weight, blood pressure, fasting glucose only.", +}; + +interface UntakenRead { + file: string; + line: number; + typeExpr: string; +} + +function scopedFiles(): string[] { + return walkSourceFiles(SRC, { floor: 3000 }) + .filter((p) => SCOPED_DIRS.some((d) => p.startsWith(d))) + .filter((p) => !p.includes("__tests__")) + .filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx")) + .sort(); +} + +function untakenReads(): { all: number; untaken: UntakenRead[] } { + let all = 0; + const untaken: UntakenRead[] = []; + for (const file of scopedFiles()) { + const src = readFileSync(join(SRC, file), "utf8"); + // Whitespace-tolerant: a call split across lines still matches. + const re = /\.measurement\s*\.\s*findMany\s*\(/g; + for (let m = re.exec(src); m; m = re.exec(src)) { + all += 1; + let i = m.index + m[0].length; + let depth = 1; + while (depth > 0 && i < src.length) { + if (src[i] === "(") depth += 1; + else if (src[i] === ")") depth -= 1; + i += 1; + } + const call = src.slice(m.index, i); + if (/\btake\b/.test(call)) continue; + const typed = /\btype\s*:\s*([^,\n}]+)/.exec(call); + const typeExpr = typed + ? typed[1].trim() + : /\btype\s*,/.test(call) + ? "(shorthand)" + : "(none)"; + untaken.push({ + file, + line: src.slice(0, m.index).split("\n").length, + typeExpr, + }); + } + } + return { all, untaken }; +} + +function sparseLiteral(typeExpr: string): boolean { + const lit = /^"([A-Z_0-9]+)"/.exec(typeExpr); + return lit !== null && !DENSE_TYPES.has(lit[1]); +} + +describe("dense measurement reads stay bounded (#1023)", () => { + const { all, untaken } = untakenReads(); + + it("finds the reads it is meant to police", () => { + // Guard against a matcher that silently matches nothing. + expect(all).toBeGreaterThanOrEqual(40); + expect(untaken.length).toBeGreaterThanOrEqual(15); + }); + + it("every untaken read names a sparse literal type or is listed with its reason", () => { + const offenders = untaken + .filter((r) => !sparseLiteral(r.typeExpr)) + .filter((r) => !(`${r.file}::${r.typeExpr}` in ALLOWED)) + .map((r) => `${r.file}:${r.line} type ${r.typeExpr}`); + expect(offenders).toEqual([]); + }); + + it("every listed exception still exists", () => { + const present = new Set(untaken.map((r) => `${r.file}::${r.typeExpr}`)); + expect(Object.keys(ALLOWED).filter((k) => !present.has(k))).toEqual([]); + }); + + it("the readers fixed for #1023 fold in SQL rather than reading rows", () => { + for (const file of [ + "lib/insights/graded-series.ts", + "lib/rollups/tiered-context.ts", + "lib/jobs/coach-plan-review.ts", + "lib/measurements/daily-series-read.ts", + ]) { + const src = readFileSync(join(SRC, file), "utf8"); + expect(src, file).not.toMatch(/\.measurement\s*\.\s*findMany\s*\(/); + } + const baseline = readFileSync( + join(SRC, "lib/insights/derived/baseline.ts"), + "utf8", + ); + expect(baseline).not.toMatch(/\.measurement\s*\.\s*findMany\s*\(/); + }); +}); diff --git a/src/app/api/insights/cards/__tests__/route.test.ts b/src/app/api/insights/cards/__tests__/route.test.ts index fbd8d1a08..d3252833b 100644 --- a/src/app/api/insights/cards/__tests__/route.test.ts +++ b/src/app/api/insights/cards/__tests__/route.test.ts @@ -1,6 +1,14 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import { NextRequest } from "next/server"; +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, diff --git a/src/lib/insights/__tests__/blood-pressure-status.test.ts b/src/lib/insights/__tests__/blood-pressure-status.test.ts index 437c065da..ce3d9f7a0 100644 --- a/src/lib/insights/__tests__/blood-pressure-status.test.ts +++ b/src/lib/insights/__tests__/blood-pressure-status.test.ts @@ -1,5 +1,13 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; +// The graded series folds per-day aggregates in SQL; the fake folds the +// mocked `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, diff --git a/src/lib/insights/__tests__/bmi-status.test.ts b/src/lib/insights/__tests__/bmi-status.test.ts index d0cc8da06..9db3e07a5 100644 --- a/src/lib/insights/__tests__/bmi-status.test.ts +++ b/src/lib/insights/__tests__/bmi-status.test.ts @@ -1,5 +1,13 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; +// The graded series folds per-day aggregates in SQL; the fake folds the +// mocked `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, diff --git a/src/lib/insights/__tests__/graded-series-rollups.test.ts b/src/lib/insights/__tests__/graded-series-rollups.test.ts index a6af77a54..1f386cfb8 100644 --- a/src/lib/insights/__tests__/graded-series-rollups.test.ts +++ b/src/lib/insights/__tests__/graded-series-rollups.test.ts @@ -1,7 +1,6 @@ import { afterEach, describe, expect, it, vi } from "vitest"; const findMany = vi.fn(); -const queryRaw = vi.fn(); vi.mock("@/lib/db", () => ({ prisma: { measurement: { @@ -10,12 +9,14 @@ vi.mock("@/lib/db", () => ({ // loadUserSourcePriority lazy-loads the source-priority blob; null // (default findUnique) falls back to the default ladders. user: { findUnique: vi.fn() }, - // v1.28.25 — dense types (BLOOD_GLUCOSE / PULSE) day-bucket the - // cold-tier fallback in SQL. - $queryRaw: (...args: unknown[]) => queryRaw(...args), }, })); +const readDayAggregates = vi.fn(); +vi.mock("@/lib/measurements/day-aggregates", () => ({ + readDayAggregates: (...args: unknown[]) => readDayAggregates(...args), +})); + const readBestGranularityRollups = vi.fn(); const rollupRow = (bucketStart: Date, mean: number) => ({ bucketStart, @@ -31,147 +32,160 @@ const rollupRow = (bucketStart: Date, mean: number) => ({ vi.mock("@/lib/rollups/measurement-read-wmy", () => ({ readBestGranularityRollups: (...args: unknown[]) => readBestGranularityRollups(...args), - // aggregateWmyBuckets is no longer imported by graded-series, but keep the - // module shape intact for any transitive consumer. aggregateWmyBuckets: vi.fn(), })); -import { buildGradedSeriesWithRollups } from "../graded-series"; +import { + buildGradedSeriesFromDayAggregates, + buildGradedSeriesFromPoints, + buildGradedSeriesWithRollups, +} from "../graded-series"; +import { foldDayAggregates } from "@/lib/measurements/__tests__/fake-day-aggregates"; +import type { ReadDayAggregatesOptions } from "@/lib/measurements/day-aggregates"; const dayMs = 24 * 60 * 60 * 1000; -/** Daily readings spanning `days` back from `now`. */ -function dailyRows(days: number, now: Date) { +/** Readings spanning `days` back from `now`, `perDay` per day. */ +function points(days: number, now: Date, perDay = 1) { const rows: Array<{ measuredAt: Date; value: number }> = []; for (let i = 0; i < days; i++) { - rows.push({ measuredAt: new Date(now.getTime() - i * dayMs), value: 80 }); + for (let k = 0; k < perDay; k++) { + rows.push({ + measuredAt: new Date(now.getTime() - i * dayMs - k * 3_600_000), + value: 60 + ((i * 7 + k * 3) % 41), + }); + } } - return rows; + return rows.reverse(); +} + +/** Serve `readDayAggregates` from an in-memory row set. */ +function serveFrom(rows: Array<{ measuredAt: Date; value: number }>) { + readDayAggregates.mockImplementation(async (opts: ReadDayAggregatesOptions) => + foldDayAggregates(rows, opts), + ); } afterEach(() => { vi.clearAllMocks(); }); -describe("buildGradedSeriesWithRollups — rollup-miss fallback", () => { +describe("buildGradedSeriesWithRollups — bounded reads (#1023)", () => { const now = new Date("2026-05-31T12:00:00Z"); + it("never materialises raw rows: the recent read is a per-day aggregate", async () => { + serveFrom(points(80, now, 24)); + readBestGranularityRollups.mockResolvedValue({ + granularity: "MONTH", + rows: [rollupRow(new Date("2025-09-01T00:00:00Z"), 79)], + }); + + await buildGradedSeriesWithRollups("u1", "PULSE", now); + + expect(findMany).not.toHaveBeenCalled(); + const opts = readDayAggregates.mock.calls[0][0] as ReadDayAggregatesOptions; + expect(opts).toMatchObject({ userId: "u1", type: "PULSE" }); + expect(opts.until).toEqual(now); + expect(opts.timeZone).toBe("Europe/Berlin"); + expect(now.getTime() - opts.since.getTime()).toBe(91 * dayMs); + // The same age partition the parity test below proves exact: a reading + // exactly 21 / 91 / 451 days old belongs to the older slice. + expect(opts.segmentStarts).toEqual( + [21, 91, 451].map((d) => new Date(now.getTime() - d * dayMs + 1)), + ); + }); + it("folds monthly/yearly from the bounded fallback when the tier has no coverage", async () => { - // The bounded recent read (~90 d) returns only recent points; the - // fallback read returns 2+ years so the fold has monthly/yearly. - findMany - // 1st call — bounded recent read (gte: since) - .mockResolvedValueOnce(dailyRows(80, now)) - // 2nd call — bounded fallback read (gte: 1095-day horizon) - .mockResolvedValueOnce(dailyRows(800, now)); - - // Tier miss: no MONTH / YEAR coverage at all. + serveFrom(points(800, now)); readBestGranularityRollups.mockResolvedValue(null); const series = await buildGradedSeriesWithRollups("u1", "WEIGHT", now); - // The bug: before the fix monthly/yearly were derived from the - // bounded ~90-day read and so were always empty even when years of - // raw history existed. After the fix they fold from the fallback. - expect(series.monthly.length).toBeGreaterThan(0); - // 800 days of history puts at least one year into the yearly tail. - expect(series.yearly.length).toBeGreaterThan(0); - // The fallback read must have been issued... - expect(findMany).toHaveBeenCalledTimes(2); - // ...and bounded (v1.28.25): it must carry a `measuredAt.gte` floor - // ~1095 days back, never an unwindowed full-history walk. - const fallbackArgs = findMany.mock.calls[1][0] as { - where: { measuredAt?: { gte?: Date } }; - }; - const gte = fallbackArgs.where.measuredAt?.gte; - expect(gte).toBeInstanceOf(Date); - expect(gte!.getTime()).toBe(now.getTime() - 1095 * dayMs); - }); - - it("day-buckets the dense-type fallback in SQL instead of a raw walk (v1.28.25)", async () => { - // BLOOD_GLUCOSE on a cold tier (heavy import before the rollup - // backfill warms) used to findMany the entire CGM history. The - // fallback now runs a single day-bucket aggregate; only the bounded - // recent read touches findMany. - findMany.mockResolvedValueOnce(dailyRows(80, now)); // bounded recent - const bucketRows: Array<{ bucket_start: Date; mean: number }> = []; - for (let i = 0; i < 800; i++) { - bucketRows.push({ - bucket_start: new Date(now.getTime() - i * dayMs), - mean: 100, - }); - } - queryRaw.mockResolvedValueOnce(bucketRows.reverse()); - - readBestGranularityRollups.mockResolvedValue(null); - - const series = await buildGradedSeriesWithRollups( - "u1", - "BLOOD_GLUCOSE", - now, - ); - expect(series.monthly.length).toBeGreaterThan(0); expect(series.yearly.length).toBeGreaterThan(0); - // Raw findMany ran once (the bounded recent read) — the fallback - // itself went through the SQL aggregate. - expect(findMany).toHaveBeenCalledTimes(1); - expect(queryRaw).toHaveBeenCalledTimes(1); + expect(readDayAggregates).toHaveBeenCalledTimes(2); + const fallback = readDayAggregates.mock + .calls[1][0] as ReadDayAggregatesOptions; + // Bounded to the yearly router's horizon, never an unwindowed walk. + expect(fallback.since.getTime()).toBe(now.getTime() - 1095 * dayMs); + expect(findMany).not.toHaveBeenCalled(); }); - it("reads monthly/yearly from the tier when it has coverage and skips the full read", async () => { - findMany.mockResolvedValueOnce(dailyRows(80, now)); - - // MONTH coverage for the 1-year window, YEAR coverage for the 3-year. + it("reads monthly/yearly from the tier when it has coverage and skips the fallback", async () => { + serveFrom(points(80, now)); readBestGranularityRollups.mockImplementation( - async (_u: string, _t: string, windowDays: number) => { - if (windowDays === 365) { - return { - granularity: "MONTH", - rows: [ - rollupRow(new Date("2025-09-01T00:00:00Z"), 79), - rollupRow(new Date("2025-10-01T00:00:00Z"), 80), - ], - }; - } - return { - granularity: "YEAR", - rows: [rollupRow(new Date("2023-01-01T00:00:00Z"), 82)], - }; - }, + async (_u: string, _t: string, windowDays: number) => + windowDays === 365 + ? { + granularity: "MONTH", + rows: [ + rollupRow(new Date("2025-09-01T00:00:00Z"), 79), + rollupRow(new Date("2025-10-01T00:00:00Z"), 80), + ], + } + : { + granularity: "YEAR", + rows: [rollupRow(new Date("2023-01-01T00:00:00Z"), 82)], + }, ); const series = await buildGradedSeriesWithRollups("u1", "WEIGHT", now); expect(series.monthly.length).toBeGreaterThan(0); expect(series.yearly.length).toBeGreaterThan(0); - // Only the bounded recent read — no full-history fallback when the - // tier covers both coarse slices. - expect(findMany).toHaveBeenCalledTimes(1); + expect(readDayAggregates).toHaveBeenCalledTimes(1); }); - it("falls back to a full read for only the slice the tier misses", async () => { - findMany - .mockResolvedValueOnce(dailyRows(80, now)) // bounded recent - .mockResolvedValueOnce(dailyRows(800, now)); // full-history fallback - - // MONTH covered, YEAR missing. + it("falls back only for the slice the tier misses", async () => { + serveFrom(points(800, now)); readBestGranularityRollups.mockImplementation( - async (_u: string, _t: string, windowDays: number) => { - if (windowDays === 365) { - return { - granularity: "MONTH", - rows: [rollupRow(new Date("2025-10-01T00:00:00Z"), 80)], - }; - } - return null; - }, + async (_u: string, _t: string, windowDays: number) => + windowDays === 365 + ? { + granularity: "MONTH", + rows: [rollupRow(new Date("2025-10-01T00:00:00Z"), 80)], + } + : null, ); const series = await buildGradedSeriesWithRollups("u1", "WEIGHT", now); - expect(series.monthly.length).toBeGreaterThan(0); + expect(series.monthly).toHaveLength(1); expect(series.yearly.length).toBeGreaterThan(0); - expect(findMany).toHaveBeenCalledTimes(2); + expect(readDayAggregates).toHaveBeenCalledTimes(2); + }); +}); + +describe("buildGradedSeriesFromDayAggregates — parity with the raw fold", () => { + const now = new Date("2026-05-31T12:00:00Z"); + const segmentStarts = [21, 91, 451].map( + (d) => new Date(now.getTime() - d * dayMs + 1), + ); + + it("matches the raw-row fold for recent / weekly / monthly and yearly stats", () => { + // Several readings per day, including readings exactly on the segment + // edges, so the partition itself is under test. + const rows = [ + ...points(900, now, 5), + { measuredAt: new Date(now.getTime() - 21 * dayMs), value: 99 }, + { measuredAt: new Date(now.getTime() - 91 * dayMs), value: 41 }, + ]; + const raw = buildGradedSeriesFromPoints(rows, now); + const agg = buildGradedSeriesFromDayAggregates( + foldDayAggregates(rows, { + since: new Date(0), + until: now, + timeZone: "Europe/Berlin", + segmentStarts, + }), + ); + expect(agg.recent).toEqual(raw.recent); + expect(agg.weekly).toEqual(raw.weekly); + expect(agg.monthly).toEqual(raw.monthly); + // The yearly slope runs over day means rather than single readings; + // every other yearly figure is exact. + expect(agg.yearly.map(({ slope: _s, ...b }) => b)).toEqual( + raw.yearly.map(({ slope: _s, ...b }) => b), + ); }); }); diff --git a/src/lib/insights/__tests__/metric-status.test.ts b/src/lib/insights/__tests__/metric-status.test.ts index d930ac00f..89f68e479 100644 --- a/src/lib/insights/__tests__/metric-status.test.ts +++ b/src/lib/insights/__tests__/metric-status.test.ts @@ -1,5 +1,13 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; +// The graded series folds per-day aggregates in SQL; the fake folds the +// mocked `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, diff --git a/src/lib/insights/__tests__/payload-budget.test.ts b/src/lib/insights/__tests__/payload-budget.test.ts index 11de84479..36c84eb29 100644 --- a/src/lib/insights/__tests__/payload-budget.test.ts +++ b/src/lib/insights/__tests__/payload-budget.test.ts @@ -12,6 +12,14 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; * `runStatusCompletion` is stubbed to capture it. */ +// The graded series folds per-day aggregates in SQL; the fake folds the +// mocked `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, diff --git a/src/lib/insights/__tests__/pulse-status-timeout.test.ts b/src/lib/insights/__tests__/pulse-status-timeout.test.ts index 508cff3b2..7ecb3c025 100644 --- a/src/lib/insights/__tests__/pulse-status-timeout.test.ts +++ b/src/lib/insights/__tests__/pulse-status-timeout.test.ts @@ -19,6 +19,14 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; * generation on every navigation while a provider is degraded. */ +// The graded series folds per-day aggregates in SQL; the fake folds the +// mocked `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, diff --git a/src/lib/insights/__tests__/pulse-status.test.ts b/src/lib/insights/__tests__/pulse-status.test.ts index 1aed06d5c..ddf2632b2 100644 --- a/src/lib/insights/__tests__/pulse-status.test.ts +++ b/src/lib/insights/__tests__/pulse-status.test.ts @@ -1,5 +1,13 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; +// The graded series folds per-day aggregates in SQL; the fake folds the +// mocked `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, diff --git a/src/lib/insights/__tests__/weight-status.test.ts b/src/lib/insights/__tests__/weight-status.test.ts index ec4e98a37..5f715f219 100644 --- a/src/lib/insights/__tests__/weight-status.test.ts +++ b/src/lib/insights/__tests__/weight-status.test.ts @@ -1,5 +1,13 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; +// The graded series folds per-day aggregates in SQL; the fake folds the +// mocked `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { auditLog: { findFirst: vi.fn(), create: vi.fn() }, diff --git a/src/lib/insights/derived/__tests__/baseline-plausibility.test.ts b/src/lib/insights/derived/__tests__/baseline-plausibility.test.ts index 7a83f5b61..37a894773 100644 --- a/src/lib/insights/derived/__tests__/baseline-plausibility.test.ts +++ b/src/lib/insights/derived/__tests__/baseline-plausibility.test.ts @@ -12,9 +12,27 @@ * The numbers below are shaped like the real thing: a resting-pulse series in * the 60s and 70s, one impossible sample dropped into it. */ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; -import { buildBaselineBand, dayMeansFromRows } from "../baseline"; +import type { ReadDayAggregatesOptions } from "@/lib/measurements/day-aggregates"; + +// The live day-mean read folds in SQL; the fake folds `rowsForRead` with the +// same day / range rules, so these tests see the filter the read asks for. +let rowsForRead: { value: number; measuredAt: Date }[] = []; +const seenOptions: ReadDayAggregatesOptions[] = []; +vi.mock("@/lib/db", () => ({ prisma: {} })); +vi.mock("@/lib/measurements/day-aggregates", async () => { + const { foldDayAggregates } = + await import("@/lib/measurements/__tests__/fake-day-aggregates"); + return { + readDayAggregates: async (opts: ReadDayAggregatesOptions) => { + seenOptions.push(opts); + return foldDayAggregates(rowsForRead, opts); + }, + }; +}); + +import { buildBaselineBand, readDayMeanSeries } from "../baseline"; import { latestDayMeanFromRows } from "../coincident-deviation"; import { isPlausibleMetricValue } from "@/lib/measurements/value-domain"; @@ -47,28 +65,49 @@ describe("metric plausibility domain", () => { }); }); +/** The live (rollup-miss) day-mean read over `rows`. */ +async function dayMeansFromRows( + rows: { value: number; measuredAt: Date }[], + type: "PULSE", +) { + rowsForRead = rows; + const { points } = await readDayMeanSeries( + "u1", + type, + 3650, + new Date("2026-09-01T00:00:00Z"), + new Map(), + ); + return points; +} + describe("per-day means behind the personal band", () => { - it("keeps the day mean on the real readings when one sample is impossible", () => { + it("asks the read for the metric's own plausibility domain", async () => { + await dayMeansFromRows(pulseDay("2026-08-01", [66]), "PULSE"); + expect(seenOptions.at(-1)?.valueRange).toEqual({ min: 20, max: 300 }); + }); + + it("keeps the day mean on the real readings when one sample is impossible", async () => { const rows = pulseDay("2026-08-01", [66, 70, 111287531.01, 68]); - const [point] = dayMeansFromRows(rows, "PULSE"); + const [point] = await dayMeansFromRows(rows, "PULSE"); expect(point.day).toBe("2026-08-01"); expect(point.mean).toBeCloseTo(68, 6); }); - it("drops a day whose every reading is impossible rather than inventing one", () => { + it("drops a day whose every reading is impossible rather than inventing one", async () => { const rows = [ ...pulseDay("2026-08-01", [66, 70, 68]), ...pulseDay("2026-08-02", [36016.75, 111287531.01]), ]; - const points = dayMeansFromRows(rows, "PULSE"); + const points = await dayMeansFromRows(rows, "PULSE"); expect(points.map((p) => p.day)).toEqual(["2026-08-01"]); }); - it("holds the band inside the metric's own domain across a poisoned month", () => { + it("holds the band inside the metric's own domain across a poisoned month", async () => { const rows: { value: number; measuredAt: Date }[] = []; for (let d = 1; d <= 28; d += 1) { const day = `2026-08-${String(d).padStart(2, "0")}`; @@ -81,7 +120,7 @@ describe("per-day means behind the personal band", () => { rows.push(...pulseDay("2026-08-26", [98211.5])); const band = buildBaselineBand( - dayMeansFromRows(rows, "PULSE").map((p) => p.mean), + (await dayMeansFromRows(rows, "PULSE")).map((p) => p.mean), "PULSE", ); diff --git a/src/lib/insights/derived/__tests__/baseline.test.ts b/src/lib/insights/derived/__tests__/baseline.test.ts index a4742a89a..dd30f888f 100644 --- a/src/lib/insights/derived/__tests__/baseline.test.ts +++ b/src/lib/insights/derived/__tests__/baseline.test.ts @@ -1,5 +1,13 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn() } }, })); diff --git a/src/lib/insights/derived/__tests__/coincident-deviation.test.ts b/src/lib/insights/derived/__tests__/coincident-deviation.test.ts index c101ef457..8c0fc800f 100644 --- a/src/lib/insights/derived/__tests__/coincident-deviation.test.ts +++ b/src/lib/insights/derived/__tests__/coincident-deviation.test.ts @@ -1,5 +1,13 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn() } }, })); diff --git a/src/lib/insights/derived/__tests__/dispatch-window-days.test.ts b/src/lib/insights/derived/__tests__/dispatch-window-days.test.ts index 1a6618097..c99ca3301 100644 --- a/src/lib/insights/derived/__tests__/dispatch-window-days.test.ts +++ b/src/lib/insights/derived/__tests__/dispatch-window-days.test.ts @@ -17,6 +17,14 @@ */ import { describe, it, expect, vi, beforeEach } from "vitest"; +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { diff --git a/src/lib/insights/derived/__tests__/dispatch.test.ts b/src/lib/insights/derived/__tests__/dispatch.test.ts index 267fc592c..02458a7ed 100644 --- a/src/lib/insights/derived/__tests__/dispatch.test.ts +++ b/src/lib/insights/derived/__tests__/dispatch.test.ts @@ -1,5 +1,13 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn().mockResolvedValue([]) }, diff --git a/src/lib/insights/derived/__tests__/readiness.test.ts b/src/lib/insights/derived/__tests__/readiness.test.ts index 0bb22411e..5371acb2f 100644 --- a/src/lib/insights/derived/__tests__/readiness.test.ts +++ b/src/lib/insights/derived/__tests__/readiness.test.ts @@ -1,5 +1,13 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn() }, diff --git a/src/lib/insights/derived/__tests__/trajectory.test.ts b/src/lib/insights/derived/__tests__/trajectory.test.ts index 705cfb00a..37c361161 100644 --- a/src/lib/insights/derived/__tests__/trajectory.test.ts +++ b/src/lib/insights/derived/__tests__/trajectory.test.ts @@ -1,5 +1,13 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn() } }, })); diff --git a/src/lib/insights/derived/baseline.ts b/src/lib/insights/derived/baseline.ts index 881ae202d..d6319880f 100644 --- a/src/lib/insights/derived/baseline.ts +++ b/src/lib/insights/derived/baseline.ts @@ -29,12 +29,13 @@ * composed-bucket baseline matches the raw-DAY baseline within tolerance. */ import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; -import { prisma } from "@/lib/db"; +import { readDayAggregates } from "@/lib/measurements/day-aggregates"; import { getAgeFromDateOfBirth } from "@/lib/analytics/pulse-targets"; import { toProfileSex } from "@/lib/profile/sex"; import { clampDerivedLowerBound, isPlausibleMetricValue, + plausibleMetricRange, } from "@/lib/measurements/value-domain"; import type { ProfileSex } from "@/lib/profile/sex"; import { @@ -212,39 +213,6 @@ export function buildBaselineBand( }; } -/** - * Group raw readings into per-day means, keeping only the values that sit - * inside the metric's own plausibility domain. - * - * Pure, and exported because this is the arithmetic the band rests on: a - * single stored value the application itself declares impossible drags the - * day's mean with it, the median then carries it into the band, and the - * personal "usual range" reads in the tens of thousands. Filtering here rather - * than at the edge of the card keeps the number wrong in exactly one place — - * nowhere — instead of right on one surface and wrong on the next. - * - * A day whose every reading is implausible produces no point at all. That is - * the honest outcome: the day contributed no measurement, and an absent day is - * something the coverage floor already knows how to report. - */ -export function dayMeansFromRows( - rows: readonly { value: number; measuredAt: Date }[], - type: MeasurementType, -): DayMeanPoint[] { - const byDay = new Map(); - for (const row of rows) { - if (!isPlausibleMetricValue(type, row.value)) continue; - const day = row.measuredAt.toISOString().slice(0, 10); - const acc = byDay.get(day) ?? { sum: 0, count: 0 }; - acc.sum += row.value; - acc.count += 1; - byDay.set(day, acc); - } - return [...byDay.entries()] - .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) - .map(([day, acc]) => ({ day, mean: acc.sum / acc.count })); -} - // ── reads ──────────────────────────────────────────────────────────── /** @@ -292,23 +260,22 @@ export async function readDayMeanSeries( // the spread is still DAY-native. } - // Per-type live fallback — bounded raw read, grouped into per-day - // means. Honest provenance: source = "live". + // Per-type live fallback — per-day means folded in SQL, so the read is + // one row per day however densely the metric is sampled (#1023). Only + // values inside the metric's own plausibility domain count: a single + // stored value the application declares impossible would drag the day's + // mean with it and the median would carry it into the band. A day whose + // every reading is implausible produces no point at all. Days are UTC + // days, matching the DAY rollup path above. const since = new Date(now.getTime() - windowDays * 24 * 60 * 60 * 1000); - const rows = await prisma.measurement.findMany({ - where: { - userId, - type, - deletedAt: null, - measuredAt: { gte: since }, - }, - orderBy: { measuredAt: "asc" }, - select: { value: true, measuredAt: true }, + const days = await readDayAggregates({ + userId, + type, + since, + timeZone: "UTC", + valueRange: plausibleMetricRange(type), }); - if (rows.length === 0) { - return { points: [], source: "none" }; - } - const points = dayMeansFromRows(rows, type); + const points = days.map((d) => ({ day: d.day, mean: d.sum / d.n })); if (points.length === 0) { return { points: [], source: "none" }; } diff --git a/src/lib/insights/graded-series.ts b/src/lib/insights/graded-series.ts index 0053d6d5d..cf31d08f9 100644 --- a/src/lib/insights/graded-series.ts +++ b/src/lib/insights/graded-series.ts @@ -48,12 +48,15 @@ */ import type { MeasurementType } from "@/generated/prisma/client"; -import { prisma } from "@/lib/db"; import { readBestGranularityRollups, type RollupBucketRow, } from "@/lib/rollups/measurement-read-wmy"; import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { + readDayAggregates, + type DayAggregateRow, +} from "@/lib/measurements/day-aggregates"; import { toBerlinYmd } from "@/lib/tz/resolver"; const MS_PER_DAY = 24 * 60 * 60 * 1000; @@ -83,18 +86,6 @@ const MONTHLY_WINDOW_MS = WEEKLY_WINDOW_MS + MONTHLY_MONTHS * 30 * MS_PER_DAY; */ const COLD_FALLBACK_WINDOW_DAYS = 1095; -/** - * Sample-dense types whose cold-tier fallback is day-bucketed in SQL - * (one AVG row per day) instead of read raw — same dense set the series - * route caps (CGM glucose ~288 rows/day, PULSE per-sample / hourly). - * The fold's output shape is unchanged; on the cold path only, a dense - * type's monthly/yearly min/max/n derive from day means rather than - * individual samples — the warm rollup tier restores sample-exact - * figures as soon as the backfill lands. - */ -const DENSE_FALLBACK_TYPES: ReadonlySet = - new Set(["BLOOD_GLUCOSE", "PULSE"]); - export interface RecentDayBucket { /** Berlin YYYY-MM-DD. */ date: string; @@ -171,6 +162,11 @@ function berlinYearKey(date: Date): string { */ function berlinIsoWeekKey(date: Date): string { const { year, month, day } = toBerlinYmd(date); + return isoWeekKeyOfYmd(year, month, day); +} + +/** ISO week key of a calendar date given as `YYYY`, `MM`, `DD`. */ +function isoWeekKeyOfYmd(year: string, month: string, day: string): string { const localMidnight = new Date( Date.UTC(Number(year), Number(month) - 1, Number(day)), ); @@ -309,6 +305,133 @@ export function buildGradedSeriesFromPoints( return { recent, weekly, monthly, yearly }; } +function mergeInto( + map: Map, + key: string, + row: DayAggregateRow, +): void { + const existing = map.get(key); + if (existing) { + existing.sum += row.sum; + existing.n += row.n; + if (row.min < existing.min) existing.min = row.min; + if (row.max > existing.max) existing.max = row.max; + } else { + map.set(key, { sum: row.sum, min: row.min, max: row.max, n: row.n }); + } +} + +/** Segment boundaries matching `buildGradedSeriesFromPoints`' age split. */ +function gradedSegmentStarts(now: Date): Date[] { + const nowMs = now.getTime(); + return [ + new Date(nowMs - RECENT_WINDOW_MS), + new Date(nowMs - WEEKLY_WINDOW_MS), + new Date(nowMs - MONTHLY_WINDOW_MS), + ]; +} + +/** + * Read the graded series' per-day aggregates for one type, folded in SQL. + * + * Replaces the raw `findMany` over the window (#1023): a watch that streams + * heart rate every few seconds put a million rows into that read, which is + * what exhausted the worker's heap. The rows here are one per Berlin day per + * age segment, and their sum / count / min / max are exact, so the recent, + * weekly and monthly slices come out identical to folding the raw rows. + * + * The segment boundaries use the same edges as `buildGradedSeriesFromPoints`: + * a row exactly `RECENT_WINDOW_MS` old is weekly there, so it is segment 1 + * here too. + */ +async function readGradedDayAggregates( + userId: string, + type: MeasurementType, + since: Date, + now: Date, +): Promise { + return readDayAggregates({ + userId, + type, + since, + // Future-dated readings were skipped by the in-memory fold. + until: now, + timeZone: "Europe/Berlin", + // `readDayAggregates` puts a row at segment k when it is older than k + // starts (`measured_at < start`). The in-memory fold treats an age of + // exactly the window as the older bucket, so shift each start by 1 ms. + segmentStarts: gradedSegmentStarts(now).map( + (d) => new Date(d.getTime() + 1), + ), + }); +} + +/** + * The graded series folded from per-day aggregates (segment = age class). + * + * recent / weekly / monthly and every bucket's min / max / mean / n match + * the raw-row fold exactly. The yearly slope is taken over the day means in + * date order rather than over individual readings: a direction signal either + * way, and the only form that stays bounded on a dense stream. + */ +export function buildGradedSeriesFromDayAggregates( + rows: readonly DayAggregateRow[], +): GradedSeries { + const recentAgg = new Map(); + const weeklyAgg = new Map(); + const monthlyAgg = new Map(); + const yearlyAgg = new Map(); + const yearlyValues = new Map(); + + const ordered = [...rows].sort((a, b) => + a.day < b.day ? -1 : a.day > b.day ? 1 : b.segment - a.segment, + ); + for (const row of ordered) { + const [year, month, day] = row.day.split("-"); + if (row.segment === 0) { + mergeInto(recentAgg, row.day, row); + } else if (row.segment === 1) { + mergeInto(weeklyAgg, isoWeekKeyOfYmd(year, month, day), row); + } else if (row.segment === 2) { + mergeInto(monthlyAgg, `${year}-${month}`, row); + } else { + mergeInto(yearlyAgg, year, row); + const list = yearlyValues.get(year); + const dayMean = row.sum / row.n; + if (list) list.push(dayMean); + else yearlyValues.set(year, [dayMean]); + } + } + + const toBuckets = (map: Map) => + Array.from(map.entries()) + .map(([key, a]) => ({ + key, + min: round(a.min), + max: round(a.max), + mean: round(a.sum / a.n), + n: a.n, + })) + .sort((x, y) => x.key.localeCompare(y.key)); + + return { + recent: toBuckets(recentAgg).map(({ key, ...b }) => ({ date: key, ...b })), + weekly: toBuckets(weeklyAgg).map(({ key, ...b }) => ({ + weekISO: key, + ...b, + })), + monthly: toBuckets(monthlyAgg).map(({ key, ...b }) => ({ + month: key, + ...b, + })), + yearly: toBuckets(yearlyAgg).map(({ key, ...b }) => ({ + year: key, + ...b, + slope: leastSquaresSlope(yearlyValues.get(key) ?? []), + })), + }; +} + /** * Project a rollup bucket row to a monthly/yearly graded bucket. The * tier already carries min/max/mean/slope per bucket — no JS folding. @@ -335,24 +458,25 @@ function rollupYearly(rows: RollupBucketRow[]): YearlyBucket[] { } /** - * Build the graded series with the recent / weekly slices folded from a - * bounded raw read and the monthly / yearly slices read from the + * Build the graded series with the recent / weekly slices folded from + * per-day aggregates and the monthly / yearly slices read from the * pre-aggregated rollup tier. * - * The recent / weekly read is capped to the recent + weekly horizon - * (last ~90 days), so the common warm-tier path never pulls the full - * multi-year history into memory. The rollup router picks MONTH for the - * ~1-year window and YEAR for the multi-year tail. + * The recent / weekly read covers the recent + weekly horizon (last ~90 + * days) and is folded in SQL to one row per day, so its size depends on + * the number of days, never on how densely a device samples (#1023). The + * rollup router picks MONTH for the ~1-year window and YEAR for the + * multi-year tail. * * Coverage-miss fallback: when the tier has no MONTH (resp. YEAR) * coverage for a metric — a fresh account the boot-backfill has not yet * caught up on — the bounded ~90-day read can NOT supply the monthly / * yearly slices (every row in it folds into recent / weekly), so naively * reusing it would ship empty monthly / yearly arrays even when years of - * raw history exist. On that miss this falls back to an in-memory fold - * over a bounded window (1095 days — the yearly router's own horizon; - * sample-dense types are additionally day-bucketed in SQL) so the coarse - * slices are populated from the raw rows rather than left falsely empty. + * raw history exist. On that miss this falls back to the same per-day + * aggregate over a bounded window (1095 days — the yearly router's own + * horizon) so the coarse slices are populated from the stored rows rather + * than left falsely empty. * The fallback read only happens when the tier is cold, which is exactly * the case the write-amplified tier was meant to avoid — so the hot path * stays bounded and the cold path stays correct. @@ -363,15 +487,8 @@ export async function buildGradedSeriesWithRollups( now: Date, ): Promise { const since = new Date(now.getTime() - WEEKLY_WINDOW_MS); - const rawRecent = await prisma.measurement.findMany({ - where: { userId, type, deletedAt: null, measuredAt: { gte: since } }, - orderBy: { measuredAt: "asc" }, - select: { measuredAt: true, value: true }, - }); - - const recentGraded = buildGradedSeriesFromPoints( - rawRecent.map((r) => ({ measuredAt: r.measuredAt, value: r.value })), - now, + const recentGraded = buildGradedSeriesFromDayAggregates( + await readGradedDayAggregates(userId, type, since, now), ); // v1.11.2 — load the source-priority blob once and thread it into both @@ -420,57 +537,19 @@ export async function buildGradedSeriesWithRollups( } // Tier coverage miss for one or both coarse slices: the bounded - // ~90-day read can't supply them, so fold a BOUNDED history window in - // memory and take whichever slices the tier left empty. The fallback - // read is the exception (cold-tier accounts), never the warm-tier - // norm. v1.28.25 — the read is bounded to the same 1095-day horizon - // the yearly rollup router serves (it was previously unwindowed), and - // the sample-dense types are day-bucketed in SQL so a heavy import - // ahead of the rollup backfill can't drag six-figure row sets into JS. + // ~90-day read can't supply them, so fold a BOUNDED history window and + // take whichever slices the tier left empty. The fallback read is the + // exception (cold-tier accounts), never the warm-tier norm. It is + // bounded to the same 1095-day horizon the yearly rollup router serves, + // and folded per day in SQL for every type, so a heavy import ahead of + // the rollup backfill can't drag six-figure row sets into JS. if (!monthlyCovered || !yearlyCovered) { const fallbackSince = new Date( now.getTime() - COLD_FALLBACK_WINDOW_DAYS * MS_PER_DAY, ); - let fallbackPoints: Point[]; - if (DENSE_FALLBACK_TYPES.has(type)) { - // Parameter-bound day-bucket aggregate, mirroring the rollup tier's - // `date_trunc(...) GROUP BY` (measurement-rollups.ts) — at most - // ~1095 rows regardless of sample density. - const bucketRows = await prisma.$queryRaw< - Array<{ bucket_start: Date; mean: number }> - >` - SELECT - date_trunc('day', m."measured_at") AS bucket_start, - AVG(m."value")::double precision AS mean - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."type" = ${type}::"measurement_type" - AND m."measured_at" >= ${fallbackSince} - AND m."deleted_at" IS NULL - GROUP BY date_trunc('day', m."measured_at") - ORDER BY bucket_start ASC - `; - fallbackPoints = bucketRows.map((r) => ({ - measuredAt: r.bucket_start, - value: r.mean, - })); - } else { - const rows = await prisma.measurement.findMany({ - where: { - userId, - type, - deletedAt: null, - measuredAt: { gte: fallbackSince }, - }, - orderBy: { measuredAt: "asc" }, - select: { measuredAt: true, value: true }, - }); - fallbackPoints = rows.map((r) => ({ - measuredAt: r.measuredAt, - value: r.value, - })); - } - const fullGraded = buildGradedSeriesFromPoints(fallbackPoints, now); + const fullGraded = buildGradedSeriesFromDayAggregates( + await readGradedDayAggregates(userId, type, fallbackSince, now), + ); if (!monthlyCovered) monthly = fullGraded.monthly; if (!yearlyCovered) yearly = fullGraded.yearly; } diff --git a/src/lib/jobs/coach-plan-review.ts b/src/lib/jobs/coach-plan-review.ts index 1b753f1a1..acf500618 100644 --- a/src/lib/jobs/coach-plan-review.ts +++ b/src/lib/jobs/coach-plan-review.ts @@ -24,6 +24,7 @@ */ import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; import { encryptToBytes } from "@/lib/ai/coach/bytes-codec"; +import { readDayAggregates } from "@/lib/measurements/day-aggregates"; export const COACH_PLAN_REVIEW_QUEUE = "coach-plan-review"; // Daily at 05:25 Europe/Berlin — just after the 05:20 reminder sweep, so the @@ -49,7 +50,7 @@ export interface CoachPlanReviewSummary { type ReviewPrisma = Pick< PrismaClient, - "coachReminder" | "coachPlan" | "measurement" + "coachReminder" | "coachPlan" | "$queryRaw" >; /** Metric valence — whether a higher reading is the better outcome. */ @@ -282,31 +283,22 @@ export async function runCoachPlanReviewTick( const since = new Date( plan.createdAt.getTime() - BEFORE_WINDOW_DAYS * MS_PER_DAY, ); - const rows = await prisma.measurement.findMany({ - where: { - userId: plan.userId, - type: resolved.type, - deletedAt: null, - measuredAt: { gte: since, lte: now }, - }, - orderBy: { measuredAt: "asc" }, - select: { value: true, measuredAt: true }, + // Day means folded in SQL (UTC days), one row per day however + // densely the metric is sampled (#1023). + const byDay = await readDayAggregates({ + userId: plan.userId, + type: resolved.type, + since, + until: now, + timeZone: "UTC", + db: prisma, }); - // Day-mean the raw rows, then split at the experiment start. - const byDay = new Map(); - for (const row of rows) { - const day = row.measuredAt.toISOString().slice(0, 10); - const agg = byDay.get(day) ?? { sum: 0, count: 0 }; - agg.sum += row.value; - agg.count += 1; - byDay.set(day, agg); - } const startKey = plan.createdAt.toISOString().slice(0, 10); const before: number[] = []; const after: number[] = []; const all: number[] = []; - for (const [day, agg] of byDay) { - const dayMean = agg.sum / agg.count; + for (const { day, sum, n } of byDay) { + const dayMean = sum / n; all.push(dayMean); if (day < startKey) before.push(dayMean); else after.push(dayMean); diff --git a/src/lib/measurements/__tests__/fake-day-aggregates.ts b/src/lib/measurements/__tests__/fake-day-aggregates.ts new file mode 100644 index 000000000..311e8350c --- /dev/null +++ b/src/lib/measurements/__tests__/fake-day-aggregates.ts @@ -0,0 +1,73 @@ +/** + * In-memory stand-in for `readDayAggregates`, for unit tests that mock + * `@/lib/db`. + * + * It reads the rows through the mocked `prisma.measurement.findMany` (so a + * test's existing row fixture keeps feeding the reader) and folds them with + * the same day / segment / range rules the SQL applies. The integration test + * `tests/integration/dense-read-bounded.test.ts` pins that this fold and the + * SQL agree on real Postgres. + */ +import { prisma } from "@/lib/db"; +import type { + DayAggregateRow, + ReadDayAggregatesOptions, +} from "@/lib/measurements/day-aggregates"; +import { userDayKey } from "@/lib/tz/format"; + +export function foldDayAggregates( + rows: ReadonlyArray<{ measuredAt: Date; value: number }>, + opts: Omit, +): DayAggregateRow[] { + const starts = opts.segmentStarts ?? []; + const byKey = new Map(); + for (const r of rows) { + const t = r.measuredAt.getTime(); + if (t < opts.since.getTime()) continue; + if (opts.until && t > opts.until.getTime()) continue; + if ( + opts.valueRange && + (r.value < opts.valueRange.min || r.value > opts.valueRange.max) + ) { + continue; + } + const day = userDayKey(r.measuredAt, opts.timeZone); + const segment = starts.filter((s) => t < s.getTime()).length; + const key = `${day}|${segment}`; + const acc = byKey.get(key); + if (acc) { + acc.n += 1; + acc.sum += r.value; + if (r.value < acc.min) acc.min = r.value; + if (r.value > acc.max) acc.max = r.value; + } else { + byKey.set(key, { + day, + segment, + n: 1, + sum: r.value, + min: r.value, + max: r.value, + }); + } + } + return [...byKey.values()].sort((a, b) => + a.day < b.day ? -1 : a.day > b.day ? 1 : b.segment - a.segment, + ); +} + +export async function fakeReadDayAggregates( + opts: ReadDayAggregatesOptions, +): Promise { + const rows = ((await prisma.measurement.findMany({ + where: { + userId: opts.userId, + type: opts.type, + deletedAt: null, + measuredAt: { gte: opts.since, ...(opts.until && { lte: opts.until }) }, + }, + orderBy: { measuredAt: "asc" }, + select: { measuredAt: true, value: true }, + })) ?? []) as Array<{ measuredAt: Date; value: number }>; + return foldDayAggregates(rows, opts); +} diff --git a/src/lib/measurements/day-aggregates.ts b/src/lib/measurements/day-aggregates.ts new file mode 100644 index 000000000..083b59b78 --- /dev/null +++ b/src/lib/measurements/day-aggregates.ts @@ -0,0 +1,122 @@ +/** + * Per-day aggregates of one measurement type, folded in Postgres. + * + * Several background readers (the status cards' graded series, the vitals + * baseline, the tiered context, the plan review) used to `findMany` every raw + * row in a time window and then fold them into days in JavaScript. The window + * bounded the time span but not the row count, so an account whose watch + * streams heart rate every few seconds materialised hundreds of thousands to + * millions of objects per read. In the worker that was enough to exhaust the + * heap, and the read itself outran the statement timeout. + * + * This reader returns at most one row per (day, segment), whatever the sample + * density: the fold runs in SQL and only the aggregates cross the wire. `sum`, + * `n`, `min` and `max` are exact, so every consumer that averaged, counted or + * took extremes over the raw rows gets the same numbers from these rows. + * + * Days are cut in `timeZone`. `segmentStarts` optionally splits the window at + * instants (newest first): a row's segment is the number of starts it lies + * before, so a caller that treated "the last 21 days" and "the 70 days before + * that" differently keeps exactly the same per-row partition. + */ +import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import { Prisma } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { isValidTimezone } from "@/lib/tz/format"; + +export interface DayAggregateRow { + /** Calendar day in the requested zone, `YYYY-MM-DD`. */ + day: string; + /** Segment index: 0 for rows at or after `segmentStarts[0]`, and so on. */ + segment: number; + n: number; + sum: number; + min: number; + max: number; +} + +export interface ReadDayAggregatesOptions { + userId: string; + type: MeasurementType; + /** Inclusive lower bound on `measuredAt`. */ + since: Date; + /** Inclusive upper bound on `measuredAt`; omit for no upper bound. */ + until?: Date; + /** IANA zone the day boundary is cut in. */ + timeZone: string; + /** Rows outside `[min, max]` are dropped before the fold. */ + valueRange?: { min: number; max: number }; + /** Segment boundaries, newest first. */ + segmentStarts?: readonly Date[]; + /** Client to read through; defaults to the shared one. */ + db?: Pick; +} + +export async function readDayAggregates( + opts: ReadDayAggregatesOptions, +): Promise { + const timeZone = isValidTimezone(opts.timeZone) ? opts.timeZone : "UTC"; + const until = opts.until + ? Prisma.sql`AND m."measured_at" <= ${opts.until}` + : Prisma.empty; + const range = opts.valueRange + ? Prisma.sql`AND m."value" >= ${opts.valueRange.min} AND m."value" <= ${opts.valueRange.max}` + : Prisma.empty; + const starts = opts.segmentStarts ?? []; + const segment = + starts.length === 0 + ? Prisma.sql`0` + : Prisma.sql`(${Prisma.join( + starts.map( + (s) => + Prisma.sql`(CASE WHEN m."measured_at" < ${s} THEN 1 ELSE 0 END)`, + ), + " + ", + )})`; + + // The day and segment are computed once per row in `src` and grouped by + // name, so the bound zone / boundary parameters never have to match + // themselves across SELECT and GROUP BY. + const rows = await (opts.db ?? prisma).$queryRaw< + Array<{ + day: string; + segment: number; + n: number; + sum: number; + min: number; + max: number; + }> + >` + WITH src AS ( + SELECT + to_char((m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}, 'YYYY-MM-DD') AS day, + ${segment} AS segment, + m."value" AS value + FROM measurements m + WHERE m."user_id" = ${opts.userId} + AND m."type" = ${opts.type}::measurement_type + AND m."deleted_at" IS NULL + AND m."measured_at" >= ${opts.since} + ${until} + ${range} + ) + SELECT + day, + segment::int AS segment, + COUNT(*)::int AS n, + SUM(value)::double precision AS sum, + MIN(value)::double precision AS min, + MAX(value)::double precision AS max + FROM src + GROUP BY day, segment + ORDER BY day ASC, segment DESC + `; + return rows.map((r) => ({ + day: r.day, + segment: Number(r.segment), + n: Number(r.n), + sum: Number(r.sum), + min: Number(r.min), + max: Number(r.max), + })); +} diff --git a/src/lib/measurements/value-domain.ts b/src/lib/measurements/value-domain.ts index fdfad09c4..d6e485827 100644 --- a/src/lib/measurements/value-domain.ts +++ b/src/lib/measurements/value-domain.ts @@ -105,12 +105,22 @@ export function isPlausibleMetricValue( value: number, ): boolean { if (!Number.isFinite(value)) return false; - if (!type) return true; - const range = VALUE_RANGES[type]; + const range = plausibleMetricRange(type); if (range === undefined) return true; return value >= range.min && value <= range.max; } +/** + * The plausibility domain of a metric, for a reader that filters in SQL + * rather than row by row (`readDayAggregates`' `valueRange`). `undefined` + * when the metric declares none, which admits every value. + */ +export function plausibleMetricRange( + type: string | null | undefined, +): { min: number; max: number } | undefined { + return type ? VALUE_RANGES[type] : undefined; +} + /** * True when EVERY named metric forbids negative values (and at least one was * named). Used by the chart, which can paint several series in one axis: a diff --git a/src/lib/rollups/__tests__/tiered-context.test.ts b/src/lib/rollups/__tests__/tiered-context.test.ts index 4998abc97..64610312b 100644 --- a/src/lib/rollups/__tests__/tiered-context.test.ts +++ b/src/lib/rollups/__tests__/tiered-context.test.ts @@ -23,6 +23,14 @@ const mocks = vi.hoisted(() => ({ queryRaw: vi.fn(), })); +// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadDayAggregates, +})); + vi.mock("../measurement-rollups", () => ({ readRollupBuckets: mocks.readRollupBuckets, ensureUserRollupsFresh: mocks.ensureUserRollupsFresh, diff --git a/src/lib/rollups/tiered-context.ts b/src/lib/rollups/tiered-context.ts index 1546f3460..fdf43b62c 100644 --- a/src/lib/rollups/tiered-context.ts +++ b/src/lib/rollups/tiered-context.ts @@ -33,12 +33,13 @@ import type { MeasurementType, RollupGranularity, } from "@/generated/prisma/client"; -import { isNearUtc, userDayKey } from "@/lib/tz/format"; +import { isNearUtc } from "@/lib/tz/format"; import { ensureUserRollupsFresh, readRollupBuckets, } from "./measurement-rollups"; import { probeRollupCoverage } from "./measurement-coverage"; +import { readDayAggregates } from "@/lib/measurements/day-aggregates"; const DAY_MS = 24 * 60 * 60 * 1000; @@ -112,7 +113,7 @@ export interface TieredSeries { export interface BuildTieredSeriesOptions { /** * The user's IANA timezone. REQUIRED — the raw 0–14d band keys its - * daily fold on the user's local calendar day (`userDayKey`) so a + * daily fold on the user's local calendar day so a * late-evening reading lands on the right day across DST boundaries * and for non-UTC users. The coarser DAY/WEEK/MONTH/YEAR bands read * straight from the rollup table, whose `bucketStart` is UTC-anchored @@ -244,8 +245,8 @@ function extractAnomalies( } /** - * Read the raw 0–14d daily points for one type, folding same-day rows. - * Days are keyed on the user's local calendar via `userDayKey(tz)`, so a + * Read the 0–14d daily points for one type, folding same-day rows. + * Days are keyed on the user's local calendar in `tz`, so a * 23:30-local reading folds onto its local day rather than the UTC day — * correct across DST transitions and for non-UTC users. */ @@ -256,26 +257,19 @@ async function readRecentDaily( tz: string, ): Promise { const since = new Date(now - TIERED_BANDS.rawDays * DAY_MS); - const rows = await prisma.measurement.findMany({ - where: { userId, type, measuredAt: { gte: since }, deletedAt: null }, - select: { value: true, measuredAt: true }, - orderBy: { measuredAt: "asc" }, + // Folded per local day in SQL: one row per day however densely the type + // is sampled, never the raw stream (#1023). + const days = await readDayAggregates({ + userId, + type, + since, + timeZone: tz, }); - const byDay = new Map(); - for (const r of rows) { - const key = userDayKey(r.measuredAt, tz); - const entry = byDay.get(key) ?? { sum: 0, count: 0 }; - entry.sum += r.value; - entry.count += 1; - byDay.set(key, entry); - } - return [...byDay.entries()] - .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) - .map(([date, { sum, count }]) => ({ - date, - value: round(sum / count), - count, - })); + return days.map((d) => ({ + date: d.day, + value: round(d.sum / d.n), + count: d.n, + })); } /** diff --git a/tests/integration/dense-read-bounded.test.ts b/tests/integration/dense-read-bounded.test.ts new file mode 100644 index 000000000..009a566a6 --- /dev/null +++ b/tests/integration/dense-read-bounded.test.ts @@ -0,0 +1,209 @@ +/** + * #1023 — a status card over a dense heart-rate stream, measured. + * + * A single account syncing heart rate from a watch killed the worker within + * half an hour: the pulse status card's recent-window read was bounded in + * time (91 days) but not in rows, so every refresh materialised the whole + * stream as objects, and the read itself ran past the statement timeout. + * The read now folds per day in Postgres. + * + * The fixture is one reading a minute for 91 days, 131 040 rows, seeded in + * one `INSERT … generate_series`. Each half of the memory test reads the + * same window: the per-day fold the card now uses, and the raw `findMany` it + * used to issue. Without the second half the budget could be any number and + * the first half would still pass. Every reading of the heap is taken after a + * forced collection, so what is compared is what each read HOLDS; the + * transient figure (taken before collecting) is reported alongside. + * + * The parity case pins that the SQL fold and the in-memory fold the unit + * tests use agree on real Postgres: same days in a non-UTC zone, same + * segment edges, same plausibility filter. + */ +import v8 from "node:v8"; +import vm from "node:vm"; + +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { buildGradedSeriesWithRollups } from "@/lib/insights/graded-series"; +import { readDayAggregates } from "@/lib/measurements/day-aggregates"; +import { foldDayAggregates } from "@/lib/measurements/__tests__/fake-day-aggregates"; +import { localHmAsUtc } from "@/lib/tz/local-day"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const prisma = getPrismaClient(); +const DENSE_USER = "dense-pulse-owner"; +const PARITY_USER = "dense-parity-owner"; +const DAYS = 91; +const ROWS = DAYS * 24 * 60; +const NOW = new Date("2026-09-20T18:00:00.000Z"); + +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(`[dense-read] ${line}\n`); +} + +/** Run `read`, returning what it holds, its transient peak and its time. */ +async function measure(read: () => Promise): Promise<{ + result: T; + heldBytes: number; + transientBytes: number; + ms: number; +}> { + const before = liveHeapBytes(); + const t0 = performance.now(); + const result = await read(); + const ms = performance.now() - t0; + const transientBytes = process.memoryUsage().heapUsed - before; + const heldBytes = liveHeapBytes() - before; + return { result, heldBytes, transientBytes, ms }; +} + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: DENSE_USER, + username: DENSE_USER, + timezone: "Europe/Berlin", + }, + }); + await prisma.$executeRawUnsafe( + `INSERT INTO measurements ( + id, user_id, type, value, unit, source, measured_at, + created_at, updated_at, sync_version) + SELECT + 'dp' || lpad(g::text, 8, '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(1, $3::int) g`, + DENSE_USER, + NOW.toISOString(), + ROWS, + ); + await prisma.$executeRawUnsafe(`ANALYZE measurements`); +}, 300_000); + +afterAll(async () => { + await truncateAllTables(prisma); +}); + +describe("pulse status over a dense stream (#1023)", () => { + it("builds the graded series without materialising the stream", async () => { + const graded = await measure(() => + buildGradedSeriesWithRollups(DENSE_USER, "PULSE", NOW), + ); + const series = graded.result; + // Every reading in the 91-day window is accounted for, one bucket per day + // or week, not one object per reading. The oldest reading sits exactly 91 + // days back, which the graded split files under the older slice. + const counted = [...series.recent, ...series.weekly].reduce( + (sum, b) => sum + b.n, + 0, + ); + expect(counted).toBe(ROWS - 1); + expect(series.recent.length).toBeLessThanOrEqual(22); + expect(series.weekly.length).toBeLessThanOrEqual(12); + + const raw = await measure(() => + prisma.measurement.findMany({ + where: { + userId: DENSE_USER, + type: "PULSE", + deletedAt: null, + measuredAt: { gte: new Date(NOW.getTime() - DAYS * 86_400_000) }, + }, + orderBy: { measuredAt: "asc" }, + select: { measuredAt: true, value: true }, + }), + ); + expect(raw.result.length).toBe(ROWS); + + report( + `graded series (per-day fold, cold tier): held ${mb(graded.heldBytes)} MB, ` + + `transient ${mb(graded.transientBytes)} MB, ${graded.ms.toFixed(0)} ms`, + ); + report( + `raw 91-day findMany (the old read): held ${mb(raw.heldBytes)} MB, ` + + `transient ${mb(raw.transientBytes)} MB, ${raw.ms.toFixed(0)} ms, ` + + `${raw.result.length} rows`, + ); + + // The fold holds a few kilobytes; the raw read holds the stream. The + // transient figure is the one that killed the worker: the old builder + // held little once it returned but allocated the whole stream on the + // way (62 MB on this fixture, against 1.6 MB for the fold). + expect(graded.heldBytes).toBeLessThan(2 * 1024 * 1024); + expect(graded.transientBytes).toBeLessThan(12 * 1024 * 1024); + expect(raw.heldBytes).toBeGreaterThan(20 * Math.max(graded.heldBytes, 1)); + }, 120_000); + + it("the SQL fold agrees with the in-memory fold on real Postgres", async () => { + const tz = "Asia/Kolkata"; + await prisma.user.create({ + data: { id: PARITY_USER, username: PARITY_USER, timezone: tz }, + }); + const day = new Date("2026-09-10T12:00:00Z"); + const edge = new Date(NOW.getTime() - 21 * 86_400_000); + const rows = [ + { at: localHmAsUtc(day, tz, 0, 5), value: 61 }, + { at: localHmAsUtc(day, tz, 23, 55), value: 99 }, + { at: localHmAsUtc(day, tz, 12, 0), value: 111_287_531 }, // impossible + { at: edge, value: 70 }, + { at: new Date(edge.getTime() + 1), value: 72 }, + { at: new Date(NOW.getTime() - 200 * 86_400_000), value: 58 }, + ]; + await prisma.measurement.createMany({ + data: rows.map((r, i) => ({ + id: `parity-${i}`, + userId: PARITY_USER, + type: "PULSE" as const, + value: r.value, + unit: "bpm", + source: "MANUAL" as const, + measuredAt: r.at, + })), + }); + const opts = { + since: new Date(NOW.getTime() - 400 * 86_400_000), + until: NOW, + timeZone: tz, + valueRange: { min: 20, max: 300 }, + segmentStarts: [edge, new Date(NOW.getTime() - 91 * 86_400_000)], + }; + const sql = await readDayAggregates({ + userId: PARITY_USER, + type: "PULSE", + ...opts, + }); + const memory = foldDayAggregates( + rows.map((r) => ({ measuredAt: r.at, value: r.value })), + opts, + ); + expect(sql).toEqual(memory); + // Sanity: the impossible value was dropped and the edge row split off. + expect(sql.reduce((n, r) => n + r.n, 0)).toBe(5); + }); +});