diff --git a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts index 8389b3bc9..11772072c 100644 --- a/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts +++ b/src/lib/measurements/__tests__/drain-per-sample-cumulative.test.ts @@ -324,3 +324,95 @@ describe("drainPerSampleCumulative — cutoffHours", () => { expect(call.where.measuredAt).toBeUndefined(); }); }); + +// A1 — a late watch sync after a day's total was already collapsed must +// FOLD the new per-sample rows into the existing total, never overwrite +// it with just the partial late sum (the original samples are gone). +describe("drainPerSampleCumulative — late-sync fold into existing total", () => { + function buildFoldMock(existingValue: number | null) { + const findManyUser = vi.fn().mockResolvedValue([ + { id: "user-1", timezone: "Europe/Berlin" }, + ]); + + // Only ACTIVITY_STEPS yields the late per-sample rows; every other + // cumulative type returns an empty scan so the run stays a single + // bucket. + const findManyMeasurement = vi.fn( + async (args: { where: { type: string } }) => { + if (args.where.type === "ACTIVITY_STEPS") { + return [ + { + id: "late-1", + type: "ACTIVITY_STEPS", + value: 300, + measuredAt: new Date("2026-05-16T20:00:00.000Z"), + externalId: "hk-uuid-late-1", + }, + { + id: "late-2", + type: "ACTIVITY_STEPS", + value: 200, + measuredAt: new Date("2026-05-16T21:00:00.000Z"), + externalId: "hk-uuid-late-2", + }, + ]; + } + return []; + }, + ); + + const upsert = vi.fn().mockResolvedValue({}); + const deleteMany = vi.fn().mockResolvedValue({ count: 2 }); + const findUnique = vi.fn().mockResolvedValue( + existingValue === null ? null : { value: existingValue }, + ); + const findFirst = vi.fn().mockResolvedValue({ unit: "count" }); + + const tx = { + measurement: { upsert, deleteMany, findUnique, findFirst }, + }; + + return { + prisma: { + user: { findMany: findManyUser }, + measurement: { findMany: findManyMeasurement }, + $transaction: vi.fn(async (cb: (t: typeof tx) => Promise) => + cb(tx), + ), + } as unknown as PrismaClient, + upsert, + findUnique, + }; + } + + it("folds late samples into a pre-existing collapsed total (no shrink)", async () => { + // Day already collapsed to 9000 on an earlier run; a late sync of + // 300 + 200 = 500 must take the row to 9500, not down to 500. + const { prisma, upsert, findUnique } = buildFoldMock(9000); + + await drainPerSampleCumulative(prisma, { log: () => {} }); + + expect(findUnique).toHaveBeenCalledTimes(1); + expect(upsert).toHaveBeenCalledTimes(1); + const upsertArg = upsert.mock.calls[0]?.[0] as { + update: { value: number }; + }; + expect(upsertArg.update.value).toBe(9500); + }); + + it("mints the partial sum when no collapsed total exists yet", async () => { + const { prisma, upsert } = buildFoldMock(null); + + await drainPerSampleCumulative(prisma, { log: () => {} }); + + const upsertArg = upsert.mock.calls[0]?.[0] as { + create: { value: number }; + update: { value: number }; + }; + // Fresh day: create branch carries the raw sum; update branch (taken + // on an idempotent re-run with no prior total) also resolves to the + // raw sum since `existing` was null. + expect(upsertArg.create.value).toBe(500); + expect(upsertArg.update.value).toBe(500); + }); +}); diff --git a/src/lib/measurements/drain-per-sample-cumulative.ts b/src/lib/measurements/drain-per-sample-cumulative.ts index 9b8202e31..e77ba3e6d 100644 --- a/src/lib/measurements/drain-per-sample-cumulative.ts +++ b/src/lib/measurements/drain-per-sample-cumulative.ts @@ -228,6 +228,40 @@ export async function drainPerSampleCumulative( }): Promise => { let removed = 0; await pc.$transaction(async (tx) => { + // Fold late per-sample rows INTO an existing collapsed daily + // total rather than overwriting it. `bucketRowsByDay` strips + // already-collapsed `stats:` rows from the bucket, so when a day + // was collapsed on an earlier run and fresh per-sample rows + // arrive late (a trailing watch sync after the day's total was + // already minted), this bucket holds ONLY those late samples. + // A blind `update: { value: reducedValue }` would replace the + // full day total with just the partial late sum — and the + // original contributing samples were already hard-deleted, so + // the lost count is unrecoverable. + // + // The probe runs inside the same transaction as the upsert + + // delete so the read-modify-write of the total is atomic against + // a concurrent drain. The late samples are genuine additional + // readings (unlike the legacy-step pass, where the existing + // total is HealthKit's own aggregate and the legacy rows would + // double-count — so that sibling deliberately does NOT fold), + // therefore the correct merge is `existing + late sum`. The + // per-sample rows are hard-deleted either way so a re-run cannot + // re-accumulate them. + const existing = await tx.measurement.findUnique({ + where: { + userId_type_source_externalId: { + userId, + type, + source: "APPLE_HEALTH", + externalId, + }, + }, + select: { value: true }, + }); + const mergedValue = + existing !== null ? existing.value + reducedValue : reducedValue; + // Upsert the daily-aggregated row first, then drop the // per-sample rows. The unique index // (userId, type, source, externalId) makes the upsert @@ -257,7 +291,7 @@ export async function drainPerSampleCumulative( externalId, }, update: { - value: reducedValue, + value: mergedValue, measuredAt: canonicalTimestamp, }, });