From 901e7d345dd6c030cc8d88bc050f4e8db68b2307 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 5 Jun 2026 03:18:23 +0200 Subject: [PATCH] fix(insights): resolve cross-source double-count and read truncation in mood crosstab MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tag × metric crosstab summed cumulative metrics (steps, active energy, sleep duration) per day across every source with no canonical pick, so once two sources reported the same day — Fitbit + Apple steps, Fitbit + WHOOP sleep — the per-day total double-counted and inflated the with/without averages and the Welch delta the surface ranks on. Run the metric rows through the same per-day source-priority + device-type picker the analytics steps/sleep path uses before bucketing, keyed by the metric's priority ladder and the user's Berlin calendar day. Same-source per-stage rows still sum into the night total; only cross-source twins collapse. Thread the user's source-priority blob through the read so the pick honours the configured ladder. The cross-metric read was a flat `take: 5000` over the 365-day window ordered `measuredAt desc`, unioned across per-stage sleep + per-sample pulse/HRV; a multi-source, high-frequency user blew past it and silently dropped the oldest months, biasing the crosstab and the long-window correlations toward recent data. Lift the cap to cover the realistic worst case for the window and annotate a wide event when the cap is hit so a truncated read is observable rather than silently wrong. The read stays raw rather than moving to the DAY rollup tier because that tier buckets on UTC midnight while these aggregates key on the user's Berlin day, which would skew the day pairing at the boundary. Co-Authored-By: Claude Opus 4.8 (1M context) (cherry picked from commit aa72ffba6e3bcac9f97be4adc26d689d4f352e98) --- .../__tests__/mood-aggregates.test.ts | 109 ++++++++++++ src/lib/insights/mood-aggregates.ts | 162 ++++++++++++++++-- .../mood-insights-crosstab.test.ts | 68 ++++++++ 3 files changed, 324 insertions(+), 15 deletions(-) diff --git a/src/lib/insights/__tests__/mood-aggregates.test.ts b/src/lib/insights/__tests__/mood-aggregates.test.ts index 31b5bd56a..47bf54902 100644 --- a/src/lib/insights/__tests__/mood-aggregates.test.ts +++ b/src/lib/insights/__tests__/mood-aggregates.test.ts @@ -984,4 +984,113 @@ describe("computeTagMetricCrosstab", () => { expect(CROSSTAB_MAX_ROWS).toBeGreaterThan(0); expect(CROSSTAB_MIN_ABSENT_DAYS).toBeGreaterThan(0); }); + + // v1.12.1 — cross-source double-count guard. A cumulative metric reported + // by two sources on the same day must be summed once (canonical source), + // not double-counted, or the with/without averages and the Welch delta + // are inflated. + function sourcedMeas( + offset: number, + type: string, + value: number, + source: CrossMetricMeasurement["source"], + ): CrossMetricMeasurement { + return { + type, + value, + measuredAt: new Date(NOW.getTime() - offset * dayMs), + source, + }; + } + + it("counts a cumulative metric once when two sources report the same day (Fitbit + Apple)", () => { + const entries: MoodAggregateEntry[] = []; + const measurements: CrossMetricMeasurement[] = []; + // 12 present days: BOTH Apple and Fitbit report ~600 kcal active energy. + // A naïve per-day sum would double this to ~1200; the canonical pick + // keeps only Apple (activeEnergy ladder: APPLE_HEALTH > … > FITBIT). + for (let i = 0; i < 12; i++) { + entries.push(structuredEntry(i, 4, [WORKOUT])); + const v = 600 + (i % 2 === 0 ? 20 : -20); + measurements.push(sourcedMeas(i, "ACTIVE_ENERGY_BURNED", v, "APPLE_HEALTH")); + measurements.push(sourcedMeas(i, "ACTIVE_ENERGY_BURNED", v, "FITBIT")); + } + // 12 absent days: single source ~350. + for (let i = 12; i < 24; i++) { + entries.push(structuredEntry(i, 3, [])); + measurements.push( + sourcedMeas(i, "ACTIVE_ENERGY_BURNED", 350 + (i % 2 === 0 ? 20 : -20), "APPLE_HEALTH"), + ); + } + const rows = computeTagMetricCrosstab({ entries, measurements, now: NOW }); + const row = rows.find((r) => r.tag === "worked_out" && r.metricKey === "activeEnergy"); + expect(row).toBeDefined(); + // ~600, NOT ~1200 — the Fitbit twin is dropped by the canonical pick. + expect(row!.withAvg).toBeCloseTo(600, 0); + expect(row!.withoutAvg).toBeCloseTo(350, 0); + expect(row!.delta).toBeCloseTo(250, 0); + }); + + it("counts sleep once when WHOOP and Fitbit both report the same night", () => { + const SLEEP: StructuredTagRef = { + key: "slept_well", + categoryKey: "health", + labelKey: "mood.tag.sleptWell", + icon: "Moon", + }; + const entries: MoodAggregateEntry[] = []; + const measurements: CrossMetricMeasurement[] = []; + // Present nights: WHOOP + Fitbit both report 480 min (8 h). Sleep ladder + // is WHOOP > FITBIT > … so WHOOP wins; the night total stays 8 h. + for (let i = 0; i < 12; i++) { + entries.push(structuredEntry(i, 4, [SLEEP])); + const v = 480 + (i % 2 === 0 ? 6 : -6); + measurements.push(sourcedMeas(i, "SLEEP_DURATION", v, "WHOOP")); + measurements.push(sourcedMeas(i, "SLEEP_DURATION", v, "FITBIT")); + } + for (let i = 12; i < 24; i++) { + entries.push(structuredEntry(i, 3, [])); + measurements.push(sourcedMeas(i, "SLEEP_DURATION", 360 + (i % 2 === 0 ? 6 : -6), "WHOOP")); + } + const rows = computeTagMetricCrosstab({ entries, measurements, now: NOW }); + const row = rows.find((r) => r.metricKey === "sleepDuration"); + expect(row).toBeDefined(); + // 8 h, not 16 h — the Fitbit twin is dropped, not summed on top. + expect(row!.withAvg).toBeCloseTo(8, 1); + expect(row!.withoutAvg).toBeCloseTo(6, 1); + }); + + it("sums per-stage rows from the SAME source into the night total (no over-collapse)", () => { + const SLEEP: StructuredTagRef = { + key: "slept_well", + categoryKey: "health", + labelKey: "mood.tag.sleptWell", + icon: "Moon", + }; + const entries: MoodAggregateEntry[] = []; + const measurements: CrossMetricMeasurement[] = []; + // Present nights: WHOOP reports the night as 3 stage rows (160 each = + // 480 min = 8 h). The canonical pick must keep ALL three same-source + // stage rows so the sum is the night total — it only drops cross-source + // duplicates, never same-source stages. + for (let i = 0; i < 12; i++) { + entries.push(structuredEntry(i, 4, [SLEEP])); + const j = i % 2 === 0 ? 2 : -2; + measurements.push(sourcedMeas(i, "SLEEP_DURATION", 160 + j, "WHOOP")); + measurements.push(sourcedMeas(i, "SLEEP_DURATION", 160 + j, "WHOOP")); + measurements.push(sourcedMeas(i, "SLEEP_DURATION", 160 + j, "WHOOP")); + } + for (let i = 12; i < 24; i++) { + entries.push(structuredEntry(i, 3, [])); + const j = i % 2 === 0 ? 2 : -2; + measurements.push(sourcedMeas(i, "SLEEP_DURATION", 120 + j, "WHOOP")); + measurements.push(sourcedMeas(i, "SLEEP_DURATION", 120 + j, "WHOOP")); + measurements.push(sourcedMeas(i, "SLEEP_DURATION", 120 + j, "WHOOP")); + } + const rows = computeTagMetricCrosstab({ entries, measurements, now: NOW }); + const row = rows.find((r) => r.metricKey === "sleepDuration"); + expect(row).toBeDefined(); + expect(row!.withAvg).toBeCloseTo(8, 1); // 3 × 160 = 480 min = 8 h + expect(row!.withoutAvg).toBeCloseTo(6, 1); // 3 × 120 = 360 min = 6 h + }); }); diff --git a/src/lib/insights/mood-aggregates.ts b/src/lib/insights/mood-aggregates.ts index d35e6bb37..3570fe17c 100644 --- a/src/lib/insights/mood-aggregates.ts +++ b/src/lib/insights/mood-aggregates.ts @@ -33,7 +33,12 @@ import { import { welchTTest } from "@/lib/insights/correlations"; import { benjaminiHochberg as fdrAdjust } from "@/lib/insights/correlation-discovery"; import { toBerlinYmd } from "@/lib/tz/resolver"; +import { annotate } from "@/lib/logging/context"; +import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority"; +import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import type { MeasurementType } from "@/generated/prisma/enums"; +import type { MeasurementSource } from "@/generated/prisma/client"; const MS_PER_DAY = 24 * 60 * 60 * 1000; @@ -106,13 +111,40 @@ export interface MoodAggregateEntry { structuredTags?: StructuredTagRef[]; } -/** Cross-metric row shape (WEIGHT / BLOOD_PRESSURE_SYS / PULSE / …). */ +/** + * Cross-metric row shape (WEIGHT / BLOOD_PRESSURE_SYS / PULSE / …). + * + * `source` / `deviceType` are optional so the correlation fixtures and + * legacy single-source tests keep their narrow shape, but when present + * they feed the crosstab's canonical-source pick: a cumulative metric + * (steps / active energy / sleep) that two sources both report for the + * same day must resolve to ONE source before summing, or the per-day + * total double-counts (see `metricDayMap`). The picker keys off + * `source` (the ladder axis) and `deviceType` (the watch>phone>scale + * axis) exactly like the analytics steps/sleep path. + */ export interface CrossMetricMeasurement { type: string; value: number; measuredAt: Date; + source?: MeasurementSource | null; + deviceType?: string | null; } +/** + * v1.12.1 — cap on the cross-metric measurement read in + * `fetchMoodAggregates`. Sized to keep a full data-rich 365-day window + * intact rather than the old flat 5,000 which silently dropped the + * oldest months (the read is `measuredAt desc`) once a multi-source, + * high-frequency user (per-stage sleep + per-sample pulse/HRV across + * Apple + Fitbit + WHOOP) blew past it. The crosstab/correlation only + * need per-day values, but the read stays raw because the DAY rollup + * buckets on UTC midnight while these aggregates key on the user's + * Berlin calendar day — see the read comment. Exported so the read's + * `take` and the truncation-annotation threshold share one constant. + */ +export const MOOD_CROSS_METRIC_ROW_CAP = 50_000; + // --- Target bands (mirror mood-chart VALUE_BANDS + mood-status) --- /** @@ -795,6 +827,18 @@ export interface TagMetricCrosstabRow { confidence: InfluenceConfidence; } +const CROSSTAB_SUM_TYPES = new Set([ + "ACTIVE_ENERGY_BURNED", + "SLEEP_DURATION", + "ACTIVITY_STEPS", +]); + +/** Berlin-calendar day key (`YYYY-MM-DD`) for a row's `measuredAt`. */ +function berlinDayKey(measuredAt: Date): string { + const { year, month, day } = toBerlinYmd(measuredAt); + return `${year}-${month}-${day}`; +} + /** * Build a Berlin-day-keyed metric map with the right aggregation. Energy * and step totals are SUMMED per day (HealthKit `stats:` rows are already @@ -802,22 +846,53 @@ export interface TagMetricCrosstabRow { * is SUMMED across per-stage rows to get the night total; everything else * (a once-daily score) is MEANED. Returns minutes/kcal/raw — the display * conversion happens at row-build time. + * + * Cross-source de-dup: before bucketing, the rows for this metric run + * through the SAME canonical-source picker the analytics steps/sleep path + * uses (`pickCanonicalSourceRows`, keyed by `metricKeyForType` + the + * Berlin day key). Without it, the moment two sources report the same day + * (Fitbit + Apple steps, Fitbit + WHOOP sleep) the SUM channels would + * double-count and bias the Welch delta. The picker collapses each day to + * one source (and one device-type within it), so the sum reflects one + * stream. MEAN channels (a once-daily score like recovery) gain the same + * single-source guarantee for free. Rows without a `source` (the legacy + * test fixtures, or a metric whose source isn't in the ladder) fall + * through the picker's pass-through branch unchanged. */ function metricDayMap( measurements: CrossMetricMeasurement[], type: string, + userPriorityJson: unknown, ): Map { - const SUM_TYPES = new Set([ - "ACTIVE_ENERGY_BURNED", - "SLEEP_DURATION", - "ACTIVITY_STEPS", - ]); - const summed = SUM_TYPES.has(type); + const summed = CROSSTAB_SUM_TYPES.has(type); + + const typeRows = measurements.filter((m) => m.type === type); + if (typeRows.length === 0) return new Map(); + + // Resolve the canonical source per day. `metricKeyForType` maps the + // crosstab's MeasurementType to its priority ladder; a metric with no + // ladder (or rows without a source) keeps every row via the picker's + // documented pass-through fallback, so behaviour is identical to the + // pre-fix sum for single-source data. + const metricKey = metricKeyForType(type as MeasurementType); + const canonicalRows = metricKey + ? pickCanonicalSourceRows( + typeRows.map((m) => ({ + measuredAt: m.measuredAt, + source: (m.source ?? "MANUAL") as MeasurementSource, + deviceType: m.deviceType ?? null, + type: type as MeasurementType, + value: m.value, + })), + metricKey, + userPriorityJson, + berlinDayKey, + ).canonicalRows + : typeRows; + const byDay = new Map(); - for (const m of measurements) { - if (m.type !== type) continue; - const { year, month, day } = toBerlinYmd(m.measuredAt); - const key = `${year}-${month}-${day}`; + for (const m of canonicalRows) { + const key = berlinDayKey(m.measuredAt); const cur = byDay.get(key) ?? { sum: 0, count: 0 }; cur.sum += m.value; cur.count += 1; @@ -867,9 +942,16 @@ export function computeTagMetricCrosstab(args: { measurements: CrossMetricMeasurement[]; now: Date; windowDays?: number; + /** + * The user's source-priority blob. Threaded into `metricDayMap` so the + * per-day canonical-source pick honours the user's ladder. `null` (the + * test default) resolves to the default ladders. + */ + userPriorityJson?: unknown; }): TagMetricCrosstabRow[] { const { entries, measurements, now } = args; const windowDays = args.windowDays ?? 365; + const userPriorityJson = args.userPriorityJson ?? null; const dayMap = collapseToTaggedDays(entries, now, windowDays); if (dayMap.size === 0) return []; @@ -888,7 +970,7 @@ export function computeTagMetricCrosstab(args: { for (const [metricKey, cfg] of Object.entries(CROSSTAB_METRICS) as Array< [CrosstabMetricKey, (typeof CROSSTAB_METRICS)[CrosstabMetricKey]] >) { - const metricByDay = metricDayMap(measurements, cfg.type); + const metricByDay = metricDayMap(measurements, cfg.type, userPriorityJson); if (metricByDay.size === 0) continue; for (const [tagKey, ref] of structuredMeta) { @@ -1422,8 +1504,15 @@ export function computeMoodAggregates(args: { entries: MoodAggregateEntry[]; measurements: CrossMetricMeasurement[]; now: Date; + /** + * The user's source-priority blob, threaded into the crosstab's + * per-day canonical-source pick. `null` (the test default) resolves to + * the default ladders. + */ + userPriorityJson?: unknown; }): MoodAggregates { const { entries, measurements, now } = args; + const userPriorityJson = args.userPriorityJson ?? null; const moodPoints = entries.map((entry) => ({ measuredAt: entry.moodLoggedAt, @@ -1481,6 +1570,7 @@ export function computeMoodAggregates(args: { entries, measurements, now, + userPriorityJson, }); // Distinct day keys the user logged on — drives the streak takeaway. @@ -1606,10 +1696,52 @@ export async function fetchMoodAggregates( }, }, orderBy: { measuredAt: "desc" }, - take: 5000, - select: { type: true, value: true, measuredAt: true }, + // v1.12.1 — scope the cap to the worst-case row count over the + // 365-day window instead of the old flat 5,000. The crosstab + + // correlation channels are per-day aggregates, but the read is raw + // because the DAY rollup buckets on UTC midnight while the crosstab + // / correlations key on the user's Berlin calendar day — feeding a + // UTC-bucketed metric series against a Berlin-bucketed mood series + // would skew the day pairing by up to a few hours at the boundary. + // So we stay on raw rows but lift the cap above the realistic worst + // case: SLEEP_DURATION (~6 stage rows/night) and PULSE / HRV ingest + // the most rows/day, and a user syncing several sources multiplies + // that — a flat 5,000 silently dropped the oldest months (the order + // is `measuredAt desc`). The cap is the floor that keeps a full + // data-rich year intact; if it is ever hit we annotate `truncated` + // so the drop is observable rather than silently recency-biased. + take: MOOD_CROSS_METRIC_ROW_CAP, + select: { + type: true, + value: true, + measuredAt: true, + // v1.12.1 — source + deviceType drive the crosstab's canonical + // per-day pick so a cumulative metric reported by two sources on + // the same day (Fitbit + Apple steps, Fitbit + WHOOP sleep) is + // summed once, not double-counted. + source: true, + deviceType: true, + }, }) .then((rows) => rows.reverse()); - return computeMoodAggregates({ entries, measurements, now }); + if (measurements.length >= MOOD_CROSS_METRIC_ROW_CAP) { + // The read hit the cap — the oldest rows in the 365-day window were + // dropped (order is `measuredAt desc`). Surface it on the wide event + // so a silently-truncated, recency-biased crosstab/correlation is at + // least observable instead of looking like clean data. + annotate({ + action: { name: "mood.insights.cross_metric.truncated" }, + meta: { rows: measurements.length, cap: MOOD_CROSS_METRIC_ROW_CAP }, + }); + } + + const userPriorityJson = await loadUserSourcePriority(userId); + + return computeMoodAggregates({ + entries, + measurements, + now, + userPriorityJson, + }); } diff --git a/tests/integration/mood-insights-crosstab.test.ts b/tests/integration/mood-insights-crosstab.test.ts index 90b7ce69d..df1e9860d 100644 --- a/tests/integration/mood-insights-crosstab.test.ts +++ b/tests/integration/mood-insights-crosstab.test.ts @@ -100,4 +100,72 @@ describe("mood-insights tag × metric crosstab (DB read path)", () => { expect(row!.confidence).toBe("high"); expect(row!.qValue).toBeLessThanOrEqual(0.1); }); + + // v1.12.1 — cross-source double-count guard over the real read path. Two + // sources (Apple + Fitbit) report the same active-energy total each present + // day; the read's canonical-source pick must keep one (Apple, per the + // activeEnergy ladder) so the per-day sum is ~600, not ~1200. + it("counts active energy once when Apple + Fitbit report the same day", async () => { + const prisma = getPrismaClient(); + const tag = await prisma.moodTag.findFirst({ + include: { category: true }, + orderBy: { key: "asc" }, + }); + expect(tag).toBeTruthy(); + + for (let i = 0; i < 24; i++) { + const present = i < 12; + const ts = new Date(NOW.getTime() - i * dayMs); + await prisma.moodEntry.create({ + data: { + userId: TEST_USER_ID, + date: dayKey(i), + mood: "GUT", + score: present ? 4 : 3, + source: "WEB", + moodLoggedAt: ts, + ...(present ? { tagLinks: { create: [{ moodTagId: tag!.id }] } } : {}), + }, + }); + const value = present + ? 600 + (i % 2 === 0 ? 20 : -20) + : 350 + (i % 2 === 0 ? 20 : -20); + // Apple stream every day. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "ACTIVE_ENERGY_BURNED", + value, + unit: "kcal", + source: "APPLE_HEALTH", + measuredAt: ts, + externalId: `apple:${dayKey(i)}`, + }, + }); + // Fitbit twin on the present days — must NOT add on top of Apple's total. + if (present) { + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "ACTIVE_ENERGY_BURNED", + value, + unit: "kcal", + source: "FITBIT", + measuredAt: ts, + externalId: `fitbit:${dayKey(i)}`, + }, + }); + } + } + + const aggregates = await fetchMoodAggregates(TEST_USER_ID, NOW); + const row = aggregates.tagMetricCrosstab.find( + (r) => r.tag === tag!.key && r.metricKey === "activeEnergy", + ); + expect(row).toBeDefined(); + // ~600 (Apple wins), NOT ~1200 (Apple + Fitbit double-counted). + expect(row!.withAvg).toBeCloseTo(600, 0); + expect(row!.withoutAvg).toBeCloseTo(350, 0); + expect(row!.delta).toBeCloseTo(250, 0); + }); });