diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b83046cc..6b6c7e4c7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,34 @@ ## [Unreleased] +## [1.32.18] — 2026-07-24 + +When a Fitbit token was revoked or expired, the hourly sync noticed the failure +but then stamped the same cycle as a success. The connection flipped back to +connected, the sync position moved forward, and after you reconnected the +catch-up window could no longer reach the days that were missed. So the data +was skipped for good, and in the meantime the failing token was retried every +hour. + +- **A revoked Fitbit token no longer reports a successful sync.** A failed token + refresh now fails the whole cycle. The connection stays parked until you + reconnect, the sync position holds instead of advancing past real data, and + the once-per-failure alert stops repeating every hour. +- **Data after a reconnect is no longer skipped.** Because the position is held + back on a failed cycle, the next successful sync still covers the days that + were missed rather than starting after them. +- **A write failure during a sync holds its place too.** If rows were fetched + but could not be saved, the cycle keeps its position so the next run fetches + them again instead of letting them fall outside the catch-up window. +- **An interrupted history backfill no longer marks itself complete.** A partial + first-time import is retried until a full pass finishes, so the one-year + history walk fills in rather than freezing halfway. + +If your Fitbit history already has a gap from before this release: reconnect +Fitbit if prompted, then open Settings, Integrations, Fitbit and run a full +sync. It re-walks the last year and fills the gap in place. Gaps older than a +year are outside the backfill window and cannot be recovered. + ## [1.32.17] — 2026-07-24 The mood correlations on the Insights page could line up a day off. Your mood diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 8eae52a6f..6f97b1e17 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -1,7 +1,7 @@ openapi: 3.1.0 info: title: HealthLog API - version: 1.32.17 + version: 1.32.18 description: >- Self-hosted personal-health-tracking PWA — public API surface for the iOS native client and external ingest. diff --git a/package.json b/package.json index 081084681..cb098d507 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "healthlog", - "version": "1.32.17", + "version": "1.32.18", "description": "Self-hosted personal-health-tracking PWA with Withings integration, AI insights, and doctor-report PDF export.", "license": "PolyForm-Noncommercial-1.0.0", "homepage": "https://healthlog.dev", diff --git a/public/sw.js b/public/sw.js index 845fd4148..0ba82d22d 100644 --- a/public/sw.js +++ b/public/sw.js @@ -36,7 +36,7 @@ try { // v1.4.38.4 → v1.4.42. Do not hand-edit; bump `package.json` and rebuild. const CACHE_VERSION = (typeof self !== "undefined" && self.__APP_VERSION__) || - /* @sw-version-fallback */ "v1.32.17"; + /* @sw-version-fallback */ "v1.32.18"; const STATIC_CACHE = `healthlog-static-${CACHE_VERSION}`; const PAGE_CACHE = `healthlog-pages-${CACHE_VERSION}`; // v1.18.6 — read-only data cache for a curated allowlist of safe GET `/api/*` diff --git a/src/app/api/fitbit/sync/route.ts b/src/app/api/fitbit/sync/route.ts index 5197ef8e8..2f7641fc7 100644 --- a/src/app/api/fitbit/sync/route.ts +++ b/src/app/api/fitbit/sync/route.ts @@ -58,6 +58,6 @@ export const POST = apiHandler(async (request: NextRequest) => { } } - const imported = await syncUserFitbit(user.id, { fullSync }); + const { imported } = await syncUserFitbit(user.id, { fullSync }); return apiSuccess({ imported, fullSync }); }); diff --git a/src/lib/fitbit/__tests__/sync-dead-token.test.ts b/src/lib/fitbit/__tests__/sync-dead-token.test.ts new file mode 100644 index 000000000..ece6ddf3a --- /dev/null +++ b/src/lib/fitbit/__tests__/sync-dead-token.test.ts @@ -0,0 +1,243 @@ +/** + * Pins the dead-token verdict for Fitbit — the parity mirror of + * `google-health/__tests__/sync-failsoft.test.ts`'s "getValidToken — + * dead-token cycle verdict" block. + * + * `getValidToken`'s three null-return paths (credentials missing, refresh + * failure, and the Fitbit-only vanished-connection CAS loss) must register on + * the cycle's hard-fail ledger, and a cycle whose token is dead must NOT stamp + * `markSynced` / `recordSyncSuccess`. Otherwise every resource returns 0 + * without failures, the cycle reads as clean, `recordSyncSuccess` un-parks + * `error_reauth`, and the hourly cohort hammers the dead refresh token forever + * while `lastSyncedAt` advances past real data. + * + * Fitbit has no one-shot leaf job, so no `runWith…HardFailLedger` wrapper is + * ported; the tests scope the ledger through the exported `hardFailStorage.run` + * directly. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { + recordSyncFailureMock, + recordSyncSuccessMock, + prismaMock, + getUserFitbitCredentialsMock, + refreshAccessTokenMock, + resourceFake, +} = vi.hoisted(() => ({ + recordSyncFailureMock: vi.fn(async () => {}), + recordSyncSuccessMock: vi.fn(async () => {}), + prismaMock: { + fitbitConnection: { + findUnique: vi.fn(), + update: vi.fn(async () => ({})), + updateMany: vi.fn(async () => ({ count: 1 })), + }, + }, + getUserFitbitCredentialsMock: vi.fn(async (): Promise => null), + refreshAccessTokenMock: vi.fn(), + // The dead-token cycle test drives the real `syncUserFitbit`; each resource + // module is mocked to do exactly what the real ones do first — resolve the + // token — so the ledger registration happens inside the cycle's ALS scope. + resourceFake: async (userId: string): Promise => { + const { getValidToken } = await import("../sync-core"); + return (await getValidToken(userId)) ? 1 : 0; + }, +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/crypto", () => ({ + encrypt: (s: string) => `enc(${s})`, + decrypt: (s: string) => s.replace(/^enc\(|\)$/g, ""), +})); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: vi.fn(async () => false), + recordSyncFailure: recordSyncFailureMock, + recordSyncSuccess: recordSyncSuccessMock, +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: vi.fn(() => []), + recomputeBucketsForMeasurement: vi.fn(async () => {}), + recomputeUserRollups: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); +vi.mock("@/lib/logging/context", () => ({ + getEvent: () => null, + annotate: () => {}, +})); +vi.mock("../credentials", () => ({ + getUserFitbitCredentials: getUserFitbitCredentialsMock, +})); +vi.mock("../client", () => ({ refreshAccessToken: refreshAccessTokenMock })); + +vi.mock("../sync-metrics", () => ({ syncUserMetrics: resourceFake })); +vi.mock("../sync-activity", () => ({ syncUserActivity: resourceFake })); +vi.mock("../sync-sleep", () => ({ syncUserSleep: resourceFake })); +vi.mock("../sync-workout", () => ({ syncUserWorkout: resourceFake })); + +import { + FITBIT_TOKEN_HARD_FAIL, + getValidToken, + hardFailStorage, +} from "../sync-core"; +import { syncUserFitbit } from "../sync"; +import { FitbitApiError } from "../response-classifier"; + +/** Run `fn` inside a fresh hard-fail ledger scope and return its failures. */ +async function withLedger( + fn: () => Promise, +): Promise<{ result: T; failures: string[] }> { + const tracker = { failures: [] as string[] }; + const result = await hardFailStorage.run(tracker, fn); + return { result, failures: tracker.failures }; +} + +/** A connection whose access token is inside the 5-min refresh buffer. */ +const EXPIRED_CONNECTION = { + id: "conn-1", + userId: "user-1", + fitbitUserId: "fb-user-1", + accessToken: "enc(access)", + refreshToken: "enc(refresh)", + tokenExpiresAt: new Date(Date.now() - 60_000), + lastSyncedAt: new Date("2026-07-01T00:00:00.000Z"), +}; + +const CREDS = { clientId: "id", clientSecret: "secret" }; + +beforeEach(() => { + recordSyncFailureMock.mockClear(); + recordSyncSuccessMock.mockClear(); + prismaMock.fitbitConnection.findUnique.mockReset(); + prismaMock.fitbitConnection.update.mockReset().mockResolvedValue({} as never); + prismaMock.fitbitConnection.updateMany + .mockReset() + .mockResolvedValue({ count: 1 } as never); + getUserFitbitCredentialsMock.mockReset().mockResolvedValue(null); + refreshAccessTokenMock.mockReset(); +}); + +describe("getValidToken — dead-token cycle verdict", () => { + it("registers the credentials-missing path on the hard-fail ledger", async () => { + prismaMock.fitbitConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([FITBIT_TOKEN_HARD_FAIL]); + expect(recordSyncFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ + userId: "user-1", + integration: "fitbit", + kind: "reauth_required", + errorCode: "credentials_missing", + }), + ); + }); + + it("registers a refresh failure on the hard-fail ledger", async () => { + prismaMock.fitbitConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserFitbitCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockRejectedValue( + new FitbitApiError({ + verb: "refreshToken", + classification: "reauth_required", + httpStatus: 401, + reason: "invalid_grant", + }), + ); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([FITBIT_TOKEN_HARD_FAIL]); + expect(recordSyncFailureMock).toHaveBeenCalledWith( + expect.objectContaining({ kind: "reauth_required" }), + ); + }); + + it("registers a vanished-connection CAS loss on the ledger", async () => { + // Fitbit-only path (no Google twin): the CAS update touches zero rows AND + // the peer re-read finds no row — the connection was deleted mid-flight. + prismaMock.fitbitConnection.findUnique + .mockResolvedValueOnce(EXPIRED_CONNECTION as never) // initial read + .mockResolvedValueOnce(null as never); // peer re-read: row gone + getUserFitbitCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockResolvedValue({ + access_token: "new-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + prismaMock.fitbitConnection.updateMany.mockResolvedValue({ + count: 0, + } as never); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result).toBeNull(); + expect(failures).toEqual([FITBIT_TOKEN_HARD_FAIL]); + }); + + it("a successful refresh leaves the ledger empty", async () => { + // Guard against over-failing: a healthy refresh must not touch the ledger. + prismaMock.fitbitConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserFitbitCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockResolvedValue({ + access_token: "new-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + prismaMock.fitbitConnection.updateMany.mockResolvedValue({ + count: 1, + } as never); + + const { result, failures } = await withLedger(() => + getValidToken("user-1"), + ); + + expect(result?.accessToken).toBe("new-access"); + expect(failures).toEqual([]); + }); + + it("a dead-token cycle fails the verdict and never stamps markSynced / recordSyncSuccess", async () => { + // Every resource resolves the token, the refresh fails each time — before + // the ledger registration the cycle read as CLEAN (all resources returned + // 0 without failures), stamped the watermark, and un-parked error_reauth. + prismaMock.fitbitConnection.findUnique.mockResolvedValue( + EXPIRED_CONNECTION as never, + ); + getUserFitbitCredentialsMock.mockResolvedValue(CREDS); + refreshAccessTokenMock.mockRejectedValue( + new FitbitApiError({ + verb: "refreshToken", + classification: "reauth_required", + httpStatus: 401, + reason: "invalid_grant", + }), + ); + + const res = await syncUserFitbit("user-1"); + + expect(res).toEqual({ imported: 0, failed: true }); + expect(recordSyncSuccessMock).not.toHaveBeenCalled(); + // No connection write may carry the watermark stamp. + for (const call of prismaMock.fitbitConnection.update.mock.calls) { + const arg = (call as unknown[])[0] as { data: Record }; + expect(arg.data).not.toHaveProperty("lastSyncedAt"); + } + }); +}); diff --git a/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts index 26c1ae2a9..6fd5448f2 100644 --- a/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts +++ b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts @@ -92,6 +92,7 @@ vi.mock("../sync-workout", () => ({ import { handleCollectionFetchError, + noteHardFailure, upsertFitbitMeasurements, } from "../sync-core"; import { syncUserFitbit } from "../sync"; @@ -139,7 +140,7 @@ describe("syncUserFitbit — all-403 looks-healthy guard", () => { syncUserSleep.mockImplementation(softSkip403); syncUserWorkout.mockImplementation(softSkip403); - const total = await syncUserFitbit("user1"); + const { imported: total } = await syncUserFitbit("user1"); expect(total).toBe(0); expect(recordSyncSuccess).not.toHaveBeenCalled(); @@ -155,7 +156,7 @@ describe("syncUserFitbit — all-403 looks-healthy guard", () => { syncUserSleep.mockImplementation(softSkip403); syncUserWorkout.mockImplementation(softSkip403); - const total = await syncUserFitbit("user1"); + const { imported: total } = await syncUserFitbit("user1"); expect(total).toBe(5); expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "fitbit"); @@ -165,16 +166,16 @@ describe("syncUserFitbit — all-403 looks-healthy guard", () => { syncUserMetrics.mockResolvedValue(3); syncUserActivity.mockResolvedValue(2); - const total = await syncUserFitbit("user1"); + const res = await syncUserFitbit("user1"); - expect(total).toBe(5); + expect(res).toEqual({ imported: 5, failed: false }); expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "fitbit"); }); it("stamps success on a quiet cycle (no soft-skip, nothing new)", async () => { syncUserMetrics.mockResolvedValue(0); - const total = await syncUserFitbit("user1"); + const { imported: total } = await syncUserFitbit("user1"); expect(total).toBe(0); // No collection was 403'd → this is a genuine "nothing changed" tick. @@ -184,7 +185,7 @@ describe("syncUserFitbit — all-403 looks-healthy guard", () => { it("short-circuits without stamping success when parked at error_reauth", async () => { isReauthRequired.mockResolvedValue(true); - const total = await syncUserFitbit("user1"); + const { imported: total } = await syncUserFitbit("user1"); expect(total).toBe(0); expect(recordSyncSuccess).not.toHaveBeenCalled(); @@ -219,7 +220,7 @@ describe("syncUserFitbit — hard-fail ledger (F-SYNC-4)", () => { syncUserWorkout.mockResolvedValue(0); // Resolves (no rethrow) with the siblings' imports. - const total = await syncUserFitbit("user1"); + const { imported: total } = await syncUserFitbit("user1"); expect(total).toBe(3); // Every sibling resource still ran despite the metrics hard failure. @@ -243,6 +244,27 @@ describe("syncUserFitbit — hard-fail ledger (F-SYNC-4)", () => { expect(recordSyncFailure).toHaveBeenCalled(); }); + it("a write-failure ledger entry holds the watermark and fails the verdict", async () => { + // A resource that fetched rows but failed to write them notes the write + // label on the ledger and returns 0. The F-SYNC-4 test above covers a + // collection label; this pins the WRITE label (F-2) through the same gate. + syncUserMetrics.mockImplementation(async () => { + noteHardFailure("measurements:create"); + return 0; + }); + // A sibling imports so the run is not degenerate — only the ledger keeps it + // honest. + syncUserActivity.mockResolvedValue(4); + + const { imported, failed } = await syncUserFitbit("user1"); + + expect(imported).toBe(4); + expect(failed).toBe(true); + // The write failure holds the watermark: no stamp, no success. + expect(update).not.toHaveBeenCalled(); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + }); + it("a failed natural-key rescue probe notes the ledger: no success stamp, no watermark", async () => { // The rescue probe (second measurement.findMany inside the upsert) dies. // The rows still go to createMany (skipDuplicates absorbs a twin @@ -270,7 +292,7 @@ describe("syncUserFitbit — hard-fail ledger (F-SYNC-4)", () => { }); // Resolves (no rethrow) — the insert was still attempted. - const total = await syncUserFitbit("user1"); + const { imported: total } = await syncUserFitbit("user1"); expect(total).toBe(1); // The ledger flips anyFailed → success is NOT stamped and the watermark is @@ -424,7 +446,7 @@ describe("syncUserFitbit — backfill collapses rollup recompute to one pass", ( it("returns 0 without running resources when the connection is gone", async () => { findUnique.mockResolvedValue(null); - const total = await syncUserFitbit("user1"); + const { imported: total } = await syncUserFitbit("user1"); expect(total).toBe(0); expect(syncUserMetrics).not.toHaveBeenCalled(); diff --git a/src/lib/fitbit/__tests__/sync.test.ts b/src/lib/fitbit/__tests__/sync.test.ts index 25f3e5ef9..9974e0b50 100644 --- a/src/lib/fitbit/__tests__/sync.test.ts +++ b/src/lib/fitbit/__tests__/sync.test.ts @@ -77,6 +77,7 @@ vi.mock("@/lib/integrations/status", () => ({ import { classificationToFailureKind, getValidToken, + hardFailStorage, upsertFitbitMeasurements, } from "../sync-core"; @@ -430,6 +431,97 @@ describe("upsertFitbitMeasurements — batched write, tombstones resurrect", () }); }); +describe("upsertFitbitMeasurements — write-catch ledger (F-2)", () => { + it("a createMany failure notes measurements:create on the ledger and does not throw", async () => { + const measuredAt = new Date("2026-05-10T07:00:00.000Z"); + // Probe: one live row (its externalId) → update branch; the other key is + // fresh → create branch. Second findMany is the natural-key rescue probe. + prismaMock.measurement.findMany + .mockResolvedValueOnce([ + { id: "m-live", type: "WEIGHT", externalId: "live-key" }, + ]) + .mockResolvedValueOnce([]); // natural-key rescue: no twin + prismaMock.measurement.createManyAndReturn.mockRejectedValueOnce( + new Error("createMany exploded"), + ); + + const tracker = { failures: [] as string[] }; + const { imported } = await hardFailStorage.run(tracker, () => + upsertFitbitMeasurements("user1", [ + { + type: "WEIGHT", + value: 1, + unit: "kg", + measuredAt, + externalId: "live-key", + }, + { + type: "WEIGHT", + value: 2, + unit: "kg", + measuredAt, + externalId: "fresh-key", + }, + ]), + ); + + // No rethrow: the sibling live row still wrote through the update branch. + expect(prismaMock.measurement.update).toHaveBeenCalledTimes(1); + expect(imported).toBe(1); + // The failed create is recorded so the cycle's watermark holds. + expect(tracker.failures).toContain("measurements:create"); + }); + + it("an update failure notes measurements:update on the ledger and does not throw", async () => { + const measuredAt = new Date("2026-05-10T07:00:00.000Z"); + // Two live rows (both updates) + one fresh row (create). The first update + // throws; the loop must continue to the second, and the create still runs. + prismaMock.measurement.findMany + .mockResolvedValueOnce([ + { id: "m-live-1", type: "WEIGHT", externalId: "live-1" }, + { id: "m-live-2", type: "WEIGHT", externalId: "live-2" }, + ]) + .mockResolvedValueOnce([]); // natural-key rescue: no twin + prismaMock.measurement.update + .mockRejectedValueOnce(new Error("update exploded")) + .mockResolvedValueOnce({}); + + const tracker = { failures: [] as string[] }; + const { imported } = await hardFailStorage.run(tracker, () => + upsertFitbitMeasurements("user1", [ + { + type: "WEIGHT", + value: 1, + unit: "kg", + measuredAt, + externalId: "live-1", + }, + { + type: "WEIGHT", + value: 2, + unit: "kg", + measuredAt, + externalId: "live-2", + }, + { + type: "WEIGHT", + value: 3, + unit: "kg", + measuredAt, + externalId: "fresh-3", + }, + ]), + ); + + // No rethrow: the sibling update ran and the fresh create still wrote. + expect(prismaMock.measurement.update).toHaveBeenCalledTimes(2); + expect(prismaMock.measurement.createManyAndReturn).toHaveBeenCalledTimes(1); + // create (1) + the one surviving update (1) landed. + expect(imported).toBe(2); + expect(tracker.failures).toContain("measurements:update"); + }); +}); + describe("classificationToFailureKind", () => { it("maps each classification onto a FailureKind", () => { expect(classificationToFailureKind("reauth_required")).toBe( diff --git a/src/lib/fitbit/sync-core.ts b/src/lib/fitbit/sync-core.ts index 2f1bcc721..d6816ef8e 100644 --- a/src/lib/fitbit/sync-core.ts +++ b/src/lib/fitbit/sync-core.ts @@ -98,6 +98,12 @@ export async function getValidToken( message: "Fitbit credentials missing — token refresh skipped", errorCode: "credentials_missing", }); + // A dead token means every resource "succeeds" with 0 rows — without a + // ledger entry the cycle would read as clean, stamp `markSynced`, and + // `recordSyncSuccess` would flip a parked connection back to connected + // (an hourly invalid-refresh hammer with an advancing watermark). Fail + // the cycle's verdict so the park holds. + noteHardFailure(FITBIT_TOKEN_HARD_FAIL); return null; } @@ -143,7 +149,16 @@ export async function getValidToken( }, ); - if (!persistedAccessToken) return null; + if (!persistedAccessToken) { + // `persistRotatedToken` returns null ONLY when the connection row + // vanished mid-flight (deleted between the read and the CAS re-read) — + // the normal lost-race path reuses the peer's rotated token above. A + // vanished-row cycle fetches nothing; register on the ledger so the + // same verdict rule as the refresh branch holds and the cycle does not + // stamp success over a connection that no longer exists. + noteHardFailure(FITBIT_TOKEN_HARD_FAIL); + return null; + } return { accessToken: persistedAccessToken, @@ -157,6 +172,14 @@ export async function getValidToken( `Fitbit token refresh failed for user ${userId}: ${err}`, ); await recordFitbitSyncFailure(userId, err); + // Same verdict rule as the credentials-missing branch above: a refresh + // failure must fail the cycle, or the all-zeros run stamps success, + // un-parks `error_reauth`, and the hourly cohort retries the dead refresh + // token forever while `lastSyncedAt` advances past real data. Fitbit's + // reconnect CTA is driven by the status-ledger park (`error_reauth`) that + // `recordFitbitSyncFailure` sets — there is no `needsReauth` column to + // flag (Google's connection-flag write is deliberately not ported). + noteHardFailure(FITBIT_TOKEN_HARD_FAIL); return null; } } @@ -223,6 +246,27 @@ export interface HardFailTracker { } export const hardFailStorage = new AsyncLocalStorage(); +/** + * Ledger label for a dead-token failure (`getValidToken` returning null on a + * missing-credentials, failed-refresh, or vanished-connection path). Distinct + * from a fetch/write label so a future one-shot leaf job could tell "the token + * is dead" (park, don't retry) apart from "a fetch/write hard-failed" (retry). + * Ported from Google Health's `GOOGLE_HEALTH_TOKEN_HARD_FAIL`. + */ +export const FITBIT_TOKEN_HARD_FAIL = "token"; + +/** + * Record a hard failure on the ambient per-cycle ledger, if one is in scope. + * No-op outside a ledger scope (e.g. a direct call from a REPL script). Used by + * `getValidToken`'s dead-token paths and the measurement write catches — + * anything that must fail the cycle's verdict without aborting sibling work. + * Ported from Google Health's `noteHardFailure`. + */ +export function noteHardFailure(label: string): void { + const tracker = hardFailStorage.getStore(); + if (tracker) tracker.failures.push(label); +} + /** * Single-source the per-resource collection-fetch error handling. A 403 on one * data class soft-skips it (warn + return 0) so sibling resources still sync; a @@ -251,8 +295,7 @@ export async function handleCollectionFetchError( } await recordFitbitSyncFailure(userId, err); getEvent()?.addWarning(`fitbit ${resource} failed for ${userId}: ${err}`); - const hardTracker = hardFailStorage.getStore(); - if (hardTracker) hardTracker.failures.push(resource); + noteHardFailure(resource); return 0; } @@ -570,8 +613,7 @@ export async function upsertFitbitMeasurements( // ambient ledger so the watermark holds and the next tick retries the // rescue rather than losing the rows for good. getEvent()?.addWarning(`Fitbit: natural-key rescue probe failed: ${err}`); - const hardTracker = hardFailStorage.getStore(); - if (hardTracker) hardTracker.failures.push("measurements:rescue"); + noteHardFailure("measurements:rescue"); } } @@ -612,6 +654,11 @@ export async function upsertFitbitMeasurements( } } catch (err) { getEvent()?.addWarning(`Fitbit: failed to create measurements: ${err}`); + // Fetched-but-unwritten rows are lost for good once they age out of the + // 24 h overlap — fail the cycle's verdict so the watermark holds and the + // next tick re-fetches. No rethrow: sibling chunks and resources must + // keep writing. + noteHardFailure("measurements:create"); } } @@ -643,6 +690,10 @@ export async function upsertFitbitMeasurements( imported++; } catch (err) { getEvent()?.addWarning(`Fitbit: failed to update measurement: ${err}`); + // Same rule as the create catch: a swallowed update loses the re-fetched + // value once it ages out of the overlap — hold the watermark so the next + // tick retries. No rethrow, so the remaining rows still write. + noteHardFailure("measurements:update"); } } diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index f2596f334..bb692496e 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -50,12 +50,12 @@ import { syncUserWorkout } from "./sync-workout"; export async function syncUserFitbit( userId: string, opts: { fullSync?: boolean } = {}, -): Promise { +): Promise<{ imported: number; failed: boolean }> { if (await isReauthRequired(userId, "fitbit")) { getEvent()?.addWarning( `fitbit sync skipped for ${userId}: parked at error_reauth`, ); - return 0; + return { imported: 0, failed: true }; } // Snapshot the incremental window ONCE for the whole cycle. Every resource @@ -65,7 +65,7 @@ export async function syncUserFitbit( where: { userId }, select: { lastSyncedAt: true }, }); - if (!connection) return 0; + if (!connection) return { imported: 0, failed: true }; const end = new Date(); const start = incrementalStart(connection.lastSyncedAt, { fullSync: opts.fullSync, @@ -151,8 +151,9 @@ export async function syncUserFitbit( // partial cycle (some rows imported, or at least one resource that did not // soft-skip) stamps success as normal. const allSoftSkipped = tracker.count >= resources.length && total === 0; + const failed = anyFailed || allSoftSkipped; - if (!anyFailed && !allSoftSkipped) { + if (!failed) { // Stamp the watermark ONCE for the whole cycle, only on a non-degenerate // run. Every resource already saw the snapshot `start`, so stamping now() // here can't shrink a later resource's window. @@ -161,9 +162,9 @@ export async function syncUserFitbit( } annotate({ - action: { name: "fitbit.sync", details: { imported: total } }, + action: { name: "fitbit.sync", details: { imported: total, failed } }, }); - return total; + return { imported: total, failed }; } /** @@ -201,7 +202,9 @@ export async function runFitbitPollCohort( onUserSynced?: (userId: string, imported: number) => void; } = {}, ): Promise<{ usersSynced: number; measurementsImported: number }> { - const sync = opts.sync ?? ((userId: string) => syncUserFitbit(userId)); + const sync = + opts.sync ?? + ((userId: string) => syncUserFitbit(userId).then((r) => r.imported)); const limit = pLimit(opts.concurrency ?? FITBIT_POLL_CONCURRENCY); let usersSynced = 0; diff --git a/src/lib/jobs/__tests__/fitbit-backfill.test.ts b/src/lib/jobs/__tests__/fitbit-backfill.test.ts index 63134ea15..9c4b9306b 100644 --- a/src/lib/jobs/__tests__/fitbit-backfill.test.ts +++ b/src/lib/jobs/__tests__/fitbit-backfill.test.ts @@ -7,16 +7,18 @@ */ import { beforeEach, describe, expect, it, vi } from "vitest"; -const { prismaMock, bossSend, syncUserFitbit } = vi.hoisted(() => ({ - prismaMock: { - fitbitConnection: { - findMany: vi.fn(), - update: vi.fn(), +const { prismaMock, bossSend, syncUserFitbit, isReauthRequiredMock } = + vi.hoisted(() => ({ + prismaMock: { + fitbitConnection: { + findMany: vi.fn(), + update: vi.fn(), + }, }, - }, - bossSend: vi.fn(), - syncUserFitbit: vi.fn(), -})); + bossSend: vi.fn(), + syncUserFitbit: vi.fn(), + isReauthRequiredMock: vi.fn(async () => false), + })); vi.mock("@/lib/db", () => ({ prisma: prismaMock })); @@ -28,6 +30,10 @@ vi.mock("@/lib/fitbit/sync", () => ({ syncUserFitbit: (...a: unknown[]) => syncUserFitbit(...a), })); +vi.mock("@/lib/integrations/status", () => ({ + isReauthRequired: isReauthRequiredMock, +})); + vi.mock("@/lib/logging/context", () => ({ annotate: () => {}, getEvent: () => null, @@ -40,6 +46,8 @@ import { beforeEach(() => { vi.clearAllMocks(); + isReauthRequiredMock.mockResolvedValue(false); + prismaMock.fitbitConnection.update.mockResolvedValue({}); }); describe("enqueueBootTimeFitbitBackfill — discovery", () => { @@ -87,17 +95,33 @@ describe("enqueueBootTimeFitbitBackfill — discovery", () => { }); }); -describe("runFitbitBackfillForUser", () => { - it("runs a full sync and stamps backfillCompletedAt", async () => { - syncUserFitbit.mockResolvedValue(123); - prismaMock.fitbitConnection.update.mockResolvedValue({}); +describe("runFitbitBackfillForUser — verdict-gated marker", () => { + it("stamps backfillCompletedAt after a CLEAN full-history run", async () => { + syncUserFitbit.mockResolvedValue({ imported: 42, failed: false }); const { imported } = await runFitbitBackfillForUser("u1"); - expect(imported).toBe(123); + expect(imported).toBe(42); expect(syncUserFitbit).toHaveBeenCalledWith("u1", { fullSync: true }); const updateArg = prismaMock.fitbitConnection.update.mock.calls[0]![0]; expect(updateArg.where).toEqual({ userId: "u1" }); expect(updateArg.data.backfillCompletedAt).toBeInstanceOf(Date); }); + + it("a failed verdict THROWS without stamping — pg-boss retries become real", async () => { + syncUserFitbit.mockResolvedValue({ imported: 7, failed: true }); + + await expect(runFitbitBackfillForUser("u1")).rejects.toThrow(/incomplete/); + expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled(); + }); + + it("a connection parked at error_reauth returns WITHOUT running the sync or stamping", async () => { + isReauthRequiredMock.mockResolvedValue(true); + + await expect(runFitbitBackfillForUser("u1")).resolves.toEqual({ + imported: 0, + }); + expect(syncUserFitbit).not.toHaveBeenCalled(); + expect(prismaMock.fitbitConnection.update).not.toHaveBeenCalled(); + }); }); diff --git a/src/lib/jobs/fitbit-backfill.ts b/src/lib/jobs/fitbit-backfill.ts index 32a5fce36..fc57592fa 100644 --- a/src/lib/jobs/fitbit-backfill.ts +++ b/src/lib/jobs/fitbit-backfill.ts @@ -14,6 +14,7 @@ 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 { syncUserFitbit } from "@/lib/fitbit/sync"; export const FITBIT_BACKFILL_QUEUE = "fitbit-backfill"; @@ -32,14 +33,45 @@ export interface FitbitBackfillPayload { /** * Per-user backfill handler. Runs a full-history sync for one account and stamps - * `backfillCompletedAt` so the discovery query drops it. Idempotent: the - * per-resource upserts are key-stable, so a re-run (e.g. a reboot mid-walk) - * overwrites rather than duplicating. Mirrors `runWhoopBackfillForUser`. + * `backfillCompletedAt` ONLY on a clean run, so the discovery query drops it. + * Idempotent: the per-resource upserts are key-stable, so a re-run (e.g. a + * reboot mid-walk) overwrites rather than duplicating. Mirrors + * `runWhoopBackfillForUser` (whose identical unconditional-stamp bug is fixed + * separately — this is the template for that port). + * + * Verdict-gated: a partial hard failure (one collection 400ing, a write + * failing, a dead token) THROWS so pg-boss retries the job — stamping the + * marker over a partial walk would freeze the history gap in forever and the + * configured `retryLimit` would be dead code. A connection parked at + * `error_reauth` is NOT clean either, but throwing against a dead grant would + * just burn the retry budget on the same no-op: return WITHOUT stamping — the + * boot discovery re-enqueues the account on the next boot, and the stamp lands + * once the user reconnects and a clean walk completes. */ export async function runFitbitBackfillForUser( userId: string, ): Promise<{ imported: number }> { - const imported = await syncUserFitbit(userId, { fullSync: true }); + if (await isReauthRequired(userId, "fitbit")) { + annotate({ + action: { + name: "fitbit.backfill.skipped_reauth", + details: { imported: 0 }, + }, + }); + return { imported: 0 }; + } + + const { imported, failed } = await syncUserFitbit(userId, { fullSync: true }); + + if (failed) { + // Surface through the pg-boss retry path (retryLimit 3, backoff, set by + // `integrationBackfillSourceOptions`). If the failure parked the connection + // at error_reauth meanwhile, the next attempt takes the reauth return above + // instead of failing again. + throw new Error( + `fitbit backfill incomplete for user ${userId} — marker not stamped`, + ); + } await prisma.fitbitConnection.update({ where: { userId },