From 337f024a87f113260d701d7a7f40184ed8b7f74f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 4 Jun 2026 01:19:35 +0200 Subject: [PATCH] feat(whoop): pg-boss sync workers + self-converging backfill Build the WHOOP sync layer on the W2 client, mirroring the Withings queues. A per-resource sync (recovery/sleep/cycle/workout) pulls since the last cursor with a 24h overlap on recovery/sleep to absorb WHOOP's after-the-fact re-scoring, maps via the W2 mappers, and upserts Measurements with source=WHOOP keyed on (userId,type,source,externalId) so a re-post overwrites in place. Workouts upsert into Workout with per-workout strain + HR zones in metadata and kJ->kcal energy. Token refresh persists BOTH rotated tokens (WHOOP invalidates the prior access and refresh token on every refresh). A self-converging boot backfill enqueues one full-history sync per un-backfilled connection and stamps backfillCompletedAt; the discovery query drops completed accounts, idempotent across reboots. Every new queue is registered in the worker's allQueues + cron schedule (recovery/sleep/workout poll catch-all, cycle poll-only, backfill, oauth-state sweep). IntegrationStatus gains the whoop key + label. Tests: rotating-refresh persistence, idempotent measurement upsert, incremental-window overlap, backfill self-convergence, and a queue-registration inventory guard. --- src/lib/integrations/status.ts | 8 +- src/lib/jobs/__tests__/whoop-backfill.test.ts | 101 ++++++ src/lib/jobs/__tests__/whoop-queues.test.ts | 102 ++++++ src/lib/jobs/reminder-worker.ts | 225 ++++++++++++ src/lib/jobs/whoop-backfill.ts | 122 +++++++ src/lib/jobs/whoop-oauth-state-cleanup.ts | 23 ++ src/lib/whoop/__tests__/sync.test.ts | 243 +++++++++++++ src/lib/whoop/sync-cycle.ts | 60 ++++ src/lib/whoop/sync-recovery.ts | 68 ++++ src/lib/whoop/sync-sleep.ts | 65 ++++ src/lib/whoop/sync-workout.ts | 132 +++++++ src/lib/whoop/sync.ts | 338 ++++++++++++++++++ 12 files changed, 1485 insertions(+), 2 deletions(-) create mode 100644 src/lib/jobs/__tests__/whoop-backfill.test.ts create mode 100644 src/lib/jobs/__tests__/whoop-queues.test.ts create mode 100644 src/lib/jobs/whoop-backfill.ts create mode 100644 src/lib/jobs/whoop-oauth-state-cleanup.ts create mode 100644 src/lib/whoop/__tests__/sync.test.ts create mode 100644 src/lib/whoop/sync-cycle.ts create mode 100644 src/lib/whoop/sync-recovery.ts create mode 100644 src/lib/whoop/sync-sleep.ts create mode 100644 src/lib/whoop/sync-workout.ts create mode 100644 src/lib/whoop/sync.ts diff --git a/src/lib/integrations/status.ts b/src/lib/integrations/status.ts index cff5dd0f7..40a0d8e57 100644 --- a/src/lib/integrations/status.ts +++ b/src/lib/integrations/status.ts @@ -39,7 +39,7 @@ import { auditLog } from "@/lib/auth/audit"; import { getEvent } from "@/lib/logging/context"; import { dispatchNotification } from "@/lib/notifications/dispatcher"; -export type IntegrationKey = "withings" | "moodlog"; +export type IntegrationKey = "withings" | "whoop" | "moodlog"; /** * Failure kinds carried into `recordSyncFailure`. @@ -820,7 +820,11 @@ export function formatAdminAlertPayload(input: AlertInput): { metadata: Record; } { const integrationLabel = - input.integration === "withings" ? "Withings" : "moodLog"; + input.integration === "withings" + ? "Withings" + : input.integration === "whoop" + ? "WHOOP" + : "moodLog"; const subjectLabel = input.subjectLabel ?? input.userId; const { reason: reasonLabel, action: actionLabel } = FAILURE_KIND_COPY[input.kind]; diff --git a/src/lib/jobs/__tests__/whoop-backfill.test.ts b/src/lib/jobs/__tests__/whoop-backfill.test.ts new file mode 100644 index 000000000..e6573ad2d --- /dev/null +++ b/src/lib/jobs/__tests__/whoop-backfill.test.ts @@ -0,0 +1,101 @@ +/** + * v1.11.0 — WHOOP backfill self-convergence tests (mocked). + * - discovery only matches un-backfilled connections; + * - a completed backfill stamps `backfillCompletedAt` so the next discovery + * pass drops the account (idempotent across reboots); + * - the discovery enqueue is singleton-keyed per user. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { prismaMock, bossSend, syncUserWhoop } = vi.hoisted(() => ({ + prismaMock: { + whoopConnection: { + findMany: vi.fn(), + update: vi.fn(), + }, + }, + bossSend: vi.fn(), + syncUserWhoop: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => ({ send: bossSend }), +})); + +vi.mock("@/lib/whoop/sync", () => ({ + syncUserWhoop: (...a: unknown[]) => syncUserWhoop(...a), +})); + +vi.mock("@/lib/logging/context", () => ({ + annotate: () => {}, + getEvent: () => null, +})); + +import { + enqueueBootTimeWhoopBackfill, + runWhoopBackfillForUser, +} from "../whoop-backfill"; + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe("enqueueBootTimeWhoopBackfill — discovery", () => { + it("queries only un-backfilled connections", async () => { + prismaMock.whoopConnection.findMany.mockResolvedValue([ + { userId: "u1" }, + { userId: "u2" }, + ]); + bossSend.mockResolvedValue("job-id"); + + const result = await enqueueBootTimeWhoopBackfill(); + + expect(prismaMock.whoopConnection.findMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: { backfillCompletedAt: null }, + }), + ); + expect(result.enqueued).toBe(2); + // Singleton-keyed per user so a fast restart never double-enqueues. + expect(bossSend).toHaveBeenCalledWith( + "whoop-backfill", + expect.objectContaining({ userId: "u1" }), + expect.objectContaining({ singletonKey: "whoop-backfill|u1" }), + ); + }); + + it("self-converges: no un-backfilled connections → nothing enqueued", async () => { + prismaMock.whoopConnection.findMany.mockResolvedValue([]); + + const result = await enqueueBootTimeWhoopBackfill(); + + expect(result.enqueued).toBe(0); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("never throws — surfaces a discovery error through the result value", async () => { + prismaMock.whoopConnection.findMany.mockRejectedValue(new Error("db down")); + + const result = await enqueueBootTimeWhoopBackfill(); + + expect(result.error).toBe("db down"); + expect(result.enqueued).toBe(0); + }); +}); + +describe("runWhoopBackfillForUser", () => { + it("runs a full sync and stamps backfillCompletedAt", async () => { + syncUserWhoop.mockResolvedValue(123); + prismaMock.whoopConnection.update.mockResolvedValue({}); + + const { imported } = await runWhoopBackfillForUser("u1"); + + expect(imported).toBe(123); + expect(syncUserWhoop).toHaveBeenCalledWith("u1", { fullSync: true }); + const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0]; + expect(updateArg.where).toEqual({ userId: "u1" }); + expect(updateArg.data.backfillCompletedAt).toBeInstanceOf(Date); + }); +}); diff --git a/src/lib/jobs/__tests__/whoop-queues.test.ts b/src/lib/jobs/__tests__/whoop-queues.test.ts new file mode 100644 index 000000000..ded3c83eb --- /dev/null +++ b/src/lib/jobs/__tests__/whoop-queues.test.ts @@ -0,0 +1,102 @@ +/** + * v1.11.0 — pg-boss queue registration for the WHOOP sync layer. + * + * The reminder-worker imports heavy infrastructure (pg-boss, Prisma adapter, + * notification dispatchers, …) so we don't boot the whole worker here. Instead + * we read the source as text and assert the queue constants + cron schedules + + * handler wiring + allQueues + boot-enqueue are present — the fast guard that + * catches the v1.4.37 dead-queue class (a queue declared but never registered + * in `allQueues`, which silently never drains). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts"); +const source = readFileSync(REMINDER_WORKER_PATH, "utf8"); + +const WHOOP_QUEUE_CONSTS = [ + "WHOOP_RECOVERY_SYNC_QUEUE", + "WHOOP_SLEEP_SYNC_QUEUE", + "WHOOP_WORKOUT_SYNC_QUEUE", + "WHOOP_CYCLE_SYNC_QUEUE", + "WHOOP_BACKFILL_QUEUE", + "WHOOP_OAUTH_STATE_CLEANUP_QUEUE", +] as const; + +describe("reminder-worker — WHOOP sync queues", () => { + it("declares every WHOOP sync queue constant", () => { + expect(source).toMatch( + /WHOOP_RECOVERY_SYNC_QUEUE\s*=\s*["']whoop-recovery-sync["']/, + ); + expect(source).toMatch( + /WHOOP_SLEEP_SYNC_QUEUE\s*=\s*["']whoop-sleep-sync["']/, + ); + expect(source).toMatch( + /WHOOP_WORKOUT_SYNC_QUEUE\s*=\s*["']whoop-workout-sync["']/, + ); + expect(source).toMatch( + /WHOOP_CYCLE_SYNC_QUEUE\s*=\s*["']whoop-cycle-sync["']/, + ); + }); + + it("registers EVERY WHOOP queue in allQueues (v1.4.37 dead-queue guard)", () => { + // Isolate the allQueues array body so the assertion can't be satisfied by + // an incidental mention elsewhere in the file. + const block = /const allQueues = \[([\s\S]*?)\];/.exec(source); + expect(block).not.toBeNull(); + const body = block![1]!; + for (const q of WHOOP_QUEUE_CONSTS) { + expect(body).toContain(q); + } + }); + + it("schedules the four poll crons + the daily oauth-state sweep", () => { + expect(source).toMatch( + /\[WHOOP_RECOVERY_SYNC_QUEUE,\s*WHOOP_RECOVERY_SYNC_CRON\]/, + ); + expect(source).toMatch( + /\[WHOOP_SLEEP_SYNC_QUEUE,\s*WHOOP_SLEEP_SYNC_CRON\]/, + ); + expect(source).toMatch( + /\[WHOOP_WORKOUT_SYNC_QUEUE,\s*WHOOP_WORKOUT_SYNC_CRON\]/, + ); + expect(source).toMatch( + /\[WHOOP_CYCLE_SYNC_QUEUE,\s*WHOOP_CYCLE_SYNC_CRON\]/, + ); + expect(source).toMatch( + /\[WHOOP_OAUTH_STATE_CLEANUP_QUEUE,\s*WHOOP_OAUTH_STATE_CLEANUP_CRON\]/, + ); + }); + + it("registers a boss.work handler for every WHOOP sync queue", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,160}WHOOP_RECOVERY_SYNC_QUEUE[\s\S]{0,160}handleWhoopRecoverySync/, + ); + expect(source).toMatch( + /boss\.work[\s\S]{0,160}WHOOP_SLEEP_SYNC_QUEUE[\s\S]{0,160}handleWhoopSleepSync/, + ); + expect(source).toMatch( + /boss\.work[\s\S]{0,160}WHOOP_WORKOUT_SYNC_QUEUE[\s\S]{0,160}handleWhoopWorkoutSync/, + ); + expect(source).toMatch( + /boss\.work[\s\S]{0,160}WHOOP_CYCLE_SYNC_QUEUE[\s\S]{0,160}handleWhoopCycleSync/, + ); + expect(source).toMatch( + /boss\.work[\s\S]{0,200}WHOOP_BACKFILL_QUEUE[\s\S]{0,200}runWhoopBackfillForUser/, + ); + }); + + it("wires the self-converging WHOOP backfill boot discovery", () => { + expect(source).toMatch(/enqueueBootTimeWhoopBackfill\(\)/); + }); + + it("imports the WHOOP per-resource sync routines", () => { + expect(source).toMatch( + /from\s*["']@\/lib\/whoop\/sync-recovery["']/, + ); + expect(source).toMatch(/from\s*["']@\/lib\/whoop\/sync-cycle["']/); + expect(source).toMatch(/from\s*["']@\/lib\/whoop\/sync-workout["']/); + }); +}); diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 69c6e3c3a..3656e28f8 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -20,6 +20,18 @@ import { import { syncUserMeasurements } from "@/lib/withings/sync"; import { syncUserActivity } from "@/lib/withings/sync-activity"; import { syncUserSleep } from "@/lib/withings/sync-sleep"; +import { syncUserRecovery } from "@/lib/whoop/sync-recovery"; +import { syncUserSleep as syncWhoopSleep } from "@/lib/whoop/sync-sleep"; +import { syncUserCycle } from "@/lib/whoop/sync-cycle"; +import { syncUserWorkout } from "@/lib/whoop/sync-workout"; +import { + WHOOP_BACKFILL_QUEUE, + WHOOP_BACKFILL_CONCURRENCY, + runWhoopBackfillForUser, + enqueueBootTimeWhoopBackfill, + type WhoopBackfillPayload, +} from "@/lib/jobs/whoop-backfill"; +import { cleanupExpiredWhoopOAuthStates } from "@/lib/jobs/whoop-oauth-state-cleanup"; import { generateGeneralStatusForUser } from "@/lib/insights/general-status"; import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status"; import { generateWeightStatusForUser } from "@/lib/insights/weight-status"; @@ -259,6 +271,24 @@ const AUDIT_LOG_CLEANUP_CRON = "15 3 * * *"; // daily at 03:15 (Europe/Berlin) // Withings approval tab without bouncing back to the callback URL. const WITHINGS_OAUTH_STATE_CLEANUP_QUEUE = "withings-oauth-state-cleanup"; const WITHINGS_OAUTH_STATE_CLEANUP_CRON = "20 3 * * *"; +// v1.11.0 — WHOOP sync queues. Webhook-primary + cron-safety-net, mirroring +// the Withings activity/sleep crons. Recovery / sleep / workout each have a +// WHOOP webhook (`*.updated`) that enqueues the matching per-resource job; the +// crons below are the catch-net for dropped deliveries. Cycle has NO webhook, +// so its cron is the only driver. Minutes are staggered off the Withings crons +// (:00/:15) to spread DB load. +const WHOOP_RECOVERY_SYNC_QUEUE = "whoop-recovery-sync"; +const WHOOP_RECOVERY_SYNC_CRON = "5 * * * *"; // every hour at :05 +const WHOOP_SLEEP_SYNC_QUEUE = "whoop-sleep-sync"; +const WHOOP_SLEEP_SYNC_CRON = "20 * * * *"; // every hour at :20 +const WHOOP_WORKOUT_SYNC_QUEUE = "whoop-workout-sync"; +const WHOOP_WORKOUT_SYNC_CRON = "35 * * * *"; // every hour at :35 +const WHOOP_CYCLE_SYNC_QUEUE = "whoop-cycle-sync"; +const WHOOP_CYCLE_SYNC_CRON = "50 * * * *"; // every hour at :50 (poll-only) +// v1.11.0 — daily sweep for the WHOOP OAuth state ledger. Slots at 03:22, +// next to the Withings sweep (03:20), inside the maintenance window. +const WHOOP_OAUTH_STATE_CLEANUP_QUEUE = "whoop-oauth-state-cleanup"; +const WHOOP_OAUTH_STATE_CLEANUP_CRON = "22 3 * * *"; const OFFHOST_BACKUP_QUEUE = "data-backup-offhost"; // 02:30 Europe/Berlin — runs after audit-log/idempotency cleanups so old // rows are gone before they're snapshotted, but before the existing @@ -1070,6 +1100,89 @@ async function handleWithingsSleepSync(jobs: Job[]) { }); } +/** + * v1.11.0 — WHOOP per-resource sync payload. Two enqueue paths feed each + * WHOOP sync queue: + * + * 1. Webhook (`recovery.updated` / `sleep.updated` / `workout.updated`) — + * payload carries `userId`, the handler syncs that one user. + * 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. + */ +interface WhoopSyncPayload { + userId?: string; +} + +/** + * 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. + */ +async function runWhoopResourceSync( + taskName: string, + jobs: Job[], + syncFn: (userId: string) => Promise, +): Promise { + await withBackgroundEvent(taskName, async (evt) => { + const prisma = getWorkerPrisma(); + try { + const targets: Array<{ userId: string }> = []; + for (const job of jobs) { + if (job.data?.userId) targets.push({ userId: job.data.userId }); + } + if (targets.length === 0) { + const connections = await prisma.whoopConnection.findMany({ + select: { userId: true }, + }); + targets.push(...connections); + } + if (targets.length === 0) return; + + let usersSynced = 0; + let measurementsImported = 0; + for (const { userId } of targets) { + try { + measurementsImported += await syncFn(userId); + usersSynced++; + } catch (err) { + evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); + } + } + + evt.setBackground({ + task_name: taskName, + result: { + users_synced: usersSynced, + total: targets.length, + measurements_imported: measurementsImported, + }, + }); + } catch (err) { + evt.setError(err); + recordError(); + throw err; + } + }); +} + +function handleWhoopRecoverySync(jobs: Job[]) { + return runWhoopResourceSync("job.whoop_recovery_sync", jobs, syncUserRecovery); +} + +function handleWhoopSleepSync(jobs: Job[]) { + return runWhoopResourceSync("job.whoop_sleep_sync", jobs, syncWhoopSleep); +} + +function handleWhoopWorkoutSync(jobs: Job[]) { + return runWhoopResourceSync("job.whoop_workout_sync", jobs, syncUserWorkout); +} + +function handleWhoopCycleSync(jobs: Job[]) { + return runWhoopResourceSync("job.whoop_cycle_sync", jobs, syncUserCycle); +} + async function handleGeneralStatusGenerate(jobs: Job[]) { void jobs; await withBackgroundEvent("job.insights.general", async (evt) => { @@ -1562,6 +1675,25 @@ async function handleWithingsOAuthStateCleanup( }); } +interface WhoopOAuthStateCleanupPayload { + triggeredAt?: string; +} + +async function handleWhoopOAuthStateCleanup( + jobs: Job[], +) { + void jobs; + await withBackgroundEvent("job.whoop_oauth_state_cleanup", async (evt) => { + const p = getWorkerPrisma(); + try { + const deleted = await cleanupExpiredWhoopOAuthStates(p); + evt.addMeta("whoop_oauth_state_cleanup_deleted", deleted); + } catch (err) { + evt.addWarning(`whoop-oauth-state-cleanup failed: ${err}`); + } + }); +} + async function handleHostMetricSample(jobs: Job[]) { void jobs; await withBackgroundEvent("job.host_metric_sample", async (evt) => { @@ -2065,6 +2197,21 @@ export async function startReminderWorker() { // crons; without this entry the schedule below silently no-ops // and abandoned rows pile up. WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, + // v1.11.0 — WHOOP sync queues. Webhook-primary + cron-safety-net for + // recovery / sleep / workout; cycle is poll-only (no WHOOP webhook). + // Every queue MUST be registered here or pg-boss never provisions it and + // both the webhook enqueue AND the cron schedule below silently no-op (the + // v1.4.37 dead-queue class). + WHOOP_RECOVERY_SYNC_QUEUE, + WHOOP_SLEEP_SYNC_QUEUE, + WHOOP_WORKOUT_SYNC_QUEUE, + WHOOP_CYCLE_SYNC_QUEUE, + // v1.11.0 — self-converging boot backfill for newly connected WHOOP + // accounts. Discovery enqueues one full-history sync per un-backfilled + // connection; idempotent across reboots. + WHOOP_BACKFILL_QUEUE, + // v1.11.0 — daily sweep for the WHOOP OAuth state ledger. + WHOOP_OAUTH_STATE_CLEANUP_QUEUE, OFFHOST_BACKUP_QUEUE, HOST_METRIC_QUEUE, FEEDBACK_AGGREGATOR_QUEUE, @@ -2219,6 +2366,15 @@ export async function startReminderWorker() { [IDEMPOTENCY_CLEANUP_QUEUE, IDEMPOTENCY_CLEANUP_CRON], [AUDIT_LOG_CLEANUP_QUEUE, AUDIT_LOG_CLEANUP_CRON], [WITHINGS_OAUTH_STATE_CLEANUP_QUEUE, WITHINGS_OAUTH_STATE_CLEANUP_CRON], + // v1.11.0 — WHOOP poll-fallback crons. Recovery/sleep/workout catch + // dropped webhooks; cycle is the sole driver (no webhook). Staggered off + // the Withings crons so the hourly ticks don't pile up on one boss poll. + [WHOOP_RECOVERY_SYNC_QUEUE, WHOOP_RECOVERY_SYNC_CRON], + [WHOOP_SLEEP_SYNC_QUEUE, WHOOP_SLEEP_SYNC_CRON], + [WHOOP_WORKOUT_SYNC_QUEUE, WHOOP_WORKOUT_SYNC_CRON], + [WHOOP_CYCLE_SYNC_QUEUE, WHOOP_CYCLE_SYNC_CRON], + // v1.11.0 — daily 03:22 Europe/Berlin prune for expired WHOOP OAuth states. + [WHOOP_OAUTH_STATE_CLEANUP_QUEUE, WHOOP_OAUTH_STATE_CLEANUP_CRON], [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON], [HOST_METRIC_QUEUE, HOST_METRIC_CRON], [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON], @@ -2300,6 +2456,51 @@ export async function startReminderWorker() { { localConcurrency: 1 }, handleWithingsSleepSync, ); + // v1.11.0 — WHOOP per-resource sync handlers. Webhook-driven per-user + + // cron full-iteration. Serial concurrency so a backfill-heavy tick never + // crowds the request pool and stays inside WHOOP's 100 req/min app cap. + await boss.work( + WHOOP_RECOVERY_SYNC_QUEUE, + { localConcurrency: 1 }, + handleWhoopRecoverySync, + ); + await boss.work( + WHOOP_SLEEP_SYNC_QUEUE, + { localConcurrency: 1 }, + handleWhoopSleepSync, + ); + await boss.work( + WHOOP_WORKOUT_SYNC_QUEUE, + { localConcurrency: 1 }, + handleWhoopWorkoutSync, + ); + await boss.work( + WHOOP_CYCLE_SYNC_QUEUE, + { localConcurrency: 1 }, + handleWhoopCycleSync, + ); + // v1.11.0 — self-converging WHOOP backfill. The boot enqueue below sends one + // full-history sync per un-backfilled connection; this handler runs it and + // stamps `backfillCompletedAt` so the discovery query drops the account. + await boss.work( + WHOOP_BACKFILL_QUEUE, + { localConcurrency: WHOOP_BACKFILL_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + const { imported } = await runWhoopBackfillForUser(userId); + workerLog( + "info", + `[whoop-backfill] user=${userId} imported=${imported}`, + ); + } + }, + ); + await boss.work( + WHOOP_OAUTH_STATE_CLEANUP_QUEUE, + { localConcurrency: 1 }, + handleWhoopOAuthStateCleanup, + ); await boss.work( GENERAL_STATUS_QUEUE, { localConcurrency: 1 }, @@ -2915,6 +3116,30 @@ export async function startReminderWorker() { ); } + // v1.11.0 — fire-and-forget boot discovery for the WHOOP backfill. Finds + // every WHOOP connection not yet backfilled and enqueues one full-history + // sync per account. Idempotent across reboots: a completed backfill stamps + // `backfillCompletedAt`, dropping the connection from the discovery set. + // Errors come back through the helper's result value — the worker boot never + // fails because of a backfill miss. + try { + const { enqueued, skipped, error } = await enqueueBootTimeWhoopBackfill(); + if (error) { + workerLog("error", `[whoop-backfill] boot discovery failed: ${error}`); + } else { + workerLog( + "info", + `[whoop-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[whoop-backfill] boot discovery threw an unexpected error", + err, + ); + } + // v1.7.0 — fire-and-forget boot discovery for the daily-mean // consolidation pass. Finds every user holding live per-sample // high-frequency mean-type rows and enqueues one job per account. diff --git a/src/lib/jobs/whoop-backfill.ts b/src/lib/jobs/whoop-backfill.ts new file mode 100644 index 000000000..fb9e59cc4 --- /dev/null +++ b/src/lib/jobs/whoop-backfill.ts @@ -0,0 +1,122 @@ +/** + * v1.11.0 — pg-boss queue + boot-time self-converging backfill for newly + * connected WHOOP accounts. Modelled on the `rollup-full-backfill` / + * step-consolidation boot pattern: a discovery query enqueues one job per + * connection that has NOT yet been backfilled, the per-user handler runs a + * full-history sync, and the pass is idempotent across reboots (the discovery + * predicate `backfill_completed_at IS NULL` drops a connection once its + * backfill finishes). + * + * The full sync walks each collection from the far-past anchor to now via + * `next_token` (the client caps `limit` at 25 and stops on an empty cursor). + * Multi-year history is low-thousands of requests — well under WHOOP's + * 10 000 req/day app cap. + * + * The queue name MUST be registered in `allQueues` in + * `src/lib/jobs/reminder-worker.ts` or pg-boss never provisions it and the + * boot enqueue silently never drains (the v1.4.37 dead-queue class). + */ +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { syncUserWhoop } from "@/lib/whoop/sync"; + +export const WHOOP_BACKFILL_QUEUE = "whoop-backfill"; + +/** + * Serial concurrency — a backfill walks years of history for one account and + * is rate-bounded by WHOOP's 100 req/min cap; concurrency-1 keeps it from + * crowding the request pool, matching `ROLLUP_FULL_BACKFILL_CONCURRENCY`. + */ +export const WHOOP_BACKFILL_CONCURRENCY = 1; + +export interface WhoopBackfillPayload { + userId: string; + enqueuedAt: string; +} + +/** + * Per-user backfill handler. Runs a full-history sync for one account and + * stamps `backfillCompletedAt` so the discovery query drops it. Idempotent: + * the per-resource upserts are key-stable, so a re-run (e.g. a reboot mid-walk) + * overwrites rather than duplicating. + */ +export async function runWhoopBackfillForUser( + userId: string, +): Promise<{ imported: number }> { + const imported = await syncUserWhoop(userId, { fullSync: true }); + + await prisma.whoopConnection.update({ + where: { userId }, + data: { backfillCompletedAt: new Date() }, + }); + + annotate({ + action: { + name: "whoop.backfill.complete", + details: { imported }, + }, + }); + return { imported }; +} + +/** + * Boot-time discovery. Finds every WHOOP connection not yet backfilled + * (`backfill_completed_at IS NULL`) and enqueues one backfill job per account. + * + * Idempotent across reboots: once a connection's backfill completes, + * `backfillCompletedAt` is set and the predicate drops it from the discovery + * set. pg-boss `singletonKey` coalesces duplicate sends so a fast restart + * while a job is queued doesn't double up. + * + * Best-effort: errors are returned through the result value so the worker boot + * never fails because of a backfill miss. + */ +export async function enqueueBootTimeWhoopBackfill(): Promise<{ + enqueued: number; + skipped: number; + error: string | null; +}> { + const boss = getGlobalBoss(); + if (!boss) { + return { enqueued: 0, skipped: 0, error: null }; + } + + try { + const connections = await prisma.whoopConnection.findMany({ + where: { backfillCompletedAt: null }, + select: { userId: true }, + }); + + if (connections.length === 0) { + return { enqueued: 0, skipped: 0, error: null }; + } + + let enqueued = 0; + let skipped = 0; + for (const { userId } of connections) { + const payload: WhoopBackfillPayload = { + userId, + enqueuedAt: new Date().toISOString(), + }; + const jobId = await boss.send(WHOOP_BACKFILL_QUEUE, payload, { + retryLimit: 3, + retryDelay: 60, + retryBackoff: true, + singletonKey: `whoop-backfill|${userId}`, + }); + if (jobId) { + enqueued += 1; + } else { + skipped += 1; + } + } + return { enqueued, skipped, error: null }; + } catch (err) { + return { + enqueued: 0, + skipped: 0, + error: err instanceof Error ? err.message : String(err), + }; + } +} diff --git a/src/lib/jobs/whoop-oauth-state-cleanup.ts b/src/lib/jobs/whoop-oauth-state-cleanup.ts new file mode 100644 index 000000000..62d2e2f0d --- /dev/null +++ b/src/lib/jobs/whoop-oauth-state-cleanup.ts @@ -0,0 +1,23 @@ +/** + * v1.11.0 — daily cleanup for the `whoop_oauth_states` table. + * + * Mirrors `withings-oauth-state-cleanup`. The OAuth state ledger is single-use + * — every happy-path + error-path branch in `whoop/callback` consumes its row, + * so an abandoned row only lingers when the user closed the WHOOP approval tab + * without bouncing back to the callback URL. The cookie + row TTL is 10 + * minutes, so once the timestamp blows the row is dead weight. + * + * Idempotent: a re-run within the same window matches zero rows the second time + * because the first pass deleted everything older than `now()`. + */ +import type { PrismaClient } from "@/generated/prisma/client"; + +export async function cleanupExpiredWhoopOAuthStates( + prisma: PrismaClient, + now: Date = new Date(), +): Promise { + const { count } = await prisma.whoopOAuthState.deleteMany({ + where: { expiresAt: { lt: now } }, + }); + return count; +} diff --git a/src/lib/whoop/__tests__/sync.test.ts b/src/lib/whoop/__tests__/sync.test.ts new file mode 100644 index 000000000..eb9768f64 --- /dev/null +++ b/src/lib/whoop/__tests__/sync.test.ts @@ -0,0 +1,243 @@ +/** + * v1.11.0 — WHOOP sync layer tests (mocked). Covers: + * - the rotating refresh token persists BOTH new tokens; + * - an incremental recovery sync maps + upserts idempotently (a re-post with + * the same externalId routes through the upsert key, never a duplicate); + * - the incremental window helper picks the right overlap. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +// ── Module mocks ──────────────────────────────────────────────── +const { + prismaMock, + refreshAccessTokenMock, + fetchRecoveriesMock, + recordSyncFailure, + recordSyncSuccess, + isReauthRequired, +} = vi.hoisted(() => ({ + prismaMock: { + whoopConnection: { + findUnique: vi.fn(), + update: vi.fn(), + findMany: vi.fn(), + }, + measurement: { + upsert: vi.fn(), + }, + }, + refreshAccessTokenMock: vi.fn(), + fetchRecoveriesMock: vi.fn(), + recordSyncFailure: vi.fn(async () => {}), + recordSyncSuccess: vi.fn(async () => {}), + isReauthRequired: vi.fn(async () => false), +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +vi.mock("@/lib/crypto", () => ({ + encrypt: (s: string) => `enc(${s})`, + decrypt: (s: string) => s.replace(/^enc\(|\)$/g, ""), +})); + +vi.mock("../client", async (orig) => { + const actual = await orig(); + return { + ...actual, + refreshAccessToken: (...a: unknown[]) => refreshAccessTokenMock(...a), + fetchRecoveries: (...a: unknown[]) => fetchRecoveriesMock(...a), + }; +}); + +vi.mock("../credentials", () => ({ + getUserWhoopCredentials: vi.fn(async () => ({ + clientId: "cid", + clientSecret: "csecret", + })), +})); + +vi.mock("@/lib/integrations/status", () => ({ + recordSyncFailure: (...a: unknown[]) => recordSyncFailure(...a), + recordSyncSuccess: (...a: unknown[]) => recordSyncSuccess(...a), + isReauthRequired: (...a: unknown[]) => isReauthRequired(...a), +})); + +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: ( + rows: Array<{ type: string; measuredAt: Date }>, + ) => rows.map((r) => ({ type: r.type, measuredAt: r.measuredAt })), + recomputeBucketsForMeasurement: vi.fn(async () => {}), +})); + +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); + +vi.mock("@/lib/logging/context", () => ({ + getEvent: () => null, + annotate: () => {}, +})); + +import { + getValidToken, + incrementalStart, + upsertWhoopMeasurements, + WHOOP_DEFAULT_OVERLAP_MS, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, +} from "../sync"; + +beforeEach(() => { + vi.clearAllMocks(); + isReauthRequired.mockResolvedValue(false); +}); + +afterEach(() => { + vi.clearAllMocks(); +}); + +describe("getValidToken — rotating refresh", () => { + it("persists BOTH the new access AND refresh token on refresh", async () => { + prismaMock.whoopConnection.findUnique.mockResolvedValue({ + id: "conn1", + whoopUserId: "42", + accessToken: "enc(old-access)", + refreshToken: "enc(old-refresh)", + // Expired (past) so the refresh path fires. + tokenExpiresAt: new Date(Date.now() - 1000), + }); + refreshAccessTokenMock.mockResolvedValue({ + access_token: "new-access", + refresh_token: "new-refresh", + expires_in: 3600, + }); + prismaMock.whoopConnection.update.mockResolvedValue({}); + + const result = await getValidToken("user1"); + + expect(result?.accessToken).toBe("new-access"); + expect(refreshAccessTokenMock).toHaveBeenCalledWith("old-refresh", { + clientId: "cid", + clientSecret: "csecret", + }); + const updateArg = prismaMock.whoopConnection.update.mock.calls[0]![0]; + expect(updateArg.data.accessToken).toBe("enc(new-access)"); + expect(updateArg.data.refreshToken).toBe("enc(new-refresh)"); + }); + + it("returns the stored token without refresh when not near expiry", async () => { + prismaMock.whoopConnection.findUnique.mockResolvedValue({ + id: "conn1", + whoopUserId: "42", + accessToken: "enc(live-access)", + refreshToken: "enc(live-refresh)", + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + }); + + const result = await getValidToken("user1"); + + expect(result?.accessToken).toBe("live-access"); + expect(refreshAccessTokenMock).not.toHaveBeenCalled(); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + }); + + it("records a reauth failure when credentials are missing on refresh", async () => { + const { getUserWhoopCredentials } = await import("../credentials"); + (getUserWhoopCredentials as ReturnType).mockResolvedValueOnce( + null, + ); + prismaMock.whoopConnection.findUnique.mockResolvedValue({ + id: "conn1", + whoopUserId: "42", + accessToken: "enc(a)", + refreshToken: "enc(r)", + tokenExpiresAt: new Date(Date.now() - 1000), + }); + + const result = await getValidToken("user1"); + + expect(result).toBeNull(); + expect(recordSyncFailure).toHaveBeenCalledWith( + expect.objectContaining({ integration: "whoop", kind: "reauth_required" }), + ); + }); +}); + +describe("incrementalStart", () => { + it("returns undefined for a full sync", () => { + expect(incrementalStart(new Date(), { fullSync: true })).toBeUndefined(); + }); + + it("subtracts the overlap from lastSyncedAt", () => { + const last = new Date("2026-06-01T12:00:00Z"); + const got = incrementalStart(last, { + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + }); + expect(got!.getTime()).toBe( + last.getTime() - WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + ); + }); + + it("defaults to the 1 h overlap when none supplied", () => { + const last = new Date("2026-06-01T12:00:00Z"); + const got = incrementalStart(last); + expect(got!.getTime()).toBe(last.getTime() - WHOOP_DEFAULT_OVERLAP_MS); + }); +}); + +describe("upsertWhoopMeasurements — idempotent upsert", () => { + it("routes each reading through the (userId,type,source,externalId) key", async () => { + prismaMock.measurement.upsert.mockResolvedValue({}); + + const n = await upsertWhoopMeasurements("user1", [ + { + type: "RECOVERY_SCORE", + value: 71, + unit: "score", + measuredAt: new Date("2026-06-01T08:00:00Z"), + externalId: "sleep-uuid:recovery", + }, + { + type: "HRV_RMSSD", + value: 64, + unit: "ms", + measuredAt: new Date("2026-06-01T08:00:00Z"), + externalId: "sleep-uuid:hrv_rmssd", + }, + ]); + + expect(n).toBe(2); + const firstWhere = + prismaMock.measurement.upsert.mock.calls[0]![0].where + .userId_type_source_externalId; + expect(firstWhere).toEqual({ + userId: "user1", + type: "RECOVERY_SCORE", + source: "WHOOP", + externalId: "sleep-uuid:recovery", + }); + }); + + it("a re-post with the same externalId is one upsert, not two rows", async () => { + prismaMock.measurement.upsert.mockResolvedValue({}); + const reading = { + type: "RECOVERY_SCORE", + value: 60, + unit: "score", + measuredAt: new Date("2026-06-01T08:00:00Z"), + externalId: "sleep-uuid:recovery", + }; + + await upsertWhoopMeasurements("user1", [reading]); + await upsertWhoopMeasurements("user1", [{ ...reading, value: 75 }]); + + // Two upsert calls, both against the SAME key — the DB collapses them. + expect(prismaMock.measurement.upsert).toHaveBeenCalledTimes(2); + const a = + prismaMock.measurement.upsert.mock.calls[0]![0].where + .userId_type_source_externalId; + const b = + prismaMock.measurement.upsert.mock.calls[1]![0].where + .userId_type_source_externalId; + expect(a).toEqual(b); + }); +}); diff --git a/src/lib/whoop/sync-cycle.ts b/src/lib/whoop/sync-cycle.ts new file mode 100644 index 000000000..3eb585f17 --- /dev/null +++ b/src/lib/whoop/sync-cycle.ts @@ -0,0 +1,60 @@ +/** + * WHOOP cycle (day) sync. There is NO webhook for cycles — this is poll-only, + * driven by the hourly fallback cron. Maps each scored cycle via `mapCycle` + * (DAY_STRAIN + ENERGY_EXPENDITURE_KJ) and upserts as `source = WHOOP`. + * + * Cycle id is an int64 (not a UUID); externalId = `cycle::`. + */ +import { fetchCycles, mapCycle } from "./client"; +import { + getValidToken, + incrementalStart, + markSynced, + recordWhoopSyncFailure, + upsertWhoopMeasurements, + type WhoopMeasurementUpsert, +} from "./sync"; +import { prisma } from "@/lib/db"; + +export async function syncUserCycle( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + const connection = await prisma.whoopConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true }, + }); + if (!connection) return 0; + + const start = incrementalStart(connection.lastSyncedAt, { + fullSync: opts.fullSync, + }); + + let records: Awaited>; + try { + records = await fetchCycles(tokenInfo.accessToken, { start }); + } catch (err) { + await recordWhoopSyncFailure(userId, err); + throw err; + } + + const readings: WhoopMeasurementUpsert[] = []; + for (const c of records) { + for (const m of mapCycle(c)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: `cycle:${c.id}:${m.fieldTag}`, + }); + } + } + + const imported = await upsertWhoopMeasurements(userId, readings); + await markSynced(userId); + return imported; +} diff --git a/src/lib/whoop/sync-recovery.ts b/src/lib/whoop/sync-recovery.ts new file mode 100644 index 000000000..7c70cf375 --- /dev/null +++ b/src/lib/whoop/sync-recovery.ts @@ -0,0 +1,68 @@ +/** + * WHOOP recovery sync. Fetches recovery records since the incremental cursor + * (24 h overlap to absorb WHOOP's after-the-fact re-scoring), maps each scored + * record via `mapRecovery`, and upserts the readings as `source = WHOOP`. + * + * Each recovery object yields several Measurement rows (recovery-score, RMSSD, + * RHR, SpO2, skin-temp), disambiguated by the field-tag in the externalId: + * `:`. The recovery record is keyed off its associated + * 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 { + getValidToken, + incrementalStart, + markSynced, + recordWhoopSyncFailure, + upsertWhoopMeasurements, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + type WhoopMeasurementUpsert, +} from "./sync"; +import { prisma } from "@/lib/db"; + +export async function syncUserRecovery( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + const connection = await prisma.whoopConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true }, + }); + if (!connection) return 0; + + const start = incrementalStart(connection.lastSyncedAt, { + fullSync: opts.fullSync, + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + }); + + let records: Awaited>; + try { + records = await fetchRecoveries(tokenInfo.accessToken, { start }); + } catch (err) { + await recordWhoopSyncFailure(userId, err); + throw err; + } + + const readings: WhoopMeasurementUpsert[] = []; + for (const r of records) { + // `sleep_id` is the stable v2 anchor for the recovery record. + const anchor = r.sleep_id; + for (const m of mapRecovery(r)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: `${anchor}:${m.fieldTag}`, + }); + } + } + + const imported = await upsertWhoopMeasurements(userId, readings); + await markSynced(userId); + return imported; +} diff --git a/src/lib/whoop/sync-sleep.ts b/src/lib/whoop/sync-sleep.ts new file mode 100644 index 000000000..d3d1676b3 --- /dev/null +++ b/src/lib/whoop/sync-sleep.ts @@ -0,0 +1,65 @@ +/** + * WHOOP sleep sync. Fetches sleep activity records since the incremental + * cursor (24 h overlap for the re-score lag), maps each scored record via + * `mapSleep` (per-stage SLEEP_DURATION rows, SLEEP_NEED, the SLEEP_* + * percentages, RESPIRATORY_RATE), and upserts as `source = WHOOP`. + * + * 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 { + getValidToken, + incrementalStart, + markSynced, + recordWhoopSyncFailure, + upsertWhoopMeasurements, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + type WhoopMeasurementUpsert, +} from "./sync"; +import { prisma } from "@/lib/db"; + +export async function syncUserSleep( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + const connection = await prisma.whoopConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true }, + }); + if (!connection) return 0; + + const start = incrementalStart(connection.lastSyncedAt, { + fullSync: opts.fullSync, + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + }); + + let records: Awaited>; + try { + records = await fetchSleeps(tokenInfo.accessToken, { start }); + } catch (err) { + await recordWhoopSyncFailure(userId, err); + throw err; + } + + const readings: WhoopMeasurementUpsert[] = []; + for (const s of records) { + for (const m of mapSleep(s)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: `${s.id}:${m.fieldTag}`, + sleepStage: m.sleepStage ?? null, + }); + } + } + + const imported = await upsertWhoopMeasurements(userId, readings); + await markSynced(userId); + return imported; +} diff --git a/src/lib/whoop/sync-workout.ts b/src/lib/whoop/sync-workout.ts new file mode 100644 index 000000000..9c044160f --- /dev/null +++ b/src/lib/whoop/sync-workout.ts @@ -0,0 +1,132 @@ +/** + * WHOOP workout sync. Fetches workout activities since the incremental cursor + * and upserts each into the `Workout` table as `source = WHOOP`, keyed on + * `(userId, source, externalId)` so a re-score overwrites in place. + * + * Per-workout strain (`WORKOUT_STRAIN`), HR-zone durations, `percent_recorded`, + * and altitude live in `Workout.metadata` (tied to the workout row) rather than + * as free-floating Measurements — a phantom strain Measurement would survive + * if the workout were later de-duped away by the read-time picker. Energy is + * converted kJ→kcal via `KJ_TO_KCAL` for `totalEnergyKcal`. + * + * A WHOOP run and the same run via Apple Health remain two distinct `Workout` + * 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 { + getValidToken, + incrementalStart, + markSynced, + recordWhoopSyncFailure, +} from "./sync"; +import { prisma } from "@/lib/db"; +import { getEvent } from "@/lib/logging/context"; +import type { Prisma } from "@/generated/prisma/client"; + +/** WHOOP reports a numeric `sport_id`; fall back to a generic label. */ +function sportLabel(sportId: number | undefined, sportName?: string): string { + if (sportName) return sportName; + if (typeof sportId === "number") return `whoop_sport_${sportId}`; + return "workout"; +} + +export async function syncUserWorkout( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + const connection = await prisma.whoopConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true }, + }); + if (!connection) return 0; + + const start = incrementalStart(connection.lastSyncedAt, { + fullSync: opts.fullSync, + }); + + let records: Awaited>; + try { + records = await fetchWorkouts(tokenInfo.accessToken, { start }); + } catch (err) { + await recordWhoopSyncFailure(userId, err); + throw err; + } + + 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; + + 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 } + : {}), + }; + + try { + await prisma.workout.upsert({ + where: { + userId_source_externalId: { + userId, + source: "WHOOP", + externalId: w.id, + }, + }, + create: { + 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}`); + } + } + + await markSynced(userId); + return imported; +} diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts new file mode 100644 index 000000000..bd23b9c45 --- /dev/null +++ b/src/lib/whoop/sync.ts @@ -0,0 +1,338 @@ +/** + * WHOOP sync service — token refresh (rotating refresh token), the shared + * Measurement upsert/rollup-fold tail, and the per-resource sync entry points. + * + * Mirrors `src/lib/withings/sync.ts`: + * - `getValidToken` decrypts the stored pair, refreshes at + * `tokenExpiresAt - 5 min`, and persists BOTH rotated tokens (WHOOP + * invalidates the prior access AND refresh token on every refresh — the + * same discipline Withings uses for its rotating refresh token). + * - Each per-resource sync (`sync-recovery` / `sync-sleep` / `sync-cycle` / + * `sync-workout`) upserts into `Measurement` / `Workout` keyed on + * `(userId, type, source = WHOOP, externalId)` so a re-post (WHOOP + * re-scores recovery/sleep after the fact) overwrites in place rather than + * minting a duplicate. After the upserts the rollup tier is re-folded + * (`recomputeBucketsForMeasurement`) and the status-insight caches are + * invalidated, identical to the Withings tail. + * + * The incremental window starts from `lastSyncedAt - overlap`. WHOOP re-scores + * recovery/sleep hours after the night, so the overlap must comfortably cover + * the re-score lag — `WHOOP_RECOVERY_SLEEP_OVERLAP_MS` is 24 h; workout/cycle + * use the smaller `WHOOP_DEFAULT_OVERLAP_MS`. + */ +import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; +import { encrypt, decrypt } from "@/lib/crypto"; +import { getEvent } from "@/lib/logging/context"; +import { + isReauthRequired, + recordSyncFailure, + recordSyncSuccess, + type FailureKind, +} from "@/lib/integrations/status"; +import { + collapseToTypeDayKeys, + recomputeBucketsForMeasurement, +} from "@/lib/rollups/measurement-rollups"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { refreshAccessToken } from "./client"; +import { getUserWhoopCredentials } from "./credentials"; +import { + WhoopApiError, + classifyWhoopError, + type WhoopClassification, +} from "./response-classifier"; + +/** Refresh the access token this many ms before `tokenExpiresAt`. */ +const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; + +/** + * Overlap window for the incremental sync, in ms. WHOOP re-scores recovery + * and sleep after the fact (the final recovery/sleep score can land hours + * after the night), so the recovery/sleep overlap is a full 24 h to make sure + * the re-scored record is re-fetched on the next tick. Workout/cycle settle + * fast — a smaller overlap suffices and keeps the page count down. + */ +export const WHOOP_RECOVERY_SLEEP_OVERLAP_MS = 24 * 60 * 60 * 1000; +export const WHOOP_DEFAULT_OVERLAP_MS = 60 * 60 * 1000; // 1 h + +export interface WhoopTokenInfo { + accessToken: string; + connection: { id: string; whoopUserId: string }; +} + +/** + * Resolve a valid WHOOP access token for a user, refreshing if it is within + * the 5-minute expiry buffer. On refresh, persists BOTH rotated tokens. + * Returns null when there is no connection, no credentials, or the refresh + * fails (the failure is recorded so scheduled syncs back off). + */ +export async function getValidToken( + userId: string, +): Promise { + const connection = await prisma.whoopConnection.findUnique({ + where: { userId }, + }); + if (!connection) return null; + + const accessToken = decrypt(connection.accessToken); + const refreshToken = decrypt(connection.refreshToken); + + if ( + connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < + Date.now() + ) { + try { + const creds = await getUserWhoopCredentials(userId); + if (!creds) { + getEvent()?.addWarning( + `No WHOOP credentials found for user ${userId} during token refresh`, + ); + await recordSyncFailure({ + userId, + integration: "whoop", + kind: "reauth_required", + message: "WHOOP credentials missing — token refresh skipped", + errorCode: "credentials_missing", + }); + return null; + } + + const newTokens = await refreshAccessToken(refreshToken, creds); + const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000); + + // WHOOP rotates the refresh token on every refresh — persist BOTH the + // new access token AND the new refresh token, or the next refresh + // reuses an invalidated token and the connection drops to reauth. + await prisma.whoopConnection.update({ + where: { id: connection.id }, + data: { + accessToken: encrypt(newTokens.access_token), + refreshToken: encrypt(newTokens.refresh_token), + tokenExpiresAt: expiresAt, + }, + }); + + return { + accessToken: newTokens.access_token, + connection: { + id: connection.id, + whoopUserId: connection.whoopUserId, + }, + }; + } catch (err) { + getEvent()?.addWarning( + `WHOOP token refresh failed for user ${userId}: ${err}`, + ); + await recordWhoopSyncFailure(userId, err); + return null; + } + } + + return { + accessToken, + connection: { + id: connection.id, + whoopUserId: connection.whoopUserId, + }, + }; +} + +/** + * 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. + */ +export function incrementalStart( + lastSyncedAt: Date | null, + opts: { fullSync?: boolean; overlapMs?: number } = {}, +): Date | undefined { + if (opts.fullSync) return undefined; + const overlap = opts.overlapMs ?? WHOOP_DEFAULT_OVERLAP_MS; + if (lastSyncedAt) return new Date(lastSyncedAt.getTime() - overlap); + return new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); +} + +/** + * One mapped reading destined for a `Measurement` row, with the resource uuid + * already resolved into a full `externalId`. The per-resource syncs build these + * from the client mappers (`:`). + */ +export interface WhoopMeasurementUpsert { + type: string; + value: number; + unit: string; + measuredAt: Date; + externalId: string; + sleepStage?: "CORE" | "DEEP" | "REM" | "AWAKE" | "IN_BED" | null; +} + +/** + * Upsert a batch of mapped WHOOP readings for one user and fold the rollup + * tier + invalidate status-insight caches once at the end (mirrors the + * Withings sync tail). Idempotent: the `(userId, type, source, externalId)` + * unique key makes a re-post (WHOOP re-score) overwrite in place. Returns the + * count of rows written. + * + * Best-effort on the rollup fold + insight invalidate — a populator hiccup + * never fails the user's sync. + */ +export async function upsertWhoopMeasurements( + userId: string, + readings: WhoopMeasurementUpsert[], +): Promise { + if (readings.length === 0) return 0; + + let imported = 0; + const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; + + for (const r of readings) { + const type = r.type as MeasurementType; + try { + await prisma.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type, + source: "WHOOP", + externalId: r.externalId, + }, + }, + create: { + userId, + type, + source: "WHOOP", + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + externalId: r.externalId, + sleepStage: r.sleepStage ?? null, + }, + update: { + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + sleepStage: r.sleepStage ?? null, + // Surface the server-side mutation to the iOS LWW reconciler. + syncVersion: { increment: 1 }, + }, + }); + touched.push({ type, measuredAt: r.measuredAt }); + imported++; + } catch (err) { + getEvent()?.addWarning(`WHOOP: failed to upsert measurement: ${err}`); + } + } + + try { + const keys = collapseToTypeDayKeys(touched); + for (const k of keys) { + await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); + } + invalidateStatusInsightsForTypes( + userId, + keys.map((k) => k.type), + ).catch((err) => { + getEvent()?.addWarning( + `whoop: status-insight invalidate failed for ${userId}: ${err}`, + ); + }); + } catch (err) { + getEvent()?.addWarning( + `whoop: rollup recompute failed for ${userId}: ${err}`, + ); + } + + return imported; +} + +/** Stamp `lastSyncedAt = now` after a successful resource sync. */ +export async function markSynced(userId: string): Promise { + await prisma.whoopConnection.update({ + where: { userId }, + data: { lastSyncedAt: new Date() }, + }); +} + +/** + * 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 + * manual `/api/whoop/sync` trigger. + * + * Parks immediately when the connection is at `error_reauth` (the user must + * reconnect first) — returns 0, matching the Withings no-op contract. + */ +export async function syncUserWhoop( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + if (await isReauthRequired(userId, "whoop")) { + getEvent()?.addWarning( + `whoop sync skipped for ${userId}: parked at error_reauth`, + ); + return 0; + } + + const { syncUserRecovery } = await import("./sync-recovery"); + const { syncUserSleep } = await import("./sync-sleep"); + const { syncUserCycle } = await import("./sync-cycle"); + const { syncUserWorkout } = await import("./sync-workout"); + + let total = 0; + let anyFailed = false; + for (const fn of [ + syncUserRecovery, + syncUserSleep, + syncUserCycle, + syncUserWorkout, + ]) { + try { + total += await fn(userId, opts); + } catch (err) { + anyFailed = true; + getEvent()?.addWarning(`whoop ${fn.name} failed for ${userId}: ${err}`); + } + } + + if (!anyFailed) { + await recordSyncSuccess(userId, "whoop"); + } + return total; +} + +/** + * Map a WHOOP response classification onto a `FailureKind` and record it. + * Shared by every per-resource catch-block and the token-refresh path. + */ +export async function recordWhoopSyncFailure( + userId: string, + err: unknown, +): Promise { + const message = err instanceof Error ? err.message : String(err); + await recordSyncFailure({ + userId, + integration: "whoop", + kind: classificationToFailureKind(classifyWhoopError(err)), + message, + errorCode: + err instanceof WhoopApiError ? err.httpStatus?.toString() : undefined, + }); +} + +export function classificationToFailureKind( + classification: WhoopClassification, +): FailureKind { + switch (classification) { + case "reauth_required": + return "reauth_required"; + case "persistent": + return "persistent"; + case "transient": + return "transient"; + case "success": + // A caller asking for the FailureKind of a success is a contract bug; + // surface it as transient so the audit log still records the anomaly. + return "transient"; + } +}