feat: on-chain refunds, plan_changed event, webhook circuit breaker, …

backend/src/lib/webhookRegistry.ts

@@ -43,6 +43,70 @@ let retryTimerHandle: NodeJS.Timeout | null = null;
 let isShuttingDown = false;
 let db: any;
 
+// ── Circuit breaker ──────────────────────────────────────────────────────────
+// After CIRCUIT_FAILURE_THRESHOLD consecutive delivery failures to a given
+// URL, stop attempting deliveries to it for CIRCUIT_COOLDOWN_MS. This avoids
+// hammering an endpoint that's known to be down and lets it recover.
+const CIRCUIT_FAILURE_THRESHOLD = 10;
+const CIRCUIT_COOLDOWN_MS = 5 * 60 * 1000;
+
+const consecutiveFailures = new Map<string, number>();
+const circuitOpenUntil = new Map<string, number>();
+
+/** True if the circuit for `url` is currently open (deliveries paused). */
+function isCircuitOpen(url: string): boolean {
+  const openUntil = circuitOpenUntil.get(url);
+  if (openUntil === undefined) return false;
+  if (Date.now() >= openUntil) {
+    // Cooldown elapsed — close the circuit and give the endpoint a fresh start.
+    circuitOpenUntil.delete(url);
+    consecutiveFailures.set(url, 0);
+    return false;
+  }
+  return true;
+}
+
+function recordDeliverySuccess(url: string): void {
+  consecutiveFailures.set(url, 0);
+  circuitOpenUntil.delete(url);
+}
+
+function recordDeliveryFailure(url: string): void {
+  const count = (consecutiveFailures.get(url) ?? 0) + 1;
+  consecutiveFailures.set(url, count);
+  if (count >= CIRCUIT_FAILURE_THRESHOLD && !circuitOpenUntil.has(url)) {
+    const openUntil = Date.now() + CIRCUIT_COOLDOWN_MS;
+    circuitOpenUntil.set(url, openUntil);
+    logger.error("Webhook circuit breaker opened after consecutive failures", {
+      url,
+      consecutiveFailures: count,
+      cooldownMs: CIRCUIT_COOLDOWN_MS,
+      resumesAt: new Date(openUntil).toISOString(),
+    });
+  }
+}
+
+/** Exposed for observability/tests. */
+export function getCircuitBreakerState(url: string): {
+  open: boolean;
+  consecutiveFailures: number;
+  openUntil: string | null;
+} {
+  return {
+    open: isCircuitOpen(url),
+    consecutiveFailures: consecutiveFailures.get(url) ?? 0,
+    openUntil: circuitOpenUntil.has(url)
+      ? new Date(circuitOpenUntil.get(url)!).toISOString()
+      : null,
+  };
+}
+
+/** Reset all circuit breaker state. Exposed for tests. */
+export function resetCircuitBreakers(): void {
+  consecutiveFailures.clear();
+  circuitOpenUntil.clear();
+}
+
 function openDatabase() {
   fs.mkdirSync(path.dirname(DB_PATH), { recursive: true });
   const database = new Database(DB_PATH);
@@ -172,6 +236,17 @@ async function fireWebhookInternal(
 ): Promise<void> {
   // Fall back to the current async context's request ID when not explicitly supplied
   const effectiveCorrelationId = correlationId ?? getReqId();
+
+  if (isCircuitOpen(url)) {
+    logger.warn("Webhook circuit breaker open, skipping delivery", {
+      url,
+      attempt: attempt + 1,
+      attemptedAt: new Date().toISOString(),
+      correlationId: effectiveCorrelationId,
+    });
+    return;
+  }
+
   const headers: Record<string, string> = { "Content-Type": "application/json" };
   if (effectiveCorrelationId) {
     headers["X-Request-ID"] = effectiveCorrelationId;
@@ -205,18 +280,22 @@ async function fireWebhookInternal(
     }
 
     succeeded = true;
+    recordDeliverySuccess(url);
     webhookDeliveries.inc({ status: "success", attempt: String(attempt + 1) });
 
     if (attempt > 0) {
       logger.info("Webhook delivery succeeded after retry", { url, attempt, correlationId: effectiveCorrelationId });
     }
   } catch (err: any) {
     deliveryError = err.message;
+    recordDeliveryFailure(url);
     webhookDeliveries.inc({ status: "failure", attempt: String(attempt + 1) });
     logger.warn("Webhook delivery failed", {
       url,
       attempt: attempt + 1,
       maxRetries: MAX_RETRIES,
+      attemptedAt,
+      httpStatus,
       error: err.message,
       correlationId: effectiveCorrelationId,
     });
@@ -232,6 +311,7 @@ async function fireWebhookInternal(
       logger.error("Webhook delivery failed permanently after max retries", {
         url,
         attempts: MAX_RETRIES + 1,
+        httpStatus,
         correlationId: effectiveCorrelationId,
       });
     }

backend/tests/webhookRegistry.integration.test.ts

@@ -0,0 +1,116 @@
+/**
+ * Covers the webhook circuit breaker: after CIRCUIT_FAILURE_THRESHOLD (10)
+ * consecutive delivery failures to a URL, further deliveries to that URL are
+ * skipped (no fetch attempt) until the cooldown window elapses.
+ */
+
+import { describe, it, expect, beforeEach } from "vitest";
+
+process.env.WEBHOOKS_DB_PATH = ":memory:";
+
+import {
+  fireWebhook,
+  getCircuitBreakerState,
+  resetCircuitBreakers,
+} from "../src/lib/webhookRegistry.js";
+
+function installFetchSpy(status: number) {
+  const calls: string[] = [];
+  const original = globalThis.fetch;
+  globalThis.fetch = ((input: string | URL | Request) => {
+    calls.push(String(input));
+    return Promise.resolve(new Response(status === 200 ? "{}" : "error", { status }));
+  }) as typeof fetch;
+  return { calls, restore: () => { globalThis.fetch = original; } };
+}
+
+const URL = "https://circuit.example.com/hook";
+
+describe("webhook circuit breaker", () => {
+  beforeEach(() => {
+    resetCircuitBreakers();
+  });
+
+  it("opens after 10 consecutive failures and skips further delivery attempts", async () => {
+    const spy = installFetchSpy(500);
+    try {
+      for (let i = 0; i < 10; i++) {
+        await fireWebhook(URL, JSON.stringify({ i }));
+      }
+
+      const state = getCircuitBreakerState(URL);
+      expect(state.open).toBe(true);
+      expect(state.consecutiveFailures).toBeGreaterThanOrEqual(10);
+
+      const callsBeforeEleventh = spy.calls.length;
+      await fireWebhook(URL, JSON.stringify({ eleventh: true }));
+
+      // Circuit is open — no additional fetch attempt should have been made.
+      expect(spy.calls.length).toBe(callsBeforeEleventh);
+    } finally {
+      spy.restore();
+    }
+  });
+
+  it("does not open the circuit for a URL that stays under the failure threshold", async () => {
+    const spy = installFetchSpy(500);
+    try {
+      for (let i = 0; i < 9; i++) {
+        await fireWebhook(URL, JSON.stringify({ i }));
+      }
+      expect(getCircuitBreakerState(URL).open).toBe(false);
+    } finally {
+      spy.restore();
+    }
+  });
+
+  it("closes and resumes deliveries once the cooldown window elapses", async () => {
+    const failing = installFetchSpy(500);
+    try {
+      for (let i = 0; i < 10; i++) {
+        await fireWebhook(URL, JSON.stringify({ i }));
+      }
+      expect(getCircuitBreakerState(URL).open).toBe(true);
+    } finally {
+      failing.restore();
+    }
+
+    const realDateNow = Date.now;
+    try {
+      // Fast-forward past the 5-minute cooldown.
+      Date.now = () => realDateNow() + 5 * 60 * 1000 + 1;
+
+      const succeeding = installFetchSpy(200);
+      try {
+        await fireWebhook(URL, JSON.stringify({ resumed: true }));
+        expect(succeeding.calls.length).toBe(1);
+        expect(getCircuitBreakerState(URL).open).toBe(false);
+        expect(getCircuitBreakerState(URL).consecutiveFailures).toBe(0);
+      } finally {
+        succeeding.restore();
+      }
+    } finally {
+      Date.now = realDateNow;
+    }
+  });
+
+  it("a success resets the consecutive-failure count", async () => {
+    const failing = installFetchSpy(500);
+    try {
+      for (let i = 0; i < 5; i++) {
+        await fireWebhook(URL, JSON.stringify({ i }));
+      }
+      expect(getCircuitBreakerState(URL).consecutiveFailures).toBe(5);
+    } finally {
+      failing.restore();
+    }
+
+    const succeeding = installFetchSpy(200);
+    try {
+      await fireWebhook(URL, JSON.stringify({ ok: true }));
+      expect(getCircuitBreakerState(URL).consecutiveFailures).toBe(0);
+    } finally {
+      succeeding.restore();
+    }
+  });
+});