fix(@app/serverless): common circular dependencies problem in inmemory implementations addressed

cjuega · cjuega · commit 979399c1b817 · 2024-10-25T22:13:40.000+02:00
diff --git a/src/apps/example-context/serverless-backend/src/config/dependency-injection/index.ts b/src/apps/example-context/serverless-backend/src/config/dependency-injection/index.ts
@@ -1,5 +1,6 @@
 import appRegister from '@src/config/dependency-injection/app/application.di';
 import exampleAggregateRegister from '@src/config/dependency-injection/example-aggregate/application.di';
+import setupBuses from '@src/config/dependency-injection/setupBuses';
 import sharedRegister from '@src/config/dependency-injection/shared/application.di';
 import { ContainerBuilder } from 'node-dependency-injection';
 
@@ -9,4 +10,6 @@ sharedRegister(container);
 appRegister(container);
 exampleAggregateRegister(container);
 
+setupBuses(container);
+
 export default container;
diff --git a/src/apps/example-context/serverless-backend/src/config/dependency-injection/setupBuses.ts b/src/apps/example-context/serverless-backend/src/config/dependency-injection/setupBuses.ts
@@ -0,0 +1,43 @@
+import type DomainEvent from '@context/shared/domain/eventBus/domainEvent';
+import type { DomainEventSubscriber } from '@context/shared/domain/eventBus/domainEventSubscriber';
+import type CommandHandlersInformation from '@context/shared/infrastructure/commandBus/commandHandlersInformation';
+import type InMemoryCommandBus from '@context/shared/infrastructure/commandBus/inMemoryCommandBus';
+import type InMemorySyncEventBus from '@context/shared/infrastructure/eventBus/inMemorySyncEventBus';
+import type InMemoryQueryBus from '@context/shared/infrastructure/queryBus/inMemoryQueryBus';
+import type QueryHandlersInformation from '@context/shared/infrastructure/queryBus/queryHandlersInformation';
+import type { ContainerBuilder } from 'node-dependency-injection';
+
+function setupInMemoryCommandBus(container: ContainerBuilder): void {
+    const mapping = container.get<CommandHandlersInformation>('Shared.CommandHandlersInformation'),
+        commandBus = container.get<InMemoryCommandBus>('Shared.CommandBus');
+
+    commandBus.registerHandlers(mapping);
+}
+
+function setupInMemoryQueryBus(container: ContainerBuilder): void {
+    const mapping = container.get<QueryHandlersInformation>('Shared.QueryHandlersInformation'),
+        queryBus = container.get<InMemoryQueryBus>('Shared.QueryBus');
+
+    queryBus.registerHandlers(mapping);
+}
+
+function setupInMemoryEventBus(container: ContainerBuilder): void {
+    const subscriberDefinitions = container.findTaggedServiceIds('eventSubscriber'),
+        eventBus = container.get<InMemorySyncEventBus>('Shared.InMemoryEventBus'),
+        subscribers: DomainEventSubscriber<DomainEvent>[] = [];
+
+    for (const { id } of subscriberDefinitions) {
+        subscribers.push(container.get(id));
+    }
+
+    eventBus.registerSubscribers(subscribers);
+}
+
+// This functions help to break circular dependencies in the DI container. Instead of injecting the mappings directly into the buses'
+// constructors (which would attempt to instanciate handlers and would lead to a circular dependency), we register handlers and
+// subscribers after the buses have been created.
+export default function setupBuses(container: ContainerBuilder): void {
+    setupInMemoryCommandBus(container);
+    setupInMemoryQueryBus(container);
+    setupInMemoryEventBus(container);
+}
diff --git a/src/apps/example-context/serverless-backend/src/config/dependency-injection/shared/application.di.ts b/src/apps/example-context/serverless-backend/src/config/dependency-injection/shared/application.di.ts
@@ -1,4 +1,6 @@
 import DomainEventMapping from '@context/shared/domain/eventBus/domainEventMapping';
+import CommandHandlersInformation from '@context/shared/infrastructure/commandBus/commandHandlersInformation';
+import InMemoryCommandBus from '@context/shared/infrastructure/commandBus/inMemoryCommandBus';
 import CurrentTimeClock from '@context/shared/infrastructure/currentTimeClock';
 import EventBridgeClientFactory from '@context/shared/infrastructure/eventBus/eventBridge/eventBridgeClientFactory';
 import EventBridgeEventBus from '@context/shared/infrastructure/eventBus/eventBridge/eventBridgeEventBus';
