fix(strava): resume an unfinished backfill where the last run stopped

One run reads at most forty pages of a hundred activities. A longer history
ended as incomplete with no note of how far it got, so every boot read the
same newest pages again, spent up to 190 of the instance's 200 requests per
fifteen minutes on them, and never reached the older activities.

The run now records the start of the oldest activity it read and sends the
next run for after the rate window; that run asks Strava for the activities
before it. A run that gets no further records nothing and sends nothing. The
place is cleared when the backfill completes.

prisma/migrations/0413_strava_backfill_resume/migration.sql

@@ -0,0 +1,12 @@
+-- Where an unfinished Strava backfill picks up.
+--
+-- One backfill run reads at most forty pages of a hundred activities. A
+-- longer history ended the run as incomplete without a note of how far it
+-- got, so every boot read the same newest forty pages again, up to 190 of
+-- the 200 requests the whole instance may make in fifteen minutes, and the
+-- older history was never reached. The column holds the start of the oldest
+-- activity the last run read; the next run asks for activities before it.
+-- Nullable and unset for every existing row: an unfinished backfill starts
+-- once more from the newest page and records its place from then on.
+
+ALTER TABLE "users" ADD COLUMN "strava_backfill_resume_before" TIMESTAMP(3);

prisma/schema.prisma

