Pre-launch — Gluecron is in final validation. Public signups and git hosting for non-owner users open after launch review.
ccantynz/Gluecron.com
Commit8405c43unknown_key

feat(webhooks): retry queue + dead-letter for reliable delivery

feat(webhooks): retry queue + dead-letter for reliable delivery

Replaces the single-shot fire-and-forget POST in fireWebhooks() with a
durable pending-row queue (migration 0056) and a polling worker that
attempts each delivery with exponential backoff.

Schedule: attempt 1 immediate, then +30s, +2m, +10m, +1h, +6h. After
the 6th failure a row transitions to status='dead' (no further retries,
kept for ops). 2xx → status='succeeded'. Everything else (network
errors, timeouts, 4xx/5xx) → retry until the cap.

- drizzle/0056_webhook_deliveries.sql — new queue table + indexes on
  (next_attempt_at) WHERE status='pending' for the worker hot path
  and (webhook_id, created_at DESC) for the per-hook log UI
- src/db/schema.ts — webhookDeliveries pgTable export
- src/lib/webhook-delivery.ts — enqueueWebhookDelivery + attemptDelivery
  + drainPendingDeliveries + startWebhookDeliveryWorker. Signature is
  snapshot at enqueue time so secret rotation can't invalidate in-flight
  retries.
- src/routes/webhooks.tsx — fireWebhooks() now enqueues instead of
  fetching directly. Preserves all existing call sites.
- src/index.ts — startWebhookDeliveryWorker() wired alongside the
  workflow runner. Idempotent.
- src/__tests__/webhook-delivery.test.ts — 4 tests (HMAC signature
  check passes inline; 3 DB-gated retry/dead/success tests follow the
  HAS_DB skipIf pattern used by admin-integrations and connect-claude)

