From 49ff2349544dd4850710e0fb06e2e4c36ef00b53 Mon Sep 17 00:00:00 2001 From: Jakob Wennberg <149234542+jakobwennberg@users.noreply.github.com> Date: Tue, 28 Jul 2026 18:08:24 +0200 Subject: [PATCH] feat(webhooks): dispatch on emit instead of waiting for the next cron tick (#1256) * feat(webhooks): dispatch on emit instead of waiting for the next cron tick The webhook dispatcher ran only on a per-minute cron, so the floor on delivery latency was up to 60 seconds plus the request. An external consumer that wanted to react as a transaction landed had only one alternative: polling /api/events, which the 100 rpm per-key limit makes expensive and which still cannot beat the tick interval. Schedules one dispatch cycle as soon as deliveries are enqueued. The cron is unchanged and remains the retry and sweep path; this only moves the first attempt forward. Wired into the event-bus fanout plus the two routes that enqueue a delivery directly: the :test verb, whose entire purpose is telling someone whether their receiver works, and the manual delivery retry. Three properties are load-bearing and covered by tests. The kick is never awaited, because eventBus.emit is awaited at ~99 call sites including journal_entry.committed and each delivery can burn a 10 s receiver timeout. It coalesces per function instance, so a bulk booking that emits once per row does not schedule one claim round trip per row. It claims 5 rows rather than the cron's 50, because it runs on the tail of a user-facing request. Double delivery is not a risk: claim_due_webhook_deliveries already claims FOR UPDATE SKIP LOCKED and flips rows to in_flight in the same statement, so a kick racing the cron sees disjoint rows. Does not close #1201, which asks for a realtime stream for API consumers. This is the cheap half. Refs #1201 Co-Authored-By: Claude Opus 5 (1M context) * docs(webhooks): stop claiming the kick makes double delivery impossible Adversarial review of the previous commit caught an overstatement in its own comments. SKIP LOCKED keeps a kick and the cron from claiming the same row at the same moment, but claim_due_webhook_deliveries autocommits before any POST is issued, so from then on ownership is only status='in_flight' and a later cycle's recoverStuckInFlight sweep can re-arm a row still queued behind an earlier cycle's serial loop. Delivery is at-least-once, which is what the public docs already tell receivers ("the same delivery id may arrive more than once ... idempotency is on you"). The comments contradicted that. No behaviour change. The kick does not create this window: the cron claims 50 rows serially against the same 20 s stuck threshold, which is wider than what a batch of 5 can open. Refs #1201 Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Opus 5 (1M context) --- DECISIONS.md | 1 + .../[companyId]/webhooks/[id]/test/route.ts | 12 +- .../v1/webhook-deliveries/[id]/retry/route.ts | 6 + lib/docs/content/changelog.ts | 2 +- lib/docs/content/webhooks.ts | 4 +- lib/webhooks/__tests__/dispatch-kick.test.ts | 110 ++++++++++++++++ lib/webhooks/dispatch-kick.ts | 121 ++++++++++++++++++ lib/webhooks/dispatcher.ts | 10 +- lib/webhooks/handler.ts | 12 +- 9 files changed, 266 insertions(+), 12 deletions(-) create mode 100644 lib/webhooks/__tests__/dispatch-kick.test.ts create mode 100644 lib/webhooks/dispatch-kick.ts diff --git a/DECISIONS.md b/DECISIONS.md index 38e28114..ad792194 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -628,3 +628,4 @@ One line per decision: `[YYYY-MM-DD] : `. Appended by agents and [2026-07-27] Supplier-invoice edits (#1230): block invoice_date / supplier_invoice_number once registration_journal_entry_id is set, rather than propagating the change into the verifikat via correct_entry_metadata: the friendlier propagate option turns a metadata PUT into a bookkeeping write (voucher rättelse, rattelse-log, period-lock checks) and needs a deliberate product call; blocking is the minimal legally correct behaviour and leaves due_date/payment_reference/notes editable for the aged-invoice flow (#1206). [2026-07-27] SKV 403 classification (#1155 item 1): the MuleSoft APIGW body "The required scopes are not authorized" is checked BEFORE the token-scope patterns and maps to ACCESS_DENIED (user mode) / SYSTEM_AUTH_FAILED with a subscription message (system mode). Substring matching on 'required scope' collided with it and produced MISSING_SCOPE, which is a RECONSENT code, so every reconnect re-flagged the token row. Token-scope detection is now a positive match on invalid_scope or SKV's documented "required scope has been requested" sentence, not a loose substring. [2026-07-27] Docs export (#1247): scripts/export-docs-to-website.mts stubs `server-only` via a Module._load hook instead of untangling the import chain. The chain is real (lib/api/v1/load-routes -> every v1 route -> lib/init -> posthog-observability -> posthog-server) and the script only reads exported markdown builders, so breaking the chain would mean restructuring route imports for a build-time script's benefit. +[2026-07-27] Webhook delivery latency (#1201): implemented option (a), an emit-triggered kick of the existing dispatchDueDeliveries, not option (b), an authenticated SSE stream over event_log. The kick is a new lib/webhooks/dispatch-kick.ts wired into fanOutToWebhooks plus the two routes that enqueue a delivery directly (the :test verb and the manual delivery retry), and it does NOT close #1201: the issue asks for a realtime stream for API consumers and that remains open. Three constraints shaped it. It is never awaited: eventBus.emit is awaited at ~99 call sites including journal_entry.committed, and each delivery can burn a 10 s receiver timeout, so awaiting would put a stranger's HTTP endpoint on the critical path of committing a verifikat. It is coalesced per function instance, because a bulk operation emits once per row and would otherwise schedule one claim round trip per row. Its batch size is 5 rather than the cron's 50, because this work runs on the tail of a user-facing request. The SKIP LOCKED claim keeps a kick and the cron from claiming the same row at the same moment, but that is a claim-time guarantee only: the RPC autocommits before any POST, so a later cycle's recoverStuckInFlight sweep can re-arm a row still queued behind an earlier serial loop. Delivery stays at-least-once as the public docs already promise, and the kick's batch of 5 opens a narrower window than the cron's existing batch of 50 against the same 20 s stuck threshold; an early draft of this change claimed double delivery was impossible, which was wrong and is now corrected in the code comments. Scheduling uses next/server after() with a deferred-microtask fallback outside a request scope, mirroring the enable-banking callback; the fallback must stay deferred rather than inline, or the coalescing flag clears before the next kick in the same tick can see it. No API_V1_VERSION bump: no new event types and no payload change, only latency. diff --git a/app/api/v1/companies/[companyId]/webhooks/[id]/test/route.ts b/app/api/v1/companies/[companyId]/webhooks/[id]/test/route.ts index 769121aa..27f0f63e 100644 --- a/app/api/v1/companies/[companyId]/webhooks/[id]/test/route.ts +++ b/app/api/v1/companies/[companyId]/webhooks/[id]/test/route.ts @@ -2,14 +2,15 @@ * /api/v1/companies/{companyId}/webhooks/{id}/test: POST :test verb. * * Enqueues a synthetic `webhook.test` delivery against the configured - * receiver. The dispatcher cron picks it up at next-minute boundary - * exactly as it would a real event. The response returns the + * receiver and kicks a dispatch cycle immediately, exactly as a real event + * does (#1201); the cron remains the retry path. The response returns the * webhook_delivery_id so the caller can poll * GET /webhooks/{id}/deliveries?delivery_id=... to see the outcome. */ import { z } from 'zod' import { ok } from '@/lib/api/v1/response' +import { kickWebhookDispatch } from '@/lib/webhooks/dispatch-kick' import { registerEndpoint, dataEnvelope } from '@/lib/api/v1/registry' import { withApiV1 } from '@/lib/api/v1/with-api-v1' import { v1ErrorResponse, v1ErrorResponseFromCode } from '@/lib/api/v1/errors' @@ -26,7 +27,7 @@ registerEndpoint({ path: '/api/v1/companies/:companyId/webhooks/:id/test', summary: 'Send a synthetic test event to a webhook.', description: - 'Enqueues a webhook.test delivery against the configured receiver. The dispatcher delivers it on the next per-minute cron tick. Use the returned webhook_delivery_id to poll GET /webhooks/{id}/deliveries for the outcome.', + 'Enqueues a webhook.test delivery against the configured receiver and dispatches it immediately, so the outcome is normally available within a second or two rather than on the next per-minute cron tick. Use the returned webhook_delivery_id to poll GET /webhooks/{id}/deliveries for the outcome.', useWhen: 'After creating or modifying a webhook, before relying on it in production: to validate that the receiver is reachable and that signature verification works on the receiver side.', doNotUseFor: @@ -111,6 +112,11 @@ export const POST = withApiV1<{ params: Promise<{ companyId: string; id: string }) } + // Deliver now rather than on the next cron tick. This verb exists to tell + // someone whether their receiver works; making them wait up to a minute + // for that answer is the whole complaint in #1201. + kickWebhookDispatch() + return ok( { webhook_delivery_id: (delivery as { id: string }).id, status: 'pending' as const }, { requestId: ctx.requestId }, diff --git a/app/api/v1/webhook-deliveries/[id]/retry/route.ts b/app/api/v1/webhook-deliveries/[id]/retry/route.ts index 35931fd2..858f13b7 100644 --- a/app/api/v1/webhook-deliveries/[id]/retry/route.ts +++ b/app/api/v1/webhook-deliveries/[id]/retry/route.ts @@ -23,6 +23,7 @@ import { ok } from '@/lib/api/v1/response' import { registerEndpoint, dataEnvelope } from '@/lib/api/v1/registry' import { withApiV1 } from '@/lib/api/v1/with-api-v1' import { v1ErrorResponse, v1ErrorResponseFromCode } from '@/lib/api/v1/errors' +import { kickWebhookDispatch } from '@/lib/webhooks/dispatch-kick' import { minimisePayload } from '@/lib/webhooks/handler' import { validateWebhookUrl } from '@/lib/webhooks/url-guard' import { hasScope } from '@/lib/auth/api-keys' @@ -213,6 +214,11 @@ export const POST = withApiV1<{ params: Promise<{ id: string }> }>( }) } + // Deliver now rather than on the next cron tick (#1201). A manual retry + // is someone watching a failed delivery and asking for it again; the + // cron stays the path for the automatic backoff retries. + kickWebhookDispatch() + return ok( { webhook_delivery_id: (replay as { id: string }).id, status: 'pending' as const }, { requestId: ctx.requestId }, diff --git a/lib/docs/content/changelog.ts b/lib/docs/content/changelog.ts index 89c3d057..1fe903c0 100644 --- a/lib/docs/content/changelog.ts +++ b/lib/docs/content/changelog.ts @@ -55,7 +55,7 @@ The first stable release of the public REST API. Six phases of development cover ### 2026-05-15 — Webhooks (Phase 6 PR-1) - **Subscriptions**: \`POST /webhooks\` (HMAC secret returned exactly once), GET list + detail, PATCH, DELETE. Per-event-type elevated scope check (\`salary_run.*\` and \`agi.generated\` require \`payroll:read\`). -- **Delivery substrate**: per-minute Vercel cron at \`/api/webhooks/dispatch/cron\`. Due rows are claimed atomically via the \`claim_due_webhook_deliveries\` SQL function (\`FOR UPDATE SKIP LOCKED\`), with \`*.pg.test.ts\` coverage for the claim path and the webhook DB triggers. Exponential backoff \`1m / 5m / 30m / 2h / 12h / 24h / 48h\` (7 retries, ~87h total). HTTP 410 from receiver auto-disables the webhook. +- **Delivery substrate**: dispatched immediately after the event is enqueued, with a per-minute Vercel cron at \`/api/webhooks/dispatch/cron\` as the retry and sweep path. Due rows are claimed atomically via the \`claim_due_webhook_deliveries\` SQL function (\`FOR UPDATE SKIP LOCKED\`), with \`*.pg.test.ts\` coverage for the claim path and the webhook DB triggers. Exponential backoff \`1m / 5m / 30m / 2h / 12h / 24h / 48h\` (7 retries, ~87h total). HTTP 410 from receiver auto-disables the webhook. - **Signature**: \`X-Gnubok-Signature: t=,v1=\`. Stripe-format. Sample receivers in [Node + Python](/docs/api/webhooks#verifying-signatures). - **SSRF protection**: webhook_url must be HTTPS; resolved IPs in private/loopback/link-local/CGNAT/cloud-metadata ranges are rejected at create AND dispatch time. Dispatch pins the validated IP through a DNS-rebinding-safe \`node:https.request\` agent (\`lib/webhooks/pinned-fetch.ts\`); redirects are refused on every outbound POST (any 3xx is treated as a blocked redirect). - **Audit + retention**: webhook delivery rows are *behandlingshistorik* per BFNAR 2013:2 kap 8 §: immutable once terminal so the audit trail of what an integration was notified of stays intact. Delivery rows are NOT räkenskapsinformation themselves; the 7-year statutory retention under BFL 7 kap 1 § applies only to the underlying verifikation / faktura / AGI XML in its own table, NOT to the delivery envelope. Accounted keeps accounting-event delivery rows for 7 years as a voluntary operational policy (the duration aligns with BFL 7 kap on the underlying records but is not itself a statutory obligation on delivery rows). Webhook DELETE preserves the delivery audit trail (\`ON DELETE SET NULL\` on \`webhook_id\`). Webhook lifecycle events (create / update / delete, plus dispatcher auto-disable) each write a V16 \`audit_log\` entry. diff --git a/lib/docs/content/webhooks.ts b/lib/docs/content/webhooks.ts index 49f9f086..1cc5ffa1 100644 --- a/lib/docs/content/webhooks.ts +++ b/lib/docs/content/webhooks.ts @@ -8,7 +8,7 @@ If you've used [Stripe webhooks](https://docs.stripe.com/webhooks), the model is 1. **Register a receiver** with [\`POST /api/v1/companies/{companyId}/webhooks\`](/docs/api/reference/webhooks#post-webhooks-create). The response includes an HMAC signing secret returned **exactly once**: store it on the receiver side immediately. If you lose it, rotate it with [\`POST /api/v1/companies/{companyId}/webhooks/{webhookId}/rotate-secret\`](/docs/api/reference/webhooks#post-webhooks-rotate_secret): a fresh secret is issued in place and the old one is invalidated immediately, with no change to the webhook's id or delivery history. 2. **Accounted emits events** internally (e.g. an invoice is marked paid via the dashboard or another API call). The webhook handler enqueues a delivery row. -3. **The dispatcher cron runs every minute**, signs the payload with HMAC-SHA256, and POSTs to your URL with a 10-second timeout. +3. **The dispatcher runs immediately after the event**, signs the payload with HMAC-SHA256, and POSTs to your URL with a 10-second timeout. A cron sweep every minute picks up anything the immediate pass did not get to, so first-attempt latency is normally a second or two but is never guaranteed to be: treat delivery as prompt, not synchronous. 4. **Your receiver verifies the signature**, processes the event idempotently, and returns 2xx. 5. **Failed deliveries retry** at \`1m / 5m / 30m / 2h / 12h / 24h / 48h\` (7 retries, ~87 hours total, about 3.6 days). After all attempts the delivery is marked \`dead\`. HTTP 410 from your receiver short-circuits to \`dead\` immediately and **auto-disables** the webhook. @@ -219,7 +219,7 @@ Use [\`GET /api/v1/companies/{companyId}/webhooks/{webhookId}/deliveries\`](/doc To replay a \`dead\` or \`delivered\` delivery, call [\`POST /api/v1/webhook-deliveries/{deliveryId}/retry\`](/docs/api/reference/webhooks#post-webhook_deliveries-retry). The retry creates a fresh delivery row pointing at the same payload: the original audit row stays in place. Receivers must be idempotent on the \`X-Gnubok-Delivery\` header. -To send a synthetic test event without driving real state, call [\`POST /api/v1/companies/{companyId}/webhooks/{webhookId}/test\`](/docs/api/reference/webhooks#post-webhooks-test). The dispatcher delivers a \`webhook.test\` event with a static payload on the next per-minute tick. +To send a synthetic test event without driving real state, call [\`POST /api/v1/companies/{companyId}/webhooks/{webhookId}/test\`](/docs/api/reference/webhooks#post-webhooks-test). The dispatcher delivers a \`webhook.test\` event with a static payload immediately, so the outcome is normally visible within a second or two. ## Auto-disable behaviour diff --git a/lib/webhooks/__tests__/dispatch-kick.test.ts b/lib/webhooks/__tests__/dispatch-kick.test.ts new file mode 100644 index 00000000..f1b20c8c --- /dev/null +++ b/lib/webhooks/__tests__/dispatch-kick.test.ts @@ -0,0 +1,110 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' + +vi.mock('@/lib/auth/api-keys', () => ({ + createServiceClientNoCookies: vi.fn(() => ({ __client: true })), +})) + +// `after` is a no-op outside a request scope in these tests; the module falls +// back to a floating promise, which is the path a plain node server takes too. +vi.mock('next/server', () => ({ + after: () => { + throw new Error('after() called outside a request scope') + }, +})) + +import { kickWebhookDispatch, resetKickStateForTests, KICK_BATCH_SIZE } from '../dispatch-kick' + +/** Resolves after the microtask queue drains, so floating work has run. */ +const flush = () => new Promise((resolve) => setTimeout(resolve, 0)) + +beforeEach(() => { + vi.clearAllMocks() + resetKickStateForTests() +}) + +describe('kickWebhookDispatch', () => { + it('runs one dispatch cycle with the small kick batch size', async () => { + const dispatch = vi.fn().mockResolvedValue({ picked: 1, delivered: 1, failed: 0, dead: 0 }) + + kickWebhookDispatch(dispatch) + await flush() + + expect(dispatch).toHaveBeenCalledTimes(1) + expect(dispatch.mock.calls[0][0]).toMatchObject({ batchSize: KICK_BATCH_SIZE }) + }) + + it('claims far fewer rows than the cron, so a backlog cannot own a request tail', () => { + // Each delivery can burn the full 10 s receiver timeout; the cron's 50 is + // fine on a dedicated invocation, not on the tail of a user request. + expect(KICK_BATCH_SIZE).toBeLessThan(50) + }) + + it('returns synchronously: the emitter never waits on receiver HTTP', async () => { + let settled = false + const dispatch = vi.fn( + () => + new Promise((resolve) => { + setTimeout(() => { + settled = true + resolve({ picked: 0, delivered: 0, failed: 0, dead: 0 }) + }, 20) + }), + ) + + const returned = kickWebhookDispatch(dispatch as never) + + // eventBus.emit awaits its subscribers, so a kick that blocked here would + // put a stranger's slow endpoint on the critical path of committing a + // verifikat. + expect(returned).toBeUndefined() + expect(settled).toBe(false) + + await new Promise((resolve) => setTimeout(resolve, 40)) + expect(settled).toBe(true) + }) + + it('coalesces a burst of kicks into a single cycle', async () => { + // A bulk booking emits once per row. Without coalescing, 100 rows would + // mean 100 claim round trips against the same handful of due deliveries. + const dispatch = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 }) + + for (let i = 0; i < 100; i++) kickWebhookDispatch(dispatch) + await flush() + + expect(dispatch).toHaveBeenCalledTimes(1) + }) + + it('accepts a new kick once the previous cycle has started', async () => { + const dispatch = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 }) + + kickWebhookDispatch(dispatch) + await flush() + kickWebhookDispatch(dispatch) + await flush() + + expect(dispatch).toHaveBeenCalledTimes(2) + }) + + it('swallows a dispatch failure: a failed kick is latency, not a lost delivery', async () => { + const dispatch = vi.fn().mockRejectedValue(new Error('receiver unreachable')) + + expect(() => kickWebhookDispatch(dispatch)).not.toThrow() + await flush() + + expect(dispatch).toHaveBeenCalledTimes(1) + }) + + it('does not wedge the coalescing flag when a cycle fails', async () => { + // The cron still covers every row, but a stuck flag would silently + // disable the fast path for the life of the function instance. + const failing = vi.fn().mockRejectedValue(new Error('boom')) + kickWebhookDispatch(failing) + await flush() + + const ok = vi.fn().mockResolvedValue({ picked: 0, delivered: 0, failed: 0, dead: 0 }) + kickWebhookDispatch(ok) + await flush() + + expect(ok).toHaveBeenCalledTimes(1) + }) +}) diff --git a/lib/webhooks/dispatch-kick.ts b/lib/webhooks/dispatch-kick.ts new file mode 100644 index 00000000..48f031e8 --- /dev/null +++ b/lib/webhooks/dispatch-kick.ts @@ -0,0 +1,121 @@ +/** + * Emit-triggered webhook dispatch kick. + * + * Without this, the floor on webhook delivery latency is the per-minute cron + * at /api/webhooks/dispatch/cron: a delivery enqueued one second after a tick + * waits the remaining 59 before anyone looks at it. External consumers had no + * way around that except polling /api/events, which the 100 rpm per-key limit + * makes expensive and which still cannot beat the tick interval (issue #1201). + * + * So: once fanOutToWebhooks has inserted delivery rows, schedule one dispatch + * cycle immediately. The cron is unchanged and remains the retry and sweep + * path; this only moves the FIRST attempt forward. + * + * Three properties matter and are all load-bearing: + * + * 1. NEVER awaited by the emitter. eventBus.emit() is awaited at ~99 call + * sites, including journal_entry.committed, and dispatchDueDeliveries POSTs + * to receivers with a 10 s timeout each. Awaiting it here would put a + * stranger's slow HTTP endpoint on the critical path of committing a + * verifikat. kickWebhookDispatch() is synchronous and returns immediately. + * + * 2. Coalesced per instance. A bulk operation emits once per row, and each + * emission would otherwise schedule its own cycle: 100 bulk-booked + * transactions would mean 100 claim round trips. While one kick is + * outstanding, further kicks are dropped; the rows they enqueued are picked + * up by that kick, the next one, or the cron. + * + * 3. Never throws. It runs inside an event-bus subscriber, where an + * unhandled rejection would surface as a failed emit on a route that has + * already done its real work. + * + * claim_due_webhook_deliveries claims with FOR UPDATE SKIP LOCKED and flips + * rows to in_flight in the same statement, so a kick and the cron never claim + * the same row at the same moment. That is a claim-time guarantee only, and it + * is worth being precise about what it does NOT buy: the RPC autocommits, so + * its locks are gone before any POST is issued, and from then on ownership is + * just status='in_flight'. A later cycle's recoverStuckInFlight sweep can + * re-arm a row that is still queued behind an earlier cycle's serial loop. + * Delivery therefore stays at-least-once, exactly as the public docs promise + * ("the same delivery id may arrive more than once ... idempotency is on + * you"). The kick does not change that contract: the cron already claims 50 + * rows serially against the same 20 s stuck threshold, which is a wider window + * than this batch of 5 can open. + */ + +import { after } from 'next/server' +import { dispatchDueDeliveries } from './dispatcher' +import { createServiceClientNoCookies } from '@/lib/auth/api-keys' +import { createLogger } from '@/lib/logger' + +const log = createLogger('webhooks/dispatch-kick') + +/** + * Deliveries claimed per emit-triggered cycle. Deliberately far below the + * cron's 50: this work runs after a user-facing request on the same function + * instance, and each delivery can burn up to REQUEST_TIMEOUT_MS. Five covers + * the realistic fanout (one or two receivers for the event that just fired) + * without letting a backlog turn one request's tail into a minute of work. + * Anything beyond it is the cron's job, which is what the cron is for. + */ +export const KICK_BATCH_SIZE = 5 + +/** + * True while a kick is scheduled but has not started running. Module scope, + * so it is per function instance rather than global: two concurrent instances + * each get one in-flight kick, which is fine (SKIP LOCKED makes the claims + * disjoint). + */ +let kickPending = false + +/** Test seam: swap the dispatch implementation without a live Supabase. */ +type DispatchFn = typeof dispatchDueDeliveries + +/** + * Schedule one webhook dispatch cycle to run after the current response. + * Returns immediately; the caller must not await the delivery work. + */ +export function kickWebhookDispatch(dispatchImpl?: DispatchFn): void { + if (kickPending) return + kickPending = true + + const dispatch = dispatchImpl ?? dispatchDueDeliveries + + const run = async (): Promise => { + // Cleared first, not last: a kick scheduled while this cycle is running + // targets rows this cycle may already have passed, so it deserves its own + // slot. Clearing here also means a cycle that dies mid-flight cannot + // wedge the flag on for the life of the instance. + kickPending = false + try { + await dispatch({ + supabase: createServiceClientNoCookies(), + batchSize: KICK_BATCH_SIZE, + }) + } catch (err) { + // The cron will retry every one of these rows. A failed kick is a + // latency regression, never a lost delivery, so warn rather than error. + log.warn('emit-triggered webhook dispatch failed; cron will retry', { + error: err instanceof Error ? err.message : String(err), + }) + } + } + + try { + // Keeps the serverless instance alive past the response so the POSTs + // actually complete. Same pattern as the enable-banking callback. + after(() => run()) + } catch { + // Outside a request scope (unit tests, scripts, a plain node server): + // run it as a floating promise instead. Deferred rather than called + // inline, so this path coalesces a synchronous burst exactly like the + // after() path does: calling run() here would clear kickPending before + // the next kick in the same tick ever sees it. run() never rejects. + queueMicrotask(() => void run()) + } +} + +/** Test-only: reset the coalescing flag between cases. */ +export function resetKickStateForTests(): void { + kickPending = false +} diff --git a/lib/webhooks/dispatcher.ts b/lib/webhooks/dispatcher.ts index 90eaf5f7..5e9db252 100644 --- a/lib/webhooks/dispatcher.ts +++ b/lib/webhooks/dispatcher.ts @@ -1,10 +1,12 @@ /** * Webhook delivery dispatcher. * - * Invoked from the per-minute cron at /api/webhooks/dispatch/cron. Picks up - * pending + retry-due deliveries (FOR UPDATE SKIP LOCKED so multiple cron - * invocations don't double-deliver), POSTs each one with HMAC signature, - * and updates the row to one of: + * Invoked from two places: the per-minute cron at /api/webhooks/dispatch/cron, + * and an emit-triggered kick right after deliveries are enqueued + * (lib/webhooks/dispatch-kick.ts), which is what keeps first-attempt latency + * off the cron interval. Picks up pending + retry-due deliveries (FOR UPDATE + * SKIP LOCKED, so a kick racing the cron claims disjoint rows), POSTs each one + * with HMAC signature, and updates the row to one of: * * - delivered (2xx response) : terminal * - failed (5xx / network / 4xx : non-terminal until attempts diff --git a/lib/webhooks/handler.ts b/lib/webhooks/handler.ts index 813d78f0..386cd6d7 100644 --- a/lib/webhooks/handler.ts +++ b/lib/webhooks/handler.ts @@ -3,8 +3,9 @@ * * Subscribes to every CoreEventType the v1 API surface emits and converts * each emission into N rows in `webhook_deliveries`: one per active webhook - * subscribed to (company_id, event_type). The dispatcher cron picks them - * up at next-minute boundary and POSTs to the receiver. + * subscribed to (company_id, event_type). A dispatch cycle is then scheduled + * immediately (see dispatch-kick.ts); the per-minute cron stays as the retry + * and sweep path. * * Wired from lib/init.ts via registerWebhookHandler() so every API route * that calls ensureInitialized() gets the subscription wired exactly once. @@ -26,6 +27,7 @@ import type { CoreEventType } from '@/lib/events/types' import { createServiceClientNoCookies } from '@/lib/auth/api-keys' import { createLogger } from '@/lib/logger' import { API_V1_VERSION } from '@/lib/api/v1/version' +import { kickWebhookDispatch } from './dispatch-kick' const log = createLogger('webhooks/handler') @@ -184,5 +186,11 @@ async function fanOutToWebhooks(args: { eventType: args.eventType, webhookCount: rows.length, }) + return } + + // Deliver now instead of waiting for the next cron tick (#1201). Scheduled, + // never awaited: see lib/webhooks/dispatch-kick.ts for why the emitter must + // not block on a receiver's HTTP endpoint. + kickWebhookDispatch() }