diff --git a/src/lib/analytics/source-rank-sql.ts b/src/lib/analytics/source-rank-sql.ts index f675c7a12..664b98170 100644 --- a/src/lib/analytics/source-rank-sql.ts +++ b/src/lib/analytics/source-rank-sql.ts @@ -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") diff --git a/src/lib/insights/__tests__/features.test.ts b/src/lib/insights/__tests__/features.test.ts index 261810cf2..2fa796a92 100644 --- a/src/lib/insights/__tests__/features.test.ts +++ b/src/lib/insights/__tests__/features.test.ts @@ -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[], + 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, diff --git a/src/lib/insights/__tests__/metric-status.test.ts b/src/lib/insights/__tests__/metric-status.test.ts index 091f2cf74..d118a93ec 100644 --- a/src/lib/insights/__tests__/metric-status.test.ts +++ b/src/lib/insights/__tests__/metric-status.test.ts @@ -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()), loadUserSourcePriority: vi.fn(), })); diff --git a/src/lib/measurements/daily-series-read.ts b/src/lib/measurements/daily-series-read.ts index 0ce67d26f..945345dd4 100644 --- a/src/lib/measurements/daily-series-read.ts +++ b/src/lib/measurements/daily-series-read.ts @@ -159,9 +159,10 @@ const TRUNC_UNIT: Record, 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) => { diff --git a/src/lib/rollups/__tests__/measurement-read-wmy.test.ts b/src/lib/rollups/__tests__/measurement-read-wmy.test.ts index 39d0e3f0d..31dd66dbb 100644 --- a/src/lib/rollups/__tests__/measurement-read-wmy.test.ts +++ b/src/lib/rollups/__tests__/measurement-read-wmy.test.ts @@ -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", + ]); }); }); }); diff --git a/src/lib/rollups/measurement-read-wmy.ts b/src/lib/rollups/measurement-read-wmy.ts index 5121ce427..92a40afc5 100644 --- a/src/lib/rollups/measurement-read-wmy.ts +++ b/src/lib/rollups/measurement-read-wmy.ts @@ -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 { - // 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, + })), + }; } diff --git a/src/lib/rollups/measurement-read.ts b/src/lib/rollups/measurement-read.ts index bffad4ad5..93c377028 100644 --- a/src/lib/rollups/measurement-read.ts +++ b/src/lib/rollups/measurement-read.ts @@ -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(, 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, + granularity: RollupGranularity, +): CanonicalRollupBucket[] { + if (granularity === "DAY") return days.map((d) => ({ ...d })); + const groups = new Map(); + 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 { + 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; + }); +} diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index c2aa7c74d..6afdac3d4 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -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), + }; } /** diff --git a/tests/integration/rollup-per-day-canonical.test.ts b/tests/integration/rollup-per-day-canonical.test.ts new file mode 100644 index 000000000..8cc597e26 --- /dev/null +++ b/tests/integration/rollup-per-day-canonical.test.ts @@ -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); + }); +});