Merge branch 'fix/v1437-gh-adopt' into release/v1.43.7

src/__tests__/page-walk-guard.test.ts

@@ -26,20 +26,37 @@ const PROVIDER_DIRS = [
   "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.
+ * Page loops that stay outside `walkPages` on purpose, with how many matches
+ * each file may hold and why. A file with more matches than its allowance has
+ * grown a loop of its own and fails like any other.
+ *
+ * Google Health's list walk (`forEachDataPointPage`) is bounded by progress,
+ * not by a cursor ceiling: it stops after `GOOGLE_HEALTH_NO_PROGRESS_PAGES`
+ * pages in a row that reach no further back (the same page token handed out
+ * again counts toward that, rather than ending the walk at once as
+ * `walkPages` does), streams each page to the writer instead of collecting
+ * the collection (a dense heart-rate history does not fit in the heap), stops
+ * mid-walk on the backfill slice's budget and on a rate limit it cannot wait
+ * out, passes over pages beyond the resume edge without charging them to its
+ * ceiling, and keeps its place in the list for the next slice. Its truncation
+ * reaches the backfill through the client outcome, and the backfill decides
+ * from it with its own sliced verdict (continue, wait for a retry, stuck,
+ * complete; `jobs/google-health-backfill.ts`), which `backfillVerdict`'s
+ * three outcomes cannot carry.
  */
-const PENDING: Readonly<Record<string, string>> = {
-  "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",
+const EXCEPTIONS: Readonly<Record<string, { matches: number; why: string }>> = {
+  "google-health/client.ts": {
+    matches: 1,
+    why: "progress-bounded streaming list walk with slice budget and resume (forEachDataPointPage)",
+  },
 };
 
-/** A page counter compared against a bound, in any of the shapes used. */
+/**
+ * A page counter compared against a bound, in any of the shapes used. A
+ * literal bound of 0 or 1 is a validity check (`page > 0`), not a ceiling.
+ */
 const PAGE_LOOP =
-  /\b(?:pages?|pageCount|pagesRead|page)\s*(?:<|<=|>|>=)\s*(?:[A-Z_][A-Z0-9_]*|maxPages|\d+)\b/;
+  /\b(?:pages?|pageCount|pagesRead|page)(?:\s*-\s*[A-Za-z_]\w*)?\s*(?:<|<=|>|>=)\s*(?:[A-Z_][A-Z0-9_]*|maxPages|[2-9]|[1-9]\d+)\b/g;
 
 function strip(text: string): string {
   return text
@@ -54,19 +71,26 @@ describe("provider clients walk pages through walkPages", () => {
       !f.includes("__tests__") &&
       !f.endsWith(".test.ts"),
   );
-  const offending = (f: string) =>
-    PAGE_LOOP.test(strip(readFileSync(join(LIB, f), "utf8")));
+  const matches = (f: string) =>
+    strip(readFileSync(join(LIB, f), "utf8")).match(PAGE_LOOP)?.length ?? 0;
 
   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("finds no page loop of their own beyond the exceptions", () => {
+    const over = files
+      .map((f) => ({ f, n: matches(f), allowed: EXCEPTIONS[f]?.matches ?? 0 }))
+      .filter(({ n, allowed }) => n > allowed)
+      .map(({ f, n }) => `${f}: ${n}`);
+    expect(over).toEqual([]);
   });
 
-  it("keeps the pending list honest", () => {
-    expect(Object.keys(PENDING).filter((f) => !offending(f))).toEqual([]);
+  it("keeps the exceptions honest", () => {
+    const stale = Object.entries(EXCEPTIONS)
+      .filter(([f, { matches: allowed }]) => matches(f) !== allowed)
+      .map(([f]) => f);
+    expect(stale).toEqual([]);
   });
 
   it("catches the shapes the loops used", () => {
@@ -74,11 +98,16 @@ describe("provider clients walk pages through walkPages", () => {
       "while (pages < WORKOUT_MAX_PAGES) {",
       "if (pageCount > 1000) break;",
       "} while (nextToken && pageCount < maxPages);",
+      "} while (pageToken && pageCount - skipped < maxPages);",
+      "for (let page = 1; page <= maxPages; page++) {",
     ]) {
-      expect(PAGE_LOOP.test(planted), planted).toBe(true);
+      expect(planted.match(PAGE_LOOP), planted).not.toBeNull();
+    }
+    for (const fine of [
+      "const walk = await walkPages(label, 12, f);",
+      "r.page > 0 ? r.page : undefined",
+    ]) {
+      expect(fine.match(PAGE_LOOP), fine).toBeNull();
     }
-    expect(PAGE_LOOP.test("const walk = await walkPages(label, 12, f);")).toBe(
-      false,
-    );
   });
 });

src/__tests__/provider-rounding-guard.test.ts

@@ -26,17 +26,7 @@ const PROVIDER_DIRS = [
   "whoop",
   "withings",
 ];
-/**
- * Files that still round on their own, with the reason. Google Health's
- * mapper is the one left: it moves to `importValue` with the next change to
- * that directory.
- */
-const PENDING: Readonly<Record<string, string>> = {
-  "google-health/mappers.ts":
-    "moves to importValue with the next Google Health change",
-};
-
-const ROUNDING = /\.toFixed\s*\(|\bfunction\s+round2\b/;
+const ROUNDING = /\.toFixed\s*\(|\b(?:function|const|let)\s+round2\b/;
 
 function strip(text: string): string {
   return text
@@ -56,25 +46,17 @@ describe("provider clients round through importValue", () => {
     expect(files.length).toBeGreaterThanOrEqual(60);
   });
 
-  it("finds no rounding of their own outside the pending list", () => {
-    const offenders = files.filter(
-      (f) =>
-        !(f in PENDING) &&
-        ROUNDING.test(strip(readFileSync(join(LIB, f), "utf8"))),
+  it("finds no rounding of their own", () => {
+    const offenders = files.filter((f) =>
+      ROUNDING.test(strip(readFileSync(join(LIB, f), "utf8"))),
     );
     expect(offenders).toEqual([]);
   });
 
-  it("keeps the pending list honest", () => {
-    const stale = Object.keys(PENDING).filter(
-      (f) => !ROUNDING.test(strip(readFileSync(join(LIB, f), "utf8"))),
-    );
-    expect(stale).toEqual([]);
-  });
-
   it("catches a planted copy", () => {
     expect(ROUNDING.test("const v = parseFloat(n.toFixed(2));")).toBe(true);
     expect(ROUNDING.test("function round2(n: number) {}")).toBe(true);
+    expect(ROUNDING.test("const round2 = (n: number) => n;")).toBe(true);
     expect(ROUNDING.test("value: importValue(n)")).toBe(false);
   });
 });

src/lib/google-health/__tests__/mappers.test.ts

@@ -73,6 +73,43 @@ describe("mapWeight — spot sample", () => {
   });
 });
 
+describe("imported values round half away from zero (importValue)", () => {
+  // Each of these sits on a half at the third decimal whose double lies just
+  // below it, where `toFixed(2)` rounded down.
+  it("rounds a gram-precise weight up on the half", () => {
+    const [r] = mapWeight({
+      weight: {
+        weightGrams: 72345,
+        sampleTime: { physicalTime: "2026-06-01T08:00:00.000Z" },
+      },
+    });
+    expect(r!.value).toBe(72.35);
+  });
+
+  it("rounds a daily distance total up on the half", () => {
+    const [r] = mapDistance({
+      civilStartTime: { date: { year: 2026, month: 6, day: 1 } },
+      distance: { millimetersSum: "1000005" },
+    });
+    expect(r!.value).toBe(1000.01);
+  });
+
+  it("rounds a session distance up on the half", () => {
+    const w = mapWorkout({
+      name: "users/me/dataTypes/exercise/dataPoints/half",
+      exercise: {
+        exerciseType: "RUNNING",
+        interval: {
+          startTime: "2026-06-03T07:00:00.000Z",
+          endTime: "2026-06-03T07:45:00.000Z",
+        },
+        metricsSummary: { distanceMillimeters: "8000025" },
+      },
+    });
+    expect(w!.totalDistanceM).toBe(8000.03);
+  });
+});
+
 describe("mapBodyFat — camelCase union key", () => {
   it("reads the percentage under the camelCase bodyFat key", () => {
     const rows = mapBodyFat({

src/lib/google-health/__tests__/rollup-request.test.ts

@@ -26,6 +26,7 @@ import {
   buildDailyRollUpBody,
   extractGoogleApiErrorDetail,
   fetchDailyRollUp,
+  runWithGoogleHealthClientOutcome,
   type GoogleHealthRollupShape,
 } from "../client";
 import { GoogleHealthApiError } from "../response-classifier";
@@ -219,6 +220,82 @@ describe("fetchDailyRollUp — chunk walk + conservative fallback", () => {
   });
 });
 
+describe("fetchDailyRollUp — pages within a chunk (walkPages)", () => {
+  const oneChunk = { start: new Date("2026-06-20T00:00:00Z") };
+
+  it("follows nextPageToken to the end of a chunk", async () => {
+    safeFetchMock
+      .mockResolvedValueOnce(
+        jsonResponse(200, {
+          rollupDataPoints: [{ steps: { countSum: "1" } }],
+          nextPageToken: "p2",
+        }),
+      )
+      .mockResolvedValueOnce(
+        jsonResponse(200, {
+          rollupDataPoints: [{ steps: { countSum: "2" } }],
+          nextPageToken: "",
+        }),
+      );
+    const { result, outcome } = await runWithGoogleHealthClientOutcome(() =>
+      fetchDailyRollUp(
+        GOOGLE_HEALTH_DATA_TYPES.steps,
+        "token",
+        "fetchSteps",
+        oneChunk,
+      ),
+    );
+    expect(result).toHaveLength(2);
+    expect(safeFetchMock).toHaveBeenCalledTimes(2);
+    expect(
+      JSON.parse((safeFetchMock.mock.calls[1]![1] as { body: string }).body)
+        .pageToken,
+    ).toBe("p2");
+    expect(outcome).toMatchObject({ pages: 2, truncated: false });
+  });
+
+  it("stops a chunk whose page token comes back again and reports it truncated", async () => {
+    safeFetchMock.mockResolvedValue(
+      jsonResponse(200, {
+        rollupDataPoints: [{ steps: { countSum: "1" } }],
+        nextPageToken: "again",
+      }),
+    );
+    const { result, outcome } = await runWithGoogleHealthClientOutcome(() =>
+      fetchDailyRollUp(
+        GOOGLE_HEALTH_DATA_TYPES.steps,
+        "token",
+        "fetchSteps",
+        oneChunk,
+      ),
+    );
+    // The first page hands out "again", the second hands it out again.
+    expect(safeFetchMock).toHaveBeenCalledTimes(2);
+    expect(result).toHaveLength(2);
+    expect(outcome).toMatchObject({
+      truncated: true,
+      reasonCode: "existing_page_limit",
+    });
+  });
+
+  it("stops a chunk at its page ceiling and reports it truncated", async () => {
+    let n = 0;
+    safeFetchMock.mockImplementation(async () =>
+      jsonResponse(200, { rollupDataPoints: [], nextPageToken: `p${++n}` }),
+    );
+    const { outcome } = await runWithGoogleHealthClientOutcome(() =>
+      fetchDailyRollUp(
+        GOOGLE_HEALTH_DATA_TYPES.steps,
+        "token",
+        "fetchSteps",
+        oneChunk,
+      ),
+    );
+    expect(safeFetchMock).toHaveBeenCalledTimes(100);
+    expect(outcome).toMatchObject({ pages: 100, truncated: true });
+  });
+});
+
 describe("extractGoogleApiErrorDetail — AIP-193 field violations", () => {
   it("keeps the plain STATUS: message shape without details", () => {
     expect(

src/lib/google-health/client.ts

@@ -55,6 +55,7 @@ import {
   googleHealthWalkProgress,
   noteGoogleHealthWalkPage,
 } from "./backfill-bounds";
+import { walkPages } from "@/lib/integrations/page-walk";
 
 export * from "./mappers";
 
@@ -1605,6 +1606,13 @@ export const GOOGLE_HEALTH_ROLLUP_FALLBACK_RANGE_DAYS = 14;
  */
 export const GOOGLE_HEALTH_ROLLUP_BACKFILL_DAYS = 5 * 365;
 
+/**
+ * Page ceiling for one dailyRollUp chunk. A chunk of at most 90 daily windows
+ * fits on one page of the documented default size, so a chunk that pages at
+ * all is unusual and one that reaches this is a runaway token.
+ */
+const GOOGLE_HEALTH_ROLLUP_MAX_PAGES = 100;
+
 /** A civil calendar date (1-based month), the dailyRollUp range unit. */
 export interface GoogleHealthCivilDate {
   year: number;
@@ -1778,56 +1786,61 @@ export async function fetchDailyRollUp(
         truncated = true;
         break;
       }
-      let pageToken: string | null | undefined;
-      let pageCount = 0;
-      do {
-        const body = buildDailyRollUpBody(chunk, pageToken ?? undefined);
+      const index = chunkIndex;
+      const chunkWalk = await walkPages<string, GoogleHealthRollupPoint>(
+        `google_health.rollup.${dataType.path}`,
+        GOOGLE_HEALTH_ROLLUP_MAX_PAGES,
+        async (pageToken, pageCount) => {
+          const body = buildDailyRollUpBody(chunk, pageToken);
 
-        requestCount += 1;
-        const { json, verdict, apiErrorDetail } =
-          await googleHealthDataRequest<GoogleHealthRollupPage>(
-            accessToken,
-            `${verb}(chunk=${chunkIndex},page=${pageCount})`,
-            (token) =>
-              safeFetch(
-                `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`,
-                {
-                  method: "POST",
-                  headers: {
-                    Authorization: `Bearer ${token}`,
-                    "Content-Type": "application/json",
+          requestCount += 1;
+          const { json, verdict, apiErrorDetail } =
+            await googleHealthDataRequest<GoogleHealthRollupPage>(
+              accessToken,
+              `${verb}(chunk=${index},page=${pageCount})`,
+              (token) =>
+                safeFetch(
+                  `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp`,
+                  {
+                    method: "POST",
+                    headers: {
+                      Authorization: `Bearer ${token}`,
+                      "Content-Type": "application/json",
+                    },
+                    body: JSON.stringify(body),
                   },
-                  body: JSON.stringify(body),
-                },
-              ),
-          );
-        if (verdict.classification !== "success") {
-          throw new GoogleHealthApiError({
-            verb,
-            classification: verdict.classification,
-            httpStatus: verdict.httpStatus,
-            reason: verdict.reason,
-            upstreamError: apiErrorDetail,
-          });
-        }
+                ),
+            );
+          if (verdict.classification !== "success") {
+            throw new GoogleHealthApiError({
+              verb,
+              classification: verdict.classification,
+              httpStatus: verdict.httpStatus,
+              reason: verdict.reason,
+              upstreamError: apiErrorDetail,
+            });
+          }
 
-        // Surface the raw envelope key names of the very first page (names
-        // only, never values) — see `RollupQuery.onEnvelopeKeys`.
-        if (
-          chunkIndex === 0 &&
-          pageCount === 0 &&
-          json &&
-          typeof json === "object"
-        ) {
-          query.onEnvelopeKeys?.(Object.keys(json));
-        }
-        for (const p of json?.rollupDataPoints ?? []) points.push(p);
-        pageToken = json?.nextPageToken ?? null;
-        pageCount += 1;
-        noteSlicePage();
-      } while (pageToken && pageCount < 100);
-      totalPages += pageCount;
-      truncated ||= Boolean(pageToken);
+          // Surface the raw envelope key names of the very first page (names
+          // only, never values) — see `RollupQuery.onEnvelopeKeys`.
+          if (
+            index === 0 &&
+            pageCount === 0 &&
+            json &&
+            typeof json === "object"
+          ) {
+            query.onEnvelopeKeys?.(Object.keys(json));
+          }
+          noteSlicePage();
+          return {
+            items: json?.rollupDataPoints ?? [],
+            next: json?.nextPageToken || null,
+          };
+        },
+      );
+      for (const p of chunkWalk.items) points.push(p);
+      totalPages += chunkWalk.pages;
+      truncated ||= chunkWalk.truncated;
       chunkIndex += 1;
     }
     notePagination(totalPages, points.length, truncated);

src/lib/google-health/mappers.ts

@@ -14,6 +14,7 @@
  * this file performs I/O — everything is synchronous and deterministic given a
  * payload (+ optional user timezone).
  */
+import { importValue } from "@/lib/integrations/import-value";
 import { canonicalDailyTimestamp } from "@/lib/measurements/consolidation-tz";
 import { zonedWallClockToUtc } from "@/lib/tz/wall-clock";
 
@@ -217,10 +218,6 @@ export interface GoogleHealthRollupPoint {
 /** Millimetres → centimetres (Google `height.heightMillimeters` → `User.heightCm`). */
 const MM_TO_CM = 0.1;
 
-function round2(n: number): number {
-  return parseFloat(n.toFixed(2));
-}
-
 /**
  * Coerce a raw payload value to a finite number, or null. proto3 int64 fields
  * arrive as JSON strings (`"12345"`); `Number()` handles both int and decimal
@@ -445,7 +442,7 @@ function mapSimple(
   return [
     {
       type: spec.type,
-      value: round2(value),
+      value: importValue(value),
       unit: spec.unit,
       measuredAt: resolveMeasuredAt(point, dataType, new Date(), tz),
       fieldTag: `${externalAnchor(point, dataType)}:${spec.fieldTag}`,
@@ -678,7 +675,7 @@ export function mapHeight(
   if (mm === null) return null;
   const t = readPath(point, `${dt.key}.sampleTime.physicalTime`);
   const sampledAt = typeof t === "string" ? parseLocalInstant(t) : null;
-  return { cm: round2(mm * MM_TO_CM), sampledAt };
+  return { cm: importValue(mm * MM_TO_CM), sampledAt };
 }
 
 // ─── Activity mappers: daily roll-up totals ────────────────────
@@ -732,7 +729,7 @@ function mapDailyRollup(
   return [
     {
       type: spec.type,
-      value: round2(value),
+      value: importValue(value),
       unit: spec.unit,
       measuredAt: canonicalDailyTimestamp(dayKey, tz),
       // `stats:<type-tag>:<YYYY-MM-DD>` — the sync layer reads `cumulativeDaily`
@@ -833,7 +830,7 @@ export function mapVo2Max(
   return [
     {
       type: "VO2_MAX",
-      value: round2(value),
+      value: importValue(value),
       unit: "mL/(kg·min)",
       measuredAt,
       fieldTag: `vo2_max:${externalAnchor(point, dt)}`,
@@ -1035,7 +1032,7 @@ export function mapSleepSessionDetailed(
     if (!windowEnd || end > windowEnd) windowEnd = end;
     rows.push({
       type: "SLEEP_DURATION",
-      value: round2(mins),
+      value: importValue(mins),
       unit: "minutes",
       measuredAt: end,
       fieldTag: `${anchor}:sleep:${start.toISOString()}`,
@@ -1287,7 +1284,9 @@ export function mapWorkout(
     durationSec,
     totalEnergyKcal: energyKcal !== null ? Math.round(energyKcal) : null,
     totalDistanceM:
-      distanceMm !== null && distanceMm >= 0 ? round2(distanceMm / 1000) : null,
+      distanceMm !== null && distanceMm >= 0
+        ? importValue(distanceMm / 1000)
+        : null,
     avgHeartRate: avgHr !== null && avgHr > 0 ? Math.round(avgHr) : null,
     // metricsSummary carries no maximum/minimum heart rate.
     maxHeartRate: null,

src/lib/integrations/page-walk.ts

@@ -19,11 +19,14 @@
  *   - 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.
+ * Google Health's dailyRollUp chunks walk through here. Its `dataPoints.list`
+ * walk (`forEachDataPointPage`) stays outside on purpose: it is bounded by
+ * progress in time rather than by a cursor ceiling (`google-health/
+ * backfill-bounds.ts`), streams pages instead of collecting them, stops on a
+ * backfill slice's budget and keeps its place for the next slice, and its
+ * backfill decides with a sliced verdict these three outcomes cannot carry.
+ * `src/__tests__/page-walk-guard.test.ts` keeps hand-rolled page loops out of
+ * the provider directories and names that one exception with its reasons.
  */
 import { AsyncLocalStorage } from "node:async_hooks";
 

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 },

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: <T>(err: T) => err,
 }));

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: <T>(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<typeof import("../client")>()),
+  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();
+  });
+});

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<number, StravaWorkoutRow>(
+      "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);