mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 13:01:00 +00:00
fix(rollups): choose the canonical source per day inside weekly, monthly and yearly buckets
The rollup writer keeps one row per bucket and source. Collapsing the stored WEEK / MONTH / YEAR rows picked one source for the whole bucket, so a month of manual blood-pressure readings plus one synced reading reported the synced day alone, and a step month that switched devices lost every day the losing device carried. The live weekly and monthly chart fold made the same pick. Every coarse bucket is now folded from the canonical DAY buckets: the source ladder picks one source per day, then the week, month or year counts every one of its days. Spread, slope and r² are composed from the summed day accumulators. The live fold picks per local day before folding into weeks and months. Ties between sources without a ladder rank break alphabetically on the SQL paths too, matching the rollup reader.
This commit is contained in:
Marc-André Bombeck committed
1 parent
4af6b87a2a
commit
d51ba9f134
9 files changed
+968
-534
No files matched your search
@@ -147,7 +147,7 @@ export function canonicalMeasurementsCte(
|
||||
WHERE "user_id" = $1
|
||||
AND "deleted_at" IS NULL
|
||||
${sinceInner}
|
||||
ORDER BY "type", date_trunc('day', "measured_at"), (${rankUnqualified}), "source"
|
||||
ORDER BY "type", date_trunc('day', "measured_at"), (${rankUnqualified}), "source"::text
|
||||
) c
|
||||
ON c.t = mm."type"
|
||||
AND c.d = date_trunc('day', mm."measured_at")
|
||||
|
||||
@@ -78,19 +78,43 @@ const dayMs = 24 * 60 * 60 * 1000;
|
||||
const TZ_USER = "user-tz";
|
||||
|
||||
function rollupRow(daysAgo: number, mean: number, count: number) {
|
||||
const now = new Date();
|
||||
return {
|
||||
bucketStart: new Date(Date.now() - daysAgo * dayMs),
|
||||
// A stored DAY bucket starts at UTC midnight.
|
||||
bucketStart: new Date(
|
||||
Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()) -
|
||||
daysAgo * dayMs,
|
||||
),
|
||||
source: "MANUAL",
|
||||
count,
|
||||
mean,
|
||||
minValue: mean,
|
||||
maxValue: mean,
|
||||
sumValue: count * mean,
|
||||
sd: 0,
|
||||
slope: 0,
|
||||
r2: 0,
|
||||
sumX: null,
|
||||
sumXy: null,
|
||||
sumXx: null,
|
||||
sumYy: null,
|
||||
computedAt: new Date(),
|
||||
};
|
||||
}
|
||||
|
||||
/** Keep the rows a DAY read's `bucketStart` window would return. */
|
||||
function inWindow(
|
||||
rows: ReturnType<typeof rollupRow>[],
|
||||
where: { bucketStart: { gte: Date; lt?: Date } },
|
||||
) {
|
||||
return rows.filter(
|
||||
(r) =>
|
||||
r.bucketStart >= where.bucketStart.gte &&
|
||||
(where.bucketStart.lt === undefined ||
|
||||
r.bucketStart < where.bucketStart.lt),
|
||||
);
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetAllMocks();
|
||||
prismaMock.user.findUnique.mockResolvedValue({
|
||||
@@ -122,22 +146,25 @@ describe("extractFeatures — v1.4.36 W3 bucketed payload", () => {
|
||||
});
|
||||
|
||||
it("attaches DAY / WEEK / MONTH buckets from measurement_rollups when includeRaw=true", async () => {
|
||||
// Two WEIGHT DAY buckets in the 0-90d window and one MONTH bucket
|
||||
// in the 365-1825d window. The reader is called once per
|
||||
// (type, granularity) combination — return data only for the two
|
||||
// we care about, empty for the rest.
|
||||
// Two WEIGHT DAY buckets in the 0-90d window and one day in the
|
||||
// 365-1825d window, which surfaces as a MONTH bucket. Every window is
|
||||
// folded from the DAY tier, so the mock answers DAY reads by window.
|
||||
const weightDays = [
|
||||
rollupRow(10, 82.5, 2),
|
||||
rollupRow(20, 82.7, 1),
|
||||
rollupRow(400, 85.1, 28),
|
||||
];
|
||||
prismaMock.measurementRollup.findMany.mockImplementation(
|
||||
async (args: { where: { type: string; granularity: string } }) => {
|
||||
if (args.where.type === "WEIGHT" && args.where.granularity === "DAY") {
|
||||
return [rollupRow(10, 82.5, 2), rollupRow(20, 82.7, 1)];
|
||||
}
|
||||
if (
|
||||
args.where.type === "WEIGHT" &&
|
||||
args.where.granularity === "MONTH"
|
||||
) {
|
||||
return [rollupRow(400, 85.1, 28)];
|
||||
}
|
||||
return [];
|
||||
async (args: {
|
||||
where: {
|
||||
type: string;
|
||||
granularity: string;
|
||||
bucketStart: { gte: Date; lt?: Date };
|
||||
};
|
||||
}) => {
|
||||
if (args.where.type !== "WEIGHT") return [];
|
||||
expect(args.where.granularity).toBe("DAY");
|
||||
return inWindow(weightDays, args.where);
|
||||
},
|
||||
);
|
||||
|
||||
@@ -188,19 +215,17 @@ describe("extractFeatures — v1.4.36 W3 bucketed payload", () => {
|
||||
{ timeout: 30_000 },
|
||||
async () => {
|
||||
// Fabricate an absurdly long bucket list so the JSON dump crosses
|
||||
// the ceiling. ~200 KB per series × 28 series ≈ ~5.6 MB.
|
||||
const giant = new Array(200_000).fill(null).map((_, i) => ({
|
||||
bucketStart: new Date(Date.now() - i * dayMs),
|
||||
count: i,
|
||||
mean: i,
|
||||
minValue: i,
|
||||
maxValue: i,
|
||||
sd: 0,
|
||||
slope: 0,
|
||||
r2: 0,
|
||||
computedAt: new Date(),
|
||||
}));
|
||||
prismaMock.measurementRollup.findMany.mockResolvedValue(giant);
|
||||
// the ceiling: every DAY read answers 200 000 distinct buckets inside
|
||||
// its own window, one second apart.
|
||||
prismaMock.measurementRollup.findMany.mockImplementation(
|
||||
async (args: { where: { bucketStart: { gte: Date } } }) =>
|
||||
new Array(200_000).fill(null).map((_, i) => ({
|
||||
...rollupRow(0, i, i),
|
||||
bucketStart: new Date(
|
||||
args.where.bucketStart.gte.getTime() + i * 1000,
|
||||
),
|
||||
})),
|
||||
);
|
||||
|
||||
await expect(extractFeatures("user-1", true)).rejects.toThrow(
|
||||
FeaturesPayloadTooLargeError,
|
||||
|
||||
@@ -62,7 +62,8 @@ vi.mock("@/lib/tz/resolver", async () => {
|
||||
return { ...actual, resolveUserTimezone: vi.fn() };
|
||||
});
|
||||
|
||||
vi.mock("@/lib/rollups/measurement-read", () => ({
|
||||
vi.mock("@/lib/rollups/measurement-read", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("@/lib/rollups/measurement-read")>()),
|
||||
loadUserSourcePriority: vi.fn(),
|
||||
}));
|
||||
|
||||
|
||||
@@ -159,9 +159,10 @@ const TRUNC_UNIT: Record<Exclude<AggregateGrain, "raw">, string> = {
|
||||
* the bucket's own day for every offset, including half-hour and negative
|
||||
* ones, and across a DST change.
|
||||
*
|
||||
* Per bucket the overlapping sources collapse to the ladder-canonical one
|
||||
* before the fold (the same pick the rollup reader makes), cumulative types
|
||||
* SUM and every other type averages. The result set is at most `cap` rows,
|
||||
* Per local day the overlapping sources collapse to the ladder-canonical one
|
||||
* (the same pick the rollup reader makes), and the days then fold into the
|
||||
* bucket: cumulative types SUM and every other type averages over every
|
||||
* reading of the canonical days. The result set is at most `cap` rows,
|
||||
* however dense the underlying stream: the fold runs in Postgres.
|
||||
*
|
||||
* `type` narrows to one type; `null` folds every type in the window (the
|
||||
@@ -190,9 +191,7 @@ export async function readLiveBuckets(opts: {
|
||||
const cumulativeList = [...CUMULATIVE_HK_TYPES]
|
||||
.map((t) => `'${t}'`)
|
||||
.join(",");
|
||||
const aggregator = Prisma.raw(
|
||||
`(CASE WHEN m."type"::text IN (${cumulativeList}) THEN SUM(m."value") ELSE AVG(m."value") END)::double precision`,
|
||||
);
|
||||
const isCumulative = Prisma.raw(`c."type"::text IN (${cumulativeList})`);
|
||||
const rankRaw = Prisma.raw(
|
||||
buildSourceRankCase(priorityJson, 'p."type"', 'p."source"'),
|
||||
);
|
||||
@@ -202,14 +201,14 @@ export async function readLiveBuckets(opts: {
|
||||
// measured_at is a UTC wall-clock timestamp: pin it to UTC, re-read it in
|
||||
// the user's zone and truncate there.
|
||||
//
|
||||
// Fold first, pick second. Every source is folded per bucket in one hashed
|
||||
// pass (a few rows per day, however dense the stream), and only then does
|
||||
// each bucket keep its ladder-canonical source. The fold of the canonical
|
||||
// source's rows is exactly the fold of that source's group, so the result
|
||||
// is the same as collapsing the rows first; what changes is the cost. A
|
||||
// year of per-minute heart rate is half a million rows, and ranking them
|
||||
// before the fold sorted them twice (about 0.55 s); folding first never
|
||||
// sorts more than one row per source per day.
|
||||
// The canonical source is chosen per local DAY, whatever the grain: each
|
||||
// source is folded per day in one hashed pass (a few rows per day, however
|
||||
// dense the stream), each day keeps its ladder-canonical source, and only
|
||||
// then are the days folded into the week or month. Picking the source per
|
||||
// week or month instead dropped every day the losing source carried, so a
|
||||
// month of manual readings plus one synced reading reported one day.
|
||||
// Folding first and picking second is the same result as collapsing the
|
||||
// raw rows first; it just never sorts more than one row per source per day.
|
||||
const buckets = await prisma.$queryRaw<
|
||||
Array<{
|
||||
type: string;
|
||||
@@ -224,8 +223,8 @@ export async function readLiveBuckets(opts: {
|
||||
SELECT
|
||||
m."type",
|
||||
m."source",
|
||||
date_trunc(${unit}, (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS d,
|
||||
${aggregator} AS avg,
|
||||
date_trunc('day', (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${timeZone}) AS day,
|
||||
SUM(m."value")::double precision AS total,
|
||||
COUNT(*)::int AS cnt,
|
||||
MIN(m."value") AS min_value,
|
||||
MAX(m."value") AS max_value
|
||||
@@ -238,20 +237,31 @@ export async function readLiveBuckets(opts: {
|
||||
GROUP BY m."type", m."source", 3
|
||||
),
|
||||
canon AS (
|
||||
SELECT DISTINCT ON (p."type", p.d)
|
||||
p."type", p.d, p.avg, p.cnt, p.min_value, p.max_value
|
||||
SELECT DISTINCT ON (p."type", p.day)
|
||||
p."type", p.day, p.total, p.cnt, p.min_value, p.max_value
|
||||
FROM per_source p
|
||||
ORDER BY p."type", p.d, (${rankRaw}), p."source"
|
||||
ORDER BY p."type", p.day, (${rankRaw}), p."source"::text
|
||||
),
|
||||
folded AS (
|
||||
SELECT
|
||||
c."type",
|
||||
date_trunc(${unit}, c.day) AS d,
|
||||
(CASE WHEN ${isCumulative} THEN SUM(c.total) ELSE SUM(c.total) / SUM(c.cnt) END)::double precision AS avg,
|
||||
SUM(c.cnt)::int AS cnt,
|
||||
MIN(c.min_value) AS min_value,
|
||||
MAX(c.max_value) AS max_value
|
||||
FROM canon c
|
||||
GROUP BY c."type", 2
|
||||
)
|
||||
SELECT
|
||||
c."type"::text AS type,
|
||||
c.d AT TIME ZONE ${timeZone} AS bucket_start,
|
||||
c.avg,
|
||||
c.cnt,
|
||||
c.min_value,
|
||||
c.max_value
|
||||
FROM canon c
|
||||
ORDER BY c.d ASC
|
||||
f."type"::text AS type,
|
||||
f.d AT TIME ZONE ${timeZone} AS bucket_start,
|
||||
f.avg,
|
||||
f.cnt,
|
||||
f.min_value,
|
||||
f.max_value
|
||||
FROM folded f
|
||||
ORDER BY f.d ASC
|
||||
LIMIT ${cap}
|
||||
`;
|
||||
return buckets.map((b) => {
|
||||
|
||||
@@ -5,13 +5,12 @@
|
||||
*
|
||||
* `prisma.measurementRollup.findMany` is mocked at the module level
|
||||
* so the test pins:
|
||||
* - each per-granularity reader filters by the correct
|
||||
* `granularity` literal,
|
||||
* - every granularity is folded from the canonical DAY buckets (the
|
||||
* source is chosen per DAY, never once per week / month / year),
|
||||
* - empty rollups return `null` so the caller can branch on
|
||||
* coverage miss,
|
||||
* - the auto-router walks YEAR → MONTH → WEEK → DAY and stops at
|
||||
* the first granularity whose floor the window clears AND which
|
||||
* has coverage,
|
||||
* - the auto-router picks the coarsest granularity whose floor the
|
||||
* window clears,
|
||||
* - `aggregateWmyBuckets` composes `count / min / max / mean / sum`
|
||||
* linearly across coarser buckets (the same compositional
|
||||
* contract `rollup-read.ts:aggregateBuckets` carries for DAY).
|
||||
@@ -86,7 +85,69 @@ afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
/** One stored DAY rollup row, per source, as `findMany` returns it. */
|
||||
function dayRow(
|
||||
day: string,
|
||||
source: string,
|
||||
count: number,
|
||||
mean: number,
|
||||
extra: Partial<{ minValue: number; maxValue: number }> = {},
|
||||
) {
|
||||
const bucketStart = new Date(`${day}T00:00:00.000Z`);
|
||||
// Accumulators for `count` readings of `mean` at the day's midday.
|
||||
const x = bucketStart.getTime() / DAY_MS_ROWS - 18262 + 0.5;
|
||||
return {
|
||||
bucketStart,
|
||||
source,
|
||||
count,
|
||||
mean,
|
||||
minValue: extra.minValue ?? mean,
|
||||
maxValue: extra.maxValue ?? mean,
|
||||
sumValue: count * mean,
|
||||
sd: 0,
|
||||
slope: null,
|
||||
r2: null,
|
||||
sumX: count * x,
|
||||
sumXy: count * x * mean,
|
||||
sumXx: count * x * x,
|
||||
sumYy: count * mean * mean,
|
||||
computedAt: new Date("2026-06-01T00:00:00.000Z"),
|
||||
};
|
||||
}
|
||||
const DAY_MS_ROWS = 86_400_000;
|
||||
|
||||
/** `n` consecutive days from `start`, one row each. */
|
||||
function days(
|
||||
start: string,
|
||||
n: number,
|
||||
source: string,
|
||||
count: number,
|
||||
mean: number,
|
||||
) {
|
||||
const out = [];
|
||||
const t0 = new Date(`${start}T00:00:00.000Z`).getTime();
|
||||
for (let i = 0; i < n; i++) {
|
||||
out.push(
|
||||
dayRow(
|
||||
new Date(t0 + i * DAY_MS_ROWS).toISOString().slice(0, 10),
|
||||
source,
|
||||
count,
|
||||
mean,
|
||||
),
|
||||
);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
describe("readBestGranularityRollups", () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
vi.setSystemTime(new Date("2026-01-15T12:00:00.000Z"));
|
||||
});
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("returns null on a non-positive window", async () => {
|
||||
expect(await readBestGranularityRollups("user", "WEIGHT", 0)).toBeNull();
|
||||
expect(await readBestGranularityRollups("user", "WEIGHT", -10)).toBeNull();
|
||||
@@ -94,65 +155,111 @@ describe("readBestGranularityRollups", () => {
|
||||
expect(findMany).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("routes a 90-day window to DAY (every coarser floor is too low)", async () => {
|
||||
findMany.mockResolvedValueOnce([bucket("2026-04-01T00:00:00.000Z")]);
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 90);
|
||||
expect(result).not.toBeNull();
|
||||
expect(result?.granularity).toBe("DAY");
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("DAY");
|
||||
});
|
||||
it.each([
|
||||
[90, "DAY"],
|
||||
[365, "MONTH"],
|
||||
[540, "MONTH"],
|
||||
[1095, "YEAR"],
|
||||
[120, "WEEK"],
|
||||
] as const)(
|
||||
"routes a %i-day window to %s, folded from the DAY tier",
|
||||
async (windowDays, granularity) => {
|
||||
findMany.mockResolvedValueOnce([dayRow("2025-12-20", "MANUAL", 1, 80)]);
|
||||
const result = await readBestGranularityRollups(
|
||||
"user",
|
||||
"WEIGHT",
|
||||
windowDays,
|
||||
);
|
||||
expect(result?.granularity).toBe(granularity);
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("DAY");
|
||||
},
|
||||
);
|
||||
|
||||
it("routes a 365-day window to MONTH (clears the 62-day MONTH floor)", async () => {
|
||||
findMany.mockResolvedValueOnce([bucket("2026-01-01T00:00:00.000Z")]);
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 365);
|
||||
expect(result).not.toBeNull();
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("MONTH");
|
||||
});
|
||||
|
||||
it("routes a 1095-day window to YEAR (clears the 730-day YEAR floor)", async () => {
|
||||
findMany.mockResolvedValueOnce([bucket("2024-01-01T00:00:00.000Z")]);
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 1095);
|
||||
expect(result).not.toBeNull();
|
||||
expect(result?.granularity).toBe("YEAR");
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("YEAR");
|
||||
});
|
||||
|
||||
it("falls back to MONTH when YEAR has no coverage", async () => {
|
||||
// 1095-day window asks for YEAR first; coverage miss → MONTH.
|
||||
findMany
|
||||
.mockResolvedValueOnce([]) // YEAR miss
|
||||
.mockResolvedValueOnce([bucket("2024-01-01T00:00:00.000Z")]); // MONTH hit
|
||||
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 1095);
|
||||
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("YEAR");
|
||||
expect(findMany.mock.calls[1][0].where.granularity).toBe("MONTH");
|
||||
});
|
||||
|
||||
it("falls all the way through to DAY when every coarser tier misses", async () => {
|
||||
findMany
|
||||
.mockResolvedValueOnce([]) // YEAR miss
|
||||
.mockResolvedValueOnce([]) // MONTH miss
|
||||
.mockResolvedValueOnce([]) // WEEK miss
|
||||
.mockResolvedValueOnce([bucket("2024-01-01T00:00:00.000Z")]); // DAY hit
|
||||
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 1095);
|
||||
|
||||
expect(result?.granularity).toBe("DAY");
|
||||
expect(findMany).toHaveBeenCalledTimes(4);
|
||||
});
|
||||
|
||||
it("returns null when no granularity carries any coverage", async () => {
|
||||
it("returns null when the window holds no DAY buckets", async () => {
|
||||
findMany.mockResolvedValue([]);
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 1095);
|
||||
expect(result).toBeNull();
|
||||
// YEAR, MONTH, WEEK, DAY all probed.
|
||||
expect(findMany).toHaveBeenCalledTimes(4);
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
// The canonical source is a per-DAY decision. Collapsing the stored
|
||||
// per-source MONTH rows picked WITHINGS for the whole month (it leads the
|
||||
// blood-pressure ladder) and reported its one reading: count 1, mean 120.
|
||||
it("keeps every manual day of a month that also carries one synced day", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
...days("2025-11-01", 20, "MANUAL", 1, 150),
|
||||
dayRow("2025-11-25", "WITHINGS", 1, 120),
|
||||
]);
|
||||
const result = await readBestGranularityRollups(
|
||||
"user",
|
||||
"BLOOD_PRESSURE_SYS",
|
||||
365,
|
||||
);
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
expect(result?.rows).toHaveLength(1);
|
||||
expect(result?.rows[0].count).toBe(21);
|
||||
expect(result?.rows[0].mean).toBeCloseTo((20 * 150 + 120) / 21, 9);
|
||||
expect(result?.rows[0].minValue).toBe(120);
|
||||
expect(result?.rows[0].maxValue).toBe(150);
|
||||
});
|
||||
|
||||
it("sums a step month whose days switch device, one source per day", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
...days("2025-11-01", 10, "FITBIT", 1, 8000),
|
||||
...days("2025-11-11", 20, "APPLE_HEALTH", 1, 8000),
|
||||
]);
|
||||
const result = await readBestGranularityRollups(
|
||||
"user",
|
||||
"ACTIVITY_STEPS",
|
||||
365,
|
||||
);
|
||||
expect(aggregateWmyBuckets(result?.rows ?? []).sum).toBe(240_000);
|
||||
});
|
||||
|
||||
it("still keeps one source on a day both sources carry", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
dayRow("2025-11-03", "APPLE_HEALTH", 1, 9000),
|
||||
dayRow("2025-11-03", "FITBIT", 1, 7000),
|
||||
dayRow("2025-11-04", "FITBIT", 1, 7000),
|
||||
]);
|
||||
const result = await readBestGranularityRollups(
|
||||
"user",
|
||||
"ACTIVITY_STEPS",
|
||||
365,
|
||||
);
|
||||
expect(result?.rows[0].sumValue).toBe(16_000);
|
||||
expect(result?.rows[0].count).toBe(2);
|
||||
});
|
||||
|
||||
it("cuts weeks on the ISO Monday in UTC", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
dayRow("2025-11-02", "MANUAL", 1, 1), // Sunday → week of 2025-10-27
|
||||
dayRow("2025-11-03", "MANUAL", 1, 2), // Monday → week of 2025-11-03
|
||||
dayRow("2025-11-09", "MANUAL", 1, 3), // Sunday → week of 2025-11-03
|
||||
]);
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 120);
|
||||
expect(result?.granularity).toBe("WEEK");
|
||||
expect(result?.rows.map((r) => r.bucketStart.toISOString())).toEqual([
|
||||
"2025-10-27T00:00:00.000Z",
|
||||
"2025-11-03T00:00:00.000Z",
|
||||
]);
|
||||
expect(result?.rows.map((r) => r.count)).toEqual([1, 2]);
|
||||
});
|
||||
|
||||
it("composes the bucket's spread and slope from the day accumulators", async () => {
|
||||
// Three days of one reading each: 70, 72, 74 → slope 2 per day,
|
||||
// population sd sqrt(8/3), r² 1.
|
||||
findMany.mockResolvedValueOnce([
|
||||
dayRow("2025-11-03", "MANUAL", 1, 70),
|
||||
dayRow("2025-11-04", "MANUAL", 1, 72),
|
||||
dayRow("2025-11-05", "MANUAL", 1, 74),
|
||||
]);
|
||||
const result = await readBestGranularityRollups("user", "WEIGHT", 365);
|
||||
const month = result?.rows[0];
|
||||
expect(month?.slope).toBeCloseTo(2, 6);
|
||||
expect(month?.r2).toBeCloseTo(1, 6);
|
||||
expect(month?.sd).toBeCloseTo(Math.sqrt(8 / 3), 6);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -232,30 +339,6 @@ describe("aggregateWmyBuckets", () => {
|
||||
* A floor adjustment must update this single routing contract.
|
||||
*/
|
||||
describe("readBestGranularityRollups — cross-consumer routing parity", () => {
|
||||
it("pins the 90 / 365 / 1095 day routing targets the consumers depend on", async () => {
|
||||
// Three sequential probes with non-empty results so the router
|
||||
// stops at the first matching tier per window.
|
||||
findMany
|
||||
.mockResolvedValueOnce([bucket("2026-02-15T00:00:00.000Z")]) // 90d → DAY
|
||||
.mockResolvedValueOnce([bucket("2025-08-01T00:00:00.000Z")]) // 365d → MONTH
|
||||
.mockResolvedValueOnce([bucket("2024-01-01T00:00:00.000Z")]); // 1095d → YEAR
|
||||
|
||||
const ninety = await readBestGranularityRollups("user", "WEIGHT", 90);
|
||||
const yearLong = await readBestGranularityRollups("user", "WEIGHT", 365);
|
||||
const threeYear = await readBestGranularityRollups("user", "WEIGHT", 1095);
|
||||
|
||||
expect(ninety?.granularity).toBe("DAY");
|
||||
expect(yearLong?.granularity).toBe("MONTH");
|
||||
expect(threeYear?.granularity).toBe("YEAR");
|
||||
// Each consumer-relevant window only fires one round-trip on the
|
||||
// happy path; the per-tier walk is reserved for coverage-miss
|
||||
// scenarios already pinned above.
|
||||
expect(findMany).toHaveBeenCalledTimes(3);
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("DAY");
|
||||
expect(findMany.mock.calls[1][0].where.granularity).toBe("MONTH");
|
||||
expect(findMany.mock.calls[2][0].where.granularity).toBe("YEAR");
|
||||
});
|
||||
|
||||
it("aggregates MONTH buckets to a byte-identical mean compared to the underlying DAY buckets", async () => {
|
||||
// The compositional contract `count / mean` are linearly
|
||||
// composable across granularities — pin it by simulating "MONTH
|
||||
@@ -313,15 +396,14 @@ describe("readTieredRollupSeries", () => {
|
||||
expect(findMany).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reads the MONTH tier for a multi-year window and spans the whole history", async () => {
|
||||
// 10-year window → MONTH tier. The fixture carries buckets from the
|
||||
// OLDEST month through the most recent — the result must keep the
|
||||
// oldest bucket, proving no recent-slice truncation.
|
||||
const months: RollupBucketRow[] = [];
|
||||
it("folds a multi-year window into MONTH buckets and spans the whole history", async () => {
|
||||
// Ten years of first-of-January readings → MONTH tier; the oldest
|
||||
// bucket survives (no recent-slice truncation).
|
||||
const rows = [];
|
||||
for (let y = 2017; y <= 2026; y++) {
|
||||
months.push(bucket(`${y}-01-01T00:00:00.000Z`, { count: 12, mean: 80 }));
|
||||
rows.push(dayRow(`${y}-01-01`, "MANUAL", 12, 80));
|
||||
}
|
||||
findMany.mockResolvedValueOnce(months);
|
||||
findMany.mockResolvedValueOnce(rows);
|
||||
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
@@ -329,28 +411,24 @@ describe("readTieredRollupSeries", () => {
|
||||
...win(3650),
|
||||
});
|
||||
|
||||
expect(result).not.toBeNull();
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("MONTH");
|
||||
// Whole-history coverage: the earliest 2017 bucket survives.
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("DAY");
|
||||
expect(result?.rows[0].measuredAt).toBe("2017-01-01T00:00:00.000Z");
|
||||
expect(result?.rows.at(-1)?.measuredAt).toBe("2026-01-01T00:00:00.000Z");
|
||||
expect(result?.rows.length).toBe(10);
|
||||
});
|
||||
|
||||
// v1.26.0 SEAM-N2 — a historic window entirely in the past must bound the
|
||||
// rollup read on BOTH ends. Pre-fix the reader anchored on `now − windowDays`
|
||||
// and would have queried `bucketStart >= now − 730d` (mid-2024 onward),
|
||||
// returning recent buckets for a 2020–2022 ask.
|
||||
it("bounds the rollup read on BOTH ends for a historic past window", async () => {
|
||||
// rollup read on BOTH ends.
|
||||
it("bounds the DAY read on BOTH ends for a historic past window", async () => {
|
||||
const from = new Date("2020-01-01T00:00:00.000Z");
|
||||
const to = new Date("2022-01-01T00:00:00.000Z");
|
||||
const inWindow: RollupBucketRow[] = [
|
||||
bucket("2020-02-01T00:00:00.000Z", { count: 12, mean: 80 }),
|
||||
bucket("2021-06-01T00:00:00.000Z", { count: 12, mean: 81 }),
|
||||
bucket("2021-12-01T00:00:00.000Z", { count: 12, mean: 82 }),
|
||||
];
|
||||
findMany.mockResolvedValueOnce(inWindow);
|
||||
findMany.mockResolvedValueOnce([
|
||||
dayRow("2020-02-01", "MANUAL", 12, 80),
|
||||
dayRow("2021-06-01", "MANUAL", 12, 81),
|
||||
dayRow("2021-12-01", "MANUAL", 12, 82),
|
||||
]);
|
||||
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
@@ -359,60 +437,20 @@ describe("readTieredRollupSeries", () => {
|
||||
to,
|
||||
});
|
||||
|
||||
// The `bucketStart` filter carries BOTH the requested lower AND upper
|
||||
// bound — not a trailing "to now" slice.
|
||||
const where = findMany.mock.calls[0][0].where;
|
||||
expect(where.bucketStart).toEqual({ gte: from, lte: to });
|
||||
// Every returned bucket_start falls inside the requested window.
|
||||
expect(findMany.mock.calls[0][0].where.bucketStart).toEqual({
|
||||
gte: from,
|
||||
lte: to,
|
||||
});
|
||||
for (const row of result?.rows ?? []) {
|
||||
const t = new Date(row.measuredAt).getTime();
|
||||
expect(t).toBeGreaterThanOrEqual(from.getTime());
|
||||
expect(t).toBeLessThanOrEqual(to.getTime());
|
||||
}
|
||||
// Tier width is 2 years (731 days) → still the MONTH tier for >730 d.
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
});
|
||||
|
||||
// The web-chart "All" path passes `to ≈ now`; prove that case still bounds
|
||||
// on both ends (no regression) — the historic fix must not perturb it.
|
||||
it("bounds a current-window (to = now) read on both ends too", async () => {
|
||||
findMany.mockResolvedValueOnce([bucket("2026-01-01T00:00:00.000Z")]);
|
||||
const { from, to } = win(3650);
|
||||
await readTieredRollupSeries({ userId: "u", type: "WEIGHT", from, to });
|
||||
expect(findMany.mock.calls[0][0].where.bucketStart).toEqual({
|
||||
gte: from,
|
||||
lte: to,
|
||||
});
|
||||
});
|
||||
|
||||
it("reads the WEEK tier for a 1–2 year window", async () => {
|
||||
findMany.mockResolvedValueOnce([bucket("2025-08-04T00:00:00.000Z")]);
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "WEIGHT",
|
||||
...win(540),
|
||||
});
|
||||
// Data span ≈ 22 months → WEEK (the request alone would call for MONTH).
|
||||
expect(result?.granularity).toBe("WEEK");
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("WEEK");
|
||||
});
|
||||
|
||||
it("falls FINER (MONTH → WEEK → DAY) when the target tier has no coverage", async () => {
|
||||
findMany
|
||||
.mockResolvedValueOnce([]) // MONTH miss
|
||||
.mockResolvedValueOnce([]) // WEEK miss
|
||||
.mockResolvedValueOnce([bucket("2024-03-01T00:00:00.000Z")]); // DAY hit
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "WEIGHT",
|
||||
...win(3650),
|
||||
});
|
||||
expect(result?.granularity).toBe("DAY");
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("MONTH");
|
||||
expect(findMany.mock.calls[1][0].where.granularity).toBe("WEEK");
|
||||
expect(findMany.mock.calls[2][0].where.granularity).toBe("DAY");
|
||||
});
|
||||
|
||||
it("returns null when no tier at or below the target carries coverage", async () => {
|
||||
it("returns null when the window holds no DAY buckets", async () => {
|
||||
findMany.mockResolvedValue([]);
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
@@ -420,25 +458,20 @@ describe("readTieredRollupSeries", () => {
|
||||
...win(3650),
|
||||
});
|
||||
expect(result).toBeNull();
|
||||
// MONTH, WEEK, DAY all probed (never coarser than the target).
|
||||
expect(findMany).toHaveBeenCalledTimes(3);
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("surfaces the cumulative summed total for SUM metrics and drops the spread", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
bucket("2025-01-01T00:00:00.000Z", {
|
||||
count: 30,
|
||||
mean: 8000,
|
||||
sumValue: 240_000,
|
||||
minValue: 0,
|
||||
maxValue: 20_000,
|
||||
}),
|
||||
...days("2024-01-01", 30, "APPLE_HEALTH", 1, 8000),
|
||||
dayRow("2026-01-01", "APPLE_HEALTH", 1, 5000),
|
||||
]);
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "ACTIVITY_STEPS",
|
||||
...win(3650),
|
||||
});
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
expect(result?.rows[0].value).toBe(240_000);
|
||||
expect(result?.rows[0].minValue).toBeUndefined();
|
||||
expect(result?.rows[0].maxValue).toBeUndefined();
|
||||
@@ -446,19 +479,17 @@ describe("readTieredRollupSeries", () => {
|
||||
|
||||
it("carries the count-weighted mean + spread for spot metrics", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
bucket("2025-01-01T00:00:00.000Z", {
|
||||
count: 12,
|
||||
mean: 81.5,
|
||||
minValue: 78,
|
||||
maxValue: 85,
|
||||
}),
|
||||
dayRow("2024-01-02", "MANUAL", 2, 80, { minValue: 78, maxValue: 82 }),
|
||||
dayRow("2024-01-03", "MANUAL", 1, 84, { minValue: 84, maxValue: 85 }),
|
||||
dayRow("2026-01-01", "MANUAL", 1, 80),
|
||||
]);
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "WEIGHT",
|
||||
...win(3650),
|
||||
});
|
||||
expect(result?.rows[0].value).toBe(81.5);
|
||||
expect(result?.rows[0].value).toBeCloseTo((2 * 80 + 84) / 3, 9);
|
||||
expect(result?.rows[0].count).toBe(3);
|
||||
expect(result?.rows[0].minValue).toBe(78);
|
||||
expect(result?.rows[0].maxValue).toBe(85);
|
||||
});
|
||||
@@ -466,93 +497,40 @@ describe("readTieredRollupSeries", () => {
|
||||
// v1.37.29 — the tier keys off the ACTUAL data span, not the requested
|
||||
// window width. The "All" tab always requests ~3650 days, so pre-fix a
|
||||
// record whose history spans four months came back as four MONTH means
|
||||
// while the chart captioned them from the real span — "weekly average"
|
||||
// over monthly buckets.
|
||||
// while the chart captioned them from the real span.
|
||||
describe("span refinement", () => {
|
||||
it("re-reads at the tier the real span calls for (4 months of data on an All request → DAY)", async () => {
|
||||
const months = [
|
||||
bucket("2026-02-01T00:00:00.000Z"),
|
||||
bucket("2026-03-01T00:00:00.000Z"),
|
||||
bucket("2026-04-01T00:00:00.000Z"),
|
||||
bucket("2026-05-01T00:00:00.000Z"),
|
||||
];
|
||||
const days = [
|
||||
bucket("2026-02-03T00:00:00.000Z"),
|
||||
bucket("2026-03-14T00:00:00.000Z"),
|
||||
bucket("2026-04-09T00:00:00.000Z"),
|
||||
bucket("2026-05-28T00:00:00.000Z"),
|
||||
];
|
||||
findMany.mockResolvedValueOnce(months).mockResolvedValueOnce(days);
|
||||
|
||||
it("serves DAY rows for four months of data on an All request", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
dayRow("2026-02-03", "MANUAL", 1, 80),
|
||||
dayRow("2026-03-14", "MANUAL", 1, 80),
|
||||
dayRow("2026-04-09", "MANUAL", 1, 80),
|
||||
dayRow("2026-05-28", "MANUAL", 1, 80),
|
||||
]);
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "WEIGHT",
|
||||
...win(3650),
|
||||
});
|
||||
|
||||
// Span of the MONTH rows ≈ 89 days → the display ladder wants DAY,
|
||||
// and the refinement jumps straight there (no WEEK stop-over).
|
||||
expect(findMany.mock.calls[0][0].where.granularity).toBe("MONTH");
|
||||
expect(findMany.mock.calls[1][0].where.granularity).toBe("DAY");
|
||||
expect(findMany).toHaveBeenCalledTimes(2);
|
||||
expect(result?.granularity).toBe("DAY");
|
||||
expect(result?.rows).toHaveLength(4);
|
||||
expect(result?.rows[0].measuredAt).toBe("2026-02-03T00:00:00.000Z");
|
||||
});
|
||||
|
||||
it("refines to WEEK for a span between one and two years", async () => {
|
||||
const months = [
|
||||
bucket("2025-01-01T00:00:00.000Z"),
|
||||
bucket("2026-05-01T00:00:00.000Z"),
|
||||
];
|
||||
const weeks = [
|
||||
bucket("2024-12-30T00:00:00.000Z"),
|
||||
bucket("2026-04-27T00:00:00.000Z"),
|
||||
];
|
||||
findMany.mockResolvedValueOnce(months).mockResolvedValueOnce(weeks);
|
||||
|
||||
it("serves WEEK rows for a span between one and two years", async () => {
|
||||
findMany.mockResolvedValueOnce([
|
||||
dayRow("2025-01-01", "MANUAL", 1, 80),
|
||||
dayRow("2026-05-01", "MANUAL", 1, 80),
|
||||
]);
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "WEIGHT",
|
||||
...win(3650),
|
||||
});
|
||||
|
||||
expect(findMany.mock.calls[1][0].where.granularity).toBe("WEEK");
|
||||
expect(result?.granularity).toBe("WEEK");
|
||||
});
|
||||
|
||||
it("keeps the coarser rows when the finer tier has no coverage", async () => {
|
||||
const months = [
|
||||
bucket("2026-01-01T00:00:00.000Z"),
|
||||
bucket("2026-05-01T00:00:00.000Z"),
|
||||
];
|
||||
findMany.mockResolvedValueOnce(months).mockResolvedValueOnce([]);
|
||||
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "WEIGHT",
|
||||
...win(3650),
|
||||
});
|
||||
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
expect(result?.rows).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("does not probe when the rows already span the requested tier", async () => {
|
||||
const months = [
|
||||
bucket("2018-01-01T00:00:00.000Z"),
|
||||
bucket("2026-05-01T00:00:00.000Z"),
|
||||
];
|
||||
findMany.mockResolvedValueOnce(months);
|
||||
|
||||
const result = await readTieredRollupSeries({
|
||||
userId: "u",
|
||||
type: "WEIGHT",
|
||||
...win(3650),
|
||||
});
|
||||
|
||||
expect(findMany).toHaveBeenCalledTimes(1);
|
||||
expect(result?.granularity).toBe("MONTH");
|
||||
expect(result?.rows.map((r) => r.measuredAt)).toEqual([
|
||||
"2024-12-30T00:00:00.000Z",
|
||||
"2026-04-27T00:00:00.000Z",
|
||||
]);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,71 +1,47 @@
|
||||
/**
|
||||
* v1.4.39 W-WMY — WEEK / MONTH / YEAR rollup readers.
|
||||
* WEEK / MONTH / YEAR readers over the measurement rollup tier.
|
||||
*
|
||||
* Background — the read-side gap
|
||||
* ------------------------------
|
||||
* Every measurement write fans out via `recomputeBucketsForMeasurement`
|
||||
* into a synchronous DAY upsert plus pg-boss-queued WEEK / MONTH / YEAR
|
||||
* recomputes. The boot-time backfill (`enqueueBootTimeRollupBackfill`)
|
||||
* mints the full WEEK / MONTH / YEAR history once per uncovered user.
|
||||
*
|
||||
* As of v1.4.38 the only reader that consults `measurement_rollups`
|
||||
* (`rollup-read.ts`, plus the per-fast-path branches in
|
||||
* `summaries-slice`, `comprehensive-aggregator`, etc.) caps at DAY and
|
||||
* a trailing 90-day window. The WEEK / MONTH / YEAR buckets sit in
|
||||
* Postgres as pure write amplification — populated every write,
|
||||
* surfaced nowhere on read. The v1.4.38 perf audit
|
||||
* (`.planning/round-v1438-perf-analysis.md` §2 + §5 P6) calls this out
|
||||
* as the largest unused investment in the rollup tier.
|
||||
*
|
||||
* What this module adds
|
||||
* ---------------------
|
||||
* Helpers that read the trailing N-bucket window for a single
|
||||
* `(userId, type)` pair at a chosen granularity, plus an auto-router
|
||||
* that picks the largest granularity that still resolves the requested
|
||||
* window. The shape returned is identical to `readRollupBuckets` so
|
||||
* callers can interleave WEEK / MONTH / YEAR rows with DAY rows
|
||||
* downstream without branching on the source granularity.
|
||||
* `(userId, type)` pair at a chosen granularity, plus an auto-router that
|
||||
* picks the largest granularity that still resolves the requested window.
|
||||
* The shape returned is identical to `readRollupBuckets` so callers can
|
||||
* interleave WEEK / MONTH / YEAR rows with DAY rows downstream without
|
||||
* branching on the source granularity.
|
||||
*
|
||||
* Built from the DAY tier
|
||||
* -----------------------
|
||||
* Every coarse bucket is folded from the canonical DAY buckets
|
||||
* (`readCanonicalRollupBuckets`): the source-priority ladder picks one
|
||||
* source per DAY, and the week, month or year then counts every one of its
|
||||
* days. The stored per-source WEEK / MONTH / YEAR rows cannot express that:
|
||||
* collapsing them picks one source for the whole bucket, so a month of
|
||||
* manual readings plus one synced reading reported only the synced day.
|
||||
*
|
||||
* Compositional contract
|
||||
* ----------------------
|
||||
* `count / min / max / mean / sumValue` are linearly composable across
|
||||
* any granularity — aggregating these stats over WEEK buckets returns
|
||||
* the same numbers as aggregating the underlying DAY buckets or the
|
||||
* raw measurements. `sd / slope / r2` are NOT linearly composable; the
|
||||
* routed callers either fall back to live SQL for those or accept the
|
||||
* per-bucket stat as-is. The auto-router is a granularity selector
|
||||
* only — it does not attempt to reconstruct slope across coarser
|
||||
* buckets.
|
||||
* `count / min / max / mean / sumValue` compose exactly across days.
|
||||
* `sd / slope / r2` are composed from the summed regression accumulators,
|
||||
* which is exact over the canonical rows too; a bucket with any day that
|
||||
* lacks the accumulators reports them as `null`.
|
||||
*
|
||||
* Coverage-miss semantics
|
||||
* -----------------------
|
||||
* Each reader returns `null` when the requested granularity yields
|
||||
* zero rows for `(userId, type, since)`. The caller decides whether
|
||||
* that is a real "no data" case (the user has not logged this type in
|
||||
* the window) or a coverage miss the boot-backfill / worker has not
|
||||
* caught up on. The default policy is to fall back to a finer
|
||||
* granularity (MONTH → WEEK → DAY) before giving up — see
|
||||
* `readBestGranularityRollups`.
|
||||
*
|
||||
* Scope vs. `rollup-read.ts`
|
||||
* --------------------------
|
||||
* `rollup-read.ts` carries the existing DAY-only aggregator. This
|
||||
* module is deliberately separate so the WEEK / MONTH / YEAR wiring
|
||||
* is reviewable in isolation and so the v1.4.39 W-SUM agent's
|
||||
* concurrent edits to the writer side don't collide with the reader
|
||||
* additions.
|
||||
* Each reader returns `null` when the window holds no DAY buckets for
|
||||
* `(userId, type)`. The caller decides whether that is a real "no data"
|
||||
* case or a coverage miss the boot backfill has not caught up on, and
|
||||
* usually falls through to live SQL.
|
||||
*/
|
||||
import type {
|
||||
MeasurementType,
|
||||
RollupGranularity,
|
||||
} from "@/generated/prisma/client";
|
||||
|
||||
import { prisma } from "@/lib/db";
|
||||
import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import {
|
||||
collapseRollupRowsBySource,
|
||||
foldCanonicalDays,
|
||||
loadUserSourcePriority,
|
||||
readCanonicalRollupBuckets,
|
||||
} from "@/lib/rollups/measurement-read";
|
||||
|
||||
/**
|
||||
@@ -125,11 +101,7 @@ const GRANULARITY_FLOORS: Array<{
|
||||
|
||||
/**
|
||||
* Pick the largest granularity that can still resolve the requested
|
||||
* `windowDays` window — WEEK for >14 d, MONTH for >62 d, YEAR for
|
||||
* >730 d. Falls back to a finer granularity on coverage miss so a
|
||||
* user with WEEK / MONTH coverage but no YEAR buckets (e.g. a tenant
|
||||
* who joined less than two years ago) still gets a usable trend
|
||||
* series.
|
||||
* `windowDays` window (see `GRANULARITY_FLOORS`) and read it.
|
||||
*
|
||||
* Returns the granularity the helper resolved against plus the row
|
||||
* shape the internal `readGranularity` reader produces. `null` when
|
||||
@@ -153,28 +125,25 @@ export async function readBestGranularityRollups(
|
||||
userPriorityJson !== undefined
|
||||
? userPriorityJson
|
||||
: await loadUserSourcePriority(userId);
|
||||
// Walk the floors from coarsest to finest and return the first
|
||||
// granularity whose floor the window clears AND which has coverage
|
||||
// for `(userId, type, since)`. The `if rows == null` fall-through
|
||||
// is what makes the helper resilient to partial coverage.
|
||||
for (const floor of GRANULARITY_FLOORS) {
|
||||
if (windowDays < floor.minWindowDays) continue;
|
||||
const rows = await readGranularity(
|
||||
userId,
|
||||
type,
|
||||
floor.granularity,
|
||||
since,
|
||||
// Trailing-window semantics: no upper bound. This router serves the
|
||||
// "last N days to now" probes (summaries-slice / health-score); the
|
||||
// requested-window bounding lives on `readTieredRollupSeries`.
|
||||
null,
|
||||
priority,
|
||||
);
|
||||
if (rows && rows.length > 0) {
|
||||
return { granularity: floor.granularity, rows };
|
||||
}
|
||||
}
|
||||
return null;
|
||||
// Every tier is folded from the same canonical DAY buckets, so a tier
|
||||
// either has coverage or none has: pick the coarsest floor the window
|
||||
// clears and read it once.
|
||||
const floor = GRANULARITY_FLOORS.find((f) => windowDays >= f.minWindowDays);
|
||||
if (!floor) return null;
|
||||
const rows = await readGranularity(
|
||||
userId,
|
||||
type,
|
||||
floor.granularity,
|
||||
since,
|
||||
// Trailing-window semantics: no upper bound. This router serves the
|
||||
// "last N days to now" probes (summaries-slice / health-score); the
|
||||
// requested-window bounding lives on `readTieredRollupSeries`.
|
||||
null,
|
||||
priority,
|
||||
);
|
||||
return rows && rows.length > 0
|
||||
? { granularity: floor.granularity, rows }
|
||||
: null;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -225,9 +194,17 @@ export function aggregateWmyBuckets(rows: RollupBucketRow[]): {
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal — shared `findMany` projection. Returns `null` when the
|
||||
* window has zero rows so the caller can branch on coverage miss
|
||||
* without a separate count round-trip.
|
||||
* Internal — one type's canonical buckets at `granularity`, built from the
|
||||
* DAY tier (`readCanonicalRollupBuckets`: canonical source per DAY, then the
|
||||
* fold). Returns `null` when the window has no rows so the caller can branch
|
||||
* on a coverage miss without a separate count round-trip.
|
||||
*
|
||||
* When `to` is supplied the window is bounded on BOTH ends (`bucketStart` in
|
||||
* `[from, to]`), so a caller asking for an arbitrary historic window reads
|
||||
* the buckets INSIDE it rather than the trailing "to now" slice, the same
|
||||
* contract the DAY-tier `readRollup` and the live-SQL fallback keep.
|
||||
* `to === null` keeps the trailing-window semantics for the
|
||||
* `readBestGranularityRollups` router.
|
||||
*/
|
||||
async function readGranularity(
|
||||
userId: string,
|
||||
@@ -237,43 +214,17 @@ async function readGranularity(
|
||||
to: Date | null,
|
||||
userPriorityJson: unknown,
|
||||
): Promise<RollupBucketRow[] | null> {
|
||||
// Bounded `findMany`: `(userId, type, granularity, bucketStart, source)`
|
||||
// is the composite primary key so the planner picks the index path
|
||||
// every time. When `to` is supplied the `bucketStart` filter is bounded
|
||||
// on BOTH ends (`gte from` AND `lte to`) so a caller asking for an
|
||||
// arbitrary historic window — `[2020-01-01, 2022-01-01]` — reads the
|
||||
// buckets INSIDE that window rather than the trailing "to now" slice.
|
||||
// This mirrors the DAY-tier `readRollup` in `daily-series-read.ts`
|
||||
// (`bucketStart: { gte: from, lte: to }`) so the tiered read reproduces
|
||||
// the same window contract the live-SQL fallback bounds on both ends.
|
||||
// `to === null` keeps the legacy trailing-window semantics for the
|
||||
// `readBestGranularityRollups` router.
|
||||
const rows = await prisma.measurementRollup.findMany({
|
||||
where: {
|
||||
userId,
|
||||
type,
|
||||
granularity,
|
||||
bucketStart: to === null ? { gte: from } : { gte: from, lte: to },
|
||||
},
|
||||
orderBy: { bucketStart: "asc" },
|
||||
select: {
|
||||
bucketStart: true,
|
||||
// v1.11.1 — source drives the per-bucket canonical collapse below.
|
||||
source: true,
|
||||
count: true,
|
||||
mean: true,
|
||||
sd: true,
|
||||
slope: true,
|
||||
r2: true,
|
||||
sumValue: true,
|
||||
minValue: true,
|
||||
maxValue: true,
|
||||
},
|
||||
const rows = await readCanonicalRollupBuckets({
|
||||
userId,
|
||||
type,
|
||||
granularity,
|
||||
from,
|
||||
to,
|
||||
toInclusive: true,
|
||||
userPriorityJson,
|
||||
});
|
||||
if (rows.length === 0) return null;
|
||||
// v1.11.1 — collapse overlapping sources to the ladder-canonical reading
|
||||
// per bucket, then drop the source column from the normalised shape.
|
||||
return collapseRollupRowsBySource(rows, type, userPriorityJson).map((r) => ({
|
||||
return rows.map((r) => ({
|
||||
bucketStart: r.bucketStart,
|
||||
count: r.count,
|
||||
mean: r.mean,
|
||||
@@ -318,9 +269,6 @@ export interface TieredSeriesRow {
|
||||
* ladders are not identical — this one may be finer (DAY rows for a
|
||||
* 91–365 d span; the client folds them into the weeks it captions) —
|
||||
* and a test pins the never-coarser direction across every span.
|
||||
*
|
||||
* The finer fallback in `readTieredRollupSeries` still rescues a tenant
|
||||
* whose coarse buckets the worker has not minted yet.
|
||||
*/
|
||||
export function pickRollupGranularityForWindow(
|
||||
windowDays: number,
|
||||
@@ -330,7 +278,7 @@ export function pickRollupGranularityForWindow(
|
||||
return "DAY";
|
||||
}
|
||||
|
||||
/** DAY → WEEK → MONTH → YEAR ordering for the finer-fallback walk. */
|
||||
/** DAY → WEEK → MONTH → YEAR, finest first. */
|
||||
const TIER_ORDER: RollupGranularity[] = ["DAY", "WEEK", "MONTH", "YEAR"];
|
||||
|
||||
/**
|
||||
@@ -354,12 +302,9 @@ const TIER_ORDER: RollupGranularity[] = ["DAY", "WEEK", "MONTH", "YEAR"];
|
||||
* range band on spot metrics; cumulative metrics (steps, energy,
|
||||
* distance) surface the bucket's summed total and drop the spread.
|
||||
*
|
||||
* Coverage handling: the target tier is read first; on a coverage miss
|
||||
* (the boot backfill / worker has not minted the coarse buckets yet) the
|
||||
* reader walks FINER (MONTH → WEEK → DAY) so a tenant still gets a usable
|
||||
* series rather than an empty chart. Returns `null` only when no tier at
|
||||
* or below the target carries any buckets for the window — the caller
|
||||
* then falls through to its live-SQL path.
|
||||
* Coverage handling: the window's canonical DAY buckets are read once and
|
||||
* folded into the chosen tier. Returns `null` when the window holds no DAY
|
||||
* buckets; the caller then falls through to its live-SQL path.
|
||||
*
|
||||
* The result is NOT capped — the tier selection bounds the row count
|
||||
* (≤ ~104 weeks for the WEEK tier, ≤ ~12 months/year for MONTH). If a
|
||||
@@ -397,77 +342,58 @@ export async function readTieredRollupSeries(opts: {
|
||||
: await loadUserSourcePriority(userId);
|
||||
|
||||
const target = pickRollupGranularityForWindow(windowDays);
|
||||
// Walk the target tier first, then finer (never coarser — coarser would
|
||||
// drop detail the chart can render). Stop at the first tier with coverage.
|
||||
const targetIdx = TIER_ORDER.indexOf(target);
|
||||
for (let i = targetIdx; i >= 0; i--) {
|
||||
let granularity = TIER_ORDER[i];
|
||||
let rows = await readGranularity(
|
||||
userId,
|
||||
const days = await readCanonicalRollupBuckets({
|
||||
userId,
|
||||
type,
|
||||
granularity: "DAY",
|
||||
from,
|
||||
to,
|
||||
toInclusive: true,
|
||||
userPriorityJson: priority,
|
||||
});
|
||||
if (days.length === 0) return null;
|
||||
// Refine the tier by the ACTUAL data span, not the requested window. The
|
||||
// "Alle" tab always requests ~3650 days, so keying the tier off the request
|
||||
// width alone handed EVERY account MONTH buckets: a record whose history
|
||||
// spans four months came back as four monthly means while the client
|
||||
// (which captions from the real span) labelled them "weekly average". The
|
||||
// tier is never coarser than the request calls for.
|
||||
const spanDays = Math.ceil(
|
||||
(days[days.length - 1].bucketStart.getTime() -
|
||||
days[0].bucketStart.getTime()) /
|
||||
86_400_000,
|
||||
);
|
||||
const refined = pickRollupGranularityForWindow(Math.max(1, spanDays));
|
||||
const granularity =
|
||||
TIER_ORDER.indexOf(refined) < TIER_ORDER.indexOf(target) ? refined : target;
|
||||
// Fold the window's own days; a leading bucket that starts before `from`
|
||||
// would hold only part of its days, so it is left out as the stored coarse
|
||||
// rows were.
|
||||
const rows = foldCanonicalDays(days, granularity).filter(
|
||||
(b) => b.bucketStart.getTime() >= from.getTime(),
|
||||
);
|
||||
if (rows.length === 0) return null;
|
||||
const useSum = CUMULATIVE_HK_TYPES.has(type);
|
||||
annotate({
|
||||
action: { name: "measurement.list" },
|
||||
meta: {
|
||||
total: rows.length,
|
||||
type,
|
||||
aggregate: "tiered",
|
||||
granularity,
|
||||
from,
|
||||
to,
|
||||
priority,
|
||||
);
|
||||
if (rows && rows.length > 0) {
|
||||
// Refine the tier by the ACTUAL data span, not the requested window.
|
||||
// The "Alle" tab always requests ~3650 days, so keying the tier off
|
||||
// the request width alone handed EVERY account MONTH buckets — a
|
||||
// record whose history spans four months came back as four monthly
|
||||
// means while the client (which captions from the real span) labelled
|
||||
// them "weekly average". Re-reading at the tier the real span calls
|
||||
// for returns the resolution the chart will actually caption. A
|
||||
// coverage miss on the finer tier keeps the coarser rows — finer
|
||||
// tiers are the base tiers the workers mint first, so in practice
|
||||
// this probe only fires when the refinement is genuinely available.
|
||||
// The walk is strictly finer-only, so it terminates.
|
||||
for (;;) {
|
||||
const spanDays = Math.ceil(
|
||||
(rows[rows.length - 1].bucketStart.getTime() -
|
||||
rows[0].bucketStart.getTime()) /
|
||||
86_400_000,
|
||||
);
|
||||
const refined = pickRollupGranularityForWindow(Math.max(1, spanDays));
|
||||
if (TIER_ORDER.indexOf(refined) >= TIER_ORDER.indexOf(granularity)) {
|
||||
break;
|
||||
}
|
||||
const finerRows = await readGranularity(
|
||||
userId,
|
||||
type,
|
||||
refined,
|
||||
from,
|
||||
to,
|
||||
priority,
|
||||
);
|
||||
if (!finerRows || finerRows.length === 0) break;
|
||||
granularity = refined;
|
||||
rows = finerRows;
|
||||
}
|
||||
const useSum = CUMULATIVE_HK_TYPES.has(type);
|
||||
annotate({
|
||||
action: { name: "measurement.list" },
|
||||
meta: {
|
||||
total: rows.length,
|
||||
type,
|
||||
aggregate: "tiered",
|
||||
granularity,
|
||||
target_granularity: target,
|
||||
source: "rollup",
|
||||
},
|
||||
});
|
||||
return {
|
||||
granularity,
|
||||
rows: rows.map((r) => ({
|
||||
type,
|
||||
value: useSum ? (r.sumValue ?? r.mean * r.count) : r.mean,
|
||||
measuredAt: r.bucketStart.toISOString(),
|
||||
count: r.count,
|
||||
minValue: useSum ? undefined : r.minValue,
|
||||
maxValue: useSum ? undefined : r.maxValue,
|
||||
})),
|
||||
};
|
||||
}
|
||||
}
|
||||
return null;
|
||||
target_granularity: target,
|
||||
source: "rollup",
|
||||
},
|
||||
});
|
||||
return {
|
||||
granularity,
|
||||
rows: rows.map((r) => ({
|
||||
type,
|
||||
value: useSum ? (r.sumValue ?? r.mean * r.count) : r.mean,
|
||||
measuredAt: r.bucketStart.toISOString(),
|
||||
count: r.count,
|
||||
minValue: useSum ? undefined : r.minValue,
|
||||
maxValue: useSum ? undefined : r.maxValue,
|
||||
})),
|
||||
};
|
||||
}
|
||||
@@ -23,9 +23,11 @@
|
||||
import type {
|
||||
MeasurementSource,
|
||||
MeasurementType,
|
||||
RollupGranularity,
|
||||
} from "@/generated/prisma/client";
|
||||
import { prisma } from "@/lib/db";
|
||||
import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum";
|
||||
import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
|
||||
import {
|
||||
getSourceLadder,
|
||||
parseSourcePriority,
|
||||
@@ -412,3 +414,235 @@ export function aggregateBuckets(rows: DailyMeanRow[]): {
|
||||
mean: sumWeighted / totalCount,
|
||||
};
|
||||
}
|
||||
|
||||
const DAY_MS = 86_400_000;
|
||||
|
||||
/**
|
||||
* Start of the UTC bucket containing `at` at the given granularity. The same
|
||||
* cut Postgres `date_trunc(<unit>, measured_at)` makes on the rollup writer's
|
||||
* UTC wall-clock `measured_at` (weeks are ISO weeks starting on Monday).
|
||||
*/
|
||||
export function utcBucketStart(at: Date, granularity: RollupGranularity): Date {
|
||||
switch (granularity) {
|
||||
case "DAY":
|
||||
return startOfUtcDay(at);
|
||||
case "WEEK": {
|
||||
const day = startOfUtcDay(at);
|
||||
// getUTCDay(): Sunday = 0 … Saturday = 6; ISO weeks start on Monday.
|
||||
const mondayOffset = (day.getUTCDay() + 6) % 7;
|
||||
return new Date(day.getTime() - mondayOffset * DAY_MS);
|
||||
}
|
||||
case "MONTH":
|
||||
return new Date(Date.UTC(at.getUTCFullYear(), at.getUTCMonth(), 1));
|
||||
case "YEAR":
|
||||
return new Date(Date.UTC(at.getUTCFullYear(), 0, 1));
|
||||
}
|
||||
}
|
||||
|
||||
/** Start of the UTC bucket after the one containing `at`. */
|
||||
export function utcBucketEnd(at: Date, granularity: RollupGranularity): Date {
|
||||
const start = utcBucketStart(at, granularity);
|
||||
switch (granularity) {
|
||||
case "DAY":
|
||||
return new Date(start.getTime() + DAY_MS);
|
||||
case "WEEK":
|
||||
return new Date(start.getTime() + 7 * DAY_MS);
|
||||
case "MONTH":
|
||||
return new Date(
|
||||
Date.UTC(start.getUTCFullYear(), start.getUTCMonth() + 1, 1),
|
||||
);
|
||||
case "YEAR":
|
||||
return new Date(Date.UTC(start.getUTCFullYear() + 1, 0, 1));
|
||||
}
|
||||
}
|
||||
|
||||
/** One canonical-source bucket, at any granularity. */
|
||||
export interface CanonicalRollupBucket {
|
||||
bucketStart: Date;
|
||||
count: number;
|
||||
mean: number;
|
||||
minValue: number;
|
||||
maxValue: number;
|
||||
sumValue: number | null;
|
||||
sd: number | null;
|
||||
slope: number | null;
|
||||
r2: number | null;
|
||||
sumX: number | null;
|
||||
sumXy: number | null;
|
||||
sumXx: number | null;
|
||||
sumYy: number | null;
|
||||
computedAt: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fold canonical DAY buckets (one row per day, already source-collapsed)
|
||||
* into buckets of `granularity`.
|
||||
*
|
||||
* The canonical source is a per-DAY decision: a week, month or year in which
|
||||
* the user logged most days by hand and one day from a device holds readings
|
||||
* from both, and each of those days counts. Collapsing the stored per-source
|
||||
* WEEK / MONTH / YEAR rows instead picked ONE source for the whole bucket, so
|
||||
* a month of manual blood-pressure readings plus a single synced one reported
|
||||
* the synced day alone, and a step history that switched devices mid-month
|
||||
* lost every day logged by the losing device.
|
||||
*
|
||||
* `count`, `min`, `max`, `sum` and the count-weighted `mean` compose exactly.
|
||||
* `sd`, `slope` and `r2` are composed from the summed regression accumulators
|
||||
* (`composeRegression`); a bucket with any day lacking them reports `null`.
|
||||
* DAY input at DAY granularity passes through unchanged.
|
||||
*/
|
||||
export function foldCanonicalDays(
|
||||
days: ReadonlyArray<CanonicalRollupBucket>,
|
||||
granularity: RollupGranularity,
|
||||
): CanonicalRollupBucket[] {
|
||||
if (granularity === "DAY") return days.map((d) => ({ ...d }));
|
||||
const groups = new Map<number, CanonicalRollupBucket[]>();
|
||||
for (const d of days) {
|
||||
const key = utcBucketStart(d.bucketStart, granularity).getTime();
|
||||
const slot = groups.get(key);
|
||||
if (slot) slot.push(d);
|
||||
else groups.set(key, [d]);
|
||||
}
|
||||
const out: CanonicalRollupBucket[] = [];
|
||||
for (const [key, members] of [...groups.entries()].sort(
|
||||
(a, b) => a[0] - b[0],
|
||||
)) {
|
||||
let count = 0;
|
||||
let sumY = 0;
|
||||
let min = Infinity;
|
||||
let max = -Infinity;
|
||||
let computedAt = members[0].computedAt;
|
||||
let accumulatorsComplete = true;
|
||||
let sumX = 0;
|
||||
let sumXy = 0;
|
||||
let sumXx = 0;
|
||||
let sumYy = 0;
|
||||
for (const m of members) {
|
||||
count += m.count;
|
||||
sumY += m.sumValue ?? m.mean * m.count;
|
||||
if (m.minValue < min) min = m.minValue;
|
||||
if (m.maxValue > max) max = m.maxValue;
|
||||
if (m.computedAt.getTime() > computedAt.getTime()) {
|
||||
computedAt = m.computedAt;
|
||||
}
|
||||
if (
|
||||
m.sumX === null ||
|
||||
m.sumXy === null ||
|
||||
m.sumXx === null ||
|
||||
m.sumYy === null
|
||||
) {
|
||||
accumulatorsComplete = false;
|
||||
} else {
|
||||
sumX += m.sumX;
|
||||
sumXy += m.sumXy;
|
||||
sumXx += m.sumXx;
|
||||
sumYy += m.sumYy;
|
||||
}
|
||||
}
|
||||
if (count === 0) continue;
|
||||
const mean = sumY / count;
|
||||
const regression = accumulatorsComplete
|
||||
? composeRegression([
|
||||
{ count, mean, sumValue: sumY, sumX, sumXy, sumXx, sumYy },
|
||||
])
|
||||
: { slope: null, r2: null, sdPop: null };
|
||||
out.push({
|
||||
bucketStart: new Date(key),
|
||||
count,
|
||||
mean,
|
||||
minValue: min,
|
||||
maxValue: max,
|
||||
sumValue: sumY,
|
||||
// A single reading has no spread; STDDEV_POP of one value is 0.
|
||||
sd: count === 1 && accumulatorsComplete ? 0 : regression.sdPop,
|
||||
slope: regression.slope,
|
||||
r2: regression.r2,
|
||||
sumX: accumulatorsComplete ? sumX : null,
|
||||
sumXy: accumulatorsComplete ? sumXy : null,
|
||||
sumXx: accumulatorsComplete ? sumXx : null,
|
||||
sumYy: accumulatorsComplete ? sumYy : null,
|
||||
computedAt,
|
||||
});
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read one type's canonical rollup buckets at `granularity`.
|
||||
*
|
||||
* Every granularity is built from the DAY tier: the day's rows are collapsed
|
||||
* to the ladder-canonical source first (`collapseRollupRowsBySource`), then
|
||||
* folded into weeks, months or years (`foldCanonicalDays`). The stored
|
||||
* per-source WEEK / MONTH / YEAR rows are not read, because the canonical
|
||||
* source cannot be chosen once per coarse bucket.
|
||||
*
|
||||
* A bucket is returned when its start lies in the window: `bucketStart >=
|
||||
* from`, and `< to` (or `<= to` with `toInclusive`). `to === null` leaves the
|
||||
* window open towards now. A returned bucket always carries all of its days,
|
||||
* including days past `to`, the way the stored coarse rows did.
|
||||
*/
|
||||
export async function readCanonicalRollupBuckets(opts: {
|
||||
userId: string;
|
||||
type: MeasurementType;
|
||||
granularity: RollupGranularity;
|
||||
from: Date;
|
||||
to: Date | null;
|
||||
toInclusive?: boolean;
|
||||
/** The user's source-priority blob; `undefined` loads it on demand. */
|
||||
userPriorityJson?: unknown;
|
||||
}): Promise<CanonicalRollupBucket[]> {
|
||||
const { userId, type, granularity, from, to } = opts;
|
||||
// The DAY read reaches the end of the last bucket that can start inside
|
||||
// the window, so that bucket folds complete.
|
||||
const dayUpper =
|
||||
to === null
|
||||
? undefined
|
||||
: granularity === "DAY"
|
||||
? to
|
||||
: utcBucketEnd(to, granularity);
|
||||
const dayRows = await prisma.measurementRollup.findMany({
|
||||
where: {
|
||||
userId,
|
||||
type,
|
||||
granularity: "DAY",
|
||||
bucketStart:
|
||||
dayUpper === undefined
|
||||
? { gte: from }
|
||||
: granularity === "DAY" && opts.toInclusive
|
||||
? { gte: from, lte: dayUpper }
|
||||
: { gte: from, lt: dayUpper },
|
||||
},
|
||||
orderBy: { bucketStart: "asc" },
|
||||
});
|
||||
if (dayRows.length === 0) return [];
|
||||
const priority =
|
||||
opts.userPriorityJson !== undefined
|
||||
? opts.userPriorityJson
|
||||
: await loadUserSourcePriority(userId);
|
||||
const canonicalDays = collapseRollupRowsBySource(dayRows, type, priority).map(
|
||||
(r): CanonicalRollupBucket => ({
|
||||
bucketStart: r.bucketStart,
|
||||
count: r.count,
|
||||
mean: r.mean,
|
||||
minValue: r.minValue,
|
||||
maxValue: r.maxValue,
|
||||
sumValue: r.sumValue,
|
||||
sd: r.sd,
|
||||
slope: r.slope,
|
||||
r2: r.r2,
|
||||
sumX: r.sumX,
|
||||
sumXy: r.sumXy,
|
||||
sumXx: r.sumXx,
|
||||
sumYy: r.sumYy,
|
||||
computedAt: r.computedAt,
|
||||
}),
|
||||
);
|
||||
const fromMs = from.getTime();
|
||||
const toMs = to?.getTime() ?? null;
|
||||
return foldCanonicalDays(canonicalDays, granularity).filter((b) => {
|
||||
const start = b.bucketStart.getTime();
|
||||
if (start < fromMs) return false;
|
||||
if (toMs === null) return true;
|
||||
return opts.toInclusive ? start <= toMs : start < toMs;
|
||||
});
|
||||
}
|
||||
@@ -38,9 +38,10 @@ import { prisma } from "@/lib/db";
|
||||
import { getGlobalBoss } from "@/lib/jobs/boss-instance";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import {
|
||||
collapseRollupRowsBySource,
|
||||
loadUserSourcePriority,
|
||||
readCanonicalRollupBuckets,
|
||||
REGRESSION_X_ORIGIN_DAYS,
|
||||
utcBucketEnd,
|
||||
utcBucketStart,
|
||||
} from "@/lib/rollups/measurement-read";
|
||||
import { startOfUtcDay } from "@/lib/tz/start-of-utc-day";
|
||||
import type {
|
||||
@@ -354,8 +355,10 @@ export async function enqueueRollupRecompute(input: {
|
||||
* Read rollup rows for `(userId, type, granularity)` in `[from, to)`.
|
||||
* Returns rows sorted ascending by `bucketStart`.
|
||||
*
|
||||
* v1.11.1 — rows are stored per source; this collapses overlapping sources to
|
||||
* the ladder-canonical reading so the return shape stays one row per bucket.
|
||||
* Rows are stored per source. Every granularity is built from the DAY tier,
|
||||
* collapsed to the ladder-canonical source per DAY and then folded into the
|
||||
* requested bucket (`readCanonicalRollupBuckets`), so the return shape stays
|
||||
* one row per bucket and a coarse bucket counts every day it covers.
|
||||
* Pass `userPriorityJson` to avoid a per-call user lookup when the caller
|
||||
* already loaded it (e.g. a loop over many types); omit it to lazy-load.
|
||||
*/
|
||||
@@ -379,20 +382,15 @@ export async function readRollupBuckets(
|
||||
computedAt: Date;
|
||||
}>
|
||||
> {
|
||||
const rows = await prisma.measurementRollup.findMany({
|
||||
where: {
|
||||
userId,
|
||||
type,
|
||||
granularity,
|
||||
bucketStart: { gte: from, lt: to },
|
||||
},
|
||||
orderBy: { bucketStart: "asc" },
|
||||
const rows = await readCanonicalRollupBuckets({
|
||||
userId,
|
||||
type,
|
||||
granularity,
|
||||
from,
|
||||
to,
|
||||
userPriorityJson,
|
||||
});
|
||||
const priority =
|
||||
userPriorityJson !== undefined
|
||||
? userPriorityJson
|
||||
: await loadUserSourcePriority(userId);
|
||||
return collapseRollupRowsBySource(rows, type, priority).map((r) => ({
|
||||
return rows.map((r) => ({
|
||||
bucketStart: r.bucketStart,
|
||||
count: r.count,
|
||||
mean: r.mean,
|
||||
@@ -800,34 +798,10 @@ export function bucketSpan(
|
||||
measuredAt: Date,
|
||||
granularity: RollupGranularity,
|
||||
): { from: Date; to: Date } {
|
||||
switch (granularity) {
|
||||
case "DAY": {
|
||||
const from = startOfUtcDay(measuredAt);
|
||||
return { from, to: new Date(from.getTime() + 24 * 60 * 60 * 1000) };
|
||||
}
|
||||
case "WEEK": {
|
||||
const day = startOfUtcDay(measuredAt);
|
||||
// Postgres ISO week: Monday is day 1. JS getUTCDay(): Sunday=0..Saturday=6.
|
||||
const dayOfWeek = day.getUTCDay(); // 0=Sun
|
||||
const mondayOffset = (dayOfWeek + 6) % 7; // 0 for Mon … 6 for Sun
|
||||
const from = new Date(day.getTime() - mondayOffset * 24 * 60 * 60 * 1000);
|
||||
return { from, to: new Date(from.getTime() + 7 * 24 * 60 * 60 * 1000) };
|
||||
}
|
||||
case "MONTH": {
|
||||
const from = new Date(
|
||||
Date.UTC(measuredAt.getUTCFullYear(), measuredAt.getUTCMonth(), 1),
|
||||
);
|
||||
const to = new Date(
|
||||
Date.UTC(measuredAt.getUTCFullYear(), measuredAt.getUTCMonth() + 1, 1),
|
||||
);
|
||||
return { from, to };
|
||||
}
|
||||
case "YEAR": {
|
||||
const from = new Date(Date.UTC(measuredAt.getUTCFullYear(), 0, 1));
|
||||
const to = new Date(Date.UTC(measuredAt.getUTCFullYear() + 1, 0, 1));
|
||||
return { from, to };
|
||||
}
|
||||
}
|
||||
return {
|
||||
from: utcBucketStart(measuredAt, granularity),
|
||||
to: utcBucketEnd(measuredAt, granularity),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,286 @@
|
||||
/**
|
||||
* Coarse buckets pick the canonical source per DAY, on both read paths.
|
||||
*
|
||||
* The rollup writer mints one row per (type, bucket, source). Collapsing the
|
||||
* stored WEEK / MONTH / YEAR rows picked ONE source for the whole bucket, so a
|
||||
* month of manual blood-pressure readings plus one synced reading reported the
|
||||
* synced day alone, and a step month that switched devices lost every day the
|
||||
* losing device carried. The live weekly / monthly fold made the same pick.
|
||||
*
|
||||
* This suite seeds real rows, folds the rollups through the real writer, and
|
||||
* pins that the rollup readers (`readRollupBuckets`,
|
||||
* `readBestGranularityRollups`) and the live fold (`readLiveBuckets`) agree
|
||||
* with each other and with the per-day arithmetic.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
import {
|
||||
readRollupBuckets,
|
||||
recomputeUserRollups,
|
||||
} from "@/lib/rollups/measurement-rollups";
|
||||
import { readLiveBuckets } from "@/lib/measurements/daily-series-read";
|
||||
import type { MeasurementType } from "@/generated/prisma/client";
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: vi.fn(() => null),
|
||||
}));
|
||||
|
||||
const DAY_MS = 86_400_000;
|
||||
const FROM = new Date("2025-11-01T00:00:00.000Z");
|
||||
const TO = new Date("2025-12-01T00:00:00.000Z");
|
||||
|
||||
let seq = 0;
|
||||
async function seedUser() {
|
||||
seq += 1;
|
||||
return getPrismaClient().user.create({
|
||||
data: {
|
||||
username: `per-day-canonical-${seq}`,
|
||||
email: `per-day-canonical-${seq}@example.test`,
|
||||
role: "USER",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
type Row = {
|
||||
type: MeasurementType;
|
||||
source: "MANUAL" | "WITHINGS" | "APPLE_HEALTH" | "FITBIT";
|
||||
value: number;
|
||||
at: Date;
|
||||
};
|
||||
|
||||
async function seed(userId: string, rows: Row[]) {
|
||||
await getPrismaClient().measurement.createMany({
|
||||
data: rows.map((r) => ({
|
||||
userId,
|
||||
type: r.type,
|
||||
value: r.value,
|
||||
unit: r.type === "ACTIVITY_STEPS" ? "count" : "mmHg",
|
||||
source: r.source,
|
||||
measuredAt: r.at,
|
||||
})),
|
||||
});
|
||||
await recomputeUserRollups(userId, {
|
||||
from: new Date("2025-10-01T00:00:00.000Z"),
|
||||
to: new Date("2026-01-01T00:00:00.000Z"),
|
||||
});
|
||||
}
|
||||
|
||||
function day(n: number, hour = 8): Date {
|
||||
return new Date(FROM.getTime() + n * DAY_MS + hour * 3_600_000);
|
||||
}
|
||||
|
||||
async function live(
|
||||
userId: string,
|
||||
type: MeasurementType,
|
||||
grain: "weekly" | "monthly",
|
||||
) {
|
||||
return readLiveBuckets({
|
||||
userId,
|
||||
type,
|
||||
from: FROM,
|
||||
to: new Date(TO.getTime() - 1),
|
||||
cap: 100,
|
||||
priorityJson: null,
|
||||
grain,
|
||||
timeZone: "UTC",
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
await truncateAllTables(getPrismaClient());
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("coarse buckets — canonical source per day", () => {
|
||||
it("counts every manual day of a month that also carries one synced day", async () => {
|
||||
const user = await seedUser();
|
||||
const rows: Row[] = [];
|
||||
for (let i = 0; i < 20; i++) {
|
||||
rows.push({
|
||||
type: "BLOOD_PRESSURE_SYS",
|
||||
source: "MANUAL",
|
||||
value: 150,
|
||||
at: day(i),
|
||||
});
|
||||
}
|
||||
// WITHINGS leads the blood-pressure ladder; it owns day 24 only.
|
||||
rows.push({
|
||||
type: "BLOOD_PRESSURE_SYS",
|
||||
source: "WITHINGS",
|
||||
value: 120,
|
||||
at: day(24),
|
||||
});
|
||||
await seed(user.id, rows);
|
||||
|
||||
const [month] = await readRollupBuckets(
|
||||
user.id,
|
||||
"BLOOD_PRESSURE_SYS",
|
||||
"MONTH",
|
||||
FROM,
|
||||
TO,
|
||||
);
|
||||
expect(month.count).toBe(21);
|
||||
expect(month.mean).toBeCloseTo((20 * 150 + 120) / 21, 9);
|
||||
|
||||
const [liveMonth] = await live(user.id, "BLOOD_PRESSURE_SYS", "monthly");
|
||||
expect(liveMonth.count).toBe(21);
|
||||
expect(liveMonth.value).toBeCloseTo(month.mean, 9);
|
||||
});
|
||||
|
||||
it("sums a step month that switches device, and keeps one source on a shared day", async () => {
|
||||
const user = await seedUser();
|
||||
const rows: Row[] = [];
|
||||
for (let i = 0; i < 10; i++) {
|
||||
rows.push({
|
||||
type: "ACTIVITY_STEPS",
|
||||
source: "FITBIT",
|
||||
value: 8000,
|
||||
at: day(i),
|
||||
});
|
||||
}
|
||||
for (let i = 10; i < 30; i++) {
|
||||
rows.push({
|
||||
type: "ACTIVITY_STEPS",
|
||||
source: "APPLE_HEALTH",
|
||||
value: 8000,
|
||||
at: day(i),
|
||||
});
|
||||
}
|
||||
// Day 12 also carries a Fitbit total; Apple Health leads the ladder.
|
||||
rows.push({
|
||||
type: "ACTIVITY_STEPS",
|
||||
source: "FITBIT",
|
||||
value: 9999,
|
||||
at: day(12),
|
||||
});
|
||||
await seed(user.id, rows);
|
||||
|
||||
const monthly = await readRollupBuckets(
|
||||
user.id,
|
||||
"ACTIVITY_STEPS",
|
||||
"MONTH",
|
||||
FROM,
|
||||
TO,
|
||||
);
|
||||
expect(monthly).toHaveLength(1);
|
||||
expect(monthly[0].count).toBe(30);
|
||||
expect(monthly[0].mean * monthly[0].count).toBeCloseTo(240_000, 6);
|
||||
|
||||
const [liveMonth] = await live(user.id, "ACTIVITY_STEPS", "monthly");
|
||||
expect(liveMonth.value).toBeCloseTo(240_000, 6);
|
||||
expect(liveMonth.count).toBe(30);
|
||||
});
|
||||
|
||||
it("the rollup and live weekly folds agree bucket for bucket", async () => {
|
||||
const user = await seedUser();
|
||||
const rows: Row[] = [];
|
||||
for (let i = 0; i < 30; i++) {
|
||||
// Two manual readings on most days, a synced one every fifth day, and
|
||||
// both sources on every tenth day.
|
||||
if (i % 5 !== 0 || i % 10 === 0) {
|
||||
rows.push({
|
||||
type: "BLOOD_PRESSURE_SYS",
|
||||
source: "MANUAL",
|
||||
value: 130 + (i % 7),
|
||||
at: day(i, 7),
|
||||
});
|
||||
rows.push({
|
||||
type: "BLOOD_PRESSURE_SYS",
|
||||
source: "MANUAL",
|
||||
value: 140 - (i % 4),
|
||||
at: day(i, 20),
|
||||
});
|
||||
}
|
||||
if (i % 5 === 0) {
|
||||
rows.push({
|
||||
type: "BLOOD_PRESSURE_SYS",
|
||||
source: "WITHINGS",
|
||||
value: 118 + i,
|
||||
at: day(i, 9),
|
||||
});
|
||||
}
|
||||
}
|
||||
await seed(user.id, rows);
|
||||
|
||||
const rollupWeeks = await readRollupBuckets(
|
||||
user.id,
|
||||
"BLOOD_PRESSURE_SYS",
|
||||
"WEEK",
|
||||
new Date("2025-10-27T00:00:00.000Z"),
|
||||
TO,
|
||||
);
|
||||
const liveWeeks = await readLiveBuckets({
|
||||
userId: user.id,
|
||||
type: "BLOOD_PRESSURE_SYS",
|
||||
from: new Date("2025-10-27T00:00:00.000Z"),
|
||||
to: new Date(TO.getTime() - 1),
|
||||
cap: 100,
|
||||
priorityJson: null,
|
||||
grain: "weekly",
|
||||
timeZone: "UTC",
|
||||
});
|
||||
|
||||
expect(liveWeeks.map((w) => w.measuredAt)).toEqual(
|
||||
rollupWeeks.map((w) => w.bucketStart.toISOString()),
|
||||
);
|
||||
for (let i = 0; i < rollupWeeks.length; i++) {
|
||||
expect(liveWeeks[i].count).toBe(rollupWeeks[i].count);
|
||||
expect(liveWeeks[i].value).toBeCloseTo(rollupWeeks[i].mean, 9);
|
||||
expect(liveWeeks[i].minValue).toBe(rollupWeeks[i].minValue);
|
||||
expect(liveWeeks[i].maxValue).toBe(rollupWeeks[i].maxValue);
|
||||
}
|
||||
// Every day counts once with its canonical source: 24 manual-only days
|
||||
// with two readings each, and six synced days (WITHINGS leads the ladder,
|
||||
// so it wins the three days that carry both).
|
||||
const total = rollupWeeks.reduce((n, w) => n + w.count, 0);
|
||||
expect(total).toBe(24 * 2 + 6);
|
||||
});
|
||||
|
||||
it("breaks a tie between unranked sources alphabetically on both paths", async () => {
|
||||
const user = await seedUser();
|
||||
// GLUCOSE carries no ladder: APPLE_HEALTH sorts before MANUAL by name,
|
||||
// although MANUAL comes first in the enum's declaration order.
|
||||
await getPrismaClient().measurement.createMany({
|
||||
data: [
|
||||
{
|
||||
userId: user.id,
|
||||
type: "BLOOD_GLUCOSE",
|
||||
value: 90,
|
||||
unit: "mg/dL",
|
||||
source: "MANUAL",
|
||||
measuredAt: day(3, 7),
|
||||
},
|
||||
{
|
||||
userId: user.id,
|
||||
type: "BLOOD_GLUCOSE",
|
||||
value: 110,
|
||||
unit: "mg/dL",
|
||||
source: "APPLE_HEALTH",
|
||||
measuredAt: day(3, 9),
|
||||
},
|
||||
],
|
||||
});
|
||||
await recomputeUserRollups(user.id, {
|
||||
from: new Date("2025-10-01T00:00:00.000Z"),
|
||||
to: new Date("2026-01-01T00:00:00.000Z"),
|
||||
});
|
||||
|
||||
const [rollupDay] = await readRollupBuckets(
|
||||
user.id,
|
||||
"BLOOD_GLUCOSE",
|
||||
"DAY",
|
||||
FROM,
|
||||
TO,
|
||||
);
|
||||
const [liveMonth] = await live(user.id, "BLOOD_GLUCOSE", "monthly");
|
||||
expect(rollupDay.mean).toBe(110);
|
||||
expect(liveMonth.value).toBe(110);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user