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

@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

[oskardudycz]

@arturwojnar, I added some additional fixes around starting/stopping/restarting, in general resilience alignment with other consumers.

I'll also need to check why some compatibility tests are not passing (currently commented out).

I also changed nested Mongo Resume Token to URN-based string formatted text, and added message position to it (as there can be more than one event appended in one changeset). I will generalise that to assume that checkpoint can be number, bigint or string, but in a separate PR.

Nevertheless, great work, thank you for doing it, and I appreciate your patience waiting for my final review. 👏 🥳