MongoDB: Change Streams：リアルタイム変更監視

// === コレクション変更を監視 ===
const changeStream = db.products.watch();

changeStream.on('change', (change) => {
  console.log('変更を検出:', change);
});

// === 特定の操作でフィルタリング ===
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('更新されたドキュメント:', change.fullDocument);
    // 完全なドキュメント（デフォルトは変更後の値）
  }
});

// === 変更フィールドのみを返す ===
const changeStream = db.products.watch([], {
  fullDocument: 'whenAvailable'  // デフォルト
});

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

// 商品コレクションの変更を監視
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._id} を受領しました`);

  // プッシュ通知
  await pushNotification(order.userId, {
    title: '新規注文',
    body: `注文合計: $${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(`${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('保存されたトークンから再開');
  }

  // 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('ストリームエラー:', err.message);
    // 4. エラー後、遅延再接続
    setTimeout(startResumableStream, 5000);
  });
}

startResumableStream();

// シーン：新規注文を監視、リアルタイムでメール送信 + 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監視中...');
}

// 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インデックス

// シーン：ShopHub管理ダッシュボードでライブ注文統計を表示
const mongoose = require('mongoose');

async function startOrderAnalytics() {
  const Order = mongoose.model('Order');

  // インメモリで日次注文数と売上を追跡（ダッシュボード用）
  let stats = {
    date: new Date().toISOString().split('T')[0],
    orderCount: 0,
    totalRevenue: 0,
    lastOrderId: null
  };

  // 挿入操作のみ監視（新規注文）
  const changeStream = Order.watch([
    { $match: { operationType: 'insert' } }
  ], { fullDocument: true });

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

    // インメモリ統計を更新
    stats.orderCount++;
    stats.totalRevenue += order.total || 0;
    stats.lastOrderId = order._id;

    // WebSocketクライアントにブロードキャスト（ダッシュボード）
    broadcastToDashboard({
      event: 'new_order',
      data: {
        orderId: order._id,
        total: order.total,
        stats: stats
      }
    });

    // デモ用にコンソールに出力
    console.log(`[LIVE] 注文 #${order.orderNumber} - $${order.total}`);
    console.log(`   本日: ${stats.orderCount}件の注文、$${stats.totalRevenue}の売上`);
  });

  changeStream.on('error', (err) => {
    console.error('Change Streamエラー:', err);
    // 再接続ロジックをここに追加
  });

  console.log('注文分析ストリームを開始...');
}

// WebSocketブロードキャストをシミュレート
function broadcastToDashboard(data) {
  // 本番環境: io.emit('order_update', data)
}

startOrderAnalytics();