feat(spec,middleware): batched webhook delivery

docs/current/SPECIFICATION.md

@@ -673,6 +673,63 @@ conditional requests (§3.2.1), compression (§3.2.2) and subscription filters
 (§5.1.3) first: those remove whole responses and whole deliveries rather than
 shrinking them.
 
+### 5.2.2 Batched delivery (normative)
+
+§5.2 mandates one event per `POST`. A high-frequency entity therefore pays a
+TLS handshake, a signature computation and a 10-second acknowledgement
+round-trip **per event** — the dominant per-delivery cost, and one that does
+not shrink with a smaller payload.
+
+Publishers MAY offer batching; subscribers opt in with `max_batch_size` on the
+subscription request. The wire format is the CloudEvents batched JSON
+encoding, so this is a media type change rather than an EEP invention:
+
+```http
+POST /hooks/eep
+content-type: application/cloudevents-batch+json
+webhook-id: msg_batch_01HN3QK7GX
+webhook-timestamp: 1708123456
+webhook-signature: v1,…
+
+[ { "specversion": "1.0", "id": "evt-1", … }, { "specversion": "1.0", "id": "evt-2", … } ]
+```
+
+- The body is a JSON array of envelopes. A batch MUST NOT be empty.
+- `max_batch_size` bounds the array. `1`, or omitted, preserves today's
+  one-event-per-POST behaviour exactly.
+- A subscription that opted into batching MUST always receive
+  `application/cloudevents-batch+json`, even when only one event is ready.
+  Switching format based on how many events happened to be available would
+  give the subscriber two parsing paths and no way to predict which it gets.
+- `max_batch_wait_ms` bounds the latency batching adds: the publisher MUST
+  deliver once it elapses even if the batch is not full. Without it, a
+  low-traffic subscription could hold an event indefinitely waiting for a
+  batch that never fills.
+- Events within a batch MUST be ordered as they were emitted for a given
+  `source`.
+- All events in one batch MUST belong to one subscription. Batching is a
+  transport optimisation and MUST NOT become a way to deliver events across
+  subscription boundaries.
+
+**Signing.** The signature covers the raw body — the whole array — exactly as
+in §5.3. The `webhook-id` identifies the *batch*, not any single event.
+
+**Acknowledgement is all-or-nothing.** A `2xx` acknowledges every event in the
+batch; any other response fails the whole batch, which is retried whole. A
+subscriber that cannot process one event in a batch MUST NOT return `2xx` and
+then drop it. This is why batching is opt-in: a subscriber must be able to
+process a batch atomically, or ask for `max_batch_size: 1`.
+
+**Deduplication is per event, not per batch.** Each envelope keeps its own
+`id`, so the idempotency rule in
+[delivery_guarantees.md](./delivery_guarantees.md) §6 applies unchanged: on a
+retry a subscriber discards the events it already processed and processes the
+rest.
+
+**Failure accounting.** A failed batch counts as **one** failed delivery
+against the §10 counter, not one per event — otherwise a single failure of a
+100-event batch would pause a subscription instantly.
+
 ### 5.3 Webhook signature verification
 
 The `webhook-signature` header contains an HMAC-SHA256 signature over the concatenation of:

packages/@eep-dev/middleware/src/core/eep-server.ts

