diff --git a/prisma/migrations/0417_measurement_rollup_reading_extremes/migration.sql b/prisma/migrations/0417_measurement_rollup_reading_extremes/migration.sql new file mode 100644 index 000000000..a42916f48 --- /dev/null +++ b/prisma/migrations/0417_measurement_rollup_reading_extremes/migration.sql @@ -0,0 +1,24 @@ +-- The lowest and highest reading value per rollup bucket. +-- +-- The full-history extremes the insight features carry (`allTimeMin`, +-- `allTimeMax`, `allTimeAvg` for weight, blood pressure and pulse) were read +-- from the raw `measurements` rows on every call: two passes over every +-- reading of those types the account ever recorded, the largest read on the +-- database. The DAY rollups already hold everything that fold needs except +-- the extremes of the readings themselves: `min_value` / `max_value` widen to +-- a row's own `value_min` / `value_max`, the intra-hour range an hourly +-- heart-rate statistic carries, so they are not the same figure. +-- +-- Two nullable columns. Existing rows read NULL until the boot re-fold +-- (`rollup-full-backfill`, discovered by the NULL arm beside `sum_xy`) +-- refills them; new writes fill them from the same aggregate as every other +-- column. A reader treats a type with a NULL in its window as uncovered and +-- reads the raw rows, as before. +-- +-- Reversibility: +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "reading_min_value"; +-- ALTER TABLE "measurement_rollups" DROP COLUMN IF EXISTS "reading_max_value"; + +ALTER TABLE "measurement_rollups" ADD COLUMN IF NOT EXISTS "reading_min_value" DOUBLE PRECISION; + +ALTER TABLE "measurement_rollups" ADD COLUMN IF NOT EXISTS "reading_max_value" DOUBLE PRECISION; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index c38d79ad9..f3337df8c 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -2253,31 +2253,31 @@ enum RollupGranularity { } model MeasurementRollup { - userId String @map("user_id") - type MeasurementType - granularity RollupGranularity - bucketStart DateTime @map("bucket_start") @db.Timestamptz(3) + userId String @map("user_id") + type MeasurementType + granularity RollupGranularity + bucketStart DateTime @map("bucket_start") @db.Timestamptz(3) /// v1.11.1 — source-aware grain. Rollup rows are minted per source so the /// read path can collapse overlapping sources (e.g. WHOOP + Apple Watch /// resting heart rate) to the ladder-canonical reading, while cumulative /// metrics (steps, energy) sum per source. Always non-null; migration 0115 /// purged the legacy source-blind aggregates and the boot backfill re-mints /// per-source rows. - source MeasurementSource - count Int - mean Float - minValue Float @map("min_value") - maxValue Float @map("max_value") - sd Float? - slope Float? - r2 Float? + source MeasurementSource + count Int + mean Float + minValue Float @map("min_value") + maxValue Float @map("max_value") + sd Float? + slope Float? + r2 Float? /// v1.4.39 W-SUM — cumulative-metric sum for the bucket. Populated /// alongside `mean / count` for every type; cumulative read paths /// (steps, flights, distance, daylight, active-energy) consume this /// directly instead of re-deriving from raw measurements. Nullable /// to keep the migration additive — the boot backfill fills NULLs /// for existing rows; new rows always carry the sum. - sumValue Float? @map("sum_value") + 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 @@ -2291,11 +2291,21 @@ model MeasurementRollup { /// parity-matches the cross-bucket compose. 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) + sumX Float? @map("sum_x") + sumXy Float? @map("sum_xy") + sumXx Float? @map("sum_xx") + sumYy Float? @map("sum_yy") + /// The lowest and highest reading `value` of the bucket. `minValue` / + /// `maxValue` widen to a row's own range (`Measurement.valueMin` / + /// `valueMax`, the intra-hour range an hourly heart-rate statistic + /// carries); these do not, so a reader whose figure is over the readings + /// themselves (the full-history extremes in `feature-blocks.ts`) can fold + /// the DAY tier instead of the raw rows. Nullable to keep migration 0417 + /// additive: the boot re-fold fills existing rows, new rows always carry + /// them, and a reader treats a type with a NULL in its range as uncovered. + readingMinValue Float? @map("reading_min_value") + readingMaxValue Float? @map("reading_max_value") + computedAt DateTime @default(now()) @map("computed_at") @db.Timestamptz(3) user User @relation(fields: [userId], references: [id], onDelete: Cascade) diff --git a/src/lib/insights/feature-blocks.ts b/src/lib/insights/feature-blocks.ts index c9b75c195..70024c404 100644 --- a/src/lib/insights/feature-blocks.ts +++ b/src/lib/insights/feature-blocks.ts @@ -17,11 +17,17 @@ import { foldMeanSql, isHourlyMeanTypeSql, } from "@/lib/measurements/day-mean"; -import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { + loadUserSourcePriority, + utcBucketEnd, +} from "@/lib/rollups/measurement-read"; import { classifyAgainstEffectiveRange } from "@/lib/labs/reference-range"; import { calendarDaysUntil } from "@/lib/measurement-reminders/due-day"; import { resolveLabFields } from "@/lib/labs/serialise"; -import { readRollupBuckets } from "@/lib/rollups/measurement-rollups"; +import { + readRollupBuckets, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; import { deriveBucketedTypes } from "@/lib/signals/adapters/correlation"; import type { AggregatedFeatures, @@ -40,11 +46,10 @@ interface AllTimeExtremes { } /** - * v1.18.11 — full-history min / max / mean per measurement type via ONE - * grouped SQL aggregation, with NO row materialisation in JS. Fills the - * `allTime*` feature fields; the windowed `summarize()` covers trends + recent - * windows, and this covers the long-horizon figures the prompt labels - * "allTime". + * v1.18.11 — full-history min / max / mean per measurement type, folded in + * SQL with NO row materialisation in JS. Fills the `allTime*` feature fields; + * the windowed `summarize()` covers trends + recent windows, and this covers + * the long-horizon figures the prompt labels "allTime". * * One source per (type, day): each source is folded per day first, each day * keeps its ladder-canonical source, then the days fold into the figures — @@ -52,6 +57,25 @@ interface AllTimeExtremes { * row let a device that samples densely (a watch's pulse) outweigh the * readings the source ladder prefers on the same days. * + * Where the days come from. A day is the same per-source fold either way — + * count, sum, the extremes of the readings and, for an hourly-mean type, the + * day value — and the DAY rollup tier already holds exactly that per (type, + * source, day), recomputed on every write. Reading the raw rows instead meant + * two passes over every reading of these types the account ever recorded on + * every call, the largest read on the database. So the figures fold: + * + * - the DAY tier, for every whole UTC day from `tierFrom` on, for each type + * whose tier rows in that range all carry the reading extremes + * (migration 0417; the boot re-fold fills older rows); + * - the raw rows for the rest: the days before `tierFrom` (older than the + * fold window, or the partial day the `floor` falls in), and every day of + * a type the tier does not cover yet. + * + * The split is at a UTC midnight, so no day is folded from both, and the + * canonical pick per day is the same on both sides. The tier's pulse day + * value follows the stored day-mean method; the raw fold follows this + * process's mirror of it, which trails a change by at most a minute. + * * Only the four types that expose `allTime*` fields are aggregated (weight, * systolic, diastolic, pulse). Returns a map keyed by `MeasurementType`; a type * with no rows is simply absent. @@ -67,15 +91,88 @@ export async function readAllTimeExtremes( ): Promise> { const out = new Map(); if (types.length === 0) return out; + // The first whole UTC day the tier may answer: after the fold window's + // edge (the fold writes nothing older, and the edge day may be partial) + // and after the floor's own day (only part of it is inside the floor). + const windowEdge = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + const tierFrom = utcBucketEnd( + floor !== null && floor > windowEdge ? floor : windowEdge, + "DAY", + ); + const typeList = (list: readonly MeasurementType[]) => + Prisma.join(list.map((t) => Prisma.sql`${t}::measurement_type`)); + + const [priorityJson, tierRows] = await Promise.all([ + loadUserSourcePriority(userId), + prisma.$queryRaw>` + SELECT r."type"::text AS type + FROM measurement_rollups r + WHERE r."user_id" = ${userId} + AND r."granularity" = 'DAY' + AND r."type" IN (${typeList(types)}) + AND r."bucket_start" >= ${tierFrom} + GROUP BY r."type" + HAVING bool_and( + r."reading_min_value" IS NOT NULL AND r."reading_max_value" IS NOT NULL + ) + `, + ]); + const tierTypes = new Set(tierRows.map((r) => r.type)); + const fromTier = types.filter((t) => tierTypes.has(t)); + const rawOnly = types.filter((t) => !tierTypes.has(t)); + const since = floor === null ? Prisma.empty : Prisma.sql`AND m."measured_at" >= ${floor}`; - const priorityJson = await loadUserSourcePriority(userId); - const rank = Prisma.raw( - buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'), - ); - const typeList = Prisma.join( - types.map((t) => Prisma.sql`${t}::measurement_type`), - ); + const rawArms: Prisma.Sql[] = []; + if (rawOnly.length > 0) { + rawArms.push(Prisma.sql` + SELECT m."type", m."source", m."measured_at", m."value" + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."deleted_at" IS NULL + AND m."type" IN (${typeList(rawOnly)}) + ${since}`); + } + if (fromTier.length > 0) { + rawArms.push(Prisma.sql` + SELECT m."type", m."source", m."measured_at", m."value" + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."deleted_at" IS NULL + AND m."type" IN (${typeList(fromTier)}) + ${since} + AND m."measured_at" < ${tierFrom}`); + } + const rank = (alias: string) => + Prisma.raw( + buildSourceRankCase(priorityJson, `${alias}."type"`, `${alias}."source"`), + ); + // The tier's canonical days, in the shape of the raw fold's: the day's + // reading-weighted sum and count, its reading extremes, and its day value + // with a weight of one (a DAY row's `mean` is the day value for an + // hourly-mean type). + const tierCanon = + fromTier.length === 0 + ? Prisma.empty + : Prisma.sql` + UNION ALL + SELECT * FROM ( + SELECT DISTINCT ON (r."type", r."bucket_start") + r."type", + r."count" AS cnt, + COALESCE(r."sum_value", r."count" * r."mean")::double precision + AS total, + r."reading_min_value" AS min_value, + r."reading_max_value" AS max_value, + r."mean" AS wsum, + 1.0::double precision AS wdays + FROM measurement_rollups r + WHERE r."user_id" = ${userId} + AND r."granularity" = 'DAY' + AND r."type" IN (${typeList(fromTier)}) + AND r."bucket_start" >= ${tierFrom} + ORDER BY r."type", r."bucket_start", (${rank("r")}), r."source"::text + ) tier`; // A day of pulse is the mean of its hours' means and the full history the // mean of its days (`day-mean.ts`). `weighted` carries, per source and day, // the day-weighted sum and the day's weight (one) for those types only; @@ -92,14 +189,10 @@ export async function readAllTimeExtremes( const rows = await prisma.$queryRaw< Array<{ type: string; mean: number; min: number; max: number }> >` - WITH hourly_src AS ( - SELECT m.* - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."deleted_at" IS NULL - AND m."type" IN (${typeList}) - ${since} - AND ${Prisma.raw(isHourlyMeanTypeSql('m."type"'))} + WITH live AS (${Prisma.join(rawArms, " UNION ALL ")}), + hourly_src AS ( + SELECT * FROM live l + WHERE ${Prisma.raw(isHourlyMeanTypeSql('l."type"'))} ), weighted AS ( SELECT @@ -113,34 +206,32 @@ export async function readAllTimeExtremes( ), per_source AS ( SELECT - m."type", - m."source", - date_trunc('day', m."measured_at") AS day, + l."type", + l."source", + date_trunc('day', l."measured_at") AS day, COUNT(*)::int AS cnt, - SUM(m."value")::double precision AS total, - MIN(m."value")::double precision AS min_value, - MAX(m."value")::double precision AS max_value - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."deleted_at" IS NULL - AND m."type" IN (${typeList}) - ${since} - GROUP BY m."type", m."source", 3 - ), - canon AS ( - SELECT DISTINCT ON (p."type", p.day) - p."type", p.cnt, p.total, p.min_value, p.max_value, w.wsum, w.wdays - FROM per_source p - LEFT JOIN weighted w - ON w."type" = p."type" AND w."source" = p."source" AND w.day = p.day - ORDER BY p."type", p.day, (${rank}), p."source"::text + SUM(l."value")::double precision AS total, + MIN(l."value")::double precision AS min_value, + MAX(l."value")::double precision AS max_value + FROM live l + GROUP BY l."type", l."source", 3 ) SELECT c."type"::text AS type, ${mean}::double precision AS mean, MIN(c.min_value)::double precision AS min, MAX(c.max_value)::double precision AS max - FROM canon c + FROM ( + SELECT * FROM ( + SELECT DISTINCT ON (p."type", p.day) + p."type", p.cnt, p.total, p.min_value, p.max_value, w.wsum, w.wdays + FROM per_source p + LEFT JOIN weighted w + ON w."type" = p."type" AND w."source" = p."source" AND w.day = p.day + ORDER BY p."type", p.day, (${rank("p")}), p."source"::text + ) raw + ${tierCanon} + ) c GROUP BY c."type" `; for (const r of rows) { diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index 3426575fa..69161ada6 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -188,6 +188,8 @@ describe("recomputeBucketsForMeasurement", () => { sd: 2.0, slope: -0.1, r2: 0.9, + reading_min_value: 80.5, + reading_max_value: 84.5, }, ]); getGlobalBossMock.mockReturnValue({ send: bossSend }); @@ -210,14 +212,14 @@ describe("recomputeBucketsForMeasurement", () => { // instead of raising (or being dropped). Param layout per row: // [userId, type, granularity, bucketStart, source, count, mean, // min, max, sum, sd, slope, r2, sumX, sumXy, sumXx, sumYy, - // computedAt]. + // readingMin, readingMax, computedAt]. expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1); const [sql, ...params] = txExecuteRawUnsafe.mock.calls[0]; expect(sql).toContain( "ON CONFLICT (user_id, type, granularity, bucket_start, source)", ); expect(sql).toContain("DO UPDATE SET"); - expect(params).toHaveLength(18); + expect(params).toHaveLength(20); expect(params[0]).toBe("user-1"); expect(params[1]).toBe("WEIGHT"); expect(params[2]).toBe("DAY"); @@ -227,6 +229,10 @@ describe("recomputeBucketsForMeasurement", () => { // v1.4.39 — sumValue flows through. Cumulative read paths // (steps, flights, distance, daylight, active-energy) consume it. expect(params[9]).toBe(247.5); + // The reading extremes, beside the range-widened min / max. + expect(params[7]).toBe(80.0); + expect(params[17]).toBe(80.5); + expect(params[18]).toBe(84.5); // WEEK / MONTH / YEAR — three enqueues against the worker queue. expect(bossSend).toHaveBeenCalledTimes(3); @@ -453,13 +459,13 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { expect(deleteMany.mock.calls[0][0].where.userId).toBe("user-1"); expect(deleteMany.mock.calls[0][0].where.granularity).toBe("DAY"); - // 3 chunks of 18 params per row, every one an ON CONFLICT DO UPDATE + // 3 chunks of 20 params per row, every one an ON CONFLICT DO UPDATE // upsert — tolerant of a mid-transaction hook insert AND of a // concurrent fold on the same partition (v1.28.33, issue #486). expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(3); - expect(txExecuteRawUnsafe.mock.calls[0]).toHaveLength(1 + 500 * 18); - expect(txExecuteRawUnsafe.mock.calls[1]).toHaveLength(1 + 500 * 18); - expect(txExecuteRawUnsafe.mock.calls[2]).toHaveLength(1 + 200 * 18); + expect(txExecuteRawUnsafe.mock.calls[0]).toHaveLength(1 + 500 * 20); + expect(txExecuteRawUnsafe.mock.calls[1]).toHaveLength(1 + 500 * 20); + expect(txExecuteRawUnsafe.mock.calls[2]).toHaveLength(1 + 200 * 20); for (const call of txExecuteRawUnsafe.mock.calls) { expect(call[0]).toContain("ON CONFLICT"); expect(call[0]).toContain("DO UPDATE SET"); @@ -575,7 +581,7 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { expect(transaction).toHaveBeenCalledTimes(1); expect(typeof transaction.mock.calls[0][0]).toBe("function"); expect(txExecuteRawUnsafe).toHaveBeenCalledTimes(1); - expect(txExecuteRawUnsafe.mock.calls[0]).toHaveLength(1 + 2 * 18); + expect(txExecuteRawUnsafe.mock.calls[0]).toHaveLength(1 + 2 * 20); expect(result.rowsUpserted).toBe(2); }); }); diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index cf33a2c6d..c3cd5f397 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -184,6 +184,10 @@ interface RollupRow { sum_xy: number | null; sum_xx: number | null; sum_yy: number | null; + // The extremes of the readings' own `value`, without the per-row range + // `min_value` / `max_value` widen to. See `MeasurementRollup.readingMinValue`. + reading_min_value: number | null; + reading_max_value: number | null; } /** @@ -546,7 +550,9 @@ async function runRollupAggregate( SUM(${REBASED_X})::double precision AS sum_x, SUM(${REBASED_X} * m."value")::double precision AS sum_xy, SUM(${REBASED_X} * ${REBASED_X})::double precision AS sum_xx, - SUM(m."value" * m."value")::double precision AS sum_yy + SUM(m."value" * m."value")::double precision AS sum_yy, + MIN(m."value")::double precision AS reading_min_value, + MAX(m."value")::double precision AS reading_max_value FROM measurements m WHERE m."user_id" = $1 AND m."type" IN (${typeList}) @@ -594,7 +600,9 @@ async function runRollupAggregate( SUM(${REBASED_X})::double precision AS sum_x, SUM(${REBASED_X} * m."value")::double precision AS sum_xy, SUM(${REBASED_X} * ${REBASED_X})::double precision AS sum_xx, - SUM(m."value" * m."value")::double precision AS sum_yy + SUM(m."value" * m."value")::double precision AS sum_yy, + MIN(m."value")::double precision AS reading_min_value, + MAX(m."value")::double precision AS reading_max_value FROM measurements m WHERE m."user_id" = $1 AND m."measured_at" >= $2 @@ -921,20 +929,24 @@ function buildRollupUpsert( row.sum_xy ?? null, row.sum_xx ?? null, row.sum_yy ?? null, + row.reading_min_value ?? null, + row.reading_max_value ?? null, computedAt, ); tuples.push( `($${base + 1}, $${base + 2}::"measurement_type", $${base + 3}::"measurement_rollup_granularity", ` + `$${base + 4}::timestamptz, $${base + 5}::"measurement_source", $${base + 6}::int, ` + `$${base + 7}, $${base + 8}, $${base + 9}, $${base + 10}, $${base + 11}, $${base + 12}, ` + - `$${base + 13}, $${base + 14}, $${base + 15}, $${base + 16}, $${base + 17}, $${base + 18}::timestamptz)`, + `$${base + 13}, $${base + 14}, $${base + 15}, $${base + 16}, $${base + 17}, $${base + 18}, ` + + `$${base + 19}, $${base + 20}::timestamptz)`, ); } const sql = ` INSERT INTO measurement_rollups (user_id, type, granularity, bucket_start, source, count, mean, min_value, max_value, sum_value, sd, slope, r2, - sum_x, sum_xy, sum_xx, sum_yy, computed_at) + sum_x, sum_xy, sum_xx, sum_yy, reading_min_value, reading_max_value, + computed_at) VALUES ${tuples.join(", ")} ON CONFLICT (user_id, type, granularity, bucket_start, source) DO UPDATE SET @@ -950,6 +962,8 @@ function buildRollupUpsert( sum_xy = EXCLUDED.sum_xy, sum_xx = EXCLUDED.sum_xx, sum_yy = EXCLUDED.sum_yy, + reading_min_value = EXCLUDED.reading_min_value, + reading_max_value = EXCLUDED.reading_max_value, computed_at = EXCLUDED.computed_at `; return { sql, params }; @@ -1201,7 +1215,9 @@ export async function enqueueBootTimeRollupBackfill( // 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. + // never re-flag a user on every boot. The same arm refills the + // reading extremes of migration 0417 (`reading_min_value IS NULL`); + // the extremes reader treats a NULL as uncovered until it lands. const users = await prisma.$queryRaw>` SELECT DISTINCT id FROM ( SELECT DISTINCT mt."user_id" AS id @@ -1227,7 +1243,7 @@ export async function enqueueBootTimeRollupBackfill( FROM measurement_rollups r WHERE r."granularity" = 'DAY' AND r."bucket_start" >= ${foldWindowStart} - AND r."sum_xy" IS NULL + AND (r."sum_xy" IS NULL OR r."reading_min_value" IS NULL) ) discovered `; diff --git a/tests/integration/all-time-extremes-tier-parity.test.ts b/tests/integration/all-time-extremes-tier-parity.test.ts new file mode 100644 index 000000000..74c16102a --- /dev/null +++ b/tests/integration/all-time-extremes-tier-parity.test.ts @@ -0,0 +1,321 @@ +/** + * The full-history extremes read the DAY rollup tier and give the figures the + * raw-row fold gave. + * + * `readAllTimeExtremes` used to fold every raw reading of the four types on + * each call. It now folds the DAY tier for the whole days inside the fold + * window and the raw rows for the rest. The reference below is that earlier + * query, verbatim, so the comparison is against what the function used to + * answer, not against a second derivation of the same idea. + * + * The seed has what the two paths could disagree on: two sources on the same + * days (weight by scale and by hand, pulse from a watch's hourly statistics + * with their own range and a strap's dense samples), readings older than the + * fold window, readings either side of a UTC midnight, soft-deleted rows, and + * a reading deleted after the tier was built (through the write path's own + * rollup recompute). + * + * Means are compared to nine significant digits: the tier sums the same day + * values in a different order. Extremes are compared exactly. + */ +import { beforeAll, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { Prisma, type MeasurementType } from "@/generated/prisma/client"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; +import { + dayWeightedRowsSql, + foldMeanSql, + isHourlyMeanTypeSql, +} from "@/lib/measurements/day-mean"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { afterMeasurementMutation } from "@/lib/rollups/after-measurement-mutation"; + +const prisma = getPrismaClient(); +const USER = "all-time-tier-parity"; +const TYPES: MeasurementType[] = [ + "WEIGHT", + "BLOOD_PRESSURE_SYS", + "BLOOD_PRESSURE_DIA", + "PULSE", +]; +const HOUR = 3_600_000; +const DAY = 24 * HOUR; +const today = Date.UTC( + new Date().getUTCFullYear(), + new Date().getUTCMonth(), + new Date().getUTCDate(), +); + +/** The query `readAllTimeExtremes` ran before it read the DAY tier. */ +async function referenceExtremes( + userId: string, + types: readonly MeasurementType[], + floor: Date | null, +) { + const out = new Map(); + const since = + floor === null ? Prisma.empty : Prisma.sql`AND m."measured_at" >= ${floor}`; + const priorityJson = await loadUserSourcePriority(userId); + const rank = Prisma.raw( + buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'), + ); + const typeList = Prisma.join( + types.map((t) => Prisma.sql`${t}::measurement_type`), + ); + const mean = Prisma.raw( + foldMeanSql({ + typeColumn: 'c."type"', + total: "c.total", + count: "c.cnt", + weightedSum: "c.wsum", + weightSum: "c.wdays", + }), + ); + const rows = await prisma.$queryRaw< + Array<{ type: string; mean: number; min: number; max: number }> + >` + WITH hourly_src AS ( + SELECT m.* + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."deleted_at" IS NULL + AND m."type" IN (${typeList}) + ${since} + AND ${Prisma.raw(isHourlyMeanTypeSql('m."type"'))} + ), + weighted AS ( + SELECT + w."type", + w."source", + date_trunc('day', w."measured_at") AS day, + SUM(w."value" * w.day_weight)::double precision AS wsum, + SUM(w.day_weight)::double precision AS wdays + FROM ${dayWeightedRowsSql("hourly_src", null, { bySource: true })} w + GROUP BY w."type", w."source", 3 + ), + per_source AS ( + SELECT + m."type", + m."source", + date_trunc('day', m."measured_at") AS day, + COUNT(*)::int AS cnt, + SUM(m."value")::double precision AS total, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."deleted_at" IS NULL + AND m."type" IN (${typeList}) + ${since} + GROUP BY m."type", m."source", 3 + ), + canon AS ( + SELECT DISTINCT ON (p."type", p.day) + p."type", p.cnt, p.total, p.min_value, p.max_value, w.wsum, w.wdays + FROM per_source p + LEFT JOIN weighted w + ON w."type" = p."type" AND w."source" = p."source" AND w.day = p.day + ORDER BY p."type", p.day, (${rank}), p."source"::text + ) + SELECT + c."type"::text AS type, + ${mean}::double precision AS mean, + MIN(c.min_value)::double precision AS min, + MAX(c.max_value)::double precision AS max + FROM canon c + GROUP BY c."type" + `; + for (const r of rows) { + out.set(r.type, { mean: r.mean, min: r.min, max: r.max }); + } + return out; +} + +/** A small deterministic generator, so a failure reproduces. */ +function lcg(seed: number): () => number { + let s = seed; + return () => { + s = (s * 1_103_515_245 + 12_345) % 2_147_483_648; + return s / 2_147_483_648; + }; +} + +function seed(): Prisma.MeasurementCreateManyInput[] { + const rnd = lcg(7); + const rows: Prisma.MeasurementCreateManyInput[] = []; + let n = 0; + const add = ( + type: MeasurementType, + unit: string, + source: Prisma.MeasurementCreateManyInput["source"], + at: number, + value: number, + extra: Partial = {}, + ) => { + n += 1; + rows.push({ + id: `${USER}-${n}`, + userId: USER, + type, + unit, + source, + value, + measuredAt: new Date(at), + ...extra, + }); + }; + // Weight and blood pressure across six years: older than the fold window + // too. Every fifth day a hand entry beside the scale's, every 23rd day a + // reading a minute either side of UTC midnight, and a few tombstones. + for (let d = 6 * 365; d >= 0; d -= 1) { + const day = today - d * DAY; + add("WEIGHT", "kg", "WITHINGS", day + 6 * HOUR, 70 + 10 * rnd()); + if (d % 5 === 0) + add("WEIGHT", "kg", "MANUAL", day + 20 * HOUR, 69 + 12 * rnd()); + for (const [type, base] of [ + ["BLOOD_PRESSURE_SYS", 110], + ["BLOOD_PRESSURE_DIA", 65], + ] as const) { + add(type, "mmHg", "WITHINGS", day + 7 * HOUR, base + 40 * rnd()); + if (d % 23 === 0) { + add(type, "mmHg", "WITHINGS", day - 60_000, base + 45 * rnd()); + add(type, "mmHg", "WITHINGS", day + 60_000, base - 5 + 50 * rnd()); + } + if (d % 31 === 0) { + add(type, "mmHg", "WITHINGS", day + 9 * HOUR, base + 80, { + deletedAt: new Date(), + }); + } + } + } + // Pulse over the last 400 days: a watch's hourly statistics carrying the + // hour's own min / max, and a strap's dense samples on every other day. + for (let d = 400; d >= 0; d -= 1) { + const day = today - d * DAY; + for (let h = 0; h < 24; h += 3) { + const avg = 55 + 30 * rnd(); + add("PULSE", "bpm", "APPLE_HEALTH", day + h * HOUR, avg, { + valueMin: avg - 15, + valueMax: avg + 40, + externalId: `stats:HKQuantityTypeIdentifierHeartRate:${day + h * HOUR}`, + }); + } + if (d % 2 === 0) { + for (let i = 0; i < 30; i += 1) { + add( + "PULSE", + "bpm", + "WHOOP", + day + 10 * HOUR + i * 120_000, + 50 + 70 * rnd(), + ); + } + } + } + return rows; +} + +async function expectParity(floor: Date | null) { + const [actual, reference] = await Promise.all([ + readAllTimeExtremes(USER, TYPES, floor), + referenceExtremes(USER, TYPES, floor), + ]); + expect([...actual.keys()].sort()).toEqual([...reference.keys()].sort()); + expect(reference.size).toBe(4); + for (const [type, ref] of reference) { + const got = actual.get(type as MeasurementType)!; + expect(got.min, type).toBe(ref.min); + expect(got.max, type).toBe(ref.max); + expect(got.mean! / ref.mean, type).toBeCloseTo(1, 9); + } +} + +beforeAll(async () => { + await truncateAllTables(prisma); + await prisma.user.create({ + data: { id: USER, username: USER, timezone: "Pacific/Auckland" }, + }); + const rows = seed(); + for (let i = 0; i < rows.length; i += 5000) { + await prisma.measurement.createMany({ data: rows.slice(i, i + 5000) }); + } + await recomputeUserRollups(USER, { granularities: ["DAY"] }); +}, 300_000); + +describe("readAllTimeExtremes over the DAY tier", () => { + it("reads the tier for every type", async () => { + const covered = await prisma.measurementRollup.groupBy({ + by: ["type"], + where: { + userId: USER, + granularity: "DAY", + readingMinValue: { not: null }, + }, + }); + expect(covered.map((c) => c.type).sort()).toEqual([...TYPES].sort()); + }); + + it("gives the raw fold's figures over the whole history", async () => { + await expectParity(null); + }); + + it("gives the raw fold's figures from a floor inside a day", async () => { + await expectParity(new Date(today - 200 * DAY + 13 * HOUR + 17 * 60_000)); + }); + + it("follows a deletion through the write path's recompute", async () => { + const hottest = await prisma.measurement.findFirst({ + where: { userId: USER, type: "PULSE", source: "WHOOP", deletedAt: null }, + orderBy: { value: "desc" }, + }); + await prisma.measurement.update({ + where: { id: hottest!.id }, + data: { deletedAt: new Date() }, + }); + await afterMeasurementMutation( + USER, + [{ type: "PULSE", measuredAt: hottest!.measuredAt }], + { source: "manual" }, + ); + await expectParity(null); + }); + + it("is the tier that answers, not the raw rows", async () => { + // Write it, then break it: a tier row raised past every reading must + // move the answer, or the parity above proved nothing about the tier. + const day = new Date(today - 3 * DAY); + await prisma.measurementRollup.updateMany({ + where: { + userId: USER, + type: "WEIGHT", + granularity: "DAY", + bucketStart: day, + }, + data: { readingMaxValue: 999 }, + }); + const out = await readAllTimeExtremes(USER, ["WEIGHT"]); + expect(out.get("WEIGHT")?.max).toBe(999); + }); + + it("reads the raw rows for a type with a tier row lacking the extremes", async () => { + const day = new Date(today - 3 * DAY); + await prisma.measurementRollup.updateMany({ + where: { + userId: USER, + type: "WEIGHT", + granularity: "DAY", + bucketStart: day, + }, + data: { readingMinValue: null }, + }); + const [actual, reference] = await Promise.all([ + readAllTimeExtremes(USER, ["WEIGHT"]), + referenceExtremes(USER, ["WEIGHT"], null), + ]); + expect(actual.get("WEIGHT")?.max).toBe(reference.get("WEIGHT")?.max); + expect(actual.get("WEIGHT")?.min).toBe(reference.get("WEIGHT")?.min); + }); +});