From 5cfcb69ef2ffeb7105e5747964cce87bfe5a27d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Wed, 29 Jul 2026 14:49:12 +0200 Subject: [PATCH] feat(01-21): add guarded Google sync progress - persist one bounded owner-scoped current-run envelope - reject stale writers and materialize interrupted runs safely --- .../migration.sql | 2 + prisma/schema.prisma | 3 + src/lib/google-health/sync-progress.ts | 315 ++++++++++++++++++ 3 files changed, 320 insertions(+) create mode 100644 prisma/migrations/0287_google_health_sync_progress/migration.sql create mode 100644 src/lib/google-health/sync-progress.ts diff --git a/prisma/migrations/0287_google_health_sync_progress/migration.sql b/prisma/migrations/0287_google_health_sync_progress/migration.sql new file mode 100644 index 000000000..a2cdf3fc1 --- /dev/null +++ b/prisma/migrations/0287_google_health_sync_progress/migration.sql @@ -0,0 +1,2 @@ +ALTER TABLE "google_health_connections" +ADD COLUMN "sync_progress" JSONB; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 4c6b771d2..5d0562c6f 100644 --- a/prisma/schema.prisma +++ b/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") diff --git a/src/lib/google-health/sync-progress.ts b/src/lib/google-health/sync-progress.ts new file mode 100644 index 000000000..e34e53cd3 --- /dev/null +++ b/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>(); + +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) + : ({} as Record); + 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) + : ({} as Record); + 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( + userId: string, + operation: () => Promise, +): Promise { + const previous = progressWriteTails.get(userId) ?? Promise.resolve(); + let release = () => {}; + const tail = new Promise((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 { + 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 { + 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, + now = new Date(), +): Promise { + // 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 { + 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).runId === "string" + ? String((stored as Record).runId).slice(0, 128) + : ""; + if (!runId) return null; + const updatedAt = isoDate( + (stored as Record).updatedAt, + new Date(0), + ); + return progressEnvelope(stored, { + runId, + now: new Date(updatedAt), + }); +} + +export async function readGoogleHealthSyncProgress( + userId: string, + now = new Date(), +): Promise { + 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); +}