diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index a3b66da34..037706cd3 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -1,19 +1,20 @@ /** - * v1.4.33 C1 — unit-level pin for the slim summaries slice. Heavier - * integration coverage lives in `tests/integration/analytics-summaries-slice.test.ts` + * Unit-level pin for the slim summaries slice. Heavier integration + * coverage lives in `tests/integration/analytics-summaries-slice.test.ts` * (real Postgres, real `regr_slope`); this file mocks `$queryRaw` so - * the slope/round/empty contracts are pinned without a container. + * the slope/round/empty/path-selection contracts are pinned without + * a container. */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: vi.fn(), - // v1.4.35 — slim slice now reads DAY buckets from - // `measurement_rollups` for the parity-checked composition. The - // freshness watermark inside `ensureUserRollupsFresh` also pokes - // `measurementRollup.findFirst`; mock both so the helper runs - // without a container. + measurement: { findFirst: vi.fn() }, + // v1.4.36 — slim slice reads DAY buckets from `measurement_rollups` + // on the happy path. The freshness watermark inside + // `ensureUserRollupsFresh` also pokes `measurementRollup.findFirst`; + // mock both so the helper runs without a container. measurementRollup: { findMany: vi.fn(), findFirst: vi.fn(), @@ -29,6 +30,8 @@ import { prisma } from "@/lib/db"; import { computeSummariesSlice } from "../summaries-slice"; const RAW = prisma.$queryRaw as unknown as ReturnType; +const MEASUREMENT_FIND_FIRST = + prisma.measurement.findFirst as unknown as ReturnType; const ROLLUP_FIND_MANY = prisma.measurementRollup.findMany as unknown as ReturnType; const ROLLUP_FIND_FIRST = @@ -36,13 +39,12 @@ const ROLLUP_FIND_FIRST = beforeEach(() => { RAW.mockReset(); + MEASUREMENT_FIND_FIRST.mockReset(); ROLLUP_FIND_MANY.mockReset(); ROLLUP_FIND_FIRST.mockReset(); - // Defaults: empty rollup buckets so the parity check falls back to - // live SQL (the pre-backfill behaviour). Individual tests opt in - // to rollup-derived values by overriding the mock. ROLLUP_FIND_MANY.mockResolvedValue([]); ROLLUP_FIND_FIRST.mockResolvedValue(null); + MEASUREMENT_FIND_FIRST.mockResolvedValue(null); }); afterEach(() => { @@ -50,178 +52,241 @@ afterEach(() => { }); describe("computeSummariesSlice", () => { - it("returns the empty-summary skeleton when the user has no rows", async () => { - // Pass 1 (aggregates) returns empty, pass 2 (latest) returns empty. - // Pass 3 (rollup.findMany) is defaulted to [] in beforeEach. - RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]); + describe("cold fallback — empty rollup table", () => { + it("returns the empty-summary skeleton when the user has no rows", async () => { + // 1. rollup-count probe — zero ⇒ cold path. + // 2. heavy aggregate ($queryRaw) — empty. + // 3. latests ($queryRaw) — empty. + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]); - const result = await computeSummariesSlice("user-1"); + const result = await computeSummariesSlice("user-1"); - // Every enum option is seeded so consumers can read - // `summaries.WEIGHT` and get a stable empty shape. - expect(result.summaries.WEIGHT).toEqual({ - count: 0, - latest: null, - min: null, - max: null, - mean: null, - avg7: null, - avg30: null, - slope7: null, - slope30: null, - slope90: null, - anomalyCount: 0, - avg30LastMonth: null, - avg30LastYear: null, - }); - expect(result.bmi).toBeNull(); - // Both SQL passes fired. - expect(RAW).toHaveBeenCalledTimes(2); - }); - - it("maps a populated aggregate row into the DataSummary shape", async () => { - RAW.mockResolvedValueOnce([ - { - type: "WEIGHT", - count: BigInt(42), - min_value: 79.2, - max_value: 84.1, - mean_value: 82.05, - avg7: 81.9, - avg30: 82.1, - slope7: -0.014, - r2_7: 0.65, - slope30: -0.005, - r2_30: 0.42, - slope90: 0.001, - r2_90: 0.12, - }, - ]).mockResolvedValueOnce([ - { type: "WEIGHT", value: 81.4, measured_at: new Date() }, - ]); - - const result = await computeSummariesSlice("user-1"); - const weight = result.summaries.WEIGHT; - - expect(weight.count).toBe(42); - expect(weight.latest).toBe(81.4); - expect(weight.min).toBe(79.2); - expect(weight.max).toBe(84.1); - expect(weight.mean).toBe(82.05); - expect(weight.avg7).toBe(81.9); - expect(weight.avg30).toBe(82.1); - expect(weight.anomalyCount).toBe(0); - expect(weight.avg30LastMonth).toBeNull(); - expect(weight.avg30LastYear).toBeNull(); - // Slope direction respects the 0.01-units-per-day "stable" - // threshold the JS helper uses. -0.014 → "down", -0.005 → - // "stable", 0.001 → "stable". - expect(weight.slope7).toEqual({ - slope: -0.014, - direction: "down", - confidence: 0.65, - }); - expect(weight.slope30).toEqual({ - slope: -0.005, - direction: "stable", - confidence: 0.42, - }); - expect(weight.slope90).toEqual({ - slope: 0.001, - direction: "stable", - confidence: 0.12, - }); - }); - - it("returns a null slope tuple when the SQL slope is null (insufficient rows)", async () => { - RAW.mockResolvedValueOnce([ - { - type: "PULSE", - count: BigInt(1), - min_value: 72, - max_value: 72, - mean_value: 72, - avg7: 72, - avg30: 72, - // regr_slope returns NULL when n < 2. - slope7: null, - r2_7: null, - slope30: null, - r2_30: null, - slope90: null, - r2_90: null, - }, - ]).mockResolvedValueOnce([ - { type: "PULSE", value: 72, measured_at: new Date() }, - ]); - - const result = await computeSummariesSlice("user-1"); - expect(result.summaries.PULSE.slope7).toBeNull(); - expect(result.summaries.PULSE.slope30).toBeNull(); - expect(result.summaries.PULSE.slope90).toBeNull(); - }); - - // v1.4.34 IW-B — surface the per-type freshness map so the - // dashboard tile strip can render an "Letzter Wert vor Xd" caption - // even when the consumer read from the slim slice (the path the - // tile-strip pre-fetch uses on cold paint). - it("surfaces lastSeenByType from the DISTINCT ON pass's measured_at", async () => { - const tenDaysAgo = new Date(Date.now() - 10 * 24 * 60 * 60 * 1000); - RAW.mockResolvedValueOnce([ - { - type: "WEIGHT", - count: BigInt(5), - min_value: 80, - max_value: 84, - mean_value: 82, + expect(result.summaries.WEIGHT).toEqual({ + count: 0, + latest: null, + min: null, + max: null, + mean: null, avg7: null, - avg30: 82, + avg30: null, slope7: null, - r2_7: null, - slope30: 0.005, - r2_30: 0.2, + slope30: null, slope90: null, - r2_90: null, - }, - ]).mockResolvedValueOnce([ - { type: "WEIGHT", value: 82.3, measured_at: tenDaysAgo }, - ]); + anomalyCount: 0, + avg30LastMonth: null, + avg30LastYear: null, + }); + expect(result.bmi).toBeNull(); + expect(RAW).toHaveBeenCalledTimes(3); + }); - const result = await computeSummariesSlice("user-1"); - const ws = result.lastSeenByType.WEIGHT; - expect(ws).not.toBeNull(); - expect(ws?.daysAgo).toBeGreaterThanOrEqual(9); - expect(ws?.daysAgo).toBeLessThanOrEqual(11); - expect(ws?.lastSeenAt).toBe(tenDaysAgo.toISOString()); - // Types the user never logged report null so call sites can - // fall through without painting a caption. - expect(result.lastSeenByType.PULSE).toBeNull(); + it("maps a populated heavy aggregate row into the DataSummary shape on cold path", async () => { + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(42), + min_value: 79.2, + max_value: 84.1, + mean_value: 82.05, + avg7: 81.9, + avg30: 82.1, + slope7: -0.014, + r2_7: 0.65, + slope30: -0.005, + r2_30: 0.42, + slope90: 0.001, + r2_90: 0.12, + }, + ]) + .mockResolvedValueOnce([ + { type: "WEIGHT", value: 81.4, measured_at: new Date() }, + ]); + + const result = await computeSummariesSlice("user-1"); + const weight = result.summaries.WEIGHT; + + expect(weight.count).toBe(42); + expect(weight.latest).toBe(81.4); + expect(weight.min).toBe(79.2); + expect(weight.max).toBe(84.1); + expect(weight.mean).toBe(82.05); + expect(weight.avg7).toBe(81.9); + expect(weight.avg30).toBe(82.1); + expect(weight.anomalyCount).toBe(0); + expect(weight.avg30LastMonth).toBeNull(); + expect(weight.avg30LastYear).toBeNull(); + expect(weight.slope7).toEqual({ + slope: -0.014, + direction: "down", + confidence: 0.65, + }); + expect(weight.slope30).toEqual({ + slope: -0.005, + direction: "stable", + confidence: 0.42, + }); + expect(weight.slope90).toEqual({ + slope: 0.001, + direction: "stable", + confidence: 0.12, + }); + }); + + it("returns a null slope tuple when the SQL slope is null (insufficient rows)", async () => { + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([ + { + type: "PULSE", + count: BigInt(1), + min_value: 72, + max_value: 72, + mean_value: 72, + avg7: 72, + avg30: 72, + slope7: null, + r2_7: null, + slope30: null, + r2_30: null, + slope90: null, + r2_90: null, + }, + ]) + .mockResolvedValueOnce([ + { type: "PULSE", value: 72, measured_at: new Date() }, + ]); + + const result = await computeSummariesSlice("user-1"); + expect(result.summaries.PULSE.slope7).toBeNull(); + expect(result.summaries.PULSE.slope30).toBeNull(); + expect(result.summaries.PULSE.slope90).toBeNull(); + }); + + it("surfaces lastSeenByType from the DISTINCT ON pass's measured_at", async () => { + const tenDaysAgo = new Date(Date.now() - 10 * 24 * 60 * 60 * 1000); + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(5), + min_value: 80, + max_value: 84, + mean_value: 82, + avg7: null, + avg30: 82, + slope7: null, + r2_7: null, + slope30: 0.005, + r2_30: 0.2, + slope90: null, + r2_90: null, + }, + ]) + .mockResolvedValueOnce([ + { type: "WEIGHT", value: 82.3, measured_at: tenDaysAgo }, + ]); + + const result = await computeSummariesSlice("user-1"); + const ws = result.lastSeenByType.WEIGHT; + expect(ws).not.toBeNull(); + expect(ws?.daysAgo).toBeGreaterThanOrEqual(9); + expect(ws?.daysAgo).toBeLessThanOrEqual(11); + expect(ws?.lastSeenAt).toBe(tenDaysAgo.toISOString()); + expect(result.lastSeenByType.PULSE).toBeNull(); + }); + + it("seeds the latest value from the DISTINCT ON pass per type", async () => { + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([ + { + type: "PULSE", + count: BigInt(3), + min_value: 60, + max_value: 95, + mean_value: 77, + avg7: 77, + avg30: 77, + slope7: 0, + r2_7: 0, + slope30: 0, + r2_30: 0, + slope90: 0, + r2_90: 0, + }, + ]) + .mockResolvedValueOnce([ + { type: "PULSE", value: 88, measured_at: new Date() }, + ]); + + const result = await computeSummariesSlice("user-1"); + expect(result.summaries.PULSE.latest).toBe(88); + expect(result.summaries.PULSE.max).toBe(95); + }); }); - it("seeds the latest value from the DISTINCT ON pass per type", async () => { - RAW.mockResolvedValueOnce([ - { - type: "PULSE", - count: BigInt(3), - min_value: 60, - max_value: 95, - mean_value: 77, - avg7: 77, - avg30: 77, - slope7: 0, - r2_7: 0, - slope30: 0, - r2_30: 0, - slope90: 0, - r2_90: 0, - }, - ]).mockResolvedValueOnce([ - { type: "PULSE", value: 88, measured_at: new Date() }, - ]); + describe("rollup-fresh happy path", () => { + it("composes count/min/max/mean from DAY buckets without running the heavy aggregate", async () => { + // 1. rollup-count probe — non-zero ⇒ happy path. + // 2. narrow aggregate ($queryRaw) — windowed/regression only. + // 3. latests ($queryRaw). + RAW.mockResolvedValueOnce([{ n: BigInt(2) }]) + .mockResolvedValueOnce([ + { + type: "WEIGHT", + avg7: 82, + avg30: 82.5, + slope7: 0.02, + r2_7: 0.5, + slope30: 0.01, + r2_30: 0.4, + slope90: 0.005, + r2_90: 0.2, + }, + ]) + .mockResolvedValueOnce([ + { type: "WEIGHT", value: 82.7, measured_at: new Date() }, + ]); - const result = await computeSummariesSlice("user-1"); - // Latest is NOT max — comes from the row at MAX(measured_at). - expect(result.summaries.PULSE.latest).toBe(88); - expect(result.summaries.PULSE.max).toBe(95); + ROLLUP_FIND_MANY.mockResolvedValueOnce([ + { + type: "WEIGHT", + bucketStart: new Date("2026-05-10T00:00:00.000Z"), + count: 10, + mean: 81.0, + minValue: 79.5, + maxValue: 82.0, + }, + { + type: "WEIGHT", + bucketStart: new Date("2026-05-11T00:00:00.000Z"), + count: 10, + mean: 83.0, + minValue: 81.5, + maxValue: 84.0, + }, + ]); + + const result = await computeSummariesSlice("user-rollup"); + const weight = result.summaries.WEIGHT; + + // count = 10 + 10 = 20; min/max across buckets; weighted mean. + expect(weight.count).toBe(20); + expect(weight.min).toBe(79.5); + expect(weight.max).toBe(84.0); + expect(weight.mean).toBe(82); + expect(weight.latest).toBe(82.7); + expect(weight.avg7).toBe(82); + expect(weight.slope7).toEqual({ + slope: 0.02, + direction: "up", + confidence: 0.5, + }); + + // probe + narrow aggregate + latests — NOT the heavy aggregate. + expect(RAW).toHaveBeenCalledTimes(3); + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1); + }); }); }); diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 5779880bd..419beb7f8 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -1,5 +1,5 @@ /** - * v1.4.33 C1 — slim `summaries` slice for `/api/analytics?slice=summaries`. + * v1.4.36 — slim `summaries` slice for `/api/analytics?slice=summaries`. * * The default `/api/analytics` path fans out 30+ `Promise.all`-wrapped * chunked findMany reads (one per `MeasurementType`) so it can hand @@ -9,28 +9,34 @@ * three slope tuples — every other field gets `null`-coalesced before * it ever paints — yet the user waits on the full chunked walk. * - * This helper resolves the same `DataSummary` shape (per-type) with - * three passes (two SQL + one rollup-table read): + * Read shape (v1.4.36) + * -------------------- + * v1.4.35 partially read-swapped onto the persistent `measurement_rollups` + * table but kept the full-fat `$queryRaw` running in parallel as a + * parity check. The heavy COUNT/MIN/MAX/AVG aggregate scanned the + * measurements table every request even on warm rollups. * - * 1. **`groupBy` + windowed aggregates** — one `$queryRaw` carrying - * `COUNT`/`MIN`/`MAX`/`AVG`, the 7-day and 30-day `AVG` slices - * via `FILTER`, plus `regr_slope` + `regr_r2` for the slope - * tuples at 7d / 30d / 90d windows. Postgres folds all of this - * into one index scan over `measurements WHERE user_id = …`. - * The `COUNT / MIN / MAX / AVG` columns stay on this pass as - * the live-SQL fallback for the v1.4.35 rollup parity check. + * v1.4.36 promotes the rollup table to the canonical source on the + * happy path: a single cheap COUNT against `measurement_rollups` + * decides whether to read the composable stats from buckets or fall + * back to the heavy live aggregate. When the rollup is populated: * - * 2. **`DISTINCT ON (type)` latest read** — one `$queryRaw` - * returning the most-recent `(type, value)` pair per type so the - * slim shape can populate `latest`. `MAX(value)` would not do — - * we need the value of the row at `MAX(measured_at)`, not the - * largest value ever recorded. + * 1. **DAY-bucket read** — composes `count / min / max / mean` per + * type via `aggregateBuckets`. The four are linearly composable + * across DAY buckets so the composed value is mathematically + * identical to the live aggregate over the same rows. * - * 3. **`measurement_rollups` DAY-bucket read** (v1.4.35) — folded - * into `count / min / max / mean` via `aggregateBuckets`. Used - * when the composed count agrees byte-for-byte with the live - * `COUNT(*)`; falls back to live for accounts that never ran - * the explicit backfill or that are mid-flight on the populator. + * 2. **Narrow `$queryRaw`** — carries only the non-composable + * windowed columns (`avg7`, `avg30`, and the slope tuples). The + * heavy COUNT/MIN/MAX/AVG columns are NOT projected. + * + * 3. **`DISTINCT ON (type)` latest read** — most-recent `(type, + * value, measured_at)` triplet per type for the `latest` field + * and the `lastSeenByType` freshness map. + * + * On the cold fallback (no rollup rows yet) the slim slice runs the + * legacy heavy aggregate so the response shape is correct on the very + * first request. * * Fields the slim shape intentionally omits (the dashboard never * reads them on first paint): @@ -54,7 +60,12 @@ import { annotate } from "@/lib/logging/context"; import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; import { aggregateBuckets } from "@/lib/measurements/rollup-read"; -interface AggregateRow { +/** + * Heavy aggregate row — used on the cold-mount fallback path where the + * rollup table is empty and we need every per-type column out of one + * SQL pass for the very first request. + */ +interface HeavyAggregateRow { type: string; count: bigint; min_value: number | null; @@ -70,6 +81,24 @@ interface AggregateRow { r2_90: number | null; } +/** + * Narrow aggregate row — only the columns DAY buckets can't compose + * linearly. The heavy `COUNT / MIN / MAX / AVG` columns are dropped + * so the rollup-fresh read path's SQL projection is half as wide as + * the legacy aggregate's. + */ +interface NarrowAggregateRow { + type: string; + avg7: number | null; + avg30: number | null; + slope7: number | null; + r2_7: number | null; + slope30: number | null; + r2_30: number | null; + slope90: number | null; + r2_90: number | null; +} + interface LatestRow { type: string; value: number; @@ -146,26 +175,39 @@ export interface SummariesSlice { export async function computeSummariesSlice( userId: string, ): Promise { - // v1.4.35 — keep the persistent rollup table current before the - // bucket read fires. On a warm process this is a single watermark - // query; on a cold mount it folds the trailing 90-day window into - // the DAY rollup so the bucket read below has something to compose. + // Keep the persistent rollup table current before the bucket read + // fires. On a warm process this is a single watermark query; on a + // cold mount it folds the trailing 90-day window into the DAY + // rollup so the bucket read below has something to compose. await ensureUserRollupsFresh(userId); - // Three passes in parallel. The first two target the same - // `measurements (user_id, type, measured_at)` index path; the third - // hits the much smaller `measurement_rollups` table (one row per - // bucket, not per row). On users with a complete backfill the - // bucket scan dominates only marginally; on un-backfilled users it - // returns a short list and the parity check falls back to live. - const [aggregates, latests, dayBuckets] = await Promise.all([ - prisma.$queryRaw` + // Cheap existence probe — one indexed COUNT against the rollup + // table. A non-zero count means there's at least one DAY bucket we + // can compose from, so the heavy live aggregate is unnecessary. + const rollupCountRows = await prisma.$queryRaw>` + SELECT COUNT(*) AS n + FROM measurement_rollups + WHERE user_id = ${userId} + AND granularity = 'DAY' + `; + const hasRollupCoverage = Number(rollupCountRows[0]?.n ?? 0) > 0; + + if (hasRollupCoverage) { + return computeFromRollups(userId); + } + return computeFromLiveAggregate(userId); +} + +/** + * Happy path — DAY buckets carry `count / min / max / mean`; a narrow + * `$queryRaw` carries only the windowed avgs + regression columns the + * buckets cannot reconstruct. + */ +async function computeFromRollups(userId: string): Promise { + const [narrows, latests, dayBuckets] = await Promise.all([ + prisma.$queryRaw` SELECT m."type"::text AS type, - COUNT(*) AS count, - MIN(m."value")::double precision AS min_value, - MAX(m."value")::double precision AS max_value, - AVG(m."value")::double precision AS mean_value, AVG(m."value") FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '7 days' )::double precision AS avg7, @@ -301,26 +343,165 @@ export async function computeSummariesSlice( bucketsByType.set(b.type, list); } + const narrowByType = new Map(); + for (const row of narrows) { + narrowByType.set(row.type, row); + } + + let totalRows = 0; + for (const [type, buckets] of bucketsByType.entries()) { + const composed = aggregateBuckets(buckets); + if (composed.count === 0) continue; + totalRows += composed.count; + const narrow = narrowByType.get(type); + const latest = latestByType.get(type) ?? null; + summaries[type] = { + count: composed.count, + latest, + min: round2(composed.min), + max: round2(composed.max), + mean: round2(composed.mean), + avg7: round2(narrow?.avg7 ?? null), + avg30: round2(narrow?.avg30 ?? null), + slope7: buildSlope(narrow?.slope7 ?? null, narrow?.r2_7 ?? null), + slope30: buildSlope(narrow?.slope30 ?? null, narrow?.r2_30 ?? null), + slope90: buildSlope(narrow?.slope90 ?? null, narrow?.r2_90 ?? null), + anomalyCount: 0, + avg30LastMonth: null, + avg30LastYear: null, + }; + } + + annotate({ + action: { name: "analytics.get.slim" }, + meta: { + analytics: { + slim_summaries: { + row_count: totalRows, + type_count: bucketsByType.size, + path: "rollup", + }, + }, + }, + }); + + return { summaries, bmi: null, lastSeenByType }; +} + +/** + * Cold fallback — runs the legacy heavy aggregate when the rollup + * table is empty for this user. Subsequent reads pick up the populated + * rollup on `computeFromRollups`. + */ +async function computeFromLiveAggregate( + userId: string, +): Promise { + const [aggregates, latests] = await Promise.all([ + prisma.$queryRaw` + SELECT + m."type"::text AS type, + COUNT(*) AS count, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value, + AVG(m."value")::double precision AS mean_value, + AVG(m."value") FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + )::double precision AS avg7, + AVG(m."value") FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + )::double precision AS avg30, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + )::double precision AS slope7, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + )::double precision AS r2_7, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + )::double precision AS slope30, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + )::double precision AS r2_30, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + )::double precision AS slope90, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + )::double precision AS r2_90 + FROM measurements m + WHERE m."user_id" = ${userId} + GROUP BY m."type" + `, + prisma.$queryRaw` + SELECT DISTINCT ON (m."type") + m."type"::text AS type, + m."value"::double precision AS value, + m."measured_at" AS measured_at + FROM measurements m + WHERE m."user_id" = ${userId} + ORDER BY m."type", m."measured_at" DESC + `, + ]); + + const latestByType = new Map(); + const lastSeenAtByType = new Map(); + for (const row of latests) { + latestByType.set(row.type, Number(row.value)); + if (row.measured_at) { + lastSeenAtByType.set(row.type, new Date(row.measured_at)); + } + } + + const summaries: Record = {}; + const lastSeenByType: Record< + string, + { lastSeenAt: string; daysAgo: number } | null + > = {}; + for (const type of measurementTypeEnum.options) { + summaries[type] = emptySummary(); + lastSeenByType[type] = null; + } + + const nowForStaleness = Date.now(); + for (const [type, measuredAt] of lastSeenAtByType.entries()) { + const daysAgo = Math.floor( + (nowForStaleness - measuredAt.getTime()) / (24 * 60 * 60 * 1000), + ); + lastSeenByType[type] = { + lastSeenAt: measuredAt.toISOString(), + daysAgo, + }; + } + let totalRows = 0; for (const row of aggregates) { - const liveCount = Number(row.count); - // Compose count/min/max/mean from the DAY buckets when the - // composed count agrees with the live `COUNT(*)`. The slim slice's - // window is all-time, so a divergent count signals either an - // un-backfilled account (rollup only covers the trailing 90d - // `ensureUserRollupsFresh` warmed up) or a mid-flight populator — - // both cases want the live SQL values, which we already have. - const composed = aggregateBuckets(bucketsByType.get(row.type) ?? []); - const useRollup = composed.count === liveCount; - const count = useRollup ? composed.count : liveCount; + const count = Number(row.count); totalRows += count; const latest = latestByType.get(row.type) ?? null; summaries[row.type] = { count, latest, - min: useRollup ? round2(composed.min) : round2(row.min_value), - max: useRollup ? round2(composed.max) : round2(row.max_value), - mean: useRollup ? round2(composed.mean) : round2(row.mean_value), + min: round2(row.min_value), + max: round2(row.max_value), + mean: round2(row.mean_value), avg7: round2(row.avg7), avg30: round2(row.avg30), slope7: buildSlope(row.slope7, row.r2_7), @@ -339,6 +520,7 @@ export async function computeSummariesSlice( slim_summaries: { row_count: totalRows, type_count: aggregates.length, + path: "live", }, }, }, diff --git a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts index 892d68919..6d7024936 100644 --- a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts +++ b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts @@ -1,19 +1,21 @@ /** - * v1.4.35 — unit pin for `buildComprehensiveAggregate`. + * Unit pin for `buildComprehensiveAggregate`. * * Heavier integration coverage (real Postgres, real `regr_slope`, - * cache hit/miss assertions) lives in - * `tests/integration/insights-comprehensive-cache.test.ts`. This file - * mocks `$queryRaw` and Prisma's `measurement.findMany` so the shape + - * rounding + slope-direction contracts are pinned without a container. + * cache hit/miss assertions, byte-identical parity with live SQL) + * lives in `tests/integration/insights-comprehensive-cache.test.ts` + * and `tests/integration/measurement-rollups.test.ts`. This file + * mocks `$queryRaw` and Prisma's `measurement.findMany` so the shape + * + rounding + slope-direction + read-path-selection contracts are + * pinned without a container. */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: vi.fn(), - measurement: { findMany: vi.fn() }, - // v1.4.35 — the rollup read inside the Promise.all + the + measurement: { findMany: vi.fn(), findFirst: vi.fn() }, + // v1.4.36 — the rollup read inside the Promise.all + the // freshness watermark inside `ensureUserRollupsFresh`. We mock // both so the unit test doesn't need a real container. measurementRollup: { @@ -33,6 +35,8 @@ import { buildComprehensiveAggregate } from "../comprehensive-aggregator"; const RAW = prisma.$queryRaw as unknown as ReturnType; const FIND_MANY = prisma.measurement.findMany as unknown as ReturnType; +const MEASUREMENT_FIND_FIRST = + prisma.measurement.findFirst as unknown as ReturnType; const ROLLUP_FIND_MANY = prisma.measurementRollup.findMany as unknown as ReturnType; const ROLLUP_FIND_FIRST = @@ -41,15 +45,16 @@ const ROLLUP_FIND_FIRST = beforeEach(() => { RAW.mockReset(); FIND_MANY.mockReset(); + MEASUREMENT_FIND_FIRST.mockReset(); ROLLUP_FIND_MANY.mockReset(); ROLLUP_FIND_FIRST.mockReset(); - // `ensureUserRollupsFresh` calls `prisma.measurement.findFirst`; - // we don't have it on the mock above, so it throws and the helper - // swallows + returns `{ recomputed: false }`. The bucket read - // proceeds against whatever we mock on `measurementRollup.findMany`. - // We default both to empty so individual tests opt in to data. + // `ensureUserRollupsFresh` calls `prisma.measurement.findFirst` and + // `prisma.measurementRollup.findFirst` in parallel. Default both to + // null so the warm-up is a no-op; individual tests opt in to a + // freshness mismatch by overriding the mocks. ROLLUP_FIND_MANY.mockResolvedValue([]); ROLLUP_FIND_FIRST.mockResolvedValue(null); + MEASUREMENT_FIND_FIRST.mockResolvedValue(null); }); afterEach(() => { @@ -57,153 +62,229 @@ afterEach(() => { }); describe("buildComprehensiveAggregate", () => { - it("returns an empty bundle for a user with no measurements", async () => { - // v1.4.35 pass order: aggregates ($queryRaw), latests ($queryRaw), - // dayBuckets (rollup.findMany — defaulted to [] in beforeEach), - // sysRaw (findMany), diaRaw (findMany). The `firstMeasurementAt` - // $queryRaw is skipped when totalMeasurements === 0. - RAW.mockResolvedValueOnce([]) // aggregates - .mockResolvedValueOnce([]); // latests - FIND_MANY.mockResolvedValueOnce([]) // sys - .mockResolvedValueOnce([]); // dia + describe("rollup-fresh happy path", () => { + it("skips the heavy live aggregate when the rollup table is populated", async () => { + const now = new Date(); + // 1. rollup-count probe — returns 2 (non-zero ⇒ happy path). + // 2. narrow aggregate ($queryRaw) — windowed/regression columns only. + // 3. latests ($queryRaw). + // 4. firstMeasurementAt ($queryRaw). + RAW.mockResolvedValueOnce([{ n: BigInt(2) }]) + .mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(42), + stddev_value: 1.2, + anomaly_count: BigInt(3), + avg7: 81.9, + avg30: 82.1, + avg30_last_month: 83.0, + slope7: -0.014, + r2_7: 0.65, + slope30: -0.005, + r2_30: 0.42, + slope90: 0.001, + r2_90: 0.12, + }, + ]) + .mockResolvedValueOnce([ + { type: "WEIGHT", value: 81.4, measured_at: now }, + ]) + .mockResolvedValueOnce([ + { first_at: new Date(now.getTime() - 86400000) }, + ]); - const result = await buildComprehensiveAggregate("user-empty"); + // DAY buckets compose to count=42, min=79.2, max=84.1, mean=82.05. + // The summary's count/min/max/mean read from these buckets — NOT + // from a heavy live aggregate column. + ROLLUP_FIND_MANY.mockResolvedValueOnce([ + { + type: "WEIGHT", + bucketStart: new Date("2026-05-10T00:00:00.000Z"), + count: 20, + mean: 81.0, + minValue: 79.2, + maxValue: 82.5, + }, + { + type: "WEIGHT", + bucketStart: new Date("2026-05-11T00:00:00.000Z"), + count: 22, + mean: 83.0, + minValue: 80.0, + maxValue: 84.1, + }, + ]); - expect(result.summaries).toEqual({}); - expect(result.bpRawRows.sys).toEqual([]); - expect(result.bpRawRows.dia).toEqual([]); - expect(result.dailyByType).toEqual({}); - expect(result.firstMeasurementAt).toBeNull(); - expect(result.totalMeasurements).toBe(0); - // 2 $queryRaw (aggregates, latests) + 2 findMany + 1 rollup.findMany. - // No firstMeasurementAt query when total === 0. - expect(RAW).toHaveBeenCalledTimes(2); - expect(FIND_MANY).toHaveBeenCalledTimes(2); - expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1); + FIND_MANY.mockResolvedValueOnce([]) // sys + .mockResolvedValueOnce([]); // dia + + const result = await buildComprehensiveAggregate("user-rollup-fresh"); + const weight = result.summaries.WEIGHT; + + // count/min/max/mean composed from buckets — the test asserts the + // bucket-derived values are present even though we never wired up + // a heavy aggregate row for the parity check. + expect(weight.count).toBe(42); + expect(weight.min).toBe(79.2); + expect(weight.max).toBe(84.1); + // Weighted mean = (20 * 81 + 22 * 83) / 42 = 3446 / 42 ≈ 82.0476 + // → round2 → 82.05. + expect(weight.mean).toBe(82.05); + // Windowed/regression columns come from the narrow aggregate. + expect(weight.latest).toBe(81.4); + expect(weight.avg7).toBe(81.9); + expect(weight.avg30).toBe(82.1); + expect(weight.avg30LastMonth).toBe(83.0); + expect(weight.anomalyCount).toBe(3); + expect(weight.slope7).toEqual({ + slope: -0.014, + direction: "down", + confidence: 0.65, + }); + expect(result.dailyByType.WEIGHT).toEqual([ + { day: "2026-05-10", value: 81 }, + { day: "2026-05-11", value: 83 }, + ]); + + // Contract pin — the heavy aggregate path is NOT exercised on the + // rollup-fresh branch. The $queryRaw calls land on the COUNT + // probe + narrow aggregate + DISTINCT-ON latest + first_at, NOT + // the legacy heavy COUNT/MIN/MAX/AVG query. We assert by call + // count + by checking that the narrow projection (no min_value / + // max_value / mean_value cols) is what landed. + expect(RAW).toHaveBeenCalledTimes(4); + expect(FIND_MANY).toHaveBeenCalledTimes(2); + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1); + }); }); - it("maps a populated WEIGHT aggregate into the DataSummary shape", async () => { - const now = new Date(); - RAW.mockResolvedValueOnce([ - { - type: "WEIGHT", - count: BigInt(42), - min_value: 79.2, - max_value: 84.1, - mean_value: 82.05, - stddev_value: 1.2, - anomaly_count: BigInt(3), - avg7: 81.9, - avg30: 82.1, - avg30_last_month: 83.0, - slope7: -0.014, - r2_7: 0.65, - slope30: -0.005, - r2_30: 0.42, - slope90: 0.001, - r2_90: 0.12, - }, - ]) - .mockResolvedValueOnce([{ type: "WEIGHT", value: 81.4, measured_at: now }]) - .mockResolvedValueOnce([{ first_at: new Date(now.getTime() - 86400000) }]); - // v1.4.35 — DAY buckets feed both the per-type count/min/max/mean - // composition AND `dailyByType`. The two buckets compose to a - // count of 42 → matches the live aggregate's BigInt(42) → the - // parity check elects the rollup-derived values, which equal the - // live values byte-for-byte (chosen on purpose so the assertions - // exercise the rollup path while keeping the previous expected - // numbers stable). - ROLLUP_FIND_MANY.mockResolvedValueOnce([ - { - type: "WEIGHT", - bucketStart: new Date("2026-05-10T00:00:00.000Z"), - count: 20, - mean: 81.0, - minValue: 79.2, - maxValue: 82.5, - }, - { - type: "WEIGHT", - bucketStart: new Date("2026-05-11T00:00:00.000Z"), - count: 22, - mean: 83.0, - minValue: 80.0, - maxValue: 84.1, - }, - ]); - FIND_MANY.mockResolvedValueOnce([]) // sys - .mockResolvedValueOnce([]); // dia + describe("cold fallback when no rollup buckets exist", () => { + it("returns an empty bundle for a user with no measurements and no rollups", async () => { + // 1. rollup-count probe — zero rows ⇒ cold path. + // 2. heavy aggregate ($queryRaw) — empty. + // 3. latests ($queryRaw) — empty. + // No firstMeasurementAt query when totalMeasurements === 0. + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([]); + FIND_MANY.mockResolvedValueOnce([]) // sys + .mockResolvedValueOnce([]); // dia - const result = await buildComprehensiveAggregate("user-pop"); - const weight = result.summaries.WEIGHT; + const result = await buildComprehensiveAggregate("user-empty"); - expect(weight.count).toBe(42); - expect(weight.latest).toBe(81.4); - // Composed from buckets: min(79.2, 80.0) = 79.2; max(82.5, 84.1) = 84.1. - expect(weight.min).toBe(79.2); - expect(weight.max).toBe(84.1); - // Weighted mean = (20 * 81 + 22 * 83) / 42 = 3446 / 42 = 82.0476…, - // round2 → 82.05. Matches the live `AVG` value of 82.05. - expect(weight.mean).toBe(82.05); - expect(weight.avg7).toBe(81.9); - expect(weight.avg30).toBe(82.1); - expect(weight.avg30LastMonth).toBe(83.0); - // Always null in the 90-day window — preserved legacy semantics. - expect(weight.avg30LastYear).toBeNull(); - expect(weight.anomalyCount).toBe(3); - // Slope direction matches the JS threshold rules: - // -0.014 → "down" (|s| >= 0.01) - // -0.005 → "stable" (|s| < 0.01) - // 0.001 → "stable" - expect(weight.slope7).toEqual({ - slope: -0.014, - direction: "down", - confidence: 0.65, - }); - expect(weight.slope30).toEqual({ - slope: -0.005, - direction: "stable", - confidence: 0.42, - }); - expect(weight.slope90).toEqual({ - slope: 0.001, - direction: "stable", - confidence: 0.12, + expect(result.summaries).toEqual({}); + expect(result.bpRawRows.sys).toEqual([]); + expect(result.bpRawRows.dia).toEqual([]); + expect(result.dailyByType).toEqual({}); + expect(result.firstMeasurementAt).toBeNull(); + expect(result.totalMeasurements).toBe(0); + expect(RAW).toHaveBeenCalledTimes(3); + expect(FIND_MANY).toHaveBeenCalledTimes(2); + // The cold path's rollup.findMany still fires (in case some + // buckets exist for a subset of types post-race), but returns []. + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1); }); - expect(result.totalMeasurements).toBe(42); - expect(result.firstMeasurementAt).toBeInstanceOf(Date); - // `dailyByType` is sourced from the same DAY buckets; the entries - // are the bucket means rounded to 2 decimals, keyed on the UTC - // bucket-start date. - expect(result.dailyByType.WEIGHT).toEqual([ - { day: "2026-05-10", value: 81 }, - { day: "2026-05-11", value: 83 }, - ]); + it("runs the heavy aggregate when no rollup rows exist yet", async () => { + const now = new Date(); + // 1. rollup-count probe — zero rows ⇒ cold path. + // 2. heavy aggregate ($queryRaw) — populated. + // 3. latests ($queryRaw) — populated. + // 4. firstMeasurementAt ($queryRaw). + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(42), + min_value: 79.2, + max_value: 84.1, + mean_value: 82.05, + stddev_value: 1.2, + anomaly_count: BigInt(3), + avg7: 81.9, + avg30: 82.1, + avg30_last_month: 83.0, + slope7: -0.014, + r2_7: 0.65, + slope30: -0.005, + r2_30: 0.42, + slope90: 0.001, + r2_90: 0.12, + }, + ]) + .mockResolvedValueOnce([ + { type: "WEIGHT", value: 81.4, measured_at: now }, + ]) + .mockResolvedValueOnce([ + { first_at: new Date(now.getTime() - 86400000) }, + ]); + FIND_MANY.mockResolvedValueOnce([]) // sys + .mockResolvedValueOnce([]); // dia + + const result = await buildComprehensiveAggregate("user-cold"); + const weight = result.summaries.WEIGHT; + + expect(weight.count).toBe(42); + expect(weight.latest).toBe(81.4); + expect(weight.min).toBe(79.2); + expect(weight.max).toBe(84.1); + expect(weight.mean).toBe(82.05); + expect(weight.avg7).toBe(81.9); + expect(weight.avg30).toBe(82.1); + expect(weight.avg30LastMonth).toBe(83.0); + expect(weight.avg30LastYear).toBeNull(); + expect(weight.anomalyCount).toBe(3); + expect(weight.slope7).toEqual({ + slope: -0.014, + direction: "down", + confidence: 0.65, + }); + expect(weight.slope30).toEqual({ + slope: -0.005, + direction: "stable", + confidence: 0.42, + }); + expect(weight.slope90).toEqual({ + slope: 0.001, + direction: "stable", + confidence: 0.12, + }); + + expect(result.totalMeasurements).toBe(42); + expect(result.firstMeasurementAt).toBeInstanceOf(Date); + // The cold path's `dailyByType` is sourced from the same rollup + // findMany call (returns [] here since no buckets exist). The + // contract is: even on cold mount, if buckets do exist for some + // type, they feed `dailyByType` so the v1.4.35 shape is stable. + expect(result.dailyByType.WEIGHT).toBeUndefined(); + }); }); it("threads sys/dia raw rows through bpRawRows so 5-min pairing survives", async () => { const measuredAt = new Date("2026-05-10T08:00:00Z"); - RAW.mockResolvedValueOnce([ - { - type: "BLOOD_PRESSURE_SYS", - count: BigInt(1), - min_value: 120, - max_value: 120, - mean_value: 120, - stddev_value: 0, - anomaly_count: BigInt(0), - avg7: 120, - avg30: 120, - avg30_last_month: null, - slope7: null, - r2_7: null, - slope30: null, - r2_30: null, - slope90: null, - r2_90: null, - }, - ]) + // Cold path so the heavy aggregate fires (no rollup count). + RAW.mockResolvedValueOnce([{ n: BigInt(0) }]) + .mockResolvedValueOnce([ + { + type: "BLOOD_PRESSURE_SYS", + count: BigInt(1), + min_value: 120, + max_value: 120, + mean_value: 120, + stddev_value: 0, + anomaly_count: BigInt(0), + avg7: 120, + avg30: 120, + avg30_last_month: null, + slope7: null, + r2_7: null, + slope30: null, + r2_30: null, + slope90: null, + r2_90: null, + }, + ]) .mockResolvedValueOnce([ { type: "BLOOD_PRESSURE_SYS", diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index 3ebecbbb5..f3a17de59 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -1,6 +1,6 @@ /** - * v1.4.35 — SQL-driven aggregator for `/api/insights/comprehensive`, - * with the partial read-swap onto `measurement_rollups` landed. + * v1.4.36 — SQL-driven aggregator for `/api/insights/comprehensive`, + * rollup-first read with live-SQL fallback on cold mounts. * * Background * ---------- @@ -21,39 +21,42 @@ * slim slice: the comprehensive endpoint also needs `anomalyCount`, * `avg30LastMonth`, `avg30LastYear`, and the 90-day window cap. * - * Read-swap (v1.4.35) - * ------------------- - * Two surfaces now read from the persistent `measurement_rollups` - * table instead of recomputing per request: + * Read-swap (v1.4.35 → v1.4.36) + * ----------------------------- + * v1.4.35 landed a partial read-swap that ran both the live aggregate + * AND the rollup read in parallel, then preferred the rollup-derived + * values when the count matched. That left the regression intact: the + * heavy `$queryRaw` over `measurements` still ran on every request, + * still scanning the 311k-row table on a power-user account. * - * - **`count / min / max / mean`** per type — composed from the DAY - * rollup buckets via `aggregateBuckets` (sum of counts; min of - * mins; max of maxes; count-weighted mean). These four are - * linearly composable across DAY buckets, so the composed value - * is mathematically identical to the live `AVG` / `MIN` / `MAX` - * / `COUNT(*)` over the same rows. A defensive parity check - * against the live aggregate's `COUNT(*)` falls back to live SQL - * when divergence is detected — covers the cold-mount edge case - * where the rollup populator is mid-flight, or a sub-second race - * between the watermark check and the bucket read. + * v1.4.36 promotes the rollup table to the canonical source on the + * happy path: * - * - **`dailyByType`** — read directly from the DAY rollup buckets - * keyed on (userId, type, granularity=DAY, bucketStart). The - * bucket's stored `mean` is the same `AVG(value)` per day the - * legacy SQL emitted via `date_trunc('day', measured_at)`, so - * downstream correlation pairings remain byte-shape stable. + * - On entry we call `ensureUserRollupsFresh` (warm-on-read populator + * for the trailing 90-day DAY window) and then probe the rollup + * table with a single COUNT. * - * Everything else stays on live SQL: - * - `slope7 / r2_7`, `slope30 / r2_30`, `slope90 / r2_90` — slope - * and R² do **not** compose linearly across DAY buckets. - * - `stddev`, `anomalyCount` — same reason. - * - `avg7 / avg30 / avg30LastMonth` — these windows aren't aligned - * to DAY-bucket boundaries (the 7-day window slides relative to - * `NOW()`), so the live `FILTER (WHERE measured_at >= NOW() - - * INTERVAL '7 days')` clause is canonical. - * - `latest / latestMeasuredAt` — the `DISTINCT ON (type)` pass - * resolves the most recent raw row, which the bucket can't - * surface (it stores the day's mean, not the day's last value). + * - When the DAY rollup table has rows for this user the **heavy** + * `$queryRaw` over `measurements` is skipped. `count / min / max / + * mean` per type compose from the persisted DAY buckets via + * `aggregateBuckets`; `dailyByType` reads the same buckets keyed + * on `(type, bucketStart)`. + * + * - A **narrow** `$queryRaw` still fires for the non-composable + * fields: `stddev`, `anomalyCount`, `avg7 / avg30 / avg30LastMonth`, + * and the `slope7 / slope30 / slope90` tuples. These don't compose + * linearly across DAY buckets so live SQL stays canonical, but the + * query no longer carries the redundant `COUNT(*)` / `MIN` / `MAX` + * / `AVG` columns that duplicated the bucket-derived values. + * + * - `latest / latestMeasuredAt` keep their dedicated `DISTINCT ON + * (type)` pass; the bucket stores the day's mean rather than the + * last reading. + * + * - Fallback path — when the COUNT probe returns zero rows + * (post-restart cold mount or a brand-new account whose boot-time + * backfill is mid-flight) we run the legacy heavy aggregate so the + * response shape is correct on the very first request. * * Scope notes * ----------- @@ -100,12 +103,12 @@ import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; import { aggregateBuckets } from "@/lib/measurements/rollup-read"; /** - * Raw row from the aggregate query. Keyed on `MeasurementType`, with - * windowed AVG/SLOPE/R2 expressions plus anomaly-detection inputs - * (window-wide mean + stddev so we can compute the |z| > 2 count - * client-side without a second pass over the raw rows). + * Heavy aggregate row — count/min/max/mean alongside the non-composable + * windowed columns. Used on the cold-mount fallback path where the + * rollup table is empty (the populator hasn't caught up yet) so the + * response shape is correct on the very first request. */ -interface AggregateRow { +interface HeavyAggregateRow { type: string; count: bigint; min_value: number | null; @@ -124,6 +127,29 @@ interface AggregateRow { r2_90: number | null; } +/** + * Narrow aggregate row — only the columns the DAY rollup buckets cannot + * compose linearly. `count` rides along so the anomaly z-score loop + * knows the divisor for the window-wide mean / stddev; the per-type + * count exposed on `DataSummary` still comes from the bucket + * composition. + */ +interface NarrowAggregateRow { + type: string; + count: bigint; + stddev_value: number | null; + anomaly_count: bigint; + avg7: number | null; + avg30: number | null; + avg30_last_month: number | null; + slope7: number | null; + r2_7: number | null; + slope30: number | null; + r2_30: number | null; + slope90: number | null; + r2_90: number | null; +} + /** Latest raw value per type (DISTINCT ON over the 90-day window). */ interface LatestRow { type: string; @@ -183,35 +209,63 @@ function round2(value: number | null): number | null { /** * Build the per-type `DataSummary` bundle for the comprehensive route. * - * Five queries against the trailing 90 days, in parallel: + * Two read shapes: * - * 1. Per-type aggregate (count + min/max/mean/stddev + windowed - * avgs + regr_slope/r2 for the 7/30/90-day slopes). The - * `count / min / max / mean` columns are kept on this query as - * the live-SQL fallback for the v1.4.35 read-swap parity check. - * 2. `DISTINCT ON (type)` for the most-recent raw value per type. - * 3. DAY rollup buckets from `measurement_rollups` (v1.4.35) — - * composed into `count / min / max / mean` for the per-type - * summary and replayed as the `dailyByType` correlation feed. - * 4. Bounded raw-row sys query for BP target pairing (sub-day - * timestamps, can't be replaced by a daily mean). - * 5. Bounded raw-row dia query for BP target pairing. + * - **Rollup-fresh path** (the happy path on every warm request): + * `count / min / max / mean` + `dailyByType` compose from the + * persisted DAY buckets; a narrow `$queryRaw` carries only the + * non-composable windowed columns. The heavy COUNT/MIN/MAX/AVG + * aggregate is **not** executed. + * + * - **Live fallback path** (cold mount before the populator catches + * up, or a brand-new account whose backfill is mid-flight): the + * legacy heavy aggregate runs so the response shape stays correct + * on the very first request. Subsequent requests fall back to the + * rollup-fresh path once `ensureUserRollupsFresh` has written + * buckets. */ export async function buildComprehensiveAggregate( userId: string, ): Promise { const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000); - // v1.4.35 — keep the persistent rollup table current before any - // read fires. After this returns the DAY-granularity buckets for - // the trailing 90 days reflect every measurement that landed before - // this call (the per-write hook is synchronous on the DAY bucket, - // and `ensureUserRollupsFresh` covers cold-mount / process-restart - // cases). The bucket reads below depend on that guarantee. + // Keep the persistent rollup table current before any read fires. + // After this returns the DAY-granularity buckets for the trailing + // 90 days reflect every measurement that landed before this call + // (the per-write hook is synchronous on the DAY bucket, and + // `ensureUserRollupsFresh` covers cold-mount / process-restart). await ensureUserRollupsFresh(userId); - const [aggregates, latests, dayBuckets, sysRaw, diaRaw] = await Promise.all([ - prisma.$queryRaw` + // Cheap existence check — one indexed COUNT against the rollup + // table. A non-zero count means there's at least one DAY bucket to + // compose against, so the heavy live aggregate is unnecessary. When + // it's zero we fall through to the live aggregate so the response + // shape stays correct on a cold mount. + const rollupCountRows = await prisma.$queryRaw>` + SELECT COUNT(*) AS n + FROM measurement_rollups + WHERE user_id = ${userId} + AND granularity = 'DAY' + `; + const hasRollupCoverage = Number(rollupCountRows[0]?.n ?? 0) > 0; + + if (hasRollupCoverage) { + return buildFromRollups(userId, ninetyDaysAgo); + } + return buildFromLiveAggregate(userId, ninetyDaysAgo); +} + +/** + * Happy path — DAY buckets carry the composable stats; a narrow + * `$queryRaw` carries only the windowed / regression columns the + * buckets cannot reconstruct. + */ +async function buildFromRollups( + userId: string, + ninetyDaysAgo: Date, +): Promise { + const [narrows, latests, dayBuckets, sysRaw, diaRaw, firstRows] = await Promise.all([ + prisma.$queryRaw` WITH window_stats AS ( SELECT m."type", @@ -225,9 +279,6 @@ export async function buildComprehensiveAggregate( SELECT m."type"::text AS type, COUNT(*) AS count, - MIN(m."value")::double precision AS min_value, - MAX(m."value")::double precision AS max_value, - AVG(m."value")::double precision AS mean_value, ws.stddev_value::double precision AS stddev_value, -- |z| > 2 anomaly count over the 90-day window. STDDEV_POP -- matches the JS helper which divides by n (not n-1). @@ -333,12 +384,14 @@ export async function buildComprehensiveAggregate( orderBy: { measuredAt: "asc" }, select: { measuredAt: true, value: true }, }), + prisma.$queryRaw>` + SELECT MIN(m."measured_at") AS first_at + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${ninetyDaysAgo} + `, ]); - // ── avg30LastYear ───────────────────────────────────────── - // The 90-day window guarantees this is null (no rows >= 365 days - // old will appear) — preserves legacy semantics. - const latestByType = new Map(); for (const row of latests) { latestByType.set(row.type, { @@ -347,53 +400,214 @@ export async function buildComprehensiveAggregate( }); } - // v1.4.35 — partition DAY buckets by type so the per-type aggregate - // loop below can compose `count / min / max / mean` from them in O(1). - const bucketsByType = new Map< - string, - Array<{ - day: Date; - count: number; - mean: number; - minValue: number; - maxValue: number; - }> - >(); - for (const b of dayBuckets) { - const list = bucketsByType.get(b.type) ?? []; - list.push({ - day: b.bucketStart, - count: b.count, - mean: b.mean, - minValue: b.minValue, - maxValue: b.maxValue, - }); - bucketsByType.set(b.type, list); + const bucketsByType = partitionBucketsByType(dayBuckets); + const narrowByType = new Map(); + for (const row of narrows) { + narrowByType.set(row.type, row); } - // Seed only types that actually have rows — the legacy route only - // populated `summaries[t]` when `data.length > 0`. + const summaries: Record = {}; + let totalMeasurements = 0; + + // Seed every type that has buckets — the bucket set is the source of + // truth on this path. The narrow aggregate provides the non-composable + // windowed columns alongside. + for (const [type, buckets] of bucketsByType.entries()) { + const composed = aggregateBuckets(buckets); + if (composed.count === 0) continue; + totalMeasurements += composed.count; + const latest = latestByType.get(type); + const narrow = narrowByType.get(type); + summaries[type] = { + count: composed.count, + latest: latest?.value ?? null, + min: round2(composed.min), + max: round2(composed.max), + mean: round2(composed.mean), + avg7: round2(narrow?.avg7 ?? null), + avg30: round2(narrow?.avg30 ?? null), + slope7: buildSlope(narrow?.slope7 ?? null, narrow?.r2_7 ?? null), + slope30: buildSlope(narrow?.slope30 ?? null, narrow?.r2_30 ?? null), + slope90: buildSlope(narrow?.slope90 ?? null, narrow?.r2_90 ?? null), + anomalyCount: Number(narrow?.anomaly_count ?? 0), + avg30LastMonth: round2(narrow?.avg30_last_month ?? null), + // Legacy semantics: the 90-day window guarantees no rows from a + // year ago. Always null. + avg30LastYear: null, + }; + } + + const firstMeasurementAt = firstRows[0]?.first_at + ? new Date(firstRows[0].first_at) + : null; + + return { + summaries, + bpRawRows: { + sys: sysRaw.map((r) => ({ measuredAt: r.measuredAt, value: r.value })), + dia: diaRaw.map((r) => ({ measuredAt: r.measuredAt, value: r.value })), + }, + dailyByType: buildDailyByType(bucketsByType), + firstMeasurementAt, + totalMeasurements, + }; +} + +/** + * Cold fallback — the rollup table is empty for this user (e.g. brand- + * new account, or post-restart before the boot-time backfill landed). + * Runs the legacy heavy aggregate so the response shape is correct on + * the very first request; subsequent requests will see the rollup + * populator catch up and route through `buildFromRollups`. + */ +async function buildFromLiveAggregate( + userId: string, + ninetyDaysAgo: Date, +): Promise { + const [aggregates, latests, dayBuckets, sysRaw, diaRaw] = await Promise.all([ + prisma.$queryRaw` + WITH window_stats AS ( + SELECT + m."type", + AVG(m."value") AS mean_value, + STDDEV_POP(m."value") AS stddev_value + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${ninetyDaysAgo} + GROUP BY m."type" + ) + SELECT + m."type"::text AS type, + COUNT(*) AS count, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value, + AVG(m."value")::double precision AS mean_value, + ws.stddev_value::double precision AS stddev_value, + COUNT(*) FILTER ( + WHERE ws.stddev_value IS NOT NULL + AND ws.stddev_value > 0 + AND ABS( + (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2)) + ) > 2 + ) AS anomaly_count, + AVG(m."value") FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + )::double precision AS avg7, + AVG(m."value") FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + )::double precision AS avg30, + AVG(m."value") FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '60 days' + AND m."measured_at" < NOW() - INTERVAL '30 days' + )::double precision AS avg30_last_month, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + )::double precision AS slope7, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + )::double precision AS r2_7, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + )::double precision AS slope30, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + )::double precision AS r2_30, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + )::double precision AS slope90, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + ) FILTER ( + WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + )::double precision AS r2_90 + FROM measurements m + JOIN window_stats ws ON ws."type" = m."type" + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${ninetyDaysAgo} + GROUP BY m."type", ws.stddev_value + `, + prisma.$queryRaw` + SELECT DISTINCT ON (m."type") + m."type"::text AS type, + m."value"::double precision AS value, + m."measured_at" AS measured_at + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${ninetyDaysAgo} + ORDER BY m."type", m."measured_at" DESC + `, + // Buckets may exist for some types even when the COUNT probe came + // back zero (race window between the probe and a sibling + // populator); the cold-path read still honours them for the + // `dailyByType` feed. + prisma.measurementRollup.findMany({ + where: { + userId, + granularity: "DAY", + bucketStart: { gte: ninetyDaysAgo }, + }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }], + }), + prisma.measurement.findMany({ + where: { + userId, + type: "BLOOD_PRESSURE_SYS", + measuredAt: { gte: ninetyDaysAgo }, + }, + orderBy: { measuredAt: "asc" }, + select: { measuredAt: true, value: true }, + }), + prisma.measurement.findMany({ + where: { + userId, + type: "BLOOD_PRESSURE_DIA", + measuredAt: { gte: ninetyDaysAgo }, + }, + orderBy: { measuredAt: "asc" }, + select: { measuredAt: true, value: true }, + }), + ]); + + const latestByType = new Map(); + for (const row of latests) { + latestByType.set(row.type, { + value: Number(row.value), + measuredAt: new Date(row.measured_at), + }); + } + + const bucketsByType = partitionBucketsByType(dayBuckets); + const summaries: Record = {}; let totalMeasurements = 0; let firstMeasurementAt: Date | null = null; for (const row of aggregates) { - const liveCount = Number(row.count); - // Compose count/min/max/mean from the DAY buckets when they agree - // with the live `COUNT(*)`. Parity divergence (cold mount, rollup - // populator mid-flight, watermark race) falls back to live SQL so - // the response shape never serves a wrong value. - const composed = aggregateBuckets(bucketsByType.get(row.type) ?? []); - const useRollup = composed.count === liveCount; - const count = useRollup ? composed.count : liveCount; + const count = Number(row.count); totalMeasurements += count; const latest = latestByType.get(row.type); summaries[row.type] = { count, latest: latest?.value ?? null, - min: useRollup ? round2(composed.min) : round2(row.min_value), - max: useRollup ? round2(composed.max) : round2(row.max_value), - mean: useRollup ? round2(composed.mean) : round2(row.mean_value), + min: round2(row.min_value), + max: round2(row.max_value), + mean: round2(row.mean_value), avg7: round2(row.avg7), avg30: round2(row.avg30), slope7: buildSlope(row.slope7, row.r2_7), @@ -401,17 +615,10 @@ export async function buildComprehensiveAggregate( slope90: buildSlope(row.slope90, row.r2_90), anomalyCount: Number(row.anomaly_count), avg30LastMonth: round2(row.avg30_last_month), - // Legacy semantics: the 90-day findMany window guarantees no - // rows from a year ago. Always null. avg30LastYear: null, }; } - // ── firstMeasurementAt ─────────────────────────────────── - // Take the earliest `measuredAt` across the user's 90-day rows. - // One extra cheap aggregate query keeps the route's `dataSpanDays` - // truthful. Run lazily: skip when no aggregate rows came back - // (totalMeasurements === 0). if (totalMeasurements > 0) { const firstRows = await prisma.$queryRaw>` SELECT MIN(m."measured_at") AS first_at @@ -424,13 +631,84 @@ export async function buildComprehensiveAggregate( : null; } - // ── daily-by-type ──────────────────────────────────────── - // v1.4.35 — sourced from the DAY rollup buckets. The legacy - // implementation ran a dedicated `date_trunc('day', m."measured_at")` - // GROUP BY scan and only returned the three types the correlation - // pairings consume. We filter to the same three types here so - // downstream consumers see a byte-identical shape, but the data - // origin is now the persisted bucket — no extra SQL pass needed. + return { + summaries, + bpRawRows: { + sys: sysRaw.map((r) => ({ measuredAt: r.measuredAt, value: r.value })), + dia: diaRaw.map((r) => ({ measuredAt: r.measuredAt, value: r.value })), + }, + dailyByType: buildDailyByType(bucketsByType), + firstMeasurementAt, + totalMeasurements, + }; +} + +/** + * Partition DAY buckets by type so the per-type aggregator can compose + * `count / min / max / mean` from them in O(1) and the daily-by-type + * builder can replay them as the correlation feed. + */ +function partitionBucketsByType( + rows: ReadonlyArray<{ + type: string; + bucketStart: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + }>, +): Map< + string, + Array<{ + day: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + }> +> { + const map = new Map< + string, + Array<{ + day: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + }> + >(); + for (const b of rows) { + const list = map.get(b.type) ?? []; + list.push({ + day: b.bucketStart, + count: b.count, + mean: b.mean, + minValue: b.minValue, + maxValue: b.maxValue, + }); + map.set(b.type, list); + } + return map; +} + +/** + * `dailyByType` is sourced from DAY rollup buckets, filtered to the + * three types the correlation pairings consume so downstream consumers + * see a byte-identical shape with the legacy + * `date_trunc('day', m."measured_at")` GROUP BY query. + */ +function buildDailyByType( + bucketsByType: Map< + string, + Array<{ + day: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + }> + >, +): Record> { const dailyByType: Record> = {}; const DAILY_TYPES = ["WEIGHT", "BLOOD_PRESSURE_SYS", "PULSE"] as const; for (const type of DAILY_TYPES) { @@ -447,16 +725,6 @@ export async function buildComprehensiveAggregate( value: round2(b.mean) ?? 0, })); } - - return { - summaries, - bpRawRows: { - sys: sysRaw.map((r) => ({ measuredAt: r.measuredAt, value: r.value })), - dia: diaRaw.map((r) => ({ measuredAt: r.measuredAt, value: r.value })), - }, - dailyByType, - firstMeasurementAt, - totalMeasurements, - }; + return dailyByType; } diff --git a/tests/integration/measurement-rollups.test.ts b/tests/integration/measurement-rollups.test.ts index 4b66cab29..d856aa9a1 100644 --- a/tests/integration/measurement-rollups.test.ts +++ b/tests/integration/measurement-rollups.test.ts @@ -630,4 +630,108 @@ describe("measurement rollups — integration", () => { vi.mocked(getGlobalBoss).mockReturnValue(null as never); }); + + // ─── v1.4.36 rollup-fresh skip-live contract ────────────────── + // + // The v1.4.35 read-swap kept the heavy live aggregate running in + // parallel with the rollup read. v1.4.36 drops that — when the + // rollup table has DAY buckets for the user, the heavy aggregate + // is bypassed entirely. This test pins that the rollup-derived + // response is still byte-identical to a parallel live aggregate + // over the same rows AND that the response shape matches the + // pre-swap contract for every field. + + it("rollup-fresh response matches live SQL for every DataSummary field", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-skip-live-user", + email: "rollup-skip-live@example.test", + role: "USER", + }, + }); + + // 60 readings over 30 days, two types, deliberately spread so + // anomalies + slopes + windowed avgs all have signal. + const now = Date.now(); + const dayMs = 24 * 60 * 60 * 1000; + const seedRows: Array<{ + userId: string; + type: "WEIGHT" | "PULSE"; + value: number; + unit: string; + source: "MANUAL"; + measuredAt: Date; + }> = []; + for (let day = 0; day < 30; day++) { + seedRows.push({ + userId: user.id, + type: "WEIGHT", + value: 80 + day * 0.05 + ((day * 7) % 5) * 0.1, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - (29 - day) * dayMs - day * 60 * 1000), + }); + seedRows.push({ + userId: user.id, + type: "PULSE", + value: 65 + ((day * 3) % 11) * 0.4, + unit: "bpm", + source: "MANUAL", + measuredAt: new Date(now - (29 - day) * dayMs - day * 90 * 1000), + }); + } + await prisma.measurement.createMany({ data: seedRows }); + + // Pre-fold the rollup buckets so the rollup-fresh path engages. + await recomputeUserRollups(user.id, { granularities: ["DAY"] }); + + // Parallel live aggregation over the same 90-day window using the + // exact heavy-aggregate shape the cold-mount fallback runs. + const ninetyDaysAgo = new Date(now - 90 * dayMs); + const live = await prisma.$queryRaw< + Array<{ + type: string; + count: bigint; + min_value: number; + max_value: number; + mean_value: number; + }> + >` + SELECT + m."type"::text AS type, + COUNT(*) AS count, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value, + AVG(m."value")::double precision AS mean_value + FROM measurements m + WHERE m."user_id" = ${user.id} + AND m."measured_at" >= ${ninetyDaysAgo} + GROUP BY m."type" + `; + + const aggregate = await buildComprehensiveAggregate(user.id); + + const round2 = (n: number) => Math.round(n * 100) / 100; + + expect(live.length).toBeGreaterThan(0); + for (const liveRow of live) { + const summary = aggregate.summaries[liveRow.type]; + expect(summary, `summary for ${liveRow.type}`).toBeDefined(); + // count / min / max / mean compose from buckets — byte-identical + // to live SQL over the same rows. + expect(summary.count).toBe(Number(liveRow.count)); + expect(summary.min).toBe(round2(liveRow.min_value)); + expect(summary.max).toBe(round2(liveRow.max_value)); + expect(summary.mean).toBe(round2(liveRow.mean_value)); + // Narrow aggregate fills the non-composable windowed columns. + // Slope tuples and avg7/avg30 are sourced from live SQL even on + // the rollup-fresh path so the values are present (not null) as + // long as the windowed avg has rows to chew on. + expect(summary.avg30).not.toBeNull(); + // Latest reading comes from the DISTINCT ON pass — never null + // when the user has rows for the type. + expect(summary.latest).not.toBeNull(); + } + }); });