Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
CodeIssuesDiscussionsWikiPull RequestsProjectsCommitsActionsReleasesContributorsPulse● GatesSecuritySettingsDeploymentsPipelineInsightsAgents✨ Explain✨ Ask AI✨ Workspace✨ Spec✨ Tests▓ Debt Map✨ NL Search🏛 Archaeology
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.tsBlame432 lines · 1 contributor
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";
33import { webhookDeliveries, webhooks } from "../db/schema";
bc519aeClaude34import { assertPublicUrl } from "./ssrf-guard";
8405c43Claude35
36// ---------------------------------------------------------------------------
37// Tunables
38// ---------------------------------------------------------------------------
39
40/** How many attempts in total before a row goes to status='dead'. */
41export const MAX_ATTEMPTS = 6;
42
43/**
44 * Backoff schedule, indexed by the *next* attempt number we're scheduling.
45 * After attemptCount=N fails, we schedule attempt N+1 at now() + BACKOFF_MS[N].
46 * Index 0 is unused (attempt 1 is queued at now() by enqueue, not by retry).
47 */
48export const BACKOFF_MS: number[] = [
49 0, // [0] unused
50 30_000, // after attempt 1 fails → +30s for attempt 2
51 120_000, // after attempt 2 fails → +2m for attempt 3
52 600_000, // after attempt 3 fails → +10m for attempt 4
53 3_600_000, // after attempt 4 fails → +1h for attempt 5
54 21_600_000, // after attempt 5 fails → +6h for attempt 6
55];
56
57/** How often the worker scans for due pending rows. */
58const DEFAULT_POLL_INTERVAL_MS = 5_000;
59
60/** Cap on rows pulled in a single tick. */
61const BATCH_SIZE = 10;
62
63/** Per-delivery HTTP timeout. */
64const DELIVERY_TIMEOUT_MS = 10_000;
65
66/** Cap on stored last_error string. */
67const ERROR_CAP = 2_000;
68
69// ---------------------------------------------------------------------------
70// Signing
71// ---------------------------------------------------------------------------
72
73/**
74 * Compute the `sha256=<hex>` HMAC signature for a payload. Returns the empty
75 * string when the hook has no secret — caller should still POST but skip the
76 * `X-Gluecron-Signature` header.
77 */
78export async function computeSignature(
79 secret: string | null,
80 payloadJson: string
81): Promise<string> {
82 if (!secret) return "";
83 const encoder = new TextEncoder();
84 const key = await crypto.subtle.importKey(
85 "raw",
86 encoder.encode(secret),
87 { name: "HMAC", hash: "SHA-256" },
88 false,
89 ["sign"]
90 );
91 const signature = await crypto.subtle.sign(
92 "HMAC",
93 key,
94 encoder.encode(payloadJson)
95 );
96 return (
97 "sha256=" +
98 Array.from(new Uint8Array(signature))
99 .map((b) => b.toString(16).padStart(2, "0"))
100 .join("")
101 );
102}
103
104// ---------------------------------------------------------------------------
105// Enqueue
106// ---------------------------------------------------------------------------
107
108/**
109 * Insert one pending delivery row. Caller passes hook + event + payload; we
110 * snapshot the signature now so future schedule-time secret rotation can't
111 * silently invalidate in-flight retries. Returns the new row id, or null on
112 * insert failure (logged; never throws).
113 */
114export async function enqueueWebhookDelivery(input: {
115 webhookId: string;
116 secret: string | null;
117 event: string;
118 payload: Record<string, unknown>;
119}): Promise<string | null> {
120 try {
121 const payloadJson = JSON.stringify(input.payload);
122 const signature = await computeSignature(input.secret, payloadJson);
123
124 const [row] = await db
125 .insert(webhookDeliveries)
126 .values({
127 webhookId: input.webhookId,
128 event: input.event,
129 payload: payloadJson,
130 signature,
131 attemptCount: 0,
132 nextAttemptAt: new Date(),
133 status: "pending",
134 })
135 .returning({ id: webhookDeliveries.id });
136
137 return row?.id ?? null;
138 } catch (err) {
139 console.error("[webhook-delivery] enqueue failed:", err);
140 return null;
141 }
142}
143
144// ---------------------------------------------------------------------------
145// One attempt
146// ---------------------------------------------------------------------------
147
148/**
149 * Run a single delivery attempt against the given row. Looks up the hook URL
150 * fresh (so a deleted hook short-circuits), POSTs, then updates the row to
151 * succeeded / pending (with new next_attempt_at) / dead based on the result.
152 *
153 * Returns 'succeeded' | 'retry' | 'dead' | 'gone' (hook was deleted).
154 */
155export async function attemptDelivery(
156 deliveryId: string
157): Promise<"succeeded" | "retry" | "dead" | "gone"> {
158 // Pull the delivery row + the hook URL in one shot.
159 const rows = await db
160 .select({
161 delivery: webhookDeliveries,
162 url: webhooks.url,
163 isActive: webhooks.isActive,
164 })
165 .from(webhookDeliveries)
166 .leftJoin(webhooks, eq(webhooks.id, webhookDeliveries.webhookId))
167 .where(eq(webhookDeliveries.id, deliveryId))
168 .limit(1);
169
170 const row = rows[0];
171 if (!row) return "gone";
172 if (!row.url || row.isActive === false) {
173 // Hook deleted or disabled between enqueue and attempt — mark dead so
174 // we don't keep polling it.
175 await db
176 .update(webhookDeliveries)
177 .set({
178 status: "dead",
179 lastError: "hook deleted or disabled",
180 lastAttemptedAt: new Date(),
181 })
182 .where(eq(webhookDeliveries.id, deliveryId));
183 return "gone";
184 }
185
186 const d = row.delivery;
187 const attemptNumber = d.attemptCount + 1;
188
bc519aeClaude189 // SSRF guard (BUILD_BIBLE §7): refuse to POST to private/internal
190 // addresses. Permanent condition — retrying won't change the URL — so
191 // the row goes straight to 'dead' with a clear reason. Never throws.
192 const guard = assertPublicUrl(row.url);
193 if (!guard.ok) {
194 const blockedAt = new Date();
195 await db
196 .update(webhookDeliveries)
197 .set({
198 status: "dead",
199 attemptCount: attemptNumber,
200 lastAttemptedAt: blockedAt,
201 lastStatusCode: null,
202 lastError: `blocked: ${guard.reason} (SSRF protection)`,
203 nextAttemptAt: null,
204 })
205 .where(eq(webhookDeliveries.id, deliveryId));
206
207 try {
208 await db
209 .update(webhooks)
210 .set({ lastDeliveredAt: blockedAt, lastStatus: 0 })
211 .where(eq(webhooks.id, d.webhookId));
212 } catch {
213 /* swallow */
214 }
215
216 return "dead";
217 }
218
8405c43Claude219 // Perform the POST.
220 let statusCode: number | null = null;
221 let errorMessage: string | null = null;
222 let success = false;
223
224 try {
225 const headers: Record<string, string> = {
226 "Content-Type": "application/json",
227 "X-Gluecron-Event": d.event,
228 "X-Gluecron-Delivery": d.id,
229 };
230 if (d.signature) headers["X-Gluecron-Signature"] = d.signature;
231
232 const res = await fetch(row.url, {
233 method: "POST",
234 headers,
235 body: d.payload,
236 signal: AbortSignal.timeout(DELIVERY_TIMEOUT_MS),
237 });
238 statusCode = res.status;
239 success = res.status >= 200 && res.status < 300;
240 if (!success) {
241 errorMessage = `HTTP ${res.status}`;
242 }
243 } catch (err) {
244 errorMessage =
245 err instanceof Error ? err.message : String(err ?? "unknown error");
246 if (errorMessage.length > ERROR_CAP) {
247 errorMessage = errorMessage.slice(0, ERROR_CAP);
248 }
249 }
250
251 const now = new Date();
252
253 if (success) {
254 await db
255 .update(webhookDeliveries)
256 .set({
257 status: "succeeded",
258 attemptCount: attemptNumber,
259 lastAttemptedAt: now,
260 lastStatusCode: statusCode,
261 lastError: null,
262 succeededAt: now,
263 nextAttemptAt: null,
264 })
265 .where(eq(webhookDeliveries.id, deliveryId));
266
267 // Best-effort sync of the parent hook row for the legacy "last status"
268 // surface in /settings/webhooks. Never throws out.
269 try {
270 await db
271 .update(webhooks)
272 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
273 .where(eq(webhooks.id, d.webhookId));
274 } catch {
275 /* swallow */
276 }
277
278 return "succeeded";
279 }
280
281 // Failure path.
282 if (attemptNumber >= MAX_ATTEMPTS) {
283 await db
284 .update(webhookDeliveries)
285 .set({
286 status: "dead",
287 attemptCount: attemptNumber,
288 lastAttemptedAt: now,
289 lastStatusCode: statusCode,
290 lastError: errorMessage,
291 nextAttemptAt: null,
292 })
293 .where(eq(webhookDeliveries.id, deliveryId));
294
295 try {
296 await db
297 .update(webhooks)
298 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
299 .where(eq(webhooks.id, d.webhookId));
300 } catch {
301 /* swallow */
302 }
303
304 return "dead";
305 }
306
307 // Schedule the next attempt.
308 const backoff = BACKOFF_MS[attemptNumber] ?? BACKOFF_MS[BACKOFF_MS.length - 1];
309 const nextAt = new Date(now.getTime() + backoff);
310
311 await db
312 .update(webhookDeliveries)
313 .set({
314 status: "pending",
315 attemptCount: attemptNumber,
316 lastAttemptedAt: now,
317 lastStatusCode: statusCode,
318 lastError: errorMessage,
319 nextAttemptAt: nextAt,
320 })
321 .where(eq(webhookDeliveries.id, deliveryId));
322
323 try {
324 await db
325 .update(webhooks)
326 .set({ lastDeliveredAt: now, lastStatus: statusCode ?? 0 })
327 .where(eq(webhooks.id, d.webhookId));
328 } catch {
329 /* swallow */
330 }
331
332 return "retry";
333}
334
335// ---------------------------------------------------------------------------
336// Drain — claim up to BATCH_SIZE due rows and attempt them.
337// ---------------------------------------------------------------------------
338
339/** Claim and attempt up to BATCH_SIZE due rows. Returns count attempted. */
340export async function drainPendingDeliveries(): Promise<number> {
341 const now = new Date();
342 let due: { id: string }[] = [];
343 try {
344 due = await db
345 .select({ id: webhookDeliveries.id })
346 .from(webhookDeliveries)
347 .where(
348 and(
349 eq(webhookDeliveries.status, "pending"),
350 lte(webhookDeliveries.nextAttemptAt, now)
351 )
352 )
353 .orderBy(asc(webhookDeliveries.nextAttemptAt))
354 .limit(BATCH_SIZE);
355 } catch (err) {
356 console.error("[webhook-delivery] poll failed:", err);
357 return 0;
358 }
359
360 if (due.length === 0) return 0;
361
362 // Run attempts in parallel — they're IO-bound and target different URLs.
363 await Promise.all(
364 due.map((row) =>
365 attemptDelivery(row.id).catch((err) => {
366 console.error(
367 `[webhook-delivery] attempt for ${row.id} threw:`,
368 err
369 );
370 })
371 )
372 );
373
374 return due.length;
375}
376
377// ---------------------------------------------------------------------------
378// Worker
379// ---------------------------------------------------------------------------
380
381let workerStarted = false;
382
383/**
384 * Background poll loop. Idempotent — calling twice is a no-op. Returns a
385 * stop function (used in tests; production never stops).
386 */
387export function startWebhookDeliveryWorker(opts?: {
388 intervalMs?: number;
389}): () => void {
390 if (workerStarted) return () => {};
391 workerStarted = true;
392
393 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
394 let stopped = false;
395 let active = false;
396
397 const tick = async () => {
398 if (stopped || active) return;
399 active = true;
400 try {
401 // Keep draining while there's work — many due rows can pile up
402 // after an outage of the downstream service.
403 let n = await drainPendingDeliveries();
404 while (n >= BATCH_SIZE && !stopped) {
405 n = await drainPendingDeliveries();
406 }
407 } catch (err) {
408 console.error("[webhook-delivery] worker tick:", err);
409 } finally {
410 active = false;
411 }
412 };
413
414 const handle = setInterval(() => {
415 void tick();
416 }, intervalMs);
417
418 // Best-effort: don't keep the process alive in tests/CLIs.
419 if (typeof (handle as { unref?: () => void }).unref === "function") {
420 (handle as { unref?: () => void }).unref?.();
421 }
422
423 return () => {
424 stopped = true;
425 workerStarted = false;
426 clearInterval(handle);
427 };
428}
429
430// Silence unused-import warnings for `sql` (kept in case future schedule
431// migrations want raw expressions — drizzle's lte/eq cover all current uses).
432void sql;