MongoDB: Change Streams：实时变更监听

// === 监听集合变更 ===
const changeStream = db.products.watch();

changeStream.on('change', (change) => {
  console.log('Change detected:', change);
  // {
  //   _id: { _data: '...' },           // resumeToken（断点续传用）
  //   operationType: 'insert',          // 操作类型
  //   fullDocument: { _id: ..., sku: ..., title: ..., ... },  // 完整文档
  //   ns: { db: 'shopdb', coll: 'products' },  // 命名空间
  //   documentKey: { _id: ObjectId('...') }    // 文档主键
  // }
});

// === 过滤特定操作 ===
const changeStream = db.products.watch([
  { $match: { operationType: 'insert' } }
]);

// === 过滤特定字段 ===
const changeStream = db.products.watch([
  { $match: { 'fullDocument.category': 'Electronics' } }
]);

// === 过滤价格变化 ===
const changeStream = db.products.watch([
  {
    $match: {
      $or: [
        { operationType: 'insert' },
        { operationType: 'update', 'updateDescription.updatedFields.price': { $exists: true } }
      ]
    }
  }
]);

// === update 时返回完整文档 ===
const changeStream = db.products.watch([], {
  fullDocument: 'updateLookup'
});

changeStream.on('change', (change) => {
  if (change.operationType === 'update') {
    console.log('Updated doc:', change.fullDocument);
    // 完整文档（默认值变化后）
  }
});

// === 仅返回变更字段 ===
const changeStream = db.products.watch([], {
  fullDocument: 'whenAvailable'  // 默认
});

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

// 监听 Product 集合变更
const changeStream = Product.watch();

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

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

// === 监听新订单，发送通知 ===
const OrderStream = db.orders.watch([
  { $match: { operationType: 'insert' } }
]);

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

  // 发送邮件
  await sendEmail(order.userId, 'Order confirmation', `Order ${order._id} received`);

  // 推送通知
  await pushNotification(order.userId, {
    title: 'New Order',
    body: `Order total: $${order.total}`
  });

  // WebSocket 实时推送
  io.emit('new_order', order);
});

// === 监听 MongoDB 变更，同步到 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;
  }
});

// === 监听商品变更，清除 Redis 缓存 ===
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 的 TechCorp 系统：Change Stream 断点续传，重启后不丢事件
async function startResumableStream() {
  // 1. 从 Redis 获取上次保存的 resumeToken
  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. 启动监听
  const changeStream = db.products.watch([], options);

  changeStream.on('change', async (change) => {
    // 处理变更...
    console.log(`${change.operationType}: ${change.documentKey._id}`);

    // 3. 每次事件后保存 resumeToken
    await redis.set('product_stream_token', JSON.stringify(change._id));
  });

  changeStream.on('error', async (err) => {
    console.error('Stream error:', err.message);
    // 4. 出错后延迟重连
    setTimeout(startResumableStream, 5000);
  });
}

startResumableStream();

Resuming from saved token
Stream error:

// 场景：监听新订单，实时发送邮件 + WebSocket 推送 + 同步到 Elasticsearch

// 1. 启动监听（在 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');

  // 监听订单集合的所有变更
  const changeStream = orders.watch([
    {
      $match: {
        operationType: 'insert',  // 仅监听插入
        'fullDocument.status': 'paid'  // 仅处理已支付订单
      }
    }
  ]);

  changeStream.on('change', async (change) => {
    const order = change.fullDocument;
    console.log(`新订单: ${order._id}, 金额: $${order.total}`);

    // 1. 发送邮件通知
    await sendEmail(order.userId, {
      subject: '订单确认',
      body: `您的订单 ${order._id} 已提交，总金额 $${order.total}`
    });

    // 2. WebSocket 实时推送给商家后台
    io.to('merchant-dashboard').emit('new_order', {
      orderId: order._id,
      total: order.total,
      items: order.items,
      timestamp: order.createdAt
    });

    // 3. 同步到 Elasticsearch 供搜索
    await elasticsearch.index({
      index: 'orders',
      id: order._id.toString(),
      body: order
    });
  });

  // 4. 监听商品变更，同步更新 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. 断点续传（应用重启后恢复监听进度）
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) => {
    // 处理变更...
    // 保存 resume token
    await redis.set('change_stream_resume_token', JSON.stringify(change._id));
  });
}

