tom/HealthLog
1
0 Fork 0
mirror of https://github.com/MBombeck/HealthLog.git synced 2026-11-08 12:32:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(insights): skip heavy live aggregate when rollups are populated

Browse Source 
v1.4.35 left the regression intact: comprehensive-aggregator and the
slim summaries slice still ran the full COUNT/MIN/MAX/AVG `$queryRaw`
in parallel with the rollup read, only preferring the bucket-derived
values when the count happened to match. Net round-trip count saved
was zero on Marc's 311k-row account.

Promote the rollup table to the canonical source on the happy path:
gate on a cheap `SELECT COUNT(*) FROM measurement_rollups` and, when
non-zero, drop the heavy aggregate. The bucket composition feeds
count/min/max/mean, a narrow `$queryRaw` keeps the non-composable
windowed columns (stddev/anomalies/avg7/avg30/avg30_last_month + the
three slope tuples), and the legacy heavy aggregate stays as the
cold-mount fallback for accounts whose backfill is mid-flight.

Unit tests pin the contract: the heavy aggregate is NOT called on the
rollup-fresh branch. Integration test verifies byte-identical count /
min / max / mean against a parallel live SQL pass on a populated
fixture.
This commit is contained in:
Marc-André Bombeck committed 2026-05-17 12:28:01 +02:00
1 parent 3f55b6f782
commit 13f8905ea1
5 files changed
+1196 -496
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/lib/analytics/__tests__/summaries-slice.test.ts
+240 -175
View File
@@ -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<typeof vi.fn>;
const MEASUREMENT_FIND_FIRST =
prisma.measurement.findFirst 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 =
@@ -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);
});
});
});
src/lib/analytics/summaries-slice.ts
+232 -50
View File
@@ -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<SummariesSlice> {
// 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<AggregateRow[]>`
// 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<Array<{ n: bigint }>>`
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<SummariesSlice> {
const [narrows, latests, dayBuckets] = await Promise.all([
prisma.$queryRaw<NarrowAggregateRow[]>`
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<string, NarrowAggregateRow>();
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<SummariesSlice> {
const [aggregates, latests] = await Promise.all([
prisma.$queryRaw<HeavyAggregateRow[]>`
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<LatestRow[]>`
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<string, number>();
const lastSeenAtByType = new Map<string, Date>();
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<string, DataSummary> = {};
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",
},
},
},
src/lib/insights/__tests__/comprehensive-aggregator.test.ts
+228 -147
View File
@@ -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<typeof vi.fn>;
const FIND_MANY =
prisma.measurement.findMany as unknown as ReturnType<typeof vi.fn>;
const MEASUREMENT_FIND_FIRST =
prisma.measurement.findFirst 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 =
@@ -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",
src/lib/insights/comprehensive-aggregator.ts
+392 -124
View File
@@ -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<ComprehensiveAggregate> {
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<AggregateRow[]>`
// 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<Array<{ n: bigint }>>`
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<ComprehensiveAggregate> {
const [narrows, latests, dayBuckets, sysRaw, diaRaw, firstRows] = await Promise.all([
prisma.$queryRaw<NarrowAggregateRow[]>`
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<Array<{ first_at: Date | null }>>`
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<string, { value: number; measuredAt: Date }>();
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<string, NarrowAggregateRow>();
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<string, DataSummary> = {};
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<ComprehensiveAggregate> {
const [aggregates, latests, dayBuckets, sysRaw, diaRaw] = await Promise.all([
prisma.$queryRaw<HeavyAggregateRow[]>`
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<LatestRow[]>`
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<string, { value: number; measuredAt: Date }>();
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<string, DataSummary> = {};
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<Array<{ first_at: Date | null }>>`
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<string, Array<{ day: string; value: number }>> {
const dailyByType: Record<string, Array<{ day: string; value: number }>> = {};
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;
}
tests/integration/measurement-rollups.test.ts
+104
View File
@@ -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();
}
});
});