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); +});