Change Streams: Real-Time Change Monitoring

Change Streams is the foundation for MongoDB's real-time change notifications—CDC (Change Data Capture)—and event-driven applications.

1. What You'll Learn

• The Concept of Change Streams (Based on oplog)
• db.collection.watch() Listening
• Pipeline filtering ($match)
• fullDocument: 'updateLookup'
• Integration with Mongoose
• Typical Scenarios (Real-Time Notifications, Data Synchronization)

graph LR
    A[Applications 1<br/>Insert Order] -->|oplog| DB[(MongoDB<br/>oplog)]
    B[Applications 2<br/>Update Order] -->|oplog| DB
    C[Applications 3<br/>Delete Order] -->|oplog| DB

    DB -->|Change Stream| CS[db.collection.watch]

    CS -->|operationType=insert| N1[Email Notification<br/>WebSocket Push]
    CS -->|operationType=update| N2[Elasticsearch<br/>Synchronize]
    CS -->|operationType=delete| N3[Redis Cache<br/>Invalidation]

    style CS fill:#d4edda
    style N1 fill:#cce5ff

graph LR
    A[Applications 1<br/>Insert Order] -->|oplog| DB[(MongoDB<br/>oplog)]
    B[Applications 2<br/>Update Order] -->|oplog| DB
    C[Applications 3<br/>Delete Order] -->|oplog| DB

    DB -->|Change Stream| CS[db.collection.watch]

    CS -->|operationType=insert| N1[Email Notification<br/>WebSocket Push]
    CS -->|operationType=update| N2[Elasticsearch<br/>Synchronize]
    CS -->|operationType=delete| N3[Redis Cache<br/>Invalidation]

    style CS fill:#d4edda
    style N1 fill:#cce5ff

2. Change Streams Basics

Concept Overview: Change Streams is a real-time change monitoring API introduced in MongoDB 3.6 and later, implemented using the replica set oplog. It allows applications to detect data changes in real time without polling—serving as the foundation for CDC (Change Data Capture) and event-driven architectures.

OpLog Principles: The Oplog (Operation Log) is a special capped collection that records all write operations in a replica set. The Primary appends each write operation (insert/update/delete) to the Oplog, and the Secondary replicates data by tailing the Oplog. Change Streams are essentially a structured encapsulation of the oplog—they convert raw oplog entries into human-readable change events and provide advanced features such as filtering and resume-from-breakpoint functionality.

graph TB
    subgraph "OpLog How It Works"
        W1[Write Operation] --> OP[(oplog<br/>Fixed Set)]
        OP --> S1[Secondary 1<br/>tailing oplog]
        OP --> S2[Secondary 2<br/>tailing oplog]
        OP --> CS[Change Stream<br/>Structured Events]
    end

    subgraph "Change Stream vs Polling"
        P1[Polling: Queries per second] --> C1[High latency<br/>High Load<br/>Duplicate Query]
        CS --> C2[Real-time Notifications<br/>Low load<br/>No duplicates]
    end

    style CS fill:#d4edda
    style C2 fill:#d4edda
    style C1 fill:#f8d7da

graph TB
    subgraph "OpLog How It Works"
        W1[Write Operation] --> OP[(oplog<br/>Fixed Set)]
        OP --> S1[Secondary 1<br/>tailing oplog]
        OP --> S2[Secondary 2<br/>tailing oplog]
        OP --> CS[Change Stream<br/>Structured Events]
    end

    subgraph "Change Stream vs Polling"
        P1[Polling: Queries per second] --> C1[High latency<br/>High Load<br/>Duplicate Query]
        CS --> C2[Real-time Notifications<br/>Low load<br/>No duplicates]
    end

    style CS fill:#d4edda
    style C2 fill:#d4edda
    style C1 fill:#f8d7da

Change Streams vs. Polling: A Comparison:

Change Event Structure:

// === Monitor Set Changes ===
const changeStream = db.products.watch();

