diff --git a/src/app/api/analytics/route.ts b/src/app/api/analytics/route.ts index f56344441..f34518523 100644 --- a/src/app/api/analytics/route.ts +++ b/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(); - 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(); + 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> { - const out: Array<{ measuredAt: Date; value: number }> = []; + type: MeasurementType, + options: { since?: Date; includeSleepStage?: boolean } = {}, +): Promise { + 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; diff --git a/tests/integration/analytics-bp-aggregate-paged.test.ts b/tests/integration/analytics-bp-aggregate-paged.test.ts new file mode 100644 index 000000000..64b03259b --- /dev/null +++ b/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 + )(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); + }); +});