mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 04:50:59 +00:00
v1.4.35 — persistent measurement rollups + partial read-swap
Foundation release for the persistent measurement_rollups cache tier. Adds the additive table + enum, populates it from every measurement write hook, lands the Apple-Health-import end-of-job backfill, and ships the parity-checked read-swap on two reader surfaces. The comprehensive insights aggregator and the slim analytics summaries slice now source count / min / max / mean per type from the DAY rollup buckets via aggregateBuckets, plus the comprehensive dailyByType correlation feed straight from the bucket means. Slope / R² / standard deviation / anomaly counts stay on live SQL — they do not compose across DAY buckets. A defensive parity check against the live aggregate's COUNT(*) falls back to live SQL when divergence is detected. Includes the new scripts/backfill-rollups.ts CLI for the one-shot fold of historical measurements into all four granularities. Tests: 4280 unit + 228 integration. tsc + lint clean. Migration 0067_v1434_measurement_rollups is idempotent under re-run.
This commit is contained in:
Marc-André Bombeck committed
1 parent
9715831f82
commit
8dc9edb65e
26 files changed
+2650
-65
No files matched your search
@@ -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(() => {
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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<typeof vi.fn>;
|
||||
const ROLLUP_FIND_MANY =
|
||||
prisma.measurementRollup.findMany as unknown as ReturnType<typeof vi.fn>;
|
||||
const ROLLUP_FIND_FIRST =
|
||||
prisma.measurementRollup.findFirst as unknown as ReturnType<typeof vi.fn>;
|
||||
|
||||
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");
|
||||
|
||||
@@ -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<SummariesSlice> {
|
||||
// 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<AggregateRow[]>`
|
||||
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<string, number>();
|
||||
@@ -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),
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
@@ -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<typeof vi.fn>;
|
||||
const FIND_MANY =
|
||||
prisma.measurement.findMany as unknown as ReturnType<typeof vi.fn>;
|
||||
const ROLLUP_FIND_MANY =
|
||||
prisma.measurementRollup.findMany as unknown as ReturnType<typeof vi.fn>;
|
||||
const ROLLUP_FIND_FIRST =
|
||||
prisma.measurementRollup.findFirst as unknown as ReturnType<typeof vi.fn>;
|
||||
|
||||
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 },
|
||||
|
||||
@@ -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<ComprehensiveAggregate> {
|
||||
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<AggregateRow[]>`
|
||||
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<DailyAggregateRow[]>`
|
||||
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<string, DataSummary> = {};
|
||||
@@ -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<string, Array<{ day: string; value: number }>> = {};
|
||||
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 {
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
Reference in new issue
Block a user