diff --git a/src/app/api/measurements/__tests__/range-aggregation-route.test.ts b/src/app/api/measurements/__tests__/range-aggregation-route.test.ts index c2bb37dc8..8ead7e75e 100644 --- a/src/app/api/measurements/__tests__/range-aggregation-route.test.ts +++ b/src/app/api/measurements/__tests__/range-aggregation-route.test.ts @@ -27,6 +27,22 @@ vi.mock("@/lib/db", () => ({ }, })); +// v1.4.39.2 — the rollup branch's coverage-mismatch fallback may call +// `recomputeUserRollups` inline. Mock it so tests can assert whether +// the inline fold fired without exercising the real populator. +vi.mock("@/lib/measurements/rollups", async () => { + const actual = await vi.importActual< + typeof import("@/lib/measurements/rollups") + >("@/lib/measurements/rollups"); + return { + ...actual, + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeUserRollups: vi + .fn() + .mockResolvedValue({ rowsUpserted: 0, durationMs: 0 }), + }; +}); + vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); vi.mock("@/lib/auth/audit", () => ({ auditLog: vi.fn() })); vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); @@ -45,6 +61,7 @@ vi.mock("next/headers", () => ({ import { GET } from "../route"; import { prisma } from "@/lib/db"; import { getSession } from "@/lib/auth/session"; +import { recomputeUserRollups } from "@/lib/measurements/rollups"; const SESSION_OK = { session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) }, @@ -320,6 +337,186 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => { expect(json.data.measurements.length).toBe(7); }); + it("source=rollup folds the partition inline and re-reads when the rollup is under-converged (v1.4.39.2)", async () => { + // Marc's pre-fix production trace: 30-day BP_SYS window returned + // `total: 2` from the rollup branch even though `measurements` + // carried ~30 distinct days. The chart tripped its `< 3 daily + // points` empty-state because the route short-circuited on + // `rollupRows.length > 0` and never reconciled with the live + // table. The coverage-mismatch fallback runs an inline + // `recomputeUserRollups` for the (user, type, DAY, window) + // partition and re-reads so the request returns the converged + // rows. + const sparseRollup = [ + { + type: "BLOOD_PRESSURE_SYS", + bucketStart: new Date(Date.UTC(2026, 4, 1, 0, 0, 0)), + mean: 128, + count: 1, + sumValue: null, + }, + { + type: "BLOOD_PRESSURE_SYS", + bucketStart: new Date(Date.UTC(2026, 4, 10, 0, 0, 0)), + mean: 132, + count: 1, + sumValue: null, + }, + ]; + const convergedRollup = Array.from({ length: 28 }, (_, i) => ({ + type: "BLOOD_PRESSURE_SYS", + bucketStart: new Date(Date.UTC(2026, 4, 1 + i, 0, 0, 0)), + mean: 128 + (i % 5), + count: 2, + sumValue: null, + })); + // First read returns the sparse rows; second (post-fold) read + // returns the converged rows. + vi.mocked(prisma.measurementRollup.findMany) + .mockResolvedValueOnce(sparseRollup as never) + .mockResolvedValueOnce(convergedRollup as never); + // Coverage probe — live `measurements` has 28 distinct days. + vi.mocked(prisma.$queryRaw).mockResolvedValueOnce([ + { days: BigInt(28) }, + ] as never); + + const res = await GET( + getRequest( + "type=BLOOD_PRESSURE_SYS&from=2026-04-30T00:00:00Z&to=2026-05-30T00:00:00Z&aggregate=daily&source=rollup&limit=5000", + ), + ); + expect(res.status).toBe(200); + const json = (await res.json()) as { + data: { measurements: Array }; + }; + // The inline fold fired exactly once with the (type, DAY, window) + // bound so the partition write stays small. + expect(recomputeUserRollups).toHaveBeenCalledTimes(1); + expect(recomputeUserRollups).toHaveBeenCalledWith( + "user-1", + expect.objectContaining({ + types: ["BLOOD_PRESSURE_SYS"], + granularities: ["DAY"], + }), + ); + // The route re-read the rollup after the fold so the response + // carries the converged 28 buckets — well above the chart's + // `< 3 daily points` empty-state threshold. + expect(prisma.measurementRollup.findMany).toHaveBeenCalledTimes(2); + expect(json.data.measurements.length).toBe(28); + }); + + it("source=rollup skips the coverage probe on a covered window (v1.4.39.2 hot-path)", async () => { + // Happy case: 30-day window, 28 rollup rows present. The route + // must short-circuit on the first read and skip the probe so the + // covered-tenant hot path stays a single indexed read. + const buckets = Array.from({ length: 28 }, (_, i) => ({ + type: "WEIGHT", + bucketStart: new Date(Date.UTC(2026, 3, 16 + i, 0, 0, 0)), + mean: 81 + i * 0.05, + count: 2, + sumValue: null, + })); + vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue( + buckets as never, + ); + + const res = await GET( + getRequest( + "type=WEIGHT&from=2026-04-15T00:00:00Z&to=2026-05-15T00:00:00Z&aggregate=daily&source=rollup&limit=5000", + ), + ); + expect(res.status).toBe(200); + expect(prisma.measurementRollup.findMany).toHaveBeenCalledTimes(1); + expect(recomputeUserRollups).not.toHaveBeenCalled(); + expect(prisma.$queryRaw).not.toHaveBeenCalled(); + }); + + it("source=rollup skips the coverage fallback on short windows (v1.4.39.2)", async () => { + // A 5-day window with 2 rollup rows is plausibly genuine sparsity + // for a user who measures BP every other day. The coverage + // fallback gates on `windowDays >= 7` so the probe + inline fold + // never fire on these short windows. + const sparseRollup = [ + { + type: "BLOOD_PRESSURE_SYS", + bucketStart: new Date(Date.UTC(2026, 4, 1, 0, 0, 0)), + mean: 128, + count: 1, + sumValue: null, + }, + { + type: "BLOOD_PRESSURE_SYS", + bucketStart: new Date(Date.UTC(2026, 4, 3, 0, 0, 0)), + mean: 132, + count: 1, + sumValue: null, + }, + ]; + vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue( + sparseRollup as never, + ); + + const res = await GET( + getRequest( + "type=BLOOD_PRESSURE_SYS&from=2026-05-01T00:00:00Z&to=2026-05-06T00:00:00Z&aggregate=daily&source=rollup&limit=5000", + ), + ); + expect(res.status).toBe(200); + expect(recomputeUserRollups).not.toHaveBeenCalled(); + expect(prisma.$queryRaw).not.toHaveBeenCalled(); + const json = (await res.json()) as { + data: { measurements: Array }; + }; + expect(json.data.measurements.length).toBe(2); + }); + + it("source=rollup keeps the sparse rollup when live `measurements` agrees (v1.4.39.2)", async () => { + // Genuine sparsity: the user measured BP twice in 30 days. The + // probe finds the same 2 distinct days in `measurements` so the + // route returns the 2 rollup rows without firing the inline + // fold. The chart's `< 3 daily points` empty-state is the right + // surface for this case — the data really is sparse. + const sparseRollup = [ + { + type: "BLOOD_PRESSURE_SYS", + bucketStart: new Date(Date.UTC(2026, 4, 1, 0, 0, 0)), + mean: 128, + count: 1, + sumValue: null, + }, + { + type: "BLOOD_PRESSURE_SYS", + bucketStart: new Date(Date.UTC(2026, 4, 10, 0, 0, 0)), + mean: 132, + count: 1, + sumValue: null, + }, + ]; + vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue( + sparseRollup as never, + ); + // Probe agrees — 2 distinct days in `measurements` too. + vi.mocked(prisma.$queryRaw).mockResolvedValueOnce([ + { days: BigInt(2) }, + ] as never); + + const res = await GET( + getRequest( + "type=BLOOD_PRESSURE_SYS&from=2026-04-30T00:00:00Z&to=2026-05-30T00:00:00Z&aggregate=daily&source=rollup&limit=5000", + ), + ); + expect(res.status).toBe(200); + // Probe fired (suspicious-low rollup) but the inline fold did + // not (live and rollup agree). + expect(prisma.$queryRaw).toHaveBeenCalledTimes(1); + expect(recomputeUserRollups).not.toHaveBeenCalled(); + const json = (await res.json()) as { + data: { measurements: Array }; + }; + expect(json.data.measurements.length).toBe(2); + }); + it("returns a weekly series when the all-time window is under two years", async () => { // Three months past one year → falls in the weekly grain band. const weeks = Array.from({ length: 60 }, (_, i) => ({ diff --git a/src/app/api/measurements/route.ts b/src/app/api/measurements/route.ts index 36ee85b15..2229d21ba 100644 --- a/src/app/api/measurements/route.ts +++ b/src/app/api/measurements/route.ts @@ -29,6 +29,7 @@ import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { recomputeBucketsForMeasurement, collapseToTypeDayKeys, + recomputeUserRollups, } from "@/lib/measurements/rollups"; import { NextRequest } from "next/server"; import type { @@ -293,7 +294,113 @@ export const GET = apiHandler(async (request: NextRequest) => { sumValue: true, }, }); - if (rollupRows.length > 0) { + // v1.4.39.2 — coverage-mismatch fallback. v1.4.39.1 wired the + // rollup write hook into the previously-bypassed Withings sync, + // /api/import, and admin-restore paths, and widened the boot-time + // backfill discovery to per-(user, type, day). Together those + // changes ensure new writes and post-deploy restarts converge the + // rollup tier for every measurement type the user has logged. They + // do NOT, however, guarantee convergence inside a single request + // when the boot backfill has not yet finished folding the user's + // historic rows: a chart query lands first, the rollup table still + // carries 2 rows for a 30-day BP_SYS window, and the route's pre- + // v1.4.39.2 "rollup wins on length > 0" short-circuit returns the + // sparse rollup. The chart trips its `< 3 daily points` empty-state + // even though `measurements` holds the full 30 days. + // + // This branch closes that gap. When the rollup row count is + // suspiciously low for the requested window (≥ 7 days asked and + // < 3 rollup rows present), we probe `measurements` for the actual + // distinct-day count over the same window. If `measurements` + // carries strictly more days than the rollup, the rollup is + // under-converged: we fold the (user, type, DAY, [from, to]) + // partition inline, re-read the rollup, and return the refreshed + // rows. The inline fold is bounded by the window + type so it is + // small (~30 buckets × one type for the dashboard chart's 30-day + // window) and finishes well inside the request budget. + // + // The probe runs only on suspicious rollups so the covered-tenant + // happy path stays a single indexed read. Marc's tenant on a 30-day + // BP_SYS window pre-fix returned 2 rollup rows and hit `total: 2` + // in the live trace; post-fix the probe finds ~30 distinct days in + // `measurements`, folds them, and the chart paints all 30. + const windowMs = to.getTime() - from.getTime(); + const windowDays = Math.ceil(windowMs / 86_400_000); + const SUSPICIOUS_ROW_FLOOR = 3; + const SUSPICIOUS_WINDOW_DAYS = 7; + let effectiveRows = rollupRows; + let coverageFallbackFired = false; + if ( + rollupRows.length > 0 && + rollupRows.length < SUSPICIOUS_ROW_FLOOR && + windowDays >= SUSPICIOUS_WINDOW_DAYS + ) { + // Cheap probe — one indexed range scan over + // `(user_id, type, measured_at)`. Returns the count of distinct + // calendar days the live table holds for the window. + const liveDayCountRows = await prisma.$queryRaw< + Array<{ days: bigint }> + >` + SELECT COUNT(DISTINCT date_trunc('day', m."measured_at"))::bigint AS days + FROM measurements m + WHERE m."user_id" = ${user.id} + AND m."type" = ${type}::measurement_type + AND m."measured_at" >= ${from} + AND m."measured_at" <= ${to} + `; + const liveDays = Number(liveDayCountRows[0]?.days ?? BigInt(0)); + if (liveDays > rollupRows.length) { + coverageFallbackFired = true; + try { + // Inline fold of the (user, type, DAY, window) partition. + // Bounded by the window so it stays small even on a + // power-user account; the upsert is idempotent against the + // composite primary key. + await recomputeUserRollups(user.id, { + types: [type as MeasurementType], + granularities: ["DAY"], + from, + to, + }); + // Re-read so this request returns the converged rows. The + // next chart paint hits the same warm rollup without paying + // the fallback cost. + effectiveRows = await prisma.measurementRollup.findMany({ + where: { + userId: user.id, + type: type as MeasurementType, + granularity: "DAY", + bucketStart: { gte: from, lte: to }, + }, + orderBy: { bucketStart: "asc" }, + take: cap, + select: { + type: true, + bucketStart: true, + mean: true, + count: true, + sumValue: true, + }, + }); + } catch (err) { + // Best-effort — fall back to the legacy live aggregate path + // below if the fold itself failed so the chart paints + // something sensible regardless. `annotate` carries the + // surface so ops can spot a silent populator regression. + const message = err instanceof Error ? err.message : String(err); + annotate({ + meta: { + rollup_coverage_fallback_failed: true, + rollup_coverage_fallback_error: message, + type, + }, + }); + // Drop into the live `date_trunc` branch below. + effectiveRows = []; + } + } + } + if (effectiveRows.length > 0) { // Cumulative metrics (steps, active energy, distance, flights, // daylight) store `mean` per bucket but the chart needs the // daily SUM. v1.4.39 W-SUM — read `sumValue` directly; fall back @@ -303,7 +410,7 @@ export const GET = apiHandler(async (request: NextRequest) => { // SUM(value) for a single-source day because AVG = SUM / COUNT. const useSum = type != null && CUMULATIVE_HK_TYPES.has(type as MeasurementType); - const measurements = rollupRows.map((r) => ({ + const measurements = effectiveRows.map((r) => ({ type: r.type, value: useSum ? (r.sumValue ?? r.mean * r.count) @@ -318,6 +425,7 @@ export const GET = apiHandler(async (request: NextRequest) => { type, aggregate: "daily", source: "rollup", + coverage_fallback: coverageFallbackFired, }, }); return apiSuccess({ @@ -332,7 +440,8 @@ export const GET = apiHandler(async (request: NextRequest) => { } // Fall through to the live aggregate path when the rollup is // empty for the requested window — covers brand-new accounts + - // the race between a recent write hook and the read. + // the race between a recent write hook and the read, and the + // coverage-fallback inline-fold-failed branch above. } if (aggregate && aggregate !== "raw" && from && to) {