diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index 7cf3c7943..061f49718 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -103,7 +103,9 @@ const WRITE_SITES: Record = { "lib/jobs/compaction-tombstone-purge.ts": "hard-deletes compaction tombstones, rows already soft-deleted and read nowhere", "lib/jobs/measurement-fold-repair.ts": - "corrects the value of a folded stats: mean in place and refolds the DAY rollup of a corrected daily mean itself; no instant, type or live set changes", + "corrects the value of a folded stats: mean in place, takes the window's live samples into it, and refolds the DAY rollup of a touched daily mean itself; no instant or type changes", + "lib/measurements/fold-constituents.ts": + "soft-deletes the samples a fold or the fold repair has just taken into a recomputed stats: mean, which accounts for them from then on", }; function sourceFiles(): string[] { diff --git a/src/lib/export/restore-lock.ts b/src/lib/export/restore-lock.ts index e57c979b3..991ea6f63 100644 --- a/src/lib/export/restore-lock.ts +++ b/src/lib/export/restore-lock.ts @@ -78,3 +78,25 @@ export async function holdAccountAgainstRestore( `; if (!row?.held) throw new AccountRestoreInProgressError(userId); } + +/** + * Serialise the passes that rewrite an account's `stats:` means from their + * samples: the daily-mean consolidation, the dense hourly fold and the + * one-time fold repair (`measurement-fold-repair.ts`). Each day transaction + * of those passes takes it right after {@link holdAccountAgainstRestore}, and + * reads the day's samples only after it has it, so two of them never compute + * the same window from two different views of its samples, and never write + * the same row at once. It waits rather than refusing: the other side holds + * it for one day's transaction. + * + * The two-key form keeps it apart from the restore lock's single-key space. + */ +export async function holdAccountFoldLock( + tx: LockClient, + userId: string, +): Promise { + await tx.$queryRaw` + SELECT 1 AS locked + FROM pg_advisory_xact_lock(hashtext('measurement-fold'), hashtext(${userId})) + `; +} diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 5ed38d850..8f863f1b6 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -143,6 +143,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ // with the mean over every sample, and those it corrected. Counts only. "means_checked", "means_corrected", + // Windows before the horizon it left as stored, and live samples it took + // into a checked mean. + "means_samples_absorbed", + "means_skipped_beyond_horizon", "mean_rows_soft_deleted", "measurements_imported", "measurements_migrated", diff --git a/src/lib/jobs/mean-consolidation.ts b/src/lib/jobs/mean-consolidation.ts index 7c787755a..6e9cef9bf 100644 --- a/src/lib/jobs/mean-consolidation.ts +++ b/src/lib/jobs/mean-consolidation.ts @@ -65,6 +65,7 @@ export async function runMeanConsolidationForUser( days: summary.totals.daysConsolidated, per_sample_rows_soft_deleted: summary.totals.perSampleRowsSoftDeleted, daily_rows_upserted: summary.totals.dailyRowsUpserted, + days_left_as_stored: summary.totals.daysLeftAsStored, stopped_early: summary.stoppedEarly, }, }, diff --git a/src/lib/jobs/measurement-fold-repair.ts b/src/lib/jobs/measurement-fold-repair.ts index 9fa1019e3..86b7f625e 100644 --- a/src/lib/jobs/measurement-fold-repair.ts +++ b/src/lib/jobs/measurement-fold-repair.ts @@ -8,25 +8,51 @@ * later part alone: every mean-type day on the night run, and the boundary * hour of every dense day (heart rate, HRV, SpO2), plus the derived resting * figure of a folded heart-rate day. The folds soft-deleted the samples they - * took, so for the last 75 days the samples of both runs are still in the - * table. This pass recomputes each such mean from all of them. + * took, and a tombstone is kept for 75 days. This pass recomputes such a mean + * from all of its samples, and only where all of them are provably still in + * the table. * - * Per account, per folded type and source, day by day in `measuredAt` order: - * - read the day's raw samples: the live ones and the ones a fold - * soft-deleted (`fold-constituents.ts`; a person's deletions are left - * out); - * - a day without fold leftovers has nothing to repair and is passed over; - * - for each hour (dense) or the day (mean) with a live `stats:` row, the - * mean over every sample is compared with the stored value, and only a - * different value is written; + * The horizon. A day is repaired only when it starts at or after + * `foldConstituentHorizon` (`fold-constituents.ts`): every fold tombstone of + * such a day is younger than the retention, so the live samples plus the + * fold leftovers are the whole day. For the daily means that is the last 75 + * days; for the dense hourly means, folded 90 days after the fact, the 75 + * days before the raw window. An older day may hold only a fragment of its + * samples (a re-upload after an anchor reset, the rows of one later run, the + * v1.28.31 rebuild), and a mean over those can be far off a stored mean that + * was computed from the full day. Older days are left exactly as they are and + * counted as `windowsSkippedBeyondHorizon`. + * + * Per account, per folded type and source, day by day in `measuredAt` order, + * over the local days that ended before the fold boundary (`foldBoundary`; a + * day the fold has not finished is the regular fold's to complete): + * - the scan finds the days with fold leftovers (`fold-constituents.ts`; a + * person's deletions are left out); a day without any has nothing to + * repair and is passed over; + * - inside the day's transaction, under the fold lock, the day's live raw + * samples and its fold leftovers are read again; + * - for each hour (dense) or the day (mean) with a live `stats:` row and at + * least one fold leftover, the mean over every sample is compared with the + * stored value, and only a different value is written. A window without + * leftovers is left alone: its samples in hand are not all of it; + * - the live samples of every window it checked are soft-deleted as fold + * leftovers (`absorbIntoFold`): they are now part of the stored mean, and + * a second run then computes the same mean from the same samples; * - for an Apple Health heart-rate day, the derived resting row is * recomputed from the whole day the same way, when one exists. * - * Each day commits in its own transaction under the restore lock. The pass - * holds one page of rows and one day at a time. It stops between days when the - * job's budget runs out and sends a follow-up that resumes at the next day; an - * account under restore is left and retried a few minutes later. Writing only - * values that differ makes a repeated run a no-op, so the pass is idempotent. + * Concurrency. The day transaction takes the restore lock and then the fold + * lock (`restore-lock.ts`), which the daily-mean consolidation and the dense + * fold take too, and reads the day's samples only after that. A fold and the + * repair therefore never work on one account's means at the same moment, and + * each sees what the other committed: samples one of them took into a mean + * are leftovers for the other, never counted twice and never missing. + * + * The pass holds one page of rows and one day at a time. It stops between + * days when the job's budget runs out and sends a follow-up that resumes at + * the next day; an account under restore is left and retried a few minutes + * later. Writing only values that differ makes a repeated run a no-op, so + * the pass is idempotent. * * When an account is through, a `MeasurementFoldRepair` row records it. The * compaction-tombstone purge deletes exactly the tombstones this pass reads, @@ -34,7 +60,9 @@ * purge run as it finishes. * * Reported per run: one outcome per account (`reportJobRun`) and, per type, - * how many windows were checked and corrected. Counts only, never a value. + * how many windows were checked, corrected and left beyond the horizon, and + * how many live samples were taken into a checked mean. Counts only, never a + * value. * * The queue name MUST be registered in `allQueues` in * `src/lib/jobs/reminder/register-maintenance.ts`, or the sends silently @@ -52,6 +80,7 @@ import { prisma as defaultPrisma } from "@/lib/db"; import { AccountRestoreInProgressError, holdAccountAgainstRestore, + holdAccountFoldLock, } from "@/lib/export/restore-lock"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { @@ -77,6 +106,7 @@ import { import { CONSOLIDATION_GRACE_CUTOFF_HOURS, canonicalDailyTimestamp, + foldBoundary, hourOfDayForUserTz, } from "@/lib/measurements/consolidation-tz"; import { @@ -87,9 +117,14 @@ import { hourlyStatsExternalId, } from "@/lib/measurements/dense-intraday-retention"; import { + absorbIntoFold, + foldConstituentHorizon, isFoldLeftover, + loadFoldLeftovers, + loadLiveSamples, meanDiffers, meanOf, + type FoldConstituent, } from "@/lib/measurements/fold-constituents"; import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; import { localDayWindow } from "@/lib/tz/local-day"; @@ -158,6 +193,24 @@ export interface FoldRepairTypeCounts { windowsCorrected: number; /** Derived resting rows corrected (heart rate only). */ restingCorrected: number; + /** + * Hours or days with fold leftovers on a day before the horizon, left as + * they are (see the header). + */ + windowsSkippedBeyondHorizon: number; + /** Live samples taken into a checked mean and kept as fold leftovers. */ + samplesAbsorbed: number; +} + +/** All-zero counts. */ +export function emptyFoldRepairCounts(): FoldRepairTypeCounts { + return { + windowsChecked: 0, + windowsCorrected: 0, + restingCorrected: 0, + windowsSkippedBeyondHorizon: 0, + samplesAbsorbed: 0, + }; } export interface FoldRepairResult { @@ -165,7 +218,7 @@ export interface FoldRepairResult { /** Where the next run starts, when `status` is `stopped`. */ resume?: FoldRepairResume; byType: Record; - /** Days with fold leftovers the pass looked at. */ + /** Days with fold leftovers inside the horizon the pass looked at. */ daysWithLeftovers: number; } @@ -185,7 +238,7 @@ function countsFor( ): FoldRepairTypeCounts { let counts = result.byType[type]; if (!counts) { - counts = { windowsChecked: 0, windowsCorrected: 0, restingCorrected: 0 }; + counts = emptyFoldRepairCounts(); result.byType[type] = counts; } return counts; @@ -203,6 +256,8 @@ export async function repairFoldedMeans( resume?: FoldRepairResume; shouldStop?: () => boolean; pageSize?: number; + /** The instant the horizon and the fold boundary are read against. */ + now?: Date; } = {}, ): Promise { const result: FoldRepairResult = { @@ -218,6 +273,7 @@ export async function repairFoldedMeans( const tz = resolveUserTimezone(user.timezone); const shouldStop = options.shouldStop ?? (() => false); const firstStep = options.resume?.step ?? 0; + const now = options.now ?? new Date(); for ( let stepIndex = firstStep; @@ -240,11 +296,10 @@ export async function repairFoldedMeans( select: { id: true }, }); if (!anyStats) continue; - // No fold ever reached past `now - threshold`, so neither its tombstones - // nor a day it folded in part lie later than the day after that; the - // live samples after it (most of the dense tier's 90 raw days) are not - // read. - const upper = new Date(Date.now() - step.foldedAfterMs + 2 * DAY_MS); + // Only the local days the fold has finished: a day after the boundary + // is the regular fold's to complete, from the same samples. + const upper = foldBoundary(now, step.foldedAfterMs, tz); + const horizon = foldConstituentHorizon(now, step.foldedAfterMs); const after = stepIndex === firstStep && options.resume?.after ? new Date(options.resume.after) @@ -289,19 +344,22 @@ export async function repairFoldedMeans( isFoldLeftover(row, step.foldedAfterMs), ); if (leftovers.length === 0) continue; + const window = localDayWindow(dateKey, tz); + if (window.dayStart < horizon) { + countsFor(result, step.type).windowsSkippedBeyondHorizon += new Set( + leftovers.map((row) => windowId(step, hk, dateKey, row, tz)), + ).size; + continue; + } result.daysWithLeftovers += 1; - const constituents = [ - ...dayRows.filter((row) => row.deletedAt === null), - ...leftovers, - ]; try { await repairDay(prisma, { userId, tz, dateKey, + window, hk, step, - constituents, counts: countsFor(result, step.type), }); } catch (err) { @@ -320,36 +378,40 @@ export async function repairFoldedMeans( return result; } -/** Recompute one day's means from its constituents; write only differences. */ +/** The `stats:` id of the window a sample belongs to. */ +function windowId( + step: RepairStep, + hk: string, + dateKey: string, + row: { measuredAt: Date }, + tz: string, +): string { + return step.grain === "day" + ? dailyStatsExternalId(hk, dateKey) + : hourlyStatsExternalId( + hk, + dateKey, + hourOfDayForUserTz(row.measuredAt, tz), + ); +} + +/** + * Recompute one day's means from all of its samples; write only differences. + * The samples are read inside the transaction, under the fold lock. + */ async function repairDay( prisma: PrismaClient, input: { userId: string; tz: string; dateKey: string; + window: { dayStart: Date; dayEnd: Date }; hk: string; step: RepairStep; - constituents: ScanRow[]; counts: FoldRepairTypeCounts; }, ): Promise { - const { userId, tz, dateKey, hk, step, constituents, counts } = input; - - // Window id → the values of every sample in it. - const byWindow = new Map(); - for (const row of constituents) { - const id = - step.grain === "day" - ? dailyStatsExternalId(hk, dateKey) - : hourlyStatsExternalId( - hk, - dateKey, - hourOfDayForUserTz(row.measuredAt, tz), - ); - const values = byWindow.get(id) ?? []; - values.push(row.value); - byWindow.set(id, values); - } + const { userId, tz, dateKey, window, hk, step, counts } = input; const restingExternalId = step.type === "PULSE" && step.source === "APPLE_HEALTH" @@ -361,36 +423,87 @@ async function repairDay( ) : null; - let correctedDayRow: Date | null = null; + let dayRowChanged: Date | null = null; + let checked = 0; + let corrected = 0; + let resting = 0; + let absorbed = 0; await prisma.$transaction(async (tx) => { // First, before any reading is touched: see `restore-lock.ts`. await holdAccountAgainstRestore(tx, userId); + // The folds rewrite the same rows; one of the two at a time. + await holdAccountFoldLock(tx, userId); + const span = { + userId, + type: step.type, + source: step.source, + from: window.dayStart, + to: window.dayEnd, + }; + const live = await loadLiveSamples(tx, span); + const leftovers = await loadFoldLeftovers(tx, { + ...span, + foldedAfterMs: step.foldedAfterMs, + }); + if (leftovers.length === 0) return; + + // Window id → its samples. Only a window with a fold leftover is + // checked: one without has no record of what its stored mean was taken + // from beyond the rows in hand. + const byWindow = new Map< + string, + { values: number[]; liveIds: string[]; leftovers: number } + >(); + const add = (row: FoldConstituent, isLive: boolean) => { + const id = windowId(step, hk, dateKey, row, tz); + let entry = byWindow.get(id); + if (!entry) { + entry = { values: [], liveIds: [], leftovers: 0 }; + byWindow.set(id, entry); + } + entry.values.push(row.value); + if (isLive) entry.liveIds.push(row.id); + else entry.leftovers += 1; + }; + for (const row of live) add(row, true); + for (const row of leftovers) add(row, false); + const stored = await tx.measurement.findMany({ where: { userId, type: step.type, source: step.source, - externalId: { in: [...byWindow.keys()] }, + externalId: { + in: [...byWindow.entries()] + .filter(([, entry]) => entry.leftovers > 0) + .map(([id]) => id), + }, deletedAt: null, }, select: { id: true, externalId: true, value: true, measuredAt: true }, }); + const takeIds: string[] = []; for (const row of stored) { - const values = row.externalId ? byWindow.get(row.externalId) : undefined; - if (!values || values.length === 0) continue; - counts.windowsChecked += 1; - const mean = meanOf(values); + const entry = row.externalId ? byWindow.get(row.externalId) : undefined; + if (!entry) continue; + checked += 1; + takeIds.push(...entry.liveIds); + if (entry.liveIds.length > 0 && step.grain === "day") { + dayRowChanged = row.measuredAt; + } + const mean = meanOf(entry.values); if (!meanDiffers(row.value, mean)) continue; await tx.measurement.update({ where: { id: row.id }, data: { value: mean, syncVersion: { increment: 1 } }, }); - counts.windowsCorrected += 1; - if (step.grain === "day") correctedDayRow = row.measuredAt; + corrected += 1; + if (step.grain === "day") dayRowChanged = row.measuredAt; } + absorbed = await absorbIntoFold(tx, takeIds); if (restingExternalId) { - const resting = await tx.measurement.findFirst({ + const restingRow = await tx.measurement.findFirst({ where: { userId, type: "RESTING_HEART_RATE", @@ -400,32 +513,37 @@ async function repairDay( }, select: { id: true, value: true }, }); - const derived = resting + const derived = restingRow ? deriveDailyRestingFromPulse( - constituents.map((row) => ({ + [...live, ...leftovers].map((row) => ({ id: row.id, - type: row.type, + type: step.type, value: row.value, measuredAt: row.measuredAt, externalId: null, })), ) : null; - if (resting && derived !== null && resting.value !== derived) { + if (restingRow && derived !== null && restingRow.value !== derived) { await tx.measurement.update({ - where: { id: resting.id }, + where: { id: restingRow.id }, data: { value: derived, syncVersion: { increment: 1 } }, }); - counts.restingCorrected += 1; + resting += 1; } } }); + counts.windowsChecked += checked; + counts.windowsCorrected += corrected; + counts.restingCorrected += resting; + counts.samplesAbsorbed += absorbed; - // A daily mean is the only live row of its day, so the DAY rollup follows - // it as the consolidation's own recompute does. The dense tier's DAY - // bucket was taken from the raw day before the fold and is left alone. - if (correctedDayRow !== null) { - await recomputeBucketsForMeasurement(userId, step.type, correctedDayRow); + // A daily mean is the only live row of its day once its samples are taken + // in, so the DAY rollup follows it as the consolidation's own recompute + // does. The dense tier's DAY bucket was taken from the raw day before the + // fold and is left alone. + if (dayRowChanged !== null) { + await recomputeBucketsForMeasurement(userId, step.type, dayRowChanged); } } @@ -460,7 +578,14 @@ export async function handleMeasurementFoldRepair( return withBackgroundEvent("job.measurement_fold_repair", async (evt) => { const startedAt = Date.now(); const candidates: JobRunCandidate[] = []; - const totals = { checked: 0, corrected: 0, resting: 0, completed: 0 }; + const totals = { + checked: 0, + corrected: 0, + resting: 0, + skipped: 0, + absorbed: 0, + completed: 0, + }; const byType: Record = {}; const boss = getGlobalBoss(); @@ -487,17 +612,17 @@ export async function handleMeasurementFoldRepair( continue; } for (const [type, counts] of Object.entries(result.byType)) { - const sum = (byType[type] ??= { - windowsChecked: 0, - windowsCorrected: 0, - restingCorrected: 0, - }); + const sum = (byType[type] ??= emptyFoldRepairCounts()); sum.windowsChecked += counts.windowsChecked; sum.windowsCorrected += counts.windowsCorrected; sum.restingCorrected += counts.restingCorrected; + sum.windowsSkippedBeyondHorizon += counts.windowsSkippedBeyondHorizon; + sum.samplesAbsorbed += counts.samplesAbsorbed; totals.checked += counts.windowsChecked; totals.corrected += counts.windowsCorrected; totals.resting += counts.restingCorrected; + totals.skipped += counts.windowsSkippedBeyondHorizon; + totals.absorbed += counts.samplesAbsorbed; } if (result.status === "completed") { @@ -557,11 +682,15 @@ export async function handleMeasurementFoldRepair( evt.addMeta("fold_repair_means_checked", totals.checked); evt.addMeta("fold_repair_means_corrected", totals.corrected); evt.addMeta("fold_repair_resting_corrected", totals.resting); + evt.addMeta("fold_repair_skipped_beyond_horizon", totals.skipped); + evt.addMeta("fold_repair_samples_absorbed", totals.absorbed); return jobDone({ means_checked: totals.checked, means_corrected: totals.corrected, resting_corrected: totals.resting, + means_skipped_beyond_horizon: totals.skipped, + means_samples_absorbed: totals.absorbed, accounts_completed: totals.completed, duration_ms: Date.now() - startedAt, }); diff --git a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts index 283e97940..8dfb3c19c 100644 --- a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts +++ b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts @@ -22,6 +22,7 @@ import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({ ...(await importOriginal()), holdAccountAgainstRestore: vi.fn(async () => {}), + holdAccountFoldLock: vi.fn(async () => {}), })); function row( @@ -434,56 +435,166 @@ describe("consolidateDailyMean — canonical-slot collision (second unique index expect(summary.totals.daysFailed).toBe(0); }); - it("re-run path: a slot row already carrying the target externalId is updated in place, with the mean over every sample of the day", async () => { - const { mock, upsert, update, txFindFirst, txFindMany } = buildPrismaMock({ - WALKING_SPEED: [row("late", 1.2, "2026-05-01T18:00:00.000Z")], - }); - occupySlot(txFindFirst, { - id: "mean-1", - externalId: targetExternalId, - deletedAt: null, - }); - // A sample an earlier fold of the same day soft-deleted. - txFindMany.mockResolvedValue([ - { - id: "early", - value: 1.0, - measuredAt: new Date("2026-05-01T07:00:00.000Z"), - externalId: "uuid-early", - deletedAt: new Date("2026-05-03T03:00:00.000Z"), - syncVersion: 1, - }, - ]); + /** The live daily row the stored-day branch finds, with its value. */ + function storedDay( + txFindFirst: ReturnType, + stored: { id: string; value: number }, + ) { + txFindFirst.mockImplementation( + async (args: { where: { measuredAt?: Date; externalId?: string } }) => + args.where.measuredAt === undefined && + args.where.externalId === targetExternalId + ? stored + : null, + ); + } - await consolidateDailyMean(mock, { log: () => {} }); + /** Live samples and fold leftovers of the day, told apart by `deletedAt`. */ + function daySamples( + txFindMany: ReturnType, + liveRows: Array<{ id: string; value: number; iso: string }>, + leftovers: Array<{ id: string; value: number; iso: string }>, + deletedAt: Date, + ) { + txFindMany.mockImplementation( + async (args: { where: { deletedAt: null | { not: null } } }) => + args.where.deletedAt === null + ? liveRows.map((r) => ({ + id: r.id, + value: r.value, + measuredAt: new Date(r.iso), + })) + : leftovers.map((r) => ({ + id: r.id, + value: r.value, + measuredAt: new Date(r.iso), + externalId: `uuid-${r.id}`, + deletedAt, + syncVersion: 1, + })), + ); + } - // No shift, no yield — the upsert's update branch owns the row. - expect(update).not.toHaveBeenCalled(); - const upsertArg = upsert.mock.calls[0]?.[0] as { - create: { measuredAt: Date }; - update: { value: number }; - }; - expect(upsertArg.create.measuredAt.getTime()).toBe(noon.getTime()); - // Not the late sample alone: the whole day. - expect(upsertArg.update.value).toBeCloseTo(1.1, 9); - // The live-row lookup and the single slot probe — no free-instant search. - expect(txFindFirst).toHaveBeenCalledTimes(2); + it("a day with a live daily row inside the horizon gets the mean over every sample, and its live rows stay as leftovers", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-05-04T03:00:00.000Z")); + try { + const { + mock, + upsert, + update, + updateMany, + deleteMany, + txFindFirst, + txFindMany, + } = buildPrismaMock({ + WALKING_SPEED: [row("late", 1.2, "2026-05-01T18:00:00.000Z")], + }); + storedDay(txFindFirst, { id: "mean-1", value: 1.2 }); + updateMany.mockResolvedValue({ count: 1 }); + daySamples( + txFindMany, + [{ id: "late", value: 1.2, iso: "2026-05-01T18:00:00.000Z" }], + [{ id: "early", value: 1.0, iso: "2026-05-01T07:00:00.000Z" }], + new Date("2026-05-03T03:00:00.000Z"), + ); + + const summary = await consolidateDailyMean(mock, { log: () => {} }); + + // Not re-minted, not the late sample alone: the whole day. + expect(upsert).not.toHaveBeenCalled(); + expect(update).toHaveBeenCalledTimes(1); + const updArg = update.mock.calls[0]?.[0] as { + where: { id: string }; + data: { value: number }; + }; + expect(updArg.where.id).toBe("mean-1"); + expect(updArg.data.value).toBeCloseTo(1.1, 9); + // The live sample becomes a leftover, never deleted outright. + expect(deleteMany).not.toHaveBeenCalled(); + const absorbArg = updateMany.mock.calls[0]?.[0] as { + where: { id: { in: string[] } }; + data: { deletedAt: Date }; + }; + expect(absorbArg.where.id.in).toEqual(["late"]); + expect(absorbArg.data.deletedAt).toBeInstanceOf(Date); + expect(summary.totals.daysLeftAsStored).toBe(0); + } finally { + vi.useRealTimers(); + } }); - it("re-run path without fold leftovers keeps the stored mean and drops the raw rows", async () => { - const { mock, upsert, deleteMany, txFindFirst } = buildPrismaMock({ - WALKING_SPEED: [row("late", 1.2, "2026-05-01T18:00:00.000Z")], - }); - occupySlot(txFindFirst, { - id: "mean-1", - externalId: targetExternalId, - deletedAt: null, - }); + it("a day with a live daily row inside the horizon but no fold leftovers is left as stored", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-05-04T03:00:00.000Z")); + try { + const { + mock, + upsert, + update, + updateMany, + deleteMany, + txFindFirst, + txFindMany, + } = buildPrismaMock({ + WALKING_SPEED: [row("late", 1.2, "2026-05-01T18:00:00.000Z")], + }); + storedDay(txFindFirst, { id: "mean-1", value: 1.0 }); + daySamples( + txFindMany, + [{ id: "late", value: 1.2, iso: "2026-05-01T18:00:00.000Z" }], + [], + new Date("2026-05-03T03:00:00.000Z"), + ); - await consolidateDailyMean(mock, { log: () => {} }); + const summary = await consolidateDailyMean(mock, { log: () => {} }); - expect(upsert).not.toHaveBeenCalled(); - expect(deleteMany).toHaveBeenCalledTimes(1); + expect(upsert).not.toHaveBeenCalled(); + expect(update).not.toHaveBeenCalled(); + expect(updateMany).not.toHaveBeenCalled(); + expect(deleteMany).not.toHaveBeenCalled(); + expect(summary.totals.daysLeftAsStored).toBe(1); + expect(summary.totals.daysConsolidated).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("a day with a live daily row before the horizon keeps its mean and its live rows, whatever the leftovers", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + // 2026-05-01 lies about 160 days back: tombstones of it may be purged. + vi.setSystemTime(new Date("2026-10-08T03:00:00.000Z")); + try { + const { + mock, + upsert, + update, + updateMany, + deleteMany, + txFindFirst, + txFindMany, + } = buildPrismaMock({ + WALKING_SPEED: [row("late", 9, "2026-05-01T18:00:00.000Z")], + }); + storedDay(txFindFirst, { id: "mean-1", value: 1.0 }); + daySamples( + txFindMany, + [{ id: "late", value: 9, iso: "2026-05-01T18:00:00.000Z" }], + [{ id: "early", value: 9, iso: "2026-05-01T07:00:00.000Z" }], + new Date("2026-09-01T03:00:00.000Z"), + ); + + const summary = await consolidateDailyMean(mock, { log: () => {} }); + + expect(upsert).not.toHaveBeenCalled(); + expect(update).not.toHaveBeenCalled(); + expect(updateMany).not.toHaveBeenCalled(); + expect(deleteMany).not.toHaveBeenCalled(); + expect(summary.totals.daysLeftAsStored).toBe(1); + expect(recomputeBucketsForMeasurement).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } }); it("a failing day bucket does not abort the pass — later buckets still consolidate", async () => { diff --git a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts index f725067a1..191415400 100644 --- a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts @@ -41,6 +41,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock"; vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({ ...(await importOriginal()), holdAccountAgainstRestore: vi.fn(async () => {}), + holdAccountFoldLock: vi.fn(async () => {}), })); function pulseRow(id: string, value: number, iso: string): PerSampleRow { diff --git a/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts b/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts index 84a247c37..a4f99c8f0 100644 --- a/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts @@ -40,6 +40,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock"; vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({ ...(await importOriginal()), holdAccountAgainstRestore: vi.fn(async () => {}), + holdAccountFoldLock: vi.fn(async () => {}), })); /** A day well outside the retention window, so the fold definitely runs. */ diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts index fb1586acd..2d76f13ae 100644 --- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts @@ -34,6 +34,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock"; vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({ ...(await importOriginal()), holdAccountAgainstRestore: vi.fn(async () => {}), + holdAccountFoldLock: vi.fn(async () => {}), })); function row( diff --git a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts index d0cddd2f6..e5cabd28b 100644 --- a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts @@ -37,6 +37,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock"; vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({ ...(await importOriginal()), holdAccountAgainstRestore: vi.fn(async () => {}), + holdAccountFoldLock: vi.fn(async () => {}), })); function spo2Row(id: string, value: number, iso: string): PerSampleRow { diff --git a/src/lib/measurements/__tests__/hourly-mint-mock.ts b/src/lib/measurements/__tests__/hourly-mint-mock.ts index a5131cee9..2798052a5 100644 --- a/src/lib/measurements/__tests__/hourly-mint-mock.ts +++ b/src/lib/measurements/__tests__/hourly-mint-mock.ts @@ -29,14 +29,15 @@ export function createManyVia(create: CreateSpy) { /** * The day's candidate lookup. With `existingId`, one row already occupies the * first slot's anchor instant (the adopt-in-place case); otherwise nothing is - * in the way. + * in the way. The fold's own lookup of live hourly rows (no `OR`) finds none: + * the row in the way is not one of them. */ export function candidateLookup(existingId: string | null = null) { return vi.fn( async (args: { - where: { OR: [unknown, { measuredAt: { in: Date[] } }] }; + where: { OR?: [unknown, { measuredAt: { in: Date[] } }] }; }) => - existingId === null + existingId === null || args.where.OR === undefined ? [] : [ { diff --git a/src/lib/measurements/consolidate-daily-mean.ts b/src/lib/measurements/consolidate-daily-mean.ts index c7416aedc..e7febac45 100644 --- a/src/lib/measurements/consolidate-daily-mean.ts +++ b/src/lib/measurements/consolidate-daily-mean.ts @@ -30,8 +30,10 @@ * 4. UPSERT the canonical daily row keyed by * `stats::` at local-noon, `value = mean`. A day * that already has a live daily row gets the mean over every sample of - * the day, its fold leftovers included (`fold-constituents.ts`), or - * keeps its stored mean when none are left. + * the day, its fold leftovers included (`fold-constituents.ts`), but + * only inside the tombstone horizon, where those are provably all of + * them; any other such day keeps its stored mean and its live rows + * (`refoldStoredDay`). * Because mean types are NOT in `CUMULATIVE_HK_TYPES`, the daily * read path averages over the single stats row (count = 1) and * returns it unchanged — no reader change needed. @@ -49,7 +51,10 @@ * production standalone image strips `tsx`. Modelled on the * `step-consolidation` boot-time converging-backfill pattern. */ -import { holdAccountAgainstRestore } from "@/lib/export/restore-lock"; +import { + holdAccountAgainstRestore, + holdAccountFoldLock, +} from "@/lib/export/restore-lock"; import type { MeasurementType, Prisma, @@ -70,7 +75,14 @@ import { import { type PerSampleRow } from "./drain-per-sample-cumulative"; import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; import { localDayWindow } from "@/lib/tz/local-day"; -import { loadFoldLeftovers, meanOf } from "./fold-constituents"; +import { + absorbIntoFold, + foldConstituentHorizon, + loadFoldLeftovers, + loadLiveSamples, + meanDiffers, + meanOf, +} from "./fold-constituents"; /** * Grace window — rows newer than `now() - cutoffHours` stay raw so @@ -112,6 +124,12 @@ export interface MeanConsolidationSummary { daysConsolidated: number; perSampleRowsSoftDeleted: number; dailyRowsUpserted: number; + /** + * Days that already had a live daily row whose samples could not be shown + * to be complete (`fold-constituents.ts`): the stored mean and the day's + * live rows were left as they are. + */ + daysLeftAsStored: number; /** * Day buckets whose write failed and was stepped over. A failed day * keeps its per-sample rows live, so the next nightly run retries it; @@ -208,6 +226,73 @@ async function nextFreeInstant( ); } +/** + * The fold of a day that already has a live daily row, and live samples the + * scan found for it. Before v1.42 the fold cut at `now - 36 h` and folded a + * day in two runs; since then a day is folded whole, and the `folded_window` + * ingest guard refuses a sample of a folded day, so this is a day an older + * release folded in part, or one whose samples reached the table before the + * guard existed. + * + * The stored mean is replaced only by the mean over every sample of the day, + * and only when the samples present are provably all of them: the day starts + * at or after the horizon of `fold-constituents.ts` (inside it no fold + * tombstone has been purged) and the fold left tombstones for it. The live + * samples it takes are then soft-deleted as fold leftovers, so the one-time + * repair and any later pass compute the same mean from the same samples. + * + * Otherwise the stored mean was taken from samples that are no longer all in + * the table, and the live rows may be a re-upload of some of them or a + * fragment of the day. Up to v1.41 the fold overwrote the stored mean with + * the mean of these rows alone; the v1.42 first cut replaced it with the mean + * of these rows plus whatever tombstones were left. Both compute a mean from + * part of a day and both can move a correct mean far off. So the day is left + * exactly as it is: the stored mean stays, and the live rows stay as they + * were synced, nothing lost. They are few (the guard refuses new ones) and + * the nightly pass passes over them again without writing. + */ +async function refoldStoredDay( + tx: Prisma.TransactionClient, + input: { + userId: string; + type: MeasurementType; + stored: { id: string; value: number }; + tz: string; + dateKey: string; + }, +): Promise { + const { userId, type, stored, tz, dateKey } = input; + const foldedAfterMs = MEAN_CONSOLIDATION_CUTOFF_HOURS * 3_600_000; + const { dayStart, dayEnd } = localDayWindow(dateKey, tz); + if (dayStart < foldConstituentHorizon(new Date(), foldedAfterMs)) { + return { kind: "left-as-stored" }; + } + const window = { + userId, + type, + source: "APPLE_HEALTH" as const, + from: dayStart, + to: dayEnd, + }; + // Read under the fold lock: the repair may have taken them meanwhile. + const live = await loadLiveSamples(tx, window); + if (live.length === 0) return { kind: "written", sourceRowsRemoved: 0 }; + const leftovers = await loadFoldLeftovers(tx, { ...window, foldedAfterMs }); + if (leftovers.length === 0) return { kind: "left-as-stored" }; + const mean = meanOf([...live, ...leftovers].map((row) => row.value)); + if (meanDiffers(stored.value, mean)) { + await tx.measurement.update({ + where: { id: stored.id }, + data: { value: mean, syncVersion: { increment: 1 } }, + }); + } + const absorbed = await absorbIntoFold( + tx, + live.map((row) => row.id), + ); + return { kind: "written", sourceRowsRemoved: absorbed }; +} + /** * Run the daily-mean consolidation. Idempotent — re-invocation after a * successful pass reports zero days because the per-sample rows are @@ -230,6 +315,7 @@ export async function consolidateDailyMean( daysConsolidated: 0, perSampleRowsSoftDeleted: 0, dailyRowsUpserted: 0, + daysLeftAsStored: 0, daysFailed: 0, }, }; @@ -280,10 +366,42 @@ export async function consolidateDailyMean( // and never reads a soft-deleted row's unit — the scan already // filters `deletedAt: null`. const unit = dayRows[0]?.unit ?? "unknown"; - let removed = 0; + // Assigned inside the transaction callback, so not narrowed here. + let outcome = { + kind: "written", + sourceRowsRemoved: 0, + } as DayWriteOutcome; await pc.$transaction(async (tx) => { // First, before any reading is touched: see `restore-lock.ts`. await holdAccountAgainstRestore(tx, userId); + // The fold repair rewrites the same rows; one of the two at a time. + await holdAccountFoldLock(tx, userId); + + // A live daily row already holds this day: a day an earlier run + // folded (in part, before the boundary was aligned to the local + // day) that these rows reached afterwards. It is never re-minted + // from the rows in hand; see `refoldStoredDay`. + const existing = await tx.measurement.findFirst({ + where: { + userId, + type, + source: "APPLE_HEALTH", + externalId, + deletedAt: null, + }, + select: { id: true, value: true }, + }); + if (existing) { + outcome = await refoldStoredDay(tx, { + userId, + type, + stored: existing, + tz, + dateKey, + }); + return; + } + // The upsert below arbiters on (userId, type, source, externalId), // but the Measurement model carries a SECOND full unique — // (userId, type, measuredAt, source, sleepStage), NULLS NOT @@ -348,77 +466,34 @@ export async function consolidateDailyMean( } } - // A live daily row already holds this day. The pass folds only - // complete local days now, so that is a day an earlier run folded - // (in part, before the boundary was aligned) and these rows reached - // it afterwards. Its stored mean is replaced only by the mean over - // EVERY sample of the day: the live rows in hand plus the ones an - // earlier fold soft-deleted (`fold-constituents.ts`). When none of - // those are left the stored mean is the only account of the earlier - // samples, so it stays as it is, and the live rows go the way of - // any folded sample. - const existing = await tx.measurement.findFirst({ - where: { - userId, - type, - source: "APPLE_HEALTH", - externalId, - deletedAt: null, - }, - select: { id: true }, - }); - let value = reducedValue; - let keepStored = false; - if (existing) { - const { dayStart, dayEnd } = localDayWindow(dateKey, tz); - const leftovers = await loadFoldLeftovers(tx, { - userId, - type, - source: "APPLE_HEALTH", - from: dayStart, - to: dayEnd, - foldedAfterMs: MEAN_CONSOLIDATION_CUTOFF_HOURS * 3_600_000, - }); - if (leftovers.length > 0) { - value = meanOf([ - ...dayRows.map((row) => row.value), - ...leftovers.map((row) => row.value), - ]); - } else { - keepStored = true; - } - } - // Mint / refresh the canonical daily-mean row. The unique // index (userId, type, source, externalId) makes the upsert // idempotent across re-runs. Built field-by-field (no spread) // per the no-mass-assignment convention. - if (!keepStored) { - await tx.measurement.upsert({ - where: { - userId_type_source_externalId: { - userId, - type, - source: "APPLE_HEALTH", - externalId, - }, - }, - create: { + await tx.measurement.upsert({ + where: { + userId_type_source_externalId: { userId, type, - value, - unit, source: "APPLE_HEALTH", - measuredAt: mintAt, externalId, }, - update: { - value, - measuredAt: mintAt, - deletedAt: null, - }, - }); - } + }, + create: { + userId, + type, + value: reducedValue, + unit, + source: "APPLE_HEALTH", + measuredAt: mintAt, + externalId, + }, + update: { + value: reducedValue, + measuredAt: mintAt, + deletedAt: null, + }, + }); // Delete the per-sample rows in the same transaction. // @@ -429,8 +504,9 @@ export async function consolidateDailyMean( const del = await tx.measurement.deleteMany({ where: { id: { in: sourceRowIds }, deletedAt: null }, }); - removed = del.count; + outcome = { kind: "written", sourceRowsRemoved: del.count }; }); + if (outcome.kind === "left-as-stored") return outcome; // T3 — recompute the affected (user, type, day) rollup buckets after // the consolidation commits, the same way the batch route does on @@ -442,7 +518,7 @@ export async function consolidateDailyMean( // lands the recompute on the correct UTC day bucket. await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); - return { kind: "written", sourceRowsRemoved: removed }; + return outcome; }, recordBucket: ({ userId, @@ -463,6 +539,10 @@ export async function consolidateDailyMean( canonicalTimestamp: canonicalTimestamp.toISOString(), externalId, }); + if (outcome?.kind === "left-as-stored") { + summary.totals.daysLeftAsStored += 1; + return; + } summary.totals.daysConsolidated += 1; summary.totals.dailyRowsUpserted += 1; summary.totals.perSampleRowsSoftDeleted += @@ -493,7 +573,7 @@ export async function consolidateDailyMean( summary.totals.daysFailed = walk.daysFailed; log( - `[mean-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFailed=${summary.totals.daysFailed}${options.dryRun ? " (dry-run)" : ""}`, + `[mean-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} perSampleRowsSoftDeleted=${summary.totals.perSampleRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysLeftAsStored=${summary.totals.daysLeftAsStored} daysFailed=${summary.totals.daysFailed}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; diff --git a/src/lib/measurements/consolidation-base.ts b/src/lib/measurements/consolidation-base.ts index ad6b0b121..0f1b21a73 100644 --- a/src/lib/measurements/consolidation-base.ts +++ b/src/lib/measurements/consolidation-base.ts @@ -145,9 +145,16 @@ export function bucketRowsByDay( return byDay; } -/** Outcome of writing a single per-day bucket. */ +/** + * Outcome of writing a single per-day bucket. `left-as-stored`: the day + * already has its `stats:` mean and the samples in hand cannot be shown to be + * all of the day (`fold-constituents.ts`), so the stored mean and the rows + * stay exactly as they are. + */ export type DayWriteOutcome = - { kind: "written"; sourceRowsRemoved: number } | { kind: "skipped-conflict" }; + | { kind: "written"; sourceRowsRemoved: number } + | { kind: "skipped-conflict" } + | { kind: "left-as-stored" }; /** * Context a per-day write strategy receives. The strategy owns its own diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index 0097b4097..e23fb9433 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -50,7 +50,10 @@ * The one-shot history rebuild that converts ALREADY-folded daily rows to * the hourly grain lives in `dense-intraday-hourly-rebuild.ts`. */ -import { holdAccountAgainstRestore } from "@/lib/export/restore-lock"; +import { + holdAccountAgainstRestore, + holdAccountFoldLock, +} from "@/lib/export/restore-lock"; import type { MeasurementSource, MeasurementType, @@ -75,7 +78,10 @@ import type { PerSampleRow } from "./drain-per-sample-cumulative"; import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; import { localDayWindow } from "@/lib/tz/local-day"; import { + absorbIntoFold, + foldConstituentHorizon, loadFoldLeftovers, + loadLiveSamples, meanOf, type FoldConstituent, } from "./fold-constituents"; @@ -466,6 +472,12 @@ export interface DenseIntradayRetentionSummary { * resting signal the raw-PULSE proxy can no longer compute after fold. */ derivedRestingRowsUpserted: number; + /** + * Hours that already had a live hourly row whose samples could not be + * shown to be complete (`fold-constituents.ts`): the stored mean and the + * hour's live rows were left as they are. + */ + hoursLeftAsStored: number; }; } @@ -524,6 +536,7 @@ export async function runDenseIntradayRetention( hourlyRowsUpserted: 0, dailyRowsRetired: 0, derivedRestingRowsUpserted: 0, + hoursLeftAsStored: 0, }, }; @@ -655,35 +668,20 @@ export async function runDenseIntradayRetention( deletedAt: null, }, })) > 0; - // The samples an earlier fold soft-deleted from this day: together - // with the live rows in hand they are the whole day again - // (`fold-constituents.ts`). Read once; no fold writes tombstones any - // more, so nothing adds to them while the day is folded. - let leftovers: FoldConstituent[] = []; - if (dayAlreadyFolded) { - const { dayStart, dayEnd } = localDayWindow(dateKey, tz); - leftovers = await loadFoldLeftovers(pc, { - userId, - type, - source, - from: dayStart, - to: dayEnd, - foldedAfterMs: retentionDays * 24 * 3_600_000, - }); - } else { + if (!dayAlreadyFolded) { await recomputeBucketsForMeasurement( userId, type, canonicalTimestamp, ); } - const leftoversByHour = new Map(); - for (const row of leftovers) { - const hour = hourOfDayForUserTz(row.measuredAt, tz); - const slot = leftoversByHour.get(hour) ?? []; - slot.push(row); - leftoversByHour.set(hour, slot); - } + // Whether the fold leftovers of this day are provably all still in + // the table (`foldConstituentHorizon`). Only then is a stored hourly + // mean, or the derived resting figure, recomputed. + const { dayStart, dayEnd } = localDayWindow(dateKey, tz); + const foldedAfterMs = retentionDays * 24 * 3_600_000; + const withinHorizon = + dayStart >= foldConstituentHorizon(new Date(), foldedAfterMs); // Fold the day into per-LOCAL-HOUR mean rows, anchored at local // HH:30. Sorted for a deterministic write order across runs. @@ -695,47 +693,97 @@ export async function runDenseIntradayRetention( removed: number; retiredDaily: boolean; slotsWritten: number; + hoursLeftAsStored: number; + /** Every raw sample of the day, when that is provably all of it. */ + wholeDay: FoldConstituent[] | null; }> => { let removed = 0; let retiredDaily = false; let slotsWritten = 0; + let hoursLeftAsStored = 0; + let wholeDay: FoldConstituent[] | null = null; await pc.$transaction(async (tx) => { // First, before any reading is touched: see `restore-lock.ts`. await holdAccountAgainstRestore(tx, userId); - // An hour that already has a live hourly row gets the mean over - // every sample of the hour (live + fold leftovers), never the - // mean of this batch alone. With no leftovers the stored mean is - // the only account of the hour's earlier samples and stays. + // The fold repair rewrites the same rows; one of the two at a + // time. + await holdAccountFoldLock(tx, userId); + // An hour that already has a live hourly row is never re-minted + // from the samples in hand. Inside the horizon, with fold + // leftovers for the hour, it gets the mean over every sample of + // the hour (live + leftovers) and the live samples are kept as + // leftovers (`absorbIntoFold`), so the repair and any later pass + // compute the same mean. Otherwise the stored mean was taken from + // samples that are no longer all in the table, and the hour is + // left exactly as it is, its live samples included: a mean of + // part of the hour must not replace it (see `refoldStoredDay` in + // `consolidate-daily-mean.ts`). const hourIds = byHour.map(([hour]) => hourlyStatsExternalId(hkIdentifier, dateKey, hour), ); - const liveHourly = dayAlreadyFolded - ? new Set( - ( - await tx.measurement.findMany({ - where: { - userId, - type, - source, - externalId: { in: hourIds }, - deletedAt: null, - }, - select: { externalId: true }, - }) - ).map((row) => row.externalId), - ) - : new Set(); + const liveHourly = new Set( + ( + await tx.measurement.findMany({ + where: { + userId, + type, + source, + externalId: { in: hourIds }, + deletedAt: null, + }, + select: { externalId: true }, + }) + ).map((row) => row.externalId), + ); + // Read under the fold lock: the repair may have taken some of the + // samples meanwhile. + let leftovers: FoldConstituent[] = []; + let liveIds = new Set(sourceRowIds); + if (liveHourly.size > 0 && withinHorizon) { + const window = { + userId, + type, + source, + from: dayStart, + to: dayEnd, + }; + liveIds = new Set( + (await loadLiveSamples(tx, window)).map((row) => row.id), + ); + leftovers = await loadFoldLeftovers(tx, { + ...window, + foldedAfterMs, + }); + } + const leftoversByHour = new Map(); + for (const row of leftovers) { + const hour = hourOfDayForUserTz(row.measuredAt, tz); + const slot = leftoversByHour.get(hour) ?? []; + slot.push(row); + leftoversByHour.set(hour, slot); + } + const slots: HourlySlot[] = []; + const deleteIds: string[] = []; + const absorbIds: string[] = []; for (const [hour, hourRows] of byHour) { const slotId = hourlyStatsExternalId(hkIdentifier, dateKey, hour); let value = meanBucketValue(hourRows); if (liveHourly.has(slotId)) { const earlier = leftoversByHour.get(hour) ?? []; - if (earlier.length === 0) continue; + if (!withinHorizon || earlier.length === 0) { + hoursLeftAsStored += 1; + continue; + } + const live = hourRows.filter((row) => liveIds.has(row.id)); + if (live.length === 0) continue; value = meanOf([ - ...hourRows.map((row) => row.value), + ...live.map((row) => row.value), ...earlier.map((row) => row.value), ]); + absorbIds.push(...live.map((row) => row.id)); + } else { + deleteIds.push(...hourRows.map((row) => row.id)); } slots.push({ externalId: slotId, @@ -776,10 +824,11 @@ export async function runDenseIntradayRetention( retiredDaily = true; } - // Delete the out-of-window per-sample rows in the same - // transaction. 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 removing it would erase the fold. + // Delete the out-of-window per-sample rows of the hours folded + // here in the same transaction. 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 removing + // it would erase the fold. // // v1.42 — deleted outright, like the cumulative drain's rows, // where they used to be tombstoned for the 75-day retention. No @@ -790,20 +839,43 @@ export async function runDenseIntradayRetention( // of the table and its indexes. const del = await tx.measurement.deleteMany({ where: { - id: { in: sourceRowIds, notIn: canonicalRowIds }, + id: { in: deleteIds, notIn: canonicalRowIds }, deletedAt: null, }, }); - removed = del.count; + const absorbed = await absorbIntoFold( + tx, + absorbIds.filter((id) => !canonicalRowIds.includes(id)), + ); + removed = del.count + absorbed; + + // The day's raw samples, for the derived resting figure: the rows + // in hand for a day folded here for the first time; for a day + // folded before, the live rows plus the leftovers, but only when + // every hour of it could be recomputed. + if (!dayAlreadyFolded && liveHourly.size === 0) { + wholeDay = dayRows; + } else if ( + withinHorizon && + leftovers.length > 0 && + hoursLeftAsStored === 0 + ) { + wholeDay = [ + ...dayRows.filter((row) => liveIds.has(row.id)), + ...leftovers, + ]; + } }); - return { removed, retiredDaily, slotsWritten }; + return { + removed, + retiredDaily, + slotsWritten, + hoursLeftAsStored, + wholeDay, + }; }; - let foldResult: { - removed: number; - retiredDaily: boolean; - slotsWritten: number; - }; + let foldResult: Awaited>; try { foldResult = await foldOnce(); } catch (err) { @@ -816,6 +888,7 @@ export async function runDenseIntradayRetention( // Counters mutate OUTSIDE the retryable transaction so a P2002 retry // cannot double-count. summary.totals.hourlyRowsUpserted += foldResult.slotsWritten; + summary.totals.hoursLeftAsStored += foldResult.hoursLeftAsStored; if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1; // iOS#34 — preserve the resting signal for PROXY users. For a user @@ -831,22 +904,18 @@ export async function runDenseIntradayRetention( // would only double-count). // // On a day an earlier run already folded, the figure comes from the - // whole day (live rows + fold leftovers); with no leftovers the rows - // in hand are only part of the day and the stored figure stays. - if (isPulse && (!dayAlreadyFolded || leftovers.length > 0)) { + // whole day (live rows + fold leftovers) and only when that is + // provably the whole day; otherwise the stored figure stays. + const wholeDay = foldResult.wholeDay; + if (isPulse && wholeDay !== null) { const restingValue = deriveDailyRestingFromPulse( - dayAlreadyFolded - ? [ - ...dayRows, - ...leftovers.map((row) => ({ - id: row.id, - type, - value: row.value, - measuredAt: row.measuredAt, - externalId: null, - })), - ] - : dayRows, + wholeDay.map((row) => ({ + id: row.id, + type, + value: row.value, + measuredAt: row.measuredAt, + externalId: null, + })), ); if (restingValue !== null && !(await userHasNativeResting(userId))) { const restingExternalId = dailyStatsExternalId( @@ -950,7 +1019,7 @@ export async function runDenseIntradayRetention( } log( - `[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)" : ""}`, + `[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} hoursLeftAsStored=${summary.totals.hoursLeftAsStored}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; diff --git a/src/lib/measurements/fold-constituents.ts b/src/lib/measurements/fold-constituents.ts index 97493b3f9..3af332551 100644 --- a/src/lib/measurements/fold-constituents.ts +++ b/src/lib/measurements/fold-constituents.ts @@ -23,12 +23,54 @@ * day folded in two runs left the tombstones of its first run in a window * that was not yet complete, so they are class B (kept for the 75-day * retention) and still constituents of the day's mean. + * + * The horizon. Tombstones are purged 75 days after they were written, so for + * an older window the samples present may be a fragment: what a re-upload + * after an anchor reset put back, the rows of one later run, nothing of the + * samples the stored mean was taken from. A mean recomputed from those can be + * far off a stored mean that was computed from the whole window. Every fold + * tombstone of a day was written after `dayStart + foldedAfter` (the fold + * only took samples older than its threshold), so when that instant is still + * inside the retention, none of them has been purged yet, and the live rows + * plus the fold leftovers are the whole window. {@link foldConstituentHorizon} + * is the earliest day start for which that holds; a pass recomputes a stored + * mean only for a day starting at or after it, and leaves an older one as it + * is. For the daily means that is the last 75 days; for the dense hourly + * means, which are folded 90 days late, the 75 days before the raw window. + * + * A pass that takes a window's live samples into a stored mean it recomputed + * soft-deletes them ({@link absorbIntoFold}) rather than deleting them, so they + * stay fold leftovers: a second pass over the same window then computes the + * same mean from the same samples instead of from the leftovers alone. */ import type { MeasurementSource, MeasurementType, Prisma, } from "@/generated/prisma/client"; +import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; + +const DAY_MS = 86_400_000; + +/** + * Slack on the horizon: a pass runs for minutes, and a tombstone right at the + * edge of the retention must not be purged by the cleanup while it does. + */ +export const FOLD_HORIZON_MARGIN_MS = DAY_MS; + +/** + * The earliest day start whose fold leftovers are certainly all still in the + * table, for a fold that takes samples `foldedAfterMs` after they were + * measured. See the header. Pure. + */ +export function foldConstituentHorizon(now: Date, foldedAfterMs: number): Date { + return new Date( + now.getTime() - + TOMBSTONE_RETENTION_DAYS * DAY_MS - + foldedAfterMs + + FOLD_HORIZON_MARGIN_MS, + ); +} export interface FoldConstituent { id: string; @@ -115,6 +157,58 @@ export async function loadFoldLeftovers( })); } +/** + * The live raw samples of one `[from, to)` span of one account, type and + * source: the rows a fold of that span would take. A `stats:` row and a + * collision retirement are never samples. + */ +export async function loadLiveSamples( + client: Pick, + input: { + userId: string; + type: MeasurementType; + source: MeasurementSource; + from: Date; + to: Date; + }, +): Promise { + return client.measurement.findMany({ + where: { + userId: input.userId, + type: input.type, + source: input.source, + deletedAt: null, + externalId: { not: null }, + NOT: [ + { externalId: { startsWith: "stats:" } }, + { externalId: { startsWith: "retired:" } }, + ], + measuredAt: { gte: input.from, lt: input.to }, + }, + select: { id: true, value: true, measuredAt: true }, + }); +} + +/** + * Soft-delete live samples a pass has just taken into a stored mean it + * recomputed, leaving them as fold leftovers (see the header). Raw HealthKit + * and Health Connect samples are never updated in place, so they keep + * `syncVersion = 1`, which {@link isFoldLeftover} reads. Returns how many rows + * it took. + */ +export async function absorbIntoFold( + tx: Pick, + ids: readonly string[], + at: Date = new Date(), +): Promise { + if (ids.length === 0) return 0; + const res = await tx.measurement.updateMany({ + where: { id: { in: [...ids] }, deletedAt: null }, + data: { deletedAt: at }, + }); + return res.count; +} + /** Arithmetic mean of a non-empty list of values. */ export function meanOf(values: readonly number[]): number { let acc = 0; diff --git a/tests/integration/dense-intraday-retention.test.ts b/tests/integration/dense-intraday-retention.test.ts index c1959e9f6..6b5422282 100644 --- a/tests/integration/dense-intraday-retention.test.ts +++ b/tests/integration/dense-intraday-retention.test.ts @@ -12,7 +12,7 @@ * adopts, retires a pre-hourly daily row atomically, stays idempotent, * and preserves the dense-tier scope. */ -import { beforeEach, describe, expect, it } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import { getPrismaClient, truncateAllTables } from "./setup"; import { runDenseIntradayRetention } from "@/lib/measurements/dense-intraday-retention"; @@ -223,12 +223,21 @@ describe("runDenseIntradayRetention (real Postgres)", () => { }, }); - // Must not throw P2002 — the fold adopts the existing hourly row. - const summary = await runDenseIntradayRetention(prisma, { - userId: TEST_USER_ID, - retentionDays: 0, - log: () => {}, - }); + // Must not throw P2002 — the fold adopts the existing hourly row. Run + // a few days after the fold, inside the tombstone horizon, where the + // stored hour may be recomputed from all of its samples. + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date("2026-05-05T00:00:00.000Z")); + let summary: Awaited>; + try { + summary = await runDenseIntradayRetention(prisma, { + userId: TEST_USER_ID, + retentionDays: 0, + log: () => {}, + }); + } finally { + vi.useRealTimers(); + } expect(summary.totals.daysConsolidated).toBe(1); const live = await prisma.measurement.findMany({ diff --git a/tests/integration/fold-complete-windows.test.ts b/tests/integration/fold-complete-windows.test.ts index d448e1556..f092ddf39 100644 --- a/tests/integration/fold-complete-windows.test.ts +++ b/tests/integration/fold-complete-windows.test.ts @@ -406,6 +406,9 @@ describe("folded_window", () => { }); }); +/** The instant the repair tests read the horizon against. */ +const REPAIR_NOW = "2026-10-08T12:00:00.000Z"; + /** The prod shape: a mean an old two-run fold left, and its tombstones. */ async function seedPartialFolds(userId: string) { const prisma = getPrismaClient(); @@ -511,6 +514,8 @@ async function seedPartialFolds(userId: string) { describe("fold repair", () => { it("corrects a partial mean from the tombstones, leaves a correct one alone, and is idempotent", async () => { + // Inside the horizon of every seeded window. + at(REPAIR_NOW); const prisma = getPrismaClient(); await account("repair"); await seedPartialFolds("repair"); @@ -528,11 +533,15 @@ describe("fold repair", () => { windowsChecked: 2, windowsCorrected: 1, restingCorrected: 0, + windowsSkippedBeyondHorizon: 0, + samplesAbsorbed: 0, }); expect(first.byType.HEART_RATE_VARIABILITY).toEqual({ windowsChecked: 1, windowsCorrected: 1, restingCorrected: 0, + windowsSkippedBeyondHorizon: 0, + samplesAbsorbed: 0, }); expect( (await live("repair", "WALKING_SPEED", `stats:${WS_HK}:2026-09-18`)) @@ -564,6 +573,8 @@ describe("fold repair", () => { }); it("stops between days and resumes where it stopped", async () => { + // Inside the horizon of every seeded window. + at(REPAIR_NOW); const prisma = getPrismaClient(); await account("repair-resume"); await seedPartialFolds("repair-resume"); @@ -600,6 +611,8 @@ describe("fold repair", () => { }); it("the handler marks the account and the purge refuses before it", async () => { + // Inside the horizon of every seeded window. + at(REPAIR_NOW); const prisma = getPrismaClient(); await account("gate"); await seedPartialFolds("gate"); diff --git a/tests/integration/fold-horizon.test.ts b/tests/integration/fold-horizon.test.ts new file mode 100644 index 000000000..32eb92a34 --- /dev/null +++ b/tests/integration/fold-horizon.test.ts @@ -0,0 +1,533 @@ +/** + * v1.42 — a stored `stats:` mean is recomputed from its samples only where + * those are provably all still in the table, and never from a fragment. + * Against Postgres. + * + * - The fold repair corrects a partial mean inside the tombstone horizon, + * takes the window's live samples into it, and leaves a day before the + * horizon exactly as it is, counting it as skipped. A second run corrects + * nothing. + * - The daily-mean consolidation and the dense hourly fold leave a stored + * day or hour before the horizon as it is, its live samples included, and + * still fold a day before the horizon that has no stored mean yet. + * - The repair and the consolidation over the same account, at the same + * time, end on the mean over every sample, and a repair after them + * corrects nothing. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const WS_HK = "HKQuantityTypeIdentifierWalkingSpeed"; +const HRV_HK = "HKQuantityTypeIdentifierHeartRateVariabilitySDNN"; + +/** The instant every test reads the horizon against. */ +const NOW = "2026-10-08T12:00:00.000Z"; +/** About 210 days back: before the daily horizon and the hourly one. */ +const OLD_DAY = "2026-03-10"; +/** Inside the daily horizon. */ +const RECENT_DAY = "2026-09-18"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date(NOW)); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +async function account(id: string) { + await getPrismaClient().user.create({ + data: { + id, + username: id, + email: `${id}@example.test`, + timezone: "UTC", + }, + }); +} + +async function liveValue(userId: string, type: string, externalId: string) { + const row = await getPrismaClient().measurement.findFirst({ + where: { userId, type: type as never, externalId, deletedAt: null }, + select: { value: true }, + }); + return row?.value; +} + +async function liveSampleIds(userId: string, type: string) { + const rows = await getPrismaClient().measurement.findMany({ + where: { + userId, + type: type as never, + deletedAt: null, + NOT: { externalId: { startsWith: "stats:" } }, + }, + select: { id: true }, + orderBy: { id: "asc" }, + }); + return rows.map((row) => row.id); +} + +type Sample = { + id: string; + type: "WALKING_SPEED" | "HEART_RATE_VARIABILITY"; + value: number; + at: string; + deletedAt?: string; +}; + +async function seed(userId: string, stats: Sample[], samples: Sample[]) { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + ...stats.map((row) => ({ + id: `${userId}-${row.id}`, + userId, + type: row.type, + value: row.value, + unit: row.type === "WALKING_SPEED" ? "m/s" : "ms", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(row.at), + externalId: + row.type === "WALKING_SPEED" + ? `stats:${WS_HK}:${row.at.slice(0, 10)}` + : `stats:${HRV_HK}:${row.at.slice(0, 13)}`, + })), + ...samples.map((row) => ({ + id: `${userId}-${row.id}`, + userId, + type: row.type, + value: row.value, + unit: row.type === "WALKING_SPEED" ? "m/s" : "ms", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(row.at), + externalId: `uuid-${userId}-${row.id}`, + deletedAt: row.deletedAt ? new Date(row.deletedAt) : null, + syncVersion: 1, + })), + ], + }); +} + +/** + * An old day whose mean was taken from its full day long ago (the samples' + * tombstones are purged), and a fragment of it present now: two samples a + * later re-upload put back and a fold took again in September, and one more + * re-uploaded today. A mean over the fragment is 1; the stored mean is 3. + */ +async function seedOldDay(userId: string) { + await seed( + userId, + [ + { + id: "old-mean", + type: "WALKING_SPEED", + value: 3, + at: `${OLD_DAY}T12:00:00.000Z`, + }, + ], + [ + { + id: "old-t1", + type: "WALKING_SPEED", + value: 1, + at: `${OLD_DAY}T08:00:00.000Z`, + deletedAt: "2026-09-01T03:00:00.000Z", + }, + { + id: "old-t2", + type: "WALKING_SPEED", + value: 1, + at: `${OLD_DAY}T09:00:00.000Z`, + deletedAt: "2026-09-01T03:00:00.000Z", + }, + { + id: "old-live", + type: "WALKING_SPEED", + value: 1, + at: `${OLD_DAY}T10:00:00.000Z`, + }, + ], + ); +} + +/** + * A recent day an old release folded in two runs (stored: the evening only, + * 2), plus one sample synced after the second run. Every sample is here: the + * mean over all of them is (1 + 1 + 2 + 2 + 2) / 5 = 1.6. + */ +async function seedRecentDay(userId: string) { + await seed( + userId, + [ + { + id: "recent-mean", + type: "WALKING_SPEED", + value: 2, + at: `${RECENT_DAY}T12:00:00.000Z`, + }, + ], + [ + { + id: "recent-t1", + type: "WALKING_SPEED", + value: 1, + at: `${RECENT_DAY}T08:00:00.000Z`, + deletedAt: "2026-09-20T03:00:00.000Z", + }, + { + id: "recent-t2", + type: "WALKING_SPEED", + value: 1, + at: `${RECENT_DAY}T10:00:00.000Z`, + deletedAt: "2026-09-20T03:00:00.000Z", + }, + { + id: "recent-t3", + type: "WALKING_SPEED", + value: 2, + at: `${RECENT_DAY}T20:00:00.000Z`, + deletedAt: "2026-09-21T03:00:00.000Z", + }, + { + id: "recent-t4", + type: "WALKING_SPEED", + value: 2, + at: `${RECENT_DAY}T22:00:00.000Z`, + deletedAt: "2026-09-21T03:00:00.000Z", + }, + { + id: "recent-live", + type: "WALKING_SPEED", + value: 2, + at: `${RECENT_DAY}T23:30:00.000Z`, + }, + ], + ); +} + +describe("fold repair horizon", () => { + it("repairs inside the horizon, leaves an older day as it is, and a second run corrects nothing", async () => { + const prisma = getPrismaClient(); + await account("horizon-repair"); + await seedOldDay("horizon-repair"); + await seedRecentDay("horizon-repair"); + const oldTombstonesBefore = await prisma.measurement.findMany({ + where: { userId: "horizon-repair", deletedAt: { not: null } }, + select: { id: true, deletedAt: true }, + orderBy: { id: "asc" }, + }); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + + const first = await repairFoldedMeans(prisma, "horizon-repair"); + expect(first.status).toBe("completed"); + expect(first.byType.WALKING_SPEED).toEqual({ + windowsChecked: 1, + windowsCorrected: 1, + restingCorrected: 0, + windowsSkippedBeyondHorizon: 1, + samplesAbsorbed: 1, + }); + expect( + await liveValue( + "horizon-repair", + "WALKING_SPEED", + `stats:${WS_HK}:${OLD_DAY}`, + ), + ).toBe(3); + expect( + await liveValue( + "horizon-repair", + "WALKING_SPEED", + `stats:${WS_HK}:${RECENT_DAY}`, + ), + ).toBeCloseTo(1.6, 9); + // The old day's live sample is still live; the recent one is a leftover. + expect(await liveSampleIds("horizon-repair", "WALKING_SPEED")).toEqual([ + "horizon-repair-old-live", + ]); + const oldTombstonesAfter = await prisma.measurement.findMany({ + where: { + userId: "horizon-repair", + deletedAt: { not: null }, + id: { in: oldTombstonesBefore.map((row) => row.id) }, + }, + select: { id: true, deletedAt: true }, + orderBy: { id: "asc" }, + }); + expect(oldTombstonesAfter).toEqual(oldTombstonesBefore); + + const again = await repairFoldedMeans(prisma, "horizon-repair"); + expect(again.byType.WALKING_SPEED).toEqual({ + windowsChecked: 1, + windowsCorrected: 0, + restingCorrected: 0, + windowsSkippedBeyondHorizon: 1, + samplesAbsorbed: 0, + }); + expect( + await liveValue( + "horizon-repair", + "WALKING_SPEED", + `stats:${WS_HK}:${RECENT_DAY}`, + ), + ).toBeCloseTo(1.6, 9); + }); + + it("leaves an hourly mean before the hourly horizon as it is", async () => { + const prisma = getPrismaClient(); + await account("horizon-repair-hourly"); + await seed( + "horizon-repair-hourly", + [ + { + id: "hour-mean", + type: "HEART_RATE_VARIABILITY", + value: 60, + at: `${OLD_DAY}T10:30:00.000Z`, + }, + ], + [ + { + id: "hour-t1", + type: "HEART_RATE_VARIABILITY", + value: 20, + at: `${OLD_DAY}T10:05:00.000Z`, + // Folded 90 days on, kept 75 more: purged by now in reality, here + // the one survivor of a fragment. + deletedAt: "2026-06-10T10:30:00.000Z", + }, + ], + ); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + + const result = await repairFoldedMeans(prisma, "horizon-repair-hourly"); + expect(result.byType.HEART_RATE_VARIABILITY).toMatchObject({ + windowsChecked: 0, + windowsCorrected: 0, + windowsSkippedBeyondHorizon: 1, + }); + expect( + await liveValue( + "horizon-repair-hourly", + "HEART_RATE_VARIABILITY", + `stats:${HRV_HK}:${OLD_DAY}T10`, + ), + ).toBe(60); + }); +}); + +describe("regular folds before the horizon", () => { + it("the daily-mean consolidation keeps an old stored mean and its live sample, and folds an old day without one", async () => { + const prisma = getPrismaClient(); + await account("horizon-mean"); + await seedOldDay("horizon-mean"); + // An old day nobody folded yet: no stored mean to protect. + await seed( + "horizon-mean", + [], + [ + { + id: "new-a", + type: "WALKING_SPEED", + value: 1, + at: "2026-03-12T08:00:00.000Z", + }, + { + id: "new-b", + type: "WALKING_SPEED", + value: 2, + at: "2026-03-12T09:00:00.000Z", + }, + ], + ); + const { consolidateDailyMean } = + await import("@/lib/measurements/consolidate-daily-mean"); + + const summary = await consolidateDailyMean(prisma, { + userId: "horizon-mean", + cutoffHours: 36, + log: () => {}, + }); + + expect(summary.totals.daysLeftAsStored).toBe(1); + expect(summary.totals.daysConsolidated).toBe(1); + expect( + await liveValue( + "horizon-mean", + "WALKING_SPEED", + `stats:${WS_HK}:${OLD_DAY}`, + ), + ).toBe(3); + expect( + await liveValue( + "horizon-mean", + "WALKING_SPEED", + `stats:${WS_HK}:2026-03-12`, + ), + ).toBeCloseTo(1.5, 9); + expect(await liveSampleIds("horizon-mean", "WALKING_SPEED")).toEqual([ + "horizon-mean-old-live", + ]); + + // The next night changes nothing either. + const again = await consolidateDailyMean(prisma, { + userId: "horizon-mean", + cutoffHours: 36, + log: () => {}, + }); + expect(again.totals.daysConsolidated).toBe(0); + expect( + await liveValue( + "horizon-mean", + "WALKING_SPEED", + `stats:${WS_HK}:${OLD_DAY}`, + ), + ).toBe(3); + }); + + it("the dense hourly fold keeps an old stored hour and its live sample", async () => { + const prisma = getPrismaClient(); + await account("horizon-dense"); + await seed( + "horizon-dense", + [ + { + id: "hour-mean", + type: "HEART_RATE_VARIABILITY", + value: 60, + at: `${OLD_DAY}T10:30:00.000Z`, + }, + ], + [ + { + id: "hour-t1", + type: "HEART_RATE_VARIABILITY", + value: 20, + at: `${OLD_DAY}T10:05:00.000Z`, + deletedAt: "2026-06-10T10:30:00.000Z", + }, + { + id: "hour-live", + type: "HEART_RATE_VARIABILITY", + value: 20, + at: `${OLD_DAY}T10:10:00.000Z`, + }, + // Another hour of the same day, never folded: folds as usual. + { + id: "other-hour", + type: "HEART_RATE_VARIABILITY", + value: 50, + at: `${OLD_DAY}T14:10:00.000Z`, + }, + ], + ); + const { runDenseIntradayRetention } = + await import("@/lib/measurements/dense-intraday-retention"); + + const summary = await runDenseIntradayRetention(prisma, { + userId: "horizon-dense", + log: () => {}, + }); + + expect(summary.totals.hoursLeftAsStored).toBe(1); + expect( + await liveValue( + "horizon-dense", + "HEART_RATE_VARIABILITY", + `stats:${HRV_HK}:${OLD_DAY}T10`, + ), + ).toBe(60); + expect( + await liveValue( + "horizon-dense", + "HEART_RATE_VARIABILITY", + `stats:${HRV_HK}:${OLD_DAY}T14`, + ), + ).toBe(50); + expect( + await liveSampleIds("horizon-dense", "HEART_RATE_VARIABILITY"), + ).toEqual(["horizon-dense-hour-live"]); + }); +}); + +describe("repair and consolidation together", () => { + it("end on the mean over every sample, and a later repair corrects nothing", async () => { + const prisma = getPrismaClient(); + await account("together"); + await seedRecentDay("together"); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + const { consolidateDailyMean } = + await import("@/lib/measurements/consolidate-daily-mean"); + + await Promise.all([ + repairFoldedMeans(prisma, "together"), + consolidateDailyMean(prisma, { + userId: "together", + cutoffHours: 36, + log: () => {}, + }), + ]); + expect( + await liveValue( + "together", + "WALKING_SPEED", + `stats:${WS_HK}:${RECENT_DAY}`, + ), + ).toBeCloseTo(1.6, 9); + expect(await liveSampleIds("together", "WALKING_SPEED")).toEqual([]); + + const again = await repairFoldedMeans(prisma, "together"); + expect(again.byType.WALKING_SPEED?.windowsCorrected).toBe(0); + expect( + await liveValue( + "together", + "WALKING_SPEED", + `stats:${WS_HK}:${RECENT_DAY}`, + ), + ).toBeCloseTo(1.6, 9); + }); + + it("a consolidation that ran first leaves the repair nothing to correct", async () => { + const prisma = getPrismaClient(); + await account("fold-first"); + await seedRecentDay("fold-first"); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + const { consolidateDailyMean } = + await import("@/lib/measurements/consolidate-daily-mean"); + + await consolidateDailyMean(prisma, { + userId: "fold-first", + cutoffHours: 36, + log: () => {}, + }); + expect( + await liveValue( + "fold-first", + "WALKING_SPEED", + `stats:${WS_HK}:${RECENT_DAY}`, + ), + ).toBeCloseTo(1.6, 9); + + const repair = await repairFoldedMeans(prisma, "fold-first"); + expect(repair.byType.WALKING_SPEED?.windowsCorrected).toBe(0); + expect( + await liveValue( + "fold-first", + "WALKING_SPEED", + `stats:${WS_HK}:${RECENT_DAY}`, + ), + ).toBeCloseTo(1.6, 9); + }); +});