コレクションでの変更ストリームの開始

// Verify you're on a replica set before using change streams
// In mongosh:
rs.status() // should show replica set members, not an error

// If running locally without a replica set, start one:
// mongod --replSet rs0 --port 27017 --dbpath /data/db
// Then in mongosh: rs.initiate()

const { MongoClient } = require('mongodb');
const client = new MongoClient(process.env.MONGODB_URI);

async function watchCollection() {
  const db = client.db('ecommerce');
  const orders = db.collection('orders');

  // Open a change stream on the orders collection
  const changeStream = orders.watch();

  console.log('Watching for changes...');
  // The stream is now open and ready to deliver events
  return changeStream;
}

async function processChanges() {
  const changeStream = db.collection('orders').watch();
  try {
    for await (const change of changeStream) {
      console.log('Change received:', change.operationType);
      console.log('Document key:', change.documentKey._id);
      // Handle the change event
      await handleOrderChange(change);
    }
  } finally {
    await changeStream.close();
  }
}

const changeStream = db.collection('products').watch();

changeStream.on('change', (event) => {
  console.log('Product changed:', event.operationType, event.documentKey._id);
  if (event.operationType === 'update') {
    invalidateProductCache(event.documentKey._id);
  }
});

changeStream.on('error', (error) => {
  console.error('Change stream error:', error);
  // Implement reconnect logic or close gracefully
});

// Close when done
// changeStream.close();

// Collection-level (most common)
const stream1 = db.collection('orders').watch();

// Database-level — all collections in 'ecommerce'
const stream2 = client.db('ecommerce').watch();

// Client-level — everything in the deployment
const stream3 = client.watch();

// Events at db/client scope include 'ns' field to identify which collection changed
for await (const change of stream2) {
  console.log('Changed collection:', change.ns.coll);
}

// Receive the full document in update events
const changeStream = db.collection('users').watch(
  [], // empty pipeline = all events
  { fullDocument: 'updateLookup' }
);

for await (const change of changeStream) {
  if (change.operationType === 'update') {
    // change.fullDocument is now the complete updated user document
    console.log('Updated user:', change.fullDocument.email);
    await syncToSearchIndex(change.fullDocument);
  }
}

// Server: push new orders to dashboard clients via WebSocket
async function startOrderWatcher(io) { // io = socket.io instance
  const changeStream = db.collection('orders').watch([
    { $match: { operationType: 'insert' } }
  ]);

  for await (const change of changeStream) {
    const newOrder = change.fullDocument;
    // Broadcast to all connected dashboard clients
    io.to('dashboard').emit('newOrder', {
      id: newOrder._id,
      customer: newOrder.customerId,
      amount: newOrder.total,
      timestamp: newOrder.createdAt
    });
  }
}

// Inventory service reacts to confirmed orders
async function inventoryWatcher() {
  const changeStream = db.collection('orders').watch([
    {
      $match: {
        operationType: 'update',
        'updateDescription.updatedFields.status': 'confirmed'
      }
    }
  ], { fullDocument: 'updateLookup' });

  for await (const change of changeStream) {
    const order = change.fullDocument;
    for (const item of order.items) {
      await reserveInventory(item.productId, item.quantity);
    }
  }
}

const changeStream = db.collection('events').watch();

// Graceful shutdown
process.on('SIGTERM', async () => {
  console.log('Shutting down...');
  await changeStream.close();
  await client.close();
  process.exit(0);
});

// Handle SIGINT (Ctrl+C in development)
process.on('SIGINT', async () => {
  await changeStream.close();
  await client.close();
  process.exit(0);
});