konnosDashboard
outbound-webhooks.ts554 lines · main
1import { randomBytes } from 'node:crypto';
2
3import { newId, NotFoundError, ValidationError } from '@briven/shared';
4import { and, asc, desc, eq, isNull, lte } from 'drizzle-orm';
5
6import { getDb } from '../db/client.js';
7import {
8 webhookOutboundDeliveries,
9 webhookSubscribers,
10 type WebhookOutboundDelivery,
11 type WebhookOutboundStatus,
12 type WebhookSubscriber,
13} from '../db/schema.js';
14import { decryptValue, encryptValue } from './project-env.js';
15
16/**
17 * Platform → customer outbound webhooks. Each subscriber declares a target
18 * URL + which event types they care about (`*` matches everything). When
19 * the platform emits an event via `publishEvent`, we fan out to every
20 * matching subscriber by inserting a `pending` delivery row. A separate
21 * worker (workers/outbound-webhook-dispatcher.ts) drains the queue,
22 * POSTs the payload + HMAC signature, and retries failures with
23 * exponential backoff up to MAX_ATTEMPTS.
24 *
25 * The signing scheme matches the inbound webhook surface — customers
26 * verify our requests the same way external services verify theirs:
27 * X-Briven-Signature: v1=<hex_hmac_sha256>
28 * X-Briven-Timestamp: <unix-milliseconds>
29 * X-Briven-Event: <event-type>
30 * X-Briven-Event-Id: <stable-event-id-for-dedupe>
31 *
32 * Idempotency: customers should dedupe on `event_id`. Retries reuse the
33 * same id; a successful function that processes the same event twice
34 * after a network blip must be a no-op on the customer's side.
35 */
36
37const NAME_RE = /^[a-z0-9][a-z0-9_-]{0,63}$/i;
38const EVENT_TYPE_RE = /^\*$|^[a-z][a-z0-9._-]{0,63}(,[a-z][a-z0-9._-]{0,63})*$/;
39
40/**
41 * Canonical event types the platform emits. Subscribers either list
42 * specific types (`abuse.report.opened,deploy.failed`) or use `*` to
43 * receive everything. New types are added here as they're wired up by
44 * the call sites that emit them — keep this list narrow on purpose.
45 */
46export const KNOWN_EVENT_TYPES = [
47 'abuse.report.opened',
48 'deploy.succeeded',
49 'deploy.failed',
50 'tier.changed',
51 'project.suspended',
52 'project.resumed',
53 // briven auth — fan-out of authentication lifecycle events to customer
54 // endpoints. Emitted from `apps/api/src/services/auth-audit.ts` when the
55 // matching action row lands; subscribers opt in by listing the type or
56 // by using `*`.
57 'auth.signup',
58 'auth.signin',
59 'auth.signout',
60 'auth.session.revoked',
61 'auth.account.linked',
62 'auth.account.unlinked',
63 'auth.user.deleted',
64 // Phase 1 — security foundation events.
65 'auth.user.banned',
66 'auth.user.unbanned',
67 'auth.user.suspended',
68 'auth.user.unsuspended',
69 'auth.waitlist.approved',
70 'auth.waitlist.rejected',
71] as const;
72export type KnownEventType = (typeof KNOWN_EVENT_TYPES)[number];
73
74/**
75 * The subset of `KNOWN_EVENT_TYPES` produced by briven auth. The dashboard
76 * Auth → Webhooks panel exposes only these as toggle checkboxes; the
77 * general project webhooks panel surfaces every type.
78 */
79export const AUTH_EVENT_TYPES = [
80 'auth.signup',
81 'auth.signin',
82 'auth.signout',
83 'auth.session.revoked',
84 'auth.account.linked',
85 'auth.account.unlinked',
86 'auth.user.deleted',
87 'auth.user.banned',
88 'auth.user.unbanned',
89 'auth.user.suspended',
90 'auth.user.unsuspended',
91 'auth.waitlist.approved',
92 'auth.waitlist.rejected',
93] as const;
94export type AuthEventType = (typeof AUTH_EVENT_TYPES)[number];
95
96function generateSigningSecret(): string {
97 return randomBytes(32).toString('hex');
98}
99
100function validateName(name: string): void {
101 if (!NAME_RE.test(name)) {
102 throw new ValidationError(
103 'subscriber name must be 1-64 chars: alphanumerics, underscore, hyphen; must start with alphanumeric',
104 );
105 }
106}
107
108function validateTargetUrl(url: string): void {
109 let parsed: URL;
110 try {
111 parsed = new URL(url);
112 } catch {
113 throw new ValidationError('target_url must be a valid URL');
114 }
115 if (parsed.protocol !== 'https:' && parsed.protocol !== 'http:') {
116 throw new ValidationError('target_url must use http or https');
117 }
118 // Refuse private-network targets up front. The runtime also has IP
119 // deny lists for customer code; we apply the same shape here so a
120 // misconfigured subscriber can't be used to scan our internal infra.
121 const host = parsed.hostname.toLowerCase();
122 if (
123 host === 'localhost' ||
124 host === '127.0.0.1' ||
125 host === '0.0.0.0' ||
126 host.endsWith('.local') ||
127 host.endsWith('.internal') ||
128 /^10\./.test(host) ||
129 /^192\.168\./.test(host) ||
130 /^172\.(1[6-9]|2[0-9]|3[0-1])\./.test(host) ||
131 /^169\.254\./.test(host)
132 ) {
133 throw new ValidationError('target_url cannot point at a private network');
134 }
135}
136
137function validateEventTypes(eventTypes: string): void {
138 if (!EVENT_TYPE_RE.test(eventTypes)) {
139 throw new ValidationError(
140 'event_types must be `*` or a comma-separated list of types (e.g. abuse.report.opened,deploy.failed)',
141 );
142 }
143}
144
145export interface PublicSubscriber {
146 id: string;
147 projectId: string;
148 name: string;
149 targetUrl: string;
150 eventTypes: string;
151 enabled: boolean;
152 allowedIps: string;
153 lastDeliveryAt: Date | null;
154 lastDeliveryStatus: WebhookOutboundStatus | null;
155 createdAt: Date;
156 updatedAt: Date;
157}
158
159function redact(row: WebhookSubscriber): PublicSubscriber {
160 return {
161 id: row.id,
162 projectId: row.projectId,
163 name: row.name,
164 targetUrl: row.targetUrl,
165 eventTypes: row.eventTypes,
166 enabled: row.enabled,
167 allowedIps: row.allowedIps,
168 lastDeliveryAt: row.lastDeliveryAt,
169 lastDeliveryStatus: row.lastDeliveryStatus,
170 createdAt: row.createdAt,
171 updatedAt: row.updatedAt,
172 };
173}
174
175export interface CreateSubscriberInput {
176 projectId: string;
177 name: string;
178 targetUrl: string;
179 eventTypes?: string;
180 enabled?: boolean;
181 allowedIps?: string;
182 createdBy: string | null;
183}
184
185export async function listSubscribers(projectId: string): Promise<PublicSubscriber[]> {
186 const db = getDb();
187 const rows = await db
188 .select()
189 .from(webhookSubscribers)
190 .where(and(eq(webhookSubscribers.projectId, projectId), isNull(webhookSubscribers.deletedAt)))
191 .orderBy(asc(webhookSubscribers.name));
192 return rows.map(redact);
193}
194
195export async function getSubscriberRaw(
196 subscriberId: string,
197 projectId: string,
198): Promise<WebhookSubscriber> {
199 const db = getDb();
200 const rows = await db
201 .select()
202 .from(webhookSubscribers)
203 .where(
204 and(
205 eq(webhookSubscribers.id, subscriberId),
206 eq(webhookSubscribers.projectId, projectId),
207 isNull(webhookSubscribers.deletedAt),
208 ),
209 )
210 .limit(1);
211 const row = rows[0];
212 if (!row) throw new NotFoundError('webhook_subscriber', subscriberId);
213 return row;
214}
215
216const IP_RE = /^(\d{1,3}\.){3}\d{1,3}(\/\d{1,2})?$/;
217
218function validateAllowedIps(raw: string): void {
219 if (!raw) return;
220 const parts = raw.split(',').map((s) => s.trim()).filter(Boolean);
221 for (const ip of parts) {
222 if (!IP_RE.test(ip)) {
223 throw new ValidationError(
224 `allowed_ips must be comma-separated IPv4 addresses or CIDR blocks (e.g. 1.2.3.4, 10.0.0.0/8). Invalid: ${ip}`,
225 );
226 }
227 }
228}
229
230export async function createSubscriber(
231 input: CreateSubscriberInput,
232): Promise<{ subscriber: PublicSubscriber; plaintextSecret: string }> {
233 validateName(input.name);
234 validateTargetUrl(input.targetUrl);
235 const eventTypes = input.eventTypes ?? '*';
236 validateEventTypes(eventTypes);
237 if (input.allowedIps) validateAllowedIps(input.allowedIps);
238 const plaintextSecret = generateSigningSecret();
239 const db = getDb();
240 const id = newId('whs');
241 try {
242 const inserted = await db
243 .insert(webhookSubscribers)
244 .values({
245 id,
246 projectId: input.projectId,
247 name: input.name,
248 targetUrl: input.targetUrl,
249 eventTypes,
250 signingSecretEncrypted: encryptValue(plaintextSecret),
251 enabled: input.enabled ?? true,
252 allowedIps: input.allowedIps ?? '',
253 createdBy: input.createdBy,
254 })
255 .returning();
256 const row = inserted[0];
257 if (!row) throw new Error('insert returned no row');
258 return { subscriber: redact(row), plaintextSecret };
259 } catch (err) {
260 if (err instanceof Error && /unique|duplicate/i.test(err.message)) {
261 throw new ValidationError(
262 `a subscriber named "${input.name}" already exists for this project`,
263 );
264 }
265 throw err;
266 }
267}
268
269export interface UpdateSubscriberInput {
270 name?: string;
271 targetUrl?: string;
272 eventTypes?: string;
273 enabled?: boolean;
274 allowedIps?: string;
275}
276
277export async function updateSubscriber(
278 subscriberId: string,
279 projectId: string,
280 patch: UpdateSubscriberInput,
281): Promise<PublicSubscriber> {
282 const existing = await getSubscriberRaw(subscriberId, projectId);
283 const updates: Partial<WebhookSubscriber> = { updatedAt: new Date() };
284 if (patch.name !== undefined && patch.name !== existing.name) {
285 validateName(patch.name);
286 updates.name = patch.name;
287 }
288 if (patch.targetUrl !== undefined && patch.targetUrl !== existing.targetUrl) {
289 validateTargetUrl(patch.targetUrl);
290 updates.targetUrl = patch.targetUrl;
291 }
292 if (patch.eventTypes !== undefined && patch.eventTypes !== existing.eventTypes) {
293 validateEventTypes(patch.eventTypes);
294 updates.eventTypes = patch.eventTypes;
295 }
296 if (patch.enabled !== undefined && patch.enabled !== existing.enabled) {
297 updates.enabled = patch.enabled;
298 }
299 if (patch.allowedIps !== undefined && patch.allowedIps !== existing.allowedIps) {
300 validateAllowedIps(patch.allowedIps);
301 updates.allowedIps = patch.allowedIps;
302 }
303 const db = getDb();
304 const result = await db
305 .update(webhookSubscribers)
306 .set(updates)
307 .where(
308 and(eq(webhookSubscribers.id, subscriberId), eq(webhookSubscribers.projectId, projectId)),
309 )
310 .returning();
311 if (!result[0]) throw new NotFoundError('webhook_subscriber', subscriberId);
312 return redact(result[0]);
313}
314
315export async function rotateSubscriberSecret(
316 subscriberId: string,
317 projectId: string,
318): Promise<{ subscriber: PublicSubscriber; plaintextSecret: string }> {
319 await getSubscriberRaw(subscriberId, projectId);
320 const plaintextSecret = generateSigningSecret();
321 const db = getDb();
322 const result = await db
323 .update(webhookSubscribers)
324 .set({
325 signingSecretEncrypted: encryptValue(plaintextSecret),
326 updatedAt: new Date(),
327 })
328 .where(
329 and(eq(webhookSubscribers.id, subscriberId), eq(webhookSubscribers.projectId, projectId)),
330 )
331 .returning();
332 if (!result[0]) throw new NotFoundError('webhook_subscriber', subscriberId);
333 return { subscriber: redact(result[0]), plaintextSecret };
334}
335
336export async function deleteSubscriber(
337 subscriberId: string,
338 projectId: string,
339): Promise<void> {
340 const db = getDb();
341 const result = await db
342 .update(webhookSubscribers)
343 .set({ deletedAt: new Date(), enabled: false, updatedAt: new Date() })
344 .where(
345 and(
346 eq(webhookSubscribers.id, subscriberId),
347 eq(webhookSubscribers.projectId, projectId),
348 isNull(webhookSubscribers.deletedAt),
349 ),
350 )
351 .returning({ id: webhookSubscribers.id });
352 if (!result[0]) throw new NotFoundError('webhook_subscriber', subscriberId);
353}
354
355export async function listOutboundDeliveries(
356 subscriberId: string,
357 projectId: string,
358 opts: { limit?: number; status?: WebhookOutboundStatus } = {},
359): Promise<WebhookOutboundDelivery[]> {
360 const db = getDb();
361 const whereClauses = [
362 eq(webhookOutboundDeliveries.subscriberId, subscriberId),
363 eq(webhookOutboundDeliveries.projectId, projectId),
364 ];
365 if (opts.status) whereClauses.push(eq(webhookOutboundDeliveries.status, opts.status));
366 return db
367 .select()
368 .from(webhookOutboundDeliveries)
369 .where(and(...whereClauses))
370 .orderBy(desc(webhookOutboundDeliveries.createdAt))
371 .limit(opts.limit ?? 100);
372}
373
374/**
375 * Fan an event out to every matching subscriber on a project. Inserts one
376 * `pending` delivery row per match; the dispatcher worker drains them.
377 *
378 * Idempotency: callers pass a stable `eventId` if they want retries on the
379 * caller's side to dedupe on briven's side. Without one, every call
380 * inserts fresh rows — fine for fire-and-forget events.
381 */
382export async function publishEvent(input: {
383 projectId: string;
384 eventType: KnownEventType;
385 payload: Record<string, unknown>;
386 eventId?: string;
387}): Promise<{ deliveriesQueued: number }> {
388 const db = getDb();
389 const subscribers = await db
390 .select()
391 .from(webhookSubscribers)
392 .where(
393 and(
394 eq(webhookSubscribers.projectId, input.projectId),
395 eq(webhookSubscribers.enabled, true),
396 isNull(webhookSubscribers.deletedAt),
397 ),
398 );
399 const matching = subscribers.filter((s) => matchesEventType(s.eventTypes, input.eventType));
400 if (matching.length === 0) return { deliveriesQueued: 0 };
401
402 const eventId = input.eventId ?? newId('wev');
403 const now = new Date();
404 await db.insert(webhookOutboundDeliveries).values(
405 matching.map((s) => ({
406 id: newId('whod'),
407 subscriberId: s.id,
408 projectId: input.projectId,
409 eventId,
410 eventType: input.eventType,
411 payload: input.payload,
412 status: 'pending' as const,
413 attemptCount: '0',
414 nextAttemptAt: now,
415 })),
416 );
417 return { deliveriesQueued: matching.length };
418}
419
420export function matchesEventType(filter: string, eventType: string): boolean {
421 if (filter === '*') return true;
422 return filter.split(',').some((t) => t.trim() === eventType);
423}
424
425/**
426 * Dispatcher claim path. Selects up to `limit` deliveries whose
427 * next_attempt_at is in the past and which are still pending. The
428 * caller does the POST + records the result via recordDeliveryResult.
429 */
430export async function claimDueDeliveries(
431 now: Date,
432 limit: number,
433): Promise<WebhookOutboundDelivery[]> {
434 const db = getDb();
435 return db
436 .select()
437 .from(webhookOutboundDeliveries)
438 .where(
439 and(
440 eq(webhookOutboundDeliveries.status, 'pending'),
441 lte(webhookOutboundDeliveries.nextAttemptAt, now),
442 ),
443 )
444 .orderBy(asc(webhookOutboundDeliveries.nextAttemptAt))
445 .limit(limit);
446}
447
448export const MAX_ATTEMPTS = 5;
449
450/**
451 * Exponential backoff schedule (in ms) for the next retry attempt
452 * AFTER the n-th failure. Attempt 0 retries 30s later, attempt 4
453 * (the 5th try) is the terminal state — record it as 'failed' instead
454 * of scheduling a 6th. Cap kept short: a webhook that hasn't recovered
455 * in ~30 minutes isn't going to.
456 */
457export function retryDelayMs(attemptCount: number): number {
458 const schedule = [30_000, 60_000, 180_000, 600_000, 1_800_000];
459 return schedule[Math.min(attemptCount, schedule.length - 1)] ?? 1_800_000;
460}
461
462export async function recordDeliveryResult(input: {
463 deliveryId: string;
464 attemptCount: number;
465 statusCode: number | null;
466 durationMs: number;
467 errorMessage: string | null;
468 ok: boolean;
469 ranAt: Date;
470}): Promise<void> {
471 const db = getDb();
472 const nextAttempt = input.attemptCount + 1;
473 const terminal = input.ok || nextAttempt >= MAX_ATTEMPTS;
474 const nextStatus: WebhookOutboundStatus = input.ok
475 ? 'ok'
476 : terminal
477 ? 'failed'
478 : 'pending';
479 const nextAttemptAt = terminal
480 ? input.ranAt
481 : new Date(input.ranAt.getTime() + retryDelayMs(input.attemptCount));
482
483 await db
484 .update(webhookOutboundDeliveries)
485 .set({
486 status: nextStatus,
487 attemptCount: String(nextAttempt),
488 lastAttemptAt: input.ranAt,
489 statusCode: input.statusCode == null ? null : String(input.statusCode),
490 durationMs: String(input.durationMs),
491 errorMessage: input.errorMessage,
492 nextAttemptAt,
493 })
494 .where(eq(webhookOutboundDeliveries.id, input.deliveryId));
495
496 // Roll the subscriber's last_delivery_* summary forward so the
497 // dashboard list view stays cheap (no aggregate needed).
498 const delivery = await db
499 .select({ subscriberId: webhookOutboundDeliveries.subscriberId })
500 .from(webhookOutboundDeliveries)
501 .where(eq(webhookOutboundDeliveries.id, input.deliveryId))
502 .limit(1);
503 const subscriberId = delivery[0]?.subscriberId;
504 if (subscriberId) {
505 await db
506 .update(webhookSubscribers)
507 .set({
508 lastDeliveryAt: input.ranAt,
509 lastDeliveryStatus: nextStatus,
510 updatedAt: new Date(),
511 })
512 .where(eq(webhookSubscribers.id, subscriberId));
513 }
514}
515
516/**
517 * Replay a delivery — reset it to pending with attemptCount=0 so the
518 * dispatcher will retry it on the next tick.
519 */
520export async function replayDelivery(
521 deliveryId: string,
522 projectId: string,
523): Promise<void> {
524 const db = getDb();
525 const delivery = await db
526 .select({ id: webhookOutboundDeliveries.id })
527 .from(webhookOutboundDeliveries)
528 .where(
529 and(
530 eq(webhookOutboundDeliveries.id, deliveryId),
531 eq(webhookOutboundDeliveries.projectId, projectId),
532 ),
533 )
534 .limit(1);
535 if (!delivery[0]) throw new NotFoundError('webhook_delivery', deliveryId);
536
537 await db
538 .update(webhookOutboundDeliveries)
539 .set({
540 status: 'pending',
541 attemptCount: '0',
542 nextAttemptAt: new Date(),
543 lastAttemptAt: null,
544 statusCode: null,
545 durationMs: null,
546 errorMessage: null,
547 })
548 .where(eq(webhookOutboundDeliveries.id, deliveryId));
549}
550
551/** Decrypt a subscriber's stored signing secret. Reveals plaintext. */
552export function decryptSubscriberSecret(row: WebhookSubscriber): string {
553 return decryptValue(row.signingSecretEncrypted);
554}