diff --git a/.changeset/configure-workflow-without-resolving.md b/.changeset/configure-workflow-without-resolving.md new file mode 100644 index 00000000..0f22bc4d --- /dev/null +++ b/.changeset/configure-workflow-without-resolving.md @@ -0,0 +1,7 @@ +--- +'@node-ts/bus-core': minor +--- + +**Breaking:** the bus reads a class workflow's `configureWorkflow()` without creating the workflow (#354). It's called once when the bus provisions or initializes, on an instance created from the class' prototype without running its constructor, and the workflow is only resolved from the container, or constructed, to handle a message. A request-scoped workflow, or one whose dependencies have side effects, is no longer created with no message just to read its mapping. + +A `configureWorkflow()` that uses the workflow's fields or injected dependencies now finds them `undefined`; if it throws, `provision()` or `initialize()` throws the new `WorkflowConfigurationFailed`, naming the workflow. `startedBy` and `when` throw the new `WorkflowMappingInvalid`, which it wraps, when given no message, a handler name that isn't a string, or a lookup without a `lookup` function or `mapsTo` field, so a value read from a field fails at startup rather than on every message. A `configureWorkflow` declared as an arrow function property must become a method. A workflow the container can't resolve now fails the messages it handles with `WorkflowHandlerFailed`, as a class handler does, rather than failing `initialize()`. `testWorkflow()` reads `configureWorkflow()` the same way, so `createWorkflow` is only called for each message. See `MIGRATING.md`. diff --git a/.changeset/nestjs-module.md b/.changeset/nestjs-module.md index 93b979fd..73baa2a5 100644 --- a/.changeset/nestjs-module.md +++ b/.changeset/nestjs-module.md @@ -8,7 +8,7 @@ Add `@node-ts/bus-nestjs`, a NestJS module for the bus (#268), for NestJS 11 and - `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. +- 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` 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`. diff --git a/CLAUDE.md b/CLAUDE.md index 2bdf0a1a..82e65676 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -78,7 +78,7 @@ pnpm exec dotenv -e test.env -- jest packages/bus-sqs/src/sqs-transport.spec.ts ### Workflows (bus-core/src/workflow) -- A workflow extends `Workflow` and implements `configureWorkflow(mapper)`, declaring `startedBy(Message, 'handlerName')` and `when(Message, 'handlerName', { lookup, mapsTo })`. Handlers return a partial state that is merged and saved; `completeWorkflow()` / `discardWorkflow()` end or drop it. +- A workflow extends `Workflow` and implements `configureWorkflow(mapper)`, declaring `startedBy(Message, 'handlerName')` and `when(Message, 'handlerName', { lookup, mapsTo })`. Handlers return a partial state that is merged and saved; `completeWorkflow()` / `discardWorkflow()` end or drop it. `configureClassWorkflow` (`workflow/configure-class-workflow.ts`, used by the registry and `testWorkflow()`) calls `configureWorkflow` on `Object.create(prototype)`, never constructing or resolving the workflow, throws `WorkflowConfigurationFailed` if it isn't a method, and wraps what it throws in it (other than the mapper's `WorkflowAlreadyStartedByMessage`/`WorkflowAlreadyHandlesMessage`; the mapper's `startedBy`/`when` throw `WorkflowMappingInvalid` for an undefined message, non-string handler name or bad lookup, so field-held values fail at startup); the workflow is only resolved, with the message, to handle one (#354). - `defineWorkflow(State)` (`workflow/define-workflow.ts`) is the function equivalent: an immutable builder whose `.startedBy(Message, fn)`, `.when(Message, fn)` and `.when(Message, { lookup, mapsTo }, fn)` handlers get `(message, state, ctx)`, where `ctx` is a `WorkflowContext` (handler context + `attributes`, `complete()`, `discard()`; `workflowContext()` builds a recording fake for tests, and `testWorkflow()` runs either workflow style message by message with the registry's merge rules from `apply-workflow-state-change.ts`). It's named after its state's `$name`. Each handler is a generic `THandler` checked by `CheckedWorkflowHandler` (`workflow-state-change.ts`), which finds returned fields that aren't in the state at any depth; a handler with an annotated return type is only checked against the annotation. Class workflows reuse it: the mapper's `startedBy`/`when` take a generic handler name checked by `CheckedClassWorkflowHandler` (`workflow.ts`) against the method's parameters and return, and both of `WorkflowMapper`'s type parameters are `out`, so `configureWorkflow(mapper: WorkflowMapper)` accepts subclasses but not an unrelated workflow's mapper, and checking it stays cheap (without `out` on the state, tsc measures it as unreliable and falls back to structural checks costing ~200k instantiations). Keep the stored handler names `string`, and write the argument checks in terms of the generic `THandlerName` (`WorkflowHandlerArgument`), since anything else typed by the workflow's keys breaks that variance (TS2636). The handler lists live on the internal `FunctionWorkflowDefinition` (`function-workflow-definition.ts`, not exported). `WorkflowRegistry` turns both styles into the same `WorkflowHandlers` before registering them, and always keeps the persisted `$workflowId`, `$version` and `$name` over what a handler returns. - `WorkflowRegistry` turns these into handler registrations and uses `Persistence` (`getWorkflowState`, `saveWorkflowState`) to load state by the mapped property; the storage of each workflow state is created by `Persistence.provision()`. State carries `$workflowId`, `$status`, and `$version` for optimistic concurrency. When no custom lookup is given, `when` matches on `stickyAttributes.workflowId` → `$workflowId`; the id is put into the sticky attributes when the workflow starts, so messages sent from that point on (and their replies) route back to the same workflow instance. diff --git a/MIGRATING.md b/MIGRATING.md index da72713f..8e45aefd 100644 --- a/MIGRATING.md +++ b/MIGRATING.md @@ -90,6 +90,7 @@ Two things behave differently: `WorkflowHandler`'s parameters are now required, `completeWorkflow()` and `discardWorkflow()` return `WorkflowStateChange`, and `WorkflowMapper`'s `onStartedBy` and `onWhen` store handler names as `string` (`OnWhenHandler` no longer takes type arguments). +- **Class workflows are only created to handle a message.** When the bus provisions or initializes, it reads `configureWorkflow()` from an instance created from the class' prototype, without running its constructor, instead of first resolving the workflow from the container, or constructing it. A `configureWorkflow()` that uses the workflow's fields or injected dependencies now finds them `undefined`, and if it throws or passes the mapper an undefined message, handler name or lookup, `provision()` or `initialize()` throws `WorkflowConfigurationFailed` naming the workflow. A `configureWorkflow` declared as an arrow function property must become a method. Map messages with only the mapper and values that don't come from the instance, and use dependencies in the handler methods. A workflow the container can't resolve now fails each message it handles with `WorkflowHandlerFailed`, as a class handler does, rather than failing `initialize()`. `testWorkflow()` reads `configureWorkflow()` the same way, so its `createWorkflow` is only called for each message. - **`Transport` has a required `endpointName`**, the name of the queue the bus receives from. A custom transport must add it, such as `get endpointName() { return this.configuration.queueName }`. `InMemoryQueue` takes an optional `endpointName`, `in-memory` by default. - **Every message the bus sends has a `messageId` and a `sentAt`** in its `MessageAttributes`: a new UUID and an ISO 8601 timestamp, unless you pass your own to `send` or `publish`. Unlike `correlationId`, they aren't copied from the message being handled. A custom transport must carry them with the message and keep them across retries and in the dead letter queue (see [Custom transports](https://node-ts.github.io/bus/transports/custom)). Tests that compare a handler's attributes with `toEqual` may need `toMatchObject` or `expect.objectContaining`. - **Retry strategies are replaced by a recoverability policy.** `withRetryStrategy()`, `RetryStrategy`, `DefaultRetryStrategy` and `CoreDependencies.retryStrategy` are removed. The bus now decides both the delay and when a message is out of attempts, with the policy passed to `withRecoverability()`: a function of the failure that returns `retry(delay)` or `deadLetter()` (see [Recoverability](https://node-ts.github.io/bus/guide/recoverability)). The default, `defaultRecoverability()`, makes 10 attempts with the same exponential delays as `DefaultRetryStrategy`, so a bus that didn't configure retries behaves as before. diff --git a/docs/guide/nestjs.md b/docs/guide/nestjs.md index d7679a8a..0920c436 100644 --- a/docs/guide/nestjs.md +++ b/docs/guide/nestjs.md @@ -104,7 +104,7 @@ The bus resolves class handlers and workflows for each message. Request-scoped p <<< @/snippets/nestjs.ts#request-scope -A request-scoped class workflow is also resolved once when the bus initializes, to read its `configureWorkflow()`, with no message, so `REQUEST` is `undefined` then. Keep the request in its constructor, and the constructors of the request-scoped providers it depends on, and read its message and attributes when a message is handled. If resolving it fails then, the application fails to start with `WorkflowResolvedWithoutMessage`. +Class workflows are only resolved to handle a message, so a request-scoped workflow, and the request-scoped providers it depends on, can read `REQUEST` in their constructors. The bus reads a class workflow's `configureWorkflow()` without creating it (see [Creating a workflow](/guide/workflows/creating-a-workflow#declaring-the-workflow)). Outside `BusModule`, `nestContainer(moduleRef)` gives `withContainer()` the same container adapter. diff --git a/docs/guide/workflows/creating-a-workflow.md b/docs/guide/workflows/creating-a-workflow.md index 4657cf4c..649fc8fa 100644 --- a/docs/guide/workflows/creating-a-workflow.md +++ b/docs/guide/workflows/creating-a-workflow.md @@ -25,7 +25,9 @@ With functions, `defineWorkflow(State)` returns a workflow that `startedBy` and ::: -A function workflow reaches its dependencies through closures and needs no container. A class workflow is constructed with no arguments, or resolved from a [container](/guide/dependency-injection) when the bus has one. +A function workflow reaches its dependencies through closures and needs no container. A class workflow is constructed with no arguments, or resolved from a [container](/guide/dependency-injection) when the bus has one, for each message it handles. + +The bus calls `configureWorkflow` once when it provisions or initializes, on an instance created from the class' prototype without running its constructor, so it can't use the workflow's fields or dependencies; use those in the handler methods. Declare it as a method, not as a property holding an arrow function, and pass the mapper the message, handler name and lookup directly rather than from a field. If it throws, or passes the mapper an undefined value, `provision()` or `initialize()` throws `WorkflowConfigurationFailed`, naming the workflow. Type a class workflow's mapper with the class itself, as `WorkflowMapper`. The compiler then checks every handler name against the method it names: diff --git a/packages/bus-core/src/service-bus/bus-instance-provisioning.integration.ts b/packages/bus-core/src/service-bus/bus-instance-provisioning.integration.ts index def76c94..2d5cb338 100644 --- a/packages/bus-core/src/service-bus/bus-instance-provisioning.integration.ts +++ b/packages/bus-core/src/service-bus/bus-instance-provisioning.integration.ts @@ -16,6 +16,7 @@ import { PersistenceInitializationOptions, PersistenceProvisionOptions, Workflow, + WorkflowConfigurationFailed, WorkflowMapper } from '../workflow' import { @@ -28,18 +29,13 @@ import { BusConfiguration } from './bus-configuration' import { BusInstance } from './bus-instance' /** - * A workflow that can't be constructed, so registering it fails + * A workflow whose configureWorkflow() throws, so registering it fails */ -class UnconstructableWorkflow extends Workflow { - constructor() { - super() - throw new Error('UnconstructableWorkflow failed to construct') - } - +class UnconfigurableWorkflow extends Workflow { configureWorkflow( - mapper: WorkflowMapper + _mapper: WorkflowMapper ): void { - mapper.withState(TestWorkflowState) + throw new Error('UnconfigurableWorkflow failed to configure') } } @@ -439,7 +435,7 @@ describe('BusInstance provisioning', () => { .withLogger(() => Mock.ofType().object) .withMessageTypes(testMessageTypes) .withTransport(new ProvisioningQueue()) - .withWorkflow(UnconstructableWorkflow) + .withWorkflow(UnconfigurableWorkflow) .build() provisionError = await sut.provision().catch((e: unknown) => e) initializeError = await sut.initialize().catch((e: unknown) => e) @@ -448,8 +444,9 @@ describe('BusInstance provisioning', () => { afterAll(async () => sut.dispose()) it('should fail provisioning', () => { - expect((provisionError as Error).message).toEqual( - 'UnconstructableWorkflow failed to construct' + expect(provisionError).toBeInstanceOf(WorkflowConfigurationFailed) + expect((provisionError as WorkflowConfigurationFailed).cause).toEqual( + new Error('UnconfigurableWorkflow failed to configure') ) }) diff --git a/packages/bus-core/src/testing/test-workflow.spec.ts b/packages/bus-core/src/testing/test-workflow.spec.ts index b4d15a4e..1e21e67b 100644 --- a/packages/bus-core/src/testing/test-workflow.spec.ts +++ b/packages/bus-core/src/testing/test-workflow.spec.ts @@ -1,6 +1,10 @@ import { ReturnAddressMissing } from '../error' import { InvalidDeliveryOptions } from '../outgoing-message' -import { defineWorkflow, WorkflowStateNotProvided } from '../workflow' +import { + defineWorkflow, + WorkflowConfigurationFailed, + WorkflowStateNotProvided +} from '../workflow' import { TestCommand } from '../workflow/test/test-command' import { TestDiscardedWorkflow } from '../workflow/test/test-discarded-workflow' import { @@ -419,8 +423,8 @@ describe('testWorkflow', () => { await sut.when(TestPaymentReceived({ orderId: '1' })) }) - it('should create one to configure it and one for each message', () => { - expect(created).toEqual(3) + it('should create one for each message, and none to configure it', () => { + expect(created).toEqual(2) }) }) @@ -441,6 +445,26 @@ describe('testWorkflow', () => { }) }) + describe('when a class workflow reads its fields in configureWorkflow', () => { + class FieldReadingWorkflow extends Workflow { + private readonly options = { stateType: TestScenarioState } + + configureWorkflow( + mapper: WorkflowMapper + ): void { + mapper.withState(this.options.stateType) + } + } + + it('should throw WorkflowConfigurationFailed, as the bus does', () => { + expect(() => + testWorkflow(FieldReadingWorkflow, { + createWorkflow: () => new FieldReadingWorkflow() + }) + ).toThrow(WorkflowConfigurationFailed) + }) + }) + describe('when a handler replies', () => { let result: WorkflowScenarioResult diff --git a/packages/bus-core/src/testing/test-workflow.ts b/packages/bus-core/src/testing/test-workflow.ts index 38a487a4..347dbd61 100644 --- a/packages/bus-core/src/testing/test-workflow.ts +++ b/packages/bus-core/src/testing/test-workflow.ts @@ -8,12 +8,11 @@ import { MessageWorkflowMapping, Workflow, WorkflowHandlerResult, - WorkflowMapper, WorkflowState, - WorkflowStateNotProvided, WorkflowStatus } from '../workflow' import { applyWorkflowStateChange } from '../workflow/apply-workflow-state-change' +import { configureClassWorkflow } from '../workflow/configure-class-workflow' import { FunctionWorkflowDefinition, isFunctionWorkflow @@ -131,14 +130,8 @@ const readClassWorkflow = ( throw new WorkflowFactoryMissing(workflowType.name) } const create = createWorkflow ?? (() => new workflowType()) - const mapper = new WorkflowMapper>( - workflowType - ) - create().configureWorkflow(mapper) - const workflowStateType = mapper.workflowStateCtor - if (!workflowStateType) { - throw new WorkflowStateNotProvided(workflowType.name) - } + // Read as the bus reads it, without creating the workflow + const { mapper, workflowStateType } = configureClassWorkflow(workflowType) // A new workflow handles each message, as on a bus const handlerOf = @@ -285,6 +278,8 @@ const outgoingAttributes = ( * @param options how to create a class workflow, members of every handler's context, and when the clock starts * @returns the scenario, with no workflow instance yet * @throws WorkflowStateNotProvided if a class workflow doesn't declare its state with `mapper.withState()` + * @throws WorkflowConfigurationFailed if a class workflow's `configureWorkflow()` throws, which is called without + * creating the workflow, as on a bus * @throws WorkflowFactoryMissing if a class workflow's constructor takes arguments and `createWorkflow` isn't given * @example * const scenario = testWorkflow(orderWorkflow) diff --git a/packages/bus-core/src/workflow/configure-class-workflow.ts b/packages/bus-core/src/workflow/configure-class-workflow.ts new file mode 100644 index 00000000..245c2bd3 --- /dev/null +++ b/packages/bus-core/src/workflow/configure-class-workflow.ts @@ -0,0 +1,70 @@ +import { ClassConstructor } from '../util' +import { + WorkflowAlreadyHandlesMessage, + WorkflowAlreadyStartedByMessage, + WorkflowConfigurationFailed, + WorkflowMappingInvalid, + WorkflowStateNotProvided +} from './error' +import { Workflow, WorkflowMapper } from './workflow' +import { WorkflowState } from './workflow-state' + +/** + * A class workflow's mapping, as its `configureWorkflow()` declared it + */ +export interface ClassWorkflowConfiguration< + TWorkflowState extends WorkflowState +> { + mapper: WorkflowMapper> + workflowStateType: ClassConstructor +} + +/** + * Reads a class workflow's mapping by calling its `configureWorkflow()` on an instance created from its prototype, + * without running its constructor, so neither the workflow nor its dependencies are created just to read it + * @param workflowType the class workflow + * @returns the mapper it configured, and its state + * @throws WorkflowConfigurationFailed if `configureWorkflow` isn't a method, or throws an error other than the + * mapper's `WorkflowAlreadyStartedByMessage` or `WorkflowAlreadyHandlesMessage`, such as `WorkflowMappingInvalid` + * @throws WorkflowStateNotProvided if it doesn't declare its state with `mapper.withState()` + */ +export const configureClassWorkflow = ( + workflowType: ClassConstructor> +): ClassWorkflowConfiguration => { + // An arrow function property only exists on instances, which aren't created here + if (typeof workflowType.prototype.configureWorkflow !== 'function') { + throw new WorkflowConfigurationFailed( + workflowType.name, + new TypeError(`configureWorkflow isn't a method of ${workflowType.name}`), + `Declare ${workflowType.name}.configureWorkflow(mapper) as a method, not as a property holding an arrow` + + ' function: the bus calls it without constructing the workflow, so properties are never set.' + ) + } + const mapper = new WorkflowMapper>( + workflowType + ) + const unconstructed = Object.create( + workflowType.prototype as object + ) as Workflow + try { + unconstructed.configureWorkflow(mapper) + } catch (error) { + if ( + error instanceof WorkflowAlreadyStartedByMessage || + error instanceof WorkflowAlreadyHandlesMessage + ) { + throw error + } + throw new WorkflowConfigurationFailed( + workflowType.name, + error, + error instanceof WorkflowMappingInvalid ? error.help : undefined + ) + } + + const workflowStateType = mapper.workflowStateCtor + if (!workflowStateType) { + throw new WorkflowStateNotProvided(workflowType.name) + } + return { mapper, workflowStateType } +} diff --git a/packages/bus-core/src/workflow/error/index.ts b/packages/bus-core/src/workflow/error/index.ts index f1b32e43..964dbe03 100644 --- a/packages/bus-core/src/workflow/error/index.ts +++ b/packages/bus-core/src/workflow/error/index.ts @@ -1,8 +1,10 @@ export * from './workflow-already-handles-message' export * from './workflow-already-initialized' export * from './workflow-already-started-by-message' +export * from './workflow-configuration-failed' export * from './workflow-does-not-handle-message' export * from './workflow-handler-failed' +export * from './workflow-mapping-invalid' export * from './workflow-name-already-registered' export * from './workflow-not-recognized' export * from './workflow-registered-after-initialization' diff --git a/packages/bus-core/src/workflow/error/workflow-configuration-failed.ts b/packages/bus-core/src/workflow/error/workflow-configuration-failed.ts new file mode 100644 index 00000000..ac61d806 --- /dev/null +++ b/packages/bus-core/src/workflow/error/workflow-configuration-failed.ts @@ -0,0 +1,31 @@ +/** + * Thrown when a class workflow's `configureWorkflow()` throws, or isn't a method. The bus calls it once when it + * provisions or initializes, on an instance created from the workflow's prototype without running its constructor, + * so fields set in the constructor or by a container are `undefined` there. + */ +export class WorkflowConfigurationFailed extends Error { + readonly help: string + + /** + * @param workflowName Name of the workflow class whose `configureWorkflow()` failed + * @param cause The error it threw, or why it couldn't be called + * @param help How to fix it, if not by keeping the workflow's fields and dependencies out of `configureWorkflow()` + */ + constructor( + readonly workflowName: string, + // A parameter property rather than `super(message, { cause })`, so it's enumerable and logged with the error + readonly cause: unknown, + help?: string + ) { + super( + `${workflowName}.configureWorkflow() failed: ${cause instanceof Error ? cause.message : String(cause)}` + ) + this.help = + help ?? + `${workflowName}.configureWorkflow() is called on an instance created without running its constructor, so it` + + " can't use the workflow's fields or dependencies. Map messages with only the mapper and values that don't" + + ' come from the instance, and use dependencies in the handler methods.' + + Object.setPrototypeOf(this, new.target.prototype) + } +} diff --git a/packages/bus-core/src/workflow/error/workflow-mapping-invalid.ts b/packages/bus-core/src/workflow/error/workflow-mapping-invalid.ts new file mode 100644 index 00000000..1ae44c3a --- /dev/null +++ b/packages/bus-core/src/workflow/error/workflow-mapping-invalid.ts @@ -0,0 +1,30 @@ +/** + * Thrown when a class workflow's `configureWorkflow()` passes `startedBy` or `when` a value that can't be used: no + * message, a handler name that isn't a string, or a lookup without a `lookup` function or a `mapsTo` field. The bus + * reads `configureWorkflow()` without constructing the workflow, so a value held in one of its fields is `undefined` + * there. + */ +export class WorkflowMappingInvalid extends Error { + readonly help: string + + /** + * @param workflowName Name of the workflow class + * @param mapperMethod The mapper method that was given the value, `startedBy` or `when` + * @param problem What's wrong with the value, such as "a lookup that isn't a function" + */ + constructor( + readonly workflowName: string, + readonly mapperMethod: 'startedBy' | 'when', + readonly problem: string + ) { + super( + `Workflow ${workflowName} calls mapper.${mapperMethod}() with ${problem}` + ) + this.help = + `Pass mapper.${mapperMethod}() the message, the name of a handler method and, for when(), a lookup with a` + + " lookup function and a mapsTo field. Don't read them from the workflow's fields, which aren't set when" + + ` ${workflowName}.configureWorkflow() is called.` + + Object.setPrototypeOf(this, new.target.prototype) + } +} diff --git a/packages/bus-core/src/workflow/registry/workflow-registry.spec.ts b/packages/bus-core/src/workflow/registry/workflow-registry.spec.ts index 3d9884ef..21ac1402 100644 --- a/packages/bus-core/src/workflow/registry/workflow-registry.spec.ts +++ b/packages/bus-core/src/workflow/registry/workflow-registry.spec.ts @@ -1,4 +1,5 @@ import { MessageAttributes, messageAttributes } from '@node-ts/bus-messages' +import { EventEmitter, once } from 'node:events' import { IMock, It, Mock, Times } from 'typemoq' import { ContainerAdapter } from '../../container' import { @@ -7,18 +8,24 @@ import { Handler, HandlerContext, HandlerDefinition, + HandlerDispatchRejected, HandlerRegistry } from '../../handler' import { DebugLogger, Logger } from '../../logger' import { MessageHandlingContext } from '../../message-handling-context' import { MessageLifecycleContext } from '../../message-lifecycle-context' import { UnitOfWorkContext } from '../../outbox/unit-of-work-context' +import { deadLetter } from '../../recoverability' import { Bus, BusInstance } from '../../service-bus' import { testMessageTypes } from '../../test' import { InMemoryQueue } from '../../transport' import { CoreDependencies, sleep } from '../../util' import { FunctionWorkflow } from '../define-workflow' import { + WorkflowAlreadyStartedByMessage, + WorkflowConfigurationFailed, + WorkflowHandlerFailed, + WorkflowMappingInvalid, WorkflowNameAlreadyRegistered, WorkflowNotRecognized, WorkflowRegisteredAfterInitialization, @@ -162,60 +169,317 @@ describe('WorkflowRegistry', () => { }) }) - describe('when initializing', () => { - beforeEach(() => { + describe('when initializing a class workflow', () => { + let container: IMock + let constructed: number + + beforeAll(async () => { + constructed = 0 + class CountingWorkflow extends TestWorkflow { + constructor(bus: BusInstance) { + super(bus) + constructed++ + } + } + container = Mock.ofType() sut = new WorkflowRegistry() - sut.register(TestWorkflow) sut.prepare( - { - loggerFactory: (name: string) => new DebugLogger(name) - } as unknown as CoreDependencies, + coreDependencies, persistence.object, new MessageHandlingContext(), new MessageLifecycleContext(), new UnitOfWorkContext() ) + sut.register(CountingWorkflow) + await sut.initialize(new DefaultHandlerRegistry(), container.object) }) - describe('without a container', () => { - it('should construct workflow instances', async () => { - await sut.initialize(new DefaultHandlerRegistry(), undefined) - }) + it('should not construct the workflow', () => { + expect(constructed).toEqual(0) }) - describe('with a container', () => { - let container: IMock + it('should not resolve the workflow from the container', () => { + container.verify(c => c.get(It.isAny(), It.isAny()), Times.never()) + }) - beforeEach(() => { - container = Mock.ofType() - container - .setup(c => c.get(TestWorkflow)) - .returns(() => new TestWorkflow(Mock.ofType().object)) - .verifiable(Times.once()) - }) + it('should read its state from configureWorkflow', () => { + expect(sut.getWorkflowStateNames()).toEqual([TestWorkflowState.NAME]) + }) + }) - it('should fetch workflows from the container', async () => { - await sut.initialize(new DefaultHandlerRegistry(), container.object) - container.verifyAll() - }) + describe('when a class workflow reads its fields in configureWorkflow', () => { + let error: unknown + + beforeAll(async () => { + class FieldReadingWorkflow extends Workflow { + private readonly startedBy = { handler: 'start' as const } + + configureWorkflow( + mapper: WorkflowMapper + ): void { + mapper + .withState(TestWorkflowState) + .startedBy(TestCommand, this.startedBy.handler) + } + + start() { + return {} + } + } + sut = new WorkflowRegistry() + sut.prepare( + coreDependencies, + persistence.object, + new MessageHandlingContext(), + new MessageLifecycleContext(), + new UnitOfWorkContext() + ) + sut.register(FieldReadingWorkflow) + error = await catchError(() => + sut.initialize(new DefaultHandlerRegistry(), undefined) + ) }) - describe('with an async container', () => { - let container: IMock - beforeEach(() => { - container = Mock.ofType() - container - .setup(c => c.get(TestWorkflow)) - .returns(() => - Promise.resolve(new TestWorkflow(Mock.ofType().object)) - ) - .verifiable(Times.once()) - }) + it('should throw WorkflowConfigurationFailed naming the workflow, with the error as its cause and the fix', () => { + expect(error).toBeInstanceOf(WorkflowConfigurationFailed) + const configurationFailed = error as WorkflowConfigurationFailed + expect(configurationFailed.workflowName).toEqual('FieldReadingWorkflow') + expect(configurationFailed.cause).toBeInstanceOf(TypeError) + expect(configurationFailed.help).toContain( + 'without running its constructor' + ) + }) + }) - it('should fetch workflows from the container', async () => { - await sut.initialize(new DefaultHandlerRegistry(), container.object) - container.verifyAll() - }) + describe('when a class workflow is started by a message twice in configureWorkflow', () => { + let error: unknown + + beforeAll(async () => { + class TwiceStartedWorkflow extends Workflow { + configureWorkflow( + mapper: WorkflowMapper + ): void { + mapper + .withState(TestWorkflowState) + .startedBy(TestCommand, 'start') + .startedBy(TestCommand, 'start') + } + + start() { + return {} + } + } + sut = new WorkflowRegistry() + sut.prepare( + coreDependencies, + persistence.object, + new MessageHandlingContext(), + new MessageLifecycleContext(), + new UnitOfWorkContext() + ) + sut.register(TwiceStartedWorkflow) + error = await catchError(() => + sut.initialize(new DefaultHandlerRegistry(), undefined) + ) + }) + + it("should throw the mapper's own error", () => { + expect(error).toBeInstanceOf(WorkflowAlreadyStartedByMessage) + }) + }) + + describe('when a class workflow reads a lookup from a field in configureWorkflow', () => { + let error: unknown + + beforeAll(async () => { + class FieldLookupWorkflow extends Workflow { + private readonly byValue = (message: TaskRan) => message.value + + configureWorkflow( + mapper: WorkflowMapper + ): void { + mapper + .withState(TestWorkflowState) + .startedBy(TestCommand, 'start') + .when(TaskRan, 'ran', { + lookup: this.byValue, + mapsTo: 'property1' + }) + } + + start() { + return {} + } + + ran() { + return {} + } + } + sut = new WorkflowRegistry() + sut.prepare( + coreDependencies, + persistence.object, + new MessageHandlingContext(), + new MessageLifecycleContext(), + new UnitOfWorkContext() + ) + sut.register(FieldLookupWorkflow) + error = await catchError(() => + sut.initialize(new DefaultHandlerRegistry(), undefined) + ) + }) + + it('should throw WorkflowConfigurationFailed, caused by WorkflowMappingInvalid naming the lookup', () => { + expect(error).toBeInstanceOf(WorkflowConfigurationFailed) + const cause = (error as WorkflowConfigurationFailed).cause + expect(cause).toBeInstanceOf(WorkflowMappingInvalid) + expect((cause as WorkflowMappingInvalid).workflowName).toEqual( + 'FieldLookupWorkflow' + ) + expect((cause as WorkflowMappingInvalid).mapperMethod).toEqual('when') + expect((cause as WorkflowMappingInvalid).problem).toEqual( + "a lookup whose lookup isn't a function" + ) + expect((error as WorkflowConfigurationFailed).help).toEqual( + (cause as WorkflowMappingInvalid).help + ) + }) + }) + + describe('when a class workflow reads a handler name from a field in configureWorkflow', () => { + let error: unknown + + beforeAll(async () => { + class FieldHandlerNameWorkflow extends Workflow { + private readonly startHandler = 'start' as const + + configureWorkflow( + mapper: WorkflowMapper + ): void { + mapper + .withState(TestWorkflowState) + .startedBy(TestCommand, this.startHandler) + } + + start() { + return {} + } + } + sut = new WorkflowRegistry() + sut.prepare( + coreDependencies, + persistence.object, + new MessageHandlingContext(), + new MessageLifecycleContext(), + new UnitOfWorkContext() + ) + sut.register(FieldHandlerNameWorkflow) + error = await catchError(() => + sut.initialize(new DefaultHandlerRegistry(), undefined) + ) + }) + + it('should throw WorkflowConfigurationFailed, caused by WorkflowMappingInvalid naming the handler name', () => { + expect(error).toBeInstanceOf(WorkflowConfigurationFailed) + const cause = (error as WorkflowConfigurationFailed).cause + expect(cause).toBeInstanceOf(WorkflowMappingInvalid) + expect((cause as WorkflowMappingInvalid).mapperMethod).toEqual( + 'startedBy' + ) + expect((cause as WorkflowMappingInvalid).problem).toContain( + "a handler name that isn't a string" + ) + }) + }) + + describe('when a class workflow declares configureWorkflow as an arrow function property', () => { + let error: unknown + + beforeAll(async () => { + class ArrowConfiguredWorkflow extends Workflow { + configureWorkflow: ( + mapper: WorkflowMapper + ) => void = mapper => { + mapper.withState(TestWorkflowState).startedBy(TestCommand, 'start') + } + + start() { + return {} + } + } + sut = new WorkflowRegistry() + sut.prepare( + coreDependencies, + persistence.object, + new MessageHandlingContext(), + new MessageLifecycleContext(), + new UnitOfWorkContext() + ) + sut.register(ArrowConfiguredWorkflow) + error = await catchError(() => + sut.initialize(new DefaultHandlerRegistry(), undefined) + ) + }) + + it('should throw WorkflowConfigurationFailed, saying to declare it as a method', () => { + expect(error).toBeInstanceOf(WorkflowConfigurationFailed) + const configurationFailed = error as WorkflowConfigurationFailed + expect(configurationFailed.workflowName).toEqual( + 'ArrowConfiguredWorkflow' + ) + expect(configurationFailed.message).not.toContain('unconstructed') + expect(configurationFailed.help).toContain('as a method') + }) + }) + + describe('when the container fails to resolve a class workflow for a message', () => { + const containerError = new Error('Workflow is not bound') + const errors = new EventEmitter() + const queue = new InMemoryQueue() + let bus: BusInstance + let dispatchError: unknown + + beforeAll(async () => { + bus = Bus.configure() + .withLogger(() => Mock.ofType().object) + .withMessageTypes(testMessageTypes) + .withWorkflow(TestWorkflow) + .withContainer({ + get: () => { + throw containerError + } + }) + .withTransport(queue) + .withRecoverability(() => deadLetter()) + .withMiddleware({ + incoming: async (_, next) => { + try { + await next() + } catch (error) { + errors.emit('failed', error) + throw error + } + } + }) + .build() + + await bus.initialize() + await bus.start() + const failed = once(errors, 'failed') + await bus.send(new TestCommand('abc')) + ;[dispatchError] = await failed + }) + + afterAll(async () => bus.dispose()) + + it('should fail the message with WorkflowHandlerFailed, keeping the container error as its cause', () => { + expect(dispatchError).toBeInstanceOf(HandlerDispatchRejected) + const [rejection] = (dispatchError as HandlerDispatchRejected).rejections + expect(rejection).toBeInstanceOf(WorkflowHandlerFailed) + expect((rejection as WorkflowHandlerFailed).workflowName).toEqual( + 'TestWorkflow' + ) + expect((rejection as WorkflowHandlerFailed).cause).toBe(containerError) }) }) diff --git a/packages/bus-core/src/workflow/registry/workflow-registry.ts b/packages/bus-core/src/workflow/registry/workflow-registry.ts index d63c360b..d96f3e78 100644 --- a/packages/bus-core/src/workflow/registry/workflow-registry.ts +++ b/packages/bus-core/src/workflow/registry/workflow-registry.ts @@ -21,14 +21,14 @@ import { import { TransportMessage } from '../../transport' import { ClassConstructor, CoreDependencies } from '../../util' import { applyWorkflowStateChange } from '../apply-workflow-state-change' +import { configureClassWorkflow } from '../configure-class-workflow' import { FunctionWorkflow } from '../define-workflow' import { WorkflowAlreadyInitialized, WorkflowHandlerFailed, WorkflowNameAlreadyRegistered, WorkflowNotRecognized, - WorkflowRegisteredAfterInitialization, - WorkflowStateNotProvided + WorkflowRegisteredAfterInitialization } from '../error' import { FunctionWorkflowDefinition, @@ -38,7 +38,7 @@ import { import { hasLookupValue } from '../has-lookup-value' import { MessageWorkflowMapping } from '../message-workflow-mapping' import { PersistedWorkflow, Persistence } from '../persistence' -import { Workflow, WorkflowMapper } from '../workflow' +import { Workflow } from '../workflow' import { WorkflowContext } from '../workflow-context' import { WorkflowState, WorkflowStatus } from '../workflow-state' import { WorkflowHandlerResult } from '../workflow-state-change' @@ -227,7 +227,7 @@ export class WorkflowRegistry { const workflowHandlers = isFunctionWorkflow(workflow) ? this.getFunctionWorkflowHandlers(workflow) - : await this.getClassWorkflowHandlers(workflow, container) + : this.getClassWorkflowHandlers(workflow, container) this.workflowStateNames.push( new workflowHandlers.workflowStateType().$name @@ -281,32 +281,15 @@ export class WorkflowRegistry { } /** - * Reads the handlers of a class workflow by calling its `configureWorkflow`. The workflow is resolved from the - * container, or constructed with no arguments, each time it handles a message. + * Reads the handlers of a class workflow from its `configureWorkflow`, which is called without constructing the + * workflow. The workflow is resolved from the container, or constructed with no arguments, only to handle a + * message, each time it handles one. */ - private async getClassWorkflowHandlers( + private getClassWorkflowHandlers( WorkflowCtor: ClassConstructor>, container: ContainerAdapter | undefined - ): Promise { - let workflowInstance - if (container) { - const workflowInstanceFromContainer = container.get(WorkflowCtor) - if (workflowInstanceFromContainer instanceof Promise) { - workflowInstance = await workflowInstanceFromContainer - } else { - workflowInstance = workflowInstanceFromContainer - } - } else { - workflowInstance = new WorkflowCtor() - } - const mapper = new WorkflowMapper(WorkflowCtor) - workflowInstance.configureWorkflow(mapper) - - if (!mapper.workflowStateCtor) { - throw new WorkflowStateNotProvided( - WorkflowCtor.prototype.constructor.name - ) - } + ): WorkflowHandlers { + const { mapper, workflowStateType } = configureClassWorkflow(WorkflowCtor) const invokerFor = ( @@ -341,7 +324,7 @@ export class WorkflowRegistry { return { workflowName: WorkflowCtor.name, - workflowStateType: mapper.workflowStateCtor, + workflowStateType, startedBy: new Map( Array.from(mapper.onStartedBy, ([messageType, options]) => [ messageType, diff --git a/packages/bus-core/src/workflow/workflow.integration.ts b/packages/bus-core/src/workflow/workflow.integration.ts index 6324c7c0..928965b8 100644 --- a/packages/bus-core/src/workflow/workflow.integration.ts +++ b/packages/bus-core/src/workflow/workflow.integration.ts @@ -39,13 +39,18 @@ describe('Workflow', () => { const CONSUME_TIMEOUT = 2000 let bus: BusInstance const inMemoryPersistence = new InMemoryPersistence() + const resolvedFor: (Message | undefined)[] = [] beforeAll(async () => { bus = Bus.configure() .withMessageTypes(testMessageTypes) .withPersistence(inMemoryPersistence) .withContainer({ - get(workflowType: ClassConstructor) { + get( + workflowType: ClassConstructor, + context?: { message?: Message } + ) { + resolvedFor.push(context?.message) return new workflowType(bus) } }) @@ -64,6 +69,13 @@ describe('Workflow', () => { await bus.dispose() }) + describe('when the bus has a container', () => { + it('should resolve class workflows from it only to handle a message', () => { + expect(resolvedFor.length).toBeGreaterThan(0) + expect(resolvedFor).not.toContain(undefined) + }) + }) + describe('when a message that starts a workflow is received', () => { const propertyMapping: MessageWorkflowMapping< TestCommand, diff --git a/packages/bus-core/src/workflow/workflow.ts b/packages/bus-core/src/workflow/workflow.ts index c564fff2..e05ec0c3 100644 --- a/packages/bus-core/src/workflow/workflow.ts +++ b/packages/bus-core/src/workflow/workflow.ts @@ -7,7 +7,8 @@ import { HandlerContext } from '../handler' import { ClassConstructor } from '../util' import { WorkflowAlreadyHandlesMessage, - WorkflowAlreadyStartedByMessage + WorkflowAlreadyStartedByMessage, + WorkflowMappingInvalid } from './error' import { MessageWorkflowMapping } from './message-workflow-mapping' import { WorkflowState, WorkflowStatus } from './workflow-state' @@ -150,6 +151,36 @@ export type OnWhenHandler = { customLookup: MessageWorkflowMapping | undefined } +/** + * What's wrong with the arguments given to `startedBy` or `when`, if anything + */ +const findMappingProblem = ( + message: unknown, + workflowHandler: unknown, + customLookup: unknown +): string | undefined => { + if (message === undefined || message === null) { + return `no message (${String(message)})` + } + if (typeof workflowHandler !== 'string') { + return `a handler name that isn't a string (${String(workflowHandler)})` + } + if (customLookup === undefined) { + return undefined + } + if (typeof customLookup !== 'object' || customLookup === null) { + return `a lookup that isn't an object (${String(customLookup)})` + } + const { lookup, mapsTo } = customLookup as Partial + if (typeof lookup !== 'function') { + return "a lookup whose lookup isn't a function" + } + if (typeof mapsTo !== 'string') { + return "a lookup whose mapsTo isn't a string" + } + return undefined +} + /** * A workflow configuration that describes how to map incoming messages to handlers within the workflow. */ @@ -192,6 +223,7 @@ export class WorkflowMapper< * @param workflowHandler The name of the workflow method that handles `message`. It must take `message` and return * changes to the workflow state, or nothing, with no fields that aren't in the state. * @throws WorkflowAlreadyStartedByMessage if the workflow is already started by `message` + * @throws WorkflowMappingInvalid if `message` is missing or `workflowHandler` isn't a string * @example * mapper.withState(OrderState).startedBy(OrderPlaced, 'start') */ @@ -207,6 +239,7 @@ export class WorkflowMapper< WorkflowStateType > ): this { + this.assertMapping('startedBy', message, workflowHandler, undefined) if (this.onStartedBy.has(message)) { throw new WorkflowAlreadyStartedByMessage(this.workflow.name, message) } @@ -225,6 +258,8 @@ export class WorkflowMapper< * @param customLookup How to find the workflow instance for `message`. By default it's found by the `workflowId` * sticky attribute that's added to messages sent from the workflow. * @throws WorkflowAlreadyHandlesMessage if the workflow already handles `message` + * @throws WorkflowMappingInvalid if `message` is missing, `workflowHandler` isn't a string, or `customLookup` has + * no `lookup` function or `mapsTo` field * @example * mapper.when(CardCharged, 'charged', { lookup: message => message.orderId, mapsTo: 'orderId' }) */ @@ -241,6 +276,7 @@ export class WorkflowMapper< >, customLookup?: MessageWorkflowMapping ): this { + this.assertMapping('when', message, workflowHandler, customLookup) if (this.onWhen.has(message)) { throw new WorkflowAlreadyHandlesMessage(this.workflow.name, message) } @@ -254,6 +290,27 @@ export class WorkflowMapper< }) return this } + + /** + * Checks what `startedBy` or `when` was given can be used, since a value that's undefined at runtime, such as one + * read from a field of the workflow that `configureWorkflow` is called without, would otherwise only fail when a + * message is handled + */ + private assertMapping( + mapperMethod: 'startedBy' | 'when', + message: unknown, + workflowHandler: unknown, + customLookup: unknown + ): void { + const problem = findMappingProblem(message, workflowHandler, customLookup) + if (problem) { + throw new WorkflowMappingInvalid( + this.workflow.name, + mapperMethod, + problem + ) + } + } } /** @@ -273,7 +330,10 @@ export class WorkflowMapper< */ export abstract class Workflow { /** - * Maps the messages the workflow handles to its handler methods + * Maps the messages the workflow handles to its handler methods. The bus calls it once when it provisions or + * initializes, on an instance created from the class' prototype without running its constructor, so it can't use + * the workflow's fields or dependencies; use those in the handler methods. Declare it as a method, not as an arrow + * function property. * @param mapper Declares the workflow state, and which methods start the workflow or handle messages. Type it with * the workflow class, such as `WorkflowMapper`, so handler names are checked. With * `any` no handler name compiles. diff --git a/packages/bus-nestjs/CLAUDE.md b/packages/bus-nestjs/CLAUDE.md index 604cd2d0..9397f164 100644 --- a/packages/bus-nestjs/CLAUDE.md +++ b/packages/bus-nestjs/CLAUDE.md @@ -11,7 +11,7 @@ A NestJS module that runs a bus in a Nest application. Read the root `CLAUDE.md` - `forFeature`/`forFeatureAsync` only register: they never make classes providers (a dynamic module only sees its own `imports`, so a class provided there couldn't get its module's providers). Users keep classes in their own module's `providers`, and `BusClassNotProvided` reports one that isn't. `forFeatureAsync`'s `imports`/`inject` are only for its factory. - `BusLifecycleHost.onModuleInit` finds registrations with `DiscoveryService`: `BusFeatureRegistration` instances (from `forFeature`'s value provider or `forFeatureAsync`'s factory, each under a `Symbol` whose description is `BUS_FEATURE_TOKEN_PREFIX` + bus name, so one Nest didn't create, because it injects a request-scoped provider, throws `BusFeatureNotStatic`), and decorated providers (`classOf()` reads a factory or value provider's class from its instance, as Nest's discovery does). Every host checks for duplicate bus names (`BusAlreadyRegistered`) and registrations for unknown buses (`BusNotRegistered`), then registers its own, checks each class handler and workflow with `moduleRef.introspect` (`BusClassNotProvided`), builds, and throws `BusCoreVersionNotSupported` if the bus has no boolean `canStart` (an older bus-core peer). `forRootAsync` warns when the factory returns a configuration other than the seed it was given. - Lifecycle: `onApplicationBootstrap` initializes and, when `bus.canStart` (bus-core), starts, unless `lifecycle: 'manual'`, and disposes the bus (once) before rethrowing if either fails; `onModuleDestroy` stops a started bus; `onApplicationShutdown` disposes. `BusModule` is global, and Nest runs each shutdown hook of global modules after that hook of every non-global module, so the bus keeps handling messages through other modules' `onModuleDestroy`, stops before any `beforeApplicationShutdown`, and is disposed after other modules' `onApplicationShutdown` (pinned by the shutdown-order test; the same in Nest 11.0.0). Other `@Global()` modules run their hooks in import order relative to `BusModule`: one imported before `forRoot()` bootstraps before the bus is initialized, and one imported after it runs `onModuleDestroy` while the bus is still handling messages. Docs tell users to release resources in `beforeApplicationShutdown`/`onApplicationShutdown`, or use `lifecycle: 'manual'` and `bus.stop()` before `app.close()`. In an application from `createBusForProvisioning()` (which adds a global module providing `BUS_PROVISIONING`), hosts only build. The app is never closed there: `bus provision` disposes the bus and `bus.mjs` exits. -- `nestContainer(moduleRef)`: `introspect` gives the scope (`REQUEST` also for providers that depend on request-scoped ones). Singletons use `moduleRef.get(type, { strict: false })`; others `moduleRef.resolve` with a `ContextId` per message object (a `WeakMap`), registering `{ message, attributes }` (`BusRequest`) as `REQUEST` once. No message (bus-core resolving a class workflow to read `configureWorkflow()`) gets a fresh context with no request, and a failure then is rethrown as `WorkflowResolvedWithoutMessage`. The request scope is keyed by the message object, so the in-memory queue's retries of the same object share it (#353). Reading `configureWorkflow()` without resolving the workflow is #354. +- `nestContainer(moduleRef)`: `introspect` gives the scope (`REQUEST` also for providers that depend on request-scoped ones). Singletons use `moduleRef.get(type, { strict: false })`; others `moduleRef.resolve` with a `ContextId` per message object (a `WeakMap`), registering `{ message, attributes }` (`BusRequest`) as `REQUEST` once. bus-core only resolves classes with a message (it reads a class workflow's `configureWorkflow()` from its prototype, #354), so a call with no message, only from users, gets a fresh context with no request. The request scope is keyed by the message object, so the in-memory queue's retries of the same object share it (#353). ## Tests diff --git a/packages/bus-nestjs/src/bus-module.integration.ts b/packages/bus-nestjs/src/bus-module.integration.ts index ad46603f..35d38211 100644 --- a/packages/bus-nestjs/src/bus-module.integration.ts +++ b/packages/bus-nestjs/src/bus-module.integration.ts @@ -29,8 +29,7 @@ import { BusClassNotProvided, BusFeatureNotStatic, BusNotBuilt, - BusNotRegistered, - WorkflowResolvedWithoutMessage + BusNotRegistered } from './error' import { getBusToken } from './get-bus-token' import { InjectBus } from './inject-bus' @@ -460,6 +459,36 @@ describe('BusModule', () => { }) }) + describe('when a request-scoped class workflow reads REQUEST in its constructor', () => { + const queue = new InMemoryQueue() + let app: TestingModule + let recorder: Recorder + + beforeAll(async () => { + app = await startApp({ + imports: [ + recorderModule(), + BusModule.forRoot({ + configure: configuration => + configuration.withTransport(queue).withMessageTypes(messageTypes) + }), + BusModule.forFeature({ workflows: [RequestScopedWorkflow] }) + ], + providers: [RequestScopedWorkflow] + }) + recorder = app.get(Recorder) + await app.get(BusInstance).publish(new OrderPlaced('order-1')) + await queue.idle() + }) + + afterAll(async () => app.close()) + + it('should create the workflow with the message it handles as the request', () => { + const [started] = recorder.by(RequestScopedWorkflow.name) + expect(started.detail).toEqual('order-1') + }) + }) + describe('when an application has several buses', () => { const defaultQueue = new InMemoryQueue() const billingQueue = new InMemoryQueue() @@ -675,40 +704,6 @@ describe('BusModule', () => { }) }) - describe('and a request-scoped class workflow reads REQUEST in its constructor', () => { - const queue = new ProvisionedQueue() - let error: unknown - - beforeAll(async () => { - error = await startFails({ - imports: [ - BusModule.forRoot({ - configure: configuration => - configuration - .withTransport(queue) - .withMessageTypes(messageTypes) - }), - BusModule.forFeature({ workflows: [RequestScopedWorkflow] }) - ], - providers: [RequestScopedWorkflow] - }) - }) - - it('should throw WorkflowResolvedWithoutMessage, naming the workflow', () => { - expect(error).toBeInstanceOf(WorkflowResolvedWithoutMessage) - const resolvedWithoutMessage = error as WorkflowResolvedWithoutMessage - expect(resolvedWithoutMessage.className).toEqual( - RequestScopedWorkflow.name - ) - expect(resolvedWithoutMessage.cause).toBeInstanceOf(TypeError) - expect(resolvedWithoutMessage.help).toContain('REQUEST') - }) - - it('should dispose the bus', () => { - expect(queue.calls).toEqual(['dispose']) - }) - }) - describe('and the transport fails to initialize', () => { const transportError = new Error('Connection refused') const queue = new ProvisionedQueue([], transportError) diff --git a/packages/bus-nestjs/src/bus-request.ts b/packages/bus-nestjs/src/bus-request.ts index 5464fc1b..51adf8e1 100644 --- a/packages/bus-nestjs/src/bus-request.ts +++ b/packages/bus-nestjs/src/bus-request.ts @@ -4,9 +4,6 @@ import { MessageAttributes } from '@node-ts/bus-messages' * What a request-scoped provider gets when it injects Nest's `REQUEST` while the bus resolves it to handle a * message: the message and its attributes. Each received message has its own request scope, shared by every class * handler and workflow that handles it. - * - * A request-scoped class workflow is also resolved once without a message, when the bus reads its - * `configureWorkflow()`, so `REQUEST` is `undefined` then. * @example * ```ts * @Injectable({ scope: Scope.REQUEST }) diff --git a/packages/bus-nestjs/src/bus-workflow.ts b/packages/bus-nestjs/src/bus-workflow.ts index fd31c6fa..86509333 100644 --- a/packages/bus-nestjs/src/bus-workflow.ts +++ b/packages/bus-nestjs/src/bus-workflow.ts @@ -3,10 +3,9 @@ import { BUS_WORKFLOW_METADATA } from './bus-discovery-metadata' import { BusRegistrationOptions } from './bus-registration-options' /** - * Registers a class workflow that's a Nest provider with a bus, which resolves it from Nest's container when it - * reads the workflow's `configureWorkflow()` and for each message it handles. Only classes that extend `Workflow` - * can be decorated. The class must still be in a module's `providers`. Workflows declared with `defineWorkflow()` - * are registered with `BusModule.forFeature()` instead. + * Registers a class workflow that's a Nest provider with a bus, which resolves it from Nest's container for each + * message it handles. Only classes that extend `Workflow` can be decorated. The class must still be in a module's + * `providers`. Workflows declared with `defineWorkflow()` are registered with `BusModule.forFeature()` instead. * @param options which bus to register it with * @example * ```ts diff --git a/packages/bus-nestjs/src/error/index.ts b/packages/bus-nestjs/src/error/index.ts index 03a64483..cfeebdfe 100644 --- a/packages/bus-nestjs/src/error/index.ts +++ b/packages/bus-nestjs/src/error/index.ts @@ -4,4 +4,3 @@ export * from './bus-core-version-not-supported' export * from './bus-feature-not-static' export * from './bus-not-built' export * from './bus-not-registered' -export * from './workflow-resolved-without-message' diff --git a/packages/bus-nestjs/src/error/workflow-resolved-without-message.ts b/packages/bus-nestjs/src/error/workflow-resolved-without-message.ts deleted file mode 100644 index 0d82bd9c..00000000 --- a/packages/bus-nestjs/src/error/workflow-resolved-without-message.ts +++ /dev/null @@ -1,25 +0,0 @@ -/** - * Thrown when a request-scoped class workflow, or a class workflow that depends on a request-scoped provider, can't - * be created without a message. The bus creates each class workflow once when it initializes, to read its - * `configureWorkflow()`, and there's no message then, so Nest's `REQUEST` is `undefined`. - */ -export class WorkflowResolvedWithoutMessage extends Error { - readonly help: string - - /** - * @param className the class workflow - * @param cause the error creating it threw - */ - constructor( - readonly className: string, - cause?: unknown - ) { - super( - `${className} couldn't be created without a message, which the bus does once when it initializes to read its configureWorkflow()`, - { cause } - ) - this.help = `Don't read REQUEST in the constructors of ${className} or the request-scoped providers it depends on: keep the request, and read its message and attributes when a message is handled, e.g. in a getter or the handler method.` - - Object.setPrototypeOf(this, new.target.prototype) - } -} diff --git a/packages/bus-nestjs/src/nest-container.spec.ts b/packages/bus-nestjs/src/nest-container.spec.ts index 6854c475..bcf034d9 100644 --- a/packages/bus-nestjs/src/nest-container.spec.ts +++ b/packages/bus-nestjs/src/nest-container.spec.ts @@ -4,7 +4,7 @@ import { ContainerAdapter } from '@node-ts/bus-core' import { MessageAttributes } from '@node-ts/bus-messages' import { Mock } from 'typemoq' import { BusRequest } from './bus-request' -import { BusClassNotProvided, WorkflowResolvedWithoutMessage } from './error' +import { BusClassNotProvided } from './error' import { nestContainer } from './nest-container' import { OrderPlaced } from './test/order-placed' @@ -117,12 +117,8 @@ describe('nestContainer', () => { } }) - it('should throw WorkflowResolvedWithoutMessage, naming the class, with the error as its cause', () => { - expect(error).toBeInstanceOf(WorkflowResolvedWithoutMessage) - expect((error as WorkflowResolvedWithoutMessage).className).toEqual( - 'ScopedHandler' - ) - expect((error as WorkflowResolvedWithoutMessage).cause).toBe(cause) + it('should rethrow the error as it is', () => { + expect(error).toBe(cause) }) }) diff --git a/packages/bus-nestjs/src/nest-container.ts b/packages/bus-nestjs/src/nest-container.ts index cca47e79..ddcf4a76 100644 --- a/packages/bus-nestjs/src/nest-container.ts +++ b/packages/bus-nestjs/src/nest-container.ts @@ -3,7 +3,7 @@ import { ContextId, ContextIdFactory, ModuleRef } from '@nestjs/core' import { ClassConstructor, ContainerAdapter } from '@node-ts/bus-core' import { Message, MessageAttributes } from '@node-ts/bus-messages' import { BusRequest } from './bus-request' -import { BusClassNotProvided, WorkflowResolvedWithoutMessage } from './error' +import { BusClassNotProvided } from './error' /** * Creates a `ContainerAdapter` that resolves class handlers and workflows from Nest's container. `BusModule` @@ -16,8 +16,6 @@ import { BusClassNotProvided, WorkflowResolvedWithoutMessage } from './error' * @param moduleRef a `ModuleRef` from the application, which finds providers in any of its modules * @returns the adapter, for `withContainer()` * @throws BusClassNotProvided from `get`, when the class isn't a provider in the application - * @throws WorkflowResolvedWithoutMessage from `get`, when a request-scoped class workflow can't be created without a - * message * @example * Bus.configure().withContainer(nestContainer(app.get(ModuleRef))) */ @@ -59,19 +57,11 @@ export const nestContainer = (moduleRef: ModuleRef): ContainerAdapter => { if (scope === Scope.DEFAULT) { return moduleRef.get(type, { strict: false }) } - const message = context?.message - const resolved = moduleRef.resolve( + return moduleRef.resolve( type, - scopeFor(message, context?.messageAttributes), + scopeFor(context?.message, context?.messageAttributes), { strict: false } ) - if (message !== undefined) { - return resolved - } - // Only class workflows are resolved without a message, when the bus reads their configureWorkflow() - return resolved.catch((error: unknown) => { - throw new WorkflowResolvedWithoutMessage(type.name, error) - }) } } } diff --git a/packages/bus-nestjs/src/test/workflows.ts b/packages/bus-nestjs/src/test/workflows.ts index fba661c3..a6cfb142 100644 --- a/packages/bus-nestjs/src/test/workflows.ts +++ b/packages/bus-nestjs/src/test/workflows.ts @@ -41,13 +41,15 @@ export const shippingWorkflow = (recorder: Recorder) => }) /** - * A request-scoped class workflow that reads the message from `REQUEST` in its constructor, which fails when the - * bus creates it without a message to read its `configureWorkflow()` + * A request-scoped class workflow that reads the message from `REQUEST` in its constructor, and records it */ export class RequestScopedWorkflow extends Workflow { - readonly orderId: string + private readonly orderId: string - constructor(request: BusRequest) { + constructor( + request: BusRequest, + private readonly recorder: Recorder + ) { super() this.orderId = (request.message as OrderPlaced).orderId } @@ -58,9 +60,15 @@ export class RequestScopedWorkflow extends Workflow { mapper.withState(FulfilmentState).startedBy(OrderPlaced, 'start') } - async start(): Promise> { + async start(event: OrderPlaced): Promise> { + this.recorder.record({ + by: RequestScopedWorkflow.name, + message: event, + detail: this.orderId + }) return { orderId: this.orderId } } } Injectable({ scope: Scope.REQUEST })(RequestScopedWorkflow) Inject(REQUEST)(RequestScopedWorkflow, undefined, 0) +Inject(Recorder)(RequestScopedWorkflow, undefined, 1)