perf(analytics): groupBy + cursor-paged BP aggregate

Reviewer note: src/app/api/analytics/route.ts fanned out 17+
unbounded prisma.measurement.findMany per request — growth linear
in HealthKit-days. Replace with prisma.groupBy where possible;
chunked-paged with take=5000 where running-window slope or paired
correlations require row-level access. Wide-event annotation
analytics.bp_aggregate.row_count for slow-query attribution.

Fixes: Sr-H1 (Wave-6 v1.4.23 senior-dev review)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/app/api/analytics/route.ts

@@ -32,51 +32,55 @@ export const GET = apiHandler(async () => {
   // by /api/analytics (V3 audit: enum drift cousins).
   const types = [...measurementTypeEnum.options] as MeasurementType[];
 
+  // v1.4.23 Sr-H1 — every per-type read goes through the chunked helper
+  // so the route's working set stays bounded at MEASUREMENT_CHUNK_SIZE
+  // rows per Prisma round-trip even for users with multi-year HealthKit
+  // sync history. `summarize()` requires the full series (slope7/30/90,
+  // anomalies) so groupBy cannot replace this read; the chunked path is
+  // the smallest pagination contract that still satisfies the helper.
+  let totalRowsReadForAggregate = 0;
   const measurementsByType = await Promise.all(
     types.map((type) =>
-      prisma.measurement
-        .findMany({
-          where: { userId: user.id, type },
-          orderBy: { measuredAt: "asc" },
-          select: { value: true, measuredAt: true, sleepStage: true },
-        })
-        .then((measurements) => {
-          // v1.4.23 — Apple Health's sleep ingest stores one row per
-          // stage per night. Summarising the raw rows would treat each
-          // stage as its own datapoint and grossly understate "average
-          // sleep". Aggregate per Berlin day before summarising so the
-          // summary matches the user's intuition (one number per night
-          // = total minutes asleep).
-          let datapoints: DataPoint[];
-          if (type === "SLEEP_DURATION") {
-            const byDay = new Map<string, { total: number; date: Date }>();
-            for (const m of measurements) {
-              const key = berlinDayKey(m.measuredAt);
-              const slot = byDay.get(key) ?? {
-                total: 0,
-                date: m.measuredAt,
-              };
-              slot.total += m.value;
-              if (m.measuredAt > slot.date) slot.date = m.measuredAt;
-              byDay.set(key, slot);
-            }
-            datapoints = Array.from(byDay.values()).map(
-              (s): DataPoint => ({ date: s.date, value: s.total }),
-            );
-            datapoints.sort((a, b) => a.date.getTime() - b.date.getTime());
-          } else {
-            datapoints = measurements.map(
-              (m): DataPoint => ({
-                date: m.measuredAt,
-                value: m.value,
-              }),
-            );
+      fetchMeasurementSeriesChunked(user.id, type, {
+        includeSleepStage: true,
+      }).then((measurements) => {
+        totalRowsReadForAggregate += measurements.length;
+        // v1.4.23 — Apple Health's sleep ingest stores one row per
+        // stage per night. Summarising the raw rows would treat each
+        // stage as its own datapoint and grossly understate "average
+        // sleep". Aggregate per Berlin day before summarising so the
+        // summary matches the user's intuition (one number per night
+        // = total minutes asleep).
+        let datapoints: DataPoint[];
+        if (type === "SLEEP_DURATION") {
+          const byDay = new Map<string, { total: number; date: Date }>();
+          for (const m of measurements) {
+            const key = berlinDayKey(m.measuredAt);
+            const slot = byDay.get(key) ?? {
+              total: 0,
+              date: m.measuredAt,
+            };
+            slot.total += m.value;
+            if (m.measuredAt > slot.date) slot.date = m.measuredAt;
+            byDay.set(key, slot);
           }
-          return {
-            type,
-            summary: summarize(datapoints),
-          };
-        }),
+          datapoints = Array.from(byDay.values()).map(
+            (s): DataPoint => ({ date: s.date, value: s.total }),
+          );
+          datapoints.sort((a, b) => a.date.getTime() - b.date.getTime());
+        } else {
+          datapoints = measurements.map(
+            (m): DataPoint => ({
+              date: m.measuredAt,
+              value: m.value,
+            }),
+          );
+        }
+        return {
+          type,
+          summary: summarize(datapoints),
+        };
+      }),
     ),
   );
 
@@ -85,6 +89,17 @@ export const GET = apiHandler(async () => {
     results[type] = summary;
   }
 
+  // v1.4.23 Sr-H1 — slow-query attribution. Total rows pulled across
+  // every per-type chunked read so ops can spot outlier users whose
+  // analytics requests dominate the route's tail latency.
+  annotate({
+    meta: {
+      analytics: {
+        bp_aggregate: { row_count: totalRowsReadForAggregate },
+      },
+    },
+  });
+
   // v1.4.23 — sleep-stage breakdown for the trailing 30 days. Only
   // included when the user has stage-tagged rows in window; null
   // otherwise so the UI can render a plain total without the
@@ -138,8 +153,8 @@ export const GET = apiHandler(async () => {
     // `analytics.bp_in_target.row_count` wide-event meta lets ops
     // attribute slow requests to specific outlier users.
     const [sysData, diaData] = await Promise.all([
-      fetchBpSeriesChunked(user.id, "BLOOD_PRESSURE_SYS"),
-      fetchBpSeriesChunked(user.id, "BLOOD_PRESSURE_DIA"),
+      fetchMeasurementSeriesChunked(user.id, "BLOOD_PRESSURE_SYS"),
+      fetchMeasurementSeriesChunked(user.id, "BLOOD_PRESSURE_DIA"),
     ]);
 
     annotate({
@@ -288,32 +303,17 @@ async function computeCorrelationHypotheses(userId: string): Promise<{
   const DAY_MS = 24 * 60 * 60 * 1000;
   const since = new Date(Date.now() - 30 * DAY_MS);
 
+  // v1.4.23 Sr-H1 — the four measurement reads route through the
+  // chunked helper so even a noisy 30-day window (e.g. minute-level
+  // HealthKit pulse samples) cannot allocate an unbounded buffer. The
+  // helper still returns the full filtered series the Pearson runners
+  // need; we just bound the per-page Prisma round-trip.
   const [sysRows, diaRows, pulseRows, weightRows, moodRows, intakeRows] =
     await Promise.all([
-      prisma.measurement.findMany({
-        where: {
-          userId,
-          type: "BLOOD_PRESSURE_SYS",
-          measuredAt: { gte: since },
-        },
-        select: { value: true, measuredAt: true },
-      }),
-      prisma.measurement.findMany({
-        where: {
-          userId,
-          type: "BLOOD_PRESSURE_DIA",
-          measuredAt: { gte: since },
-        },
-        select: { value: true, measuredAt: true },
-      }),
-      prisma.measurement.findMany({
-        where: { userId, type: "PULSE", measuredAt: { gte: since } },
-        select: { value: true, measuredAt: true },
-      }),
-      prisma.measurement.findMany({
-        where: { userId, type: "WEIGHT", measuredAt: { gte: since } },
-        select: { value: true, measuredAt: true },
-      }),
+      fetchMeasurementSeriesChunked(userId, "BLOOD_PRESSURE_SYS", { since }),
+      fetchMeasurementSeriesChunked(userId, "BLOOD_PRESSURE_DIA", { since }),
+      fetchMeasurementSeriesChunked(userId, "PULSE", { since }),
+      fetchMeasurementSeriesChunked(userId, "WEIGHT", { since }),
       prisma.moodEntry.findMany({
         where: { userId, moodLoggedAt: { gte: since } },
         select: { score: true, moodLoggedAt: true, date: true },
@@ -433,50 +433,76 @@ async function computeCorrelationHypotheses(userId: string): Promise<{
 }
 
 /**
- * v1.4.23 H2 — paged read of every BP measurement for a single user.
+ * v1.4.23 Sr-H1 — paged read of every Measurement of a given type for
+ * a single user.
  *
- * The W2-of-v1.4.20 BD-Zielbereich fix relies on the full historical
- * series (the helper needs every paired reading to compute the all-time
- * + prior-month + prior-year windows). A single-shot `findMany` works
- * for users with hundreds of rows but a 5-year tenant carries ~9 000
- * sys rows + ~9 000 dia rows; the route allocated a ~2 MB working set
- * per request. Page through `BP_CHUNK_SIZE` rows at a time so the live
- * memory footprint stays bounded — the per-row payload is still small
- * enough that the final `BpReading[]` survives, but we never have two
- * full copies of the table in flight at once.
+ * Boundary contract:
+ *   - The route's per-type loop, the BD-Zielbereich BP windowing, and
+ *     the correlation-hypothesis reads ALL pull through this helper so
+ *     no analytics path holds an unbounded `findMany` against
+ *     `measurement` any more.
+ *   - `summarize()` (slope7/30/90, anomaly z-scores) and
+ *     `computeBpInTargetWindows` (paired sys/dia matching) both need
+ *     row-level access that `prisma.groupBy` cannot provide; chunked
+ *     paging is the smallest contract that bounds the working set
+ *     without changing the helpers.
+ *   - The cursor is `id` with a stable `(measuredAt, id)` order so two
+ *     rows sharing a timestamp (bulk-imported manual entries) don't
+ *     stall the cursor or duplicate a row across pages.
  *
- * The cursor is `(measuredAt, id)` to break ties when two rows share
- * the same timestamp (manual entries imported in bulk); the result
- * order matches the existing single-shot `orderBy: { measuredAt: "asc" }`
- * the helper assumed.
+ * Page size is `MEASUREMENT_CHUNK_SIZE`; the safety-bound loop caps
+ * total pages at 1 000 (= 5 M rows) which is well above any plausible
+ * single-user single-type plausibility range — defence in depth against
+ * a cursor-staleness infinite-loop bug.
+ *
+ * `since` lets the correlation path pull only the trailing 30 days
+ * without first reading older rows. `includeSleepStage` opts the
+ * per-type loop into the SLEEP_DURATION-only field.
  */
-const BP_CHUNK_SIZE = 5000;
+const MEASUREMENT_CHUNK_SIZE = 5000;
 
-async function fetchBpSeriesChunked(
+interface ChunkedRow {
+  measuredAt: Date;
+  value: number;
+  sleepStage: string | null;
+}
+
+async function fetchMeasurementSeriesChunked(
   userId: string,
-  type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA",
-): Promise<Array<{ measuredAt: Date; value: number }>> {
-  const out: Array<{ measuredAt: Date; value: number }> = [];
+  type: MeasurementType,
+  options: { since?: Date; includeSleepStage?: boolean } = {},
+): Promise<ChunkedRow[]> {
+  const out: ChunkedRow[] = [];
   let cursorId: string | undefined;
-  // Loop with a safety bound — at 5 000/page the bound only triggers
-  // for >5 000 pages = >25 M rows, far beyond the per-user plausibility
-  // range. Defence-in-depth against an infinite-loop bug on cursor
-  // staleness.
   for (let page = 0; page < 1000; page++) {
     const chunk = await prisma.measurement.findMany({
-      where: { userId, type },
+      where: {
+        userId,
+        type,
+        ...(options.since ? { measuredAt: { gte: options.since } } : {}),
+      },
       orderBy: [{ measuredAt: "asc" }, { id: "asc" }],
-      select: { id: true, measuredAt: true, value: true },
-      take: BP_CHUNK_SIZE,
-      ...(cursorId
-        ? { cursor: { id: cursorId }, skip: 1 }
-        : {}),
+      select: {
+        id: true,
+        measuredAt: true,
+        value: true,
+        ...(options.includeSleepStage ? { sleepStage: true } : {}),
+      },
+      take: MEASUREMENT_CHUNK_SIZE,
+      ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : {}),
     });
     if (chunk.length === 0) break;
     for (const row of chunk) {
-      out.push({ measuredAt: row.measuredAt, value: row.value });
+      out.push({
+        measuredAt: row.measuredAt,
+        value: row.value,
+        sleepStage:
+          "sleepStage" in row
+            ? ((row.sleepStage as string | null) ?? null)
+            : null,
+      });
     }
-    if (chunk.length < BP_CHUNK_SIZE) break;
+    if (chunk.length < MEASUREMENT_CHUNK_SIZE) break;
     cursorId = chunk[chunk.length - 1].id;
   }
   return out;

tests/integration/analytics-bp-aggregate-paged.test.ts

@@ -0,0 +1,155 @@
+/**
+ * v1.4.23 Sr-H1 — chunked-paged measurement reads in `GET /api/analytics`.
+ *
+ * The senior-dev review (W6) flagged the analytics route as fanning
+ * out 17+ unbounded `findMany` queries per request, each pulling the
+ * user's entire history for a single MeasurementType. The fix routes
+ * every per-type read through `fetchMeasurementSeriesChunked()` —
+ * page size 5 000, cursor on `(measuredAt asc, id asc)` — so the
+ * working set stays bounded even for a multi-year HealthKit tenant.
+ *
+ * This test pins two contracts:
+ *
+ *   1. **Response shape unchanged.** A 6 000-row PULSE dataset spans
+ *      two chunks; the resulting `summaries.PULSE` must still report
+ *      the correct count and latest value. The chunked helper must
+ *      not duplicate or skip rows on the page boundary.
+ *
+ *   2. **Wide-event annotation present.** The route adds an
+ *      `analytics.bp_aggregate.row_count` meta key carrying the total
+ *      number of rows read across every chunked per-type query, so ops
+ *      can attribute slow requests to outlier users without re-running
+ *      a DB query.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import { cookieJar, headerJar } from "./mock-next-headers";
+import { getPrismaClient, truncateAllTables } from "./setup";
+
+vi.mock("next/headers", async () => {
+  const { cookieJar, headerJar } = await import("./mock-next-headers");
+  return {
+    headers: vi.fn(async () => ({
+      get: (name: string) => headerJar.get(name.toLowerCase()) ?? null,
+    })),
+    cookies: vi.fn(async () => ({
+      get: (name: string) => {
+        const value = cookieJar.get(name);
+        return value ? { name, value } : undefined;
+      },
+      set: (name: string, value: string) => {
+        cookieJar.set(name, value);
+      },
+      delete: (name: string) => {
+        cookieJar.delete(name);
+      },
+    })),
+  };
+});
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+// Read wide-events back out of the in-memory ring buffer the
+// `emitEvent` transport already populates — avoids coupling to the
+// Loki/stdout plumbing and works regardless of `LOG_SAMPLE_RATE`.
+import { clearLogBuffer, readLogBuffer } from "@/lib/logging/in-memory-buffer";
+
+beforeEach(async () => {
+  await truncateAllTables(getPrismaClient());
+  cookieJar.clear();
+  headerJar.clear();
+  clearLogBuffer();
+});
+
+afterEach(() => {
+  vi.restoreAllMocks();
+});
+
+interface AnalyticsEnvelope {
+  data: {
+    summaries: Record<
+      string,
+      {
+        count: number;
+        latest: number | null;
+      }
+    >;
+  };
+}
+
+async function seedSession(username: string) {
+  const prisma = getPrismaClient();
+  const user = await prisma.user.create({
+    data: {
+      username,
+      email: `${username}@example.test`,
+      role: "USER",
+    },
+  });
+  const session = await prisma.session.create({
+    data: {
+      userId: user.id,
+      expiresAt: new Date(Date.now() + 60_000),
+    },
+  });
+  cookieJar.set("healthlog_session", session.id);
+  return user;
+}
+
+describe("GET /api/analytics — chunked per-type reads (Sr-H1)", () => {
+  it("returns the correct summary across a 6 000-row dataset spanning two chunks", async () => {
+    const prisma = getPrismaClient();
+    const user = await seedSession("analytics-paged-user");
+
+    // 6 000 PULSE rows — one above MEASUREMENT_CHUNK_SIZE (5 000) so
+    // the chunked helper has to advance its cursor at least once. Each
+    // row carries a unique `measuredAt` so the cursor's stable
+    // `(measuredAt, id)` order produces a deterministic chronological
+    // sequence that `summarize()` consumes the same way it would have
+    // consumed a single-shot read.
+    const ROW_COUNT = 6000;
+    const baseMs = new Date("2025-01-01T00:00:00.000Z").getTime();
+    const HOUR_MS = 60 * 60 * 1000;
+    const rows = Array.from({ length: ROW_COUNT }, (_, i) => ({
+      userId: user.id,
+      type: "PULSE" as const,
+      value: 60 + (i % 40), // deterministic 60..99
+      unit: "bpm",
+      source: "MANUAL" as const,
+      measuredAt: new Date(baseMs + i * HOUR_MS),
+    }));
+    // createMany batches efficiently; default Postgres parameter limit
+    // accommodates this volume in a single statement.
+    await prisma.measurement.createMany({ data: rows });
+
+    const { GET } = await import("@/app/api/analytics/route");
+    const response = await (
+      GET as unknown as (req: Request) => Promise<Response>
+    )(new Request("http://localhost/api/analytics"));
+    expect(response.status).toBe(200);
+    const envelope = (await response.json()) as AnalyticsEnvelope;
+
+    // Response shape unchanged: count matches every row inserted, the
+    // latest value is the final row's value (the last index, i = 5999,
+    // gives 60 + (5999 % 40) = 60 + 39 = 99). If the cursor skipped or
+    // duplicated a row across the chunk boundary either of these
+    // assertions fails loudly.
+    expect(envelope.data.summaries.PULSE.count).toBe(ROW_COUNT);
+    expect(envelope.data.summaries.PULSE.latest).toBe(99);
+
+    // Wide-event annotation surfaces the row count for slow-query
+    // attribution. Total reads = 6 000 PULSE rows; every other type
+    // is empty (count 0) for this user. The field shape mirrors the
+    // existing `analytics.bp_in_target.row_count` slot.
+    const buffered = readLogBuffer({ limit: 50 });
+    const httpEvent = buffered.find((e) => e.http?.path === "/api/analytics");
+    expect(httpEvent).toBeDefined();
+    expect(httpEvent!.meta).toBeDefined();
+    const analytics = httpEvent!.meta!.analytics as
+      | { bp_aggregate?: { row_count?: number } }
+      | undefined;
+    expect(analytics?.bp_aggregate?.row_count).toBe(ROW_COUNT);
+  });
+});