mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 04:50:59 +00:00
fix(insights): fold dense measurement reads per day in SQL
A single account syncing heart rate from a watch took the worker down within half an hour. The status cards' graded series read the last 91 days of raw rows, bounded in time but not in rows, so a stream sampled every minute or faster materialised six or seven figures of objects on every refresh and the read ran into the statement timeout. Add readDayAggregates, which folds one type into per-day sum, count, min and max in Postgres (optionally split into age segments and filtered to the metric's plausibility domain), and move the readers that folded raw rows into days in memory onto it: the graded series (recent, weekly and the cold-tier fallback, for every type), the vitals day-mean fallback behind the personal band, the tiered context's recent days and the plan review. Recent, weekly and monthly buckets are identical to the raw fold; the cold-tier yearly slope now runs over day means. On 91 days of per-minute pulse the graded series now allocates 1.6 MB where it allocated 62 MB, while the raw read alone holds 22 MB. A structural guard freezes the remaining untaken measurement reads in the worker and insight directories: each must name a sparse literal type or carry a listed reason. Refs #1023
This commit is contained in:
Marc-André Bombeck committed
1 parent
88cea09c7b
commit
5a0b6ee153
26 files changed
+1083
-287
No files matched your search
@@ -0,0 +1,177 @@
|
||||
/**
|
||||
* Structural guard: background and insight readers must not materialise a
|
||||
* dense measurement stream.
|
||||
*
|
||||
* #1023: a single account syncing heart rate from a watch took the worker
|
||||
* down within half an hour. The status-card read was bounded in time
|
||||
* (`measuredAt >= now − 91 days`) but not in rows, and a stream sampled every
|
||||
* minute or faster turns 91 days into six or seven figures of objects. The
|
||||
* read also ran into the statement timeout. The cure was to fold per day in
|
||||
* SQL (`readDayAggregates`, `readLiveBuckets`), so the reader's size tracks
|
||||
* the number of days, not the sampling rate.
|
||||
*
|
||||
* What this guard pins, in the directories that run on the worker or feed an
|
||||
* insight: every `measurement.findMany` without a `take` reads a type that is
|
||||
* named as a literal and is not one a device streams. A read whose type is a
|
||||
* variable could be handed a dense type by the next caller, so it must be
|
||||
* listed below with the reason it is safe. The list is frozen both ways: a new
|
||||
* unlisted read fails, and an entry whose read is gone fails too.
|
||||
*
|
||||
* It is a tripwire, not a proof: the matcher reads source text, and a read
|
||||
* written some other way (a raw SELECT of rows, a helper in another
|
||||
* directory) passes it. The floor below fails the guard if the matcher stops
|
||||
* finding reads at all.
|
||||
*/
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { walkSourceFiles } from "./helpers/source-files";
|
||||
import {
|
||||
CUMULATIVE_HK_TYPES,
|
||||
HIGH_FREQUENCY_MEAN_TYPES,
|
||||
} from "@/lib/measurements/apple-health-mapping";
|
||||
|
||||
const SRC = join(process.cwd(), "src");
|
||||
|
||||
const SCOPED_DIRS = [
|
||||
"lib/insights/",
|
||||
"lib/jobs/",
|
||||
"lib/rollups/",
|
||||
"lib/ai/",
|
||||
"lib/analytics/",
|
||||
"lib/dashboard/",
|
||||
"lib/daily/",
|
||||
];
|
||||
|
||||
/** Types a device writes at sampling rate rather than a few times a day. */
|
||||
const DENSE_TYPES: ReadonlySet<string> = new Set([
|
||||
"PULSE",
|
||||
"BLOOD_GLUCOSE",
|
||||
"HEART_RATE_VARIABILITY",
|
||||
"OXYGEN_SATURATION",
|
||||
...CUMULATIVE_HK_TYPES,
|
||||
...HIGH_FREQUENCY_MEAN_TYPES,
|
||||
]);
|
||||
|
||||
/**
|
||||
* Untaken reads whose type is not a sparse literal, keyed
|
||||
* `<file>::<type expression>`, with why each stays bounded.
|
||||
*/
|
||||
const ALLOWED: Record<string, string> = {
|
||||
"lib/insights/comprehensive-generate.ts::(shorthand)":
|
||||
'Only reached for SLEEP_DURATION (guarded by `type === "SLEEP_DURATION"`); a few stage rows per night.',
|
||||
"lib/insights/derived/vascular-age.ts::VASCULAR_AGE_TYPE":
|
||||
"A constant for a type written at most a few times a day.",
|
||||
"lib/insights/derived/fitness-age.ts::VO2_MAX_TYPE":
|
||||
"A constant for VO2 max, written at most daily.",
|
||||
"lib/insights/derived/six-minute-walk.ts::SIX_MINUTE_WALK_TYPE":
|
||||
"A constant for six-minute-walk distance, written at most daily.",
|
||||
"lib/insights/derived/wellness-scores.ts::measurementType":
|
||||
"Daily computed / provider scores (recovery, strain, readiness), one row per day.",
|
||||
"lib/jobs/reaction-line.ts::{ in: [...types]":
|
||||
"Exact-instant match (`measuredAt: row.occurredAt`), not a window.",
|
||||
"lib/jobs/step-consolidation-repair.ts::STEP_TYPE":
|
||||
"Repair job over one user's step rows for a bounded day range; runs once per repair, not per status refresh.",
|
||||
'lib/ai/coach/snapshot.ts::"BLOOD_GLUCOSE" as never':
|
||||
"Clinical CGM metrics (time in range, GMI, CV) need individual readings; 30 days at a sensor's fixed rate (at most one a minute) stays below 45 000 rows.",
|
||||
'lib/dashboard/snapshot.ts::"BLOOD_GLUCOSE"':
|
||||
"Same 30-day clinical glucose panel as the coach snapshot, grouped by meal context.",
|
||||
"lib/ai/coach/cycle-snapshot.ts::{ in: PHASE_CROSSTAB_METRIC_TYPES":
|
||||
"Cycle-phase crosstab over 365 days. Its dense members (steps, CGM glucose) are the known remaining case; steps are drained to one row per day nightly, CGM is not. Listed so it cannot grow unseen.",
|
||||
"lib/analytics/score/reader.ts::{ in: types":
|
||||
"Health-score inputs: steps (drained to one row per day nightly), sleep, waist, weight, blood pressure, fasting glucose only.",
|
||||
};
|
||||
|
||||
interface UntakenRead {
|
||||
file: string;
|
||||
line: number;
|
||||
typeExpr: string;
|
||||
}
|
||||
|
||||
function scopedFiles(): string[] {
|
||||
return walkSourceFiles(SRC, { floor: 3000 })
|
||||
.filter((p) => SCOPED_DIRS.some((d) => p.startsWith(d)))
|
||||
.filter((p) => !p.includes("__tests__"))
|
||||
.filter((p) => !p.endsWith(".test.ts") && !p.endsWith(".test.tsx"))
|
||||
.sort();
|
||||
}
|
||||
|
||||
function untakenReads(): { all: number; untaken: UntakenRead[] } {
|
||||
let all = 0;
|
||||
const untaken: UntakenRead[] = [];
|
||||
for (const file of scopedFiles()) {
|
||||
const src = readFileSync(join(SRC, file), "utf8");
|
||||
// Whitespace-tolerant: a call split across lines still matches.
|
||||
const re = /\.measurement\s*\.\s*findMany\s*\(/g;
|
||||
for (let m = re.exec(src); m; m = re.exec(src)) {
|
||||
all += 1;
|
||||
let i = m.index + m[0].length;
|
||||
let depth = 1;
|
||||
while (depth > 0 && i < src.length) {
|
||||
if (src[i] === "(") depth += 1;
|
||||
else if (src[i] === ")") depth -= 1;
|
||||
i += 1;
|
||||
}
|
||||
const call = src.slice(m.index, i);
|
||||
if (/\btake\b/.test(call)) continue;
|
||||
const typed = /\btype\s*:\s*([^,\n}]+)/.exec(call);
|
||||
const typeExpr = typed
|
||||
? typed[1].trim()
|
||||
: /\btype\s*,/.test(call)
|
||||
? "(shorthand)"
|
||||
: "(none)";
|
||||
untaken.push({
|
||||
file,
|
||||
line: src.slice(0, m.index).split("\n").length,
|
||||
typeExpr,
|
||||
});
|
||||
}
|
||||
}
|
||||
return { all, untaken };
|
||||
}
|
||||
|
||||
function sparseLiteral(typeExpr: string): boolean {
|
||||
const lit = /^"([A-Z_0-9]+)"/.exec(typeExpr);
|
||||
return lit !== null && !DENSE_TYPES.has(lit[1]);
|
||||
}
|
||||
|
||||
describe("dense measurement reads stay bounded (#1023)", () => {
|
||||
const { all, untaken } = untakenReads();
|
||||
|
||||
it("finds the reads it is meant to police", () => {
|
||||
// Guard against a matcher that silently matches nothing.
|
||||
expect(all).toBeGreaterThanOrEqual(40);
|
||||
expect(untaken.length).toBeGreaterThanOrEqual(15);
|
||||
});
|
||||
|
||||
it("every untaken read names a sparse literal type or is listed with its reason", () => {
|
||||
const offenders = untaken
|
||||
.filter((r) => !sparseLiteral(r.typeExpr))
|
||||
.filter((r) => !(`${r.file}::${r.typeExpr}` in ALLOWED))
|
||||
.map((r) => `${r.file}:${r.line} type ${r.typeExpr}`);
|
||||
expect(offenders).toEqual([]);
|
||||
});
|
||||
|
||||
it("every listed exception still exists", () => {
|
||||
const present = new Set(untaken.map((r) => `${r.file}::${r.typeExpr}`));
|
||||
expect(Object.keys(ALLOWED).filter((k) => !present.has(k))).toEqual([]);
|
||||
});
|
||||
|
||||
it("the readers fixed for #1023 fold in SQL rather than reading rows", () => {
|
||||
for (const file of [
|
||||
"lib/insights/graded-series.ts",
|
||||
"lib/rollups/tiered-context.ts",
|
||||
"lib/jobs/coach-plan-review.ts",
|
||||
"lib/measurements/daily-series-read.ts",
|
||||
]) {
|
||||
const src = readFileSync(join(SRC, file), "utf8");
|
||||
expect(src, file).not.toMatch(/\.measurement\s*\.\s*findMany\s*\(/);
|
||||
}
|
||||
const baseline = readFileSync(
|
||||
join(SRC, "lib/insights/derived/baseline.ts"),
|
||||
"utf8",
|
||||
);
|
||||
expect(baseline).not.toMatch(/\.measurement\s*\.\s*findMany\s*\(/);
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,14 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
import { NextRequest } from "next/server";
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique: vi.fn() },
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
|
||||
// The graded series folds per-day aggregates in SQL; the fake folds the
|
||||
// mocked `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique: vi.fn() },
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
|
||||
// The graded series folds per-day aggregates in SQL; the fake folds the
|
||||
// mocked `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique: vi.fn() },
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const findMany = vi.fn();
|
||||
const queryRaw = vi.fn();
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
measurement: {
|
||||
@@ -10,12 +9,14 @@ vi.mock("@/lib/db", () => ({
|
||||
// loadUserSourcePriority lazy-loads the source-priority blob; null
|
||||
// (default findUnique) falls back to the default ladders.
|
||||
user: { findUnique: vi.fn() },
|
||||
// v1.28.25 — dense types (BLOOD_GLUCOSE / PULSE) day-bucket the
|
||||
// cold-tier fallback in SQL.
|
||||
$queryRaw: (...args: unknown[]) => queryRaw(...args),
|
||||
},
|
||||
}));
|
||||
|
||||
const readDayAggregates = vi.fn();
|
||||
vi.mock("@/lib/measurements/day-aggregates", () => ({
|
||||
readDayAggregates: (...args: unknown[]) => readDayAggregates(...args),
|
||||
}));
|
||||
|
||||
const readBestGranularityRollups = vi.fn();
|
||||
const rollupRow = (bucketStart: Date, mean: number) => ({
|
||||
bucketStart,
|
||||
@@ -31,147 +32,160 @@ const rollupRow = (bucketStart: Date, mean: number) => ({
|
||||
vi.mock("@/lib/rollups/measurement-read-wmy", () => ({
|
||||
readBestGranularityRollups: (...args: unknown[]) =>
|
||||
readBestGranularityRollups(...args),
|
||||
// aggregateWmyBuckets is no longer imported by graded-series, but keep the
|
||||
// module shape intact for any transitive consumer.
|
||||
aggregateWmyBuckets: vi.fn(),
|
||||
}));
|
||||
|
||||
import { buildGradedSeriesWithRollups } from "../graded-series";
|
||||
import {
|
||||
buildGradedSeriesFromDayAggregates,
|
||||
buildGradedSeriesFromPoints,
|
||||
buildGradedSeriesWithRollups,
|
||||
} from "../graded-series";
|
||||
import { foldDayAggregates } from "@/lib/measurements/__tests__/fake-day-aggregates";
|
||||
import type { ReadDayAggregatesOptions } from "@/lib/measurements/day-aggregates";
|
||||
|
||||
const dayMs = 24 * 60 * 60 * 1000;
|
||||
|
||||
/** Daily readings spanning `days` back from `now`. */
|
||||
function dailyRows(days: number, now: Date) {
|
||||
/** Readings spanning `days` back from `now`, `perDay` per day. */
|
||||
function points(days: number, now: Date, perDay = 1) {
|
||||
const rows: Array<{ measuredAt: Date; value: number }> = [];
|
||||
for (let i = 0; i < days; i++) {
|
||||
rows.push({ measuredAt: new Date(now.getTime() - i * dayMs), value: 80 });
|
||||
for (let k = 0; k < perDay; k++) {
|
||||
rows.push({
|
||||
measuredAt: new Date(now.getTime() - i * dayMs - k * 3_600_000),
|
||||
value: 60 + ((i * 7 + k * 3) % 41),
|
||||
});
|
||||
}
|
||||
}
|
||||
return rows;
|
||||
return rows.reverse();
|
||||
}
|
||||
|
||||
/** Serve `readDayAggregates` from an in-memory row set. */
|
||||
function serveFrom(rows: Array<{ measuredAt: Date; value: number }>) {
|
||||
readDayAggregates.mockImplementation(async (opts: ReadDayAggregatesOptions) =>
|
||||
foldDayAggregates(rows, opts),
|
||||
);
|
||||
}
|
||||
|
||||
afterEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("buildGradedSeriesWithRollups — rollup-miss fallback", () => {
|
||||
describe("buildGradedSeriesWithRollups — bounded reads (#1023)", () => {
|
||||
const now = new Date("2026-05-31T12:00:00Z");
|
||||
|
||||
it("never materialises raw rows: the recent read is a per-day aggregate", async () => {
|
||||
serveFrom(points(80, now, 24));
|
||||
readBestGranularityRollups.mockResolvedValue({
|
||||
granularity: "MONTH",
|
||||
rows: [rollupRow(new Date("2025-09-01T00:00:00Z"), 79)],
|
||||
});
|
||||
|
||||
await buildGradedSeriesWithRollups("u1", "PULSE", now);
|
||||
|
||||
expect(findMany).not.toHaveBeenCalled();
|
||||
const opts = readDayAggregates.mock.calls[0][0] as ReadDayAggregatesOptions;
|
||||
expect(opts).toMatchObject({ userId: "u1", type: "PULSE" });
|
||||
expect(opts.until).toEqual(now);
|
||||
expect(opts.timeZone).toBe("Europe/Berlin");
|
||||
expect(now.getTime() - opts.since.getTime()).toBe(91 * dayMs);
|
||||
// The same age partition the parity test below proves exact: a reading
|
||||
// exactly 21 / 91 / 451 days old belongs to the older slice.
|
||||
expect(opts.segmentStarts).toEqual(
|
||||
[21, 91, 451].map((d) => new Date(now.getTime() - d * dayMs + 1)),
|
||||
);
|
||||
});
|
||||
|
||||
it("folds monthly/yearly from the bounded fallback when the tier has no coverage", async () => {
|
||||
// The bounded recent read (~90 d) returns only recent points; the
|
||||
// fallback read returns 2+ years so the fold has monthly/yearly.
|
||||
findMany
|
||||
// 1st call — bounded recent read (gte: since)
|
||||
.mockResolvedValueOnce(dailyRows(80, now))
|
||||
// 2nd call — bounded fallback read (gte: 1095-day horizon)
|
||||
.mockResolvedValueOnce(dailyRows(800, now));
|
||||
|
||||
// Tier miss: no MONTH / YEAR coverage at all.
|
||||
serveFrom(points(800, now));
|
||||
readBestGranularityRollups.mockResolvedValue(null);
|
||||
|
||||
const series = await buildGradedSeriesWithRollups("u1", "WEIGHT", now);
|
||||
|
||||
// The bug: before the fix monthly/yearly were derived from the
|
||||
// bounded ~90-day read and so were always empty even when years of
|
||||
// raw history existed. After the fix they fold from the fallback.
|
||||
expect(series.monthly.length).toBeGreaterThan(0);
|
||||
// 800 days of history puts at least one year into the yearly tail.
|
||||
expect(series.yearly.length).toBeGreaterThan(0);
|
||||
// The fallback read must have been issued...
|
||||
expect(findMany).toHaveBeenCalledTimes(2);
|
||||
// ...and bounded (v1.28.25): it must carry a `measuredAt.gte` floor
|
||||
// ~1095 days back, never an unwindowed full-history walk.
|
||||
const fallbackArgs = findMany.mock.calls[1][0] as {
|
||||
where: { measuredAt?: { gte?: Date } };
|
||||
};
|
||||
const gte = fallbackArgs.where.measuredAt?.gte;
|
||||
expect(gte).toBeInstanceOf(Date);
|
||||
expect(gte!.getTime()).toBe(now.getTime() - 1095 * dayMs);
|
||||
});
|
||||
|
||||
it("day-buckets the dense-type fallback in SQL instead of a raw walk (v1.28.25)", async () => {
|
||||
// BLOOD_GLUCOSE on a cold tier (heavy import before the rollup
|
||||
// backfill warms) used to findMany the entire CGM history. The
|
||||
// fallback now runs a single day-bucket aggregate; only the bounded
|
||||
// recent read touches findMany.
|
||||
findMany.mockResolvedValueOnce(dailyRows(80, now)); // bounded recent
|
||||
const bucketRows: Array<{ bucket_start: Date; mean: number }> = [];
|
||||
for (let i = 0; i < 800; i++) {
|
||||
bucketRows.push({
|
||||
bucket_start: new Date(now.getTime() - i * dayMs),
|
||||
mean: 100,
|
||||
});
|
||||
}
|
||||
queryRaw.mockResolvedValueOnce(bucketRows.reverse());
|
||||
|
||||
readBestGranularityRollups.mockResolvedValue(null);
|
||||
|
||||
const series = await buildGradedSeriesWithRollups(
|
||||
"u1",
|
||||
"BLOOD_GLUCOSE",
|
||||
now,
|
||||
);
|
||||
|
||||
expect(series.monthly.length).toBeGreaterThan(0);
|
||||
expect(series.yearly.length).toBeGreaterThan(0);
|
||||
// Raw findMany ran once (the bounded recent read) — the fallback
|
||||
// itself went through the SQL aggregate.
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(queryRaw).toHaveBeenCalledTimes(1);
|
||||
expect(readDayAggregates).toHaveBeenCalledTimes(2);
|
||||
const fallback = readDayAggregates.mock
|
||||
.calls[1][0] as ReadDayAggregatesOptions;
|
||||
// Bounded to the yearly router's horizon, never an unwindowed walk.
|
||||
expect(fallback.since.getTime()).toBe(now.getTime() - 1095 * dayMs);
|
||||
expect(findMany).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reads monthly/yearly from the tier when it has coverage and skips the full read", async () => {
|
||||
findMany.mockResolvedValueOnce(dailyRows(80, now));
|
||||
|
||||
// MONTH coverage for the 1-year window, YEAR coverage for the 3-year.
|
||||
it("reads monthly/yearly from the tier when it has coverage and skips the fallback", async () => {
|
||||
serveFrom(points(80, now));
|
||||
readBestGranularityRollups.mockImplementation(
|
||||
async (_u: string, _t: string, windowDays: number) => {
|
||||
if (windowDays === 365) {
|
||||
return {
|
||||
granularity: "MONTH",
|
||||
rows: [
|
||||
rollupRow(new Date("2025-09-01T00:00:00Z"), 79),
|
||||
rollupRow(new Date("2025-10-01T00:00:00Z"), 80),
|
||||
],
|
||||
};
|
||||
}
|
||||
return {
|
||||
granularity: "YEAR",
|
||||
rows: [rollupRow(new Date("2023-01-01T00:00:00Z"), 82)],
|
||||
};
|
||||
},
|
||||
async (_u: string, _t: string, windowDays: number) =>
|
||||
windowDays === 365
|
||||
? {
|
||||
granularity: "MONTH",
|
||||
rows: [
|
||||
rollupRow(new Date("2025-09-01T00:00:00Z"), 79),
|
||||
rollupRow(new Date("2025-10-01T00:00:00Z"), 80),
|
||||
],
|
||||
}
|
||||
: {
|
||||
granularity: "YEAR",
|
||||
rows: [rollupRow(new Date("2023-01-01T00:00:00Z"), 82)],
|
||||
},
|
||||
);
|
||||
|
||||
const series = await buildGradedSeriesWithRollups("u1", "WEIGHT", now);
|
||||
|
||||
expect(series.monthly.length).toBeGreaterThan(0);
|
||||
expect(series.yearly.length).toBeGreaterThan(0);
|
||||
// Only the bounded recent read — no full-history fallback when the
|
||||
// tier covers both coarse slices.
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(readDayAggregates).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("falls back to a full read for only the slice the tier misses", async () => {
|
||||
findMany
|
||||
.mockResolvedValueOnce(dailyRows(80, now)) // bounded recent
|
||||
.mockResolvedValueOnce(dailyRows(800, now)); // full-history fallback
|
||||
|
||||
// MONTH covered, YEAR missing.
|
||||
it("falls back only for the slice the tier misses", async () => {
|
||||
serveFrom(points(800, now));
|
||||
readBestGranularityRollups.mockImplementation(
|
||||
async (_u: string, _t: string, windowDays: number) => {
|
||||
if (windowDays === 365) {
|
||||
return {
|
||||
granularity: "MONTH",
|
||||
rows: [rollupRow(new Date("2025-10-01T00:00:00Z"), 80)],
|
||||
};
|
||||
}
|
||||
return null;
|
||||
},
|
||||
async (_u: string, _t: string, windowDays: number) =>
|
||||
windowDays === 365
|
||||
? {
|
||||
granularity: "MONTH",
|
||||
rows: [rollupRow(new Date("2025-10-01T00:00:00Z"), 80)],
|
||||
}
|
||||
: null,
|
||||
);
|
||||
|
||||
const series = await buildGradedSeriesWithRollups("u1", "WEIGHT", now);
|
||||
|
||||
expect(series.monthly.length).toBeGreaterThan(0);
|
||||
expect(series.monthly).toHaveLength(1);
|
||||
expect(series.yearly.length).toBeGreaterThan(0);
|
||||
expect(findMany).toHaveBeenCalledTimes(2);
|
||||
expect(readDayAggregates).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("buildGradedSeriesFromDayAggregates — parity with the raw fold", () => {
|
||||
const now = new Date("2026-05-31T12:00:00Z");
|
||||
const segmentStarts = [21, 91, 451].map(
|
||||
(d) => new Date(now.getTime() - d * dayMs + 1),
|
||||
);
|
||||
|
||||
it("matches the raw-row fold for recent / weekly / monthly and yearly stats", () => {
|
||||
// Several readings per day, including readings exactly on the segment
|
||||
// edges, so the partition itself is under test.
|
||||
const rows = [
|
||||
...points(900, now, 5),
|
||||
{ measuredAt: new Date(now.getTime() - 21 * dayMs), value: 99 },
|
||||
{ measuredAt: new Date(now.getTime() - 91 * dayMs), value: 41 },
|
||||
];
|
||||
const raw = buildGradedSeriesFromPoints(rows, now);
|
||||
const agg = buildGradedSeriesFromDayAggregates(
|
||||
foldDayAggregates(rows, {
|
||||
since: new Date(0),
|
||||
until: now,
|
||||
timeZone: "Europe/Berlin",
|
||||
segmentStarts,
|
||||
}),
|
||||
);
|
||||
expect(agg.recent).toEqual(raw.recent);
|
||||
expect(agg.weekly).toEqual(raw.weekly);
|
||||
expect(agg.monthly).toEqual(raw.monthly);
|
||||
// The yearly slope runs over day means rather than single readings;
|
||||
// every other yearly figure is exact.
|
||||
expect(agg.yearly.map(({ slope: _s, ...b }) => b)).toEqual(
|
||||
raw.yearly.map(({ slope: _s, ...b }) => b),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
|
||||
// The graded series folds per-day aggregates in SQL; the fake folds the
|
||||
// mocked `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique: vi.fn() },
|
||||
|
||||
@@ -12,6 +12,14 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
* `runStatusCompletion` is stubbed to capture it.
|
||||
*/
|
||||
|
||||
// The graded series folds per-day aggregates in SQL; the fake folds the
|
||||
// mocked `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique: vi.fn() },
|
||||
|
||||
@@ -19,6 +19,14 @@ import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
* generation on every navigation while a provider is degraded.
|
||||
*/
|
||||
|
||||
// The graded series folds per-day aggregates in SQL; the fake folds the
|
||||
// mocked `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique: vi.fn() },
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
|
||||
// The graded series folds per-day aggregates in SQL; the fake folds the
|
||||
// mocked `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
user: { findUnique: vi.fn() },
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, expect, it, vi, beforeEach } from "vitest";
|
||||
|
||||
// The graded series folds per-day aggregates in SQL; the fake folds the
|
||||
// mocked `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
auditLog: { findFirst: vi.fn(), create: vi.fn() },
|
||||
|
||||
@@ -12,9 +12,27 @@
|
||||
* The numbers below are shaped like the real thing: a resting-pulse series in
|
||||
* the 60s and 70s, one impossible sample dropped into it.
|
||||
*/
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { buildBaselineBand, dayMeansFromRows } from "../baseline";
|
||||
import type { ReadDayAggregatesOptions } from "@/lib/measurements/day-aggregates";
|
||||
|
||||
// The live day-mean read folds in SQL; the fake folds `rowsForRead` with the
|
||||
// same day / range rules, so these tests see the filter the read asks for.
|
||||
let rowsForRead: { value: number; measuredAt: Date }[] = [];
|
||||
const seenOptions: ReadDayAggregatesOptions[] = [];
|
||||
vi.mock("@/lib/db", () => ({ prisma: {} }));
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => {
|
||||
const { foldDayAggregates } =
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates");
|
||||
return {
|
||||
readDayAggregates: async (opts: ReadDayAggregatesOptions) => {
|
||||
seenOptions.push(opts);
|
||||
return foldDayAggregates(rowsForRead, opts);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
import { buildBaselineBand, readDayMeanSeries } from "../baseline";
|
||||
import { latestDayMeanFromRows } from "../coincident-deviation";
|
||||
import { isPlausibleMetricValue } from "@/lib/measurements/value-domain";
|
||||
|
||||
@@ -47,28 +65,49 @@ describe("metric plausibility domain", () => {
|
||||
});
|
||||
});
|
||||
|
||||
/** The live (rollup-miss) day-mean read over `rows`. */
|
||||
async function dayMeansFromRows(
|
||||
rows: { value: number; measuredAt: Date }[],
|
||||
type: "PULSE",
|
||||
) {
|
||||
rowsForRead = rows;
|
||||
const { points } = await readDayMeanSeries(
|
||||
"u1",
|
||||
type,
|
||||
3650,
|
||||
new Date("2026-09-01T00:00:00Z"),
|
||||
new Map(),
|
||||
);
|
||||
return points;
|
||||
}
|
||||
|
||||
describe("per-day means behind the personal band", () => {
|
||||
it("keeps the day mean on the real readings when one sample is impossible", () => {
|
||||
it("asks the read for the metric's own plausibility domain", async () => {
|
||||
await dayMeansFromRows(pulseDay("2026-08-01", [66]), "PULSE");
|
||||
expect(seenOptions.at(-1)?.valueRange).toEqual({ min: 20, max: 300 });
|
||||
});
|
||||
|
||||
it("keeps the day mean on the real readings when one sample is impossible", async () => {
|
||||
const rows = pulseDay("2026-08-01", [66, 70, 111287531.01, 68]);
|
||||
|
||||
const [point] = dayMeansFromRows(rows, "PULSE");
|
||||
const [point] = await dayMeansFromRows(rows, "PULSE");
|
||||
|
||||
expect(point.day).toBe("2026-08-01");
|
||||
expect(point.mean).toBeCloseTo(68, 6);
|
||||
});
|
||||
|
||||
it("drops a day whose every reading is impossible rather than inventing one", () => {
|
||||
it("drops a day whose every reading is impossible rather than inventing one", async () => {
|
||||
const rows = [
|
||||
...pulseDay("2026-08-01", [66, 70, 68]),
|
||||
...pulseDay("2026-08-02", [36016.75, 111287531.01]),
|
||||
];
|
||||
|
||||
const points = dayMeansFromRows(rows, "PULSE");
|
||||
const points = await dayMeansFromRows(rows, "PULSE");
|
||||
|
||||
expect(points.map((p) => p.day)).toEqual(["2026-08-01"]);
|
||||
});
|
||||
|
||||
it("holds the band inside the metric's own domain across a poisoned month", () => {
|
||||
it("holds the band inside the metric's own domain across a poisoned month", async () => {
|
||||
const rows: { value: number; measuredAt: Date }[] = [];
|
||||
for (let d = 1; d <= 28; d += 1) {
|
||||
const day = `2026-08-${String(d).padStart(2, "0")}`;
|
||||
@@ -81,7 +120,7 @@ describe("per-day means behind the personal band", () => {
|
||||
rows.push(...pulseDay("2026-08-26", [98211.5]));
|
||||
|
||||
const band = buildBaselineBand(
|
||||
dayMeansFromRows(rows, "PULSE").map((p) => p.mean),
|
||||
(await dayMeansFromRows(rows, "PULSE")).map((p) => p.mean),
|
||||
"PULSE",
|
||||
);
|
||||
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: { measurement: { findMany: vi.fn() } },
|
||||
}));
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: { measurement: { findMany: vi.fn() } },
|
||||
}));
|
||||
|
||||
@@ -17,6 +17,14 @@
|
||||
*/
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
measurement: {
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
measurement: { findMany: vi.fn().mockResolvedValue([]) },
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
measurement: { findMany: vi.fn() },
|
||||
|
||||
@@ -1,5 +1,13 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: { measurement: { findMany: vi.fn() } },
|
||||
}));
|
||||
|
||||
@@ -29,12 +29,13 @@
|
||||
* composed-bucket baseline matches the raw-DAY baseline within tolerance.
|
||||
*/
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { readDayAggregates } from "@/lib/measurements/day-aggregates";
|
||||
import { getAgeFromDateOfBirth } from "@/lib/analytics/pulse-targets";
|
||||
import { toProfileSex } from "@/lib/profile/sex";
|
||||
import {
|
||||
clampDerivedLowerBound,
|
||||
isPlausibleMetricValue,
|
||||
plausibleMetricRange,
|
||||
} from "@/lib/measurements/value-domain";
|
||||
import type { ProfileSex } from "@/lib/profile/sex";
|
||||
import {
|
||||
@@ -212,39 +213,6 @@ export function buildBaselineBand(
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Group raw readings into per-day means, keeping only the values that sit
|
||||
* inside the metric's own plausibility domain.
|
||||
*
|
||||
* Pure, and exported because this is the arithmetic the band rests on: a
|
||||
* single stored value the application itself declares impossible drags the
|
||||
* day's mean with it, the median then carries it into the band, and the
|
||||
* personal "usual range" reads in the tens of thousands. Filtering here rather
|
||||
* than at the edge of the card keeps the number wrong in exactly one place —
|
||||
* nowhere — instead of right on one surface and wrong on the next.
|
||||
*
|
||||
* A day whose every reading is implausible produces no point at all. That is
|
||||
* the honest outcome: the day contributed no measurement, and an absent day is
|
||||
* something the coverage floor already knows how to report.
|
||||
*/
|
||||
export function dayMeansFromRows(
|
||||
rows: readonly { value: number; measuredAt: Date }[],
|
||||
type: MeasurementType,
|
||||
): DayMeanPoint[] {
|
||||
const byDay = new Map<string, { sum: number; count: number }>();
|
||||
for (const row of rows) {
|
||||
if (!isPlausibleMetricValue(type, row.value)) continue;
|
||||
const day = row.measuredAt.toISOString().slice(0, 10);
|
||||
const acc = byDay.get(day) ?? { sum: 0, count: 0 };
|
||||
acc.sum += row.value;
|
||||
acc.count += 1;
|
||||
byDay.set(day, acc);
|
||||
}
|
||||
return [...byDay.entries()]
|
||||
.sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
|
||||
.map(([day, acc]) => ({ day, mean: acc.sum / acc.count }));
|
||||
}
|
||||
|
||||
// ── reads ────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
@@ -292,23 +260,22 @@ export async function readDayMeanSeries(
|
||||
// the spread is still DAY-native.
|
||||
}
|
||||
|
||||
// Per-type live fallback — bounded raw read, grouped into per-day
|
||||
// means. Honest provenance: source = "live".
|
||||
// Per-type live fallback — per-day means folded in SQL, so the read is
|
||||
// one row per day however densely the metric is sampled (#1023). Only
|
||||
// values inside the metric's own plausibility domain count: a single
|
||||
// stored value the application declares impossible would drag the day's
|
||||
// mean with it and the median would carry it into the band. A day whose
|
||||
// every reading is implausible produces no point at all. Days are UTC
|
||||
// days, matching the DAY rollup path above.
|
||||
const since = new Date(now.getTime() - windowDays * 24 * 60 * 60 * 1000);
|
||||
const rows = await prisma.measurement.findMany({
|
||||
where: {
|
||||
userId,
|
||||
type,
|
||||
deletedAt: null,
|
||||
measuredAt: { gte: since },
|
||||
},
|
||||
orderBy: { measuredAt: "asc" },
|
||||
select: { value: true, measuredAt: true },
|
||||
const days = await readDayAggregates({
|
||||
userId,
|
||||
type,
|
||||
since,
|
||||
timeZone: "UTC",
|
||||
valueRange: plausibleMetricRange(type),
|
||||
});
|
||||
if (rows.length === 0) {
|
||||
return { points: [], source: "none" };
|
||||
}
|
||||
const points = dayMeansFromRows(rows, type);
|
||||
const points = days.map((d) => ({ day: d.day, mean: d.sum / d.n }));
|
||||
if (points.length === 0) {
|
||||
return { points: [], source: "none" };
|
||||
}
|
||||
|
||||
@@ -48,12 +48,15 @@
|
||||
*/
|
||||
|
||||
import type { MeasurementType } from "@/generated/prisma/client";
|
||||
import { prisma } from "@/lib/db";
|
||||
import {
|
||||
readBestGranularityRollups,
|
||||
type RollupBucketRow,
|
||||
} from "@/lib/rollups/measurement-read-wmy";
|
||||
import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
|
||||
import {
|
||||
readDayAggregates,
|
||||
type DayAggregateRow,
|
||||
} from "@/lib/measurements/day-aggregates";
|
||||
import { toBerlinYmd } from "@/lib/tz/resolver";
|
||||
|
||||
const MS_PER_DAY = 24 * 60 * 60 * 1000;
|
||||
@@ -83,18 +86,6 @@ const MONTHLY_WINDOW_MS = WEEKLY_WINDOW_MS + MONTHLY_MONTHS * 30 * MS_PER_DAY;
|
||||
*/
|
||||
const COLD_FALLBACK_WINDOW_DAYS = 1095;
|
||||
|
||||
/**
|
||||
* Sample-dense types whose cold-tier fallback is day-bucketed in SQL
|
||||
* (one AVG row per day) instead of read raw — same dense set the series
|
||||
* route caps (CGM glucose ~288 rows/day, PULSE per-sample / hourly).
|
||||
* The fold's output shape is unchanged; on the cold path only, a dense
|
||||
* type's monthly/yearly min/max/n derive from day means rather than
|
||||
* individual samples — the warm rollup tier restores sample-exact
|
||||
* figures as soon as the backfill lands.
|
||||
*/
|
||||
const DENSE_FALLBACK_TYPES: ReadonlySet<MeasurementType> =
|
||||
new Set<MeasurementType>(["BLOOD_GLUCOSE", "PULSE"]);
|
||||
|
||||
export interface RecentDayBucket {
|
||||
/** Berlin YYYY-MM-DD. */
|
||||
date: string;
|
||||
@@ -171,6 +162,11 @@ function berlinYearKey(date: Date): string {
|
||||
*/
|
||||
function berlinIsoWeekKey(date: Date): string {
|
||||
const { year, month, day } = toBerlinYmd(date);
|
||||
return isoWeekKeyOfYmd(year, month, day);
|
||||
}
|
||||
|
||||
/** ISO week key of a calendar date given as `YYYY`, `MM`, `DD`. */
|
||||
function isoWeekKeyOfYmd(year: string, month: string, day: string): string {
|
||||
const localMidnight = new Date(
|
||||
Date.UTC(Number(year), Number(month) - 1, Number(day)),
|
||||
);
|
||||
@@ -309,6 +305,133 @@ export function buildGradedSeriesFromPoints(
|
||||
return { recent, weekly, monthly, yearly };
|
||||
}
|
||||
|
||||
function mergeInto(
|
||||
map: Map<string, Agg>,
|
||||
key: string,
|
||||
row: DayAggregateRow,
|
||||
): void {
|
||||
const existing = map.get(key);
|
||||
if (existing) {
|
||||
existing.sum += row.sum;
|
||||
existing.n += row.n;
|
||||
if (row.min < existing.min) existing.min = row.min;
|
||||
if (row.max > existing.max) existing.max = row.max;
|
||||
} else {
|
||||
map.set(key, { sum: row.sum, min: row.min, max: row.max, n: row.n });
|
||||
}
|
||||
}
|
||||
|
||||
/** Segment boundaries matching `buildGradedSeriesFromPoints`' age split. */
|
||||
function gradedSegmentStarts(now: Date): Date[] {
|
||||
const nowMs = now.getTime();
|
||||
return [
|
||||
new Date(nowMs - RECENT_WINDOW_MS),
|
||||
new Date(nowMs - WEEKLY_WINDOW_MS),
|
||||
new Date(nowMs - MONTHLY_WINDOW_MS),
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the graded series' per-day aggregates for one type, folded in SQL.
|
||||
*
|
||||
* Replaces the raw `findMany` over the window (#1023): a watch that streams
|
||||
* heart rate every few seconds put a million rows into that read, which is
|
||||
* what exhausted the worker's heap. The rows here are one per Berlin day per
|
||||
* age segment, and their sum / count / min / max are exact, so the recent,
|
||||
* weekly and monthly slices come out identical to folding the raw rows.
|
||||
*
|
||||
* The segment boundaries use the same edges as `buildGradedSeriesFromPoints`:
|
||||
* a row exactly `RECENT_WINDOW_MS` old is weekly there, so it is segment 1
|
||||
* here too.
|
||||
*/
|
||||
async function readGradedDayAggregates(
|
||||
userId: string,
|
||||
type: MeasurementType,
|
||||
since: Date,
|
||||
now: Date,
|
||||
): Promise<DayAggregateRow[]> {
|
||||
return readDayAggregates({
|
||||
userId,
|
||||
type,
|
||||
since,
|
||||
// Future-dated readings were skipped by the in-memory fold.
|
||||
until: now,
|
||||
timeZone: "Europe/Berlin",
|
||||
// `readDayAggregates` puts a row at segment k when it is older than k
|
||||
// starts (`measured_at < start`). The in-memory fold treats an age of
|
||||
// exactly the window as the older bucket, so shift each start by 1 ms.
|
||||
segmentStarts: gradedSegmentStarts(now).map(
|
||||
(d) => new Date(d.getTime() + 1),
|
||||
),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* The graded series folded from per-day aggregates (segment = age class).
|
||||
*
|
||||
* recent / weekly / monthly and every bucket's min / max / mean / n match
|
||||
* the raw-row fold exactly. The yearly slope is taken over the day means in
|
||||
* date order rather than over individual readings: a direction signal either
|
||||
* way, and the only form that stays bounded on a dense stream.
|
||||
*/
|
||||
export function buildGradedSeriesFromDayAggregates(
|
||||
rows: readonly DayAggregateRow[],
|
||||
): GradedSeries {
|
||||
const recentAgg = new Map<string, Agg>();
|
||||
const weeklyAgg = new Map<string, Agg>();
|
||||
const monthlyAgg = new Map<string, Agg>();
|
||||
const yearlyAgg = new Map<string, Agg>();
|
||||
const yearlyValues = new Map<string, number[]>();
|
||||
|
||||
const ordered = [...rows].sort((a, b) =>
|
||||
a.day < b.day ? -1 : a.day > b.day ? 1 : b.segment - a.segment,
|
||||
);
|
||||
for (const row of ordered) {
|
||||
const [year, month, day] = row.day.split("-");
|
||||
if (row.segment === 0) {
|
||||
mergeInto(recentAgg, row.day, row);
|
||||
} else if (row.segment === 1) {
|
||||
mergeInto(weeklyAgg, isoWeekKeyOfYmd(year, month, day), row);
|
||||
} else if (row.segment === 2) {
|
||||
mergeInto(monthlyAgg, `${year}-${month}`, row);
|
||||
} else {
|
||||
mergeInto(yearlyAgg, year, row);
|
||||
const list = yearlyValues.get(year);
|
||||
const dayMean = row.sum / row.n;
|
||||
if (list) list.push(dayMean);
|
||||
else yearlyValues.set(year, [dayMean]);
|
||||
}
|
||||
}
|
||||
|
||||
const toBuckets = (map: Map<string, Agg>) =>
|
||||
Array.from(map.entries())
|
||||
.map(([key, a]) => ({
|
||||
key,
|
||||
min: round(a.min),
|
||||
max: round(a.max),
|
||||
mean: round(a.sum / a.n),
|
||||
n: a.n,
|
||||
}))
|
||||
.sort((x, y) => x.key.localeCompare(y.key));
|
||||
|
||||
return {
|
||||
recent: toBuckets(recentAgg).map(({ key, ...b }) => ({ date: key, ...b })),
|
||||
weekly: toBuckets(weeklyAgg).map(({ key, ...b }) => ({
|
||||
weekISO: key,
|
||||
...b,
|
||||
})),
|
||||
monthly: toBuckets(monthlyAgg).map(({ key, ...b }) => ({
|
||||
month: key,
|
||||
...b,
|
||||
})),
|
||||
yearly: toBuckets(yearlyAgg).map(({ key, ...b }) => ({
|
||||
year: key,
|
||||
...b,
|
||||
slope: leastSquaresSlope(yearlyValues.get(key) ?? []),
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Project a rollup bucket row to a monthly/yearly graded bucket. The
|
||||
* tier already carries min/max/mean/slope per bucket — no JS folding.
|
||||
@@ -335,24 +458,25 @@ function rollupYearly(rows: RollupBucketRow[]): YearlyBucket[] {
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the graded series with the recent / weekly slices folded from a
|
||||
* bounded raw read and the monthly / yearly slices read from the
|
||||
* Build the graded series with the recent / weekly slices folded from
|
||||
* per-day aggregates and the monthly / yearly slices read from the
|
||||
* pre-aggregated rollup tier.
|
||||
*
|
||||
* The recent / weekly read is capped to the recent + weekly horizon
|
||||
* (last ~90 days), so the common warm-tier path never pulls the full
|
||||
* multi-year history into memory. The rollup router picks MONTH for the
|
||||
* ~1-year window and YEAR for the multi-year tail.
|
||||
* The recent / weekly read covers the recent + weekly horizon (last ~90
|
||||
* days) and is folded in SQL to one row per day, so its size depends on
|
||||
* the number of days, never on how densely a device samples (#1023). The
|
||||
* rollup router picks MONTH for the ~1-year window and YEAR for the
|
||||
* multi-year tail.
|
||||
*
|
||||
* Coverage-miss fallback: when the tier has no MONTH (resp. YEAR)
|
||||
* coverage for a metric — a fresh account the boot-backfill has not yet
|
||||
* caught up on — the bounded ~90-day read can NOT supply the monthly /
|
||||
* yearly slices (every row in it folds into recent / weekly), so naively
|
||||
* reusing it would ship empty monthly / yearly arrays even when years of
|
||||
* raw history exist. On that miss this falls back to an in-memory fold
|
||||
* over a bounded window (1095 days — the yearly router's own horizon;
|
||||
* sample-dense types are additionally day-bucketed in SQL) so the coarse
|
||||
* slices are populated from the raw rows rather than left falsely empty.
|
||||
* raw history exist. On that miss this falls back to the same per-day
|
||||
* aggregate over a bounded window (1095 days — the yearly router's own
|
||||
* horizon) so the coarse slices are populated from the stored rows rather
|
||||
* than left falsely empty.
|
||||
* The fallback read only happens when the tier is cold, which is exactly
|
||||
* the case the write-amplified tier was meant to avoid — so the hot path
|
||||
* stays bounded and the cold path stays correct.
|
||||
@@ -363,15 +487,8 @@ export async function buildGradedSeriesWithRollups(
|
||||
now: Date,
|
||||
): Promise<GradedSeries> {
|
||||
const since = new Date(now.getTime() - WEEKLY_WINDOW_MS);
|
||||
const rawRecent = await prisma.measurement.findMany({
|
||||
where: { userId, type, deletedAt: null, measuredAt: { gte: since } },
|
||||
orderBy: { measuredAt: "asc" },
|
||||
select: { measuredAt: true, value: true },
|
||||
});
|
||||
|
||||
const recentGraded = buildGradedSeriesFromPoints(
|
||||
rawRecent.map((r) => ({ measuredAt: r.measuredAt, value: r.value })),
|
||||
now,
|
||||
const recentGraded = buildGradedSeriesFromDayAggregates(
|
||||
await readGradedDayAggregates(userId, type, since, now),
|
||||
);
|
||||
|
||||
// v1.11.2 — load the source-priority blob once and thread it into both
|
||||
@@ -420,57 +537,19 @@ export async function buildGradedSeriesWithRollups(
|
||||
}
|
||||
|
||||
// Tier coverage miss for one or both coarse slices: the bounded
|
||||
// ~90-day read can't supply them, so fold a BOUNDED history window in
|
||||
// memory and take whichever slices the tier left empty. The fallback
|
||||
// read is the exception (cold-tier accounts), never the warm-tier
|
||||
// norm. v1.28.25 — the read is bounded to the same 1095-day horizon
|
||||
// the yearly rollup router serves (it was previously unwindowed), and
|
||||
// the sample-dense types are day-bucketed in SQL so a heavy import
|
||||
// ahead of the rollup backfill can't drag six-figure row sets into JS.
|
||||
// ~90-day read can't supply them, so fold a BOUNDED history window and
|
||||
// take whichever slices the tier left empty. The fallback read is the
|
||||
// exception (cold-tier accounts), never the warm-tier norm. It is
|
||||
// bounded to the same 1095-day horizon the yearly rollup router serves,
|
||||
// and folded per day in SQL for every type, so a heavy import ahead of
|
||||
// the rollup backfill can't drag six-figure row sets into JS.
|
||||
if (!monthlyCovered || !yearlyCovered) {
|
||||
const fallbackSince = new Date(
|
||||
now.getTime() - COLD_FALLBACK_WINDOW_DAYS * MS_PER_DAY,
|
||||
);
|
||||
let fallbackPoints: Point[];
|
||||
if (DENSE_FALLBACK_TYPES.has(type)) {
|
||||
// Parameter-bound day-bucket aggregate, mirroring the rollup tier's
|
||||
// `date_trunc(...) GROUP BY` (measurement-rollups.ts) — at most
|
||||
// ~1095 rows regardless of sample density.
|
||||
const bucketRows = await prisma.$queryRaw<
|
||||
Array<{ bucket_start: Date; mean: number }>
|
||||
>`
|
||||
SELECT
|
||||
date_trunc('day', m."measured_at") AS bucket_start,
|
||||
AVG(m."value")::double precision AS mean
|
||||
FROM measurements m
|
||||
WHERE m."user_id" = ${userId}
|
||||
AND m."type" = ${type}::"measurement_type"
|
||||
AND m."measured_at" >= ${fallbackSince}
|
||||
AND m."deleted_at" IS NULL
|
||||
GROUP BY date_trunc('day', m."measured_at")
|
||||
ORDER BY bucket_start ASC
|
||||
`;
|
||||
fallbackPoints = bucketRows.map((r) => ({
|
||||
measuredAt: r.bucket_start,
|
||||
value: r.mean,
|
||||
}));
|
||||
} else {
|
||||
const rows = await prisma.measurement.findMany({
|
||||
where: {
|
||||
userId,
|
||||
type,
|
||||
deletedAt: null,
|
||||
measuredAt: { gte: fallbackSince },
|
||||
},
|
||||
orderBy: { measuredAt: "asc" },
|
||||
select: { measuredAt: true, value: true },
|
||||
});
|
||||
fallbackPoints = rows.map((r) => ({
|
||||
measuredAt: r.measuredAt,
|
||||
value: r.value,
|
||||
}));
|
||||
}
|
||||
const fullGraded = buildGradedSeriesFromPoints(fallbackPoints, now);
|
||||
const fullGraded = buildGradedSeriesFromDayAggregates(
|
||||
await readGradedDayAggregates(userId, type, fallbackSince, now),
|
||||
);
|
||||
if (!monthlyCovered) monthly = fullGraded.monthly;
|
||||
if (!yearlyCovered) yearly = fullGraded.yearly;
|
||||
}
|
||||
|
||||
@@ -24,6 +24,7 @@
|
||||
*/
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
import { encryptToBytes } from "@/lib/ai/coach/bytes-codec";
|
||||
import { readDayAggregates } from "@/lib/measurements/day-aggregates";
|
||||
|
||||
export const COACH_PLAN_REVIEW_QUEUE = "coach-plan-review";
|
||||
// Daily at 05:25 Europe/Berlin — just after the 05:20 reminder sweep, so the
|
||||
@@ -49,7 +50,7 @@ export interface CoachPlanReviewSummary {
|
||||
|
||||
type ReviewPrisma = Pick<
|
||||
PrismaClient,
|
||||
"coachReminder" | "coachPlan" | "measurement"
|
||||
"coachReminder" | "coachPlan" | "$queryRaw"
|
||||
>;
|
||||
|
||||
/** Metric valence — whether a higher reading is the better outcome. */
|
||||
@@ -282,31 +283,22 @@ export async function runCoachPlanReviewTick(
|
||||
const since = new Date(
|
||||
plan.createdAt.getTime() - BEFORE_WINDOW_DAYS * MS_PER_DAY,
|
||||
);
|
||||
const rows = await prisma.measurement.findMany({
|
||||
where: {
|
||||
userId: plan.userId,
|
||||
type: resolved.type,
|
||||
deletedAt: null,
|
||||
measuredAt: { gte: since, lte: now },
|
||||
},
|
||||
orderBy: { measuredAt: "asc" },
|
||||
select: { value: true, measuredAt: true },
|
||||
// Day means folded in SQL (UTC days), one row per day however
|
||||
// densely the metric is sampled (#1023).
|
||||
const byDay = await readDayAggregates({
|
||||
userId: plan.userId,
|
||||
type: resolved.type,
|
||||
since,
|
||||
until: now,
|
||||
timeZone: "UTC",
|
||||
db: prisma,
|
||||
});
|
||||
// Day-mean the raw rows, then split at the experiment start.
|
||||
const byDay = new Map<string, { sum: number; count: number }>();
|
||||
for (const row of rows) {
|
||||
const day = row.measuredAt.toISOString().slice(0, 10);
|
||||
const agg = byDay.get(day) ?? { sum: 0, count: 0 };
|
||||
agg.sum += row.value;
|
||||
agg.count += 1;
|
||||
byDay.set(day, agg);
|
||||
}
|
||||
const startKey = plan.createdAt.toISOString().slice(0, 10);
|
||||
const before: number[] = [];
|
||||
const after: number[] = [];
|
||||
const all: number[] = [];
|
||||
for (const [day, agg] of byDay) {
|
||||
const dayMean = agg.sum / agg.count;
|
||||
for (const { day, sum, n } of byDay) {
|
||||
const dayMean = sum / n;
|
||||
all.push(dayMean);
|
||||
if (day < startKey) before.push(dayMean);
|
||||
else after.push(dayMean);
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
/**
|
||||
* In-memory stand-in for `readDayAggregates`, for unit tests that mock
|
||||
* `@/lib/db`.
|
||||
*
|
||||
* It reads the rows through the mocked `prisma.measurement.findMany` (so a
|
||||
* test's existing row fixture keeps feeding the reader) and folds them with
|
||||
* the same day / segment / range rules the SQL applies. The integration test
|
||||
* `tests/integration/dense-read-bounded.test.ts` pins that this fold and the
|
||||
* SQL agree on real Postgres.
|
||||
*/
|
||||
import { prisma } from "@/lib/db";
|
||||
import type {
|
||||
DayAggregateRow,
|
||||
ReadDayAggregatesOptions,
|
||||
} from "@/lib/measurements/day-aggregates";
|
||||
import { userDayKey } from "@/lib/tz/format";
|
||||
|
||||
export function foldDayAggregates(
|
||||
rows: ReadonlyArray<{ measuredAt: Date; value: number }>,
|
||||
opts: Omit<ReadDayAggregatesOptions, "userId" | "type">,
|
||||
): DayAggregateRow[] {
|
||||
const starts = opts.segmentStarts ?? [];
|
||||
const byKey = new Map<string, DayAggregateRow>();
|
||||
for (const r of rows) {
|
||||
const t = r.measuredAt.getTime();
|
||||
if (t < opts.since.getTime()) continue;
|
||||
if (opts.until && t > opts.until.getTime()) continue;
|
||||
if (
|
||||
opts.valueRange &&
|
||||
(r.value < opts.valueRange.min || r.value > opts.valueRange.max)
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const day = userDayKey(r.measuredAt, opts.timeZone);
|
||||
const segment = starts.filter((s) => t < s.getTime()).length;
|
||||
const key = `${day}|${segment}`;
|
||||
const acc = byKey.get(key);
|
||||
if (acc) {
|
||||
acc.n += 1;
|
||||
acc.sum += r.value;
|
||||
if (r.value < acc.min) acc.min = r.value;
|
||||
if (r.value > acc.max) acc.max = r.value;
|
||||
} else {
|
||||
byKey.set(key, {
|
||||
day,
|
||||
segment,
|
||||
n: 1,
|
||||
sum: r.value,
|
||||
min: r.value,
|
||||
max: r.value,
|
||||
});
|
||||
}
|
||||
}
|
||||
return [...byKey.values()].sort((a, b) =>
|
||||
a.day < b.day ? -1 : a.day > b.day ? 1 : b.segment - a.segment,
|
||||
);
|
||||
}
|
||||
|
||||
export async function fakeReadDayAggregates(
|
||||
opts: ReadDayAggregatesOptions,
|
||||
): Promise<DayAggregateRow[]> {
|
||||
const rows = ((await prisma.measurement.findMany({
|
||||
where: {
|
||||
userId: opts.userId,
|
||||
type: opts.type,
|
||||
deletedAt: null,
|
||||
measuredAt: { gte: opts.since, ...(opts.until && { lte: opts.until }) },
|
||||
},
|
||||
orderBy: { measuredAt: "asc" },
|
||||
select: { measuredAt: true, value: true },
|
||||
})) ?? []) as Array<{ measuredAt: Date; value: number }>;
|
||||
return foldDayAggregates(rows, opts);
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
/**
|
||||
* Per-day aggregates of one measurement type, folded in Postgres.
|
||||
*
|
||||
* Several background readers (the status cards' graded series, the vitals
|
||||
* baseline, the tiered context, the plan review) used to `findMany` every raw
|
||||
* row in a time window and then fold them into days in JavaScript. The window
|
||||
* bounded the time span but not the row count, so an account whose watch
|
||||
* streams heart rate every few seconds materialised hundreds of thousands to
|
||||
* millions of objects per read. In the worker that was enough to exhaust the
|
||||
* heap, and the read itself outran the statement timeout.
|
||||
*
|
||||
* This reader returns at most one row per (day, segment), whatever the sample
|
||||
* density: the fold runs in SQL and only the aggregates cross the wire. `sum`,
|
||||
* `n`, `min` and `max` are exact, so every consumer that averaged, counted or
|
||||
* took extremes over the raw rows gets the same numbers from these rows.
|
||||
*
|
||||
* Days are cut in `timeZone`. `segmentStarts` optionally splits the window at
|
||||
* instants (newest first): a row's segment is the number of starts it lies
|
||||
* before, so a caller that treated "the last 21 days" and "the 70 days before
|
||||
* that" differently keeps exactly the same per-row partition.
|
||||
*/
|
||||
import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
|
||||
import { Prisma } from "@/generated/prisma/client";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { isValidTimezone } from "@/lib/tz/format";
|
||||
|
||||
export interface DayAggregateRow {
|
||||
/** Calendar day in the requested zone, `YYYY-MM-DD`. */
|
||||
day: string;
|
||||
/** Segment index: 0 for rows at or after `segmentStarts[0]`, and so on. */
|
||||
segment: number;
|
||||
n: number;
|
||||
sum: number;
|
||||
min: number;
|
||||
max: number;
|
||||
}
|
||||
|
||||
export interface ReadDayAggregatesOptions {
|
||||
userId: string;
|
||||
type: MeasurementType;
|
||||
/** Inclusive lower bound on `measuredAt`. */
|
||||
since: Date;
|
||||
/** Inclusive upper bound on `measuredAt`; omit for no upper bound. */
|
||||
until?: Date;
|
||||
/** IANA zone the day boundary is cut in. */
|
||||
timeZone: string;
|
||||
/** Rows outside `[min, max]` are dropped before the fold. */
|
||||
valueRange?: { min: number; max: number };
|
||||
/** Segment boundaries, newest first. */
|
||||
segmentStarts?: readonly Date[];
|
||||
/** Client to read through; defaults to the shared one. */
|
||||
db?: Pick<PrismaClient, "$queryRaw">;
|
||||
}
|
||||
|
||||
export async function readDayAggregates(
|
||||
opts: ReadDayAggregatesOptions,
|
||||
): Promise<DayAggregateRow[]> {
|
||||
const timeZone = isValidTimezone(opts.timeZone) ? opts.timeZone : "UTC";
|
||||
const until = opts.until
|
||||
? Prisma.sql`AND m."measured_at" <= ${opts.until}`
|
||||
: Prisma.empty;
|
||||
const range = opts.valueRange
|
||||
? Prisma.sql`AND m."value" >= ${opts.valueRange.min} AND m."value" <= ${opts.valueRange.max}`
|
||||
: Prisma.empty;
|
||||
const starts = opts.segmentStarts ?? [];
|
||||
const segment =
|
||||
starts.length === 0
|
||||
? Prisma.sql`0`
|
||||
: Prisma.sql`(${Prisma.join(
|
||||
starts.map(
|
||||
(s) =>
|
||||
Prisma.sql`(CASE WHEN m."measured_at" < ${s} THEN 1 ELSE 0 END)`,
|
||||
),
|
||||
" + ",
|
||||
)})`;
|
||||
|
||||
// The day and segment are computed once per row in `src` and grouped by
|
||||
// name, so the bound zone / boundary parameters never have to match
|
||||
// themselves across SELECT and GROUP BY.
|
||||
const rows = await (opts.db ?? prisma).$queryRaw<
|
||||
Array<{
|
||||
day: string;
|
||||
segment: number;
|
||||
n: number;
|
||||
sum: number;
|
||||
min: number;
|
||||
max: number;
|
||||
}>
|
||||
>`
|
||||
WITH src AS (
|
||||
SELECT
|
||||
to_char((m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}, 'YYYY-MM-DD') AS day,
|
||||
${segment} AS segment,
|
||||
m."value" AS value
|
||||
FROM measurements m
|
||||
WHERE m."user_id" = ${opts.userId}
|
||||
AND m."type" = ${opts.type}::measurement_type
|
||||
AND m."deleted_at" IS NULL
|
||||
AND m."measured_at" >= ${opts.since}
|
||||
${until}
|
||||
${range}
|
||||
)
|
||||
SELECT
|
||||
day,
|
||||
segment::int AS segment,
|
||||
COUNT(*)::int AS n,
|
||||
SUM(value)::double precision AS sum,
|
||||
MIN(value)::double precision AS min,
|
||||
MAX(value)::double precision AS max
|
||||
FROM src
|
||||
GROUP BY day, segment
|
||||
ORDER BY day ASC, segment DESC
|
||||
`;
|
||||
return rows.map((r) => ({
|
||||
day: r.day,
|
||||
segment: Number(r.segment),
|
||||
n: Number(r.n),
|
||||
sum: Number(r.sum),
|
||||
min: Number(r.min),
|
||||
max: Number(r.max),
|
||||
}));
|
||||
}
|
||||
@@ -105,12 +105,22 @@ export function isPlausibleMetricValue(
|
||||
value: number,
|
||||
): boolean {
|
||||
if (!Number.isFinite(value)) return false;
|
||||
if (!type) return true;
|
||||
const range = VALUE_RANGES[type];
|
||||
const range = plausibleMetricRange(type);
|
||||
if (range === undefined) return true;
|
||||
return value >= range.min && value <= range.max;
|
||||
}
|
||||
|
||||
/**
|
||||
* The plausibility domain of a metric, for a reader that filters in SQL
|
||||
* rather than row by row (`readDayAggregates`' `valueRange`). `undefined`
|
||||
* when the metric declares none, which admits every value.
|
||||
*/
|
||||
export function plausibleMetricRange(
|
||||
type: string | null | undefined,
|
||||
): { min: number; max: number } | undefined {
|
||||
return type ? VALUE_RANGES[type] : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* True when EVERY named metric forbids negative values (and at least one was
|
||||
* named). Used by the chart, which can paint several series in one axis: a
|
||||
|
||||
@@ -23,6 +23,14 @@ const mocks = vi.hoisted(() => ({
|
||||
queryRaw: vi.fn(),
|
||||
}));
|
||||
|
||||
// Day-mean reads fold per-day aggregates in SQL; the fake folds the mocked
|
||||
// `measurement.findMany` rows with the same rules.
|
||||
vi.mock("@/lib/measurements/day-aggregates", async () => ({
|
||||
readDayAggregates: (
|
||||
await import("@/lib/measurements/__tests__/fake-day-aggregates")
|
||||
).fakeReadDayAggregates,
|
||||
}));
|
||||
|
||||
vi.mock("../measurement-rollups", () => ({
|
||||
readRollupBuckets: mocks.readRollupBuckets,
|
||||
ensureUserRollupsFresh: mocks.ensureUserRollupsFresh,
|
||||
|
||||
@@ -33,12 +33,13 @@ import type {
|
||||
MeasurementType,
|
||||
RollupGranularity,
|
||||
} from "@/generated/prisma/client";
|
||||
import { isNearUtc, userDayKey } from "@/lib/tz/format";
|
||||
import { isNearUtc } from "@/lib/tz/format";
|
||||
import {
|
||||
ensureUserRollupsFresh,
|
||||
readRollupBuckets,
|
||||
} from "./measurement-rollups";
|
||||
import { probeRollupCoverage } from "./measurement-coverage";
|
||||
import { readDayAggregates } from "@/lib/measurements/day-aggregates";
|
||||
|
||||
const DAY_MS = 24 * 60 * 60 * 1000;
|
||||
|
||||
@@ -112,7 +113,7 @@ export interface TieredSeries {
|
||||
export interface BuildTieredSeriesOptions {
|
||||
/**
|
||||
* The user's IANA timezone. REQUIRED — the raw 0–14d band keys its
|
||||
* daily fold on the user's local calendar day (`userDayKey`) so a
|
||||
* daily fold on the user's local calendar day so a
|
||||
* late-evening reading lands on the right day across DST boundaries
|
||||
* and for non-UTC users. The coarser DAY/WEEK/MONTH/YEAR bands read
|
||||
* straight from the rollup table, whose `bucketStart` is UTC-anchored
|
||||
@@ -244,8 +245,8 @@ function extractAnomalies(
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the raw 0–14d daily points for one type, folding same-day rows.
|
||||
* Days are keyed on the user's local calendar via `userDayKey(tz)`, so a
|
||||
* Read the 0–14d daily points for one type, folding same-day rows.
|
||||
* Days are keyed on the user's local calendar in `tz`, so a
|
||||
* 23:30-local reading folds onto its local day rather than the UTC day —
|
||||
* correct across DST transitions and for non-UTC users.
|
||||
*/
|
||||
@@ -256,26 +257,19 @@ async function readRecentDaily(
|
||||
tz: string,
|
||||
): Promise<TieredRawPoint[]> {
|
||||
const since = new Date(now - TIERED_BANDS.rawDays * DAY_MS);
|
||||
const rows = await prisma.measurement.findMany({
|
||||
where: { userId, type, measuredAt: { gte: since }, deletedAt: null },
|
||||
select: { value: true, measuredAt: true },
|
||||
orderBy: { measuredAt: "asc" },
|
||||
// Folded per local day in SQL: one row per day however densely the type
|
||||
// is sampled, never the raw stream (#1023).
|
||||
const days = await readDayAggregates({
|
||||
userId,
|
||||
type,
|
||||
since,
|
||||
timeZone: tz,
|
||||
});
|
||||
const byDay = new Map<string, { sum: number; count: number }>();
|
||||
for (const r of rows) {
|
||||
const key = userDayKey(r.measuredAt, tz);
|
||||
const entry = byDay.get(key) ?? { sum: 0, count: 0 };
|
||||
entry.sum += r.value;
|
||||
entry.count += 1;
|
||||
byDay.set(key, entry);
|
||||
}
|
||||
return [...byDay.entries()]
|
||||
.sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
|
||||
.map(([date, { sum, count }]) => ({
|
||||
date,
|
||||
value: round(sum / count),
|
||||
count,
|
||||
}));
|
||||
return days.map((d) => ({
|
||||
date: d.day,
|
||||
value: round(d.sum / d.n),
|
||||
count: d.n,
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
/**
|
||||
* #1023 — a status card over a dense heart-rate stream, measured.
|
||||
*
|
||||
* A single account syncing heart rate from a watch killed the worker within
|
||||
* half an hour: the pulse status card's recent-window read was bounded in
|
||||
* time (91 days) but not in rows, so every refresh materialised the whole
|
||||
* stream as objects, and the read itself ran past the statement timeout.
|
||||
* The read now folds per day in Postgres.
|
||||
*
|
||||
* The fixture is one reading a minute for 91 days, 131 040 rows, seeded in
|
||||
* one `INSERT … generate_series`. Each half of the memory test reads the
|
||||
* same window: the per-day fold the card now uses, and the raw `findMany` it
|
||||
* used to issue. Without the second half the budget could be any number and
|
||||
* the first half would still pass. Every reading of the heap is taken after a
|
||||
* forced collection, so what is compared is what each read HOLDS; the
|
||||
* transient figure (taken before collecting) is reported alongside.
|
||||
*
|
||||
* The parity case pins that the SQL fold and the in-memory fold the unit
|
||||
* tests use agree on real Postgres: same days in a non-UTC zone, same
|
||||
* segment edges, same plausibility filter.
|
||||
*/
|
||||
import v8 from "node:v8";
|
||||
import vm from "node:vm";
|
||||
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
import { buildGradedSeriesWithRollups } from "@/lib/insights/graded-series";
|
||||
import { readDayAggregates } from "@/lib/measurements/day-aggregates";
|
||||
import { foldDayAggregates } from "@/lib/measurements/__tests__/fake-day-aggregates";
|
||||
import { localHmAsUtc } from "@/lib/tz/local-day";
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
const prisma = getPrismaClient();
|
||||
const DENSE_USER = "dense-pulse-owner";
|
||||
const PARITY_USER = "dense-parity-owner";
|
||||
const DAYS = 91;
|
||||
const ROWS = DAYS * 24 * 60;
|
||||
const NOW = new Date("2026-09-20T18:00:00.000Z");
|
||||
|
||||
const forceGc = ((): (() => void) => {
|
||||
v8.setFlagsFromString("--expose-gc");
|
||||
const gc = vm.runInNewContext("gc") as () => void;
|
||||
v8.setFlagsFromString("--no-expose-gc");
|
||||
return gc;
|
||||
})();
|
||||
|
||||
function liveHeapBytes(): number {
|
||||
forceGc();
|
||||
forceGc();
|
||||
return process.memoryUsage().heapUsed;
|
||||
}
|
||||
|
||||
const mb = (bytes: number): string => (bytes / 1024 / 1024).toFixed(1);
|
||||
|
||||
function report(line: string): void {
|
||||
process.stderr.write(`[dense-read] ${line}\n`);
|
||||
}
|
||||
|
||||
/** Run `read`, returning what it holds, its transient peak and its time. */
|
||||
async function measure<T>(read: () => Promise<T>): Promise<{
|
||||
result: T;
|
||||
heldBytes: number;
|
||||
transientBytes: number;
|
||||
ms: number;
|
||||
}> {
|
||||
const before = liveHeapBytes();
|
||||
const t0 = performance.now();
|
||||
const result = await read();
|
||||
const ms = performance.now() - t0;
|
||||
const transientBytes = process.memoryUsage().heapUsed - before;
|
||||
const heldBytes = liveHeapBytes() - before;
|
||||
return { result, heldBytes, transientBytes, ms };
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
await truncateAllTables(prisma);
|
||||
await prisma.user.create({
|
||||
data: {
|
||||
id: DENSE_USER,
|
||||
username: DENSE_USER,
|
||||
timezone: "Europe/Berlin",
|
||||
},
|
||||
});
|
||||
await prisma.$executeRawUnsafe(
|
||||
`INSERT INTO measurements (
|
||||
id, user_id, type, value, unit, source, measured_at,
|
||||
created_at, updated_at, sync_version)
|
||||
SELECT
|
||||
'dp' || lpad(g::text, 8, '0'),
|
||||
$1,
|
||||
'PULSE'::measurement_type,
|
||||
55 + (g % 50),
|
||||
'bpm',
|
||||
'GOOGLE_HEALTH'::measurement_source,
|
||||
($2::timestamptz AT TIME ZONE 'UTC') - (g * interval '1 minute'),
|
||||
now(), now(), 1
|
||||
FROM generate_series(1, $3::int) g`,
|
||||
DENSE_USER,
|
||||
NOW.toISOString(),
|
||||
ROWS,
|
||||
);
|
||||
await prisma.$executeRawUnsafe(`ANALYZE measurements`);
|
||||
}, 300_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await truncateAllTables(prisma);
|
||||
});
|
||||
|
||||
describe("pulse status over a dense stream (#1023)", () => {
|
||||
it("builds the graded series without materialising the stream", async () => {
|
||||
const graded = await measure(() =>
|
||||
buildGradedSeriesWithRollups(DENSE_USER, "PULSE", NOW),
|
||||
);
|
||||
const series = graded.result;
|
||||
// Every reading in the 91-day window is accounted for, one bucket per day
|
||||
// or week, not one object per reading. The oldest reading sits exactly 91
|
||||
// days back, which the graded split files under the older slice.
|
||||
const counted = [...series.recent, ...series.weekly].reduce(
|
||||
(sum, b) => sum + b.n,
|
||||
0,
|
||||
);
|
||||
expect(counted).toBe(ROWS - 1);
|
||||
expect(series.recent.length).toBeLessThanOrEqual(22);
|
||||
expect(series.weekly.length).toBeLessThanOrEqual(12);
|
||||
|
||||
const raw = await measure(() =>
|
||||
prisma.measurement.findMany({
|
||||
where: {
|
||||
userId: DENSE_USER,
|
||||
type: "PULSE",
|
||||
deletedAt: null,
|
||||
measuredAt: { gte: new Date(NOW.getTime() - DAYS * 86_400_000) },
|
||||
},
|
||||
orderBy: { measuredAt: "asc" },
|
||||
select: { measuredAt: true, value: true },
|
||||
}),
|
||||
);
|
||||
expect(raw.result.length).toBe(ROWS);
|
||||
|
||||
report(
|
||||
`graded series (per-day fold, cold tier): held ${mb(graded.heldBytes)} MB, ` +
|
||||
`transient ${mb(graded.transientBytes)} MB, ${graded.ms.toFixed(0)} ms`,
|
||||
);
|
||||
report(
|
||||
`raw 91-day findMany (the old read): held ${mb(raw.heldBytes)} MB, ` +
|
||||
`transient ${mb(raw.transientBytes)} MB, ${raw.ms.toFixed(0)} ms, ` +
|
||||
`${raw.result.length} rows`,
|
||||
);
|
||||
|
||||
// The fold holds a few kilobytes; the raw read holds the stream. The
|
||||
// transient figure is the one that killed the worker: the old builder
|
||||
// held little once it returned but allocated the whole stream on the
|
||||
// way (62 MB on this fixture, against 1.6 MB for the fold).
|
||||
expect(graded.heldBytes).toBeLessThan(2 * 1024 * 1024);
|
||||
expect(graded.transientBytes).toBeLessThan(12 * 1024 * 1024);
|
||||
expect(raw.heldBytes).toBeGreaterThan(20 * Math.max(graded.heldBytes, 1));
|
||||
}, 120_000);
|
||||
|
||||
it("the SQL fold agrees with the in-memory fold on real Postgres", async () => {
|
||||
const tz = "Asia/Kolkata";
|
||||
await prisma.user.create({
|
||||
data: { id: PARITY_USER, username: PARITY_USER, timezone: tz },
|
||||
});
|
||||
const day = new Date("2026-09-10T12:00:00Z");
|
||||
const edge = new Date(NOW.getTime() - 21 * 86_400_000);
|
||||
const rows = [
|
||||
{ at: localHmAsUtc(day, tz, 0, 5), value: 61 },
|
||||
{ at: localHmAsUtc(day, tz, 23, 55), value: 99 },
|
||||
{ at: localHmAsUtc(day, tz, 12, 0), value: 111_287_531 }, // impossible
|
||||
{ at: edge, value: 70 },
|
||||
{ at: new Date(edge.getTime() + 1), value: 72 },
|
||||
{ at: new Date(NOW.getTime() - 200 * 86_400_000), value: 58 },
|
||||
];
|
||||
await prisma.measurement.createMany({
|
||||
data: rows.map((r, i) => ({
|
||||
id: `parity-${i}`,
|
||||
userId: PARITY_USER,
|
||||
type: "PULSE" as const,
|
||||
value: r.value,
|
||||
unit: "bpm",
|
||||
source: "MANUAL" as const,
|
||||
measuredAt: r.at,
|
||||
})),
|
||||
});
|
||||
const opts = {
|
||||
since: new Date(NOW.getTime() - 400 * 86_400_000),
|
||||
until: NOW,
|
||||
timeZone: tz,
|
||||
valueRange: { min: 20, max: 300 },
|
||||
segmentStarts: [edge, new Date(NOW.getTime() - 91 * 86_400_000)],
|
||||
};
|
||||
const sql = await readDayAggregates({
|
||||
userId: PARITY_USER,
|
||||
type: "PULSE",
|
||||
...opts,
|
||||
});
|
||||
const memory = foldDayAggregates(
|
||||
rows.map((r) => ({ measuredAt: r.at, value: r.value })),
|
||||
opts,
|
||||
);
|
||||
expect(sql).toEqual(memory);
|
||||
// Sanity: the impossible value was dropped and the edge row split off.
|
||||
expect(sql.reduce((n, r) => n + r.n, 0)).toBe(5);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user