Recompute stored pulse rollup means to the hourly-mean statistic

Rows written before this change keep the reading mean: the boot backfill and
the freshness check fill only missing days, so nothing rewrites them. Migration
0373 recomputes the mean of every stored PULSE row from the readings, DAY as
the mean of its hours' means and WEEK / MONTH / YEAR as the mean of their days,
and leaves count, min, max and the sums alone. The writer stores the coarse
rows the same way, so a refold and the migration agree.

prisma/migrations/0373_pulse_hourly_mean_day/migration.sql

@@ -0,0 +1,113 @@
+-- Pulse: a day is the mean of its hours' means, a coarser bucket the mean of
+-- its days.
+--
+-- A watch samples pulse about every 5 to 10 minutes at rest and about every
+-- 5 seconds in a workout, so the plain mean over a day's readings let one
+-- workout hour outweigh the other twenty-three. The rollup writer now stores a
+-- PULSE DAY bucket's `mean` as the mean of its per-hour means, and a WEEK /
+-- MONTH / YEAR bucket's as the mean of its days' values (see
+-- `src/lib/measurements/day-statistic.ts`). Nothing rewrites a stored bucket
+-- on its own: the boot backfill and the freshness check fill only missing
+-- days, so rows written before this release would keep the reading mean
+-- until a write happened to touch their day. This migration recomputes them.
+--
+-- Scope, exactly:
+--   - only rows of type PULSE, in every granularity;
+--   - only the `mean` column. `count`, `min_value`, `max_value`, `sum_value`,
+--     `sd`, `slope`, `r2` and the regression sums stay over every reading, as
+--     the writer keeps them;
+--   - hours, days, weeks, months and years are `date_trunc` on `measured_at`
+--     in the session zone, exactly as the writer groups them, per user / source
+--     over non-deleted readings.
+--
+-- Recomputing from `measurements` makes the migration idempotent: a rerun
+-- derives the same means whatever the column held before. A stored row whose
+-- readings are all gone is left as it is; the writer prunes such rows on the
+-- next fold of their range.
+--
+-- The hour means are built once into a temporary table, so the readings are
+-- scanned once for all four granularities.
+--
+-- Reversibility (down): recompute `mean` as `AVG(value)` over the same groups,
+-- or refold the PULSE rollups on a release before this one. No column changes
+-- shape.
+
+CREATE TEMPORARY TABLE pulse_hour_means AS
+SELECT
+  m."user_id"                         AS user_id,
+  m."source"                          AS source,
+  date_trunc('hour', m."measured_at") AS hr,
+  AVG(m."value")                      AS hour_mean
+FROM measurements m
+WHERE m."type" = 'PULSE'
+  AND m."deleted_at" IS NULL
+GROUP BY m."user_id", m."source", date_trunc('hour', m."measured_at");
+
+CREATE TEMPORARY TABLE pulse_day_means AS
+SELECT
+  h.user_id,
+  h.source,
+  date_trunc('day', h.hr) AS day,
+  AVG(h.hour_mean)        AS day_mean
+FROM pulse_hour_means h
+GROUP BY h.user_id, h.source, date_trunc('day', h.hr);
+
+CREATE INDEX pulse_day_means_key ON pulse_day_means (user_id, source, day);
+
+-- DAY buckets: the mean of the day's hour means.
+UPDATE "measurement_rollups" r
+SET "mean" = d.day_mean
+FROM pulse_day_means d
+WHERE r."granularity" = 'DAY'
+  AND r."type"         = 'PULSE'
+  AND r."user_id"      = d.user_id
+  AND r."source"       = d.source
+  AND r."bucket_start" = d.day;
+
+-- WEEK buckets: the mean of the week's day values.
+UPDATE "measurement_rollups" r
+SET "mean" = s.mean
+FROM (
+  SELECT d.user_id, d.source, date_trunc('week', d.day) AS bucket_start,
+         AVG(d.day_mean) AS mean
+  FROM pulse_day_means d
+  GROUP BY d.user_id, d.source, date_trunc('week', d.day)
+) s
+WHERE r."granularity" = 'WEEK'
+  AND r."type"         = 'PULSE'
+  AND r."user_id"      = s.user_id
+  AND r."source"       = s.source
+  AND r."bucket_start" = s.bucket_start;
+
+-- MONTH buckets.
+UPDATE "measurement_rollups" r
+SET "mean" = s.mean
+FROM (
+  SELECT d.user_id, d.source, date_trunc('month', d.day) AS bucket_start,
+         AVG(d.day_mean) AS mean
+  FROM pulse_day_means d
+  GROUP BY d.user_id, d.source, date_trunc('month', d.day)
+) s
+WHERE r."granularity" = 'MONTH'
+  AND r."type"         = 'PULSE'
+  AND r."user_id"      = s.user_id
+  AND r."source"       = s.source
+  AND r."bucket_start" = s.bucket_start;
+
+-- YEAR buckets.
+UPDATE "measurement_rollups" r
+SET "mean" = s.mean
+FROM (
+  SELECT d.user_id, d.source, date_trunc('year', d.day) AS bucket_start,
+         AVG(d.day_mean) AS mean
+  FROM pulse_day_means d
+  GROUP BY d.user_id, d.source, date_trunc('year', d.day)
+) s
+WHERE r."granularity" = 'YEAR'
+  AND r."type"         = 'PULSE'
+  AND r."user_id"      = s.user_id
+  AND r."source"       = s.source
+  AND r."bucket_start" = s.bucket_start;
+
+DROP TABLE pulse_day_means;
+DROP TABLE pulse_hour_means;