@@ -9,6 +11,8 @@ import DdbOneTableClientFactory from '@context/shared/infrastructure/persistence
 import DdbOneTableDomainEventRepository from '@context/shared/infrastructure/persistence/ddbOneTable/ddbOneTableDomainEventRepository';
 import DynamodbClientFactory from '@context/shared/infrastructure/persistence/dynamodb/dynamodbClientFactory';
 import DynamodbStreamsOutboxConsumer from '@context/shared/infrastructure/persistence/dynamodb/dynamodbStreamsOutboxConsumer';
+import InMemoryQueryBus from '@context/shared/infrastructure/queryBus/inMemoryQueryBus';
+import QueryHandlersInformation from '@context/shared/infrastructure/queryBus/queryHandlersInformation';
 import config from '@src/config/config';
 import { type ContainerBuilder, Definition, Reference, TagReference } from 'node-dependency-injection';
 
@@ -85,6 +89,12 @@ const serviceName = config.get('serviceName'),
             // .addArgument(new Reference('Shared.EventBus.NoFailover'))
             .addArgument(failoverOrOutboxConfig)
             .addTag('dynamodbStreamProcessor');
+
+        container.register('Shared.QueryHandlersInformation', QueryHandlersInformation).addArgument(new TagReference('queryHandler'));
+        container.register('Shared.QueryBus', InMemoryQueryBus);
+
+        container.register('Shared.CommandHandlersInformation', CommandHandlersInformation).addArgument(new TagReference('commandHandler'));
+        container.register('Shared.CommandBus', InMemoryCommandBus);
     };
 
 export default register;
diff --git a/src/apps/example-context/serverless-backend/src/subscribers/domainEvents/registerSubscribers.ts b/src/apps/example-context/serverless-backend/src/subscribers/domainEvents/registerSubscribers.ts
diff --git a/src/apps/example-context/serverless-backend/src/subscribers/domainEvents/subscribers.ts b/src/apps/example-context/serverless-backend/src/subscribers/domainEvents/subscribers.ts
@@ -5,12 +5,11 @@ import type { DomainEventUnmarshaller } from '@context/shared/domain/eventBus/do
 import type { EventBus } from '@context/shared/domain/eventBus/eventBus';
 import type { Logger } from '@context/shared/domain/logger';
 import container from '@src/config/dependency-injection';
-import registerSubscribers from '@src/subscribers/domainEvents/registerSubscribers';
 
 const logger: Logger = container.get('Shared.Logger'),
     clock: Clock = container.get('Shared.Clock'),
     unmarshaller: DomainEventUnmarshaller = container.get('Shared.EventBus.EventMarshaller'),
