@btravstack/amqp-worker

Reference. A complete, structured description of the AMQP starter's public surface: every export of @btravstack/amqp-worker, its options and defaults, what a delivery becomes, and how its drain meets the kernel's deadline. For the task, see Consume AMQP messages; for the reasoning, Starters and The kernel maps nothing; for the worked example, Order AMQP worker. Generated signatures are under API reference.

Exports

packages/amqp-worker/src/index.ts exports exactly this:

Export	Kind	What it is
AmqpModule	value	AmqpModule(name)({ contract, handlers, url?, connectionOptions?, defaultConsumerOptions?, connectTimeoutMs?, unit?, imports?, provides?, exports?, needs? }) — a di Module(name)({...}) that also takes the handlers provider
AmqpModuleOptions	type	The options object AmqpModule(name) takes
AmqpHandlers	value	AmqpHandlers(contract) — the builder on the starter's own handlers port, typed for contract, so the next call is { inject: { name: Dep }, unit?, sync }, or ([pieces]) to compose one provider per consumer/rpc
HandlersPortOf<C>	type	The handlers port's class typed for C — what a composed orderHandlers's .port is
HandlersInstanceOf<C>	type	That port's instance typed for C (service WorkerInferHandlers<C>)
AmqpHandler	value	AmqpHandler(contract, key) — one consumer or rpc as a provider of its own, typed by key alone; the next call is { inject: { name: Dep }, unit?, sync }, and the piece is what AmqpHandlers(contract)([...]) composes
HandlerPortOf<C, K>	type	One piece's port class, typed for the one key K it implements
AmqpMessage	value	AmqpMessage(contract) — the port the delivery is seeded on, typed by that contract's own messages; a unit module names it in needs and injects it
AmqpMessageOf<C>	type	The union of every message C declares — what AmqpMessage(C) carries
AmqpMessagePortOf<C>	type	AmqpMessage(C)'s port class, typed for C
amqp	value	amqp({ contract, … }) — the starter module itself, needing the handlers port for contract; what AmqpModule imports
AmqpOptions	type	amqp()'s options
AmqpConnectionOptions	type	The connection tuning TypedAmqpWorker.create accepts — heartbeat, reconnect interval, findServers, TLS/socket options; reached by index, since the library does not export it by name
AmqpRuntime	value	class AmqpRuntime extends RuntimePort<Runtime<never, AmqpInfo>> {} — the runtime's port
AmqpConfig	value	class AmqpConfig extends Port("AmqpConfig")<{ url: string; connectTimeoutMs: number }> {} — the broker, bound from AMQP_URL and AMQP_CONNECT_TIMEOUT_MS; a publisher sharing the consumer's broker reads it too
AmqpInfo	type	{ readonly queues: readonly string[] } — published on Serving.info once consuming

HandlersPortOf<C> / HandlersInstanceOf<C> / HandlerPortOf<C, K> are exported as types only, and only because declaration emit forces it: an application that composes orderHandlers = AmqpHandlers(contract)([piece, piece]) and exports it by name (or a slice that exports one piece by name) needs to be able to print that type, and a type built from an unexported alias fails TS4023 ("has or is using name 'ID' … but cannot be named") the moment it tries. AnyAmqpContract — Parameters<typeof TypedAmqpWorker.create>[0]["contract"], the bound on contract — lives in src/amqp-runtime.ts and is not exported from the entry point; it is extracted from the worker's own signature so @amqp-contract/contract stays out of the peer range. The values behind the two handlers ports stay unexported on the same terms as AnyAmqpContract: AmqpHandlersPort — Port("AmqpHandlers"), the starter's own handlers port, declared once — and HANDLER_PREFIX (handler.ts), the string prefix a piece's port id carries. Nothing outside this package legitimately constructs a provider against either bare port — a consumer always goes through AmqpHandlers(contract) or AmqpHandler(contract, key), both of which cast it to the typed alias — so there is nothing a value export would help with. HandlerKeyOf<C> (handler.ts) is unexported for the same reason: nothing outside that file needs to name a bare key. The port is reached as provider.port when a caller needs it.

AmqpModule(name)({...})

Everything Module(name)({...}) takes, plus the contract, the handlers provider and the starter's own options. It appends amqp({ contract, … }) to imports, prepends handlers to provides, prepends AmqpRuntime to exports, and hands the augmented tuples to di's own Module(name).

