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