feat: MongoDB Consumer based on MongoDB ChangeStream subscription #258

arturwojnar · 2025-07-31T08:31:24Z

This PR intends to provide a consumer relying on the MongoDB Change Streams.

Example of what caused me thinking about this feature 👇

switch (type) {
  case 'B2BInstitutionUsedLicencesIncreased':
    if (state._disabled) {
      throw new InstitutionInvalidState({
        details: { currentState: 'deleted' },
        message: `Cannot use a licence of a disabled institution.`,
      })
    }

    // Hack.
    const licenceId = generateLicenceId()
    await getEventStore().appendToStream<LicenceEvent>(
      toStreamName(LICENCES_STREAM_TYPE, event.patientId.toString()),
      [
        literalObject<B2BLincenceObtained>({
          type: 'B2BLincenceObtained',
          metadata: {
            patientId: event.patientId,
            institutionId: metadata.institutionId,
            licenceId,
          },
          data: { obtainedAt: event.timestamp, duration: state.licenceDuration as Duration },
        }),
      ],
    )

    return literalObject<B2BInstitution>({
      ...state,
      usedLicences: state.usedLicences + 1,
      availableLicences: state.availableLicences - 1,
    })
}

☝️ This is clipping of a projection's evolve function. As you see, to achieve the reactiveness I did something very ugly 🤮 - I converted the evolve into an asynchronous function and on processing the B2BInstitutionUsedLicencesIncreased event, I append to a stream another event - B2BLincenceObtained.

The latter event is a consequence of the first, but it should not be implicitly and directly coupled to the first one. Especially, if that happens in the projection's internals.

So, the reactiveness is the clue of this PR.

MongoDB brings the Change Stream functionality. This is a simple pull mechanism on MongoDB's oplog.

By subscribing to the Change Stream and storing last processed message's position (here's called a token) we implement, in fact, a simple version of the outbox pattern.

Here's the fixed version:

consumer.reactor<B2BInstitutionUsedLicencesIncreased>({
  processorId: v4(),
  eachMessage: async (event) => {
    await getEventStore().appendToStream<LicenceEvent>(
      toStreamName(LICENCES_STREAM_TYPE, event.patientId.toString()),
      [
        literalObject<B2BLincenceObtained>({
          type: 'B2BLincenceObtained',
          metadata: {
            patientId: event.patientId,
            institutionId: metadata.institutionId,
            licenceId,
          },
          data: { obtainedAt: event.timestamp, duration: state.licenceDuration },
        }),
      ],
    )
  },
  connectionOptions: {
    client,
  },
});

oskardudycz · 2025-08-01T11:10:15Z

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventsConsumer.ts

+  processors?: MongoDBProcessor<MessageType>[];
+};
+
+export type EventStoreDBEventStoreConsumerConfig<


FIX: There are some leftovers from EventStoreDB inspiration in the naming, this will need to be adjusted 🙂

oskardudycz · 2025-08-01T11:12:43Z

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventsConsumer.ts