Option	Required	Default	What it is
contract	yes	—	an amqp-contract contract; the queues consumed are read off its consumers and rpcs
handlers	yes	—	the handlers provider — a Provider<HandlersInstanceOf<TContract>, E, N>, what AmqpHandlers(contract)({ inject, unit?, sync }) returns for this contract, one entry per consumers / rpcs key; one built for another contract fails at the call
url	no	read from AMQP_URL	pins the broker — a test's container
connectionOptions	no	unset — amqp-connection-manager's own defaults	AmqpConnectionOptions, the connection tuning TypedAmqpWorker.create accepts: heartbeat, reconnect interval, findServers, TLS/socket options
defaultConsumerOptions	no	unset — the broker's own defaults, so no prefetch cap	@amqp-contract/worker's ConsumerOptions, applied to every handler: prefetch (the throughput knob), priority, arguments, consumerTag, exclusive
connectTimeoutMs	no	read from AMQP_CONNECT_TIMEOUT_MS (default 5000)	pins how long create waits for the connection; a top-level CreateWorkerOptions field, not one under connectionOptions, where setting it is silently inert. The library's own default is 30 s — longer than most orchestrators wait before restarting the pod
unit	no	unset — dispatch runs unchanged	{ message?: Unit }, the unit module forked around every delivery and seeded with the validated message on AmqpMessage(contract) — built after validation, before the handler runs, torn down when the unit closes; a bound module's own unmet needs join this root's (less the seeded port), refused the same way an unmet handlers port is. Gated against what the pieces declared — see The unit
imports	no	[]	the application's modules
provides	no	[]	the application's own providers
exports	no	[]	the application's own exports; AmqpRuntime is added

The worked composition root, from examples/order-amqp-worker/src/module.ts:

export const OrderAmqpWorker = AmqpModule("OrderAmqpWorker")({
  needs: [Env],
  contract: orderContract,
  handlers: orderHandlers,
  imports: [
    OrderPersistenceModule,
    NotificationsSlice,
    AuditSlice,
    observability(),
    otel(),
  ],
  provides: [relayConfig, outboxRelay],
  // Forked per delivery, after the message is validated: where the envelope's
  // `tenantId` becomes the fork's `Tenant`.
  unit: { message: MessageUnitModule },
  // Everything the fork and the relay read out of the application scope.
  exports: [Outbox, OrderDatabase, Logger, Tracer],
});

NotificationsSlice and AuditSlice are each a slice module exporting one piece of orderHandlers — see the composing form below and Split a worker into slices — imported here because orderHandlers's own deps are the pieces' ports, and di discovers a provider only through a module's imports / provides, never through another provider's deps.

observability() is a second starter, not this package's business: it brings the Logger the handlers and the relay write to, bound from LOG_LEVEL, JSON per line on stdout, every line carrying the delivery's own unit.

RED metrics, reported always and collected when you ask

The runtime REPORTS rate, errors and duration at the unit seam — the one place a framework that owns the unit lifecycle gets them for free — and an observer is what turns a report into a measurement. Reporting always happens; collection happens when otel() is composed, and not before:

Instrument	Kind	Dimensions
btravstack.amqp.operations	counter	handler, outcome
btravstack.amqp.duration	histogram (ms)	the same two

instrumented is gone. Every unit is handed to Observers, and this module contributes a no-op member of its own — so a graph composing no observability owes nothing, and an operation costs one inert call per module that reads the port. Composing observability() writes the failures as lines; composing otel() beside it opens the spans and mints btravstack.<component>.operations and .duration.

The dimensions are chosen for cardinality, and what is absent matters more than what is present. handler is the consumers/rpcs key, so the contract bounds it; the payload is nowhere near the attributes. outcome counts a defect as an error, not as a silence — a count that skipped defects would report a healthy rate while every delivery went to the dead-letter queue.

AmqpHandlers(contract)

