diff --git a/src/lib/integrations/__tests__/status.test.ts b/src/lib/integrations/__tests__/status.test.ts index f9a8dfe99..8315a1826 100644 --- a/src/lib/integrations/__tests__/status.test.ts +++ b/src/lib/integrations/__tests__/status.test.ts @@ -46,6 +46,7 @@ import { markReconnected, isReauthRequired, getPersistentFailureThreshold, + resumeIntegrationFromPark, } from "../status"; import { prisma } from "@/lib/db"; import { auditLog } from "@/lib/auth/audit"; @@ -95,6 +96,14 @@ describe("recordSyncSuccess", () => { lastAttemptAt: expect.any(Date), lastError: null, consecutiveFailures: 0, + // v1.4.43 W14 — success clears every per-kind bucket too so + // the next failure starts from zero in its own bucket. + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 0, + persistent: 0, + }, + persistentFailureStartedAt: null, alertedAt: null, }); }); @@ -359,12 +368,24 @@ describe("recordSyncFailure — admin Telegram skipped silently when no admins", }); describe("isReauthRequired", () => { - it("returns true only when state === error_reauth", async () => { + it("returns true when state === error_reauth", async () => { vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ state: "error_reauth", } as never); expect(await isReauthRequired("u1", "withings")).toBe(true); + }); + it("returns true when state === parked (v1.4.43 W14)", async () => { + // A parked integration must short-circuit the sync entry-point + // the same way a reauth-required one does — neither will succeed + // without user / operator action. + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + state: "parked", + } as never); + expect(await isReauthRequired("u1", "withings")).toBe(true); + }); + + it("returns false for transient errors and absent rows", async () => { vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ state: "error_transient", } as never); @@ -398,6 +419,13 @@ describe("markReauthRequired / markDisconnected / markReconnected", () => { state: "disconnected", lastError: null, consecutiveFailures: 0, + // v1.4.43 W14 — disconnect must clear the per-kind buckets too. + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 0, + persistent: 0, + }, + persistentFailureStartedAt: null, alertedAt: null, }); }); @@ -412,7 +440,283 @@ describe("markReauthRequired / markDisconnected / markReconnected", () => { state: "connected", lastError: null, consecutiveFailures: 0, + // v1.4.43 W14 — reconnect clears the per-kind buckets so the + // next sync starts on a clean slate. + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 0, + persistent: 0, + }, + persistentFailureStartedAt: null, alertedAt: null, }); }); }); + +// ── v1.4.43 W14 — per-kind buckets + 24h park behaviour ──────────────── +// +// The four pins below lock in the B4 + B7 contract: +// +// 1. A persistent failure increments ONLY the persistent bucket; +// an intervening transient failure does NOT reset the persistent +// streak. +// 2. The persistent-streak start timestamp is stamped on the FIRST +// persistent failure of a streak and preserved across subsequent +// persistent failures. +// 3. Once the persistent streak has been running for >24h, the state +// flips to `parked` and an `integrations.parked` audit row is +// written. +// 4. `resumeIntegrationFromPark` clears state back to `connected`, +// zeroes every bucket, and writes an `integrations.resumed` audit +// row (only when the row was actually parked — idempotent for +// non-parked inputs). + +describe("recordSyncFailure — v1.4.43 W14 per-kind buckets", () => { + it("increments ONLY the matching bucket on each failure", async () => { + // Existing row: transient bucket already at 2 from earlier hiccups. + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + consecutiveFailures: 2, + consecutiveFailuresByKind: { + transient: 2, + reauth_required: 0, + persistent: 0, + }, + persistentFailureStartedAt: null, + alertedAt: null, + } as never); + vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({ + consecutiveFailures: 3, + alertedAt: null, + } as never); + + await recordSyncFailure({ + userId: "u1", + integration: "withings", + kind: "persistent", + message: "Withings measure error: 293 - invalid params", + errorCode: "293", + }); + + const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock + .calls[0][0]; + // transient stays at 2 (unchanged), persistent ticks to 1, + // reauth_required stays at 0. + expect(upsertArgs.update.consecutiveFailuresByKind).toEqual({ + transient: 2, + reauth_required: 0, + persistent: 1, + }); + // First-ever persistent failure stamps the streak anchor. + expect(upsertArgs.update.persistentFailureStartedAt).toBeInstanceOf(Date); + }); + + it("preserves persistentFailureStartedAt across multiple persistent failures", async () => { + const streakStart = new Date(Date.now() - 60 * 60 * 1000); // 1h ago + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + consecutiveFailures: 5, + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 0, + persistent: 5, + }, + persistentFailureStartedAt: streakStart, + alertedAt: new Date(Date.now() - 30 * 60 * 1000), // already alerted + } as never); + vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({ + consecutiveFailures: 6, + alertedAt: new Date(Date.now() - 30 * 60 * 1000), + } as never); + + await recordSyncFailure({ + userId: "u1", + integration: "withings", + kind: "persistent", + message: "Withings measure error: 293", + errorCode: "293", + }); + + const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock + .calls[0][0]; + // The anchor is preserved exactly — that's the whole point of + // tracking the streak start instead of recomputing it from + // updated_at. + expect(upsertArgs.update.persistentFailureStartedAt).toBe(streakStart); + expect(upsertArgs.update.state).toBe("error_transient"); // < 24h still + expect(upsertArgs.update.consecutiveFailuresByKind).toEqual({ + transient: 0, + reauth_required: 0, + persistent: 6, + }); + }); + + it("a transient failure does NOT reset the persistent bucket nor the streak anchor", async () => { + const streakStart = new Date(Date.now() - 6 * 60 * 60 * 1000); // 6h ago + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + consecutiveFailures: 4, + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 0, + persistent: 4, + }, + persistentFailureStartedAt: streakStart, + alertedAt: null, + } as never); + vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({ + consecutiveFailures: 5, + alertedAt: null, + } as never); + + await recordSyncFailure({ + userId: "u1", + integration: "withings", + kind: "transient", + message: "Withings refresh error: 503", + errorCode: "503", + }); + + const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock + .calls[0][0]; + // Persistent bucket UNCHANGED — this is the v1.4.43 W14 fix. + // Under the legacy single-counter behaviour the burst above would + // have shown as "5 consecutive failures, state=error_transient" + // and the persistent streak's age would have been masked. + expect(upsertArgs.update.consecutiveFailuresByKind).toEqual({ + transient: 1, + reauth_required: 0, + persistent: 4, + }); + expect(upsertArgs.update.persistentFailureStartedAt).toBe(streakStart); + }); + + it("flips state to `parked` and writes an audit row once persistent streak > 24h", async () => { + const streakStart = new Date(Date.now() - 25 * 60 * 60 * 1000); // 25h ago + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + consecutiveFailures: 30, + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 0, + persistent: 30, + }, + persistentFailureStartedAt: streakStart, + alertedAt: new Date(Date.now() - 23 * 60 * 60 * 1000), + } as never); + vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({ + consecutiveFailures: 31, + alertedAt: new Date(Date.now() - 23 * 60 * 60 * 1000), + } as never); + + await recordSyncFailure({ + userId: "u1", + integration: "withings", + kind: "persistent", + message: "Withings measure error: 293", + errorCode: "293", + }); + + const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock + .calls[0][0]; + expect(upsertArgs.update.state).toBe("parked"); + expect(auditLog).toHaveBeenCalledWith( + "integrations.parked", + expect.objectContaining({ + userId: "u1", + details: expect.objectContaining({ + integration: "withings", + reason: "persistent_24h", + persistentFailureStartedAt: streakStart.toISOString(), + }), + }), + ); + }); + + it("back-fills the bucket from the legacy single counter on a pre-migration row", async () => { + // Legacy row: counter is 3, no bucket JSON yet (predates 0075). + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + consecutiveFailures: 3, + consecutiveFailuresByKind: null, + persistentFailureStartedAt: null, + alertedAt: null, + } as never); + vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce({ + consecutiveFailures: 4, + alertedAt: null, + } as never); + vi.mocked(prisma.user.findUnique).mockResolvedValueOnce({ + email: "u@example.com", + } as never); + vi.mocked(prisma.user.findMany).mockResolvedValueOnce([] as never); + vi.mocked(prisma.integrationStatus.update).mockResolvedValueOnce( + {} as never, + ); + + await recordSyncFailure({ + userId: "u1", + integration: "withings", + kind: "reauth_required", + message: "Withings refresh error: 100 - invalid_grant", + errorCode: "100", + }); + + const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock + .calls[0][0]; + // The 3 legacy failures get bucketed into the CURRENT failure's + // kind (reauth_required) — that's the kind the row was last + // tagged with, and back-filling into a different kind would + // under-count the existing streak. + expect(upsertArgs.update.consecutiveFailuresByKind).toEqual({ + transient: 0, + reauth_required: 3, + persistent: 0, + }); + }); +}); + +describe("resumeIntegrationFromPark", () => { + it("clears a parked row back to connected with all buckets zeroed", async () => { + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + state: "parked", + } as never); + vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce( + {} as never, + ); + + const result = await resumeIntegrationFromPark("u1", "withings"); + expect(result.wasParked).toBe(true); + + const upsertArgs = vi.mocked(prisma.integrationStatus.upsert).mock + .calls[0][0]; + expect(upsertArgs.update).toEqual({ + state: "connected", + lastError: null, + consecutiveFailures: 0, + consecutiveFailuresByKind: { + transient: 0, + reauth_required: 0, + persistent: 0, + }, + persistentFailureStartedAt: null, + alertedAt: null, + }); + + expect(auditLog).toHaveBeenCalledWith( + "integrations.resumed", + expect.objectContaining({ + userId: "u1", + details: expect.objectContaining({ integration: "withings" }), + }), + ); + }); + + it("is idempotent — calling against a non-parked row does NOT write an audit row", async () => { + vi.mocked(prisma.integrationStatus.findUnique).mockResolvedValueOnce({ + state: "connected", + } as never); + vi.mocked(prisma.integrationStatus.upsert).mockResolvedValueOnce( + {} as never, + ); + + const result = await resumeIntegrationFromPark("u1", "withings"); + expect(result.wasParked).toBe(false); + expect(auditLog).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/integrations/status.ts b/src/lib/integrations/status.ts index 5b3f848f4..948881b9d 100644 --- a/src/lib/integrations/status.ts +++ b/src/lib/integrations/status.ts @@ -41,11 +41,52 @@ import { dispatchNotification } from "@/lib/notifications/dispatcher"; export type IntegrationKey = "withings" | "moodlog"; +/** + * Failure kinds carried into `recordSyncFailure`. + * + * - `transient` : retry on the next sync; user not blocked. + * - `reauth_required` : permanent revoke / invalid_grant; park at + * `error_reauth` until the user reconnects. + * - `persistent` : contract mismatch (invalid params, missing + * field, unknown action). Surfaces in the + * integration-status card AND audit log so an + * operator can investigate, but does NOT skip + * future sync attempts — those may succeed + * once the upstream side resolves. v1.4.43 + * W14: after 24h of unbroken persistent + * failures the row is `parked` and the + * integration stops attempting until the user + * / operator reconnects. + * + * v1.4.42 W6 extended this union from `transient | reauth_required` to + * the three-state taxonomy above; the state-mapping function turns + * `persistent` into `error_transient` for now (a Withings 293 still + * lets the next sync run), but the audit detail carries the explicit + * kind so operations can filter. + */ +export type FailureKind = "transient" | "reauth_required" | "persistent"; + +/** + * Recognised IntegrationStatus states. + * + * `parked` (v1.4.43 W14) is set when an integration's persistent + * failure streak has exceeded `PARK_PERSISTENT_FAILURE_AFTER_MS` (24h + * by default). A parked integration STOPS RETRYING — the next + * scheduled sync skips, and the Settings UI surfaces a "Paused — + * reconnect manually" pill with a "Wieder verbinden" CTA that POSTs + * to `/api/integrations/withings/resume` to clear the park. This is + * intentionally heavier than `error_transient`: a contract-mismatch + * that's been failing for a full day is no longer "the upstream might + * recover on its own" — it's an operator-shaped problem, and + * retrying every 15 minutes for another week just buries the audit + * trail. + */ export type IntegrationState = | "connected" | "error_transient" | "error_reauth" - | "disconnected"; + | "disconnected" + | "parked"; /** * The ladder at which a streak of failures escalates from "user-visible @@ -73,6 +114,16 @@ export function getPersistentFailureThreshold(): number { */ const ALERT_REPEAT_WINDOW_MS = 24 * 60 * 60 * 1000; +/** + * v1.4.43 W14 — once a `persistent` failure streak has been running + * for this long without a single intervening success, flip the + * integration to `parked`. 24h matches the same window the alert + * ladder uses for re-paging and gives the operator a full business + * day to notice a 293/294 surge before the integration disables + * itself. + */ +const PARK_PERSISTENT_FAILURE_AFTER_MS = 24 * 60 * 60 * 1000; + export interface IntegrationStatusSnapshot { integration: IntegrationKey; state: IntegrationState; @@ -80,6 +131,72 @@ export interface IntegrationStatusSnapshot { lastAttemptAt: string | null; lastError: string | null; consecutiveFailures: number; + /** v1.4.43 W14 — per-kind bucketed counters; `null` for rows + * that predate the migration (back-filled on next write). */ + consecutiveFailuresByKind: ConsecutiveFailuresByKind | null; +} + +/** + * Bucketed consecutive-failure counters keyed by `FailureKind`. + * v1.4.43 W14 — exposed so the response shape stays explicit and + * tests can pin the bucket increments. + */ +export type ConsecutiveFailuresByKind = Record; + +/** + * Type guard for the JSON payload Prisma returns for the + * `consecutiveFailuresByKind` column. Anything that's not a plain + * object with three numeric keys is treated as "no value yet" so the + * back-fill path runs on the next write. + */ +function isFailureBucketObject( + value: unknown, +): value is ConsecutiveFailuresByKind { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return false; + } + const obj = value as Record; + return ( + typeof obj.transient === "number" && + typeof obj.reauth_required === "number" && + typeof obj.persistent === "number" + ); +} + +/** + * Read the JSON `consecutiveFailuresByKind` column off the Prisma row + * with a typed view. Returns `null` when the row predates the + * migration so callers can back-fill from the legacy + * `consecutiveFailures` integer. + */ +function readBucketColumn(value: unknown): ConsecutiveFailuresByKind | null { + return isFailureBucketObject(value) ? value : null; +} + +/** + * Zero-bucket envelope. Inlined where we need the literal so eslint + * doesn't flag a re-assignment of the shared constant. + */ +function zeroBuckets(): ConsecutiveFailuresByKind { + return { transient: 0, reauth_required: 0, persistent: 0 }; +} + +/** + * Back-fill the per-kind buckets for a legacy row that predates the + * v1.4.43 W14 migration. We bucket the existing + * `consecutiveFailures` count into the kind matching the CURRENT + * failure (so a row sitting at `error_reauth` with 4 failures + * back-fills into `reauth_required: 4`). Subsequent failures of a + * different kind then increment their own bucket without losing the + * legacy count. + */ +function backfillBuckets( + legacyCount: number, + currentKind: FailureKind, +): ConsecutiveFailuresByKind { + const buckets = zeroBuckets(); + buckets[currentKind] = Math.max(0, legacyCount); + return buckets; } /** @@ -102,6 +219,7 @@ export async function getIntegrationStatus( lastAttemptAt: null, lastError: null, consecutiveFailures: 0, + consecutiveFailuresByKind: null, }; } return { @@ -111,12 +229,20 @@ export async function getIntegrationStatus( lastAttemptAt: row.lastAttemptAt?.toISOString() ?? null, lastError: row.lastError ? safeDecryptError(row.lastError) : null, consecutiveFailures: row.consecutiveFailures, + consecutiveFailuresByKind: readBucketColumn(row.consecutiveFailuresByKind), }; } /** * Read the current `state` cheaply — used by the sync entry-points * to short-circuit when reauth is required. + * + * v1.4.43 W14 — `parked` is treated as "reauth required" for the + * purpose of sync short-circuit: the user has to call `/resume` (or + * complete the OAuth flow again) before any further sync work + * happens. Returning `true` for both states means existing call sites + * keep their current "skip the cron tick" behaviour without churn, + * and the Settings UI distinguishes the two via the pill copy. */ export async function isReauthRequired( userId: string, @@ -126,7 +252,7 @@ export async function isReauthRequired( where: { userId_integration: { userId, integration } }, select: { state: true }, }); - return row?.state === "error_reauth"; + return row?.state === "error_reauth" || row?.state === "parked"; } /** @@ -141,6 +267,10 @@ export async function recordSyncSuccess( integration: IntegrationKey, ): Promise { const now = new Date(); + // v1.4.43 W14 — a success resets ALL per-kind buckets back to zero + // and clears the persistent-streak start timestamp. The legacy + // single-column `consecutiveFailures` zero stays for backward + // compatibility with v1.4.43 readers that haven't migrated yet. await prisma.integrationStatus.upsert({ where: { userId_integration: { userId, integration } }, create: { @@ -150,6 +280,7 @@ export async function recordSyncSuccess( lastSuccessAt: now, lastAttemptAt: now, consecutiveFailures: 0, + consecutiveFailuresByKind: zeroBuckets(), }, update: { state: "connected", @@ -157,32 +288,13 @@ export async function recordSyncSuccess( lastAttemptAt: now, lastError: null, consecutiveFailures: 0, + consecutiveFailuresByKind: zeroBuckets(), + persistentFailureStartedAt: null, alertedAt: null, }, }); } -/** - * Failure kinds carried into `recordSyncFailure`. - * - * - `transient` : retry on the next sync; user not blocked. - * - `reauth_required` : permanent revoke / invalid_grant; park at - * `error_reauth` until the user reconnects. - * - `persistent` : contract mismatch (invalid params, missing - * field, unknown action). Surfaces in the - * integration-status card AND audit log so an - * operator can investigate, but does NOT skip - * future sync attempts — those may succeed once - * the upstream side resolves. - * - * v1.4.42 W6 extended this union from `transient | reauth_required` to - * the three-state taxonomy above; the state-mapping function turns - * `persistent` into `error_transient` for now (a Withings 293 still - * lets the next sync run), but the audit detail carries the explicit - * kind so operations can filter. - */ -export type FailureKind = "transient" | "reauth_required" | "persistent"; - export interface RecordSyncFailureInput { userId: string; integration: IntegrationKey; @@ -194,36 +306,131 @@ export interface RecordSyncFailureInput { /** * Record a sync failure. Always: - * - increments `consecutive_failures` + * - increments the per-kind bucket in `consecutiveFailuresByKind` + * (and the legacy `consecutive_failures` integer for one release + * as a fallback for un-migrated readers) * - persists the encrypted error message * - writes one `AuditLog` row with `integrations.sync.failed` * * If `kind === "reauth_required"` the row is parked at `error_reauth` - * so the next scheduled sync skips. Otherwise the state is marked - * `error_transient` and the next sync will try again. + * so the next scheduled sync skips. If `kind === "persistent"` and + * the persistent streak has been running for >24h, the row is parked + * at `parked` so the next sync also skips — the operator / user has + * to call `resumeIntegrationFromPark` to clear it. Otherwise the state + * is `error_transient` and the next sync will try again. * - * If the post-update `consecutive_failures` count crosses the alerting - * threshold AND the alerting window has lapsed, dispatch a Telegram - * notification to all admins. Failures here are best-effort: a failed - * dispatch DOES NOT swallow the audit log. + * If the post-update bucket max crosses the alerting threshold AND the + * alerting window has lapsed, dispatch a Telegram notification to all + * admins. Failures here are best-effort: a failed dispatch DOES NOT + * swallow the audit log. + * + * v1.4.43 W14 — per-kind counter migration. Each failure increments + * ONLY its own bucket; a transient hiccup followed by a persistent + * failure no longer masks the persistent streak's true age. The + * legacy single-column `consecutiveFailures` integer continues to be + * incremented so the v1.4.43 alert ladder + UI consumers keep working + * unchanged; v1.4.44 will drop it once every reader migrates. */ export async function recordSyncFailure( input: RecordSyncFailureInput, ): Promise { const { userId, integration, kind, message, errorCode } = input; const now = new Date(); + const encryptedError = safeEncryptError(message); + + // Read the current row first so we can: + // (a) compute the new per-kind bucket value + // (b) decide whether the persistent streak has exceeded the + // park threshold + // (c) back-fill `consecutiveFailuresByKind` for rows that + // predate the v1.4.43 W14 migration. + const existing = await prisma.integrationStatus.findUnique({ + where: { userId_integration: { userId, integration } }, + select: { + consecutiveFailures: true, + consecutiveFailuresByKind: true, + persistentFailureStartedAt: true, + alertedAt: true, + }, + }); + + // Resolve the starting bucket envelope: + // - existing JSON value if present (post-migration row) + // - back-fill from the legacy integer if the row predates the + // migration (counts go into the CURRENT failure's bucket so the + // persistent streak is dated from this write, not from zero) + // - zero envelope for the first-ever write + const startingBuckets = readBucketColumn(existing?.consecutiveFailuresByKind); + + // Snapshot the persistent-bucket count BEFORE we increment so the + // streak-anchor decision below can see the pre-increment value. + // Otherwise we'd inspect the freshly-incremented bucket and never + // recognise a "first persistent failure of a fresh streak". + const persistentStreakBefore = startingBuckets?.persistent ?? 0; + + // Now build the new bucket envelope. `buckets` is always a fresh + // object so the upsert payload doesn't share a reference with the + // existing-row snapshot (which would couple the in-memory mutation + // to the audit-log read path below). + const buckets: ConsecutiveFailuresByKind = startingBuckets + ? { ...startingBuckets } + : existing + ? backfillBuckets(existing.consecutiveFailures, kind) + : zeroBuckets(); + + // Increment only the bucket matching this failure's kind. + // The other two buckets stay at their current value so a persistent + // streak isn't reset by an intervening transient hiccup. + if (startingBuckets) { + // Post-migration row — increment the actual bucket. + buckets[kind] = (buckets[kind] ?? 0) + 1; + } else if (!existing) { + // First-ever write — bucket starts at zero, this failure makes 1. + buckets[kind] = 1; + } + // else: back-fill path already seeded `buckets[kind]` with the legacy + // count; the current failure IS counted in that seed so no second + // increment is needed (a row at 4 failures sitting in error_reauth + // back-fills to reauth_required: 4 and this 5th failure makes the + // legacy +1 path bring the bucket to 5 below). + + // Track the persistent-streak start so the >24h park check has a + // wall-clock anchor. Only stamped on the FIRST persistent failure of + // a streak; cleared on success or when the persistent bucket goes + // back to zero (which today only happens via success, but the logic + // is symmetric for future "transient drained the streak" rules). + const isPersistent = kind === "persistent"; + let persistentFailureStartedAt: Date | null = + existing?.persistentFailureStartedAt ?? null; + if (isPersistent && persistentStreakBefore === 0) { + persistentFailureStartedAt = now; + } + + // Park decision: a persistent failure whose streak has exceeded the + // 24h window flips the state to `parked`. This is sticky — once + // parked, the row stays parked until either a success arrives + // (unlikely, since the sync entry-point short-circuits) or the + // user calls `resumeIntegrationFromPark` via the API. + const persistentStreakAgeMs = + isPersistent && persistentFailureStartedAt + ? now.getTime() - persistentFailureStartedAt.getTime() + : 0; + const shouldPark = + isPersistent && persistentStreakAgeMs > PARK_PERSISTENT_FAILURE_AFTER_MS; + // State mapping: // reauth_required → error_reauth (sync entry-point short-circuits) // transient → error_transient (next sync still runs) - // persistent → error_transient (next sync still runs, but the - // audit detail carries `kind: "persistent"` so - // operations can grep for contract-bug bursts) - const newState: IntegrationState = - kind === "reauth_required" ? "error_reauth" : "error_transient"; - - // Encrypt the error before persisting. crypto.encrypt() handles all - // edge cases (empty string, etc.) — we just have to give it a string. - const encryptedError = safeEncryptError(message); + // persistent → error_transient unless we just crossed the + // 24h park threshold, in which case → parked + // (sync entry-point short-circuits, audit detail + // carries the explicit kind so operations can + // grep for contract-bug bursts) + const newState: IntegrationState = shouldPark + ? "parked" + : kind === "reauth_required" + ? "error_reauth" + : "error_transient"; const row = await prisma.integrationStatus.upsert({ where: { userId_integration: { userId, integration } }, @@ -234,12 +441,16 @@ export async function recordSyncFailure( lastAttemptAt: now, lastError: encryptedError, consecutiveFailures: 1, + consecutiveFailuresByKind: buckets, + persistentFailureStartedAt: isPersistent ? now : null, }, update: { state: newState, lastAttemptAt: now, lastError: encryptedError, consecutiveFailures: { increment: 1 }, + consecutiveFailuresByKind: buckets, + persistentFailureStartedAt, }, }); @@ -255,6 +466,7 @@ export async function recordSyncFailure( errorCode: errorCode ?? null, message, attemptNumber: row.consecutiveFailures, + bucketCount: buckets[kind], state: newState, }, }); @@ -266,8 +478,19 @@ export async function recordSyncFailure( // Both conditions matter: (1) prevents premature paging, (2) // prevents loops where a flapping integration that fails once an // hour pages every hour. + // + // v1.4.43 W14 — the threshold check reads `Math.max(...buckets)` so + // a row with a 3-deep persistent streak still pages even when the + // transient bucket sat at 0 (the legacy integer was the running + // total and would have paged at the same moment). const threshold = getPersistentFailureThreshold(); - if (row.consecutiveFailures >= threshold) { + const alertSignal = Math.max( + row.consecutiveFailures, + buckets.transient, + buckets.reauth_required, + buckets.persistent, + ); + if (alertSignal >= threshold) { const previouslyAlerted = row.alertedAt && now.getTime() - row.alertedAt.getTime() < ALERT_REPEAT_WINDOW_MS; @@ -278,7 +501,7 @@ export async function recordSyncFailure( kind, message, errorCode, - consecutiveFailures: row.consecutiveFailures, + consecutiveFailures: alertSignal, }).catch((err) => { getEvent()?.addWarning( `Admin alert dispatch failed: ${err instanceof Error ? err.message : String(err)}`, @@ -293,6 +516,76 @@ export async function recordSyncFailure( }); } } + + // Park-event audit row — written once per transition into the + // `parked` state so the operations trail shows "this integration + // disabled itself after 24h of persistent failures" without the + // operator having to correlate the sync.failed rows by timestamp. + if (shouldPark && existing?.persistentFailureStartedAt) { + await auditLog("integrations.parked", { + userId, + details: { + integration, + reason: "persistent_24h", + persistentFailureStartedAt: + existing.persistentFailureStartedAt.toISOString(), + persistentStreakAgeMs, + errorCode: errorCode ?? null, + message, + }, + }); + } +} + +/** + * v1.4.43 W14 — clear a `parked` integration so the next scheduled + * sync runs again. Used by `/api/integrations/withings/resume` and + * the OAuth-callback path. The state moves to `connected`; all + * per-kind buckets reset to zero; the persistent-streak anchor + * clears; the alert window resets so the next genuine 3-strike burst + * still pages admins. + * + * Idempotent: calling against a connected row is a no-op (same + * post-state, no audit row). Calling against any non-parked error + * state also clears the row — the resume CTA is the universal + * "unstick this integration" button. + */ +export async function resumeIntegrationFromPark( + userId: string, + integration: IntegrationKey, +): Promise<{ wasParked: boolean }> { + const existing = await prisma.integrationStatus.findUnique({ + where: { userId_integration: { userId, integration } }, + select: { state: true }, + }); + const wasParked = existing?.state === "parked"; + + await prisma.integrationStatus.upsert({ + where: { userId_integration: { userId, integration } }, + create: { + userId, + integration, + state: "connected", + consecutiveFailuresByKind: zeroBuckets(), + }, + update: { + state: "connected", + lastError: null, + consecutiveFailures: 0, + consecutiveFailuresByKind: zeroBuckets(), + persistentFailureStartedAt: null, + alertedAt: null, + }, + }); + + if (wasParked) { + await auditLog("integrations.resumed", { + userId, + details: { integration, source: "user_resume" }, + }); + } + + return { wasParked }; } /** @@ -417,11 +710,14 @@ export async function markDisconnected( userId, integration, state: "disconnected", + consecutiveFailuresByKind: zeroBuckets(), }, update: { state: "disconnected", lastError: null, consecutiveFailures: 0, + consecutiveFailuresByKind: zeroBuckets(), + persistentFailureStartedAt: null, alertedAt: null, }, }); @@ -442,11 +738,14 @@ export async function markReconnected( userId, integration, state: "connected", + consecutiveFailuresByKind: zeroBuckets(), }, update: { state: "connected", lastError: null, consecutiveFailures: 0, + consecutiveFailuresByKind: zeroBuckets(), + persistentFailureStartedAt: null, alertedAt: null, }, });