diff --git a/docs/ops/measurement-maintenance.md b/docs/ops/measurement-maintenance.md index 070b7b0a6..c6890196b 100644 --- a/docs/ops/measurement-maintenance.md +++ b/docs/ops/measurement-maintenance.md @@ -4,19 +4,57 @@ Since v1.42 the nightly folds of Apple Health data delete the raw samples they fold into hourly or daily rows, instead of keeping them as deleted rows for 75 days. On an instance that has synced Apple Health for a while, those leftovers ("compaction tombstones") were most of the `measurements` table and of its -indexes. Two pieces clean them up: +indexes. Three pieces deal with them, in this order: -1. **The backlog purge** runs by itself. Every worker boot queues one +1. **The fold repair** runs by itself, once per account, before anything is + deleted. Up to v1.42 the folds could fold a day in two runs, and the + second run stored the mean of the later part of the day only. The repair + recomputes those hourly and daily means from all of the day's samples, + including the compaction tombstones, and corrects a stored mean only when + it differs. When it has finished an account it records that in + `measurement_fold_repairs`. +2. **The backlog purge** runs by itself, after the repair. Every worker boot queues one `compaction-tombstone-purge` job. It deletes the leftovers 5,000 rows at a time, one account at a time, pausing briefly between batches, and stops after 40 batches; a run that stopped with work left queues the next one a minute later. Rows a person deleted are never touched by it: they keep the - 75-day retention of the nightly tombstone cleanup. -2. **The table maintenance** is yours to start, after the purge. It runs + 75-day retention of the nightly tombstone cleanup. The purge leaves an + account alone until the repair has finished it, and a boot queues no purge + run while the repair has finished no account. +3. **The table maintenance** is yours to start, after the purge. It runs `VACUUM (ANALYZE)` on `measurements` and then rebuilds each of its indexes with `REINDEX INDEX CONCURRENTLY`, largest first. The purge leaves the indexes full of empty pages, and only a rebuild gives that space back. +## Repair before purge + +The order matters because the purge deletes what the repair reads. Do not +delete compaction tombstones by hand, and do not insert rows into +`measurement_fold_repairs` to hurry the purge along: a mean the repair has not +seen keeps its wrong value for good once its tombstones are gone. + +The repair reports every run on the wide event `job.measurement_fold_repair` +(`fold_repair_means_checked`, `fold_repair_means_corrected`, +`fold_repair_resting_corrected`, and `fold_repair_by_type` with the same +counts per type; counts only, never a value). It is finished when every +account has a row: + +```sql +SELECT count(*) AS accounts_left +FROM users u +LEFT JOIN measurement_fold_repairs r ON r.user_id = u.id +WHERE r.user_id IS NULL; +``` + +Each run works on one account, a day at a time, and stops on its time budget +with a follow-up that resumes at the next day; an account under restore is +tried again five minutes later. While accounts are left, purge runs report +them as `compaction_purge_awaiting_repair_accounts`. + +A restore of a backup exported before the account's repair finished writes +the compaction tombstones back, removes the account's row and queues the +repair again; the purge follows once the repair is through. + ## Is the purge finished? The purge reports every run on the wide event `job.compaction_tombstone_purge` diff --git a/prisma/migrations/0389_measurement_fold_repair/migration.sql b/prisma/migrations/0389_measurement_fold_repair/migration.sql new file mode 100644 index 000000000..c53691304 --- /dev/null +++ b/prisma/migrations/0389_measurement_fold_repair/migration.sql @@ -0,0 +1,12 @@ +-- One row per account once the fold repair has recomputed the daily and +-- hourly means an older release folded in two runs. The compaction-tombstone +-- purge waits for it, because the tombstones it deletes are what the repair +-- reads. Holds no health data; the account's deletion takes it along. +CREATE TABLE "measurement_fold_repairs" ( + "user_id" TEXT NOT NULL, + "completed_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "measurement_fold_repairs_pkey" PRIMARY KEY ("user_id") +); + +ALTER TABLE "measurement_fold_repairs" ADD CONSTRAINT "measurement_fold_repairs_user_id_fkey" FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 72acf6c71..23d5a7755 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1002,6 +1002,7 @@ model User { // move, a birth, a new job. Shared under the `profile` domain; never sent to // a model. lifeEvents LifeEvent[] + measurementFoldRepair MeasurementFoldRepair? // v1.7.0 profile — optional patient-identity fields surfaced on the // health-record export cover (PDF) and the FHIR `Patient` resource. @@ -9641,3 +9642,24 @@ enum LifeEventPrecision { @@map("life_event_precision") } + +/// v1.42 — the fold repair has been through this account. +/// +/// Before v1.42 the daily-mean consolidation and the dense hourly fold could +/// fold a local day in two runs, the second overwriting the stored mean with +/// the mean of the later part alone. The one-time repair +/// (`src/lib/jobs/measurement-fold-repair.ts`) recomputes every such mean from +/// the samples still in the table, the fold's own tombstones among them. The +/// compaction-tombstone purge deletes exactly those tombstones, so it leaves an +/// account alone until this row exists, and a restore that writes back a file +/// older than `completedAt`'s run removes the row so the repair goes again. +model MeasurementFoldRepair { + userId String @id @map("user_id") + /// When the repair finished the account. A restore compares it with the + /// file's export instant; the purge only asks whether the row exists. + completedAt DateTime @map("completed_at") + + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@map("measurement_fold_repairs") +} diff --git a/src/__tests__/measurement-write-tail-guard.test.ts b/src/__tests__/measurement-write-tail-guard.test.ts index ad4cca5a1..7cf3c7943 100644 --- a/src/__tests__/measurement-write-tail-guard.test.ts +++ b/src/__tests__/measurement-write-tail-guard.test.ts @@ -102,6 +102,8 @@ const WRITE_SITES: Record = { "hard-deletes rows that were already soft-deleted and read nowhere", "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", }; function sourceFiles(): string[] { diff --git a/src/lib/data-wipe/wipe-plan.ts b/src/lib/data-wipe/wipe-plan.ts index 29281526d..df6254c80 100644 --- a/src/lib/data-wipe/wipe-plan.ts +++ b/src/lib/data-wipe/wipe-plan.ts @@ -54,6 +54,9 @@ export const WIPE_MODELS = [ // record and nothing else holds it, so a wipe that leaves it behind leaves // behind an account's activity pattern. "IntradayCumulativeProfile", + // That the fold repair went through this account's means. With the readings + // gone it describes nothing; without it the repair simply runs once more. + "MeasurementFoldRepair", // The health score as it was shown, one row per local day. Not a cache of // the live computation — nothing can rebuild a past day once the readings // behind it have moved — so leaving it behind leaves behind a record of how diff --git a/src/lib/export/backup-plan.ts b/src/lib/export/backup-plan.ts index fc5dc798f..9a07f0bce 100644 --- a/src/lib/export/backup-plan.ts +++ b/src/lib/export/backup-plan.ts @@ -720,6 +720,8 @@ export const NOT_IN_BACKUP_MODELS: Readonly> = { "The record of a restore this host ran from one of its own stored copies. It names a backup row that exists only in this database, so carrying it would describe a restore the receiving host never performed.", OffhostBackupState: "When this host last put this account's copy in this operator's bucket, and how big it was. It describes one deployment's relationship with one bucket, so restoring it elsewhere would assert an off-host copy that host has never written.", + MeasurementFoldRepair: + "That this host's one-time fold repair has recomputed the account's hourly and daily means, which lets the compaction-tombstone purge delete the tombstones the repair reads. It describes this database, not the file: a restore of a file exported before the repair ran writes back the means and tombstones of that moment, so the restore removes the row and the repair runs again before anything is purged.", BackupPassAttempt: "When this host's backup passes last started and finished this account, kept so a pass the process died under is taken last next time and named on the backups page. It describes this deployment's own runs, so restoring it elsewhere would report an interrupted run that host never had, or hide one it did.", DocumentImportKey: diff --git a/src/lib/export/restore-backup.ts b/src/lib/export/restore-backup.ts index acdbf8d73..55275dbd0 100644 --- a/src/lib/export/restore-backup.ts +++ b/src/lib/export/restore-backup.ts @@ -125,9 +125,9 @@ import { invalidateUserData } from "@/lib/cache/invalidate"; import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; import { FOLDED_TYPES, - coveringStatsExternalId, - isFoldedWindowCandidate, + compactionTombstoneCoverId, } from "@/lib/measurements/folded-window"; +import { enqueueMeasurementFoldRepair } from "@/lib/jobs/measurement-fold-repair"; import { resolveUserTimezone } from "@/lib/measurements/consolidation-base"; import { stampSyncReset } from "@/lib/sync/reset"; import { @@ -429,19 +429,28 @@ export async function restoreBackup( // the transaction reads them a second time, in batches, as it writes them. // v1.42 — the live `stats:` ids of the folded types, gathered on the first // read, so the measurement pass can tell a compaction tombstone from a - // person's deletion without a third read of the file. + // person's deletion without a third read of the file. Keyed by source as + // well: Health Connect folds into the same `stats:` ids as Apple Health, and + // an Apple Health hour must not make a Health Connect deletion look folded. const liveFoldStatsIds = new Set(); const foldedTypes = new Set(FOLDED_TYPES); + const foldSources = new Set(["APPLE_HEALTH", "HEALTH_CONNECT"]); + const foldStatsKey = (source: string, type: string, externalId: string) => + `${source}|${type}|${externalId}`; const noteLiveFoldRow = (row: BackupMeasurement) => { if ( row.deletedAt == null && - row.source === "APPLE_HEALTH" && + row.source !== undefined && + foldSources.has(row.source) && foldedTypes.has(row.type) && row.externalId?.startsWith("stats:") ) { - liveFoldStatsIds.add(`${row.type}|${row.externalId}`); + liveFoldStatsIds.add(foldStatsKey(row.source, row.type, row.externalId)); } }; + // Set inside the transaction when the restore takes the account's fold + // repair marker away; the repair is queued once the transaction commits. + let foldRepairReset = false; let raw: unknown; let payload; let streamed: StreamedBackup; @@ -1013,8 +1022,9 @@ export async function restoreBackup( Date.now() - TOMBSTONE_RETENTION_DAYS * 86_400_000; // // v1.42 — nor is a compaction tombstone, whatever its age: a raw - // Apple Health sample the folds soft-deleted before v1.42, whose hour - // or day a live `stats:` row in the same file covers. The folds + // sample the folds soft-deleted before v1.42, whose whole hour or day + // the fold had reached when it was deleted and which a live `stats:` + // row of the same source in the same file covers. The folds // delete those outright now and the backlog purge removes the old // ones, so writing one back would only hand the purge the same row // again; the `folded_window` ingest guard is what keeps the sample @@ -1032,21 +1042,47 @@ export async function restoreBackup( )?.timezone ?? null, ); + // Only once the fold repair has been through the means this file + // holds. The repair recomputes a mean an older release folded from + // part of its day, and these tombstones are what it reads. A file + // exported after the account's repair finished carries repaired + // means, so its compaction tombstones can go. Any other file is + // written back whole, the account's repair marker is removed, and the + // repair runs again once the transaction commits; the purge then + // removes the tombstones as usual. + const foldRepair = await tx.measurementFoldRepair.findUnique({ + where: { userId: ownerId }, + select: { completedAt: true }, + }); + const dropCompactionTombstones = + foldRepair !== null && + new Date(payload.exportedAt).getTime() > + foldRepair.completedAt.getTime(); + if (!dropCompactionTombstones) { + await tx.measurementFoldRepair.deleteMany({ + where: { userId: ownerId }, + }); + foldRepairReset = true; + } const isCompactionTombstone = (measurement: BackupMeasurement) => { + if (!dropCompactionTombstones) return false; if (measurement.deletedAt == null) return false; - const row = { - type: measurement.type, - source: measurement.source ?? "MANUAL", - externalId: measurement.externalId ?? null, - measuredAt: new Date(measurement.measuredAt), - }; - if (!isFoldedWindowCandidate(row, new Date(measurement.deletedAt))) { - return false; - } - const covering = coveringStatsExternalId(row, foldTz); + const source = measurement.source ?? "MANUAL"; + const covering = compactionTombstoneCoverId( + { + type: measurement.type, + source, + externalId: measurement.externalId ?? null, + measuredAt: new Date(measurement.measuredAt), + deletedAt: new Date(measurement.deletedAt), + }, + foldTz, + ); return ( covering !== null && - liveFoldStatsIds.has(`${measurement.type}|${covering}`) + liveFoldStatsIds.has( + foldStatsKey(source, measurement.type, covering), + ) ); }; const isExpiredTombstone = (measurement: BackupMeasurement) => @@ -2808,6 +2844,24 @@ export async function restoreBackup( // ran past the per-request statement limit, the fold was dropped, and the // account's charts stayed empty until the next boot (#1031). It gets the // time the deadline check above keeps free for it instead. + // The restore took the fold repair marker away (see the measurement pass): + // queue the repair now that the restored rows are committed. Best-effort; + // the next boot's discovery queues it otherwise, and the purge waits + // either way. + if (foldRepairReset) { + try { + await enqueueMeasurementFoldRepair(ownerId); + } catch (err) { + annotate({ + meta: { + fold_repair_enqueue_failed: true, + fold_repair_enqueue_error: + err instanceof Error ? err.message : String(err), + }, + }); + } + } + if (measurementsRestored > 0) { try { await recomputeUserRollups(ownerId, { diff --git a/src/lib/import/health-connect/import.ts b/src/lib/import/health-connect/import.ts index ec20132fc..a36b8e20f 100644 --- a/src/lib/import/health-connect/import.ts +++ b/src/lib/import/health-connect/import.ts @@ -65,7 +65,10 @@ import { isFoldedWindowCandidate, loadFoldTimezone, } from "@/lib/measurements/folded-window"; -import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz"; +import { + canonicalDailyTimestamp, + foldBoundary, +} from "@/lib/measurements/consolidation-tz"; import { validateMeasurementRange } from "@/lib/validations/measurement"; import { isModuleEnabled } from "@/lib/modules/gate"; import { isCycleAvailableForUser } from "@/lib/cycle/gate"; @@ -482,7 +485,18 @@ class ImportRun { ) { return; } - const cutoff = this.now.getTime() - HC_RAW_PULSE_WINDOW_MS; + // The nightly fold folds this source's raw samples once their local day + // is over in full, 90 days back (`foldBoundary`). Splitting the export at + // the same instant keeps every day on one side: a day cut at + // `now - 90 days` itself got an hourly mean of its first part here and + // raw samples for the rest, and the fold then overwrote the boundary + // hour's mean with the mean of the raw part alone. + this.foldTz ??= await loadFoldTimezone(this.prisma, this.userId); + const cutoff = foldBoundary( + this.now, + HC_RAW_PULSE_WINDOW_MS, + this.foldTz, + ).getTime(); await this.importHourlyPulse(cutoff); await this.importRawPulse(cutoff); } diff --git a/src/lib/jobs/compaction-tombstone-purge.ts b/src/lib/jobs/compaction-tombstone-purge.ts index 3c9d17fa7..0b2c87690 100644 --- a/src/lib/jobs/compaction-tombstone-purge.ts +++ b/src/lib/jobs/compaction-tombstone-purge.ts @@ -35,6 +35,13 @@ * and every boot queues one run. * - Every run reports its outcome per account (`reportJobRun`) and its * totals as job facts. + * + * Repair first. Some of these tombstones are the only remaining record of + * samples whose day an older release folded in two runs, leaving a mean of + * part of the day (`measurement-fold-repair.ts`). The purge therefore leaves + * an account alone until the repair has been through it (a + * `MeasurementFoldRepair` row), and the boot sends no run while no account + * has one. The repair queues a run as it finishes an account. */ import { createHash } from "node:crypto"; import type { Job } from "pg-boss"; @@ -84,6 +91,10 @@ export interface CompactionTombstonePurgeOutcome { scanned: number; /** The deferred accounts' ids. Never leaves the process unhashed. */ deferredUserIds: string[]; + /** Accounts left alone because the fold repair has not finished them. */ + awaitingRepairAccounts: number; + /** Their ids. Never leaves the process unhashed. */ + awaitingRepairUserIds: string[]; } export interface PurgeCompactionTombstonesOptions { @@ -115,6 +126,8 @@ export async function purgeCompactionTombstones( const scanPageSize = options.scanPageSize ?? PURGE_BATCH_SIZE; const pauseMs = options.pauseMs ?? BATCH_PAUSE_MS; const timezoneOf = new Map(); + const repaired = new Map(); + const awaitingRepair = new Set(); let scanned = 0; // The walk's position. Rows behind it are classified; the class-A ones wait @@ -135,6 +148,17 @@ export async function purgeCompactionTombstones( } } + async function loadRepaired(userIds: string[]): Promise { + const missing = userIds.filter((id) => !repaired.has(id)); + if (missing.length === 0) return; + const rows = await prisma.measurementFoldRepair.findMany({ + where: { userId: { in: missing } }, + select: { userId: true }, + }); + const done = new Set(rows.map((row) => row.userId)); + for (const id of missing) repaired.set(id, done.has(id)); + } + async function classifyNextPage(skipUserIds: ReadonlySet) { const page: ScannedRow[] = await prisma.measurement.findMany({ where: { @@ -175,7 +199,13 @@ export async function purgeCompactionTombstones( else byAccount.set(row.userId, [row]); } await timezoneFor([...byAccount.keys()]); + await loadRepaired([...byAccount.keys()]); for (const [userId, rows] of byAccount) { + // The repair reads these tombstones; nothing goes before it has run. + if (!repaired.get(userId)) { + awaitingRepair.add(userId); + continue; + } const tz = timezoneOf.get(userId); // An account deleted since the page was read has nothing left to purge. if (!tz) continue; @@ -234,6 +264,8 @@ export async function purgeCompactionTombstones( deferredAccounts: outcome.deferredAccounts, scanned, deferredUserIds: outcome.deferredUserIds, + awaitingRepairAccounts: awaitingRepair.size, + awaitingRepairUserIds: [...awaitingRepair], }; } @@ -259,13 +291,18 @@ export async function handleCompactionTombstonePurge( return withBackgroundEvent("job.compaction_tombstone_purge", async (evt) => { const startedAt = Date.now(); const result = await purgeCompactionTombstones(prisma, { shouldStop }); - const candidates: JobRunCandidate[] = result.deferredUserIds.map( - (userId) => ({ + const candidates: JobRunCandidate[] = [ + ...result.deferredUserIds.map((userId): JobRunCandidate => ({ key: accountKey(userId), outcome: "deferred", cause: "restore_in_progress", - }), - ); + })), + ...result.awaitingRepairUserIds.map((userId): JobRunCandidate => ({ + key: accountKey(userId), + outcome: "skipped", + cause: "awaiting_fold_repair", + })), + ]; reportJobRun({ queue: COMPACTION_TOMBSTONE_PURGE_QUEUE, runId: jobs[0]?.id ?? "unknown", @@ -276,6 +313,10 @@ export async function handleCompactionTombstonePurge( evt.addMeta("compaction_purge_drained", result.drained); evt.addMeta("compaction_purge_deferred_accounts", result.deferredAccounts); evt.addMeta("compaction_purge_continuation", continuation); + evt.addMeta( + "compaction_purge_awaiting_repair_accounts", + result.awaitingRepairAccounts, + ); // A run that stopped with work left sends the next one, but only when it // made progress, so a run that cannot delete anything cannot loop. @@ -318,13 +359,21 @@ export async function handleCompactionTombstonePurge( /** * Queue one purge run when the worker boots. The run is self-limiting and - * idempotent, so a boot that finds no backlog costs one cheap scan. + * idempotent, so a boot that finds no backlog costs one cheap scan. While the + * fold repair has finished no account, there is nothing the run may delete, + * so none is sent; the repair sends one as it finishes an account. */ -export async function enqueueBootTimeCompactionTombstonePurge(): Promise<{ +export async function enqueueBootTimeCompactionTombstonePurge( + prisma: Pick = defaultPrisma, +): Promise<{ enqueued: boolean; }> { const boss = getGlobalBoss(); if (!boss) return { enqueued: false }; + const anyRepaired = await prisma.measurementFoldRepair.findFirst({ + select: { userId: true }, + }); + if (!anyRepaired) return { enqueued: false }; const id = await boss.send( COMPACTION_TOMBSTONE_PURGE_QUEUE, { continuation: 0 }, diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 61f9733b5..5ed38d850 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -43,6 +43,7 @@ export type JobFacts = Readonly>; */ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "access_tokens_deleted", + "accounts_completed", // v1.40 — the document AI run reaper's three outcomes. "ai_runs_deleted", "ai_runs_timed_out", @@ -138,6 +139,10 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "manual_mints_removed", "markers", "mean_days_consolidated", + // v1.42 — the one-time fold repair: windows whose stored mean was compared + // with the mean over every sample, and those it corrected. Counts only. + "means_checked", + "means_corrected", "mean_rows_soft_deleted", "measurements_imported", "measurements_migrated", @@ -188,6 +193,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "refused", "reminders_due", "removed", + "resting_corrected", "rows_written", // v1.39.1 — the background restore of a stored backup, and the sweep that // re-queues one a stopped worker left running. Counts and a claim flag. diff --git a/src/lib/jobs/measurement-fold-repair.ts b/src/lib/jobs/measurement-fold-repair.ts new file mode 100644 index 000000000..9fa1019e3 --- /dev/null +++ b/src/lib/jobs/measurement-fold-repair.ts @@ -0,0 +1,626 @@ +/** + * One-time repair of the hourly and daily means a fold computed from part of + * their window (v1.42). + * + * Up to v1.42 the daily-mean consolidation and the dense hourly fold cut at + * `now - threshold` instead of a local day boundary. A day was folded in two + * runs, and the second run overwrote the stored mean with the mean of the + * 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. + * + * 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; + * - 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. + * + * When an account is through, a `MeasurementFoldRepair` row records it. The + * compaction-tombstone purge deletes exactly the tombstones this pass reads, + * so it leaves an account alone until that row exists, and the repair queues a + * 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. + * + * The queue name MUST be registered in `allQueues` in + * `src/lib/jobs/reminder/register-maintenance.ts`, or the sends silently + * no-op. + */ +import { createHash } from "node:crypto"; +import type { Job } from "pg-boss"; + +import type { + MeasurementSource, + MeasurementType, + PrismaClient, +} from "@/generated/prisma/client"; +import { prisma as defaultPrisma } from "@/lib/db"; +import { + AccountRestoreInProgressError, + holdAccountAgainstRestore, +} from "@/lib/export/restore-lock"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { + COMPACTION_TOMBSTONE_PURGE_QUEUE, + COMPACTION_TOMBSTONE_PURGE_SINGLETON_KEY, +} from "@/lib/jobs/compaction-tombstone-purge"; +import { jobBudget } from "@/lib/jobs/job-budget"; +import { jobDone, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { reportJobRun, type JobRunCandidate } from "@/lib/jobs/job-run-report"; +import { annotate } from "@/lib/logging/context"; +import { logCaught } from "@/lib/logging/signal"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { + HIGH_FREQUENCY_MEAN_TYPES, + dailyStatsExternalId, + hkIdentifierForType, +} from "@/lib/measurements/apple-health-mapping"; +import { + iterateDayBuckets, + iterateSourcePages, + resolveUserTimezone, +} from "@/lib/measurements/consolidation-base"; +import { + CONSOLIDATION_GRACE_CUTOFF_HOURS, + canonicalDailyTimestamp, + hourOfDayForUserTz, +} from "@/lib/measurements/consolidation-tz"; +import { + DENSE_INTRADAY_FOLD_SOURCES, + DENSE_INTRADAY_RETENTION_DAYS, + DENSE_INTRADAY_RETENTION_TYPES, + deriveDailyRestingFromPulse, + hourlyStatsExternalId, +} from "@/lib/measurements/dense-intraday-retention"; +import { + isFoldLeftover, + meanDiffers, + meanOf, +} from "@/lib/measurements/fold-constituents"; +import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; +import { localDayWindow } from "@/lib/tz/local-day"; + +export const MEASUREMENT_FOLD_REPAIR_QUEUE = "measurement-fold-repair"; + +/** Delay before the follow-up of a run that stopped early, in seconds. */ +export const FOLD_REPAIR_CONTINUATION_DELAY_SECONDS = 30; + +/** Delay before retrying an account a restore was holding, in seconds. */ +export const FOLD_REPAIR_RESTORE_RETRY_SECONDS = 300; + +/** Follow-up runs one account's chain may send. */ +export const FOLD_REPAIR_MAX_CONTINUATIONS = 200; + +const HOUR_MS = 3_600_000; +const DAY_MS = 24 * HOUR_MS; + +/** HealthKit identifier of the derived resting row (`dense-intraday-retention.ts`). */ +const RESTING_HK_IDENTIFIER = "HKQuantityTypeIdentifierRestingHeartRate"; + +/** One (type, source) the folds write means for, and the grain. */ +interface RepairStep { + type: MeasurementType; + source: MeasurementSource; + grain: "day" | "hour"; + foldedAfterMs: number; +} + +/** Fixed order, so a resume position (`step`) means the same on every run. */ +export const FOLD_REPAIR_STEPS: readonly RepairStep[] = [ + ...[...HIGH_FREQUENCY_MEAN_TYPES].map((type): RepairStep => ({ + type, + source: "APPLE_HEALTH", + grain: "day", + foldedAfterMs: CONSOLIDATION_GRACE_CUTOFF_HOURS * HOUR_MS, + })), + ...DENSE_INTRADAY_FOLD_SOURCES.flatMap((source) => + [...DENSE_INTRADAY_RETENTION_TYPES].map((type): RepairStep => ({ + type, + source, + grain: "hour", + foldedAfterMs: DENSE_INTRADAY_RETENTION_DAYS * DAY_MS, + })), + ), +]; + +export interface FoldRepairResume { + /** Index into {@link FOLD_REPAIR_STEPS}. */ + step: number; + /** ISO instant: the step resumes at rows measured at or after it. */ + after?: string; +} + +export interface MeasurementFoldRepairPayload { + userId: string; + resume?: FoldRepairResume; + continuation?: number; +} + +/** Per-type counts. Never a value. */ +export interface FoldRepairTypeCounts { + /** Hours or days with a live `stats:` row and fold leftovers. */ + windowsChecked: number; + /** Of those, the ones whose stored mean was corrected. */ + windowsCorrected: number; + /** Derived resting rows corrected (heart rate only). */ + restingCorrected: number; +} + +export interface FoldRepairResult { + status: "completed" | "stopped" | "restore_in_progress"; + /** Where the next run starts, when `status` is `stopped`. */ + resume?: FoldRepairResume; + byType: Record; + /** Days with fold leftovers the pass looked at. */ + daysWithLeftovers: number; +} + +type ScanRow = { + id: string; + type: MeasurementType; + value: number; + measuredAt: Date; + externalId: string | null; + deletedAt: Date | null; + syncVersion: number; +}; + +function countsFor( + result: FoldRepairResult, + type: MeasurementType, +): FoldRepairTypeCounts { + let counts = result.byType[type]; + if (!counts) { + counts = { windowsChecked: 0, windowsCorrected: 0, restingCorrected: 0 }; + result.byType[type] = counts; + } + return counts; +} + +/** + * Repair one account. Resolves with `stopped` and a resume position when + * `shouldStop` ends the walk between two days, with `restore_in_progress` + * when a restore of the account holds it, and with `completed` otherwise. + */ +export async function repairFoldedMeans( + prisma: PrismaClient, + userId: string, + options: { + resume?: FoldRepairResume; + shouldStop?: () => boolean; + pageSize?: number; + } = {}, +): Promise { + const result: FoldRepairResult = { + status: "completed", + byType: {}, + daysWithLeftovers: 0, + }; + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { timezone: true }, + }); + if (!user) return result; + const tz = resolveUserTimezone(user.timezone); + const shouldStop = options.shouldStop ?? (() => false); + const firstStep = options.resume?.step ?? 0; + + for ( + let stepIndex = firstStep; + stepIndex < FOLD_REPAIR_STEPS.length; + stepIndex++ + ) { + const step = FOLD_REPAIR_STEPS[stepIndex]; + const hk = hkIdentifierForType(step.type); + if (!hk) continue; + + // Nothing of this type and source was ever folded: no `stats:` row. + const anyStats = await prisma.measurement.findFirst({ + where: { + userId, + type: step.type, + source: step.source, + externalId: { startsWith: `stats:${hk}:` }, + deletedAt: null, + }, + 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); + const after = + stepIndex === firstStep && options.resume?.after + ? new Date(options.resume.after) + : null; + + const pages = iterateSourcePages( + prisma, + { + userId, + type: step.type, + source: step.source, + externalId: { not: null }, + NOT: [ + { externalId: { startsWith: "stats:" } }, + { externalId: { startsWith: "retired:" } }, + ], + OR: [{ deletedAt: null }, { deletedAt: { not: null }, syncVersion: 1 }], + measuredAt: after ? { gte: after, lt: upper } : { lt: upper }, + }, + { + id: true, + type: true, + value: true, + measuredAt: true, + externalId: true, + deletedAt: true, + syncVersion: true, + }, + options.pageSize, + ); + + for await (const [dateKey, dayRows] of iterateDayBuckets(pages, tz, null)) { + if (shouldStop()) { + result.status = "stopped"; + result.resume = { + step: stepIndex, + after: localDayWindow(dateKey, tz).dayStart.toISOString(), + }; + return result; + } + const leftovers = dayRows.filter((row) => + isFoldLeftover(row, step.foldedAfterMs), + ); + if (leftovers.length === 0) continue; + result.daysWithLeftovers += 1; + const constituents = [ + ...dayRows.filter((row) => row.deletedAt === null), + ...leftovers, + ]; + try { + await repairDay(prisma, { + userId, + tz, + dateKey, + hk, + step, + constituents, + counts: countsFor(result, step.type), + }); + } catch (err) { + if (err instanceof AccountRestoreInProgressError) { + result.status = "restore_in_progress"; + result.resume = { + step: stepIndex, + after: localDayWindow(dateKey, tz).dayStart.toISOString(), + }; + return result; + } + throw err; + } + } + } + return result; +} + +/** Recompute one day's means from its constituents; write only differences. */ +async function repairDay( + prisma: PrismaClient, + input: { + userId: string; + tz: string; + dateKey: string; + 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 restingExternalId = + step.type === "PULSE" && step.source === "APPLE_HEALTH" + ? dailyStatsExternalId( + RESTING_HK_IDENTIFIER, + // The fold keys the derived row by the UTC date of local noon. + // eslint-disable-next-line healthlog/no-utc-day-key -- UTC by design: matches the derived resting row's stats: key in dense-intraday-retention.ts + canonicalDailyTimestamp(dateKey, tz).toISOString().slice(0, 10), + ) + : null; + + let correctedDayRow: Date | null = null; + await prisma.$transaction(async (tx) => { + // First, before any reading is touched: see `restore-lock.ts`. + await holdAccountAgainstRestore(tx, userId); + const stored = await tx.measurement.findMany({ + where: { + userId, + type: step.type, + source: step.source, + externalId: { in: [...byWindow.keys()] }, + deletedAt: null, + }, + select: { id: true, externalId: true, value: true, measuredAt: true }, + }); + 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); + 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; + } + + if (restingExternalId) { + const resting = await tx.measurement.findFirst({ + where: { + userId, + type: "RESTING_HEART_RATE", + source: "COMPUTED", + externalId: restingExternalId, + deletedAt: null, + }, + select: { id: true, value: true }, + }); + const derived = resting + ? deriveDailyRestingFromPulse( + constituents.map((row) => ({ + id: row.id, + type: row.type, + value: row.value, + measuredAt: row.measuredAt, + externalId: null, + })), + ) + : null; + if (resting && derived !== null && resting.value !== derived) { + await tx.measurement.update({ + where: { id: resting.id }, + data: { value: derived, syncVersion: { increment: 1 } }, + }); + counts.restingCorrected += 1; + } + } + }); + + // 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); + } +} + +/** Record that the repair has been through an account. */ +export async function markFoldRepairComplete( + prisma: PrismaClient, + userId: string, + at: Date = new Date(), +): Promise { + await prisma.measurementFoldRepair.upsert({ + where: { userId }, + create: { userId, completedAt: at }, + update: { completedAt: at }, + }); +} + +/** A stable, non-identifying key for one account in the run report. */ +function accountKey(userId: string): string { + return createHash("sha256").update(userId).digest("hex").slice(0, 12); +} + +/** + * One run per account. A finished account gets its marker and queues a purge + * run; one the budget stopped sends its own follow-up; one under restore is + * tried again a few minutes later. + */ +export async function handleMeasurementFoldRepair( + jobs: Job[], + prisma: PrismaClient = defaultPrisma, +): Promise { + const shouldStop = jobBudget(jobs); + 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 byType: Record = {}; + const boss = getGlobalBoss(); + + for (const job of jobs) { + const userId = job.data?.userId; + if (!userId) continue; + const continuation = job.data?.continuation ?? 0; + let result: FoldRepairResult; + try { + result = await repairFoldedMeans(prisma, userId, { + resume: job.data?.resume, + shouldStop, + }); + } catch (err) { + // On the run report as `failed` (which raises `job.run.failed` / + // `job.run.partial`), with the cause logged here. The next boot + // retries the account, its marker still absent. + logCaught("measurement.fold_repair.failed", err); + candidates.push({ + key: accountKey(userId), + outcome: "failed", + cause: "repair_error", + }); + continue; + } + for (const [type, counts] of Object.entries(result.byType)) { + const sum = (byType[type] ??= { + windowsChecked: 0, + windowsCorrected: 0, + restingCorrected: 0, + }); + sum.windowsChecked += counts.windowsChecked; + sum.windowsCorrected += counts.windowsCorrected; + sum.restingCorrected += counts.restingCorrected; + totals.checked += counts.windowsChecked; + totals.corrected += counts.windowsCorrected; + totals.resting += counts.restingCorrected; + } + + if (result.status === "completed") { + await markFoldRepairComplete(prisma, userId); + totals.completed += 1; + candidates.push({ key: accountKey(userId), outcome: "ok" }); + continue; + } + candidates.push({ + key: accountKey(userId), + outcome: "deferred", + cause: + result.status === "stopped" ? "job_budget" : "restore_in_progress", + }); + if (boss && continuation < FOLD_REPAIR_MAX_CONTINUATIONS) { + const payload: MeasurementFoldRepairPayload = { + userId, + resume: result.resume, + continuation: continuation + 1, + }; + await boss.send(MEASUREMENT_FOLD_REPAIR_QUEUE, payload, { + singletonKey: foldRepairSingletonKey(userId), + startAfter: + result.status === "stopped" + ? FOLD_REPAIR_CONTINUATION_DELAY_SECONDS + : FOLD_REPAIR_RESTORE_RETRY_SECONDS, + retryLimit: 2, + retryDelay: 60, + }); + } else { + evt.addWarning( + "measurement-fold-repair stopped an account without a follow-up; the next boot resumes it", + ); + } + } + + // A finished account unblocks its share of the purge; one queued run + // covers every account finished so far. + if (boss && totals.completed > 0) { + await boss.send( + COMPACTION_TOMBSTONE_PURGE_QUEUE, + { continuation: 0 }, + { + singletonKey: COMPACTION_TOMBSTONE_PURGE_SINGLETON_KEY, + startAfter: FOLD_REPAIR_CONTINUATION_DELAY_SECONDS, + retryLimit: 0, + }, + ); + } + + await reportJobRun({ + queue: MEASUREMENT_FOLD_REPAIR_QUEUE, + runId: jobs[0]?.id ?? "unknown", + candidates, + }); + annotate({ meta: { fold_repair_by_type: byType } }); + evt.addMeta("fold_repair_means_checked", totals.checked); + evt.addMeta("fold_repair_means_corrected", totals.corrected); + evt.addMeta("fold_repair_resting_corrected", totals.resting); + + return jobDone({ + means_checked: totals.checked, + means_corrected: totals.corrected, + resting_corrected: totals.resting, + accounts_completed: totals.completed, + duration_ms: Date.now() - startedAt, + }); + }); +} + +/** One key per account, so a boot's send collapses into a queued run. */ +export function foldRepairSingletonKey(userId: string): string { + return `measurement-fold-repair|${userId}`; +} + +/** + * Queue one repair run for every account the repair has not been through. + * Converges: a finished account has its marker and drops out. Best-effort: + * errors come back in the result so worker boot never fails on it. + */ +export async function enqueueBootTimeMeasurementFoldRepair( + prisma: PrismaClient = defaultPrisma, +): Promise<{ enqueued: number; skipped: number; error: string | null }> { + const boss = getGlobalBoss(); + if (!boss) return { enqueued: 0, skipped: 0, error: null }; + try { + const users = await prisma.user.findMany({ + where: { measurementFoldRepair: null }, + select: { id: true }, + }); + let enqueued = 0; + let skipped = 0; + for (const { id } of users) { + const payload: MeasurementFoldRepairPayload = { userId: id }; + const jobId = await boss.send(MEASUREMENT_FOLD_REPAIR_QUEUE, payload, { + singletonKey: foldRepairSingletonKey(id), + retryLimit: 2, + retryDelay: 60, + }); + if (jobId) enqueued += 1; + else skipped += 1; + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} + +/** Queue a repair run for one account, e.g. after a restore reset its marker. */ +export async function enqueueMeasurementFoldRepair( + userId: string, +): Promise { + const boss = getGlobalBoss(); + if (!boss) return false; + const payload: MeasurementFoldRepairPayload = { userId }; + const id = await boss.send(MEASUREMENT_FOLD_REPAIR_QUEUE, payload, { + singletonKey: foldRepairSingletonKey(userId), + retryLimit: 2, + retryDelay: 60, + }); + return id !== null; +} diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 7baf50ab8..9d3e564c6 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -373,6 +373,14 @@ export const QUEUE_RUNTIME: Readonly> = { "compaction-tombstone-purge": short( "At most 40 batches of 5 000 rows per run; the predicate is idempotent, so the next run resumes.", ), + "measurement-fold-repair": { + runtime: "budgeted", + budget: { + file: "lib/jobs/measurement-fold-repair.ts", + fn: "handleMeasurementFoldRepair", + }, + why: "One account's folded means, a day at a time; stops at three quarters of the default expiry and sends a follow-up that resumes at the next day.", + }, "measurement-maintenance": { runtime: "long", expireInSeconds: MEASUREMENT_MAINTENANCE_SEND_OPTIONS.expireInSeconds, diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 6c1d2d210..72a5a0989 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -67,6 +67,12 @@ import { handleCompactionTombstonePurge, type CompactionTombstonePurgePayload, } from "@/lib/jobs/compaction-tombstone-purge"; +import { + MEASUREMENT_FOLD_REPAIR_QUEUE, + enqueueBootTimeMeasurementFoldRepair, + handleMeasurementFoldRepair, + type MeasurementFoldRepairPayload, +} from "@/lib/jobs/measurement-fold-repair"; import { MEASUREMENT_MAINTENANCE_QUEUE, handleMeasurementMaintenance, @@ -462,6 +468,10 @@ const allQueues = [ // v1.42 — backlog purge of compaction tombstones, queued once at boot. // Without this entry the boot enqueue silently no-ops. COMPACTION_TOMBSTONE_PURGE_QUEUE, + // v1.42 — one-time repair of the means older releases folded from part of + // their window; the purge above waits for it. Without this entry the boot + // enqueue silently no-ops and the purge never runs. + MEASUREMENT_FOLD_REPAIR_QUEUE, // v1.42 — operator-triggered VACUUM / REINDEX of `measurements`. No // cron: the admin route enqueues it. Without this entry the trigger // silently no-ops. @@ -806,6 +816,16 @@ const queuePolicies: QueuePolicyTable = { reason: "Fixed singleton key; collapse queued duplicates only, so the active run can queue its own follow-up.", }, + // v1.42 — per-account fold repair. `short` for the same reason: a run the + // budget stopped sends its own follow-up for the account while it is still + // active. A boot's send for a queued account collapses into it; a duplicate + // beside an active run rewrites nothing, since only differing means are + // written. + [MEASUREMENT_FOLD_REPAIR_QUEUE]: { + policy: "short", + reason: + "Per-account singleton key; collapse queued duplicates only, so the active run can queue its own follow-up.", + }, // Per-document, enqueued on upload. `short`, NOT `exclusive`: each handler // re-reads the document when it starts, so collapsing sends that arrive while @@ -1147,6 +1167,14 @@ export async function registerMaintenanceQueues( { localConcurrency: 1 }, handleCompactionTombstonePurge, ); + // v1.42 — fold repair. Serial: each day holds the account against a + // concurrent restore, and the purge waits for it. + await createAndWork( + boss, + MEASUREMENT_FOLD_REPAIR_QUEUE, + { localConcurrency: 1 }, + handleMeasurementFoldRepair, + ); // v1.42 — measurement table maintenance, operator-triggered only. await createAndWork( boss, @@ -1895,8 +1923,32 @@ export async function enqueueMaintenanceBootDiscovery(): Promise { ); } + // v1.42 — the fold repair, once per account; the purge below waits for it. + try { + const { enqueued, skipped, error } = + await enqueueBootTimeMeasurementFoldRepair(); + if (error) { + workerLog( + "error", + `[measurement-fold-repair] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[measurement-fold-repair] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[measurement-fold-repair] boot discovery threw an unexpected error", + err, + ); + } + // v1.42 — one compaction-tombstone purge run per boot. The run is // self-limiting and idempotent, so a boot with no backlog costs one query. + // It sends nothing until the fold repair has finished an account. try { const { enqueued } = await enqueueBootTimeCompactionTombstonePurge(); workerLog( diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index adee880ec..2d144f23d 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -116,6 +116,7 @@ const subsystemSurface = { "apple-health-import-reconcile": { audience: "account" }, "health-connect-import": { audience: "account" }, "compaction-tombstone-purge": { audience: "system" }, + "measurement-fold-repair": { audience: "system" }, "measurement-maintenance": { audience: "system" }, "medication-intake-import": { audience: "system" }, "intake-slot-dedup": { audience: "system" }, diff --git a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts index 53bf25ef6..283e97940 100644 --- a/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts +++ b/src/lib/measurements/__tests__/consolidate-daily-mean.test.ts @@ -99,6 +99,8 @@ describe("consolidateDailyMean — drain flow (mocked Prisma)", () => { // Canonical-slot probe inside the write transaction — defaults to an // unoccupied slot (the pre-collision-fix happy path). const txFindFirst = vi.fn().mockResolvedValue(null); + // The fold leftovers of a day that already has a live daily row. + const txFindMany = vi.fn().mockResolvedValue([]); const findManyMeasurement = vi.fn( async (args: { where: { type: string } }) => rowsByType[args.where.type] ?? [], @@ -110,6 +112,7 @@ describe("consolidateDailyMean — drain flow (mocked Prisma)", () => { deleteMany, update, findFirst: txFindFirst, + findMany: txFindMany, }, }; return { @@ -289,6 +292,8 @@ describe("consolidateDailyMean — canonical-slot collision (second unique index const deleteMany = vi.fn().mockResolvedValue({ count: 0 }); const update = vi.fn().mockResolvedValue({}); const txFindFirst = vi.fn().mockResolvedValue(null); + // The fold leftovers of a day that already has a live daily row. + const txFindMany = vi.fn().mockResolvedValue([]); const findManyMeasurement = vi.fn( async (args: { where: { type: string } }) => rowsByType[args.where.type] ?? [], @@ -300,6 +305,7 @@ describe("consolidateDailyMean — canonical-slot collision (second unique index deleteMany, update, findFirst: txFindFirst, + findMany: txFindMany, }, }; return { @@ -319,6 +325,7 @@ describe("consolidateDailyMean — canonical-slot collision (second unique index deleteMany, update, txFindFirst, + txFindMany, }; } @@ -328,8 +335,21 @@ describe("consolidateDailyMean — canonical-slot collision (second unique index occupant: { id: string; externalId: string | null; deletedAt: Date | null }, ) { txFindFirst.mockImplementation( - async (args: { where: { measuredAt: Date } }) => - args.where.measuredAt.getTime() === noon.getTime() ? occupant : null, + async (args: { + where: { measuredAt?: Date; externalId?: string; deletedAt?: null }; + }) => { + // The live-daily-row lookup (no instant): the occupant answers it + // when it is the live daily row itself. + if (args.where.measuredAt === undefined) { + return occupant.externalId === args.where.externalId && + occupant.deletedAt === null + ? { id: occupant.id } + : null; + } + return args.where.measuredAt.getTime() === noon.getTime() + ? occupant + : null; + }, ); } @@ -414,8 +434,44 @@ 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", async () => { - const { mock, upsert, update, txFindFirst } = buildPrismaMock({ + 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, + }, + ]); + + await consolidateDailyMean(mock, { log: () => {} }); + + // 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("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, { @@ -426,14 +482,8 @@ describe("consolidateDailyMean — canonical-slot collision (second unique index await consolidateDailyMean(mock, { log: () => {} }); - // 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 }; - }; - expect(upsertArg.create.measuredAt.getTime()).toBe(noon.getTime()); - // Only the single slot probe ran — no free-instant search. - expect(txFindFirst).toHaveBeenCalledTimes(1); + expect(upsert).not.toHaveBeenCalled(); + expect(deleteMany).toHaveBeenCalledTimes(1); }); 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 714fd0819..f725067a1 100644 --- a/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-pulse-summary.test.ts @@ -103,7 +103,12 @@ function buildPrismaMock(opts: { .fn() .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]), }, - measurement: { findMany, findFirst: topFindFirst, upsert }, + measurement: { + findMany, + findFirst: topFindFirst, + upsert, + count: vi.fn().mockResolvedValue(0), + }, $transaction: vi.fn(async (cb: (t: unknown) => Promise) => cb(tx)), } as unknown as PrismaClient; 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 9c9c0ce2a..84a247c37 100644 --- a/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-resting-mint.test.ts @@ -116,6 +116,7 @@ function buildPrismaMock(existingResting: Array<{ source: string }>) { ), findFirst: probeFindFirst, upsert, + count: vi.fn().mockResolvedValue(0), }, $transaction: vi.fn(async (cb: (t: unknown) => Promise) => cb(tx), diff --git a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts index e660d5731..fb1586acd 100644 --- a/src/lib/measurements/__tests__/dense-intraday-retention.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-retention.test.ts @@ -24,6 +24,7 @@ import { DENSE_INTRADAY_RETENTION_TYPES, } from "../dense-intraday-retention"; import type { PerSampleRow } from "../drain-per-sample-cumulative"; +import { foldBoundary } from "../consolidation-tz"; import type { MeasurementType, PrismaClient } from "@/generated/prisma/client"; import { candidateLookup, createManyVia } from "./hourly-mint-mock"; @@ -87,7 +88,10 @@ function buildPrismaMock( .fn() .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]), }, - measurement: { findMany: findManyMeasurement }, + measurement: { + findMany: findManyMeasurement, + count: vi.fn().mockResolvedValue(0), + }, $transaction: vi.fn(async (cb: (t: unknown) => Promise) => cb(tx), ), @@ -119,16 +123,29 @@ describe("runDenseIntradayRetention — retention bound", () => { ).where; expect(where.measuredAt?.lt).toBeInstanceOf(Date); } - // The cutoff is ~DENSE_INTRADAY_RETENTION_DAYS in the past. + // The cutoff is the start of the account's local day that + // DENSE_INTRADAY_RETENTION_DAYS back falls on: only complete local days + // fold, so no day is folded in two runs. const firstCutoff = ( findManyMeasurement.mock.calls[0]?.[0] as unknown as { where: { measuredAt: { lt: Date } }; } - ).where.measuredAt.lt.getTime(); - const expected = - Date.now() - DENSE_INTRADAY_RETENTION_DAYS * 24 * 60 * 60 * 1000; - // Within a generous 60s of the expected boundary. - expect(Math.abs(firstCutoff - expected)).toBeLessThan(60_000); + ).where.measuredAt.lt; + const expected = foldBoundary( + new Date(), + DENSE_INTRADAY_RETENTION_DAYS * 24 * 60 * 60 * 1000, + "Europe/Berlin", + ); + expect(firstCutoff.getTime()).toBe(expected.getTime()); + // Local midnight in Berlin. + expect( + new Intl.DateTimeFormat("en-GB", { + timeZone: "Europe/Berlin", + hour: "2-digit", + minute: "2-digit", + hourCycle: "h23", + }).format(firstCutoff), + ).toBe("00:00"); }); it("folds everything when retentionDays = 0 (no window)", async () => { 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 23c4e2fa7..d0cddd2f6 100644 --- a/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts +++ b/src/lib/measurements/__tests__/dense-intraday-spo2-summary.test.ts @@ -93,7 +93,12 @@ function buildPrismaMock(opts: { spo2Rows: PerSampleRow[] }) { .fn() .mockResolvedValue([{ id: "user-1", timezone: "Europe/Berlin" }]), }, - measurement: { findMany, findFirst: topFindFirst, upsert }, + measurement: { + findMany, + findFirst: topFindFirst, + upsert, + count: vi.fn().mockResolvedValue(0), + }, $transaction: vi.fn(async (cb: (t: unknown) => Promise) => cb(tx)), } as unknown as PrismaClient; diff --git a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts index ba1a18552..53368ffe3 100644 --- a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts +++ b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts @@ -1,3 +1,4 @@ +import { foldBoundary } from "../consolidation-tz"; import { describe, expect, it, vi } from "vitest"; // The SUM drain now recomputes the affected DAY rollup bucket after each @@ -320,11 +321,13 @@ describe("drainPerSampleCumulative — cutoffHours", () => { expect(call.where.source).toBe("APPLE_HEALTH"); expect(call.where.measuredAt?.lt).toBeInstanceOf(Date); + // The start of the account's local day that the 36-hour grace ends in: + // only complete local days fold. const cutoff = call.where.measuredAt!.lt as Date; - const expectedMin = beforeAt - 36 * 60 * 60 * 1000; - const expectedMax = afterAt - 36 * 60 * 60 * 1000; - expect(cutoff.getTime()).toBeGreaterThanOrEqual(expectedMin); - expect(cutoff.getTime()).toBeLessThanOrEqual(expectedMax); + const expected = [beforeAt, afterAt].map((t) => + foldBoundary(new Date(t), 36 * 60 * 60 * 1000, "Europe/Berlin").getTime(), + ); + expect(expected).toContain(cutoff.getTime()); }); it("omits the cutoff filter when cutoffHours is not provided (CLI / admin one-shot)", async () => { diff --git a/src/lib/measurements/__tests__/folded-window.test.ts b/src/lib/measurements/__tests__/folded-window.test.ts index 4c9bc9b0d..d83a3d464 100644 --- a/src/lib/measurements/__tests__/folded-window.test.ts +++ b/src/lib/measurements/__tests__/folded-window.test.ts @@ -114,7 +114,9 @@ describe("coveringStatsExternalId", () => { }); }); -function client(live: Array<{ type: string; externalId: string }>) { +function client( + live: Array<{ type: string; externalId: string; measuredAt: Date }>, +) { const findMany = vi.fn(async () => live); return { findMany, @@ -142,11 +144,12 @@ describe("findFoldedWindowDuplicates", () => { expect(findMany).not.toHaveBeenCalled(); }); - it("returns the rows a live `stats:` row covers, in one live-only read", async () => { + it("returns the rows a live `stats:` row covers, in live-only reads", async () => { const { client: c, findMany } = client([ { type: "PULSE", externalId: "stats:HKQuantityTypeIdentifierHeartRate:2026-03-01T10", + measuredAt: new Date("2026-03-01T10:30:00.000Z"), }, ]); const found = await findFoldedWindowDuplicates( @@ -161,13 +164,18 @@ describe("findFoldedWindowDuplicates", () => { ], { now: NOW }, ); + // The 11:15 sample is 45 minutes from the 10:30 anchor: outside the + // half hour any zone's hour anchor can be from a sample of its hour. expect([...found]).toEqual([0]); - expect(findMany).toHaveBeenCalledTimes(1); - const where = ( - findMany.mock.calls[0] as unknown as [{ where: Record }] - )[0].where; - expect(where.deletedAt).toBeNull(); - expect(where.source).toBe("APPLE_HEALTH"); + // One read by externalId, one for the timezone-change reach of the + // sample it did not cover; both live rows of the sample's own source. + expect(findMany).toHaveBeenCalledTimes(2); + for (const call of findMany.mock.calls) { + const where = (call as unknown as [{ where: Record }])[0] + .where; + expect(where.deletedAt).toBeNull(); + expect(where.source).toBe("APPLE_HEALTH"); + } }); }); @@ -177,6 +185,7 @@ describe("findCompactionTombstones", () => { { type: "PULSE", externalId: "stats:HKQuantityTypeIdentifierHeartRate:2026-03-01T10", + measuredAt: new Date("2026-03-01T10:30:00.000Z"), }, ]; @@ -207,6 +216,7 @@ describe("findCompactionTombstones", () => { { type: "PULSE", externalId: "stats:HKQuantityTypeIdentifierHeartRate:2026-03-01", + measuredAt: new Date("2026-03-01T12:00:00.000Z"), }, ]); const found = await findCompactionTombstones(c, "u1", "UTC", [ @@ -218,6 +228,22 @@ describe("findCompactionTombstones", () => { expect(found.size).toBe(0); }); + it("leaves a deletion alone whose day the fold had not finished yet", async () => { + // Past the 90 days by an hour, but the rest of the local day was still + // inside them: an older release folded such a day in two runs, and a + // person deleting a sample of it deleted something the fold had not + // taken. + const { client: c, findMany } = client(covering); + const found = await findCompactionTombstones(c, "u1", "UTC", [ + { + ...raw("PULSE", measuredAt), + deletedAt: new Date(measuredAt.getTime() + 90 * DAY + 3_600_000), + }, + ]); + expect(found.size).toBe(0); + expect(findMany).not.toHaveBeenCalled(); + }); + it("leaves a live row alone", async () => { const { client: c } = client(covering); const found = await findCompactionTombstones(c, "u1", "UTC", [ @@ -300,6 +326,9 @@ describe("Health Connect", () => { (call) => (call as unknown as [{ where: { source: string } }])[0].where.source, ); - expect(sources.sort()).toEqual(["APPLE_HEALTH", "HEALTH_CONNECT"]); + expect([...new Set(sources)].sort()).toEqual([ + "APPLE_HEALTH", + "HEALTH_CONNECT", + ]); }); }); diff --git a/src/lib/measurements/consolidate-daily-mean.ts b/src/lib/measurements/consolidate-daily-mean.ts index 46d8f27cf..c7416aedc 100644 --- a/src/lib/measurements/consolidate-daily-mean.ts +++ b/src/lib/measurements/consolidate-daily-mean.ts @@ -22,12 +22,16 @@ * Per user × type × completed calendar day (anchored to `User.timezone`): * 1. SELECT live (`deletedAt IS NULL`) `source = 'APPLE_HEALTH'` rows * for the type whose externalId is NOT the daily-stats shape and - * whose `measuredAt` is older than the 36-hour grace cutoff (keeps - * today's in-flight watch syncs raw for the live "today" view). + * whose local day ended before the day the 36-hour grace ends in + * (`foldBoundary`; keeps today's in-flight watch syncs raw for the + * live "today" view, and never folds a day in two runs). * 2. Group into per-day buckets in the user's timezone. * 3. Compute the MEAN of the per-sample values for the day. * 4. UPSERT the canonical daily row keyed by - * `stats::` at local-noon, `value = mean`. + * `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. * 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. @@ -65,6 +69,8 @@ import { } from "./consolidation-base"; 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"; /** * Grace window — rows newer than `now() - cutoffHours` stay raw so @@ -266,6 +272,8 @@ export async function consolidateDailyMean( reducedValue, dayRows, sourceRowIds, + tz, + dateKey, }): Promise => { // Carry the canonical unit straight off the in-hand live day rows // (units are homogeneous per type). Avoids an extra per-day query @@ -340,34 +348,77 @@ 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. - await tx.measurement.upsert({ - where: { - userId_type_source_externalId: { + if (!keepStored) { + await tx.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type, + source: "APPLE_HEALTH", + externalId, + }, + }, + create: { userId, type, + value, + unit, source: "APPLE_HEALTH", + measuredAt: mintAt, externalId, }, - }, - create: { - userId, - type, - value: reducedValue, - unit, - source: "APPLE_HEALTH", - measuredAt: mintAt, - externalId, - }, - update: { - value: reducedValue, - measuredAt: mintAt, - deletedAt: null, - }, - }); + update: { + value, + measuredAt: mintAt, + deletedAt: null, + }, + }); + } // Delete the per-sample rows in the same transaction. // diff --git a/src/lib/measurements/consolidation-base.ts b/src/lib/measurements/consolidation-base.ts index 592f787b8..ad6b0b121 100644 --- a/src/lib/measurements/consolidation-base.ts +++ b/src/lib/measurements/consolidation-base.ts @@ -40,6 +40,7 @@ import { CONSOLIDATION_GRACE_CUTOFF_HOURS, canonicalDailyTimestamp, dayKeyForUserTz, + foldBoundary, type PerSampleRow, } from "./consolidation-tz"; import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; @@ -84,15 +85,17 @@ export function resolveUserTimezone(timezone: string | null): string { } /** - * Compute the cutoff instant for a grace window. Returns `null` when no - * positive `cutoffHours` is supplied (drain everything the caller points - * at), matching the one-shot / CLI default the drains carry. + * The grace window in milliseconds, or `null` when no positive `cutoffHours` + * is supplied (drain everything the caller points at), matching the one-shot + * / CLI default the drains carry. The instant a pass stops at is resolved per + * account with {@link foldBoundary}: the start of the local day the window + * ends in, so only complete local days are folded. */ -export function resolveCutoffInstant( +export function resolveCutoffMs( cutoffHours: number | undefined, -): Date | null { +): number | null { return typeof cutoffHours === "number" && cutoffHours > 0 - ? new Date(Date.now() - cutoffHours * 60 * 60 * 1000) + ? cutoffHours * 60 * 60 * 1000 : null; } @@ -432,7 +435,8 @@ export async function runConsolidation( }> { const { prismaClient, options } = params; const dryRun = options.dryRun ?? false; - const cutoffAt = resolveCutoffInstant(options.cutoffHours); + const cutoffMs = resolveCutoffMs(options.cutoffHours); + const startedAt = new Date(); const scanSelect = params.scanSelect ?? DEFAULT_SCAN_SELECT; const shouldStop = options.shouldStop ?? (() => false); let daysFailed = 0; @@ -444,6 +448,10 @@ export async function runConsolidation( walk: for (const user of users) { const tz = resolveUserTimezone(user.timezone); + // Aligned to the account's local day: a day is folded whole or not at + // all, never in two runs (see `foldBoundary`). + const cutoffAt = + cutoffMs === null ? null : foldBoundary(startedAt, cutoffMs, tz); params.onUserStart?.({ userId: user.id, tz, dryRun }); for (const type of params.types) { diff --git a/src/lib/measurements/consolidation-tz.ts b/src/lib/measurements/consolidation-tz.ts index 133cf4f16..15811d27d 100644 --- a/src/lib/measurements/consolidation-tz.ts +++ b/src/lib/measurements/consolidation-tz.ts @@ -29,6 +29,7 @@ import type { MeasurementType } from "@/generated/prisma/client"; // node:module pull) so importing it keeps this dependency-free module clean. import { userDayKey } from "@/lib/tz/format"; import { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; +import { startOfLocalDayInTz } from "@/lib/tz/local-day"; /** * Canonical grace window shared by the cumulative + mean drains. Rows @@ -39,6 +40,26 @@ import { zonedWallClockToUtc } from "@/lib/tz/wall-clock"; */ export const CONSOLIDATION_GRACE_CUTOFF_HOURS = 36; +/** + * The instant a fold stops at: the first instant of the local day that + * `now - cutoffMs` falls on, in the account's zone. Rows before it belong + * to local days that are over in full, so a fold never cuts a day (or any + * hour of it) in two. + * + * Cutting at `now - cutoffMs` itself used to fold the first part of a day on + * one run and the rest on the next, and the second run overwrote the stored + * mean with the mean of the later part alone. The day is the grain for the + * hourly fold too: its pre-fold DAY rollup and the derived resting figure are + * computed from the whole day's raw rows, which only a complete day has. + * + * The `folded_window` ingest guard and the compaction-tombstone classifier + * read the same boundary (`folded-window.ts`), so "folded" means the same + * thing on all three paths. + */ +export function foldBoundary(now: Date, cutoffMs: number, tz: string): Date { + return startOfLocalDayInTz(new Date(now.getTime() - cutoffMs), tz); +} + /** * Per-sample row shape the drains scan and bucket. Exposed for unit * testing the bucketing semantics without booting Prisma. diff --git a/src/lib/measurements/dense-intraday-retention.ts b/src/lib/measurements/dense-intraday-retention.ts index 1cbf1ab9d..0097b4097 100644 --- a/src/lib/measurements/dense-intraday-retention.ts +++ b/src/lib/measurements/dense-intraday-retention.ts @@ -27,7 +27,9 @@ * mean-consolidation drain skeleton (same `runConsolidation` base, * same soft-delete + rollup handling) but with the grace cutoff * widened from 36 hours to the retention window, and the fold grain - * set to the LOCAL HOUR: per-sample rows OLDER than the window fold + * set to the LOCAL HOUR: per-sample rows of the local days that ended + * before the window's boundary (`foldBoundary`; a day folds whole, in + * one run) fold * to one `stats::T` hourly-mean row per user × * type × local hour and the raw rows are deleted (tombstoned before * v1.42); rows INSIDE the @@ -71,6 +73,12 @@ import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollup import { percentile } from "@/lib/insights/strain-score"; import type { PerSampleRow } from "./drain-per-sample-cumulative"; import { DEFAULT_TIMEZONE } from "@/lib/tz/format"; +import { localDayWindow } from "@/lib/tz/local-day"; +import { + loadFoldLeftovers, + meanOf, + type FoldConstituent, +} from "./fold-constituents"; /** * iOS#34 / #69 — heart-rate (PULSE) is the densest spot signal after @@ -629,7 +637,53 @@ export async function runDenseIntradayRetention( // live, and the post-fold DAY recompute is skipped entirely. The // WEEK/MONTH/YEAR recomputes this call enqueues run post-commit // against whatever is live then, as before. - await recomputeBucketsForMeasurement(userId, type, canonicalTimestamp); + // + // A day that already carries hourly rows of this source is one an + // earlier run folded in part (before the boundary was aligned to the + // local day) or that these rows reached afterwards. Its live rows are + // then a mix of hourly means and raw samples, and a DAY recompute + // from them would overwrite the figure the first fold took from the + // whole raw day, so it is left alone. + const hourlyPrefix = `stats:${hkIdentifier}:${dateKey}T`; + const dayAlreadyFolded = + (await pc.measurement.count({ + where: { + userId, + type, + source, + externalId: { startsWith: hourlyPrefix }, + 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 { + 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); + } // Fold the day into per-LOCAL-HOUR mean rows, anchored at local // HH:30. Sorted for a deterministic write order across runs. @@ -640,23 +694,63 @@ export async function runDenseIntradayRetention( const foldOnce = async (): Promise<{ removed: number; retiredDaily: boolean; + slotsWritten: number; }> => { let removed = 0; let retiredDaily = false; + let slotsWritten = 0; 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. + 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 slots: HourlySlot[] = []; + 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; + value = meanOf([ + ...hourRows.map((row) => row.value), + ...earlier.map((row) => row.value), + ]); + } + slots.push({ + externalId: slotId, + anchor: canonicalHourlyTimestamp(dateKey, hour, tz), + value, + }); + } const canonicalRowIds = await adoptOrMintHourlyRows(tx, { userId, type, source, unit, - slots: byHour.map(([hour, hourRows]) => ({ - externalId: hourlyStatsExternalId(hkIdentifier, dateKey, hour), - anchor: canonicalHourlyTimestamp(dateKey, hour, tz), - value: meanBucketValue(hourRows), - })), + slots, }); + slotsWritten = slots.length; // Retire a live pre-hourly DAILY `stats:` row for this day (the // late-sync path: raw samples arriving for a day folded to the @@ -702,10 +796,14 @@ export async function runDenseIntradayRetention( }); removed = del.count; }); - return { removed, retiredDaily }; + return { removed, retiredDaily, slotsWritten }; }; - let foldResult: { removed: number; retiredDaily: boolean }; + let foldResult: { + removed: number; + retiredDaily: boolean; + slotsWritten: number; + }; try { foldResult = await foldOnce(); } catch (err) { @@ -717,7 +815,7 @@ export async function runDenseIntradayRetention( } // Counters mutate OUTSIDE the retryable transaction so a P2002 retry // cannot double-count. - summary.totals.hourlyRowsUpserted += byHour.length; + summary.totals.hourlyRowsUpserted += foldResult.slotsWritten; if (foldResult.retiredDaily) summary.totals.dailyRowsRetired += 1; // iOS#34 — preserve the resting signal for PROXY users. For a user @@ -731,8 +829,25 @@ export async function runDenseIntradayRetention( // the resolver prefers. Skipped for users WITH native resting rows // (the resolver ignores the proxy entirely for them, so a derived row // would only double-count). - if (isPulse) { - const restingValue = deriveDailyRestingFromPulse(dayRows); + // + // 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)) { + const restingValue = deriveDailyRestingFromPulse( + dayAlreadyFolded + ? [ + ...dayRows, + ...leftovers.map((row) => ({ + id: row.id, + type, + value: row.value, + measuredAt: row.measuredAt, + externalId: null, + })), + ] + : dayRows, + ); if (restingValue !== null && !(await userHasNativeResting(userId))) { const restingExternalId = dailyStatsExternalId( RESTING_HK_IDENTIFIER, diff --git a/src/lib/measurements/fold-constituents.ts b/src/lib/measurements/fold-constituents.ts new file mode 100644 index 000000000..97493b3f9 --- /dev/null +++ b/src/lib/measurements/fold-constituents.ts @@ -0,0 +1,128 @@ +/** + * The raw samples a folded hour or day was computed from, as far as they are + * still in the table. + * + * A fold writes one `stats:` mean per local hour (dense types) or local day + * (mean types). Before v1.42 it soft-deleted the samples it folded, and until + * the fold boundary was aligned to the local day it could fold a day in two + * runs, the second overwriting the stored mean with the mean of the later + * part only. Recomputing such a mean correctly needs every sample of the + * window: the ones still live and the ones a fold soft-deleted. + * + * Which tombstones a fold left, as opposed to a person's deletions: + * - every path that deletes on a person's behalf (`DELETE + * /api/measurements/:id`, the bulk delete, the by-external-ids delete) + * raises `syncVersion`; the folds never did, and a raw HealthKit or Health + * Connect sample is never updated in place, so a fold's tombstone carries + * `syncVersion = 1`; + * - a fold only took samples already past its threshold, so the sample was + * older than the threshold when it was deleted. + * A collision retirement (`retired:` id) and a `stats:` row are never samples. + * + * This is deliberately wider than the class-A rule in `folded-window.ts`: a + * 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. + */ +import type { + MeasurementSource, + MeasurementType, + Prisma, +} from "@/generated/prisma/client"; + +export interface FoldConstituent { + id: string; + value: number; + measuredAt: Date; +} + +/** The fields a row needs for {@link isFoldLeftover}. */ +export interface FoldLeftoverCandidate { + externalId: string | null; + measuredAt: Date; + deletedAt: Date | null; + syncVersion: number; +} + +/** Whether a soft-deleted row is a sample a fold removed. Pure. */ +export function isFoldLeftover( + row: FoldLeftoverCandidate, + foldedAfterMs: number, +): boolean { + if (row.deletedAt === null) return false; + if (row.syncVersion !== 1) return false; + if (row.externalId === null) return false; + if (row.externalId.startsWith("stats:")) return false; + if (row.externalId.startsWith("retired:")) return false; + return row.measuredAt.getTime() < row.deletedAt.getTime() - foldedAfterMs; +} + +/** Prisma filter for the rows {@link isFoldLeftover} then decides on. */ +export function foldLeftoverWhere(input: { + userId: string; + type: MeasurementType; + source: MeasurementSource; +}): Prisma.MeasurementWhereInput { + return { + userId: input.userId, + type: input.type, + source: input.source, + deletedAt: { not: null }, + syncVersion: 1, + externalId: { not: null }, + NOT: [ + { externalId: { startsWith: "stats:" } }, + { externalId: { startsWith: "retired:" } }, + ], + }; +} + +/** + * The fold leftovers of one `[from, to)` span of one account, type and source. + * The span is one local day, so the result is at most a day of samples. + */ +export async function loadFoldLeftovers( + client: Pick, + input: { + userId: string; + type: MeasurementType; + source: MeasurementSource; + from: Date; + to: Date; + foldedAfterMs: number; + }, +): Promise { + const rows = await client.measurement.findMany({ + where: { + ...foldLeftoverWhere(input), + measuredAt: { gte: input.from, lt: input.to }, + }, + select: { + id: true, + value: true, + measuredAt: true, + externalId: true, + deletedAt: true, + syncVersion: true, + }, + }); + return rows + .filter((row) => isFoldLeftover(row, input.foldedAfterMs)) + .map((row) => ({ + id: row.id, + value: row.value, + measuredAt: row.measuredAt, + })); +} + +/** Arithmetic mean of a non-empty list of values. */ +export function meanOf(values: readonly number[]): number { + let acc = 0; + for (const v of values) acc += v; + return acc / values.length; +} + +/** Whether a recomputed mean differs from the stored one beyond float noise. */ +export function meanDiffers(stored: number, recomputed: number): boolean { + return Math.abs(stored - recomputed) > 1e-9 * Math.max(1, Math.abs(stored)); +} diff --git a/src/lib/measurements/folded-window.ts b/src/lib/measurements/folded-window.ts index b6fbdb227..9fa41a6e1 100644 --- a/src/lib/measurements/folded-window.ts +++ b/src/lib/measurements/folded-window.ts @@ -18,15 +18,34 @@ * {@link findCompactionTombstones}, so the three can never disagree about * which rows are compaction leftovers and which are a person's deletions. * + * "Folded" means the fold has been able to reach the sample's whole window. + * The folds stop at the start of the local day the threshold ends in + * (`foldBoundary` in `consolidation-tz.ts`), so a window counts as folded only + * when its local day lies entirely before that boundary, read in the account's + * zone at the instant in question. A sample of a day the fold has not finished + * is never refused, even if a `stats:` row for the day exists (a day an older + * release folded in part): refusing it would lose a genuine late sample. + * * Which `stats:` row covers a sample is decided by its externalId, built from * the account's current timezone exactly as the fold builds it * (`stats::T` for a dense hour, `stats::` * for a mean day). A dense day still folded at the pre-v1.28.31 daily grain is * deliberately NOT covered: the one-shot hourly rebuild reads those days' * tombstones, so they are neither refused at ingest by this rule nor purged. - * An account whose timezone changed after a fold may see an old day keyed in - * the other zone; such a row is then simply not recognised, which is the - * pre-v1.42 behaviour, never a wrong refusal. + * + * An account whose timezone changed after a fold keys its old windows in the + * other zone, so the externalId alone misses them and an old sample would + * re-enter beside its own mean. The ingest guard therefore also counts a + * window as covered when a live `stats:` row of the same source, type and + * grain sits within reach of the sample: the anchor of the window containing + * an instant is at most 13 hours from it for a day (local noon, a 25-hour DST + * day included; the reach allows 14) and 30 minutes for an hour (HH:30), in + * any zone. Near local midnight the reach of a day also takes in the + * neighbouring day's row, so a sample there whose own day has no row but + * whose neighbour has one is refused too; the price of a check that cannot + * know the zone the fold ran in. The classifier + * of tombstones does not use that reach; a tombstone it misses stays class B, + * which keeps a row longer, never loses one. * * Health Connect folds its raw heart-rate history the same way under its own * source (the importer writes samples of the last 90 days raw, and the @@ -64,22 +83,61 @@ export const FOLDED_WINDOW_REASON: FoldedWindowReason = "folded_window"; const HOUR_MS = 3_600_000; -/** Which sources and types fold into which window, and from what age. */ +/** + * Which sources and types fold into which window, and from what age. + * `reachMs` is how far from a sample the anchor of its window can sit in any + * zone (see the module comment), for the timezone-change check. + */ export const FOLDED_WINDOW_RULES = [ { window: "hour", sources: DENSE_INTRADAY_FOLD_SOURCES, types: DENSE_INTRADAY_RETENTION_TYPES, foldedAfterMs: DENSE_INTRADAY_RETENTION_DAYS * 24 * HOUR_MS, + reachMs: HOUR_MS / 2, }, { window: "day", sources: ["APPLE_HEALTH"] as const, types: HIGH_FREQUENCY_MEAN_TYPES, foldedAfterMs: CONSOLIDATION_GRACE_CUTOFF_HOURS * HOUR_MS, + reachMs: 14 * HOUR_MS, }, ] as const; +type FoldedWindowRule = (typeof FOLDED_WINDOW_RULES)[number]; + +function ruleFor( + row: Pick, +): FoldedWindowRule | null { + return ( + FOLDED_WINDOW_RULES.find( + (rule) => + (rule.sources as readonly MeasurementSource[]).includes(row.source) && + rule.types.has(row.type), + ) ?? null + ); +} + +/** + * Whether the fold had reached a row's whole window at instant `at`: its + * local day (in `tz`) lies entirely before the fold boundary of that instant. + * Day keys compare as strings. Pure. + */ +export function isWindowFoldedAt( + row: Pick, + at: Date, + tz: string, +): boolean { + const rule = ruleFor(row); + if (!rule) return false; + const firstOpenDay = dayKeyForUserTz( + new Date(at.getTime() - rule.foldedAfterMs), + tz, + ); + return dayKeyForUserTz(row.measuredAt, tz) < firstOpenDay; +} + /** Every type either rule folds. */ export const FOLDED_TYPES: readonly MeasurementType[] = [ ...DENSE_INTRADAY_RETENTION_TYPES, @@ -95,9 +153,12 @@ export interface FoldedWindowCandidateRow { } /** - * Whether a row is old enough, and of the right kind, for its hour or day to - * have been folded already. Pure: whether a live `stats:` row actually covers - * it is the lookup's question. `now` is the instant the age is measured at: + * Whether a row is old enough, and of the right kind, for its hour or day + * possibly to have been folded already. A cheap prefilter that needs no + * timezone: every row {@link isWindowFoldedAt} accepts passes it, because a + * day before the boundary ended before `now - threshold`. Pure: the aligned + * check and whether a live `stats:` row actually covers the row are the + * lookup's questions. `now` is the instant the age is measured at: * the request time for an upload, the deletion time for a tombstone. */ export function isFoldedWindowCandidate( @@ -109,12 +170,10 @@ export function isFoldedWindowCandidate( // A collision retirement (`reconcile-external-measurement.ts`) parks the // losing row at the epoch under a `retired:` id. It is not a folded sample. if (row.externalId.startsWith("retired:")) return false; - const age = now.getTime() - row.measuredAt.getTime(); - return FOLDED_WINDOW_RULES.some( - (rule) => - (rule.sources as readonly MeasurementSource[]).includes(row.source) && - rule.types.has(row.type) && - age > rule.foldedAfterMs, + const rule = ruleFor(row); + return ( + rule !== null && + now.getTime() - row.measuredAt.getTime() > rule.foldedAfterMs ); } @@ -160,10 +219,40 @@ export async function loadFoldTimezone( return resolveUserTimezone(user?.timezone ?? null); } +/** A live `stats:` row the coverage check reads. */ +type LiveStatsRow = { + type: MeasurementType; + externalId: string | null; + measuredAt: Date; +}; + +const HOURLY_KEY_RE = /:\d{4}-\d{2}-\d{2}T\d{2}$/; +const DAILY_KEY_RE = /:\d{4}-\d{2}-\d{2}$/; + +/** Whether a `stats:` id is the grain the rule folds into, for that type. */ +function isRuleGrain( + rule: FoldedWindowRule, + type: MeasurementType, + externalId: string | null, +): boolean { + const hk = hkIdentifierForType(type); + if (!hk || externalId === null) return false; + if (!externalId.startsWith(`stats:${hk}:`)) return false; + return rule.window === "hour" + ? HOURLY_KEY_RE.test(externalId) + : DAILY_KEY_RE.test(externalId); +} + +/** Candidates whose time span per query stays within this many days. */ +const REACH_QUERY_SPAN_MS = 31 * 24 * HOUR_MS; + /** * Indexes (into `rows`) whose covering `stats:` row is live under the row's * own source. One query per source present, each over the unique - * `(user_id, type, source, external_id)` index for the whole set. + * `(user_id, type, source, external_id)` index for the whole set. With + * `reach`, a row is also covered by a live `stats:` row of the same source, + * type and grain anchored within the rule's reach of it (the timezone-change + * check), read in spans of at most a month so the read stays bounded. */ async function coveredIndexes( client: Pick, @@ -175,6 +264,7 @@ async function coveredIndexes( source: MeasurementSource; measuredAt: Date; }>, + options: { reach: boolean }, ): Promise> { const covered = new Set(); if (rows.length === 0) return covered; @@ -216,13 +306,72 @@ async function coveredIndexes( if (liveKeys.has(key)) covered.add(index); } } + if (!options.reach) return covered; + + // The timezone-change check, for the rows the externalId did not cover. + const open = rows + .filter((row) => !covered.has(row.index) && ruleFor(row) !== null) + .sort((a, b) => a.measuredAt.getTime() - b.measuredAt.getTime()); + const bySourceOpen = new Map(); + for (const row of open) { + const list = bySourceOpen.get(row.source) ?? []; + list.push(row); + bySourceOpen.set(row.source, list); + } + const maxReach = Math.max(...FOLDED_WINDOW_RULES.map((r) => r.reachMs)); + for (const [source, list] of bySourceOpen) { + let start = 0; + while (start < list.length) { + const spanStart = list[start].measuredAt.getTime(); + let end = start; + while ( + end + 1 < list.length && + list[end + 1].measuredAt.getTime() - spanStart <= REACH_QUERY_SPAN_MS + ) { + end += 1; + } + const span = list.slice(start, end + 1); + start = end + 1; + const types = [...new Set(span.map((row) => row.type))]; + const nearby: LiveStatsRow[] = await client.measurement.findMany({ + where: { + userId, + type: { in: types }, + source, + externalId: { startsWith: "stats:" }, + deletedAt: null, + measuredAt: { + gte: new Date(spanStart - maxReach), + lte: new Date( + span[span.length - 1].measuredAt.getTime() + maxReach, + ), + }, + }, + select: { type: true, externalId: true, measuredAt: true }, + orderBy: { measuredAt: "asc" }, + }); + for (const row of span) { + const rule = ruleFor(row); + if (!rule) continue; + const t = row.measuredAt.getTime(); + const hit = nearby.some( + (stats) => + stats.type === row.type && + isRuleGrain(rule, row.type, stats.externalId) && + Math.abs(stats.measuredAt.getTime() - t) <= rule.reachMs, + ); + if (hit) covered.add(row.index); + } + } + } return covered; } /** - * Indexes (into `rows`) of the incoming rows whose hour or day a live `stats:` - * row already covers. Rows too young to have been folded are never looked up, - * so a batch of current samples costs no query at all. + * Indexes (into `rows`) of the incoming rows whose hour or day the fold has + * finished and a live `stats:` row covers (by its externalId, or within reach + * after a timezone change). Rows too young to have been folded are never + * looked up, so a batch of current samples costs no query at all. */ export async function findFoldedWindowDuplicates( client: Pick & @@ -250,7 +399,14 @@ export async function findFoldedWindowDuplicates( if (!client.user) throw new Error("folded_window lookup needs a timezone"); tz = await loadFoldTimezone({ user: client.user }, userId); } - return coveredIndexes(client, userId, tz, candidates); + const zone = tz; + return coveredIndexes( + client, + userId, + zone, + candidates.filter((row) => isWindowFoldedAt(row, now, zone)), + { reach: true }, + ); } /** A tombstone as the classifier reads it. */ @@ -259,11 +415,32 @@ export interface TombstoneCandidateRow extends FoldedWindowCandidateRow { } /** - * Indexes (into `rows`) of the compaction tombstones ("class A"): soft-deleted - * raw samples of a folded source and type that were already past the fold - * threshold when they were deleted, and whose hour or day a live `stats:` row - * covers. Everything else, in particular a person's own deletions of recent - * samples, is class B and keeps the 75-day tombstone retention. + * The covering `stats:` id of a tombstone that has the shape of a compaction + * tombstone, or null: a soft-deleted raw sample of a folded source and type + * whose whole window the fold had reached when it was deleted. Whether that + * `stats:` row is live is the caller's question. Pure; the restore asks it of + * a backup file's rows. + * + * The window must have been complete at deletion, not merely the sample old + * enough: an older release folded a day in two runs, and a person who + * deleted a sample of such a half-folded day deleted something the fold had + * not taken. That deletion is class B. + */ +export function compactionTombstoneCoverId( + row: TombstoneCandidateRow, + tz: string, +): string | null { + if (row.deletedAt === null) return null; + if (!isFoldedWindowCandidate(row, row.deletedAt)) return null; + if (!isWindowFoldedAt(row, row.deletedAt, tz)) return null; + return coveringStatsExternalId(row, tz); +} + +/** + * Indexes (into `rows`) of the compaction tombstones ("class A"): rows that + * {@link compactionTombstoneCoverId} accepts and whose covering `stats:` row + * is live under the row's own source. Everything else, in particular a + * person's own deletions, is class B and keeps the 75-day tombstone retention. */ export async function findCompactionTombstones( client: Pick, @@ -272,7 +449,7 @@ export async function findCompactionTombstones( rows: readonly TombstoneCandidateRow[], ): Promise> { const candidates = rows.flatMap((row, index) => - row.deletedAt !== null && isFoldedWindowCandidate(row, row.deletedAt) + compactionTombstoneCoverId(row, tz) !== null ? [ { index, @@ -283,5 +460,5 @@ export async function findCompactionTombstones( ] : [], ); - return coveredIndexes(client, userId, tz, candidates); + return coveredIndexes(client, userId, tz, candidates, { reach: false }); } diff --git a/tests/integration/compaction-tombstones.test.ts b/tests/integration/compaction-tombstones.test.ts index 6fd879a8f..a4afa1c42 100644 --- a/tests/integration/compaction-tombstones.test.ts +++ b/tests/integration/compaction-tombstones.test.ts @@ -88,8 +88,14 @@ async function account(id: string, signIn = false) { * One account's mix: the live hourly row, five class-A tombstones under it, * and class-B rows the purge must leave alone. */ -async function seedMix(userId: string) { +async function seedMix(userId: string, repaired = true) { const prisma = getPrismaClient(); + // The purge waits for the fold repair; these accounts have been through it. + if (repaired) { + await prisma.measurementFoldRepair.create({ + data: { userId, completedAt: new Date(NOW - HOUR) }, + }); + } const compactedAt = new Date(NOW - DAY); await prisma.measurement.create({ data: { @@ -141,7 +147,9 @@ async function seedMix(userId: string) { source: "APPLE_HEALTH", measuredAt: meanAt, externalId: `uuid-${userId}-rr`, - deletedAt: new Date(meanAt.getTime() + 2 * DAY), + // Late enough that the whole local day was past the grace: a fold + // only takes complete days. + deletedAt: new Date(meanAt.getTime() + 3 * DAY), }, }); await prisma.measurement.createMany({ @@ -326,6 +334,22 @@ describe("compaction-tombstone purge", () => { }, 60_000); }); +describe("compaction-tombstone purge before the fold repair", () => { + it("leaves an account the repair has not finished alone", async () => { + const prisma = getPrismaClient(); + await account("purge-unrepaired"); + await seedMix("purge-unrepaired", false); + const { purgeCompactionTombstones } = + await import("@/lib/jobs/compaction-tombstone-purge"); + const outcome = await purgeCompactionTombstones(prisma, { pauseMs: 0 }); + expect(outcome.deleted).toBe(0); + expect(outcome.awaitingRepairAccounts).toBe(1); + for (const id of CLASS_A("purge-unrepaired")) { + expect(await ids("purge-unrepaired")).toContain(id); + } + }); +}); + describe("/api/sync/changes", () => { it("leaves compaction tombstones out and still reports a person's deletion", async () => { await account("feed", true); @@ -405,6 +429,44 @@ describe("restore", () => { expect(restored).toContain("restore-a-hourly"); expect(restored).toContain("restore-a-live"); }, 120_000); + + it("writes class-A tombstones back from a file older than the fold repair, and runs the repair again", async () => { + const prisma = getPrismaClient(); + await account("restore-old", true); + await seedMix("restore-old", false); + const { storeBackupBlob } = await import("@/lib/export/store-backup-blob"); + const { streamFullBackupJson } = + await import("@/lib/export/full-backup-stream"); + const { id: backupId } = await storeBackupBlob( + prisma, + { userId: "restore-old", type: "WEEKLY_AUTO" }, + (write) => + streamFullBackupJson(prisma, "restore-old", write, { + purpose: "disaster-recovery", + }), + ); + // The repair finished after the file was exported. + await prisma.measurementFoldRepair.create({ + data: { userId: "restore-old", completedAt: new Date(Date.now() + 1) }, + }); + const { POST } = await import("./restore-job-driver"); + const res = await POST( + new Request(`http://localhost/api/admin/backups/${backupId}/restore`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ confirm: "RESTORE" }), + }) as never, + { params: Promise.resolve({ id: backupId }) }, + ); + expect(res.status).toBe(200); + const restored = await ids("restore-old"); + for (const id of CLASS_A("restore-old")) expect(restored).toContain(id); + expect( + await prisma.measurementFoldRepair.findUnique({ + where: { userId: "restore-old" }, + }), + ).toBeNull(); + }, 120_000); }); type BatchBody = { diff --git a/tests/integration/dense-intraday-retention.test.ts b/tests/integration/dense-intraday-retention.test.ts index 06a7d6fba..c1959e9f6 100644 --- a/tests/integration/dense-intraday-retention.test.ts +++ b/tests/integration/dense-intraday-retention.test.ts @@ -209,6 +209,20 @@ describe("runDenseIntradayRetention (real Postgres)", () => { }, }); + // The sample the prior fold took for that hour, soft-deleted by it. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "HEART_RATE_VARIABILITY", + value: 20, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-01T06:05:00.000Z"), + externalId: "hk-hrv-early", + deletedAt: new Date("2026-05-03T00:00:00.000Z"), + }, + }); + // Must not throw P2002 — the fold adopts the existing hourly row. const summary = await runDenseIntradayRetention(prisma, { userId: TEST_USER_ID, @@ -226,8 +240,9 @@ describe("runDenseIntradayRetention (real Postgres)", () => { }); expect(live).toHaveLength(1); expect(live[0].externalId).toBe(`stats:${HRV_HK}:${DAY_KEY}T08`); - // Refreshed to the late sample's hourly mean, overwriting the stale 999. - expect(live[0].value).toBeCloseTo(40, 6); + // Refreshed to the mean over every sample of the hour, the prior fold's + // tombstone included, never to the late sample alone. + expect(live[0].value).toBeCloseTo(30, 6); expect(live[0].measuredAt.toISOString()).toBe("2026-05-01T06:30:00.000Z"); }); diff --git a/tests/integration/fold-complete-windows.test.ts b/tests/integration/fold-complete-windows.test.ts new file mode 100644 index 000000000..d448e1556 --- /dev/null +++ b/tests/integration/fold-complete-windows.test.ts @@ -0,0 +1,729 @@ +/** + * v1.42 — the folds take complete local days only, a re-fold recomputes from + * every sample of the window, and the one-time repair puts right the means an + * older release folded from part of their day. Against Postgres. + * + * - Mean consolidation and the dense hourly fold, run on two nights over a + * day the old `now - threshold` cut split, store the mean of the whole + * day / hour, for Apple Health and Health Connect alike. + * - A re-fold of a window that already has a `stats:` row takes the mean + * over the live rows AND the fold's own tombstones. + * - The `folded_window` guard refuses a sample only when its local day lies + * before the aligned boundary and a live `stats:` row covers it, the + * latter also after a timezone change. + * - A person's deletion inside a half-folded day is class B. + * - The repair corrects a partial mean from the tombstones, leaves a correct + * one untouched, resumes, and is idempotent; the purge waits for it. + * - The restore keys the live `stats:` ids by source. + */ +import { + afterAll, + afterEach, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; + +import { cookieJar, headerJar } from "./mock-next-headers"; +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("next/headers", async () => { + const { cookieJar, headerJar } = await import("./mock-next-headers"); + return { + headers: vi.fn(async () => ({ + get: (name: string) => headerJar.get(name.toLowerCase()) ?? null, + })), + cookies: vi.fn(async () => ({ + get: (name: string) => { + const value = cookieJar.get(name); + return value ? { name, value } : undefined; + }, + set: (name: string, value: string) => { + cookieJar.set(name, value); + }, + delete: (name: string) => { + cookieJar.delete(name); + }, + })), + }; +}); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +const WS_HK = "HKQuantityTypeIdentifierWalkingSpeed"; +const HRV_HK = "HKQuantityTypeIdentifierHeartRateVariabilitySDNN"; +const HR_HK = "HKQuantityTypeIdentifierHeartRate"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + cookieJar.clear(); + headerJar.clear(); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +afterAll(async () => { + await truncateAllTables(getPrismaClient()); +}); + +async function account(id: string, timezone = "UTC", signIn = false) { + const prisma = getPrismaClient(); + await prisma.user.create({ + data: { + id, + username: id, + email: `${id}@example.test`, + role: "ADMIN", + timezone, + }, + }); + if (signIn) { + const session = await prisma.session.create({ + data: { userId: id, expiresAt: new Date(Date.now() + 3_600_000) }, + }); + cookieJar.set("healthlog_session", session.id); + } +} + +function at(iso: string) { + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date(iso)); +} + +async function live(userId: string, type: string, externalId: string) { + return getPrismaClient().measurement.findFirst({ + where: { userId, type: type as never, externalId, deletedAt: null }, + select: { id: true, value: true, updatedAt: true }, + }); +} + +describe("folds take complete local days", () => { + it("mean consolidation over two nights stores the whole day's mean", async () => { + const prisma = getPrismaClient(); + await account("mean-two-nights"); + await prisma.measurement.createMany({ + data: ( + [ + ["08:00", 1], + ["10:00", 1], + ["20:00", 2], + ["22:00", 2], + ] as const + ).map(([hm, value], i) => ({ + userId: "mean-two-nights", + type: "WALKING_SPEED" as const, + value, + unit: "m/s", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(`2026-05-16T${hm}:00.000Z`), + externalId: `uuid-ws-${i}`, + })), + }); + const { consolidateDailyMean } = + await import("@/lib/measurements/consolidate-daily-mean"); + + at("2026-05-18T03:00:00.000Z"); + await consolidateDailyMean(prisma, { + userId: "mean-two-nights", + cutoffHours: 36, + log: () => {}, + }); + at("2026-05-19T03:00:00.000Z"); + await consolidateDailyMean(prisma, { + userId: "mean-two-nights", + cutoffHours: 36, + log: () => {}, + }); + + const row = await live( + "mean-two-nights", + "WALKING_SPEED", + `stats:${WS_HK}:2026-05-16`, + ); + expect(row?.value).toBeCloseTo(1.5, 9); + }); + + it("the dense fold over two nights stores the whole hour's mean (Apple Health and Health Connect)", async () => { + const prisma = getPrismaClient(); + await account("dense-two-nights"); + for (const source of ["APPLE_HEALTH", "HEALTH_CONNECT"] as const) { + await prisma.measurement.createMany({ + data: ( + [ + ["10:05", 40], + ["10:10", 40], + ["10:50", 80], + ["10:55", 80], + ] as const + ).map(([hm, value], i) => ({ + userId: "dense-two-nights", + type: "HEART_RATE_VARIABILITY" as const, + value, + unit: "ms", + source, + measuredAt: new Date(`2026-01-10T${hm}:00.000Z`), + externalId: `${source}-hrv-${i}`, + })), + }); + } + const { runDenseIntradayRetention } = + await import("@/lib/measurements/dense-intraday-retention"); + + at("2026-04-10T10:30:00.000Z"); + await runDenseIntradayRetention(prisma, { + userId: "dense-two-nights", + log: () => {}, + }); + at("2026-04-11T10:30:00.000Z"); + await runDenseIntradayRetention(prisma, { + userId: "dense-two-nights", + log: () => {}, + }); + + const rows = await prisma.measurement.findMany({ + where: { + userId: "dense-two-nights", + type: "HEART_RATE_VARIABILITY", + deletedAt: null, + }, + select: { source: true, externalId: true, value: true }, + orderBy: { source: "asc" }, + }); + expect(rows).toEqual([ + { + source: "APPLE_HEALTH", + externalId: `stats:${HRV_HK}:2026-01-10T10`, + value: 60, + }, + { + source: "HEALTH_CONNECT", + externalId: `stats:${HRV_HK}:2026-01-10T10`, + value: 60, + }, + ]); + }); + + it("a re-fold of a day an older release folded in part takes the mean over live rows and its tombstones", async () => { + const prisma = getPrismaClient(); + await account("refold"); + // What the old cut left behind: the morning folded into a daily row and + // soft-deleted, the evening still raw. + await prisma.measurement.create({ + data: { + userId: "refold", + type: "WALKING_SPEED", + value: 1, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + externalId: `stats:${WS_HK}:2026-05-16`, + }, + }); + await prisma.measurement.createMany({ + data: [ + ["08:00", 1, true], + ["10:00", 1, true], + ["20:00", 2, false], + ["22:00", 2, false], + ].map(([hm, value, folded], i) => ({ + userId: "refold", + type: "WALKING_SPEED" as const, + value: value as number, + unit: "m/s", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(`2026-05-16T${hm}:00.000Z`), + externalId: `uuid-refold-${i}`, + deletedAt: folded ? new Date("2026-05-18T03:00:00.000Z") : null, + })), + }); + const { consolidateDailyMean } = + await import("@/lib/measurements/consolidate-daily-mean"); + at("2026-05-19T03:00:00.000Z"); + await consolidateDailyMean(prisma, { + userId: "refold", + cutoffHours: 36, + log: () => {}, + }); + const row = await live( + "refold", + "WALKING_SPEED", + `stats:${WS_HK}:2026-05-16`, + ); + expect(row?.value).toBeCloseTo(1.5, 9); + expect( + await prisma.measurement.count({ + where: { userId: "refold", deletedAt: null }, + }), + ).toBe(1); + }); +}); + +describe("folded_window", () => { + it("accepts a late sample of a day the fold has not finished, refuses one of a finished day", async () => { + const prisma = getPrismaClient(); + await account("guard"); + for (const day of ["2026-05-15", "2026-05-16"]) { + await prisma.measurement.create({ + data: { + userId: "guard", + type: "WALKING_SPEED", + value: 1, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date(`${day}T12:00:00.000Z`), + externalId: `stats:${WS_HK}:${day}`, + }, + }); + } + const { findFoldedWindowDuplicates } = + await import("@/lib/measurements/folded-window"); + const found = await findFoldedWindowDuplicates( + prisma, + "guard", + [ + // 37 hours old: its day is not over before the boundary. + { + type: "WALKING_SPEED", + source: "APPLE_HEALTH", + externalId: "uuid-late", + measuredAt: new Date("2026-05-16T21:00:00.000Z"), + }, + { + type: "WALKING_SPEED", + source: "APPLE_HEALTH", + externalId: "uuid-old", + measuredAt: new Date("2026-05-15T21:00:00.000Z"), + }, + ], + { now: new Date("2026-05-18T10:00:00.000Z") }, + ); + expect([...found]).toEqual([1]); + }); + + it("refuses an old sample after a timezone change (day and hour grain)", async () => { + const prisma = getPrismaClient(); + // Folded while the account was in Los Angeles / Kolkata; now in Tokyo / + // UTC. + await account("tz-day", "Asia/Tokyo"); + await prisma.measurement.create({ + data: { + userId: "tz-day", + type: "WALKING_SPEED", + value: 1, + unit: "m/s", + source: "APPLE_HEALTH", + // LA local noon of 2026-03-10. + measuredAt: new Date("2026-03-10T19:00:00.000Z"), + externalId: `stats:${WS_HK}:2026-03-10`, + }, + }); + await account("tz-hour", "UTC"); + await prisma.measurement.create({ + data: { + userId: "tz-hour", + type: "PULSE", + value: 60, + unit: "bpm", + source: "APPLE_HEALTH", + // Kolkata 10:30 local of 2026-01-10 (UTC+5:30). + measuredAt: new Date("2026-01-10T05:00:00.000Z"), + externalId: `stats:${HR_HK}:2026-01-10T10`, + }, + }); + const { findFoldedWindowDuplicates } = + await import("@/lib/measurements/folded-window"); + const day = await findFoldedWindowDuplicates( + prisma, + "tz-day", + [ + { + type: "WALKING_SPEED", + source: "APPLE_HEALTH", + externalId: "uuid-la-evening", + // 23:00 in LA on 2026-03-10, 2026-03-11 in Tokyo. + measuredAt: new Date("2026-03-11T06:00:00.000Z"), + }, + ], + { now: new Date("2026-03-20T00:00:00.000Z") }, + ); + expect([...day]).toEqual([0]); + const hour = await findFoldedWindowDuplicates( + prisma, + "tz-hour", + [ + { + type: "PULSE", + source: "APPLE_HEALTH", + externalId: "uuid-kolkata", + // 10:50 in Kolkata, 05:20 UTC: hour 05 in the new zone. + measuredAt: new Date("2026-01-10T05:20:00.000Z"), + }, + ], + { now: new Date("2026-06-01T00:00:00.000Z") }, + ); + expect([...hour]).toEqual([0]); + }); + + it("classes a person's deletion inside a half-folded day as B", async () => { + const prisma = getPrismaClient(); + await account("half-folded"); + await prisma.measurement.create({ + data: { + userId: "half-folded", + type: "WALKING_SPEED", + value: 1, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + externalId: `stats:${WS_HK}:2026-05-16`, + }, + }); + const { findCompactionTombstones } = + await import("@/lib/measurements/folded-window"); + const found = await findCompactionTombstones(prisma, "half-folded", "UTC", [ + { + type: "WALKING_SPEED", + source: "APPLE_HEALTH", + externalId: "uuid-deleted-by-person", + measuredAt: new Date("2026-05-16T20:00:00.000Z"), + // 38 hours later: past the grace, but the day was not over yet. + deletedAt: new Date("2026-05-18T10:00:00.000Z"), + }, + { + type: "WALKING_SPEED", + source: "APPLE_HEALTH", + externalId: "uuid-folded", + measuredAt: new Date("2026-05-16T08:00:00.000Z"), + deletedAt: new Date("2026-05-19T03:00:00.000Z"), + }, + ]); + expect([...found]).toEqual([1]); + }); +}); + +/** The prod shape: a mean an old two-run fold left, and its tombstones. */ +async function seedPartialFolds(userId: string) { + const prisma = getPrismaClient(); + const run1 = new Date("2026-09-20T03:00:00.000Z"); + const run2 = new Date("2026-09-21T03:00:00.000Z"); + await prisma.measurement.createMany({ + data: [ + // Wrong: the mean of the evening alone. + { + id: `${userId}-ws-wrong`, + userId, + type: "WALKING_SPEED", + value: 2, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-09-18T12:00:00.000Z"), + externalId: `stats:${WS_HK}:2026-09-18`, + }, + // Right already: one run took the whole day. + { + id: `${userId}-ws-right`, + userId, + type: "WALKING_SPEED", + value: 3, + unit: "m/s", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-09-17T12:00:00.000Z"), + externalId: `stats:${WS_HK}:2026-09-17`, + }, + // Wrong: the boundary hour holds the later half only. + { + id: `${userId}-hrv-wrong`, + userId, + type: "HEART_RATE_VARIABILITY", + value: 80, + unit: "ms", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-06-20T10:30:00.000Z"), + externalId: `stats:${HRV_HK}:2026-06-20T10`, + }, + ], + }); + const tomb = ( + id: string, + type: "WALKING_SPEED" | "HEART_RATE_VARIABILITY", + value: number, + iso: string, + deletedAt: Date, + syncVersion = 1, + ) => ({ + id: `${userId}-${id}`, + userId, + type, + value, + unit: type === "WALKING_SPEED" ? "m/s" : "ms", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(iso), + externalId: `uuid-${userId}-${id}`, + deletedAt, + syncVersion, + }); + await prisma.measurement.createMany({ + data: [ + tomb("ws1", "WALKING_SPEED", 1, "2026-09-18T08:00:00.000Z", run1), + tomb("ws2", "WALKING_SPEED", 1, "2026-09-18T10:00:00.000Z", run1), + tomb("ws3", "WALKING_SPEED", 2, "2026-09-18T20:00:00.000Z", run2), + tomb("ws4", "WALKING_SPEED", 2, "2026-09-18T22:00:00.000Z", run2), + // A person deleted this one: not part of the mean. + tomb("ws5", "WALKING_SPEED", 9, "2026-09-18T23:00:00.000Z", run2, 2), + tomb("ok1", "WALKING_SPEED", 3, "2026-09-17T08:00:00.000Z", run1), + tomb("ok2", "WALKING_SPEED", 3, "2026-09-17T20:00:00.000Z", run1), + tomb( + "hrv1", + "HEART_RATE_VARIABILITY", + 40, + "2026-06-20T10:05:00.000Z", + new Date("2026-09-18T10:30:00.000Z"), + ), + tomb( + "hrv2", + "HEART_RATE_VARIABILITY", + 40, + "2026-06-20T10:10:00.000Z", + new Date("2026-09-18T10:30:00.000Z"), + ), + tomb( + "hrv3", + "HEART_RATE_VARIABILITY", + 80, + "2026-06-20T10:50:00.000Z", + new Date("2026-09-19T10:30:00.000Z"), + ), + tomb( + "hrv4", + "HEART_RATE_VARIABILITY", + 80, + "2026-06-20T10:55:00.000Z", + new Date("2026-09-19T10:30:00.000Z"), + ), + ], + }); +} + +describe("fold repair", () => { + it("corrects a partial mean from the tombstones, leaves a correct one alone, and is idempotent", async () => { + const prisma = getPrismaClient(); + await account("repair"); + await seedPartialFolds("repair"); + const rightBefore = await live( + "repair", + "WALKING_SPEED", + `stats:${WS_HK}:2026-09-17`, + ); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + + const first = await repairFoldedMeans(prisma, "repair"); + expect(first.status).toBe("completed"); + expect(first.byType.WALKING_SPEED).toEqual({ + windowsChecked: 2, + windowsCorrected: 1, + restingCorrected: 0, + }); + expect(first.byType.HEART_RATE_VARIABILITY).toEqual({ + windowsChecked: 1, + windowsCorrected: 1, + restingCorrected: 0, + }); + expect( + (await live("repair", "WALKING_SPEED", `stats:${WS_HK}:2026-09-18`)) + ?.value, + ).toBeCloseTo(1.5, 9); + expect( + ( + await live( + "repair", + "HEART_RATE_VARIABILITY", + `stats:${HRV_HK}:2026-06-20T10`, + ) + )?.value, + ).toBe(60); + const rightAfter = await live( + "repair", + "WALKING_SPEED", + `stats:${WS_HK}:2026-09-17`, + ); + expect(rightAfter).toEqual(rightBefore); + + const again = await repairFoldedMeans(prisma, "repair"); + expect( + Object.values(again.byType).reduce( + (sum, c) => sum + c.windowsCorrected, + 0, + ), + ).toBe(0); + }); + + it("stops between days and resumes where it stopped", async () => { + const prisma = getPrismaClient(); + await account("repair-resume"); + await seedPartialFolds("repair-resume"); + const { repairFoldedMeans } = + await import("@/lib/jobs/measurement-fold-repair"); + let calls = 0; + const stopped = await repairFoldedMeans(prisma, "repair-resume", { + shouldStop: () => ++calls > 1, + }); + expect(stopped.status).toBe("stopped"); + expect(stopped.resume).toBeDefined(); + const rest = await repairFoldedMeans(prisma, "repair-resume", { + resume: stopped.resume, + }); + expect(rest.status).toBe("completed"); + expect( + ( + await live( + "repair-resume", + "WALKING_SPEED", + `stats:${WS_HK}:2026-09-18`, + ) + )?.value, + ).toBeCloseTo(1.5, 9); + expect( + ( + await live( + "repair-resume", + "HEART_RATE_VARIABILITY", + `stats:${HRV_HK}:2026-06-20T10`, + ) + )?.value, + ).toBe(60); + }); + + it("the handler marks the account and the purge refuses before it", async () => { + const prisma = getPrismaClient(); + await account("gate"); + await seedPartialFolds("gate"); + const { + purgeCompactionTombstones, + enqueueBootTimeCompactionTombstonePurge, + } = await import("@/lib/jobs/compaction-tombstone-purge"); + + const before = await purgeCompactionTombstones(prisma, { pauseMs: 0 }); + expect(before.deleted).toBe(0); + expect(before.awaitingRepairAccounts).toBe(1); + expect( + await prisma.measurement.count({ + where: { userId: "gate", deletedAt: { not: null } }, + }), + ).toBe(11); + + const send = vi.fn().mockResolvedValue("job-id"); + const { setGlobalBoss } = await import("@/lib/jobs/boss-instance"); + setGlobalBoss({ send } as never); + expect(await enqueueBootTimeCompactionTombstonePurge(prisma)).toEqual({ + enqueued: false, + }); + + const { handleMeasurementFoldRepair } = + await import("@/lib/jobs/measurement-fold-repair"); + await handleMeasurementFoldRepair( + [ + { + id: "run-1", + data: { userId: "gate" }, + expireInSeconds: 900, + } as never, + ], + prisma, + ); + expect( + await prisma.measurementFoldRepair.findUnique({ + where: { userId: "gate" }, + }), + ).not.toBeNull(); + // Finishing the account queued a purge run. + expect(send).toHaveBeenCalledWith( + "compaction-tombstone-purge", + expect.anything(), + expect.anything(), + ); + expect(await enqueueBootTimeCompactionTombstonePurge(prisma)).toEqual({ + enqueued: true, + }); + setGlobalBoss(null); + + const after = await purgeCompactionTombstones(prisma, { pauseMs: 0 }); + expect(after.awaitingRepairAccounts).toBe(0); + expect(after.deleted).toBeGreaterThan(0); + }); +}); + +describe("restore", () => { + it("keys the live stats ids by source: a Health Connect deletion under an Apple Health hour comes back", async () => { + const prisma = getPrismaClient(); + await account("restore-src", "UTC", true); + const hourId = `stats:${HR_HK}:2026-01-10T10`; + await prisma.measurement.createMany({ + data: [ + { + id: "restore-src-ah-hourly", + userId: "restore-src", + type: "PULSE", + value: 60, + unit: "bpm", + source: "APPLE_HEALTH", + measuredAt: new Date("2026-01-10T10:30:00.000Z"), + externalId: hourId, + }, + { + id: "restore-src-hc-deleted", + userId: "restore-src", + type: "PULSE", + value: 70, + unit: "bpm", + source: "HEALTH_CONNECT", + measuredAt: new Date("2026-01-10T10:20:00.000Z"), + externalId: "hc:deleted", + deletedAt: new Date(Date.now() - 2 * 86_400_000), + syncVersion: 2, + }, + ], + }); + // The repair finished before the file was exported, so the restore drops + // compaction tombstones; this one is not one. + await prisma.measurementFoldRepair.create({ + data: { + userId: "restore-src", + completedAt: new Date(Date.now() - 86_400_000), + }, + }); + const { storeBackupBlob } = await import("@/lib/export/store-backup-blob"); + const { streamFullBackupJson } = + await import("@/lib/export/full-backup-stream"); + const { id: backupId } = await storeBackupBlob( + prisma, + { userId: "restore-src", type: "WEEKLY_AUTO" }, + (write) => + streamFullBackupJson(prisma, "restore-src", write, { + purpose: "disaster-recovery", + }), + ); + const { POST } = await import("./restore-job-driver"); + const res = await POST( + new Request(`http://localhost/api/admin/backups/${backupId}/restore`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ confirm: "RESTORE" }), + }) as never, + { params: Promise.resolve({ id: backupId }) }, + ); + expect(res.status).toBe(200); + const ids = ( + await prisma.measurement.findMany({ + where: { userId: "restore-src" }, + select: { id: true }, + }) + ).map((row) => row.id); + expect(ids).toContain("restore-src-hc-deleted"); + }, 120_000); +});