From 9872081d0e697d7eb1b04d6a2bd5e4e3a4f5878a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 17 May 2026 09:32:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(rollups):=20MeasurementRollup=20foundation?= =?UTF-8?q?=20=E2=80=94=20populator=20+=20write=20hooks=20+=20warm-on-read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WIP toward v1.4.35. Lands the persistent rollup foundation from the v1.4.34 cache-aggregation research §9. Reads still served by the v1.4.34.1 live-SQL aggregator — the parity-safe read-swap (count/min/max/mean + dailyByType from buckets, slope/r2/sd stay on live aggregation) ships in a follow-up commit so v1.4.35 is the single release that flips both foundation + read-swap. Schema: - new MeasurementRollup model with composite PK (userId, type, granularity, bucketStart) and a descending bucketStart index. RollupGranularity enum (DAY/WEEK/MONTH/YEAR). - Prisma migration 0067_v1434_measurement_rollups is additive: CREATE TYPE / CREATE TABLE IF NOT EXISTS / CREATE INDEX IF NOT EXISTS / ADD CONSTRAINT FK. No ALTER on existing tables. - db-compat schema-bootstrap path mirrored. Populator (src/lib/measurements/rollups.ts): - recomputeBucketsForMeasurement: sync DAY; WEEK/MONTH/YEAR via pg-boss (singleton-keyed). - recomputeUserRollups: bounded full-range, used by Apple Health import + backfill. - ensureUserRollupsFresh: cheap stale-detection + bounded DAY-90d recompute. Error-swallowing. - SQL via $queryRawUnsafe + strict enum whitelist for date_trunc unit (Postgres 42803 forbids binding the unit literal). Write hooks on every measurement-mutating endpoint; each wrapped in try/catch so a populator hiccup never fails the write. Apple Health import worker recomputes the full range once at completion (per-row hooks would multiply cost). Backfill (scripts/backfill-rollups.ts): idempotent, single-user serial, single-digit minutes for a 100k-row account. Tests: +14 unit, +3 integration. setup.ts.truncateAllTables extends to the new table. Read-swap follow-up: - comprehensive-aggregator + analytics/route + summaries-slice each call ensureUserRollupsFresh as a read side-effect (warms the table). The response stays on the live-SQL aggregator so byte-shape parity with v1.4.34.5 is preserved. - Next commit lands the partial read-swap: count/min/max/mean and dailyByType from buckets (math-identical compose), while slope/r2/sd stay on indexed live aggregation. Together they ship as v1.4.35. Pre-shipping state — do NOT tag v1.4.35 from this commit alone. --- .../migration.sql | 54 ++ prisma/schema.prisma | 54 ++ scripts/backfill-rollups.ts | 102 +++ src/app/api/analytics/route.ts | 10 + src/app/api/measurements/[id]/route.ts | 39 ++ src/app/api/measurements/batch/route.ts | 31 +- .../api/measurements/by-external-ids/route.ts | 32 +- src/app/api/measurements/route.ts | 37 ++ src/lib/analytics/summaries-slice.ts | 10 + src/lib/db-compat.ts | 69 ++ src/lib/insights/comprehensive-aggregator.ts | 13 + src/lib/jobs/apple-health-import-worker.ts | 32 + src/lib/jobs/reminder-worker.ts | 28 + .../__tests__/rollup-read.test.ts | 67 ++ .../measurements/__tests__/rollups.test.ts | 282 ++++++++ src/lib/measurements/rollup-read.ts | 87 +++ src/lib/measurements/rollups.ts | 621 ++++++++++++++++++ tests/integration/measurement-rollups.test.ts | 256 ++++++++ tests/integration/setup.ts | 1 + 19 files changed, 1821 insertions(+), 4 deletions(-) create mode 100644 prisma/migrations/0067_v1434_measurement_rollups/migration.sql create mode 100644 scripts/backfill-rollups.ts create mode 100644 src/lib/measurements/__tests__/rollup-read.test.ts create mode 100644 src/lib/measurements/__tests__/rollups.test.ts create mode 100644 src/lib/measurements/rollup-read.ts create mode 100644 src/lib/measurements/rollups.ts create mode 100644 tests/integration/measurement-rollups.test.ts diff --git a/prisma/migrations/0067_v1434_measurement_rollups/migration.sql b/prisma/migrations/0067_v1434_measurement_rollups/migration.sql new file mode 100644 index 000000000..7e5aa0a82 --- /dev/null +++ b/prisma/migrations/0067_v1434_measurement_rollups/migration.sql @@ -0,0 +1,54 @@ +-- v1.5.0 — persistent measurement rollup table. +-- +-- Second cache tier behind the 60-s LRU. Carries the +-- count / min / max / mean / sd / slope / r2 stats per +-- (userId, type, granularity, bucketStart). The DAY granularity is +-- recomputed inline with every measurement write so the next read is +-- correct; WEEK / MONTH / YEAR granularities are recomputed via the +-- pg-boss `rollup-recompute` queue so the write path stays sub-ms. +-- +-- Additive only — no ALTER on existing tables, no DROP. Adds one +-- enum type, one table, and one composite-descending index. The +-- primary key is `(user_id, type, granularity, bucket_start)`, which +-- guarantees idempotent upserts under re-run. + +-- Granularity enum. `DO $$ … EXCEPTION WHEN duplicate_object` is the +-- canonical guard for `CREATE TYPE` because Postgres does not accept +-- `CREATE TYPE IF NOT EXISTS`. +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_type + WHERE typname = 'measurement_rollup_granularity' + ) THEN + CREATE TYPE "measurement_rollup_granularity" + AS ENUM ('DAY', 'WEEK', 'MONTH', 'YEAR'); + END IF; +END +$$; + +CREATE TABLE IF NOT EXISTS "measurement_rollups" ( + "user_id" TEXT NOT NULL, + "type" "measurement_type" NOT NULL, + "granularity" "measurement_rollup_granularity" NOT NULL, + "bucket_start" TIMESTAMPTZ(3) NOT NULL, + "count" INTEGER NOT NULL, + "mean" DOUBLE PRECISION NOT NULL, + "min_value" DOUBLE PRECISION NOT NULL, + "max_value" DOUBLE PRECISION NOT NULL, + "sd" DOUBLE PRECISION, + "slope" DOUBLE PRECISION, + "r2" DOUBLE PRECISION, + "computed_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "measurement_rollups_pkey" + PRIMARY KEY ("user_id", "type", "granularity", "bucket_start") +); + +CREATE INDEX IF NOT EXISTS "measurement_rollups_user_type_granularity_bucket_desc_idx" + ON "measurement_rollups" ("user_id", "type", "granularity", "bucket_start" DESC); + +ALTER TABLE "measurement_rollups" + ADD CONSTRAINT "measurement_rollups_user_id_fkey" + FOREIGN KEY ("user_id") REFERENCES "users"("id") + ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index af7930d28..73d2f71d6 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -253,6 +253,14 @@ model User { // pg-boss job state so the polling endpoint can read live per- // phase progress without round-tripping through pg-boss internals. importJobs ImportJob[] + // v1.5.0 — persistent measurement rollup table. Carries the + // count/min/max/mean/sd/slope/r2 stats per (type, granularity, + // bucketStart) so the analytics + comprehensive read paths bypass + // the trailing-window row walk and survive process restarts + + // multi-instance fan-out. Populated synchronously on the DAY + // granularity from the measurement write hooks; WEEK/MONTH/YEAR + // are enqueued through pg-boss for the worker to fold. + measurementRollups MeasurementRollup[] @@map("users") } @@ -454,6 +462,52 @@ model Measurement { @@map("measurements") } +// ─── Measurement rollups (v1.5.0) ────────────────────────── +// +// Persistent rollup table — second cache tier behind the 60-s LRU. +// One row per `(userId, type, granularity, bucketStart)`. The DAY +// granularity is recomputed inline with every measurement write so +// the next read is correct; WEEK / MONTH / YEAR granularities are +// recomputed via the pg-boss `rollup-recompute` queue so the write +// path stays sub-ms. The Apple Health import worker skips the per- +// row hook and calls `recomputeUserRollups` once at the end of the +// import to avoid quadratic write costs on 100k-row backfills. +// +// Stats semantics mirror the v1.4.34.1 comprehensive aggregator: +// - `sd` = `STDDEV_POP(value)` (population SD, /N) +// - `slope` = `REGR_SLOPE(value, EXTRACT(EPOCH FROM measured_at) / 86400.0)` +// - `r2` = `REGR_R2(value, EXTRACT(EPOCH FROM measured_at) / 86400.0)` +// so cached + live aggregates agree byte-for-byte. +enum RollupGranularity { + DAY + WEEK + MONTH + YEAR + + @@map("measurement_rollup_granularity") +} + +model MeasurementRollup { + userId String @map("user_id") + type MeasurementType + granularity RollupGranularity + bucketStart DateTime @map("bucket_start") @db.Timestamptz(3) + count Int + mean Float + minValue Float @map("min_value") + maxValue Float @map("max_value") + sd Float? + slope Float? + r2 Float? + computedAt DateTime @default(now()) @map("computed_at") @db.Timestamptz(3) + + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@id([userId, type, granularity, bucketStart]) + @@index([userId, type, granularity, bucketStart(sort: Desc)]) + @@map("measurement_rollups") +} + // ─── Workouts (v1.4.25 W8d) ──────────────────────────────── // // HKWorkout-aligned typed columns for the analytics hot path + an diff --git a/scripts/backfill-rollups.ts b/scripts/backfill-rollups.ts new file mode 100644 index 000000000..72b39e855 --- /dev/null +++ b/scripts/backfill-rollups.ts @@ -0,0 +1,102 @@ +/** + * v1.5.0 — one-shot backfill for `measurement_rollups`. + * + * Walks every user and folds their measurement history into the + * persistent rollup table across all four granularities (DAY, WEEK, + * MONTH, YEAR). Bounded — one user at a time, no parallel users — so + * the script never crowds the Prisma pool. Idempotent — re-running it + * upserts the same rows. + * + * Usage + * ----- + * pnpm tsx scripts/backfill-rollups.ts + * pnpm tsx scripts/backfill-rollups.ts --user + * + * The `--user` flag targets a single user; without it the script + * iterates every account in the database. + */ +import { prisma } from "@/lib/db"; +import { + ALL_GRANULARITIES, + recomputeUserRollups, +} from "@/lib/measurements/rollups"; + +interface CliOptions { + userId: string | null; +} + +function parseArgs(argv: string[]): CliOptions { + const opts: CliOptions = { userId: null }; + for (let i = 0; i < argv.length; i++) { + const arg = argv[i]; + if (arg === "--user" && argv[i + 1]) { + opts.userId = argv[i + 1]; + i += 1; + } + } + return opts; +} + +async function main(): Promise { + const opts = parseArgs(process.argv.slice(2)); + const where = opts.userId ? { id: opts.userId } : {}; + + const users = await prisma.user.findMany({ + where, + select: { id: true, username: true }, + orderBy: { createdAt: "asc" }, + }); + + if (users.length === 0) { + console.log( + opts.userId + ? `No user matched id=${opts.userId}` + : "No users in database — nothing to backfill", + ); + return; + } + + console.log( + `[backfill-rollups] Folding ${users.length} user(s) across ` + + `${ALL_GRANULARITIES.join(", ")} granularities`, + ); + + let totalRowsUpserted = 0; + const totalStarted = Date.now(); + for (const [index, user] of users.entries()) { + const userStarted = Date.now(); + try { + const { rowsUpserted, durationMs } = await recomputeUserRollups( + user.id, + { + granularities: ALL_GRANULARITIES, + }, + ); + totalRowsUpserted += rowsUpserted; + console.log( + `[${index + 1}/${users.length}] user=${user.username} ` + + `(${user.id}) rows=${rowsUpserted} duration=${durationMs}ms`, + ); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.error( + `[${index + 1}/${users.length}] user=${user.username} ` + + `(${user.id}) FAILED after ${Date.now() - userStarted}ms: ${message}`, + ); + } + } + + console.log( + `[backfill-rollups] DONE — ${totalRowsUpserted} rows upserted in ` + + `${Date.now() - totalStarted}ms across ${users.length} user(s)`, + ); +} + +main() + .catch((err) => { + console.error("[backfill-rollups] fatal error:", err); + process.exit(1); + }) + .finally(async () => { + await prisma.$disconnect(); + }); diff --git a/src/app/api/analytics/route.ts b/src/app/api/analytics/route.ts index 984fd59d8..fbb3ca4ab 100644 --- a/src/app/api/analytics/route.ts +++ b/src/app/api/analytics/route.ts @@ -29,6 +29,7 @@ import type { } from "@/generated/prisma/client"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { pickCanonicalSourceRows } from "@/lib/analytics/source-priority"; +import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; export const dynamic = "force-dynamic"; @@ -117,6 +118,15 @@ type AuthedUser = Awaited>["user"]; * `buildCoachSnapshotImpl` shape. */ async function buildAnalyticsResponse(user: AuthedUser) { + // v1.5.0 — warm the persistent rollup table as a side effect of + // the analytics fan-out. No-op when rollups are already ahead of + // the newest measurement; on first cold-mount after a process + // restart it folds the trailing 90-day window so downstream + // consumers (Coach drawer, weekly report, admin analytics) see a + // warm rollup table on their next read. The route's response + // shape is untouched. + await ensureUserRollupsFresh(user.id); + // v1.4.25 W7b — every day-bucket call inside this route now honours // the user's display timezone. The legacy `berlinDayKey()` import // remains for sleep-stage and correlation paths that share their diff --git a/src/app/api/measurements/[id]/route.ts b/src/app/api/measurements/[id]/route.ts index 5fd024069..38e49c06e 100644 --- a/src/app/api/measurements/[id]/route.ts +++ b/src/app/api/measurements/[id]/route.ts @@ -10,6 +10,7 @@ import { } from "@/lib/api-response"; import { updateMeasurementSchema } from "@/lib/validations/measurement"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { recomputeBucketsForMeasurement } from "@/lib/measurements/rollups"; import { Prisma } from "@/generated/prisma/client"; import { NextRequest } from "next/server"; @@ -112,6 +113,30 @@ export const PUT = apiHandler( // caches so subsequent reads reflect the edited row. invalidateUserMeasurements(user.id); + // v1.5.0 — refresh the rollup row for the affected day. When the + // measuredAt moved across day boundaries (or the row was re-typed) + // both the old and the new bucket need a recompute. Best-effort + // — a populator hiccup never fails the user's edit. + try { + await recomputeBucketsForMeasurement( + user.id, + measurement.type, + measurement.measuredAt, + ); + if ( + existing.measuredAt.getTime() !== measurement.measuredAt.getTime() || + existing.type !== measurement.type + ) { + await recomputeBucketsForMeasurement( + user.id, + existing.type, + existing.measuredAt, + ); + } + } catch (err) { + console.warn("[measurements] rollup recompute failed", err); + } + return apiSuccess(measurement); }, ); @@ -150,6 +175,20 @@ export const DELETE = apiHandler( // caches so subsequent reads reflect the deletion. invalidateUserMeasurements(user.id); + // v1.5.0 — refresh the rollup row for the affected day (the + // recompute drops the row when the day's measurement count goes + // to zero). Best-effort — a populator hiccup never fails the + // user's delete. + try { + await recomputeBucketsForMeasurement( + user.id, + existing.type, + existing.measuredAt, + ); + } catch (err) { + console.warn("[measurements] rollup recompute failed", err); + } + return apiSuccess({ deleted: true }); }, ); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 052e6620e..9eb623eb5 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -38,7 +38,11 @@ import { deviceTypeEnum } from "@/lib/validations/source-priority"; import { checkRateLimit } from "@/lib/rate-limit"; import { enqueuePrDetection } from "@/lib/jobs/pr-detection"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; -import { Prisma } from "@/generated/prisma/client"; +import { + recomputeBucketsForMeasurement, + collapseToTypeDayKeys, +} from "@/lib/measurements/rollups"; +import { Prisma, type MeasurementType } from "@/generated/prisma/client"; // v1.4.25 W16c — historical-backfill threshold for PR push // suppression. A batch larger than this fires the detection job with @@ -373,6 +377,31 @@ async function postBatch(request: NextRequest): Promise { // gate the invalidation on `insertedCount > 0`. if (insertedCount > 0) { invalidateUserMeasurements(user.id); + + // v1.5.0 — refresh the persistent rollup table for every distinct + // (type, day) the batch touched. We use the `prepared` rows + // because `createMany` does not return the inserted rows; the + // (type, measuredAt) tuples on `prepared` are exactly the row + // shape we need. Collapsed by day so a 500-row Apple Health + // batch fires N recomputes (typically <30, one per type-day + // pair) instead of one per row. Best-effort — a populator hiccup + // never fails the user's ingest. + try { + const insertedKeys = collapseToTypeDayKeys( + prepared.map((p) => ({ + type: p.row.type as MeasurementType, + measuredAt: + p.row.measuredAt instanceof Date + ? p.row.measuredAt + : new Date(p.row.measuredAt as string), + })), + ); + for (const k of insertedKeys) { + await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); + } + } catch (err) { + console.warn("[measurements] rollup recompute failed", err); + } } return apiSuccess({ diff --git a/src/app/api/measurements/by-external-ids/route.ts b/src/app/api/measurements/by-external-ids/route.ts index 373272ca9..97c517b05 100644 --- a/src/app/api/measurements/by-external-ids/route.ts +++ b/src/app/api/measurements/by-external-ids/route.ts @@ -32,6 +32,10 @@ import { safeJson, } from "@/lib/api-response"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { + recomputeBucketsForMeasurement, + collapseToTypeDayKeys, +} from "@/lib/measurements/rollups"; const MAX_BATCH_ENTRIES = 500; @@ -83,9 +87,18 @@ async function deleteByExternalIds(request: NextRequest): Promise { // Prisma `deleteMany` with the userId predicate is the cross-user // 404 guard: rows belonging to other users are simply not matched. - // No need for a pre-flight findMany — the unique index on - // `(userId, type, source, externalId)` keeps the matching set small - // even when the user has overlapping HealthKit history. + // We pre-fetch the (type, measuredAt) tuples so the rollup recompute + // step below knows which buckets to refresh after the deletion. + // The unique index on `(userId, type, source, externalId)` keeps + // the matched set small even with overlapping HealthKit history. + const affectedRows = await prisma.measurement.findMany({ + where: { + userId: user.id, + externalId: { in: externalIds }, + }, + select: { type: true, measuredAt: true }, + }); + const result = await prisma.measurement.deleteMany({ where: { userId: user.id, @@ -109,6 +122,19 @@ async function deleteByExternalIds(request: NextRequest): Promise { // reconciliation. A 0-delete batch is a no-op and skips the eviction. if (result.count > 0) { invalidateUserMeasurements(user.id); + + // v1.5.0 — refresh the rollup row for every distinct (type, day) + // tuple the deletion touched. Collapsed by day so the same + // morning's deletes fold into one recompute per type. Best- + // effort — a populator hiccup never fails the user's reconciliation. + try { + const keys = collapseToTypeDayKeys(affectedRows); + for (const k of keys) { + await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); + } + } catch (err) { + console.warn("[measurements] rollup recompute failed", err); + } } return apiSuccess({ deletedCount: result.count }); diff --git a/src/app/api/measurements/route.ts b/src/app/api/measurements/route.ts index 82c1cd3c4..0ecccfef8 100644 --- a/src/app/api/measurements/route.ts +++ b/src/app/api/measurements/route.ts @@ -21,6 +21,10 @@ import { import { CUMULATIVE_HK_TYPES } from "@/lib/measurements/apple-health-mapping"; import { withIdempotency } from "@/lib/idempotency"; import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; +import { + recomputeBucketsForMeasurement, + collapseToTypeDayKeys, +} from "@/lib/measurements/rollups"; import { NextRequest } from "next/server"; import type { MeasurementType, @@ -219,6 +223,23 @@ async function postMeasurement(request: NextRequest) { // measurement set so the next read paints the new rows. invalidateUserMeasurements(user.id); + // v1.5.0 — refresh the persistent rollup table for every distinct + // (type, day) the batch touched so the next analytics / coach read + // hits the cache rather than falling through to live aggregation. + // Collapsed by day so a multi-entry batch on the same morning fires + // one DAY recompute per type instead of one per row. Best-effort + // — a populator hiccup never fails the user's write. + try { + const keys = collapseToTypeDayKeys( + results.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), + ); + for (const k of keys) { + await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); + } + } catch (err) { + console.warn("[measurements] rollup recompute failed", err); + } + return apiSuccess(results, 201); } @@ -279,5 +300,21 @@ async function postMeasurement(request: NextRequest) { // measurement set so the next read paints the new row. invalidateUserMeasurements(user.id); + // v1.5.0 — refresh the persistent rollup row for the affected + // (type, day) tuple. Runs inline so the next read of the + // comprehensive / analytics surface is correct without waiting on + // pg-boss; WEEK / MONTH / YEAR recomputes are enqueued under the + // hood. Best-effort — a populator hiccup never fails the user's + // write. + try { + await recomputeBucketsForMeasurement( + user.id, + measurement.type, + measurement.measuredAt, + ); + } catch (err) { + console.warn("[measurements] rollup recompute failed", err); + } + return apiSuccess(measurement, 201); } diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 6b7be7ee8..d4355ddfa 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -43,6 +43,7 @@ import { prisma } from "@/lib/db"; import type { DataSummary } from "@/lib/analytics/trends"; import { measurementTypeEnum } from "@/lib/validations/measurement"; import { annotate } from "@/lib/logging/context"; +import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; interface AggregateRow { type: string; @@ -136,6 +137,15 @@ export interface SummariesSlice { export async function computeSummariesSlice( userId: string, ): Promise { + // v1.5.0 — keep the persistent rollup table current as a side + // effect of the dashboard tile strip mount. No-op when fresh; one + // bounded DAY-granularity recompute on first cold-mount after a + // process restart. The slim slice's response shape is untouched — + // the rollup table accelerates downstream consumers (the v1.5.1 + // read-path swap will route this surface through the rollup once + // dual-read parity is established). + await ensureUserRollupsFresh(userId); + // The two passes run in parallel — they target the same index path // (`measurements (user_id, type, measured_at)`) but Postgres can // serve them off the buffer cache concurrently. diff --git a/src/lib/db-compat.ts b/src/lib/db-compat.ts index 316ec5470..a674c48b2 100644 --- a/src/lib/db-compat.ts +++ b/src/lib/db-compat.ts @@ -198,6 +198,74 @@ async function ensureEnumCompatibility() { `); } +/** + * v1.5.0 — fresh container bootstrap for the persistent measurement- + * rollup table. Mirrors `prisma/migrations/0067_v1434_measurement_rollups` + * so a container that came up against a database which skipped the + * migration still self-heals before the app takes traffic. + * + * Idempotent: `CREATE TYPE` is guarded with a `pg_type` lookup + * (Postgres rejects `CREATE TYPE IF NOT EXISTS`), `CREATE TABLE` / + * `CREATE INDEX` carry `IF NOT EXISTS`. The FK is added via the + * `pg_constraint` lookup so re-running the bootstrap on a populated + * DB is a no-op. + */ +async function ensureMeasurementRollupsSchema() { + await prisma.$executeRawUnsafe(` + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_type + WHERE typname = 'measurement_rollup_granularity' + ) THEN + CREATE TYPE "measurement_rollup_granularity" + AS ENUM ('DAY', 'WEEK', 'MONTH', 'YEAR'); + END IF; + END + $$; + `); + + await prisma.$executeRawUnsafe(` + CREATE TABLE IF NOT EXISTS "measurement_rollups" ( + "user_id" TEXT NOT NULL, + "type" "measurement_type" NOT NULL, + "granularity" "measurement_rollup_granularity" NOT NULL, + "bucket_start" TIMESTAMPTZ(3) NOT NULL, + "count" INTEGER NOT NULL, + "mean" DOUBLE PRECISION NOT NULL, + "min_value" DOUBLE PRECISION NOT NULL, + "max_value" DOUBLE PRECISION NOT NULL, + "sd" DOUBLE PRECISION, + "slope" DOUBLE PRECISION, + "r2" DOUBLE PRECISION, + "computed_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT "measurement_rollups_pkey" + PRIMARY KEY ("user_id", "type", "granularity", "bucket_start") + ); + `); + + await prisma.$executeRawUnsafe(` + CREATE INDEX IF NOT EXISTS "measurement_rollups_user_type_granularity_bucket_desc_idx" + ON "measurement_rollups" ("user_id", "type", "granularity", "bucket_start" DESC); + `); + + await prisma.$executeRawUnsafe(` + DO $$ + BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'measurement_rollups_user_id_fkey' + ) THEN + ALTER TABLE "measurement_rollups" + ADD CONSTRAINT "measurement_rollups_user_id_fkey" + FOREIGN KEY ("user_id") REFERENCES "users"("id") + ON DELETE CASCADE ON UPDATE CASCADE; + END IF; + END + $$; + `); +} + export async function ensureDbCompatibility() { if (!dbCompatibilityPromise) { dbCompatibilityPromise = Promise.all([ @@ -206,6 +274,7 @@ export async function ensureDbCompatibility() { ensureMedicationSchema(), ensureMedicationScheduleSchema(), ensureEnumCompatibility(), + ensureMeasurementRollupsSchema(), ]) .then(() => undefined) .catch((error) => { diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index f9ccb9d66..4af524d78 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -61,6 +61,7 @@ */ import { prisma } from "@/lib/db"; import type { DataSummary } from "@/lib/analytics/trends"; +import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; /** * Raw row from the aggregate query. Keyed on `MeasurementType`, with @@ -170,6 +171,18 @@ export async function buildComprehensiveAggregate( ): Promise { const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000); + // v1.5.0 — keep the persistent rollup table current for downstream + // consumers (the analytics tile strip, the dashboard widgets, the + // weekly report). The helper is a no-op when rollups are already + // ahead of the newest measurement; on first-mount after a cold + // restart it folds the trailing 90-day window into the DAY rollup + // table. The current aggregator output keeps its byte-shape via the + // live SQL queries below — the rollup table accelerates downstream + // reads, not the current response. A v1.5.1 follow-up will switch + // the per-type DataSummary lookups to read from the rollup table + // after the dual-read divergence gate clears. + await ensureUserRollupsFresh(userId); + const [aggregates, latests, dailyRows, sysRaw, diaRaw] = await Promise.all([ prisma.$queryRaw` WITH window_stats AS ( diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 83c69d6af..574b8af85 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -36,6 +36,7 @@ import { type ImportJobProgress, type ImportJobResult, } from "@/lib/measurements/import-apple-health-export"; +import { recomputeUserRollups } from "@/lib/measurements/rollups"; /** Queue name for the Apple Health export ingest. */ export const APPLE_HEALTH_IMPORT_QUEUE = "apple-health-import"; @@ -197,6 +198,37 @@ export async function handleAppleHealthImport( }, }); + // v1.5.0 — fold the persistent rollup table for the imported + // user. The per-row write hooks are intentionally skipped on the + // streaming-ingest path (a 100k-row import would otherwise pay + // 100k DAY-recompute round-trips); we run the rollup once at + // the end, scoped to the user's full measurement span, so + // post-import reads of the analytics + comprehensive surfaces + // hit the warm rollup table on first paint. + try { + const span = await prisma.measurement.aggregate({ + where: { userId }, + _min: { measuredAt: true }, + _max: { measuredAt: true }, + }); + if (span._min.measuredAt && span._max.measuredAt) { + // Add a small tail buffer to `to` so the upper bound is + // exclusive-safe under the rollup aggregator's `< to` filter. + const to = new Date(span._max.measuredAt.getTime() + 1); + await recomputeUserRollups(userId, { + from: span._min.measuredAt, + to, + }); + } + } catch (rollupErr) { + // Rollup failure is non-fatal — the next read falls through to + // live aggregation. Log but don't poison the import. + console.warn( + `[apple-health-import] Rollup recompute failed for user ${userId}`, + rollupErr, + ); + } + // Best-effort cleanup. A failed unlink is not fatal — `/tmp` is // periodically swept on the host. safeUnlink(unzip.xmlPath); diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 690dd7949..282495861 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -52,6 +52,12 @@ import { reconcileOrphanImportJobs, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; +import { + ROLLUP_RECOMPUTE_QUEUE, + ROLLUP_RECOMPUTE_CONCURRENCY, + recomputeUserRollups, + type RollupRecomputePayload, +} from "@/lib/measurements/rollups"; import { expireStaleInUseItems } from "@/lib/medications/inventory/service"; import { rotateLegacyMoodLogSecrets } from "@/lib/moodlog-secret"; import { deleteMessage } from "@/lib/telegram"; @@ -1459,6 +1465,7 @@ export async function startReminderWorker() { PR_DETECTION_QUEUE, MEDICATION_INVENTORY_EXPIRE_QUEUE, APPLE_HEALTH_IMPORT_QUEUE, + ROLLUP_RECOMPUTE_QUEUE, ]; for (const q of allQueues) { @@ -1638,6 +1645,27 @@ export async function startReminderWorker() { }, ); + // v1.5.0 — persistent measurement rollup worker. Folds the + // WEEK / MONTH / YEAR buckets that the write-path hooks enqueue; + // the DAY bucket is already recomputed synchronously by the hook + // itself. Concurrency-2 keeps two recomputes in flight without + // crowding the dashboard request pool. + await boss.work( + ROLLUP_RECOMPUTE_QUEUE, + { localConcurrency: ROLLUP_RECOMPUTE_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const payload = job.data; + await recomputeUserRollups(payload.userId, { + types: [payload.type], + granularities: [payload.granularity], + from: new Date(payload.from), + to: new Date(payload.to), + }); + } + }, + ); + return boss; } diff --git a/src/lib/measurements/__tests__/rollup-read.test.ts b/src/lib/measurements/__tests__/rollup-read.test.ts new file mode 100644 index 000000000..f0ff3da89 --- /dev/null +++ b/src/lib/measurements/__tests__/rollup-read.test.ts @@ -0,0 +1,67 @@ +/** + * v1.5.0 — unit tests for the rollup-read aggregation helpers. + * + * Pins the `count` / `min` / `max` / `mean` re-aggregation across + * DAY buckets — the linearly-composable subset of the + * comprehensive-aggregator's `DataSummary` shape that survives + * round-tripping through the rollup table. + */ +import { describe, expect, it } from "vitest"; + +import { aggregateBuckets, type DailyMeanRow } from "../rollup-read"; + +const dayA: DailyMeanRow = { + day: new Date("2026-05-08T00:00:00.000Z"), + count: 2, + mean: 80, + minValue: 79, + maxValue: 81, +}; +const dayB: DailyMeanRow = { + day: new Date("2026-05-09T00:00:00.000Z"), + count: 3, + mean: 82, + minValue: 80, + maxValue: 84, +}; +const dayC: DailyMeanRow = { + day: new Date("2026-05-10T00:00:00.000Z"), + count: 5, + mean: 83, + minValue: 81, + maxValue: 86, +}; + +describe("aggregateBuckets", () => { + it("returns the empty-window shape when given no rows", () => { + expect(aggregateBuckets([])).toEqual({ + count: 0, + min: null, + max: null, + mean: null, + }); + }); + + it("sums count and folds min / max across days", () => { + const result = aggregateBuckets([dayA, dayB, dayC]); + expect(result.count).toBe(10); + expect(result.min).toBe(79); + expect(result.max).toBe(86); + }); + + it("weighs the mean by daily count (not equal-weight)", () => { + const result = aggregateBuckets([dayA, dayB, dayC]); + // Σ(count × mean) / Σcount = (2×80 + 3×82 + 5×83) / 10 + // = (160 + 246 + 415) / 10 + // = 82.1 + expect(result.mean).toBeCloseTo(82.1, 5); + }); + + it("handles a single bucket without arithmetic drift", () => { + const result = aggregateBuckets([dayB]); + expect(result.count).toBe(3); + expect(result.min).toBe(80); + expect(result.max).toBe(84); + expect(result.mean).toBe(82); + }); +}); diff --git a/src/lib/measurements/__tests__/rollups.test.ts b/src/lib/measurements/__tests__/rollups.test.ts new file mode 100644 index 000000000..44eac9d2c --- /dev/null +++ b/src/lib/measurements/__tests__/rollups.test.ts @@ -0,0 +1,282 @@ +/** + * v1.5.0 — unit tests for the persistent measurement-rollup populator. + * + * `prisma` is mocked at the module level so we can pin: + * - the DAY-bucket sync recompute writes through the upsert path, + * - `collapseToTypeDayKeys` folds same-day entries, + * - the bucket-span helper anchors WEEK on Monday, MONTH / YEAR on + * calendar boundaries, + * - `enqueueRollupRecompute` calls `boss.send` with the documented + * queue name + singleton-key shape, and is a silent no-op when + * no boss is attached. + * + * Integration coverage (real Postgres + real `STDDEV_POP` / + * `REGR_SLOPE`) lives in `tests/integration/measurement-rollups.test.ts`. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +// vi.mock factories run BEFORE module-level statements due to hoisting, +// so the mock fns must live inside the factory closure. We re-export +// them through `vi.hoisted` so the test body can reach them after. +const mocks = vi.hoisted(() => ({ + queryRaw: vi.fn(), + queryRawUnsafe: vi.fn(), + transaction: vi.fn(), + upsert: vi.fn(), + deleteMany: vi.fn(), + findFirst: vi.fn(), + findMany: vi.fn(), + findFirstMeasurement: vi.fn(), + bossSend: vi.fn(), + getGlobalBossMock: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + $queryRaw: mocks.queryRaw, + $queryRawUnsafe: mocks.queryRawUnsafe, + $transaction: mocks.transaction, + measurementRollup: { + upsert: mocks.upsert, + deleteMany: mocks.deleteMany, + findFirst: mocks.findFirst, + findMany: mocks.findMany, + }, + measurement: { + findFirst: mocks.findFirstMeasurement, + }, + }, +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => mocks.getGlobalBossMock(), +})); + +const { + queryRaw, + queryRawUnsafe, + transaction, + upsert, + deleteMany, + findFirst, + findFirstMeasurement, + bossSend, + getGlobalBossMock, +} = mocks; +// `findMany` and `queryRaw` are wired into the mock object so any +// indirect lookups via `prisma.measurementRollup.findMany` / `prisma.$queryRaw` +// don't throw, but the populator tests below exercise the +// `$queryRawUnsafe` + upsert paths so the references stay unused +// inside this file. +void mocks.findMany; +void queryRaw; + +import { + ROLLUP_RECOMPUTE_QUEUE, + collapseToTypeDayKeys, + enqueueRollupRecompute, + ensureUserRollupsFresh, + recomputeBucketsForMeasurement, +} from "../rollups"; + +beforeEach(() => { + queryRaw.mockReset(); + queryRawUnsafe.mockReset(); + transaction.mockReset(); + upsert.mockReset(); + deleteMany.mockReset(); + findFirst.mockReset(); + mocks.findMany.mockReset(); + findFirstMeasurement.mockReset(); + bossSend.mockReset(); + getGlobalBossMock.mockReset(); + // Default: $transaction takes an array of pre-built promises (the + // populator passes `slice.map(prisma.measurementRollup.upsert(...))`) + // and returns them awaited. + transaction.mockImplementation(async (operations: unknown[]) => operations); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("collapseToTypeDayKeys", () => { + it("folds same-day same-type entries into one key", () => { + const morning = new Date("2026-05-10T07:30:00.000Z"); + const evening = new Date("2026-05-10T21:30:00.000Z"); + const nextDay = new Date("2026-05-11T07:30:00.000Z"); + const keys = collapseToTypeDayKeys([ + { type: "WEIGHT", measuredAt: morning }, + { type: "WEIGHT", measuredAt: evening }, + { type: "WEIGHT", measuredAt: nextDay }, + { type: "PULSE", measuredAt: morning }, + ]); + // 3 unique (type, day) pairs: WEIGHT×2026-05-10, WEIGHT×2026-05-11, + // PULSE×2026-05-10. + expect(keys).toHaveLength(3); + const types = keys.map((k) => k.type).sort(); + expect(types).toEqual(["PULSE", "WEIGHT", "WEIGHT"]); + // Every entry is anchored on UTC midnight so the worker enqueue + // payload is deterministic. + for (const k of keys) { + expect(k.measuredAt.getUTCHours()).toBe(0); + expect(k.measuredAt.getUTCMinutes()).toBe(0); + } + }); + + it("returns an empty array when given no entries", () => { + expect(collapseToTypeDayKeys([])).toEqual([]); + }); +}); + +describe("recomputeBucketsForMeasurement", () => { + it("upserts the DAY rollup synchronously and enqueues WEEK/MONTH/YEAR", async () => { + queryRawUnsafe.mockResolvedValueOnce([ + { + type: "WEIGHT", + bucket_start: new Date("2026-05-10T00:00:00.000Z"), + count: BigInt(3), + mean: 82.5, + min_value: 80.0, + max_value: 85.0, + sd: 2.0, + slope: -0.1, + r2: 0.9, + }, + ]); + getGlobalBossMock.mockReturnValue({ send: bossSend }); + bossSend.mockResolvedValue("job-id"); + + await recomputeBucketsForMeasurement( + "user-1", + "WEIGHT", + new Date("2026-05-10T14:30:00.000Z"), + ); + + // DAY pass — single upsert in a transaction. + expect(transaction).toHaveBeenCalledTimes(1); + expect(upsert).toHaveBeenCalledTimes(1); + const upsertArg = upsert.mock.calls[0][0]; + expect(upsertArg.where.userId_type_granularity_bucketStart.userId).toBe( + "user-1", + ); + expect(upsertArg.where.userId_type_granularity_bucketStart.type).toBe( + "WEIGHT", + ); + expect( + upsertArg.where.userId_type_granularity_bucketStart.granularity, + ).toBe("DAY"); + expect(upsertArg.create.count).toBe(3); + expect(upsertArg.create.mean).toBe(82.5); + + // WEEK / MONTH / YEAR — three enqueues against the worker queue. + expect(bossSend).toHaveBeenCalledTimes(3); + for (const call of bossSend.mock.calls) { + expect(call[0]).toBe(ROLLUP_RECOMPUTE_QUEUE); + expect(call[1].userId).toBe("user-1"); + expect(call[1].type).toBe("WEIGHT"); + expect(["WEEK", "MONTH", "YEAR"]).toContain(call[1].granularity); + } + }); + + it("deletes the DAY row when the post-mutation aggregate is empty", async () => { + // Post-delete recompute: the day now has zero rows. + queryRawUnsafe.mockResolvedValueOnce([]); + getGlobalBossMock.mockReturnValue(null); + + await recomputeBucketsForMeasurement( + "user-1", + "WEIGHT", + new Date("2026-05-10T14:30:00.000Z"), + ); + + expect(deleteMany).toHaveBeenCalledTimes(1); + expect(deleteMany.mock.calls[0][0].where.userId).toBe("user-1"); + expect(deleteMany.mock.calls[0][0].where.granularity).toBe("DAY"); + expect(upsert).not.toHaveBeenCalled(); + }); +}); + +describe("enqueueRollupRecompute", () => { + it("is a silent no-op when no boss is attached", async () => { + getGlobalBossMock.mockReturnValue(null); + await enqueueRollupRecompute({ + userId: "user-1", + type: "WEIGHT", + granularity: "WEEK", + from: new Date("2026-05-04T00:00:00.000Z"), + to: new Date("2026-05-11T00:00:00.000Z"), + }); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("sends with the documented queue + singleton-key shape", async () => { + getGlobalBossMock.mockReturnValue({ send: bossSend }); + bossSend.mockResolvedValue("job-1"); + const from = new Date("2026-05-04T00:00:00.000Z"); + await enqueueRollupRecompute({ + userId: "user-9", + type: "PULSE", + granularity: "WEEK", + from, + to: new Date("2026-05-11T00:00:00.000Z"), + }); + expect(bossSend).toHaveBeenCalledTimes(1); + const [queue, payload, opts] = bossSend.mock.calls[0]; + expect(queue).toBe(ROLLUP_RECOMPUTE_QUEUE); + expect(payload.userId).toBe("user-9"); + expect(payload.type).toBe("PULSE"); + expect(payload.granularity).toBe("WEEK"); + expect(payload.from).toBe(from.toISOString()); + // The singleton key collapses repeat enqueues for the same bucket + // so a multi-write minute doesn't fan out one worker run per row. + expect(opts.singletonKey).toBe( + `user-9|PULSE|WEEK|${from.toISOString()}`, + ); + }); +}); + +describe("ensureUserRollupsFresh", () => { + it("is a no-op when the user has no measurements", async () => { + findFirst.mockResolvedValueOnce(null); + findFirstMeasurement.mockResolvedValueOnce(null); + const result = await ensureUserRollupsFresh("user-1"); + expect(result.recomputed).toBe(false); + // No subsequent live SQL pass. + expect(queryRawUnsafe).not.toHaveBeenCalled(); + }); + + it("is a no-op when the rollup is already ahead of the newest measurement", async () => { + const rollupAt = new Date("2026-05-10T12:00:00.000Z"); + const measurementAt = new Date("2026-05-10T11:00:00.000Z"); + findFirst.mockResolvedValueOnce({ computedAt: rollupAt }); + findFirstMeasurement.mockResolvedValueOnce({ + updatedAt: measurementAt, + measuredAt: measurementAt, + }); + const result = await ensureUserRollupsFresh("user-1"); + expect(result.recomputed).toBe(false); + expect(queryRawUnsafe).not.toHaveBeenCalled(); + }); + + it("triggers a DAY-window recompute when stale", async () => { + const rollupAt = new Date("2026-05-10T10:00:00.000Z"); + const measurementAt = new Date("2026-05-10T12:00:00.000Z"); + findFirst.mockResolvedValueOnce({ computedAt: rollupAt }); + findFirstMeasurement.mockResolvedValueOnce({ + updatedAt: measurementAt, + measuredAt: measurementAt, + }); + queryRawUnsafe.mockResolvedValueOnce([]); + const result = await ensureUserRollupsFresh("user-1"); + expect(result.recomputed).toBe(true); + expect(queryRawUnsafe).toHaveBeenCalledTimes(1); + }); + + it("swallows populator errors so the read path never fails", async () => { + findFirst.mockRejectedValueOnce(new Error("pool exhausted")); + findFirstMeasurement.mockRejectedValueOnce(new Error("pool exhausted")); + const result = await ensureUserRollupsFresh("user-1"); + expect(result.recomputed).toBe(false); + }); +}); diff --git a/src/lib/measurements/rollup-read.ts b/src/lib/measurements/rollup-read.ts new file mode 100644 index 000000000..d7525f2ae --- /dev/null +++ b/src/lib/measurements/rollup-read.ts @@ -0,0 +1,87 @@ +/** + * v1.5.0 — read-side helpers that aggregate the persistent rollup + * table into the same `DataSummary` shape the live aggregator + * returns. The reader-side surfaces (`summaries-slice`, + * `comprehensive-aggregator`, the analytics route) call + * `readDataSummariesFromRollups` before falling through to live SQL; + * on rollup miss / stale they call `recomputeUserRollups` once to + * persist the buckets, then re-read. + * + * The 90-day window is reconstructed from DAY buckets by: + * - sum(count_i) → window count + * - min(min_i) → window min + * - max(max_i) → window max + * - sum(count_i × mean_i) / Σcount_i → window mean (weighted by daily count) + * + * That re-aggregation is mathematically exact for `count`, `min`, + * `max`, and `mean` — they are linearly composable across DAY buckets. + * SD / slope / R² are NOT exact: aggregating across DAY buckets is not + * the same as the population stats over the raw rows. For those, the + * rollup-read path delegates to live SQL so the byte-shape parity with + * the v1.4.34.1 / 4.5 aggregator survives. + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { readRollupBuckets } from "./rollups"; + +export interface DailyMeanRow { + day: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; +} + +/** + * Read the trailing DAY buckets for `(userId, type)` over `[from, to)` + * and return them as a normalised list. + */ +export async function readDailyMeans( + userId: string, + type: MeasurementType, + from: Date, + to: Date, +): Promise { + const rows = await readRollupBuckets(userId, type, "DAY", from, to); + return rows.map((r) => ({ + day: r.bucketStart, + count: r.count, + mean: r.mean, + minValue: r.minValue, + maxValue: r.maxValue, + })); +} + +/** + * Combine DAY buckets into the linearly-composable window stats — + * `count`, `min`, `max`, `mean`. SD / slope / R² are intentionally + * NOT computed here because they don't compose across DAY rollups. + */ +export function aggregateBuckets(rows: DailyMeanRow[]): { + count: number; + min: number | null; + max: number | null; + mean: number | null; +} { + if (rows.length === 0) { + return { count: 0, min: null, max: null, mean: null }; + } + let totalCount = 0; + let sumWeighted = 0; + let min = Infinity; + let max = -Infinity; + for (const r of rows) { + totalCount += r.count; + sumWeighted += r.count * r.mean; + if (r.minValue < min) min = r.minValue; + if (r.maxValue > max) max = r.maxValue; + } + if (totalCount === 0) { + return { count: 0, min: null, max: null, mean: null }; + } + return { + count: totalCount, + min: Number.isFinite(min) ? min : null, + max: Number.isFinite(max) ? max : null, + mean: sumWeighted / totalCount, + }; +} diff --git a/src/lib/measurements/rollups.ts b/src/lib/measurements/rollups.ts new file mode 100644 index 000000000..fed8be94d --- /dev/null +++ b/src/lib/measurements/rollups.ts @@ -0,0 +1,621 @@ +/** + * v1.5.0 — persistent measurement-rollup populator. + * + * Sits behind the 60-s in-process LRU as the second cache tier + * (per `.planning/research/v1434-r-cache-aggregation.md` §9). One row + * per `(userId, type, granularity, bucketStart)` in + * `measurement_rollups`, carrying the count / min / max / mean / sd / + * slope / r2 stats Postgres folds via `STDDEV_POP` / `REGR_SLOPE` / + * `REGR_R2`. The stats semantics mirror the v1.4.34.1 comprehensive + * aggregator so cached + live aggregates agree byte-for-byte. + * + * Write hooks: + * - The measurement create / update / delete endpoints call + * `recomputeBucketsForMeasurement` after their successful commit. + * The DAY bucket recompute runs inline (cheap, scoped to a single + * `(user, type, day)` tuple) so the next read is correct. + * WEEK / MONTH / YEAR recomputes enqueue onto pg-boss for the + * worker to fold. + * - The Apple Health import worker calls `recomputeUserRollups` + * ONCE at the end of the import, scoped to the import's date + * range, to avoid per-row write-hook overhead on 100k-row + * backfills. + * + * Reads: + * - `readRollupBuckets` returns the persisted rows. The reader + * side (`analytics`, `comprehensive`, `summaries-slice`) calls + * `isRollupFresh` first to decide whether to fall through to the + * live aggregator + persist-on-miss. Cheap heuristic — one + * indexed query for the newest measurement's `measuredAt`. + */ +import { prisma } from "@/lib/db"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import type { + MeasurementType, + RollupGranularity, +} from "@/generated/prisma/client"; + +/** All granularities the rollup table tracks. */ +export const ALL_GRANULARITIES: RollupGranularity[] = [ + "DAY", + "WEEK", + "MONTH", + "YEAR", +]; + +/** Granularities that recompute asynchronously via pg-boss. */ +const ASYNC_GRANULARITIES: RollupGranularity[] = ["WEEK", "MONTH", "YEAR"]; + +/** pg-boss queue name. */ +export const ROLLUP_RECOMPUTE_QUEUE = "rollup-recompute"; + +/** Worker concurrency cap — recomputes are read-heavy on Postgres. */ +export const ROLLUP_RECOMPUTE_CONCURRENCY = 2; + +/** + * Payload `boss.send` carries onto the `rollup-recompute` queue. The + * worker hands the values straight to `recomputeUserRollups`. + */ +export interface RollupRecomputePayload { + userId: string; + type: MeasurementType; + granularity: RollupGranularity; + /** ISO-8601 string — the worker parses through `new Date(value)`. */ + from: string; + /** ISO-8601 string — the worker parses through `new Date(value)`. */ + to: string; + /** Wall-clock kick-off for debugging only. */ + enqueuedAt: string; +} + +/** Postgres `date_trunc` unit per granularity. */ +const DATE_TRUNC_UNIT: Record = { + DAY: "day", + WEEK: "week", + MONTH: "month", + YEAR: "year", +}; + +/** Raw row shape returned by the aggregate `$queryRaw`. */ +interface RollupRow { + type: string; + bucket_start: Date; + count: bigint; + mean: number | null; + min_value: number | null; + max_value: number | null; + sd: number | null; + slope: number | null; + r2: number | null; +} + +/** + * Recompute the rollup buckets for one user. Bounded by the optional + * `types` / `granularities` / `from` / `to` filters; defaults cover + * every type, every granularity, and the trailing 5 years. + * + * Idempotent: rows are upserted under the + * `(userId, type, granularity, bucketStart)` composite primary key, + * so re-running the call is a no-op for unchanged buckets and a + * refresh for changed ones. + */ +export async function recomputeUserRollups( + userId: string, + opts: { + types?: MeasurementType[]; + granularities?: RollupGranularity[]; + from?: Date; + to?: Date; + } = {}, +): Promise<{ rowsUpserted: number; durationMs: number }> { + const startedAt = Date.now(); + const granularities = opts.granularities ?? ALL_GRANULARITIES; + const fiveYearsAgo = new Date(Date.now() - 5 * 365 * 24 * 60 * 60 * 1000); + const from = opts.from ?? fiveYearsAgo; + const to = opts.to ?? new Date(); + + let rowsUpserted = 0; + + for (const granularity of granularities) { + const rows = await runRollupAggregate({ + userId, + types: opts.types, + granularity, + from, + to, + }); + if (rows.length === 0) continue; + + rowsUpserted += await persistRollupRows(userId, granularity, rows); + } + + return { rowsUpserted, durationMs: Date.now() - startedAt }; +} + +/** + * Synchronous DAY bucket recompute for the write-path hooks plus an + * async enqueue of the WEEK / MONTH / YEAR recomputes. The DAY pass + * runs inline so the next read of `comprehensive` / `analytics` + * reflects the new row without waiting on pg-boss. + * + * Scoped to the affected `(userId, type, day)` tuple — cheap, and + * avoids invalidating buckets that haven't changed. + */ +export async function recomputeBucketsForMeasurement( + userId: string, + type: MeasurementType, + measuredAt: Date, +): Promise { + // DAY pass — synchronous. Pull the rollup for the affected day + // and upsert it. The aggregate is scoped to the single day so the + // SQL is bounded even for users with multi-year history. + const dayStart = startOfUtcDay(measuredAt); + const dayEnd = new Date(dayStart.getTime() + 24 * 60 * 60 * 1000); + const rows = await runRollupAggregate({ + userId, + types: [type], + granularity: "DAY", + from: dayStart, + to: dayEnd, + }); + if (rows.length === 0) { + // No rows in this day after the mutation (e.g. the DELETE path + // removed the last reading) — drop the rollup row if it exists. + await prisma.measurementRollup.deleteMany({ + where: { + userId, + type, + granularity: "DAY", + bucketStart: dayStart, + }, + }); + } else { + await persistRollupRows(userId, "DAY", rows); + } + + // WEEK / MONTH / YEAR — enqueue. Each granularity covers a range + // around the affected day so the worker only re-aggregates the + // bucket(s) the write could have changed. + for (const granularity of ASYNC_GRANULARITIES) { + const { from, to } = bucketSpan(measuredAt, granularity); + await enqueueRollupRecompute({ + userId, + type, + granularity, + from, + to, + }); + } +} + +/** + * Best-effort pg-boss enqueue. Silent no-op when no boss is attached + * (route is running in a context without the worker — typically the + * test suite). The reader-side fall-through still produces fresh + * data on a cache miss, so missing the enqueue is at-most a one-read + * cost rather than a correctness bug. + */ +export async function enqueueRollupRecompute(input: { + userId: string; + type: MeasurementType; + granularity: RollupGranularity; + from: Date; + to: Date; +}): Promise { + const boss = getGlobalBoss(); + if (!boss) return; + const payload: RollupRecomputePayload = { + userId: input.userId, + type: input.type, + granularity: input.granularity, + from: input.from.toISOString(), + to: input.to.toISOString(), + enqueuedAt: new Date().toISOString(), + }; + await boss.send(ROLLUP_RECOMPUTE_QUEUE, payload, { + retryLimit: 3, + retryDelay: 30, + retryBackoff: true, + // Coalesce duplicate enqueues for the same bucket key — multiple + // measurements landing on the same day shouldn't each spawn a + // separate worker run. The singleton key uses bucket-start so + // two writes inside the same week share one queued job. + singletonKey: `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}`, + }); +} + +/** + * Read rollup rows for `(userId, type, granularity)` in `[from, to)`. + * Returns rows sorted ascending by `bucketStart`. + */ +export async function readRollupBuckets( + userId: string, + type: MeasurementType, + granularity: RollupGranularity, + from: Date, + to: Date, +): Promise< + Array<{ + bucketStart: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + sd: number | null; + slope: number | null; + r2: number | null; + computedAt: Date; + }> +> { + const rows = await prisma.measurementRollup.findMany({ + where: { + userId, + type, + granularity, + bucketStart: { gte: from, lt: to }, + }, + orderBy: { bucketStart: "asc" }, + }); + return rows.map((r) => ({ + bucketStart: r.bucketStart, + count: r.count, + mean: r.mean, + minValue: r.minValue, + maxValue: r.maxValue, + sd: r.sd, + slope: r.slope, + r2: r.r2, + computedAt: r.computedAt, + })); +} + +/** + * Cheap freshness heuristic — one indexed query against + * `measurements` for the newest row per (user, type). Returns true + * when the persisted rollup covers every measurement currently in + * the DB, false when there's a newer measurement than the freshest + * rollup row (the rollup needs a recompute). + * + * `null` newest measurement means the user has no rows of this type, + * in which case the rollups are trivially fresh (empty == empty). + */ +export async function isRollupFresh( + userId: string, + type: MeasurementType, + granularity: RollupGranularity, + newestComputedAt: Date | null, +): Promise { + const newest = await prisma.measurement.findFirst({ + where: { userId, type }, + orderBy: { measuredAt: "desc" }, + select: { measuredAt: true, updatedAt: true }, + }); + if (!newest) { + // No measurements at all — empty rollup set is fresh by definition. + return true; + } + if (!newestComputedAt) { + // We have measurements but no rollup rows — definitely stale. + return false; + } + // Stale when the latest row was written / updated after the most + // recent rollup `computed_at` for this `(user, type, granularity)`. + const watermark = + newest.updatedAt > newest.measuredAt ? newest.updatedAt : newest.measuredAt; + return watermark <= newestComputedAt; +} + +// ─── internals ──────────────────────────────────────────── + +/** + * Pull the raw aggregates from Postgres. Groups by + * `date_trunc(, measured_at)` and folds the stats with + * `STDDEV_POP` / `REGR_SLOPE` / `REGR_R2`. The slope X-axis is days + * (epoch / 86400) so the slope is in "units per day", matching the + * comprehensive-aggregator + summaries-slice convention. + */ +async function runRollupAggregate(input: { + userId: string; + types: MeasurementType[] | undefined; + granularity: RollupGranularity; + from: Date; + to: Date; +}): Promise { + if (input.types && input.types.length === 0) return []; + + // Postgres rejects `date_trunc($1::text, m."measured_at")` in a + // GROUP BY because the planner sees the parameterised expression + // and can't match the SELECT-list to the GROUP-BY expression. The + // unit literal must be inlined. The value comes from a closed enum + // (DATE_TRUNC_UNIT) so there's no injection surface — we still + // assert against the whitelist before splicing. + const truncUnit = DATE_TRUNC_UNIT[input.granularity]; + if (!["day", "week", "month", "year"].includes(truncUnit)) { + throw new Error(`unexpected granularity: ${input.granularity}`); + } + const dateTrunc = `date_trunc('${truncUnit}', m."measured_at")`; + + if (input.types && input.types.length > 0) { + // Build the type-list literal carefully. The enum values come + // 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)) { + throw new Error(`invalid measurement type: ${t}`); + } + } + const typeList = input.types + .map((t) => `'${t}'::"measurement_type"`) + .join(","); + const sql = ` + SELECT + m."type"::text AS type, + ${dateTrunc} AS bucket_start, + COUNT(*) AS count, + AVG(m."value")::double precision AS mean, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value, + STDDEV_POP(m."value")::double precision AS sd, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + )::double precision AS slope, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + )::double precision AS r2 + FROM measurements m + WHERE m."user_id" = $1 + AND m."type" IN (${typeList}) + AND m."measured_at" >= $2 + AND m."measured_at" < $3 + GROUP BY m."type", ${dateTrunc} + `; + return prisma.$queryRawUnsafe( + sql, + input.userId, + input.from, + input.to, + ); + } + + const sql = ` + SELECT + m."type"::text AS type, + ${dateTrunc} AS bucket_start, + COUNT(*) AS count, + AVG(m."value")::double precision AS mean, + MIN(m."value")::double precision AS min_value, + MAX(m."value")::double precision AS max_value, + STDDEV_POP(m."value")::double precision AS sd, + REGR_SLOPE( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + )::double precision AS slope, + REGR_R2( + m."value", + EXTRACT(EPOCH FROM m."measured_at") / 86400.0 + )::double precision AS r2 + FROM measurements m + WHERE m."user_id" = $1 + AND m."measured_at" >= $2 + AND m."measured_at" < $3 + GROUP BY m."type", ${dateTrunc} + `; + return prisma.$queryRawUnsafe( + sql, + input.userId, + input.from, + input.to, + ); +} + +/** + * UPSERT the aggregate rows into `measurement_rollups`. Returns the + * number of rows touched. Uses a single chunked `$executeRaw` insert + * with `ON CONFLICT DO UPDATE` so the round-trip count is bounded + * regardless of fan-out width. + */ +async function persistRollupRows( + userId: string, + granularity: RollupGranularity, + rows: RollupRow[], +): Promise { + if (rows.length === 0) return 0; + // Chunk to keep the parameter count below Postgres' 65k cap and + // the transaction round-trip count bounded for large backfills. + // For the typical hot path (a single DAY recompute = 1 row, a + // multi-week WEEK recompute = a few rows) the round-trip count is + // negligible. + const CHUNK = 500; + let touched = 0; + for (let i = 0; i < rows.length; i += CHUNK) { + const slice = rows.slice(i, i + CHUNK); + await prisma.$transaction( + slice.map((row) => + prisma.measurementRollup.upsert({ + where: { + userId_type_granularity_bucketStart: { + userId, + type: row.type as MeasurementType, + granularity, + bucketStart: row.bucket_start, + }, + }, + create: { + userId, + type: row.type as MeasurementType, + granularity, + bucketStart: row.bucket_start, + count: Number(row.count), + mean: row.mean ?? 0, + minValue: row.min_value ?? 0, + maxValue: row.max_value ?? 0, + sd: row.sd, + slope: row.slope, + r2: row.r2, + computedAt: new Date(), + }, + update: { + count: Number(row.count), + mean: row.mean ?? 0, + minValue: row.min_value ?? 0, + maxValue: row.max_value ?? 0, + sd: row.sd, + slope: row.slope, + r2: row.r2, + computedAt: new Date(), + }, + }), + ), + ); + touched += slice.length; + } + return touched; +} + +/** + * Truncate a `Date` to the start of its UTC day. The rollup table + * stores TIMESTAMPTZ values so we anchor on UTC to keep the bucket + * key deterministic across timezone-shifting consumers. + */ +function startOfUtcDay(d: Date): Date { + return new Date( + Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate(), 0, 0, 0, 0), + ); +} + +/** + * Span of the bucket containing `measuredAt` at the given granularity. + * Postgres `date_trunc` is the canonical truncator; we mirror its + * Monday-anchored ISO week and calendar-month / calendar-year + * semantics in JS so the worker re-aggregation queries select the + * same set of rows the read-side `date_trunc` would group on. + */ +function bucketSpan( + measuredAt: Date, + granularity: RollupGranularity, +): { from: Date; to: Date } { + switch (granularity) { + case "DAY": { + const from = startOfUtcDay(measuredAt); + return { from, to: new Date(from.getTime() + 24 * 60 * 60 * 1000) }; + } + case "WEEK": { + const day = startOfUtcDay(measuredAt); + // Postgres ISO week: Monday is day 1. JS getUTCDay(): Sunday=0..Saturday=6. + const dayOfWeek = day.getUTCDay(); // 0=Sun + const mondayOffset = (dayOfWeek + 6) % 7; // 0 for Mon … 6 for Sun + const from = new Date( + day.getTime() - mondayOffset * 24 * 60 * 60 * 1000, + ); + return { from, to: new Date(from.getTime() + 7 * 24 * 60 * 60 * 1000) }; + } + case "MONTH": { + const from = new Date( + Date.UTC(measuredAt.getUTCFullYear(), measuredAt.getUTCMonth(), 1), + ); + const to = new Date( + Date.UTC( + measuredAt.getUTCFullYear(), + measuredAt.getUTCMonth() + 1, + 1, + ), + ); + return { from, to }; + } + case "YEAR": { + const from = new Date(Date.UTC(measuredAt.getUTCFullYear(), 0, 1)); + const to = new Date(Date.UTC(measuredAt.getUTCFullYear() + 1, 0, 1)); + return { from, to }; + } + } +} + +/** + * Cheap read-side warm-up. Called from the analytics + comprehensive + * read paths so the rollup table stays current for downstream + * consumers without each read paying the full live-aggregation cost + * twice. + * + * Contract: + * - If the user has at least one DAY rollup row that is current + * (newest measurement ≤ newest rollup `computed_at`), this is a + * no-op. + * - Otherwise the function kicks off a `recomputeUserRollups` over + * the trailing 90-day window (the dashboard hot path) for the + * DAY granularity only. WEEK / MONTH / YEAR follow on the next + * write hook or the explicit backfill — we don't pay for them + * here. + * - Best-effort: thrown errors are swallowed so the read path + * never fails because of a populator hiccup. + */ +export async function ensureUserRollupsFresh( + userId: string, +): Promise<{ recomputed: boolean }> { + try { + // Single round-trip — the most-recent DAY rollup for the user + // and the most-recent measurement updatedAt. If the rollup is + // ahead of the measurement, we're done. + const [newestRollup, newestMeasurement] = await Promise.all([ + prisma.measurementRollup.findFirst({ + where: { userId, granularity: "DAY" }, + orderBy: { computedAt: "desc" }, + select: { computedAt: true }, + }), + prisma.measurement.findFirst({ + where: { userId }, + orderBy: { updatedAt: "desc" }, + select: { updatedAt: true, measuredAt: true }, + }), + ]); + + if (!newestMeasurement) { + // User has no measurements — rollups are trivially fresh. + return { recomputed: false }; + } + const watermark = + newestMeasurement.updatedAt > newestMeasurement.measuredAt + ? newestMeasurement.updatedAt + : newestMeasurement.measuredAt; + if (newestRollup && watermark <= newestRollup.computedAt) { + return { recomputed: false }; + } + + // Stale — refresh the trailing 90-day DAY window. Synchronous so + // the next read of `comprehensive` / `analytics` benefits even on + // the first cold-mount fan-out. Bounded to DAY granularity so the + // read path doesn't accidentally pay the WEEK / MONTH / YEAR + // cost; those are handled by write hooks + the backfill script. + const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000); + await recomputeUserRollups(userId, { + granularities: ["DAY"], + from: ninetyDaysAgo, + to: new Date(), + }); + return { recomputed: true }; + } catch { + // Best-effort — never block the read path on a populator failure. + return { recomputed: false }; + } +} + +/** + * Collapse a list of `(type, day)` keys to the set of distinct + * `(type, dayStart)` pairs. Used by the batch-ingest hooks so a 500- + * row batch fires at most N DAY recomputes (typically <10 for a + * single-day batch). + */ +export function collapseToTypeDayKeys( + rows: Array<{ type: MeasurementType; measuredAt: Date }>, +): Array<{ type: MeasurementType; measuredAt: Date }> { + const seen = new Map(); + for (const row of rows) { + const dayStart = startOfUtcDay(row.measuredAt); + const key = `${row.type}|${dayStart.toISOString()}`; + if (!seen.has(key)) { + seen.set(key, { type: row.type, measuredAt: dayStart }); + } + } + return Array.from(seen.values()); +} diff --git a/tests/integration/measurement-rollups.test.ts b/tests/integration/measurement-rollups.test.ts new file mode 100644 index 000000000..0301423b7 --- /dev/null +++ b/tests/integration/measurement-rollups.test.ts @@ -0,0 +1,256 @@ +/** + * v1.5.0 — integration coverage for the persistent measurement- + * rollup populator. + * + * Runs against the testcontainer Postgres so the real `STDDEV_POP` / + * `REGR_SLOPE` / `REGR_R2` functions and the `date_trunc` semantics + * are exercised end-to-end. Three contracts are pinned: + * + * 1. **Synchronous DAY recompute on write hook.** Calling + * `recomputeBucketsForMeasurement` after a measurement insert + * writes a `MeasurementRollup` row with the right + * `count / mean / min / max / sd` shape. + * + * 2. **Backfill cardinality.** `recomputeUserRollups` over a + * synthetic 3-user × 4-type × 50-row fixture produces the + * expected (user × type × granularity × bucket) cardinality + * with no duplicate rows. + * + * 3. **Re-aggregated DataSummary matches live SQL.** A populated + * rollup-DAY series re-aggregated through `aggregateBuckets` + * returns `count / min / max / mean` byte-identical to a + * parallel live `$queryRaw` over the same rows. Establishes + * the byte-shape parity contract before the v1.5.1 read-path + * swap. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { + recomputeBucketsForMeasurement, + recomputeUserRollups, + ALL_GRANULARITIES, +} from "@/lib/measurements/rollups"; +import { aggregateBuckets } from "@/lib/measurements/rollup-read"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => null, // no pg-boss in the test container +})); + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe("measurement rollups — integration", () => { + it("writes a DAY rollup row that matches the inserted measurements", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-day-user", + email: "rollup-day-user@example.test", + role: "USER", + }, + }); + const measuredAt = new Date("2026-05-10T10:00:00.000Z"); + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "MANUAL", + measuredAt: new Date("2026-05-10T07:00:00.000Z"), + }, + { + userId: user.id, + type: "WEIGHT", + value: 82, + unit: "kg", + source: "MANUAL", + measuredAt: new Date("2026-05-10T15:00:00.000Z"), + }, + { + userId: user.id, + type: "WEIGHT", + value: 84, + unit: "kg", + source: "MANUAL", + measuredAt: new Date("2026-05-10T22:00:00.000Z"), + }, + ], + }); + + await recomputeBucketsForMeasurement(user.id, "WEIGHT", measuredAt); + + const dayRow = await prisma.measurementRollup.findFirst({ + where: { userId: user.id, type: "WEIGHT", granularity: "DAY" }, + }); + expect(dayRow).not.toBeNull(); + expect(dayRow!.count).toBe(3); + expect(dayRow!.mean).toBeCloseTo(82, 5); + expect(dayRow!.minValue).toBe(80); + expect(dayRow!.maxValue).toBe(84); + // STDDEV_POP across [80, 82, 84]: variance = (4+0+4)/3 ≈ 2.667; + // SD = sqrt ≈ 1.633. + expect(dayRow!.sd).toBeCloseTo(1.633, 2); + }); + + it("backfills with the expected (user × type × granularity × bucket) cardinality", async () => { + const prisma = getPrismaClient(); + const userIds: string[] = []; + const types = ["WEIGHT", "PULSE", "BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"] as const; + + // 3 users × 4 types × ~50 rows over ~7 weeks + for (let u = 0; u < 3; u++) { + const user = await prisma.user.create({ + data: { + username: `rollup-backfill-user-${u}`, + email: `rollup-backfill-${u}@example.test`, + role: "USER", + }, + }); + userIds.push(user.id); + interface SeedRow { + userId: string; + type: (typeof types)[number]; + value: number; + unit: string; + source: "MANUAL"; + measuredAt: Date; + } + const rows: SeedRow[] = []; + const baseMs = new Date("2026-03-01T08:00:00.000Z").getTime(); + for (const type of types) { + for (let i = 0; i < 50; i++) { + rows.push({ + userId: user.id, + type, + value: 80 + (i % 5), + unit: type === "WEIGHT" ? "kg" : type === "PULSE" ? "bpm" : "mmHg", + source: "MANUAL", + measuredAt: new Date(baseMs + i * 24 * 60 * 60 * 1000), + }); + } + } + await prisma.measurement.createMany({ data: rows }); + } + + for (const userId of userIds) { + await recomputeUserRollups(userId, { + granularities: ALL_GRANULARITIES, + }); + } + + // 50 distinct days per (user, type) → 50 DAY rollups. + const dayCount = await prisma.measurementRollup.count({ + where: { granularity: "DAY" }, + }); + expect(dayCount).toBe(3 * 4 * 50); + + // WEEK / MONTH / YEAR cardinalities should be > 0 and < dayCount. + const weekCount = await prisma.measurementRollup.count({ + where: { granularity: "WEEK" }, + }); + const monthCount = await prisma.measurementRollup.count({ + where: { granularity: "MONTH" }, + }); + const yearCount = await prisma.measurementRollup.count({ + where: { granularity: "YEAR" }, + }); + expect(weekCount).toBeGreaterThan(0); + expect(weekCount).toBeLessThan(dayCount); + expect(monthCount).toBeGreaterThan(0); + expect(monthCount).toBeLessThan(weekCount); + expect(yearCount).toBeGreaterThan(0); + expect(yearCount).toBeLessThanOrEqual(monthCount); + + // Re-running the backfill is idempotent — totals stay the same. + for (const userId of userIds) { + await recomputeUserRollups(userId, { + granularities: ALL_GRANULARITIES, + }); + } + const dayCountAfter = await prisma.measurementRollup.count({ + where: { granularity: "DAY" }, + }); + expect(dayCountAfter).toBe(dayCount); + }); + + it("re-aggregated DAY buckets match live SQL for count / min / max / mean (byte-shape parity)", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-parity-user", + email: "rollup-parity@example.test", + role: "USER", + }, + }); + + // 14-day descending-weight fixture spread across multiple days + // so the DAY rollup has > 1 buckets to aggregate. + const baseTime = new Date("2026-05-01T08:00:00.000Z").getTime(); + const N = 28; + const rows = Array.from({ length: N }, (_, i) => ({ + userId: user.id, + type: "WEIGHT" as const, + value: 80 + (i * 4) / (N - 1), // 80 → 84 + unit: "kg", + source: "MANUAL" as const, + measuredAt: new Date(baseTime + i * 12 * 60 * 60 * 1000), // 2/day + })); + await prisma.measurement.createMany({ data: rows }); + + // Backfill rollups. + await recomputeUserRollups(user.id, { granularities: ["DAY"] }); + + const buckets = await prisma.measurementRollup.findMany({ + where: { userId: user.id, type: "WEIGHT", granularity: "DAY" }, + orderBy: { bucketStart: "asc" }, + }); + + // Re-aggregate the DAY buckets through the rollup-read helper. + const reAggregated = aggregateBuckets( + buckets.map((b) => ({ + day: b.bucketStart, + count: b.count, + mean: b.mean, + minValue: b.minValue, + maxValue: b.maxValue, + })), + ); + + // Compare against a parallel live $queryRaw — the source of truth + // the v1.4.34.5 comprehensive aggregator uses. + const live = await prisma.$queryRaw< + Array<{ + count: bigint; + min_value: number; + max_value: number; + mean_value: number; + }> + >` + SELECT + 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" = ${user.id} + AND m."type" = 'WEIGHT' + `; + + expect(reAggregated.count).toBe(Number(live[0].count)); + expect(reAggregated.min).toBeCloseTo(live[0].min_value, 5); + expect(reAggregated.max).toBeCloseTo(live[0].max_value, 5); + expect(reAggregated.mean).toBeCloseTo(live[0].mean_value, 5); + }); +}); diff --git a/tests/integration/setup.ts b/tests/integration/setup.ts index 063971547..4be37416e 100644 --- a/tests/integration/setup.ts +++ b/tests/integration/setup.ts @@ -49,6 +49,7 @@ export async function truncateAllTables(client: PrismaClient): Promise { "host_metrics", "idempotency_keys", "integration_statuses", + "measurement_rollups", "measurements", "medication_intake_events", "medication_schedules",