MongoDB: تدفقات التغييرات: مراقبة التغييرات في الوقت الفعلي

// === Monitor Set Changes ===
const changeStream = db.products.watch();

changeStream.on('change', (change) => {
  console.log('Change detected:', change);
  // {
  //   _id: { _data: '...' },           // resumeToken(For resuming downloads)
  //   operationType: 'insert',          // Operation Type
  //   fullDocument: { _id: ..., sku: ..., title: ..., ... },  // Complete Documentation
  //   ns: { db: 'shopdb', coll: 'products' },  // Namespace
  //   documentKey: { _id: ObjectId('...') }    // Document Primary Key
  // }
});

// === Filter by Specific Actions ===
const changeStream = db.products.watch([
  { $match: { operationType: 'insert' } }
]);

// === Filter by Specific Fields ===
const changeStream = db.products.watch([
  { $match: { 'fullDocument.category': 'Electronics' } }
]);

// === Filter by Price Changes ===
const changeStream = db.products.watch([
  {
    $match: {
      $or: [
        { operationType: 'insert' },
        { operationType: 'update', 'updateDescription.updatedFields.price': { $exists: true } }
      ]
    }
  }
]);

// === update Return to the full document ===
const changeStream = db.products.watch([], {
  fullDocument: 'updateLookup'
});

changeStream.on('change', (change) => {
  if (change.operationType === 'update') {
    console.log('Updated doc:', change.fullDocument);
    // Complete Documentation(After the default value changes)
  }
});

// === Return only the fields that have changed ===
const changeStream = db.products.watch([], {
  fullDocument: 'whenAvailable'  // Default
});

// === mongoose Change Streams(mongoose 6+)===
const Product = mongoose.model('Product', productSchema);

// Monitoring Product Set Changes
const changeStream = Product.watch();

changeStream.on('change', (change) => {
  console.log(`${change.operationType}:`, change.fullDocument);
});

// Filter
const filteredStream = Product.watch([
  { $match: { operationType: { $in: ['insert', 'update'] } } }
]);

// === Monitor New Orders，Send a notification ===
const OrderStream = db.orders.watch([
  { $match: { operationType: 'insert' } }
]);

OrderStream.on('change', async (change) => {
  const order = change.fullDocument;

  // Send an Email
  await sendEmail(order.userId, 'Order confirmation', `Order ${order._id} received`);

  // Push Notifications
  await pushNotification(order.userId, {
    title: 'New Order',
    body: `Order total: $${order.total}`
  });

  // WebSocket Real-time Notifications
  io.emit('new_order', order);
});

// === Monitoring MongoDB Change，Sync to Elasticsearch ===
const ProductStream = db.products.watch();

ProductStream.on('change', async (change) => {
  switch (change.operationType) {
    case 'insert':
    case 'update':
    case 'replace':
      await elasticsearch.index({
        index: 'products',
        id: change.documentKey._id.toString(),
        body: change.fullDocument
      });
      break;
    case 'delete':
      await elasticsearch.delete({
        index: 'products',
        id: change.documentKey._id.toString()
      });
      break;
  }
});

// === Monitor Product Changes，Invalidate Redis cache ===
const ProductStream = db.products.watch();

ProductStream.on('change', async (change) => {
  const productId = change.documentKey._id.toString();
  await redis.del(`product:${productId}`);
  console.log(`Cache cleared for ${productId}`);
});

// Alice's TechCorp System: Change Stream Resume, Events Are Not Lost After a Restart
async function startResumableStream() {
  // 1. Retrieve last saved resumeToken from Redis
  let resumeToken = await redis.get('product_stream_token');
  let options = { fullDocument: 'updateLookup' };

  if (resumeToken) {
    options.resumeAfter = JSON.parse(resumeToken);
    console.log('Resuming from saved token');
  }

  // 2. Start Listening
  const changeStream = db.products.watch([], options);

  changeStream.on('change', async (change) => {
    // Processing Changes...
    console.log(`${change.operationType}: ${change.documentKey._id}`);

    // 3. Save after each event resumeToken
    await redis.set('product_stream_token', JSON.stringify(change._id));
  });

  changeStream.on('error', async (err) => {
    console.error('Stream error:', err.message);
    // 4. Delayed Reconnection After an Error
    setTimeout(startResumableStream, 5000);
  });
}

startResumableStream();

