diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index cda75930f..c6e430dee 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -643,6 +643,12 @@ async function postBatch(request: NextRequest): Promise { let insertedCount = 0; let updatedCount = 0; + /** + * `stats:` re-posts reported `updated` with nothing to write (see + * `isUnchangedStatsRow`). Counted in the response's `updated` like before, + * and in nothing that starts work. + */ + let unchangedStatsCount = 0; let duplicateCount = 0; let failedCount = 0; // v1.44 — subset of `insertedCount` written by the set-based insert. @@ -846,6 +852,105 @@ async function postBatch(request: NextRequest): Promise { ); } + // An unchanged `stats:` re-post. The phone re-sends the same daily + // totals on every sync, and each used to go through the reconciler + // (about ten round trips in its own transaction) only to write the same + // values back, and then counted as `updated`, which started the whole + // tail: the DAY refold, rollup jobs, PR detection and a cache stale for + // the account. Such a row is still reported `updated`, exactly as + // before, but nothing is written and it starts nothing. + // + // "Unchanged" is the reconciler's own update with nothing to change: a + // live row holds the entry's external key AND its natural identity + // (both indexes are unique, so that row is the only candidate), already + // carries the HealthKit-statistics provenance, and every column the + // update would write already has the value it would write. Like the + // sample prefilter above, it applies only where neither key occurs a + // second time in the batch, and never to a value the plausibility gate + // would refuse. A concurrent writer is no different from one that ran a + // moment later: its write lands after this no-op either way. + const statsExternalIds = [ + ...new Set( + prepared + .map((p) => p.row.externalId as string) + .filter((externalId) => isStatsExternalId(externalId)), + ), + ]; + const storedStatsByExternalKey = new Map< + string, + Record & { measuredAt: Date } + >(); + if (statsExternalIds.length > 0) { + const stored = await prisma.measurement.findMany({ + where: { + userId: user.id, + externalId: { in: statsExternalIds }, + deletedAt: null, + }, + select: { + type: true, + source: true, + externalId: true, + measuredAt: true, + sleepStage: true, + value: true, + valueMin: true, + valueMax: true, + unit: true, + externalSourceVersion: true, + glucoseContext: true, + rhythmClassification: true, + deviceType: true, + aggregationProvenance: true, + }, + }); + for (const row of stored) { + storedStatsByExternalKey.set( + `${row.type}::${row.source}::${row.externalId}`, + row, + ); + } + } + function isUnchangedStatsRow(p: Prepared): boolean { + const externalKey = `${p.row.type}::${p.row.source}::${p.row.externalId}`; + const stored = storedStatsByExternalKey.get(externalKey); + if (!stored) return false; + const measuredAt = p.row.measuredAt as Date; + const naturalKey = naturalKeyOf( + p.row.type, + p.row.source, + measuredAt, + p.row.sleepStage, + ); + if ( + !(stored.measuredAt instanceof Date) || + stored.measuredAt.getTime() !== measuredAt.getTime() || + externalKeyCounts.get(externalKey) !== 1 || + naturalKeyCounts.get(naturalKey) !== 1 || + stored.aggregationProvenance !== "HEALTHKIT_STATISTICS" || + classifyRangeRejection( + p.row.type as MeasurementType, + p.row.value as number, + ) !== null + ) { + return false; + } + const desired = desiredFor(p); + const same = (column: keyof typeof desired) => + (stored[column] ?? null) === (desired[column] ?? null); + return ( + same("value") && + same("valueMin") && + same("valueMax") && + same("unit") && + same("sleepStage") && + same("externalSourceVersion") && + same("glucoseContext") && + same("rhythmClassification") && + same("deviceType") + ); + } + // v1.42 — the `folded_window` guard (`folded-window.ts`). A raw Apple // Health sample of a dense or mean type, old enough to have been folded, // whose hour or day a live `stats:` row already covers, is a duplicate: @@ -1063,6 +1168,13 @@ async function postBatch(request: NextRequest): Promise { continue; } + if (statsRow && isUnchangedStatsRow(p)) { + results[p.index] = { index: p.index, status: "updated" }; + unchangedStatsCount++; + pushMergeCandidate(p); + continue; + } + if (!statsRow && hasSameReadingSibling(p)) { results[p.index] = { index: p.index, @@ -1311,7 +1423,7 @@ async function postBatch(request: NextRequest): Promise { details: { processed: entries.length, inserted: insertedCount, - updated: updatedCount, + updated: updatedCount + unchangedStatsCount, duplicates: duplicateCount, skipped: skipped.length, failed: failedCount, @@ -1449,7 +1561,8 @@ async function postBatch(request: NextRequest): Promise { meta: { processed: entries.length, inserted: insertedCount, - updated: updatedCount, + updated: updatedCount + unchangedStatsCount, + stats_unchanged: unchangedStatsCount, duplicates: duplicateCount, skipped: skipped.length, // v1.32.1 (issue #585) — per-reason skip breakdown; see the @@ -1563,7 +1676,7 @@ async function postBatch(request: NextRequest): Promise { const response = apiSuccess({ processed: entries.length, inserted: insertedCount + hourInsertedCount, - updated: updatedCount + hourUpdatedCount, + updated: updatedCount + unchangedStatsCount + hourUpdatedCount, duplicates: duplicateCount, failed: failedCount, skipped, diff --git a/tests/integration/routes/measurements-batch-unchanged-stats.test.ts b/tests/integration/routes/measurements-batch-unchanged-stats.test.ts new file mode 100644 index 000000000..d4a0b6cd2 --- /dev/null +++ b/tests/integration/routes/measurements-batch-unchanged-stats.test.ts @@ -0,0 +1,189 @@ +/** + * `POST /api/measurements/batch`: a `stats:` row re-posted with nothing + * changed, through the real API kit against the real database. + * + * The phone re-sends the same daily totals on every sync. Such a re-post: + * - answers exactly what it answered before: per entry `updated`, and the + * same `updated` count in the summary; + * - writes nothing (the stored row keeps its `syncVersion` and + * `updatedAt`), and starts nothing: no rollup refold, no cache + * invalidation, no PR detection; + * - goes the old way as soon as one column differs (the value, the unit, + * the device), and for a tombstoned row (a resurrection), so those are + * written and start the tail as before; + * - costs a handful of statements for a batch of them, instead of a + * reconciler transaction each. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { POST } from "@/app/api/measurements/batch/route"; + +import { callRoute } from "../../kit/call-route"; +import { + makeSession, + makeUser, + type SessionFixture, +} from "../../kit/factories"; +import { countStatements } from "../query-counter"; +import { getPrismaClient, truncateAllTables } from "../setup"; + +const tail = vi.hoisted(() => ({ rollups: 0, invalidations: 0, prs: 0 })); +vi.mock("@/lib/rollups/after-measurement-mutation", async () => { + const actual = await vi.importActual< + typeof import("@/lib/rollups/after-measurement-mutation") + >("@/lib/rollups/after-measurement-mutation"); + return { + ...actual, + afterMeasurementMutation: ( + ...args: Parameters + ) => { + tail.rollups += 1; + return actual.afterMeasurementMutation(...args); + }, + }; +}); +vi.mock("@/lib/cache/invalidate", async () => { + const actual = await vi.importActual( + "@/lib/cache/invalidate", + ); + return { + ...actual, + invalidateUserMeasurements: ( + ...args: Parameters + ) => { + tail.invalidations += 1; + return actual.invalidateUserMeasurements(...args); + }, + }; +}); +vi.mock("@/lib/jobs/pr-detection", async () => { + const actual = await vi.importActual< + typeof import("@/lib/jobs/pr-detection") + >("@/lib/jobs/pr-detection"); + return { + ...actual, + enqueuePrDetection: async () => { + tail.prs += 1; + }, + }; +}); + +const prisma = getPrismaClient(); +const ROUTE = "/api/measurements/batch"; +const DAYS = ["2026-10-01", "2026-10-02", "2026-10-03", "2026-10-04"]; + +function stepsEntry(day: string, value: number, deviceType = "watch") { + return { + hkIdentifier: "HKQuantityTypeIdentifierStepCount", + value, + unit: "count", + startDate: `${day}T00:00:00.000Z`, + endDate: `${day}T23:59:59.000Z`, + externalId: `stats:HKQuantityTypeIdentifierStepCount:${day}`, + deviceType, + }; +} + +let auth: SessionFixture; +let userId: string; + +async function post(entries: unknown[]) { + return callRoute<{ + updated: number; + inserted: number; + entries: Array<{ status: string }>; + }>(POST, { method: "POST", route: ROUTE, auth, body: { entries } }); +} + +function resetTail() { + tail.rollups = 0; + tail.invalidations = 0; + tail.prs = 0; +} + +beforeEach(async () => { + await truncateAllTables(prisma); + const user = await makeUser({ timezone: "UTC" }); + userId = user.id; + auth = await makeSession(user); + const first = await post(DAYS.map((d, i) => stepsEntry(d, 1000 + i))); + expect(first.status).toBe(200); + expect(first.data.inserted).toBe(DAYS.length); + resetTail(); +}); + +async function storedVersions() { + const rows = await prisma.measurement.findMany({ + where: { userId, externalId: { startsWith: "stats:" } }, + orderBy: { externalId: "asc" }, + select: { syncVersion: true, updatedAt: true, value: true }, + }); + return rows; +} + +describe("an unchanged stats: re-post", () => { + it("answers `updated` as before, writes nothing and starts nothing", async () => { + const before = await storedVersions(); + const measured = await countStatements(() => + post(DAYS.map((d, i) => stepsEntry(d, 1000 + i))), + ); + const res = measured.result; + expect(res.status).toBe(200); + expect(res.data.updated).toBe(DAYS.length); + expect(res.data.inserted).toBe(0); + expect(res.data.entries.map((e) => e.status)).toEqual( + DAYS.map(() => "updated"), + ); + expect(await storedVersions()).toEqual(before); + expect(tail).toEqual({ rollups: 0, invalidations: 0, prs: 0 }); + // Four reconciler transactions used to be about forty statements. + expect(measured.count).toBeLessThanOrEqual(12); + }); + + it("writes and starts the tail when one column differs", async () => { + const before = (await storedVersions()).map((r) => r.syncVersion); + const res = await post([ + stepsEntry(DAYS[0]!, 1000), // unchanged + stepsEntry(DAYS[1]!, 5555), // new value + stepsEntry(DAYS[2]!, 1002, "phone"), // new device + ]); + expect(res.data.entries.map((e) => e.status)).toEqual([ + "updated", + "updated", + "updated", + ]); + expect(res.data.updated).toBe(3); + const after = await storedVersions(); + expect(after.map((r) => r.syncVersion)).toEqual([ + before[0], + before[1]! + 1, + before[2]! + 1, + before[3], + ]); + expect(after[1]!.value).toBe(5555); + expect(tail.rollups).toBe(1); + expect(tail.invalidations).toBeGreaterThanOrEqual(1); + expect(tail.prs).toBe(1); + }); + + it("resurrects a tombstoned row the old way", async () => { + await prisma.measurement.updateMany({ + where: { + userId, + externalId: `stats:HKQuantityTypeIdentifierStepCount:${DAYS[3]}`, + }, + data: { deletedAt: new Date() }, + }); + const res = await post([stepsEntry(DAYS[3]!, 1003)]); + expect(res.data.entries[0]!.status).toBe("updated"); + const row = await prisma.measurement.findFirst({ + where: { + userId, + externalId: `stats:HKQuantityTypeIdentifierStepCount:${DAYS[3]}`, + }, + select: { deletedAt: true }, + }); + expect(row?.deletedAt).toBeNull(); + expect(tail.rollups).toBe(1); + }); +});