mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 04:50:59 +00:00
feat(rollups): MeasurementRollup foundation — populator + write hooks + warm-on-read
WIP toward v1.4.35. Lands the persistent rollup foundation from the
v1.4.34 cache-aggregation research §9. Reads still served by the
v1.4.34.1 live-SQL aggregator — the parity-safe read-swap
(count/min/max/mean + dailyByType from buckets, slope/r2/sd stay on
live aggregation) ships in a follow-up commit so v1.4.35 is the
single release that flips both foundation + read-swap.
Schema:
- new MeasurementRollup model with composite PK
(userId, type, granularity, bucketStart) and a descending
bucketStart index. RollupGranularity enum (DAY/WEEK/MONTH/YEAR).
- Prisma migration 0067_v1434_measurement_rollups is additive:
CREATE TYPE / CREATE TABLE IF NOT EXISTS / CREATE INDEX IF NOT
EXISTS / ADD CONSTRAINT FK. No ALTER on existing tables.
- db-compat schema-bootstrap path mirrored.
Populator (src/lib/measurements/rollups.ts):
- recomputeBucketsForMeasurement: sync DAY; WEEK/MONTH/YEAR via
pg-boss (singleton-keyed).
- recomputeUserRollups: bounded full-range, used by Apple Health
import + backfill.
- ensureUserRollupsFresh: cheap stale-detection + bounded DAY-90d
recompute. Error-swallowing.
- SQL via $queryRawUnsafe + strict enum whitelist for date_trunc
unit (Postgres 42803 forbids binding the unit literal).
Write hooks on every measurement-mutating endpoint; each wrapped
in try/catch so a populator hiccup never fails the write.
Apple Health import worker recomputes the full range once at
completion (per-row hooks would multiply cost).
Backfill (scripts/backfill-rollups.ts): idempotent, single-user
serial, single-digit minutes for a 100k-row account.
Tests: +14 unit, +3 integration. setup.ts.truncateAllTables
extends to the new table.
Read-swap follow-up:
- comprehensive-aggregator + analytics/route + summaries-slice
each call ensureUserRollupsFresh as a read side-effect (warms
the table). The response stays on the live-SQL aggregator so
byte-shape parity with v1.4.34.5 is preserved.
- Next commit lands the partial read-swap: count/min/max/mean
and dailyByType from buckets (math-identical compose), while
slope/r2/sd stay on indexed live aggregation. Together they
ship as v1.4.35.
Pre-shipping state — do NOT tag v1.4.35 from this commit alone.
This commit is contained in:
Marc-André Bombeck committed
1 parent
2982ed1f0b
commit
9872081d0e
19 files changed
+1821
-4
No files matched your search
@@ -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;
|
||||
@@ -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
|
||||
|
||||
@@ -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 <userId>
|
||||
*
|
||||
* 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<void> {
|
||||
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();
|
||||
});
|
||||
@@ -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<ReturnType<typeof requireAuth>>["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
|
||||
|
||||
@@ -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 });
|
||||
},
|
||||
);
|
||||
@@ -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<Response> {
|
||||
// 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({
|
||||
|
||||
@@ -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<Response> {
|
||||
|
||||
// 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<Response> {
|
||||
// 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 });
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
@@ -43,6 +43,7 @@ import { prisma } from "@/lib/db";
|
||||
import type { DataSummary } from "@/lib/analytics/trends";
|
||||
import { measurementTypeEnum } from "@/lib/validations/measurement";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { ensureUserRollupsFresh } from "@/lib/measurements/rollups";
|
||||
|
||||
interface AggregateRow {
|
||||
type: string;
|
||||
@@ -136,6 +137,15 @@ export interface SummariesSlice {
|
||||
export async function computeSummariesSlice(
|
||||
userId: string,
|
||||
): Promise<SummariesSlice> {
|
||||
// v1.5.0 — keep the persistent rollup table current as a side
|
||||
// effect of the dashboard tile strip mount. No-op when fresh; one
|
||||
// bounded DAY-granularity recompute on first cold-mount after a
|
||||
// process restart. The slim slice's response shape is untouched —
|
||||
// the rollup table accelerates downstream consumers (the v1.5.1
|
||||
// read-path swap will route this surface through the rollup once
|
||||
// dual-read parity is established).
|
||||
await ensureUserRollupsFresh(userId);
|
||||
|
||||
// The two passes run in parallel — they target the same index path
|
||||
// (`measurements (user_id, type, measured_at)`) but Postgres can
|
||||
// serve them off the buffer cache concurrently.
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
@@ -61,6 +61,7 @@
|
||||
*/
|
||||
import { prisma } from "@/lib/db";
|
||||
import type { DataSummary } from "@/lib/analytics/trends";
|
||||
import { ensureUserRollupsFresh } from "@/lib/measurements/rollups";
|
||||
|
||||
/**
|
||||
* Raw row from the aggregate query. Keyed on `MeasurementType`, with
|
||||
@@ -170,6 +171,18 @@ export async function buildComprehensiveAggregate(
|
||||
): Promise<ComprehensiveAggregate> {
|
||||
const ninetyDaysAgo = new Date(Date.now() - 90 * 24 * 60 * 60 * 1000);
|
||||
|
||||
// v1.5.0 — keep the persistent rollup table current for downstream
|
||||
// consumers (the analytics tile strip, the dashboard widgets, the
|
||||
// weekly report). The helper is a no-op when rollups are already
|
||||
// ahead of the newest measurement; on first-mount after a cold
|
||||
// restart it folds the trailing 90-day window into the DAY rollup
|
||||
// table. The current aggregator output keeps its byte-shape via the
|
||||
// live SQL queries below — the rollup table accelerates downstream
|
||||
// reads, not the current response. A v1.5.1 follow-up will switch
|
||||
// the per-type DataSummary lookups to read from the rollup table
|
||||
// after the dual-read divergence gate clears.
|
||||
await ensureUserRollupsFresh(userId);
|
||||
|
||||
const [aggregates, latests, dailyRows, sysRaw, diaRaw] = await Promise.all([
|
||||
prisma.$queryRaw<AggregateRow[]>`
|
||||
WITH window_stats AS (
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<RollupRecomputePayload>(
|
||||
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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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<DailyMeanRow[]> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
@@ -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<RollupGranularity, string> = {
|
||||
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<void> {
|
||||
// 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<void> {
|
||||
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<boolean> {
|
||||
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(<granularity>, 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<RollupRow[]> {
|
||||
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<RollupRow[]>(
|
||||
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<RollupRow[]>(
|
||||
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<number> {
|
||||
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<string, { type: MeasurementType; measuredAt: Date }>();
|
||||
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());
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
/**
|
||||
* v1.5.0 — integration coverage for the persistent measurement-
|
||||
* rollup populator.
|
||||
*
|
||||
* Runs against the testcontainer Postgres so the real `STDDEV_POP` /
|
||||
* `REGR_SLOPE` / `REGR_R2` functions and the `date_trunc` semantics
|
||||
* are exercised end-to-end. Three contracts are pinned:
|
||||
*
|
||||
* 1. **Synchronous DAY recompute on write hook.** Calling
|
||||
* `recomputeBucketsForMeasurement` after a measurement insert
|
||||
* writes a `MeasurementRollup` row with the right
|
||||
* `count / mean / min / max / sd` shape.
|
||||
*
|
||||
* 2. **Backfill cardinality.** `recomputeUserRollups` over a
|
||||
* synthetic 3-user × 4-type × 50-row fixture produces the
|
||||
* expected (user × type × granularity × bucket) cardinality
|
||||
* with no duplicate rows.
|
||||
*
|
||||
* 3. **Re-aggregated DataSummary matches live SQL.** A populated
|
||||
* rollup-DAY series re-aggregated through `aggregateBuckets`
|
||||
* returns `count / min / max / mean` byte-identical to a
|
||||
* parallel live `$queryRaw` over the same rows. Establishes
|
||||
* the byte-shape parity contract before the v1.5.1 read-path
|
||||
* swap.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
import {
|
||||
recomputeBucketsForMeasurement,
|
||||
recomputeUserRollups,
|
||||
ALL_GRANULARITIES,
|
||||
} from "@/lib/measurements/rollups";
|
||||
import { aggregateBuckets } from "@/lib/measurements/rollup-read";
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: () => null, // no pg-boss in the test container
|
||||
}));
|
||||
|
||||
beforeEach(async () => {
|
||||
await truncateAllTables(getPrismaClient());
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("measurement rollups — integration", () => {
|
||||
it("writes a DAY rollup row that matches the inserted measurements", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const user = await prisma.user.create({
|
||||
data: {
|
||||
username: "rollup-day-user",
|
||||
email: "rollup-day-user@example.test",
|
||||
role: "USER",
|
||||
},
|
||||
});
|
||||
const measuredAt = new Date("2026-05-10T10:00:00.000Z");
|
||||
await prisma.measurement.createMany({
|
||||
data: [
|
||||
{
|
||||
userId: user.id,
|
||||
type: "WEIGHT",
|
||||
value: 80,
|
||||
unit: "kg",
|
||||
source: "MANUAL",
|
||||
measuredAt: new Date("2026-05-10T07:00:00.000Z"),
|
||||
},
|
||||
{
|
||||
userId: user.id,
|
||||
type: "WEIGHT",
|
||||
value: 82,
|
||||
unit: "kg",
|
||||
source: "MANUAL",
|
||||
measuredAt: new Date("2026-05-10T15:00:00.000Z"),
|
||||
},
|
||||
{
|
||||
userId: user.id,
|
||||
type: "WEIGHT",
|
||||
value: 84,
|
||||
unit: "kg",
|
||||
source: "MANUAL",
|
||||
measuredAt: new Date("2026-05-10T22:00:00.000Z"),
|
||||
},
|
||||
],
|
||||
});
|
||||
|
||||
await recomputeBucketsForMeasurement(user.id, "WEIGHT", measuredAt);
|
||||
|
||||
const dayRow = await prisma.measurementRollup.findFirst({
|
||||
where: { userId: user.id, type: "WEIGHT", granularity: "DAY" },
|
||||
});
|
||||
expect(dayRow).not.toBeNull();
|
||||
expect(dayRow!.count).toBe(3);
|
||||
expect(dayRow!.mean).toBeCloseTo(82, 5);
|
||||
expect(dayRow!.minValue).toBe(80);
|
||||
expect(dayRow!.maxValue).toBe(84);
|
||||
// STDDEV_POP across [80, 82, 84]: variance = (4+0+4)/3 ≈ 2.667;
|
||||
// SD = sqrt ≈ 1.633.
|
||||
expect(dayRow!.sd).toBeCloseTo(1.633, 2);
|
||||
});
|
||||
|
||||
it("backfills with the expected (user × type × granularity × bucket) cardinality", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const userIds: string[] = [];
|
||||
const types = ["WEIGHT", "PULSE", "BLOOD_PRESSURE_SYS", "BLOOD_PRESSURE_DIA"] as const;
|
||||
|
||||
// 3 users × 4 types × ~50 rows over ~7 weeks
|
||||
for (let u = 0; u < 3; u++) {
|
||||
const user = await prisma.user.create({
|
||||
data: {
|
||||
username: `rollup-backfill-user-${u}`,
|
||||
email: `rollup-backfill-${u}@example.test`,
|
||||
role: "USER",
|
||||
},
|
||||
});
|
||||
userIds.push(user.id);
|
||||
interface SeedRow {
|
||||
userId: string;
|
||||
type: (typeof types)[number];
|
||||
value: number;
|
||||
unit: string;
|
||||
source: "MANUAL";
|
||||
measuredAt: Date;
|
||||
}
|
||||
const rows: SeedRow[] = [];
|
||||
const baseMs = new Date("2026-03-01T08:00:00.000Z").getTime();
|
||||
for (const type of types) {
|
||||
for (let i = 0; i < 50; i++) {
|
||||
rows.push({
|
||||
userId: user.id,
|
||||
type,
|
||||
value: 80 + (i % 5),
|
||||
unit: type === "WEIGHT" ? "kg" : type === "PULSE" ? "bpm" : "mmHg",
|
||||
source: "MANUAL",
|
||||
measuredAt: new Date(baseMs + i * 24 * 60 * 60 * 1000),
|
||||
});
|
||||
}
|
||||
}
|
||||
await prisma.measurement.createMany({ data: rows });
|
||||
}
|
||||
|
||||
for (const userId of userIds) {
|
||||
await recomputeUserRollups(userId, {
|
||||
granularities: ALL_GRANULARITIES,
|
||||
});
|
||||
}
|
||||
|
||||
// 50 distinct days per (user, type) → 50 DAY rollups.
|
||||
const dayCount = await prisma.measurementRollup.count({
|
||||
where: { granularity: "DAY" },
|
||||
});
|
||||
expect(dayCount).toBe(3 * 4 * 50);
|
||||
|
||||
// WEEK / MONTH / YEAR cardinalities should be > 0 and < dayCount.
|
||||
const weekCount = await prisma.measurementRollup.count({
|
||||
where: { granularity: "WEEK" },
|
||||
});
|
||||
const monthCount = await prisma.measurementRollup.count({
|
||||
where: { granularity: "MONTH" },
|
||||
});
|
||||
const yearCount = await prisma.measurementRollup.count({
|
||||
where: { granularity: "YEAR" },
|
||||
});
|
||||
expect(weekCount).toBeGreaterThan(0);
|
||||
expect(weekCount).toBeLessThan(dayCount);
|
||||
expect(monthCount).toBeGreaterThan(0);
|
||||
expect(monthCount).toBeLessThan(weekCount);
|
||||
expect(yearCount).toBeGreaterThan(0);
|
||||
expect(yearCount).toBeLessThanOrEqual(monthCount);
|
||||
|
||||
// Re-running the backfill is idempotent — totals stay the same.
|
||||
for (const userId of userIds) {
|
||||
await recomputeUserRollups(userId, {
|
||||
granularities: ALL_GRANULARITIES,
|
||||
});
|
||||
}
|
||||
const dayCountAfter = await prisma.measurementRollup.count({
|
||||
where: { granularity: "DAY" },
|
||||
});
|
||||
expect(dayCountAfter).toBe(dayCount);
|
||||
});
|
||||
|
||||
it("re-aggregated DAY buckets match live SQL for count / min / max / mean (byte-shape parity)", async () => {
|
||||
const prisma = getPrismaClient();
|
||||
const user = await prisma.user.create({
|
||||
data: {
|
||||
username: "rollup-parity-user",
|
||||
email: "rollup-parity@example.test",
|
||||
role: "USER",
|
||||
},
|
||||
});
|
||||
|
||||
// 14-day descending-weight fixture spread across multiple days
|
||||
// so the DAY rollup has > 1 buckets to aggregate.
|
||||
const baseTime = new Date("2026-05-01T08:00:00.000Z").getTime();
|
||||
const N = 28;
|
||||
const rows = Array.from({ length: N }, (_, i) => ({
|
||||
userId: user.id,
|
||||
type: "WEIGHT" as const,
|
||||
value: 80 + (i * 4) / (N - 1), // 80 → 84
|
||||
unit: "kg",
|
||||
source: "MANUAL" as const,
|
||||
measuredAt: new Date(baseTime + i * 12 * 60 * 60 * 1000), // 2/day
|
||||
}));
|
||||
await prisma.measurement.createMany({ data: rows });
|
||||
|
||||
// Backfill rollups.
|
||||
await recomputeUserRollups(user.id, { granularities: ["DAY"] });
|
||||
|
||||
const buckets = await prisma.measurementRollup.findMany({
|
||||
where: { userId: user.id, type: "WEIGHT", granularity: "DAY" },
|
||||
orderBy: { bucketStart: "asc" },
|
||||
});
|
||||
|
||||
// Re-aggregate the DAY buckets through the rollup-read helper.
|
||||
const reAggregated = aggregateBuckets(
|
||||
buckets.map((b) => ({
|
||||
day: b.bucketStart,
|
||||
count: b.count,
|
||||
mean: b.mean,
|
||||
minValue: b.minValue,
|
||||
maxValue: b.maxValue,
|
||||
})),
|
||||
);
|
||||
|
||||
// Compare against a parallel live $queryRaw — the source of truth
|
||||
// the v1.4.34.5 comprehensive aggregator uses.
|
||||
const live = await prisma.$queryRaw<
|
||||
Array<{
|
||||
count: bigint;
|
||||
min_value: number;
|
||||
max_value: number;
|
||||
mean_value: number;
|
||||
}>
|
||||
>`
|
||||
SELECT
|
||||
COUNT(*) AS count,
|
||||
MIN(m."value")::double precision AS min_value,
|
||||
MAX(m."value")::double precision AS max_value,
|
||||
AVG(m."value")::double precision AS mean_value
|
||||
FROM measurements m
|
||||
WHERE m."user_id" = ${user.id}
|
||||
AND m."type" = 'WEIGHT'
|
||||
`;
|
||||
|
||||
expect(reAggregated.count).toBe(Number(live[0].count));
|
||||
expect(reAggregated.min).toBeCloseTo(live[0].min_value, 5);
|
||||
expect(reAggregated.max).toBeCloseTo(live[0].max_value, 5);
|
||||
expect(reAggregated.mean).toBeCloseTo(live[0].mean_value, 5);
|
||||
});
|
||||
});
|
||||
@@ -49,6 +49,7 @@ export async function truncateAllTables(client: PrismaClient): Promise<void> {
|
||||
"host_metrics",
|
||||
"idempotency_keys",
|
||||
"integration_statuses",
|
||||
"measurement_rollups",
|
||||
"measurements",
|
||||
"medication_intake_events",
|
||||
"medication_schedules",
|
||||
|
||||
Reference in new issue
Block a user