feat(rollups): store per-bucket regression accumulators

Windowed slope, r2 and population sd still ran a live REGR_*/STDDEV_POP
query 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 do not compose linearly across DAY buckets.

Add the four regression-accumulator sums Postgres folds for REGR_* /
STDDEV_POP - sum_x, sum_xy, sum_xx, sum_yy - per (user, type,
granularity, bucketStart, source) over the same canonical-source rows
the live query reads, with the x-axis in epoch days. With n = count and
sum_y = mean*count already stored, all six terms are additive across
buckets, so a 7/30/90-day slope/r2/sd composes from the summed
accumulators with a bit-identical result to the live closed form.

Migration 0190 adds the four nullable columns; nullable keeps it
additive so existing rows read NULL until the boot re-fold refills them
(mirrors the v1.4.39 sum_value rollout). The populator emits the four
SUM terms inside the existing aggregate (no extra round-trip) and
persistRollupRows writes them. composeRegression / composeWindowedRegression
evaluate the closed form from already-collapsed buckets and return a full
miss when any in-window bucket lacks accumulators, so the reader falls
back to live SQL rather than composing a partial window.

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;

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)

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<RegressionAccumulators>,
+): 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<T extends SourcedBucketRow>(
   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<AccumulatorBucketRow>,
+  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

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(),
   });