fix(series): follow the source priority ladder in the dense and raw series reads

The rollup tier and sleep resolve every (type, day) to the source the person's
ladder picks before they aggregate. The dense and raw reads of
/api/measurements/series filtered on user and type only, so with a second
provider a day's value blended the two, a reading present in both showed
twice, and the configured ladder had no effect on those charts.

Both reads now take the ladder-winning source per day through the shared
canonicalMeasurementsCte helper, as the other live readers do:

- the day and hour bucket read of the dense kinds (pulse, glucose, HRV, SpO2),
  and the stats strip over its rows;
- the generic raw read (weight, body fat, steps, resting heart rate, ...).

The collapse is per day for the hour grain too, as everywhere else. The helper
gains an optional single-type scope so a series scans one metric, not every row
the person has in the window.

Only a type that has a ladder is collapsed. For a type without one (glucose,
body water, bone mass) the helper breaks the tie by source name, which would
silently hide one of two real sources, so those keep every source as before.
A single-source account reads exactly what is stored.

Tests: a real-database test covering the ladder in both directions, the
per-day fallback, a reading mirrored into two sources, the day-bucket, hour-
bucket and raw reads, a type with no ladder and a single-source account. The
route's unit test file reads its fixtures through a small adapter for the new
raw statements, and the dense-read guard drops the exception for the removed
findMany. The OpenAPI description states the rule.

Refs #1108

docs/api/openapi.yaml

@@ -3885,7 +3885,10 @@ paths:
         explicit `unit` token and a summary `stats` block. `bloodPressure` pairs systolic + diastolic (`secondary`)
         within ±5 min so one fetch renders the dual-line chart; `sleep` collapses per-stage rows into one per-night
         point carrying time-asleep in hours (`sleepStages` holds the per-stage breakdown) and is internally capped to
-        365 days. `glucose` values + `unit` resolve to the user's mg/dL|mmol/L preference. Auth via cookie or Bearer.
+        365 days. `glucose` values + `unit` resolve to the user's mg/dL|mmol/L preference. Where a metric has a
+        source-priority ladder (pulse, weight, SpO2, HRV, ...), each day shows the readings of the source the person's
+        ladder picks for that day, so a second provider never blends into or duplicates a day; a metric without a ladder
+        (glucose, body water, bone mass) keeps every source. Auth via cookie or Bearer.
       parameters:
         - in: query
           name: kind

src/__tests__/dense-measurement-read-guard.test.ts

@@ -101,8 +101,6 @@ const ALLOWED: Record<string, string> = {
     "Health-score inputs: steps (drained to one row per day nightly), sleep, waist, weight, blood pressure, fasting glucose only.",
   'app/api/analytics/route.ts::"BLOOD_GLUCOSE"':
     "The 30-day per-context glucose summary, the same window and bound as the dashboard and coach glucose panels: at a sensor's fixed rate, below 45 000 rows.",
-  "app/api/measurements/series/route.ts::(shorthand)":
-    "The raw branch of the per-kind series. Every dense kind reaches it only inside 90 days and under 10 000 rows (counted first); the sparse kinds are a few readings a day.",
   "app/api/measurements/batch/route.ts::p.row.type as MeasurementType":
     "Cross-source merge probe: one ±tolerance window around each posted reading's instant, so the rows found are bounded by the batch size.",
   "app/api/measurements/route.ts::(shorthand)":

src/app/api/measurements/series/__tests__/route.test.ts

@@ -9,6 +9,9 @@ vi.mock("@/lib/db", () => ({
     // v1.28.25 — dense kinds (glucose / pulse) day-bucket in SQL for
     // windows beyond the 90-day raw cap.
     $queryRaw: vi.fn(),
+    // The series reads go through the ladder-winning source per day (see
+    // `seriesRowsFrom`), which is raw SQL with the user id as `$1`.
+    $queryRawUnsafe: vi.fn(),
   },
 }));
 
@@ -65,8 +68,47 @@ function req(query: string): NextRequest {
   return new NextRequest(`http://localhost/api/measurements/series?${query}`);
 }
 
