mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
feat(whoop): per-resource cursors, by-id refresh, and full-sync backfill
Give each WHOOP collection its own incremental cursor in a resource_cursors map so a slow or rejecting collection no longer drags the sync window forward for its siblings; legacy connections fall back to the shared last-synced timestamp. The update webhook now carries the record id and triggers a targeted single-record refresh instead of re-walking the collection, and a full sync anchors at a fixed deep start so a backfill covers the span an incremental tick skips. Cursor advances are a single atomic update guarded by GREATEST so the independent resource queues cannot race each other backward. Ingest row shape, source tagging, and upsert keys are unchanged.
This commit is contained in:
commit
cbde0d573f
16 files changed
+958
-125
No files matched your search
@@ -0,0 +1,18 @@
|
||||
-- v1.16.16 (iOS #17) — per-resource WHOOP sync cursors.
|
||||
--
|
||||
-- WHOOP's collection endpoints filter on a record's OWN time range, not on
|
||||
-- when it reached the cloud, and each resource (recovery / sleep / workout /
|
||||
-- cycle) re-scores and arrives on its own latency. A single shared
|
||||
-- `last_synced_at` cursor advanced by the fastest resource then drags the
|
||||
-- incremental `start` forward for a slow or failing resource too, so its
|
||||
-- late-arriving records fall permanently before the window and are never
|
||||
-- ingested. A 403/timeout on ONE collection likewise must not poison the
|
||||
-- cursor every sibling reads.
|
||||
--
|
||||
-- `resource_cursors` is a small JSON map `{ "recovery": "<iso>", ... }` of the
|
||||
-- per-resource last-synced instant. Each resource advances only its own key on
|
||||
-- a successful fetch+upsert; a stalled resource leaves its key untouched and
|
||||
-- keeps re-fetching from where it left off without holding back the others.
|
||||
-- Null = legacy connection — the readers fall back to the shared
|
||||
-- `last_synced_at` so no historical state is lost.
|
||||
ALTER TABLE "whoop_connections" ADD COLUMN "resource_cursors" JSONB;
|
||||
@@ -2168,6 +2168,15 @@ model WhoopConnection {
|
||||
/// a refresh token. Null = legacy connection that predates scope tracking.
|
||||
scope String? @map("scope")
|
||||
lastSyncedAt DateTime? @map("last_synced_at")
|
||||
/// v1.16.16 (iOS #17) — per-resource sync cursors. A JSON map
|
||||
/// `{ "recovery": "<iso>", "sleep": "<iso>", "workout": "<iso>",
|
||||
/// "cycle": "<iso>" }` of each resource's own last-synced instant. The
|
||||
/// incremental `start` for a resource is `cursor − overlap`; a resource
|
||||
/// advances ONLY its own key on a successful fetch+upsert, so a slow or
|
||||
/// 403-ing collection never drags the cursor forward for its siblings and
|
||||
/// its own late-arriving records stay inside the window. Null / a missing
|
||||
/// key falls back to the shared `lastSyncedAt` (legacy connection).
|
||||
resourceCursors Json? @map("resource_cursors")
|
||||
/// Set once the self-converging history backfill walks every collection to
|
||||
/// completion. Null = backfill still in flight (or never started).
|
||||
backfillCompletedAt DateTime? @map("backfill_completed_at")
|
||||
|
||||
@@ -162,8 +162,11 @@ describe("POST /api/whoop/webhook/[token]", () => {
|
||||
where: { whoopUserId: "42" },
|
||||
select: { userId: true },
|
||||
});
|
||||
// v1.16.16 (iOS #17) — the resource id rides through so the worker can do
|
||||
// a targeted fetch-by-id refresh instead of re-walking the collection.
|
||||
expect(send).toHaveBeenCalledWith("whoop-recovery-sync", {
|
||||
userId: "user-1",
|
||||
resourceId: "abc",
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,115 @@
|
||||
/**
|
||||
* v1.16.16 (iOS #17) — `runWhoopResourceSync` dispatch tests (mocked syncs).
|
||||
*
|
||||
* The webhook now carries the resource id; the worker must do a targeted
|
||||
* fetch-by-id refresh when it is present, fall back to the per-user collection
|
||||
* walk when a webhook job omits the id, and walk every connection only on the
|
||||
* cron tick (no userId on any job). One user's failure never starves the
|
||||
* cohort.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { Job } from "pg-boss";
|
||||
|
||||
const {
|
||||
syncUserWorkout,
|
||||
syncWhoopWorkoutById,
|
||||
findManyMock,
|
||||
} = vi.hoisted(() => ({
|
||||
syncUserWorkout: vi.fn<(...a: unknown[]) => Promise<number>>(async () => 0),
|
||||
syncWhoopWorkoutById: vi.fn<(...a: unknown[]) => Promise<number>>(
|
||||
async () => 0,
|
||||
),
|
||||
findManyMock: vi.fn<(...a: unknown[]) => Promise<Array<{ userId: string }>>>(
|
||||
async () => [],
|
||||
),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/whoop/sync-recovery", () => ({
|
||||
syncUserRecovery: vi.fn(async () => 0),
|
||||
syncWhoopRecoveryById: vi.fn(async () => 0),
|
||||
}));
|
||||
vi.mock("@/lib/whoop/sync-sleep", () => ({
|
||||
syncUserSleep: vi.fn(async () => 0),
|
||||
syncWhoopSleepById: vi.fn(async () => 0),
|
||||
}));
|
||||
vi.mock("@/lib/whoop/sync-cycle", () => ({
|
||||
syncUserCycle: vi.fn(async () => 0),
|
||||
}));
|
||||
vi.mock("@/lib/whoop/sync-workout", () => ({
|
||||
syncUserWorkout: (...a: unknown[]) => syncUserWorkout(...a),
|
||||
syncWhoopWorkoutById: (...a: unknown[]) => syncWhoopWorkoutById(...a),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/jobs/worker-status", () => ({ recordError: vi.fn() }));
|
||||
|
||||
vi.mock("@/lib/logging/background", () => ({
|
||||
withBackgroundEvent: async (
|
||||
_name: string,
|
||||
fn: (evt: {
|
||||
setBackground: () => void;
|
||||
setError: () => void;
|
||||
addWarning: () => void;
|
||||
}) => Promise<void>,
|
||||
) =>
|
||||
fn({
|
||||
setBackground: vi.fn(),
|
||||
setError: vi.fn(),
|
||||
addWarning: vi.fn(),
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock("../reminder/shared", () => ({
|
||||
getWorkerPrisma: () => ({
|
||||
whoopConnection: { findMany: (...a: unknown[]) => findManyMock(...a) },
|
||||
}),
|
||||
}));
|
||||
|
||||
import { handleWhoopWorkoutSync } from "../reminder/whoop-sync";
|
||||
|
||||
function job(data: { userId?: string; resourceId?: string }): Job<{
|
||||
userId?: string;
|
||||
resourceId?: string;
|
||||
}> {
|
||||
return { data } as Job<{ userId?: string; resourceId?: string }>;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
describe("runWhoopResourceSync dispatch", () => {
|
||||
it("a webhook job with a resourceId does a targeted fetch-by-id refresh", async () => {
|
||||
await handleWhoopWorkoutSync([job({ userId: "u1", resourceId: "w-9" })]);
|
||||
|
||||
expect(syncWhoopWorkoutById).toHaveBeenCalledWith("u1", "w-9");
|
||||
expect(syncUserWorkout).not.toHaveBeenCalled();
|
||||
expect(findManyMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("a webhook job without a resourceId falls back to the per-user collection", async () => {
|
||||
await handleWhoopWorkoutSync([job({ userId: "u1" })]);
|
||||
|
||||
expect(syncUserWorkout).toHaveBeenCalledWith("u1");
|
||||
expect(syncWhoopWorkoutById).not.toHaveBeenCalled();
|
||||
expect(findManyMock).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("a cron tick (no userId) walks every connection's collection", async () => {
|
||||
findManyMock.mockResolvedValue([{ userId: "a" }, { userId: "b" }]);
|
||||
|
||||
await handleWhoopWorkoutSync([job({})]);
|
||||
|
||||
expect(findManyMock).toHaveBeenCalled();
|
||||
expect(syncUserWorkout).toHaveBeenCalledTimes(2);
|
||||
expect(syncWhoopWorkoutById).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("one user's failure never starves the rest of the cohort", async () => {
|
||||
findManyMock.mockResolvedValue([{ userId: "a" }, { userId: "b" }]);
|
||||
syncUserWorkout.mockRejectedValueOnce(new Error("boom for a"));
|
||||
|
||||
await expect(handleWhoopWorkoutSync([job({})])).resolves.toBeUndefined();
|
||||
// Both users were attempted despite the first throwing.
|
||||
expect(syncUserWorkout).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
@@ -7,10 +7,19 @@
|
||||
import { type Job } from "pg-boss";
|
||||
import { recordError } from "@/lib/jobs/worker-status";
|
||||
import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
import { syncUserRecovery } from "@/lib/whoop/sync-recovery";
|
||||
import { syncUserSleep as syncWhoopSleep } from "@/lib/whoop/sync-sleep";
|
||||
import {
|
||||
syncUserRecovery,
|
||||
syncWhoopRecoveryById,
|
||||
} from "@/lib/whoop/sync-recovery";
|
||||
import {
|
||||
syncUserSleep as syncWhoopSleep,
|
||||
syncWhoopSleepById,
|
||||
} from "@/lib/whoop/sync-sleep";
|
||||
import { syncUserCycle } from "@/lib/whoop/sync-cycle";
|
||||
import { syncUserWorkout } from "@/lib/whoop/sync-workout";
|
||||
import {
|
||||
syncUserWorkout,
|
||||
syncWhoopWorkoutById,
|
||||
} from "@/lib/whoop/sync-workout";
|
||||
import { getWorkerPrisma } from "./shared";
|
||||
|
||||
/**
|
||||
@@ -18,34 +27,59 @@ import { getWorkerPrisma } from "./shared";
|
||||
* WHOOP sync queue:
|
||||
*
|
||||
* 1. Webhook (`recovery.updated` / `sleep.updated` / `workout.updated`) —
|
||||
* payload carries `userId`, the handler syncs that one user.
|
||||
* payload carries `userId` and (v1.16.16) the `resourceId`; the handler
|
||||
* does a targeted fetch-by-id refresh of that single record.
|
||||
* 2. Cron — payload has no `userId`; the handler iterates every WHOOP
|
||||
* connection and re-syncs each, catching dropped webhook deliveries.
|
||||
* Cycle has no webhook, so its cron is the sole driver.
|
||||
* connection and re-syncs each collection, catching dropped webhook
|
||||
* deliveries. Cycle has no webhook, so its cron is the sole driver.
|
||||
*
|
||||
* A webhook job that carries a `userId` but no `resourceId` (a legacy in-flight
|
||||
* job, or a delivery that omitted the id) falls back to the per-user collection
|
||||
* walk rather than touching every connection.
|
||||
*/
|
||||
export interface WhoopSyncPayload {
|
||||
userId?: string;
|
||||
/** v1.16.16 — webhook resource id for a targeted fetch-by-id refresh. */
|
||||
resourceId?: string;
|
||||
}
|
||||
|
||||
/** A per-user collection walk (cron + the no-id webhook fallback). */
|
||||
type CollectionSync = (userId: string) => Promise<number>;
|
||||
/** A webhook-driven single-record refresh by resource id. */
|
||||
type ByIdSync = (userId: string, resourceId: string) => Promise<number>;
|
||||
|
||||
/**
|
||||
* Shared driver for the per-resource WHOOP sync handlers. Resolves the target
|
||||
* set (per-user from the webhook payload, or every connection on the cron
|
||||
* tick) and runs `syncFn` per user. One user's parked-at-reauth state never
|
||||
* starves the rest of the cohort on the cron path.
|
||||
* Shared driver for the per-resource WHOOP sync handlers.
|
||||
*
|
||||
* - A webhook job with a `resourceId` runs the targeted `byIdFn` (when the
|
||||
* resource supports fetch-by-id) — landing the exact record immediately.
|
||||
* - A webhook job with a `userId` but no `resourceId` runs the per-user
|
||||
* collection `syncFn`.
|
||||
* - A cron tick (no `userId` on any job) walks every connection's collection.
|
||||
*
|
||||
* One user's parked-at-reauth state never starves the rest of the cohort.
|
||||
*/
|
||||
export async function runWhoopResourceSync(
|
||||
taskName: string,
|
||||
jobs: Job<WhoopSyncPayload>[],
|
||||
syncFn: (userId: string) => Promise<number>,
|
||||
syncFn: CollectionSync,
|
||||
byIdFn?: ByIdSync,
|
||||
): Promise<void> {
|
||||
await withBackgroundEvent(taskName, async (evt) => {
|
||||
const prisma = getWorkerPrisma();
|
||||
try {
|
||||
const targets: Array<{ userId: string }> = [];
|
||||
// (userId, optional resourceId) work items from the webhook payloads.
|
||||
const targets: Array<{ userId: string; resourceId?: string }> = [];
|
||||
for (const job of jobs) {
|
||||
if (job.data?.userId) targets.push({ userId: job.data.userId });
|
||||
if (job.data?.userId) {
|
||||
targets.push({
|
||||
userId: job.data.userId,
|
||||
resourceId: job.data.resourceId,
|
||||
});
|
||||
}
|
||||
}
|
||||
if (targets.length === 0) {
|
||||
// Cron tick — re-walk every connection's collection.
|
||||
const connections = await prisma.whoopConnection.findMany({
|
||||
select: { userId: true },
|
||||
});
|
||||
@@ -55,9 +89,12 @@ export async function runWhoopResourceSync(
|
||||
|
||||
let usersSynced = 0;
|
||||
let measurementsImported = 0;
|
||||
for (const { userId } of targets) {
|
||||
for (const { userId, resourceId } of targets) {
|
||||
try {
|
||||
measurementsImported += await syncFn(userId);
|
||||
measurementsImported +=
|
||||
resourceId && byIdFn
|
||||
? await byIdFn(userId, resourceId)
|
||||
: await syncFn(userId);
|
||||
usersSynced++;
|
||||
} catch (err) {
|
||||
evt.addWarning(`${taskName} failed for user ${userId}: ${err}`);
|
||||
@@ -85,17 +122,29 @@ export function handleWhoopRecoverySync(jobs: Job<WhoopSyncPayload>[]) {
|
||||
"job.whoop_recovery_sync",
|
||||
jobs,
|
||||
syncUserRecovery,
|
||||
syncWhoopRecoveryById,
|
||||
);
|
||||
}
|
||||
|
||||
export function handleWhoopSleepSync(jobs: Job<WhoopSyncPayload>[]) {
|
||||
return runWhoopResourceSync("job.whoop_sleep_sync", jobs, syncWhoopSleep);
|
||||
return runWhoopResourceSync(
|
||||
"job.whoop_sleep_sync",
|
||||
jobs,
|
||||
syncWhoopSleep,
|
||||
syncWhoopSleepById,
|
||||
);
|
||||
}
|
||||
|
||||
export function handleWhoopWorkoutSync(jobs: Job<WhoopSyncPayload>[]) {
|
||||
return runWhoopResourceSync("job.whoop_workout_sync", jobs, syncUserWorkout);
|
||||
return runWhoopResourceSync(
|
||||
"job.whoop_workout_sync",
|
||||
jobs,
|
||||
syncUserWorkout,
|
||||
syncWhoopWorkoutById,
|
||||
);
|
||||
}
|
||||
|
||||
export function handleWhoopCycleSync(jobs: Job<WhoopSyncPayload>[]) {
|
||||
// Cycle has no webhook (poll-only), so no fetch-by-id path.
|
||||
return runWhoopResourceSync("job.whoop_cycle_sync", jobs, syncUserCycle);
|
||||
}
|
||||
@@ -6,6 +6,10 @@ import {
|
||||
WHOOP_PAGE_LIMIT,
|
||||
exchangeCode,
|
||||
fetchBodyMeasurement,
|
||||
fetchRecoveryByCycleId,
|
||||
fetchSleepById,
|
||||
fetchWorkoutById,
|
||||
fetchCycleById,
|
||||
fetchCycles,
|
||||
fetchProfile,
|
||||
fetchRecoveries,
|
||||
@@ -198,6 +202,39 @@ describe("single-object endpoints", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("fetch-by-id (webhook-driven single-record refresh)", () => {
|
||||
it("resolves one record at the documented v2 single-record path", async () => {
|
||||
const cases: Array<[() => Promise<unknown>, string]> = [
|
||||
[() => fetchSleepById("at", "s1"), "/v2/activity/sleep/s1"],
|
||||
[() => fetchWorkoutById("at", "w1"), "/v2/activity/workout/w1"],
|
||||
[() => fetchCycleById("at", "123"), "/v2/cycle/123"],
|
||||
[() => fetchRecoveryByCycleId("at", "123"), "/v2/cycle/123/recovery"],
|
||||
];
|
||||
for (const [call, path] of cases) {
|
||||
const fetchMock = installFetchMock([{ status: 200, body: { id: "x" } }]);
|
||||
await call();
|
||||
const [url] = fetchMock.mock.calls[0] as unknown as [string];
|
||||
expect(url).toContain(path);
|
||||
vi.unstubAllGlobals();
|
||||
}
|
||||
});
|
||||
|
||||
it("returns the single record body (one workout by id)", async () => {
|
||||
installFetchMock([
|
||||
{ status: 200, body: { id: "w1", start: "s", end: "e" } },
|
||||
]);
|
||||
const w = await fetchWorkoutById("at", "w1");
|
||||
expect(w.id).toBe("w1");
|
||||
});
|
||||
|
||||
it("url-encodes a resource id with reserved characters", async () => {
|
||||
const fetchMock = installFetchMock([{ status: 200, body: { id: "x" } }]);
|
||||
await fetchWorkoutById("at", "a/b?c");
|
||||
const [url] = fetchMock.mock.calls[0] as unknown as [string];
|
||||
expect(url).toContain("/v2/activity/workout/a%2Fb%3Fc");
|
||||
});
|
||||
});
|
||||
|
||||
describe("mapRecovery", () => {
|
||||
const base: WhoopRecovery = {
|
||||
cycle_id: 1,
|
||||
|
||||
@@ -9,30 +9,33 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const fetchWorkouts = vi.fn();
|
||||
const fetchWorkoutById = vi.fn();
|
||||
|
||||
vi.mock("../client", () => ({
|
||||
fetchWorkouts: (...a: unknown[]) => fetchWorkouts(...a),
|
||||
fetchWorkoutById: (...a: unknown[]) => fetchWorkoutById(...a),
|
||||
KJ_TO_KCAL: 0.239006,
|
||||
}));
|
||||
|
||||
const getValidToken = vi.fn();
|
||||
const markSynced = vi.fn();
|
||||
const markResourceSynced = vi.fn();
|
||||
vi.mock("../sync", async () => {
|
||||
const actual = await vi.importActual<typeof import("../sync")>("../sync");
|
||||
return {
|
||||
...actual,
|
||||
getValidToken: (...a: unknown[]) => getValidToken(...a),
|
||||
markSynced: (...a: unknown[]) => markSynced(...a),
|
||||
markResourceSynced: (...a: unknown[]) => markResourceSynced(...a),
|
||||
};
|
||||
});
|
||||
|
||||
const whoopConnectionFindUnique = vi.fn();
|
||||
const workoutUpsert = vi.fn();
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
whoopConnection: {
|
||||
findUnique: (...a: unknown[]) => whoopConnectionFindUnique(...a),
|
||||
},
|
||||
workout: { upsert: vi.fn() },
|
||||
workout: { upsert: (...a: unknown[]) => workoutUpsert(...a) },
|
||||
},
|
||||
}));
|
||||
|
||||
@@ -40,8 +43,11 @@ vi.mock("@/lib/logging/context", () => ({
|
||||
getEvent: () => ({ addWarning: vi.fn() }),
|
||||
}));
|
||||
|
||||
import { syncUserWorkout } from "../sync-workout";
|
||||
import { WHOOP_RECOVERY_SLEEP_OVERLAP_MS } from "../sync";
|
||||
import { syncUserWorkout, syncWhoopWorkoutById } from "../sync-workout";
|
||||
import {
|
||||
WHOOP_FULL_SYNC_ANCHOR,
|
||||
WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
} from "../sync";
|
||||
|
||||
const LAST_SYNCED = new Date("2026-06-10T12:00:00.000Z");
|
||||
|
||||
@@ -51,8 +57,12 @@ beforeEach(() => {
|
||||
accessToken: "tok",
|
||||
connection: { id: "c1", whoopUserId: "w1" },
|
||||
});
|
||||
whoopConnectionFindUnique.mockResolvedValue({ lastSyncedAt: LAST_SYNCED });
|
||||
markSynced.mockResolvedValue(undefined);
|
||||
// No per-resource cursor yet → falls back to the shared lastSyncedAt.
|
||||
whoopConnectionFindUnique.mockResolvedValue({
|
||||
lastSyncedAt: LAST_SYNCED,
|
||||
resourceCursors: null,
|
||||
});
|
||||
markResourceSynced.mockResolvedValue(undefined);
|
||||
fetchWorkouts.mockResolvedValue([]);
|
||||
});
|
||||
|
||||
@@ -70,9 +80,81 @@ describe("syncUserWorkout — late-synced workout overlap window", () => {
|
||||
expect(arg.start!.getTime()).toBeLessThanOrEqual(sixHoursBefore);
|
||||
});
|
||||
|
||||
it("fullSync passes no start (deep backfill anchor handles history)", async () => {
|
||||
it("fullSync anchors at the deep-history anchor, not the cursor", async () => {
|
||||
await syncUserWorkout("user-1", { fullSync: true });
|
||||
const arg = fetchWorkouts.mock.calls[0]![1] as { start?: Date };
|
||||
expect(arg.start).toBeUndefined();
|
||||
expect(arg.start).toEqual(WHOOP_FULL_SYNC_ANCHOR);
|
||||
});
|
||||
|
||||
it("uses the per-resource workout cursor when present, ignoring siblings", async () => {
|
||||
const workoutCursor = new Date("2026-06-04T12:00:00.000Z");
|
||||
whoopConnectionFindUnique.mockResolvedValue({
|
||||
// recovery synced far more recently — must NOT pull the workout window
|
||||
// forward.
|
||||
lastSyncedAt: new Date("2026-06-12T00:00:00.000Z"),
|
||||
resourceCursors: {
|
||||
recovery: "2026-06-12T00:00:00.000Z",
|
||||
workout: workoutCursor.toISOString(),
|
||||
},
|
||||
});
|
||||
|
||||
await syncUserWorkout("user-1");
|
||||
|
||||
const arg = fetchWorkouts.mock.calls[0]![1] as { start?: Date };
|
||||
expect(arg.start!.getTime()).toBe(
|
||||
workoutCursor.getTime() - WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("syncWhoopWorkoutById — webhook fetch-by-id", () => {
|
||||
const SCORED_WORKOUT = {
|
||||
id: "w-123",
|
||||
start: "2026-06-14T07:00:00.000Z",
|
||||
end: "2026-06-14T08:00:00.000Z",
|
||||
sport_id: 1,
|
||||
score: {
|
||||
strain: 12.3,
|
||||
average_heart_rate: 130,
|
||||
max_heart_rate: 165,
|
||||
kilojoule: 1500,
|
||||
percent_recorded: 99,
|
||||
},
|
||||
};
|
||||
|
||||
it("resolves the ONE record by id and upserts it (no collection walk)", async () => {
|
||||
fetchWorkoutById.mockResolvedValue(SCORED_WORKOUT);
|
||||
workoutUpsert.mockResolvedValue({});
|
||||
|
||||
const imported = await syncWhoopWorkoutById("user-1", "w-123");
|
||||
|
||||
expect(imported).toBe(1);
|
||||
// Targeted fetch-by-id, never the collection.
|
||||
expect(fetchWorkoutById).toHaveBeenCalledWith("tok", "w-123");
|
||||
expect(fetchWorkouts).not.toHaveBeenCalled();
|
||||
const upsertArg = workoutUpsert.mock.calls[0]![0];
|
||||
expect(upsertArg.where.userId_source_externalId).toEqual({
|
||||
userId: "user-1",
|
||||
source: "WHOOP",
|
||||
externalId: "w-123",
|
||||
});
|
||||
});
|
||||
|
||||
it("a single-id refresh does NOT advance the resource cursor", async () => {
|
||||
fetchWorkoutById.mockResolvedValue(SCORED_WORKOUT);
|
||||
workoutUpsert.mockResolvedValue({});
|
||||
|
||||
await syncWhoopWorkoutById("user-1", "w-123");
|
||||
|
||||
expect(markResourceSynced).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("an unscored record stores nothing", async () => {
|
||||
fetchWorkoutById.mockResolvedValue({ ...SCORED_WORKOUT, score: null });
|
||||
|
||||
const imported = await syncWhoopWorkoutById("user-1", "w-123");
|
||||
|
||||
expect(imported).toBe(0);
|
||||
expect(workoutUpsert).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -25,6 +25,7 @@ const {
|
||||
measurement: {
|
||||
upsert: vi.fn(),
|
||||
},
|
||||
$executeRaw: vi.fn(),
|
||||
},
|
||||
refreshAccessTokenMock: vi.fn(),
|
||||
fetchRecoveriesMock: vi.fn(),
|
||||
@@ -88,8 +89,11 @@ import {
|
||||
getValidToken,
|
||||
incrementalStart,
|
||||
isCollectionForbidden,
|
||||
markResourceSynced,
|
||||
resolveResourceCursor,
|
||||
upsertWhoopMeasurements,
|
||||
WHOOP_DEFAULT_OVERLAP_MS,
|
||||
WHOOP_FULL_SYNC_ANCHOR,
|
||||
WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
} from "../sync";
|
||||
import { WhoopApiError } from "../response-classifier";
|
||||
@@ -172,8 +176,11 @@ describe("getValidToken — rotating refresh", () => {
|
||||
});
|
||||
|
||||
describe("incrementalStart", () => {
|
||||
it("returns undefined for a full sync", () => {
|
||||
expect(incrementalStart(new Date(), { fullSync: true })).toBeUndefined();
|
||||
it("anchors a full sync at the deep-history anchor, ignoring the cursor", () => {
|
||||
const got = incrementalStart(new Date("2026-06-01T12:00:00Z"), {
|
||||
fullSync: true,
|
||||
});
|
||||
expect(got).toEqual(WHOOP_FULL_SYNC_ANCHOR);
|
||||
});
|
||||
|
||||
it("subtracts the overlap from lastSyncedAt", () => {
|
||||
@@ -191,6 +198,109 @@ describe("incrementalStart", () => {
|
||||
const got = incrementalStart(last);
|
||||
expect(got!.getTime()).toBe(last.getTime() - WHOOP_DEFAULT_OVERLAP_MS);
|
||||
});
|
||||
|
||||
it("a fullSync reaches a span an incremental tick would skip", () => {
|
||||
// Incremental from a recent cursor only looks back the overlap (days); a
|
||||
// fullSync anchors years back, so it covers history the incremental misses.
|
||||
const recentCursor = new Date("2026-06-01T12:00:00Z");
|
||||
const incremental = incrementalStart(recentCursor, {
|
||||
overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
});
|
||||
const full = incrementalStart(recentCursor, { fullSync: true });
|
||||
const yearOld = new Date("2025-01-01T00:00:00Z");
|
||||
// A year-old record is BEFORE the incremental start (skipped) but AFTER
|
||||
// the full-sync anchor (covered).
|
||||
expect(yearOld.getTime()).toBeLessThan(incremental!.getTime());
|
||||
expect(yearOld.getTime()).toBeGreaterThan(full!.getTime());
|
||||
});
|
||||
});
|
||||
|
||||
describe("resolveResourceCursor — per-resource cursor", () => {
|
||||
it("prefers the per-resource cursor over the shared lastSyncedAt", () => {
|
||||
const got = resolveResourceCursor(
|
||||
{
|
||||
resourceCursors: { workout: "2026-06-10T00:00:00.000Z" },
|
||||
lastSyncedAt: new Date("2026-06-12T00:00:00.000Z"),
|
||||
},
|
||||
"workout",
|
||||
);
|
||||
expect(got?.toISOString()).toBe("2026-06-10T00:00:00.000Z");
|
||||
});
|
||||
|
||||
it("falls back to lastSyncedAt for a resource with no cursor key (legacy)", () => {
|
||||
const shared = new Date("2026-06-12T00:00:00.000Z");
|
||||
const got = resolveResourceCursor(
|
||||
{ resourceCursors: { recovery: "2026-06-10T00:00:00.000Z" }, lastSyncedAt: shared },
|
||||
"workout",
|
||||
);
|
||||
expect(got).toEqual(shared);
|
||||
});
|
||||
|
||||
it("falls back to lastSyncedAt when the column is null", () => {
|
||||
const shared = new Date("2026-06-12T00:00:00.000Z");
|
||||
expect(
|
||||
resolveResourceCursor({ resourceCursors: null, lastSyncedAt: shared }, "sleep"),
|
||||
).toEqual(shared);
|
||||
});
|
||||
|
||||
it("a stalled resource cursor does not advance with a sibling's", () => {
|
||||
// workout last synced a week ago; recovery synced an hour ago. Resolving
|
||||
// workout must still return the OLD workout cursor — a sibling's progress
|
||||
// never drags the stalled resource's window forward.
|
||||
const connection = {
|
||||
resourceCursors: {
|
||||
recovery: "2026-06-12T11:00:00.000Z",
|
||||
workout: "2026-06-05T12:00:00.000Z",
|
||||
},
|
||||
lastSyncedAt: new Date("2026-06-12T11:00:00.000Z"),
|
||||
};
|
||||
const workoutStart = incrementalStart(
|
||||
resolveResourceCursor(connection, "workout"),
|
||||
{ overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS },
|
||||
);
|
||||
// Workout still re-fetches from a week-old cursor minus the overlap, NOT
|
||||
// from recovery's recent one — the late-synced workout stays in range.
|
||||
expect(workoutStart!.getTime()).toBe(
|
||||
new Date("2026-06-05T12:00:00.000Z").getTime() -
|
||||
WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("markResourceSynced — atomic, independent cursor advance", () => {
|
||||
it("issues ONE atomic jsonb_set upsert binding the resource key + instant", async () => {
|
||||
prismaMock.$executeRaw.mockResolvedValue(1);
|
||||
|
||||
const at = new Date("2026-06-14T08:00:00.000Z");
|
||||
await markResourceSynced("user1", "workout", at);
|
||||
|
||||
// A single atomic statement, never a read-then-write (which would race the
|
||||
// concurrent sibling resource queues).
|
||||
expect(prismaMock.whoopConnection.findUnique).not.toHaveBeenCalled();
|
||||
expect(prismaMock.whoopConnection.update).not.toHaveBeenCalled();
|
||||
expect(prismaMock.$executeRaw).toHaveBeenCalledTimes(1);
|
||||
|
||||
const call = prismaMock.$executeRaw.mock.calls[0]!;
|
||||
const sql = (call[0] as TemplateStringsArray).join("?");
|
||||
const values = call.slice(1);
|
||||
// Merge-in-place + monotonic last_synced_at + scoped to the user.
|
||||
expect(sql).toContain("jsonb_set");
|
||||
expect(sql).toContain("GREATEST");
|
||||
// Resource key + ISO instant are parameter-bound, not spliced.
|
||||
expect(values).toContain("workout");
|
||||
expect(values).toContain(at.toISOString());
|
||||
expect(values).toContain(at);
|
||||
expect(values).toContain("user1");
|
||||
});
|
||||
|
||||
it("refuses a resource outside the closed whitelist (no statement issued)", async () => {
|
||||
await markResourceSynced(
|
||||
"user1",
|
||||
"evil" as unknown as Parameters<typeof markResourceSynced>[1],
|
||||
new Date(),
|
||||
);
|
||||
expect(prismaMock.$executeRaw).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("upsertWhoopMeasurements — idempotent upsert", () => {
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
/**
|
||||
* v1.16.16 (iOS #17) — `processWhoopNotification` enqueue tests (mocked).
|
||||
*
|
||||
* A `*.updated` webhook must enqueue the per-resource sync job carrying the
|
||||
* resource id so the worker can do a targeted fetch-by-id refresh (landing the
|
||||
* exact record immediately rather than waiting for the next overlap window).
|
||||
* The existing 200-always + unknown-user + delete-soft-delete contract is
|
||||
* preserved.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const { findFirst, bossSend, updateMany, deleteMany } = vi.hoisted(() => ({
|
||||
findFirst: vi.fn(),
|
||||
bossSend: vi.fn(),
|
||||
updateMany: vi.fn(),
|
||||
deleteMany: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
whoopConnection: { findFirst: (...a: unknown[]) => findFirst(...a) },
|
||||
measurement: { updateMany: (...a: unknown[]) => updateMany(...a) },
|
||||
workout: { deleteMany: (...a: unknown[]) => deleteMany(...a) },
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: () => ({ send: (...a: unknown[]) => bossSend(...a) }),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/rate-limit", () => ({
|
||||
checkRateLimit: vi.fn(),
|
||||
rateLimitHeaders: () => ({}),
|
||||
}));
|
||||
vi.mock("@/lib/api-response", () => ({ getClientIp: () => "1.2.3.4" }));
|
||||
vi.mock("@/lib/logging/context", () => ({
|
||||
annotate: () => {},
|
||||
getEvent: () => ({ setAuth: vi.fn(), addWarning: vi.fn() }),
|
||||
}));
|
||||
|
||||
import { processWhoopNotification } from "../webhook-handler";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
findFirst.mockResolvedValue({ userId: "user-1" });
|
||||
bossSend.mockResolvedValue("job-id");
|
||||
});
|
||||
|
||||
describe("processWhoopNotification — fetch-by-id enqueue", () => {
|
||||
it("enqueues the workout queue with the resource id", async () => {
|
||||
const res = await processWhoopNotification({
|
||||
user_id: 42,
|
||||
id: "w-123",
|
||||
type: "workout.updated",
|
||||
});
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(bossSend).toHaveBeenCalledWith("whoop-workout-sync", {
|
||||
userId: "user-1",
|
||||
resourceId: "w-123",
|
||||
});
|
||||
});
|
||||
|
||||
it("carries a numeric resource id (cycle-based recovery) as a string", async () => {
|
||||
await processWhoopNotification({
|
||||
user_id: 42,
|
||||
id: 9981,
|
||||
type: "recovery.updated",
|
||||
});
|
||||
|
||||
expect(bossSend).toHaveBeenCalledWith("whoop-recovery-sync", {
|
||||
userId: "user-1",
|
||||
resourceId: "9981",
|
||||
});
|
||||
});
|
||||
|
||||
it("returns 200 for an unknown user without enqueuing", async () => {
|
||||
findFirst.mockResolvedValue(null);
|
||||
|
||||
const res = await processWhoopNotification({
|
||||
user_id: 7,
|
||||
id: "w-1",
|
||||
type: "workout.updated",
|
||||
});
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(bossSend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("soft-deletes on a *.deleted event (no enqueue)", async () => {
|
||||
updateMany.mockResolvedValue({ count: 3 });
|
||||
deleteMany.mockResolvedValue({ count: 1 });
|
||||
|
||||
const res = await processWhoopNotification({
|
||||
user_id: 42,
|
||||
id: "w-123",
|
||||
type: "workout.deleted",
|
||||
});
|
||||
|
||||
expect(res.status).toBe(200);
|
||||
expect(bossSend).not.toHaveBeenCalled();
|
||||
expect(deleteMany).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -402,6 +402,67 @@ export function fetchWorkouts(
|
||||
);
|
||||
}
|
||||
|
||||
// ─── Single-record reads (webhook-driven fetch-by-id) ──────────
|
||||
// A `*.updated` webhook carries only the resource id, not the data. Re-fetch
|
||||
// that ONE record by id for a targeted refresh — far cheaper than re-walking
|
||||
// the whole collection, and it lands the record immediately rather than on the
|
||||
// next overlap window. WHOOP v2 single-record paths (re-verify at build):
|
||||
// sleep GET /v2/activity/sleep/{sleepId}
|
||||
// workout GET /v2/activity/workout/{workoutId}
|
||||
// cycle GET /v2/cycle/{cycleId}
|
||||
// recovery GET /v2/cycle/{cycleId}/recovery (no recovery-by-recovery-id)
|
||||
// `fetchSingle` throws a `WhoopApiError` on a non-2xx (incl. a 404 for a since-
|
||||
// deleted id) so the caller's existing classify/soft-skip path handles it.
|
||||
|
||||
export function fetchSleepById(
|
||||
accessToken: string,
|
||||
sleepId: string,
|
||||
): Promise<WhoopSleep> {
|
||||
return fetchSingle<WhoopSleep>(
|
||||
`/v2/activity/sleep/${encodeURIComponent(sleepId)}`,
|
||||
accessToken,
|
||||
"fetchSleepById",
|
||||
);
|
||||
}
|
||||
|
||||
export function fetchWorkoutById(
|
||||
accessToken: string,
|
||||
workoutId: string,
|
||||
): Promise<WhoopWorkout> {
|
||||
return fetchSingle<WhoopWorkout>(
|
||||
`/v2/activity/workout/${encodeURIComponent(workoutId)}`,
|
||||
accessToken,
|
||||
"fetchWorkoutById",
|
||||
);
|
||||
}
|
||||
|
||||
export function fetchCycleById(
|
||||
accessToken: string,
|
||||
cycleId: string,
|
||||
): Promise<WhoopCycle> {
|
||||
return fetchSingle<WhoopCycle>(
|
||||
`/v2/cycle/${encodeURIComponent(cycleId)}`,
|
||||
accessToken,
|
||||
"fetchCycleById",
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* WHOOP v2 has no recovery-by-recovery-id route — a recovery is read through
|
||||
* its cycle (`GET /v2/cycle/{cycleId}/recovery`). The recovery webhook carries
|
||||
* the cycle id, so the targeted refresh resolves the recovery from there.
|
||||
*/
|
||||
export function fetchRecoveryByCycleId(
|
||||
accessToken: string,
|
||||
cycleId: string,
|
||||
): Promise<WhoopRecovery> {
|
||||
return fetchSingle<WhoopRecovery>(
|
||||
`/v2/cycle/${encodeURIComponent(cycleId)}/recovery`,
|
||||
accessToken,
|
||||
"fetchRecoveryByCycleId",
|
||||
);
|
||||
}
|
||||
|
||||
export function fetchBodyMeasurement(
|
||||
accessToken: string,
|
||||
): Promise<WhoopBodyMeasurement> {
|
||||
|
||||
@@ -10,7 +10,8 @@ import {
|
||||
getValidToken,
|
||||
incrementalStart,
|
||||
handleCollectionFetchError,
|
||||
markSynced,
|
||||
markResourceSynced,
|
||||
resolveResourceCursor,
|
||||
upsertWhoopMeasurements,
|
||||
type WhoopMeasurementUpsert,
|
||||
} from "./sync";
|
||||
@@ -25,11 +26,11 @@ export async function syncUserCycle(
|
||||
|
||||
const connection = await prisma.whoopConnection.findUnique({
|
||||
where: { userId },
|
||||
select: { lastSyncedAt: true },
|
||||
select: { lastSyncedAt: true, resourceCursors: true },
|
||||
});
|
||||
if (!connection) return 0;
|
||||
|
||||
const start = incrementalStart(connection.lastSyncedAt, {
|
||||
const start = incrementalStart(resolveResourceCursor(connection, "cycle"), {
|
||||
fullSync: opts.fullSync,
|
||||
});
|
||||
|
||||
@@ -54,6 +55,6 @@ export async function syncUserCycle(
|
||||
}
|
||||
|
||||
const imported = await upsertWhoopMeasurements(userId, readings);
|
||||
await markSynced(userId);
|
||||
await markResourceSynced(userId, "cycle");
|
||||
return imported;
|
||||
}
|
||||
@@ -9,12 +9,17 @@
|
||||
* sleep UUID (v2 recovery carries `sleep_id`, not a stable recovery id) so the
|
||||
* externalId is stable across re-scores.
|
||||
*/
|
||||
import { fetchRecoveries, mapRecovery } from "./client";
|
||||
import {
|
||||
fetchRecoveries,
|
||||
fetchRecoveryByCycleId,
|
||||
mapRecovery,
|
||||
} from "./client";
|
||||
import {
|
||||
getValidToken,
|
||||
incrementalStart,
|
||||
handleCollectionFetchError,
|
||||
markSynced,
|
||||
markResourceSynced,
|
||||
resolveResourceCursor,
|
||||
upsertWhoopMeasurements,
|
||||
WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
type WhoopMeasurementUpsert,
|
||||
@@ -30,14 +35,17 @@ export async function syncUserRecovery(
|
||||
|
||||
const connection = await prisma.whoopConnection.findUnique({
|
||||
where: { userId },
|
||||
select: { lastSyncedAt: true },
|
||||
select: { lastSyncedAt: true, resourceCursors: true },
|
||||
});
|
||||
if (!connection) return 0;
|
||||
|
||||
const start = incrementalStart(connection.lastSyncedAt, {
|
||||
fullSync: opts.fullSync,
|
||||
overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
});
|
||||
const start = incrementalStart(
|
||||
resolveResourceCursor(connection, "recovery"),
|
||||
{
|
||||
fullSync: opts.fullSync,
|
||||
overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
},
|
||||
);
|
||||
|
||||
let records: Awaited<ReturnType<typeof fetchRecoveries>>;
|
||||
try {
|
||||
@@ -62,6 +70,43 @@ export async function syncUserRecovery(
|
||||
}
|
||||
|
||||
const imported = await upsertWhoopMeasurements(userId, readings);
|
||||
await markSynced(userId);
|
||||
await markResourceSynced(userId, "recovery");
|
||||
return imported;
|
||||
}
|
||||
|
||||
/**
|
||||
* Webhook-driven targeted refresh: resolve ONE recovery by its cycle id and
|
||||
* upsert its readings, instead of re-walking the whole collection. WHOOP v2
|
||||
* reads a recovery through its cycle (`/v2/cycle/{cycleId}/recovery`), and the
|
||||
* recovery webhook carries the cycle id. An unscored / since-deleted record
|
||||
* yields nothing. Does NOT advance the resource cursor — a single-id refresh
|
||||
* proves nothing about records between the cursor and now, so the cron/overlap
|
||||
* path stays responsible for moving the cursor.
|
||||
*/
|
||||
export async function syncWhoopRecoveryById(
|
||||
userId: string,
|
||||
cycleId: string,
|
||||
): Promise<number> {
|
||||
const tokenInfo = await getValidToken(userId);
|
||||
if (!tokenInfo) return 0;
|
||||
|
||||
let record: Awaited<ReturnType<typeof fetchRecoveryByCycleId>>;
|
||||
try {
|
||||
record = await fetchRecoveryByCycleId(tokenInfo.accessToken, cycleId);
|
||||
} catch (err) {
|
||||
return handleCollectionFetchError("recovery", userId, err);
|
||||
}
|
||||
|
||||
const readings: WhoopMeasurementUpsert[] = [];
|
||||
for (const m of mapRecovery(record)) {
|
||||
readings.push({
|
||||
type: m.type,
|
||||
value: m.value,
|
||||
unit: m.unit,
|
||||
measuredAt: m.measuredAt,
|
||||
externalId: `${record.sleep_id}:${m.fieldTag}`,
|
||||
});
|
||||
}
|
||||
|
||||
return upsertWhoopMeasurements(userId, readings);
|
||||
}
|
||||
@@ -7,12 +7,13 @@
|
||||
* Per-stage rows carry the `sleepStage` axis so the five stage rows for one
|
||||
* night stay distinct under the dedup key. externalId = `<sleep_id>:<fieldTag>`.
|
||||
*/
|
||||
import { fetchSleeps, mapSleep } from "./client";
|
||||
import { fetchSleeps, fetchSleepById, mapSleep } from "./client";
|
||||
import {
|
||||
getValidToken,
|
||||
incrementalStart,
|
||||
handleCollectionFetchError,
|
||||
markSynced,
|
||||
markResourceSynced,
|
||||
resolveResourceCursor,
|
||||
upsertWhoopMeasurements,
|
||||
WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
type WhoopMeasurementUpsert,
|
||||
@@ -28,11 +29,11 @@ export async function syncUserSleep(
|
||||
|
||||
const connection = await prisma.whoopConnection.findUnique({
|
||||
where: { userId },
|
||||
select: { lastSyncedAt: true },
|
||||
select: { lastSyncedAt: true, resourceCursors: true },
|
||||
});
|
||||
if (!connection) return 0;
|
||||
|
||||
const start = incrementalStart(connection.lastSyncedAt, {
|
||||
const start = incrementalStart(resolveResourceCursor(connection, "sleep"), {
|
||||
fullSync: opts.fullSync,
|
||||
overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
});
|
||||
@@ -59,6 +60,42 @@ export async function syncUserSleep(
|
||||
}
|
||||
|
||||
const imported = await upsertWhoopMeasurements(userId, readings);
|
||||
await markSynced(userId);
|
||||
await markResourceSynced(userId, "sleep");
|
||||
return imported;
|
||||
}
|
||||
|
||||
/**
|
||||
* Webhook-driven targeted refresh: resolve ONE sleep activity by its uuid and
|
||||
* upsert its readings, instead of re-walking the whole collection. An unscored
|
||||
* / since-deleted record yields nothing. Does NOT advance the resource cursor
|
||||
* (a single-id refresh proves nothing about the records between the cursor and
|
||||
* now).
|
||||
*/
|
||||
export async function syncWhoopSleepById(
|
||||
userId: string,
|
||||
sleepId: string,
|
||||
): Promise<number> {
|
||||
const tokenInfo = await getValidToken(userId);
|
||||
if (!tokenInfo) return 0;
|
||||
|
||||
let record: Awaited<ReturnType<typeof fetchSleepById>>;
|
||||
try {
|
||||
record = await fetchSleepById(tokenInfo.accessToken, sleepId);
|
||||
} catch (err) {
|
||||
return handleCollectionFetchError("sleep", userId, err);
|
||||
}
|
||||
|
||||
const readings: WhoopMeasurementUpsert[] = [];
|
||||
for (const m of mapSleep(record)) {
|
||||
readings.push({
|
||||
type: m.type,
|
||||
value: m.value,
|
||||
unit: m.unit,
|
||||
measuredAt: m.measuredAt,
|
||||
externalId: `${record.id}:${m.fieldTag}`,
|
||||
sleepStage: m.sleepStage ?? null,
|
||||
});
|
||||
}
|
||||
|
||||
return upsertWhoopMeasurements(userId, readings);
|
||||
}
|
||||
+113
-71
@@ -13,12 +13,18 @@
|
||||
* rows (different `source`); the read-time `pickCanonicalWorkoutRows` picker
|
||||
* (E-slice, W6) collapses the cross-source twin at read time.
|
||||
*/
|
||||
import { fetchWorkouts, KJ_TO_KCAL } from "./client";
|
||||
import {
|
||||
fetchWorkouts,
|
||||
fetchWorkoutById,
|
||||
KJ_TO_KCAL,
|
||||
type WhoopWorkout,
|
||||
} from "./client";
|
||||
import {
|
||||
getValidToken,
|
||||
incrementalStart,
|
||||
handleCollectionFetchError,
|
||||
markSynced,
|
||||
markResourceSynced,
|
||||
resolveResourceCursor,
|
||||
WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
} from "./sync";
|
||||
import { prisma } from "@/lib/db";
|
||||
@@ -41,7 +47,7 @@ export async function syncUserWorkout(
|
||||
|
||||
const connection = await prisma.whoopConnection.findUnique({
|
||||
where: { userId },
|
||||
select: { lastSyncedAt: true },
|
||||
select: { lastSyncedAt: true, resourceCursors: true },
|
||||
});
|
||||
if (!connection) return 0;
|
||||
|
||||
@@ -53,10 +59,13 @@ export async function syncUserWorkout(
|
||||
// the same window recovery/sleep already use for their late re-scores. The
|
||||
// `(userId, WHOOP, externalId)` upsert keeps the re-fetch idempotent and a
|
||||
// handful of workouts/day keeps the page count down.
|
||||
const start = incrementalStart(connection.lastSyncedAt, {
|
||||
fullSync: opts.fullSync,
|
||||
overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
});
|
||||
const start = incrementalStart(
|
||||
resolveResourceCursor(connection, "workout"),
|
||||
{
|
||||
fullSync: opts.fullSync,
|
||||
overlapMs: WHOOP_RECOVERY_SLEEP_OVERLAP_MS,
|
||||
},
|
||||
);
|
||||
|
||||
let records: Awaited<ReturnType<typeof fetchWorkouts>>;
|
||||
try {
|
||||
@@ -67,75 +76,108 @@ export async function syncUserWorkout(
|
||||
|
||||
let imported = 0;
|
||||
for (const w of records) {
|
||||
if (!w.score) continue; // unscored workout — nothing to store yet
|
||||
const startedAt = new Date(w.start);
|
||||
const endedAt = new Date(w.end);
|
||||
const durationSec = Math.max(
|
||||
0,
|
||||
Math.round((endedAt.getTime() - startedAt.getTime()) / 1000),
|
||||
);
|
||||
const energyKcal =
|
||||
typeof w.score.kilojoule === "number"
|
||||
? Math.round(w.score.kilojoule * KJ_TO_KCAL)
|
||||
: null;
|
||||
imported += await upsertWhoopWorkout(userId, w);
|
||||
}
|
||||
|
||||
const metadata: Prisma.InputJsonValue = {
|
||||
whoopWorkoutStrain: w.score.strain,
|
||||
percentRecorded: w.score.percent_recorded,
|
||||
...(w.score.altitude_gain_meter != null
|
||||
? { altitudeGainMeter: w.score.altitude_gain_meter }
|
||||
: {}),
|
||||
...(w.score.altitude_change_meter != null
|
||||
? { altitudeChangeMeter: w.score.altitude_change_meter }
|
||||
: {}),
|
||||
...(w.score.zone_durations
|
||||
? { zoneDurations: w.score.zone_durations }
|
||||
: {}),
|
||||
};
|
||||
await markResourceSynced(userId, "workout");
|
||||
return imported;
|
||||
}
|
||||
|
||||
try {
|
||||
await prisma.workout.upsert({
|
||||
where: {
|
||||
userId_source_externalId: {
|
||||
userId,
|
||||
source: "WHOOP",
|
||||
externalId: w.id,
|
||||
},
|
||||
},
|
||||
create: {
|
||||
/**
|
||||
* Webhook-driven targeted refresh: resolve ONE workout by its uuid and upsert
|
||||
* it, instead of re-walking the whole collection. This is the direct fix for
|
||||
* iOS #17 — a `workout.updated` webhook lands the exact workout immediately,
|
||||
* with no dependence on the overlap window catching it. An unscored /
|
||||
* since-deleted record yields nothing. Does NOT advance the resource cursor.
|
||||
*/
|
||||
export async function syncWhoopWorkoutById(
|
||||
userId: string,
|
||||
workoutId: string,
|
||||
): Promise<number> {
|
||||
const tokenInfo = await getValidToken(userId);
|
||||
if (!tokenInfo) return 0;
|
||||
|
||||
let record: WhoopWorkout;
|
||||
try {
|
||||
record = await fetchWorkoutById(tokenInfo.accessToken, workoutId);
|
||||
} catch (err) {
|
||||
return handleCollectionFetchError("workout", userId, err);
|
||||
}
|
||||
|
||||
return upsertWhoopWorkout(userId, record);
|
||||
}
|
||||
|
||||
/**
|
||||
* Upsert one WHOOP workout into the `Workout` table keyed on
|
||||
* `(userId, source = WHOOP, externalId)`. Returns 1 when a row was written, 0
|
||||
* for an unscored record (nothing to store yet) or an upsert failure (logged,
|
||||
* never thrown — a single bad record must not fail the surrounding sync).
|
||||
* Shared by the collection walk and the webhook-driven fetch-by-id refresh so
|
||||
* both write an identical row shape.
|
||||
*/
|
||||
export async function upsertWhoopWorkout(
|
||||
userId: string,
|
||||
w: WhoopWorkout,
|
||||
): Promise<number> {
|
||||
if (!w.score) return 0; // unscored workout — nothing to store yet
|
||||
const startedAt = new Date(w.start);
|
||||
const endedAt = new Date(w.end);
|
||||
const durationSec = Math.max(
|
||||
0,
|
||||
Math.round((endedAt.getTime() - startedAt.getTime()) / 1000),
|
||||
);
|
||||
const energyKcal =
|
||||
typeof w.score.kilojoule === "number"
|
||||
? Math.round(w.score.kilojoule * KJ_TO_KCAL)
|
||||
: null;
|
||||
|
||||
const metadata: Prisma.InputJsonValue = {
|
||||
whoopWorkoutStrain: w.score.strain,
|
||||
percentRecorded: w.score.percent_recorded,
|
||||
...(w.score.altitude_gain_meter != null
|
||||
? { altitudeGainMeter: w.score.altitude_gain_meter }
|
||||
: {}),
|
||||
...(w.score.altitude_change_meter != null
|
||||
? { altitudeChangeMeter: w.score.altitude_change_meter }
|
||||
: {}),
|
||||
...(w.score.zone_durations
|
||||
? { zoneDurations: w.score.zone_durations }
|
||||
: {}),
|
||||
};
|
||||
|
||||
const row = {
|
||||
sportType: sportLabel(w.sport_id, w.sport_name),
|
||||
startedAt,
|
||||
endedAt,
|
||||
durationSec,
|
||||
totalEnergyKcal: energyKcal,
|
||||
totalDistanceM: w.score.distance_meter ?? null,
|
||||
avgHeartRate: w.score.average_heart_rate ?? null,
|
||||
maxHeartRate: w.score.max_heart_rate ?? null,
|
||||
elevationM: w.score.altitude_gain_meter ?? null,
|
||||
metadata,
|
||||
};
|
||||
|
||||
try {
|
||||
await prisma.workout.upsert({
|
||||
where: {
|
||||
userId_source_externalId: {
|
||||
userId,
|
||||
source: "WHOOP",
|
||||
externalId: w.id,
|
||||
sportType: sportLabel(w.sport_id, w.sport_name),
|
||||
startedAt,
|
||||
endedAt,
|
||||
durationSec,
|
||||
totalEnergyKcal: energyKcal,
|
||||
totalDistanceM: w.score.distance_meter ?? null,
|
||||
avgHeartRate: w.score.average_heart_rate ?? null,
|
||||
maxHeartRate: w.score.max_heart_rate ?? null,
|
||||
elevationM: w.score.altitude_gain_meter ?? null,
|
||||
metadata,
|
||||
},
|
||||
update: {
|
||||
sportType: sportLabel(w.sport_id, w.sport_name),
|
||||
startedAt,
|
||||
endedAt,
|
||||
durationSec,
|
||||
totalEnergyKcal: energyKcal,
|
||||
totalDistanceM: w.score.distance_meter ?? null,
|
||||
avgHeartRate: w.score.average_heart_rate ?? null,
|
||||
maxHeartRate: w.score.max_heart_rate ?? null,
|
||||
elevationM: w.score.altitude_gain_meter ?? null,
|
||||
metadata,
|
||||
},
|
||||
});
|
||||
imported++;
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(`WHOOP: failed to upsert workout: ${err}`);
|
||||
}
|
||||
},
|
||||
create: {
|
||||
userId,
|
||||
source: "WHOOP",
|
||||
externalId: w.id,
|
||||
...row,
|
||||
},
|
||||
update: row,
|
||||
});
|
||||
return 1;
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(`WHOOP: failed to upsert workout: ${err}`);
|
||||
return 0;
|
||||
}
|
||||
|
||||
await markSynced(userId);
|
||||
return imported;
|
||||
}
|
||||
+122
-7
@@ -117,6 +117,61 @@ const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000;
|
||||
export const WHOOP_RECOVERY_SLEEP_OVERLAP_MS = 7 * 24 * 60 * 60 * 1000;
|
||||
export const WHOOP_DEFAULT_OVERLAP_MS = 60 * 60 * 1000; // 1 h
|
||||
|
||||
/**
|
||||
* The far-past anchor a `fullSync` walks from. Incremental ticks start at
|
||||
* `cursor − overlap` (days at most); a fullSync deliberately ignores the
|
||||
* cursor and re-walks the deep history WHOOP retains (multi-year), which an
|
||||
* incremental run would never reach. WHOOP's first public data predates this,
|
||||
* so a fixed anchor covers every realistic account; the client's `next_token`
|
||||
* pagination + the idempotent upsert keep the re-walk bounded and safe.
|
||||
*/
|
||||
export const WHOOP_FULL_SYNC_ANCHOR = new Date("2020-01-01T00:00:00.000Z");
|
||||
|
||||
/**
|
||||
* The four pollable WHOOP collections that carry an independent sync cursor.
|
||||
* Body-measurement is a single profile object (no time-range cursor) and is
|
||||
* excluded. Each resource advances ONLY its own cursor key so a slow/failing
|
||||
* collection never drags the incremental window forward for its siblings.
|
||||
*/
|
||||
export type WhoopResource = "recovery" | "sleep" | "workout" | "cycle";
|
||||
|
||||
/**
|
||||
* Shape of the `WhoopConnection.resourceCursors` JSON map: a partial
|
||||
* `resource → ISO-8601 last-synced instant`. A missing key (or a legacy null
|
||||
* column) means the resource has no per-resource cursor yet and falls back to
|
||||
* the shared `lastSyncedAt`.
|
||||
*/
|
||||
type ResourceCursorMap = Partial<Record<WhoopResource, string>>;
|
||||
|
||||
function parseResourceCursors(raw: unknown): ResourceCursorMap {
|
||||
if (!raw || typeof raw !== "object" || Array.isArray(raw)) return {};
|
||||
return raw as ResourceCursorMap;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the incremental cursor for one resource. Prefers the per-resource
|
||||
* cursor; falls back to the connection's shared `lastSyncedAt` for a legacy
|
||||
* connection (or a resource that has never synced under the new column) so no
|
||||
* historical state is lost on the first tick after the migration. Returns null
|
||||
* when neither exists (the very first sync — `incrementalStart` then anchors 30
|
||||
* days back).
|
||||
*/
|
||||
export function resolveResourceCursor(
|
||||
connection: {
|
||||
resourceCursors?: unknown;
|
||||
lastSyncedAt?: Date | null;
|
||||
},
|
||||
resource: WhoopResource,
|
||||
): Date | null {
|
||||
const map = parseResourceCursors(connection.resourceCursors);
|
||||
const iso = map[resource];
|
||||
if (iso) {
|
||||
const parsed = new Date(iso);
|
||||
if (!Number.isNaN(parsed.getTime())) return parsed;
|
||||
}
|
||||
return connection.lastSyncedAt ?? null;
|
||||
}
|
||||
|
||||
export interface WhoopTokenInfo {
|
||||
accessToken: string;
|
||||
connection: { id: string; whoopUserId: string };
|
||||
@@ -200,18 +255,24 @@ export async function getValidToken(
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute the incremental `start` for a resource sync. `fullSync` returns
|
||||
* undefined (the backfill anchor handles the deep history). Otherwise start
|
||||
* from `lastSyncedAt - overlap`, or 30 days back on the very first incremental
|
||||
* tick.
|
||||
* Compute the `start` bound for a resource fetch.
|
||||
*
|
||||
* A `fullSync` deliberately ignores the cursor and anchors at
|
||||
* `WHOOP_FULL_SYNC_ANCHOR` — the deep historical backfill an incremental run
|
||||
* would never reach. (It previously returned `undefined`; an explicit anchor
|
||||
* pins the lower bound so the walk is reproducible and the cursor a stalled
|
||||
* incremental left behind cannot silently narrow a backfill.)
|
||||
*
|
||||
* Otherwise it is an incremental tick: start from `cursor − overlap`, or 30
|
||||
* days back on the very first sync (no cursor yet).
|
||||
*/
|
||||
export function incrementalStart(
|
||||
lastSyncedAt: Date | null,
|
||||
cursor: Date | null,
|
||||
opts: { fullSync?: boolean; overlapMs?: number } = {},
|
||||
): Date | undefined {
|
||||
if (opts.fullSync) return undefined;
|
||||
if (opts.fullSync) return WHOOP_FULL_SYNC_ANCHOR;
|
||||
const overlap = opts.overlapMs ?? WHOOP_DEFAULT_OVERLAP_MS;
|
||||
if (lastSyncedAt) return new Date(lastSyncedAt.getTime() - overlap);
|
||||
if (cursor) return new Date(cursor.getTime() - overlap);
|
||||
return new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
|
||||
}
|
||||
|
||||
@@ -316,6 +377,60 @@ export async function markSynced(userId: string): Promise<void> {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* The closed set of resources allowed in the `resourceCursors` JSON path. The
|
||||
* key is passed to `jsonb_set` below — assert it against this whitelist so a
|
||||
* caller can never splice an arbitrary path, even though `resource` is already
|
||||
* the typed `WhoopResource` union.
|
||||
*/
|
||||
const WHOOP_CURSOR_RESOURCES: ReadonlySet<WhoopResource> = new Set([
|
||||
"recovery",
|
||||
"sleep",
|
||||
"workout",
|
||||
"cycle",
|
||||
]);
|
||||
|
||||
/**
|
||||
* Advance ONE resource's cursor after its successful fetch+upsert. Sibling
|
||||
* cursors are preserved untouched, so a resource that errored or soft-skipped
|
||||
* this tick keeps its own (older) cursor and re-fetches from there next time.
|
||||
* Also keeps the shared `lastSyncedAt` moving (monotonically) for any legacy
|
||||
* reader still on it.
|
||||
*
|
||||
* The merge is a SINGLE atomic `jsonb_set` upsert, not a read-modify-write. The
|
||||
* four WHOOP resources are independent pg-boss queues that run concurrently for
|
||||
* the same user (a cron tick fans out all four at once), so a read-then-write
|
||||
* would let two resources read the same map and the second write clobber the
|
||||
* first's cursor. `jsonb_set` merges in-place under the row lock, and
|
||||
* `GREATEST` keeps `last_synced_at` from being dragged backward by a slower
|
||||
* sibling finishing with an older `at`.
|
||||
*/
|
||||
export async function markResourceSynced(
|
||||
userId: string,
|
||||
resource: WhoopResource,
|
||||
at: Date = new Date(),
|
||||
): Promise<void> {
|
||||
// Defence-in-depth: the typed union already bounds `resource`, but the value
|
||||
// reaches a JSON path, so refuse anything outside the closed whitelist.
|
||||
if (!WHOOP_CURSOR_RESOURCES.has(resource)) return;
|
||||
|
||||
const iso = at.toISOString();
|
||||
// `resource` is parameter-bound ($2), never string-spliced. `jsonb_set`
|
||||
// builds the path array from it; `coalesce` seeds an empty object for a
|
||||
// legacy null column; `GREATEST` advances `last_synced_at` only forward.
|
||||
await prisma.$executeRaw`
|
||||
UPDATE "whoop_connections"
|
||||
SET "resource_cursors" = jsonb_set(
|
||||
coalesce("resource_cursors", '{}'::jsonb),
|
||||
ARRAY[${resource}],
|
||||
to_jsonb(${iso}::text),
|
||||
true
|
||||
),
|
||||
"last_synced_at" = GREATEST("last_synced_at", ${at})
|
||||
WHERE "user_id" = ${userId}
|
||||
`;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
|
||||
@@ -203,13 +203,18 @@ export async function processWhoopNotification(
|
||||
}
|
||||
|
||||
// `*.updated` (creates arrive as updates too). Enqueue the per-user
|
||||
// resource sync; the worker re-fetches by id (payload carries no data).
|
||||
// resource sync. The payload carries the resource id so the worker can do a
|
||||
// targeted fetch-by-id refresh — landing the exact record immediately rather
|
||||
// than waiting for the next overlap window to re-walk the collection.
|
||||
const boss = getGlobalBoss();
|
||||
if (!boss) {
|
||||
getEvent()?.addWarning("whoop-webhook: pg-boss not initialised");
|
||||
return NextResponse.json({ status: "ok" }, { status: 200 });
|
||||
}
|
||||
await boss.send(queue, { userId: connection.userId });
|
||||
await boss.send(queue, {
|
||||
userId: connection.userId,
|
||||
resourceId: resourceId ?? undefined,
|
||||
});
|
||||
|
||||
annotate({
|
||||
action: { name: "whoop.webhook.enqueued" },
|
||||
|
||||
Reference in new issue
Block a user