feat: add command interceptors

package.json

@@ -25,8 +25,9 @@
     "@commitlint/config-angular": "18.6.0",
     "@nestjs/common": "10.3.1",
     "@nestjs/core": "10.3.1",
-    "@types/node": "20.11.7",
+    "@nestjs/testing": "^10.3.1",
     "@types/jest": "29.5.11",
+    "@types/node": "20.11.7",
     "@typescript-eslint/eslint-plugin": "6.19.1",
     "@typescript-eslint/parser": "6.19.1",
     "eslint": "8.56.0",

src/command-bus.ts

@@ -16,6 +16,7 @@ import {
   ICommandPublisher,
 } from './interfaces/index';
 import { ObservableBus } from './utils/observable-bus';
+import { CommandInterceptionExecutor } from './command-interception-executor';
 
 export type CommandHandlerType = Type<ICommandHandler<ICommand>>;
 
@@ -27,7 +28,10 @@ export class CommandBus<CommandBase extends ICommand = ICommand>
   private handlers = new Map<string, ICommandHandler<CommandBase>>();
   private _publisher: ICommandPublisher<CommandBase>;
 
-  constructor(private readonly moduleRef: ModuleRef) {
+  constructor(
+    private readonly moduleRef: ModuleRef,
+    private readonly commandInterceptor: CommandInterceptionExecutor,
+  ) {
     super();
     this.useDefaultPublisher();
   }
@@ -61,8 +65,13 @@ export class CommandBus<CommandBase extends ICommand = ICommand>
       const commandName = this.getCommandName(command);
       throw new CommandHandlerNotFoundException(commandName);
     }
