From 686bc1bf30ff3c00783e735b2ce28eda0e5d3c6c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 4 Jun 2026 22:15:00 +0200 Subject: [PATCH] feat(fitbit): sync Google Health metrics with backfill and poll jobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extend the Fitbit/Google Health provider with the read path: the dataPoints.list client half, the metrics-bundle sync, the poll + backfill jobs, and the connection-management routes. Builds on the existing OAuth + credentials + token-refresh layer. - client: paginated `fetchDataPoints` walker over `safeFetch` keyed on `nextPageToken`, encoding each data type kebab-case in the path and snake_case in the filter predicate; defensive per-type mappers (finite + positive guards) for weight, body fat, daily SpO2, daily HRV, daily resting heart rate, daily respiratory rate, intraday heart rate, sleep-temperature derivations, and a height profile-seed extractor. `FITBIT_FIELD_MAP` + `mapping.md` are the source of truth. - mapping (no new MeasurementType): weight→WEIGHT, body-fat→BODY_FAT, SpO2→OXYGEN_SATURATION, HRV→HEART_RATE_VARIABILITY (SDNN slot, not HRV_RMSSD), resting-HR→RESTING_HEART_RATE, respiratory-rate→RESPIRATORY_RATE, heart-rate→PULSE, sleep-temp→WRIST_TEMPERATURE, height→User.heightCm when null. Steps/distance/energy/floors/VO2 + sleep + workouts stay for a later wave. - sync: `syncUserFitbit` orchestrator with the reauth short-circuit, the per-data-class 403 soft-skip (the six Restricted bundles grant independently), the all-soft-skipped success guard, and incremental-vs-full windows; `upsertFitbitMeasurements` keyed (userId,type,FITBIT,externalId) with a 24 h overlap for after-the-fact daily re-rolls, recomputing rollups and invalidating status caches. Google refresh tokens do not rotate, so the stored refresh token is kept unless a new one is returned. - jobs: hourly `fitbit-sync` poll (poll-only; Pub/Sub deferred), the self-converging boot backfill, and the daily OAuth-state sweep — every queue registered in both `allQueues` and a `boss.work` binding, staggered off the WHOOP/Withings crons. - routes: status / sync / disconnect mirroring the WHOOP shape. Tests cover the mappers, the casing invariant, `fetchDataPoints`, the orchestrator soft-skip guard, the idempotent upsert, the backfill discovery, and the queue registration; an integration test pins the upsert idempotency, the DAY rollup fold, and the Fitbit-vs-Withings per-source no-collapse against the Postgres testcontainer. --- src/app/api/fitbit/disconnect/route.ts | 42 ++ src/app/api/fitbit/status/route.ts | 57 ++ src/app/api/fitbit/sync/route.ts | 27 + src/lib/fitbit/__tests__/client.test.ts | 166 ++++++ .../fitbit/__tests__/sync-user-fitbit.test.ts | 117 ++++ src/lib/fitbit/__tests__/sync.test.ts | 61 +++ src/lib/fitbit/client.ts | 516 ++++++++++++++++++ src/lib/fitbit/mapping.md | 59 ++ src/lib/fitbit/sync-metrics.ts | 189 +++++++ src/lib/fitbit/sync.ts | 244 ++++++++- .../jobs/__tests__/fitbit-backfill.test.ts | 101 ++++ src/lib/jobs/__tests__/fitbit-queues.test.ts | 73 +++ src/lib/jobs/fitbit-backfill.ts | 32 +- src/lib/jobs/fitbit-oauth-state-cleanup.ts | 23 + src/lib/jobs/reminder-worker.ts | 156 ++++++ tests/integration/fitbit-metrics-sync.test.ts | 194 +++++++ 16 files changed, 2050 insertions(+), 7 deletions(-) create mode 100644 src/app/api/fitbit/disconnect/route.ts create mode 100644 src/app/api/fitbit/status/route.ts create mode 100644 src/app/api/fitbit/sync/route.ts create mode 100644 src/lib/fitbit/__tests__/sync-user-fitbit.test.ts create mode 100644 src/lib/fitbit/mapping.md create mode 100644 src/lib/fitbit/sync-metrics.ts create mode 100644 src/lib/jobs/__tests__/fitbit-backfill.test.ts create mode 100644 src/lib/jobs/__tests__/fitbit-queues.test.ts create mode 100644 src/lib/jobs/fitbit-oauth-state-cleanup.ts create mode 100644 tests/integration/fitbit-metrics-sync.test.ts diff --git a/src/app/api/fitbit/disconnect/route.ts b/src/app/api/fitbit/disconnect/route.ts new file mode 100644 index 000000000..96fa88fea --- /dev/null +++ b/src/app/api/fitbit/disconnect/route.ts @@ -0,0 +1,42 @@ +import { prisma } from "@/lib/db"; +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { auditLog } from "@/lib/auth/audit"; +import { apiSuccess, apiError } from "@/lib/api-response"; +import { annotate } from "@/lib/logging/context"; +import { markDisconnected } from "@/lib/integrations/status"; + +/** + * Disconnect the Fitbit / Google Health integration for the current user + * (v1.12.0). + * + * Clears the stored OAuth tokens by deleting the `FitbitConnection` row and + * parks the integration status at `disconnected`. There is no per-user + * subscription to tear down (Pub/Sub is deferred — poll-only at launch), so no + * upstream unsubscribe call is needed. The BYO-key credentials on `User` are + * left intact so a reconnect doesn't force the user to re-paste them; use the + * credentials DELETE endpoint to remove those too. + */ +export const POST = apiHandler(async () => { + const { user } = await requireAuth(); + annotate({ action: { name: "fitbit.disconnect" } }); + + const connection = await prisma.fitbitConnection.findUnique({ + where: { userId: user.id }, + }); + + if (!connection) { + return apiError("No Fitbit connection", 404); + } + + await prisma.fitbitConnection.delete({ + where: { userId: user.id }, + }); + + await auditLog("fitbit.disconnect", { + userId: user.id, + }); + + await markDisconnected(user.id, "fitbit"); + + return apiSuccess({ disconnected: true }); +}); diff --git a/src/app/api/fitbit/status/route.ts b/src/app/api/fitbit/status/route.ts new file mode 100644 index 000000000..ac51d14f1 --- /dev/null +++ b/src/app/api/fitbit/status/route.ts @@ -0,0 +1,57 @@ +import { prisma } from "@/lib/db"; +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { annotate } from "@/lib/logging/context"; +import { apiSuccess } from "@/lib/api-response"; + +/** + * Get Fitbit / Google Health connection status for the current user (v1.12.0). + * + * Mirrors the WHOOP status route: `configured` checks the per-user BYO-key + * credentials, the connection row reports last-sync + token expiry + backfill + * progress. `tokenExpired` is reported straight off the row; the sync path + * refreshes lazily via `getValidToken`. + */ +export const GET = apiHandler(async () => { + const { user } = await requireAuth(); + annotate({ action: { name: "fitbit.status" } }); + + const dbUser = await prisma.user.findUnique({ + where: { id: user.id }, + select: { + fitbitClientIdEncrypted: true, + fitbitClientSecretEncrypted: true, + }, + }); + + const configured = + !!dbUser?.fitbitClientIdEncrypted && !!dbUser?.fitbitClientSecretEncrypted; + + const connection = await prisma.fitbitConnection.findUnique({ + where: { userId: user.id }, + select: { + fitbitUserId: true, + lastSyncedAt: true, + tokenExpiresAt: true, + backfillCompletedAt: true, + createdAt: true, + scope: true, + }, + }); + + if (!connection) { + return apiSuccess({ connected: false, configured }); + } + + const tokenExpired = connection.tokenExpiresAt.getTime() <= Date.now(); + + return apiSuccess({ + connected: true, + configured, + lastSyncedAt: connection.lastSyncedAt, + connectedAt: connection.createdAt, + tokenExpired, + tokenExpiresAt: connection.tokenExpiresAt, + backfillCompleted: !!connection.backfillCompletedAt, + scope: connection.scope, + }); +}); diff --git a/src/app/api/fitbit/sync/route.ts b/src/app/api/fitbit/sync/route.ts new file mode 100644 index 000000000..e99c8873f --- /dev/null +++ b/src/app/api/fitbit/sync/route.ts @@ -0,0 +1,27 @@ +import { apiHandler, requireAuth } from "@/lib/api-handler"; +import { apiSuccess } from "@/lib/api-response"; +import { annotate } from "@/lib/logging/context"; +import { syncUserFitbit } from "@/lib/fitbit/sync"; +import { NextRequest } from "next/server"; + +/** + * Manually trigger a Fitbit / Google Health sync for the current user (v1.12.0). + * + * Mirrors the WHOOP manual-sync route: incremental by default, full history when + * `{ fullSync: true }` is posted. + */ +export const POST = apiHandler(async (request: NextRequest) => { + const { user } = await requireAuth(); + annotate({ action: { name: "fitbit.sync" } }); + + let fullSync = false; + try { + const body = await request.json(); + fullSync = body?.fullSync === true; + } catch { + // no body provided -> default incremental sync + } + + const imported = await syncUserFitbit(user.id, { fullSync }); + return apiSuccess({ imported, fullSync }); +}); diff --git a/src/lib/fitbit/__tests__/client.test.ts b/src/lib/fitbit/__tests__/client.test.ts index 9afec3831..63b03b70e 100644 --- a/src/lib/fitbit/__tests__/client.test.ts +++ b/src/lib/fitbit/__tests__/client.test.ts @@ -1,10 +1,19 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { FITBIT_API_BASE, + FITBIT_DATA_TYPES, + FITBIT_FIELD_MAP, FITBIT_OAUTH_SCOPE, exchangeCode, + fetchDataPoints, fetchProfile, getAuthorizationUrl, + mapBodyFat, + mapHeartRateVariability, + mapHeightCm, + mapOxygenSaturation, + mapRestingHeartRate, + mapWeight, refreshAccessToken, resolveFitbitUserId, } from "../client"; @@ -160,3 +169,160 @@ describe("fetchProfile + resolveFitbitUserId", () => { expect(resolveFitbitUserId({})).toBe("me"); }); }); + +describe("fetchDataPoints", () => { + it("encodes the data type kebab-case in the path and snake_case in the filter, walking nextPageToken", async () => { + const fetchMock = installFetchMock([ + { + status: 200, + body: { + dataPoints: [{ a: 1 }], + nextPageToken: "tok2", + }, + }, + { status: 200, body: { dataPoints: [{ a: 2 }] } }, + ]); + + const points = await fetchDataPoints( + FITBIT_DATA_TYPES.bodyFat, + "at", + "fetchBodyFat", + { start: new Date("2026-01-01T00:00:00.000Z") }, + ); + expect(points).toHaveLength(2); + + const [url1] = fetchMock.mock.calls[0] as unknown as [string]; + // Path uses kebab-case; filter uses snake_case. + expect(url1).toContain( + `${FITBIT_API_BASE}/users/me/dataTypes/body-fat/dataPoints`, + ); + const parsed = new URL(url1); + expect(parsed.searchParams.get("filter")).toBe( + 'body_fat.sample_time.physical_time >= "2026-01-01T00:00:00.000Z"', + ); + + // Second page carries the pageToken from the first response. + const [url2] = fetchMock.mock.calls[1] as unknown as [string]; + expect(new URL(url2).searchParams.get("pageToken")).toBe("tok2"); + }); + + it("filters daily summaries on the civil date, not the sample time", async () => { + const fetchMock = installFetchMock([ + { status: 200, body: { dataPoints: [] } }, + ]); + await fetchDataPoints( + FITBIT_DATA_TYPES.restingHeartRate, + "at", + "fetchRhr", + { start: new Date("2026-03-04T10:00:00.000Z") }, + ); + const [url] = fetchMock.mock.calls[0] as unknown as [string]; + expect(new URL(url).searchParams.get("filter")).toBe( + 'daily_resting_heart_rate.date >= "2026-03-04"', + ); + }); + + it("throws a classified FitbitApiError on a non-2xx page", async () => { + installFetchMock([{ status: 403, body: {} }]); + await expect( + fetchDataPoints(FITBIT_DATA_TYPES.weight, "at", "fetchWeight"), + ).rejects.toMatchObject({ + name: "FitbitApiError", + classification: "reauth_required", + httpStatus: 403, + }); + }); +}); + +describe("metric mappers", () => { + it("maps weight in kg with a stable sample-time anchor + field-tag externalId", () => { + const point = { + weight: { + kilograms: 81.456, + sample_time: { physical_time: "2026-05-10T07:30:00.000Z" }, + }, + }; + const out = mapWeight(point); + expect(out).toHaveLength(1); + expect(out[0]).toMatchObject({ + type: "WEIGHT", + value: 81.46, + unit: "kg", + }); + expect(out[0]!.measuredAt.toISOString()).toBe("2026-05-10T07:30:00.000Z"); + // externalId field-tag: :; anchor is the sample time ISO string. + expect(out[0]!.fieldTag).toBe("2026-05-10T07:30:00.000Z:weight"); + }); + + it("maps body fat percentage", () => { + const out = mapBodyFat({ body_fat: { percentage: 18.2 } }); + expect(out[0]).toMatchObject({ type: "BODY_FAT", value: 18.2, unit: "%" }); + }); + + it("maps daily SpO2 with a civil-date anchor", () => { + const out = mapOxygenSaturation({ + daily_oxygen_saturation: { + average_percentage: 97.4, + date: { year: 2026, month: 5, day: 10 }, + }, + }); + expect(out[0]).toMatchObject({ + type: "OXYGEN_SATURATION", + value: 97.4, + unit: "%", + }); + // Daily anchor is the civil date (YYYY-MM-DD), so a re-fetch of the same day + // overwrites in place. + expect(out[0]!.fieldTag).toBe("2026-05-10:spo2"); + }); + + it("maps Fitbit HRV into the SDNN HEART_RATE_VARIABILITY slot, NOT HRV_RMSSD", () => { + const out = mapHeartRateVariability({ + daily_heart_rate_variability: { + rmssd_milliseconds: 42.7, + date: { year: 2026, month: 5, day: 10 }, + }, + }); + expect(out[0]!.type).toBe("HEART_RATE_VARIABILITY"); + expect(out[0]!.type).not.toBe("HRV_RMSSD"); + expect(out[0]).toMatchObject({ value: 42.7, unit: "ms" }); + }); + + it("maps daily resting heart rate", () => { + const out = mapRestingHeartRate({ + daily_resting_heart_rate: { + beats_per_minute: 54, + date: { year: 2026, month: 5, day: 10 }, + }, + }); + expect(out[0]).toMatchObject({ type: "RESTING_HEART_RATE", value: 54 }); + }); + + it("drops a garbage / non-positive reading rather than minting a row", () => { + expect(mapWeight({ weight: { kilograms: 0 } })).toHaveLength(0); + expect(mapWeight({ weight: { kilograms: Number.NaN } })).toHaveLength(0); + expect(mapWeight({})).toHaveLength(0); + }); + + it("resolves height to cm from either centimetres or metres, never as a Measurement", () => { + expect(mapHeightCm({ height: { centimeters: 178 } })).toBe(178); + expect(mapHeightCm({ height: { meters: 1.78 } })).toBe(178); + expect(mapHeightCm({ height: {} })).toBeNull(); + }); +}); + +describe("FITBIT_FIELD_MAP ↔ FITBIT_DATA_TYPES casing parity", () => { + it("keeps the kebab-path / snake-filter pair consistent across both tables", () => { + for (const [, entry] of Object.entries(FITBIT_FIELD_MAP)) { + // Path is kebab-case, filter is snake_case — they must never carry the + // wrong separator (the design §A.1 casing gotcha). + expect(entry.path).not.toContain("_"); + expect(entry.filter).not.toContain("-"); + } + // Every FITBIT_DATA_TYPES entry has matching casing too. + for (const dt of Object.values(FITBIT_DATA_TYPES)) { + expect(dt.path).not.toContain("_"); + expect(dt.filter).not.toContain("-"); + } + }); +}); diff --git a/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts new file mode 100644 index 000000000..ab2e10fca --- /dev/null +++ b/src/lib/fitbit/__tests__/sync-user-fitbit.test.ts @@ -0,0 +1,117 @@ +/** + * v1.12.0 — `syncUserFitbit` orchestration tests (mocked resource sync). + * + * Covers the all-403 "looks-healthy" guard: a genuine grant-revoke 403s the + * metrics collection. The resource sync soft-skips (returns 0, records no + * failure) via `handleCollectionFetchError`, so the orchestrator sees + * `total === 0` with no thrown error — and without the guard would stamp + * `recordSyncSuccess`, keeping a dead connection "connected" until the + * token-refresh path catches the 401. The guard refuses to stamp success when + * every enabled resource soft-skipped and nothing imported. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { + recordSyncSuccess, + recordSyncFailure, + isReauthRequired, + syncUserMetrics, +} = vi.hoisted(() => ({ + recordSyncSuccess: vi.fn<(...a: unknown[]) => Promise>(async () => {}), + recordSyncFailure: vi.fn<(...a: unknown[]) => Promise>(async () => {}), + isReauthRequired: vi.fn<(...a: unknown[]) => Promise>( + async () => false, + ), + syncUserMetrics: vi.fn<(...a: unknown[]) => Promise>(), +})); + +vi.mock("@/lib/db", () => ({ prisma: {} })); +vi.mock("@/lib/crypto", () => ({ + encrypt: (s: string) => s, + decrypt: (s: string) => s, +})); +vi.mock("@/lib/logging/context", () => ({ + getEvent: () => null, + annotate: () => {}, +})); +vi.mock("@/lib/integrations/status", () => ({ + recordSyncFailure: (...a: unknown[]) => recordSyncFailure(...a), + recordSyncSuccess: (...a: unknown[]) => recordSyncSuccess(...a), + isReauthRequired: (...a: unknown[]) => isReauthRequired(...a), +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: () => [], + recomputeBucketsForMeasurement: vi.fn(async () => {}), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); + +vi.mock("../sync-metrics", () => ({ + syncUserMetrics: (...a: unknown[]) => syncUserMetrics(...a), +})); + +import { handleCollectionFetchError, syncUserFitbit } from "../sync"; +import { FitbitApiError } from "../response-classifier"; + +/** A resource sync that 403s its collection and soft-skips, like the real one. */ +async function softSkip403(...args: unknown[]): Promise { + return handleCollectionFetchError( + "test-resource", + String(args[0]), + new FitbitApiError({ + verb: "fetchTest", + classification: "reauth_required", + httpStatus: 403, + reason: "http_403", + }), + ); +} + +beforeEach(() => { + vi.clearAllMocks(); + isReauthRequired.mockResolvedValue(false); +}); + +describe("syncUserFitbit — all-403 looks-healthy guard", () => { + it("does NOT stamp success when the resource soft-skipped and nothing imported", async () => { + syncUserMetrics.mockImplementation(softSkip403); + + const total = await syncUserFitbit("user1"); + + expect(total).toBe(0); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + // Soft-skips record no failure either — the status is left exactly as-is so + // the token-refresh 401 path is the one that flips it to reauth. + expect(recordSyncFailure).not.toHaveBeenCalled(); + }); + + it("stamps success on a cycle that imported rows", async () => { + syncUserMetrics.mockResolvedValue(5); + + const total = await syncUserFitbit("user1"); + + expect(total).toBe(5); + 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"); + + expect(total).toBe(0); + // No collection was 403'd → this is a genuine "nothing changed" tick. + expect(recordSyncSuccess).toHaveBeenCalledWith("user1", "fitbit"); + }); + + it("short-circuits without stamping success when parked at error_reauth", async () => { + isReauthRequired.mockResolvedValue(true); + + const total = await syncUserFitbit("user1"); + + expect(total).toBe(0); + expect(recordSyncSuccess).not.toHaveBeenCalled(); + expect(syncUserMetrics).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/fitbit/__tests__/sync.test.ts b/src/lib/fitbit/__tests__/sync.test.ts index db7ee54f8..b3bb8fbab 100644 --- a/src/lib/fitbit/__tests__/sync.test.ts +++ b/src/lib/fitbit/__tests__/sync.test.ts @@ -14,6 +14,11 @@ const { prismaMock, refreshAccessTokenMock, recordSyncFailure } = vi.hoisted( findUnique: vi.fn(), update: vi.fn(), }, + measurement: { + upsert: vi.fn<(arg: Record) => Promise>( + async () => ({}), + ), + }, }, refreshAccessTokenMock: vi.fn(), recordSyncFailure: vi.fn<(...a: unknown[]) => Promise>(async () => {}), @@ -22,6 +27,15 @@ const { prismaMock, refreshAccessTokenMock, recordSyncFailure } = vi.hoisted( vi.mock("@/lib/db", () => ({ prisma: prismaMock })); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + collapseToTypeDayKeys: () => [], + recomputeBucketsForMeasurement: vi.fn(async () => {}), +})); + +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn(async () => {}), +})); + vi.mock("@/lib/crypto", () => ({ encrypt: (s: string) => `enc(${s})`, decrypt: (s: string) => s.replace(/^enc\(|\)$/g, ""), @@ -49,6 +63,7 @@ vi.mock("@/lib/integrations/status", () => ({ import { classificationToFailureKind, getValidToken, + upsertFitbitMeasurements, } from "../sync"; beforeEach(() => { @@ -180,6 +195,52 @@ describe("getValidToken — non-rotating refresh", () => { }); }); +describe("upsertFitbitMeasurements — idempotent FITBIT upsert", () => { + it("upserts each reading on the (userId, type, source=FITBIT, externalId) key with field-by-field data", async () => { + const measuredAt = new Date("2026-05-10T07:00:00.000Z"); + const imported = await upsertFitbitMeasurements("user1", [ + { + type: "WEIGHT", + value: 80.5, + unit: "kg", + measuredAt, + externalId: "2026-05-10T07:00:00.000Z:weight", + }, + ]); + + expect(imported).toBe(1); + const arg = prismaMock.measurement.upsert.mock.calls[0]![0] as { + where: { userId_type_source_externalId: Record }; + create: Record; + update: { syncVersion: unknown }; + }; + // The composite unique key pins source = FITBIT — a re-post overwrites in + // place rather than minting a duplicate. + expect(arg.where.userId_type_source_externalId).toEqual({ + userId: "user1", + type: "WEIGHT", + source: "FITBIT", + externalId: "2026-05-10T07:00:00.000Z:weight", + }); + // create builds the data field-by-field (no mass-assignment spread). + expect(arg.create).toMatchObject({ + userId: "user1", + type: "WEIGHT", + source: "FITBIT", + value: 80.5, + unit: "kg", + }); + // update bumps syncVersion for the iOS LWW reconciler. + expect(arg.update.syncVersion).toEqual({ increment: 1 }); + }); + + it("returns 0 without touching prisma for an empty batch", async () => { + const imported = await upsertFitbitMeasurements("user1", []); + expect(imported).toBe(0); + expect(prismaMock.measurement.upsert).not.toHaveBeenCalled(); + }); +}); + describe("classificationToFailureKind", () => { it("maps each classification onto a FailureKind", () => { expect(classificationToFailureKind("reauth_required")).toBe( diff --git a/src/lib/fitbit/client.ts b/src/lib/fitbit/client.ts index e6bfaea28..8f9f47b3d 100644 --- a/src/lib/fitbit/client.ts +++ b/src/lib/fitbit/client.ts @@ -233,3 +233,519 @@ export function resolveFitbitUserId(profile: FitbitProfile): string { if (profile.id) return profile.id; return "me"; } + +// ─── Data-point reads (Google Health `dataPoints.list`) ──────── +// +// The Google Health API exposes one uniform `DataPoint` resource shape across +// every data type, read through `GET /v4/users/me/dataTypes/{dataType}/dataPoints` +// with `nextPageToken` pagination. This mirrors the WHOOP `fetchCollection` +// page-walk (`whoop/client.ts:270`), but Google's value-field JSON is NOT fully +// published yet — so the mappers below are intentionally defensive: every value +// is pulled out of a small set of candidate field shapes and guarded by a +// finite-positive check before it becomes a Measurement. Capture the real +// per-type value-field JSON into `src/lib/fitbit/mapping.md` against a live test +// account at build and tighten the extractors then. +// +// Casing gotcha (design §A.1): the data-type id is **kebab-case in the path** +// (`body-fat`) and **snake_case in the `filter`** (`body_fat`). `FITBIT_DATA_TYPES` +// pins both forms so a fetcher can never encode the wrong one. + +/** + * Page-size ceiling for `dataPoints.list`. The daily/intraday reads default to + * 1440 (one-per-minute) and cap at 10 000; sleep/exercise cap at 25 (same as + * WHOOP's `WHOOP_PAGE_LIMIT`). The launch metrics use the daily/spot reads, so + * the default page size is the larger value. + */ +export const FITBIT_PAGE_SIZE = 1000; +/** Sleep/exercise read cap — matches the Google Health 25-cap for those types. */ +export const FITBIT_ACTIVITY_PAGE_SIZE = 25; + +/** + * One data type's two on-the-wire encodings. `path` is the kebab-case segment + * spliced into the request URL; `filter` is the snake_case prefix used to build + * the incremental `filter=` predicate. The `timeField` names the value object's + * time anchor (a sample time for spot readings, a date for daily summaries) so + * the incremental filter targets the right field. + */ +export interface FitbitDataType { + /** kebab-case segment for the request path. */ + path: string; + /** snake_case prefix for the `filter` predicate. */ + filter: string; + /** + * Which time anchor the filter / measuredAt resolution targets: + * - `sample` → spot reading (`{type}.sample_time.physical_time`). + * - `date` → daily summary (`{type}.date`). + */ + timeField: "sample" | "date"; +} + +/** + * The launch (W3 metrics) data types. Each entry pins the kebab-path + snake-filter + * pair so the two encodings can never drift. Activity/sleep/workout types land in + * W5; only the health-metrics bundle is read here. + */ +export const FITBIT_DATA_TYPES = { + weight: { path: "weight", filter: "weight", timeField: "sample" }, + bodyFat: { path: "body-fat", filter: "body_fat", timeField: "sample" }, + oxygenSaturation: { + path: "daily-oxygen-saturation", + filter: "daily_oxygen_saturation", + timeField: "date", + }, + heartRateVariability: { + path: "daily-heart-rate-variability", + filter: "daily_heart_rate_variability", + timeField: "date", + }, + restingHeartRate: { + path: "daily-resting-heart-rate", + filter: "daily_resting_heart_rate", + timeField: "date", + }, + respiratoryRate: { + path: "daily-respiratory-rate", + filter: "daily_respiratory_rate", + timeField: "date", + }, + heartRate: { path: "heart-rate", filter: "heart_rate", timeField: "sample" }, + height: { path: "height", filter: "height", timeField: "sample" }, + sleepTemperature: { + path: "daily-sleep-temperature-derivations", + filter: "daily_sleep_temperature_derivations", + timeField: "date", + }, +} as const satisfies Record; + +export type FitbitDataTypeKey = keyof typeof FITBIT_DATA_TYPES; + +/** Google Health `DataPoint` — value object is type-keyed + carries a time anchor. */ +export interface FitbitDataPoint { + [key: string]: unknown; +} + +/** `dataPoints.list` envelope: `{ dataPoints, nextPageToken }`. */ +interface FitbitDataPointPage { + dataPoints?: FitbitDataPoint[]; + nextPageToken?: string | null; +} + +interface DataPointQuery { + /** Lower-bound incremental cursor; omitted on a full backfill. */ + start?: Date; + /** Page size (defaults to `FITBIT_PAGE_SIZE`). */ + pageSize?: number; + /** Hard ceiling on pages walked (defence against a runaway cursor). */ + maxPages?: number; +} + +/** + * Walk every `DataPoint` for one data type since the incremental cursor. + * Mirrors the WHOOP `fetchCollection` page-loop: `nextPageToken` pagination, a + * 1000-page `maxPages` ceiling, and per-page `addExternalCall` telemetry. The + * data-type id is kebab-cased in the path; the `filter` predicate is built from + * the snake_case form against the type's time anchor. + */ +export async function fetchDataPoints( + dataType: FitbitDataType, + accessToken: string, + verb: string, + query: DataPointQuery = {}, +): Promise { + const points: FitbitDataPoint[] = []; + let pageToken: string | null | undefined; + let pageCount = 0; + const maxPages = query.maxPages ?? 1000; + const pageSize = query.pageSize ?? FITBIT_PAGE_SIZE; + + do { + const params = new URLSearchParams({ pageSize: String(pageSize) }); + if (query.start) { + // snake_case filter prefix; sample readings filter on the sample time, + // daily summaries on the civil date. + const field = + dataType.timeField === "sample" + ? `${dataType.filter}.sample_time.physical_time` + : `${dataType.filter}.date`; + const bound = + dataType.timeField === "sample" + ? query.start.toISOString() + : query.start.toISOString().slice(0, 10); + params.set("filter", `${field} >= "${bound}"`); + } + if (pageToken) params.set("pageToken", pageToken); + + const pageStart = performance.now(); + const res = await safeFetch( + `${FITBIT_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`, + { + method: "GET", + headers: { Authorization: `Bearer ${accessToken}` }, + }, + ); + + const json = (await res.json().catch(() => null)) as FitbitDataPointPage | null; + const verdict = classifyFitbitResponse(res.status); + getEvent()?.addExternalCall({ + service: "fitbit", + method: `${verb}(page=${pageCount})`, + duration_ms: Math.round(performance.now() - pageStart), + status: res.status, + error: verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new FitbitApiError({ + verb, + classification: verdict.classification, + httpStatus: verdict.httpStatus, + reason: verdict.reason, + }); + } + + for (const p of json?.dataPoints ?? []) points.push(p); + pageToken = json?.nextPageToken ?? null; + pageCount += 1; + } while (pageToken && pageCount < maxPages); + + return points; +} + +// ─── Field → Measurement mapping ─────────────────────────────── +// The single source of truth is `src/lib/fitbit/mapping.md` — keep both in +// sync when adding entries. + +/** Metres → centimetres (Fitbit `height` → `User.heightCm`). */ +const M_TO_CM = 100; + +function round2(n: number): number { + return parseFloat(n.toFixed(2)); +} + +/** Finite + strictly-positive guard — the WHOOP `positive()` discipline. */ +function positive(n: unknown): n is number { + return typeof n === "number" && Number.isFinite(n) && n > 0; +} + +/** + * A single mapped Fitbit reading destined for one `Measurement` row. The + * `source` (`FITBIT`) and `externalId` (`:`) are stamped by the + * sync layer; the mapper emits only type/value/unit/measuredAt + the field-tag + * that disambiguates the externalId. Mirrors WHOOP's `MappedMeasurement`. + */ +export interface FitbitMappedMeasurement { + type: string; + value: number; + unit: string; + measuredAt: Date; + /** Disambiguator appended to the per-point anchor to form the externalId. */ + fieldTag: string; +} + +/** + * Pull the first finite number out of a list of candidate value paths on a + * `DataPoint`. The Google value-field JSON is undocumented, so each mapper hands + * the small set of shapes the field is likely to take (a bare `value`, a typed + * sub-object, a `{ value: { fpVal } }` wrapper, …); the first finite hit wins. + */ +function firstNumber( + point: FitbitDataPoint, + paths: string[], +): number | null { + for (const path of paths) { + const v = readPath(point, path); + if (positive(v)) return v; + // A value can legitimately be zero for some metrics, but the launch metrics + // (weight, body-fat, SpO2, HRV, RHR, respiratory rate, HR, height) are all + // strictly positive — a zero is a garbage/empty reading, so `positive` is + // the right guard. + } + return null; +} + +/** Resolve a dotted path against a nested object; undefined on any miss. */ +function readPath(obj: unknown, path: string): unknown { + let cur: unknown = obj; + for (const seg of path.split(".")) { + if (cur === null || typeof cur !== "object") return undefined; + cur = (cur as Record)[seg]; + } + return cur; +} + +/** + * Resolve a `DataPoint`'s measurement timestamp. Sample readings carry a + * `sample_time.physical_time`; daily summaries carry a `date` (a civil date or a + * `{year,month,day}` object). Falls back to the fetch time only when nothing + * parses, so a row is never dropped for a missing anchor. + */ +function resolveMeasuredAt( + point: FitbitDataPoint, + dataType: FitbitDataType, + fallback: Date, +): Date { + if (dataType.timeField === "sample") { + const t = readPath(point, `${dataType.filter}.sample_time.physical_time`); + if (typeof t === "string") { + const d = new Date(t); + if (!Number.isNaN(d.getTime())) return d; + } + } else { + const dateVal = readPath(point, `${dataType.filter}.date`); + if (typeof dateVal === "string") { + const d = new Date(dateVal); + if (!Number.isNaN(d.getTime())) return d; + } + if (dateVal && typeof dateVal === "object") { + const o = dateVal as Record; + if ( + typeof o.year === "number" && + typeof o.month === "number" && + typeof o.day === "number" + ) { + // Google civil dates are 1-based months; anchor at UTC midday so a + // timezone shift can't roll the civil day across a boundary. + return new Date(Date.UTC(o.year, o.month - 1, o.day, 12)); + } + } + } + return fallback; +} + +/** + * Stable anchor for a `DataPoint`'s externalId. A spot reading anchors on its + * sample time; a daily summary on its civil date. Combined with a type-specific + * field-tag this makes the upsert key idempotent across re-fetches (a re-sync of + * the same window overwrites in place rather than minting a duplicate). + */ +function externalAnchor(point: FitbitDataPoint, dataType: FitbitDataType): string { + const at = resolveMeasuredAt(point, dataType, new Date(0)); + if (dataType.timeField === "date") return at.toISOString().slice(0, 10); + return at.toISOString(); +} + +/** + * Map one data point of a simple single-value metric into a Measurement reading. + * `valuePaths` lists the candidate value shapes; the first finite-positive hit + * wins. Returns an empty array when no value parses (an empty/garbage point). + */ +function mapSimple( + point: FitbitDataPoint, + dataType: FitbitDataType, + spec: { type: string; unit: string; fieldTag: string; valuePaths: string[]; factor?: number }, +): FitbitMappedMeasurement[] { + let value = firstNumber(point, spec.valuePaths); + if (value === null) return []; + if (spec.factor) value = value * spec.factor; + return [ + { + type: spec.type, + value: round2(value), + unit: spec.unit, + measuredAt: resolveMeasuredAt(point, dataType, new Date()), + fieldTag: `${externalAnchor(point, dataType)}:${spec.fieldTag}`, + }, + ]; +} + +/** Candidate value-field shapes a Google `DataPoint` is likely to carry. */ +function valuePaths(filterKey: string, leaf: string): string[] { + return [ + `${filterKey}.${leaf}`, + `${filterKey}.value.${leaf}`, + `${filterKey}.value`, + `value.${leaf}`, + `value`, + ]; +} + +export function mapWeight(point: FitbitDataPoint): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.weight; + return mapSimple(point, dt, { + type: "WEIGHT", + unit: "kg", + fieldTag: "weight", + valuePaths: valuePaths(dt.filter, "kilograms"), + }); +} + +export function mapBodyFat(point: FitbitDataPoint): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.bodyFat; + return mapSimple(point, dt, { + type: "BODY_FAT", + unit: "%", + fieldTag: "body_fat", + valuePaths: valuePaths(dt.filter, "percentage"), + }); +} + +export function mapOxygenSaturation( + point: FitbitDataPoint, +): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.oxygenSaturation; + return mapSimple(point, dt, { + type: "OXYGEN_SATURATION", + unit: "%", + fieldTag: "spo2", + valuePaths: [ + ...valuePaths(dt.filter, "average_percentage"), + ...valuePaths(dt.filter, "percentage"), + ], + }); +} + +export function mapHeartRateVariability( + point: FitbitDataPoint, +): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.heartRateVariability; + // Fitbit reports a nightly RMSSD-style HRV. Per the design decision it lands + // in the SDNN-lineage `HEART_RATE_VARIABILITY` slot (Apple-comparable), NOT + // WHOOP's `HRV_RMSSD` (reserved for the WHOOP-native estimator). Re-confirm + // the estimator against a live account and revisit if it warrants `HRV_RMSSD`. + return mapSimple(point, dt, { + type: "HEART_RATE_VARIABILITY", + unit: "ms", + fieldTag: "hrv", + valuePaths: [ + ...valuePaths(dt.filter, "rmssd_milliseconds"), + ...valuePaths(dt.filter, "milliseconds"), + ], + }); +} + +export function mapRestingHeartRate( + point: FitbitDataPoint, +): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.restingHeartRate; + return mapSimple(point, dt, { + type: "RESTING_HEART_RATE", + unit: "bpm", + fieldTag: "rhr", + valuePaths: valuePaths(dt.filter, "beats_per_minute"), + }); +} + +export function mapRespiratoryRate( + point: FitbitDataPoint, +): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.respiratoryRate; + return mapSimple(point, dt, { + type: "RESPIRATORY_RATE", + unit: "breaths/min", + fieldTag: "resp_rate", + valuePaths: [ + ...valuePaths(dt.filter, "breaths_per_minute"), + ...valuePaths(dt.filter, "average_breaths_per_minute"), + ], + }); +} + +export function mapHeartRate(point: FitbitDataPoint): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.heartRate; + return mapSimple(point, dt, { + type: "PULSE", + unit: "bpm", + fieldTag: "hr", + valuePaths: valuePaths(dt.filter, "beats_per_minute"), + }); +} + +export function mapSleepTemperature( + point: FitbitDataPoint, +): FitbitMappedMeasurement[] { + const dt = FITBIT_DATA_TYPES.sleepTemperature; + // Google surfaces a sleeping skin/wrist temperature derivation. `WRIST_TEMPERATURE` + // is the closest semantic slot (Apple sleeping-wrist-temp). Confirm absolute-vs- + // baseline at build; the guard rejects a non-positive (baseline-delta) reading. + return mapSimple(point, dt, { + type: "WRIST_TEMPERATURE", + unit: "celsius", + fieldTag: "wrist_temp", + valuePaths: [ + ...valuePaths(dt.filter, "nightly_temperature_celsius"), + ...valuePaths(dt.filter, "celsius"), + ], + }); +} + +/** + * Extract the profile height (in cm) from a Fitbit `height` data point, or null + * when nothing parses. Height is a one-time `User.heightCm` profile seed (written + * only when the user has no height yet) — NOT a Measurement, mirroring WHOOP's + * `mapBody` height handling. Returns cm. + */ +export function mapHeightCm(point: FitbitDataPoint): number | null { + const dt = FITBIT_DATA_TYPES.height; + // Google may report height in metres or centimetres; try cm-direct first, then + // metres × 100. Both pass through the positive guard. + const cm = firstNumber(point, valuePaths(dt.filter, "centimeters")); + if (cm !== null) return round2(cm); + const m = firstNumber(point, valuePaths(dt.filter, "meters")); + if (m !== null) return round2(m * M_TO_CM); + return null; +} + +/** + * Field→Measurement mapping table (mirror of `mapping.md`). Documents which + * Fitbit/Google-Health source field becomes which MeasurementType + unit, and + * pins the kebab-path / snake-filter pair for each. Used as the single-glance + * reference and by the mapper tests; the mappers above are the executable form. + */ +export const FITBIT_FIELD_MAP: Record< + string, + { type: string; unit: string; path: string; filter: string; note?: string } +> = { + weight: { + type: "WEIGHT", + unit: "kg", + path: "weight", + filter: "weight", + note: "picker ranks a real Withings scale above Fitbit", + }, + bodyFat: { type: "BODY_FAT", unit: "%", path: "body-fat", filter: "body_fat" }, + oxygenSaturation: { + type: "OXYGEN_SATURATION", + unit: "%", + path: "daily-oxygen-saturation", + filter: "daily_oxygen_saturation", + }, + heartRateVariability: { + type: "HEART_RATE_VARIABILITY", + unit: "ms", + path: "daily-heart-rate-variability", + filter: "daily_heart_rate_variability", + note: "SDNN slot (Apple-comparable), NOT HRV_RMSSD; confirm estimator at build", + }, + restingHeartRate: { + type: "RESTING_HEART_RATE", + unit: "bpm", + path: "daily-resting-heart-rate", + filter: "daily_resting_heart_rate", + }, + respiratoryRate: { + type: "RESPIRATORY_RATE", + unit: "breaths/min", + path: "daily-respiratory-rate", + filter: "daily_respiratory_rate", + }, + heartRate: { + type: "PULSE", + unit: "bpm", + path: "heart-rate", + filter: "heart_rate", + note: "intraday spot HR", + }, + height: { + type: "User.heightCm", + unit: "cm", + path: "height", + filter: "height", + note: "profile seed — written to User.heightCm only when null, never as a Measurement", + }, + sleepTemperature: { + type: "WRIST_TEMPERATURE", + unit: "celsius", + path: "daily-sleep-temperature-derivations", + filter: "daily_sleep_temperature_derivations", + note: "overnight wrist-temp; confirm absolute-vs-baseline at build", + }, +}; diff --git a/src/lib/fitbit/mapping.md b/src/lib/fitbit/mapping.md new file mode 100644 index 000000000..a557be31b --- /dev/null +++ b/src/lib/fitbit/mapping.md @@ -0,0 +1,59 @@ +# Fitbit / Google Health field → Measurement mapping + +Single source of truth for the Fitbit (Google Health API v4) data-type → +HealthLog `Measurement` mapping. Keep this in sync with `FITBIT_FIELD_MAP` and +the per-type mappers in `client.ts`. Every Fitbit row ingests server-side with +`source = FITBIT` and `externalId = :` (the anchor is the data +point's sample time for spot readings or its civil date for daily summaries; the +field-tag disambiguates the metric). + +Transport docs: `developers.google.com/health` (v4, post-Fitbit-Web-API). The +per-type value-field JSON is **NOT fully published** — the mappers read a small +set of candidate value paths defensively (`firstNumber` walks each shape and +takes the first finite-positive hit). **Re-verify against a live test account at +build** and tighten the extractors. Design: `.planning/v1.12.0-google-health-design.md`. + +## Casing gotcha (design §A.1) + +The data-type id is **kebab-case in the request path** (`body-fat`) and +**snake_case in the `filter` predicate** (`body_fat`). `FITBIT_DATA_TYPES` pins +both forms per type so a fetcher can never encode the wrong one. + +## Health-metrics bundle (W3) + +Scope: `googlehealth.health_metrics_and_measurements.readonly`. + +| Data type (path / filter) | MeasurementType | Unit | fieldTag | Grain | Note | +|---|---|---|---|---|---| +| `weight` / `weight` | `WEIGHT` | kg | `weight` | sample | Picker ranks a real Withings scale above Fitbit. | +| `body-fat` / `body_fat` | `BODY_FAT` | % | `body_fat` | sample | | +| `daily-oxygen-saturation` / `daily_oxygen_saturation` | `OXYGEN_SATURATION` | % | `spo2` | daily | Already 0–100. | +| `daily-heart-rate-variability` / `daily_heart_rate_variability` | `HEART_RATE_VARIABILITY` | ms | `hrv` | daily | **Decision:** SDNN slot (Apple-comparable), NOT `HRV_RMSSD`. Fitbit reports an RMSSD-style nightly HRV — confirm the estimator at build and reconsider `HRV_RMSSD` if warranted. | +| `daily-resting-heart-rate` / `daily_resting_heart_rate` | `RESTING_HEART_RATE` | bpm | `rhr` | daily | | +| `daily-respiratory-rate` / `daily_respiratory_rate` | `RESPIRATORY_RATE` | breaths/min | `resp_rate` | daily | | +| `heart-rate` / `heart_rate` | `PULSE` | bpm | `hr` | sample | Intraday spot HR. | +| `daily-sleep-temperature-derivations` / `daily_sleep_temperature_derivations` | `WRIST_TEMPERATURE` | celsius | `wrist_temp` | daily | Closest semantic slot to Apple sleeping-wrist-temp. Confirm absolute-vs-baseline at build; the positive guard rejects a baseline-delta reading. | +| `height` / `height` | `User.heightCm` | cm | — | sample | Profile seed — written to `User.heightCm` ONLY when null; never minted as a Measurement (WHOOP `mapBody` pattern). m→cm when reported in metres. | + +Every value passes a finite + strictly-positive guard (the launch metrics are +all positive — a zero/NaN is a garbage/empty reading and is dropped). + +## Deferred (Google Health Q3-2026 roadmap — slots exist, NOT in launch set) + +`blood-glucose` → `BLOOD_GLUCOSE`, `blood-pressure` → `BLOOD_PRESSURE_SYS`+`_DIA`, +`basal-body-temperature` → `BODY_TEMPERATURE`. These light up for free when +Google ships the data types. + +## Later waves (W5) + +`steps` → `ACTIVITY_STEPS`, `distance` → `WALKING_RUNNING_DISTANCE`, +`total-calories` → `ACTIVE_ENERGY_BURNED`, `floors` → `FLIGHTS_CLIMBED`, +`vo2-max`/`run-vo2-max` → `VO2_MAX`, `sleep` → per-stage `SLEEP_DURATION`, +`exercise` → `Workout` rows. `active-zone-minutes` is **skipped** at launch (no +slot; not blocking). + +## Idempotency + +`(userId, type, source: "FITBIT", externalId)` unique. `externalId = +:` — a re-fetch of the same window (daily summaries re-roll +after the fact, so the incremental overlap is 24 h) overwrites in place. diff --git a/src/lib/fitbit/sync-metrics.ts b/src/lib/fitbit/sync-metrics.ts new file mode 100644 index 000000000..142ec3d9c --- /dev/null +++ b/src/lib/fitbit/sync-metrics.ts @@ -0,0 +1,189 @@ +/** + * Fitbit / Google Health "health metrics & measurements" bundle sync (v1.12.0). + * + * Reads the launch metric data types from the + * `health_metrics_and_measurements.readonly` Restricted bundle and upserts each + * mapped reading as `source = FITBIT`: + * + * - weight → WEIGHT + * - body-fat → BODY_FAT + * - daily-oxygen-saturation→ OXYGEN_SATURATION + * - daily-HRV → HEART_RATE_VARIABILITY (SDNN slot; see mapper) + * - daily-resting-HR → RESTING_HEART_RATE + * - daily-respiratory-rate → RESPIRATORY_RATE + * - heart-rate (intraday) → PULSE + * - daily sleep-temp deriv.→ WRIST_TEMPERATURE + * - height → User.heightCm (profile seed, NOT a Measurement) + * + * Each data point yields at most one Measurement row, disambiguated by the + * per-point anchor + field-tag in the externalId (`:`) so a + * re-fetch of the same window overwrites in place rather than duplicating. + * + * A per-data-class 403 soft-skips THAT class (returns 0, leaves the connection + * connected) rather than parking the whole integration — the six Restricted + * bundles can be granted independently in the Google consent flow. + * + * Activity / sleep / workout land in W5; this is the first (and W3-only) resource. + */ +import { + FITBIT_DATA_TYPES, + type FitbitDataType, + type FitbitMappedMeasurement, + fetchDataPoints, + mapBodyFat, + mapHeartRate, + mapHeartRateVariability, + mapHeightCm, + mapOxygenSaturation, + mapRespiratoryRate, + mapRestingHeartRate, + mapSleepTemperature, + mapWeight, +} from "./client"; +import { + getValidToken, + handleCollectionFetchError, + incrementalStart, + markSynced, + upsertFitbitMeasurements, + type FitbitMeasurementUpsert, +} from "./sync"; +import { prisma } from "@/lib/db"; +import { getEvent } from "@/lib/logging/context"; + +/** One mappable metric: its data-type encoding + the per-point mapper + a verb. */ +interface MetricResource { + dataType: FitbitDataType; + map: (point: Record) => FitbitMappedMeasurement[]; + verb: string; +} + +/** The launch metric resources (Measurement-producing). Height is handled separately. */ +const METRIC_RESOURCES: MetricResource[] = [ + { dataType: FITBIT_DATA_TYPES.weight, map: mapWeight, verb: "fetchWeight" }, + { dataType: FITBIT_DATA_TYPES.bodyFat, map: mapBodyFat, verb: "fetchBodyFat" }, + { + dataType: FITBIT_DATA_TYPES.oxygenSaturation, + map: mapOxygenSaturation, + verb: "fetchSpo2", + }, + { + dataType: FITBIT_DATA_TYPES.heartRateVariability, + map: mapHeartRateVariability, + verb: "fetchHrv", + }, + { + dataType: FITBIT_DATA_TYPES.restingHeartRate, + map: mapRestingHeartRate, + verb: "fetchRhr", + }, + { + dataType: FITBIT_DATA_TYPES.respiratoryRate, + map: mapRespiratoryRate, + verb: "fetchRespiratoryRate", + }, + { + dataType: FITBIT_DATA_TYPES.heartRate, + map: mapHeartRate, + verb: "fetchHeartRate", + }, + { + dataType: FITBIT_DATA_TYPES.sleepTemperature, + map: mapSleepTemperature, + verb: "fetchSleepTemperature", + }, +]; + +export async function syncUserMetrics( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + const connection = await prisma.fitbitConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true }, + }); + if (!connection) return 0; + + const start = incrementalStart(connection.lastSyncedAt, { + fullSync: opts.fullSync, + }); + + let imported = 0; + + // Each metric data type is fetched + mapped independently so a per-class 403 + // soft-skips only that class. + for (const resource of METRIC_RESOURCES) { + let points: Record[]; + try { + points = await fetchDataPoints( + resource.dataType, + tokenInfo.accessToken, + resource.verb, + { start }, + ); + } catch (err) { + imported += await handleCollectionFetchError(resource.verb, userId, err); + continue; + } + + const readings: FitbitMeasurementUpsert[] = []; + for (const point of points) { + for (const m of resource.map(point)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.fieldTag, + }); + } + } + imported += await upsertFitbitMeasurements(userId, readings); + } + + // Height → User.heightCm, only when the user has no height yet. Never mint a + // Measurement; never overwrite a user-set value. Mirrors WHOOP's mapBody seed. + try { + let heightPoints: Record[]; + try { + heightPoints = await fetchDataPoints( + FITBIT_DATA_TYPES.height, + tokenInfo.accessToken, + "fetchHeight", + { start }, + ); + } catch (err) { + heightPoints = []; + await handleCollectionFetchError("fetchHeight", userId, err); + } + + // Latest non-null parsed height wins (the points are time-ordered ascending). + let heightCm: number | null = null; + for (const point of heightPoints) { + const cm = mapHeightCm(point); + if (cm !== null) heightCm = cm; + } + if (heightCm !== null) { + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { heightCm: true }, + }); + if (user && user.heightCm === null) { + await prisma.user.update({ + where: { id: userId }, + data: { heightCm }, + }); + } + } + } catch (err) { + getEvent()?.addWarning( + `fitbit: failed to seed heightCm for ${userId}: ${err}`, + ); + } + + await markSynced(userId); + return imported; +} diff --git a/src/lib/fitbit/sync.ts b/src/lib/fitbit/sync.ts index d9844c48f..1bae89550 100644 --- a/src/lib/fitbit/sync.ts +++ b/src/lib/fitbit/sync.ts @@ -13,13 +13,22 @@ * collection walkers, and the 403 soft-skip orchestration) lands in a later * wave and extends this file. */ +import { AsyncLocalStorage } from "node:async_hooks"; import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; import { encrypt, decrypt } from "@/lib/crypto"; -import { getEvent } from "@/lib/logging/context"; +import { annotate, getEvent } from "@/lib/logging/context"; import { + isReauthRequired, recordSyncFailure, + recordSyncSuccess, type FailureKind, } from "@/lib/integrations/status"; +import { + collapseToTypeDayKeys, + recomputeBucketsForMeasurement, +} from "@/lib/rollups/measurement-rollups"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; import { refreshAccessToken } from "./client"; import { getUserFitbitCredentials } from "./credentials"; import { @@ -31,6 +40,14 @@ import { /** Refresh the access token this many ms before `tokenExpiresAt`. */ const TOKEN_REFRESH_BUFFER_MS = 5 * 60 * 1000; +/** + * Overlap window for the incremental sync, in ms. Google re-rolls daily + * summaries after the fact (a daily SpO2 / HRV / RHR can finalise hours after + * the night), so the default overlap is a full 24 h to make sure the re-rolled + * row is re-fetched on the next tick. Mirrors WHOOP's recovery/sleep overlap. + */ +export const FITBIT_DEFAULT_OVERLAP_MS = 24 * 60 * 60 * 1000; + export interface FitbitTokenInfo { accessToken: string; connection: { id: string; fitbitUserId: string }; @@ -121,9 +138,232 @@ export async function getValidToken( }; } +/** + * True when a caught error is a per-data-class 403 (forbidden). The six Google + * Health Restricted bundles are granted independently in the consent flow, so a + * 403 on ONE data class is a scope gate on THAT class — soft-skip it and keep + * the connection connected rather than parking the whole integration at + * `error_reauth`. Connection-wide reauth is reserved for a 401 (token rejected) + * and for a 403 on the token-refresh / profile path (a genuine grant revoke), + * which run outside the per-resource catch blocks. Matters more here than for + * WHOOP because partial grants are likely with six independent bundles. + */ +export function isCollectionForbidden(err: unknown): boolean { + return err instanceof FitbitApiError && err.httpStatus === 403; +} + +/** + * Per-`syncUserFitbit`-cycle counter for collection-403 soft-skips. Scoped by + * AsyncLocalStorage so the count never bleeds across concurrent per-user + * pg-boss jobs. Mirrors the WHOOP `softSkipStorage` tracker. + */ +interface SoftSkipTracker { + count: number; +} +const softSkipStorage = new AsyncLocalStorage(); + +/** + * 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; + * anything else records a classified sync failure and rethrows. A soft-skip + * increments the ambient tracker so `syncUserFitbit` can refuse to stamp success + * on an all-403 grant-revoke cycle that imported nothing. + */ +export async function handleCollectionFetchError( + resource: string, + userId: string, + err: unknown, +): Promise { + if (isCollectionForbidden(err)) { + getEvent()?.addWarning( + `fitbit ${resource} sync skipped for ${userId}: collection 403 (soft-skip)`, + ); + const tracker = softSkipStorage.getStore(); + if (tracker) tracker.count += 1; + return 0; + } + await recordFitbitSyncFailure(userId, err); + throw err; +} + +/** + * Compute the incremental `start` for a resource sync. `fullSync` returns + * undefined (the backfill walks deep history without a lower bound). Otherwise + * start from `lastSyncedAt - overlap`, or 30 days back on the very first + * incremental tick. Mirrors WHOOP's `incrementalStart`. + */ +export function incrementalStart( + lastSyncedAt: Date | null, + opts: { fullSync?: boolean; overlapMs?: number } = {}, +): Date | undefined { + if (opts.fullSync) return undefined; + const overlap = opts.overlapMs ?? FITBIT_DEFAULT_OVERLAP_MS; + if (lastSyncedAt) return new Date(lastSyncedAt.getTime() - overlap); + return new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); +} + +/** + * One mapped reading destined for a `Measurement` row, with the per-point anchor + * already resolved into a full `externalId`. The per-resource syncs build these + * from the client mappers (`:`). Mirrors WHOOP's + * `WhoopMeasurementUpsert` (no `sleepStage` — sleep lands in W5). + */ +export interface FitbitMeasurementUpsert { + type: string; + value: number; + unit: string; + measuredAt: Date; + externalId: string; +} + +/** + * Upsert a batch of mapped Fitbit readings for one user and fold the rollup tier + * + invalidate status-insight caches once at the end (mirrors the WHOOP / Withings + * sync tail). Idempotent: the `(userId, type, source, externalId)` unique key + * makes a re-post (a re-fetched daily summary) overwrite in place rather than + * minting a duplicate. Returns the count of rows written. + * + * Best-effort on the rollup fold + insight invalidate — a populator hiccup never + * fails the user's sync. + */ +export async function upsertFitbitMeasurements( + userId: string, + readings: FitbitMeasurementUpsert[], +): Promise { + if (readings.length === 0) return 0; + + let imported = 0; + const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; + + for (const r of readings) { + const type = r.type as MeasurementType; + try { + await prisma.measurement.upsert({ + where: { + userId_type_source_externalId: { + userId, + type, + source: "FITBIT", + externalId: r.externalId, + }, + }, + create: { + userId, + type, + source: "FITBIT", + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + externalId: r.externalId, + }, + update: { + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + // Surface the server-side mutation to the iOS LWW reconciler. + syncVersion: { increment: 1 }, + }, + }); + touched.push({ type, measuredAt: r.measuredAt }); + imported++; + } catch (err) { + getEvent()?.addWarning(`Fitbit: failed to upsert measurement: ${err}`); + } + } + + try { + const keys = collapseToTypeDayKeys(touched); + for (const k of keys) { + await recomputeBucketsForMeasurement(userId, k.type, k.measuredAt); + } + invalidateStatusInsightsForTypes( + userId, + keys.map((k) => k.type), + ).catch((err) => { + getEvent()?.addWarning( + `fitbit: status-insight invalidate failed for ${userId}: ${err}`, + ); + }); + } catch (err) { + getEvent()?.addWarning( + `fitbit: rollup recompute failed for ${userId}: ${err}`, + ); + } + + return imported; +} + +/** Stamp `lastSyncedAt = now` after a successful resource sync. */ +export async function markSynced(userId: string): Promise { + await prisma.fitbitConnection.update({ + where: { userId }, + data: { lastSyncedAt: new Date() }, + }); +} + +/** + * Full per-user sync across every Fitbit resource. Drives the hourly poll + * catch-all, the manual `/api/fitbit/sync` trigger, and the boot-time backfill + * (`fullSync: true`). Poll-only — there is no Fitbit webhook at launch (Pub/Sub + * deferred), so there is no per-resource webhook enqueue path. + * + * Parks immediately when the connection is at `error_reauth` (the user must + * reconnect first) — returns 0, matching the WHOOP / Withings no-op contract. + * + * W3 wires the health-metrics resource; activity / sleep / workout join the + * resource list in W5. + */ +export async function syncUserFitbit( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + if (await isReauthRequired(userId, "fitbit")) { + getEvent()?.addWarning( + `fitbit sync skipped for ${userId}: parked at error_reauth`, + ); + return 0; + } + + const { syncUserMetrics } = await import("./sync-metrics"); + + const resources = [syncUserMetrics]; + + const tracker: SoftSkipTracker = { count: 0 }; + let total = 0; + let anyFailed = false; + await softSkipStorage.run(tracker, async () => { + for (const fn of resources) { + try { + total += await fn(userId, opts); + } catch (err) { + anyFailed = true; + getEvent()?.addWarning(`fitbit ${fn.name} failed for ${userId}: ${err}`); + } + } + }); + + // A genuine grant-revoke 403s EVERY collection: each resource soft-skips + // (returns 0, records no failure), so `anyFailed` stays false and `total` is + // 0 — yet the connection is dead until the token-refresh path next catches the + // 401. Don't stamp success when the whole cycle was soft-skipped and nothing + // imported; leave the status as-is so the "looks-healthy" window closes. A + // 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; + + if (!anyFailed && !allSoftSkipped) { + await recordSyncSuccess(userId, "fitbit"); + } + + annotate({ + action: { name: "fitbit.sync", details: { imported: total } }, + }); + return total; +} + /** * Map a Fitbit response classification onto a `FailureKind` and record it. - * Shared by the token-refresh path and (later) every per-resource catch block. + * Shared by the token-refresh path and every per-resource catch block. */ export async function recordFitbitSyncFailure( userId: string, diff --git a/src/lib/jobs/__tests__/fitbit-backfill.test.ts b/src/lib/jobs/__tests__/fitbit-backfill.test.ts new file mode 100644 index 000000000..2f164bc08 --- /dev/null +++ b/src/lib/jobs/__tests__/fitbit-backfill.test.ts @@ -0,0 +1,101 @@ +/** + * v1.12.0 — Fitbit backfill self-convergence tests (mocked). + * - discovery only matches un-backfilled connections; + * - a completed backfill stamps `backfillCompletedAt` so the next discovery + * pass drops the account (idempotent across reboots); + * - the discovery enqueue is singleton-keyed per user. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { prismaMock, bossSend, syncUserFitbit } = vi.hoisted(() => ({ + prismaMock: { + fitbitConnection: { + findMany: vi.fn(), + update: vi.fn(), + }, + }, + bossSend: vi.fn(), + syncUserFitbit: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ prisma: prismaMock })); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => ({ send: bossSend }), +})); + +vi.mock("@/lib/fitbit/sync", () => ({ + syncUserFitbit: (...a: unknown[]) => syncUserFitbit(...a), +})); + +vi.mock("@/lib/logging/context", () => ({ + annotate: () => {}, + getEvent: () => null, +})); + +import { + enqueueBootTimeFitbitBackfill, + runFitbitBackfillForUser, +} from "../fitbit-backfill"; + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe("enqueueBootTimeFitbitBackfill — discovery", () => { + it("queries only un-backfilled connections", async () => { + prismaMock.fitbitConnection.findMany.mockResolvedValue([ + { userId: "u1" }, + { userId: "u2" }, + ]); + bossSend.mockResolvedValue("job-id"); + + const result = await enqueueBootTimeFitbitBackfill(); + + expect(prismaMock.fitbitConnection.findMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: { backfillCompletedAt: null }, + }), + ); + expect(result.enqueued).toBe(2); + // Singleton-keyed per user so a fast restart never double-enqueues. + expect(bossSend).toHaveBeenCalledWith( + "fitbit-backfill", + expect.objectContaining({ userId: "u1" }), + expect.objectContaining({ singletonKey: "fitbit-backfill|u1" }), + ); + }); + + it("self-converges: no un-backfilled connections → nothing enqueued", async () => { + prismaMock.fitbitConnection.findMany.mockResolvedValue([]); + + const result = await enqueueBootTimeFitbitBackfill(); + + expect(result.enqueued).toBe(0); + expect(bossSend).not.toHaveBeenCalled(); + }); + + it("never throws — surfaces a discovery error through the result value", async () => { + prismaMock.fitbitConnection.findMany.mockRejectedValue(new Error("db down")); + + const result = await enqueueBootTimeFitbitBackfill(); + + expect(result.error).toBe("db down"); + expect(result.enqueued).toBe(0); + }); +}); + +describe("runFitbitBackfillForUser", () => { + it("runs a full sync and stamps backfillCompletedAt", async () => { + syncUserFitbit.mockResolvedValue(123); + prismaMock.fitbitConnection.update.mockResolvedValue({}); + + const { imported } = await runFitbitBackfillForUser("u1"); + + expect(imported).toBe(123); + 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); + }); +}); diff --git a/src/lib/jobs/__tests__/fitbit-queues.test.ts b/src/lib/jobs/__tests__/fitbit-queues.test.ts new file mode 100644 index 000000000..5049be89b --- /dev/null +++ b/src/lib/jobs/__tests__/fitbit-queues.test.ts @@ -0,0 +1,73 @@ +/** + * v1.12.0 — pg-boss queue registration for the Fitbit / Google Health sync layer. + * + * The reminder-worker imports heavy infrastructure (pg-boss, Prisma adapter, + * notification dispatchers, …) so we don't boot the whole worker here. Instead we + * read the source as text and assert the queue constants + cron schedules + + * handler wiring + allQueues + boot-enqueue are present — the fast guard that + * catches the v1.4.37 dead-queue class (a queue declared but never registered in + * `allQueues`, which silently never drains). + * + * Fitbit is poll-only at launch (no webhook — Pub/Sub deferred), so a single + * hourly `fitbit-sync` queue drives the per-user `syncUserFitbit` driver across + * every connection, alongside the backfill + the daily OAuth-state sweep. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +const REMINDER_WORKER_PATH = join(__dirname, "..", "reminder-worker.ts"); +const source = readFileSync(REMINDER_WORKER_PATH, "utf8"); + +const FITBIT_QUEUE_CONSTS = [ + "FITBIT_SYNC_QUEUE", + "FITBIT_BACKFILL_QUEUE", + "FITBIT_OAUTH_STATE_CLEANUP_QUEUE", +] as const; + +describe("reminder-worker — Fitbit sync queues", () => { + it("declares the Fitbit sync + oauth-state-cleanup queue constants", () => { + expect(source).toMatch(/FITBIT_SYNC_QUEUE\s*=\s*["']fitbit-sync["']/); + expect(source).toMatch( + /FITBIT_OAUTH_STATE_CLEANUP_QUEUE\s*=\s*["']fitbit-oauth-state-cleanup["']/, + ); + }); + + it("registers EVERY Fitbit queue in allQueues (v1.4.37 dead-queue guard)", () => { + const block = /const allQueues = \[([\s\S]*?)\];/.exec(source); + expect(block).not.toBeNull(); + const body = block![1]!; + for (const q of FITBIT_QUEUE_CONSTS) { + expect(body).toContain(q); + } + }); + + it("schedules the hourly poll cron + the daily oauth-state sweep", () => { + expect(source).toMatch(/\[FITBIT_SYNC_QUEUE,\s*FITBIT_SYNC_CRON\]/); + expect(source).toMatch( + /\[FITBIT_OAUTH_STATE_CLEANUP_QUEUE,\s*FITBIT_OAUTH_STATE_CLEANUP_CRON\]/, + ); + }); + + it("registers a boss.work handler for every Fitbit queue", () => { + expect(source).toMatch( + /boss\.work[\s\S]{0,160}FITBIT_SYNC_QUEUE[\s\S]{0,160}handleFitbitSync/, + ); + expect(source).toMatch( + /boss\.work[\s\S]{0,200}FITBIT_BACKFILL_QUEUE[\s\S]{0,200}runFitbitBackfillForUser/, + ); + expect(source).toMatch( + /boss\.work[\s\S]{0,160}FITBIT_OAUTH_STATE_CLEANUP_QUEUE[\s\S]{0,160}handleFitbitOAuthStateCleanup/, + ); + }); + + it("wires the self-converging Fitbit backfill boot discovery", () => { + expect(source).toMatch(/enqueueBootTimeFitbitBackfill\(\)/); + }); + + it("imports the Fitbit sync driver + backfill exports", () => { + expect(source).toMatch(/from\s*["']@\/lib\/fitbit\/sync["']/); + expect(source).toMatch(/from\s*["']@\/lib\/jobs\/fitbit-backfill["']/); + }); +}); diff --git a/src/lib/jobs/fitbit-backfill.ts b/src/lib/jobs/fitbit-backfill.ts index 1aed63144..4936bcc57 100644 --- a/src/lib/jobs/fitbit-backfill.ts +++ b/src/lib/jobs/fitbit-backfill.ts @@ -6,17 +6,14 @@ * across reboots (the predicate `backfill_completed_at IS NULL` drops a * connection once its backfill finishes). * - * The per-user backfill handler (`runFitbitBackfillForUser`) and the - * reminder-worker registration land in the data-sync wave; this module exports - * the queue contract + the discovery enqueue so the OAuth callback can request - * a backfill on connect. - * * The queue name MUST be registered in `allQueues` in * `src/lib/jobs/reminder-worker.ts` or pg-boss never provisions it and the boot * enqueue silently never drains (the v1.4.37 dead-queue class). */ import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { syncUserFitbit } from "@/lib/fitbit/sync"; export const FITBIT_BACKFILL_QUEUE = "fitbit-backfill"; @@ -32,6 +29,31 @@ export interface FitbitBackfillPayload { enqueuedAt: string; } +/** + * 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`. + */ +export async function runFitbitBackfillForUser( + userId: string, +): Promise<{ imported: number }> { + const imported = await syncUserFitbit(userId, { fullSync: true }); + + await prisma.fitbitConnection.update({ + where: { userId }, + data: { backfillCompletedAt: new Date() }, + }); + + annotate({ + action: { + name: "fitbit.backfill.complete", + details: { imported }, + }, + }); + return { imported }; +} + /** * Boot-time discovery. Finds every Fitbit connection not yet backfilled * (`backfill_completed_at IS NULL`) and enqueues one backfill job per account. diff --git a/src/lib/jobs/fitbit-oauth-state-cleanup.ts b/src/lib/jobs/fitbit-oauth-state-cleanup.ts new file mode 100644 index 000000000..97b1dec8e --- /dev/null +++ b/src/lib/jobs/fitbit-oauth-state-cleanup.ts @@ -0,0 +1,23 @@ +/** + * v1.12.0 — daily cleanup for the `fitbit_oauth_states` table. + * + * Mirrors `whoop-oauth-state-cleanup`. The OAuth state ledger is single-use — + * every happy-path + error-path branch in `fitbit/callback` consumes its row, so + * an abandoned row only lingers when the user closed the Google approval tab + * without bouncing back to the callback URL. The cookie + row TTL is 10 minutes, + * so once the timestamp blows the row is dead weight. + * + * Idempotent: a re-run within the same window matches zero rows the second time + * because the first pass deleted everything older than `now()`. + */ +import type { PrismaClient } from "@/generated/prisma/client"; + +export async function cleanupExpiredFitbitOAuthStates( + prisma: PrismaClient, + now: Date = new Date(), +): Promise { + const { count } = await prisma.fitbitOAuthState.deleteMany({ + where: { expiresAt: { lt: now } }, + }); + return count; +} diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index 5ac070485..0e3da5503 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -32,6 +32,15 @@ import { type WhoopBackfillPayload, } from "@/lib/jobs/whoop-backfill"; import { cleanupExpiredWhoopOAuthStates } from "@/lib/jobs/whoop-oauth-state-cleanup"; +import { syncUserFitbit } from "@/lib/fitbit/sync"; +import { + FITBIT_BACKFILL_QUEUE, + FITBIT_BACKFILL_CONCURRENCY, + runFitbitBackfillForUser, + enqueueBootTimeFitbitBackfill, + type FitbitBackfillPayload, +} from "@/lib/jobs/fitbit-backfill"; +import { cleanupExpiredFitbitOAuthStates } from "@/lib/jobs/fitbit-oauth-state-cleanup"; import { generateGeneralStatusForUser } from "@/lib/insights/general-status"; import { generateBloodPressureStatusForUser } from "@/lib/insights/blood-pressure-status"; import { generateWeightStatusForUser } from "@/lib/insights/weight-status"; @@ -301,6 +310,17 @@ const WHOOP_CYCLE_SYNC_CRON = "50 * * * *"; // every hour at :50 (poll-only) // next to the Withings sweep (03:20), inside the maintenance window. const WHOOP_OAUTH_STATE_CLEANUP_QUEUE = "whoop-oauth-state-cleanup"; const WHOOP_OAUTH_STATE_CLEANUP_CRON = "22 3 * * *"; +// v1.12.0 — Fitbit / Google Health poll-only sync. There is no Fitbit webhook +// at launch (Pub/Sub deferred), so a single hourly cron drives the per-user +// `syncUserFitbit` driver across every connection. Minute staggered off the +// WHOOP slots (:05/:20/:35/:50) and the Withings slots (:00/:15) so the hourly +// ticks don't pile up on one boss poll. +const FITBIT_SYNC_QUEUE = "fitbit-sync"; +const FITBIT_SYNC_CRON = "8 * * * *"; // every hour at :08 +// v1.12.0 — daily sweep for the Fitbit OAuth state ledger. Slots at 03:24, next +// to the WHOOP sweep (03:22), inside the maintenance window. +const FITBIT_OAUTH_STATE_CLEANUP_QUEUE = "fitbit-oauth-state-cleanup"; +const FITBIT_OAUTH_STATE_CLEANUP_CRON = "24 3 * * *"; const OFFHOST_BACKUP_QUEUE = "data-backup-offhost"; // 02:30 Europe/Berlin — runs after audit-log/idempotency cleanups so old // rows are gone before they're snapshotted, but before the existing @@ -1706,6 +1726,78 @@ async function handleWhoopOAuthStateCleanup( }); } +/** + * v1.12.0 — Fitbit poll-sync payload. Poll-only (no webhook at launch): the + * single hourly cron tick carries no `userId`, so the handler iterates every + * Fitbit connection and re-syncs each via `syncUserFitbit`. One user's + * parked-at-reauth state never starves the rest of the cohort. + */ +interface FitbitSyncPayload { + userId?: string; +} + +async function handleFitbitSync(jobs: Job[]) { + await withBackgroundEvent("job.fitbit_sync", async (evt) => { + const prisma = getWorkerPrisma(); + try { + const targets: Array<{ userId: string }> = []; + for (const job of jobs) { + if (job.data?.userId) targets.push({ userId: job.data.userId }); + } + if (targets.length === 0) { + const connections = await prisma.fitbitConnection.findMany({ + select: { userId: true }, + }); + targets.push(...connections); + } + if (targets.length === 0) return; + + let usersSynced = 0; + let measurementsImported = 0; + for (const { userId } of targets) { + try { + measurementsImported += await syncUserFitbit(userId); + usersSynced++; + } catch (err) { + evt.addWarning(`job.fitbit_sync failed for user ${userId}: ${err}`); + } + } + + evt.setBackground({ + task_name: "job.fitbit_sync", + result: { + users_synced: usersSynced, + total: targets.length, + measurements_imported: measurementsImported, + }, + }); + } catch (err) { + evt.setError(err); + recordError(); + throw err; + } + }); +} + +interface FitbitOAuthStateCleanupPayload { + triggeredAt?: string; +} + +async function handleFitbitOAuthStateCleanup( + jobs: Job[], +) { + void jobs; + await withBackgroundEvent("job.fitbit_oauth_state_cleanup", async (evt) => { + const p = getWorkerPrisma(); + try { + const deleted = await cleanupExpiredFitbitOAuthStates(p); + evt.addMeta("fitbit_oauth_state_cleanup_deleted", deleted); + } catch (err) { + evt.addWarning(`fitbit-oauth-state-cleanup failed: ${err}`); + } + }); +} + async function handleHostMetricSample(jobs: Job[]) { void jobs; await withBackgroundEvent("job.host_metric_sample", async (evt) => { @@ -2224,6 +2316,13 @@ export async function startReminderWorker() { WHOOP_BACKFILL_QUEUE, // v1.11.0 — daily sweep for the WHOOP OAuth state ledger. WHOOP_OAUTH_STATE_CLEANUP_QUEUE, + // v1.12.0 — Fitbit / Google Health poll-only sync (no webhook at launch), + // self-converging boot backfill, and the daily OAuth-state ledger sweep. + // Every queue MUST be registered here or pg-boss never provisions it and the + // schedule + boot enqueue silently no-op (the v1.4.37 dead-queue class). + FITBIT_SYNC_QUEUE, + FITBIT_BACKFILL_QUEUE, + FITBIT_OAUTH_STATE_CLEANUP_QUEUE, OFFHOST_BACKUP_QUEUE, HOST_METRIC_QUEUE, FEEDBACK_AGGREGATOR_QUEUE, @@ -2396,6 +2495,10 @@ export async function startReminderWorker() { [WHOOP_CYCLE_SYNC_QUEUE, WHOOP_CYCLE_SYNC_CRON], // v1.11.0 — daily 03:22 Europe/Berlin prune for expired WHOOP OAuth states. [WHOOP_OAUTH_STATE_CLEANUP_QUEUE, WHOOP_OAUTH_STATE_CLEANUP_CRON], + // v1.12.0 — hourly Fitbit poll (:08, staggered off WHOOP/Withings) + the + // daily 03:24 Europe/Berlin prune for expired Fitbit OAuth states. + [FITBIT_SYNC_QUEUE, FITBIT_SYNC_CRON], + [FITBIT_OAUTH_STATE_CLEANUP_QUEUE, FITBIT_OAUTH_STATE_CLEANUP_CRON], [OFFHOST_BACKUP_QUEUE, OFFHOST_BACKUP_CRON], [HOST_METRIC_QUEUE, HOST_METRIC_CRON], [FEEDBACK_AGGREGATOR_QUEUE, FEEDBACK_AGGREGATOR_CRON], @@ -2526,6 +2629,35 @@ export async function startReminderWorker() { { localConcurrency: 1 }, handleWhoopOAuthStateCleanup, ); + // v1.12.0 — Fitbit poll-sync (cron full-iteration; no webhook). Serial + // concurrency so a backfill-heavy tick never crowds the request pool. + await boss.work( + FITBIT_SYNC_QUEUE, + { localConcurrency: 1 }, + handleFitbitSync, + ); + // v1.12.0 — self-converging Fitbit backfill. The boot enqueue below sends one + // full-history sync per un-backfilled connection; this handler runs it and + // stamps `backfillCompletedAt` so the discovery query drops the account. + await boss.work( + FITBIT_BACKFILL_QUEUE, + { localConcurrency: FITBIT_BACKFILL_CONCURRENCY }, + async (jobs) => { + for (const job of jobs) { + const { userId } = job.data; + const { imported } = await runFitbitBackfillForUser(userId); + workerLog( + "info", + `[fitbit-backfill] user=${userId} imported=${imported}`, + ); + } + }, + ); + await boss.work( + FITBIT_OAUTH_STATE_CLEANUP_QUEUE, + { localConcurrency: 1 }, + handleFitbitOAuthStateCleanup, + ); await boss.work( GENERAL_STATUS_QUEUE, { localConcurrency: 1 }, @@ -3213,6 +3345,30 @@ export async function startReminderWorker() { ); } + // v1.12.0 — fire-and-forget boot discovery for the Fitbit backfill. Finds + // every Fitbit connection not yet backfilled and enqueues one full-history + // sync per account. Idempotent across reboots: a completed backfill stamps + // `backfillCompletedAt`, dropping the connection from the discovery set. + // Errors come back through the helper's result value — the worker boot never + // fails because of a backfill miss. + try { + const { enqueued, skipped, error } = await enqueueBootTimeFitbitBackfill(); + if (error) { + workerLog("error", `[fitbit-backfill] boot discovery failed: ${error}`); + } else { + workerLog( + "info", + `[fitbit-backfill] boot discovery: enqueued=${enqueued} skipped=${skipped}`, + ); + } + } catch (err) { + workerLog( + "error", + "[fitbit-backfill] boot discovery threw an unexpected error", + err, + ); + } + // v1.7.0 — fire-and-forget boot discovery for the daily-mean // consolidation pass. Finds every user holding live per-sample // high-frequency mean-type rows and enqueues one job per account. diff --git a/tests/integration/fitbit-metrics-sync.test.ts b/tests/integration/fitbit-metrics-sync.test.ts new file mode 100644 index 000000000..527fa9ce0 --- /dev/null +++ b/tests/integration/fitbit-metrics-sync.test.ts @@ -0,0 +1,194 @@ +/** + * v1.12.0 — Fitbit / Google Health metrics-sync end-to-end. + * + * Drives `syncUserMetrics` against a mocked Google Health `dataPoints.list` + * response and the real Postgres testcontainer. Asserts: + * - Each mapped data point writes one Measurement row keyed + * `(userId, type, source=FITBIT, externalId)`. + * - A second sync of the same window UPSERTS in place (no duplicates) and + * bumps `syncVersion` — the idempotency contract. + * - A DAY measurement_rollup row is folded for the touched type/day. + * - A Fitbit weight and a Withings weight on the same day both persist (the + * per-source first-write-wins contract — no ingest-time collapse for a + * server-owned source pair). + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +// Crypto reads `ENCRYPTION_KEY` lazily on first encrypt(). Pin a deterministic +// 32-byte test key before any `@/lib/crypto` import so the spec is self-contained. +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { encrypt } from "@/lib/crypto"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +// `recomputeBucketsForMeasurement` enqueues WEEK/MONTH/YEAR jobs via pg-boss; +// detach the boss so the DAY fold runs without a live queue. +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: vi.fn(() => null), +})); + +const TEST_USER_ID = "user-fitbit-metrics-sync"; + +beforeEach(async () => { + const prisma = getPrismaClient(); + await truncateAllTables(prisma); + await prisma.user.create({ + data: { + id: TEST_USER_ID, + username: "fitbit-metrics-sync", + email: "fitbit-metrics-sync@example.test", + }, + }); + await prisma.fitbitConnection.create({ + data: { + userId: TEST_USER_ID, + fitbitUserId: "gh-1", + accessToken: encrypt("access-token"), + refreshToken: encrypt("refresh-token"), + // Token valid for an hour — keeps `getValidToken` off the refresh path. + tokenExpiresAt: new Date(Date.now() + 60 * 60 * 1000), + scope: "https://www.googleapis.com/auth/googlehealth.health_metrics_and_measurements.readonly", + }, + }); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); +}); + +/** + * Stub global fetch so each metric data-type read returns its own data points. + * The metrics sync hits one URL per data type (kebab-cased in the path); match + * on the path segment and return the matching point set, or an empty page for + * any type we don't seed. + */ +function stubGoogleHealth(byPathSegment: Record) { + vi.stubGlobal( + "fetch", + vi.fn(async (url: string) => { + const match = /dataTypes\/([^/]+)\/dataPoints/.exec(url); + const seg = match?.[1] ?? ""; + const dataPoints = byPathSegment[seg] ?? []; + return { + status: 200, + json: async () => ({ dataPoints }), + }; + }), + ); +} + +describe("syncUserMetrics — integration", () => { + it("writes one Measurement row per mapped data point and folds a DAY rollup", async () => { + stubGoogleHealth({ + weight: [ + { + weight: { + kilograms: 80.5, + sample_time: { physical_time: "2026-05-10T07:00:00.000Z" }, + }, + }, + ], + "daily-resting-heart-rate": [ + { + daily_resting_heart_rate: { + beats_per_minute: 55, + date: { year: 2026, month: 5, day: 10 }, + }, + }, + ], + }); + + const { syncUserMetrics } = await import("@/lib/fitbit/sync-metrics"); + const imported = await syncUserMetrics(TEST_USER_ID); + expect(imported).toBe(2); + + const prisma = getPrismaClient(); + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, source: "FITBIT" }, + orderBy: { type: "asc" }, + }); + expect(rows).toHaveLength(2); + const weight = rows.find((r) => r.type === "WEIGHT"); + expect(weight?.value).toBe(80.5); + expect(weight?.unit).toBe("kg"); + expect(weight?.externalId).toBe("2026-05-10T07:00:00.000Z:weight"); + + // A DAY rollup folded for the touched weight day. + const rollup = await prisma.measurementRollup.findMany({ + where: { + userId: TEST_USER_ID, + type: "WEIGHT", + granularity: "DAY", + source: "FITBIT", + }, + }); + expect(rollup.length).toBeGreaterThanOrEqual(1); + }); + + it("upserts in place on a re-sync of the same window (no duplicates) and bumps syncVersion", async () => { + const point = (kg: number) => ({ + weight: { + kilograms: kg, + sample_time: { physical_time: "2026-05-10T07:00:00.000Z" }, + }, + }); + + stubGoogleHealth({ weight: [point(80.5)] }); + const { syncUserMetrics } = await import("@/lib/fitbit/sync-metrics"); + await syncUserMetrics(TEST_USER_ID); + + // Re-fetch the same window with a corrected value (same anchor → same key). + stubGoogleHealth({ weight: [point(81.2)] }); + await syncUserMetrics(TEST_USER_ID); + + const prisma = getPrismaClient(); + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, source: "FITBIT", type: "WEIGHT" }, + }); + // Exactly one row — the second sync overwrote in place. + expect(rows).toHaveLength(1); + expect(rows[0]!.value).toBe(81.2); + // syncVersion bumped on the update path. + expect(rows[0]!.syncVersion).toBeGreaterThanOrEqual(2); + }); + + it("keeps a Fitbit weight and a Withings weight on the same day as separate rows", async () => { + const prisma = getPrismaClient(); + // A pre-existing Withings weight for the same day. + await prisma.measurement.create({ + data: { + userId: TEST_USER_ID, + type: "WEIGHT", + value: 79.9, + unit: "kg", + source: "WITHINGS", + measuredAt: new Date("2026-05-10T06:30:00.000Z"), + externalId: "withings:weight:1", + }, + }); + + stubGoogleHealth({ + weight: [ + { + weight: { + kilograms: 80.5, + sample_time: { physical_time: "2026-05-10T07:00:00.000Z" }, + }, + }, + ], + }); + const { syncUserMetrics } = await import("@/lib/fitbit/sync-metrics"); + await syncUserMetrics(TEST_USER_ID); + + const rows = await prisma.measurement.findMany({ + where: { userId: TEST_USER_ID, type: "WEIGHT" }, + orderBy: { source: "asc" }, + }); + // Both sources persist — no ingest-time collapse for a server-owned pair. + expect(rows).toHaveLength(2); + expect(rows.map((r) => r.source).sort()).toEqual(["FITBIT", "WITHINGS"]); + }); +});