Incoming and outgoing pipeline behaviours (middleware) #263

Today the only extension points are a read-side middleware chain over TransportMessage (withMessageReadMiddleware) and eight synchronous emitters on BusInstance (beforeSend, beforePublish, afterSend, afterPublish, afterReceive, beforeDispatch, afterDispatch, onError). Nothing wraps an outgoing send or publish, or a single handler. Transports build their native headers only from MessageAttributes: bus-rabbitmq writes attributes/stickyAttributes as JSON headers, and bus-sqs writes attributes.<key>/stickyAttributes.<key>/correlationId SNS attributes. So #202 (raw RabbitMQ headers such as x-delay) can't be done.

Problems found on master:

• A read middleware that doesn't call next() leaves the message neither deleted nor returned. It stays in flight in memory, unacked on RabbitMQ, and SQS redelivers it after the visibility timeout.
• beforeDispatch is emitted after handlers.map(dispatchMessageToHandler), so function handlers have already started when it fires.
• MiddlewareDispatcher and TypedEmitter are exported from the package root, and BusInstance.dispatchMessageToHandler is public by accident.

#258, #264 and #151 build on this, and #260/#261 hook into it.

Goal

One small, typed, async (context, next) middleware model with three stages: incoming per message, per handler, and outgoing. It replaces read middleware and the emitters, and lets outgoing middleware set transport-native headers.

Decisions

Answered by the maintainer:

1. Name: "middleware" (matches the existing Middleware type, /guide/middleware and the Koa idiom).
2. Registration: a single withMiddleware(...middleware: BusMiddleware[]), where BusMiddleware = { incoming?, handler?, outgoing? }.
3. Emitters: remove all eight and withMessageReadMiddleware, with no shims.
4. Per-handler stage: ships in this issue.
5. Native headers: a generic headers bag on the outgoing context, set from middleware only; per-call headers wait for Delayed / deferred message delivery #258's send options. Incoming native headers are read from transportMessage.raw.

Decided from the principles and the dependent issues:

6. Outgoing timing: the outgoing chain runs when send()/publish() is called, after correlation and sticky attributes are merged and before the outbox. Its terminal step buffers inside a handler and calls the transport otherwise. The outbox flush calls the transport with the finished message and doesn't re-run middleware. This lets a durable outbox (Durable transactional outbox #260) store the final message, lets OTel inject context while the sender's span is active, and lets validation fail the handler that sent.
7. Order: registration order within each stage, with the first registered outermost. There are no priorities or named slots.
8. Short-circuit: not calling next() means handled. Incoming: handlers are skipped and the message is deleted (fixes the stuck message). Handler: that handler is skipped and counts as succeeded. Outgoing: the message is dropped (debug log). next() can be called at most once; a second call throws MiddlewareNextCalledTwice (names the stage).
9. Errors: a throw from an incoming middleware, or an error it lets through from next(), returns the message for retry as a handler failure does today. Catching without rethrowing marks the message handled. A throw from a handler middleware fails that handler only (its outbox is discarded, siblings unaffected, the message is retried via HandlerDispatchRejected). A throw from an outgoing middleware rejects that send()/publish(), and nothing is buffered.
10. Incoming is read-only: message and attributes stay frozen. Middleware passes derived data through its own AsyncLocalStorage or sticky attributes.
11. Per-bus: the chains live in BusConfiguration and are passed to BusInstance. Nothing is module-level. A stateful plugin is a factory called once per bus.

Non-goals

• Core messageId, sentAt and Transport.endpointName: these are in the messageId issue, which runs in parallel with this one.
• Normalised incoming native headers (use transportMessage.raw).
• Per-call headers on send()/publish() (left to Delayed / deferred message delivery #258's send options).
• Middleware filtered by message type, and middleware that replaces the message object.
• Attempt count or failure metadata on the incoming context (Configurable recoverability policy and error queue with failure metadata #262).
• Changing failMessage semantics: the outbox still flushes and the delete still runs after it; on RabbitMQ, failMessage followed by a throw both dead-letters and retries (Configurable recoverability policy and error queue with failure metadata #262).
• Test fakes for middleware contexts (Handler and workflow test helpers #265).
• A bus-test conformance case for native headers (each transport tests its own).

Approach

How the dependent issues hook in:

• OpenTelemetry tracing and metrics #264: openTelemetry() returns a BusMiddleware. Incoming opens the process span, extracting traceparent from attributes. Handler adds optional per-handler child spans. Outgoing creates the span and injects traceparent/tracestate into attributes. Message ids and send times come from the messageId issue.
• Add Request/Response pattern #151: outgoing stamps the request id; incoming resolves replies without calling next(). Correlating on message ids relies on the messageId issue.
• Delayed / deferred message delivery #258: delivery options join TransportSendOptions and the outgoing context. The persistence scheduler fallback is a core terminal step.
• Durable transactional outbox #260: stores the finished outgoing message; the per-handler stage is where the transaction/session scope lives.
• Inbox / idempotent consumer (message deduplication) #261: duplicate skip is an incoming short-circuit; the stable message id it dedups on comes from the messageId issue and is visible to outgoing middleware.
• Configurable recoverability policy and error queue with failure metadata #262: adds attempt and failure metadata to the incoming context.

Acceptance criteria

• withMiddleware() on BusConfiguration (JSDoc, BusAlreadyInitialized guard, returns this) takes BusMiddleware objects with incoming, handler and outgoing stages, run in registration order (first is outermost).
• Incoming middleware runs for polled and Receiver messages, including messages with no handler. Not calling next() deletes the message. A throw returns it for retry.
• Handler middleware wraps each handler and workflow handler inside its outbox. A throw fails that handler only, and its sends are dropped.
• Outgoing middleware runs at send()/publish(), before the outbox. It can mutate attributes and headers, drop the message by not calling next(), or reject the send by throwing. Flushing doesn't re-run it.
• A second next() call throws MiddlewareNextCalledTwice.
• Transport.send/publish take TransportSendOptions. bus-rabbitmq and bus-sqs write headers natively and throw TransportHeaderReserved on a reserved name. Each has an integration test proving a header arrives (RabbitMQ also through a retry). InMemoryQueue keeps them.
• withMessageReadMiddleware, MiddlewareDispatcher, the eight emitters, their payload types and TypedEmitter are removed. Tests are ported to middleware.
• Middleware is unit-testable as a plain function with a plain-object context and next, with no bus. At least one spec does this.
• Two buses in one process don't see each other's middleware or context (isolation test).
• Docs: /guide/middleware rewritten; /guide/lifecycle-hooks deleted and redirected to it, with links fixed; snippets type-check.
• MIGRATING.md entry, a minor Breaking: changeset and CLAUDE.md updates.