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] 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); + }); +});