Change Streams

// ===== Requirements =====
// - Replica set or sharded cluster (NOT standalone)
// - Driver that supports change streams (Node, Python, Go, Java, etc)

// ===== Watch a collection =====
const orders = db.collection('orders');
const stream = orders.watch();

for await (const change of stream) {
  console.log(change);
  /*
  {
    _id: { _data: '...' },   // resume token
    operationType: 'insert' | 'update' | 'replace' | 'delete' | 'invalidate' | 'rename',
    fullDocument: { ... },
    documentKey: { _id: ... },
    updateDescription: { updatedFields: {...}, removedFields: [...] },
    clusterTime: Timestamp,
    ns: { db: 'shop', coll: 'orders' },
  }
  */
}

// ===== Filter changes =====
const pipeline = [
  { $match: {
      operationType: { $in: ['insert', 'update'] },
      'fullDocument.status': 'paid',
  } },
];
const stream2 = orders.watch(pipeline, { fullDocument: 'updateLookup' });

// fullDocument: 'updateLookup' includes the document after update (extra cost).

// ===== Resume after disconnect =====
let resumeToken;
const stream3 = orders.watch([], { resumeAfter: resumeToken });
for await (const change of stream3) {
  resumeToken = change._id;   // save for restart
  // ...
}

// On restart, pass resumeAfter or startAfter to continue from where you stopped.

// ===== Watch at database / cluster level =====
db.watch();              // all collections in db
client.watch();          // all changes in the cluster

// ===== Use cases =====
// - Cache invalidation (Redis tied to Mongo)
// - Sync to a search index (Elastic / Meilisearch)
// - Real-time UI updates via WebSocket / SSE
// - Event-driven workflows (transactional outbox alternative)

// ===== Performance =====
// - Each watcher consumes an oplog tail; oplog sized for retention window
// - Heavy aggregation pipelines in watch() add CPU per change
// - Distributed worker should partition by collection or shard

// ===== Patterns =====
// - Persist resume token in durable store (DB / Redis)
// - Heartbeat / observe NO-OP events to verify liveness
// - Limit projection ({ $project } in the pipeline) to reduce bandwidth
// - One watcher per logical concern; do not multiplex

// ===== Pitfalls =====
// - Standalone server (change streams require replica set)
// - Long pause -> token expires -> needs full reset (lost events)
// - Heavy filtering done client-side instead of in the watch pipeline
// - Multiple watchers fighting for the same updates without coordination

const stream = db.collection('orders').watch();
stream.on('change', e => console.log(e.operationType, e.fullDocument));