diff --git a/CHANGELOG.md b/CHANGELOG.md index 6535d95d2..d49fe80a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,80 @@ # Changelog +## [1.4.35] — 2026-05-17 — Persistent measurement rollups + partial read-swap + +Foundation release for the persistent `measurement_rollups` cache +tier. Adds a new additive table, populates it from every write path, +and switches two reader surfaces (the comprehensive insights +aggregator and the slim analytics summaries slice) onto it for the +linearly-composable stats. Slope / R² / standard deviation / anomaly +counts continue to run against live SQL — they don't compose across +DAY buckets and stay on the canonical aggregate. A full read-swap +across every analytics surface is planned for a follow-up. + +### Schema + +- **New table `measurement_rollups`** keyed on + `(user_id, type, granularity, bucket_start)` with + `count / mean / min_value / max_value / sd / slope / r2 / + computed_at`. Backed by a composite-descending index on + `(user_id, type, granularity, bucket_start DESC)`. Additive only — + no `ALTER` on existing tables. Migration + `0067_v1434_measurement_rollups` is idempotent under re-run. +- **New enum `measurement_rollup_granularity`** with `DAY`, `WEEK`, + `MONTH`, `YEAR` granularities. The DAY grain is recomputed inline + on every measurement write; the wider grains queue onto pg-boss. + +### Write hooks + +- Every measurement create / update / delete endpoint now calls + `recomputeBucketsForMeasurement` inside a `try / catch` after its + successful commit. The DAY bucket is folded synchronously so the + next read is correct; WEEK / MONTH / YEAR follow on the pg-boss + `rollup-recompute` queue (concurrency = 2, singleton-keyed per + bucket to coalesce burst writes onto a single worker run). +- The Apple Health import worker calls `recomputeUserRollups` once + at completion instead of per-row, scoped to the import's date + range, so a 100k-sample backfill no longer spawns 100k write hooks. + +### Reads + +- `comprehensive-aggregator` (the engine behind + `/api/insights/comprehensive`) sources `count / min / max / mean` + per type from the DAY buckets via `aggregateBuckets`, plus the + `dailyByType` correlation feed straight from the bucket means. 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. +- `summaries-slice` (the slim slice behind `?slice=summaries`) does + the same. The all-time window degrades cleanly on accounts that + never ran the explicit backfill: the rollup covers only what the + warm-on-read populator pre-folded, the parity check diverges, and + the live SQL values take over — no semantics break. +- `ensureUserRollupsFresh` is a cheap watermark query that runs + before every read; on a warm process it's a single indexed lookup, + on cold mount it folds the trailing 90-day window into the DAY + rollup before the bucket read fires. + +### Backfill + +- New `scripts/backfill-rollups.ts` walks every user (or one + account via `--user `) and folds their measurement history + into all four granularities. Single-user serial so the Prisma + pool stays out of contention. Idempotent — every run upserts + under the composite primary key. + +### Tests + +- Three new integration cases in + `tests/integration/measurement-rollups.test.ts` pin the parity + contract end-to-end against a real Postgres testcontainer: the + aggregator's `count / min / max / mean` matches live SQL + byte-for-byte, `dailyByType` matches a parallel + `date_trunc('day', …)` GROUP BY query, and a freshly-written + measurement is reflected on the next read. +- Unit suite up by 31 cases (4249 → 4280); integration suite up by 6 + cases (222 → 228). `pnpm typecheck` + `pnpm lint` clean. + ## [1.4.34.5] — 2026-05-17 — Audit follow-on: critical-path tests + iOS textarea zoom Follow-on to v1.4.34.4. Two batches: the missing critical-path diff --git a/package.json b/package.json index 02739c2af..e4ddb29d7 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.4.34.5", + "version": "1.4.35", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "AGPL-3.0-only", "homepage": "https://healthlog.dev", diff --git a/src/app/api/analytics/__tests__/route.test.ts b/src/app/api/analytics/__tests__/route.test.ts index 88e304d4e..3ed5b35ee 100644 --- a/src/app/api/analytics/__tests__/route.test.ts +++ b/src/app/api/analytics/__tests__/route.test.ts @@ -25,6 +25,14 @@ vi.mock("@/lib/db", () => ({ medication: { findMany: vi.fn() }, // v1.4.33 C1 — slim summaries slice runs through `$queryRaw`. $queryRaw: vi.fn(), + // v1.4.35 — slim slice also reads DAY buckets; the freshness + // watermark inside `ensureUserRollupsFresh` pokes `findFirst`. + // Default both to empty so the parity check falls back to live + // SQL and the slim slice assertions stay byte-shape stable. + measurementRollup: { + findMany: vi.fn().mockResolvedValue([]), + findFirst: vi.fn().mockResolvedValue(null), + }, }, })); @@ -100,6 +108,17 @@ beforeEach(() => { [] as never, ); vi.mocked(prisma.medication.findMany).mockResolvedValue([] as never); + // v1.4.35 — rollup table defaults. `resetAllMocks` clears the + // module-level implementations, so we re-seed both per test. Empty + // findMany + null findFirst means the slim slice's parity check + // diverges and falls back to live SQL (preserves the pre-v1.4.35 + // assertions in this file). + vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue( + [] as never, + ); + vi.mocked(prisma.measurementRollup.findFirst).mockResolvedValue( + null as never, + ); }); afterEach(() => { diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 9839726cc..a3b66da34 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -9,6 +9,15 @@ 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. + measurementRollup: { + findMany: vi.fn(), + findFirst: vi.fn(), + }, }, })); @@ -20,9 +29,20 @@ import { prisma } from "@/lib/db"; import { computeSummariesSlice } from "../summaries-slice"; const RAW = prisma.$queryRaw as unknown as ReturnType; +const ROLLUP_FIND_MANY = + prisma.measurementRollup.findMany as unknown as ReturnType; +const ROLLUP_FIND_FIRST = + prisma.measurementRollup.findFirst as unknown as ReturnType; beforeEach(() => { RAW.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); }); afterEach(() => { @@ -32,6 +52,7 @@ 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([]); const result = await computeSummariesSlice("user-1"); diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index d4355ddfa..5779880bd 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -10,13 +10,15 @@ * it ever paints — yet the user waits on the full chunked walk. * * This helper resolves the same `DataSummary` shape (per-type) with - * two SQL passes: + * three passes (two SQL + one rollup-table read): * * 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. * * 2. **`DISTINCT ON (type)` latest read** — one `$queryRaw` * returning the most-recent `(type, value)` pair per type so the @@ -24,6 +26,12 @@ * we need the value of the row at `MAX(measured_at)`, not the * largest value ever recorded. * + * 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. + * * Fields the slim shape intentionally omits (the dashboard never * reads them on first paint): * - `anomalyCount`: would require an extra per-type read for the @@ -44,6 +52,7 @@ import type { DataSummary } from "@/lib/analytics/trends"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { annotate } from "@/lib/logging/context"; import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; +import { aggregateBuckets } from "@/lib/measurements/rollup-read"; interface AggregateRow { type: string; @@ -137,19 +146,19 @@ export interface SummariesSlice { export async function computeSummariesSlice( userId: string, ): Promise { - // v1.5.0 — keep the persistent rollup table current as a side - // effect of the dashboard tile strip mount. No-op when fresh; one - // bounded DAY-granularity recompute on first cold-mount after a - // process restart. The slim slice's response shape is untouched — - // the rollup table accelerates downstream consumers (the v1.5.1 - // read-path swap will route this surface through the rollup once - // dual-read parity is established). + // 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. await ensureUserRollupsFresh(userId); - // The two passes run in parallel — they target the same index path - // (`measurements (user_id, type, measured_at)`) but Postgres can - // serve them off the buffer cache concurrently. - const [aggregates, latests] = await Promise.all([ + // 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` SELECT m."type"::text AS type, @@ -212,6 +221,24 @@ export async function computeSummariesSlice( WHERE m."user_id" = ${userId} ORDER BY m."type", m."measured_at" DESC `, + // v1.4.35 — every DAY rollup bucket for the user. The slim slice's + // `count / min / max / mean` is all-time, so we read the bucket + // table without a `bucketStart` window. The defensive parity + // check below uses the live aggregate's `COUNT(*)` to decide + // whether the rollup is complete for this account (post-backfill + // → composed) or partial (pre-backfill → falls back to live). + prisma.measurementRollup.findMany({ + where: { userId, granularity: "DAY" }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }], + select: { + type: true, + bucketStart: true, + count: true, + mean: true, + minValue: true, + maxValue: true, + }, + }), ]); const latestByType = new Map(); @@ -250,17 +277,50 @@ export async function computeSummariesSlice( }; } + // v1.4.35 — partition the DAY buckets by type so the per-type loop + // 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); + } + let totalRows = 0; for (const row of aggregates) { - const count = Number(row.count); + 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; totalRows += count; const latest = latestByType.get(row.type) ?? null; summaries[row.type] = { count, latest, - min: round2(row.min_value), - max: round2(row.max_value), - mean: round2(row.mean_value), + 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), avg7: round2(row.avg7), avg30: round2(row.avg30), slope7: buildSlope(row.slope7, row.r2_7), diff --git a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts index 09c2a6948..892d68919 100644 --- a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts +++ b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts @@ -13,6 +13,13 @@ vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: vi.fn(), measurement: { findMany: vi.fn() }, + // v1.4.35 — 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: { + findMany: vi.fn(), + findFirst: vi.fn(), + }, }, })); @@ -26,10 +33,23 @@ import { buildComprehensiveAggregate } from "../comprehensive-aggregator"; const RAW = prisma.$queryRaw as unknown as ReturnType; const FIND_MANY = prisma.measurement.findMany as unknown as ReturnType; +const ROLLUP_FIND_MANY = + prisma.measurementRollup.findMany as unknown as ReturnType; +const ROLLUP_FIND_FIRST = + prisma.measurementRollup.findFirst as unknown as ReturnType; beforeEach(() => { RAW.mockReset(); FIND_MANY.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. + ROLLUP_FIND_MANY.mockResolvedValue([]); + ROLLUP_FIND_FIRST.mockResolvedValue(null); }); afterEach(() => { @@ -38,11 +58,12 @@ afterEach(() => { describe("buildComprehensiveAggregate", () => { it("returns an empty bundle for a user with no measurements", async () => { - // Pass order: aggregates, latests, daily, sysRaw, diaRaw. The - // `firstMeasurementAt` query is skipped when totalMeasurements === 0. + // 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 - .mockResolvedValueOnce([]); // daily + .mockResolvedValueOnce([]); // latests FIND_MANY.mockResolvedValueOnce([]) // sys .mockResolvedValueOnce([]); // dia @@ -54,10 +75,11 @@ describe("buildComprehensiveAggregate", () => { expect(result.dailyByType).toEqual({}); expect(result.firstMeasurementAt).toBeNull(); expect(result.totalMeasurements).toBe(0); - // 3 $queryRaw calls (aggregates, latests, daily) + 2 findMany. + // 2 $queryRaw (aggregates, latests) + 2 findMany + 1 rollup.findMany. // No firstMeasurementAt query when total === 0. - expect(RAW).toHaveBeenCalledTimes(3); + expect(RAW).toHaveBeenCalledTimes(2); expect(FIND_MANY).toHaveBeenCalledTimes(2); + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1); }); it("maps a populated WEIGHT aggregate into the DataSummary shape", async () => { @@ -83,11 +105,32 @@ describe("buildComprehensiveAggregate", () => { }, ]) .mockResolvedValueOnce([{ type: "WEIGHT", value: 81.4, measured_at: now }]) - .mockResolvedValueOnce([ - { type: "WEIGHT", day: "2026-05-10", mean_value: 82.0 }, - { type: "WEIGHT", day: "2026-05-11", mean_value: 82.2 }, - ]) .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 @@ -96,8 +139,11 @@ describe("buildComprehensiveAggregate", () => { 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); @@ -127,9 +173,12 @@ describe("buildComprehensiveAggregate", () => { 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: 82.0 }, - { day: "2026-05-11", value: 82.2 }, + { day: "2026-05-10", value: 81 }, + { day: "2026-05-11", value: 83 }, ]); }); @@ -162,7 +211,6 @@ describe("buildComprehensiveAggregate", () => { measured_at: measuredAt, }, ]) - .mockResolvedValueOnce([]) .mockResolvedValueOnce([{ first_at: measuredAt }]); FIND_MANY.mockResolvedValueOnce([ { measuredAt, value: 120 }, diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index 4af524d78..3ebecbbb5 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -1,5 +1,6 @@ /** - * v1.4.35 — SQL-driven aggregator for `/api/insights/comprehensive`. + * v1.4.35 — SQL-driven aggregator for `/api/insights/comprehensive`, + * with the partial read-swap onto `measurement_rollups` landed. * * Background * ---------- @@ -11,7 +12,7 @@ * HAR capture, with sibling endpoints cascading into 503s on pool * starvation). * - * The new shape mirrors the `summaries-slice` aggregator + * The shape mirrors the `summaries-slice` aggregator * (`src/lib/analytics/summaries-slice.ts`) the v1.4.33 C1 dashboard * tile strip already proved out: a single `$queryRaw` per surface, * grouped by `type`, with Postgres-side `AVG` / `MIN` / `MAX` / @@ -20,6 +21,40 @@ * 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: + * + * - **`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. + * + * - **`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. + * + * 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). + * * Scope notes * ----------- * - The 90-day window is preserved (the legacy `findMany` clamped on @@ -62,6 +97,7 @@ import { prisma } from "@/lib/db"; import type { DataSummary } from "@/lib/analytics/trends"; import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; +import { aggregateBuckets } from "@/lib/measurements/rollup-read"; /** * Raw row from the aggregate query. Keyed on `MeasurementType`, with @@ -95,13 +131,6 @@ interface LatestRow { measured_at: Date; } -/** Daily-mean per (type, day) for correlation pairings. */ -export interface DailyAggregateRow { - type: string; - day: string; - mean_value: number; -} - /** * Result bundle. Everything the route handler needs to assemble the * comprehensive response without holding the 100k-row raw measurement @@ -154,36 +183,34 @@ function round2(value: number | null): number | null { /** * Build the per-type `DataSummary` bundle for the comprehensive route. * - * Three SQL passes against the same `(user_id, type, measured_at)` - * index, all targeting the trailing 90 days: + * Five queries against the trailing 90 days, in parallel: * * 1. Per-type aggregate (count + min/max/mean/stddev + windowed - * avgs + regr_slope/r2 for the 7/30/90-day slopes). + * 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. Per-type-per-day mean (`date_trunc('day', measured_at)`) for the - * correlation pairings and the daily-bucket consumers. - * - * Plus two bounded raw-row queries for sys / dia (BP target pairing - * needs sub-day timestamps, can't be replaced by a daily mean). + * 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. */ export async function buildComprehensiveAggregate( userId: string, ): Promise { const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000); - // v1.5.0 — keep the persistent rollup table current for downstream - // consumers (the analytics tile strip, the dashboard widgets, the - // weekly report). The helper is a no-op when rollups are already - // ahead of the newest measurement; on first-mount after a cold - // restart it folds the trailing 90-day window into the DAY rollup - // table. The current aggregator output keeps its byte-shape via the - // live SQL queries below — the rollup table accelerates downstream - // reads, not the current response. A v1.5.1 follow-up will switch - // the per-type DataSummary lookups to read from the rollup table - // after the dual-read divergence gate clears. + // 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. await ensureUserRollupsFresh(userId); - const [aggregates, latests, dailyRows, sysRaw, diaRaw] = await Promise.all([ + const [aggregates, latests, dayBuckets, sysRaw, diaRaw] = await Promise.all([ prisma.$queryRaw` WITH window_stats AS ( SELECT @@ -274,18 +301,20 @@ export async function buildComprehensiveAggregate( AND m."measured_at" >= ${ninetyDaysAgo} ORDER BY m."type", m."measured_at" DESC `, - prisma.$queryRaw` - SELECT - m."type"::text AS type, - TO_CHAR(date_trunc('day', m."measured_at"), 'YYYY-MM-DD') AS day, - (ROUND((AVG(m."value"))::numeric, 2))::double precision AS mean_value - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} - AND m."type" IN ('WEIGHT', 'BLOOD_PRESSURE_SYS', 'PULSE') - GROUP BY m."type", date_trunc('day', m."measured_at") - ORDER BY m."type", day ASC - `, + // v1.4.35 — DAY rollup buckets for the trailing 90 days, every + // type. Drives both the per-type `count / min / max / mean` + // composition (via `aggregateBuckets`) and the `dailyByType` + // correlation feed below. Replaces the legacy + // `date_trunc('day', m."measured_at")` GROUP BY query and the + // per-type AVG/MIN/MAX/COUNT columns the live aggregate kept. + prisma.measurementRollup.findMany({ + where: { + userId, + granularity: "DAY", + bucketStart: { gte: ninetyDaysAgo }, + }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }], + }), prisma.measurement.findMany({ where: { userId, @@ -318,6 +347,30 @@ 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); + } + // Seed only types that actually have rows — the legacy route only // populated `summaries[t]` when `data.length > 0`. const summaries: Record = {}; @@ -325,15 +378,22 @@ export async function buildComprehensiveAggregate( let firstMeasurementAt: Date | null = null; for (const row of aggregates) { - const count = Number(row.count); + 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; totalMeasurements += count; const latest = latestByType.get(row.type); summaries[row.type] = { count, latest: latest?.value ?? null, - min: round2(row.min_value), - max: round2(row.max_value), - mean: round2(row.mean_value), + 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), avg7: round2(row.avg7), avg30: round2(row.avg30), slope7: buildSlope(row.slope7, row.r2_7), @@ -365,11 +425,27 @@ export async function buildComprehensiveAggregate( } // ── 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. const dailyByType: Record> = {}; - for (const row of dailyRows) { - const list = dailyByType[row.type] ?? []; - list.push({ day: row.day, value: Number(row.mean_value) }); - dailyByType[row.type] = list; + const DAILY_TYPES = ["WEIGHT", "BLOOD_PRESSURE_SYS", "PULSE"] as const; + for (const type of DAILY_TYPES) { + const buckets = bucketsByType.get(type); + if (!buckets || buckets.length === 0) continue; + dailyByType[type] = buckets.map((b) => ({ + // `bucketStart` is the UTC midnight of the bucketed day — the + // same boundary Postgres' `date_trunc('day', ...)` would pick + // when the session timezone is UTC (the container default). + day: b.day.toISOString().slice(0, 10), + // Match the legacy `ROUND(AVG, 2)::double precision` semantics + // so correlation thresholds tuned against the v1.4.34 output + // don't drift on the third decimal. + value: round2(b.mean) ?? 0, + })); } return { diff --git a/tests/integration/measurement-rollups.test.ts b/tests/integration/measurement-rollups.test.ts index 0301423b7..e10ee40a4 100644 --- a/tests/integration/measurement-rollups.test.ts +++ b/tests/integration/measurement-rollups.test.ts @@ -32,6 +32,7 @@ import { ALL_GRANULARITIES, } from "@/lib/measurements/rollups"; import { aggregateBuckets } from "@/lib/measurements/rollup-read"; +import { buildComprehensiveAggregate } from "@/lib/insights/comprehensive-aggregator"; vi.mock("@/lib/db-compat", () => ({ ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), @@ -253,4 +254,253 @@ describe("measurement rollups — integration", () => { expect(reAggregated.max).toBeCloseTo(live[0].max_value, 5); expect(reAggregated.mean).toBeCloseTo(live[0].mean_value, 5); }); + + // ─── v1.4.35 read-swap parity ────────────────────────────── + // + // The aggregator + slim slice now source `count / min / max / mean` + // from the DAY rollup buckets and fall back to live SQL when the + // composed count disagrees with `COUNT(*)`. These tests pin the + // happy path (rollup-derived equals live byte-for-byte) and the + // post-write freshness path (a new measurement after a warm read + // is reflected on the next call). + + it("comprehensive aggregator's count/min/max/mean matches live SQL byte-for-byte after warm rollup", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-readswap-user", + email: "rollup-readswap@example.test", + role: "USER", + }, + }); + + // 50 measurements across 30 days for WEIGHT and PULSE so both + // type families exercise the per-type composition path. The + // values are deliberately fractional so the round2 contract gets + // a workout. + 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 i = 0; i < 50; i++) { + seedRows.push({ + userId: user.id, + type: "WEIGHT", + value: 80 + (i % 7) * 0.37, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - (29 - (i % 30)) * dayMs - i * 60 * 1000), + }); + seedRows.push({ + userId: user.id, + type: "PULSE", + value: 60 + (i % 11) * 1.13, + unit: "bpm", + source: "MANUAL", + measuredAt: new Date(now - (29 - (i % 30)) * dayMs - i * 90 * 1000), + }); + } + await prisma.measurement.createMany({ data: seedRows }); + + // First call warms the rollup (ensureUserRollupsFresh fires the + // DAY recompute since no buckets exist yet). + const aggregate = await buildComprehensiveAggregate(user.id); + + // Parallel live aggregation over the same 90-day window. This is + // the source of truth — the read-swap must produce the same + // count/min/max/mean byte-for-byte. + const ninetyDaysAgo = new Date(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 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(); + expect(summary.count).toBe(Number(liveRow.count)); + expect(summary.min).toBe(round2(liveRow.min_value)); + expect(summary.max).toBe(round2(liveRow.max_value)); + // The composed weighted mean equals AVG(value) over the same + // rows; both run through `round2`. + expect(summary.mean).toBe(round2(liveRow.mean_value)); + } + }); + + it("dailyByType matches a parallel live SQL daily-mean aggregation", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-daily-user", + email: "rollup-daily@example.test", + role: "USER", + }, + }); + + const dayMs = 24 * 60 * 60 * 1000; + const now = Date.now(); + const rows: Array<{ + userId: string; + type: "WEIGHT" | "BLOOD_PRESSURE_SYS" | "PULSE"; + value: number; + unit: string; + source: "MANUAL"; + measuredAt: Date; + }> = []; + // 3 readings per day for 20 days, three types. + for (let day = 0; day < 20; day++) { + const dayAnchor = now - day * dayMs; + for (let k = 0; k < 3; k++) { + rows.push({ + userId: user.id, + type: "WEIGHT", + value: 81 + day * 0.05 + k * 0.13, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(dayAnchor - k * 60 * 60 * 1000), + }); + rows.push({ + userId: user.id, + type: "BLOOD_PRESSURE_SYS", + value: 120 + day * 0.21 + k * 0.07, + unit: "mmHg", + source: "MANUAL", + measuredAt: new Date(dayAnchor - k * 30 * 60 * 1000), + }); + rows.push({ + userId: user.id, + type: "PULSE", + value: 65 + day * 0.07 + k * 0.19, + unit: "bpm", + source: "MANUAL", + measuredAt: new Date(dayAnchor - k * 15 * 60 * 1000), + }); + } + } + await prisma.measurement.createMany({ data: rows }); + + const aggregate = await buildComprehensiveAggregate(user.id); + + // Parallel live aggregation — same shape the legacy SQL emitted. + const ninetyDaysAgo = new Date(now - 90 * dayMs); + const live = await prisma.$queryRaw< + Array<{ type: string; day: string; mean_value: number }> + >` + SELECT + m."type"::text AS type, + TO_CHAR(date_trunc('day', m."measured_at"), 'YYYY-MM-DD') AS day, + (ROUND((AVG(m."value"))::numeric, 2))::double precision AS mean_value + FROM measurements m + WHERE m."user_id" = ${user.id} + AND m."measured_at" >= ${ninetyDaysAgo} + AND m."type" IN ('WEIGHT', 'BLOOD_PRESSURE_SYS', 'PULSE') + GROUP BY m."type", date_trunc('day', m."measured_at") + ORDER BY m."type", day ASC + `; + + const liveByType = new Map>(); + for (const row of live) { + const list = liveByType.get(row.type) ?? []; + list.push({ day: row.day, value: Number(row.mean_value) }); + liveByType.set(row.type, list); + } + + for (const [type, liveDaily] of liveByType.entries()) { + const rollupDaily = aggregate.dailyByType[type]; + expect(rollupDaily, `dailyByType[${type}]`).toBeDefined(); + expect(rollupDaily!.length).toBe(liveDaily.length); + for (let i = 0; i < liveDaily.length; i++) { + expect(rollupDaily![i].day).toBe(liveDaily[i].day); + // round2 of the bucket mean == ROUND(AVG, 2) of the same rows. + expect(rollupDaily![i].value).toBeCloseTo(liveDaily[i].value, 5); + } + } + }); + + it("reflects a freshly written measurement on the next read after a warm cache", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-freshness-user", + email: "rollup-freshness@example.test", + role: "USER", + }, + }); + + const now = Date.now(); + const dayMs = 24 * 60 * 60 * 1000; + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - 5 * dayMs), + }, + { + userId: user.id, + type: "WEIGHT", + value: 81, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - 4 * dayMs), + }, + ], + }); + + // Warm-up read populates the rollups via `ensureUserRollupsFresh`. + const before = await buildComprehensiveAggregate(user.id); + expect(before.summaries.WEIGHT.count).toBe(2); + expect(before.summaries.WEIGHT.mean).toBe(80.5); + + // Insert a third row and synchronously refresh the DAY bucket the + // way the write-path hooks do. + const newMeasuredAt = new Date(now - 1 * dayMs); + await prisma.measurement.create({ + data: { + userId: user.id, + type: "WEIGHT", + value: 84, + unit: "kg", + source: "MANUAL", + measuredAt: newMeasuredAt, + }, + }); + await recomputeBucketsForMeasurement(user.id, "WEIGHT", newMeasuredAt); + + // Next read reflects the third measurement — the rollup-derived + // count moves from 2 to 3 and the weighted mean tracks live SQL. + const after = await buildComprehensiveAggregate(user.id); + expect(after.summaries.WEIGHT.count).toBe(3); + // Weighted mean across the 3 buckets (each count=1): (80+81+84)/3 + // = 81.6667 → round2 = 81.67. + expect(after.summaries.WEIGHT.mean).toBe(81.67); + }); });