feat(01-21): add fair Google sync outcomes

- finish workout and low-cost resources before dense metrics
- report bounded terminal outcomes and invalidate workout caches

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<GoogleHealthClientOutcome>();
+
+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<GoogleHealthClientOutcome["reasonCode"], null>,
+): void {
+  const tracker = clientOutcomeStorage.getStore();
+  if (tracker && !tracker.reasonCode) tracker.reasonCode = reasonCode;
+}
+
+export async function runWithGoogleHealthClientOutcome<T>(
+  fn: () => Promise<T>,
+): 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<GoogleHealthRollupPoint[]> => {
     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;
   };
 

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 };
 }
 

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 } },

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<GoogleHealthResourceStatus>([
+  "pending",
+  "complete",
+  "partial",
+  "empty",
+  "truncated",
+  "failed",
+]);
+const REASON_CODES = new Set<GoogleHealthReasonCode>([
+  "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<string, unknown>)
+      : {};
+  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<GoogleHealthSyncResult> {
   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<void> => {
+    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;