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.

webhook-delivery.tsBlame401 lines · 1 contributor

Normal view Raw

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