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 1/4] 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, + }; +} From c094f217fc4542d56ae83397bf9af08460de1231 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 15 Oct 2026 11:07:57 +0200 Subject: [PATCH 2/4] feat(integrations): watch connections for silence and lasting trouble An hourly and a daily pass flag a continuous source that delivered on most of the previous 14 days and has gone quiet, and open one operator incident per connection when a connection stays degraded or needs action for more than a day. The incident is announced once through the existing admin alert path, adopted rather than repeated when the sync ledger already paged, and closed with one recovery note. Scales, cuffs and workout logs are watched for errors only, never for silence. --- src/lib/data-wipe/wipe-plan.ts | 1 + src/lib/export/backup-plan.ts | 2 + .../__tests__/connection-health.test.ts | 209 +++++++++ .../__tests__/watchdog-classify.test.ts | 320 +++++++++++++ src/lib/integrations/watchdog.ts | 430 ++++++++++++++++++ src/lib/jobs/integration-watchdog.ts | 71 +++ src/lib/jobs/job-outcome.ts | 5 + src/lib/jobs/queue-runtime.ts | 6 + .../reminder/register-integration-sync.ts | 40 ++ src/lib/jobs/subsystem-surface.ts | 2 + .../integration/integration-watchdog.test.ts | 280 ++++++++++++ 11 files changed, 1366 insertions(+) create mode 100644 src/lib/integrations/__tests__/connection-health.test.ts create mode 100644 src/lib/integrations/__tests__/watchdog-classify.test.ts create mode 100644 src/lib/integrations/watchdog.ts create mode 100644 src/lib/jobs/integration-watchdog.ts create mode 100644 tests/integration/integration-watchdog.test.ts diff --git a/src/lib/data-wipe/wipe-plan.ts b/src/lib/data-wipe/wipe-plan.ts index da0dbb4af..a247fd823 100644 --- a/src/lib/data-wipe/wipe-plan.ts +++ b/src/lib/data-wipe/wipe-plan.ts @@ -231,6 +231,7 @@ export const WIPE_MODELS = [ "GoogleHealthOAuthState", "McpOAuthConnection", "IntegrationStatus", + "IntegrationWatch", // ── Notification transports ───────────────────────────────────────────── "NotificationChannel", diff --git a/src/lib/export/backup-plan.ts b/src/lib/export/backup-plan.ts index 340e4b68e..94d737678 100644 --- a/src/lib/export/backup-plan.ts +++ b/src/lib/export/backup-plan.ts @@ -670,6 +670,8 @@ export const NOT_IN_BACKUP_MODELS: Readonly> = { "A person's API token for their Paperless-ngx or Papra instance. A backup file is not a credential store, same reasoning as ApiToken; and restored on another host it would point at an address that host's operator never allowed in DOCUMENT_SOURCE_ORIGINS. The documents imported through it are carried with the vault; reconnecting takes one form.", IntegrationStatus: "A ledger of what this deployment's syncs did. Restoring it onto another host would assert a sync history that host never had.", + IntegrationWatch: + "What this host's integration watchdog observed and which operator incidents it announced. Restored elsewhere it would assert silences and alerts that host never saw.", // Credentials and device identity. ApiToken: diff --git a/src/lib/integrations/__tests__/connection-health.test.ts b/src/lib/integrations/__tests__/connection-health.test.ts new file mode 100644 index 000000000..5c0db5353 --- /dev/null +++ b/src/lib/integrations/__tests__/connection-health.test.ts @@ -0,0 +1,209 @@ +import { describe, expect, it } from "vitest"; +import { + readRecentCalls, + resolveConnectionHealth, + safeErrorCode, + type ConnectionHealthInput, +} from "../connection-health"; +import type { SyncHealth } from "../sync-verdict"; + +const NOW = Date.parse("2026-10-15T12:00:00.000Z"); +const HOUR = 60 * 60 * 1000; + +function input( + verdict: SyncHealth["verdict"], + extra: Partial = {}, +): ConnectionHealthInput { + return { + syncHealth: { + verdict, + since: verdict === "fresh" ? null : "2026-10-14T08:00:00.000Z", + }, + connected: true, + ledgerState: "connected", + lastSuccessAt: "2026-10-15T11:00:00.000Z", + lastAttemptAt: "2026-10-15T11:00:00.000Z", + failingSince: null, + failingLegs: [], + recentCalls: [], + lastDataAt: "2026-10-15T10:00:00.000Z", + silenceWindowMs: 24 * HOUR, + now: NOW, + ...extra, + }; +} + +describe("resolveConnectionHealth", () => { + it("returns null for a connection that does not exist", () => { + expect( + resolveConnectionHealth(input("fresh", { connected: false })), + ).toBeNull(); + expect(resolveConnectionHealth(input("disconnected"))).toBeNull(); + }); + + it("maps a fresh verdict to ok / data_arriving", () => { + const h = resolveConnectionHealth(input("fresh"))!; + expect(h.state).toBe("ok"); + expect(h.reason).toBe("data_arriving"); + expect(h.action).toBeNull(); + expect(h.since).toBeNull(); + }); + + it("maps a pending first sync to ok / first_sync_pending, not to a problem", () => { + const h = resolveConnectionHealth(input("pending_first_sync"))!; + expect(h.state).toBe("ok"); + expect(h.reason).toBe("first_sync_pending"); + }); + + it.each([ + ["reauth_required", "reauth_required", "reconnect"], + ["parked", "paused_after_errors", "resume"], + ] as const)( + "maps %s to action_required with exactly one action", + (verdict, reason, action) => { + const h = resolveConnectionHealth(input(verdict))!; + expect(h.state).toBe("action_required"); + expect(h.reason).toBe(reason); + expect(h.action).toBe(action); + expect(h.since).toBe("2026-10-14T08:00:00.000Z"); + }, + ); + + it("treats a provider-side reauth flag as action_required even on a fresh ledger", () => { + const h = resolveConnectionHealth(input("fresh", { needsReauth: true }))!; + expect(h.state).toBe("action_required"); + expect(h.action).toBe("reconnect"); + }); + + it.each([ + ["failing", "provider_errors"], + ["stalled", "sync_stalled"], + ["stale", "no_new_data"], + ] as const)("maps %s to degraded / %s, no action", (verdict, reason) => { + const h = resolveConnectionHealth(input(verdict))!; + expect(h.state).toBe("degraded"); + expect(h.reason).toBe(reason); + expect(h.action).toBeNull(); + expect(h.since).not.toBeNull(); + }); + + it("lists plan-withheld types as limited without an action", () => { + const h = resolveConnectionHealth( + input("fresh", { planUnavailable: ["spo2", "ecg", "spo2"] }), + )!; + expect(h.state).toBe("limited"); + expect(h.reason).toBe("plan_limited"); + expect(h.action).toBeNull(); + expect(h.unavailable).toEqual([ + { item: "spo2", cause: "plan" }, + { item: "ecg", cause: "plan" }, + ]); + }); + + it("offers one reconnect when a permission gap is part of the limit", () => { + const h = resolveConnectionHealth( + input("fresh", { + planUnavailable: ["ecg"], + permissionUnavailable: ["spo2"], + }), + )!; + expect(h.state).toBe("limited"); + expect(h.reason).toBe("permission_limited"); + expect(h.action).toBe("reconnect"); + }); + + it("lets a worse state win over limited, keeping the unavailable list", () => { + const h = resolveConnectionHealth( + input("failing", { planUnavailable: ["ecg"] }), + )!; + expect(h.state).toBe("degraded"); + expect(h.unavailable).toEqual([{ item: "ecg", cause: "plan" }]); + }); + + it("surfaces the watchdog silence flag as degraded / unexpected_silence", () => { + const h = resolveConnectionHealth( + input("fresh", { + lastDataAt: "2026-10-13T12:00:00.000Z", + silentSince: "2026-10-13T12:00:00.000Z", + }), + )!; + expect(h.state).toBe("degraded"); + expect(h.reason).toBe("unexpected_silence"); + expect(h.since).toBe("2026-10-13T12:00:00.000Z"); + }); + + it("drops a stale silence flag as soon as fresh data has arrived", () => { + const h = resolveConnectionHealth( + input("fresh", { + lastDataAt: "2026-10-15T11:30:00.000Z", + silentSince: "2026-10-13T12:00:00.000Z", + }), + )!; + expect(h.state).toBe("ok"); + }); + + it("lets a reauth outrank the silence flag", () => { + const h = resolveConnectionHealth( + input("reauth_required", { + lastDataAt: "2026-10-13T12:00:00.000Z", + silentSince: "2026-10-13T12:00:00.000Z", + }), + )!; + expect(h.state).toBe("action_required"); + }); +}); + +describe("recent-call diagnostics", () => { + it("keeps only closed-shape codes", () => { + expect(safeErrorCode("401")).toBe("401"); + expect(safeErrorCode("invalid_grant")).toBe("invalid_grant"); + expect(safeErrorCode("Bearer abc def")).toBeNull(); + expect(safeErrorCode("https://host/api?token=x")).toBeNull(); + expect(safeErrorCode(null)).toBeNull(); + }); + + it("reads well-formed entries and drops anything carrying free text", () => { + const calls = readRecentCalls([ + { + endpoint: "sleep", + status: "transient", + at: "2026-10-15T10:00:00Z", + errorCode: "503", + message: "upstream said something with a token", + }, + { + endpoint: "https://evil/?t=1", + status: "ok", + at: "2026-10-15T09:00:00Z", + errorCode: "token=secret value", + }, + { endpoint: "sleep", status: "exploded", at: "2026-10-15T08:00:00Z" }, + "not an object", + ]); + expect(calls).toEqual([ + { + endpoint: "sleep", + status: "transient", + at: "2026-10-15T10:00:00.000Z", + errorCode: "503", + }, + { + endpoint: "sync", + status: "ok", + at: "2026-10-15T09:00:00.000Z", + errorCode: null, + }, + ]); + expect(JSON.stringify(calls)).not.toContain("token"); + }); + + it("caps the published list at ten", () => { + const many = Array.from({ length: 15 }, (_, i) => ({ + endpoint: "sync", + status: "ok", + at: new Date(NOW - i * HOUR).toISOString(), + errorCode: null, + })); + expect(readRecentCalls(many)).toHaveLength(10); + }); +}); diff --git a/src/lib/integrations/__tests__/watchdog-classify.test.ts b/src/lib/integrations/__tests__/watchdog-classify.test.ts new file mode 100644 index 000000000..0e2992e46 --- /dev/null +++ b/src/lib/integrations/__tests__/watchdog-classify.test.ts @@ -0,0 +1,320 @@ +import { describe, expect, it } from "vitest"; +import type { ConnectionHealth } from "../connection-health"; +import { + NORMAL_DELIVERY_MIN_DAYS, + SOURCE_PROFILES, + classifySilence, + currentIncident, + decideWatchTransition, + type WatchRow, +} from "../watchdog-classify"; +import { formatWatchdogMessage } from "../watchdog"; + +const NOW = new Date("2026-10-15T12:00:00.000Z"); +const HOUR = 60 * 60 * 1000; +const ago = (hours: number) => new Date(NOW.getTime() - hours * HOUR); + +function health( + state: ConnectionHealth["state"], + reason: ConnectionHealth["reason"], + since: Date | null, +): ConnectionHealth { + return { + state, + reason, + since: since?.toISOString() ?? null, + action: null, + unavailable: [], + lastSuccessAt: null, + lastDataAt: null, + diagnostics: { + ledgerState: null, + failingLegs: [], + failingSince: null, + lastAttemptAt: null, + verdict: "fresh", + recentCalls: [], + }, + }; +} + +const openRow = (overrides: Partial = {}): WatchRow => ({ + silentSince: null, + incidentReason: "unexpected_silence", + incidentOpenedAt: ago(5), + alertedAt: ago(5), + ...overrides, +}); + +describe("classifySilence", () => { + const oura = SOURCE_PROFILES.oura; + const withings = SOURCE_PROFILES.withings; + + it("never flags an on-demand source for silence, however long and however regular", () => { + expect(withings.kind).toBe("on_demand"); + expect(SOURCE_PROFILES.strava.kind).toBe("on_demand"); + expect( + classifySilence({ + profile: withings, + lastDataAt: ago(24 * 30), + deliveredDays: 14, + currentSilentSince: null, + now: NOW, + }), + ).toEqual({ kind: "clear" }); + }); + + it("is clear while a continuous source delivers inside its window", () => { + expect( + classifySilence({ + profile: oura, + lastDataAt: ago(20), + deliveredDays: null, + currentSilentSince: null, + now: NOW, + }), + ).toEqual({ kind: "clear" }); + }); + + it("asks for the history before flagging (the cheap hourly pass reads it lazily)", () => { + expect( + classifySilence({ + profile: oura, + lastDataAt: ago(30), + deliveredDays: null, + currentSilentSince: null, + now: NOW, + }), + ).toEqual({ kind: "needs_history" }); + }); + + it("flags a silent source that delivered on most of the previous 14 days", () => { + expect( + classifySilence({ + profile: oura, + lastDataAt: ago(30), + deliveredDays: NORMAL_DELIVERY_MIN_DAYS, + currentSilentSince: null, + now: NOW, + }), + ).toEqual({ kind: "flag", silentSince: ago(30) }); + }); + + it("does not flag a source its owner wears only now and then", () => { + expect( + classifySilence({ + profile: oura, + lastDataAt: ago(30), + deliveredDays: NORMAL_DELIVERY_MIN_DAYS - 1, + currentSilentSince: null, + now: NOW, + }), + ).toEqual({ kind: "not_expected" }); + }); + + it("keeps a standing flag on the hourly pass without re-reading history", () => { + expect( + classifySilence({ + profile: oura, + lastDataAt: ago(50), + deliveredDays: null, + currentSilentSince: ago(50), + now: NOW, + }), + ).toEqual({ kind: "flag", silentSince: ago(50) }); + }); + + it("drops a standing flag on the daily re-check once the history no longer says it delivers", () => { + expect( + classifySilence({ + profile: oura, + lastDataAt: ago(24 * 6), + deliveredDays: 8, + currentSilentSince: ago(24 * 6), + now: NOW, + }), + ).toEqual({ kind: "not_expected" }); + }); + + it("gives Apple Health a wider window than a wearable", () => { + const apple = SOURCE_PROFILES["apple-health"]; + expect( + classifySilence({ + profile: apple, + lastDataAt: ago(30), + deliveredDays: 14, + currentSilentSince: null, + now: NOW, + }), + ).toEqual({ kind: "clear" }); + }); +}); + +describe("currentIncident", () => { + it("opens nothing for a degraded state younger than a day", () => { + expect( + currentIncident({ + health: health("degraded", "provider_errors", ago(5)), + silentSince: null, + ledgerAlerted: false, + now: NOW, + }), + ).toBeNull(); + }); + + it("opens an incident once degraded or action_required has lasted more than a day", () => { + expect( + currentIncident({ + health: health("action_required", "reauth_required", ago(25)), + silentSince: null, + ledgerAlerted: false, + now: NOW, + }), + ).toEqual({ reason: "reauth_required", since: ago(25) }); + }); + + it("follows the ledger's own page at once, so its recovery is reported too", () => { + expect( + currentIncident({ + health: health("degraded", "provider_errors", ago(3)), + silentSince: null, + ledgerAlerted: true, + now: NOW, + }), + ).toEqual({ reason: "provider_errors", since: ago(3) }); + }); + + it("opens a silence incident from the flag", () => { + expect( + currentIncident({ + health: health("ok", "data_arriving", null), + silentSince: ago(30), + ledgerAlerted: false, + now: NOW, + }), + ).toEqual({ reason: "unexpected_silence", since: ago(30) }); + }); + + it("has no incident for a limited connection", () => { + expect( + currentIncident({ + health: health("limited", "plan_limited", null), + silentSince: null, + ledgerAlerted: false, + now: NOW, + }), + ).toBeNull(); + }); +}); + +describe("decideWatchTransition", () => { + const incident = { reason: "unexpected_silence" as const, since: ago(30) }; + + it("opens and announces a new incident", () => { + expect( + decideWatchTransition({ row: null, incident, ledgerAlerted: false }), + ).toEqual({ kind: "open", ...incident }); + }); + + it("adopts an incident the ledger already paged about instead of paging again", () => { + expect( + decideWatchTransition({ row: null, incident, ledgerAlerted: true }), + ).toEqual({ kind: "adopt", ...incident }); + }); + + it("sends nothing while an announced incident stays open (no repeats)", () => { + expect( + decideWatchTransition({ + row: openRow(), + incident, + ledgerAlerted: false, + }), + ).toEqual({ kind: "keep", reason: "unexpected_silence", announce: false }); + }); + + it("treats a changed reason as the same incident", () => { + expect( + decideWatchTransition({ + row: openRow(), + incident: { reason: "reauth_required", since: ago(2) }, + ledgerAlerted: false, + }), + ).toEqual({ kind: "keep", reason: "reauth_required", announce: false }); + }); + + it("retries the announcement of an incident no channel accepted", () => { + expect( + decideWatchTransition({ + row: openRow({ alertedAt: null }), + incident, + ledgerAlerted: false, + }), + ).toEqual({ kind: "keep", reason: "unexpected_silence", announce: true }); + }); + + it("closes with one recovery note when the announced incident ends", () => { + expect( + decideWatchTransition({ + row: openRow(), + incident: null, + ledgerAlerted: false, + }), + ).toEqual({ + kind: "close", + notify: true, + previousReason: "unexpected_silence", + }); + }); + + it("closes quietly when the operator was never told it started", () => { + expect( + decideWatchTransition({ + row: openRow({ alertedAt: null }), + incident: null, + ledgerAlerted: false, + }), + ).toEqual({ + kind: "close", + notify: false, + previousReason: "unexpected_silence", + }); + }); + + it("does nothing when there was and is no incident", () => { + expect( + decideWatchTransition({ + row: null, + incident: null, + ledgerAlerted: false, + }), + ).toEqual({ kind: "none" }); + }); +}); + +describe("formatWatchdogMessage", () => { + it("names the provider, the account and the instant, in plain text", () => { + const m = formatWatchdogMessage({ + kind: "incident", + key: "oura", + subjectLabel: "someone@example.test", + reason: "unexpected_silence", + since: ago(30), + }); + expect(m.title).toBe("Oura needs attention for someone@example.test"); + expect(m.message).toContain("2026-10-14 06:00 UTC"); + expect(m.message).not.toMatch(/[—–·→]/); + }); + + it("says why a recovered incident ended", () => { + const m = formatWatchdogMessage({ + kind: "recovery", + key: "apple-health", + subjectLabel: "someone@example.test", + reason: "unexpected_silence", + since: null, + closedBecause: "disconnected", + }); + expect(m.title).toBe("Apple Health recovered for someone@example.test"); + expect(m.message).toContain("removed"); + }); +}); diff --git a/src/lib/integrations/watchdog.ts b/src/lib/integrations/watchdog.ts new file mode 100644 index 000000000..5fbe4f211 --- /dev/null +++ b/src/lib/integrations/watchdog.ts @@ -0,0 +1,430 @@ +/** + * The integration watchdog: notices a connection that has gone quiet or stayed + * broken, flags it for the connection health read, and tells the operator once. + * + * Runs on two schedules (`src/lib/jobs/integration-watchdog.ts`): + * + * hourly — cheap. One health load per account with a connection (the same + * loader the routes use), plus a 14-day history count only for a + * continuous source that is silent and not flagged yet. + * daily — the same pass, and it re-reads the history behind every standing + * silence flag, so a source its owner simply stopped using drops out + * once its own history no longer says it delivers. + * + * Judgement lives in `./watchdog-classify` (pure, unit-tested); this module + * gathers inputs and performs the side effects: one `IntegrationWatch` row per + * watched connection, one operator alert per incident through the existing + * admin path (`dispatchAdminSystemAlert`), one recovery note when it ends. + * + * Operator messages carry the provider, the account's email (as the ledger's + * own alerts do), a reason and an instant. Never a token, a value or an error + * message from the provider. + */ +import { prisma } from "@/lib/db"; +import type { MeasurementSource } from "@/generated/prisma/client"; +import { getEvent } from "@/lib/logging/context"; +import { dispatchAdminSystemAlert } from "./status"; +import { CONNECTION_KEYS, type ConnectionKey } from "./connection-health"; +import { + loadConnectionFacts, + type ConnectionFacts, +} from "./connection-health-load"; +import { INTEGRATION_MEASUREMENT_SOURCE } from "./metric-freshness"; +import { + HISTORY_DAYS, + SOURCE_PROFILES, + classifySilence, + currentIncident, + decideWatchTransition, + type IncidentReason, + type WatchRow, +} from "./watchdog-classify"; + +export type WatchdogMode = "hourly" | "daily"; + +export interface WatchdogSummary { + usersScanned: number; + usersFailed: number; + connectionsConsidered: number; + silencesFlagged: number; + incidentsOpened: number; + incidentsClosed: number; + notified: number; +} + +const DAY_MS = 24 * 60 * 60 * 1000; + +const LABELS: Record = { + withings: "Withings", + whoop: "WHOOP", + fitbit: "Fitbit", + "google-health": "Google Health", + polar: "Polar", + oura: "Oura", + strava: "Strava", + nightscout: "Nightscout", + "apple-health": "Apple Health", +}; + +function measurementSourceFor(key: ConnectionKey): MeasurementSource | null { + if (key === "apple-health") return "APPLE_HEALTH"; + return INTEGRATION_MEASUREMENT_SOURCE[key] ?? null; +} + +/** + * Distinct UTC days with data in the `HISTORY_DAYS` before the silence window. + * The window ends where silence starts to count, so the history being read is + * the one the silence is measured against. + */ +async function countDeliveredDays( + userId: string, + key: ConnectionKey, + now: Date, +): Promise { + const source = measurementSourceFor(key); + if (!source) return 0; + const to = new Date(now.getTime() - SOURCE_PROFILES[key].silenceWindowMs); + const from = new Date(to.getTime() - HISTORY_DAYS * DAY_MS); + const rows = await prisma.$queryRaw>` + SELECT COUNT(DISTINCT date_trunc('day', m."measured_at"))::int AS "days" + FROM "measurements" m + WHERE m."user_id" = ${userId} + AND m."source" = ${source}::"measurement_source" + AND m."deleted_at" IS NULL + AND m."measured_at" >= ${from} + AND m."measured_at" < ${to} + `; + return Number(rows[0]?.days ?? 0); +} + +/** + * Every account that has a connection the watchdog covers, or a watch row that + * still holds a flag or an incident (so a removed connection gets its incident + * closed). + */ +async function listWatchedUserIds(): Promise { + const [withings, whoop, fitbit, google, userRows, healthKit, watches] = + await Promise.all([ + prisma.withingsConnection.findMany({ select: { userId: true } }), + prisma.whoopConnection.findMany({ + where: { whoopUserId: { not: null } }, + select: { userId: true }, + }), + prisma.fitbitConnection.findMany({ select: { userId: true } }), + prisma.googleHealthConnection.findMany({ select: { userId: true } }), + prisma.user.findMany({ + where: { + OR: [ + { polarAccessTokenEncrypted: { not: null } }, + { ouraAccessTokenEncrypted: { not: null } }, + { stravaAccessTokenEncrypted: { not: null } }, + { nightscoutUrlEncrypted: { not: null } }, + { healthKitLastSyncedAt: { not: null } }, + ], + }, + select: { id: true }, + }), + prisma.healthKitTypeSync.findMany({ + distinct: ["userId"], + select: { userId: true }, + }), + prisma.integrationWatch.findMany({ + where: { + OR: [ + { silentSince: { not: null } }, + { incidentReason: { not: null } }, + ], + }, + select: { userId: true }, + }), + ]); + const ids = new Set(); + for (const r of [...withings, ...whoop, ...fitbit, ...google, ...healthKit]) { + ids.add(r.userId); + } + for (const r of userRows) ids.add(r.id); + for (const r of watches) ids.add(r.userId); + return [...ids].sort(); +} + +function formatInstant(value: Date): string { + return `${value.toISOString().slice(0, 16).replace("T", " ")} UTC`; +} + +const INCIDENT_COPY: Record string> = { + unexpected_silence: (l) => + `${l} has delivered nothing although it normally delivers every day. The connection reports no error.`, + reauth_required: (l) => `${l} needs to be reconnected by the account owner.`, + paused_after_errors: (l) => + `${l} paused itself after a day of failing syncs. Resume or reconnect it in the integration settings.`, + provider_errors: (l) => + `${l} syncs have been failing for more than a day. HealthLog keeps retrying.`, + sync_stalled: (l) => `${l} sync has not run for more than a day.`, + no_new_data: (l) => `${l} has delivered no new data for a while.`, + // The remaining reasons belong to healthy states and never open an + // incident; the copy exists so the table stays total. + data_arriving: (l) => `${l} is delivering data.`, + first_sync_pending: (l) => `${l} is waiting for its first sync.`, + plan_limited: (l) => `${l} withholds some data types under its plan.`, + permission_limited: (l) => `${l} lacks permission for some data types.`, +}; + +/** + * Pure formatter for the two operator messages, exported for tests. Plain + * text: the dispatcher sends it to Telegram without a parse mode, and the only + * interpolations are a provider label, an email and server-built instants. + */ +export function formatWatchdogMessage(input: { + kind: "incident" | "recovery"; + key: ConnectionKey; + subjectLabel: string; + reason: string; + since: Date | null; + closedBecause?: "healthy" | "disconnected" | "not_expected"; +}): { title: string; message: string } { + const label = LABELS[input.key]; + if (input.kind === "incident") { + const copy = + INCIDENT_COPY[input.reason as IncidentReason]?.(label) ?? + `${label} needs attention.`; + const since = input.since ? ` Since ${formatInstant(input.since)}.` : ""; + return { + title: `${label} needs attention for ${input.subjectLabel}`, + message: `${copy}${since}`, + }; + } + const why = + input.closedBecause === "disconnected" + ? "The connection was removed." + : input.closedBecause === "not_expected" + ? "It has not delivered regularly for a while, so the watchdog no longer expects data from it." + : "Data is arriving again."; + return { + title: `${label} recovered for ${input.subjectLabel}`, + message: `${label} is fine again. ${why}`, + }; +} + +/** Run one watchdog pass over every watched account. */ +export async function runIntegrationWatchdog(opts: { + mode: WatchdogMode; + now?: Date; +}): Promise { + const now = opts.now ?? new Date(); + const summary: WatchdogSummary = { + usersScanned: 0, + usersFailed: 0, + connectionsConsidered: 0, + silencesFlagged: 0, + incidentsOpened: 0, + incidentsClosed: 0, + notified: 0, + }; + const userIds = await listWatchedUserIds(); + for (const userId of userIds) { + summary.usersScanned += 1; + try { + await watchUser(userId, opts.mode, now, summary); + } catch (err) { + // One account's failure must not stop the pass for everyone else. + summary.usersFailed += 1; + getEvent()?.addWarning( + `integration watchdog failed for one account: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + return summary; +} + +/** Watch every connection of one account. Exported for tests. */ +export async function watchUser( + userId: string, + mode: WatchdogMode, + now: Date, + summary: WatchdogSummary, +): Promise { + const [facts, rows, subject] = await Promise.all([ + loadConnectionFacts(userId, CONNECTION_KEYS, now), + prisma.integrationWatch.findMany({ where: { userId } }), + prisma.user.findUnique({ where: { id: userId }, select: { email: true } }), + ]); + const rowByKey = new Map(rows.map((r) => [r.integration, r])); + const subjectLabel = subject?.email ?? "an account"; + + for (const key of CONNECTION_KEYS) { + const fact = facts.get(key); + const row = rowByKey.get(key) ?? null; + if (!fact) continue; + if (!fact.connected && row === null) continue; + summary.connectionsConsidered += 1; + await watchConnection({ + userId, + key, + fact, + row, + mode, + now, + subjectLabel, + summary, + }); + } +} + +async function watchConnection(input: { + userId: string; + key: ConnectionKey; + fact: ConnectionFacts; + row: (WatchRow & { id: string }) | null; + mode: WatchdogMode; + now: Date; + subjectLabel: string; + summary: WatchdogSummary; +}): Promise { + const { userId, key, fact, row, mode, now, summary } = input; + const profile = SOURCE_PROFILES[key]; + + // ── Silence ──────────────────────────────────────────────────────── + let silentSince: Date | null = null; + let closedBecause: "healthy" | "disconnected" | "not_expected" = + fact.connected ? "healthy" : "disconnected"; + if (fact.connected) { + const base = { + profile, + lastDataAt: fact.lastDataAt, + currentSilentSince: row?.silentSince ?? null, + now, + }; + // The daily pass re-reads the history behind a standing flag. + const recheck = mode === "daily" && row?.silentSince != null; + let verdict = classifySilence({ + ...base, + deliveredDays: recheck + ? await countDeliveredDays(userId, key, now) + : null, + }); + if (verdict.kind === "needs_history") { + verdict = classifySilence({ + ...base, + deliveredDays: await countDeliveredDays(userId, key, now), + }); + } + if (verdict.kind === "flag") silentSince = verdict.silentSince; + if (verdict.kind === "not_expected") closedBecause = "not_expected"; + } + if (silentSince && !row?.silentSince) summary.silencesFlagged += 1; + + // ── Incident ─────────────────────────────────────────────────────── + const incident = currentIncident({ + health: fact.connected ? fact.healthWithoutSilence : null, + silentSince, + ledgerAlerted: fact.ledgerAlerted, + now, + }); + const transition = decideWatchTransition({ + row, + incident, + ledgerAlerted: fact.ledgerAlerted, + }); + + let incidentReason = row?.incidentReason ?? null; + let incidentOpenedAt = row?.incidentOpenedAt ?? null; + let alertedAt = row?.alertedAt ?? null; + + const announce = async (reason: string, since: Date | null) => { + const text = formatWatchdogMessage({ + kind: "incident", + key, + subjectLabel: input.subjectLabel, + reason, + since, + }); + const outcome = await dispatchAdminSystemAlert({ + subjectUserId: userId, + title: text.title, + message: text.message, + metadata: { integration: key, reason, watchdog: "incident" }, + }); + if (outcome.channelsAttempted > 0) { + alertedAt = now; + summary.notified += 1; + } + }; + + switch (transition.kind) { + case "none": + break; + case "open": + summary.incidentsOpened += 1; + incidentReason = transition.reason; + incidentOpenedAt = now; + await announce(transition.reason, transition.since); + break; + case "adopt": + // The ledger already paged about this streak; record it as announced so + // the operator gets the recovery note and no second page. + summary.incidentsOpened += 1; + incidentReason = transition.reason; + incidentOpenedAt = now; + alertedAt = now; + break; + case "keep": + incidentReason = transition.reason; + if (alertedAt === null && fact.ledgerAlerted) { + alertedAt = now; + } else if (transition.announce && incident) { + // Opened earlier while no admin channel accepted the alert. + await announce(transition.reason, incident.since); + } + break; + case "close": + summary.incidentsClosed += 1; + if (transition.notify) { + const text = formatWatchdogMessage({ + kind: "recovery", + key, + subjectLabel: input.subjectLabel, + reason: transition.previousReason, + since: null, + closedBecause, + }); + const outcome = await dispatchAdminSystemAlert({ + subjectUserId: userId, + title: text.title, + message: text.message, + metadata: { + integration: key, + reason: transition.previousReason, + watchdog: "recovery", + }, + }); + if (outcome.channelsAttempted > 0) summary.notified += 1; + } + incidentReason = null; + incidentOpenedAt = null; + alertedAt = null; + break; + } + + // Nothing to remember and nothing remembered: no row. + if (row === null && silentSince === null && incidentReason === null) return; + + await prisma.integrationWatch.upsert({ + where: { userId_integration: { userId, integration: key } }, + create: { + userId, + integration: key, + silentSince, + incidentReason, + incidentOpenedAt, + alertedAt, + checkedAt: now, + }, + update: { + silentSince, + incidentReason, + incidentOpenedAt, + alertedAt, + checkedAt: now, + }, + }); +} diff --git a/src/lib/jobs/integration-watchdog.ts b/src/lib/jobs/integration-watchdog.ts new file mode 100644 index 000000000..ce6373037 --- /dev/null +++ b/src/lib/jobs/integration-watchdog.ts @@ -0,0 +1,71 @@ +/** + * Queues, schedules and handler for the integration watchdog + * (`src/lib/integrations/watchdog.ts`). + * + * Two queues rather than one queue with two payloads, so each schedule shows + * up under its own name in the job ledger. Both passes take the same advisory + * lock: the daily pass is a superset of the hourly one, and two passes side by + * side could each open the same incident and page twice. + */ +import type { Job } from "pg-boss"; +import { withJobLock } from "@/lib/jobs/job-lock"; +import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome"; +import { withBackgroundEvent } from "@/lib/logging/background"; +import { + runIntegrationWatchdog, + type WatchdogMode, +} from "@/lib/integrations/watchdog"; + +export const INTEGRATION_WATCHDOG_HOURLY_QUEUE = "integration-watchdog-hourly"; +export const INTEGRATION_WATCHDOG_DAILY_QUEUE = "integration-watchdog-daily"; + +/** + * :47 is a free minute off every hourly sync tick (used: 0, 5, 8, 11, 13, 15, + * 17, 18, 20, 30, 35, 41, 50), late enough in the hour that the syncs before it + * have written what they fetched. + */ +export const INTEGRATION_WATCHDOG_HOURLY_CRON = "47 * * * *"; +/** Once a day, inside the maintenance window, off the hourly :47. */ +export const INTEGRATION_WATCHDOG_DAILY_CRON = "52 4 * * *"; + +const LOCK_KEY = "job:integration-watchdog:pass"; + +export interface IntegrationWatchdogPayload { + triggeredAt?: string; +} + +function handlerFor(mode: WatchdogMode) { + return async ( + jobs: Job[], + ): Promise => { + void jobs; + return withBackgroundEvent( + `job.integration_watchdog.${mode}`, + async (evt) => { + try { + const run = await withJobLock(LOCK_KEY, () => + runIntegrationWatchdog({ mode }), + ); + if (!run.ran) return jobDone({ already_running: 1 }); + const s = run.result; + evt.addMeta("integration_watchdog", s); + return jobDone({ + users_scanned: s.usersScanned, + users_failed: s.usersFailed, + considered: s.connectionsConsidered, + silences_flagged: s.silencesFlagged, + incidents_opened: s.incidentsOpened, + incidents_closed: s.incidentsClosed, + notified: s.notified, + }); + } catch (err) { + evt.addWarning(`integration watchdog failed: ${err}`); + return jobFailed("integration watchdog pass failed", err); + } + }, + ); + }; +} + +export const handleIntegrationWatchdogHourly = handlerFor("hourly"); +export const handleIntegrationWatchdogDaily = handlerFor("daily"); diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 8f8a8a2cc..f62c00e65 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -125,6 +125,9 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "hourly_rows_upserted", "import_staging_swept", "imported", + // v1.42.4 — integration watchdog: operator incidents opened / closed. + "incidents_closed", + "incidents_opened", "in_slot", "in_window", "insufficient", @@ -242,6 +245,8 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "sent", "served", "side_effects_migrated", + // v1.42.4 — integration watchdog: sources newly flagged as silent. + "silences_flagged", "single_user", "skipped", "skipped_already_dispatched", diff --git a/src/lib/jobs/queue-runtime.ts b/src/lib/jobs/queue-runtime.ts index 2a9a88692..4efb8728d 100644 --- a/src/lib/jobs/queue-runtime.ts +++ b/src/lib/jobs/queue-runtime.ts @@ -295,6 +295,12 @@ export const QUEUE_RUNTIME: Readonly> = { "geo-backfill": short("At most 500 audit rows per pass."), "geolite2-fetch": short("One database download."), "tls-pin-monitor": short("One TLS handshake."), + "integration-watchdog-hourly": short( + "A few indexed reads per account with a connection and no provider call; a 14-day history count only for a silent source.", + ), + "integration-watchdog-daily": short( + "The hourly pass plus one 14-day history count per standing silence flag.", + ), "mood-reminder-cleanup": short("One retention DELETE."), "push-attempt-cleanup": short("One retention DELETE."), "arrival-reaction-cleanup": short("One retention DELETE."), diff --git a/src/lib/jobs/reminder/register-integration-sync.ts b/src/lib/jobs/reminder/register-integration-sync.ts index f241b4b62..9e5b55bec 100644 --- a/src/lib/jobs/reminder/register-integration-sync.ts +++ b/src/lib/jobs/reminder/register-integration-sync.ts @@ -70,6 +70,15 @@ import { type IntegrationBackfillAdmissionPayload, } from "@/lib/jobs/integration-backfill-admission"; import { drainIntegrationBackfillAdmission } from "@/lib/jobs/integration-backfill-drain"; +import { + INTEGRATION_WATCHDOG_DAILY_CRON, + INTEGRATION_WATCHDOG_DAILY_QUEUE, + INTEGRATION_WATCHDOG_HOURLY_CRON, + INTEGRATION_WATCHDOG_HOURLY_QUEUE, + handleIntegrationWatchdogDaily, + handleIntegrationWatchdogHourly, + type IntegrationWatchdogPayload, +} from "@/lib/jobs/integration-watchdog"; import { workerLog } from "./shared"; import { jobDone } from "@/lib/jobs/job-outcome"; import { @@ -320,6 +329,10 @@ const allQueues = [ STRAVA_BACKFILL_QUEUE, // v1.30.x — daily sweep for the native OIDC handoff ledger. OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, + // v1.42.4 — the integration watchdog: an hourly cheap pass and a daily + // pass that re-reads the history behind standing silence flags. + INTEGRATION_WATCHDOG_HOURLY_QUEUE, + INTEGRATION_WATCHDOG_DAILY_QUEUE, ]; const schedules: ScheduleEntry[] = [ @@ -377,6 +390,18 @@ const schedules: ScheduleEntry[] = [ [OURA_SYNC_QUEUE, OURA_SYNC_CRON], // v1.28.x — hourly Strava (:17) OAuth poll. [STRAVA_SYNC_QUEUE, STRAVA_SYNC_CRON], + // v1.42.4 — integration watchdog, hourly at :47 and daily at 04:52. The + // next tick is the retry: a missed pass only delays a flag by an hour. + [ + INTEGRATION_WATCHDOG_HOURLY_QUEUE, + INTEGRATION_WATCHDOG_HOURLY_CRON, + cronIsTheRetry, + ], + [ + INTEGRATION_WATCHDOG_DAILY_QUEUE, + INTEGRATION_WATCHDOG_DAILY_CRON, + cronIsTheRetry, + ], // v1.30.x — daily 03:28 Europe/Berlin prune for expired native OIDC handoffs. [ OIDC_NATIVE_HANDOFF_CLEANUP_QUEUE, @@ -725,6 +750,21 @@ export async function registerIntegrationSyncQueues( { localConcurrency: 1 }, handleStravaSync, ); + // v1.42.4 — integration watchdog. Single-flight per process; both queues + // also share one advisory lock, so the daily and hourly passes never run + // side by side on any replica. + await createAndWork( + boss, + INTEGRATION_WATCHDOG_HOURLY_QUEUE, + { localConcurrency: 1 }, + handleIntegrationWatchdogHourly, + ); + await createAndWork( + boss, + INTEGRATION_WATCHDOG_DAILY_QUEUE, + { localConcurrency: 1 }, + handleIntegrationWatchdogDaily, + ); // v1.28.x — self-converging Strava backfill. This source queue retains the // per-connection singleton and forwards admitted work to the shared drain. await createAndWork( diff --git a/src/lib/jobs/subsystem-surface.ts b/src/lib/jobs/subsystem-surface.ts index e4fc73d43..c4f51546c 100644 --- a/src/lib/jobs/subsystem-surface.ts +++ b/src/lib/jobs/subsystem-surface.ts @@ -43,6 +43,8 @@ const subsystemSurface = { "polar-sync": { audience: "account" }, "oura-sync": { audience: "account" }, "strava-sync": { audience: "account" }, + "integration-watchdog-hourly": { audience: "system" }, + "integration-watchdog-daily": { audience: "system" }, "strava-backfill": { audience: "system" }, "oidc-native-handoff-cleanup": { audience: "system" }, diff --git a/tests/integration/integration-watchdog.test.ts b/tests/integration/integration-watchdog.test.ts new file mode 100644 index 000000000..81c5ac97d --- /dev/null +++ b/tests/integration/integration-watchdog.test.ts @@ -0,0 +1,280 @@ +/** + * The integration watchdog and the connection health read against real + * Postgres: the 14-day history count, the one-alert-per-incident ledger, the + * recovery note, the on-demand exemption, the adoption of a page the sync + * ledger already sent, and the recent-call ring the diagnostics read. + * + * The dispatcher is mocked: what is asserted is how many operator messages + * the watchdog asks for, not their delivery. + */ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +process.env.ENCRYPTION_KEY ??= + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +import { getPrismaClient, truncateAllTables } from "./setup"; + +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); + +vi.mock("@/lib/notifications/dispatcher", () => ({ + dispatchNotification: vi.fn().mockResolvedValue({ + dispatched: true, + channelsAttempted: 1, + channelsSucceeded: 1, + }), +})); + +import { dispatchNotification } from "@/lib/notifications/dispatcher"; +import { runIntegrationWatchdog } from "@/lib/integrations/watchdog"; +import { loadConnectionHealth } from "@/lib/integrations/connection-health-load"; +import { + recordSyncFailure, + recordSyncSuccess, +} from "@/lib/integrations/status"; + +const HOUR = 60 * 60 * 1000; +const DAY = 24 * HOUR; +const NOW = new Date("2026-10-15T12:00:00.000Z"); +const USER = "user-watchdog-subject"; +const ADMIN = "user-watchdog-admin"; + +const prisma = () => getPrismaClient(); + +async function seedDailyReadings( + userId: string, + source: "OURA" | "WITHINGS", + days: number[], + newestHoursAgo: number, +) { + const rows = days.map((d) => ({ + userId, + type: "RESTING_HEART_RATE" as const, + value: 60, + unit: "bpm", + source, + measuredAt: new Date(NOW.getTime() - d * DAY - 6 * HOUR), + })); + rows.push({ + userId, + type: "RESTING_HEART_RATE" as const, + value: 60, + unit: "bpm", + source, + measuredAt: new Date(NOW.getTime() - newestHoursAgo * HOUR), + }); + await prisma().measurement.createMany({ data: rows }); +} + +function dispatchedTitles(): string[] { + return vi + .mocked(dispatchNotification) + .mock.calls.map((c) => String((c[0] as { title: string }).title)); +} + +beforeEach(async () => { + await truncateAllTables(prisma()); + await prisma().user.create({ + data: { + id: ADMIN, + username: "wd-admin", + email: "admin@example.test", + role: "ADMIN", + }, + }); + await prisma().user.create({ + data: { + id: USER, + username: "wd-subject", + email: "subject@example.test", + ouraAccessTokenEncrypted: "not-a-real-token", + }, + }); + vi.mocked(dispatchNotification).mockClear(); +}); + +describe("integration watchdog", () => { + it("flags a normally-daily wearable gone silent, alerts once, then sends one recovery note", async () => { + // Fourteen consecutive days of data before the window, newest 30 h ago. + await seedDailyReadings( + USER, + "OURA", + Array.from({ length: 13 }, (_, i) => i + 2), + 30, + ); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(NOW); + try { + await recordSyncSuccess(USER, "oura"); + + const first = await runIntegrationWatchdog({ mode: "hourly", now: NOW }); + expect(first.silencesFlagged).toBe(1); + expect(first.incidentsOpened).toBe(1); + expect(dispatchedTitles()).toEqual([ + "Oura needs attention for subject@example.test", + ]); + + const watch = await prisma().integrationWatch.findUniqueOrThrow({ + where: { userId_integration: { userId: USER, integration: "oura" } }, + }); + expect(watch.incidentReason).toBe("unexpected_silence"); + expect(watch.silentSince).not.toBeNull(); + expect(watch.alertedAt).not.toBeNull(); + + const health = await loadConnectionHealth(USER, ["oura"]); + expect(health.oura?.state).toBe("degraded"); + expect(health.oura?.reason).toBe("unexpected_silence"); + + // Next hour, and the daily pass: still silent, nothing new is sent. + const later = new Date(NOW.getTime() + HOUR); + await runIntegrationWatchdog({ mode: "hourly", now: later }); + await runIntegrationWatchdog({ mode: "daily", now: later }); + expect(dispatchNotification).toHaveBeenCalledTimes(1); + + // Data arrives again: the incident closes with one recovery note. + await prisma().measurement.create({ + data: { + userId: USER, + type: "RESTING_HEART_RATE", + value: 61, + unit: "bpm", + source: "OURA", + measuredAt: new Date(later.getTime() - 10 * 60 * 1000), + }, + }); + const recovered = await runIntegrationWatchdog({ + mode: "hourly", + now: new Date(later.getTime() + HOUR), + }); + expect(recovered.incidentsClosed).toBe(1); + expect(dispatchedTitles()).toEqual([ + "Oura needs attention for subject@example.test", + "Oura recovered for subject@example.test", + ]); + const cleared = await prisma().integrationWatch.findUniqueOrThrow({ + where: { userId_integration: { userId: USER, integration: "oura" } }, + }); + expect(cleared.silentSince).toBeNull(); + expect(cleared.incidentReason).toBeNull(); + + // Healthy and quiet: further passes send nothing. + await runIntegrationWatchdog({ + mode: "hourly", + now: new Date(later.getTime() + 2 * HOUR), + }); + expect(dispatchNotification).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + it("leaves a wearable alone when its owner wears it only now and then", async () => { + await seedDailyReadings(USER, "OURA", [3, 7, 11], 30); + await recordSyncSuccess(USER, "oura"); + const summary = await runIntegrationWatchdog({ mode: "hourly", now: NOW }); + expect(summary.silencesFlagged).toBe(0); + expect(dispatchNotification).not.toHaveBeenCalled(); + expect(await prisma().integrationWatch.count()).toBe(0); + }); + + it("never alerts on silence for an on-demand source such as a scale", async () => { + await prisma().user.update({ + where: { id: USER }, + data: { ouraAccessTokenEncrypted: null }, + }); + await prisma().withingsConnection.create({ + data: { + userId: USER, + withingsUserId: "w-1", + accessToken: "x", + refreshToken: "y", + tokenExpiresAt: new Date(NOW.getTime() + DAY), + scope: "user.metrics,user.activity", + }, + }); + // Daily weigh-ins for two weeks, then nothing for three days. + await seedDailyReadings( + USER, + "WITHINGS", + Array.from({ length: 13 }, (_, i) => i + 4), + 72, + ); + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(NOW); + try { + await recordSyncSuccess(USER, "withings"); + const summary = await runIntegrationWatchdog({ mode: "daily", now: NOW }); + expect(summary.connectionsConsidered).toBe(1); + expect(summary.silencesFlagged).toBe(0); + expect(dispatchNotification).not.toHaveBeenCalled(); + const health = await loadConnectionHealth(USER, ["withings"]); + expect(health.withings?.state).toBe("ok"); + } finally { + vi.useRealTimers(); + } + }); + + it("adopts a failure streak the ledger already paged about and reports its end once", async () => { + await seedDailyReadings(USER, "OURA", [2], 5); + for (let i = 0; i < 3; i++) { + await recordSyncFailure({ + userId: USER, + integration: "oura", + kind: "transient", + message: "Oura error: 503 with a body that must never be published", + errorCode: "503", + }); + } + // The ledger's own 3-strike page. + expect(dispatchNotification).toHaveBeenCalledTimes(1); + + const now = new Date(); + const adopted = await runIntegrationWatchdog({ mode: "hourly", now }); + expect(adopted.incidentsOpened).toBe(1); + expect(dispatchNotification).toHaveBeenCalledTimes(1); + + await recordSyncSuccess(USER, "oura"); + const closed = await runIntegrationWatchdog({ mode: "hourly", now }); + expect(closed.incidentsClosed).toBe(1); + expect(dispatchNotification).toHaveBeenCalledTimes(2); + expect(dispatchedTitles()[1]).toBe( + "Oura recovered for subject@example.test", + ); + }); +}); + +describe("recent-call ring", () => { + it("keeps the newest ten attempts, newest first, without any message text", async () => { + for (let i = 0; i < 7; i++) { + await recordSyncFailure({ + userId: USER, + integration: "oura", + kind: "transient", + message: `secret-bearing upstream body ${i}`, + errorCode: i === 6 ? "token=abc def" : "503", + leg: "sleep", + }); + } + for (let i = 0; i < 5; i++) { + await recordSyncSuccess(USER, "oura", { leg: "sleep" }); + } + const row = await prisma().integrationStatus.findUniqueOrThrow({ + where: { userId_integration: { userId: USER, integration: "oura" } }, + }); + const calls = row.recentCalls as Array>; + expect(calls).toHaveLength(10); + expect(calls.slice(0, 5).every((c) => c.status === "ok")).toBe(true); + expect(calls[5]).toMatchObject({ + endpoint: "sleep", + status: "transient", + errorCode: null, + }); + expect(calls[6]).toMatchObject({ errorCode: "503" }); + expect(JSON.stringify(calls)).not.toContain("secret"); + + const health = await loadConnectionHealth(USER, ["oura"]); + expect(health.oura?.diagnostics.recentCalls).toHaveLength(10); + expect(JSON.stringify(health.oura)).not.toContain(USER); + }); +}); From 8162afc0e1c37fc5b06863f1e26585104b68c350 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 15 Oct 2026 11:09:50 +0200 Subject: [PATCH 3/4] docs(api): publish the connection health state --- docs/api/openapi.yaml | 216 +++++++++++++++++++++++++ src/lib/openapi/routes/integrations.ts | 142 ++++++++++++++++ 2 files changed, 358 insertions(+) diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index c1dbd36c3..317c79c73 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -73461,6 +73461,12 @@ components: anyOf: - $ref: "#/components/schemas/AppleHealthSyncProgress" - type: "null" + health: + description: The unified Apple Health connection state. Null when the phone has never delivered anything (or the read + failed). Present on the GET read; omitted from the PATCH echo. + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" required: - entries - lastSyncedAt @@ -73598,6 +73604,171 @@ components: additionalProperties: false description: "Server-side Apple Health backfill progress: how many rows have been accepted and how far back in time they reach. Only what the server actually knows — no ETA, no device-side queue state." + ConnectionHealth: + type: object + properties: + state: + type: string + enum: + - ok + - limited + - degraded + - action_required + description: "`ok` data is arriving (or the first sync is still due). `limited` it works, but the types in `unavailable` + cannot be delivered (provider plan or missing permission). `degraded` temporary provider trouble or + unexpected silence; the server keeps retrying, `since` dates it. `action_required` the user has to do the + one thing in `action`." + reason: + type: string + enum: + - data_arriving + - first_sync_pending + - plan_limited + - permission_limited + - provider_errors + - sync_stalled + - no_new_data + - unexpected_silence + - reauth_required + - paused_after_errors + description: "Why the state is what it is. ok: `data_arriving`, `first_sync_pending`. limited: `plan_limited`, + `permission_limited`. degraded: `provider_errors`, `sync_stalled`, `no_new_data`, `unexpected_silence` (the + integration watchdog saw a source that normally delivers daily go quiet). action_required: + `reauth_required`, `paused_after_errors`. Pick the sentence from this code alone." + since: + anyOf: + - type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + - type: "null" + description: When the current degraded / action_required condition started. Null for ok and limited. + action: + anyOf: + - type: string + enum: + - reconnect + - resume + - type: "null" + description: "The one action that resolves the state: `reconnect` (run the provider's connect flow again) or `resume` + (POST the provider's resume route). Also `reconnect` on a `limited` connection whose gap is a missing + permission. Null when nothing the user can do helps." + unavailable: + type: array + items: + type: object + properties: + item: + type: string + description: Provider data-type key, e.g. `spo2`, `vo2Max`, `activity`. + cause: + type: string + enum: + - plan + - permission + required: + - item + - cause + additionalProperties: false + description: Data types this connection cannot deliver. Present in every state, empty when nothing is withheld. + lastSuccessAt: + anyOf: + - type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + - type: "null" + description: Last successful sync. For Apple Health, the last accepted batch. + lastDataAt: + anyOf: + - type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + - type: "null" + description: "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." + diagnostics: + type: object + properties: + ledgerState: + anyOf: + - type: string + - type: "null" + description: The stored sync-ledger state. Null for Apple Health. + failingLegs: + type: array + items: + type: string + failingSince: + anyOf: + - type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + - type: "null" + lastAttemptAt: + anyOf: + - type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + - type: "null" + verdict: + type: string + description: The `SyncHealth` verdict the state was derived from. + recentCalls: + maxItems: 10 + type: array + items: + type: object + properties: + endpoint: + type: string + description: Sync leg name (`sleep`, `measures`, `sync`, `token_refresh`), or `batch:` for Apple Health. Never a + URL. + status: + type: string + enum: + - ok + - transient + - persistent + - reauth_required + at: + type: string + format: date-time + pattern: ^(?:(?:\d\d[2468][048]|\d\d[13579][26]|\d\d0[48]|[02468][048]00|[13579][26]00)-02-29|\d{4}-(?:(?:0[13578]|1[02])-(?:0[1-9]|[12]\d|3[01])|(?:0[469]|11)-(?:0[1-9]|[12]\d|30)|(?:02)-(?:0[1-9]|1\d|2[0-8])))T(?:(?:[01]\d|2[0-3]):[0-5]\d(?::[0-5]\d(?:\.\d+)?)?(?:Z|([+-](?:[01]\d|2[0-3]):[0-5]\d)))$ + errorCode: + anyOf: + - type: string + - type: "null" + description: Short error code (`401`, `invalid_grant`, `scope_missing`); anything not shaped like one is dropped. + required: + - endpoint + - status + - at + - errorCode + additionalProperties: false + description: The last ten sync attempts, newest first. + required: + - ledgerState + - failingLegs + - failingSince + - lastAttemptAt + - verdict + - recentCalls + additionalProperties: false + description: Redacted diagnostics for a bug report. Carries no token, no error message, no health value and no account + identifier, so a client may offer it for copying as is. + required: + - state + - reason + - since + - action + - unavailable + - lastSuccessAt + - lastDataAt + - diagnostics + additionalProperties: false + description: The one server-resolved health state of a connection (v1.42.4). It folds the ledger state, the liveness + verdict, plan and permission gaps, provider reauth flags and the integration watchdog's silence flag into four + states. Clients render it and never recompute it. The older health-shaped fields beside it stay for one release + after the clients read `health`, then go. HealthKitConfigPatchEnvelope: type: object properties: @@ -73933,6 +74104,12 @@ components: - type: "null" syncHealth: $ref: "#/components/schemas/SyncHealth" + health: + description: The unified connection state. Present whenever the connection exists; null in the rare case it resolved to + none. Absent on the not-connected body. + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" required: - connected - configured @@ -74129,6 +74306,12 @@ components: anyOf: - type: string - type: "null" + health: + description: The unified connection state. Present whenever the connection exists; null in the rare case it resolved to + none. Absent on the not-connected body. + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" required: - connected - configured @@ -74217,6 +74400,12 @@ components: hasActivityScope: description: Pre-computed convenience flag the reconnect banner reads. A null `scope` counts as missing activity. type: boolean + health: + description: The unified connection state. Present whenever the connection exists; null in the rare case it resolved to + none. Absent on the not-connected body. + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" required: - connected - configured @@ -74531,6 +74720,12 @@ components: anyOf: - type: string - type: "null" + health: + description: The unified connection state. Present whenever the connection exists; null in the rare case it resolved to + none. Absent on the not-connected body. + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" required: - connected - configured @@ -74694,6 +74889,12 @@ components: anyOf: - type: string - type: "null" + health: + description: The unified connection state. Present whenever the connection exists; null in the rare case it resolved to + none. Absent on the not-connected body. + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" needsReauth: description: "The stored refresh token was rejected with `invalid_grant` — Google expires it after 7 days while the OAuth client is in Testing publishing mode, and a user-revoked grant lands here too. Distinct from `parked` @@ -74929,6 +75130,12 @@ components: - type: "null" syncHealth: $ref: "#/components/schemas/SyncHealth" + health: + description: The unified connection state. Present whenever the connection exists; null in the rare case it resolved to + none. Absent on the not-connected body. + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" required: - connected - configured @@ -75468,6 +75675,14 @@ components: allowPrivateHost: description: Nightscout only. type: boolean + health: + anyOf: + - $ref: "#/components/schemas/ConnectionHealth" + - type: "null" + description: The unified connection state, null when the provider is not connected. It supersedes `state`, `lastError`, + `consecutiveFailuresByKind`, `syncHealth`, `tokenExpired`, `needsReauth`, `hasActivityScope`, + `planUnavailable` and `scopeReconnect` on this entry, which go once the settings surface renders `health` + (no other client reads this envelope). required: - integration - state @@ -75479,6 +75694,7 @@ components: - metricFreshness - connected - configured + - health additionalProperties: false description: "One provider's entry in the consolidated envelope: the ledger core and the liveness verdict, which every entry carries, plus the extras that provider has. The optional fields are not optional per request — each is diff --git a/src/lib/openapi/routes/integrations.ts b/src/lib/openapi/routes/integrations.ts index eb8be83ed..9919a20c4 100644 --- a/src/lib/openapi/routes/integrations.ts +++ b/src/lib/openapi/routes/integrations.ts @@ -31,6 +31,12 @@ * come from `src/lib/validations/*` so the wire contract stays single-source. */ import { z } from "zod/v4"; +import { + CONNECTION_HEALTH_ACTIONS, + CONNECTION_HEALTH_REASONS, + CONNECTION_HEALTH_STATES, + RECENT_CALL_STATUSES, +} from "@/lib/integrations/connection-health"; import type { ZodOpenApiObject } from "zod-openapi"; import { fitbitCredentialsSchema } from "@/lib/validations/fitbit"; import { googleHealthCredentialsSchema } from "@/lib/validations/google-health"; @@ -97,6 +103,101 @@ const syncHealth = z "Server-resolved sync-health verdict. The single source of liveness truth — `lastSyncedAt` alone cannot express 'this pipe stopped delivering'.", }); +// Mirrors `ConnectionHealth` in `src/lib/integrations/connection-health.ts`: +// the one server-resolved state per connection every client renders. +const connectionHealth = z + .object({ + state: z + .enum(CONNECTION_HEALTH_STATES) + .describe( + "`ok` data is arriving (or the first sync is still due). `limited` it works, but the types in `unavailable` cannot be delivered (provider plan or missing permission). `degraded` temporary provider trouble or unexpected silence; the server keeps retrying, `since` dates it. `action_required` the user has to do the one thing in `action`.", + ), + reason: z + .enum(CONNECTION_HEALTH_REASONS) + .describe( + "Why the state is what it is. ok: `data_arriving`, `first_sync_pending`. limited: `plan_limited`, `permission_limited`. degraded: `provider_errors`, `sync_stalled`, `no_new_data`, `unexpected_silence` (the integration watchdog saw a source that normally delivers daily go quiet). action_required: `reauth_required`, `paused_after_errors`. Pick the sentence from this code alone.", + ), + since: z.iso + .datetime({ offset: true }) + .nullable() + .describe( + "When the current degraded / action_required condition started. Null for ok and limited.", + ), + action: z + .enum(CONNECTION_HEALTH_ACTIONS) + .nullable() + .describe( + "The one action that resolves the state: `reconnect` (run the provider's connect flow again) or `resume` (POST the provider's resume route). Also `reconnect` on a `limited` connection whose gap is a missing permission. Null when nothing the user can do helps.", + ), + unavailable: z + .array( + z.object({ + item: z + .string() + .describe( + "Provider data-type key, e.g. `spo2`, `vo2Max`, `activity`.", + ), + cause: z.enum(["plan", "permission"]), + }), + ) + .describe( + "Data types this connection cannot deliver. Present in every state, empty when nothing is withheld.", + ), + lastSuccessAt: z.iso + .datetime({ offset: true }) + .nullable() + .describe( + "Last successful sync. For Apple Health, the last accepted batch.", + ), + lastDataAt: z.iso + .datetime({ offset: true }) + .nullable() + .describe( + "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.", + ), + diagnostics: z + .object({ + ledgerState: z + .string() + .nullable() + .describe("The stored sync-ledger state. Null for Apple Health."), + failingLegs: z.array(z.string()), + failingSince: z.iso.datetime({ offset: true }).nullable(), + lastAttemptAt: z.iso.datetime({ offset: true }).nullable(), + verdict: z + .string() + .describe("The `SyncHealth` verdict the state was derived from."), + recentCalls: z + .array( + z.object({ + endpoint: z + .string() + .describe( + "Sync leg name (`sleep`, `measures`, `sync`, `token_refresh`), or `batch:` for Apple Health. Never a URL.", + ), + status: z.enum(RECENT_CALL_STATUSES), + at: z.iso.datetime({ offset: true }), + errorCode: z + .string() + .nullable() + .describe( + "Short error code (`401`, `invalid_grant`, `scope_missing`); anything not shaped like one is dropped.", + ), + }), + ) + .max(10) + .describe("The last ten sync attempts, newest first."), + }) + .describe( + "Redacted diagnostics for a bug report. Carries no token, no error message, no health value and no account identifier, so a client may offer it for copying as is.", + ), + }) + .meta({ + id: "ConnectionHealth", + description: + "The one server-resolved health state of a connection (v1.42.4). It folds the ledger state, the liveness verdict, plan and permission gaps, provider reauth flags and the integration watchdog's silence flag into four states. Clients render it and never recompute it. The older health-shaped fields beside it stay for one release after the clients read `health`, then go.", + }); + const metricFreshnessEntry = z .object({ type: z @@ -212,6 +313,12 @@ const healthKitConfigResponse = z .describe( "Backfill/sync progress summary. Present on the GET read (null when the progress read failed); omitted from the PATCH echo.", ), + health: connectionHealth + .nullable() + .optional() + .describe( + "The unified Apple Health connection state. Null when the phone has never delivered anything (or the read failed). Present on the GET read; omitted from the PATCH echo.", + ), }) .meta({ id: "HealthKitConfigResponse", @@ -405,6 +512,12 @@ const nightscoutStatusResponse = z "Last recorded failure message, decrypted from the ledger. Null when the last attempt succeeded or none has run.", ), syncHealth: syncHealth.optional(), + health: connectionHealth + .nullable() + .optional() + .describe( + "The unified connection state. Present whenever the connection exists; null in the rare case it resolved to none. Absent on the not-connected body.", + ), }) .meta({ id: "NightscoutStatusResponse", @@ -483,6 +596,12 @@ const whoopStatusResponse = z .describe( "Space-separated OAuth scope string granted by WHOOP. Null on a legacy connection that predates scope tracking.", ), + health: connectionHealth + .nullable() + .optional() + .describe( + "The unified connection state. Present whenever the connection exists; null in the rare case it resolved to none. Absent on the not-connected body.", + ), }) .meta({ id: "WhoopStatusResponse", @@ -529,6 +648,12 @@ const withingsStatusResponse = z .describe( "Pre-computed convenience flag the reconnect banner reads. A null `scope` counts as missing activity.", ), + health: connectionHealth + .nullable() + .optional() + .describe( + "The unified connection state. Present whenever the connection exists; null in the rare case it resolved to none. Absent on the not-connected body.", + ), }) .meta({ id: "WithingsStatusResponse", @@ -810,6 +935,12 @@ function connectionStatusShape() { .describe( "OAuth scope string granted at authorisation. Null on a legacy connection that predates scope tracking.", ), + health: connectionHealth + .nullable() + .optional() + .describe( + "The unified connection state. Present whenever the connection exists; null in the rare case it resolved to none. Absent on the not-connected body.", + ), }; } @@ -863,6 +994,12 @@ const oauthProviderStatusResponse = z lastAttemptAt: z.iso.datetime({ offset: true }).nullable().optional(), lastError: z.string().nullable().optional(), syncHealth: syncHealth.optional(), + health: connectionHealth + .nullable() + .optional() + .describe( + "The unified connection state. Present whenever the connection exists; null in the rare case it resolved to none. Absent on the not-connected body.", + ), }) .meta({ id: "OAuthProviderStatusResponse", @@ -947,6 +1084,11 @@ const integrationEnvelopeEntry = z ), hasToken: z.boolean().optional().describe("Nightscout only."), allowPrivateHost: z.boolean().optional().describe("Nightscout only."), + health: connectionHealth + .nullable() + .describe( + "The unified connection state, null when the provider is not connected. It supersedes `state`, `lastError`, `consecutiveFailuresByKind`, `syncHealth`, `tokenExpired`, `needsReauth`, `hasActivityScope`, `planUnavailable` and `scopeReconnect` on this entry, which go once the settings surface renders `health` (no other client reads this envelope).", + ), }) .meta({ id: "IntegrationStatusEntry", From f2646bc186ef35c99a57b66f511652c123ccbef5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 15 Oct 2026 11:15:06 +0200 Subject: [PATCH 4/4] test(integrations): cover the health field on the status routes --- .../status/__tests__/route.test.ts | 40 +++++++++++++++++++ .../api/oura/status/__tests__/route.test.ts | 3 ++ .../api/polar/status/__tests__/route.test.ts | 3 ++ .../api/whoop/status/__tests__/route.test.ts | 3 ++ 4 files changed, 49 insertions(+) diff --git a/src/app/api/integrations/status/__tests__/route.test.ts b/src/app/api/integrations/status/__tests__/route.test.ts index 25b08cc78..67daaed0a 100644 --- a/src/app/api/integrations/status/__tests__/route.test.ts +++ b/src/app/api/integrations/status/__tests__/route.test.ts @@ -5,6 +5,9 @@ vi.mock("@/lib/api-handler", () => ({ requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), })); +vi.mock("@/lib/integrations/connection-health-load", () => ({ + loadConnectionHealth: vi.fn().mockResolvedValue({}), +})); vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() }, @@ -444,3 +447,40 @@ describe("/api/integrations/status — what a provider withholds by contract", ( ).toEqual([]); }); }); + +describe("/api/integrations/status — the unified connection health", () => { + beforeEach(() => { + vi.clearAllMocks(); + for (const key of Object.keys(perIntegrationLedger)) { + delete perIntegrationLedger[key]; + } + }); + + it("puts each provider's resolved health on its own entry, null where not connected", async () => { + const { loadConnectionHealth } = + await import("@/lib/integrations/connection-health-load"); + const ouraHealth = { state: "limited", reason: "permission_limited" }; + vi.mocked(loadConnectionHealth).mockResolvedValueOnce({ + withings: null, + whoop: null, + fitbit: null, + "google-health": null, + polar: null, + oura: ouraHealth, + strava: null, + nightscout: null, + "apple-health": null, + } as never); + userFind.mockResolvedValue({ ouraAccessTokenEncrypted: "tok" }); + + const entries = (await fetchEntries()) as Array< + Entry & { health?: unknown } + >; + expect(entries.find((e) => e.integration === "oura")?.health).toEqual( + ouraHealth, + ); + for (const entry of entries.filter((e) => e.integration !== "oura")) { + expect(entry.health).toBeNull(); + } + }); +}); diff --git a/src/app/api/oura/status/__tests__/route.test.ts b/src/app/api/oura/status/__tests__/route.test.ts index dc73142d0..6dc845f78 100644 --- a/src/app/api/oura/status/__tests__/route.test.ts +++ b/src/app/api/oura/status/__tests__/route.test.ts @@ -5,6 +5,9 @@ vi.mock("@/lib/api-handler", () => ({ requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), })); +vi.mock("@/lib/integrations/connection-health-load", () => ({ + loadOneConnectionHealth: vi.fn().mockResolvedValue(null), +})); vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() } }, })); diff --git a/src/app/api/polar/status/__tests__/route.test.ts b/src/app/api/polar/status/__tests__/route.test.ts index 7a7d30162..f9b4adbfd 100644 --- a/src/app/api/polar/status/__tests__/route.test.ts +++ b/src/app/api/polar/status/__tests__/route.test.ts @@ -5,6 +5,9 @@ vi.mock("@/lib/api-handler", () => ({ requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), })); +vi.mock("@/lib/integrations/connection-health-load", () => ({ + loadOneConnectionHealth: vi.fn().mockResolvedValue(null), +})); vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() } }, })); diff --git a/src/app/api/whoop/status/__tests__/route.test.ts b/src/app/api/whoop/status/__tests__/route.test.ts index 3744084de..80d07e6cf 100644 --- a/src/app/api/whoop/status/__tests__/route.test.ts +++ b/src/app/api/whoop/status/__tests__/route.test.ts @@ -5,6 +5,9 @@ vi.mock("@/lib/api-handler", () => ({ requireAuth: vi.fn(async () => ({ user: { id: "u1" } })), })); +vi.mock("@/lib/integrations/connection-health-load", () => ({ + loadOneConnectionHealth: vi.fn().mockResolvedValue(null), +})); vi.mock("@/lib/db", () => ({ prisma: { user: { findUnique: vi.fn() },