MongoDB Change Streams & Triggers — reactive, reliable DB-driven events

Modern apps increasingly react to data changes in real time: notify users when orders ship, denormalize data into read models, invalidate caches, or feed analytics pipelines. MongoDB Change Streams give you a supported, server-side way to subscribe to and react to changes in collections, databases, or an entire deployment — without tailing the oplog yourself. They're the foundation for event-driven features and (when hosted in Atlas) the basis for serverless Atlas Triggers that run functions in response to those events.

What a Change Stream is (short version)

A change stream is a cursor that yields change-event documents describing operations that modify data (inserts, updates, replaces, deletes, etc.). Because the server exposes change streams as part of the aggregation framework, you can filter and transform events with a pipeline before your app receives them. Change streams are resumable using resume tokens, letting your consumer pick up where it left off after a restart or transient failure.

Important deployment note: change streams depend on MongoDB's replication (oplog), so they are available on replica sets and sharded clusters (not on plain standalone instances). Some storage engine and topology constraints apply (see official docs for availability details).

Key concepts

• Change event: the document emitted for a single operation (insert/update/delete/replace/rename/ invalidate, etc.). It includes fields like operationType, ns (namespace), documentKey, fullDocument (optionally), updateDescription, and an _id resume token.
• Resume token: a server-provided token (in the _id field of events) you can persist to resume a stream from a particular point.
• Aggregation pipeline filtering: you can pass a pipeline to the change stream to limit events server-side (e.g., only operationType: "update" or updates to a particular field).
• Resumability options: resumeAfter, startAfter, and startAtOperationTime let you pick an entry point for the stream.
• fullDocument option: e.g., fullDocument: "updateLookup" returns the post-image of modified documents for update events.

When to use change streams (example patterns)

• Real-time notifications: send emails/push when order.status changes to shipped.
• Cache invalidation: remove/update entries in Redis after writes.
• Denormalization: update summary/lookup collections when source data changes (e.g., maintain counters).
• ETL / streaming: push change events into a data pipeline (Kafka, Kinesis, BigQuery, etc.).
• Cross-service coordination: trigger serverless workflows, lambdas, or Atlas Functions in response to DB changes.

Example 1 — Node.js (native MongoDB driver) — basic watcher + resume tokens

This example shows collection.watch() with an aggregation-style pipeline that only forwards update events where the status field changed. It persists resume tokens in a simple store (here: a collection called changestream_resume) so you can resume after restarts.

// watcher.js
const { MongoClient } = require('mongodb');

const uri = process.env.MONGODB_URI || 'mongodb://localhost:27017/?replicaSet=rs0';
const client = new MongoClient(uri, { useNewUrlParser: true, useUnifiedTopology: true });

async function saveResumeToken(name, token) {
  const coll = client.db('admin').collection('changestream_resume');
  await coll.updateOne({ _id: name }, { $set: { token } }, { upsert: true });
}

async function getResumeToken(name) {
  const coll = client.db('admin').collection('changestream_resume');
  const doc = await coll.findOne({ _id: name });
  return doc ? doc.token : null;
}

async function run() {
  await client.connect();
  const db = client.db('shop');
  const orders = db.collection('orders');

  const resumeToken = await getResumeToken('orders-watcher');

  // Server-side filter: only update events where updateDescription.modifiedFields.status exists
  const pipeline = [
    { $match: {
        operationType: 'update',
        'updateDescription.updatedFields.status': { $exists: true }
    }}
  ];

  const changeStream = orders.watch(pipeline, { fullDocument: 'updateLookup', resumeAfter: resumeToken });

  changeStream.on('change', async (change) => {
    try {
      console.log('Got change:', change);
      // Example: send notification when status becomes 'shipped'
      const full = change.fullDocument;
      if (full && full.status === 'shipped') {
        // call notification service...
        console.log(`Order ${full._id} shipped — notify user ${full.userId}`);
      }

      // persist resume token
      await saveResumeToken('orders-watcher', change._id);
    } catch (err) {
      console.error('Error processing change', err);
      // production: instrument, backoff if needed
    }
  });

  changeStream.on('error', (err) => {
    console.error('Change stream error', err);
    // in many cases you may want to recreate the stream with exponential backoff
  });
}

run().catch(console.dir);

Notes:

