mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 20:42:43 +00:00
fix(whoop): a dead token no longer reports a successful sync
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.
This commit is contained in:
1 parent
8be7458c7a
commit
9a69e4e557
14 files changed
+672
-65
No files matched your search
@@ -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 } };
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
@@ -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);
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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<number>>(async () => 0),
|
||||
syncWhoopWorkoutById: vi.fn<(...a: unknown[]) => Promise<number>>(
|
||||
async () => 0,
|
||||
),
|
||||
findManyMock: vi.fn<
|
||||
(...a: unknown[]) => Promise<Array<{ userId: string }>>
|
||||
>(async () => []),
|
||||
}),
|
||||
);
|
||||
const {
|
||||
syncUserWorkout,
|
||||
syncWhoopWorkoutById,
|
||||
findManyMock,
|
||||
recordWhoopSyncFailureMock,
|
||||
recordSyncSuccessMock,
|
||||
} = vi.hoisted(() => ({
|
||||
syncUserWorkout: vi.fn<(...a: unknown[]) => Promise<number>>(async () => 0),
|
||||
syncWhoopWorkoutById: vi.fn<(...a: unknown[]) => Promise<number>>(
|
||||
async () => 0,
|
||||
),
|
||||
findManyMock: vi.fn<(...a: unknown[]) => Promise<Array<{ userId: string }>>>(
|
||||
async () => [],
|
||||
),
|
||||
recordWhoopSyncFailureMock: vi.fn<(...a: unknown[]) => Promise<void>>(
|
||||
async () => {},
|
||||
),
|
||||
recordSyncSuccessMock: vi.fn<(...a: unknown[]) => Promise<void>>(
|
||||
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<typeof import("@/lib/integrations/status")>()),
|
||||
recordSyncSuccess: recordSyncSuccessMock,
|
||||
}));
|
||||
vi.mock("@/lib/whoop/sync-core", async (orig) => ({
|
||||
...(await orig<typeof import("@/lib/whoop/sync-core")>()),
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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}`);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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() },
|
||||
|
||||
@@ -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 },
|
||||
|
||||
@@ -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<typeof import("../sync-core")>("../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);
|
||||
});
|
||||
});
|
||||
@@ -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<unknown> => 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<number> => {
|
||||
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: <T>(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<typeof import("../client")>();
|
||||
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<T>(
|
||||
fn: () => Promise<T>,
|
||||
): 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");
|
||||
});
|
||||
});
|
||||
@@ -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: <T>(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();
|
||||
|
||||
@@ -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<T>(
|
||||
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<HardFailTracker>();
|
||||
|
||||
/**
|
||||
* 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<number>,
|
||||
): Promise<number> {
|
||||
// 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;
|
||||
|
||||
@@ -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:<id>:<fieldTag>` upsert keeps the overwrite idempotent.
|
||||
const start = incrementalStart(resolveResourceCursor(connection, "cycle"), {
|
||||
fullSync: opts.fullSync,
|
||||
overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
});
|
||||
|
||||
let records: Awaited<ReturnType<typeof fetchCycles>>;
|
||||
|
||||
+47
-19
@@ -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<number> {
|
||||
): 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 };
|
||||
}
|
||||
Reference in new issue
Block a user