Reacting to Data Changes with MongoDB Change Streams

Change streams let your application subscribe to changes in a collection, database or entire cluster. Instead of polling, you get each insert, update and delete as it happens.

Watching a collection

const stream = orders.watch(
  [{ $match: { operationType: { $in: ['insert', 'update'] } } }],
  { fullDocument: 'updateLookup' }      // include the full document on updates
);

for await (const change of stream) {
  if (change.operationType === 'insert') {
    await notifyWarehouse(change.fullDocument);
  } else if (change.updateDescription.updatedFields.status === 'shipped') {
    await sendShippingEmail(change.fullDocument);
  }
}

Resume after a restart

Each event has a resume token (change._id). Store the last processed token, and continue from there after a restart, without missing events (as long as they're still in the oplog):

const saved = await state.findOne({ _id: 'orders-stream' });
const stream = orders.watch(pipeline, {
  fullDocument: 'updateLookup',
  ...(saved && { resumeAfter: saved.token }),
});

for await (const change of stream) {
  await handle(change);
  await state.updateOne({ _id: 'orders-stream' }, { $set: { token: change._id } }, { upsert: true });
}

Use cases

• Push live updates to browsers over WebSockets.
• Keep a search index, cache or analytics store in sync.
• Trigger workflows (emails, webhooks) when data changes.
• Audit logs of who changed what.

Things to know

• Requires a replica set or sharded cluster.
• The oplog has a limited window; if your consumer is down longer than that, it can't resume and must resync.
• Processing is at-least-once when you store tokens after handling, so make handlers idempotent.
• Every open change stream uses server resources; prefer one stream per purpose with a $match filter over many streams.