add @node-ts/bus-nestjs, a NestJS module for the bus

.changeset/nestjs-module.md

@@ -0,0 +1,14 @@
+---
+'@node-ts/bus-nestjs': minor
+'@node-ts/bus-cli': minor
+'@node-ts/bus-core': minor
+---
+
+Add `@node-ts/bus-nestjs`, a NestJS module for the bus (#268), for NestJS 11 and 12.
+
+- `BusModule.forRoot({ configure })` and `BusModule.forRootAsync({ imports, inject, useFactory })` register a bus, configured by a factory that's given a `BusConfiguration` (logging through Nest's `Logger`, with no interrupt signals) and returns it with the transport, persistence and message types. The bus is injected as `BusInstance`; named buses with `@InjectBus(name)` and `getBusToken(name)`.
+- Class handlers and workflows are providers of your modules, registered with the bus by `@BusHandler()` and `@BusWorkflow()` or by listing them in `BusModule.forFeature()`, and resolved from Nest's container for each message by `nestContainer(moduleRef)`. Function handlers and `defineWorkflow()` workflows are registered with `BusModule.forFeatureAsync()`, given the providers they close over. Request-scoped providers get a request scope per received message, with the message and its attributes as Nest's `REQUEST` (`BusRequest`).
+- The bus is built when the application initializes, and initialized and started when it bootstraps (or by the application, with `lifecycle: 'manual'`); if that fails, it's disposed. Since `BusModule` is global, Nest stops it after the `onModuleDestroy` of every module that isn't global, and disposes it after their `onApplicationShutdown`, so release what handlers use in `beforeApplicationShutdown` or `onApplicationShutdown`. Startup fails with `BusClassNotProvided`, `BusNotRegistered`, `BusAlreadyRegistered`, `BusFeatureNotStatic`, `WorkflowResolvedWithoutMessage` or `BusCoreVersionNotSupported`, naming the fix.
+- `createBusForProvisioning(AppModule)` returns the application's bus, built but not initialized, for `bus provision`.
+
+`bus provision` in `@node-ts/bus-cli` also accepts a module that exports a built, uninitialized bus, or a function that returns one, and exits once it has printed its report, even if the module left handles open. `BusInstance.canStart` in `@node-ts/bus-core` says whether `start()` can be called: false for a send-only bus or one with a `Receiver`.

CLAUDE.md

@@ -50,16 +50,17 @@ pnpm exec dotenv -e test.env -- jest packages/bus-sqs/src/sqs-transport.spec.ts
 - Persistence (workflow state and delayed messages): `bus-postgres`, `bus-mongodb`.
 - `bus-sqs-lambda` — a `Receiver` that feeds Lambda SQS events into the bus instead of the bus polling the transport.
 - `bus-opentelemetry` — `openTelemetry()` middleware for spans, W3C trace context (carried in message attributes) and messaging metrics, with only `@opentelemetry/api` as a peer. See `packages/bus-opentelemetry/CLAUDE.md`.
