#watch() API

#Overview

watch() returns the MongoDB driver's native ChangeStream<T> object. Use it when you want direct driver-level event handling. If you need persisted resume tokens, multi-target fanout, idempotency markers, and lifecycle stats, use ChangeStreamSyncManager.

#Basic usage

import MonSQLize from 'monsqlize';
const msq = new MonSQLize({
  type: 'mongodb',
  databaseName: 'mydb',
  config: { uri: 'mongodb://localhost:27017' }
});

await msq.connect();
const collection = msq.collection('users');

//Listen for all changes
const watcher = collection.watch();

watcher.on('change', (change) => {
  console.log('Data changes:', change.operationType);
  console.log('Documentation:', change.fullDocument);
});

#API Reference

#collection.watch([pipeline], [options])

Monitor collection data changes.

Parameters:

• pipeline (Array, optional): aggregation pipeline for filtering events
• options (Object, optional): Configuration options

Return value: ChangeStream<TSchema> — MongoDB driver native ChangeStream object

⚠️ collection.watch() directly returns the MongoDB driven ChangeStream without additional packaging. Please refer to MongoDB ChangeStream official documentation.

#Configuration options

#MongoDB native options

#ChangeStream native method

watch() returns the MongoDB-driven ChangeStream<T> object, which supports the following native APIs:

#cs.on(event, handler)

Listen for events (EventEmitter interface).

Event List:

• 'change': Data change
• 'error': Error
• 'close': Close
• 'end': end of stream

#cs.once(event, handler)

Listen for events (one-time).

#cs.close()

Close ChangeStream.

Return value: Promise<void>

#cs.closed

Read-only property that checks whether the ChangeStream has been closed.

Type: boolean

#cs.resumeToken

Read-only attribute, obtains the latest resumeToken (used for resumed transmission).

Type: unknown

#cs.next()

Explicitly get the next change event (iterator pattern).

Return value: Promise<ChangeStreamDocument<T>>

💡 If you want resume-token persistence, multi-target synchronization, and lifecycle stats for supervised restarts, use ChangeStreamSyncManager.

#Usage example

#Example 1: Basic monitoring

const watcher = collection.watch();

watcher.on('change', (change) => {
  console.log('Operation type:', change.operationType);
  console.log('Document ID:', change.documentKey._id);
});

#Example 2: Filter events

//Only listen to insert and update
const watcher = collection.watch([
  { $match: { operationType: { $in: ['insert', 'update'] } } }
]);

watcher.on('change', (change) => {
  console.log('Add or modify:', change.fullDocument);
});

#Example 3: Error handling

const cs = collection.watch();

cs.on('error', (error) => {
  console.error('Change Stream error:', error);
});

cs.on('close', () => {
  console.log('Change Stream is closed');
});

#Example 4: Statistics monitoring (via ChangeStreamSyncManager)

For built-in statistics (eventCount / syncedCount / errorCount), please use ChangeStreamSyncManager:

import MonSQLize from 'monsqlize';

const syncManager = new MonSQLize.ChangeStreamSyncManager({ db, config: { ... } });
await syncManager.start();

setInterval(() => {
  const stats = syncManager.getStats();
  console.log('Synced events:', stats.syncedCount, 'Error:', stats.errorCount);
}, 60000);

#Example 5: Graceful shutdown

const cs = collection.watch();

process.on('SIGTERM', async () => {
  console.log('Closing watch...');
  await cs.close();
  console.log('watch is closed');
  process.exit(0);
});

#Cache invalidation integration

⚠️ collection.watch() itself does not provide built-in cache invalidation function. If you want to integrate watch with cache, there are two options:

#Solution 1: Manual processing (lightweight scenario recommended)

const cs = collection.watch();

cs.on('change', async (change) => {
  //Manually invalidate corresponding cache keys based on operation type
  if (['insert', 'update', 'replace', 'delete'].includes(change.operationType)) {
    myCache.delete('user-list');
    myCache.delete(`user:${change.documentKey?._id}`);
  }
});

#Solution 2: ChangeStreamSyncManager (recommended for production scenarios)

ChangeStreamSyncManager has built-in breakpoint resumption, multi-target synchronization and statistical capabilities, and can handle caching in the apply callback:

