From c41df33dd361426716a8da184cc44ce19729c2a4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 24 Sep 2026 08:06:47 +0200 Subject: [PATCH] fix(cycle): read the phase crosstab as per-source day aggregates The coach's cycle snapshot read a year of raw readings for the phase crosstab, including steps, heart-rate variability and CGM glucose. On a CGM account that is over 100 000 objects per snapshot. The crosstab only uses each day's sum and count after picking one source and device per day, and the pick depends only on which sources and devices are present. readSourceDayAggregates returns one row per type, day, source and device, and the day map now carries each row's reading count, so the day values are the ones the raw readings give. The 30-day clinical glucose reads stay raw: time in range, SD, J-index and the risk indices are functions of individual readings, and a sensor's rate bounds the window below 45 000 rows. Refs #1023 --- .../dense-measurement-read-guard.test.ts | 6 +- .../ai/coach/__tests__/cycle-snapshot.test.ts | 8 ++ src/lib/ai/coach/cycle-snapshot.ts | 31 +++-- src/lib/insights/mood-aggregates.ts | 6 + src/lib/insights/mood-crosstab.ts | 3 +- .../__tests__/fake-day-aggregates.ts | 59 +++++++++ src/lib/measurements/day-aggregates.ts | 85 ++++++++++++- .../cycle-crosstab-day-aggregates.test.ts | 114 ++++++++++++++++++ 8 files changed, 289 insertions(+), 23 deletions(-) create mode 100644 tests/integration/cycle-crosstab-day-aggregates.test.ts diff --git a/src/__tests__/dense-measurement-read-guard.test.ts b/src/__tests__/dense-measurement-read-guard.test.ts index d669c99d4..8414051a4 100644 --- a/src/__tests__/dense-measurement-read-guard.test.ts +++ b/src/__tests__/dense-measurement-read-guard.test.ts @@ -74,11 +74,9 @@ const ALLOWED: Record = { "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.", + "Clinical CGM metrics (time in range, SD / CV, J-index, LBGI / HBGI, reading span) are functions of the individual readings, which a per-day aggregate cannot reproduce. 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.", + "Same 30-day clinical glucose panel as the coach snapshot (per-reading metrics), grouped by meal context.", "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.", }; diff --git a/src/lib/ai/coach/__tests__/cycle-snapshot.test.ts b/src/lib/ai/coach/__tests__/cycle-snapshot.test.ts index 7ac766bf2..25869b76d 100644 --- a/src/lib/ai/coach/__tests__/cycle-snapshot.test.ts +++ b/src/lib/ai/coach/__tests__/cycle-snapshot.test.ts @@ -10,6 +10,14 @@ vi.mock("@/lib/db", () => ({ }, })); +// The crosstab reads per-source day aggregates; the fake groups the mocked +// `measurement.findMany` rows with the same rules. +vi.mock("@/lib/measurements/day-aggregates", async () => ({ + readSourceDayAggregates: ( + await import("@/lib/measurements/__tests__/fake-day-aggregates") + ).fakeReadSourceDayAggregates, +})); + import { buildCycleSnapshotBlock } from "../cycle-snapshot"; import { prisma } from "@/lib/db"; diff --git a/src/lib/ai/coach/cycle-snapshot.ts b/src/lib/ai/coach/cycle-snapshot.ts index c4f0b4cad..297496213 100644 --- a/src/lib/ai/coach/cycle-snapshot.ts +++ b/src/lib/ai/coach/cycle-snapshot.ts @@ -40,6 +40,7 @@ import { } from "@/lib/cycle/phase-crosstab"; import type { CrossMetricMeasurement } from "@/lib/insights/mood-aggregates"; import { addDays } from "@/lib/cycle/day-math"; +import { readSourceDayAggregates } from "@/lib/measurements/day-aggregates"; import { DEFAULT_TIMEZONE, moodDateKey } from "@/lib/mood/date-key"; /** Trailing window the phase contrast walks (days). Mirrors the read route. */ @@ -195,21 +196,16 @@ export async function buildCycleSnapshotBlock( orderBy: { measuredAt: "asc" }, select: { measuredAt: true, value: true }, }), - prisma.measurement.findMany({ - where: { - userId, - deletedAt: null, - type: { in: PHASE_CROSSTAB_METRIC_TYPES }, - measuredAt: { gte: new Date(Date.parse(`${from}T00:00:00Z`)) }, - }, - orderBy: { measuredAt: "asc" }, - select: { - type: true, - value: true, - measuredAt: true, - source: true, - deviceType: true, - }, + // A year of steps, heart-rate variability and CGM glucose is far too + // many readings to hold as objects (#1023). The crosstab only needs + // each day's sum and count per source and device, which is what the + // per-source picker decides on, so the fold runs in SQL. Days are the + // Berlin days `metricDayMap` keys on. + readSourceDayAggregates({ + userId, + types: PHASE_CROSSTAB_METRIC_TYPES, + since: new Date(Date.parse(`${from}T00:00:00Z`)), + timeZone: "Europe/Berlin", }), prisma.user.findUnique({ where: { id: userId }, @@ -285,8 +281,9 @@ export async function buildCycleSnapshotBlock( ); const measurements: CrossMetricMeasurement[] = measurementRows.map((m) => ({ type: m.type, - value: m.value, - measuredAt: m.measuredAt, + value: m.sum, + count: m.n, + measuredAt: m.firstAt, source: m.source, deviceType: m.deviceType, })); diff --git a/src/lib/insights/mood-aggregates.ts b/src/lib/insights/mood-aggregates.ts index ed2c0a633..a6e634cd9 100644 --- a/src/lib/insights/mood-aggregates.ts +++ b/src/lib/insights/mood-aggregates.ts @@ -254,6 +254,12 @@ export interface CrossMetricMeasurement { measuredAt: Date; source?: MeasurementSource | null; deviceType?: string | null; + /** + * Readings this row stands for. Absent on a raw reading (one). A row + * pre-aggregated per day, source and device carries its readings' sum in + * `value` and their number here, so the day's mean stays exact. + */ + count?: number; } /** diff --git a/src/lib/insights/mood-crosstab.ts b/src/lib/insights/mood-crosstab.ts index d349febb6..181754cad 100644 --- a/src/lib/insights/mood-crosstab.ts +++ b/src/lib/insights/mood-crosstab.ts @@ -231,6 +231,7 @@ export function metricDayMap( deviceType: m.deviceType ?? null, type: type as MeasurementType, value: m.value, + count: m.count, })), metricKey, userPriorityJson, @@ -243,7 +244,7 @@ export function metricDayMap( const key = berlinDayKey(m.measuredAt); const cur = byDay.get(key) ?? { sum: 0, count: 0 }; cur.sum += m.value; - cur.count += 1; + cur.count += m.count ?? 1; byDay.set(key, cur); } const out = new Map(); diff --git a/src/lib/measurements/__tests__/fake-day-aggregates.ts b/src/lib/measurements/__tests__/fake-day-aggregates.ts index 311e8350c..096727fb9 100644 --- a/src/lib/measurements/__tests__/fake-day-aggregates.ts +++ b/src/lib/measurements/__tests__/fake-day-aggregates.ts @@ -9,9 +9,14 @@ * SQL agree on real Postgres. */ import { prisma } from "@/lib/db"; +import type { + MeasurementSource, + MeasurementType, +} from "@/generated/prisma/client"; import type { DayAggregateRow, ReadDayAggregatesOptions, + SourceDayAggregateRow, } from "@/lib/measurements/day-aggregates"; import { userDayKey } from "@/lib/tz/format"; @@ -71,3 +76,57 @@ export async function fakeReadDayAggregates( })) ?? []) as Array<{ measuredAt: Date; value: number }>; return foldDayAggregates(rows, opts); } + +/** + * In-memory stand-in for `readSourceDayAggregates`: reads the mocked + * `measurement.findMany` rows and groups them per type, day, source and + * device with the SQL's rules. + */ +export async function fakeReadSourceDayAggregates(opts: { + userId: string; + types: readonly MeasurementType[]; + since: Date; + timeZone: string; +}): Promise { + const rows = ((await prisma.measurement.findMany({ + where: { + userId: opts.userId, + deletedAt: null, + type: { in: [...opts.types] }, + measuredAt: { gte: opts.since }, + }, + orderBy: { measuredAt: "asc" }, + })) ?? []) as Array<{ + type: MeasurementType; + value: number; + measuredAt: Date; + source?: MeasurementSource | null; + deviceType?: string | null; + }>; + const groups = new Map(); + for (const r of rows) { + if (r.measuredAt.getTime() < opts.since.getTime()) continue; + if (!opts.types.includes(r.type)) continue; + const day = userDayKey(r.measuredAt, opts.timeZone); + const source = (r.source ?? "MANUAL") as MeasurementSource; + const deviceType = r.deviceType ?? null; + const key = `${r.type}|${day}|${source}|${deviceType}`; + const g = groups.get(key); + if (g) { + g.n += 1; + g.sum += r.value; + if (r.measuredAt < g.firstAt) g.firstAt = r.measuredAt; + } else { + groups.set(key, { + type: r.type, + day, + source, + deviceType, + n: 1, + sum: r.value, + firstAt: r.measuredAt, + }); + } + } + return [...groups.values()]; +} diff --git a/src/lib/measurements/day-aggregates.ts b/src/lib/measurements/day-aggregates.ts index 083b59b78..48d2966b1 100644 --- a/src/lib/measurements/day-aggregates.ts +++ b/src/lib/measurements/day-aggregates.ts @@ -19,7 +19,11 @@ * 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 type { + MeasurementSource, + MeasurementType, + PrismaClient, +} from "@/generated/prisma/client"; import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; import { isValidTimezone } from "@/lib/tz/format"; @@ -120,3 +124,82 @@ export async function readDayAggregates( max: Number(r.max), })); } + +export interface SourceDayAggregateRow { + type: MeasurementType; + /** Calendar day in the requested zone, `YYYY-MM-DD`. */ + day: string; + source: MeasurementSource; + deviceType: string | null; + n: number; + sum: number; + /** Earliest reading of the group; lies inside `day` in the zone. */ + firstAt: Date; +} + +/** + * Per-day aggregates of several types, kept apart per source and device. + * + * For readers that pick one source per day (`pickCanonicalSourceRows`) and + * then add up or average that day. The picker decides from which sources and + * device types are present on a day, never from individual values, so it + * makes the same pick over these groups as over the raw rows, and the day's + * sum and count follow exactly. Size: one row per type, day, source and + * device, however densely a type is sampled. + */ +export async function readSourceDayAggregates(opts: { + userId: string; + types: readonly MeasurementType[]; + since: Date; + timeZone: string; + db?: Pick; +}): Promise { + if (opts.types.length === 0) return []; + const timeZone = isValidTimezone(opts.timeZone) ? opts.timeZone : "UTC"; + const types = Prisma.join( + opts.types.map((t) => Prisma.sql`${t}::measurement_type`), + ); + const rows = await (opts.db ?? prisma).$queryRaw< + Array<{ + type: string; + day: string; + source: string; + device_type: string | null; + n: number; + sum: number; + first_at: Date; + }> + >` + WITH src AS ( + SELECT + m."type"::text AS type, + to_char((m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}, 'YYYY-MM-DD') AS day, + m."source"::text AS source, + m."device_type" AS device_type, + m."value" AS value, + m."measured_at" AS measured_at + FROM measurements m + WHERE m."user_id" = ${opts.userId} + AND m."type" IN (${types}) + AND m."deleted_at" IS NULL + AND m."measured_at" >= ${opts.since} + ) + SELECT + type, day, source, device_type, + COUNT(*)::int AS n, + SUM(value)::double precision AS sum, + MIN(measured_at) AS first_at + FROM src + GROUP BY type, day, source, device_type + ORDER BY day ASC, type ASC, source ASC + `; + return rows.map((r) => ({ + type: r.type as MeasurementType, + day: r.day, + source: r.source as MeasurementSource, + deviceType: r.device_type, + n: Number(r.n), + sum: Number(r.sum), + firstAt: r.first_at, + })); +} diff --git a/tests/integration/cycle-crosstab-day-aggregates.test.ts b/tests/integration/cycle-crosstab-day-aggregates.test.ts new file mode 100644 index 000000000..cdbc2df2e --- /dev/null +++ b/tests/integration/cycle-crosstab-day-aggregates.test.ts @@ -0,0 +1,114 @@ +/** + * The cycle-phase crosstab reads per-day, per-source aggregates instead of a + * year of raw readings (#1023). This pins that the day values it tests are + * the ones the raw readings give: for every crosstab metric, `metricDayMap` + * over the aggregates equals `metricDayMap` over the raw rows, on a fixture + * with two sources on the same days, two device types within one source, + * dense CGM glucose, summed steps and readings next to Berlin midnight. + */ +import { beforeAll, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { readSourceDayAggregates } from "@/lib/measurements/day-aggregates"; +import { metricDayMap } from "@/lib/insights/mood-crosstab"; +import { PHASE_CROSSTAB_METRIC_TYPES } from "@/lib/cycle/phase-crosstab"; +import type { CrossMetricMeasurement } from "@/lib/insights/mood-aggregates"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const prisma = getPrismaClient(); +const USER = "cycle-crosstab-agg"; +const SINCE = new Date("2026-06-01T00:00:00Z"); + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.user.create({ data: { id: USER, username: USER } }); + // CGM glucose every 5 minutes for 60 days from two sources. + await prisma.$executeRawUnsafe( + `INSERT INTO measurements (id, user_id, type, value, unit, source, measured_at, created_at, updated_at, sync_version) + SELECT 'cg' || s || g, $1, 'BLOOD_GLUCOSE'::measurement_type, 80 + (g % 60), 'mg/dL', + (CASE WHEN s = 0 THEN 'APPLE_HEALTH' ELSE 'NIGHTSCOUT' END)::measurement_source, + timestamp '2026-06-02 00:00:00' + (g * interval '5 minutes') + (s * interval '1 second'), + now(), now(), 1 + FROM generate_series(0, 60 * 288 - 1) g, generate_series(0, 1) s`, + USER, + ); + // Steps every 15 minutes from a watch and a phone, plus Google totals on + // some days; resting heart rate once a day near Berlin midnight. + await prisma.$executeRawUnsafe( + `INSERT INTO measurements (id, user_id, type, value, unit, source, device_type, measured_at, created_at, updated_at, sync_version) + SELECT 'st' || d || g, $1, 'ACTIVITY_STEPS'::measurement_type, 10 + (g % 7), 'count', + 'APPLE_HEALTH'::measurement_source, CASE WHEN d = 0 THEN 'watch' ELSE 'phone' END, + timestamp '2026-06-02 00:00:00' + (g * interval '15 minutes') + (d * interval '2 seconds'), + now(), now(), 1 + FROM generate_series(0, 60 * 96 - 1) g, generate_series(0, 1) d + WHERE d = 0 OR g % 3 = 0`, + USER, + ); + await prisma.$executeRawUnsafe( + `INSERT INTO measurements (id, user_id, type, value, unit, source, measured_at, created_at, updated_at, sync_version) + SELECT 'gs' || g, $1, 'ACTIVITY_STEPS'::measurement_type, 5000 + g, 'count', + 'GOOGLE_HEALTH'::measurement_source, + timestamp '2026-06-02 12:00:00' + (g * interval '1 day'), now(), now(), 1 + FROM generate_series(0, 59, 4) g`, + USER, + ); + await prisma.$executeRawUnsafe( + `INSERT INTO measurements (id, user_id, type, value, unit, source, measured_at, created_at, updated_at, sync_version) + SELECT 'rh' || g, $1, 'RESTING_HEART_RATE'::measurement_type, 55 + (g % 9), 'bpm', + 'WITHINGS'::measurement_source, + timestamp '2026-06-02 22:30:00' + (g * interval '1 day'), now(), now(), 1 + FROM generate_series(0, 59) g`, + USER, + ); +}, 120_000); + +describe("cycle crosstab day values (#1023)", () => { + it("per-source day aggregates give the raw rows' day values for every metric", async () => { + const raw = await prisma.measurement.findMany({ + where: { + userId: USER, + deletedAt: null, + type: { in: PHASE_CROSSTAB_METRIC_TYPES }, + measuredAt: { gte: SINCE }, + }, + select: { + type: true, + value: true, + measuredAt: true, + source: true, + deviceType: true, + }, + }); + const agg = await readSourceDayAggregates({ + userId: USER, + types: PHASE_CROSSTAB_METRIC_TYPES, + since: SINCE, + timeZone: "Europe/Berlin", + }); + const aggRows: CrossMetricMeasurement[] = agg.map((a) => ({ + type: a.type, + value: a.sum, + count: a.n, + measuredAt: a.firstAt, + source: a.source, + deviceType: a.deviceType, + })); + + expect(raw.length).toBeGreaterThan(40_000); + expect(aggRows.length).toBeLessThan(500); + + for (const type of PHASE_CROSSTAB_METRIC_TYPES) { + const expected = metricDayMap(raw, type, null); + const actual = metricDayMap(aggRows, type, null); + expect([...actual.keys()].sort(), type).toEqual( + [...expected.keys()].sort(), + ); + for (const [day, value] of expected) { + expect(actual.get(day), `${type} ${day}`).toBeCloseTo(value, 9); + } + } + }, 60_000); +});