From d1ec6715d9823fcdd8e23026d07910b184c99970 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 14:14:22 +0200 Subject: [PATCH] fix(series): pick one source per day for both halves of a blood-pressure reading The systolic and diastolic reads each took the ladder's source per day on their own. Whenever a source held only half of a reading on a day (a deleted diastolic, an orphaned sample, a sub-floor systolic), the two picks diverged and the pairing joined one source's systolic to another source's diastolic. A sub-floor systolic could also claim a day and then contribute nothing, leaving the day empty although a lower-ranked source had a valid reading. The source is now picked per day from the systolic rows that pass the floor, both halves are read from it in one statement, and a pair is only made within one source, which also covers a reading that straddles midnight. --- .../series/__tests__/route.test.ts | 70 ++++++++++ src/app/api/measurements/series/route.ts | 132 ++++++++++++------ .../integration/series-source-ladder.test.ts | 40 +++++- 3 files changed, 196 insertions(+), 46 deletions(-) diff --git a/src/app/api/measurements/series/__tests__/route.test.ts b/src/app/api/measurements/series/__tests__/route.test.ts index 37146ef82..2d82924ff 100644 --- a/src/app/api/measurements/series/__tests__/route.test.ts +++ b/src/app/api/measurements/series/__tests__/route.test.ts @@ -90,10 +90,32 @@ function adaptSeriesSql(sql: string): Promise { id: string; value: number; measuredAt: Date; + source?: string; valueMin?: number | null; valueMax?: number | null; }> >; + // The blood-pressure read selects both halves in one statement, each row + // carrying its type and source. + if (sql.includes(`'BLOOD_PRESSURE_DIA'::"measurement_type"`)) { + return Promise.all( + (["BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"] as const).map((t) => + find({ where: { type: t } }).then((rows) => + rows.map((r) => ({ + id: r.id, + type: t, + source: r.source ?? "MANUAL", + value: r.value, + measured_at: r.measuredAt, + })), + ), + ), + ).then(([sys, dia]) => + [...sys, ...dia].sort( + (a, b) => a.measured_at.getTime() - b.measured_at.getTime(), + ), + ); + } return find({ where: { type } }).then((rows) => rows.map((r) => ({ id: r.id, @@ -211,6 +233,54 @@ describe("GET /api/measurements/series", () => { ]); }); + it("pairs a systolic only with a diastolic from its own source", async () => { + // Across midnight the two days may have picked different sources; the + // nearest diastolic then belongs to another source and must not be used. + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + const at = (iso: string) => new Date(iso); + vi.mocked(prisma.measurement.findMany).mockImplementation((( + args: unknown, + ) => { + const a = args as { where: { type: string } }; + if (a.where.type === "BLOOD_PRESSURE_SYS") { + return Promise.resolve([ + { + id: "s1", + value: 150, + source: "WITHINGS", + measuredAt: at("2026-05-01T23:59:00Z"), + }, + { + id: "s2", + value: 125, + source: "APPLE_HEALTH", + measuredAt: at("2026-05-02T00:01:00Z"), + }, + ]) as never; + } + if (a.where.type === "BLOOD_PRESSURE_DIA") { + return Promise.resolve([ + { + id: "d2", + value: 85, + source: "APPLE_HEALTH", + measuredAt: at("2026-05-02T00:01:00Z"), + }, + ]) as never; + } + return Promise.resolve([]) as never; + }) as never); + const res = await GET(req("kind=bloodPressure&days=30")); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { points: Array<{ id: string; secondary: number | null }> }; + }; + expect(body.data.points.map((p) => [p.id, p.secondary])).toEqual([ + ["s1", null], + ["s2", 85], + ]); + }); + it("returns single-value series for kind=weight", async () => { vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); vi.mocked(prisma.measurement.findMany).mockResolvedValue([ diff --git a/src/app/api/measurements/series/route.ts b/src/app/api/measurements/series/route.ts index 461048430..e98cdfc3b 100644 --- a/src/app/api/measurements/series/route.ts +++ b/src/app/api/measurements/series/route.ts @@ -27,6 +27,7 @@ import { reconstructSleepNights } from "@/lib/analytics/sleep-night"; import { VALUE_RANGES } from "@/lib/validations/measurement"; import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import { seriesRowsFrom } from "@/lib/measurements/series-canonical"; +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; import { resolveUserTimezone, userDayKey } from "@/lib/tz/resolver"; import { convertGlucose, resolveGlucoseUnit } from "@/lib/glucose"; @@ -329,65 +330,106 @@ export const GET = apiHandler(async (request: NextRequest) => { // enforced — a seed-era systolic-0, iOS #33 — must never surface as the // latest reading. Floors mirror VALUE_RANGES (the input validator's min); // this is a read-side selection guard, not a tightening of the write floor. - // Each of the two types reads the ladder-winning source per day (see - // `seriesRowsFrom`), like the other kinds, so a second provider is not - // blended into the pairs and a reading present in both appears once. - const bpPriority = await loadUserSourcePriority(user.id); - const readBp = ( - type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA", - floor: number, - ) => - prisma - .$queryRawUnsafe< - Array<{ id: string; value: number; measured_at: Date }> - >( - ` - SELECT m."id", m."value", m."measured_at" - FROM ${seriesRowsFrom(bpPriority, type, days)} - WHERE m."measured_at" >= $2 - AND m."value" >= $3 + // + // The ladder picks ONE source per day for the reading, not one per type: + // the source is chosen from the day's systolic rows that pass the floor, + // and both halves are read from it. Two per-type picks diverge whenever a + // source holds only half of a reading on a day (a deleted diastolic, a + // sub-floor systolic, an orphaned sample), and the pairing below then + // joined one source's systolic to another source's diastolic. Picking on + // floor-passing systolic rows also keeps a corrupt row from claiming a day + // it then contributes nothing to. The day is the UTC day, as in every + // other ladder read (`canonicalMeasurementsCte`), and the pick looks at + // the whole edge day, not only the slice after `since`. + // + // `rank` is the closed-enum whitelist splice from `buildSourceRankCase`; + // every request value is a positional parameter. + const rank = buildSourceRankCase( + await loadUserSourcePriority(user.id), + '"type"', + '"source"', + ); + const bpRows = await prisma.$queryRawUnsafe< + Array<{ + id: string; + type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA"; + source: string; + value: number; + measured_at: Date; + }> + >( + ` + SELECT m."id", m."type"::text AS type, m."source"::text AS source, + m."value", m."measured_at" + FROM measurements m + JOIN ( + SELECT DISTINCT ON (date_trunc('day', "measured_at")) + date_trunc('day', "measured_at") AS d, + "source" AS canon + FROM measurements + WHERE "user_id" = $1 + AND "type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" + AND "deleted_at" IS NULL + AND "value" >= $3 + AND "measured_at" >= date_trunc('day', $2::timestamptz) + ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text + ) c + ON c.d = date_trunc('day', m."measured_at") + AND c.canon = m."source" + WHERE m."user_id" = $1 + AND m."deleted_at" IS NULL + AND m."measured_at" >= $2 + AND ( + (m."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" AND m."value" >= $3) + OR (m."type" = 'BLOOD_PRESSURE_DIA'::"measurement_type" AND m."value" >= $4) + ) ORDER BY m."measured_at" ASC, m."id" ASC `, - user.id, - since, - floor, - ) - .then((rows) => - rows.map((r) => ({ - id: r.id, - value: r.value, - measuredAt: r.measured_at, - })), - ); - const [sys, dia] = await Promise.all([ - readBp("BLOOD_PRESSURE_SYS", VALUE_RANGES.BLOOD_PRESSURE_SYS.min), - readBp("BLOOD_PRESSURE_DIA", VALUE_RANGES.BLOOD_PRESSURE_DIA.min), - ]); + user.id, + since, + VALUE_RANGES.BLOOD_PRESSURE_SYS.min, + VALUE_RANGES.BLOOD_PRESSURE_DIA.min, + ); + const sys = bpRows.filter((r) => r.type === "BLOOD_PRESSURE_SYS"); + // Diastolic rows per source: a pair is only ever made inside one source, + // which also holds across midnight, where the two days may have picked + // different sources. + const diaBySource = new Map(); + for (const r of bpRows) { + if (r.type !== "BLOOD_PRESSURE_DIA") continue; + const list = diaBySource.get(r.source); + if (list) list.push(r); + else diaBySource.set(r.source, [r]); + } // Pair by closest timestamp within ±5 minutes. Both reads arrive // sorted ascending, so the nearest-diastolic index is non-decreasing - // across the systolic walk — a two-pointer merge finds each pair in - // O(sys + dia) instead of the previous full diastolic rescan per - // systolic row (O(sys × dia) — quadratic on dense imports). Ties keep - // the earlier diastolic row, matching the old scan's strict-`<` keep. + // across the systolic walk — a two-pointer merge (one pointer per + // source) finds each pair in O(sys + dia) instead of the previous full + // diastolic rescan per systolic row (O(sys × dia) — quadratic on dense + // imports). Ties keep the earlier diastolic row, matching the old + // scan's strict-`<` keep. const PAIR_WINDOW_MS = 5 * 60_000; - let diaIdx = 0; + const diaIdx = new Map(); points = sys.map((s) => { - const sysMs = s.measuredAt.getTime(); + const sysMs = s.measured_at.getTime(); + const dia = diaBySource.get(s.source) ?? []; + let idx = diaIdx.get(s.source) ?? 0; while ( - diaIdx + 1 < dia.length && - Math.abs(dia[diaIdx + 1].measuredAt.getTime() - sysMs) < - Math.abs(dia[diaIdx].measuredAt.getTime() - sysMs) + idx + 1 < dia.length && + Math.abs(dia[idx + 1].measured_at.getTime() - sysMs) < + Math.abs(dia[idx].measured_at.getTime() - sysMs) ) { - diaIdx += 1; + idx += 1; } - const best = dia[diaIdx]; + diaIdx.set(s.source, idx); + const best = dia[idx]; const paired = best !== undefined && - Math.abs(best.measuredAt.getTime() - sysMs) <= PAIR_WINDOW_MS; + Math.abs(best.measured_at.getTime() - sysMs) <= PAIR_WINDOW_MS; return { id: s.id, - at: s.measuredAt.toISOString(), + at: s.measured_at.toISOString(), value: s.value, secondary: paired ? best.value : null, }; diff --git a/tests/integration/series-source-ladder.test.ts b/tests/integration/series-source-ladder.test.ts index 6cb75c099..440d9fc71 100644 --- a/tests/integration/series-source-ladder.test.ts +++ b/tests/integration/series-source-ladder.test.ts @@ -55,6 +55,7 @@ const TZ = "Europe/Berlin"; const DUAL = "ladder-dual"; const DENSE = "ladder-dense"; const SOLO = "ladder-solo"; +const BP_MIX = "ladder-bp-mix"; const NOW = new Date(Math.floor(Date.now() / 60_000) * 60_000 - 60_000); const DAY = 86_400_000; @@ -128,7 +129,7 @@ async function series(userId: string, kind: string, days: number) { beforeAll(async () => { await truncateAllTables(prisma); - for (const id of [DUAL, DENSE, SOLO]) { + for (const id of [DUAL, DENSE, SOLO, BP_MIX]) { await prisma.user.create({ data: { id, username: id, timezone: TZ } }); invalidateUserTimezone(id); } @@ -201,6 +202,29 @@ beforeAll(async () => { ); } + // Blood pressure where one source holds only half of a reading. Day 1: the + // cuff's systolic 150 is live but its diastolic was deleted, while Apple + // Health holds the same moment as a full 125/85. Day 2: the cuff's only + // systolic that day is a sub-floor 0 next to a diastolic 80, while Apple + // Health holds 126/86. A per-type pick takes the cuff's systolic and Apple's + // diastolic on day 1 and pairs them across sources, and on day 2 picks the + // cuff for a systolic the floor then removes, which empties the day. + await prisma.measurement.createMany({ + data: [ + row(BP_MIX, "BLOOD_PRESSURE_SYS", "mmHg", "WITHINGS", 150, at(1, 10)), + { + ...row(BP_MIX, "BLOOD_PRESSURE_DIA", "mmHg", "WITHINGS", 95, at(1, 10)), + deletedAt: new Date(), + }, + row(BP_MIX, "BLOOD_PRESSURE_SYS", "mmHg", "APPLE_HEALTH", 125, at(1, 10)), + row(BP_MIX, "BLOOD_PRESSURE_DIA", "mmHg", "APPLE_HEALTH", 85, at(1, 10)), + row(BP_MIX, "BLOOD_PRESSURE_SYS", "mmHg", "WITHINGS", 0, at(2, 10)), + row(BP_MIX, "BLOOD_PRESSURE_DIA", "mmHg", "WITHINGS", 80, at(2, 10)), + row(BP_MIX, "BLOOD_PRESSURE_SYS", "mmHg", "APPLE_HEALTH", 126, at(2, 12)), + row(BP_MIX, "BLOOD_PRESSURE_DIA", "mmHg", "APPLE_HEALTH", 86, at(2, 12)), + ], + }); + // A single-source account. await prisma.measurement.createMany({ data: [1, 2, 3, 4].map((d) => @@ -274,6 +298,20 @@ describe("blood pressure follows the ladder", () => { }); }); +describe("blood pressure keeps a reading's two halves from one source", () => { + it("never pairs one source's systolic with another source's diastolic", async () => { + await setLadder(BP_MIX, { bloodPressure: ["WITHINGS", "APPLE_HEALTH"] }); + const points = (await series(BP_MIX, "bloodPressure", 30)).points; + // Day 1 is the cuff's day: its systolic stands alone rather than borrowing + // Apple's diastolic. Day 2 is Apple's, because the cuff has no systolic + // that passes the floor there. + expect(points.map((p) => [p.value, p.secondary])).toEqual([ + [126, 86], + [150, null], + ]); + }); +}); + describe("a reading mirrored into two sources", () => { it("appears once", async () => { await setLadder(DUAL, { weight: ["WITHINGS", "MANUAL"] });