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) <noreply@anthropic.com>

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(() => {

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<RollupBucketRow[] | null> {
-  // 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,

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<unknown> {
+  const user = await prisma.user.findUnique({
+    where: { id: userId },
+    select: { sourcePriorityJson: true },
+  });
+  return user?.sourcePriorityJson ?? null;
+}
+
 export interface DailyMeanRow {
   day: Date;
   count: number;

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,