@@ -316,6 +316,11 @@ model User {
   // First-connect full-history backfill marker (self-converging discovery
   // predicate `strava_backfill_completed_at IS NULL`), mirroring WHOOP/Fitbit.
   stravaBackfillCompletedAt   DateTime? @map("strava_backfill_completed_at")
+  // Where an unfinished backfill picks up: the start of the oldest activity
+  // the last run read. A history longer than one run's page ceiling is read
+  // in runs that each walk further back from here, instead of every run
+  // starting again at the newest page. Cleared when the backfill completes.
+  stravaBackfillResumeBefore  DateTime? @map("strava_backfill_resume_before")
 
   // Locale (null = use system/admin default)
   locale String? @map("locale") // "de" | "en" | null

src/lib/data-wipe/wipe-plan.ts

@@ -469,6 +469,7 @@ export const USER_RESET = {
   stravaAthleteId: null,
   stravaLastActivityAt: null,
   stravaBackfillCompletedAt: null,
+  stravaBackfillResumeBefore: null,
   healthKitConfigJson: Prisma.DbNull,
   healthKitLastSyncedAt: null,
   healthKitLastSyncTrigger: null,

src/lib/export/backup-plan.ts

@@ -960,6 +960,7 @@ export const USER_COLUMN_BACKUP_CLASS = {
   coachLastSeenAt: "OPERATIONAL",
   stravaLastActivityAt: "OPERATIONAL",
   stravaBackfillCompletedAt: "OPERATIONAL",
+  stravaBackfillResumeBefore: "OPERATIONAL",
   healthKitLastSyncedAt: "OPERATIONAL",
   healthKitLastSyncTrigger: "OPERATIONAL",
   healthKitLastBackgroundSyncAt: "OPERATIONAL",

src/lib/jobs/integration-backfill-drain.ts

@@ -132,11 +132,15 @@ async function runAdmitted(
       return;
     }
     case "strava-backfill": {
-      const { imported } = await runStravaBackfillForUser(userId);
+      const { imported, continued } = await runStravaBackfillForUser(
+        userId,
+        data.slice ?? 0,
+      );
       totals.imported += imported;
+      if (continued) totals.continued += 1;
       workerLog(
         "info",
-        `[strava-backfill] user=${userId} imported=${imported}`,
+        `[strava-backfill] user=${userId} imported=${imported}${continued ? " continued" : ""}`,
       );
       return;
     }

src/lib/jobs/strava-backfill.ts

@@ -17,13 +17,16 @@
 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 {
+  enqueueIntegrationBackfillAdmission,
+  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";
+import { syncStravaHistory } from "@/lib/strava/sync";
 
 export const STRAVA_BACKFILL_QUEUE = "strava-backfill";
 
@@ -40,23 +43,38 @@ export interface StravaBackfillPayload {
 }
 
 /**
- * Per-user backfill handler. Runs a full-history sync for one account and
- * 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
+ * The pause before the next run of an unfinished backfill: Strava's
+ * short-term rate window. One run spends up to forty page reads and 150
+ * detail reads of the 200 the app may make in fifteen minutes, so the next
+ * run waits for a fresh window rather than opening on a 429 or starving the
+ * hourly syncs.
+ */
+export const STRAVA_BACKFILL_CONTINUATION_DELAY_SECONDS = 15 * 60;
+
+/**
+ * Per-user backfill handler. Runs one stretch of the full-history walk for
+ * one account and stamps `stravaBackfillCompletedAt` only when the walk read
+ * to the start of the 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.
+ * WHOOP and Fitbit backfills. A failed sync throws out of the walk (failure
+ * recorded first) and pg-boss retries. A walk that reached its page ceiling
+ * with history left is not complete: it records where it got to
+ * (`stravaBackfillResumeBefore`, the start of the oldest activity it read)
+ * and sends the next run for after the rate window, which asks for the
+ * activities before that point. Without the note every run began again at
+ * the newest page, read the same forty pages and never reached the rest. A
+ * run that got no further than the last one records nothing and sends
+ * nothing, and the next boot 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 }> {
+  slice: number = 0,
+): Promise<{ imported: number; continued: boolean }> {
   if (await isReauthRequired(userId, "strava")) {
     annotate({
       action: {
@@ -64,27 +82,56 @@ export async function runStravaBackfillForUser(
         details: { imported: 0 },
       },
     });
-    return { imported: 0 };
+    return { imported: 0, continued: false };
   }
 
-  const { result: imported, truncated } = await withPageWalkLedger(() =>
-    syncUserStrava(userId, { fullSync: true }),
+  const stored = await prisma.user.findUnique({
+    where: { id: userId },
+    select: { stravaBackfillResumeBefore: true },
+  });
+  const before = stored?.stravaBackfillResumeBefore ?? undefined;
+  const { result, truncated } = await withPageWalkLedger(() =>
+    syncStravaHistory(userId, before),
   );
+  const { imported, oldestStart } = result;
   // 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") {
+    // One second past the oldest start, so an activity that shares it is
+    // read again rather than skipped; the upsert makes the overlap free.
+    const resumeBefore = oldestStart
+      ? new Date(oldestStart.getTime() + 1000)
+      : null;
+    const progressed =
+      resumeBefore !== null &&
+      (before === undefined || resumeBefore.getTime() < before.getTime());
+    if (progressed) {
+      await prisma.user.update({
+        where: { id: userId },
+        data: { stravaBackfillResumeBefore: resumeBefore },
+      });
+    }
+    const continued = progressed && (await sendNextRun(userId, slice));
     annotate({
       action: {
         name: "strava.backfill.truncated",
-        details: { imported, collections: truncated.join(",") },
+        details: {
+          imported,
+          collections: truncated.join(","),
+          progressed,
+          continued,
+        },
       },
     });
-    return { imported };
+    return { imported, continued };
   }
 
   await prisma.user.update({
     where: { id: userId },
-    data: { stravaBackfillCompletedAt: new Date() },
+    data: {
+      stravaBackfillCompletedAt: new Date(),
+      stravaBackfillResumeBefore: null,
+    },
   });
 
   annotate({
@@ -93,7 +140,21 @@ export async function runStravaBackfillForUser(
       details: { imported },
     },
   });
-  return { imported };
+  return { imported, continued: false };
+}
+
+async function sendNextRun(userId: string, slice: number): Promise<boolean> {
+  const boss = getGlobalBoss();
+  if (!boss) return false;
+  const sent = await enqueueIntegrationBackfillAdmission(
+    boss,
+    {
+      kind: "strava-backfill",
+      data: { userId, enqueuedAt: new Date().toISOString(), slice: slice + 1 },
+    },
+    STRAVA_BACKFILL_CONTINUATION_DELAY_SECONDS,
+  );
+  return Boolean(sent);
 }
 
 /**

src/lib/strava/__tests__/sync-cursor-durability.test.ts

@@ -182,6 +182,7 @@ describe("syncUserStrava cursor durability", () => {
   it("advances the backfill watermark after every workout is persisted", async () => {
     await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({
       imported: 3,
+      continued: false,
     });
 
     expect(workouts.size).toBe(3);

src/lib/strava/__tests__/sync-page-walk.test.ts

@@ -11,13 +11,15 @@ const mocks = vi.hoisted(() => ({
   fetchActivities: vi.fn(),
   annotate: vi.fn(),
   updateUser: vi.fn(),
+  findUser: vi.fn(),
   isReauthRequired: vi.fn(),
+  send: vi.fn(),
 }));
 
 vi.mock("@/lib/db", () => ({
   prisma: {
     user: {
-      findUnique: vi.fn(async () => ({ stravaLastActivityAt: null })),
+      findUnique: mocks.findUser,
       updateMany: vi.fn(async () => ({ count: 1 })),
       update: mocks.updateUser,
     },
@@ -62,7 +64,7 @@ vi.mock("../client", async (importOriginal) => ({
   fetchActivityById: vi.fn(async () => ({})),
 }));
 vi.mock("@/lib/jobs/boss-instance", () => ({
-  getGlobalBoss: vi.fn(() => null),
+  getGlobalBoss: vi.fn(() => ({ send: mocks.send })),
 }));
 
 import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill";
@@ -71,27 +73,36 @@ 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,
-    };
-  });
+const HOUR = 3_600_000;
+const T0 = Date.UTC(2020, 0, 1);
+
+/** Activity `id` of a history; a higher id started later. */
+function activity(id: number) {
+  return {
+    id,
+    sport_type: "Run",
+    start_date: new Date(T0 + id * HOUR).toISOString(),
+    moving_time: 1800,
+    elapsed_time: 1900,
+    average_heartrate: 140,
+  };
 }
 
-/** A history of `total` activities, served page by page. */
+/**
+ * A history of `total` activities served as Strava serves a walk without
+ * `after`: newest first, `before` (epoch seconds) cutting off the newer ones.
+ */
 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)));
+    async (
+      _t: string,
+      q: { page: number; perPage: number; before?: number },
+    ) => {
+      const ids = Array.from({ length: total }, (_, i) => total - i).filter(
+        (id) => q.before === undefined || T0 + id * HOUR < q.before * 1000,
+      );
+      const from = (q.page - 1) * q.perPage;
+      return ids.slice(from, from + q.perPage).map(activity);
     },
   );
 }
