fix(analytics): collapse overlapping sources on the cold BP path

The raw BP pulls in the comprehensive aggregator and the bp-in-target live
fallback read every source's row for BLOOD_PRESSURE_SYS/DIA. BP is a
multi-source metric — a Withings reading mirrored into Apple Health is two
rows for one event — so each shared reading was counted twice in the
in-target denominator and twice in the recency-weighted grade. The rollup
branch of the same fast-path reads source-collapsed measurement_rollups, so a
rollup-warm account and a rollup-cold account returned different BP
percentages and grades for identical data — a web/iOS and cold/warm parity
break.

Route both raw pulls through canonicalMeasurementsFrom so the cold/live path
picks the ladder-winning source per (type, day), matching the warm/rollup
path. Per-event rows of the winning source survive the collapse, so the
downstream 5-minute pairing still sees individual readings.
This commit is contained in:
Marc-André Bombeck committed 2026-06-20 16:52:00 +02:00
1 parent fb4e3921e2
commit 013241ba74
4 files changed
+215 -92

No files matched your search

@@ -15,6 +15,10 @@ vi.mock("@/lib/db", () => ({
// v1.11.1 — readRollupBuckets lazy-loads the source-priority blob; null
// (default findUnique) falls back to the default ladders.
user: { findUnique: vi.fn() },
// v1.18.10 I-1 — the live BP fallback now reads source-collapsed rows via
// `$queryRawUnsafe` (`canonicalMeasurementsFrom`) instead of a chunked
// `findMany`, so cold == warm. Mock it to feed the SYS / DIA series.
$queryRawUnsafe: vi.fn(),
},
}));
@@ -39,6 +43,7 @@ const MEASUREMENT_FIND_MANY = prisma.measurement
.findMany as unknown as ReturnType<typeof vi.fn>;
const ROLLUP_FIND_MANY = prisma.measurementRollup
.findMany as unknown as ReturnType<typeof vi.fn>;
const QUERY_RAW = prisma.$queryRawUnsafe as unknown as ReturnType<typeof vi.fn>;
const PROBE = probeRollupCoverage as unknown as ReturnType<typeof vi.fn>;
const FULLY_COVERED = isFullyCovered as unknown as ReturnType<typeof vi.fn>;
const ANNOTATE = annotate as unknown as ReturnType<typeof vi.fn>;
@@ -53,6 +58,7 @@ const TARGETS_UNDER_65 = {
beforeEach(() => {
MEASUREMENT_FIND_MANY.mockReset();
ROLLUP_FIND_MANY.mockReset();
QUERY_RAW.mockReset();
PROBE.mockReset();
FULLY_COVERED.mockReset();
ANNOTATE.mockReset();
@@ -215,7 +221,7 @@ describe("computeBpInTargetFastPath", () => {
});
describe("live fallback — partial or no coverage", () => {
it("falls back to the chunked findMany when SYS bucket coverage is missing", async () => {
it("falls back to the source-collapsed live read when SYS bucket coverage is missing", async () => {
const coverage = new Map<string, boolean>([
["BLOOD_PRESSURE_SYS", false],
["BLOOD_PRESSURE_DIA", true],
@@ -223,17 +229,16 @@ describe("computeBpInTargetFastPath", () => {
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
// First findMany = SYS rows, second = DIA rows.
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([
// First raw query = SYS rows, second = DIA rows. Rows arrive in the raw
// SQL snake_case shape (`measured_at` + `value`).
QUERY_RAW.mockResolvedValueOnce([
{
id: "1",
measuredAt: new Date("2026-05-15T08:00:00.000Z"),
measured_at: new Date("2026-05-15T08:00:00.000Z"),
value: 122,
},
]).mockResolvedValueOnce([
{
id: "2",
measuredAt: new Date("2026-05-15T08:00:00.000Z"),
measured_at: new Date("2026-05-15T08:00:00.000Z"),
value: 76,
},
]);
@@ -249,7 +254,8 @@ describe("computeBpInTargetFastPath", () => {
// One paired in-target reading.
expect(result.last30Days).toEqual({ pct: 100, pairs: 1 });
expect(ROLLUP_FIND_MANY).not.toHaveBeenCalled();
expect(MEASUREMENT_FIND_MANY).toHaveBeenCalledTimes(2);
expect(MEASUREMENT_FIND_MANY).not.toHaveBeenCalled();
expect(QUERY_RAW).toHaveBeenCalledTimes(2);
});
it("pins the path:live annotate", async () => {
@@ -259,7 +265,7 @@ describe("computeBpInTargetFastPath", () => {
]);
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
await computeBpInTargetFastPath({
userId: "user-partial",
@@ -275,11 +281,51 @@ describe("computeBpInTargetFastPath", () => {
expect(bpCall).toBeDefined();
});
it("collapses overlapping sources on the live path (v1.18.10 I-1 — cold == warm)", async () => {
// Two sources mirror the SAME reading on the same day (Withings + its
// Apple-Health mirror). The source-collapse picks one canonical source
// per (type, day) BEFORE pairing, so the in-target denominator counts the
// shared reading once — matching the rollup path. The collapse runs in
// SQL (`canonicalMeasurementsFrom`), so the mock just confirms the live
// query routes through it and the result reflects ONE pair, not two.
const coverage = new Map<string, boolean>([
["BLOOD_PRESSURE_SYS", false],
["BLOOD_PRESSURE_DIA", false],
]);
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
// The canonical subquery has already collapsed the duplicate source — it
// returns ONE SYS row and ONE DIA row for the day.
QUERY_RAW.mockResolvedValueOnce([
{ measured_at: new Date("2026-05-15T08:00:00.000Z"), value: 122 },
]).mockResolvedValueOnce([
{ measured_at: new Date("2026-05-15T08:00:00.000Z"), value: 76 },
]);
const result = await computeBpInTargetFastPath({
userId: "user-dual-source",
targets: TARGETS_UNDER_65,
now: new Date("2026-05-17T12:00:00.000Z"),
coverage,
});
expect(result.path).toBe("live");
expect(result.last30Days).toEqual({ pct: 100, pairs: 1 });
// Both live reads must route through the canonical-source subquery so the
// collapse fires in SQL — pin the marker so a regression that reverts to
// a raw `findMany` (double-counting) fails here.
for (const call of QUERY_RAW.mock.calls) {
const sql = String(call[0]);
expect(sql).toContain("DISTINCT ON");
expect(sql).toContain("date_trunc('day'");
}
});
it("probes coverage when the caller omits the map", async () => {
const coverage = new Map<string, boolean>([]);
FULLY_COVERED.mockReturnValue(false);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
const result = await computeBpInTargetFastPath({
userId: "user-fresh",
@@ -331,7 +377,7 @@ describe("computeBpInTargetFastPath", () => {
]);
FULLY_COVERED.mockReturnValue(true);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
const result = await computeBpInTargetFastPath({
userId: "user-honolulu",
@@ -343,7 +389,7 @@ describe("computeBpInTargetFastPath", () => {
expect(result.path).toBe("live");
expect(ROLLUP_FIND_MANY).not.toHaveBeenCalled();
expect(MEASUREMENT_FIND_MANY).toHaveBeenCalledTimes(2);
expect(QUERY_RAW).toHaveBeenCalledTimes(2);
const calls = ANNOTATE.mock.calls.map((c) => c[0]);
const bpCall = calls.find(
@@ -362,7 +408,7 @@ describe("computeBpInTargetFastPath", () => {
]);
FULLY_COVERED.mockReturnValue(true);
PROBE.mockResolvedValue(coverage);
MEASUREMENT_FIND_MANY.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
QUERY_RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
const result = await computeBpInTargetFastPath({
userId: "user-tokyo",
+60 -26
View File
@@ -54,6 +54,10 @@ import { prisma } from "@/lib/db";
import { annotate } from "@/lib/logging/context";
import { readRollupBuckets } from "@/lib/rollups/measurement-rollups";
import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
import {
buildSourceRankCase,
canonicalMeasurementsFrom,
} from "@/lib/analytics/source-rank-sql";
import {
probeRollupCoverage,
type RollupCoverageMap,
@@ -454,9 +458,33 @@ async function computeFromLive(
const DAY_MS = 24 * 60 * 60 * 1000;
const bpInTargetSince = new Date(now.getTime() - 365 * DAY_MS);
// v1.18.10 I-1 — collapse overlapping sources before windowing. BP is a
// multi-source metric (a Withings reading mirrored into Apple Health is two
// rows for one event). Without the collapse the live/cold path double-counts
// the shared reading in both the in-target denominator and the
// recency-weighted grade, returning a different number than the rollup/warm
// path (which reads source-collapsed `measurement_rollups`). Pick the
// ladder-winning source per (type, day) so cold == warm (web↔iOS parity).
const priorityJson = await loadUserSourcePriority(userId);
const rankUnqualified = buildSourceRankCase(
priorityJson,
'"type"',
'"source"',
);
const [sysData, diaData] = await Promise.all([
fetchSeriesChunked(userId, "BLOOD_PRESSURE_SYS", bpInTargetSince),
fetchSeriesChunked(userId, "BLOOD_PRESSURE_DIA", bpInTargetSince),
fetchSeriesChunked(
userId,
"BLOOD_PRESSURE_SYS",
bpInTargetSince,
rankUnqualified,
),
fetchSeriesChunked(
userId,
"BLOOD_PRESSURE_DIA",
bpInTargetSince,
rankUnqualified,
),
]);
const rowCount = sysData.length + diaData.length;
@@ -499,34 +527,40 @@ async function computeFromLive(
}
/**
* Cursor-paged read of one BP type. Mirrors the route's
* `fetchMeasurementSeriesChunked` shape but pruned to the only two
* fields the in-target helper consumes (`measuredAt` + `value`) so the
* Prisma round-trip stays minimal on the cold fallback path.
* Source-collapsed read of one BP type over the trailing 365 days.
*
* v1.18.10 I-1 — routes through `canonicalMeasurementsFrom` so a reading that
* exists under two sources (Withings + its Apple-Health mirror) is counted
* once per (type, day), matching the rollup fast-path. Pruned to the two
* fields the in-target helper consumes (`measuredAt` + `value`). Per-event
* rows of the winning source survive the collapse, so the downstream ±5-min
* pairing still sees individual readings. BP is low-volume even on multi-year
* accounts, so one source-collapsed query replaces the prior cursor paging.
*/
const CHUNK = 5000;
async function fetchSeriesChunked(
userId: string,
type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA",
// The 365-day window is expressed as a SQL interval inside the canonical
// subquery; the bound `Date` arg is retained for call-site symmetry.
since: Date,
rankUnqualified: string,
): Promise<BpReading[]> {
const out: BpReading[] = [];
let cursorId: string | undefined;
for (let page = 0; page < 1000; page++) {
const chunk = await prisma.measurement.findMany({
where: { userId, type, measuredAt: { gte: since }, deletedAt: null },
orderBy: [{ measuredAt: "asc" }, { id: "asc" }],
select: { id: true, measuredAt: true, value: true },
take: CHUNK,
...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : {}),
});
if (chunk.length === 0) break;
for (const row of chunk) {
out.push({ measuredAt: row.measuredAt, value: row.value });
}
if (chunk.length < CHUNK) break;
cursorId = chunk[chunk.length - 1].id;
}
return out;
void since;
const rows = await prisma.$queryRawUnsafe<
Array<{ measured_at: Date; value: number }>
>(
`
SELECT
m."measured_at" AS measured_at,
m."value"::double precision AS value
FROM ${canonicalMeasurementsFrom(rankUnqualified, "365 days")}
WHERE m."type" = '${type}'
ORDER BY m."measured_at" ASC
`,
userId,
);
return rows.map((r) => ({
measuredAt: new Date(r.measured_at),
value: Number(r.value),
}));
}
@@ -170,12 +170,10 @@ describe("buildComprehensiveAggregate", () => {
},
]);
// v1.4.38 W-F — sys + dia merged into a single `findMany`
// (`type: { in: [...] }`) → one mock call carrying both rows
// (empty here since the rollup-fresh fixture doesn't exercise
// the BP pairing). The aggregator partitions by type in JS so
// the bpRawRows.sys / .dia byte-shape is preserved.
FIND_MANY.mockResolvedValueOnce([]);
// v1.18.10 I-1 — bpRawRows now reads source-collapsed rows via
// `$queryRawUnsafe` (3rd UNSAFE call: narrow, latests, bpRaw). Empty
// here since the rollup-fresh fixture doesn't exercise BP pairing.
UNSAFE.mockResolvedValueOnce([]);
const result = await buildComprehensiveAggregate("user-rollup-fresh");
const weight = result.summaries.WEIGHT;
@@ -210,9 +208,10 @@ describe("buildComprehensiveAggregate", () => {
// DISTINCT-ON latest moved to `$queryRawUnsafe` (2 calls). Neither
// path runs the legacy heavy COUNT/MIN/MAX/AVG query.
expect(RAW).toHaveBeenCalledTimes(2);
expect(UNSAFE).toHaveBeenCalledTimes(2);
// v1.4.38 W-F — sys + dia merged into a single round-trip.
expect(FIND_MANY).toHaveBeenCalledTimes(1);
// v1.18.10 I-1 — narrow + latests + bpRaw all run on `$queryRawUnsafe`.
expect(UNSAFE).toHaveBeenCalledTimes(3);
// bpRawRows no longer uses `measurement.findMany`.
expect(FIND_MANY).not.toHaveBeenCalled();
expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1);
});
});
@@ -226,9 +225,10 @@ describe("buildComprehensiveAggregate", () => {
// 2. latests — empty.
// No firstMeasurementAt query when totalMeasurements === 0.
RAW.mockResolvedValueOnce([]);
UNSAFE.mockResolvedValueOnce([]).mockResolvedValueOnce([]);
// v1.4.38 W-F — sys + dia merged into one `findMany`.
FIND_MANY.mockResolvedValueOnce([]);
// heavy, latests, bpRaw — all empty.
UNSAFE.mockResolvedValueOnce([])
.mockResolvedValueOnce([])
.mockResolvedValueOnce([]);
const result = await buildComprehensiveAggregate("user-empty");
@@ -239,9 +239,9 @@ describe("buildComprehensiveAggregate", () => {
expect(result.firstMeasurementAt).toBeNull();
expect(result.totalMeasurements).toBe(0);
expect(RAW).toHaveBeenCalledTimes(1);
expect(UNSAFE).toHaveBeenCalledTimes(2);
// v1.4.38 W-F — sys + dia merged → single findMany.
expect(FIND_MANY).toHaveBeenCalledTimes(1);
// v1.18.10 I-1 — heavy + latests + bpRaw all on `$queryRawUnsafe`.
expect(UNSAFE).toHaveBeenCalledTimes(3);
expect(FIND_MANY).not.toHaveBeenCalled();
// 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);
@@ -280,11 +280,12 @@ describe("buildComprehensiveAggregate", () => {
slope90: 0.001,
r2_90: 0.12,
},
]).mockResolvedValueOnce([
{ type: "WEIGHT", value: 81.4, measured_at: now },
]);
// v1.4.38 W-F — sys + dia merged into one round-trip.
FIND_MANY.mockResolvedValueOnce([]);
])
.mockResolvedValueOnce([
{ type: "WEIGHT", value: 81.4, measured_at: now },
])
// v1.18.10 I-1 — bpRaw (empty) is the 3rd UNSAFE call.
.mockResolvedValueOnce([]);
const result = await buildComprehensiveAggregate("user-cold");
const weight = result.summaries.WEIGHT;
@@ -352,23 +353,34 @@ describe("buildComprehensiveAggregate", () => {
slope90: null,
r2_90: null,
},
]).mockResolvedValueOnce([
{
type: "BLOOD_PRESSURE_SYS",
value: 120,
measured_at: measuredAt,
},
]);
// v1.4.38 W-F — single merged `findMany` returning both sys + dia
// rows tagged by `type`. The aggregator partitions in JS, so the
// bpRawRows.sys / .dia shape stays byte-identical.
FIND_MANY.mockResolvedValueOnce([
{ type: "BLOOD_PRESSURE_SYS", measuredAt, value: 120 },
{ type: "BLOOD_PRESSURE_DIA", measuredAt, value: 80 },
]);
])
.mockResolvedValueOnce([
{
type: "BLOOD_PRESSURE_SYS",
value: 120,
measured_at: measuredAt,
},
])
// v1.18.10 I-1 — bpRaw runs on `$queryRawUnsafe` through the
// source-collapsed canonical subquery. The query returns the
// already-collapsed sys + dia rows in raw SQL snake_case shape; the
// aggregator partitions by type in JS so the bpRawRows.sys / .dia
// byte-shape stays identical.
.mockResolvedValueOnce([
{ type: "BLOOD_PRESSURE_SYS", measured_at: measuredAt, value: 120 },
{ type: "BLOOD_PRESSURE_DIA", measured_at: measuredAt, value: 80 },
]);
const result = await buildComprehensiveAggregate("user-bp");
expect(result.bpRawRows.sys).toEqual([{ measuredAt, value: 120 }]);
expect(result.bpRawRows.dia).toEqual([{ measuredAt, value: 80 }]);
// The BP pull must route through the canonical-source subquery (the
// collapse fires in SQL); pin the marker so a revert to a plain findMany
// (double-counting overlapping sources) fails here.
const bpSql = UNSAFE.mock.calls
.map((c) => String(c[0]))
.find((sql) => sql.includes("BLOOD_PRESSURE_SYS"));
expect(bpSql).toContain("user_id");
expect(FIND_MANY).not.toHaveBeenCalled();
});
});
+50 -19
View File
@@ -302,38 +302,67 @@ function timeSubquery<T>(
});
}
interface BpRawRow {
type: string;
measured_at: Date;
value: number;
}
/**
* v1.4.38 W-F — fetch the trailing-90d sys + dia raw rows in a single
* `findMany` (`type: { in: [...] }`) rather than two parallel queries.
* Saves one round-trip + lets Postgres consolidate the index scan.
* Partitioning by type in JS preserves the byte-shape of the legacy
* sys / dia arrays (ASC by measuredAt — already enforced by the SQL
* ORDER BY).
* round-trip rather than two parallel queries.
*
* v1.18.10 I-1 — collapse overlapping sources before returning. BP is a
* multi-source metric (a Withings reading mirrored into Apple Health is two
* rows for one event). The legacy `findMany` returned every source's row, so
* `computeBpInTargetPct` and the recency-weighted grade double-counted the
* shared reading — and a rollup-warm account (which reads source-collapsed
* `measurement_rollups` via `bp-in-target-fast-path`) returned a different
* BP percentage / grade than a rollup-cold account on identical data. Routing
* both raw pulls through `canonicalMeasurementsFrom` picks the ladder-winning
* source per (type, day) so the cold/live path matches the warm/rollup path
* (web↔iOS parity). `BLOOD_PRESSURE_SYS/DIA` are already in `RANKED_TYPES`.
*
* Per-event rows survive the collapse (the subquery keeps every reading of the
* winning source per day, not a daily mean), so the downstream ±5-min pairing
* still sees individual readings. Partitioning by type in JS preserves the
* byte-shape of the legacy sys / dia arrays (ASC by measuredAt).
*/
async function fetchBpRawRows(
userId: string,
// The 90-day window is expressed as a SQL interval inside the canonical
// subquery (parity with the rollup fast-path's `"90 days"`); the bound
// `Date` arg is retained for call-site symmetry and is exactly 90 days.
ninetyDaysAgo: Date,
// The source-rank CASE built once by the caller (over unqualified
// `"type"`/`"source"` columns) — reused so the BP pull doesn't re-load the
// user's source-priority blob the caller already resolved.
rankUnqualified: string,
): Promise<{
sys: Array<{ measuredAt: Date; value: number }>;
dia: Array<{ measuredAt: Date; value: number }>;
}> {
const rows = await prisma.measurement.findMany({
where: {
userId,
type: { in: ["BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"] },
measuredAt: { gte: ninetyDaysAgo },
deletedAt: null,
},
orderBy: { measuredAt: "asc" },
select: { type: true, measuredAt: true, value: true },
});
void ninetyDaysAgo;
const rows = await prisma.$queryRawUnsafe<BpRawRow[]>(
`
SELECT
m."type"::text AS type,
m."measured_at" AS measured_at,
m."value"::double precision AS value
FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
WHERE m."type" IN ('BLOOD_PRESSURE_SYS', 'BLOOD_PRESSURE_DIA')
ORDER BY m."measured_at" ASC
`,
userId,
);
const sys: Array<{ measuredAt: Date; value: number }> = [];
const dia: Array<{ measuredAt: Date; value: number }> = [];
for (const r of rows) {
const measuredAt = new Date(r.measured_at);
if (r.type === "BLOOD_PRESSURE_SYS") {
sys.push({ measuredAt: r.measuredAt, value: r.value });
sys.push({ measuredAt, value: Number(r.value) });
} else if (r.type === "BLOOD_PRESSURE_DIA") {
dia.push({ measuredAt: r.measuredAt, value: r.value });
dia.push({ measuredAt, value: Number(r.value) });
}
}
return { sys, dia };
@@ -481,7 +510,7 @@ async function buildFromRollups(
// readings; the per-type partition in `fetchBpRawRows` preserves
// the legacy `bpRawRows.sys` / `bpRawRows.dia` byte-shape.
timeSubquery(timings, "bpRaw", () =>
fetchBpRawRows(userId, ninetyDaysAgo),
fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified),
),
timeSubquery(
timings,
@@ -694,7 +723,9 @@ async function buildFromLiveAggregate(
}),
),
// v1.4.38 W-F — bp sys + dia in one round-trip (see `fetchBpRawRows`).
timeSubquery(timings, "bpRaw", () => fetchBpRawRows(userId, ninetyDaysAgo)),
timeSubquery(timings, "bpRaw", () =>
fetchBpRawRows(userId, ninetyDaysAgo, rankUnqualified),
),
]);
const latestByType = new Map<string, { value: number; measuredAt: Date }>();