feat(webhook): signed delivery with retries and dead-letter queue

src/index.ts

@@ -6,6 +6,7 @@ import helmet from "helmet";
 import { logger } from "./logger";
 import { openApiSpec } from "./openapi";
 import { isSafeWebhookUrl } from "./utils/webhookUrl";
+import { generateWebhookSecret, deliverEventToWebhooks, webhookDlq } from "./webhook/delivery";
 import { resolveClientIp } from "./utils/clientIp";
 import { getStoreAdapter } from "./persistence";
 import {
@@ -23,6 +24,7 @@ import {
   pairKey,
   defaultMeta,
   recordEvent,
+  setWebhookDeliveryHook,
   trimEventLog,
   EVENT_LOG_CAP,
   EVENT_LOG_CAP_MAX,
@@ -131,6 +133,10 @@ const EVENT_PAYLOAD_MAX_ARRAY_ITEMS = 32;
 const EVENT_PAYLOAD_MAX_DEPTH = 3;
 
 const app = express();
+// Register webhook delivery hook so every recorded event is delivered to subscribers.
+setWebhookDeliveryHook((event) => {
+  deliverEventToWebhooks(event, webhookStore).catch(() => {});
+});
 
 // --- Persistence Hydration on startup ---
 export const hydrationPromise = (async () => {
@@ -1997,10 +2003,11 @@ app.post(
     const deduped = validateWebhookEvents(res, req, events);
     if (deduped === null) return;
     const id = `wh_${randomUUID().replace(/-/g, "").slice(0, 16)}`;
-    webhookStore.set(id, { url, events: deduped, createdAt: Date.now() });
+    const secret = generateWebhookSecret();
+    webhookStore.set(id, { url, events: deduped, createdAt: Date.now(), secret });
     // Record id and url only — never any webhook secret material.
     recordEvent("webhook.created", { id, url });
-    res.status(201).json({ id, url, events: deduped });
+    res.status(201).json({ id, url, events: deduped, secret });
   },
 );
 
@@ -2117,6 +2124,47 @@ app.patch("/api/v1/webhooks/:id", (req: Request, res: Response) => {
   res.json({ id, ...updated });
 });
 
+/**
+ * List dead-lettered webhook deliveries.
+ *
+ * @route GET /api/v1/webhooks/dlq
+ */
+app.get("/api/v1/webhooks/dlq", (req: Request, res: Response) => {
+  const query = (req.query ?? {}) as Record<string, unknown>;
+  const limit = Math.min(
+    Math.max(1, Number(query["limit"] ?? 20)),
+    100,
+  );
+  const entries = Array.from(webhookDlq.values())
+    .sort((a, b) => b.deadAt - a.deadAt)
+    .slice(0, limit);
+  res.json({ entries, total: webhookDlq.size });
+});
+
+/**
+ * Replay a dead-lettered webhook delivery.
+ *
+ * @route POST /api/v1/webhooks/dlq/:id/replay
+ */
+app.post("/api/v1/webhooks/dlq/:id/replay", (req: Request, res: Response) => {
+  const id = req.params.id ?? "";
+  const entry = webhookDlq.get(id);
+  if (!entry) {
+    sendError(res, req, 404, "not_found", `dlq entry ${id} not found`);
+    return;
+  }
+  const record = webhookStore.get(entry.webhookId);
+  if (!record) {
+    sendError(res, req, 404, "not_found", `webhook ${entry.webhookId} not found`);
+    return;
+  }
+  webhookDlq.delete(id);
+  deliverEventToWebhooks(entry.event, new Map([[entry.webhookId, record]])).catch(
+    () => {},
+  );
+  res.json({ replayed: true, id });
+});
+
 /**
  * Normalize the `:source`/`:destination` route params to their canonical asset
  * codes via {@link normalizeAsset}. On invalid input a `400 invalid_request` is

src/stores.ts

@@ -156,6 +156,7 @@ export const verifyApiKeySecret = (
 
 /** Record stored for each registered webhook. */
 export type WebhookRecord = {
+  secret: string;
   url: string;
   events: string[];
   createdAt: number;
@@ -300,6 +301,14 @@ export const trimEventLog = (cap: number): void => {
  *   enforces this at the call site so stray string literals are caught at compile time.
  * @param payload - Arbitrary structured data attached to the event.
  */
+/** Optional hook for delivering recorded events to webhook subscribers. */
+let webhookDeliveryHook: ((event: AppEvent) => void) | undefined;
+
+/** Register a callback that fires after every event is recorded. */
+export function setWebhookDeliveryHook(hook: (event: AppEvent) => void): void {
+  webhookDeliveryHook = hook;
+}
+
 export const recordEvent = (
   type: EventType,
   payload: Record<string, unknown>,
@@ -313,6 +322,13 @@ export const recordEvent = (
   eventLog.push(event);
   const cap = effectiveEventLogCap();
   if (eventLog.length > cap) eventLog.shift();
+  if (webhookDeliveryHook) {
+    try {
+      webhookDeliveryHook(event);
+    } catch {
+      // Delivery errors must not break event recording.
+    }
+  }
   return event;
 };
 
@@ -480,7 +496,12 @@ export const hydrateFromSnapshot = (snapshot: unknown): void => {
             item.length === 2 &&
             typeof item[0] === "string"
           ) {
-            webhookStore.set(item[0], item[1] as WebhookRecord);
+            const record = item[1] as WebhookRecord;
+            // Backfill secret for webhooks created before the signing feature.
+            if (!record.secret) {
+              record.secret = randomBytes(32).toString("base64url");
+            }
+            webhookStore.set(item[0], record);
           }
         }
       }

src/webhook/delivery.test.ts

@@ -0,0 +1,120 @@
+import { describe, it, expect, vi } from "vitest";
+import {
+  generateWebhookSecret,
+  signWebhookPayload,
+  verifyWebhookPayload,
+  webhookDlq,
+  deliverEventToWebhooks,
+  WEBHOOK_MAX_RETRIES,
+} from "./delivery";
+import type { AppEvent, WebhookRecord } from "../stores";
+
+describe("webhook delivery", () => {
+  it("generates a URL-safe secret", () => {
+    const secret = generateWebhookSecret();
+    expect(secret).toMatch(/^[A-Za-z0-9_-]+$/);
+    expect(secret.length).toBeGreaterThanOrEqual(40);
+  });
+
+  it("signs and verifies a payload", () => {
+    const secret = generateWebhookSecret();
+    const payload = JSON.stringify({ event: "test" });
+    const { signature, timestamp } = signWebhookPayload(secret, payload);
+    expect(signature).toMatch(/^t=\d+,v0=[0-9a-f]{64}$/);
+    expect(verifyWebhookPayload(secret, payload, signature)).toBe(true);
+  });
+
+  it("rejects a tampered signature", () => {
+    const secret = generateWebhookSecret();
+    const payload = JSON.stringify({ event: "test" });
+    const { signature } = signWebhookPayload(secret, payload);
+    expect(verifyWebhookPayload(secret, payload + "x", signature)).toBe(false);
+  });
+
+  it("rejects an expired signature", () => {
+    const secret = generateWebhookSecret();
+    const payload = JSON.stringify({ event: "test" });
+    const oldTs = Math.floor(Date.now() / 1000) - 400;
+    const { signature } = signWebhookPayload(secret, payload, oldTs);
+    expect(verifyWebhookPayload(secret, payload, signature)).toBe(false);
+  });
+
+  it("delivers to matching webhooks", async () => {
+    webhookDlq.clear();
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true, status: 200 });
+    const event: AppEvent = {
+      id: "evt-1",
+      ts: Date.now(),
+      type: "pair.registered",
+      payload: { source: "XLM", destination: "USDC" },
+    };
+    const webhooks = new Map<string, WebhookRecord>([
+      [
+        "wh-1",
+        {
+          url: "https://example.com/hook",
+          events: ["pair.registered"],
+          createdAt: Date.now(),
+          secret: generateWebhookSecret(),
+        },
+      ],
+      [
+        "wh-2",
+        {
+          url: "https://example.com/other",
+          events: ["pair.disabled"],
+          createdAt: Date.now(),
+          secret: generateWebhookSecret(),
+        },
+      ],
+    ]);
+
+    await deliverEventToWebhooks(event, webhooks, fetchMock as any);
+    // Give async delivery a tick
+    await new Promise((r) => setTimeout(r, 10));
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(fetchMock).toHaveBeenCalledWith(
+      "https://example.com/hook",
+      expect.objectContaining({
+        method: "POST",
+        headers: expect.objectContaining({
+          "Content-Type": "application/json",
+          "X-Signature": expect.stringMatching(/^t=\d+,v0=[0-9a-f]{64}$/),
+        }),
+      }),
+    );
+  });
+
+  it("dead-letters after max retries on 5xx", async () => {
+    webhookDlq.clear();
+    const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 503 });
+    const event: AppEvent = {
+      id: "evt-2",
+      ts: Date.now(),
+      type: "pair.registered",
+      payload: {},
+    };
+    const webhooks = new Map<string, WebhookRecord>([
+      [
+        "wh-1",
+        {
+          url: "https://example.com/hook",
+          events: ["pair.registered"],
+          createdAt: Date.now(),
+          secret: generateWebhookSecret(),
+        },
+      ],
+    ]);
+
+    await deliverEventToWebhooks(event, webhooks, fetchMock as any);
+    // Wait for all retries + backoff
+    await new Promise((r) => setTimeout(r, 8000));
+
+    expect(fetchMock).toHaveBeenCalledTimes(WEBHOOK_MAX_RETRIES);
+    expect(webhookDlq.size).toBe(1);
+    const entry = Array.from(webhookDlq.values())[0];
+    expect(entry.webhookId).toBe("wh-1");
+    expect(entry.attempts).toBe(WEBHOOK_MAX_RETRIES);
+  });
+});