feat(rollups): v1.4.35 — partial read-swap onto measurement_rollups

Switches two reader surfaces onto the persistent rollup table for the
linearly-composable stats. The comprehensive insights aggregator and
the slim analytics summaries slice now source count / min / max / mean
from the DAY rollup buckets via aggregateBuckets, plus the comprehensive
dailyByType correlation feed straight from the bucket means. Slope /
R² / standard deviation / anomaly counts stay on live SQL — they do
not compose across DAY buckets.

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, and degrades
cleanly on the slim slice's all-time window for accounts that never
ran the explicit backfill.

Three new integration cases 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.

tests: 4249 → 4280 unit + 222 → 228 integration. tsc + lint clean.
This commit is contained in:
Marc-André Bombeck committed 2026-05-17 09:53:41 +02:00
1 parent aede0e9377
commit 67ccd55398
8 files changed
+627 -78

No files matched your search

+75
View File
@@ -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 <id>`) 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
+1 -1
View File
@@ -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",
@@ -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(() => {
@@ -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<typeof vi.fn>;
const ROLLUP_FIND_MANY =
prisma.measurementRollup.findMany as unknown as ReturnType<typeof vi.fn>;
const ROLLUP_FIND_FIRST =
prisma.measurementRollup.findFirst as unknown as ReturnType<typeof vi.fn>;
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");
+76 -16
View File
@@ -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<SummariesSlice> {
// 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<AggregateRow[]>`
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<string, number>();
@@ -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),
@@ -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<typeof vi.fn>;
const FIND_MANY =
prisma.measurement.findMany as unknown as ReturnType<typeof vi.fn>;
const ROLLUP_FIND_MANY =
prisma.measurementRollup.findMany as unknown as ReturnType<typeof vi.fn>;
const ROLLUP_FIND_FIRST =
prisma.measurementRollup.findFirst as unknown as ReturnType<typeof vi.fn>;
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 },
+124 -48
View File
@@ -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<ComprehensiveAggregate> {
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<AggregateRow[]>`
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<DailyAggregateRow[]>`
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<string, DataSummary> = {};
@@ -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<string, Array<{ day: string; value: number }>> = {};
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 {
@@ -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<string, Array<{ day: string; value: number }>>();
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);
});
});