Partial #169 - MongoDB EventStore onAfterCommit hook

src/packages/emmett-mongodb/src/eventStore/mongoDBEventStore.ts

@@ -1,7 +1,8 @@
 import {
-  assertExpectedVersionMatchesCurrent,
   ExpectedVersionConflictError,
+  assertExpectedVersionMatchesCurrent,
   filterProjections,
+  tryPublishMessagesAfterCommit,
   type AggregateStreamOptions,
   type AggregateStreamResult,
   type AppendToStreamOptions,
@@ -14,6 +15,7 @@ import {
   type ReadEventMetadataWithoutGlobalPosition,
   type ReadStreamOptions,
   type ReadStreamResult,
+  type DefaultEventStoreOptions,
 } from '@event-driven-io/emmett';
 import {
   MongoClient,
@@ -178,7 +180,8 @@ export type MongoDBEventStoreOptions = {
     MongoDBProjectionInlineHandlerContext
   >[];
   storage?: MongoDBEventStoreStorageOptions;
-} & MongoDBEventStoreConnectionOptions;
+} & MongoDBEventStoreConnectionOptions &
+  DefaultEventStoreOptions<MongoDBEventStore>;
 
 export type MongoDBEventStore = EventStore<MongoDBReadEventMetadata> & {
   projections: ProjectionQueries<StreamType>;
@@ -189,13 +192,15 @@ export type MongoDBEventStore = EventStore<MongoDBReadEventMetadata> & {
 
 class MongoDBEventStoreImplementation implements MongoDBEventStore, Closeable {
   private readonly client: MongoClient;
-  private shouldManageClientLifetime: boolean;
   private readonly inlineProjections: MongoDBInlineProjectionDefinition[];
+  private shouldManageClientLifetime: boolean;
   private isClosed: boolean = false;
-  public projections: ProjectionQueries<StreamType>;
   private storage: MongoDBEventStoreStorage;
+  private options: MongoDBEventStoreOptions;
+  public projections: ProjectionQueries<StreamType>;
 
   constructor(options: MongoDBEventStoreOptions) {
+    this.options = options;
     this.client =
       'client' in options && options.client
         ? options.client
@@ -388,6 +393,15 @@ class MongoDBEventStoreImplementation implements MongoDBEventStore, Closeable {
       );
     }
 
+    await tryPublishMessagesAfterCommit<MongoDBEventStore>(
+      // @ts-expect-error Issues with `globalPosition` not being present causing the type for metadata to expect `never`
+      eventsToAppend,
+      this.options.hooks,
+      // {
+      // TODO: same context as InlineProjectionHandlerContext for mongodb?
+      // },
+    );
+
     return {
       nextExpectedStreamVersion:
         currentStreamVersion + BigInt(eventsToAppend.length),

src/packages/emmett-mongodb/src/eventStore/mongoDBEventstore.onAfterCommit.e2e.spec.ts

@@ -0,0 +1,153 @@
+import { after, before, describe, it } from 'node:test';
+import { v7 as uuid } from 'uuid';
+import { type Event, assertEqual } from '@event-driven-io/emmett';
+import {
+  getMongoDBEventStore,
+  type MongoDBReadEvent,
+} from './mongoDBEventStore';
+import {
+  MongoDBContainer,
+  StartedMongoDBContainer,
+} from '@testcontainers/mongodb';
+import { MongoClient } from 'mongodb';
+
+type TestEvent = Event<'test', { counter: number }, { some: boolean }>;
+
+void describe('MongoDBEventStore onAfterCommit', () => {
+  let mongodb: StartedMongoDBContainer;
+  let client: MongoClient;
+
+  before(async () => {
+    mongodb = await new MongoDBContainer().start();
+    client = new MongoClient(mongodb.getConnectionString(), {
+      directConnection: true,
+    });
+
+    await client.connect();
+  });
+
+  after(async () => {
+    try {
+      await client.close();
+      await mongodb.stop();
+    } catch (error) {
+      console.log(error);
+    }
+  });
+
+  void it('calls onAfterCommit hook after events append', async () => {
+    // Given
+    const appendedEvents: MongoDBReadEvent[] = [];
+    const eventStore = getMongoDBEventStore({
+      client,
+      hooks: {
+        onAfterCommit: (events) => {
+          appendedEvents.push(...events);
+        },
+      },
+    });
+    const streamName = `test:${uuid()}`;
+    let counter = 0;
+    const events: TestEvent[] = [
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: true },
+      },
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: false },
+      },
+    ];
+
+    // When
+    await eventStore.appendToStream(streamName, events);
+
+    // Then
+    assertEqual(2, appendedEvents.length);
+  });
+
+  void it('calls onAfterCommit hook exactly once for each events append', async () => {
+    // Given
+    const appendedEvents: MongoDBReadEvent[] = [];
+    const eventStore = getMongoDBEventStore({
+      client,
+      hooks: {
+        onAfterCommit: (events) => {
+          appendedEvents.push(...events);
+        },
+      },
+    });
+    const streamName = `test:${uuid()}`;
+    let counter = 0;
+    const events: TestEvent[] = [
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: true },
+      },
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: false },
+      },
+    ];
+    const nextEvents: TestEvent[] = [
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: true },
+      },
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: false },
+      },
+    ];
+
+    // When
+    await eventStore.appendToStream(streamName, events);
+    await eventStore.appendToStream(streamName, nextEvents);
+
+    // Then
+    assertEqual(4, appendedEvents.length);
+  });
+
+  void it('silently fails when onAfterCommit hook failed but still keeps events', async () => {
+    // Given
+    const appendedEvents: MongoDBReadEvent[] = [];
+    const eventStore = getMongoDBEventStore({
+      client,
+      hooks: {
+        onAfterCommit: (events) => {
+          appendedEvents.push(...events);
+          throw new Error('onAfterCommit failed!');
+        },
+      },
+    });
+
+    const streamName = `test:${uuid()}`;
+    let counter = 0;
+    const events: TestEvent[] = [
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: true },
+      },
+      {
+        type: 'test',
+        data: { counter: ++counter },
+        metadata: { some: false },
+      },
+    ];
+
+    // When
+    await eventStore.appendToStream(streamName, events);
+
+    // Then
+    assertEqual(2, appendedEvents.length);
+    const { events: eventsInStore } = await eventStore.readStream(streamName);
+    assertEqual(2, eventsInStore.length);
+  });
+});