From 8f1e3c5d4403d25910cb04bd5ee9e20db6b29685 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 21 Jun 2026 00:29:35 +0200 Subject: [PATCH] feat(measurements): preserve resting + daily min/max when folding dense heart rate The dense intra-day retention drain folds out-of-window per-sample PULSE rows to one daily-mean row to bound the high-frequency ingest. The mean-only fold erased the two clinically-meaningful daily figures derived from the raw stream: the resting estimate (20th-percentile of each day's PULSE, the fallback for users without a native resting series) and the daily min/max held in the persistent rollup DAY bucket. Preserve both in the PULSE facet of the fold: - Recompute the PULSE DAY rollup from the raw rows before the fold, while they are still live, so the bucket keeps the true intraday min/max/mean; skip the post-fold recompute that would collapse them. - For users with no native resting series, mint one derived RESTING_HEART_RATE row per folded day from the day's 20th percentile, sourced COMPUTED so it never collides with measured resting rows and reads back as the clean resting series. Users with a native series are skipped to avoid double-counting; the probe is memoised per user. Add a per-day failure boundary so one poisoned day can no longer abort the whole walk. Surface the derived-resting count through the handler return, the wide-event annotation, and the worker log. --- .../stress-strain-retention-queue.test.ts | 7 + src/lib/jobs/dense-intraday-retention.ts | 17 +- src/lib/jobs/reminder/register-rollup.ts | 9 +- .../dense-intraday-pulse-summary.test.ts | 341 ++++++++++++++++++ .../measurements/dense-intraday-retention.ts | 211 ++++++++++- 5 files changed, 578 insertions(+), 7 deletions(-) create mode 100644 src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts diff --git a/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts b/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts index 576dc93b4..a529fa6fb 100644 --- a/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts +++ b/src/lib/jobs/__tests__/stress-strain-retention-queue.test.ts @@ -140,4 +140,11 @@ describe("reminder-worker — dense intra-day retention wiring", () => { // same tick as the cumulative + mean drains. expect(source).toMatch(/runDenseIntradayRetention\(\s*getWorkerPrisma\(\)/); }); + + it("surfaces the iOS#34 derived-resting count in the per-user worker log", () => { + // The PULSE facet mints derived RESTING_HEART_RATE rows for proxy users; + // the per-user handler returns the count and the worker logs it so the + // signal-preservation is observable in production. + expect(source).toMatch(/derivedRestingRowsUpserted/); + }); }); diff --git a/src/lib/jobs/dense-intraday-retention.ts b/src/lib/jobs/dense-intraday-retention.ts index f10a445ce..7654e50fe 100644 --- a/src/lib/jobs/dense-intraday-retention.ts +++ b/src/lib/jobs/dense-intraday-retention.ts @@ -89,11 +89,19 @@ export interface DenseIntradayRetentionPayload { */ export async function runDenseIntradayRetentionForUser( userId: string, -): Promise<{ daysConsolidated: number; perSampleRowsSoftDeleted: number }> { +): Promise<{ + daysConsolidated: number; + perSampleRowsSoftDeleted: number; + derivedRestingRowsUpserted: number; +}> { // Kill-switch: when an operator opts out, no-op so any already-queued // backlog completes cleanly (drains the queue) instead of running the fold. if (!DENSE_INTRADAY_RETENTION_ENABLED) { - return { daysConsolidated: 0, perSampleRowsSoftDeleted: 0 }; + return { + daysConsolidated: 0, + perSampleRowsSoftDeleted: 0, + derivedRestingRowsUpserted: 0, + }; } const summary = await runDenseIntradayRetention(prisma, { userId, @@ -108,12 +116,17 @@ export async function runDenseIntradayRetentionForUser( days: summary.totals.daysConsolidated, per_sample_rows_soft_deleted: summary.totals.perSampleRowsSoftDeleted, daily_rows_upserted: summary.totals.dailyRowsUpserted, + // 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: + summary.totals.derivedRestingRowsUpserted, }, }, }); return { daysConsolidated: summary.totals.daysConsolidated, perSampleRowsSoftDeleted: summary.totals.perSampleRowsSoftDeleted, + derivedRestingRowsUpserted: summary.totals.derivedRestingRowsUpserted, }; } diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index 2420b4b07..56ccd1b8d 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -238,11 +238,14 @@ export async function registerRollupQueues( for (const job of jobs) { const { userId } = job.data; try { - const { daysConsolidated, perSampleRowsSoftDeleted } = - await runDenseIntradayRetentionForUser(userId); + const { + daysConsolidated, + perSampleRowsSoftDeleted, + derivedRestingRowsUpserted, + } = await runDenseIntradayRetentionForUser(userId); workerLog( "info", - `[dense-intraday-retention] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted}`, + `[dense-intraday-retention] user=${userId} days=${daysConsolidated} perSampleRowsSoftDeleted=${perSampleRowsSoftDeleted} derivedRestingRowsUpserted=${derivedRestingRowsUpserted}`, ); } catch (err) { recordError(); diff --git a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts new file mode 100644 index 000000000..a2926ef0f --- /dev/null +++ b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts @@ -0,0 +1,341 @@ +/** + * 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 + * figures the app derives from the raw stream: + * + * 1. Resting HR — the read-path resolver derives it from the 20th-percentile + * of each day's RAW PULSE for users with no native RESTING_HEART_RATE. + * 2. Daily min/max — the persistent rollup DAY bucket, which aggregates LIVE + * rows, would collapse min == max == mean after the fold. + * + * This suite pins the PULSE facet that preserves both: the pre-fold DAY-rollup + * recompute (true min/max captured while raw rows are live), the derived + * RESTING_HEART_RATE row minted from the day's 20th percentile for proxy users + * only, the no-double-count guard for native-resting users, the derive helper's + * percentile + sample-floor semantics, idempotency, and the per-day failure + * boundary. + */ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +const { recomputeBucketsForMeasurement } = vi.hoisted(() => ({ + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + recomputeBucketsForMeasurement, +})); + +import { + runDenseIntradayRetention, + deriveDailyRestingFromPulse, +} from "../dense-intraday-retention"; +import type { PerSampleRow } from "../drain-per-sample-cumulative"; +import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; + +function pulseRow(id: string, value: number, iso: string): PerSampleRow { + return { + id, + type: "PULSE" as MeasurementType, + value, + measuredAt: new Date(iso), + externalId: null, + unit: "bpm", + }; +} + +/** + * Prisma mock covering the full PULSE-summary flow: the scan (per-type rows), + * the per-user native-resting probe (top-level `findFirst`), the fold + * transaction (`create`/`update`/`findFirst`/`updateMany`), and the derived + * resting `upsert` (top-level). + */ +function buildPrismaMock(opts: { + pulseRows: PerSampleRow[]; + hasNativeResting?: boolean; +}) { + // Inside the fold transaction: no pre-existing canonical row → create path. + const txCreate = vi.fn().mockResolvedValue({ id: "minted-daily" }); + const txUpdate = vi.fn().mockResolvedValue({}); + const txFindFirst = vi.fn().mockResolvedValue(null); + const txUpdateMany = vi + .fn() + .mockResolvedValue({ count: opts.pulseRows.length }); + const tx = { + measurement: { + create: txCreate, + update: txUpdate, + findFirst: txFindFirst, + updateMany: txUpdateMany, + }, + }; + + // Top-level client. + const findMany = vi.fn(async (args: { where: { type: string } }) => + args.where.type === "PULSE" ? opts.pulseRows : [], + ); + // Native-resting probe is the only top-level `findFirst` call. + const topFindFirst = vi + .fn() + .mockResolvedValue(opts.hasNativeResting ? { id: "native-rhr" } : null); + const upsert = vi.fn().mockResolvedValue({ id: "derived-rhr" }); + + const mock = { + user: { + findMany: vi + .fn() + .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]), + }, + measurement: { findMany, findFirst: topFindFirst, upsert }, + $transaction: vi.fn(async (cb: (t: unknown) => Promise) => cb(tx)), + } as unknown as PrismaClient; + + return { mock, txCreate, txUpdateMany, upsert, topFindFirst }; +} + +beforeEach(() => { + recomputeBucketsForMeasurement.mockClear(); +}); + +describe("deriveDailyRestingFromPulse", () => { + it("returns the rounded 20th percentile of the day's samples", () => { + // 10 samples 60..69 → p20 ≈ 61.8 → round 62. + const rows = Array.from({ length: 10 }, (_, i) => + pulseRow(`r${i}`, 60 + i, "2026-05-01T08:00:00.000Z"), + ); + expect(deriveDailyRestingFromPulse(rows)).toBe(62); + }); + + it("excludes the workout burst — a dense high tail cannot drag resting up", () => { + // 8 resting samples ~58 + 4 workout samples ~150. p20 stays in the low band. + const rows = [ + ...[57, 58, 58, 59, 59, 60, 60, 61].map((v, i) => + pulseRow(`a${i}`, v, "2026-05-01T07:00:00.000Z"), + ), + ...[140, 150, 155, 160].map((v, i) => + pulseRow(`w${i}`, v, "2026-05-01T18:00:00.000Z"), + ), + ]; + const resting = deriveDailyRestingFromPulse(rows); + expect(resting).toBeLessThan(70); + expect(resting).toBeGreaterThan(50); + }); + + it("returns null below the minimum daily sample floor", () => { + expect( + deriveDailyRestingFromPulse([ + pulseRow("a", 130, "2026-05-01T18:00:00.000Z"), + pulseRow("b", 135, "2026-05-01T18:05:00.000Z"), + ]), + ).toBeNull(); + }); +}); + +describe("PULSE fold — min/max preservation via pre-fold rollup recompute", () => { + it("recomputes the PULSE DAY rollup from RAW rows BEFORE the fold, and NOT after", async () => { + const rows = [ + pulseRow("a", 55, "2026-05-01T06:00:00.000Z"), + pulseRow("b", 70, "2026-05-01T12:00:00.000Z"), + pulseRow("c", 160, "2026-05-01T18:00:00.000Z"), + ]; + const { mock, txCreate } = buildPrismaMock({ + pulseRows: rows, + hasNativeResting: true, // isolate the min/max path from the resting mint + }); + + await runDenseIntradayRetention(mock, { retentionDays: 0, log: () => {} }); + + // The fold minted the daily mean row. + expect(txCreate).toHaveBeenCalledTimes(1); + + // 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. + const pulseRecomputes = recomputeBucketsForMeasurement.mock.calls.filter( + (c) => c[1] === "PULSE", + ); + expect(pulseRecomputes).toHaveLength(1); + }); +}); + +describe("PULSE fold — derived resting row for proxy users", () => { + it("mints a COMPUTED RESTING_HEART_RATE row from the day's 20th percentile when no native resting exists", async () => { + const rows = [ + ...[58, 59, 60, 61, 62].map((v, i) => + pulseRow(`a${i}`, v, "2026-05-01T07:00:00.000Z"), + ), + pulseRow("w", 155, "2026-05-01T18:00:00.000Z"), + ]; + const { mock, upsert } = buildPrismaMock({ + pulseRows: rows, + hasNativeResting: false, + }); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + expect(upsert).toHaveBeenCalledTimes(1); + const arg = upsert.mock.calls[0]?.[0] as { + create: { type: string; source: string; value: number; unit: string }; + where: { + userId_type_source_externalId: { source: string; externalId: string }; + }; + }; + expect(arg.create.type).toBe("RESTING_HEART_RATE"); + // COMPUTED keeps it off both unique indexes Apple's own resting rows use. + expect(arg.create.source).toBe("COMPUTED"); + expect(arg.where.userId_type_source_externalId.source).toBe("COMPUTED"); + expect( + arg.where.userId_type_source_externalId.externalId.startsWith( + "stats:HKQuantityTypeIdentifierRestingHeartRate:", + ), + ).toBe(true); + // Resting value is in the low band, not the workout-polluted mean. + expect(arg.create.value).toBeLessThan(70); + expect(summary.totals.derivedRestingRowsUpserted).toBe(1); + }); + + it("does NOT mint a derived resting row when the user already has native resting (no double-count)", async () => { + const rows = [58, 59, 60, 61, 62].map((v, i) => + pulseRow(`a${i}`, v, "2026-05-01T07:00:00.000Z"), + ); + const { mock, upsert } = buildPrismaMock({ + pulseRows: rows, + hasNativeResting: true, + }); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + expect(upsert).not.toHaveBeenCalled(); + expect(summary.totals.derivedRestingRowsUpserted).toBe(0); + }); + + it("does NOT mint a derived resting row for a day below the sample floor", async () => { + const rows = [ + pulseRow("a", 130, "2026-05-01T18:00:00.000Z"), + pulseRow("b", 135, "2026-05-01T18:05:00.000Z"), + ]; + const { mock, upsert } = buildPrismaMock({ + pulseRows: rows, + hasNativeResting: false, + }); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + expect(upsert).not.toHaveBeenCalled(); + expect(summary.totals.derivedRestingRowsUpserted).toBe(0); + }); + + it("probes native resting once per user (memoised across days)", async () => { + const rows = [ + ...[58, 59, 60, 61, 62].map((v, i) => + pulseRow(`a${i}`, v, "2026-05-01T07:00:00.000Z"), + ), + ...[57, 58, 59, 60, 61].map((v, i) => + pulseRow(`b${i}`, v, "2026-05-02T07:00:00.000Z"), + ), + ]; + const { mock, upsert, topFindFirst } = buildPrismaMock({ + pulseRows: rows, + hasNativeResting: false, + }); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + // Two folded days → two derived resting rows, but the native probe runs + // once (memoised). + expect(summary.totals.derivedRestingRowsUpserted).toBe(2); + expect(upsert).toHaveBeenCalledTimes(2); + expect(topFindFirst).toHaveBeenCalledTimes(1); + }); +}); + +describe("PULSE fold — idempotency + dry-run", () => { + it("converges to zero work when the scan returns no live rows (re-run)", async () => { + const { mock, upsert, txCreate } = buildPrismaMock({ + pulseRows: [], + hasNativeResting: false, + }); + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + expect(txCreate).not.toHaveBeenCalled(); + expect(upsert).not.toHaveBeenCalled(); + expect(summary.totals.daysConsolidated).toBe(0); + expect(summary.totals.derivedRestingRowsUpserted).toBe(0); + }); + + it("writes nothing on a dry-run (no fold, no pre-fold recompute, no resting mint)", async () => { + const rows = [58, 59, 60, 61, 62].map((v, i) => + pulseRow(`a${i}`, v, "2026-05-01T07:00:00.000Z"), + ); + const { mock, upsert, txCreate } = buildPrismaMock({ + pulseRows: rows, + hasNativeResting: false, + }); + await runDenseIntradayRetention(mock, { + retentionDays: 0, + dryRun: true, + log: () => {}, + }); + expect(txCreate).not.toHaveBeenCalled(); + expect(upsert).not.toHaveBeenCalled(); + expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled(); + }); +}); + +describe("PULSE fold — per-day failure boundary", () => { + it("steps over a day whose fold throws and keeps draining the rest", async () => { + const day1 = [58, 59, 60, 61, 62].map((v, i) => + pulseRow(`a${i}`, v, "2026-05-01T07:00:00.000Z"), + ); + const day2 = [57, 58, 59, 60, 61].map((v, i) => + pulseRow(`b${i}`, v, "2026-05-02T07:00:00.000Z"), + ); + const { mock, upsert } = buildPrismaMock({ + pulseRows: [...day1, ...day2], + hasNativeResting: false, + }); + + // Make the FIRST day's fold transaction throw a non-P2002 error; the + // boundary must absorb it and continue to the second day. + let call = 0; + (mock.$transaction as ReturnType).mockImplementation( + async (cb: (t: unknown) => Promise) => { + call += 1; + if (call === 1) throw new Error("boom — poisoned day"); + return cb({ + measurement: { + create: vi.fn().mockResolvedValue({ id: "minted" }), + update: vi.fn(), + findFirst: vi.fn().mockResolvedValue(null), + updateMany: vi.fn().mockResolvedValue({ count: 5 }), + }, + }); + }, + ); + + const summary = await runDenseIntradayRetention(mock, { + retentionDays: 0, + log: () => {}, + }); + + // The second day still folded + minted its resting row despite the first + // day's failure — the boundary isolated the poison. + expect(summary.totals.daysConsolidated).toBe(1); + expect(upsert).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index 057171e3c..b8934410c 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -50,6 +50,53 @@ import { import { runConsolidation, type DayWriteOutcome } from "./consolidation-base"; import { meanBucketValue } from "./consolidate-daily-mean"; import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; +import { percentile } from "@/lib/insights/strain-score"; +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: + * + * 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). + */ /** * The dense-tier types. These are the daytime intra-day signals the Stress @@ -75,6 +122,54 @@ export const DENSE_INTRADAY_RETENTION_DAYS = 14; /** The daily-stats externalId prefix marks an already-collapsed row. */ const DAILY_STATS_PREFIX = "stats:"; +/** + * The HealthKit identifier the derived resting row mints its `stats:` + * externalId under. It is the SAME identifier `RESTING_HEART_RATE` maps to + * in `apple-health-mapping.ts`, so the row reads back as a canonical daily + * resting figure; the `COMPUTED` source keeps it distinct from Apple's own + * `RESTING_HEART_RATE` rows on both unique indexes. + */ +const RESTING_HK_IDENTIFIER = "HKQuantityTypeIdentifierRestingHeartRate"; + +/** + * Percentile of a day's raw PULSE samples used as that day's resting + * estimate. Mirrors `RESTING_PROXY_DAILY_PERCENTILE` in + * `src/lib/analytics/resting-pulse.ts` — the 20th percentile is the day's + * low band (above the rare sleeping-bradycardia outlier, below the bulk of + * waking + workout HR), so it tracks resting HR without being pulled up by a + * dense workout burst. Kept in lockstep with the read-path proxy so the + * resting figure a folded day persists equals the figure the proxy would + * have produced from the now-deleted raw rows. + */ +const RESTING_DERIVE_PERCENTILE = 20; + +/** + * Minimum PULSE samples a day needs before it contributes a derived resting + * row. Mirrors `RESTING_PROXY_MIN_DAILY_SAMPLES` — a one/two-sample day + * cannot distinguish a resting floor from a lone workout reading, so it is + * skipped rather than persisting a workout-level number as "resting". + */ +const RESTING_DERIVE_MIN_DAILY_SAMPLES = 3; + +/** + * Derive the resting estimate for a single folded PULSE day from its raw + * 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`. + */ +export function deriveDailyRestingFromPulse( + rows: readonly PerSampleRow[], +): number | null { + if (rows.length < RESTING_DERIVE_MIN_DAILY_SAMPLES) return null; + return Math.round( + percentile( + rows.map((r) => r.value), + RESTING_DERIVE_PERCENTILE, + ), + ); +} + export interface DenseIntradayRetentionSummary { dryRun: boolean; totals: { @@ -82,6 +177,12 @@ export interface DenseIntradayRetentionSummary { daysConsolidated: number; perSampleRowsSoftDeleted: number; dailyRowsUpserted: number; + /** + * Derived `RESTING_HEART_RATE` rows minted from folded PULSE days for + * proxy users (zero native resting rows). iOS#34 — preserves the + * resting signal the raw-PULSE proxy can no longer compute after fold. + */ + derivedRestingRowsUpserted: number; }; } @@ -132,9 +233,30 @@ export async function runDenseIntradayRetention( daysConsolidated: 0, perSampleRowsSoftDeleted: 0, dailyRowsUpserted: 0, + derivedRestingRowsUpserted: 0, }, }; + // iOS#34 — per-user memo of whether the user has ANY native + // `RESTING_HEART_RATE` row. The read-path resolver + // (`resolveRestingPulseSeries`) is all-or-nothing: a single native + // resting row makes it ignore the PULSE proxy entirely. So the derived + // resting row is minted ONLY for proxy users (zero native rows) — minting + // alongside native rows would double-count. Memoised so the per-day fold + // does not re-query, and cleared per user at the start of the walk. + const nativeRestingByUser = new Map(); + async function userHasNativeResting(userId: string): Promise { + const cached = nativeRestingByUser.get(userId); + if (cached !== undefined) return cached; + const native = await prismaClient.measurement.findFirst({ + where: { userId, type: "RESTING_HEART_RATE", deletedAt: null }, + select: { id: true }, + }); + const has = native !== null; + nativeRestingByUser.set(userId, has); + return has; + } + const { usersScanned } = await runConsolidation({ prismaClient, options: { ...options, cutoffHours }, @@ -179,6 +301,18 @@ export async function runDenseIntradayRetention( sourceRowIds, }): Promise => { const unit = dayRows[0]?.unit ?? "unknown"; + const isPulse = type === "PULSE"; + + // iOS#34 — PULSE min/max preservation. 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 + // PULSE, 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 — and the post-fold recompute is SKIPPED below. (HRV has + // no min/max consumer, so it keeps the simpler post-fold recompute.) + if (isPulse && !options.dryRun) { + await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); + } // Fold the day to one canonical daily-mean row at local-noon. // @@ -319,11 +453,70 @@ export async function runDenseIntradayRetention( removed = await foldOnce(); } + // 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). + if (isPulse) { + const restingValue = deriveDailyRestingFromPulse(dayRows); + if (restingValue !== null && !(await userHasNativeResting(userId))) { + const restingExternalId = dailyStatsExternalId( + RESTING_HK_IDENTIFIER, + // The fold's canonical timestamp is local-noon of the day; reuse + // its UTC date for the resting row's `stats:` key so a re-run + // upserts the same row instead of minting a sibling. + canonicalTimestamp.toISOString().slice(0, 10), + ); + await pc.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type: "RESTING_HEART_RATE", + source: "COMPUTED", + externalId: restingExternalId, + }, + }, + create: { + userId, + type: "RESTING_HEART_RATE", + value: restingValue, + unit: "bpm", + source: "COMPUTED", + measuredAt: canonicalTimestamp, + externalId: restingExternalId, + }, + update: { value: restingValue, deletedAt: null }, + }); + summary.totals.derivedRestingRowsUpserted += 1; + // The derived row is its own (type, day); recompute its DAY bucket + // so the rollup tier serves it on a covered read. + await recomputeBucketsForMeasurement( + userId, + "RESTING_HEART_RATE", + canonicalTimestamp, + ); + } + } + // 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. - await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); + // + // iOS#34 — for PULSE this would COLLAPSE the true intraday min/max the + // pre-fold recompute above just captured (the single folded row makes + // min == max == mean), so the PULSE 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 (!isPulse) { + await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); + } return { kind: "written", sourceRowsRemoved: removed }; }, @@ -340,12 +533,26 @@ export async function runDenseIntradayRetention( `[dense-intraday-retention] user=${userId} tz=${tz}${dryRun ? " (dry-run)" : ""}`, ); }, + // Per-day failure boundary (v1.18.10 lesson). A day whose fold throws — + // e.g. a residual unique-index collision the adopt-in-place path could + // not absorb, or a derived-resting upsert that races a native row — is + // logged and STEPPED OVER so the global walk keeps draining every other + // user / type / day. The failed day keeps its raw rows live (the fold + // soft-deletes only on a committed transaction), so the next nightly run + // retries it. Without this, one poisoned day aborted the whole walk and + // stranded every later account. + onBucketError: ({ userId, type, dateKey, error }) => { + const reason = error instanceof Error ? error.message : String(error); + log( + `[dense-intraday-retention] user=${userId} type=${type} day=${dateKey} failed — ${reason}; continuing with the next bucket`, + ); + }, }); 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}${options.dryRun ? " (dry-run)" : ""}`, + `[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)" : ""}`, ); return summary;