mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 04:32:42 +00:00
Merge pull request #494 from MBombeck/release/v1.28.37
release/v1.28.37 — provider step totals survive a restart
This commit is contained in:
12 files changed
+1007
-24
No files matched your search
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
+1
-1
@@ -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",
|
||||
|
||||
+1
-1
@@ -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/*`
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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<StepConsolidationRepairPayload>(
|
||||
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<void> {
|
||||
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,
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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:<day>`) 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…:<day>` total the bug minted, but
|
||||
* ONLY for a day where a live GOOGLE_HEALTH / FITBIT `stats:steps:<day>`
|
||||
* 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:<tag>:<day>` 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<StepConsolidationRepairSummary> {
|
||||
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<string, Date>();
|
||||
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<Array<{ id: string }>>`
|
||||
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),
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -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:<HK>:<YYYY-MM-DD>`. 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<Array<{ id: string }>>`
|
||||
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:%'
|
||||
)
|
||||
`;
|
||||
|
||||
|
||||
@@ -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:<day>)", () => {
|
||||
// Regression: Google + Fitbit daily steps carry `stats:steps:<day>`,
|
||||
// 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:<day> 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(
|
||||
|
||||
@@ -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:<day>`, and any future
|
||||
* stats-contract provider carries `stats:<tag>:<day>`. 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:<day>`, 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<MeasurementSource> = 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<string, PerSampleRow[]> {
|
||||
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;
|
||||
|
||||
@@ -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:<day>`, 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({
|
||||
|
||||
@@ -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:<day>` 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…:<day>` 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);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user