MongoDB Change Streams for Real-Time Data

const { MongoClient } = require("mongodb");

async function watchOrders() {
  const client = await MongoClient.connect(process.env.MONGODB_URI);
  const db = client.db("myapp");
  const orders = db.collection("orders");

  const changeStream = orders.watch();

  changeStream.on("change", (change) => {
    console.log("Change detected:", change.operationType);
    console.log("Document:", change.fullDocument);
    console.log("Resume token:", change._id);
  });

  changeStream.on("error", (error) => {
    console.error("Change stream error:", error);
  });
}

// Only watch for specific operations and conditions
const pipeline = [
  { $match: {
    operationType: { $in: ["insert", "update"] },
    "fullDocument.status": "completed"
  }},
  { $project: {
    operationType: 1,
    "fullDocument.orderId": 1,
    "fullDocument.amount": 1,
    "fullDocument.customerId": 1
  }}
];

const changeStream = orders.watch(pipeline, {
  fullDocument: "updateLookup"  // Include full doc on updates
});

let resumeToken = loadResumeTokenFromStorage(); // Your persistence layer

async function watchWithResume() {
  const options = resumeToken
    ? { resumeAfter: resumeToken, fullDocument: "updateLookup" }
    : { fullDocument: "updateLookup" };

  const changeStream = orders.watch([], options);

  changeStream.on("change", async (change) => {
    // Process the change
    await processOrderChange(change);

    // Persist resume token for crash recovery
    resumeToken = change._id;
    await saveResumeTokenToStorage(resumeToken);
  });
}

// Watch all collections in a database
const dbChangeStream = db.watch();

// Watch entire cluster (all databases)
const clusterChangeStream = client.watch();

// Filter to specific collections
const pipeline = [
  { $match: {
    "ns.coll": { $in: ["orders", "payments", "shipments"] }
  }}
];
const filteredStream = db.watch(pipeline);