From d31a33e14c1d299de61427699f00234d091a9dfc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Mon, 20 Jul 2026 11:08:50 +0200 Subject: [PATCH] fix(strava): preserve cursors on workout write failure --- .../__tests__/sync-cursor-durability.test.ts | 208 ++++++++++++++++++ src/lib/strava/sync.ts | 73 +++--- 2 files changed, 254 insertions(+), 27 deletions(-) create mode 100644 src/lib/strava/__tests__/sync-cursor-durability.test.ts diff --git a/src/lib/strava/__tests__/sync-cursor-durability.test.ts b/src/lib/strava/__tests__/sync-cursor-durability.test.ts new file mode 100644 index 000000000..18019f19a --- /dev/null +++ b/src/lib/strava/__tests__/sync-cursor-durability.test.ts @@ -0,0 +1,208 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { StravaWorkoutRow } from "../client"; + +const mocks = vi.hoisted(() => ({ + fetchActivities: vi.fn(), + fetchActivityById: vi.fn(), + mapActivity: vi.fn(), + findUnique: vi.fn(), + updateMany: vi.fn(), + updateUser: vi.fn(), + createManyAndReturn: vi.fn(), + updateWorkout: vi.fn(), + getStravaConnection: vi.fn(), + recordSyncFailure: vi.fn(), + recordSyncSuccess: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + findUnique: mocks.findUnique, + updateMany: mocks.updateMany, + update: mocks.updateUser, + }, + workout: { + createManyAndReturn: mocks.createManyAndReturn, + update: mocks.updateWorkout, + }, + }, +})); +vi.mock("@/lib/arrivals/workout-emit", () => ({ + emitInsertedWorkoutArrival: vi.fn(async () => {}), +})); +vi.mock("@/lib/logging/context", () => ({ + annotate: vi.fn(), + getEvent: () => ({ addWarning: vi.fn() }), +})); +vi.mock("@/lib/integrations/status", () => ({ + recordSyncFailure: mocks.recordSyncFailure, + recordSyncSuccess: mocks.recordSyncSuccess, + toFailureKind: vi.fn(() => "transient"), +})); +vi.mock("../credentials", () => ({ + getStravaConnection: mocks.getStravaConnection, + getStravaClientCredentials: vi.fn(), + storeStravaTokens: vi.fn(), +})); +vi.mock("../client", () => ({ + fetchActivities: mocks.fetchActivities, + fetchActivityById: mocks.fetchActivityById, + mapActivity: mocks.mapActivity, + refreshAccessToken: vi.fn(), + summaryHasHeartRate: vi.fn(() => true), +})); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill"; +import { syncUserStrava } from "../sync"; + +const previousCursor = new Date("2026-07-01T08:00:00.000Z"); +const firstStart = new Date("2026-07-18T08:00:00.000Z"); +const secondStart = new Date("2026-07-19T08:00:00.000Z"); +const thirdStart = new Date("2026-07-20T08:00:00.000Z"); + +function workout(externalId: string, startedAt: Date): StravaWorkoutRow { + return { + externalId, + sportType: "running", + startedAt, + endedAt: new Date(startedAt.getTime() + 3_600_000), + durationSec: 3_600, + totalEnergyKcal: 500, + totalDistanceM: 10_000, + avgHeartRate: 145, + maxHeartRate: 170, + elevationM: 80, + metadata: {}, + }; +} + +function activity(id: number, row: StravaWorkoutRow) { + return { + id, + start_date: row.startedAt.toISOString(), + row, + }; +} + +describe("syncUserStrava cursor durability", () => { + let cursor: Date | null; + let failExternalId: string | null; + let workouts: Map; + let backfillCompletedAt: Date | null; + + beforeEach(() => { + vi.clearAllMocks(); + cursor = previousCursor; + failExternalId = null; + workouts = new Map(); + backfillCompletedAt = null; + + const rows = [ + workout("strava-1", firstStart), + workout("strava-2", secondStart), + workout("strava-3", thirdStart), + ]; + mocks.getStravaConnection.mockResolvedValue({ + accessToken: "access-token", + refreshToken: "refresh-token", + refreshTokenCiphertext: Buffer.from("ciphertext"), + }); + mocks.findUnique.mockImplementation(async () => ({ + stravaLastActivityAt: cursor, + })); + mocks.fetchActivities.mockResolvedValue( + rows.map((row, index) => activity(index + 1, row)), + ); + mocks.fetchActivityById.mockResolvedValue({}); + mocks.mapActivity.mockImplementation((source) => source.row); + mocks.createManyAndReturn.mockImplementation(async ({ data }) => { + if (data.externalId === failExternalId) { + throw new Error("database unavailable"); + } + if (workouts.has(data.externalId)) return []; + workouts.set(data.externalId, data); + return [{ id: `workout-${data.externalId}`, startedAt: data.startedAt }]; + }); + mocks.updateWorkout.mockImplementation(async ({ where, data }) => { + const externalId = where.userId_source_externalId.externalId; + const existing = workouts.get(externalId); + if (!existing) throw new Error("missing workout"); + workouts.set(externalId, { ...existing, ...data }); + return { id: `workout-${externalId}` }; + }); + mocks.updateMany.mockImplementation(async ({ data }) => { + cursor = data.stravaLastActivityAt; + return { count: 1 }; + }); + mocks.updateUser.mockImplementation(async ({ data }) => { + backfillCompletedAt = data.stravaBackfillCompletedAt; + return { id: "user-1" }; + }); + mocks.recordSyncFailure.mockResolvedValue(undefined); + mocks.recordSyncSuccess.mockResolvedValue(undefined); + }); + + it("leaves the previous cursor unchanged and rejects when a workout write fails", async () => { + failExternalId = "strava-2"; + + await expect(syncUserStrava("user-1")).rejects.toThrow( + "database unavailable", + ); + + expect(cursor).toEqual(previousCursor); + expect(mocks.updateMany).not.toHaveBeenCalled(); + expect(workouts.has("strava-3")).toBe(true); + expect(mocks.recordSyncFailure).toHaveBeenCalledWith( + expect.objectContaining({ + userId: "user-1", + integration: "strava", + message: "database unavailable", + }), + ); + expect(mocks.recordSyncSuccess).not.toHaveBeenCalled(); + }); + + it("leaves the backfill watermark unchanged when a workout write fails", async () => { + failExternalId = "strava-2"; + + await expect(runStravaBackfillForUser("user-1")).rejects.toThrow( + "database unavailable", + ); + + expect(backfillCompletedAt).toBeNull(); + expect(mocks.updateUser).not.toHaveBeenCalled(); + }); + + it("advances the backfill watermark after every workout is persisted", async () => { + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: 3, + }); + + expect(workouts.size).toBe(3); + expect(backfillCompletedAt).toBeInstanceOf(Date); + expect(mocks.updateUser).toHaveBeenCalledTimes(1); + }); + + it("advances the cursor only after every fetched workout is persisted", async () => { + await expect(syncUserStrava("user-1")).resolves.toBe(3); + + expect(workouts.size).toBe(3); + expect(cursor).toEqual(thirdStart); + expect(mocks.updateMany).toHaveBeenCalledTimes(1); + expect(mocks.recordSyncSuccess).toHaveBeenCalledWith("user-1", "strava"); + }); + + it("replays by external identity without creating duplicate workouts", async () => { + await expect(syncUserStrava("user-1")).resolves.toBe(3); + await expect(syncUserStrava("user-1")).resolves.toBe(3); + + expect(workouts.size).toBe(3); + expect(mocks.createManyAndReturn).toHaveBeenCalledTimes(6); + expect(mocks.updateWorkout).toHaveBeenCalledTimes(3); + expect(cursor).toEqual(thirdStart); + }); +}); diff --git a/src/lib/strava/sync.ts b/src/lib/strava/sync.ts index 4c195512c..800316eaf 100644 --- a/src/lib/strava/sync.ts +++ b/src/lib/strava/sync.ts @@ -68,6 +68,19 @@ export function classifyStravaFailure(err: unknown): FailureKind { return toFailureKind(classifyStravaError(err)); } +async function recordStravaFailure(userId: string, err: unknown): Promise { + await recordSyncFailure({ + userId, + integration: "strava", + kind: classifyStravaFailure(err), + message: err instanceof Error ? err.message : String(err), + errorCode: + err instanceof StravaApiError && err.httpStatus != null + ? String(err.httpStatus) + : undefined, + }); +} + export interface SyncUserStravaOptions { fullSync?: boolean; lookbackDays?: number; @@ -198,35 +211,32 @@ export async function syncUserStrava( if (activities.length < PER_PAGE) break; } } catch (err) { - await recordSyncFailure({ - userId, - integration: "strava", - kind: classifyStravaFailure(err), - message: err instanceof Error ? err.message : String(err), - errorCode: - err instanceof StravaApiError && err.httpStatus != null - ? String(err.httpStatus) - : undefined, - }); + await recordStravaFailure(userId, err); throw err; } - const imported = await upsertStravaWorkouts(userId, rows); + let imported: number; + try { + imported = await upsertStravaWorkouts(userId, rows); - // Advance the incremental cursor to the newest activity start we saw, but - // never move it backwards (a bounded backfill page-walk may not reach the - // very newest activity in one pass). - if (newestStart) { - await prisma.user.updateMany({ - where: { - id: userId, - OR: [ - { stravaLastActivityAt: null }, - { stravaLastActivityAt: { lt: newestStart } }, - ], - }, - data: { stravaLastActivityAt: newestStart }, - }); + // Advance the incremental cursor to the newest activity start we saw, but + // never move it backwards (a bounded backfill page-walk may not reach the + // very newest activity in one pass). + if (newestStart) { + await prisma.user.updateMany({ + where: { + id: userId, + OR: [ + { stravaLastActivityAt: null }, + { stravaLastActivityAt: { lt: newestStart } }, + ], + }, + data: { stravaLastActivityAt: newestStart }, + }); + } + } catch (err) { + await recordStravaFailure(userId, err); + throw err; } await recordSyncSuccess(userId, "strava"); @@ -239,8 +249,10 @@ export async function syncUserStrava( /** * Upsert a batch of mapped Strava activities into `Workout`, keyed on * `(userId, source = STRAVA, externalId)` so a re-fetch overwrites in place. - * Best-effort per row: a single bad row is logged, never thrown, so it can't - * fail the surrounding sync. + * `(userId, source = STRAVA, externalId)` so replays overwrite in place. + * Failed rows are logged while the rest of the batch is attempted, then the + * first error is rethrown so the caller cannot advance its cursor or backfill + * watermark past a partially persisted batch. */ export async function upsertStravaWorkouts( userId: string, @@ -249,6 +261,8 @@ export async function upsertStravaWorkouts( if (rows.length === 0) return 0; let imported = 0; + let firstError: unknown; + let hadError = false; for (const r of rows) { const data = { sportType: r.sportType, @@ -292,8 +306,13 @@ export async function upsertStravaWorkouts( imported += 1; } catch (err) { getEvent()?.addWarning(`strava: failed to upsert workout: ${err}`); + if (!hadError) { + firstError = err; + hadError = true; + } } } + if (hadError) throw firstError; annotate({ action: { name: "strava.workout.ingest", details: { imported } },