diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index 5bf8a3e56..851305e3d 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -25,6 +25,7 @@ const { measurement: { upsert: vi.fn(), }, + $executeRaw: vi.fn(), }, refreshAccessTokenMock: vi.fn(), fetchRecoveriesMock: vi.fn(), @@ -266,38 +267,39 @@ describe("resolveResourceCursor — per-resource cursor", () => { }); }); -describe("markResourceSynced — independent cursor advance", () => { - it("advances ONLY the named resource's key, preserving siblings", async () => { - prismaMock.whoopConnection.findUnique.mockResolvedValue({ - resourceCursors: { recovery: "2026-06-01T00:00:00.000Z" }, - }); - prismaMock.whoopConnection.update.mockResolvedValue({}); +describe("markResourceSynced — atomic, independent cursor advance", () => { + it("issues ONE atomic jsonb_set upsert binding the resource key + instant", async () => { + prismaMock.$executeRaw.mockResolvedValue(1); const at = new Date("2026-06-14T08:00:00.000Z"); await markResourceSynced("user1", "workout", at); - const data = prismaMock.whoopConnection.update.mock.calls[0]![0].data; - // recovery's older cursor is preserved; workout is set to `at`. - expect(data.resourceCursors).toEqual({ - recovery: "2026-06-01T00:00:00.000Z", - workout: at.toISOString(), - }); - // The shared cursor keeps moving for any legacy reader. - expect(data.lastSyncedAt).toEqual(at); + // A single atomic statement, never a read-then-write (which would race the + // concurrent sibling resource queues). + expect(prismaMock.whoopConnection.findUnique).not.toHaveBeenCalled(); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1); + + const call = prismaMock.$executeRaw.mock.calls[0]!; + const sql = (call[0] as TemplateStringsArray).join("?"); + const values = call.slice(1); + // Merge-in-place + monotonic last_synced_at + scoped to the user. + expect(sql).toContain("jsonb_set"); + expect(sql).toContain("GREATEST"); + // Resource key + ISO instant are parameter-bound, not spliced. + expect(values).toContain("workout"); + expect(values).toContain(at.toISOString()); + expect(values).toContain(at); + expect(values).toContain("user1"); }); - it("seeds the map from empty when the column is null", async () => { - prismaMock.whoopConnection.findUnique.mockResolvedValue({ - resourceCursors: null, - }); - prismaMock.whoopConnection.update.mockResolvedValue({}); - - const at = new Date("2026-06-14T08:00:00.000Z"); - await markResourceSynced("user1", "sleep", at); - - expect( - prismaMock.whoopConnection.update.mock.calls[0]![0].data.resourceCursors, - ).toEqual({ sleep: at.toISOString() }); + it("refuses a resource outside the closed whitelist (no statement issued)", async () => { + await markResourceSynced( + "user1", + "evil" as unknown as Parameters[1], + new Date(), + ); + expect(prismaMock.$executeRaw).not.toHaveBeenCalled(); }); }); diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index 45f2f1c9b..5cabd885f 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -24,7 +24,7 @@ */ import { AsyncLocalStorage } from "node:async_hooks"; import { prisma } from "@/lib/db"; -import type { MeasurementType, Prisma } from "@/generated/prisma/client"; +import type { MeasurementType } from "@/generated/prisma/client"; import { encrypt, decrypt } from "@/lib/crypto"; import { getEvent } from "@/lib/logging/context"; import { @@ -378,38 +378,57 @@ export async function markSynced(userId: string): Promise { } /** - * Advance ONE resource's cursor after its successful fetch+upsert. Reads the - * current `resourceCursors` map, sets only this resource's key to `now`, and - * writes the merged map back — sibling cursors are preserved untouched, so a - * resource that errored or soft-skipped this tick keeps its own (older) cursor - * and re-fetches from there next time. Also keeps the shared `lastSyncedAt` - * moving for any legacy reader still on it. + * The closed set of resources allowed in the `resourceCursors` JSON path. The + * key is passed to `jsonb_set` below — assert it against this whitelist so a + * caller can never splice an arbitrary path, even though `resource` is already + * the typed `WhoopResource` union. + */ +const WHOOP_CURSOR_RESOURCES: ReadonlySet = new Set([ + "recovery", + "sleep", + "workout", + "cycle", +]); + +/** + * Advance ONE resource's cursor after its successful fetch+upsert. Sibling + * cursors are preserved untouched, so a resource that errored or soft-skipped + * this tick keeps its own (older) cursor and re-fetches from there next time. + * Also keeps the shared `lastSyncedAt` moving (monotonically) for any legacy + * reader still on it. * - * Best-effort merge: the read+write is not transactional, but each resource - * runs serially within a user's sync (the per-resource pg-boss queues are - * `localConcurrency: 1`), so two cursors for the same user never race. + * The merge is a SINGLE atomic `jsonb_set` upsert, not a read-modify-write. The + * four WHOOP resources are independent pg-boss queues that run concurrently for + * the same user (a cron tick fans out all four at once), so a read-then-write + * would let two resources read the same map and the second write clobber the + * first's cursor. `jsonb_set` merges in-place under the row lock, and + * `GREATEST` keeps `last_synced_at` from being dragged backward by a slower + * sibling finishing with an older `at`. */ export async function markResourceSynced( userId: string, resource: WhoopResource, at: Date = new Date(), ): Promise { - const connection = await prisma.whoopConnection.findUnique({ - where: { userId }, - select: { resourceCursors: true }, - }); - if (!connection) return; + // Defence-in-depth: the typed union already bounds `resource`, but the value + // reaches a JSON path, so refuse anything outside the closed whitelist. + if (!WHOOP_CURSOR_RESOURCES.has(resource)) return; - const map = parseResourceCursors(connection.resourceCursors); - const next: ResourceCursorMap = { ...map, [resource]: at.toISOString() }; - - await prisma.whoopConnection.update({ - where: { userId }, - data: { - resourceCursors: next as Prisma.InputJsonValue, - lastSyncedAt: at, - }, - }); + const iso = at.toISOString(); + // `resource` is parameter-bound ($2), never string-spliced. `jsonb_set` + // builds the path array from it; `coalesce` seeds an empty object for a + // legacy null column; `GREATEST` advances `last_synced_at` only forward. + await prisma.$executeRaw` + UPDATE "whoop_connections" + SET "resource_cursors" = jsonb_set( + coalesce("resource_cursors", '{}'::jsonb), + ARRAY[${resource}], + to_jsonb(${iso}::text), + true + ), + "last_synced_at" = GREATEST("last_synced_at", ${at}) + WHERE "user_id" = ${userId} + `; } /**