diff --git a/src/lib/google-health/client.ts b/src/lib/google-health/client.ts index 237500921..659ccc75a 100644 --- a/src/lib/google-health/client.ts +++ b/src/lib/google-health/client.ts @@ -33,6 +33,7 @@ * `mappers.ts` and are imported here type-only — that direction keeps the * module graph acyclic (`client` → `mappers`, never back). */ +import { AsyncLocalStorage } from "node:async_hooks"; import { createHash, randomBytes } from "node:crypto"; import { annotate, getEvent } from "@/lib/logging/context"; import { safeFetch } from "@/lib/safe-fetch"; @@ -50,6 +51,75 @@ import type { export * from "./mappers"; +export interface GoogleHealthClientOutcome { + pages: number; + fetched: number; + mapped: number; + written: number; + truncated: boolean; + reasonCode: + | "collection_failed" + | "token_failed" + | "upsert_failed" + | "rollup_failed" + | "existing_page_limit" + | null; +} + +const clientOutcomeStorage = new AsyncLocalStorage(); + +function addOutcomeCount(current: number, value: number): number { + if (!Number.isFinite(value) || value <= 0) return current; + return Math.min(2_147_483_647, current + Math.trunc(value)); +} + +function notePagination( + pages: number, + fetched: number, + truncated: boolean, +): void { + const tracker = clientOutcomeStorage.getStore(); + if (!tracker) return; + tracker.pages = addOutcomeCount(tracker.pages, pages); + tracker.fetched = addOutcomeCount(tracker.fetched, fetched); + if (truncated) { + tracker.truncated = true; + tracker.reasonCode = "existing_page_limit"; + } +} + +export function noteGoogleHealthMapped(count: number): void { + const tracker = clientOutcomeStorage.getStore(); + if (tracker) tracker.mapped = addOutcomeCount(tracker.mapped, count); +} + +export function noteGoogleHealthWritten(count: number): void { + const tracker = clientOutcomeStorage.getStore(); + if (tracker) tracker.written = addOutcomeCount(tracker.written, count); +} + +export function noteGoogleHealthOutcomeFailure( + reasonCode: Exclude, +): void { + const tracker = clientOutcomeStorage.getStore(); + if (tracker && !tracker.reasonCode) tracker.reasonCode = reasonCode; +} + +export async function runWithGoogleHealthClientOutcome( + fn: () => Promise, +): Promise<{ result: T; outcome: GoogleHealthClientOutcome }> { + const outcome: GoogleHealthClientOutcome = { + pages: 0, + fetched: 0, + mapped: 0, + written: 0, + truncated: false, + reasonCode: null, + }; + const result = await clientOutcomeStorage.run(outcome, fn); + return { result, outcome }; +} + 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"; @@ -711,6 +781,7 @@ export async function fetchDataPoints( pageCount += 1; } while (pageToken && pageCount < maxPages); + notePagination(pageCount, points.length, Boolean(pageToken)); return points; }; @@ -935,6 +1006,8 @@ export async function fetchDailyRollUp( const walk = async (maxDays: number): Promise => { const points: GoogleHealthRollupPoint[] = []; let chunkIndex = 0; + let totalPages = 0; + let truncated = false; for (const chunk of chunkCivilRange(startCivil, endCivil, maxDays)) { let pageToken: string | null | undefined; let pageCount = 0; @@ -998,8 +1071,11 @@ export async function fetchDailyRollUp( pageToken = json?.nextPageToken ?? null; pageCount += 1; } while (pageToken && pageCount < 100); + totalPages += pageCount; + truncated ||= Boolean(pageToken); chunkIndex += 1; } + notePagination(totalPages, points.length, truncated); return points; }; diff --git a/src/lib/google-health/sync-core.ts b/src/lib/google-health/sync-core.ts index e2ff839b2..27ccaa139 100644 --- a/src/lib/google-health/sync-core.ts +++ b/src/lib/google-health/sync-core.ts @@ -27,7 +27,12 @@ import { recomputeBucketsForMeasurement, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; -import { refreshAccessToken } from "./client"; +import { + noteGoogleHealthMapped, + noteGoogleHealthOutcomeFailure, + noteGoogleHealthWritten, + refreshAccessToken, +} from "./client"; import { getUserGoogleHealthCredentials } from "./credentials"; import { GoogleHealthApiError, @@ -227,6 +232,13 @@ export const GOOGLE_HEALTH_TOKEN_HARD_FAIL = "token"; export function noteHardFailure(label: string): void { const tracker = hardFailStorage.getStore(); if (tracker) tracker.failures.push(label); + noteGoogleHealthOutcomeFailure( + label === GOOGLE_HEALTH_TOKEN_HARD_FAIL + ? "token_failed" + : label.includes("rollup") + ? "rollup_failed" + : "upsert_failed", + ); } /** @@ -314,6 +326,7 @@ export async function handleCollectionFetchError( if (tracker) tracker.count += 1; return 0; } + noteGoogleHealthOutcomeFailure("collection_failed"); await recordGoogleHealthSyncFailure(userId, err); getEvent()?.addWarning( `google-health ${resource} failed for ${userId}: ${err}`, @@ -485,6 +498,7 @@ export async function upsertGoogleHealthMeasurements( measuredAt: Date; }>; }> { + noteGoogleHealthMapped(readings.length); if (readings.length === 0) return { imported: 0, touched: [], inserted: [] }; // Probe EVERY existing row (live AND tombstoned) for the batch's externalIds @@ -796,6 +810,7 @@ export async function upsertGoogleHealthMeasurements( if (tracker) { for (const t of touched) tracker.keys.push(t); } + noteGoogleHealthWritten(imported); return { imported, touched, inserted: insertedRows }; } @@ -816,8 +831,10 @@ export async function upsertGoogleHealthMeasurements( getEvent()?.addWarning( `google-health: rollup recompute failed for ${userId}: ${err}`, ); + noteHardFailure("rollup"); } + noteGoogleHealthWritten(imported); return { imported, touched, inserted: insertedRows }; } diff --git a/src/lib/google-health/sync-workout.ts b/src/lib/google-health/sync-workout.ts index 80e1c252a..fd559fb3e 100644 --- a/src/lib/google-health/sync-workout.ts +++ b/src/lib/google-health/sync-workout.ts @@ -31,6 +31,7 @@ import { } from "./sync-core"; import { prisma } from "@/lib/db"; import { emitInsertedWorkoutArrival } from "@/lib/arrivals/workout-emit"; +import { invalidateUserMeasurements } from "@/lib/cache/invalidate"; import { annotate, getEvent } from "@/lib/logging/context"; import { resolveUserTimezone } from "@/lib/tz/resolver"; @@ -129,6 +130,10 @@ export async function syncUserWorkout( } } + if (imported > 0) { + await invalidateUserMeasurements(userId); + } + // `markSynced` is owned by the orchestrator (`syncUserGoogleHealth`). annotate({ action: { name: "googleHealth.workout.sync", details: { imported } }, diff --git a/src/lib/google-health/sync.ts b/src/lib/google-health/sync.ts index 791f3a031..90ef9fe7b 100644 --- a/src/lib/google-health/sync.ts +++ b/src/lib/google-health/sync.ts @@ -7,6 +7,10 @@ import { recomputeUserRollups, } from "@/lib/rollups/measurement-rollups"; import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate"; +import { + runWithGoogleHealthClientOutcome, + type GoogleHealthClientOutcome, +} from "./client"; import { syncUserActivity } from "./sync-activity"; import { syncUserMetrics } from "./sync-metrics"; import { syncUserSleep } from "./sync-sleep"; @@ -18,6 +22,108 @@ import { runWithGoogleHealthSyncCycle, type GoogleHealthResourceSyncOptions, } from "./sync-core"; +import { + startGoogleHealthSyncProgress, + updateGoogleHealthSyncProgress, + type GoogleHealthReasonCode, + type GoogleHealthResourceOutcome, + type GoogleHealthResourceStatus, + type GoogleHealthSyncState, +} from "./sync-progress"; + +export interface GoogleHealthSyncResult { + runId?: string; + state?: GoogleHealthSyncState; + imported: number; + failed: boolean; + resources?: GoogleHealthResourceOutcome[]; +} + +const RESOURCE_STATUSES = new Set([ + "pending", + "complete", + "partial", + "empty", + "truncated", + "failed", +]); +const REASON_CODES = new Set([ + "collection_failed", + "token_failed", + "upsert_failed", + "rollup_failed", + "existing_page_limit", +]); + +function boundedCount(value: unknown): number { + return typeof value === "number" && Number.isFinite(value) + ? Math.min(2_147_483_647, Math.max(0, Math.trunc(value))) + : 0; +} + +function boundedOutcome(value: unknown): GoogleHealthResourceOutcome { + const item = + value && typeof value === "object" + ? (value as Record) + : {}; + const status = RESOURCE_STATUSES.has( + item.status as GoogleHealthResourceStatus, + ) + ? (item.status as GoogleHealthResourceStatus) + : "failed"; + const reasonCode = REASON_CODES.has(item.reasonCode as GoogleHealthReasonCode) + ? (item.reasonCode as GoogleHealthReasonCode) + : null; + return { + resource: + typeof item.resource === "string" + ? item.resource + .toLowerCase() + .replace(/[^a-z0-9-]/g, "-") + .slice(0, 48) + : "unknown", + pages: boundedCount(item.pages), + fetched: boundedCount(item.fetched), + mapped: boundedCount(item.mapped), + written: boundedCount(item.written), + status, + durationMs: Math.min(86_400_000, boundedCount(item.durationMs)), + truncated: item.truncated === true, + reasonCode, + }; +} + +function terminalResourceOutcome( + resource: string, + imported: number, + outcome: GoogleHealthClientOutcome, + durationMs: number, +): GoogleHealthResourceOutcome { + const written = outcome.written || imported; + const mapped = outcome.mapped || imported; + const fetched = outcome.fetched || imported; + const pages = outcome.pages || (fetched > 0 ? 1 : 0); + const status: GoogleHealthResourceStatus = outcome.truncated + ? "truncated" + : outcome.reasonCode + ? written > 0 + ? "partial" + : "failed" + : fetched === 0 && written === 0 + ? "empty" + : "complete"; + return boundedOutcome({ + resource, + pages, + fetched, + mapped, + written, + status, + durationMs, + truncated: outcome.truncated, + reasonCode: outcome.reasonCode, + }); +} /** * Full per-user sync across every Google Health resource. The incremental @@ -27,19 +133,41 @@ import { export async function syncUserGoogleHealth( userId: string, opts: { fullSync?: boolean } = {}, -): Promise<{ imported: number; failed: boolean }> { +): Promise { if (await isReauthRequired(userId, GOOGLE_HEALTH_INTEGRATION_KEY)) { getEvent()?.addWarning( `google-health sync skipped for ${userId}: parked at error_reauth`, ); - return { imported: 0, failed: true }; + return { state: "failed", imported: 0, failed: true, resources: [] }; } const connection = await prisma.googleHealthConnection.findUnique({ where: { userId }, select: { lastSyncedAt: true }, }); - if (!connection) return { imported: 0, failed: true }; + if (!connection) { + return { state: "failed", imported: 0, failed: true, resources: [] }; + } + + const progress = await startGoogleHealthSyncProgress(userId); + const persistProgress = async ( + state: GoogleHealthSyncState, + imported: number, + failed: boolean, + resources: GoogleHealthResourceOutcome[], + ): Promise => { + await updateGoogleHealthSyncProgress(userId, progress.runId, { + state, + startedAt: progress.startedAt, + imported, + failed, + resources, + }).catch((err) => { + getEvent()?.addWarning( + `google-health progress write failed for ${userId}: ${err}`, + ); + }); + }; const start = incrementalStart(connection.lastSyncedAt, { fullSync: opts.fullSync, @@ -50,30 +178,64 @@ export async function syncUserGoogleHealth( deferRollup: opts.fullSync === true, }; const resources = [ - { name: "metrics", fn: syncUserMetrics }, - { name: "activity", fn: syncUserActivity }, - { name: "sleep", fn: syncUserSleep }, { name: "workout", fn: syncUserWorkout }, + { name: "sleep", fn: syncUserSleep }, + { name: "activity", fn: syncUserActivity }, + { name: "dense-heart-rate", fn: syncUserMetrics }, ]; const cycle = await runWithGoogleHealthSyncCycle(async () => { let total = 0; let anyFailed = false; + const resourceOutcomes: GoogleHealthResourceOutcome[] = []; for (const { name, fn } of resources) { + const startedAt = performance.now(); try { - total += await fn(userId, resourceOpts); + const tracked = await runWithGoogleHealthClientOutcome(() => + fn(userId, resourceOpts), + ); + total += tracked.result; + const outcome = terminalResourceOutcome( + name, + tracked.result, + tracked.outcome, + performance.now() - startedAt, + ); + resourceOutcomes.push(outcome); + if ( + outcome.status === "failed" || + outcome.status === "partial" || + outcome.status === "truncated" + ) { + anyFailed = true; + } } catch (err) { anyFailed = true; + resourceOutcomes.push( + boundedOutcome({ + resource: name, + status: "failed", + durationMs: performance.now() - startedAt, + reasonCode: "collection_failed", + }), + ); getEvent()?.addWarning( `google-health ${name} sync failed for ${userId}: ${err}`, ); } + await persistProgress("in_progress", total, anyFailed, resourceOutcomes); } - return { total, anyFailed }; + return { total, anyFailed, resources: resourceOutcomes }; }); const total = cycle.result.total; - const anyFailed = cycle.result.anyFailed || cycle.hardFailures.length > 0; + let anyFailed = cycle.result.anyFailed || cycle.hardFailures.length > 0; + const suppliedResources = (cycle as unknown as { resources?: unknown }) + .resources; + const outcomeSource = Array.isArray(suppliedResources) + ? suppliedResources + : cycle.result.resources; + const resourceOutcomes = outcomeSource.map(boundedOutcome); if (opts.fullSync && cycle.deferredRollupKeys.length > 0) { try { @@ -91,6 +253,18 @@ export async function syncUserGoogleHealth( ); }); } catch (err) { + anyFailed = true; + resourceOutcomes.push( + boundedOutcome({ + resource: "rollup", + fetched: cycle.deferredRollupKeys.length, + mapped: cycle.deferredRollupKeys.length, + written: total, + status: "failed", + durationMs: 0, + reasonCode: "rollup_failed", + }), + ); getEvent()?.addWarning( `google-health: backfill rollup recompute failed for ${userId}: ${err}`, ); @@ -98,7 +272,8 @@ export async function syncUserGoogleHealth( } const allSoftSkipped = cycle.softSkipCount >= resources.length && total === 0; - const failed = anyFailed || allSoftSkipped; + const truncated = resourceOutcomes.some((resource) => resource.truncated); + const failed = anyFailed || allSoftSkipped || truncated; if (!failed) { await markSynced(userId); @@ -108,7 +283,23 @@ export async function syncUserGoogleHealth( annotate({ action: { name: "googleHealth.sync", details: { imported: total, failed } }, }); - return { imported: total, failed }; + const state: GoogleHealthSyncState = truncated + ? "truncated" + : failed + ? total > 0 + ? "partial" + : "failed" + : total === 0 + ? "zero" + : "complete"; + await persistProgress(state, total, failed, resourceOutcomes); + return { + runId: progress.runId, + state, + imported: total, + failed, + resources: resourceOutcomes, + }; } /** Bounded fan-out width for the hourly Google Health cohort poll. */ @@ -128,7 +319,10 @@ export async function runGoogleHealthPollCohort( ): Promise<{ usersSynced: number; measurementsImported: number }> { const sync = opts.sync ?? - ((userId: string) => syncUserGoogleHealth(userId).then((r) => r.imported)); + (async (userId: string) => { + const result = await syncUserGoogleHealth(userId); + return result.imported; + }); const limit = pLimit(opts.concurrency ?? GOOGLE_HEALTH_POLL_CONCURRENCY); let usersSynced = 0;