From fb17a4e9c6fdda13acfe05110cacb97660e54910 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Fri, 3 Jul 2026 14:13:16 +0200 Subject: [PATCH] feat(google-health): add the Google Health API transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the transport that reads a user's device data from the Google Health API (health.googleapis.com/v4): Google OAuth with PKCE and offline access, the paginated dataPoints reader with a daily-rollup fallback for the types that lack a list method, the metric/activity/sleep/workout mappers onto the shared measurement shape, and a response classifier that separates retryable errors, hard rejects, and the re-consent case. Access tokens live an hour and refresh tokens are reused, not rotated. A refresh that comes back invalid_grant raises a distinct reauth-required signal so the app can prompt a fresh consent rather than fail silently — the periodic re-consent a Testing-mode Google client needs. The four core readonly scopes ship by default; the Pixel Watch ECG and irregular-rhythm scopes stay behind an opt-in flag. Only the transport is new: the mapped-measurement output, the upsert/dedup/rollup tail, and the source ladders are the shared ones. --- .../google-health/__tests__/mappers.test.ts | 262 +++ .../__tests__/response-classifier.test.ts | 122 ++ .../google-health/__tests__/scopes.test.ts | 73 + .../__tests__/setup-docs-parity.test.ts | 76 + src/lib/google-health/client.ts | 1729 +++++++++++++++++ src/lib/google-health/credentials.ts | 41 + src/lib/google-health/mapping.md | 136 ++ src/lib/google-health/oauth-state.ts | 38 + src/lib/google-health/response-classifier.ts | 114 ++ src/lib/google-health/sync-activity.ts | 140 ++ src/lib/google-health/sync-metrics.ts | 195 ++ src/lib/google-health/sync-sleep.ts | 83 + src/lib/google-health/sync-workout.ts | 105 + src/lib/google-health/sync.ts | 729 +++++++ 14 files changed, 3843 insertions(+) create mode 100644 src/lib/google-health/__tests__/mappers.test.ts create mode 100644 src/lib/google-health/__tests__/response-classifier.test.ts create mode 100644 src/lib/google-health/__tests__/scopes.test.ts create mode 100644 src/lib/google-health/__tests__/setup-docs-parity.test.ts create mode 100644 src/lib/google-health/client.ts create mode 100644 src/lib/google-health/credentials.ts create mode 100644 src/lib/google-health/mapping.md create mode 100644 src/lib/google-health/oauth-state.ts create mode 100644 src/lib/google-health/response-classifier.ts create mode 100644 src/lib/google-health/sync-activity.ts create mode 100644 src/lib/google-health/sync-metrics.ts create mode 100644 src/lib/google-health/sync-sleep.ts create mode 100644 src/lib/google-health/sync-workout.ts create mode 100644 src/lib/google-health/sync.ts diff --git a/src/lib/google-health/__tests__/mappers.test.ts b/src/lib/google-health/__tests__/mappers.test.ts new file mode 100644 index 000000000..ce044e25f --- /dev/null +++ b/src/lib/google-health/__tests__/mappers.test.ts @@ -0,0 +1,262 @@ +/** + * Representative-payload tests for the Google Health data-point mappers. + * + * The Google value-field JSON is undocumented, so each mapper hands a small set + * of candidate shapes; these tests pin the launch-metric happy paths — the + * MeasurementType, unit, value (incl. km→m conversion), the resolved + * `measuredAt` timestamp/timezone, and the externalId `fieldTag` grain (spot + * anchor vs the `stats:`-style daily-total overwrite key vs per-sleep-stage). + */ +import { describe, expect, it } from "vitest"; + +import { + mapActiveEnergy, + mapDistance, + mapGoogleHealthSleepStage, + mapGoogleHealthSportType, + mapOxygenSaturation, + mapRestingHeartRate, + mapSleepSession, + mapSteps, + mapWeight, + mapWorkout, +} from "../client"; + +describe("mapWeight — spot sample", () => { + it("maps a kilograms sample to a WEIGHT reading anchored on its physical time", () => { + const rows = mapWeight({ + weight: { + kilograms: 72.53, + sample_time: { physical_time: "2026-06-01T08:00:00.000Z" }, + }, + }); + expect(rows).toHaveLength(1); + const r = rows[0]!; + expect(r.type).toBe("WEIGHT"); + expect(r.unit).toBe("kg"); + expect(r.value).toBe(72.53); + expect(r.measuredAt.toISOString()).toBe("2026-06-01T08:00:00.000Z"); + // Spot readings anchor on the instant → first-write-wins immutable externalId. + expect(r.fieldTag).toBe("2026-06-01T08:00:00.000Z:weight"); + expect(r.cumulativeDaily).toBeUndefined(); + }); + + it("returns nothing when no value parses", () => { + expect(mapWeight({ weight: { sample_time: {} } })).toEqual([]); + }); +}); + +describe("mapOxygenSaturation — daily civil-date summary", () => { + it("maps a nightly average to a percentage anchored on the civil day", () => { + const rows = mapOxygenSaturation({ + oxygen_saturation: { average_percentage: 97.4, date: "2026-06-01" }, + }); + expect(rows).toHaveLength(1); + const r = rows[0]!; + expect(r.type).toBe("OXYGEN_SATURATION"); + expect(r.unit).toBe("%"); + expect(r.value).toBe(97.4); + // A civil-date summary anchors at UTC midnight and keys per-day. + expect(r.measuredAt.toISOString()).toBe("2026-06-01T00:00:00.000Z"); + expect(r.fieldTag).toBe("2026-06-01:spo2"); + }); +}); + +describe("mapRestingHeartRate", () => { + it("maps beats_per_minute to a RESTING_HEART_RATE reading in bpm", () => { + const rows = mapRestingHeartRate({ + daily_resting_heart_rate: { beats_per_minute: 54, date: "2026-06-01" }, + }); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + type: "RESTING_HEART_RATE", + unit: "bpm", + value: 54, + fieldTag: "2026-06-01:rhr", + }); + }); +}); + +describe("mapSteps — daily cumulative total", () => { + it("maps an interval daily total to a cumulative ACTIVITY_STEPS row", () => { + const rows = mapSteps({ + steps: { + count: 8500, + interval: { + start_time: "2026-06-01T00:00:00.000Z", + end_time: "2026-06-01T23:59:59.000Z", + }, + }, + }); + expect(rows).toHaveLength(1); + const r = rows[0]!; + expect(r.type).toBe("ACTIVITY_STEPS"); + expect(r.unit).toBe("steps"); + expect(r.value).toBe(8500); + // Daily totals carry the overwrite-in-place grain (cumulativeDaily flag + + // per-day fieldTag) so a re-fetched day replaces rather than duplicates. + expect(r.cumulativeDaily).toBe(true); + expect(r.fieldTag).toBe("steps:2026-06-01"); + expect(r.measuredAt.toISOString()).toBe("2026-06-01T00:00:00.000Z"); + }); +}); + +describe("mapDistance — unit conversion", () => { + it("converts a kilometers daily total to metres", () => { + const rows = mapDistance({ + distance: { + kilometers: 5.2, + interval: { start_time: "2026-06-01T00:00:00.000Z" }, + }, + }); + expect(rows).toHaveLength(1); + const r = rows[0]!; + expect(r.type).toBe("WALKING_RUNNING_DISTANCE"); + expect(r.unit).toBe("m"); + expect(r.value).toBe(5200); // 5.2 km → 5200 m + expect(r.cumulativeDaily).toBe(true); + }); + + it("prefers an explicit metres field when present", () => { + const rows = mapDistance({ + distance: { + meters: 8000, + interval: { start_time: "2026-06-01T00:00:00.000Z" }, + }, + }); + expect(rows[0]!.value).toBe(8000); + }); +}); + +describe("mapActiveEnergy — preserves a legitimate zero", () => { + it("keeps a rest-day 0 kcal (a real reading, not a gap)", () => { + const rows = mapActiveEnergy({ + active_energy_burned: { + active_kilocalories: 0, + interval: { start_time: "2026-06-01T00:00:00.000Z" }, + }, + }); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + type: "ACTIVE_ENERGY_BURNED", + unit: "kcal", + value: 0, + cumulativeDaily: true, + }); + }); +}); + +describe("mapSleepSession — per-stage segments", () => { + it("emits one SLEEP_DURATION row per known stage, skipping unknown labels", () => { + const rows = mapSleepSession({ + sleep: { + interval: { end_time: "2026-06-02T06:30:00.000Z" }, + stages: [ + { + stage: "deep", + start_time: "2026-06-02T02:00:00.000Z", + end_time: "2026-06-02T02:45:00.000Z", + }, + { + stage: "rem", + start_time: "2026-06-02T03:00:00.000Z", + end_time: "2026-06-02T03:30:00.000Z", + }, + { + stage: "totally_unknown_label", + start_time: "2026-06-02T04:00:00.000Z", + end_time: "2026-06-02T04:10:00.000Z", + }, + ], + }, + }); + expect(rows).toHaveLength(2); + + const deep = rows[0]!; + expect(deep.type).toBe("SLEEP_DURATION"); + expect(deep.unit).toBe("minutes"); + expect(deep.value).toBe(45); + expect(deep.sleepStage).toBe("DEEP"); + expect(deep.measuredAt.toISOString()).toBe("2026-06-02T02:45:00.000Z"); + // Per-stage rows key off the session anchor + an indexed stage tag. + expect(deep.fieldTag).toBe("2026-06-02T06:30:00.000Z:sleep_deep:0"); + + const rem = rows[1]!; + expect(rem.value).toBe(30); + expect(rem.sleepStage).toBe("REM"); + expect(rem.fieldTag).toBe("2026-06-02T06:30:00.000Z:sleep_rem:1"); + }); + + it("maps a 'light' stage onto the shared CORE band", () => { + expect(mapGoogleHealthSleepStage("light")).toBe("CORE"); + expect(mapGoogleHealthSleepStage("REM")).toBe("REM"); + expect(mapGoogleHealthSleepStage("in-bed")).toBe("IN_BED"); + expect(mapGoogleHealthSleepStage("nonsense")).toBeNull(); + }); +}); + +describe("mapWorkout — exercise session", () => { + it("maps a running session to a Workout with duration, energy, distance and HR", () => { + const w = mapWorkout({ + exercise: { + activity_type: "running", + session_id: "abc123", + interval: { + start_time: "2026-06-03T07:00:00.000Z", + end_time: "2026-06-03T07:45:00.000Z", + }, + active_kilocalories: 420, + distance: { meters: 8000 }, + average_heart_rate: { beats_per_minute: 150 }, + maximum_heart_rate: { beats_per_minute: 172 }, + }, + }); + expect(w).not.toBeNull(); + expect(w).toMatchObject({ + externalId: "abc123", + sportType: "running", + durationSec: 2700, // 45 min + totalEnergyKcal: 420, + totalDistanceM: 8000, + avgHeartRate: 150, + maxHeartRate: 172, + minHeartRate: null, + }); + expect(w!.startedAt.toISOString()).toBe("2026-06-03T07:00:00.000Z"); + expect(w!.endedAt.toISOString()).toBe("2026-06-03T07:45:00.000Z"); + }); + + it("falls back to a start-anchored externalId when no session id is present", () => { + const w = mapWorkout({ + exercise: { + type: "walk", + interval: { + start_time: "2026-06-03T09:00:00.000Z", + end_time: "2026-06-03T09:30:00.000Z", + }, + }, + }); + expect(w!.externalId).toBe("exercise:2026-06-03T09:00:00.000Z"); + expect(w!.sportType).toBe("walking"); + }); + + it("returns null for a session with no usable time span", () => { + expect(mapWorkout({ exercise: {} })).toBeNull(); + expect( + mapWorkout({ + exercise: { + interval: { + start_time: "2026-06-03T09:00:00.000Z", + end_time: "2026-06-03T09:00:00.000Z", // zero-length + }, + }, + }), + ).toBeNull(); + }); + + it("resolves unknown sport types to a generic label", () => { + expect(mapGoogleHealthSportType("kitesurfing")).toBe("other"); + expect(mapGoogleHealthSportType("")).toBe("other"); + expect(mapGoogleHealthSportType("Cycling")).toBe("cycling"); + }); +}); diff --git a/src/lib/google-health/__tests__/response-classifier.test.ts b/src/lib/google-health/__tests__/response-classifier.test.ts new file mode 100644 index 000000000..5b2a5ab70 --- /dev/null +++ b/src/lib/google-health/__tests__/response-classifier.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, it } from "vitest"; + +import { + GoogleHealthApiError, + classifyGoogleHealthError, + classifyGoogleHealthResponse, + isGoogleHealthInvalidGrant, + isGoogleHealthReauthRequired, +} from "../response-classifier"; + +describe("classifyGoogleHealthResponse", () => { + it("classifies 2xx as success", () => { + for (const status of [200, 201, 204, 299]) { + const v = classifyGoogleHealthResponse(status); + expect(v.classification).toBe("success"); + expect(v.httpStatus).toBe(status); + } + }); + + it("classifies 429 as transient (rate-limited — honour Google backoff)", () => { + const v = classifyGoogleHealthResponse(429); + expect(v.classification).toBe("transient"); + expect(v.reason).toBe("http_429"); + }); + + it("classifies 5xx as transient", () => { + for (const status of [500, 502, 503, 504]) { + expect(classifyGoogleHealthResponse(status).classification).toBe( + "transient", + ); + } + }); + + it("classifies 401 and 403 as reauth_required", () => { + expect(classifyGoogleHealthResponse(401).classification).toBe( + "reauth_required", + ); + expect(classifyGoogleHealthResponse(403).classification).toBe( + "reauth_required", + ); + }); + + it("classifies other 4xx as persistent (hard reject)", () => { + for (const status of [400, 404, 422]) { + expect(classifyGoogleHealthResponse(status).classification).toBe( + "persistent", + ); + } + }); +}); + +describe("classifyGoogleHealthError", () => { + it("reads the verdict off a typed GoogleHealthApiError", () => { + const err = new GoogleHealthApiError({ + verb: "fetchWeight", + classification: "transient", + httpStatus: 503, + reason: "http_503", + }); + expect(classifyGoogleHealthError(err)).toBe("transient"); + }); + + it("defaults a plain Error to transient (never hard-disables on an unknown)", () => { + expect(classifyGoogleHealthError(new Error("socket hang up"))).toBe( + "transient", + ); + }); + + it("recovers the verdict from the legacy message shape across a lost prototype", () => { + // A pg-boss retry can strip the prototype; the message-shape fallback still + // recovers a 401 → reauth_required. + const plain = new Error("GoogleHealth fetchSleep error: 401"); + expect(classifyGoogleHealthError(plain)).toBe("reauth_required"); + }); +}); + +describe("isGoogleHealthInvalidGrant / isGoogleHealthReauthRequired", () => { + it("lifts a 400 + invalid_grant onto reauth_required (the 7-day expiry signal)", () => { + const err = new GoogleHealthApiError({ + verb: "refreshToken", + classification: "persistent", // a bare 400 classifies persistent… + httpStatus: 400, + reason: "http_400", + upstreamError: "invalid_grant", + }); + expect(isGoogleHealthInvalidGrant(err)).toBe(true); + // …but the reauth predicate still routes it to the reconnect CTA. + expect(isGoogleHealthReauthRequired(err)).toBe(true); + }); + + it("keeps a 400 invalid_client as persistent (bad secret ≠ reconnect)", () => { + const err = new GoogleHealthApiError({ + verb: "refreshToken", + classification: "persistent", + httpStatus: 400, + reason: "http_400", + upstreamError: "invalid_client", + }); + expect(isGoogleHealthInvalidGrant(err)).toBe(false); + expect(isGoogleHealthReauthRequired(err)).toBe(false); + }); + + it("treats a 401 as reauth_required", () => { + const err = new GoogleHealthApiError({ + verb: "fetchProfile", + classification: "reauth_required", + httpStatus: 401, + reason: "http_401", + }); + expect(isGoogleHealthReauthRequired(err)).toBe(true); + }); + + it("does not flag a transient 503 as reauth", () => { + const err = new GoogleHealthApiError({ + verb: "fetchSteps", + classification: "transient", + httpStatus: 503, + reason: "http_503", + }); + expect(isGoogleHealthReauthRequired(err)).toBe(false); + }); +}); diff --git a/src/lib/google-health/__tests__/scopes.test.ts b/src/lib/google-health/__tests__/scopes.test.ts new file mode 100644 index 000000000..8c69cf462 --- /dev/null +++ b/src/lib/google-health/__tests__/scopes.test.ts @@ -0,0 +1,73 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { + GOOGLE_HEALTH_CORE_SCOPES, + GOOGLE_HEALTH_EXPERIMENTAL_SCOPES, + experimentalScopesEnabled, + getGoogleHealthScopeString, + resolveGoogleHealthScopes, +} from "../client"; + +describe("Google Health scope resolver", () => { + const KEY = "GOOGLE_HEALTH_EXPERIMENTAL_SCOPES"; + let prev: string | undefined; + + beforeEach(() => { + prev = process.env[KEY]; + delete process.env[KEY]; + }); + afterEach(() => { + if (prev === undefined) delete process.env[KEY]; + else process.env[KEY] = prev; + }); + + it("ships exactly the four core Restricted read scopes by default", () => { + delete process.env[KEY]; + const scopes = resolveGoogleHealthScopes(); + expect(scopes).toEqual([...GOOGLE_HEALTH_CORE_SCOPES]); + expect(scopes).toHaveLength(4); + // Every core scope is a Restricted read-only googlehealth scope. + for (const s of scopes) { + expect(s).toMatch( + /^https:\/\/www\.googleapis\.com\/auth\/googlehealth\..+\.readonly$/, + ); + } + }); + + it("keeps the ECG/IRN experimental scopes off unless opted in", () => { + delete process.env[KEY]; + expect(experimentalScopesEnabled()).toBe(false); + const scopes = resolveGoogleHealthScopes(); + for (const ex of GOOGLE_HEALTH_EXPERIMENTAL_SCOPES) { + expect(scopes).not.toContain(ex); + } + }); + + it('adds ECG + IRN when the flag is "true" or "1"', () => { + for (const flag of ["true", "1"]) { + process.env[KEY] = flag; + expect(experimentalScopesEnabled()).toBe(true); + const scopes = resolveGoogleHealthScopes(); + expect(scopes).toEqual([ + ...GOOGLE_HEALTH_CORE_SCOPES, + ...GOOGLE_HEALTH_EXPERIMENTAL_SCOPES, + ]); + expect(scopes).toHaveLength(6); + } + }); + + it("treats any other flag value as off", () => { + for (const flag of ["", "false", "yes", "0"]) { + process.env[KEY] = flag; + expect(experimentalScopesEnabled()).toBe(false); + } + }); + + it("joins the resolved scopes with a single space for the authorize request", () => { + delete process.env[KEY]; + expect(getGoogleHealthScopeString()).toBe( + GOOGLE_HEALTH_CORE_SCOPES.join(" "), + ); + expect(getGoogleHealthScopeString().split(" ")).toHaveLength(4); + }); +}); diff --git a/src/lib/google-health/__tests__/setup-docs-parity.test.ts b/src/lib/google-health/__tests__/setup-docs-parity.test.ts new file mode 100644 index 000000000..4f295e2db --- /dev/null +++ b/src/lib/google-health/__tests__/setup-docs-parity.test.ts @@ -0,0 +1,76 @@ +/** + * Doc ↔ code parity guard for the Google Health operator runbook. + * + * The setup-guide link on the Settings card sends a mid-setup user to + * `docs.healthlog.dev/integrations/google-health`, rendered from + * `docs/integrations/google-health.md`. If the client ever requests a scope the + * runbook doesn't list, or the walkthrough regresses back to a wrong provider + * (the file was once a Fitbit-Web-API walkthrough), a self-hoster follows a + * broken recipe and the OAuth consent screen rejects the connection. + * + * This test pins the doc to the code: every core scope the client requests must + * appear verbatim in the runbook, alongside the real Google OAuth endpoints, the + * fixed callback path, and the Google Cloud console setup. It also asserts the + * provider key is wired into the setup-guide link so the card actually points + * here. + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; + +import { GOOGLE_HEALTH_CORE_SCOPES } from "../client"; +import { + INTEGRATION_DOCS_BASE, + integrationDocsHref, +} from "@/components/settings/integrations/setup-guide-link"; + +const ROOT = join(__dirname, "../../../.."); +const DOC_PATH = join(ROOT, "docs/integrations/google-health.md"); +const doc = readFileSync(DOC_PATH, "utf8"); + +describe("google-health setup runbook parity", () => { + it("mentions every core Restricted scope the client requests", () => { + for (const scope of GOOGLE_HEALTH_CORE_SCOPES) { + expect(doc).toContain(scope); + } + }); + + it("documents the real Google OAuth endpoints", () => { + expect(doc).toContain("https://accounts.google.com/o/oauth2/v2/auth"); + expect(doc).toContain("https://oauth2.googleapis.com/token"); + }); + + it("documents the fixed callback path", () => { + expect(doc).toContain("/api/google-health/callback"); + }); + + it("walks the Google Cloud console setup (not the retired Fitbit walkthrough)", () => { + expect(doc).toContain("console.cloud.google.com"); + expect(doc.toLowerCase()).toContain("google health api"); + // Guard against the pre-rewrite Fitbit-developer-console walkthrough + // creeping back in. + expect(doc).not.toContain("dev.fitbit.com"); + expect(doc).not.toContain("api.fitbit.com"); + }); + + it("documents the honest data + re-consent caveats", () => { + const lower = doc.toLowerCase(); + expect(lower).toContain("stress"); + expect(lower).toContain("readiness"); + expect(doc).toContain("7-day"); + expect(doc).toContain("GOOGLE_HEALTH_REDIRECT_URI"); + }); + + it("names the 100-user / CASA verification ceiling", () => { + expect(doc).toContain("100 users"); + expect(doc).toContain("CASA"); + }); + + it("resolves the setup-guide href for the google-health provider", () => { + // The literal argument also compile-checks that "google-health" is a member + // of the IntegrationDocsProvider union. + expect(integrationDocsHref("google-health")).toBe( + `${INTEGRATION_DOCS_BASE}/google-health`, + ); + }); +}); diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts new file mode 100644 index 000000000..3a6f5c69e --- /dev/null +++ b/src/lib/google-health/client.ts @@ -0,0 +1,1729 @@ +/** + * Google Health API client — OAuth + data-point reads (v1.27.0). + * Docs: https://developers.google.com/health (health.googleapis.com/v4, the + * successor to the Fitbit Web API, unifying Fitbit + Pixel Watch + Fitbit Air). + * + * Hand-rolled fetch over `safeFetch` (no SDK): the OAuth handshake + * (`getAuthorizationUrl` / `exchangeCode` / `refreshAccessToken`), a single + * profile fetch for the connection's external user id, and the paginated + * `dataPoints.list` walker + per-type mappers. + * + * KEY OAUTH SEMANTICS (verified 2026 contract): + * - Access-token TTL = 1 h (contrast the classic Fitbit Web API's 8 h). + * - Refresh tokens do NOT rotate — a routine refresh returns a fresh + * `access_token` WITHOUT a `refresh_token`; the sync layer keeps the stored + * one (see `getValidToken` in `sync.ts`). Refresh tokens are time-based: + * they expire after 6 months of disuse, or — in a consent screen still in + * "Testing" publishing mode — after 7 DAYS, at which point the user must + * re-consent. A revoked / expired refresh token surfaces `invalid_grant` + * (or a 401) on the token endpoint; `postToken` lifts that onto the + * `reauth_required` class so the connection prompts a reconnect rather than + * a generic hard error. + * - PKCE (S256): the authorize request carries a `code_challenge`; the + * callback presents the matching `code_verifier` on exchange. Google's + * web-server (confidential) client also sends the client secret via HTTP + * Basic auth (RFC 6749 §2.3.1) — Basic + PKCE together. + * - `access_type=offline` + `prompt=consent` are required to reliably receive + * a refresh token (and force one on every re-consent). + */ +import { createHash, randomBytes } from "node:crypto"; +import { getEvent } from "@/lib/logging/context"; +import { safeFetch } from "@/lib/safe-fetch"; +import { + GoogleHealthApiError, + classifyGoogleHealthResponse, + type GoogleHealthClassification, +} from "./response-classifier"; + +export const GOOGLE_HEALTH_API_BASE = "https://health.googleapis.com/v4"; +const GOOGLE_OAUTH_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"; +const GOOGLE_OAUTH_TOKEN_URL = "https://oauth2.googleapis.com/token"; + +export interface GoogleHealthCredentials { + clientId: string; + clientSecret: string; +} + +/** + * Resolve the OAuth `redirect_uri`, then assert it against an allowlist + * (defence-in-depth). + * + * The value is operator-controlled config (`GOOGLE_HEALTH_REDIRECT_URI`, else + * derived from `NEXT_PUBLIC_APP_URL`), not user input, so Google's + * registered-redirect check is the real backstop. But a misconfigured or + * `Host`-coerced `NEXT_PUBLIC_APP_URL` (a mis-deployed reverse proxy reflecting + * a forwarded Host) would otherwise send the authorization code's landing URL + * off-origin. Pin the target here so a malformed origin fails fast at the + * handshake rather than silently redirecting elsewhere: + * - must be an absolute, parseable URL, + * - must be https (the one exception is a localhost/loopback dev host, which + * Google itself permits over http), + * - must land on the fixed `/api/google-health/callback` path, + * - when derived from `NEXT_PUBLIC_APP_URL`, must stay same-origin with it. + */ +function getRedirectUri(): string { + const explicit = process.env.GOOGLE_HEALTH_REDIRECT_URI; + const appUrl = process.env.NEXT_PUBLIC_APP_URL; + const raw = + explicit ?? (appUrl ? `${appUrl}/api/google-health/callback` : undefined); + + if (!raw) { + throw new Error( + "Google Health redirect_uri is not configured — set GOOGLE_HEALTH_REDIRECT_URI or NEXT_PUBLIC_APP_URL", + ); + } + + let parsed: URL; + try { + parsed = new URL(raw); + } catch { + throw new Error( + `Google Health redirect_uri is not an absolute URL: ${raw}`, + ); + } + + const isLoopback = + parsed.hostname === "localhost" || + parsed.hostname === "127.0.0.1" || + parsed.hostname === "[::1]" || + parsed.hostname === "::1"; + if ( + parsed.protocol !== "https:" && + !(parsed.protocol === "http:" && isLoopback) + ) { + throw new Error( + `Google Health redirect_uri must be https (or http on localhost): ${parsed.origin}`, + ); + } + + if (parsed.pathname !== "/api/google-health/callback") { + throw new Error( + `Google Health redirect_uri must target /api/google-health/callback, got ${parsed.pathname}`, + ); + } + + // When an explicit GOOGLE_HEALTH_REDIRECT_URI is set alongside + // NEXT_PUBLIC_APP_URL, require them to share an origin so the pinned value + // can't drift to an unexpected host relative to the app's own base URL. + if (explicit && appUrl) { + let appOrigin: string; + try { + appOrigin = new URL(appUrl).origin; + } catch { + throw new Error(`NEXT_PUBLIC_APP_URL is not an absolute URL: ${appUrl}`); + } + if (parsed.origin !== appOrigin) { + throw new Error( + `Google Health redirect_uri origin ${parsed.origin} does not match NEXT_PUBLIC_APP_URL origin ${appOrigin}`, + ); + } + } + + return parsed.toString(); +} + +/** + * The four core Restricted read scopes HealthLog requests for v1. Every scope + * is Restricted → the operator's OAuth client needs Google verification + an + * annual CASA assessment before it leaves "Testing" publishing mode (staying in + * Testing with ≤100 test users avoids CASA at the cost of a 7-day refresh-token + * expiry). + */ +export const GOOGLE_HEALTH_CORE_SCOPES = [ + "https://www.googleapis.com/auth/googlehealth.activity_and_fitness.readonly", + "https://www.googleapis.com/auth/googlehealth.health_metrics_and_measurements.readonly", + "https://www.googleapis.com/auth/googlehealth.sleep.readonly", + "https://www.googleapis.com/auth/googlehealth.profile.readonly", +] as const; + +/** + * Optional Pixel-Watch clinical scopes (ECG + irregular-rhythm notifications). + * Kept defined but OFF by default: they widen the Restricted-scope review + * surface and only apply to Pixel Watch sources, so they are opt-in behind the + * `GOOGLE_HEALTH_EXPERIMENTAL_SCOPES` env flag until they land cleanly. Both are + * read-only categories (no `.writeonly` variant exists upstream). + */ +export const GOOGLE_HEALTH_EXPERIMENTAL_SCOPES = [ + "https://www.googleapis.com/auth/googlehealth.ecg.readonly", + "https://www.googleapis.com/auth/googlehealth.irn.readonly", +] as const; + +/** + * True when the operator has opted into the experimental ECG/IRN scopes via + * `GOOGLE_HEALTH_EXPERIMENTAL_SCOPES=true`. Read lazily so tests can toggle it + * per case and so a deploy can flip it without a rebuild. + */ +export function experimentalScopesEnabled(): boolean { + const raw = process.env.GOOGLE_HEALTH_EXPERIMENTAL_SCOPES; + return raw === "true" || raw === "1"; +} + +/** + * Resolve the scope list HealthLog requests: the four core Restricted bundles, + * plus the ECG/IRN bundles only when the experimental flag is set. + */ +export function resolveGoogleHealthScopes(): string[] { + return experimentalScopesEnabled() + ? [...GOOGLE_HEALTH_CORE_SCOPES, ...GOOGLE_HEALTH_EXPERIMENTAL_SCOPES] + : [...GOOGLE_HEALTH_CORE_SCOPES]; +} + +/** The space-separated scope string sent on the authorize request. */ +export function getGoogleHealthScopeString(): string { + return resolveGoogleHealthScopes().join(" "); +} + +// ─── PKCE ────────────────────────────────────────────────────── +// +// Google's authorization-code flow accepts S256 PKCE. The verifier is a +// high-entropy random string; the challenge is BASE64URL(SHA256(verifier)). The +// verifier is stashed on the OAuth-state row at connect and presented on the +// token exchange at callback — never in the cookie or the URL. + +export interface GoogleHealthPkcePair { + verifier: string; + challenge: string; +} + +/** + * Mint a PKCE verifier + S256 challenge. 64 random bytes → 86 base64url chars, + * comfortably inside the RFC 7636 43–128 char verifier range and well past the + * 256-bit entropy floor. + */ +export function generatePkcePair(): GoogleHealthPkcePair { + const verifier = randomBytes(64).toString("base64url"); + const challenge = createHash("sha256").update(verifier).digest("base64url"); + return { verifier, challenge }; +} + +/** + * Build the Google authorization URL (a browser redirect, not a fetch). `state` + * is the opaque CSRF nonce minted by `oauth-state.ts`; `codeChallenge` is the + * S256 PKCE challenge whose verifier the callback presents on token exchange. + * + * `access_type=offline` + `prompt=consent` are Google's requirement to receive + * a refresh token; `prompt=consent` forces the consent screen so a re-connect + * always returns a fresh refresh token even if the user previously granted. + */ +export function getAuthorizationUrl( + state: string, + creds: GoogleHealthCredentials, + codeChallenge: string, +): string { + const params = new URLSearchParams({ + response_type: "code", + client_id: creds.clientId, + redirect_uri: getRedirectUri(), + scope: getGoogleHealthScopeString(), + access_type: "offline", + prompt: "consent", + code_challenge: codeChallenge, + code_challenge_method: "S256", + state, + }); + return `${GOOGLE_OAUTH_AUTH_URL}?${params}`; +} + +export interface GoogleHealthTokenResponse { + access_token: string; + /** + * Present on the initial code exchange and whenever Google issues a new + * refresh token; ABSENT on a routine refresh because Google does not rotate. + * The sync layer keeps the stored token when this is missing. + */ + refresh_token?: string; + expires_in: number; + scope?: string; + token_type?: string; +} + +/** Basic-auth header carrying the confidential client credentials. */ +function basicAuthHeader(creds: GoogleHealthCredentials): string { + const raw = `${creds.clientId}:${creds.clientSecret}`; + return `Basic ${Buffer.from(raw, "utf8").toString("base64")}`; +} + +async function postToken( + params: URLSearchParams, + creds: GoogleHealthCredentials, + verb: string, +): Promise { + const start = performance.now(); + const res = await safeFetch(GOOGLE_OAUTH_TOKEN_URL, { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + Authorization: basicAuthHeader(creds), + }, + body: params.toString(), + }); + + const json = await res.json().catch(() => null); + const verdict = classifyGoogleHealthResponse(res.status); + const upstreamError = + typeof json?.error === "string" ? (json.error as string) : undefined; + + // Google signals a revoked / expired refresh token — the user disconnected + // the app upstream, OR the 7-day "Testing"-mode refresh window lapsed — via + // `invalid_grant` on the token endpoint (sometimes a 400, sometimes a 401). A + // bare status classification buckets a 400 as `persistent` (never prompts a + // reconnect); lift an `invalid_grant` onto `reauth_required` so the connection + // surfaces the reconnect CTA. A 401 already classifies as `reauth_required`. + const classification: GoogleHealthClassification = + upstreamError === "invalid_grant" + ? "reauth_required" + : verdict.classification; + + getEvent()?.addExternalCall({ + service: "google-health", + method: verb, + duration_ms: Math.round(performance.now() - start), + status: res.status, + error: verdict.classification === "success" ? undefined : verdict.reason, + }); + if (classification !== "success") { + throw new GoogleHealthApiError({ + verb, + classification, + httpStatus: verdict.httpStatus, + reason: verdict.reason, + upstreamError, + }); + } + return json as GoogleHealthTokenResponse; +} + +/** + * Exchange an authorization code for the initial token pair, presenting the + * PKCE verifier. `redirect_uri` must exactly match the one sent to the + * authorize endpoint. + */ +export async function exchangeCode( + code: string, + codeVerifier: string, + creds: GoogleHealthCredentials, +): Promise { + return postToken( + new URLSearchParams({ + grant_type: "authorization_code", + code, + client_id: creds.clientId, + redirect_uri: getRedirectUri(), + code_verifier: codeVerifier, + }), + creds, + "exchangeCode", + ); +} + +/** + * Refresh an expired access token. Google does NOT rotate refresh tokens — the + * response carries a fresh `access_token` + `expires_in` but usually omits + * `refresh_token`. The caller persists the new access token + expiry and keeps + * the stored refresh token unless a new one is returned. The original scope is + * preserved by Google, so no `scope` param is re-sent. A revoked / expired + * refresh token throws a `GoogleHealthApiError` classified `reauth_required`. + */ +export async function refreshAccessToken( + refreshToken: string, + creds: GoogleHealthCredentials, +): Promise { + return postToken( + new URLSearchParams({ + grant_type: "refresh_token", + refresh_token: refreshToken, + client_id: creds.clientId, + }), + creds, + "refreshAccessToken", + ); +} + +/** + * The Google Health profile carries the external user id used as the + * connection's `externalUserId`. A single GET, not a paginated collection. + */ +export interface GoogleHealthProfile { + /** + * External user identifier. Google returns `me`-relative profile data; the + * stable id is surfaced under `name` (a `users/{id}` resource name) or `id` + * depending on the API surface — both are captured here and resolved at the + * call site. Re-verify the exact field against a live account at build. + */ + name?: string; + id?: string; +} + +export async function fetchProfile( + accessToken: string, +): Promise { + const start = performance.now(); + const res = await safeFetch(`${GOOGLE_HEALTH_API_BASE}/users/me/profile`, { + method: "GET", + headers: { Authorization: `Bearer ${accessToken}` }, + }); + const json = (await res + .json() + .catch(() => null)) as GoogleHealthProfile | null; + const verdict = classifyGoogleHealthResponse(res.status); + getEvent()?.addExternalCall({ + service: "google-health", + method: "fetchProfile", + duration_ms: Math.round(performance.now() - start), + status: res.status, + error: verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new GoogleHealthApiError({ + verb: "fetchProfile", + classification: verdict.classification, + httpStatus: verdict.httpStatus, + reason: verdict.reason, + }); + } + return (json ?? {}) as GoogleHealthProfile; +} + +/** + * Resolve the stable external user id from a Google Health profile. The + * `users/{id}` resource name is the canonical anchor; fall back to a bare `id` + * or "me" so the connection always persists a non-empty `externalUserId`. + */ +export function resolveGoogleHealthUserId( + profile: GoogleHealthProfile, +): string { + if (profile.name) { + const tail = profile.name.split("/").pop(); + if (tail) return tail; + } + if (profile.id) return profile.id; + return "me"; +} + +// ─── Data-point reads (Google Health `dataPoints.list` / `:dailyRollUp`) ────── +// +// 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. Google's value-field JSON is NOT fully +// published, 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 `mapping.md` against a live test account at +// build and tighten the extractors then. +// +// Casing gotcha: the data-type id is **kebab-case in the path** (`body-fat`) +// and **snake_case in the `filter`** (`body_fat`). `GOOGLE_HEALTH_DATA_TYPES` +// pins both forms so a fetcher can never encode the wrong one. +// +// READ METHOD: most data types support `:list`. Two types — +// `total-calories` and `calories-in-heart-rate-zone` — support ONLY the roll-up +// surface (no `:list`); they carry `readMethod: "dailyRollUp"` and route through +// the single-day roll-up walker instead. Every other type stays on `:list`. + +/** + * 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. The launch + * metrics use the daily/spot reads, so the default page size is the larger. + */ +export const GOOGLE_HEALTH_PAGE_SIZE = 1000; +/** Sleep/exercise read cap — matches the Google Health 25-cap for those types. */ +export const GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE = 25; + +/** + * One data type's 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. `timeField` names the value object's + * time anchor. `readMethod` selects the read surface (`list` by default; + * `dailyRollUp` for the two roll-up-only types). + */ +export interface GoogleHealthDataType { + /** 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 keyed on a civil date (`{type}.date`). + * - `interval` → an INTERVAL data type (steps / distance / calories / + * floors, sleep, exercise), anchored on `{type}.interval.start_time` + * (the physical instant) with `{type}.interval.civil_start_time` as the + * civil fallback — NOT a `sample_time` (which 400s/empties for interval + * types and stalls the incremental filter) and NOT a bare `date`. + */ + timeField: "sample" | "date" | "interval"; + /** + * Which read surface to use. Defaults to `list`. `total-calories` and + * `calories-in-heart-rate-zone` are roll-up-only (no `:list`) and MUST use + * `dailyRollUp` — the list walker refuses them. + */ + readMethod?: "list" | "dailyRollUp"; +} + +/** + * The launch data types. Each entry pins the kebab-path + snake-filter pair so + * the two encodings can never drift. Identifiers verified against the 2026 + * Google Health contract (API-RESEARCH §3/§4). + */ +export const GOOGLE_HEALTH_DATA_TYPES = { + weight: { path: "weight", filter: "weight", timeField: "sample" }, + bodyFat: { path: "body-fat", filter: "body_fat", timeField: "sample" }, + oxygenSaturation: { + path: "oxygen-saturation", + filter: "oxygen_saturation", + timeField: "date", + }, + heartRateVariability: { + path: "heart-rate-variability", + filter: "heart_rate_variability", + timeField: "date", + }, + restingHeartRate: { + path: "daily-resting-heart-rate", + filter: "daily_resting_heart_rate", + timeField: "date", + }, + respiratoryRate: { + path: "respiratory-rate", + filter: "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", + }, + // ── Activity bundle — daily cumulative totals ────────────────── + // Scope: `googlehealth.activity_and_fitness.readonly`. These are INTERVAL + // data types: Google buckets a daily total into an `interval` (a `start_time` + // + `end_time`, with `civil_start_time` for the calendar-day grain). The + // incremental filter targets `interval.start_time` and the externalId carries + // the `stats:` prefix so a re-fetched day overwrites in place (mirrors the + // Apple-Health `stats::` daily-total overwrite contract). + steps: { path: "steps", filter: "steps", timeField: "interval" }, + distance: { path: "distance", filter: "distance", timeField: "interval" }, + // Active energy — canonical id `active-energy-burned` (release-notes + // 2026-05-26); this is the ACTIVE portion only, NOT `total-calories` (which + // folds in BMR and is roll-up-only, see below). + activeEnergy: { + path: "active-energy-burned", + filter: "active_energy_burned", + timeField: "interval", + }, + floors: { path: "floors", filter: "floors", timeField: "interval" }, + // VO2 max is a daily-summary metric (one civil-date reading), not an interval + // bucket — keep the `date` anchor. + vo2Max: { path: "vo2-max", filter: "vo2_max", timeField: "date" }, + // ── Roll-up-only types (no `:list`) ─────────────────────────── + // `total-calories` and `calories-in-heart-rate-zone` support ONLY the roll-up + // surface. Defined here so the read layer routes them through `:dailyRollUp` + // and never accidentally `:list`s them; not yet wired to a Measurement mapper + // (active energy already covers the active-calories slot; total-calories folds + // in BMR and needs a modelling decision before it lands as a metric). + totalCalories: { + path: "total-calories", + filter: "total_calories", + timeField: "interval", + readMethod: "dailyRollUp", + }, + caloriesInHeartRateZone: { + path: "calories-in-heart-rate-zone", + filter: "calories_in_heart_rate_zone", + timeField: "interval", + readMethod: "dailyRollUp", + }, + // ── Sleep bundle ─────────────────────────────────────────────── + // Scope: `googlehealth.sleep.readonly`. A sleep session is an INTERVAL data + // type (a start + end span carrying a per-stage breakdown); the incremental + // filter must target `interval.start_time`, not a `sample_time` (which 400s + // for interval types). Mapped to per-stage SLEEP_DURATION rows. + sleep: { path: "sleep", filter: "sleep", timeField: "interval" }, + // ── Exercise bundle ──────────────────────────────────────────── + // Scope: `googlehealth.activity_and_fitness.readonly`. An exercise session is + // an INTERVAL data type (a start + end span) → a `Workout` row (NOT a + // Measurement). The incremental filter targets `interval.start_time`. + exercise: { path: "exercise", filter: "exercise", timeField: "interval" }, +} as const satisfies Record; + +/** Google Health `DataPoint` — value object is type-keyed + carries a time anchor. */ +export interface GoogleHealthDataPoint { + [key: string]: unknown; +} + +/** `dataPoints.list` envelope: `{ dataPoints, nextPageToken }`. */ +interface GoogleHealthDataPointPage { + dataPoints?: GoogleHealthDataPoint[]; + nextPageToken?: string | null; +} + +/** + * `dataPoints:dailyRollUp` envelope. The response shape is not fully published; + * both `dataPoints` and `rollupDataPoints` are tolerated so a live-verified + * shape doesn't require a code change here. (OPEN — confirm at build.) + */ +interface GoogleHealthDailyRollUpPage { + dataPoints?: GoogleHealthDataPoint[]; + rollupDataPoints?: GoogleHealthDataPoint[]; + nextPageToken?: string | null; +} + +interface DataPointQuery { + /** Lower-bound incremental cursor; omitted on a full backfill. */ + start?: Date; + /** Page size (defaults to `GOOGLE_HEALTH_PAGE_SIZE`). */ + pageSize?: number; + /** Hard ceiling on pages walked (defence against a runaway cursor). */ + maxPages?: number; + /** Hard ceiling on days walked for a `:dailyRollUp` read. */ + maxDays?: number; +} + +/** Default backfill horizon (days) when a roll-up read has no incremental start. */ +const DAILY_ROLLUP_BACKFILL_DAYS = 90; +const MS_PER_DAY = 24 * 60 * 60 * 1000; + +/** + * Build the incremental `filter` field + bound for one data type. Centralised + * so the predicate and the read-time anchor resolution can never drift. + * - `sample` → spot reading, filter on `{filter}.sample_time.physical_time`. + * - `interval` → INTERVAL type, filter on `{filter}.interval.start_time`. + * - `date` → daily civil-date summary, filter on `{filter}.date`. + */ +export function incrementalFilter( + dataType: GoogleHealthDataType, + start: Date, +): { field: string; bound: string } { + switch (dataType.timeField) { + case "sample": + return { + field: `${dataType.filter}.sample_time.physical_time`, + bound: start.toISOString(), + }; + case "interval": + return { + field: `${dataType.filter}.interval.start_time`, + bound: start.toISOString(), + }; + case "date": + return { + field: `${dataType.filter}.date`, + bound: start.toISOString().slice(0, 10), + }; + } +} + +/** + * Walk every `DataPoint` for one data type since the incremental cursor. + * Routes by `readMethod`: `list` types walk `dataPoints.list` with + * `nextPageToken` pagination; the two roll-up-only types walk `:dailyRollUp` + * one day at a time. 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: GoogleHealthDataType, + accessToken: string, + verb: string, + query: DataPointQuery = {}, +): Promise { + if (dataType.readMethod === "dailyRollUp") { + return fetchDailyRollUp(dataType, accessToken, verb, query); + } + + const points: GoogleHealthDataPoint[] = []; + let pageToken: string | null | undefined; + let pageCount = 0; + const maxPages = query.maxPages ?? 1000; + const pageSize = query.pageSize ?? GOOGLE_HEALTH_PAGE_SIZE; + + do { + const params = new URLSearchParams({ pageSize: String(pageSize) }); + if (query.start) { + const { field, bound } = incrementalFilter(dataType, query.start); + params.set("filter", `${field} >= "${bound}"`); + } + if (pageToken) params.set("pageToken", pageToken); + + const pageStart = performance.now(); + const res = await safeFetch( + `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`, + { + method: "GET", + headers: { Authorization: `Bearer ${accessToken}` }, + }, + ); + + const json = (await res + .json() + .catch(() => null)) as GoogleHealthDataPointPage | null; + const verdict = classifyGoogleHealthResponse(res.status); + getEvent()?.addExternalCall({ + service: "google-health", + 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 GoogleHealthApiError({ + 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; +} + +/** + * Walk `dataPoints:dailyRollUp` one civil day at a time for a roll-up-only data + * type (`total-calories`, `calories-in-heart-rate-zone`). Those types have no + * `:list`; the daily-summary endpoint returns one aggregate per day. Bounded by + * `maxDays` (default 90) and by the incremental `start` when present. + * + * The exact request/response shape of `:dailyRollUp` is not fully published + * (OPEN — confirm at build); this reads defensively and tolerates either a + * `dataPoints` or a `rollupDataPoints` array so a live-verified shape needs no + * code change. The per-day `date` filter is the documented single-day selector. + */ +export async function fetchDailyRollUp( + dataType: GoogleHealthDataType, + accessToken: string, + verb: string, + query: DataPointQuery = {}, +): Promise { + const points: GoogleHealthDataPoint[] = []; + const maxDays = query.maxDays ?? DAILY_ROLLUP_BACKFILL_DAYS; + const now = Date.now(); + const startMs = query.start + ? query.start.getTime() + : now - maxDays * MS_PER_DAY; + + let dayCount = 0; + for ( + let cursor = startMs; + cursor <= now && dayCount < maxDays; + cursor += MS_PER_DAY, dayCount += 1 + ) { + const day = new Date(cursor).toISOString().slice(0, 10); + const params = new URLSearchParams({ + // The daily-summary selector: the single civil day this roll-up covers. + filter: `${dataType.filter}.date = "${day}"`, + }); + + const dayStart = performance.now(); + const res = await safeFetch( + `${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints:dailyRollUp?${params}`, + { + method: "GET", + headers: { Authorization: `Bearer ${accessToken}` }, + }, + ); + + const json = (await res + .json() + .catch(() => null)) as GoogleHealthDailyRollUpPage | null; + const verdict = classifyGoogleHealthResponse(res.status); + getEvent()?.addExternalCall({ + service: "google-health", + method: `${verb}(day=${dayCount})`, + duration_ms: Math.round(performance.now() - dayStart), + status: res.status, + error: verdict.classification === "success" ? undefined : verdict.reason, + }); + if (verdict.classification !== "success") { + throw new GoogleHealthApiError({ + verb, + classification: verdict.classification, + httpStatus: verdict.httpStatus, + reason: verdict.reason, + }); + } + + for (const p of json?.dataPoints ?? json?.rollupDataPoints ?? []) { + points.push(p); + } + } + + return points; +} + +// ─── Field → Measurement mapping ─────────────────────────────── +// The single source of truth is `mapping.md` — keep both in sync when adding +// entries. + +/** Metres → centimetres (Google `height` → `User.heightCm`). */ +const M_TO_CM = 100; + +function round2(n: number): number { + return parseFloat(n.toFixed(2)); +} + +/** Finite + strictly-positive guard. */ +function positive(n: unknown): n is number { + return typeof n === "number" && Number.isFinite(n) && n > 0; +} + +/** + * A single mapped reading destined for one `Measurement` row. The `source` + * (`GOOGLE_HEALTH`) and `externalId` (`:`) are stamped by the + * sync layer; the mapper emits only type/value/unit/measuredAt + the field-tag + * that disambiguates the externalId. + */ +export interface GoogleHealthMappedMeasurement { + type: string; + value: number; + unit: string; + measuredAt: Date; + /** Disambiguator appended to the per-point anchor to form the externalId. */ + fieldTag: string; + /** Per-stage sleep rows carry the SleepStage; everything else omits it. */ + sleepStage?: GoogleHealthSleepStage; + /** + * `true` when the externalId carries the `stats:` daily-total prefix (the + * cumulative activity metrics). The sync layer stamps the externalId; this + * flag lets it pick the right shape (`stats::` vs the + * `:` spot shape) without re-deriving the grain. + */ + cumulativeDaily?: boolean; +} + +/** HealthLog `SleepStage` values a Google sleep stage maps onto. */ +export type GoogleHealthSleepStage = + "IN_BED" | "AWAKE" | "ASLEEP" | "REM" | "CORE" | "DEEP"; + +/** + * 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: GoogleHealthDataPoint, + 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; +} + +/** Parse a `{year,month,day}` civil-date object into a UTC-midday Date, or null. */ +function parseCivilDateObject(val: unknown): Date | null { + if (!val || typeof val !== "object") return null; + const o = val 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 null; +} + +/** + * Resolve a `DataPoint`'s measurement timestamp. + * - `sample` → `sample_time.physical_time` (a spot ISO instant). + * - `interval` → `interval.start_time` (the bucket's start ISO instant), with + * `interval.civil_start_time` (a civil string / `{year,month,day}`) as the + * fallback. + * - `date` → `date` (a civil string or `{year,month,day}` object). + * Falls back to `fallback` only when nothing parses, so a row is never dropped + * for a missing anchor. + */ +function resolveMeasuredAt( + point: GoogleHealthDataPoint, + dataType: GoogleHealthDataType, + 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 if (dataType.timeField === "interval") { + const t = readPath(point, `${dataType.filter}.interval.start_time`); + if (typeof t === "string") { + const d = new Date(t); + if (!Number.isNaN(d.getTime())) return d; + } + const civil = readPath( + point, + `${dataType.filter}.interval.civil_start_time`, + ); + if (typeof civil === "string") { + const d = new Date(civil); + if (!Number.isNaN(d.getTime())) return d; + } + const civilObj = parseCivilDateObject(civil); + if (civilObj) return civilObj; + } else { + const dateVal = readPath(point, `${dataType.filter}.date`); + if (typeof dateVal === "string") { + const d = new Date(dateVal); + if (!Number.isNaN(d.getTime())) return d; + } + const civilObj = parseCivilDateObject(dateVal); + if (civilObj) return civilObj; + } + 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. + */ +function externalAnchor( + point: GoogleHealthDataPoint, + dataType: GoogleHealthDataType, +): string { + const at = resolveMeasuredAt(point, dataType, new Date(0)); + // The mapper-driven interval types (steps/distance/calories/floors) are daily + // totals, so they share the civil-day externalId grain with `date` summaries — + // a re-fetched day overwrites in place. (Sleep/exercise are interval too but + // mint their own per-session anchors and never reach this helper.) + if (dataType.timeField === "date" || dataType.timeField === "interval") { + 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. + */ +function mapSimple( + point: GoogleHealthDataPoint, + dataType: GoogleHealthDataType, + spec: { + type: string; + unit: string; + fieldTag: string; + valuePaths: string[]; + factor?: number; + }, +): GoogleHealthMappedMeasurement[] { + 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: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.weight; + return mapSimple(point, dt, { + type: "WEIGHT", + unit: "kg", + fieldTag: "weight", + valuePaths: valuePaths(dt.filter, "kilograms"), + }); +} + +export function mapBodyFat( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.bodyFat; + return mapSimple(point, dt, { + type: "BODY_FAT", + unit: "%", + fieldTag: "body_fat", + valuePaths: valuePaths(dt.filter, "percentage"), + }); +} + +export function mapOxygenSaturation( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_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: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.heartRateVariability; + // Google 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: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_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: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_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: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.heartRate; + return mapSimple(point, dt, { + type: "PULSE", + unit: "bpm", + fieldTag: "hr", + valuePaths: valuePaths(dt.filter, "beats_per_minute"), + }); +} + +export function mapSleepTemperature( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_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 Google `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. Returns cm. + */ +export function mapHeightCm(point: GoogleHealthDataPoint): number | null { + const dt = GOOGLE_HEALTH_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 + * 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 GOOGLE_HEALTH_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 Google Health", + }, + bodyFat: { + type: "BODY_FAT", + unit: "%", + path: "body-fat", + filter: "body_fat", + }, + oxygenSaturation: { + type: "OXYGEN_SATURATION", + unit: "%", + path: "oxygen-saturation", + filter: "oxygen_saturation", + }, + heartRateVariability: { + type: "HEART_RATE_VARIABILITY", + unit: "ms", + path: "heart-rate-variability", + filter: "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: "respiratory-rate", + filter: "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", + }, + // ── Activity bundle — daily cumulative ────────────────────────── + steps: { + type: "ACTIVITY_STEPS", + unit: "steps", + path: "steps", + filter: "steps", + note: "daily total; stats: externalId overwrites on re-fetch; 0 is a valid rest day", + }, + distance: { + type: "WALKING_RUNNING_DISTANCE", + unit: "m", + path: "distance", + filter: "distance", + note: "daily total metres; stats: externalId overwrites on re-fetch", + }, + activeEnergy: { + type: "ACTIVE_ENERGY_BURNED", + unit: "kcal", + path: "active-energy-burned", + filter: "active_energy_burned", + note: "ACTIVE portion only (NOT total-calories, which folds in BMR); stats: externalId overwrites on re-fetch", + }, + floors: { + type: "FLIGHTS_CLIMBED", + unit: "flights", + path: "floors", + filter: "floors", + note: "daily total floors; stats: externalId overwrites on re-fetch", + }, + vo2Max: { + type: "VO2_MAX", + unit: "mL/(kg·min)", + path: "vo2-max", + filter: "vo2_max", + note: "daily latest-wins; daily-anchor externalId overwrites on re-fetch", + }, + sleep: { + type: "SLEEP_DURATION", + unit: "minutes", + path: "sleep", + filter: "sleep", + note: "per-stage rows (IN_BED/AWAKE/REM/CORE/DEEP); measuredAt = stage END", + }, + exercise: { + type: "Workout", + unit: "—", + path: "exercise", + filter: "exercise", + note: "exercise session → Workout row (NOT a Measurement); cross-source dedup at read time", + }, +}; + +// ─── Activity mappers: daily cumulative ──────────────────────── + +/** Finite + non-negative guard — cumulative metrics admit a legitimate 0. */ +function nonNegative(n: unknown): n is number { + return typeof n === "number" && Number.isFinite(n) && n >= 0; +} + +/** + * Pull the first finite NON-negative number out of a list of candidate value + * paths. Unlike `firstNumber` (strictly positive), this admits a legitimate + * zero — a day of rest still records 0 steps / 0 floors / 0 active kcal, and + * dropping the zero would leave a hole the chart misreads as missing data. + */ +function firstNonNegativeNumber( + point: GoogleHealthDataPoint, + paths: string[], +): number | null { + for (const path of paths) { + const v = readPath(point, path); + if (nonNegative(v)) return v; + } + return null; +} + +/** + * Map one daily cumulative-activity data point into a single Measurement + * reading. The externalId is the `stats:`-prefixed daily-total shape so a + * re-fetched day overwrites in place rather than minting a duplicate — the same + * overwrite contract the Apple-Health `stats::` daily totals + * use. A zero is preserved (a rest day is real data, not a gap). `latestWins` + * (VO2 max) carries the same daily anchor; it is daily latest-wins rather than a + * running sum, but the per-day overwrite key is identical. + */ +function mapDailyCumulative( + point: GoogleHealthDataPoint, + dataType: GoogleHealthDataType, + spec: { + type: string; + unit: string; + fieldTag: string; + valuePaths: string[]; + factor?: number; + /** VO2 max is daily latest-wins, not a running total — still one row/day. */ + latestWins?: boolean; + }, +): GoogleHealthMappedMeasurement[] { + // VO2 max is strictly positive; the running totals admit a legitimate zero. + let value = spec.latestWins + ? firstNumber(point, spec.valuePaths) + : firstNonNegativeNumber(point, spec.valuePaths); + if (value === null) return []; + if (spec.factor) value = value * spec.factor; + const dayKey = externalAnchor(point, dataType); // YYYY-MM-DD for a daily type + return [ + { + type: spec.type, + value: round2(value), + unit: spec.unit, + measuredAt: resolveMeasuredAt(point, dataType, new Date()), + // `stats::` — the sync layer reads `cumulativeDaily` + // to assemble the externalId, matching the Apple-Health daily-total shape. + fieldTag: `${spec.fieldTag}:${dayKey}`, + cumulativeDaily: true, + }, + ]; +} + +export function mapSteps( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.steps; + return mapDailyCumulative(point, dt, { + type: "ACTIVITY_STEPS", + unit: "steps", + fieldTag: "steps", + valuePaths: [ + ...valuePaths(dt.filter, "count"), + ...valuePaths(dt.filter, "steps"), + ], + }); +} + +export function mapDistance( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.distance; + // Google may report metres or kilometres; prefer the explicit-metres field, + // else convert km → m. Both pass the non-negative guard. + const meters = firstNonNegativeNumber(point, valuePaths(dt.filter, "meters")); + if (meters !== null) { + return [ + { + type: "WALKING_RUNNING_DISTANCE", + value: round2(meters), + unit: "m", + measuredAt: resolveMeasuredAt(point, dt, new Date()), + fieldTag: `distance:${externalAnchor(point, dt)}`, + cumulativeDaily: true, + }, + ]; + } + return mapDailyCumulative(point, dt, { + type: "WALKING_RUNNING_DISTANCE", + unit: "m", + fieldTag: "distance", + valuePaths: valuePaths(dt.filter, "kilometers"), + factor: 1000, + }); +} + +export function mapActiveEnergy( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.activeEnergy; + // ACTIVE energy only — NOT total-calories (which folds in BMR). The candidate + // paths target the active-portion field explicitly. + return mapDailyCumulative(point, dt, { + type: "ACTIVE_ENERGY_BURNED", + unit: "kcal", + fieldTag: "active_energy", + valuePaths: [ + ...valuePaths(dt.filter, "active_kilocalories"), + ...valuePaths(dt.filter, "kilocalories"), + ...valuePaths(dt.filter, "calories"), + ], + }); +} + +export function mapFloors( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.floors; + return mapDailyCumulative(point, dt, { + type: "FLIGHTS_CLIMBED", + unit: "flights", + fieldTag: "floors", + valuePaths: [ + ...valuePaths(dt.filter, "count"), + ...valuePaths(dt.filter, "floors"), + ], + }); +} + +export function mapVo2Max( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const dt = GOOGLE_HEALTH_DATA_TYPES.vo2Max; + return mapDailyCumulative(point, dt, { + type: "VO2_MAX", + unit: "mL/(kg·min)", + fieldTag: "vo2_max", + valuePaths: [ + ...valuePaths(dt.filter, "milliliters_per_kilogram_per_minute"), + ...valuePaths(dt.filter, "value"), + ], + latestWins: true, // strictly positive, one daily reading; not a running sum + }); +} + +// ── Sleep ────────────────────────────────────────────────────── +// +// A Google Health sleep session carries a list of per-stage segments, each with +// a stage label + a start + end. HealthLog stores one SLEEP_DURATION row per +// stage segment with `measuredAt = stage END` (so the night-total + hypnogram +// readers consume the same enum WHOOP / Apple write). The stage labels are +// harmonised onto the shared `SleepStage` enum. + +/** Google Health sleep stage label → HealthLog `SleepStage`. */ +const GOOGLE_HEALTH_SLEEP_STAGE_MAP: Record = { + // Canonical Google Health stage names (snake / lower variants accepted). + in_bed: "IN_BED", + inbed: "IN_BED", + awake: "AWAKE", + wake: "AWAKE", + light: "CORE", // "light" ↔ Apple "core" (same shallow-NREM band) + core: "CORE", + rem: "REM", + deep: "DEEP", + // The classic (non-stages) sleep log uses asleep/restless/wake. + asleep: "ASLEEP", + restless: "AWAKE", +} as const; + +/** + * Normalise a raw Google sleep-stage label to a `SleepStage`, or null for an + * unknown label (skipped rather than mis-bucketed). + */ +export function mapGoogleHealthSleepStage( + raw: unknown, +): GoogleHealthSleepStage | null { + if (typeof raw !== "string") return null; + const key = raw + .trim() + .toLowerCase() + .replace(/[\s-]+/g, "_"); + return ( + GOOGLE_HEALTH_SLEEP_STAGE_MAP[key] ?? + GOOGLE_HEALTH_SLEEP_STAGE_MAP[key.replace(/_/g, "")] ?? + null + ); +} + +/** One sleep-stage segment pulled defensively off a Google sleep session. */ +interface GoogleHealthSleepSegment { + stage: string; + startTime?: string; + endTime?: string; +} + +/** Minutes between two ISO instants, or null if either is unparseable. */ +function minutesBetween(startIso?: string, endIso?: string): number | null { + if (!startIso || !endIso) return null; + const s = new Date(startIso).getTime(); + const e = new Date(endIso).getTime(); + if (Number.isNaN(s) || Number.isNaN(e) || e <= s) return null; + return (e - s) / 60_000; +} + +/** + * Read the per-stage segments off a Google sleep `DataPoint`, tolerating the + * undocumented JSON shape. Candidate locations: `sleep.segments`, + * `sleep.stages`, `sleep.levels.data`, or a bare top-level `segments`/`stages`. + */ +function readSleepSegments( + point: GoogleHealthDataPoint, +): GoogleHealthSleepSegment[] { + const candidates = [ + readPath(point, "sleep.segments"), + readPath(point, "sleep.stages"), + readPath(point, "sleep.levels.data"), + readPath(point, "segments"), + readPath(point, "stages"), + readPath(point, "levels.data"), + ]; + for (const c of candidates) { + if (!Array.isArray(c)) continue; + const out: GoogleHealthSleepSegment[] = []; + for (const raw of c) { + if (!raw || typeof raw !== "object") continue; + const o = raw as Record; + const stage = + (typeof o.stage === "string" && o.stage) || + (typeof o.level === "string" && o.level) || + (typeof o.type === "string" && o.type) || + ""; + const startTime = + (typeof o.startTime === "string" && o.startTime) || + (typeof o.start_time === "string" && o.start_time) || + (typeof o.dateTime === "string" && o.dateTime) || + undefined; + const endTime = + (typeof o.endTime === "string" && o.endTime) || + (typeof o.end_time === "string" && o.end_time) || + undefined; + if (stage) out.push({ stage, startTime, endTime }); + } + if (out.length > 0) return out; + } + return []; +} + +/** + * The stable session anchor for a sleep `DataPoint`'s externalId. The session + * end (or start) ISO instant is unique per night, so per-stage rows key as + * `:sleep_` — a re-scored night overwrites in place. + */ +function sleepSessionAnchor(point: GoogleHealthDataPoint): string { + const end = + readPath(point, "sleep.interval.end_time") ?? + readPath(point, "interval.end_time") ?? + readPath(point, "sleep.endTime") ?? + readPath(point, "sleep.end_time") ?? + readPath(point, "endTime") ?? + readPath(point, "end_time"); + if (typeof end === "string") { + const d = new Date(end); + if (!Number.isNaN(d.getTime())) return d.toISOString(); + } + const start = + readPath(point, "sleep.interval.start_time") ?? + readPath(point, "interval.start_time") ?? + readPath(point, "sleep.startTime") ?? + readPath(point, "sleep.start_time") ?? + readPath(point, "startTime"); + if (typeof start === "string") { + const d = new Date(start); + if (!Number.isNaN(d.getTime())) return d.toISOString(); + } + return new Date(0).toISOString(); +} + +/** + * Map one Google sleep session into per-SEGMENT `SLEEP_DURATION` rows. The + * Google sleep payload carries a real per-stage segment series (each with its + * own start/end), so one row is emitted PER SEGMENT — `measuredAt = that + * segment's END` — rather than collapsing a stage's segments onto a single + * lastEnd instant. The timeline is MEASURED (real onsets), so these rows are NOT + * flagged reconstructed — unlike WHOOP, which has no onsets. + * + * Each segment carries a stage-scoped, INDEXED fieldTag so the several segments + * of one stage stay distinct under the `(userId, type, source, externalId)` + * dedup key. Unknown stage labels are skipped; a session with no parseable + * segment yields nothing. + */ +export function mapSleepSession( + point: GoogleHealthDataPoint, +): GoogleHealthMappedMeasurement[] { + const segments = readSleepSegments(point); + if (segments.length === 0) return []; + + const anchor = sleepSessionAnchor(point); + const out: GoogleHealthMappedMeasurement[] = []; + let segIndex = 0; + for (const seg of segments) { + const stage = mapGoogleHealthSleepStage(seg.stage); + if (!stage) continue; + const mins = minutesBetween(seg.startTime, seg.endTime); + if (mins === null || !(mins > 0)) continue; + const end = new Date(seg.endTime as string); + out.push({ + type: "SLEEP_DURATION", + value: round2(mins), + unit: "minutes", + measuredAt: end, + fieldTag: `${anchor}:sleep_${stage.toLowerCase()}:${segIndex}`, + sleepStage: stage, + }); + segIndex += 1; + } + return out; +} + +// ── Workouts (exercise sessions) ─────────────────────────────── + +/** + * Google Health exercise-activity-type → HealthLog `WorkoutSportType`. Unknown + * types fall through to a generic label; the column is free-text so an unmapped + * type still persists (just not under a canonical sport bucket). + */ +const GOOGLE_HEALTH_EXERCISE_TYPE_MAP: Record = { + walk: "walking", + walking: "walking", + run: "running", + running: "running", + treadmill: "running", + bike: "cycling", + biking: "cycling", + cycling: "cycling", + spinning: "cycling", + hike: "hiking", + hiking: "hiking", + swim: "swimming", + swimming: "swimming", + rowing: "rowing", + elliptical: "elliptical", + stairclimber: "stairClimber", + yoga: "yoga", + pilates: "mindAndBody", + weights: "strength", + strength: "strength", + workout: "strength", + hiit: "hiit", + interval_workout: "hiit", + dance: "dance", + golf: "golf", + tennis: "tennis", + basketball: "basketball", + soccer: "soccer", + bootcamp: "crossTraining", + circuit_training: "crossTraining", + sport: "mixedCardio", +} as const; + +/** Resolve a Google exercise activity type to a canonical sport label. */ +export function mapGoogleHealthSportType(raw: unknown): string { + if (typeof raw !== "string" || raw.trim() === "") return "other"; + const key = raw + .trim() + .toLowerCase() + .replace(/[\s-]+/g, "_"); + return ( + GOOGLE_HEALTH_EXERCISE_TYPE_MAP[key] ?? + GOOGLE_HEALTH_EXERCISE_TYPE_MAP[key.replace(/_/g, "")] ?? + "other" + ); +} + +/** One mapped Google exercise session destined for a `Workout` row. */ +export interface GoogleHealthMappedWorkout { + externalId: string; + sportType: string; + startedAt: Date; + endedAt: Date; + durationSec: number; + totalEnergyKcal: number | null; + totalDistanceM: number | null; + avgHeartRate: number | null; + maxHeartRate: number | null; + minHeartRate: number | null; +} + +/** Read a finite number off a list of candidate paths (any sign), or null. */ +function readNumber( + point: GoogleHealthDataPoint, + paths: string[], +): number | null { + for (const path of paths) { + const v = readPath(point, path); + if (typeof v === "number" && Number.isFinite(v)) return v; + } + return null; +} + +/** Read an ISO/string instant off a list of candidate paths, or null. */ +function readInstant( + point: GoogleHealthDataPoint, + paths: string[], +): Date | null { + for (const path of paths) { + const v = readPath(point, path); + if (typeof v === "string") { + const d = new Date(v); + if (!Number.isNaN(d.getTime())) return d; + } + } + return null; +} + +/** + * Map one Google exercise session `DataPoint` into a `Workout` shape. Returns + * null when there is no usable start/end (a session with no time span is not a + * workout). The externalId anchors on the session id when present, else on the + * start instant, so a re-fetch overwrites the same `Workout` row in place. + * Energy is the active session energy in kcal; HR fields are optional. + */ +export function mapWorkout( + point: GoogleHealthDataPoint, +): GoogleHealthMappedWorkout | null { + const f = GOOGLE_HEALTH_DATA_TYPES.exercise.filter; + const startedAt = readInstant(point, [ + `${f}.interval.start_time`, + `${f}.startTime`, + `${f}.start_time`, + `${f}.sample_time.physical_time`, + "interval.start_time", + "startTime", + "start_time", + ]); + const endedAt = readInstant(point, [ + `${f}.interval.end_time`, + `${f}.endTime`, + `${f}.end_time`, + "interval.end_time", + "endTime", + "end_time", + ]); + if (!startedAt || !endedAt || endedAt <= startedAt) return null; + + const durationSec = Math.round( + (endedAt.getTime() - startedAt.getTime()) / 1000, + ); + + const sessionId = + (typeof readPath(point, `${f}.session_id`) === "string" && + (readPath(point, `${f}.session_id`) as string)) || + (typeof readPath(point, `${f}.id`) === "string" && + (readPath(point, `${f}.id`) as string)) || + (typeof readPath(point, "name") === "string" && + (readPath(point, "name") as string)) || + null; + const externalId = sessionId ?? `exercise:${startedAt.toISOString()}`; + + const sportRaw = + readPath(point, `${f}.activity_type`) ?? + readPath(point, `${f}.exercise_type`) ?? + readPath(point, `${f}.type`) ?? + readPath(point, "activityType"); + + const energyKcal = readNumber(point, [ + `${f}.active_kilocalories`, + `${f}.calories`, + `${f}.energy.kilocalories`, + ]); + const distanceM = readNumber(point, [ + `${f}.distance.meters`, + `${f}.distance_meters`, + `${f}.distance`, + ]); + const avgHr = readNumber(point, [ + `${f}.average_heart_rate.beats_per_minute`, + `${f}.average_heart_rate`, + `${f}.heart_rate.average`, + ]); + const maxHr = readNumber(point, [ + `${f}.maximum_heart_rate.beats_per_minute`, + `${f}.max_heart_rate`, + `${f}.heart_rate.maximum`, + ]); + const minHr = readNumber(point, [ + `${f}.minimum_heart_rate.beats_per_minute`, + `${f}.min_heart_rate`, + `${f}.heart_rate.minimum`, + ]); + + return { + externalId, + sportType: mapGoogleHealthSportType(sportRaw), + startedAt, + endedAt, + durationSec, + totalEnergyKcal: energyKcal !== null ? Math.round(energyKcal) : null, + totalDistanceM: + distanceM !== null && distanceM >= 0 ? round2(distanceM) : null, + avgHeartRate: avgHr !== null && avgHr > 0 ? Math.round(avgHr) : null, + maxHeartRate: maxHr !== null && maxHr > 0 ? Math.round(maxHr) : null, + minHeartRate: minHr !== null && minHr > 0 ? Math.round(minHr) : null, + }; +} diff --git a/src/lib/google-health/credentials.ts b/src/lib/google-health/credentials.ts new file mode 100644 index 000000000..c39a6a4b8 --- /dev/null +++ b/src/lib/google-health/credentials.ts @@ -0,0 +1,41 @@ +/** + * Helper to retrieve per-user Google Health OAuth credentials from the database. + * Mirrors `src/lib/whoop/credentials.ts`. + * + * Google Health ships per-user BYO-keys: each self-hoster registers their own + * Google Cloud OAuth client and pastes the client id/secret into Settings, + * stored encrypted on `User` (the Restricted-scope verification + annual CASA + * assessment is per-OAuth-client, so a single shared app is unworkable for a + * multi-operator product). + */ +import { prisma } from "@/lib/db"; +import { decrypt } from "@/lib/crypto"; +import type { GoogleHealthCredentials } from "./client"; + +/** + * Fetch and decrypt the user's Google Health API credentials. + * Returns null if the user has not configured credentials. + */ +export async function getUserGoogleHealthCredentials( + userId: string, +): Promise { + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { + googleHealthClientIdEncrypted: true, + googleHealthClientSecretEncrypted: true, + }, + }); + + if ( + !user?.googleHealthClientIdEncrypted || + !user?.googleHealthClientSecretEncrypted + ) { + return null; + } + + return { + clientId: decrypt(user.googleHealthClientIdEncrypted), + clientSecret: decrypt(user.googleHealthClientSecretEncrypted), + }; +} diff --git a/src/lib/google-health/mapping.md b/src/lib/google-health/mapping.md new file mode 100644 index 000000000..6619aaef4 --- /dev/null +++ b/src/lib/google-health/mapping.md @@ -0,0 +1,136 @@ +# Google Health field → Measurement mapping + +Single source of truth for the Google Health API v4 data-type → HealthLog +`Measurement` mapping. Keep this in sync with `GOOGLE_HEALTH_FIELD_MAP` and the +per-type mappers in `client.ts`. Every row ingests server-side with +`source = GOOGLE_HEALTH` and `externalId = :` (the anchor is the +data point's sample time for spot readings, its civil date for daily summaries, +or its `interval.start_time` for INTERVAL types — steps/distance/calories/floors, +sleep, exercise; the field-tag disambiguates the metric). + +Google Health is a **separate, coexisting provider** from the classic Fitbit +integration (`src/lib/fitbit/*`, `source = FITBIT`). They never share a token, +connection row, source enum, or cookie. + +## Time anchor by grain + +Google Health exposes three time-anchor shapes. The incremental `filter` and the +read-time `measuredAt`/externalId resolution must target the right one or the +incremental sync stalls (a `sample_time` filter 400s/empties for an INTERVAL type): + +- **sample** — spot reading; anchor `{filter}.sample_time.physical_time`. +- **date** — daily-summary metric (SpO2, HRV, RHR, respiratory rate, wrist temp, + VO2 max); anchor `{filter}.date` (a civil date / `{year,month,day}`). +- **interval** — INTERVAL type (steps/distance/calories/floors daily totals, plus + sleep + exercise sessions); anchor `{filter}.interval.start_time` (a physical + instant) with `{filter}.interval.civil_start_time` as the civil fallback. + +Transport docs: `developers.google.com/health` (v4). 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. + +## Read method + +Most data types support `dataPoints.list` (`readMethod: "list"`, the default). +Two types — `total-calories` and `calories-in-heart-rate-zone` — have **no +`:list`** and are read via `dataPoints:dailyRollUp` (`readMethod: "dailyRollUp"`, +one civil day per request). They are defined in `GOOGLE_HEALTH_DATA_TYPES` so the +read layer routes them correctly, but are **not yet wired to a Measurement +mapper** (active energy already covers the active-calories slot; `total-calories` +folds in BMR and needs a modelling decision before it lands as a metric). + +## Casing gotcha + +The data-type id is **kebab-case in the request path** (`body-fat`) and +**snake_case in the `filter` predicate** (`body_fat`). `GOOGLE_HEALTH_DATA_TYPES` +pins both forms per type so a fetcher can never encode the wrong one. + +## Health-metrics bundle + +Scope: `googlehealth.health_metrics_and_measurements.readonly`. Identifiers +verified against the 2026 contract (API-RESEARCH §3/§4). + +| Data type (path / filter) | MeasurementType | Unit | fieldTag | Grain | Note | +| ----------------------------------------------------------------------------- | ------------------------ | ----------- | ------------ | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `weight` / `weight` | `WEIGHT` | kg | `weight` | sample | Picker ranks a real Withings scale above Google Health. | +| `body-fat` / `body_fat` | `BODY_FAT` | % | `body_fat` | sample | | +| `oxygen-saturation` / `oxygen_saturation` | `OXYGEN_SATURATION` | % | `spo2` | daily | Already 0–100. | +| `heart-rate-variability` / `heart_rate_variability` | `HEART_RATE_VARIABILITY` | ms | `hrv` | daily | **Decision:** SDNN slot (Apple-comparable), NOT `HRV_RMSSD`. Google reports an RMSSD-style nightly HRV — confirm the estimator at build and reconsider if warranted. | +| `daily-resting-heart-rate` / `daily_resting_heart_rate` | `RESTING_HEART_RATE` | bpm | `rhr` | daily | | +| `respiratory-rate` / `respiratory_rate` | `RESPIRATORY_RATE` | breaths/min | `resp_rate` | daily | Identifier appears in spec prose; confirm `:list` support at build (API-RESEARCH OPEN §4). | +| `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`, `electrocardiogram` (ECG) and +`irregular-rhythm-notification` (gated behind the experimental ECG/IRN scopes). +These light up when Google ships the data types + the operator opts in. + +## Activity bundle — daily cumulative + +Scope: `googlehealth.activity_and_fitness.readonly`. Each is a per-day summary +(one value per calendar day). The externalId carries the `stats:` daily-total +prefix — `stats::` — so a re-fetched day **overwrites** in +place rather than minting a duplicate, matching the Apple-Health +`stats::` daily-total overwrite contract. The running totals +(steps/distance/floors/active-energy) **preserve a 0** (a rest day is real data, +not a gap); VO2 max stays strictly positive (daily latest-wins). + +| Data type (path / filter) | MeasurementType | Unit | fieldTag | Note | +| ----------------------------------------------- | -------------------------- | ----------- | --------------- | ------------------------------------------------------------------- | +| `steps` / `steps` | `ACTIVITY_STEPS` | steps | `steps` | daily total; 0 valid | +| `distance` / `distance` | `WALKING_RUNNING_DISTANCE` | m | `distance` | metres (km → m when reported in km) | +| `active-energy-burned` / `active_energy_burned` | `ACTIVE_ENERGY_BURNED` | kcal | `active_energy` | **ACTIVE portion only** — NOT `total-calories` (which folds in BMR) | +| `floors` / `floors` | `FLIGHTS_CLIMBED` | flights | `floors` | daily total; 0 valid | +| `vo2-max` / `vo2_max` | `VO2_MAX` | mL/(kg·min) | `vo2_max` | daily latest-wins; strictly positive | + +## Roll-up-only types (no `:list`) + +Scope: `googlehealth.activity_and_fitness.readonly`. Read via `:dailyRollUp`; not +yet mapped to a MeasurementType (see **Read method** above). + +| Data type (path / filter) | Status | +| ------------------------------------------------------------- | ------------------------------------------------------- | +| `total-calories` / `total_calories` | roll-up-only; folds in BMR; modelling decision pending | +| `calories-in-heart-rate-zone` / `calories_in_heart_rate_zone` | roll-up-only; per-zone breakdown; no HealthLog slot yet | + +## Sleep bundle + +Scope: `googlehealth.sleep.readonly`. A sleep session carries per-stage segments +(stage label + start + end). HealthLog stores one `SLEEP_DURATION` row per stage +SEGMENT, `measuredAt = that segment's END instant`, harmonised onto the shared +`SleepStage` enum the night-total + hypnogram readers consume. externalId = +`:sleep_:` (session end ISO instant + indexed segment), +so a re-scored night overwrites in place. Stage map: `light → CORE` ("light" ↔ +Apple "core" shallow-NREM band), `deep → DEEP`, `rem → REM`, +`awake`/`wake`/`restless → AWAKE`, `in_bed → IN_BED`, classic `asleep → ASLEEP`. +Unknown stage labels are skipped, not mis-bucketed. + +## Exercise bundle — Workouts + +Scope: `googlehealth.activity_and_fitness.readonly`. Each exercise session → one +`Workout` row (NOT a Measurement), keyed `(userId, source: "GOOGLE_HEALTH", +externalId)` where externalId is the session id (or `exercise:` when +absent). Fields: sportType (Google activity type → canonical `WorkoutSportType`, +`other` fallback), startedAt/endedAt, durationSec, totalEnergyKcal (active session +energy), totalDistanceM, avg/max/min HR (optional). A Google Health run and the +same run via Apple Health / WHOOP stay distinct rows; the read-time +`pickCanonicalWorkoutRows` picker collapses the cross-source twin per the user's +source ladder. + +## Idempotency + +`(userId, type, source: "GOOGLE_HEALTH", externalId)` unique. Spot/daily metric +rows: `externalId = :`. Daily cumulative activity rows: +`externalId = stats::` (Apple-Health overwrite shape). +Workouts: `(userId, source: "GOOGLE_HEALTH", externalId)` on the `Workout` table. +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/google-health/oauth-state.ts b/src/lib/google-health/oauth-state.ts new file mode 100644 index 000000000..2f8a9fc5a --- /dev/null +++ b/src/lib/google-health/oauth-state.ts @@ -0,0 +1,38 @@ +/** + * Shared constants + mint helper for the Google Health OAuth state nonce ledger. + * Mirrors `src/lib/whoop/oauth-state.ts`. + * + * The nonce is random + opaque; a short-lived `GoogleHealthOAuthState` row + * carries the `(nonce → userId)` mapping (plus the PKCE code verifier) so the + * user id and the verifier never leak into request logs or network captures via + * the OAuth `state` param. The connect route, the callback route, and the + * cleanup cron all read these constants so the contract has one source of truth + * and the unit tests can pin it without importing either route file. + * + * The string keys are deliberately distinct from the classic Fitbit provider's + * (`fitbit_state`) so the two coexisting integrations never collide on a cookie. + */ +import { randomBytes } from "node:crypto"; + +/** + * Cookie name carrying the in-flight state nonce. httpOnly + Secure + + * sameSite:lax at the connect route (mirrors the WHOOP cookie shape). Distinct + * from the classic Fitbit `fitbit_state` cookie. + */ +export const GOOGLE_HEALTH_OAUTH_STATE_COOKIE = "google_health_state" as const; + +/** + * 10-minute TTL on the state row. Matches the cookie `maxAge`. Long enough to + * cover a user approving the Google consent prompt; short enough that an + * abandoned handshake doesn't strand a row for hours. + */ +export const GOOGLE_HEALTH_OAUTH_STATE_TTL_MS = 10 * 60 * 1000; + +/** + * Mint a fresh state nonce. 16 random bytes → 22 base64url chars without + * padding — 128 bits of entropy, aligned with the OAuth 2.0 §10.10 "at least + * 128 bits" recommendation. + */ +export function mintGoogleHealthOAuthStateNonce(): string { + return randomBytes(16).toString("base64url"); +} diff --git a/src/lib/google-health/response-classifier.ts b/src/lib/google-health/response-classifier.ts new file mode 100644 index 000000000..75104614c --- /dev/null +++ b/src/lib/google-health/response-classifier.ts @@ -0,0 +1,114 @@ +/** + * Off-response classifier for Google Health API replies (v1.27.0). + * + * Like WHOOP / Fitbit, Google signals failure through the HTTP status code + * itself (no `status` field embedded in a 200 body), so the classification is + * status-driven. The shared core lives in + * `src/lib/integrations/http-status-classifier.ts`; this module re-exports it + * under the Google-Health-specific names callers already use. + * + * - `success` → HTTP 2xx + * - `transient` → retryable: HTTP 429 (rate-limited — honour Google's + * truncated exponential backoff at the call site), 5xx + * upstream outages, 3xx CDN/network glitches, and an + * unknown status where we never reached the envelope. + * - `reauth_required` → HTTP 401 / 403 — the access token is rejected or the + * grant was revoked; the user must redo OAuth. (The + * per-data-class 403 soft-skip lives in `sync.ts`'s + * `isCollectionForbidden`, not here, so a Restricted + * scope granted independently doesn't park the whole + * connection.) + * - `persistent` → any other 4xx (400, 404, 422, …) — a malformed + * request or a contract mismatch retrying won't fix. + * + * NEEDS-REAUTH SIGNAL. Google surfaces a revoked / expired refresh token — a + * user-revoked grant OR the 7-day "Testing"-mode refresh-token expiry — via + * `invalid_grant` on the token endpoint (sometimes a 400, sometimes a 401). A + * plain status classification would bucket that 400 as `persistent` and never + * prompt a reconnect; the `client.ts` token path lifts an `invalid_grant` onto + * `reauth_required` before constructing the error, and a 401 already classifies + * that way. Downstream, `reauth_required` maps through + * `classificationToFailureKind` → `FailureKind "reauth_required"` → + * `IntegrationState "error_reauth"`, which the status card renders as a + * "reconnect" CTA. Use `isGoogleHealthReauthRequired()` to read that signal off + * any caught error. + */ + +import { + classifyHttpStatus, + classifyIntegrationError, + isOAuthInvalidGrant, + IntegrationApiError, + type ClassifiedHttpResponse, + type IntegrationClassification, +} from "@/lib/integrations/http-status-classifier"; + +/** Wire label used in the error-message prefix + the message-shape fallbacks. */ +export const GOOGLE_HEALTH_VENDOR = "GoogleHealth"; + +/** Outcome buckets for a Google Health response. */ +export type GoogleHealthClassification = IntegrationClassification; + +export type ClassifiedGoogleHealthResponse = ClassifiedHttpResponse; + +/** Classify a single Google Health API response by its HTTP status. */ +export function classifyGoogleHealthResponse( + httpStatus: number, +): ClassifiedGoogleHealthResponse { + return classifyHttpStatus(httpStatus); +} + +/** + * Typed error carrying the classification verdict so downstream `try/catch` + * blocks can branch without re-parsing the message. Extends the shared base + * (message shape `GoogleHealth error: ` is preserved). + */ +export class GoogleHealthApiError extends IntegrationApiError { + constructor(opts: { + verb: string; + classification: GoogleHealthClassification; + httpStatus: number | undefined; + reason: string; + upstreamError?: string; + }) { + super({ vendor: GOOGLE_HEALTH_VENDOR, ...opts }); + this.name = "GoogleHealthApiError"; + } +} + +/** + * Read the classification from any caught error. Returns `transient` for a + * non-`GoogleHealthApiError` input so a call site surfacing a plain `Error` + * retries rather than permanently disabling the integration. Falls back to + * parsing the HTTP status out of the legacy `"GoogleHealth error: + * "` message shape for callers that lose the original prototype across a + * pg-boss retry. + */ +export function classifyGoogleHealthError( + err: unknown, +): GoogleHealthClassification { + return classifyIntegrationError(err, GOOGLE_HEALTH_VENDOR); +} + +/** + * True when a caught error carries an OAuth `invalid_grant` on a 400 from the + * token endpoint — the canonical signal that the stored refresh token was + * revoked or expired (a user-revoked grant, or the 7-day "Testing"-mode + * expiry). Robust to a lost prototype across a pg-boss retry. + */ +export function isGoogleHealthInvalidGrant(err: unknown): boolean { + return isOAuthInvalidGrant(err, GOOGLE_HEALTH_VENDOR); +} + +/** + * True when a caught error means the user must reconnect — either a rejected + * token / revoked grant (401/403 → `reauth_required`) or a token-endpoint + * `invalid_grant`. This is the single predicate downstream routes / UI consume + * to decide whether to show the "reconnect" CTA. + */ +export function isGoogleHealthReauthRequired(err: unknown): boolean { + return ( + classifyGoogleHealthError(err) === "reauth_required" || + isGoogleHealthInvalidGrant(err) + ); +} diff --git a/src/lib/google-health/sync-activity.ts b/src/lib/google-health/sync-activity.ts new file mode 100644 index 000000000..4aad7a757 --- /dev/null +++ b/src/lib/google-health/sync-activity.ts @@ -0,0 +1,140 @@ +/** + * Google Health activity-bundle sync (v1.27.0). + * + * Reads the daily cumulative-activity data types from the + * `activity_and_fitness.readonly` Restricted bundle and upserts each mapped + * daily total as `source = GOOGLE_HEALTH`: + * + * - steps → ACTIVITY_STEPS (count) + * - distance → WALKING_RUNNING_DISTANCE (metres) + * - active-energy-burned→ ACTIVE_ENERGY_BURNED (kcal — ACTIVE portion only) + * - floors → FLIGHTS_CLIMBED (count) + * - vo2-max → VO2_MAX (mL/(kg·min); daily latest-wins) + * + * These are per-day summaries (one value per calendar day). The externalId is + * minted with the `stats:` daily-total prefix — `stats::` + * — so a re-fetched day OVERWRITES the existing row rather than minting a + * duplicate, matching the Apple-Health `stats::` overwrite + * contract. A day of rest legitimately records 0 steps / 0 floors / 0 active + * kcal, so the cumulative mappers preserve a zero; VO2 max stays strictly + * positive. + * + * A per-data-class 403 soft-skips THAT class (returns 0, leaves the connection + * connected) — the Restricted bundles are granted independently. + */ +import { + GOOGLE_HEALTH_DATA_TYPES, + type GoogleHealthDataType, + type GoogleHealthMappedMeasurement, + fetchDataPoints, + mapActiveEnergy, + mapDistance, + mapFloors, + mapSteps, + mapVo2Max, +} from "./client"; +import { + getValidToken, + handleCollectionFetchError, + upsertGoogleHealthMeasurements, + type GoogleHealthMeasurementUpsert, + type GoogleHealthResourceSyncOptions, +} from "./sync"; +import { annotate } from "@/lib/logging/context"; + +/** One mappable activity metric: its data-type encoding + the mapper + a verb. */ +interface ActivityResource { + dataType: GoogleHealthDataType; + map: (point: Record) => GoogleHealthMappedMeasurement[]; + verb: string; +} + +const ACTIVITY_RESOURCES: ActivityResource[] = [ + { + dataType: GOOGLE_HEALTH_DATA_TYPES.steps, + map: mapSteps, + verb: "fetchSteps", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.distance, + map: mapDistance, + verb: "fetchDistance", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.activeEnergy, + map: mapActiveEnergy, + verb: "fetchActiveEnergy", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.floors, + map: mapFloors, + verb: "fetchFloors", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.vo2Max, + map: mapVo2Max, + verb: "fetchVo2Max", + }, +]; + +/** + * Assemble the stored externalId for a mapped activity reading. A cumulative + * daily metric carries the `cumulativeDaily` flag from its mapper — its + * externalId gets the `stats:` daily-total prefix so a re-fetched day overwrites + * in place (the mapper's `fieldTag` is already the `:` form). + */ +function externalIdFor(m: GoogleHealthMappedMeasurement): string { + return m.cumulativeDaily ? `stats:${m.fieldTag}` : m.fieldTag; +} + +export async function syncUserActivity( + userId: string, + opts: GoogleHealthResourceSyncOptions = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + // Cycle-wide watermark snapshotted once by `syncUserGoogleHealth`; undefined + // on a full/backfill run. + const start = opts.start; + + let imported = 0; + for (const resource of ACTIVITY_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: GoogleHealthMeasurementUpsert[] = []; + 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: externalIdFor(m), + }); + } + } + imported += ( + await upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + } + + // `markSynced` is owned by the orchestrator (`syncUserGoogleHealth`). + annotate({ + action: { name: "googleHealth.activity.sync", details: { imported } }, + }); + return imported; +} diff --git a/src/lib/google-health/sync-metrics.ts b/src/lib/google-health/sync-metrics.ts new file mode 100644 index 000000000..7065eaac3 --- /dev/null +++ b/src/lib/google-health/sync-metrics.ts @@ -0,0 +1,195 @@ +/** + * Google Health "health metrics & measurements" bundle sync (v1.27.0). + * + * Reads the launch metric data types from the + * `health_metrics_and_measurements.readonly` Restricted bundle and upserts each + * mapped reading as `source = GOOGLE_HEALTH`: + * + * - weight → WEIGHT + * - body-fat → BODY_FAT + * - oxygen-saturation → OXYGEN_SATURATION + * - heart-rate-variability → HEART_RATE_VARIABILITY (SDNN slot; see mapper) + * - daily-resting-HR → RESTING_HEART_RATE + * - 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 Restricted bundles + * can be granted independently in the Google consent flow. + */ +import { + GOOGLE_HEALTH_DATA_TYPES, + type GoogleHealthDataType, + type GoogleHealthMappedMeasurement, + fetchDataPoints, + mapBodyFat, + mapHeartRate, + mapHeartRateVariability, + mapHeightCm, + mapOxygenSaturation, + mapRespiratoryRate, + mapRestingHeartRate, + mapSleepTemperature, + mapWeight, +} from "./client"; +import { + getValidToken, + handleCollectionFetchError, + upsertGoogleHealthMeasurements, + type GoogleHealthMeasurementUpsert, + type GoogleHealthResourceSyncOptions, +} from "./sync"; +import { getEvent } from "@/lib/logging/context"; +import { prisma } from "@/lib/db"; + +/** One mappable metric: its data-type encoding + the per-point mapper + a verb. */ +interface MetricResource { + dataType: GoogleHealthDataType; + map: (point: Record) => GoogleHealthMappedMeasurement[]; + verb: string; +} + +/** The launch metric resources (Measurement-producing). Height handled separately. */ +const METRIC_RESOURCES: MetricResource[] = [ + { + dataType: GOOGLE_HEALTH_DATA_TYPES.weight, + map: mapWeight, + verb: "fetchWeight", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.bodyFat, + map: mapBodyFat, + verb: "fetchBodyFat", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.oxygenSaturation, + map: mapOxygenSaturation, + verb: "fetchSpo2", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.heartRateVariability, + map: mapHeartRateVariability, + verb: "fetchHrv", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.restingHeartRate, + map: mapRestingHeartRate, + verb: "fetchRhr", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.respiratoryRate, + map: mapRespiratoryRate, + verb: "fetchRespiratoryRate", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate, + map: mapHeartRate, + verb: "fetchHeartRate", + }, + { + dataType: GOOGLE_HEALTH_DATA_TYPES.sleepTemperature, + map: mapSleepTemperature, + verb: "fetchSleepTemperature", + }, +]; + +export async function syncUserMetrics( + userId: string, + opts: GoogleHealthResourceSyncOptions = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + // The incremental lower bound is the cycle-wide watermark snapshotted once by + // `syncUserGoogleHealth` — never re-read here, so a sibling resource's stamp + // can't shrink this one's window. Undefined on a full/backfill run. + const start = opts.start; + + 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: GoogleHealthMeasurementUpsert[] = []; + 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 upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + } + + // 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( + GOOGLE_HEALTH_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( + `google-health: failed to seed heightCm for ${userId}: ${err}`, + ); + } + + // `markSynced` is owned by the orchestrator (`syncUserGoogleHealth`), stamped + // once at the end of the cycle — never here, so the watermark can't move + // mid-cycle. + return imported; +} diff --git a/src/lib/google-health/sync-sleep.ts b/src/lib/google-health/sync-sleep.ts new file mode 100644 index 000000000..b75dd8279 --- /dev/null +++ b/src/lib/google-health/sync-sleep.ts @@ -0,0 +1,83 @@ +/** + * Google Health sleep-bundle sync (v1.27.0). + * + * Reads sleep sessions from the `sleep.readonly` Restricted bundle and maps each + * into per-SEGMENT `SLEEP_DURATION` rows via `mapSleepSession` (one row per stage + * segment; `measuredAt = that segment's END` instant; harmonised onto the shared + * `SleepStage` enum IN_BED/AWAKE/ASLEEP/REM/CORE/DEEP that the night-total + + * hypnogram readers already consume for WHOOP / Apple). Google carries a real + * per-segment series, so the rows lay each block at its true clock time (a + * MEASURED timeline, not reconstructed). Upserts as `source = GOOGLE_HEALTH`. + * + * Each segment carries the `sleepStage` axis and an indexed fieldTag so the + * several segments of one stage stay distinct under the + * `(userId, type, source, externalId)` dedup key. externalId = + * `:sleep_:` — a re-scored night overwrites in place. + * A 24 h overlap covers Google's after-the-fact re-score. + * + * A per-data-class 403 soft-skips the resource — the sleep bundle is granted + * independently of activity / metrics in the Google consent flow. + */ +import { + GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE, + GOOGLE_HEALTH_DATA_TYPES, + fetchDataPoints, + mapSleepSession, +} from "./client"; +import { + getValidToken, + handleCollectionFetchError, + upsertGoogleHealthMeasurements, + type GoogleHealthMeasurementUpsert, + type GoogleHealthResourceSyncOptions, +} from "./sync"; +import { annotate } from "@/lib/logging/context"; + +export async function syncUserSleep( + userId: string, + opts: GoogleHealthResourceSyncOptions = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + // Cycle-wide watermark snapshotted once by `syncUserGoogleHealth`; undefined + // on a full/backfill run. + const start = opts.start; + + let points: Record[]; + try { + points = await fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.sleep, + tokenInfo.accessToken, + "fetchSleep", + { start, pageSize: GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE }, + ); + } catch (err) { + return handleCollectionFetchError("fetchSleep", userId, err); + } + + const readings: GoogleHealthMeasurementUpsert[] = []; + for (const point of points) { + for (const m of mapSleepSession(point)) { + readings.push({ + type: m.type, + value: m.value, + unit: m.unit, + measuredAt: m.measuredAt, + externalId: m.fieldTag, + sleepStage: m.sleepStage ?? null, + }); + } + } + + const imported = ( + await upsertGoogleHealthMeasurements(userId, readings, { + deferRollup: opts.deferRollup, + }) + ).imported; + // `markSynced` is owned by the orchestrator (`syncUserGoogleHealth`). + annotate({ + action: { name: "googleHealth.sleep.sync", details: { imported } }, + }); + return imported; +} diff --git a/src/lib/google-health/sync-workout.ts b/src/lib/google-health/sync-workout.ts new file mode 100644 index 000000000..6a7cbd9ca --- /dev/null +++ b/src/lib/google-health/sync-workout.ts @@ -0,0 +1,105 @@ +/** + * Google Health exercise-session sync (v1.27.0). + * + * Reads exercise sessions from the `activity_and_fitness.readonly` Restricted + * bundle and upserts each into the `Workout` table as `source = GOOGLE_HEALTH`, + * keyed on `(userId, source, externalId)` so a re-fetch overwrites in place. + * + * A Google Health run and the same run via Apple Health (or WHOOP) stay distinct + * `Workout` rows (different `source`); the read-time `pickCanonicalWorkoutRows` + * picker collapses the cross-source twin at read time per the user's source + * ladder. There is no ingest-time collapse for a server-owned source pair. + * + * A per-data-class 403 soft-skips the resource — the activity/fitness bundle is + * granted independently in the Google consent flow. + */ +import { + GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE, + GOOGLE_HEALTH_DATA_TYPES, + fetchDataPoints, + mapWorkout, + type GoogleHealthMappedWorkout, +} from "./client"; +import { + getValidToken, + handleCollectionFetchError, + type GoogleHealthResourceSyncOptions, +} from "./sync"; +import { prisma } from "@/lib/db"; +import { annotate, getEvent } from "@/lib/logging/context"; + +export async function syncUserWorkout( + userId: string, + opts: GoogleHealthResourceSyncOptions = {}, +): Promise { + const tokenInfo = await getValidToken(userId); + if (!tokenInfo) return 0; + + // Cycle-wide watermark snapshotted once by `syncUserGoogleHealth`; undefined + // on a full/backfill run. + const start = opts.start; + + let points: Record[]; + try { + points = await fetchDataPoints( + GOOGLE_HEALTH_DATA_TYPES.exercise, + tokenInfo.accessToken, + "fetchExercise", + { start, pageSize: GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE }, + ); + } catch (err) { + return handleCollectionFetchError("fetchExercise", userId, err); + } + + let imported = 0; + for (const point of points) { + const w: GoogleHealthMappedWorkout | null = mapWorkout(point); + if (!w) continue; // no usable time span — not a workout + + try { + await prisma.workout.upsert({ + where: { + userId_source_externalId: { + userId, + source: "GOOGLE_HEALTH", + externalId: w.externalId, + }, + }, + create: { + userId, + source: "GOOGLE_HEALTH", + externalId: w.externalId, + sportType: w.sportType, + startedAt: w.startedAt, + endedAt: w.endedAt, + durationSec: w.durationSec, + totalEnergyKcal: w.totalEnergyKcal, + totalDistanceM: w.totalDistanceM, + avgHeartRate: w.avgHeartRate, + maxHeartRate: w.maxHeartRate, + minHeartRate: w.minHeartRate, + }, + update: { + sportType: w.sportType, + startedAt: w.startedAt, + endedAt: w.endedAt, + durationSec: w.durationSec, + totalEnergyKcal: w.totalEnergyKcal, + totalDistanceM: w.totalDistanceM, + avgHeartRate: w.avgHeartRate, + maxHeartRate: w.maxHeartRate, + minHeartRate: w.minHeartRate, + }, + }); + imported++; + } catch (err) { + getEvent()?.addWarning(`google-health: failed to upsert workout: ${err}`); + } + } + + // `markSynced` is owned by the orchestrator (`syncUserGoogleHealth`). + annotate({ + action: { name: "googleHealth.workout.sync", details: { imported } }, + }); + return imported; +} diff --git a/src/lib/google-health/sync.ts b/src/lib/google-health/sync.ts new file mode 100644 index 000000000..b161bd269 --- /dev/null +++ b/src/lib/google-health/sync.ts @@ -0,0 +1,729 @@ +/** + * Google Health sync service (v1.27.0). + * + * Token-management + per-resource orchestration: + * - `getValidToken` decrypts the stored token, refreshes at + * `tokenExpiresAt - 5 min`, persists the new access token + expiry, and — + * following Google's semantics — only overwrites the stored refresh token + * when the response carries a fresh one (Google does NOT rotate refresh + * tokens). A revoked / expired refresh token (the 7-day "Testing"-mode + * expiry, or a user-revoked grant) surfaces `invalid_grant`/401 → + * `reauth_required`, which parks the connection at `error_reauth` so the + * card prompts a reconnect. + * - `recordGoogleHealthSyncFailure` / `classificationToFailureKind` map a + * classified API error onto the shared integration-status ledger. + * - `syncUserGoogleHealth` drives the hourly poll, the manual trigger, and the + * boot-time backfill across every resource. + */ +import { AsyncLocalStorage } from "node:async_hooks"; +import pLimit from "p-limit"; +import { prisma } from "@/lib/db"; +import type { MeasurementType } from "@/generated/prisma/client"; +import { encrypt, decrypt } from "@/lib/crypto"; +import { annotate, getEvent } from "@/lib/logging/context"; +import { + isReauthRequired, + recordSyncFailure, + recordSyncSuccess, + type FailureKind, +} from "@/lib/integrations/status"; +import { + collapseToTypeDayKeys, + recomputeBucketsForMeasurement, + recomputeUserRollups, +} from "@/lib/rollups/measurement-rollups"; +import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { refreshAccessToken } from "./client"; +import { getUserGoogleHealthCredentials } from "./credentials"; +import { + GoogleHealthApiError, + classifyGoogleHealthError, + isGoogleHealthReauthRequired, + type GoogleHealthClassification, +} from "./response-classifier"; + +/** + * The status-ledger key for this integration. Distinct from the classic Fitbit + * `"fitbit"` key so the two coexisting providers keep independent status rows. + */ +export const GOOGLE_HEALTH_INTEGRATION_KEY = "google-health"; + +/** 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 GOOGLE_HEALTH_DEFAULT_OVERLAP_MS = 24 * 60 * 60 * 1000; + +export interface GoogleHealthTokenInfo { + accessToken: string; + connection: { id: string; googleUserId: string }; +} + +/** + * Resolve a valid Google Health access token for a user, refreshing if it is + * within the 5-minute expiry buffer. Returns null when there is no connection, + * no credentials, or the refresh fails (the failure is recorded so scheduled + * syncs back off). + * + * Google does not rotate refresh tokens. On refresh, persist the new access + * token + expiry and overwrite the stored `refreshToken` ONLY when the response + * carries a fresh one — otherwise keep the existing refresh token so the next + * refresh still authenticates. + */ +export async function getValidToken( + userId: string, +): Promise { + const connection = await prisma.googleHealthConnection.findUnique({ + where: { userId }, + }); + if (!connection) return null; + + const accessToken = decrypt(connection.accessToken); + const refreshToken = decrypt(connection.refreshToken); + + if ( + connection.tokenExpiresAt.getTime() - TOKEN_REFRESH_BUFFER_MS < + Date.now() + ) { + try { + const creds = await getUserGoogleHealthCredentials(userId); + if (!creds) { + getEvent()?.addWarning( + `No Google Health credentials found for user ${userId} during token refresh`, + ); + await recordSyncFailure({ + userId, + integration: GOOGLE_HEALTH_INTEGRATION_KEY, + kind: "reauth_required", + message: "Google Health credentials missing — token refresh skipped", + errorCode: "credentials_missing", + }); + return null; + } + + const newTokens = await refreshAccessToken(refreshToken, creds); + const expiresAt = new Date(Date.now() + newTokens.expires_in * 1000); + + // Google does NOT rotate refresh tokens: a routine refresh returns a new + // access token but usually omits `refresh_token`. Persist the new access + // token + expiry and only overwrite the stored refresh token when the + // response actually carries a fresh one — otherwise keep the existing one. + await prisma.googleHealthConnection.update({ + where: { id: connection.id }, + data: { + accessToken: encrypt(newTokens.access_token), + tokenExpiresAt: expiresAt, + ...(newTokens.refresh_token + ? { refreshToken: encrypt(newTokens.refresh_token) } + : {}), + }, + }); + + return { + accessToken: newTokens.access_token, + connection: { + id: connection.id, + googleUserId: connection.googleUserId, + }, + }; + } catch (err) { + getEvent()?.addWarning( + `Google Health token refresh failed for user ${userId}: ${err}`, + ); + // A revoked / expired refresh token (`invalid_grant` / 401 — the 7-day + // Testing-mode refresh expiry, or a user-revoked grant) is a SOFT + // disconnect: flag `needsReauth` on the connection so the card surfaces a + // "reconnect" CTA rather than a hard error. The callback clears it on a + // fresh consent. A transient refresh failure (network blip, 5xx) leaves + // the flag untouched so we don't nag the user to reconnect needlessly. + if (isGoogleHealthReauthRequired(err)) { + await prisma.googleHealthConnection + .update({ + where: { id: connection.id }, + data: { needsReauth: true }, + }) + .catch((e) => + getEvent()?.addWarning( + `google-health: needsReauth flag write failed for ${userId}: ${e}`, + ), + ); + } + await recordGoogleHealthSyncFailure(userId, err); + return null; + } + } + + return { + accessToken, + connection: { + id: connection.id, + googleUserId: connection.googleUserId, + }, + }; +} + +/** + * True when a caught error is a per-data-class 403 (forbidden). The 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. + */ +export function isCollectionForbidden(err: unknown): boolean { + return err instanceof GoogleHealthApiError && err.httpStatus === 403; +} + +/** + * Per-`syncUserGoogleHealth`-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(); + +/** + * Per-cycle accumulator of the `(type, day)` keys every resource's writes + * touched. Only populated on a `fullSync` backfill (when the inline per-day + * rollup hook is deferred); the orchestrator drains it into ONE + * `recomputeUserRollups(from, to)` pass at the end of the cycle. Scoped by + * AsyncLocalStorage so concurrent per-user jobs never cross-pollinate. + */ +interface RollupDeferTracker { + keys: Array<{ type: MeasurementType; measuredAt: Date }>; +} +const rollupDeferStorage = 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 `syncUserGoogleHealth` 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( + `google-health ${resource} sync skipped for ${userId}: collection 403 (soft-skip)`, + ); + const tracker = softSkipStorage.getStore(); + if (tracker) tracker.count += 1; + return 0; + } + await recordGoogleHealthSyncFailure(userId, err); + throw err; +} + +/** + * Options threaded from `syncUserGoogleHealth` to each per-resource sync. The + * `start` watermark is snapshotted ONCE by the orchestrator (a single + * `lastSyncedAt` read) so every resource fetches from the same lower bound; no + * resource re-reads `lastSyncedAt` or stamps `markSynced` itself. On a + * full/backfill run `start` is undefined (no lower bound). The orchestrator owns + * the single end-of-cycle `markSynced`. + */ +export interface GoogleHealthResourceSyncOptions { + fullSync?: boolean; + /** The incremental lower bound, snapshotted once by the orchestrator. */ + start?: Date; + /** + * When true, `upsertGoogleHealthMeasurements` writes the rows but SKIPS the + * inline per-(type,day) DAY-rollup recompute + status-insight invalidate, + * returning the touched type-days to the caller instead. The orchestrator + * collapses them into ONE `recomputeUserRollups(from, to)` pass at the end of + * a `fullSync` cycle, so a multi-year backfill pays a single range-recompute + * rather than thousands of per-day round-trips. The incremental path leaves + * this unset and keeps the inline per-day hook. + */ + deferRollup?: boolean; +} + +/** + * 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 ?? GOOGLE_HEALTH_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 (`:`). + */ +export interface GoogleHealthMeasurementUpsert { + type: string; + value: number; + unit: string; + measuredAt: Date; + externalId: string; + /** + * Per-stage sleep rows carry the `SleepStage` axis so the (up to six) stage + * rows for one night stay distinct under the dedup key; every other row omits + * it (null). Sleep is the only producer. + */ + sleepStage?: "IN_BED" | "AWAKE" | "ASLEEP" | "REM" | "CORE" | "DEEP" | null; +} + +/** Chunk size for the batched `createMany` insert of fresh readings. */ +const GOOGLE_HEALTH_CREATE_CHUNK = 500; + +/** + * Write a batch of mapped Google Health readings for one user and (unless + * deferred) fold the rollup tier + invalidate status-insight caches once at the + * end (mirrors the WHOOP / Withings sync tail). Returns the count of rows + * written plus the distinct `(type, day)` keys the write touched. + * + * TOMBSTONE-SAFE: the live-row probe filters `deletedAt: null`, so a row the + * user soft-deleted (iOS LWW reconciler) is NOT matched — a fresh insert is + * created instead of resurrecting the tombstone on the next hourly sync. The + * partial unique index keeps the fresh insert from colliding with the live set. + * + * OVERWRITE CONTRACT preserved: a re-fetched daily summary (`stats:`-keyed and + * every other stable per-point anchor) maps to the SAME live `externalId`, so + * the probe finds the existing live row and takes the in-place update branch. + * + * BATCHING: the existence probe is a single `findMany` over the batch's + * externalIds; fresh rows go through chunked `createMany`, and only the + * already-live rows take a per-row `update` for their differing values. + * + * Best-effort on the rollup fold + insight invalidate — a populator hiccup never + * fails the user's sync. + */ +export async function upsertGoogleHealthMeasurements( + userId: string, + readings: GoogleHealthMeasurementUpsert[], + opts: { deferRollup?: boolean } = {}, +): Promise<{ + imported: number; + touched: Array<{ type: MeasurementType; measuredAt: Date }>; +}> { + if (readings.length === 0) return { imported: 0, touched: [] }; + + // Probe the LIVE rows (deletedAt: null) for every externalId in the batch in + // a single query. A tombstoned row deliberately does NOT appear here, so it is + // treated as absent → a fresh insert, not a resurrecting update. + const externalIds = readings.map((r) => r.externalId); + let liveByKey = new Map(); + try { + const existing = await prisma.measurement.findMany({ + where: { + userId, + source: "GOOGLE_HEALTH", + deletedAt: null, + externalId: { in: externalIds }, + }, + select: { id: true, type: true, externalId: true }, + }); + liveByKey = new Map( + existing + .filter((e) => e.externalId !== null) + .map((e) => [`${e.type} ${e.externalId}`, { id: e.id }]), + ); + } catch (err) { + // A probe failure must not strand the whole batch; fall back to treating + // every row as fresh (the partial unique index still rejects a genuine + // live-row collision, which the per-create catch swallows). + getEvent()?.addWarning(`google-health: live-row probe failed: ${err}`); + } + + const toCreate: Array<{ + type: MeasurementType; + value: number; + unit: string; + measuredAt: Date; + externalId: string; + sleepStage: GoogleHealthMeasurementUpsert["sleepStage"]; + }> = []; + const toUpdate: Array<{ id: string; r: GoogleHealthMeasurementUpsert }> = []; + const touched: Array<{ type: MeasurementType; measuredAt: Date }> = []; + + // A batch can carry the same (type, externalId) twice (an overlap re-fetch); + // collapse to last-write-wins so we never emit two creates for one live key. + const plannedCreateKeys = new Set(); + for (const r of readings) { + const type = r.type as MeasurementType; + const key = `${type} ${r.externalId}`; + const live = liveByKey.get(key); + if (live) { + toUpdate.push({ id: live.id, r }); + } else if (!plannedCreateKeys.has(key)) { + plannedCreateKeys.add(key); + toCreate.push({ + type, + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + externalId: r.externalId, + sleepStage: r.sleepStage ?? null, + }); + } else { + // A duplicate fresh key inside the same batch — overwrite the planned + // create's payload so last-write-wins, matching the prior upsert loop. + const idx = toCreate.findIndex( + (c) => `${c.type} ${c.externalId}` === key, + ); + if (idx >= 0) { + toCreate[idx] = { + type, + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + externalId: r.externalId, + sleepStage: r.sleepStage ?? null, + }; + } + } + } + + let imported = 0; + + // Fresh inserts: chunked `createMany` (server-owned rows, field-by-field). + // `skipDuplicates` guards the partial-unique index in the rare race where a + // concurrent run inserted the same live key between the probe and the write. + for (let i = 0; i < toCreate.length; i += GOOGLE_HEALTH_CREATE_CHUNK) { + const chunk = toCreate.slice(i, i + GOOGLE_HEALTH_CREATE_CHUNK); + try { + const res = await prisma.measurement.createMany({ + data: chunk.map((c) => ({ + userId, + type: c.type, + source: "GOOGLE_HEALTH" as const, + value: c.value, + unit: c.unit, + measuredAt: c.measuredAt, + externalId: c.externalId, + sleepStage: c.sleepStage, + })), + skipDuplicates: true, + }); + imported += res.count; + for (const c of chunk) { + touched.push({ type: c.type, measuredAt: c.measuredAt }); + } + } catch (err) { + getEvent()?.addWarning( + `google-health: failed to create measurements: ${err}`, + ); + } + } + + // Live-row overwrites: per-row update (differing values) on the live id, so + // the re-fetched daily summary overwrites in place and bumps `syncVersion`. + for (const { id, r } of toUpdate) { + try { + await prisma.measurement.update({ + where: { id }, + data: { + value: r.value, + unit: r.unit, + measuredAt: r.measuredAt, + sleepStage: r.sleepStage ?? null, + // Surface the server-side mutation to the iOS LWW reconciler. + syncVersion: { increment: 1 }, + }, + }); + touched.push({ + type: r.type as MeasurementType, + measuredAt: r.measuredAt, + }); + imported++; + } catch (err) { + getEvent()?.addWarning( + `google-health: failed to update measurement: ${err}`, + ); + } + } + + // On a `fullSync` backfill the caller defers the rollup fold: it collapses + // every resource's touched type-days into a SINGLE range-recompute at the end + // of the cycle. The incremental path keeps the inline per-day hook here. + if (opts.deferRollup) { + const tracker = rollupDeferStorage.getStore(); + if (tracker) tracker.keys.push(...touched); + return { imported, touched }; + } + + 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( + `google-health: status-insight invalidate failed for ${userId}: ${err}`, + ); + }); + } catch (err) { + getEvent()?.addWarning( + `google-health: rollup recompute failed for ${userId}: ${err}`, + ); + } + + return { imported, touched }; +} + +/** + * Stamp `lastSyncedAt = now`. Called ONCE per cycle by `syncUserGoogleHealth` + * after a non-degenerate run — never per resource, so the watermark can't move + * mid-cycle and shrink a later resource's fetch window. + */ +export async function markSynced(userId: string): Promise { + await prisma.googleHealthConnection.update({ + where: { userId }, + data: { lastSyncedAt: new Date() }, + }); +} + +/** + * Full per-user sync across every Google Health resource. Drives the hourly + * poll catch-all, the manual `/api/google-health/sync` trigger, and the + * boot-time backfill (`fullSync: true`). Poll-only — there is no Google Health + * webhook at launch (the Pub/Sub-style webhook uses a Tink keyset rotating every + * 30 days, 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. + * + * Resources: health-metrics, daily-cumulative activity, per-stage sleep, and + * exercise workouts. Each runs independently so a per-resource 403 (a Restricted + * bundle the user did not grant) soft-skips only that resource. + * + * WATERMARK: the incremental `start` is snapshotted ONCE here and threaded to + * every resource, and `markSynced` is stamped ONCE at the end. A per-resource + * read+stamp would let the first resource move `lastSyncedAt` to now() so later + * resources only see the last overlap window — silently dropping the gap after + * an outage longer than the overlap. + */ +export async function syncUserGoogleHealth( + userId: string, + opts: { fullSync?: boolean } = {}, +): Promise { + if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) { + getEvent()?.addWarning( + `google-health sync skipped for ${userId}: parked at error_reauth`, + ); + return 0; + } + + // Snapshot the incremental watermark ONCE for the whole cycle. Every resource + // sees the same `start`; no resource re-reads `lastSyncedAt` mid-cycle. On a + // full/backfill run `incrementalStart` ignores the snapshot and returns + // undefined (no lower bound). + const connection = await prisma.googleHealthConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true }, + }); + if (!connection) return 0; + const start = incrementalStart(connection.lastSyncedAt, { + fullSync: opts.fullSync, + }); + const resourceOpts: GoogleHealthResourceSyncOptions = { + fullSync: opts.fullSync, + start, + // A backfill walks years of daily summaries — defer the per-(type,day) + // rollup hook on every write and run ONE range-recompute at the end of the + // cycle. The hourly incremental keeps the inline hook (small touched set). + deferRollup: opts.fullSync === true, + }; + + const [ + { syncUserMetrics }, + { syncUserActivity }, + { syncUserSleep }, + { syncUserWorkout }, + ] = await Promise.all([ + import("./sync-metrics"), + import("./sync-activity"), + import("./sync-sleep"), + import("./sync-workout"), + ]); + + const resources = [ + syncUserMetrics, + syncUserActivity, + syncUserSleep, + syncUserWorkout, + ]; + + const tracker: SoftSkipTracker = { count: 0 }; + const deferTracker: RollupDeferTracker = { keys: [] }; + let total = 0; + let anyFailed = false; + await softSkipStorage.run(tracker, async () => { + await rollupDeferStorage.run(deferTracker, async () => { + for (const fn of resources) { + try { + total += await fn(userId, resourceOpts); + } catch (err) { + anyFailed = true; + getEvent()?.addWarning( + `google-health ${fn.name} failed for ${userId}: ${err}`, + ); + } + } + }); + }); + + // On a `fullSync` backfill the per-write inline rollup hook was deferred; the + // accumulated touched type-days collapse into ONE range-recompute spanning the + // touched days. Best-effort: a populator hiccup never fails the backfill. + if (opts.fullSync && deferTracker.keys.length > 0) { + try { + const days = collapseToTypeDayKeys(deferTracker.keys); + const types = Array.from(new Set(days.map((k) => k.type))); + const sorted = days + .map((k) => k.measuredAt.getTime()) + .sort((a, b) => a - b); + const from = new Date(sorted[0]!); + // The keys are UTC day-starts; extend `to` past the last touched day so + // the aggregator's `< to` upper bound covers it. + const to = new Date(sorted[sorted.length - 1]! + 24 * 60 * 60 * 1000); + await recomputeUserRollups(userId, { types, from, to }); + invalidateStatusInsightsForTypes(userId, types).catch((err) => { + getEvent()?.addWarning( + `google-health: status-insight invalidate failed for ${userId}: ${err}`, + ); + }); + } catch (err) { + getEvent()?.addWarning( + `google-health: backfill rollup recompute 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. + const allSoftSkipped = tracker.count >= resources.length && total === 0; + + if (!anyFailed && !allSoftSkipped) { + // 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. + await markSynced(userId); + await recordSyncSuccess(userId, GOOGLE_HEALTH_INTEGRATION_KEY); + } + + annotate({ + action: { name: "googleHealth.sync", details: { imported: total } }, + }); + return total; +} + +/** + * Map a Google Health response classification onto a `FailureKind` and record + * it. Shared by the token-refresh path and every per-resource catch block. + */ +export async function recordGoogleHealthSyncFailure( + userId: string, + err: unknown, +): Promise { + const message = err instanceof Error ? err.message : String(err); + await recordSyncFailure({ + userId, + integration: GOOGLE_HEALTH_INTEGRATION_KEY, + kind: classificationToFailureKind(classifyGoogleHealthError(err)), + message, + errorCode: + err instanceof GoogleHealthApiError + ? err.httpStatus?.toString() + : undefined, + }); +} + +export function classificationToFailureKind( + classification: GoogleHealthClassification, +): FailureKind { + switch (classification) { + case "reauth_required": + return "reauth_required"; + case "persistent": + return "persistent"; + case "transient": + return "transient"; + case "success": + // A caller asking for the FailureKind of a success is a contract bug; + // surface it as transient so the audit log still records the anomaly. + return "transient"; + } +} + +/** + * Bounded fan-out width for the hourly Google Health cohort poll. The cron tick + * carries no `userId`, so the worker resolves every connection and hands the + * cohort here. A small pool (rather than a strict serial loop) means one slow + * Google response can't stall the whole cohort, while the cap keeps the burst of + * per-user resource fetches + rollup writes from crowding the worker's DB pool. + */ +export const GOOGLE_HEALTH_POLL_CONCURRENCY = 4; + +/** + * Run a Google Health hourly-poll cohort with bounded concurrency + per-user + * error isolation. Each user's `syncUserGoogleHealth` runs inside the pool; a + * single user's failure (or a slow Google response) is captured through + * `onUserError` and never aborts the rest of the pass. Returns the per-cohort + * totals for the wide-event envelope. + */ +export async function runGoogleHealthPollCohort( + userIds: string[], + opts: { + concurrency?: number; + sync?: (userId: string) => Promise; + onUserError?: (userId: string, err: unknown) => void; + /** Invoked after a user's sync lands `imported` measurements, so the worker + * can fire the eventful Vorsorge satisfaction enqueue without the + * google-health lib reaching into the job layer. */ + onUserSynced?: (userId: string, imported: number) => void; + } = {}, +): Promise<{ usersSynced: number; measurementsImported: number }> { + const sync = opts.sync ?? ((userId: string) => syncUserGoogleHealth(userId)); + const limit = pLimit(opts.concurrency ?? GOOGLE_HEALTH_POLL_CONCURRENCY); + + let usersSynced = 0; + let measurementsImported = 0; + await Promise.all( + userIds.map((userId) => + limit(async () => { + try { + // Read the accumulators AFTER the await resolves. A compound + // assignment (`x += await f()`) captures the left value BEFORE the + // await and would lose increments under overlapping pool tasks. + const n = await sync(userId); + measurementsImported = measurementsImported + n; + usersSynced = usersSynced + 1; + opts.onUserSynced?.(userId, n); + } catch (err) { + opts.onUserError?.(userId, err); + } + }), + ), + ); + + return { usersSynced, measurementsImported }; +}