From 483c84b42eb63f345b3ca49cb8a1becf51993baa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 05:06:21 +0200 Subject: [PATCH] fix(integrations): report a page walk that stops short instead of ending quietly The WHOOP, Oura and Withings list walks and the Fitbit exercise walk each kept a page loop that, at its ceiling or on a cursor that did not advance, returned what it had read and said nothing, and the backfill around it stamped the history complete. `walkPages` keeps a ceiling but reports a walk that stops with a page pending, or on a cursor it already followed, as truncated. A backfill reads that through one verdict: a failure retries, a truncated collection leaves the marker unset without a retry, a clean run stamps it. The Fitbit exercise walk may read a hundred pages during a backfill instead of twelve. A guard keeps page-counting loops out of the provider directories, with Google Health and Strava listed until their next change. --- src/__tests__/page-walk-guard.test.ts | 84 +++++++++++ src/lib/fitbit/sync-workout.ts | 66 +++++--- .../integrations/__tests__/page-walk.test.ts | 66 ++++++++ src/lib/integrations/page-walk.ts | 112 ++++++++++++++ .../jobs/__tests__/fitbit-backfill.test.ts | 21 +++ src/lib/jobs/fitbit-backfill.ts | 46 ++++-- src/lib/jobs/whoop-backfill.ts | 48 ++++-- src/lib/oura/client.ts | 84 +++++------ src/lib/whoop/client.ts | 82 +++++----- src/lib/withings/client.ts | 141 ++++++++++-------- src/lib/withings/sync-activity.ts | 101 +++++++------ src/lib/withings/sync-ecg.ts | 99 ++++++------ 12 files changed, 655 insertions(+), 295 deletions(-) create mode 100644 src/__tests__/page-walk-guard.test.ts create mode 100644 src/lib/integrations/__tests__/page-walk.test.ts create mode 100644 src/lib/integrations/page-walk.ts diff --git a/src/__tests__/page-walk-guard.test.ts b/src/__tests__/page-walk-guard.test.ts new file mode 100644 index 000000000..3d7f7900f --- /dev/null +++ b/src/__tests__/page-walk-guard.test.ts @@ -0,0 +1,84 @@ +/** + * Structural guard: a provider client walks a paged collection through + * `walkPages` (`src/lib/integrations/page-walk.ts`), never a loop of its own. + * + * Each hand-rolled loop ended at its page ceiling by returning what it had, + * silently, and the backfill around it stamped the history complete. A + * `while (pages < MAX)`, a `pageCount > N` break or a `do … while (next && + * pageCount < max)` in a provider directory is that class coming back. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; + +import { walkSourceFiles } from "./helpers/source-files"; + +const LIB = join(process.cwd(), "src", "lib"); +const PROVIDER_DIRS = [ + "dawarich", + "fitbit", + "google-health", + "nightscout", + "oura", + "polar", + "strava", + "whoop", + "withings", +]; +/** + * Loops still counting pages on their own, with the reason. Each is bounded + * and reports nothing at its bound; each moves to `walkPages` with the next + * change to its file. + */ +const PENDING: Readonly> = { + "google-health/client.ts": + "progress-bounded walk (backfill-bounds.ts); moves with the next Google Health change", + "strava/sync.ts": + "activity pages per sync; moves with the next Strava change", +}; + +/** A page counter compared against a bound, in any of the shapes used. */ +const PAGE_LOOP = + /\b(?:pages?|pageCount|pagesRead|page)\s*(?:<|<=|>|>=)\s*(?:[A-Z_][A-Z0-9_]*|maxPages|\d+)\b/; + +function strip(text: string): string { + return text + .replace(/\/\*[\s\S]*?\*\//g, "") + .replace(/(^|[^:])\/\/.*$/gm, "$1"); +} + +describe("provider clients walk pages through walkPages", () => { + const files = walkSourceFiles(LIB, { floor: 1000 }).filter( + (f) => + PROVIDER_DIRS.some((d) => f.startsWith(`${d}/`)) && + !f.includes("__tests__") && + !f.endsWith(".test.ts"), + ); + const offending = (f: string) => + PAGE_LOOP.test(strip(readFileSync(join(LIB, f), "utf8"))); + + it("walks the provider directories", () => { + expect(files.length).toBeGreaterThanOrEqual(60); + }); + + it("finds no page loop of their own outside the pending list", () => { + expect(files.filter((f) => !(f in PENDING) && offending(f))).toEqual([]); + }); + + it("keeps the pending list honest", () => { + expect(Object.keys(PENDING).filter((f) => !offending(f))).toEqual([]); + }); + + it("catches the shapes the loops used", () => { + for (const planted of [ + "while (pages < WORKOUT_MAX_PAGES) {", + "if (pageCount > 1000) break;", + "} while (nextToken && pageCount < maxPages);", + ]) { + expect(PAGE_LOOP.test(planted), planted).toBe(true); + } + expect(PAGE_LOOP.test("const walk = await walkPages(label, 12, f);")).toBe( + false, + ); + }); +}); diff --git a/src/lib/fitbit/sync-workout.ts b/src/lib/fitbit/sync-workout.ts index 5a7312cad..249a1755d 100644 --- a/src/lib/fitbit/sync-workout.ts +++ b/src/lib/fitbit/sync-workout.ts @@ -13,8 +13,10 @@ * collapse for a server-owned source pair. * * The page-walk is bounded (the 150 req/h budget is tight): a small page cap - * covers the incremental window comfortably; a deep backfill is bounded by the - * backfill horizon `afterDate`. A per-endpoint 403 soft-skips the resource. + * covers the incremental window comfortably, and a backfill, bounded by the + * horizon `afterDate`, may read up to ten thousand sessions. A walk that + * reaches its cap is reported truncated (`walkPages`), so the backfill is not + * stamped complete over it. A per-endpoint 403 soft-skips the resource. */ import { fetchActivityList, mapWorkout, readActivityList } from "./client"; import type { FitbitMappedWorkout } from "./client"; @@ -24,11 +26,14 @@ import { prisma, toJson } from "@/lib/db"; import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit"; import { annotate, getEvent } from "@/lib/logging/context"; import { resolveUserTimezone } from "@/lib/tz/resolver"; +import { walkPages } from "@/lib/integrations/page-walk"; /** Activities per page on the `activities/list` walk. */ const WORKOUT_PAGE_SIZE = 100; -/** Hard ceiling on pages walked per cycle (rate-budget guard). */ +/** Ceiling on pages walked per incremental cycle (rate-budget guard). */ const WORKOUT_MAX_PAGES = 12; +/** Ceiling on pages walked by a backfill. */ +const WORKOUT_BACKFILL_MAX_PAGES = 100; function metadataRecord(value: unknown): Record { return typeof value === "object" && value !== null && !Array.isArray(value) @@ -36,6 +41,32 @@ function metadataRecord(value: unknown): Record { : {}; } +function walkActivityList( + accessToken: string, + afterDate: Date, + maxPages: number, +) { + return walkPages[number]>( + "fitbit.exercise", + maxPages, + async (cursor) => { + const offset = cursor ?? 0; + const page = readActivityList( + await fetchActivityList(accessToken, afterDate, { + limit: WORKOUT_PAGE_SIZE, + offset, + }), + ); + return { + items: page, + next: + page.length < WORKOUT_PAGE_SIZE ? null : offset + WORKOUT_PAGE_SIZE, + }; + }, + 0, + ); +} + export async function syncUserWorkout( userId: string, opts: FitbitResourceSyncOptions = {}, @@ -50,28 +81,19 @@ export async function syncUserWorkout( const afterDate = opts.start ?? new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); - const entries: ReturnType = []; - let offset = 0; - let pages = 0; - while (pages < WORKOUT_MAX_PAGES) { - let body: unknown; - try { - body = await fetchActivityList(tokenInfo.accessToken, afterDate, { - limit: WORKOUT_PAGE_SIZE, - offset, - }); - } catch (err) { - return handleCollectionFetchError("fetchExercise", userId, err); - } - const page = readActivityList(body); - entries.push(...page); - pages += 1; - if (page.length < WORKOUT_PAGE_SIZE) break; - offset += WORKOUT_PAGE_SIZE; + let walk: Awaited>; + try { + walk = await walkActivityList( + tokenInfo.accessToken, + afterDate, + opts.fullSync ? WORKOUT_BACKFILL_MAX_PAGES : WORKOUT_MAX_PAGES, + ); + } catch (err) { + return handleCollectionFetchError("fetchExercise", userId, err); } let imported = 0; - for (const entry of entries) { + for (const entry of walk.items) { const w: FitbitMappedWorkout | null = mapWorkout(entry, tz); if (!w) continue; // no usable time span — not a workout diff --git a/src/lib/integrations/__tests__/page-walk.test.ts b/src/lib/integrations/__tests__/page-walk.test.ts new file mode 100644 index 000000000..69980ff49 --- /dev/null +++ b/src/lib/integrations/__tests__/page-walk.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); + +import { backfillVerdict, walkPages, withPageWalkLedger } from "../page-walk"; + +describe("walkPages", () => { + it("reads until a page has no next", async () => { + const walk = await walkPages( + "test.done", + 10, + async (cursor) => ({ + items: [`p${cursor ?? 0}`], + next: (cursor ?? 0) < 2 ? (cursor ?? 0) + 1 : null, + }), + ); + expect(walk).toEqual({ + items: ["p0", "p1", "p2"], + pages: 3, + truncated: false, + }); + }); + + it("reports a walk that reaches its ceiling with a page pending", async () => { + const walk = await walkPages( + "test.ceiling", + 4, + async (cursor) => ({ items: [cursor ?? 0], next: (cursor ?? 0) + 1 }), + ); + expect(walk.truncated).toBe(true); + expect(walk.pages).toBe(4); + expect(walk.items).toEqual([0, 1, 2, 3]); + }); + + it("stops on a cursor it already followed, the first one included", async () => { + const fetchPage = vi.fn(async () => ({ items: [1], next: 0 })); + const walk = await walkPages( + "test.repeat", + 50, + fetchPage, + 0, + ); + expect(walk.truncated).toBe(true); + expect(fetchPage).toHaveBeenCalledTimes(1); + }); +}); + +describe("withPageWalkLedger and backfillVerdict", () => { + it("collects the truncated walks of a run", async () => { + const { result, truncated } = await withPageWalkLedger(async () => { + await walkPages("b.walk", 1, async () => ({ items: [], next: "x" })); + await walkPages("a.walk", 5, async () => ({ items: [], next: null })); + return "done"; + }); + expect(result).toBe("done"); + expect(truncated).toEqual(["b.walk"]); + }); + + it("retries a failure, leaves a truncated history incomplete, stamps a clean one", () => { + expect(backfillVerdict({ failed: true, truncated: ["a"] })).toBe("retry"); + expect(backfillVerdict({ failed: false, truncated: ["a"] })).toBe( + "incomplete", + ); + expect(backfillVerdict({ failed: false, truncated: [] })).toBe("complete"); + }); +}); diff --git a/src/lib/integrations/page-walk.ts b/src/lib/integrations/page-walk.ts new file mode 100644 index 000000000..df126c8a5 --- /dev/null +++ b/src/lib/integrations/page-walk.ts @@ -0,0 +1,112 @@ +/** + * The one way a provider client walks a paged collection, and the contract a + * history backfill reads off it. + * + * Every client used to keep its own `while (pages < MAX)` loop, and each one + * ended the same way at its ceiling: it returned what it had read, said + * nothing, and the backfill around it stamped the connection complete. A + * history longer than the ceiling, or a cursor that never advanced, was cut + * off for good without a trace. `walkPages` keeps the ceiling, because a + * runaway cursor must stop somewhere, but a walk that stops with a page still + * pending, or on a cursor it has already followed, reports itself + * `truncated`: on the run's wide event, and in the ledger a backfill opens + * with `withPageWalkLedger`. + * + * The backfill contract (`backfillVerdict`), the same for every provider: + * - a failed collection is transient: the job throws and pg-boss retries; + * - a truncated collection is not complete: the marker is not stamped, and + * boot discovery offers the account again, without a retry storm; + * - a connection parked for re-authorisation is permanent until the person + * reconnects: the runner returns without stamping and without a retry. + * + * Google Health keeps its own progress-based walk (`google-health/ + * backfill-bounds.ts`), whose `no_progress` end is this module's runaway case + * measured in time rather than in cursors; it moves onto this contract with + * its next change. `src/__tests__/page-walk-guard.test.ts` keeps hand-rolled + * page loops out of the provider directories. + */ +import { AsyncLocalStorage } from "node:async_hooks"; + +import { annotate } from "@/lib/logging/context"; + +/** One page as the client read it. A null `next` ends the walk. */ +export interface WalkPage { + items: readonly T[]; + next: C | null | undefined; +} + +/** What the walk came to. */ +export interface PageWalk { + items: T[]; + pages: number; + /** Stopped at the ceiling or on a repeated cursor with history left. */ + truncated: boolean; +} + +const ledger = new AsyncLocalStorage>(); + +/** + * Walk `fetchPage` from `first` (or no cursor) until a page has no `next`, + * at most `maxPages` pages. `label` names the collection (`whoop.recovery`) + * on the wide event and in the ledger. + */ +export async function walkPages( + label: string, + maxPages: number, + fetchPage: (cursor: C | undefined, page: number) => Promise>, + first?: C, +): Promise> { + const items: T[] = []; + const seen = new Set(first === undefined ? [] : [first]); + let cursor = first; + for (let page = 0; page < maxPages; page += 1) { + const result = await fetchPage(cursor, page); + items.push(...result.items); + const next = result.next; + if (next === null || next === undefined) { + return { items, pages: page + 1, truncated: false }; + } + if (seen.has(next)) return truncated(label, items, page + 1, "repeat"); + seen.add(next); + cursor = next; + } + return truncated(label, items, maxPages, "ceiling"); +} + +function truncated( + label: string, + items: T[], + pages: number, + reason: "ceiling" | "repeat", +): PageWalk { + annotate({ + action: { name: "integration.page_walk.truncated" }, + meta: { walk: label, pages, reason }, + }); + ledger.getStore()?.add(label); + return { items, pages, truncated: true }; +} + +/** + * Run `run` with a ledger of the walks inside it that ended truncated. A + * backfill wraps its full sync in this and reads the ledger before it + * stamps its marker. + */ +export async function withPageWalkLedger( + run: () => Promise, +): Promise<{ result: R; truncated: string[] }> { + const walks = new Set(); + const result = await ledger.run(walks, run); + return { result, truncated: [...walks].sort() }; +} + +/** What a backfill does with its full sync's outcome. */ +export type BackfillVerdict = "complete" | "retry" | "incomplete"; + +export function backfillVerdict(outcome: { + failed: boolean; + truncated: readonly string[]; +}): BackfillVerdict { + if (outcome.failed) return "retry"; + return outcome.truncated.length > 0 ? "incomplete" : "complete"; +} diff --git a/src/lib/jobs/__tests__/fitbit-backfill.test.ts b/src/lib/jobs/__tests__/fitbit-backfill.test.ts index 9c4b9306b..b34e9972a 100644 --- a/src/lib/jobs/__tests__/fitbit-backfill.test.ts +++ b/src/lib/jobs/__tests__/fitbit-backfill.test.ts @@ -125,3 +125,24 @@ describe("runFitbitBackfillForUser — verdict-gated marker", () => { expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled(); }); }); + +describe("runFitbitBackfillForUser — a walk that reached its ceiling", () => { + it("leaves the marker unset and does not throw", async () => { + const { walkPages } = await import("@/lib/integrations/page-walk"); + syncUserFitbit.mockImplementation(async () => { + // A history longer than the walk may read: every page has a next. + await walkPages( + "fitbit.exercise", + 3, + async (cursor) => ({ items: [1], next: (cursor ?? 0) + 1 }), + 0, + ); + return { imported: 3, failed: false }; + }); + + await expect(runFitbitBackfillForUser("u1")).resolves.toEqual({ + imported: 3, + }); + expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/jobs/fitbit-backfill.ts b/src/lib/jobs/fitbit-backfill.ts index fc57592fa..7fbb0e8f4 100644 --- a/src/lib/jobs/fitbit-backfill.ts +++ b/src/lib/jobs/fitbit-backfill.ts @@ -15,6 +15,10 @@ 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 { + backfillVerdict, + withPageWalkLedger, +} from "@/lib/integrations/page-walk"; import { syncUserFitbit } from "@/lib/fitbit/sync"; export const FITBIT_BACKFILL_QUEUE = "fitbit-backfill"; @@ -39,14 +43,17 @@ export interface FitbitBackfillPayload { * `runWhoopBackfillForUser` (whose identical unconditional-stamp bug is fixed * separately — this is the template for that port). * - * Verdict-gated: a partial hard failure (one collection 400ing, a write - * failing, a dead token) THROWS so pg-boss retries the job — stamping the - * marker over a partial walk would freeze the history gap in 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 — the - * boot discovery re-enqueues the account on the next boot, and the stamp lands - * once the user reconnects and a clean walk completes. + * Verdict-gated (`backfillVerdict`, `integrations/page-walk.ts`): a partial + * hard failure (one collection 400ing, a write failing, a dead token) THROWS + * so pg-boss retries the job — stamping the marker over a partial walk would + * freeze the history gap in forever and the configured `retryLimit` would be + * dead code. A collection whose walk reached its page ceiling leaves the + * marker unset WITHOUT a retry: nothing failed, and boot discovery offers the + * account again. 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 — the boot discovery re-enqueues the + * account on the next boot, and the stamp lands once the user reconnects and + * a clean walk completes. */ export async function runFitbitBackfillForUser( userId: string, @@ -61,9 +68,28 @@ export async function runFitbitBackfillForUser( return { imported: 0 }; } - const { imported, failed } = await syncUserFitbit(userId, { fullSync: true }); + const { + result: { imported, failed }, + truncated, + } = await withPageWalkLedger(() => + syncUserFitbit(userId, { fullSync: true }), + ); + const verdict = backfillVerdict({ failed, truncated }); - if (failed) { + if (verdict === "incomplete") { + // A collection longer than its walk may read in one run: the history is + // not complete, but nothing failed, so no retry. Boot discovery offers + // the account again while the marker stays unset. + annotate({ + action: { + name: "fitbit.backfill.truncated", + details: { imported, collections: truncated.join(",") }, + }, + }); + return { imported }; + } + + if (verdict === "retry") { // Surface through the pg-boss retry path (retryLimit 3, backoff, set by // `integrationBackfillSourceOptions`). If the failure parked the connection // at error_reauth meanwhile, the next attempt takes the reauth return above diff --git a/src/lib/jobs/whoop-backfill.ts b/src/lib/jobs/whoop-backfill.ts index 488ae6b1c..83d87ad82 100644 --- a/src/lib/jobs/whoop-backfill.ts +++ b/src/lib/jobs/whoop-backfill.ts @@ -21,6 +21,10 @@ 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 { + backfillVerdict, + withPageWalkLedger, +} from "@/lib/integrations/page-walk"; import { syncUserWhoop } from "@/lib/whoop/sync"; export const WHOOP_BACKFILL_QUEUE = "whoop-backfill"; @@ -44,16 +48,19 @@ export interface WhoopBackfillPayload { * 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. + * Verdict-gated (`backfillVerdict`, `integrations/page-walk.ts`): 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 collection whose walk reached its page ceiling leaves + * the marker unset WITHOUT a retry: nothing failed, and boot discovery offers + * the account again. 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, @@ -68,9 +75,26 @@ export async function runWhoopBackfillForUser( return { imported: 0 }; } - const { imported, failed } = await syncUserWhoop(userId, { fullSync: true }); + const { + result: { imported, failed }, + truncated, + } = await withPageWalkLedger(() => syncUserWhoop(userId, { fullSync: true })); + const verdict = backfillVerdict({ failed, truncated }); - if (failed) { + if (verdict === "incomplete") { + // A collection longer than its walk may read in one run: the history is + // not complete, but nothing failed, so no retry. Boot discovery offers + // the account again while the marker stays unset. + annotate({ + action: { + name: "whoop.backfill.truncated", + details: { imported, collections: truncated.join(",") }, + }, + }); + return { imported }; + } + + if (verdict === "retry") { // 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. diff --git a/src/lib/oura/client.ts b/src/lib/oura/client.ts index 4b9c92078..9718ce696 100644 --- a/src/lib/oura/client.ts +++ b/src/lib/oura/client.ts @@ -22,6 +22,7 @@ import { safeFetch } from "@/lib/safe-fetch"; import { OuraApiError, classifyOuraResponse } from "./response-classifier"; import { envOr, envValue } from "@/lib/env"; import { importValue } from "@/lib/integrations/import-value"; +import { walkPages } from "@/lib/integrations/page-walk"; const OURA_API_BASE = "https://api.ouraring.com"; const OURA_OAUTH_AUTH_URL = "https://cloud.ouraring.com/oauth/authorize"; @@ -476,51 +477,48 @@ async function fetchCollection( verb: string, query: DateRangeQuery, ): Promise { - const records: T[] = []; - let nextToken: string | null | undefined; - let pageCount = 0; - const maxPages = query.maxPages ?? 100; - - do { - const params = new URLSearchParams({ - start_date: query.startDate, - end_date: query.endDate, - }); - if (nextToken) params.set("next_token", nextToken); - - const pageStart = performance.now(); - const res = await safeFetch(`${OURA_API_BASE}${path}?${params}`, { - method: "GET", - headers: { - Authorization: `Bearer ${accessToken}`, - Accept: "application/json", - }, - }); - const json = (await res - .json() - .catch(() => null)) as OuraCollection | null; - const verdict = classifyOuraResponse(res.status); - getEvent()?.addExternalCall({ - service: "oura", - method: `${verb}(page=${pageCount})`, - duration_ms: Math.round(performance.now() - pageStart), - status: res.status, - error: verdict.classification === "success" ? undefined : verdict.reason, - }); - if (verdict.classification !== "success") { - throw new OuraApiError({ - verb, - classification: verdict.classification, - httpStatus: verdict.httpStatus, - reason: verdict.reason, + const walk = await walkPages( + `oura.${verb}`, + query.maxPages ?? 100, + async (nextToken, page) => { + const params = new URLSearchParams({ + start_date: query.startDate, + end_date: query.endDate, }); - } - for (const r of json?.data ?? []) records.push(r); - nextToken = json?.next_token ?? null; - pageCount += 1; - } while (nextToken && pageCount < maxPages); + if (nextToken) params.set("next_token", nextToken); - return records; + const pageStart = performance.now(); + const res = await safeFetch(`${OURA_API_BASE}${path}?${params}`, { + method: "GET", + headers: { + Authorization: `Bearer ${accessToken}`, + Accept: "application/json", + }, + }); + const json = (await res + .json() + .catch(() => null)) as OuraCollection | null; + const verdict = classifyOuraResponse(res.status); + getEvent()?.addExternalCall({ + service: "oura", + method: `${verb}(page=${page})`, + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: + verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new OuraApiError({ + verb, + classification: verdict.classification, + httpStatus: verdict.httpStatus, + reason: verdict.reason, + }); + } + return { items: json?.data ?? [], next: json?.next_token ?? null }; + }, + ); + return walk.items; } export function fetchSleep( diff --git a/src/lib/whoop/client.ts b/src/lib/whoop/client.ts index 533a2866c..f6b76aa69 100644 --- a/src/lib/whoop/client.ts +++ b/src/lib/whoop/client.ts @@ -21,6 +21,7 @@ import { reconstructContiguousSleepTimeline } from "@/lib/sleep/reconstruct-time import { WhoopApiError, classifyWhoopResponse } from "./response-classifier"; import { envOr, envValue } from "@/lib/env"; import { importValue } from "@/lib/integrations/import-value"; +import { walkPages } from "@/lib/integrations/page-walk"; const WHOOP_API_BASE = "https://api.prod.whoop.com/developer"; const WHOOP_OAUTH_AUTH_URL = "https://api.prod.whoop.com/oauth/oauth2/auth"; @@ -269,7 +270,10 @@ export interface WhoopProfile { interface CollectionQuery { start?: Date; end?: Date; - /** Hard ceiling on pages walked (defence against a runaway cursor). */ + /** + * Ceiling on pages walked (defence against a runaway cursor). A walk that + * reaches it is reported truncated (`walkPages`). + */ maxPages?: number; } @@ -279,49 +283,45 @@ async function fetchCollection( verb: string, query: CollectionQuery = {}, ): Promise { - const records: T[] = []; - let nextToken: string | null | undefined; - let pageCount = 0; - const maxPages = query.maxPages ?? 1000; + const walk = await walkPages( + `whoop.${verb}`, + query.maxPages ?? 1000, + async (nextToken, page) => { + const params = new URLSearchParams({ limit: String(WHOOP_PAGE_LIMIT) }); + if (query.start) params.set("start", query.start.toISOString()); + if (query.end) params.set("end", query.end.toISOString()); + if (nextToken) params.set("nextToken", nextToken); - do { - const params = new URLSearchParams({ limit: String(WHOOP_PAGE_LIMIT) }); - if (query.start) params.set("start", query.start.toISOString()); - if (query.end) params.set("end", query.end.toISOString()); - if (nextToken) params.set("nextToken", nextToken); - - const pageStart = performance.now(); - const res = await safeFetch(`${WHOOP_API_BASE}${path}?${params}`, { - method: "GET", - headers: { Authorization: `Bearer ${accessToken}` }, - }); - - const json = (await res - .json() - .catch(() => null)) as WhoopCollection | null; - const verdict = classifyWhoopResponse(res.status); - getEvent()?.addExternalCall({ - service: "whoop", - method: `${verb}(page=${pageCount})`, - duration_ms: Math.round(performance.now() - pageStart), - status: res.status, - error: verdict.classification === "success" ? undefined : verdict.reason, - }); - if (verdict.classification !== "success") { - throw new WhoopApiError({ - verb, - classification: verdict.classification, - httpStatus: verdict.httpStatus, - reason: verdict.reason, + const pageStart = performance.now(); + const res = await safeFetch(`${WHOOP_API_BASE}${path}?${params}`, { + method: "GET", + headers: { Authorization: `Bearer ${accessToken}` }, }); - } - for (const r of json?.records ?? []) records.push(r); - nextToken = json?.next_token ?? null; - pageCount += 1; - } while (nextToken && pageCount < maxPages); - - return records; + const json = (await res + .json() + .catch(() => null)) as WhoopCollection | null; + const verdict = classifyWhoopResponse(res.status); + getEvent()?.addExternalCall({ + service: "whoop", + method: `${verb}(page=${page})`, + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: + verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new WhoopApiError({ + verb, + classification: verdict.classification, + httpStatus: verdict.httpStatus, + reason: verdict.reason, + }); + } + return { items: json?.records ?? [], next: json?.next_token ?? null }; + }, + ); + return walk.items; } async function fetchSingle( diff --git a/src/lib/withings/client.ts b/src/lib/withings/client.ts index c02117675..6cf3b760e 100644 --- a/src/lib/withings/client.ts +++ b/src/lib/withings/client.ts @@ -10,6 +10,7 @@ import { } from "./response-classifier"; import { envOr, envValue } from "@/lib/env"; import { importValue } from "@/lib/integrations/import-value"; +import { walkPages } from "@/lib/integrations/page-walk"; const WITHINGS_OAUTH_URL = "https://account.withings.com/oauth2_user/authorize2"; @@ -249,6 +250,24 @@ export interface WithingsMeasureGroup { export const WITHINGS_MEASTYPES: readonly number[] = Object.keys(MEASURE_TYPE_MAP).map(Number); +/** Ceiling on the pages one Withings list walk reads (`walkPages`). */ +export const WITHINGS_MAX_PAGES = 1000; + +/** + * The offset of the next page of a Withings list body, or null when it has + * no more. A body that says `more` without an offset past the current one + * hands back the current offset, which `walkPages` reports as a repeat + * instead of stopping quietly. + */ +export function withingsNextOffset( + body: { more?: unknown; offset?: unknown }, + offset: number, +): number | null { + if (body.more !== true && body.more !== 1) return null; + const next = Number(body.offset); + return Number.isFinite(next) && next > offset ? next : offset; +} + /** * Fetch measurements from Withings API. * @@ -273,77 +292,69 @@ export async function fetchMeasurements( baseParams.enddate = String(Math.floor(endDate.getTime() / 1000)); } - const results: WithingsMeasure[] = []; - let offset = 0; - let pageCount = 0; - - while (true) { - const params = new URLSearchParams({ - ...baseParams, - offset: String(offset), - }); - - const pageStart = performance.now(); - const res = await safeFetch(WITHINGS_MEASURE_URL, { - method: "POST", - headers: { - "Content-Type": "application/x-www-form-urlencoded", - Authorization: `Bearer ${accessToken}`, - }, - body: params.toString(), - }); - - const json = await res.json(); - const verdict = classifyWithingsResponse(res.status, json); - getEvent()?.addExternalCall({ - service: "withings", - method: `fetchMeasurements(page=${pageCount})`, - duration_ms: Math.round(performance.now() - pageStart), - status: res.status, - error: verdict.classification === "success" ? undefined : verdict.reason, - }); - if (verdict.classification !== "success") { - throw new WithingsApiError({ - verb: "measure", - classification: verdict.classification, - withingsStatus: verdict.withingsStatus, - reason: verdict.reason, - upstreamError: typeof json?.error === "string" ? json.error : undefined, + const walk = await walkPages( + "withings.measure", + WITHINGS_MAX_PAGES, + async (cursor, page) => { + const offset = cursor ?? 0; + const params = new URLSearchParams({ + ...baseParams, + offset: String(offset), }); - } - const body = json.body ?? {}; - const groups: WithingsMeasureGroup[] = body.measuregrps ?? []; + const pageStart = performance.now(); + const res = await safeFetch(WITHINGS_MEASURE_URL, { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + Authorization: `Bearer ${accessToken}`, + }, + body: params.toString(), + }); - for (const group of groups) { - const measuredAt = new Date(group.date * 1000); - - for (const m of group.measures) { - const mapping = MEASURE_TYPE_MAP[m.type]; - if (!mapping) continue; - - // Withings stores value * 10^unit (unit is usually negative) - const value = m.value * Math.pow(10, m.unit); - - results.push({ - type: mapping.type, - value: importValue(value), - measuredAt, + const json = await res.json(); + const verdict = classifyWithingsResponse(res.status, json); + getEvent()?.addExternalCall({ + service: "withings", + method: `fetchMeasurements(page=${page})`, + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: + verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new WithingsApiError({ + verb: "measure", + classification: verdict.classification, + withingsStatus: verdict.withingsStatus, + reason: verdict.reason, + upstreamError: + typeof json?.error === "string" ? json.error : undefined, }); } - } - const hasMore = body.more === true || body.more === 1; - if (!hasMore) break; - - const nextOffset = Number(body.offset); - if (!Number.isFinite(nextOffset) || nextOffset <= offset) break; - offset = nextOffset; - pageCount += 1; - if (pageCount > 1000) break; - } - - return results; + const body = json.body ?? {}; + const groups: WithingsMeasureGroup[] = body.measuregrps ?? []; + const items: WithingsMeasure[] = []; + for (const group of groups) { + const measuredAt = new Date(group.date * 1000); + for (const m of group.measures) { + const mapping = MEASURE_TYPE_MAP[m.type]; + if (!mapping) continue; + // Withings stores value * 10^unit (unit is usually negative) + const value = m.value * Math.pow(10, m.unit); + items.push({ + type: mapping.type, + value: importValue(value), + measuredAt, + }); + } + } + return { items, next: withingsNextOffset(body, offset) }; + }, + 0, + ); + return walk.items; } /** diff --git a/src/lib/withings/sync-activity.ts b/src/lib/withings/sync-activity.ts index 1f2284f06..991bf43a0 100644 --- a/src/lib/withings/sync-activity.ts +++ b/src/lib/withings/sync-activity.ts @@ -49,7 +49,11 @@ import { } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; -import { hasActivityScope } from "./client"; +import { + hasActivityScope, + WITHINGS_MAX_PAGES, + withingsNextOffset, +} from "./client"; import { WithingsApiError, classifyWithingsResponse, @@ -66,6 +70,7 @@ import { recordSyncFailure, recordSyncSuccess, } from "@/lib/integrations/status"; +import { walkPages } from "@/lib/integrations/page-walk"; const WITHINGS_MEASURE_URL = "https://wbsapi.withings.net/v2/measure"; @@ -145,57 +150,51 @@ export async function fetchWithingsActivity( data_fields: "steps,distance,calories", }; - const results: WithingsActivityEntry[] = []; - let offset = 0; - let pageCount = 0; - while (true) { - const params = new URLSearchParams({ - ...baseParams, - offset: String(offset), - }); - const pageStart = performance.now(); - const res = await safeFetch(WITHINGS_MEASURE_URL, { - method: "POST", - headers: { - "Content-Type": "application/x-www-form-urlencoded", - Authorization: `Bearer ${accessToken}`, - }, - body: params.toString(), - }); - const json = await res.json(); - const verdict = classifyWithingsResponse(res.status, json); - getEvent()?.addExternalCall({ - service: "withings", - method: `fetchWithingsActivity(page=${pageCount})`, - duration_ms: Math.round(performance.now() - pageStart), - status: res.status, - error: verdict.classification === "success" ? undefined : verdict.reason, - }); - if (verdict.classification !== "success") { - throw new WithingsApiError({ - verb: "activity", - classification: verdict.classification, - withingsStatus: verdict.withingsStatus, - reason: verdict.reason, - upstreamError: typeof json?.error === "string" ? json.error : undefined, + const walk = await walkPages( + "withings.activity", + WITHINGS_MAX_PAGES, + async (cursor, page) => { + const offset = cursor ?? 0; + const params = new URLSearchParams({ + ...baseParams, + offset: String(offset), }); - } - - const body = json.body ?? {}; - const activities: WithingsActivityEntry[] = body.activities ?? []; - for (const entry of activities) { - results.push(entry); - } - - const hasMore = body.more === true || body.more === 1; - if (!hasMore) break; - const nextOffset = Number(body.offset); - if (!Number.isFinite(nextOffset) || nextOffset <= offset) break; - offset = nextOffset; - pageCount += 1; - if (pageCount > 1000) break; - } - return results; + const pageStart = performance.now(); + const res = await safeFetch(WITHINGS_MEASURE_URL, { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + Authorization: `Bearer ${accessToken}`, + }, + body: params.toString(), + }); + const json = await res.json(); + const verdict = classifyWithingsResponse(res.status, json); + getEvent()?.addExternalCall({ + service: "withings", + method: `fetchWithingsActivity(page=${page})`, + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: + verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new WithingsApiError({ + verb: "activity", + classification: verdict.classification, + withingsStatus: verdict.withingsStatus, + reason: verdict.reason, + upstreamError: + typeof json?.error === "string" ? json.error : undefined, + }); + } + const body = json.body ?? {}; + const activities: WithingsActivityEntry[] = body.activities ?? []; + return { items: activities, next: withingsNextOffset(body, offset) }; + }, + 0, + ); + return walk.items; } /** diff --git a/src/lib/withings/sync-ecg.ts b/src/lib/withings/sync-ecg.ts index 4721853fe..b4fe85fe7 100644 --- a/src/lib/withings/sync-ecg.ts +++ b/src/lib/withings/sync-ecg.ts @@ -80,6 +80,8 @@ import { recordSyncFailure, recordSyncSuccess, } from "@/lib/integrations/status"; +import { walkPages } from "@/lib/integrations/page-walk"; +import { WITHINGS_MAX_PAGES, withingsNextOffset } from "./client"; const WITHINGS_HEART_URL = "https://wbsapi.withings.net/v2/heart"; @@ -139,60 +141,55 @@ export async function fetchWithingsHeartList( startdate: number, enddate: number, ): Promise { - const results: WithingsHeartEntry[] = []; - let offset = 0; - let pageCount = 0; - - while (true) { - const params = new URLSearchParams({ - action: "list", - startdate: String(startdate), - enddate: String(enddate), - offset: String(offset), - }); - - const pageStart = performance.now(); - const res = await safeFetch(WITHINGS_HEART_URL, { - method: "POST", - headers: { - "Content-Type": "application/x-www-form-urlencoded", - Authorization: `Bearer ${accessToken}`, - }, - body: params.toString(), - }); - const json = await res.json(); - const verdict = classifyWithingsResponse(res.status, json); - getEvent()?.addExternalCall({ - service: "withings", - method: `fetchWithingsHeartList(page=${pageCount})`, - duration_ms: Math.round(performance.now() - pageStart), - status: res.status, - error: verdict.classification === "success" ? undefined : verdict.reason, - }); - if (verdict.classification !== "success") { - throw new WithingsApiError({ - verb: "heart", - classification: verdict.classification, - withingsStatus: verdict.withingsStatus, - reason: verdict.reason, - upstreamError: typeof json?.error === "string" ? json.error : undefined, + const walk = await walkPages( + "withings.heart", + WITHINGS_MAX_PAGES, + async (cursor, page) => { + const offset = cursor ?? 0; + const params = new URLSearchParams({ + action: "list", + startdate: String(startdate), + enddate: String(enddate), + offset: String(offset), }); - } - const body = json.body ?? {}; - const series: WithingsHeartEntry[] = body.series ?? []; - results.push(...series); + const pageStart = performance.now(); + const res = await safeFetch(WITHINGS_HEART_URL, { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + Authorization: `Bearer ${accessToken}`, + }, + body: params.toString(), + }); + const json = await res.json(); + const verdict = classifyWithingsResponse(res.status, json); + getEvent()?.addExternalCall({ + service: "withings", + method: `fetchWithingsHeartList(page=${page})`, + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: + verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new WithingsApiError({ + verb: "heart", + classification: verdict.classification, + withingsStatus: verdict.withingsStatus, + reason: verdict.reason, + upstreamError: + typeof json?.error === "string" ? json.error : undefined, + }); + } - const hasMore = body.more === true || body.more === 1; - if (!hasMore) break; - const nextOffset = Number(body.offset); - if (!Number.isFinite(nextOffset) || nextOffset <= offset) break; - offset = nextOffset; - pageCount += 1; - if (pageCount > 1000) break; - } - - return results; + const body = json.body ?? {}; + const series: WithingsHeartEntry[] = body.series ?? []; + return { items: series, next: withingsNextOffset(body, offset) }; + }, + 0, + ); + return walk.items; } /**