diff --git a/src/__tests__/page-walk-guard.test.ts b/src/__tests__/page-walk-guard.test.ts index 3d7f7900f..bd843d704 100644 --- a/src/__tests__/page-walk-guard.test.ts +++ b/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> = { - "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> = { + "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, - ); }); }); diff --git a/src/__tests__/provider-rounding-guard.test.ts b/src/__tests__/provider-rounding-guard.test.ts index 79ff4ba24..10fe0b655 100644 --- a/src/__tests__/provider-rounding-guard.test.ts +++ b/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> = { - "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); }); }); diff --git a/src/lib/google-health/__tests__/mappers.test.ts b/src/lib/google-health/__tests__/mappers.test.ts index c20559389..16d119f23 100644 --- a/src/lib/google-health/__tests__/mappers.test.ts +++ b/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({ diff --git a/src/lib/google-health/__tests__/rollup-request.test.ts b/src/lib/google-health/__tests__/rollup-request.test.ts index bed00a773..770010be4 100644 --- a/src/lib/google-health/__tests__/rollup-request.test.ts +++ b/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( diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index fcf70ecd5..d12006fc1 100644 --- a/src/lib/google-health/client.ts +++ b/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( + `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( - 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( + 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); diff --git a/src/lib/google-health/mappers.ts b/src/lib/google-health/mappers.ts index 194a782ff..0e6bfe02a 100644 --- a/src/lib/google-health/mappers.ts +++ b/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::` — 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, diff --git a/src/lib/integrations/page-walk.ts b/src/lib/integrations/page-walk.ts index df126c8a5..08af1771d 100644 --- a/src/lib/integrations/page-walk.ts +++ b/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"; 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);