Merge pull request #477 from MBombeck/release/v1.28.31

release/v1.28.31 — heart-rate history keeps its daily shape
This commit is contained in:
Marc-André Bombeck authored and GitHub committed 2026-07-12 12:19:23 +02:00
commit 14b61e1084
20 files changed
+2397 -413

No files matched your search

+16
View File
@@ -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.
+1 -1
View File
@@ -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.
+1 -1
View File
@@ -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",
+1 -1
View File
@@ -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/*`
@@ -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/,
);
});
});
@@ -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<HH>` 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<string[]> {
const types = Array.from(DENSE_INTRADAY_RETENTION_TYPES).map(String);
if (types.length === 0) return [];
const users = await prismaClient.$queryRaw<Array<{ user_id: string }>>`
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),
};
}
}
+6 -5
View File
@@ -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:
+83 -5
View File
@@ -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<DenseIntradayRetentionPayload>(
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<DenseIntradayHourlyRebuildPayload>(
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<void> {
);
}
// 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
@@ -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:<HK>:<YYYY-MM-DD>T<HH>` 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:<HK>:<YYYY-MM-DD>T<HH> 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<HH>.
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);
});
});
@@ -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<string, MockRow[]>;
tombstonesByType: Record<string, MockRow[]>;
}) {
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<unknown>) =>
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<typeof vi.fn>).mockImplementation(
async (cb: (t: unknown) => Promise<unknown>) => {
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);
});
});
@@ -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",
);
@@ -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<string, unknown[]>,
// 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<typeof vi.fn>,
};
})();
// 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 () => {
@@ -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",
);
@@ -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<TType extends MeasurementType>(
reducedValue,
dayRows,
sourceRowIds,
tz,
dateKey,
shouldMint,
});
}
+66
View File
@@ -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<string, Intl.DateTimeFormat>();
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
@@ -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:<HK>:<YYYY-MM-DD>` 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:<HK>:<YYYY-MM-DD>` row exists (the
* pre-hourly fold's output — hourly rows carry a `T<HH>` 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:<HK>:` — exactly a calendar-day key. Hourly rows
* (`…:<YYYY-MM-DD>T<HH>`) and the iOS hourly-HR wire rows
* (`…:<ISO-instant>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<DenseIntradayHourlyRebuildSummary> {
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 };
}
+329 -232
View File
@@ -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:<HK>:<day>` 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:<HK>:<YYYY-MM-DD>T<HH>` 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<MeasurementType> =
new Set<MeasurementType>([
@@ -130,27 +134,8 @@ export const DENSE_INTRADAY_RETENTION_TYPES: ReadonlySet<MeasurementType> =
]);
/**
* 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<MeasurementType> =
new Set<MeasurementType>(["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<MeasurementType> =
*/
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:<HKIdentifier>:<YYYY-MM-DD>T<HH>` with a zero-padded local hour.
* Extends the locked `stats:<HK>:<YYYY-MM-DD>` 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<number, PerSampleRow[]> {
const byHour = new Map<number, PerSampleRow[]>();
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<string> {
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<MeasurementType>({
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<DayWriteOutcome> => {
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<number> => {
// 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;
+6 -1
View File
@@ -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 } {
@@ -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<void> {
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([]);
});
});
@@ -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 () => {