-- `bus-cli` — the `bus` command line. `bus provision <module>` builds the `BusConfiguration` a module exports and calls `bus.provision()` (see Provisioning below). `bus generate-message-types` reads message classes with the TypeScript compiler API and writes a file exporting a `MessageTypes` map, which users pass to `Bus.configure().withMessageTypes()`. There's no global registry: the bus merges the maps it's given (`mergeMessageTypes` in bus-messages) and passes them to its serializer on every read, so Dates, Maps, Sets, bigints and nested classes are restored without decorators. See `packages/bus-cli/CLAUDE.md`.
-- `bus-test` — shared conformance suites. New transports call `transportTests(transport, publishSystemMessage, systemMessageTopicIdentifier, readAllFromDeadLetterQueue)` from their `*.integration.ts` (see bus-sqs / bus-rabbitmq); persistence adapters call `workflowStateRoundTripTests(persistence)` and, when they store outgoing messages, `scheduledMessageRoundTripTests(persistence)` with an instance on its own schema or database, and when they implement `beginTransaction()`, `outboxTests(persistence)` and `inboxTests(persistence)` the same way. Its fixtures' message types are generated into `src/helpers/message-types.generated.ts`, exported as `messageTypes`, and its suites pass them to their buses. Every adapter generates the message types of its own test fixtures the same way (bus-sqs, bus-rabbitmq, bus-sqs-lambda and bus-opentelemetry in `src/test`, bus-postgres and bus-mongodb in `test/`), and every test bus that receives messages calls `.withMessageTypes(...)` with them. bus-core can't depend on bus-cli, so its tests use `testMessageTypes` and `messageTypesFor(...)` from `src/test`. After changing a fixture run `pnpm --filter <package> run generate:message-types` and commit the file (CI runs `pnpm check:message-types`).
+- `bus-nestjs` — `BusModule` for NestJS (peers `@nestjs/common` and `@nestjs/core` 11 or 12; 12 is ESM-only, loaded through `require(esm)`): `forRoot`/`forRootAsync` with a factory given a seeded `BusConfiguration`, `forFeature`/`forFeatureAsync` and `@BusHandler()`/`@BusWorkflow()` (via `DiscoveryService.createDecorator()`), `nestContainer(moduleRef)` with a per-message request scope, the bus run from Nest's lifecycle hooks, and `createBusForProvisioning()`. See `packages/bus-nestjs/CLAUDE.md`.
+- `bus-cli` — the `bus` command line. `bus provision <module>` builds the `BusConfiguration` a module exports (or takes a built, uninitialized `BusInstance`) and calls `bus.provision()` (see Provisioning below). `bus generate-message-types` reads message classes with the TypeScript compiler API and writes a file exporting a `MessageTypes` map, which users pass to `Bus.configure().withMessageTypes()`. There's no global registry: the bus merges the maps it's given (`mergeMessageTypes` in bus-messages) and passes them to its serializer on every read, so Dates, Maps, Sets, bigints and nested classes are restored without decorators. See `packages/bus-cli/CLAUDE.md`.
+- `bus-test` — shared conformance suites. New transports call `transportTests(transport, publishSystemMessage, systemMessageTopicIdentifier, readAllFromDeadLetterQueue)` from their `*.integration.ts` (see bus-sqs / bus-rabbitmq); persistence adapters call `workflowStateRoundTripTests(persistence)` and, when they store outgoing messages, `scheduledMessageRoundTripTests(persistence)` with an instance on its own schema or database, and when they implement `beginTransaction()`, `outboxTests(persistence)` and `inboxTests(persistence)` the same way. Its fixtures' message types are generated into `src/helpers/message-types.generated.ts`, exported as `messageTypes`, and its suites pass them to their buses. Every adapter generates the message types of its own test fixtures the same way (bus-sqs, bus-rabbitmq, bus-sqs-lambda, bus-opentelemetry and bus-nestjs in `src/test`, bus-postgres and bus-mongodb in `test/`), and every test bus that receives messages calls `.withMessageTypes(...)` with them. bus-core can't depend on bus-cli, so its tests use `testMessageTypes` and `messageTypesFor(...)` from `src/test`. After changing a fixture run `pnpm --filter <package> run generate:message-types` and commit the file (CI runs `pnpm check:message-types`).
 
 ### Bus lifecycle (bus-core/src/service-bus)
 
 - `Bus.configure()` returns a fluent `BusConfiguration` (`withTransport`, `withHandler`, `withCustomHandler`, `withWorkflow`, `withPersistence`, `withMessageTypes`, `withContainer`, `withConcurrency`, `withRecoverability`, `withMiddleware`, `asSendOnly`, `withReceiver`, ...). `build()` merges the message types, assembles `CoreDependencies` (logger factory, serializer, message serializer, message types, handler registry, container), creates the bus' `MessageHandlingContext`, calls `prepare(coreDependencies)` on transport/persistence/workflow registry, and returns a `BusInstance`. Configuration methods throw `BusAlreadyInitialized` after build.
 - **Buses share nothing**, as if each ran in its own process (design principle 3). The handling context, lifecycle context, message types and default logger factory are per bus; don't add module-level or `globalThis` state that holds per-bus or per-message data. A transport instance is one queue and one connection, so `build()` throws `TransportAlreadyInUse` for a transport another bus uses (a `WeakSet` in `bus-configuration.ts`). Serializers and persistence may be shared: serializers get the bus' message types per call, persistence stores plain JSON while `WorkflowRegistry` does `toPlain`/`toClass` with its own bus' serializer and types, and a shared persistence is only disposed by the last bus using it (a `WeakMap` count in `bus-instance.ts`, released once per bus however often it's disposed). Every bus prepares its persistence, including send-only ones, and `initialize()` initializes it when the bus has workflows or the persistence stores outgoing messages.
 - At `initialize()`, a bus that receives (handlers or workflows, polling or with a `Receiver`) throws `MessageTypesMissing` unless its message types cover every handled `$name` and workflow state. Send-only buses and buses with no handlers aren't checked, and custom-handler messages are exempt.
 - **Provisioning** (`provisioning/`, #333): nothing is created at runtime by default. `bus.provision({ dryRun? })` registers the workflows, calls the optional `Persistence.provision({ workflows, dryRun })` (when the bus uses persistence) then, after `connect()` unless a dry run, `Transport.provision({ handlerRegistry, sendOnly, messageNames, sendsAnyMessage, dryRun })` (`sendsAnyMessage` is true for a scheduler, so its runtime permissions allow publishing anywhere; a send-only bus with no message types logs a warning), and returns their `ProvisioningPlan`s (`adapter`, `resources` of `{ type, name, properties? }`, `runtimePermissions` `{ format, document }`); it never initializes or starts, so callers `dispose()`. `messageNames` (`getMessageNames` in `bus-instance.ts`) is every handled `$name` plus every message-types entry that isn't a workflow state (by registered state name, or a class extending `WorkflowState`, also matched by base class name for another copy of bus-core), so send-only buses provision topics for their message types. `initialize()` creates nothing: adapters get `verifyResources` (from `withResourceVerification()`, default true, false right after auto-provisioning) and throw `ResourcesNotProvisioned(adapterName, missing)` from read-only checks; transports also get `autoProvision` and may only create things at runtime (e.g. a topic on first publish) when it's set. `withAutoProvision()` runs `provision()` at the start of `initialize()`. The transport is connected once (`connectTransport`) and workflows registered once (`registerWorkflows` caches the registration promise, so a failure is rethrown, never skipped), whichever runs first. Transports check only what the bus receives through at startup. `WorkflowRegistry.initialize` only registers handlers and records `getPersistedWorkflows()`; `Persistence.initializeWorkflow` no longer exists. Test buses on real infra use `.withAutoProvision()`; bus-test's suites call `bus.provision()` then `initialize()`, so provision-then-run is covered without runtime provisioning.
-- `BusInstance.initialize()` connects the transport and lets it check its queues/subscriptions; `start()` runs `concurrency` application loops that `readNextMessage()` from the transport; after an empty read a loop sleeps only what's left of 500 ms (`EMPTY_QUEUE_SLEEP_MS`), so a transport that already waited, like the in-memory queue or an SQS long poll, isn't slowed. With a `Receiver` configured, messages are pushed in via `receive()` instead. The policy still applies: a retried message is returned to the transport with the delay and its error rethrown to the host, and a dead-lettered one goes through `transport.fail` and is reported as handled.
+- `BusInstance.initialize()` connects the transport and lets it check its queues/subscriptions; `start()` (allowed when `canStart`: not send-only and no `Receiver`) runs `concurrency` application loops that `readNextMessage()` from the transport; after an empty read a loop sleeps only what's left of 500 ms (`EMPTY_QUEUE_SLEEP_MS`), so a transport that already waited, like the in-memory queue or an SQS long poll, isn't slowed. With a `Receiver` configured, messages are pushed in via `receive()` instead. The policy still applies: a retried message is returned to the transport with the delay and its error rethrown to the host, and a dead-lettered one goes through `transport.fail` and is reported as handled.
 - Message flow: transport message → the bus' `messageHandlingContext.run` (an `AsyncLocalStorage` per bus; deep code reads the current message with `bus.getHandlingContext()`) → the bus' `messageLifecycleContext` → `incoming` middleware → dispatch to all matching handlers/workflows, which share the message's outbox, each wrapped in the `handler` middleware. The delete, return or dead-letter happens once, after the incoming chain settles, outside it (`settleMessage` in `bus-instance.ts`), so an incoming middleware that doesn't call `next()` has its message deleted. Success deletes the message. `failMessage()` only flags the lifecycle context, and the message is dead-lettered (`transport.fail(message, failure)`) even if something then throws. A throw or `returnMessage()` asks the `RecoverabilityPolicy` (`recoverability/`, set by `withRecoverability`, default `defaultRecoverability()`), with `failedAttempts` = `TransportMessage.failedAttempts + 1`: `retry(delay)` calls `transport.returnMessage(message, delay)` and `deadLetter()` calls `transport.fail`. A policy that throws, or returns anything but `deadLetter()` or `retry()` with a finite delay ≥ 0 (`isRecoverabilityAction`), dead-letters. Transports never count attempts or dead-letter returned messages themselves; `fail` writes the `MessageFailure` (`createMessageFailure`) as one JSON `bus-failure` header (`FAILURE_HEADER`, `toFailureHeader`/`fromFailureHeader`), which core rejects in `dispatchOutgoing` and every transport also reserves. A message with no handler is discarded (deleted), not returned, retried or dead-lettered. This holds for any queue or transport, including one that checks before queuing (`transport/in-memory-queue.ts`).
 - **Outbox**: `send`/`publish` run the `outgoing` middleware first (after correlation and sticky attributes are merged), and its last step buffers the message in the outbox, which is per received message and shared by all its handlers (`runInOutbox` around `dispatchMessageToHandlers`, innermost in the incoming chain), so it's only dispatched once every handler resolves, and dropped if any fails. Workflow state is held there too: `runInOutbox` puts a store in the bus' internal `UnitOfWorkContext` (`outbox/unit-of-work-context.ts`, shared with `WorkflowRegistry` through `prepare`), which is the transaction with `withOutbox()` or a `BufferedWorkflowStateStore` without it; the registry reads and saves through it, and `flushOutbox` applies the buffered saves (each still checking `$version`, a conflict failing the message) before storing delayed messages and dispatching (`dispatchOutboxed`). Each buffered message and save is tagged with its `UnitOfWorkScope` (`UnitOfWorkContext.runInScope`: one per handler call in `dispatchMessageToHandler`, and one per workflow instance in the registry's `dispatchMessageToWorkflowInstance`, so a `when` that matches several instances tags each separately); when a save fails part way, `sendForSavedScopes` still sends the messages of the scopes whose own state was kept (a retry's guard would skip them), then the outbox is discarded and the message fails. If `failMessage()` or `returnMessage()` was called by then, `flushOutbox` discards it all first (`discardOutbox`), before anything is dispatched, and the workflow registry doesn't save the state it returned. Sends with no outbox (incoming middleware, or after the handlers resolved) go straight to the transport; late sends after a failure are dropped with a warning. Flushing the outbox calls the transport with the finished message and doesn't rerun middleware. `correlationId` and `stickyAttributes` propagate from the incoming message to outgoing ones; `messageId` and `sentAt` never do: the bus stamps new ones on every outgoing message unless the caller passes them, before the outgoing middleware runs, so middleware sees them.
 - **Middleware** (`middleware/`): `withMiddleware(...BusMiddleware[])` with `incoming`, `handler` and `outgoing` stages, each an async `(context, next)` function run in registration order (first outermost) by the internal `MiddlewarePipeline`, which the configuration builds per bus. Not calling `next()` short-circuits (message deleted / handler skipped / message dropped); a second `next()` throws `MiddlewareNextCalledTwice`. Outgoing middleware sets transport-native `headers`, passed to `Transport.send/publish` as `TransportSendOptions`. The terminal step snapshots the context (attributes two levels deep, headers) and calls the optional `Transport.assertSendOptions` before buffering or sending, so a reserved name (`TransportHeaderReserved`) rejects the caller's `send()` rather than failing the outbox flush. `IncomingContext.requestedSettlement()` reads whether `failMessage()`/`returnMessage()` was called (from the lifecycle context), and `OutgoingContext.dispatched` settles when the message reaches the transport or a delayed one is stored (`storeAndSettle`; `dueAt` is on the send and publish contexts), or rejects with the transport or persistence error or `OutgoingMessageDropped` (`dropOutgoing`/`discardOutbox`; a buffered message is flushed inside an `AsyncLocalStorage.snapshot()` of its `next()`). There are no lifecycle emitters: extend the middleware contexts instead.

README.md

@@ -23,6 +23,7 @@ Requires Node.js 24 or later.
 - [@node-ts/bus-postgres](https://github.com/node-ts/bus/tree/master/packages/bus-postgres) - A Postgres persistence for workflow state
 - [@node-ts/bus-mongodb](https://github.com/node-ts/bus/tree/master/packages/bus-mongodb) - A MongoDB persistence for workflow state
 - [@node-ts/bus-opentelemetry](https://github.com/node-ts/bus/tree/master/packages/bus-opentelemetry) - OpenTelemetry tracing and metrics, as bus middleware
+- [@node-ts/bus-nestjs](https://github.com/node-ts/bus/tree/master/packages/bus-nestjs) - A NestJS module: handlers and workflows as providers, and the bus run by Nest's lifecycle
 - [@node-ts/bus-test](https://github.com/node-ts/bus/tree/master/packages/bus-test) - The conformance test suites for transport and persistence adapters
 
 ## Contributing

docs/.vitepress/config.mts

@@ -92,6 +92,7 @@ const guide: DefaultTheme.SidebarItem[] = [
       { text: 'Transactional outbox', link: '/guide/outbox' },
       { text: 'OpenTelemetry', link: '/guide/opentelemetry' },
       { text: 'Dependency injection', link: '/guide/dependency-injection' },
+      { text: 'NestJS', link: '/guide/nestjs' },
       {
         text: 'Long running processes',
         link: '/guide/long-running-processes'

docs/guide/dependency-injection.md

@@ -27,7 +27,10 @@ To construct them with their dependencies, pass an adapter to your IoC container
 
 `get` is also given the message being handled and its attributes, for containers that resolve differently per message, and may return a promise.
 
+In a NestJS application, [`@node-ts/bus-nestjs`](/guide/nestjs) registers handlers and workflows as providers and resolves them from Nest's container.
+
 ## See also
 
 - [Commands](/guide/messages/commands), for class handlers
+- [NestJS](/guide/nestjs)
 - [`ContainerAdapter`](/api/bus-core/interfaces/ContainerAdapter) in the API reference