Merge pull request #1142 from MBombeck/ready/pr-1134

fix(series): follow the source priority ladder in the blood-pressure series

src/app/api/measurements/series/__tests__/route.test.ts

@@ -90,10 +90,32 @@ function adaptSeriesSql(sql: string): Promise<unknown> {
       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([

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,55 +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.
-    const [sys, dia] = await Promise.all([
-      prisma.measurement.findMany({
-        where: {
-          userId: user.id,
-          type: "BLOOD_PRESSURE_SYS",
-          measuredAt: { gte: since },
-          deletedAt: null,
-          value: { gte: VALUE_RANGES.BLOOD_PRESSURE_SYS.min },
-        },
-        orderBy: { measuredAt: "asc" },
-        select: { id: true, value: true, measuredAt: true },
-      }),
-      prisma.measurement.findMany({
-        where: {
-          userId: user.id,
-          type: "BLOOD_PRESSURE_DIA",
-          measuredAt: { gte: since },
-          deletedAt: null,
-          value: { gte: VALUE_RANGES.BLOOD_PRESSURE_DIA.min },
-        },
-        orderBy: { measuredAt: "asc" },
-        select: { id: true, value: true, measuredAt: true },
-      }),
-    ]);
+    //
+    // 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,
+      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<string, typeof bpRows>();
+    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<string, number>();
     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,
       };

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;
 
@@ -114,7 +115,12 @@ async function series(userId: string, kind: string, days: number) {
   return (
     (await res.json()) as {
       data: {
-        points: Array<{ id: string; at: string; value: number }>;
+        points: Array<{
+          id: string;
+          at: string;
+          value: number;
+          secondary: number | null;
+        }>;
         stats: { count: number; mean: number; min: number; max: number };
       };
     }
@@ -123,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);
   }
@@ -146,6 +152,25 @@ beforeAll(async () => {
     dual.push(row(DUAL, "WEIGHT", "kg", "MANUAL", 80 + d, at(d)));
     dual.push(row(DUAL, "WEIGHT", "kg", "WITHINGS", 80 + d, at(d)));
   }
+  // Blood pressure: Apple Health (120/80, 122/82) on days 1-2, Withings
+  // (140/90, 142/92) on days 2-3. Day 2 has both sources; day 1 only Apple,
+  // day 3 only Withings. A sub-floor systolic (0) on day 1 must stay out.
+  const bp = (
+    source: MeasurementSource,
+    d: number,
+    sys: number,
+    dia: number,
+  ) => {
+    dual.push(row(DUAL, "BLOOD_PRESSURE_SYS", "mmHg", source, sys, at(d, 10)));
+    dual.push(row(DUAL, "BLOOD_PRESSURE_DIA", "mmHg", source, dia, at(d, 10)));
+  };
+  bp("APPLE_HEALTH", 1, 120, 80);
+  bp("APPLE_HEALTH", 2, 122, 82);
+  bp("WITHINGS", 2, 142, 92);
+  bp("WITHINGS", 3, 140, 90);
+  dual.push(
+    row(DUAL, "BLOOD_PRESSURE_SYS", "mmHg", "APPLE_HEALTH", 0, at(1, 20)),
+  );
   // Glucose has no ladder: both sources must stay.
   for (const d of [1, 2]) {
     dual.push(row(DUAL, "BLOOD_GLUCOSE", "mg/dL", "APPLE_HEALTH", 100, at(d)));
@@ -177,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) =>
@@ -220,6 +268,50 @@ describe("pulse follows the ladder (raw read, 30 days)", () => {
   });
 });
 
+describe("blood pressure follows the ladder", () => {
+  const bpSeries = async () => (await series(DUAL, "bloodPressure", 30)).points;
+
+  it("shows the ladder's source on a day both cover, and the other where only it exists", async () => {
+    await setLadder(DUAL, { bloodPressure: ["APPLE_HEALTH", "WITHINGS"] });
+    const points = await bpSeries();
+    // Day 1 Apple, day 2 Apple (Withings' 142/92 is out), day 3 only Withings.
+    expect(points.map((p) => [p.value, p.secondary])).toEqual([
+      [140, 90],
+      [122, 82],
+      [120, 80],
+    ]);
+  });
+
+  it("flips with the ladder", async () => {
+    await setLadder(DUAL, { bloodPressure: ["WITHINGS", "APPLE_HEALTH"] });
+    const points = await bpSeries();
+    expect(points.map((p) => [p.value, p.secondary])).toEqual([
+      [140, 90],
+      [142, 92],
+      [120, 80],
+    ]);
+  });
+
+  it("keeps a non-physiological systolic out of the series", async () => {
+    await setLadder(DUAL, { bloodPressure: ["APPLE_HEALTH", "WITHINGS"] });
+    expect((await bpSeries()).every((p) => p.value >= 40)).toBe(true);
+  });
+});
+
+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"] });