Change Streams

Change streams let your application subscribe to real-time changes happening in a collection, database, or entire deployment — without polling. They're built on the replication oplog, so they require a replica set (or sharded cluster); a standalone mongod cannot produce them.

Opening a Change Stream

// Watch a single collection
const changeStream = db.collection("orders").watch()

changeStream.on("change", (event) => {
  console.log(event)
})

// Watch an entire database
const dbStream = db.watch()

// Watch the entire deployment (all databases)
const clusterStream = client.watch()

Anatomy of a Change Event

db.orders.updateOne({ _id: orderId }, { $set: { status: "shipped" } })

{
  _id: { _data: '82655...' },              // resume token
  operationType: 'update',
  clusterTime: Timestamp({ t: 1720000000, i: 1 }),
  ns: { db: 'shop', coll: 'orders' },
  documentKey: { _id: ObjectId("...") },
  updateDescription: {
    updatedFields: { status: 'shipped' },
    removedFields: [],
    truncatedArrays: []
  }
}

Field	Meaning
`operationType`	`insert`, `update`, `replace`, `delete`, `drop`, `rename`, `invalidate`
`ns`	Namespace (database + collection) the change happened in
`documentKey`	The `_id` of the affected document
`updateDescription`	For updates: exactly which fields changed, were removed, or arrays truncated
`fullDocument`	The current full document — only present by default on inserts/replaces
`_id`	The resume token — a durable pointer to this event's position in the stream

fullDocument: "updateLookup"

By default, update events only include the delta (updateDescription), not the whole document. Pass fullDocument: "updateLookup" to have MongoDB fetch the current full document alongside every update event.

const changeStream = db.collection("orders").watch([], {
  fullDocument: "updateLookup"
})

changeStream.on("change", (event) => {
  if (event.operationType === "update") {
    console.log(event.fullDocument) // the full order document, post-update
  }
})

Warning

updateLookup re-fetches the document at read time — if it was deleted or changed again after the event fired, thefullDocument may be missing or reflect a later state. Don't treat it as an exact snapshot of the moment the update happened.

Resume Tokens — Reliability

If your application restarts or the connection drops, you can resume exactly where you left off using the last processed event's resume token, instead of missing or replaying events from the start.

let lastResumeToken = loadResumeTokenFromDisk() // however you persist it

const changeStream = db.collection("orders").watch([], {
  resumeAfter: lastResumeToken
})

changeStream.on("change", (event) => {
  processEvent(event)
  saveResumeTokenToDisk(event._id) // persist after each successful process
})

Note

MongoDB only retains oplog history for a limited window (configurable via minOplogRetentionHours on Atlas, or the oplog size for self-hosted). If your app is down longer than that window, the resume token expires and you must re-sync from a fresh snapshot.

Filtering Events with an Aggregation Pipeline

watch() accepts an aggregation pipeline to filter which events you receive server-side, instead of filtering in application code.

const changeStream = db.collection("orders").watch([
  { $match: {
      operationType: { $in: ["insert", "update"] },
      "fullDocument.status": "shipped"
  } }
], { fullDocument: "updateLookup" })

Use Case: Cache Invalidation

db.collection("products").watch().on("change", async (event) => {
  if (["update", "replace", "delete"].includes(event.operationType)) {
    await redisClient.del(`product:${event.documentKey._id}`)
  }
})

Use Case: Real-Time Notifications

db.collection("messages").watch([
  { $match: { operationType: "insert" } }
]).on("change", (event) => {
  const msg = event.fullDocument
  websocketServer.broadcast(msg.roomId, msg)
})

Use Case: ETL / Sync to Another System

Stream inserts/updates from MongoDB into a search index (Elasticsearch/Atlas Search) as they happen.
Replicate specific collections into a data warehouse for analytics without a nightly batch job.
Trigger downstream microservice events (e.g. "order shipped" → send email) directly off the data change, no separate event bus needed for simple cases.

Requirements and Limitations

Requires a replica set or sharded cluster — not available on a standalone mongod.
Requires MongoDB 3.6+ (4.0+ for whole-database watch, 4.2+ for whole-deployment watch).
Events are only available for operations that write to the oplog — mongoimport, direct file manipulation, etc. still generate events since they perform real writes.
Consumers must handle the invalidate event (collection dropped/renamed) — the stream closes and must be re-established against the new namespace if applicable.

Tip

Change streams are resumable, but they are not a durable message queue — always persist the resume token, and design consumers to be idempotent in case of reconnects replaying an event.

Summary

Change streams give push-based, real-time notifications of inserts/updates/deletes/etc. built on the oplog.
fullDocument: "updateLookup" fetches the current document alongside update events.
Resume tokens make streams resilient to disconnects — persist the last token you processed.
Common uses: cache invalidation, notifications, and lightweight ETL/sync — always requires a replica set.