diff --git a/backend/src/migrations/v009_create_webhook_queue.ts b/backend/src/migrations/v009_create_webhook_queue.ts index 6923cab6..a7b4869c 100644 --- a/backend/src/migrations/v009_create_webhook_queue.ts +++ b/backend/src/migrations/v009_create_webhook_queue.ts @@ -26,8 +26,11 @@ export default { id TEXT PRIMARY KEY, type TEXT NOT NULL, payload TEXT NOT NULL, - status TEXT NOT NULL CHECK(status IN ('pending','processing','success','failed')), - enqueued_at TEXT NOT NULL DEFAULT (datetime('now')) + status TEXT NOT NULL CHECK(status IN ('pending','processing','success','failed','dead_letter')), + enqueued_at TEXT NOT NULL DEFAULT (datetime('now')), + attempts INTEGER NOT NULL DEFAULT 0, + max_attempts INTEGER NOT NULL DEFAULT 5, + next_attempt_at TEXT ) `); @@ -37,6 +40,12 @@ export default { ON webhook_queue(status, enqueued_at) `); + // Create index for scheduler claim checks by status and next attempt time + await ctx.db.exec(` + CREATE INDEX IF NOT EXISTS idx_webhook_queue_status_next_attempt + ON webhook_queue(status, next_attempt_at) + `); + // Create queue metadata table for overflow count await ctx.db.exec(` CREATE TABLE IF NOT EXISTS queue_metadata ( diff --git a/backend/src/services/webhookDeliveryRepo.ts b/backend/src/services/webhookDeliveryRepo.ts index ba11a985..c1a747e9 100644 --- a/backend/src/services/webhookDeliveryRepo.ts +++ b/backend/src/services/webhookDeliveryRepo.ts @@ -86,7 +86,8 @@ function rowToDelivery(row: WebhookDeliveryRow): WebhookDelivery { export function computeNextRetry(attemptCount: number): string | null { if (attemptCount >= MAX_RETRY_ATTEMPTS) return null; - const baseDelay = RETRY_DELAYS_MS[attemptCount]; + const delayIndex = Math.max(0, attemptCount - 1); + const baseDelay = RETRY_DELAYS_MS[delayIndex] ?? RETRY_DELAYS_MS[RETRY_DELAYS_MS.length - 1]; const jitter = Math.round(Math.random() * baseDelay * 0.5); return new Date(Date.now() + baseDelay + jitter).toISOString(); } @@ -168,8 +169,12 @@ export class WebhookDeliveryRepo { } getPending(): WebhookDelivery[] { + return this.claimDue(new Date()); + } + + claimDue(now: Date): WebhookDelivery[] { const db = getDatabase(); - const now = new Date().toISOString(); + const nowIso = now.toISOString(); const rows = db .prepare( `SELECT * FROM webhook_deliveries @@ -177,7 +182,7 @@ export class WebhookDeliveryRepo { OR (status = 'failed' AND next_retry_at <= ?) ORDER BY enqueued_at ASC` ) - .all(now, now) as WebhookDeliveryRow[]; + .all(nowIso, nowIso) as WebhookDeliveryRow[]; return rows.map(rowToDelivery); } diff --git a/backend/src/services/webhookQueueService.ts b/backend/src/services/webhookQueueService.ts index 4dfc7155..a833178f 100644 --- a/backend/src/services/webhookQueueService.ts +++ b/backend/src/services/webhookQueueService.ts @@ -44,6 +44,7 @@ const WebhookQueueStatsSchema = z.object({ pendingCount: z.number().int().min(0), successCount: z.number().int().min(0), failureCount: z.number().int().min(0), + deadLetterCount: z.number().int().min(0), oldestTimestamp: z.string().datetime().nullable(), }); @@ -70,11 +71,7 @@ class WebhookQueueService { } public static resetInstance(): void { - if (WebhookQueueService.instance) { - WebhookQueueService.instance.db = getDatabase(); - } else { - WebhookQueueService.instance = new WebhookQueueService(); - } + WebhookQueueService.instance = new WebhookQueueService(); } enqueue(type: string, payload?: unknown): WebhookEvent { @@ -132,6 +129,7 @@ class WebhookQueueService { pendingCount: s.pending, successCount: s.success, failureCount: s.failed, + deadLetterCount: s.deadLetter, oldestTimestamp: s.oldestPending, }; } @@ -152,6 +150,10 @@ class WebhookQueueService { return webhookDeliveryRepo.getPending(); } + claimDue(now: Date = new Date()): WebhookEvent[] { + return webhookDeliveryRepo.claimDue(now).map(deliveryToEvent); + } + getDeadLetters(): WebhookDelivery[] { return webhookDeliveryRepo.getDeadLetters(); } diff --git a/backend/src/tests/webhook-delivery-repo.test.ts b/backend/src/tests/webhook-delivery-repo.test.ts index 1e866bb8..c9942a66 100644 --- a/backend/src/tests/webhook-delivery-repo.test.ts +++ b/backend/src/tests/webhook-delivery-repo.test.ts @@ -252,11 +252,30 @@ describe("WebhookDeliveryRepo", () => { }); describe("computeNextRetry", () => { + let dateNowSpy: jest.SpyInstance; + let randomSpy: jest.SpyInstance; + + beforeEach(() => { + dateNowSpy = jest.spyOn(Date, "now").mockReturnValue( + new Date("2026-07-03T00:00:00.000Z").getTime() + ); + randomSpy = jest.spyOn(Math, "random").mockReturnValue(0); + }); + + afterEach(() => { + dateNowSpy.mockRestore(); + randomSpy.mockRestore(); + }); + it("should return null beyond max attempts", () => { expect(computeNextRetry(MAX_RETRY_ATTEMPTS)).toBeNull(); expect(computeNextRetry(99)).toBeNull(); }); + it("should compute a deterministic first retry delay with zero jitter", () => { + expect(computeNextRetry(1)).toBe("2026-07-03T00:01:00.000Z"); + }); + it("should return a future date for valid attempts", () => { const next = computeNextRetry(0); expect(next).not.toBeNull(); @@ -324,6 +343,34 @@ describe("WebhookQueueService integration", () => { const deadLetters = webhookQueueService.getDeadLetters(); expect(deadLetters.length).toBe(2); + + const stats = webhookQueueService.getStats(); + expect(stats.deadLetterCount).toBe(2); + }); + + it("should claim only immediately due retryable deliveries", () => { + const db = getDatabase(); + const ready = webhookQueueService.enqueue("ready.pending", {}); + const dueRetry = webhookQueueService.enqueue("due.retry", {}); + const futureRetry = webhookQueueService.enqueue("future.retry", {}); + const deadLetter = webhookQueueService.enqueueWithSubscriber("dead.retry", {}, "sub-1"); + + webhookQueueService.markFailed(dueRetry.id); + webhookQueueService.markFailed(futureRetry.id); + for (let i = 0; i < 5; i++) { + webhookQueueService.markFailed(deadLetter.id); + } + + db.prepare( + "UPDATE webhook_deliveries SET next_retry_at = ? WHERE id = ?" + ).run("2026-07-03T00:00:00.000Z", dueRetry.id); + db.prepare( + "UPDATE webhook_deliveries SET next_retry_at = ? WHERE id = ?" + ).run("2026-07-03T00:10:00.000Z", futureRetry.id); + + const claimed = webhookQueueService.claimDue(new Date("2026-07-03T00:05:00.000Z")); + expect(claimed.map((event) => event.id).sort()).toEqual([dueRetry.id, ready.id].sort()); + expect(claimed.every((event) => event.status !== "dead_letter")).toBe(true); }); it("should cleanup old deliveries", () => { diff --git a/backend/src/tests/webhook-queue-migration.test.ts b/backend/src/tests/webhook-queue-migration.test.ts new file mode 100644 index 00000000..6c67b7c5 --- /dev/null +++ b/backend/src/tests/webhook-queue-migration.test.ts @@ -0,0 +1,32 @@ +import migration from "../migrations/v009_create_webhook_queue"; +import type { MigrationContext } from "../lib/migrations/types"; + +describe("v009 webhook queue migration", () => { + it("creates retry scheduling and dead-letter queue columns", async () => { + const execStatements: string[] = []; + const ctx = { + db: { + exec: jest.fn(async (sql: string) => { + execStatements.push(sql); + return []; + }), + get: jest.fn(), + run: jest.fn(async () => ({ lastInsertRowId: 0, changes: 0 })), + transaction: jest.fn((fn: (db: MigrationContext["db"]) => unknown) => fn(ctx.db)), + }, + env: {}, + isProduction: false, + isTest: true, + } satisfies MigrationContext; + + await migration.up(ctx); + + const createQueueSql = execStatements.find((sql) => sql.includes("CREATE TABLE IF NOT EXISTS webhook_queue")); + const indexSql = execStatements.find((sql) => sql.includes("idx_webhook_queue_status_next_attempt")); + expect(createQueueSql).toContain("attempts INTEGER NOT NULL DEFAULT 0"); + expect(createQueueSql).toContain("max_attempts INTEGER NOT NULL DEFAULT 5"); + expect(createQueueSql).toContain("next_attempt_at TEXT"); + expect(createQueueSql).toContain("'dead_letter'"); + expect(indexSql).toContain("status, next_attempt_at"); + }); +});