From 4e79098cc09a26381bbdbae3e4458e9b7e30932c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 25 Oct 2026 06:27:44 +0100 Subject: [PATCH] perf(measurements): fold per-source days without joining a regrouping back Three readers grouped the same rows twice and joined the second grouping back to the first: the live buckets joined each hourly-mean day's hour means to the per-source days, the all-time extremes joined the day weights the same way, and the doctor report's dense buckets joined each bucket's latest reading and hour means to the bucket aggregates. On a measurements table without planner statistics (a new install, or an import under the autoanalyze threshold) the planner estimated one row on each side and ran every one of those joins as a nested loop that re-ran the inner grouping for each outer row. On ten years of readings over six types (51,000 rows, no statistics) the dense buckets and the every-type live buckets ran into the 60 s statement timeout and the all-time extremes took 11 s. With statistics all three answered in 0.1 to 0.2 s. Each is now one chain of folds over one scan. The live buckets fold the readings per local hour and the hours per local day, carrying the day of hours beside the counts. The all-time extremes take the day weights from the same windowed rows they count, the shape the pre-fold remainder already has. The dense buckets carry each bucket's latest reading and each hour's count and mean on the rows as window values, and one grouping folds them all. Same fixture without statistics: 0.2 s, 0.1 s and 0.1 s, the same as with them. A parity suite runs the joined statements as the reference against the readers across ladders, zones with whole, half-hour and negative offsets, grains, overlapping and off-ladder sources, glucose contexts and device types, tombstones and both DST changes; sums and means agree to floating-point summation order, everything else exactly. The unanalysed-table guard now runs the three readers under EXPLAIN ANALYZE: the joins measured 200 to 3,000 row visits per row stored, the folds about 10. The dense-bucket loader drops under the 200-line function cap, so its baseline entry goes. --- .../baselines/source-function-lines.json | 5 +- src/lib/doctor-report/dense-buckets.ts | 112 +-- src/lib/insights/feature-blocks.ts | 53 +- src/lib/measurements/daily-series-read.ts | 67 +- .../canonical-source-pick-unanalysed.test.ts | 124 ++- .../per-source-fold-parity.test.ts | 752 ++++++++++++++++++ 6 files changed, 965 insertions(+), 148 deletions(-) create mode 100644 tests/integration/per-source-fold-parity.test.ts diff --git a/src/__tests__/baselines/source-function-lines.json b/src/__tests__/baselines/source-function-lines.json index e29dc0c14..3f0155aef 100644 --- a/src/__tests__/baselines/source-function-lines.json +++ b/src/__tests__/baselines/source-function-lines.json @@ -1,7 +1,7 @@ { "$comment": "Production functions over 200 lines, keyed \"file :: name\", and their length. Maintained by src/__tests__/source-size-ratchet.test.ts.", - "count": 349, - "total": 113048, + "count": 348, + "total": 112838, "entries": { "app/achievements/page.tsx :: AchievementsPage": 206, "app/api/admin/backups/upload/route.ts :: apiHandler(arg)": 314, @@ -275,7 +275,6 @@ "lib/doctor-report-pdf/measurements-charts-section.ts :: buildMeasurementsChartsSection": 384, "lib/doctor-report-pdf/medication-mood-wellness-section.ts :: buildMedicationMoodWellnessSection": 509, "lib/doctor-report/collect.ts :: collectDoctorReportData": 667, - "lib/doctor-report/dense-buckets.ts :: loadDenseMeasurementBuckets": 210, "lib/documents/fenced-chat.ts :: streamFencedReply": 309, "lib/documents/ingest.ts :: ingestDocument": 272, "lib/export/backup-json-scan.ts :: scanBackupJson": 225, diff --git a/src/lib/doctor-report/dense-buckets.ts b/src/lib/doctor-report/dense-buckets.ts index 3810fcda7..c01baed17 100644 --- a/src/lib/doctor-report/dense-buckets.ts +++ b/src/lib/doctor-report/dense-buckets.ts @@ -81,6 +81,13 @@ export async function loadDenseMeasurementBuckets(params: { const hourlyMeanType = Prisma.raw(isHourlyMeanTypeSql('"type"')); // Each hour's weight under the active pulse day-mean method (code-built). const hourWeight = Prisma.raw(hourWeightSql("n")); + // One grouping over one scan, no join. The latest reading and the hour + // means used to be groupings of their own, joined back to the buckets; on + // a measurements table without planner statistics (a new install, an + // import under the autoanalyze threshold) those joins ran as nested loops + // that re-ran a grouping for every bucket, and a long report window ran + // into the statement timeout. `tests/integration/per-source-fold-parity.test.ts` + // holds the joined statement as the reference. return prisma.$queryRaw` WITH localized AS ( SELECT @@ -106,6 +113,9 @@ export async function loadDenseMeasurementBuckets(params: { AND m."deleted_at" IS NULL AND m."type" IN (${typeList}) ), + -- Every row carries its bucket's latest reading and its local hour's + -- reading count and mean, with one row per hour marked, so the one + -- grouping below folds them all. scored AS ( SELECT localized.*, @@ -113,72 +123,18 @@ export async function loadDenseMeasurementBuckets(params: { WHEN "value" > 0 THEN 1.509 * (power(ln("value"), 1.084) - 5.381) ELSE NULL - END AS risk_transform + END AS risk_transform, + FIRST_VALUE("value") OVER ( + PARTITION BY local_day, "type", "source", "device_type", "glucose_context" + ORDER BY "measured_at" DESC, "id" DESC + ) AS latest_value, + COUNT(*) OVER hour AS n, + AVG("value") OVER hour AS hour_mean, + ROW_NUMBER() OVER hour = 1 AS hour_first FROM localized - ), - latest_rows AS ( - SELECT DISTINCT ON ( - local_day, - "type", - "source", - "device_type", - "glucose_context" + WINDOW hour AS ( + PARTITION BY local_day, local_hour, "type", "source", "device_type", "glucose_context" ) - local_day, - "type", - "source", - "device_type", - "glucose_context", - "measured_at" AS latest_at, - "value" AS latest_value - FROM localized - ORDER BY - local_day, - "type", - "source", - "device_type", - "glucose_context", - "measured_at" DESC, - "id" DESC - ), - -- The hourly-mean day (pulse): each local hour's mean and reading - -- count, then the weighted sum of the means and the sum of the hour - -- weights per bucket. Empty for every other type. - hour_means AS ( - SELECT - local_day, - "type", - "source", - "device_type", - "glucose_context", - AVG("value") AS hour_mean, - COUNT(*) AS n - FROM localized - WHERE ${hourlyMeanType} - GROUP BY - local_day, - local_hour, - "type", - "source", - "device_type", - "glucose_context" - ), - hour_agg AS ( - SELECT - local_day, - "type", - "source", - "device_type", - "glucose_context", - SUM(hour_mean * ${hourWeight})::double precision AS hour_weighted_sum, - SUM(${hourWeight})::double precision AS hour_weight - FROM hour_means - GROUP BY - local_day, - "type", - "source", - "device_type", - "glucose_context" ), grouped AS ( SELECT @@ -188,6 +144,14 @@ export async function loadDenseMeasurementBuckets(params: { "device_type", "glucose_context", COUNT(*)::int AS sample_count, + MAX("measured_at") AS latest_at, + MAX(latest_value)::double precision AS latest_value, + SUM(hour_mean * ${hourWeight}) + FILTER (WHERE hour_first AND ${hourlyMeanType})::double precision + AS hour_weighted_sum, + SUM(${hourWeight}) + FILTER (WHERE hour_first AND ${hourlyMeanType})::double precision + AS hour_weight, SUM("value")::double precision AS sum_value, MIN("value")::double precision AS min_value, MAX("value")::double precision AS max_value, @@ -242,8 +206,8 @@ export async function loadDenseMeasurementBuckets(params: { grouped.sum_value AS "sumValue", grouped.min_value AS "minValue", grouped.max_value AS "maxValue", - latest_rows.latest_at AS "latestAt", - latest_rows.latest_value::double precision AS "latestValue", + grouped.latest_at AS "latestAt", + grouped.latest_value AS "latestValue", grouped.clinical_count AS "clinicalCount", grouped.clinical_sum AS "clinicalSum", grouped.clinical_sum_squares AS "clinicalSumSquares", @@ -256,21 +220,9 @@ export async function loadDenseMeasurementBuckets(params: { grouped.clinical_tar2_count AS "clinicalTar2Count", grouped.clinical_low_risk_sum AS "clinicalLowRiskSum", grouped.clinical_high_risk_sum AS "clinicalHighRiskSum", - hour_agg.hour_weighted_sum AS "hourWeightedSum", - hour_agg.hour_weight AS "hourWeight" + grouped.hour_weighted_sum AS "hourWeightedSum", + grouped.hour_weight AS "hourWeight" FROM grouped - INNER JOIN latest_rows - ON latest_rows.local_day = grouped.local_day - AND latest_rows."type" = grouped."type" - AND latest_rows."source" = grouped."source" - AND latest_rows."device_type" IS NOT DISTINCT FROM grouped."device_type" - AND latest_rows."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" - LEFT JOIN hour_agg - ON hour_agg.local_day = grouped.local_day - AND hour_agg."type" = grouped."type" - AND hour_agg."source" = grouped."source" - AND hour_agg."device_type" IS NOT DISTINCT FROM grouped."device_type" - AND hour_agg."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" ORDER BY "bucketStart" ASC, grouped."type" ASC `; diff --git a/src/lib/insights/feature-blocks.ts b/src/lib/insights/feature-blocks.ts index acd01bf57..63229ce86 100644 --- a/src/lib/insights/feature-blocks.ts +++ b/src/lib/insights/feature-blocks.ts @@ -12,11 +12,7 @@ import { prisma } from "@/lib/db"; import { Prisma } from "@/generated/prisma/client"; import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; -import { - dayWeightedRowsSql, - foldMeanSql, - isHourlyMeanTypeSql, -} from "@/lib/measurements/day-mean"; +import { dayWeightedRowsSql, foldMeanSql } from "@/lib/measurements/day-mean"; import { loadUserSourcePriority, utcBucketEnd, @@ -207,9 +203,18 @@ export async function readAllTimeExtremes( 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; - // every other type keeps the reading-weighted mean below, unchanged. + // mean of its days (`day-mean.ts`). `per_source` carries, per source and + // day, the day-weighted sum and the day's weight (one) beside the reading + // count and sum; the mean below reads the weighted pair for those types + // only, and every other type keeps the reading-weighted mean, unchanged. + // + // One fold of one scan, no join. The weighted pair used to be a second + // grouping of the same rows, joined back to the per-source days before the + // pick; on a measurements table without planner statistics (a new install, + // an import under the autoanalyze threshold) that join ran as a nested + // loop that re-ran the weighting for every source day, and ten years of + // readings took seconds. `tests/integration/per-source-fold-parity.test.ts` + // holds the joined statement as the reference. const mean = Prisma.raw( foldMeanSql({ typeColumn: 'c."type"', @@ -223,30 +228,18 @@ export async function readAllTimeExtremes( Array<{ type: string; mean: number; min: number; max: number }> >` WITH live AS (${Prisma.join(rawArms, " UNION ALL ")}), - hourly_src AS ( - SELECT * FROM live l - WHERE ${Prisma.raw(isHourlyMeanTypeSql('l."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 l."type", l."source", - date_trunc('day', l."measured_at") AS day, - COUNT(*)::int AS cnt, - 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 + date_trunc('day', l."measured_at") AS day, + COUNT(*)::int AS cnt, + SUM(l."value")::double precision AS total, + MIN(l."value")::double precision AS min_value, + MAX(l."value")::double precision AS max_value, + SUM(l."value" * l.day_weight)::double precision AS wsum, + SUM(l.day_weight)::double precision AS wdays + FROM ${dayWeightedRowsSql("live", null, { bySource: true })} l GROUP BY l."type", l."source", 3 ) SELECT @@ -257,10 +250,8 @@ export async function readAllTimeExtremes( 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 + p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.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} diff --git a/src/lib/measurements/daily-series-read.ts b/src/lib/measurements/daily-series-read.ts index b25cf5a7a..702af8782 100644 --- a/src/lib/measurements/daily-series-read.ts +++ b/src/lib/measurements/daily-series-read.ts @@ -217,13 +217,22 @@ export async function readLiveBuckets(opts: { // the user's zone and truncate there. // // The canonical source is chosen per local DAY, whatever the grain: each - // source is folded per day in one hashed pass (a few rows per day, however - // dense the stream), each day keeps its ladder-canonical source, and only - // then are the days folded into the week or month. Picking the source per + // source is folded per hour and then per day in hashed passes (a few rows + // per day, however dense the stream), each day keeps its ladder-canonical + // source, and only then are the days folded into the week or month. Picking the source per // week or month instead dropped every day the losing source carried, so a // month of manual readings plus one synced reading reported one day. // Folding first and picking second is the same result as collapsing the // raw rows first; it just never sorts more than one row per source per day. + // + // One chain of folds over one scan, no join. The hour means used to be a + // second scan joined back to the per-source days before the pick; on a + // measurements table without planner statistics (a new install, an import + // under the autoanalyze threshold) that join ran as a nested loop that + // re-ran the hour fold for every source day, and a ten-year read of every + // type ran into the statement timeout. + // `tests/integration/per-source-fold-parity.test.ts` holds the joined + // statement as the reference. const buckets = await prisma.$queryRaw< Array<{ type: string; @@ -234,13 +243,20 @@ export async function readLiveBuckets(opts: { max_value: number | null; }> >` - WITH per_source AS ( + -- Each source's readings fold per local hour, then the hours per local + -- day. A day of a type with an activity-dependent sampling rate is the + -- weighted mean of its local hours' means, not the mean of its readings: + -- a workout hour holds hundreds of times as many readings as a resting + -- one. NULL for every other type. + WITH per_hour AS ( SELECT m."type", m."source", date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, + date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, SUM(m."value")::double precision AS total, - COUNT(*)::int AS cnt, + COUNT(*) AS n, + AVG(m."value") AS hour_mean, MIN(m."value") AS min_value, MAX(m."value") AS max_value FROM measurements m @@ -249,40 +265,25 @@ export async function readLiveBuckets(opts: { AND m."measured_at" <= ${to} AND m."deleted_at" IS NULL ${typeFilter} - GROUP BY m."type", m."source", 3 + GROUP BY m."type", m."source", 3, 4 ), - -- A day of a type with an activity-dependent sampling rate is the weighted - -- mean of its local hours' means, not the mean of its readings: a workout - -- hour holds hundreds of times as many readings as a resting one. - -- Empty for every other type. - hour_mean AS ( - SELECT h."type", h."source", h.day, - ${dayOfHours}::double precision AS day_mean - FROM ( - SELECT - m."type", - m."source", - date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, - date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, - AVG(m."value") AS hour_mean, - COUNT(*) AS n - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${from} - AND m."measured_at" <= ${to} - AND m."deleted_at" IS NULL - AND m."type" IN (${hourlyMeanList}) - ${typeFilter} - GROUP BY m."type", m."source", 3, 4 - ) h + per_source AS ( + SELECT + h."type", + h."source", + h.day, + SUM(h.total)::double precision AS total, + SUM(h.n)::int AS cnt, + MIN(h.min_value) AS min_value, + MAX(h.max_value) AS max_value, + (CASE WHEN h."type" IN (${hourlyMeanList}) THEN ${dayOfHours} END)::double precision AS day_mean + FROM per_hour h GROUP BY h."type", h."source", h.day ), canon AS ( SELECT DISTINCT ON (p."type", p.day) - p."type", p.day, p.total, p.cnt, p.min_value, p.max_value, hm.day_mean + p."type", p.day, p.total, p.cnt, p.min_value, p.max_value, p.day_mean FROM per_source p - LEFT JOIN hour_mean hm - ON hm."type" = p."type" AND hm."source" = p."source" AND hm.day = p.day ORDER BY p."type", p.day, (${rankRaw}), p."source"::text ), folded AS ( diff --git a/tests/integration/canonical-source-pick-unanalysed.test.ts b/tests/integration/canonical-source-pick-unanalysed.test.ts index 40389cea3..f968ae8e5 100644 --- a/tests/integration/canonical-source-pick-unanalysed.test.ts +++ b/tests/integration/canonical-source-pick-unanalysed.test.ts @@ -15,8 +15,14 @@ * stay within a small multiple of the rows seeded. A plan that revisits the * table per row lands thousands of times over that budget, whatever the * machine. + * + * The same holds for the readers that fold per source and day without the + * helper (the blood-pressure series, the live buckets, the all-time + * extremes, the doctor report's dense buckets): each used to join a second + * grouping of the same rows back to its days, which the planner ran the same + * way on such a table. */ -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import type { MeasurementSource, @@ -32,6 +38,10 @@ import { canonicalMeasurementsCte, type CanonicalWindow, } from "@/lib/analytics/source-rank-sql"; +import { prisma as appPrisma } from "@/lib/db"; +import { loadDenseMeasurementBuckets } from "@/lib/doctor-report/dense-buckets"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { readLiveBuckets } from "@/lib/measurements/daily-series-read"; import { bloodPressureSeriesSql } from "@/lib/measurements/series/read-blood-pressure"; import { VALUE_RANGES } from "@/lib/validations/measurement"; @@ -49,6 +59,14 @@ const DAY = 86_400_000; */ const WORK_PER_ROW = 8; +/** + * The same for a reader that folds every source's rows per hour or per day + * before it picks: a few more passes over the rows (a sort and a grouping + * per level), so about 10 per row seeded. The joins these readers used to + * make measured 200 to 3,000. + */ +const FOLD_WORK_PER_ROW = 16; + /** * Long enough for any linear plan on this fixture many times over, short * enough that a quadratic one fails here instead of at the test timeout. @@ -166,6 +184,55 @@ const SHAPES: Array< let seeded = 0; +/** + * Run `read` with every `$queryRaw` it issues also run under `EXPLAIN + * ANALYZE` on a transaction with the statement timeout, and hold each plan to + * the budget. The reader still gets its real rows. + */ +async function expectBoundedReader( + read: () => Promise, + workPerRow: number, +) { + const original = appPrisma.$queryRaw.bind(appPrisma) as ( + ...args: unknown[] + ) => Promise; + const statements: Array<[TemplateStringsArray, unknown[]]> = []; + vi.spyOn(appPrisma, "$queryRaw").mockImplementation((( + strings: TemplateStringsArray, + ...values: unknown[] + ) => { + statements.push([strings, values]); + return original(strings, ...values); + }) as never); + try { + await read(); + } finally { + vi.restoreAllMocks(); + } + expect(statements.length).toBeGreaterThan(0); + for (const [strings, values] of statements) { + const head = "EXPLAIN (ANALYZE, FORMAT JSON) "; + const explain = Object.assign([head + strings[0], ...strings.slice(1)], { + raw: [head + strings.raw[0], ...strings.raw.slice(1)], + }); + const [{ "QUERY PLAN": plan }] = await prisma.$transaction(async (tx) => { + await tx.$executeRawUnsafe( + `SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}'`, + ); + return tx.$queryRaw }>>( + explain, + ...values, + ); + }); + const visits = rowVisits(plan[0].Plan); + expect( + visits, + `${visits} row visits for ${seeded} rows seeded:\n` + + JSON.stringify(plan[0].Plan, null, 1), + ).toBeLessThanOrEqual(workPerRow * seeded); + } +} + /** * Run `sql` under `EXPLAIN ANALYZE` and hold its row visits to the budget. * The statement timeout fails a quadratic plan here rather than at the test @@ -275,3 +342,58 @@ describe("blood-pressure series pick on a table without statistics", () => { expect(rows.every((r) => r.source === "WITHINGS")).toBe(true); }); }); + +/** + * The readers that fold each source's days themselves and then pick: the + * live buckets of every type over ten years, the all-time extremes, and the + * doctor report's dense buckets over ten years. Each used to join a second + * grouping of the same rows (hour means, day weights, latest readings) back + * to the per-source days; on this fixture those joins measured about 200 + * (dense buckets), 1,200 (extremes) and 3,000 (live buckets) row visits per + * row seeded. + */ +describe("per-source folds on a table without statistics", () => { + const from = new Date(NOW.getTime() - 3650 * DAY); + it.each<[string, () => Promise>]>([ + [ + "live buckets, every type, ten years", + () => + readLiveBuckets({ + userId: USER, + type: null, + from, + to: NOW, + cap: 100_000, + priorityJson: null, + grain: "daily", + timeZone: "Europe/Berlin", + }), + ], + [ + "all-time extremes", + () => + readAllTimeExtremes(USER, [ + "BLOOD_PRESSURE_SYS", + "BLOOD_PRESSURE_DIA", + "PULSE", + ]), + ], + [ + "dense buckets, ten years", + () => + loadDenseMeasurementBuckets({ + userId: USER, + start: from, + end: NOW, + reportTz: "Europe/Berlin", + denseTypes: ["PULSE"], + }), + ], + ])("reads the window a bounded number of times: %s", async (_label, read) => { + let out: unknown[] | Map = []; + await expectBoundedReader(async () => { + out = await read(); + }, FOLD_WORK_PER_ROW); + expect(out instanceof Map ? out.size : out.length).toBeGreaterThan(0); + }); +}); diff --git a/tests/integration/per-source-fold-parity.test.ts b/tests/integration/per-source-fold-parity.test.ts new file mode 100644 index 000000000..2a4bce63d --- /dev/null +++ b/tests/integration/per-source-fold-parity.test.ts @@ -0,0 +1,752 @@ +/** + * Three per-source folds, against a real Postgres: each reader as it ships + * (one fold over one scan) against the statement it replaced, kept verbatim + * below as the reference, in which a second grouping of the same rows was + * joined back to the per-source days or buckets. + * + * - `readLiveBuckets`: the local hours' means for an hourly-mean type, + * joined to the per-source days before the ladder pick; + * - `readAllTimeExtremes`: the day-weighted sums for an hourly-mean type, + * joined to the per-source days before the ladder pick; + * - `loadDenseMeasurementBuckets`: each bucket's latest reading and its + * hour means, joined to the bucket aggregates. + * + * Every row must match: keys, counts, timestamps and extremes exactly, sums + * and means to within floating-point summation order (each statement adds + * the same readings, but not always in the same order or grouping, and + * neither statement fixed that order). The fixture carries what could tell + * them apart: overlapping sources per day, a source off every ladder, a + * cumulative type, an hourly-mean type with several readings in one hour, + * glucose contexts and device types including null ones, tombstones, + * readings a minute either side of UTC midnight, both DST changes in the + * user's zone, and empty days. Shapes run under the default ladders and a + * user ladder, in zones with whole, half-hour and negative offsets. + * + * The fixture holds no rollups, so `readAllTimeExtremes` reads every type + * from the raw rows; its rollup arm is a separate `UNION ALL` the change did + * not touch. + */ +import { afterEach, beforeAll, describe, expect, it, vi } from "vitest"; + +import { + Prisma, + type GlucoseContext, + type MeasurementSource, + type MeasurementType, +} from "@/generated/prisma/client"; +import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql"; +import { prisma as appPrisma } from "@/lib/db"; +import { loadDenseMeasurementBuckets } from "@/lib/doctor-report/dense-buckets"; +import { readAllTimeExtremes } from "@/lib/insights/feature-blocks"; +import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; +import { readLiveBuckets } from "@/lib/measurements/daily-series-read"; +import { + dayWeightedRowsSql, + foldMeanSql, + hourWeightSql, + isHourlyMeanTypeSql, + weightedHourMeanSql, +} from "@/lib/measurements/day-mean"; +import { HOURLY_MEAN_DAY_TYPES } from "@/lib/measurements/day-statistic"; + +import { makeUser } from "../kit/factories"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +const prisma = getPrismaClient(); +const USER_ID = "user-per-source-fold-parity"; +const DAY_MS = 86_400_000; +const NOW = Date.now(); +const TODAY_UTC = Math.floor(NOW / DAY_MS) * DAY_MS; + +type Row = Record; + +/** The rows each `$queryRaw` the reader issued returned, in call order. */ +async function capture(read: () => Promise): Promise { + const results: Row[][] = []; + const original = appPrisma.$queryRaw.bind(appPrisma) as ( + ...args: unknown[] + ) => Promise; + vi.spyOn(appPrisma, "$queryRaw").mockImplementation(((...args: unknown[]) => + original(...args).then((rows) => { + results.push(rows); + return rows; + })) as never); + await read(); + return results; +} + +afterEach(() => { + vi.restoreAllMocks(); +}); + +/** Compare two row sets: numbers to a relative 1e-9, everything else exact. */ +function expectSameRows(now: Row[], before: Row[], keys: string[]) { + const key = (r: Row) => + keys + .map((k) => + r[k] instanceof Date ? (r[k] as Date).toISOString() : String(r[k]), + ) + .join("|"); + const sort = (rows: Row[]) => + [...rows].sort((a, b) => (key(a) < key(b) ? -1 : key(a) > key(b) ? 1 : 0)); + const a = sort(now); + const b = sort(before); + expect(a.map(key)).toEqual(b.map(key)); + for (let i = 0; i < a.length; i++) { + expect(Object.keys(a[i]).sort()).toEqual(Object.keys(b[i]).sort()); + for (const [col, value] of Object.entries(a[i])) { + const other = b[i][col]; + if (typeof value === "number" && typeof other === "number") { + expect( + Math.abs(value - other), + `${col} of ${key(a[i])}: ${value} vs ${other}`, + ).toBeLessThanOrEqual(1e-9 * Math.max(1, Math.abs(other))); + } else { + expect(value, `${col} of ${key(a[i])}`).toEqual(other); + } + } + } +} + +/** Small seeded generator, so the fixture is the same on every run. */ +function rng(seed: number) { + let a = seed >>> 0; + return () => { + a = (a + 0x6d2b79f5) >>> 0; + let t = a; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +const SOURCES: MeasurementSource[] = [ + "APPLE_HEALTH", + "WITHINGS", + "MANUAL", + "OURA", + // Off every ladder: the name tiebreak decides. + "NIGHTSCOUT", +]; +const TYPES: MeasurementType[] = [ + "PULSE", + "BLOOD_GLUCOSE", + "HEART_RATE_VARIABILITY", + "WEIGHT", + "ACTIVITY_STEPS", + "BLOOD_PRESSURE_SYS", +]; +const CONTEXTS: Array = [null, "FASTING", "RANDOM"]; +const DEVICES: Array = [null, "watch", "phone"]; + +async function seed(): Promise { + await makeUser({ id: USER_ID, timezone: "Europe/Berlin" }); + const random = rng(11); + const rows: Prisma.MeasurementCreateManyInput[] = []; + // 400 days: over both DST changes in the user's zone. + for (let day = 0; day < 400; day++) { + if (day % 17 === 8) continue; // an empty day + const dayStart = TODAY_UTC - day * DAY_MS; + for (const type of TYPES) { + const sources = [...SOURCES] + .sort(() => random() - 0.5) + .slice(0, 1 + Math.floor(random() * 3)); + for (const source of sources) { + const readings = 1 + Math.floor(random() * 6); + // Several readings share an hour, so an hour mean differs from a + // reading mean. + const hour = Math.floor(random() * 24); + for (let r = 0; r < readings; r++) { + const offset = + r === 0 && random() < 0.2 + ? random() < 0.5 + ? 60_000 + : DAY_MS - 60_000 + : random() < 0.5 + ? hour * 3_600_000 + Math.floor(random() * 3_600_000) + : Math.floor(random() * DAY_MS); + rows.push({ + userId: USER_ID, + type, + source, + value: Math.round(random() * 2_000_000) / 10_000 + 1, + unit: "x", + measuredAt: new Date(dayStart + offset), + deviceType: DEVICES[Math.floor(random() * DEVICES.length)], + glucoseContext: + type === "BLOOD_GLUCOSE" + ? CONTEXTS[Math.floor(random() * CONTEXTS.length)] + : null, + // A tombstone, sometimes the only row its source had that day. + deletedAt: random() < 0.06 ? new Date() : null, + }); + } + } + } + } + // The natural identity is unique; drop accidental collisions. + const seen = new Set(); + const unique = rows.filter((row) => { + const key = `${row.type}|${row.source}|${(row.measuredAt as Date).getTime()}`; + if (seen.has(key)) return false; + seen.add(key); + return true; + }); + await prisma.measurement.createMany({ data: unique }); + return unique.length; +} + +const LADDERS: Array<[string, Prisma.InputJsonValue | null]> = [ + ["default ladders", null], + [ + "a user ladder", + { + pulse: ["OURA", "WITHINGS", "APPLE_HEALTH"], + weight: ["MANUAL", "WITHINGS"], + steps: ["WITHINGS", "APPLE_HEALTH"], + bloodPressure: ["APPLE_HEALTH", "MANUAL", "WITHINGS"], + }, + ], +]; +const ZONES = ["Europe/Berlin", "America/Los_Angeles", "Asia/Kolkata"]; + +async function setLadder(ladder: Prisma.InputJsonValue | null) { + await prisma.user.update({ + where: { id: USER_ID }, + data: { sourcePriorityJson: ladder ?? Prisma.DbNull }, + }); +} + +// --------------------------------------------------------------------------- +// The references, verbatim but for their inputs, which the readers build the +// same way. + +function legacyLiveBuckets(opts: { + userId: string; + type: MeasurementType | null; + from: Date; + to: Date; + cap: number; + priorityJson: unknown; + unitName: string; + timeZone: string; +}) { + const { userId, type, from, to, cap, priorityJson, timeZone } = opts; + const unit = Prisma.raw(`'${opts.unitName}'`); + const cumulativeList = [...CUMULATIVE_HK_TYPES] + .map((t) => `'${t}'`) + .join(","); + const isCumulative = Prisma.raw(`c."type"::text IN (${cumulativeList})`); + const hourlyMeanList = Prisma.raw( + [...HOURLY_MEAN_DAY_TYPES].map((t) => `'${t}'::measurement_type`).join(","), + ); + const rankRaw = Prisma.raw( + buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'), + ); + const dayOfHours = Prisma.raw( + weightedHourMeanSql("h.hour_mean", hourWeightSql("h.n")), + ); + const typeFilter = type + ? Prisma.sql`AND m."type" = ${type}::measurement_type` + : Prisma.empty; + return prisma.$queryRaw` + WITH per_source AS ( + SELECT + m."type", + m."source", + date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, + SUM(m."value")::double precision AS total, + COUNT(*)::int AS cnt, + MIN(m."value") AS min_value, + MAX(m."value") AS max_value + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${from} + AND m."measured_at" <= ${to} + AND m."deleted_at" IS NULL + ${typeFilter} + GROUP BY m."type", m."source", 3 + ), + -- A day of a type with an activity-dependent sampling rate is the weighted + -- mean of its local hours' means, not the mean of its readings: a workout + -- hour holds hundreds of times as many readings as a resting one. + -- Empty for every other type. + hour_mean AS ( + SELECT h."type", h."source", h.day, + ${dayOfHours}::double precision AS day_mean + FROM ( + SELECT + m."type", + m."source", + date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day, + date_trunc('hour', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS hr, + AVG(m."value") AS hour_mean, + COUNT(*) AS n + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${from} + AND m."measured_at" <= ${to} + AND m."deleted_at" IS NULL + AND m."type" IN (${hourlyMeanList}) + ${typeFilter} + GROUP BY m."type", m."source", 3, 4 + ) h + GROUP BY h."type", h."source", h.day + ), + canon AS ( + SELECT DISTINCT ON (p."type", p.day) + p."type", p.day, p.total, p.cnt, p.min_value, p.max_value, hm.day_mean + FROM per_source p + LEFT JOIN hour_mean hm + ON hm."type" = p."type" AND hm."source" = p."source" AND hm.day = p.day + ORDER BY p."type", p.day, (${rankRaw}), p."source"::text + ), + folded AS ( + SELECT + c."type", + date_trunc(${unit}, c.day) AS d, + (CASE + WHEN ${isCumulative} THEN SUM(c.total) + -- Hourly-mean types: every day once, so a window of days does not + -- hand a workout day its thousands of readings as weight. + WHEN COUNT(c.day_mean) = COUNT(*) THEN AVG(c.day_mean) + ELSE SUM(c.total) / SUM(c.cnt) + END)::double precision AS avg, + SUM(c.cnt)::int AS cnt, + MIN(c.min_value) AS min_value, + MAX(c.max_value) AS max_value + FROM canon c + GROUP BY c."type", 2 + ) + SELECT + f."type"::text AS type, + f.d AT TIME ZONE ${timeZone} AS bucket_start, + f.avg, + f.cnt, + f.min_value, + f.max_value + FROM folded f + ORDER BY f.d ASC + LIMIT ${cap} + `; +} + +function legacyAllTimeExtremes( + userId: string, + types: readonly MeasurementType[], + floor: Date | null, + priorityJson: unknown, +) { + const typeList = (list: readonly MeasurementType[]) => + Prisma.join(list.map((t) => Prisma.sql`${t}::measurement_type`)); + const since = + floor === null ? Prisma.empty : Prisma.sql`AND m."measured_at" >= ${floor}`; + // No rollups in the fixture: every type is read raw, in one arm. + const rawArms = [ + 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(types)}) + ${since}`, + ]; + const tierCanon = Prisma.empty; + const rank = (alias: string) => + Prisma.raw( + buildSourceRankCase(priorityJson, `${alias}."type"`, `${alias}."source"`), + ); + const mean = Prisma.raw( + foldMeanSql({ + typeColumn: 'c."type"', + total: "c.total", + count: "c.cnt", + weightedSum: "c.wsum", + weightSum: "c.wdays", + }), + ); + return prisma.$queryRaw` + WITH live AS (${Prisma.join(rawArms, " UNION ALL ")}), + hourly_src AS ( + SELECT * FROM live l + WHERE ${Prisma.raw(isHourlyMeanTypeSql('l."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 + l."type", + l."source", + date_trunc('day', l."measured_at") AS day, + COUNT(*)::int AS cnt, + 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 ( + 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" + `; +} + +function legacyDenseBuckets(params: { + userId: string; + start: Date; + end: Date; + reportTz: string; + denseTypes: readonly MeasurementType[]; +}) { + const { userId, start, end, reportTz, denseTypes } = params; + const typeList = Prisma.join( + denseTypes.map((type) => Prisma.sql`${type}::"measurement_type"`), + ); + const hourlyMeanType = Prisma.raw(isHourlyMeanTypeSql('"type"')); + const hourWeight = Prisma.raw(hourWeightSql("n")); + return prisma.$queryRaw` + WITH localized AS ( + SELECT + m."id", + m."type", + m."source", + m."device_type", + m."glucose_context", + m."value", + m."measured_at", + date_trunc( + 'day', + (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${reportTz} + ) AS local_day, + date_trunc( + 'hour', + (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${reportTz} + ) AS local_hour + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."measured_at" >= ${start} + AND m."measured_at" <= ${end} + AND m."deleted_at" IS NULL + AND m."type" IN (${typeList}) + ), + scored AS ( + SELECT + localized.*, + CASE + WHEN "value" > 0 + THEN 1.509 * (power(ln("value"), 1.084) - 5.381) + ELSE NULL + END AS risk_transform + FROM localized + ), + latest_rows AS ( + SELECT DISTINCT ON ( + local_day, + "type", + "source", + "device_type", + "glucose_context" + ) + local_day, + "type", + "source", + "device_type", + "glucose_context", + "measured_at" AS latest_at, + "value" AS latest_value + FROM localized + ORDER BY + local_day, + "type", + "source", + "device_type", + "glucose_context", + "measured_at" DESC, + "id" DESC + ), + -- The hourly-mean day (pulse): each local hour's mean and reading + -- count, then the weighted sum of the means and the sum of the hour + -- weights per bucket. Empty for every other type. + hour_means AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + AVG("value") AS hour_mean, + COUNT(*) AS n + FROM localized + WHERE ${hourlyMeanType} + GROUP BY + local_day, + local_hour, + "type", + "source", + "device_type", + "glucose_context" + ), + hour_agg AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + SUM(hour_mean * ${hourWeight})::double precision AS hour_weighted_sum, + SUM(${hourWeight})::double precision AS hour_weight + FROM hour_means + GROUP BY + local_day, + "type", + "source", + "device_type", + "glucose_context" + ), + grouped AS ( + SELECT + local_day, + "type", + "source", + "device_type", + "glucose_context", + COUNT(*)::int AS sample_count, + SUM("value")::double precision AS sum_value, + MIN("value")::double precision AS min_value, + MAX("value")::double precision AS max_value, + COUNT(*) FILTER (WHERE "value" > 0)::int AS clinical_count, + SUM("value") FILTER (WHERE "value" > 0)::double precision + AS clinical_sum, + SUM("value" * "value") FILTER (WHERE "value" > 0)::double precision + AS clinical_sum_squares, + MIN("measured_at") FILTER (WHERE "value" > 0) AS clinical_first_at, + MAX("measured_at") FILTER (WHERE "value" > 0) AS clinical_last_at, + COUNT(*) FILTER (WHERE "value" BETWEEN 70 AND 180)::int + AS clinical_tir_count, + COUNT(*) FILTER (WHERE "value" > 0 AND "value" < 70)::int + AS clinical_tbr1_count, + COUNT(*) FILTER (WHERE "value" > 0 AND "value" < 54)::int + AS clinical_tbr2_count, + COUNT(*) FILTER (WHERE "value" > 180)::int + AS clinical_tar1_count, + COUNT(*) FILTER (WHERE "value" > 250)::int + AS clinical_tar2_count, + SUM( + CASE + WHEN risk_transform < 0 + THEN 10 * risk_transform * risk_transform + ELSE 0 + END + ) FILTER (WHERE risk_transform IS NOT NULL)::double precision + AS clinical_low_risk_sum, + SUM( + CASE + WHEN risk_transform > 0 + THEN 10 * risk_transform * risk_transform + ELSE 0 + END + ) FILTER (WHERE risk_transform IS NOT NULL)::double precision + AS clinical_high_risk_sum + FROM scored + GROUP BY + local_day, + "type", + "source", + "device_type", + "glucose_context" + ) + SELECT + grouped.local_day AT TIME ZONE ${reportTz} AS "bucketStart", + grouped."type", + grouped."source", + grouped."device_type" AS "deviceType", + grouped."glucose_context" AS "glucoseContext", + grouped.sample_count AS "count", + grouped.sum_value AS "sumValue", + grouped.min_value AS "minValue", + grouped.max_value AS "maxValue", + latest_rows.latest_at AS "latestAt", + latest_rows.latest_value::double precision AS "latestValue", + grouped.clinical_count AS "clinicalCount", + grouped.clinical_sum AS "clinicalSum", + grouped.clinical_sum_squares AS "clinicalSumSquares", + grouped.clinical_first_at AS "clinicalFirstAt", + grouped.clinical_last_at AS "clinicalLastAt", + grouped.clinical_tir_count AS "clinicalTirCount", + grouped.clinical_tbr1_count AS "clinicalTbr1Count", + grouped.clinical_tbr2_count AS "clinicalTbr2Count", + grouped.clinical_tar1_count AS "clinicalTar1Count", + grouped.clinical_tar2_count AS "clinicalTar2Count", + grouped.clinical_low_risk_sum AS "clinicalLowRiskSum", + grouped.clinical_high_risk_sum AS "clinicalHighRiskSum", + hour_agg.hour_weighted_sum AS "hourWeightedSum", + hour_agg.hour_weight AS "hourWeight" + FROM grouped + INNER JOIN latest_rows + ON latest_rows.local_day = grouped.local_day + AND latest_rows."type" = grouped."type" + AND latest_rows."source" = grouped."source" + AND latest_rows."device_type" IS NOT DISTINCT FROM grouped."device_type" + AND latest_rows."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" + LEFT JOIN hour_agg + ON hour_agg.local_day = grouped.local_day + AND hour_agg."type" = grouped."type" + AND hour_agg."source" = grouped."source" + AND hour_agg."device_type" IS NOT DISTINCT FROM grouped."device_type" + AND hour_agg."glucose_context" IS NOT DISTINCT FROM grouped."glucose_context" + ORDER BY "bucketStart" ASC, grouped."type" ASC + +`; +} + +// --------------------------------------------------------------------------- + +describe("per-source fold parity", () => { + let seeded = 0; + beforeAll(async () => { + await truncateAllTables(prisma); + seeded = await seed(); + }, 120_000); + + it("seeds a fixture worth comparing", () => { + expect(seeded).toBeGreaterThan(5000); + }); + + describe("readLiveBuckets", () => { + const SHAPES: Array< + [string, MeasurementType | null, "daily" | "weekly" | "monthly", number] + > = [ + ["every type, daily, 400 days", null, "daily", 400], + ["every type, weekly, 400 days", null, "weekly", 400], + ["every type, monthly, 400 days", null, "monthly", 400], + ["pulse, daily, 30 days", "PULSE", "daily", 30], + ["steps, weekly, 120 days", "ACTIVITY_STEPS", "weekly", 120], + ]; + const UNIT = { daily: "day", weekly: "week", monthly: "month" } as const; + for (const [ladderLabel, ladder] of LADDERS) { + for (const zone of ZONES) { + for (const [label, type, grain, days] of SHAPES) { + it(`returns the same buckets: ${ladderLabel}, ${zone}, ${label}`, async () => { + await setLadder(ladder); + // A window edge in the middle of a day. + const from = new Date(NOW - days * DAY_MS - 5 * 3_600_000); + const to = new Date(NOW); + const opts = { + userId: USER_ID, + type, + from, + to, + cap: 100_000, + priorityJson: ladder, + timeZone: zone, + }; + const [now] = await capture(() => + readLiveBuckets({ ...opts, grain }), + ); + const before = await legacyLiveBuckets({ + ...opts, + unitName: UNIT[grain], + }); + expect(now.length).toBeGreaterThan(0); + expectSameRows(now, before, ["type", "bucket_start", "cnt"]); + }); + } + } + } + }); + + describe("readAllTimeExtremes", () => { + const TYPES_READ: MeasurementType[] = [ + "WEIGHT", + "BLOOD_PRESSURE_SYS", + "PULSE", + "ACTIVITY_STEPS", + ]; + for (const [ladderLabel, ladder] of LADDERS) { + for (const [floorLabel, floor] of [ + ["all time", null], + ["from a floor mid-day", new Date(NOW - 90 * DAY_MS - 7 * 3_600_000)], + ] as const) { + it(`returns the same figures: ${ladderLabel}, ${floorLabel}`, async () => { + await setLadder(ladder); + const results = await capture(() => + readAllTimeExtremes(USER_ID, TYPES_READ, floor), + ); + // The tier check first, then the fold; no type came from the tier. + expect(results).toHaveLength(2); + expect(results[0]).toEqual([]); + const before = await legacyAllTimeExtremes( + USER_ID, + TYPES_READ, + floor, + ladder, + ); + expect(results[1]).toHaveLength(TYPES_READ.length); + expectSameRows(results[1], before, ["type"]); + }); + } + } + }); + + describe("loadDenseMeasurementBuckets", () => { + const DENSE: MeasurementType[] = [ + "PULSE", + "BLOOD_GLUCOSE", + "HEART_RATE_VARIABILITY", + ]; + for (const zone of ZONES) { + for (const [label, days] of [ + ["400 days", 400], + ["120 days", 120], + ] as const) { + it(`returns the same buckets: ${zone}, ${label}`, async () => { + const params = { + userId: USER_ID, + start: new Date(NOW - days * DAY_MS - 5 * 3_600_000), + end: new Date(NOW), + reportTz: zone, + denseTypes: DENSE, + }; + const now = (await loadDenseMeasurementBuckets( + params, + )) as unknown as Row[]; + const before = await legacyDenseBuckets(params); + expect(now.length).toBeGreaterThan(0); + // Some buckets carry an hour fold and some do not. + expect(now.some((r) => r.hourWeight !== null)).toBe(true); + expect(now.some((r) => r.hourWeight === null)).toBe(true); + expectSameRows(now, before, [ + "bucketStart", + "type", + "source", + "deviceType", + "glucoseContext", + ]); + }); + } + } + }); +});