From 8390b4d24e58936b3f6410192c6ef3d8c0fcf3ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 13 Aug 2026 22:00:27 +0200 Subject: [PATCH 1/4] Serve correlation-discovery spot channels from the DAY rollup tier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The discovery scan read every raw measurement in the 180-day window for all nine measurement channels and bucketed them into daily means in JS, paying tens of thousands of rows on every cache miss (a CGM glucose channel alone can exceed the read cap). The DAY rollup tier already stores those per-day aggregates, so eligible spot channels now read one pre-aggregated row per day per source and compose the all-source count-weighted mean — the exact reduction toDailyMeans applies. Read-swap replaces the raw read per channel, with fallback-on-miss: structural ineligibility (SLEEP_DURATION's per-night reconstruction, the cumulative types' device-type collapse axis the rollup rows do not carry), a far-from-UTC profile timezone (the v1.4.38 W-A near-UTC guard), a coverage-probe miss, or an empty in-window bucket read all route that channel to the existing raw path. The SWR cache cell stays in front; the swap only lowers the cost of a cache miss. --- src/app/api/insights/correlations/route.ts | 40 ++-- .../insights/correlation-channel-series.ts | 204 ++++++++++++++++++ 2 files changed, 221 insertions(+), 23 deletions(-) diff --git a/src/app/api/insights/correlations/route.ts b/src/app/api/insights/correlations/route.ts index 1dd15e271..bde103c8f 100644 --- a/src/app/api/insights/correlations/route.ts +++ b/src/app/api/insights/correlations/route.ts @@ -45,16 +45,14 @@ import { syncAcceptedPatterns, } from "@/lib/insights/correlation-patterns"; import { - buildMeasurementDailySeries, fetchComplianceSeries, fetchCustomMetricBehaviourSeries, fetchEnvironmentSeries, fetchLabDraws, - fetchMeasurementWindowSeries, + fetchMeasurementDailySeriesTiered, fetchMoodWindowSeries, fetchSymptomSeries, } from "@/lib/insights/correlation-channel-series"; -import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; export const dynamic = "force-dynamic"; @@ -139,23 +137,21 @@ async function buildCorrelationsResponse( DISCOVERY_OUTCOMES, ) as MeasurementType[]; - // v1.30.3 (QA F1) — the fetch + desc/cap/resort discipline now lives in - // `fetchMeasurementWindowSeries` / `fetchMoodWindowSeries` - // (`correlation-channel-series.ts`), shared with the Coach tool, the - // per-metric card, and the period narrative so a fourth independently- - // maintained copy can't drift the way the period-narrative one did. - const [ - { byType: measurementsByType, measurementsCapped }, - moodWindow, - priorityJson, - ] = await Promise.all([ - fetchMeasurementWindowSeries(userId, since, [ + // Rollup read-swap for the measurement channels: eligible spot channels + // read their per-day means from the DAY rollup tier, with per-channel + // fallback to the raw `fetchMeasurementWindowSeries` path on any miss + // (coverage gap, far-from-UTC profile tz, sleep / cumulative grain) — + // see `fetchMeasurementDailySeriesTiered` for the full parity contract. + // The SWR cache cell above stays in front of this read: the swap only + // lowers the cost of a cache MISS, it does not replace the cache. + const [measurementDaily, moodWindow] = await Promise.all([ + fetchMeasurementDailySeriesTiered(userId, tz, since, [ ...behaviourTypes, ...outcomeTypes, ]), fetchMoodWindowSeries(userId, tz, since), - loadUserSourcePriority(userId), ]); + const { measurementsCapped } = measurementDaily; const { moodDaily, moodCapped } = moodWindow; // v1.21.0 (FDREXTEND) — build the two non-measurement, non-mood channels from @@ -180,14 +176,7 @@ async function buildCorrelationsResponse( ]); const points = (key: string): DailySeriesPoint[] => - key === "MOOD" - ? moodDaily - : buildMeasurementDailySeries( - key as MeasurementType, - measurementsByType.get(key) ?? [], - tz, - priorityJson, - ); + key === "MOOD" ? moodDaily : (measurementDaily.byType.get(key) ?? []); const series: NamedSeries[] = []; for (const key of DISCOVERY_BEHAVIOURS) { @@ -287,7 +276,12 @@ async function buildCorrelationsResponse( // capped read still covers the recent window `discoverEmergingCorrelations` // needs; this only tells a dashboard the retrospective scan's older // half of the window may be thin. + // The cap can only apply to raw-path channels — rollup-served + // channels read one row per day per source and cannot reach it. measurements_capped: measurementsCapped, + // Rollup read-swap reach: how many measurement channels rode the DAY + // rollup tier on this build (the rest took the raw fallback path). + measurement_rollup_channels: measurementDaily.rollupTypes.length, mood_entries_capped: moodCapped, // FDREXTEND — per-channel day-counts so a dashboard can see whether the // two sparse new channels reached the n ≥ 20 floor or degraded to absent. diff --git a/src/lib/insights/correlation-channel-series.ts b/src/lib/insights/correlation-channel-series.ts index caac368f2..57ae69b63 100644 --- a/src/lib/insights/correlation-channel-series.ts +++ b/src/lib/insights/correlation-channel-series.ts @@ -46,6 +46,9 @@ import { pickMainNightAndNaps, type SleepStageRow, } from "@/lib/analytics/sleep-night"; +import { isNearUtc } from "@/lib/tz/format"; +import { probeRollupCoverage } from "@/lib/rollups/measurement-coverage"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import type { MeasurementSource, MeasurementType, @@ -556,6 +559,207 @@ function buildCumulativeDailySeries( .sort((a, b) => (a.day < b.day ? -1 : 1)); } +/** One tiered measurement daily-series fetch's result. */ +export interface TieredMeasurementDailySeries { + /** + * Per-type daily series, same shape `buildMeasurementDailySeries` + * produces. A requested type with no data maps to an empty array or is + * absent — callers treat both as "channel degrades to absent". + */ + byType: Map; + /** + * True when the RAW fallback read hit {@link MEASUREMENT_READ_CAP} — + * rollup-served channels read one row per day per source and cannot cap, + * so the flag now describes only the channels that took the raw path. + */ + measurementsCapped: boolean; + /** Channel types served from the DAY rollup tier on this call. */ + rollupTypes: MeasurementType[]; +} + +/** + * True for the discovery measurement channels whose daily reduction the + * DAY rollup tier can reproduce. Two families are structurally excluded: + * + * - `SLEEP_DURATION` — the daily figure is the MAIN night's total asleep + * minutes via `reconstructSleepSessions` (writer-dedup, midnight-spanning + * session clustering, nap exclusion). A DAY bucket over per-STAGE rows + * carries none of that; no composition of `count/mean/sum` recovers it. + * - `CUMULATIVE_HK_TYPES` (steps, daylight, …) — the raw path collapses on + * TWO axes via `pickCanonicalSourceRows`: source ladder, then device-type + * ladder WITHIN the picked source (watch beats phone beats scale). Rollup + * rows are per `(day, source)` only — the device-type axis is folded away + * at write time, so a multi-device day would re-inflate the total the raw + * path deliberately de-duplicates. + * + * Both stay on the raw path rather than shipping a silent value change. + */ +function isRollupSwapEligible(type: MeasurementType): boolean { + return type !== "SLEEP_DURATION" && !CUMULATIVE_HK_TYPES.has(type); +} + +/** + * Collapse per-`(day, source)` DAY rollup rows into per-day MEANS across + * ALL sources — the exact reduction `toDailyMeans` applies to the raw + * rows. Deliberately NOT `collapseRollupRowsBySource`: the raw spot-metric + * path averages every reading of the day regardless of source, so the + * ladder collapse would CHANGE a multi-source day's value. The + * count-weighted compose (Σ sum / Σ count, preferring the exact stored + * `sumValue` over the `mean·count` round trip) reproduces the all-rows + * mean exactly up to float addition order. + */ +function composeRollupDailyMeans( + rows: Array<{ + bucketStart: Date; + count: number; + mean: number; + sumValue: number | null; + }>, +): DailySeriesPoint[] { + const byDay = new Map(); + for (const r of rows) { + if (r.count <= 0) continue; + const sum = r.sumValue ?? r.mean * r.count; + if (!Number.isFinite(sum)) continue; + // DAY buckets are minted at UTC midnight; the UTC date IS the bucket's + // day key (the same convention graded-series and the derived baselines + // read the tier under). + const day = r.bucketStart.toISOString().slice(0, 10); + const acc = byDay.get(day) ?? { sum: 0, count: 0 }; + acc.sum += sum; + acc.count += r.count; + byDay.set(day, acc); + } + return [...byDay.entries()] + .map(([day, acc]) => ({ day, value: acc.sum / acc.count })) + .sort((a, b) => (a.day < b.day ? -1 : 1)); +} + +/** + * Rollup read-swap for the correlation-discovery measurement channels: the + * per-day means the discovery scan needs already sit in the DAY rollup + * tier, so eligible spot channels read one pre-aggregated row per day per + * source instead of every raw reading in the 180-day window (a CGM + * glucose channel alone can hold tens of thousands of raw rows). + * + * Read-swap semantics (standing rule: replace, with fallback-on-miss — + * never both): per channel, EITHER the rollup tier serves the series OR + * the raw `fetchMeasurementWindowSeries` + `buildMeasurementDailySeries` + * path does. A channel falls back when + * - the type is structurally ineligible ({@link isRollupSwapEligible}), + * - the user's profile timezone is outside the near-UTC band (below), + * - the coverage probe reports no DAY buckets for the type, or + * - the in-window bucket read comes back empty (backfill pending). + * + * Timezone contract — the same v1.4.38 W-A guard the correlation + * hypotheses fast path carries: DAY buckets group rows by UTC day while + * the raw path keys days in the profile tz. Inside the ±3 h `isNearUtc` + * band the two calendars agree for every reading logged more than the + * tz-offset away from local midnight; readings inside that band attribute + * to the neighbouring day (the documented, accepted tolerance — the + * n ≥ 20 / FDR gates absorb the single-day phase shift). Outside the band + * the calendars diverge for most of the day, so the guard forces the raw + * path and day-key parity is preserved exactly. + * + * Window contract: buckets are read `bucketStart >= since`, so the + * partial OLDEST day (the one `since` lands inside) is not served — the + * rollup series covers full days only, while the raw path would have + * included that day's post-`since` readings. One day at the 180-day + * horizon, equivalent to the user not logging that day. + * + * NOT `readBestGranularityRollups`: its floor table routes a >90-day + * window to the WEEK tier first, which cannot feed a daily series — the + * discovery scan needs the DAY grain unconditionally. + */ +export async function fetchMeasurementDailySeriesTiered( + userId: string, + tz: string, + since: Date, + types: MeasurementType[], +): Promise { + const priorityJson = await loadUserSourcePriority(userId); + + const rawTypes: MeasurementType[] = []; + const rollupCandidates: MeasurementType[] = []; + const nearUtc = isNearUtc(tz); + for (const type of types) { + if (nearUtc && isRollupSwapEligible(type)) rollupCandidates.push(type); + else rawTypes.push(type); + } + + const byType = new Map(); + const rollupTypes: MeasurementType[] = []; + + if (rollupCandidates.length > 0) { + const coverage = await probeRollupCoverage(userId); + const covered: MeasurementType[] = []; + for (const type of rollupCandidates) { + if (coverage.get(type) === true) covered.push(type); + else rawTypes.push(type); + } + if (covered.length > 0) { + const buckets = await prisma.measurementRollup.findMany({ + where: { + userId, + type: { in: covered }, + granularity: "DAY", + bucketStart: { gte: since }, + }, + orderBy: { bucketStart: "asc" }, + select: { + type: true, + bucketStart: true, + count: true, + mean: true, + sumValue: true, + }, + }); + const rowsByType = new Map(); + for (const bucket of buckets) { + const list = rowsByType.get(bucket.type) ?? []; + list.push(bucket); + rowsByType.set(bucket.type, list); + } + for (const type of covered) { + const series = composeRollupDailyMeans(rowsByType.get(type) ?? []); + if (series.length === 0) { + // Coverage said "has buckets" but the window resolved empty — + // backfill pending or all data older than the window. Fall back + // so a partially-backfilled account never reads a thinner series + // than the raw path would produce. + rawTypes.push(type); + } else { + byType.set(type, series); + rollupTypes.push(type); + } + } + } + } + + let measurementsCapped = false; + if (rawTypes.length > 0) { + const rawFetch = await fetchMeasurementWindowSeries( + userId, + since, + rawTypes, + ); + measurementsCapped = rawFetch.measurementsCapped; + for (const type of rawTypes) { + byType.set( + type, + buildMeasurementDailySeries( + type, + rawFetch.byType.get(type) ?? [], + tz, + priorityJson, + ), + ); + } + } + + return { byType, measurementsCapped, rollupTypes }; +} + /** * v1.22 — build the user's lab draws for the labs ↔ outcome correlation pass. * From f9ac7a3327cbaa6c88a44079159c3c9dd17bb4c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 13 Aug 2026 22:06:10 +0200 Subject: [PATCH 2/4] Pin the rollup swap's parity, fallback, and eligibility contracts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both paths read identical fixtures — per-(day, source) DAY rollup rows derived from the same raw rows the fallback reads — and must produce equivalent daily series: the plain daily-mean case, the multi-source day (the all-source weighted mean, never a ladder-collapsed single source), and the pre-sum_value mean-times-count compose. A far-from-UTC profile timezone must force the raw path so day keys stay profile-tz keyed; the near-UTC midnight band's single-day attribution shift is pinned explicitly so it stays a documented decision rather than drift. Coverage misses and empty in-window bucket reads fall back per channel; sleep and the cumulative types never enter the rollup query. Every assertion family was broken on purpose and observed failing with the site named before restoring green; the record sits in the test's doc comment. The route test mocks the new coverage probe and bucket read as empty so every channel rides the raw path unchanged there. --- .../correlations/__tests__/route.test.ts | 12 + ...elation-channel-series-rollup-swap.test.ts | 413 ++++++++++++++++++ 2 files changed, 425 insertions(+) create mode 100644 src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts diff --git a/src/app/api/insights/correlations/__tests__/route.test.ts b/src/app/api/insights/correlations/__tests__/route.test.ts index 2e19d0ea1..0a110f9ed 100644 --- a/src/app/api/insights/correlations/__tests__/route.test.ts +++ b/src/app/api/insights/correlations/__tests__/route.test.ts @@ -9,6 +9,12 @@ vi.mock("@/lib/db", () => ({ findUnique: vi.fn().mockResolvedValue({ timezone: "Europe/Berlin" }), }, measurement: { findMany: vi.fn().mockResolvedValue([]) }, + // Rollup read-swap — the tiered measurement fetch probes DAY-bucket + // coverage ($queryRaw) and reads `measurement_rollups`. Default both + // empty so every channel takes the raw fallback path and the existing + // assertions ride through unchanged. + measurementRollup: { findMany: vi.fn().mockResolvedValue([]) }, + $queryRaw: vi.fn().mockResolvedValue([]), moodEntry: { findMany: vi.fn().mockResolvedValue([]) }, // FDREXTEND — the route now folds two non-measurement channels in (built by // the shared `correlation-channel-series` helpers). Default to empty corpora @@ -115,6 +121,12 @@ beforeEach(() => { (prisma.measurement.findMany as ReturnType).mockResolvedValue( [], ); + // Rollup read-swap — empty coverage probe + bucket read keep every + // measurement channel on the raw fallback path. + ( + prisma.measurementRollup.findMany as ReturnType + ).mockResolvedValue([]); + (prisma.$queryRaw as ReturnType).mockResolvedValue([]); (prisma.moodEntry.findMany as ReturnType).mockResolvedValue([]); // FDREXTEND — the two non-measurement channels read their own models; default // them empty so neither channel produces a point. diff --git a/src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts b/src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts new file mode 100644 index 000000000..de6d72f18 --- /dev/null +++ b/src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts @@ -0,0 +1,413 @@ +/** + * Parity battery for the correlation-discovery rollup read-swap + * (`fetchMeasurementDailySeriesTiered`). + * + * Every parity case seeds IDENTICAL fixtures through both paths: the raw + * measurement rows the fallback path reads, and per-`(day, source)` DAY + * rollup rows DERIVED from those same raw rows (count / mean / sumValue, + * UTC-midnight `bucketStart` — the exact shape the rollup writer mints). + * The resulting daily series must be equivalent. Fixtures are now-anchored + * (never fixed calendar dates) so the suite cannot age out of a retention + * or DST window. + * + * Watched-red record (each assertion family was broken on purpose, seen + * failing with the site named, then restored green): + * - skewing `composeRollupDailyMeans` (`acc.sum += sum + 1`) → the + * spot-parity and multi-source parity tests failed on the composed + * values (57.75 vs 58.25, 81 vs 81.5); + * - inverting the `isNearUtc` gate in `fetchMeasurementDailySeriesTiered` + * → the far-from-UTC test failed on `rollupFindMany` being called and + * on UTC-keyed days where profile-tz keys were required; + * - removing the empty-window `rawTypes.push(type)` fallback → the + * fallback-on-miss test failed with an empty series; + * - removing the `isRollupSwapEligible` sleep/cumulative exclusion → the + * eligibility test failed on the rollup query's `type.in` list. + */ +import { describe, it, expect, vi, beforeEach, afterAll } from "vitest"; +import type { MeasurementSource } from "@/generated/prisma/client"; + +const measurementFindMany = vi.fn(); +const rollupFindMany = vi.fn(); +const userFindUnique = vi.fn(); +const queryRaw = vi.fn(); +vi.mock("@/lib/db", () => ({ + prisma: { + measurement: { findMany: (a: unknown) => measurementFindMany(a) }, + measurementRollup: { findMany: (a: unknown) => rollupFindMany(a) }, + user: { findUnique: (a: unknown) => userFindUnique(a) }, + $queryRaw: (...a: unknown[]) => queryRaw(...a), + }, +})); + +import { + fetchMeasurementDailySeriesTiered, + toDailyMeans, +} from "@/lib/insights/correlation-channel-series"; + +const DAY_MS = 24 * 60 * 60 * 1000; +const USER = "user-1"; +const SINCE = () => new Date(Date.now() - 30 * DAY_MS); + +/** Now-anchored instant: `daysAgo` days back, pinned to a UTC wall time. */ +function at(daysAgo: number, hourUtc: number, minute = 0): Date { + const d = new Date(Date.now() - daysAgo * DAY_MS); + d.setUTCHours(hourUtc, minute, 0, 0); + return d; +} + +/** UTC calendar-date key of an instant — the DAY bucket's day key. */ +function utcDayKey(d: Date): string { + return d.toISOString().slice(0, 10); +} + +interface RawRow { + type: string; + value: number; + measuredAt: Date; + source: MeasurementSource; + deviceType: string | null; + sleepStage: string | null; +} + +function raw( + type: string, + measuredAt: Date, + value: number, + source: MeasurementSource = "APPLE_HEALTH", +): RawRow { + return { + type, + value, + measuredAt, + source, + deviceType: null, + sleepStage: null, + }; +} + +/** + * Derive the per-`(type, UTC day, source)` DAY rollup rows the writer + * would mint from the given raw rows. `nullSumValueDays` simulates rows + * that predate the `sum_value` column (the compose must fall back to + * `mean·count`). + */ +function deriveDayRollups( + rows: RawRow[], + nullSumValueDays: string[] = [], +): Array<{ + type: string; + bucketStart: Date; + count: number; + mean: number; + sumValue: number | null; +}> { + const groups = new Map< + string, + { type: string; day: string; source: string; values: number[] } + >(); + for (const r of rows) { + const day = utcDayKey(r.measuredAt); + const key = `${r.type}|${day}|${r.source}`; + const g = groups.get(key) ?? { + type: r.type, + day, + source: r.source, + values: [], + }; + g.values.push(r.value); + groups.set(key, g); + } + return [...groups.values()].map((g) => { + const sum = g.values.reduce((s, v) => s + v, 0); + return { + type: g.type, + bucketStart: new Date(`${g.day}T00:00:00.000Z`), + count: g.values.length, + mean: sum / g.values.length, + sumValue: nullSumValueDays.includes(g.day) ? null : sum, + }; + }); +} + +function mockCoverage(entries: Array<[string, boolean]>): void { + queryRaw.mockResolvedValue( + entries.map(([type, has]) => ({ type, has_buckets: has })), + ); +} + +beforeEach(() => { + vi.resetAllMocks(); + // `loadUserSourcePriority` reads the priority blob; null = default ladders. + userFindUnique.mockResolvedValue({ sourcePriorityJson: null }); + measurementFindMany.mockResolvedValue([]); + rollupFindMany.mockResolvedValue([]); + queryRaw.mockResolvedValue([]); +}); + +describe("rollup ↔ raw parity — plain daily-mean type", () => { + // Wall times sit mid-day, far from every local midnight band, so the + // UTC bucket day and the Europe/Berlin profile day agree by construction. + const rows = [ + raw("RESTING_HEART_RATE", at(10, 8), 55), + raw("RESTING_HEART_RATE", at(10, 20), 61), + raw("RESTING_HEART_RATE", at(9, 8), 58), + raw("RESTING_HEART_RATE", at(5, 12), 60.5), + ]; + + it("serves the identical daily series from the DAY rollup tier (raw path not consulted)", async () => { + mockCoverage([["RESTING_HEART_RATE", true]]); + // One bucket rides the pre-`sum_value` shape to pin the mean·count fallback. + rollupFindMany.mockResolvedValue( + deriveDayRollups(rows, [utcDayKey(at(9, 8))]), + ); + + const viaRollup = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["RESTING_HEART_RATE"], + ); + + expect(viaRollup.rollupTypes).toEqual(["RESTING_HEART_RATE"]); + expect(viaRollup.measurementsCapped).toBe(false); + // Read-swap REPLACES the raw read — it must not run in parallel. + expect(measurementFindMany).not.toHaveBeenCalled(); + + mockCoverage([["RESTING_HEART_RATE", false]]); + measurementFindMany.mockResolvedValue(rows); + const viaRaw = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["RESTING_HEART_RATE"], + ); + + const rollupSeries = viaRollup.byType.get("RESTING_HEART_RATE")!; + const rawSeries = viaRaw.byType.get("RESTING_HEART_RATE")!; + expect(rollupSeries.map((p) => p.day)).toEqual(rawSeries.map((p) => p.day)); + rollupSeries.forEach((p, i) => + expect(p.value).toBeCloseTo(rawSeries[i].value, 10), + ); + // Spot-check the multi-reading day collapses to the day MEAN. + expect( + rollupSeries.find((p) => p.day === utcDayKey(at(10, 8)))?.value, + ).toBeCloseTo(58, 10); + expect(rollupSeries).toHaveLength(3); + }); +}); + +describe("rollup ↔ raw parity — multi-source day (source collapse)", () => { + const rows = [ + raw("WEIGHT", at(8, 7), 80.4, "APPLE_HEALTH"), + raw("WEIGHT", at(8, 7, 5), 81.6, "WITHINGS"), + raw("WEIGHT", at(6, 7), 80.0, "APPLE_HEALTH"), + ]; + + it("composes the ALL-source day mean, matching toDailyMeans — never a ladder-collapsed single source", async () => { + mockCoverage([["WEIGHT", true]]); + rollupFindMany.mockResolvedValue(deriveDayRollups(rows)); + const viaRollup = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["WEIGHT"], + ); + + mockCoverage([["WEIGHT", false]]); + measurementFindMany.mockResolvedValue(rows); + const viaRaw = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["WEIGHT"], + ); + + const rollupSeries = viaRollup.byType.get("WEIGHT")!; + const rawSeries = viaRaw.byType.get("WEIGHT")!; + expect(rollupSeries.map((p) => p.day)).toEqual(rawSeries.map((p) => p.day)); + rollupSeries.forEach((p, i) => + expect(p.value).toBeCloseTo(rawSeries[i].value, 10), + ); + + // The dual-source day is the MEAN across both sources' readings (81.0). + // A ladder collapse would have returned one source's value (80.4 or + // 81.6) — pin that it did not. + const dualDay = rollupSeries.find((p) => p.day === utcDayKey(at(8, 7)))!; + expect(dualDay.value).toBeCloseTo(81.0, 10); + expect(dualDay.value).not.toBeCloseTo(80.4, 2); + expect(dualDay.value).not.toBeCloseTo(81.6, 2); + }); +}); + +describe("rollup ↔ raw parity — far-from-UTC profile timezone", () => { + const savedTz = process.env.TZ; + afterAll(() => { + process.env.TZ = savedTz; + }); + + it("forces the raw path when UTC and profile-tz day keys diverge, preserving the tz-keyed series", async () => { + // Process tz deliberately differs from BOTH UTC and the profile tz so + // any accidental host-clock dependence would surface here. + process.env.TZ = "America/Chicago"; + + // 13:00 UTC is 01:00–02:00 the NEXT calendar day in Pacific/Auckland + // (+12 NZST / +13 NZDT) — every fixture's UTC day key and profile-tz + // day key differ. + const rows = [ + raw("RESTING_HEART_RATE", at(9, 13), 52), + raw("RESTING_HEART_RATE", at(7, 13), 56), + ]; + // Coverage TRUE and buckets present: if the guard fails, the rollup + // path would serve UTC-keyed days and this test names it. + mockCoverage([["RESTING_HEART_RATE", true]]); + rollupFindMany.mockResolvedValue(deriveDayRollups(rows)); + measurementFindMany.mockResolvedValue(rows); + + const result = await fetchMeasurementDailySeriesTiered( + USER, + "Pacific/Auckland", + SINCE(), + ["RESTING_HEART_RATE"], + ); + + expect(rollupFindMany).not.toHaveBeenCalled(); + expect(result.rollupTypes).toEqual([]); + const series = result.byType.get("RESTING_HEART_RATE")!; + // Byte-identical to the pure tz-keyed reduction the raw path pins. + expect(series).toEqual( + toDailyMeans( + rows.map((r) => ({ value: r.value, at: r.measuredAt })), + "Pacific/Auckland", + ), + ); + // And those keys are the profile-tz (next-day) dates, not the UTC dates. + expect(series.map((p) => p.day)).toEqual([ + utcDayKey(new Date(at(9, 13).getTime() + DAY_MS)), + utcDayKey(new Date(at(7, 13).getTime() + DAY_MS)), + ]); + }); +}); + +describe("near-UTC midnight band — the documented attribution tolerance", () => { + it("pins that a reading inside the offset band lands on the UTC day via rollups and the local day via raw", async () => { + // 23:30 UTC is 01:30 the next Berlin day. Inside the ±3 h near-UTC + // band the rollup swap accepts this single-day attribution shift (the + // v1.4.38 W-A tolerance the correlations fast path documents); this + // test pins the shift so it stays a decision, never drift. + const reading = raw("RESTING_HEART_RATE", at(10, 23, 30), 60); + + mockCoverage([["RESTING_HEART_RATE", true]]); + rollupFindMany.mockResolvedValue(deriveDayRollups([reading])); + const viaRollup = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["RESTING_HEART_RATE"], + ); + + mockCoverage([["RESTING_HEART_RATE", false]]); + measurementFindMany.mockResolvedValue([reading]); + const viaRaw = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["RESTING_HEART_RATE"], + ); + + const utcDay = utcDayKey(reading.measuredAt); + const berlinDay = utcDayKey( + new Date(reading.measuredAt.getTime() + DAY_MS), + ); + expect(viaRollup.byType.get("RESTING_HEART_RATE")).toEqual([ + { day: utcDay, value: 60 }, + ]); + expect(viaRaw.byType.get("RESTING_HEART_RATE")).toEqual([ + { day: berlinDay, value: 60 }, + ]); + }); +}); + +describe("fallback-on-miss", () => { + it("takes the raw path for a channel with no rollup coverage and produces the same series as before", async () => { + const rows = [ + raw("BLOOD_GLUCOSE", at(4, 9), 95), + raw("BLOOD_GLUCOSE", at(4, 15), 105), + ]; + mockCoverage([["BLOOD_GLUCOSE", false]]); + measurementFindMany.mockResolvedValue(rows); + + const result = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["BLOOD_GLUCOSE"], + ); + + expect(rollupFindMany).not.toHaveBeenCalled(); + expect(result.rollupTypes).toEqual([]); + expect(result.byType.get("BLOOD_GLUCOSE")).toEqual( + toDailyMeans( + rows.map((r) => ({ value: r.value, at: r.measuredAt })), + "Europe/Berlin", + ), + ); + }); + + it("falls back per channel when coverage exists but the window's buckets are empty (backfill pending)", async () => { + const rows = [raw("WEIGHT", at(3, 7), 79.5)]; + mockCoverage([["WEIGHT", true]]); + rollupFindMany.mockResolvedValue([]); // covered, but nothing in-window + measurementFindMany.mockResolvedValue(rows); + + const result = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["WEIGHT"], + ); + + expect(result.rollupTypes).toEqual([]); + expect(result.byType.get("WEIGHT")).toEqual([ + { day: utcDayKey(at(3, 7)), value: 79.5 }, + ]); + }); +}); + +describe("structural eligibility", () => { + it("keeps sleep and the cumulative types on the raw path even with full coverage", async () => { + mockCoverage([ + ["SLEEP_DURATION", true], + ["ACTIVITY_STEPS", true], + ["TIME_IN_DAYLIGHT", true], + ["WEIGHT", true], + ]); + const weightRows = [raw("WEIGHT", at(3, 7), 79.5)]; + rollupFindMany.mockResolvedValue(deriveDayRollups(weightRows)); + measurementFindMany.mockResolvedValue([]); + + const result = await fetchMeasurementDailySeriesTiered( + USER, + "Europe/Berlin", + SINCE(), + ["SLEEP_DURATION", "ACTIVITY_STEPS", "TIME_IN_DAYLIGHT", "WEIGHT"], + ); + + // The rollup query may only ever name the eligible spot type. + expect(rollupFindMany).toHaveBeenCalledTimes(1); + const rollupWhere = rollupFindMany.mock.calls[0][0] as { + where: { type: { in: string[] } }; + }; + expect(rollupWhere.where.type.in).toEqual(["WEIGHT"]); + expect(result.rollupTypes).toEqual(["WEIGHT"]); + + // The raw read carries exactly the three structurally-excluded types. + const rawWhere = measurementFindMany.mock.calls[0][0] as { + where: { type: { in: string[] } }; + }; + expect(rawWhere.where.type.in.sort()).toEqual([ + "ACTIVITY_STEPS", + "SLEEP_DURATION", + "TIME_IN_DAYLIGHT", + ]); + }); +}); From 35d798ad688ac4801291c0fcbacfc3ba35831944 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 13 Aug 2026 22:07:57 +0200 Subject: [PATCH 3/4] Round-trip the rollup swap through the real writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unit battery derives rollup fixtures in JS; this pins the same parity through the pipe itself. Seed raw measurements, read the tiered fetch before the fold (coverage miss, raw path baseline), fold with the real recomputeUserRollups DAY pass, read again, and assert the two daily series are equivalent — including a dual-source weight day whose per-source rows must compose to the all-source mean. A drifted writer projection, coverage-probe SQL, or bucket compose surfaces here. --- .../correlations-rollup-swap.test.ts | 192 ++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 tests/integration/correlations-rollup-swap.test.ts diff --git a/tests/integration/correlations-rollup-swap.test.ts b/tests/integration/correlations-rollup-swap.test.ts new file mode 100644 index 000000000..c160aa598 --- /dev/null +++ b/tests/integration/correlations-rollup-swap.test.ts @@ -0,0 +1,192 @@ +/** + * Correlation-discovery rollup read-swap — DB round trip. + * + * The unit battery + * (`src/lib/insights/__tests__/correlation-channel-series-rollup-swap.test.ts`) + * derives rollup fixtures in JS; this suite proves the SAME parity through + * the real pipe: seed raw measurements, fold them with the real rollup + * writer (`recomputeUserRollups` — the Postgres `AVG` / `SUM` / `COUNT` + * grouped per `(type, day, source)`), then read the tiered fetch twice — + * once BEFORE the fold (coverage probe misses → raw path) and once AFTER + * (DAY buckets serve) — and assert the two daily series are equivalent. + * A drifted writer projection, coverage-probe SQL, or bucket compose all + * surface here instead of in production. + * + * Fixtures are now-anchored (never fixed calendar dates) and pinned to + * mid-day UTC wall times, so the UTC bucket day and the Europe/Berlin + * profile day agree by construction and the suite cannot age out. + * + * Watched-red record: skewing `composeRollupDailyMeans` (`acc.sum += sum + * + 1`) failed the post-fold parity assertions here naming the composed + * values; restored green. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { fetchMeasurementDailySeriesTiered } from "@/lib/insights/correlation-channel-series"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +// The DAY-only recompute never touches pg-boss; leave the boss detached for +// parity with the sibling rollup suites (`isolate: false` shares the mock). +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** Now-anchored instant: `daysAgo` days back, pinned to a UTC wall time. */ +function at(daysAgo: number, hourUtc: number, minute = 0): Date { + const d = new Date(Date.now() - daysAgo * DAY_MS); + d.setUTCHours(hourUtc, minute, 0, 0); + return d; +} + +let seq = 0; +async function seedUser(prisma: ReturnType) { + seq += 1; + return prisma.user.create({ + data: { + username: `corr-rollup-swap-${seq}`, + email: `corr-rollup-swap-${seq}@example.test`, + role: "USER", + timezone: "Europe/Berlin", + }, + }); +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("correlations rollup swap — real writer round trip", () => { + it("serves the same daily series from the folded DAY buckets as from the raw rows", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + const since = new Date(Date.now() - 30 * DAY_MS); + const types = [ + "RESTING_HEART_RATE", + "WEIGHT", + "BLOOD_GLUCOSE", + ] as const satisfies readonly string[]; + + await prisma.measurement.createMany({ + data: [ + // RHR — two readings on one day (day-mean collapse), one on another. + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 55, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: at(10, 8), + }, + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 61, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: at(10, 18), + }, + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 58, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: at(8, 9), + }, + // WEIGHT — a dual-source day: the discovery series is the ALL-source + // mean (81.0), which the reader must compose across the two + // per-source rollup rows the writer mints. + { + userId: user.id, + type: "WEIGHT", + value: 80.4, + unit: "kg", + source: "APPLE_HEALTH", + measuredAt: at(7, 7), + }, + { + userId: user.id, + type: "WEIGHT", + value: 81.6, + unit: "kg", + source: "WITHINGS", + measuredAt: at(7, 7, 30), + }, + // GLUCOSE — several readings across a day. + { + userId: user.id, + type: "BLOOD_GLUCOSE", + value: 92, + unit: "mg/dL", + source: "MANUAL", + measuredAt: at(5, 8), + }, + { + userId: user.id, + type: "BLOOD_GLUCOSE", + value: 118, + unit: "mg/dL", + source: "MANUAL", + measuredAt: at(5, 13), + }, + { + userId: user.id, + type: "BLOOD_GLUCOSE", + value: 101, + unit: "mg/dL", + source: "MANUAL", + measuredAt: at(5, 19), + }, + ], + }); + + // BEFORE the fold: the coverage probe finds no DAY buckets, so every + // channel takes the raw path — this is the pre-swap behaviour baseline. + const viaRaw = await fetchMeasurementDailySeriesTiered( + user.id, + "Europe/Berlin", + since, + [...types], + ); + expect(viaRaw.rollupTypes).toEqual([]); + + // Fold with the REAL writer (per-source DAY rows via Postgres AVG/SUM). + await recomputeUserRollups(user.id, { granularities: ["DAY"] }); + + const viaRollup = await fetchMeasurementDailySeriesTiered( + user.id, + "Europe/Berlin", + since, + [...types], + ); + expect(viaRollup.rollupTypes.sort()).toEqual([...types].sort()); + + for (const type of types) { + const rawSeries = viaRaw.byType.get(type)!; + const rollupSeries = viaRollup.byType.get(type)!; + expect(rollupSeries.map((p) => p.day)).toEqual( + rawSeries.map((p) => p.day), + ); + rollupSeries.forEach((p, i) => + expect(p.value).toBeCloseTo(rawSeries[i].value, 10), + ); + } + + // Anchor the dual-source WEIGHT day explicitly: all-source mean, never + // a ladder-collapsed single source's value. + const weightDay = viaRollup.byType.get("WEIGHT")!; + expect(weightDay).toHaveLength(1); + expect(weightDay[0].value).toBeCloseTo(81.0, 10); + }); +}); From 82a535c9b5629a12b0c9b21487232c5a979afab2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 13 Aug 2026 22:58:38 +0200 Subject: [PATCH 4/4] =?UTF-8?q?chore(release):=20v1.37.17=20=E2=80=94=20th?= =?UTF-8?q?e=20discovery=20reads=20days,=20not=20readings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 6 ++++++ docs/api/openapi.yaml | 2 +- package.json | 2 +- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d41e1c65..7f182fcb7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## [1.37.17] — 2026-08-13 + +### Changed + +- The correlations discovery reads its measurement series from the pre-aggregated daily tier instead of walking every raw reading over 180 days. For a dense record, glucose from a continuous sensor most of all, this turns tens of thousands of rows into at most one row per day and source. Values stay the same: the daily figures are composed to match the previous per-reading averages exactly, a channel without aggregate coverage falls back to the raw path on its own, and sleep duration, steps and daylight time keep their raw reads because their daily figures depend on session reconstruction and device de-duplication that the aggregates cannot express. Profiles far from UTC also stay on the raw path so day boundaries keep matching their timezone. + ## [1.37.16] — 2026-08-13 ### Changed diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 43481fd39..5e4118d0c 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.37.16 + version: 1.37.17 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index b04ff5c14..5bac2a0fa 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.37.16", + "version": "1.37.17", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev",