Change Streams and real-time data processing

Basic Concepts of Change Streams

Change Streams are a mechanism provided by MongoDB for real-time data change monitoring, allowing applications to subscribe to change events in the database. Unlike traditional polling methods, Change Streams adopt a push model, immediately notifying clients when data changes occur in a collection, database, or the entire cluster. This mechanism is built on MongoDB's oplog (operation log) and can capture operations such as insertions, updates, replacements, and deletions.

Key features of Change Streams include:

• Real-time: Changes are propagated to subscribed clients almost immediately.
• Reliability: Utilizes MongoDB's replica set features to ensure no events are lost.
• Flexibility: Can monitor changes at the level of a single collection, an entire database, or the cluster.
• Recoverability: Supports resuming monitoring from a specific point in time or operation token.

// Basic Change Stream monitoring example
const { MongoClient } = require('mongodb');

async function watchCollection() {
  const client = new MongoClient('mongodb://localhost:27017');
  await client.connect();
  
  const collection = client.db('test').collection('inventory');
  const changeStream = collection.watch();
  
  changeStream.on('change', (change) => {
    console.log('Change detected:', change);
  });
}

watchCollection().catch(console.error);

How Change Streams Work

The core of MongoDB Change Streams relies on the replica set's oplog mechanism. When Change Streams are enabled, MongoDB creates a cursor that continuously tracks new entries in the oplog. For sharded clusters, Change Streams aggregate oplog changes from all shards, ensuring clients receive a globally consistent sequence of changes.

Change Stream events contain the following key information:

• _id: A unique identifier for the event, which can be used to resume interrupted monitoring.
• operationType: The type of operation (insert/update/replace/delete, etc.).
• ns: The namespace (database and collection name).
• documentKey: The _id of the affected document.
• fullDocument: For insert and replace operations, contains the complete document.
• updateDescription: For update operations, includes the modified fields and new values.

// Typical structure of a Change Stream event
{
  _id: { _data: '82638B...' },
  operationType: 'update',
  ns: { db: 'test', coll: 'inventory' },
  documentKey: { _id: ObjectId("5f8d...") },
  updateDescription: {
    updatedFields: { quantity: 95 },
    removedFields: [],
    truncatedArrays: []
  },
  clusterTime: Timestamp({ t: 160272..., i: 1 })
}

Filtering and Configuring Change Streams

Change Streams support powerful filtering capabilities, allowing you to focus on specific types of changes. Filters can be defined using pipeline expressions, similar to the syntax of aggregation pipelines. Common filtering conditions include operation types and changes in document field values.

// Change Stream with filtering
const pipeline = [
  { $match: { 
    $or: [
      { operationType: 'insert' },
      { 
        operationType: 'update',
        'updateDescription.updatedFields.quantity': { $lt: 10 }
      }
    ]
  }}
];

const changeStream = collection.watch(pipeline);

Change Streams also support various configuration options:

• fullDocument: Controls whether to return the full document (defaults to only returning updated fields).
• fullDocumentBeforeChange: Captures the document state before the change.
• resumeAfter: Resumes monitoring from a specific event.
• startAtOperationTime: Starts monitoring from a specific point in time.
• batchSize: Controls the number of events returned per batch.

Real-Time Data Processing Use Cases

Change Streams are particularly suitable for applications requiring real-time responses:

Inventory Management System: Triggers restocking processes immediately when inventory levels fall below a threshold.

// Inventory alert system
const inventoryAlertStream = inventoryCollection.watch([
  { $match: { 
    operationType: 'update',
    'updateDescription.updatedFields.quantity': { $lt: 5 }
  }}
]);

inventoryAlertStream.on('change', async (change) => {
  const productId = change.documentKey._id;
  await sendReorderRequest(productId);
  console.log(`Restock request initiated for product ${productId}`);
});

Chat Application: Pushes new messages to online users in real time.

// Chat message push
const messageStream = chatCollection.watch([
  { $match: { operationType: 'insert' } }
]);

messageStream.on('change', (change) => {
  const message = change.fullDocument;
  io.to(message.roomId).emit('new_message', message);
});

IoT Data Processing: Processes sensor data in real time and triggers alerts.

// Temperature monitoring system
const tempStream = sensorsCollection.watch([
  { $match: { 
    operationType: 'update',
    'updateDescription.updatedFields.temperature': { $gt: 30 }
  }}
]);

tempStream.on('change', (change) => {
  const sensorId = change.documentKey._id;
  triggerAlarm(`Sensor ${sensorId} temperature too high!`);
});

Integrating Change Streams with Frontend

Frontend applications can connect to backend services via WebSocket, where the backend uses Change Streams to monitor database changes and push them to the frontend in real time. This architecture enables true full-stack real-time applications.

// Backend WebSocket service example
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });

// MongoDB Change Stream monitoring
const changeStream = db.collection('orders').watch();