-    eventBus: EventBus = registerSubscribers(container);
+    eventBus: EventBus = container.get('Shared.InMemoryEventBus');
 
 export const on: EventBridgeHandler<string, Record<string, unknown>, void> = async (event) => {
     const domainEvents = [unmarshaller.unmarshall(JSON.stringify(event.detail))];
diff --git a/src/contexts/shared/src/infrastructure/commandBus/inMemoryCommandBus.test.ts b/src/contexts/shared/src/infrastructure/commandBus/inMemoryCommandBus.test.ts
@@ -23,12 +23,23 @@ class MyCommandHandler implements CommandHandler<HandledCommand> {
 }
 
 describe('inMemoryCommandBus', () => {
+    it('throws an error when no handlers are registered', async () => {
+        expect.hasAssertions();
+
+        const unhandledCommand = new UnhandledCommand(),
+            commandBus = new InMemoryCommandBus();
+
+        await expect(commandBus.dispatch(unhandledCommand)).rejects.toThrow(Error);
+    });
+
     it('throws an error if dispatches a command without handler', async () => {
         expect.hasAssertions();
 
         const unhandledCommand = new UnhandledCommand(),
             commandHandlersInformation = new CommandHandlersInformation([]),
-            commandBus = new InMemoryCommandBus(commandHandlersInformation);
+            commandBus = new InMemoryCommandBus();
+
+        commandBus.registerHandlers(commandHandlersInformation);
 
         await expect(commandBus.dispatch(unhandledCommand)).rejects.toThrow(CommandNotRegisteredError);
     });
@@ -37,7 +48,9 @@ describe('inMemoryCommandBus', () => {
         const handledCommand = new HandledCommand(),
             myCommandHandler = new MyCommandHandler(),
             commandHandlersInformation = new CommandHandlersInformation([myCommandHandler]),
-            commandBus = new InMemoryCommandBus(commandHandlersInformation);
+            commandBus = new InMemoryCommandBus();
+
+        commandBus.registerHandlers(commandHandlersInformation);
 
         await commandBus.dispatch(handledCommand);
     });
diff --git a/src/contexts/shared/src/infrastructure/commandBus/inMemoryCommandBus.ts b/src/contexts/shared/src/infrastructure/commandBus/inMemoryCommandBus.ts
@@ -3,13 +3,17 @@ import type { CommandBus } from '@src/domain/commandBus/commandBus';
 import type CommandHandlersInformation from '@src/infrastructure/commandBus/commandHandlersInformation';
 
 export default class InMemoryCommandBus implements CommandBus {
-    private commandHandlersInformation: CommandHandlersInformation;
+    private commandHandlersInformation?: CommandHandlersInformation;
 
-    constructor(commandHandlersInformation: CommandHandlersInformation) {
+    registerHandlers(commandHandlersInformation: CommandHandlersInformation): void {
         this.commandHandlersInformation = commandHandlersInformation;
     }
 
     async dispatch(command: Command): Promise<void> {
+        if (!this.commandHandlersInformation) {
+            throw new Error('No command handlers registered');
+        }
+
         const handler = this.commandHandlersInformation.search(command);
 
         await handler.handle(command);
diff --git a/src/contexts/shared/src/infrastructure/eventBus/inMemoryAsyncEventBus.test.ts b/src/contexts/shared/src/infrastructure/eventBus/inMemoryAsyncEventBus.test.ts
@@ -45,12 +45,23 @@ class DomainEventSubscriberDummy implements DomainEventSubscriber<DummyEvent> {
 }
 
 describe('inMemoryAsyncEventBus', () => {
+    it('throws an error when no subscribers are registered', async () => {
+        expect.hasAssertions();
+
+        const event = new DummyEvent({ id: ObjectMother.uuid() }),
+            eventBus = new InMemoryAsyncEventBus();
+
+        await expect(eventBus.publish([event])).rejects.toThrow(Error);
+    });
+
     it('the subscriber should be called when the event it is subscribed to is published', async () => {
         expect.assertions(1);
 
         const event = new DummyEvent({ id: ObjectMother.uuid() }),
             subscriber = new DomainEventSubscriberDummy(),
-            eventBus = new InMemoryAsyncEventBus([subscriber]);
+            eventBus = new InMemoryAsyncEventBus();
+
+        eventBus.registerSubscribers([subscriber]);
 
         subscriber.setExpectation((actual: DummyEvent) => {
             expect(actual).toStrictEqual(event);
diff --git a/src/contexts/shared/src/infrastructure/eventBus/inMemoryAsyncEventBus.ts b/src/contexts/shared/src/infrastructure/eventBus/inMemoryAsyncEventBus.ts
@@ -4,13 +4,17 @@ import type { EventBus } from '@src/domain/eventBus/eventBus';
 import EventEmitterBus from '@src/infrastructure/eventBus/eventEmitterBus';
 
 export default class InMemoryAsyncEventBus implements EventBus {
-    private readonly bus: EventEmitterBus;
+    private bus?: EventEmitterBus;
 
-    constructor(subscribers: DomainEventSubscriber<DomainEvent>[]) {
+    registerSubscribers(subscribers: DomainEventSubscriber<DomainEvent>[]): void {
         this.bus = new EventEmitterBus(subscribers);
     }
 
     publish(events: DomainEvent[]): Promise<void> {
+        if (!this.bus) {
+            return Promise.reject(new Error('No event subscribers registered'));
+        }
+
         this.bus.publish(events);
         return Promise.resolve();
     }
diff --git a/src/contexts/shared/src/infrastructure/eventBus/inMemorySyncEventBus.test.ts b/src/contexts/shared/src/infrastructure/eventBus/inMemorySyncEventBus.test.ts
@@ -50,7 +50,9 @@ describe('inMemorySyncEventBus', () => {
 
         const event = new DummyEvent({ id: ObjectMother.uuid() }),
             subscriber = new DomainEventSubscriberDummy(),
-            eventBus = new InMemorySyncEventBus([subscriber]);
+            eventBus = new InMemorySyncEventBus();
+
+        eventBus.registerSubscribers([subscriber]);
 
         subscriber.setExpectation((actual: DummyEvent) => {
             expect(actual).toStrictEqual(event);
diff --git a/src/contexts/shared/src/infrastructure/eventBus/inMemorySyncEventBus.ts b/src/contexts/shared/src/infrastructure/eventBus/inMemorySyncEventBus.ts
@@ -6,10 +6,6 @@ export default class InMemorySyncEventBus implements EventBus {
     // biome-ignore lint/complexity/noBannedTypes: <explanation>
     private readonly subscriptions: Map<string, Function[]> = new Map();
 
-    constructor(subscribers: DomainEventSubscriber<DomainEvent>[]) {
-        this.registerSubscribers(subscribers);
-    }
-
     async publish(events: DomainEvent[]): Promise<void> {
         const executions: Promise<void>[] = [];
 
@@ -24,7 +20,7 @@ export default class InMemorySyncEventBus implements EventBus {
         await Promise.all(executions);
     }
 
-    private registerSubscribers(subscribers: DomainEventSubscriber<DomainEvent>[]): void {
+    registerSubscribers(subscribers: DomainEventSubscriber<DomainEvent>[]): void {
         for (const subscriber of subscribers) {
             for (const e of subscriber.subscribedTo()) {
                 this.subscribe(e.eventName, subscriber);
diff --git a/src/contexts/shared/src/infrastructure/eventBus/kafka/kafkaEventBusConsumer.ts b/src/contexts/shared/src/infrastructure/eventBus/kafka/kafkaEventBusConsumer.ts
@@ -21,7 +21,8 @@ export default class KafkaEventBusConsumer implements EventBusConsumer {
         private readonly unmarshaller: DomainEventUnmarshaller,
         private readonly config: KafkaConfig
     ) {
-        this.internalBus = new InMemorySyncEventBus(subscribers);
+        this.internalBus = new InMemorySyncEventBus();
+        this.internalBus.registerSubscribers(subscribers);
         this.consumer = kafka.consumer({ groupId: this.config.groupId });
 
         this.removeDisconnectListener = this.consumer.on(this.consumer.events.DISCONNECT, () => {
diff --git a/src/contexts/shared/src/infrastructure/queryBus/inMemoryQueryBus.test.ts b/src/contexts/shared/src/infrastructure/queryBus/inMemoryQueryBus.test.ts
@@ -26,12 +26,23 @@ class MyQueryHandler implements QueryHandler<Query, Response> {
 }
 
 describe('inMemoryQueryBus', () => {
+    it('throws an error when no handlers are registered', async () => {
+        expect.hasAssertions();
+
+        const unhandledQuery = new UnhandledQuery(),
+            queryBus = new InMemoryQueryBus();
+
+        await expect(queryBus.ask(unhandledQuery)).rejects.toThrow(Error);
+    });
+
     it('throws an error if dispatches a query without handler', async () => {
         expect.hasAssertions();
 
         const unhandledQuery = new UnhandledQuery(),
             queryHandlersInformation = new QueryHandlersInformation([]),
-            queryBus = new InMemoryQueryBus(queryHandlersInformation);
+            queryBus = new InMemoryQueryBus();
+
+        queryBus.registerHandlers(queryHandlersInformation);
 
         await expect(queryBus.ask(unhandledQuery)).rejects.toThrow(QueryNotRegisteredError);
     });
@@ -40,7 +51,9 @@ describe('inMemoryQueryBus', () => {
         const handledQuery = new HandledQuery(),
             myQueryHandler = new MyQueryHandler(),
             queryHandlersInformation = new QueryHandlersInformation([myQueryHandler]),
-            queryBus = new InMemoryQueryBus(queryHandlersInformation);
+            queryBus = new InMemoryQueryBus();
+
+        queryBus.registerHandlers(queryHandlersInformation);
 
         await queryBus.ask(handledQuery);
     });
diff --git a/src/contexts/shared/src/infrastructure/queryBus/inMemoryQueryBus.ts b/src/contexts/shared/src/infrastructure/queryBus/inMemoryQueryBus.ts
@@ -4,13 +4,17 @@ import type { Response } from '@src/domain/queryBus/response';
 import type QueryHandlersInformation from '@src/infrastructure/queryBus/queryHandlersInformation';
 
 export default class InMemoryQueryBus implements QueryBus {
-    private queryHandlersInformation: QueryHandlersInformation;
+    private queryHandlersInformation?: QueryHandlersInformation;
 
-    constructor(queryHandlersInformation: QueryHandlersInformation) {
+    registerHandlers(queryHandlersInformation: QueryHandlersInformation): void {
         this.queryHandlersInformation = queryHandlersInformation;
     }
 
     async ask<R extends Response>(query: Query): Promise<R> {
+        if (!this.queryHandlersInformation) {
+            throw new Error('No query handlers registered');
+        }
+
         const handler = this.queryHandlersInformation.search(query),
             response = (await handler.handle(query)) as R;
 
