From 8cdb53cb0e144411084c43d0f9abda755c75b9f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 20:13:13 +0200 Subject: [PATCH] perf(withings): write a measure sync in sets and refold only what changed A manual full sync re-reads the whole history, and nearly all of it is already stored. Each reading cost a findFirst plus an update, and every re-read day was marked dirty, so the run refolded the DAY tier and queued three coarse recomputes for every day of the account: about thirty seconds of database work behind 1.4 s of Withings calls. The measures leg now reads the stored rows of the window once, skips the unchanged ones, updates the changed or tombstoned ones and inserts the new ones in batches. Only inserted or changed readings count as touched. recomputeBucketsForKeys refolds exactly the touched DAY buckets in one aggregate and one transaction, and queues one coarse job per run of adjacent buckets instead of three per day. --- src/lib/rollups/measurement-rollups.ts | 170 ++++++++++++++- .../__tests__/sync-measure-watermark.test.ts | 49 +++-- src/lib/withings/__tests__/sync-plan.test.ts | 31 +-- .../__tests__/sync-plausibility.test.ts | 42 ++-- src/lib/withings/sync.ts | 171 +++++++++------ tests/integration/measurement-rollups.test.ts | 164 ++++++++++++++ .../integration/withings-sync-volume.test.ts | 204 ++++++++++++++++++ 7 files changed, 725 insertions(+), 106 deletions(-) create mode 100644 tests/integration/withings-sync-volume.test.ts diff --git a/src/lib/rollups/measurement-rollups.ts b/src/lib/rollups/measurement-rollups.ts index cf33a2c6d..bfaf64b92 100644 --- a/src/lib/rollups/measurement-rollups.ts +++ b/src/lib/rollups/measurement-rollups.ts @@ -358,6 +358,143 @@ export async function recomputeBucketsForMeasurement( ); } +/** Keys one set-based DAY refold carries; bounds the statement's arrays. */ +const KEYED_REFOLD_CHUNK = 2_000; + +/** + * The set-based sibling of {@link recomputeBucketsForMeasurement}, for a write + * that touched many `(type, day)` buckets at once (a sync, an import, a batch). + * + * The per-key hook costs an aggregate, a transaction and three queue sends per + * bucket, so a sync that re-read two years of history paid that seven thousand + * times over. Here the DAY tier is refolded for exactly the touched buckets in + * one aggregate and one transaction per chunk, and the coarser tiers are + * enqueued once per contiguous run of buckets rather than once per day. The + * result is the one the per-key loop produced: every touched day re-derived + * from its live rows, a day left empty dropped, untouched days left alone. + */ +export async function recomputeBucketsForKeys( + userId: string, + rows: Array<{ type: MeasurementType; measuredAt: Date }>, +): Promise { + const keys = collapseToTypeDayKeys(rows); + for (let i = 0; i < keys.length; i += KEYED_REFOLD_CHUNK) { + await refoldDayBuckets(userId, keys.slice(i, i + KEYED_REFOLD_CHUNK)); + } + for (const granularity of ASYNC_GRANULARITIES) { + for (const run of bucketRuns(keys, granularity)) { + await enqueueRollupRecompute({ userId, granularity, ...run }); + } + } +} + +async function refoldDayBuckets( + userId: string, + keys: Array<{ type: MeasurementType; measuredAt: Date }>, +): Promise { + if (keys.length === 0) return; + const types = [...new Set(keys.map((k) => k.type))]; + const days = [...new Set(keys.map((k) => k.measuredAt.getTime()))].sort( + (a, b) => a - b, + ); + const wanted = new Set( + keys.map((k) => `${k.type}|${k.measuredAt.getTime()}`), + ); + const aggregate = await runRollupAggregate({ + userId, + types, + granularity: "DAY", + from: new Date(days[0]!), + to: new Date(days[days.length - 1]! + 24 * 60 * 60 * 1000), + days: days.map((d) => new Date(d)), + }); + // The day filter is per day, not per (type, day): another touched type's + // day comes back too. Its aggregate would be right, but it was not touched, + // so it is not rewritten. + const ordered = aggregate + .filter((r) => + wanted.has(`${r.type}|${new Date(r.bucket_start).getTime()}`), + ) + .sort((a, b) => + a.type !== b.type + ? a.type < b.type + ? -1 + : 1 + : a.source !== b.source + ? a.source < b.source + ? -1 + : 1 + : a.bucket_start.getTime() - b.bucket_start.getTime(), + ); + // Delete exactly the touched buckets (every source, so a source that left a + // day does not strand its row, and a day that emptied out is dropped), then + // upsert the fresh aggregate, in one transaction like `persistRollupRows`. + await prisma.$transaction( + async (tx) => { + await tx.$executeRawUnsafe( + `DELETE FROM measurement_rollups + WHERE user_id = $1 + AND granularity = 'DAY'::"measurement_rollup_granularity" + AND (type, bucket_start) IN ( + SELECT k.t::"measurement_type", k.b + FROM unnest($2::text[], $3::timestamptz[]) AS k(t, b) + )`, + userId, + keys.map((k) => k.type), + keys.map((k) => k.measuredAt.toISOString()), + ); + for (let i = 0; i < ordered.length; i += 500) { + const { sql, params } = buildRollupUpsert( + userId, + "DAY", + ordered.slice(i, i + 500), + ); + await tx.$executeRawUnsafe(sql, ...params); + } + }, + { maxWait: 10_000, timeout: 120_000 }, + ); +} + +/** + * The coarse buckets the touched days fall in, merged per type into runs of + * adjacent buckets: one recompute job per run, each covering whole buckets. + */ +function bucketRuns( + keys: Array<{ type: MeasurementType; measuredAt: Date }>, + granularity: RollupGranularity, +): Array<{ type: MeasurementType; from: Date; to: Date }> { + const spans = new Map< + string, + { type: MeasurementType; from: Date; to: Date } + >(); + for (const k of keys) { + const span = bucketSpan(k.measuredAt, granularity); + spans.set(`${k.type}|${span.from.getTime()}`, { type: k.type, ...span }); + } + const sorted = [...spans.values()].sort((a, b) => + a.type !== b.type + ? a.type < b.type + ? -1 + : 1 + : a.from.getTime() - b.from.getTime(), + ); + const runs: Array<{ type: MeasurementType; from: Date; to: Date }> = []; + for (const span of sorted) { + const last = runs[runs.length - 1]; + if ( + last && + last.type === span.type && + last.to.getTime() === span.from.getTime() + ) { + last.to = span.to; + } else { + runs.push({ ...span }); + } + } + return runs; +} + /** * Best-effort pg-boss enqueue. Silent no-op when no boss is attached * (route is running in a context without the worker — typically the @@ -399,7 +536,16 @@ export async function enqueueRollupRecompute(input: { // is still QUEUED — safe here precisely because the handler re-reads the // bucket's live rows when it starts, so the queued job still sees writes // that land after it was enqueued. - singletonKey: `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}`, + // + // A job covering a run of several buckets (`recomputeBucketsForKeys`) + // carries its end in the key too: under the bare start it would collapse + // into a queued single-bucket job and the rest of the run would be lost. + singletonKey: + `${input.userId}|${input.type}|${input.granularity}|${input.from.toISOString()}` + + (bucketSpan(input.from, input.granularity).to.getTime() === + input.to.getTime() + ? "" + : `|${input.to.toISOString()}`), }); } @@ -480,10 +626,21 @@ async function runRollupAggregate( granularity: RollupGranularity; from: Date; to: Date; + /** + * Only the readings of these UTC days, inside `[from, to)`. A keyed + * refold (`recomputeBucketsForKeys`) passes the touched days so a handful + * spread across years does not aggregate everything in between. + */ + days?: Date[]; }, db: Pick = prisma, ): Promise { if (input.types && input.types.length === 0) return []; + // Bound as the fourth parameter, never spliced. + const dayFilter = input.days + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + : ""; + const dayParams = input.days ? [input.days.map((d) => d.toISOString())] : []; // Postgres rejects `date_trunc($1::text, m."measured_at")` in a // GROUP BY because the planner sees the parameterised expression @@ -553,6 +710,7 @@ async function runRollupAggregate( AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL + ${dayFilter} GROUP BY m."type", m."source", ${dateTrunc} `; return applyHourlyMeanDay( @@ -563,6 +721,7 @@ async function runRollupAggregate( input.userId, input.from, input.to, + ...dayParams, ), ); } @@ -600,6 +759,7 @@ async function runRollupAggregate( AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL + ${dayFilter} GROUP BY m."type", m."source", ${dateTrunc} `; return applyHourlyMeanDay( @@ -610,6 +770,7 @@ async function runRollupAggregate( input.userId, input.from, input.to, + ...dayParams, ), ); } @@ -635,6 +796,7 @@ async function applyHourlyMeanDay( granularity: RollupGranularity; from: Date; to: Date; + days?: Date[]; }, rows: RollupRow[], ): Promise { @@ -685,6 +847,11 @@ async function applyHourlyMeanDay( AND m."measured_at" >= $2 AND m."measured_at" < $3 AND m."deleted_at" IS NULL + ${ + input.days + ? `AND date_trunc('day', m."measured_at") = ANY($4::timestamptz[])` + : "" + } GROUP BY m."type", m."source", date_trunc('hour', m."measured_at") ) h GROUP BY 1, 2, 3 @@ -694,6 +861,7 @@ async function applyHourlyMeanDay( input.userId, input.from, input.to, + ...(input.days ? [input.days.map((d) => d.toISOString())] : []), ); const byKey = new Map( means.map((m) => [ diff --git a/src/lib/withings/__tests__/sync-measure-watermark.test.ts b/src/lib/withings/__tests__/sync-measure-watermark.test.ts index 3dcb7ae9b..a242e89c0 100644 --- a/src/lib/withings/__tests__/sync-measure-watermark.test.ts +++ b/src/lib/withings/__tests__/sync-measure-watermark.test.ts @@ -16,8 +16,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { measurement: { - findFirst: vi.fn(), - create: vi.fn(), + findMany: vi.fn(), + createManyAndReturn: vi.fn(), update: vi.fn(), }, withingsConnection: { @@ -66,7 +66,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { >("@/lib/rollups/measurement-rollups"); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined), }; }); @@ -96,10 +96,6 @@ import { import { getValidToken, syncUserMeasurements } from "../sync"; -const P2002 = Object.assign(new Error("Unique constraint failed"), { - code: "P2002", -}); - beforeEach(() => { vi.clearAllMocks(); vi.mocked(prisma.$transaction).mockImplementation(async (run) => @@ -116,8 +112,15 @@ beforeEach(() => { lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000), } as never); vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation( + (async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((row, i) => ({ + id: `m-${i}`, + type: row.type, + measuredAt: row.measuredAt, + }))) as never, + ); vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); }); @@ -168,10 +171,20 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () measuredAt: new Date("2026-05-16T08:05:00Z"), }, ] as never); - // The first write hard-fails (a transient DB error), the second succeeds. - vi.mocked(prisma.measurement.create) - .mockRejectedValueOnce(new Error("connection reset by peer")) - .mockResolvedValue({} as never); + // The weight is stored with an older value; its update hard-fails (a + // transient DB error). The pulse is new and inserts. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + id: "m-weight", + type: "WEIGHT", + measuredAt: new Date("2026-05-16T08:00:00Z"), + value: 79, + deletedAt: null, + }, + ] as never); + vi.mocked(prisma.measurement.update).mockRejectedValueOnce( + new Error("connection reset by peer"), + ); const result = await syncUserMeasurements("user-1"); @@ -187,7 +200,7 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () ); }); - it("does NOT hold the watermark for a benign P2002 idempotent-write collision", async () => { + it("does NOT hold the watermark for a duplicate a concurrent run inserted first", async () => { vi.mocked(fetchMeasurements).mockResolvedValue([ { type: "WEIGHT", @@ -195,11 +208,15 @@ describe("syncUserMeasurements — watermark hold on hard row failure (F4)", () measuredAt: new Date("2026-05-16T08:00:00Z"), }, ] as never); - vi.mocked(prisma.measurement.create).mockRejectedValueOnce(P2002); + // A concurrent run inserted the same reading first: the batched insert + // skips it as a duplicate. + vi.mocked(prisma.measurement.createManyAndReturn).mockResolvedValueOnce( + [] as never, + ); const { imported } = await syncUserMeasurements("user-1"); - // P2002 means the row is already present — nothing lost — so the cycle is + // A skipped duplicate is already present — nothing lost — so the cycle is // still healthy: stamp the watermark + success, record no failure. expect(imported).toBe(0); expect(prisma.withingsConnection.update).toHaveBeenCalledTimes(1); diff --git a/src/lib/withings/__tests__/sync-plan.test.ts b/src/lib/withings/__tests__/sync-plan.test.ts index 89834898e..1cf80a2a5 100644 --- a/src/lib/withings/__tests__/sync-plan.test.ts +++ b/src/lib/withings/__tests__/sync-plan.test.ts @@ -10,8 +10,8 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { measurement: { - findFirst: vi.fn(), - create: vi.fn(), + findMany: vi.fn(), + createManyAndReturn: vi.fn(), update: vi.fn(), }, withingsConnection: { @@ -61,7 +61,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { >("@/lib/rollups/measurement-rollups"); return { ...actual, - recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined), }; }); @@ -179,8 +179,10 @@ function persistedLegs(): Array<{ leg: unknown; value: unknown }> { function createdTypes(): string[] { return vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0] as { data: { type: string } }).data.type); + .mocked(prisma.measurement.createManyAndReturn) + .mock.calls.flatMap((c) => + (c[0] as { data: Array<{ type: string }> }).data.map((d) => d.type), + ); } beforeEach(() => { @@ -193,14 +195,15 @@ beforeEach(() => { connection() as never, ); vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockImplementation((async (args: { - data: { type: string; measuredAt: Date }; - }) => ({ - id: `m-${args.data.type}`, - type: args.data.type, - measuredAt: args.data.measuredAt, - })) as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation( + (async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((d) => ({ + id: `m-${d.type}`, + type: d.type, + measuredAt: d.measuredAt, + }))) as never, + ); vi.mocked(prisma.$executeRaw).mockResolvedValue(1 as never); }); @@ -338,7 +341,7 @@ describe("syncUserMeasurements inside the Withings API plan", () => { for (let run = 0; run < 4; run++) { vi.mocked(prisma.$executeRaw).mockClear(); - vi.mocked(prisma.measurement.create).mockClear(); + vi.mocked(prisma.measurement.createManyAndReturn).mockClear(); requestedStartdates.length = 0; await syncUserMeasurements("user-1"); const [persisted] = persistedLegs(); diff --git a/src/lib/withings/__tests__/sync-plausibility.test.ts b/src/lib/withings/__tests__/sync-plausibility.test.ts index c980998a5..b80a6715c 100644 --- a/src/lib/withings/__tests__/sync-plausibility.test.ts +++ b/src/lib/withings/__tests__/sync-plausibility.test.ts @@ -14,10 +14,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: { measurement: { - findFirst: vi.fn(), findMany: vi.fn(), - create: vi.fn(), createMany: vi.fn(), + createManyAndReturn: vi.fn(), update: vi.fn(), }, withingsConnection: { @@ -67,6 +66,7 @@ vi.mock("@/lib/rollups/measurement-rollups", async () => { return { ...actual, recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + recomputeBucketsForKeys: vi.fn().mockResolvedValue(undefined), }; }); @@ -125,13 +125,15 @@ beforeEach(() => { lastSyncedAt: new Date(Date.now() - 60 * 60 * 1000), } as never); vi.mocked(prisma.withingsConnection.update).mockResolvedValue({} as never); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); - vi.mocked(prisma.measurement.create).mockResolvedValue({ - id: "m-1", - type: "PULSE", - measuredAt: new Date(), - } as never); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation( + (async (args: { data: Array<{ type: string; measuredAt: Date }> }) => + args.data.map((d, i) => ({ + id: `m-${i}`, + type: d.type, + measuredAt: d.measuredAt, + }))) as never, + ); vi.mocked(prisma.measurement.createMany).mockResolvedValue({ count: 1, } as never); @@ -157,7 +159,11 @@ describe("syncUserMeasurements — plausibility gate", () => { ); expect(result).toEqual({ imported: 1, failed: false }); - expect(prisma.measurement.create).toHaveBeenCalledTimes(1); + expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1); + expect( + vi.mocked(prisma.measurement.createManyAndReturn).mock.calls[0]?.[0] + ?.data, + ).toHaveLength(1); expect(tally()).toBeUndefined(); }); @@ -182,7 +188,11 @@ describe("syncUserMeasurements — plausibility gate", () => { // Only the real reading landed, and the run is still a clean success: one // impossible sample must not strand the rest behind a held watermark. expect(result).toEqual({ imported: 1, failed: false }); - expect(prisma.measurement.create).toHaveBeenCalledTimes(1); + expect(prisma.measurement.createManyAndReturn).toHaveBeenCalledTimes(1); + expect( + vi.mocked(prisma.measurement.createManyAndReturn).mock.calls[0]?.[0] + ?.data, + ).toHaveLength(1); expect(tally()).toEqual({ total: 1, buckets: [ @@ -196,9 +206,15 @@ describe("syncUserMeasurements — plausibility gate", () => { // The slot already holds a real pulse; the provider re-sends the same // instant carrying a decode slip. Ungated, the update branch rewrote the // good value with the bad one. - vi.mocked(prisma.measurement.findFirst).mockResolvedValue({ - id: "existing-row", - } as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValueOnce([ + { + id: "existing-row", + type: "PULSE", + measuredAt: new Date("2026-05-16T08:00:00Z"), + value: 61, + deletedAt: null, + }, + ] as never); vi.mocked(fetchMeasurements).mockResolvedValue([ { type: "PULSE", diff --git a/src/lib/withings/sync.ts b/src/lib/withings/sync.ts index 1c84a09b0..6bf90b248 100644 --- a/src/lib/withings/sync.ts +++ b/src/lib/withings/sync.ts @@ -29,7 +29,6 @@ import { } from "./api-plan"; import { getUserWithingsCredentials } from "./credentials"; import { annotate, getEvent } from "@/lib/logging/context"; -import { isP2002 } from "@/lib/prisma-errors"; import { emitInsertedMeasurementArrivals } from "@/lib/arrivals/measurement-emit"; import { dropImplausibleMeasurements } from "@/lib/measurements/plausibility-gate"; import { @@ -45,7 +44,7 @@ import { } from "./response-classifier"; import { collapseToTypeDayKeys, - recomputeBucketsForMeasurement, + recomputeBucketsForKeys, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { @@ -349,7 +348,8 @@ export async function syncUserMeasurements( // through Withings. We collect first, fold once at the end so a 100- // row monthly catch-up costs at most ~N (type, day) recomputes rather // than N per-row hooks. Best-effort: a populator hiccup never fails - // the user's sync. + // the user's sync. Only readings this run inserted or changed count as + // touched: a re-read reading that is already stored dirties nothing. const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; const insertedArrivals: Array<{ id: string; @@ -367,69 +367,118 @@ export async function syncUserMeasurements( let anyRowFailed = false; let firstRowError: unknown = null; + // The run reads a window (the whole history on a full sync) and most of it + // is usually stored already. One read of the stored rows in that window + // splits it three ways: unchanged (nothing to write, nothing to refold), + // changed or tombstoned (an update), and new (one batched insert). The + // per-row `findFirst` + `update` this replaced cost two round trips per + // reading and marked every re-read day dirty, which is where a full sync's + // forty seconds went. + const latest = new Map(); for (const m of measures) { - const measType = m.type as MeasurementType; - try { - // v1.4.25 — Migration 0055 added `sleepStage` to the - // composite unique with `NULLS NOT DISTINCT`, so a non-sleep - // upsert keyed on (userId, type, measuredAt, source) with - // sleepStage IS NULL is still safe. But Prisma's typed compound - // input requires a non-null `sleepStage`, so we model the - // idempotent write as a `findFirst` + `create`/`update`. The - // unique index still serializes concurrent inserts, so the worst - // case is a Prisma P2002 we catch and ignore. - const existing = await prisma.measurement.findFirst({ - where: { - userId, - type: measType, - measuredAt: m.measuredAt, - source: "WITHINGS", - sleepStage: null, + latest.set(`${m.type}|${m.measuredAt.getTime()}`, m); + } + const incoming = [...latest.values()]; + const existingByKey = new Map< + string, + { id: string; value: number; deletedAt: Date | null } + >(); + if (incoming.length > 0) { + let first = incoming[0]!.measuredAt; + let last = first; + for (const m of incoming) { + if (m.measuredAt < first) first = m.measuredAt; + if (m.measuredAt > last) last = m.measuredAt; + } + const stored = await prisma.measurement.findMany({ + where: { + userId, + source: "WITHINGS", + sleepStage: null, + type: { + in: [...new Set(incoming.map((m) => m.type as MeasurementType))], }, - select: { id: true }, + measuredAt: { gte: first, lte: last }, + }, + select: { + id: true, + type: true, + measuredAt: true, + value: true, + deletedAt: true, + }, + }); + for (const row of stored) { + existingByKey.set(`${row.type}|${row.measuredAt.getTime()}`, row); + } + } + + const toCreate: typeof incoming = []; + for (const m of incoming) { + const measType = m.type as MeasurementType; + const existing = existingByKey.get(`${m.type}|${m.measuredAt.getTime()}`); + if (!existing) { + toCreate.push(m); + continue; + } + if (existing.value === m.value && existing.deletedAt === null) { + imported++; + continue; + } + try { + await prisma.measurement.update({ + where: { id: existing.id }, + // `deletedAt: null` is a no-op on a live row, a deliberate + // RESURRECTION on a tombstoned one — Withings is the source of + // truth for its own rows, so a re-fetched measure brings the row + // back (mirrors Google / Fitbit). + data: { value: m.value, deletedAt: null }, }); - if (existing) { - await prisma.measurement.update({ - where: { id: existing.id }, - // `deletedAt: null` is a no-op on a live row, a deliberate - // RESURRECTION on a tombstoned one — Withings is the source of - // truth for its own rows, so a re-fetched measure brings the row - // back (mirrors Google / Fitbit). - data: { value: m.value, deletedAt: null }, - }); - } else { - const created = await prisma.measurement.create({ - data: { - userId, - type: measType, - value: m.value, - unit: getUnitForType(m.type), - measuredAt: m.measuredAt, - source: "WITHINGS", - }, - select: { id: true, type: true, measuredAt: true }, - }); - insertedArrivals.push(created); - } touched.push({ type: measType, measuredAt: m.measuredAt }); imported++; } catch (err) { - if (isP2002(err)) { - // Benign: the composite unique index serialised a concurrent insert of - // this exact reading — the row is present, nothing is lost. Warn and - // continue WITHOUT holding the watermark (this is the expected - // idempotent-write race the findFirst+create models around). - getEvent()?.addWarning( - `Withings measure upsert hit an idempotent P2002 (row already present): ${err}`, - ); - } else { - // A genuine write failure strands this reading. Flag the cycle so the - // watermark is held and the next scheduled tick refetches the overlap - // window and retries. - anyRowFailed = true; - if (firstRowError === null) firstRowError = err; - getEvent()?.addWarning(`Failed to upsert measure: ${err}`); + // A genuine write failure strands this reading. Flag the cycle so the + // watermark is held and the next scheduled tick refetches the overlap + // window and retries. + anyRowFailed = true; + if (firstRowError === null) firstRowError = err; + getEvent()?.addWarning(`Failed to update measure: ${err}`); + } + } + + // New readings in one statement per chunk. `skipDuplicates` absorbs the + // idempotent-write race the unique index serialises (a concurrent webhook + // run inserting the same reading): that row is present and nothing is lost, + // exactly the P2002 the per-row path caught and ignored. + for (let i = 0; i < toCreate.length; i += 1_000) { + const chunk = toCreate.slice(i, i + 1_000); + try { + const created = await prisma.measurement.createManyAndReturn({ + data: chunk.map((m) => ({ + userId, + type: m.type as MeasurementType, + value: m.value, + unit: getUnitForType(m.type), + measuredAt: m.measuredAt, + source: "WITHINGS" as const, + })), + skipDuplicates: true, + select: { id: true, type: true, measuredAt: true }, + }); + for (const row of created) { + insertedArrivals.push(row); + touched.push({ type: row.type, measuredAt: row.measuredAt }); } + imported += created.length; + if (created.length < chunk.length) { + getEvent()?.addWarning( + `Withings measure insert skipped ${chunk.length - created.length} rows already present (idempotent race)`, + ); + } + } catch (err) { + anyRowFailed = true; + if (firstRowError === null) firstRowError = err; + getEvent()?.addWarning(`Failed to insert measures: ${err}`); } } @@ -446,9 +495,7 @@ export async function syncUserMeasurements( // chart at the 30-day range under-counted recent days. try { const keys = collapseToTypeDayKeys(touched); - for (const k of keys) { - await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); - } + await recomputeBucketsForKeys(userId, keys); // v1.8.0 — a Withings sync that lands new weight / BP / pulse / // body-comp rows dirties the matching per-metric assessment caches. diff --git a/tests/integration/measurement-rollups.test.ts b/tests/integration/measurement-rollups.test.ts index 2d91a56e1..2e382c9a5 100644 --- a/tests/integration/measurement-rollups.test.ts +++ b/tests/integration/measurement-rollups.test.ts @@ -29,6 +29,7 @@ import { getPrismaClient, truncateAllTables } from "./setup"; import { ROLLUP_FULL_BACKFILL_QUEUE, enqueueBootTimeRollupBackfill, + recomputeBucketsForKeys, recomputeBucketsForMeasurement, recomputeUserRollups, ALL_GRANULARITIES, @@ -954,3 +955,166 @@ describe("measurement rollups — integration", () => { expect(aggregate.summaries.BLOOD_PRESSURE_SYS).toBeDefined(); }); }); + +describe("keyed DAY refold", () => { + /** The DAY tier as comparable rows, `computedAt` aside. */ + async function dayTier(userId: string) { + const rows = await getPrismaClient().measurementRollup.findMany({ + where: { userId, granularity: "DAY" }, + orderBy: [{ type: "asc" }, { bucketStart: "asc" }, { source: "asc" }], + }); + // Floating sums depend on the order the scan feeds them; the last bits + // of a double are not what the comparison is about. + return rows.map(({ computedAt: _computedAt, ...rest }) => + Object.fromEntries( + Object.entries(rest).map(([k, v]) => [ + k, + typeof v === "number" ? Number(v.toPrecision(12)) : v, + ]), + ), + ); + } + + it("leaves the tier the per-key hook leaves, touching nothing else", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { username: "keyed-refold", email: "keyed-refold@example.test" }, + }); + const at = (iso: string) => new Date(iso); + const reading = ( + type: "PULSE" | "WEIGHT", + source: "MANUAL" | "WITHINGS", + iso: string, + value: number, + ) => ({ + userId: user.id, + type, + source, + value, + unit: type === "PULSE" ? "bpm" : "kg", + measuredAt: at(iso), + }); + await prisma.measurement.createMany({ + data: [ + // Two sources on one day, pulse over several hours (its day is a + // mean of hourly means), a weight two years earlier, and a day whose + // only reading is about to be deleted. + reading("PULSE", "WITHINGS", "2026-05-10T07:10:00.000Z", 58), + reading("PULSE", "WITHINGS", "2026-05-10T07:40:00.000Z", 66), + reading("PULSE", "MANUAL", "2026-05-10T19:00:00.000Z", 71), + reading("WEIGHT", "WITHINGS", "2024-03-02T06:00:00.000Z", 82.4), + reading("WEIGHT", "WITHINGS", "2026-05-11T06:00:00.000Z", 81.0), + reading("WEIGHT", "MANUAL", "2025-01-15T06:00:00.000Z", 83.3), + ], + }); + const keys = [ + { type: "PULSE" as const, measuredAt: at("2026-05-10T07:10:00.000Z") }, + { type: "WEIGHT" as const, measuredAt: at("2024-03-02T06:00:00.000Z") }, + { type: "WEIGHT" as const, measuredAt: at("2026-05-11T06:00:00.000Z") }, + ]; + // An untouched bucket, folded once, must keep its row as it is. + await recomputeBucketsForMeasurement( + user.id, + "WEIGHT", + at("2025-01-15T06:00:00.000Z"), + ); + for (const k of keys) { + await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); + } + await prisma.measurement.updateMany({ + where: { userId: user.id, measuredAt: at("2026-05-11T06:00:00.000Z") }, + data: { deletedAt: new Date() }, + }); + await prisma.measurement.create({ + data: reading("PULSE", "WITHINGS", "2026-05-10T12:00:00.000Z", 90), + }); + const untouched = await prisma.measurementRollup.findFirstOrThrow({ + where: { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: at("2025-01-15T00:00:00.000Z"), + }, + }); + + // The per-key hook's answer, then the keyed refold's, from the same rows. + for (const k of keys) { + await recomputeBucketsForMeasurement(user.id, k.type, k.measuredAt); + } + const perKey = await dayTier(user.id); + await prisma.measurementRollup.deleteMany({ + where: { userId: user.id, bucketStart: { not: untouched.bucketStart } }, + }); + + const sends: Array<{ singletonKey: string }> = []; + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn(async (_queue: string, _payload: unknown, opts: never) => { + sends.push(opts); + return "job"; + }), + } as never); + await recomputeBucketsForKeys(user.id, keys); + + expect(await dayTier(user.id)).toEqual(perKey); + expect( + perKey.some( + (r) => + (r.bucketStart as Date).getTime() === + at("2026-05-11T00:00:00Z").getTime(), + ), + ).toBe(false); + const after = await prisma.measurementRollup.findFirstOrThrow({ + where: { + userId: user.id, + type: "WEIGHT", + granularity: "DAY", + bucketStart: untouched.bucketStart, + }, + }); + expect(after.computedAt).toEqual(untouched.computedAt); + + // One coarse job per bucket here: no two touched buckets of one type are + // adjacent, so no run merges. + expect(sends.map((o) => o.singletonKey).sort()).toEqual( + [ + `${user.id}|PULSE|WEEK|2026-05-04T00:00:00.000Z`, + `${user.id}|PULSE|MONTH|2026-05-01T00:00:00.000Z`, + `${user.id}|PULSE|YEAR|2026-01-01T00:00:00.000Z`, + `${user.id}|WEIGHT|WEEK|2024-02-26T00:00:00.000Z`, + `${user.id}|WEIGHT|WEEK|2026-05-11T00:00:00.000Z`, + `${user.id}|WEIGHT|MONTH|2024-03-01T00:00:00.000Z`, + `${user.id}|WEIGHT|MONTH|2026-05-01T00:00:00.000Z`, + `${user.id}|WEIGHT|YEAR|2024-01-01T00:00:00.000Z`, + `${user.id}|WEIGHT|YEAR|2026-01-01T00:00:00.000Z`, + ].sort(), + ); + }); + + it("merges adjacent buckets into one ranged job", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { username: "keyed-runs", email: "keyed-runs@example.test" }, + }); + const sends: Array<{ singletonKey: string }> = []; + vi.mocked(getGlobalBoss).mockReturnValue({ + send: vi.fn(async (_queue: string, _payload: unknown, opts: never) => { + sends.push(opts); + return "job"; + }), + } as never); + // Three consecutive weeks of a daily reading. + const keys = Array.from({ length: 21 }, (_, i) => ({ + type: "WEIGHT" as const, + measuredAt: new Date(Date.UTC(2026, 4, 4 + i, 6)), + })); + + await recomputeBucketsForKeys(user.id, keys); + + const weekKeys = sends + .map((o) => o.singletonKey) + .filter((k) => k.includes("|WEEK|")); + expect(weekKeys).toEqual([ + `${user.id}|WEIGHT|WEEK|2026-05-04T00:00:00.000Z|2026-05-25T00:00:00.000Z`, + ]); + }); +}); diff --git a/tests/integration/withings-sync-volume.test.ts b/tests/integration/withings-sync-volume.test.ts new file mode 100644 index 000000000..c2b97d9aa --- /dev/null +++ b/tests/integration/withings-sync-volume.test.ts @@ -0,0 +1,204 @@ +/** + * A Withings measure sync at the volume a full re-read of a real account + * carries: two years of daily weigh-ins (seven body-composition values each) + * and a daily blood-pressure reading, about 7 300 measures over 7 300 + * (type, day) buckets. + * + * The manual full sync re-reads the whole history, and nearly all of it is + * already stored. The run must therefore cost what CHANGED, not what was + * READ: one set-based read and write for the measures, and a rollup refold + * only for the days that gained or changed a reading. The timings are logged + * for the record; the assertions are on behaviour, so a slow CI host cannot + * fail them. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { encrypt } from "@/lib/crypto"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/withings/client", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, fetchMeasurements: vi.fn() }; +}); + +import { fetchMeasurements } from "@/lib/withings/client"; +import { syncUserMeasurements } from "@/lib/withings/sync"; + +const USER = "user-withings-volume"; +const DAY_MS = 86_400_000; +const DAYS = 730; + +const prisma = getPrismaClient(); + +const WEIGH_IN: Array<[string, number]> = [ + ["WEIGHT", 81.4], + ["BODY_FAT", 21.5], + ["FAT_MASS", 17.5], + ["FAT_FREE_MASS", 63.9], + ["MUSCLE_MASS", 60.7], + ["BONE_MASS", 3.2], + ["TOTAL_BODY_WATER", 46.1], +]; +const BP: Array<[string, number]> = [ + ["BLOOD_PRESSURE_SYS", 124], + ["BLOOD_PRESSURE_DIA", 79], + ["PULSE", 64], +]; + +/** Two years of history ending yesterday, at fixed UTC minutes. */ +function history(anchor: number) { + const out: Array<{ type: string; value: number; measuredAt: Date }> = []; + for (let d = DAYS; d >= 1; d -= 1) { + const day = anchor - d * DAY_MS; + for (const [type, base] of WEIGH_IN) { + out.push({ + type, + value: base + (d % 7) * 0.1, + measuredAt: new Date(day + 6 * 3_600_000 + 12 * 60_000), + }); + } + for (const [type, base] of BP) { + out.push({ + type, + value: base + (d % 5), + measuredAt: new Date(day + 20 * 3_600_000 + 41 * 60_000), + }); + } + } + return out; +} + +beforeEach(async () => { + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: USER, + username: "withings-volume", + email: "withings-volume@example.test", + }, + }); + await prisma.withingsConnection.create({ + data: { + userId: USER, + withingsUserId: "wu-volume", + accessToken: encrypt("access-token"), + refreshToken: encrypt("refresh-token"), + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + scope: "user.metrics,user.activity", + }, + }); +}); + +afterEach(() => { + vi.mocked(fetchMeasurements).mockReset(); +}); + +describe("Withings full sync at account volume", () => { + it("writes the history once, then re-reads it at the cost of what changed", async () => { + const anchor = Date.UTC(2026, 9, 20); + const all = history(anchor); + vi.mocked(fetchMeasurements).mockResolvedValue(all); + + let started = performance.now(); + const first = await syncUserMeasurements(USER, { fullSync: true }); + const firstMs = performance.now() - started; + expect(first).toEqual({ imported: all.length, failed: false }); + expect( + await prisma.measurement.count({ + where: { userId: USER, source: "WITHINGS" }, + }), + ).toBe(all.length); + expect( + await prisma.measurementRollup.count({ + where: { userId: USER, granularity: "DAY" }, + }), + ).toBe(DAYS * (WEIGH_IN.length + BP.length)); + + // The re-read: the same history, one corrected weigh-in, one reading + // that was deleted here and comes back, and a new reading today. + const corrected = all.find((m) => m.type === "WEIGHT")!; + const resurrected = all.find((m) => m.type === "PULSE")!; + await prisma.measurement.updateMany({ + where: { + userId: USER, + type: "PULSE", + measuredAt: resurrected.measuredAt, + }, + data: { deletedAt: new Date() }, + }); + const today = { + type: "WEIGHT", + value: 80.9, + measuredAt: new Date(anchor + 7 * 3_600_000), + }; + const reread = [ + ...all.map((m) => (m === corrected ? { ...m, value: 79.0 } : m)), + today, + ]; + vi.mocked(fetchMeasurements).mockResolvedValue(reread); + + const before = new Date(); + started = performance.now(); + const second = await syncUserMeasurements(USER, { fullSync: true }); + const secondMs = performance.now() - started; + console.log( + `withings full sync: ${all.length} measures first run ${firstMs.toFixed(0)} ms, re-read ${secondMs.toFixed(0)} ms`, + ); + + expect(second).toEqual({ imported: reread.length, failed: false }); + const stored = await prisma.measurement.findMany({ + where: { + userId: USER, + source: "WITHINGS", + OR: [ + { type: "WEIGHT", measuredAt: corrected.measuredAt }, + { type: "PULSE", measuredAt: resurrected.measuredAt }, + { type: "WEIGHT", measuredAt: today.measuredAt }, + ], + }, + select: { type: true, value: true, measuredAt: true, deletedAt: true }, + }); + expect(stored).toHaveLength(3); + expect(stored.every((m) => m.deletedAt === null)).toBe(true); + expect( + stored.find( + (m) => m.measuredAt.getTime() === corrected.measuredAt.getTime(), + )?.value, + ).toBe(79.0); + + // Only the three days that changed were refolded; the other buckets keep + // the aggregate the first run wrote. + const refolded = await prisma.measurementRollup.findMany({ + where: { + userId: USER, + granularity: "DAY", + computedAt: { gte: before }, + }, + select: { type: true, bucketStart: true, mean: true }, + }); + const dayOf = (d: Date) => + new Date(Math.floor(d.getTime() / DAY_MS) * DAY_MS); + expect( + refolded.map((r) => [r.type, r.bucketStart.toISOString()]).sort(), + ).toEqual([ + ["PULSE", dayOf(resurrected.measuredAt).toISOString()], + ["WEIGHT", dayOf(corrected.measuredAt).toISOString()], + ["WEIGHT", dayOf(today.measuredAt).toISOString()], + ]); + expect( + refolded.find( + (r) => + r.type === "WEIGHT" && + r.bucketStart.getTime() === dayOf(corrected.measuredAt).getTime(), + )?.mean, + ).toBe(79.0); + }); +});