From 3634f6c5206e602253ac5bf70dc8afa233cbe16b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 4 Oct 2026 11:22:17 +0200 Subject: [PATCH] Average pulse by day in the dashboard, insights and series stats The dashboard summaries, the insights aggregate and the series stats strip still took a plain mean over pulse readings for their 7-day, 30-day, prior-month and all-time figures, while the year-ago figure and the chart already weighed each day once, so the year delta compared two different statistics. They now go through the day-mean helper: each day is the mean of its hours' means and a window the mean of its days, in the frame each reader buckets its days by. Count, min, max, the median and the spread stay over every reading; every other type is unchanged. --- src/app/api/measurements/series/route.ts | 86 +++++-- .../__tests__/summaries-slice.test.ts | 29 +-- src/lib/analytics/summaries-slice.ts | 140 +++++++---- src/lib/insights/comprehensive-aggregator.ts | 64 +++-- .../integration/pulse-day-mean-zones.test.ts | 221 ++++++++++++++++++ .../pulse-window-mean-readers.test.ts | 209 +++++++++++++++++ 6 files changed, 658 insertions(+), 91 deletions(-) create mode 100644 tests/integration/pulse-day-mean-zones.test.ts create mode 100644 tests/integration/pulse-window-mean-readers.test.ts diff --git a/src/app/api/measurements/series/route.ts b/src/app/api/measurements/series/route.ts index 794660abe..f91091ced 100644 --- a/src/app/api/measurements/series/route.ts +++ b/src/app/api/measurements/series/route.ts @@ -27,6 +27,12 @@ import { reconstructSleepNights } from "@/lib/analytics/sleep-night"; import { VALUE_RANGES } from "@/lib/validations/measurement"; import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; import { usesHourlyMeanDay } from "@/lib/measurements/day-statistic"; +import { + dayWeightedRows, + readingsMean, + windowMeanSql, + zoneDayFrame, +} from "@/lib/measurements/day-mean"; import { seriesRowsFrom } from "@/lib/measurements/series-canonical"; import { resolveUserTimezone, userDayKey } from "@/lib/tz/resolver"; import { convertGlucose, resolveGlucoseUnit } from "@/lib/glucose"; @@ -487,17 +493,50 @@ export const GET = apiHandler(async (request: NextRequest) => { // strip's mean/min/max/stdDev/count are the same figures the raw path // computed: `AVG`/`MIN`/`MAX` over `value`, `STDDEV_POP` matching the // JS population-variance helper, exact row count. Aggregate-only — - // no row transfer. - const [rawAgg] = await prisma.$queryRawUnsafe< - Array<{ - n: number; - mean: number | null; - min: number | null; - max: number | null; - sd: number | null; - }> - >( - ` + // no row transfer. For an hourly-mean type (pulse) the mean is the mean + // of the local days' values, each day the mean of its local hours' + // means, so it is the mean of the day points the chart draws; count, + // min, max and the spread stay over every reading. + const [rawAgg] = usesHourlyMeanDay(type) + ? await prisma.$queryRawUnsafe< + Array<{ + n: number; + mean: number | null; + min: number | null; + max: number | null; + sd: number | null; + }> + >( + ` + WITH src AS ( + SELECT m.* + FROM ${seriesRows} + WHERE m."measured_at" >= $2 + ) + SELECT + COUNT(*)::int AS n, + ${windowMeanSql({ typeColumn: 'm."type"', value: 'm."value"', weight: "m.day_weight" })}::double precision + AS mean, + MIN(m."value")::double precision AS min, + MAX(m."value")::double precision AS max, + STDDEV_POP(m."value")::double precision AS sd + FROM ${dayWeightedRows("src", zoneDayFrame("$3"))} m + GROUP BY m."type" + `, + user.id, + since, + userTz, + ) + : await prisma.$queryRawUnsafe< + Array<{ + n: number; + mean: number | null; + min: number | null; + max: number | null; + sd: number | null; + }> + >( + ` SELECT COUNT(*)::int AS n, AVG(m."value")::double precision AS mean, @@ -507,9 +546,9 @@ export const GET = apiHandler(async (request: NextRequest) => { FROM ${seriesRows} WHERE m."measured_at" >= $2 `, - user.id, - since, - ); + user.id, + since, + ); const round2 = (v: number) => Math.round(v * 100) / 100; const dayId = (d: Date) => grain === "day" @@ -687,6 +726,25 @@ export const GET = apiHandler(async (request: NextRequest) => { valueMax: row.valueMax ?? null, }; }); + // An hourly-mean type's strip mean is the mean of its local days' + // values, each day the mean of its local hours' means, so a dense + // workout hour does not outweigh the rest of the window. Count, min, + // max and the spread stay over every reading. + if (usesHourlyMeanDay(type) && rows.length > 0) { + const userTz = await resolveUserTimezone(user.id); + const values = rows.map((r) => r.value); + const rawSummary = summarize( + rows.map((r) => ({ date: r.measuredAt, value: r.value })), + ); + const mean = readingsMean(type, rows, userTz); + statsOverride = { + mean: mean === null ? 0 : Math.round(mean * 100) / 100, + min: rawSummary.min ?? 0, + max: rawSummary.max ?? 0, + stdDev: stdDev(values), + count: rawSummary.count, + }; + } } else { points = rows.map((r) => ({ id: r.id, diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 9425ee5cf..f6b58107c 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -45,11 +45,9 @@ vi.mock("@/lib/logging/context", () => ({ // 90-day FILTER caps the slice itself writes — is exercised without // coupling the slice unit test to the rank builder's enum-whitelist // internals. -vi.mock("@/lib/analytics/source-rank-sql", () => ({ - buildSourceRankCase: vi.fn(() => "90"), - canonicalMeasurementsFrom: vi.fn( - (_rank: string, sinceInterval?: string) => - `( +vi.mock("@/lib/analytics/source-rank-sql", () => { + const cte = (_rank: string, sinceInterval?: string) => + ` SELECT mm.* FROM measurements mm WHERE mm."user_id" = $1 @@ -58,13 +56,15 @@ vi.mock("@/lib/analytics/source-rank-sql", () => ({ sinceInterval ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'` : "" - } - ) m`, - ), -})); + }`; + return { + buildSourceRankCase: vi.fn(() => "90"), + canonicalMeasurementsCte: vi.fn(cte), + }; +}); import { prisma } from "@/lib/db"; -import { canonicalMeasurementsFrom } from "@/lib/analytics/source-rank-sql"; +import { canonicalMeasurementsCte } from "@/lib/analytics/source-rank-sql"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { computeSummariesSlice } from "../summaries-slice"; @@ -79,7 +79,7 @@ const ROLLUP_FIND_MANY = prisma.measurementRollup .findMany as unknown as ReturnType; const ROLLUP_FIND_FIRST = prisma.measurementRollup .findFirst as unknown as ReturnType; -const CANONICAL_FROM = canonicalMeasurementsFrom as unknown as ReturnType< +const CANONICAL_FROM = canonicalMeasurementsCte as unknown as ReturnType< typeof vi.fn >; @@ -609,11 +609,12 @@ describe("computeSummariesSlice", () => { expect(narrowsCap?.length).toBeGreaterThanOrEqual(1); // v1.11.1 — the cold-fallback `windowed` scan delegates its outer - // 90-day cap to `canonicalMeasurementsFrom(rank, "90 days")`. The + // 90-day cap to `canonicalMeasurementsCte(rank, "90 days")` (a CTE + // now, so the day weights can read it twice). The // helper lives in `@/lib/analytics/source-rank-sql` (stubbed // above), so pin the contract at the call boundary: the slice // must ask for the 90-day window. - expect(canonicalMeasurementsFrom).toHaveBeenCalledWith( + expect(canonicalMeasurementsCte).toHaveBeenCalledWith( expect.any(String), "90 days", ); @@ -621,7 +622,7 @@ describe("computeSummariesSlice", () => { // 15-year scan-DoS floor (far beyond any real history) so a // coverage-miss cold read can't trigger an unbounded full-partition // scan. It no longer calls the helper without an interval. - expect(canonicalMeasurementsFrom).toHaveBeenCalledWith( + expect(canonicalMeasurementsCte).toHaveBeenCalledWith( expect.any(String), "15 years", ); diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 76dc34e6d..0eb7ea343 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -60,7 +60,18 @@ import type { MeasurementType } from "@/generated/prisma/client"; import { Prisma } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; -import { windowWeighting } from "@/lib/measurements/day-statistic"; +import { + usesHourlyMeanDay, + windowWeighting, +} from "@/lib/measurements/day-statistic"; +import { + SESSION_DAY_FRAME, + dayWeightedRows, + dayWeightedRowsSql, + foldMeanSql, + isHourlyMeanTypeSql, + windowMeanSql, +} from "@/lib/measurements/day-mean"; import type { DataSummary } from "@/lib/analytics/trends"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { annotate } from "@/lib/logging/context"; @@ -80,7 +91,7 @@ import { import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { buildSourceRankCase, - canonicalMeasurementsFrom, + canonicalMeasurementsCte, } from "@/lib/analytics/source-rank-sql"; import { isFullyCovered, @@ -137,11 +148,33 @@ interface AllTimeAggregateRow { interface RollupAggregateRow { type: string; count: number; + /** Days that hold a reading: the weight of the mean for an hourly-mean type. */ + days: number; min: number; max: number; mean: number; } +/** + * The window means over canonical readings that carry a `day_weight` (see + * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse), + * the unchanged `AVG(value)` for every other type. Days are UTC days, the + * frame the canonical-source pick and the rollup tier bucket by. + */ +function windowedMean(filter: string): string { + return windowMeanSql({ + typeColumn: 'm."type"', + value: 'm."value"', + weight: "m.day_weight", + filter, + }); +} +const AVG7_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '7 days'`); +const AVG30_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '30 days'`); +const AVG30_LAST_MONTH_SQL = windowedMean( + `m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" < NOW() - INTERVAL '30 days'`, +); + interface NarrowAggregateRow { type: string; avg7: number | null; @@ -455,26 +488,23 @@ async function computeFromRollups(userId: string): Promise { // 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 window means weigh each day once for an hourly-mean type + // (`windowedMean`); the median stays over every reading. prisma.$queryRawUnsafe( ` + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} + ) SELECT m."type"::text AS type, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS avg7, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS avg30, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '60 days' - AND m."measured_at" < NOW() - INTERVAL '30 days' - )::double precision AS avg30_last_month, + ${AVG7_SQL}::double precision AS avg7, + ${AVG30_SQL}::double precision AS avg30, + ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month, PERCENTILE_CONT(0.5) WITHIN GROUP ( ORDER BY m."value" ) FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '90 days' )::double precision AS median - FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} + FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m GROUP BY m."type" `, userId, @@ -527,12 +557,19 @@ async function computeFromRollups(userId: string): Promise { SELECT "type"::text AS type, SUM("count")::int AS count, + COUNT(*)::int AS days, MIN("min_value")::double precision AS min, MAX("max_value")::double precision AS max, - ( - SUM("count" * "mean")::double precision - / NULLIF(SUM("count")::double precision, 0) - ) AS mean + -- An hourly-mean type (pulse): every day once, the DAY mean already + -- being the mean of its hours' means. Every other type: every + -- reading once. + (CASE WHEN ${isHourlyMeanTypeSql('"type"')} + THEN AVG("mean")::double precision + ELSE ( + SUM("count" * "mean")::double precision + / NULLIF(SUM("count")::double precision, 0) + ) + END) AS mean FROM collapsed GROUP BY "type" `, @@ -576,16 +613,27 @@ async function computeFromRollups(userId: string): Promise { // day first (a hash aggregate, a few rows per day however dense the // stream), each day then keeps its ladder-canonical source, so the // sort never sees more than one row per source per day. + // An hourly-mean type folds each source's day as the mean of its + // hours' means (`wsum` / `wdays`, see `day-mean.ts`) and the days into + // the remainder once each, the statistic the DAY rollup stores. prisma.$queryRaw< Array<{ type: string; count: number; + days: number; min: number; max: number; mean: number; }> >` - WITH per_source AS ( + WITH src AS ( + SELECT * + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."deleted_at" IS NULL + AND m."measured_at" < ${foldSplit} + ), + per_source AS ( SELECT m."type", m."source", @@ -593,25 +641,33 @@ async function computeFromRollups(userId: string): Promise { COUNT(*)::int AS cnt, SUM(m."value")::double precision AS total, MIN(m."value")::double precision AS min_value, - MAX(m."value")::double precision AS max_value - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."deleted_at" IS NULL - AND m."measured_at" < ${foldSplit} + MAX(m."value")::double precision AS max_value, + SUM(m."value" * m.day_weight)::double precision AS wsum, + SUM(m.day_weight)::double precision AS wdays + FROM ${dayWeightedRowsSql("src", null, { bySource: true })} m GROUP BY m."type", m."source", 3 ), canon AS ( SELECT DISTINCT ON (p."type", p.day) - p."type", p.cnt, p.total, p.min_value, p.max_value + p."type", p.cnt, p.total, p.min_value, p.max_value, p.wsum, p.wdays FROM per_source p ORDER BY p."type", p.day, (${Prisma.raw(rankPerSource)}), p."source"::text ) SELECT c."type"::text AS type, SUM(c.cnt)::int AS count, + COUNT(*)::int AS days, MIN(c.min_value)::double precision AS min, MAX(c.max_value)::double precision AS max, - (SUM(c.total) / SUM(c.cnt))::double precision AS mean + ${Prisma.raw( + foldMeanSql({ + typeColumn: 'c."type"', + total: "c.total", + count: "c.cnt", + weightedSum: "c.wsum", + weightSum: "c.wdays", + }), + )}::double precision AS mean FROM canon c GROUP BY c."type" `, @@ -619,11 +675,12 @@ async function computeFromRollups(userId: string): Promise { const preFoldByType = new Map< string, - { count: number; min: number; max: number; mean: number } + { count: number; days: number; min: number; max: number; mean: number } >(); for (const row of preFoldRemainder) { preFoldByType.set(row.type, { count: Number(row.count), + days: Number(row.days), min: Number(row.min), max: Number(row.max), mean: Number(row.mean), @@ -754,8 +811,13 @@ async function computeFromRollups(userId: string): Promise { const allCount = row.count + (pre?.count ?? 0); const allMin = pre ? Math.min(row.min, pre.min) : row.min; const allMax = pre ? Math.max(row.max, pre.max) : row.max; + // Both halves weigh their days once for an hourly-mean type (pulse), so + // they combine by day count; every other type combines by readings. const allMean = pre - ? (row.mean * row.count + pre.mean * pre.count) / allCount + ? usesHourlyMeanDay(row.type) + ? (row.mean * Number(row.days) + pre.mean * pre.days) / + (Number(row.days) + pre.days) + : (row.mean * row.count + pre.mean * pre.count) / allCount : row.mean; summaries[row.type] = { count: allCount, @@ -889,31 +951,29 @@ async function computeFromLiveAggregate( const [allTime, windowed, latests] = await Promise.all([ prisma.$queryRawUnsafe( ` + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)} + ) SELECT m."type"::text AS type, COUNT(*) AS count, MIN(m."value")::double precision AS min_value, MAX(m."value")::double precision AS max_value, - AVG(m."value")::double precision AS mean_value - FROM ${canonicalMeasurementsFrom(rankUnqualified, ALL_TIME_LIVE_CAP_INTERVAL)} + ${windowMeanSql({ typeColumn: 'm."type"', value: 'm."value"', weight: "m.day_weight" })}::double precision + AS mean_value + FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m GROUP BY m."type" `, userId, ), prisma.$queryRawUnsafe( ` + WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} + ) SELECT m."type"::text AS type, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS avg7, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS avg30, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '60 days' - AND m."measured_at" < NOW() - INTERVAL '30 days' - )::double precision AS avg30_last_month, + ${AVG7_SQL}::double precision AS avg7, + ${AVG30_SQL}::double precision AS avg30, + ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month, PERCENTILE_CONT(0.5) WITHIN GROUP ( ORDER BY m."value" ) FILTER ( @@ -956,7 +1016,7 @@ async function computeFromLiveAggregate( ) FILTER ( WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC' )::double precision AS r2_30 - FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} + FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m GROUP BY m."type" `, userId, diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index e828eb189..8d562abd0 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -114,6 +114,11 @@ import { type AccumulatorBucketRow, } from "@/lib/rollups/measurement-read"; import { windowWeighting } from "@/lib/measurements/day-statistic"; +import { + SESSION_DAY_FRAME, + dayWeightedRows, + windowMeanSql, +} from "@/lib/measurements/day-mean"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import { buildSourceRankCase, @@ -126,6 +131,26 @@ import { } from "@/lib/rollups/measurement-coverage"; import { dateOnlyKey } from "@/lib/tz/date-only"; +/** + * Window means over canonical readings carrying a `day_weight` (see + * `day-mean.ts`): the mean of the day values for an hourly-mean type (pulse), + * the unchanged `AVG(value)` otherwise. Days are UTC days, the frame the + * canonical-source pick buckets by. + */ +function windowedMean(filter: string): string { + return windowMeanSql({ + typeColumn: 'm."type"', + value: 'm."value"', + weight: "m.day_weight", + filter, + }); +} +const AVG7_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '7 days'`); +const AVG30_SQL = windowedMean(`m."measured_at" >= NOW() - INTERVAL '30 days'`); +const AVG30_LAST_MONTH_SQL = windowedMean( + `m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" < NOW() - INTERVAL '30 days'`, +); + /** * Heavy aggregate row — count/min/max/mean alongside the non-composable * windowed columns. Used on the cold-mount fallback path where the @@ -419,6 +444,9 @@ async function buildFromRollups( -- instead of twice per rebuild. Output is identical. WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} ), + -- The z-score's mean and spread stay over every reading: they measure + -- how far one reading sits from the others, a per-reading dispersion, + -- not the level the window means below report. window_stats AS ( SELECT m."type", @@ -440,18 +468,11 @@ async function buildFromRollups( (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2)) ) > 2 ) AS anomaly_count, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS avg7, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS avg30, + ${AVG7_SQL}::double precision AS avg7, + ${AVG30_SQL}::double precision AS avg30, -- Prior 30d window (days [30, 60) ago) used by tile delta callout. - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '60 days' - AND m."measured_at" < NOW() - INTERVAL '30 days' - )::double precision AS avg30_last_month - FROM cm m + ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month + FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m JOIN window_stats ws ON ws."type" = m."type" GROUP BY m."type", ws.stddev_value `, @@ -662,6 +683,9 @@ async function buildFromLiveAggregate( -- so the DISTINCT-ON pick runs once. Output is identical. WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")} ), + -- The z-score's mean and spread stay over every reading: they measure + -- how far one reading sits from the others, a per-reading dispersion, + -- not the level the window means below report. window_stats AS ( SELECT m."type", @@ -675,7 +699,8 @@ async function buildFromLiveAggregate( COUNT(*) AS count, MIN(m."value")::double precision AS min_value, MAX(m."value")::double precision AS max_value, - AVG(m."value")::double precision AS mean_value, + ${windowMeanSql({ typeColumn: 'm."type"', value: 'm."value"', weight: "m.day_weight" })}::double precision + AS mean_value, ws.stddev_value::double precision AS stddev_value, COUNT(*) FILTER ( WHERE ws.stddev_value IS NOT NULL @@ -684,16 +709,9 @@ async function buildFromLiveAggregate( (ROUND(((m."value" - ws.mean_value) / ws.stddev_value)::numeric, 2)) ) > 2 ) AS anomaly_count, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' - )::double precision AS avg7, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' - )::double precision AS avg30, - AVG(m."value") FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '60 days' - AND m."measured_at" < NOW() - INTERVAL '30 days' - )::double precision AS avg30_last_month, + ${AVG7_SQL}::double precision AS avg7, + ${AVG30_SQL}::double precision AS avg30, + ${AVG30_LAST_MONTH_SQL}::double precision AS avg30_last_month, -- A2-M2 — the regression windows anchor on the UTC-midnight day -- boundary, NOT the wall-clock instant, so this cold-fallback path -- returns the SAME boundary-day membership the warm rollup path @@ -731,7 +749,7 @@ async function buildFromLiveAggregate( ) FILTER ( WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC' )::double precision AS r2_30 - FROM cm m + FROM ${dayWeightedRows("cm", SESSION_DAY_FRAME)} m JOIN window_stats ws ON ws."type" = m."type" GROUP BY m."type", ws.stddev_value `, diff --git a/tests/integration/pulse-day-mean-zones.test.ts b/tests/integration/pulse-day-mean-zones.test.ts new file mode 100644 index 000000000..da0bdc286 --- /dev/null +++ b/tests/integration/pulse-day-mean-zones.test.ts @@ -0,0 +1,221 @@ +/** + * A day of pulse is the mean of its LOCAL hours' means, and the days are LOCAL + * days, for a profile zone that is not UTC. + * + * Two zones that each break a UTC assumption: + * + * - Asia/Kolkata (UTC+05:30). Its hours start at half past a UTC hour, so + * 10:20Z, 10:25Z (both 15:5x local) and 10:40Z (16:10 local) are two local + * hours (100 and 40, day value 70), though one UTC hour (80). 18:45Z is + * already 00:15 the next local day. + * - Europe/Berlin. 23:30Z is the next local day, so 10:00Z, 11:00Z (three + * readings) and 23:30Z are two local days (100 and 40), though one UTC day + * (80). + * + * Asserted through the live per-day read the chart serves, the series route's + * day bucket, and the shared SQL day weight every other live reader uses. + */ +import { NextRequest } from "next/server"; +import { beforeAll, describe, expect, it, vi } from "vitest"; + +import { cookieJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; +import type { Prisma } from "@/generated/prisma/client"; +import { readLiveBuckets } from "@/lib/measurements/daily-series-read"; +import { + dayWeightedRows, + dayWeightedRowsSql, + readingsMean, + windowMeanSql, + zoneDayFrame, +} from "@/lib/measurements/day-mean"; +import { invalidateUserTimezone } from "@/lib/tz/resolver"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => cookieJar.set(name, value), + delete: (name: string) => cookieJar.delete(name), + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const prisma = getPrismaClient(); +const DAY = 86_400_000; +const now = new Date(); +/** A UTC midnight three weeks back: inside every window the reads use. */ +const D = Date.UTC( + now.getUTCFullYear(), + now.getUTCMonth(), + now.getUTCDate() - 21, +); +const iso = (ms: number) => new Date(ms).toISOString(); +const hm = (h: number, m: number) => D + h * 3_600_000 + m * 60_000; + +const KOLKATA = { + user: "pulse-zone-kolkata", + tz: "Asia/Kolkata", + readings: [ + [hm(10, 20), 100], + [hm(10, 25), 100], + [hm(10, 40), 40], + [hm(18, 45), 90], + ] as const, + /** Local days: 70 (two local hours) and 90. */ + days: [70, 90], + /** What a UTC frame would give: one day, hours 80 and 90. */ + utcDay: 85, +}; + +const BERLIN = { + user: "pulse-zone-berlin", + tz: "Europe/Berlin", + readings: [ + [hm(10, 0), 100], + [hm(11, 0), 100], + [hm(11, 10), 100], + [hm(11, 20), 100], + [hm(23, 30), 40], + ] as const, + /** Local days: 100 and 40. */ + days: [100, 40], + /** What a UTC frame would give: one day, hours 100, 100 and 40. */ + utcDay: 80, +}; + +const CASES = [KOLKATA, BERLIN]; + +beforeAll(async () => { + await truncateAllTables(prisma); + for (const c of CASES) { + await prisma.user.create({ + data: { id: c.user, username: c.user, timezone: c.tz }, + }); + invalidateUserTimezone(c.user); + const rows: Prisma.MeasurementCreateManyInput[] = c.readings.map( + ([at, value], i) => ({ + id: `${c.user}-${i}`, + userId: c.user, + type: "PULSE", + unit: "bpm", + source: "APPLE_HEALTH", + value, + measuredAt: new Date(at), + }), + ); + await prisma.measurement.createMany({ data: rows }); + } +}, 120_000); + +describe.each(CASES)("profile zone $tz", (c) => { + it("the TypeScript helper reads local hours and local days", () => { + const rows = c.readings.map(([at, value]) => ({ + value, + measuredAt: new Date(at), + })); + expect(readingsMean("PULSE", rows, c.tz)).toBeCloseTo( + (c.days[0] + c.days[1]) / 2, + 9, + ); + expect(readingsMean("PULSE", rows, "UTC")).toBeCloseTo(c.utcDay, 9); + }); + + it("the live per-day read buckets by local hour and local day", async () => { + const rows = await readLiveBuckets({ + userId: c.user, + type: "PULSE", + from: new Date(D - DAY), + to: new Date(D + 2 * DAY), + cap: 400, + priorityJson: null, + grain: "daily", + timeZone: c.tz, + }); + expect(rows.map((r) => r.value)).toEqual( + c.days.map((v) => expect.closeTo(v, 6)), + ); + // Count and band stay over every reading. + expect(rows.reduce((s, r) => s + (r.count ?? 0), 0)).toBe( + c.readings.length, + ); + }); + + it("the shared SQL day weight gives the mean of the local days", async () => { + const [row] = await prisma.$queryRawUnsafe>( + ` + WITH src AS ( + SELECT * FROM measurements + WHERE "user_id" = $1 AND "deleted_at" IS NULL + ) + SELECT ${windowMeanSql({ + typeColumn: 'm."type"', + value: 'm."value"', + weight: "m.day_weight", + })}::double precision AS mean + FROM ${dayWeightedRows("src", zoneDayFrame("$2"))} m + GROUP BY m."type" + `, + c.user, + c.tz, + ); + expect(row.mean).toBeCloseTo((c.days[0] + c.days[1]) / 2, 9); + + const [tagged] = await prisma.$queryRaw>` + WITH src AS ( + SELECT * FROM measurements + WHERE "user_id" = ${c.user} AND "deleted_at" IS NULL + ) + SELECT SUM(m.day_weight)::double precision AS days + FROM ${dayWeightedRowsSql("src", c.tz)} m + `; + // Each local day's weights add up to one. + expect(tagged.days).toBeCloseTo(2, 9); + }); + + it("the series route's day bucket is the local day of local hours", async () => { + const s = await prisma.session.create({ + data: { userId: c.user, expiresAt: new Date(Date.now() + 3_600_000) }, + }); + cookieJar.clear(); + cookieJar.set("healthlog_session", s.id); + const { GET } = await import("@/app/api/measurements/series/route"); + const res = await GET( + new NextRequest( + "http://localhost/api/measurements/series?kind=pulse&days=365", + ), + ); + expect(res.status).toBe(200); + const body = (await res.json()) as { + data: { + points: Array<{ id: string; value: number }>; + stats: { mean: number; count: number }; + }; + }; + expect(body.data.points.every((p) => p.id.startsWith("day:"))).toBe(true); + expect(body.data.points.map((p) => p.value)).toEqual( + c.days.map((v) => expect.closeTo(v, 2)), + ); + // The stats strip: the mean of the day values, the count of readings. + expect(body.data.stats.mean).toBeCloseTo((c.days[0] + c.days[1]) / 2, 2); + expect(body.data.stats.count).toBe(c.readings.length); + }); +}); + +it("the fixture really separates the frames", () => { + expect(iso(hm(18, 45))).toMatch(/T18:45/); + for (const c of CASES) { + expect((c.days[0] + c.days[1]) / 2).not.toBeCloseTo(c.utcDay, 2); + } +}); diff --git a/tests/integration/pulse-window-mean-readers.test.ts b/tests/integration/pulse-window-mean-readers.test.ts new file mode 100644 index 000000000..086adfef8 --- /dev/null +++ b/tests/integration/pulse-window-mean-readers.test.ts @@ -0,0 +1,209 @@ +/** + * The dashboard's and the insights' window means of pulse weigh each day once, + * each day the mean of its hours' means; heart-rate variability, seeded with + * the very same readings, keeps the plain mean over its readings. + * + * A workout day: twelve readings at 150 bpm inside one hour, then three + * resting hours of one reading at 60 bpm. Its day value is (150 + 3 x 60) / 4 + * = 82.5, its readings' mean (12 x 150 + 3 x 60) / 15 = 132. A quiet day: one + * reading at 60. + * + * last 30 days workout day + quiet day 71.25 (plain 127.5) + * days 30 to 60 workout day + quiet day 71.25 (plain 127.5) + * six years ago workout day alone 82.5 (plain 132) + * all time five days 73.5 (plain 6060 / 47) + * + * Asserted on both paths of the dashboard summaries slice (the rollup tier, + * with the six-year-old day in the pre-fold remainder, and the live + * fallback), both paths of the insights aggregate, and the stats strip of the + * series route's raw window. + */ +import { NextRequest } from "next/server"; +import { beforeAll, describe, expect, it, vi } from "vitest"; + +import { cookieJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; +import type { Prisma } from "@/generated/prisma/client"; +import { computeSummariesSlice } from "@/lib/analytics/summaries-slice"; +import { buildComprehensiveAggregate } from "@/lib/insights/comprehensive-aggregator"; +import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { invalidateUserTimezone } from "@/lib/tz/resolver"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => cookieJar.set(name, value), + delete: (name: string) => cookieJar.delete(name), + })), + }; +}); + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +// The reads kick off a background rollup refresh; keep it off so the live +// user stays on the live path for the whole file. +vi.mock("@/lib/rollups/measurement-rollups", async (importOriginal) => { + const actual = + await importOriginal(); + return { ...actual, ensureUserRollupsFresh: vi.fn(async () => undefined) }; +}); + +const prisma = getPrismaClient(); +const DAY = 86_400_000; +const now = new Date(); +const T0 = Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()); + +const LIVE = "pulse-window-live"; +const ROLLUP = "pulse-window-rollup"; + +const WORKOUT_DAYS = [T0 - 3 * DAY, T0 - 45 * DAY, T0 - 6 * 365 * DAY]; +const QUIET_DAYS = [T0 - 2 * DAY, T0 - 44 * DAY]; + +const HOURLY = { + window: (82.5 + 60) / 2, + allTime: (82.5 * 3 + 60 * 2) / 5, +}; +const PLAIN = { + window: (12 * 150 + 3 * 60 + 60) / 16, + allTime: (3 * (12 * 150 + 3 * 60) + 2 * 60) / 47, +}; + +function rows( + userId: string, + type: "PULSE" | "HEART_RATE_VARIABILITY", +): Prisma.MeasurementCreateManyInput[] { + const unit = type === "PULSE" ? "bpm" : "ms"; + const out: Prisma.MeasurementCreateManyInput[] = []; + const push = (at: number, value: number) => + out.push({ + id: `${userId}-${type}-${at}`, + userId, + type, + unit, + source: "APPLE_HEALTH", + value, + measuredAt: new Date(at), + }); + for (const d of WORKOUT_DAYS) { + for (let i = 0; i < 12; i += 1) push(d + 10 * 3_600_000 + i * 300_000, 150); + for (const h of [12, 14, 16]) push(d + h * 3_600_000, 60); + } + for (const d of QUIET_DAYS) push(d + 8 * 3_600_000, 60); + return out; +} + +beforeAll(async () => { + await truncateAllTables(prisma); + for (const user of [LIVE, ROLLUP]) { + await prisma.user.create({ + data: { id: user, username: user, timezone: "UTC" }, + }); + invalidateUserTimezone(user); + await prisma.measurement.createMany({ + data: [...rows(user, "PULSE"), ...rows(user, "HEART_RATE_VARIABILITY")], + }); + } + // The rollup user's DAY tier covers the last 400 days; the six-year-old + // workout day stays outside it, in the pre-fold remainder. + await recomputeUserRollups(ROLLUP, { + types: ["PULSE", "HEART_RATE_VARIABILITY"], + from: new Date(T0 - 400 * DAY), + to: new Date(T0 + DAY), + }); +}, 120_000); + +describe.each([ + { path: "rollup", user: ROLLUP }, + { path: "live", user: LIVE }, +])("dashboard summaries slice, $path path", ({ user }) => { + it("weighs each day once for pulse", async () => { + const { summaries } = await computeSummariesSlice(user); + const pulse = summaries.PULSE; + expect(pulse.avg7).toBeCloseTo(HOURLY.window, 2); + expect(pulse.avg30).toBeCloseTo(HOURLY.window, 2); + expect(pulse.avg30LastMonth).toBeCloseTo(HOURLY.window, 2); + expect(pulse.mean).toBeCloseTo(HOURLY.allTime, 2); + // Count, min and max stay over every reading. + expect(pulse.count).toBe(47); + expect(pulse.min).toBe(60); + expect(pulse.max).toBe(150); + }); + + it("keeps the plain mean for HRV", async () => { + const { summaries } = await computeSummariesSlice(user); + const hrv = summaries.HEART_RATE_VARIABILITY; + expect(hrv.avg7).toBeCloseTo(PLAIN.window, 2); + expect(hrv.avg30).toBeCloseTo(PLAIN.window, 2); + expect(hrv.avg30LastMonth).toBeCloseTo(PLAIN.window, 2); + expect(hrv.mean).toBeCloseTo(PLAIN.allTime, 2); + }); +}); + +describe.each([ + { path: "rollup", user: ROLLUP }, + { path: "live", user: LIVE }, +])("insights aggregate, $path path", ({ user }) => { + it("weighs each day once for pulse and keeps the plain mean for HRV", async () => { + const { summaries } = await buildComprehensiveAggregate(user); + expect(summaries.PULSE.avg7).toBeCloseTo(HOURLY.window, 2); + expect(summaries.PULSE.avg30).toBeCloseTo(HOURLY.window, 2); + expect(summaries.PULSE.avg30LastMonth).toBeCloseTo(HOURLY.window, 2); + // The 90-day window holds two workout days and two quiet days. + expect(summaries.PULSE.mean).toBeCloseTo(HOURLY.window, 2); + expect(summaries.PULSE.count).toBe(32); + const hrv = summaries.HEART_RATE_VARIABILITY; + expect(hrv.avg7).toBeCloseTo(PLAIN.window, 2); + expect(hrv.avg30LastMonth).toBeCloseTo(PLAIN.window, 2); + expect(hrv.mean).toBeCloseTo(PLAIN.window, 2); + }); +}); + +describe("series route, raw window", () => { + async function stats(kind: string) { + const s = await prisma.session.create({ + data: { userId: LIVE, expiresAt: new Date(Date.now() + 3_600_000) }, + }); + cookieJar.clear(); + cookieJar.set("healthlog_session", s.id); + const { GET } = await import("@/app/api/measurements/series/route"); + const res = await GET( + new NextRequest( + `http://localhost/api/measurements/series?kind=${kind}&days=30`, + ), + ); + expect(res.status).toBe(200); + return ( + (await res.json()) as { + data: { + points: unknown[]; + stats: { mean: number; count: number; min: number; max: number }; + }; + } + ).data; + } + + it("gives pulse the mean of its days on the stats strip", async () => { + const data = await stats("pulse"); + // Raw points: every reading of the last 30 days. + expect(data.points).toHaveLength(16); + expect(data.stats.mean).toBeCloseTo(HOURLY.window, 2); + expect(data.stats.count).toBe(16); + expect(data.stats.min).toBe(60); + expect(data.stats.max).toBe(150); + }); + + it("keeps the plain mean for HRV", async () => { + const data = await stats("heartRateVariability"); + expect(data.stats.mean).toBeCloseTo(PLAIN.window, 2); + }); +});