From d06eec6acea1f52c2c89f7b10cb99dbe683d963a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 4 Jun 2026 09:37:25 +0200 Subject: [PATCH] feat(rollups): collapse sources in the DAY + WMY rollup readers readRollupBuckets and the WEEK/MONTH/YEAR readers now select the source column and run collapseRollupRowsBySource before composing, so every rollup consumer (dashboard sparkline, fast-paths, derived metrics, range deltas) sees one canonical row per bucket. loadUserSourcePriority loads the ladder blob once; readBestGranularityRollups threads it through each granularity probe so the per-floor walk never re-queries the user. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../__tests__/measurement-read-wmy.test.ts | 9 ++++++ src/lib/rollups/measurement-read-wmy.ts | 28 +++++++++++++++++-- src/lib/rollups/measurement-read.ts | 15 ++++++++++ src/lib/rollups/measurement-rollups.ts | 16 ++++++++++- 4 files changed, 64 insertions(+), 4 deletions(-) diff --git a/src/lib/rollups/__tests__/measurement-read-wmy.test.ts b/src/lib/rollups/__tests__/measurement-read-wmy.test.ts index 04531c6fb..f47c6dffa 100644 --- a/src/lib/rollups/__tests__/measurement-read-wmy.test.ts +++ b/src/lib/rollups/__tests__/measurement-read-wmy.test.ts @@ -24,6 +24,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ findMany: vi.fn(), + userFindUnique: vi.fn(), })); vi.mock("@/lib/db", () => ({ @@ -31,6 +32,12 @@ vi.mock("@/lib/db", () => ({ measurementRollup: { findMany: mocks.findMany, }, + // v1.11.1 — readBestGranularityRollups loads the source-priority blob via + // loadUserSourcePriority; default to null so the collapse uses the default + // ladders (these routing fixtures are single-source per bucket anyway). + user: { + findUnique: mocks.userFindUnique, + }, }, })); @@ -62,6 +69,8 @@ function bucket( beforeEach(() => { findMany.mockReset(); + mocks.userFindUnique.mockReset(); + mocks.userFindUnique.mockResolvedValue({ sourcePriorityJson: null }); }); afterEach(() => { diff --git a/src/lib/rollups/measurement-read-wmy.ts b/src/lib/rollups/measurement-read-wmy.ts index dc6a3fc72..7d9222ec7 100644 --- a/src/lib/rollups/measurement-read-wmy.ts +++ b/src/lib/rollups/measurement-read-wmy.ts @@ -61,6 +61,10 @@ import type { } from "@/generated/prisma/client"; import { prisma } from "@/lib/db"; +import { + collapseRollupRowsBySource, + loadUserSourcePriority, +} from "@/lib/rollups/measurement-read"; /** * Normalised bucket row returned by every WMY reader. Mirrors the @@ -134,19 +138,32 @@ export async function readBestGranularityRollups( userId: string, type: MeasurementType, windowDays: number, + userPriorityJson?: unknown, ): Promise<{ granularity: RollupGranularity; rows: RollupBucketRow[]; } | null> { if (!Number.isFinite(windowDays) || windowDays <= 0) return null; const since = new Date(Date.now() - windowDays * 24 * 60 * 60 * 1000); + // v1.11.1 — load the source-priority blob once and thread it into every + // granularity probe so the collapse never re-queries the user per floor. + const priority = + 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); + const rows = await readGranularity( + userId, + type, + floor.granularity, + since, + priority, + ); if (rows && rows.length > 0) { return { granularity: floor.granularity, rows }; } @@ -211,8 +228,9 @@ async function readGranularity( type: MeasurementType, granularity: RollupGranularity, since: Date, + userPriorityJson: unknown, ): Promise { - // Bounded `findMany`: `(userId, type, granularity, bucketStart)` + // Bounded `findMany`: `(userId, type, granularity, bucketStart, source)` // is the composite primary key so the planner picks the index path // every time. `bucketStart >= since` is the same shape // `readRollupBuckets` uses; we don't carry an upper bound because @@ -227,6 +245,8 @@ async function readGranularity( 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, @@ -238,7 +258,9 @@ async function readGranularity( }, }); if (rows.length === 0) return null; - return rows.map((r) => ({ + // 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) => ({ bucketStart: r.bucketStart, count: r.count, mean: r.mean, diff --git a/src/lib/rollups/measurement-read.ts b/src/lib/rollups/measurement-read.ts index e03f469b3..9db48f5ef 100644 --- a/src/lib/rollups/measurement-read.ts +++ b/src/lib/rollups/measurement-read.ts @@ -24,12 +24,27 @@ import type { MeasurementSource, MeasurementType, } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; import { getSourceLadder, parseSourcePriority, } from "@/lib/validations/source-priority"; +/** + * v1.11.1 — load a user's source-priority blob for the rollup collapse. + * `null` makes `collapseRollupRowsBySource` fall back to the default ladders. + * Callers that read many types in a loop should load it once and thread it + * through, rather than paying one lookup per read. + */ +export async function loadUserSourcePriority(userId: string): Promise { + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { sourcePriorityJson: true }, + }); + return user?.sourcePriorityJson ?? null; +} + export interface DailyMeanRow { day: Date; count: number; diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 8457ed781..855108353 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -30,6 +30,10 @@ import { prisma } from "@/lib/db"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { annotate } from "@/lib/logging/context"; +import { + collapseRollupRowsBySource, + loadUserSourcePriority, +} from "@/lib/rollups/measurement-read"; import { startOfUtcDay } from "@/lib/tz/start-of-utc-day"; import type { MeasurementSource, @@ -262,6 +266,11 @@ 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. + * 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. */ export async function readRollupBuckets( userId: string, @@ -269,6 +278,7 @@ export async function readRollupBuckets( granularity: RollupGranularity, from: Date, to: Date, + userPriorityJson?: unknown, ): Promise< Array<{ bucketStart: Date; @@ -291,7 +301,11 @@ export async function readRollupBuckets( }, orderBy: { bucketStart: "asc" }, }); - return rows.map((r) => ({ + const priority = + userPriorityJson !== undefined + ? userPriorityJson + : await loadUserSourcePriority(userId); + return collapseRollupRowsBySource(rows, type, priority).map((r) => ({ bucketStart: r.bucketStart, count: r.count, mean: r.mean,