From 2cb84c17e916a95c08e7b29468cb3f2369864144 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 14 Jun 2026 17:19:09 +0200 Subject: [PATCH 1/2] Add per-resource cursors, fetch-by-id, and full backfill to WHOOP sync Replace the single shared last-synced cursor with a per-resource cursor map on the WHOOP connection. Recovery, sleep, workout, and cycle each advance only their own cursor on a successful fetch+upsert, so a slow or 403-ing collection never drags the incremental window forward for its siblings and its own late-arriving records stay inside the overlap. Legacy connections fall back to the shared last-synced instant, so no historical state is lost. Add single-record fetch-by-id client routes and per-resource by-id syncs. The update webhook now carries the resource id through the queue payload, so the worker resolves and upserts that one record immediately rather than re-walking the whole collection on the next overlap tick. Anchor a full sync at a fixed deep-history point instead of returning no lower bound, so a backfill covers the multi-year span an incremental tick would skip and a stale cursor cannot silently narrow it. --- .../migration.sql | 18 ++ prisma/schema.prisma | 9 + .../webhook/[token]/__tests__/route.test.ts | 3 + .../__tests__/whoop-resource-sync.test.ts | 115 +++++++++++ src/lib/jobs/reminder/whoop-sync.ts | 83 ++++++-- src/lib/whoop/__tests__/client.test.ts | 37 ++++ .../__tests__/sync-workout-overlap.test.ts | 100 +++++++++- src/lib/whoop/__tests__/sync.test.ts | 112 ++++++++++- .../__tests__/webhook-notification.test.ts | 104 ++++++++++ src/lib/whoop/client.ts | 61 ++++++ src/lib/whoop/sync-cycle.ts | 9 +- src/lib/whoop/sync-recovery.ts | 61 +++++- src/lib/whoop/sync-sleep.ts | 47 ++++- src/lib/whoop/sync-workout.ts | 184 +++++++++++------- src/lib/whoop/sync.ts | 112 ++++++++++- src/lib/whoop/webhook-handler.ts | 9 +- 16 files changed, 938 insertions(+), 126 deletions(-) create mode 100644 prisma/migrations/0158_v11616_whoop_per_resource_cursor/migration.sql create mode 100644 src/lib/jobs/__tests__/whoop-resource-sync.test.ts create mode 100644 src/lib/whoop/__tests__/webhook-notification.test.ts diff --git a/prisma/migrations/0158_v11616_whoop_per_resource_cursor/migration.sql b/prisma/migrations/0158_v11616_whoop_per_resource_cursor/migration.sql new file mode 100644 index 000000000..c5b46e06e --- /dev/null +++ b/prisma/migrations/0158_v11616_whoop_per_resource_cursor/migration.sql @@ -0,0 +1,18 @@ +-- v1.16.16 (iOS #17) — per-resource WHOOP sync cursors. +-- +-- WHOOP's collection endpoints filter on a record's OWN time range, not on +-- when it reached the cloud, and each resource (recovery / sleep / workout / +-- cycle) re-scores and arrives on its own latency. A single shared +-- `last_synced_at` cursor advanced by the fastest resource then drags the +-- incremental `start` forward for a slow or failing resource too, so its +-- late-arriving records fall permanently before the window and are never +-- ingested. A 403/timeout on ONE collection likewise must not poison the +-- cursor every sibling reads. +-- +-- `resource_cursors` is a small JSON map `{ "recovery": "", ... }` of the +-- per-resource last-synced instant. Each resource advances only its own key on +-- a successful fetch+upsert; a stalled resource leaves its key untouched and +-- keeps re-fetching from where it left off without holding back the others. +-- Null = legacy connection — the readers fall back to the shared +-- `last_synced_at` so no historical state is lost. +ALTER TABLE "whoop_connections" ADD COLUMN "resource_cursors" JSONB; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 6a09c701b..8fcc91704 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -2168,6 +2168,15 @@ model WhoopConnection { /// a refresh token. Null = legacy connection that predates scope tracking. scope String? @map("scope") lastSyncedAt DateTime? @map("last_synced_at") + /// v1.16.16 (iOS #17) — per-resource sync cursors. A JSON map + /// `{ "recovery": "", "sleep": "", "workout": "", + /// "cycle": "" }` of each resource's own last-synced instant. The + /// incremental `start` for a resource is `cursor − overlap`; a resource + /// advances ONLY its own key on a successful fetch+upsert, so a slow or + /// 403-ing collection never drags the cursor forward for its siblings and + /// its own late-arriving records stay inside the window. Null / a missing + /// key falls back to the shared `lastSyncedAt` (legacy connection). + resourceCursors Json? @map("resource_cursors") /// Set once the self-converging history backfill walks every collection to /// completion. Null = backfill still in flight (or never started). backfillCompletedAt DateTime? @map("backfill_completed_at") diff --git a/src/app/api/whoop/webhook/[token]/__tests__/route.test.ts b/src/app/api/whoop/webhook/[token]/__tests__/route.test.ts index 442198c10..83dc167a2 100644 --- a/src/app/api/whoop/webhook/[token]/__tests__/route.test.ts +++ b/src/app/api/whoop/webhook/[token]/__tests__/route.test.ts @@ -162,8 +162,11 @@ describe("POST /api/whoop/webhook/[token]", () => { where: { whoopUserId: "42" }, select: { userId: true }, }); + // v1.16.16 (iOS #17) — the resource id rides through so the worker can do + // a targeted fetch-by-id refresh instead of re-walking the collection. expect(send).toHaveBeenCalledWith("whoop-recovery-sync", { userId: "user-1", + resourceId: "abc", }); }); diff --git a/src/lib/jobs/__tests__/whoop-resource-sync.test.ts b/src/lib/jobs/__tests__/whoop-resource-sync.test.ts new file mode 100644 index 000000000..0a304b7ae --- /dev/null +++ b/src/lib/jobs/__tests__/whoop-resource-sync.test.ts @@ -0,0 +1,115 @@ +/** + * v1.16.16 (iOS #17) — `runWhoopResourceSync` dispatch tests (mocked syncs). + * + * The webhook now carries the resource id; the worker must do a targeted + * fetch-by-id refresh when it is present, fall back to the per-user collection + * walk when a webhook job omits the id, and walk every connection only on the + * cron tick (no userId on any job). One user's failure never starves the + * cohort. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Job } from "pg-boss"; + +const { + syncUserWorkout, + syncWhoopWorkoutById, + findManyMock, +} = vi.hoisted(() => ({ + syncUserWorkout: vi.fn<(...a: unknown[]) => Promise>(async () => 0), + syncWhoopWorkoutById: vi.fn<(...a: unknown[]) => Promise>( + async () => 0, + ), + findManyMock: vi.fn<(...a: unknown[]) => Promise>>( + async () => [], + ), +})); + +vi.mock("@/lib/whoop/sync-recovery", () => ({ + syncUserRecovery: vi.fn(async () => 0), + syncWhoopRecoveryById: vi.fn(async () => 0), +})); +vi.mock("@/lib/whoop/sync-sleep", () => ({ + syncUserSleep: vi.fn(async () => 0), + syncWhoopSleepById: vi.fn(async () => 0), +})); +vi.mock("@/lib/whoop/sync-cycle", () => ({ + syncUserCycle: vi.fn(async () => 0), +})); +vi.mock("@/lib/whoop/sync-workout", () => ({ + syncUserWorkout: (...a: unknown[]) => syncUserWorkout(...a), + syncWhoopWorkoutById: (...a: unknown[]) => syncWhoopWorkoutById(...a), +})); + +vi.mock("@/lib/jobs/worker-status", () => ({ recordError: vi.fn() })); + +vi.mock("@/lib/logging/background", () => ({ + withBackgroundEvent: async ( + _name: string, + fn: (evt: { + setBackground: () => void; + setError: () => void; + addWarning: () => void; + }) => Promise, + ) => + fn({ + setBackground: vi.fn(), + setError: vi.fn(), + addWarning: vi.fn(), + }), +})); + +vi.mock("../reminder/shared", () => ({ + getWorkerPrisma: () => ({ + whoopConnection: { findMany: (...a: unknown[]) => findManyMock(...a) }, + }), +})); + +import { handleWhoopWorkoutSync } from "../reminder/whoop-sync"; + +function job(data: { userId?: string; resourceId?: string }): Job<{ + userId?: string; + resourceId?: string; +}> { + return { data } as Job<{ userId?: string; resourceId?: string }>; +} + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe("runWhoopResourceSync dispatch", () => { + it("a webhook job with a resourceId does a targeted fetch-by-id refresh", async () => { + await handleWhoopWorkoutSync([job({ userId: "u1", resourceId: "w-9" })]); + + expect(syncWhoopWorkoutById).toHaveBeenCalledWith("u1", "w-9"); + expect(syncUserWorkout).not.toHaveBeenCalled(); + expect(findManyMock).not.toHaveBeenCalled(); + }); + + it("a webhook job without a resourceId falls back to the per-user collection", async () => { + await handleWhoopWorkoutSync([job({ userId: "u1" })]); + + expect(syncUserWorkout).toHaveBeenCalledWith("u1"); + expect(syncWhoopWorkoutById).not.toHaveBeenCalled(); + expect(findManyMock).not.toHaveBeenCalled(); + }); + + it("a cron tick (no userId) walks every connection's collection", async () => { + findManyMock.mockResolvedValue([{ userId: "a" }, { userId: "b" }]); + + await handleWhoopWorkoutSync([job({})]); + + expect(findManyMock).toHaveBeenCalled(); + expect(syncUserWorkout).toHaveBeenCalledTimes(2); + expect(syncWhoopWorkoutById).not.toHaveBeenCalled(); + }); + + it("one user's failure never starves the rest of the cohort", async () => { + findManyMock.mockResolvedValue([{ userId: "a" }, { userId: "b" }]); + syncUserWorkout.mockRejectedValueOnce(new Error("boom for a")); + + await expect(handleWhoopWorkoutSync([job({})])).resolves.toBeUndefined(); + // Both users were attempted despite the first throwing. + expect(syncUserWorkout).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/lib/jobs/reminder/whoop-sync.ts b/src/lib/jobs/reminder/whoop-sync.ts index 7f5513124..aab16721d 100644 --- a/src/lib/jobs/reminder/whoop-sync.ts +++ b/src/lib/jobs/reminder/whoop-sync.ts @@ -7,10 +7,19 @@ import { type Job } from "pg-boss"; import { recordError } from "@/lib/jobs/worker-status"; import { withBackgroundEvent } from "@/lib/logging/background"; -import { syncUserRecovery } from "@/lib/whoop/sync-recovery"; -import { syncUserSleep as syncWhoopSleep } from "@/lib/whoop/sync-sleep"; +import { + syncUserRecovery, + syncWhoopRecoveryById, +} from "@/lib/whoop/sync-recovery"; +import { + syncUserSleep as syncWhoopSleep, + syncWhoopSleepById, +} from "@/lib/whoop/sync-sleep"; import { syncUserCycle } from "@/lib/whoop/sync-cycle"; -import { syncUserWorkout } from "@/lib/whoop/sync-workout"; +import { + syncUserWorkout, + syncWhoopWorkoutById, +} from "@/lib/whoop/sync-workout"; import { getWorkerPrisma } from "./shared"; /** @@ -18,34 +27,59 @@ import { getWorkerPrisma } from "./shared"; * WHOOP sync queue: * * 1. Webhook (`recovery.updated` / `sleep.updated` / `workout.updated`) — - * payload carries `userId`, the handler syncs that one user. + * payload carries `userId` and (v1.16.16) the `resourceId`; the handler + * does a targeted fetch-by-id refresh of that single record. * 2. Cron — payload has no `userId`; the handler iterates every WHOOP - * connection and re-syncs each, catching dropped webhook deliveries. - * Cycle has no webhook, so its cron is the sole driver. + * connection and re-syncs each collection, catching dropped webhook + * deliveries. Cycle has no webhook, so its cron is the sole driver. + * + * A webhook job that carries a `userId` but no `resourceId` (a legacy in-flight + * job, or a delivery that omitted the id) falls back to the per-user collection + * walk rather than touching every connection. */ export interface WhoopSyncPayload { userId?: string; + /** v1.16.16 — webhook resource id for a targeted fetch-by-id refresh. */ + resourceId?: string; } +/** A per-user collection walk (cron + the no-id webhook fallback). */ +type CollectionSync = (userId: string) => Promise; +/** A webhook-driven single-record refresh by resource id. */ +type ByIdSync = (userId: string, resourceId: string) => Promise; + /** - * Shared driver for the per-resource WHOOP sync handlers. Resolves the target - * set (per-user from the webhook payload, or every connection on the cron - * tick) and runs `syncFn` per user. One user's parked-at-reauth state never - * starves the rest of the cohort on the cron path. + * Shared driver for the per-resource WHOOP sync handlers. + * + * - A webhook job with a `resourceId` runs the targeted `byIdFn` (when the + * resource supports fetch-by-id) — landing the exact record immediately. + * - A webhook job with a `userId` but no `resourceId` runs the per-user + * collection `syncFn`. + * - A cron tick (no `userId` on any job) walks every connection's collection. + * + * One user's parked-at-reauth state never starves the rest of the cohort. */ export async function runWhoopResourceSync( taskName: string, jobs: Job[], - syncFn: (userId: string) => Promise, + syncFn: CollectionSync, + byIdFn?: ByIdSync, ): Promise { await withBackgroundEvent(taskName, async (evt) => { const prisma = getWorkerPrisma(); try { - const targets: Array<{ userId: string }> = []; + // (userId, optional resourceId) work items from the webhook payloads. + const targets: Array<{ userId: string; resourceId?: string }> = []; for (const job of jobs) { - if (job.data?.userId) targets.push({ userId: job.data.userId }); + if (job.data?.userId) { + targets.push({ + userId: job.data.userId, + resourceId: job.data.resourceId, + }); + } } if (targets.length === 0) { + // Cron tick — re-walk every connection's collection. const connections = await prisma.whoopConnection.findMany({ select: { userId: true }, }); @@ -55,9 +89,12 @@ export async function runWhoopResourceSync( let usersSynced = 0; let measurementsImported = 0; - for (const { userId } of targets) { + for (const { userId, resourceId } of targets) { try { - measurementsImported += await syncFn(userId); + measurementsImported += + resourceId && byIdFn + ? await byIdFn(userId, resourceId) + : await syncFn(userId); usersSynced++; } catch (err) { evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); @@ -85,17 +122,29 @@ export function handleWhoopRecoverySync(jobs: Job[]) { "job.whoop_recovery_sync", jobs, syncUserRecovery, + syncWhoopRecoveryById, ); } export function handleWhoopSleepSync(jobs: Job[]) { - return runWhoopResourceSync("job.whoop_sleep_sync", jobs, syncWhoopSleep); + return runWhoopResourceSync( + "job.whoop_sleep_sync", + jobs, + syncWhoopSleep, + syncWhoopSleepById, + ); } export function handleWhoopWorkoutSync(jobs: Job[]) { - return runWhoopResourceSync("job.whoop_workout_sync", jobs, syncUserWorkout); + return runWhoopResourceSync( + "job.whoop_workout_sync", + jobs, + syncUserWorkout, + syncWhoopWorkoutById, + ); } export function handleWhoopCycleSync(jobs: Job[]) { + // Cycle has no webhook (poll-only), so no fetch-by-id path. return runWhoopResourceSync("job.whoop_cycle_sync", jobs, syncUserCycle); } diff --git a/src/lib/whoop/__tests__/client.test.ts b/src/lib/whoop/__tests__/client.test.ts index d100da50b..74d3f12bb 100644 --- a/src/lib/whoop/__tests__/client.test.ts +++ b/src/lib/whoop/__tests__/client.test.ts @@ -6,6 +6,10 @@ import { WHOOP_PAGE_LIMIT, exchangeCode, fetchBodyMeasurement, + fetchRecoveryByCycleId, + fetchSleepById, + fetchWorkoutById, + fetchCycleById, fetchCycles, fetchProfile, fetchRecoveries, @@ -198,6 +202,39 @@ describe("single-object endpoints", () => { }); }); +describe("fetch-by-id (webhook-driven single-record refresh)", () => { + it("resolves one record at the documented v2 single-record path", async () => { + const cases: Array<[() => Promise, string]> = [ + [() => fetchSleepById("at", "s1"), "/v2/activity/sleep/s1"], + [() => fetchWorkoutById("at", "w1"), "/v2/activity/workout/w1"], + [() => fetchCycleById("at", "123"), "/v2/cycle/123"], + [() => fetchRecoveryByCycleId("at", "123"), "/v2/cycle/123/recovery"], + ]; + for (const [call, path] of cases) { + const fetchMock = installFetchMock([{ status: 200, body: { id: "x" } }]); + await call(); + const [url] = fetchMock.mock.calls[0] as unknown as [string]; + expect(url).toContain(path); + vi.unstubAllGlobals(); + } + }); + + it("returns the single record body (one workout by id)", async () => { + installFetchMock([ + { status: 200, body: { id: "w1", start: "s", end: "e" } }, + ]); + const w = await fetchWorkoutById("at", "w1"); + expect(w.id).toBe("w1"); + }); + + it("url-encodes a resource id with reserved characters", async () => { + const fetchMock = installFetchMock([{ status: 200, body: { id: "x" } }]); + await fetchWorkoutById("at", "a/b?c"); + const [url] = fetchMock.mock.calls[0] as unknown as [string]; + expect(url).toContain("/v2/activity/workout/a%2Fb%3Fc"); + }); +}); + describe("mapRecovery", () => { const base: WhoopRecovery = { cycle_id: 1, diff --git a/src/lib/whoop/__tests__/sync-workout-overlap.test.ts b/src/lib/whoop/__tests__/sync-workout-overlap.test.ts index 3462568e7..c90cfaf4e 100644 --- a/src/lib/whoop/__tests__/sync-workout-overlap.test.ts +++ b/src/lib/whoop/__tests__/sync-workout-overlap.test.ts @@ -9,30 +9,33 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; const fetchWorkouts = vi.fn(); +const fetchWorkoutById = vi.fn(); vi.mock("../client", () => ({ fetchWorkouts: (...a: unknown[]) => fetchWorkouts(...a), + fetchWorkoutById: (...a: unknown[]) => fetchWorkoutById(...a), KJ_TO_KCAL: 0.239006, })); const getValidToken = vi.fn(); -const markSynced = vi.fn(); +const markResourceSynced = vi.fn(); vi.mock("../sync", async () => { const actual = await vi.importActual("../sync"); return { ...actual, getValidToken: (...a: unknown[]) => getValidToken(...a), - markSynced: (...a: unknown[]) => markSynced(...a), + markResourceSynced: (...a: unknown[]) => markResourceSynced(...a), }; }); const whoopConnectionFindUnique = vi.fn(); +const workoutUpsert = vi.fn(); vi.mock("@/lib/db", () => ({ prisma: { whoopConnection: { findUnique: (...a: unknown[]) => whoopConnectionFindUnique(...a), }, - workout: { upsert: vi.fn() }, + workout: { upsert: (...a: unknown[]) => workoutUpsert(...a) }, }, })); @@ -40,8 +43,11 @@ vi.mock("@/lib/logging/context", () => ({ getEvent: () => ({ addWarning: vi.fn() }), })); -import { syncUserWorkout } from "../sync-workout"; -import { WHOOP_RECOVERY_SLEEP_OVERLAP_MS } from "../sync"; +import { syncUserWorkout, syncWhoopWorkoutById } from "../sync-workout"; +import { + WHOOP_FULL_SYNC_ANCHOR, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, +} from "../sync"; const LAST_SYNCED = new Date("2026-06-10T12:00:00.000Z"); @@ -51,8 +57,12 @@ beforeEach(() => { accessToken: "tok", connection: { id: "c1", whoopUserId: "w1" }, }); - whoopConnectionFindUnique.mockResolvedValue({ lastSyncedAt: LAST_SYNCED }); - markSynced.mockResolvedValue(undefined); + // No per-resource cursor yet → falls back to the shared lastSyncedAt. + whoopConnectionFindUnique.mockResolvedValue({ + lastSyncedAt: LAST_SYNCED, + resourceCursors: null, + }); + markResourceSynced.mockResolvedValue(undefined); fetchWorkouts.mockResolvedValue([]); }); @@ -70,9 +80,81 @@ describe("syncUserWorkout — late-synced workout overlap window", () => { expect(arg.start!.getTime()).toBeLessThanOrEqual(sixHoursBefore); }); - it("fullSync passes no start (deep backfill anchor handles history)", async () => { + it("fullSync anchors at the deep-history anchor, not the cursor", async () => { await syncUserWorkout("user-1", { fullSync: true }); const arg = fetchWorkouts.mock.calls[0]![1] as { start?: Date }; - expect(arg.start).toBeUndefined(); + expect(arg.start).toEqual(WHOOP_FULL_SYNC_ANCHOR); + }); + + it("uses the per-resource workout cursor when present, ignoring siblings", async () => { + const workoutCursor = new Date("2026-06-04T12:00:00.000Z"); + whoopConnectionFindUnique.mockResolvedValue({ + // recovery synced far more recently — must NOT pull the workout window + // forward. + lastSyncedAt: new Date("2026-06-12T00:00:00.000Z"), + resourceCursors: { + recovery: "2026-06-12T00:00:00.000Z", + workout: workoutCursor.toISOString(), + }, + }); + + await syncUserWorkout("user-1"); + + const arg = fetchWorkouts.mock.calls[0]![1] as { start?: Date }; + expect(arg.start!.getTime()).toBe( + workoutCursor.getTime() - WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + ); + }); +}); + +describe("syncWhoopWorkoutById — webhook fetch-by-id", () => { + const SCORED_WORKOUT = { + id: "w-123", + start: "2026-06-14T07:00:00.000Z", + end: "2026-06-14T08:00:00.000Z", + sport_id: 1, + score: { + strain: 12.3, + average_heart_rate: 130, + max_heart_rate: 165, + kilojoule: 1500, + percent_recorded: 99, + }, + }; + + it("resolves the ONE record by id and upserts it (no collection walk)", async () => { + fetchWorkoutById.mockResolvedValue(SCORED_WORKOUT); + workoutUpsert.mockResolvedValue({}); + + const imported = await syncWhoopWorkoutById("user-1", "w-123"); + + expect(imported).toBe(1); + // Targeted fetch-by-id, never the collection. + expect(fetchWorkoutById).toHaveBeenCalledWith("tok", "w-123"); + expect(fetchWorkouts).not.toHaveBeenCalled(); + const upsertArg = workoutUpsert.mock.calls[0]![0]; + expect(upsertArg.where.userId_source_externalId).toEqual({ + userId: "user-1", + source: "WHOOP", + externalId: "w-123", + }); + }); + + it("a single-id refresh does NOT advance the resource cursor", async () => { + fetchWorkoutById.mockResolvedValue(SCORED_WORKOUT); + workoutUpsert.mockResolvedValue({}); + + await syncWhoopWorkoutById("user-1", "w-123"); + + expect(markResourceSynced).not.toHaveBeenCalled(); + }); + + it("an unscored record stores nothing", async () => { + fetchWorkoutById.mockResolvedValue({ ...SCORED_WORKOUT, score: null }); + + const imported = await syncWhoopWorkoutById("user-1", "w-123"); + + expect(imported).toBe(0); + expect(workoutUpsert).not.toHaveBeenCalled(); }); }); diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts index e22b21f8f..5bf8a3e56 100644 --- a/src/lib/whoop/__tests__/sync.test.ts +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -88,8 +88,11 @@ import { getValidToken, incrementalStart, isCollectionForbidden, + markResourceSynced, + resolveResourceCursor, upsertWhoopMeasurements, WHOOP_DEFAULT_OVERLAP_MS, + WHOOP_FULL_SYNC_ANCHOR, WHOOP_RECOVERY_SLEEP_OVERLAP_MS, } from "../sync"; import { WhoopApiError } from "../response-classifier"; @@ -172,8 +175,11 @@ describe("getValidToken — rotating refresh", () => { }); describe("incrementalStart", () => { - it("returns undefined for a full sync", () => { - expect(incrementalStart(new Date(), { fullSync: true })).toBeUndefined(); + it("anchors a full sync at the deep-history anchor, ignoring the cursor", () => { + const got = incrementalStart(new Date("2026-06-01T12:00:00Z"), { + fullSync: true, + }); + expect(got).toEqual(WHOOP_FULL_SYNC_ANCHOR); }); it("subtracts the overlap from lastSyncedAt", () => { @@ -191,6 +197,108 @@ describe("incrementalStart", () => { const got = incrementalStart(last); expect(got!.getTime()).toBe(last.getTime() - WHOOP_DEFAULT_OVERLAP_MS); }); + + it("a fullSync reaches a span an incremental tick would skip", () => { + // Incremental from a recent cursor only looks back the overlap (days); a + // fullSync anchors years back, so it covers history the incremental misses. + const recentCursor = new Date("2026-06-01T12:00:00Z"); + const incremental = incrementalStart(recentCursor, { + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + }); + const full = incrementalStart(recentCursor, { fullSync: true }); + const yearOld = new Date("2025-01-01T00:00:00Z"); + // A year-old record is BEFORE the incremental start (skipped) but AFTER + // the full-sync anchor (covered). + expect(yearOld.getTime()).toBeLessThan(incremental!.getTime()); + expect(yearOld.getTime()).toBeGreaterThan(full!.getTime()); + }); +}); + +describe("resolveResourceCursor — per-resource cursor", () => { + it("prefers the per-resource cursor over the shared lastSyncedAt", () => { + const got = resolveResourceCursor( + { + resourceCursors: { workout: "2026-06-10T00:00:00.000Z" }, + lastSyncedAt: new Date("2026-06-12T00:00:00.000Z"), + }, + "workout", + ); + expect(got?.toISOString()).toBe("2026-06-10T00:00:00.000Z"); + }); + + it("falls back to lastSyncedAt for a resource with no cursor key (legacy)", () => { + const shared = new Date("2026-06-12T00:00:00.000Z"); + const got = resolveResourceCursor( + { resourceCursors: { recovery: "2026-06-10T00:00:00.000Z" }, lastSyncedAt: shared }, + "workout", + ); + expect(got).toEqual(shared); + }); + + it("falls back to lastSyncedAt when the column is null", () => { + const shared = new Date("2026-06-12T00:00:00.000Z"); + expect( + resolveResourceCursor({ resourceCursors: null, lastSyncedAt: shared }, "sleep"), + ).toEqual(shared); + }); + + it("a stalled resource cursor does not advance with a sibling's", () => { + // workout last synced a week ago; recovery synced an hour ago. Resolving + // workout must still return the OLD workout cursor — a sibling's progress + // never drags the stalled resource's window forward. + const connection = { + resourceCursors: { + recovery: "2026-06-12T11:00:00.000Z", + workout: "2026-06-05T12:00:00.000Z", + }, + lastSyncedAt: new Date("2026-06-12T11:00:00.000Z"), + }; + const workoutStart = incrementalStart( + resolveResourceCursor(connection, "workout"), + { overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS }, + ); + // Workout still re-fetches from a week-old cursor minus the overlap, NOT + // from recovery's recent one — the late-synced workout stays in range. + expect(workoutStart!.getTime()).toBe( + new Date("2026-06-05T12:00:00.000Z").getTime() - + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + ); + }); +}); + +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({}); + + 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); + }); + + 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() }); + }); }); describe("upsertWhoopMeasurements — idempotent upsert", () => { diff --git a/src/lib/whoop/__tests__/webhook-notification.test.ts b/src/lib/whoop/__tests__/webhook-notification.test.ts new file mode 100644 index 000000000..09f6d5171 --- /dev/null +++ b/src/lib/whoop/__tests__/webhook-notification.test.ts @@ -0,0 +1,104 @@ +/** + * v1.16.16 (iOS #17) — `processWhoopNotification` enqueue tests (mocked). + * + * A `*.updated` webhook must enqueue the per-resource sync job carrying the + * resource id so the worker can do a targeted fetch-by-id refresh (landing the + * exact record immediately rather than waiting for the next overlap window). + * The existing 200-always + unknown-user + delete-soft-delete contract is + * preserved. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { findFirst, bossSend, updateMany, deleteMany } = vi.hoisted(() => ({ + findFirst: vi.fn(), + bossSend: vi.fn(), + updateMany: vi.fn(), + deleteMany: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + whoopConnection: { findFirst: (...a: unknown[]) => findFirst(...a) }, + measurement: { updateMany: (...a: unknown[]) => updateMany(...a) }, + workout: { deleteMany: (...a: unknown[]) => deleteMany(...a) }, + }, +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => ({ send: (...a: unknown[]) => bossSend(...a) }), +})); + +vi.mock("@/lib/rate-limit", () => ({ + checkRateLimit: vi.fn(), + rateLimitHeaders: () => ({}), +})); +vi.mock("@/lib/api-response", () => ({ getClientIp: () => "1.2.3.4" })); +vi.mock("@/lib/logging/context", () => ({ + annotate: () => {}, + getEvent: () => ({ setAuth: vi.fn(), addWarning: vi.fn() }), +})); + +import { processWhoopNotification } from "../webhook-handler"; + +beforeEach(() => { + vi.clearAllMocks(); + findFirst.mockResolvedValue({ userId: "user-1" }); + bossSend.mockResolvedValue("job-id"); +}); + +describe("processWhoopNotification — fetch-by-id enqueue", () => { + it("enqueues the workout queue with the resource id", async () => { + const res = await processWhoopNotification({ + user_id: 42, + id: "w-123", + type: "workout.updated", + }); + + expect(res.status).toBe(200); + expect(bossSend).toHaveBeenCalledWith("whoop-workout-sync", { + userId: "user-1", + resourceId: "w-123", + }); + }); + + it("carries a numeric resource id (cycle-based recovery) as a string", async () => { + await processWhoopNotification({ + user_id: 42, + id: 9981, + type: "recovery.updated", + }); + + expect(bossSend).toHaveBeenCalledWith("whoop-recovery-sync", { + userId: "user-1", + resourceId: "9981", + }); + }); + + it("returns 200 for an unknown user without enqueuing", async () => { + findFirst.mockResolvedValue(null); + + const res = await processWhoopNotification({ + user_id: 7, + id: "w-1", + type: "workout.updated", + }); + + expect(res.status).toBe(200); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("soft-deletes on a *.deleted event (no enqueue)", async () => { + updateMany.mockResolvedValue({ count: 3 }); + deleteMany.mockResolvedValue({ count: 1 }); + + const res = await processWhoopNotification({ + user_id: 42, + id: "w-123", + type: "workout.deleted", + }); + + expect(res.status).toBe(200); + expect(bossSend).not.toHaveBeenCalled(); + expect(deleteMany).toHaveBeenCalled(); + }); +}); diff --git a/src/lib/whoop/client.ts b/src/lib/whoop/client.ts index 90cf76fd2..a75ad1e06 100644 --- a/src/lib/whoop/client.ts +++ b/src/lib/whoop/client.ts @@ -402,6 +402,67 @@ export function fetchWorkouts( ); } +// ─── Single-record reads (webhook-driven fetch-by-id) ────────── +// A `*.updated` webhook carries only the resource id, not the data. Re-fetch +// that ONE record by id for a targeted refresh — far cheaper than re-walking +// the whole collection, and it lands the record immediately rather than on the +// next overlap window. WHOOP v2 single-record paths (re-verify at build): +// sleep GET /v2/activity/sleep/{sleepId} +// workout GET /v2/activity/workout/{workoutId} +// cycle GET /v2/cycle/{cycleId} +// recovery GET /v2/cycle/{cycleId}/recovery (no recovery-by-recovery-id) +// `fetchSingle` throws a `WhoopApiError` on a non-2xx (incl. a 404 for a since- +// deleted id) so the caller's existing classify/soft-skip path handles it. + +export function fetchSleepById( + accessToken: string, + sleepId: string, +): Promise { + return fetchSingle( + `/v2/activity/sleep/${encodeURIComponent(sleepId)}`, + accessToken, + "fetchSleepById", + ); +} + +export function fetchWorkoutById( + accessToken: string, + workoutId: string, +): Promise { + return fetchSingle( + `/v2/activity/workout/${encodeURIComponent(workoutId)}`, + accessToken, + "fetchWorkoutById", + ); +} + +export function fetchCycleById( + accessToken: string, + cycleId: string, +): Promise { + return fetchSingle( + `/v2/cycle/${encodeURIComponent(cycleId)}`, + accessToken, + "fetchCycleById", + ); +} + +/** + * WHOOP v2 has no recovery-by-recovery-id route — a recovery is read through + * its cycle (`GET /v2/cycle/{cycleId}/recovery`). The recovery webhook carries + * the cycle id, so the targeted refresh resolves the recovery from there. + */ +export function fetchRecoveryByCycleId( + accessToken: string, + cycleId: string, +): Promise { + return fetchSingle( + `/v2/cycle/${encodeURIComponent(cycleId)}/recovery`, + accessToken, + "fetchRecoveryByCycleId", + ); +} + export function fetchBodyMeasurement( accessToken: string, ): Promise { diff --git a/src/lib/whoop/sync-cycle.ts b/src/lib/whoop/sync-cycle.ts index 81bb2a9bb..9819094e8 100644 --- a/src/lib/whoop/sync-cycle.ts +++ b/src/lib/whoop/sync-cycle.ts @@ -10,7 +10,8 @@ import { getValidToken, incrementalStart, handleCollectionFetchError, - markSynced, + markResourceSynced, + resolveResourceCursor, upsertWhoopMeasurements, type WhoopMeasurementUpsert, } from "./sync"; @@ -25,11 +26,11 @@ export async function syncUserCycle( const connection = await prisma.whoopConnection.findUnique({ where: { userId }, - select: { lastSyncedAt: true }, + select: { lastSyncedAt: true, resourceCursors: true }, }); if (!connection) return 0; - const start = incrementalStart(connection.lastSyncedAt, { + const start = incrementalStart(resolveResourceCursor(connection, "cycle"), { fullSync: opts.fullSync, }); @@ -54,6 +55,6 @@ export async function syncUserCycle( } const imported = await upsertWhoopMeasurements(userId, readings); - await markSynced(userId); + await markResourceSynced(userId, "cycle"); return imported; } diff --git a/src/lib/whoop/sync-recovery.ts b/src/lib/whoop/sync-recovery.ts index c348db88d..f9a032889 100644 --- a/src/lib/whoop/sync-recovery.ts +++ b/src/lib/whoop/sync-recovery.ts @@ -9,12 +9,17 @@ * sleep UUID (v2 recovery carries `sleep_id`, not a stable recovery id) so the * externalId is stable across re-scores. */ -import { fetchRecoveries, mapRecovery } from "./client"; +import { + fetchRecoveries, + fetchRecoveryByCycleId, + mapRecovery, +} from "./client"; import { getValidToken, incrementalStart, handleCollectionFetchError, - markSynced, + markResourceSynced, + resolveResourceCursor, upsertWhoopMeasurements, WHOOP_RECOVERY_SLEEP_OVERLAP_MS, type WhoopMeasurementUpsert, @@ -30,14 +35,17 @@ export async function syncUserRecovery( const connection = await prisma.whoopConnection.findUnique({ where: { userId }, - select: { lastSyncedAt: true }, + select: { lastSyncedAt: true, resourceCursors: true }, }); if (!connection) return 0; - const start = incrementalStart(connection.lastSyncedAt, { - fullSync: opts.fullSync, - overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, - }); + const start = incrementalStart( + resolveResourceCursor(connection, "recovery"), + { + fullSync: opts.fullSync, + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + }, + ); let records: Awaited>; try { @@ -62,6 +70,43 @@ export async function syncUserRecovery( } const imported = await upsertWhoopMeasurements(userId, readings); - await markSynced(userId); + await markResourceSynced(userId, "recovery"); return imported; } + +/** + * Webhook-driven targeted refresh: resolve ONE recovery by its cycle id and + * upsert its readings, instead of re-walking the whole collection. WHOOP v2 + * reads a recovery through its cycle (`/v2/cycle/{cycleId}/recovery`), and the + * recovery webhook carries the cycle id. An unscored / since-deleted record + * yields nothing. Does NOT advance the resource cursor — a single-id refresh + * proves nothing about records between the cursor and now, so the cron/overlap + * path stays responsible for moving the cursor. + */ +export async function syncWhoopRecoveryById( + userId: string, + cycleId: string, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + let record: Awaited>; + try { + record = await fetchRecoveryByCycleId(tokenInfo.accessToken, cycleId); + } catch (err) { + return handleCollectionFetchError("recovery", userId, err); + } + + const readings: WhoopMeasurementUpsert[] = []; + for (const m of mapRecovery(record)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: `${record.sleep_id}:${m.fieldTag}`, + }); + } + + return upsertWhoopMeasurements(userId, readings); +} diff --git a/src/lib/whoop/sync-sleep.ts b/src/lib/whoop/sync-sleep.ts index 82f0f7b84..234052ad0 100644 --- a/src/lib/whoop/sync-sleep.ts +++ b/src/lib/whoop/sync-sleep.ts @@ -7,12 +7,13 @@ * Per-stage rows carry the `sleepStage` axis so the five stage rows for one * night stay distinct under the dedup key. externalId = `:`. */ -import { fetchSleeps, mapSleep } from "./client"; +import { fetchSleeps, fetchSleepById, mapSleep } from "./client"; import { getValidToken, incrementalStart, handleCollectionFetchError, - markSynced, + markResourceSynced, + resolveResourceCursor, upsertWhoopMeasurements, WHOOP_RECOVERY_SLEEP_OVERLAP_MS, type WhoopMeasurementUpsert, @@ -28,11 +29,11 @@ export async function syncUserSleep( const connection = await prisma.whoopConnection.findUnique({ where: { userId }, - select: { lastSyncedAt: true }, + select: { lastSyncedAt: true, resourceCursors: true }, }); if (!connection) return 0; - const start = incrementalStart(connection.lastSyncedAt, { + const start = incrementalStart(resolveResourceCursor(connection, "sleep"), { fullSync: opts.fullSync, overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, }); @@ -59,6 +60,42 @@ export async function syncUserSleep( } const imported = await upsertWhoopMeasurements(userId, readings); - await markSynced(userId); + await markResourceSynced(userId, "sleep"); return imported; } + +/** + * Webhook-driven targeted refresh: resolve ONE sleep activity by its uuid and + * upsert its readings, instead of re-walking the whole collection. An unscored + * / since-deleted record yields nothing. Does NOT advance the resource cursor + * (a single-id refresh proves nothing about the records between the cursor and + * now). + */ +export async function syncWhoopSleepById( + userId: string, + sleepId: string, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + let record: Awaited>; + try { + record = await fetchSleepById(tokenInfo.accessToken, sleepId); + } catch (err) { + return handleCollectionFetchError("sleep", userId, err); + } + + const readings: WhoopMeasurementUpsert[] = []; + for (const m of mapSleep(record)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: `${record.id}:${m.fieldTag}`, + sleepStage: m.sleepStage ?? null, + }); + } + + return upsertWhoopMeasurements(userId, readings); +} diff --git a/src/lib/whoop/sync-workout.ts b/src/lib/whoop/sync-workout.ts index 8e46f8fe9..661d070ef 100644 --- a/src/lib/whoop/sync-workout.ts +++ b/src/lib/whoop/sync-workout.ts @@ -13,12 +13,18 @@ * rows (different `source`); the read-time `pickCanonicalWorkoutRows` picker * (E-slice, W6) collapses the cross-source twin at read time. */ -import { fetchWorkouts, KJ_TO_KCAL } from "./client"; +import { + fetchWorkouts, + fetchWorkoutById, + KJ_TO_KCAL, + type WhoopWorkout, +} from "./client"; import { getValidToken, incrementalStart, handleCollectionFetchError, - markSynced, + markResourceSynced, + resolveResourceCursor, WHOOP_RECOVERY_SLEEP_OVERLAP_MS, } from "./sync"; import { prisma } from "@/lib/db"; @@ -41,7 +47,7 @@ export async function syncUserWorkout( const connection = await prisma.whoopConnection.findUnique({ where: { userId }, - select: { lastSyncedAt: true }, + select: { lastSyncedAt: true, resourceCursors: true }, }); if (!connection) return 0; @@ -53,10 +59,13 @@ export async function syncUserWorkout( // the same window recovery/sleep already use for their late re-scores. The // `(userId, WHOOP, externalId)` upsert keeps the re-fetch idempotent and a // handful of workouts/day keeps the page count down. - const start = incrementalStart(connection.lastSyncedAt, { - fullSync: opts.fullSync, - overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, - }); + const start = incrementalStart( + resolveResourceCursor(connection, "workout"), + { + fullSync: opts.fullSync, + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + }, + ); let records: Awaited>; try { @@ -67,75 +76,108 @@ export async function syncUserWorkout( let imported = 0; for (const w of records) { - if (!w.score) continue; // unscored workout — nothing to store yet - const startedAt = new Date(w.start); - const endedAt = new Date(w.end); - const durationSec = Math.max( - 0, - Math.round((endedAt.getTime() - startedAt.getTime()) / 1000), - ); - const energyKcal = - typeof w.score.kilojoule === "number" - ? Math.round(w.score.kilojoule * KJ_TO_KCAL) - : null; + imported += await upsertWhoopWorkout(userId, w); + } - const metadata: Prisma.InputJsonValue = { - whoopWorkoutStrain: w.score.strain, - percentRecorded: w.score.percent_recorded, - ...(w.score.altitude_gain_meter != null - ? { altitudeGainMeter: w.score.altitude_gain_meter } - : {}), - ...(w.score.altitude_change_meter != null - ? { altitudeChangeMeter: w.score.altitude_change_meter } - : {}), - ...(w.score.zone_durations - ? { zoneDurations: w.score.zone_durations } - : {}), - }; + await markResourceSynced(userId, "workout"); + return imported; +} - try { - await prisma.workout.upsert({ - where: { - userId_source_externalId: { - userId, - source: "WHOOP", - externalId: w.id, - }, - }, - create: { +/** + * Webhook-driven targeted refresh: resolve ONE workout by its uuid and upsert + * it, instead of re-walking the whole collection. This is the direct fix for + * iOS #17 — a `workout.updated` webhook lands the exact workout immediately, + * with no dependence on the overlap window catching it. An unscored / + * since-deleted record yields nothing. Does NOT advance the resource cursor. + */ +export async function syncWhoopWorkoutById( + userId: string, + workoutId: string, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + let record: WhoopWorkout; + try { + record = await fetchWorkoutById(tokenInfo.accessToken, workoutId); + } catch (err) { + return handleCollectionFetchError("workout", userId, err); + } + + return upsertWhoopWorkout(userId, record); +} + +/** + * Upsert one WHOOP workout into the `Workout` table keyed on + * `(userId, source = WHOOP, externalId)`. Returns 1 when a row was written, 0 + * for an unscored record (nothing to store yet) or an upsert failure (logged, + * never thrown — a single bad record must not fail the surrounding sync). + * Shared by the collection walk and the webhook-driven fetch-by-id refresh so + * both write an identical row shape. + */ +export async function upsertWhoopWorkout( + userId: string, + w: WhoopWorkout, +): Promise { + if (!w.score) return 0; // unscored workout — nothing to store yet + const startedAt = new Date(w.start); + const endedAt = new Date(w.end); + const durationSec = Math.max( + 0, + Math.round((endedAt.getTime() - startedAt.getTime()) / 1000), + ); + const energyKcal = + typeof w.score.kilojoule === "number" + ? Math.round(w.score.kilojoule * KJ_TO_KCAL) + : null; + + const metadata: Prisma.InputJsonValue = { + whoopWorkoutStrain: w.score.strain, + percentRecorded: w.score.percent_recorded, + ...(w.score.altitude_gain_meter != null + ? { altitudeGainMeter: w.score.altitude_gain_meter } + : {}), + ...(w.score.altitude_change_meter != null + ? { altitudeChangeMeter: w.score.altitude_change_meter } + : {}), + ...(w.score.zone_durations + ? { zoneDurations: w.score.zone_durations } + : {}), + }; + + const row = { + sportType: sportLabel(w.sport_id, w.sport_name), + startedAt, + endedAt, + durationSec, + totalEnergyKcal: energyKcal, + totalDistanceM: w.score.distance_meter ?? null, + avgHeartRate: w.score.average_heart_rate ?? null, + maxHeartRate: w.score.max_heart_rate ?? null, + elevationM: w.score.altitude_gain_meter ?? null, + metadata, + }; + + try { + await prisma.workout.upsert({ + where: { + userId_source_externalId: { userId, source: "WHOOP", externalId: w.id, - sportType: sportLabel(w.sport_id, w.sport_name), - startedAt, - endedAt, - durationSec, - totalEnergyKcal: energyKcal, - totalDistanceM: w.score.distance_meter ?? null, - avgHeartRate: w.score.average_heart_rate ?? null, - maxHeartRate: w.score.max_heart_rate ?? null, - elevationM: w.score.altitude_gain_meter ?? null, - metadata, }, - update: { - sportType: sportLabel(w.sport_id, w.sport_name), - startedAt, - endedAt, - durationSec, - totalEnergyKcal: energyKcal, - totalDistanceM: w.score.distance_meter ?? null, - avgHeartRate: w.score.average_heart_rate ?? null, - maxHeartRate: w.score.max_heart_rate ?? null, - elevationM: w.score.altitude_gain_meter ?? null, - metadata, - }, - }); - imported++; - } catch (err) { - getEvent()?.addWarning(`WHOOP: failed to upsert workout: ${err}`); - } + }, + create: { + userId, + source: "WHOOP", + externalId: w.id, + ...row, + }, + update: row, + }); + return 1; + } catch (err) { + getEvent()?.addWarning(`WHOOP: failed to upsert workout: ${err}`); + return 0; } - - await markSynced(userId); - return imported; } diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index 48ff81fe8..45f2f1c9b 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 } from "@/generated/prisma/client"; +import type { MeasurementType, Prisma } from "@/generated/prisma/client"; import { encrypt, decrypt } from "@/lib/crypto"; import { getEvent } from "@/lib/logging/context"; import { @@ -117,6 +117,61 @@ const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; export const WHOOP_RECOVERY_SLEEP_OVERLAP_MS = 7 * 24 * 60 * 60 * 1000; export const WHOOP_DEFAULT_OVERLAP_MS = 60 * 60 * 1000; // 1 h +/** + * The far-past anchor a `fullSync` walks from. Incremental ticks start at + * `cursor − overlap` (days at most); a fullSync deliberately ignores the + * cursor and re-walks the deep history WHOOP retains (multi-year), which an + * incremental run would never reach. WHOOP's first public data predates this, + * so a fixed anchor covers every realistic account; the client's `next_token` + * pagination + the idempotent upsert keep the re-walk bounded and safe. + */ +export const WHOOP_FULL_SYNC_ANCHOR = new Date("2020-01-01T00:00:00.000Z"); + +/** + * The four pollable WHOOP collections that carry an independent sync cursor. + * Body-measurement is a single profile object (no time-range cursor) and is + * excluded. Each resource advances ONLY its own cursor key so a slow/failing + * collection never drags the incremental window forward for its siblings. + */ +export type WhoopResource = "recovery" | "sleep" | "workout" | "cycle"; + +/** + * Shape of the `WhoopConnection.resourceCursors` JSON map: a partial + * `resource → ISO-8601 last-synced instant`. A missing key (or a legacy null + * column) means the resource has no per-resource cursor yet and falls back to + * the shared `lastSyncedAt`. + */ +type ResourceCursorMap = Partial>; + +function parseResourceCursors(raw: unknown): ResourceCursorMap { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return {}; + return raw as ResourceCursorMap; +} + +/** + * Resolve the incremental cursor for one resource. Prefers the per-resource + * cursor; falls back to the connection's shared `lastSyncedAt` for a legacy + * connection (or a resource that has never synced under the new column) so no + * historical state is lost on the first tick after the migration. Returns null + * when neither exists (the very first sync — `incrementalStart` then anchors 30 + * days back). + */ +export function resolveResourceCursor( + connection: { + resourceCursors?: unknown; + lastSyncedAt?: Date | null; + }, + resource: WhoopResource, +): Date | null { + const map = parseResourceCursors(connection.resourceCursors); + const iso = map[resource]; + if (iso) { + const parsed = new Date(iso); + if (!Number.isNaN(parsed.getTime())) return parsed; + } + return connection.lastSyncedAt ?? null; +} + export interface WhoopTokenInfo { accessToken: string; connection: { id: string; whoopUserId: string }; @@ -200,18 +255,24 @@ export async function getValidToken( } /** - * Compute the incremental `start` for a resource sync. `fullSync` returns - * undefined (the backfill anchor handles the deep history). Otherwise start - * from `lastSyncedAt - overlap`, or 30 days back on the very first incremental - * tick. + * Compute the `start` bound for a resource fetch. + * + * A `fullSync` deliberately ignores the cursor and anchors at + * `WHOOP_FULL_SYNC_ANCHOR` — the deep historical backfill an incremental run + * would never reach. (It previously returned `undefined`; an explicit anchor + * pins the lower bound so the walk is reproducible and the cursor a stalled + * incremental left behind cannot silently narrow a backfill.) + * + * Otherwise it is an incremental tick: start from `cursor − overlap`, or 30 + * days back on the very first sync (no cursor yet). */ export function incrementalStart( - lastSyncedAt: Date | null, + cursor: Date | null, opts: { fullSync?: boolean; overlapMs?: number } = {}, ): Date | undefined { - if (opts.fullSync) return undefined; + if (opts.fullSync) return WHOOP_FULL_SYNC_ANCHOR; const overlap = opts.overlapMs ?? WHOOP_DEFAULT_OVERLAP_MS; - if (lastSyncedAt) return new Date(lastSyncedAt.getTime() - overlap); + if (cursor) return new Date(cursor.getTime() - overlap); return new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); } @@ -316,6 +377,41 @@ 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. + * + * 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. + */ +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; + + 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, + }, + }); +} + /** * Full per-user sync across every WHOOP resource. Webhook-driven syncs enqueue * a single per-resource job; this drives the hourly poll catch-all and the diff --git a/src/lib/whoop/webhook-handler.ts b/src/lib/whoop/webhook-handler.ts index 7b20702cc..45f733688 100644 --- a/src/lib/whoop/webhook-handler.ts +++ b/src/lib/whoop/webhook-handler.ts @@ -203,13 +203,18 @@ export async function processWhoopNotification( } // `*.updated` (creates arrive as updates too). Enqueue the per-user - // resource sync; the worker re-fetches by id (payload carries no data). + // resource sync. The payload carries the resource id so the worker can do a + // targeted fetch-by-id refresh — landing the exact record immediately rather + // than waiting for the next overlap window to re-walk the collection. const boss = getGlobalBoss(); if (!boss) { getEvent()?.addWarning("whoop-webhook: pg-boss not initialised"); return NextResponse.json({ status: "ok" }, { status: 200 }); } - await boss.send(queue, { userId: connection.userId }); + await boss.send(queue, { + userId: connection.userId, + resourceId: resourceId ?? undefined, + }); annotate({ action: { name: "whoop.webhook.enqueued" }, From 7206615213ad5053766b0d7f66f11b01aac3ba50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sun, 14 Jun 2026 17:23:55 +0200 Subject: [PATCH 2/2] 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. --- src/lib/whoop/__tests__/sync.test.ts | 54 +++++++++++----------- src/lib/whoop/sync.ts | 69 ++++++++++++++++++---------- 2 files changed, 72 insertions(+), 51 deletions(-) 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} + `; } /**