/ opencodehub / tests / unit / webhook-queue.test.ts
swadhinbiswas feat: finalize platform polish - discussions, gists, webhook queue, realtime, push-mirror, auth/metrics hardening · 8/28/2026
f02976d
415 lines · 13.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
/**
 * Unit: Webhook delivery queue (src/lib/webhooks.ts)
 *
 * Contract after queue decoupling:
 * - triggerWebhooks only ENQUEUES pending delivery rows (no HTTP dispatch).
 * - deliverWebhookDelivery performs exactly one attempt and transitions the
 *   row: delivered / pending-with-backoff / dead.
 * - Retries follow the 1s→16s exponential backoff schedule and die after
 *   WEBHOOK_MAX_RETRIES retries (+initial attempt).
 * - 4xx responses are non-retryable and dead-letter immediately.
 * - processWebhookQueue claims rows atomically (FOR UPDATE SKIP LOCKED) and
 *   requeues locks stuck beyond STALE_WEBHOOK_LOCK_SECS.
 */
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";

const { validateWebhookUrlMock, isOfflineModeMock } = vi.hoisted(() => ({
  validateWebhookUrlMock: vi.fn(async (): Promise<{ valid: boolean; reason?: string }> => ({ valid: true })),
  isOfflineModeMock: vi.fn(() => false),
}));

vi.mock("@/db", () => ({
  getDatabase: () => mockDb.db,
  schema: {
    webhooks: { deliveryCount: "__col_delivery_count__" },
    webhookDeliveries: {},
  },
}));

vi.mock("@/lib/url-validator", () => ({
  validateWebhookUrl: validateWebhookUrlMock,
}));

vi.mock("@/lib/config", () => ({
  isOfflineMode: isOfflineModeMock,
}));

import { schema } from "@/db";
import { PgDialect } from "drizzle-orm/pg-core";
import {
  deliverWebhookDelivery,
  processWebhookQueue,
  reclaimStaleLocks,
  triggerWebhooks,
} from "@/lib/webhooks";

const WEBHOOKS_TABLE = Symbol("webhooks");
const DELIVERIES_TABLE = Symbol("webhookDeliveries");
// The mocked schema above hands the lib marker objects; mirror their identity
// so the fake db can tell updates apart.
(schema as any).webhooks = { ...(schema as any).webhooks, __table: WEBHOOKS_TABLE };
(schema as any).webhookDeliveries = { __table: DELIVERIES_TABLE };

interface UpdateRecord {
  table: symbol;
  values: Record<string, any>;
}

function makeDb(options: {
  hooks?: any[];
  /** FIFO results served by webhookDeliveries.findFirst */
  deliveries?: any[];
  /** ids returned by the atomic claim statement */
  claimIds?: string[];
  /** ids returned by the stale-sweep .returning() */
  reclaimedIds?: string[];
} = {}) {
  const state = {
    insertedDeliveries: [] as any[],
    deliveryUpdates: [] as UpdateRecord[],
    webhookUpdates: [] as UpdateRecord[],
    claimStatements: [] as string[],
  };

  function recordUpdate(table: any, values: Record<string, any>) {
    const record = { table, values };
    if (table === (schema as any).webhookDeliveries) state.deliveryUpdates.push(record);
    else if (table === (schema as any).webhooks) state.webhookUpdates.push(record);
    return record;
  }

  const db = {
    query: {
      webhooks: {
        findMany: async () => options.hooks ?? [],
        findFirst: async () => options.hooks?.[0] ?? null,
      },
      webhookDeliveries: {
        findFirst: async () => {
          const next = options.deliveries?.shift();
          if (next instanceof Error) throw next; // simulate a worker crash
          return next ?? null;
        },
      },
    },
    insert: (_table: any) => ({
      values: async (rows: any) => {
        for (const row of Array.isArray(rows) ? rows : [rows]) {
          state.insertedDeliveries.push(row);
        }
        return {};
      },
    }),
    update: (table: any) => ({
      set: (values: Record<string, any>) => ({
        where: () => {
          const record = recordUpdate(table, values);
          return {
            returning: async () => {
              if (table === (schema as any).webhookDeliveries && values.lockedAt === null && values.status === "pending") {
                return (options.reclaimedIds ?? []).map((id) => ({ id }));
              }
              return [];
            },
            then: (resolve: (v: any) => void, reject: (e: any) => void) =>
              Promise.resolve({}).then(() => resolve({}), reject),
          } as any;
        },
      }),
    }),
    execute: async (query: any) => {
      state.claimStatements.push(new PgDialect().sqlToQuery(query).sql);
      return { rows: (options.claimIds ?? []).map((id) => ({ id })) };
    },
  };

  void recordUpdate;
  return { db, state };
}

