tom/HealthLog
1
0 Fork 0
mirror of https://github.com/MBombeck/HealthLog.git synced 2026-11-07 04:12:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'fix/analytics-live-regression' into fix/review-meds-docs

Browse Source
This commit is contained in:
Marc-André Bombeck committed 2026-10-25 06:28:23 +01:00
2 parents 4d4a83fb7d 4e79098cc0
commit 8aab59cdc1
16 files changed
+1671 -267
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/__tests__/baselines/source-function-lines.json
+2 -3
View File
@@ -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,
src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts
+1 -1
View File
@@ -316,7 +316,7 @@ describe("computeBpInTargetFastPath", () => {
// a raw `findMany` (double-counting) fails here.
for (const call of QUERY_RAW.mock.calls) {
const sql = String(call[0]);
expect(sql).toContain("DISTINCT ON");
expect(sql).toContain("canonical_source");
expect(sql).toContain("date_trunc('day'");
}
});
src/lib/analytics/__tests__/source-rank-sql.test.ts
+12 -9
View File
@@ -7,8 +7,10 @@
* self-join. These tests pin:
* 1. the alias-wrapped `canonicalMeasurementsFrom` is exactly the CTE body
* wrapped in `( … ) m` — so the single-reference call site is unchanged;
* 2. the canonical body carries exactly ONE DISTINCT-ON pick — so a fold
* that references the CTE twice runs the expensive pick once;
* 2. the canonical body reads `measurements` once and picks with one window
* function, with no join for the planner to run as a nested loop — so a
* fold that references the CTE twice runs the pick once, and a table
* without statistics cannot turn it quadratic;
* 3. the interval whitelist still rejects an unsafe splice;
* 4. the window counts from an instant the caller took, never from the
* database clock.
@@ -34,21 +36,22 @@ describe("canonicalMeasurementsCte / canonicalMeasurementsFrom", () => {
expect(from).toBe(`(${cte}\n ) m`);
});
it("runs exactly one DISTINCT-ON canonical pick in the body", () => {
it("picks with one window over one scan, with no join", () => {
const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS);
const matches = cte.match(/DISTINCT ON/g) ?? [];
expect(matches).toHaveLength(1);
expect(cte.match(/FROM measurements\b/g) ?? []).toHaveLength(1);
expect(cte.match(/FIRST_VALUE\(/g) ?? []).toHaveLength(1);
expect(cte).not.toMatch(/\bJOIN\b|DISTINCT ON/);
});
it("binds the user id as $1 and scopes both the inner and outer windows", () => {
it("binds the user id as $1 and scopes the scan to the window", () => {
const cte = canonicalMeasurementsCte(RANK, NINETY_DAYS);
// user id is always a bound parameter, never spliced.
expect(cte).toContain('"user_id" = $1');
// The trailing-window bound appears on both the inner pick and the outer
// filter so the folded references stay scoped identically.
// The trailing-window bound scopes the one scan, so the pick and the rows
// it keeps are cut by the same window.
const intervalMatches =
cte.match(/>= \$2::timestamp - INTERVAL '90 days'/g) ?? [];
expect(intervalMatches).toHaveLength(2);
expect(intervalMatches).toHaveLength(1);
});
it("never cuts its window on the database clock", () => {
src/lib/analytics/source-rank-sql.ts
+55 -60
View File
@@ -121,35 +121,45 @@ export interface CanonicalWindow {
/**
* Build the canonical-source SELECT body — the raw `measurements` rows
* restricted to the ladder-winning source per (type, day) — WITHOUT the
* outer `( … ) m` alias wrapper. The inner `DISTINCT ON` picks the
* canonical source for each day and the join keeps only that source's
* readings.
* outer `( … ) m` alias wrapper. One scan reads the window's rows; a window
* function names each (type, day)'s canonical source on every row, and the
* rows of any other source are dropped.
*
* Use this when the same canonical set must be referenced more than once in
* one statement (e.g. a `window_stats` CTE plus the outer aggregate): bind it
* once via `WITH cm AS (${canonicalMeasurementsCte(...)})` and select
* `FROM cm m` everywhere, so the expensive DISTINCT-ON self-join runs once
* rather than per reference. For the single-reference case use
* `canonicalMeasurementsFrom`, which wraps this body in the alias.
* `FROM cm m` everywhere, so the pick runs once rather than per reference.
* For the single-reference case use `canonicalMeasurementsFrom`, which wraps
* this body in the alias.
*
* The rows carry every `measurements` column plus `canonical_source`, the
* picked source (equal to `source` on every row that survives). Readers
* select the columns they need, so the extra one never reaches them.
*
* The user id is bound as `$1`. `rankUnqualified` must be a CASE built with
* unqualified `"type"`/`"source"` columns. `window`, when given, scopes both
* the inner pick and the outer filter to the readings from `from` on, or from
* `interval` (e.g. `"90 days"`, whitelisted) before it. `from` is an instant
* the caller took (`boundInstantSql` or `instantLiteralSql`, see
* `request-clock-sql.ts`), the request's `now` for a trailing window, so the
* cut runs on the caller's clock and never on the database's. `onlyType`, when given, scopes both
* halves to that one metric.
* unqualified `"type"`/`"source"` columns. `window`, when given, scopes the
* scan to the readings from `from` on, or from `interval` (e.g. `"90 days"`,
* whitelisted) before it. `from` is an instant the caller took
* (`boundInstantSql` or `instantLiteralSql`, see `request-clock-sql.ts`), the
* request's `now` for a trailing window, so the cut runs on the caller's
* clock and never on the database's. `onlyType`, when given, scopes the scan
* to that one metric.
*
* The pick groups before it sorts. Its order (rank, then source name) depends
* only on a row's `(type, source)`, so every reading of one source on one day
* sorts alike, and the distinct `(type, day, source)` triples carry the whole
* decision. A hash aggregate reduces the window's rows to those triples (a
* few thousand) and `DISTINCT ON` sorts only them; sorting every raw row by
* a CASE expression no index can supply spilled to disk on a large account.
* The picked source per day is the same by construction;
* `tests/integration/canonical-source-pick-parity.test.ts` holds the old
* statement as its reference.
* The pick is the first source of the day in ladder order (rank, then source
* name), the same order the `DISTINCT ON` pick it replaced used, over the
* same rows, so every (type, day) keeps the same source;
* `tests/integration/canonical-source-pick-parity.test.ts` holds the
* `DISTINCT ON` statement as its reference.
*
* Why no join: the earlier shape grouped the window to `(type, day, source)`,
* picked per day with `DISTINCT ON`, and joined the picks back to the rows.
* Whether that join ran as a hash join or as a nested loop that rescanned the
* pick for every row was the planner's call, made from table statistics. On
* a table without them (a new install, or an import still under the
* autoanalyze threshold) it estimated one row and chose the nested loop: on a
* 6,000-row fixture the all-time read took seconds, on a ten-year account it
* ran past the statement timeout. A window over one scan has no such choice
* to get wrong; `canonical-source-pick-unanalysed.test.ts` holds that.
*/
export function canonicalMeasurementsCte(
rankUnqualified: string,
@@ -161,52 +171,37 @@ export function canonicalMeasurementsCte(
: window.interval === undefined
? assertInstantSql(window.from)
: `${assertInstantSql(window.from)} - INTERVAL '${assertInterval(window.interval)}'`;
const sinceInner = bound ? `AND "measured_at" >= ${bound}` : "";
const sinceOuter = bound ? `AND mm."measured_at" >= ${bound}` : "";
// A single-type reader (a series for one metric) scopes both halves to that
// type, so the pick scans one metric's rows instead of every row the user
const since = bound ? `AND mm."measured_at" >= ${bound}` : "";
// A single-type reader (a series for one metric) scopes the scan to that
// type, so the pick reads one metric's rows instead of every row the user
// has in the window.
const typeInner = onlyType
? `AND "type" = '${assertEnumLiteral(onlyType)}'::"measurement_type"`
: "";
const typeOuter = onlyType
const typeOnly = onlyType
? `AND mm."type" = '${assertEnumLiteral(onlyType)}'::"measurement_type"`
: "";
return `
SELECT mm.*
FROM measurements mm
JOIN (
SELECT DISTINCT ON ("type", d)
"type" AS t,
d,
"source" AS canon
FROM (
SELECT "type", date_trunc('day', "measured_at") AS d, "source"
FROM measurements
WHERE "user_id" = $1
AND "deleted_at" IS NULL
${sinceInner}
${typeInner}
GROUP BY "type", date_trunc('day', "measured_at"), "source"
) g
ORDER BY "type", d, (${rankUnqualified}), "source"::text
) c
ON c.t = mm."type"
AND c.d = date_trunc('day', mm."measured_at")
AND c.canon = mm."source"
WHERE mm."user_id" = $1
AND mm."deleted_at" IS NULL
${sinceOuter}
${typeOuter}`;
SELECT w.*
FROM (
SELECT mm.*,
FIRST_VALUE(mm."source") OVER (
PARTITION BY mm."type", date_trunc('day', mm."measured_at")
ORDER BY (${rankUnqualified}), mm."source"::text
) AS canonical_source
FROM measurements mm
WHERE mm."user_id" = $1
AND mm."deleted_at" IS NULL
${since}
${typeOnly}
) w
WHERE w."source" = w.canonical_source`;
}
/**
* Build a FROM-clause subquery (aliased `m`) that restricts raw `measurements`
* to the canonical-source rows per (type, day): the inner `DISTINCT ON` picks
* the ladder-winning source for each day, and the join keeps only that source's
* readings. Use it in place of `FROM measurements m` so a live aggregate over
* an overlapping vital never blends two devices — the same collapse the rollup
* readers apply, kept in lockstep for live/rollup parity.
* to the canonical-source rows per (type, day): the ladder-winning source for
* each day, and only that source's readings. Use it in place of
* `FROM measurements m` so a live aggregate over an overlapping vital never
* blends two devices — the same collapse the rollup readers apply, kept in
* lockstep for live/rollup parity.
*
* The user id is bound as `$1`; `rankUnqualified`, `window` and `onlyType` are
* those of `canonicalMeasurementsCte`.
src/lib/analytics/summaries-slice.ts
+4 -4
View File
@@ -573,10 +573,10 @@ async function computeFromRollups(
const [narrows, latests, dayBuckets, accumulatorRows, preFoldRemainder] =
await Promise.all([
// The FROM is restricted to the canonical-source rows per (type, day): the
// inner DISTINCT ON picks the ladder-winning source for each day, and the
// join keeps only that source's readings so the 90-day AVG / median / slope
// never blend two devices that both reported the same vital.
// The FROM is restricted to the canonical-source rows per (type, day):
// the ladder-winning source for each day, and only that source's
// readings, so the 90-day AVG / median / slope never blend two devices
// that both reported the same vital.
// The window means weigh each day once for an hourly-mean type
// (`windowedMean`); the median stays over every reading.
prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
src/lib/doctor-report/dense-buckets.ts
+32 -80
View File
@@ -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<DenseMeasurementBucket[]>`
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
`;
src/lib/insights/comprehensive-aggregator.ts
+5 -5
View File
@@ -403,10 +403,10 @@ async function buildFromRollups(
timeSubquery(timings, "narrow", () =>
prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
`
-- v1.18.11 perf#3a — the canonical-source self-join is the dominant
-- v1.18.11 perf#3a — the canonical-source pick is the dominant
-- cost; bind it ONCE as a CTE and reference it from both window_stats
-- and the outer aggregate so the DISTINCT-ON pick runs a single time
-- instead of twice per rebuild. Output is identical.
-- and the outer aggregate so the pick runs a single time instead of
-- twice per rebuild. Output is identical.
WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)}
),
-- The z-score's mean and spread stay over every reading: they measure
@@ -626,9 +626,9 @@ async function buildFromLiveAggregate(
timeSubquery(timings, "heavy", () =>
prisma.$queryRawUnsafe<HeavyAggregateRow[]>(
`
-- v1.18.11 perf#3a — bind the canonical-source self-join once (see the
-- v1.18.11 perf#3a — bind the canonical-source pick once (see the
-- warm narrow query). window_stats + the outer aggregate both read cm,
-- so the DISTINCT-ON pick runs once. Output is identical.
-- so the pick runs once. Output is identical.
WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, NINETY_DAYS)}
),
-- The z-score's mean and spread stay over every reading: they measure
src/lib/insights/feature-blocks.ts
+22 -31
View File
@@ -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}
src/lib/measurements/__tests__/series-canonical.test.ts
+6 -7
View File
@@ -10,9 +10,8 @@ const SINCE = new Date("2026-03-01T06:30:00.000Z");
describe("seriesRowsFrom", () => {
it("collapses a type that has a ladder to the ladder's source per day", () => {
const sql = seriesRowsFrom(null, "PULSE", SINCE);
expect(sql).toContain("DISTINCT ON");
// Scoped to the one metric, in both halves of the helper.
expect(sql).toContain(`"type" = 'PULSE'::"measurement_type"`);
expect(sql).toContain("canonical_source");
// Scoped to the one metric in the helper's one scan.
expect(sql).toContain(`mm."type" = 'PULSE'::"measurement_type"`);
expect(sql.trimEnd().endsWith(") m")).toBe(true);
});
@@ -21,7 +20,7 @@ describe("seriesRowsFrom", () => {
const sql = seriesRowsFrom(null, "WEIGHT", SINCE);
expect(
sql.match(/>= '2026-03-01 06:30:00.000'::timestamp - INTERVAL '1 day'/g),
).toHaveLength(2);
).toHaveLength(1);
// Counted from the caller's `since`, never from the database clock.
expect(sql).not.toMatch(/NOW\(\)/i);
});
@@ -46,7 +45,7 @@ describe("seriesRowsFrom", () => {
"keeps every source for %s, which has no ladder",
(type) => {
const sql = seriesRowsFrom(null, type, SINCE);
expect(sql).not.toContain("DISTINCT ON");
expect(sql).not.toContain("canonical_source");
expect(sql).toContain(`mm."type" = '${type}'::"measurement_type"`);
expect(sql).toContain(`mm."user_id" = $1`);
expect(sql).toContain(`mm."deleted_at" IS NULL`);
@@ -69,9 +68,9 @@ describe("canonicalMeasurementsCte, one-metric scope", () => {
expect(sql).not.toContain("measurement_type");
});
it("scopes the pick and the join to the type when one is given", () => {
it("scopes the scan to the type when one is given", () => {
const sql = canonicalMeasurementsCte(rank, NINETY_DAYS, "PULSE");
expect(sql.match(/'PULSE'::"measurement_type"/g)).toHaveLength(2);
expect(sql.match(/'PULSE'::"measurement_type"/g)).toHaveLength(1);
});
it("refuses anything that is not a closed-enum literal", () => {
src/lib/measurements/daily-series-read.ts
+34 -33
View File
@@ -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 (
src/lib/measurements/series/read-blood-pressure.ts
+57 -28
View File
@@ -32,7 +32,8 @@ export async function readBloodPressureSeries(
// the whole edge day, not only the slice after `since`.
//
// `rank` is the closed-enum whitelist splice from `buildSourceRankCase`;
// every request value is a positional parameter.
// every request value is a positional parameter. The statement and why it
// is a window rather than a join are under `bloodPressureSeriesSql`.
const rank = buildSourceRankCase(
await loadUserSourcePriority(userId),
'"type"',
@@ -47,33 +48,7 @@ export async function readBloodPressureSeries(
measured_at: Date;
}>
>(
`
SELECT m."id", m."type"::text AS type, m."source"::text AS source,
m."value", m."measured_at"
FROM measurements m
JOIN (
SELECT DISTINCT ON (date_trunc('day', "measured_at"))
date_trunc('day', "measured_at") AS d,
"source" AS canon
FROM measurements
WHERE "user_id" = $1
AND "type" = 'BLOOD_PRESSURE_SYS'::"measurement_type"
AND "deleted_at" IS NULL
AND "value" >= $3
AND "measured_at" >= date_trunc('day', $2::timestamptz AT TIME ZONE 'UTC')
ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text
) c
ON c.d = date_trunc('day', m."measured_at")
AND c.canon = m."source"
WHERE m."user_id" = $1
AND m."deleted_at" IS NULL
AND m."measured_at" >= $2
AND (
(m."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" AND m."value" >= $3)
OR (m."type" = 'BLOOD_PRESSURE_DIA'::"measurement_type" AND m."value" >= $4)
)
ORDER BY m."measured_at" ASC, m."id" ASC
`,
bloodPressureSeriesSql(rank),
userId,
since,
VALUE_RANGES.BLOOD_PRESSURE_SYS.min,
@@ -125,3 +100,57 @@ export async function readBloodPressureSeries(
});
return { points };
}
/**
* The series statement: every floor-passing systolic and diastolic row from
* `$2` on, of the source the ladder picks for its UTC day. `$1` is the user,
* `$2` the window start, `$3` and `$4` the systolic and diastolic floors;
* `rank` is a `buildSourceRankCase` CASE over unqualified `"type"` and
* `"source"`.
*
* One scan reads the window from the start of its first UTC day, so the pick
* sees the whole edge day. A window function names each day's source on
* every row: the day's rows are ordered systolic first, then by ladder rank,
* then by source name, and the first row's source is taken only when that
* row is a systolic one, so a day with no floor-passing systolic picks no
* source and shows nothing, as before. Rows of any other source, and rows
* of the edge day before `$2`, are dropped after the pick.
*
* Why no join: the pick used to be a `DISTINCT ON` over the day's systolic
* rows joined back to the readings. On a table without planner statistics
* (a new install, an import still under the autoanalyze threshold) the
* planner estimated one row on each side and ran that join as a nested loop
* that rescanned the pick for every reading: ten years of readings cost
* about 5,000 row visits per row. The window has no such choice to get
* wrong; `tests/integration/canonical-source-pick-unanalysed.test.ts` holds
* that, and `tests/integration/bp-series-pick-parity.test.ts` holds the old
* statement as the reference for the rows it returns.
*/
export function bloodPressureSeriesSql(rank: string): string {
return `
SELECT w."id", w."type"::text AS type, w."source"::text AS source,
w."value", w."measured_at"
FROM (
SELECT mm."id", mm."type", mm."source", mm."value", mm."measured_at",
FIRST_VALUE(
CASE WHEN mm."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type"
THEN mm."source" END
) OVER (
PARTITION BY date_trunc('day', mm."measured_at")
ORDER BY mm."type" <> 'BLOOD_PRESSURE_SYS'::"measurement_type",
(${rank}), mm."source"::text
) AS canonical_source
FROM measurements mm
WHERE mm."user_id" = $1
AND mm."deleted_at" IS NULL
AND mm."measured_at" >= date_trunc('day', $2::timestamptz AT TIME ZONE 'UTC')
AND (
(mm."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" AND mm."value" >= $3)
OR (mm."type" = 'BLOOD_PRESSURE_DIA'::"measurement_type" AND mm."value" >= $4)
)
) w
WHERE w."source" = w.canonical_source
AND w."measured_at" >= $2
ORDER BY w."measured_at" ASC, w."id" ASC
`;
}
src/lib/provenance/__tests__/source-provenance.test.ts
+5 -3
View File
@@ -115,10 +115,12 @@ describe("the provenance query orders like the canonical pick", () => {
`ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text`,
),
);
// The shared pick groups to (type, day, source) first and orders the
// groups by the same keys, the UTC day named `d`.
// The shared pick partitions by type and UTC day and orders each day's
// rows by the same keys.
expect(squash(canonicalMeasurementsCte(rank))).toContain(
squash(`ORDER BY "type", d, (${rank}), "source"::text`),
squash(
`PARTITION BY mm."type", date_trunc('day', mm."measured_at") ORDER BY (${rank}), mm."source"::text`,
),
);
// Values are bound, never spliced.
expect(userId).toBe("u1");
tests/integration/bp-series-pick-parity.test.ts
+282
View File
@@ -0,0 +1,282 @@
/**
* The blood-pressure series pick, against a real Postgres: the shipped
* statement (one window function over one scan names each UTC day's source)
* against the `DISTINCT ON` pick over the day's systolic rows joined back to
* the readings, kept verbatim below as the reference.
*
* Both must return exactly the same rows in the same order, under the
* default ladder and under user ladders that reorder it, for window starts
* mid-day, on and either side of UTC midnight and on a local midnight, under
* a UTC session and under sessions in other zones (the day cut reads the
* session zone in both statements), over a fixture that carries what could
* tell them apart: overlapping sources on one day, sources off the ladder
* (the name tiebreak decides), a source whose only systolic that day is
* below the floor, a diastolic below its floor, a diastolic with no
* systolic, a systolic with no diastolic, tombstones, readings a minute
* either side of UTC midnight, both DST changes in the user's zone, empty
* days, and several readings per source per day.
*/
import { beforeAll, describe, expect, it } from "vitest";
import type { MeasurementSource } from "@/generated/prisma/client";
import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql";
import { bloodPressureSeriesSql } from "@/lib/measurements/series/read-blood-pressure";
import { VALUE_RANGES } from "@/lib/validations/measurement";
import { makeUser } from "../kit/factories";
import { getPrismaClient, truncateAllTables } from "./setup";
const USER_ID = "user-bp-series-parity";
const DAY_MS = 86_400_000;
const SYS_MIN = VALUE_RANGES.BLOOD_PRESSURE_SYS.min;
const DIA_MIN = VALUE_RANGES.BLOOD_PRESSURE_DIA.min;
/** The statement as it stood before the window: a pick joined back. */
function legacyBloodPressureSeriesSql(rank: string): string {
return `
SELECT m."id", m."type"::text AS type, m."source"::text AS source,
m."value", m."measured_at"
FROM measurements m
JOIN (
SELECT DISTINCT ON (date_trunc('day', "measured_at"))
date_trunc('day', "measured_at") AS d,
"source" AS canon
FROM measurements
WHERE "user_id" = $1
AND "type" = 'BLOOD_PRESSURE_SYS'::"measurement_type"
AND "deleted_at" IS NULL
AND "value" >= $3
AND "measured_at" >= date_trunc('day', $2::timestamptz AT TIME ZONE 'UTC')
ORDER BY date_trunc('day', "measured_at"), (${rank}), "source"::text
) c
ON c.d = date_trunc('day', m."measured_at")
AND c.canon = m."source"
WHERE m."user_id" = $1
AND m."deleted_at" IS NULL
AND m."measured_at" >= $2
AND (
(m."type" = 'BLOOD_PRESSURE_SYS'::"measurement_type" AND m."value" >= $3)
OR (m."type" = 'BLOOD_PRESSURE_DIA'::"measurement_type" AND m."value" >= $4)
)
ORDER BY m."measured_at" ASC, m."id" ASC
`;
}
/** 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[] = [
"WITHINGS",
"APPLE_HEALTH",
"MANUAL",
// Off the blood-pressure ladder: rank alike, the name tiebreak decides.
"OURA",
"WHOOP",
];
const NOW = Date.now();
const TODAY_UTC = Math.floor(NOW / DAY_MS) * DAY_MS;
type Row = {
userId: string;
type: "BLOOD_PRESSURE_SYS" | "BLOOD_PRESSURE_DIA";
source: MeasurementSource;
value: number;
unit: string;
measuredAt: Date;
deletedAt: Date | null;
};
async function seed(): Promise<number> {
const prisma = getPrismaClient();
await makeUser({ id: USER_ID, timezone: "Europe/Berlin" });
const random = rng(7);
const rows: Row[] = [];
const push = (
type: Row["type"],
source: MeasurementSource,
value: number,
at: number,
) =>
rows.push({
userId: USER_ID,
type,
source,
value,
unit: "mmHg",
measuredAt: new Date(at),
// A tombstone, sometimes the only row its source had that day.
deletedAt: random() < 0.06 ? new Date() : null,
});
// 400 days: over both DST changes in the user's zone.
for (let day = 0; day < 400; day++) {
if (day % 13 === 6) continue; // an empty day
const dayStart = TODAY_UTC - day * DAY_MS;
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() * 3);
for (let r = 0; r < readings; r++) {
// Mostly through the day, sometimes a minute either side of UTC
// midnight.
const at =
dayStart +
(r === 0 && random() < 0.25
? random() < 0.5
? 60_000
: DAY_MS - 60_000
: Math.floor(random() * DAY_MS));
const shape = random();
if (shape < 0.08) {
// A systolic below the floor: it must not claim the day.
push(
"BLOOD_PRESSURE_SYS",
source,
Math.floor(random() * SYS_MIN),
at,
);
push("BLOOD_PRESSURE_DIA", source, 80, at);
} else if (shape < 0.14) {
// A diastolic below its floor.
push("BLOOD_PRESSURE_SYS", source, 125, at);
push(
"BLOOD_PRESSURE_DIA",
source,
Math.floor(random() * DIA_MIN),
at,
);
} else if (shape < 0.2) {
// A diastolic alone: it never picks a source.
push("BLOOD_PRESSURE_DIA", source, 78, at);
} else if (shape < 0.25) {
// A systolic alone.
push("BLOOD_PRESSURE_SYS", source, 131, at);
} else {
const sys = 100 + Math.round(random() * 600) / 10;
push("BLOOD_PRESSURE_SYS", source, sys, at);
push(
"BLOOD_PRESSURE_DIA",
source,
60 + Math.round(random() * 300) / 10,
at + Math.floor(random() * 4) * 60_000,
);
}
}
}
}
// The natural identity is unique; drop accidental collisions.
const seen = new Set<string>();
const unique = rows.filter((row) => {
const key = `${row.type}|${row.source}|${row.measuredAt.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, unknown]> = [
["default ladder", null],
[
"a user ladder that reorders it",
{ bloodPressure: ["MANUAL", "APPLE_HEALTH", "WITHINGS"] },
],
];
const STARTS: Array<[string, Date]> = [
["ten years", new Date(NOW - 3650 * DAY_MS)],
["a year, mid-day", new Date(TODAY_UTC - 365 * DAY_MS + 13 * 3_600_000)],
["thirty days, on UTC midnight", new Date(TODAY_UTC - 30 * DAY_MS)],
[
"a week, a minute before UTC midnight",
new Date(TODAY_UTC - 7 * DAY_MS - 60_000),
],
[
"a week, a minute after UTC midnight",
new Date(TODAY_UTC - 7 * DAY_MS + 60_000),
],
// 22:00 UTC is Berlin midnight in summer, 23:00 in winter.
[
"ninety days, on a local midnight",
new Date(TODAY_UTC - 91 * DAY_MS + 22 * 3_600_000),
],
["two days, mid-day", new Date(TODAY_UTC - 2 * DAY_MS + 9 * 3_600_000)],
];
const ZONES = [
"UTC",
"Europe/Berlin",
"Pacific/Auckland",
"America/Los_Angeles",
];
describe("blood-pressure series pick parity", () => {
let seeded = 0;
beforeAll(async () => {
await truncateAllTables(getPrismaClient());
seeded = await seed();
});
it("seeds a fixture worth comparing", () => {
expect(seeded).toBeGreaterThan(2000);
});
for (const [ladderLabel, ladder] of LADDERS) {
for (const zone of ZONES) {
for (const [startLabel, since] of STARTS) {
it(`returns the same rows: ${ladderLabel}, ${zone} session, ${startLabel}`, async () => {
const prisma = getPrismaClient();
const rank = buildSourceRankCase(ladder, '"type"', '"source"');
const select = (sql: string) =>
prisma.$transaction(async (tx) => {
await tx.$executeRawUnsafe(`SET LOCAL TIME ZONE '${zone}'`);
return tx.$queryRawUnsafe<
Array<{
id: string;
type: string;
source: string;
value: number;
measured_at: Date;
}>
>(sql, USER_ID, since, SYS_MIN, DIA_MIN);
});
const now = await select(bloodPressureSeriesSql(rank));
const before = await select(legacyBloodPressureSeriesSql(rank));
expect(now.length).toBeGreaterThan(0);
expect(now).toEqual(before);
// Not trivial: both halves came back, from more than one source
// over the window, and the pick dropped rows that were live,
// floor-passing and inside the window.
expect(new Set(now.map((r) => r.type)).size).toBe(2);
const live = await prisma.$queryRawUnsafe<Array<{ n: number }>>(
`SELECT COUNT(*)::int AS n FROM measurements
WHERE "user_id" = $1 AND "deleted_at" IS NULL
AND "measured_at" >= $2
AND (("type" = 'BLOOD_PRESSURE_SYS' AND "value" >= $3)
OR ("type" = 'BLOOD_PRESSURE_DIA' AND "value" >= $4))`,
USER_ID,
since,
SYS_MIN,
DIA_MIN,
);
expect(now.length).toBeLessThan(live[0].n);
if (since.getTime() < TODAY_UTC - 30 * DAY_MS) {
expect(new Set(now.map((r) => r.source)).size).toBeGreaterThan(1);
}
});
}
}
}
});
tests/integration/canonical-source-pick-parity.test.ts
+3 -3
View File
@@ -1,8 +1,8 @@
/**
* The canonical-source pick, against a real Postgres: the shipped statement
* (group the window to `(type, day, source)` first, then `DISTINCT ON`)
* against the statement it replaced (`DISTINCT ON` over every raw row),
* kept verbatim below as the reference. Only its window anchor changed: it
* (one window function over one scan names each day's source) against the
* `DISTINCT ON` pick joined back to the rows, kept verbatim below as the
* reference. Only its window anchor changed: it
* counts from the same bound instant the shipped helper takes, where it read
* `NOW()`, so both cut one window.
*
tests/integration/canonical-source-pick-unanalysed.test.ts
+399
View File
@@ -0,0 +1,399 @@
/**
* The canonical-source pick (`canonicalMeasurementsCte`) must not depend on
* planner statistics. A new install, or an account whose import is still
* under the autoanalyze threshold, reads against a `measurements` table the
* planner has no statistics for. The pick used to join its per-day choices
* back to the rows; on such a table the planner estimated one row on each
* side and ran the join as a nested loop, rescanning the whole pick for every
* reading. `GET /api/analytics` on 6,000 blood-pressure rows took seconds,
* and on a ten-year account the all-time read ran past the statement timeout.
*
* Latency on a testcontainer says little, and the statement count did not
* move at all, so this file holds the work instead: every reader shape the
* helper serves is run under `EXPLAIN ANALYZE` on a table with no statistics,
* and the rows its plan nodes produce or discard, times their loops, must
* 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, vi } from "vitest";
import type {
MeasurementSource,
MeasurementType,
Prisma,
} from "@/generated/prisma/client";
import {
boundInstantSql,
instantLiteralSql,
} from "@/lib/analytics/request-clock-sql";
import {
buildSourceRankCase,
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";
import { getPrismaClient, truncateAllTables } from "./setup";
const prisma = getPrismaClient();
const USER = "canonical-pick-unanalysed";
const DAY = 86_400_000;
/**
* Rows produced or discarded, summed over the plan, per row seeded. The
* window pick sits at 4 or under on every shape below; the nested loop this
* guards against measured 200 on the 90-day read and over 3,000 on the
* one-type read, and ran past a minute on the all-time read.
*/
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.
*/
const STATEMENT_TIMEOUT = "10s";
/**
* Ten years of paired blood pressure from one source with a second source on
* every third day, and a year of pulse from two sources a day: overlapping
* sources, so the pick has days to decide.
*/
function seedRows(): Prisma.MeasurementCreateManyInput[] {
const rows: Prisma.MeasurementCreateManyInput[] = [];
const now = Date.now();
const push = (
type: MeasurementType,
source: MeasurementSource,
value: number,
at: number,
) =>
rows.push({
userId: USER,
type,
source,
value,
unit: type === "PULSE" ? "bpm" : "mmHg",
measuredAt: new Date(at),
});
for (let d = 0; d < 3000; d += 1) {
const at = now - (d + 0.25) * DAY;
push("BLOOD_PRESSURE_SYS", "WITHINGS", 120 + (d % 30), at);
push("BLOOD_PRESSURE_DIA", "WITHINGS", 75 + (d % 15), at);
if (d % 3 === 0) {
push("BLOOD_PRESSURE_SYS", "MANUAL", 130 + (d % 20), at + 600_000);
push("BLOOD_PRESSURE_DIA", "MANUAL", 80 + (d % 10), at + 600_000);
}
}
for (let d = 0; d < 365; d += 1) {
for (const h of [7, 13, 21]) {
const at = now - d * DAY - h * 3_600_000;
push("PULSE", "WITHINGS", 60 + ((d + h) % 20), at);
push("PULSE", "APPLE_HEALTH", 62 + ((d + h) % 18), at + 60_000);
}
}
return rows;
}
interface PlanNode {
"Actual Rows"?: number;
"Actual Loops"?: number;
"Rows Removed by Filter"?: number;
"Rows Removed by Join Filter"?: number;
"Rows Removed by Index Recheck"?: number;
Plans?: PlanNode[];
}
/**
* Every row a node returned or threw away, over all its loops. EXPLAIN
* reports both per loop, so a node rescanned once per outer row counts every
* rescan.
*/
function rowVisits(node: PlanNode): number {
const perLoop =
(node["Actual Rows"] ?? 0) +
(node["Rows Removed by Filter"] ?? 0) +
(node["Rows Removed by Join Filter"] ?? 0) +
(node["Rows Removed by Index Recheck"] ?? 0);
const own = perLoop * (node["Actual Loops"] ?? 1);
return own + (node.Plans ?? []).reduce((sum, p) => sum + rowVisits(p), 0);
}
const NOW = new Date();
const SHAPES: Array<
[
label: string,
window: CanonicalWindow | undefined,
type: MeasurementType | undefined,
from: Date | undefined,
]
> = [
// The summaries slice's all-time read.
[
"every type, 15 years",
{ from: boundInstantSql(2), interval: "15 years" },
undefined,
NOW,
],
// The summaries slice's and the comprehensive aggregator's 90-day reads.
[
"every type, 90 days",
{ from: boundInstantSql(2), interval: "90 days" },
undefined,
NOW,
],
// The BP-in-target fast path: from its 365-day start, no interval.
[
"every type, from an instant",
{ from: boundInstantSql(2) },
undefined,
new Date(NOW.getTime() - 365 * DAY),
],
// A metric series: one type, from a literal instant one day wider.
[
"one type, ten years",
{
from: instantLiteralSql(new Date(NOW.getTime() - 3650 * DAY)),
interval: "1 day",
},
"BLOOD_PRESSURE_SYS",
undefined,
],
["every type, no window", undefined, undefined, undefined],
];
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<unknown>,
workPerRow: number,
) {
const original = appPrisma.$queryRaw.bind(appPrisma) as (
...args: unknown[]
) => Promise<unknown>;
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<Array<{ "QUERY PLAN": Array<{ Plan: PlanNode }> }>>(
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
* timeout.
*/
async function expectBoundedPlan(sql: string, params: unknown[]) {
const [{ "QUERY PLAN": plan }] = await prisma.$transaction(async (tx) => {
await tx.$executeRawUnsafe(
`SET LOCAL statement_timeout = '${STATEMENT_TIMEOUT}'`,
);
return tx.$queryRawUnsafe<
Array<{ "QUERY PLAN": Array<{ Plan: PlanNode }> }>
>(`EXPLAIN (ANALYZE, FORMAT JSON) ${sql}`, ...params);
});
const visits = rowVisits(plan[0].Plan);
expect(
visits,
`${visits} row visits for ${seeded} rows seeded:\n` +
JSON.stringify(plan[0].Plan, null, 1),
).toBeLessThanOrEqual(WORK_PER_ROW * seeded);
}
beforeAll(async () => {
await truncateAllTables(prisma);
await prisma.$executeRawUnsafe(
"ALTER TABLE measurements SET (autovacuum_enabled = false)",
);
// Statistics another file in this worker gathered survive its TRUNCATE.
await prisma.$executeRawUnsafe(
"DELETE FROM pg_statistic WHERE starelid = 'measurements'::regclass",
);
await prisma.user.create({ data: { id: USER, username: USER } });
const rows = seedRows();
seeded = rows.length;
await prisma.measurement.createMany({ data: rows });
}, 120_000);
afterAll(async () => {
await prisma.$executeRawUnsafe(
"ALTER TABLE measurements RESET (autovacuum_enabled)",
);
});
describe("canonical-source pick on a table without statistics", () => {
it("runs against a table the planner has no statistics for", async () => {
expect(seeded).toBeGreaterThan(10_000);
const [stats] = await prisma.$queryRawUnsafe<
Array<{ columns: number; reltuples: number }>
>(
`SELECT (SELECT COUNT(*) FROM pg_statistic
WHERE starelid = 'measurements'::regclass)::int AS columns,
reltuples::float8 AS reltuples
FROM pg_class WHERE oid = 'measurements'::regclass`,
);
expect(stats).toEqual({ columns: 0, reltuples: -1 });
});
it.each(SHAPES)(
"reads the window a bounded number of times: %s",
async (_label, window, onlyType, from) => {
const rank = buildSourceRankCase(null, '"type"', '"source"');
const sql = `WITH cm AS (${canonicalMeasurementsCte(rank, window, onlyType)}
)
SELECT COUNT(*)::int AS n, SUM(cm."value")::float8 AS total FROM cm`;
const params: unknown[] = from ? [USER, from] : [USER];
await expectBoundedPlan(sql, params);
// The pick kept rows and dropped the second source's.
const [{ n }] = await prisma.$queryRawUnsafe<Array<{ n: number }>>(
sql,
...params,
);
expect(n).toBeGreaterThan(0);
expect(n).toBeLessThan(seeded);
},
);
});
/**
* The blood-pressure series picks its own source per day (from the day's
* systolic rows above the floor, applied to both halves of the reading), so
* it does not go through `canonicalMeasurementsCte` and needs its own row
* here. Its windows are the series route's: ten years, a quarter, a week.
* The `DISTINCT ON` pick joined back to the readings measured about 4,900
* row visits per row seeded on the ten-year read; the window stays at 4.
*/
describe("blood-pressure series pick on a table without statistics", () => {
it.each([
["3650 days", 3650],
["90 days", 90],
["7 days", 7],
])("reads the window a bounded number of times: %s", async (_label, days) => {
const rank = buildSourceRankCase(null, '"type"', '"source"');
const params: unknown[] = [
USER,
new Date(NOW.getTime() - days * DAY),
VALUE_RANGES.BLOOD_PRESSURE_SYS.min,
VALUE_RANGES.BLOOD_PRESSURE_DIA.min,
];
const sql = bloodPressureSeriesSql(rank);
await expectBoundedPlan(sql, params);
// Both halves came back, and the second source's days were dropped.
const rows = await prisma.$queryRawUnsafe<
Array<{ type: string; source: string }>
>(sql, ...params);
expect(rows.some((r) => r.type === "BLOOD_PRESSURE_SYS")).toBe(true);
expect(rows.some((r) => r.type === "BLOOD_PRESSURE_DIA")).toBe(true);
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<unknown[] | Map<unknown, unknown>>]>([
[
"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<unknown, unknown> = [];
await expectBoundedReader(async () => {
out = await read();
}, FOLD_WORK_PER_ROW);
expect(out instanceof Map ? out.size : out.length).toBeGreaterThan(0);
});
});
tests/integration/per-source-fold-parity.test.ts
+752
View File
@@ -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<string, unknown>;
/** The rows each `$queryRaw` the reader issued returned, in call order. */
async function capture(read: () => Promise<unknown>): Promise<Row[][]> {
const results: Row[][] = [];
const original = appPrisma.$queryRaw.bind(appPrisma) as (
...args: unknown[]
) => Promise<Row[]>;
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<GlucoseContext | null> = [null, "FASTING", "RANDOM"];
const DEVICES: Array<string | null> = [null, "watch", "phone"];
async function seed(): Promise<number> {
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<string>();
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<Row[]>`
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<Row[]>`
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<Row[]>`
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",
]);
});
}
}
});
});