startOrderListener().catch(console.error);

// 2. 在 mongosh 中手动测试
// 触发事件：插入新订单
db.orders.insertOne({
  userId: 'user_001',
  items: [{ sku: 'PHONE-001', qty: 1, price: 599 }],
  total: 599,
  status: 'paid',
  createdAt: new Date()
});
// 应用立即收到通知：发送邮件 + WebSocket 推送 + ES 索引

const { WebSocketServer } = require('ws');
const http = require('http');

// === 1. WebSocket 服务器 ===
const server = http.createServer(app);
const wss = new WebSocketServer({ server });

// 客户端连接管理：按角色分组
const clients = new Map(); // ws → { role, subscriptions }
wss.on('connection', (ws, req) => {
  const params = new URL(req.url, `http://${req.headers.host}`).searchParams;
  const role = params.get('role') || 'viewer';
  clients.set(ws, { role, subscriptions: new Set() });

  ws.on('message', (data) => {
    const msg = JSON.parse(data);
    if (msg.action === 'subscribe') {
      clients.get(ws).subscriptions.add(msg.collection);
    }
  });

  ws.on('close', () => clients.delete(ws));
});

// === 2. Change Stream 监听多个集合 ===
const collections = ['orders', 'products', 'users'];
collections.forEach(coll => {
  const stream = mongoose.connection.collection(coll).watch();

  stream.on('change', (change) => {
    const event = {
      collection: coll,
      operation: change.operationType,
      documentId: change.documentKey?._id,
      data: change.fullDocument || change.updateDescription,
      timestamp: change.clusterTime
    };

    // 广播给订阅了该集合的客户端
    broadcast(coll, event);

    // 异常检测：订单取消/退款 → 告警
    if (coll === 'orders' && change.operationType === 'update') {
      const updatedFields = change.updateDescription?.updatedFields || {};
      if (updatedFields.status === 'cancelled') {
        broadcastAlert('order_cancelled', event);
      }
    }
  });

  stream.on('error', (err) => {
    console.error(`Change Stream ${coll} 错误:`, err.message);
    // 自动重连
    setTimeout(() => startChangeStream(coll), 5000);
  });
});

// === 3. 广播与告警函数 ===
function broadcast(collection, event) {
  const message = JSON.stringify({ type: 'change', ...event });
  clients.forEach((meta, ws) => {
    if (ws.readyState === 1 && meta.subscriptions.has(collection)) {
      ws.send(message);
    }
  });
}

function broadcastAlert(alertType, event) {
  const message = JSON.stringify({ type: 'alert', alertType, ...event });
  clients.forEach((meta, ws) => {
    if (ws.readyState === 1 && meta.role === 'admin') {
      ws.send(message);
    }
  });
}

// === 4. 聚合管道过滤：只监听高价值订单 ===
const highValueStream = mongoose.connection.collection('orders').watch([
  { $match: { 'fullDocument.total': { $gte: 500 }, operationType: { $in: ['insert', 'update'] } } },
  {
    $project: {
      operationType: 1,
      documentKey: 1,
      'fullDocument.orderId': 1,
      'fullDocument.total': 1,
      'fullDocument.status': 1,
      'fullDocument.userId': 1,
      updateDescription: 1
    }
  }
]);

highValueStream.on('change', (change) => {
  broadcastAlert('high_value_order', {
    collection: 'orders',
    orderId: change.fullDocument?.orderId,
    total: change.fullDocument?.total,
    operation: change.operationType
  });
});

server.listen(3000);