A change stream is a real-time feed of every document change in a collection, database, or whole cluster. It's MongoDB's built-in change data capture (CDC) — no Kafka Connect, no debezium, no polling. Subscribe, receive insert / update / delete / replace events as they happen, with resumable tokens for crash recovery. This post covers opening change streams at every scope, filtering with an aggregation pipeline, resume tokens, full-document options (including the new fullDocumentBeforeChange), and the canonical use cases: cache invalidation, audit logs, materialised-view refresh, cross-store sync.

Series map: 00 Foundations → … → 11 Sharding → 12 Change Streams (this post) → 13 Security → … → 17 Drivers & Case Studies.

Learning Outcome

  • Open a change stream on a collection, database, or cluster.
  • Filter events with an aggregation pipeline.
  • Resume after a crash using a resume token.
  • Use fullDocument and fullDocumentBeforeChange options.
  • Apply the standard patterns: cache invalidation, audit log, materialised view, sync.

Chapter 1 — Mechanics

Change streams are implemented over the same oplog that powers replication (Post 10). The server tails the oplog, applies your aggregation filter, transforms each event into a stable shape, and streams them to your client.

Requirements:

  • Replica set (or sharded cluster). Standalone mongod doesn't have an oplog.
  • Reads from the primary by default. Can be served from secondaries with read preference (less common).

Per-event latency: typically tens of milliseconds.

Chapter 2 — Open a change stream

Collection scope

1
2
3
4
5
6
const cs = db.collection('orders').watch();
 
for await (const change of cs) {
  console.log(change);
}

Each change document looks like:

1
2
3
4
5
6
7
8
9
10
{
  "_id": { "_data": "82..." },                    // resume token
  "operationType": "insert",                       // insert | update | replace | delete | drop | rename | invalidate
  "clusterTime": Timestamp(...),
  "wallTime": ISODate(...),
  "ns": { "db": "shop", "coll": "orders" },
  "documentKey": { "_id": "o5" },
  "fullDocument": { "_id": "o5", "user_id": 1, ... }   // present for inserts
}

For updates:

1
2
3
4
5
6
7
8
9
10
{
  "operationType": "update",
  "documentKey": { "_id": "o3" },
  "updateDescription": {
    "updatedFields": { "status": "shipped" },
    "removedFields": [],
    "truncatedArrays": []
  }
}

For deletes, only documentKey is present (the full document is gone).

Database scope

1
2
const cs = client.db('shop').watch();      // all collections in shop

Cluster scope

1
2
const cs = client.watch();                 // every collection in every database

Useful for cross-DB sync + cluster-wide audit. Requires admin privilege.

Chapter 3 — Filter with a pipeline

The first argument to watch() is a normal aggregation pipeline applied to events:

1
2
3
4
5
6
7
8
const cs = db.collection('orders').watch([
  { $match: {
    operationType: { $in: ["insert", "update"] },
    "fullDocument.status": "paid"
  }},
  { $project: { _id: 1, documentKey: 1, fullDocument: 1, operationType: 1 } }
])

The server applies the filter before sending — only matching events reach you.

For update events the fullDocument is not present by default; see next chapter.

Chapter 4 — fullDocument options

1
2
3
4
db.collection('orders').watch([], {
  fullDocument: "updateLookup"
})

fullDocument values:

Value Effect
"default" only present on insert, replace
"updateLookup" re-fetches the doc as it exists now after each update (may differ from the update time if more writes happened)
"whenAvailable" (7.0+, requires changeStreamPreAndPostImages) include post-image if recorded
"required" (7.0+) error if post-image unavailable

Pre-image (the change before)

1
2
3
4
5
6
7
8
9
10
db.runCommand({
  collMod: "orders",
  changeStreamPreAndPostImages: { enabled: true }
})
 
db.collection('orders').watch([], {
  fullDocument:            "required",
  fullDocumentBeforeChange:"required"
})

Now every event includes both:

  • fullDocument — the doc as-of after the change.
  • fullDocumentBeforeChange — the doc as-of before.

Useful for: diff-based audit, before/after replication to another store, "what exactly changed".

Cost: storing pre-images increases write overhead; enable per collection only when needed.

