diff --git a/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts b/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts index 79f28b182..3b6bde237 100644 --- a/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts +++ b/src/app/api/measurements/batch/__tests__/aggregated-hr-wire.test.ts @@ -409,3 +409,89 @@ describe("POST /api/measurements/batch — stats tombstone resurrection", () => expect(prisma.measurement.createMany).not.toHaveBeenCalled(); }); }); + +/** + * Two entries carrying the SAME `stats:` key inside ONE batch. Neither is in + * the DB yet, so before the fix both missed the existing-row probe, both + * landed in the bulk insert, and `skipDuplicates` kept whichever the unique + * index saw first — the OLDER snapshot won and the newer, larger figure was + * dropped silently while both entries still reported `inserted`. + * + * `stats:` rows are overwrite-by-contract, so the LAST entry for a key wins. + */ +describe("POST /api/measurements/batch — intra-batch `stats:` supersession", () => { + it("keeps only the LAST entry when one batch carries the same stats: key twice", async () => { + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ count: 1 }); + + const res = await POST( + makeRequest({ + entries: [ + hrBucketEntry(HR_BUCKET_ID, 70), // earlier snapshot of a filling bucket + hrBucketEntry(HR_BUCKET_ID, 78), // later, authoritative snapshot + ], + }), + ); + expect(res.status).toBe(200); + const { data } = await readJson(res); + + // Exactly ONE row reaches the insert, carrying the NEWER value. + const createArg = vi.mocked(prisma.measurement.createMany).mock.calls[0][0]; + const rows = (createArg as { data: { value: number }[] }).data; + expect(rows).toHaveLength(1); + expect(rows[0].value).toBe(78); + + // The superseded entry is reported honestly, not as a phantom insert. + expect(data.entries[0].status).toBe("duplicate"); + expect(data.entries[0].reason).toBe("superseded_in_batch"); + expect(data.entries[1].status).toBe("inserted"); + expect(data.inserted).toBe(1); + expect(data.duplicates).toBe(1); + }); + + it("overwrites an existing row exactly once, with the LAST value in the batch", async () => { + // The key is already stored, so both entries take the overwrite branch. + // Only the authoritative last snapshot may reach the DB. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { type: "PULSE", source: "APPLE_HEALTH", externalId: HR_BUCKET_ID }, + ] as never); + + const res = await POST( + makeRequest({ + entries: [ + hrBucketEntry(HR_BUCKET_ID, 70), + hrBucketEntry(HR_BUCKET_ID, 78), + ], + }), + ); + expect(res.status).toBe(200); + const { data } = await readJson(res); + + expect(prisma.measurement.updateMany).toHaveBeenCalledTimes(1); + const updateArg = vi.mocked(prisma.measurement.updateMany).mock.calls[0][0]; + expect((updateArg as { data: { value: number } }).data.value).toBe(78); + + expect(data.entries[0].status).toBe("duplicate"); + expect(data.entries[0].reason).toBe("superseded_in_batch"); + expect(data.entries[1].status).toBe("updated"); + expect(data.updated).toBe(1); + }); + + it("leaves DISTINCT stats: keys in one batch untouched", async () => { + const OTHER = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:10:00.000Z"; + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ count: 2 }); + + const res = await POST( + makeRequest({ + entries: [hrBucketEntry(HR_BUCKET_ID, 70), hrBucketEntry(OTHER, 78)], + }), + ); + expect(res.status).toBe(200); + const { data } = await readJson(res); + + const createArg = vi.mocked(prisma.measurement.createMany).mock.calls[0][0]; + expect((createArg as { data: unknown[] }).data).toHaveLength(2); + expect(data.inserted).toBe(2); + expect(data.duplicates).toBe(0); + }); +}); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index 1bdce805b..96a4e60e6 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -480,10 +480,45 @@ async function postBatch(request: NextRequest): Promise { return false; } + // Intra-batch `stats:` supersession. Two entries carrying the same + // (type, source, externalId) `stats:` key inside ONE batch both miss + // `existingSet` — neither is in the DB yet — so the overwrite branch + // never runs, both land in `toInsert`, and + // `createMany({ skipDuplicates: true })` keeps whichever row the unique + // index saw first. The OLDER snapshot of a per-day cumulative total (or + // of a still-filling ten-minute heart-rate bucket) therefore won and the + // newer, larger figure was dropped silently: both entries reported + // `inserted`, the counters still summed, and the client checkpointed + // past both. `stats:` rows carry overwrite semantics by contract — a + // re-post replaces value / unit / measuredAt — so within one batch the + // LAST entry for a key is the authoritative one. Resolve it here and + // mark the superseded earlier entries `duplicate`, the same status the + // client already checkpoints past for a composite-key duplicate. + const lastStatsIndexByKey = new Map(); + for (const p of prepared) { + if (!isStatsExternalId(p.row.externalId as string)) continue; + lastStatsIndexByKey.set( + `${p.row.type}::${p.row.source}::${p.row.externalId}`, + p.index, + ); + } + const toInsert: Prisma.MeasurementCreateManyInput[] = []; const toOverwrite: Prepared[] = []; for (const p of prepared) { const key = `${p.row.type}::${p.row.source}::${p.row.externalId}`; + if ( + isStatsExternalId(p.row.externalId as string) && + lastStatsIndexByKey.get(key) !== p.index + ) { + results[p.index] = { + index: p.index, + status: "duplicate", + reason: "superseded_in_batch", + }; + duplicateCount += 1; + continue; + } if (existingSet.has(key)) { // v1.5.0 issue #213 — per-day cumulative `stats:*` rows are // intentionally overwritten on a re-post so today's tile reflects diff --git a/src/lib/analytics/__tests__/resting-pulse.test.ts b/src/lib/analytics/__tests__/resting-pulse.test.ts index af41b0e4d..af7e7fc6b 100644 --- a/src/lib/analytics/__tests__/resting-pulse.test.ts +++ b/src/lib/analytics/__tests__/resting-pulse.test.ts @@ -187,6 +187,79 @@ describe("resolveRestingPulseSeries", () => { resolveRestingPulseSeries({ restingSamples: [], pulseSamples: [] }), ).toEqual({ series: [], which: "none" }); }); + + /** + * The shape a proxy account actually has once the retention fold has run + * for a while: derived RESTING_HEART_RATE rows for the OLD days the fold + * has already processed, and nothing but raw PULSE for the RECENT days it + * has not reached yet. The all-or-nothing resolver saw the derived rows, + * took the resting branch, and dropped the proxy — so the series the user + * saw stopped dead at the fold horizon. + */ + it("carries the series past the fold horizon — derived rows for old days, PULSE for recent ones", () => { + // Days 01-05: folded, one derived resting row each. No PULSE left — + // the fold tombstoned the raw readings it derived them from. + const restingSamples: PulseSample[] = [ + day("2026-06-01T04:00:00", 61), + day("2026-06-02T04:00:00", 62), + day("2026-06-03T04:00:00", 63), + day("2026-06-04T04:00:00", 64), + day("2026-06-05T04:00:00", 65), + ]; + // Days 06-08: not yet folded. Raw PULSE only, no resting row at all. + const pulseSamples: PulseSample[] = []; + for (const d of ["06", "07", "08"]) { + pulseSamples.push( + day(`2026-06-${d}T07:00:00`, 66), + day(`2026-06-${d}T07:10:00`, 68), + day(`2026-06-${d}T07:20:00`, 70), + day(`2026-06-${d}T18:00:00`, 155), // workout burst, must not win + ); + } + + const { series, which } = resolveRestingPulseSeries({ + restingSamples, + pulseSamples, + }); + + // Five folded days + three proxy days — the recent days MUST appear. + expect(series).toHaveLength(8); + const days = series.map((p) => + p.measuredAt.toLocaleDateString("en-CA", { timeZone: PROXY_DAY_ZONE }), + ); + expect(days).toEqual([ + "2026-06-01", + "2026-06-02", + "2026-06-03", + "2026-06-04", + "2026-06-05", + "2026-06-06", + "2026-06-07", + "2026-06-08", + ]); + // The proxy days track the resting floor, not the workout burst. + for (const point of series.slice(5)) { + expect(point.value).toBeLessThan(100); + } + // Mixed series — hedge the label rather than call an estimate native. + expect(which).toBe("proxy"); + }); + + it("keeps the resting row on a day that has both a resting row and PULSE", () => { + // A native day must not be second-guessed by the proxy, and a native + // account with no gap days keeps the honest 'resting' label. + const { series, which } = resolveRestingPulseSeries({ + restingSamples: [day("2026-06-01T04:00:00", 58)], + pulseSamples: [ + day("2026-06-01T07:00:00", 80), + day("2026-06-01T07:10:00", 82), + day("2026-06-01T07:20:00", 84), + ], + }); + expect(series).toHaveLength(1); + expect(series[0].value).toBe(58); + expect(which).toBe("resting"); + }); }); describe("deriveRestingProxyFromPulse — the day bucket is the PROFILE day", () => { diff --git a/src/lib/analytics/intraday-pulse-io.ts b/src/lib/analytics/intraday-pulse-io.ts index 8a063a21d..9a4df4243 100644 --- a/src/lib/analytics/intraday-pulse-io.ts +++ b/src/lib/analytics/intraday-pulse-io.ts @@ -68,6 +68,13 @@ export interface IntradayPulseResult { resolution: IntradayResolution; } +/** + * How old the newest RESTING_HEART_RATE row may be before the baseline stops + * trusting the resting rows alone and pulls the PULSE history in as well. Two + * days tolerates a single missed night without widening the read. + */ +const RESTING_STALE_AFTER_MS = 2 * 24 * 60 * 60 * 1000; + /** Median (p50), or null on an empty series. */ function median(values: number[]): number | null { if (values.length === 0) return null; @@ -103,9 +110,21 @@ async function resolveBaseline( select: { value: true, measuredAt: true }, }); - // Only reach for the heavier PULSE-history read when resting rows are thin. + // Only reach for the heavier PULSE-history read when the resting rows + // cannot carry the baseline on their own: too few of them, OR the newest + // one is already stale. Staleness matters because the retention fold mints + // derived resting rows only for the days it has ALREADY folded, so a proxy + // account accumulates plenty of resting rows for OLD days while the RECENT + // days still hold nothing but raw PULSE. Gating on the row count alone + // pinned the baseline to the fold horizon and never advanced it past it. + // A genuinely native account reports a resting row daily, never trips the + // staleness arm, and pays for no extra read. + const newestRestingAt = restingRows[0]?.measuredAt ?? null; + const restingIsStale = + newestRestingAt === null || + Date.now() - newestRestingAt.getTime() > RESTING_STALE_AFTER_MS; const pulseHistory = - restingRows.length >= MIN_BASELINE_DAYS + restingRows.length >= MIN_BASELINE_DAYS && !restingIsStale ? [] : await prisma.measurement.findMany({ where: { userId, type: "PULSE", deletedAt: null }, diff --git a/src/lib/analytics/resting-pulse.ts b/src/lib/analytics/resting-pulse.ts index 5e62139c4..652079dae 100644 --- a/src/lib/analytics/resting-pulse.ts +++ b/src/lib/analytics/resting-pulse.ts @@ -19,10 +19,12 @@ * * The fix: judge a RESTING band against a RESTING series. * - * 1. **Prefer `RESTING_HEART_RATE`** when the user has it (Apple + * 1. **Prefer `RESTING_HEART_RATE`** on every day that has one (Apple * already separated the clean signal for us). - * 2. **Fallback proxy from `PULSE`** when no `RESTING_HEART_RATE` - * exists (manual-only users, older imports): take a robust + * 2. **Fallback proxy from `PULSE`** on the days that have no + * `RESTING_HEART_RATE` (manual-only users, older imports, and the + * recent days of an account whose derived rows only reach as far + * as the retention fold has run): take a robust * low-percentile of each calendar day's PULSE samples — the resting * heart rate is the floor of the day, not its average, so the daily * 20th percentile estimates resting while excluding the workout @@ -112,11 +114,27 @@ export function deriveRestingProxyFromPulse( /** * Resolve the resting-pulse series a resting-target surface should - * score. Prefers the clean `RESTING_HEART_RATE` rows; falls back to the - * low-percentile PULSE proxy when the user has no resting rows. + * score. Merges PER DAY: a day that has a `RESTING_HEART_RATE` row is + * represented by that row; a day that has none falls back to the + * low-percentile PULSE proxy. * - * `which` reports the branch taken so callers can annotate / label - * honestly ("resting heart rate" vs "estimated from heart rate"). + * v1.30.16 made the dense-intraday retention fold mint derived + * `RESTING_HEART_RATE` rows (source `COMPUTED`) for the days it folds. + * A proxy account therefore now carries derived rows for OLD folded + * days and nothing but raw `PULSE` for RECENT not-yet-folded days. The + * previous all-or-nothing shape returned the resting branch the moment + * a single resting row existed and never consulted the proxy again, so + * the visible series stopped dead at the fold horizon. Merging per day + * keeps the clean signal where it exists and carries the series + * forward across the days that only have the proxy. + * + * `which` reports how to LABEL the series, and it is deliberately + * conservative: `"resting"` only when EVERY point came from a real + * resting row, `"proxy"` as soon as one point is proxy-derived. The two + * error directions are not symmetric — calling a partly-estimated + * series "resting heart rate" is a false statement about the estimated + * days, whereas calling a partly-native series "estimated from heart + * rate" merely hedges. A labelling signal should hedge, never lie. */ export function resolveRestingPulseSeries(input: { restingSamples: ReadonlyArray; @@ -124,13 +142,24 @@ export function resolveRestingPulseSeries(input: { /** Optional day-key for the proxy buckets (defaults to Berlin-day). */ dayKeyOf?: (d: Date) => string; }): { series: PulseSample[]; which: "resting" | "proxy" | "none" } { - if (input.restingSamples.length > 0) { - const series = [...input.restingSamples].sort( - (a, b) => a.measuredAt.getTime() - b.measuredAt.getTime(), - ); - return { series, which: "resting" }; + const dayKeyOf = input.dayKeyOf ?? toBerlinDayKey; + const byTime = (a: PulseSample, b: PulseSample) => + a.measuredAt.getTime() - b.measuredAt.getTime(); + + const resting = [...input.restingSamples].sort(byTime); + // Days already covered by a real resting row — the proxy must not + // second-guess them. + const restingDays = new Set(resting.map((s) => dayKeyOf(s.measuredAt))); + const proxy = deriveRestingProxyFromPulse( + input.pulseSamples, + dayKeyOf, + ).filter((p) => !restingDays.has(dayKeyOf(p.measuredAt))); + + if (resting.length === 0) { + return proxy.length > 0 + ? { series: proxy, which: "proxy" } + : { series: [], which: "none" }; } - const proxy = deriveRestingProxyFromPulse(input.pulseSamples, input.dayKeyOf); - if (proxy.length > 0) return { series: proxy, which: "proxy" }; - return { series: [], which: "none" }; + if (proxy.length === 0) return { series: resting, which: "resting" }; + return { series: [...resting, ...proxy].sort(byTime), which: "proxy" }; } diff --git a/src/lib/insights/__tests__/pulse-status.test.ts b/src/lib/insights/__tests__/pulse-status.test.ts index e267d9fd2..1aed06d5c 100644 --- a/src/lib/insights/__tests__/pulse-status.test.ts +++ b/src/lib/insights/__tests__/pulse-status.test.ts @@ -124,9 +124,18 @@ describe("generatePulseStatusForUser — A2 resting-target in-target %", () => { measuredAt: new Date(now.getTime() - (i % 30) * dayMs), }); } - // Clean resting series, comfortably inside a 60-100 band. + // Clean resting series, comfortably inside a 60-100 band, covering the + // SAME 30-day span as the PULSE above. The span has to match for the + // fixture to mean what the test name claims: the resolver merges per + // day, so a day with no resting row of its own is estimated from that + // day's PULSE rather than dropped. Leaving ten of the thirty days + // resting-free made this assert the merge's gap-fill behaviour by + // accident — on days whose only readings are a 150 bpm workout, the + // honest estimate IS out of band. Gap-day behaviour is pinned directly + // in `resting-pulse.test.ts`; this case is about preferring the clean + // signal over workout HR where both exist. const restingRecords: Array<{ value: number; measuredAt: Date }> = []; - for (let d = 0; d < 20; d++) { + for (let d = 0; d < 30; d++) { restingRecords.push({ value: 72, measuredAt: new Date(now.getTime() - d * dayMs), diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index 2bc1a9455..515f8ea19 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -423,12 +423,15 @@ export async function runDenseIntradayRetention( }; // 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. + // `RESTING_HEART_RATE` row. The derived resting row is minted ONLY for + // proxy users (zero native rows): a native account already reports the + // clean signal for those days, so a derived row would add nothing and + // would let a fold artefact outrank a device reading. The read-path + // resolver (`resolveRestingPulseSeries`) merges resting and proxy PER DAY + // and prefers a resting row wherever one exists, so the mint's job is + // narrow — preserve the day's figure before the fold tombstones the raw + // readings it was derived from. 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); diff --git a/src/lib/rollups/__tests__/measurement-rollups.test.ts b/src/lib/rollups/__tests__/measurement-rollups.test.ts index 21d23ab4b..d8a7a2747 100644 --- a/src/lib/rollups/__tests__/measurement-rollups.test.ts +++ b/src/lib/rollups/__tests__/measurement-rollups.test.ts @@ -487,6 +487,70 @@ describe("persistRollupRows — large path (via recomputeUserRollups)", () => { expect(deleteMany).toHaveBeenCalledTimes(1); }); + /** + * The aggregate filters measurements by `measured_at` but GROUPS them by + * `date_trunc`, and `persistRollupRows` then deletes every rollup row in + * the bucket range it writes. A `from` that lands mid-bucket therefore + * replaced a COMPLETE bucket with an aggregate built only from the + * readings after that instant — and because every caller passes a moving + * instant, a different oldest bucket was corrupted on each run and nothing + * repaired it. The window must be snapped out to whole buckets first. + */ + describe("bucket alignment of the recompute window", () => { + it("snaps a mid-day `from` down to the start of its DAY bucket", async () => { + queryRawUnsafe.mockResolvedValueOnce(syntheticRows(1)); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + // 14:37:12Z on 2024-03-10 — squarely inside a day bucket. + const midDay = new Date(Date.UTC(2024, 2, 10, 14, 37, 12, 500)); + await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from: midDay, + to: new Date(Date.UTC(2024, 2, 20, 9, 5, 0)), + }); + + // runRollupAggregate → $queryRawUnsafe(sql, userId, from, to) + const [, userId, fromArg, toArg] = queryRawUnsafe.mock.calls[0]; + expect(userId).toBe("user-1"); + // Whole-bucket lower edge — the morning of the 10th must be included. + expect((fromArg as Date).toISOString()).toBe("2024-03-10T00:00:00.000Z"); + // Upper edge snaps out to the end of the closing bucket. + expect((toArg as Date).toISOString()).toBe("2024-03-21T00:00:00.000Z"); + }); + + it("snaps a mid-month `from` back to the 1st for MONTH granularity", async () => { + queryRawUnsafe.mockResolvedValueOnce(syntheticRows(1)); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + await recomputeUserRollups("user-1", { + granularities: ["MONTH"], + from: new Date(Date.UTC(2024, 4, 17, 8, 0, 0)), + to: new Date(Date.UTC(2024, 6, 3, 0, 0, 0)), + }); + + const [, , fromArg, toArg] = queryRawUnsafe.mock.calls[0]; + expect((fromArg as Date).toISOString()).toBe("2024-05-01T00:00:00.000Z"); + expect((toArg as Date).toISOString()).toBe("2024-08-01T00:00:00.000Z"); + }); + + it("leaves an already-aligned window untouched", async () => { + queryRawUnsafe.mockResolvedValueOnce(syntheticRows(1)); + txExecuteRawUnsafe.mockResolvedValueOnce(1); + + const from = new Date(Date.UTC(2024, 2, 10)); + const to = new Date(Date.UTC(2024, 2, 20)); + await recomputeUserRollups("user-1", { + granularities: ["DAY"], + from, + to, + }); + + const [, , fromArg, toArg] = queryRawUnsafe.mock.calls[0]; + expect((fromArg as Date).getTime()).toBe(from.getTime()); + expect((toArg as Date).getTime()).toBe(to.getTime()); + }); + }); + it("the ≤500-row path runs through the same interactive upsert transaction", async () => { queryRawUnsafe.mockResolvedValueOnce(syntheticRows(2)); txExecuteRawUnsafe.mockResolvedValueOnce(2); diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index b4f4727ca..519750ed2 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -190,12 +190,32 @@ export async function recomputeUserRollups( let rowsUpserted = 0; for (const granularity of granularities) { + // Snap the window out to whole buckets before aggregating. The + // aggregate filters rows by `measured_at` but GROUPS by + // `date_trunc(, measured_at)`, and `persistRollupRows` + // then deletes every rollup row in the [min, max] bucket range it is + // about to write. A `from` that lands mid-bucket therefore replaced a + // COMPLETE bucket with an aggregate computed only from the readings + // after that instant — and because the callers pass a moving instant + // (`Date.now() - 90d` on the read-path refresh, an arbitrary + // client-supplied instant on the coverage fallback, `now - 5y` on the + // boot backfill and the restore path), a different oldest day was + // corrupted on every invocation and nothing ever repaired it. Same + // hazard on the `to` edge for any row measured later in the closing + // bucket. `bucketSpan` is the canonical truncator — its own contract + // note already required every recompute to cover whole spans; this is + // the caller that did not. + const alignedFrom = bucketSpan(from, granularity).from; + const alignedTo = + to > alignedFrom + ? bucketSpan(new Date(to.getTime() - 1), granularity).to + : bucketSpan(from, granularity).to; const rows = await runRollupAggregate({ userId, types: opts.types, granularity, - from, - to, + from: alignedFrom, + to: alignedTo, }); if (rows.length === 0) continue;