feat: distributed multi-node webhook delivery engine with HMAC signing and DLQ recovery

apps/api/src/app.ts

@@ -49,6 +49,8 @@ import { swapDisputeRoutes } from "./routes/swap-dispute.js";
 import { SwapDisputeStore } from "./lib/swapDisputeStore.js";
 import { getChatInfrastructure } from "./lib/chat-infrastructure.js";
 import { juryArbitrationRoutes } from "./routes/jury-arbitration.js";
+import { webhookRoutes } from "./routes/webhooks.js";
+import { WebhookDeliveryStore } from "./lib/webhookDeliveryStore.js";
 
 const MAX_PAYMENTS_CACHE = 10000;
 const usedPayments = new Map<string, number>();
@@ -458,6 +460,15 @@ app.register(swapDisputeRoutes, {
   prefix: "/api/v1",
   store: new SwapDisputeStore(pgPool ?? null),
 });
+// Distributed Multi-Node Webhook Event Delivery Engine & DLQ Recovery
+// (#445): developer-registered endpoint registration, delivery-log lookup,
+// and dead-letter replay. Shares the pool so DLQ replay's SELECT ... FOR
+// UPDATE coordinates with webhookDeliveryWorker; degrades to an in-memory
+// store in dev like the routes above.
+app.register(webhookRoutes, {
+  prefix: "/api/v1",
+  store: new WebhookDeliveryStore(pgPool ?? null),
+});
 // (#404) Decentralized Jury Dispute Arbitration: commit-reveal voting,
 // VRF juror selection, and automated escrow resolution with stake slashing.
 app.register(juryArbitrationRoutes, { prefix: "/api/v1" });

apps/api/src/db/migrations/030_add_distributed_webhook_pipeline.sql

@@ -0,0 +1,52 @@
+BEGIN;
+
+-- Distributed Multi-Node Webhook Event Delivery Engine & DLQ Recovery (#445).
+--
+-- Velo's ops-alert webhook (lib/webhook.ts's sendWebhookAlert) posts to a
+-- single Slack/Discord URL and is fine to fire-and-forget. This is a
+-- different surface: developers register their own target URL to receive
+-- signed trade-status events (COMPLETED / REFUNDED). Sending those inline in
+-- the request thread means one slow or dead client endpoint blocks a real API
+-- response and, with no signature, lets a malicious third party spoof status
+-- callbacks against anyone who trusts them unsigned.
+--
+-- webhook_endpoints is one row per developer-registered destination, holding
+-- the HMAC secret used to sign every delivery to it. webhook_delivery_logs is
+-- one row per attempted delivery, carrying its own attempt count and status
+-- so a stuck delivery can be inspected and, once dead-lettered, replayed
+-- without re-deriving anything from the original trigger.
+
+CREATE TYPE webhook_delivery_status AS ENUM ('QUEUED', 'DELIVERED', 'FAILED', 'DEAD_LETTER');
+
+CREATE TABLE webhook_endpoints (
+    endpoint_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    user_id VARCHAR(64) NOT NULL,
+    target_url TEXT NOT NULL,
+    secret_key VARCHAR(64) NOT NULL,
+    is_active BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Lookup for "which active endpoints does this developer have" — the query
+-- the enqueue path runs on every trade-status event.
+CREATE INDEX idx_webhook_endpoints_user ON webhook_endpoints(user_id) WHERE is_active;
+
+CREATE TABLE webhook_delivery_logs (
+    delivery_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    endpoint_id UUID NOT NULL REFERENCES webhook_endpoints(endpoint_id) ON DELETE CASCADE,
+    event_type VARCHAR(64) NOT NULL,
+    payload JSONB NOT NULL,
+    signature_header VARCHAR(64) NOT NULL,
+    attempt_count INT NOT NULL DEFAULT 0,
+    status webhook_delivery_status NOT NULL DEFAULT 'QUEUED',
+    last_response_code INT NULL,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
+);
+
+-- The DLQ replay endpoint's hot query is "find this dead-lettered delivery to
+-- claim it"; the operator dashboard's is "list an endpoint's recent
+-- deliveries" — both covered by leading with endpoint_id.
+CREATE INDEX idx_webhook_delivery_endpoint ON webhook_delivery_logs(endpoint_id, created_at DESC);
+CREATE INDEX idx_webhook_delivery_status ON webhook_delivery_logs(status) WHERE status = 'DEAD_LETTER';
+
+COMMIT;

apps/api/src/index.ts

@@ -17,6 +17,8 @@ import { createEnterpriseStore } from "./lib/enterprise-store.js";
 import { startApprovalTimeoutWorker } from "./lib/workers/approvalTimeoutWorker.js";
 import { startCollateralCooldownWorker } from "./lib/workers/cooldownWorker.js";
 import { CollateralGuardStore } from "./lib/collateralGuard.js";
+import { startWebhookDeliveryWorker } from "./lib/workers/webhookDeliveryWorker.js";
+import { WebhookDeliveryStore } from "./lib/webhookDeliveryStore.js";
 
 const port = Number(process.env.PORT ?? 3000);
 
@@ -125,6 +127,34 @@ async function startServer() {
     } catch (error) {
       app.log.error(error, "approval timeout worker failed to start");
     }
+
+    // (#445) Distributed Multi-Node Webhook Event Delivery Engine: drains
+    // velo:webhook-delivery-queue, delivers signed developer-webhook
+    // payloads, and dead-letters deliveries that exhaust their retries.
+    // Needs both the delivery log (Postgres) and the queue (Redis).
+    if (process.env.REDIS_URL) {
+      try {
+        const deliveryQueueRedis = createClient({ url: process.env.REDIS_URL });
+        deliveryQueueRedis.on("error", (error) =>
+          app.log.error(error, "webhook delivery queue error"),
+        );
+        await deliveryQueueRedis.connect();
+        startWebhookDeliveryWorker({
+          store: new WebhookDeliveryStore(pgPool ?? null),
+          redis: deliveryQueueRedis,
+          onEvent: (event) => {
+            if (event.type === "dead-letter") {
+              app.log.warn(event, "webhook delivery dead-lettered");
+            }
+          },
+        });
+      } catch (error) {
+        // A Redis outage must not stop the API from serving requests.
+        app.log.error(error, "webhook delivery worker failed to start");
+      }
+    } else {
+      app.log.warn("REDIS_URL is not configured; webhook delivery worker is disabled");
+    }
   } catch (err) {
     app.log.error(err);
     process.exit(1);

apps/api/src/lib/webhook.ts

@@ -1,4 +1,8 @@
 import "dotenv/config";
+import { createHmac } from "node:crypto";
+import { createClient, type RedisClientType } from "redis";
+import { WEBHOOK_DELIVERY_QUEUE } from "@velo/shared";
+import type { WebhookDeliveryStore } from "./webhookDeliveryStore.js";
 
 const WEBHOOK_URL = process.env.REFUND_WEBHOOK_URL;
 
@@ -212,3 +216,86 @@ export async function sendSwapExpiryWarningAlert(params: {
     },
   });
 }
