From 8ddcea45c1d2d6479b3488088111c3cf7f579264 Mon Sep 17 00:00:00 2001 From: Nikos Benakis Date: Fri, 13 Nov 2026 23:07:03 +0200 Subject: [PATCH 1/2] feat: envelope, deliveryHeaders and nonRetryableStatuses options For receivers with their own delivery conventions, such as MCP Events webhook mode: send the payload as the body, add per-attempt headers, and fail a delivery at once on statuses like 410 without hurting endpoint health. Co-Authored-By: Claude Opus 5.5 --- .changeset/delivery-options.md | 5 + docs/reference/configuration.md | 3 + src/hookgres.ts | 117 +++++++++++++++++++++++- src/index.ts | 1 + test/delivery-options.test.ts | 156 ++++++++++++++++++++++++++++++++ 5 files changed, 279 insertions(+), 3 deletions(-) create mode 100644 .changeset/delivery-options.md create mode 100644 test/delivery-options.test.ts diff --git a/.changeset/delivery-options.md b/.changeset/delivery-options.md new file mode 100644 index 0000000..4d65308 --- /dev/null +++ b/.changeset/delivery-options.md @@ -0,0 +1,5 @@ +--- +"hookgres": minor +--- + +Add three delivery options for receivers with their own conventions, such as MCP Events webhook mode: `envelope: false` sends the payload itself as the body, `deliveryHeaders` adds request headers per attempt, and `nonRetryableStatuses` fails a delivery at once on statuses like `410` without counting against the endpoint's health. diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index c03e329..dfbb168 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -44,6 +44,9 @@ const hookgres = new Hookgres({ | `responseBodyLimit` | `4096` | Characters of each receiver response to store in the attempt log, from 0 (none) to 4096. See [Privacy and GDPR](/guides/privacy#limit-stored-response-bodies). | | `blockPrivateAddresses` | `false` | `true` refuses to send to private and internal addresses; `{ allow: [...] }` does the same except for the listed IPs and CIDRs. See [Block internal addresses](/guides/production#block-internal-addresses-ssrf). | | `fetch` | `globalThis.fetch` | Custom fetch, for a proxy or tests. With `blockPrivateAddresses`, only endpoint URLs are checked; delivery uses this fetch as is. | +| `envelope` | `true` | `true` wraps each payload in the Standard Webhooks body `{"type", "timestamp", "data"}`. `false` sends the payload itself as the body, for receivers that define their own, like MCP Events. Workers and senders on one schema must agree. | +| `deliveryHeaders` | — | `(delivery) => headers`: extra request headers per attempt, e.g. a routing id. Gets `deliveryId`, `messageId`, `endpointId`, `tenantId`, `eventType` and `attemptNumber`. Can't set `content-type`, `user-agent` or `webhook-*`; if it throws, the attempt fails and is retried. | +| `nonRetryableStatuses` | `[]` | Response statuses (300–599) that fail a delivery at once with no more retries, e.g. `[410, 413]`. They don't count against the endpoint's health. | | `userAgent` | `"Hookgres/0.1"` | The `user-agent` header on every request. | | `onError` | `console.error` | Called with unexpected worker errors, like a lost database connection, and with idle-connection errors from a pool Hookgres created itself. Failed deliveries are not errors. | | `onAttempt` | — | Called with an [`AttemptEvent`](/reference/api#types) after every HTTP attempt. See [Metrics and tracing](/guides/production#metrics-and-tracing). | diff --git a/src/hookgres.ts b/src/hookgres.ts index 1c3c623..14c2d64 100644 --- a/src/hookgres.ts +++ b/src/hookgres.ts @@ -104,6 +104,26 @@ const DOCS = "https://hookgres.dev/docs"; * * @see https://hookgres.dev/docs/reference/configuration */ +/** What `deliveryHeaders` gets for each attempt. */ +export interface DeliveryHeadersInput { + deliveryId: string; + messageId: string; + endpointId: string; + tenantId: string; + eventType: string; + /** 1 for the first attempt. */ + attemptNumber: number; +} + +/** Headers `deliveryHeaders` can't set: Hookgres owns them. */ +const RESERVED_HEADERS = new Set([ + "content-type", + "user-agent", + "webhook-id", + "webhook-timestamp", + "webhook-signature", +]); + export interface HookgresOptions { /** * An existing `pg.Pool`, usually the one your app already uses, so @@ -198,6 +218,31 @@ export interface HookgresOptions { * endpoint URLs are still checked, but delivery uses this fetch as is. */ fetch?: typeof fetch; + /** + * `true` (the default) wraps each payload in the Standard Webhooks body + * `{"type", "timestamp", "data"}`. `false` sends the payload itself as the + * request body, for receivers that define their own body, like MCP Events. + * Workers and senders sharing a schema must use the same setting. + * @see https://hookgres.dev/docs/reference/configuration + */ + envelope?: boolean; + /** + * Extra request headers for each attempt, e.g. a routing id the receiver + * reads before it parses the body. Called once per attempt. It can't set + * `content-type`, `user-agent` or the `webhook-*` headers. If it throws, the + * attempt fails with that error and is retried. + * + * @example + * deliveryHeaders: ({ tenantId }) => ({ "x-account-id": tenantId }) + */ + deliveryHeaders?: (delivery: DeliveryHeadersInput) => Record; + /** + * Response statuses that fail a delivery at once, with no more retries, e.g. + * `[410, 413]` for receivers that use them to reject one delivery. They + * don't count against the endpoint's health. Defaults to `[]`: every + * non-2xx response is retried. + */ + nonRetryableStatuses?: number[]; /** `User-Agent` header sent with every webhook. Defaults to `Hookgres/0.1`. */ userAgent?: string; /** Called for unexpected worker errors, and for idle-connection errors on a pool Hookgres created from `connectionString`. Defaults to `console.error`. */ @@ -327,6 +372,9 @@ export class Hookgres { private readonly transport: Transport; private readonly closeTransport: () => void; private readonly userAgent: string; + private readonly envelope: boolean; + private readonly deliveryHeaders: HookgresOptions["deliveryHeaders"]; + private readonly nonRetryableStatuses: Set; private readonly onError: (error: unknown) => void; private readonly onAttempt: HookgresOptions["onAttempt"]; private readonly onEndpointDisabled: HookgresOptions["onEndpointDisabled"]; @@ -509,6 +557,33 @@ export class Hookgres { } } this.userAgent = options.userAgent ?? "Hookgres/0.1"; + this.envelope = options.envelope ?? true; + if (typeof this.envelope !== "boolean") { + throw new HookgresValidationError( + `envelope must be true (Standard Webhooks body, the default) or false (the payload is the body). See ${DOCS}/reference/configuration`, + ); + } + if ( + options.deliveryHeaders !== undefined && + typeof options.deliveryHeaders !== "function" + ) { + throw new HookgresValidationError( + `deliveryHeaders must be a function that returns extra headers, e.g. ({ tenantId }) => ({ "x-account-id": tenantId }). See ${DOCS}/reference/configuration`, + ); + } + this.deliveryHeaders = options.deliveryHeaders; + const statuses = options.nonRetryableStatuses ?? []; + if ( + !Array.isArray(statuses) || + statuses.some( + (status) => !Number.isInteger(status) || status < 300 || status > 599, + ) + ) { + throw new HookgresValidationError( + `nonRetryableStatuses must be HTTP status codes from 300 to 599, e.g. [410, 413]. See ${DOCS}/reference/configuration`, + ); + } + this.nonRetryableStatuses = new Set(statuses); const onError = options.onError ?? ((error) => console.error("[hookgres] worker error:", error)); @@ -2042,7 +2117,9 @@ export class Hookgres { let body = bodies.get(delivery.message_id); if (body === undefined) { const timestamp = new Date(delivery.message_created_ms).toISOString(); - body = `{"type":${JSON.stringify(delivery.event_type)},"timestamp":"${timestamp}","data":${payload}}`; + body = this.envelope + ? `{"type":${JSON.stringify(delivery.event_type)},"timestamp":"${timestamp}","data":${payload}}` + : payload; bodies.set(delivery.message_id, body); } delivery.body = body; @@ -2365,7 +2442,12 @@ export class Hookgres { return; } - const isFinal = attemptNumber > this.retrySchedule.length; + // A status the receiver uses to reject this one delivery: fail it now, + // and leave the endpoint's health alone, since the receiver answered. + const rejected = + outcome.statusCode !== null && + this.nonRetryableStatuses.has(outcome.statusCode); + const isFinal = rejected || attemptNumber > this.retrySchedule.length; const baseDelayMs = isFinal ? 0 : (this.retrySchedule[attemptNumber - 1] ?? 0); @@ -2417,7 +2499,7 @@ export class Hookgres { ELSE disabled_xid END, updated_at = now() - WHERE id IN (SELECT endpoint_id FROM cur) + WHERE id IN (SELECT endpoint_id FROM cur) AND NOT $11::boolean RETURNING id, disabled AND disabled_xid = pg_current_xact_id() AS disabled_now, -- Disabled automatically, now or earlier: no more retries. -- A manual disable is a pause, so retries continue. @@ -2454,6 +2536,7 @@ export class Hookgres { isFinal, delayMs, this.disableEndpointAfterMs, + rejected, ], this.workerDb, ); @@ -2479,6 +2562,33 @@ export class Hookgres { } } + /** `deliveryHeaders`' result, lowercased and checked. Throws on misuse. */ + private extraHeaders(delivery: ClaimedDelivery): Record { + if (!this.deliveryHeaders) return {}; + const headers = this.deliveryHeaders({ + deliveryId: delivery.id, + messageId: delivery.message_id, + endpointId: delivery.endpoint_id, + tenantId: delivery.tenant_id, + eventType: delivery.event_type, + attemptNumber: delivery.attempt_count, + }); + const result: Record = {}; + for (const [name, value] of Object.entries(headers ?? {})) { + const key = name.toLowerCase(); + if (RESERVED_HEADERS.has(key)) { + throw new Error(`deliveryHeaders can't set "${key}"`); + } + if (typeof value !== "string") { + throw new Error( + `deliveryHeaders returned a non-string value for "${key}"`, + ); + } + result[key] = value; + } + return result; + } + private async attempt(delivery: ClaimedDelivery): Promise { if (delivery.loadError) { return { @@ -2503,6 +2613,7 @@ export class Hookgres { const init: TransportRequest = { method: "POST", headers: { + ...this.extraHeaders(delivery), "content-type": "application/json", "user-agent": this.userAgent, "webhook-id": delivery.message_id, diff --git a/src/index.ts b/src/index.ts index 2ec0b83..63e2ddf 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,6 +1,7 @@ export { HookgresValidationError } from "./errors.js"; export { DEFAULT_RETRY_SCHEDULE, + type DeliveryHeadersInput, Hookgres, type HookgresOptions, } from "./hookgres.js"; diff --git a/test/delivery-options.test.ts b/test/delivery-options.test.ts new file mode 100644 index 0000000..803dd27 --- /dev/null +++ b/test/delivery-options.test.ts @@ -0,0 +1,156 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { HookgresValidationError, verifyWebhook } from "../src/index.js"; +import { drain, type Receiver, startReceiver, useHookgres } from "./helpers.js"; + +const ctx = useHookgres(); +let receiver: Receiver; + +beforeEach(async () => { + receiver = await startReceiver(); +}); +afterEach(async () => { + await receiver.close(); +}); + +describe("envelope: false", () => { + it("sends the payload itself as the signed body", async () => { + const hw = ctx.create({ envelope: false }); + const ep = await hw.createEndpoint({ tenantId: "acme", url: receiver.url }); + const payload = { + eventId: "evt_1", + name: "incident.created", + data: { a: 1 }, + }; + await hw.send({ tenantId: "acme", eventType: "incident.created", payload }); + await drain(hw); + + expect(receiver.requests).toHaveLength(1); + const request = receiver.requests[0]; + if (!request) throw new Error("no request"); + expect(JSON.parse(request.body)).toEqual(payload); + expect(() => + verifyWebhook({ + secret: ep.secret, + headers: request.headers, + body: request.body, + }), + ).not.toThrow(); + }); + + it("keeps the Standard Webhooks envelope by default", async () => { + const hw = ctx.create(); + await hw.createEndpoint({ tenantId: "acme", url: receiver.url }); + await hw.send({ tenantId: "acme", eventType: "a.b", payload: { x: 1 } }); + await drain(hw); + const body = JSON.parse(receiver.requests[0]?.body ?? "{}"); + expect(body).toMatchObject({ type: "a.b", data: { x: 1 } }); + }); + + it("rejects a non-boolean", () => { + expect(() => ctx.create({ envelope: "no" as unknown as boolean })).toThrow( + HookgresValidationError, + ); + }); +}); + +describe("deliveryHeaders", () => { + it("adds headers to every attempt, with the attempt number", async () => { + const seen: number[] = []; + const hw = ctx.create({ + deliveryHeaders: ({ tenantId, attemptNumber }) => { + seen.push(attemptNumber); + return { "X-Account-Id": tenantId }; + }, + }); + receiver.respond = (n) => (n === 1 ? 500 : 200); + await hw.createEndpoint({ tenantId: "acme", url: receiver.url }); + await hw.send({ tenantId: "acme", eventType: "a.b", payload: {} }); + await drain(hw); + + expect(receiver.requests).toHaveLength(2); + for (const request of receiver.requests) { + expect(request.headers["x-account-id"]).toBe("acme"); + } + expect(seen).toEqual([1, 2]); + }); + + it("can't override the signature headers: the attempt fails and is retried", async () => { + let calls = 0; + const hw = ctx.create({ + deliveryHeaders: () => { + calls++; + return calls === 1 + ? { "Webhook-Id": "forged" } + : ({} as Record); + }, + }); + const ep = await hw.createEndpoint({ tenantId: "acme", url: receiver.url }); + await hw.send({ tenantId: "acme", eventType: "a.b", payload: {} }); + await drain(hw); + + expect(receiver.requests).toHaveLength(1); + expect(receiver.requests[0]?.headers["webhook-id"]).not.toBe("forged"); + const [delivery] = await hw.listDeliveries({ endpointId: ep.id }); + const attempts = await hw.listAttempts(delivery?.id ?? ""); + expect(attempts[0]?.error).toContain(`can't set "webhook-id"`); + expect(delivery?.status).toBe("succeeded"); + }); + + it("rejects a non-function", () => { + expect(() => + ctx.create({ + deliveryHeaders: {} as unknown as () => Record, + }), + ).toThrow(HookgresValidationError); + }); +}); + +describe("nonRetryableStatuses", () => { + it("fails the delivery on the first such response, without retries", async () => { + const hw = ctx.create({ nonRetryableStatuses: [410, 413] }); + receiver.respond = () => 410; + const ep = await hw.createEndpoint({ tenantId: "acme", url: receiver.url }); + await hw.send({ tenantId: "acme", eventType: "a.b", payload: {} }); + await drain(hw); + + expect(receiver.requests).toHaveLength(1); + const [delivery] = await hw.listDeliveries({ endpointId: ep.id }); + expect(delivery?.status).toBe("failed"); + }); + + it("doesn't count against the endpoint's health", async () => { + const hw = ctx.create({ + nonRetryableStatuses: [410], + disableEndpointAfterMs: 1, + }); + receiver.respond = () => 410; + const ep = await hw.createEndpoint({ tenantId: "acme", url: receiver.url }); + await hw.send({ tenantId: "acme", eventType: "a.b", payload: {} }); + await drain(hw); + await new Promise((resolve) => setTimeout(resolve, 10)); + await hw.send({ tenantId: "acme", eventType: "a.b", payload: {} }); + await drain(hw); + + const endpoint = await hw.getEndpoint(ep.id); + expect(endpoint?.disabled).toBe(false); + expect(endpoint?.failingSince).toBeNull(); + expect(receiver.requests).toHaveLength(2); + }); + + it("still retries other statuses", async () => { + const hw = ctx.create({ nonRetryableStatuses: [410] }); + receiver.respond = (n) => (n === 1 ? 500 : 200); + await hw.createEndpoint({ tenantId: "acme", url: receiver.url }); + await hw.send({ tenantId: "acme", eventType: "a.b", payload: {} }); + await drain(hw); + expect(receiver.requests).toHaveLength(2); + }); + + it("rejects values that aren't 300-599 status codes", () => { + for (const bad of [[200], [600], [410.5], "410"]) { + expect(() => + ctx.create({ nonRetryableStatuses: bad as unknown as number[] }), + ).toThrow(HookgresValidationError); + } + }); +}); From a3af4736672c0937504d9103eebc6f5fb7dd054d Mon Sep 17 00:00:00 2001 From: Nikos Benakis Date: Fri, 13 Nov 2026 23:23:34 +0200 Subject: [PATCH 2/2] docs: envelope is decided by the worker Co-Authored-By: Claude Opus 5.5 --- docs/reference/configuration.md | 2 +- src/hookgres.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index dfbb168..f70f70d 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -44,7 +44,7 @@ const hookgres = new Hookgres({ | `responseBodyLimit` | `4096` | Characters of each receiver response to store in the attempt log, from 0 (none) to 4096. See [Privacy and GDPR](/guides/privacy#limit-stored-response-bodies). | | `blockPrivateAddresses` | `false` | `true` refuses to send to private and internal addresses; `{ allow: [...] }` does the same except for the listed IPs and CIDRs. See [Block internal addresses](/guides/production#block-internal-addresses-ssrf). | | `fetch` | `globalThis.fetch` | Custom fetch, for a proxy or tests. With `blockPrivateAddresses`, only endpoint URLs are checked; delivery uses this fetch as is. | -| `envelope` | `true` | `true` wraps each payload in the Standard Webhooks body `{"type", "timestamp", "data"}`. `false` sends the payload itself as the body, for receivers that define their own, like MCP Events. Workers and senders on one schema must agree. | +| `envelope` | `true` | `true` wraps each payload in the Standard Webhooks body `{"type", "timestamp", "data"}`. `false` sends the payload itself as the body, for receivers that define their own, like MCP Events. The body is built when a worker sends, so the workers' setting is the one that counts. | | `deliveryHeaders` | — | `(delivery) => headers`: extra request headers per attempt, e.g. a routing id. Gets `deliveryId`, `messageId`, `endpointId`, `tenantId`, `eventType` and `attemptNumber`. Can't set `content-type`, `user-agent` or `webhook-*`; if it throws, the attempt fails and is retried. | | `nonRetryableStatuses` | `[]` | Response statuses (300–599) that fail a delivery at once with no more retries, e.g. `[410, 413]`. They don't count against the endpoint's health. | | `userAgent` | `"Hookgres/0.1"` | The `user-agent` header on every request. | diff --git a/src/hookgres.ts b/src/hookgres.ts index 14c2d64..cdaccda 100644 --- a/src/hookgres.ts +++ b/src/hookgres.ts @@ -222,7 +222,7 @@ export interface HookgresOptions { * `true` (the default) wraps each payload in the Standard Webhooks body * `{"type", "timestamp", "data"}`. `false` sends the payload itself as the * request body, for receivers that define their own body, like MCP Events. - * Workers and senders sharing a schema must use the same setting. + * The body is built at delivery time, so the workers' setting is what counts. * @see https://hookgres.dev/docs/reference/configuration */ envelope?: boolean;