feat: implement indexer status SSE endpoint and robust webhook delive…

app/api/indexer/status/route.ts

@@ -13,11 +13,19 @@
 export const dynamic = "force-dynamic";
 export const runtime = "nodejs";
 
+import { isCircuitBreakerOpen } from "@/app/lib/admin-guard";
+
 interface IndexerStatus {
   ledgerCursor: number;
   lagMs: number;
   queueDepth: number;
   syncedAt: string;
+  /**
+   * True when an admin has tripped the indexer circuit breaker via
+   * POST /api/admin/circuit-breaker. Ingestion is halted; cursor and lag
+   * are frozen at their last values and will not advance until reset.
+   */
+  breakerOpen: boolean;
 }
 
 function getIndexerStatus(): IndexerStatus {
@@ -27,6 +35,7 @@ function getIndexerStatus(): IndexerStatus {
     lagMs: Math.floor(Math.random() * 3000),
     queueDepth: Math.floor(Math.random() * 50),
     syncedAt: new Date().toISOString(),
+    breakerOpen: isCircuitBreakerOpen("indexer"),
   };
 }
 

app/lib/webhook-circuit-breaker.test.ts

@@ -0,0 +1,184 @@
+/** @jest-environment node */
+/**
+ * Enforcement tests for the GLOBAL admin circuit breaker.
+ *
+ * These cover the wiring between `POST /api/admin/circuit-breaker` (control
+ * plane) and the webhook dispatch path (data plane). The admin breaker is
+ * distinct from the per-endpoint failure breaker in WebhookDeliveryClient —
+ * see the webhook-delivery module docblock. The critical behavioural
+ * difference asserted here: the admin breaker DEFERS deliveries, it does not
+ * DLQ them.
+ */
+import { WebhookDeliveryClient, WebhookEndpoint, WebhookEvent } from "./webhook-delivery";
+import { WebhookDeliveryWorker } from "./webhook-delivery-worker";
+import { webhookDeliveryStore } from "./webhook-delivery-store";
+import { webhookOutboxStore } from "./webhook-outbox";
+import {
+  setCircuitBreaker,
+  isCircuitBreakerOpen,
+  _resetAdminStateForTesting,
+} from "./admin-guard";
+
+const ADMIN_ADDRESS = "GADMIN_TEST_ADDRESS_12345";
+
+/** Admin-authenticated request used to drive the breaker in tests. */
+function adminRequest(): Request {
+  return new Request("http://localhost/api/admin/circuit-breaker", {
+    method: "POST",
+    headers: { "Actor-Wallet-Address": ADMIN_ADDRESS },
+  });
+}
+
+/** Trip or reset the global breaker through the real admin-guard entry point. */
+function toggleBreaker(target: "indexer" | "webhook", open: boolean): void {
+  const result = setCircuitBreaker(adminRequest(), target, open);
+  if (!("circuitBreakers" in result)) {
+    throw new Error("Failed to toggle breaker — admin auth rejected in test setup");
+  }
+}
+
+const endpoint: WebhookEndpoint = {
+  id: "ep-breaker-test",
+  url: "https://example.test/hook",
+  maxRetries: 3,
+};
+
+const event: WebhookEvent = {
+  id: "evt-1",
+  eventType: "stream.created",
+  streamId: "stream-1",
+  data: { amount: "100" },
+  timestamp: new Date().toISOString(),
+};
+
+const fetchMock = jest.fn();
+
+beforeEach(() => {
+  _resetAdminStateForTesting(ADMIN_ADDRESS);
+  // Both stores are module-level singletons — clear them so DLQ and outbox
+  // assertions cannot be polluted by a previous test.
+  webhookDeliveryStore.clear();
+  webhookOutboxStore.clear();
+  fetchMock.mockReset();
+  global.fetch = fetchMock as unknown as typeof fetch;
+});
+
+describe("admin breaker → attemptDelivery", () => {
+  it("does not contact the endpoint while the webhook breaker is open", async () => {
+    toggleBreaker("webhook", true);
+    const client = new WebhookDeliveryClient();
+
+    const result = await client.attemptDelivery(endpoint, event, "d-1", 1);
+
+    expect(result.success).toBe(false);
+    expect(result.deferred).toBe(true);
+    expect(result.shouldRetry).toBe(false);
+    expect(result.error).toMatch(/admin circuit breaker/i);
+    // The decisive assertion: no outbound HTTP happened at all.
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  it("delivers normally once the breaker is reset", async () => {
+    toggleBreaker("webhook", true);
+    toggleBreaker("webhook", false);
+    fetchMock.mockResolvedValue({ status: 200, statusText: "OK" });
+    const client = new WebhookDeliveryClient();
+
+    const result = await client.attemptDelivery(endpoint, event, "d-2", 1);
+
+    expect(result.success).toBe(true);
+    expect(result.deferred).toBeUndefined();
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("is unaffected by the indexer breaker (targets are independent)", async () => {
+    toggleBreaker("indexer", true);
+    fetchMock.mockResolvedValue({ status: 200, statusText: "OK" });
+    const client = new WebhookDeliveryClient();
+
+    const result = await client.attemptDelivery(endpoint, event, "d-3", 1);
+
+    expect(result.success).toBe(true);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not record an endpoint failure when deferring", async () => {
+    toggleBreaker("webhook", true);
+    const client = new WebhookDeliveryClient();
+
+    // Defer many times — well past the per-endpoint threshold of 5.
+    for (let i = 0; i < 10; i++) {
+      await client.attemptDelivery(endpoint, event, `d-loop-${i}`, 1);
+    }
+
+    // The endpoint is blameless; its own breaker must stay closed so that
+    // traffic resumes immediately when the admin resets the global breaker.
+    expect(client.isCircuitOpen(endpoint.id)).toBe(false);
+  });
+});
+
+describe("admin breaker → processDelivery", () => {
+  it("defers instead of moving the delivery to the DLQ", async () => {
+    toggleBreaker("webhook", true);
+    const worker = new WebhookDeliveryWorker({}, () => Promise.resolve());
+
+    const result = await worker.processDelivery(endpoint, event, "d-4");
+
+    expect(result.success).toBe(false);
+    expect(result.deferred).toBe(true);
+    expect(result.dlqed).toBeUndefined();
+
+    // Regression guard: an operator-initiated pause must never discard events.
+    const dlqIds = webhookDeliveryStore.getAllDLQEntries().map((e) => e.deliveryId);
+    expect(dlqIds).not.toContain("d-4");
+  });
+
+  it("still DLQs on a genuine non-retryable endpoint failure", async () => {
+    // Breaker closed — confirms the deferral path did not swallow real failures.
+    fetchMock.mockResolvedValue({ status: 400, statusText: "Bad Request" });
+    const worker = new WebhookDeliveryWorker({}, () => Promise.resolve());
+
+    const result = await worker.processDelivery(endpoint, event, "d-5");
+
+    expect(result.success).toBe(false);
+    expect(result.dlqed).toBe(true);
+    expect(result.deferred).toBeUndefined();
+  });
+});
+
+describe("admin breaker → drainOutbox", () => {
+  it("skips the drain entirely while the breaker is open", async () => {
+    webhookOutboxStore.addToOutbox(endpoint, event);
+    toggleBreaker("webhook", true);
+    const worker = new WebhookDeliveryWorker({}, () => Promise.resolve());
+
+    await worker.drainOutbox();
+
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  it("leaves deferred outbox entries retryable rather than failed", async () => {
+    const entry = webhookOutboxStore.addToOutbox(endpoint, event);
+    toggleBreaker("webhook", true);
+    const worker = new WebhookDeliveryWorker({}, () => Promise.resolve());
+
+    await worker.processOutboxEntry(entry);
+
+    // Back to pending, not "failed" or "dlq" — it must be picked up again
+    // by the next drain once the operator resets the breaker.
+    const entries = webhookOutboxStore.getPendingOutboxEntries(100);
+    expect(entries.map((e) => e.id)).toContain(entry.id);
+  });
+});
+
+describe("isCircuitBreakerOpen", () => {
+  it("reports per-target state independently", () => {
+    expect(isCircuitBreakerOpen("indexer")).toBe(false);
+    expect(isCircuitBreakerOpen("webhook")).toBe(false);
+
+    toggleBreaker("indexer", true);
+
+    expect(isCircuitBreakerOpen("indexer")).toBe(true);
+    expect(isCircuitBreakerOpen("webhook")).toBe(false);
+  });
+});

app/lib/webhook-delivery-worker.ts

@@ -11,6 +11,7 @@ import {
 } from "./webhook-delivery";
 import { webhookDeliveryStore } from "./webhook-delivery-store";
 import { webhookOutboxStore, WebhookOutboxEntry } from "./webhook-outbox";
+import { isCircuitBreakerOpen } from "./admin-guard";
 
 export class WebhookDeliveryWorker {
   private client: WebhookDeliveryClient;
@@ -49,7 +50,14 @@ export class WebhookDeliveryWorker {
     event: WebhookEvent,
     deliveryId: string,
     reissuedFrom?: string,
-  ): Promise<{ success: boolean; deliveryId: string; attempts: number; dlqed?: boolean }> {
+  ): Promise<{
+    success: boolean;
+    deliveryId: string;
+    attempts: number;
+    dlqed?: boolean;
+    /** Held by the global admin breaker — retryable once the breaker resets. */
+    deferred?: boolean;
+  }> {
     const ctx = getCorrelationContext();
     withWebhookContext(deliveryId);
     const maxAttempts = this.maxRetries;
@@ -69,6 +77,19 @@ export class WebhookDeliveryWorker {
           webhookDeliveryStore.getDelivery(deliveryId)?.attempts ?? [],
         );
 
+        // ── Global admin breaker → hold, do NOT DLQ ──────────────────────────
+        // The operator paused dispatch deliberately. Leave the delivery pending
+        // so it can be retried once the breaker is reset; DLQing here would
+        // discard every in-flight event for the duration of an incident.
+        if (result.deferred) {
+          logger.warn("Webhook delivery deferred — admin circuit breaker open", {
+            delivery_id: deliveryId, endpoint_id: endpoint.id,
+            event_id: event.id, breaker: "admin.webhook",
+            correlation_id: ctx?.correlation_id,
+          });
+          return { success: false, deliveryId, attempts: 0, deferred: true };
+        }
+
         if (result.error?.includes("Circuit breaker open")) {
           webhookDeliveryStore.moveToDLQ(deliveryId, result.error);
           return { success: false, deliveryId, attempts: 0, dlqed: true };
@@ -168,6 +189,13 @@ export class WebhookDeliveryWorker {
         status: "delivered",
         attempts: result.attempts,
       });
+    } else if (result.deferred) {
+      // Admin breaker is open — return the entry to the pending pool untouched
+      // so the next drain (after the breaker resets) picks it up again.
+      webhookOutboxStore.updateOutboxEntry(entry.id, {
+        status: "pending",
+        attempts: result.attempts,
+      });
     } else if (result.dlqed) {
       webhookOutboxStore.updateOutboxEntry(entry.id, {
         status: "dlq",
@@ -187,6 +215,17 @@ export class WebhookDeliveryWorker {
    * Drain the outbox: process all pending entries
    */
   async drainOutbox(limit: number = 100): Promise<void> {
+    // Skip the whole drain while the admin breaker is open. Without this the
+    // worker would churn every pending entry through processDelivery once per
+    // tick, only to defer each one back to pending.
+    if (isCircuitBreakerOpen("webhook")) {
+      logger.warn("Outbox drain skipped — admin circuit breaker open", {
+        breaker: "admin.webhook",
+        correlation_id: getCorrelationContext()?.correlation_id,
+      });
+      return;
+    }
+
     const entries = webhookOutboxStore.getPendingOutboxEntries(limit);
 
     logger.info("Draining webhook outbox", {

app/lib/webhook-delivery.ts

@@ -1,6 +1,7 @@
 import crypto from 'crypto';
 import { logger, withWebhookContext, getCorrelationContext } from './logger';
 import { getActiveSigningSecrets } from '@/app/lib/webhook-secrets';
+import { isCircuitBreakerOpen } from '@/app/lib/admin-guard';
 
 /**
  * Idempotent webhook delivery with exponential backoff + full jitter and DLQ support.
@@ -20,6 +21,21 @@ import { getActiveSigningSecrets } from '@/app/lib/webhook-secrets';
  * - 4xx (except 408, 429) → non-retryable client error, go straight to DLQ
  * - 408, 429, 5xx    → retryable server/transient error
  * - network timeout  → retryable
+ *
+ * ## Two distinct circuit breakers
+ * This module is subject to two independent breakers. They are NOT the same
+ * mechanism and must not be conflated:
+ *
+ * 1. **Global admin breaker** — `isCircuitBreakerOpen("webhook")` from
+ *    admin-guard. Manually toggled subsystem-wide via
+ *    `POST /api/admin/circuit-breaker`. Checked FIRST. When open, no endpoint
+ *    is contacted at all. Deliveries are held (`deferred: true`), NOT DLQed —
+ *    an operator pausing dispatch during an incident must not lose events.
+ *
+ * 2. **Per-endpoint failure breaker** — `WebhookDeliveryClient.isCircuitOpen`.
+ *    Trips itself after N consecutive failures to one endpoint and auto-resets
+ *    after 5 min. When open, that endpoint's delivery goes to DLQ, because the
+ *    endpoint itself is demonstrably broken.
  */
 
 // ── Types ─────────────────────────────────────────────────────────────────────
@@ -390,10 +406,32 @@ export class WebhookDeliveryClient {
     error?: string;
     shouldRetry: boolean;
     nextRetryAt?: string;
+    /**
+     * Set when the global admin breaker held this delivery. The caller must
+     * leave the delivery pending rather than DLQing it — see the module
+     * docblock on the two breakers.
+     */
+    deferred?: boolean;
   }> {
     const context = getCorrelationContext();
     withWebhookContext(deliveryId);
 
+    // ── Global admin breaker (checked before anything endpoint-specific) ─────
+    // Operator-controlled kill switch. Hold the delivery; do not DLQ it and do
+    // not record a failure against the endpoint — the endpoint is not at fault.
+    if (isCircuitBreakerOpen('webhook')) {
+      const error = 'Webhook dispatch halted: admin circuit breaker is open';
+      logger.warn('Webhook delivery deferred by admin circuit breaker', {
+        delivery_id: deliveryId,
+        endpoint_id: endpoint.id,
+        endpoint_url: endpoint.url,
+        event_id: event.id,
+        breaker: 'admin.webhook',
+        correlation_id: context?.correlation_id,
+      });
+      return { success: false, error, shouldRetry: false, deferred: true };
+    }
+
     if (this.isCircuitOpen(endpoint.id, endpoint.circuitBreakerThreshold)) {
       const error = 'Circuit breaker open: endpoint experiencing repeated failures';
       logger.warn('Webhook delivery blocked by circuit breaker', {