Chapter 5 — Resume tokens

Change stream flow from writes to consumer, and resuming after a crash with the last saved resume token

Every change event has _id = a resume token. If your consumer crashes, persist the last processed token, then resume:

1
2
3
4
const cs = db.collection('orders').watch([], {
  resumeAfter: lastToken     // or startAfter: lastToken
});
  • resumeAfter — fails if the token is no longer in the oplog.
  • startAfter — like resumeAfter but tolerant of invalidate events (e.g., collection drop).
  • startAtOperationTime — start at a cluster time instead of a token.

Pattern:

1
2
3
4
5
6
7
8
9
10
11
let lastToken = await loadLastToken();
const cs = db.collection('orders').watch([], {
  resumeAfter: lastToken,
  fullDocument: 'updateLookup'
});
 
for await (const change of cs) {
  await processChange(change);
  await saveLastToken(change._id);
}

If the consumer is down longer than the oplog window (Post 10), the token expires → must re-sync from scratch.

Chapter 6 — Use case: cache invalidation

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const cs = db.collection('products').watch([], { fullDocument: 'updateLookup' });
 
for await (const change of cs) {
  const id = change.documentKey._id;
  switch (change.operationType) {
    case 'insert':
    case 'update':
    case 'replace':
      await redis.set(`product:${id}`, JSON.stringify(change.fullDocument));
      break;
    case 'delete':
      await redis.del(`product:${id}`);
      break;
  }
}

App reads from Redis; cache stays warm + accurate automatically. Better than a TTL-based cache when staleness matters.

Chapter 7 — Use case: audit log

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const cs = db.collection('orders').watch([], {
  fullDocument:            'required',
  fullDocumentBeforeChange:'required'
});
 
for await (const change of cs) {
  await db.collection('audit_orders').insertOne({
    at:           change.wallTime,
    op:           change.operationType,
    document_id:  change.documentKey._id,
    before:       change.fullDocumentBeforeChange,
    after:        change.fullDocument,
    by:           change.lsid     // session id, if available
  });
}

The audit collection becomes a tamper-evident change history (especially if combined with TTL for retention + S3 Object Lock for archived snapshots).

Chapter 8 — Use case: materialised view refresh

Pre-compute daily_revenue with $merge (Post 06), then keep it fresh via change streams:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
const cs = db.collection('orders').watch([
  { $match: { 'fullDocument.status': 'paid' } }
], { fullDocument: 'updateLookup' });
 
for await (const change of cs) {
  if (!change.fullDocument) continue;
  const day = trunc(change.fullDocument.at, 'day');
  await db.collection('daily_revenue').updateOne(
    { _id: day },
    { $inc: { revenue: change.fullDocument.total, orders: 1 } },
    { upsert: true }
  );
}

Reactive instead of batch — daily_revenue updates within seconds of the order.

For exactly-once semantics, watch out: if your consumer crashes after applying but before persisting the resume token, you'll double-count on restart. Mitigations:

  1. Idempotent updates (e.g., use the order _id to dedupe).
  2. Transactional consumer (persist token + apply update in one transaction).
  3. Outbox table on the source side (the order is the source of truth; the view is rebuildable).

Chapter 9 — Use case: cross-store sync

Stream MongoDB changes into Elasticsearch / OpenSearch / Algolia for full-text search; into Redis for caching; into BigQuery / Snowflake for analytics; into another MongoDB cluster for cross-region replication.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
const cs = db.collection('articles').watch([], { fullDocument: 'updateLookup' });
 
for await (const change of cs) {
  switch (change.operationType) {
    case 'insert':
    case 'update':
    case 'replace':
      await openSearch.index({
        index: 'articles',
        id: change.documentKey._id,
        body: change.fullDocument
      });
      break;
    case 'delete':
      await openSearch.delete({ index: 'articles', id: change.documentKey._id });
      break;
  }
}

Atlas does this natively via Atlas Triggers (Post 16) — a serverless function fires on every change event. No infra to run.

Chapter 10 — Operational concerns

Long-running consumers

