Change stream events (insert, update, delete) in MongoDB - Mini Project: Build & Apply

Track Product Inventory Changes with MongoDB Change Streams

📖 Scenario: You manage a small store's product inventory using MongoDB. You want to watch for any changes to the products collection so you can react when products are added, updated, or removed.

🎯 Goal: Create a MongoDB change stream to listen for insert, update, and delete events on the products collection.

📋 What You'll Learn

Create a products collection with initial product documents.

Define a change stream pipeline to filter for insert, update, and delete events.

Open the change stream on the products collection using the pipeline.

Add a listener to handle and log the change events.

💡 Why This Matters

🌍 Real World

Change streams help businesses monitor live changes in their databases, such as inventory updates or user activity, enabling real-time reactions.

💼 Career

Understanding change streams is valuable for roles like database administrators, backend developers, and data engineers who build reactive and event-driven applications.

1

Create the products collection with initial data

Create a products collection and insert these exact documents: { _id: 1, name: "Pen", quantity: 100 }, { _id: 2, name: "Notebook", quantity: 200 }, and { _id: 3, name: "Eraser", quantity: 150 }.

MongoDB

// Insert initial products into the products collection
// Your code here

Hint

Use db.products.insertMany() with an array of product objects.

2

Define a change stream pipeline to filter events

Create a variable called pipeline that filters change events to only include insert, update, and delete operations using { $match: { operationType: { $in: ["insert", "update", "delete"] } } }.

MongoDB

db.products.insertMany([
  { _id: 1, name: "Pen", quantity: 100 },
  { _id: 2, name: "Notebook", quantity: 200 },
  { _id: 3, name: "Eraser", quantity: 150 }
])

// Define the change stream pipeline to filter for insert, update, and delete events
// Your code here

Hint

Use const pipeline = [{ $match: { operationType: { $in: ["insert", "update", "delete"] } } }].

3

Open the change stream on the products collection

Create a variable called changeStream that opens a change stream on db.products using the pipeline variable.

MongoDB

db.products.insertMany([
  { _id: 1, name: "Pen", quantity: 100 },
  { _id: 2, name: "Notebook", quantity: 200 },
  { _id: 3, name: "Eraser", quantity: 150 }
])

const pipeline = [
  { $match: { operationType: { $in: ["insert", "update", "delete"] } } }
]

// Open the change stream on the products collection using the pipeline
// Your code here

Hint

Use const changeStream = db.products.watch(pipeline) to open the change stream.

4

Add a listener to handle and log change events

Use changeStream.on("change", (change) => { ... }) to add a listener that logs the change object to the console.

MongoDB

db.products.insertMany([
  { _id: 1, name: "Pen", quantity: 100 },
  { _id: 2, name: "Notebook", quantity: 200 },
  { _id: 3, name: "Eraser", quantity: 150 }
])

const pipeline = [
  { $match: { operationType: { $in: ["insert", "update", "delete"] } } }
]

const changeStream = db.products.watch(pipeline)

// Add a listener to the changeStream to log change events
// Your code here

Hint

Use changeStream.on("change", (change) => { console.log("Change detected:", change) }).

Practice

1. What does a MongoDB change stream primarily allow you to do?

easy

A. Backup the entire database automatically

B. Listen to real-time insert, update, and delete events on a collection

C. Run queries faster by caching results

D. Create indexes on collections

Solution

1. Step 1: Understand the purpose of change streams

   Change streams provide a way to watch for changes like inserts, updates, and deletes in real time.

2. Step 2: Compare options with the purpose

   Only Listen to real-time insert, update, and delete events on a collection describes listening to these real-time events, while others describe unrelated features.

3. Final Answer:

   Listen to real-time insert, update, and delete events on a collection -> Option B

4. Quick Check:

   Change streams = real-time event listening [OK]

Hint: Change streams = real-time watch on data changes [OK]

Common Mistakes:

• Confusing change streams with backups
• Thinking change streams speed up queries
• Mixing change streams with indexing

2. Which of the following is the correct way to start watching a collection for changes in MongoDB?

easy

A. db.collection.observe()

B. db.collection.listen()

C. db.collection.watch()

D. db.collection.monitor()

Solution

1. Step 1: Recall the MongoDB method for change streams

   The official method to watch changes is watch() on a collection object.

2. Step 2: Verify options

   Only db.collection.watch() uses the correct method name watch(). Others are invalid method names.

3. Final Answer:

   db.collection.watch() -> Option C

4. Quick Check:

   Correct method = watch() [OK]

Hint: Use .watch() to start change streams [OK]

Common Mistakes:

• Using listen() or observe() which don't exist
• Confusing with monitor() which is not a method
• Forgetting to call watch() on the collection

3. Given this code snippet, what type of change event will be logged when a document is deleted?

const changeStream = db.collection('users').watch();
changeStream.on('change', (change) => {
  console.log(change.operationType);
});
// Then a document is deleted from 'users'

medium

A. "delete"

B. "update"

C. "insert"

D. "replace"

Solution

1. Step 1: Understand operationType values in change streams

   When a document is deleted, the operationType is "delete".

2. Step 2: Match the event with the logged output

   The code logs change.operationType, so deleting a document logs "delete".

3. Final Answer:

   "delete" -> Option A

4. Quick Check:

   Delete event logs "delete" [OK]

Hint: Delete event = operationType "delete" [OK]

Common Mistakes:

• Confusing delete with update or insert
• Expecting 'remove' instead of 'delete'
• Not knowing operationType values

4. You wrote this code to watch changes but it throws an error:

const changeStream = db.collection('orders').watch;
changeStream.on('change', (change) => {
  console.log(change);
});

What is the error?

medium

A. Using 'const' instead of 'let' for changeStream

B. Incorrect event name 'change'

C. No error, code is correct

D. Missing parentheses after watch() method call

Solution

1. Step 1: Identify how watch() is called

   The code assigns db.collection('orders').watch without parentheses, so it assigns the function itself, not the stream.

2. Step 2: Understand the error cause

   Without calling watch() with parentheses, changeStream is not a stream and has no on method, causing an error.

3. Final Answer:

   Missing parentheses after watch() method call -> Option D

4. Quick Check:

   Call watch() with () to get stream [OK]

Hint: Always call watch() with parentheses () [OK]

Common Mistakes:

• Forgetting parentheses after watch
• Thinking event name 'change' is wrong
• Confusing const vs let for streams

5. You want to listen only for insert events on a collection using change streams. Which of these code snippets correctly filters for inserts only?

const changeStream = db.collection('products').watch([
  { $match: { 'operationType': 'insert' } }
]);
changeStream.on('change', (change) => {
  console.log(change.fullDocument);
});

hard

A. The code correctly filters for insert events only

B. The $match stage should use 'operation_type' instead of 'operationType'

C. You must filter inside the on('change') callback, not in watch()

D. Change streams do not support filtering by operationType

Solution

1. Step 1: Understand change stream pipeline filtering

   You can pass an aggregation pipeline to watch() to filter events, using $match on 'operationType'.

2. Step 2: Verify the filter syntax

   The code uses { $match: { 'operationType': 'insert' } }, which is the correct way to filter for insert events.

3. Final Answer:

   The code correctly filters for insert events only -> Option A

4. Quick Check:

   Filter inserts with $match on operationType [OK]

Hint: Filter inserts using $match on operationType in watch() [OK]

Common Mistakes:

• Using wrong field name like operation_type
• Filtering inside callback instead of pipeline
• Thinking filtering is not supported