feat(rollups): collapse overlapping sources across all read paths

Complete the source-aware rollup read path: the slim summaries slice
(rollup + live), the comprehensive aggregator (both paths, including the
dailyByType correlation feed), and the /api/measurements route (rollup
read + the live date_trunc fallback) now resolve overlapping sources to
the source-priority ladder's canonical reading per (type, day) before
aggregating. A dual-source vital (e.g. WHOOP + Apple Watch resting heart
rate) surfaces one canonical series instead of a blend or a doubled
count; cumulative metrics keep summing per canonical source.

The collapse runs in SQL where the aggregate must stay a single
round-trip (buildSourceRankCase emits a per-user ladder-rank CASE;
canonicalMeasurementsFrom restricts a live read to canonical-source rows)
and in application code (collapseRollupRowsBySource) for the row readers,
both driven by the same resolved ladders so live and rollup agree.

Also widens the enum-literal SQL whitelist to accept digits so VO2_MAX
no longer throws on the typed write-hook recompute or the rank CASE.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/app/api/analytics/__tests__/route.test.ts

@@ -44,7 +44,12 @@ vi.mock("@/lib/db", () => ({
       findFirst: vi.fn().mockResolvedValue(null),
       deleteMany: vi.fn().mockResolvedValue({ count: 0 }),
       upsert: vi.fn().mockResolvedValue({}),
+      // v1.11.1 — the writer now delete-then-inserts via createMany.
+      createMany: vi.fn().mockResolvedValue({ count: 0 }),
     },
+    // v1.11.1 — the source-aware rollup readers load the user's
+    // source-priority blob; null → default ladders.
+    user: { findUnique: vi.fn().mockResolvedValue(null) },
     $transaction: vi.fn().mockImplementation(async (queries: unknown[]) => {
       if (Array.isArray(queries)) return Promise.all(queries);
       return undefined;
@@ -256,8 +261,13 @@ describe("GET /api/analytics", () => {
     // two parallel ones (all-time + windowed), so the cold fixture
     // now mocks 4 `$queryRaw` calls: coverage + allTime + windowed +
     // latest.
-    vi.mocked(prisma.$queryRaw)
-      .mockResolvedValueOnce([{ n: BigInt(0) }] as never)
+    // The coverage probe rides `$queryRaw` (n:0 → cold path). v1.11.1 — the
+    // cold path's data aggregates (allTime + windowed + latest) now run via
+    // `$queryRawUnsafe` (the source-aware collapse splices a whitelisted CASE).
+    vi.mocked(prisma.$queryRaw).mockResolvedValueOnce([
+      { n: BigInt(0) },
+    ] as never);
+    vi.mocked(prisma.$queryRawUnsafe as any)
       .mockResolvedValueOnce([
         {
           type: "WEIGHT",

src/app/api/measurements/__tests__/range-aggregation-route.test.ts

@@ -23,6 +23,11 @@ vi.mock("@/lib/db", () => ({
     measurementRollup: {
       findMany: vi.fn(),
     },
+    // v1.11.1 — the rollup path calls `loadUserSourcePriority`
+    // (prisma.user.findUnique) to build the source-rank ladders before
+    // `collapseRollupRowsBySource` runs. This is a separate read from the
+    // date_trunc `$queryRaw`; `null` → default ladders.
+    user: { findUnique: vi.fn() },
     $queryRaw: vi.fn(),
   },
 }));
@@ -77,6 +82,9 @@ function getRequest(query: string): NextRequest {
 beforeEach(() => {
   vi.resetAllMocks();
   vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
+  // v1.11.1 — null source-priority blob → default rank ladders for the
+  // rollup-path source collapse.
+  vi.mocked(prisma.user.findUnique).mockResolvedValue(null as never);
 });
 
 describe("GET /api/measurements — aggregation gate (C2)", () => {
@@ -223,6 +231,8 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
       bucketStart: new Date(Date.UTC(2026, 3, 16 + i, 0, 0, 0)),
       mean: 81 + i * 0.05,
       count: 5,
+      // v1.11.1 — single source per day → collapse is identity.
+      source: "APPLE_HEALTH",
     }));
     vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue(
       buckets as never,
@@ -261,6 +271,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
       mean: 2000, // would yield 10000 if multiplied by count=5
       count: 5,
       sumValue: 11000 + i * 250, // distinct from mean × count
+      source: "APPLE_HEALTH",
     }));
     vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue(
       buckets as never,
@@ -292,6 +303,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
         mean: 2000,
         count: 5,
         sumValue: null,
+        source: "APPLE_HEALTH",
       },
     ];
     vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue(
@@ -354,6 +366,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
         mean: 128,
         count: 1,
         sumValue: null,
+        source: "APPLE_HEALTH",
       },
       {
         type: "BLOOD_PRESSURE_SYS",
@@ -361,6 +374,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
         mean: 132,
         count: 1,
         sumValue: null,
+        source: "APPLE_HEALTH",
       },
     ];
     const convergedRollup = Array.from({ length: 28 }, (_, i) => ({
@@ -369,6 +383,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
       mean: 128 + (i % 5),
       count: 2,
       sumValue: null,
+      source: "APPLE_HEALTH",
     }));
     // First read returns the sparse rows; second (post-fold) read
     // returns the converged rows.
@@ -416,6 +431,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
       mean: 81 + i * 0.05,
       count: 2,
       sumValue: null,
+      source: "APPLE_HEALTH",
     }));
     vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue(
       buckets as never,
@@ -444,6 +460,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
         mean: 128,
         count: 1,
         sumValue: null,
+        source: "APPLE_HEALTH",
       },
       {
         type: "BLOOD_PRESSURE_SYS",
@@ -451,6 +468,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
         mean: 132,
         count: 1,
         sumValue: null,
+        source: "APPLE_HEALTH",
       },
     ];
     vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue(
@@ -484,6 +502,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
         mean: 128,
         count: 1,
         sumValue: null,
+        source: "APPLE_HEALTH",
       },
       {
         type: "BLOOD_PRESSURE_SYS",
@@ -491,6 +510,7 @@ describe("GET /api/measurements — all-time semantics (SD-H1)", () => {
         mean: 132,
         count: 1,
         sumValue: null,
+        source: "APPLE_HEALTH",
       },
     ];
     vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue(

src/app/api/measurements/route.ts

@@ -34,6 +34,11 @@ import {
   collapseToTypeDayKeys,
   recomputeUserRollups,
 } from "@/lib/rollups/measurement-rollups";
+import {
+  collapseRollupRowsBySource,
+  loadUserSourcePriority,
+} from "@/lib/rollups/measurement-read";
+import { buildSourceRankCase } from "@/lib/analytics/source-rank-sql";
 import { NextRequest } from "next/server";
 import type {
   MeasurementType,
@@ -308,30 +313,40 @@ export const GET = apiHandler(async (request: NextRequest) => {
     to
   ) {
     const cap = Math.min(limit, BUCKET_CAP.daily);
-    const rollupRows = await prisma.measurementRollup.findMany({
-      where: {
-        userId: user.id,
-        type: type as MeasurementType,
-        granularity: "DAY",
-        bucketStart: { gte: from, lte: to },
-      },
-      orderBy: { bucketStart: "asc" },
-      take: cap,
-      select: {
-        type: true,
-        bucketStart: true,
-        mean: true,
-        count: true,
-        // v1.4.39 W-SUM — the writer now populates `sum_value` on
-        // every fold. The cumulative metric path consumes the column
-        // directly instead of reconstructing `mean * count`; the
-        // legacy fallback below covers the boot-backfill convergence
-        // window when an existing bucket pre-dates v1.4.39.
-        sumValue: true,
-        minValue: true,
-        maxValue: true,
-      },
-    });
+    // v1.11.1 — rollup rows are per source; collapse overlapping sources to the
+    // ladder-canonical reading per day before the chart consumes them, so a
+    // dual-source vital paints one line, not two stacked points per day.
+    const priorityJson = await loadUserSourcePriority(user.id);
+    const rollupRows = collapseRollupRowsBySource(
+      await prisma.measurementRollup.findMany({
+        where: {
+          userId: user.id,
+          type: type as MeasurementType,
+          granularity: "DAY",
+          bucketStart: { gte: from, lte: to },
+        },
+        orderBy: { bucketStart: "asc" },
+        // Fetch beyond the bucket cap pre-collapse — N sources per day can
+        // exceed `cap` rows before the per-day collapse reduces them.
+        select: {
+          type: true,
+          source: true,
+          bucketStart: true,
+          mean: true,
+          count: true,
+          // v1.4.39 W-SUM — the writer now populates `sum_value` on
+          // every fold. The cumulative metric path consumes the column
+          // directly instead of reconstructing `mean * count`; the
+          // legacy fallback below covers the boot-backfill convergence
+          // window when an existing bucket pre-dates v1.4.39.
+          sumValue: true,
+          minValue: true,
+          maxValue: true,
+        },
+      }),
+      type as MeasurementType,
+      priorityJson,
+    ).slice(0, cap);
     // v1.4.39.2 — coverage-mismatch fallback. v1.4.39.1 wired the
     // rollup write hook into the previously-bypassed Withings sync,
     // /api/import, and admin-restore paths, and widened the boot-time
@@ -403,26 +418,31 @@ export const GET = apiHandler(async (request: NextRequest) => {
           });
           // Re-read so this request returns the converged rows. The
           // next chart paint hits the same warm rollup without paying
-          // the fallback cost.
-          effectiveRows = await prisma.measurementRollup.findMany({
-            where: {
-              userId: user.id,
-              type: type as MeasurementType,
-              granularity: "DAY",
-              bucketStart: { gte: from, lte: to },
-            },
-            orderBy: { bucketStart: "asc" },
-            take: cap,
-            select: {
-              type: true,
-              bucketStart: true,
-              mean: true,
-              count: true,
-              sumValue: true,
-              minValue: true,
-              maxValue: true,
-            },
-          });
+          // the fallback cost. v1.11.1 — collapse per-source rows to the
+          // canonical reading per day, same as the initial read.
+          effectiveRows = collapseRollupRowsBySource(
+            await prisma.measurementRollup.findMany({
+              where: {
+                userId: user.id,
+                type: type as MeasurementType,
+                granularity: "DAY",
+                bucketStart: { gte: from, lte: to },
+              },
+              orderBy: { bucketStart: "asc" },
+              select: {
+                type: true,
+                source: true,
+                bucketStart: true,
+                mean: true,
+                count: true,
+                sumValue: true,
+                minValue: true,
+                maxValue: true,
+              },
+            }),
+            type as MeasurementType,
+            priorityJson,
+          ).slice(0, cap);
         } catch (err) {
           // Best-effort — fall back to the legacy live aggregate path
           // below if the fold itself failed so the chart paints
@@ -520,20 +540,48 @@ export const GET = apiHandler(async (request: NextRequest) => {
     const aggregator = useSum
       ? Prisma.raw(`SUM(m."value")::double precision`)
       : Prisma.raw(`AVG(m."value")::double precision`);
+    // v1.11.1 — collapse overlapping sources to the ladder-canonical reading
+    // per (type, bucket) before aggregating, so this cold/uncovered fallback
+    // agrees with the warm rollup path. `canon` picks the winning source for
+    // each (type, bucket) via the per-user rank; the join keeps only its rows.
+    const priorityJson = await loadUserSourcePriority(user.id);
+    const rankRaw = Prisma.raw(
+      buildSourceRankCase(priorityJson, 'm."type"', 'm."source"'),
+    );
+    const typeFilter = type
+      ? Prisma.sql`AND m."type" = ${type}::measurement_type`
+      : Prisma.empty;
     const buckets = await prisma.$queryRaw<
       Array<{ type: string; bucket_start: Date; avg: number; cnt: number }>
     >`
+      WITH canon AS (
+        SELECT DISTINCT ON (m."type", date_trunc(${truncUnitLiteral}, m."measured_at"))
+          m."type"                                          AS t,
+          date_trunc(${truncUnitLiteral}, m."measured_at")  AS d,
+          m."source"                                        AS canon
+        FROM measurements m
+        WHERE m."user_id" = ${user.id}
+          AND m."measured_at" >= ${from}
+          AND m."measured_at" <= ${to}
+          AND m."deleted_at" IS NULL
+          ${typeFilter}
+        ORDER BY m."type", date_trunc(${truncUnitLiteral}, m."measured_at"), (${rankRaw}), m."source"
+      )
       SELECT
         m."type"::text AS type,
         date_trunc(${truncUnitLiteral}, m."measured_at") AS bucket_start,
         ${aggregator} AS avg,
         COUNT(*)::int AS cnt
       FROM measurements m
+      JOIN canon c
+        ON c.t = m."type"
+        AND c.d = date_trunc(${truncUnitLiteral}, m."measured_at")
+        AND c.canon = m."source"
       WHERE m."user_id" = ${user.id}
         AND m."measured_at" >= ${from}
         AND m."measured_at" <= ${to}
         AND m."deleted_at" IS NULL
-        ${type ? Prisma.sql`AND m."type" = ${type}::measurement_type` : Prisma.empty}
+        ${typeFilter}
       GROUP BY m."type", bucket_start
       ORDER BY bucket_start ASC
       LIMIT ${cap}