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", + ]); + }); + } + } + }); +});