diff --git a/prisma/migrations/0373_pulse_hourly_mean_day/migration.sql b/prisma/migrations/0373_pulse_hourly_mean_day/migration.sql new file mode 100644 index 000000000..4b7110762 --- /dev/null +++ b/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; diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index f5982bd4b..df8e52bb2 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/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, @@ -604,7 +609,10 @@ async function applyHourlyMeanDay( }, rows: RollupRow[], ): Promise { - 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, diff --git a/tests/integration/pulse-hourly-mean-migration.test.ts b/tests/integration/pulse-hourly-mean-migration.test.ts new file mode 100644 index 000000000..040aa53fa --- /dev/null +++ b/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 { + // 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` + 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); + }); +});