MongoDB $changeStream Stage

Example 1 — Sample Data

Confirm replica set mode, then seed one order. Open the change stream in a second mongosh session when you try Examples 2–3.

// Must be a replica set (or sharded cluster)
rs.status()

db.orders.drop()
db.orders.insertOne({
  _id: 1,
  product: "Keyboard",
  status: "pending",
  amount: 45
})

How the Data Is Stored in MongoDB

After the insert, MongoDB keeps one document in orders. Run find() to see it:

db.orders.find()

[
  { _id: 1, product: "Keyboard", status: "pending", amount: 45 }
]

Read the important fields like this:

How It Works

1. Check deployment. rs.status() confirms replica set mode—required for change streams.
2. Seed one document. A starting order gives you something to update in later demos.
3. Two sessions help. Keep a stream open in Session A; write in Session B.

Example 2 — Basic Watch

Goal: open a change stream, insert a new order, and print the insert event.

// Session A — open the stream
const stream = db.orders.watch()

// Session B — make a change
db.orders.insertOne({
  _id: 2,
  product: "Mouse",
  status: "pending",
  amount: 25
})

// Session A — read one event
const event = stream.next()
printjson(event)

// Same idea with the explicit stage:
// const stream = db.orders.aggregate([
//   { $changeStream: {} }
// ])

operationType: "insert"
ns: { db: "test", coll: "orders" }
documentKey: { _id: 2 }
fullDocument: {
  _id: 2,
  product: "Mouse",
  status: "pending",
  amount: 25
}
_id: { _data: "<resume-token>" }

How It Works

1. Open the cursor. watch() starts listening; it blocks until a change arrives.
2. Write in another session. The insert becomes a committed oplog entry.
3. Read the event. stream.next() returns one change document with operationType and fullDocument.
4. Save _id. That resume token lets you reconnect later with resumeAfter.

Example 3 — Filter Inserts

Goal: keep only insert events. Updates are ignored by the pipeline. Also show how updateLookup loads the full document on updates when you do want them.

// Inserts only — pipeline filters after the stream opens
const inserts = db.orders.watch([
  { $match: { operationType: "insert" } }
])

db.orders.updateOne(
  { _id: 1 },
  { $set: { status: "shipped" } }
)
// → no event on this inserts-only stream

db.orders.insertOne({
  _id: 3,
  product: "Monitor",
  status: "pending",
  amount: 180
})
// → inserts.next() returns the insert event

// When you need the full document after updates:
const withLookup = db.orders.watch([], {
  fullDocument: "updateLookup"
})
// Explicit stage form:
// db.orders.aggregate([
//   { $changeStream: { fullDocument: "updateLookup" } }
// ])

updateOne → ignored by $match
insertOne  → operationType: "insert"
             fullDocument.product: "Monitor"

// with updateLookup on a later update:
operationType: "update"
fullDocument.status: "shipped"
updateDescription.updatedFields: { status: "shipped" }

How It Works

1. Pipeline filters events. Stages after the stream act on each change document.
2. $match on operationType. Updates never reach your app on the inserts-only cursor.
3. Options vs stage. Pass fullDocument as the second argument to watch(), or inside { $changeStream: { … } }.
4. Use case. Inserts-only feeds power ETL; updateLookup powers cache/search sync.