tom/HealthLog
1
0 Fork 0
mirror of https://github.com/MBombeck/HealthLog.git synced 2026-11-07 04:12:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(measurements): recompute a stored fold mean only where all its samples remain

Browse Source 
The fold repair and the regular folds rewrote an existing daily or hourly
stats: mean from the samples still in the table. Fold tombstones are purged
75 days after they were written, so for an older window those samples can
be a fragment: a re-upload after an anchor reset, the rows of one later
fold run. A mean over them replaced a stored mean that had been taken from
the whole day, and on old days it moved a lot.

A stored mean is now recomputed only for a day that starts at or after the
tombstone horizon (foldConstituentHorizon): every fold tombstone of such a
day was written after dayStart + the fold threshold and is still inside the
retention, so the live samples plus the fold leftovers are the whole day.
That is the last 75 days for daily means and, since dense days fold 90 days
late, the 75 days before the raw window for hourly means. A window also
needs at least one fold leftover; without one the rows in hand are not
known to be all of it.

Beyond the horizon:
- The repair leaves the day as it is and counts it per type as
  windowsSkippedBeyondHorizon.
- The daily-mean consolidation and the dense hourly fold leave a day or
  hour that already has a live stats: row exactly as it is, the stored
  mean and its live samples alike (daysLeftAsStored, hoursLeftAsStored).
  Up to v1.41 the fold overwrote such a row with the mean of the live rows
  alone, and the first v1.42 cut with the mean of those plus whatever
  tombstones were left; both are a mean of part of a day. Dropping the
  live rows instead would lose samples that may not be in the stored
  mean, so they stay as synced. They are few, the folded_window guard
  refuses new ones, and the nightly pass walks over them without writing.
  A day or hour without a stored mean still folds as before.

Inside the horizon, a pass that recomputes a stored mean soft-deletes the
live samples it took in as fold leftovers instead of deleting them, so a
second pass computes the same mean from the same samples. Before, a fold
that deleted them left the repair a fragment, and the repair then rewrote
the mean from the tombstones alone.

The repair, the mean consolidation and the dense fold now take a per-account
advisory lock after the restore lock in every day transaction, and the
repair reads the day's samples inside it, so the two never compute the
same window from two views of its samples or write the same row at once.
The repair also stops at the fold boundary: a day the fold has not
finished is the fold's to complete.
This commit is contained in:
Marc-André Bombeck committed 2026-10-09 01:03:02 +02:00
1 parent bb5e439553
commit cbc3b3118d
18 files changed
+1353 -274
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/__tests__/measurement-write-tail-guard.test.ts
+3 -1
View File
@@ -103,7 +103,9 @@ const WRITE_SITES: Record<string, string> = {
"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[] {
src/lib/export/restore-lock.ts
+22
View File
@@ -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<void> {
await tx.$queryRaw`
SELECT 1 AS locked
FROM pg_advisory_xact_lock(hashtext('measurement-fold'), hashtext(${userId}))
`;
}
src/lib/jobs/job-outcome.ts
+4
View File
@@ -143,6 +143,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet<string> = 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",
src/lib/jobs/mean-consolidation.ts
+1
View File
@@ -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,
},
},
src/lib/jobs/measurement-fold-repair.ts
+202 -73
View File
@@ -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<string, FoldRepairTypeCounts>;
/** 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<FoldRepairResult> {
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<void> {
const { userId, tz, dateKey, hk, step, constituents, counts } = input;
// Window id → the values of every sample in it.
const byWindow = new Map<string, number[]>();
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<string, FoldRepairTypeCounts> = {};
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,
});
src/lib/measurements/__tests__/consolidate-daily-mean.test.ts
+155 -44
View File
@@ -22,6 +22,7 @@ import type { MeasurementType, PrismaClient } from "@/generated/prisma/client";
vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
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<typeof vi.fn>,
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<typeof vi.fn>,
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 () => {
src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts
+1
View File
@@ -41,6 +41,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock";
vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
holdAccountAgainstRestore: vi.fn(async () => {}),
holdAccountFoldLock: vi.fn(async () => {}),
}));
function pulseRow(id: string, value: number, iso: string): PerSampleRow {
src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts
+1
View File
@@ -40,6 +40,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock";
vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
holdAccountAgainstRestore: vi.fn(async () => {}),
holdAccountFoldLock: vi.fn(async () => {}),
}));
/** A day well outside the retention window, so the fold definitely runs. */
src/lib/measurements/__tests__/dense-intraday-retention.test.ts
+1
View File
@@ -34,6 +34,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock";
vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
holdAccountAgainstRestore: vi.fn(async () => {}),
holdAccountFoldLock: vi.fn(async () => {}),
}));
function row(
src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts
+1
View File
@@ -37,6 +37,7 @@ import { candidateLookup, createManyVia } from "./hourly-mint-mock";
vi.mock("@/lib/export/restore-lock", async (importOriginal) => ({
...(await importOriginal<typeof import("@/lib/export/restore-lock")>()),
holdAccountAgainstRestore: vi.fn(async () => {}),
holdAccountFoldLock: vi.fn(async () => {}),
}));
function spo2Row(id: string, value: number, iso: string): PerSampleRow {
src/lib/measurements/__tests__/hourly-mint-mock.ts
+4 -3
View File
@@ -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
? []
: [
{
src/lib/measurements/consolidate-daily-mean.ts
+150 -70
View File
@@ -30,8 +30,10 @@
* 4. UPSERT the canonical daily row keyed by
* `stats:<HKIdentifier>:<dateKey>` 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<DayWriteOutcome> {
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;
src/lib/measurements/consolidation-base.ts
+9 -2
View File
@@ -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
src/lib/measurements/dense-intraday-retention.ts
+143 -74
View File
@@ -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<number, FoldConstituent[]>();
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<string | null>();
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<number, FoldConstituent[]>();
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<ReturnType<typeof foldOnce>>;
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;
src/lib/measurements/fold-constituents.ts
+94
View File
@@ -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<Prisma.TransactionClient, "measurement">,
input: {
userId: string;
type: MeasurementType;
source: MeasurementSource;
from: Date;
to: Date;
},
): Promise<FoldConstituent[]> {
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<Prisma.TransactionClient, "measurement">,
ids: readonly string[],
at: Date = new Date(),
): Promise<number> {
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;
tests/integration/dense-intraday-retention.test.ts
+16 -7
View File
@@ -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<ReturnType<typeof runDenseIntradayRetention>>;
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({
tests/integration/fold-complete-windows.test.ts
+13
View File
@@ -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");
tests/integration/fold-horizon.test.ts
+533
View File
@@ -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);
});
});