diff --git a/src/app/api/insights/correlations/route.ts b/src/app/api/insights/correlations/route.ts index 4256e6274..e741f1607 100644 --- a/src/app/api/insights/correlations/route.ts +++ b/src/app/api/insights/correlations/route.ts @@ -38,11 +38,15 @@ import { type NamedSeries, } from "@/lib/insights/correlation-discovery"; import { + buildMeasurementDailySeries, fetchComplianceSeries, fetchEnvironmentSeries, fetchLabDraws, fetchSymptomSeries, + toDailyMeans, + type MeasurementSeriesRow, } from "@/lib/insights/correlation-channel-series"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; export const dynamic = "force-dynamic"; @@ -58,25 +62,6 @@ function tzDayKey(at: Date, tz: string): string { return `${year}-${mm}-${dd}`; } -/** Collapse rows to per-day means keyed in the user's tz. */ -function toDailyMeans( - rows: Array<{ value: number; at: Date }>, - tz: string, -): DailySeriesPoint[] { - const byDay = new Map(); - for (const r of rows) { - if (!Number.isFinite(r.value)) continue; - const day = tzDayKey(r.at, tz); - const acc = byDay.get(day) ?? { sum: 0, count: 0 }; - acc.sum += r.value; - acc.count += 1; - 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)); -} - export const GET = apiHandler(async () => { const { user } = await requireAuth(); @@ -112,7 +97,11 @@ export const GET = apiHandler(async () => { DISCOVERY_OUTCOMES, ) as MeasurementType[]; - const [measurements, moodEntries] = await Promise.all([ + // v1.29.6 — `source` + `deviceType` + `sleepStage` feed + // `buildMeasurementDailySeries`'s per-type grain resolution below (source + // collapse for cumulative types, per-night reconstruction for sleep); the + // rest of the channels never look at them. + const [measurements, moodEntries, priorityJson] = await Promise.all([ prisma.measurement.findMany({ where: { userId: user.id, @@ -122,7 +111,14 @@ export const GET = apiHandler(async () => { }, orderBy: { measuredAt: "asc" }, take: 20000, - select: { type: true, value: true, measuredAt: true }, + select: { + type: true, + value: true, + measuredAt: true, + source: true, + deviceType: true, + sleepStage: true, + }, }), prisma.moodEntry.findMany({ where: { userId: user.id, deletedAt: null, moodLoggedAt: { gte: since } }, @@ -130,15 +126,19 @@ export const GET = apiHandler(async () => { take: 5000, select: { score: true, moodLoggedAt: true }, }), + loadUserSourcePriority(user.id), ]); - const measurementsByType = new Map< - string, - Array<{ value: number; at: Date }> - >(); + const measurementsByType = new Map(); for (const m of measurements) { const list = measurementsByType.get(m.type) ?? []; - list.push({ value: m.value, at: m.measuredAt }); + list.push({ + value: m.value, + at: m.measuredAt, + source: m.source, + deviceType: m.deviceType, + sleepStage: m.sleepStage, + }); measurementsByType.set(m.type, list); } @@ -167,7 +167,12 @@ export const GET = apiHandler(async () => { const points = (key: string): DailySeriesPoint[] => key === "MOOD" ? moodDaily - : toDailyMeans(measurementsByType.get(key) ?? [], tz); + : buildMeasurementDailySeries( + key as MeasurementType, + measurementsByType.get(key) ?? [], + tz, + priorityJson, + ); const series: NamedSeries[] = []; for (const key of DISCOVERY_BEHAVIOURS) { diff --git a/src/app/api/measurements/__tests__/group-by-day.test.ts b/src/app/api/measurements/__tests__/group-by-day.test.ts index 3d193d221..d22a10677 100644 --- a/src/app/api/measurements/__tests__/group-by-day.test.ts +++ b/src/app/api/measurements/__tests__/group-by-day.test.ts @@ -27,6 +27,11 @@ vi.mock("@/lib/db", () => ({ measurementRollup: { findMany: vi.fn(), }, + // v1.29.6 — the groupBy=day branch now runs its raw rows through + // `pickCanonicalSourceRows`, which loads the user's source-priority + // blob via `loadUserSourcePriority` (prisma.user.findUnique). `null` + // resolves to the default ladders. + user: { findUnique: vi.fn() }, // v1.4.43 W6 — validation-failed paths write an audit breadcrumb. auditLog: { create: vi.fn().mockResolvedValue({}) }, $queryRaw: vi.fn(), @@ -75,6 +80,9 @@ beforeEach(() => { // write; the route swallows rejections so the test only needs a // resolved mock to keep the catch-block silent. vi.mocked(prisma.auditLog.create).mockResolvedValue({} as never); + // v1.29.6 — null source-priority blob → default rank ladders for the + // groupBy=day canonical-source collapse. + vi.mocked(prisma.user.findUnique).mockResolvedValue(null as never); }); describe("GET /api/measurements — groupBy=day (W7c collapsed list)", () => { @@ -165,6 +173,80 @@ describe("GET /api/measurements — groupBy=day (W7c collapsed list)", () => { expect(day15.measuredAt).toContain("2026-05-15T10:00:00"); }); + // v1.29.6 — the raw-row scan must read the MOST RECENT window before + // bucketing, mirroring the workouts-list ordering fix (05f80ad37): the + // canonical/day-bucket picker doesn't reorder its input, but the + // underlying `orderBy: asc` + `take: limit` sliced the OLDEST page of + // rows before the day-bucketing loop ran — on a long history the most + // recent days never appeared at all. + it("scans the raw rows newest-first before bucketing", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + + await GET(getRequest("type=ACTIVITY_STEPS&groupBy=day")); + + const call = vi.mocked(prisma.measurement.findMany).mock.calls[0]?.[0] as { + orderBy?: { measuredAt?: "asc" | "desc" }; + }; + expect(call.orderBy).toEqual({ measuredAt: "desc" }); + }); + + // v1.29.6 — a day with both an Apple Health AND a Withings write for the + // same cumulative metric must collapse to the ladder-canonical source + // before summing, exactly like the dashboard rollup tile + // (`collapseRollupRowsBySource`) does. Without the collapse, 9000 + // Apple-Health steps + 8800 Withings steps summed to a fabricated 17800 — + // a number the tile never showed. + it("collapses cross-source rows to the ladder-canonical source before summing", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + id: "m-apple", + type: "ACTIVITY_STEPS", + value: 9000, + unit: "steps", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-15T08:00:00.000Z"), + notes: null, + deviceType: null, + }, + { + id: "m-withings", + type: "ACTIVITY_STEPS", + value: 8800, + unit: "steps", + source: "WITHINGS", + measuredAt: new Date("2026-05-15T09:00:00.000Z"), + notes: null, + deviceType: null, + }, + ] as never); + + const res = await GET( + getRequest( + "type=ACTIVITY_STEPS&groupBy=day&from=2026-05-15T00:00:00Z&to=2026-05-16T00:00:00Z", + ), + ); + expect(res.status).toBe(200); + const json = (await res.json()) as { + data: { + measurements: Array<{ + dayKey: string; + value: number; + source: string; + sampleCount: number; + }>; + meta: { droppedDuplicates?: number }; + }; + }; + expect(json.data.measurements.length).toBe(1); + const [day] = json.data.measurements; + // APPLE_HEALTH beats WITHINGS on the default `steps` ladder — the + // Withings row drops out of the sum entirely rather than adding on top. + expect(day.value).toBe(9000); + expect(day.source).toBe("APPLE_HEALTH"); + expect(day.sampleCount).toBe(1); + expect(json.data.meta.droppedDuplicates).toBe(1); + }); + it("falls back to the legacy findMany path when groupBy=day is set on a non-cumulative type", async () => { vi.mocked(prisma.measurement.findMany).mockResolvedValue([ { diff --git a/src/app/api/measurements/route.ts b/src/app/api/measurements/route.ts index 50ec6ebd8..52fbac848 100644 --- a/src/app/api/measurements/route.ts +++ b/src/app/api/measurements/route.ts @@ -21,6 +21,8 @@ import { type AggregateGrain, } from "@/lib/measurements/range-aggregation"; import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; +import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority"; import { canonicalDailyTimestamp, dayKeyForUserTz, @@ -214,9 +216,17 @@ export const GET = apiHandler(async (request: NextRequest) => { user.timezone && user.timezone.length > 0 ? user.timezone : "Europe/Berlin"; + // v1.29.6 — DESC + take so a long history returns the MOST RECENT + // window of raw samples for bucketing. The previous `orderBy: asc` + // + `take: limit` sliced the OLDEST page of rows before the + // day-bucketing loop ran below — on an account with more than + // `limit` cumulative samples in its history, the buckets were built + // entirely from year-old rows and the recent days silently never + // appeared. Mirrors the workouts-list ordering fix (05f80ad37): + // re-establish "most recent" as the page the caller actually gets. const rows = await prisma.measurement.findMany({ where, - orderBy: { measuredAt: "asc" }, + orderBy: { measuredAt: "desc" }, // Per-sample row scan within the requested window — bounded by // the existing `limit` cap (default 100, max 5000) so a wide // multi-year window can still walk the whole range when the @@ -230,9 +240,34 @@ export const GET = apiHandler(async (request: NextRequest) => { source: true, measuredAt: true, notes: true, + deviceType: true, }, }); + // v1.29.6 — collapse overlapping sources to the ladder-canonical + // reading per day BEFORE summing. Without this, a day with both an + // Apple Health AND a Withings write for the same cumulative metric + // (e.g. steps) summed BOTH sources into one inflated total — a day + // the dashboard tile (which reads through the source-collapsed + // rollup, `collapseRollupRowsBySource` in + // `src/lib/rollups/measurement-read.ts`) showed correctly. Uses the + // same `metricKeyForType` ladder lookup the rollup collapse uses, so + // this list and the tile agree on which source wins. A type with no + // ladder (e.g. TIME_IN_DAYLIGHT, FALL_COUNT) has nothing to collapse + // against — every row passes through unchanged, matching the + // pass-through fallback used at the other `pickCanonicalSourceRows` + // call sites. + const metricKey = metricKeyForType(type as MeasurementType); + const priorityJson = await loadUserSourcePriority(user.id); + const canonicalRows = metricKey + ? pickCanonicalSourceRows( + rows.map((r) => ({ ...r, type: r.type as MeasurementType })), + metricKey, + priorityJson, + (d) => dayKeyForUserTz(d, tz), + ).canonicalRows + : rows; + // Group by the user's calendar day, sum values, keep a representative // unit/source from the bucket's first row so the rendered row carries // a sensible label even when the bucket mixes sources. @@ -245,7 +280,7 @@ export const GET = apiHandler(async (request: NextRequest) => { source: string; }; const buckets = new Map(); - for (const row of rows) { + for (const row of canonicalRows) { const key = dayKeyForUserTz(row.measuredAt, tz); const slot = buckets.get(key); if (!slot) { @@ -293,6 +328,7 @@ export const GET = apiHandler(async (request: NextRequest) => { type, groupBy: "day", rowsScanned: rows.length, + droppedDuplicates: rows.length - canonicalRows.length, }, }); return apiSuccess({ @@ -302,6 +338,7 @@ export const GET = apiHandler(async (request: NextRequest) => { limit, offset: 0, groupBy: "day", + droppedDuplicates: rows.length - canonicalRows.length, }, }); } diff --git a/src/lib/ai/coach/tools/correlations-read.ts b/src/lib/ai/coach/tools/correlations-read.ts index ebeb3e2d5..f1d9c8d85 100644 --- a/src/lib/ai/coach/tools/correlations-read.ts +++ b/src/lib/ai/coach/tools/correlations-read.ts @@ -35,14 +35,17 @@ import { EARLY_WINDOW_DAYS, MEDICATION_COMPLIANCE_CHANNEL_KEY, SYMPTOM_SEVERITY_CHANNEL_KEY, - type DailySeriesPoint, type NamedSeries, } from "@/lib/insights/correlation-discovery"; import { + buildMeasurementDailySeries, fetchComplianceSeries, fetchLabDraws, fetchSymptomSeries, + toDailyMeans, + type MeasurementSeriesRow, } from "@/lib/insights/correlation-channel-series"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import { computeCoincidentDeviation, loadBaselineProfile, @@ -127,25 +130,6 @@ function tzDayKey(at: Date, tz: string): string { return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`; } -/** Collapse rows to per-day means keyed in the user's tz. */ -function toDailyMeans( - rows: Array<{ value: number; at: Date }>, - tz: string, -): DailySeriesPoint[] { - const byDay = new Map(); - for (const r of rows) { - if (!Number.isFinite(r.value)) continue; - const day = tzDayKey(r.at, tz); - const acc = byDay.get(day) ?? { sum: 0, count: 0 }; - acc.sum += r.value; - acc.count += 1; - 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)); -} - /** Natural labels for the non-measurement channel keys (read cleanly in prose). */ const CHANNEL_LABELS: Record = { [MEDICATION_COMPLIANCE_CHANNEL_KEY]: "medication adherence", @@ -186,41 +170,68 @@ export async function readCoachCorrelations( DISCOVERY_OUTCOMES, ) as MeasurementType[]; - const [measurements, moodEntries, coincidentDerived] = await Promise.all([ - prisma.measurement.findMany({ - where: { - userId, - deletedAt: null, - measuredAt: { gte: since }, - type: { in: [...behaviourTypes, ...outcomeTypes] }, - }, - orderBy: { measuredAt: "asc" }, - take: 20000, - select: { type: true, value: true, measuredAt: true }, - }), - prisma.moodEntry.findMany({ - where: { userId, deletedAt: null, moodLoggedAt: { gte: since } }, - orderBy: { moodLoggedAt: "asc" }, - take: 5000, - select: { score: true, moodLoggedAt: true }, - }), - // Coincident-deviation is its own derived metric — fail-soft to null so a - // baseline hiccup never sinks the whole correlations read. D2-8: pass the - // user's tz so the "today" grouping matches the user's calendar day, not - // UTC's, before the fired flag is narrated as "out of band TODAY". - computeCoincidentDeviation(userId, profile, { tz }).catch(() => null), - ]); + // v1.29.6 — `source` + `deviceType` + `sleepStage` feed + // `buildMeasurementDailySeries`'s per-type grain resolution below (see + // the insights route's identical comment for why). + const [measurements, moodEntries, coincidentDerived, priorityJson] = + await Promise.all([ + prisma.measurement.findMany({ + where: { + userId, + deletedAt: null, + measuredAt: { gte: since }, + type: { in: [...behaviourTypes, ...outcomeTypes] }, + }, + orderBy: { measuredAt: "asc" }, + take: 20000, + select: { + type: true, + value: true, + measuredAt: true, + source: true, + deviceType: true, + sleepStage: true, + }, + }), + prisma.moodEntry.findMany({ + where: { userId, deletedAt: null, moodLoggedAt: { gte: since } }, + orderBy: { moodLoggedAt: "asc" }, + take: 5000, + select: { score: true, moodLoggedAt: true }, + }), + // Coincident-deviation is its own derived metric — fail-soft to null so a + // baseline hiccup never sinks the whole correlations read. D2-8: pass the + // user's tz so the "today" grouping matches the user's calendar day, not + // UTC's, before the fired flag is narrated as "out of band TODAY". + computeCoincidentDeviation(userId, profile, { tz }).catch(() => null), + loadUserSourcePriority(userId), + ]); - const byType = new Map>(); + const byType = new Map(); for (const m of measurements) { const list = byType.get(m.type) ?? []; - list.push({ value: m.value, at: m.measuredAt }); + list.push({ + value: m.value, + at: m.measuredAt, + source: m.source, + deviceType: m.deviceType, + sleepStage: m.sleepStage, + }); byType.set(m.type, list); } const moodDaily = toDailyMeans( moodEntries.map((e) => ({ value: e.score, at: e.moodLoggedAt })), tz, ); + const seriesPoints = (key: string) => + key === "MOOD" + ? moodDaily + : buildMeasurementDailySeries( + key as MeasurementType, + byType.get(key) ?? [], + tz, + priorityJson, + ); // The two non-measurement, non-mood channels come from their own sources // (the dose-history ledger + the illness day-log), folded in below exactly @@ -241,18 +252,14 @@ export async function readCoachCorrelations( } else if (key === SYMPTOM_SEVERITY_CHANNEL_KEY) { series.push({ ...symptomSeries, role: "behaviour" }); } else { - const points = - key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz); - series.push({ key, role: "behaviour", points }); + series.push({ key, role: "behaviour", points: seriesPoints(key) }); } } for (const key of DISCOVERY_OUTCOMES) { if (key === SYMPTOM_SEVERITY_CHANNEL_KEY) { series.push({ ...symptomSeries, role: "outcome" }); } else { - const points = - key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz); - series.push({ key, role: "outcome", points }); + series.push({ key, role: "outcome", points: seriesPoints(key) }); } } diff --git a/src/lib/insights/__tests__/correlation-channel-series.test.ts b/src/lib/insights/__tests__/correlation-channel-series.test.ts new file mode 100644 index 000000000..65b3dc833 --- /dev/null +++ b/src/lib/insights/__tests__/correlation-channel-series.test.ts @@ -0,0 +1,132 @@ +import { describe, it, expect } from "vitest"; +import type { SleepStage } from "@/generated/prisma/client"; +import { + buildMeasurementDailySeries, + toDailyMeans, + type MeasurementSeriesRow, +} from "@/lib/insights/correlation-channel-series"; + +/** Build a raw measurement row for `buildMeasurementDailySeries`. */ +function row( + iso: string, + value: number, + source: MeasurementSeriesRow["source"] = "APPLE_HEALTH", + sleepStage: SleepStage | null = null, + deviceType: string | null = null, +): MeasurementSeriesRow { + return { at: new Date(iso), value, source, deviceType, sleepStage }; +} + +describe("buildMeasurementDailySeries — grain consistency (v1.29.6)", () => { + it("sums (not averages) a cumulative type across mixed per-sample-chunk and drained daily-total rows", () => { + const rows: MeasurementSeriesRow[] = [ + // 2026-06-04: two ~mid-morning / afternoon per-sample chunks — the + // "not-yet-drained" shape. + row("2026-06-04T08:00:00.000Z", 350), + row("2026-06-04T14:00:00.000Z", 300), + // 2026-06-05: a single drained `stats:` daily total. + row("2026-06-05T12:00:00.000Z", 8400), + ]; + + const points = buildMeasurementDailySeries( + "ACTIVITY_STEPS", + rows, + "UTC", + null, + ); + + expect(points).toEqual([ + { day: "2026-06-04", value: 650 }, + { day: "2026-06-05", value: 8400 }, + ]); + + // The old `toDailyMeans` reduction would have produced meaningless + // blended figures — pin that the two grains disagree, so a future + // regression that reverts to `toDailyMeans` for this type is caught. + const meanPoints = toDailyMeans( + rows.map((r) => ({ value: r.value, at: r.at })), + "UTC", + ); + expect(meanPoints.find((p) => p.day === "2026-06-04")?.value).toBe(325); // mean of 350/300 + expect(meanPoints.find((p) => p.day === "2026-06-05")?.value).toBe(8400); // single row, coincides + }); + + it("collapses overlapping sources to the ladder-canonical reading before summing a cumulative type", () => { + const rows: MeasurementSeriesRow[] = [ + row("2026-06-04T08:00:00.000Z", 9000, "APPLE_HEALTH"), + row("2026-06-04T08:05:00.000Z", 8800, "WITHINGS"), + ]; + + const points = buildMeasurementDailySeries( + "ACTIVITY_STEPS", + rows, + "UTC", + null, + ); + + // Default `steps` ladder ranks APPLE_HEALTH above WITHINGS — the + // Withings row must drop out of the sum entirely, not add on top. + expect(points).toEqual([{ day: "2026-06-04", value: 9000 }]); + }); + + it("sums a night's per-stage segments into one total instead of averaging them", () => { + // One night: CORE 240 + DEEP 90 + REM 80 = 410 minutes asleep. + // `measuredAt` is the END of each segment; the reconstructor derives + // the start from `measuredAt - value minutes` (mirrors sleep-night.test.ts). + const rows: MeasurementSeriesRow[] = [ + row("2026-06-04T04:00:00.000Z", 240, "APPLE_HEALTH", "CORE"), + row("2026-06-04T05:30:00.000Z", 90, "APPLE_HEALTH", "DEEP"), + row("2026-06-04T07:00:00.000Z", 80, "APPLE_HEALTH", "REM"), + ]; + + const points = buildMeasurementDailySeries( + "SLEEP_DURATION", + rows, + "UTC", + null, + ); + + expect(points).toHaveLength(1); + // The pre-fix `toDailyMeans` reduction would have averaged the three + // segment durations (240 + 90 + 80) / 3 ≈ 137 — a number with no + // clinical meaning. The correct grain is the night's TOTAL. + expect(points[0].value).toBe(410); + expect(points[0].value).not.toBeCloseTo((240 + 90 + 80) / 3, 0); + }); + + it("collapses an overlapping-source night (WHOOP + Apple Health) to one canonical total, never double-counted", () => { + // Same night, two writers: WHOOP (wins the default `sleep` ladder) + // reports a shorter granular breakdown; Apple Health separately + // reports its own (different) total for the same night. Without the + // writer-dedup a naive per-row collapse could blend or double both. + const rows: MeasurementSeriesRow[] = [ + row("2026-06-04T04:00:00.000Z", 200, "WHOOP", "CORE"), + row("2026-06-04T05:00:00.000Z", 60, "WHOOP", "DEEP"), + row("2026-06-04T04:30:00.000Z", 300, "APPLE_HEALTH", "CORE"), + ]; + + const points = buildMeasurementDailySeries( + "SLEEP_DURATION", + rows, + "UTC", + null, + ); + + expect(points).toHaveLength(1); + // WHOOP wins the default `sleep` ladder — the night's total is the + // WHOOP-only sum (260), not a blend with Apple Health's 300 and not + // the two summed together (560). + expect(points[0].value).toBe(260); + }); + + it("keeps the MEAN grain for spot metrics (unchanged behaviour)", () => { + const rows: MeasurementSeriesRow[] = [ + row("2026-06-04T08:00:00.000Z", 60), + row("2026-06-04T20:00:00.000Z", 64), + ]; + + const points = buildMeasurementDailySeries("PULSE", rows, "UTC", null); + + expect(points).toEqual([{ day: "2026-06-04", value: 62 }]); + }); +}); diff --git a/src/lib/insights/correlation-channel-series.ts b/src/lib/insights/correlation-channel-series.ts index fa3d4f90e..378106dae 100644 --- a/src/lib/insights/correlation-channel-series.ts +++ b/src/lib/insights/correlation-channel-series.ts @@ -38,6 +38,18 @@ import { type SymptomEpisodeSpan, } from "@/lib/insights/correlation-series-builders"; import type { DoseHistoryRow } from "@/lib/medications/scheduling/dose-history"; +import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; +import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority"; +import { + reconstructSleepSessions, + pickMainNightAndNaps, + type SleepStageRow, +} from "@/lib/analytics/sleep-night"; +import type { + MeasurementSource, + MeasurementType, +} from "@/generated/prisma/client"; /** * v1.21.0 (FDREXTEND) — build the user's MEDICATION_COMPLIANCE daily series. @@ -244,6 +256,148 @@ function tzDayKey(at: Date, tz: string): string { return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`; } +/** + * v1.29.6 — collapse rows to per-day MEANS keyed in the user's tz. This is + * the correct grain for spot metrics (BP, glucose, HRV, resting HR, weight, + * mood) where a day's reduction is the average of its readings. Cumulative + * metrics and sleep must NOT use this — see `buildMeasurementDailySeries`. + * + * Hoisted here from the two former call sites (`/api/insights/correlations` + * and `metric-correlation-context.ts`) so both surfaces stay byte-identical + * instead of drifting as two independently-maintained copies — the same + * class of drift that let the cumulative/sleep grain bug below slip into + * one file and not the other. + */ +export function toDailyMeans( + rows: Array<{ value: number; at: Date }>, + tz: string, +): DailySeriesPoint[] { + const byDay = new Map(); + for (const r of rows) { + if (!Number.isFinite(r.value)) continue; + const day = tzDayKey(r.at, tz); + const acc = byDay.get(day) ?? { sum: 0, count: 0 }; + acc.sum += r.value; + acc.count += 1; + 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)); +} + +/** Minimum row shape `buildMeasurementDailySeries` needs per raw reading. */ +export interface MeasurementSeriesRow { + value: number; + at: Date; + source: MeasurementSource; + deviceType: string | null; + /** Only populated (and only consulted) for SLEEP_DURATION rows. */ + sleepStage: SleepStageRow["sleepStage"] | null; +} + +/** + * v1.29.6 — collapse one MeasurementType's raw rows to a single-grain daily + * series, keyed in the user's tz. Fixes a correlation-discovery distortion: + * `ACTIVITY_STEPS` and other cumulative HK types were being reduced with + * `toDailyMeans`, which blends per-sample chunk averages (~350 steps, from + * the not-yet-nightly-drained window) with drained `stats:` daily totals + * (~8400 steps) into one meaningless per-day figure. `SLEEP_DURATION` was + * averaging per-STAGE segment durations (~45 min) instead of summing a + * night's total time asleep, and without collapsing overlapping sources + * first (a WHOOP + Apple Health night double-counted). + * + * - `SLEEP_DURATION` → per-night TOTAL time asleep for the MAIN session + * (naps excluded, matching the dashboard/list convention), via + * `reconstructSleepSessions` — the same writer-dedup + per-night + * collapse the sleep list route and dashboard tile use. + * - `CUMULATIVE_HK_TYPES` (steps, active energy, distance, flights, + * daylight, falls) → source-collapsed per-day SUM, via + * `pickCanonicalSourceRows` (a type with no ladder passes every row + * through unchanged, matching the picker's documented fallback). + * - everything else → per-day MEAN via `toDailyMeans` (unchanged). + */ +export function buildMeasurementDailySeries( + type: MeasurementType, + rows: MeasurementSeriesRow[], + tz: string, + priorityJson: unknown, +): DailySeriesPoint[] { + if (type === "SLEEP_DURATION") { + return buildSleepDailySeries(rows, tz, priorityJson); + } + if (CUMULATIVE_HK_TYPES.has(type)) { + return buildCumulativeDailySeries(type, rows, tz, priorityJson); + } + return toDailyMeans( + rows.map((r) => ({ value: r.value, at: r.at })), + tz, + ); +} + +function buildSleepDailySeries( + rows: MeasurementSeriesRow[], + tz: string, + priorityJson: unknown, +): DailySeriesPoint[] { + const stageRows: SleepStageRow[] = rows.map((r) => ({ + value: r.value, + measuredAt: r.at, + sleepStage: r.sleepStage, + source: r.source, + deviceType: r.deviceType, + })); + const sessions = reconstructSleepSessions(stageRows, tz, priorityJson); + + const byNight = new Map(); + for (const s of sessions) { + const list = byNight.get(s.night) ?? []; + list.push(s); + byNight.set(s.night, list); + } + + const points: DailySeriesPoint[] = []; + for (const [night, nightSessions] of byNight) { + const { main } = pickMainNightAndNaps(nightSessions); + if (!main) continue; + points.push({ day: night, value: main.asleepMinutes }); + } + return points.sort((a, b) => (a.day < b.day ? -1 : 1)); +} + +function buildCumulativeDailySeries( + type: MeasurementType, + rows: MeasurementSeriesRow[], + tz: string, + priorityJson: unknown, +): DailySeriesPoint[] { + const metricKey = metricKeyForType(type); + const canonicalRows = metricKey + ? pickCanonicalSourceRows( + rows.map((r) => ({ + measuredAt: r.at, + source: r.source, + deviceType: r.deviceType, + type, + value: r.value, + })), + metricKey, + priorityJson, + (d) => tzDayKey(d, tz), + ).canonicalRows + : rows.map((r) => ({ measuredAt: r.at, value: r.value })); + + const byDay = new Map(); + for (const row of canonicalRows) { + if (!Number.isFinite(row.value)) continue; + const key = tzDayKey(row.measuredAt, tz); + byDay.set(key, (byDay.get(key) ?? 0) + row.value); + } + return [...byDay.entries()] + .map(([day, value]) => ({ day, value })) + .sort((a, b) => (a.day < b.day ? -1 : 1)); +} + /** * v1.22 — build the user's lab draws for the labs ↔ outcome correlation pass. * diff --git a/src/lib/insights/metric-correlation-context.ts b/src/lib/insights/metric-correlation-context.ts index 2d18cd396..c7557aec9 100644 --- a/src/lib/insights/metric-correlation-context.ts +++ b/src/lib/insights/metric-correlation-context.ts @@ -22,15 +22,19 @@ */ import type { MeasurementType } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; -import { wallClockInTz } from "@/lib/tz/wall-clock"; import { discoverCorrelations, discoveryMeasurementTypes, DISCOVERY_BEHAVIOURS, DISCOVERY_OUTCOMES, - type DailySeriesPoint, type NamedSeries, } from "@/lib/insights/correlation-discovery"; +import { + buildMeasurementDailySeries, + toDailyMeans, + type MeasurementSeriesRow, +} from "@/lib/insights/correlation-channel-series"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import type { RelevantCorrelation } from "@/lib/insights/assessment-context"; const MS_PER_DAY = 24 * 60 * 60 * 1000; @@ -55,31 +59,6 @@ function channelKeyForType(type: MeasurementType): string | null { return null; } -/** Day key (YYYY-MM-DD) for an instant in the user's display timezone. */ -function tzDayKey(at: Date, tz: string): string { - const { year, month, day } = wallClockInTz(at, tz); - return `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}`; -} - -/** Collapse rows to per-day means keyed in the user's tz. */ -function toDailyMeans( - rows: Array<{ value: number; at: Date }>, - tz: string, -): DailySeriesPoint[] { - const byDay = new Map(); - for (const r of rows) { - if (!Number.isFinite(r.value)) continue; - const day = tzDayKey(r.at, tz); - const acc = byDay.get(day) ?? { sum: 0, count: 0 }; - acc.sum += r.value; - acc.count += 1; - 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)); -} - /** * Fetch the FDR-surviving correlations that involve `measurementType`. * @@ -117,7 +96,10 @@ export async function getRelevantCorrelationsForMetric( DISCOVERY_OUTCOMES, ) as MeasurementType[]; - const [measurements, moodEntries] = await Promise.all([ + // v1.29.6 — `source` + `deviceType` + `sleepStage` feed + // `buildMeasurementDailySeries`'s per-type grain resolution below (see + // the route's identical comment for why). + const [measurements, moodEntries, priorityJson] = await Promise.all([ prisma.measurement.findMany({ where: { userId, @@ -127,7 +109,14 @@ export async function getRelevantCorrelationsForMetric( }, orderBy: { measuredAt: "asc" }, take: 20000, - select: { type: true, value: true, measuredAt: true }, + select: { + type: true, + value: true, + measuredAt: true, + source: true, + deviceType: true, + sleepStage: true, + }, }), prisma.moodEntry.findMany({ where: { userId, deletedAt: null, moodLoggedAt: { gte: since } }, @@ -135,12 +124,19 @@ export async function getRelevantCorrelationsForMetric( take: 5000, select: { score: true, moodLoggedAt: true }, }), + loadUserSourcePriority(userId), ]); - const byType = new Map>(); + const byType = new Map(); for (const m of measurements) { const list = byType.get(m.type) ?? []; - list.push({ value: m.value, at: m.measuredAt }); + list.push({ + value: m.value, + at: m.measuredAt, + source: m.source, + deviceType: m.deviceType, + sleepStage: m.sleepStage, + }); byType.set(m.type, list); } const moodDaily = toDailyMeans( @@ -148,16 +144,22 @@ export async function getRelevantCorrelationsForMetric( tz, ); + const points = (key: string) => + key === "MOOD" + ? moodDaily + : buildMeasurementDailySeries( + key as MeasurementType, + byType.get(key) ?? [], + tz, + priorityJson, + ); + const series: NamedSeries[] = []; for (const key of DISCOVERY_BEHAVIOURS) { - const points = - key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz); - series.push({ key, role: "behaviour", points }); + series.push({ key, role: "behaviour", points: points(key) }); } for (const key of DISCOVERY_OUTCOMES) { - const points = - key === "MOOD" ? moodDaily : toDailyMeans(byType.get(key) ?? [], tz); - series.push({ key, role: "outcome", points }); + series.push({ key, role: "outcome", points: points(key) }); } const result = discoverCorrelations(series); diff --git a/src/lib/personal-records/__tests__/pr-detection-worker.test.ts b/src/lib/personal-records/__tests__/pr-detection-worker.test.ts index 08561385f..c4377c3d0 100644 --- a/src/lib/personal-records/__tests__/pr-detection-worker.test.ts +++ b/src/lib/personal-records/__tests__/pr-detection-worker.test.ts @@ -36,6 +36,7 @@ interface FakeMeasurementRow { source: string; externalId: string | null; userId: string; + deviceType?: string | null; } interface FakeWorkoutRow { @@ -100,14 +101,6 @@ function makeFakePrisma(state: { return true; } - // REG-9 day bucket key. Mirrors `date_trunc('day', measured_at)` on - // UTC — the production query and this fake share the same bucket - // contract so a day-sum test exercises the same partitioning the - // SQL aggregate performs. - function dayKey(d: Date): string { - return `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}`; - } - function matchesWorkoutWhere(row: FakeWorkoutRow, where: FakeWhere): boolean { if (where.userId !== undefined && row.userId !== where.userId) return false; if (where.sportType !== undefined && row.sportType !== where.sportType) @@ -140,79 +133,33 @@ function makeFakePrisma(state: { return true; } - // REG-9 (v1.4.46): the cumulative-kind path runs a `$queryRaw` - // template that SUMs each `(user_id, type)` partition by - // `date_trunc('day', measured_at)` and `ORDER BY day_total {DESC|ASC}`. - // The fake re-implements the same shape against the in-memory rows - // so the worker-level tests exercise the real day-bucket reducer - // logic, not a stub. - function runCumulativeDaySum( - userId: string, - type: MeasurementType, - direction: "desc" | "asc", - ): Array<{ day_total: number; max_measured_at: Date }> { - const matches = state.measurements.filter( - (r) => r.userId === userId && r.type === type, - ); - if (matches.length === 0) return []; - - const byDay = new Map< - string, - { day_total: number; max_measured_at: Date } - >(); - for (const r of matches) { - const key = dayKey(r.measuredAt); - const slot = byDay.get(key) ?? { - day_total: 0, - max_measured_at: r.measuredAt, - }; - slot.day_total += r.value; - if (r.measuredAt > slot.max_measured_at) { - slot.max_measured_at = r.measuredAt; - } - byDay.set(key, slot); - } - - const sorted = Array.from(byDay.values()).sort((a, b) => - direction === "desc" - ? b.day_total - a.day_total - : a.day_total - b.day_total, - ); - return [sorted[0]]; - } - return { - $queryRaw: vi.fn( - async (sql: { strings: string[]; values: unknown[]; sql?: string }) => { - // Prisma.sql flattens nested fragments at construction time — - // the ASC/DESC fragment is inlined into `strings`, only the - // userId + type land in the bound `values` list. We detect the - // cumulative day-sum probe by the SQL preamble and read the - // direction off the flattened SQL text. - const fullSql = sql.sql ?? sql.strings.join("?"); - if (!fullSql.includes("date_trunc('day', m.\"measured_at\")")) { - throw new Error(`Unhandled $queryRaw in fake Prisma:\n${fullSql}`); - } - const [userIdVal, typeVal] = sql.values as [string, string]; - // The ORDER BY direction was interpolated as a literal — recover - // it from the flattened SQL so the fake exercises the same - // direction the worker asked for. - const dirMatch = /ORDER BY day_total\s+(ASC|DESC)/i.exec(fullSql); - const direction: "asc" | "desc" = - dirMatch?.[1]?.toUpperCase() === "ASC" ? "asc" : "desc"; - return runCumulativeDaySum( - userIdVal, - typeVal as MeasurementType, - direction, - ); - }, - ), + // v1.29.6 — the cumulative-kind path no longer runs a `$queryRaw` + // UTC `date_trunc` aggregate; it reads every matching row via + // `measurement.findMany` and does the source-collapse + tz-aware + // day-bucket SUM in JS (`pickCanonicalSourceRows` + + // `dayKeyForUserTz`), matching every other cumulative-day surface. + user: { + findUnique: vi.fn(async () => ({ + // Every fixture in this file is written in `Date.UTC(...)` — + // "UTC" as the resolved IANA zone keeps the day-bucket keys + // identical to the pre-fix UTC `date_trunc` partitioning, so + // the existing day-sum assertions stay valid. + timezone: "UTC", + sourcePriorityJson: null, + })), + }, measurement: { count: vi.fn(async ({ where }: { where: FakeWhere }) => { return state.measurements.filter((r) => matchesMeasurementWhere(r, where), ).length; }), + findMany: vi.fn(async ({ where }: { where: FakeWhere }) => { + return state.measurements.filter((r) => + matchesMeasurementWhere(r, where), + ); + }), findFirst: vi.fn( async ({ where, @@ -900,6 +847,84 @@ describe("detectPersonalRecordsForUser — REG-9 cumulative day-sum (v1.4.46)", expect(pr?.value, `${type} day-sum`).toBe(expectedValue); } }); + + // v1.29.6 — a day with both an Apple Health AND a Withings write for + // the same cumulative metric must NOT be double-counted into a false + // "record" day. Before the fix, the day-sum probe summed every row on + // the day regardless of source, so 9 000 Apple-Health steps + 8 800 + // Withings steps on one day (17 800) could out-rank a later day's real + // 12 000-step single-source total — a fabricated record that also + // suppressed the later, genuine best. + it("does not double-count an Apple Health + Withings write on the same day into a false record", async () => { + const samples: FakeMeasurementRow[] = []; + for (let i = 0; i < PR_DETECTION_WARMUP_THRESHOLD; i++) { + samples.push({ + id: `m-seed-${i}`, + type: "ACTIVITY_STEPS", + value: 4000 + i * 100, + unit: "x", + measuredAt: new Date(Date.UTC(2026, 4, 4 + i, 12, 0, 0)), + source: "APPLE_HEALTH", + externalId: null, + userId: USER, + }); + } + // Cross-source day: Apple Health (the higher-priority default ladder + // source) and Withings both write the SAME day's steps. Summed + // naively this is 17 800 — bigger than any single-source day below. + // Canonically collapsed (Apple Health wins), it is 9 000. + samples.push({ + id: "m-dual-apple", + type: "ACTIVITY_STEPS", + value: 9000, + unit: "x", + measuredAt: new Date(Date.UTC(2026, 4, 11, 8, 0, 0)), + source: "APPLE_HEALTH", + externalId: null, + userId: USER, + }); + samples.push({ + id: "m-dual-withings", + type: "ACTIVITY_STEPS", + value: 8800, + unit: "x", + measuredAt: new Date(Date.UTC(2026, 4, 11, 8, 5, 0)), + source: "WITHINGS", + externalId: null, + userId: USER, + }); + // A later, genuine single-source day that beats the collapsed + // cross-source day (9 000) but NOT the naive double-counted sum + // (17 800). A pre-fix worker would suppress this as "no improvement". + samples.push({ + id: "m-later-best", + type: "ACTIVITY_STEPS", + value: 12000, + unit: "x", + measuredAt: new Date(Date.UTC(2026, 4, 13, 9, 0, 0)), + source: "APPLE_HEALTH", + externalId: null, + userId: USER, + }); + + const state = { + measurements: samples, + workouts: [], + personalRecords: [] as FakePersonalRecordRow[], + }; + const prisma = makeFakePrisma(state); + + await detectPersonalRecordsForUser(USER, { prisma }); + + const stepsPR = state.personalRecords.find( + (r) => r.metricType === "ACTIVITY_STEPS" && r.metricSlot === null, + ); + expect(stepsPR).toBeDefined(); + expect(stepsPR?.value).toBe(12000); + expect(stepsPR?.achievedAt.getTime()).toBe( + new Date(Date.UTC(2026, 4, 13, 9, 0, 0)).getTime(), + ); + }); }); describe("compareToCurrentBest — pure helper", () => { diff --git a/src/lib/personal-records/pr-detection-worker.ts b/src/lib/personal-records/pr-detection-worker.ts index 31e5165a2..a50fd1767 100644 --- a/src/lib/personal-records/pr-detection-worker.ts +++ b/src/lib/personal-records/pr-detection-worker.ts @@ -39,11 +39,13 @@ */ import { prisma as defaultPrisma } from "@/lib/db"; import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { dayKeyForUserTz } from "@/lib/measurements/consolidation-tz"; +import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; +import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { getPRDirection, isPRTrackable } from "./pr-direction"; import { PersonalRecordDirection, - Prisma, type MeasurementType, type PrismaClient, } from "@/generated/prisma/client"; @@ -207,6 +209,26 @@ export async function detectPersonalRecordsForUser( let ties = 0; let scanned = 0; + // v1.29.6 — resolved once per run and threaded into the cumulative-day + // picker: the source-collapse ladder needs the user's priority JSON, and + // the day bucketing needs the user's IANA timezone so a "best day" lines + // up with the calendar day the user actually experienced, matching every + // other cumulative-day surface (the `groupBy=day` list branch, the + // dashboard tile). Read directly via the injected `prisma` client + // (not `resolveUserTimezone` / `loadUserSourcePriority`, which both + // reach for the module-level singleton) so the worker's Prisma + // dependency injection — the whole reason `options.prisma` exists — + // stays intact for tests. + const userRow = await prisma.user.findUnique({ + where: { id: userId }, + select: { timezone: true, sourcePriorityJson: true }, + }); + const tz = + userRow?.timezone && userRow.timezone.length > 0 + ? userRow.timezone + : "Europe/Berlin"; + const priorityJson = userRow?.sourcePriorityJson ?? null; + // ── Measurement-driven PRs (metricSlot = null) ───────────── for (const type of measurementTypeEnum.options) { const metricType = type as MeasurementType; @@ -228,6 +250,8 @@ export async function detectPersonalRecordsForUser( userId, metricType, direction, + tz, + priorityJson, ); if (!best) continue; @@ -404,6 +428,8 @@ async function findBestMeasurement( userId: string, type: MeasurementType, direction: PersonalRecordDirection, + tz: string, + priorityJson: unknown, ): Promise { // REG-9 (v1.4.46): cumulative HealthKit kinds (ACTIVITY_STEPS, // ACTIVE_ENERGY_BURNED, FLIGHTS_CLIMBED, WALKING_RUNNING_DISTANCE, @@ -414,23 +440,30 @@ async function findBestMeasurement( // 4 000 steps plus every other slice on the same calendar day) was // ignored. // - // For cumulative kinds we instead bucket-and-sum per calendar day - // (UTC for now — analytics surfaces still day-bucket in UTC; the - // dashboard tile already runs the same shape via - // `pickCumulativeDaySum`). The picked row's `measuredAt` is the - // latest slice on the winning day so the PR's `achievedAt` - // timestamp still points at a real ingested sample (the unique - // index `(userId, metricType, metricSlot, achievedAt)` needs a - // stable timestamp). For MAX-direction cumulative kinds — which is - // every cumulative type today per `pr-direction.ts` — we pick the - // day with the largest sum. The MIN branch is wired symmetrically - // for future-proofing; if a MIN cumulative kind ever lands the - // worker will pick the day with the smallest sum. + // For cumulative kinds we instead bucket-and-sum per calendar day, + // anchored to the user's IANA timezone (matches every other + // cumulative-day surface — the `groupBy=day` list branch, the + // dashboard tile). The picked row's `measuredAt` is the latest + // slice on the winning day so the PR's `achievedAt` timestamp + // still points at a real ingested sample (the unique index + // `(userId, metricType, metricSlot, achievedAt)` needs a stable + // timestamp). For MAX-direction cumulative kinds — which is every + // cumulative type today per `pr-direction.ts` — we pick the day + // with the largest sum. The MIN branch is wired symmetrically for + // future-proofing; if a MIN cumulative kind ever lands the worker + // will pick the day with the smallest sum. // // Spot kinds (resting HR, VO2 max, HRV, body composition) keep the // existing per-row pick — each row is already the day's measurement. if (CUMULATIVE_HK_TYPES.has(type)) { - return findBestCumulativeDay(prisma, userId, type, direction); + return findBestCumulativeDay( + prisma, + userId, + type, + direction, + tz, + priorityJson, + ); } const row = await prisma.measurement.findFirst({ @@ -460,54 +493,93 @@ async function findBestMeasurement( } /** - * REG-9 (v1.4.46): day-bucket SUM picker for cumulative HK kinds. + * REG-9 (v1.4.46) / v1.29.6 — day-bucket SUM picker for cumulative HK + * kinds. * - * Postgres `date_trunc('day', measured_at)` groups every slice on the - * same UTC calendar day. The outer `ORDER BY day_sum {DESC|ASC} LIMIT 1` - * picks the winning day; the inner `MAX(measured_at)` returns a - * stable timestamp for the PR's `achievedAt`. The latest slice on the - * winning day also carries the row's `unit / source / externalId` — - * we re-read it as a separate `findFirst` so the response shape - * stays identical to the spot-kind path (the caller's downstream - * `personalRecord.create` reaches for the original row's metadata). + * v1.29.6 fix: the original implementation summed EVERY row on a + * calendar day regardless of source. A day with both an Apple Health + * AND a Withings write for the same cumulative metric (e.g. steps) + * summed both sources into one inflated total — a false "record" day + * that could permanently outrank a later day's real single-source + * total, which would never win the comparison again. The fix + * collapses overlapping sources to the ladder-canonical reading per + * day BEFORE summing (`pickCanonicalSourceRows`, the same picker the + * doctor report and mood crosstab use), and buckets on the user's + * IANA timezone instead of the UTC calendar day, matching every other + * cumulative-day surface. * * Order of operations: - * 1. Sum-per-day aggregate via `$queryRaw` (one round-trip). - * 2. Re-read the slice at `MAX(measured_at)` on the winning day - * to recover the row's metadata. + * 1. Read every row for the type (already scoped by the caller's + * warm-up gate to a metric with at least + * `PR_DETECTION_WARMUP_THRESHOLD` samples). + * 2. Collapse to the canonical source per local day. + * 3. Bucket-and-sum the canonical rows per local day; pick the + * winning day by `direction`. + * 4. Re-read the winning day's latest slice so the PR row carries a + * real row's unit/source/externalId (the day-sum overrides + * `value` below). * The unique-index contract is unaffected — a re-run picks the same - * day and the same MAX-slice, so `achievedAt` is stable. + * day and the same latest-slice timestamp, so `achievedAt` is stable. */ async function findBestCumulativeDay( prisma: PrismaClient, userId: string, type: MeasurementType, direction: PersonalRecordDirection, + tz: string, + priorityJson: unknown, ): Promise { - const orderDir = - direction === PersonalRecordDirection.MAX - ? Prisma.sql`DESC` - : Prisma.sql`ASC`; + const rows = await prisma.measurement.findMany({ + where: { userId, type, deletedAt: null }, + orderBy: { measuredAt: "asc" }, + select: { + id: true, + value: true, + unit: true, + measuredAt: true, + source: true, + externalId: true, + deviceType: true, + }, + }); + if (rows.length === 0) return null; - const rows = await prisma.$queryRaw< - Array<{ day_total: number; max_measured_at: Date }> - >( - Prisma.sql` - SELECT - SUM(m."value")::double precision AS day_total, - MAX(m."measured_at") AS max_measured_at - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."type"::text = ${type} - AND m."deleted_at" IS NULL - GROUP BY date_trunc('day', m."measured_at") - ORDER BY day_total ${orderDir} - LIMIT 1 - `, - ); + const metricKey = metricKeyForType(type); + const canonicalRows = metricKey + ? pickCanonicalSourceRows( + rows.map((r) => ({ ...r, type })), + metricKey, + priorityJson, + (d) => dayKeyForUserTz(d, tz), + ).canonicalRows + : rows; - const winner = rows[0]; - if (!winner) return null; + const byDay = new Map(); + for (const row of canonicalRows) { + const key = dayKeyForUserTz(row.measuredAt, tz); + const slot = byDay.get(key); + if (!slot) { + byDay.set(key, { total: row.value, latest: row.measuredAt }); + } else { + slot.total += row.value; + if (row.measuredAt > slot.latest) slot.latest = row.measuredAt; + } + } + + let winnerTotal: number | null = null; + let winnerLatest: Date | null = null; + for (const { total, latest } of byDay.values()) { + const better = + winnerTotal === null || + (direction === PersonalRecordDirection.MAX + ? total > winnerTotal + : total < winnerTotal); + if (better) { + winnerTotal = total; + winnerLatest = latest; + } + } + if (winnerTotal === null || winnerLatest === null) return null; // Re-read the latest slice on the winning day so the PR row carries // a real row's unit/source/externalId. The slice's `value` is the @@ -518,7 +590,7 @@ async function findBestCumulativeDay( userId, type, deletedAt: null, - measuredAt: winner.max_measured_at, + measuredAt: winnerLatest, }, select: { id: true, @@ -532,7 +604,7 @@ async function findBestCumulativeDay( return { id: slice.id, - value: winner.day_total, + value: winnerTotal, unit: slice.unit, measuredAt: slice.measuredAt, source: slice.source,