Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
CodeIssuesPull RequestsActionsSecurityInsightsSettings
✨ AI
ExplainAsk AIWorkspaceSpec to PRTestsNL SearchDebt MapArchaeology
More
CommitsDiscussionsWikiProjectsReleasesContributorsPulseGatesDeploymentsPipelineAgents
Blame · Line-by-line history

webhook-delivery.ts

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

Gluecron.com/src/lib/webhook-delivery.ts
webhook-delivery.tsBlame543 lines · 3 contributors
Normal viewRaw
8405c43Claude1/**
2 * Reliable webhook delivery (migration 0056).
3 *
4 * Replaces the inline single-shot fetch in src/routes/webhooks.tsx with a
5 * durable pending-row queue. `enqueueWebhookDelivery()` precomputes the HMAC
6 * signature, inserts one row per (hook, event) into `webhook_deliveries`
7 * with `status='pending'` and `next_attempt_at=now()`, then kicks the
8 * worker. The worker picks claims rows whose `next_attempt_at <= now()` and
9 * attempts each POST.
10 *
11 * Retry schedule (after attempt #1 fires immediately):
12 * attempt 2 → +30s
13 * attempt 3 → +2m
14 * attempt 4 → +10m
15 * attempt 5 → +1h
16 * attempt 6 → +6h
17 * after attempt 6 → status='dead' (no further retries; row kept for ops)
18 *
19 * 2xx response → status='succeeded' + succeeded_at + last_status_code.
20 * Anything else (including network errors and timeouts) → counted as a
21 * failed attempt and rescheduled.
22 *
23 * Public surface:
24 * - enqueueWebhookDelivery({...}) — fire-and-forget queue insert
25 * - attemptDelivery(deliveryId) — single attempt (exported for tests)
26 * - drainPendingDeliveries() — drain a batch (exported for tests)
27 * - startWebhookDeliveryWorker() — background poll loop
28 * - MAX_ATTEMPTS, BACKOFF_MS — exported for tests
29 */
30
31import { and, asc, eq, lte, sql } from "drizzle-orm";
32import { db } from "../db";
e2206a6Test User33import { webhookDeliveries, webhooks, repositories, users } from "../db/schema";
bc519aeClaude34import { assertPublicUrl } from "./ssrf-guard";
e2206a6Test User35import { notify } from "./notify";
8405c43Claude36
37// ---------------------------------------------------------------------------
38// Tunables
39// ---------------------------------------------------------------------------
40
41/** How many attempts in total before a row goes to status='dead'. */
42export const MAX_ATTEMPTS = 6;
43
44/**
45 * Backoff schedule, indexed by the *next* attempt number we're scheduling.
46 * After attemptCount=N fails, we schedule attempt N+1 at now() + BACKOFF_MS[N].
47 * Index 0 is unused (attempt 1 is queued at now() by enqueue, not by retry).
48 */
49export const BACKOFF_MS: number[] = [
50 0, // [0] unused
51 30_000, // after attempt 1 fails → +30s for attempt 2
52 120_000, // after attempt 2 fails → +2m for attempt 3
53 600_000, // after attempt 3 fails → +10m for attempt 4
54 3_600_000, // after attempt 4 fails → +1h for attempt 5
55 21_600_000, // after attempt 5 fails → +6h for attempt 6
56];
57
58/** How often the worker scans for due pending rows. */
59const DEFAULT_POLL_INTERVAL_MS = 5_000;
60
61/** Cap on rows pulled in a single tick. */
62const BATCH_SIZE = 10;
63
64/** Per-delivery HTTP timeout. */
65const DELIVERY_TIMEOUT_MS = 10_000;
66
67/** Cap on stored last_error string. */
68const ERROR_CAP = 2_000;
69
6088927ccantynz-alt70/** How many redirect hops a webhook delivery may follow. */
71const MAX_REDIRECTS = 3;
72
73/**
74 * POST to a user-supplied webhook URL, re-running the SSRF guard on every
75 * redirect hop.
76 *
77 * Two holes this closes:
78 *
79 * 1. `fetch()` defaults to `redirect: "follow"`. assertPublicUrl only ever saw
80 * the URL the user registered, so a perfectly public endpoint could answer
81 * 302 → http://169.254.169.254/latest/meta-data/ (or any RFC1918 host) and
82 * fetch would obediently follow it. The guard was bypassed entirely by one
83 * redirect.
84 *
85 * 2. assertPublicUrl is a synchronous literal check. resolvesToPrivate() — the
86 * DNS layer that catches a public-looking hostname pointing at a private
87 * address — existed, was fully unit-tested, and had ZERO production
88 * callers: only the test file imported it. `evil.example.com A 127.0.0.1`
89 * sailed through.
90 *
91 * Redirects are followed rather than refused because http→https upgrades are
92 * a legitimate and common webhook endpoint behaviour; refusing them outright
93 * would break working integrations.
94 *
95 * Residual risk, deliberately not solved here: DNS rebinding. The resolve and
96 * the connect are separate operations, so a hostname can return a public
97 * address to our lookup and a private one to the socket. Eliminating that
98 * needs connect-time pinning (resolve, validate, then dial the literal IP with
99 * a Host header), which Bun's fetch does not expose. This raises the bar from
100 * "trivially bypassable with a redirect" to "requires an active rebinding
101 * attack".
102 */
103async function deliverFollowingRedirects(
104 startUrl: string,
105 init: { method: string; headers: Record<string, string>; body: string }
106): Promise<{ status: number; blockedReason?: string }> {
107 const { assertPublicUrl, resolvesToPrivate, ssrfPrivateAllowed } =
108 await import("./ssrf-guard");
109
110 // Honour the same escape hatch assertPublicUrl uses. Without this the DNS
111 // layer would still block 127.0.0.1 under SSRF_ALLOW_PRIVATE=1 and in the
112 // test env, where suites legitimately point webhooks at a local Bun.serve().
113 const enforcing = !ssrfPrivateAllowed();
114
115 let url = startUrl;
116 for (let hop = 0; hop <= MAX_REDIRECTS; hop++) {
117 const guard = assertPublicUrl(url);
118 if (!guard.ok) return { status: 0, blockedReason: guard.reason };
119 if (enforcing && (await resolvesToPrivate(guard.url.hostname))) {
120 return { status: 0, blockedReason: "hostname resolves to a private address" };
121 }
122
123 const res = await fetch(url, {
124 ...init,
125 redirect: "manual",
126 signal: AbortSignal.timeout(DELIVERY_TIMEOUT_MS),
127 });
128
129 if (res.status < 300 || res.status >= 400) return { status: res.status };
130
131 const location = res.headers.get("location");
132 if (!location) return { status: res.status };
133 // Resolve relative Locations against the current hop.
134 url = new URL(location, url).toString();
135 }
136 return { status: 0, blockedReason: `more than ${MAX_REDIRECTS} redirects` };
137}
138
8405c43Claude139// ---------------------------------------------------------------------------
140// Signing
141// ---------------------------------------------------------------------------
142
143/**
144 * Compute the `sha256=<hex>` HMAC signature for a payload. Returns the empty
145 * string when the hook has no secret — caller should still POST but skip the
146 * `X-Gluecron-Signature` header.
147 */
148export async function computeSignature(
149 secret: string | null,
150 payloadJson: string
151): Promise<string> {
152 if (!secret) return "";
153 const encoder = new TextEncoder();
154 const key = await crypto.subtle.importKey(
155 "raw",
156 encoder.encode(secret),
157 { name: "HMAC", hash: "SHA-256" },
158 false,
159 ["sign"]
160 );
161 const signature = await crypto.subtle.sign(
162 "HMAC",
163 key,
164 encoder.encode(payloadJson)
165 );
166 return (
167 "sha256=" +
168 Array.from(new Uint8Array(signature))
169 .map((b) => b.toString(16).padStart(2, "0"))
170 .join("")
171 );
172}
173
174// ---------------------------------------------------------------------------
175// Enqueue
176// ---------------------------------------------------------------------------
177
178/**
179 * Insert one pending delivery row. Caller passes hook + event + payload; we
180 * snapshot the signature now so future schedule-time secret rotation can't
181 * silently invalidate in-flight retries. Returns the new row id, or null on
182 * insert failure (logged; never throws).
183 */
184export async function enqueueWebhookDelivery(input: {
185 webhookId: string;
186 secret: string | null;
187 event: string;
188 payload: Record<string, unknown>;
189}): Promise<string | null> {
190 try {
191 const payloadJson = JSON.stringify(input.payload);
192 const signature = await computeSignature(input.secret, payloadJson);
193
194 const [row] = await db
195 .insert(webhookDeliveries)
196 .values({
197 webhookId: input.webhookId,
198 event: input.event,
199 payload: payloadJson,
200 signature,
201 attemptCount: 0,
202 nextAttemptAt: new Date(),
203 status: "pending",
204 })
205 .returning({ id: webhookDeliveries.id });
206
207 return row?.id ?? null;
208 } catch (err) {
209 console.error("[webhook-delivery] enqueue failed:", err);
210 return null;
211 }
212}
213
e2206a6Test User214/**
215 * Alert the repo owner that a webhook has permanently stopped delivering —
216 * previously the only signal was a console.error nobody reads (chat-
217 * notifier.ts) or a "no deliveries yet" pill on a settings page nobody was
218 * visiting. Fire-and-forget, in-app notification only; never throws — a bug
219 * here must not affect delivery bookkeeping.
220 */
221async function notifyWebhookDead(webhookId: string, reason: string): Promise<void> {
222 try {
223 const [row] = await db
224 .select({
225 url: webhooks.url,
226 repoName: repositories.name,
227 repoId: repositories.id,
228 ownerId: repositories.ownerId,
229 ownerUsername: users.username,
230 })
231 .from(webhooks)
232 .innerJoin(repositories, eq(repositories.id, webhooks.repositoryId))
233 .innerJoin(users, eq(users.id, repositories.ownerId))
234 .where(eq(webhooks.id, webhookId))
235 .limit(1);
236 if (!row) return;
237
238 await notify(row.ownerId, {
239 kind: "integration_dead",
240 title: `Webhook delivery failed permanently — ${row.repoName}`,
241 body: `A webhook to ${row.url} has stopped retrying (${reason}). It will not be attempted again until you re-check its configuration.`,
242 url: `/${row.ownerUsername}/${row.repoName}/settings/webhooks`,
243 repositoryId: row.repoId,
244 });
245 } catch (err) {
246 console.error("[webhook-delivery] notifyWebhookDead failed:", err);
247 }
248}
249
8405c43Claude250// ---------------------------------------------------------------------------
251// One attempt
252// ---------------------------------------------------------------------------
253
254/**
255 * Run a single delivery attempt against the given row. Looks up the hook URL
256 * fresh (so a deleted hook short-circuits), POSTs, then updates the row to
257 * succeeded / pending (with new next_attempt_at) / dead based on the result.
258 *
259 * Returns 'succeeded' | 'retry' | 'dead' | 'gone' (hook was deleted).
260 */
261export async function attemptDelivery(
262 deliveryId: string
263): Promise<"succeeded" | "retry" | "dead" | "gone"> {
264 // Pull the delivery row + the hook URL in one shot.
265 const rows = await db
266 .select({
267 delivery: webhookDeliveries,
268 url: webhooks.url,
269 isActive: webhooks.isActive,
270 })
271 .from(webhookDeliveries)
272 .leftJoin(webhooks, eq(webhooks.id, webhookDeliveries.webhookId))
273 .where(eq(webhookDeliveries.id, deliveryId))
274 .limit(1);
275
276 const row = rows[0];
277 if (!row) return "gone";
278 if (!row.url || row.isActive === false) {
279 // Hook deleted or disabled between enqueue and attempt — mark dead so
280 // we don't keep polling it.
281 await db
282 .update(webhookDeliveries)
283 .set({
284 status: "dead",
285 lastError: "hook deleted or disabled",
286 lastAttemptedAt: new Date(),
287 })
288 .where(eq(webhookDeliveries.id, deliveryId));
289 return "gone";
290 }
291
292 const d = row.delivery;
293 const attemptNumber = d.attemptCount + 1;
294
bc519aeClaude295 // SSRF guard (BUILD_BIBLE §7): refuse to POST to private/internal
296 // addresses. Permanent condition — retrying won't change the URL — so
297 // the row goes straight to 'dead' with a clear reason. Never throws.
298 const guard = assertPublicUrl(row.url);
299 if (!guard.ok) {
300 const blockedAt = new Date();
301 await db
302 .update(webhookDeliveries)
303 .set({
304 status: "dead",
305 attemptCount: attemptNumber,
306 lastAttemptedAt: blockedAt,
307 lastStatusCode: null,
308 lastError: `blocked: ${guard.reason} (SSRF protection)`,
309 nextAttemptAt: null,
310 })
311 .where(eq(webhookDeliveries.id, deliveryId));
312
313 try {
314 await db
315 .update(webhooks)
316 .set({ lastDeliveredAt: blockedAt, lastStatus: 0 })
317 .where(eq(webhooks.id, d.webhookId));
318 } catch {
319 /* swallow */
320 }
321
e2206a6Test User322 void notifyWebhookDead(d.webhookId, `blocked: ${guard.reason}`);
323
bc519aeClaude324 return "dead";
325 }
326
8405c43Claude327 // Perform the POST.
328 let statusCode: number | null = null;
329 let errorMessage: string | null = null;
330 let success = false;
331
332 try {
333 const headers: Record<string, string> = {
334 "Content-Type": "application/json",
335 "X-Gluecron-Event": d.event,
336 "X-Gluecron-Delivery": d.id,
337 };
338 if (d.signature) headers["X-Gluecron-Signature"] = d.signature;
339
6088927ccantynz-alt340 const res = await deliverFollowingRedirects(row.url, {
8405c43Claude341 method: "POST",
342 headers,
343 body: d.payload,
344 });
345 statusCode = res.status;
346 success = res.status >= 200 && res.status < 300;
347 if (!success) {
6088927ccantynz-alt348 errorMessage = res.blockedReason
349 ? `blocked: ${res.blockedReason} (SSRF protection)`
350 : `HTTP ${res.status}`;
8405c43Claude351 }
352 } catch (err) {
353 errorMessage =
354 err instanceof Error ? err.message : String(err ?? "unknown error");
355 if (errorMessage.length > ERROR_CAP) {
356 errorMessage = errorMessage.slice(0, ERROR_CAP);
357 }
358 }
359
360 const now = new Date();
361
362 if (success) {
363 await db
364 .update(webhookDeliveries)
365 .set({
366 status: "succeeded",
367 attemptCount: attemptNumber,
368 lastAttemptedAt: now,
369 lastStatusCode: statusCode,
370 lastError: null,
371 succeededAt: now,
372 nextAttemptAt: null,
373 })
374 .where(eq(webhookDeliveries.id, deliveryId));
375
376 // Best-effort sync of the parent hook row for the legacy "last status"
377 // surface in /settings/webhooks. Never throws out.
378 try {
379 await db
380 .update(webhooks)
381 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
382 .where(eq(webhooks.id, d.webhookId));
383 } catch {
384 /* swallow */
385 }
386
387 return "succeeded";
388 }
389
390 // Failure path.
391 if (attemptNumber >= MAX_ATTEMPTS) {
392 await db
393 .update(webhookDeliveries)
394 .set({
395 status: "dead",
396 attemptCount: attemptNumber,
397 lastAttemptedAt: now,
398 lastStatusCode: statusCode,
399 lastError: errorMessage,
400 nextAttemptAt: null,
401 })
402 .where(eq(webhookDeliveries.id, deliveryId));
403
404 try {
405 await db
406 .update(webhooks)
407 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
408 .where(eq(webhooks.id, d.webhookId));
409 } catch {
410 /* swallow */
411 }
412
e2206a6Test User413 void notifyWebhookDead(d.webhookId, errorMessage ?? `HTTP ${statusCode ?? "?"} after ${MAX_ATTEMPTS} attempts`);
414
8405c43Claude415 return "dead";
416 }
417
418 // Schedule the next attempt.
419 const backoff = BACKOFF_MS[attemptNumber] ?? BACKOFF_MS[BACKOFF_MS.length - 1];
420 const nextAt = new Date(now.getTime() + backoff);
421
422 await db
423 .update(webhookDeliveries)
424 .set({
425 status: "pending",
426 attemptCount: attemptNumber,
427 lastAttemptedAt: now,
428 lastStatusCode: statusCode,
429 lastError: errorMessage,
430 nextAttemptAt: nextAt,
431 })
432 .where(eq(webhookDeliveries.id, deliveryId));
433
434 try {
435 await db
436 .update(webhooks)
437 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
438 .where(eq(webhooks.id, d.webhookId));
439 } catch {
440 /* swallow */
441 }
442
443 return "retry";
444}
445
446// ---------------------------------------------------------------------------
447// Drain — claim up to BATCH_SIZE due rows and attempt them.
448// ---------------------------------------------------------------------------
449
450/** Claim and attempt up to BATCH_SIZE due rows. Returns count attempted. */
451export async function drainPendingDeliveries(): Promise<number> {
452 const now = new Date();
453 let due: { id: string }[] = [];
454 try {
455 due = await db
456 .select({ id: webhookDeliveries.id })
457 .from(webhookDeliveries)
458 .where(
459 and(
460 eq(webhookDeliveries.status, "pending"),
461 lte(webhookDeliveries.nextAttemptAt, now)
462 )
463 )
464 .orderBy(asc(webhookDeliveries.nextAttemptAt))
465 .limit(BATCH_SIZE);
466 } catch (err) {
467 console.error("[webhook-delivery] poll failed:", err);
468 return 0;
469 }
470
471 if (due.length === 0) return 0;
472
473 // Run attempts in parallel — they're IO-bound and target different URLs.
474 await Promise.all(
475 due.map((row) =>
476 attemptDelivery(row.id).catch((err) => {
477 console.error(
478 `[webhook-delivery] attempt for ${row.id} threw:`,
479 err
480 );
481 })
482 )
483 );
484
485 return due.length;
486}
487
488// ---------------------------------------------------------------------------
489// Worker
490// ---------------------------------------------------------------------------
491
492let workerStarted = false;
493
494/**
495 * Background poll loop. Idempotent — calling twice is a no-op. Returns a
496 * stop function (used in tests; production never stops).
497 */
498export function startWebhookDeliveryWorker(opts?: {
499 intervalMs?: number;
500}): () => void {
501 if (workerStarted) return () => {};
502 workerStarted = true;
503
504 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
505 let stopped = false;
506 let active = false;
507
508 const tick = async () => {
509 if (stopped || active) return;
510 active = true;
511 try {
512 // Keep draining while there's work — many due rows can pile up
513 // after an outage of the downstream service.
514 let n = await drainPendingDeliveries();
515 while (n >= BATCH_SIZE && !stopped) {
516 n = await drainPendingDeliveries();
517 }
518 } catch (err) {
519 console.error("[webhook-delivery] worker tick:", err);
520 } finally {
521 active = false;
522 }
523 };
524
525 const handle = setInterval(() => {
526 void tick();
527 }, intervalMs);
528
529 // Best-effort: don't keep the process alive in tests/CLIs.
530 if (typeof (handle as { unref?: () => void }).unref === "function") {
531 (handle as { unref?: () => void }).unref?.();
532 }
533
534 return () => {
535 stopped = true;
536 workerStarted = false;
537 clearInterval(handle);
538 };
539}
540
541// Silence unused-import warnings for `sql` (kept in case future schedule
542// migrations want raw expressions — drizzle's lte/eq cover all current uses).
543void sql;