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);