@@ -102,10 +113,30 @@ function truncatedAnnotations() {
   );
 }
 
+/** The resume point the user row holds, as the backfill writes it. */
+let resumeBefore: Date | null = null;
+
 beforeEach(() => {
   vi.clearAllMocks();
+  resumeBefore = null;
   mocks.isReauthRequired.mockResolvedValue(false);
-  mocks.updateUser.mockResolvedValue({ id: "user-1" });
+  mocks.findUser.mockImplementation(async () => ({
+    stravaLastActivityAt: null,
+    stravaBackfillResumeBefore: resumeBefore,
+  }));
+  mocks.updateUser.mockImplementation(
+    async ({
+      data,
+    }: {
+      data: { stravaBackfillResumeBefore?: Date | null };
+    }) => {
+      if ("stravaBackfillResumeBefore" in data) {
+        resumeBefore = data.stravaBackfillResumeBefore ?? null;
+      }
+      return { id: "user-1" };
+    },
+  );
+  mocks.send.mockResolvedValue("job-next");
 });
 
 describe("Strava activity walk", () => {
@@ -157,20 +188,42 @@ describe("Strava backfill verdict", () => {
 
     await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({
       imported: 3 * PER_PAGE - 1,
+      continued: false,
     });
     expect(mocks.updateUser).toHaveBeenCalledWith({
       where: { id: "user-1" },
-      data: { stravaBackfillCompletedAt: expect.any(Date) },
+      data: {
+        stravaBackfillCompletedAt: expect.any(Date),
+        stravaBackfillResumeBefore: null,
+      },
     });
+    expect(mocks.send).not.toHaveBeenCalled();
   });
 
-  it("leaves the marker unset, without a retry, when the walk was cut off", async () => {
+  it("records where a cut-off walk got to and sends the next run after the rate window", async () => {
     history((BACKFILL_MAX_PAGES + 1) * PER_PAGE);
 
     await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({
       imported: BACKFILL_MAX_PAGES * PER_PAGE,
+      continued: true,
     });
-    expect(mocks.updateUser).not.toHaveBeenCalled();
+    // The oldest activity read is id 101; the next run asks for those before.
+    expect(resumeBefore).toEqual(new Date(T0 + 101 * HOUR + 1000));
+    expect(mocks.updateUser).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: expect.objectContaining({
+          stravaBackfillCompletedAt: expect.any(Date),
+        }),
+      }),
+    );
+    expect(mocks.send).toHaveBeenCalledWith(
+      "integration-backfill-admission",
+      expect.objectContaining({
+        kind: "strava-backfill",
+        data: expect.objectContaining({ userId: "user-1", slice: 1 }),
+      }),
+      expect.objectContaining({ startAfter: 15 * 60 }),
+    );
     expect(mocks.annotate).toHaveBeenCalledWith(
       expect.objectContaining({
         action: expect.objectContaining({ name: "strava.backfill.truncated" }),
@@ -178,12 +231,63 @@ describe("Strava backfill verdict", () => {
     );
   });
 
+  it("goes on from the recorded place instead of the newest page, and finishes", async () => {
+    const total = (BACKFILL_MAX_PAGES + 10) * PER_PAGE;
+    history(total);
+
+    await runStravaBackfillForUser("user-1");
+    const firstRun = mocks.fetchActivities.mock.calls.length;
+    mocks.fetchActivities.mockClear();
+    await expect(runStravaBackfillForUser("user-1", 1)).resolves.toEqual({
+      imported: 10 * PER_PAGE + 1,
+      continued: false,
+    });
+
+    expect(firstRun).toBe(BACKFILL_MAX_PAGES);
+    // Every page of the second run asked for activities before the place
+    // the first one recorded (the oldest it read was id 1001), and the walk
+    // ended at the oldest activity.
+    const queries = mocks.fetchActivities.mock.calls.map(([, q]) => q);
+    expect(new Set(queries.map((q) => q.before))).toEqual(
+      new Set([(T0 + 1001 * HOUR) / 1000 + 1]),
+    );
+    expect(queries.length).toBe(11);
+    expect(resumeBefore).toBeNull();
+    expect(mocks.updateUser).toHaveBeenLastCalledWith({
+      where: { id: "user-1" },
+      data: {
+        stravaBackfillCompletedAt: expect.any(Date),
+        stravaBackfillResumeBefore: null,
+      },
+    });
+  });
+
+  it("records nothing and sends nothing when a run got no further than the last", async () => {
+    history((BACKFILL_MAX_PAGES + 1) * PER_PAGE);
+    // A place older than anything the walk can read past: the provider kept
+    // handing back the same newest pages.
+    mocks.fetchActivities.mockImplementation(async (_t, q) =>
+      Array.from({ length: PER_PAGE }, (_, i) =>
+        activity(5_000 - (q.page - 1) * PER_PAGE - i),
+      ),
+    );
+    resumeBefore = new Date(T0);
+
+    await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({
+      imported: BACKFILL_MAX_PAGES * PER_PAGE,
+      continued: false,
+    });
+    expect(resumeBefore).toEqual(new Date(T0));
+    expect(mocks.send).not.toHaveBeenCalled();
+  });
+
   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,
+      continued: false,
     });
     expect(mocks.fetchActivities).not.toHaveBeenCalled();
     expect(mocks.updateUser).not.toHaveBeenCalled();

src/lib/strava/client.ts

@@ -225,6 +225,8 @@ export interface StravaDetailedActivity extends StravaSummaryActivity {
 export interface FetchActivitiesQuery {
   /** Epoch SECONDS cursor — return activities started after this instant. */
   after?: number;
+  /** Epoch SECONDS — return activities started before this instant. */
+  before?: number;
   page?: number;
   perPage?: number;
 }
@@ -269,6 +271,8 @@ export async function fetchActivities(
   const params = new URLSearchParams();
   if (typeof query.after === "number")
     params.set("after", String(Math.floor(query.after)));
+  if (typeof query.before === "number")
+    params.set("before", String(Math.floor(query.before)));
   params.set("page", String(query.page ?? 1));
   params.set("per_page", String(query.perPage ?? 100));
   const json = await stravaGet<StravaSummaryActivity[] | null>(

src/lib/strava/sync.ts

@@ -102,10 +102,43 @@ export async function syncUserStrava(
   userId: string,
   opts: SyncUserStravaOptions = {},
 ): Promise<number> {
+  return (await runStravaSync(userId, opts)).imported;
+}
+
+/** What one history run read, for the backfill to resume from. */
+export interface StravaHistoryRun {
+  imported: number;
+  /**
+   * The start of the oldest activity the run read, when the pages came
+   * newest first as Strava serves a walk without `after`; null otherwise,
+   * so a place is never recorded from pages in an order the resume does not
+   * expect.
+   */
+  oldestStart: Date | null;
+}
+
+/**
+ * One run of the full-history walk, from `before` (or the newest activity)
+ * back as far as one run's page ceiling reaches.
+ */
+export async function syncStravaHistory(
+  userId: string,
+  before?: Date,
+): Promise<StravaHistoryRun> {
+  return runStravaSync(userId, { fullSync: true }, before);
+}
+
+async function runStravaSync(
+  userId: string,
+  opts: SyncUserStravaOptions,
+  before?: Date,
+): Promise<StravaHistoryRun> {
   const conn = await getStravaConnection(userId);
-  if (!conn) return 0;
+  if (!conn) return { imported: 0, oldestStart: null };
 
   const fullSync = opts.fullSync ?? false;
+  const beforeEpoch =
+    before === undefined ? undefined : Math.floor(before.getTime() / 1000);
 
   // Resolve the incremental cursor. On a full backfill we walk all history
   // (`after` omitted). Otherwise start at the stored newest-activity instant
@@ -167,6 +200,9 @@ export async function syncUserStrava(
 
   let rows: StravaWorkoutRow[];
   let newestStart: Date | null = null;
+  let oldestStart: Date | null = null;
+  let previousStart: number | null = null;
+  let newestFirst = true;
 
   try {
     // Strava pages by number; a short page is the last one. A walk that
@@ -178,7 +214,12 @@ export async function syncUserStrava(
       async (cursor) => {
         const page = cursor ?? 1;
         const activities = await authed((t) =>
-          fetchActivities(t, { after: afterEpoch, page, perPage: PER_PAGE }),
+          fetchActivities(t, {
+            after: afterEpoch,
+            before: beforeEpoch,
+            page,
+            perPage: PER_PAGE,
+          }),
         );
         const mapped: StravaWorkoutRow[] = [];
         for (const a of activities) {
@@ -212,11 +253,13 @@ export async function syncUserStrava(
             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 (!Number.isNaN(d.getTime())) {
+                if (!newestStart || d > newestStart) newestStart = d;
+                if (!oldestStart || d < oldestStart) oldestStart = d;
+                if (previousStart !== null && d.getTime() > previousStart) {
+                  newestFirst = false;
+                }
+                previousStart = d.getTime();
               }
             }
           }
@@ -270,7 +313,7 @@ export async function syncUserStrava(
   annotate({
     action: { name: "strava.sync.complete", details: { imported } },
   });
-  return imported;
+  return { imported, oldestStart: newestFirst ? oldestStart : null };
 }
 
 /**