From 9a69e4e55768b7e07f498256c20a39db2a4f5203 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 24 Jul 2026 20:05:41 +0200 Subject: [PATCH] fix(whoop): a dead token no longer reports a successful sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port the hard-fail ledger the Fitbit transport already carries. A revoked WHOOP grant makes every resource short-circuit and return zero rows without throwing, so the cycle read as clean: it stamped success, un-parked the reconnect state, reset the failure streak, and the hourly cohort hammered the dead refresh token on every tick. Record the three dead-token null-return paths on the ledger and fold that verdict into both drivers — the full orchestrator and the per-resource cron path that runs most often — so a dead-token cycle refuses the success stamp and the park holds. Return an imported-plus-failed verdict from the orchestrator and gate the history backfill on it: an interrupted full-history walk now throws for a pg-boss retry instead of being stamped complete, and a connection parked at reauth returns without burning the retry budget. Thread the same verdict into the one-shot sleep-timeline backfill. Also catch a write-path throw on both drivers so a persistent measurement-write failure surfaces on the status ledger instead of leaving a stale-green pill, and widen the cycle overlap to the seven-day re-score window the other collections use so a late-revised day's strain and energy are re-fetched rather than stepped past. --- .../api/whoop/sync/__tests__/route.test.ts | 4 +- src/app/api/whoop/sync/route.ts | 2 +- .../__tests__/sleep-timeline-backfill.test.ts | 14 +- src/lib/jobs/__tests__/whoop-backfill.test.ts | 50 +++- .../__tests__/whoop-resource-sync.test.ts | 73 ++++- src/lib/jobs/reminder/whoop-sync.ts | 13 +- src/lib/jobs/sleep-timeline-backfill.ts | 12 +- src/lib/jobs/whoop-backfill.ts | 42 ++- .../__tests__/sync-cycle-overlap.test.ts | 83 ++++++ .../whoop/__tests__/sync-dead-token.test.ts | 277 ++++++++++++++++++ .../whoop/__tests__/sync-user-whoop.test.ts | 12 +- src/lib/whoop/sync-core.ts | 80 ++++- src/lib/whoop/sync-cycle.ts | 9 + src/lib/whoop/sync.ts | 66 +++-- 14 files changed, 672 insertions(+), 65 deletions(-) create mode 100644 src/lib/whoop/__tests__/sync-cycle-overlap.test.ts create mode 100644 src/lib/whoop/__tests__/sync-dead-token.test.ts diff --git a/src/app/api/whoop/sync/__tests__/route.test.ts b/src/app/api/whoop/sync/__tests__/route.test.ts index 953ae2c02..6e5f461e7 100644 --- a/src/app/api/whoop/sync/__tests__/route.test.ts +++ b/src/app/api/whoop/sync/__tests__/route.test.ts @@ -30,7 +30,7 @@ describe("POST /api/whoop/sync", () => { beforeEach(() => vi.clearAllMocks()); it("triggers an incremental sync by default", async () => { - sync.mockResolvedValue(7); + sync.mockResolvedValue({ imported: 7, failed: false }); const res = (await ( POST as unknown as (r: NextRequest) => Promise<{ data: unknown }> )(req({}))) as { data: { imported: number; fullSync: boolean } }; @@ -40,7 +40,7 @@ describe("POST /api/whoop/sync", () => { }); it("honours fullSync: true", async () => { - sync.mockResolvedValue(99); + sync.mockResolvedValue({ imported: 99, failed: false }); const res = (await ( POST as unknown as (r: NextRequest) => Promise<{ data: unknown }> )(req({ fullSync: true }))) as { data: { fullSync: boolean } }; diff --git a/src/app/api/whoop/sync/route.ts b/src/app/api/whoop/sync/route.ts index 17d069ad4..d3a331aca 100644 --- a/src/app/api/whoop/sync/route.ts +++ b/src/app/api/whoop/sync/route.ts @@ -27,6 +27,6 @@ export const POST = apiHandler(async (request: NextRequest) => { // no body provided -> default incremental sync } - const imported = await syncUserWhoop(user.id, { fullSync }); + const { imported } = await syncUserWhoop(user.id, { fullSync }); return apiSuccess({ imported, fullSync }); }); diff --git a/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts b/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts index 710f923b9..7b816dc90 100644 --- a/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts +++ b/src/lib/jobs/__tests__/sleep-timeline-backfill.test.ts @@ -107,7 +107,7 @@ describe("enqueueBootTimeSleepTimelineBackfill — discovery", () => { describe("runSleepTimelineBackfillForUser", () => { it("WHOOP: deletes the source's sleep rows, full-syncs, and stamps the marker", async () => { prismaMock.measurement.deleteMany.mockResolvedValue({ count: 5 }); - syncUserWhoop.mockResolvedValue(42); + syncUserWhoop.mockResolvedValue({ imported: 42, failed: false }); prismaMock.whoopConnection.update.mockResolvedValue({}); const { deleted, imported } = await runSleepTimelineBackfillForUser( @@ -126,6 +126,18 @@ describe("runSleepTimelineBackfillForUser", () => { expect(updateArg.data.sleepTimelineBackfillAt).toBeInstanceOf(Date); }); + it("WHOOP: a failed verdict THROWS without stamping the marker (gated re-sync)", async () => { + // The rows were already deleted; a partial re-sync must not stamp the + // completion marker over the gap — it throws so pg-boss retries. + prismaMock.measurement.deleteMany.mockResolvedValue({ count: 5 }); + syncUserWhoop.mockResolvedValue({ imported: 3, failed: true }); + + await expect( + runSleepTimelineBackfillForUser("w1", "WHOOP"), + ).rejects.toThrow(/incomplete/); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + }); + it("WITHINGS: deletes the source's sleep rows, re-syncs, and stamps the marker", async () => { prismaMock.measurement.deleteMany.mockResolvedValue({ count: 3 }); syncWithingsSleep.mockResolvedValue(7); diff --git a/src/lib/jobs/__tests__/whoop-backfill.test.ts b/src/lib/jobs/__tests__/whoop-backfill.test.ts index e6573ad2d..7cb6404e4 100644 --- a/src/lib/jobs/__tests__/whoop-backfill.test.ts +++ b/src/lib/jobs/__tests__/whoop-backfill.test.ts @@ -7,16 +7,18 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { prismaMock, bossSend, syncUserWhoop } = vi.hoisted(() => ({ - prismaMock: { - whoopConnection: { - findMany: vi.fn(), - update: vi.fn(), +const { prismaMock, bossSend, syncUserWhoop, isReauthRequiredMock } = + vi.hoisted(() => ({ + prismaMock: { + whoopConnection: { + findMany: vi.fn(), + update: vi.fn(), + }, }, - }, - bossSend: vi.fn(), - syncUserWhoop: vi.fn(), -})); + bossSend: vi.fn(), + syncUserWhoop: vi.fn(), + isReauthRequiredMock: vi.fn(async () => false), + })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); @@ -28,6 +30,10 @@ vi.mock("@/lib/whoop/sync", () => ({ syncUserWhoop: (...a: unknown[]) => syncUserWhoop(...a), })); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: isReauthRequiredMock, +})); + vi.mock("@/lib/logging/context", () => ({ annotate: () => {}, getEvent: () => null, @@ -40,6 +46,8 @@ import { beforeEach(() => { vi.clearAllMocks(); + isReauthRequiredMock.mockResolvedValue(false); + prismaMock.whoopConnection.update.mockResolvedValue({}); }); describe("enqueueBootTimeWhoopBackfill — discovery", () => { @@ -85,10 +93,9 @@ describe("enqueueBootTimeWhoopBackfill — discovery", () => { }); }); -describe("runWhoopBackfillForUser", () => { - it("runs a full sync and stamps backfillCompletedAt", async () => { - syncUserWhoop.mockResolvedValue(123); - prismaMock.whoopConnection.update.mockResolvedValue({}); +describe("runWhoopBackfillForUser — verdict-gated marker", () => { + it("stamps backfillCompletedAt after a CLEAN full-history run", async () => { + syncUserWhoop.mockResolvedValue({ imported: 123, failed: false }); const { imported } = await runWhoopBackfillForUser("u1"); @@ -98,4 +105,21 @@ describe("runWhoopBackfillForUser", () => { expect(updateArg.where).toEqual({ userId: "u1" }); expect(updateArg.data.backfillCompletedAt).toBeInstanceOf(Date); }); + + it("a failed verdict THROWS without stamping — pg-boss retries become real", async () => { + syncUserWhoop.mockResolvedValue({ imported: 7, failed: true }); + + await expect(runWhoopBackfillForUser("u1")).rejects.toThrow(/incomplete/); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + }); + + it("a connection parked at error_reauth returns WITHOUT running the sync or stamping", async () => { + isReauthRequiredMock.mockResolvedValue(true); + + await expect(runWhoopBackfillForUser("u1")).resolves.toEqual({ + imported: 0, + }); + expect(syncUserWhoop).not.toHaveBeenCalled(); + expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled(); + }); }); diff --git a/src/lib/jobs/__tests__/whoop-resource-sync.test.ts b/src/lib/jobs/__tests__/whoop-resource-sync.test.ts index a7469d718..295ae98e6 100644 --- a/src/lib/jobs/__tests__/whoop-resource-sync.test.ts +++ b/src/lib/jobs/__tests__/whoop-resource-sync.test.ts @@ -10,17 +10,38 @@ 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 () => []), - }), -); +const { + syncUserWorkout, + syncWhoopWorkoutById, + findManyMock, + recordWhoopSyncFailureMock, + recordSyncSuccessMock, +} = 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 () => [], + ), + recordWhoopSyncFailureMock: vi.fn<(...a: unknown[]) => Promise>( + async () => {}, + ), + recordSyncSuccessMock: vi.fn<(...a: unknown[]) => Promise>( + async () => {}, + ), +})); + +// Keep the real marker helpers (isSyncFailureRecorded / markSyncFailureRecorded) +// but stub the DB-touching writers so this stays a unit test. +vi.mock("@/lib/integrations/status", async (orig) => ({ + ...(await orig()), + recordSyncSuccess: recordSyncSuccessMock, +})); +vi.mock("@/lib/whoop/sync-core", async (orig) => ({ + ...(await orig()), + recordWhoopSyncFailure: recordWhoopSyncFailureMock, +})); vi.mock("@/lib/whoop/sync-recovery", () => ({ syncUserRecovery: vi.fn(async () => 0), @@ -63,6 +84,7 @@ vi.mock("../reminder/shared", () => ({ })); import { handleWhoopWorkoutSync } from "../reminder/whoop-sync"; +import { markSyncFailureRecorded } from "@/lib/integrations/status"; function job(data: { userId?: string; resourceId?: string }): Job<{ userId?: string; @@ -111,3 +133,32 @@ describe("runWhoopResourceSync dispatch", () => { expect(syncUserWorkout).toHaveBeenCalledTimes(2); }); }); + +describe("runWhoopResourceSync — WH-4 write-path ledger honesty", () => { + it("records an UNMARKED write-path throw once (a stale-green pill otherwise)", async () => { + // A `MeasurementReconciliationError` out of the upsert reaches no ledger on + // the cron path before WH-4 — the pill stays green through a persistent + // write failure. The catch now records it once. + syncUserWorkout.mockRejectedValueOnce(new Error("write failed")); + + await handleWhoopWorkoutSync([job({ userId: "u1" })]); + + expect(recordWhoopSyncFailureMock).toHaveBeenCalledTimes(1); + expect(recordWhoopSyncFailureMock).toHaveBeenCalledWith( + "u1", + expect.objectContaining({ message: "write failed" }), + ); + }); + + it("does NOT re-record a MARKED fetch error (no double-record)", async () => { + // A fetch hard-fail was already recorded + marked in + // `handleCollectionFetchError`; the cron catch must skip it. + syncUserWorkout.mockRejectedValueOnce( + markSyncFailureRecorded(new Error("already recorded fetch error")), + ); + + await handleWhoopWorkoutSync([job({ userId: "u1" })]); + + expect(recordWhoopSyncFailureMock).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/jobs/reminder/whoop-sync.ts b/src/lib/jobs/reminder/whoop-sync.ts index a99c33358..b8df364f5 100644 --- a/src/lib/jobs/reminder/whoop-sync.ts +++ b/src/lib/jobs/reminder/whoop-sync.ts @@ -21,7 +21,11 @@ import { syncUserWorkout, syncWhoopWorkoutById, } from "@/lib/whoop/sync-workout"; -import { syncWhoopResourceWithStatus } from "@/lib/whoop/sync-core"; +import { + recordWhoopSyncFailure, + syncWhoopResourceWithStatus, +} from "@/lib/whoop/sync-core"; +import { isSyncFailureRecorded } from "@/lib/integrations/status"; import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy"; import { getWorkerPrisma } from "./shared"; @@ -110,6 +114,13 @@ export async function runWhoopResourceSync( }); } } catch (err) { + // WH-4: a write-path throw on the hourly cron reaches no ledger + // otherwise (stale-green pill). Fetch hard-fails already recorded + + // marked in `handleCollectionFetchError`, so record only the UNMARKED + // escapes here. + if (!isSyncFailureRecorded(err)) { + await recordWhoopSyncFailure(userId, err); + } evt.addWarning(`${taskName} failed for user ${userId}: ${err}`); } } diff --git a/src/lib/jobs/sleep-timeline-backfill.ts b/src/lib/jobs/sleep-timeline-backfill.ts index 5cee00849..33aed68bf 100644 --- a/src/lib/jobs/sleep-timeline-backfill.ts +++ b/src/lib/jobs/sleep-timeline-backfill.ts @@ -71,7 +71,17 @@ export async function runSleepTimelineBackfillForUser( let imported = 0; if (provider === "WHOOP") { - imported = await syncUserWhoop(userId, { fullSync: true }); + // `syncUserWhoop` now returns a verdict; gate the completion marker on a + // clean run so a partial re-sync (dead token, write failure) throws for a + // pg-boss retry instead of freezing the timeline gap under a stamped + // marker — the same discipline the WHOOP/Fitbit backfills use. + const result = await syncUserWhoop(userId, { fullSync: true }); + imported = result.imported; + if (result.failed) { + throw new Error( + `sleep-timeline backfill incomplete for user ${userId} (whoop) — marker not stamped`, + ); + } await prisma.whoopConnection.update({ where: { userId }, data: { sleepTimelineBackfillAt: new Date() }, diff --git a/src/lib/jobs/whoop-backfill.ts b/src/lib/jobs/whoop-backfill.ts index aded98c8e..488ae6b1c 100644 --- a/src/lib/jobs/whoop-backfill.ts +++ b/src/lib/jobs/whoop-backfill.ts @@ -20,6 +20,7 @@ import { prisma } from "@/lib/db"; import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; +import { isReauthRequired } from "@/lib/integrations/status"; import { syncUserWhoop } from "@/lib/whoop/sync"; export const WHOOP_BACKFILL_QUEUE = "whoop-backfill"; @@ -37,15 +38,46 @@ export interface WhoopBackfillPayload { } /** - * 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. + * Per-user backfill handler. Runs a full-history sync for one account and stamps + * `backfillCompletedAt` ONLY on a clean run, 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. Mirrors the landed + * `runFitbitBackfillForUser`. + * + * Verdict-gated: a partial hard failure (a dead token, a write failing, a + * fetch hard-fail) THROWS so pg-boss retries the job — stamping the marker over + * a partial multi-year walk would freeze the history gap forever and the + * configured `retryLimit` would be dead code. A connection parked at + * `error_reauth` is NOT clean either, but throwing against a dead grant would + * just burn the retry budget on the same no-op: return WITHOUT stamping — boot + * discovery re-offers the account after the user reconnects and a clean walk + * completes. `syncUserWhoop` has its own reauth short-circuit (returns + * `failed: true`), but the guard must still live here so a parked account + * RETURNS instead of throwing through three retries. */ export async function runWhoopBackfillForUser( userId: string, ): Promise<{ imported: number }> { - const imported = await syncUserWhoop(userId, { fullSync: true }); + if (await isReauthRequired(userId, "whoop")) { + annotate({ + action: { + name: "whoop.backfill.skipped_reauth", + details: { imported: 0 }, + }, + }); + return { imported: 0 }; + } + + const { imported, failed } = await syncUserWhoop(userId, { fullSync: true }); + + if (failed) { + // Surface through the pg-boss retry path (retryLimit + backoff set by + // `integrationBackfillSourceOptions`). If the failure parked the connection + // at error_reauth meanwhile, the next attempt takes the reauth return above. + throw new Error( + `whoop backfill incomplete for user ${userId} — marker not stamped`, + ); + } await prisma.whoopConnection.update({ where: { userId }, diff --git a/src/lib/whoop/__tests__/sync-cycle-overlap.test.ts b/src/lib/whoop/__tests__/sync-cycle-overlap.test.ts new file mode 100644 index 000000000..27b865782 --- /dev/null +++ b/src/lib/whoop/__tests__/sync-cycle-overlap.test.ts @@ -0,0 +1,83 @@ +/** + * Q8 — WHOOP cycle ingest used the narrow 1 h default overlap while workout / + * recovery / sleep use the 7-day re-score overlap. A cycle re-scored through the + * day (or fetched just after the cursor stepped past it) then fell before the + * incremental `start` and its final value was never re-fetched. This pins cycle + * onto the SAME overlap the other collections use (the upsert is idempotent). + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const fetchCycles = vi.fn(); + +vi.mock("../client", () => ({ + fetchCycles: (...a: unknown[]) => fetchCycles(...a), + mapCycle: () => [], +})); + +const getValidToken = vi.fn(); +const markResourceSynced = vi.fn(); +const upsertWhoopMeasurements = vi.fn(); +vi.mock("../sync-core", async () => { + const actual = + await vi.importActual("../sync-core"); + return { + ...actual, + getValidToken: (...a: unknown[]) => getValidToken(...a), + markResourceSynced: (...a: unknown[]) => markResourceSynced(...a), + upsertWhoopMeasurements: (...a: unknown[]) => upsertWhoopMeasurements(...a), + }; +}); + +const whoopConnectionFindUnique = vi.fn(); +vi.mock("@/lib/db", () => ({ + prisma: { + whoopConnection: { + findUnique: (...a: unknown[]) => whoopConnectionFindUnique(...a), + }, + }, +})); + +import { syncUserCycle } from "../sync-cycle"; +import { + WHOOP_FULL_SYNC_ANCHOR, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, +} from "../sync-core"; + +const LAST_SYNCED = new Date("2026-06-10T12:00:00.000Z"); + +beforeEach(() => { + vi.clearAllMocks(); + getValidToken.mockResolvedValue({ + accessToken: "tok", + connection: { id: "c1", whoopUserId: "w1" }, + }); + whoopConnectionFindUnique.mockResolvedValue({ + lastSyncedAt: LAST_SYNCED, + resourceCursors: null, + }); + markResourceSynced.mockResolvedValue(undefined); + upsertWhoopMeasurements.mockResolvedValue(0); + fetchCycles.mockResolvedValue([]); +}); + +describe("syncUserCycle — 7-day re-score overlap window", () => { + it("fetches from lastSyncedAt minus the 7-day overlap, not 1 h", async () => { + await syncUserCycle("user-1"); + + expect(fetchCycles).toHaveBeenCalledTimes(1); + const arg = fetchCycles.mock.calls[0]![1] as { start?: Date }; + expect(arg.start?.getTime()).toBe( + LAST_SYNCED.getTime() - WHOOP_RECOVERY_SLEEP_OVERLAP_MS, + ); + // A cycle still settling 6 h after the cursor is inside the fetch range; + // under the old 1 h overlap it would be missed. + const sixHoursBefore = LAST_SYNCED.getTime() - 6 * 60 * 60 * 1000; + expect(arg.start!.getTime()).toBeLessThanOrEqual(sixHoursBefore); + }); + + it("fullSync anchors at the deep-history anchor, not the cursor", async () => { + await syncUserCycle("user-1", { fullSync: true }); + const arg = fetchCycles.mock.calls[0]![1] as { start?: Date }; + expect(arg.start).toEqual(WHOOP_FULL_SYNC_ANCHOR); + }); +}); diff --git a/src/lib/whoop/__tests__/sync-dead-token.test.ts b/src/lib/whoop/__tests__/sync-dead-token.test.ts new file mode 100644 index 000000000..3065f74a2 --- /dev/null +++ b/src/lib/whoop/__tests__/sync-dead-token.test.ts @@ -0,0 +1,277 @@ +/** + * WH-1 — pins the dead-token verdict for WHOOP, the parity mirror of + * `fitbit/__tests__/sync-dead-token.test.ts`. + * + * `getValidToken`'s three null-return paths (credentials missing, refresh + * failure inside the advisory-lock transaction, and the in-lock vanished + * connection) must register on the cycle's hard-fail ledger, and a cycle whose + * token is dead must NOT stamp `recordSyncSuccess`. Otherwise every resource + * returns 0 without failures, the cycle reads as clean, `recordSyncSuccess` + * un-parks `error_reauth`, and the hourly cohort hammers the dead refresh token + * forever while `lastSyncedAt` advances past real data. + * + * Includes the WHOOP-specific pin Fitbit has no twin for: the per-resource + * driver `syncWhoopResourceWithStatus` (the dominant hourly-cron path) must + * also refuse the success stamp when a hard failure is noted inside `run`. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { + recordSyncFailureMock, + recordSyncSuccessMock, + isReauthRequiredMock, + prismaMock, + getUserWhoopCredentialsMock, + refreshAccessTokenMock, + resourceFake, +} = vi.hoisted(() => ({ + recordSyncFailureMock: vi.fn(async () => {}), + recordSyncSuccessMock: vi.fn(async () => {}), + isReauthRequiredMock: vi.fn(async () => false), + prismaMock: { + whoopConnection: { + findUnique: vi.fn(), + update: vi.fn(async () => ({})), + }, + $transaction: vi.fn(), + }, + getUserWhoopCredentialsMock: vi.fn(async (): Promise => null), + refreshAccessTokenMock: vi.fn(), + // The dead-token cycle test drives the real `syncUserWhoop`; each resource + // module resolves the token first — exactly what the real leaves do — so the + // ledger registration happens inside the cycle's ALS scope. + resourceFake: async (userId: string): Promise => { + const { getValidToken } = await import("../sync-core"); + return (await getValidToken(userId)) ? 1 : 0; + }, +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/crypto", () => ({ + encrypt: (s: string) => `enc(${s})`, + decrypt: (s: string) => s.replace(/^enc\(|\)$/g, ""), +})); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: isReauthRequiredMock, + recordSyncFailure: recordSyncFailureMock, + recordSyncSuccess: recordSyncSuccessMock, + isSyncFailureRecorded: () => false, + markSyncFailureRecorded: (err: T) => err, +})); +vi.mock("@/lib/integrations/oauth-refresh", () => ({ + acquireProviderTokenRefreshLock: vi.fn(async () => {}), + PROVIDER_REFRESH_TRANSACTION_OPTIONS: { maxWait: 10_000, timeout: 60_000 }, +})); +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: vi.fn(), + MeasurementReconciliationError: class extends Error {}, +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: vi.fn(() => []), + recomputeBucketsForMeasurement: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); +vi.mock("@/lib/arrivals/measurement-emit", () => ({ + emitInsertedMeasurementArrivals: vi.fn(async () => {}), +})); +vi.mock("@/lib/logging/context", () => ({ + getEvent: () => null, + annotate: () => {}, +})); +vi.mock("../credentials", () => ({ + getUserWhoopCredentials: getUserWhoopCredentialsMock, +})); +vi.mock("../client", async (orig) => { + const actual = await orig(); + return { ...actual, refreshAccessToken: refreshAccessTokenMock }; +}); + +// Each resource resolves the token first (the ledger-registration seam). +vi.mock("../sync-recovery", () => ({ syncUserRecovery: resourceFake })); +vi.mock("../sync-sleep", () => ({ syncUserSleep: resourceFake })); +vi.mock("../sync-cycle", () => ({ syncUserCycle: resourceFake })); +vi.mock("../sync-workout", () => ({ syncUserWorkout: resourceFake })); +vi.mock("../sync-body", () => ({ syncUserBody: resourceFake })); + +import { + WHOOP_TOKEN_HARD_FAIL, + getValidToken, + hardFailStorage, + noteHardFailure, + syncWhoopResourceWithStatus, +} from "../sync-core"; +import { syncUserWhoop } from "../sync"; +import { WhoopApiError } from "../response-classifier"; + +/** Run `fn` inside a fresh hard-fail ledger scope and return its failures. */ +async function withLedger( + fn: () => Promise, +): Promise<{ result: T; failures: string[] }> { + const tracker = { failures: [] as string[] }; + const result = await hardFailStorage.run(tracker, fn); + return { result, failures: tracker.failures }; +} + +/** A connection whose access token is inside the 5-min refresh buffer. */ +const EXPIRED_CONNECTION = { + id: "conn-1", + whoopUserId: "wu-1", + accessToken: "enc(access)", + refreshToken: "enc(refresh)", + tokenExpiresAt: new Date(Date.now() - 60_000), + lastSyncedAt: new Date("2026-07-01T00:00:00.000Z"), +}; + +const CREDS = { clientId: "id", clientSecret: "secret" }; + +beforeEach(() => { + vi.clearAllMocks(); + isReauthRequiredMock.mockResolvedValue(false); + prismaMock.whoopConnection.findUnique.mockReset(); + prismaMock.whoopConnection.update.mockReset().mockResolvedValue({}); + getUserWhoopCredentialsMock.mockReset().mockResolvedValue(null); + refreshAccessTokenMock.mockReset(); + // The advisory-lock transaction runs its callback against the same mock. + prismaMock.$transaction + .mockReset() + .mockImplementation(async (run: (tx: unknown) => unknown) => + run(prismaMock), + ); +}); + +describe("getValidToken — dead-token cycle verdict", () => { + it("registers the credentials-missing path on the hard-fail ledger", async () => { + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(null); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([WHOOP_TOKEN_HARD_FAIL]); + expect(recordSyncFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "whoop", + kind: "reauth_required", + errorCode: "credentials_missing", + }), + ); + }); + + it("registers a refresh failure on the hard-fail ledger", async () => { + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockRejectedValue( + new WhoopApiError({ + verb: "refreshToken", + classification: "reauth_required", + httpStatus: 401, + reason: "invalid_grant", + }), + ); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([WHOOP_TOKEN_HARD_FAIL]); + expect(recordSyncFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + integration: "whoop", + kind: "reauth_required", + }), + ); + }); + + it("registers an in-lock vanished connection on the ledger", async () => { + // The connection was deleted between the first read and the in-lock re-read + // (disconnect / webhook race). + prismaMock.whoopConnection.findUnique + .mockResolvedValueOnce(EXPIRED_CONNECTION as never) // initial read + .mockResolvedValueOnce(null as never); // in-lock re-read: row gone + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([WHOOP_TOKEN_HARD_FAIL]); + expect(refreshAccessTokenMock).not.toHaveBeenCalled(); + }); + + it("a successful refresh leaves the ledger empty", async () => { + // Guard against over-failing: a healthy refresh must not touch the ledger. + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockResolvedValue({ + access_token: "new-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result?.accessToken).toBe("new-access"); + expect(failures).toEqual([]); + }); +}); + +describe("syncUserWhoop — dead-token cycle refuses the success stamp", () => { + it("a dead-token cycle returns { imported: 0, failed: true } and never stamps success", async () => { + // Every resource resolves the token; the refresh fails each time — before + // the ledger fix the cycle read as CLEAN (all resources returned 0 without + // throwing), stamped success, and un-parked error_reauth. + prismaMock.whoopConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserWhoopCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockRejectedValue( + new WhoopApiError({ + verb: "refreshToken", + classification: "reauth_required", + httpStatus: 401, + reason: "invalid_grant", + }), + ); + + const res = await syncUserWhoop("user-1"); + + expect(res).toEqual({ imported: 0, failed: true }); + expect(recordSyncSuccessMock).not.toHaveBeenCalled(); + }); +}); + +describe("syncWhoopResourceWithStatus — dominant per-resource cron path", () => { + it("does NOT stamp success when a hard failure is noted inside run", async () => { + // The hourly-cron arm Fitbit has no twin for: a dead token noted by + // `getValidToken` inside `run` must block the per-resource success stamp, + // or the release is inert on the path that runs most often. + const imported = await syncWhoopResourceWithStatus("user-1", async () => { + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); + return 0; + }); + + expect(imported).toBe(0); + expect(recordSyncSuccessMock).not.toHaveBeenCalled(); + }); + + it("stamps success on a clean per-resource run (no hard failure)", async () => { + const imported = await syncWhoopResourceWithStatus("user-1", async () => 3); + + expect(imported).toBe(3); + expect(recordSyncSuccessMock).toHaveBeenCalledWith("user-1", "whoop"); + }); +}); diff --git a/src/lib/whoop/__tests__/sync-user-whoop.test.ts b/src/lib/whoop/__tests__/sync-user-whoop.test.ts index a1ebab5b9..2c5fcfbdc 100644 --- a/src/lib/whoop/__tests__/sync-user-whoop.test.ts +++ b/src/lib/whoop/__tests__/sync-user-whoop.test.ts @@ -46,6 +46,8 @@ vi.mock("@/lib/integrations/status", () => ({ recordSyncFailure: (...a: unknown[]) => recordSyncFailure(...a), recordSyncSuccess: (...a: unknown[]) => recordSyncSuccess(...a), isReauthRequired: (...a: unknown[]) => isReauthRequired(...a), + isSyncFailureRecorded: () => false, + markSyncFailureRecorded: (err: T) => err, })); vi.mock("@/lib/rollups/measurement-rollups", () => ({ collapseToTypeDayKeys: () => [], @@ -109,7 +111,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { fn.mockImplementation(softSkip403); } - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(0); expect(recordSyncSuccess).not.toHaveBeenCalled(); @@ -125,7 +127,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { syncUserWorkout.mockImplementation(softSkip403); syncUserBody.mockImplementation(softSkip403); - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(3); expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop"); @@ -138,7 +140,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { syncUserWorkout.mockResolvedValue(0); syncUserBody.mockResolvedValue(0); - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(3); expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "whoop"); @@ -155,7 +157,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { fn.mockResolvedValue(0); } - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(0); // No collection was 403'd → this is a genuine "nothing changed" tick. @@ -165,7 +167,7 @@ describe("syncUserWhoop — all-403 looks-healthy guard", () => { it("short-circuits without stamping success when parked at error_reauth", async () => { isReauthRequired.mockResolvedValue(true); - const total = await syncUserWhoop("user1"); + const { imported: total } = await syncUserWhoop("user1"); expect(total).toBe(0); expect(recordSyncSuccess).not.toHaveBeenCalled(); diff --git a/src/lib/whoop/sync-core.ts b/src/lib/whoop/sync-core.ts index 9c89cdf36..dc54db379 100644 --- a/src/lib/whoop/sync-core.ts +++ b/src/lib/whoop/sync-core.ts @@ -30,6 +30,7 @@ import type { MeasurementType } from "@/generated/prisma/client"; import { encrypt, decrypt } from "@/lib/crypto"; import { getEvent } from "@/lib/logging/context"; import { + markSyncFailureRecorded, recordSyncFailure, recordSyncSuccess, type FailureKind, @@ -97,6 +98,42 @@ export async function runWithWhoopSoftSkipTracking( return { result, softSkipCount: tracker.count }; } +/** + * Per-cycle ledger of HARD failures that must fail the cycle's verdict WITHOUT + * aborting sibling resources — a dead token (`getValidToken` returning null on + * a missing-credentials, failed-refresh, or in-lock vanished-connection path). + * Every resource leaf short-circuits `if (!tokenInfo) return 0` on a null + * token, so without this ledger a dead-token cycle reads as CLEAN: every + * resource "succeeds" with 0 rows, no failure is recorded, and + * `recordSyncSuccess` un-parks `error_reauth`, resets the failure buckets + + * `alertedAt`, and the hourly cohort hammers the dead refresh token forever. + * The orchestrator + the per-resource driver read this ledger to refuse the + * success stamp. Ported from Fitbit R2 / Google Health's `hardFailStorage`. + * Scoped by AsyncLocalStorage so concurrent per-user jobs never cross-pollinate. + */ +export interface HardFailTracker { + failures: string[]; +} +export const hardFailStorage = new AsyncLocalStorage(); + +/** + * Ledger label for a dead-token failure (`getValidToken` returning null on a + * missing-credentials, failed-refresh, or in-lock vanished-connection path). + * Ported from Fitbit's `FITBIT_TOKEN_HARD_FAIL`. + */ +export const WHOOP_TOKEN_HARD_FAIL = "token"; + +/** + * Record a hard failure on the ambient per-cycle ledger, if one is in scope. + * No-op outside a ledger scope (e.g. a direct REPL call). Used by + * `getValidToken`'s dead-token paths — anything that must fail the cycle's + * verdict without aborting sibling work. Ported from Fitbit's `noteHardFailure`. + */ +export function noteHardFailure(label: string): void { + const tracker = hardFailStorage.getStore(); + if (tracker) tracker.failures.push(label); +} + /** * Single-source the per-resource collection-fetch error handling. A 403 on one * data class soft-skips it (warn + return 0) so sibling resources still sync; @@ -122,7 +159,9 @@ export async function handleCollectionFetchError( return 0; } await recordWhoopSyncFailure(userId, err); - throw err; + // Marked so the orchestrator / cron catch (which records unmarked write-path + // throws for WH-4) does not record this fetch failure a SECOND time. + throw markSyncFailureRecorded(err); } /** Refresh the access token this many ms before `tokenExpiresAt`. */ @@ -137,8 +176,14 @@ const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; * narrow overlap; the night's per-stage rows then never reach the DB and * every sleep surface keeps showing the parallel coarse source for that * night. Seven days re-fetches a handful of records per tick (the upserts - * are idempotent) and closes the gap for every realistic lag. Workout/cycle - * settle fast — a smaller overlap suffices and keeps the page count down. + * are idempotent) and closes the gap for every realistic lag. Workout AND + * cycle now ride the same 7-day overlap: a workout can reach the WHOOP cloud + * days after a late phone sync, and a cycle is re-scored through the day — + * both collections filter on the record's own time range, so a narrow overlap + * steps the cursor past a still-settling record and misses it. One record per + * day keeps the re-fetch to a single page. `WHOOP_DEFAULT_OVERLAP_MS` stays + * the narrow fallback `incrementalStart` uses when no resource passes an + * explicit overlap. */ export const WHOOP_RECOVERY_SLEEP_OVERLAP_MS = 7 * 24 * 60 * 60 * 1000; export const WHOOP_DEFAULT_OVERLAP_MS = 60 * 60 * 1000; // 1 h @@ -242,6 +287,11 @@ export async function getValidToken( message: "WHOOP credentials missing — token refresh skipped", errorCode: "credentials_missing", }); + // A dead token means every resource "succeeds" with 0 rows — without a + // ledger entry the cycle reads as clean and `recordSyncSuccess` un-parks + // `error_reauth` while the hourly cohort hammers the dead refresh token. + // Fail the cycle's verdict so the park holds. + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); return null; } @@ -252,7 +302,14 @@ export async function getValidToken( const current = await tx.whoopConnection.findUnique({ where: { userId }, }); - if (!current?.whoopUserId) return null; + if (!current?.whoopUserId) { + // The connection vanished on the re-read inside the advisory lock + // (disconnect/webhook race). Register on the ledger so the cycle does + // not stamp success over a connection that no longer exists — the same + // invariant as the refresh branch below. + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); + return null; + } if ( current.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS >= @@ -295,6 +352,9 @@ export async function getValidToken( `WHOOP token refresh failed for user ${userId}: ${err}`, ); await recordWhoopSyncFailure(userId, err); + // The WH-1 kill shot: an all-zeros refresh-failure run must not stamp + // success, un-park `error_reauth`, and re-arm the alert. Fail the verdict. + noteHardFailure(WHOOP_TOKEN_HARD_FAIL); return null; } } @@ -535,10 +595,18 @@ export async function syncWhoopResourceWithStatus( userId: string, run: () => Promise, ): Promise { + // Run under BOTH trackers: soft-skip (a 403 scope gate) and hard-fail (a dead + // token noted by `getValidToken`). This is the dominant path — the four + // hourly crons walk every connection through here — so without the hard-fail + // gate a dead-token tick would stamp success, un-park `error_reauth`, and the + // whole release would be inert on the path that runs most often. + const hardFailTracker: HardFailTracker = { failures: [] }; const { result: imported, softSkipCount } = - await runWithWhoopSoftSkipTracking(run); + await runWithWhoopSoftSkipTracking(() => + hardFailStorage.run(hardFailTracker, run), + ); const softSkippedOnly = softSkipCount > 0 && imported === 0; - if (!softSkippedOnly) { + if (!softSkippedOnly && hardFailTracker.failures.length === 0) { await recordSyncSuccess(userId, "whoop"); } return imported; diff --git a/src/lib/whoop/sync-cycle.ts b/src/lib/whoop/sync-cycle.ts index c4068ca6f..39c0488ad 100644 --- a/src/lib/whoop/sync-cycle.ts +++ b/src/lib/whoop/sync-cycle.ts @@ -13,6 +13,7 @@ import { markResourceSynced, resolveResourceCursor, upsertWhoopMeasurements, + WHOOP_RECOVERY_SLEEP_OVERLAP_MS, type WhoopMeasurementUpsert, } from "./sync-core"; import { prisma } from "@/lib/db"; @@ -30,8 +31,16 @@ export async function syncUserCycle( }); if (!connection) return 0; + // WHOOP re-scores a cycle THROUGH the day (day strain + energy accrue and + // settle after the fact) and the collection filters on the cycle's own time + // range, not on when it updated — so a narrow 1 h overlap could step the + // cursor past a cycle still being revised and miss the final value. Ride the + // same 7-day re-score overlap workout/recovery/sleep already use; one cycle + // per day means ~7 records re-fetched per tick (one page), and the + // `cycle::` upsert keeps the overwrite idempotent. const start = incrementalStart(resolveResourceCursor(connection, "cycle"), { fullSync: opts.fullSync, + overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS, }); let records: Awaited>; diff --git a/src/lib/whoop/sync.ts b/src/lib/whoop/sync.ts index a7e74bc0c..06855227c 100644 --- a/src/lib/whoop/sync.ts +++ b/src/lib/whoop/sync.ts @@ -6,13 +6,22 @@ * the static leaf graph and the full per-user sequencing policy. */ import { getEvent } from "@/lib/logging/context"; -import { isReauthRequired, recordSyncSuccess } from "@/lib/integrations/status"; +import { + isReauthRequired, + isSyncFailureRecorded, + recordSyncSuccess, +} from "@/lib/integrations/status"; import { syncUserRecovery } from "./sync-recovery"; import { syncUserSleep } from "./sync-sleep"; import { syncUserCycle } from "./sync-cycle"; import { syncUserWorkout } from "./sync-workout"; import { syncUserBody } from "./sync-body"; -import { runWithWhoopSoftSkipTracking } from "./sync-core"; +import { + hardFailStorage, + recordWhoopSyncFailure, + runWithWhoopSoftSkipTracking, + type HardFailTracker, +} from "./sync-core"; /** * Full per-user sync across every WHOOP resource. Webhook-driven syncs enqueue @@ -20,17 +29,18 @@ import { runWithWhoopSoftSkipTracking } from "./sync-core"; * 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. + * reconnect first) — returns `{ imported: 0, failed: true }`, matching the + * landed Fitbit verdict contract so the backfill caller can gate its marker. */ export async function syncUserWhoop( userId: string, opts: { fullSync?: boolean } = {}, -): Promise { +): Promise<{ imported: number; failed: boolean }> { if (await isReauthRequired(userId, "whoop")) { getEvent()?.addWarning( `whoop sync skipped for ${userId}: parked at error_reauth`, ); - return 0; + return { imported: 0, failed: true }; } const resources = [ @@ -42,23 +52,41 @@ export async function syncUserWhoop( ]; let anyFailed = false; + // Nest the hard-fail ledger inside the soft-skip tracker (manual nest, not a + // combined wrapper) so a dead token noted by `getValidToken` — or a marked + // fetch hard-fail — fails the verdict just like a thrown resource error. + const hardFailTracker: HardFailTracker = { failures: [] }; const { result: total, softSkipCount } = await runWithWhoopSoftSkipTracking( - async () => { - let imported = 0; - for (const fn of resources) { - try { - imported += await fn(userId, opts); - } catch (err) { - anyFailed = true; - getEvent()?.addWarning( - `whoop ${fn.name} failed for ${userId}: ${err}`, - ); + () => + hardFailStorage.run(hardFailTracker, async () => { + let imported = 0; + for (const fn of resources) { + try { + imported += await fn(userId, opts); + } catch (err) { + anyFailed = true; + // WH-4: a write-path throw (`MeasurementReconciliationError` out of + // `upsertWhoopMeasurements`) reaches no ledger otherwise — a + // stale-green pill during a persistent write failure. Fetch + // hard-fails already recorded + marked in `handleCollectionFetchError`, + // so record only the UNMARKED escapes here (no double-record). + if (!isSyncFailureRecorded(err)) { + await recordWhoopSyncFailure(userId, err); + } + getEvent()?.addWarning( + `whoop ${fn.name} failed for ${userId}: ${err}`, + ); + } } - } - return imported; - }, + return imported; + }), ); + // A dead token / vanished connection notes the hard-fail ledger without + // throwing (each leaf short-circuits `if (!tokenInfo) return 0`); fold that + // into the verdict so an all-zeros dead-token cycle does not stamp success. + anyFailed = anyFailed || hardFailTracker.failures.length > 0; + // A genuine grant-revoke 403s EVERY collection: each resource soft-skips // (returns 0, records no failure), so `anyFailed` stays false and `total` is // 0 — yet the connection is dead until the token-refresh path next catches @@ -71,5 +99,5 @@ export async function syncUserWhoop( if (!anyFailed && !allSoftSkipped) { await recordSyncSuccess(userId, "whoop"); } - return total; + return { imported: total, failed: anyFailed || allSoftSkipped }; }