diff --git a/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts b/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts index a3a27e096..4115377c3 100644 --- a/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts +++ b/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts @@ -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; const ROLLUP_FIND_MANY = prisma.measurementRollup .findMany as unknown as ReturnType; +const QUERY_RAW = prisma.$queryRawUnsafe as unknown as ReturnType; const PROBE = probeRollupCoverage as unknown as ReturnType; const FULLY_COVERED = isFullyCovered as unknown as ReturnType; const ANNOTATE = annotate as unknown as ReturnType; @@ -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([ ["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([ + ["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([]); 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", diff --git a/src/lib/analytics/bp-in-target-fast-path.ts b/src/lib/analytics/bp-in-target-fast-path.ts index 6dfba6775..0d8bae265 100644 --- a/src/lib/analytics/bp-in-target-fast-path.ts +++ b/src/lib/analytics/bp-in-target-fast-path.ts @@ -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 { - 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), + })); } diff --git a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts index a2fc9143f..a07c4c200 100644 --- a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts +++ b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts @@ -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(); }); }); diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index 6e3e72226..42ffb8540 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -302,38 +302,67 @@ function timeSubquery( }); } +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( + ` + 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();