-    this._publisher.publish(command);
-    return handler.execute(command);
+
+    const next = (): Promise<R> => {
+      this._publisher.publish(command);
+      return handler.execute(command);
+    };
+
+    return this.commandInterceptor.intercept(command, next);
   }
 
   bind<T extends CommandBase>(handler: ICommandHandler<T>, id: string) {

src/command-interception-executor.ts

@@ -0,0 +1,48 @@
+import { Injectable, Type } from '@nestjs/common';
+import { ICommand, ICommandInterceptor } from './interfaces';
+import { ModuleRef } from '@nestjs/core';
+
+export type CommandInterceptorType = Type<ICommandInterceptor>;
+
+@Injectable()
+export class CommandInterceptionExecutor {
+  private interceptors: ICommandInterceptor[] = [];
+
+  constructor(private readonly moduleRef: ModuleRef) {}
+
+  intercept<T extends ICommand, R>(
+    command: T,
+    next: () => Promise<R>,
+  ): Promise<R> {
+    if (this.interceptors.length === 0) {
+      return next();
+    }
+
+    const nextFn = async (i = 0): Promise<R> => {
+      if (i >= this.interceptors.length) {
+        return next();
+      }
+
+      const handler = () => nextFn(i + 1);
+      return this.interceptors[i].intercept(command, handler);
+    };
+
+    return nextFn();
+  }
+
+  register(interceptors: CommandInterceptorType[] = []) {
+    interceptors.forEach((interceptor) =>
+      this.registerInterceptor(interceptor),
+    );
+  }
+
+  private registerInterceptor(interceptor: CommandInterceptorType) {
+    const instance = this.moduleRef.get(interceptor, { strict: false });
+
+    if (!instance) {
+      return;
+    }
+
+    this.interceptors.push(instance);
+  }
+}

src/command-interception.spec.ts

@@ -0,0 +1,140 @@
+import { Test } from '@nestjs/testing';
+import {
+  CqrsModule,
+  CommandBus,
+  ICommandHandler,
+  ICommand,
+  CommandHandler,
+  CommandInterceptor,
+  ICommandInterceptor,
+} from './index';
+import { Type } from '@nestjs/common';
+
+class TestCommand implements ICommand {}
+
+@CommandHandler(TestCommand)
+class TestCommandHandler implements ICommandHandler {
+  execute(): Promise<any> {
+    return Promise.resolve(undefined);
+  }
+}
+
+@CommandInterceptor()
+class FirstCommandInterceptor implements ICommandInterceptor {
+  intercept(_: unknown, next: () => Promise<unknown>) {
+    return next();
+  }
+}
+
+@CommandInterceptor()
+class SecondCommandInterceptor implements ICommandInterceptor {
+  intercept(_: unknown, next: () => Promise<unknown>) {
+    return next();
+  }
+}
+
+describe('Command interception', () => {
+  const bootstrap = async (
+    ...interceptors: Type<ICommandInterceptor>[]
+  ): Promise<{
+    commandBus: CommandBus;
+    commandHandler: TestCommandHandler;
+    interceptors: ICommandInterceptor[];
+  }> => {
+    const moduleRef = await Test.createTestingModule({
+      providers: [TestCommandHandler, ...interceptors],
+      imports: [CqrsModule],
+    }).compile();
+    await moduleRef.init();
+
+    return {
+      commandBus: moduleRef.get(CommandBus),
+      commandHandler: moduleRef.get(TestCommandHandler),
+      interceptors: interceptors.map((interceptor) =>
+        moduleRef.get(interceptor),
+      ),
+    };
+  };
+
+  it('should invoke command handler', async () => {
+    const { commandBus, commandHandler } = await bootstrap(
+      FirstCommandInterceptor,
+      SecondCommandInterceptor,
+    );
+
+    const fakeResult = {};
+    const commandExecuteSpy = jest
+      .spyOn(commandHandler, 'execute')
+      .mockImplementation(() => Promise.resolve(fakeResult));
+
+    const command = new TestCommand();
+    const executionResult = await commandBus.execute(command);
+
+    expect(commandExecuteSpy).toHaveBeenCalledWith(command);
+    expect(executionResult).toEqual(fakeResult);
+  });
+
+  it('should invoke every interceptor', async () => {
+    const {
+      commandBus,
+      interceptors: [firstCommandInterceptor, secondCommandInterceptor],
+    } = await bootstrap(FirstCommandInterceptor, SecondCommandInterceptor);
+
+    const firstHandlerInterceptSpy = jest.spyOn(
+      firstCommandInterceptor,
+      'intercept',
+    );
+    const secondHandlerInterceptSpy = jest.spyOn(
+      secondCommandInterceptor,
+      'intercept',
+    );
+
+    const command = new TestCommand();
+    await commandBus.execute(command);
+
+    expect(firstHandlerInterceptSpy).toHaveBeenCalledWith(
+      command,
+      expect.anything(),
+    );
+    expect(secondHandlerInterceptSpy).toHaveBeenCalledWith(
+      command,
+      expect.anything(),
+    );
+  });
+
+  it('should allow modification of a command', async () => {
+    const {
+      commandBus,
+      interceptors: [commandInterceptor],
+    } = await bootstrap(FirstCommandInterceptor);
+
+    const fakeResult = {};
+    jest
+      .spyOn(commandInterceptor, 'intercept')
+      .mockImplementation(() => Promise.resolve(fakeResult));
+
+    const executionResult = commandBus.execute(new TestCommand());
+    await expect(executionResult).resolves.toEqual(fakeResult);
+  });
+
+  it('should propagate errors and stop execution', async () => {
+    const {
+      commandBus,
+      interceptors: [firstCommandInterceptor, secondCommandInterceptor],
+    } = await bootstrap(FirstCommandInterceptor, SecondCommandInterceptor);
+
+    const fakeError = new Error('FAKE_ERROR');
+    jest
+      .spyOn(firstCommandInterceptor, 'intercept')
+      .mockImplementation(() => Promise.reject(fakeError));
+    const secondInterceptorInterceptSpy = jest.spyOn(
+      secondCommandInterceptor,
+      'intercept',
+    );
+
+    await expect(commandBus.execute(new TestCommand())).rejects.toEqual(
+      fakeError,
+    );
+    expect(secondInterceptorInterceptSpy).not.toHaveBeenCalled();
+  });
+});

src/cqrs.module.ts

@@ -6,6 +6,7 @@ import { IEvent } from './interfaces';
 import { QueryBus } from './query-bus';
 import { ExplorerService } from './services/explorer.service';
 import { UnhandledExceptionBus } from './unhandled-exception-bus';
+import { CommandInterceptionExecutor } from './command-interception-executor';
 
 @Module({
   providers: [
@@ -15,6 +16,7 @@ import { UnhandledExceptionBus } from './unhandled-exception-bus';
     UnhandledExceptionBus,
     EventPublisher,
     ExplorerService,
+    CommandInterceptionExecutor,
   ],
   exports: [
     CommandBus,
@@ -43,14 +45,17 @@ export class CqrsModule<EventBase extends IEvent = IEvent>
     private readonly eventBus: EventBus<EventBase>,
     private readonly commandBus: CommandBus,
     private readonly queryBus: QueryBus,
+    private readonly commandInterceptor: CommandInterceptionExecutor,
   ) {}
 
   onApplicationBootstrap() {
-    const { events, queries, sagas, commands } = this.explorerService.explore();
+    const { events, queries, sagas, commands, interceptors } =
+      this.explorerService.explore();
 
     this.eventBus.register(events);
     this.commandBus.register(commands);
     this.queryBus.register(queries);
     this.eventBus.registerSagas(sagas);
+    this.commandInterceptor.register(interceptors);
   }
 }

src/decorators/command-interceptor.decorator.ts

@@ -0,0 +1,15 @@
+import 'reflect-metadata';
+import { COMMAND_INTERCEPTOR_METADATA } from './constants';
+
+/**
+ * Decorator that marks a class as a Nest command interceptor. A command interceptor
+ * intercepts commands (actions) executed by your application code and allows you to implement
+ * cross-cutting concerns.
+ *
+ * The decorated class must implement the `ICommandInterceptor` interface.
+ */
+export const CommandInterceptor = (): ClassDecorator => {
+  return (target: object) => {
+    Reflect.defineMetadata(COMMAND_INTERCEPTOR_METADATA, {}, target);
+  };
+};

src/decorators/constants.ts

@@ -5,3 +5,4 @@ export const QUERY_HANDLER_METADATA = '__queryHandler__';
 export const EVENT_METADATA = '__event__';
 export const EVENTS_HANDLER_METADATA = '__eventsHandler__';
 export const SAGA_METADATA = '__saga__';
+export const COMMAND_INTERCEPTOR_METADATA = '__commandInterceptor__';

src/decorators/index.ts

@@ -2,3 +2,4 @@ export * from './command-handler.decorator';
 export * from './events-handler.decorator';
 export * from './query-handler.decorator';
 export * from './saga.decorator';
+export * from './command-interceptor.decorator';

src/interfaces/commands/command-interceptor.interface.ts

@@ -0,0 +1,15 @@
+import { ICommand } from './command.interface';
+
+/**
+ * Interface describing implementation of a command interceptor
+ *
+ * @publicApi
+ */
+export interface ICommandInterceptor<T extends ICommand = any, R = any> {
+  /**
+   * Method to implement a custom command interceptor.
+   * @param command the command to execute.
+   * @param next a reference to the function, which provides access to the command handler
+   */
+  intercept(command: T, next: () => Promise<R>): Promise<R>;
+}

src/interfaces/cqrs-options.interface.ts

@@ -2,10 +2,12 @@ import { Type } from '@nestjs/common';
 import { ICommandHandler } from './commands/command-handler.interface';
 import { IEventHandler } from './events/event-handler.interface';
 import { IQueryHandler } from './queries/query-handler.interface';
+import { ICommandInterceptor } from './commands/command-interceptor.interface';
 
 export interface CqrsOptions {
   events?: Type<IEventHandler>[];
   queries?: Type<IQueryHandler>[];
   commands?: Type<ICommandHandler>[];
   sagas?: Type<any>[];
+  interceptors?: Type<ICommandInterceptor>[];
 }

src/interfaces/index.ts

@@ -2,6 +2,7 @@ export * from './commands/command-bus.interface';
 export * from './commands/command-handler.interface';
 export * from './commands/command-publisher.interface';
 export * from './commands/command.interface';
+export * from './commands/command-interceptor.interface';
 export * from './events/event-bus.interface';
 export * from './events/event-handler.interface';
 export * from './events/event-publisher.interface';