The first call fixes the contract type (the value is otherwise unused) and returns a builder on the starter's handlers port, typed for C — so the second call is { inject, unit?, sync }, whose sync hands back the whole handlers record, checked against the contract before any module sees it (a record missing a consumer, or with a typo'd key, is refused here), and the provider carries the port as provider.port. It is { inject, unit?, sync } rather than di's whole arm set for parity: api.OrpcRouter(contract) and all three packages' piece factories spell it that way, so one arm across the family is one surface to learn and one to keep. unit declares the ports every entry of the record reads off context.unit, resolved out of the per-delivery fork exactly as a piece's are and gated by AmqpModule the same way — so a worker that has not outgrown one function needs no slicing to reach it. There is no name to give: a consumer serves one handlers record as it boots one runtime, so the port is the starter's — one Port("AmqpHandlers"), generic at the value level and fixed per contract at the type level (HandlersPortOf<C>, the move the kernel's RuntimePort makes) — and two handlers providers in one graph are di's duplicate-provider defect at build. Each handler is a bare function of the message its consumer declares; WorkerInferHandlers<C> accepts it with nothing wrapped around it, and no context is injected. A record covers every consumer and rpc the contract declares — orderContract has two, orderNotifications and orderAudit, both reading the one orderChanged event on their own queue:

export const orderHandlers = AmqpHandlers(orderContract)({
  inject: { logger: Logger },
  sync: ({ logger }) => ({
    orderNotifications: ({ input: message }) => {
      const { tenantId, id, payload } = message.payload;
      logger.info(
        payload === null
          ? "order gone — notifying"
          : "order placed — notifying",
        {
          tenantId,
          orderId: id,
          ...(payload === null ? {} : { quantity: payload.quantity }),
        },
      );
      return OkAsync();
    },
    orderAudit: ({ input: message }) => {
      const { tenantId, id, occurredAt, payload } = message.payload;
      logger.info("recording an order change", {
        tenantId,
        orderId: id,
        occurredAt,
        change: payload === null ? "removed" : "placed",
      });
      return OkAsync();
    },
  }),
});

examples/order-amqp-worker no longer calls AmqpHandlers this way — its two consumers are each a slice's own piece instead (below) — but the monolithic form is unchanged, and still what Consume AMQP messages teaches for a worker that has not outgrown one function.

A third call composes several pieces instead of one record: AmqpHandlers(contract)([piece, piece, ...]), where each piece is what AmqpHandler(contract, key)({ inject: { name: Dep }, unit?, sync }) returns. Di constructs every piece first — they are the composed provider's own deps, declared under the very key each piece's port id carries, so the services record IS the handlers record. Every key the contract declares must be covered: an array missing one is refused at the call, against an "UNCOVERED HANDLERS — the contract declares a consumer this array does not cover" marker. The diagnostic is a three-line TS2769 and the sentence is at the tail of the third line, past three hundred characters of the caller's own contract type — measured, and not shortenable from inside this package. The missing key is named beside it, whatever the array's length: the refusal is a tuple as long as the array you wrote, so TypeScript lines the two up element by element and reports one error on the trailing element — measured on a one-element array, is not assignable to type 'readonly ["UNCOVERED HANDLERS — …", "right"]'. A piece built for another contract is refused too, structurally, since its port's service is that contract's handler for the key. Uncovered checks coverage, not injectivity, so two pieces claiming the same key still type-check together; di's duplicate-provider defect at build catches it only once both end up discharged as providers in the same graph — wire in just one and the other is silently unregistered, with no diagnostic. The composed provider's own deps are the pieces' ports, not what a piece closes over, so the pieces themselves still need discharging like any other need — typically provides: [...] on the module, or a slice module that exports its own piece.

AmqpHandler(contract, key)

One consumer or rpc, as a provider of its own: the port id carries the contract key (`AmqpHandler:${key}`, HANDLER_PREFIX stripped by the composing form to recover it), so two slices claiming one consumer is di's duplicate-provider defect rather than a silent merge. contract types key and the handler; a key the contract does not declare is refused at the call — there is nothing to type it by — and a handler whose message has drifted is a compile error here rather than at the root. There is no name to give and nothing minted by hand: the provider carries its port as provider.port (HandlerPortOf<C, K>).

The options are { inject, unit?, sync } — this package's own record, not di's whole arm set. unit names the ports the handler reads off context.unit, and sync's return is typed by that record while the port it lands on keeps the context-free handler shape. value could have carried the same record — the declared record is what types it either way — and was dropped for consistency with @btravstack/http-server's OrpcController, so one arm reads the same on all three transports. A piece with no services is { inject: {}, sync: () => handler }.

const orderNotifications = AmqpHandler(
  orderContract,
  "orderNotifications",
)({
  inject: { logger: Logger },
  sync:
    ({ logger }) =>
    ({ input: message }) => {
      logger.info("order changed", { orderId: message.payload.id });
      return OkAsync(undefined);
    },
});

const orderAudit = AmqpHandler(
  orderContract,
  "orderAudit",
)({
  inject: { logger: Logger },
  sync:
    ({ logger }) =>
    ({ input: message }) => {
      logger.info("order audited", { orderId: message.payload.id });
      return OkAsync(undefined);
    },
});

const orderHandlers = AmqpHandlers(orderContract)([
  orderNotifications,
  orderAudit,
]);

amqp(options)

const amqp: <TContract extends AnyAmqpContract, Unit extends AnyUnitModule | undefined = undefined>(
  options: AmqpOptions<TContract, Unit>,
) => Module<
  AmqpRuntime | AmqpConfig,
  ConfigInvalid,
  Env | HandlersInstanceOf<TContract> | UnitNeedsOf<Unit>
>;

The primitive AmqpModule delegates to. AmqpOptions<TContract, Unit> has the sugar's fields minus handlers / imports / provides / exports: the handlers are not an option but the module's need. It provides and exports AmqpRuntime and AmqpConfig, and needs Env (the kernel discharges it), the handlers port typed for contract (HandlersInstanceOf<TContract>) — the runtime provider depends on it through di, so a root that imports the starter without providing the handlers, or provides one built for another contract, is refused at start, whose module parameter accepts no need but Env and Scope — and a bound unit.message module's own unmet needs (UnitNeedsOf<Unit>).

The declared type is the same with url pinned or not, so a pinned composition still carries ConfigInvalid in its error channel.

AmqpConfig, and the environment

Bound through Config.provider; each option pins its field — explicit > environment > default, per field.

Variable	Default	Parsed by	Notes
AMQP_URL	amqp://127.0.0.1:5672	Config.string	the broker
AMQP_CONNECT_TIMEOUT_MS	5000	Config.integer	how long create waits before an unreachable broker is a RuntimeStartFailed

A blank value is a ConfigInvalid — startFailed and exit 78 under runMain.

AmqpRuntime and AmqpInfo

Declared over the kernel's RuntimePort with service Runtime<never, AmqpInfo> — it resolves nothing. Its start calls TypedAmqpWorker.create({ contract, handlers, middleware, urls: [url], … }). create reports an unreachable broker as a modeled Err(ConnectionError); the starter names that tag and maps it to Err(RuntimeStartFailed({ runtime: "amqp", cause })), which is what keeps an unreachable broker at exit 1 rather than 70. Everything else create can fail with — a topology the broker refuses, a bad option, a bug in a provider — stays a defect and exits 70, which is the distinction the blanket recoverDefect that used to sit here could not make.

AmqpInfo.queues is derived — every queue named by the contract's consumers and rpcs, sorted and de-duplicated — never configured, so Serving.info cannot disagree with what the worker consumes.

The unit

One unit per delivery, kind: "delivery", opened by the starter's own WorkerMiddleware. With no unit bound it calls next() unchanged, and context.unit is {} on every piece. With one bound, the middleware forks it — after the message is validated, before the handler runs — and tears the fork down when the unit closes. The ambient currentUnit() record is the only route to the unit's AbortSignal from inside a handler: currentUnit()?.signal, aborted at the kernel's drainTimeoutMs. This transport has no cancellation of its own to defer to — an un-acked delivery is redelivered, which is recovery, not cancellation — so answering a RetryableError on an aborted signal is what hands the message to the next worker.

UnitMeta field	Value
id	randomUUID(), minted per delivery
traceId	the trace id of a W3C traceparent header, else the publisher's messageId, else correlationId (an RPC-shaped message), else the minted id — non-blank only

A delivery tag is not a unit id: tags are per-channel and restart at 1 after a reconnect, which amqp-connection-manager performs silently underneath the worker — the one identifier that looks unique per delivery is not, across exactly the event this library exists to handle. consumerTag + deliveryTag almost fixes it, until ConsumerOptions lets a caller pin consumerTag. Minting is the only form of the rule that survives. A blank messageId is ignored rather than adopted, since "" is not nullish and would otherwise give every delivery the same trace id. A traceparent header outranks messageId because it is the one value minted to span processes.

AmqpMessage(contract) — the one seeded port

The fork is seeded with the validated message, on AmqpMessage(contract). That is the only entry the worker seeds, and it is what lets a unit module derive a tenant — or anything else it scopes by — from the delivery rather than from an ambient record:

const Message = AmqpMessage(orderContract);

export const MessageUnit = Module("MessageUnit")({
  needs: [Message],
  provides: [
    Provider(Tenant)({
      inject: { message: Message },
      sync: ({ message }) => message.payload.tenantId,
    }),
  ],
  exports: [Tenant],
});

One Port("AmqpMessage") call, cast per contract at the type level, so no contract instantiating it warns about a duplicate id while a module built for one contract still cannot read another's message. A module naming that port in needs owes the composition root nothing for it — the seed discharges it, and it is subtracted from what unit contributes to the starter's Needs; everything else the module needs still surfaces at start.

context.unit, and the gate on the module bound

A piece declares the unit-scoped ports its handler may read as unit: beside inject, and the handler reads them off context.unit.name. The whole-record arm takes the same unit:, applied to every entry of the record it hands back, so a worker that has not outgrown one function reaches context.unit without slicing first. Entries are lazy getters over the forked context — neither writable nor configurable, so a handler reads what the fork holds and cannot reshape the record under the next delivery.

That declaration is a promise the root has to keep, and nothing else checks it: the piece and the root are typed independently. So AmqpModule gatesunit.message against what was declared — the union of every piece's record, collected by AmqpHandlers(contract)([...]) where the pieces are known, or the record arm's own unit: where the worker is one function — and a bound module that does not export a declared port is refused against a "UNIT DOES NOT PROVIDE — a piece injects a port the bound unit module does not export" marker, carrying the offending port. The gate rides the whole options record, not the unit property, because a gate on a property is not read when the property is absent — and a root that declares a piece's unit: and then binds no module at all is exactly the case worth catching.

amqp() is not gated, structurally: it takes its handlers as a need, never as a value, so there is nothing to read the declarations off. That is the one path where a declared port goes unchecked, and it fails two ways. With no module bound the record is empty, so context.unit.tenant is undefined and the next property access throws a TypeError; with a module bound that does not export Tenant, the getter runs and di throws [di] no service registered for port …, naming the port. Either way the handler is invoked from inside one of the library's own combinators, so even a synchronous throw is qualified as a Defect: the delivery is nacked once and goes straight to the dead-letter queue with that error in the report — loud, on the first message, never a silent wrong answer.

The drain, and the one deadline

Serving.drain(signal) calls worker.close({ drainTimeoutMs: null }) — cancel every consumer, let in-flight handlers finish so their acks land on a still-open channel, then close — raced against signal, the kernel's deadline. stop() reuses whatever deadline drain armed, so a signal-driven shutdown never waits twice; called alone, it waits on the worker's own close.

drainTimeoutMs: null is deliberate: the library's own default drain timeout is 30 s, above the kernel's 20 s default, and would quietly win. Passing null removes the second clock instead of requiring it be kept under the first — one deadline in the process, the kernel's.

When the deadline wins, close() keeps running underneath: the connection stays open and the in-flight handler keeps heading toward its own ack or nack on the library's clock. The kernel reports the unit abandoned, which is honest about what it waited for. Redelivery happens only once the connection actually drops — when the process dies, not when the kernel's deadline passes.

Peer dependencies

@btravstack/core, @btravstack/config, @btravstack/di, unthrown, @amqp-contract/worker, @opentelemetry/api. @opentelemetry/api is a peer because @amqp-contract/worker itself peers on it; @amqp-contract/contract is not in the list — it is a devDependency of the package, used only to type its own tests. Node >=22.

Deliberately not included

• Result → ack / retry / dead-letter. On this transport that is a three-way split, and the package owns none of it. amqp-contract's own dispatch routes a modeled RetryableError / NonRetryableError against the queue's retry config. A Defect is not routed that way: it is nacked once, immediately, under its original routing key, never touching the retry budget — so a handler that wants "infrastructure comes back" must recover its own Defects into a RetryableError explicitly, or an infrastructure failure is parked on the first attempt exactly like a permanent domain error. Note also that retry: { mode: "ttl-backoff", maxRetries: 3 } means four total attempts, not the three Temporal's maximumAttempts: 3 names.
• A publisher. The starter runs a consumer; publishing is @amqp-contract/client's job (the worked example's outbox relay creates its own client from the same AmqpConfig).
• A context channel. A handler reads nothing out of a context; what it needs, its provider declares.

Testing

The package's own suite needs a Docker daemon: it runs against a real RabbitMQ, because the retry and dead-letter routing it relies on is the broker's behaviour, not something an in-memory fake could stand in for. The broker is the one container @btravstack/internal-test-infra/rabbitmq starts for the whole repository and every workspace reuses, and each test gets a vhost of its own from @amqp-contract/testing's it extension. amqp-runtime.spec.ts covers the published info, the unreachable broker, the environment binding, the unit boundary and its fork, and the drain; handler.spec.ts composes a broadcast with two consumers of one publisher from two pieces, pinning that both run and that each was built from the ports its own provider declared rather than a record closing over both, and drives the seeded fork through a piece and through the record arm. handler.test-d.ts pins the composing form's compile-time gates: a piece typed by its own key, an array covering every declared key, a missing key refused and named, and a piece built for another contract refused structurally.