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] 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. *