• Use resumeAfter only if you have a token and you want to resume after it; startAtOperationTime is another option if you prefer timestamps. resumeAfter / startAfter / startAtOperationTime are mutually exclusive.

Example 2 — Node.js with async iterator (modern approach)

async function watchWithAsyncIterator() {
  await client.connect();
  const coll = client.db('shop').collection('products');
  const pipeline = [{ $match: { 'fullDocument.inventory': { $lte: 10 } } }];

  const changeStream = coll.watch(pipeline, { fullDocument: 'updateLookup' });

  try {
    for await (const change of changeStream) {
      console.log('Event:', change.operationType, change.documentKey);
      // handle change...
      // store change._id as resume token if desired
    }
  } finally {
    await changeStream.close();
  }
}

The async iterator pattern is very handy for linear control flow and error handling.

Example 3 — Mongoose (integration)

Mongoose exposes Model.watch() (or connection.watch()) and will receive change events from anywhere (not only changes made through Mongoose). Example:

const mongoose = require('mongoose');
await mongoose.connect(process.env.MONGO_URI);

const changeStream = mongoose.connection.collection('orders').watch(
  [{ $match: { operationType: 'insert' } }],
  { fullDocument: 'default' }
);

changeStream.on('change', (change) => {
  console.log('New order inserted:', change.fullDocument);
});

Mongoose docs: change streams are server constructs and pick up changes from any client.

Example 4 — Python (PyMongo)

# watcher.py
from pymongo import MongoClient
import time

client = MongoClient("mongodb://localhost:27017/?replicaSet=rs0")
db = client['shop']
coll = db['orders']

# optional: load resume token from storage
resume_token = None

pipeline = [
    { '$match': { 'operationType': 'update',
                  'updateDescription.updatedFields.status': { '$exists': True } } }
]

with coll.watch(pipeline, full_document='updateLookup', resume_after=resume_token) as stream:
    for change in stream:
        print("Change:", change)
        # persist resume token
        resume_token = change['_id']
        # process...

PyMongo supports the same options as other drivers (resume_after, start_at_operation_time, full_document, etc.). See PyMongo docs for details.

Atlas Triggers — serverless glue built on change streams

If you use MongoDB Atlas, Atlas App Services (formerly Realm) provide Database Triggers that are backed by change streams. A trigger watches a collection (with a filter) and runs a serverless function on each matching change event. Triggers help remove the need to host a long-lived consumer application — Atlas keeps the stream open and runs your code. Typical use-cases: send notifications, enrich documents, call external APIs, or enqueue tasks on other services. (MongoDB)

Minimal Atlas trigger function example (Atlas UI/Function):

exports = function(changeEvent) {
  const doc = changeEvent.fullDocument;
  if (!doc) return;
  if (doc.status === "shipped") {
    // call 3rd-party email API (use context.http)
    console.log(`Order ${doc._id} shipped; sending email.`);
  }
};

In Atlas you configure:

• Trigger type: Database (change stream) or Scheduled.
• Namespace: database.collection to watch.
• Operation types: insert, update, replace, delete, etc.
• Optional filter (match expression) to limit triggers server-side.

How to shape change events with pipelines (server-side filtering & projection)

Change streams accept an aggregation-style pipeline which runs on server and returns only events that match. Use $match to avoid shipping unneeded events to your consumer, and $project to reshape events into the fields you expect.

Example pipeline to only receive update events and project a small payload:

const pipeline = [
  { $match: { operationType: 'update' } },
  { $project: {
      docId: '$documentKey._id',
      updatedFields: '$updateDescription.updatedFields',
      clusterTime: '$clusterTime'
  }}
];

If you expect high throughput, pushing filters into the pipeline drastically reduces network and app CPU cost.

See $changeStream and aggregation docs.

Resumability and error handling — production considerations

1. Persist resume tokens: after processing an event (or after a safe checkpoint), store the event's _id so you can call resumeAfter or startAfter on restarts. Resume tokens are opaque — treat them as binary blobs. (MongoDB)
2. Handle invalidation: a change stream can emit an invalidate event (e.g., on drop/rename). If this happens you may need to re-establish streams (and possibly full re-sync depending on your app).
3. Oplog window: resume tokens are valid only while the needed oplog entries still exist. If your process is down longer than the oplog retention window, resuming with a token might fail — be prepared for a fallback re-sync.
4. Backoff & retry: on transient errors, implement exponential backoff when recreating streams; on fatal errors, plan a safe restart / reinitialize.
5. Idempotency: your change-event handlers should be idempotent or deduplicating, since in some failure modes a change event might be delivered more than once.
6. Limit what you ask for: use server-side filters, fullDocument judiciously, and avoid requesting large post-images unless needed.
7. Monitoring: monitor change stream errors, lag, and the number of open cursors. On Atlas, monitor trigger failures in App Services logs.