A change stream client holds a cursor on the server. Treat it like a long-running connection:

  • Always wrap in error handling — network blips throw; reconnect with the last token.
  • On clean shutdown, persist the last token + close the cursor.
  • Use one consumer per collection per business purpose. Don't share streams across handlers.

Performance impact on the cluster

Change streams read from the oplog — extra reads, but typically cheap. Many streams = many oplog readers = some CPU + memory cost on the primary. Atlas handles thousands of streams comfortably.

Pre-image cost

Recording pre-images doubles per-update write work + storage. Enable only where the audit / before-image is needed.

Sharded clusters

Change streams on sharded clusters merge events from every shard, ordered by cluster time. Latency a touch higher than on a single replica set.

Chapter 11 — Limits

Limit Default
Single event size 16 MB (BSON)
Resume window depends on oplog window (Post 10)
Pipeline operators subset of full aggregation; basic $match, $project, $addFields, $replaceRoot, $replaceWith, $redact
Pre/post images per-collection opt-in; cost on writes

Large updates may produce events too big — adjust your pipeline to project only what consumers need.

Chapter 12 — Atlas Triggers (preview)

For serverless use, Atlas runs your code on every change event:

1
2
3
4
5
6
// Atlas Trigger function — runs on each change
exports = async function(changeEvent) {
  const fullDoc = changeEvent.fullDocument;
  // ... your logic
};

No infrastructure; pay per invocation. Post 16 covers Triggers + Atlas Functions deeply.

For local / self-hosted, the long-running consumer pattern (Chapter 5) is the equivalent.

Chapter 13 — Common pitfalls

Pitfall Cause Fix
Resume fails with "token expired" consumer down > oplog window enlarge oplog; or accept full re-sync from startAtOperationTime
Missing fullDocument on updates default mode use fullDocument: 'updateLookup' (or required pre-image)
Pre-image undefined collection not enabled collMod with changeStreamPreAndPostImages
Double-processed events on crash persisted token after apply but apply non-idempotent idempotent updates OR transactional consumer
Big event (> 16 MB) full-document mode + huge doc project subset before sending
Stream slow on busy collection lots of unrelated events; client filters them move filter to the pipeline ($match) — server side
invalidate event collection dropped or renamed use startAfter instead of resumeAfter
One consumer for everything bottleneck one consumer per use case

Practice Exercises

Exercise 1 — Print every change

Open a change stream on shop.orders. In another window insert + update + delete docs. Watch events print.

Exercise 2 — Resume token

Persist the last token to a JSON file after each event. Kill the consumer. Insert a few docs. Restart; confirm catch-up.

Exercise 3 — Cache invalidator

Run a Node consumer that pushes product changes into a Map (in-memory cache). Read from the cache in another script. Update a product; confirm the cache reflects the change in < 1 s.

Exercise 4 — Audit log

Enable pre-images on orders. Build the audit collection from Chapter 7. Update an order; inspect the audit doc.

Exercise 5 — Materialised view

Pipe paid-order events into daily_revenue. Insert a few paid orders for the same day; confirm daily_revenue totals match.

Exercise 6 — Stream-to-Elasticsearch (simulation)

Print to stdout in OpenSearch-bulk format every article change. Validate end-to-end pattern.

Cheat-sheet

What How
Open collection db.coll.watch([pipeline], options)
Open db client.db('x').watch()
Open cluster client.watch()
Iterate for await (const e of cs) { ... }
Full doc on updates { fullDocument: 'updateLookup' }
Pre-image enable on collection + fullDocumentBeforeChange: 'required'
Resume after crash persist change._id; pass resumeAfter (or startAfter)
From cluster time { startAtOperationTime: ts }
Filter events [{ $match: { ... } }] first stage
Single event size 16 MB cap
One consumer per business purpose; not shared
Atlas Triggers for serverless event handling
Sharded events ordered by cluster time across shards

What's next

Post 13 covers security — authentication (SCRAM, x.509, LDAP, IAM), authorization (RBAC + custom roles), TLS, encryption at rest, Client-Side Field-Level Encryption (CSFLE), and Queryable Encryption (GA in 7.0).

References

Note: this post is in progress. Planned additions: outbox-pattern walkthrough; multi-region cross-cluster sync; bench numbers on pre-image overhead.