Make WHOOP resource-cursor advance a single atomic upsert

The four WHOOP resources are independent pg-boss queues that run concurrently
for the same user, so the read-modify-write that merged one cursor key into the
map and wrote it back could let a sibling's write clobber a just-advanced
cursor. Replace it with a single jsonb_set UPDATE that merges in place under the
row lock, and GREATEST so the shared last-synced instant only moves forward.
The resource key is parameter-bound and asserted against the closed resource set
before it reaches the JSON path.
This commit is contained in:
Marc-André Bombeck committed 2026-06-14 17:23:55 +02:00
1 parent 2cb84c17e9
commit 7206615213
2 files changed
+72 -51

No files matched your search

+28 -26
View File
@@ -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<typeof markResourceSynced>[1],
new Date(),
);
expect(prismaMock.$executeRaw).not.toHaveBeenCalled();
});
});
+44 -25
View File
@@ -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<void> {
}
/**
* 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<WhoopResource> = 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<void> {
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}
`;
}
/**