Change streams: MongoDB & NoSQL best practices explained

Change streams: MongoDB & NoSQL best practices explained

Real-time data synchronization is no longer a nice-to-have feature—it's fundamental to modern applications. Change streams let you watch MongoDB collections and react instantly to inserts, updates, and deletes, enabling everything from live notifications to cache invalidation without polling or message queues.

What they are — A stream of inserts, updates and deletes

Change streams are a MongoDB feature that allows your application to subscribe to data changes within a collection, database, or deployment. Rather than polling the database repeatedly or relying on external systems to detect changes, change streams push change events to your application in real time.

Each change event contains:

• Operation type: insert, update, replace, delete, or invalidate
• Document keys: The _id of the modified document
• Full document: The complete document after the change (if enabled)
• Update description: Field-level changes (for updates)
• Cluster time: When the change occurred
• Resume token: A unique marker to resume from this point

This is particularly valuable in e-commerce platforms where orders flow through multiple systems. Instead of having your cache layer, notification service, and search index all independently querying for new orders, they can all subscribe to the same stream and react instantly.

Watch a collection — Subscribe from your application

Watching a collection is straightforward. You call the watch() method on a collection, optionally passing an aggregation pipeline to filter which changes you care about, then iterate over the resulting stream.

const mongodb = require('mongodb');
const { MongoClient } = mongodb;

const client = new MongoClient('mongodb://localhost:27017');
const db = client.db('ecommerce');
const ordersCollection = db.collection('orders');

async function watchOrderInserts() {
  const stream = ordersCollection.watch([
    { $match: { operationType: 'insert' } }
  ]);

  for await (const change of stream) {
    const order = change.fullDocument;
    console.log('New order:', order._id);
    await notifyWarehouse(order);
    await updateSearchIndex(order);
  }
}

watchOrderInserts().catch(console.error);

The pipeline filters the stream at the server level, reducing bandwidth. You can match on operation type, document fields, or any combination. For a multi-tenant SaaS platform, you might filter by tenant ID; for a trading system, by account or symbol.

If you need to watch at the database or cluster level (all collections), use db.watch() or client.watch() respectively, though collection-level watching is more common for focused use cases.

In practice, you'll want to wrap the stream in proper error handling and a reconnection strategy, because network interruptions are inevitable in production:

async function watchWithResilience(collectionName, resumeToken = null) {
  const collection = db.collection(collectionName);
  let currentToken = resumeToken;

  while (true) {
    try {
      const options = currentToken ? { resumeAfter: currentToken } : {};
      const stream = collection.watch(
        [{ $match: { operationType: { $in: ['insert', 'update'] } } }],
        options
      );

      for await (const change of stream) {
        currentToken = change._id;
        await handleChange(change);
      }
    } catch (error) {
      console.error('Stream error:', error.message);
      await delay(5000); // Back off before retrying
    }
  }
}