let mockDb: ReturnType<typeof makeDb>;

const hookRow = {
  id: "hook-1",
  repositoryId: "repo-1",
  url: "https://receiver.example.com/hook",
  secret: "topsecret",
  events: ["push"],
  active: true,
};

const starHookRow = {
  ...hookRow,
  id: "hook-star",
  events: ["*"],
};

const otherHookRow = {
  ...hookRow,
  id: "hook-other",
  events: ["issues"],
};

function pendingDelivery(overrides: Partial<any> = {}) {
  return {
    id: "delivery-1",
    webhookId: "hook-1",
    event: "push",
    payload: JSON.stringify({ ref: "refs/heads/main" }),
    status: "pending",
    attempts: 0,
    nextAttemptAt: new Date(Date.now() - 1000),
    lockedAt: null,
    failureReason: null,
    responseCode: null,
    responseBody: null,
    durationMs: null,
    error: null,
    requestHeaders: null,
    responseHeaders: null,
    ...overrides,
  };
}

function stubFetch(status: number, body = "") {
  const fetchMock = vi.fn(async () =>
    new Response(body, { status, headers: { "x-reply": "yes" } }),
  );
  vi.stubGlobal("fetch", fetchMock);
  return fetchMock;
}

beforeEach(() => {
  mockDb = makeDb();
});

afterEach(() => {
  vi.unstubAllGlobals();
  vi.unstubAllEnvs();
  vi.clearAllMocks();
});

describe("triggerWebhooks (enqueue-only)", () => {
  it("enqueues one pending delivery per matching hook and performs no HTTP dispatch", async () => {
    const fetchMock = stubFetch(200);
    mockDb = makeDb({ hooks: [hookRow, starHookRow, otherHookRow] });

    const enqueued = await triggerWebhooks("repo-1", "push", { after: "abc" });

    expect(enqueued).toBe(2); // push-subscriber + wildcard; 'issues'-only hook excluded
    expect(mockDb.state.insertedDeliveries).toHaveLength(2);

    for (const row of mockDb.state.insertedDeliveries) {
      expect(row.status).toBe("pending");
      expect(row.attempts).toBe(0);
      expect(row.event).toBe("push");
      expect(row.nextAttemptAt).toBeInstanceOf(Date);
      expect(row.webhookId).not.toBe("hook-other");
      expect(JSON.parse(row.payload)).toEqual({ after: "abc" });
      expect(row.id).toBeTruthy();
    }
    expect(fetchMock).not.toHaveBeenCalled();
  });

  it("returns 0 and inserts nothing when no hooks subscribe to the event", async () => {
    mockDb = makeDb({ hooks: [otherHookRow] });
    const enqueued = await triggerWebhooks("repo-1", "push", {});
    expect(enqueued).toBe(0);
    expect(mockDb.state.insertedDeliveries).toHaveLength(0);
  });
});