+> = {
+  consumerId?: string;
+
+  processors?: MongoDBProcessor<MessageType>[];


CONCERN: Do we need MongoDBProcessor here? The intention behind consumers is that you can plug any processors you want, so consuming messages is decoupled from handling.

No, it was my mistake. Good catch!

oskardudycz · 2025-08-01T11:13:56Z

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventsConsumer.ts

+        connectionString: string;
+        clientOptions?: MongoClientOptions;
+        client?: never;
+        onHandleStart?: (


QUESTION: Could you expand on the reasoning behind onHandleStart and onHandleEnd?

Yes. I needed tha ton some point of the development, but now I dont :) Removed.

oskardudycz · 2025-08-01T11:14:58Z

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventsConsumer.ts

+  | FullDocument<EventType, EventMetaDataType, T>
+  | UpdateDescription<ReadEvent<EventType, EventMetaDataType>>;
+
+export const mongoDBEventsConsumer = <


Suggested change

export const mongoDBEventsConsumer = <

export const mongoDBMessagesConsumer = <

oskardudycz · 2025-08-01T11:16:12Z

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventsConsumer.ts

+  const processors = options.processors ?? [];
+  const subscribe = _subscribe(
+    options.changeStreamFullDocumentPolicy,
+    client.db(),


SUGGESTION: Eventually, we should also allow passing db, right?

That happens here, right?

oskardudycz · 2025-08-01T11:21:04Z

src/packages/emmett-mongodb/src/eventStore/consumers/readProcessorCheckpoint.ts

+  options: { processorId: string; partition?: string; collectionName?: string },
+): Promise<ReadProcessorCheckpointResult> => {
+  const result = await client
+    .db()


SUGGESTION: Here, we'll also need to provide an option to pass db name through options.

oskardudycz · 2025-08-01T11:22:51Z

src/packages/emmett-mongodb/src/eventStore/consumers/subscriptions/index.ts

+  };
+};
+
+export const generateVersionPolicies = async (db: Db) => {


QUESTION: Could you explain more on why is it needed and provide some reference links for that?

It checks the MongoDB version and returns a version-dependent configuration. Hm, actually, there is a difference between versions 5 and 6. What is your minimum support version?

oskardudycz · 2025-08-01T11:30:53Z

src/packages/emmett-mongodb/src/eventStore/mongoDBEventStore.subscription.e2e.spec.ts

+          event.data.productItem.productId,
+        );
+
+        if (event.data.productItem.productId === lastProductItemId) {


SUGGESTION: You can use a built-in stop after capabilities here to make this check easier. See

emmett/src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreConsumer.handling.int.spec.ts

Line 189 in bd65823

stopAfter: (event) =>

oskardudycz · 2025-08-01T11:31:47Z

src/packages/emmett-mongodb/src/eventStore/consumers/storeProcessorCheckpoint.ts

+    }
+  | { success: false; reason: 'IGNORED' | 'MISMATCH' };
+
+export const storeProcessorCheckpoint = async <Position extends string | null>(


SUGGESTION: It'd be great to add some integration tests for that.

oskardudycz · 2025-08-01T11:33:11Z

src/packages/emmett-mongodb/src/eventStore/consumers/storeProcessorCheckpoint.ts

+  options: {
+    processorId: string;
+    version: number | undefined;
+    newPosition: null extends Position


FIX: In the end we need to support here also other types of positions. We'll need to be passing checkpoint as we're doing in ESDB. Without it, we'll be able to only use mongo processor in the MongoDB event store, and we want in the end to use it also with other event stores.

esdb supports only the bigint

export const storeProcessorCheckpoint = async <Position extends bigint | null>

Check out my changes, is it sth you were thinking of?

oskardudycz · 2025-08-01T11:36:20Z

src/packages/emmett-mongodb/src/eventStore/consumers/storeProcessorCheckpoint.ts

+      ? { success: true, newPosition: options.newPosition }
+      : {
+          success: false,
+          reason: result.matchedCount === 0 ? 'IGNORED' : 'MISMATCH',


SUGGESTION: I might be wrong, but this doesn't seem to be correct. IGNORED is meant to be used when the stored checkpoint is bigger than the one we're passing (it means that it was already handled). Checking for matched count may mean multiple reasons.

oskardudycz

@arturwojnar thanks for doing it!

I did a first round of review, I added some questions/comments/suggestions but overall, it looks like a great first step, after clarifications, adding more tests I'll be happy to merge it!

It'd be great also if you could update descriptions with the highlights of this PR.

oskardudycz · 2025-08-01T11:38:43Z

@arturwojnar could you ensure that linter, build and tests are passing? See: https://github.com/event-driven-io/emmett/actions/runs/16644021116/job/47194905100?pr=258. Thanks in advance!

alex-laycalvert

I think this is a good start based off of the other consumers. Nothing that @oskardudycz hasn't already mentioned. I think some of the boillerplate can be removed since the addition of centralized consumer types and stuff.

Probably want to add some documentation that this consumer implementation requires change streams be enabled and can't be used in single-instance environments

arturwojnar · 2025-08-15T18:30:39Z

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventsConsumer.ts

+
+        stream = subscribe<Event, CheckpointType>(startFrom);
+
+        void (async () => {


@oskardudycz I do it this way, let's say sequentially, one by one, instead relying on stream events (stream.on('data')).

It is because, as it is done in tests for three following stream appends, the value of the lastCheckpoint (processors.ts) doesn't have time to change because of lasting I/O operations. That causes that storeProcessorCheckpoint throws MISMATCH and checkpoints get no persisted.

This solution ain't perfect, though, because there is still possibility for the race because at line 293 we have asynchronous handling of a processor function.

In my Hermes repo I handle this case by implementing a queue of messages to ACK, so messages can be finished processing out of order but still they will be stored / ACK in order.

@arturwojnar check ESDB implementation, there it’s handled with custom transform and callbacks

emmett/src/packages/emmett-esdb/src/eventStore/consumers/subscriptions/index.ts

Line 135 in bd65823

async _transform(

This gives a guarantee of sequential processing and passing it to producer. That should eliminate getting out of order and race conditions. I’m considering to also apply the similar pattern inside processors.

arturwojnar · 2025-08-15T18:31:00Z

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventsConsumer.ts

+      }
+    },
+    close: async () => {
+      if (stream) {


the same as stop... :/

arturwojnar added 2 commits July 12, 2025 23:06

feat: mongodb processor and subscription

8c6af5c

feat: receiving updates on messages

bb80bf3

arturwojnar marked this pull request as draft July 31, 2025 08:31

oskardudycz reviewed Aug 1, 2025

View reviewed changes

oskardudycz requested changes Aug 1, 2025

View reviewed changes

oskardudycz added enhancement New feature or request projections mongodb consumers labels Aug 1, 2025

oskardudycz requested a review from alex-laycalvert August 1, 2025 11:39

alex-laycalvert reviewed Aug 5, 2025

View reviewed changes

arturwojnar added 8 commits August 6, 2025 16:19

refactor: removed the esdb copy-paste leftovers

bb1ebfe

fix: revert the processors type back to generic MessageProcessor

ae79330

fix: removed onHandleStart

753103c

refactor: to mongoDbEventsConsumer renamed to mongoDBMessagesConsumer

592a866

feat: databaseName as parameter for the readProcessorCheckpoint

30fa044

test: storeProcessorCheckpoint and readProcessorCheckpoint tests

2de9d86

refactor: storeProcessorCheckpoint

fc53348

chore: eslint fix

847580d

arturwojnar added 5 commits August 9, 2025 21:52

feat: handling an unknown Position

fa8cb43

feat: starting from the earliest position

4c03a71

fix: eslint all fixed

e4d5904

feat: processing messages one by one

ead380c

fix: removed incorrect change

13afb1e

arturwojnar commented Aug 15, 2025

View reviewed changes

	export const mongoDBEventsConsumer = <
	export const mongoDBMessagesConsumer = <


		stream = subscribe<Event, CheckpointType>(startFrom);

		void (async () => {

Uh oh!

feat: MongoDB Consumer based on MongoDB ChangeStream subscription #258

Are you sure you want to change the base?

feat: MongoDB Consumer based on MongoDB ChangeStream subscription #258

Uh oh!

Conversation

arturwojnar commented Jul 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

oskardudycz left a comment

Choose a reason for hiding this comment

Uh oh!

oskardudycz commented Aug 1, 2025

Uh oh!

alex-laycalvert left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

arturwojnar commented Jul 31, 2025 •

edited

Loading