From be37497118dde6d01ccabd4220e10c305483a2cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 22 Jun 2026 07:13:06 +0200 Subject: [PATCH] perf(rollups): compose windowed slope/r2/sd from accumulators The summaries slice (computeFromRollups) and the comprehensive aggregator (buildFromRollups) each ran a narrow live $queryRaw over the raw measurements partition on EVERY warm-rollup request to fold slope7/30/90 and r2_7/30/90 via REGR_SLOPE / REGR_R2 - the un-eliminated live scan on the happy path. Drop the six REGR_* columns from both narrow queries and compose the windowed slope / r2 (and the population sd via composeRegression) from the per-bucket accumulators stored in measurement_rollups. The 90-day DAY buckets are loaded once (already loaded in comprehensive; one bounded findMany added in the slim slice), source-collapsed to the canonical source, then summed over the day-aligned 7/30/90-day sub-windows. A window whose buckets lack accumulators (pre-backfill) composes to null, the same shape the empty REGR_* FILTER returned, so a transitional read never returns a wrong partial regression. median (PERCENTILE_CONT) and the exact |z|>2 anomaly count stay on the minimal live query - both are non-composable across buckets. The cold coverage-miss path keeps the live REGR_* regression. Annotate slope_source:"rollup" + median_source/anomaly_source:"live" so a regression to the heavy path stays observable (no-silent-cap rule). --- .../__tests__/summaries-slice.test.ts | 69 ++++++-- src/lib/analytics/summaries-slice.ts | 152 +++++++++++++----- .../comprehensive-aggregator.test.ts | 105 ++++++++---- src/lib/insights/comprehensive-aggregator.ts | 143 ++++++++-------- 4 files changed, 315 insertions(+), 154 deletions(-) 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, })), ); }