From 1375d20a9d82d5bc99a32c4826973d036dd7e58f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 12 Jul 2026 12:07:41 +0200 Subject: [PATCH] =?UTF-8?q?chore(release):=20v1.28.31=20=E2=80=94=20heart-?= =?UTF-8?q?rate=20history=20keeps=20its=20daily=20shape?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Long-term retention for the dense intra-day tier (PULSE, HRV, SpO2) folds to per-local-hour means instead of a single daily mean: the last 14 days keep every raw sample, older days keep up to 24 points per metric — the day's shape survives permanently at ~9k rows/metric/year. Hourly rows key as stats::T anchored at local HH:30, adopt-in-place against both unique indexes with the P2002 retry outside the transaction. The pre-fold DAY-rollup recompute extends to HRV (mean fidelity, not only min/max) and the post-fold recompute is gone for the whole tier. Derived resting rows for proxy users still mint from the day's raw samples. A one-shot rebuild reconstructs the hourly shape for every already-folded day from the soft-deleted raw samples and retires the daily row in the same transaction — at no instant are both grains live, so no AVG-over-live-rows reader can double-count. Boot discovery pairs live daily-grain rows with surviving tombstones (parameter-bound SQL, pinned against real Postgres) and converges to zero once converted; days whose tombstones were pruned keep their daily mean. PULSE/SpO2 DAY buckets keep the true pre-fold stats; HRV DAY buckets recompute from the hourly rows (strictly better than the collapsed fold-time bucket); WEEK/MONTH/YEAR respans enqueue whole-bucket recomputes. --- CHANGELOG.md | 16 + docs/api/openapi.yaml | 2 +- package.json | 2 +- public/sw.js | 2 +- .../stress-strain-retention-queue.test.ts | 42 ++ src/lib/jobs/dense-intraday-hourly-rebuild.ts | 225 +++++++ src/lib/jobs/dense-intraday-retention.ts | 11 +- src/lib/jobs/reminder/register-rollup.ts | 88 ++- .../dense-intraday-hourly-fold.test.ts | 162 +++++ .../dense-intraday-hourly-rebuild.test.ts | 459 ++++++++++++++ .../dense-intraday-pulse-summary.test.ts | 14 +- .../dense-intraday-retention.test.ts | 148 ++++- .../dense-intraday-spo2-summary.test.ts | 51 +- src/lib/measurements/consolidation-base.ts | 10 + src/lib/measurements/consolidation-tz.ts | 66 +++ .../dense-intraday-hourly-rebuild.ts | 367 ++++++++++++ .../measurements/dense-intraday-retention.ts | 561 ++++++++++-------- src/lib/rollups/measurement-rollups.ts | 7 +- .../dense-intraday-hourly-rebuild.test.ts | 365 ++++++++++++ .../dense-intraday-retention.test.ts | 212 +++---- 20 files changed, 2397 insertions(+), 413 deletions(-) create mode 100644 src/lib/jobs/dense-intraday-hourly-rebuild.ts create mode 100644 src/lib/measurements/__tests__/dense-intraday-hourly-fold.test.ts create mode 100644 src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts create mode 100644 src/lib/measurements/dense-intraday-hourly-rebuild.ts create mode 100644 tests/integration/dense-intraday-hourly-rebuild.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 47886ea99..be10733a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,22 @@ ## [Unreleased] +## [1.28.31] — 2026-07-12 — Heart-rate history keeps its daily shape + +Long-term retention for the dense intra-day signals (heart rate, HRV, blood +oxygen) now folds to hourly means instead of a single daily mean. The last +14 days keep every raw sample as before; older days keep up to 24 points — +enough to see how a day actually went, at roughly nine thousand rows per +metric per year instead of an unbounded raw stream. + +Because the old fold soft-deleted rather than erased, a one-time rebuild +reconstructs the hourly shape for every already-folded day from the retained +raw samples, then retires the old daily row in the same transaction — no +reader can ever double-count a day. The rebuild runs automatically once per +installation after the update; days whose raw samples are no longer +available keep their daily mean. Daily min/max/mean statistics captured at +fold time stay untouched throughout. + ## [1.28.30] — 2026-07-12 — The daily briefing stops silently skipping days The recurring "no briefing today" had a chain of causes, now closed end to end. diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 0bd7fed38..f9ca2e624 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.28.30 + version: 1.28.31 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index da490eb20..8fc6fa2cd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.28.30", + "version": "1.28.31", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index 903ab1775..d13f2e178 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.28.30"; + /* @sw-version-fallback */ "v1.28.31"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts b/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts index a529fa6fb..839c2b575 100644 --- a/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts +++ b/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts @@ -148,3 +148,45 @@ describe("reminder-worker — dense intra-day retention wiring", () => { expect(source).toMatch(/derivedRestingRowsUpserted/); }); }); + +describe("reminder-worker — dense hourly-rebuild wiring", () => { + it("imports the queue symbols from the dense-intraday-hourly-rebuild module", () => { + expect(source).toMatch( + /from\s*["']@\/lib\/jobs\/dense-intraday-hourly-rebuild["']/, + ); + expect(source).toMatch(/\bDENSE_INTRADAY_HOURLY_REBUILD_QUEUE\b/); + expect(source).toMatch(/\brunDenseIntradayHourlyRebuildForUser\b/); + expect(source).toMatch(/\benqueueBootTimeDenseIntradayHourlyRebuild\b/); + }); + + it("registers the hourly-rebuild queue in allQueues", () => { + expect(allQueuesBlock()).toMatch(/\bDENSE_INTRADAY_HOURLY_REBUILD_QUEUE\b/); + }); + + it("registers a boss.work handler against the hourly-rebuild queue", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,250}DENSE_INTRADAY_HOURLY_REBUILD_QUEUE[\s\S]{0,500}runDenseIntradayHourlyRebuildForUser/, + ); + }); + + it("fires the boot-discovery enqueue helper", () => { + expect(source).toMatch( + /await enqueueBootTimeDenseIntradayHourlyRebuild\(\)/, + ); + }); + + it("defers the boot-discovery rebuild past the startup storm (P2028 guard)", () => { + const REBUILD_PATH = join( + __dirname, + "..", + "dense-intraday-hourly-rebuild.ts", + ); + const rebuildSource = readFileSync(REBUILD_PATH, "utf8"); + expect(rebuildSource).toMatch( + /DENSE_INTRADAY_HOURLY_REBUILD_BOOT_DELAY_SECONDS\s*=\s*\d+/, + ); + expect(rebuildSource).toMatch( + /startAfter:\s*DENSE_INTRADAY_HOURLY_REBUILD_BOOT_DELAY_SECONDS/, + ); + }); +}); diff --git a/src/lib/jobs/dense-intraday-hourly-rebuild.ts b/src/lib/jobs/dense-intraday-hourly-rebuild.ts new file mode 100644 index 000000000..9f451d577 --- /dev/null +++ b/src/lib/jobs/dense-intraday-hourly-rebuild.ts @@ -0,0 +1,225 @@ +/** + * v1.28.31 — pg-boss queue + boot-time one-shot discovery for the dense-tier + * hourly history rebuild (`runDenseIntradayHourlyRebuild`). + * + * Modelled on the self-converging boot backfills (`lab-biomarker-backfill`, + * `dense-intraday-retention`): a discovery query enqueues one job per user + * that still holds rebuildable days, and the pass is idempotent across + * reboots. The durable marker is the DATA, not a schema column: a rebuilt + * day's daily `stats:` row is tombstoned in the same transaction as its + * hourly rows, so the "live daily row PAIRED with tombstoned raw rows" + * predicate below stops matching it — once every rebuildable day is + * converted, the discovery returns zero users on every later boot and the + * rebuild has run exactly once per install. + * + * The pairing in the discovery is deliberately coarse (±14 h around the + * daily row's local-noon anchor covers the whole local day for every zone); + * the per-user handler re-derives the exact tz-correct day window and skips + * days whose tombstones actually belong to a neighbour. A residual + * false-positive user costs one cheap no-op job per boot and disappears at + * the latest when the tombstone-retention prune removes the neighbouring + * raw tombstones. + * + * The queue name MUST be registered in `allQueues` in + * `src/lib/jobs/reminder/register-rollup.ts` so pg-boss provisions it at + * boot; an unregistered queue silently never drains. + */ +import { Prisma, type PrismaClient } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { DENSE_INTRADAY_RETENTION_ENABLED } from "@/lib/jobs/dense-intraday-retention"; +import { + DENSE_INTRADAY_RETENTION_DAYS, + DENSE_INTRADAY_RETENTION_TYPES, +} from "@/lib/measurements/dense-intraday-retention"; +import { runDenseIntradayHourlyRebuild } from "@/lib/measurements/dense-intraday-hourly-rebuild"; + +export const DENSE_INTRADAY_HOURLY_REBUILD_QUEUE = + "dense-intraday-hourly-rebuild"; + +/** + * Serial concurrency — a rebuild walks every folded day for one account + * with a transaction per day; concurrency-1 keeps it from crowding the + * request pool, matching the retention drain's convention. + */ +export const DENSE_INTRADAY_HOURLY_REBUILD_CONCURRENCY = 1; + +/** + * Boot-discovery jobs start only after this delay (seconds). The rebuild + * shares the retention drain's P2028 reasoning (transaction-per-day work + * must never contend with the deploy storm for the shared pool) and is + * staged AFTER the retention drain's 600 s defer so the two dense-tier + * walks never overlap at boot. + */ +export const DENSE_INTRADAY_HOURLY_REBUILD_BOOT_DELAY_SECONDS = 900; + +export interface DenseIntradayHourlyRebuildPayload { + userId: string; + enqueuedAt: string; +} + +/** + * Per-user queue handler. Runs the hourly history rebuild for one account + * and returns the summary totals so the worker can log them. + */ +export async function runDenseIntradayHourlyRebuildForUser( + userId: string, +): Promise<{ + daysRebuilt: number; + hourlyRowsUpserted: number; + dailyRowsRetired: number; + daysSkippedNoTombstones: number; +}> { + // Shares the retention drain's kill-switch: when an operator disables the + // dense tier, no-op so any already-queued backlog drains cleanly. + if (!DENSE_INTRADAY_RETENTION_ENABLED) { + return { + daysRebuilt: 0, + hourlyRowsUpserted: 0, + dailyRowsRetired: 0, + daysSkippedNoTombstones: 0, + }; + } + const summary = await runDenseIntradayHourlyRebuild(prisma, { + userId, + log: () => { + // Silent inside the queue handler — the worker logs the totals. + }, + }); + annotate({ + action: { + name: "retention.hourly_rebuild.complete", + details: { + days_rebuilt: summary.totals.daysRebuilt, + hourly_rows_upserted: summary.totals.hourlyRowsUpserted, + daily_rows_retired: summary.totals.dailyRowsRetired, + days_skipped_no_tombstones: summary.totals.daysSkippedNoTombstones, + days_failed: summary.totals.daysFailed, + }, + }, + }); + return { + daysRebuilt: summary.totals.daysRebuilt, + hourlyRowsUpserted: summary.totals.hourlyRowsUpserted, + dailyRowsRetired: summary.totals.dailyRowsRetired, + daysSkippedNoTombstones: summary.totals.daysSkippedNoTombstones, + }; +} + +/** + * Discovery query for the boot enqueue: every user holding at least one + * live daily-grain dense-tier `stats:` row older than the retention window + * that is PAIRED with tombstoned raw rows of the same type within ±14 h of + * the daily row's anchor (the coarse whole-local-day pairing; the per-user + * handler re-derives the exact tz-correct window). + * + * The daily-grain shape is asserted with a constant regex — hourly rows + * (`…T` suffix) and the iOS hourly-HR wire rows (ISO-instant suffix) + * never match, so a converted day can never re-qualify. Prisma cannot + * express the regex or the pairing, hence the parameter-bound `$queryRaw`; + * the only spliced text is the constant pattern. Exported so the + * integration suite pins the SQL against a real Postgres — a malformed + * query here would surface only as a swallowed boot-discovery error and + * the rebuild would silently never run. + */ +export async function findHourlyRebuildCandidateUserIds( + prismaClient: PrismaClient, + windowStart: Date, +): Promise { + const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES).map(String); + if (types.length === 0) return []; + + const users = await prismaClient.$queryRaw>` + SELECT DISTINCT m."user_id" + FROM "measurements" m + WHERE m."source" = 'APPLE_HEALTH' + AND m."type"::text IN (${Prisma.join(types)}) + AND m."deleted_at" IS NULL + AND m."external_id" ~ '^stats:[A-Za-z0-9]+:[0-9]{4}-[0-9]{2}-[0-9]{2}$' + AND m."measured_at" < ${windowStart} + AND EXISTS ( + SELECT 1 + FROM "measurements" t + WHERE t."user_id" = m."user_id" + AND t."type" = m."type" + AND t."source" = 'APPLE_HEALTH' + AND t."deleted_at" IS NOT NULL + AND t."external_id" NOT LIKE 'stats:%' + AND t."measured_at" >= m."measured_at" - interval '14 hours' + AND t."measured_at" < m."measured_at" + interval '14 hours' + ) + `; + return users.map((u) => u.user_id); +} + +/** + * Boot-time discovery. Runs `findHourlyRebuildCandidateUserIds` and + * enqueues one rebuild job per account. + * + * Idempotent across reboots: rebuilding a day tombstones its daily row, + * which removes the pair. pg-boss `singletonKey` coalesces duplicate + * sends. Best-effort: errors are returned through the result value so + * worker boot never fails because of a rebuild miss. + */ +export async function enqueueBootTimeDenseIntradayHourlyRebuild(): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + if (!DENSE_INTRADAY_RETENTION_ENABLED) { + return { enqueued: 0, skipped: 0, error: null }; + } + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + const windowStart = new Date( + Date.now() - DENSE_INTRADAY_RETENTION_DAYS * 24 * 60 * 60 * 1000, + ); + const userIds = await findHourlyRebuildCandidateUserIds( + prisma, + windowStart, + ); + + if (userIds.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const userId of userIds) { + const payload: DenseIntradayHourlyRebuildPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send( + DENSE_INTRADAY_HOURLY_REBUILD_QUEUE, + payload, + { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + // Defer past the boot storm AND past the retention drain's own + // 600 s defer so the two dense-tier walks never overlap at boot. + startAfter: DENSE_INTRADAY_HOURLY_REBUILD_BOOT_DELAY_SECONDS, + singletonKey: `dense-intraday-hourly-rebuild|${userId}`, + }, + ); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index 61c1d0202..bd86101b7 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/src/lib/jobs/dense-intraday-retention.ts @@ -9,10 +9,10 @@ * so they drop off the discovery list). * * Unlike the daily-mean consolidation, this pass scopes to the dense-tier - * types (`HEART_RATE_VARIABILITY`, `PULSE`) and keeps the last - * `DENSE_INTRADAY_RETENTION_DAYS` of raw per-sample rows so the Stress - * engine still has its intra-day SDNN shape; only out-of-window samples - * fold to a daily mean. + * types (`HEART_RATE_VARIABILITY`, `PULSE`, `OXYGEN_SATURATION`) and keeps + * the last `DENSE_INTRADAY_RETENTION_DAYS` of raw per-sample rows so the + * Stress engine still has its intra-day SDNN shape; only out-of-window + * samples fold — to hourly means since v1.28.31. * * The queue name MUST be registered in `allQueues` in * `src/lib/jobs/reminder-worker.ts` so pg-boss provisions it at boot; an @@ -115,7 +115,8 @@ export async function runDenseIntradayRetentionForUser( details: { days: summary.totals.daysConsolidated, per_sample_rows_soft_deleted: summary.totals.perSampleRowsSoftDeleted, - daily_rows_upserted: summary.totals.dailyRowsUpserted, + hourly_rows_upserted: summary.totals.hourlyRowsUpserted, + daily_rows_retired: summary.totals.dailyRowsRetired, // iOS#34 — derived RESTING_HEART_RATE rows minted from folded PULSE // days for proxy users, preserving the resting signal post-fold. derived_resting_rows_upserted: diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index c38d8e41d..2affd68f8 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -76,6 +76,13 @@ import { type DenseIntradayRetentionPayload, } from "@/lib/jobs/dense-intraday-retention"; import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; +import { + DENSE_INTRADAY_HOURLY_REBUILD_QUEUE, + DENSE_INTRADAY_HOURLY_REBUILD_CONCURRENCY, + runDenseIntradayHourlyRebuildForUser, + enqueueBootTimeDenseIntradayHourlyRebuild, + type DenseIntradayHourlyRebuildPayload, +} from "@/lib/jobs/dense-intraday-hourly-rebuild"; import { getWorkerPrisma, workerLog, @@ -123,6 +130,10 @@ const allQueues = [ // driven like mean-consolidation; the steady-state nightly walk folds // onto the drain-cumulative tick). DENSE_INTRADAY_RETENTION_QUEUE, + // v1.28.31 — one-shot hourly history rebuild for pre-hourly folded + // dense-tier days. Boot-discovery driven and self-converging (a rebuilt + // day's retired daily row drops it from the discovery pairing). + DENSE_INTRADAY_HOURLY_REBUILD_QUEUE, // v1.4.37 W7c — explicit createQueue is required before the nightly // schedule below registers (pg-boss v12 contract). Without this entry the // drain schedule silently no-ops and the per-sample APPLE_HEALTH rows @@ -231,10 +242,11 @@ export async function registerRollupQueues( // v1.10.0 WX-E — dense intra-day retention per-user backfill worker. The // boot enqueue helper below sends one job per user holding live per-sample - // dense-tier (HRV / HR) rows older than the retention window; this handler - // folds those out-of-window samples to a daily mean and soft-deletes the - // originals, keeping the in-window intra-day shape intact for the Stress - // engine. Serial concurrency so the backfill never crowds the request pool. + // dense-tier (HRV / HR / SpO2) rows older than the retention window; this + // handler folds those out-of-window samples to hourly means (v1.28.31) and + // soft-deletes the originals, keeping the in-window intra-day shape intact + // for the Stress engine. Serial concurrency so the backfill never crowds + // the request pool. await boss.work( DENSE_INTRADAY_RETENTION_QUEUE, { localConcurrency: DENSE_INTRADAY_RETENTION_CONCURRENCY }, @@ -264,6 +276,43 @@ export async function registerRollupQueues( }, ); + // v1.28.31 — one-shot hourly history rebuild worker. The boot enqueue + // helper below sends one job per user still holding a pre-hourly daily + // `stats:` row paired with tombstoned raw rows; this handler reconstructs + // hourly means from the tombstones and retires the daily row atomically. + // Self-converging: a rebuilt day drops out of the discovery pairing, so + // the pass runs once per install. Serial concurrency so the rebuild never + // crowds the request pool. + await boss.work( + DENSE_INTRADAY_HOURLY_REBUILD_QUEUE, + { localConcurrency: DENSE_INTRADAY_HOURLY_REBUILD_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + try { + const { + daysRebuilt, + hourlyRowsUpserted, + dailyRowsRetired, + daysSkippedNoTombstones, + } = await runDenseIntradayHourlyRebuildForUser(userId); + workerLog( + "info", + `[dense-intraday-hourly-rebuild] user=${userId} daysRebuilt=${daysRebuilt} hourlyRowsUpserted=${hourlyRowsUpserted} dailyRowsRetired=${dailyRowsRetired} daysSkippedNoTombstones=${daysSkippedNoTombstones}`, + ); + } catch (err) { + recordError(); + workerLog( + "error", + `[dense-intraday-hourly-rebuild] user=${userId} failed`, + err, + ); + throw err; + } + } + }, + ); + // v1.4.39 W-MOOD — mood-rollup per-bucket worker. Folds the // WEEK / MONTH / YEAR buckets that the mood-entry write hooks // enqueue; the DAY bucket runs synchronously in the hook itself. @@ -429,7 +478,7 @@ export async function registerRollupQueues( ); workerLog( "info", - `[dense-intraday-retention] triggeredAt=${job.data.triggeredAt} usersScanned=${denseSummary.totals.usersScanned} daysConsolidated=${denseSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${denseSummary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${denseSummary.totals.dailyRowsUpserted}`, + `[dense-intraday-retention] triggeredAt=${job.data.triggeredAt} usersScanned=${denseSummary.totals.usersScanned} daysConsolidated=${denseSummary.totals.daysConsolidated} perSampleRowsSoftDeleted=${denseSummary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${denseSummary.totals.hourlyRowsUpserted} dailyRowsRetired=${denseSummary.totals.dailyRowsRetired}`, ); } } catch (err) { @@ -578,6 +627,35 @@ export async function enqueueRollupBootDiscovery(): Promise { ); } + // v1.28.31 — one-shot hourly history rebuild. Finds every user still + // holding a pre-hourly daily dense-tier `stats:` row paired with tombstoned + // raw rows and enqueues one rebuild job per account. Rebuilt days retire + // their daily row, so the pairing predicate converges to zero across + // boots — the retired row is the durable once-per-install marker. + try { + // Self-defers past both the boot window and the retention drain's own + // 600 s stage via its larger constant; no stagger argument threaded. + const { enqueued, skipped, error } = + await enqueueBootTimeDenseIntradayHourlyRebuild(); + if (error) { + workerLog( + "error", + `[dense-intraday-hourly-rebuild] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[dense-intraday-hourly-rebuild] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[dense-intraday-hourly-rebuild] boot discovery threw an unexpected error", + err, + ); + } + // v1.4.39 W-MOOD — mood rollup tier. Mirrors the v1.4.35.1 // measurement-rollup pattern: one job per user with mood entries but no // rollup coverage. Idempotent across reboots and singleton-keyed inside diff --git a/src/lib/measurements/__tests__/dense-intraday-hourly-fold.test.ts b/src/lib/measurements/__tests__/dense-intraday-hourly-fold.test.ts new file mode 100644 index 000000000..2c9281eb9 --- /dev/null +++ b/src/lib/measurements/__tests__/dense-intraday-hourly-fold.test.ts @@ -0,0 +1,162 @@ +/** + * v1.28.31 — hourly-grain primitives of the dense intra-day retention tier. + * + * Pins the pure helpers the hourly fold and the history rebuild share: the + * per-LOCAL-hour bucketing (tz-correct, DST-safe on both transition days), + * the `stats::T` externalId shape (the `stats:` prefix + * is load-bearing for the scan's re-fold exclusion), and the local-HH:30 + * anchor instant (mirrors the local-noon daily convention one grain down). + */ +import { describe, it, expect } from "vitest"; + +import { + bucketRowsByLocalHour, + hourlyStatsExternalId, +} from "../dense-intraday-retention"; +import { + canonicalHourlyTimestamp, + dayKeyForUserTz, + hourOfDayForUserTz, +} from "../consolidation-tz"; +import type { PerSampleRow } from "../consolidation-tz"; +import type { MeasurementType } from "@/generated/prisma/client"; + +function row(id: string, iso: string, value = 50): PerSampleRow { + return { + id, + type: "HEART_RATE_VARIABILITY" as MeasurementType, + value, + measuredAt: new Date(iso), + externalId: null, + unit: "ms", + }; +} + +describe("hourlyStatsExternalId", () => { + it("emits stats::T with a zero-padded hour", () => { + expect( + hourlyStatsExternalId( + "HKQuantityTypeIdentifierHeartRate", + "2026-05-01", + 7, + ), + ).toBe("stats:HKQuantityTypeIdentifierHeartRate:2026-05-01T07"); + expect( + hourlyStatsExternalId( + "HKQuantityTypeIdentifierHeartRate", + "2026-05-01", + 23, + ), + ).toBe("stats:HKQuantityTypeIdentifierHeartRate:2026-05-01T23"); + }); + + it("stays under the stats: prefix (the scan's re-fold exclusion)", () => { + const id = hourlyStatsExternalId("X", "2026-05-01", 0); + expect(id.startsWith("stats:")).toBe(true); + }); + + it("never collides with the daily externalId of the same day", () => { + // The daily id is `stats:X:2026-05-01`; every hourly id appends T. + for (let hour = 0; hour < 24; hour++) { + expect(hourlyStatsExternalId("X", "2026-05-01", hour)).not.toBe( + "stats:X:2026-05-01", + ); + } + }); +}); + +describe("bucketRowsByLocalHour — timezone correctness", () => { + it("buckets by the USER-LOCAL hour, not the UTC hour", () => { + // Berlin is UTC+2 on 2026-05-01: 08:15Z and 08:45Z are local hour 10. + const buckets = bucketRowsByLocalHour( + [ + row("a", "2026-05-01T08:15:00.000Z"), + row("b", "2026-05-01T08:45:00.000Z"), + row("c", "2026-05-01T09:15:00.000Z"), // local hour 11 + ], + "Europe/Berlin", + ); + expect([...buckets.keys()].sort((x, y) => x - y)).toEqual([10, 11]); + expect(buckets.get(10)).toHaveLength(2); + expect(buckets.get(11)).toHaveLength(1); + }); + + it("west-of-UTC zones bucket into their own local hours", () => { + // New York is UTC-4 on 2026-05-01: 02:30Z is local hour 22 (prev day). + const buckets = bucketRowsByLocalHour( + [row("a", "2026-05-01T02:30:00.000Z")], + "America/New_York", + ); + expect([...buckets.keys()]).toEqual([22]); + }); + + it("spring-forward day: the skipped local hour never appears", () => { + // Europe/Berlin 2026-03-29: 02:00 CET jumps to 03:00 CEST — local hour + // 2 does not exist. 00:30Z = 01:30 CET (hour 1); 01:30Z = 03:30 CEST + // (hour 3). + const buckets = bucketRowsByLocalHour( + [ + row("a", "2026-03-29T00:30:00.000Z"), + row("b", "2026-03-29T01:30:00.000Z"), + ], + "Europe/Berlin", + ); + expect([...buckets.keys()].sort((x, y) => x - y)).toEqual([1, 3]); + }); + + it("fall-back day: both instants of the repeated local hour share ONE bucket", () => { + // Europe/Berlin 2026-10-25: 03:00 CEST falls back to 02:00 CET — the + // 02:xx wall-clock hour happens twice. 00:30Z = 02:30 CEST; 01:30Z = + // 02:30 CET. One bucket, two rows. + const buckets = bucketRowsByLocalHour( + [ + row("a", "2026-10-25T00:30:00.000Z"), + row("b", "2026-10-25T01:30:00.000Z"), + ], + "Europe/Berlin", + ); + expect([...buckets.keys()]).toEqual([2]); + expect(buckets.get(2)).toHaveLength(2); + }); +}); + +describe("canonicalHourlyTimestamp — local HH:30 anchor", () => { + it("anchors at the middle of the local hour", () => { + // Berlin UTC+2: local 09:30 on 2026-05-01 is 07:30Z. + expect( + canonicalHourlyTimestamp("2026-05-01", 9, "Europe/Berlin").toISOString(), + ).toBe("2026-05-01T07:30:00.000Z"); + // UTC: identity. + expect(canonicalHourlyTimestamp("2026-05-01", 9, "UTC").toISOString()).toBe( + "2026-05-01T09:30:00.000Z", + ); + }); + + it("round-trips back to its own (day, hour) for regular days", () => { + for (const tz of ["Europe/Berlin", "America/New_York", "Asia/Kathmandu"]) { + for (const hour of [0, 6, 12, 23]) { + const anchor = canonicalHourlyTimestamp("2026-05-01", hour, tz); + expect(dayKeyForUserTz(anchor, tz)).toBe("2026-05-01"); + expect(hourOfDayForUserTz(anchor, tz)).toBe(hour); + } + } + }); + + it("resolves the ambiguous fall-back hour to an instant INSIDE that local hour", () => { + // 2026-10-25, Europe/Berlin: local 02:30 exists twice. Whichever + // instant is picked must read back as (2026-10-25, hour 2) — that is + // all the fold needs, the externalId is the row identity. + const anchor = canonicalHourlyTimestamp("2026-10-25", 2, "Europe/Berlin"); + expect(dayKeyForUserTz(anchor, "Europe/Berlin")).toBe("2026-10-25"); + expect(hourOfDayForUserTz(anchor, "Europe/Berlin")).toBe(2); + }); + + it("produces 24 DISTINCT anchors per regular day (index-B identities cannot collide)", () => { + const anchors = new Set( + Array.from({ length: 24 }, (_, hour) => + canonicalHourlyTimestamp("2026-05-01", hour, "Europe/Berlin").getTime(), + ), + ); + expect(anchors.size).toBe(24); + }); +}); diff --git a/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts b/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts new file mode 100644 index 000000000..de394991c --- /dev/null +++ b/src/lib/measurements/__tests__/dense-intraday-hourly-rebuild.test.ts @@ -0,0 +1,459 @@ +/** + * v1.28.31 — one-shot hourly history rebuild for pre-hourly folded + * dense-tier days. + * + * Pins the rebuild contract: a folded day (live daily `stats:` row + + * tombstoned raw rows) is converted to hourly means AND the daily row is + * retired in the SAME transaction (no instant where both grains are live — + * the double-count invariant); days with zero tombstoned raws keep their + * daily row; the day scan is bounded by the retention window; a re-run + * converges to zero work; non-APPLE_HEALTH sources are never touched; the + * DAY-rollup handling is per type (HRV recomputed from the hourly rows, + * PULSE/SpO2 untouched with only whole-span WEEK/MONTH/YEAR enqueues). + */ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +const { recomputeBucketsForMeasurement, enqueueRollupRecompute, bucketSpan } = + vi.hoisted(() => ({ + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + enqueueRollupRecompute: vi.fn().mockResolvedValue(undefined), + bucketSpan: vi.fn((measuredAt: Date) => ({ + from: measuredAt, + to: measuredAt, + })), + })); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + recomputeBucketsForMeasurement, + enqueueRollupRecompute, + bucketSpan, +})); + +import { runDenseIntradayHourlyRebuild } from "../dense-intraday-hourly-rebuild"; +import type { PrismaClient } from "@/generated/prisma/client"; + +const HRV_DAILY_ID = + "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01"; +const PULSE_DAILY_ID = "stats:HKQuantityTypeIdentifierHeartRate:2026-05-01"; + +interface MockRow { + id: string; + externalId: string | null; + type?: string; + value?: number; + measuredAt?: Date; + unit?: string; +} + +function tombstone( + id: string, + value: number, + iso: string, + type: string, + unit = "ms", +): MockRow { + return { + id, + type, + value, + measuredAt: new Date(iso), + externalId: `hk-${id}`, + unit, + }; +} + +/** + * Prisma mock for the rebuild flow. `findMany` discriminates the two scans + * by their predicates: the daily-candidate scan filters live rows + * (`deletedAt: null` + `externalId.startsWith`), the tombstone scan filters + * `deletedAt: { not: null }`. + */ +function buildPrismaMock(opts: { + dailyByType: Record; + tombstonesByType: Record; +}) { + const txCreate = vi.fn().mockResolvedValue({ id: "minted-hourly" }); + const txUpdate = vi.fn().mockResolvedValue({}); + const txFindFirst = vi.fn().mockResolvedValue(null); + const tx = { + measurement: { + create: txCreate, + update: txUpdate, + findFirst: txFindFirst, + }, + }; + + const findMany = vi.fn( + async (args: { + where: { + type: string; + deletedAt?: null | { not: null }; + externalId?: { startsWith: string }; + }; + }) => { + if (args.where.deletedAt === null) { + return opts.dailyByType[args.where.type] ?? []; + } + return opts.tombstonesByType[args.where.type] ?? []; + }, + ); + const topUpdate = vi.fn().mockResolvedValue({}); + const $transaction = vi.fn(async (cb: (t: unknown) => Promise) => + cb(tx), + ); + + const mock = { + user: { + findMany: vi + .fn() + .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]), + }, + measurement: { findMany, update: topUpdate }, + $transaction, + } as unknown as PrismaClient; + + return { + mock, + findMany, + txCreate, + txUpdate, + txFindFirst, + topUpdate, + $transaction, + }; +} + +beforeEach(() => { + recomputeBucketsForMeasurement.mockClear(); + enqueueRollupRecompute.mockClear(); + bucketSpan.mockClear(); +}); + +describe("runDenseIntradayHourlyRebuild — rebuild flow", () => { + it("mints hourly means from the tombstoned raws and retires the daily row in the SAME transaction", async () => { + const { mock, txCreate, txUpdate, topUpdate, $transaction } = + buildPrismaMock({ + dailyByType: { + HEART_RATE_VARIABILITY: [{ id: "daily-1", externalId: HRV_DAILY_ID }], + }, + tombstonesByType: { + HEART_RATE_VARIABILITY: [ + tombstone( + "a", + 40, + "2026-05-01T08:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + tombstone( + "b", + 60, + "2026-05-01T09:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + ], + }, + }); + + const summary = await runDenseIntradayHourlyRebuild(mock, { + log: () => {}, + }); + + // Hourly rows: Berlin UTC+2 → local hours 10 and 11, each its own mean. + expect(txCreate).toHaveBeenCalledTimes(2); + const createArgs = txCreate.mock.calls.map( + (c) => (c[0] as { data: { externalId: string; value: number } }).data, + ); + expect(createArgs.map((d) => d.externalId)).toEqual([ + `${HRV_DAILY_ID}T10`, + `${HRV_DAILY_ID}T11`, + ]); + expect(createArgs.map((d) => d.value)).toEqual([40, 60]); + + // The daily row is tombstoned via the TRANSACTION client — never the + // top-level client — inside the single per-day transaction, so no + // reader ever sees the daily and hourly rows live at once. + expect($transaction).toHaveBeenCalledTimes(1); + expect(topUpdate).not.toHaveBeenCalled(); + const retire = txUpdate.mock.calls.find( + (c) => (c[0] as { where: { id: string } }).where.id === "daily-1", + ); + expect(retire).toBeDefined(); + expect( + (retire?.[0] as { data: { deletedAt: Date } }).data.deletedAt, + ).toBeInstanceOf(Date); + + expect(summary.totals.daysRebuilt).toBe(1); + expect(summary.totals.hourlyRowsUpserted).toBe(2); + expect(summary.totals.dailyRowsRetired).toBe(1); + expect(summary.totals.daysSkippedNoTombstones).toBe(0); + }); + + it("HRV: recomputes the DAY bucket from the now-live hourly rows after the rebuild", async () => { + const { mock } = buildPrismaMock({ + dailyByType: { + HEART_RATE_VARIABILITY: [{ id: "daily-1", externalId: HRV_DAILY_ID }], + }, + tombstonesByType: { + HEART_RATE_VARIABILITY: [ + tombstone( + "a", + 40, + "2026-05-01T08:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + ], + }, + }); + + await runDenseIntradayHourlyRebuild(mock, { log: () => {} }); + + // Historical HRV DAY buckets were collapsed to min == max == mean at + // fold time; hourly-derived stats are strictly better. + expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(1); + expect(recomputeBucketsForMeasurement).toHaveBeenCalledWith( + "user-1", + "HEART_RATE_VARIABILITY", + expect.any(Date), + ); + // The DAY recompute call owns the WEEK/MONTH/YEAR enqueue for HRV. + expect(enqueueRollupRecompute).not.toHaveBeenCalled(); + }); + + it("PULSE: leaves the DAY bucket untouched (true pre-fold min/max/mean) and enqueues WEEK/MONTH/YEAR only", async () => { + const { mock } = buildPrismaMock({ + dailyByType: { + PULSE: [{ id: "daily-p", externalId: PULSE_DAILY_ID }], + }, + tombstonesByType: { + PULSE: [tombstone("a", 60, "2026-05-01T08:00:00.000Z", "PULSE", "bpm")], + }, + }); + + await runDenseIntradayHourlyRebuild(mock, { log: () => {} }); + + expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled(); + const granularities = enqueueRollupRecompute.mock.calls + .map((c) => (c[0] as { granularity: string }).granularity) + .sort(); + expect(granularities).toEqual(["MONTH", "WEEK", "YEAR"]); + }); + + it("skips days with ZERO tombstoned raws — the daily row is the only surviving representation", async () => { + const { mock, $transaction, txUpdate, topUpdate } = buildPrismaMock({ + dailyByType: { + HEART_RATE_VARIABILITY: [{ id: "daily-1", externalId: HRV_DAILY_ID }], + }, + tombstonesByType: {}, + }); + + const summary = await runDenseIntradayHourlyRebuild(mock, { + log: () => {}, + }); + + expect($transaction).not.toHaveBeenCalled(); + expect(txUpdate).not.toHaveBeenCalled(); + expect(topUpdate).not.toHaveBeenCalled(); + expect(summary.totals.daysRebuilt).toBe(0); + expect(summary.totals.dailyRowsRetired).toBe(0); + expect(summary.totals.daysSkippedNoTombstones).toBe(1); + }); + + it("never considers days inside the retention window (bounded candidate scan)", async () => { + const { mock, findMany } = buildPrismaMock({ + dailyByType: {}, + tombstonesByType: {}, + }); + + await runDenseIntradayHourlyRebuild(mock, { log: () => {} }); + + // Every daily-candidate scan carries the window bound. + const candidateCalls = findMany.mock.calls.filter( + (c) => (c[0] as { where: { deletedAt?: null } }).where.deletedAt === null, + ); + expect(candidateCalls.length).toBeGreaterThan(0); + for (const call of candidateCalls) { + const where = (call[0] as { where: { measuredAt?: { lt: Date } } }).where; + expect(where.measuredAt?.lt).toBeInstanceOf(Date); + } + }); + + it("scopes every scan to APPLE_HEALTH and excludes stats: rows from the tombstone read", async () => { + const { mock, findMany } = buildPrismaMock({ + dailyByType: { + HEART_RATE_VARIABILITY: [{ id: "daily-1", externalId: HRV_DAILY_ID }], + }, + tombstonesByType: { + HEART_RATE_VARIABILITY: [ + tombstone( + "a", + 40, + "2026-05-01T08:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + ], + }, + }); + + await runDenseIntradayHourlyRebuild(mock, { log: () => {} }); + + for (const call of findMany.mock.calls) { + const where = (call[0] as unknown as { where: { source: string } }).where; + expect(where.source).toBe("APPLE_HEALTH"); + } + // The tombstone scan must never feed a `stats:` row (e.g. a retired + // daily row of a neighbouring day) into an hourly mean. + const tombstoneCall = findMany.mock.calls.find( + (c) => + (c[0] as { where: { deletedAt?: { not: null } } }).where.deletedAt !== + null, + ); + expect( + ( + tombstoneCall?.[0] as unknown as { + where: { NOT: { externalId: { startsWith: string } } }; + } + ).where.NOT.externalId.startsWith, + ).toBe("stats:"); + }); + + it("ignores hourly-shaped and iOS wire-shaped stats externalIds in the candidate set", async () => { + const { mock, $transaction } = buildPrismaMock({ + dailyByType: { + PULSE: [ + // Already-hourly row (this tier's own output) — not a candidate. + { id: "h", externalId: `${PULSE_DAILY_ID}T10` }, + // iOS hourly-HR wire row (ISO-instant suffix) — not a candidate. + { + id: "ios", + externalId: + "stats:HKQuantityTypeIdentifierHeartRate:2026-05-01T14:00:00.000Z", + }, + ], + }, + tombstonesByType: { + PULSE: [tombstone("a", 60, "2026-05-01T08:00:00.000Z", "PULSE", "bpm")], + }, + }); + + const summary = await runDenseIntradayHourlyRebuild(mock, { + log: () => {}, + }); + + expect($transaction).not.toHaveBeenCalled(); + expect(summary.totals.daysRebuilt).toBe(0); + expect(summary.totals.daysSkippedNoTombstones).toBe(0); + }); + + it("converges: a run with no live daily rows does zero work", async () => { + const { mock, $transaction } = buildPrismaMock({ + dailyByType: {}, + tombstonesByType: { + HEART_RATE_VARIABILITY: [ + tombstone( + "a", + 40, + "2026-05-01T08:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + ], + }, + }); + + const summary = await runDenseIntradayHourlyRebuild(mock, { + log: () => {}, + }); + + expect($transaction).not.toHaveBeenCalled(); + expect(summary.totals.daysRebuilt).toBe(0); + expect(summary.totals.hourlyRowsUpserted).toBe(0); + expect(summary.totals.dailyRowsRetired).toBe(0); + }); + + it("writes nothing on a dry-run", async () => { + const { mock, $transaction, txCreate, txUpdate, topUpdate } = + buildPrismaMock({ + dailyByType: { + HEART_RATE_VARIABILITY: [{ id: "daily-1", externalId: HRV_DAILY_ID }], + }, + tombstonesByType: { + HEART_RATE_VARIABILITY: [ + tombstone( + "a", + 40, + "2026-05-01T08:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + ], + }, + }); + + const summary = await runDenseIntradayHourlyRebuild(mock, { + dryRun: true, + log: () => {}, + }); + + expect($transaction).not.toHaveBeenCalled(); + expect(txCreate).not.toHaveBeenCalled(); + expect(txUpdate).not.toHaveBeenCalled(); + expect(topUpdate).not.toHaveBeenCalled(); + expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled(); + expect(enqueueRollupRecompute).not.toHaveBeenCalled(); + // The preview still reports what a real run would do. + expect(summary.totals.daysRebuilt).toBe(1); + expect(summary.totals.hourlyRowsUpserted).toBe(1); + }); + + it("per-day failure boundary: a poisoned day is stepped over and the rest still rebuilds", async () => { + const { mock, $transaction } = buildPrismaMock({ + dailyByType: { + HEART_RATE_VARIABILITY: [ + { id: "daily-1", externalId: HRV_DAILY_ID }, + { + id: "daily-2", + externalId: + "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-02", + }, + ], + }, + tombstonesByType: { + HEART_RATE_VARIABILITY: [ + tombstone( + "a", + 40, + "2026-05-01T08:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + tombstone( + "b", + 44, + "2026-05-02T08:00:00.000Z", + "HEART_RATE_VARIABILITY", + ), + ], + }, + }); + + // First day's transaction throws a non-P2002 error; the boundary must + // absorb it and continue to the second day. + let call = 0; + ($transaction as ReturnType).mockImplementation( + async (cb: (t: unknown) => Promise) => { + call += 1; + if (call === 1) throw new Error("boom — poisoned day"); + return cb({ + measurement: { + create: vi.fn().mockResolvedValue({ id: "minted" }), + update: vi.fn().mockResolvedValue({}), + findFirst: vi.fn().mockResolvedValue(null), + }, + }); + }, + ); + + const summary = await runDenseIntradayHourlyRebuild(mock, { + log: () => {}, + }); + + expect(summary.totals.daysFailed).toBe(1); + expect(summary.totals.daysRebuilt).toBe(1); + }); +}); diff --git a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts index a2926ef0f..6d7973204 100644 --- a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts @@ -2,8 +2,8 @@ * iOS#34 / #69 — server-side aggregation of high-frequency heart rate. * * The dense intra-day retention drain folds out-of-window per-sample PULSE - * rows to one daily-MEAN `stats:` row to bound the ~16k-rows-per-user bloat. - * The plain mean-only fold would erase the two clinically-meaningful daily + * rows to hourly-MEAN `stats:` rows to bound the ~16k-rows-per-user bloat. + * The mean fold alone would erase the two clinically-meaningful daily * figures the app derives from the raw stream: * * 1. Resting HR — the read-path resolver derives it from the 20th-percentile @@ -146,12 +146,14 @@ describe("PULSE fold — min/max preservation via pre-fold rollup recompute", () await runDenseIntradayRetention(mock, { retentionDays: 0, log: () => {} }); - // The fold minted the daily mean row. - expect(txCreate).toHaveBeenCalledTimes(1); + // The fold minted one hourly mean row per distinct local hour (06:00Z / + // 12:00Z / 18:00Z land in three different Berlin hours). + expect(txCreate).toHaveBeenCalledTimes(3); // PULSE recompute happens EXACTLY once — the pre-fold capture. A second - // (post-fold) PULSE recompute would aggregate the single mean row and - // collapse min == max == mean, erasing the intraday bounds. + // (post-fold) PULSE recompute would aggregate the hourly mean rows and + // degrade min/max to hourly-mean extremes (and the DAY mean to an + // unweighted mean-of-hourly-means), erasing the intraday bounds. const pulseRecomputes = recomputeBucketsForMeasurement.mock.calls.filter( (c) => c[1] === "PULSE", ); diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts index ff246cae0..9f10409db 100644 --- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts @@ -1,11 +1,13 @@ /** * v1.10.0 — computed scores (WX-E). Dense intra-day retention drain. + * v1.28.31 — the fold target is HOURLY means (user-local hours). * * Pins the retention BOUND: the scan only folds per-sample dense-tier rows * OLDER than the retention window (`measuredAt < now - retentionDays`), so * the recent intra-day shape the Stress engine reads is never collapsed. - * Also pins the dense-tier scope (HRV + PULSE), the APPLE_HEALTH source - * scope, the MEAN reduction + soft-delete, and the rollup recompute. + * Also pins the dense-tier scope (HRV + PULSE + SpO2), the APPLE_HEALTH + * source scope, the per-local-hour MEAN reduction + soft-delete, the + * hourly `stats:` externalId shape, and the pre-fold rollup recompute. */ import { describe, it, expect, vi, beforeEach } from "vitest"; @@ -37,12 +39,12 @@ function row( function buildPrismaMock( rowsByType: Record, - // When set, the canonical-row lookup inside `writeDay` resolves to this - // existing row id — simulating a daily row already sitting on the - // canonical local-noon instant (the P2002 coexistence case). + // When set, every canonical-slot lookup inside `writeDay` resolves to + // this existing row id — simulating a row already sitting on the target + // externalId / anchor instant (the P2002 coexistence case). existingCanonicalId: string | null = null, ) { - const create = vi.fn().mockResolvedValue({ id: "minted-daily" }); + const create = vi.fn().mockResolvedValue({ id: "minted-hourly" }); const update = vi.fn().mockResolvedValue({}); const findFirst = vi .fn() @@ -134,9 +136,10 @@ describe("runDenseIntradayRetention — retention bound", () => { }); }); -describe("runDenseIntradayRetention — fold flow", () => { - it("creates the day MEAN and soft-deletes the out-of-window rows when no canonical row exists", async () => { - // Two HRV samples on an out-of-window day. +describe("runDenseIntradayRetention — hourly fold flow", () => { + it("creates one MEAN row per LOCAL hour and soft-deletes the out-of-window rows", async () => { + // Two HRV samples in DIFFERENT local hours (Berlin is UTC+2 on this + // date: 08:00Z → 10:xx local, 09:00Z → 11:xx local). const hrvRows = [ row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"), row("b", 60, "2026-05-01T09:00:00.000Z", "HEART_RATE_VARIABILITY"), @@ -150,15 +153,39 @@ describe("runDenseIntradayRetention — fold flow", () => { log: () => {}, }); - // No pre-existing canonical row → create, not update. + // No pre-existing canonical rows → create per hour, never update. expect(update).not.toHaveBeenCalled(); - const createArg = create.mock.calls[0]?.[0] as { - data: { value: number; source: string; type: string }; - }; - // value is the MEAN (50), not the sum (100). - expect(createArg.data.value).toBeCloseTo(50, 6); - expect(createArg.data.source).toBe("APPLE_HEALTH"); - expect(createArg.data.type).toBe("HEART_RATE_VARIABILITY"); + expect(create).toHaveBeenCalledTimes(2); + const createArgs = create.mock.calls.map( + (c) => + ( + c[0] as { + data: { + value: number; + source: string; + type: string; + externalId: string; + measuredAt: Date; + }; + } + ).data, + ); + // Each local hour carries ITS OWN mean — the intraday shape survives at + // hour resolution (a single daily mean of 50 would erase it). + expect(createArgs.map((d) => d.value)).toEqual([40, 60]); + expect(createArgs.map((d) => d.externalId)).toEqual([ + "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01T10", + "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01T11", + ]); + // Anchored mid-hour: local HH:30 (Berlin UTC+2 → 08:30Z / 09:30Z). + expect(createArgs.map((d) => d.measuredAt.toISOString())).toEqual([ + "2026-05-01T08:30:00.000Z", + "2026-05-01T09:30:00.000Z", + ]); + for (const d of createArgs) { + expect(d.source).toBe("APPLE_HEALTH"); + expect(d.type).toBe("HEART_RATE_VARIABILITY"); + } // soft-delete, never hard delete. const updArg = updateMany.mock.calls[0]?.[0] as { @@ -166,15 +193,37 @@ describe("runDenseIntradayRetention — fold flow", () => { }; expect(updArg.data.deletedAt).toBeInstanceOf(Date); expect(summary.totals.daysConsolidated).toBe(1); - expect(recomputeBucketsForMeasurement).toHaveBeenCalledTimes(1); + expect(summary.totals.hourlyRowsUpserted).toBe(2); + }); + + it("folds same-hour samples into ONE hourly mean row", async () => { + const hrvRows = [ + row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"), + row("b", 60, "2026-05-01T08:40:00.000Z", "HEART_RATE_VARIABILITY"), + ]; + const { mock, create } = buildPrismaMock({ + HEART_RATE_VARIABILITY: hrvRows, + }); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + expect(create).toHaveBeenCalledTimes(1); + const createArg = create.mock.calls[0]?.[0] as { + data: { value: number }; + }; + // MEAN of (40, 60) = 50 for the shared local hour. + expect(createArg.data.value).toBeCloseTo(50, 6); + expect(summary.totals.hourlyRowsUpserted).toBe(1); }); it("adopts an existing canonical row in place instead of colliding (P2002 coexistence)", async () => { const hrvRows = [ row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"), - row("b", 60, "2026-05-01T09:00:00.000Z", "HEART_RATE_VARIABILITY"), ]; - // A daily row already sits on the canonical local-noon instant. + // A row already sits on the hourly slot (externalId or anchor). const { mock, create, update, updateMany } = buildPrismaMock( { HEART_RATE_VARIABILITY: hrvRows }, "existing-canonical-row", @@ -183,23 +232,68 @@ describe("runDenseIntradayRetention — fold flow", () => { await runDenseIntradayRetention(mock, { retentionDays: 0, log: () => {} }); // The fold adopts the existing row in place — no create (which would - // collide on the measured_at unique index), an update with the stats - // externalId + refreshed mean. + // collide on the measured_at unique index), an update with the hourly + // stats externalId + refreshed mean. expect(create).not.toHaveBeenCalled(); const updateArg = update.mock.calls[0]?.[0] as { where: { id: string }; data: { value: number; externalId: string; deletedAt: null }; }; expect(updateArg.where.id).toBe("existing-canonical-row"); - expect(updateArg.data.value).toBeCloseTo(50, 6); - expect(updateArg.data.externalId.startsWith("stats:")).toBe(true); + expect(updateArg.data.value).toBeCloseTo(40, 6); + expect(updateArg.data.externalId).toBe( + "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01T10", + ); expect(updateArg.data.deletedAt).toBeNull(); - // The soft-delete excludes the adopted canonical row id. + // The soft-delete excludes the adopted canonical row ids. const updManyArg = updateMany.mock.calls[0]?.[0] as { - where: { id: { not: string } }; + where: { id: { notIn: string[] } }; }; - expect(updManyArg.where.id.not).toBe("existing-canonical-row"); + expect(updManyArg.where.id.notIn).toContain("existing-canonical-row"); + }); + + it("retires a live pre-hourly DAILY stats row in the same fold transaction", async () => { + const hrvRows = [ + row("a", 40, "2026-05-01T08:00:00.000Z", "HEART_RATE_VARIABILITY"), + ]; + const { mock, update, findFirst, $transaction } = (() => { + const built = buildPrismaMock({ HEART_RATE_VARIABILITY: hrvRows }); + return { + ...built, + $transaction: built.mock.$transaction as ReturnType, + }; + })(); + + // Discriminating lookup mock: the hourly slot lookups miss, the daily + // retirement lookup (keyed by the DAILY `stats:` externalId + live + // filter) hits. + findFirst.mockImplementation( + async (args: { where: { externalId?: string; deletedAt?: null } }) => + args.where.externalId === + "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01" && + args.where.deletedAt === null + ? { id: "daily-legacy-row" } + : null, + ); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + // The daily row is tombstoned inside the (single) fold transaction — + // never a moment where the daily and hourly rows are both live. + expect($transaction).toHaveBeenCalledTimes(1); + const retire = update.mock.calls.find( + (c) => + (c[0] as { where: { id: string } }).where.id === "daily-legacy-row", + ); + expect(retire).toBeDefined(); + expect( + (retire?.[0] as { data: { deletedAt: Date } }).data.deletedAt, + ).toBeInstanceOf(Date); + expect(summary.totals.dailyRowsRetired).toBe(1); }); it("does not write on a dry-run", async () => { diff --git a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts index 5ce32dbb8..cf19ba322 100644 --- a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts @@ -5,16 +5,17 @@ * through the day and overnight; the readings map to `aggregation: "latest"` * and belong to NEITHER `CUMULATIVE_HK_TYPES` nor `HIGH_FREQUENCY_MEAN_TYPES`, * so every sample piled up raw forever. The dense intra-day retention drain - * folds the out-of-window per-sample rows to one daily-MEAN `stats:` row, - * mirroring the PULSE facet's min/max preservation: the daily MIN is the + * folds the out-of-window per-sample rows to hourly-MEAN `stats:` rows, + * mirroring the PULSE facet's fidelity handling: the daily MIN is the * overnight-desaturation nadir — the single most clinically meaningful SpO2 * figure — so the DAY rollup bucket is recomputed from the RAW rows BEFORE the - * fold and the post-fold recompute that would collapse min == max == mean is - * skipped. + * fold and the post-fold recompute that would degrade min/max (and drift the + * mean to an unweighted mean-of-hourly-means) is skipped. * * This suite pins the SpO2 facet: the pre-fold (and only) DAY-rollup recompute, - * the MEAN fold + soft-delete, the no-derived-resting-row guard (resting HR is - * a PULSE-only concern), idempotency, and the per-day failure boundary. + * the hourly MEAN fold + soft-delete, the no-derived-resting-row guard + * (resting HR is a PULSE-only concern), idempotency, and the per-day failure + * boundary. */ import { describe, it, expect, vi, beforeEach } from "vitest"; @@ -86,9 +87,9 @@ beforeEach(() => { recomputeBucketsForMeasurement.mockClear(); }); -describe("SpO2 fold — MEAN collapse + soft-delete", () => { - it("folds the day to one MEAN row (not the latest, not the min) and soft-deletes the raw rows", async () => { - // Overnight dip to 91, daytime band 96..98. Mean ~95.2, NOT the 91 nadir. +describe("SpO2 fold — hourly MEAN collapse + soft-delete", () => { + it("folds each LOCAL hour to its own MEAN row and soft-deletes the raw rows", async () => { + // Overnight dip to 91, daytime band 96..98 — four distinct local hours. const rows = [ spo2Row("a", 91, "2026-05-01T03:00:00.000Z"), spo2Row("b", 96, "2026-05-01T09:00:00.000Z"), @@ -104,13 +105,23 @@ describe("SpO2 fold — MEAN collapse + soft-delete", () => { log: () => {}, }); - const createArg = txCreate.mock.calls[0]?.[0] as { - data: { value: number; type: string; source: string; unit: string }; - }; - expect(createArg.data.type).toBe("OXYGEN_SATURATION"); - expect(createArg.data.source).toBe("APPLE_HEALTH"); - expect(createArg.data.unit).toBe("%"); - expect(createArg.data.value).toBeCloseTo((91 + 96 + 97 + 98) / 4, 6); + // One hourly row per distinct local hour; the overnight 91 nadir hour + // keeps its own value instead of vanishing into a daily mean of ~95.5. + expect(txCreate).toHaveBeenCalledTimes(4); + const createArgs = txCreate.mock.calls.map( + (c) => + ( + c[0] as { + data: { value: number; type: string; source: string; unit: string }; + } + ).data, + ); + expect(createArgs.map((d) => d.value)).toEqual([91, 96, 97, 98]); + for (const d of createArgs) { + expect(d.type).toBe("OXYGEN_SATURATION"); + expect(d.source).toBe("APPLE_HEALTH"); + expect(d.unit).toBe("%"); + } // Soft-delete, never hard delete. const updArg = txUpdateMany.mock.calls[0]?.[0] as { @@ -119,6 +130,7 @@ describe("SpO2 fold — MEAN collapse + soft-delete", () => { expect(updArg.data.deletedAt).toBeInstanceOf(Date); expect(summary.totals.daysConsolidated).toBe(1); expect(summary.totals.perSampleRowsSoftDeleted).toBe(rows.length); + expect(summary.totals.hourlyRowsUpserted).toBe(4); }); }); @@ -133,11 +145,12 @@ describe("SpO2 fold — min/max preservation (overnight desat nadir)", () => { await runDenseIntradayRetention(mock, { retentionDays: 0, log: () => {} }); - expect(txCreate).toHaveBeenCalledTimes(1); + // Three distinct local hours → three hourly rows. + expect(txCreate).toHaveBeenCalledTimes(3); // EXACTLY one SpO2 recompute — the pre-fold capture. A second (post-fold) - // recompute would aggregate the single mean row and collapse the daily MIN - // (the overnight-desaturation nadir) into the mean. + // recompute would aggregate the hourly mean rows and degrade the daily MIN + // (the overnight-desaturation nadir) toward the hourly means. const spo2Recomputes = recomputeBucketsForMeasurement.mock.calls.filter( (c) => c[1] === "OXYGEN_SATURATION", ); diff --git a/src/lib/measurements/consolidation-base.ts b/src/lib/measurements/consolidation-base.ts index 8142a251b..a94dce1b2 100644 --- a/src/lib/measurements/consolidation-base.ts +++ b/src/lib/measurements/consolidation-base.ts @@ -144,6 +144,14 @@ export interface DayWriteContext { reducedValue: number; dayRows: PerSampleRow[]; sourceRowIds: string[]; + /** + * The user's resolved IANA timezone and the bucket's calendar-day key. + * A drain whose write grain is finer than the day (the dense-tier hourly + * fold) needs both to derive per-hour sub-buckets; the day-grain drains + * ignore them. + */ + tz: string; + dateKey: string; } /** Inputs to one per-user-type-day consolidation pass. */ @@ -426,6 +434,8 @@ export async function runConsolidation( reducedValue, dayRows, sourceRowIds, + tz, + dateKey, shouldMint, }); } diff --git a/src/lib/measurements/consolidation-tz.ts b/src/lib/measurements/consolidation-tz.ts index 41eb8e578..b387505bd 100644 --- a/src/lib/measurements/consolidation-tz.ts +++ b/src/lib/measurements/consolidation-tz.ts @@ -121,6 +121,72 @@ export function canonicalDailyTimestamp(dateKey: string, tz?: string): Date { return new Date(utcNoon.getTime() - offsetMinutes * 60 * 1000); } +/** + * Per-timezone cache for the hour-of-day formatter. The dense-tier hourly + * fold calls `hourOfDayForUserTz` once per scanned sample (thousands of + * rows per heavy user-day); constructing an `Intl.DateTimeFormat` per call + * dominates the bucketing cost, so the formatter is built once per zone. + */ +const hourFormatterByTz = new Map(); + +function hourFormatterFor(tz: string): Intl.DateTimeFormat { + const cached = hourFormatterByTz.get(tz); + if (cached) return cached; + const fmt = new Intl.DateTimeFormat("en-US", { + timeZone: tz, + hour: "2-digit", + hour12: false, + }); + hourFormatterByTz.set(tz, fmt); + return fmt; +} + +/** + * Resolve the user's local hour-of-day (0–23) for a given instant + + * timezone. On a fall-back DST day the repeated wall-clock hour maps both + * of its instants to the same hour value — deliberate: the hourly fold + * buckets by LOCAL hour, so the 25-hour day folds its duplicated hour into + * one bucket rather than minting two rows for the same wall-clock hour. + */ +export function hourOfDayForUserTz(date: Date, tz: string): number { + const part = hourFormatterFor(tz) + .formatToParts(date) + .find((p) => p.type === "hour")?.value; + const hour = part ? Number.parseInt(part, 10) : 0; + // Some ICU builds render midnight as "24" under hourCycle h24. + return hour === 24 ? 0 : hour; +} + +/** + * Compute the canonical anchor instant for a (calendar-day, local-hour) + * slot: the user's local HH:30 — the middle of the local hour, mirroring + * the local-noon convention of `canonicalDailyTimestamp` one grain down. + * Anchoring mid-hour keeps the instant a full 30 minutes inside its hour, + * so it round-trips back to the same (day, hour) through + * `dayKeyForUserTz` + `hourOfDayForUserTz` for every zone. + * + * DST: the offset is read at the first-guess instant and re-read once at + * the candidate — a transition between the two shifts the offset by at + * most one step, so the second read converges for every wall-clock time + * that exists. On a fall-back day the ambiguous repeated HH:30 resolves + * deterministically to one of its two instants (whichever the re-read + * lands on), which is all the fold needs — the hourly externalId, not the + * instant, is the row's identity. + */ +export function canonicalHourlyTimestamp( + dateKey: string, + hour: number, + tz: string, +): Date { + const hh = String(hour).padStart(2, "0"); + const guess = new Date(`${dateKey}T${hh}:30:00.000Z`); + const offsetFirst = tzOffsetMinutesAt(guess, tz); + const candidate = new Date(guess.getTime() - offsetFirst * 60 * 1000); + const offsetAtCandidate = tzOffsetMinutesAt(candidate, tz); + if (offsetAtCandidate === offsetFirst) return candidate; + return new Date(guess.getTime() - offsetAtCandidate * 60 * 1000); +} + /** * v1.4.37 W10 — Compute the JS-Date instant at the user's local 00:00 * for a calendar-day key. Robust on DST transitions because the offset diff --git a/src/lib/measurements/dense-intraday-hourly-rebuild.ts b/src/lib/measurements/dense-intraday-hourly-rebuild.ts new file mode 100644 index 000000000..455e4b613 --- /dev/null +++ b/src/lib/measurements/dense-intraday-hourly-rebuild.ts @@ -0,0 +1,367 @@ +/** + * v1.28.31 — one-shot history rebuild: hourly means for already-folded + * dense-tier days. + * + * Before v1.28.31 the dense intra-day retention drain folded out-of-window + * per-sample APPLE_HEALTH rows (`HEART_RATE_VARIABILITY`, `PULSE`, + * `OXYGEN_SATURATION`) into ONE daily-mean `stats::` row + * per day, erasing the intraday shape. The fold soft-deletes rather than + * hard-deletes, so for every day folded within the tombstone-retention + * horizon the raw samples are still in the table — enough to reconstruct + * the hourly grain the fold now writes. + * + * Scope, per user × dense-tier type × local day OLDER than the retention + * window: a day qualifies only when BOTH + * (a) a live daily-grain `stats::` row exists (the + * pre-hourly fold's output — hourly rows carry a `T` suffix and + * never match the daily shape), AND + * (b) tombstoned non-`stats:` APPLE_HEALTH raw rows exist inside the + * day's local window (the fold's soft-deleted inputs). + * + * Per qualifying day: compute per-local-hour means from the tombstoned raw + * rows (same `meanBucketValue` reducer + `adoptOrMintHourlyRow` + * adopt-in-place slot logic as the live fold, so a re-run converges), then + * tombstone the daily row IN THE SAME TRANSACTION as the hourly mint — at + * no instant are the daily row and its hourly rows both live, so an + * AVG-over-live-rows reader can never double-count the day. + * + * Days with no tombstoned raw rows keep their daily row untouched (their + * tombstones were pruned past the retention horizon — the daily mean is + * the only surviving representation). Days inside the retention window are + * never touched (the daily-row scan is bounded by the window cutoff). + * Non-APPLE_HEALTH sources are never touched (every predicate is + * source-scoped). + * + * DAY-rollup handling after a rebuilt day: + * - `PULSE` + `OXYGEN_SATURATION`: DAY buckets are left untouched — they + * hold the TRUE pre-fold min/max/mean the fold captured from raw rows; + * recomputing from hourly means would degrade them. + * - `HEART_RATE_VARIABILITY`: the pre-v1.28.31 fold recomputed the HRV + * DAY bucket AFTER the fold, collapsing it to min == max == mean of + * the single daily row. Hourly-derived stats are strictly better, so + * the DAY bucket is recomputed from the now-live hourly rows. + * - WEEK/MONTH/YEAR aggregate from live measurement rows (not from DAY + * buckets — see `runRollupAggregate`), and the rebuild changes the + * live row population of the day, so whole-bucket recomputes are + * enqueued for every affected span (via `bucketSpan`, never a partial + * span). + * + * Idempotent / self-converging: retiring the daily row removes the day + * from the (a)+(b) pairing, so a re-run — and the boot-time discovery in + * `src/lib/jobs/dense-intraday-hourly-rebuild.ts` — converges to zero + * work. The retired daily row IS the durable marker; no schema column is + * needed (the `lab-biomarker-backfill` "the data is the marker" pattern). + * + * Runs on pg-boss — the production standalone image strips `tsx`, so this + * can never be a CLI script. + */ +import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import { isP2002 as isUniqueConstraintViolation } from "@/lib/prisma-errors"; + +import { hkIdentifierForType } from "./apple-health-mapping"; +import { + loadConsolidationUsers, + resolveUserTimezone, +} from "./consolidation-base"; +import { + canonicalDailyTimestamp, + canonicalHourlyTimestamp, + localDayWindow, + type PerSampleRow, +} from "./consolidation-tz"; +import { meanBucketValue } from "./consolidate-daily-mean"; +import { + DENSE_INTRADAY_RETENTION_DAYS, + DENSE_INTRADAY_RETENTION_TYPES, + adoptOrMintHourlyRow, + bucketRowsByLocalHour, + hourlyStatsExternalId, +} from "./dense-intraday-retention"; +import { + bucketSpan, + enqueueRollupRecompute, + recomputeBucketsForMeasurement, +} from "@/lib/rollups/measurement-rollups"; + +/** + * The daily-grain suffix a pre-hourly fold row carries after + * `stats::` — exactly a calendar-day key. Hourly rows + * (`…:T`) and the iOS hourly-HR wire rows + * (`…:Z`) both fail this shape, so the rebuild can never + * mistake them for a daily row. + */ +const DAILY_SUFFIX_RE = /^\d{4}-\d{2}-\d{2}$/; + +export interface DenseIntradayHourlyRebuildSummary { + dryRun: boolean; + totals: { + usersScanned: number; + /** Days converted from the daily to the hourly grain. */ + daysRebuilt: number; + /** Hourly `stats:` rows minted / adopted across all rebuilt days. */ + hourlyRowsUpserted: number; + /** Daily `stats:` rows tombstoned (one per rebuilt day). */ + dailyRowsRetired: number; + /** + * Days whose daily row survives because zero tombstoned raw rows + * remain for the day (pruned past the tombstone-retention horizon) — + * nothing to reconstruct from. + */ + daysSkippedNoTombstones: number; + /** Days whose rebuild threw and was stepped over (retried next run). */ + daysFailed: number; + }; +} + +export interface DenseIntradayHourlyRebuildOptions { + /** Limit the pass to a single user. Default = every user. */ + userId?: string; + /** Preview-only mode — no DB writes. */ + dryRun?: boolean; + /** Logger sink — defaults to `console.log`. */ + log?: (line: string) => void; + /** + * Retention window in days — daily rows anchored INSIDE the window are + * never touched. Defaults to `DENSE_INTRADAY_RETENTION_DAYS`; tests can + * pass `0` to lift the bound. + */ + retentionDays?: number; +} + +/** + * Run the hourly history rebuild. See the module header for scope and + * invariants. Does NOT enforce any auth gate — the queue handler owns + * that concern. + */ +export async function runDenseIntradayHourlyRebuild( + prismaClient: PrismaClient, + options: DenseIntradayHourlyRebuildOptions = {}, +): Promise { + const log = options.log ?? ((line) => console.log(line)); + const dryRun = options.dryRun ?? false; + const retentionDays = options.retentionDays ?? DENSE_INTRADAY_RETENTION_DAYS; + const cutoffAt = + retentionDays > 0 + ? new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000) + : null; + + const summary: DenseIntradayHourlyRebuildSummary = { + dryRun, + totals: { + usersScanned: 0, + daysRebuilt: 0, + hourlyRowsUpserted: 0, + dailyRowsRetired: 0, + daysSkippedNoTombstones: 0, + daysFailed: 0, + }, + }; + + const users = await loadConsolidationUsers(prismaClient, options.userId); + summary.totals.usersScanned = users.length; + + for (const user of users) { + const tz = resolveUserTimezone(user.timezone); + + for (const type of DENSE_INTRADAY_RETENTION_TYPES) { + const hkIdentifier = hkIdentifierForType(type); + if (!hkIdentifier) continue; + + // Candidate days: live daily-grain stats rows older than the window. + // The startsWith narrows to this type's stats family; the exact + // daily shape is asserted in JS below (Prisma has no regex filter), + // which also bounds the scan — at most one daily row per folded day. + const statsPrefix = `stats:${hkIdentifier}:`; + const dailyRows = await prismaClient.measurement.findMany({ + where: { + userId: user.id, + type, + source: "APPLE_HEALTH", + deletedAt: null, + externalId: { startsWith: statsPrefix }, + ...(cutoffAt ? { measuredAt: { lt: cutoffAt } } : {}), + }, + select: { id: true, externalId: true }, + orderBy: { measuredAt: "asc" }, + }); + + const candidates: Array<{ id: string; dateKey: string }> = []; + for (const row of dailyRows) { + const suffix = row.externalId?.slice(statsPrefix.length) ?? ""; + if (DAILY_SUFFIX_RE.test(suffix)) { + candidates.push({ id: row.id, dateKey: suffix }); + } + } + if (candidates.length === 0) continue; + + for (const candidate of candidates) { + // Per-day failure boundary: one poisoned day is stepped over so the + // walk keeps rebuilding every other user / type / day; the failed + // day keeps its live daily row, so the discovery re-finds it. + try { + const rebuilt = await rebuildDay(prismaClient, { + userId: user.id, + type, + hkIdentifier, + tz, + dateKey: candidate.dateKey, + dailyRowId: candidate.id, + dryRun, + }); + if (rebuilt === null) { + summary.totals.daysSkippedNoTombstones += 1; + continue; + } + summary.totals.daysRebuilt += 1; + summary.totals.hourlyRowsUpserted += rebuilt.hourlyRows; + summary.totals.dailyRowsRetired += rebuilt.dailyRetired ? 1 : 0; + + if (dryRun) continue; + + const dayAnchor = canonicalDailyTimestamp(candidate.dateKey, tz); + if (type === "HEART_RATE_VARIABILITY") { + // Historical HRV DAY buckets were collapsed to + // min == max == mean at fold time; the now-live hourly rows + // carry strictly better stats. Recompute DAY inline (and let + // the call enqueue the WEEK/MONTH/YEAR spans). + await recomputeBucketsForMeasurement(user.id, type, dayAnchor); + } else { + // PULSE + SpO2: the DAY bucket holds the TRUE pre-fold + // min/max/mean — leave it untouched. Only the WEEK/MONTH/YEAR + // buckets aggregate live rows whose population this rebuild + // changed, so enqueue whole-span recomputes for them. + await Promise.all( + (["WEEK", "MONTH", "YEAR"] as const).map((granularity) => { + const { from, to } = bucketSpan(dayAnchor, granularity); + return enqueueRollupRecompute({ + userId: user.id, + type, + granularity, + from, + to, + }); + }), + ); + } + } catch (err) { + summary.totals.daysFailed += 1; + const reason = err instanceof Error ? err.message : String(err); + log( + `[dense-intraday-hourly-rebuild] user=${user.id} type=${type} day=${candidate.dateKey} failed — ${reason}; continuing with the next day`, + ); + } + } + } + } + + log( + `[dense-intraday-hourly-rebuild] done — usersScanned=${summary.totals.usersScanned} daysRebuilt=${summary.totals.daysRebuilt} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} daysSkippedNoTombstones=${summary.totals.daysSkippedNoTombstones} daysFailed=${summary.totals.daysFailed}${dryRun ? " (dry-run)" : ""}`, + ); + + return summary; +} + +/** + * Rebuild one (user, type, local day): hourly means from the day's + * tombstoned raw rows + retirement of the daily row, atomically. Returns + * `null` when the day has no tombstoned raw rows (skip — the daily row is + * the only surviving representation), otherwise the per-day counts. + */ +async function rebuildDay( + prismaClient: PrismaClient, + input: { + userId: string; + type: MeasurementType; + hkIdentifier: string; + tz: string; + dateKey: string; + dailyRowId: string; + dryRun: boolean; + }, +): Promise<{ hourlyRows: number; dailyRetired: boolean } | null> { + const { dayStart, dayEnd } = localDayWindow(input.dateKey, input.tz); + + // The fold's soft-deleted inputs: tombstoned per-sample rows inside the + // day's local window. Never `stats:` rows (a tombstoned daily row from a + // neighbouring rebuild must not feed the mean), never another source. + const tombstoned = (await prismaClient.measurement.findMany({ + where: { + userId: input.userId, + type: input.type, + source: "APPLE_HEALTH", + deletedAt: { not: null }, + NOT: { externalId: { startsWith: "stats:" } }, + measuredAt: { gte: dayStart, lt: dayEnd }, + }, + select: { + id: true, + type: true, + value: true, + measuredAt: true, + externalId: true, + unit: true, + }, + orderBy: [{ measuredAt: "asc" }, { id: "asc" }], + })) as PerSampleRow[]; + + if (tombstoned.length === 0) return null; + + const unit = tombstoned[0]?.unit ?? "unknown"; + const byHour = [ + ...bucketRowsByLocalHour(tombstoned, input.tz).entries(), + ].sort((a, b) => a[0] - b[0]); + + if (input.dryRun) { + return { hourlyRows: byHour.length, dailyRetired: true }; + } + + const rebuildOnce = async (): Promise<{ dailyRetired: boolean }> => { + let dailyRetired = false; + await prismaClient.$transaction(async (tx) => { + const canonicalRowIds: string[] = []; + for (const [hour, hourRows] of byHour) { + const rowId = await adoptOrMintHourlyRow(tx, { + userId: input.userId, + type: input.type, + externalId: hourlyStatsExternalId( + input.hkIdentifier, + input.dateKey, + hour, + ), + anchor: canonicalHourlyTimestamp(input.dateKey, hour, input.tz), + value: meanBucketValue(hourRows), + unit, + }); + canonicalRowIds.push(rowId); + } + + // Retire the daily row in the SAME transaction — at no instant are + // the daily row and the hourly rows both live (an AVG-over-live-rows + // reader would double-count the day). Guarded against the degenerate + // adopt where the daily row itself became an hourly slot (a + // DST-shifted anchor landing on the daily instant): tombstoning it + // then would erase the rebuild. + if (!canonicalRowIds.includes(input.dailyRowId)) { + await tx.measurement.update({ + where: { id: input.dailyRowId }, + data: { deletedAt: new Date() }, + }); + dailyRetired = true; + } + }); + return { dailyRetired }; + }; + + let outcome: { dailyRetired: boolean }; + try { + outcome = await rebuildOnce(); + } catch (err) { + if (!isUniqueConstraintViolation(err)) throw err; + // A concurrent writer (the nightly fold racing this rebuild) won a slot + // mid-transaction. Retry once: the deterministic lookup inside + // `adoptOrMintHourlyRow` now resolves the winning row and adopts it. + outcome = await rebuildOnce(); + } + + return { hourlyRows: byHour.length, dailyRetired: outcome.dailyRetired }; +} diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index dfbdd6802..5d3b593e4 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -1,5 +1,8 @@ /** - * v1.10.0 — dense intra-day retention tier for daytime HRV / heart-rate. + * v1.10.0 — dense intra-day retention tier for daytime HRV / heart-rate / + * SpO2. v1.28.31 — the out-of-window fold target is HOURLY means (up to 24 + * `stats:` rows per day, user-local hours) instead of one daily mean, so + * the intraday shape of a folded day survives at hour resolution. * * The Stress engine (`src/lib/insights/stress-score.ts`) reads the * intra-day SDNN SHAPE across the day — the spread of a day's @@ -21,26 +24,34 @@ * * 2. A BOUNDED retention window contains the volume the exemption would * otherwise let grow forever. This pass REUSES the proven - * mean-consolidation drain (same `runConsolidation` base, same - * idempotent `stats::` fold, same soft-delete + rollup - * recompute) but with the grace cutoff widened from 36 hours to the - * retention window: per-sample HRV / HR rows OLDER than the window are - * folded to one daily-mean row and the raw rows are tombstoned; rows - * INSIDE the window stay raw so the Stress engine always has its - * intra-day inputs for the days it scores. + * mean-consolidation drain skeleton (same `runConsolidation` base, + * same soft-delete + rollup handling) but with the grace cutoff + * widened from 36 hours to the retention window, and the fold grain + * set to the LOCAL HOUR: per-sample rows OLDER than the window fold + * to one `stats::T` hourly-mean row per user × + * type × local hour and the raw rows are tombstoned; rows INSIDE the + * window stay raw so the Stress engine always has its per-sample + * inputs for the days it scores. * * The net effect: the last `DENSE_INTRADAY_RETENTION_DAYS` carry the dense - * intra-day samples; everything older folds to the same daily-mean grain - * the rest of the high-frequency metrics already use, and the - * `MeasurementRollup` DAY/WEEK/MONTH/YEAR tiers keep serving the long - * history. No new table, no new value column — the retention bound lives - * here in code so tuning it never needs a migration. + * per-sample stream; everything older folds to at most 24 hourly-mean rows + * per day per type (~9k rows/year/type — bounded, vs the unbounded raw + * stream), and the `MeasurementRollup` DAY/WEEK/MONTH/YEAR tiers keep + * serving the long history. No new table, no new value column — the + * retention bound and the fold grain live here in code so tuning either + * never needs a migration. * * Runs on pg-boss (`DENSE_INTRADAY_RETENTION_QUEUE`), modelled on the * `mean-consolidation` boot-time converging-backfill pattern — the * production standalone image strips `tsx`, so this can never be a CLI. + * The one-shot history rebuild that converts ALREADY-folded daily rows to + * the hourly grain lives in `dense-intraday-hourly-rebuild.ts`. */ -import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; +import type { + MeasurementType, + Prisma, + PrismaClient, +} from "@/generated/prisma/client"; import { isP2002 as isUniqueConstraintViolation } from "@/lib/prisma-errors"; import { @@ -48,6 +59,10 @@ import { hkIdentifierForType, } from "./apple-health-mapping"; import { runConsolidation, type DayWriteOutcome } from "./consolidation-base"; +import { + canonicalHourlyTimestamp, + hourOfDayForUserTz, +} from "./consolidation-tz"; import { meanBucketValue } from "./consolidate-daily-mean"; import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; import { percentile } from "@/lib/insights/strain-score"; @@ -55,47 +70,33 @@ import type { PerSampleRow } from "./drain-per-sample-cumulative"; /** * iOS#34 / #69 — heart-rate (PULSE) is the densest spot signal after - * walking-distance: ~16k raw rows per Apple-Health user. Collapsing a - * folded PULSE day to a single MEAN row (the plain dense-tier fold) keeps - * the chart mean + bounds the bloat, but it would erase the two - * clinically-meaningful daily figures the rest of the app derives from the - * raw stream: + * walking-distance: ~16k raw rows per Apple-Health user. The hourly fold + * bounds that bloat while keeping the intraday shape at hour resolution, + * but two clinically-meaningful daily figures still need explicit + * preservation because they derive from the RAW stream: * * 1. **Resting HR.** `resolveRestingPulseSeries` * (`src/lib/analytics/resting-pulse.ts`) prefers a native * `RESTING_HEART_RATE` series and, for users who have none, falls back * to the 20th-percentile-of-each-day's-raw-PULSE proxy (the day's - * floor, excluding the workout burst in the upper tail). Once the raw - * PULSE rows are folded that proxy can no longer be computed — a - * single mean row is `n = 1` and is skipped by the proxy's - * `RESTING_PROXY_MIN_DAILY_SAMPLES` floor. So for a proxy user the - * resting tile would silently lose every folded day. - * 2. **Daily min / max.** The persistent `MeasurementRollup` DAY bucket - * stores `min_value` / `max_value` / `mean`, but it aggregates LIVE - * `deleted_at IS NULL` rows — recomputing it AFTER the fold collapses - * min == max == mean. - * - * The PULSE facet of this drain therefore preserves the full daily signal - * WITHOUT polluting the PULSE read path (the rollup tier averages every - * live PULSE row, so extra min/max rows of type PULSE would corrupt the - * mean): - * - * - Recompute the DAY rollup from the RAW rows BEFORE the fold, so the - * persistent tier keeps the true intraday min / max / mean, then SKIP - * the post-fold PULSE-DAY recompute that would collapse them. - * - Fold the raw PULSE day to one MEAN `stats:` PULSE row — the chart + - * rollup-mean read path is unchanged (AVG over one row == the mean). - * - For a PROXY user (zero native `RESTING_HEART_RATE` rows), mint one - * derived `RESTING_HEART_RATE` row per folded day from the day's - * 20th-percentile floor, sourced `COMPUTED` so it never collides with - * Apple's own resting rows and reads back as the clean resting series - * the resolver prefers. Users WITH native resting rows already have the - * clean signal; minting alongside would double-count, so they are - * skipped. - * - * HRV keeps the plain mean-only fold — out-of-window HRV has no resting / - * min / max consumer (the Stress engine reads the IN-window shape, which - * this drain never folds). + * floor, excluding the workout burst in the upper tail). Hourly means + * flatten the tails, so for a proxy user the resting tile would drift + * up on every folded day. The fold therefore mints one derived + * `RESTING_HEART_RATE` row per folded PULSE day from the day's RAW + * rows at fold time, sourced `COMPUTED` (see below). + * 2. **DAY-rollup fidelity (mean AND min/max).** The persistent + * `MeasurementRollup` DAY bucket stores `min_value` / `max_value` / + * `mean`, but it aggregates LIVE `deleted_at IS NULL` rows. + * Recomputing it AFTER the fold would aggregate the hourly rows: the + * min/max collapse to the extreme HOURLY means (hiding the true + * intraday extremes — for SpO2 the overnight-desaturation nadir), and + * the mean becomes an UNWEIGHTED mean-of-hourly-means that drifts + * whenever the per-hour sample counts differ. So for EVERY dense-tier + * type the DAY bucket is recomputed from the RAW rows BEFORE the fold + * — capturing the true mean, min, and max — and the post-fold DAY + * recompute is skipped. WEEK/MONTH/YEAR recomputes (enqueued by the + * pre-fold call) aggregate whatever is live at run time, exactly as + * they did against the raw stream's successor rows before. */ /** @@ -109,7 +110,7 @@ import type { PerSampleRow } from "./drain-per-sample-cumulative"; * periodically through the day and overnight; the readings map to * `aggregation: "latest"` and sit in NEITHER `CUMULATIVE_HK_TYPES` nor * `HIGH_FREQUENCY_MEAN_TYPES`, so every sample piled up raw forever. - * Folded here to one daily-mean row with the daily min/max preserved on + * Folded here to hourly-mean rows with the daily min/max preserved on * the rollup tier (the lowest sample is the overnight-desaturation * nadir — clinically the meaningful figure, not the mean). * @@ -121,6 +122,9 @@ import type { PerSampleRow } from "./drain-per-sample-cumulative"; * `RESPIRATORY_RATE` is deliberately NOT here: it is already a member of * `HIGH_FREQUENCY_MEAN_TYPES`, so the nightly mean-consolidation already * collapses it to a daily-mean row. Adding it here would double-fold it. + * + * One fold grain per tier: every member folds HOURLY. A per-type grain + * split would fork the write path and the rebuild for no reader benefit. */ export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet = new Set([ @@ -130,27 +134,8 @@ export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet = ]); /** - * Dense-tier types whose daily MIN / MAX are clinically meaningful and must - * survive the fold. For these, the DAY rollup bucket is recomputed from the - * RAW rows BEFORE the fold (while min/max are still computable from live - * rows) and the post-fold recompute that would collapse min == max == mean is - * SKIPPED — identical to the PULSE facet (iOS#34), generalised: - * - * - `PULSE` — daily min/max bound the resting floor + workout peak. - * - `OXYGEN_SATURATION` — the daily MIN is the overnight-desaturation - * nadir, the single most clinically meaningful SpO2 figure; the mean - * alone would hide a transient nocturnal dip. - * - * `HEART_RATE_VARIABILITY` is absent: out-of-window HRV has no min/max - * consumer (the Stress engine reads the IN-window shape this drain never - * folds), so it keeps the simpler post-fold recompute against its mean row. - */ -const DENSE_MIN_MAX_PRESERVING_TYPES: ReadonlySet = - new Set(["PULSE", "OXYGEN_SATURATION"]); - -/** - * Retention bound (days). Per-sample HRV / HR rows older than this window - * fold to one daily-mean row and the raw rows tombstone; rows inside the + * Retention bound (days). Per-sample dense-tier rows older than this window + * fold to hourly-mean rows and the raw rows tombstone; rows inside the * window stay raw so the Stress engine has its intra-day inputs. * * 14 days covers the Stress engine's 7-day reference window plus a margin @@ -159,9 +144,151 @@ const DENSE_MIN_MAX_PRESERVING_TYPES: ReadonlySet = */ export const DENSE_INTRADAY_RETENTION_DAYS = 14; -/** The daily-stats externalId prefix marks an already-collapsed row. */ +/** The stats externalId prefix marks an already-collapsed row. */ const DAILY_STATS_PREFIX = "stats:"; +/** + * Hourly `stats:` externalId for one (type, local-day, local-hour) slot: + * `stats::T` with a zero-padded local hour. + * Extends the locked `stats::` daily family one grain down + * — the `stats:` prefix is load-bearing: the retention scan's + * `NOT startsWith('stats:')` predicate is what keeps folded rows out of + * re-folding, so every row this tier mints MUST stay under it. + */ +export function hourlyStatsExternalId( + hkIdentifier: string, + dateKey: string, + hour: number, +): string { + return `stats:${hkIdentifier}:${dateKey}T${String(hour).padStart(2, "0")}`; +} + +/** + * Group per-sample rows into per-LOCAL-HOUR buckets (0–23) for the user's + * timezone. A spring-forward day yields no bucket for the skipped hour (no + * sample can carry a nonexistent wall-clock time); a fall-back day folds + * both instants of the repeated hour into one bucket. Pure; exposed for + * unit testing without Prisma. + */ +export function bucketRowsByLocalHour( + rows: readonly PerSampleRow[], + tz: string, +): Map { + const byHour = new Map(); + for (const row of rows) { + const hour = hourOfDayForUserTz(row.measuredAt, tz); + const slot = byHour.get(hour) ?? []; + slot.push(row); + byHour.set(hour, slot); + } + return byHour; +} + +/** + * Adopt-or-mint one hourly `stats:` row inside the caller's transaction. + * + * TWO unique indexes can collide on the mint: + * A) `(userId, type, source, externalId)` — a row already carrying the + * target hourly `stats:` externalId (a prior fold / rebuild pass). + * B) `(userId, type, measuredAt, source, sleepStage)` (NULLS NOT + * DISTINCT, tombstones included) — any row sitting on the hourly + * anchor instant, which may carry a DIFFERENT externalId (a + * per-sample row that happened at HH:30, or a sibling stats row). + * + * Determinism (VECTOR 1): resolve by index A FIRST — the row already + * holding the target externalId IS the canonical hourly row by + * construction. Only when no such row exists fall back to the index-B row + * physically occupying the anchor. A stable `orderBy: id` pins the pick. + * Adopting the wrong sibling and stamping the target externalId onto it + * would otherwise collide with the real `stats:` row on index A. + * + * Concurrency (VECTOR 2): the resolve→create is a check-then-act not + * serialised by the unique index under READ COMMITTED. If a concurrent + * writer wins the slot between the lookup and the INSERT, the `create` + * throws P2002 and Postgres aborts the whole transaction. Callers + * therefore catch P2002 OUTSIDE their transaction (a caught error inside + * would leave the tx in the aborted `25P02` state) and retry the fold + * once: on the retry the deterministic lookup finds the now-existing row + * and ADOPTS it, so the create path never fires twice. Built + * field-by-field (no spread) per the no-mass-assignment convention. + */ +export async function adoptOrMintHourlyRow( + tx: Prisma.TransactionClient, + input: { + userId: string; + type: MeasurementType; + /** Target hourly `stats:` externalId (index-A identity). */ + externalId: string; + /** Local HH:30 anchor instant (index-B identity). */ + anchor: Date; + value: number; + unit: string; + }, +): Promise { + const eidRow = await tx.measurement.findFirst({ + where: { + userId: input.userId, + type: input.type, + source: "APPLE_HEALTH", + externalId: input.externalId, + }, + select: { id: true }, + orderBy: { id: "asc" }, + }); + // Index-B occupant (`sleepStage` is NULL for these continuous types, + // matched via the NULLS-NOT-DISTINCT index). At most one row can exist. + const slotRow = await tx.measurement.findFirst({ + where: { + userId: input.userId, + type: input.type, + source: "APPLE_HEALTH", + measuredAt: input.anchor, + sleepStage: null, + }, + select: { id: true }, + orderBy: { id: "asc" }, + }); + + const adoptTarget = eidRow ?? slotRow; + if (adoptTarget) { + // Pin the adopted row to the anchor, but only when the anchor slot is + // free or already this row — a DIFFERENT row occupying the slot would + // otherwise collide on index B. In that rare case the row keeps its + // existing instant; the `stats:` externalId is the identity, + // measuredAt is secondary. + const slotIsFreeForTarget = + slotRow === null || slotRow.id === adoptTarget.id; + // Adopt: refresh value, stamp the hourly externalId, optionally + // re-anchor, and un-tombstone. This coexists with any pre-existing row + // on the slot instead of colliding with it. + await tx.measurement.update({ + where: { id: adoptTarget.id }, + data: { + value: input.value, + unit: input.unit, + externalId: input.externalId, + deletedAt: null, + ...(slotIsFreeForTarget ? { measuredAt: input.anchor } : {}), + }, + }); + return adoptTarget.id; + } + + const created = await tx.measurement.create({ + data: { + userId: input.userId, + type: input.type, + value: input.value, + unit: input.unit, + source: "APPLE_HEALTH", + measuredAt: input.anchor, + externalId: input.externalId, + }, + select: { id: true }, + }); + return created.id; +} + /** * The HealthKit identifier the derived resting row mints its `stats:` * externalId under. It is the SAME identifier `RESTING_HEART_RATE` maps to @@ -196,7 +323,8 @@ const RESTING_DERIVE_MIN_DAILY_SAMPLES = 3; * samples — the rounded 20th percentile, or `null` when the day has too few * samples to tell a resting floor from a lone reading. Pure; exported for * unit testing without Prisma. Kept value-for-value in step with - * `deriveRestingProxyFromPulse`. + * `deriveRestingProxyFromPulse`. Always fed the day's RAW rows (never the + * hourly means — those flatten the low tail the percentile reads). */ export function deriveDailyRestingFromPulse( rows: readonly PerSampleRow[], @@ -216,7 +344,16 @@ export interface DenseIntradayRetentionSummary { usersScanned: number; daysConsolidated: number; perSampleRowsSoftDeleted: number; - dailyRowsUpserted: number; + /** Hourly `stats:` rows minted / refreshed by the fold. */ + hourlyRowsUpserted: number; + /** + * Pre-hourly daily `stats:` rows retired (tombstoned) because the fold + * minted hourly rows for their day — the late-sync path: raw samples + * arriving for a day that was folded to the daily grain before + * v1.28.31. Retired in the SAME transaction as the hourly mint so no + * reader ever sees the daily and hourly rows live at once. + */ + dailyRowsRetired: number; /** * Derived `RESTING_HEART_RATE` rows minted from folded PULSE days for * proxy users (zero native resting rows). iOS#34 — preserves the @@ -244,13 +381,13 @@ export interface DenseIntradayRetentionOptions { } /** - * Run the dense intra-day retention drain. Folds per-sample HRV / HR rows - * older than the retention window into one daily-mean `stats:` row per user - * × type × day and soft-deletes the raw rows; rows inside the window stay - * raw. Idempotent — re-invocation after a successful pass converges to zero - * work because the folded rows are soft-deleted and so excluded from the - * live scan, and the minted stats row is excluded by the `NOT - * startsWith('stats:')` predicate. + * Run the dense intra-day retention drain. Folds per-sample dense-tier + * rows older than the retention window into hourly-mean `stats:` rows + * (user-local hours, anchored at local HH:30) and soft-deletes the raw + * rows; rows inside the window stay raw. Idempotent — re-invocation after + * a successful pass converges to zero work because the folded rows are + * soft-deleted and so excluded from the live scan, and the minted hourly + * rows are excluded by the `NOT startsWith('stats:')` predicate. * * Does NOT enforce any auth gate — the queue handler owns that concern. */ @@ -272,7 +409,8 @@ export async function runDenseIntradayRetention( usersScanned: 0, daysConsolidated: 0, perSampleRowsSoftDeleted: 0, - dailyRowsUpserted: 0, + hourlyRowsUpserted: 0, + dailyRowsRetired: 0, derivedRestingRowsUpserted: 0, }, }; @@ -297,6 +435,10 @@ export async function runDenseIntradayRetention( return has; } + // The base walks users serially; recordBucket needs the current user's tz + // for the dry-run hourly fan-out estimate (writeDay never runs on dry-run). + let currentTz = "Europe/Berlin"; + const { usersScanned } = await runConsolidation({ prismaClient, options: { ...options, cutoffHours }, @@ -304,13 +446,14 @@ export async function runDenseIntradayRetention( hkIdentifierForType, dailyStatsExternalId, statsPrefix: DAILY_STATS_PREFIX, - // Mean is the correct out-of-window reduction — the same reducer the - // daily-mean consolidation uses for the other high-frequency spot - // metrics. Reused verbatim so the two surfaces can never drift. + // The base's per-day reducedValue is the day MEAN; the hourly fold + // recomputes per-hour means itself (same `meanBucketValue` reducer, so + // the two surfaces can never drift), and the day-level value only + // feeds the summary/dry-run reporting. reduce: meanBucketValue, // Live per-sample rows for the type, source-scoped to APPLE_HEALTH so - // manual + Withings spot rows survive. The single minted stats row is - // excluded by the NOT-startsWith predicate so it is never re-folded; + // manual + Withings spot rows survive. The minted hourly stats rows are + // excluded by the NOT-startsWith predicate so they are never re-folded; // soft-deleted rows are excluded so a re-run converges. `cutoffAt` is // the retention boundary — rows newer than it (inside the window) are // NOT scanned, so the dense intra-day shape is preserved. @@ -336,176 +479,129 @@ export async function runDenseIntradayRetention( type, externalId, canonicalTimestamp, - reducedValue, dayRows, sourceRowIds, + tz, + dateKey, }): Promise => { const unit = dayRows[0]?.unit ?? "unknown"; const isPulse = type === "PULSE"; - const preservesMinMax = DENSE_MIN_MAX_PRESERVING_TYPES.has(type); - - // iOS#34 — min/max preservation (PULSE + SpO2). The persistent - // `MeasurementRollup` DAY bucket stores min/max/mean but aggregates LIVE - // rows; recomputing it AFTER the fold (one mean row) collapses - // min == max == mean. So for the min/max-preserving types, recompute the - // DAY bucket from the RAW rows BEFORE the fold, while they are still live - // — the bucket then keeps the TRUE intraday min/max/mean (for SpO2 the - // MIN is the overnight-desaturation nadir) — and the post-fold recompute - // is SKIPPED below. HRV has no min/max consumer, so it keeps the simpler - // post-fold recompute. - if (preservesMinMax && !options.dryRun) { - await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); + const hkIdentifier = hkIdentifierForType(type); + // Unreachable: the base skips types without an HK identifier before + // scanning; asserted so a future type-set edit fails loud, not by + // minting a malformed externalId. + if (!hkIdentifier) { + throw new Error(`no HK identifier for dense-tier type ${type}`); } - // Fold the day to one canonical daily-mean row at local-noon. - // - // Unlike `consolidate-daily-mean` (whose types never share the - // canonical local-noon instant with another path), the dense-tier - // types HRV / PULSE are dense enough that a per-sample row can land - // exactly on the canonical timestamp, and a previously-minted daily - // row can already occupy it. TWO unique indexes can collide: - // A) `(userId, type, source, externalId)` — the row already - // carrying the target `stats:` externalId. - // B) `(userId, type, measuredAt, source, sleepStage)` (NULLS NOT - // DISTINCT) — any row sitting on the canonical local-noon - // instant, which may carry a DIFFERENT externalId (a sibling - // consolidation path, a manual daily entry, or a per-sample row - // that happened at local noon). - // - // Determinism (VECTOR 1): resolve the canonical row by index A FIRST — - // the row that already holds the target `stats:` externalId IS the - // canonical daily row by construction. Only when no such row exists do - // we fall back to the index-B row physically occupying the canonical - // instant. A stable `orderBy: id` pins the pick. Adopting the wrong - // sibling and stamping the target externalId onto it would otherwise - // collide with the real `stats:` row on index A. - // - // Concurrency (VECTOR 2): the resolve→create is a check-then-act not - // serialised by the unique index under READ COMMITTED. If a concurrent - // writer wins the canonical slot between the lookup and the INSERT, the - // `create` throws P2002 and Postgres aborts the whole transaction. We - // therefore catch P2002 OUTSIDE the transaction (a caught error inside - // would leave the tx in the aborted `25P02` state) and retry the fold - // once: on the retry the deterministic lookup finds the now-existing - // row and ADOPTS it, so the create path never fires twice. Mirrors the - // `consolidate-legacy-steps.ts` P2002 classification, but recovers in - // place rather than skipping, because the colliding row IS the - // canonical daily row this fold owns. Built field-by-field (no spread) - // per the no-mass-assignment convention. - const foldOnce = async (): Promise => { + // DAY-rollup fidelity — mean AND min/max (all dense-tier types). The + // persistent `MeasurementRollup` DAY bucket aggregates LIVE rows; + // recomputing it AFTER the fold would aggregate the hourly means: the + // min/max collapse to the extreme hourly MEANS (for SpO2 hiding the + // overnight-desaturation nadir; for PULSE the resting floor + workout + // peak) and the mean becomes an unweighted mean-of-hourly-means that + // drifts whenever per-hour sample counts differ. So the DAY bucket is + // recomputed from the RAW rows BEFORE the fold, while they are still + // live, and the post-fold DAY recompute is skipped entirely. The + // WEEK/MONTH/YEAR recomputes this call enqueues run post-commit + // against whatever is live then, as before. + await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); + + // Fold the day into per-LOCAL-HOUR mean rows, anchored at local + // HH:30. Sorted for a deterministic write order across runs. + const byHour = [...bucketRowsByLocalHour(dayRows, tz).entries()].sort( + (a, b) => a[0] - b[0], + ); + + const foldOnce = async (): Promise<{ + removed: number; + retiredDaily: boolean; + }> => { let removed = 0; + let retiredDaily = false; await pc.$transaction(async (tx) => { - // Index-A row: already carries the target `stats:` externalId. By - // construction this drain only ever mints `stats:` rows at the - // canonical instant, so in the common case this row IS the - // canonical daily row already sitting at local-noon. - const eidRow = await tx.measurement.findFirst({ - where: { userId, type, source: "APPLE_HEALTH", externalId }, - select: { id: true, measuredAt: true }, - orderBy: { id: "asc" }, - }); - // Index-B row: occupies the canonical local-noon instant - // (`sleepStage` is NULL for these continuous types, matched via the - // NULLS-NOT-DISTINCT index). At most one such row can exist. - const slotRow = await tx.measurement.findFirst({ + const canonicalRowIds: string[] = []; + for (const [hour, hourRows] of byHour) { + const rowId = await adoptOrMintHourlyRow(tx, { + userId, + type, + externalId: hourlyStatsExternalId(hkIdentifier, dateKey, hour), + anchor: canonicalHourlyTimestamp(dateKey, hour, tz), + value: meanBucketValue(hourRows), + unit, + }); + canonicalRowIds.push(rowId); + } + + // Retire a live pre-hourly DAILY `stats:` row for this day (the + // late-sync path: raw samples arriving for a day folded to the + // daily grain before v1.28.31). Same transaction as the hourly + // mint — an AVG-over-live-rows reader must never see the daily + // row and the hourly rows live at once (double count). + const dailyRow = await tx.measurement.findFirst({ where: { userId, type, source: "APPLE_HEALTH", - measuredAt: canonicalTimestamp, - sleepStage: null, + externalId, + deletedAt: null, }, select: { id: true }, orderBy: { id: "asc" }, }); - - // Prefer the externalId-carrying row (index A) — adopting a sibling - // and stamping the target externalId onto it would collide with it. - const adoptTarget = eidRow ?? slotRow; - - let canonicalRowId: string; - if (adoptTarget) { - // Pin the adopted row to local-noon, but only when the canonical - // slot is free or already this row — a DIFFERENT row occupying - // the slot (a per-sample row that fell on local-noon, now being - // folded) would otherwise collide on index B. In that rare case - // (only reachable on a tz/DST shift between mints) the row keeps - // its existing instant; the `stats:` externalId is the identity, - // measuredAt is secondary. - const slotIsFreeForTarget = - slotRow === null || slotRow.id === adoptTarget.id; - // Adopt: refresh value, stamp the `stats:` externalId, optionally - // re-anchor to local-noon, and un-tombstone. This coexists with a - // pre-existing daily row instead of colliding with it. + if (dailyRow && !canonicalRowIds.includes(dailyRow.id)) { await tx.measurement.update({ - where: { id: adoptTarget.id }, - data: { - value: reducedValue, - unit, - externalId, - deletedAt: null, - ...(slotIsFreeForTarget - ? { measuredAt: canonicalTimestamp } - : {}), - }, + where: { id: dailyRow.id }, + data: { deletedAt: new Date() }, }); - canonicalRowId = adoptTarget.id; - } else { - const created = await tx.measurement.create({ - data: { - userId, - type, - value: reducedValue, - unit, - source: "APPLE_HEALTH", - measuredAt: canonicalTimestamp, - externalId, - }, - select: { id: true }, - }); - canonicalRowId = created.id; + retiredDaily = true; } // Soft-delete the out-of-window per-sample rows in the same - // transaction — tombstone, never hard-delete; they remain as an - // audit trail and drop off the live read + this pass's re-run - // discovery. EXCLUDE the canonical row itself: a per-sample row - // that happened to fall on the canonical local-noon instant is the - // row we just adopted as the daily mean, so tombstoning it would + // transaction — tombstone, never hard-delete; they remain until + // the tombstone-retention prune and drop off the live read + this + // pass's re-run discovery. EXCLUDE the adopted canonical rows: a + // per-sample row that happened to fall on an hourly anchor is the + // row just adopted as that hour's mean, so tombstoning it would // erase the fold. const del = await tx.measurement.updateMany({ where: { - id: { in: sourceRowIds, not: canonicalRowId }, + id: { in: sourceRowIds, notIn: canonicalRowIds }, deletedAt: null, }, data: { deletedAt: new Date() }, }); removed = del.count; }); - return removed; + return { removed, retiredDaily }; }; - let removed: number; + let foldResult: { removed: number; retiredDaily: boolean }; try { - removed = await foldOnce(); + foldResult = await foldOnce(); } catch (err) { if (!isUniqueConstraintViolation(err)) throw err; - // A concurrent writer won the canonical slot mid-transaction. Retry - // once: the deterministic lookup now resolves the winning row and - // adopts it, so this pass cannot create a duplicate. - removed = await foldOnce(); + // A concurrent writer won a slot mid-transaction. Retry once: the + // deterministic lookup now resolves the winning row and adopts it, + // so this pass cannot create a duplicate. + foldResult = await foldOnce(); } + // Counters mutate OUTSIDE the retryable transaction so a P2002 retry + // cannot double-count. + summary.totals.hourlyRowsUpserted += byHour.length; + if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1; // iOS#34 — preserve the resting signal for PROXY users. For a user // with zero native `RESTING_HEART_RATE` rows, the read-path resolver // derives resting from the 20th-percentile of each day's RAW PULSE; the // fold has just deleted those rows, so that day's resting figure would - // vanish. Mint one derived `RESTING_HEART_RATE` row from the same - // percentile, sourced `COMPUTED` (distinct from Apple's own resting - // rows on both unique indexes), so the day reads back as the clean - // resting series the resolver prefers. Skipped for users WITH native - // resting rows (the resolver ignores the proxy entirely for them, so a - // derived row would only double-count). + // vanish (the hourly means flatten the low tail the percentile reads). + // Mint one derived `RESTING_HEART_RATE` row from the same percentile, + // sourced `COMPUTED` (distinct from Apple's own resting rows on both + // unique indexes), so the day reads back as the clean resting series + // the resolver prefers. Skipped for users WITH native resting rows + // (the resolver ignores the proxy entirely for them, so a derived row + // would only double-count). if (isPulse) { const restingValue = deriveDailyRestingFromPulse(dayRows); if (restingValue !== null && !(await userHasNativeResting(userId))) { @@ -547,30 +643,31 @@ export async function runDenseIntradayRetention( } } - // Recompute the affected (user, type, day) rollup buckets after the - // fold commits — the rollup tier reads only `deleted_at IS NULL`, so - // the stale DAY bucket must re-aggregate against the single - // consolidated row, exactly as the mean-consolidation drain does. - // - // iOS#34 — for the min/max-preserving types (PULSE + SpO2) this would - // COLLAPSE the true intraday min/max the pre-fold recompute above just - // captured (the single folded row makes min == max == mean), so their - // DAY bucket is left as the pre-fold recompute set it. HRV (no min/max - // consumer) keeps the post-fold recompute against its single mean row. - if (!preservesMinMax) { - await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); - } + // No post-fold DAY recompute for ANY dense-tier type — it would + // overwrite the true raw-derived mean/min/max the pre-fold recompute + // above captured with hourly-mean-derived approximations (see the + // fidelity note there). - return { kind: "written", sourceRowsRemoved: removed }; + return { kind: "written", sourceRowsRemoved: foldResult.removed }; }, recordBucket: ({ dayRows, outcome }) => { summary.totals.daysConsolidated += 1; - summary.totals.dailyRowsUpserted += 1; + if (outcome === null) { + // Dry-run: writeDay never ran, so estimate the hourly fan-out here + // from the same bucketing the real fold would use. + summary.totals.hourlyRowsUpserted += bucketRowsByLocalHour( + dayRows, + currentTz, + ).size; + } summary.totals.perSampleRowsSoftDeleted += outcome?.kind === "written" ? outcome.sourceRowsRemoved : dayRows.length; }, + onUserStart: ({ tz }) => { + currentTz = tz; + }, onUserComplete: ({ userId, tz, dryRun }) => { log( `[dense-intraday-retention] user=${userId} tz=${tz}${dryRun ? " (dry-run)" : ""}`, @@ -595,7 +692,7 @@ export async function runDenseIntradayRetention( summary.totals.usersScanned = usersScanned; log( - `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`, + `[dense-intraday-retention] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} hourlyRowsUpserted=${summary.totals.hourlyRowsUpserted} dailyRowsRetired=${summary.totals.dailyRowsRetired} derivedRestingRowsUpserted=${summary.totals.derivedRestingRowsUpserted}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index c87db714c..7747182bf 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -650,8 +650,13 @@ async function persistRollupRows( * 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. + * + * Exported for callers that enqueue their own `enqueueRollupRecompute` + * (the dense-tier hourly rebuild) — a recompute over a PARTIAL bucket + * span would replace the full bucket with a partial aggregate, so every + * enqueue must cover the whole span this helper resolves. */ -function bucketSpan( +export function bucketSpan( measuredAt: Date, granularity: RollupGranularity, ): { from: Date; to: Date } { diff --git a/tests/integration/dense-intraday-hourly-rebuild.test.ts b/tests/integration/dense-intraday-hourly-rebuild.test.ts new file mode 100644 index 000000000..450bc9672 --- /dev/null +++ b/tests/integration/dense-intraday-hourly-rebuild.test.ts @@ -0,0 +1,365 @@ +/** + * v1.28.31 — real-Postgres integration coverage for + * `runDenseIntradayHourlyRebuild()`: converts a pre-hourly folded day + * (live daily `stats:` row + tombstoned raw rows) to the hourly grain. + * + * Pins against live unique indexes: the hourly mints coexist with the + * daily row (different instants, different externalIds), the daily row is + * retired in the SAME transaction as the hourly mint (no live overlap → + * no double count for an AVG-over-live-rows reader), a re-run converges + * to zero work, days without tombstones keep their daily row, and days + * inside the retention window are untouched. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { runDenseIntradayHourlyRebuild } from "@/lib/measurements/dense-intraday-hourly-rebuild"; +import { findHourlyRebuildCandidateUserIds } from "@/lib/jobs/dense-intraday-hourly-rebuild"; +import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; + +const TEST_USER_ID = "user-hourly-rebuild"; +const TZ = "Europe/Berlin"; +const DAY_KEY = "2026-05-01"; +const CANONICAL = canonicalDailyTimestamp(DAY_KEY, TZ); +const HRV_HK = "HKQuantityTypeIdentifierHeartRateVariabilitySDNN"; +const HRV_DAILY_STATS_ID = `stats:${HRV_HK}:${DAY_KEY}`; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + await getPrismaClient().user.create({ + data: { + id: TEST_USER_ID, + username: "hourly-rebuild", + email: "hourly-rebuild@example.test", + timezone: TZ, + }, + }); +}); + +/** Seed the exact shape the pre-hourly fold left behind for one HRV day. */ +async function seedFoldedDay(): Promise { + const prisma = getPrismaClient(); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 50, // the old daily mean of (40, 60) + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: CANONICAL, + externalId: HRV_DAILY_STATS_ID, + }, + }); + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 40, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T08:00:00.000Z"), // local hour 10 + externalId: "hk-hrv-1", + deletedAt: new Date("2026-05-15T02:00:00.000Z"), // fold tombstone + }, + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 60, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T09:00:00.000Z"), // local hour 11 + externalId: "hk-hrv-2", + deletedAt: new Date("2026-05-15T02:00:00.000Z"), + }, + ], + }); +} + +describe("runDenseIntradayHourlyRebuild (real Postgres)", () => { + it("rebuilds hourly means from the tombstones and retires the daily row atomically", async () => { + const prisma = getPrismaClient(); + await seedFoldedDay(); + + const summary = await runDenseIntradayHourlyRebuild(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + + expect(summary.totals.daysRebuilt).toBe(1); + expect(summary.totals.hourlyRowsUpserted).toBe(2); + expect(summary.totals.dailyRowsRetired).toBe(1); + expect(summary.totals.daysFailed).toBe(0); + + const live = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + orderBy: { measuredAt: "asc" }, + }); + // ONLY the hourly rows are live — never the daily row alongside them. + expect(live).toHaveLength(2); + expect(live.map((r) => r.externalId)).toEqual([ + `stats:${HRV_HK}:${DAY_KEY}T10`, + `stats:${HRV_HK}:${DAY_KEY}T11`, + ]); + expect(live.map((r) => r.value)).toEqual([40, 60]); + expect(live.map((r) => r.measuredAt.toISOString())).toEqual([ + "2026-05-01T08:30:00.000Z", + "2026-05-01T09:30:00.000Z", + ]); + + // The daily row is tombstoned, and the raw tombstones stayed tombstoned + // (the rebuild reads them, never resurrects them). + const daily = await prisma.measurement.findFirst({ + where: { userId: TEST_USER_ID, externalId: HRV_DAILY_STATS_ID }, + }); + expect(daily?.deletedAt).not.toBeNull(); + const rawTombstones = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + externalId: { in: ["hk-hrv-1", "hk-hrv-2"] }, + }, + }); + expect(rawTombstones.every((r) => r.deletedAt !== null)).toBe(true); + }); + + it("re-run converges to zero work (the retired daily row IS the marker)", async () => { + const prisma = getPrismaClient(); + await seedFoldedDay(); + + const first = await runDenseIntradayHourlyRebuild(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + expect(first.totals.daysRebuilt).toBe(1); + + const second = await runDenseIntradayHourlyRebuild(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + expect(second.totals.daysRebuilt).toBe(0); + expect(second.totals.hourlyRowsUpserted).toBe(0); + expect(second.totals.dailyRowsRetired).toBe(0); + expect(second.totals.daysSkippedNoTombstones).toBe(0); + + // The hourly rows are unchanged by the second pass. + const live = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + orderBy: { measuredAt: "asc" }, + }); + expect(live.map((r) => r.value)).toEqual([40, 60]); + }); + + it("keeps the daily row for days whose tombstones were pruned (nothing to reconstruct)", async () => { + const prisma = getPrismaClient(); + // Daily row only — the raw tombstones were hard-deleted by the + // tombstone-retention prune long ago. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 50, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: CANONICAL, + externalId: HRV_DAILY_STATS_ID, + }, + }); + + const summary = await runDenseIntradayHourlyRebuild(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + + expect(summary.totals.daysRebuilt).toBe(0); + expect(summary.totals.daysSkippedNoTombstones).toBe(1); + + const daily = await prisma.measurement.findFirst({ + where: { userId: TEST_USER_ID, externalId: HRV_DAILY_STATS_ID }, + }); + // Untouched and still live — it is the day's only representation. + expect(daily?.deletedAt).toBeNull(); + expect(daily?.value).toBeCloseTo(50, 6); + }); + + it("never touches days inside the retention window", async () => { + const prisma = getPrismaClient(); + // A daily row anchored 2 days ago — inside the default 14-day window. + // (Synthetic: the real fold never produces one this recent, but the + // window bound must hold regardless of how the row got there.) + const recentNoon = new Date(Date.now() - 2 * 24 * 60 * 60 * 1000); + const recentDayKey = recentNoon.toISOString().slice(0, 10); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 50, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: recentNoon, + externalId: `stats:${HRV_HK}:${recentDayKey}`, + }, + }); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 40, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date(recentNoon.getTime() - 60 * 60 * 1000), + externalId: "hk-hrv-recent", + deletedAt: new Date(), + }, + }); + + const summary = await runDenseIntradayHourlyRebuild(prisma, { + userId: TEST_USER_ID, + // Default retention window. + log: () => {}, + }); + + expect(summary.totals.daysRebuilt).toBe(0); + const daily = await prisma.measurement.findFirst({ + where: { + userId: TEST_USER_ID, + externalId: `stats:${HRV_HK}:${recentDayKey}`, + }, + }); + expect(daily?.deletedAt).toBeNull(); + }); + + it("never touches non-APPLE_HEALTH sources", async () => { + const prisma = getPrismaClient(); + // A Withings daily-shaped row + a tombstoned Withings raw row must both + // survive untouched — every rebuild predicate is source-scoped. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 48, + unit: "ms", + source: "WITHINGS", + measuredAt: CANONICAL, + externalId: HRV_DAILY_STATS_ID, + }, + }); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 44, + unit: "ms", + source: "WITHINGS", + measuredAt: new Date("2026-05-01T08:00:00.000Z"), + externalId: "withings-hrv-1", + deletedAt: new Date(), + }, + }); + + const summary = await runDenseIntradayHourlyRebuild(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + + expect(summary.totals.daysRebuilt).toBe(0); + const withingsDaily = await prisma.measurement.findFirst({ + where: { userId: TEST_USER_ID, source: "WITHINGS", deletedAt: null }, + }); + expect(withingsDaily?.value).toBeCloseTo(48, 6); + }); +}); + +describe("findHourlyRebuildCandidateUserIds (boot-discovery SQL, real Postgres)", () => { + // Discovery bound "older than now" so the seeded 2026-05-01 day is in scope. + const WINDOW_START = new Date(); + + it("finds a user with a folded day (live daily row + paired tombstones)", async () => { + await seedFoldedDay(); + const users = await findHourlyRebuildCandidateUserIds( + getPrismaClient(), + WINDOW_START, + ); + expect(users).toEqual([TEST_USER_ID]); + }); + + it("converges: after the rebuild the pairing no longer matches", async () => { + const prisma = getPrismaClient(); + await seedFoldedDay(); + await runDenseIntradayHourlyRebuild(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + const users = await findHourlyRebuildCandidateUserIds(prisma, WINDOW_START); + // The retired daily row is the durable marker — zero candidates left, + // so the boot discovery enqueues nothing on every later boot. + expect(users).toEqual([]); + }); + + it("ignores a daily row without tombstoned raws, and hourly-shaped rows", async () => { + const prisma = getPrismaClient(); + // Daily row alone (tombstones pruned) — no pair. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 50, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: CANONICAL, + externalId: HRV_DAILY_STATS_ID, + }, + }); + // Hourly-grain row + a tombstoned raw in its window — the hourly + // externalId shape must not qualify as a daily candidate. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "PULSE", + value: 60, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-02T08:30:00.000Z"), + externalId: "stats:HKQuantityTypeIdentifierHeartRate:2026-05-02T10", + }, + }); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "PULSE", + value: 62, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-02T08:00:00.000Z"), + externalId: "hk-pulse-raw", + deletedAt: new Date(), + }, + }); + + const users = await findHourlyRebuildCandidateUserIds(prisma, WINDOW_START); + expect(users).toEqual([]); + }); + + it("ignores days inside the retention window (windowStart bound)", async () => { + await seedFoldedDay(); + // A window start BEFORE the seeded day excludes it. + const users = await findHourlyRebuildCandidateUserIds( + getPrismaClient(), + new Date("2026-04-01T00:00:00.000Z"), + ); + expect(users).toEqual([]); + }); +}); diff --git a/tests/integration/dense-intraday-retention.test.ts b/tests/integration/dense-intraday-retention.test.ts index 1fcd9a63b..420d1dcad 100644 --- a/tests/integration/dense-intraday-retention.test.ts +++ b/tests/integration/dense-intraday-retention.test.ts @@ -1,15 +1,16 @@ /** * v1.10.2 — real-Postgres integration coverage for - * `runDenseIntradayRetention()`. The mocked unit tests could not surface - * the v1.10.0.2 P2002 fold-coexistence bug: the fold mints a canonical - * daily-mean row at the user's local-noon instant, but a row may ALREADY - * sit on that `(userId, type, measuredAt, source, sleepStage)` unique - * index from another path. An externalId-keyed upsert misses the lookup - * and the INSERT then violates the measured_at index with P2002. + * `runDenseIntradayRetention()`. v1.28.31 — hourly fold grain. * - * This file seeds that exact colliding shape against a live Postgres and - * asserts the reworked fold coexists with the pre-existing daily row, - * stays idempotent, and preserves the dense-tier scope. + * The mocked unit tests could not surface the v1.10.0.2 P2002 + * fold-coexistence bug class: the fold mints canonical `stats:` rows at + * deterministic instants, but a row may ALREADY sit on the + * `(userId, type, measuredAt, source, sleepStage)` unique index from + * another path, or already carry the target externalId on + * `(userId, type, source, externalId)`. This file seeds those colliding + * shapes against a live Postgres and asserts the hourly fold coexists / + * adopts, retires a pre-hourly daily row atomically, stays idempotent, + * and preserves the dense-tier scope. */ import { beforeEach, describe, expect, it } from "vitest"; @@ -20,10 +21,10 @@ import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; const TEST_USER_ID = "user-dense-retention"; const TZ = "Europe/Berlin"; const DAY_KEY = "2026-05-01"; -// The canonical local-noon instant the fold writes for the seeded day. +// The canonical local-noon instant the PRE-hourly fold wrote for the day. const CANONICAL = canonicalDailyTimestamp(DAY_KEY, TZ); -const HRV_STATS_ID = - "stats:HKQuantityTypeIdentifierHeartRateVariabilitySDNN:2026-05-01"; +const HRV_HK = "HKQuantityTypeIdentifierHeartRateVariabilitySDNN"; +const HRV_DAILY_STATS_ID = `stats:${HRV_HK}:${DAY_KEY}`; beforeEach(async () => { await truncateAllTables(getPrismaClient()); @@ -38,7 +39,7 @@ beforeEach(async () => { }); describe("runDenseIntradayRetention (real Postgres)", () => { - it("folds out-of-window HRV per-sample rows into one daily MEAN and soft-deletes them", async () => { + it("folds out-of-window HRV per-sample rows into hourly MEANs and soft-deletes them", async () => { const prisma = getPrismaClient(); await prisma.measurement.createMany({ data: [ @@ -48,6 +49,7 @@ describe("runDenseIntradayRetention (real Postgres)", () => { value: 40, unit: "ms", source: "APPLE_HEALTH", + // Berlin UTC+2 → local hour 10. measuredAt: new Date("2026-05-01T08:00:00.000Z"), externalId: "hk-hrv-1", }, @@ -57,9 +59,20 @@ describe("runDenseIntradayRetention (real Postgres)", () => { value: 60, unit: "ms", source: "APPLE_HEALTH", - measuredAt: new Date("2026-05-01T09:00:00.000Z"), + // Same local hour 10 — folds into the SAME hourly mean. + measuredAt: new Date("2026-05-01T08:40:00.000Z"), externalId: "hk-hrv-2", }, + { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 70, + unit: "ms", + source: "APPLE_HEALTH", + // Local hour 11 — its own hourly row. + measuredAt: new Date("2026-05-01T09:10:00.000Z"), + externalId: "hk-hrv-3", + }, ], }); @@ -70,7 +83,8 @@ describe("runDenseIntradayRetention (real Postgres)", () => { }); expect(summary.totals.daysConsolidated).toBe(1); - expect(summary.totals.perSampleRowsSoftDeleted).toBe(2); + expect(summary.totals.perSampleRowsSoftDeleted).toBe(3); + expect(summary.totals.hourlyRowsUpserted).toBe(2); const live = await prisma.measurement.findMany({ where: { @@ -78,35 +92,35 @@ describe("runDenseIntradayRetention (real Postgres)", () => { type: "HEART_RATE_VARIABILITY", deletedAt: null, }, + orderBy: { measuredAt: "asc" }, }); - expect(live).toHaveLength(1); - // MEAN of (40, 60) = 50. + expect(live).toHaveLength(2); + // Hour 10: MEAN of (40, 60) = 50, anchored at local 10:30 (08:30Z). + expect(live[0].externalId).toBe(`stats:${HRV_HK}:${DAY_KEY}T10`); expect(live[0].value).toBeCloseTo(50, 6); - expect(live[0].externalId).toBe(HRV_STATS_ID); - expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); + expect(live[0].measuredAt.toISOString()).toBe("2026-05-01T08:30:00.000Z"); + // Hour 11: its own value, anchored at local 11:30 (09:30Z). + expect(live[1].externalId).toBe(`stats:${HRV_HK}:${DAY_KEY}T11`); + expect(live[1].value).toBeCloseTo(70, 6); + expect(live[1].measuredAt.toISOString()).toBe("2026-05-01T09:30:00.000Z"); }); - it("coexists with a pre-existing daily row on the canonical instant (the P2002 case)", async () => { + it("retires a pre-hourly DAILY stats row in the same pass (late-sync path, no P2002, no double count)", async () => { const prisma = getPrismaClient(); - // A previously-collapsed daily `stats:` row ALREADY sits on the - // canonical local-noon instant under a DIFFERENT HK stats id than the - // one the fold mints (e.g. a row minted by a sibling consolidation path, - // or a legacy stats id). The scan EXCLUDES it (NOT startsWith 'stats:'), - // so it is never folded into the mean — but pre-v1.10.2 the fold's - // externalId-keyed upsert missed it on lookup and the INSERT then - // collided on (userId, type, measuredAt, source, sleepStage) with P2002. + // A pre-v1.28.31 fold already collapsed this day to the daily grain: + // the canonical daily row sits at local-noon with the daily stats id. await prisma.measurement.create({ data: { userId: TEST_USER_ID, type: "HEART_RATE_VARIABILITY", - value: 999, // stale value to be overwritten by the fold + value: 999, // stale daily mean — superseded by the hourly rows unit: "ms", source: "APPLE_HEALTH", measuredAt: CANONICAL, - externalId: "stats:legacy-hrv-id:2026-05-01", + externalId: HRV_DAILY_STATS_ID, }, }); - // Out-of-window per-sample rows that should fold into the canonical row. + // Late-synced raw samples arrive for the already-folded day. await prisma.measurement.createMany({ data: [ { @@ -137,6 +151,7 @@ describe("runDenseIntradayRetention (real Postgres)", () => { log: () => {}, }); expect(summary.totals.daysConsolidated).toBe(1); + expect(summary.totals.dailyRowsRetired).toBe(1); const live = await prisma.measurement.findMany({ where: { @@ -144,75 +159,51 @@ describe("runDenseIntradayRetention (real Postgres)", () => { type: "HEART_RATE_VARIABILITY", deletedAt: null, }, + orderBy: { measuredAt: "asc" }, }); - // Exactly one live daily row: the pre-existing one, adopted in place. - expect(live).toHaveLength(1); - expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); - // The fold refreshed the value to the per-sample MEAN (50), overwriting - // the stale 999, and stamped the canonical `stats:` externalId. - expect(live[0].value).toBeCloseTo(50, 6); - expect(live[0].externalId).toBe(HRV_STATS_ID); + // ONLY the hourly rows are live — the daily row is tombstoned in the + // same transaction, so an AVG-over-live-rows reader never double-counts. + expect(live).toHaveLength(2); + expect(live.map((r) => r.externalId)).toEqual([ + `stats:${HRV_HK}:${DAY_KEY}T08`, + `stats:${HRV_HK}:${DAY_KEY}T20`, + ]); + expect(live.map((r) => r.value)).toEqual([40, 60]); - // The two out-of-window samples are tombstoned. - const tombstoned = await prisma.measurement.findMany({ - where: { - userId: TEST_USER_ID, - type: "HEART_RATE_VARIABILITY", - deletedAt: { not: null }, - }, + const daily = await prisma.measurement.findFirst({ + where: { userId: TEST_USER_ID, externalId: HRV_DAILY_STATS_ID }, }); - expect(tombstoned).toHaveLength(2); + expect(daily?.deletedAt).not.toBeNull(); }); - it("adopts the row already carrying the target stats externalId on the canonical instant (VECTOR 1)", async () => { + it("adopts a row already carrying the target hourly stats externalId (VECTOR 1, no P2002)", async () => { const prisma = getPrismaClient(); - // BOTH-present collision (VECTOR 1). A prior fold already minted the - // canonical daily row carrying the TARGET `stats:` externalId and it - // sits on the canonical local-noon instant. A fresh batch of - // out-of-window per-sample rows then arrives for the same day. The fold - // must adopt the EXISTING canonical row in place — refreshing its value - // and re-anchoring it — never INSERT a second row at local-noon (that - // would collide on the measured_at composite) and never stamp the target - // externalId onto a sibling (that would collide on the externalId - // composite with this very row). The pre-fix `findFirst` had no - // externalId discrimination and no deterministic `orderBy`, so it could - // pick the wrong row and trip P2002 on the externalId index. + // A prior hourly fold already minted the hour-08 row at its anchor. await prisma.measurement.create({ data: { userId: TEST_USER_ID, type: "HEART_RATE_VARIABILITY", - value: 999, // stale — the fold overwrites it with the day's mean + value: 999, // stale — refreshed by the re-fold unit: "ms", source: "APPLE_HEALTH", - measuredAt: CANONICAL, - externalId: HRV_STATS_ID, // the TARGET canonical externalId + measuredAt: new Date("2026-05-01T06:30:00.000Z"), // local 08:30 + externalId: `stats:${HRV_HK}:${DAY_KEY}T08`, }, }); - // Out-of-window per-sample rows that fold into the canonical row. - await prisma.measurement.createMany({ - data: [ - { - userId: TEST_USER_ID, - type: "HEART_RATE_VARIABILITY", - value: 40, - unit: "ms", - source: "APPLE_HEALTH", - measuredAt: new Date("2026-05-01T06:00:00.000Z"), - externalId: "hk-hrv-1", - }, - { - userId: TEST_USER_ID, - type: "HEART_RATE_VARIABILITY", - value: 60, - unit: "ms", - source: "APPLE_HEALTH", - measuredAt: new Date("2026-05-01T18:00:00.000Z"), - externalId: "hk-hrv-2", - }, - ], + // A late raw sample lands in the same local hour. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 40, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T06:10:00.000Z"), + externalId: "hk-hrv-late", + }, }); - // Must not throw P2002. + // Must not throw P2002 — the fold adopts the existing hourly row. const summary = await runDenseIntradayRetention(prisma, { userId: TEST_USER_ID, retentionDays: 0, @@ -227,30 +218,17 @@ describe("runDenseIntradayRetention (real Postgres)", () => { deletedAt: null, }, }); - // Exactly ONE live canonical row remains — the row that already carried - // the target `stats:` externalId, adopted in place and folded. expect(live).toHaveLength(1); - expect(live[0].externalId).toBe(HRV_STATS_ID); - expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); - // MEAN of the two folded per-sample values (40, 60) = 50, overwriting - // the stale 999. - expect(live[0].value).toBeCloseTo(50, 6); - - // Both scanned per-sample rows are tombstoned. - const tombstoned = await prisma.measurement.findMany({ - where: { - userId: TEST_USER_ID, - type: "HEART_RATE_VARIABILITY", - deletedAt: { not: null }, - }, - }); - expect(tombstoned).toHaveLength(2); + expect(live[0].externalId).toBe(`stats:${HRV_HK}:${DAY_KEY}T08`); + // Refreshed to the late sample's hourly mean, overwriting the stale 999. + expect(live[0].value).toBeCloseTo(40, 6); + expect(live[0].measuredAt.toISOString()).toBe("2026-05-01T06:30:00.000Z"); }); - it("does not tombstone a per-sample row that happens to fall on the canonical instant", async () => { + it("does not tombstone a per-sample row that happens to fall on an hourly anchor", async () => { const prisma = getPrismaClient(); - // One per-sample row sits exactly on the canonical local-noon instant; - // it must become the adopted daily row, not get soft-deleted. + // One per-sample row sits exactly on the local-10:30 anchor instant; + // it must become the adopted hourly row, not get soft-deleted. await prisma.measurement.createMany({ data: [ { @@ -259,8 +237,8 @@ describe("runDenseIntradayRetention (real Postgres)", () => { value: 50, unit: "ms", source: "APPLE_HEALTH", - measuredAt: CANONICAL, - externalId: "hk-hrv-noon", + measuredAt: new Date("2026-05-01T08:30:00.000Z"), // local 10:30 + externalId: "hk-hrv-anchor", }, { userId: TEST_USER_ID, @@ -268,8 +246,8 @@ describe("runDenseIntradayRetention (real Postgres)", () => { value: 70, unit: "ms", source: "APPLE_HEALTH", - measuredAt: new Date("2026-05-01T06:00:00.000Z"), - externalId: "hk-hrv-morning", + measuredAt: new Date("2026-05-01T08:00:00.000Z"), // same local hour + externalId: "hk-hrv-sibling", }, ], }); @@ -289,13 +267,13 @@ describe("runDenseIntradayRetention (real Postgres)", () => { }, }); expect(live).toHaveLength(1); - expect(live[0].measuredAt.getTime()).toBe(CANONICAL.getTime()); - // MEAN of (50, 70) = 60 — the adopted row carries the day's mean. + expect(live[0].measuredAt.toISOString()).toBe("2026-05-01T08:30:00.000Z"); + // MEAN of (50, 70) = 60 — the adopted row carries the hour's mean. expect(live[0].value).toBeCloseTo(60, 6); - expect(live[0].externalId).toBe(HRV_STATS_ID); + expect(live[0].externalId).toBe(`stats:${HRV_HK}:${DAY_KEY}T10`); }); - it("is idempotent — a second run is a no-op and the value stays correct", async () => { + it("is idempotent — a second run is a no-op and the values stay correct", async () => { const prisma = getPrismaClient(); await prisma.measurement.createMany({ data: [ @@ -327,9 +305,11 @@ describe("runDenseIntradayRetention (real Postgres)", () => { }); expect(first.totals.daysConsolidated).toBe(1); expect(first.totals.perSampleRowsSoftDeleted).toBe(2); + expect(first.totals.hourlyRowsUpserted).toBe(2); // Second run must converge to zero work and never collide on the - // canonical row it minted on the first pass. + // hourly rows it minted on the first pass (the `stats:` prefix keeps + // them out of the scan). const second = await runDenseIntradayRetention(prisma, { userId: TEST_USER_ID, retentionDays: 0, @@ -337,13 +317,15 @@ describe("runDenseIntradayRetention (real Postgres)", () => { }); expect(second.totals.daysConsolidated).toBe(0); expect(second.totals.perSampleRowsSoftDeleted).toBe(0); + expect(second.totals.hourlyRowsUpserted).toBe(0); const live = await prisma.measurement.findMany({ where: { userId: TEST_USER_ID, type: "PULSE", deletedAt: null }, + orderBy: { measuredAt: "asc" }, }); - expect(live).toHaveLength(1); - // MEAN of (60, 80) = 70. - expect(live[0].value).toBeCloseTo(70, 6); + expect(live).toHaveLength(2); + // Each hour keeps its own mean (60 and 80) — never a blended 70. + expect(live.map((r) => r.value)).toEqual([60, 80]); }); it("keeps in-window per-sample rows raw (retention bound)", async () => {