Add middleware

Cross-cutting concerns — authentication, tracing, timing, dependency injection — belong in one place rather than at the top of every handler. There are two mechanisms:

• Worker middleware wraps handler invocation and can inject typed context.
• Client interceptors wrap publish() and call() and can patch the outgoing message.

Both take (args, next) and call next() to continue. Everything stays inside AsyncResult; nothing throws.

Inject typed context into handlers

A middleware proves something once and passes the result downstream. Handlers receive it as helpers.context, in their first argument.

import {
  composeMiddleware,
  declareMiddleware,
  NonRetryableError,
  TypedAmqpWorker,
  type EmptyContext,
} from "@amqp-contract/worker";
import { ErrAsync } from "unthrown";

const auth = declareMiddleware<EmptyContext, { tenantId: string }>((args, next) => {
  const tenantId = args.rawMessage.properties.headers?.["x-tenant-id"];
  if (typeof tenantId !== "string") {
    // Short-circuit: routes like any handler error. The handler never runs.
    return ErrAsync(new NonRetryableError("Missing x-tenant-id header"));
  }
  return next({ context: { tenantId } });
});

const worker = await TypedAmqpWorker.create({
  contract,
  middleware: auth,
  handlers: {
    // context is typed { tenantId: string } — proven by the middleware
    processOrder: ({ context, input: { payload } }) => processFor(context.tenantId, payload),
  },
  urls: ["amqp://localhost"],
}).getOrThrow();

The two type parameters on declareMiddleware are the context going in and the context coming out. Without any middleware or createContext, handlers get EmptyContext.

Chain several middleware

const timing = declareMiddleware<{ tenantId: string }, { tenantId: string }>((args, next) => {
  const start = Date.now();
  return next().tap(() => {
    console.log(`${args.handlerName} (${args.context.tenantId}): ${Date.now() - start}ms`);
  });
});

middleware: composeMiddleware(auth, timing),

composeMiddleware(outermost, …, innermost) runs left to right. Context types accumulate along the chain, so timing sees the tenantId that auth added, and handlers see the final accumulated type.

The middleware option also accepts an array directly — middleware: [auth, timing] — with the first entry outermost, mirroring the client's interceptor arrays. At runtime it composes exactly like composeMiddleware(...), but the array form does not thread the stepwise context types across entries: when middleware accumulate typed context for the handlers, pre-compose with composeMiddleware so the chain's final context type is inferred into helpers.context.

Because next() returns the handler's AsyncResult, a middleware can post-process it with .tap, .mapErr or .flatMapErr.

Inject dependencies per message

createContext seeds the chain. It runs once per message after validation, so it can build request-scoped values; close over singletons for anything per-worker.

const worker = await TypedAmqpWorker.create({
  contract,
  createContext: (info) => ({
    log: baseLogger.child({
      handler: info.handlerName,
      correlationId: info.rawMessage.properties.correlationId,
    }),
    orderRepo,
  }),
  middleware: auth, // seeded with { log, orderRepo }
  handlers: {
    // context: { log, orderRepo } & { tenantId: string }
    processOrder: ({ context, input: { payload } }) => context.orderRepo.process(payload),
  },
  urls: ["amqp://localhost"],
}).getOrThrow();

If createContext throws or rejects, the message is dead-lettered as a NonRetryableError and the handler never runs.

For a dependency-injection graph, demesne's Layer.forkScope is the recommended implementation: build the graph once at startup, fork a scope per message.

Substitute the payload

next({ payload }) replaces the payload for everything downstream.

const normalize = declareMiddleware<EmptyContext, EmptyContext>((args, next) =>
  next({ payload: { ...args.message.payload, email: args.message.payload.email.toLowerCase() } }),
);

The substituted payload is re-validated against the consumer's schema before the handler runs. An invalid substitution is a NonRetryableError, so middleware cannot smuggle unvalidated data past the contract boundary.

Short-circuit without running the handler

Return a result instead of calling next(). It routes exactly like a handler result:

Returned	Effect
Err(new RetryableError(…))	Queue retry mode applies
Err(new NonRetryableError(…))	Dead-lettered
Err(rpcError(code, data))	Typed error reply to the caller (RPC with a declared errors map)
Ok(value)	Handler skipped. On an RPC, value is validated against the response schema and published as the reply

That last row is how you build a cache: check the cache in middleware, return Ok(hit) to reply without invoking the handler.

Use typed error constructors in RPC handlers

RPC handlers with a declared errors map get typed constructors in helpers.errors, with per-code data inference:

handlers: {
  getOrder: ({ errors, input: { payload } }) =>
    orders.has(payload.orderId)
      ? OkAsync(orders.get(payload.orderId))
      : ErrAsync(errors.ORDER_NOT_FOUND({ orderId: payload.orderId })),
},

Equivalent to rpcError("ORDER_NOT_FOUND", { orderId }), but autocompleted and checked against the contract.

Know what middleware can and cannot see

• Middleware runs after payload and header validation, so args.message is already schema-checked. Parse and validation failures dead-letter the message before any middleware runs.
• The chain wraps consumers and RPC handlers. args.isRpc discriminates; args.handlerName carries the contract key.
• args.rawMessage is the underlying AMQP message, which is where raw headers live.

Stamp headers on every publish

Client interceptors run outside validation, so a patched message is validated exactly like an unpatched one.

import { TypedAmqpClient, type PublishInterceptor } from "@amqp-contract/client";

const stampTenant: PublishInterceptor = (args, next) =>
  next({
    options: {
      ...args.options,
      headers: { ...args.options.headers, "x-tenant-id": currentTenantId() },
    },
  });

const client = await TypedAmqpClient.create({
  contract,
  urls: ["amqp://localhost"],
  publishInterceptors: [stampTenant],
}).getOrThrow();

The first interceptor in the array is the outermost.

Retry a timed-out RPC call

Call interceptors wrap the whole round trip — request validation, publish, and awaiting the reply — so calling next() again reissues the call.

import { RpcTimeoutError, type CallInterceptor } from "@amqp-contract/client";
import { ErrAsync, P } from "unthrown";

const retryTimeoutsOnce: CallInterceptor = (args, next) =>
  next().flatMapErrCases((matcher) =>
    matcher.with(
      P.tag("@amqp-contract/MessageValidationError"),
      P.tag("@amqp-contract/PublishError"),
      P.tag("@amqp-contract/RpcTimeoutError"),
      P.tag("@amqp-contract/RpcCancelledError"),
      P.tag("@amqp-contract/RpcError"),
      (error) => (error instanceof RpcTimeoutError ? next() : ErrAsync(error)),
    ),
  );

They can also adjust timeoutMs or patch the request before it goes out.

A request the broker side refused (timeout, nack, closed channel) is a modeled PublishError, so it reaches the matcher like any other case; this interceptor passes it on. Only a failure core cannot classify is a defect, which flows through flatMapErrCases untouched — use .recoverDefect(…) if you need to act on one.

Trace context needs neither

Trace context crosses the broker on its own: publish and call inject the producer span's context into the message headers (through the propagator your OpenTelemetry SDK registered), and the worker runs createContext, middleware and the handler with the consume span active, continuing the publisher's trace. No interceptor or middleware is needed, and a traceparent stamped by hand is overwritten by the injected one whenever an SDK is registered.

Telemetry spans sit outside the interceptor chain, so interceptor work is already covered by the built-in instrumentation. See instrument with OpenTelemetry.

Where next

• Consume messages — handler signatures.
• Error model — how short-circuit errors route.
• Add logging — the built-in logger, which may be all you need.