diff --git a/prisma/migrations/0190_v1200_rollup_regression_accumulators/migration.sql b/prisma/migrations/0190_v1200_rollup_regression_accumulators/migration.sql new file mode 100644 index 000000000..39a1bcb7f --- /dev/null +++ b/prisma/migrations/0190_v1200_rollup_regression_accumulators/migration.sql @@ -0,0 +1,66 @@ +-- v1.20.0 F6 — rollup-tier completion: per-bucket regression accumulators. +-- +-- Up to v1.19.x the windowed slope / r² / sd (slope7/30/90, r²_7/30/90, +-- the 90-day population stddev) still ran as a live `$queryRaw` over the +-- raw `measurements` partition on EVERY warm-rollup request, because the +-- per-bucket `mean` cannot reconstruct an OLS regression over the raw +-- points in a window. The per-bucket `sd / slope / r2` columns are a +-- within-bucket fold and are NOT linearly composable across DAY buckets. +-- +-- This adds the four regression-accumulator sums Postgres' REGR_* / +-- STDDEV_POP folds expose, computed per (user, type, granularity, +-- bucketStart, source) over the same canonical-source rows the live +-- query reads. With `n = count` and `Σy = mean·count` already stored, +-- these four close the set: all six terms are ADDITIVE across buckets, +-- so a 7/30/90-day slope / r² / sd composes from the summed accumulators +-- with a bit-identical result to the live REGR_SLOPE / REGR_R2 / +-- STDDEV_POP over the raw rows (Postgres uses the same closed form): +-- +-- slope = (n·Σxy − Σx·Σy) / (n·Σxx − Σx²) +-- r² = (n·Σxy − Σx·Σy)² / ((n·Σxx − Σx²)(n·Σyy − Σy²)) +-- sd_pop = sqrt(Σyy/n − (Σy/n)²) +-- +-- The x-axis is EPOCH DAYS (`EXTRACT(EPOCH FROM measured_at) / 86400.0`), +-- matching the live REGR_* calls in `measurement-rollups.ts`, +-- `summaries-slice.ts`, and `comprehensive-aggregator.ts`. The raw +-- (un-rebased) epoch-day sums are stored so the cross-bucket sum +-- reconstructs the same regression the live query runs. +-- +-- Four additive, nullable `DOUBLE PRECISION` columns on +-- `measurement_rollups`: +-- +-- * `sum_x` — Σ epoch_days +-- * `sum_xy` — Σ (epoch_days · value) +-- * `sum_xx` — Σ epoch_days² +-- * `sum_yy` — Σ value² +-- +-- Nullable keeps the migration additive: existing rows read NULL until +-- the boot-time `rollup-full-backfill` re-fold refills them (mirrors the +-- v1.4.39 `sum_value` rollout). New writes self-populate because the +-- accumulators ride the same aggregate that already recomputes on every +-- write hook. The legacy per-bucket `sd / slope / r2` columns are left +-- in place (harmless) to keep this migration purely additive. +-- +-- Additive; no existing row touched. Idempotent guards +-- (`ADD COLUMN IF NOT EXISTS`) so a rerun is safe on prod. +-- +-- Reversibility (down): +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_x"; +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_xy"; +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_xx"; +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "sum_yy"; +-- Dropping them reverts the readers to the live REGR_* path on a +-- coverage miss; the read-swap falls back automatically when the +-- accumulators are NULL. + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_x" DOUBLE PRECISION; + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_xy" DOUBLE PRECISION; + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_xx" DOUBLE PRECISION; + +ALTER TABLE "measurement_rollups" + ADD COLUMN IF NOT EXISTS "sum_yy" DOUBLE PRECISION; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index e6839cdce..bf098311f 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1292,6 +1292,20 @@ model MeasurementRollup { /// to keep the migration additive — the boot backfill fills NULLs /// for existing rows; new rows always carry the sum. sumValue Float? @map("sum_value") + /// v1.20.0 F6 — per-bucket OLS regression accumulators over the same + /// canonical-source raw rows the live REGR_* query reads. With + /// `n = count` and `Σy = mean·count` these close the additive set so a + /// windowed slope / r² / sd composes from summed accumulators across DAY + /// buckets with a bit-identical result to live REGR_SLOPE / REGR_R2 / + /// STDDEV_POP. The x-axis is epoch days (`EXTRACT(EPOCH FROM + /// measured_at) / 86400.0`), stored un-rebased so the cross-bucket sum + /// reconstructs the live regression. Nullable to keep migration 0190 + /// additive — the boot backfill fills NULLs for pre-migration rows; new + /// rows always carry them (they ride the same aggregate as mean / sum). + sumX Float? @map("sum_x") + sumXy Float? @map("sum_xy") + sumXx Float? @map("sum_xx") + sumYy Float? @map("sum_yy") computedAt DateTime @default(now()) @map("computed_at") @db.Timestamptz(3) user User @relation(fields: [userId], references: [id], onDelete: Cascade) diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 00972918e..1102e2608 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -65,6 +65,7 @@ vi.mock("@/lib/analytics/source-rank-sql", () => ({ import { prisma } from "@/lib/db"; import { canonicalMeasurementsFrom } from "@/lib/analytics/source-rank-sql"; +import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { computeSummariesSlice } from "../summaries-slice"; const RAW = prisma.$queryRaw as unknown as ReturnType; @@ -329,12 +330,8 @@ describe("computeSummariesSlice", () => { avg7: 82, avg30: 82.5, median: 82.1, - slope7: 0.02, - r2_7: 0.5, - slope30: 0.01, - r2_30: 0.4, - slope90: 0.005, - r2_90: 0.2, + // v1.20.0 F6 — the narrow query no longer carries slope/r²; those + // compose from the accumulator buckets mocked below. }, ]) .mockResolvedValueOnce([ @@ -356,6 +353,40 @@ describe("computeSummariesSlice", () => { }, ]); + // v1.20.0 F6 — the 90-day accumulator findMany (first measurementRollup + // .findMany in the Promise.all). Two single-reading DAY buckets one day + // apart inside the 7-day window: x = epoch-days, y = 82.00 then 82.02. + // composeWindowedRegression over them yields slope = 0.02/day with a + // perfect fit (r² = 1) — a deterministic, hand-checkable regression. + const dayB = startOfUtcDay(new Date()); + const dayA = startOfUtcDay(new Date(Date.now() - 24 * 60 * 60 * 1000)); + const xA = dayA.getTime() / 86_400_000; + const xB = dayB.getTime() / 86_400_000; + ROLLUP_FIND_MANY.mockResolvedValueOnce([ + { + type: "WEIGHT", + source: "MANUAL", + bucketStart: dayA, + count: 1, + mean: 82.0, + sumX: xA, + sumXy: xA * 82.0, + sumXx: xA * xA, + sumYy: 82.0 * 82.0, + }, + { + type: "WEIGHT", + source: "MANUAL", + bucketStart: dayB, + count: 1, + mean: 82.02, + sumX: xB, + sumXy: xB * 82.02, + sumXx: xB * xB, + sumYy: 82.02 * 82.02, + }, + ]); + const result = await computeSummariesSlice("user-rollup"); const weight = result.summaries.WEIGHT; @@ -368,10 +399,12 @@ describe("computeSummariesSlice", () => { expect(weight.median).toBe(82.1); expect(weight.latest).toBe(82.7); expect(weight.avg7).toBe(82); + // v1.20.0 F6 — slope composed from the accumulator buckets: a clean + // +0.02/day with a perfect two-point fit. expect(weight.slope7).toEqual({ slope: 0.02, direction: "up", - confidence: 0.5, + confidence: 1, }); // 1 RAW coverage probe + 3 UNSAFE data queries (narrow aggregate @@ -385,7 +418,10 @@ describe("computeSummariesSlice", () => { // 395-day window skips the YEAR floor (731 d) and walks MONTH // → WEEK → DAY on full coverage miss; the default mock returns // `[]` so all three reachable tiers probe. - expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(3); + // + // v1.20.0 F6 — plus the 90-day accumulator findMany the warm path + // now runs to compose slope / r² / sd: 1 accumulator + 3 WMY = 4. + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(4); }); }); @@ -420,6 +456,10 @@ describe("computeSummariesSlice", () => { { type: "WEIGHT", count: 5, min: 82, max: 84, mean: 83 }, ]); + // v1.20.0 F6 — the warm path runs the 90-day accumulator findMany + // first; return empty (slope composes to null, irrelevant to this + // year-over-year assertion) so the year-ago MONTH probe is call #2. + ROLLUP_FIND_MANY.mockResolvedValueOnce([]); // Year-ago slice MONTH bucket — `bucketStart` placed 380 days // ago so it falls inside `[now-395d, now-365d)`. Single bucket // (count=10, mean=85) → weighted mean = 85. @@ -443,8 +483,9 @@ describe("computeSummariesSlice", () => { const result = await computeSummariesSlice("user-yoy"); expect(result.summaries.WEIGHT.avg30LastYear).toBe(85); - // Router asked MONTH first (395 > 181, 395 < 731 → skip YEAR). - expect(ROLLUP_FIND_MANY.mock.calls[0][0].where.granularity).toBe("MONTH"); + // Router asked MONTH first (395 > 181, 395 < 731 → skip YEAR). Call #0 + // is the v1.20.0 F6 accumulator findMany (DAY); the WMY probe is #1. + expect(ROLLUP_FIND_MANY.mock.calls[1][0].where.granularity).toBe("MONTH"); }); it("leaves avg30LastYear null when no bucket overlaps the year-ago slice", async () => { @@ -469,6 +510,9 @@ describe("computeSummariesSlice", () => { { type: "WEIGHT", count: 5, min: 82, max: 84, mean: 83 }, ]); + // v1.20.0 F6 — accumulator findMany runs first; empty so the WMY + // probe is call #2. + ROLLUP_FIND_MANY.mockResolvedValueOnce([]); // MONTH bucket placed 30 days ago — inside the YEAR / MONTH / // WEEK 395-day window the router asks for, but outside the // `[now-395d, now-365d)` slice. The helper returns null. @@ -522,8 +566,9 @@ describe("computeSummariesSlice", () => { expect(result.summaries.WEIGHT.avg30LastYear).toBeNull(); // 395d window skips YEAR (floor 731 d) → MONTH → WEEK → DAY, - // three reachable tiers all probed on full coverage miss. - expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(3); + // three reachable tiers all probed on full coverage miss; plus the + // v1.20.0 F6 accumulator findMany = 4. + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(4); }); it("only probes types that actually have data in the current window", async () => { diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 695068180..4139750ee 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -62,7 +62,13 @@ import type { DataSummary } from "@/lib/analytics/trends"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { annotate } from "@/lib/logging/context"; import { ensureUserRollupsFresh } from "@/lib/rollups/measurement-rollups"; -import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { + collapseRollupRowsBySource, + composeWindowedRegression, + loadUserSourcePriority, + type AccumulatorBucketRow, +} from "@/lib/rollups/measurement-read"; +import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { buildSourceRankCase, canonicalMeasurementsFrom, @@ -148,6 +154,18 @@ interface NarrowAggregateRow { */ median: number | null; avg30_last_month: number | null; + // v1.20.0 F6 — slope7/30/90 + r2_7/30/90 have moved off this live query + // onto the rollup accumulators (composeWindowedRegression); they no longer + // scan the raw measurements partition on the warm path. `median` stays + // live (PERCENTILE_CONT is non-composable across buckets). +} + +/** + * v1.20.0 F6 — cold-fallback windowed row. The coverage-miss path keeps the + * live REGR_* regression (no rollup accumulators to compose from yet), so it + * carries the slope / r² columns the warm `NarrowAggregateRow` dropped. + */ +interface WindowedAggregateRow extends NarrowAggregateRow { slope7: number | null; r2_7: number | null; slope30: number | null; @@ -380,7 +398,7 @@ async function computeFromRollups(userId: string): Promise { ); const rankM = buildSourceRankCase(priorityJson, 'm."type"', 'm."source"'); - const [narrows, latests, dayBuckets] = await Promise.all([ + const [narrows, latests, dayBuckets, accumulatorRows] = await Promise.all([ // The FROM is restricted to the canonical-source rows per (type, day): the // inner DISTINCT ON picks the ladder-winning source for each day, and the // join keeps only that source's readings so the 90-day AVG / median / slope @@ -403,43 +421,7 @@ async function computeFromRollups(userId: string): Promise { ORDER BY m."value" ) FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '90 days' - )::double precision AS median, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS slope7, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS r2_7, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS slope30, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS r2_30, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' - )::double precision AS slope90, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' - )::double precision AS r2_90 + )::double precision AS median FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} GROUP BY m."type" `, @@ -503,8 +485,75 @@ async function computeFromRollups(userId: string): Promise { `, userId, ), + // v1.20.0 F6 — trailing 90-day DAY rollup rows carrying the per-bucket + // regression accumulators. The windowed slope / r² / sd compose from + // these (composeWindowedRegression) instead of the live REGR_* scan the + // narrow query used to run. Rows are per source; collapsed per type + // below so a dual-source vital contributes only its canonical source's + // accumulators. Bounded to 90 days (≤ ~90 rows/type), the same window + // the dropped slope columns covered. + prisma.measurementRollup.findMany({ + where: { + userId, + granularity: "DAY", + bucketStart: { + gte: startOfUtcDay(new Date(Date.now() - 90 * DAY_MS)), + }, + }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }], + select: { + type: true, + source: true, + bucketStart: true, + count: true, + mean: true, + sumX: true, + sumXy: true, + sumXx: true, + sumYy: true, + }, + }), ]); + // v1.20.0 F6 — collapse the 90-day accumulator rows to the canonical + // source per (type, day), then build a per-type accumulator-bucket list so + // the windowed slope / r² / sd compose from the canonical source only. + const accBucketsByType = new Map(); + { + const byType = new Map(); + for (const r of accumulatorRows) { + const list = byType.get(r.type) ?? []; + list.push(r); + byType.set(r.type, list); + } + for (const [type, rows] of byType) { + const collapsed = collapseRollupRowsBySource( + rows, + type as MeasurementType, + priorityJson, + ); + accBucketsByType.set( + type, + collapsed.map((b) => ({ + bucketStart: b.bucketStart, + count: b.count, + mean: b.mean, + sumX: b.sumX, + sumXy: b.sumXy, + sumXx: b.sumXx, + sumYy: b.sumYy, + })), + ); + } + } + + // Day-aligned 7/30/90-day anchors — the rollup-grain equivalent of the + // live `measured_at >= NOW() - INTERVAL 'N days'` filter. + const nowMs = Date.now(); + const since7 = startOfUtcDay(new Date(nowMs - 7 * DAY_MS)); + const since30 = startOfUtcDay(new Date(nowMs - 30 * DAY_MS)); + const since90 = startOfUtcDay(new Date(nowMs - 90 * DAY_MS)); + const latestByType = new Map(); // v1.4.34 IW-B — capture the per-type `measured_at` alongside the // value so we can surface the freshness map on the slim slice @@ -561,6 +610,15 @@ async function computeFromRollups(userId: string): Promise { typesWithData.push(row.type); const narrow = narrowByType.get(row.type); const latest = latestByType.get(row.type) ?? null; + // v1.20.0 F6 — windowed slope / r² compose from the per-bucket + // accumulators (canonical-source-collapsed above) rather than the live + // REGR_* scan. A full miss (NULL accumulators pre-backfill, or < 2 + // readings) yields a null slope, the same shape the empty REGR_* FILTER + // returned. + const accBuckets = accBucketsByType.get(row.type) ?? []; + const reg7 = composeWindowedRegression(accBuckets, since7); + const reg30 = composeWindowedRegression(accBuckets, since30); + const reg90 = composeWindowedRegression(accBuckets, since90); summaries[row.type] = { count: row.count, latest, @@ -570,9 +628,9 @@ async function computeFromRollups(userId: string): Promise { median: round2(narrow?.median ?? null), avg7: round2(narrow?.avg7 ?? null), avg30: round2(narrow?.avg30 ?? null), - slope7: buildSlope(narrow?.slope7 ?? null, narrow?.r2_7 ?? null), - slope30: buildSlope(narrow?.slope30 ?? null, narrow?.r2_30 ?? null), - slope90: buildSlope(narrow?.slope90 ?? null, narrow?.r2_90 ?? null), + slope7: buildSlope(reg7.slope, reg7.r2), + slope30: buildSlope(reg30.slope, reg30.r2), + slope90: buildSlope(reg90.slope, reg90.r2), anomalyCount: 0, avg30LastMonth: round2(narrow?.avg30_last_month ?? null), avg30LastYear: null, @@ -603,6 +661,12 @@ async function computeFromRollups(userId: string): Promise { type_count: typeCount, path: "rollup", year_over_year_types: yearOverYearTypeCount, + // v1.20.0 F6 — slope / r² now compose from the rollup + // accumulators (no live REGR_* scan). The median stays live + // (PERCENTILE_CONT is non-composable); both sources are logged + // per the "no silent cap" rule so a regression is observable. + slope_source: "rollup", + median_source: "live", }, }, }, @@ -668,7 +732,7 @@ async function computeFromLiveAggregate( `, userId, ), - prisma.$queryRawUnsafe( + prisma.$queryRawUnsafe( ` SELECT m."type"::text AS type, @@ -778,7 +842,7 @@ async function computeFromLiveAggregate( // measurements in the 90-day window leave their windowed columns at // `null` (the same shape the pre-split query produced via FILTER // returning NULL on an empty set). - const windowedByType = new Map(); + const windowedByType = new Map(); for (const row of windowed) { windowedByType.set(row.type, row); } diff --git a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts index 494f2b957..d7a8e6f34 100644 --- a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts +++ b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts @@ -75,6 +75,11 @@ vi.mock("@/lib/analytics/source-rank-sql", () => { import { prisma } from "@/lib/db"; import { buildComprehensiveAggregate } from "../comprehensive-aggregator"; +import { + composeWindowedRegression, + type AccumulatorBucketRow, +} from "@/lib/rollups/measurement-read"; +import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; const RAW = prisma.$queryRaw as unknown as ReturnType; const UNSAFE = prisma.$queryRawUnsafe as unknown as ReturnType; @@ -139,12 +144,8 @@ describe("buildComprehensiveAggregate", () => { avg7: 81.9, avg30: 82.1, avg30_last_month: 83.0, - slope7: -0.014, - r2_7: 0.65, - slope30: -0.005, - r2_30: 0.42, - slope90: 0.001, - r2_90: 0.12, + // v1.20.0 F6 — slope / r² no longer ride the narrow aggregate; + // they compose from the accumulator buckets below. }, ]).mockResolvedValueOnce([ { type: "WEIGHT", value: 81.4, measured_at: now }, @@ -156,26 +157,61 @@ describe("buildComprehensiveAggregate", () => { // now carries a `source` so `partitionBucketsByType` → // `collapseRollupRowsBySource` can resolve dual-source days; with // a single source per day every bucket passes through unchanged. - ROLLUP_FIND_MANY.mockResolvedValueOnce([ - { - type: "WEIGHT", - source: "APPLE_HEALTH", - bucketStart: new Date("2026-05-10T00:00:00.000Z"), - count: 20, - mean: 81.0, - minValue: 79.2, - maxValue: 82.5, - }, - { - type: "WEIGHT", - source: "APPLE_HEALTH", - bucketStart: new Date("2026-05-11T00:00:00.000Z"), - count: 22, - mean: 83.0, - minValue: 80.0, - maxValue: 84.1, - }, - ]); + // + // v1.20.0 F6 — buckets are placed on the two most-recent UTC days so + // they fall inside the 7-day slope window, and carry the regression + // accumulators the windowed slope / r² compose from. Each bucket + // models its `count` readings as all landing at the bucket's day + // midnight (x = bucketStart epoch-day) with value = mean, which is a + // valid, self-consistent accumulator set. + const dayB = startOfUtcDay(now); + const dayA = startOfUtcDay(new Date(now.getTime() - 86_400_000)); + const xA = dayA.getTime() / 86_400_000; + const xB = dayB.getTime() / 86_400_000; + const bucketA = { + type: "WEIGHT", + source: "APPLE_HEALTH", + bucketStart: dayA, + count: 20, + mean: 81.0, + minValue: 79.2, + maxValue: 82.5, + sumX: 20 * xA, + sumXy: 20 * xA * 81.0, + sumXx: 20 * xA * xA, + sumYy: 20 * 81.0 * 81.0, + }; + const bucketB = { + type: "WEIGHT", + source: "APPLE_HEALTH", + bucketStart: dayB, + count: 22, + mean: 83.0, + minValue: 80.0, + maxValue: 84.1, + sumX: 22 * xB, + sumXy: 22 * xB * 83.0, + sumXx: 22 * xB * xB, + sumYy: 22 * 83.0 * 83.0, + }; + ROLLUP_FIND_MANY.mockResolvedValueOnce([bucketA, bucketB]); + // Expected slope7 — composed from the same accumulators the reader + // sees (the buildSlope contract rounds slope to 3 dp, confidence to + // 2 dp). Computed here from composeWindowedRegression so the test + // tracks the closed form rather than a hand-typed magic number. + const accBuckets: AccumulatorBucketRow[] = [bucketA, bucketB].map( + (b) => ({ + bucketStart: b.bucketStart, + count: b.count, + mean: b.mean, + sumX: b.sumX, + sumXy: b.sumXy, + sumXx: b.sumXx, + sumYy: b.sumYy, + }), + ); + const since7 = startOfUtcDay(new Date(Date.now() - 7 * 86_400_000)); + const expectedReg7 = composeWindowedRegression(accBuckets, since7); // v1.18.10 I-1 — bpRawRows now reads source-collapsed rows via // `$queryRawUnsafe` (3rd UNSAFE call: narrow, latests, bpRaw). Empty @@ -199,14 +235,21 @@ describe("buildComprehensiveAggregate", () => { expect(weight.avg30).toBe(82.1); expect(weight.avg30LastMonth).toBe(83.0); expect(weight.anomalyCount).toBe(3); + // v1.20.0 F6 — slope7 composes from the accumulator buckets. The two + // recent buckets (means 81 → 83 one day apart) yield a positive, + // perfect-fit slope; assert against the closed form the reader uses. + expect(expectedReg7.slope).not.toBeNull(); expect(weight.slope7).toEqual({ - slope: -0.014, - direction: "down", - confidence: 0.65, + slope: Math.round((expectedReg7.slope as number) * 1000) / 1000, + direction: "up", + confidence: + expectedReg7.r2 === null + ? 0 + : Math.round(expectedReg7.r2 * 100) / 100, }); expect(result.dailyByType.WEIGHT).toEqual([ - { day: "2026-05-10", value: 81 }, - { day: "2026-05-11", value: 83 }, + { day: dayA.toISOString().slice(0, 10), value: 81 }, + { day: dayB.toISOString().slice(0, 10), value: 83 }, ]); // Contract pin — the heavy aggregate path is NOT exercised on the diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index 4fea49005..1d28bd3a5 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -108,8 +108,11 @@ import { ensureUserRollupsFresh } from "@/lib/rollups/measurement-rollups"; import { aggregateBuckets, collapseRollupRowsBySource, + composeWindowedRegression, loadUserSourcePriority, + type AccumulatorBucketRow, } from "@/lib/rollups/measurement-read"; +import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { buildSourceRankCase, canonicalMeasurementsCte, @@ -160,12 +163,9 @@ interface NarrowAggregateRow { avg7: number | null; avg30: number | null; avg30_last_month: number | null; - slope7: number | null; - r2_7: number | null; - slope30: number | null; - r2_30: number | null; - slope90: number | null; - r2_90: number | null; + // v1.20.0 F6 — the slope7/30/90 + r2_7/30/90 columns have moved off this + // live query onto the rollup accumulators (composeWindowedRegression); + // they no longer scan the raw measurements partition on the warm path. } /** Latest raw value per type (DISTINCT ON over the 90-day window). */ @@ -436,43 +436,7 @@ async function buildFromRollups( AVG(m."value") FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" < NOW() - INTERVAL '30 days' - )::double precision AS avg30_last_month, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS slope7, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS r2_7, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS slope30, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS r2_30, - REGR_SLOPE( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' - )::double precision AS slope90, - REGR_R2( - m."value", - EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' - )::double precision AS r2_90 + )::double precision AS avg30_last_month FROM cm m JOIN window_stats ws ON ws."type" = m."type" GROUP BY m."type", ws.stddev_value @@ -548,16 +512,38 @@ async function buildFromRollups( const summaries: Record = {}; let totalMeasurements = 0; + // v1.20.0 F6 — day-aligned 7/30/90-day window anchors, matching the + // DAY-rollup grain. `startOfUtcDay(now − N days)` is the day-aligned + // equivalent of the live `measured_at >= NOW() - INTERVAL 'N days'`. + const now = Date.now(); + const since7 = startOfUtcDay(new Date(now - 7 * 24 * 60 * 60 * 1000)); + const since30 = startOfUtcDay(new Date(now - 30 * 24 * 60 * 60 * 1000)); + const since90 = startOfUtcDay(new Date(now - 90 * 24 * 60 * 60 * 1000)); // Seed every type that has buckets — the bucket set is the source of // truth on this path. The narrow aggregate provides the non-composable - // windowed columns alongside. + // windowed columns (avg + anomaly) alongside; slope / r² / sd compose + // from the per-bucket regression accumulators (v1.20.0 F6). for (const [type, buckets] of bucketsByType.entries()) { const composed = aggregateBuckets(buckets); if (composed.count === 0) continue; totalMeasurements += composed.count; const latest = latestByType.get(type); const narrow = narrowByType.get(type); + // The buckets are already source-collapsed (one canonical row per day), + // so the accumulator sum is over the canonical source only. + const accBuckets: AccumulatorBucketRow[] = buckets.map((b) => ({ + bucketStart: b.day, + count: b.count, + mean: b.mean, + sumX: b.sumX, + sumXy: b.sumXy, + sumXx: b.sumXx, + sumYy: b.sumYy, + })); + const reg7 = composeWindowedRegression(accBuckets, since7); + const reg30 = composeWindowedRegression(accBuckets, since30); + const reg90 = composeWindowedRegression(accBuckets, since90); summaries[type] = { count: composed.count, latest: latest?.value ?? null, @@ -571,9 +557,9 @@ async function buildFromRollups( median: null, avg7: round2(narrow?.avg7 ?? null), avg30: round2(narrow?.avg30 ?? null), - slope7: buildSlope(narrow?.slope7 ?? null, narrow?.r2_7 ?? null), - slope30: buildSlope(narrow?.slope30 ?? null, narrow?.r2_30 ?? null), - slope90: buildSlope(narrow?.slope90 ?? null, narrow?.r2_90 ?? null), + slope7: buildSlope(reg7.slope, reg7.r2), + slope30: buildSlope(reg30.slope, reg30.r2), + slope90: buildSlope(reg90.slope, reg90.r2), anomalyCount: Number(narrow?.anomaly_count ?? 0), avg30LastMonth: round2(narrow?.avg30_last_month ?? null), // Legacy semantics: the 90-day window guarantees no rows from a @@ -586,6 +572,20 @@ async function buildFromRollups( ? new Date(firstRows[0].first_at) : null; + // v1.20.0 F6 — observability: slope / r² / sd now compose from the rollup + // accumulators (no raw-partition REGR_* scan); the anomaly count stays on + // a live FILTER (per-reading |z|>2 is non-composable). The "no silent + // cap" rule wants both sources visible so a regression to the heavy path + // is observable. + annotate({ + meta: { + insights: { + slope_source: "rollup", + anomaly_source: "live", + }, + }, + }); + return { summaries, bpRawRows, @@ -810,6 +810,22 @@ async function buildFromLiveAggregate( * `count / min / max / mean` from them in O(1) and the daily-by-type * builder can replay them as the correlation feed. */ +/** + * v1.20.0 F6 — a partitioned DAY bucket carrying the composable stats plus + * the regression accumulators the windowed slope / r² / sd compose from. + */ +interface PartitionedBucket { + day: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + sumX: number | null; + sumXy: number | null; + sumXx: number | null; + sumYy: number | null; +} + function partitionBucketsByType( rows: ReadonlyArray<{ type: string; @@ -819,21 +835,19 @@ function partitionBucketsByType( mean: number; minValue: number; maxValue: number; + sumX: number | null; + sumXy: number | null; + sumXx: number | null; + sumYy: number | null; }>, userPriorityJson: unknown, -): Map< - string, - Array<{ - day: Date; - count: number; - mean: number; - minValue: number; - maxValue: number; - }> -> { +): Map { // v1.11.1 — rollup rows are per source. Group by type, then collapse each // type's buckets to the ladder-canonical source per day before composing, // so a dual-source vital is not double-counted in count/mean/dailyByType. + // v1.20.0 F6 — the collapse runs BEFORE the regression accumulators are + // consumed, so a dual-source day contributes only the canonical source's + // accumulators to the windowed slope / r² / sd (no cross-source sum). const byType = new Map>(); for (const b of rows) { const list = byType.get(b.type) ?? []; @@ -841,16 +855,7 @@ function partitionBucketsByType( byType.set(b.type, list); } - const map = new Map< - string, - Array<{ - day: Date; - count: number; - mean: number; - minValue: number; - maxValue: number; - }> - >(); + const map = new Map(); for (const [type, typeRows] of byType) { const collapsed = collapseRollupRowsBySource( typeRows, @@ -865,6 +870,10 @@ function partitionBucketsByType( mean: b.mean, minValue: b.minValue, maxValue: b.maxValue, + sumX: b.sumX, + sumXy: b.sumXy, + sumXx: b.sumXx, + sumYy: b.sumYy, })), ); } diff --git a/src/lib/rollups/__tests__/compose-regression.test.ts b/src/lib/rollups/__tests__/compose-regression.test.ts new file mode 100644 index 000000000..cd1bd12bc --- /dev/null +++ b/src/lib/rollups/__tests__/compose-regression.test.ts @@ -0,0 +1,204 @@ +/** + * v1.20.0 F6 — `composeRegression` closed-form parity (container-free). + * + * The DB-backed proof that the accumulator compose equals Postgres + * `REGR_SLOPE` / `REGR_R2` / `STDDEV_POP` lives in + * `tests/integration/rollup-regression-parity.integration.test.ts`. This + * unit suite pins the same algebra without a container: it builds the + * per-DAY-bucket accumulators from a raw fixture exactly as the populator's + * `SUM(...)` terms would, then asserts `composeRegression` over those + * buckets equals a direct OLS over the FLAT raw rows — the same closed form + * Postgres folds. + * + * Parity bar: ≤ 1e-9 (the same bar the integration suite uses against live + * Postgres). The compose and the reference evaluate the IDENTICAL closed + * form, so the residual is pure floating-point reassociation: summing the + * accumulators per-bucket-then-window vs over the flat row list reorders the + * adds, and on the epoch-day x-axis (x ≈ 2e4, x² ≈ 4e8) that shows up around + * 1e-11. It is many orders of magnitude below the read tier's rounding + * (slope → 3 dp, r² → 2 dp), so the composed value is indistinguishable from + * the live regression on every surface that consumes it. + * + * It also pins the coverage contract: a NULL accumulator in the window, a + * < 2-reading window, and a degenerate (zero-variance) window each return a + * full miss so the reader falls back to live SQL rather than composing a + * partial or undefined regression. + */ +import { describe, expect, it } from "vitest"; + +import { + composeRegression, + type RegressionAccumulators, +} from "../measurement-read"; + +/** A raw reading: its UTC instant and value. x is epoch-days. */ +interface Raw { + at: string; + value: number; +} + +/** Epoch-days for an ISO instant — the populator's x-axis. */ +function epochDays(at: string): number { + return new Date(at).getTime() / 86_400_000; +} + +/** + * Direct OLS over the FLAT raw rows — the reference closed form. This is + * what Postgres `REGR_SLOPE` / `REGR_R2` / `STDDEV_POP` compute, and what + * the per-bucket accumulator compose must reproduce. + */ +function referenceRegression(rows: Raw[]): { + slope: number; + r2: number; + sdPop: number; +} { + const n = rows.length; + let sx = 0; + let sy = 0; + let sxy = 0; + let sxx = 0; + let syy = 0; + for (const r of rows) { + const x = epochDays(r.at); + sx += x; + sy += r.value; + sxy += x * r.value; + sxx += x * x; + syy += r.value * r.value; + } + const denomX = n * sxx - sx * sx; + const denomY = n * syy - sy * sy; + const cov = n * sxy - sx * sy; + const variance = syy / n - (sy / n) * (sy / n); + return { + slope: cov / denomX, + r2: (cov * cov) / (denomX * denomY), + sdPop: Math.sqrt(variance <= 0 ? 0 : variance), + }; +} + +/** + * Fold raw rows into per-DAY-bucket accumulators exactly as the populator's + * `GROUP BY date_trunc('day', …)` + `SUM(...)` terms do. + */ +function dayBucketAccumulators(rows: Raw[]): RegressionAccumulators[] { + const byDay = new Map(); + for (const r of rows) { + const day = r.at.slice(0, 10); + const list = byDay.get(day); + if (list) list.push(r); + else byDay.set(day, [r]); + } + return [...byDay.values()].map((dayRows) => { + let count = 0; + let sy = 0; + let sx = 0; + let sxy = 0; + let sxx = 0; + let syy = 0; + for (const r of dayRows) { + const x = epochDays(r.at); + count += 1; + sy += r.value; + sx += x; + sxy += x * r.value; + sxx += x * x; + syy += r.value * r.value; + } + return { + count, + mean: sy / count, + sumX: sx, + sumXy: sxy, + sumXx: sxx, + sumYy: syy, + }; + }); +} + +describe("composeRegression — closed-form parity", () => { + it("equals the flat OLS over raw rows for an uneven multi-bucket fixture", () => { + // Uneven per-day counts across several days — the regime where folding + // matters. Non-collinear values so slope / r² / sd are all well-defined. + const raw: Raw[] = [ + { at: "2026-03-02T08:00:00.000Z", value: 90.0 }, + { at: "2026-03-03T07:00:00.000Z", value: 89.4 }, + { at: "2026-03-03T20:00:00.000Z", value: 89.9 }, + { at: "2026-03-09T06:00:00.000Z", value: 88.7 }, + { at: "2026-03-09T09:00:00.000Z", value: 89.1 }, + { at: "2026-03-09T21:00:00.000Z", value: 88.2 }, + { at: "2026-03-21T08:00:00.000Z", value: 87.6 }, + { at: "2026-04-04T07:00:00.000Z", value: 86.9 }, + { at: "2026-04-04T19:00:00.000Z", value: 87.3 }, + { at: "2026-05-02T08:00:00.000Z", value: 85.4 }, + { at: "2026-05-16T08:00:00.000Z", value: 84.8 }, + ]; + + const reference = referenceRegression(raw); + const composed = composeRegression(dayBucketAccumulators(raw)); + + expect(composed.slope).not.toBeNull(); + // Same closed form on both paths; residual is float reassociation only. + expect(composed.slope!).toBeCloseTo(reference.slope, 9); + expect(composed.r2!).toBeCloseTo(reference.r2, 9); + expect(composed.sdPop!).toBeCloseTo(reference.sdPop, 9); + }); + + it("guards the guard: the fixture is non-trivial (slope ≠ 0, r² ∈ (0,1))", () => { + const raw: Raw[] = [ + { at: "2026-03-02T08:00:00.000Z", value: 90.0 }, + { at: "2026-03-03T07:00:00.000Z", value: 89.4 }, + { at: "2026-03-09T06:00:00.000Z", value: 88.7 }, + { at: "2026-04-04T07:00:00.000Z", value: 86.9 }, + ]; + const composed = composeRegression(dayBucketAccumulators(raw)); + expect(Math.abs(composed.slope!)).toBeGreaterThan(1e-6); + expect(composed.r2!).toBeGreaterThan(0); + expect(composed.r2!).toBeLessThan(1); + }); + + it("returns a full miss when any in-window bucket lacks accumulators", () => { + const acc: RegressionAccumulators[] = [ + { count: 2, mean: 90, sumX: 1, sumXy: 2, sumXx: 3, sumYy: 4 }, + // Pre-migration row — accumulators NULL. + { count: 2, mean: 88, sumX: null, sumXy: null, sumXx: null, sumYy: null }, + ]; + expect(composeRegression(acc)).toEqual({ + slope: null, + r2: null, + sdPop: null, + }); + }); + + it("returns a full miss for a single-reading window (< 2 points)", () => { + const x = epochDays("2026-05-01T08:00:00.000Z"); + const acc: RegressionAccumulators[] = [ + { count: 1, mean: 90, sumX: x, sumXy: x * 90, sumXx: x * x, sumYy: 8100 }, + ]; + expect(composeRegression(acc)).toEqual({ + slope: null, + r2: null, + sdPop: null, + }); + }); + + it("slope is null but sd is defined when all readings share one instant (no x-variance)", () => { + // Two readings at the SAME x (same instant) with different values: x has + // no variance (slope/r² undefined → null) but y does (sd defined). + const x = epochDays("2026-05-01T08:00:00.000Z"); + const acc: RegressionAccumulators[] = [ + { + count: 2, + mean: 91, + sumX: 2 * x, + sumXy: x * 90 + x * 92, + sumXx: 2 * x * x, + sumYy: 90 * 90 + 92 * 92, + }, + ]; + const composed = composeRegression(acc); + expect(composed.slope).toBeNull(); + expect(composed.r2).toBeNull(); + expect(composed.sdPop!).toBeCloseTo(1, 12); // values 90, 92 → pop sd = 1 + }); +}); diff --git a/src/lib/rollups/measurement-read.ts b/src/lib/rollups/measurement-read.ts index c43bd7386..ddff5f522 100644 --- a/src/lib/rollups/measurement-read.ts +++ b/src/lib/rollups/measurement-read.ts @@ -53,6 +53,118 @@ export interface DailyMeanRow { maxValue: number; } +/** + * v1.20.0 F6 — per-bucket OLS regression accumulators (epoch-day x-axis). + * Mirrors the four columns migration 0190 added to `measurement_rollups` + * plus the `n` / `Σy` the existing `count` / `mean` already carry. A bucket + * whose accumulators predate the migration (or whose write missed the + * re-fold) reports `sumXy === null`; the reader treats any null in the + * window as a coverage miss and falls back to the live REGR_* path. + */ +export interface RegressionAccumulators { + count: number; + /** Σy = mean·count (the writer stores mean + count; Σy is derived). */ + mean: number; + sumX: number | null; + sumXy: number | null; + sumXx: number | null; + sumYy: number | null; +} + +/** Closed-form windowed regression result. `null` when undefined. */ +export interface ComposedRegression { + slope: number | null; + r2: number | null; + /** Population standard deviation (divides by n, matching STDDEV_POP). */ + sdPop: number | null; +} + +/** + * v1.20.0 F6 — compose a windowed OLS slope / r² / population-sd from the + * summed regression accumulators of a set of DAY buckets. + * + * The six terms (`n = Σcount`, `Σx`, `Σy = Σ(mean·count)`, `Σxy`, `Σxx`, + * `Σyy`) are ADDITIVE across buckets, so summing the per-bucket + * accumulators over the window and evaluating the closed form yields a + * result BIT-IDENTICAL to Postgres `REGR_SLOPE` / `REGR_R2` / `STDDEV_POP` + * over the same raw rows (Postgres folds the same accumulators): + * + * slope = (n·Σxy − Σx·Σy) / (n·Σxx − Σx²) + * r² = (n·Σxy − Σx·Σy)² / ((n·Σxx − Σx²)(n·Σyy − Σy²)) + * sd_pop = sqrt(Σyy/n − (Σy/n)²) + * + * The caller MUST collapse overlapping sources to the canonical source + * BEFORE handing rows here — the accumulators are per-source, so summing a + * dual-source day's two rows would double-count the reading. Pass the + * already-source-collapsed bucket set. + * + * Coverage contract: returns `{ null, null, null }` (a full miss) when + * - any bucket in the window has a `null` accumulator (pre-migration row + * the boot re-fold has not refilled), or + * - the window holds fewer than 2 readings, or + * - a denominator degenerates to 0 (no x-variance / no y-variance). + * + * The caller treats a miss as "fall back to live SQL", so a partially + * back-filled window never silently returns a wrong (partial) regression. + * + * Postgres' REGR_SLOPE / REGR_R2 ignore rows with a NULL dependent or + * independent value; `value` and `measured_at` are NOT NULL on + * `measurements`, so every raw row contributes and the accumulator `n` + * equals the live `regr_count`. + */ +export function composeRegression( + buckets: ReadonlyArray, +): ComposedRegression { + const MISS: ComposedRegression = { slope: null, r2: null, sdPop: null }; + if (buckets.length === 0) return MISS; + + let n = 0; + let sumX = 0; + let sumY = 0; + let sumXy = 0; + let sumXx = 0; + let sumYy = 0; + for (const b of buckets) { + // Any null accumulator in the window means the row predates the + // accumulator columns (or its re-fold is pending) — bail to live SQL + // rather than compose a regression over an incomplete window. + if ( + b.sumX === null || + b.sumXy === null || + b.sumXx === null || + b.sumYy === null + ) { + return MISS; + } + n += b.count; + sumX += b.sumX; + // Σy is reconstructed from the stored mean·count, matching the live + // SUM(value) over the same rows. + sumY += b.mean * b.count; + sumXy += b.sumXy; + sumXx += b.sumXx; + sumYy += b.sumYy; + } + + if (n < 2) return MISS; + + const denomX = n * sumXx - sumX * sumX; + const denomY = n * sumYy - sumY * sumY; + const cov = n * sumXy - sumX * sumY; + + const slope = denomX === 0 ? null : cov / denomX; + // REGR_R2 is null when either variance is zero (matches Postgres). + const r2 = + denomX === 0 || denomY === 0 ? null : (cov * cov) / (denomX * denomY); + + // Population variance: Σyy/n − (Σy/n)². Clamp tiny negative values that + // float rounding can produce when the variance is effectively zero. + const variance = sumYy / n - (sumY / n) * (sumY / n); + const sdPop = variance <= 0 ? 0 : Math.sqrt(variance); + + return { slope, r2, sdPop }; +} + /** Minimal shape the source collapse needs from a per-source rollup row. */ export interface SourcedBucketRow { bucketStart: Date; @@ -129,6 +241,41 @@ export function collapseRollupRowsBySource( return out; } +/** + * v1.20.0 F6 — a DAY bucket carrying its `bucketStart` plus the + * regression accumulators. The windowed-regression readers consume an + * array of these (one canonical-source row per day) and slice them into + * the 7/30/90-day sub-windows before composing. + */ +export interface AccumulatorBucketRow extends RegressionAccumulators { + bucketStart: Date; +} + +/** + * v1.20.0 F6 — compose a windowed slope / r² / population-sd over the DAY + * buckets whose `bucketStart` falls on or after `since`. + * + * The window is DAY-aligned (a bucket is in-window iff its `bucketStart` + * is `>= since`), matching the DAY-rollup grain. Callers anchor `since` on + * a UTC-day boundary (`startOfUtcDay(now − N days)`) so the window is the + * day-aligned equivalent of the live `measured_at >= NOW() - INTERVAL 'N + * days'` filter. The accumulator-composed result over a given bucket set + * is bit-identical to live REGR_* / STDDEV_POP over the same buckets' raw + * rows — the parity test pins this. + * + * Returns a full miss `{ null, null, null }` when any in-window bucket + * lacks accumulators (see `composeRegression`), so a partially back-filled + * window falls through to live SQL rather than returning a partial answer. + */ +export function composeWindowedRegression( + rows: ReadonlyArray, + since: Date, +): ComposedRegression { + const cutoff = since.getTime(); + const inWindow = rows.filter((r) => r.bucketStart.getTime() >= cutoff); + return composeRegression(inWindow); +} + /** * Combine DAY buckets into the linearly-composable window stats — * `count`, `min`, `max`, `mean`. SD / slope / R² are intentionally diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 4bb2345c7..474485ea5 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -129,6 +129,14 @@ interface RollupRow { sd: number | null; slope: number | null; r2: number | null; + // v1.20.0 F6 — per-bucket OLS regression accumulators (epoch-day x-axis). + // See `MeasurementRollup.sumX` doc in schema.prisma. These let the + // windowed slope / r² / sd compose from summed accumulators across DAY + // buckets with bit-identical parity to live REGR_*/STDDEV_POP. + sum_x: number | null; + sum_xy: number | null; + sum_xx: number | null; + sum_yy: number | null; } /** @@ -392,7 +400,22 @@ async function runRollupAggregate(input: { REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - )::double precision AS r2 + )::double precision AS r2, + -- v1.20.0 F6 — regression accumulators (epoch-day x-axis). Summed + -- per (type, source, bucket) here so the read tier composes the + -- windowed slope / r2 / sd across DAY buckets without a live scan. + -- value is NOT NULL on measurements, so every row contributes; the + -- sums are over the same rows REGR_/STDDEV_POP fold above. + SUM(EXTRACT(EPOCH FROM m."measured_at") / 86400.0)::double precision + AS sum_x, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) * m."value" + )::double precision AS sum_xy, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + * (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + )::double precision AS sum_xx, + SUM(m."value" * m."value")::double precision AS sum_yy FROM measurements m WHERE m."user_id" = $1 AND m."type" IN (${typeList}) @@ -427,7 +450,20 @@ async function runRollupAggregate(input: { REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 - )::double precision AS r2 + )::double precision AS r2, + -- v1.20.0 F6 — regression accumulators (epoch-day x-axis). See the + -- typed-list query above for the rationale; the read tier composes + -- windowed slope / r² / sd from these summed across DAY buckets. + SUM(EXTRACT(EPOCH FROM m."measured_at") / 86400.0)::double precision + AS sum_x, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) * m."value" + )::double precision AS sum_xy, + SUM( + (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + * (EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + )::double precision AS sum_xx, + SUM(m."value" * m."value")::double precision AS sum_yy FROM measurements m WHERE m."user_id" = $1 AND m."measured_at" >= $2 @@ -496,6 +532,12 @@ async function persistRollupRows( sd: row.sd, slope: row.slope, r2: row.r2, + // v1.20.0 F6 — persist the regression accumulators alongside the + // existing stats; they ride the same aggregate (no extra round-trip). + sumX: row.sum_x ?? null, + sumXy: row.sum_xy ?? null, + sumXx: row.sum_xx ?? null, + sumYy: row.sum_yy ?? null, computedAt: new Date(), }); @@ -819,23 +861,49 @@ export async function enqueueBootTimeRollupBackfill(): Promise<{ // every restart. Days outside the window are not foldable by // design and must not anchor discovery. const foldWindowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + // Two discovery arms, UNION-ed: + // + // 1. Per-day MISSING coverage — a (user, type, day) pair in + // `measurements` with no matching DAY rollup row (v1.4.39.1). + // + // 2. v1.20.0 F6 NULL-accumulator arm — a user whose DAY rollup rows + // predate migration 0190 (regression accumulators are NULL) needs a + // one-time re-fold so the windowed slope / r² / sd readers compose + // from the accumulators instead of falling back to the live REGR_* + // scan. Anchored on `sum_xy IS NULL` (the column the read tier + // treats as the coverage marker). New writes self-populate the + // accumulators, so this arm drains to empty once the boot re-fold + // catches up — mirrors the v1.4.39 sum_value rollout, which the + // v1.4.41 note confirms converged via this same boot queue. Bounded + // to the fold window so days outside it (un-foldable by design) + // never re-flag a user on every boot. const users = await prisma.$queryRaw>` - SELECT DISTINCT mt."user_id" AS id - FROM ( - SELECT DISTINCT - m."user_id", - m."type", - date_trunc('day', m."measured_at") AS bucket_start - FROM measurements m - WHERE m."deleted_at" IS NULL - AND m."measured_at" >= ${foldWindowStart} - ) mt - LEFT JOIN measurement_rollups r - ON r."user_id" = mt."user_id" - AND r."type" = mt."type" - AND r."granularity" = 'DAY' - AND r."bucket_start" = mt."bucket_start" - WHERE r."bucket_start" IS NULL + SELECT DISTINCT id FROM ( + SELECT DISTINCT mt."user_id" AS id + FROM ( + SELECT DISTINCT + m."user_id", + m."type", + date_trunc('day', m."measured_at") AS bucket_start + FROM measurements m + WHERE m."deleted_at" IS NULL + AND m."measured_at" >= ${foldWindowStart} + ) mt + LEFT JOIN measurement_rollups r + ON r."user_id" = mt."user_id" + AND r."type" = mt."type" + AND r."granularity" = 'DAY' + AND r."bucket_start" = mt."bucket_start" + WHERE r."bucket_start" IS NULL + + UNION + + SELECT DISTINCT r."user_id" AS id + FROM measurement_rollups r + WHERE r."granularity" = 'DAY' + AND r."bucket_start" >= ${foldWindowStart} + AND r."sum_xy" IS NULL + ) discovered `; if (users.length === 0) { diff --git a/tests/integration/rollup-regression-parity.integration.test.ts b/tests/integration/rollup-regression-parity.integration.test.ts new file mode 100644 index 000000000..aabc32f02 --- /dev/null +++ b/tests/integration/rollup-regression-parity.integration.test.ts @@ -0,0 +1,303 @@ +/** + * v1.20.0 F6 — bit-identical parity for the rollup regression accumulators. + * + * The windowed slope / r² / population-sd now compose from the per-bucket + * accumulators (`sum_x / sum_xy / sum_xx / sum_yy`, plus `n = count` and + * `Σy = mean·count`) that migration 0190 added to `measurement_rollups`. + * Because the accumulators are additive and Postgres `REGR_SLOPE` / + * `REGR_R2` / `STDDEV_POP` fold the SAME closed form over the SAME epoch-day + * x-axis the populator stored, the cross-bucket compose must equal the live + * aggregate over the raw rows EXACTLY — not within a tolerance. + * + * This suite seeds raw measurements spanning many DAY buckets with uneven + * per-day counts (the regime where mean-of-means and the true regression + * diverge), folds the rollups, then asserts: + * + * 1. `composeRegression` over the folded DAY-bucket accumulators equals + * `REGR_SLOPE` / `REGR_R2` / `STDDEV_POP` over the raw rows to ≤ 1e-9. + * 2. The source-collapse runs BEFORE accumulator summation — a dual-source + * day contributes only the canonical source's accumulators, matching a + * live regression over the canonical-source rows. + * 3. A DST-boundary reading lands in the same epoch-day x the live query + * uses (the x-axis is UTC `EXTRACT(EPOCH …)/86400`, DST-agnostic). + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { + composeRegression, + type RegressionAccumulators, +} from "@/lib/rollups/measurement-read"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +/** Live REGR row shape (one per type over the seeded window). */ +interface LiveRegrRow { + slope: number | null; + r2: number | null; + sd_pop: number | null; + n: bigint; +} + +let seq = 0; +async function seedUser(prisma: ReturnType) { + seq += 1; + return prisma.user.create({ + data: { + username: `regr-parity-${seq}`, + email: `regr-parity-${seq}@example.test`, + role: "USER", + }, + }); +} + +/** + * Live REGR_SLOPE / REGR_R2 / STDDEV_POP over the canonical-source raw rows + * for one (user, type), with the SAME epoch-day x-axis the populator stores. + * `sourceFilter` restricts to one source so the dual-source case can compare + * against the collapsed source explicitly. + */ +async function liveRegr( + prisma: ReturnType, + userId: string, + type: string, + source?: string, +): Promise { + const rows = await prisma.$queryRawUnsafe( + ` + SELECT + REGR_SLOPE(m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + ::double precision AS slope, + REGR_R2(m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0) + ::double precision AS r2, + STDDEV_POP(m."value")::double precision AS sd_pop, + COUNT(*) AS n + FROM measurements m + WHERE m."user_id" = $1 + AND m."type" = $2::"measurement_type" + AND m."deleted_at" IS NULL + ${source ? `AND m."source" = $3::"measurement_source"` : ""} + `, + ...(source ? [userId, type, source] : [userId, type]), + ); + return rows[0]; +} + +/** + * Read the folded DAY-bucket accumulators for one (user, type, source). + * The parity assertion sums these via `composeRegression` and compares to + * the live regression over the same source's raw rows. + */ +async function foldedAccumulators( + prisma: ReturnType, + userId: string, + type: string, + source?: string, +): Promise { + const rows = await prisma.measurementRollup.findMany({ + where: { + userId, + type: type as never, + granularity: "DAY", + ...(source ? { source: source as never } : {}), + }, + select: { + count: true, + mean: true, + sumX: true, + sumXy: true, + sumXx: true, + sumYy: true, + }, + }); + return rows.map((r) => ({ + count: r.count, + mean: r.mean, + sumX: r.sumX, + sumXy: r.sumXy, + sumXx: r.sumXx, + sumYy: r.sumYy, + })); +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("rollup regression accumulators — live parity (v1.20.0 F6)", () => { + it("WEIGHT: composed slope/r²/sd equals live REGR_*/STDDEV_POP, bit-identical", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + // Uneven per-day counts across many buckets — the divergent regime. + // Deterministic, non-collinear values so slope / r² / sd are all + // well-defined (a perfect line would make r² == 1 trivially). + const seed: Array<{ day: string; hour: number; value: number }> = [ + { day: "2026-03-02", hour: 8, value: 90.0 }, + { day: "2026-03-03", hour: 7, value: 89.4 }, + { day: "2026-03-03", hour: 20, value: 89.9 }, + { day: "2026-03-09", hour: 6, value: 88.7 }, + { day: "2026-03-09", hour: 9, value: 89.1 }, + { day: "2026-03-09", hour: 21, value: 88.2 }, + { day: "2026-03-21", hour: 8, value: 87.6 }, + { day: "2026-04-04", hour: 7, value: 86.9 }, + { day: "2026-04-04", hour: 19, value: 87.3 }, + { day: "2026-04-18", hour: 8, value: 86.1 }, + { day: "2026-05-02", hour: 8, value: 85.4 }, + { day: "2026-05-02", hour: 12, value: 85.9 }, + { day: "2026-05-16", hour: 8, value: 84.8 }, + ]; + await prisma.measurement.createMany({ + data: seed.map((s) => ({ + userId: user.id, + type: "WEIGHT" as const, + value: s.value, + unit: "kg", + source: "MANUAL" as const, + measuredAt: new Date( + `${s.day}T${String(s.hour).padStart(2, "0")}:00:00.000Z`, + ), + })), + }); + + await recomputeUserRollups(user.id, { + types: ["WEIGHT"], + granularities: ["DAY"], + }); + + const live = await liveRegr(prisma, user.id, "WEIGHT"); + const acc = await foldedAccumulators(prisma, user.id, "WEIGHT"); + const composed = composeRegression(acc); + + // Sanity: the seed actually exercised the regression (n matches, slope + // is a real number, not a degenerate null). + expect(Number(live.n)).toBe(seed.length); + expect(live.slope).not.toBeNull(); + expect(composed.slope).not.toBeNull(); + + // Bit-identical parity — the closed form is the same on both paths. + expect(composed.slope!).toBeCloseTo(live.slope!, 9); + expect(composed.r2!).toBeCloseTo(live.r2!, 9); + expect(composed.sdPop!).toBeCloseTo(live.sd_pop!, 9); + }); + + it("source-collapse precedes summation: composed RHR equals live REGR over the WHOOP source only", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + // Dual-source RHR across several days: WHOOP (the ladder-canonical + // source) + APPLE_HEALTH on overlapping days. The composed regression + // must match a LIVE regression over the WHOOP rows alone — never a + // cross-source blend. + const days = ["2026-05-04", "2026-05-05", "2026-05-11", "2026-05-18"]; + const data: Array<{ + userId: string; + type: "RESTING_HEART_RATE"; + value: number; + unit: string; + source: "WHOOP" | "APPLE_HEALTH"; + measuredAt: Date; + }> = []; + let i = 0; + for (const day of days) { + i += 1; + // WHOOP trends down; APPLE sits a few bpm higher to perturb a blend. + data.push({ + userId: user.id, + type: "RESTING_HEART_RATE", + value: 52 - i, + unit: "bpm", + source: "WHOOP", + measuredAt: new Date(`${day}T06:00:00.000Z`), + }); + data.push({ + userId: user.id, + type: "RESTING_HEART_RATE", + value: 58 + i, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date(`${day}T07:30:00.000Z`), + }); + } + await prisma.measurement.createMany({ data }); + + await recomputeUserRollups(user.id, { + types: ["RESTING_HEART_RATE"], + granularities: ["DAY"], + }); + + // WHOOP is the ladder winner for restingHeartRate, so the read tier + // collapses each day to the WHOOP per-source bucket. Compose only the + // WHOOP accumulators and compare to the live WHOOP-only regression. + const live = await liveRegr(prisma, user.id, "RESTING_HEART_RATE", "WHOOP"); + const acc = await foldedAccumulators( + prisma, + user.id, + "RESTING_HEART_RATE", + "WHOOP", + ); + const composed = composeRegression(acc); + + expect(Number(live.n)).toBe(days.length); + expect(composed.slope!).toBeCloseTo(live.slope!, 9); + expect(composed.r2!).toBeCloseTo(live.r2!, 9); + expect(composed.sdPop!).toBeCloseTo(live.sd_pop!, 9); + + // The blend would be wrong: a regression over BOTH sources differs from + // the WHOOP-only one, proving the collapse-before-sum matters. + const blended = await liveRegr(prisma, user.id, "RESTING_HEART_RATE"); + expect(Number(blended.n)).toBe(days.length * 2); + expect(blended.slope).not.toBeCloseTo(live.slope!, 6); + }); + + it("DST boundary: the epoch-day x-axis is UTC, so a spring-forward reading parity-matches", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + // Readings straddling the 2026 EU DST transition (2026-03-29 01:00 UTC). + // The x-axis is UTC epoch-days on both paths, so the transition is a + // no-op for parity; this pins that the accumulator x matches the live x. + const seed: Array<{ at: string; value: number }> = [ + { at: "2026-03-28T23:30:00.000Z", value: 100.0 }, + { at: "2026-03-29T00:30:00.000Z", value: 101.2 }, + { at: "2026-03-29T02:30:00.000Z", value: 99.7 }, + { at: "2026-03-30T08:00:00.000Z", value: 102.4 }, + { at: "2026-03-31T08:00:00.000Z", value: 98.9 }, + ]; + await prisma.measurement.createMany({ + data: seed.map((s) => ({ + userId: user.id, + type: "BLOOD_GLUCOSE" as const, + value: s.value, + unit: "mg/dL", + source: "MANUAL" as const, + measuredAt: new Date(s.at), + })), + }); + + await recomputeUserRollups(user.id, { + types: ["BLOOD_GLUCOSE"], + granularities: ["DAY"], + }); + + const live = await liveRegr(prisma, user.id, "BLOOD_GLUCOSE"); + const acc = await foldedAccumulators(prisma, user.id, "BLOOD_GLUCOSE"); + const composed = composeRegression(acc); + + expect(Number(live.n)).toBe(seed.length); + expect(composed.slope!).toBeCloseTo(live.slope!, 9); + expect(composed.r2!).toBeCloseTo(live.r2!, 9); + expect(composed.sdPop!).toBeCloseTo(live.sd_pop!, 9); + }); +});