From 86d917c5e299c9539c44c037c2e5dc2782dd14a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 26 Jun 2026 13:16:00 +0200 Subject: [PATCH 1/3] fix(insights): align cold-fallback slope window to the rollup day boundary The rollup read path windows the 7/30/90-day regression on startOfUtcDay(now - N days), while the live-SQL fallback filtered the REGR_* aggregates on the wall-clock NOW() - INTERVAL bound. An account crossing a coverage boundary (warm to cold) therefore saw the boundary day's readings flip in or out of the slope window depending only on whether the rollup tier was populated, returning a cache-dependent slope for the same request. Anchor both cold-fallback paths on date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL 'N days', the SQL twin of startOfUtcDay, so warm and cold return identical boundary-day membership. The avg7/avg30/median windows stay wall-clock since the warm narrow query computes them the same way. --- .../__tests__/summaries-slice.test.ts | 50 +++++++++++++++++++ src/lib/analytics/summaries-slice.ts | 25 +++++++--- src/lib/insights/comprehensive-aggregator.ts | 25 +++++++--- 3 files changed, 88 insertions(+), 12 deletions(-) diff --git a/src/lib/analytics/__tests__/summaries-slice.test.ts b/src/lib/analytics/__tests__/summaries-slice.test.ts index 1102e2608..2d18dcbd6 100644 --- a/src/lib/analytics/__tests__/summaries-slice.test.ts +++ b/src/lib/analytics/__tests__/summaries-slice.test.ts @@ -652,4 +652,54 @@ describe("computeSummariesSlice", () => { ); }); }); + + /** + * A2-M2 — read-swap boundary-day consistency. The warm + * `computeFromRollups` path windows the regression on UTC-midnight + * (`startOfUtcDay(now − N days)` + `composeWindowedRegression`). The + * cold-fallback `windowed` query must anchor its REGR_* FILTERs on the + * SAME UTC-midnight boundary — `date_trunc('day', NOW() AT TIME ZONE + * 'UTC') − INTERVAL 'N days'` — so a warm→cold coverage transition + * returns identical boundary-day membership for slope7/30/90 instead of + * a cache-dependent answer to the same request. + */ + describe("A2-M2 regression-window boundary alignment", () => { + it("anchors the cold-fallback slope FILTERs on the UTC-midnight day boundary, not the wall-clock instant", async () => { + const queries: string[] = []; + UNSAFE.mockImplementation((sql: string) => { + queries.push(sql); + return Promise.resolve([]); + }); + + // Empty coverage ⇒ cold-fallback path ⇒ the `windowed` query runs. + RAW.mockResolvedValueOnce([]); + await computeSummariesSlice("user-a2m2"); + + const windowedSql = queries.find( + (sql) => sql.includes("AS slope7") && sql.includes("REGR_SLOPE"), + ); + expect(windowedSql).toBeDefined(); + const sql = windowedSql as string; + + // Every regression window (7/30/90 for slope + r²) anchors on the + // day-truncated UTC boundary. Six FILTERs total. + for (const days of ["7 days", "30 days", "90 days"]) { + expect(sql).toContain( + `(date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '${days}') AT TIME ZONE 'UTC'`, + ); + } + const anchored = sql.match( + /date_trunc\('day', NOW\(\) AT TIME ZONE 'UTC'\) - INTERVAL '(?:7|30|90) days'\) AT TIME ZONE 'UTC'/g, + ); + expect(anchored?.length).toBe(6); + + // The regression windows must NOT fall back to the bare wall-clock + // `NOW() - INTERVAL 'N days'` bound the warm path never uses for the + // slope. Isolate the slope columns (everything from the first + // REGR_SLOPE onward) so the avg7/avg30 windows above — which DO stay + // wall-clock by design — don't trip the assertion. + const slopeBlock = sql.slice(sql.indexOf("REGR_SLOPE")); + expect(slopeBlock).not.toMatch(/WHERE m\."measured_at" >= NOW\(\) -/); + }); + }); }); diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index 454e3c50d..fe43a1ad2 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -772,41 +772,54 @@ async function computeFromLiveAggregate( ) FILTER ( WHERE m."measured_at" >= NOW() - INTERVAL '90 days' )::double precision AS median, + -- A2-M2 — anchor the regression windows on the UTC-midnight day + -- boundary so this cold-fallback path returns the SAME boundary-day + -- membership the warm computeFromRollups composes via + -- startOfUtcDay(now - N days) + composeWindowedRegression. The + -- expression date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL + -- 'N days' is the SQL twin of startOfUtcDay; the trailing AT TIME + -- ZONE 'UTC' re-materialises the naive truncation as a UTC instant + -- so the measured_at (timestamptz) comparison lands on the same wall + -- the rollup bucketStart sits on. The wall-clock NOW() - INTERVAL + -- bound let the boundary day flip in/out of the slope window across + -- a warm-to-cold coverage transition (cache-dependent slope on the + -- same request); the avg7/30/median windows above stay wall-clock + -- because the warm narrow query computes THOSE the same way. REGR_SLOPE( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC' )::double precision AS slope7, REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC' )::double precision AS r2_7, REGR_SLOPE( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC' )::double precision AS slope30, REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC' )::double precision AS r2_30, REGR_SLOPE( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC' )::double precision AS slope90, REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC' )::double precision AS r2_90 FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")} GROUP BY m."type" diff --git a/src/lib/insights/comprehensive-aggregator.ts b/src/lib/insights/comprehensive-aggregator.ts index e00dc87be..2968a8885 100644 --- a/src/lib/insights/comprehensive-aggregator.ts +++ b/src/lib/insights/comprehensive-aggregator.ts @@ -682,41 +682,54 @@ async function buildFromLiveAggregate( WHERE m."measured_at" >= NOW() - INTERVAL '60 days' AND m."measured_at" < NOW() - INTERVAL '30 days' )::double precision AS avg30_last_month, + -- A2-M2 — the regression windows anchor on the UTC-midnight day + -- boundary, NOT the wall-clock instant, so this cold-fallback path + -- returns the SAME boundary-day membership the warm rollup path + -- composes from (startOfUtcDay(now - N days) + + -- composeWindowedRegression). The expression + -- date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL 'N days' + -- is the SQL twin of startOfUtcDay; the trailing AT TIME ZONE 'UTC' + -- re-materialises the naive truncation as a UTC instant so the + -- measured_at (timestamptz) comparison is on the same wall the + -- rollup bucketStart sits on. Without this anchor an account that + -- crosses a coverage boundary (warm to cold) saw the boundary day + -- flip in/out of the slope window, giving a cache-dependent answer + -- to the same request. REGR_SLOPE( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC' )::double precision AS slope7, REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '7 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '7 days') AT TIME ZONE 'UTC' )::double precision AS r2_7, REGR_SLOPE( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC' )::double precision AS slope30, REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '30 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '30 days') AT TIME ZONE 'UTC' )::double precision AS r2_30, REGR_SLOPE( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC' )::double precision AS slope90, REGR_R2( m."value", EXTRACT(EPOCH FROM m."measured_at") / 86400.0 ) FILTER ( - WHERE m."measured_at" >= NOW() - INTERVAL '90 days' + WHERE m."measured_at" >= (date_trunc('day', NOW() AT TIME ZONE 'UTC') - INTERVAL '90 days') AT TIME ZONE 'UTC' )::double precision AS r2_90 FROM cm m JOIN window_stats ws ON ws."type" = m."type" From be7c5c5f5ad62557588dd9d5442f6fc678204239 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 26 Jun 2026 13:18:54 +0200 Subject: [PATCH 2/3] perf(withings): batch the activity sync reads and writes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The activity sync ran a per-(entry, field) findFirst followed by an update or create, roughly entries x 3 x 2 round-trips — about 2,190 on a 365-day backfill. Flatten the entry/field grid into the concrete rows the sync wants to land, resolve every overwrite-vs-insert decision in one findMany keyed on the same four dedup columns, then run a single createMany for fresh slots and one transaction for the changed-value overwrites. A backfill is now a bounded handful of queries regardless of the day count, and an unchanged re-sync is a no-op on the write path. The overwrite contract is unchanged: an existing WITHINGS row for a (type, measuredAt) slot keeps its row and only rewrites value; a fresh slot inserts with the stable externalId. createMany uses skipDuplicates so a concurrent sibling sync landing the same slot is tolerated by the four-column unique. --- .../withings/__tests__/sync-activity.test.ts | 203 +++++++++++++----- src/lib/withings/sync-activity.ts | 131 ++++++++--- 2 files changed, 255 insertions(+), 79 deletions(-) diff --git a/src/lib/withings/__tests__/sync-activity.test.ts b/src/lib/withings/__tests__/sync-activity.test.ts index fc463323e..6a926c349 100644 --- a/src/lib/withings/__tests__/sync-activity.test.ts +++ b/src/lib/withings/__tests__/sync-activity.test.ts @@ -7,18 +7,30 @@ */ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -vi.mock("@/lib/db", () => ({ - prisma: { - measurement: { - findFirst: vi.fn(), - create: vi.fn(), - update: vi.fn(), +vi.mock("@/lib/db", () => { + // v1.20.2 F3 — the write path batches: one `findMany` resolves the + // overwrite-vs-insert decision for the whole sync, `createMany` lands + // every fresh slot, and changed-value overwrites run inside a + // `$transaction([...update])`. The mock `$transaction` awaits the array + // of update promises so the per-row `update` mocks still fire and the + // call counts assert the batched shape. + const measurement = { + findMany: vi.fn(), + createMany: vi.fn(), + update: vi.fn(), + }; + return { + prisma: { + measurement, + withingsConnection: { + findUnique: vi.fn(), + }, + $transaction: vi.fn((ops: unknown) => + Array.isArray(ops) ? Promise.all(ops) : Promise.resolve(), + ), }, - withingsConnection: { - findUnique: vi.fn(), - }, - }, -})); + }; +}); vi.mock("@/lib/integrations/status", () => ({ isReauthRequired: vi.fn().mockResolvedValue(false), @@ -155,22 +167,35 @@ describe("fetchWithingsActivity", () => { }); }); +/** + * v1.20.2 F3 — helper: pull the flat `data[]` the batched `createMany` + * received (the writer always passes a single call with every fresh + * slot). Returns `[]` when `createMany` never fired. + */ +function createManyRows(): Array> { + const calls = vi.mocked(prisma.measurement.createMany).mock.calls; + if (calls.length === 0) return []; + return (calls[0][0] as { data: Array> }).data; +} + describe("syncUserActivity — field mapping + idempotency", () => { it("writes one row per (date, metric) the first time a day shows up", async () => { installFetchMock([ { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 }, ]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + // No existing slot rows → every planned row is an insert. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 3, + } as never); const imported = await syncUserActivity("user-1"); expect(imported).toBe(3); - expect(prisma.measurement.create).toHaveBeenCalledTimes(3); + // v1.20.2 F3 — one batched insert carrying all three metric rows. + expect(prisma.measurement.createMany).toHaveBeenCalledTimes(1); - const types = vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0].data as { type: string }).type); + const types = createManyRows().map((r) => r.type as string); expect(types.sort()).toEqual([ "ACTIVE_ENERGY_BURNED", "ACTIVITY_STEPS", @@ -178,33 +203,67 @@ describe("syncUserActivity — field mapping + idempotency", () => { ]); }); - it("updates instead of inserting when the same (date, metric) already exists", async () => { + it("updates instead of inserting when the same (date, metric) already exists with a different value", async () => { installFetchMock([ { date: "2026-05-12", steps: 9001, distance: 7000, calories: 450 }, ]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue({ - id: "row-existing", - } as never); + const measuredAt = new Date("2026-05-12T12:00:00.000Z"); + // Each slot already holds a row with a DIFFERENT value → all three + // overwrite (the value moved). Keyed by the same four columns the + // batched read matches on. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { id: "row-steps", type: "ACTIVITY_STEPS", measuredAt, value: 1 }, + { + id: "row-distance", + type: "WALKING_RUNNING_DISTANCE", + measuredAt, + value: 1, + }, + { + id: "row-calories", + type: "ACTIVE_ENERGY_BURNED", + measuredAt, + value: 1, + }, + ] as never); vi.mocked(prisma.measurement.update).mockResolvedValue({} as never); const imported = await syncUserActivity("user-1"); expect(imported).toBe(3); - expect(prisma.measurement.create).not.toHaveBeenCalled(); + expect(prisma.measurement.createMany).not.toHaveBeenCalled(); + // One batched `$transaction` carrying three per-row updates. + expect(prisma.$transaction).toHaveBeenCalledTimes(1); expect(prisma.measurement.update).toHaveBeenCalledTimes(3); }); + it("skips the write entirely when an existing slot already holds the same value (idempotent re-sync)", async () => { + installFetchMock([{ date: "2026-05-12", steps: 8420 }]); + const measuredAt = new Date("2026-05-12T12:00:00.000Z"); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { id: "row-steps", type: "ACTIVITY_STEPS", measuredAt, value: 8420 }, + ] as never); + + const imported = await syncUserActivity("user-1"); + + // The slot is still "touched" (rollup re-fold + imported count), but + // no value moved so neither the insert nor the update path runs. + expect(imported).toBe(1); + expect(prisma.measurement.createMany).not.toHaveBeenCalled(); + expect(prisma.measurement.update).not.toHaveBeenCalled(); + expect(prisma.$transaction).not.toHaveBeenCalled(); + }); + it("anchors measuredAt at noon UTC so the instant lands inside the local day for every supported tz", async () => { installFetchMock([{ date: "2026-05-12", steps: 1000 }]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 1, + } as never); await syncUserActivity("user-1"); - const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as { - data: { measuredAt: Date }; - }; - const measuredAt = createArg.data.measuredAt; + const measuredAt = createManyRows()[0].measuredAt as Date; expect(measuredAt.toISOString()).toBe("2026-05-12T12:00:00.000Z"); // Regression: anchoring at noon UTC keeps the row inside the @@ -233,15 +292,15 @@ describe("syncUserActivity — field mapping + idempotency", () => { // `distance` absent — Withings drops the field rather than emitting 0 // when there's no GPS. Steps + calories still write. installFetchMock([{ date: "2026-05-12", steps: 8000, calories: 380 }]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 2, + } as never); const imported = await syncUserActivity("user-1"); expect(imported).toBe(2); - const types = vi - .mocked(prisma.measurement.create) - .mock.calls.map((c) => (c[0].data as { type: string }).type); + const types = createManyRows().map((r) => r.type as string); expect(types).not.toContain("WALKING_RUNNING_DISTANCE"); }); @@ -252,8 +311,10 @@ describe("syncUserActivity — field mapping + idempotency", () => { installFetchMock([ { date: "2026-05-12", steps: 0, distance: 0, calories: 0 }, ]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 3, + } as never); const imported = await syncUserActivity("user-1"); expect(imported).toBe(3); @@ -261,8 +322,10 @@ describe("syncUserActivity — field mapping + idempotency", () => { it("calls recordSyncSuccess after a clean round-trip", async () => { installFetchMock([{ date: "2026-05-12", steps: 1000 }]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 1, + } as never); await syncUserActivity("user-1"); expect(recordSyncSuccess).toHaveBeenCalledWith("user-1", "withings"); @@ -270,18 +333,52 @@ describe("syncUserActivity — field mapping + idempotency", () => { it("tags every row with the canonical externalId so future replays dedup", async () => { installFetchMock([{ date: "2026-05-12", steps: 8420 }]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 1, + } as never); await syncUserActivity("user-1"); - const createArg = vi.mocked(prisma.measurement.create).mock.calls[0][0] as { - data: { externalId: string }; - }; - expect(createArg.data.externalId).toBe( + expect(createManyRows()[0].externalId).toBe( "withings:activity:user-1:2026-05-12:steps", ); }); + + it("batches a multi-day backfill into one read and one insert (F3)", async () => { + // 5 days × 3 metrics = 15 planned rows. The pre-v1.20.2 shape ran + // 15 findFirst + 15 create = 30 round-trips; the batched path is one + // findMany + one createMany regardless of the day count. + installFetchMock( + Array.from({ length: 5 }, (_, i) => ({ + date: `2026-05-1${i}`, + steps: 1000 + i, + distance: 2000 + i, + calories: 300 + i, + })), + ); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 15, + } as never); + + const imported = await syncUserActivity("user-1"); + + expect(imported).toBe(15); + expect(prisma.measurement.findMany).toHaveBeenCalledTimes(1); + expect(prisma.measurement.createMany).toHaveBeenCalledTimes(1); + expect(createManyRows()).toHaveLength(15); + // The single read is keyed by the full distinct (type, measuredAt) + // sets the batch will write — `skipDuplicates` guards the insert. + const readArg = vi.mocked(prisma.measurement.findMany).mock.calls[0][0] as { + where: { type: { in: unknown[] }; measuredAt: { in: unknown[] } }; + }; + expect(readArg.where.type.in).toHaveLength(3); + expect(readArg.where.measuredAt.in).toHaveLength(5); + const createArg = vi.mocked(prisma.measurement.createMany).mock + .calls[0][0] as { skipDuplicates: boolean }; + expect(createArg.skipDuplicates).toBe(true); + }); }); describe("syncUserActivity — scope-skip guard (v1.4.26)", () => { @@ -300,7 +397,7 @@ describe("syncUserActivity — scope-skip guard (v1.4.26)", () => { expect(imported).toBe(0); expect(fetchMock).not.toHaveBeenCalled(); - expect(prisma.measurement.create).not.toHaveBeenCalled(); + expect(prisma.measurement.createMany).not.toHaveBeenCalled(); expect(recordSyncSuccess).not.toHaveBeenCalled(); }); @@ -391,8 +488,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => { { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 }, { date: "2026-05-13", steps: 9100, distance: 7100, calories: 430 }, ]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 0, + } as never); await syncUserActivity("user-1"); @@ -418,8 +517,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => { installFetchMock([ { date: "2026-05-12", steps: 8420, distance: 6720, calories: 412 }, ]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 0, + } as never); await syncUserActivity("user-1"); @@ -431,8 +532,10 @@ describe("syncUserActivity — measurement rollup hook (v1.4.39.1)", () => { // hiccup must never fail the user's sync. The sync's `imported` // return value should still reflect the rows written. installFetchMock([{ date: "2026-05-12", steps: 8420 }]); - vi.mocked(prisma.measurement.findFirst).mockResolvedValue(null); - vi.mocked(prisma.measurement.create).mockResolvedValue({} as never); + vi.mocked(prisma.measurement.findMany).mockResolvedValue([] as never); + vi.mocked(prisma.measurement.createMany).mockResolvedValue({ + count: 0, + } as never); vi.mocked(recomputeBucketsForMeasurement).mockRejectedValueOnce( new Error("simulated rollup failure"), ); diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index d9e032af3..b3c00e2f2 100644 --- a/src/lib/withings/sync-activity.ts +++ b/src/lib/withings/sync-activity.ts @@ -316,6 +316,22 @@ export async function syncUserActivity( // See sync.ts header for the full rationale; the chart's // `source=rollup` fast-path otherwise misses Withings activity days. const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; + + // v1.20.2 F3 — flatten (entry × field) into the concrete rows the sync + // wants to land, then batch the reads + writes. The pre-v1.20.2 shape + // ran a per-(entry, field) `findFirst` + `update`/`create`, i.e. + // ~entries × 3 × 2 round-trips (≈ 2,190 on a 365-day backfill). The + // overwrite contract is unchanged: an existing WITHINGS row for the + // `(type, measuredAt)` slot has its `value` rewritten (the four-column + // dedup keys on `(userId, type, measuredAt, source, sleepStage=null)`), + // a fresh slot is inserted with the stable `externalId`. + interface PlannedRow { + type: MeasurementType; + measuredAt: Date; + value: number; + externalId: string; + } + const planned: PlannedRow[] = []; for (const entry of entries) { if (!entry.date) continue; const measuredAt = activityMeasuredAt(entry.date); @@ -325,42 +341,99 @@ export async function syncUserActivity( // valid (a day of rest still records 0 steps). Only skip // undefined / null / NaN. if (raw == null || !Number.isFinite(raw)) continue; - try { - const existing = await prisma.measurement.findFirst({ - where: { - userId, - type, - measuredAt, - source: "WITHINGS", - sleepStage: null, - }, - select: { id: true }, + planned.push({ + type, + measuredAt, + value: raw, + externalId: `withings:activity:${userId}:${entry.date}:${field}`, + }); + } + } + + if (planned.length > 0) { + const slotKey = (type: MeasurementType, measuredAt: Date) => + `${type}|${measuredAt.getTime()}`; + + try { + // One read for the whole batch: every existing WITHINGS / non-sleep + // row that occupies a slot we intend to write. Keyed by the same + // four columns the per-row `findFirst` matched on, so the + // overwrite-vs-insert decision is identical — just resolved in + // memory instead of N round-trips. + const distinctMeasuredAt = [ + ...new Map(planned.map((p) => [p.measuredAt.getTime(), p.measuredAt])), + ].map(([, d]) => d); + const distinctTypes = [...new Set(planned.map((p) => p.type))]; + const existingRows = await prisma.measurement.findMany({ + where: { + userId, + source: "WITHINGS", + sleepStage: null, + type: { in: distinctTypes }, + measuredAt: { in: distinctMeasuredAt }, + }, + select: { id: true, type: true, measuredAt: true, value: true }, + }); + const existingBySlot = new Map(); + for (const row of existingRows) { + existingBySlot.set(slotKey(row.type, row.measuredAt), { + id: row.id, + value: row.value, }); + } + + const toCreate: PlannedRow[] = []; + const toUpdate: Array<{ id: string; value: number }> = []; + for (const p of planned) { + const existing = existingBySlot.get(slotKey(p.type, p.measuredAt)); if (existing) { - await prisma.measurement.update({ - where: { id: existing.id }, - data: { value: raw }, - }); + // Only write when the value actually moved — a re-sync of an + // unchanged day is then a pure no-op on the write path. + if (existing.value !== p.value) { + toUpdate.push({ id: existing.id, value: p.value }); + } } else { - await prisma.measurement.create({ - data: { - userId, - type, - value: raw, - unit: getUnitForType(type), - measuredAt, - source: "WITHINGS", - externalId: `withings:activity:${userId}:${entry.date}:${field}`, - }, - }); + toCreate.push(p); } - touched.push({ type, measuredAt }); + touched.push({ type: p.type, measuredAt: p.measuredAt }); imported++; - } catch (err) { - getEvent()?.addWarning( - `Failed to upsert activity row (${type}, ${entry.date}): ${err}`, + } + + // One insert for every fresh slot. `skipDuplicates` tolerates a + // concurrent sibling sync landing the same slot between the read + // and this write (the four-column unique catches it). + if (toCreate.length > 0) { + await prisma.measurement.createMany({ + data: toCreate.map((p) => ({ + userId, + type: p.type, + value: p.value, + unit: getUnitForType(p.type), + measuredAt: p.measuredAt, + source: "WITHINGS" as const, + externalId: p.externalId, + })), + skipDuplicates: true, + }); + } + + // Values differ per row, so the overwrites can't collapse into one + // `updateMany`; run them in a single transaction so the batch is one + // round-trip rather than one per row. + if (toUpdate.length > 0) { + await prisma.$transaction( + toUpdate.map((u) => + prisma.measurement.update({ + where: { id: u.id }, + data: { value: u.value }, + }), + ), ); } + } catch (err) { + getEvent()?.addWarning( + `Failed to batch-upsert ${planned.length} activity rows for ${userId}: ${err}`, + ); } } From 9dcdf94db3ea1936d07079758afdeb80c0e97c97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 26 Jun 2026 13:21:23 +0200 Subject: [PATCH 3/3] perf(medications): batch the intake slot-dedup reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The slot-dedup pass ran a findFirst (latest takenAt) plus a findMany (all live rows) per medication — 2N round-trips, 40 for a 20-med user. Pull every live intake row for the user's medication set in one ordered findMany, group in memory by medicationId, and derive each medication's lastIntakeAt (max non-null takenAt) from the same grouped slice the findFirst used to scan. The read cost is now one query regardless of the medication count. Behaviour is unchanged: the grouped rows arrive in the same scheduledFor-asc order the heal and snap passes depend on, and the derived lastIntakeAt matches the old desc-ordered findFirst. --- .../intake-slot-dedup-isolation.test.ts | 66 +++++++++++++++++++ src/lib/medications/intake-slot-dedup.ts | 65 ++++++++++++------ 2 files changed, 112 insertions(+), 19 deletions(-) diff --git a/src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts b/src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts index 6d784a435..586640b6b 100644 --- a/src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts +++ b/src/lib/medications/__tests__/intake-slot-dedup-isolation.test.ts @@ -177,3 +177,69 @@ describe("dedupeUserIntakeSlots — per-slot error isolation", () => { expect(prisma.medicationIntakeEvent.updateMany).toHaveBeenCalledTimes(2); }); }); + +/** + * v1.20.2 F4 — the per-medication dedup read path is batched. The + * pre-v1.20.2 shape issued a `findFirst` (latest takenAt) + a `findMany` + * (all rows) PER medication → 2N round-trips. The batched path pulls + * every live row for the user's medication set in ONE `findMany`, groups + * in memory, and derives the per-med `lastIntakeAt` from the grouped set, + * so the read cost is one query regardless of the medication count. + */ +describe("dedupeUserIntakeSlots — batched read path (F4)", () => { + it("reads every medication's rows in a single findMany and never per-med findFirst", async () => { + const MED_2 = { ...MED, id: "med-2" }; + vi.mocked(prisma.medication.findMany).mockResolvedValue([ + MED, + MED_2, + ] as never); + + // Rows for BOTH medications come back in one ordered result set; the + // grouping keys them apart by `medicationId`. med-2 carries a single + // row (no duplicate slot) so it is a no-op — proving the grouping + // routes each med's slice correctly. + const med2Row = { + id: "r-m2", + medicationId: "med-2", + scheduledFor: new Date("2026-06-15T05:00:00.000Z"), + takenAt: new Date("2026-06-15T05:02:00.000Z"), + skipped: false, + syncVersion: 0, + createdAt: new Date("2026-06-15T00:00:00Z"), + attributionSource: "AUTO", + inventoryConsumption: null, + }; + // `inventoryConsumption: null` on every row keeps the loser-refund + // path (which itself runs a `findFirst` inside `restoreForIntake`) + // out of the picture, so the findFirst assertion isolates the removed + // lastIntake probe. + const med1Rows = seedRows().map((r) => ({ + ...r, + inventoryConsumption: null, + })); + vi.mocked(prisma.medicationIntakeEvent.findMany).mockResolvedValue([ + ...med1Rows, + med2Row, + ] as never); + vi.mocked(prisma.medicationIntakeEvent.update).mockResolvedValue( + {} as never, + ); + + const summary = await dedupeUserIntakeSlots("u1"); + + // ONE findMany for the whole medication set; the per-med findFirst is + // gone entirely. + expect(prisma.medicationIntakeEvent.findMany).toHaveBeenCalledTimes(1); + expect(prisma.medicationIntakeEvent.findFirst).not.toHaveBeenCalled(); + // The single read is keyed by the medication-id set, not per med. + const whereArg = vi.mocked(prisma.medicationIntakeEvent.findMany).mock + .calls[0][0] as { + where: { medicationId: { in: string[] } }; + }; + expect(whereArg.where.medicationId.in.sort()).toEqual(["med-1", "med-2"]); + + // med-1's two slots still collapse from its grouped slice; med-2's + // lone row is left alone. + expect(summary.slotsCollapsed).toBe(2); + }); +}); diff --git a/src/lib/medications/intake-slot-dedup.ts b/src/lib/medications/intake-slot-dedup.ts index 073c47db4..328d24c25 100644 --- a/src/lib/medications/intake-slot-dedup.ts +++ b/src/lib/medications/intake-slot-dedup.ts @@ -242,25 +242,24 @@ export async function dedupeUserIntakeSlots( // recompute after the collapse. Deduped so a busy day folds once. const daysToRecompute = new Set(); - for (const med of medications) { - if (med.schedules.length === 0) continue; + // v1.20.2 F4 — batch the per-medication reads. The pre-v1.20.2 shape + // issued a `findFirst` (latest takenAt) + a `findMany` (all live rows) + // PER medication → 2N round-trips (20 meds = 40). Pull every live row + // for the user's medication set in ONE ordered `findMany`, group in + // memory by `medicationId`, and derive the per-med `lastIntakeAt` (max + // non-null takenAt) from the same grouped set the `findFirst` would + // have scanned. The grouping preserves the `scheduledFor asc` order the + // snap + heal passes depend on (a stable Postgres sort keeps the + // per-medication slices in the same order a per-med query returned). + const medIds = medications + .filter((m) => m.schedules.length > 0) + .map((m) => m.id); - // Latest non-tombstoned takenAt — only rolling schedules consult it, - // but the snap resolver needs it threaded in. - const lastIntake = await prisma.medicationIntakeEvent.findFirst({ - where: { - userId, - medicationId: med.id, - deletedAt: null, - takenAt: { not: null }, - }, - orderBy: { takenAt: "desc" }, - select: { takenAt: true }, - }); - const lastIntakeAt = lastIntake?.takenAt ?? null; - - const rows = (await prisma.medicationIntakeEvent.findMany({ - where: { userId, medicationId: med.id, deletedAt: null }, + const rowsByMedication = new Map(); + const lastIntakeAtByMedication = new Map(); + if (medIds.length > 0) { + const allRows = (await prisma.medicationIntakeEvent.findMany({ + where: { userId, medicationId: { in: medIds }, deletedAt: null }, select: { id: true, medicationId: true, @@ -276,9 +275,37 @@ export async function dedupeUserIntakeSlots( // v1.16.10 — stamped losers refund their consumption on collapse. inventoryConsumption: true, }, - orderBy: { scheduledFor: "asc" }, + // `medicationId` first so the per-med slices are contiguous, then the + // `scheduledFor asc` the loop relies on. + orderBy: [{ medicationId: "asc" }, { scheduledFor: "asc" }], })) as DedupIntakeRow[]; + for (const row of allRows) { + const list = rowsByMedication.get(row.medicationId); + if (list) { + list.push(row); + } else { + rowsByMedication.set(row.medicationId, [row]); + } + // Latest non-tombstoned takenAt per medication — the same value the + // per-med `findFirst({ takenAt: { not: null }, orderBy: desc })` + // returned. Only rolling schedules consult it, but the snap resolver + // needs it threaded in. + if (row.takenAt !== null) { + const prev = lastIntakeAtByMedication.get(row.medicationId) ?? null; + if (prev === null || row.takenAt.getTime() > prev.getTime()) { + lastIntakeAtByMedication.set(row.medicationId, row.takenAt); + } + } + } + } + + for (const med of medications) { + if (med.schedules.length === 0) continue; + + const lastIntakeAt = lastIntakeAtByMedication.get(med.id) ?? null; + const rows = rowsByMedication.get(med.id) ?? []; + if (rows.length < 2) continue; // v1.16.9 — heal pass FIRST: converge ad-hoc actioned API/REMINDER