@@ -430,6 +430,14 @@ export class EEPServer {
         failure_count: 0,
         expires_at: expiresAt.toISOString(),
         ...(filter ? { filter } : {}),
+        ...(typeof body.max_batch_size === "number" && body.max_batch_size > 1
+          ? {
+              max_batch_size: Math.min(Math.floor(body.max_batch_size), 500),
+              ...(typeof body.max_batch_wait_ms === "number"
+                ? { max_batch_wait_ms: Math.max(0, Math.min(Math.floor(body.max_batch_wait_ms), 30_000)) }
+                : {})
+            }
+          : {}),
         ...(body.delivery_format === "cloudevents/v1.0-binary"
           ? { delivery_format: "cloudevents/v1.0-binary" as const }
           : {}),

packages/@eep-dev/middleware/src/core/request-handler.ts

@@ -94,6 +94,13 @@ export type SubscriptionRecord = {
    * Defaults to structured.
    */
   delivery_format?: DeliveryFormat;
+  /**
+   * Maximum events combined into one delivery (SPECIFICATION.md §5.2.2).
+   * 1 or absent preserves one-event-per-POST.
+   */
+  max_batch_size?: number;
+  /** Milliseconds the publisher may hold an event waiting for the batch to fill. */
+  max_batch_wait_ms?: number;
   /** Requested access tier; matched against gate config on delivery. */
   tier?: string;
   created_at: string;

packages/@eep-dev/middleware/src/dispatcher/content-mode.ts

@@ -66,3 +66,21 @@ export function renderDelivery(event: CloudEvent, format: DeliveryFormat | undef
 
     return { body, headers };
 }
+
+/** CloudEvents batched JSON media type (SPECIFICATION.md §5.2.2). */
+export const BATCH_CONTENT_TYPE = "application/cloudevents-batch+json";
+
+/**
+ * Render a batch of events as one delivery.
+ *
+ * Batching is structured-mode only: binary content mode maps attributes to
+ * headers, and a batch has many events with different attribute values, so
+ * there is nowhere for them to go. A subscriber asking for both gets
+ * structured batches.
+ */
+export function renderBatch(events: CloudEvent[]): RenderedDelivery {
+    return {
+        body: JSON.stringify(events),
+        headers: { "content-type": BATCH_CONTENT_TYPE },
+    };
+}

packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from "vitest";
 import { EEPSigner, generateSigningKeyPair, verifyEd25519 } from "@eep-dev/signer";
 import { WebhookDispatcher, DEFAULT_RETRY_SCHEDULE_MS, type WebhookHttpClient } from "./webhook-dispatcher.js";
+import { BATCH_CONTENT_TYPE } from "./content-mode.js";
 import { InMemoryDBAdapter } from "../db/in-memory.js";
 import { InMemoryEventBusAdapter } from "../event-bus/in-memory.js";
 import { TEST_DELIVERY_EVENT_TYPE } from "../core/request-handler.js";
@@ -59,6 +60,121 @@ describe("WebhookDispatcher", () => {
     expect(DEFAULT_RETRY_SCHEDULE_MS).toEqual([0, 5_000, 30_000, 120_000, 900_000, 3_600_000, 21_600_000]);
   });
 
+  // SPECIFICATION.md §5.2.2 — one event per POST means a high-frequency
+  // entity pays a TLS handshake, a signature and a 10s ack round-trip per
+  // event. Batching amortises all three.
+  describe("batched delivery (§5.2.2)", () => {
+    const setup = async (maxBatchSize: number, script: Array<number | "throw"> = [200]) => {
+      const db = new InMemoryDBAdapter();
+      await db.saveSubscription(subscription({ max_batch_size: maxBatchSize }));
+      const { client, calls } = mockClient(script);
+      const dispatcher = new WebhookDispatcher({ db, httpClient: client, retryScheduleMs: NO_DELAY });
+      return { db, dispatcher, calls };
+    };
+
+    const batch = (n: number) => Array.from({ length: n }, (_, i) => event({ id: `evt_${i + 1}` }));
+
+    it("combines events into one POST with the CloudEvents batch media type", async () => {
+      const { dispatcher, calls } = await setup(10);
+      const results = await dispatcher.dispatchBatch(batch(3), "sub_1");
+
+      expect(calls).toHaveLength(1);
+      expect(calls[0]!.headers["content-type"]).toBe(BATCH_CONTENT_TYPE);
+      expect(JSON.parse(calls[0]!.body)).toHaveLength(3);
+      // One result per event, so callers still see per-event outcomes.
+      expect(results).toHaveLength(3);
+      expect(results.every((r) => r.delivered)).toBe(true);
+    });
+
+    it("keeps each envelope's own id so per-event dedup still works", async () => {
+      const { dispatcher, calls } = await setup(10);
+      await dispatcher.dispatchBatch(batch(3), "sub_1");
+      const sent = JSON.parse(calls[0]!.body) as Array<{ id: string }>;
+      expect(sent.map((e) => e.id)).toEqual(["evt_1", "evt_2", "evt_3"]);
+    });
+
+    it("signs the whole array once", async () => {
+      const { dispatcher, calls } = await setup(10);
+      await dispatcher.dispatchBatch(batch(3), "sub_1");
+      const call = calls[0]!;
+      expect(
+        new EEPSigner(SECRET).verify(
+          call.headers["webhook-id"]!,
+          call.headers["webhook-timestamp"]!,
+          call.headers["webhook-signature"]!,
+          call.body
+        )
+      ).toBe(true);
+    });
+
+    it("splits into several deliveries at max_batch_size", async () => {
+      const { dispatcher, calls } = await setup(2, [200, 200]);
+      const results = await dispatcher.dispatchBatch(batch(4), "sub_1");
+      expect(calls).toHaveLength(2);
+      expect(JSON.parse(calls[0]!.body)).toHaveLength(2);
+      expect(JSON.parse(calls[1]!.body)).toHaveLength(2);
+      expect(results).toHaveLength(4);
+    });
+
+    // max_batch_size 1 must be byte-identical to the unbatched path.
+    it("sends a single event unbatched when max_batch_size is 1", async () => {
+      const { dispatcher, calls } = await setup(1);
+      await dispatcher.dispatchBatch(batch(1), "sub_1");
+      expect(calls[0]!.headers["content-type"]).toBe("application/json");
+      expect(JSON.parse(calls[0]!.body)).toMatchObject({ id: "evt_1" });
+    });
+
+    it("applies the subscription filter before batching", async () => {
+      const db = new InMemoryDBAdapter();
+      await db.saveSubscription(
+        subscription({
+          max_batch_size: 10,
+          filter: { match: "all", conditions: [{ path: "id", op: "eq", value: "evt_2" }] }
+        })
+      );
+      const { client, calls } = mockClient([200]);
+      const dispatcher = new WebhookDispatcher({ db, httpClient: client, retryScheduleMs: NO_DELAY });
+      await dispatcher.dispatchBatch(batch(3), "sub_1");
+      const sent = JSON.parse(calls[0]!.body) as Array<{ id: string }>;
+      expect(sent).toHaveLength(1);
+      expect(sent[0]!.id).toBe("evt_2");
+      // Still the batch media type: a subscriber that opted into batching gets
+      // one parsing path regardless of how many events were ready.
+      expect(calls[0]!.headers["content-type"]).toBe(BATCH_CONTENT_TYPE);
+    });
+
+    // A failed batch is ONE failed delivery, not one per event — otherwise a
+    // single failure of a large batch would pause a subscription instantly.
+    it("counts a failed batch as one failed delivery", async () => {
+      const db = new InMemoryDBAdapter();
+      await db.saveSubscription(subscription({ max_batch_size: 10 }));
+      const { client } = mockClient(["throw"]);
+      const dispatcher = new WebhookDispatcher({
+        db,
+        httpClient: client,
+        retryScheduleMs: NO_DELAY,
+        pauseAfterFailures: 2
+      });
+
+      await dispatcher.dispatchBatch(batch(5), "sub_1");
+      const after = await db.getSubscription("sub_1");
+      expect(after?.failure_count).toBe(1);
+      expect(after?.status).toBe("active");
+    });
+
+    it("delivers nothing for an unknown subscription", async () => {
+      const { dispatcher, calls } = await setup(10);
+      expect(await dispatcher.dispatchBatch(batch(2), "sub_missing")).toEqual([]);
+      expect(calls).toHaveLength(0);
+    });
+
+    it("delivers nothing for an empty batch", async () => {
+      const { dispatcher, calls } = await setup(10);
+      expect(await dispatcher.dispatchBatch([], "sub_1")).toEqual([]);
+      expect(calls).toHaveLength(0);
+    });
+  });
+
   // SPECIFICATION.md §5.2.1 — binary content mode relocates attributes to
   // ce-* headers so a subscriber can route without parsing the body.
   describe("content modes (§5.2.1)", () => {