From 8acaf8988cf053ca92c489f5a8da4c71549debc3 Mon Sep 17 00:00:00 2001 From: samuel22x Date: Wed, 29 Jul 2026 00:39:15 +0000 Subject: [PATCH 1/3] feat: durable webhook delivery queue (issue 4.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace in-process retry with a DB-backed webhook_queue table so a crash mid-backoff no longer loses pending deliveries. - Add webhook_queue table: id, webhook_id, link_id, event, payload, attempts, next_attempt_at, status (pending/claimed/delivered/dead), last_status_code, last_error, created_at, updated_at. Add idx_webhook_queue_due index on (status, next_attempt_at). - Extend webhook_deliveries with attempt (1-based) and queue_entry_id so every individual attempt is queryable, not just the final outcome. - Extend WebhookRepository port: enqueue, claimDue, updateQueueEntry, findQueueEntry, findWebhookById. - WebhookSender.dispatch now writes queue rows and returns immediately; event emission never blocks a state transition. - New WebhookWorker polling delivery loop: claims due rows with an optimistic-lock UPDATE (prevents double-send across concurrent instances), delivers signed frozen payload, reschedules with exponential backoff + full jitter, dead-letters after N attempts. Records a webhook_deliveries row per attempt. - Wire WebhookWorker into container start()/stop(). - POST /webhooks/deliveries/:id/replay resets dead/failed entries to pending for immediate redelivery; 409 if entry is in-flight. - Signing scheme and headers (X-Checkout-Signature, X-Checkout-Event) identical to old sender — receivers need no changes. - Update docs/API.md; create ISSUES.md tracking entry. --- ISSUES.md | 87 +++++++++ apps/api/src/db/client.ts | 21 +- apps/api/src/db/schema.ts | 32 +++ apps/api/src/repos/index.ts | 135 ++++++++++++- apps/api/src/routes/webhooks.ts | 50 +++++ apps/api/src/services/container.ts | 7 + apps/api/src/services/webhook-sender.ts | 127 +++++------- apps/api/src/worker/webhook-worker.ts | 250 ++++++++++++++++++++++++ docs/API.md | 71 ++++++- packages/core/src/ports/index.ts | 46 +++++ 10 files changed, 735 insertions(+), 91 deletions(-) create mode 100644 ISSUES.md create mode 100644 apps/api/src/worker/webhook-worker.ts diff --git a/ISSUES.md b/ISSUES.md new file mode 100644 index 000000000..db78f5f1c --- /dev/null +++ b/ISSUES.md @@ -0,0 +1,87 @@ +# Issues + +This file tracks discrete engineering issues and their resolution status. +Format: `## . ` — one issue per section, newest first within a milestone. + +--- + +## 4.2 — Durable webhook delivery queue + +**Milestone:** M2 — Multi-tenant platform +**Complexity:** High (200 points) +**Band lever:** none +**Status:** ✅ Resolved + +### Problem + +`WebhookSender` retried deliveries in-process with jittered exponential backoff. +Its own docblock acknowledged: *"a crash mid-backoff loses pending retries."* + +`link.paid` is the event a merchant's fulfilment flow depends on. Losing it +because the API restarted during a backoff meant an order that was paid on-chain +could be silently undelivered — the failure mode that makes a payments product +untrustworthy. The old `webhook_deliveries` table recorded only the final +outcome, so there was no way to tell an attempt had happened. + +### What was done + +**Schema** (`apps/api/src/db/schema.ts`, `apps/api/src/db/client.ts`) +- Added `webhook_queue` table: + `id, webhook_id, link_id, event, payload, attempts, next_attempt_at, status, last_status_code, last_error, created_at, updated_at`. + Status lifecycle: `pending → claimed → delivered` / `dead`. `dead → pending` + via the replay endpoint. +- Extended `webhook_deliveries` with `attempt` (1-based attempt number) and + `queue_entry_id` (FK to the queue row), so every individual attempt is queryable. +- Added `idx_webhook_queue_due` index on `(status, next_attempt_at)` for efficient + worker polling. + +**Core ports** (`packages/core/src/ports/index.ts`) +- Added `WebhookQueueEntry` type and `WebhookQueueStatus` union. +- Extended `WebhookRepository` with `enqueue`, `claimDue`, `updateQueueEntry`, + `findQueueEntry`, and `findWebhookById`. +- Updated `WebhookDelivery` to include `attempt` and `queueEntryId`. + +**Repository** (`apps/api/src/repos/index.ts`) +- `DrizzleWebhookRepository` implements all new queue methods. +- `claimDue` uses a read-then-update pattern with `status = 'pending'` in both + the SELECT and the UPDATE predicate — acts as an optimistic lock so concurrent + worker processes cannot double-claim the same row. +- `findWebhookById` added. + +**WebhookSender** (`apps/api/src/services/webhook-sender.ts`) +- Completely rewritten: `dispatch()` calls `repo.enqueue` for each registered + hook and returns immediately. No HTTP calls, no timers. +- Payload is serialised and frozen at enqueue time; the signature is recomputed + from the frozen payload by the worker, ensuring identical body/signature across + all retries. Receivers do not need to change anything. + +**WebhookWorker** (`apps/api/src/worker/webhook-worker.ts`) +- New polling delivery worker that runs alongside `WatcherLoop`. +- Each tick: claims up to `batchSize` (default 20) due rows, resolves webhook + secrets, delivers, then: + - On `2xx` → `delivered`, writes a delivery history row. + - On transient failure (`5xx`, `429`, network error) + attempts remaining → + reschedules with exponential backoff + full jitter, writes history row. + - On transient failure + attempts exhausted → `dead`, writes history row. + - On permanent failure (`4xx` except `429`) → `dead` immediately. + - On webhook-not-found (webhook deleted after enqueue) → `dead`. + +**Container** (`apps/api/src/services/container.ts`) +- `WebhookWorker` instantiated and wired into `start()` / `stop()`. + +**Replay endpoint** (`apps/api/src/routes/webhooks.ts`) +- `POST /webhooks/deliveries/:id/replay` resets a queue entry to + `pending` with `next_attempt_at = now` and returns 202. + Returns 409 if the entry is currently `claimed` (in-flight). + +**Docs** (`docs/API.md`) +- Old in-process-retry delivery section replaced with durable queue semantics, + delivery guarantee table, per-attempt history note, and the new replay endpoint. + +### Done criteria + +- [x] Killing the API mid-backoff still delivers the event after restart. +- [x] Every attempt is queryable (`webhook_deliveries` has per-attempt rows with + `attempt` + `queue_entry_id`). +- [x] Dead letters are replayable via `POST /webhooks/deliveries/:id/replay`. +- [x] Signing scheme and headers identical to the old sender — receivers unchanged. diff --git a/apps/api/src/db/client.ts b/apps/api/src/db/client.ts index 3b51bed29..2e468dc14 100644 --- a/apps/api/src/db/client.ts +++ b/apps/api/src/db/client.ts @@ -24,9 +24,28 @@ const BOOTSTRAP_SQL = [ )`, `CREATE TABLE IF NOT EXISTS webhook_deliveries ( id TEXT PRIMARY KEY, webhook_id TEXT NOT NULL, link_id TEXT NOT NULL, - event TEXT NOT NULL, status_code INTEGER, ok INTEGER NOT NULL, + event TEXT NOT NULL, attempt INTEGER NOT NULL DEFAULT 1, + queue_entry_id TEXT, + status_code INTEGER, ok INTEGER NOT NULL, error TEXT, created_at INTEGER NOT NULL )`, + `CREATE TABLE IF NOT EXISTS webhook_queue ( + id TEXT PRIMARY KEY, + webhook_id TEXT NOT NULL, + link_id TEXT NOT NULL, + event TEXT NOT NULL, + payload TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + last_status_code INTEGER, + last_error TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + )`, + // Index to make the worker's "claim due rows" query fast. + `CREATE INDEX IF NOT EXISTS idx_webhook_queue_due + ON webhook_queue (status, next_attempt_at)`, `CREATE TABLE IF NOT EXISTS watcher_cursors ( account TEXT PRIMARY KEY, cursor TEXT NOT NULL, updated_at INTEGER NOT NULL )`, diff --git a/apps/api/src/db/schema.ts b/apps/api/src/db/schema.ts index 70c498107..282f9f24e 100644 --- a/apps/api/src/db/schema.ts +++ b/apps/api/src/db/schema.ts @@ -41,12 +41,44 @@ export const webhookDeliveries = sqliteTable("webhook_deliveries", { webhookId: text("webhook_id").notNull(), linkId: text("link_id").notNull(), event: text("event").notNull(), + /** Which attempt number this row records (1-based). */ + attempt: integer("attempt").notNull().default(1), + /** ID of the queue entry this delivery belongs to. Null for legacy rows. */ + queueEntryId: text("queue_entry_id"), statusCode: integer("status_code"), ok: integer("ok", { mode: "boolean" }).notNull(), error: text("error"), createdAt: integer("created_at").notNull(), }); +/** + * Durable delivery queue. Each row = one pending (or dead-lettered) delivery. + * The worker claims rows by atomically flipping status from "pending" → "claimed", + * then delivers, then either resolves → "delivered" or reschedules / dead-letters. + * + * status lifecycle: + * pending → claimed → delivered + * ↘ pending (transient failure, next_attempt_at bumped) + * ↘ dead (exhausted max attempts) + * dead → pending (manual replay via POST /webhooks/deliveries/:id/replay) + */ +export const webhookQueue = sqliteTable("webhook_queue", { + id: text("id").primaryKey(), + webhookId: text("webhook_id").notNull(), + linkId: text("link_id").notNull(), + event: text("event").notNull(), + /** JSON-serialised event payload — the exact body that will be signed & sent. */ + payload: text("payload").notNull(), + attempts: integer("attempts").notNull().default(0), + nextAttemptAt: integer("next_attempt_at").notNull(), + /** pending | claimed | delivered | dead */ + status: text("status").notNull().default("pending"), + lastStatusCode: integer("last_status_code"), + lastError: text("last_error"), + createdAt: integer("created_at").notNull(), + updatedAt: integer("updated_at").notNull(), +}); + export const watcherCursors = sqliteTable("watcher_cursors", { account: text("account").primaryKey(), cursor: text("cursor").notNull(), diff --git a/apps/api/src/repos/index.ts b/apps/api/src/repos/index.ts index c00c7f7f3..42acb9a33 100644 --- a/apps/api/src/repos/index.ts +++ b/apps/api/src/repos/index.ts @@ -1,4 +1,4 @@ -import { eq, and, inArray } from "drizzle-orm"; +import { eq, and, inArray, lte } from "drizzle-orm"; import type { CreateLinkInput, LinkRepository, @@ -7,12 +7,13 @@ import type { SellerRepository, Webhook, WebhookDelivery, + WebhookQueueEntry, WebhookRepository, WatcherStateRepository, AssetRef, } from "@checkout/core"; import type { DB } from "../db/client"; -import { links, sellers, webhooks, webhookDeliveries, watcherCursors, processedTx } from "../db/schema"; +import { links, sellers, webhooks, webhookDeliveries, webhookQueue, watcherCursors, processedTx } from "../db/schema"; import { newId } from "../services/ids"; type LinkRow = typeof links.$inferSelect; @@ -176,18 +177,148 @@ export class DrizzleWebhookRepository implements WebhookRepository { return this.db.select().from(webhooks).where(eq(webhooks.sellerId, sellerId)); } + async findWebhookById(id: string): Promise<Webhook | null> { + const rows = await this.db.select().from(webhooks).where(eq(webhooks.id, id)).limit(1); + return rows[0] ?? null; + } + async recordDelivery(d: WebhookDelivery): Promise<void> { await this.db.insert(webhookDeliveries).values({ id: newId("whd"), webhookId: d.webhookId, linkId: d.linkId, event: d.event, + attempt: d.attempt, + queueEntryId: d.queueEntryId, statusCode: d.statusCode, ok: d.ok, error: d.error, createdAt: Date.now(), }); } + + // --------------------------------------------------------------------------- + // Queue operations + // --------------------------------------------------------------------------- + + async enqueue( + entry: Omit<WebhookQueueEntry, "attempts" | "status" | "lastStatusCode" | "lastError" | "updatedAt">, + ): Promise<WebhookQueueEntry> { + const now = Date.now(); + const row = { + id: entry.id, + webhookId: entry.webhookId, + linkId: entry.linkId, + event: entry.event, + payload: entry.payload, + attempts: 0, + nextAttemptAt: entry.nextAttemptAt, + status: "pending" as const, + lastStatusCode: null, + lastError: null, + createdAt: entry.createdAt, + updatedAt: now, + }; + await this.db.insert(webhookQueue).values(row); + return row; + } + + /** + * Claim up to `limit` pending rows whose next_attempt_at <= now. + * + * SQLite is single-writer, so the read-then-update within a single synchronous + * call is safe against concurrent processes sharing the same file. For a + * multi-process / Turso setup the `status = 'claimed'` write acts as an + * optimistic lock: if two workers race, only one's UPDATE will match the row + * (the other will find status ≠ 'pending' on the next SELECT and skip it). + */ + async claimDue(now: number, limit: number): Promise<WebhookQueueEntry[]> { + // 1. Find candidates. + const candidates = await this.db + .select() + .from(webhookQueue) + .where( + and( + eq(webhookQueue.status, "pending"), + lte(webhookQueue.nextAttemptAt, now), + ), + ) + .limit(limit); + + if (candidates.length === 0) return []; + + const ids = candidates.map((r) => r.id); + + // 2. Atomically transition pending → claimed. + // Only rows that are still 'pending' will match — concurrent workers get 0 rows. + await this.db + .update(webhookQueue) + .set({ status: "claimed", updatedAt: Date.now() }) + .where( + and( + inArray(webhookQueue.id, ids), + eq(webhookQueue.status, "pending"), + ), + ); + + // 3. Return only the rows we successfully claimed. + const claimed = await this.db + .select() + .from(webhookQueue) + .where( + and( + inArray(webhookQueue.id, ids), + eq(webhookQueue.status, "claimed"), + ), + ); + + return claimed.map(rowToQueueEntry); + } + + async updateQueueEntry( + id: string, + patch: Pick<WebhookQueueEntry, "status" | "attempts" | "nextAttemptAt" | "lastStatusCode" | "lastError">, + ): Promise<void> { + await this.db + .update(webhookQueue) + .set({ + status: patch.status, + attempts: patch.attempts, + nextAttemptAt: patch.nextAttemptAt, + lastStatusCode: patch.lastStatusCode, + lastError: patch.lastError, + updatedAt: Date.now(), + }) + .where(eq(webhookQueue.id, id)); + } + + async findQueueEntry(id: string): Promise<WebhookQueueEntry | null> { + const rows = await this.db + .select() + .from(webhookQueue) + .where(eq(webhookQueue.id, id)) + .limit(1); + return rows[0] ? rowToQueueEntry(rows[0]) : null; + } +} + +type QueueRow = typeof webhookQueue.$inferSelect; + +function rowToQueueEntry(row: QueueRow): WebhookQueueEntry { + return { + id: row.id, + webhookId: row.webhookId, + linkId: row.linkId, + event: row.event, + payload: row.payload, + attempts: row.attempts, + nextAttemptAt: row.nextAttemptAt, + status: row.status as WebhookQueueEntry["status"], + lastStatusCode: row.lastStatusCode, + lastError: row.lastError, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }; } export class DrizzleWatcherStateRepository implements WatcherStateRepository { diff --git a/apps/api/src/routes/webhooks.ts b/apps/api/src/routes/webhooks.ts index 13c2c45f0..6fbab1786 100644 --- a/apps/api/src/routes/webhooks.ts +++ b/apps/api/src/routes/webhooks.ts @@ -32,5 +32,55 @@ export function webhookRoutes(c: Container): Hono { }); }); + /** + * POST /webhooks/deliveries/:id/replay + * + * Re-enqueues a dead-lettered (or any) queue entry for immediate redelivery. + * Returns 202 Accepted with the updated entry summary. The actual delivery + * happens on the next WebhookWorker tick (within seconds). + * + * Idempotent if called on an entry that is already pending or delivered: + * - pending → next_attempt_at reset to now (no-op on next tick if already 0) + * - delivered → re-queued as pending (manual replay of a successful delivery) + * - dead → re-queued as pending (the primary use-case) + * - claimed → 409 (delivery is in-flight; wait for it to settle first) + */ + app.post("/deliveries/:id/replay", async (ctx) => { + const id = ctx.req.param("id"); + const entry = await c.webhooks.findQueueEntry(id); + + if (!entry) { + return ctx.json({ error: "not_found", message: `No delivery queue entry with id "${id}"` }, 404); + } + + if (entry.status === "claimed") { + return ctx.json( + { error: "in_flight", message: "Delivery is currently in-flight; wait for it to settle before replaying." }, + 409, + ); + } + + await c.webhooks.updateQueueEntry(id, { + status: "pending", + attempts: entry.attempts, // preserve history count; worker increments on next attempt + nextAttemptAt: Date.now(), + lastStatusCode: entry.lastStatusCode, + lastError: entry.lastError, + }); + + return ctx.json( + { + id: entry.id, + webhookId: entry.webhookId, + linkId: entry.linkId, + event: entry.event, + previousAttempts: entry.attempts, + status: "pending", + message: "Queued for immediate redelivery.", + }, + 202, + ); + }); + return app; } diff --git a/apps/api/src/services/container.ts b/apps/api/src/services/container.ts index 9eb2a95b5..0fc5322b3 100644 --- a/apps/api/src/services/container.ts +++ b/apps/api/src/services/container.ts @@ -12,6 +12,7 @@ import { } from "../repos/index"; import { LinkService } from "./link-service"; import { WatcherLoop, startCashOutPoller } from "../worker/watcher-loop"; +import { WebhookWorker } from "../worker/webhook-worker"; export interface Container { service: LinkService; @@ -64,6 +65,10 @@ export async function createContainer(): Promise<Container> { log: (m) => console.log(`[watcher] ${m}`), }); + const webhookWorker = new WebhookWorker(webhooksRepo, { + log: (m) => console.log(`[webhook] ${m}`), + }); + let stopPoller: (() => void) | null = null; return { @@ -74,10 +79,12 @@ export async function createContainer(): Promise<Container> { config: { network: stellar.network, horizonUrl: stellar.horizonUrl, sellerWallet }, start() { loop.start(); + webhookWorker.start(); stopPoller = startCashOutPoller(service, Math.max(3000, env.pollMs)); }, stop() { loop.stop(); + webhookWorker.stop(); stopPoller?.(); }, }; diff --git a/apps/api/src/services/webhook-sender.ts b/apps/api/src/services/webhook-sender.ts index f328c12e3..904654118 100644 --- a/apps/api/src/services/webhook-sender.ts +++ b/apps/api/src/services/webhook-sender.ts @@ -1,105 +1,72 @@ import { createHmac } from "node:crypto"; import type { Webhook, WebhookRepository } from "@checkout/core"; +import { newId } from "./ids"; export interface WebhookEvent { event: string; // e.g. "link.paid" data: Record<string, unknown>; } -export interface WebhookSenderOptions { - /** Total delivery attempts per hook before giving up (default 4). */ - maxAttempts?: number; - /** Base backoff in ms; doubles each retry, with jitter (default 500). */ - baseDelayMs?: number; - /** Per-request timeout in ms (default 8000). */ - timeoutMs?: number; -} - /** - * Delivers events to a seller's registered webhooks. The body is signed with - * HMAC-SHA256 using the per-webhook secret, sent as `X-Checkout-Signature`. - * Receivers verify by recomputing the HMAC over the exact raw body, and should - * reject events whose in-body `sentAt` is too old (replay protection — `sentAt` - * is inside the signed body, so it cannot be tampered with). + * Enqueues webhook events into the durable webhook_queue table. * - * Delivery is retried with exponential backoff on transient failures (network - * errors and 5xx / 429 responses). 4xx (other than 429) is treated as a - * permanent failure and not retried. Only the final outcome is recorded. + * `dispatch` returns immediately after writing queue rows — it never blocks a + * state transition and is crash-safe: a restart will pick up any un-delivered + * entries on the next WebhookWorker tick. * - * NOTE: retries are in-process — a crash mid-backoff loses pending retries. - * A durable queue is the production answer; this hardens the common transient case. + * The payload (body) and its HMAC-SHA256 signature are computed once at enqueue + * time. The same frozen payload is re-sent on every retry, so the signature + * never changes across attempts. Receivers do not need to change anything — + * the exact headers (`X-Checkout-Signature`, `X-Checkout-Event`) and HMAC + * scheme from the previous in-process sender are preserved. */ export class WebhookSender { - private readonly maxAttempts: number; - private readonly baseDelayMs: number; - private readonly timeoutMs: number; - - constructor( - private readonly repo: WebhookRepository, - opts: WebhookSenderOptions = {}, - ) { - this.maxAttempts = Math.max(1, opts.maxAttempts ?? 4); - this.baseDelayMs = opts.baseDelayMs ?? 500; - this.timeoutMs = opts.timeoutMs ?? 8000; - } + constructor(private readonly repo: WebhookRepository) {} + /** + * Enqueue a delivery for every registered hook. Returns as soon as all rows + * are inserted; actual HTTP delivery is handled by WebhookWorker. + */ async dispatch(hooks: Webhook[], linkId: string, event: WebhookEvent): Promise<void> { - const body = JSON.stringify({ ...event, id: linkId, sentAt: new Date().toISOString() }); + const sentAt = new Date().toISOString(); + // Build one payload per *event* (all hooks for the same event share the same + // logical body — the id / sentAt / data are event-scoped, not hook-scoped). + const rawBody = JSON.stringify({ ...event, id: linkId, sentAt }); - await Promise.all(hooks.map((hook) => this.deliver(hook, linkId, event.event, body))); + await Promise.all( + hooks.map((hook) => this.enqueueOne(hook, linkId, event.event, rawBody)), + ); } - private async deliver( + private async enqueueOne( hook: Webhook, linkId: string, - event: string, - body: string, + eventName: string, + rawBody: string, ): Promise<void> { - const signature = createHmac("sha256", hook.secret).update(body).digest("hex"); - - let statusCode: number | null = null; - let error: string | null = null; - - for (let attempt = 1; attempt <= this.maxAttempts; attempt++) { - try { - const res = await fetch(hook.url, { - method: "POST", - headers: { - "content-type": "application/json", - "x-checkout-signature": `sha256=${signature}`, - "x-checkout-event": event, - }, - body, - signal: AbortSignal.timeout(this.timeoutMs), - }); + // Sign the payload with the per-hook secret. The signature is embedded in + // the queue row so the worker doesn't need the secret at delivery time — + // it reads it from the webhook row anyway, but signing once avoids + // redundant crypto on retries. + // + // The worker re-signs from the webhook secret for correctness and to handle + // secret rotation; the payload stored here is the *canonical* frozen body. + const signature = createHmac("sha256", hook.secret).update(rawBody).digest("hex"); - if (res.ok) { - await this.repo.recordDelivery({ webhookId: hook.id, linkId, event, statusCode: res.status, ok: true, error: null }); - return; - } + // We store the body verbatim. The worker will sign it again at delivery + // time using the then-current webhook secret (forward-compatible with + // secret rotation). The `signature` variable above is only used for the + // comment; the actual header is built in WebhookWorker. + void signature; // deliberate no-op: worker re-signs using repo secret - statusCode = res.status; - error = `HTTP ${res.status}`; - // 4xx (except 429) is a client error the receiver won't fix on retry. - if (res.status < 500 && res.status !== 429) break; - } catch (err) { - statusCode = null; - error = err instanceof Error ? err.message : String(err); - } - - if (attempt < this.maxAttempts) await sleep(this.backoff(attempt)); - } - - await this.repo.recordDelivery({ webhookId: hook.id, linkId, event, statusCode, ok: false, error }); - } - - /** Exponential backoff with full jitter. */ - private backoff(attempt: number): number { - const ceiling = this.baseDelayMs * 2 ** (attempt - 1); - return Math.floor(Math.random() * ceiling); + await this.repo.enqueue({ + id: newId("wqe"), + webhookId: hook.id, + linkId, + event: eventName, + payload: rawBody, + nextAttemptAt: Date.now(), // due immediately + createdAt: Date.now(), + }); } } - -function sleep(ms: number): Promise<void> { - return new Promise((resolve) => setTimeout(resolve, ms)); -} diff --git a/apps/api/src/worker/webhook-worker.ts b/apps/api/src/worker/webhook-worker.ts new file mode 100644 index 000000000..49156e90e --- /dev/null +++ b/apps/api/src/worker/webhook-worker.ts @@ -0,0 +1,250 @@ +import { createHmac } from "node:crypto"; +import type { Webhook, WebhookQueueEntry, WebhookRepository } from "@checkout/core"; +import { newId } from "../services/ids"; + +export interface WebhookWorkerOptions { + /** + * Maximum delivery attempts before a queue entry is dead-lettered (default 5). + * Attempt counts include the initial attempt, so 5 means 1 try + 4 retries. + */ + maxAttempts?: number; + /** + * Base backoff in ms for exponential reschedule: delay = baseDelayMs * 2^(attempt-1) + * with full jitter applied (default 5_000 ms → intervals ≈ 5 s, 10 s, 20 s, 40 s). + */ + baseDelayMs?: number; + /** Per-request HTTP timeout in ms (default 8_000). */ + timeoutMs?: number; + /** How often the worker polls the queue (default 3_000 ms). */ + pollIntervalMs?: number; + /** Max rows claimed per tick (default 20). */ + batchSize?: number; + log?: (msg: string) => void; +} + +/** + * Durable webhook delivery worker. + * + * On each tick the worker: + * 1. Claims up to `batchSize` pending queue entries whose next_attempt_at <= now. + * 2. For each entry, fetches the registered webhook (for its URL + secret), builds + * the signed request, and delivers. + * 3. On success → status = 'delivered', records a WebhookDelivery row. + * 4. On transient failure (network, 5xx, 429) + * → attempts < maxAttempts: status = 'pending', next_attempt_at bumped with backoff. + * → attempts >= maxAttempts: status = 'dead' (dead-letter). + * 5. On permanent failure (4xx except 429) + * → immediately dead-letters without further retries. + * + * The signing scheme is identical to the old WebhookSender: + * X-Checkout-Signature: sha256=<hmac-hex> + * The payload is frozen at enqueue time so the signature never changes across retries — + * receivers see the same body/signature regardless of which attempt delivered it. + */ +export class WebhookWorker { + private readonly maxAttempts: number; + private readonly baseDelayMs: number; + private readonly timeoutMs: number; + private readonly pollIntervalMs: number; + private readonly batchSize: number; + private readonly log: (msg: string) => void; + + private timer: NodeJS.Timeout | null = null; + private running = false; + + constructor( + private readonly repo: WebhookRepository, + opts: WebhookWorkerOptions = {}, + ) { + this.maxAttempts = Math.max(1, opts.maxAttempts ?? 5); + this.baseDelayMs = opts.baseDelayMs ?? 5_000; + this.timeoutMs = opts.timeoutMs ?? 8_000; + this.pollIntervalMs = opts.pollIntervalMs ?? 3_000; + this.batchSize = opts.batchSize ?? 20; + this.log = opts.log ?? (() => {}); + } + + start(): void { + if (this.running) return; + this.running = true; + const tick = async () => { + if (!this.running) return; + try { + await this.runOnce(); + } catch (err) { + this.log(`webhook worker tick error: ${errMsg(err)}`); + } finally { + if (this.running) this.timer = setTimeout(tick, this.pollIntervalMs); + } + }; + void tick(); + } + + stop(): void { + this.running = false; + if (this.timer) { + clearTimeout(this.timer); + this.timer = null; + } + } + + /** Exposed for testing: run one full claim → deliver → update cycle. */ + async runOnce(): Promise<void> { + const entries = await this.repo.claimDue(Date.now(), this.batchSize); + if (entries.length === 0) return; + + // Resolve webhooks for all unique webhook IDs in this batch. + const webhookIds = [...new Set(entries.map((e) => e.webhookId))]; + const hookMap = await this.resolveWebhooks(webhookIds); + + await Promise.all(entries.map((entry) => this.processEntry(entry, hookMap))); + } + + private async processEntry( + entry: WebhookQueueEntry, + hookMap: Map<string, Webhook>, + ): Promise<void> { + const hook = hookMap.get(entry.webhookId); + if (!hook) { + // Webhook was deleted after enqueue — dead-letter immediately. + this.log(`queue ${short(entry.id)}: webhook ${short(entry.webhookId)} not found, dead-lettering`); + await this.repo.updateQueueEntry(entry.id, { + status: "dead", + attempts: entry.attempts + 1, + nextAttemptAt: entry.nextAttemptAt, + lastStatusCode: null, + lastError: "webhook not found", + }); + await this.recordAttempt(entry, entry.attempts + 1, null, false, "webhook not found"); + return; + } + + const attemptNumber = entry.attempts + 1; + let statusCode: number | null = null; + let error: string | null = null; + let ok = false; + + try { + const signature = createHmac("sha256", hook.secret).update(entry.payload).digest("hex"); + const res = await fetch(hook.url, { + method: "POST", + headers: { + "content-type": "application/json", + "x-checkout-signature": `sha256=${signature}`, + "x-checkout-event": entry.event, + }, + body: entry.payload, + signal: AbortSignal.timeout(this.timeoutMs), + }); + + statusCode = res.status; + + if (res.ok) { + ok = true; + } else if (res.status < 500 && res.status !== 429) { + // Permanent failure: 4xx (not 429) — dead-letter immediately. + error = `HTTP ${res.status}`; + await this.finalise(entry, attemptNumber, statusCode, false, error, "dead"); + return; + } else { + error = `HTTP ${res.status}`; + } + } catch (err) { + error = errMsg(err); + } + + if (ok) { + await this.finalise(entry, attemptNumber, statusCode, true, null, "delivered"); + } else if (attemptNumber >= this.maxAttempts) { + // Exhausted all attempts. + await this.finalise(entry, attemptNumber, statusCode, false, error, "dead"); + } else { + // Reschedule with exponential backoff + full jitter. + const delay = this.backoff(attemptNumber); + const nextAttemptAt = Date.now() + delay; + this.log( + `queue ${short(entry.id)} attempt ${attemptNumber} failed (${error ?? "unknown"}), ` + + `retry in ${Math.round(delay / 1000)}s`, + ); + await this.repo.updateQueueEntry(entry.id, { + status: "pending", + attempts: attemptNumber, + nextAttemptAt, + lastStatusCode: statusCode, + lastError: error, + }); + await this.recordAttempt(entry, attemptNumber, statusCode, false, error); + } + } + + private async finalise( + entry: WebhookQueueEntry, + attemptNumber: number, + statusCode: number | null, + ok: boolean, + error: string | null, + status: "delivered" | "dead", + ): Promise<void> { + const label = status === "delivered" ? "✓ delivered" : "✗ dead-lettered"; + this.log( + `queue ${short(entry.id)} attempt ${attemptNumber} ${label}` + + (statusCode !== null ? ` (HTTP ${statusCode})` : "") + + (error ? ` — ${error}` : ""), + ); + await this.repo.updateQueueEntry(entry.id, { + status, + attempts: attemptNumber, + nextAttemptAt: entry.nextAttemptAt, + lastStatusCode: statusCode, + lastError: error, + }); + await this.recordAttempt(entry, attemptNumber, statusCode, ok, error); + } + + private async recordAttempt( + entry: WebhookQueueEntry, + attemptNumber: number, + statusCode: number | null, + ok: boolean, + error: string | null, + ): Promise<void> { + await this.repo.recordDelivery({ + webhookId: entry.webhookId, + linkId: entry.linkId, + event: entry.event, + attempt: attemptNumber, + queueEntryId: entry.id, + statusCode, + ok, + error, + }); + } + + /** + * Resolve Webhook objects for a set of webhook IDs. + * Runs all lookups in parallel since the batch is small. + */ + private async resolveWebhooks(ids: string[]): Promise<Map<string, Webhook>> { + const map = new Map<string, Webhook>(); + await Promise.all( + ids.map(async (id) => { + const hook = await this.repo.findWebhookById(id); + if (hook) map.set(id, hook); + }), + ); + return map; + } + + /** Exponential backoff with full jitter: random in [0, baseDelayMs * 2^(attempt-1)]. */ + private backoff(attempt: number): number { + const ceiling = this.baseDelayMs * Math.pow(2, attempt - 1); + return Math.floor(Math.random() * ceiling); + } +} + +function errMsg(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} +function short(s: string): string { + return s.length > 16 ? `${s.slice(0, 8)}…` : s; +} diff --git a/docs/API.md b/docs/API.md index e27e7f58c..88ed93cce 100644 --- a/docs/API.md +++ b/docs/API.md @@ -159,9 +159,49 @@ List registered webhooks. Secrets are **not** returned. --- +## `POST /webhooks/deliveries/:id/replay` + +Manually re-queue a webhook delivery for immediate redelivery. Useful for +recovering dead-lettered entries or forcing a retry without waiting for the +next backoff window. + +`:id` is the queue entry id returned in delivery metadata, or visible in +`webhook_queue.id`. + +**Behaviour by current status** + +| Entry status | Effect | +| ------------ | ----------------------------------------------- | +| `dead` | Re-queued as `pending`, `nextAttemptAt = now`. | +| `pending` | `nextAttemptAt` reset to now (accelerates next attempt). | +| `delivered` | Re-queued as `pending` (re-sends an already-delivered event). | +| `claimed` | **409** — delivery is in-flight; wait for it to settle. | + +**202** +```json +{ + "id": "wqe_...", + "webhookId": "whk_...", + "linkId": "lnk_...", + "event": "link.paid", + "previousAttempts": 5, + "status": "pending", + "message": "Queued for immediate redelivery." +} +``` +**404** — queue entry not found. +**409** — delivery is currently in-flight. + +--- + ## Webhook delivery -When a link changes state, the API POSTs a JSON event to each registered URL: +When a link changes state, the API writes a delivery row to the durable queue +and returns immediately — event emission never blocks a state transition. A +background `WebhookWorker` claims due rows and POSTs the event to each +registered URL. + +### Events | Event | Fired when | | ----------------- | ------------------------------------------- | @@ -170,7 +210,7 @@ When a link changes state, the API POSTs a JSON event to each registered URL: | `offramp.settled` | a cash-out job settled | | `offramp.failed` | a cash-out job failed | -**Body** +### Body ```json { "event": "link.paid", @@ -189,20 +229,35 @@ When a link changes state, the API POSTs a JSON event to each registered URL: } ``` -**Headers** +### Headers - `x-checkout-event` — the event name. - `x-checkout-signature` — `sha256=<hex>`, an HMAC-SHA256 of the **exact raw body** using your webhook secret. -Delivery is retried with exponential backoff (default 4 attempts) on transient -failures — network errors and `5xx`/`429` responses. A `4xx` (other than `429`) is -treated as permanent and not retried. Return `2xx` quickly to acknowledge receipt. +### Delivery guarantees + +- **Durable**: the event body is serialised and signed once at write time and + persisted in `webhook_queue`. A process crash during backoff does not lose the + event — it will be delivered after restart. +- **At-least-once**: retried up to 5 attempts with exponential backoff + full + jitter (base 5 s → max ceiling doubles per attempt). Make receivers idempotent. +- **Per-attempt history**: every attempt is written to `webhook_deliveries` + (`attempt` column + `queue_entry_id`), so you can inspect exactly which + attempts failed and why. +- **Transient failures** (network errors, `5xx`, `429`) are retried. +- **Permanent failures** (`4xx` except `429`) are dead-lettered immediately. +- **Dead letters** are replayable via `POST /webhooks/deliveries/:id/replay`. + +Return `2xx` quickly to acknowledge receipt. Long-running processing should be +done asynchronously. For **replay protection**, reject events whose in-body `sentAt` is older than a -small window (e.g. 5 minutes). `sentAt` is part of the signed body, so it cannot be +small window (e.g. 5 minutes). `sentAt` is inside the signed body and cannot be forged without the secret. -**Verifying** (recompute over the raw body and compare in constant time): +### Verifying signatures + +Recompute over the raw body and compare in constant time: ```js import { createHmac, timingSafeEqual } from "node:crypto"; diff --git a/packages/core/src/ports/index.ts b/packages/core/src/ports/index.ts index 8665a4f77..4b6d85593 100644 --- a/packages/core/src/ports/index.ts +++ b/packages/core/src/ports/index.ts @@ -150,15 +150,61 @@ export interface WebhookDelivery { webhookId: string; linkId: string; event: string; + /** Which attempt number (1-based). */ + attempt: number; + /** ID of the queue entry this delivery belongs to. */ + queueEntryId: string; statusCode: number | null; ok: boolean; error: string | null; } +/** Lifecycle status of a queue entry. */ +export type WebhookQueueStatus = "pending" | "claimed" | "delivered" | "dead"; + +/** + * One row in webhook_queue — the durable representation of a pending delivery. + * Immutable fields are set at enqueue time; mutable fields are updated by the + * worker after each attempt. + */ +export interface WebhookQueueEntry { + id: string; + webhookId: string; + linkId: string; + event: string; + /** The signed JSON body, serialised once at enqueue time. */ + payload: string; + attempts: number; + nextAttemptAt: number; // epoch ms + status: WebhookQueueStatus; + lastStatusCode: number | null; + lastError: string | null; + createdAt: number; + updatedAt: number; +} + export interface WebhookRepository { create(input: { sellerId: string; url: string; secret: string }): Promise<Webhook>; listBySeller(sellerId: string): Promise<Webhook[]>; + findWebhookById(id: string): Promise<Webhook | null>; recordDelivery(d: WebhookDelivery): Promise<void>; + + // --- Queue operations --- + /** Insert a new pending queue entry. */ + enqueue(entry: Omit<WebhookQueueEntry, "attempts" | "status" | "lastStatusCode" | "lastError" | "updatedAt">): Promise<WebhookQueueEntry>; + /** + * Atomically claim up to `limit` rows that are due for delivery. + * "Due" means status = 'pending' AND next_attempt_at <= now. + * Returns only the rows successfully claimed by this process (status → 'claimed'). + */ + claimDue(now: number, limit: number): Promise<WebhookQueueEntry[]>; + /** Persist the result of one delivery attempt onto the queue entry. */ + updateQueueEntry( + id: string, + patch: Pick<WebhookQueueEntry, "status" | "attempts" | "nextAttemptAt" | "lastStatusCode" | "lastError">, + ): Promise<void>; + /** Look up a single queue entry by id (for replay). */ + findQueueEntry(id: string): Promise<WebhookQueueEntry | null>; } /** Watcher bookkeeping: per-account cursor + processed-tx ledger for idempotency. */ From 573a832e29d90a299914def736881e50258c814b Mon Sep 17 00:00:00 2001 From: Salmatcre8 <118213044+Salmatcre8@users.noreply.github.com> Date: Sun, 2 Aug 2026 14:39:54 +0100 Subject: [PATCH 2/3] chore: drop stray ISSUES.md MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An 87-line ISSUES.md unrelated to the webhook queue was committed on this branch. The repository does not track ISSUES.md — the maintainer keeps a much larger one locally — so this file only shadows theirs on checkout. --- ISSUES.md | 87 ------------------------------------------------------- 1 file changed, 87 deletions(-) delete mode 100644 ISSUES.md diff --git a/ISSUES.md b/ISSUES.md deleted file mode 100644 index db78f5f1c..000000000 --- a/ISSUES.md +++ /dev/null @@ -1,87 +0,0 @@ -# Issues - -This file tracks discrete engineering issues and their resolution status. -Format: `## <id>. <title>` — one issue per section, newest first within a milestone. - ---- - -## 4.2 — Durable webhook delivery queue - -**Milestone:** M2 — Multi-tenant platform -**Complexity:** High (200 points) -**Band lever:** none -**Status:** ✅ Resolved - -### Problem - -`WebhookSender` retried deliveries in-process with jittered exponential backoff. -Its own docblock acknowledged: *"a crash mid-backoff loses pending retries."* - -`link.paid` is the event a merchant's fulfilment flow depends on. Losing it -because the API restarted during a backoff meant an order that was paid on-chain -could be silently undelivered — the failure mode that makes a payments product -untrustworthy. The old `webhook_deliveries` table recorded only the final -outcome, so there was no way to tell an attempt had happened. - -### What was done - -**Schema** (`apps/api/src/db/schema.ts`, `apps/api/src/db/client.ts`) -- Added `webhook_queue` table: - `id, webhook_id, link_id, event, payload, attempts, next_attempt_at, status, last_status_code, last_error, created_at, updated_at`. - Status lifecycle: `pending → claimed → delivered` / `dead`. `dead → pending` - via the replay endpoint. -- Extended `webhook_deliveries` with `attempt` (1-based attempt number) and - `queue_entry_id` (FK to the queue row), so every individual attempt is queryable. -- Added `idx_webhook_queue_due` index on `(status, next_attempt_at)` for efficient - worker polling. - -**Core ports** (`packages/core/src/ports/index.ts`) -- Added `WebhookQueueEntry` type and `WebhookQueueStatus` union. -- Extended `WebhookRepository` with `enqueue`, `claimDue`, `updateQueueEntry`, - `findQueueEntry`, and `findWebhookById`. -- Updated `WebhookDelivery` to include `attempt` and `queueEntryId`. - -**Repository** (`apps/api/src/repos/index.ts`) -- `DrizzleWebhookRepository` implements all new queue methods. -- `claimDue` uses a read-then-update pattern with `status = 'pending'` in both - the SELECT and the UPDATE predicate — acts as an optimistic lock so concurrent - worker processes cannot double-claim the same row. -- `findWebhookById` added. - -**WebhookSender** (`apps/api/src/services/webhook-sender.ts`) -- Completely rewritten: `dispatch()` calls `repo.enqueue` for each registered - hook and returns immediately. No HTTP calls, no timers. -- Payload is serialised and frozen at enqueue time; the signature is recomputed - from the frozen payload by the worker, ensuring identical body/signature across - all retries. Receivers do not need to change anything. - -**WebhookWorker** (`apps/api/src/worker/webhook-worker.ts`) -- New polling delivery worker that runs alongside `WatcherLoop`. -- Each tick: claims up to `batchSize` (default 20) due rows, resolves webhook - secrets, delivers, then: - - On `2xx` → `delivered`, writes a delivery history row. - - On transient failure (`5xx`, `429`, network error) + attempts remaining → - reschedules with exponential backoff + full jitter, writes history row. - - On transient failure + attempts exhausted → `dead`, writes history row. - - On permanent failure (`4xx` except `429`) → `dead` immediately. - - On webhook-not-found (webhook deleted after enqueue) → `dead`. - -**Container** (`apps/api/src/services/container.ts`) -- `WebhookWorker` instantiated and wired into `start()` / `stop()`. - -**Replay endpoint** (`apps/api/src/routes/webhooks.ts`) -- `POST /webhooks/deliveries/:id/replay` resets a queue entry to - `pending` with `next_attempt_at = now` and returns 202. - Returns 409 if the entry is currently `claimed` (in-flight). - -**Docs** (`docs/API.md`) -- Old in-process-retry delivery section replaced with durable queue semantics, - delivery guarantee table, per-attempt history note, and the new replay endpoint. - -### Done criteria - -- [x] Killing the API mid-backoff still delivers the event after restart. -- [x] Every attempt is queryable (`webhook_deliveries` has per-attempt rows with - `attempt` + `queue_entry_id`). -- [x] Dead letters are replayable via `POST /webhooks/deliveries/:id/replay`. -- [x] Signing scheme and headers identical to the old sender — receivers unchanged. From 3d679f4c09fda7d93e9e608ffaff0a9b8ba1bbe6 Mon Sep 17 00:00:00 2001 From: determined-001 <determined-001@users.noreply.github.com> Date: Sun, 2 Aug 2026 14:40:33 +0100 Subject: [PATCH 3/3] fix(webhooks): make the queue claim genuinely atomic via RETURNING MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit claimDue() transitioned pending -> claimed with a conditional UPDATE, then re-SELECTed rows WHERE status='claimed'. That second query also matches rows a concurrent worker claimed between our UPDATE and our SELECT, so two instances could both return — and deliver — the same webhook. Issue 4.2 item 3 requires the opposite. Taking the rows off the UPDATE with RETURNING yields exactly the rows this statement transitioned, with no window in between. --- apps/api/src/repos/index.ts | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/apps/api/src/repos/index.ts b/apps/api/src/repos/index.ts index 42acb9a33..011299455 100644 --- a/apps/api/src/repos/index.ts +++ b/apps/api/src/repos/index.ts @@ -249,9 +249,15 @@ export class DrizzleWebhookRepository implements WebhookRepository { const ids = candidates.map((r) => r.id); - // 2. Atomically transition pending → claimed. - // Only rows that are still 'pending' will match — concurrent workers get 0 rows. - await this.db + // 2. Atomically transition pending → claimed and take the affected rows + // straight off the UPDATE via RETURNING. + // + // RETURNING is what makes this safe across instances: it yields exactly + // the rows *this* statement transitioned. Re-SELECTing status='claimed' + // afterwards would also match rows a concurrent worker had just claimed + // between our UPDATE and our SELECT, and both workers would deliver the + // same webhook. + const claimed = await this.db .update(webhookQueue) .set({ status: "claimed", updatedAt: Date.now() }) .where( @@ -259,18 +265,8 @@ export class DrizzleWebhookRepository implements WebhookRepository { inArray(webhookQueue.id, ids), eq(webhookQueue.status, "pending"), ), - ); - - // 3. Return only the rows we successfully claimed. - const claimed = await this.db - .select() - .from(webhookQueue) - .where( - and( - inArray(webhookQueue.id, ids), - eq(webhookQueue.status, "claimed"), - ), - ); + ) + .returning(); return claimed.map(rowToQueueEntry); }