function delay(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

Resume tokens — Store them to resume after restarts

Every change event carries a resume token—a unique identifier that marks its position in the change stream. If your application crashes or restarts, you can pick up from where you left off instead of reprocessing all historical changes or missing changes entirely.

Resume tokens are opaque and should be treated as such. Store them in a durable location—Redis, PostgreSQL, or even MongoDB itself—before processing each change. The order matters: save the token first, then process the change, so a crash doesn't cause you to lose track.

const redis = require('redis');
const redisClient = redis.createClient();

async function watchWithPersistence(collectionName) {
  const collection = db.collection(collectionName);
  
  // Load the last token from Redis
  let lastToken = await redisClient.get(`changestream:${collectionName}`);
  const options = lastToken ? { resumeAfter: JSON.parse(lastToken) } : {};

  const stream = collection.watch([], options);

  for await (const change of stream) {
    // Persist the token before processing
    await redisClient.set(
      `changestream:${collectionName}`,
      JSON.stringify(change._id)
    );

    // Now process the change safely
    await handleChange(change);
  }
}

async function handleChange(change) {
  if (change.operationType === 'insert') {
    await updateCache(change.fullDocument);
  } else if (change.operationType === 'update') {
    await invalidateCache(change.documentKey._id);
  }
}

In distributed systems, consider storing resume tokens per consumer or per service. A notification service, cache layer, and search index don't need to share the same token—they each progress independently through the stream.

Use cases — Cache invalidation, notifications and syncing search indexes

Cache invalidation: When a document changes, invalidate it from your cache immediately. This is far more reliable than time-based expiry. If a user updates their profile, subscribers see the new data within milliseconds, not minutes.

const redis = require('redis');
const cache = redis.createClient();

async function watchForCacheInvalidation() {
  const usersCollection = db.collection('users');
  const stream = usersCollection.watch([
    { $match: { operationType: { $in: ['update', 'replace', 'delete'] } } }
  ]);

  for await (const change of stream) {
    const userId = change.documentKey._id;
    await cache.del(`user:${userId}`);
    console.log(`Invalidated cache for user ${userId}`);
  }
}

watchForCacheInvalidation().catch(console.error);

Real-time notifications: Send notifications to clients as soon as relevant data changes. A ride-hailing app notifies drivers of new ride requests; a trading platform alerts users of price movements; a SaaS app notifies team members of project updates.

async function watchAndNotify() {
  const projectsCollection = db.collection('projects');
  const stream = projectsCollection.watch([
    {
      $match: {
        operationType: 'update',
        'updateDescription.updatedFields.status': { $exists: true }
      }
    }
  ]);

  for await (const change of stream) {
    const projectId = change.documentKey._id;
    const newStatus = change.fullDocument.status;
    
    // Send WebSocket message to connected clients
    await io.to(`project:${projectId}`).emit('statusChanged', {
      projectId,
      newStatus,
      timestamp: new Date()
    });
  }
}

watchAndNotify().catch(console.error);

Search index synchronization: Keep your Elasticsearch or Algolia indexes in sync with MongoDB without building complex ingestion pipelines. Every change flows through immediately.

const { Client } = require('@elastic/elasticsearch');
const esClient = new Client({ node: 'http://localhost:9200' });

async function syncSearchIndex() {
  const productsCollection = db.collection('products');
  const stream = productsCollection.watch();

  for await (const change of stream) {
    const { operationType, documentKey, fullDocument } = change;

    if (operationType === 'insert' || operationType === 'replace') {
      await esClient.index({
        index: 'products',
        id: documentKey._id.toString(),
        document: fullDocument
      });
    } else if (operationType === 'delete') {
      await esClient.delete({
        index: 'products',
        id: documentKey._id.toString()
      });
    } else if (operationType === 'update') {
      await esClient.update({
        index: 'products',
        id: documentKey._id.toString(),
        doc: fullDocument
      });
    }
  }
}

syncSearchIndex().catch(console.error);

Common mistakes

• Not persisting resume tokens: Restarting your application without saving tokens causes you to miss or replay changes. Always store tokens durably.
• Ignoring backpressure: If your change handler is slow, the stream will buffer changes in memory. Add error handling, retries, and logging to understand bottlenecks.
• Processing without idempotency: If your app crashes after handling a change but before updating the resume token, that change will be replayed. Ensure your handlers are idempotent—applying the same change twice has the same effect as once.
• Filtering all work in the pipeline: While aggregation pipelines reduce bandwidth, complex logic belongs in your application. Keep server-side filters simple and move business logic to code.
• Watching at the wrong level: Watching an entire database or cluster is rare and inefficient. Watch specific collections and filter within them.
• Not handling connection loss: Change streams can disconnect. Implement exponential backoff and reconnection logic, not just a single watch call.

Key takeaways

• Change streams let you subscribe to real-time MongoDB changes instead of polling, enabling responsive applications.
• Combine aggregation pipelines with change streams to filter at the server level and reduce bandwidth.
• Always persist resume tokens durably—in Redis, PostgreSQL, or MongoDB—so you can resume without data loss after restarts.
• Common use cases include cache invalidation, real-time notifications, and search index synchronization.
• Build resilience into your change stream handlers: handle errors, reconnect with backoff, and make handlers idempotent.
• Store resume tokens before processing changes to ensure consistency between your stream position and your application state.