+
+/* ------------------------------------------------------------------ */
+/*  Distributed Multi-Node Webhook Event Delivery Engine (#445)        */
+/* ------------------------------------------------------------------ */
+//
+// Everything above this line is the single-destination ops alert (Slack /
+// Discord) fired inline and best-effort. This section is a different
+// surface: developers register their own target URL to receive signed
+// trade-status events. Sending those inline in the request thread means one
+// slow or dead client endpoint blocks a real API response — so these are
+// always enqueued onto a Redis Stream (`velo:webhook-delivery-queue`) for
+// `webhookDeliveryWorker` to actually deliver, never fetched here.
+
+/** HMAC-SHA256 over the exact JSON string sent as the request body. */
+export function signWebhookPayload(payloadJson: string, secretKey: string): string {
+  return createHmac("sha256", secretKey).update(payloadJson).digest("hex");
+}
+
+let deliveryQueueClient: RedisClientType | undefined;
+
+async function deliveryQueue(): Promise<RedisClientType | undefined> {
+  const url = process.env.REDIS_URL;
+  if (!url) return undefined;
+  if (!deliveryQueueClient) {
+    deliveryQueueClient = createClient({ url }) as RedisClientType;
+    deliveryQueueClient.on("error", (error) =>
+      console.error("Redis webhook delivery queue error", error),
+    );
+    await deliveryQueueClient.connect();
+  }
+  return deliveryQueueClient;
+}
+
+/**
+ * Signs and enqueues one webhook delivery for every active endpoint a
+ * developer has registered, so callers (e.g. cash.ts on refund/completion)
+ * just describe the event once regardless of how many endpoints exist.
+ *
+ * A missing REDIS_URL (local dev without Redis) degrades to a no-op with a
+ * warning rather than throwing — a notification outage must never fail the
+ * trade action that triggered it.
+ */
+export async function notifyDeveloperWebhooks(
+  store: WebhookDeliveryStore,
+  userId: string,
+  eventType: string,
+  payload: Record<string, unknown>,
+): Promise<void> {
+  const endpoints = await store.listActiveEndpoints(userId);
+  if (endpoints.length === 0) return;
+
+  const client = await deliveryQueue();
+  // The envelope object is what's persisted on the delivery log, and what
+  // gets re-stringified on DLQ replay — so replay reproduces the exact bytes
+  // that were signed, and a client's signature check still passes.
+  const envelope = { type: eventType, data: payload };
+  const payloadJson = JSON.stringify(envelope);
+
+  for (const endpoint of endpoints) {
+    const signatureHeader = signWebhookPayload(payloadJson, endpoint.secretKey);
+    const log = await store.createDeliveryLog({
+      endpointId: endpoint.endpointId,
+      eventType,
+      payload: envelope,
+      signatureHeader,
+    });
+
+    if (!client) {
+      console.warn("REDIS_URL not configured; webhook delivery not enqueued", {
+        deliveryId: log.deliveryId,
+      });
+      continue;
+    }
+
+    await client.xAdd(WEBHOOK_DELIVERY_QUEUE, "*", {
+      deliveryId: log.deliveryId,
+      endpointId: endpoint.endpointId,
+      targetUrl: endpoint.targetUrl,
+      payload: payloadJson,
+      signature: signatureHeader,
+    });
+  }
+}

