fix(rollups): per-type coverage probe gates rollup vs live read

The v1.4.35/v1.4.36 read-swap gated on a global `COUNT(*) > 0`
against `measurement_rollups`. A user with BP fully rolled up who
logged their first WEIGHT measurement saw the per-write hook upsert
one WEIGHT DAY bucket, the global probe flipped to true, the read
path picked the bucket-derived branch for WEIGHT, and the all-time
WEIGHT count collapsed to whatever the trailing 90d narrow aggregate
plus the single fresh bucket could reconstruct.

The new shared probe in `rollup-coverage.ts` joins distinct
measurement types against the rollup table per user and surfaces
`{ type, has_buckets }` rows. Both `computeSummariesSlice` and
`buildComprehensiveAggregate` take the rollup-derived branch only
when every type the user has logged is covered; partial coverage
falls back to the live aggregate so the brand-new-type case stays
correct. Integration test seeds BP rolled up + WEIGHT raw-only and
asserts the WEIGHT summary count matches live SQL.
This commit is contained in:
Marc-André Bombeck committed 2026-05-17 13:00:17 +02:00
1 parent 7ee32a142e
commit 07491df6b2
6 files changed
+235 -42

No files matched your search

@@ -54,10 +54,10 @@ afterEach(() => {
describe("computeSummariesSlice", () => {
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.
// 1. per-type coverage probe — empty ⇒ cold path.
// 2. heavy aggregate ($queryRaw) — empty.
// 3. latests ($queryRaw) — empty.
RAW.mockResolvedValueOnce([{ n: BigInt(0) }])
RAW.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([]);
@@ -83,7 +83,8 @@ describe("computeSummariesSlice", () => {
});
it("maps a populated heavy aggregate row into the DataSummary shape on cold path", async () => {
RAW.mockResolvedValueOnce([{ n: BigInt(0) }])
// Coverage probe returns WEIGHT with no buckets → cold path.
RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }])
.mockResolvedValueOnce([
{
type: "WEIGHT",
@@ -136,7 +137,7 @@ describe("computeSummariesSlice", () => {
});
it("returns a null slope tuple when the SQL slope is null (insufficient rows)", async () => {
RAW.mockResolvedValueOnce([{ n: BigInt(0) }])
RAW.mockResolvedValueOnce([{ type: "PULSE", has_buckets: false }])
.mockResolvedValueOnce([
{
type: "PULSE",
@@ -166,7 +167,7 @@ describe("computeSummariesSlice", () => {
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) }])
RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }])
.mockResolvedValueOnce([
{
type: "WEIGHT",
@@ -198,7 +199,7 @@ describe("computeSummariesSlice", () => {
});
it("seeds the latest value from the DISTINCT ON pass per type", async () => {
RAW.mockResolvedValueOnce([{ n: BigInt(0) }])
RAW.mockResolvedValueOnce([{ type: "PULSE", has_buckets: false }])
.mockResolvedValueOnce([
{
type: "PULSE",
@@ -228,10 +229,10 @@ describe("computeSummariesSlice", () => {
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.
// 1. per-type coverage probe — WEIGHT fully covered ⇒ happy path.
// 2. narrow aggregate ($queryRaw) — windowed/regression only.
// 3. latests ($queryRaw).
RAW.mockResolvedValueOnce([{ n: BigInt(2) }])
RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }])
.mockResolvedValueOnce([
{
type: "WEIGHT",
+18 -12
View File
@@ -59,6 +59,10 @@ 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";
import {
isFullyCovered,
probeRollupCoverage,
} from "@/lib/measurements/rollup-coverage";
/**
* Heavy aggregate row — used on the cold-mount fallback path where the
@@ -181,18 +185,20 @@ export async function computeSummariesSlice(
// rollup so the bucket read below has something to compose.
await ensureUserRollupsFresh(userId);
// 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) {
// v1.4.36 QA C1 — per-type coverage probe replaces the legacy global
// COUNT. The previous gate returned true as soon as ANY type had at
// least one DAY bucket. Pathology: a user with BP fully rolled up
// logs their first WEIGHT measurement → the sync write-hook upserts
// one WEIGHT DAY bucket → the global probe stayed >0 → the route
// flipped to the bucket-derived path for WEIGHT as well → all-time
// count collapsed to whatever the trailing window covers because the
// narrow aggregate's windowed columns + the single fresh bucket
// can't reconstruct the prior history. We now decide per type and
// only take the rollup path when EVERY type the user has logged is
// covered. Partial coverage falls back to the live aggregate so the
// brand-new-type case stays correct.
const coverage = await probeRollupCoverage(userId);
if (isFullyCovered(coverage)) {
return computeFromRollups(userId);
}
return computeFromLiveAggregate(userId);
@@ -65,11 +65,11 @@ describe("buildComprehensiveAggregate", () => {
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).
// 1. per-type coverage probe — WEIGHT fully covered ⇒ happy path.
// 2. narrow aggregate ($queryRaw) — windowed/regression columns only.
// 3. latests ($queryRaw).
// 4. firstMeasurementAt ($queryRaw).
RAW.mockResolvedValueOnce([{ n: BigInt(2) }])
RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }])
.mockResolvedValueOnce([
{
type: "WEIGHT",
@@ -161,11 +161,11 @@ describe("buildComprehensiveAggregate", () => {
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.
// 1. per-type coverage probe — empty (no measurements) ⇒ cold path.
// 2. heavy aggregate ($queryRaw) — empty.
// 3. latests ($queryRaw) — empty.
// No firstMeasurementAt query when totalMeasurements === 0.
RAW.mockResolvedValueOnce([{ n: BigInt(0) }])
RAW.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([]);
FIND_MANY.mockResolvedValueOnce([]) // sys
@@ -188,11 +188,11 @@ describe("buildComprehensiveAggregate", () => {
it("runs the heavy aggregate when no rollup rows exist yet", async () => {
const now = new Date();
// 1. rollup-count probe — zero rows ⇒ cold path.
// 1. per-type coverage probe — WEIGHT measured but no buckets ⇒ cold path.
// 2. heavy aggregate ($queryRaw) — populated.
// 3. latests ($queryRaw) — populated.
// 4. firstMeasurementAt ($queryRaw).
RAW.mockResolvedValueOnce([{ n: BigInt(0) }])
RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }])
.mockResolvedValueOnce([
{
type: "WEIGHT",
@@ -263,8 +263,8 @@ describe("buildComprehensiveAggregate", () => {
it("threads sys/dia raw rows through bpRawRows so 5-min pairing survives", async () => {
const measuredAt = new Date("2026-05-10T08:00:00Z");
// Cold path so the heavy aggregate fires (no rollup count).
RAW.mockResolvedValueOnce([{ n: BigInt(0) }])
// Cold path so the heavy aggregate fires (BP type lacks coverage).
RAW.mockResolvedValueOnce([{ type: "BLOOD_PRESSURE_SYS", has_buckets: false }])
.mockResolvedValueOnce([
{
type: "BLOOD_PRESSURE_SYS",
+16 -14
View File
@@ -101,6 +101,10 @@ 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";
import {
isFullyCovered,
probeRollupCoverage,
} from "@/lib/measurements/rollup-coverage";
/**
* Heavy aggregate row — count/min/max/mean alongside the non-composable
@@ -236,20 +240,18 @@ export async function buildComprehensiveAggregate(
// `ensureUserRollupsFresh` covers cold-mount / process-restart).
await ensureUserRollupsFresh(userId);
// 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) {
// v1.4.36 QA C1 — per-type coverage probe. The legacy global COUNT
// returned true as soon as any one type had a DAY bucket, which made
// the bucket-derived path swallow a brand-new type's all-time history
// (the per-write hook upserts a single bucket for the new type, the
// probe flips to true, the comprehensive response then iterates only
// bucket-bearing types and the trailing-90d narrow aggregate becomes
// the new "all-time" total for that type). We now decide per type
// and only take the rollup path when EVERY type the user has logged
// is covered. Partial coverage falls back to the live aggregate so
// a fresh-type measurement doesn't break its summary card.
const coverage = await probeRollupCoverage(userId);
if (isFullyCovered(coverage)) {
return buildFromRollups(userId, ninetyDaysAgo);
}
return buildFromLiveAggregate(userId, ninetyDaysAgo);
+96
View File
@@ -0,0 +1,96 @@
/**
* v1.4.36 — per-type rollup coverage probe.
*
* The v1.4.35/v1.4.36 read-swap gates on whether the DAY rollup table
* carries buckets for a user. The first iteration checked a single
* `COUNT(*) > 0` against `measurement_rollups`. That returns `true` as
* soon as *any* type has buckets, which breaks the "first measurement
* for a brand-new type" case: the per-write hook upserts one DAY bucket
* for the new type → the global probe stays >0 → the read path picks
* the bucket-derived branch for *every* type → the brand-new type's
* narrow-aggregate windowed columns come back fine but the seed loop
* only iterates bucket-bearing types, so a type that has live
* measurements but no buckets falls out of the response entirely.
*
* The probe below joins `measurements`-distinct-types against
* `measurement_rollups` so the caller learns, per type, whether the
* bucket table covers it. Types with `hasBuckets=false` fall back to
* the live aggregator branch; types with `hasBuckets=true` ride the
* cheap composed branch. The result is a single round-trip indexed
* read regardless of how many types the user has logged.
*/
import { prisma } from "@/lib/db";
/**
* Per-type coverage map. `true` means at least one DAY rollup row
* exists for this `(user, type)` pair; the caller can safely compose
* `count / min / max / mean` from the bucket table for this type.
* `false` means the bucket table is empty for this type and the caller
* must fall back to the live aggregate.
*
* Types the user has never logged are absent from the map — callers
* never need to ask about a type with zero measurements.
*/
export type RollupCoverageMap = Map<string, boolean>;
/**
* Probe DAY-bucket coverage for every type the user has measurements
* for. The join is anchored on the smaller `DISTINCT type FROM
* measurements` set so the planner picks the per-type
* `(user_id, type, measured_at)` index path; the LEFT JOIN onto the
* rollup table uses the `(user_id, type, granularity, bucket_start)`
* composite primary key. A `COUNT > 0` per partition keeps the result
* shape stable even when a type has zero buckets.
*/
export async function probeRollupCoverage(
userId: string,
): Promise<RollupCoverageMap> {
const rows = await prisma.$queryRaw<
Array<{ type: string; has_buckets: boolean }>
>`
SELECT
m."type"::text AS type,
COUNT(r.*) > 0 AS has_buckets
FROM (
SELECT DISTINCT "type"
FROM measurements
WHERE user_id = ${userId}
) m
LEFT JOIN measurement_rollups r
ON r.user_id = ${userId}
AND r."type" = m."type"
AND r.granularity = 'DAY'
GROUP BY m."type"
`;
const coverage: RollupCoverageMap = new Map();
for (const row of rows) {
coverage.set(row.type, Boolean(row.has_buckets));
}
return coverage;
}
/**
* Returns the set of types the user has logged that do NOT yet have
* DAY-rollup coverage. The bucket-fresh read paths use this to decide
* which types still need the live aggregator branch.
*/
export function typesMissingCoverage(coverage: RollupCoverageMap): string[] {
const missing: string[] = [];
for (const [type, hasBuckets] of coverage.entries()) {
if (!hasBuckets) missing.push(type);
}
return missing;
}
/**
* Convenience — `true` when the user has at least one measurement and
* every type with measurements also has DAY-bucket coverage. The read
* path can skip the live aggregate entirely.
*/
export function isFullyCovered(coverage: RollupCoverageMap): boolean {
if (coverage.size === 0) return false;
for (const hasBuckets of coverage.values()) {
if (!hasBuckets) return false;
}
return true;
}
@@ -734,4 +734,92 @@ describe("measurement rollups — integration", () => {
expect(summary.latest).not.toBeNull();
}
});
it("falls back to the live aggregate for a type without rollup coverage", async () => {
// v1.4.36 QA C1 — pre-fix the global rollup probe flipped to true
// as soon as ANY type had a DAY bucket, which made the comprehensive
// response collapse a brand-new type's all-time count to whatever
// the bucket-derived path could see (often 1 — the freshly-written
// bucket). This guards the per-type probe so adding a first WEIGHT
// measurement to a BP-rolled-up account doesn't truncate the WEIGHT
// summary.
const prisma = getPrismaClient();
const user = await prisma.user.create({
data: {
username: "rollup-coverage-mix",
email: "rollup-coverage-mix@example.test",
role: "USER",
},
});
const now = Date.now();
const dayMs = 24 * 60 * 60 * 1000;
// Seed BP — multiple days within the 90-day window so the rollup
// path has something to compose against.
const bpRows = [];
for (let day = 0; day < 10; day += 1) {
bpRows.push({
userId: user.id,
type: "BLOOD_PRESSURE_SYS" as const,
value: 120 + day,
unit: "mmHg",
source: "MANUAL" as const,
measuredAt: new Date(now - day * dayMs),
});
}
await prisma.measurement.createMany({ data: bpRows });
// Pre-fold BP DAY buckets only. WEIGHT will be added below WITHOUT
// a populator pass so its coverage probe returns false.
await recomputeUserRollups(user.id, {
granularities: ["DAY"],
types: ["BLOOD_PRESSURE_SYS"],
});
// Insert WEIGHT measurements directly via Prisma — no rollup
// populator runs, no DAY bucket exists for WEIGHT.
const weightRows = [];
for (let day = 0; day < 5; day += 1) {
weightRows.push({
userId: user.id,
type: "WEIGHT" as const,
value: 80 + day,
unit: "kg",
source: "MANUAL" as const,
measuredAt: new Date(now - day * dayMs - 1000),
});
}
await prisma.measurement.createMany({ data: weightRows });
// Confirm setup: BP has rollups, WEIGHT does not.
const bpRollups = await prisma.measurementRollup.count({
where: { userId: user.id, type: "BLOOD_PRESSURE_SYS", granularity: "DAY" },
});
const weightRollups = await prisma.measurementRollup.count({
where: { userId: user.id, type: "WEIGHT", granularity: "DAY" },
});
expect(bpRollups).toBeGreaterThan(0);
expect(weightRollups).toBe(0);
const aggregate = await buildComprehensiveAggregate(user.id);
// Parallel live SQL — the contract is that the aggregator surfaces
// a WEIGHT summary whose count matches the actual measurement
// count, not zero (pre-fix) and not a bucket-derived partial.
const liveCounts = await prisma.$queryRaw<
Array<{ type: string; count: bigint }>
>`
SELECT m."type"::text AS type, COUNT(*) AS count
FROM measurements m
WHERE m."user_id" = ${user.id}
AND m."type" = 'WEIGHT'
GROUP BY m."type"
`;
expect(liveCounts).toHaveLength(1);
const expectedWeightCount = Number(liveCounts[0].count);
expect(aggregate.summaries.WEIGHT).toBeDefined();
expect(aggregate.summaries.WEIGHT.count).toBe(expectedWeightCount);
expect(aggregate.summaries.BLOOD_PRESSURE_SYS).toBeDefined();
});
});