// Scene: Monitor New Orders, Send emails in real time + WebSocket Push + Sync to Elasticsearch

// 1. Start Listening (in Node.js)
const { MongoClient } = require('mongodb');

async function startOrderListener() {
  const client = new MongoClient('mongodb://localhost:27017/?replicaSet=rs0');
  await client.connect();
  const orders = client.db('shopdb').collection('orders');

  // Monitor all changes to the order collection
  const changeStream = orders.watch([
    {
      $match: {
        operationType: 'insert',  // Listen-only insertion
        'fullDocument.status': 'paid'  // Process only paid orders
      }
    }
  ]);

  changeStream.on('change', async (change) => {
    const order = change.fullDocument;
    console.log(`New Orders: ${order._id}, Amount: $${order.total}`);

    // 1. Send an email notification
    await sendEmail(order.userId, {
      subject: 'Order Confirmation',
      body: `Your Order ${order._id} Submitted，Total Amount $${order.total}`
    });

    // 2. WebSocket Real-time push notifications to the merchant's backend
    io.to('merchant-dashboard').emit('new_order', {
      orderId: order._id,
      total: order.total,
      items: order.items,
      timestamp: order.createdAt
    });

    // 3. Sync to Elasticsearch For Search
    await elasticsearch.index({
      index: 'orders',
      id: order._id.toString(),
      body: order
    });
  });

  // 4. Monitor Product Changes，Synchronized Update Elasticsearch
  const productStream = client.db('shopdb').collection('products').watch();
  productStream.on('change', async (change) => {
    if (change.operationType === 'delete') {
      await elasticsearch.delete({
        index: 'products',
        id: change.documentKey._id.toString()
      });
    } else if (change.fullDocument) {
      await elasticsearch.index({
        index: 'products',
        id: change.documentKey._id.toString(),
        body: change.fullDocument
      });
    }
  });

  console.log('Change Streams listening...');
}

// 5. Resume Download(Restore progress tracking after the app restarts)
async function resumeAfterRestart() {
  const resumeToken = await redis.get('change_stream_resume_token');
  const changeStream = orders.watch([], {
    resumeAfter: JSON.parse(resumeToken),
    fullDocument: 'updateLookup'
  });

  changeStream.on('change', async (change) => {
    // Processing Changes...
    // Save resume token
    await redis.set('change_stream_resume_token', JSON.stringify(change._id));
  });
}

startOrderListener().catch(console.error);

// 2. Manual Testing in mongosh
// Trigger Event:Insert a New Order
db.orders.insertOne({
  userId: 'user_001',
  items: [{ sku: 'PHONE-001', qty: 1, price: 599 }],
  total: 599,
  status: 'paid',
  createdAt: new Date()
});
// The app receives a notification immediately:Send an Email + WebSocket Push + ES Index

// نظام TechCorp لمراقبة المخزون: تنبيه فوري عند نفاد المخزون
const Product = mongoose.model('Product', new mongoose.Schema({
  sku: String,
  title: String,
  stock: Number,
  lowStockThreshold: { type: Number, default: 10 }
}));

// مراقبة تحديثات المخزون
const stockStream = Product.watch([
  {
    $match: {
      operationType: 'update',
      'updateDescription.updatedFields.stock': { $exists: true }
    }
  }
], { fullDocument: 'updateLookup' });

stockStream.on('change', async (change) => {
  const product = change.fullDocument;

  if (product.stock <= product.lowStockThreshold) {
    // إرسال تنبيه للمخزون المنخفض
    console.log(`⚠️ تحذير: المنتج ${product.sku} - ${product.title}`);
    console.log(`   المخزون الحالي: ${product.stock}`);
    console.log(`   الحد الأدنى: ${product.lowStockThreshold}`);

    // إرسال بريد إلكتروني للمشرف
    await sendEmail('inventory@techcorp.com', {
      subject: `تحذير: مخزون منخفض - ${product.sku}`,
      body: `المنتج "${product.title}" وصل إلى ${product.stock} وحدة فقط`
    });

    // إرسال إشعار فوري للوحة التحكم
    io.emit('low_stock_alert', {
      sku: product.sku,
      title: product.title,
      currentStock: product.stock,
      threshold: product.lowStockThreshold
    });
  }
});

// محاكاة تحديث المخزون
await Product.updateOne(
  { sku: 'PHONE-001' },
  { $inc: { stock: -50 } }
);