changeStream.on('change', (change) => {
  console.log('Change detected:', change);
  // {
  //   _id: { _data: '...' },           // resumeToken(For resuming downloads)
  //   operationType: 'insert',          // Operation Type
  //   fullDocument: { _id: ..., sku: ..., title: ..., ... },  // Complete Documentation
  //   ns: { db: 'shopdb', coll: 'products' },  // Namespace
  //   documentKey: { _id: ObjectId('...') }    // Document Primary Key
  // }
});

Key Points Analysis:

1. Change Streams must run on a replica set or sharded cluster (depends on the oplog).
2. update By default, the event does not return the full document; you must set fullDocument: 'updateLookup'
3. The _id field is the resumeToken, which is used for resuming downloads—save it, and after restarting, continue listening from where you left off.

3. Pipeline Filtration

Concept Description: Change Streams supports aggregation pipeline filtering—by applying stages such as $match and $project to the oplog event stream, only events of interest are passed to the application. Filtering is performed on the server side, reducing network traffic and application-layer processing overhead.

How It Works: watch() accepts an array of aggregation pipelines as a parameter; the pipeline stages are executed on the MongoDB server side. Events first pass through the pipeline for filtering, and only those that pass are pushed to the client. It supports stages such as $match, $project, $addFields, and $replaceRoot.

Filtering Policy:

// === Filter by Specific Actions ===
const changeStream = db.products.watch([
  { $match: { operationType: 'insert' } }
]);

// === Filter by Specific Fields ===
const changeStream = db.products.watch([
  { $match: { 'fullDocument.category': 'Electronics' } }
]);

// === Filter by Price Changes ===
const changeStream = db.products.watch([
  {
    $match: {
      $or: [
        { operationType: 'insert' },
        { operationType: 'update', 'updateDescription.updatedFields.price': { $exists: true } }
      ]
    }
  }
]);

Key Points Analysis:

1. Filtering is performed on the server side, reducing unnecessary network traffic.
2. The field path for $match uses the oplog event structure (such as fullDocument.category), not the original document fields.
3. Stages that require global state, such as $group and $limit, are not supported.

4. fullDocument Configuration

Concept Description: The fullDocument option controls whether Change Stream events include the full document. By default, update events return only the changed fields (updateDescription) and do not return the full document. When fullDocument: 'updateLookup' is set, MongoDB performs an additional query to retrieve the full content of the current document.

How It Works:

• whenAvailable (default): insert/replace events include the entire document; update events include only the fields that have changed; delete events do not include any document
• updateLookup: All events include the full document (update events will additionally query the latest document), but there is additional read overhead and a slight delay.
• required: Must return the complete document; otherwise, an error will be reported.

Trade-offs of updateLookup:

sequenceDiagram
    participant App as Applications
    participant DB as MongoDB
    participant Doc as Document

    App->>DB: watch([], {fullDocument: 'updateLookup'})
    DB->>DB: oplog Generate update Event

    Note over DB: update By default, events contain only the changed fields.

    DB->>Doc: View the current complete document
    Doc-->>DB: Back to Latest Documents
    DB-->>App: Push Events + fullDocument

    Note over App: The incident includes the complete documentation

sequenceDiagram
    participant App as Applications
    participant DB as MongoDB
    participant Doc as Document

    App->>DB: watch([], {fullDocument: 'updateLookup'})
    DB->>DB: oplog Generate update Event

    Note over DB: update By default, events contain only the changed fields.

    DB->>Doc: View the current complete document
    Doc-->>DB: Back to Latest Documents
    DB-->>App: Push Events + fullDocument

    Note over App: The incident includes the complete documentation

// === update Return to the full document ===
const changeStream = db.products.watch([], {
  fullDocument: 'updateLookup'
});

changeStream.on('change', (change) => {
  if (change.operationType === 'update') {
    console.log('Updated doc:', change.fullDocument);
    // Complete Documentation(After the default value changes)
  }
});

// === Return only the fields that have changed ===
const changeStream = db.products.watch([], {
  fullDocument: 'whenAvailable'  // Default
});

Key Points Analysis:

