From d10179f9a3588523419b173be14f6cc2ae724d5f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 15 Oct 2026 11:01:16 +0200 Subject: [PATCH] feat(integrations): resolve one health state per connection Fold the ledger state, liveness verdict, plan and scope gaps and the provider reauth flag into ok / limited / degraded / action_required with a reason code, the one action, the unavailable types and redacted diagnostics. Keep the last ten sync attempts per connection for those diagnostics, publish the state on the integrations envelope, every per-provider status route and the Apple Health config read, and page the operator once per failure streak instead of once a day. --- .../migration.sql | 40 ++ prisma/schema.prisma | 47 ++ src/app/api/fitbit/status/route.ts | 5 + src/app/api/google-health/status/route.ts | 5 + src/app/api/integrations/healthkit/route.ts | 8 +- src/app/api/integrations/status/route.ts | 34 ++ src/app/api/nightscout/status/route.ts | 5 + src/app/api/oura/status/route.ts | 5 + src/app/api/polar/status/route.ts | 5 + src/app/api/strava/status/route.ts | 5 + src/app/api/whoop/status/route.ts | 5 + src/app/api/withings/status/route.ts | 5 + src/lib/integrations/__tests__/status.test.ts | 20 +- .../integrations/connection-health-load.ts | 436 ++++++++++++++++++ src/lib/integrations/connection-health.ts | 390 ++++++++++++++++ src/lib/integrations/status.ts | 150 +++++- src/lib/integrations/watchdog-classify.ts | 213 +++++++++ 17 files changed, 1351 insertions(+), 27 deletions(-) create mode 100644 prisma/migrations/0400_integration_health_watchdog/migration.sql create mode 100644 src/lib/integrations/connection-health-load.ts create mode 100644 src/lib/integrations/connection-health.ts create mode 100644 src/lib/integrations/watchdog-classify.ts diff --git a/prisma/migrations/0400_integration_health_watchdog/migration.sql b/prisma/migrations/0400_integration_health_watchdog/migration.sql new file mode 100644 index 000000000..6f13cefe8 --- /dev/null +++ b/prisma/migrations/0400_integration_health_watchdog/migration.sql @@ -0,0 +1,40 @@ +-- v1.42.4 — one health state per connection, and a watchdog behind it. +-- +-- `integration_statuses.recent_calls` keeps the last ten sync attempts per +-- connection as `{ endpoint, status, at, errorCode }`: the leg name, the +-- outcome, the instant and a short error code. No message, token or value is +-- stored in it; it is the source of the redacted diagnostics a user can copy +-- into a bug report. +-- +-- `integration_watches` holds what the integration watchdog observed per +-- connection: whether a source that normally delivers has gone silent, and +-- the one operator incident open for it, so an alert is sent once per +-- incident and a recovery note once when it closes. +-- +-- Additive only: one nullable column, one new table, no rewrite. +ALTER TABLE "integration_statuses" ADD COLUMN IF NOT EXISTS "recent_calls" JSONB; + +CREATE TABLE IF NOT EXISTS "integration_watches" ( + "id" TEXT NOT NULL, + "user_id" TEXT NOT NULL, + "integration" TEXT NOT NULL, + "silent_since" TIMESTAMP(3), + "incident_reason" TEXT, + "incident_opened_at" TIMESTAMP(3), + "alerted_at" TIMESTAMP(3), + "checked_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "integration_watches_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX IF NOT EXISTS "integration_watches_user_id_integration_key" + ON "integration_watches"("user_id", "integration"); + +DO $$ BEGIN + ALTER TABLE "integration_watches" + ADD CONSTRAINT "integration_watches_user_id_fkey" + FOREIGN KEY ("user_id") REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE; +EXCEPTION WHEN duplicate_object THEN NULL; +END $$; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 9834872e9..f00b0bfd2 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -804,6 +804,7 @@ model User { idempotencyKeys IdempotencyKey[] devices Device[] integrationStatuses IntegrationStatus[] + integrationWatches IntegrationWatch[] coachConversations CoachConversation[] coachFacts CoachFact[] coachPlans CoachPlan[] @@ -5047,6 +5048,15 @@ model IntegrationStatus { // a failure recorded without a leg), under which any success clears the row. failingLegs Json? @map("failing_legs") alertedAt DateTime? @map("alerted_at") // last time admin Telegram alert fired (idempotency guard) + // The last ten sync attempts, newest first, as a JSON array of + // `{ endpoint, status, at, errorCode }`. `endpoint` is the sync leg name the + // code passes (never a URL), `status` is `ok` or the failure kind, and + // `errorCode` is a short closed-shape code ("401", "invalid_grant") or null. + // No message, no token, no value: this is what the redacted diagnostics on + // the connection health read publish, so it is shaped to be safe to copy + // into a bug report. Written by `appendRecentCall` in + // `src/lib/integrations/status.ts`, read by `resolveConnectionHealth`. + recentCalls Json? @map("recent_calls") createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") @@ -5058,6 +5068,43 @@ model IntegrationStatus { @@map("integration_statuses") } +// What the integration watchdog (`src/lib/integrations/watchdog.ts`) knows +// about one connection: whether a source that normally delivers has gone +// silent, and whether an operator incident is open for it. One row per +// (user, connection) the watchdog has looked at. `integration` carries the +// connection key, which is the IntegrationStatus key set plus "apple-health", +// so it is a separate table rather than more columns on IntegrationStatus, +// whose readers assume every row is a polled provider. +// +// Operational, like IntegrationStatus: it records what this host observed, so +// it is excluded from backups and removed with the account's data. +model IntegrationWatch { + id String @id @default(cuid()) + userId String @map("user_id") + integration String + /// Set while the source is unexpectedly silent: the newest data instant at + /// the moment the watchdog flagged it. Null when the source is not flagged. + /// The connection health read surfaces it as `degraded` / + /// `unexpected_silence` until data arrives again. + silentSince DateTime? @map("silent_since") + /// Reason code of the operator incident currently open for this connection, + /// null when none is. One incident at a time per connection. + incidentReason String? @map("incident_reason") + incidentOpenedAt DateTime? @map("incident_opened_at") + /// When the operator was told about the open incident, or when the watchdog + /// adopted an alert the sync ledger had already sent for the same streak. + /// Null while an incident is open but not yet announced. + alertedAt DateTime? @map("alerted_at") + checkedAt DateTime @default(now()) @map("checked_at") + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @updatedAt @map("updated_at") + + user User @relation(fields: [userId], references: [id], onDelete: Cascade) + + @@unique([userId, integration]) + @@map("integration_watches") +} + // ─── App Settings ───────────────────────────────────────── model AppSettings { diff --git a/src/app/api/fitbit/status/route.ts b/src/app/api/fitbit/status/route.ts index ac51d14f1..60a695256 100644 --- a/src/app/api/fitbit/status/route.ts +++ b/src/app/api/fitbit/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; /** * Get Fitbit / Google Health connection status for the current user (v1.12.0). @@ -53,5 +54,9 @@ export const GET = apiHandler(async () => { tokenExpiresAt: connection.tokenExpiresAt, backfillCompleted: !!connection.backfillCompletedAt, scope: connection.scope, + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "fitbit"), }); }); diff --git a/src/app/api/google-health/status/route.ts b/src/app/api/google-health/status/route.ts index 38b002f82..d63ea0c12 100644 --- a/src/app/api/google-health/status/route.ts +++ b/src/app/api/google-health/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; /** * Get Google Health connection status for the current user (v1.26.0). @@ -58,5 +59,9 @@ export const GET = apiHandler(async () => { backfillCompleted: !!connection.backfillCompletedAt, needsReauth: connection.needsReauth, scope: connection.scope, + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "google-health"), }); }); diff --git a/src/app/api/integrations/healthkit/route.ts b/src/app/api/integrations/healthkit/route.ts index 886e489ea..7a3307145 100644 --- a/src/app/api/integrations/healthkit/route.ts +++ b/src/app/api/integrations/healthkit/route.ts @@ -27,6 +27,7 @@ import { withHealthKitArrivals, } from "@/lib/integrations/healthkit-type-sync"; import { getSourceFreshness } from "@/lib/integrations/metric-freshness"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; import { classifyMetricFreshness, PUSH_CADENCE, @@ -173,10 +174,14 @@ export const GET = apiHandler(async () => { // progress; the phone's queue and throttle state stay on the phone and are // not guessed at. Fail-soft for the same reason the freshness read is — // a progress line is never worth 500-ing the config the iOS client needs. - const [samples, arrivals, syncProgress] = await Promise.all([ + // `health` is the unified connection state; fail-soft to null for the same + // reason as the reads beside it, so the config the phone needs never 500s + // over a diagnostic. + const [samples, arrivals, syncProgress, health] = await Promise.all([ getSourceFreshness(user.id, "APPLE_HEALTH").catch(() => []), getHealthKitTypeArrivals(user.id).catch(() => null), getAppleHealthSyncProgress(user.id).catch(() => null), + loadOneConnectionHealth(user.id, "apple-health").catch(() => null), ]); // No `connected` hint: Apple Health has nothing to connect or disconnect, so // "never delivered" resolves to `pending_first_sync`, not `disconnected`. @@ -202,6 +207,7 @@ export const GET = apiHandler(async () => { arrivals, ), syncProgress, + health, }); }); diff --git a/src/app/api/integrations/status/route.ts b/src/app/api/integrations/status/route.ts index fe97a54a2..f7a38cf98 100644 --- a/src/app/api/integrations/status/route.ts +++ b/src/app/api/integrations/status/route.ts @@ -51,6 +51,8 @@ import { import { getPolarClientCredentials } from "@/lib/polar/credentials"; import { getStravaClientCredentials } from "@/lib/strava/credentials"; import { hasActivityScope } from "@/lib/withings/client"; +import { loadConnectionHealth } from "@/lib/integrations/connection-health-load"; +import type { ConnectionHealth } from "@/lib/integrations/connection-health"; export const dynamic = "force-dynamic"; @@ -159,6 +161,20 @@ export const GET = apiHandler(async () => { getStravaClientCredentials(user.id).then((c) => !!c), ]); + // The unified per-connection state. Resolved by the same loader the + // per-provider routes and the integration watchdog use, so the envelope and + // every other surface say the same thing about a connection. + const health = await loadConnectionHealth(user.id, [ + "withings", + "whoop", + "fitbit", + "google-health", + "polar", + "oura", + "strava", + "nightscout", + ]); + // F-SYNC-1 — per-metric-type last-value timestamps so the card can show that a // single metric type has gone silent even while the integration reads green // ("connected · 5 min ago"). Fail-soft: this is an honesty signal, never worth @@ -301,6 +317,7 @@ export const GET = apiHandler(async () => { parseWithingsApiPlanState(withingsConn.apiPlanState), ) : [], + health: health["withings"], ...resolve("withings", withingsStatus, { connected: !!withingsConn, configured: withingsConfigured, @@ -324,6 +341,7 @@ export const GET = apiHandler(async () => { backfillCompleted: whoopConnected ? !!whoopConn?.backfillCompletedAt : null, + health: health["whoop"], ...resolve("whoop", whoopStatus, { connected: whoopConnected, configured: whoopConfigured, @@ -341,6 +359,7 @@ export const GET = apiHandler(async () => { ? fitbitConn.tokenExpiresAt.getTime() <= now : null, backfillCompleted: fitbitConn ? !!fitbitConn.backfillCompletedAt : null, + health: health["fitbit"], ...resolve("fitbit", fitbitStatus, { connected: fitbitConnected, configured: fitbitConfigured, @@ -364,6 +383,7 @@ export const GET = apiHandler(async () => { // user-revoked grant) surfaced from the connection row; the card reads it // to raise a distinct "Reconnect" CTA separate from parked/disconnected. needsReauth: googleHealthConn ? googleHealthConn.needsReauth : false, + health: health["google-health"], ...resolve("google-health", googleHealthStatus, { connected: googleHealthConnected, configured: googleHealthConfigured, @@ -382,6 +402,7 @@ export const GET = apiHandler(async () => { hasOwnCredentials: !!dbUser?.polarClientIdEncrypted && !!dbUser?.polarClientSecretEncrypted, + health: health["polar"], ...resolve("polar", polarStatus, { connected: polarConnected, configured: polarConnected, @@ -400,6 +421,7 @@ export const GET = apiHandler(async () => { scopeReconnect: ouraConnected ? ouraScopeReconnectList(readOuraScopeDenied(dbUser?.ouraScopeDenied)) : [], + health: health["oura"], ...resolve("oura", ouraStatus, { connected: ouraConnected, configured: ouraConnected, @@ -416,6 +438,7 @@ export const GET = apiHandler(async () => { !!dbUser?.stravaClientSecretEncrypted, // Strava writes Workout rows, not Measurements — its freshness list // carries the `WORKOUTS` pseudo-entry alone. + health: health["strava"], ...resolve("strava", stravaStatus, { connected: stravaConnected, configured: stravaConnected, @@ -431,6 +454,7 @@ export const GET = apiHandler(async () => { configured: nightscoutConfigured, hasToken: !!dbUser?.nightscoutTokenEncrypted, allowPrivateHost: dbUser?.nightscoutAllowPrivateHost ?? false, + health: health["nightscout"], ...resolve("nightscout", nightscoutStatus, { connected: nightscoutConfigured, configured: nightscoutConfigured, @@ -442,6 +466,16 @@ export const GET = apiHandler(async () => { interface IntegrationViewModel { integration: IntegrationKey; + /** + * The one server-resolved state per connection (ok / limited / degraded / + * action_required), null when the provider is not connected. Every other + * health-shaped field on this entry (`state`, `lastError`, + * `consecutiveFailuresByKind`, `syncHealth`, `tokenExpired`, `needsReauth`, + * `hasActivityScope`, `planUnavailable`, `scopeReconnect`) is folded into + * it and goes once the settings cards render `health`; no other client + * reads this envelope. + */ + health: ConnectionHealth | null; state: string; lastSuccessAt: string | null; lastAttemptAt: string | null; diff --git a/src/app/api/nightscout/status/route.ts b/src/app/api/nightscout/status/route.ts index f2cda6aa0..dafd647ce 100644 --- a/src/app/api/nightscout/status/route.ts +++ b/src/app/api/nightscout/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; import { getIntegrationStatus } from "@/lib/integrations/status"; import { INTEGRATION_CADENCE, @@ -62,5 +63,9 @@ export const GET = apiHandler(async () => { failingSinceAt: status.failingSince, cadence: INTEGRATION_CADENCE["nightscout"], }), + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "nightscout"), }); }); diff --git a/src/app/api/oura/status/route.ts b/src/app/api/oura/status/route.ts index ba3abf574..20924f365 100644 --- a/src/app/api/oura/status/route.ts +++ b/src/app/api/oura/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; import { getOuraClientCredentials } from "@/lib/oura/credentials"; import { getIntegrationStatus } from "@/lib/integrations/status"; import { @@ -70,5 +71,9 @@ export const GET = apiHandler(async () => { failingSinceAt: status.failingSince, cadence: INTEGRATION_CADENCE["oura"], }), + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "oura"), }); }); diff --git a/src/app/api/polar/status/route.ts b/src/app/api/polar/status/route.ts index 92cd26d79..1f11d7f0e 100644 --- a/src/app/api/polar/status/route.ts +++ b/src/app/api/polar/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; import { getPolarClientCredentials } from "@/lib/polar/credentials"; import { getIntegrationStatus } from "@/lib/integrations/status"; import { @@ -71,5 +72,9 @@ export const GET = apiHandler(async () => { failingSinceAt: status.failingSince, cadence: INTEGRATION_CADENCE["polar"], }), + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "polar"), }); }); diff --git a/src/app/api/strava/status/route.ts b/src/app/api/strava/status/route.ts index c80c14a71..9906c250b 100644 --- a/src/app/api/strava/status/route.ts +++ b/src/app/api/strava/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; import { getStravaClientCredentials } from "@/lib/strava/credentials"; import { getIntegrationStatus } from "@/lib/integrations/status"; import { @@ -70,5 +71,9 @@ export const GET = apiHandler(async () => { failingSinceAt: status.failingSince, cadence: INTEGRATION_CADENCE["strava"], }), + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "strava"), }); }); diff --git a/src/app/api/whoop/status/route.ts b/src/app/api/whoop/status/route.ts index ae8ca1221..bfe462376 100644 --- a/src/app/api/whoop/status/route.ts +++ b/src/app/api/whoop/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; /** * Get WHOOP connection status for the current user (v1.11.0). @@ -54,5 +55,9 @@ export const GET = apiHandler(async () => { tokenExpiresAt: connection.tokenExpiresAt, backfillCompleted: !!connection.backfillCompletedAt, scope: connection.scope, + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "whoop"), }); }); diff --git a/src/app/api/withings/status/route.ts b/src/app/api/withings/status/route.ts index 7e4311481..baeb2d754 100644 --- a/src/app/api/withings/status/route.ts +++ b/src/app/api/withings/status/route.ts @@ -2,6 +2,7 @@ import { prisma } from "@/lib/db"; import { apiHandler, requireAuth } from "@/lib/api-handler"; import { annotate, getEvent } from "@/lib/logging/context"; import { apiSuccess } from "@/lib/api-response"; +import { loadOneConnectionHealth } from "@/lib/integrations/connection-health-load"; import { decrypt, encrypt } from "@/lib/crypto"; import { getUserWithingsCredentials } from "@/lib/withings/credentials"; import { hasActivityScope, refreshAccessToken } from "@/lib/withings/client"; @@ -118,5 +119,9 @@ export const GET = apiHandler(async () => { // never re-authed since v1.4.25 — it counts as "missing activity". scope: connection.scope, hasActivityScope: hasActivityScope(connection.scope), + // The one server-resolved state (ok / limited / degraded / + // action_required) every client renders. The fields above stay for + // clients that still read them; see `ConnectionHealth` in the API docs. + health: await loadOneConnectionHealth(user.id, "withings"), }); }); diff --git a/src/lib/integrations/__tests__/status.test.ts b/src/lib/integrations/__tests__/status.test.ts index 88eb2b38d..a4d365a01 100644 --- a/src/lib/integrations/__tests__/status.test.ts +++ b/src/lib/integrations/__tests__/status.test.ts @@ -20,6 +20,7 @@ vi.mock("@/lib/logging/context", () => ({ vi.mock("@/lib/db", () => ({ prisma: { + $executeRaw: vi.fn(), integrationStatus: { findUnique: vi.fn(), upsert: vi.fn(), @@ -363,8 +364,12 @@ describe("recordSyncFailure — alert window", () => { expect(dispatchNotification).not.toHaveBeenCalled(); }); - it("DOES dispatch a fresh alert after 24h have elapsed", async () => { - // Existing bucket at 5 → in-memory increment lands at 6. + it("does NOT re-page the same streak once 24h have elapsed", async () => { + // One page per streak. `alertedAt` is cleared by every path that ends a + // streak, so a stamped value means this streak was already announced, + // however long ago. The daily re-page this test used to pin turned a + // week-long outage into seven identical messages; the integration + // watchdog owns the rest of the incident, including its recovery note. vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ consecutiveFailuresByKind: { transient: 5, @@ -377,15 +382,6 @@ describe("recordSyncFailure — alert window", () => { vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({ alertedAt: new Date(Date.now() - 25 * 60 * 60 * 1000), } as never); - vi.mocked(prisma.user.findUnique).mockResolvedValueOnce({ - email: "u@example.com", - } as never); - vi.mocked(prisma.user.findMany).mockResolvedValueOnce([ - { id: "admin-1" }, - ] as never); - vi.mocked(prisma.integrationStatus.update).mockResolvedValueOnce( - {} as never, - ); await recordSyncFailure({ userId: "u1", @@ -394,7 +390,7 @@ describe("recordSyncFailure — alert window", () => { message: "Withings refresh error: 503 - upstream", }); - expect(dispatchNotification).toHaveBeenCalledOnce(); + expect(dispatchNotification).not.toHaveBeenCalled(); }); }); diff --git a/src/lib/integrations/connection-health-load.ts b/src/lib/integrations/connection-health-load.ts new file mode 100644 index 000000000..a309f8253 --- /dev/null +++ b/src/lib/integrations/connection-health-load.ts @@ -0,0 +1,436 @@ +/** + * Gather what `resolveConnectionHealth` needs, for one account, in one place. + * + * Every surface that publishes a connection's health (the consolidated + * integrations envelope, the per-provider status routes, the Apple Health + * config read, the watchdog) calls this loader, so the four states cannot + * drift into one opinion per route. It reads: + * + * - the sync ledger rows (state, streak, failing legs, recent calls), + * - which connections exist and the provider-side extras that bear on + * health (Withings plan + activity scope, Oura scope gaps, Google Health + * `needsReauth`), + * - the watchdog's silence flags, + * - the newest data per connected source (one index probe each). + * + * Read-only. A connection that does not exist resolves to null. + */ +import { prisma } from "@/lib/db"; +import type { MeasurementSource } from "@/generated/prisma/client"; +import type { IntegrationKey } from "./status"; +import { + INTEGRATION_CADENCE, + PUSH_CADENCE, + resolveSyncVerdict, + type SyncHealth, +} from "./sync-verdict"; +import { + CONNECTION_KEYS, + RECENT_CALLS_LIMIT, + readRecentCalls, + resolveConnectionHealth, + type ConnectionHealth, + type ConnectionKey, + type RecentCall, +} from "./connection-health"; +import { SOURCE_PROFILES } from "./watchdog-classify"; +import { + INTEGRATION_MEASUREMENT_SOURCE, + INTEGRATION_WORKOUT_SOURCE, +} from "./metric-freshness"; +import { parseWithingsApiPlanState } from "@/lib/withings/api-plan"; +import { withingsPlanUnavailableFeatures } from "@/lib/withings/plan-features"; +import { hasActivityScope } from "@/lib/withings/client"; +import { readOuraScopeDenied } from "@/lib/oura/credentials"; +import { ouraScopeReconnectList } from "@/lib/oura/scope"; + +export type ConnectionHealthMap = Record< + ConnectionKey, + ConnectionHealth | null +>; + +/** Everything the loader read for one connection, beside the resolved health. */ +export interface ConnectionFacts { + key: ConnectionKey; + connected: boolean; + syncHealth: SyncHealth; + lastDataAt: Date | null; + /** The sync ledger holds an unresolved operator page for this streak. */ + ledgerAlerted: boolean; + /** Health resolved WITHOUT the watchdog's silence flag (the watchdog's own input). */ + healthWithoutSilence: ConnectionHealth | null; + /** Health as published (silence flag folded in). */ + health: ConnectionHealth | null; + silentSince: Date | null; +} + +interface LedgerRow { + integration: string; + state: string; + lastSuccessAt: Date | null; + lastAttemptAt: Date | null; + failingSinceAt: Date | null; + failingLegs: unknown; + recentCalls: unknown; + alertedAt: Date | null; +} + +/** Newest measurement of one source: one probe on `(userId, source, measuredAt DESC)`. */ +async function newestMeasurementAt( + userId: string, + source: MeasurementSource, +): Promise { + const row = await prisma.measurement.findFirst({ + where: { userId, source, deletedAt: null }, + orderBy: { measuredAt: "desc" }, + select: { measuredAt: true }, + }); + return row?.measuredAt ?? null; +} + +async function newestWorkoutAt( + userId: string, + source: MeasurementSource, +): Promise { + const row = await prisma.workout.findFirst({ + where: { userId, source }, + orderBy: { startedAt: "desc" }, + select: { startedAt: true }, + }); + return row?.startedAt ?? null; +} + +function newestOf(...values: Array): Date | null { + let max: Date | null = null; + for (const v of values) { + if (v && (max === null || v.getTime() > max.getTime())) max = v; + } + return max; +} + +function iso(value: Date | null | undefined): string | null { + return value ? value.toISOString() : null; +} + +function readLegs(value: unknown): string[] { + if (!Array.isArray(value)) return []; + return value.filter( + (v): v is string => + typeof v === "string" && /^[A-Za-z0-9_.:-]{1,48}$/.test(v), + ); +} + +/** + * Load the facts and resolved health for every connection key of one account. + * `keys` narrows the work for a single-provider route. + */ +export async function loadConnectionFacts( + userId: string, + keys: readonly ConnectionKey[] = CONNECTION_KEYS, + now: Date = new Date(), +): Promise> { + const wanted = new Set(keys); + const [ + ledgerRows, + watchRows, + dbUser, + withingsConn, + whoopConn, + fitbitConn, + googleHealthConn, + healthKitArrivals, + ] = await Promise.all([ + prisma.integrationStatus.findMany({ + where: { userId }, + select: { + integration: true, + state: true, + lastSuccessAt: true, + lastAttemptAt: true, + failingSinceAt: true, + failingLegs: true, + recentCalls: true, + alertedAt: true, + }, + }), + prisma.integrationWatch.findMany({ + where: { userId }, + select: { integration: true, silentSince: true }, + }), + prisma.user.findUnique({ + where: { id: userId }, + select: { + polarAccessTokenEncrypted: true, + ouraAccessTokenEncrypted: true, + ouraScopeDenied: true, + stravaAccessTokenEncrypted: true, + nightscoutUrlEncrypted: true, + healthKitLastSyncedAt: true, + }, + }), + wanted.has("withings") + ? prisma.withingsConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true, scope: true, apiPlanState: true }, + }) + : null, + wanted.has("whoop") + ? prisma.whoopConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true, whoopUserId: true }, + }) + : null, + wanted.has("fitbit") + ? prisma.fitbitConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true }, + }) + : null, + wanted.has("google-health") + ? prisma.googleHealthConnection.findUnique({ + where: { userId }, + select: { lastSyncedAt: true, needsReauth: true }, + }) + : null, + wanted.has("apple-health") + ? prisma.healthKitTypeSync.findMany({ + where: { userId }, + select: { type: true, lastReceivedAt: true, lastNewSampleAt: true }, + orderBy: { lastReceivedAt: "desc" }, + }) + : [], + ]); + + const ledger = new Map( + ledgerRows.map((row) => [row.integration, row]), + ); + const silence = new Map( + watchRows.map((row) => [row.integration, row.silentSince]), + ); + + const connected: Record = { + withings: !!withingsConn, + whoop: !!whoopConn?.whoopUserId, + fitbit: !!fitbitConn, + "google-health": !!googleHealthConn, + polar: !!dbUser?.polarAccessTokenEncrypted, + oura: !!dbUser?.ouraAccessTokenEncrypted, + strava: !!dbUser?.stravaAccessTokenEncrypted, + nightscout: !!dbUser?.nightscoutUrlEncrypted, + // Apple Health has nothing to connect; it is "connected" once the phone + // has delivered anything (a live batch, or rows from the export import). + "apple-health": false, + }; + const legacyLastSyncedAt: Partial> = { + withings: withingsConn?.lastSyncedAt ?? null, + whoop: whoopConn?.whoopUserId ? (whoopConn.lastSyncedAt ?? null) : null, + fitbit: fitbitConn?.lastSyncedAt ?? null, + "google-health": googleHealthConn?.lastSyncedAt ?? null, + }; + + // Newest data per wanted, connected source. Apple Health is probed even + // before it counts as connected, because delivered data is what connects it. + const lastData = new Map(); + await Promise.all( + [...wanted].map(async (key) => { + if (key === "apple-health") { + const measured = await newestMeasurementAt(userId, "APPLE_HEALTH"); + lastData.set(key, measured); + return; + } + if (!connected[key]) return; + const mSource = INTEGRATION_MEASUREMENT_SOURCE[key as IntegrationKey]; + const wSource = INTEGRATION_WORKOUT_SOURCE[key as IntegrationKey]; + const [m, w] = await Promise.all([ + mSource ? newestMeasurementAt(userId, mSource) : null, + wSource ? newestWorkoutAt(userId, wSource) : null, + ]); + lastData.set(key, newestOf(m, w)); + }), + ); + + const out = new Map(); + const nowMs = now.getTime(); + + for (const key of keys) { + if (key === "apple-health") { + out.set( + key, + appleHealthFacts({ + lastSyncedAt: dbUser?.healthKitLastSyncedAt ?? null, + arrivals: healthKitArrivals, + newestMeasuredAt: lastData.get("apple-health") ?? null, + silentSince: silence.get(key) ?? null, + now, + }), + ); + continue; + } + + const row = ledger.get(key); + const isConnected = connected[key]; + const syncHealth = resolveSyncVerdict({ + connected: isConnected, + configured: isConnected, + state: row?.state ?? "unknown", + lastSuccessAt: row?.lastSuccessAt ?? null, + lastAttemptAt: row?.lastAttemptAt ?? null, + failingSinceAt: row?.failingSinceAt ?? null, + legacyLastSyncedAt: legacyLastSyncedAt[key] ?? null, + cadence: INTEGRATION_CADENCE[key], + now: nowMs, + }); + + let planUnavailable: string[] = []; + let permissionUnavailable: string[] = []; + if (key === "withings" && withingsConn) { + planUnavailable = withingsPlanUnavailableFeatures( + parseWithingsApiPlanState(withingsConn.apiPlanState), + ); + // A connection whose grant predates the activity scope cannot read + // activity or sleep; one reconnect asks for it. A null scope is the + // same legacy case `hasActivityScope` already treats as missing. + if (!hasActivityScope(withingsConn.scope ?? null)) { + permissionUnavailable = ["activity", "sleep"]; + } + } + if (key === "oura" && isConnected) { + permissionUnavailable = ouraScopeReconnectList( + readOuraScopeDenied(dbUser?.ouraScopeDenied), + ); + } + + const lastDataAt = lastData.get(key) ?? null; + const silentSince = silence.get(key) ?? null; + const common = { + syncHealth, + connected: isConnected, + ledgerState: row?.state ?? null, + lastSuccessAt: iso(row?.lastSuccessAt ?? legacyLastSyncedAt[key] ?? null), + lastAttemptAt: iso(row?.lastAttemptAt), + failingSince: iso(row?.failingSinceAt), + failingLegs: readLegs(row?.failingLegs), + recentCalls: readRecentCalls(row?.recentCalls), + lastDataAt: iso(lastDataAt), + needsReauth: key === "google-health" && !!googleHealthConn?.needsReauth, + planUnavailable, + permissionUnavailable, + silenceWindowMs: SOURCE_PROFILES[key].silenceWindowMs, + now: nowMs, + }; + out.set(key, { + key, + connected: isConnected, + syncHealth, + lastDataAt, + ledgerAlerted: row?.alertedAt != null, + healthWithoutSilence: resolveConnectionHealth({ + ...common, + silentSince: null, + }), + health: resolveConnectionHealth({ + ...common, + silentSince: iso(silentSince), + }), + silentSince, + }); + } + return out; +} + +function appleHealthFacts(input: { + lastSyncedAt: Date | null; + arrivals: Array<{ + type: string; + lastReceivedAt: Date; + lastNewSampleAt: Date | null; + }>; + newestMeasuredAt: Date | null; + silentSince: Date | null; + now: Date; +}): ConnectionFacts { + const newestReceived = newestOf( + ...input.arrivals.map((a) => a.lastReceivedAt), + ); + const newestNewSample = newestOf( + ...input.arrivals.map((a) => a.lastNewSampleAt), + ); + // "Last import with data": a sync that brought a new sample. Rows from the + // export import carry no arrival record, so their newest sample stands in. + const lastDataAt = newestNewSample ?? input.newestMeasuredAt; + const connected = + input.lastSyncedAt !== null || + newestReceived !== null || + input.newestMeasuredAt !== null; + // Same rule the config read uses (`newestAppleHealthDataAt`): anything that + // arrived counts as liveness for the push cadence. + const syncHealth = resolveSyncVerdict({ + configured: connected, + connected, + lastDataAt: newestOf( + input.lastSyncedAt, + newestReceived, + input.newestMeasuredAt, + ), + legacyLastSyncedAt: input.lastSyncedAt, + cadence: PUSH_CADENCE, + now: input.now, + }); + const recentCalls: RecentCall[] = input.arrivals + .slice(0, RECENT_CALLS_LIMIT) + .map((a) => ({ + endpoint: `batch:${a.type}`, + status: "ok", + at: a.lastReceivedAt.toISOString(), + errorCode: null, + })); + const common = { + syncHealth, + connected, + ledgerState: null, + lastSuccessAt: iso(input.lastSyncedAt ?? newestReceived), + lastAttemptAt: iso(newestReceived), + failingSince: null, + failingLegs: [], + recentCalls, + lastDataAt: iso(lastDataAt), + silenceWindowMs: SOURCE_PROFILES["apple-health"].silenceWindowMs, + now: input.now.getTime(), + }; + return { + key: "apple-health", + connected, + syncHealth, + lastDataAt, + ledgerAlerted: false, + healthWithoutSilence: resolveConnectionHealth({ + ...common, + silentSince: null, + }), + health: resolveConnectionHealth({ + ...common, + silentSince: iso(input.silentSince), + }), + silentSince: input.silentSince, + }; +} + +/** The published health per connection key, for routes. */ +export async function loadConnectionHealth( + userId: string, + keys: readonly ConnectionKey[] = CONNECTION_KEYS, +): Promise { + const facts = await loadConnectionFacts(userId, keys); + const out = {} as ConnectionHealthMap; + for (const key of CONNECTION_KEYS) out[key] = facts.get(key)?.health ?? null; + return out; +} + +/** One connection's published health, for a single-provider route. */ +export async function loadOneConnectionHealth( + userId: string, + key: ConnectionKey, +): Promise { + const facts = await loadConnectionFacts(userId, [key]); + return facts.get(key)?.health ?? null; +} diff --git a/src/lib/integrations/connection-health.ts b/src/lib/integrations/connection-health.ts new file mode 100644 index 000000000..fb5cd53e5 --- /dev/null +++ b/src/lib/integrations/connection-health.ts @@ -0,0 +1,390 @@ +/** + * One health state per connection, resolved on the server. + * + * Every integration used to express its condition in its own vocabulary: the + * ledger `state` (`error_transient`, `error_reauth`, `parked`), the liveness + * `syncHealth.verdict`, Withings' `planUnavailable` and `hasActivityScope`, + * Oura's `scopeReconnect`, Google Health's `needsReauth`, a token expiry, a + * backfill flag. Each card combined those itself, so the same fact could read + * as a red error on one card and as a footnote on the next. + * + * This module folds all of it into four states a person can act on: + * + * ok data is arriving (or the first sync is still due). + * limited it works, but some data types are unavailable because of + * the provider plan or a missing permission; `unavailable` + * names them. + * degraded temporary provider trouble or unexpected silence; the + * server keeps retrying. `since` dates it. + * action_required the person has to do something; `action` names the one + * thing. + * + * Plus a short reason code, the last successful sync, the newest data the + * source delivered, and a redacted diagnostics object that is safe to paste + * into a bug report: endpoint (leg) names, outcomes, instants and short error + * codes. Never a message, a token, a value or a user id. + * + * Pure: no Prisma, no I/O, type imports only, so the resolver is unit-testable + * and the types are importable from client code. The loader that gathers the + * inputs lives in `./connection-health-load`. + */ +import type { IntegrationKey } from "./status"; +import type { SyncHealth } from "./sync-verdict"; + +/** Every connection the health read covers: the polled providers + Apple Health. */ +export type ConnectionKey = IntegrationKey | "apple-health"; + +export const CONNECTION_KEYS: readonly ConnectionKey[] = [ + "withings", + "whoop", + "fitbit", + "google-health", + "polar", + "oura", + "strava", + "nightscout", + "apple-health", +]; + +export const CONNECTION_HEALTH_STATES = [ + "ok", + "limited", + "degraded", + "action_required", +] as const; +export type ConnectionHealthState = (typeof CONNECTION_HEALTH_STATES)[number]; + +/** + * Why the state is what it is. One code per state branch, so a client can pick + * its sentence from the code alone. + */ +export const CONNECTION_HEALTH_REASONS = [ + // ok + "data_arriving", + "first_sync_pending", + // limited + "plan_limited", + "permission_limited", + // degraded + "provider_errors", + "sync_stalled", + "no_new_data", + "unexpected_silence", + // action_required + "reauth_required", + "paused_after_errors", +] as const; +export type ConnectionHealthReason = (typeof CONNECTION_HEALTH_REASONS)[number]; + +/** The one thing a person can do about an `action_required` (or `limited`) connection. */ +export const CONNECTION_HEALTH_ACTIONS = ["reconnect", "resume"] as const; +export type ConnectionHealthAction = (typeof CONNECTION_HEALTH_ACTIONS)[number]; + +export interface UnavailableItem { + /** A provider data type key (`spo2`, `vo2Max`, `activity`, …). */ + item: string; + /** Withheld by the provider's plan, or not covered by the granted permission. */ + cause: "plan" | "permission"; +} + +/** Outcome of one recorded sync attempt. */ +export const RECENT_CALL_STATUSES = [ + "ok", + "transient", + "persistent", + "reauth_required", +] as const; +export type RecentCallStatus = (typeof RECENT_CALL_STATUSES)[number]; + +export interface RecentCall { + /** The sync leg name the code recorded (`sleep`, `measures`, `sync`), or for Apple Health `batch:`. */ + endpoint: string; + status: RecentCallStatus; + at: string; + /** A short error code (`401`, `invalid_grant`, `scope_missing`) or null. */ + errorCode: string | null; +} + +/** How many attempts the ring keeps and the diagnostics publish. */ +export const RECENT_CALLS_LIMIT = 10; + +export interface ConnectionDiagnostics { + /** The ledger state string as stored, for whoever reads a bug report. */ + ledgerState: string | null; + /** Sync legs currently recorded as failing. */ + failingLegs: string[]; + /** Start of the current failure streak, if any. */ + failingSince: string | null; + lastAttemptAt: string | null; + /** The liveness verdict the state was derived from. */ + verdict: SyncHealth["verdict"]; + /** Newest first, at most `RECENT_CALLS_LIMIT`. */ + recentCalls: RecentCall[]; +} + +export interface ConnectionHealth { + state: ConnectionHealthState; + reason: ConnectionHealthReason; + /** When the current non-ok condition started; null for `ok` and `limited`. */ + since: string | null; + /** The one action that resolves it; null when none is needed or possible. */ + action: ConnectionHealthAction | null; + /** Data types this connection cannot deliver; empty when nothing is withheld. */ + unavailable: UnavailableItem[]; + lastSuccessAt: string | null; + /** + * The newest data this source delivered. For polled providers, when the + * newest reading or workout was taken; for Apple Health, when a sync last + * brought a new sample. Null when nothing has arrived. + */ + lastDataAt: string | null; + diagnostics: ConnectionDiagnostics; +} + +export interface ConnectionHealthInput { + /** The liveness verdict from `resolveSyncVerdict`. */ + syncHealth: SyncHealth; + /** Not connected: there is no health to resolve. */ + connected: boolean; + ledgerState: string | null; + lastSuccessAt: string | null; + lastAttemptAt: string | null; + failingSince: string | null; + failingLegs: readonly string[]; + recentCalls: readonly RecentCall[]; + lastDataAt: string | null; + /** Provider-side reauth signal outside the ledger (Google Health `needsReauth`). */ + needsReauth?: boolean; + /** Types withheld by the provider plan (Withings). */ + planUnavailable?: readonly string[]; + /** Types the granted permission does not cover (Oura scopes, Withings activity scope). */ + permissionUnavailable?: readonly string[]; + /** + * The watchdog's unexpected-silence flag: the newest-data instant at the + * moment it flagged, or null when not flagged. + */ + silentSince?: string | null; + /** + * How long the source may go without data before the silence flag counts. + * A flag is ignored once `lastDataAt` is newer than this window, so a source + * that delivered after the last watchdog pass reads healthy at once. + */ + silenceWindowMs?: number; + now?: number; +} + +function toMs(value: string | null | undefined): number | null { + if (!value) return null; + const ms = Date.parse(value); + return Number.isFinite(ms) ? ms : null; +} + +/** + * Fold the inputs into one state. First match wins, from the state that asks + * the most of the person down to the one that asks nothing: + * + * 1. action_required — the grant is gone, or the connection paused itself. + * 2. degraded — failing, stalled, stale, or unexpectedly silent. + * 3. limited — working, with named types withheld. + * 4. ok. + * + * Returns null for a connection that does not exist: there is nothing to say + * about its health, and a state would suggest otherwise. + */ +export function resolveConnectionHealth( + input: ConnectionHealthInput, +): ConnectionHealth | null { + if (!input.connected || input.syncHealth.verdict === "disconnected") { + return null; + } + const now = input.now ?? Date.now(); + const { verdict, since: verdictSince } = input.syncHealth; + + const unavailable: UnavailableItem[] = [ + ...dedupe(input.planUnavailable).map((item): UnavailableItem => ({ + item, + cause: "plan", + })), + ...dedupe(input.permissionUnavailable).map((item): UnavailableItem => ({ + item, + cause: "permission", + })), + ]; + + const diagnostics: ConnectionDiagnostics = { + ledgerState: input.ledgerState, + failingLegs: [...input.failingLegs], + failingSince: input.failingSince, + lastAttemptAt: input.lastAttemptAt, + verdict, + recentCalls: input.recentCalls.slice(0, RECENT_CALLS_LIMIT), + }; + + const base = { + unavailable, + lastSuccessAt: input.lastSuccessAt, + lastDataAt: input.lastDataAt, + diagnostics, + }; + + // 1. The person has to act. + if (verdict === "reauth_required" || input.needsReauth) { + return { + ...base, + state: "action_required", + reason: "reauth_required", + since: verdictSince ?? input.failingSince, + action: "reconnect", + }; + } + if (verdict === "parked") { + return { + ...base, + state: "action_required", + reason: "paused_after_errors", + since: verdictSince ?? input.failingSince, + action: "resume", + }; + } + + // 2. Temporary trouble; the server keeps trying. + if (verdict === "failing") { + return { + ...base, + state: "degraded", + reason: "provider_errors", + since: verdictSince, + action: null, + }; + } + if (verdict === "stalled") { + return { + ...base, + state: "degraded", + reason: "sync_stalled", + since: verdictSince, + action: null, + }; + } + if (verdict === "stale") { + return { + ...base, + state: "degraded", + reason: "no_new_data", + since: verdictSince, + action: null, + }; + } + if (isSilenceStillTrue(input, now)) { + return { + ...base, + state: "degraded", + reason: "unexpected_silence", + since: input.silentSince ?? null, + action: null, + }; + } + + // 3. Working, with something withheld. A permission gap is fixable by one + // reconnect; a plan gap is not fixable here at all. + const permissionGap = unavailable.some((u) => u.cause === "permission"); + if (unavailable.length > 0) { + return { + ...base, + state: "limited", + reason: permissionGap ? "permission_limited" : "plan_limited", + since: null, + action: permissionGap ? "reconnect" : null, + }; + } + + // 4. Fine. + return { + ...base, + state: "ok", + reason: + verdict === "pending_first_sync" ? "first_sync_pending" : "data_arriving", + since: null, + action: null, + }; +} + +/** + * The watchdog flag stands only while no data newer than the silence window + * has arrived. The watchdog clears it on its next pass; this keeps the read + * honest in the hour between. + */ +function isSilenceStillTrue( + input: ConnectionHealthInput, + now: number, +): boolean { + if (!input.silentSince) return false; + const window = input.silenceWindowMs; + const lastData = toMs(input.lastDataAt); + if (window != null && lastData != null && now - lastData <= window) { + return false; + } + return true; +} + +function dedupe(values: readonly string[] | undefined): string[] { + if (!values) return []; + return [...new Set(values)]; +} + +// ── Recent-call ring ───────────────────────────────────────────────── + +/** Leg names and error codes are identifiers; anything else is dropped. */ +const SAFE_TOKEN = /^[A-Za-z0-9_.:-]{1,48}$/; + +/** A leg name safe to publish, or `sync` when the caller passed none or something unexpected. */ +export function safeEndpointName(value: string | null | undefined): string { + return value && SAFE_TOKEN.test(value) ? value : "sync"; +} + +/** + * An error code safe to publish, or null. Codes are short identifiers the sync + * code passes (`401`, `invalid_grant`, `scope_missing`); a value that does not + * look like one (a sentence, a URL, something carrying a token) is dropped + * rather than trimmed, because a trimmed secret is still a secret. + */ +export function safeErrorCode(value: string | null | undefined): string | null { + return value && SAFE_TOKEN.test(value) ? value : null; +} + +/** + * Read the stored JSON ring into typed entries, newest first. Anything that is + * not a well-formed entry is skipped; every string is re-checked against the + * same closed shapes the writer applies, so a hand-edited row cannot publish + * free text through the diagnostics. + */ +export function readRecentCalls(value: unknown): RecentCall[] { + if (!Array.isArray(value)) return []; + const out: RecentCall[] = []; + for (const raw of value) { + if (!raw || typeof raw !== "object") continue; + const entry = raw as Record; + const status = entry.status; + const at = entry.at; + if ( + typeof status !== "string" || + !(RECENT_CALL_STATUSES as readonly string[]).includes(status) || + typeof at !== "string" || + !Number.isFinite(Date.parse(at)) + ) { + continue; + } + out.push({ + endpoint: safeEndpointName( + typeof entry.endpoint === "string" ? entry.endpoint : null, + ), + status: status as RecentCallStatus, + at: new Date(at).toISOString(), + errorCode: safeErrorCode( + typeof entry.errorCode === "string" ? entry.errorCode : null, + ), + }); + if (out.length >= RECENT_CALLS_LIMIT) break; + } + return out; +} diff --git a/src/lib/integrations/status.ts b/src/lib/integrations/status.ts index d00fbdad5..57bac23f8 100644 --- a/src/lib/integrations/status.ts +++ b/src/lib/integrations/status.ts @@ -40,6 +40,12 @@ import { type DispatchOutcome, } from "@/lib/notifications/dispatcher"; import type { IntegrationClassification } from "@/lib/integrations/http-status-classifier"; +import { + RECENT_CALLS_LIMIT, + safeEndpointName, + safeErrorCode, + type RecentCallStatus, +} from "@/lib/integrations/connection-health"; export type IntegrationKey = | "withings" @@ -194,13 +200,14 @@ export function getPersistentFailureThreshold(): number { } /** - * Re-alert window: once we've paged on a streak we hold the alert for - * 24h before paging again on the same streak (idempotency). The streak - * is implicitly "reset" by a single success, which clears every - * per-kind bucket and `alertedAt` — so a flapping integration that - * succeeds once an hour will not page repeatedly. + * One page per failure streak. `alertedAt` is cleared by every path that ends + * a streak (a success entitled to clear the row, a reconnect, a resume, a + * disconnect), so a stamped `alertedAt` means "this streak was already + * announced". It used to re-page every 24 h for as long as the streak ran, + * which turned a week-long outage into seven identical messages; the + * integration watchdog (`./watchdog`) now owns the rest of the incident, + * including the one recovery note when it ends. */ -const ALERT_REPEAT_WINDOW_MS = 24 * 60 * 60 * 1000; /** * v1.4.43 W14 — once a `persistent` failure streak has been running @@ -474,6 +481,12 @@ export async function recordSyncSuccess( ...(ownedLastError ? { lastError: null } : {}), }, }); + await appendRecentCall(userId, integration, { + endpoint: leg, + status: "ok", + at: now, + errorCode: null, + }); // Meta only — the enclosing wide event owns its own action name (the // cron task), and overwriting it here would rename the job. annotate({ @@ -516,6 +529,12 @@ export async function recordSyncSuccess( alertedAt: null, }, }); + await appendRecentCall(userId, integration, { + endpoint: leg, + status: "ok", + at: now, + errorCode: null, + }); } export interface RecordSyncFailureInput { @@ -718,6 +737,13 @@ export async function recordSyncFailure( }, }); + await appendRecentCall(userId, integration, { + endpoint: leg, + status: effectiveKind, + at: now, + errorCode: errorCode ?? null, + }); + // Audit log entry — fire-and-await so an integration test can assert // it. The auth/audit helper is its own DB write so it's safe to call // serially without bloating latency in the success path (which never @@ -771,9 +797,7 @@ export async function recordSyncFailure( const threshold = getPersistentFailureThreshold(); const alertSignal = Math.max(...Object.values(buckets)); if (alertSignal >= threshold) { - const previouslyAlerted = - row.alertedAt && - now.getTime() - row.alertedAt.getTime() < ALERT_REPEAT_WINDOW_MS; + const previouslyAlerted = row.alertedAt != null; if (!previouslyAlerted) { const outcome = await maybeAlertAdmins({ userId, @@ -970,6 +994,13 @@ export async function parkIntegrationAtReauth(opts: { }, }); + await appendRecentCall(userId, integration, { + endpoint: leg, + status: "reauth_required", + at: now, + errorCode, + }); + if (isFreshPark) { await auditLog("integrations.reauth_required", { userId, @@ -1024,6 +1055,13 @@ export async function markReauthRequired( }, }); + await appendRecentCall(userId, integration, { + endpoint: "token_refresh", + status: "reauth_required", + at: now, + errorCode: null, + }); + await auditLog("integrations.reauth_required", { userId, details: { integration, message }, @@ -1090,6 +1128,64 @@ export async function markReconnected( // ── helpers ──────────────────────────────────────────────────────── +interface RecentCallWrite { + endpoint: string | null; + status: RecentCallStatus; + at: Date; + errorCode: string | null; +} + +/** + * Prepend one attempt to the row's `recentCalls` ring and keep the newest + * `RECENT_CALLS_LIMIT`. One statement, no read: the ring is rebuilt inside + * Postgres, so a hot sync path pays one UPDATE rather than a read-modify-write. + * + * The entry is built from closed shapes only. The leg name and error code go + * through the same filters the reader applies (`safeEndpointName`, + * `safeErrorCode`), and the failure message is never part of it, so nothing a + * provider sent can reach the diagnostics a user copies into a bug report. + * + * Fail-soft: the ring is diagnostics. A failed append must never fail or undo + * the ledger write it describes, so the error becomes a wide-event warning. + */ +async function appendRecentCall( + userId: string, + integration: IntegrationKey, + call: RecentCallWrite, +): Promise { + const entry = JSON.stringify({ + endpoint: safeEndpointName(call.endpoint), + status: call.status, + at: call.at.toISOString(), + errorCode: safeErrorCode(call.errorCode), + }); + try { + // Parameter-bound throughout; the ring is rebuilt newest-first and cut at + // the limit. A row whose column is NULL or not an array starts empty. + await prisma.$executeRaw` + UPDATE "integration_statuses" + SET "recent_calls" = ( + SELECT COALESCE(jsonb_agg(e ORDER BY ord), '[]'::jsonb) + FROM ( + SELECT e, ord + FROM jsonb_array_elements( + jsonb_build_array(${entry}::jsonb) || + CASE WHEN jsonb_typeof("recent_calls") = 'array' + THEN "recent_calls" ELSE '[]'::jsonb END + ) WITH ORDINALITY AS t(e, ord) + ORDER BY ord + LIMIT ${RECENT_CALLS_LIMIT} + ) newest + ) + WHERE "user_id" = ${userId} AND "integration" = ${integration} + `; + } catch (err) { + getEvent()?.addWarning( + `IntegrationStatus recent-call append failed: ${err instanceof Error ? err.message : String(err)}`, + ); + } +} + /** * Encrypt with a fallback that swallows crypto-config errors so a * misconfigured ENCRYPTION_KEY can never break the audit/error path. @@ -1247,7 +1343,36 @@ async function maybeAlertAdmins(input: AlertInput): Promise { where: { id: input.userId }, select: { email: true }, }); + const payload = formatAdminAlertPayload({ + ...input, + subjectLabel: subject?.email ?? input.userId, + }); + return dispatchAdminSystemAlert({ + subjectUserId: input.userId, + title: payload.title, + message: payload.message, + metadata: payload.metadata, + }); +} +/** + * Send one `SYSTEM_ALERT` to every admin through the existing dispatcher, the + * operator path the sync ledger and the integration watchdog share. Each admin + * receives it on whatever channels they enabled (Telegram, ntfy, Web Push, + * APNs); nothing here adds a channel. + * + * The returned outcome aggregates the dispatcher's real per-admin outcomes, so + * one attempted delivery cannot be hidden by another admin having no enabled + * channel. A caller stamps its own "already announced" marker only when + * `channelsAttempted > 0`. + */ +export async function dispatchAdminSystemAlert(input: { + /** The account the alert is about; picks the tap target per admin. */ + subjectUserId: string; + title: string; + message: string; + metadata: Record; +}): Promise { const admins = await prisma.user.findMany({ where: { role: "ADMIN" }, select: { id: true }, @@ -1261,10 +1386,7 @@ async function maybeAlertAdmins(input: AlertInput): Promise { }; } - const payload = formatAdminAlertPayload({ - ...input, - subjectLabel: subject?.email ?? input.userId, - }); + const payload = input; const outcome: DispatchOutcome = { dispatched: false, channelsAttempted: 0, @@ -1282,7 +1404,7 @@ async function maybeAlertAdmins(input: AlertInput): Promise { // Tap target: an admin whose own sync fails fixes it in their own // integration settings; anyone else's shows on the admin page. url: - admin.id === input.userId + admin.id === input.subjectUserId ? "/settings/integrations" : "/admin/integrations", }, diff --git a/src/lib/integrations/watchdog-classify.ts b/src/lib/integrations/watchdog-classify.ts new file mode 100644 index 000000000..cc1951e36 --- /dev/null +++ b/src/lib/integrations/watchdog-classify.ts @@ -0,0 +1,213 @@ +/** + * The integration watchdog's judgement, kept pure. + * + * Two questions per connection, answered from data the runner gathers: + * + * 1. Is a source that normally delivers unexpectedly silent? + * Only a CONTINUOUS source can be: a wearable, a CGM, a phone that + * records steps all day. Its own history decides "normally": it must have + * delivered on at least `NORMAL_DELIVERY_MIN_DAYS` of the previous + * `HISTORY_DAYS` days. An ON-DEMAND source (a scale, a blood-pressure + * cuff, a workout log) is silent whenever its owner does not measure, + * which says nothing about the pipe, so silence never counts for it; it + * is watched for errors only. + * + * 2. Is there an incident the operator should hear about? + * Unexpected silence; or a health state of `action_required` / + * `degraded` that has lasted longer than `INCIDENT_AFTER_MS`; or a + * failure streak the sync ledger already paged about (adopted, so the + * operator still gets the one recovery note when it ends). + * + * One incident per connection at a time. It is announced once and closed once; + * a change of reason while it is open (silence turning into a reauth) is the + * same incident and sends nothing. + */ +import type { + ConnectionHealth, + ConnectionHealthReason, + ConnectionKey, +} from "./connection-health"; + +export type SourceCadenceKind = "continuous" | "on_demand"; + +export interface SourceProfile { + kind: SourceCadenceKind; + /** How long a continuous source may go without data before silence counts. */ + silenceWindowMs: number; +} + +const HOUR_MS = 60 * 60 * 1000; +const DAY_MS = 24 * HOUR_MS; + +/** How many days of a connection's own history decide "normally delivers". */ +export const HISTORY_DAYS = 14; + +/** Days with data, out of `HISTORY_DAYS`, that make a source "normally delivering". */ +export const NORMAL_DELIVERY_MIN_DAYS = 10; + +/** How long a non-ok health state must last before it is an operator incident. */ +export const INCIDENT_AFTER_MS = DAY_MS; + +/** + * Per connection. Withings is mostly scales and cuffs, Strava is workouts: + * both are on-demand. Everything worn or attached around the clock is + * continuous. + * + * Apple Health gets 48 h instead of 24 h. iOS defers background delivery when + * the phone is low on storage, in low-power mode or simply not unlocked, and a + * day of that is ordinary; the history gate already limits the check to + * phones that deliver almost daily, and the longer window keeps an ordinary + * quiet day from paging anyone. + */ +export const SOURCE_PROFILES: Record = { + withings: { kind: "on_demand", silenceWindowMs: DAY_MS }, + strava: { kind: "on_demand", silenceWindowMs: DAY_MS }, + whoop: { kind: "continuous", silenceWindowMs: DAY_MS }, + oura: { kind: "continuous", silenceWindowMs: DAY_MS }, + fitbit: { kind: "continuous", silenceWindowMs: DAY_MS }, + "google-health": { kind: "continuous", silenceWindowMs: DAY_MS }, + polar: { kind: "continuous", silenceWindowMs: DAY_MS }, + nightscout: { kind: "continuous", silenceWindowMs: DAY_MS }, + "apple-health": { kind: "continuous", silenceWindowMs: 2 * DAY_MS }, +}; + +/** Incident reason codes. The silence one is the watchdog's own. */ +export type IncidentReason = ConnectionHealthReason; + +export interface WatchRow { + silentSince: Date | null; + incidentReason: string | null; + incidentOpenedAt: Date | null; + alertedAt: Date | null; +} + +export interface SilenceInput { + profile: SourceProfile; + /** Newest data instant, or null when the source never delivered. */ + lastDataAt: Date | null; + /** + * Distinct days with data in the `HISTORY_DAYS` before the silence window, + * or null when the runner has not computed it (it only does when needed). + */ + deliveredDays: number | null; + /** The current flag, so an already-flagged source is not re-judged hourly. */ + currentSilentSince: Date | null; + now: Date; +} + +export type SilenceVerdict = + /** Data is inside the window (or the source is on-demand): no flag. */ + | { kind: "clear" } + /** Silent, and the history has to be read before deciding. */ + | { kind: "needs_history" } + /** Silent, and the history says it should not be. */ + | { kind: "flag"; silentSince: Date } + /** Silent, but it does not normally deliver: the owner is not using it. */ + | { kind: "not_expected" }; + +/** + * Decide the silence flag. `deliveredDays` is only needed for a silent, + * not-yet-flagged continuous source (the hourly pass reads it lazily) and for a + * re-check of a flag (the daily pass passes it so a source its owner has + * stopped using drops out once its history no longer says it delivers). + */ +export function classifySilence(input: SilenceInput): SilenceVerdict { + const { profile, lastDataAt, now } = input; + if (profile.kind === "on_demand") return { kind: "clear" }; + // A source that never delivered has no history to break with. + if (lastDataAt === null) return { kind: "clear" }; + if (now.getTime() - lastDataAt.getTime() <= profile.silenceWindowMs) { + return { kind: "clear" }; + } + if (input.deliveredDays === null) { + // Already flagged and nothing asked for a re-check: keep the flag as is. + if (input.currentSilentSince) { + return { kind: "flag", silentSince: input.currentSilentSince }; + } + return { kind: "needs_history" }; + } + if (input.deliveredDays >= NORMAL_DELIVERY_MIN_DAYS) { + return { + kind: "flag", + silentSince: input.currentSilentSince ?? lastDataAt, + }; + } + return { kind: "not_expected" }; +} + +/** + * The incident condition right now, or null when there is none. + * + * `health` is the resolved connection health WITHOUT the watchdog's own + * silence flag folded in, so the silence condition and the state condition + * stay separate. `ledgerAlerted` is true while the sync ledger holds an + * unresolved page for the current streak. + */ +export function currentIncident(input: { + health: ConnectionHealth | null; + silentSince: Date | null; + ledgerAlerted: boolean; + now: Date; +}): { reason: IncidentReason; since: Date } | null { + const { health, silentSince, now } = input; + if (health === null) return null; + const troubled = + health.state === "action_required" || health.state === "degraded"; + if (troubled) { + // A provider-side reauth flag (Google Health) can stand without a ledger + // streak to date it; the last success is then when it last worked. + const sinceIso = health.since ?? health.lastSuccessAt; + const since = sinceIso ? new Date(sinceIso) : null; + const lasted = + since !== null && now.getTime() - since.getTime() > INCIDENT_AFTER_MS; + if (lasted || input.ledgerAlerted) { + return { reason: health.reason, since: since ?? now }; + } + } + if (silentSince) { + return { reason: "unexpected_silence", since: silentSince }; + } + return null; +} + +export type WatchTransition = + /** No incident before, none now. */ + | { kind: "none" } + /** A new incident: announce it. */ + | { kind: "open"; reason: IncidentReason; since: Date } + /** A new incident the ledger already announced: record it, send nothing. */ + | { kind: "adopt"; reason: IncidentReason; since: Date } + /** Still open: send nothing. `announce` when the open incident was never announced (no channel last time). */ + | { kind: "keep"; reason: IncidentReason; announce: boolean } + /** Over. `notify` when the operator had been told it started. */ + | { kind: "close"; notify: boolean; previousReason: string }; + +/** + * Compare the open incident (if any) with the current condition. Pure; the + * runner performs the side effects. + */ +export function decideWatchTransition(input: { + row: WatchRow | null; + incident: { reason: IncidentReason; since: Date } | null; + ledgerAlerted: boolean; +}): WatchTransition { + const open = input.row?.incidentReason ?? null; + if (open === null) { + if (input.incident === null) return { kind: "none" }; + return input.ledgerAlerted + ? { kind: "adopt", ...input.incident } + : { kind: "open", ...input.incident }; + } + if (input.incident === null) { + return { + kind: "close", + notify: input.row?.alertedAt != null, + previousReason: open, + }; + } + return { + kind: "keep", + reason: input.incident.reason, + announce: input.row?.alertedAt == null && !input.ledgerAlerted, + }; +}