[factory] Delivery options: envelope, deliveryHeaders, nonRetryableStatuses — pr-open

.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.

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. 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.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
 | `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).                                                                                                                                                                                                                                                                                                                                                                           |

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<E extends EventMap = AnyEvents> {
   /**
    * An existing `pg.Pool`, usually the one your app already uses, so
@@ -198,6 +218,31 @@ export interface HookgresOptions<E extends EventMap = AnyEvents> {
    * 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.
+   * The body is built at delivery time, so the workers' setting is what counts.
+   * @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<string, string>;
+  /**
+   * 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<E extends EventMap = AnyEvents> {
   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<number>;
   private readonly onError: (error: unknown) => void;
   private readonly onAttempt: HookgresOptions["onAttempt"];
   private readonly onEndpointDisabled: HookgresOptions["onEndpointDisabled"];
@@ -509,6 +557,33 @@ export class Hookgres<E extends EventMap = AnyEvents> {
       }
     }
     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<E extends EventMap = AnyEvents> {
       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<E extends EventMap = AnyEvents> {
       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<E extends EventMap = AnyEvents> {
                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<E extends EventMap = AnyEvents> {
         isFinal,
         delayMs,
         this.disableEndpointAfterMs,
+        rejected,
       ],
       this.workerDb,
     );
@@ -2479,6 +2562,33 @@ export class Hookgres<E extends EventMap = AnyEvents> {
     }
   }
 
+  /** `deliveryHeaders`' result, lowercased and checked. Throws on misuse. */
+  private extraHeaders(delivery: ClaimedDelivery): Record<string, string> {
+    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<string, string> = {};
+    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<HttpResult> {
     if (delivery.loadError) {
       return {
@@ -2503,6 +2613,7 @@ export class Hookgres<E extends EventMap = AnyEvents> {
       const init: TransportRequest = {
         method: "POST",
         headers: {
+          ...this.extraHeaders(delivery),
           "content-type": "application/json",
           "user-agent": this.userAgent,
           "webhook-id": delivery.message_id,

src/index.ts

@@ -1,6 +1,7 @@
 export { HookgresValidationError } from "./errors.js";
 export {
   DEFAULT_RETRY_SCHEDULE,
+  type DeliveryHeadersInput,
   Hookgres,
   type HookgresOptions,
 } from "./hookgres.js";

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<string, string>);
+      },
+    });
+    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<string, string>,
+      }),
+    ).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);
+    }
+  });
+});