describe("deliverWebhookDelivery", () => {
  it("marks the row delivered on 2xx, records response fields and bumps stats", async () => {
    const fetchMock = stubFetch(200, "ok");
    mockDb = makeDb({
      hooks: [hookRow],
      deliveries: [pendingDelivery()],
    });

    const outcome = await deliverWebhookDelivery("delivery-1");

    expect(outcome).toBe("delivered");
    expect(fetchMock).toHaveBeenCalledTimes(1);

    const update = mockDb.state.deliveryUpdates[0];
    expect(update.values.status).toBe("delivered");
    expect(update.values.attempts).toBe(1);
    expect(update.values.responseCode).toBe(200);
    expect(update.values.responseBody).toBe("ok");
    expect(update.values.error).toBeNull();
    expect(update.values.lockedAt).toBeNull();

    const headers = JSON.parse(update.values.requestHeaders);
    expect(headers["X-Hub-Signature-256"]).toMatch(/^sha256=[0-9a-f]{64}$/);
    expect(headers["X-OpenCodeHub-Delivery"]).toBe("delivery-1");
    expect(headers["X-OpenCodeHub-Event"]).toBe("push");

    const stats = mockDb.state.webhookUpdates[0];
    expect(stats.values.lastDeliveryStatus).toBe("success");
    expect(stats.values.lastDeliveryAt).toBeInstanceOf(Date);
    // Atomic COALESCE increment referencing the deliveryCount column
    expect(JSON.stringify(stats.values.deliveryCount)).toContain(
      "__col_delivery_count__",
    );
  });

  it("signs with X-Hub-Signature-256 only when a secret is configured", async () => {
    stubFetch(200);
    mockDb = makeDb({
      hooks: [{ ...hookRow, secret: null }],
      deliveries: [pendingDelivery()],
    });

    await deliverWebhookDelivery("delivery-1");

    const headers = JSON.parse(mockDb.state.deliveryUpdates[0].values.requestHeaders);
    expect(headers["X-Hub-Signature-256"]).toBeUndefined();
  });

  it("retries failures with exponential backoff, then dies after max attempts", async () => {
    vi.stubEnv("WEBHOOK_MAX_RETRIES", "1"); // total attempts = 2
    const fetchMock = stubFetch(500);
    const delivery = pendingDelivery();
    mockDb = makeDb({ hooks: [hookRow], deliveries: [delivery] });

    const first = await deliverWebhookDelivery("delivery-1");
    expect(first).toBe("retrying");
    expect(fetchMock).toHaveBeenCalledTimes(1);

    const retryValues = mockDb.state.deliveryUpdates[0].values;
    expect(retryValues.status).toBe("pending");
    expect(retryValues.attempts).toBe(1);
    expect(retryValues.lockedAt).toBeNull();
    expect(retryValues.failureReason).toBeNull();
    const delay = retryValues.nextAttemptAt.getTime() - Date.now();
    expect(delay).toBeGreaterThanOrEqual(900); // 1s backoff after 1st failure
    expect(delay).toBeLessThanOrEqual(2500);
    expect(retryValues.error).toContain("HTTP 500");

    // Second (final allowed) attempt fails → dead letter with reason
    mockDb = makeDb({
      hooks: [hookRow],
      deliveries: [pendingDelivery({ attempts: 1 })],
    });
    const second = await deliverWebhookDelivery("delivery-1");

    expect(second).toBe("dead");
    const deadValues = mockDb.state.deliveryUpdates[0].values;
    expect(deadValues.status).toBe("dead");
    expect(deadValues.attempts).toBe(2);
    expect(deadValues.failureReason).toBe("Failed after 2/2 attempts");
    expect(deadValues.lockedAt).toBeNull();
  });

  it("dead-letters 4xx responses immediately (non-retryable)", async () => {
    const fetchMock = stubFetch(422);
    mockDb = makeDb({
      hooks: [hookRow],
      deliveries: [pendingDelivery()],
    });

    const outcome = await deliverWebhookDelivery("delivery-1");

    expect(outcome).toBe("dead");
    expect(fetchMock).toHaveBeenCalledTimes(1); // no retry scheduled
    const deadValues = mockDb.state.deliveryUpdates[0].values;
    expect(deadValues.status).toBe("dead");
    expect(deadValues.attempts).toBe(1);
    expect(deadValues.failureReason).toContain("Not retryable");
    expect(deadValues.failureReason).toContain("422");
    expect(deadValues.responseCode).toBe(422);
    expect(deadValues.nextAttemptAt).toBeUndefined(); // no reschedule
  });

  it("dead-letters rows whose webhook was deleted", async () => {
    stubFetch(200);
    mockDb = makeDb({ hooks: [], deliveries: [pendingDelivery()] });

    const outcome = await deliverWebhookDelivery("delivery-1");

    expect(outcome).toBe("dead");
    expect(mockDb.state.deliveryUpdates[0].values.failureReason).toContain("no longer exist");
  });

  it("skips terminal rows without touching them", async () => {
    stubFetch(200);
    mockDb = makeDb({
      hooks: [hookRow],
      deliveries: [pendingDelivery({ status: "delivered" })],
    });

    const outcome = await deliverWebhookDelivery("delivery-1");

    expect(outcome).toBe("skipped");
    expect(mockDb.state.deliveryUpdates).toHaveLength(0);
    expect(mockDb.state.webhookUpdates).toHaveLength(0);
  });
});

