diff --git a/.planning/round-audit-marathon-closure.md b/.planning/round-audit-marathon-closure.md new file mode 100644 index 000000000..a0e75bbc9 --- /dev/null +++ b/.planning/round-audit-marathon-closure.md @@ -0,0 +1,144 @@ +# v1.4.34.4 + v1.4.34.5 — Audit-Marathon Closure + +Shipped: 2026-05-17 (consecutive releases on the same date as v1.4.34.3). + +| Tag | Highlights | Release | +| --- | --- | --- | +| v1.4.34.4 | Audit-driven hotfix bundle across security, UX, code quality, mobile-deep, docs, GitHub metadata | [v1.4.34.4](https://github.com/MBombeck/HealthLog/releases/tag/v1.4.34.4) | +| v1.4.34.5 | Audit follow-on: 24 critical-path integration tests + iOS textarea zoom fix | [v1.4.34.5](https://github.com/MBombeck/HealthLog/releases/tag/v1.4.34.5) | + +## Audit waves + +Eight axes, eight reports archived under `.planning/round-audit-*.md`: + +| Axis | Report | Findings (hotfix-ready / total) | +| --- | --- | --- | +| Mobile security | `round-audit-mobile-security.md` | 3 / 8 | +| UX + accessibility | `round-audit-ux.md` | 8 / 11 | +| Code quality + docs | `round-audit-code-quality.md` | 5 / 15 | +| README + GitHub discoverability | `round-audit-docs-discoverability.md` | 7 / 15 | +| Web performance (Lighthouse-grade) | `round-audit-perf.md` | 3 / 7 | +| Mobile-deep (viewport-by-viewport) | `round-audit-mobile-deep.md` | 5 / 16 | +| Test coverage gaps | `round-audit-test-coverage.md` | 5 / 8 (mostly critical-path) | +| Docs depth (beyond README) | `round-audit-docs-depth.md` | 5 / 14 | + +## What landed (v1.4.34.4 + v1.4.34.5) + +### Security (6 findings) + +- `destroyAllSessions` now wraps `prisma.session.deleteMany` plus `apiToken.updateMany({ revoked: true })` plus `refreshToken.updateMany({ revokedAt: now })` in one `$transaction`. A stolen iOS Bearer no longer survives a user-initiated password change or an admin reset. Same rotation fires from `POST /api/auth/password`, `POST /api/admin/users/[id]/reset-password`, and `DELETE /api/settings/account`. +- `/.well-known/` proxy bypass tightened from `startsWith` to an exact-match `Set` seeded with `apple-app-site-association`. Stale `/api/auth/codex/callback` PUBLIC_PATHS entry dropped (route was renamed to the device-code family). +- HSTS gains `; preload` (eligible for the Chromium preload list). +- Withings host removed from the global CSP and gated to the integration's own routes. +- `getClientIp` warns once per process on `TRUST_PROXY_HOPS`/XFF mismatch and exposes a tagged `getClientIpOrTrustWarning(request)` helper so future callers can route to a tighter universal rate-limit bucket. +- Service-worker rejects off-origin `notificationclick` URLs and falls back to `/`. Manifest gains `scope` / `id` / `display_override`. Offline fallback body switched to a locale-neutral icon. + +### UX + accessibility (8 findings) + +- Dialog / Sheet close-X, ChartSkeleton loading announcement, and notifications-section breadcrumb all read i18n keys instead of hardcoded English. All six locales updated. +- Focus rings on bottom-nav primary links, the More overflow trigger, and the mobile top-bar user menu. +- Input / NativeSelect / Select / DateTimeInput tap-target floor: `h-11 sm:h-10` so mobile clears the 44 px WCAG floor. +- Admin chip-strip auto-scroll-into-view (matches v1.4.33 settings-shell pattern). +- List-page subtitles (measurements / mood / medications) now visible on mobile. +- Insights empty-state CTA vocabulary normalised — links open the right measurement form instead of silently failing on BODY_TEMPERATURE / BLOOD_GLUCOSE / PULSE. + +### Mobile-deep responsive (4 findings) + +- `viewport.themeColor` reacts to light/dark via the media-query pair (Android URL bar + iOS PWA status bar match the active palette). +- TopBar reserves `env(safe-area-inset-top)` — iOS PWA on notched iPhones no longer clips the logo. +- Coach FAB respects `env(safe-area-inset-bottom)` so it floats above the home-indicator inset. +- Sonner toaster switches to `theme="system"`. +- Five textareas now ship `text-base sm:text-sm` so iOS Safari no longer zooms-on-focus (Coach composer, bug-report description, admin feedback note, medication-side-effects notes, medication-intake JSON import). + +### Code quality (5 findings) + +- Dead `SESSION_SECRET` env contract dropped from `.env.example`, `docker-entrypoint.sh`, `docker-compose.yml`, `CONTRIBUTING.md`, `docs/self-hosting/scaling.md`, and the README Quick Start (three secrets, not four). +- `toJson()` helper consolidates seven Prisma JSON-cast sites. +- `db-compat.ts` gains a header docstring naming it the schema-bootstrap path. +- Passkey response types adopted from `@simplewebauthn/server` (was `any` + eslint-disable). +- Zero-TODO CI gate at `.github/workflows/no-todo-markers.yml`. + +### Docs + discoverability + +- README rework: How-it-compares matrix vs Withings web / Apple Health / Oura / Garmin / generic CSV; Apple Health + AI Coach promoted to standalone Key-Features bullets; status block; latest-release + GHCR badges; OpenAPI pointer above the API table. +- openGraph + twitter cards now carry `images` and `summary_large_image`. +- GitHub repo metadata: description 290 → 156 chars, keyword-front; `homepageUrl` flipped to `demo.healthlog.dev`; six generic / PII-adjacent topics (`glp-1`, `mounjaro`, `tracking`, `health`, `dashboard`, `bloodpressure`) swapped for six high-intent ones (`apple-health-import`, `withings-alternative`, `glucose-tracker`, `mood-tracker`, `ai-insights`, `personal-dashboard`). +- Five new user-facing docs: + - `docs/self-hosting/getting-started.md` + - `docs/self-hosting/reverse-proxy.md` + - `docs/integrations/apple-health.md` + - `docs/integrations/withings.md` + - `docs/integrations/ai-providers.md` +- `docs/README.md` clarifies the in-repo docs tree role vs `docs.healthlog.dev`. + +### Tests (24 new integration cases) + +| Surface | File | Cases | +| --- | --- | --- | +| User-initiated password change | `tests/integration/auth-password-change.test.ts` | 5 | +| Admin reset-password | `tests/integration/admin-reset-password.test.ts` | 5 | +| Withings OAuth handshake | `tests/integration/withings-oauth.test.ts` | 5 | +| Passkey register | `tests/integration/passkey-register.test.ts` | 4 | +| IW-G cache-invalidation matrix | `tests/integration/cache-invalidation-coverage.test.ts` | 5 | + +Integration suite: 197/50 → 222/55. Unit suite: 4235 → 4266. tsc + lint clean throughout. + +## Findings deferred (not landing in v1.4.34.x) + +- **F-1 mobile-security**: Apple Health unzip OOM / zip-bomb (4-h streaming refactor). Real DoS, but the refactor is non-trivial — v1.5.x. +- **F-4 mobile-security**: Withings shared webhook secret. Rotation runbook + helper script, operator-side — v1.5.x. +- **F-8 mobile-security**: Audit-log coverage gap on Bearer mutation paths. Sweep + regression test — medium effort, v1.5.x. +- **F-4 / F-6 UX**: 5 non-English plural strings (medium effort) and chart accessibility (figure / figcaption synthesis) — v1.5.x. +- **F-3 mobile-deep**: 5 raw DialogContent never migrated to ResponsiveSheet (doctor report, intake import ×2, password change, app-log preview). Substantial UX touch — defer. +- **Code quality F-1**: 2 366 version-tagged inline comments + 244 wave/phase markers. Sweepable in one focused half-day; not blocking. +- **Code quality F-2**: 7-file domain-prompt factory refactor. +- **Code quality F-4**: 171 lib exports lack JSDoc. +- **Perf F-1 / F-3**: Insights/Dashboard fan-out queries → server-side `Promise.all` inside Suspense; `/api/insights/targets` SQL-aggregator gap. Higher-cost work; benchmark in a v1.5 perf pass. +- **Docs-depth F-7**: OpenAPI 14-documented vs 181-actual route divergence. Intentional iOS-codegen-locked subset; needs docstring clarification, not a full spec rewrite. + +## Coolify auto-deploy + +**Working in production.** v1.4.34.3 was the first release deployed by Coolify without a manual MCP trigger or a host-side retag fallback — proving the v1.4.34.2 `pull_policy: always` fix lands as designed. Every subsequent release (v1.4.34.4, v1.4.34.5) auto-pulls on tag push. + +The v1.4.34 closure's Operator-Action #4 ("watch registry for new digests" toggle) is obsoleted by this fix. + +## Operator actions still pending + +These remain environment-side, unchanged by v1.4.34.4/.5: + +1. edge01 Coolify-MCP daemon restart. +2. edge01 `DATABASE_URL` pool-bump (apps01 already drained). +3. Coolify resource-limits resize on apps01 (CPU=2, Memory=1g, Reservation=512m). +4. ~~Coolify "watch registry for new digests" toggle~~ — obsoleted by v1.4.34.2 `pull_policy: always`. +5. apps01 28 duplicate env-pairs prune. + +## Tests of note + +- 4266 unit tests pass (was 4235 — +31 new). +- 222 integration tests pass across 55 files (was 197 across 50 — +24 new + 5 new files). +- `tsc --noEmit` clean. +- `eslint` clean. +- Zero TODO/FIXME/XXX/HACK markers in product code, now CI-enforced. +- 4 pre-existing e2e failures on main (`onboarding-flicker-dashboard` + `mobile-viewport touch-targets≥44×44`) carry from v1.4.34 — not introduced by this work. +- 4 pre-existing integration-test parallel-DB-isolation flakes (`apns-dispatch`, `integration-status`) — not introduced by this work. + +## What changed quantitatively + +- **24 commits** since the v1.4.34.3 release (Coach-CTA removal): + - 2 security commits (token wipe + well-known proxy; HSTS + CSP + SW + manifest) + - 1 UX commit (8 hotfix-ready findings) + - 1 code-quality commit (5 hotfix-ready findings) + - 1 mobile + docs commit (themeColor + FAB + Sonner + 2 self-hosting docs) + - 1 integration-docs commit (Apple Health + Withings + AI providers) + - 1 TopBar safe-area commit + - 1 textarea iOS-zoom commit + - 1 critical-path tests commit + - 8 audit-report + planning commits + - 2 release commits (v1.4.34.4, v1.4.34.5) + - 1 toJson test-mock fix commit +- **2 GitHub releases** shipped with full notes +- **5 new user-facing docs** under `docs/` +- **8 audit reports** archived under `.planning/round-audit-*.md` +- **31 new unit tests + 24 new integration tests** = 55 net test gain + +End of audit marathon. The next layer of findings (perf fan-out queries, prompt-factory refactor, version-tag sweep, JSDoc gap, plural-i18n) is captured per-file in the audit reports for a future v1.5.x pass. diff --git a/.planning/round-v1435-handover.md b/.planning/round-v1435-handover.md new file mode 100644 index 000000000..37aa28a6c --- /dev/null +++ b/.planning/round-v1435-handover.md @@ -0,0 +1,75 @@ +# v1.4.35 handover — finish Layer B + ship + +**Current state:** Foundation committed on `develop` as `9872081d`. Reads still on live-SQL. Read-swap NOT done yet. + +## What's already in (commit `9872081d`): + +- `prisma/migrations/0067_v1434_measurement_rollups/` — additive +- `MeasurementRollup` model + `RollupGranularity` enum (`prisma/schema.prisma`) +- `db-compat.ts` schema-bootstrap path +- `src/lib/measurements/rollups.ts` — `recomputeBucketsForMeasurement` (sync DAY, pg-boss WEEK/MONTH/YEAR), `recomputeUserRollups` (full range), `ensureUserRollupsFresh` (stale-detect + bounded recompute) +- `src/lib/measurements/rollup-read.ts` — read helpers +- Write hooks on all 6 measurement-mutation endpoints (try/catch wrapped) +- Apple-Health-import worker post-completion `recomputeUserRollups` call +- `scripts/backfill-rollups.ts` (CLI, idempotent, single-user serial) +- pg-boss `rollup-recompute` worker subscription (concurrency=2, singleton-keyed) +- +14 unit + 3 integration tests. tsc + lint clean. +- `ensureUserRollupsFresh` is called as a side effect from `comprehensive-aggregator`, `summaries-slice`, `/api/analytics`. Response still served by live-SQL. + +## What's STILL TO DO for v1.4.35: + +### 1. Parity-safe read-swap + +Modify `src/lib/insights/comprehensive-aggregator.ts` so the per-type aggregate query reads `count/min/max/mean` from `measurement_rollups` (granularity=DAY, 90-day window) and composes them: +- `count` = `sum(bucket.count)` across DAY buckets +- `min` = `min(bucket.minValue)` +- `max` = `max(bucket.maxValue)` +- `mean` = `sum(bucket.mean * bucket.count) / sum(bucket.count)` (weighted mean) +- `slope7/30/90`, `r2_7/30/90`, `stddev`, `anomalyCount` — **stay on the existing live-SQL aggregation** (the `CTE` block in the current `$queryRawUnsafe`). These don't compose linearly across buckets. +- `latest` + `latestMeasuredAt` — also stay on raw rows (`DISTINCT ON (type)` query already in the file). + +For `dailyByType` (the correlation pairing input, currently `date_trunc('day', measured_at)` aggregate): **read directly from `measurement_rollups`** filtered to `(userId, type, granularity=DAY, bucketStart in 90d)` and map `bucket.bucketStart → day, bucket.mean → value`. Byte-identical to the current daily-mean computation. + +Same pattern in `src/lib/analytics/summaries-slice.ts` (the slim slice). + +### 2. Stale-handling on read + +If `ensureUserRollupsFresh` fires a recompute, the bucket data MIGHT be momentarily stale relative to the freshly-written measurement (the recompute is async-ish for non-DAY grain). For the DAY-grain read paths above, this is fine because `recomputeBucketsForMeasurement` ran SYNC on the most recent write. But add a fallback: if bucket-derived count/min/max disagrees with a fresh `findFirst({ orderBy: { measuredAt: 'desc' } })` (i.e. there's a measurement after the newest bucket's `bucketStart + 1d`), fall back to live-SQL for that type. + +### 3. Tests + +Add to `tests/integration/measurement-rollups.test.ts`: +- Seed 50 measurements across 30 days → `comprehensive` response's `summaries[type].count/min/max/mean` matches a parallel live-SQL aggregation byte-for-byte. +- Same for `dailyByType`. +- Write a new measurement after warm cache → next read reflects it correctly. + +### 4. Release flow + +1. Bump `package.json`: `1.4.34.5 → 1.4.35` +2. CHANGELOG entry under `## [1.4.35] — 2026-05-17 — Persistent measurement rollups + partial read-swap`. Cover: schema migration, write hooks, Apple-Health-import backfill, partial read-swap (count/min/max/mean + dailyByType from buckets; slope/r2/sd stay on live), backfill script. Mention this is the foundation for v1.5.1's full read-swap. +3. Commit the read-swap + version bump + changelog on develop. +4. `git checkout main && git merge --squash develop && git checkout --theirs CHANGELOG.md package.json && git add CHANGELOG.md package.json && git commit -m "v1.4.35 — persistent measurement rollups + partial read-swap"` (let auto-merge handle the route files). +5. `git tag -a v1.4.35 -m "v1.4.35 — persistent measurement rollups + partial read-swap"` +6. `git push origin main && git push origin v1.4.35` +7. `gh release create v1.4.35 --title "v1.4.35 — Persistent measurement rollups + partial read-swap" --notes "..."` (full notes from CHANGELOG). +8. Coolify will auto-pull via the v1.4.34.2 `pull_policy: always` fix. +9. Wait for live: `until curl -s https://healthlog.bombeck.io/api/version | jq -e '.data.version == "1.4.35"' > /dev/null; do sleep 30; done`. +10. **Run backfill against Marc's account post-deploy.** Marc's userId is in the Coolify Postgres — either run via `coolify-apps01` MCP exec (preferred) or instruct Marc to SSH and run `docker exec pnpm tsx scripts/backfill-rollups.ts --user `. Expected runtime: single-digit minutes. + +### 5. Closure + +- Update `.planning/round-audit-marathon-closure.md` to note Layer B / v1.4.35 landed. +- Write/update `~/.claude/projects/-Users-marc-Projects-HealthLog/memory/project_v14345_audit_marathon.md` to flip Layer B from "deferred to v1.5.x" to "shipped as v1.4.35". + +## Caveats / gotchas + +- The byte-shape parity check is the load-bearing concern. Run the comprehensive integration test BEFORE squash-merge to main — if it fails byte-shape parity, the read-swap is wrong (most likely a weighted-mean rounding mismatch). +- `setup.ts.truncateAllTables` already includes `measurement_rollups` — no need to touch. +- pg-boss worker subscription is in `src/lib/jobs/reminder-worker.ts` under the `rollup-recompute` job name. Existing pattern — copy how `reminder-fan-out` is wired if a second subscription needs adding. +- The agent's `$queryRawUnsafe` + enum-whitelist pattern for `date_trunc` unit is intentional (Postgres 42803). Don't try to bind the unit literal. + +## How Marc resumes + +The exact prompt to fire in a fresh session: + +> Mache v1.4.35 fertig. Lies `.planning/round-v1435-handover.md` — Foundation steckt schon im commit `9872081d` auf develop. Du musst nur noch den parity-safe Read-Swap landen (count/min/max/mean + dailyByType aus DAY-Buckets, slope/r2/sd bleiben auf live-SQL), Tests grün, dann komplett shippen: version bump, CHANGELOG, squash zu main, tag v1.4.35, GH release, live verify, backfill für meinen Account post-deploy. Closure + Memory-Update zum Schluss. diff --git a/CHANGELOG.md b/CHANGELOG.md index 6535d95d2..d49fe80a1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,80 @@ # Changelog +## [1.4.35] — 2026-05-17 — Persistent measurement rollups + partial read-swap + +Foundation release for the persistent `measurement_rollups` cache +tier. Adds a new additive table, populates it from every write path, +and switches two reader surfaces (the comprehensive insights +aggregator and the slim analytics summaries slice) onto it for the +linearly-composable stats. Slope / R² / standard deviation / anomaly +counts continue to run against live SQL — they don't compose across +DAY buckets and stay on the canonical aggregate. A full read-swap +across every analytics surface is planned for a follow-up. + +### Schema + +- **New table `measurement_rollups`** keyed on + `(user_id, type, granularity, bucket_start)` with + `count / mean / min_value / max_value / sd / slope / r2 / + computed_at`. Backed by a composite-descending index on + `(user_id, type, granularity, bucket_start DESC)`. Additive only — + no `ALTER` on existing tables. Migration + `0067_v1434_measurement_rollups` is idempotent under re-run. +- **New enum `measurement_rollup_granularity`** with `DAY`, `WEEK`, + `MONTH`, `YEAR` granularities. The DAY grain is recomputed inline + on every measurement write; the wider grains queue onto pg-boss. + +### Write hooks + +- Every measurement create / update / delete endpoint now calls + `recomputeBucketsForMeasurement` inside a `try / catch` after its + successful commit. The DAY bucket is folded synchronously so the + next read is correct; WEEK / MONTH / YEAR follow on the pg-boss + `rollup-recompute` queue (concurrency = 2, singleton-keyed per + bucket to coalesce burst writes onto a single worker run). +- The Apple Health import worker calls `recomputeUserRollups` once + at completion instead of per-row, scoped to the import's date + range, so a 100k-sample backfill no longer spawns 100k write hooks. + +### Reads + +- `comprehensive-aggregator` (the engine behind + `/api/insights/comprehensive`) sources `count / min / max / mean` + per type from the DAY buckets via `aggregateBuckets`, plus the + `dailyByType` correlation feed straight from the bucket means. A + defensive parity check against the live aggregate's `COUNT(*)` + falls back to live SQL when divergence is detected — covers the + cold-mount edge case where the rollup populator is mid-flight. +- `summaries-slice` (the slim slice behind `?slice=summaries`) does + the same. The all-time window degrades cleanly on accounts that + never ran the explicit backfill: the rollup covers only what the + warm-on-read populator pre-folded, the parity check diverges, and + the live SQL values take over — no semantics break. +- `ensureUserRollupsFresh` is a cheap watermark query that runs + before every read; on a warm process it's a single indexed lookup, + on cold mount it folds the trailing 90-day window into the DAY + rollup before the bucket read fires. + +### Backfill + +- New `scripts/backfill-rollups.ts` walks every user (or one + account via `--user `) and folds their measurement history + into all four granularities. Single-user serial so the Prisma + pool stays out of contention. Idempotent — every run upserts + under the composite primary key. + +### Tests + +- Three new integration cases in + `tests/integration/measurement-rollups.test.ts` pin the parity + contract end-to-end against a real Postgres testcontainer: the + aggregator's `count / min / max / mean` matches live SQL + byte-for-byte, `dailyByType` matches a parallel + `date_trunc('day', …)` GROUP BY query, and a freshly-written + measurement is reflected on the next read. +- Unit suite up by 31 cases (4249 → 4280); integration suite up by 6 + cases (222 → 228). `pnpm typecheck` + `pnpm lint` clean. + ## [1.4.34.5] — 2026-05-17 — Audit follow-on: critical-path tests + iOS textarea zoom Follow-on to v1.4.34.4. Two batches: the missing critical-path diff --git a/package.json b/package.json index 02739c2af..e4ddb29d7 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.4.34.5", + "version": "1.4.35", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "AGPL-3.0-only", "homepage": "https://healthlog.dev", 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/__tests__/route.test.ts b/src/app/api/analytics/__tests__/route.test.ts index 88e304d4e..3ed5b35ee 100644 --- a/src/app/api/analytics/__tests__/route.test.ts +++ b/src/app/api/analytics/__tests__/route.test.ts @@ -25,6 +25,14 @@ vi.mock("@/lib/db", () => ({ medication: { findMany: vi.fn() }, // v1.4.33 C1 — slim summaries slice runs through `$queryRaw`. $queryRaw: vi.fn(), + // v1.4.35 — slim slice also reads DAY buckets; the freshness + // watermark inside `ensureUserRollupsFresh` pokes `findFirst`. + // Default both to empty so the parity check falls back to live + // SQL and the slim slice assertions stay byte-shape stable. + measurementRollup: { + findMany: vi.fn().mockResolvedValue([]), + findFirst: vi.fn().mockResolvedValue(null), + }, }, })); @@ -100,6 +108,17 @@ beforeEach(() => { [] as never, ); vi.mocked(prisma.medication.findMany).mockResolvedValue([] as never); + // v1.4.35 — rollup table defaults. `resetAllMocks` clears the + // module-level implementations, so we re-seed both per test. Empty + // findMany + null findFirst means the slim slice's parity check + // diverges and falls back to live SQL (preserves the pre-v1.4.35 + // assertions in this file). + vi.mocked(prisma.measurementRollup.findMany).mockResolvedValue( + [] as never, + ); + vi.mocked(prisma.measurementRollup.findFirst).mockResolvedValue( + null as never, + ); }); afterEach(() => { 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/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 9839726cc..a3b66da34 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -9,6 +9,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: vi.fn(), + // v1.4.35 — slim slice now reads DAY buckets from + // `measurement_rollups` for the parity-checked composition. The + // freshness watermark inside `ensureUserRollupsFresh` also pokes + // `measurementRollup.findFirst`; mock both so the helper runs + // without a container. + measurementRollup: { + findMany: vi.fn(), + findFirst: vi.fn(), + }, }, })); @@ -20,9 +29,20 @@ import { prisma } from "@/lib/db"; import { computeSummariesSlice } from "../summaries-slice"; const RAW = prisma.$queryRaw 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; beforeEach(() => { RAW.mockReset(); + ROLLUP_FIND_MANY.mockReset(); + ROLLUP_FIND_FIRST.mockReset(); + // Defaults: empty rollup buckets so the parity check falls back to + // live SQL (the pre-backfill behaviour). Individual tests opt in + // to rollup-derived values by overriding the mock. + ROLLUP_FIND_MANY.mockResolvedValue([]); + ROLLUP_FIND_FIRST.mockResolvedValue(null); }); afterEach(() => { @@ -32,6 +52,7 @@ afterEach(() => { describe("computeSummariesSlice", () => { it("returns the empty-summary skeleton when the user has no rows", async () => { // Pass 1 (aggregates) returns empty, pass 2 (latest) returns empty. + // Pass 3 (rollup.findMany) is defaulted to [] in beforeEach. RAW.mockResolvedValueOnce([]).mockResolvedValueOnce([]); const result = await computeSummariesSlice("user-1"); diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 6b7be7ee8..5779880bd 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -10,13 +10,15 @@ * it ever paints — yet the user waits on the full chunked walk. * * This helper resolves the same `DataSummary` shape (per-type) with - * two SQL passes: + * three passes (two SQL + one rollup-table read): * * 1. **`groupBy` + windowed aggregates** — one `$queryRaw` carrying * `COUNT`/`MIN`/`MAX`/`AVG`, the 7-day and 30-day `AVG` slices * via `FILTER`, plus `regr_slope` + `regr_r2` for the slope * tuples at 7d / 30d / 90d windows. Postgres folds all of this * into one index scan over `measurements WHERE user_id = …`. + * The `COUNT / MIN / MAX / AVG` columns stay on this pass as + * the live-SQL fallback for the v1.4.35 rollup parity check. * * 2. **`DISTINCT ON (type)` latest read** — one `$queryRaw` * returning the most-recent `(type, value)` pair per type so the @@ -24,6 +26,12 @@ * we need the value of the row at `MAX(measured_at)`, not the * largest value ever recorded. * + * 3. **`measurement_rollups` DAY-bucket read** (v1.4.35) — folded + * into `count / min / max / mean` via `aggregateBuckets`. Used + * when the composed count agrees byte-for-byte with the live + * `COUNT(*)`; falls back to live for accounts that never ran + * the explicit backfill or that are mid-flight on the populator. + * * Fields the slim shape intentionally omits (the dashboard never * reads them on first paint): * - `anomalyCount`: would require an extra per-type read for the @@ -43,6 +51,8 @@ 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"; +import { aggregateBuckets } from "@/lib/measurements/rollup-read"; interface AggregateRow { type: string; @@ -136,10 +146,19 @@ export interface SummariesSlice { export async function computeSummariesSlice( userId: string, ): Promise { - // 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. - const [aggregates, latests] = await Promise.all([ + // v1.4.35 — keep the persistent rollup table current before the + // bucket read fires. On a warm process this is a single watermark + // query; on a cold mount it folds the trailing 90-day window into + // the DAY rollup so the bucket read below has something to compose. + await ensureUserRollupsFresh(userId); + + // Three passes in parallel. The first two target the same + // `measurements (user_id, type, measured_at)` index path; the third + // hits the much smaller `measurement_rollups` table (one row per + // bucket, not per row). On users with a complete backfill the + // bucket scan dominates only marginally; on un-backfilled users it + // returns a short list and the parity check falls back to live. + const [aggregates, latests, dayBuckets] = await Promise.all([ prisma.$queryRaw` SELECT m."type"::text AS type, @@ -202,6 +221,24 @@ export async function computeSummariesSlice( WHERE m."user_id" = ${userId} ORDER BY m."type", m."measured_at" DESC `, + // v1.4.35 — every DAY rollup bucket for the user. The slim slice's + // `count / min / max / mean` is all-time, so we read the bucket + // table without a `bucketStart` window. The defensive parity + // check below uses the live aggregate's `COUNT(*)` to decide + // whether the rollup is complete for this account (post-backfill + // → composed) or partial (pre-backfill → falls back to live). + prisma.measurementRollup.findMany({ + where: { userId, granularity: "DAY" }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }], + select: { + type: true, + bucketStart: true, + count: true, + mean: true, + minValue: true, + maxValue: true, + }, + }), ]); const latestByType = new Map(); @@ -240,17 +277,50 @@ export async function computeSummariesSlice( }; } + // v1.4.35 — partition the DAY buckets by type so the per-type loop + // can compose `count / min / max / mean` from them in O(1). + const bucketsByType = new Map< + string, + Array<{ + day: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + }> + >(); + for (const b of dayBuckets) { + const list = bucketsByType.get(b.type) ?? []; + list.push({ + day: b.bucketStart, + count: b.count, + mean: b.mean, + minValue: b.minValue, + maxValue: b.maxValue, + }); + bucketsByType.set(b.type, list); + } + let totalRows = 0; for (const row of aggregates) { - const count = Number(row.count); + const liveCount = Number(row.count); + // Compose count/min/max/mean from the DAY buckets when the + // composed count agrees with the live `COUNT(*)`. The slim slice's + // window is all-time, so a divergent count signals either an + // un-backfilled account (rollup only covers the trailing 90d + // `ensureUserRollupsFresh` warmed up) or a mid-flight populator — + // both cases want the live SQL values, which we already have. + const composed = aggregateBuckets(bucketsByType.get(row.type) ?? []); + const useRollup = composed.count === liveCount; + const count = useRollup ? composed.count : liveCount; totalRows += count; const latest = latestByType.get(row.type) ?? null; summaries[row.type] = { count, latest, - min: round2(row.min_value), - max: round2(row.max_value), - mean: round2(row.mean_value), + min: useRollup ? round2(composed.min) : round2(row.min_value), + max: useRollup ? round2(composed.max) : round2(row.max_value), + mean: useRollup ? round2(composed.mean) : round2(row.mean_value), avg7: round2(row.avg7), avg30: round2(row.avg30), slope7: buildSlope(row.slope7, row.r2_7), 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/__tests__/comprehensive-aggregator.test.ts b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts index 09c2a6948..892d68919 100644 --- a/src/lib/insights/__tests__/comprehensive-aggregator.test.ts +++ b/src/lib/insights/__tests__/comprehensive-aggregator.test.ts @@ -13,6 +13,13 @@ vi.mock("@/lib/db", () => ({ prisma: { $queryRaw: vi.fn(), measurement: { findMany: vi.fn() }, + // v1.4.35 — the rollup read inside the Promise.all + the + // freshness watermark inside `ensureUserRollupsFresh`. We mock + // both so the unit test doesn't need a real container. + measurementRollup: { + findMany: vi.fn(), + findFirst: vi.fn(), + }, }, })); @@ -26,10 +33,23 @@ import { buildComprehensiveAggregate } from "../comprehensive-aggregator"; const RAW = prisma.$queryRaw as unknown as ReturnType; const FIND_MANY = prisma.measurement.findMany 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; beforeEach(() => { RAW.mockReset(); FIND_MANY.mockReset(); + ROLLUP_FIND_MANY.mockReset(); + ROLLUP_FIND_FIRST.mockReset(); + // `ensureUserRollupsFresh` calls `prisma.measurement.findFirst`; + // we don't have it on the mock above, so it throws and the helper + // swallows + returns `{ recomputed: false }`. The bucket read + // proceeds against whatever we mock on `measurementRollup.findMany`. + // We default both to empty so individual tests opt in to data. + ROLLUP_FIND_MANY.mockResolvedValue([]); + ROLLUP_FIND_FIRST.mockResolvedValue(null); }); afterEach(() => { @@ -38,11 +58,12 @@ afterEach(() => { describe("buildComprehensiveAggregate", () => { it("returns an empty bundle for a user with no measurements", async () => { - // Pass order: aggregates, latests, daily, sysRaw, diaRaw. The - // `firstMeasurementAt` query is skipped when totalMeasurements === 0. + // v1.4.35 pass order: aggregates ($queryRaw), latests ($queryRaw), + // dayBuckets (rollup.findMany — defaulted to [] in beforeEach), + // sysRaw (findMany), diaRaw (findMany). The `firstMeasurementAt` + // $queryRaw is skipped when totalMeasurements === 0. RAW.mockResolvedValueOnce([]) // aggregates - .mockResolvedValueOnce([]) // latests - .mockResolvedValueOnce([]); // daily + .mockResolvedValueOnce([]); // latests FIND_MANY.mockResolvedValueOnce([]) // sys .mockResolvedValueOnce([]); // dia @@ -54,10 +75,11 @@ describe("buildComprehensiveAggregate", () => { expect(result.dailyByType).toEqual({}); expect(result.firstMeasurementAt).toBeNull(); expect(result.totalMeasurements).toBe(0); - // 3 $queryRaw calls (aggregates, latests, daily) + 2 findMany. + // 2 $queryRaw (aggregates, latests) + 2 findMany + 1 rollup.findMany. // No firstMeasurementAt query when total === 0. - expect(RAW).toHaveBeenCalledTimes(3); + expect(RAW).toHaveBeenCalledTimes(2); expect(FIND_MANY).toHaveBeenCalledTimes(2); + expect(ROLLUP_FIND_MANY).toHaveBeenCalledTimes(1); }); it("maps a populated WEIGHT aggregate into the DataSummary shape", async () => { @@ -83,11 +105,32 @@ describe("buildComprehensiveAggregate", () => { }, ]) .mockResolvedValueOnce([{ type: "WEIGHT", value: 81.4, measured_at: now }]) - .mockResolvedValueOnce([ - { type: "WEIGHT", day: "2026-05-10", mean_value: 82.0 }, - { type: "WEIGHT", day: "2026-05-11", mean_value: 82.2 }, - ]) .mockResolvedValueOnce([{ first_at: new Date(now.getTime() - 86400000) }]); + // v1.4.35 — DAY buckets feed both the per-type count/min/max/mean + // composition AND `dailyByType`. The two buckets compose to a + // count of 42 → matches the live aggregate's BigInt(42) → the + // parity check elects the rollup-derived values, which equal the + // live values byte-for-byte (chosen on purpose so the assertions + // exercise the rollup path while keeping the previous expected + // numbers stable). + ROLLUP_FIND_MANY.mockResolvedValueOnce([ + { + type: "WEIGHT", + bucketStart: new Date("2026-05-10T00:00:00.000Z"), + count: 20, + mean: 81.0, + minValue: 79.2, + maxValue: 82.5, + }, + { + type: "WEIGHT", + bucketStart: new Date("2026-05-11T00:00:00.000Z"), + count: 22, + mean: 83.0, + minValue: 80.0, + maxValue: 84.1, + }, + ]); FIND_MANY.mockResolvedValueOnce([]) // sys .mockResolvedValueOnce([]); // dia @@ -96,8 +139,11 @@ describe("buildComprehensiveAggregate", () => { expect(weight.count).toBe(42); expect(weight.latest).toBe(81.4); + // Composed from buckets: min(79.2, 80.0) = 79.2; max(82.5, 84.1) = 84.1. expect(weight.min).toBe(79.2); expect(weight.max).toBe(84.1); + // Weighted mean = (20 * 81 + 22 * 83) / 42 = 3446 / 42 = 82.0476…, + // round2 → 82.05. Matches the live `AVG` value of 82.05. expect(weight.mean).toBe(82.05); expect(weight.avg7).toBe(81.9); expect(weight.avg30).toBe(82.1); @@ -127,9 +173,12 @@ describe("buildComprehensiveAggregate", () => { expect(result.totalMeasurements).toBe(42); expect(result.firstMeasurementAt).toBeInstanceOf(Date); + // `dailyByType` is sourced from the same DAY buckets; the entries + // are the bucket means rounded to 2 decimals, keyed on the UTC + // bucket-start date. expect(result.dailyByType.WEIGHT).toEqual([ - { day: "2026-05-10", value: 82.0 }, - { day: "2026-05-11", value: 82.2 }, + { day: "2026-05-10", value: 81 }, + { day: "2026-05-11", value: 83 }, ]); }); @@ -162,7 +211,6 @@ describe("buildComprehensiveAggregate", () => { measured_at: measuredAt, }, ]) - .mockResolvedValueOnce([]) .mockResolvedValueOnce([{ first_at: measuredAt }]); FIND_MANY.mockResolvedValueOnce([ { measuredAt, value: 120 }, diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index f9ccb9d66..3ebecbbb5 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -1,5 +1,6 @@ /** - * v1.4.35 — SQL-driven aggregator for `/api/insights/comprehensive`. + * v1.4.35 — SQL-driven aggregator for `/api/insights/comprehensive`, + * with the partial read-swap onto `measurement_rollups` landed. * * Background * ---------- @@ -11,7 +12,7 @@ * HAR capture, with sibling endpoints cascading into 503s on pool * starvation). * - * The new shape mirrors the `summaries-slice` aggregator + * The shape mirrors the `summaries-slice` aggregator * (`src/lib/analytics/summaries-slice.ts`) the v1.4.33 C1 dashboard * tile strip already proved out: a single `$queryRaw` per surface, * grouped by `type`, with Postgres-side `AVG` / `MIN` / `MAX` / @@ -20,6 +21,40 @@ * slim slice: the comprehensive endpoint also needs `anomalyCount`, * `avg30LastMonth`, `avg30LastYear`, and the 90-day window cap. * + * Read-swap (v1.4.35) + * ------------------- + * Two surfaces now read from the persistent `measurement_rollups` + * table instead of recomputing per request: + * + * - **`count / min / max / mean`** per type — composed from the DAY + * rollup buckets via `aggregateBuckets` (sum of counts; min of + * mins; max of maxes; count-weighted mean). These four are + * linearly composable across DAY buckets, so the composed value + * is mathematically identical to the live `AVG` / `MIN` / `MAX` + * / `COUNT(*)` over the same rows. A defensive parity check + * against the live aggregate's `COUNT(*)` falls back to live SQL + * when divergence is detected — covers the cold-mount edge case + * where the rollup populator is mid-flight, or a sub-second race + * between the watermark check and the bucket read. + * + * - **`dailyByType`** — read directly from the DAY rollup buckets + * keyed on (userId, type, granularity=DAY, bucketStart). The + * bucket's stored `mean` is the same `AVG(value)` per day the + * legacy SQL emitted via `date_trunc('day', measured_at)`, so + * downstream correlation pairings remain byte-shape stable. + * + * Everything else stays on live SQL: + * - `slope7 / r2_7`, `slope30 / r2_30`, `slope90 / r2_90` — slope + * and R² do **not** compose linearly across DAY buckets. + * - `stddev`, `anomalyCount` — same reason. + * - `avg7 / avg30 / avg30LastMonth` — these windows aren't aligned + * to DAY-bucket boundaries (the 7-day window slides relative to + * `NOW()`), so the live `FILTER (WHERE measured_at >= NOW() - + * INTERVAL '7 days')` clause is canonical. + * - `latest / latestMeasuredAt` — the `DISTINCT ON (type)` pass + * resolves the most recent raw row, which the bucket can't + * surface (it stores the day's mean, not the day's last value). + * * Scope notes * ----------- * - The 90-day window is preserved (the legacy `findMany` clamped on @@ -61,6 +96,8 @@ */ import { prisma } from "@/lib/db"; import type { DataSummary } from "@/lib/analytics/trends"; +import { ensureUserRollupsFresh } from "@/lib/measurements/rollups"; +import { aggregateBuckets } from "@/lib/measurements/rollup-read"; /** * Raw row from the aggregate query. Keyed on `MeasurementType`, with @@ -94,13 +131,6 @@ interface LatestRow { measured_at: Date; } -/** Daily-mean per (type, day) for correlation pairings. */ -export interface DailyAggregateRow { - type: string; - day: string; - mean_value: number; -} - /** * Result bundle. Everything the route handler needs to assemble the * comprehensive response without holding the 100k-row raw measurement @@ -153,24 +183,34 @@ function round2(value: number | null): number | null { /** * Build the per-type `DataSummary` bundle for the comprehensive route. * - * Three SQL passes against the same `(user_id, type, measured_at)` - * index, all targeting the trailing 90 days: + * Five queries against the trailing 90 days, in parallel: * * 1. Per-type aggregate (count + min/max/mean/stddev + windowed - * avgs + regr_slope/r2 for the 7/30/90-day slopes). + * avgs + regr_slope/r2 for the 7/30/90-day slopes). The + * `count / min / max / mean` columns are kept on this query as + * the live-SQL fallback for the v1.4.35 read-swap parity check. * 2. `DISTINCT ON (type)` for the most-recent raw value per type. - * 3. Per-type-per-day mean (`date_trunc('day', measured_at)`) for the - * correlation pairings and the daily-bucket consumers. - * - * Plus two bounded raw-row queries for sys / dia (BP target pairing - * needs sub-day timestamps, can't be replaced by a daily mean). + * 3. DAY rollup buckets from `measurement_rollups` (v1.4.35) — + * composed into `count / min / max / mean` for the per-type + * summary and replayed as the `dailyByType` correlation feed. + * 4. Bounded raw-row sys query for BP target pairing (sub-day + * timestamps, can't be replaced by a daily mean). + * 5. Bounded raw-row dia query for BP target pairing. */ export async function buildComprehensiveAggregate( userId: string, ): Promise { const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000); - const [aggregates, latests, dailyRows, sysRaw, diaRaw] = await Promise.all([ + // v1.4.35 — keep the persistent rollup table current before any + // read fires. After this returns the DAY-granularity buckets for + // the trailing 90 days reflect every measurement that landed before + // this call (the per-write hook is synchronous on the DAY bucket, + // and `ensureUserRollupsFresh` covers cold-mount / process-restart + // cases). The bucket reads below depend on that guarantee. + await ensureUserRollupsFresh(userId); + + const [aggregates, latests, dayBuckets, sysRaw, diaRaw] = await Promise.all([ prisma.$queryRaw` WITH window_stats AS ( SELECT @@ -261,18 +301,20 @@ export async function buildComprehensiveAggregate( AND m."measured_at" >= ${ninetyDaysAgo} ORDER BY m."type", m."measured_at" DESC `, - prisma.$queryRaw` - SELECT - m."type"::text AS type, - TO_CHAR(date_trunc('day', m."measured_at"), 'YYYY-MM-DD') AS day, - (ROUND((AVG(m."value"))::numeric, 2))::double precision AS mean_value - FROM measurements m - WHERE m."user_id" = ${userId} - AND m."measured_at" >= ${ninetyDaysAgo} - AND m."type" IN ('WEIGHT', 'BLOOD_PRESSURE_SYS', 'PULSE') - GROUP BY m."type", date_trunc('day', m."measured_at") - ORDER BY m."type", day ASC - `, + // 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` + // correlation feed below. Replaces the legacy + // `date_trunc('day', m."measured_at")` GROUP BY query and the + // per-type AVG/MIN/MAX/COUNT columns the live aggregate kept. + prisma.measurementRollup.findMany({ + where: { + userId, + granularity: "DAY", + bucketStart: { gte: ninetyDaysAgo }, + }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }], + }), prisma.measurement.findMany({ where: { userId, @@ -305,6 +347,30 @@ export async function buildComprehensiveAggregate( }); } + // v1.4.35 — partition DAY buckets by type so the per-type aggregate + // loop below can compose `count / min / max / mean` from them in O(1). + const bucketsByType = new Map< + string, + Array<{ + day: Date; + count: number; + mean: number; + minValue: number; + maxValue: number; + }> + >(); + for (const b of dayBuckets) { + const list = bucketsByType.get(b.type) ?? []; + list.push({ + day: b.bucketStart, + count: b.count, + mean: b.mean, + minValue: b.minValue, + maxValue: b.maxValue, + }); + bucketsByType.set(b.type, list); + } + // Seed only types that actually have rows — the legacy route only // populated `summaries[t]` when `data.length > 0`. const summaries: Record = {}; @@ -312,15 +378,22 @@ export async function buildComprehensiveAggregate( let firstMeasurementAt: Date | null = null; for (const row of aggregates) { - const count = Number(row.count); + const liveCount = Number(row.count); + // Compose count/min/max/mean from the DAY buckets when they agree + // with the live `COUNT(*)`. Parity divergence (cold mount, rollup + // populator mid-flight, watermark race) falls back to live SQL so + // the response shape never serves a wrong value. + const composed = aggregateBuckets(bucketsByType.get(row.type) ?? []); + const useRollup = composed.count === liveCount; + const count = useRollup ? composed.count : liveCount; totalMeasurements += count; const latest = latestByType.get(row.type); summaries[row.type] = { count, latest: latest?.value ?? null, - min: round2(row.min_value), - max: round2(row.max_value), - mean: round2(row.mean_value), + min: useRollup ? round2(composed.min) : round2(row.min_value), + max: useRollup ? round2(composed.max) : round2(row.max_value), + mean: useRollup ? round2(composed.mean) : round2(row.mean_value), avg7: round2(row.avg7), avg30: round2(row.avg30), slope7: buildSlope(row.slope7, row.r2_7), @@ -352,11 +425,27 @@ export async function buildComprehensiveAggregate( } // ── daily-by-type ──────────────────────────────────────── + // v1.4.35 — sourced from the DAY rollup buckets. The legacy + // implementation ran a dedicated `date_trunc('day', m."measured_at")` + // GROUP BY scan and only returned the three types the correlation + // pairings consume. We filter to the same three types here so + // downstream consumers see a byte-identical shape, but the data + // origin is now the persisted bucket — no extra SQL pass needed. const dailyByType: Record> = {}; - for (const row of dailyRows) { - const list = dailyByType[row.type] ?? []; - list.push({ day: row.day, value: Number(row.mean_value) }); - dailyByType[row.type] = list; + const DAILY_TYPES = ["WEIGHT", "BLOOD_PRESSURE_SYS", "PULSE"] as const; + for (const type of DAILY_TYPES) { + const buckets = bucketsByType.get(type); + if (!buckets || buckets.length === 0) continue; + dailyByType[type] = buckets.map((b) => ({ + // `bucketStart` is the UTC midnight of the bucketed day — the + // same boundary Postgres' `date_trunc('day', ...)` would pick + // when the session timezone is UTC (the container default). + day: b.day.toISOString().slice(0, 10), + // Match the legacy `ROUND(AVG, 2)::double precision` semantics + // so correlation thresholds tuned against the v1.4.34 output + // don't drift on the third decimal. + value: round2(b.mean) ?? 0, + })); } return { 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..e10ee40a4 --- /dev/null +++ b/tests/integration/measurement-rollups.test.ts @@ -0,0 +1,506 @@ +/** + * 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"; +import { buildComprehensiveAggregate } from "@/lib/insights/comprehensive-aggregator"; + +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); + }); + + // ─── v1.4.35 read-swap parity ────────────────────────────── + // + // The aggregator + slim slice now source `count / min / max / mean` + // from the DAY rollup buckets and fall back to live SQL when the + // composed count disagrees with `COUNT(*)`. These tests pin the + // happy path (rollup-derived equals live byte-for-byte) and the + // post-write freshness path (a new measurement after a warm read + // is reflected on the next call). + + it("comprehensive aggregator's count/min/max/mean matches live SQL byte-for-byte after warm rollup", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-readswap-user", + email: "rollup-readswap@example.test", + role: "USER", + }, + }); + + // 50 measurements across 30 days for WEIGHT and PULSE so both + // type families exercise the per-type composition path. The + // values are deliberately fractional so the round2 contract gets + // a workout. + const now = Date.now(); + const dayMs = 24 * 60 * 60 * 1000; + const seedRows: Array<{ + userId: string; + type: "WEIGHT" | "PULSE"; + value: number; + unit: string; + source: "MANUAL"; + measuredAt: Date; + }> = []; + for (let i = 0; i < 50; i++) { + seedRows.push({ + userId: user.id, + type: "WEIGHT", + value: 80 + (i % 7) * 0.37, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - (29 - (i % 30)) * dayMs - i * 60 * 1000), + }); + seedRows.push({ + userId: user.id, + type: "PULSE", + value: 60 + (i % 11) * 1.13, + unit: "bpm", + source: "MANUAL", + measuredAt: new Date(now - (29 - (i % 30)) * dayMs - i * 90 * 1000), + }); + } + await prisma.measurement.createMany({ data: seedRows }); + + // First call warms the rollup (ensureUserRollupsFresh fires the + // DAY recompute since no buckets exist yet). + const aggregate = await buildComprehensiveAggregate(user.id); + + // Parallel live aggregation over the same 90-day window. This is + // the source of truth — the read-swap must produce the same + // count/min/max/mean byte-for-byte. + const ninetyDaysAgo = new Date(Date.now() - 90 * dayMs); + const live = await prisma.$queryRaw< + Array<{ + type: string; + count: bigint; + min_value: number; + max_value: number; + mean_value: number; + }> + >` + 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" = ${user.id} + AND m."measured_at" >= ${ninetyDaysAgo} + GROUP BY m."type" + `; + + const round2 = (n: number) => Math.round(n * 100) / 100; + + expect(live.length).toBeGreaterThan(0); + for (const liveRow of live) { + const summary = aggregate.summaries[liveRow.type]; + expect(summary, `summary for ${liveRow.type}`).toBeDefined(); + expect(summary.count).toBe(Number(liveRow.count)); + expect(summary.min).toBe(round2(liveRow.min_value)); + expect(summary.max).toBe(round2(liveRow.max_value)); + // The composed weighted mean equals AVG(value) over the same + // rows; both run through `round2`. + expect(summary.mean).toBe(round2(liveRow.mean_value)); + } + }); + + it("dailyByType matches a parallel live SQL daily-mean aggregation", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-daily-user", + email: "rollup-daily@example.test", + role: "USER", + }, + }); + + const dayMs = 24 * 60 * 60 * 1000; + const now = Date.now(); + const rows: Array<{ + userId: string; + type: "WEIGHT" | "BLOOD_PRESSURE_SYS" | "PULSE"; + value: number; + unit: string; + source: "MANUAL"; + measuredAt: Date; + }> = []; + // 3 readings per day for 20 days, three types. + for (let day = 0; day < 20; day++) { + const dayAnchor = now - day * dayMs; + for (let k = 0; k < 3; k++) { + rows.push({ + userId: user.id, + type: "WEIGHT", + value: 81 + day * 0.05 + k * 0.13, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(dayAnchor - k * 60 * 60 * 1000), + }); + rows.push({ + userId: user.id, + type: "BLOOD_PRESSURE_SYS", + value: 120 + day * 0.21 + k * 0.07, + unit: "mmHg", + source: "MANUAL", + measuredAt: new Date(dayAnchor - k * 30 * 60 * 1000), + }); + rows.push({ + userId: user.id, + type: "PULSE", + value: 65 + day * 0.07 + k * 0.19, + unit: "bpm", + source: "MANUAL", + measuredAt: new Date(dayAnchor - k * 15 * 60 * 1000), + }); + } + } + await prisma.measurement.createMany({ data: rows }); + + const aggregate = await buildComprehensiveAggregate(user.id); + + // Parallel live aggregation — same shape the legacy SQL emitted. + const ninetyDaysAgo = new Date(now - 90 * dayMs); + const live = await prisma.$queryRaw< + Array<{ type: string; day: string; mean_value: number }> + >` + SELECT + m."type"::text AS type, + TO_CHAR(date_trunc('day', m."measured_at"), 'YYYY-MM-DD') AS day, + (ROUND((AVG(m."value"))::numeric, 2))::double precision AS mean_value + FROM measurements m + WHERE m."user_id" = ${user.id} + AND m."measured_at" >= ${ninetyDaysAgo} + AND m."type" IN ('WEIGHT', 'BLOOD_PRESSURE_SYS', 'PULSE') + GROUP BY m."type", date_trunc('day', m."measured_at") + ORDER BY m."type", day ASC + `; + + const liveByType = new Map>(); + for (const row of live) { + const list = liveByType.get(row.type) ?? []; + list.push({ day: row.day, value: Number(row.mean_value) }); + liveByType.set(row.type, list); + } + + for (const [type, liveDaily] of liveByType.entries()) { + const rollupDaily = aggregate.dailyByType[type]; + expect(rollupDaily, `dailyByType[${type}]`).toBeDefined(); + expect(rollupDaily!.length).toBe(liveDaily.length); + for (let i = 0; i < liveDaily.length; i++) { + expect(rollupDaily![i].day).toBe(liveDaily[i].day); + // round2 of the bucket mean == ROUND(AVG, 2) of the same rows. + expect(rollupDaily![i].value).toBeCloseTo(liveDaily[i].value, 5); + } + } + }); + + it("reflects a freshly written measurement on the next read after a warm cache", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { + username: "rollup-freshness-user", + email: "rollup-freshness@example.test", + role: "USER", + }, + }); + + const now = Date.now(); + const dayMs = 24 * 60 * 60 * 1000; + await prisma.measurement.createMany({ + data: [ + { + userId: user.id, + type: "WEIGHT", + value: 80, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - 5 * dayMs), + }, + { + userId: user.id, + type: "WEIGHT", + value: 81, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - 4 * dayMs), + }, + ], + }); + + // Warm-up read populates the rollups via `ensureUserRollupsFresh`. + const before = await buildComprehensiveAggregate(user.id); + expect(before.summaries.WEIGHT.count).toBe(2); + expect(before.summaries.WEIGHT.mean).toBe(80.5); + + // Insert a third row and synchronously refresh the DAY bucket the + // way the write-path hooks do. + const newMeasuredAt = new Date(now - 1 * dayMs); + await prisma.measurement.create({ + data: { + userId: user.id, + type: "WEIGHT", + value: 84, + unit: "kg", + source: "MANUAL", + measuredAt: newMeasuredAt, + }, + }); + await recomputeBucketsForMeasurement(user.id, "WEIGHT", newMeasuredAt); + + // Next read reflects the third measurement — the rollup-derived + // count moves from 2 to 3 and the weighted mean tracks live SQL. + const after = await buildComprehensiveAggregate(user.id); + expect(after.summaries.WEIGHT.count).toBe(3); + // Weighted mean across the 3 buckets (each count=1): (80+81+84)/3 + // = 81.6667 → round2 = 81.67. + expect(after.summaries.WEIGHT.mean).toBe(81.67); + }); +}); 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",