From fb4e3921e27d813c817ec15180d455f44ac99510 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 20 Jun 2026 16:51:51 +0200 Subject: [PATCH 1/4] feat(withings): capture nightly sleep vitals from the summary endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-segment sleep sync wrote only the stage timeline; the per-night summary's vitals — average HR, respiratory rate, SDNN-based HRV, average SpO2 and the sleep score — were dropped at the fetch. Add a getsummary fetch that maps each vital to its existing MeasurementType (RESTING_HEART_RATE, RESPIRATORY_RATE, HEART_RATE_VARIABILITY, OXYGEN_SATURATION, SLEEP_SCORE), so no migration and no client-contract change: the rows flow to the sync delta under the WITHINGS source like every other server-side fetch. Each vital upserts on the stable (userId, type, source, externalId) composite keyed by the night's session id, so a re-sync overwrites rather than duplicates. A summary fetch failure is isolated so the stage segments already written survive. Fractional SpO2 is normalised to percent. --- src/lib/withings/__tests__/sync-sleep.test.ts | 159 ++++++++++++++ src/lib/withings/sync-sleep.ts | 201 ++++++++++++++++++ 2 files changed, 360 insertions(+) diff --git a/src/lib/withings/__tests__/sync-sleep.test.ts b/src/lib/withings/__tests__/sync-sleep.test.ts index 70c985e69..91bad4170 100644 --- a/src/lib/withings/__tests__/sync-sleep.test.ts +++ b/src/lib/withings/__tests__/sync-sleep.test.ts @@ -14,6 +14,7 @@ vi.mock("@/lib/db", () => ({ findFirst: vi.fn(), create: vi.fn(), update: vi.fn(), + upsert: vi.fn(), }, withingsConnection: { findUnique: vi.fn() }, }, @@ -69,6 +70,7 @@ import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollup import { fetchWithingsSleep, mapWithingsSleepState, + mapWithingsSleepSummary, syncUserSleep, } from "../sync-sleep"; @@ -107,6 +109,71 @@ afterEach(() => { vi.unstubAllGlobals(); }); +/** + * v1.18.10 P0 — fetch mock that returns the per-segment series on the first + * call (`action=get`) and the per-night summary series on the second + * (`action=getsummary`). Lets the sync exercise both the stage path and the + * nightly-vital path in one round-trip. + */ +function installSegmentThenSummaryFetch( + segments: FakeSegment[], + summaries: Array<{ + id?: number; + startdate: number; + enddate: number; + data?: Record; + }>, +) { + let call = 0; + const fetchMock = vi.fn(async () => { + const body = call === 0 ? { series: segments } : { series: summaries }; + call++; + return { status: 200, json: async () => ({ status: 0, body }) }; + }); + vi.stubGlobal("fetch", fetchMock); + return fetchMock; +} + +describe("mapWithingsSleepSummary", () => { + it("maps the canonical nightly vitals to existing enums", () => { + const rows = mapWithingsSleepSummary({ + hr_average: 58, + rr_average: 14, + sdnn_1: 65, + spo2_average: 96, + sleep_score: 82, + }); + const byType = Object.fromEntries(rows.map((r) => [r.type, r.value])); + expect(byType).toEqual({ + RESTING_HEART_RATE: 58, + RESPIRATORY_RATE: 14, + HEART_RATE_VARIABILITY: 65, + OXYGEN_SATURATION: 96, + SLEEP_SCORE: 82, + }); + }); + + it("normalises a fractional SpO2 (0..1) to percent", () => { + const rows = mapWithingsSleepSummary({ spo2_average: 0.97 }); + const spo2 = rows.find((r) => r.type === "OXYGEN_SATURATION"); + expect(spo2?.value).toBeCloseTo(97, 5); + }); + + it("skips omitted / non-finite fields", () => { + const rows = mapWithingsSleepSummary({ + hr_average: 60, + rr_average: null, + sdnn_1: undefined as unknown as number, + sleep_score: Number.NaN, + }); + expect(rows.map((r) => r.type)).toEqual(["RESTING_HEART_RATE"]); + }); + + it("returns nothing for an absent data block", () => { + expect(mapWithingsSleepSummary(undefined)).toEqual([]); + }); +}); + describe("mapWithingsSleepState", () => { it("maps state 0 (awake) → AWAKE", () => { expect(mapWithingsSleepState(0)).toBe("AWAKE"); @@ -271,6 +338,98 @@ describe("syncUserSleep — segment writes + idempotency", () => { expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); }); + it("upserts nightly vitals with stable per-vital externalIds (v1.18.10 P0)", async () => { + const nightEnd = Math.floor(Date.UTC(2026, 4, 13, 6) / 1000); + installSegmentThenSummaryFetch( + [], + [ + { + id: 555, + startdate: nightEnd - 8 * 3600, + enddate: nightEnd, + data: { + hr_average: 57, + rr_average: 13, + sdnn_1: 70, + spo2_average: 95, + sleep_score: 88, + }, + }, + ], + ); + vi.mocked(prisma.measurement.upsert).mockResolvedValue({} as never); + + const imported = await syncUserSleep("user-1"); + // Five vitals upserted (no stage segments in this fixture). + expect(imported).toBe(5); + + const calls = vi.mocked(prisma.measurement.upsert).mock.calls; + const byType = new Map( + calls.map((c) => { + const w = c[0] as { + where: { + userId_type_source_externalId: { type: string; externalId: string }; + }; + create: { value: number }; + }; + return [ + w.where.userId_type_source_externalId.type, + { + externalId: w.where.userId_type_source_externalId.externalId, + value: w.create.value, + }, + ]; + }), + ); + expect(byType.get("RESTING_HEART_RATE")).toEqual({ + externalId: "withings:sleep:user-1:555:hr", + value: 57, + }); + expect(byType.get("SLEEP_SCORE")).toEqual({ + externalId: "withings:sleep:user-1:555:score", + value: 88, + }); + expect(byType.get("OXYGEN_SATURATION")?.value).toBe(95); + }); + + it("keeps stage segments when the summary fetch fails (v1.18.10 P0)", async () => { + let call = 0; + vi.stubGlobal( + "fetch", + vi.fn(async () => { + const isSummary = call > 0; + call++; + return { + status: 200, + json: async () => + isSummary + ? { status: 293 } // summary errors + : { + status: 0, + body: { + series: [ + { + startdate: 1715000000, + enddate: 1715003600, + state: 2, + id: 1, + }, + ], + }, + }, + }; + }), + ); + vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); + vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + + const imported = await syncUserSleep("user-1"); + // The one stage segment is still written; the summary failure is swallowed. + expect(imported).toBe(1); + expect(prisma.measurement.create).toHaveBeenCalledTimes(1); + expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); + }); + it("tolerates a ScanWatch night with no REM segment (no all-zeros row)", async () => { // ScanWatch reports CORE + DEEP only; a missing REM should NOT // synthesise an all-zeros REM row. Withings simply omits the diff --git a/src/lib/withings/sync-sleep.ts b/src/lib/withings/sync-sleep.ts index 27b643baf..31940c073 100644 --- a/src/lib/withings/sync-sleep.ts +++ b/src/lib/withings/sync-sleep.ts @@ -61,6 +61,87 @@ import { const WITHINGS_SLEEP_URL = "https://wbsapi.withings.net/v2/sleep"; +/** + * v1.18.10 P0 — nightly sleep vitals from `action=getsummary`. + * + * The per-segment `action=get` series carries only the stage timeline; the + * per-night summary carries the physiological vitals (average HR, respiratory + * rate, SDNN-based HRV, average SpO2, the Withings sleep score). These were + * previously dropped at the fetch even though the canonical enums exist. Each + * vital maps to an EXISTING `MeasurementType` so there is no migration and no + * iOS-contract change — the rows simply flow to `/api/sync/changes` under the + * `WITHINGS` source like every other server-side fetch. + * + * Field set requested via `data_fields`. Withings nests the requested fields + * under each night's `data` object. Only the fields with a canonical enum are + * requested; snoring / breathing-disturbance intensity have no enum and are + * deliberately left out (no home to store them). + */ +const SLEEP_SUMMARY_DATA_FIELDS = + "hr_average,rr_average,sdnn_1,spo2_average,sleep_score"; + +/** + * One nightly summary row from `POST /v2/sleep?action=getsummary`. The vitals + * live under `data`; `id` is the per-night session id (stable across re-syncs, + * the same id the segment series carries) so the externalId reconciles. + */ +export interface WithingsSleepSummary { + id?: number; + startdate: number; + enddate: number; + data?: { + hr_average?: number | null; + rr_average?: number | null; + sdnn_1?: number | null; + spo2_average?: number | null; + sleep_score?: number | null; + }; +} + +/** + * One nightly vital extracted from a summary, ready to upsert. The + * `measuredAt` is the night's END (`enddate`) so the vital sorts under the + * same night as the END-stamped stage segments. + */ +interface SleepVitalRow { + type: MeasurementType; + value: number; + /** Stable per-vital externalId suffix; combined with the session id. */ + fieldTag: string; +} + +/** + * Map a nightly summary's `data` block to the canonical vital rows. Skips any + * field Withings omitted or returned as a non-finite value. `spo2_average` + * arrives as a fraction (0..1) on some firmware and a percent (0..100) on + * others — normalise the fraction form to percent so it shares the + * `OXYGEN_SATURATION` bucket with the meastype-54 spot reading. + */ +export function mapWithingsSleepSummary( + data: WithingsSleepSummary["data"] | undefined, +): SleepVitalRow[] { + if (!data) return []; + const rows: SleepVitalRow[] = []; + const push = (type: MeasurementType, value: unknown, fieldTag: string) => { + if (typeof value !== "number" || !Number.isFinite(value)) return; + rows.push({ type, value, fieldTag }); + }; + // Average HR over the sleep window → RESTING_HEART_RATE (one nightly sample, + // distinct from spot PULSE — same convention the HealthKit nightly RHR uses). + push("RESTING_HEART_RATE", data.hr_average, "hr"); + push("RESPIRATORY_RATE", data.rr_average, "rr"); + // sdnn_1 is the SDNN-based HRV estimator — the same estimator the + // HEART_RATE_VARIABILITY enum documents (kept distinct from WHOOP's RMSSD). + push("HEART_RATE_VARIABILITY", data.sdnn_1, "hrv"); + let spo2 = data.spo2_average; + if (typeof spo2 === "number" && Number.isFinite(spo2) && spo2 <= 1) { + spo2 = spo2 * 100; + } + push("OXYGEN_SATURATION", spo2, "spo2"); + push("SLEEP_SCORE", data.sleep_score, "score"); + return rows; +} + /** * Default backfill window for the first sleep sync of a connection. * Matches every other Withings ingest path so a reconnect lights up @@ -83,6 +164,16 @@ export interface WithingsSleepSegment { id?: number; } +/** + * Format a Date as `YYYY-MM-DD` in UTC for the summary endpoint's + * `startdateymd` / `enddateymd` params. The summary window is day-grained; + * UTC keeps it deterministic and matches the unix-second window the segment + * fetch derives from the same `start` / `now` Dates. + */ +function ymdUtc(d: Date): string { + return d.toISOString().slice(0, 10); +} + /** * Map Withings `state` to the HealthLog `SleepStage` enum. Returns * `null` for states we deliberately ignore (state 4 = synthetic @@ -153,6 +244,56 @@ export async function fetchWithingsSleep( return series; } +/** + * v1.18.10 P0 — fetch the per-night sleep summaries (with vitals) for the + * given window. `action=getsummary` keys by `lastupdate`/`startdateymd`; we + * use the same unix-second window the segment fetch uses, expressed as the + * `YYYY-MM-DD` form the summary endpoint expects. Returns the raw summaries + * (no DB write) so the caller composes the upsert step. + */ +export async function fetchWithingsSleepSummary( + accessToken: string, + startymd: string, + endymd: string, +): Promise { + const params = new URLSearchParams({ + action: "getsummary", + startdateymd: startymd, + enddateymd: endymd, + data_fields: SLEEP_SUMMARY_DATA_FIELDS, + }); + + const pageStart = performance.now(); + const res = await safeFetch(WITHINGS_SLEEP_URL, { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + Authorization: `Bearer ${accessToken}`, + }, + body: params.toString(), + }); + const json = await res.json(); + const verdict = classifyWithingsResponse(res.status, json); + getEvent()?.addExternalCall({ + service: "withings", + method: "fetchWithingsSleepSummary", + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new WithingsApiError({ + verb: "sleep", + classification: verdict.classification, + withingsStatus: verdict.withingsStatus, + reason: verdict.reason, + upstreamError: typeof json?.error === "string" ? json.error : undefined, + }); + } + const series: WithingsSleepSummary[] = json.body?.series ?? []; + return series; +} + /** * Sync sleep data for a single user. Walks the trailing 30-day window * (or whatever `SLEEP_BACKFILL_DAYS` is set to), writing one @@ -319,6 +460,66 @@ export async function syncUserSleep( segmentIndex++; } + // v1.18.10 P0 — nightly sleep vitals (avg HR / respiratory rate / SDNN HRV + // / avg SpO2 / sleep score) from the per-night summary. A summary-fetch + // failure must not lose the stage segments already written, so it runs in + // its own try/catch and only annotates on error. Each vital upserts on the + // stable `(userId, type, source, externalId)` composite so a re-sync + // overwrites the night's value rather than inserting a duplicate. + try { + const startYmd = ymdUtc(start); + const endYmd = ymdUtc(now); + const summaries = await fetchWithingsSleepSummary( + tokenInfo.accessToken, + startYmd, + endYmd, + ); + for (const summary of summaries) { + const measuredAt = new Date(summary.enddate * 1000); + const sessionId = summary.id ?? "no-id"; + const vitals = mapWithingsSleepSummary(summary.data); + for (const vital of vitals) { + const externalId = `withings:sleep:${userId}:${sessionId}:${vital.fieldTag}`; + try { + await prisma.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type: vital.type, + source: "WITHINGS", + externalId, + }, + }, + create: { + userId, + type: vital.type, + value: vital.value, + unit: getUnitForType(vital.type), + measuredAt, + source: "WITHINGS", + externalId, + }, + update: { + value: vital.value, + measuredAt, + syncVersion: { increment: 1 }, + }, + }); + touched.push({ type: vital.type, measuredAt }); + imported++; + } catch (err) { + getEvent()?.addWarning( + `withings sleep: failed to upsert vital (${vital.type}, ${measuredAt.toISOString()}): ${err}`, + ); + } + } + } + } catch (err) { + getEvent()?.addWarning( + `withings sleep: summary fetch failed for ${userId}: ${err}`, + ); + } + // v1.4.39.1 — refresh the persistent rollup table for every distinct // (type, day) the sync touched. Sleep segments collapse heavily — // ~10 stage rows per night land in one DAY recompute. From 013241ba746328f6b5feb6eb63ff3eb0ccc2bfa6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 20 Jun 2026 16:52:00 +0200 Subject: [PATCH 2/4] fix(analytics): collapse overlapping sources on the cold BP path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../__tests__/bp-in-target-fast-path.test.ts | 72 +++++++++++++--- src/lib/analytics/bp-in-target-fast-path.ts | 86 +++++++++++++------ .../comprehensive-aggregator.test.ts | 80 +++++++++-------- src/lib/insights/comprehensive-aggregator.ts | 69 +++++++++++---- 4 files changed, 215 insertions(+), 92 deletions(-) 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(); From 56d8cf19c58760b7b2f142f3af06db5c97258c01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 20 Jun 2026 16:52:10 +0200 Subject: [PATCH 3/4] fix(sources): ladder STRESS_SCORE so the source collapse stays deterministic STRESS_SCORE has no source-priority ladder, so metricKeyForType returns null and the SQL source collapse falls to an alphabetical rank-90 tiebreak. It is safe today because the only producer is the COMPUTED HRV-derived proxy, but the instant a device-native producer (Oura daily_stress) is enabled both rows survive and blend nondeterministically. Add a stress metric key with an OURA-above-COMPUTED default ladder, wire it through metricKeyForType and RANKED_TYPES, and surface it in the Sources settings (all six locales). A guard test asserts every RANKED_TYPES member resolves to a non-null metric key so a future ranked type can't ship unladdered. Regenerate the OpenAPI for the additive source-priority key. --- docs/api/openapi.yaml | 96 +++++++++++++++++++ messages/de.json | 3 +- messages/en.json | 3 +- messages/es.json | 3 +- messages/fr.json | 3 +- messages/it.json | 3 +- messages/pl.json | 3 +- src/components/settings/sources-section.tsx | 2 + src/lib/analytics/source-rank-sql.ts | 6 +- .../__tests__/cumulative-day-sum.test.ts | 21 ++++ src/lib/measurements/cumulative-day-sum.ts | 6 ++ src/lib/validations/source-priority.ts | 12 +++ 12 files changed, 154 insertions(+), 7 deletions(-) diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 1a90c78ad..0fd3b5992 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -3606,6 +3606,22 @@ paths: - NIGHTSCOUT - POLAR - OURA + stress: + maxItems: 8 + type: array + items: + type: string + enum: + - MANUAL + - WITHINGS + - IMPORT + - APPLE_HEALTH + - COMPUTED + - WHOOP + - FITBIT + - NIGHTSCOUT + - POLAR + - OURA metricPriority: type: object properties: @@ -3881,6 +3897,22 @@ paths: - NIGHTSCOUT - POLAR - OURA + stress: + maxItems: 8 + type: array + items: + type: string + enum: + - MANUAL + - WITHINGS + - IMPORT + - APPLE_HEALTH + - COMPUTED + - WHOOP + - FITBIT + - NIGHTSCOUT + - POLAR + - OURA deviceTypePriority: type: object properties: @@ -14406,6 +14438,22 @@ components: - NIGHTSCOUT - POLAR - OURA + stress: + maxItems: 8 + type: array + items: + type: string + enum: + - MANUAL + - WITHINGS + - IMPORT + - APPLE_HEALTH + - COMPUTED + - WHOOP + - FITBIT + - NIGHTSCOUT + - POLAR + - OURA metricPriority: type: object properties: @@ -14681,6 +14729,22 @@ components: - NIGHTSCOUT - POLAR - OURA + stress: + maxItems: 8 + type: array + items: + type: string + enum: + - MANUAL + - WITHINGS + - IMPORT + - APPLE_HEALTH + - COMPUTED + - WHOOP + - FITBIT + - NIGHTSCOUT + - POLAR + - OURA additionalProperties: false deviceTypePriority: type: object @@ -15002,6 +15066,22 @@ components: - NIGHTSCOUT - POLAR - OURA + stress: + maxItems: 8 + type: array + items: + type: string + enum: + - MANUAL + - WITHINGS + - IMPORT + - APPLE_HEALTH + - COMPUTED + - WHOOP + - FITBIT + - NIGHTSCOUT + - POLAR + - OURA metricPriority: type: object properties: @@ -15277,6 +15357,22 @@ components: - NIGHTSCOUT - POLAR - OURA + stress: + maxItems: 8 + type: array + items: + type: string + enum: + - MANUAL + - WITHINGS + - IMPORT + - APPLE_HEALTH + - COMPUTED + - WHOOP + - FITBIT + - NIGHTSCOUT + - POLAR + - OURA additionalProperties: false deviceTypePriority: type: object diff --git a/messages/de.json b/messages/de.json index 40d7ddbd8..364f6d650 100644 --- a/messages/de.json +++ b/messages/de.json @@ -3870,7 +3870,8 @@ "vo2Max": "VO2 max", "skinTemperature": "Hauttemperatur", "respiratoryRate": "Atemfrequenz", - "recovery": "Erholung" + "recovery": "Erholung", + "stress": "Stress" }, "sourceLabels": { "WITHINGS": "Withings", diff --git a/messages/en.json b/messages/en.json index 54241ad76..b1600d2b8 100644 --- a/messages/en.json +++ b/messages/en.json @@ -3870,7 +3870,8 @@ "vo2Max": "VO2 max", "skinTemperature": "Skin temperature", "respiratoryRate": "Respiratory rate", - "recovery": "Recovery" + "recovery": "Recovery", + "stress": "Stress" }, "sourceLabels": { "WITHINGS": "Withings", diff --git a/messages/es.json b/messages/es.json index 4b8561ed3..6e95bb739 100644 --- a/messages/es.json +++ b/messages/es.json @@ -3870,7 +3870,8 @@ "vo2Max": "VO2 máx.", "skinTemperature": "Temperatura de la piel", "respiratoryRate": "Frecuencia respiratoria", - "recovery": "Recuperación" + "recovery": "Recuperación", + "stress": "Estrés" }, "sourceLabels": { "WITHINGS": "Withings", diff --git a/messages/fr.json b/messages/fr.json index 71ec8f395..0efc0cb1d 100644 --- a/messages/fr.json +++ b/messages/fr.json @@ -3870,7 +3870,8 @@ "vo2Max": "VO2 max", "skinTemperature": "Température cutanée", "respiratoryRate": "Fréquence respiratoire", - "recovery": "Récupération" + "recovery": "Récupération", + "stress": "Stress" }, "sourceLabels": { "WITHINGS": "Withings", diff --git a/messages/it.json b/messages/it.json index 712b5c259..32468cad7 100644 --- a/messages/it.json +++ b/messages/it.json @@ -3870,7 +3870,8 @@ "vo2Max": "VO2 max", "skinTemperature": "Temperatura cutanea", "respiratoryRate": "Frequenza respiratoria", - "recovery": "Recupero" + "recovery": "Recupero", + "stress": "Stress" }, "sourceLabels": { "WITHINGS": "Withings", diff --git a/messages/pl.json b/messages/pl.json index 832a4d5d2..07834c1a3 100644 --- a/messages/pl.json +++ b/messages/pl.json @@ -3870,7 +3870,8 @@ "vo2Max": "VO2 max", "skinTemperature": "Temperatura skóry", "respiratoryRate": "Częstość oddechów", - "recovery": "Regeneracja" + "recovery": "Regeneracja", + "stress": "Stres" }, "sourceLabels": { "WITHINGS": "Withings", diff --git a/src/components/settings/sources-section.tsx b/src/components/settings/sources-section.tsx index 237b019c7..c07fa2ae4 100644 --- a/src/components/settings/sources-section.tsx +++ b/src/components/settings/sources-section.tsx @@ -87,6 +87,8 @@ const METRIC_LABEL_KEYS: Record = { skinTemperature: "settings.sections.sources.metrics.skinTemperature", respiratoryRate: "settings.sections.sources.metrics.respiratoryRate", recovery: "settings.sections.sources.metrics.recovery", + // v1.18.10 I-5 — native-vs-derived stress ladder. + stress: "settings.sections.sources.metrics.stress", }; const SOURCE_LABEL_KEYS: Record = { diff --git a/src/lib/analytics/source-rank-sql.ts b/src/lib/analytics/source-rank-sql.ts index 5aac2e1c4..6b78f9d2a 100644 --- a/src/lib/analytics/source-rank-sql.ts +++ b/src/lib/analytics/source-rank-sql.ts @@ -31,7 +31,7 @@ import { * absent: they only ever have one source per (type, day), so the collapse is * a no-op for them and they fall into the CASE's ELSE branch. */ -const RANKED_TYPES: readonly MeasurementType[] = [ +export const RANKED_TYPES: readonly MeasurementType[] = [ "RESTING_HEART_RATE", "HEART_RATE_VARIABILITY", "RESPIRATORY_RATE", @@ -45,6 +45,10 @@ const RANKED_TYPES: readonly MeasurementType[] = [ "PULSE", "VO2_MAX", "RECOVERY_SCORE", + // v1.18.10 I-5 — STRESS_SCORE carries a source-priority ladder (`stress`) + // so the SQL collapse ranks a future device-native producer above the + // COMPUTED proxy deterministically. No-op until a second producer lands. + "STRESS_SCORE", "SLEEP_DURATION", "ACTIVITY_STEPS", "ACTIVE_ENERGY_BURNED", diff --git a/src/lib/measurements/__tests__/cumulative-day-sum.test.ts b/src/lib/measurements/__tests__/cumulative-day-sum.test.ts index d4a63684f..eb993ce5d 100644 --- a/src/lib/measurements/__tests__/cumulative-day-sum.test.ts +++ b/src/lib/measurements/__tests__/cumulative-day-sum.test.ts @@ -6,8 +6,10 @@ import { CUMULATIVE_DAY_SUM_TYPES, cumulativeMetricKey, isCumulativeDaySumType, + metricKeyForType, pickCumulativeDaySum, } from "../cumulative-day-sum"; +import { RANKED_TYPES } from "@/lib/analytics/source-rank-sql"; /** * v1.4.36 W4c — pickCumulativeDaySum unit tests. @@ -156,3 +158,22 @@ describe("cumulativeMetricKey", () => { expect(cumulativeMetricKey("TIME_IN_DAYLIGHT")).toBeNull(); }); }); + +describe("metricKeyForType — source-priority ladder coverage (v1.18.10 I-5)", () => { + // Every type in RANKED_TYPES drives the SQL source-collapse CASE + // (`buildSourceRankCase` skips a type whose `metricKeyForType` is null). + // A ranked type that resolves to null would make `collapseRollupRowsBySource` + // fall to the alphabetical rank-90 tiebreak — nondeterministic the instant a + // second producer for that type is enabled. Pin the contract so a future + // ranked type cannot ship without its metric key. + it("resolves every RANKED_TYPES member to a non-null SourcePriorityMetricKey", () => { + for (const type of RANKED_TYPES) { + expect(metricKeyForType(type)).not.toBeNull(); + } + }); + + it("ladders STRESS_SCORE so the COMPUTED-vs-device producer is deterministic", () => { + expect(metricKeyForType("STRESS_SCORE")).toBe("stress"); + expect(RANKED_TYPES).toContain("STRESS_SCORE"); + }); +}); diff --git a/src/lib/measurements/cumulative-day-sum.ts b/src/lib/measurements/cumulative-day-sum.ts index e5c41947f..c4c55b1d1 100644 --- a/src/lib/measurements/cumulative-day-sum.ts +++ b/src/lib/measurements/cumulative-day-sum.ts @@ -172,6 +172,12 @@ export function metricKeyForType( return "vo2Max"; case "RECOVERY_SCORE": return "recovery"; + case "STRESS_SCORE": + // v1.18.10 I-5 — ladder the COMPUTED-vs-device stress producer so the + // rollup source collapse is deterministic the instant a second producer + // (Oura `daily_stress`) is enabled, rather than falling to the + // alphabetical rank-90 tiebreak in `collapseRollupRowsBySource`. + return "stress"; case "SLEEP_DURATION": return "sleep"; default: diff --git a/src/lib/validations/source-priority.ts b/src/lib/validations/source-priority.ts index 128516207..b54781693 100644 --- a/src/lib/validations/source-priority.ts +++ b/src/lib/validations/source-priority.ts @@ -131,6 +131,13 @@ export const SOURCE_PRIORITY_METRIC_KEYS = [ // `RECOVERY_SCORE` rows distinguished by source — this ladder lets the // same picker resolve native-above-proxy without a second engine. "recovery", + // v1.18.10 I-5 — native-vs-derived stress. HealthLog computes a COMPUTED + // HRV-derived `STRESS_SCORE` proxy; a device-native producer (Oura + // `daily_stress`) is the deferred second source. Laddering it now makes the + // picker deterministic the instant a second producer is enabled, instead of + // collapsing to the alphabetical rank-90 tiebreak. Single-producer today, + // so the ladder is a no-op until the second producer lands. + "stress", ] as const; export type SourcePriorityMetricKey = @@ -297,6 +304,11 @@ export const DEFAULT_SOURCE_PRIORITY: Required = { // fallback for users without a strap. v1.17.0 — Oura readiness + Polar // nightly recovery slot below WHOOP and above the computed proxy. recovery: ["WHOOP", "OURA", "POLAR", "COMPUTED"], + // v1.18.10 I-5 — native-vs-derived stress. Oura's device-native + // `daily_stress` (the deferred second producer) outranks HealthLog's + // COMPUTED HRV-derived proxy; the proxy is the fallback for users without a + // device-native stress signal. Single-producer today. + stress: ["OURA", "COMPUTED"], }; /** From 3fb1f3989c3f537df2ef18eaf0c06671e502d8ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 20 Jun 2026 16:52:19 +0200 Subject: [PATCH 4/4] perf(measurements): cover soft-delete reads and cap the all-time live scan Add a covering partial index over (user_id, type, measured_at DESC) WHERE deleted_at IS NULL (migration 0183). The hot reads all filter deleted_at and order measured_at DESC, but the base index is ASC and excludes deleted_at, so every type not served by the two existing partial indexes pays a heap re-check of deleted_at over its partition. The covering partial removes the re-check for the DISTINCT-ON-latest, list and series reads. Cap the cold-fallback all-time aggregate in summaries-slice at a 15-year window. It sits far beyond any real history, so it never changes a real all-time count/min/max/mean; it exists only so a coverage-miss cold read can't trigger an unbounded full-partition scan. --- .../migration.sql | 40 +++++++++++++++++++ prisma/schema.prisma | 6 +++ .../__tests__/summaries-slice.test.ts | 9 +++-- src/lib/analytics/summaries-slice.ts | 25 +++++++++--- 4 files changed, 72 insertions(+), 8 deletions(-) create mode 100644 prisma/migrations/0183_v11810_measurements_live_covering_idx/migration.sql diff --git a/prisma/migrations/0183_v11810_measurements_live_covering_idx/migration.sql b/prisma/migrations/0183_v11810_measurements_live_covering_idx/migration.sql new file mode 100644 index 000000000..201c3c841 --- /dev/null +++ b/prisma/migrations/0183_v11810_measurements_live_covering_idx/migration.sql @@ -0,0 +1,40 @@ +-- v1.18.10 P-5 — covering partial index for soft-delete-filtered reads. +-- +-- The hot reads on `measurements` all filter `deleted_at IS NULL` and order +-- `measured_at DESC` per type — the DISTINCT-ON-latest pass, the management +-- list, the series reads, and the per-type count. The base +-- `(user_id, type, measured_at)` index is ASC and does NOT carry +-- `deleted_at`, so those reads pay a heap re-check of `deleted_at` over the +-- type partition for every candidate row. +-- +-- Two partial indexes already cover the highest-volume types +-- (`measurements_dense_intraday_retention_idx` for HRV/PULSE, migration 0108; +-- `measurements_legacy_step_consolidation_idx` for ACTIVITY_STEPS, migration +-- 0087). This index closes the gap for every OTHER type: a covering partial +-- index keyed `(user_id, type, measured_at DESC)` over live rows only serves +-- the latest / list / series reads without the heap re-check, and — being +-- partial on `deleted_at IS NULL` — it carries no entries for tombstoned rows. +-- +-- Additive + order-safe: a pure `CREATE INDEX IF NOT EXISTS`, no column add, +-- no backfill, no rewrite of `measurements`. +-- +-- Prisma can't express a partial-predicate index, so it lives here in raw +-- migration SQL and is intentionally omitted from `schema.prisma` (the same +-- treatment migrations 0087 + 0108 use; a comment on the model documents it). +-- +-- Idempotent guard (`IF NOT EXISTS`) makes reruns safe. Forward-only. +-- +-- Lock note (conscious trade-off): a plain, non-`CONCURRENTLY` build, so it +-- runs inside Prisma's migration transaction and holds an ACCESS EXCLUSIVE +-- lock on `measurements` for the build. On the largest multi-year HealthKit +-- tenants the live row set is large, so deploy at a low-traffic window. A +-- `CONCURRENTLY` build cannot run inside the migration transaction and is not +-- used here to keep the migration atomic with the rest of the release. +-- +-- Reversibility: +-- DROP INDEX IF EXISTS "measurements_live_covering_idx"; +-- A roll-back loses only the scan acceleration; no data is touched. + +CREATE INDEX IF NOT EXISTS "measurements_live_covering_idx" + ON "measurements" ("user_id", "type", "measured_at" DESC) + WHERE "deleted_at" IS NULL; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 480c5eaa2..c571c1afd 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1113,6 +1113,12 @@ model Measurement { // — see migration 0108. It accelerates the v1.10.0 dense intra-day // retention drain's discovery + per-user scan. Same Prisma-can't- // express-a-partial-predicate reason it is omitted here. + // `measurements_live_covering_idx` is a covering partial index over + // `(user_id, type, measured_at DESC) WHERE deleted_at IS NULL` — see + // migration 0183 (v1.18.10 P-5). It serves the DISTINCT-ON-latest, list, + // and series reads for every type NOT covered by the two partials above + // and removes the `deleted_at` heap re-check. Same Prisma-can't-express-a- + // partial-predicate reason it is omitted here. @@map("measurements") } diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 8f4085299..00972918e 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -592,15 +592,18 @@ describe("computeSummariesSlice", () => { // 90-day cap to `canonicalMeasurementsFrom(rank, "90 days")`. The // helper lives in `@/lib/analytics/source-rank-sql` (stubbed // above), so pin the contract at the call boundary: the slice - // must ask for the 90-day window. The `allTime` aggregate - // deliberately calls it WITHOUT an interval (no cap — all-time - // count/min/max/mean must scan every row). + // must ask for the 90-day window. expect(canonicalMeasurementsFrom).toHaveBeenCalledWith( expect.any(String), "90 days", ); + // v1.18.10 P-7 — the `allTime` aggregate is now capped at a generous + // 15-year scan-DoS floor (far beyond any real history) so a + // coverage-miss cold read can't trigger an unbounded full-partition + // scan. It no longer calls the helper without an interval. expect(canonicalMeasurementsFrom).toHaveBeenCalledWith( expect.any(String), + "15 years", ); }); }); diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 2b4d481ba..695068180 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -81,11 +81,22 @@ import { resolveUserTimezone } from "@/lib/tz/resolver"; const DAY_MS = 24 * 60 * 60 * 1000; +/** + * v1.18.10 P-7 — upper bound on the cold-fallback `allTime` aggregate's + * scan window. Sits far beyond any real account's history (the earliest prod + * measurements are < 10 years old) so it never changes a real all-time + * count / min / max / mean; it exists purely so a coverage-miss cold read + * can't trigger an unbounded full-partition scan. Whitelisted by + * `canonicalMeasurementsFrom`'s interval regex (` years`). + */ +const ALL_TIME_LIVE_CAP_INTERVAL = "15 years" as const; + /** * v1.4.48 — all-time aggregate row used on the cold-mount fallback * path. Holds the linearly composable columns (`count / min / max / * mean`) that must reflect every row the user has ever logged for - * that type, so this query intentionally has no `measured_at` cap. + * that type, capped at `ALL_TIME_LIVE_CAP_INTERVAL` (v1.18.10 P-7) as a + * scan-DoS floor. */ interface AllTimeAggregateRow { type: string; @@ -608,9 +619,13 @@ async function computeFromRollups(userId: string): Promise { * v1.4.48 — split into two parallel queries: * * 1. `allTime` — the linearly composable columns (`count / min / - * max / mean`) keep the full-partition scan because they must - * reflect every row the user has ever logged. No `measured_at` - * cap. + * max / mean`). v1.18.10 P-7 caps this at `ALL_TIME_LIVE_CAP_INTERVAL` + * (15 years) so an un-windowed cold-fallback read on a coverage-miss + * tenant can never scan an unbounded partition. The cap is a scan-DoS + * floor, not a semantic change: it sits far beyond any real account's + * history (the earliest prod rows are < 10 years old) and the steady + * state is rollup-backed anyway, so it only ever bounds the cold path. + * * * 2. `windowed` — `avg7 / avg30` and the slope/r² tuples are all * already filtered to a 7/30/90 day window inside the SELECT. @@ -648,7 +663,7 @@ async function computeFromLiveAggregate( MIN(m."value")::double precision AS min_value, MAX(m."value")::double precision AS max_value, AVG(m."value")::double precision AS mean_value - FROM ${canonicalMeasurementsFrom(rankUnqualified)} + FROM ${canonicalMeasurementsFrom(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)} GROUP BY m."type" `, userId,