From 7860df2fbaff2a226575f272de9d84b9511e27de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 26 Sep 2026 16:56:02 +0200 Subject: [PATCH] perf(rollups): fold weekly, monthly and yearly buckets from canonical days in SQL Building the coarse buckets from the DAY tier read every day of the window into the process: a five-year yearly read moved thousands of rows per type, and the Coach's history bands read several such windows per request. The per-day source pick and the fold now run in Postgres (DISTINCT ON the day, ordered by the ladder rank and then the source name, the same order the in-process pick uses), so a coarse read returns one row per bucket. The long-range chart picks its tier from the span of the window's days and reads that tier once. The Coach snapshot hands its already-loaded source order to the history bands instead of each band reading it again. --- src/lib/ai/coach/snapshot-series.ts | 3 + src/lib/ai/coach/snapshot.ts | 3 + .../__tests__/blood-pressure-status.test.ts | 3 + src/lib/insights/__tests__/bmi-status.test.ts | 3 + src/lib/insights/__tests__/features.test.ts | 38 +- .../insights/__tests__/metric-status.test.ts | 3 + .../insights/__tests__/payload-budget.test.ts | 3 + .../insights/__tests__/pulse-status.test.ts | 17 +- .../insights/__tests__/weight-status.test.ts | 3 + .../__tests__/measurement-read-wmy.test.ts | 474 +++++++++--------- src/lib/rollups/measurement-read-wmy.ts | 52 +- src/lib/rollups/measurement-read.ts | 307 +++++++----- src/lib/rollups/tiered-context.ts | 22 +- .../rollup-per-day-canonical.test.ts | 55 ++ 14 files changed, 563 insertions(+), 423 deletions(-) diff --git a/src/lib/ai/coach/snapshot-series.ts b/src/lib/ai/coach/snapshot-series.ts index 9357b0716..cc5752508 100644 --- a/src/lib/ai/coach/snapshot-series.ts +++ b/src/lib/ai/coach/snapshot-series.ts @@ -206,6 +206,8 @@ export async function buildCoarseTimelineTail( type: MeasurementType, now: Date, tz: string, + /** The user's source-priority blob, already loaded by the caller. */ + userPriorityJson: unknown, ): Promise { // `buildTieredSeries` reads fall back on miss — a coverage miss yields empty // bands, which collapse to `undefined` here. The try/catch keeps the coarse @@ -222,6 +224,7 @@ export async function buildCoarseTimelineTail( now: now.getTime(), tz, coarseOnly: true, + userPriorityJson, }); } catch { return undefined; diff --git a/src/lib/ai/coach/snapshot.ts b/src/lib/ai/coach/snapshot.ts index fb399c94b..824f1ea87 100644 --- a/src/lib/ai/coach/snapshot.ts +++ b/src/lib/ai/coach/snapshot.ts @@ -548,6 +548,7 @@ async function buildCoachSnapshotImpl( "BLOOD_PRESSURE_SYS" as MeasurementType, now, userTz, + prefsRow?.sourcePriorityJson ?? null, ); } if (wantsWeight) { @@ -556,6 +557,7 @@ async function buildCoachSnapshotImpl( "WEIGHT" as MeasurementType, now, userTz, + prefsRow?.sourcePriorityJson ?? null, ); } if (wantsPulse) { @@ -564,6 +566,7 @@ async function buildCoachSnapshotImpl( "PULSE" as MeasurementType, now, userTz, + prefsRow?.sourcePriorityJson ?? null, ); } const [bpCoarseTail, weightCoarseTail, pulseCoarseTail] = await Promise.all([ diff --git a/src/lib/insights/__tests__/blood-pressure-status.test.ts b/src/lib/insights/__tests__/blood-pressure-status.test.ts index 191b5e50e..974623e27 100644 --- a/src/lib/insights/__tests__/blood-pressure-status.test.ts +++ b/src/lib/insights/__tests__/blood-pressure-status.test.ts @@ -18,6 +18,8 @@ vi.mock("@/lib/db", () => ({ }, measurement: { findMany: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // Weekly / monthly / yearly buckets are folded from the DAY tier in SQL. + $queryRaw: vi.fn(), medication: { findMany: vi.fn() }, medicationIntakeEvent: { findMany: vi.fn() }, moodEntry: { findMany: vi.fn() }, @@ -94,6 +96,7 @@ beforeEach(() => { // Cold rollup tier: the BP channels fold monthly/yearly from the // full-history `measurement.findMany` fallback on a tier miss. vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); }); describe("generateBloodPressureStatusForUser — graded payload", () => { diff --git a/src/lib/insights/__tests__/bmi-status.test.ts b/src/lib/insights/__tests__/bmi-status.test.ts index 0a5122681..16a1fd4fa 100644 --- a/src/lib/insights/__tests__/bmi-status.test.ts +++ b/src/lib/insights/__tests__/bmi-status.test.ts @@ -19,6 +19,8 @@ vi.mock("@/lib/db", () => ({ // v1.18.11 (P6) — the input gate probes salient inputs via groupBy. measurement: { findMany: vi.fn(), groupBy: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // Weekly / monthly / yearly buckets are folded from the DAY tier in SQL. + $queryRaw: vi.fn(), }, })); @@ -87,6 +89,7 @@ beforeEach(() => { // Cold rollup tier: the BMI graded series scales the WEIGHT tier, which // folds from the full-history `measurement.findMany` fallback on a miss. vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); // v1.18.11 (P6) — input-gate probe default: empty groups so the fingerprint // is computed but, with no cached `inputHash`, the gate misses and fixtures // build normally. The forced fixtures skip the gate entirely. diff --git a/src/lib/insights/__tests__/features.test.ts b/src/lib/insights/__tests__/features.test.ts index 07b10173f..39cda5375 100644 --- a/src/lib/insights/__tests__/features.test.ts +++ b/src/lib/insights/__tests__/features.test.ts @@ -148,14 +148,36 @@ describe("extractFeatures — v1.4.36 W3 bucketed payload", () => { }); it("attaches DAY / WEEK / MONTH buckets from measurement_rollups when includeRaw=true", async () => { - // Two WEIGHT DAY buckets in the 0-90d window and one day in the - // 365-1825d window, which surfaces as a MONTH bucket. Every window is - // folded from the DAY tier, so the mock answers DAY reads by window. - const weightDays = [ - rollupRow(10, 82.5, 2), - rollupRow(20, 82.7, 1), - rollupRow(400, 85.1, 28), - ]; + // Two WEIGHT DAY buckets in the 0-90d window (the DAY read) and one + // month in the 365-1825d window (the SQL fold of the DAY tier). + const weightDays = [rollupRow(10, 82.5, 2), rollupRow(20, 82.7, 1)]; + const monthStart = rollupRow(400, 85.1, 28).bucketStart; + prismaMock.$queryRaw.mockImplementation( + async (strings: TemplateStringsArray, ...values: unknown[]) => { + const sql = JSON.stringify([strings, values]); + if ( + !sql.includes("measurement_rollups") || + !values.includes("WEIGHT") || + !sql.includes("'month'") + ) { + return []; + } + return [ + { + bucket_start: monthStart, + count: 28, + sum_y: 28 * 85.1, + min_value: 84, + max_value: 86, + sum_x: null, + sum_xy: null, + sum_xx: null, + sum_yy: null, + computed_at: new Date(), + }, + ]; + }, + ); prismaMock.measurementRollup.findMany.mockImplementation( async (args: { where: { diff --git a/src/lib/insights/__tests__/metric-status.test.ts b/src/lib/insights/__tests__/metric-status.test.ts index d118a93ec..0f7fe5be4 100644 --- a/src/lib/insights/__tests__/metric-status.test.ts +++ b/src/lib/insights/__tests__/metric-status.test.ts @@ -18,6 +18,8 @@ vi.mock("@/lib/db", () => ({ }, measurement: { findMany: vi.fn(), count: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // Weekly / monthly / yearly buckets are folded from the DAY tier in SQL. + $queryRaw: vi.fn(), }, })); @@ -116,6 +118,7 @@ beforeEach(() => { {} as never, ); vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); // resetAllMocks clears the module-mock default impls too — restore the // benign defaults so the relations fetch and the previous-context format // are no-ops unless a test overrides them. diff --git a/src/lib/insights/__tests__/payload-budget.test.ts b/src/lib/insights/__tests__/payload-budget.test.ts index 8b2286742..717ed105b 100644 --- a/src/lib/insights/__tests__/payload-budget.test.ts +++ b/src/lib/insights/__tests__/payload-budget.test.ts @@ -31,6 +31,8 @@ vi.mock("@/lib/db", () => ({ // v1.18.11 (P6) — the slow-mover input gate probes salient inputs. measurement: { findMany: vi.fn(), groupBy: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // Weekly / monthly / yearly buckets are folded from the DAY tier in SQL. + $queryRaw: vi.fn(), medicationIntakeEvent: { findMany: vi.fn() }, moodEntry: { findMany: vi.fn(), aggregate: vi.fn() }, customMetric: { findMany: vi.fn() }, @@ -109,6 +111,7 @@ beforeEach(() => { [] as never, ); vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); // v1.18.11 (P6) — input-gate probe: empty groups + zero mood so the gate // misses (no cached inputHash) and every fixture builds normally. vi.mocked(prisma.measurement.groupBy).mockResolvedValue([] as never); diff --git a/src/lib/insights/__tests__/pulse-status.test.ts b/src/lib/insights/__tests__/pulse-status.test.ts index 0e3ef9cff..1f76a76a6 100644 --- a/src/lib/insights/__tests__/pulse-status.test.ts +++ b/src/lib/insights/__tests__/pulse-status.test.ts @@ -90,6 +90,7 @@ beforeEach(() => { // Cold rollup tier: the pulse graded series folds monthly/yearly from // the full-history `measurement.findMany` fallback on a tier miss. vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); }); describe("generatePulseStatusForUser — graded payload", () => { @@ -111,11 +112,17 @@ describe("generatePulseStatusForUser — graded payload", () => { // v1.28.25 — PULSE is a dense type, so the cold-tier fallback reads // a SQL day-bucket aggregate instead of the raw findMany walk. Feed // the same 1000 days as day buckets. - vi.mocked(prisma.$queryRaw).mockResolvedValue( - records - .map((r) => ({ bucket_start: r.measuredAt, mean: r.value })) - .reverse() as never, - ); + // The rollup tier is empty (the coarse fold over it returns nothing), + // so the monthly / yearly slices come from that fallback. + const dayBuckets = records + .map((r) => ({ bucket_start: r.measuredAt, mean: r.value })) + .reverse(); + vi.mocked(prisma.$queryRaw).mockImplementation((( + strings: TemplateStringsArray, + ) => + Promise.resolve( + strings.join("").includes("measurement_rollups") ? [] : dayBuckets, + )) as never); vi.mocked(prisma.moodEntry.findMany).mockResolvedValue([] as never); const captured: { userPrompt: string | null } = { userPrompt: null }; diff --git a/src/lib/insights/__tests__/weight-status.test.ts b/src/lib/insights/__tests__/weight-status.test.ts index b07ef29e3..b57af8a95 100644 --- a/src/lib/insights/__tests__/weight-status.test.ts +++ b/src/lib/insights/__tests__/weight-status.test.ts @@ -19,6 +19,8 @@ vi.mock("@/lib/db", () => ({ // moodEntry.aggregate before the heavy findMany build. measurement: { findMany: vi.fn(), groupBy: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // Weekly / monthly / yearly buckets are folded from the DAY tier in SQL. + $queryRaw: vi.fn(), moodEntry: { findMany: vi.fn(), aggregate: vi.fn() }, customMetric: { findMany: vi.fn() }, // v1.11.1 — the rollup readers lazy-load the user's @@ -102,6 +104,7 @@ beforeEach(() => { // Cold rollup tier: the graded builder folds monthly/yearly from the // full-history `measurement.findMany` fallback the test already mocks. vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never); // v1.11.1 — null source-priority blob → default rank ladders. vi.mocked(prisma.user.findUnique).mockResolvedValue(null as never); // v1.18.11 (P6) — input-gate probe. Default to empty groups + zero mood so diff --git a/src/lib/rollups/__tests__/measurement-read-wmy.test.ts b/src/lib/rollups/__tests__/measurement-read-wmy.test.ts index 31dd66dbb..936f20730 100644 --- a/src/lib/rollups/__tests__/measurement-read-wmy.test.ts +++ b/src/lib/rollups/__tests__/measurement-read-wmy.test.ts @@ -3,10 +3,15 @@ * readers and the auto-router that picks the largest granularity * that still resolves a requested window. * - * `prisma.measurementRollup.findMany` is mocked at the module level - * so the test pins: - * - every granularity is folded from the canonical DAY buckets (the - * source is chosen per DAY, never once per week / month / year), + * Prisma is mocked at the module level (DAY rows through `findMany`, the + * coarse fold through `$queryRaw`, the chart span through `aggregate`), so + * the test pins: + * - DAY reads the DAY tier; WEEK / MONTH / YEAR read the SQL fold of the + * canonical DAY buckets (the per-day source pick and the fold are SQL, + * pinned against real Postgres in + * `tests/integration/rollup-per-day-canonical.test.ts`), + * - a coarse bucket's spread and slope compose from its days' + * accumulators, * - empty rollups return `null` so the caller can branch on * coverage miss, * - the auto-router picks the coarsest granularity whose floor the @@ -23,6 +28,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ findMany: vi.fn(), + aggregate: vi.fn(), + queryRaw: vi.fn(), userFindUnique: vi.fn(), })); @@ -30,10 +37,9 @@ vi.mock("@/lib/db", () => ({ prisma: { measurementRollup: { findMany: mocks.findMany, + aggregate: mocks.aggregate, }, - // v1.11.1 — readBestGranularityRollups loads the source-priority blob via - // loadUserSourcePriority; default to null so the collapse uses the default - // ladders (these routing fixtures are single-source per bucket anyway). + $queryRaw: mocks.queryRaw, user: { findUnique: mocks.userFindUnique, }, @@ -49,7 +55,9 @@ import { } from "../measurement-read-wmy"; import { pickBucket } from "@/lib/charts/bucket-time-series"; -const { findMany } = mocks; +const { findMany, aggregate, queryRaw } = mocks; +const DAY_MS = 86_400_000; +const ORIGIN_DAYS = 18262; function bucket( bucketStart: string, @@ -69,14 +77,80 @@ function bucket( }; } +/** One stored DAY rollup row, as the DAY `findMany` returns it. */ +function dayRow(day: string, source: string, count: number, mean: number) { + const bucketStart = new Date(`${day}T00:00:00.000Z`); + return { + bucketStart, + source, + count, + mean, + minValue: mean, + maxValue: mean, + sumValue: count * mean, + sd: 0, + slope: null, + r2: null, + sumX: null, + sumXy: null, + sumXx: null, + sumYy: null, + computedAt: new Date("2026-06-01T00:00:00.000Z"), + }; +} + +/** + * One coarse bucket as the SQL fold returns it, from `(day, value)` single + * readings; the accumulators are the sums a real fold would carry. + */ +function folded( + bucketStart: string, + readings: Array<[string, number]>, + withAccumulators = true, +) { + let sumX = 0; + let sumXy = 0; + let sumXx = 0; + let sumYy = 0; + let sumY = 0; + for (const [day, value] of readings) { + const x = Date.parse(`${day}T12:00:00.000Z`) / DAY_MS - ORIGIN_DAYS; + sumX += x; + sumXy += x * value; + sumXx += x * x; + sumYy += value * value; + sumY += value; + } + const values = readings.map(([, v]) => v); + return { + bucket_start: new Date(bucketStart), + count: readings.length, + sum_y: sumY, + min_value: Math.min(...values), + max_value: Math.max(...values), + sum_x: withAccumulators ? sumX : null, + sum_xy: withAccumulators ? sumXy : null, + sum_xx: withAccumulators ? sumXx : null, + sum_yy: withAccumulators ? sumYy : null, + computed_at: new Date("2026-06-01T00:00:00.000Z"), + }; +} + +/** The SQL text and bound values of a `$queryRaw` call, as one string. */ +function sqlOf(call: unknown[]): string { + return JSON.stringify(call); +} + beforeEach(() => { findMany.mockReset(); - // Default to "no coverage" for any read a test did not queue explicitly — - // the v1.37.29 span-refinement probe in `readTieredRollupSeries` issues an - // extra read when a fixture's rows span less than the requested window, - // and an unqueued vi.fn() would resolve `undefined` and crash the reader - // instead of modelling an unminted tier. findMany.mockResolvedValue([]); + aggregate.mockReset(); + aggregate.mockResolvedValue({ + _min: { bucketStart: null }, + _max: { bucketStart: null }, + }); + queryRaw.mockReset(); + queryRaw.mockResolvedValue([]); mocks.userFindUnique.mockReset(); mocks.userFindUnique.mockResolvedValue({ sourcePriorityJson: null }); }); @@ -85,60 +159,6 @@ afterEach(() => { vi.restoreAllMocks(); }); -/** One stored DAY rollup row, per source, as `findMany` returns it. */ -function dayRow( - day: string, - source: string, - count: number, - mean: number, - extra: Partial<{ minValue: number; maxValue: number }> = {}, -) { - const bucketStart = new Date(`${day}T00:00:00.000Z`); - // Accumulators for `count` readings of `mean` at the day's midday. - const x = bucketStart.getTime() / DAY_MS_ROWS - 18262 + 0.5; - return { - bucketStart, - source, - count, - mean, - minValue: extra.minValue ?? mean, - maxValue: extra.maxValue ?? mean, - sumValue: count * mean, - sd: 0, - slope: null, - r2: null, - sumX: count * x, - sumXy: count * x * mean, - sumXx: count * x * x, - sumYy: count * mean * mean, - computedAt: new Date("2026-06-01T00:00:00.000Z"), - }; -} -const DAY_MS_ROWS = 86_400_000; - -/** `n` consecutive days from `start`, one row each. */ -function days( - start: string, - n: number, - source: string, - count: number, - mean: number, -) { - const out = []; - const t0 = new Date(`${start}T00:00:00.000Z`).getTime(); - for (let i = 0; i < n; i++) { - out.push( - dayRow( - new Date(t0 + i * DAY_MS_ROWS).toISOString().slice(0, 10), - source, - count, - mean, - ), - ); - } - return out; -} - describe("readBestGranularityRollups", () => { beforeEach(() => { vi.useFakeTimers({ toFake: ["Date"] }); @@ -153,114 +173,97 @@ describe("readBestGranularityRollups", () => { expect(await readBestGranularityRollups("user", "WEIGHT", -10)).toBeNull(); expect(await readBestGranularityRollups("user", "WEIGHT", NaN)).toBeNull(); expect(findMany).not.toHaveBeenCalled(); + expect(queryRaw).not.toHaveBeenCalled(); + }); + + it("routes a 90-day window to the DAY tier", async () => { + findMany.mockResolvedValueOnce([dayRow("2025-12-20", "MANUAL", 1, 80)]); + const result = await readBestGranularityRollups("user", "WEIGHT", 90); + expect(result?.granularity).toBe("DAY"); + expect(findMany.mock.calls[0][0].where.granularity).toBe("DAY"); + expect(queryRaw).not.toHaveBeenCalled(); }); it.each([ - [90, "DAY"], - [365, "MONTH"], - [540, "MONTH"], - [1095, "YEAR"], - [120, "WEEK"], + [120, "WEEK", "week"], + [365, "MONTH", "month"], + [1095, "YEAR", "year"], ] as const)( - "routes a %i-day window to %s, folded from the DAY tier", - async (windowDays, granularity) => { - findMany.mockResolvedValueOnce([dayRow("2025-12-20", "MANUAL", 1, 80)]); + "routes a %i-day window to %s, folded from the DAY tier in SQL", + async (windowDays, granularity, unit) => { + queryRaw.mockResolvedValueOnce([ + folded("2025-12-01T00:00:00.000Z", [["2025-12-02", 80]]), + ]); const result = await readBestGranularityRollups( "user", "WEIGHT", windowDays, ); expect(result?.granularity).toBe(granularity); - expect(findMany).toHaveBeenCalledTimes(1); - expect(findMany.mock.calls[0][0].where.granularity).toBe("DAY"); + expect(queryRaw).toHaveBeenCalledTimes(1); + const sql = sqlOf(queryRaw.mock.calls[0]); + expect(sql).toContain(`'${unit}'`); + // The fold reads the DAY rows and picks one source per day. + expect(sql).toContain(`granularity\\" = 'DAY'`); + expect(sql).toContain("DISTINCT ON"); + expect(findMany).not.toHaveBeenCalled(); }, ); - it("returns null when the window holds no DAY buckets", async () => { - findMany.mockResolvedValue([]); + it("returns null when the window holds no buckets", async () => { const result = await readBestGranularityRollups("user", "WEIGHT", 1095); expect(result).toBeNull(); - expect(findMany).toHaveBeenCalledTimes(1); + expect(queryRaw).toHaveBeenCalledTimes(1); }); - // The canonical source is a per-DAY decision. Collapsing the stored - // per-source MONTH rows picked WITHINGS for the whole month (it leads the - // blood-pressure ladder) and reported its one reading: count 1, mean 120. - it("keeps every manual day of a month that also carries one synced day", async () => { - findMany.mockResolvedValueOnce([ - ...days("2025-11-01", 20, "MANUAL", 1, 150), - dayRow("2025-11-25", "WITHINGS", 1, 120), - ]); - const result = await readBestGranularityRollups( - "user", - "BLOOD_PRESSURE_SYS", - 365, - ); - expect(result?.granularity).toBe("MONTH"); - expect(result?.rows).toHaveLength(1); - expect(result?.rows[0].count).toBe(21); - expect(result?.rows[0].mean).toBeCloseTo((20 * 150 + 120) / 21, 9); - expect(result?.rows[0].minValue).toBe(120); - expect(result?.rows[0].maxValue).toBe(150); - }); - - it("sums a step month whose days switch device, one source per day", async () => { - findMany.mockResolvedValueOnce([ - ...days("2025-11-01", 10, "FITBIT", 1, 8000), - ...days("2025-11-11", 20, "APPLE_HEALTH", 1, 8000), - ]); - const result = await readBestGranularityRollups( - "user", - "ACTIVITY_STEPS", - 365, - ); - expect(aggregateWmyBuckets(result?.rows ?? []).sum).toBe(240_000); - }); - - it("still keeps one source on a day both sources carry", async () => { - findMany.mockResolvedValueOnce([ - dayRow("2025-11-03", "APPLE_HEALTH", 1, 9000), - dayRow("2025-11-03", "FITBIT", 1, 7000), - dayRow("2025-11-04", "FITBIT", 1, 7000), - ]); - const result = await readBestGranularityRollups( - "user", - "ACTIVITY_STEPS", - 365, - ); - expect(result?.rows[0].sumValue).toBe(16_000); - expect(result?.rows[0].count).toBe(2); - }); - - it("cuts weeks on the ISO Monday in UTC", async () => { - findMany.mockResolvedValueOnce([ - dayRow("2025-11-02", "MANUAL", 1, 1), // Sunday → week of 2025-10-27 - dayRow("2025-11-03", "MANUAL", 1, 2), // Monday → week of 2025-11-03 - dayRow("2025-11-09", "MANUAL", 1, 3), // Sunday → week of 2025-11-03 - ]); - const result = await readBestGranularityRollups("user", "WEIGHT", 120); - expect(result?.granularity).toBe("WEEK"); - expect(result?.rows.map((r) => r.bucketStart.toISOString())).toEqual([ - "2025-10-27T00:00:00.000Z", - "2025-11-03T00:00:00.000Z", - ]); - expect(result?.rows.map((r) => r.count)).toEqual([1, 2]); - }); - - it("composes the bucket's spread and slope from the day accumulators", async () => { - // Three days of one reading each: 70, 72, 74 → slope 2 per day, - // population sd sqrt(8/3), r² 1. - findMany.mockResolvedValueOnce([ - dayRow("2025-11-03", "MANUAL", 1, 70), - dayRow("2025-11-04", "MANUAL", 1, 72), - dayRow("2025-11-05", "MANUAL", 1, 74), + it("composes a coarse bucket's mean, spread and slope from its days", async () => { + // Three days of one reading each: 70, 72, 74 → mean 72, slope 2 per + // day, population sd sqrt(8/3), r² 1. + queryRaw.mockResolvedValueOnce([ + folded("2025-11-01T00:00:00.000Z", [ + ["2025-11-03", 70], + ["2025-11-04", 72], + ["2025-11-05", 74], + ]), ]); const result = await readBestGranularityRollups("user", "WEIGHT", 365); const month = result?.rows[0]; + expect(month?.count).toBe(3); + expect(month?.mean).toBeCloseTo(72, 9); + expect(month?.sumValue).toBe(216); expect(month?.slope).toBeCloseTo(2, 6); expect(month?.r2).toBeCloseTo(1, 6); expect(month?.sd).toBeCloseTo(Math.sqrt(8 / 3), 6); }); + + it("reports no spread or slope for a bucket with a day lacking accumulators", async () => { + queryRaw.mockResolvedValueOnce([ + folded( + "2025-11-01T00:00:00.000Z", + [ + ["2025-11-03", 70], + ["2025-11-04", 72], + ], + false, + ), + ]); + const result = await readBestGranularityRollups("user", "WEIGHT", 365); + expect(result?.rows[0].mean).toBe(71); + expect(result?.rows[0].sd).toBeNull(); + expect(result?.rows[0].slope).toBeNull(); + }); + + it("leaves out a bucket that starts before the window", async () => { + // The 365-day window starts 2025-01-15; January 2025 began before it. + queryRaw.mockResolvedValueOnce([ + folded("2025-01-01T00:00:00.000Z", [["2025-01-20", 80]]), + folded("2025-02-01T00:00:00.000Z", [["2025-02-03", 81]]), + ]); + const result = await readBestGranularityRollups("user", "WEIGHT", 365); + expect(result?.rows.map((r) => r.bucketStart.toISOString())).toEqual([ + "2025-02-01T00:00:00.000Z", + ]); + }); }); describe("aggregateWmyBuckets", () => { @@ -373,16 +376,22 @@ describe("readBestGranularityRollups — cross-consumer routing parity", () => { * truncated recent slice; that the finer fallback rescues a coverage * miss; and that the wire shape mirrors the daily reader. */ +/** + * Whole-history series reader. Pins that a very long "Alle" window reads + * the display-matching tier (chosen from the real span of the data, WEEK for + * 1–2 years, MONTH beyond), that it spans the whole history, and that the + * wire shape mirrors the daily reader. + */ describe("readTieredRollupSeries", () => { - const DAY_MS = 86_400_000; - // A fixed "now-ish" anchor for the current-window cases; the historic case - // uses its own explicit past bounds. Nothing in the reader reads - // `Date.now()` any more, so these are just span endpoints. const NOW = new Date("2026-06-21T00:00:00.000Z"); const win = (days: number, to: Date = NOW) => ({ from: new Date(to.getTime() - days * DAY_MS), to, }); + const spanOf = (first: string, last: string) => ({ + _min: { bucketStart: new Date(first) }, + _max: { bucketStart: new Date(last) }, + }); it("returns null on a non-positive window without touching the db", async () => { expect( @@ -393,118 +402,86 @@ describe("readTieredRollupSeries", () => { to: NOW, }), ).toBeNull(); - expect(findMany).not.toHaveBeenCalled(); - }); - - it("folds a multi-year window into MONTH buckets and spans the whole history", async () => { - // Ten years of first-of-January readings → MONTH tier; the oldest - // bucket survives (no recent-slice truncation). - const rows = []; - for (let y = 2017; y <= 2026; y++) { - rows.push(dayRow(`${y}-01-01`, "MANUAL", 12, 80)); - } - findMany.mockResolvedValueOnce(rows); - - const result = await readTieredRollupSeries({ - userId: "u", - type: "WEIGHT", - ...win(3650), - }); - - expect(result?.granularity).toBe("MONTH"); - expect(findMany).toHaveBeenCalledTimes(1); - expect(findMany.mock.calls[0][0].where.granularity).toBe("DAY"); - expect(result?.rows[0].measuredAt).toBe("2017-01-01T00:00:00.000Z"); - expect(result?.rows.at(-1)?.measuredAt).toBe("2026-01-01T00:00:00.000Z"); - expect(result?.rows.length).toBe(10); - }); - - // v1.26.0 SEAM-N2 — a historic window entirely in the past must bound the - // rollup read on BOTH ends. - it("bounds the DAY read on BOTH ends for a historic past window", async () => { - const from = new Date("2020-01-01T00:00:00.000Z"); - const to = new Date("2022-01-01T00:00:00.000Z"); - findMany.mockResolvedValueOnce([ - dayRow("2020-02-01", "MANUAL", 12, 80), - dayRow("2021-06-01", "MANUAL", 12, 81), - dayRow("2021-12-01", "MANUAL", 12, 82), - ]); - - const result = await readTieredRollupSeries({ - userId: "u", - type: "WEIGHT", - from, - to, - }); - - expect(findMany.mock.calls[0][0].where.bucketStart).toEqual({ - gte: from, - lte: to, - }); - for (const row of result?.rows ?? []) { - const t = new Date(row.measuredAt).getTime(); - expect(t).toBeGreaterThanOrEqual(from.getTime()); - expect(t).toBeLessThanOrEqual(to.getTime()); - } - // Data span ≈ 22 months → WEEK (the request alone would call for MONTH). - expect(result?.granularity).toBe("WEEK"); + expect(aggregate).not.toHaveBeenCalled(); }); it("returns null when the window holds no DAY buckets", async () => { - findMany.mockResolvedValue([]); const result = await readTieredRollupSeries({ userId: "u", type: "WEIGHT", ...win(3650), }); expect(result).toBeNull(); - expect(findMany).toHaveBeenCalledTimes(1); + expect(queryRaw).not.toHaveBeenCalled(); + expect(findMany).not.toHaveBeenCalled(); }); - it("surfaces the cumulative summed total for SUM metrics and drops the spread", async () => { - findMany.mockResolvedValueOnce([ - ...days("2024-01-01", 30, "APPLE_HEALTH", 1, 8000), - dayRow("2026-01-01", "APPLE_HEALTH", 1, 5000), + it("serves MONTH buckets over a multi-year history and keeps the oldest", async () => { + aggregate.mockResolvedValueOnce( + spanOf("2017-01-01T00:00:00.000Z", "2026-01-01T00:00:00.000Z"), + ); + queryRaw.mockResolvedValueOnce([ + folded("2017-01-01T00:00:00.000Z", [["2017-01-01", 80]]), + folded("2026-01-01T00:00:00.000Z", [["2026-01-01", 82]]), + ]); + + const result = await readTieredRollupSeries({ + userId: "u", + type: "WEIGHT", + ...win(3650), + }); + + expect(result?.granularity).toBe("MONTH"); + expect(sqlOf(queryRaw.mock.calls[0])).toContain("'month'"); + expect(result?.rows[0].measuredAt).toBe("2017-01-01T00:00:00.000Z"); + expect(result?.rows).toHaveLength(2); + }); + + // v1.26.0 SEAM-N2 — a historic window entirely in the past bounds the read + // on BOTH ends. + it("bounds the span read on both ends for a historic window", async () => { + const from = new Date("2020-01-01T00:00:00.000Z"); + const to = new Date("2022-01-01T00:00:00.000Z"); + await readTieredRollupSeries({ userId: "u", type: "WEIGHT", from, to }); + expect(aggregate.mock.calls[0][0].where.bucketStart).toEqual({ + gte: from, + lte: to, + }); + }); + + it("serves the summed total for step-like metrics and drops the spread", async () => { + aggregate.mockResolvedValueOnce( + spanOf("2024-01-01T00:00:00.000Z", "2026-01-01T00:00:00.000Z"), + ); + queryRaw.mockResolvedValueOnce([ + folded( + "2024-01-01T00:00:00.000Z", + Array.from({ length: 30 }, (_, i): [string, number] => [ + `2024-01-${String(i + 1).padStart(2, "0")}`, + 8000, + ]), + ), ]); const result = await readTieredRollupSeries({ userId: "u", type: "ACTIVITY_STEPS", ...win(3650), }); - expect(result?.granularity).toBe("MONTH"); expect(result?.rows[0].value).toBe(240_000); expect(result?.rows[0].minValue).toBeUndefined(); expect(result?.rows[0].maxValue).toBeUndefined(); }); - it("carries the count-weighted mean + spread for spot metrics", async () => { - findMany.mockResolvedValueOnce([ - dayRow("2024-01-02", "MANUAL", 2, 80, { minValue: 78, maxValue: 82 }), - dayRow("2024-01-03", "MANUAL", 1, 84, { minValue: 84, maxValue: 85 }), - dayRow("2026-01-01", "MANUAL", 1, 80), - ]); - const result = await readTieredRollupSeries({ - userId: "u", - type: "WEIGHT", - ...win(3650), - }); - expect(result?.rows[0].value).toBeCloseTo((2 * 80 + 84) / 3, 9); - expect(result?.rows[0].count).toBe(3); - expect(result?.rows[0].minValue).toBe(78); - expect(result?.rows[0].maxValue).toBe(85); - }); - // v1.37.29 — the tier keys off the ACTUAL data span, not the requested - // window width. The "All" tab always requests ~3650 days, so pre-fix a - // record whose history spans four months came back as four MONTH means - // while the chart captioned them from the real span. + // window width. describe("span refinement", () => { it("serves DAY rows for four months of data on an All request", async () => { + aggregate.mockResolvedValueOnce( + spanOf("2026-02-03T00:00:00.000Z", "2026-05-28T00:00:00.000Z"), + ); findMany.mockResolvedValueOnce([ dayRow("2026-02-03", "MANUAL", 1, 80), - dayRow("2026-03-14", "MANUAL", 1, 80), - dayRow("2026-04-09", "MANUAL", 1, 80), - dayRow("2026-05-28", "MANUAL", 1, 80), + dayRow("2026-05-28", "MANUAL", 1, 81), ]); const result = await readTieredRollupSeries({ userId: "u", @@ -512,14 +489,16 @@ describe("readTieredRollupSeries", () => { ...win(3650), }); expect(result?.granularity).toBe("DAY"); - expect(result?.rows).toHaveLength(4); + expect(queryRaw).not.toHaveBeenCalled(); expect(result?.rows[0].measuredAt).toBe("2026-02-03T00:00:00.000Z"); }); - it("serves WEEK rows for a span between one and two years", async () => { - findMany.mockResolvedValueOnce([ - dayRow("2025-01-01", "MANUAL", 1, 80), - dayRow("2026-05-01", "MANUAL", 1, 80), + it("serves WEEK buckets for a span between one and two years", async () => { + aggregate.mockResolvedValueOnce( + spanOf("2025-01-01T00:00:00.000Z", "2026-05-01T00:00:00.000Z"), + ); + queryRaw.mockResolvedValueOnce([ + folded("2024-12-30T00:00:00.000Z", [["2025-01-01", 80]]), ]); const result = await readTieredRollupSeries({ userId: "u", @@ -527,10 +506,7 @@ describe("readTieredRollupSeries", () => { ...win(3650), }); expect(result?.granularity).toBe("WEEK"); - expect(result?.rows.map((r) => r.measuredAt)).toEqual([ - "2024-12-30T00:00:00.000Z", - "2026-04-27T00:00:00.000Z", - ]); + expect(sqlOf(queryRaw.mock.calls[0])).toContain("'week'"); }); }); }); diff --git a/src/lib/rollups/measurement-read-wmy.ts b/src/lib/rollups/measurement-read-wmy.ts index 92a40afc5..ece63f589 100644 --- a/src/lib/rollups/measurement-read-wmy.ts +++ b/src/lib/rollups/measurement-read-wmy.ts @@ -38,8 +38,8 @@ import type { import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; import { annotate } from "@/lib/logging/context"; +import { prisma } from "@/lib/db"; import { - foldCanonicalDays, loadUserSourcePriority, readCanonicalRollupBuckets, } from "@/lib/rollups/measurement-read"; @@ -302,9 +302,9 @@ const TIER_ORDER: RollupGranularity[] = ["DAY", "WEEK", "MONTH", "YEAR"]; * range band on spot metrics; cumulative metrics (steps, energy, * distance) surface the bucket's summed total and drop the spread. * - * Coverage handling: the window's canonical DAY buckets are read once and - * folded into the chosen tier. Returns `null` when the window holds no DAY - * buckets; the caller then falls through to its live-SQL path. + * Coverage handling: the tier is chosen from the span of the window's DAY + * buckets and read folded from them. Returns `null` when the window holds no + * DAY buckets; the caller then falls through to its live-SQL path. * * The result is NOT capped — the tier selection bounds the row count * (≤ ~104 weeks for the WEEK tier, ≤ ~12 months/year for MONTH). If a @@ -342,36 +342,38 @@ export async function readTieredRollupSeries(opts: { : await loadUserSourcePriority(userId); const target = pickRollupGranularityForWindow(windowDays); - const days = await readCanonicalRollupBuckets({ - userId, - type, - granularity: "DAY", - from, - to, - toInclusive: true, - userPriorityJson: priority, - }); - if (days.length === 0) return null; // Refine the tier by the ACTUAL data span, not the requested window. The // "Alle" tab always requests ~3650 days, so keying the tier off the request // width alone handed EVERY account MONTH buckets: a record whose history // spans four months came back as four monthly means while the client // (which captions from the real span) labelled them "weekly average". The // tier is never coarser than the request calls for. - const spanDays = Math.ceil( - (days[days.length - 1].bucketStart.getTime() - - days[0].bucketStart.getTime()) / - 86_400_000, - ); + const span = await prisma.measurementRollup.aggregate({ + where: { + userId, + type, + granularity: "DAY", + bucketStart: { gte: from, lte: to }, + }, + _min: { bucketStart: true }, + _max: { bucketStart: true }, + }); + const first = span._min.bucketStart; + const last = span._max.bucketStart; + if (!first || !last) return null; + const spanDays = Math.ceil((last.getTime() - first.getTime()) / 86_400_000); const refined = pickRollupGranularityForWindow(Math.max(1, spanDays)); const granularity = TIER_ORDER.indexOf(refined) < TIER_ORDER.indexOf(target) ? refined : target; - // Fold the window's own days; a leading bucket that starts before `from` - // would hold only part of its days, so it is left out as the stored coarse - // rows were. - const rows = foldCanonicalDays(days, granularity).filter( - (b) => b.bucketStart.getTime() >= from.getTime(), - ); + const rows = await readCanonicalRollupBuckets({ + userId, + type, + granularity, + from, + to, + toInclusive: true, + userPriorityJson: priority, + }); if (rows.length === 0) return null; const useSum = CUMULATIVE_HK_TYPES.has(type); annotate({ diff --git a/src/lib/rollups/measurement-read.ts b/src/lib/rollups/measurement-read.ts index 93c377028..c598645ab 100644 --- a/src/lib/rollups/measurement-read.ts +++ b/src/lib/rollups/measurement-read.ts @@ -25,7 +25,9 @@ import type { MeasurementType, RollupGranularity, } from "@/generated/prisma/client"; +import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { @@ -475,106 +477,90 @@ export interface CanonicalRollupBucket { } /** - * Fold canonical DAY buckets (one row per day, already source-collapsed) - * into buckets of `granularity`. - * - * The canonical source is a per-DAY decision: a week, month or year in which - * the user logged most days by hand and one day from a device holds readings - * from both, and each of those days counts. Collapsing the stored per-source - * WEEK / MONTH / YEAR rows instead picked ONE source for the whole bucket, so - * a month of manual blood-pressure readings plus a single synced one reported - * the synced day alone, and a step history that switched devices mid-month - * lost every day logged by the losing device. - * - * `count`, `min`, `max`, `sum` and the count-weighted `mean` compose exactly. - * `sd`, `slope` and `r2` are composed from the summed regression accumulators - * (`composeRegression`); a bucket with any day lacking them reports `null`. - * DAY input at DAY granularity passes through unchanged. + * One coarse bucket as the canonical-day fold returns it: the summed + * composable stats of the bucket's canonical days. */ -export function foldCanonicalDays( - days: ReadonlyArray, - granularity: RollupGranularity, -): CanonicalRollupBucket[] { - if (granularity === "DAY") return days.map((d) => ({ ...d })); - const groups = new Map(); - for (const d of days) { - const key = utcBucketStart(d.bucketStart, granularity).getTime(); - const slot = groups.get(key); - if (slot) slot.push(d); - else groups.set(key, [d]); - } - const out: CanonicalRollupBucket[] = []; - for (const [key, members] of [...groups.entries()].sort( - (a, b) => a[0] - b[0], - )) { - let count = 0; - let sumY = 0; - let min = Infinity; - let max = -Infinity; - let computedAt = members[0].computedAt; - let accumulatorsComplete = true; - let sumX = 0; - let sumXy = 0; - let sumXx = 0; - let sumYy = 0; - for (const m of members) { - count += m.count; - sumY += m.sumValue ?? m.mean * m.count; - if (m.minValue < min) min = m.minValue; - if (m.maxValue > max) max = m.maxValue; - if (m.computedAt.getTime() > computedAt.getTime()) { - computedAt = m.computedAt; - } - if ( - m.sumX === null || - m.sumXy === null || - m.sumXx === null || - m.sumYy === null - ) { - accumulatorsComplete = false; - } else { - sumX += m.sumX; - sumXy += m.sumXy; - sumXx += m.sumXx; - sumYy += m.sumYy; - } - } - if (count === 0) continue; - const mean = sumY / count; - const regression = accumulatorsComplete - ? composeRegression([ - { count, mean, sumValue: sumY, sumX, sumXy, sumXx, sumYy }, - ]) - : { slope: null, r2: null, sdPop: null }; - out.push({ - bucketStart: new Date(key), - count, - mean, - minValue: min, - maxValue: max, - sumValue: sumY, - // A single reading has no spread; STDDEV_POP of one value is 0. - sd: count === 1 && accumulatorsComplete ? 0 : regression.sdPop, - slope: regression.slope, - r2: regression.r2, - sumX: accumulatorsComplete ? sumX : null, - sumXy: accumulatorsComplete ? sumXy : null, - sumXx: accumulatorsComplete ? sumXx : null, - sumYy: accumulatorsComplete ? sumYy : null, - computedAt, - }); - } - return out; +interface FoldedBucketRow { + bucket_start: Date; + count: number; + sum_y: number; + min_value: number; + max_value: number; + /** Null when any day of the bucket lacks the regression accumulators. */ + sum_x: number | null; + sum_xy: number | null; + sum_xx: number | null; + sum_yy: number | null; + computed_at: Date; +} + +/** SQL `date_trunc` unit per coarse granularity (a closed map, never input). */ +const COARSE_TRUNC_UNIT: Record, string> = { + WEEK: "week", + MONTH: "month", + YEAR: "year", +}; + +/** + * Compose one coarse bucket from the summed stats of its canonical days. + * `count`, `min`, `max`, `sum` and the count-weighted `mean` are exact; + * `sd`, `slope` and `r2` come from the summed regression accumulators + * (`composeRegression`) and are `null` when a day lacks them. + */ +function composeFoldedBucket(row: FoldedBucketRow): CanonicalRollupBucket { + const count = Number(row.count); + const sumY = Number(row.sum_y); + const mean = sumY / count; + const accumulatorsComplete = + row.sum_x !== null && + row.sum_xy !== null && + row.sum_xx !== null && + row.sum_yy !== null; + const sumX = accumulatorsComplete ? Number(row.sum_x) : null; + const sumXy = accumulatorsComplete ? Number(row.sum_xy) : null; + const sumXx = accumulatorsComplete ? Number(row.sum_xx) : null; + const sumYy = accumulatorsComplete ? Number(row.sum_yy) : null; + const regression = accumulatorsComplete + ? composeRegression([ + { count, mean, sumValue: sumY, sumX, sumXy, sumXx, sumYy }, + ]) + : { slope: null, r2: null, sdPop: null }; + return { + bucketStart: new Date(row.bucket_start), + count, + mean, + minValue: Number(row.min_value), + maxValue: Number(row.max_value), + sumValue: sumY, + // A single reading has no spread; STDDEV_POP of one value is 0. + sd: count === 1 && accumulatorsComplete ? 0 : regression.sdPop, + slope: regression.slope, + r2: regression.r2, + sumX, + sumXy, + sumXx, + sumYy, + computedAt: new Date(row.computed_at), + }; } /** * Read one type's canonical rollup buckets at `granularity`. * - * Every granularity is built from the DAY tier: the day's rows are collapsed - * to the ladder-canonical source first (`collapseRollupRowsBySource`), then - * folded into weeks, months or years (`foldCanonicalDays`). The stored - * per-source WEEK / MONTH / YEAR rows are not read, because the canonical - * source cannot be chosen once per coarse bucket. + * Every granularity is built from the DAY tier. The canonical source is a + * per-DAY decision: a week, month or year in which the user logged most days + * by hand and one day from a device holds readings from both, and each of + * those days counts. Collapsing the stored per-source WEEK / MONTH / YEAR + * rows instead picked ONE source for the whole bucket, so a month of manual + * blood-pressure readings plus one synced one reported the synced day alone, + * and a step history that switched devices mid-month lost every day logged by + * the losing device. The stored coarse rows are therefore not read. + * + * DAY: the day's rows collapse to the ladder-canonical source in + * `collapseRollupRowsBySource`. WEEK / MONTH / YEAR: the same pick runs in + * SQL (`DISTINCT ON` the day, ordered by the ladder rank and then the source + * name, the order the JS collapse uses), and the days fold into their bucket + * there too, so a five-year read returns a handful of rows, not every day. * * A bucket is returned when its start lies in the window: `bucketStart >= * from`, and `< to` (or `<= to` with `toInclusive`). `to === null` leaves the @@ -592,54 +578,105 @@ export async function readCanonicalRollupBuckets(opts: { userPriorityJson?: unknown; }): Promise { const { userId, type, granularity, from, to } = opts; - // The DAY read reaches the end of the last bucket that can start inside - // the window, so that bucket folds complete. - const dayUpper = - to === null - ? undefined - : granularity === "DAY" - ? to - : utcBucketEnd(to, granularity); - const dayRows = await prisma.measurementRollup.findMany({ - where: { - userId, - type, - granularity: "DAY", - bucketStart: - dayUpper === undefined - ? { gte: from } - : granularity === "DAY" && opts.toInclusive - ? { gte: from, lte: dayUpper } - : { gte: from, lt: dayUpper }, - }, - orderBy: { bucketStart: "asc" }, - }); - if (dayRows.length === 0) return []; + if (granularity === "DAY") { + const dayRows = await prisma.measurementRollup.findMany({ + where: { + userId, + type, + granularity: "DAY", + bucketStart: + to === null + ? { gte: from } + : opts.toInclusive + ? { gte: from, lte: to } + : { gte: from, lt: to }, + }, + orderBy: { bucketStart: "asc" }, + select: { + bucketStart: true, + source: true, + count: true, + mean: true, + minValue: true, + maxValue: true, + sumValue: true, + sd: true, + slope: true, + r2: true, + sumX: true, + sumXy: true, + sumXx: true, + sumYy: true, + computedAt: true, + }, + }); + if (dayRows.length === 0) return []; + const priority = + opts.userPriorityJson !== undefined + ? opts.userPriorityJson + : await loadUserSourcePriority(userId); + return collapseRollupRowsBySource(dayRows, type, priority).map( + ({ source: _source, ...day }) => day, + ); + } + const priority = opts.userPriorityJson !== undefined ? opts.userPriorityJson : await loadUserSourcePriority(userId); - const canonicalDays = collapseRollupRowsBySource(dayRows, type, priority).map( - (r): CanonicalRollupBucket => ({ - bucketStart: r.bucketStart, - count: r.count, - mean: r.mean, - minValue: r.minValue, - maxValue: r.maxValue, - sumValue: r.sumValue, - sd: r.sd, - slope: r.slope, - r2: r.r2, - sumX: r.sumX, - sumXy: r.sumXy, - sumXx: r.sumXx, - sumYy: r.sumYy, - computedAt: r.computedAt, - }), + const unit = Prisma.raw(`'${COARSE_TRUNC_UNIT[granularity]}'`); + const rank = Prisma.raw( + buildSourceRankCase(priority, 'r."type"', 'r."source"'), ); + // The DAY read reaches the end of the last bucket that can start inside the + // window, so that bucket folds complete; buckets that start before `from` + // hold only part of their days and are left out below. + const upper = + to === null + ? Prisma.empty + : Prisma.sql`AND r."bucket_start" < ${utcBucketEnd(to, granularity)}`; + const rows = await prisma.$queryRaw` + WITH canon AS ( + SELECT DISTINCT ON (r."bucket_start") + r."bucket_start", r."count", r."mean", r."min_value", r."max_value", + r."sum_value", r."sum_x", r."sum_xy", r."sum_xx", r."sum_yy", + r."computed_at" + FROM measurement_rollups r + WHERE r."user_id" = ${userId} + AND r."type" = ${type}::measurement_type + AND r."granularity" = 'DAY' + AND r."bucket_start" >= ${from} + ${upper} + ORDER BY r."bucket_start", (${rank}), r."source"::text + ) + SELECT + date_trunc(${unit}, c."bucket_start" AT TIME ZONE 'UTC') AT TIME ZONE 'UTC' + AS bucket_start, + SUM(c."count")::int AS count, + SUM(COALESCE(c."sum_value", c."count" * c."mean"))::double precision + AS sum_y, + MIN(c."min_value")::double precision AS min_value, + MAX(c."max_value")::double precision AS max_value, + CASE WHEN COUNT(c."sum_x") = COUNT(*) AND COUNT(c."sum_xy") = COUNT(*) + AND COUNT(c."sum_xx") = COUNT(*) AND COUNT(c."sum_yy") = COUNT(*) + THEN SUM(c."sum_x") END::double precision AS sum_x, + CASE WHEN COUNT(c."sum_x") = COUNT(*) AND COUNT(c."sum_xy") = COUNT(*) + AND COUNT(c."sum_xx") = COUNT(*) AND COUNT(c."sum_yy") = COUNT(*) + THEN SUM(c."sum_xy") END::double precision AS sum_xy, + CASE WHEN COUNT(c."sum_x") = COUNT(*) AND COUNT(c."sum_xy") = COUNT(*) + AND COUNT(c."sum_xx") = COUNT(*) AND COUNT(c."sum_yy") = COUNT(*) + THEN SUM(c."sum_xx") END::double precision AS sum_xx, + CASE WHEN COUNT(c."sum_x") = COUNT(*) AND COUNT(c."sum_xy") = COUNT(*) + AND COUNT(c."sum_xx") = COUNT(*) AND COUNT(c."sum_yy") = COUNT(*) + THEN SUM(c."sum_yy") END::double precision AS sum_yy, + MAX(c."computed_at") AS computed_at + FROM canon c + GROUP BY 1 + ORDER BY 1 + `; const fromMs = from.getTime(); const toMs = to?.getTime() ?? null; - return foldCanonicalDays(canonicalDays, granularity).filter((b) => { + return rows.map(composeFoldedBucket).filter((b) => { const start = b.bucketStart.getTime(); if (start < fromMs) return false; if (toMs === null) return true; diff --git a/src/lib/rollups/tiered-context.ts b/src/lib/rollups/tiered-context.ts index 28ec0a6a6..a5b29605f 100644 --- a/src/lib/rollups/tiered-context.ts +++ b/src/lib/rollups/tiered-context.ts @@ -154,6 +154,11 @@ export interface BuildTieredSeriesOptions { * `skipRecentDaily`. */ coarseOnly?: boolean; + /** + * The user's source-priority blob, when the caller already holds it; the + * bands otherwise load it themselves. + */ + userPriorityJson?: unknown; } interface RollupBucket { @@ -390,6 +395,7 @@ async function readBand( toDaysAgo: number, now: number, tz?: string, + userPriorityJson?: unknown, ): Promise { const from = new Date(now - fromDaysAgo * DAY_MS); const to = new Date(now - toDaysAgo * DAY_MS); @@ -413,7 +419,14 @@ async function readBand( // aggregate instead of silently handing the AI nothing (the module doc // promised replace-with-fallback from day one; only the "replace" half // existed). - const buckets = await readRollupBuckets(userId, type, granularity, from, to); + const buckets = await readRollupBuckets( + userId, + type, + granularity, + from, + to, + userPriorityJson, + ); if (buckets.length > 0) return buckets; return readBandLive(userId, type, granularity, from, to); } @@ -456,6 +469,7 @@ export async function buildTieredSeries( TIERED_BANDS.rawDays, now, options.tz, + options.userPriorityJson, ), coarseOnly ? emptyBand @@ -466,6 +480,8 @@ export async function buildTieredSeries( TIERED_BANDS.weekUntil, TIERED_BANDS.dayUntil, now, + undefined, + options.userPriorityJson, ), readBand( userId, @@ -474,6 +490,8 @@ export async function buildTieredSeries( TIERED_BANDS.monthUntil, TIERED_BANDS.weekUntil, now, + undefined, + options.userPriorityJson, ), readBand( userId, @@ -482,6 +500,8 @@ export async function buildTieredSeries( TIERED_BANDS.yearUntil, TIERED_BANDS.monthUntil, now, + undefined, + options.userPriorityJson, ), ]); diff --git a/tests/integration/rollup-per-day-canonical.test.ts b/tests/integration/rollup-per-day-canonical.test.ts index a720de64b..c24606ade 100644 --- a/tests/integration/rollup-per-day-canonical.test.ts +++ b/tests/integration/rollup-per-day-canonical.test.ts @@ -20,6 +20,7 @@ import { recomputeUserRollups, } from "@/lib/rollups/measurement-rollups"; import { readLiveBuckets } from "@/lib/measurements/daily-series-read"; +import { readTieredRollupSeries } from "@/lib/rollups/measurement-read-wmy"; import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; import { computeSummariesSlice } from "@/lib/analytics/summaries-slice"; import type { MeasurementType } from "@/generated/prisma/client"; @@ -366,4 +367,58 @@ describe("coarse buckets — canonical source per day", () => { expect(slice.summaries.WEIGHT.max).toBe(90); expect(slice.summaries.WEIGHT.mean).toBe(86); }); + + it("the long-range chart and the yearly router fold the same canonical days", async () => { + const user = await seedUser(); + const rows: Row[] = []; + // Two years of blood pressure: typed in every tenth day, a synced cuff + // (higher on the ladder) on every thirtieth day, both on some days. + const start = Date.UTC(2024, 0, 1, 8); + for (let i = 0; i < 730; i += 10) { + rows.push({ + type: "BLOOD_PRESSURE_SYS", + source: "MANUAL", + value: 150, + at: new Date(start + i * DAY_MS), + }); + if (i % 30 === 0) { + rows.push({ + type: "BLOOD_PRESSURE_SYS", + source: "WITHINGS", + value: 120, + at: new Date(start + i * DAY_MS + 3_600_000), + }); + } + } + await getPrismaClient().measurement.createMany({ + data: rows.map((r) => ({ + userId: user.id, + type: r.type, + value: r.value, + unit: "mmHg", + source: r.source, + measuredAt: r.at, + })), + }); + await recomputeUserRollups(user.id, { + from: new Date("2023-12-01T00:00:00.000Z"), + to: new Date("2026-02-01T00:00:00.000Z"), + granularities: ["DAY"], + }); + + const tiered = await readTieredRollupSeries({ + userId: user.id, + type: "BLOOD_PRESSURE_SYS", + from: new Date("2016-01-01T00:00:00.000Z"), + to: new Date("2026-01-31T00:00:00.000Z"), + priorityJson: null, + }); + expect(tiered?.granularity).toBe("WEEK"); + // Every seeded day counts once, with the cuff winning the shared days. + const days = 73; + const cuffDays = 25; + expect(tiered?.rows.reduce((n, r) => n + r.count, 0)).toBe(days); + const total = tiered?.rows.reduce((n, r) => n + r.value * r.count, 0) ?? 0; + expect(total).toBeCloseTo((days - cuffDays) * 150 + cuffDays * 120, 6); + }); });