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.tsBlame473 lines · 2 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
70// ---------------------------------------------------------------------------
71// Signing
72// ---------------------------------------------------------------------------
73
74/**
75 * Compute the `sha256=<hex>` HMAC signature for a payload. Returns the empty
76 * string when the hook has no secret — caller should still POST but skip the
77 * `X-Gluecron-Signature` header.
78 */
79export async function computeSignature(
80 secret: string | null,
81 payloadJson: string
82): Promise<string> {
83 if (!secret) return "";
84 const encoder = new TextEncoder();
85 const key = await crypto.subtle.importKey(
86 "raw",
87 encoder.encode(secret),
88 { name: "HMAC", hash: "SHA-256" },
89 false,
90 ["sign"]
91 );
92 const signature = await crypto.subtle.sign(
93 "HMAC",
94 key,
95 encoder.encode(payloadJson)
96 );
97 return (
98 "sha256=" +
99 Array.from(new Uint8Array(signature))
100 .map((b) => b.toString(16).padStart(2, "0"))
101 .join("")
102 );
103}
104
105// ---------------------------------------------------------------------------
106// Enqueue
107// ---------------------------------------------------------------------------
108
109/**
110 * Insert one pending delivery row. Caller passes hook + event + payload; we
111 * snapshot the signature now so future schedule-time secret rotation can't
112 * silently invalidate in-flight retries. Returns the new row id, or null on
113 * insert failure (logged; never throws).
114 */
115export async function enqueueWebhookDelivery(input: {
116 webhookId: string;
117 secret: string | null;
118 event: string;
119 payload: Record<string, unknown>;
120}): Promise<string | null> {
121 try {
122 const payloadJson = JSON.stringify(input.payload);
123 const signature = await computeSignature(input.secret, payloadJson);
124
125 const [row] = await db
126 .insert(webhookDeliveries)
127 .values({
128 webhookId: input.webhookId,
129 event: input.event,
130 payload: payloadJson,
131 signature,
132 attemptCount: 0,
133 nextAttemptAt: new Date(),
134 status: "pending",
135 })
136 .returning({ id: webhookDeliveries.id });
137
138 return row?.id ?? null;
139 } catch (err) {
140 console.error("[webhook-delivery] enqueue failed:", err);
141 return null;
142 }
143}
144
e2206a6Test User145/**
146 * Alert the repo owner that a webhook has permanently stopped delivering —
147 * previously the only signal was a console.error nobody reads (chat-
148 * notifier.ts) or a "no deliveries yet" pill on a settings page nobody was
149 * visiting. Fire-and-forget, in-app notification only; never throws — a bug
150 * here must not affect delivery bookkeeping.
151 */
152async function notifyWebhookDead(webhookId: string, reason: string): Promise<void> {
153 try {
154 const [row] = await db
155 .select({
156 url: webhooks.url,
157 repoName: repositories.name,
158 repoId: repositories.id,
159 ownerId: repositories.ownerId,
160 ownerUsername: users.username,
161 })
162 .from(webhooks)
163 .innerJoin(repositories, eq(repositories.id, webhooks.repositoryId))
164 .innerJoin(users, eq(users.id, repositories.ownerId))
165 .where(eq(webhooks.id, webhookId))
166 .limit(1);
167 if (!row) return;
168
169 await notify(row.ownerId, {
170 kind: "integration_dead",
171 title: `Webhook delivery failed permanently — ${row.repoName}`,
172 body: `A webhook to ${row.url} has stopped retrying (${reason}). It will not be attempted again until you re-check its configuration.`,
173 url: `/${row.ownerUsername}/${row.repoName}/settings/webhooks`,
174 repositoryId: row.repoId,
175 });
176 } catch (err) {
177 console.error("[webhook-delivery] notifyWebhookDead failed:", err);
178 }
179}
180
8405c43Claude181// ---------------------------------------------------------------------------
182// One attempt
183// ---------------------------------------------------------------------------
184
185/**
186 * Run a single delivery attempt against the given row. Looks up the hook URL
187 * fresh (so a deleted hook short-circuits), POSTs, then updates the row to
188 * succeeded / pending (with new next_attempt_at) / dead based on the result.
189 *
190 * Returns 'succeeded' | 'retry' | 'dead' | 'gone' (hook was deleted).
191 */
192export async function attemptDelivery(
193 deliveryId: string
194): Promise<"succeeded" | "retry" | "dead" | "gone"> {
195 // Pull the delivery row + the hook URL in one shot.
196 const rows = await db
197 .select({
198 delivery: webhookDeliveries,
199 url: webhooks.url,
200 isActive: webhooks.isActive,
201 })
202 .from(webhookDeliveries)
203 .leftJoin(webhooks, eq(webhooks.id, webhookDeliveries.webhookId))
204 .where(eq(webhookDeliveries.id, deliveryId))
205 .limit(1);
206
207 const row = rows[0];
208 if (!row) return "gone";
209 if (!row.url || row.isActive === false) {
210 // Hook deleted or disabled between enqueue and attempt — mark dead so
211 // we don't keep polling it.
212 await db
213 .update(webhookDeliveries)
214 .set({
215 status: "dead",
216 lastError: "hook deleted or disabled",
217 lastAttemptedAt: new Date(),
218 })
219 .where(eq(webhookDeliveries.id, deliveryId));
220 return "gone";
221 }
222
223 const d = row.delivery;
224 const attemptNumber = d.attemptCount + 1;
225
bc519aeClaude226 // SSRF guard (BUILD_BIBLE §7): refuse to POST to private/internal
227 // addresses. Permanent condition — retrying won't change the URL — so
228 // the row goes straight to 'dead' with a clear reason. Never throws.
229 const guard = assertPublicUrl(row.url);
230 if (!guard.ok) {
231 const blockedAt = new Date();
232 await db
233 .update(webhookDeliveries)
234 .set({
235 status: "dead",
236 attemptCount: attemptNumber,
237 lastAttemptedAt: blockedAt,
238 lastStatusCode: null,
239 lastError: `blocked: ${guard.reason} (SSRF protection)`,
240 nextAttemptAt: null,
241 })
242 .where(eq(webhookDeliveries.id, deliveryId));
243
244 try {
245 await db
246 .update(webhooks)
247 .set({ lastDeliveredAt: blockedAt, lastStatus: 0 })
248 .where(eq(webhooks.id, d.webhookId));
249 } catch {
250 /* swallow */
251 }
252
e2206a6Test User253 void notifyWebhookDead(d.webhookId, `blocked: ${guard.reason}`);
254
bc519aeClaude255 return "dead";
256 }
257
8405c43Claude258 // Perform the POST.
259 let statusCode: number | null = null;
260 let errorMessage: string | null = null;
261 let success = false;
262
263 try {
264 const headers: Record<string, string> = {
265 "Content-Type": "application/json",
266 "X-Gluecron-Event": d.event,
267 "X-Gluecron-Delivery": d.id,
268 };
269 if (d.signature) headers["X-Gluecron-Signature"] = d.signature;
270
271 const res = await fetch(row.url, {
272 method: "POST",
273 headers,
274 body: d.payload,
275 signal: AbortSignal.timeout(DELIVERY_TIMEOUT_MS),
276 });
277 statusCode = res.status;
278 success = res.status >= 200 && res.status < 300;
279 if (!success) {
280 errorMessage = `HTTP ${res.status}`;
281 }
282 } catch (err) {
283 errorMessage =
284 err instanceof Error ? err.message : String(err ?? "unknown error");
285 if (errorMessage.length > ERROR_CAP) {
286 errorMessage = errorMessage.slice(0, ERROR_CAP);
287 }
288 }
289
290 const now = new Date();
291
292 if (success) {
293 await db
294 .update(webhookDeliveries)
295 .set({
296 status: "succeeded",
297 attemptCount: attemptNumber,
298 lastAttemptedAt: now,
299 lastStatusCode: statusCode,
300 lastError: null,
301 succeededAt: now,
302 nextAttemptAt: null,
303 })
304 .where(eq(webhookDeliveries.id, deliveryId));
305
306 // Best-effort sync of the parent hook row for the legacy "last status"
307 // surface in /settings/webhooks. Never throws out.
308 try {
309 await db
310 .update(webhooks)
311 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
312 .where(eq(webhooks.id, d.webhookId));
313 } catch {
314 /* swallow */
315 }
316
317 return "succeeded";
318 }
319
320 // Failure path.
321 if (attemptNumber >= MAX_ATTEMPTS) {
322 await db
323 .update(webhookDeliveries)
324 .set({
325 status: "dead",
326 attemptCount: attemptNumber,
327 lastAttemptedAt: now,
328 lastStatusCode: statusCode,
329 lastError: errorMessage,
330 nextAttemptAt: null,
331 })
332 .where(eq(webhookDeliveries.id, deliveryId));
333
334 try {
335 await db
336 .update(webhooks)
337 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
338 .where(eq(webhooks.id, d.webhookId));
339 } catch {
340 /* swallow */
341 }
342
e2206a6Test User343 void notifyWebhookDead(d.webhookId, errorMessage ?? `HTTP ${statusCode ?? "?"} after ${MAX_ATTEMPTS} attempts`);
344
8405c43Claude345 return "dead";
346 }
347
348 // Schedule the next attempt.
349 const backoff = BACKOFF_MS[attemptNumber] ?? BACKOFF_MS[BACKOFF_MS.length - 1];
350 const nextAt = new Date(now.getTime() + backoff);
351
352 await db
353 .update(webhookDeliveries)
354 .set({
355 status: "pending",
356 attemptCount: attemptNumber,
357 lastAttemptedAt: now,
358 lastStatusCode: statusCode,
359 lastError: errorMessage,
360 nextAttemptAt: nextAt,
361 })
362 .where(eq(webhookDeliveries.id, deliveryId));
363
364 try {
365 await db
366 .update(webhooks)
367 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
368 .where(eq(webhooks.id, d.webhookId));
369 } catch {
370 /* swallow */
371 }
372
373 return "retry";
374}
375
376// ---------------------------------------------------------------------------
377// Drain — claim up to BATCH_SIZE due rows and attempt them.
378// ---------------------------------------------------------------------------
379
380/** Claim and attempt up to BATCH_SIZE due rows. Returns count attempted. */
381export async function drainPendingDeliveries(): Promise<number> {
382 const now = new Date();
383 let due: { id: string }[] = [];
384 try {
385 due = await db
386 .select({ id: webhookDeliveries.id })
387 .from(webhookDeliveries)
388 .where(
389 and(
390 eq(webhookDeliveries.status, "pending"),
391 lte(webhookDeliveries.nextAttemptAt, now)
392 )
393 )
394 .orderBy(asc(webhookDeliveries.nextAttemptAt))
395 .limit(BATCH_SIZE);
396 } catch (err) {
397 console.error("[webhook-delivery] poll failed:", err);
398 return 0;
399 }
400
401 if (due.length === 0) return 0;
402
403 // Run attempts in parallel — they're IO-bound and target different URLs.
404 await Promise.all(
405 due.map((row) =>
406 attemptDelivery(row.id).catch((err) => {
407 console.error(
408 `[webhook-delivery] attempt for ${row.id} threw:`,
409 err
410 );
411 })
412 )
413 );
414
415 return due.length;
416}
417
418// ---------------------------------------------------------------------------
419// Worker
420// ---------------------------------------------------------------------------
421
422let workerStarted = false;
423
424/**
425 * Background poll loop. Idempotent — calling twice is a no-op. Returns a
426 * stop function (used in tests; production never stops).
427 */
428export function startWebhookDeliveryWorker(opts?: {
429 intervalMs?: number;
430}): () => void {
431 if (workerStarted) return () => {};
432 workerStarted = true;
433
434 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
435 let stopped = false;
436 let active = false;
437
438 const tick = async () => {
439 if (stopped || active) return;
440 active = true;
441 try {
442 // Keep draining while there's work — many due rows can pile up
443 // after an outage of the downstream service.
444 let n = await drainPendingDeliveries();
445 while (n >= BATCH_SIZE && !stopped) {
446 n = await drainPendingDeliveries();
447 }
448 } catch (err) {
449 console.error("[webhook-delivery] worker tick:", err);
450 } finally {
451 active = false;
452 }
453 };
454
455 const handle = setInterval(() => {
456 void tick();
457 }, intervalMs);
458
459 // Best-effort: don't keep the process alive in tests/CLIs.
460 if (typeof (handle as { unref?: () => void }).unref === "function") {
461 (handle as { unref?: () => void }).unref?.();
462 }
463
464 return () => {
465 stopped = true;
466 workerStarted = false;
467 clearInterval(handle);
468 };
469}
470
471// Silence unused-import warnings for `sql` (kept in case future schedule
472// migrations want raw expressions — drizzle's lte/eq cover all current uses).
473void sql;