mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 21:10:59 +00:00
Merge pull request #620 from MBombeck/release/v1.32.18
release/v1.32.18 — a dead Fitbit token no longer stamps a successful sync
This commit is contained in:
12 files changed
+538
-43
No files matched your search
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
+1
-1
@@ -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",
|
||||
|
||||
+1
-1
@@ -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/*`
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
@@ -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<unknown> => 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<number> => {
|
||||
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<T>(
|
||||
fn: () => Promise<T>,
|
||||
): 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<string, unknown> };
|
||||
expect(arg.data).not.toHaveProperty("lastSyncedAt");
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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();
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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<HardFailTracker>();
|
||||
|
||||
/**
|
||||
* 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");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+10
-7
@@ -50,12 +50,12 @@ import { syncUserWorkout } from "./sync-workout";
|
||||
export async function syncUserFitbit(
|
||||
userId: string,
|
||||
opts: { fullSync?: boolean } = {},
|
||||
): Promise<number> {
|
||||
): 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;
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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 },
|
||||
|
||||
Reference in new issue
Block a user