Capstok — learn by doing

Why this matters

Change streams let you subscribe to a real-time stream of all insert, update, replace, and delete operations on a collection, database, or entire cluster — without polling. This is the foundation of CDC (Change Data Capture) in MongoDB: downstream systems stay in sync without dual-writes, without scheduled batch jobs, and without tight coupling between services. Change streams use the oplog under the hood and require a replica set (or sharded cluster), which means they also work in local development with a single-node replica set — a common gotcha when change streams 'work in prod but not locally'.

Demo

Watch a collection for inserts and updates in real time using a change stream.

Try it yourself

Start a change stream watcher in one terminal. In another, insert a document. Verify the change event appears including the fullDocument field. Then run an update — confirm a separate event fires with operationType: 'update'.
Add a resume token handler: store the _id of each change event. Stop the watcher, insert 3 more documents, then restart using resumeAfter: storedToken. Verify you receive the 3 missed events.
Try to open a change stream on a standalone MongoDB instance (no replica set). Note the error. Start a single-node replica set with rs.initiate() and confirm it works.

Prompt your AI

Use these three in order. Each builds on the one before.

1. Basics & terminology

What is a MongoDB change stream? What types of operations does it capture and what are the prerequisites for using it?

2. Why it works (the mechanism)

MongoDB change streams are built on the oplog. Explain what the oplog is, how it relates to replication, and why change streams require a replica set.

3. Advanced — application & what's next

I need to fan out a change stream to 5 downstream consumers without each consumer reading the oplog independently. What architecture pattern handles this, and how do resume tokens factor into fault tolerance?

References

package main

import (
	"context"; "fmt"; "time"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

func main() {
	client, _ := mongo.Connect(context.TODO(), options.Client().ApplyURI("mongodb://localhost:27017/?replicaSet=rs0"))
	col := client.Database("demo").Collection("orders")

	// Watch only inserts
	pipeline := bson.A{bson.D{{"$match", bson.D{{"operationType", "insert"}}}}}
	stream, _ := col.Watch(context.TODO(), pipeline)
	defer stream.Close(context.TODO())

	fmt.Println("Watching for inserts...")
	ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
	defer cancel()

	for stream.Next(ctx) {
		var event bson.M
		stream.Decode(&event)
		fmt.Println("New order:", event["fullDocument"])
	}
}

Run: go run main.go

Change streams and watch()