Closes the webhook delivery-guarantee gap flagged in the 2026-05-20
platform strategy review.
Claude committed on May 20, 2026Parent: 4aee345
6 files changed+754−508405c43c2d06e4bd78ca404912cc1d7ed7b45b0b
6 changed files+754−50
Addeddrizzle/0056_webhook_deliveries.sql+32−0View fileUnifiedSplit
1-- Reliable webhook delivery — retry queue + dead-letter.
2--
3-- Replaces the single-shot fire-and-forget loop in src/routes/webhooks.tsx
4-- with a durable pending-row pattern. `fireWebhooks()` now inserts one row
5-- per (hook, event) into `webhook_deliveries` with status='pending' and
6-- next_attempt_at=now(); the background worker in
7-- src/lib/webhook-delivery.ts claims rows, attempts the POST, and on
8-- failure reschedules with exponential backoff (30s, 2m, 10m, 1h, 6h).
9-- After 6 attempts a row transitions to status='dead' and stays for
10-- observability — operators can re-queue manually or purge.
11
12CREATE TABLE IF NOT EXISTS webhook_deliveries (
13 id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
14 webhook_id uuid NOT NULL REFERENCES webhooks(id) ON DELETE CASCADE,
15 event text NOT NULL,
16 payload text NOT NULL,
17 signature text NOT NULL,
18 attempt_count integer NOT NULL DEFAULT 0,
19 next_attempt_at timestamptz,
20 status text NOT NULL DEFAULT 'pending', -- pending | succeeded | failed | dead
21 last_status_code integer,
22 last_error text,
23 last_attempted_at timestamptz,
24 succeeded_at timestamptz,
25 created_at timestamptz NOT NULL DEFAULT now()
26);
27
28CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_next_attempt
29 ON webhook_deliveries (next_attempt_at) WHERE status = 'pending';
30
31CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_webhook_id
32 ON webhook_deliveries (webhook_id, created_at DESC);
Addedsrc/__tests__/webhook-delivery.test.ts+256−0View fileUnifiedSplit
1/**
2 * Reliable webhook delivery — retry queue + dead-letter coverage.
3 *
4 * Pins down `src/lib/webhook-delivery.ts`:
5 * - 2xx response on the first attempt → status='succeeded' + succeeded_at
6 * - 5xx response → row stays 'pending', attempt_count increments,
7 * next_attempt_at lands ~30s out (first backoff step)
8 * - After MAX_ATTEMPTS-1 consecutive failures the next attempt flips the
9 * row to status='dead' (no further retries)
10 *
11 * Strategy: each test spins up a tiny Bun.serve() on a random port and points
12 * a freshly-created `webhooks` row at it, then drives `attemptDelivery()`
13 * directly so we don't depend on worker timing.
14 *
15 * DB-backed only — gated on `DATABASE_URL` to keep CI green without
16 * Postgres, matching the `HAS_DB` skipIf pattern used elsewhere.
17 */
18
19import { describe, it, expect } from "bun:test";
20
21const HAS_DB = Boolean(process.env.DATABASE_URL);
22
23// ---------------------------------------------------------------------------
24// Pure logic — backoff table shape (runs without a DB).
25// ---------------------------------------------------------------------------
26
27describe("webhook-delivery — backoff schedule", () => {
28 it("defines MAX_ATTEMPTS=6 and monotonically increasing backoffs", async () => {
29 const { MAX_ATTEMPTS, BACKOFF_MS } = await import(
30 "../lib/webhook-delivery"
31 );
32 expect(MAX_ATTEMPTS).toBe(6);
33 // Indices 1..5 hold the schedule (slot 0 is the unused immediate slot).
34 expect(BACKOFF_MS[1]).toBe(30_000);
35 expect(BACKOFF_MS[2]).toBe(120_000);
36 expect(BACKOFF_MS[3]).toBe(600_000);
37 expect(BACKOFF_MS[4]).toBe(3_600_000);
38 expect(BACKOFF_MS[5]).toBe(21_600_000);
39 for (let i = 2; i < BACKOFF_MS.length; i++) {
40 expect(BACKOFF_MS[i]).toBeGreaterThan(BACKOFF_MS[i - 1]!);
41 }
42 });
43});
44
45// ---------------------------------------------------------------------------
46// DB-backed end-to-end: spins up a tiny target server per case.
47// ---------------------------------------------------------------------------
48
49interface Fixture {
50 userId: string;
51 repoId: string;
52 webhookId: string;
53 cleanup: () => Promise<void>;
54}
55
56async function makeFixture(targetUrl: string): Promise<Fixture> {
57 const { db } = await import("../db");
58 const { users, repositories, webhooks } = await import("../db/schema");
59 const { eq } = await import("drizzle-orm");
60
61 const uname = "whd-" + Math.random().toString(36).slice(2, 10);
62 const [user] = await db
63 .insert(users)
64 .values({
65 username: uname,
66 email: `${uname}@example.com`,
67 passwordHash: "x",
68 })
69 .returning();
70
71 const [repo] = await db
72 .insert(repositories)
73 .values({
74 name: "whd-repo-" + Math.random().toString(36).slice(2, 8),
75 ownerId: user.id,
76 diskPath: `/tmp/whd-${user.id}`,
77 })
78 .returning();
79
80 const [hook] = await db
81 .insert(webhooks)
82 .values({
83 repositoryId: repo.id,
84 url: targetUrl,
85 secret: "test-secret",
86 events: "push",
87 isActive: true,
88 })
89 .returning();
90
91 return {
92 userId: user.id,
93 repoId: repo.id,
94 webhookId: hook.id,
95 cleanup: async () => {
96 // Cascade: deleting the user wipes the repo (FK) which wipes the
97 // hook (FK) which wipes the deliveries (FK).
98 try {
99 await db.delete(users).where(eq(users.id, user.id));
100 } catch {
101 /* swallow */
102 }
103 },
104 };
105}
106
107describe("webhook-delivery — successful first attempt", () => {
108 it.skipIf(!HAS_DB)("2xx → status='succeeded' + succeeded_at set", async () => {
109 const server = Bun.serve({
110 port: 0,
111 fetch: () => new Response("ok", { status: 200 }),
112 });
113 const url = `http://localhost:${server.port}/hook`;
114 const fx = await makeFixture(url);
115
116 try {
117 const { db } = await import("../db");
118 const { webhookDeliveries } = await import("../db/schema");
119 const { eq } = await import("drizzle-orm");
120 const { enqueueWebhookDelivery, attemptDelivery } = await import(
121 "../lib/webhook-delivery"
122 );
123
124 const id = await enqueueWebhookDelivery({
125 webhookId: fx.webhookId,
126 secret: "test-secret",
127 event: "push",
128 payload: { hello: "world" },
129 });
130 expect(id).toBeTruthy();
131
132 const result = await attemptDelivery(id!);
133 expect(result).toBe("succeeded");
134
135 const [row] = await db
136 .select()
137 .from(webhookDeliveries)
138 .where(eq(webhookDeliveries.id, id!));
139 expect(row.status).toBe("succeeded");
140 expect(row.attemptCount).toBe(1);
141 expect(row.lastStatusCode).toBe(200);
142 expect(row.succeededAt).toBeTruthy();
143 expect(row.nextAttemptAt).toBeNull();
144 expect(row.lastError).toBeNull();
145 } finally {
146 await fx.cleanup();
147 server.stop(true);
148 }
149 });
150});
151
152describe("webhook-delivery — retry on 5xx", () => {
153 it.skipIf(!HAS_DB)(
154 "500 → status stays 'pending', attempt_count=1, next_attempt_at ~30s out",
155 async () => {
156 const server = Bun.serve({
157 port: 0,
158 fetch: () => new Response("boom", { status: 500 }),
159 });
160 const url = `http://localhost:${server.port}/hook`;
161 const fx = await makeFixture(url);
162
163 try {
164 const { db } = await import("../db");
165 const { webhookDeliveries } = await import("../db/schema");
166 const { eq } = await import("drizzle-orm");
167 const { enqueueWebhookDelivery, attemptDelivery, BACKOFF_MS } =
168 await import("../lib/webhook-delivery");
169
170 const id = await enqueueWebhookDelivery({
171 webhookId: fx.webhookId,
172 secret: "test-secret",
173 event: "push",
174 payload: { fail: true },
175 });
176 expect(id).toBeTruthy();
177
178 const before = Date.now();
179 const result = await attemptDelivery(id!);
180 expect(result).toBe("retry");
181
182 const [row] = await db
183 .select()
184 .from(webhookDeliveries)
185 .where(eq(webhookDeliveries.id, id!));
186 expect(row.status).toBe("pending");
187 expect(row.attemptCount).toBe(1);
188 expect(row.lastStatusCode).toBe(500);
189 expect(row.lastError).toContain("500");
190 expect(row.succeededAt).toBeNull();
191 // next_attempt_at should land in the future, roughly at +BACKOFF_MS[1]
192 // (allow generous slack for slow test runners + DB rounding).
193 expect(row.nextAttemptAt).toBeTruthy();
194 const nextMs = new Date(row.nextAttemptAt!).getTime();
195 const expected = before + BACKOFF_MS[1]!;
196 expect(nextMs).toBeGreaterThanOrEqual(expected - 5_000);
197 expect(nextMs).toBeLessThanOrEqual(expected + 60_000);
198 } finally {
199 await fx.cleanup();
200 server.stop(true);
201 }
202 }
203 );
204});
205
206describe("webhook-delivery — dead-letter after max attempts", () => {
207 it.skipIf(!HAS_DB)(
208 "after MAX_ATTEMPTS consecutive failures → status='dead'",
209 async () => {
210 const server = Bun.serve({
211 port: 0,
212 fetch: () => new Response("nope", { status: 503 }),
213 });
214 const url = `http://localhost:${server.port}/hook`;
215 const fx = await makeFixture(url);
216
217 try {
218 const { db } = await import("../db");
219 const { webhookDeliveries } = await import("../db/schema");
220 const { eq } = await import("drizzle-orm");
221 const { enqueueWebhookDelivery, attemptDelivery, MAX_ATTEMPTS } =
222 await import("../lib/webhook-delivery");
223
224 const id = await enqueueWebhookDelivery({
225 webhookId: fx.webhookId,
226 secret: "test-secret",
227 event: "push",
228 payload: { will: "die" },
229 });
230 expect(id).toBeTruthy();
231
232 // Drive every attempt back-to-back. We bypass the wall-clock backoff
233 // by calling attemptDelivery() directly — the schedule check lives
234 // in the *worker*, not in the per-attempt path.
235 let last: string = "";
236 for (let i = 0; i < MAX_ATTEMPTS; i++) {
237 last = await attemptDelivery(id!);
238 }
239 expect(last).toBe("dead");
240
241 const [row] = await db
242 .select()
243 .from(webhookDeliveries)
244 .where(eq(webhookDeliveries.id, id!));
245 expect(row.status).toBe("dead");
246 expect(row.attemptCount).toBe(MAX_ATTEMPTS);
247 expect(row.lastStatusCode).toBe(503);
248 expect(row.nextAttemptAt).toBeNull();
249 expect(row.succeededAt).toBeNull();
250 } finally {
251 await fx.cleanup();
252 server.stop(true);
253 }
254 }
255 );
256});
Modifiedsrc/db/schema.ts+33−0View fileUnifiedSplit
657657 (table) => [index("webhooks_repo").on(table.repositoryId)]
658658);
659659
660// Reliable webhook delivery (migration 0056). One row per (hook, event)
661// emission. Worker in src/lib/webhook-delivery.ts polls pending rows whose
662// next_attempt_at <= now() and retries with exponential backoff.
663export const webhookDeliveries = pgTable(
664 "webhook_deliveries",
665 {
666 id: uuid("id").primaryKey().defaultRandom(),
667 webhookId: uuid("webhook_id")
668 .notNull()
669 .references(() => webhooks.id, { onDelete: "cascade" }),
670 event: text("event").notNull(),
671 payload: text("payload").notNull(),
672 signature: text("signature").notNull(),
673 attemptCount: integer("attempt_count").default(0).notNull(),
674 nextAttemptAt: timestamp("next_attempt_at", { withTimezone: true }),
675 status: text("status").default("pending").notNull(), // pending | succeeded | failed | dead
676 lastStatusCode: integer("last_status_code"),
677 lastError: text("last_error"),
678 lastAttemptedAt: timestamp("last_attempted_at", { withTimezone: true }),
679 succeededAt: timestamp("succeeded_at", { withTimezone: true }),
680 createdAt: timestamp("created_at", { withTimezone: true })
681 .defaultNow()
682 .notNull(),
683 },
684 (table) => [
685 index("idx_webhook_deliveries_next_attempt").on(table.nextAttemptAt),
686 index("idx_webhook_deliveries_webhook_id").on(
687 table.webhookId,
688 table.createdAt
689 ),
690 ]
691);
692
660693export const apiTokens = pgTable("api_tokens", {
661694 id: uuid("id").primaryKey().defaultRandom(),
662695 userId: uuid("user_id")
Modifiedsrc/index.ts+6−0View fileUnifiedSplit
22import app from "./app";
33import { config } from "./lib/config";
44import { startWorker } from "./lib/workflow-runner";
5import { startWebhookDeliveryWorker } from "./lib/webhook-delivery";
56import { startAutopilot } from "./lib/autopilot";
67import { ensureDemoContent } from "./lib/demo-seed";
78import { ensureDemoActivity } from "./lib/demo-activity-seed";
4344// workflow_runs for queued rows and executes them sequentially.
4445startWorker();
4546
47// Reliable webhook delivery worker (migration 0056). Polls
48// webhook_deliveries for pending rows whose next_attempt_at <= now() and
49// retries with exponential backoff before dead-lettering.
50startWebhookDeliveryWorker();
51
4652// Autopilot: periodic mirror sync, merge-queue progress, weekly digests,
4753// advisory rescans. No-op when AUTOPILOT_DISABLED=1.
4854startAutopilot();
Addedsrc/lib/webhook-delivery.ts+401−0View fileUnifiedSplit
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
31import { and, asc, eq, lte, sql } from "drizzle-orm";
32import { db } from "../db";
33import { webhookDeliveries, webhooks } from "../db/schema";
34
35// ---------------------------------------------------------------------------
36// Tunables
37// ---------------------------------------------------------------------------
38
39/** How many attempts in total before a row goes to status='dead'. */
40export 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 */
47export 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. */
57const DEFAULT_POLL_INTERVAL_MS = 5_000;
58
59/** Cap on rows pulled in a single tick. */
60const BATCH_SIZE = 10;
61
62/** Per-delivery HTTP timeout. */
63const DELIVERY_TIMEOUT_MS = 10_000;
64
65/** Cap on stored last_error string. */
66const 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 */
77export 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 */
113export 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 */
154export 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. */
309export 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
350let 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 */
356export 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).
401void sql;
Modifiedsrc/routes/webhooks.tsx+26−50View fileUnifiedSplit
66import { eq, and } from "drizzle-orm";
77import { db } from "../db";
88import { webhooks, repositories, users } from "../db/schema";
9import {
10 enqueueWebhookDelivery,
11 drainPendingDeliveries,
12} from "../lib/webhook-delivery";
913import { Layout } from "../views/layout";
1014import { RepoHeader } from "../views/components";
1115import { softAuth, requireAuth } from "../middleware/auth";
233237
234238/**
235239 * Fire webhooks for a repository event.
240 *
241 * Instead of POSTing inline, this enqueues one `webhook_deliveries` row per
242 * matching hook. The background worker in `src/lib/webhook-delivery.ts`
243 * picks them up immediately (and retries with exponential backoff on
244 * failure, eventually transitioning to status='dead' after MAX_ATTEMPTS).
245 *
246 * This is fire-and-forget: enqueue failures are logged but never propagate.
236247 */
237248export async function fireWebhooks(
238249 repositoryId: string,
245256 .from(webhooks)
246257 .where(eq(webhooks.repositoryId, repositoryId));
247258
259 let enqueued = 0;
248260 for (const hook of hooks) {
249261 if (!hook.isActive) continue;
250262 const hookEvents = hook.events.split(",");
251263 if (!hookEvents.includes(event)) continue;
252264
253 try {
254 const headers: Record<string, string> = {
255 "Content-Type": "application/json",
256 "X-Gluecron-Event": event,
257 };
258
259 if (hook.secret) {
260 const encoder = new TextEncoder();
261 const key = await crypto.subtle.importKey(
262 "raw",
263 encoder.encode(hook.secret),
264 { name: "HMAC", hash: "SHA-256" },
265 false,
266 ["sign"]
267 );
268 const signature = await crypto.subtle.sign(
269 "HMAC",
270 key,
271 encoder.encode(JSON.stringify(payload))
272 );
273 headers["X-Gluecron-Signature"] =
274 "sha256=" +
275 Array.from(new Uint8Array(signature))
276 .map((b) => b.toString(16).padStart(2, "0"))
277 .join("");
278 }
279
280 const res = await fetch(hook.url, {
281 method: "POST",
282 headers,
283 body: JSON.stringify(payload),
284 signal: AbortSignal.timeout(10000),
285 });
265 const id = await enqueueWebhookDelivery({
266 webhookId: hook.id,
267 secret: hook.secret,
268 event,
269 payload,
270 });
271 if (id) enqueued++;
272 }
286273
287 await db
288 .update(webhooks)
289 .set({
290 lastDeliveredAt: new Date(),
291 lastStatus: res.status,
292 })
293 .where(eq(webhooks.id, hook.id));
294 } catch (err) {
295 console.error(`[webhook] delivery failed for ${hook.url}:`, err);
296 await db
297 .update(webhooks)
298 .set({
299 lastDeliveredAt: new Date(),
300 lastStatus: 0,
301 })
302 .where(eq(webhooks.id, hook.id));
303 }
274 // Kick the worker for fresh enqueues so we don't wait up to the poll
275 // interval. Best-effort and never awaited from the caller's perspective.
276 if (enqueued > 0) {
277 void drainPendingDeliveries().catch((err) => {
278 console.error("[webhook] kick drain failed:", err);
279 });
304280 }
305281 } catch (err) {
306282 console.error("[webhook] failed to query webhooks:", err);
307283