feat(outbox): handle concurrency on publishEvents using publishEventWithConcurrencyControl method on every event

Author: NeverBreakSpaghetti

packages/ddd-toolkit/src/outbox/mongo-outbox.spec.ts

@@ -1,7 +1,7 @@
 import { MongoMemoryReplSet } from 'mongodb-memory-server';
 import { MongoClient, ObjectId } from 'mongodb';
 import { MongoOutbox } from './mongo-outbox';
-import { Event } from '../event-bus/event';
+import { Event } from '../event-bus';
 import { waitFor } from '../utils';
 
 class FooEvent extends Event<{ foo: string }> {
@@ -89,14 +89,15 @@ describe('Mongo outbox', () => {
             });
 
             describe('When publish', () => {
-                it('should call publishEventsFn once', async () => {
+                it('should call publishEventsFn for every event', async () => {
                     await outbox.publishEvents(ids);
-                    expect(PublishEventsFnMock).toBeCalled();
+                    expect(PublishEventsFnMock).toHaveBeenCalledTimes(ids.length);
                 });
 
-                it('should pass both events to publishEventsFn', async () => {
+                it('should pass one event at time to publishEventsFn', async () => {
                     await outbox.publishEvents(ids);
-                    expect(PublishEventsFnMock).toBeCalledWith(events);
+                    expect(PublishEventsFnMock).toHaveBeenCalledWith([events[0]]);
+                    expect(PublishEventsFnMock).toHaveBeenCalledWith([events[1]]);
                 });
 
                 it('should update the status of the events to published', async () => {

packages/ddd-toolkit/src/outbox/mongo-outbox.ts

@@ -1,5 +1,5 @@
 import { ClientSession, Collection, MongoClient, ObjectId } from 'mongodb';
-import { IEvent } from '../event-bus/event-bus.interface';
+import { IEvent } from '../event-bus';
 import { ILogger } from '../logger';
 import { inspect } from 'util';
 import { difference, intersection } from 'lodash';
@@ -57,42 +57,7 @@ export class MongoOutbox implements IOutbox, IInit, ITerminate {
     }
 
     public async publishEvents(scheduledEventsIds: string[]): Promise<void> {
-        const session = this.mongoClient.startSession();
-        try {
-            await session.withTransaction(async () => {
-                const outboxModels = await this.outboxCollection
-                    .find(
-                        {
-                            _id: { $in: scheduledEventsIds.map((id) => new ObjectId(id)) },
-                            status: 'scheduled',
-                        },
-                        { session },
-                    )
-                    .toArray();
-                if (!outboxModels.length) return;
-                const events = outboxModels.map((model) => model.event);
-                await this.publishEventsFn(events);
-                const publishedIds = outboxModels.map((model) => model._id);
-                await this.outboxCollection.updateMany(
-                    {
-                        _id: { $in: publishedIds },
-                        status: 'scheduled',
-                    },
-                    {
-                        $set: {
-                            status: 'published',
-                            publishedAt: new Date(),
-                        },
-                    },
-                    { session },
-                );
-                this.logger.debug(`Published events ${publishedIds.join(', ')}`);
-            });
-        } catch (e) {
-            this.logger.warn(`Failed to publish events ${scheduledEventsIds.join(', ')}. ${inspect(e)}`);
-        } finally {
-            await session.endSession();
-        }
+        await Promise.all(scheduledEventsIds.map((eventId) => this.publishEventWithConcurrencyControl(eventId)));
     }
 
     //FROM https://github.com/gpad/ms-practical-ws/blob/main/src/infra/outbox_pattern.ts
@@ -156,6 +121,8 @@ export class MongoOutbox implements IOutbox, IInit, ITerminate {
                     { session },
                 );
             });
+        } catch (e) {
+            this.logger.warn(`Failed to publish event ${eventId}. ${inspect(e)}`);
         } finally {
             await session.endSession();
         }