1. updateLookup Performs an additional query, which increases database load in scenarios with frequent updates.
2. updateLookup returns the most recent document as of the query time, which may differ slightly from the document at the time of the change (due to other concurrent modifications).
3. The delete event does not return a document even if updateLookup is set (the document no longer exists)

5. Mongoose Integration

Concept Explanation: Mongoose 6+ natively supports Change Streams, which are returned via Model.watch(). This is fully consistent with the native MongoDB driver’s API, but using it directly at the model layer aligns better with Mongoose’s development conventions.

Event-Driven Architecture: Change Streams are a core component of event-driven architecture—database changes serve as the event source, driving downstream side effects such as cache updates, search synchronization, and notification push.

graph TB
    subgraph "Source of the Incident"
        DB[(MongoDB<br/>oplog)]
    end

    subgraph "Change Stream Bus"
        CS[Model.watch<br/>Change Flow]
    end

    subgraph "Event Consumer"
        N1[Email Notification]
        N2[Elasticsearch<br/>Search Synchronization]
        N3[Redis<br/>Cache Expiration]
        N4[WebSocket<br/>Real-time Notifications]
        N5[Audit Log<br/>Compliance Record]
    end

    DB --> CS
    CS --> N1
    CS --> N2
    CS --> N3
    CS --> N4
    CS --> N5

    style CS fill:#d4edda

graph TB
    subgraph "Source of the Incident"
        DB[(MongoDB<br/>oplog)]
    end

    subgraph "Change Stream Bus"
        CS[Model.watch<br/>Change Flow]
    end

    subgraph "Event Consumer"
        N1[Email Notification]
        N2[Elasticsearch<br/>Search Synchronization]
        N3[Redis<br/>Cache Expiration]
        N4[WebSocket<br/>Real-time Notifications]
        N5[Audit Log<br/>Compliance Record]
    end

    DB --> CS
    CS --> N1
    CS --> N2
    CS --> N3
    CS --> N4
    CS --> N5

    style CS fill:#d4edda

// === mongoose Change Streams(mongoose 6+)===
const Product = mongoose.model('Product', productSchema);

// Monitoring Product Set Changes
const changeStream = Product.watch();

changeStream.on('change', (change) => {
  console.log(`${change.operationType}:`, change.fullDocument);
});

// Filter
const filteredStream = Product.watch([
  { $match: { operationType: { $in: ['insert', 'update'] } } }
]);

Key Points Analysis:

1. Mongoose's watch() returns the native MongoDB Change Stream, and the API is completely identical.
2. The database connection must be maintained during monitoring; if the connection is lost, Change Stream automatically shuts down.
3. In a production environment, it is recommended to wrap the Change Stream Manager: automatic reconnection + resumeToken persistence

6. Real-World Scenarios

Concept Overview: Change Streams have three classic use cases in production environments: real-time notifications, data synchronization (CDC), and cache invalidation. Each of these use cases is a typical implementation of an event-driven architecture.

(1) Real-Time Notification System

Scenario: ShopHub e-commerce needs to send real-time email notifications and WebSocket push notifications to the merchant’s backend when a new order is created.

// === Monitor New Orders，Send a notification ===
const OrderStream = db.orders.watch([
  { $match: { operationType: 'insert' } }
]);

OrderStream.on('change', async (change) => {
  const order = change.fullDocument;

  // Send an Email
  await sendEmail(order.userId, 'Order confirmation', `Order ${order._id} received`);

  // Push Notifications
  await pushNotification(order.userId, {
    title: 'New Order',
    body: `Order total: $${order.total}`
  });

  // WebSocket Real-time Notifications
  io.emit('new_order', order);
});

(2) Data Synchronization (CDC)

Scenario: ShopHub needs to synchronize MongoDB product data in real time to Elasticsearch to support full-text search. Change Streams enables a zero-latency CDC pipeline.

CDC Architecture:

