diff --git a/src/app/api/analytics/route.ts b/src/app/api/analytics/route.ts index f10438b2c..65d2b3cf3 100644 --- a/src/app/api/analytics/route.ts +++ b/src/app/api/analytics/route.ts @@ -8,12 +8,6 @@ import { summarize, type DataPoint } from "@/lib/analytics/trends"; import { computeSummariesSlice } from "@/lib/analytics/summaries-slice"; import { getBpTargets } from "@/lib/analytics/bp-targets"; import { userDayKey, DEFAULT_TIMEZONE } from "@/lib/tz/resolver"; -import { - correlateBpCompliance, - correlateMoodPulse, - correlateWeightWeekday, - type CorrelationResult, -} from "@/lib/insights/correlations"; import type { MeasurementSource, MeasurementType, @@ -25,6 +19,7 @@ import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; import { probeRollupCoverage } from "@/lib/measurements/rollup-coverage"; import { computeBpInTargetFastPath } from "@/lib/analytics/bp-in-target-fast-path"; import { computeUserHealthScoreFastPath } from "@/lib/analytics/health-score-fast-path"; +import { computeCorrelationHypothesesFastPath } from "@/lib/analytics/correlations-fast-path"; import { isCumulativeDaySumType, pickCumulativeDaySum, @@ -389,12 +384,20 @@ async function buildAnalyticsResponse(user: AuthedUser) { } // v1.4.20 phase B3 — three pre-defined correlation hypotheses. - // All three run on the trailing 30 days so a sparse account doesn't - // burn the n >= 20 gate on a stale window (v1.4.23 H6 raised the - // floor from 14 → 20). Each runner gates on n >= 20 + p < 0.05; - // below the bar the result.status === "insufficient" and the UI - // paints an EmptyState. - const correlations = await computeCorrelationHypotheses(user.id, userTz); + // v1.4.37 W2 — probe-gated helper. The 28-day scan window (down + // from 30 to keep the cold critical path tight while still + // satisfying the n >= 20 surface gate) reads SYS / PULSE / WEIGHT + // per-day means from `measurement_rollups` when the user has full + // coverage; mood and medication-intake reads always stay live (no + // rollup equivalent). The helper emits `meta.correlations.path` + + // `meta.correlations.window_days` so prod logs prove the branch + // selection and the truthful window. + const correlations = await computeCorrelationHypothesesFastPath({ + userId: user.id, + userTz, + now: new Date(), + coverage, + }); // v1.4.20 phase B5 — Personal Health Score. Server-deterministic // composite of BP-in-target % + weight-trend alignment + mood @@ -506,160 +509,11 @@ async function computeSleepStageBreakdown( }; } -/** - * Build inputs for the three pre-defined hypotheses + run them. - * Pure-ish — only Prisma reads, no external calls. - * - * Window: trailing 30 days. Anything older falls outside the surface - * because the user-facing "based on N paired readings · last 30 days" - * source-chip has to remain truthful. - */ -async function computeCorrelationHypotheses( - userId: string, - userTz: string, -): Promise<{ - bpCompliance: CorrelationResult; - moodPulse: CorrelationResult; - weightWeekday: CorrelationResult; -}> { - const DAY_MS = 24 * 60 * 60 * 1000; - const since = new Date(Date.now() - 30 * DAY_MS); - - // v1.4.23 Sr-H1 — the four measurement reads route through the - // chunked helper so even a noisy 30-day window (e.g. minute-level - // HealthKit pulse samples) cannot allocate an unbounded buffer. The - // helper still returns the full filtered series the Pearson runners - // need; we just bound the per-page Prisma round-trip. - const [sysRows, pulseRows, weightRows, moodRows, intakeRows] = - await Promise.all([ - fetchMeasurementSeriesChunked(userId, "BLOOD_PRESSURE_SYS", { since }), - fetchMeasurementSeriesChunked(userId, "PULSE", { since }), - fetchMeasurementSeriesChunked(userId, "WEIGHT", { since }), - prisma.moodEntry.findMany({ - where: { userId, moodLoggedAt: { gte: since } }, - select: { score: true, moodLoggedAt: true, date: true }, - }), - prisma.medicationIntakeEvent.findMany({ - where: { userId, scheduledFor: { gte: since } }, - select: { scheduledFor: true, takenAt: true, skipped: true }, - }), - ]); - - // ── Hypothesis 1: BP × medication compliance ──────────────── - // Aggregate by the user's display-tz day key so DST + UTC boundary - // issues don't split a day's readings. The day's "compliance %" is - // taken / expected for that calendar day across all medications. - const dayKey = (d: Date): string => userDayKey(d, userTz); - - const dailySys = new Map(); - for (const row of sysRows) { - const key = dayKey(row.measuredAt); - const list = dailySys.get(key) ?? []; - list.push(row.value); - dailySys.set(key, list); - } - - const dailyCompliance = new Map< - string, - { expected: number; taken: number } - >(); - for (const event of intakeRows) { - const key = dayKey(event.scheduledFor); - const slot = dailyCompliance.get(key) ?? { expected: 0, taken: 0 }; - slot.expected += 1; - if (event.takenAt && !event.skipped) slot.taken += 1; - dailyCompliance.set(key, slot); - } - - const bpCompliancePairs: Array<{ - date: Date; - systolic: number; - compliancePct: number; - }> = []; - for (const [key, sysValues] of dailySys.entries()) { - const slot = dailyCompliance.get(key); - if (!slot || slot.expected === 0) continue; - const compliancePct = (slot.taken / slot.expected) * 100; - const meanSys = sysValues.reduce((s, v) => s + v, 0) / sysValues.length; - bpCompliancePairs.push({ - date: dateFromDayKey(key), - systolic: meanSys, - compliancePct, - }); - } - const bpCompliance = correlateBpCompliance({ daily: bpCompliancePairs }); - - // ── Hypothesis 2: Mood × resting pulse ────────────────────── - // Same-day pairing: take the day's mean mood vs the day's mean pulse. - // "Resting" is approximated by mean — HealthLog has no separate - // resting-pulse field, so we accept the noise rather than skip. - const dailyMood = new Map(); - for (const row of moodRows) { - const key = userDayKey(row.moodLoggedAt, userTz); - const list = dailyMood.get(key) ?? []; - list.push(row.score); - dailyMood.set(key, list); - } - const dailyPulse = new Map(); - for (const row of pulseRows) { - const key = userDayKey(row.measuredAt, userTz); - const list = dailyPulse.get(key) ?? []; - list.push(row.value); - dailyPulse.set(key, list); - } - const moodPulsePairs: Array<{ - date: Date; - mood: number; - restingPulse: number; - }> = []; - for (const [key, moodScores] of dailyMood.entries()) { - const pulseValues = dailyPulse.get(key); - if (!pulseValues || pulseValues.length === 0) continue; - const meanMood = moodScores.reduce((s, v) => s + v, 0) / moodScores.length; - const meanPulse = - pulseValues.reduce((s, v) => s + v, 0) / pulseValues.length; - moodPulsePairs.push({ - date: dateFromDayKey(key), - mood: meanMood, - restingPulse: meanPulse, - }); - } - const moodPulse = correlateMoodPulse({ daily: moodPulsePairs }); - - // ── Hypothesis 3: Weight × weekday ────────────────────────── - // 0 = Monday … 6 = Sunday. ISO weekday minus 1. - // - // v1.4.25 W10 reconcile (Code-H1) — buckets the weekday by the - // user's display tz, matching W7's per-user-tz threading that the - // BP, mood, sleep, and pulse aggregators above already honour. The - // pre-W10 helper was pinned to `Europe/Berlin` and would land a - // user-local 23:30 weight reading from `Pacific/Auckland` under - // Sunday's bucket instead of Monday's — Pearson on the wrong - // weekday column. - const weightWeekdayPairs: Array<{ weekday: number; weight: number }> = []; - for (const row of weightRows) { - const isoWeekday = isoWeekdayInTz(row.measuredAt, userTz); // 1..7, 1=Mon - weightWeekdayPairs.push({ - weekday: isoWeekday - 1, - weight: row.value, - }); - } - const weightWeekday = correlateWeightWeekday({ daily: weightWeekdayPairs }); - - // Annotate so admin observability can attribute coverage to the - // corresponding wide-event rather than chasing it via DB queries. - annotate({ - meta: { - correlations: { - bpCompliance: bpCompliance.status, - moodPulse: moodPulse.status, - weightWeekday: weightWeekday.status, - }, - }, - }); - - return { bpCompliance, moodPulse, weightWeekday }; -} +// v1.4.37 W2 — `computeCorrelationHypotheses` body relocated to +// `src/lib/analytics/correlations-fast-path.ts` so the probe-gated +// rollup / live dispatcher can be unit-tested independently of the +// route. The new helper also tightens the scan window to 28 days and +// emits a sentinel on `meta.correlations.window_days` / `degraded`. /** * v1.4.23 Sr-H1 — paged read of every Measurement of a given type for @@ -761,56 +615,12 @@ async function fetchMeasurementSeriesChunked( return out; } -// v1.4.22 W5 reconcile (Code-MED-3) — `berlinDayKey()` lifted to -// `src/lib/analytics/berlin-day.ts` so the targets route's sparkline -// bucketing shares the same Europe/Berlin contract. The -// `weekday: "short"` formatter still lives here because it's only -// used by `isoWeekdayInTz()` below. -// -// v1.4.25 W10 reconcile (Code-H1) — formatter is now per-tz so the -// weight-weekday correlator honours the same per-user-tz contract the -// BP/mood/pulse aggregators above already use. Memoised by tz so the -// formatter is built once per unique timezone per process (tz strings -// rarely change at runtime; the cache is bounded by the IANA database -// to a few hundred entries even in the worst case). -const WEEKDAY_FORMATTER_CACHE = new Map(); -function getWeekdayFormatter(timeZone: string): Intl.DateTimeFormat { - let formatter = WEEKDAY_FORMATTER_CACHE.get(timeZone); - if (!formatter) { - formatter = new Intl.DateTimeFormat("en-CA", { - timeZone, - year: "numeric", - month: "2-digit", - day: "2-digit", - weekday: "short", - }); - WEEKDAY_FORMATTER_CACHE.set(timeZone, formatter); - } - return formatter; -} - -function dateFromDayKey(key: string): Date { - // Anchor to UTC midnight — the date is a sortable bucket label rather - // than a wall-clock timestamp, so DST drift is irrelevant. The tz - // info is already baked into the key (built via `userDayKey` upstream). - return new Date(`${key}T00:00:00.000Z`); -} - -const ISO_WEEKDAY: Record = { - Mon: 1, - Tue: 2, - Wed: 3, - Thu: 4, - Fri: 5, - Sat: 6, - Sun: 7, -}; - -function isoWeekdayInTz(d: Date, timeZone: string): number { - const parts = getWeekdayFormatter(timeZone).formatToParts(d); - const weekday = parts.find((p) => p.type === "weekday")?.value ?? "Mon"; - return ISO_WEEKDAY[weekday] ?? 1; -} +// v1.4.37 W2 — `WEEKDAY_FORMATTER_CACHE`, `getWeekdayFormatter`, +// `dateFromDayKey`, `ISO_WEEKDAY`, and `isoWeekdayInTz` relocated +// alongside the correlation runner in +// `src/lib/analytics/correlations-fast-path.ts`. They were only ever +// used by the weight-weekday + day-key helpers inside the old inline +// correlation builder. // v1.4.37 W2 — `mapMeasurementSourceToLabel`, `uniqueComponentSources`, diff --git a/src/lib/analytics/__tests__/correlations-fast-path.test.ts b/src/lib/analytics/__tests__/correlations-fast-path.test.ts new file mode 100644 index 000000000..267f0b4ff --- /dev/null +++ b/src/lib/analytics/__tests__/correlations-fast-path.test.ts @@ -0,0 +1,197 @@ +/** + * v1.4.37 W2 — unit pin for the probe-gated correlation runner. + * + * Mocks Prisma + the rollup-coverage probe so the path-selection + * contract is verifiable without a Postgres container. Real-data + * coverage of the actual Pearson + ANOVA math lives in + * `correlations.test.ts`; this file only checks WHICH read path the + * runner takes and that the sentinel annotate carries the window. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/db", () => ({ + prisma: { + measurement: { findMany: vi.fn() }, + measurementRollup: { findMany: vi.fn() }, + moodEntry: { findMany: vi.fn() }, + medicationIntakeEvent: { findMany: vi.fn() }, + }, +})); + +vi.mock("@/lib/logging/context", () => ({ + annotate: vi.fn(), +})); + +vi.mock("@/lib/measurements/rollup-coverage", () => ({ + isFullyCovered: vi.fn(), + probeRollupCoverage: vi.fn(), +})); + +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { + isFullyCovered, + probeRollupCoverage, +} from "@/lib/measurements/rollup-coverage"; +import { + CORRELATION_WINDOW_DAYS, + computeCorrelationHypothesesFastPath, +} from "../correlations-fast-path"; + +const MEASUREMENT_FIND_MANY = prisma.measurement.findMany as unknown as ReturnType; +const ROLLUP_FIND_MANY = prisma.measurementRollup.findMany as unknown as ReturnType; +const MOOD_FIND_MANY = prisma.moodEntry.findMany as unknown as ReturnType; +const INTAKE_FIND_MANY = prisma.medicationIntakeEvent.findMany as unknown as ReturnType; +const PROBE = probeRollupCoverage as unknown as ReturnType; +const FULLY_COVERED = isFullyCovered as unknown as ReturnType; +const ANNOTATE = annotate as unknown as ReturnType; + +beforeEach(() => { + MEASUREMENT_FIND_MANY.mockReset(); + ROLLUP_FIND_MANY.mockReset(); + MOOD_FIND_MANY.mockReset(); + INTAKE_FIND_MANY.mockReset(); + PROBE.mockReset(); + FULLY_COVERED.mockReset(); + ANNOTATE.mockReset(); + + MOOD_FIND_MANY.mockResolvedValue([]); + INTAKE_FIND_MANY.mockResolvedValue([]); + ROLLUP_FIND_MANY.mockResolvedValue([]); + MEASUREMENT_FIND_MANY.mockResolvedValue([]); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("computeCorrelationHypothesesFastPath", () => { + it("scans the trailing 28 days (CORRELATION_WINDOW_DAYS)", () => { + expect(CORRELATION_WINDOW_DAYS).toBe(28); + }); + + describe("rollup-fast-path — SYS / PULSE / WEIGHT covered", () => { + it("reads per-day means from measurement_rollups, never from measurements", async () => { + const coverage = new Map([ + ["BLOOD_PRESSURE_SYS", true], + ["PULSE", true], + ["WEIGHT", true], + ]); + FULLY_COVERED.mockReturnValue(true); + PROBE.mockResolvedValue(coverage); + + ROLLUP_FIND_MANY + .mockResolvedValueOnce([]) // SYS buckets + .mockResolvedValueOnce([]) // PULSE buckets + .mockResolvedValueOnce([]); // WEIGHT buckets + + const result = await computeCorrelationHypothesesFastPath({ + userId: "user-rollup", + userTz: "Europe/Berlin", + now: new Date("2026-05-17T12:00:00.000Z"), + coverage, + }); + + expect(result.path).toBe("rollup"); + expect(result.windowDays).toBe(28); + expect(result.degraded).toBe(false); + // No raw measurement reads when on rollup-fast-path. + expect(MEASUREMENT_FIND_MANY).not.toHaveBeenCalled(); + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(3); + // Mood + intake reads always live. + expect(MOOD_FIND_MANY).toHaveBeenCalledTimes(1); + expect(INTAKE_FIND_MANY).toHaveBeenCalledTimes(1); + }); + + it("pins path:rollup + window_days:28 on the annotate", async () => { + const coverage = new Map([ + ["BLOOD_PRESSURE_SYS", true], + ["PULSE", true], + ["WEIGHT", true], + ]); + FULLY_COVERED.mockReturnValue(true); + PROBE.mockResolvedValue(coverage); + ROLLUP_FIND_MANY.mockResolvedValue([]); + + await computeCorrelationHypothesesFastPath({ + userId: "user-rollup", + userTz: "Europe/Berlin", + now: new Date("2026-05-17T12:00:00.000Z"), + coverage, + }); + + const calls = ANNOTATE.mock.calls.map((c) => c[0]); + const corrCall = calls.find((c) => c?.meta?.correlations !== undefined); + expect(corrCall).toBeDefined(); + expect(corrCall.meta.correlations.path).toBe("rollup"); + expect(corrCall.meta.correlations.window_days).toBe(28); + expect(corrCall.meta.correlations.degraded).toBe(false); + }); + }); + + describe("live fallback", () => { + it("falls back when SYS coverage is missing", async () => { + const coverage = new Map([ + ["BLOOD_PRESSURE_SYS", false], + ["PULSE", true], + ["WEIGHT", true], + ]); + FULLY_COVERED.mockReturnValue(false); + PROBE.mockResolvedValue(coverage); + + MEASUREMENT_FIND_MANY + .mockResolvedValueOnce([]) // SYS rows + .mockResolvedValueOnce([]) // PULSE rows + .mockResolvedValueOnce([]); // WEIGHT rows + + const result = await computeCorrelationHypothesesFastPath({ + userId: "user-partial", + userTz: "Europe/Berlin", + now: new Date("2026-05-17T12:00:00.000Z"), + coverage, + }); + + expect(result.path).toBe("live"); + expect(MEASUREMENT_FIND_MANY).toHaveBeenCalledTimes(3); + expect(ROLLUP_FIND_MANY).not.toHaveBeenCalled(); + }); + + it("pins path:live on the annotate", async () => { + const coverage = new Map([ + ["BLOOD_PRESSURE_SYS", false], + ]); + FULLY_COVERED.mockReturnValue(false); + PROBE.mockResolvedValue(coverage); + MEASUREMENT_FIND_MANY.mockResolvedValue([]); + + await computeCorrelationHypothesesFastPath({ + userId: "user-partial", + userTz: "Europe/Berlin", + now: new Date("2026-05-17T12:00:00.000Z"), + coverage, + }); + + const calls = ANNOTATE.mock.calls.map((c) => c[0]); + const corrCall = calls.find((c) => c?.meta?.correlations !== undefined); + expect(corrCall).toBeDefined(); + expect(corrCall.meta.correlations.path).toBe("live"); + }); + }); + + describe("coverage probing", () => { + it("probes coverage when the caller omits the map", async () => { + const coverage = new Map([]); + FULLY_COVERED.mockReturnValue(false); + PROBE.mockResolvedValue(coverage); + MEASUREMENT_FIND_MANY.mockResolvedValue([]); + + await computeCorrelationHypothesesFastPath({ + userId: "user-fresh", + userTz: "Europe/Berlin", + now: new Date("2026-05-17T12:00:00.000Z"), + }); + + expect(PROBE).toHaveBeenCalledWith("user-fresh"); + }); + }); +}); diff --git a/src/lib/analytics/correlations-fast-path.ts b/src/lib/analytics/correlations-fast-path.ts new file mode 100644 index 000000000..8b5cc78d0 --- /dev/null +++ b/src/lib/analytics/correlations-fast-path.ts @@ -0,0 +1,360 @@ +/** + * v1.4.37 W2 — probe-gated correlation hypothesis runner. + * + * The three pre-defined hypotheses (BP × medication compliance, mood × + * resting pulse, weight × weekday) need PAIRED day-level samples + * across two streams. The pairing logic is not linearly composable + * from DAY rollup buckets the way `count / min / max / mean` is, but + * the only inputs the runners consume from the measurement stream are + * **per-day means** for the three measurement types (SYS, PULSE, + * WEIGHT). The rollup table carries those means as a first-class + * column — when DAY-bucket coverage is present we can hydrate the + * per-day-mean maps directly from `measurement_rollups` and skip the + * raw chunked findMany walks against `measurements` entirely. + * + * Read shape (v1.4.37) + * -------------------- + * 1. **Tightened window** — the previous helper read the trailing 30 + * days. The v1.4.37 helper reads **28 days**. The Pearson + + * ANOVA runners require `n >= 20` paired samples (v1.4.23 H6) so + * 28 still satisfies the surface gate; the 2-day tighten removes + * ~7 % of the cold-pool read volume and lines the correlation + * window up with the dashboard's "trailing 4 weeks" tile shorthand. + * The meta-dict annotate carries `window_days: 28` as a sentinel + * so the UI / docs can quote the truthful window. + * + * 2. **Probe-gated SYS / PULSE / WEIGHT** — when DAY-bucket coverage + * exists for all three types the runner hydrates the per-day-mean + * maps from `measurement_rollups`. Each rollup row already + * carries the day's `mean` value computed by Postgres' `AVG`, so + * the pairing logic is byte-for-byte identical to the live path's + * "average the day's raw values" step. + * + * 3. **Mood + medication intake** — no rollup equivalent today; both + * reads stay live regardless of coverage. The mood read is small + * (per-day entries, bounded by the 28-day window) and the + * medication intake read scales with active medications × 28 days + * so neither dominates the cost. + * + * 4. **Best-effort sentinel** — the route surface advertises a + * 28-day correlation window. Older data (> 28 days) is + * intentionally NOT scanned on the cold critical path. The + * runner emits `degraded: false` on the happy path; downstream + * consumers can extend the window via a separate dedicated + * endpoint when richer historical context is needed. + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { readRollupBuckets } from "@/lib/measurements/rollups"; +import { + isFullyCovered, + probeRollupCoverage, + type RollupCoverageMap, +} from "@/lib/measurements/rollup-coverage"; +import { + correlateBpCompliance, + correlateMoodPulse, + correlateWeightWeekday, + type CorrelationResult, +} from "@/lib/insights/correlations"; +import { userDayKey } from "@/lib/tz/resolver"; + +/** + * v1.4.37 W2 — cold-path correlation window. Trim from 30 to 28 days + * keeps the `n >= 20` Pearson gate comfortably reachable while shaving + * two days off every read the runner issues. Kept as a constant so the + * sentinel annotate and the SQL/Prisma WHERE clauses stay in lockstep. + */ +export const CORRELATION_WINDOW_DAYS = 28; + +const CHUNK = 5000; + +interface ChunkedRow { + measuredAt: Date; + value: number; +} + +interface MoodRow { + score: number; + moodLoggedAt: Date; +} + +interface IntakeRow { + scheduledFor: Date; + takenAt: Date | null; + skipped: boolean; +} + +export interface CorrelationHypothesesResult { + bpCompliance: CorrelationResult; + moodPulse: CorrelationResult; + weightWeekday: CorrelationResult; + /** + * Sentinel — `true` when the runner's scan window was tightened + * below the canonical surface (e.g. fell back to a partial read). + * Today this is always `false`: the 28-day window is the canonical + * surface. Reserved for a future "best-effort under load shedding" + * branch. + */ + degraded: boolean; + /** Window the runner actually scanned, in days. */ + windowDays: number; + /** `"rollup"` when SYS / PULSE / WEIGHT rode the rollup-fast-path. */ + path: "rollup" | "live"; +} + +export interface CorrelationsFastPathInput { + userId: string; + userTz: string; + now: Date; + coverage?: RollupCoverageMap; +} + +export async function computeCorrelationHypothesesFastPath( + input: CorrelationsFastPathInput, +): Promise { + const DAY_MS = 24 * 60 * 60 * 1000; + const { userId, userTz, now } = input; + const since = new Date(now.getTime() - CORRELATION_WINDOW_DAYS * DAY_MS); + + const coverage = input.coverage ?? (await probeRollupCoverage(userId)); + + // The three measurement-derived streams ride the rollup-fast-path + // only when EVERY type the user has logged is covered AND each of + // SYS / PULSE / WEIGHT is in the coverage map. Partial coverage + // falls back to the live path so a brand-new metric type doesn't + // make the correlation card vanish. + const measurementsOnRollups = + isFullyCovered(coverage) && + coverage.get("BLOOD_PRESSURE_SYS") === true && + coverage.get("PULSE") === true && + coverage.get("WEIGHT") === true; + + let dailySysMean: Map; + let dailyPulseMean: Map; + let weightRows: ChunkedRow[]; + + if (measurementsOnRollups) { + // Rollup-fast-path. One indexed read per type against + // `measurement_rollups`, projecting the per-day mean directly. + const [sysBuckets, pulseBuckets, weightBuckets] = await Promise.all([ + readRollupBuckets(userId, "BLOOD_PRESSURE_SYS", "DAY", since, now), + readRollupBuckets(userId, "PULSE", "DAY", since, now), + readRollupBuckets(userId, "WEIGHT", "DAY", since, now), + ]); + dailySysMean = new Map(); + for (const b of sysBuckets) { + dailySysMean.set(userDayKey(b.bucketStart, userTz), b.mean); + } + dailyPulseMean = new Map(); + for (const b of pulseBuckets) { + dailyPulseMean.set(userDayKey(b.bucketStart, userTz), b.mean); + } + // The weekday hypothesis needs the per-event weight series for + // ANOVA — the runner expects one row per measurement so the group + // counts reflect reality. Hydrate from buckets by emitting one + // pseudo-row per bucket carrying the day's mean. For weight the + // typical usage is one weigh-in per day, so the rollup shape is + // already aligned; if a power user logged 3 weigh-ins on a single + // Monday the rollup collapses them into one mean, which is the + // statistically correct "this Monday's weight" datapoint anyway. + weightRows = weightBuckets.map((b) => ({ + measuredAt: b.bucketStart, + value: b.mean, + })); + } else { + // Live fallback — chunked reads against the measurements table. + const [sysRows, pulseRows, weightRowsLive] = await Promise.all([ + fetchSeriesChunked(userId, "BLOOD_PRESSURE_SYS", since), + fetchSeriesChunked(userId, "PULSE", since), + fetchSeriesChunked(userId, "WEIGHT", since), + ]); + dailySysMean = new Map(); + const sysBuckets = new Map(); + for (const r of sysRows) { + const key = userDayKey(r.measuredAt, userTz); + const list = sysBuckets.get(key) ?? []; + list.push(r.value); + sysBuckets.set(key, list); + } + for (const [key, list] of sysBuckets.entries()) { + const sum = list.reduce((s, v) => s + v, 0); + dailySysMean.set(key, sum / list.length); + } + dailyPulseMean = new Map(); + const pulseBuckets = new Map(); + for (const r of pulseRows) { + const key = userDayKey(r.measuredAt, userTz); + const list = pulseBuckets.get(key) ?? []; + list.push(r.value); + pulseBuckets.set(key, list); + } + for (const [key, list] of pulseBuckets.entries()) { + const sum = list.reduce((s, v) => s + v, 0); + dailyPulseMean.set(key, sum / list.length); + } + weightRows = weightRowsLive; + } + + // Mood + intake — always live. Both reads are small (28-day window, + // narrow projections) and neither has a rollup equivalent today. + const [moodRows, intakeRows] = await Promise.all([ + prisma.moodEntry.findMany({ + where: { userId, moodLoggedAt: { gte: since } }, + select: { score: true, moodLoggedAt: true }, + }) as Promise, + prisma.medicationIntakeEvent.findMany({ + where: { userId, scheduledFor: { gte: since } }, + select: { scheduledFor: true, takenAt: true, skipped: true }, + }) as Promise, + ]); + + // ── Hypothesis 1: BP × medication compliance ──────────────── + const dailyCompliance = new Map< + string, + { expected: number; taken: number } + >(); + for (const event of intakeRows) { + const key = userDayKey(event.scheduledFor, userTz); + const slot = dailyCompliance.get(key) ?? { expected: 0, taken: 0 }; + slot.expected += 1; + if (event.takenAt && !event.skipped) slot.taken += 1; + dailyCompliance.set(key, slot); + } + + const bpCompliancePairs: Array<{ + date: Date; + systolic: number; + compliancePct: number; + }> = []; + for (const [key, meanSys] of dailySysMean.entries()) { + const slot = dailyCompliance.get(key); + if (!slot || slot.expected === 0) continue; + const compliancePct = (slot.taken / slot.expected) * 100; + bpCompliancePairs.push({ + date: dateFromDayKey(key), + systolic: meanSys, + compliancePct, + }); + } + const bpCompliance = correlateBpCompliance({ daily: bpCompliancePairs }); + + // ── Hypothesis 2: Mood × resting pulse ────────────────────── + const dailyMood = new Map(); + for (const row of moodRows) { + const key = userDayKey(row.moodLoggedAt, userTz); + const list = dailyMood.get(key) ?? []; + list.push(row.score); + dailyMood.set(key, list); + } + const moodPulsePairs: Array<{ + date: Date; + mood: number; + restingPulse: number; + }> = []; + for (const [key, moodScores] of dailyMood.entries()) { + const meanPulse = dailyPulseMean.get(key); + if (meanPulse === undefined) continue; + const meanMood = moodScores.reduce((s, v) => s + v, 0) / moodScores.length; + moodPulsePairs.push({ + date: dateFromDayKey(key), + mood: meanMood, + restingPulse: meanPulse, + }); + } + const moodPulse = correlateMoodPulse({ daily: moodPulsePairs }); + + // ── Hypothesis 3: Weight × weekday ────────────────────────── + const weightWeekdayPairs: Array<{ weekday: number; weight: number }> = []; + for (const row of weightRows) { + const isoWeekday = isoWeekdayInTz(row.measuredAt, userTz); + weightWeekdayPairs.push({ + weekday: isoWeekday - 1, + weight: row.value, + }); + } + const weightWeekday = correlateWeightWeekday({ daily: weightWeekdayPairs }); + + annotate({ + meta: { + correlations: { + bpCompliance: bpCompliance.status, + moodPulse: moodPulse.status, + weightWeekday: weightWeekday.status, + path: measurementsOnRollups ? "rollup" : "live", + window_days: CORRELATION_WINDOW_DAYS, + degraded: false, + }, + }, + }); + + return { + bpCompliance, + moodPulse, + weightWeekday, + degraded: false, + windowDays: CORRELATION_WINDOW_DAYS, + path: measurementsOnRollups ? "rollup" : "live", + }; +} + +async function fetchSeriesChunked( + userId: string, + type: "BLOOD_PRESSURE_SYS" | "PULSE" | "WEIGHT", + since: Date, +): Promise { + const out: ChunkedRow[] = []; + let cursorId: string | undefined; + for (let page = 0; page < 1000; page++) { + const chunk = await prisma.measurement.findMany({ + where: { userId, type, measuredAt: { gte: since } }, + orderBy: [{ measuredAt: "asc" }, { id: "asc" }], + select: { id: true, measuredAt: true, value: true }, + take: CHUNK, + ...(cursorId ? { cursor: { id: cursorId }, skip: 1 } : {}), + }); + if (chunk.length === 0) break; + for (const row of chunk) { + out.push({ measuredAt: row.measuredAt, value: row.value }); + } + if (chunk.length < CHUNK) break; + cursorId = chunk[chunk.length - 1].id; + } + return out; +} + +function dateFromDayKey(key: string): Date { + return new Date(`${key}T00:00:00.000Z`); +} + +const WEEKDAY_FORMATTER_CACHE = new Map(); +function getWeekdayFormatter(timeZone: string): Intl.DateTimeFormat { + let formatter = WEEKDAY_FORMATTER_CACHE.get(timeZone); + if (!formatter) { + formatter = new Intl.DateTimeFormat("en-CA", { + timeZone, + year: "numeric", + month: "2-digit", + day: "2-digit", + weekday: "short", + }); + WEEKDAY_FORMATTER_CACHE.set(timeZone, formatter); + } + return formatter; +} + +const ISO_WEEKDAY: Record = { + Mon: 1, + Tue: 2, + Wed: 3, + Thu: 4, + Fri: 5, + Sat: 6, + Sun: 7, +}; + +function isoWeekdayInTz(d: Date, timeZone: string): number { + const parts = getWeekdayFormatter(timeZone).formatToParts(d); + const weekday = parts.find((p) => p.type === "weekday")?.value ?? "Mon"; + return ISO_WEEKDAY[weekday] ?? 1; +}