src/lib/rollups/measurement-rollups.ts

@@ -588,11 +588,16 @@ async function runRollupAggregate(
 
 /**
  * For a type whose DAY value is the mean of its per-hour means (see
- * `day-statistic.ts`), replace the DAY bucket's `mean`. Only `mean` changes:
+ * `day-statistic.ts`), replace the bucket's `mean`: a DAY bucket's is the mean
+ * of its hours' means, a WEEK / MONTH / YEAR bucket's the mean of its days'
+ * values, each day once. The coarse rows are not read for their mean (every
+ * coarse read folds the DAY tier, `readCanonicalRollupBuckets`), but a stored
+ * row should not contradict the tier it is folded from. Only `mean` changes:
  * `count`, `min`, `max` and the sums the regression composes from stay over
  * every reading, so the band and the windowed slope are untouched.
  *
- * Hours are UTC hours, matching the UTC day this tier buckets by.
+ * Hours and days are UTC, matching the UTC day this tier buckets by; migration
+ * 0373 applies the same statistic to rows stored before it.
  */
 async function applyHourlyMeanDay(
   db: Pick<typeof prisma, "$queryRawUnsafe">,
@@ -604,7 +609,10 @@ async function applyHourlyMeanDay(
   },
   rows: RollupRow[],
 ): Promise<RollupRow[]> {
-  if (input.granularity !== "DAY") return rows;
+  const truncUnit = DATE_TRUNC_UNIT[input.granularity];
+  if (!["day", "week", "month", "year"].includes(truncUnit)) {
+    throw new Error(`unexpected granularity: ${input.granularity}`);
+  }
   const types = [
     ...new Set(
       rows
@@ -623,22 +631,28 @@ async function applyHourlyMeanDay(
     Array<{ type: string; source: string; bucket_start: Date; mean: number }>
   >(
     `
-    SELECT h."type"::text AS type,
-           h."source"::text AS source,
-           date_trunc('day', h.hr) AS bucket_start,
-           AVG(h.hour_mean)::double precision AS mean
+    SELECT d."type"::text AS type,
+           d."source"::text AS source,
+           date_trunc('${truncUnit}', d.day) AS bucket_start,
+           AVG(d.day_mean)::double precision AS mean
     FROM (
-      SELECT m."type", m."source",
-             date_trunc('hour', m."measured_at") AS hr,
-             AVG(m."value") AS hour_mean
-      FROM measurements m
-      WHERE m."user_id" = $1
-        AND m."type" IN (${typeList})
-        AND m."measured_at" >= $2
-        AND m."measured_at" <  $3
-        AND m."deleted_at" IS NULL
-      GROUP BY m."type", m."source", date_trunc('hour', m."measured_at")
-    ) h
+      SELECT h."type", h."source",
+             date_trunc('day', h.hr) AS day,
+             AVG(h.hour_mean) AS day_mean
+      FROM (
+        SELECT m."type", m."source",
+               date_trunc('hour', m."measured_at") AS hr,
+               AVG(m."value") AS hour_mean
+        FROM measurements m
+        WHERE m."user_id" = $1
+          AND m."type" IN (${typeList})
+          AND m."measured_at" >= $2
+          AND m."measured_at" <  $3
+          AND m."deleted_at" IS NULL
+        GROUP BY m."type", m."source", date_trunc('hour', m."measured_at")
+      ) h
+      GROUP BY 1, 2, 3
+    ) d
     GROUP BY 1, 2, 3
   `,
     input.userId,

tests/integration/pulse-hourly-mean-migration.test.ts

@@ -0,0 +1,201 @@
+/**
+ * Migration 0373 rewrites the `mean` of stored PULSE rollup rows to the
+ * hourly-mean statistic, and touches nothing else.
+ *
+ * The rows are first built by the writer, then their `mean` is set to a
+ * sentinel the way a row written before this release still holds the reading
+ * mean. The migration must restore exactly what the writer stores now: DAY =
+ * mean of the day's hour means, WEEK / MONTH / YEAR = mean of their days. A
+ * second type (HRV) with the same readings keeps its sentinel, every other
+ * column keeps its value, and a rerun changes nothing.
+ */
+import { readFileSync } from "node:fs";
+import { resolve } from "node:path";
+import { beforeAll, describe, expect, it } from "vitest";
+
+import { getPrismaClient, truncateAllTables } from "./setup";
+import type { Prisma } from "@/generated/prisma/client";
+import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups";
+
+const prisma = getPrismaClient();
+const USER = "pulse-mean-migration";
+const SENTINEL = 999;
+
+const MIGRATION = readFileSync(
+  resolve(
+    process.cwd(),
+    "prisma",
+    "migrations",
+    "0373_pulse_hourly_mean_day",
+    "migration.sql",
+  ),
+  "utf8",
+);
+
+function steps(): string[] {
+  return MIGRATION.split("\n")
+    .filter((line) => !line.trimStart().startsWith("--"))
+    .join("\n")
+    .split(";")
+    .map((s) => s.trim())
+    .filter((s) => s.length > 0);
+}
+
+async function runMigration(): Promise<void> {
+  // One connection: the temporary tables live in the session that made them.
+  await prisma.$transaction(async (tx) => {
+    for (const step of steps()) await tx.$executeRawUnsafe(step);
+  });
+}
+
+/** Tuesday 2026-03-03: one hard hour of 12 readings at 150, three hours of one reading at 60. */
+const TUE = Date.UTC(2026, 2, 3);
+/** Wednesday: two hours at 70. */
+const WED = Date.UTC(2026, 2, 4);
+const TUE_VALUE = (150 + 3 * 60) / 4;
+const WED_VALUE = 70;
+
+function rows(
+  type: "PULSE" | "HEART_RATE_VARIABILITY",
+): Prisma.MeasurementCreateManyInput[] {
+  const out: Prisma.MeasurementCreateManyInput[] = [];
+  const add = (at: number, value: number) =>
+    out.push({
+      id: `${type}-${at}`,
+      userId: USER,
+      type,
+      unit: type === "PULSE" ? "bpm" : "ms",
+      source: "APPLE_HEALTH",
+      value,
+      measuredAt: new Date(at),
+    });
+  for (let i = 0; i < 12; i += 1) add(TUE + 10 * 3_600_000 + i * 300_000, 150);
+  for (const h of [12, 14, 16]) add(TUE + h * 3_600_000, 60);
+  for (const h of [8, 9]) add(WED + h * 3_600_000, 70);
+  return out;
+}
+
+type Row = {
+  type: string;
+  granularity: string;
+  bucket_start: Date;
+  count: number;
+  mean: number;
+  min_value: number;
+  max_value: number;
+  sum_value: number | null;
+  sd: number | null;
+  sum_x: number | null;
+  sum_xy: number | null;
+  sum_xx: number | null;
+  sum_yy: number | null;
+};
+
+const readRows = () =>
+  prisma.$queryRaw<Row[]>`
+    SELECT "type"::text AS type, "granularity"::text AS granularity,
+           "bucket_start", "count", "mean", "min_value", "max_value",
+           "sum_value", "sd", "sum_x", "sum_xy", "sum_xx", "sum_yy"
+    FROM measurement_rollups
+    WHERE "user_id" = ${USER}
+    ORDER BY 1, 2, 3
+  `;
+
+let before: Row[] = [];
+
+beforeAll(async () => {
+  await truncateAllTables(prisma);
+  await prisma.user.create({
+    data: { id: USER, username: USER, timezone: "UTC" },
+  });
+  await prisma.measurement.createMany({
+    data: [...rows("PULSE"), ...rows("HEART_RATE_VARIABILITY")],
+  });
+  await recomputeUserRollups(USER, {
+    types: ["PULSE", "HEART_RATE_VARIABILITY"],
+    from: new Date(Date.UTC(2026, 0, 1)),
+    to: new Date(Date.UTC(2027, 0, 1)),
+  });
+  await prisma.$executeRaw`
+    UPDATE measurement_rollups SET "mean" = ${SENTINEL} WHERE "user_id" = ${USER}
+  `;
+  before = await readRows();
+  await runMigration();
+}, 120_000);
+
+const meanOf = (all: Row[], type: string, granularity: string, at?: number) =>
+  all.find(
+    (r) =>
+      r.type === type &&
+      r.granularity === granularity &&
+      (at === undefined || r.bucket_start.getTime() === at),
+  )?.mean;
+
+describe("migration 0373", () => {
+  it("covers all four granularities of both types", () => {
+    for (const type of ["PULSE", "HEART_RATE_VARIABILITY"]) {
+      for (const g of ["DAY", "WEEK", "MONTH", "YEAR"]) {
+        expect(meanOf(before, type, g)).toBe(SENTINEL);
+      }
+    }
+  });
+
+  it("gives a PULSE day the mean of its hours' means", async () => {
+    const after = await readRows();
+    expect(meanOf(after, "PULSE", "DAY", TUE)).toBeCloseTo(TUE_VALUE, 9);
+    expect(meanOf(after, "PULSE", "DAY", WED)).toBeCloseTo(WED_VALUE, 9);
+  });
+
+  it("gives a PULSE week, month and year the mean of their days", async () => {
+    const after = await readRows();
+    for (const g of ["WEEK", "MONTH", "YEAR"]) {
+      expect(meanOf(after, "PULSE", g)).toBeCloseTo(
+        (TUE_VALUE + WED_VALUE) / 2,
+        9,
+      );
+    }
+  });
+
+  it("matches what the writer stores now", async () => {
+    const migrated = await readRows();
+    await recomputeUserRollups(USER, {
+      types: ["PULSE"],
+      from: new Date(Date.UTC(2026, 0, 1)),
+      to: new Date(Date.UTC(2027, 0, 1)),
+    });
+    const written = await readRows();
+    for (const r of written.filter((x) => x.type === "PULSE")) {
+      const m = migrated.find(
+        (x) =>
+          x.type === r.type &&
+          x.granularity === r.granularity &&
+          x.bucket_start.getTime() === r.bucket_start.getTime(),
+      );
+      expect(m?.mean).toBeCloseTo(r.mean, 9);
+    }
+    // Restore the migrated state for the remaining assertions.
+    await prisma.$executeRaw`
+      UPDATE measurement_rollups SET "mean" = ${SENTINEL}
+      WHERE "user_id" = ${USER} AND "type" = 'HEART_RATE_VARIABILITY'
+    `;
+  });
+
+  it("leaves every other type's mean alone", async () => {
+    const after = await readRows();
+    for (const r of after.filter((x) => x.type !== "PULSE")) {
+      expect(r.mean).toBe(SENTINEL);
+    }
+  });
+
+  it("leaves every column but mean alone", async () => {
+    const after = await readRows();
+    const strip = ({ mean: _mean, ...rest }: Row) => rest;
+    expect(after.map(strip)).toEqual(before.map(strip));
+  });
+
+  it("is idempotent", async () => {
+    const first = await readRows();
+    await runMigration();
+    expect(await readRows()).toEqual(first);
+  });
+});