From cddade3b4f0af7c8813d70612a9255241008178a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 4 Jun 2026 09:33:30 +0200 Subject: [PATCH] feat(rollups): add source-priority collapse helper for read path collapseRollupRowsBySource resolves the per-source rollup rows minted by the source-aware writer down to one row per bucket, using the user's source-priority ladder (the same ordering the raw-row picker uses). A metricKeyForType superset maps every overlapping non-cumulative vital (RHR, HRV-SDNN, SpO2, body/skin temperature, respiratory rate, weight, body fat, blood pressure, pulse, VO2max, recovery) plus the cumulative keys onto a SourcePriorityMetricKey. Unlisted sources fall back to the highest-count row so a bucket never doubles or goes dark. Wired into the rollup readers in the following commits. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/lib/measurements/cumulative-day-sum.ts | 47 ++++++ .../collapse-rollup-rows-by-source.test.ts | 157 ++++++++++++++++++ src/lib/rollups/measurement-read.ts | 85 ++++++++++ 3 files changed, 289 insertions(+) create mode 100644 src/lib/rollups/__tests__/collapse-rollup-rows-by-source.test.ts diff --git a/src/lib/measurements/cumulative-day-sum.ts b/src/lib/measurements/cumulative-day-sum.ts index 3c09c056c..c67eea962 100644 --- a/src/lib/measurements/cumulative-day-sum.ts +++ b/src/lib/measurements/cumulative-day-sum.ts @@ -131,3 +131,50 @@ export function cumulativeMetricKey( return null; } } + +/** + * v1.11.1 — full MeasurementType → SourcePriorityMetricKey map for the + * source-aware rollup collapse. A superset of `cumulativeMetricKey`: it adds + * the overlapping NON-cumulative vitals (spot / daily readings that two or + * more sources realistically report for the same day) so the rollup read path + * can resolve the canonical source through the same ladder the raw-row picker + * uses. Cumulative types fall through to `cumulativeMetricKey`. Returns null + * for single-source types (no competing source today → source-blind grouping + * is already correct) and for types without a ladder — the collapse treats a + * null key as "no priority, keep one row deterministically". + */ +export function metricKeyForType( + type: MeasurementType, +): SourcePriorityMetricKey | null { + switch (type) { + case "RESTING_HEART_RATE": + return "restingHeartRate"; + case "HEART_RATE_VARIABILITY": + return "hrv"; + case "RESPIRATORY_RATE": + return "respiratoryRate"; + case "OXYGEN_SATURATION": + return "spo2"; + case "BODY_TEMPERATURE": + return "bodyTemperature"; + case "SKIN_TEMPERATURE": + return "skinTemperature"; + case "WEIGHT": + return "weight"; + case "BODY_FAT": + return "bodyFat"; + case "BLOOD_PRESSURE_SYS": + case "BLOOD_PRESSURE_DIA": + return "bloodPressure"; + case "PULSE": + return "pulse"; + case "VO2_MAX": + return "vo2Max"; + case "RECOVERY_SCORE": + return "recovery"; + case "SLEEP_DURATION": + return "sleep"; + default: + return cumulativeMetricKey(type); + } +} diff --git a/src/lib/rollups/__tests__/collapse-rollup-rows-by-source.test.ts b/src/lib/rollups/__tests__/collapse-rollup-rows-by-source.test.ts new file mode 100644 index 000000000..cd37fb4c0 --- /dev/null +++ b/src/lib/rollups/__tests__/collapse-rollup-rows-by-source.test.ts @@ -0,0 +1,157 @@ +/** + * v1.11.1 — source-aware rollup collapse. + * + * The writer mints one rollup row per (type, day, source). This helper + * resolves overlapping sources to ONE row per bucket via the user's + * source-priority ladder, so a day with WHOOP + Apple Watch resting heart + * rate surfaces the ladder-canonical reading instead of an AVG blend. + * + * Default ladders pinned here (from `DEFAULT_SOURCE_PRIORITY`): + * restingHeartRate: WHOOP > APPLE_HEALTH > WITHINGS + * spo2: WITHINGS > WHOOP > APPLE_HEALTH > MANUAL + * recovery: WHOOP > COMPUTED + * steps: APPLE_HEALTH > WITHINGS > MANUAL + */ +import { describe, expect, it } from "vitest"; + +import type { MeasurementSource } from "@/generated/prisma/client"; +import { collapseRollupRowsBySource } from "../measurement-read"; + +interface Row { + bucketStart: Date; + source: MeasurementSource; + count: number; + mean: number; + sumValue: number | null; +} + +const DAY1 = new Date("2026-06-01T00:00:00.000Z"); +const DAY2 = new Date("2026-06-02T00:00:00.000Z"); + +function row( + bucketStart: Date, + source: MeasurementSource, + mean: number, + count = 1, + sumValue: number | null = null, +): Row { + return { bucketStart, source, count, mean, sumValue }; +} + +describe("collapseRollupRowsBySource", () => { + it("picks WHOOP over Apple for resting heart rate (ladder head wins)", () => { + const out = collapseRollupRowsBySource( + [ + row(DAY1, "APPLE_HEALTH", 54), + row(DAY1, "WHOOP", 51), + ], + "RESTING_HEART_RATE", + null, + ); + expect(out).toHaveLength(1); + expect(out[0].source).toBe("WHOOP"); + expect(out[0].mean).toBe(51); + }); + + it("picks Withings over Apple for SpO₂", () => { + const out = collapseRollupRowsBySource( + [ + row(DAY1, "APPLE_HEALTH", 96), + row(DAY1, "WITHINGS", 98), + ], + "OXYGEN_SATURATION", + null, + ); + expect(out).toHaveLength(1); + expect(out[0].source).toBe("WITHINGS"); + expect(out[0].mean).toBe(98); + }); + + it("picks native WHOOP over the COMPUTED proxy for RECOVERY_SCORE", () => { + const out = collapseRollupRowsBySource( + [ + row(DAY1, "COMPUTED", 62), + row(DAY1, "WHOOP", 70), + ], + "RECOVERY_SCORE", + null, + ); + expect(out).toHaveLength(1); + expect(out[0].source).toBe("WHOOP"); + }); + + it("collapses cumulative steps to ONE source's sum (never cross-source sum)", () => { + const out = collapseRollupRowsBySource( + [ + row(DAY1, "WITHINGS", 3800, 1, 3800), + row(DAY1, "APPLE_HEALTH", 4000, 1, 4000), + ], + "ACTIVITY_STEPS", + null, + ); + expect(out).toHaveLength(1); + // steps ladder = APPLE_HEALTH first → canonical source's sum only, NOT 7800. + expect(out[0].source).toBe("APPLE_HEALTH"); + expect(out[0].sumValue).toBe(4000); + }); + + it("preserves multi-day series, collapsing each day independently", () => { + const out = collapseRollupRowsBySource( + [ + row(DAY1, "APPLE_HEALTH", 54), + row(DAY1, "WHOOP", 51), + row(DAY2, "WHOOP", 49), + ], + "RESTING_HEART_RATE", + null, + ); + expect(out).toHaveLength(2); + expect(out.map((r) => r.bucketStart.getTime())).toEqual([ + DAY1.getTime(), + DAY2.getTime(), + ]); + expect(out.every((r) => r.source === "WHOOP")).toBe(true); + }); + + it("leaves a single-source day unchanged (fast path)", () => { + const input = [row(DAY1, "WITHINGS", 72)]; + const out = collapseRollupRowsBySource(input, "WEIGHT", null); + expect(out).toBe(input); + }); + + it("honours a custom source-priority override", () => { + // User pins Apple above WHOOP for resting heart rate. + const custom = { + restingHeartRate: ["APPLE_HEALTH", "WHOOP", "WITHINGS"], + }; + const out = collapseRollupRowsBySource( + [ + row(DAY1, "WHOOP", 51), + row(DAY1, "APPLE_HEALTH", 54), + ], + "RESTING_HEART_RATE", + custom, + ); + expect(out).toHaveLength(1); + expect(out[0].source).toBe("APPLE_HEALTH"); + }); + + it("falls back to max-count when no ladder source is present", () => { + // IMPORT is not on the restingHeartRate ladder; keep one row (most data). + const out = collapseRollupRowsBySource( + [ + row(DAY1, "IMPORT", 60, 2), + row(DAY1, "MANUAL", 58, 5), + ], + "RESTING_HEART_RATE", + null, + ); + expect(out).toHaveLength(1); + expect(out[0].source).toBe("MANUAL"); + expect(out[0].count).toBe(5); + }); + + it("returns empty input untouched", () => { + expect(collapseRollupRowsBySource([], "WEIGHT", null)).toEqual([]); + }); +}); diff --git a/src/lib/rollups/measurement-read.ts b/src/lib/rollups/measurement-read.ts index 557c20c74..e03f469b3 100644 --- a/src/lib/rollups/measurement-read.ts +++ b/src/lib/rollups/measurement-read.ts @@ -20,6 +20,16 @@ * the v1.4.34.1 / 4.5 aggregator survives. */ +import type { + MeasurementSource, + MeasurementType, +} from "@/generated/prisma/client"; +import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; +import { + getSourceLadder, + parseSourcePriority, +} from "@/lib/validations/source-priority"; + export interface DailyMeanRow { day: Date; count: number; @@ -28,6 +38,81 @@ export interface DailyMeanRow { maxValue: number; } +/** Minimal shape the source collapse needs from a per-source rollup row. */ +export interface SourcedBucketRow { + bucketStart: Date; + source: MeasurementSource; + count: number; +} + +/** + * v1.11.1 — collapse per-source rollup rows to ONE row per bucket using the + * user's source-priority ladder. The writer mints one row per + * (type, day, source); this resolves overlapping sources (e.g. WHOOP + Apple + * Watch resting heart rate) to the ladder-canonical reading before the linear + * composition in `aggregateBuckets` / the WMY readers runs. Cumulative types + * collapse to the single canonical source too, so the caller reads that one + * source's summed `sumValue` — a day's total reflects one source, never a + * cross-source sum. + * + * Resolution per bucket: + * 1. the first source in the metric's ladder that is present → canonical; + * 2. no ladder match (an unlisted source, or a type with no ladder) → the + * row with the most readings (max `count`), tie-broken by source name, + * so the bucket neither doubles nor goes dark. + * + * Input order is preserved (buckets emit in first-seen order). A single-source + * day (the common case) short-circuits to the row unchanged. + */ +export function collapseRollupRowsBySource( + rows: T[], + type: MeasurementType, + userPriorityJson: unknown, +): T[] { + if (rows.length <= 1) return rows; + + const byBucket = new Map(); + for (const row of rows) { + const key = row.bucketStart.getTime(); + const slot = byBucket.get(key); + if (slot) slot.push(row); + else byBucket.set(key, [row]); + } + + const metricKey = metricKeyForType(type); + const ladder: readonly MeasurementSource[] = metricKey + ? getSourceLadder(parseSourcePriority(userPriorityJson), metricKey) + : []; + + const out: T[] = []; + for (const bucketRows of byBucket.values()) { + if (bucketRows.length === 1) { + out.push(bucketRows[0]); + continue; + } + let picked: T | undefined; + for (const source of ladder) { + const hit = bucketRows.find((r) => r.source === source); + if (hit) { + picked = hit; + break; + } + } + if (!picked) { + // No ladder match — keep one row deterministically (most data wins, + // tie-broken by source name) so the bucket neither doubles nor goes dark. + picked = bucketRows.reduce((best, r) => + r.count > best.count || + (r.count === best.count && r.source < best.source) + ? r + : best, + ); + } + out.push(picked); + } + return out; +} + /** * Combine DAY buckets into the linearly-composable window stats — * `count`, `min`, `max`, `mean`. SD / slope / R² are intentionally