describe("processWebhookQueue", () => {
  it("claims due rows atomically with FOR UPDATE SKIP LOCKED and delivers them", async () => {
    stubFetch(200);
    mockDb = makeDb({
      hooks: [hookRow],
      deliveries: [pendingDelivery({ id: "d1" }), pendingDelivery({ id: "d2" })],
      claimIds: ["d1", "d2"],
    });

    const result = await processWebhookQueue(20);

    expect(result.claimed).toBe(2);
    expect(result.delivered).toBe(2);
    expect(result.dead).toBe(0);
    expect(mockDb.state.claimStatements).toHaveLength(1);
    expect(mockDb.state.claimStatements[0]).toContain("FOR UPDATE SKIP LOCKED");
    expect(mockDb.state.claimStatements[0]).toContain("status = 'pending'");
    expect(mockDb.state.claimStatements[0]).toContain("next_attempt_at <= now()");
    const deliveredUpdates = mockDb.state.deliveryUpdates.filter(
      (u) => u.values.status === "delivered",
    );
    expect(deliveredUpdates).toHaveLength(2);
    // The sweep ran first and requeued nothing (no reclaimedIds configured)
    expect(mockDb.state.deliveryUpdates[0].values.status).toBe("pending");
    expect(mockDb.state.deliveryUpdates[0].values.lockedAt).toBeNull();
  });

  it("keeps delivering remaining claimed rows when one delivery crashes", async () => {
    const fetchMock = stubFetch(200);
    mockDb = makeDb({
      hooks: [hookRow],
      deliveries: [
        new Error("simulated worker crash"), // d-crash: findFirst explodes → caught by queue loop
        pendingDelivery({ id: "d-ok" }),
      ],
      claimIds: ["d-crash", "d-ok"],
    });

    const result = await processWebhookQueue(10);

    // The crashed row is not counted anywhere — it stays 'processing' until
    // the stale-lock sweep requeues it; the healthy row still gets delivered.
    expect(result.claimed).toBe(2);
    expect(result.delivered).toBe(1);
    expect(result.dead).toBe(0);
    expect(fetchMock).toHaveBeenCalledTimes(1);
  });

  it("requeues stale processing locks before claiming", async () => {
    stubFetch(200);
    mockDb = makeDb({
      hooks: [hookRow],
      deliveries: [pendingDelivery()],
      claimIds: ["delivery-1"],
      reclaimedIds: ["stale-1", "stale-2"],
    });

    const swept = await reclaimStaleLocks();

    expect(swept).toBe(2);
    expect(mockDb.state.deliveryUpdates).toHaveLength(1);
    expect(mockDb.state.deliveryUpdates[0].values.status).toBe("pending");
    expect(mockDb.state.deliveryUpdates[0].values.lockedAt).toBeNull();
  });
});