+/**
+ * The route reads its series through `$queryRawUnsafe` (the ladder-winning
+ * source per day). This file's fixtures are written as rows per type and as the
+ * two aggregate results, so the adapter routes each statement back to them:
+ * the day/hour bucket and the stats aggregate take the next `$queryRaw` result,
+ * and a raw read asks `findMany` for the type named in the SQL and returns the
+ * rows in the column names the statement selects. The SQL itself (which source
+ * wins, which rows count) is covered against a real database in
+ * `tests/integration/series-source-ladder.test.ts`.
+ */
+function adaptSeriesSql(sql: string): Promise<unknown> {
+  if (/GROUP BY local_day|STDDEV_POP/.test(sql)) {
+    return (prisma.$queryRaw as unknown as () => Promise<unknown>)();
+  }
+  const type = /'([A-Z0-9_]+)'::"measurement_type"/.exec(sql)?.[1];
+  const find = prisma.measurement.findMany as unknown as (
+    a: unknown,
+  ) => Promise<
+    Array<{
+      id: string;
+      value: number;
+      measuredAt: Date;
+      valueMin?: number | null;
+      valueMax?: number | null;
+    }>
+  >;
+  return find({ where: { type } }).then((rows) =>
+    rows.map((r) => ({
+      id: r.id,
+      value: r.value,
+      measured_at: r.measuredAt,
+      value_min: r.valueMin ?? null,
+      value_max: r.valueMax ?? null,
+    })),
+  );
+}
+
 beforeEach(() => {
   vi.resetAllMocks();
+  vi.mocked(prisma.$queryRawUnsafe).mockImplementation(((sql: string) =>
+    adaptSeriesSql(sql)) as never);
   vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
   vi.mocked(prisma.measurement.count).mockResolvedValue(0 as never);
   vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never);
@@ -216,14 +258,14 @@ describe("GET /api/measurements/series", () => {
     expect(body.data.points[1].valueMin).toBeNull();
     expect(body.data.points[1].valueMax).toBeNull();
 
-    // The pulse read selects the spread columns.
-    const selectArg = vi.mocked(prisma.measurement.findMany).mock.calls[0][0];
-    expect(
-      (selectArg as { select: Record<string, boolean> }).select.valueMin,
-    ).toBe(true);
-    expect(
-      (selectArg as { select: Record<string, boolean> }).select.valueMax,
-    ).toBe(true);
+    // The pulse read selects the spread columns, and reads the pulse rows of
+    // this person's ladder-winning source (the user id is the first parameter).
+    const [sql, userId] = vi.mocked(prisma.$queryRawUnsafe).mock
+      .calls[0] as unknown as [string, string];
+    expect(sql).toContain('m."value_min"');
+    expect(sql).toContain('m."value_max"');
+    expect(sql).toContain(`'PULSE'::"measurement_type"`);
+    expect(userId).toBe("user-1");
   });
 
   it("collapses sleep stage rows into one night point in hours (v1.11.4)", async () => {
@@ -627,8 +669,12 @@ describe("GET /api/measurements/series — dense-kind day-bucketing (v1.28.25)",
     expect(body.data.stats.count).toBe(43_200);
     expect(body.data.stats.mean).toBe(72.1);
     // The bucket read folds per hour, not per day.
-    const bucketSql = vi.mocked(prisma.$queryRaw).mock.calls[0] as unknown[];
-    expect(bucketSql.slice(1)).toContain("hour");
+    const bucketCall = vi
+      .mocked(prisma.$queryRawUnsafe)
+      .mock.calls.find(([sql]) =>
+        /GROUP BY local_day/.test(sql as string),
+      ) as unknown[];
+    expect(bucketCall.slice(1)).toContain("hour");
   });
 
   it("hour-buckets a CGM glucose window too dense to send raw, in the user's unit", async () => {
@@ -690,8 +736,12 @@ describe("GET /api/measurements/series — dense-kind day-bucketing (v1.28.25)",
       max: 14.4,
       stdDev: 2,
     });
-    const bucketSql = vi.mocked(prisma.$queryRaw).mock.calls[0] as unknown[];
-    expect(bucketSql.slice(1)).toContain("hour");
+    const bucketCall = vi
+      .mocked(prisma.$queryRawUnsafe)
+      .mock.calls.find(([sql]) =>
+        /GROUP BY local_day/.test(sql as string),
+      ) as unknown[];
+    expect(bucketCall.slice(1)).toContain("hour");
   });
 
   it("keeps fingerstick glucose at or under the row cap raw", async () => {

src/app/api/measurements/series/__tests__/unknown-query-param-open.test.ts

@@ -22,6 +22,7 @@ vi.mock("@/lib/db", () => ({
     user: { findUnique: vi.fn() },
     auditLog: { create: vi.fn() },
     $queryRaw: vi.fn(),
+    $queryRawUnsafe: vi.fn(),
   },
 }));
 
@@ -74,6 +75,7 @@ beforeEach(() => {
   vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
   vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never);
   vi.mocked(prisma.$queryRaw).mockResolvedValue([] as never);
+  vi.mocked(prisma.$queryRawUnsafe).mockResolvedValue([] as never);
   vi.mocked(prisma.auditLog.create).mockResolvedValue({} as never);
   vi.mocked(prisma.user.findUnique).mockResolvedValue({
     glucoseUnit: "mg/dL",

src/app/api/measurements/series/route.ts

@@ -26,6 +26,7 @@ import type { MeasurementType, SleepStage } from "@/generated/prisma/client";
 import { reconstructSleepNights } from "@/lib/analytics/sleep-night";
 import { VALUE_RANGES } from "@/lib/validations/measurement";
 import { loadUserSourcePriority } from "@/lib/rollups/measurement-read";
+import { seriesRowsFrom } from "@/lib/measurements/series-canonical";
 import { resolveUserTimezone, userDayKey } from "@/lib/tz/resolver";
 import { convertGlucose, resolveGlucoseUnit } from "@/lib/glucose";
 
@@ -399,49 +400,57 @@ export const GET = apiHandler(async (request: NextRequest) => {
     // v1.28.25 — long-window read of a sample-dense kind (CGM glucose,
     // per-sample / hourly pulse). Day-bucket in SQL instead of walking
     // every raw row into JS: one aggregate pass in Postgres returns at
-    // most `days` rows regardless of sample density. Parameter-bound
-    // tagged-template `$queryRaw`, mirroring the rollup tier's
-    // `date_trunc(... ) GROUP BY` aggregate (measurement-rollups.ts).
-    const userTz = await resolveUserTimezone(user.id);
+    // most `days` rows regardless of sample density. Bound parameters
+    // mirror the rollup tier's `date_trunc(... ) GROUP BY` aggregate
+    // (measurement-rollups.ts), and the rows are the ladder-winning source per
+    // day, the same rule the rollup tier and sleep apply.
+    const [userTz, priorityJson] = await Promise.all([
+      resolveUserTimezone(user.id),
+      loadUserSourcePriority(user.id),
+    ]);
     const type = KIND_TO_TYPE[kind];
-    const bucketRows = await prisma.$queryRaw<
+    const seriesRows = seriesRowsFrom(priorityJson, type, days);
+    const bucketRows = await prisma.$queryRawUnsafe<
       Array<{
         bucket_start: Date;
         mean: number;
         min_value: number;
         max_value: number;
       }>
-    >`
+    >(
+      `
       WITH localized AS (
         SELECT
           m."value",
           m."value_min",
           m."value_max",
           date_trunc(
-            ${grain},
-            (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE ${userTz}
+            $2,
+            (m."measured_at" AT TIME ZONE 'UTC') AT TIME ZONE $3
           ) AS local_day
-        FROM measurements m
-        WHERE m."user_id" = ${user.id}
-          AND m."type" = ${type}::"measurement_type"
-          AND m."measured_at" >= ${since}
-          AND m."deleted_at" IS NULL
+        FROM ${seriesRows}
+        WHERE m."measured_at" >= $4
       )
       SELECT
-        local_day AT TIME ZONE ${userTz}                         AS bucket_start,
+        local_day AT TIME ZONE $3                                AS bucket_start,
         AVG("value")::double precision                           AS mean,
         MIN(COALESCE("value_min", "value"))::double precision    AS min_value,
         MAX(COALESCE("value_max", "value"))::double precision    AS max_value
       FROM localized
       GROUP BY local_day
       ORDER BY bucket_start ASC
-    `;
+    `,
+      user.id,
+      grain,
+      userTz,
+      since,
+    );
     // Stats stay aggregated over the RAW rows (not the day buckets) so the
     // strip's mean/min/max/stdDev/count are the same figures the raw path
     // computed: `AVG`/`MIN`/`MAX` over `value`, `STDDEV_POP` matching the
     // JS population-variance helper, exact row count. Aggregate-only —
     // no row transfer.
-    const [rawAgg] = await prisma.$queryRaw<
+    const [rawAgg] = await prisma.$queryRawUnsafe<
       Array<{
         n: number;
         mean: number | null;
@@ -449,19 +458,20 @@ export const GET = apiHandler(async (request: NextRequest) => {
         max: number | null;
         sd: number | null;
       }>
-    >`
+    >(
+      `
       SELECT
         COUNT(*)::int                            AS n,
         AVG(m."value")::double precision         AS mean,
         MIN(m."value")::double precision         AS min,
         MAX(m."value")::double precision         AS max,
         STDDEV_POP(m."value")::double precision  AS sd
-      FROM measurements m
-      WHERE m."user_id" = ${user.id}
-        AND m."type" = ${type}::"measurement_type"
-        AND m."measured_at" >= ${since}
-        AND m."deleted_at" IS NULL
-    `;
+      FROM ${seriesRows}
+      WHERE m."measured_at" >= $2
+    `,
+      user.id,
+      since,
+    );
     const round2 = (v: number) => Math.round(v * 100) / 100;
     const dayId = (d: Date) =>
       grain === "day"
@@ -538,21 +548,37 @@ export const GET = apiHandler(async (request: NextRequest) => {
     // hourly average. Only PULSE rows ever carry a non-null spread (the
     // hourly HR bucket); every other kind selects the bare value shape.
     const includeSpread = kind === "pulse";
-    const rows = await prisma.measurement.findMany({
-      where: {
-        userId: user.id,
-        type,
-        measuredAt: { gte: since },
-        deletedAt: null,
-      },
-      orderBy: { measuredAt: "asc" },
-      select: {
-        id: true,
-        value: true,
-        measuredAt: true,
-        ...(includeSpread ? { valueMin: true, valueMax: true } : {}),
-      },
-    });
+    // The ladder-winning source per day (see `seriesRowsFrom`), so a second
+    // provider neither blends into nor duplicates the series. Bounded as before:
+    // every dense kind reaches this read only inside 90 days and under 10 000
+    // rows (counted first), and the sparse kinds are a few readings a day.
+    const sourcePriority = await loadUserSourcePriority(user.id);
+    const rawRows = await prisma.$queryRawUnsafe<
+      Array<{
+        id: string;
+        value: number;
+        measured_at: Date;
+        value_min: number | null;
+        value_max: number | null;
+      }>
+    >(
+      `
+      SELECT m."id", m."value", m."measured_at", m."value_min", m."value_max"
+      FROM ${seriesRowsFrom(sourcePriority, type, days)}
+      WHERE m."measured_at" >= $2
+      ORDER BY m."measured_at" ASC, m."id" ASC
+    `,
+      user.id,
+      since,
+    );
+    const rows = rawRows.map((r) => ({
+      id: r.id,
+      value: r.value,
+      measuredAt: r.measured_at,
+      ...(includeSpread
+        ? { valueMin: r.value_min, valueMax: r.value_max }
+        : {}),
+    }));
     if (kind === "glucose") {
       // v1.16.16 — glucose is stored canonical mg/dL; convert each point to
       // the user's display unit AT SERIALIZATION so the wire DTO is unit-

src/lib/analytics/source-rank-sql.ts

@@ -123,11 +123,13 @@ function assertInterval(value: string): string {
  * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with
  * unqualified `"type"`/`"source"` columns. `sinceInterval` (e.g. `"90 days"`)
  * is whitelisted and, when given, scopes both the inner pick and the outer
- * filter to a trailing window.
+ * filter to a trailing window. `onlyType`, when given, scopes both halves to
+ * that one metric.
  */
 export function canonicalMeasurementsCte(
   rankUnqualified: string,
   sinceInterval?: string,
+  onlyType?: MeasurementType,
 ): string {
   const sinceInner = sinceInterval
     ? `AND "measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'`
@@ -135,6 +137,15 @@ export function canonicalMeasurementsCte(
   const sinceOuter = sinceInterval
     ? `AND mm."measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'`
     : "";
+  // A single-type reader (a series for one metric) scopes both halves to that
+  // type, so the pick scans one metric's rows instead of every row the user
+  // has in the window.
+  const typeInner = onlyType
+    ? `AND "type" = '${assertEnumLiteral(onlyType)}'::"measurement_type"`
+    : "";
+  const typeOuter = onlyType
+    ? `AND mm."type" = '${assertEnumLiteral(onlyType)}'::"measurement_type"`
+    : "";
   return `
         SELECT mm.*
         FROM measurements mm
@@ -147,6 +158,7 @@ export function canonicalMeasurementsCte(
           WHERE "user_id" = $1
             AND "deleted_at" IS NULL
             ${sinceInner}
+            ${typeInner}
           ORDER BY "type", date_trunc('day', "measured_at"), (${rankUnqualified}), "source"::text
         ) c
           ON c.t = mm."type"
@@ -154,7 +166,8 @@ export function canonicalMeasurementsCte(
           AND c.canon = mm."source"
         WHERE mm."user_id" = $1
           AND mm."deleted_at" IS NULL
-          ${sinceOuter}`;
+          ${sinceOuter}
+          ${typeOuter}`;
 }
 
 /**
@@ -173,7 +186,8 @@ export function canonicalMeasurementsCte(
 export function canonicalMeasurementsFrom(
   rankUnqualified: string,
   sinceInterval?: string,
+  onlyType?: MeasurementType,
 ): string {
-  return `(${canonicalMeasurementsCte(rankUnqualified, sinceInterval)}
+  return `(${canonicalMeasurementsCte(rankUnqualified, sinceInterval, onlyType)}
       ) m`;
 }

src/lib/measurements/__tests__/series-canonical.test.ts

@@ -0,0 +1,74 @@
+import { describe, expect, it } from "vitest";
+
+import { canonicalMeasurementsCte } from "@/lib/analytics/source-rank-sql";
+
+import { seriesRowsFrom } from "../series-canonical";
+
+describe("seriesRowsFrom", () => {
+  it("collapses a type that has a ladder to the ladder's source per day", () => {
+    const sql = seriesRowsFrom(null, "PULSE", 30);
+    expect(sql).toContain("DISTINCT ON");
+    // Scoped to the one metric, in both halves of the helper.
+    expect(sql).toContain(`"type" = 'PULSE'::"measurement_type"`);
+    expect(sql).toContain(`mm."type" = 'PULSE'::"measurement_type"`);
+    expect(sql.trimEnd().endsWith(") m")).toBe(true);
+  });
+
+  it("reads one day wider than the request, so the edge day is picked from all its readings", () => {
+    expect(seriesRowsFrom(null, "WEIGHT", 30)).toContain("INTERVAL '31 days'");
+    expect(seriesRowsFrom(null, "WEIGHT", 3650)).toContain(
+      "INTERVAL '3651 days'",
+    );
+  });
+
+  it("applies the person's ladder, not only the default one", () => {
+    const first = seriesRowsFrom(
+      { pulse: ["FITBIT", "APPLE_HEALTH"] },
+      "PULSE",
+      7,
+    );
+    const second = seriesRowsFrom(
+      { pulse: ["APPLE_HEALTH", "FITBIT"] },
+      "PULSE",
+      7,
+    );
+    expect(first).not.toEqual(second);
+    expect(first).toMatch(/WHEN 'FITBIT' THEN 0 WHEN 'APPLE_HEALTH' THEN 1/);
+    expect(second).toMatch(/WHEN 'APPLE_HEALTH' THEN 0 WHEN 'FITBIT' THEN 1/);
+  });
+
+  it.each(["BLOOD_GLUCOSE", "TOTAL_BODY_WATER", "BONE_MASS"] as const)(
+    "keeps every source for %s, which has no ladder",
+    (type) => {
+      const sql = seriesRowsFrom(null, type, 30);
+      expect(sql).not.toContain("DISTINCT ON");
+      expect(sql).toContain(`mm."type" = '${type}'::"measurement_type"`);
+      expect(sql).toContain(`mm."user_id" = $1`);
+      expect(sql).toContain(`mm."deleted_at" IS NULL`);
+    },
+  );
+
+  it.each([0, -1, 1.5, 3651, Number.NaN])("refuses days = %s", (days) => {
+    expect(() => seriesRowsFrom(null, "PULSE", days)).toThrow(/days/);
+  });
+});
+
+describe("canonicalMeasurementsCte, one-metric scope", () => {
+  const rank = `CASE "source" WHEN 'A' THEN 0 ELSE 90 END`;
+
+  it("is unchanged when no type is given", () => {
+    const sql = canonicalMeasurementsCte(rank, "90 days");
+    expect(sql).not.toContain("measurement_type");
+  });
+
+  it("scopes the pick and the join to the type when one is given", () => {
+    const sql = canonicalMeasurementsCte(rank, "90 days", "PULSE");
+    expect(sql.match(/'PULSE'::"measurement_type"/g)).toHaveLength(2);
+  });
+
+  it("refuses anything that is not a closed-enum literal", () => {
+    expect(() =>
+      canonicalMeasurementsCte(rank, undefined, "PULSE'; DROP" as never),
+    ).toThrow(/unsafe enum literal/);
+  });
+});

src/lib/measurements/series-canonical.ts

@@ -0,0 +1,56 @@
+/**
+ * The rows a metric series reads, restricted to the source the user's priority
+ * ladder picks for each day.
+ *
+ * The rollup tier and sleep already resolve every (type, day) to the
+ * ladder-winning source before they aggregate. The dense and raw reads of
+ * `/api/measurements/series` filtered on user and type only, so with a second
+ * provider a day's value blended the two and the configured ladder had no
+ * effect on those charts. This is the one place that applies the same rule to
+ * them, through the same SQL helper the other live readers use.
+ *
+ * Only a type that HAS a ladder is collapsed (`RANKED_TYPES`). For a type
+ * without one (glucose, body water, bone mass) the helper would break the tie
+ * by source name, which silently hides one of two real sources, so those keep
+ * every source exactly as before.
+ *
+ * Returns a FROM-clause subquery aliased `m`; the user id is bound as `$1`.
+ */
+import type { MeasurementType } from "@/generated/prisma/client";
+import {
+  RANKED_TYPES,
+  buildSourceRankCase,
+  canonicalMeasurementsFrom,
+} from "@/lib/analytics/source-rank-sql";
+
+const ENUM_RE = /^[A-Z0-9_]+$/;
+
+export function seriesRowsFrom(
+  priorityJson: unknown,
+  type: MeasurementType,
+  days: number,
+): string {
+  if (!ENUM_RE.test(type)) {
+    throw new Error(`unsafe enum literal for SQL splice: ${type}`);
+  }
+  if (!Number.isInteger(days) || days < 1 || days > 3650) {
+    throw new Error(`days out of range for SQL splice: ${days}`);
+  }
+  if (!RANKED_TYPES.includes(type)) {
+    return `(
+        SELECT mm.*
+        FROM measurements mm
+        WHERE mm."user_id" = $1
+          AND mm."type" = '${type}'::"measurement_type"
+          AND mm."deleted_at" IS NULL
+      ) m`;
+  }
+  // One day wider than the request: the caller trims to its exact `since`, so
+  // the edge day's source is picked from all of its readings, not from the
+  // slice the interval happens to include.
+  return canonicalMeasurementsFrom(
+    buildSourceRankCase(priorityJson, '"type"', '"source"'),
+    `${days + 1} days`,
+    type,
+  );
+}

src/lib/openapi/routes/measurements.ts

@@ -840,7 +840,7 @@ export const measurementPaths: NonNullable<ZodOpenApiObject["paths"]> = {
       tags: ["Measurements"],
       summary: "Per-kind time series (iOS chart source)",
       description:
-        "Maps a camelCase `kind` to the canonical MeasurementType(s) and returns an ordered point series with an explicit `unit` token and a summary `stats` block. `bloodPressure` pairs systolic + diastolic (`secondary`) within ±5 min so one fetch renders the dual-line chart; `sleep` collapses per-stage rows into one per-night point carrying time-asleep in hours (`sleepStages` holds the per-stage breakdown) and is internally capped to 365 days. `glucose` values + `unit` resolve to the user's mg/dL|mmol/L preference. Auth via cookie or Bearer.",
+        "Maps a camelCase `kind` to the canonical MeasurementType(s) and returns an ordered point series with an explicit `unit` token and a summary `stats` block. `bloodPressure` pairs systolic + diastolic (`secondary`) within ±5 min so one fetch renders the dual-line chart; `sleep` collapses per-stage rows into one per-night point carrying time-asleep in hours (`sleepStages` holds the per-stage breakdown) and is internally capped to 365 days. `glucose` values + `unit` resolve to the user's mg/dL|mmol/L preference. Where a metric has a source-priority ladder (pulse, weight, SpO2, HRV, ...), each day shows the readings of the source the person's ladder picks for that day, so a second provider never blends into or duplicates a day; a metric without a ladder (glucose, body water, bone mass) keeps every source. Auth via cookie or Bearer.",
       requestParams: {
         query: seriesQuerySchema,
       },

tests/integration/series-source-ladder.test.ts

@@ -0,0 +1,286 @@
+/**
+ * The per-kind series follows the source-priority ladder.
+ *
+ * The rollup tier and sleep resolve every (type, day) to the source the
+ * person's ladder picks before they aggregate. The dense and raw reads of
+ * `/api/measurements/series` filtered on user and type only, so with a second
+ * provider a day's value blended the two, a reading present in both showed
+ * twice, and the ladder had no effect on those charts.
+ *
+ * What is asserted, against a real Postgres:
+ *   - the ladder decides which source a day shows, in both directions;
+ *   - the rule is per day, so a day only the second source covers still shows;
+ *   - a reading mirrored into two sources appears once;
+ *   - the day-bucket, hour-bucket and raw reads all follow it, and the stats
+ *     strip counts the canonical rows;
+ *   - a type with no ladder keeps every source (the collapse would otherwise
+ *     break the tie by source name and hide one of two real sources);
+ *   - a single-source account reads exactly what is stored.
+ */
+import { NextRequest } from "next/server";
+import { beforeAll, describe, expect, it, vi } from "vitest";
+
+import { cookieJar } from "./mock-next-headers";
+import { getPrismaClient, truncateAllTables } from "./setup";
+import type {
+  MeasurementSource,
+  MeasurementType,
+  Prisma,
+} from "@/generated/prisma/client";
+import { invalidateUserTimezone } from "@/lib/tz/resolver";
+
+vi.mock("next/headers", async () => {
+  const { cookieJar, headerJar } = await import("./mock-next-headers");
+  return {
+    headers: vi.fn(async () => ({
+      get: (name: string) => headerJar.get(name.toLowerCase()) ?? null,
+    })),
+    cookies: vi.fn(async () => ({
+      get: (name: string) => {
+        const value = cookieJar.get(name);
+        return value ? { name, value } : undefined;
+      },
+      set: (name: string, value: string) => cookieJar.set(name, value),
+      delete: (name: string) => cookieJar.delete(name),
+    })),
+  };
+});
+
+vi.mock("@/lib/db-compat", () => ({
+  ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
+}));
+
+const prisma = getPrismaClient();
+const TZ = "Europe/Berlin";
+const DUAL = "ladder-dual";
+const DENSE = "ladder-dense";
+const SOLO = "ladder-solo";
+const NOW = new Date(Math.floor(Date.now() / 60_000) * 60_000 - 60_000);
+const DAY = 86_400_000;
+
+/**
+ * Noon UTC of the n-th day back (plus a few minutes), so a reading falls on one
+ * UTC day and one Berlin day whenever the suite runs. Anchoring on "now" would
+ * let a reading cross midnight and make the per-day source pick depend on the
+ * clock.
+ */
+const TODAY_NOON_UTC = Date.UTC(
+  NOW.getUTCFullYear(),
+  NOW.getUTCMonth(),
+  NOW.getUTCDate(),
+  12,
+);
+const at = (daysBack: number, minuteOffset = 0) =>
+  new Date(TODAY_NOON_UTC - daysBack * DAY + minuteOffset * 60_000);
+
+let seq = 0;
+const row = (
+  userId: string,
+  type: MeasurementType,
+  unit: string,
+  source: MeasurementSource,
+  value: number,
+  measuredAt: Date,
+): Prisma.MeasurementCreateManyInput => ({
+  id: `ladder-${++seq}`,
+  userId,
+  type,
+  unit,
+  source,
+  value,
+  measuredAt,
+});
+
+async function setLadder(userId: string, ladder: Record<string, string[]>) {
+  await prisma.user.update({
+    where: { id: userId },
+    data: { sourcePriorityJson: ladder },
+  });
+}
+
+async function series(userId: string, kind: string, days: number) {
+  const s = await prisma.session.create({
+    data: { userId, expiresAt: new Date(Date.now() + 3_600_000) },
+  });
+  cookieJar.clear();
+  cookieJar.set("healthlog_session", s.id);
+  const { GET } = await import("@/app/api/measurements/series/route");
+  const res = await GET(
+    new NextRequest(
+      `http://localhost/api/measurements/series?kind=${kind}&days=${days}`,
+    ),
+  );
+  expect(res.status).toBe(200);
+  return (
+    (await res.json()) as {
+      data: {
+        points: Array<{ id: string; at: string; value: number }>;
+        stats: { count: number; mean: number; min: number; max: number };
+      };
+    }
+  ).data;
+}
+
+beforeAll(async () => {
+  await truncateAllTables(prisma);
+  for (const id of [DUAL, DENSE, SOLO]) {
+    await prisma.user.create({ data: { id, username: id, timezone: TZ } });
+    invalidateUserTimezone(id);
+  }
+
+  // Pulse: Apple Health (60-64) covers days 1-3, Fitbit (90-94) covers days 3-5,
+  // so day 3 has both sources and days 4-5 only Fitbit.
+  const dual: Prisma.MeasurementCreateManyInput[] = [];
+  for (const d of [1, 2, 3]) {
+    [60, 62, 64].forEach((v, i) =>
+      dual.push(row(DUAL, "PULSE", "bpm", "APPLE_HEALTH", v, at(d, i))),
+    );
+  }
+  for (const d of [3, 4, 5]) {
+    [90, 92, 94].forEach((v, i) =>
+      dual.push(row(DUAL, "PULSE", "bpm", "FITBIT", v, at(d, i + 3))),
+    );
+  }
+  // Weight: the same weighing recorded by hand AND by the scale, three days.
+  for (const d of [1, 2, 3]) {
+    dual.push(row(DUAL, "WEIGHT", "kg", "MANUAL", 80 + d, at(d)));
+    dual.push(row(DUAL, "WEIGHT", "kg", "WITHINGS", 80 + d, at(d)));
+  }
+  // Glucose has no ladder: both sources must stay.
+  for (const d of [1, 2]) {
+    dual.push(row(DUAL, "BLOOD_GLUCOSE", "mg/dL", "APPLE_HEALTH", 100, at(d)));
+    dual.push(row(DUAL, "BLOOD_GLUCOSE", "mg/dL", "NIGHTSCOUT", 110, at(d, 1)));
+  }
+  await prisma.measurement.createMany({ data: dual });
+
+  // A dense two-source pulse stream: past the 10 000-row cap, so it is read
+  // as hour buckets. Apple is a steady 70, Fitbit a steady 100.
+  for (const [source, value] of [
+    ["APPLE_HEALTH", 70],
+    ["FITBIT", 100],
+  ] as const) {
+    await prisma.$executeRawUnsafe(
+      `INSERT INTO measurements (
+         id, user_id, type, value, unit, source, measured_at,
+         created_at, updated_at, sync_version)
+       SELECT $4 || lpad(g::text, 7, '0'), $1, 'PULSE'::measurement_type,
+         $5, 'bpm', $6::measurement_source,
+         ($2::timestamptz AT TIME ZONE 'UTC') - (g * interval '1 minute'),
+         now(), now(), 1
+       FROM generate_series(0, $3::int - 1) g`,
+      DENSE,
+      NOW.toISOString(),
+      6_000,
+      `dn-${source}-`,
+      value,
+      source,
+    );
+  }
+
+  // A single-source account.
+  await prisma.measurement.createMany({
+    data: [1, 2, 3, 4].map((d) =>
+      row(SOLO, "WEIGHT", "kg", "MANUAL", 70 + d, at(d)),
+    ),
+  });
+  await prisma.$executeRawUnsafe(`ANALYZE measurements`);
+}, 120_000);
+
+describe("pulse follows the ladder (raw read, 30 days)", () => {
+  it("shows Apple Health when it ranks first, and fills the days only Fitbit covers", async () => {
+    await setLadder(DUAL, { pulse: ["APPLE_HEALTH", "FITBIT"] });
+    const data = await series(DUAL, "pulse", 30);
+    // Days 1-3: Apple (60-64). Day 3 is shared, so Fitbit's day-3 rows are out.
+    // Days 4-5: only Fitbit exists, so Fitbit shows.
+    const values = data.points.map((p) => p.value).sort((a, b) => a - b);
+    expect(values).toEqual([
+      60, 60, 60, 62, 62, 62, 64, 64, 64, 90, 90, 92, 92, 94, 94,
+    ]);
+    expect(data.stats.count).toBe(15);
+  });
+
+  it("shows Fitbit when it ranks first, and fills the days only Apple covers", async () => {
+    await setLadder(DUAL, { pulse: ["FITBIT", "APPLE_HEALTH"] });
+    const data = await series(DUAL, "pulse", 30);
+    const values = data.points.map((p) => p.value).sort((a, b) => a - b);
+    // Days 3-5: Fitbit. Days 1-2: only Apple exists.
+    expect(values).toEqual([
+      60, 60, 62, 62, 64, 64, 90, 90, 90, 92, 92, 92, 94, 94, 94,
+    ]);
+    expect(data.stats.count).toBe(15);
+  });
+
+  it("does not blend the two on a shared day (day-bucket read, 365 days)", async () => {
+    await setLadder(DUAL, { pulse: ["APPLE_HEALTH", "FITBIT"] });
+    const data = await series(DUAL, "pulse", 365);
+    const byDay = data.points.map((p) => p.value).sort((a, b) => a - b);
+    // One mean per local day: Apple's days 1-3 are 62, Fitbit's days 4-5 are 92.
+    expect(byDay).toEqual([62, 62, 62, 92, 92]);
+    expect(data.stats.count).toBe(15);
+  });
+});
+
+describe("a reading mirrored into two sources", () => {
+  it("appears once", async () => {
+    await setLadder(DUAL, { weight: ["WITHINGS", "MANUAL"] });
+    const data = await series(DUAL, "weight", 30);
+    expect(data.points).toHaveLength(3);
+    expect(data.points.map((p) => p.value).sort()).toEqual([81, 82, 83]);
+    expect(data.stats.count).toBe(3);
+  });
+
+  it("follows the ladder for which copy it keeps", async () => {
+    await setLadder(DUAL, { weight: ["MANUAL", "WITHINGS"] });
+    const manual = await series(DUAL, "weight", 30);
+    const rows = await prisma.measurement.findMany({
+      where: { userId: DUAL, type: "WEIGHT", source: "MANUAL" },
+      select: { id: true },
+    });
+    expect(new Set(manual.points.map((p) => p.id))).toEqual(
+      new Set(rows.map((r) => r.id)),
+    );
+  });
+});
+
+describe("an hour-bucket read (past the 10 000-row cap)", () => {
+  it("shows only the ladder's source in every bucket", async () => {
+    await setLadder(DENSE, { pulse: ["APPLE_HEALTH", "FITBIT"] });
+    const apple = await series(DENSE, "pulse", 30);
+    expect(apple.points.length).toBeGreaterThan(50);
+    expect(apple.points.every((p) => p.id.startsWith("hour:"))).toBe(true);
+    expect(apple.points.every((p) => p.value === 70)).toBe(true);
+    expect(apple.stats.count).toBe(6_000);
+
+    await setLadder(DENSE, { pulse: ["FITBIT", "APPLE_HEALTH"] });
+    const fitbit = await series(DENSE, "pulse", 30);
+    expect(fitbit.points.every((p) => p.value === 100)).toBe(true);
+    expect(fitbit.stats.count).toBe(6_000);
+  });
+});
+
+describe("a type with no ladder", () => {
+  it("keeps every source (glucose)", async () => {
+    const data = await series(DUAL, "glucose", 30);
+    expect(data.points).toHaveLength(4);
+    expect(data.stats.count).toBe(4);
+  });
+});
+
+describe("a single-source account", () => {
+  it("reads exactly what is stored", async () => {
+    const data = await series(SOLO, "weight", 30);
+    const raw = await prisma.measurement.findMany({
+      where: { userId: SOLO, type: "WEIGHT" },
+      orderBy: { measuredAt: "asc" },
+      select: { id: true, value: true, measuredAt: true },
+    });
+    expect(data.points).toEqual(
+      raw.map((r) => ({
+        id: r.id,
+        at: r.measuredAt.toISOString(),
+        value: r.value,
+        secondary: null,
+      })),
+    );
+  });
+});