diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index c610a4825..257649e72 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -41,7 +41,10 @@ import { type ImportJobProgress, type ImportJobResult, } from "@/lib/measurements/import-apple-health-export"; -import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { + recomputeUserRollups, + ROLLUP_FOLD_WINDOW_MS, +} from "@/lib/rollups/measurement-rollups"; import { withBackgroundEvent } from "@/lib/logging/background"; /** @@ -502,19 +505,23 @@ export async function handleAppleHealthImport( // contradict the row the status poll surfaces to the operator. let rollupFailed = false; try { - const span = await prisma.measurement.aggregate({ - where: { userId }, - _min: { measuredAt: true }, - _max: { measuredAt: true }, - }); - if (span._min.measuredAt && span._max.measuredAt) { - // Add a small tail buffer to `to` so the upper bound is - // exclusive-safe under the rollup aggregator's `< to` filter. - const to = new Date(span._max.measuredAt.getTime() + 1); - await recomputeUserRollups(userId, { - from: span._min.measuredAt, - to, - }); + // Folded over what this import wrote, not over the account's whole + // history, and never further back than the fold window every other + // path keeps to: buckets older than `ROLLUP_FOLD_WINDOW_MS` are by + // design never written, so reads there fall back to live SQL. The old + // span ran from the account's first measurement, whatever the import + // carried, and folded any history older than the window as well. + const span = result.measuredSpan; + if (span) { + const windowStart = new Date(Date.now() - ROLLUP_FOLD_WINDOW_MS); + const spanFrom = new Date(span.from); + const from = spanFrom > windowStart ? spanFrom : windowStart; + // `to` is exclusive in the aggregator, so one millisecond past the + // newest row. + const to = new Date(new Date(span.to).getTime() + 1); + if (to > from) { + await recomputeUserRollups(userId, { from, to }); + } } } catch (rollupErr) { // Rollup failure is non-fatal โ€” the next read falls through to diff --git a/src/lib/measurements/import-apple-health-export.ts b/src/lib/measurements/import-apple-health-export.ts index 9bc4fe8cf..d175fc5c0 100644 --- a/src/lib/measurements/import-apple-health-export.ts +++ b/src/lib/measurements/import-apple-health-export.ts @@ -125,6 +125,11 @@ export interface ImportJobResult { /** Estimated `(measurement type, local day)` aggregate rows considered. */ rows: number; }; + /** + * Oldest and newest instant this import wrote a measurement at (ISO), or + * null when it wrote none. What the rollup fold after the import covers. + */ + measuredSpan: { from: string; to: string } | null; /** Bounded auxiliary ECG outcomes. Never contains filenames or waveform data. */ ecg: { discovered: number; @@ -396,6 +401,14 @@ export async function streamParseExportXml( let rowsUpserted = 0; const cumulativeEstimatedDays = new Set(); let cumulativeEstimatedRows = 0; + // The instants this import wrote measurements at, oldest and newest. The + // worker folds the rollup table over exactly this span afterwards. + let spanFrom: Date | null = null; + let spanTo: Date | null = null; + const widenSpan = (at: Date) => { + if (spanFrom === null || at < spanFrom) spanFrom = at; + if (spanTo === null || at > spanTo) spanTo = at; + }; // Per R-1 ยง8 the percent stays best-effort and may remain null // until the parser sees the closing `` tag. We don't // currently mutate this in v1.4.34 โ€” the iOS app keeps polling on @@ -438,6 +451,7 @@ export async function streamParseExportXml( const flushSpotBatch = async (): Promise => { if (spotBatch.length === 0) return; const chunk = spotBatch.splice(0, spotBatch.length); + for (const row of chunk) widenSpan(row.measuredAt); const insertedArrivals: Array<{ id: string; type: MeasurementType; @@ -731,6 +745,7 @@ export async function streamParseExportXml( } const externalId = dailyStatsExternalId(mapping.hkIdentifier, dayKey); const measuredAt = canonicalDailyTimestamp(dayKey, userTimezone); + widenSpan(measuredAt); const rowStart = Date.now(); const verdict = await prisma.$transaction((tx) => reconcileExternalMeasurement( @@ -1202,6 +1217,13 @@ export async function streamParseExportXml( days: cumulativeEstimatedDays.size, rows: cumulativeEstimatedRows, }, + measuredSpan: + spanFrom !== null && spanTo !== null + ? { + from: (spanFrom as Date).toISOString(), + to: (spanTo as Date).toISOString(), + } + : null, ecg: { discovered: 0, imported: 0, diff --git a/tests/integration/apple-health-import-rollup-window.test.ts b/tests/integration/apple-health-import-rollup-window.test.ts new file mode 100644 index 000000000..be558605e --- /dev/null +++ b/tests/integration/apple-health-import-rollup-window.test.ts @@ -0,0 +1,116 @@ +/** + * #1031 โ€” the rollup fold after an import covers what the import wrote. + * + * The worker folded the rollup table from the account's first measurement to + * its last, whatever the import carried, and so also folded history older + * than the five-year window every other fold path keeps to. It now folds the + * span of the rows this import wrote, clamped to that window. + */ +import { createHash } from "node:crypto"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { strToU8, zipSync } from "fflate"; +import type { Job } from "pg-boss"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { + APPLE_HEALTH_IMPORT_PARSER_REVISION, + handleAppleHealthImport, + _setWorkerPrismaForTests, + type AppleHealthImportPayload, +} from "@/lib/jobs/apple-health-import-worker"; +import { ROLLUP_FOLD_WINDOW_MS } from "@/lib/rollups/measurement-rollups"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +const USER = "user-import-rollup-window"; + +function record(at: Date, value: number): string { + const stamp = at.toISOString().replace("T", " ").slice(0, 19) + " +0000"; + return ``; +} + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); + _setWorkerPrismaForTests(getPrismaClient()); + await getPrismaClient().user.create({ + data: { + id: USER, + username: USER, + email: `${USER}@example.test`, + timezone: "UTC", + }, + }); +}); + +describe("import rollup fold window", () => { + it("folds the imported span, never past the fold window, and leaves older history alone", async () => { + const prisma = getPrismaClient(); + const now = Date.now(); + const day = 86_400_000; + // Already in the account before the import: a reading eight years old. + await prisma.measurement.create({ + data: { + userId: USER, + type: "WEIGHT", + value: 90, + unit: "kg", + source: "MANUAL", + measuredAt: new Date(now - 8 * 365 * day), + }, + }); + // The import: one reading seven years old, a few recent ones. + const xml = ` + + ${record(new Date(now - 7 * 365 * day), 88)} + ${record(new Date(now - 40 * day), 84)} + ${record(new Date(now - 30 * day), 83)} + ${record(new Date(now - 20 * day), 82)} +`; + const dir = mkdtempSync(join(tmpdir(), "hl-import-window-")); + const path = join(dir, "export.zip"); + const bytes = Buffer.from( + zipSync({ "apple_health_export/export.xml": strToU8(xml) }), + ); + writeFileSync(path, bytes); + await prisma.importJob.create({ + data: { + userId: USER, + pgBossJobId: "window-job", + status: "queued", + uploadBytes: bytes.length, + uploadSha256: createHash("sha256").update(bytes).digest("hex"), + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + }, + }); + + const outcome = await handleAppleHealthImport({ + id: "window-job", + data: { + userId: USER, + uploadPath: path, + uploadBytes: bytes.length, + enqueuedAt: new Date().toISOString(), + }, + } as unknown as Job); + expect(outcome.ok).toBe(true); + + const buckets = await prisma.measurementRollup.findMany({ + where: { userId: USER, type: "WEIGHT", granularity: "DAY" }, + orderBy: { bucketStart: "asc" }, + }); + const windowStart = now - ROLLUP_FOLD_WINDOW_MS; + expect(buckets.length).toBeGreaterThanOrEqual(3); + for (const bucket of buckets) { + expect(bucket.bucketStart.getTime()).toBeGreaterThan(windowStart - day); + } + }); +});