wss.on('connection', (ws) => {
  // Send current state when a new connection is established
  const sendInitialData = async () => {
    const orders = await db.collection('orders').find().toArray();
    ws.send(JSON.stringify({ type: 'INITIAL_DATA', data: orders }));
  };
  
  sendInitialData();
  
  // Monitor changes and push to clients
  changeStream.on('change', (change) => {
    ws.send(JSON.stringify({ type: 'CHANGE', data: change }));
  });
});

Frontend can handle real-time updates like this:

// Frontend WebSocket handling
const socket = new WebSocket('ws://localhost:8080');

socket.onmessage = (event) => {
  const message = JSON.parse(event.data);
  switch(message.type) {
    case 'INITIAL_DATA':
      renderOrders(message.data);
      break;
    case 'CHANGE':
      handleOrderChange(message.data);
      break;
  }
};

function handleOrderChange(change) {
  if (change.operationType === 'insert') {
    addNewOrderToUI(change.fullDocument);
  } else if (change.operationType === 'update') {
    updateOrderInUI(change.documentKey._id, change.updateDescription);
  }
}

Performance Considerations for Change Streams

When using Change Streams, consider the following performance factors:

1. Resource Consumption: Long-running Change Streams maintain database connections, so connection pools should be managed properly.
2. Event Ordering: In sharded clusters, the order of cross-shard events may be affected.
3. Recovery Strategy: Implement checkpoint mechanisms to record resumeToken for processed events.
4. Error Handling: Re-establish monitoring when network interruptions or cursor closures occur.

// Change Stream with error handling and recovery mechanism
let resumeToken;
const startChangeStream = async () => {
  const options = resumeToken ? { resumeAfter: resumeToken } : {};
  
  const changeStream = collection.watch([], options);
  
  changeStream.on('change', (change) => {
    console.log('Processing change:', change);
    resumeToken = change._id; // Update resume token
    // Business logic processing...
  });
  
  changeStream.on('error', (err) => {
    console.error('Change Stream error:', err);
    setTimeout(startChangeStream, 1000); // Retry after 1 second
  });
};

startChangeStream();

Advanced Change Stream Patterns

For complex scenarios, combine aggregation pipelines to implement more powerful real-time processing logic:

Data Transformation: Transform data formats directly in the Change Stream.

const transformPipeline = [
  { $match: { operationType: { $in: ['insert', 'update'] } } },
  { $addFields: {
    timestamp: { $toDate: '$clusterTime' },
    modifiedFields: {
      $cond: {
        if: { $eq: ['$operationType', 'update'] },
        then: '$updateDescription.updatedFields',
        else: null
      }
    }
  }}
];

const enhancedStream = collection.watch(transformPipeline);

Join Queries: Incorporate related data in change events.

const joinPipeline = [
  { $match: { operationType: 'insert' } },
  { $lookup: {
    from: 'users',
    localField: 'userId',
    foreignField: '_id',
    as: 'user'
  }},
  { $unwind: '$user' }
];

Time Window Aggregation: Calculate metrics in sliding time windows in real time.

const timeWindowPipeline = [
  { $match: { operationType: 'insert' } },
  { $group: {
    _id: {
      $dateToString: { format: "%Y-%m-%d %H:%M", date: "$clusterTime" }
    },
    count: { $sum: 1 },
    avgValue: { $avg: "$value" }
  }},
  { $sort: { "_id": 1 } }
];

Monitoring and Managing Change Streams

In production environments, monitor the health of Change Streams:

1. Metric Collection: Track the number of processed events, latency, etc.
2. Logging: Record errors and exceptions in detail.
3. Flow Control: Implement backpressure mechanisms under high load.
4. Automatic Recovery: Implement reconnection and state recovery.

// Monitoring decorator example
function monitoredChangeStream(collection, pipeline, callback) {
  let eventsProcessed = 0;
  let lastEventTime = null;
  
  const changeStream = collection.watch(pipeline);
  
  // Collect metrics
  const metrics = {
    getStatus: () => ({
      active: changeStream.isClosed(),
      eventsProcessed,
      lastEventTime,
      latency: lastEventTime ? Date.now() - lastEventTime.getTime() : null
    })
  };
  
  changeStream.on('change', (change) => {
    try {
      callback(change);
      eventsProcessed++;
      lastEventTime = new Date();
    } catch (err) {
      console.error('Error processing change:', err);
    }
  });
  
  changeStream.on('error', (err) => {
    console.error('Change Stream error:', err);
  });
  
  return { changeStream, metrics };
}

// Using the monitoring decorator
const { changeStream, metrics } = monitoredChangeStream(
  collection, 
  [], 
  handleChange
);

// Periodically output metrics
setInterval(() => {
  console.log('Change Stream metrics:', metrics.getStatus());
}, 60000);