mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-08 12:32:43 +00:00
feat(google-health): add the Google Health API transport
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.
This commit is contained in:
Marc-André Bombeck committed
1 parent
f56268adf6
commit
fb17a4e9c6
14 files changed
+3843
No files matched your search
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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`,
|
||||
);
|
||||
});
|
||||
});
|
||||
File diff suppressed because it is too large.
Load diff
@@ -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<GoogleHealthCredentials | null> {
|
||||
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),
|
||||
};
|
||||
}
|
||||
@@ -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 = <anchor>:<fieldTag>` (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:<fieldTag>:<YYYY-MM-DD>` — so a re-fetched day **overwrites** in
|
||||
place rather than minting a duplicate, matching the Apple-Health
|
||||
`stats:<HK>:<YYYY-MM-DD>` 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 =
|
||||
`<session-anchor>:sleep_<stage>:<i>` (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:<startISO>` 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 = <anchor>:<fieldTag>`. Daily cumulative activity rows:
|
||||
`externalId = stats:<fieldTag>:<YYYY-MM-DD>` (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.
|
||||
@@ -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");
|
||||
}
|
||||
@@ -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 <verb> error: <status>` 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 <verb> error:
|
||||
* <status>"` 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)
|
||||
);
|
||||
}
|
||||
@@ -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:<fieldTag>:<YYYY-MM-DD>`
|
||||
* — so a re-fetched day OVERWRITES the existing row rather than minting a
|
||||
* duplicate, matching the Apple-Health `stats:<HK>:<YYYY-MM-DD>` 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<string, unknown>) => 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 `<tag>:<YYYY-MM-DD>` form).
|
||||
*/
|
||||
function externalIdFor(m: GoogleHealthMappedMeasurement): string {
|
||||
return m.cumulativeDaily ? `stats:${m.fieldTag}` : m.fieldTag;
|
||||
}
|
||||
|
||||
export async function syncUserActivity(
|
||||
userId: string,
|
||||
opts: GoogleHealthResourceSyncOptions = {},
|
||||
): Promise<number> {
|
||||
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<string, unknown>[];
|
||||
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;
|
||||
}
|
||||
@@ -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 (`<anchor>:<fieldTag>`) 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<string, unknown>) => 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<number> {
|
||||
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<string, unknown>[];
|
||||
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<string, unknown>[];
|
||||
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;
|
||||
}
|
||||
@@ -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 =
|
||||
* `<session-anchor>:sleep_<stage>:<i>` — 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<number> {
|
||||
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<string, unknown>[];
|
||||
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;
|
||||
}
|
||||
@@ -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<number> {
|
||||
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<string, unknown>[];
|
||||
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;
|
||||
}
|
||||
@@ -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<GoogleHealthTokenInfo | null> {
|
||||
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<SoftSkipTracker>();
|
||||
|
||||
/**
|
||||
* 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<RollupDeferTracker>();
|
||||
|
||||
/**
|
||||
* 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<number> {
|
||||
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 (`<anchor>:<fieldTag>`).
|
||||
*/
|
||||
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<string, { id: string }>();
|
||||
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<string>();
|
||||
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<void> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
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<number>;
|
||||
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 };
|
||||
}
|
||||
Reference in new issue
Block a user