apps/api/src/lib/webhookDeliveryStore.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, it } from "vitest";
+import { createHmac } from "node:crypto";
+import { signWebhookPayload } from "./webhook.js";
+import { WebhookDeliveryStore } from "./webhookDeliveryStore.js";
+
+describe("signWebhookPayload (#445)", () => {
+  it("computes HMAC-SHA256 of the exact payload bytes with the endpoint secret", () => {
+    const payload = JSON.stringify({ type: "REFUNDED", data: { trade_id: "abc" } });
+    const secret = "s3cr3t";
+    const expected = createHmac("sha256", secret).update(payload).digest("hex");
+    expect(signWebhookPayload(payload, secret)).toBe(expected);
+  });
+
+  it("produces a different signature for a different secret", () => {
+    const payload = JSON.stringify({ type: "REFUNDED", data: { trade_id: "abc" } });
+    expect(signWebhookPayload(payload, "secret-a")).not.toBe(signWebhookPayload(payload, "secret-b"));
+  });
+
+  it("produces a different signature when the payload changes by one byte", () => {
+    const secret = "s3cr3t";
+    const a = signWebhookPayload(JSON.stringify({ trade_id: "abc" }), secret);
+    const b = signWebhookPayload(JSON.stringify({ trade_id: "abd" }), secret);
+    expect(a).not.toBe(b);
+  });
+});
+
+describe("WebhookDeliveryStore (#445, in-memory mode)", () => {
+  it("registers an endpoint with a 64-hex-char secret key", async () => {
+    const store = new WebhookDeliveryStore();
+    const endpoint = await store.registerEndpoint({
+      userId: "GALICE",
+      targetUrl: "https://example.com/hook",
+    });
+    expect(endpoint.secretKey).toMatch(/^[0-9a-f]{64}$/);
+    expect(endpoint.isActive).toBe(true);
+  });
+
+  it("lists only active endpoints for the given user", async () => {
+    const store = new WebhookDeliveryStore();
+    await store.registerEndpoint({ userId: "GALICE", targetUrl: "https://a.example.com" });
+    await store.registerEndpoint({ userId: "GBOB", targetUrl: "https://b.example.com" });
+    const endpoints = await store.listActiveEndpoints("GALICE");
+    expect(endpoints).toHaveLength(1);
+    expect(endpoints[0].targetUrl).toBe("https://a.example.com");
+  });
+
+  it("claimDeadLetterForReplay is a no-op unless the delivery is DEAD_LETTER", async () => {
+    const store = new WebhookDeliveryStore();
+    const endpoint = await store.registerEndpoint({ userId: "GALICE", targetUrl: "https://a.example.com" });
+    const log = await store.createDeliveryLog({
+      endpointId: endpoint.endpointId,
+      eventType: "REFUNDED",
+      payload: { trade_id: "t1" },
+      signatureHeader: "sig",
+    });
+
+    // Still QUEUED — replay must refuse.
+    expect(await store.claimDeadLetterForReplay(log.deliveryId)).toBeNull();
+
+    await store.recordAttempt(log.deliveryId, { status: "DEAD_LETTER", lastResponseCode: 503 });
+    const claimed = await store.claimDeadLetterForReplay(log.deliveryId);
+    expect(claimed?.status).toBe("QUEUED");
+
+    // Second replay of the same (now QUEUED, not DEAD_LETTER) delivery is a no-op.
+    expect(await store.claimDeadLetterForReplay(log.deliveryId)).toBeNull();
+  });
+});