feat: MongoDB Consumer based on MongoDB ChangeStream subscription

[arturwojnar]

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]

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

[oskardudycz]

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

[arturwojnar]

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]

The minimum is MongoDB 4 for now.

[oskardudycz]

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]

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

[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]

@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]

@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.

[oskardudycz]

@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]

the same as stop... :/

[arturwojnar]

it's not the case for the esdb..

[arturwojnar]

@oskardudycz