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); + }); +});