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

src/__tests__/dense-measurement-read-guard.test.ts

@@ -74,11 +74,9 @@ const ALLOWED: Record<string, string> = {
   "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.",
 };

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";
 

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

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;
 }
 
 /**

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<string, number>();

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<SourceDayAggregateRow[]> {
+  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<string, SourceDayAggregateRow>();
+  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()];
+}

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<PrismaClient, "$queryRaw">;
+}): Promise<SourceDayAggregateRow[]> {
+  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,
+  }));
+}

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