From 93b3f36d96ced1f68bab537bb4880c47e161e957 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Tue, 20 Oct 2026 06:49:44 +0200 Subject: [PATCH] fix(strava): walk activity pages through walkPages and gate the backfill on it The activity loop stopped at its page ceiling and returned what it had read, and the backfill stamped the account complete over it: a history longer than forty pages was cut off for good. The walk now runs through walkPages, which reports a walk cut off at its ceiling, and the backfill reads backfillVerdict like the WHOOP and Fitbit backfills: a truncated walk leaves the marker unset without a retry, and a connection waiting for re-authorisation returns without walking. --- src/lib/jobs/strava-backfill.ts | 45 ++++- .../__tests__/sync-cursor-durability.test.ts | 1 + .../strava/__tests__/sync-page-walk.test.ts | 191 ++++++++++++++++++ src/lib/strava/sync.ts | 98 +++++---- 4 files changed, 289 insertions(+), 46 deletions(-) create mode 100644 src/lib/strava/__tests__/sync-page-walk.test.ts diff --git a/src/lib/jobs/strava-backfill.ts b/src/lib/jobs/strava-backfill.ts index 9301bb11e..539396dad 100644 --- a/src/lib/jobs/strava-backfill.ts +++ b/src/lib/jobs/strava-backfill.ts @@ -18,6 +18,11 @@ import { prisma } from "@/lib/db"; import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; +import { isReauthRequired } from "@/lib/integrations/status"; +import { + backfillVerdict, + withPageWalkLedger, +} from "@/lib/integrations/page-walk"; import { syncUserStrava } from "@/lib/strava/sync"; export const STRAVA_BACKFILL_QUEUE = "strava-backfill"; @@ -36,14 +41,46 @@ export interface StravaBackfillPayload { /** * Per-user backfill handler. Runs a full-history sync for one account and - * stamps `stravaBackfillCompletedAt` so the discovery query drops it. - * Idempotent: the `(userId, source, externalId)` workout upsert is key-stable, - * so a re-run (e.g. a reboot mid-walk) overwrites rather than duplicating. + * stamps `stravaBackfillCompletedAt` only when the walk read the whole + * history, so the discovery query drops it. Idempotent: the + * `(userId, source, externalId)` workout upsert is key-stable, so a re-run + * (e.g. a reboot mid-walk) overwrites rather than duplicating. + * + * Verdict-gated (`backfillVerdict`, `integrations/page-walk.ts`), like the + * WHOOP and Fitbit backfills. A failed sync throws out of `syncUserStrava` + * (failure recorded first) and pg-boss retries. An activity walk that reached + * its page ceiling with history left is not complete: the marker stays unset + * without a retry, and boot discovery offers the account again. A connection + * parked for re-authorisation returns without stamping and without a retry, + * since every attempt would meet the same dead grant. */ export async function runStravaBackfillForUser( userId: string, ): Promise<{ imported: number }> { - const imported = await syncUserStrava(userId, { fullSync: true }); + if (await isReauthRequired(userId, "strava")) { + annotate({ + action: { + name: "strava.backfill.skipped_reauth", + details: { imported: 0 }, + }, + }); + return { imported: 0 }; + } + + const { result: imported, truncated } = await withPageWalkLedger(() => + syncUserStrava(userId, { fullSync: true }), + ); + // A failure has already thrown; what is left to decide is whether the walk + // reached the beginning of the history. + if (backfillVerdict({ failed: false, truncated }) === "incomplete") { + annotate({ + action: { + name: "strava.backfill.truncated", + details: { imported, collections: truncated.join(",") }, + }, + }); + return { imported }; + } await prisma.user.update({ where: { id: userId }, diff --git a/src/lib/strava/__tests__/sync-cursor-durability.test.ts b/src/lib/strava/__tests__/sync-cursor-durability.test.ts index dc71edf55..636e529ba 100644 --- a/src/lib/strava/__tests__/sync-cursor-durability.test.ts +++ b/src/lib/strava/__tests__/sync-cursor-durability.test.ts @@ -38,6 +38,7 @@ vi.mock("@/lib/logging/context", () => ({ vi.mock("@/lib/integrations/status", () => ({ recordSyncFailure: mocks.recordSyncFailure, recordSyncSuccess: mocks.recordSyncSuccess, + isReauthRequired: vi.fn(async () => false), toFailureKind: vi.fn(() => "transient"), markSyncFailureRecorded: (err: T) => err, })); diff --git a/src/lib/strava/__tests__/sync-page-walk.test.ts b/src/lib/strava/__tests__/sync-page-walk.test.ts new file mode 100644 index 000000000..3f2f09f0e --- /dev/null +++ b/src/lib/strava/__tests__/sync-page-walk.test.ts @@ -0,0 +1,191 @@ +/** + * The Strava activity walk runs through `walkPages`, and the backfill reads + * its verdict (`integrations/page-walk.ts`): a walk that reaches its page + * ceiling with a full page last is truncated, and the backfill leaves its + * marker unset instead of stamping a history it did not finish. Before, the + * loop stopped at forty pages and the backfill stamped the account complete. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + fetchActivities: vi.fn(), + annotate: vi.fn(), + updateUser: vi.fn(), + isReauthRequired: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + findUnique: vi.fn(async () => ({ stravaLastActivityAt: null })), + updateMany: vi.fn(async () => ({ count: 1 })), + update: mocks.updateUser, + }, + workout: { + createManyAndReturn: vi.fn(async ({ data }) => [ + { id: `w-${data.externalId}`, startedAt: data.startedAt }, + ]), + update: vi.fn(), + }, + }, +})); +vi.mock("@/lib/arrivals/workout-emit", () => ({ + emitInsertedWorkoutArrival: vi.fn(async () => {}), +})); +vi.mock("@/lib/cache/invalidate", () => ({ + invalidateUserMeasurements: vi.fn(), +})); +vi.mock("@/lib/logging/context", () => ({ + annotate: mocks.annotate, + getEvent: () => ({ addWarning: vi.fn() }), +})); +vi.mock("@/lib/integrations/status", () => ({ + recordSyncFailure: vi.fn(), + recordSyncSuccess: vi.fn(), + isReauthRequired: mocks.isReauthRequired, + toFailureKind: vi.fn(() => "transient"), + markSyncFailureRecorded: (err: T) => err, +})); +vi.mock("../credentials", () => ({ + getStravaConnection: vi.fn(async () => ({ + accessToken: "access-token", + refreshToken: "refresh-token", + refreshTokenCiphertext: Buffer.from("ciphertext"), + })), + getStravaClientCredentials: vi.fn(), + storeStravaTokens: vi.fn(), +})); +vi.mock("../client", async (importOriginal) => ({ + ...(await importOriginal()), + fetchActivities: mocks.fetchActivities, + // Summaries carry HR; the detail fetch only fills calories. + fetchActivityById: vi.fn(async () => ({})), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill"; +import { syncUserStrava } from "../sync"; + +const PER_PAGE = 100; +const BACKFILL_MAX_PAGES = 40; + +/** A page of `n` activities; ids continue across pages. */ +function page(pageNo: number, n: number) { + return Array.from({ length: n }, (_, i) => { + const id = (pageNo - 1) * PER_PAGE + i + 1; + return { + id, + sport_type: "Run", + start_date: new Date(Date.UTC(2020, 0, 1) + id * 3_600_000).toISOString(), + moving_time: 1800, + elapsed_time: 1900, + average_heartrate: 140, + }; + }); +} + +/** A history of `total` activities, served page by page. */ +function history(total: number) { + mocks.fetchActivities.mockImplementation( + async (_t: string, q: { page: number; perPage: number }) => { + const remaining = total - (q.page - 1) * q.perPage; + return page(q.page, Math.max(0, Math.min(q.perPage, remaining))); + }, + ); +} + +function truncatedAnnotations() { + return mocks.annotate.mock.calls.filter( + ([e]) => e?.action?.name === "integration.page_walk.truncated", + ); +} + +beforeEach(() => { + vi.clearAllMocks(); + mocks.isReauthRequired.mockResolvedValue(false); + mocks.updateUser.mockResolvedValue({ id: "user-1" }); +}); + +describe("Strava activity walk", () => { + it("reads pages until a short one and stops there", async () => { + history(2 * PER_PAGE + 7); + + await expect(syncUserStrava("user-1", { fullSync: true })).resolves.toBe( + 207, + ); + expect(mocks.fetchActivities.mock.calls.map(([, q]) => q.page)).toEqual([ + 1, 2, 3, + ]); + expect(truncatedAnnotations()).toEqual([]); + }); + + it("ends on an empty page without calling it truncated", async () => { + history(PER_PAGE); + + await expect(syncUserStrava("user-1", { fullSync: true })).resolves.toBe( + PER_PAGE, + ); + expect(mocks.fetchActivities).toHaveBeenCalledTimes(2); + expect(truncatedAnnotations()).toEqual([]); + }); + + it("reports a walk cut off at its ceiling", async () => { + history((BACKFILL_MAX_PAGES + 5) * PER_PAGE); + + await syncUserStrava("user-1", { fullSync: true }); + + expect(mocks.fetchActivities).toHaveBeenCalledTimes(BACKFILL_MAX_PAGES); + expect(truncatedAnnotations()).toEqual([ + [ + expect.objectContaining({ + meta: { + walk: "strava.activities", + pages: BACKFILL_MAX_PAGES, + reason: "ceiling", + }, + }), + ], + ]); + }); +}); + +describe("Strava backfill verdict", () => { + it("stamps the marker when the walk reached the start of the history", async () => { + history(3 * PER_PAGE - 1); + + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: 3 * PER_PAGE - 1, + }); + expect(mocks.updateUser).toHaveBeenCalledWith({ + where: { id: "user-1" }, + data: { stravaBackfillCompletedAt: expect.any(Date) }, + }); + }); + + it("leaves the marker unset, without a retry, when the walk was cut off", async () => { + history((BACKFILL_MAX_PAGES + 1) * PER_PAGE); + + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: BACKFILL_MAX_PAGES * PER_PAGE, + }); + expect(mocks.updateUser).not.toHaveBeenCalled(); + expect(mocks.annotate).toHaveBeenCalledWith( + expect.objectContaining({ + action: expect.objectContaining({ name: "strava.backfill.truncated" }), + }), + ); + }); + + it("returns without walking or stamping while the connection waits for re-authorisation", async () => { + mocks.isReauthRequired.mockResolvedValue(true); + history(PER_PAGE); + + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: 0, + }); + expect(mocks.fetchActivities).not.toHaveBeenCalled(); + expect(mocks.updateUser).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/strava/sync.ts b/src/lib/strava/sync.ts index 38e5600ba..89070f8fd 100644 --- a/src/lib/strava/sync.ts +++ b/src/lib/strava/sync.ts @@ -19,7 +19,8 @@ * `reauth_required` on the `strava` ledger. * * Rate limits: 200 req / 15 min + 2000 / day per app. The walk is page-bounded - * and the per-activity detail fetch (for `calories`, absent from the summary) + * (`walkPages`, which reports a walk cut off at its ceiling instead of + * passing it off as the whole history) and the per-activity detail fetch (for `calories`, absent from the summary) * is budget-capped; a 429 mid-walk records a transient failure and the next * cron tick resumes from the cursor. */ @@ -34,6 +35,7 @@ import { toFailureKind, type FailureKind, } from "@/lib/integrations/status"; +import { walkPages } from "@/lib/integrations/page-walk"; import { fetchActivities, fetchActivityById, @@ -163,58 +165,70 @@ export async function syncUserStrava( ? BACKFILL_DETAIL_BUDGET : INCREMENTAL_DETAIL_BUDGET; - const rows: StravaWorkoutRow[] = []; + let rows: StravaWorkoutRow[]; let newestStart: Date | null = null; try { - for (let page = 1; page <= maxPages; page++) { - const activities = await authed((t) => - fetchActivities(t, { after: afterEpoch, page, perPage: PER_PAGE }), - ); - if (activities.length === 0) break; - - for (const a of activities) { - // Fetch detail only to fill `calories` (absent from the summary), and - // only while the budget allows. When the summary already lacks HR we - // also let the detail top it up. A detail failure is non-fatal — map - // from the summary — but a 429 aborts the whole walk (transient). - let detail: StravaDetailedActivity | null = null; - if (detailBudget > 0) { - try { - detail = await authed((t) => fetchActivityById(t, a.id)); - detailBudget -= 1; - } catch (err) { - if (err instanceof StravaApiError && err.httpStatus === 429) { - throw err; + // Strava pages by number; a short page is the last one. A walk that + // reaches its ceiling with a full page last reports itself truncated, and + // the backfill leaves its marker unset (`integrations/page-walk.ts`). + const walk = await walkPages( + "strava.activities", + maxPages, + async (cursor) => { + const page = cursor ?? 1; + const activities = await authed((t) => + fetchActivities(t, { after: afterEpoch, page, perPage: PER_PAGE }), + ); + const mapped: StravaWorkoutRow[] = []; + for (const a of activities) { + // Fetch detail only to fill `calories` (absent from the summary), + // and only while the budget allows. When the summary already lacks + // HR we also let the detail top it up. A detail failure is + // non-fatal — map from the summary — but a 429 aborts the whole + // walk (transient). + let detail: StravaDetailedActivity | null = null; + if (detailBudget > 0) { + try { + detail = await authed((t) => fetchActivityById(t, a.id)); + detailBudget -= 1; + } catch (err) { + if (err instanceof StravaApiError && err.httpStatus === 429) { + throw err; + } + getEvent()?.addWarning( + `strava: activity detail fetch failed for ${a.id}: ${err}`, + ); + detail = null; } + } else if (!summaryHasHeartRate(a)) { getEvent()?.addWarning( - `strava: activity detail fetch failed for ${a.id}: ${err}`, + `strava: detail budget exhausted; mapping activity ${a.id} from summary only`, ); - detail = null; } - } else if (!summaryHasHeartRate(a)) { - getEvent()?.addWarning( - `strava: detail budget exhausted; mapping activity ${a.id} from summary only`, - ); - } - const row = mapActivity(a, detail); - if (row) { - rows.push(row); - if (a.start_date) { - const d = new Date(a.start_date); - if ( - !Number.isNaN(d.getTime()) && - (!newestStart || d > newestStart) - ) { - newestStart = d; + const row = mapActivity(a, detail); + if (row) { + mapped.push(row); + if (a.start_date) { + const d = new Date(a.start_date); + if ( + !Number.isNaN(d.getTime()) && + (!newestStart || d > newestStart) + ) { + newestStart = d; + } } } } - } - - if (activities.length < PER_PAGE) break; - } + return { + items: mapped, + next: activities.length < PER_PAGE ? null : page + 1, + }; + }, + 1, + ); + rows = walk.items; } catch (err) { await recordStravaFailure(userId, err); throw markSyncFailureRecorded(err);