mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 21:10:59 +00:00
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.
This commit is contained in:
Marc-André Bombeck committed
1 parent
6975849bec
commit
8f1e3c5d44
5 files changed
+578
-7
No files matched your search
@@ -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/);
|
||||
});
|
||||
});
|
||||
@@ -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,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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<unknown>) => 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<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(),
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<string, boolean>();
|
||||
async function userHasNativeResting(userId: string): Promise<boolean> {
|
||||
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<MeasurementType>({
|
||||
prismaClient,
|
||||
options: { ...options, cutoffHours },
|
||||
@@ -179,6 +301,18 @@ export async function runDenseIntradayRetention(
|
||||
sourceRowIds,
|
||||
}): Promise<DayWriteOutcome> => {
|
||||
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;
|
||||
|
||||
Reference in new issue
Block a user