feat(01-21): add guarded Google sync progress

- persist one bounded owner-scoped current-run envelope
- reject stale writers and materialize interrupted runs safely

prisma/migrations/0287_google_health_sync_progress/migration.sql

@@ -0,0 +1,2 @@
+ALTER TABLE "google_health_connections"
+ADD COLUMN "sync_progress" JSONB;

prisma/schema.prisma

@@ -3857,6 +3857,9 @@ model GoogleHealthConnection {
   /// pre-fix externalId (a re-scored night minted parallel copies). Null =
   /// repair pending (or the connection predates the fix).
   sleepRepairedAt     DateTime? @map("sleep_repaired_at")
+  /// One bounded, owner-scoped current sync run. This is replaced at the start
+  /// of every run and never retains provider payloads or historical runs.
+  syncProgress        Json?     @map("sync_progress")
   createdAt           DateTime  @default(now()) @map("created_at")
   updatedAt           DateTime  @updatedAt @map("updated_at")
 

src/lib/google-health/sync-progress.ts

@@ -0,0 +1,315 @@
+import { randomBytes } from "node:crypto";
+
+import { Prisma } from "@/generated/prisma/client";
+import { prisma } from "@/lib/db";
+
+export const GOOGLE_HEALTH_PROGRESS_STALE_MS = 30 * 60 * 1000;
+export const GOOGLE_HEALTH_PROGRESS_MAX_RESOURCES = 16;
+
+const MAX_COUNTER = 2_147_483_647;
+const MAX_DURATION_MS = 24 * 60 * 60 * 1000;
+const progressWriteTails = new Map<string, Promise<void>>();
+
+const RUN_STATES = new Set([
+  "in_progress",
+  "complete",
+  "partial",
+  "zero",
+  "truncated",
+  "failed",
+  "interrupted",
+] as const);
+const RESOURCE_STATES = new Set([
+  "pending",
+  "complete",
+  "partial",
+  "empty",
+  "truncated",
+  "failed",
+] as const);
+const REASON_CODES = new Set([
+  "collection_failed",
+  "token_failed",
+  "upsert_failed",
+  "rollup_failed",
+  "existing_page_limit",
+] as const);
+
+export type GoogleHealthSyncState =
+  | "in_progress"
+  | "complete"
+  | "partial"
+  | "zero"
+  | "truncated"
+  | "failed"
+  | "interrupted";
+
+export type GoogleHealthResourceStatus =
+  "pending" | "complete" | "partial" | "empty" | "truncated" | "failed";
+
+export type GoogleHealthReasonCode =
+  | "collection_failed"
+  | "token_failed"
+  | "upsert_failed"
+  | "rollup_failed"
+  | "existing_page_limit";
+
+export interface GoogleHealthResourceOutcome {
+  resource: string;
+  pages: number;
+  fetched: number;
+  mapped: number;
+  written: number;
+  status: GoogleHealthResourceStatus;
+  durationMs: number;
+  truncated: boolean;
+  reasonCode: GoogleHealthReasonCode | null;
+}
+
+export interface GoogleHealthSyncProgress {
+  runId: string;
+  state: GoogleHealthSyncState;
+  startedAt: string;
+  updatedAt: string;
+  terminalAt?: string;
+  imported: number;
+  failed: boolean;
+  resources: GoogleHealthResourceOutcome[];
+}
+
+const INITIAL_RESOURCES = [
+  "workout",
+  "sleep",
+  "bounded-metrics",
+  "activity",
+  "dense-heart-rate",
+] as const;
+
+function boundedInteger(value: unknown, maximum = MAX_COUNTER): number {
+  if (typeof value !== "number" || !Number.isFinite(value)) return 0;
+  return Math.min(maximum, Math.max(0, Math.trunc(value)));
+}
+
+function sanitizedResourceName(value: unknown): string {
+  if (typeof value !== "string") return "unknown";
+  const normalized = value.toLowerCase().replace(/[^a-z0-9-]/g, "-");
+  return normalized.replace(/-+/g, "-").slice(0, 48) || "unknown";
+}
+
+function resourceOutcome(
+  value: unknown,
+  fallbackResource = "unknown",
+): GoogleHealthResourceOutcome {
+  const record =
+    value && typeof value === "object"
+      ? (value as Record<string, unknown>)
+      : ({} as Record<string, unknown>);
+  const status = RESOURCE_STATES.has(
+    record.status as GoogleHealthResourceStatus,
+  )
+    ? (record.status as GoogleHealthResourceStatus)
+    : "pending";
+  const reasonCode = REASON_CODES.has(
+    record.reasonCode as GoogleHealthReasonCode,
+  )
+    ? (record.reasonCode as GoogleHealthReasonCode)
+    : null;
+  return {
+    resource: sanitizedResourceName(record.resource ?? fallbackResource),
+    pages: boundedInteger(record.pages),
+    fetched: boundedInteger(record.fetched),
+    mapped: boundedInteger(record.mapped),
+    written: boundedInteger(record.written),
+    status,
+    durationMs: boundedInteger(record.durationMs, MAX_DURATION_MS),
+    truncated: record.truncated === true,
+    reasonCode,
+  };
+}
+
+function initialResources(): GoogleHealthResourceOutcome[] {
+  return INITIAL_RESOURCES.map((resource) => resourceOutcome({}, resource));
+}
+
+function isoDate(value: unknown, fallback: Date): string {
+  if (typeof value !== "string") return fallback.toISOString();
+  const parsed = new Date(value);
+  return Number.isFinite(parsed.getTime())
+    ? parsed.toISOString()
+    : fallback.toISOString();
+}
+
+function progressEnvelope(
+  value: unknown,
+  fallback: {
+    runId: string;
+    now: Date;
+    startedAt?: string;
+  },
+): GoogleHealthSyncProgress {
+  const record =
+    value && typeof value === "object"
+      ? (value as Record<string, unknown>)
+      : ({} as Record<string, unknown>);
+  const state = RUN_STATES.has(record.state as GoogleHealthSyncState)
+    ? (record.state as GoogleHealthSyncState)
+    : "in_progress";
+  const resources = Array.isArray(record.resources)
+    ? record.resources
+        .slice(0, GOOGLE_HEALTH_PROGRESS_MAX_RESOURCES)
+        .map((resource) => resourceOutcome(resource))
+    : initialResources();
+  const envelope: GoogleHealthSyncProgress = {
+    runId: fallback.runId,
+    state,
+    startedAt: isoDate(record.startedAt ?? fallback.startedAt, fallback.now),
+    updatedAt: fallback.now.toISOString(),
+    imported: boundedInteger(record.imported ?? record.written),
+    failed: record.failed === true,
+    resources,
+  };
+  if (state !== "in_progress") {
+    envelope.terminalAt = isoDate(record.terminalAt, fallback.now);
+  }
+  return envelope;
+}
+
+function asJson(value: GoogleHealthSyncProgress): Prisma.InputJsonValue {
+  return value as unknown as Prisma.InputJsonValue;
+}
+
+async function serializeProgressWrite<T>(
+  userId: string,
+  operation: () => Promise<T>,
+): Promise<T> {
+  const previous = progressWriteTails.get(userId) ?? Promise.resolve();
+  let release = () => {};
+  const tail = new Promise<void>((resolve) => {
+    release = resolve;
+  });
+  progressWriteTails.set(userId, tail);
+  await previous;
+  try {
+    return await operation();
+  } finally {
+    release();
+    if (progressWriteTails.get(userId) === tail) {
+      progressWriteTails.delete(userId);
+    }
+  }
+}
+
+export async function startGoogleHealthSyncProgress(
+  userId: string,
+  now = new Date(),
+): Promise<GoogleHealthSyncProgress> {
+  const runId = randomBytes(18).toString("base64url");
+  const envelope = progressEnvelope(
+    {
+      state: "in_progress",
+      startedAt: now.toISOString(),
+      imported: 0,
+      failed: false,
+      resources: initialResources(),
+    },
+    { runId, now },
+  );
+  await serializeProgressWrite(userId, () =>
+    prisma.googleHealthConnection.update({
+      where: { userId },
+      data: { syncProgress: asJson(envelope) },
+    }),
+  );
+  return envelope;
+}
+
+async function guardedProgressUpdate(
+  userId: string,
+  runId: string,
+  envelope: GoogleHealthSyncProgress,
+): Promise<boolean> {
+  const result = await serializeProgressWrite(userId, () =>
+    prisma.googleHealthConnection.updateMany({
+      where: {
+        userId,
+        syncProgress: {
+          path: ["runId"],
+          equals: runId,
+        },
+      },
+      data: { syncProgress: asJson(envelope) },
+    }),
+  );
+  return result.count === 1;
+}
+
+export async function updateGoogleHealthSyncProgress(
+  userId: string,
+  runId: string,
+  update: Record<string, unknown>,
+  now = new Date(),
+): Promise<boolean> {
+  // Issue the guarded write without a preceding read. Besides avoiding a
+  // read/modify/write window, this makes the database choose one winner when a
+  // delayed terminal competes with a new run. Callers pass the complete
+  // bounded envelope (including resources and startedAt) on each update.
+  const merged = progressEnvelope(update, {
+    runId,
+    now,
+    startedAt:
+      typeof update.startedAt === "string" ? update.startedAt : undefined,
+  });
+  return guardedProgressUpdate(userId, runId, merged);
+}
+
+async function loadProgress(
+  userId: string,
+): Promise<GoogleHealthSyncProgress | null> {
+  const connection = await prisma.googleHealthConnection.findUnique({
+    where: { userId },
+    select: { syncProgress: true },
+  });
+  const stored = connection?.syncProgress;
+  if (!stored || typeof stored !== "object" || Array.isArray(stored)) {
+    return null;
+  }
+  const runId =
+    typeof (stored as Record<string, unknown>).runId === "string"
+      ? String((stored as Record<string, unknown>).runId).slice(0, 128)
+      : "";
+  if (!runId) return null;
+  const updatedAt = isoDate(
+    (stored as Record<string, unknown>).updatedAt,
+    new Date(0),
+  );
+  return progressEnvelope(stored, {
+    runId,
+    now: new Date(updatedAt),
+  });
+}
+
+export async function readGoogleHealthSyncProgress(
+  userId: string,
+  now = new Date(),
+): Promise<GoogleHealthSyncProgress | null> {
+  const current = await loadProgress(userId);
+  if (!current || current.state !== "in_progress") return current;
+  const updatedAt = new Date(current.updatedAt).getTime();
+  if (now.getTime() - updatedAt <= GOOGLE_HEALTH_PROGRESS_STALE_MS) {
+    return current;
+  }
+
+  const interrupted = progressEnvelope(
+    {
+      ...current,
+      state: "interrupted",
+      failed: true,
+      terminalAt: now.toISOString(),
+    },
+    { runId: current.runId, now, startedAt: current.startedAt },
+  );
+  if (await guardedProgressUpdate(userId, current.runId, interrupted)) {
+    return interrupted;
+  }
+  return loadProgress(userId);
+}