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