const syncManager = new MonSQLize.ChangeStreamSyncManager({
  db,
  config: {
    enabled: true,
    targets: [{
      name: 'cache-invalidation',
      apply: async (event) => {
        //Handle cache invalidation in apply
        myCache.delete('user-list');
      }
    }]
  }
});
await syncManager.start();

Cross-instance cache invalidation: use the Cache API with cache.distributed configured, then follow the deployment notes in Distributed Deployment.

#Notes

#1. MongoDB version requirements

Change Streams requires MongoDB 4.0+ and a replica set or sharded cluster environment.

Single node environment will report an error:

Error: The $changeStream stage is only supported on replica sets

Solution:

Dev/Test Environment - using mongodb-memory-server:

const msq = new MonSQLize({
  type: 'mongodb',
  databaseName: 'mydb',
  config: {
    useMemoryServer: true,
    memoryServerOptions: {
      instance: {
        replSet: 'rs0'  //Enable replica set mode
      }
    }
  }
});

await msq.connect();
const collection = msq.dbInstance.collection('users');

//✅ Watch is now available
const watcher = collection.watch();

useMemoryServer uses a single-node replica set, the binary cache is pinned at .cache/mongodb-memory-server/binaries, and the automatically created temporary dbPath is pinned at .cache/mongodb-memory-server/db and cleaned up on shutdown.

Production environment - using replica sets or sharded clusters:

const msq = new MonSQLize({
  type: 'mongodb',
  databaseName: 'mydb',
  config: {
    uri: 'mongodb://host1:27017,host2:27017,host3:27017/mydb?replicaSet=rs0'
  }
});

#2. Performance impact

• watch itself has little impact on performance (natively supported by MongoDB)
• ChangeStream monitoring is asynchronous and does not block the main process

#3. resumeToken expires

The MongoDB oplog has a size limit and the resumeToken may expire (default hours).

Handling Suggestions:

• Monitor error events and detect ChangeStreamHistoryLost errors
• Close the current ChangeStream and call collection.watch() again (without resumeAfter)
• If you need to automatically handle breakpoint resumption, please use ChangeStreamSyncManager

#4. Memory management

Long-running watches need to pay attention to:

• Correctly call watcher.close() to release resources
• Listening process signal is closed gracefully
• Don’t create too many watchers (1-2 per collection is enough)

#Troubleshooting

#Problem 1: watch closes immediately

Reason: MongoDB is not a replica set environment

Solution: Use replica set or mongodb-memory-server

#Issue 2: ChangeStream closed unexpectedly

Cause: If the network is unstable or MongoDB load is too high, ChangeStream will automatically shut down and trigger the close event.

Troubleshooting:

const cs = collection.watch();

cs.on('close', () => {
  console.warn('ChangeStream is closed, please check network and MongoDB status');
  //If you need to automatically reconnect, you can call collection.watch() again here
});

cs.on('error', (err) => {
  console.error('ChangeStream error:', err.message);
});

#Question 3: Cache integration

See the Cache Invalidation Integration chapter.

#watch event vs global event

#Difference description

monSQLize has two event systems:

1. Global event (msq object):

• Monitoring object: query operation of the application
• Event type: slow-query, query, connected, error, closed
• Applicable scenarios: performance monitoring, operation and maintenance alarms
• Document: Event System

2. watch event (ChangeStream object):

• Monitoring object: MongoDB data changes
• Event types: change, error, close, end (MongoDB native events)
• Applicable scenarios: real-time data synchronization, cache invalidation
• Documentation: This document

#Usage scenario comparison

#Example: Use both

const msq = new MonSQLize({
  type: 'mongodb',
  databaseName: 'shop',
  config: { uri: 'mongodb://localhost:27017' },
  slowQueryMs: 100
});

await msq.connect();

//✅ Monitor slow queries (operation and maintenance)
msq.on('slow-query', (meta) => {
  console.warn('Slow query:', meta.operation, meta.duration + 'ms');
  //Send alert
});

//✅ Monitor data changes (business)
const collection = msq.dbInstance.collection('products');
const cs = collection.watch();

cs.on('change', (change) => {
  console.log('Data changes:', change.operationType);
  //Cache invalidation, business notification
});

#Related documents

• MongoDB Change Streams official documentation
• Distributed Deployment Guide
• Caching System
• Event System