graph LR
    A[MongoDB<br/>Product Data] -->|Change Stream| B[CDC Worker<br/>Node.jsProcess]
    B -->|insert/update| C[Elasticsearch<br/>Search Index]
    B -->|delete| C
    B -->|Change Log| D[(Redis<br/>resumeToken)]
    D -->|Restart and Recover| B

    style B fill:#d4edda

graph LR
    A[MongoDB<br/>Product Data] -->|Change Stream| B[CDC Worker<br/>Node.jsProcess]
    B -->|insert/update| C[Elasticsearch<br/>Search Index]
    B -->|delete| C
    B -->|Change Log| D[(Redis<br/>resumeToken)]
    D -->|Restart and Recover| B

    style B fill:#d4edda

// === Monitoring MongoDB Change，Sync to Elasticsearch ===
const ProductStream = db.products.watch();

ProductStream.on('change', async (change) => {
  switch (change.operationType) {
    case 'insert':
    case 'update':
    case 'replace':
      await elasticsearch.index({
        index: 'products',
        id: change.documentKey._id.toString(),
        body: change.fullDocument
      });
      break;
    case 'delete':
      await elasticsearch.delete({
        index: 'products',
        id: change.documentKey._id.toString()
      });
      break;
  }
});

(3) Cache Invalidation

Scenario: ShopHub uses Redis to cache product detail pages. When product data changes, the cache is automatically cleared to prevent stale data.

// === Monitor Product Changes，Invalidate Redis cache ===
const ProductStream = db.products.watch();

ProductStream.on('change', async (change) => {
  const productId = change.documentKey._id.toString();
  await redis.del(`product:${productId}`);
  console.log(`Cache cleared for ${productId}`);
});

▶ Example 1: Change Stream Resume After Interruption

// Alice's TechCorp System: Change Stream Resume, Events Are Not Lost After a Restart
async function startResumableStream() {
  // 1. Retrieve last saved resumeToken from Redis
  let resumeToken = await redis.get('product_stream_token');
  let options = { fullDocument: 'updateLookup' };

  if (resumeToken) {
    options.resumeAfter = JSON.parse(resumeToken);
    console.log('Resuming from saved token');
  }

  // 2. Start Listening
  const changeStream = db.products.watch([], options);

  changeStream.on('change', async (change) => {
    // Processing Changes...
    console.log(`${change.operationType}: ${change.documentKey._id}`);

    // 3. Save after each event resumeToken
    await redis.set('product_stream_token', JSON.stringify(change._id));
  });

  changeStream.on('error', async (err) => {
    console.error('Stream error:', err.message);
    // 4. Delayed Reconnection After an Error
    setTimeout(startResumableStream, 5000);
  });
}

startResumableStream();

7. Change Streams Limitations

Conceptual Note: Change Streams rely on the oplog and replica sets and have clear limitations. Understanding these limitations is essential for designing reliable real-time systems.

Detailed Explanation of Restrictions:

oplog Window Monitoring: The oplog has a fixed size, and older entries are overwritten. If Change Stream consumes data more slowly than the oplog is overwritten, the resumeToken will expire, and resume-from-breakpoint functionality will not work.

graph LR
    A[oplog Write Speed<br/>1000 ops/s] --> B[oplog Capacity<br/>5% Disk ≈ 50GB]
    B --> C[oplog Window<br/>about 72 hours]
    C --> D{Consumption Rate?}
    D -->|Keep up| E[✅ Normal]
    D -->|Behind > 72h| F[❌ resumeToken Failure<br/>A full resync is required]

    style E fill:#d4edda
    style F fill:#f8d7da

graph LR
    A[oplog Write Speed<br/>1000 ops/s] --> B[oplog Capacity<br/>5% Disk ≈ 50GB]
    B --> C[oplog Window<br/>about 72 hours]
    C --> D{Consumption Rate?}
    D -->|Keep up| E[✅ Normal]
    D -->|Behind > 72h| F[❌ resumeToken Failure<br/>A full resync is required]

    style E fill:#d4edda
    style F fill:#f8d7da

▶ Example: Hands-On Guide to Change Streams Real-Time Order Notifications

