From 0ba5334b572a9ae47964455aff665eb6299b7f48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 15 Jul 2026 22:26:00 +0200 Subject: [PATCH] fix(measurements): legacy step consolidation no longer sweeps provider daily totals WIP-commit (worktree safeguard; final assembly + version bump happen in the main tree after v1.28.36 merges). --- CHANGELOG.md | 13 + docs/api/openapi.yaml | 2 +- package.json | 2 +- public/sw.js | 2 +- .../step-consolidation-repair-queue.test.ts | 68 ++++ src/lib/jobs/reminder/register-rollup.ts | 59 ++++ src/lib/jobs/step-consolidation-repair.ts | 323 ++++++++++++++++++ src/lib/jobs/step-consolidation.ts | 42 ++- .../consolidate-legacy-steps.test.ts | 33 ++ .../measurements/consolidate-legacy-steps.ts | 146 +++++++- .../consolidate-legacy-steps.test.ts | 131 +++++++ .../step-consolidation-repair.test.ts | 210 ++++++++++++ 12 files changed, 1007 insertions(+), 24 deletions(-) create mode 100644 src/lib/jobs/__tests__/step-consolidation-repair-queue.test.ts create mode 100644 src/lib/jobs/step-consolidation-repair.ts create mode 100644 tests/integration/step-consolidation-repair.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index ba123c9de..56417186e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ ## [Unreleased] +## [1.28.37] — 2026-07-15 — Provider step totals survive a restart + +A boot-time maintenance pass that folds genuinely old raw step samples into +daily totals had too broad a reach: it also matched the daily step totals that +Google Health, Fitbit, and Withings write, and soft-deleted them on every +worker start. Recent days came back on the next sync, but older days stayed +gone until a full re-sync — which the next restart swept again. The pass now +skips any row that is already a daily total and only ever touches the ingest +paths that produced real legacy raw samples, so a provider's step history is +left alone. A one-time repair restores the totals removed by earlier restarts +(within the retention window; a single full sync recovers anything older) and +cleans up the placeholder totals the bug minted in their place. + ## [1.28.36] — 2026-07-15 — Re-importing after a failed Apple Health import actually retries The v1.28.33 rollup and staging fixes were correct but never reached the diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 188d7c073..10aac6148 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.28.36 + version: 1.28.37 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index 4163b9cd5..c93f29e06 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.28.36", + "version": "1.28.37", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index 34d07b2f8..223305f07 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.28.36"; + /* @sw-version-fallback */ "v1.28.37"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/lib/jobs/__tests__/step-consolidation-repair-queue.test.ts b/src/lib/jobs/__tests__/step-consolidation-repair-queue.test.ts new file mode 100644 index 000000000..8e0eeab75 --- /dev/null +++ b/src/lib/jobs/__tests__/step-consolidation-repair-queue.test.ts @@ -0,0 +1,68 @@ +/** + * v1.28.37 — step-consolidation repair queue registration guard + + * `extractStatsDay` unit coverage. + * + * Same source-text-grep approach as the drain + step-consolidation + + * mean-consolidation guards: assert the repair queue is registered in + * `allQueues`, a `boss.work` handler is wired against it, and the + * boot-discovery enqueue helper fires — without booting pg-boss + Prisma. + * An unregistered queue silently never drains (the dead-queue class). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { extractStatsDay } from "../step-consolidation-repair"; + +const REGISTRAR_PATH = join(__dirname, "..", "reminder", "register-rollup.ts"); +const source = readFileSync(REGISTRAR_PATH, "utf8"); + +describe("reminder-worker — step-consolidation-repair wiring", () => { + it("imports the queue symbols from the repair module", () => { + expect(source).toMatch( + /from\s*["']@\/lib\/jobs\/step-consolidation-repair["']/, + ); + expect(source).toMatch(/\bSTEP_CONSOLIDATION_REPAIR_QUEUE\b/); + expect(source).toMatch(/\brunStepConsolidationRepairForUser\b/); + expect(source).toMatch(/\benqueueBootTimeStepConsolidationRepair\b/); + }); + + it("registers the repair queue in the allQueues loop", () => { + const allQueuesMatch = source.match( + /const allQueues\s*=\s*\[([\s\S]*?)\];/, + ); + expect(allQueuesMatch).not.toBeNull(); + expect(allQueuesMatch![1]).toMatch(/\bSTEP_CONSOLIDATION_REPAIR_QUEUE\b/); + }); + + it("registers a boss.work handler against the repair queue", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,200}STEP_CONSOLIDATION_REPAIR_QUEUE[\s\S]{0,400}runStepConsolidationRepairForUser/, + ); + }); + + it("fires the boot-discovery enqueue helper", () => { + expect(source).toMatch( + /await enqueueBootTimeStepConsolidationRepair\([^)]*\)/, + ); + }); +}); + +describe("extractStatsDay", () => { + it("extracts the day key from a provider stats:steps externalId", () => { + expect(extractStatsDay("stats:steps:2026-05-16")).toBe("2026-05-16"); + }); + + it("extracts the day key from the narrow Apple mint shape", () => { + expect( + extractStatsDay("stats:HKQuantityTypeIdentifierStepCount:2026-05-16"), + ).toBe("2026-05-16"); + }); + + it("returns null for a null externalId or a non-day-suffixed key", () => { + expect(extractStatsDay(null)).toBeNull(); + expect(extractStatsDay("uuid-abc-123")).toBeNull(); + expect(extractStatsDay("stats:steps:not-a-date")).toBeNull(); + }); +}); diff --git a/src/lib/jobs/reminder/register-rollup.ts b/src/lib/jobs/reminder/register-rollup.ts index 2affd68f8..ded392022 100644 --- a/src/lib/jobs/reminder/register-rollup.ts +++ b/src/lib/jobs/reminder/register-rollup.ts @@ -60,6 +60,13 @@ import { enqueueBootTimeStepConsolidation, type StepConsolidationPayload, } from "@/lib/jobs/step-consolidation"; +import { + STEP_CONSOLIDATION_REPAIR_QUEUE, + STEP_CONSOLIDATION_REPAIR_CONCURRENCY, + runStepConsolidationRepairForUser, + enqueueBootTimeStepConsolidationRepair, + type StepConsolidationRepairPayload, +} from "@/lib/jobs/step-consolidation-repair"; import { MEAN_CONSOLIDATION_QUEUE, MEAN_CONSOLIDATION_CONCURRENCY, @@ -122,6 +129,9 @@ const allQueues = [ // v1.5.6 — boot-time legacy step consolidation. Discovery enqueues // one job per user still holding live pre-v1.5.0 granular step rows. STEP_CONSOLIDATION_QUEUE, + // v1.28.37 — one-shot repair for the provider step rows the pre-fix + // consolidation swept. Boot-discovery driven, self-converging. + STEP_CONSOLIDATION_REPAIR_QUEUE, // v1.7.0 — daily-mean consolidation for high-frequency spot HealthKit // metrics (walking speed/step length, respiratory rate, audio exposure). MEAN_CONSOLIDATION_QUEUE, @@ -219,6 +229,25 @@ export async function registerRollupQueues( }, ); + // v1.28.37 — step-consolidation repair worker. Resurrects the + // GOOGLE_HEALTH / FITBIT daily-total step rows the pre-fix consolidation + // swept and removes the shadow MANUAL totals it minted. Serial + // concurrency so the repair never crowds the dashboard request pool. + await boss.work( + STEP_CONSOLIDATION_REPAIR_QUEUE, + { localConcurrency: STEP_CONSOLIDATION_REPAIR_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + const summary = await runStepConsolidationRepairForUser(userId); + workerLog( + "info", + `[step-consolidation-repair] user=${userId} resurrected=${summary.rowsResurrected} wedgeSkipped=${summary.wedgeSkipped} manualMintsRemoved=${summary.manualMintsRemoved} daysRecomputed=${summary.daysRecomputed} failures=${summary.failures}`, + ); + } + }, + ); + // v1.7.0 — daily-mean consolidation worker. The boot enqueue helper // below sends one job per user holding live per-sample high-frequency // mean-type rows; this handler collapses each completed day to its @@ -712,4 +741,34 @@ export async function enqueueRollupBootDiscovery(): Promise { err, ); } + + // v1.28.37 — step-consolidation repair. Finds every account still holding + // a resurrectable tombstoned GOOGLE_HEALTH / FITBIT daily-total step row + // (swept by the pre-fix consolidation) and enqueues one repair job per + // account. Self-converging: a resurrected row goes live and drops out of + // the discovery predicate. + try { + // Stage 5. + const { enqueued, skipped, error } = + await enqueueBootTimeStepConsolidationRepair( + BOOT_BACKFILL_STAGGER_SECONDS * 5, + ); + if (error) { + workerLog( + "error", + `[step-consolidation-repair] boot discovery failed: ${error}`, + ); + } else { + workerLog( + "info", + `[step-consolidation-repair] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[step-consolidation-repair] boot discovery threw an unexpected error", + err, + ); + } } diff --git a/src/lib/jobs/step-consolidation-repair.ts b/src/lib/jobs/step-consolidation-repair.ts new file mode 100644 index 000000000..a0b1db8d2 --- /dev/null +++ b/src/lib/jobs/step-consolidation-repair.ts @@ -0,0 +1,323 @@ +/** + * v1.28.37 — pg-boss queue + boot-time one-shot repair for the step + * data-loss the pre-fix legacy step consolidation caused. + * + * Background: before this release the boot-time legacy step consolidation + * (`consolidate-legacy-steps.ts`) discovered any live `ACTIVITY_STEPS` row + * whose externalId was not the narrow `stats:HKQuantityTypeIdentifierStepCount:` + * shape and soft-deleted it — sweeping GOOGLE_HEALTH and FITBIT daily-total + * rows (`stats:steps:`) on every worker boot and, on days without an + * Apple total, minting a shadow `MANUAL` `stats:HK…` total from the summed + * provider value. The predicate + source-pin fix stops the sweep going + * forward; this job repairs the rows already tombstoned. + * + * What it does, per account, self-convergingly: + * 1. RESURRECT (`deletedAt: null`) tombstoned GOOGLE_HEALTH / FITBIT + * `stats:%` `ACTIVITY_STEPS` rows within the 75-day tombstone-retention + * horizon (older tombstones are already hard-pruned; those days need one + * final full sync per account). Wedge-safe: a row is resurrected ONLY if + * no live row already occupies its `(userId, type, source, measuredAt)` + * slot — the tombstone-wedge lesson (probe first). A per-row P2002 + * catch is a belt on top of the probe for the concurrent-write race. + * 2. REMOVE the shadow MANUAL `stats:HK…:` total the bug minted, but + * ONLY for a day where a live GOOGLE_HEALTH / FITBIT `stats:steps:` + * total now returns (soft-delete, recoverable — never hard-delete). A + * MANUAL total on a day with no provider row is left untouched: it may + * be a genuine Apple-legacy consolidation. + * 3. Recompute the DAY rollup bucket for every touched day so the dashboard + * tier follows the resurrected/removed rows. + * + * Converges to zero without a marker column: once a tombstoned provider row + * is resurrected it is live and drops out of the discovery predicate; a + * wedge-blocked row (a live sibling already occupies its slot — the data is + * NOT lost) is excluded from discovery by the `NOT EXISTS` clause, so it + * never keeps an account on the list. A re-run (reboot mid-walk) re-selects + * only the still-tombstoned, still-resurrectable rows. + * + * The queue name MUST be registered in `allQueues` in + * `src/lib/jobs/reminder/register-rollup.ts` or pg-boss never provisions it + * and the boot enqueue silently never drains (the v1.4.37 dead-queue class). + */ +import type { MeasurementSource } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { isP2002 as isUniqueConstraintViolation } from "@/lib/prisma-errors"; +import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; +import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; + +export const STEP_CONSOLIDATION_REPAIR_QUEUE = "step-consolidation-repair"; + +/** + * Serial concurrency — a repair walks one account's tombstoned provider + * step rows and writes per day; concurrency-1 keeps it from crowding the + * request pool, matching `STEP_CONSOLIDATION_CONCURRENCY`. + */ +export const STEP_CONSOLIDATION_REPAIR_CONCURRENCY = 1; + +/** Providers whose swept daily-total step rows this job resurrects. */ +const REPAIR_PROVIDER_SOURCES: readonly MeasurementSource[] = [ + "GOOGLE_HEALTH", + "FITBIT", +] as const; + +/** The narrow `stats:HK…` shape the bug minted its shadow MANUAL totals in. */ +const STEP_MANUAL_MINT_PREFIX = "stats:HKQuantityTypeIdentifierStepCount:"; + +const STEP_TYPE = "ACTIVITY_STEPS" as const; + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** Trailing `YYYY-MM-DD` day key of a `stats::` externalId. */ +const STATS_DAY_RE = /(\d{4}-\d{2}-\d{2})$/; + +export interface StepConsolidationRepairPayload { + userId: string; + enqueuedAt: string; +} + +export interface StepConsolidationRepairSummary { + /** Tombstoned provider daily totals brought back live. */ + rowsResurrected: number; + /** Rows left tombstoned because a live sibling occupies their slot. */ + wedgeSkipped: number; + /** Shadow MANUAL totals soft-deleted because a provider row now returns. */ + manualMintsRemoved: number; + /** Distinct DAY rollup buckets recomputed. */ + daysRecomputed: number; + /** Rows/mints stepped over after an unexpected write error. */ + failures: number; +} + +/** Extract the `YYYY-MM-DD` day key from a `stats:` externalId, or null. */ +export function extractStatsDay(externalId: string | null): string | null { + if (!externalId) return null; + const match = STATS_DAY_RE.exec(externalId); + return match ? match[1] : null; +} + +/** + * Per-user repair handler. Idempotent and self-converging — see the file + * header. Returns the summary totals so the worker can log them. + */ +export async function runStepConsolidationRepairForUser( + userId: string, +): Promise { + const summary: StepConsolidationRepairSummary = { + rowsResurrected: 0, + wedgeSkipped: 0, + manualMintsRemoved: 0, + daysRecomputed: 0, + failures: 0, + }; + + const cutoff = new Date(Date.now() - TOMBSTONE_RETENTION_DAYS * DAY_MS); + + // Touched UTC-day → a representative instant in that day, for one rollup + // recompute per affected bucket (the rollup tier keys DAY buckets by UTC + // day; a resurrected provider row and a removed local-noon mint can land + // in different UTC days, so both instants are tracked). + const touchedByUtcDay = new Map(); + const markTouched = (instant: Date) => { + touchedByUtcDay.set(instant.toISOString().slice(0, 10), instant); + }; + + // ── 1. Resurrect tombstoned provider daily totals ──────────────────── + const tombstoned = await prisma.measurement.findMany({ + where: { + userId, + type: STEP_TYPE, + source: { in: [...REPAIR_PROVIDER_SOURCES] }, + externalId: { startsWith: "stats:" }, + // Within the retention horizon: older tombstones are already pruned. + deletedAt: { not: null, gte: cutoff }, + }, + select: { id: true, source: true, measuredAt: true, externalId: true }, + }); + + for (const row of tombstoned) { + // Wedge probe: never resurrect into a slot a live row already holds — + // the second unique index `(userId, type, measuredAt, source, sleepStage)` + // covers tombstones, so a static collision cannot arise, but a + // concurrent provider sync could have re-created the day live. Skip it: + // the live row is the current truth and the data is not lost. + const liveSibling = await prisma.measurement.findFirst({ + where: { + userId, + type: STEP_TYPE, + source: row.source, + measuredAt: row.measuredAt, + sleepStage: null, + deletedAt: null, + }, + select: { id: true }, + }); + if (liveSibling) { + summary.wedgeSkipped += 1; + continue; + } + + try { + await prisma.measurement.update({ + where: { id: row.id }, + data: { deletedAt: null }, + }); + summary.rowsResurrected += 1; + markTouched(row.measuredAt); + } catch (err) { + // A racing insert between the probe and this update can trip the + // unique index. Step over the row — a later boot re-selects it. + if (isUniqueConstraintViolation(err)) { + summary.wedgeSkipped += 1; + continue; + } + summary.failures += 1; + } + } + + // ── 2. Remove shadow MANUAL mints where a provider row now returns ──── + const manualMints = await prisma.measurement.findMany({ + where: { + userId, + type: STEP_TYPE, + source: "MANUAL", + externalId: { startsWith: STEP_MANUAL_MINT_PREFIX }, + deletedAt: null, + }, + select: { id: true, measuredAt: true, externalId: true }, + }); + + for (const mint of manualMints) { + const day = extractStatsDay(mint.externalId); + if (!day) continue; + + // Only remove the mint when a LIVE provider daily total for the same + // day exists — that is the discriminator between a spurious shadow of + // provider data and a genuine Apple-legacy consolidation. + const providerRow = await prisma.measurement.findFirst({ + where: { + userId, + type: STEP_TYPE, + source: { in: [...REPAIR_PROVIDER_SOURCES] }, + externalId: `stats:steps:${day}`, + deletedAt: null, + }, + select: { id: true }, + }); + if (!providerRow) continue; + + try { + await prisma.measurement.update({ + where: { id: mint.id }, + data: { deletedAt: new Date() }, + }); + summary.manualMintsRemoved += 1; + markTouched(mint.measuredAt); + } catch { + summary.failures += 1; + } + } + + // ── 3. Recompute the DAY rollup bucket for every touched day ────────── + for (const instant of touchedByUtcDay.values()) { + try { + await recomputeBucketsForMeasurement(userId, STEP_TYPE, instant); + summary.daysRecomputed += 1; + } catch { + summary.failures += 1; + } + } + + // Stable wide-event action name. No-ops outside a request/job context. + annotate({ + action: { + name: "measurement.step.repair", + details: { + rows_resurrected: summary.rowsResurrected, + wedge_skipped: summary.wedgeSkipped, + manual_mints_removed: summary.manualMintsRemoved, + days_recomputed: summary.daysRecomputed, + failures: summary.failures, + }, + }, + }); + + return summary; +} + +/** + * Boot-time discovery. Finds every account still holding a resurrectable + * tombstoned GOOGLE_HEALTH / FITBIT `stats:%` step row (tombstoned within + * the retention horizon, with no live sibling occupying its slot) and + * enqueues one repair job per account. + * + * Self-converging: a resurrected row goes live and drops out; a + * wedge-blocked row is excluded by the `NOT EXISTS` clause so it never + * keeps an account on the list. Best-effort: errors return through the + * result so worker boot never fails on a repair miss. + */ +export async function enqueueBootTimeStepConsolidationRepair( + startAfterSeconds: number = 0, +): Promise<{ enqueued: number; skipped: number; error: string | null }> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + // Splice-free: every literal is a compile-time constant; the horizon + // is a bound parameter. `NOT EXISTS` excludes wedge-blocked rows so the + // set shrinks to empty once every resurrectable row is live. + const users = await prisma.$queryRaw>` + SELECT DISTINCT t."user_id" AS id + FROM measurements t + WHERE t."type" = 'ACTIVITY_STEPS' + AND t."deleted_at" IS NOT NULL + AND t."deleted_at" >= now() - make_interval(days => ${TOMBSTONE_RETENTION_DAYS}) + AND t."source" IN ('GOOGLE_HEALTH', 'FITBIT') + AND t."external_id" LIKE 'stats:%' + AND NOT EXISTS ( + SELECT 1 + FROM measurements l + WHERE l."user_id" = t."user_id" + AND l."type" = t."type" + AND l."source" = t."source" + AND l."measured_at" = t."measured_at" + AND l."sleep_stage" IS NULL + AND l."deleted_at" IS NULL + ) + `; + + if (users.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { id } of users) { + const payload: StepConsolidationRepairPayload = { + userId: id, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(STEP_CONSOLIDATION_REPAIR_QUEUE, payload, { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + singletonKey: `step-consolidation-repair|${id}`, + ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), + }); + 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), + }; + } +} diff --git a/src/lib/jobs/step-consolidation.ts b/src/lib/jobs/step-consolidation.ts index 0fd6ad896..e49747268 100644 --- a/src/lib/jobs/step-consolidation.ts +++ b/src/lib/jobs/step-consolidation.ts @@ -55,6 +55,20 @@ export async function runStepConsolidationForUser( }, }, }); + // Regression guardrail. `providerRowsSkipped` is 0 in normal operation — + // the source pin + broad `stats:` skip keep every provider-owned / daily- + // total step row out of the consolidation scan. A non-zero value means a + // future change re-broadened the scan and a provider's daily total was + // about to be soft-deleted (the Google/Fitbit data-loss bug). Emit the + // wide event only when the canary trips so the signal is cheap and loud. + if (summary.totals.providerRowsSkipped > 0) { + annotate({ + action: { + name: "measurement.step.provider_row_skipped", + details: { count: summary.totals.providerRowsSkipped }, + }, + }); + } return { daysConsolidated: summary.totals.daysConsolidated, legacyRowsSoftDeleted: summary.totals.legacyRowsSoftDeleted, @@ -63,9 +77,10 @@ export async function runStepConsolidationForUser( /** * Boot-time discovery. Finds every user with at least one LIVE legacy - * step row (an `ACTIVITY_STEPS` row whose `externalId` is NULL or does - * NOT start with the daily-stats prefix, and that is not tombstoned) - * and enqueues one consolidation job per account. + * step row — a non-tombstoned `ACTIVITY_STEPS` row from a writable source + * (`APPLE_HEALTH` / `MANUAL` / `IMPORT`) whose `externalId` is NULL or is + * NOT already in a provider's daily-total `stats:%` shape — and enqueues + * one consolidation job per account. * * Idempotent across reboots: once a user's legacy rows are * soft-deleted, the `deleted_at IS NULL` predicate drops them from the @@ -93,19 +108,28 @@ export async function enqueueBootTimeStepConsolidation( } try { - // The daily-total externalId shape is `stats::`. A - // live row is a legacy candidate when its externalId does NOT match - // that prefix (NULL externalId — a manual/legacy row — also - // qualifies because `LIKE` against NULL is NULL/false). Splice-free: - // the prefix is a constant compile-time literal, not user input. + // A live row is a legacy-consolidation candidate when BOTH: + // - its `source` is a client/user-writable ingest path + // (`APPLE_HEALTH` / `MANUAL` / `IMPORT`) — the only sources that + // ever wrote raw per-sample step rows; and + // - its `external_id` is NOT already in ANY provider's daily-total + // `stats:%` shape (NULL externalId — a manual/legacy row — also + // qualifies because `LIKE` against NULL is NULL/false). + // Both scopes mirror `buildScanWhere` in `consolidate-legacy-steps.ts` + // so discovery and scan agree exactly. The source pin is what protects + // WITHINGS (keyed `withings:activity:…`, not `stats:`) and the + // GOOGLE_HEALTH / FITBIT daily totals (`stats:steps:…`) from the boot + // sweep that this closes. Splice-free: every literal is a compile-time + // constant, not user input. const users = await prisma.$queryRaw>` SELECT DISTINCT m."user_id" AS id FROM measurements m WHERE m."type" = 'ACTIVITY_STEPS' AND m."deleted_at" IS NULL + AND m."source" IN ('APPLE_HEALTH', 'MANUAL', 'IMPORT') AND ( m."external_id" IS NULL - OR m."external_id" NOT LIKE 'stats:HKQuantityTypeIdentifierStepCount:%' + OR m."external_id" NOT LIKE 'stats:%' ) `; diff --git a/src/lib/measurements/__tests__/consolidate-legacy-steps.test.ts b/src/lib/measurements/__tests__/consolidate-legacy-steps.test.ts index 4f3438d75..16d358f7b 100644 --- a/src/lib/measurements/__tests__/consolidate-legacy-steps.test.ts +++ b/src/lib/measurements/__tests__/consolidate-legacy-steps.test.ts @@ -53,6 +53,39 @@ describe("bucketLegacyStepRows", () => { expect(byDay.get("2026-05-16")?.[0].id).toBe("legacy"); }); + it("skips a Google Health daily-total row (stats:steps:)", () => { + // Regression: Google + Fitbit daily steps carry `stats:steps:`, + // which does NOT start with the narrow `stats:HK…` prefix. The broad + // `stats:` skip must exclude them so they are never bucketed (and + // therefore never soft-deleted) — the data-loss bug this closes. + const tz = "Europe/Berlin"; + const rows = [ + row({ + id: "google-total", + externalId: "stats:steps:2026-05-16", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + }), + row({ id: "legacy", measuredAt: new Date("2026-05-16T08:00:00.000Z") }), + ]; + const byDay = bucketLegacyStepRows(rows, tz); + expect(byDay.get("2026-05-16")).toHaveLength(1); + expect(byDay.get("2026-05-16")?.[0].id).toBe("legacy"); + }); + + it("skips a Fitbit daily-total row (identical stats:steps: shape)", () => { + const byDay = bucketLegacyStepRows( + [ + row({ + id: "fitbit-total", + externalId: "stats:steps:2026-05-17", + measuredAt: new Date("2026-05-17T12:00:00.000Z"), + }), + ], + "Europe/Berlin", + ); + expect(byDay.has("2026-05-17")).toBe(false); + }); + it("buckets a late-evening UTC sample into the next local day", () => { // 23:30 UTC on 2026-05-16 is 01:30 on 2026-05-17 in Berlin (UTC+2). const byDay = bucketLegacyStepRows( diff --git a/src/lib/measurements/consolidate-legacy-steps.ts b/src/lib/measurements/consolidate-legacy-steps.ts index 5ddfe1519..3cdcecea7 100644 --- a/src/lib/measurements/consolidate-legacy-steps.ts +++ b/src/lib/measurements/consolidate-legacy-steps.ts @@ -11,8 +11,15 @@ * daily shape so historical step data reads consistently. * * Per user × calendar day (anchored to `User.timezone`): - * 1. SELECT live (`deletedAt IS NULL`) `ACTIVITY_STEPS` rows whose - * `externalId` does NOT start with the daily-stats prefix. + * 1. SELECT live (`deletedAt IS NULL`) `ACTIVITY_STEPS` rows that are + * genuine legacy raw per-sample rows — `source` in the writable set + * (`APPLE_HEALTH` / `MANUAL` / `IMPORT`) AND `externalId` NOT already + * in ANY provider's daily-total `stats:` shape. A row keyed `stats:%` + * is already an aggregated daily total (Apple `stats:HK…`, Google / + * Fitbit `stats:steps:…`) and must never be re-consolidated; a + * provider-owned row (`GOOGLE_HEALTH` / `FITBIT` / `WITHINGS` / …) is a + * server-side daily aggregate that this pass must never touch. See the + * scope note on `buildScanWhere` below. * 2. Group into per-day buckets in the user's timezone. * 3. SUM the legacy values for the day. * 4. UPSERT the canonical daily-total row keyed by @@ -37,7 +44,10 @@ * production standalone image strips `tsx`. Modelled on the * `rollup-full-backfill` boot-time converging-backfill pattern. */ -import type { PrismaClient } from "@/generated/prisma/client"; +import type { + MeasurementSource, + PrismaClient, +} from "@/generated/prisma/client"; import { isP2002 as isUniqueConstraintViolation } from "@/lib/prisma-errors"; import { recomputeBucketsForMeasurement } from "@/lib/rollups/measurement-rollups"; @@ -63,6 +73,60 @@ const STEP_HK_IDENTIFIER = "HKQuantityTypeIdentifierStepCount"; */ export const STEP_DAILY_STATS_PREFIX = `stats:${STEP_HK_IDENTIFIER}:`; +/** + * Broad daily-total prefix marking ANY provider's already-aggregated + * daily step row. Apple daily totals carry `stats:HKQuantityType…`, + * Google + Fitbit carry `stats:steps:`, and any future + * stats-contract provider carries `stats::`. A row whose + * externalId starts with this is, by construction, already a daily total + * and must NEVER be discovered, scanned, or bucketed for consolidation — + * consolidation targets ONLY the raw per-sample rows that predate the + * daily-stats split. + * + * This is the skip-prefix the scan `where` and the bucketer use. + * `STEP_DAILY_STATS_PREFIX` (narrow) stays the shape this pass MINTS for + * a genuinely-consolidated legacy day; the two are deliberately + * different — mint narrow, skip broad. + * + * The original v1.5.6 pass skipped only the narrow `stats:HK…` prefix. + * Google Health daily steps (`stats:steps:`, v1.27.0) and Fitbit + * (identical shape, v1.12.0) re-qualified under that narrow skip on every + * worker boot and were soft-deleted — the data-loss bug this widening + * closes. + */ +export const ANY_DAILY_STATS_PREFIX = "stats:"; + +/** + * The only `MeasurementSource`s that can hold genuine pre-v1.5.0 legacy + * raw step rows: the client/user-writable ingest paths. + * + * - `APPLE_HEALTH` — the HealthKit batch ingest that wrote per-sample + * step rows before iOS moved to the daily-`stats:` total. + * - `MANUAL` — hand-entered step readings. + * - `IMPORT` — the CSV / Apple-Health-export bulk import. + * + * Every OTHER source is a server-owned daily aggregate that must never be + * consolidated: `WITHINGS` (keyed `withings:activity:…`, NOT `stats:`, so + * the `stats:%` skip alone would NOT protect it — this pin is what does), + * `GOOGLE_HEALTH` / `FITBIT` (`stats:steps:…`), and + * `WHOOP` / `POLAR` / `OURA` / `NIGHTSCOUT` / `STRAVA` / `COMPUTED`. + * + * Pinning here matches every sibling drain (all pin `source`); the v1.5.6 + * pass was the only outlier, written when Apple + manual were the only + * step sources. The pin does NOT reduce coverage of genuine legacy raw + * rows — those are `APPLE_HEALTH` / `MANUAL` / `IMPORT` by construction. + */ +export const LEGACY_STEP_CONSOLIDATABLE_SOURCES: readonly MeasurementSource[] = + ["APPLE_HEALTH", "MANUAL", "IMPORT"] as const; + +/** + * Fast-membership set of the consolidatable sources for the in-band + * regression canary (`providerRowsSkipped`). + */ +const CONSOLIDATABLE_SOURCE_SET: ReadonlySet = new Set( + LEGACY_STEP_CONSOLIDATABLE_SOURCES, +); + /** Per-(user, day) action summary. */ export interface StepConsolidationBucket { userId: string; @@ -99,6 +163,18 @@ export interface StepConsolidationSummary { * of the pass — the pass stays converging for every other day. */ daysSkippedOnConflict: number; + /** + * Regression canary. Counts source rows that reached a per-day bucket + * (i.e. were about to be soft-deleted) yet are a provider-owned source + * OR carry a `stats:%` daily-total externalId. With the source pin + + * broad `stats:` skip in place this is ALWAYS 0 — a provider / daily-total + * row can never enter the scan. A non-zero value means a future change + * re-broadened the scan and a provider's daily total is being eaten + * again (the v1.27.0/v1.12.0 Google/Fitbit data-loss bug). Surfaced as + * the `measurement.step.provider_row_skipped` wide event so a dashboard + * alerts on the regression. + */ + providerRowsSkipped: number; }; } @@ -121,7 +197,9 @@ export function bucketLegacyStepRows( rows: readonly PerSampleRow[], tz: string, ): Map { - return bucketRowsByDay(rows, tz, STEP_DAILY_STATS_PREFIX); + // Skip on the BROAD `stats:` prefix so a daily total from ANY provider + // (Apple `stats:HK…`, Google / Fitbit `stats:steps:…`) is never bucketed. + return bucketRowsByDay(rows, tz, ANY_DAILY_STATS_PREFIX); } /** SUM the values in a per-day legacy bucket. */ @@ -154,6 +232,7 @@ export async function consolidateLegacySteps( dailyRowsUpserted: 0, daysFoldedIntoExisting: 0, daysSkippedOnConflict: 0, + providerRowsSkipped: 0, }, }; @@ -172,18 +251,41 @@ export async function consolidateLegacySteps( // Single type — the HK identifier is fixed. hkIdentifierForType: () => STEP_HK_IDENTIFIER, dailyStatsExternalId, - statsPrefix: STEP_DAILY_STATS_PREFIX, + // Skip on the BROAD `stats:` prefix (not the narrow `stats:HK…` mint + // prefix). This drives both the scan `NOT startsWith` and the bucketer + // skip, so a daily total from ANY provider is excluded at both sites. + statsPrefix: ANY_DAILY_STATS_PREFIX, reduce: sumLegacyStepValues, - // Live legacy step rows whose externalId is NOT the daily-stats - // shape. The `stats:` daily-total row (if present) is read separately - // per day; we exclude it here so it is never soft-deleted. No - // source-scope — legacy granular rows predate the source split, so - // every source is in scope. Soft-deleted rows are excluded so a - // re-run after a partial pass never re-aggregates. + // Select `source` too so the in-band regression canary + // (`providerRowsSkipped`) can inspect what reached a bucket. + scanSelect: { + id: true, + type: true, + value: true, + measuredAt: true, + externalId: true, + source: true, + }, + // Live legacy step rows that are genuine pre-v1.5.0 raw per-sample + // rows. TWO scopes, both required: + // 1. `source IN (APPLE_HEALTH, MANUAL, IMPORT)` — the only sources + // that ever wrote raw per-sample step rows. This is what protects + // WITHINGS (keyed `withings:activity:…`, NOT `stats:`, so the + // prefix skip below would miss it), plus GOOGLE_HEALTH / FITBIT / + // WHOOP / … server-owned daily aggregates. Matches every sibling + // drain, which all pin `source`. + // 2. `NOT startsWith 'stats:'` — a row already in a provider's + // daily-total shape (Apple `stats:HK…`, Google/Fitbit + // `stats:steps:…`, and the MANUAL totals this pass itself mints) + // is already aggregated and is read separately per day; excluding + // it here means it is never soft-deleted or re-consolidated. + // Soft-deleted rows are excluded so a re-run after a partial pass never + // re-aggregates. buildScanWhere: ({ userId, type, statsPrefix }) => ({ userId, type, deletedAt: null, + source: { in: [...LEGACY_STEP_CONSOLIDATABLE_SOURCES] }, NOT: { externalId: { startsWith: statsPrefix } }, }), // Does a post-v1.5.0 daily total already exist for this day? If so it @@ -307,6 +409,26 @@ export async function consolidateLegacySteps( return; } + // In-band regression canary. With the scope pins above, no + // provider-owned or `stats:%`-keyed row can reach a bucket, so this + // stays 0. If a future change re-broadens the scan, the offending + // rows show up here and the wide event fires — catching the + // Google/Fitbit data-loss regression before it eats history again. + // `source` is present because the legacy-steps `scanSelect` requests + // it; the shared `PerSampleRow` shape does not carry it, so read it + // through a local widening. + for (const r of dayRows as ReadonlyArray< + PerSampleRow & { source?: MeasurementSource } + >) { + const isProviderOwned = + r.source !== undefined && !CONSOLIDATABLE_SOURCE_SET.has(r.source); + const isDailyTotal = + r.externalId?.startsWith(ANY_DAILY_STATS_PREFIX) ?? false; + if (isProviderOwned || isDailyTotal) { + summary.totals.providerRowsSkipped += 1; + } + } + const hadExistingTotal = !shouldMint; summary.buckets.push({ userId, @@ -350,7 +472,7 @@ export async function consolidateLegacySteps( summary.totals.usersScanned = usersScanned; log( - `[step-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} legacyRowsSoftDeleted=${summary.totals.legacyRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFoldedIntoExisting=${summary.totals.daysFoldedIntoExisting} daysSkippedOnConflict=${summary.totals.daysSkippedOnConflict}${options.dryRun ? " (dry-run)" : ""}`, + `[step-consolidation] done — usersScanned=${summary.totals.usersScanned} daysConsolidated=${summary.totals.daysConsolidated} legacyRowsSoftDeleted=${summary.totals.legacyRowsSoftDeleted} dailyRowsUpserted=${summary.totals.dailyRowsUpserted} daysFoldedIntoExisting=${summary.totals.daysFoldedIntoExisting} daysSkippedOnConflict=${summary.totals.daysSkippedOnConflict} providerRowsSkipped=${summary.totals.providerRowsSkipped}${options.dryRun ? " (dry-run)" : ""}`, ); return summary; diff --git a/tests/integration/consolidate-legacy-steps.test.ts b/tests/integration/consolidate-legacy-steps.test.ts index 2698568fb..7e7336aa8 100644 --- a/tests/integration/consolidate-legacy-steps.test.ts +++ b/tests/integration/consolidate-legacy-steps.test.ts @@ -168,6 +168,137 @@ describe("consolidateLegacySteps (real Postgres)", () => { expect(tombstoned).toBe(2); }); + it("leaves Google Health AND Fitbit daily-total rows untouched (the data-loss bug, pinned for both providers)", async () => { + const prisma = getPrismaClient(); + // Provider daily totals, keyed `stats:steps:`, server-owned. + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 8123, + unit: "steps", + source: "GOOGLE_HEALTH", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + externalId: "stats:steps:2026-05-16", + }, + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 9456, + unit: "steps", + source: "FITBIT", + measuredAt: new Date("2026-05-17T12:00:00.000Z"), + externalId: "stats:steps:2026-05-17", + }, + ], + }); + + const summary = await consolidateLegacySteps(prisma, { + userId: TEST_USER_ID, + dryRun: false, + log: () => {}, + }); + + // Nothing discovered, nothing bucketed, nothing minted — the canary + // stays 0 (no provider row ever reached the scan). + expect(summary.totals.daysConsolidated).toBe(0); + expect(summary.totals.legacyRowsSoftDeleted).toBe(0); + expect(summary.totals.dailyRowsUpserted).toBe(0); + expect(summary.totals.providerRowsSkipped).toBe(0); + + // Both provider rows are still live, unchanged; no MANUAL mint created. + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS" }, + orderBy: { measuredAt: "asc" }, + }); + expect(rows).toHaveLength(2); + expect(rows.every((r) => r.deletedAt === null)).toBe(true); + expect(rows.map((r) => r.source)).toEqual(["GOOGLE_HEALTH", "FITBIT"]); + expect(rows.map((r) => r.value)).toEqual([8123, 9456]); + const manual = await prisma.measurement.count({ + where: { userId: TEST_USER_ID, source: "MANUAL" }, + }); + expect(manual).toBe(0); + }); + + it("leaves a Withings daily-total row untouched (source pin, not stats:-keyed)", async () => { + const prisma = getPrismaClient(); + // Withings keys `withings:activity:…` — NOT `stats:`. Only the source + // pin protects it; the prefix skip alone would miss it. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 7000, + unit: "steps", + source: "WITHINGS", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + externalId: `withings:activity:${TEST_USER_ID}:2026-05-16:steps`, + }, + }); + + const summary = await consolidateLegacySteps(prisma, { + userId: TEST_USER_ID, + dryRun: false, + log: () => {}, + }); + + expect(summary.totals.daysConsolidated).toBe(0); + expect(summary.totals.providerRowsSkipped).toBe(0); + const live = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS", deletedAt: null }, + }); + expect(live).toHaveLength(1); + expect(live[0].source).toBe("WITHINGS"); + expect(live[0].value).toBe(7000); + }); + + it("still consolidates genuine legacy IMPORT and MANUAL raw rows (the drain is not disabled)", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 1000, + unit: "steps", + source: "IMPORT", + measuredAt: new Date("2026-05-16T08:00:00.000Z"), + externalId: "import-legacy-1", + }, + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 500, + unit: "steps", + source: "MANUAL", + measuredAt: new Date("2026-05-16T18:00:00.000Z"), + externalId: "manual-legacy-1", + }, + ], + }); + + const summary = await consolidateLegacySteps(prisma, { + userId: TEST_USER_ID, + dryRun: false, + log: () => {}, + }); + + expect(summary.totals.daysConsolidated).toBe(1); + expect(summary.totals.legacyRowsSoftDeleted).toBe(2); + expect(summary.totals.dailyRowsUpserted).toBe(1); + expect(summary.totals.providerRowsSkipped).toBe(0); + + const live = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS", deletedAt: null }, + }); + expect(live).toHaveLength(1); + expect(live[0].value).toBe(1500); + expect(live[0].externalId).toBe(STEP_TOTAL_ID); + expect(live[0].source).toBe("MANUAL"); + }); + it("is idempotent — a second run is a no-op", async () => { const prisma = getPrismaClient(); await prisma.measurement.createMany({ diff --git a/tests/integration/step-consolidation-repair.test.ts b/tests/integration/step-consolidation-repair.test.ts new file mode 100644 index 000000000..eb29b63a5 --- /dev/null +++ b/tests/integration/step-consolidation-repair.test.ts @@ -0,0 +1,210 @@ +/** + * v1.28.37 — real-Postgres integration coverage for the step-consolidation + * repair (`runStepConsolidationRepairForUser`). Mirrors the + * `consolidate-legacy-steps` fixture: the testcontainer is hot before this + * file loads, so each case truncates + seeds + repairs. + * + * Pins the contracts the unit mocks can't: + * - tombstoned GOOGLE_HEALTH + FITBIT `stats:steps:` rows within the + * retention horizon are resurrected (both providers, parity); + * - a tombstone older than the 75-day horizon is left alone; + * - the shadow MANUAL `stats:HK…:` total is removed only where a live + * provider row for the same day returns; + * - a re-run converges to zero (no double-resurrect / no duplicate). + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { runStepConsolidationRepairForUser } from "@/lib/jobs/step-consolidation-repair"; +import { TOMBSTONE_RETENTION_DAYS } from "@/lib/auth/native-client"; + +const TEST_USER_ID = "user-step-repair"; +const DAY_MS = 24 * 60 * 60 * 1000; + +/** A tombstone timestamp `daysAgo` days in the past. */ +function tombstonedAt(daysAgo: number): Date { + return new Date(Date.now() - daysAgo * DAY_MS); +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + await getPrismaClient().user.create({ + data: { + id: TEST_USER_ID, + username: "step-repair", + email: "step-repair@example.test", + timezone: "Europe/Berlin", + }, + }); +}); + +describe("runStepConsolidationRepairForUser (real Postgres)", () => { + it("resurrects tombstoned Google Health AND Fitbit daily totals within the horizon", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 8123, + unit: "steps", + source: "GOOGLE_HEALTH", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + externalId: "stats:steps:2026-05-16", + deletedAt: tombstonedAt(3), + }, + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 9456, + unit: "steps", + source: "FITBIT", + measuredAt: new Date("2026-05-17T12:00:00.000Z"), + externalId: "stats:steps:2026-05-17", + deletedAt: tombstonedAt(3), + }, + ], + }); + + const summary = await runStepConsolidationRepairForUser(TEST_USER_ID); + + expect(summary.rowsResurrected).toBe(2); + expect(summary.wedgeSkipped).toBe(0); + expect(summary.failures).toBe(0); + + const live = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS", deletedAt: null }, + orderBy: { measuredAt: "asc" }, + }); + expect(live).toHaveLength(2); + expect(live.map((r) => r.source)).toEqual(["GOOGLE_HEALTH", "FITBIT"]); + expect(live.map((r) => r.value)).toEqual([8123, 9456]); + }); + + it("does not resurrect a tombstone older than the retention horizon", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 5000, + unit: "steps", + source: "GOOGLE_HEALTH", + measuredAt: new Date("2026-01-01T12:00:00.000Z"), + externalId: "stats:steps:2026-01-01", + deletedAt: tombstonedAt(TOMBSTONE_RETENTION_DAYS + 10), + }, + }); + + const summary = await runStepConsolidationRepairForUser(TEST_USER_ID); + expect(summary.rowsResurrected).toBe(0); + + const live = await prisma.measurement.count({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS", deletedAt: null }, + }); + expect(live).toBe(0); + }); + + it("removes the shadow MANUAL mint where a live provider row returns, and leaves it where none does", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.createMany({ + data: [ + // Day A: tombstoned Google row + shadow MANUAL mint → resurrect + remove mint. + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 8000, + unit: "steps", + source: "GOOGLE_HEALTH", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + externalId: "stats:steps:2026-05-16", + deletedAt: tombstonedAt(2), + }, + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 8000, + unit: "steps", + source: "MANUAL", + measuredAt: new Date("2026-05-16T11:00:00.000Z"), + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-16", + }, + // Day B: a MANUAL mint with NO provider row → genuine Apple-legacy, keep it. + { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 4200, + unit: "steps", + source: "MANUAL", + measuredAt: new Date("2026-05-18T11:00:00.000Z"), + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-18", + }, + ], + }); + + const summary = await runStepConsolidationRepairForUser(TEST_USER_ID); + + expect(summary.rowsResurrected).toBe(1); + expect(summary.manualMintsRemoved).toBe(1); + + // Day A: Google live, MANUAL mint tombstoned. + const dayA = await prisma.measurement.findMany({ + where: { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + externalId: { + in: [ + "stats:steps:2026-05-16", + "stats:HKQuantityTypeIdentifierStepCount:2026-05-16", + ], + }, + }, + }); + const googleA = dayA.find((r) => r.source === "GOOGLE_HEALTH"); + const manualA = dayA.find((r) => r.source === "MANUAL"); + expect(googleA?.deletedAt).toBeNull(); + expect(manualA?.deletedAt).not.toBeNull(); + + // Day B: the standalone MANUAL mint is untouched (still live). + const manualB = await prisma.measurement.findFirst({ + where: { + userId: TEST_USER_ID, + source: "MANUAL", + externalId: "stats:HKQuantityTypeIdentifierStepCount:2026-05-18", + }, + }); + expect(manualB?.deletedAt).toBeNull(); + }); + + it("converges — a second run resurrects nothing and does not double-touch", async () => { + const prisma = getPrismaClient(); + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "ACTIVITY_STEPS", + value: 6000, + unit: "steps", + source: "GOOGLE_HEALTH", + measuredAt: new Date("2026-05-16T12:00:00.000Z"), + externalId: "stats:steps:2026-05-16", + deletedAt: tombstonedAt(1), + }, + }); + + const first = await runStepConsolidationRepairForUser(TEST_USER_ID); + expect(first.rowsResurrected).toBe(1); + + const second = await runStepConsolidationRepairForUser(TEST_USER_ID); + expect(second.rowsResurrected).toBe(0); + expect(second.wedgeSkipped).toBe(0); + expect(second.manualMintsRemoved).toBe(0); + + // Exactly one live row — no duplicate minted by the resurrect path. + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "ACTIVITY_STEPS" }, + }); + expect(rows).toHaveLength(1); + expect(rows[0].deletedAt).toBeNull(); + expect(rows[0].value).toBe(6000); + }); +});