Reacting to data changes

This section documents the current MongoDB 8.x server line as published at the MongoDB Server Manual, which is the reference these pages are written and verified against. No specific patch version is pinned. Some capabilities (Atlas Search, Atlas Vector Search, and parts of encryption and backup) are Atlas-only — they are linked, not documented in depth.

This content was generated with the assistance of AI and should be verified against the official manual before being relied on in production, since MongoDB iterates quickly.

This section’s bibliography lists the reference material consulted while preparing these pages.

A change stream is an ordered feed of the data changes happening in a collection, a database, or a whole deployment. Instead of polling, an application opens the stream and receives a change-event document for every insert, update, replace, delete, and drop as it is applied. Change streams read from the replica-set oplog, so a replica set or sharded cluster is required (see Replication).

Opening a stream with watch

db.collection.watch() opens a stream scoped to one collection. db.watch() widens it to every collection in a database, and Mongo.watch() (the deployment-level form) covers every database except the internal admin, local, and config. The returned cursor blocks until the next event is available.

// One collection
const stream = db.orders.watch();

while (stream.hasNext()) {
  const event = stream.next();
  print(`${event.operationType} on ${event.ns.coll} -> ${JSON.stringify(event.documentKey)}`);
}

See Change Streams in the manual, and db.collection.watch() for the method signature.

The change-event document

Every event is a document with a common shape. The most-used fields:

_id — the resume token for this event (see Resuming a stream).
operationType — insert, update, replace, delete, drop, rename, invalidate, and others.
clusterTime — the oplog timestamp at which the change was applied.
ns — the namespace, as { db, coll }.
documentKey — the _id (plus shard-key fields on a sharded collection) of the affected document.
fullDocument — for insert and replace, the whole new document; for update, present only when the stream was opened with fullDocument: "updateLookup".
updateDescription — for update only: { updatedFields, removedFields, truncatedArrays } describing exactly what changed.

{
  "_id": { "_data": "8264A1F2C7000000012B022C0100296E5A..." },
  "operationType": "update",
  "clusterTime": { "$timestamp": { "t": 1688472263, "i": 1 } },
  "ns": { "db": "shop", "coll": "orders" },
  "documentKey": { "_id": { "$oid": "64a1f2c7e3b0a1f2c7e3b0a1" } },
  "updateDescription": {
    "updatedFields": { "status": "SHIPPED", "shippedAt": { "$date": "2023-07-04T12:04:23Z" } },
    "removedFields": [],
    "truncatedArrays": []
  }
}

See Change Events for every field of every operation type.

Resuming a stream

Each event carries a resume token in its _id. Persist the token as you process events, and after a disconnect reopen the stream from where you left off instead of from "now":

resumeAfter: <token> — resume starting with the event after the given token. The token’s event must still be in the oplog.
startAfter: <token> — like resumeAfter, but it also works to restart after an invalidate event (for example when the watched collection was dropped and recreated).
startAtOperationTime: <timestamp> — start from a cluster time rather than a token; useful for a first start at a known point.

let resumeToken = loadSavedToken();   // application-provided
const stream = db.orders.watch([], resumeToken ? { resumeAfter: resumeToken } : {});

while (stream.hasNext()) {
  const event = stream.next();
  handle(event);
  resumeToken = event._id;
  saveToken(resumeToken);
}

Change streams require readConcern: "majority" (the effective default): an event is only delivered once its change is majority-committed and therefore cannot be rolled back. See Resume a Change Stream.

Filtering with an aggregation pipeline

watch() accepts an aggregation pipeline as its first argument. The stages run on the server against each change event, so the application receives only the events it cares about and only the fields it needs. $match, $project, $addFields, $replaceRoot, and $redact are supported.

const stream = db.orders.watch([
  { $match: {
      operationType: { $in: ["insert", "update"] },
      "fullDocument.region": "EU"
  } },
  { $project: { "fullDocument.paymentDetails": 0 } }
], { fullDocument: "updateLookup" });

See Aggregation pipeline for the stage semantics and Modify Change Stream Output.

Seeing the whole document

By default an update event carries only the delta in updateDescription. Two options add the full document:

fullDocument: "updateLookup" — attach a current copy of the document (fetched at lookup time, so it may already reflect later changes) as fullDocument.
fullDocument: "whenAvailable" / "required" together with fullDocumentBeforeChange: "whenAvailable" / "required" — include the document image from before the change as fullDocumentBeforeChange. This requires changeStreamPreAndPostImages to be enabled on the collection with collMod.

db.runCommand({ collMod: "orders", changeStreamPreAndPostImages: { enabled: true } });

const stream = db.orders.watch([], {
  fullDocument: "required",
  fullDocumentBeforeChange: "required"
});

See Lookup Full Document for Update Operations.

Typical uses

Cache invalidation — evict or refresh a cached entry the moment its source document changes.
ETL and streaming — feed a downstream warehouse, search index, or message queue with a continuous, resumable stream of changes rather than periodic full scans.
Notifications — push in-app or email alerts when a watched document reaches a state (for example an order becoming SHIPPED).

Because the stream is resumable, a consumer that stores its last token can stop and restart without losing or double-counting events, as long as the token is still within the oplog window.