// Scene: Monitor New Orders, Send emails in real time + WebSocket Push + Sync to Elasticsearch

// 1. Start Listening (in Node.js)
const { MongoClient } = require('mongodb');

async function startOrderListener() {
  const client = new MongoClient('mongodb://localhost:27017/?replicaSet=rs0');
  await client.connect();
  const orders = client.db('shopdb').collection('orders');

  // Monitor all changes to the order collection
  const changeStream = orders.watch([
    {
      $match: {
        operationType: 'insert',  // Listen-only insertion
        'fullDocument.status': 'paid'  // Process only paid orders
      }
    }
  ]);

  changeStream.on('change', async (change) => {
    const order = change.fullDocument;
    console.log(`New Orders: ${order._id}, Amount: $${order.total}`);

    // 1. Send an email notification
    await sendEmail(order.userId, {
      subject: 'Order Confirmation',
      body: `Your Order ${order._id} Submitted，Total Amount $${order.total}`
    });

    // 2. WebSocket Real-time push notifications to the merchant's backend
    io.to('merchant-dashboard').emit('new_order', {
      orderId: order._id,
      total: order.total,
      items: order.items,
      timestamp: order.createdAt
    });

    // 3. Sync to Elasticsearch For Search
    await elasticsearch.index({
      index: 'orders',
      id: order._id.toString(),
      body: order
    });
  });

  // 4. Monitor Product Changes，Synchronized Update Elasticsearch
  const productStream = client.db('shopdb').collection('products').watch();
  productStream.on('change', async (change) => {
    if (change.operationType === 'delete') {
      await elasticsearch.delete({
        index: 'products',
        id: change.documentKey._id.toString()
      });
    } else if (change.fullDocument) {
      await elasticsearch.index({
        index: 'products',
        id: change.documentKey._id.toString(),
        body: change.fullDocument
      });
    }
  });

  console.log('Change Streams listening...');
}

// 5. Resume Download(Restore progress tracking after the app restarts)
async function resumeAfterRestart() {
  const resumeToken = await redis.get('change_stream_resume_token');
  const changeStream = orders.watch([], {
    resumeAfter: JSON.parse(resumeToken),
    fullDocument: 'updateLookup'
  });

  changeStream.on('change', async (change) => {
    // Processing Changes...
    // Save resume token
    await redis.set('change_stream_resume_token', JSON.stringify(change._id));
  });
}

startOrderListener().catch(console.error);

// 2. Manual Testing in mongosh
// Trigger Event:Insert a New Order
db.orders.insertOne({
  userId: 'user_001',
  items: [{ sku: 'PHONE-001', qty: 1, price: 599 }],
  total: 599,
  status: 'paid',
  createdAt: new Date()
});
// The app receives a notification immediately:Send an Email + WebSocket Push + ES Index

Output: Triggered within < 100 ms after an order is inserted, automatically completing all side effects—such as emails, push notifications, and search synchronization—without the need for polling.

❓ FAQ

📖 Summary

• Change Streams: Real-time change notifications based on the oplog
• watch() method for listening + $match filtering
• operationType: insert / update / delete / replace / drop
• fullDocument: 'updateLookup' Get the full document
• Typical scenarios: real-time notifications, CDC, cache invalidation
• Required replica sets, oplog size limit

📝 Exercises

1. Basic Problem (⭐): Monitor all changes to the products collection and print them to the console.
2. Basic Question (⭐): Use $match to filter insert operations.
3. Advanced Exercise (⭐⭐): Listen for new orders and send email notifications (using a mock email function).
4. Advanced Exercise (⭐⭐): Implement product cache expiration (Redis synchronization).
5. Challenge (⭐⭐⭐): Complete the CDC system (real-time synchronization between MongoDB and Elasticsearch).

Web-Tutorial Tech Team

A team of developers maintaining programming tutorials. Each tutorial is written and reviewed by developers with expertise in that field. We work to keep our content accurate and reliable — if you spot an issue, please let us know.

About Us