@btravstack/amqp-worker

The consuming half of AMQP for @btravstack/core — publishing belongs to @amqp-contract/client, which the outbox publisher in examples/order-amqp-worker uses directly. An amqp-contract worker under the kernel's lifecycle — one unit per delivery, and a drain with exactly one deadline, the kernel's own.

📖 Documentation · Reference · API Reference

pnpm add @btravstack/amqp-worker @btravstack/core @btravstack/config @btravstack/di unthrown \
  @amqp-contract/worker@^3.0.0-beta.12 @opentelemetry/api

All six are peer dependencies — install them (@opentelemetry/api because @amqp-contract/worker itself peers on it). Node >=22.22.

A worked example

import { Env } from "@btravstack/config";
import { AmqpHandlers, AmqpModule } from "@btravstack/amqp-worker";
import { runMain, Logger } from "@btravstack/core";
import { ErrAsync, OkAsync, P } from "unthrown";

// The handlers, as a service: one entry per consumer the contract declares,
// built by di from the use cases it lists — no injected context — on the
// starter's own handlers port, typed by the contract (a consumer serves one
// handlers record, so there is nothing to name).
const orderHandlers = AmqpHandlers(orderContract)({
  inject: { placeOrder: PlaceOrder },
  sync: ({ placeOrder }) => ({
    orderNotifications: ({ input: message }) =>
      placeOrder
        .execute(
          TenantId(message.payload.tenantId),
          message.payload.id,
          message.payload.payload?.quantity ?? 0,
        )
        .map(() => undefined)
        // A modeled domain error is permanent: dead-letter it, no retry.
        .mapErrCases((matcher) =>
          matcher.with(
            P.tag("InvalidQuantity"),
            P.tag("InvalidOrderId"),
            P.tag("DuplicateOrder"),
            (error) => new NonRetryableError(error._tag, error),
          ),
        )
        // Infrastructure failing is not: recover the defect into a retry.
        .recoverDefect((cause) =>
          ErrAsync(new RetryableError("placing the order failed", cause)),
        ),
    // Every consumer the contract declares must be covered — an uncovered
    // key is refused at this call, not on the first delivery.
    orderAudit: () => OkAsync(undefined),
  }),
});

const Worker = AmqpModule("Worker")({
  needs: [Env],
  contract: orderContract,
  handlers: orderHandlers,
  imports: [AppModule],
});

await runMain(Worker);

AmqpModule imports the starter — AmqpRuntime, AmqpConfig bound from AMQP_URL (default amqp://127.0.0.1:5672) — provides the handlers and exports the runtime port; the handlers are checked against the contract at the call, not on the first delivery. runtimeInfo() reads { queues } back once consuming.

A worker with several consumers can be several slices instead of one record: AmqpHandler(contract, key)({ inject: { name: Dep }, unit, sync }) mints a provider for ONE consumer or rpc, typed by the key alone, and AmqpHandlers(contract)([...]) composes an array of them into the same handlers provider AmqpModule takes — the array must cover every key the contract declares, and each piece's own port must still be discharged (provides), since the composed provider's deps are the pieces' ports, not what they close over. A consumer belongs to exactly one slice, and the compiler holds only half of that: coverage is checked, injectivity is not, so two slices claiming one key type-check together. It is di's duplicate-provider defect at build once both pieces are discharged as providers in one graph — wire in only one and the other's handler is silently never registered.

A piece's unit names the ports its handler reads off context.unit, resolved out of the per-delivery fork of the module bound on unit.message — which is seeded with the validated message on AmqpMessage(contract), so a unit module derives a tenant from the delivery rather than from an ambient record. The record arm above takes the same unit: — one record for every entry in it — so a worker that has not outgrown one function reaches context.unit without slicing first. The root is checked against what was declared either way: bind a module that does not export one of those ports and AmqpModule refuses it, naming the port.

Its own test suite needs a Docker daemon: the specs run against the shared rabbitmq container internal/test-infra starts once per machine and every workspace reuses. Nothing else here does.

Options

AmqpModule(name)({...}) takes the starter's options plus handlers and the module lists (imports, provides, exports, needs). It is the only way to compose the starter, which is not exported:

The full table — required/optional, defaults, and the reasoning — lives on the reference page, which is this list's one detailed home.

What it decides, and what it does not

Result → ack / retry / dead-letter is a three-way split and none of it is this package's: a RetryableError / NonRetryableError is routed by amqp-contract's own dispatch against the queue's retry config, and a Defect is nacked once, straight to the dead-letter queue, never touching that budget — which is why the handler above recovers its own defects into a RetryableError. The drain calls worker.close({ drainTimeoutMs: null }) raced against the kernel's deadline: the library is told to wait forever, so there is no second timeout in the process to keep in step. Each delivery is one unit, traceId the publisher's messageId. The rest is on the documentation site.

AmqpModule automatically exports publicly available Logger and HealthChecks ports, including whole-module re-exports and providers declared at the root. It keeps private infrastructure hidden and business exports explicit.

License

MIT © Benoit TRAVERS

Broadcast subscriptions

amqpSubscription(name, { contract, settings, inject?, sync }) consumes best-effort facts beside an HTTP runtime, with one exclusive ephemeral queue per replica. It reports failures and broker health and closes with the scope. InvalidAmqpSubscription names a refused configuration; SubscriptionHandlers<C, E> types the contract's handler record. See the subscription reference for the typed contract, settings port and delivery guarantees.