Limitations & compatibility notes

• Change streams require a replica-set or sharded cluster (i.e., they rely on the oplog). They aren't supported on standalone single-node servers unless you initialize a single-node replica set.
• Some options (e.g., startAtOperationTime) require a timestamp within the oplog window to work.
• The $changeStream aggregation stage must be the first stage if you use it directly in db.aggregate([...]).

Advanced patterns

1. Durable, multi-worker consumer
If many workers process the same stream, consider centralizing event distribution (single consumer writes events to a durable queue — Kafka, Redis Stream, RabbitMQ — then workers process from the queue). This simplifies resume token storage and partitioning.

2. Event-sourcing vs. reactive side-effects
Treat change streams as a way to produce events for side effects (notifications, caches). If you need strict event-sourcing semantics (immutable event log, replayability, strict ordering across collections), consider combining change streams with an append-only event store (or push changes into Kafka).

3. Denormalization micro-batcher
Instead of updating denormalized documents per change, aggregate events in short time windows (1–5s) and perform bulk updates to reduce write amplification.

4. Low-latency dashboards
Feed selected change events into an in-memory real-time dashboard (WebSocket server). Keep server-side filters narrow.

Security and permissions

• The user connecting must have appropriate privileges to watch the collection or DB.
• In Atlas, configure triggers and function permissions according to principle of least privilege.
• Be careful when your change handler calls external services — validate/escape data to avoid injection-like issues.

Observability & cost

• Change streams keep a cursor open on the server; each open change stream consumes resources. If you have many consumers across many collections, consider consolidating watchers or use Atlas Triggers to avoid running many long-lived processes.
• If you run change streams that return large fullDocument payloads frequently, network costs can rise — use projection in pipelines to limit size.

Troubleshooting quick checklist

• "I get $changeStream stage is only supported on replica sets": make sure you are connected to a replica set (standalone DBs don't expose an oplog). (codemia.io)
• "My resume token fails": check oplog retention; the token may be too old and the oplog entries have rolled off.
• "I'm missing events": ensure your pipeline matches the structure of change events (e.g., updateDescription.updatedFields for updated fields; fullDocument may be null unless you requested updateLookup).

Quick reference — API bits

Node.js (collection.watch):

• collection.watch(pipeline?, options?) 
  • options.fullDocument: 'default' | 'updateLookup'
  • options.resumeAfter, options.startAfter, options.startAtOperationTime
  • changeStream.resumeToken available on cursor
• Use changeStream.close() to close.

Python (PyMongo):

• collection.watch(pipeline=None, full_document=None, resume_after=None, start_at_operation_time=None)

Atlas Triggers:

• Configure DB triggers in Atlas App Services: namespace, operation types, function to run per change.

Example: end-to-end flow (Order → Notification)

1. App inserts/updates orders in shop.orders.
2. Atlas trigger configured on shop.orders for update events with a filter updateDescription.updatedFields.status == "shipped".
3. Trigger runs a function that calls an external email service to notify the buyer, and logs to notifications collection.

This pattern decouples order writes from notification side effects and works even if the write originates outside your application (e.g., admin portal, direct DB load).

Resources & further reading

• MongoDB Change Streams manual (official). (MongoDB)
• $changeStream aggregation stage docs. (MongoDB)
• Node.js driver guide: monitoring with change streams. (MongoDB)
• Atlas App Services / Database Triggers docs (Atlas Triggers). (MongoDB)
• Mongoose docs on change streams. (mongoosejs.com)

Final checklist before you put it in production

• ☐ Replica set / sharded cluster confirmed and configured (not standalone).
• ☐ Resume tokens persisted and tested across restarts.
• ☐ Backfill strategy for oplog/gap conditions (what happens if resume token is stale).
• ☐ Idempotency & deduplication in handlers.
• ☐ Server-side filters to reduce event volume.
• ☐ Monitoring/alerts for stream errors and latency.
• ☐ Proper security and least-privilege access.