konnosDashboard
outbound-webhook-dispatcher.ts218 lines · main
1import { createHmac } from 'node:crypto';
2import { lookup } from 'node:dns/promises';
3
4import { env } from '../env.js';
5import { log } from '../lib/logger.js';
6import { getDb } from '../db/client.js';
7import { webhookSubscribers } from '../db/schema.js';
8import { eq } from 'drizzle-orm';
9import {
10 claimDueDeliveries,
11 decryptSubscriberSecret,
12 recordDeliveryResult,
13} from '../services/outbound-webhooks.js';
14
15/**
16 * Outbound webhook dispatcher. Every TICK_MS:
17 * 1. claim up to BATCH_SIZE pending deliveries whose next_attempt_at
18 * has passed.
19 * 2. for each delivery: decrypt the subscriber's secret, sign the
20 * payload, POST to target_url.
21 * 3. record the outcome — on success mark `ok`; on failure schedule
22 * the next attempt with exponential backoff up to MAX_ATTEMPTS,
23 * then mark `failed`.
24 *
25 * Concurrency: deliveries to different subscribers fan out in parallel.
26 * Subscriber-level ordering is best-effort — strict per-subscriber
27 * serialisation would need explicit locking, and outbound webhooks are
28 * intentionally at-least-once with customer-side dedupe on event_id.
29 */
30
31const TICK_MS = 30_000;
32const BATCH_SIZE = 50;
33const HTTP_TIMEOUT_MS = 10_000;
34const MAX_ERROR_LEN = 500;
35// Match the constant in services/outbound-webhooks.ts (we don't want a
36// runtime import cycle just to share a number).
37const MAX_ATTEMPTS = 5;
38
39let timer: ReturnType<typeof setInterval> | null = null;
40let inflight = false;
41
42async function tick(): Promise<void> {
43 if (inflight) {
44 log.warn('outbound_webhook_dispatcher_tick_skipped_inflight');
45 return;
46 }
47 inflight = true;
48 const now = new Date();
49 try {
50 const due = await claimDueDeliveries(now, BATCH_SIZE);
51 if (due.length === 0) return;
52 log.info('outbound_webhook_dispatcher_tick', { dueCount: due.length });
53 await Promise.all(due.map((d) => fireOne(d, now)));
54 } catch (err) {
55 log.error('outbound_webhook_dispatcher_tick_failed', {
56 message: err instanceof Error ? err.message : String(err),
57 });
58 } finally {
59 inflight = false;
60 }
61}
62
63async function fireOne(
64 delivery: Awaited<ReturnType<typeof claimDueDeliveries>>[number],
65 now: Date,
66): Promise<void> {
67 const attemptCount = Number(delivery.attemptCount);
68 const db = getDb();
69 const subRows = await db
70 .select()
71 .from(webhookSubscribers)
72 .where(eq(webhookSubscribers.id, delivery.subscriberId))
73 .limit(1);
74 const subscriber = subRows[0];
75 if (!subscriber || !subscriber.enabled || subscriber.deletedAt) {
76 // Subscriber was paused or deleted between claim and dispatch. Force
77 // the row to its terminal `failed` state by passing the
78 // last-retry-slot attempt count to the recorder. No HTTP attempted.
79 await recordDeliveryResult({
80 deliveryId: delivery.id,
81 attemptCount: MAX_ATTEMPTS - 1,
82 statusCode: null,
83 durationMs: 0,
84 errorMessage: 'subscriber disabled or deleted between claim and dispatch',
85 ok: false,
86 ranAt: now,
87 });
88 return;
89 }
90
91 // IP allowlist check
92 if (subscriber.allowedIps) {
93 const allowed = subscriber.allowedIps.split(',').map((s) => s.trim()).filter(Boolean);
94 if (allowed.length > 0) {
95 let resolved: string;
96 try {
97 const hostname = new URL(subscriber.targetUrl).hostname;
98 resolved = await lookup(hostname).then((r) => r.address);
99 } catch {
100 await recordDeliveryResult({
101 deliveryId: delivery.id,
102 attemptCount: MAX_ATTEMPTS - 1,
103 statusCode: null,
104 durationMs: 0,
105 errorMessage: 'ip_allowlist: dns resolution failed',
106 ok: false,
107 ranAt: now,
108 });
109 return;
110 }
111 if (!allowed.some((a) => ipMatches(resolved, a))) {
112 await recordDeliveryResult({
113 deliveryId: delivery.id,
114 attemptCount: MAX_ATTEMPTS - 1,
115 statusCode: null,
116 durationMs: 0,
117 errorMessage: `ip_allowlist: resolved ${resolved} not in allowlist`,
118 ok: false,
119 ranAt: now,
120 });
121 return;
122 }
123 }
124 }
125
126 const secret = decryptSubscriberSecret(subscriber);
127 const body = JSON.stringify(delivery.payload);
128 const timestamp = String(now.getTime());
129 const signature = `v1=${createHmac('sha256', secret).update(`${timestamp}.${body}`).digest('hex')}`;
130
131 let statusCode: number | null = null;
132 let errorMessage: string | null = null;
133 let ok = false;
134
135 const t0 = Date.now();
136 try {
137 const res = await fetch(subscriber.targetUrl, {
138 method: 'POST',
139 headers: {
140 'content-type': 'application/json',
141 'x-briven-signature': signature,
142 'x-briven-timestamp': timestamp,
143 'x-briven-event': delivery.eventType,
144 'x-briven-event-id': delivery.eventId,
145 'user-agent': `briven-webhook/1.0 (${env.BRIVEN_API_ORIGIN})`,
146 },
147 body,
148 signal: AbortSignal.timeout(HTTP_TIMEOUT_MS),
149 });
150 statusCode = res.status;
151 ok = res.status >= 200 && res.status < 300;
152 if (!ok) {
153 const text = await res.text().catch(() => '');
154 errorMessage = `${res.status}: ${text.slice(0, MAX_ERROR_LEN)}`;
155 }
156 } catch (err) {
157 errorMessage = (err instanceof Error ? err.message : String(err)).slice(0, MAX_ERROR_LEN);
158 }
159 const durationMs = Date.now() - t0;
160
161 await recordDeliveryResult({
162 deliveryId: delivery.id,
163 attemptCount,
164 statusCode,
165 durationMs,
166 errorMessage,
167 ok,
168 ranAt: now,
169 });
170}
171
172export function startOutboundWebhookDispatcher(): void {
173 if (timer) return;
174 if (!env.BRIVEN_DATABASE_URL) {
175 log.warn('outbound_webhook_dispatcher_skipped_no_db');
176 return;
177 }
178 // 75s after boot — sits between schedule (45s) and retention (90s)
179 // so the three workers don't all hit the connection pool together.
180 setTimeout(() => {
181 void tick();
182 timer = setInterval(() => {
183 void tick();
184 }, TICK_MS);
185 }, 75_000).unref?.();
186 log.info('outbound_webhook_dispatcher_armed', { tickMs: TICK_MS, batch: BATCH_SIZE });
187}
188
189export function stopOutboundWebhookDispatcher(): void {
190 if (timer) {
191 clearInterval(timer);
192 timer = null;
193 }
194}
195
196function ipMatches(ip: string, pattern: string): boolean {
197 if (pattern.includes('/')) {
198 const [base, bits] = pattern.split('/');
199 const mask = parseInt(bits!, 10);
200 const ipNum = ipv4ToNumber(ip);
201 const baseNum = ipv4ToNumber(base!);
202 if (ipNum == null || baseNum == null) return false;
203 const shift = 32 - mask;
204 return (ipNum >> shift) === (baseNum >> shift);
205 }
206 return ip === pattern;
207}
208
209function ipv4ToNumber(ip: string): number | null {
210 const parts = ip.split('.').map(Number);
211 if (parts.length !== 4 || parts.some((n) => Number.isNaN(n) || n < 0 || n > 255)) {
212 return null;
213 }
214 return (parts[0]! << 24) | (parts[1]! << 16) | (parts[2]! << 8) | parts[3]!;
215}
216
217// Exported for tests.
218export const _internals = { tick, fireOne, ipMatches };