Publish events using MongoDB Change Streams

[bounoable]

By default, events are not published over a bus when inserted into a store. To automatically publish inserted events, a decorator can be used:

package example

func example(store event.Store, bus event.Bus) {
  // store publishes events over the given bus after insertion into the store
  store = eventstore.WithBus(store, bus)
}

This has a flaw: when the process crashes between the insert and publish, the event will never be published, causing a "data-loss" (not really, the event was inserted into the store). If such an event would trigger a projection, the projection would likely be in an invalid state until it is manually reset by a user/developer because there is no way for a projection to know that it is missing events (not even through a projection.Progressor).

Proposal

Proposal is to create a service that uses the MongoDB Change Streams feature to tail the oplog of the event store. When an event is inserted into the database, the oplog notifies the service, which then publishes the event over the event bus.

Change Streams provide a "resume token" which are used to resume tailing the oplog after a service crash/restart. This should ensure that no events are "dropped" / all inserted events are eventually published.

[bounoable]

https://github.com/damianiandrea/mongodb-nats-connector

[gedw99]

https://github.com/damianiandrea/mongodb-nats-connector

Exactly !!

This is ready for integration

I presume we put the cdc events into nats ?

Then workers react to the events and create materialise read only data into something . Maybe mongo actually . Talk about inversion !!!