Robust MongoDB Change Stream implementation

#mongodb #robust #redis

Change streams allow applications to access real-time data changes without the prior complexity and risk of manually tailing theoplog.

Change streams enable you to update other data sources like a search index in elastic search, or publish events to subscribers.

A naive implementation would listen to some collections, but won't handle the case your service that is listening crashes or has some downtime when doing a rollout deployment.

This article provides a more robust solution that will avoid making you lose any change stream event, even when your service crashes or get some downtime.

Robust Change Streams using a Resume Token

conststream=model.watch([],{fullDocument:'updateLookup',});stream.on('change',(data)=>{})

A stream will emit a ChangeStreamData for each data change in your collection.

For each data change it will also provide aresumeToken, that will let you resume the processing from this point in time.

AresumeToken is like a cursor in cursor-based pagination.

The implementation to use a resume token is easy as that:

constchangeStreamListen=async<Textendsany>(model:Model<T>,fn:(data:ChangeStreamData<T>)=>void,)=>{constname=model.collection.name;constkey=`resumetoken:${name}`;constresumeToken=awaitRedisCache.get(key);constgetResumeOptions=()=>{if(resumeToken){return{resumeAfter:resumeToken,};}return{};};conststream=model.watch([],{fullDocument:'updateLookup',...getResumeOptions(),});stream.on('change',changeStreamMiddleware(name,fn)).on('error',(err)=>{// eslint-disable-next-lineconsole.log('change error:',err);Sentry.setExtra('error',err);Sentry.captureException(err);});};

This implementation saves the resume token for a given collection in Redis, and tries to resume from it if available.

In Summary

Change Streams can simplify a lot your software architecture.
Providing a robust implementation of it is important to make sure you won't lose any data change.

References

Woovi
Woovi is a Startup that enables shoppers to pay as they like. Woovi provides instant payment solutions for merchants to accept orders to make this possible.

If you want to work with us, we arehiring!