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