diff --git a/src/app/api/analytics/__tests__/route.test.ts b/src/app/api/analytics/__tests__/route.test.ts index f499922c7..b74119b60 100644 --- a/src/app/api/analytics/__tests__/route.test.ts +++ b/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", diff --git a/src/app/api/measurements/__tests__/range-aggregation-route.test.ts b/src/app/api/measurements/__tests__/range-aggregation-route.test.ts index fddbbb06e..6f92b73eb 100644 --- a/src/app/api/measurements/__tests__/range-aggregation-route.test.ts +++ b/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( diff --git a/src/app/api/measurements/route.ts b/src/app/api/measurements/route.ts index ce3a032fb..11c5939b3 100644 --- a/src/app/api/measurements/route.ts +++ b/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} diff --git a/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts b/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts index 5e781df45..871bbaa32 100644 --- a/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts +++ b/src/lib/analytics/__tests__/bp-in-target-fast-path.test.ts @@ -12,6 +12,9 @@ vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // v1.11.1 — readRollupBuckets lazy-loads the source-priority blob; null + // (default findUnique) falls back to the default ladders. + user: { findUnique: vi.fn() }, }, })); diff --git a/src/lib/analytics/__tests__/correlations-fast-path.test.ts b/src/lib/analytics/__tests__/correlations-fast-path.test.ts index d8acd1865..6add0b368 100644 --- a/src/lib/analytics/__tests__/correlations-fast-path.test.ts +++ b/src/lib/analytics/__tests__/correlations-fast-path.test.ts @@ -13,6 +13,8 @@ vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // v1.11.1 — readRollupBuckets lazy-loads the source-priority blob. + user: { findUnique: vi.fn() }, moodEntry: { findMany: vi.fn() }, medicationIntakeEvent: { findMany: vi.fn() }, }, diff --git a/src/lib/analytics/__tests__/health-score-fast-path.test.ts b/src/lib/analytics/__tests__/health-score-fast-path.test.ts index d7f645e6d..3f5d47a22 100644 --- a/src/lib/analytics/__tests__/health-score-fast-path.test.ts +++ b/src/lib/analytics/__tests__/health-score-fast-path.test.ts @@ -12,6 +12,8 @@ vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany: vi.fn() }, measurementRollup: { findMany: vi.fn() }, + // v1.11.1 — readRollupBuckets lazy-loads the source-priority blob. + user: { findUnique: vi.fn() }, moodEntry: { findMany: vi.fn() }, medication: { findMany: vi.fn() }, medicationIntakeEvent: { findMany: vi.fn() }, diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 53d1b2286..7dcee10a7 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -10,6 +10,16 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: vi.fn(), + // v1.11.1 — the data-aggregate queries inside `computeFromRollups` + // and `computeFromLiveAggregate` now splice a whitelisted + // source-rank CASE and bind `userId` as `$1`, so they run via + // `$queryRawUnsafe(sql, userId)` rather than the tagged-template + // `$queryRaw`. The coverage probe stays on `$queryRaw`. + $queryRawUnsafe: vi.fn(), + // v1.11.1 — `loadUserSourcePriority` reads the user's + // `sourcePriorityJson` to build the rank ladders. `null` here → + // default ladders. + user: { findUnique: vi.fn() }, measurement: { findFirst: vi.fn() }, // v1.4.36 — slim slice reads DAY buckets from `measurement_rollups` // on the happy path. The freshness watermark inside @@ -26,22 +36,63 @@ vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn(), })); +// v1.11.1 — `computeFromRollups` / `computeFromLiveAggregate` build a +// source-rank CASE via `@/lib/analytics/source-rank-sql` and splice it +// into the data-aggregate SQL. The builder's own correctness is pinned +// in its dedicated suite (and the integration suite runs the real SQL); +// here we stub it to deterministic, side-effect-free fragments so the +// slice's plumbing — path selection, slope/round/empty contracts, the +// 90-day FILTER caps the slice itself writes — is exercised without +// coupling the slice unit test to the rank builder's enum-whitelist +// internals. +vi.mock("@/lib/analytics/source-rank-sql", () => ({ + buildSourceRankCase: vi.fn(() => "90"), + canonicalMeasurementsFrom: vi.fn( + (_rank: string, sinceInterval?: string) => + `( + SELECT mm.* + FROM measurements mm + WHERE mm."user_id" = $1 + AND mm."deleted_at" IS NULL + ${ + sinceInterval + ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'` + : "" + } + ) m`, + ), +})); + import { prisma } from "@/lib/db"; +import { canonicalMeasurementsFrom } from "@/lib/analytics/source-rank-sql"; import { computeSummariesSlice } from "../summaries-slice"; const RAW = prisma.$queryRaw as unknown as ReturnType; +const UNSAFE = prisma.$queryRawUnsafe as unknown as ReturnType; +const USER_FIND_UNIQUE = + prisma.user.findUnique as unknown as ReturnType; const MEASUREMENT_FIND_FIRST = prisma.measurement.findFirst as unknown as ReturnType; const ROLLUP_FIND_MANY = prisma.measurementRollup.findMany as unknown as ReturnType; const ROLLUP_FIND_FIRST = prisma.measurementRollup.findFirst as unknown as ReturnType; +const CANONICAL_FROM = + canonicalMeasurementsFrom as unknown as ReturnType; beforeEach(() => { RAW.mockReset(); + UNSAFE.mockReset(); + USER_FIND_UNIQUE.mockReset(); MEASUREMENT_FIND_FIRST.mockReset(); ROLLUP_FIND_MANY.mockReset(); ROLLUP_FIND_FIRST.mockReset(); + // clear (not reset) — preserve the FROM-clause stub implementation, + // drop cross-test call history so the cap assertion only sees this + // test's calls. + CANONICAL_FROM.mockClear(); + // null → loadUserSourcePriority returns null → default rank ladders. + USER_FIND_UNIQUE.mockResolvedValue(null); ROLLUP_FIND_MANY.mockResolvedValue([]); ROLLUP_FIND_FIRST.mockResolvedValue(null); MEASUREMENT_FIND_FIRST.mockResolvedValue(null); @@ -54,15 +105,14 @@ afterEach(() => { describe("computeSummariesSlice", () => { describe("cold fallback — empty rollup table", () => { it("returns the empty-summary skeleton when the user has no rows", async () => { - // v1.4.48 M0 — cold path now issues two aggregate queries - // (`allTime` + `windowed`) instead of one fat aggregate, so the - // $queryRaw call count is 4 instead of 3: - // 1. per-type coverage probe — empty ⇒ cold path. - // 2. all-time aggregate ($queryRaw) — empty. - // 3. windowed aggregate ($queryRaw, 90-day cap) — empty. - // 4. latests ($queryRaw) — empty. - RAW.mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) + // v1.11.1 — the coverage probe stays on `$queryRaw` (1 RAW call); + // the three data-aggregate queries moved to `$queryRawUnsafe`: + // 1. per-type coverage probe ($queryRaw) — empty ⇒ cold path. + // 2. all-time aggregate ($queryRawUnsafe) — empty. + // 3. windowed aggregate ($queryRawUnsafe, 90-day cap) — empty. + // 4. latests ($queryRawUnsafe) — empty. + RAW.mockResolvedValueOnce([]); + UNSAFE.mockResolvedValueOnce([]) .mockResolvedValueOnce([]) .mockResolvedValueOnce([]); @@ -85,7 +135,9 @@ describe("computeSummariesSlice", () => { avg30LastYear: null, }); expect(result.bmi).toBeNull(); - expect(RAW).toHaveBeenCalledTimes(4); + // 1 RAW coverage probe + 3 UNSAFE data queries. + expect(RAW).toHaveBeenCalledTimes(1); + expect(UNSAFE).toHaveBeenCalledTimes(3); }); it("maps a populated heavy aggregate row into the DataSummary shape on cold path", async () => { @@ -95,16 +147,16 @@ describe("computeSummariesSlice", () => { // 2. all-time aggregate (count / min / max / mean) // 3. windowed aggregate (avg7/30 + slope/r²) // 4. latests - RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - count: BigInt(42), - min_value: 79.2, - max_value: 84.1, - mean_value: 82.05, - }, - ]) + RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(42), + min_value: 79.2, + max_value: 84.1, + mean_value: 82.05, + }, + ]) .mockResolvedValueOnce([ { type: "WEIGHT", @@ -153,16 +205,16 @@ describe("computeSummariesSlice", () => { }); it("returns a null slope tuple when the SQL slope is null (insufficient rows)", async () => { - RAW.mockResolvedValueOnce([{ type: "PULSE", has_buckets: false }]) - .mockResolvedValueOnce([ - { - type: "PULSE", - count: BigInt(1), - min_value: 72, - max_value: 72, - mean_value: 72, - }, - ]) + RAW.mockResolvedValueOnce([{ type: "PULSE", has_buckets: false }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "PULSE", + count: BigInt(1), + min_value: 72, + max_value: 72, + mean_value: 72, + }, + ]) .mockResolvedValueOnce([ { type: "PULSE", @@ -188,16 +240,16 @@ describe("computeSummariesSlice", () => { it("surfaces lastSeenByType from the DISTINCT ON pass's measured_at", async () => { const tenDaysAgo = new Date(Date.now() - 10 * 24 * 60 * 60 * 1000); - RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - count: BigInt(5), - min_value: 80, - max_value: 84, - mean_value: 82, - }, - ]) + RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(5), + min_value: 80, + max_value: 84, + mean_value: 82, + }, + ]) .mockResolvedValueOnce([ { type: "WEIGHT", @@ -225,16 +277,16 @@ describe("computeSummariesSlice", () => { }); it("seeds the latest value from the DISTINCT ON pass per type", async () => { - RAW.mockResolvedValueOnce([{ type: "PULSE", has_buckets: false }]) - .mockResolvedValueOnce([ - { - type: "PULSE", - count: BigInt(3), - min_value: 60, - max_value: 95, - mean_value: 77, - }, - ]) + RAW.mockResolvedValueOnce([{ type: "PULSE", has_buckets: false }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "PULSE", + count: BigInt(3), + min_value: 60, + max_value: 95, + mean_value: 77, + }, + ]) .mockResolvedValueOnce([ { type: "PULSE", @@ -260,29 +312,29 @@ describe("computeSummariesSlice", () => { describe("rollup-fresh happy path", () => { it("composes count/min/max/mean from the per-type rollup GROUP BY without running the heavy aggregate", async () => { - // v1.4.37.2 — the slim slice's rollup read is now a per-type - // GROUP BY ($queryRaw) instead of a row-per-bucket findMany, - // so the mock sequence is: - // 1. per-type coverage probe — WEIGHT fully covered ⇒ happy path. - // 2. narrow aggregate — windowed/regression only. - // 3. latests. - // 4. rollup GROUP BY — one row per type with count/min/max/mean - // already composed server-side. - RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - avg7: 82, - avg30: 82.5, - median: 82.1, - slope7: 0.02, - r2_7: 0.5, - slope30: 0.01, - r2_30: 0.4, - slope90: 0.005, - r2_90: 0.2, - }, - ]) + // v1.11.1 — the per-type coverage probe stays on `$queryRaw`; the + // three rollup-path data queries moved to `$queryRawUnsafe`: + // 1. per-type coverage probe ($queryRaw) — WEIGHT fully covered + // ⇒ happy path. + // 2. narrow aggregate ($queryRawUnsafe) — windowed/regression only. + // 3. latests ($queryRawUnsafe). + // 4. rollup GROUP BY ($queryRawUnsafe) — one row per type with + // count/min/max/mean already composed server-side. + RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + avg7: 82, + avg30: 82.5, + median: 82.1, + slope7: 0.02, + r2_7: 0.5, + slope30: 0.01, + r2_30: 0.4, + slope90: 0.005, + r2_90: 0.2, + }, + ]) .mockResolvedValueOnce([ { type: "WEIGHT", value: 82.7, measured_at: new Date() }, ]) @@ -320,9 +372,11 @@ describe("computeSummariesSlice", () => { confidence: 0.5, }); - // probe + narrow aggregate + latests + rollup GROUP BY - // (v1.4.37.2 — the prior `findMany` is gone). No heavy aggregate. - expect(RAW).toHaveBeenCalledTimes(4); + // 1 RAW coverage probe + 3 UNSAFE data queries (narrow aggregate + // + latests + rollup GROUP BY; v1.4.37.2 — the prior `findMany` + // is gone). No heavy aggregate. + expect(RAW).toHaveBeenCalledTimes(1); + expect(UNSAFE).toHaveBeenCalledTimes(3); // v1.4.40 W-WMY-WIRE — the year-ago baseline probe runs // `readBestGranularityRollups(userId, type, 395)` per // type-with-data via `prisma.measurementRollup.findMany`. The @@ -343,20 +397,20 @@ describe("computeSummariesSlice", () => { describe("year-over-year wiring (avg30LastYear)", () => { it("populates avg30LastYear from MONTH buckets that overlap the year-ago slice", async () => { // Coverage probe shows WEIGHT covered → rollup happy path. - RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - avg7: 82, - avg30: 82.5, - slope7: 0, - r2_7: 0, - slope30: 0, - r2_30: 0, - slope90: 0, - r2_90: 0, - }, - ]) + RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + avg7: 82, + avg30: 82.5, + slope7: 0, + r2_7: 0, + slope30: 0, + r2_30: 0, + slope90: 0, + r2_90: 0, + }, + ]) .mockResolvedValueOnce([ { type: "WEIGHT", value: 82.7, measured_at: new Date() }, ]) @@ -392,20 +446,20 @@ describe("computeSummariesSlice", () => { }); it("leaves avg30LastYear null when no bucket overlaps the year-ago slice", async () => { - RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - avg7: 82, - avg30: 82.5, - slope7: 0, - r2_7: 0, - slope30: 0, - r2_30: 0, - slope90: 0, - r2_90: 0, - }, - ]) + RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + avg7: 82, + avg30: 82.5, + slope7: 0, + r2_7: 0, + slope30: 0, + r2_30: 0, + slope90: 0, + r2_90: 0, + }, + ]) .mockResolvedValueOnce([ { type: "WEIGHT", value: 82.7, measured_at: new Date() }, ]) @@ -437,20 +491,20 @@ describe("computeSummariesSlice", () => { }); it("leaves avg30LastYear null when every granularity misses (no coverage)", async () => { - RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - avg7: 82, - avg30: 82.5, - slope7: 0, - r2_7: 0, - slope30: 0, - r2_30: 0, - slope90: 0, - r2_90: 0, - }, - ]) + RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + avg7: 82, + avg30: 82.5, + slope7: 0, + r2_7: 0, + slope30: 0, + r2_30: 0, + slope90: 0, + r2_90: 0, + }, + ]) .mockResolvedValueOnce([ { type: "WEIGHT", value: 82.7, measured_at: new Date() }, ]) @@ -472,12 +526,12 @@ describe("computeSummariesSlice", () => { it("only probes types that actually have data in the current window", async () => { // v1.4.48 M0 — empty coverage map ⇒ `isFullyCovered` returns - // false ⇒ cold-fallback path. Cold path now issues 4 raw - // queries (coverage probe + all-time + windowed + latests), - // all returning empty arrays here ⇒ no types-with-data ⇒ the - // year-ago probe must NOT fan out. - RAW.mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) + // false ⇒ cold-fallback path. v1.11.1 — coverage probe stays on + // `$queryRaw`; all-time + windowed + latests run via + // `$queryRawUnsafe`. All return empty arrays here ⇒ no + // types-with-data ⇒ the year-ago probe must NOT fan out. + RAW.mockResolvedValueOnce([]); + UNSAFE.mockResolvedValueOnce([]) .mockResolvedValueOnce([]) .mockResolvedValueOnce([]); @@ -499,34 +553,53 @@ describe("computeSummariesSlice", () => { */ describe("90-day outer measured_at cap (v1.4.48 M0)", () => { it("applies the 90-day cap to narrows (rollup-fresh path) and windowed (cold-fallback path)", async () => { + // v1.11.1 — the cap-bearing data queries (narrows / windowed) + // now run via `$queryRawUnsafe(sql, userId)`, so the SQL is a + // plain string arg[0] rather than a tagged-template + // strings array. Capture from the UNSAFE mock. The coverage probe + // (still `$queryRaw`) drives path selection per call. const queries: string[] = []; - RAW.mockImplementation((strings: TemplateStringsArray) => { - queries.push(strings.join("?")); + UNSAFE.mockImplementation((sql: string) => { + queries.push(sql); return Promise.resolve([]); }); // Trigger the rollup-fresh path so we capture the `narrows` SQL. // Coverage probe returns one covered type ⇒ `isFullyCovered` // is true ⇒ `computeFromRollups` runs. - RAW.mockImplementationOnce((strings: TemplateStringsArray) => { - queries.push(strings.join("?")); - return Promise.resolve([{ type: "WEIGHT", has_buckets: true }]); - }); + // Then the cold-fallback path (empty coverage) so we capture the + // `windowed` SQL. + RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]); await computeSummariesSlice("user-rollup-pin"); - // Trigger the cold-fallback path so we capture the `windowed` - // SQL. Empty coverage map ⇒ `isFullyCovered` is false ⇒ - // `computeFromLiveAggregate` runs. + RAW.mockResolvedValueOnce([]); await computeSummariesSlice("user-cold-pin"); const joined = queries.join("\n---\n"); - // Both windowed scans must carry the outer 90-day cap. - const capMatches = joined.match( - /AND m\."measured_at" >= NOW\(\) - INTERVAL '90 days'/g, + // v1.11.1 — the rollup-fresh `narrows` query still writes its + // outer 90-day cap inline, now on the canonical-source subquery's + // raw alias (`mm.`) rather than the outer `m.`. Pin that the cap + // survives the source-rank refactor. + const narrowsCap = joined.match( + /AND mm\."measured_at" >= NOW\(\) - INTERVAL '90 days'/g, + ); + expect(narrowsCap).not.toBeNull(); + expect(narrowsCap?.length).toBeGreaterThanOrEqual(1); + + // v1.11.1 — the cold-fallback `windowed` scan delegates its outer + // 90-day cap to `canonicalMeasurementsFrom(rank, "90 days")`. The + // helper lives in `@/lib/analytics/source-rank-sql` (stubbed + // above), so pin the contract at the call boundary: the slice + // must ask for the 90-day window. The `allTime` aggregate + // deliberately calls it WITHOUT an interval (no cap — all-time + // count/min/max/mean must scan every row). + expect(canonicalMeasurementsFrom).toHaveBeenCalledWith( + expect.any(String), + "90 days", + ); + expect(canonicalMeasurementsFrom).toHaveBeenCalledWith( + expect.any(String), ); - expect(capMatches).not.toBeNull(); - // narrows (rollup-fresh) + windowed (cold-fallback) = 2 caps. - expect(capMatches?.length).toBeGreaterThanOrEqual(2); }); }); }); diff --git a/src/lib/analytics/source-rank-sql.ts b/src/lib/analytics/source-rank-sql.ts new file mode 100644 index 000000000..e2bcc4314 --- /dev/null +++ b/src/lib/analytics/source-rank-sql.ts @@ -0,0 +1,149 @@ +/** + * v1.11.1 — source-aware SQL collapse helpers. + * + * The rollup writer mints one row per (type, day, source); the live + * measurement reads see one raw row per reading. For overlapping standard + * vitals (e.g. WHOOP + Apple Watch resting heart rate) both the rollup + * all-time aggregate and the live 90-day aggregates must resolve to the + * source-priority ladder's canonical source per (type, day) before composing, + * or a dual-source user double-counts / blends two devices. + * + * `collapseRollupRowsBySource` does this in application code for the row-by-row + * readers. These helpers do the equivalent IN SQL so the slim slice + the + * comprehensive aggregator + the live `/api/measurements` fallback keep their + * single-round-trip `GROUP BY type` shape (no six-figure row transfer). + * + * Every spliced literal is a closed-enum value (`MeasurementType` / + * `MeasurementSource`) asserted against `/^[A-Z_]+$/`, the same whitelist + * convention the rollup writer uses for its date-trunc + type-list splices. + * The user id is always passed as a bound parameter, never spliced. + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { metricKeyForType } from "@/lib/measurements/cumulative-day-sum"; +import { + getSourceLadder, + parseSourcePriority, +} from "@/lib/validations/source-priority"; + +/** + * Measurement types that carry a source-priority ladder — the overlapping + * vitals plus the cumulative metrics. Single-source types are intentionally + * absent: they only ever have one source per (type, day), so the collapse is + * a no-op for them and they fall into the CASE's ELSE branch. + */ +const RANKED_TYPES: readonly MeasurementType[] = [ + "RESTING_HEART_RATE", + "HEART_RATE_VARIABILITY", + "RESPIRATORY_RATE", + "OXYGEN_SATURATION", + "BODY_TEMPERATURE", + "SKIN_TEMPERATURE", + "WEIGHT", + "BODY_FAT", + "BLOOD_PRESSURE_SYS", + "BLOOD_PRESSURE_DIA", + "PULSE", + "VO2_MAX", + "RECOVERY_SCORE", + "SLEEP_DURATION", + "ACTIVITY_STEPS", + "ACTIVE_ENERGY_BURNED", + "WALKING_RUNNING_DISTANCE", + "FLIGHTS_CLIMBED", +]; + +// Enum members can carry digits (e.g. VO2_MAX, BLOOD_PRESSURE_SYS) — allow +// 0-9 so the whitelist doesn't reject a legitimate closed-enum value. +const ENUM_RE = /^[A-Z0-9_]+$/; +function assertEnumLiteral(value: string): string { + if (!ENUM_RE.test(value)) { + throw new Error(`unsafe enum literal for SQL splice: ${value}`); + } + return value; +} + +/** + * Build a SQL CASE expression mapping `(typeCol, sourceCol)` to an integer + * rank where 0 = the highest-priority source on that type's ladder. A source + * absent from a type's ladder — and every row of a type without a ladder — + * gets rank 90 so a deterministic tiebreak (`source` name, or `count DESC` + * where available) decides. Resolves the user's `sourcePriorityJson` ladders; + * `null` yields the default ladders. + */ +export function buildSourceRankCase( + priorityJson: unknown, + typeCol: string, + sourceCol: string, +): string { + const resolved = parseSourcePriority(priorityJson); + const branches: string[] = []; + for (const type of RANKED_TYPES) { + const metricKey = metricKeyForType(type); + if (!metricKey) continue; + const ladder = getSourceLadder(resolved, metricKey); + if (ladder.length === 0) continue; + const whens = ladder + .map((source, i) => `WHEN '${assertEnumLiteral(source)}' THEN ${i}`) + .join(" "); + branches.push( + `WHEN ${typeCol} = '${assertEnumLiteral(type)}' THEN ` + + `(CASE ${sourceCol} ${whens} ELSE 90 END)`, + ); + } + if (branches.length === 0) return "90"; + return `CASE ${branches.join(" ")} ELSE 90 END`; +} + +const INTERVAL_RE = /^\d+ (day|days|month|months|year|years)$/; +function assertInterval(value: string): string { + if (!INTERVAL_RE.test(value)) { + throw new Error(`unsafe interval literal for SQL splice: ${value}`); + } + return value; +} + +/** + * Build a FROM-clause subquery (aliased `m`) that restricts raw `measurements` + * to the canonical-source rows per (type, day): the inner `DISTINCT ON` picks + * the ladder-winning source for each day, and the join keeps only that source's + * readings. Use it in place of `FROM measurements m` so a live aggregate over + * an overlapping vital never blends two devices — the same collapse the rollup + * readers apply, kept in lockstep for live/rollup parity. + * + * 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. + */ +export function canonicalMeasurementsFrom( + rankUnqualified: string, + sinceInterval?: string, +): string { + const sinceInner = sinceInterval + ? `AND "measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'` + : ""; + const sinceOuter = sinceInterval + ? `AND mm."measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'` + : ""; + return `( + SELECT mm.* + FROM measurements mm + JOIN ( + SELECT DISTINCT ON ("type", date_trunc('day', "measured_at")) + "type" AS t, + date_trunc('day', "measured_at") AS d, + "source" AS canon + FROM measurements + WHERE "user_id" = $1 + AND "deleted_at" IS NULL + ${sinceInner} + ORDER BY "type", date_trunc('day', "measured_at"), (${rankUnqualified}), "source" + ) c + ON c.t = mm."type" + AND c.d = date_trunc('day', mm."measured_at") + AND c.canon = mm."source" + WHERE mm."user_id" = $1 + AND mm."deleted_at" IS NULL + ${sinceOuter} + ) m`; +} diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 9a1ce8988..33045b551 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -62,6 +62,11 @@ import type { DataSummary } from "@/lib/analytics/trends"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { annotate } from "@/lib/logging/context"; import { ensureUserRollupsFresh } from "@/lib/rollups/measurement-rollups"; +import { loadUserSourcePriority } from "@/lib/rollups/measurement-read"; +import { + buildSourceRankCase, + canonicalMeasurementsFrom, +} from "@/lib/analytics/source-rank-sql"; import { isFullyCovered, probeRollupCoverage, @@ -274,8 +279,22 @@ export async function computeSummariesSlice( * multi-second cold to sub-second; output is bit-identical. */ async function computeFromRollups(userId: string): Promise { + // v1.11.1 — resolve the user's source-priority ladders once, then build the + // rank CASE expressions used to collapse overlapping sources to the canonical + // reading per (type, day) inside each query. `"type"`/`"source"` variants for + // the unqualified rollup + inner-subquery columns, `m."…"` for the latest + // read's outer alias. + const priorityJson = await loadUserSourcePriority(userId); + const rankUnqualified = buildSourceRankCase(priorityJson, '"type"', '"source"'); + const rankM = buildSourceRankCase(priorityJson, 'm."type"', 'm."source"'); + const [narrows, latests, dayBuckets] = await Promise.all([ - prisma.$queryRaw` + // The FROM is restricted to the canonical-source rows per (type, day): the + // inner DISTINCT ON picks the ladder-winning source for each day, and the + // join keeps only that source's readings so the 90-day AVG / median / slope + // never blend two devices that both reported the same vital. + prisma.$queryRawUnsafe( + ` SELECT m."type"::text AS type, AVG(m."value") FILTER ( @@ -329,22 +348,47 @@ async function computeFromRollups(userId: string): Promise { ) FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '90 days' )::double precision AS r2_90 - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."deleted_at" IS NULL - AND m."measured_at" >= NOW() - INTERVAL '90 days' + FROM ( + SELECT mm.* + FROM measurements mm + JOIN ( + SELECT DISTINCT ON ("type", date_trunc('day', "measured_at")) + "type" AS t, + date_trunc('day', "measured_at") AS d, + "source" AS canon + FROM measurements + WHERE "user_id" = $1 + AND "deleted_at" IS NULL + AND "measured_at" >= NOW() - INTERVAL '90 days' + ORDER BY "type", date_trunc('day', "measured_at"), (${rankUnqualified}), "source" + ) c + ON c.t = mm."type" + AND c.d = date_trunc('day', mm."measured_at") + AND c.canon = mm."source" + WHERE mm."user_id" = $1 + AND mm."deleted_at" IS NULL + AND mm."measured_at" >= NOW() - INTERVAL '90 days' + ) m GROUP BY m."type" `, - prisma.$queryRaw` + userId, + ), + // v1.11.1 — the latest tile reflects the canonical source for the latest + // day, matching the chart: order by latest day first, then the ladder rank + // (canonical source wins), then the latest reading of that source. + prisma.$queryRawUnsafe( + ` SELECT DISTINCT ON (m."type") m."type"::text AS type, m."value"::double precision AS value, m."measured_at" AS measured_at FROM measurements m - WHERE m."user_id" = ${userId} + WHERE m."user_id" = $1 AND m."deleted_at" IS NULL - ORDER BY m."type", m."measured_at" DESC + ORDER BY m."type", date_trunc('day', m."measured_at") DESC, (${rankM}), m."measured_at" DESC `, + userId, + ), // v1.4.37.2 hotfix — the v1.4.35 implementation read EVERY DAY // rollup bucket for the user (`findMany` without a `bucketStart` // window) and then composed `count / min / max / mean` in JS. @@ -353,11 +397,26 @@ async function computeFromRollups(userId: string): Promise { // cache miss, even with the rollup table hot. The slim slice's // contract is the all-time count / min / max / mean per type — // exactly what a SQL `GROUP BY type` returns in a single - // round-trip. Returns 8 rows - // instead of 306k, brings the cache-miss cost into the < 100 ms - // budget. The downstream `aggregateBuckets` call is bypassed - // because the per-type aggregate is already shaped server-side. - prisma.$queryRaw` + // round-trip. + // + // v1.11.1 — rows are now per source. Collapse each (type, day) to the + // ladder-canonical source via DISTINCT ON before the all-time aggregate, + // so a dual-source vital is counted once. Still one server-side pass — + // the DISTINCT ON + GROUP BY runs in Postgres and returns one row/type. + prisma.$queryRawUnsafe( + ` + WITH collapsed AS ( + SELECT DISTINCT ON ("type", "bucket_start") + "type" AS type, + "count" AS count, + "min_value" AS min_value, + "max_value" AS max_value, + "mean" AS mean + FROM "measurement_rollups" + WHERE "user_id" = $1 + AND "granularity" = 'DAY' + ORDER BY "type", "bucket_start", (${rankUnqualified}), "count" DESC, "source" + ) SELECT "type"::text AS type, SUM("count")::int AS count, @@ -367,11 +426,11 @@ async function computeFromRollups(userId: string): Promise { SUM("count" * "mean")::double precision / NULLIF(SUM("count")::double precision, 0) ) AS mean - FROM "measurement_rollups" - WHERE "user_id" = ${userId} - AND "granularity" = 'DAY' + FROM collapsed GROUP BY "type" `, + userId, + ), ]); const latestByType = new Map(); @@ -506,20 +565,31 @@ async function computeFromRollups(userId: string): Promise { async function computeFromLiveAggregate( userId: string, ): Promise { + // v1.11.1 — collapse overlapping sources to the ladder-canonical reading per + // (type, day) BEFORE aggregating, exactly as the rollup path does, so a + // coverage-miss tenant gets the same numbers as a warm one (live/rollup + // parity). `canonicalMeasurementsFrom` swaps `FROM measurements m` for a + // canonical-source-filtered subquery; the latest read uses the ladder rank. + const priorityJson = await loadUserSourcePriority(userId); + const rankUnqualified = buildSourceRankCase(priorityJson, '"type"', '"source"'); + const rankM = buildSourceRankCase(priorityJson, 'm."type"', 'm."source"'); + const [allTime, windowed, latests] = await Promise.all([ - prisma.$queryRaw` + prisma.$queryRawUnsafe( + ` SELECT m."type"::text AS type, COUNT(*) AS count, MIN(m."value")::double precision AS min_value, MAX(m."value")::double precision AS max_value, AVG(m."value")::double precision AS mean_value - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."deleted_at" IS NULL + FROM ${canonicalMeasurementsFrom(rankUnqualified)} GROUP BY m."type" `, - prisma.$queryRaw` + userId, + ), + prisma.$queryRawUnsafe( + ` SELECT m."type"::text AS type, AVG(m."value") FILTER ( @@ -573,22 +643,24 @@ async function computeFromLiveAggregate( ) FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '90 days' )::double precision AS r2_90 - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."deleted_at" IS NULL - AND m."measured_at" >= NOW() - INTERVAL '90 days' + FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} GROUP BY m."type" `, - prisma.$queryRaw` + userId, + ), + prisma.$queryRawUnsafe( + ` SELECT DISTINCT ON (m."type") m."type"::text AS type, m."value"::double precision AS value, m."measured_at" AS measured_at FROM measurements m - WHERE m."user_id" = ${userId} + WHERE m."user_id" = $1 AND m."deleted_at" IS NULL - ORDER BY m."type", m."measured_at" DESC + ORDER BY m."type", date_trunc('day', m."measured_at") DESC, (${rankM}), m."measured_at" DESC `, + userId, + ), ]); const latestByType = new Map(); diff --git a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts index 86d87dc4f..3a62e54f4 100644 --- a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts +++ b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts @@ -14,6 +14,17 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: vi.fn(), + // v1.11.1 — the data-aggregate queries inside `buildFromRollups` + // and `buildFromLiveAggregate` now splice a whitelisted source-rank + // CASE and bind `userId` as `$1`, so the narrow/heavy aggregate + + // the latests pass run via `$queryRawUnsafe(sql, userId)` rather + // than the tagged-template `$queryRaw`. The coverage probe and the + // first_at query stay on `$queryRaw`. + $queryRawUnsafe: vi.fn(), + // v1.11.1 — `loadUserSourcePriority` reads the user's + // `sourcePriorityJson` to build the rank ladders. `null` here → + // default ladders. + user: { findUnique: vi.fn() }, measurement: { findMany: vi.fn(), findFirst: vi.fn() }, // v1.4.36 — the rollup read inside the Promise.all + the // freshness watermark inside `ensureUserRollupsFresh`. We mock @@ -29,10 +40,39 @@ vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn(), })); +// v1.11.1 — `buildFromRollups` / `buildFromLiveAggregate` build a +// source-rank CASE via `@/lib/analytics/source-rank-sql` and splice it +// into the data-aggregate SQL. The builder's own correctness is pinned +// in its dedicated suite (and the integration suite runs the real SQL); +// here we stub it to deterministic, side-effect-free fragments so the +// aggregator's plumbing — path selection, bpRaw threading, dailyByType +// composition — is exercised without coupling the unit test to the rank +// builder's enum-whitelist internals. +vi.mock("@/lib/analytics/source-rank-sql", () => ({ + buildSourceRankCase: vi.fn(() => "90"), + canonicalMeasurementsFrom: vi.fn( + (_rank: string, sinceInterval?: string) => + `( + SELECT mm.* + FROM measurements mm + WHERE mm."user_id" = $1 + AND mm."deleted_at" IS NULL + ${ + sinceInterval + ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'` + : "" + } + ) m`, + ), +})); + import { prisma } from "@/lib/db"; import { buildComprehensiveAggregate } from "../comprehensive-aggregator"; const RAW = prisma.$queryRaw as unknown as ReturnType; +const UNSAFE = prisma.$queryRawUnsafe as unknown as ReturnType; +const USER_FIND_UNIQUE = + prisma.user.findUnique as unknown as ReturnType; const FIND_MANY = prisma.measurement.findMany as unknown as ReturnType; const MEASUREMENT_FIND_FIRST = @@ -44,10 +84,14 @@ const ROLLUP_FIND_FIRST = beforeEach(() => { RAW.mockReset(); + UNSAFE.mockReset(); + USER_FIND_UNIQUE.mockReset(); FIND_MANY.mockReset(); MEASUREMENT_FIND_FIRST.mockReset(); ROLLUP_FIND_MANY.mockReset(); ROLLUP_FIND_FIRST.mockReset(); + // null → loadUserSourcePriority returns null → default rank ladders. + USER_FIND_UNIQUE.mockResolvedValue(null); // `ensureUserRollupsFresh` calls `prisma.measurement.findFirst` and // `prisma.measurementRollup.findFirst` in parallel. Default both to // null so the warm-up is a no-op; individual tests opt in to a @@ -65,41 +109,47 @@ describe("buildComprehensiveAggregate", () => { describe("rollup-fresh happy path", () => { it("skips the heavy live aggregate when the rollup table is populated", async () => { const now = new Date(); - // 1. per-type coverage probe — WEIGHT fully covered ⇒ happy path. - // 2. narrow aggregate ($queryRaw) — windowed/regression columns only. - // 3. latests ($queryRaw). - // 4. firstMeasurementAt ($queryRaw). + // v1.11.1 — the coverage probe + the first_at query stay on + // `$queryRaw` (2 RAW calls); the narrow aggregate + the latests + // pass moved to `$queryRawUnsafe`: + // RAW: 1. per-type coverage probe — WEIGHT covered ⇒ happy path. + // 2. firstMeasurementAt. + // UNSAFE: 1. narrow aggregate — windowed/regression columns only. + // 2. latests. RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: true }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - count: BigInt(42), - stddev_value: 1.2, - anomaly_count: BigInt(3), - avg7: 81.9, - avg30: 82.1, - avg30_last_month: 83.0, - slope7: -0.014, - r2_7: 0.65, - slope30: -0.005, - r2_30: 0.42, - slope90: 0.001, - r2_90: 0.12, - }, - ]) - .mockResolvedValueOnce([ - { type: "WEIGHT", value: 81.4, measured_at: now }, - ]) .mockResolvedValueOnce([ { first_at: new Date(now.getTime() - 86400000) }, ]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(42), + stddev_value: 1.2, + anomaly_count: BigInt(3), + avg7: 81.9, + avg30: 82.1, + avg30_last_month: 83.0, + slope7: -0.014, + r2_7: 0.65, + slope30: -0.005, + r2_30: 0.42, + slope90: 0.001, + r2_90: 0.12, + }, + ]).mockResolvedValueOnce([ + { type: "WEIGHT", value: 81.4, measured_at: now }, + ]); // DAY buckets compose to count=42, min=79.2, max=84.1, mean=82.05. // The summary's count/min/max/mean read from these buckets — NOT - // from a heavy live aggregate column. + // from a heavy live aggregate column. v1.11.1 — each bucket row + // now carries a `source` so `partitionBucketsByType` → + // `collapseRollupRowsBySource` can resolve dual-source days; with + // a single source per day every bucket passes through unchanged. ROLLUP_FIND_MANY.mockResolvedValueOnce([ { type: "WEIGHT", + source: "APPLE_HEALTH", bucketStart: new Date("2026-05-10T00:00:00.000Z"), count: 20, mean: 81.0, @@ -108,6 +158,7 @@ describe("buildComprehensiveAggregate", () => { }, { type: "WEIGHT", + source: "APPLE_HEALTH", bucketStart: new Date("2026-05-11T00:00:00.000Z"), count: 22, mean: 83.0, @@ -151,12 +202,12 @@ describe("buildComprehensiveAggregate", () => { ]); // Contract pin — the heavy aggregate path is NOT exercised on the - // rollup-fresh branch. The $queryRaw calls land on the COUNT - // probe + narrow aggregate + DISTINCT-ON latest + first_at, NOT - // the legacy heavy COUNT/MIN/MAX/AVG query. We assert by call - // count + by checking that the narrow projection (no min_value / - // max_value / mean_value cols) is what landed. - expect(RAW).toHaveBeenCalledTimes(4); + // rollup-fresh branch. v1.11.1 — the coverage probe + first_at + // stay on `$queryRaw` (2 calls); the narrow aggregate + the + // DISTINCT-ON latest moved to `$queryRawUnsafe` (2 calls). Neither + // path runs the legacy heavy COUNT/MIN/MAX/AVG query. + expect(RAW).toHaveBeenCalledTimes(2); + expect(UNSAFE).toHaveBeenCalledTimes(2); // v1.4.38 W-F — sys + dia merged into a single round-trip. expect(FIND_MANY).toHaveBeenCalledTimes(1); expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1); @@ -165,13 +216,14 @@ describe("buildComprehensiveAggregate", () => { describe("cold fallback when no rollup buckets exist", () => { it("returns an empty bundle for a user with no measurements and no rollups", async () => { - // 1. per-type coverage probe — empty (no measurements) ⇒ cold path. - // 2. heavy aggregate ($queryRaw) — empty. - // 3. latests ($queryRaw) — empty. + // v1.11.1 — the coverage probe stays on `$queryRaw`; the heavy + // aggregate + latests moved to `$queryRawUnsafe`: + // RAW: 1. per-type coverage probe — empty ⇒ cold path. + // UNSAFE: 1. heavy aggregate — empty. + // 2. latests — empty. // No firstMeasurementAt query when totalMeasurements === 0. - RAW.mockResolvedValueOnce([]) - .mockResolvedValueOnce([]) - .mockResolvedValueOnce([]); + RAW.mockResolvedValueOnce([]); + UNSAFE.mockResolvedValueOnce([]).mockResolvedValueOnce([]); // v1.4.38 W-F — sys + dia merged into one `findMany`. FIND_MANY.mockResolvedValueOnce([]); @@ -183,7 +235,8 @@ describe("buildComprehensiveAggregate", () => { expect(result.dailyByType).toEqual({}); expect(result.firstMeasurementAt).toBeNull(); expect(result.totalMeasurements).toBe(0); - expect(RAW).toHaveBeenCalledTimes(3); + expect(RAW).toHaveBeenCalledTimes(1); + expect(UNSAFE).toHaveBeenCalledTimes(2); // v1.4.38 W-F — sys + dia merged → single findMany. expect(FIND_MANY).toHaveBeenCalledTimes(1); // The cold path's rollup.findMany still fires (in case some @@ -193,37 +246,39 @@ describe("buildComprehensiveAggregate", () => { it("runs the heavy aggregate when no rollup rows exist yet", async () => { const now = new Date(); - // 1. per-type coverage probe — WEIGHT measured but no buckets ⇒ cold path. - // 2. heavy aggregate ($queryRaw) — populated. - // 3. latests ($queryRaw) — populated. - // 4. firstMeasurementAt ($queryRaw). + // v1.11.1 — the coverage probe + first_at stay on `$queryRaw`; the + // heavy aggregate + latests moved to `$queryRawUnsafe`: + // RAW: 1. per-type coverage probe — WEIGHT measured but no + // buckets ⇒ cold path. + // 2. firstMeasurementAt (totalMeasurements > 0). + // UNSAFE: 1. heavy aggregate — populated. + // 2. latests — populated. RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }]) - .mockResolvedValueOnce([ - { - type: "WEIGHT", - count: BigInt(42), - min_value: 79.2, - max_value: 84.1, - mean_value: 82.05, - stddev_value: 1.2, - anomaly_count: BigInt(3), - avg7: 81.9, - avg30: 82.1, - avg30_last_month: 83.0, - slope7: -0.014, - r2_7: 0.65, - slope30: -0.005, - r2_30: 0.42, - slope90: 0.001, - r2_90: 0.12, - }, - ]) - .mockResolvedValueOnce([ - { type: "WEIGHT", value: 81.4, measured_at: now }, - ]) .mockResolvedValueOnce([ { first_at: new Date(now.getTime() - 86400000) }, ]); + UNSAFE.mockResolvedValueOnce([ + { + type: "WEIGHT", + count: BigInt(42), + min_value: 79.2, + max_value: 84.1, + mean_value: 82.05, + stddev_value: 1.2, + anomaly_count: BigInt(3), + avg7: 81.9, + avg30: 82.1, + avg30_last_month: 83.0, + slope7: -0.014, + r2_7: 0.65, + slope30: -0.005, + r2_30: 0.42, + slope90: 0.001, + r2_90: 0.12, + }, + ]).mockResolvedValueOnce([ + { type: "WEIGHT", value: 81.4, measured_at: now }, + ]); // v1.4.38 W-F — sys + dia merged into one round-trip. FIND_MANY.mockResolvedValueOnce([]); @@ -269,35 +324,37 @@ describe("buildComprehensiveAggregate", () => { it("threads sys/dia raw rows through bpRawRows so 5-min pairing survives", async () => { const measuredAt = new Date("2026-05-10T08:00:00Z"); // Cold path so the heavy aggregate fires (BP type lacks coverage). - RAW.mockResolvedValueOnce([{ type: "BLOOD_PRESSURE_SYS", has_buckets: false }]) - .mockResolvedValueOnce([ - { - type: "BLOOD_PRESSURE_SYS", - count: BigInt(1), - min_value: 120, - max_value: 120, - mean_value: 120, - stddev_value: 0, - anomaly_count: BigInt(0), - avg7: 120, - avg30: 120, - avg30_last_month: null, - slope7: null, - r2_7: null, - slope30: null, - r2_30: null, - slope90: null, - r2_90: null, - }, - ]) - .mockResolvedValueOnce([ - { - type: "BLOOD_PRESSURE_SYS", - value: 120, - measured_at: measuredAt, - }, - ]) - .mockResolvedValueOnce([{ first_at: measuredAt }]); + // v1.11.1 — coverage probe + first_at on `$queryRaw`; heavy + latests + // on `$queryRawUnsafe`. + RAW.mockResolvedValueOnce([ + { type: "BLOOD_PRESSURE_SYS", has_buckets: false }, + ]).mockResolvedValueOnce([{ first_at: measuredAt }]); + UNSAFE.mockResolvedValueOnce([ + { + type: "BLOOD_PRESSURE_SYS", + count: BigInt(1), + min_value: 120, + max_value: 120, + mean_value: 120, + stddev_value: 0, + anomaly_count: BigInt(0), + avg7: 120, + avg30: 120, + avg30_last_month: null, + slope7: null, + r2_7: null, + slope30: null, + r2_30: null, + slope90: null, + r2_90: null, + }, + ]).mockResolvedValueOnce([ + { + type: "BLOOD_PRESSURE_SYS", + value: 120, + measured_at: measuredAt, + }, + ]); // v1.4.38 W-F — single merged `findMany` returning both sys + dia // rows tagged by `type`. The aggregator partitions in JS, so the // bpRawRows.sys / .dia shape stays byte-identical. diff --git a/src/lib/insights/__tests__/weight-status.test.ts b/src/lib/insights/__tests__/weight-status.test.ts index 878364eb0..07b454a62 100644 --- a/src/lib/insights/__tests__/weight-status.test.ts +++ b/src/lib/insights/__tests__/weight-status.test.ts @@ -6,6 +6,10 @@ vi.mock("@/lib/db", () => ({ measurement: { findMany: vi.fn() }, measurementRollup: { findMany: vi.fn() }, moodEntry: { findMany: vi.fn() }, + // v1.11.1 — the rollup readers lazy-load the user's + // `sourcePriorityJson` via `loadUserSourcePriority`. `null` here → + // default rank ladders. + user: { findUnique: vi.fn() }, }, })); @@ -47,6 +51,8 @@ beforeEach(() => { // Cold rollup tier: the graded builder folds monthly/yearly from the // full-history `measurement.findMany` fallback the test already mocks. vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue([] as never); + // v1.11.1 — null source-priority blob → default rank ladders. + vi.mocked(prisma.user.findUnique).mockResolvedValue(null as never); }); describe("generateWeightStatusForUser — graded payload", () => { diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index 10ec3e72d..bbf6aa5bb 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -100,8 +100,20 @@ import { prisma } from "@/lib/db"; import type { DataSummary } from "@/lib/analytics/trends"; import { annotate } from "@/lib/logging/context"; +import type { + MeasurementSource, + MeasurementType, +} from "@/generated/prisma/client"; import { ensureUserRollupsFresh } from "@/lib/rollups/measurement-rollups"; -import { aggregateBuckets } from "@/lib/rollups/measurement-read"; +import { + aggregateBuckets, + collapseRollupRowsBySource, + loadUserSourcePriority, +} from "@/lib/rollups/measurement-read"; +import { + buildSourceRankCase, + canonicalMeasurementsFrom, +} from "@/lib/analytics/source-rank-sql"; import { isFullyCovered, probeRollupCoverage, @@ -339,17 +351,24 @@ async function buildFromRollups( ): Promise { // v1.4.38 W-F — bp sys + dia consolidated into one `findMany` // (`type: { in: [...] }`) plus per-sub-query wall-clock timings. + // v1.11.1 — collapse overlapping sources to the ladder-canonical reading per + // (type, day) before aggregating. `rankUnqualified` ranks the inner + // canonical-source subquery; `rankM` ranks the latest read's `m.`-aliased + // columns. The window_stats CTE + the main narrow both read from the + // canonical-source-filtered set so a dual-source vital isn't blended. + const priorityJson = await loadUserSourcePriority(userId); + const rankUnqualified = buildSourceRankCase(priorityJson, '"type"', '"source"'); + const rankM = buildSourceRankCase(priorityJson, 'm."type"', 'm."source"'); + const [narrows, latests, dayBuckets, bpRawRows, firstRows] = await Promise.all([ - timeSubquery(timings, "narrow", () => prisma.$queryRaw` + timeSubquery(timings, "narrow", () => prisma.$queryRawUnsafe( + ` WITH window_stats AS ( SELECT m."type", AVG(m."value") AS mean_value, STDDEV_POP(m."value") AS stddev_value - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} - AND m."deleted_at" IS NULL + FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} GROUP BY m."type" ) SELECT @@ -412,24 +431,26 @@ async function buildFromRollups( ) FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '90 days' )::double precision AS r2_90 - FROM measurements m + FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} JOIN window_stats ws ON ws."type" = m."type" - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} - AND m."deleted_at" IS NULL GROUP BY m."type", ws.stddev_value - `), - timeSubquery(timings, "latests", () => prisma.$queryRaw` + `, + userId, + )), + timeSubquery(timings, "latests", () => prisma.$queryRawUnsafe( + ` SELECT DISTINCT ON (m."type") m."type"::text AS type, m."value"::double precision AS value, m."measured_at" AS measured_at FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} + WHERE m."user_id" = $1 + AND m."measured_at" >= NOW() - INTERVAL '90 days' AND m."deleted_at" IS NULL - ORDER BY m."type", m."measured_at" DESC - `), + ORDER BY m."type", date_trunc('day', m."measured_at") DESC, (${rankM}), m."measured_at" DESC + `, + userId, + )), // v1.4.35 — DAY rollup buckets for the trailing 90 days, every // type. Drives both the per-type `count / min / max / mean` // composition (via `aggregateBuckets`) and the `dailyByType` @@ -466,7 +487,7 @@ async function buildFromRollups( }); } - const bucketsByType = partitionBucketsByType(dayBuckets); + const bucketsByType = partitionBucketsByType(dayBuckets, priorityJson); const narrowByType = new Map(); for (const row of narrows) { narrowByType.set(row.type, row); @@ -536,17 +557,21 @@ async function buildFromLiveAggregate( // v1.4.38 W-F — same sub-query timing + consolidated bp-raw read as // the rollup-fresh path so the prod observability covers cold mounts // too. + // v1.11.1 — same source-aware collapse as the rollup-fresh path so the cold + // start agrees with the warm path (live/rollup parity). + const priorityJson = await loadUserSourcePriority(userId); + const rankUnqualified = buildSourceRankCase(priorityJson, '"type"', '"source"'); + const rankM = buildSourceRankCase(priorityJson, 'm."type"', 'm."source"'); + const [aggregates, latests, dayBuckets, bpRawRows] = await Promise.all([ - timeSubquery(timings, "heavy", () => prisma.$queryRaw` + timeSubquery(timings, "heavy", () => prisma.$queryRawUnsafe( + ` WITH window_stats AS ( SELECT m."type", AVG(m."value") AS mean_value, STDDEV_POP(m."value") AS stddev_value - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} - AND m."deleted_at" IS NULL + FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} GROUP BY m."type" ) SELECT @@ -609,24 +634,26 @@ async function buildFromLiveAggregate( ) FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '90 days' )::double precision AS r2_90 - FROM measurements m + FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} JOIN window_stats ws ON ws."type" = m."type" - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} - AND m."deleted_at" IS NULL GROUP BY m."type", ws.stddev_value - `), - timeSubquery(timings, "latests", () => prisma.$queryRaw` + `, + userId, + )), + timeSubquery(timings, "latests", () => prisma.$queryRawUnsafe( + ` SELECT DISTINCT ON (m."type") m."type"::text AS type, m."value"::double precision AS value, m."measured_at" AS measured_at FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} + WHERE m."user_id" = $1 + AND m."measured_at" >= NOW() - INTERVAL '90 days' AND m."deleted_at" IS NULL - ORDER BY m."type", m."measured_at" DESC - `), + ORDER BY m."type", date_trunc('day', m."measured_at") DESC, (${rankM}), m."measured_at" DESC + `, + userId, + )), // Buckets may exist for some types even when the COUNT probe came // back zero (race window between the probe and a sibling // populator); the cold-path read still honours them for the @@ -651,7 +678,7 @@ async function buildFromLiveAggregate( }); } - const bucketsByType = partitionBucketsByType(dayBuckets); + const bucketsByType = partitionBucketsByType(dayBuckets, priorityJson); const summaries: Record = {}; let totalMeasurements = 0; @@ -713,12 +740,14 @@ async function buildFromLiveAggregate( function partitionBucketsByType( rows: ReadonlyArray<{ type: string; + source: MeasurementSource; bucketStart: Date; count: number; mean: number; minValue: number; maxValue: number; }>, + userPriorityJson: unknown, ): Map< string, Array<{ @@ -729,6 +758,16 @@ function partitionBucketsByType( maxValue: number; }> > { + // v1.11.1 — rollup rows are per source. Group by type, then collapse each + // type's buckets to the ladder-canonical source per day before composing, + // so a dual-source vital is not double-counted in count/mean/dailyByType. + const byType = new Map>(); + for (const b of rows) { + const list = byType.get(b.type) ?? []; + list.push(b); + byType.set(b.type, list); + } + const map = new Map< string, Array<{ @@ -739,16 +778,22 @@ function partitionBucketsByType( maxValue: number; }> >(); - for (const b of rows) { - const list = map.get(b.type) ?? []; - list.push({ - day: b.bucketStart, - count: b.count, - mean: b.mean, - minValue: b.minValue, - maxValue: b.maxValue, - }); - map.set(b.type, list); + for (const [type, typeRows] of byType) { + const collapsed = collapseRollupRowsBySource( + typeRows, + type as MeasurementType, + userPriorityJson, + ); + map.set( + type, + collapsed.map((b) => ({ + day: b.bucketStart, + count: b.count, + mean: b.mean, + minValue: b.minValue, + maxValue: b.maxValue, + })), + ); } return map; } diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index 855108353..be973b5aa 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -353,7 +353,9 @@ async function runRollupAggregate(input: { // from `MeasurementType` (Prisma-generated TS enum); we whitelist // them with a strict regex before splicing into SQL. for (const t of input.types) { - if (!/^[A-Z_]+$/.test(t)) { + // Enum members can carry digits (e.g. VO2_MAX) — allow 0-9 so a + // legitimate type is not rejected on the write-hook's typed recompute. + if (!/^[A-Z0-9_]+$/.test(t)) { throw new Error(`invalid measurement type: ${t}`); } } diff --git a/tests/integration/source-aware-rollups.integration.test.ts b/tests/integration/source-aware-rollups.integration.test.ts new file mode 100644 index 000000000..f180dab39 --- /dev/null +++ b/tests/integration/source-aware-rollups.integration.test.ts @@ -0,0 +1,407 @@ +/** + * v1.11.1 — source-aware measurement rollups, DB round-trip. + * + * The rollup grain gained `source` in its primary key + * (`@@id([userId, type, granularity, bucketStart, source])`): the populator + * now mints ONE row per (type, day, source) instead of one source-blind + * aggregate per (type, day). The read path + * (`readRollupBuckets` → `collapseRollupRowsBySource`) collapses the + * overlapping per-source rows back to ONE row per bucket using the user's + * source-priority ladder before the linear DAY-bucket composition runs. + * + * `collapseRollupRowsBySource` is unit-tested in isolation + * (`src/lib/rollups/__tests__/collapse-rollup-rows-by-source.test.ts`); this + * suite pins the same contract end-to-end through the real Postgres + * testcontainer — the populator SQL groups by `m."source"` and the reader + * lazy-loads the user's ladder from `sourcePriorityJson`. Five contracts: + * + * 1. **Point-metric ladder pick.** RHR with WHOOP + APPLE_HEALTH on one + * day → two raw rows, one collapsed row that surfaces the WHOOP value + * (WHOOP leads the restingHeartRate ladder). + * 2. **Cumulative no-blend.** ACTIVITY_STEPS with APPLE_HEALTH + WITHINGS + * on one day → two raw rows, one collapsed row reflecting ONLY the + * Apple canonical source's count/mean — never the cross-source sum. + * 3. **Native-vs-derived.** RECOVERY_SCORE with WHOOP + COMPUTED → the + * WHOOP (device-native) row wins over the COMPUTED proxy. + * 4. **Disappearing source.** Soft-deleting the WHOOP RHR rows and + * recomputing drops the stale WHOOP rollup row (delete-then-insert), + * leaving the APPLE row alone — the reader now surfaces the Apple value. + * 5. **Idempotency.** Re-running the recompute on a dual-source day keeps + * the per-source row count at two (no duplicate-per-source rows). + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { + recomputeUserRollups, + readRollupBuckets, +} from "@/lib/rollups/measurement-rollups"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +// `recomputeBucketsForMeasurement` enqueues WEEK/MONTH/YEAR jobs via pg-boss; +// the DAY-only `recomputeUserRollups` calls here never touch the queue, but the +// sibling rollups test mocks the boss instance and the integration suite runs +// with `isolate: false`, so we leave it detached (null) for parity. +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +// Deterministic UTC day used across every case. All seeded `measuredAt` +// values land inside this calendar day so each (type, source) folds to a +// single DAY bucket. +const DAY = "2026-05-20"; +const DAY_START_UTC = new Date(`${DAY}T00:00:00.000Z`); +const DAY_END_UTC = new Date(`${DAY}T23:59:59.999Z`); +// Read window [from, to) that brackets the seeded day with room to spare. +const FROM = new Date(`${DAY}T00:00:00.000Z`); +const TO = new Date("2026-05-21T00:00:00.000Z"); + +let seq = 0; +async function seedUser(prisma: ReturnType) { + seq += 1; + return prisma.user.create({ + data: { + username: `source-aware-rollup-${seq}`, + email: `source-aware-rollup-${seq}@example.test`, + role: "USER", + }, + }); +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("source-aware measurement rollups — integration (v1.11.1)", () => { + it("RESTING_HEART_RATE: keeps one raw row per source, collapses to the WHOOP ladder pick", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 51, + unit: "bpm", + source: "WHOOP", + measuredAt: new Date(`${DAY}T06:00:00.000Z`), + }, + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 54, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date(`${DAY}T07:00:00.000Z`), + }, + ], + }); + + await recomputeUserRollups(user.id, { + types: ["RESTING_HEART_RATE"], + granularities: ["DAY"], + }); + + // Raw table: one row PER source for the (type, day). + const rawRows = await prisma.measurementRollup.findMany({ + where: { + userId: user.id, + type: "RESTING_HEART_RATE", + granularity: "DAY", + bucketStart: { gte: DAY_START_UTC, lte: DAY_END_UTC }, + }, + orderBy: { source: "asc" }, + }); + expect(rawRows).toHaveLength(2); + expect(rawRows.map((r) => r.source).sort()).toEqual([ + "APPLE_HEALTH", + "WHOOP", + ]); + // Each per-source row carries that source's single reading. + const apple = rawRows.find((r) => r.source === "APPLE_HEALTH")!; + const whoop = rawRows.find((r) => r.source === "WHOOP")!; + expect(apple.count).toBe(1); + expect(apple.mean).toBeCloseTo(54, 5); + expect(whoop.count).toBe(1); + expect(whoop.mean).toBeCloseTo(51, 5); + + // Read path collapses to ONE bucket — WHOOP wins the restingHeartRate + // ladder (WHOOP > APPLE_HEALTH > WITHINGS). + const buckets = await readRollupBuckets( + user.id, + "RESTING_HEART_RATE", + "DAY", + FROM, + TO, + ); + expect(buckets).toHaveLength(1); + expect(buckets[0].mean).toBeCloseTo(51, 5); + expect(buckets[0].count).toBe(1); + }); + + it("ACTIVITY_STEPS: collapses to the Apple canonical source only — never a cross-source 7800 blend", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "ACTIVITY_STEPS", + value: 4000, + unit: "count", + source: "APPLE_HEALTH", + measuredAt: new Date(`${DAY}T08:00:00.000Z`), + }, + { + userId: user.id, + type: "ACTIVITY_STEPS", + value: 3800, + unit: "count", + source: "WITHINGS", + measuredAt: new Date(`${DAY}T09:00:00.000Z`), + }, + ], + }); + + await recomputeUserRollups(user.id, { + types: ["ACTIVITY_STEPS"], + granularities: ["DAY"], + }); + + const rawRows = await prisma.measurementRollup.findMany({ + where: { + userId: user.id, + type: "ACTIVITY_STEPS", + granularity: "DAY", + bucketStart: { gte: DAY_START_UTC, lte: DAY_END_UTC }, + }, + orderBy: { source: "asc" }, + }); + expect(rawRows).toHaveLength(2); + const apple = rawRows.find((r) => r.source === "APPLE_HEALTH")!; + const withings = rawRows.find((r) => r.source === "WITHINGS")!; + expect(apple.mean).toBeCloseTo(4000, 5); + expect(apple.sumValue).toBeCloseTo(4000, 5); + expect(withings.mean).toBeCloseTo(3800, 5); + + // Steps ladder = APPLE_HEALTH > WITHINGS > MANUAL. The collapsed bucket + // reflects ONLY the Apple row — a single canonical source's daily total, + // not the 7800 cross-source sum that a source-blind aggregate produced. + const buckets = await readRollupBuckets( + user.id, + "ACTIVITY_STEPS", + "DAY", + FROM, + TO, + ); + expect(buckets).toHaveLength(1); + expect(buckets[0].count).toBe(1); // one Apple reading, not two summed + expect(buckets[0].mean).toBeCloseTo(4000, 5); + expect(buckets[0].mean).not.toBeCloseTo(7800, 5); + expect(buckets[0].mean).not.toBeCloseTo(3900, 5); // not a cross-source avg + }); + + it("RECOVERY_SCORE: native WHOOP wins over the COMPUTED proxy", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "RECOVERY_SCORE", + value: 70, + unit: "score", + source: "WHOOP", + measuredAt: new Date(`${DAY}T05:00:00.000Z`), + }, + { + userId: user.id, + type: "RECOVERY_SCORE", + value: 62, + unit: "score", + source: "COMPUTED", + measuredAt: new Date(`${DAY}T05:30:00.000Z`), + }, + ], + }); + + await recomputeUserRollups(user.id, { + types: ["RECOVERY_SCORE"], + granularities: ["DAY"], + }); + + const rawRows = await prisma.measurementRollup.findMany({ + where: { + userId: user.id, + type: "RECOVERY_SCORE", + granularity: "DAY", + bucketStart: { gte: DAY_START_UTC, lte: DAY_END_UTC }, + }, + }); + expect(rawRows).toHaveLength(2); + + // recovery ladder = WHOOP > COMPUTED — the device-native score wins. + const buckets = await readRollupBuckets( + user.id, + "RECOVERY_SCORE", + "DAY", + FROM, + TO, + ); + expect(buckets).toHaveLength(1); + expect(buckets[0].mean).toBeCloseTo(70, 5); + }); + + it("disappearing source: soft-deleting WHOOP rows drops the stale WHOOP rollup row on recompute", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 51, + unit: "bpm", + source: "WHOOP", + measuredAt: new Date(`${DAY}T06:00:00.000Z`), + }, + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 54, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date(`${DAY}T07:00:00.000Z`), + }, + ], + }); + + await recomputeUserRollups(user.id, { + types: ["RESTING_HEART_RATE"], + granularities: ["DAY"], + }); + + // Both sources present after the first fold. + const firstFold = await prisma.measurementRollup.findMany({ + where: { + userId: user.id, + type: "RESTING_HEART_RATE", + granularity: "DAY", + bucketStart: { gte: DAY_START_UTC, lte: DAY_END_UTC }, + }, + }); + expect(firstFold).toHaveLength(2); + + // The WHOOP source disappears (user disconnected the strap / deleted the + // last reading from that device). Tombstone the WHOOP rows. + await prisma.measurement.updateMany({ + where: { userId: user.id, source: "WHOOP" }, + data: { deletedAt: new Date() }, + }); + + // Re-fold the same day. The populator delete-then-inserts the affected + // (type, bucket) partition across ALL sources, so the now-empty WHOOP + // bucket is dropped rather than stranded as a stale row a plain upsert + // would never revisit. + await recomputeUserRollups(user.id, { + types: ["RESTING_HEART_RATE"], + granularities: ["DAY"], + }); + + const secondFold = await prisma.measurementRollup.findMany({ + where: { + userId: user.id, + type: "RESTING_HEART_RATE", + granularity: "DAY", + bucketStart: { gte: DAY_START_UTC, lte: DAY_END_UTC }, + }, + }); + expect(secondFold).toHaveLength(1); + expect(secondFold[0].source).toBe("APPLE_HEALTH"); + + // The reader now surfaces the Apple value — WHOOP is gone from the ladder + // resolution because it no longer has a row. + const buckets = await readRollupBuckets( + user.id, + "RESTING_HEART_RATE", + "DAY", + FROM, + TO, + ); + expect(buckets).toHaveLength(1); + expect(buckets[0].mean).toBeCloseTo(54, 5); + }); + + it("idempotency: re-running the recompute keeps the per-source row count at two", async () => { + const prisma = getPrismaClient(); + const user = await seedUser(prisma); + + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 51, + unit: "bpm", + source: "WHOOP", + measuredAt: new Date(`${DAY}T06:00:00.000Z`), + }, + { + userId: user.id, + type: "RESTING_HEART_RATE", + value: 54, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date(`${DAY}T07:00:00.000Z`), + }, + ], + }); + + const countRows = () => + prisma.measurementRollup.count({ + where: { + userId: user.id, + type: "RESTING_HEART_RATE", + granularity: "DAY", + bucketStart: { gte: DAY_START_UTC, lte: DAY_END_UTC }, + }, + }); + + await recomputeUserRollups(user.id, { + types: ["RESTING_HEART_RATE"], + granularities: ["DAY"], + }); + expect(await countRows()).toBe(2); + + // Second pass over the identical fixture must not duplicate per-source + // rows — the delete-then-insert keys on (userId, type, granularity, + // bucketStart, source). + await recomputeUserRollups(user.id, { + types: ["RESTING_HEART_RATE"], + granularities: ["DAY"], + }); + expect(await countRows()).toBe(2); + + // And the collapsed read is still a single WHOOP-canonical bucket. + const buckets = await readRollupBuckets( + user.id, + "RESTING_HEART_RATE", + "DAY", + FROM, + TO, + ); + expect(buckets).toHaveLength(1); + expect(buckets[0].mean).toBeCloseTo(51, 5); + }); +});