MongoDB Change Streams: Real-Time Data Synchronization with SQL-Style Event Processing

-- SQL polling approach - inefficient and delayed
-- Application repeatedly checks for changes
SELECT 
  order_id,
  status,
  updated_at,
  customer_id
FROM orders
WHERE updated_at > '2025-09-05 10:00:00'
  AND status IN ('pending', 'processing')
ORDER BY updated_at DESC;

-- Problems with polling:
-- - Constant database load from repeated queries
-- - Delay between actual change and detection
-- - Missed changes between polling intervals
-- - No differentiation between insert/update/delete operations
-- - Scaling issues with high-frequency changes

-- Trigger-based approaches - complex maintenance
CREATE OR REPLACE FUNCTION notify_order_change()
RETURNS TRIGGER AS $$
BEGIN
  IF TG_OP = 'INSERT' THEN
    PERFORM pg_notify('order_changes', json_build_object(
      'operation', 'insert',
      'order_id', NEW.order_id,
      'status', NEW.status
    )::text);
  ELSIF TG_OP = 'UPDATE' THEN
    PERFORM pg_notify('order_changes', json_build_object(
      'operation', 'update', 
      'order_id', NEW.order_id,
      'old_status', OLD.status,
      'new_status', NEW.status
    )::text);
  END IF;
  RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

-- Problems: Complex setup, maintenance overhead, limited filtering

// MongoDB Change Streams - efficient real-time data monitoring
const changeStream = db.collection('orders').watch([
  {
    $match: {
      'operationType': { $in: ['insert', 'update', 'delete'] },
      'fullDocument.status': { $in: ['pending', 'processing', 'shipped'] }
    }
  }
]);

changeStream.on('change', (change) => {
  console.log('Real-time change detected:', {
    operationType: change.operationType,
    documentKey: change.documentKey,
    fullDocument: change.fullDocument,
    updateDescription: change.updateDescription,
    timestamp: change.clusterTime
  });

  // React immediately to changes
  handleOrderStatusChange(change);
});

// Benefits:
// - Zero polling overhead - push-based notifications
// - Immediate change detection with sub-second latency
// - Rich change metadata including operation type and modified fields
// - Efficient filtering at the database level
// - Automatic resume capability for fault tolerance
// - Scalable across replica sets and sharded clusters

// Change Stream implementation for various scenarios
class ChangeStreamManager {
  constructor(db) {
    this.db = db;
    this.activeStreams = new Map();
  }

  async watchCollection(collectionName, pipeline = [], options = {}) {
    const collection = this.db.collection(collectionName);

    const changeStreamOptions = {
      fullDocument: options.includeFullDocument ? 'updateLookup' : 'default',
      fullDocumentBeforeChange: options.includeBeforeDocument ? 'whenAvailable' : 'off',
      resumeAfter: options.resumeToken || null,
      startAtOperationTime: options.startTime || null,
      maxAwaitTimeMS: options.maxAwaitTime || 1000,
      batchSize: options.batchSize || 1000
    };

    const changeStream = collection.watch(pipeline, changeStreamOptions);

    // Store stream reference for management
    const streamId = `${collectionName}_${Date.now()}`;
    this.activeStreams.set(streamId, {
      stream: changeStream,
      collection: collectionName,
      pipeline: pipeline,
      startedAt: new Date()
    });

    return { streamId, changeStream };
  }

  async watchDatabase(pipeline = [], options = {}) {
    // Watch changes across entire database
    const changeStream = this.db.watch(pipeline, {
      fullDocument: options.includeFullDocument ? 'updateLookup' : 'default',
      fullDocumentBeforeChange: options.includeBeforeDocument ? 'whenAvailable' : 'off'
    });

    const streamId = `database_${Date.now()}`;
    this.activeStreams.set(streamId, {
      stream: changeStream,
      scope: 'database',
      startedAt: new Date()
    });

    return { streamId, changeStream };
  }

  async setupOrderProcessingStream() {
    // Real-time order processing workflow
    const pipeline = [
      {
        $match: {
          $or: [
            // New orders created
            {
              'operationType': 'insert',
              'fullDocument.status': 'pending'
            },
            // Order status updates
            {
              'operationType': 'update',
              'updateDescription.updatedFields.status': { $exists: true }
            },
            // Order cancellations
            {
              'operationType': 'update',
              'updateDescription.updatedFields.cancelled': true
            }
          ]
        }
      },
      {
        $project: {
          operationType: 1,
          documentKey: 1,
          fullDocument: 1,
          updateDescription: 1,
          clusterTime: 1,
          // Add computed fields for processing
          orderValue: '$fullDocument.total_amount',
          customerId: '$fullDocument.customer_id',
          priorityLevel: {
            $switch: {
              branches: [
                { case: { $gt: ['$fullDocument.total_amount', 1000] }, then: 'high' },
                { case: { $gt: ['$fullDocument.total_amount', 500] }, then: 'medium' }
              ],
              default: 'normal'
            }
          }
        }
      }
    ];

    const { streamId, changeStream } = await this.watchCollection('orders', pipeline, {
      includeFullDocument: true,
      includeBeforeDocument: true
    });

    changeStream.on('change', (change) => {
      this.processOrderChange(change);
    });

    changeStream.on('error', (error) => {
      console.error('Change stream error:', error);
      this.handleStreamError(streamId, error);
    });

    return streamId;
  }

  async processOrderChange(change) {
    const { operationType, fullDocument, updateDescription, priorityLevel } = change;

    try {
      switch (operationType) {
        case 'insert':
          // New order created
          await this.handleNewOrder(fullDocument, priorityLevel);
          break;

        case 'update':
          // Order modified
          await this.handleOrderUpdate(fullDocument, updateDescription, priorityLevel);
          break;

        case 'delete':
          // Order deleted (rare but handle gracefully)
          await this.handleOrderDeletion(change.documentKey);
          break;
      }
    } catch (error) {
      console.error('Failed to process order change:', error);
      // Implement dead letter queue or retry logic
      await this.queueFailedChange(change, error);
    }
  }

  async handleNewOrder(order, priority) {
    console.log(`New ${priority} priority order: ${order._id}`);

    // Trigger immediate actions for new orders
    const actions = [];

    // Inventory reservation
    actions.push(this.reserveInventory(order._id, order.items));

    // Payment processing for high-priority orders
    if (priority === 'high') {
      actions.push(this.expeditePaymentProcessing(order._id));
    }

    // Customer notification
    actions.push(this.notifyCustomer(order.customer_id, 'order_created', order._id));

    // Fraud detection for large orders
    if (order.total_amount > 2000) {
      actions.push(this.triggerFraudCheck(order._id));
    }

    await Promise.allSettled(actions);
  }

  async handleOrderUpdate(order, updateDescription, priority) {
    const updatedFields = updateDescription.updatedFields || {};

    // React to specific field changes
    if ('status' in updatedFields) {
      await this.handleStatusChange(order._id, updatedFields.status, priority);
    }

    if ('shipping_address' in updatedFields) {
      await this.updateShippingCalculations(order._id, updatedFields.shipping_address);
    }

    if ('items' in updatedFields) {
      await this.recalculateOrderTotal(order._id, updatedFields.items);
    }
  }

  async handleStatusChange(orderId, newStatus, priority) {
    const statusActions = {
      'confirmed': [
        () => this.initiateFullfillment(orderId),
        () => this.updateInventory(orderId, 'reserved'),
        () => this.sendCustomerNotification(orderId, 'order_confirmed')
      ],
      'shipped': [
        () => this.generateTrackingNumber(orderId),
        () => this.updateInventory(orderId, 'shipped'),
        () => this.sendShipmentNotification(orderId),
        () => this.scheduleDeliveryWindow(orderId)
      ],
      'delivered': [
        () => this.finalizeOrder(orderId),
        () => this.updateInventory(orderId, 'delivered'),
        () => this.requestCustomerFeedback(orderId),
        () => this.triggerRecommendations(orderId)
      ],
      'cancelled': [
        () => this.releaseReservedInventory(orderId),
        () => this.processRefund(orderId),
        () => this.sendCancellationNotification(orderId)
      ]
    };

    const actions = statusActions[newStatus] || [];

    if (actions.length > 0) {
      console.log(`Processing ${newStatus} status change for order ${orderId} (${priority} priority)`);

      // Execute high-priority orders first
      if (priority === 'high') {
        for (const action of actions) {
          await action();
        }
      } else {
        await Promise.allSettled(actions.map(action => action()));
      }
    }
  }

  async setupInventoryMonitoring() {
    // Real-time inventory level monitoring
    const pipeline = [
      {
        $match: {
          'operationType': 'update',
          'updateDescription.updatedFields.quantity': { $exists: true }
        }
      },
      {
        $addFields: {
          currentQuantity: '$fullDocument.quantity',
          previousQuantity: {
            $subtract: [
              '$fullDocument.quantity',
              '$updateDescription.updatedFields.quantity'
            ]
          },
          quantityChange: '$updateDescription.updatedFields.quantity',
          productId: '$fullDocument.product_id',
          threshold: '$fullDocument.reorder_threshold'
        }
      },
      {
        $match: {
          $or: [
            // Low stock alert
            { $expr: { $lt: ['$currentQuantity', '$threshold'] } },
            // Out of stock
            { currentQuantity: 0 },
            // Large quantity changes (potential issues)
            { $expr: { $gt: [{ $abs: '$quantityChange' }, 100] } }
          ]
        }
      }
    ];

    const { streamId, changeStream } = await this.watchCollection('inventory', pipeline, {
      includeFullDocument: true
    });

    changeStream.on('change', (change) => {
      this.processInventoryChange(change);
    });

    return streamId;
  }

  async processInventoryChange(change) {
    const { currentQuantity, threshold, productId, quantityChange } = change;

    if (currentQuantity === 0) {
      // Out of stock - immediate action required
      await this.handleOutOfStock(productId);
    } else if (currentQuantity <= threshold) {
      // Low stock warning
      await this.triggerReorderAlert(productId, currentQuantity, threshold);
    }

    // Detect unusual quantity changes
    if (Math.abs(quantityChange) > 100) {
      await this.flagUnusualInventoryChange(productId, quantityChange, change.clusterTime);
    }

    // Update real-time inventory dashboard
    await this.updateInventoryDashboard(productId, currentQuantity);
  }

  async handleStreamError(streamId, error) {
    console.error(`Change stream ${streamId} encountered error:`, error);

    const streamInfo = this.activeStreams.get(streamId);

    if (streamInfo) {
      // Close errored stream
      streamInfo.stream.close();

      // Attempt to resume from last known position
      if (error.resumeToken) {
        console.log(`Attempting to resume stream ${streamId}`);

        const resumeOptions = {
          resumeAfter: error.resumeToken,
          includeFullDocument: true
        };

        const { streamId: newStreamId, changeStream } = await this.watchCollection(
          streamInfo.collection,
          streamInfo.pipeline,
          resumeOptions
        );

        // Update stream reference
        this.activeStreams.delete(streamId);
        console.log(`Stream ${streamId} resumed as ${newStreamId}`);
      }
    }
  }

  async closeStream(streamId) {
    const streamInfo = this.activeStreams.get(streamId);

    if (streamInfo) {
      await streamInfo.stream.close();
      this.activeStreams.delete(streamId);
      console.log(`Stream ${streamId} closed successfully`);
    }
  }

  async closeAllStreams() {
    const closePromises = Array.from(this.activeStreams.keys()).map(
      streamId => this.closeStream(streamId)
    );

    await Promise.allSettled(closePromises);
    console.log('All change streams closed');
  }

  // Placeholder methods for business logic
  async reserveInventory(orderId, items) { /* Implementation */ }
  async expeditePaymentProcessing(orderId) { /* Implementation */ }
  async notifyCustomer(customerId, event, orderId) { /* Implementation */ }
  async triggerFraudCheck(orderId) { /* Implementation */ }
  async initiateFullfillment(orderId) { /* Implementation */ }
  async updateInventory(orderId, status) { /* Implementation */ }
  async sendCustomerNotification(orderId, type) { /* Implementation */ }
  async generateTrackingNumber(orderId) { /* Implementation */ }
  async handleOutOfStock(productId) { /* Implementation */ }
  async triggerReorderAlert(productId, current, threshold) { /* Implementation */ }
  async updateInventoryDashboard(productId, quantity) { /* Implementation */ }
}

// Advanced change stream patterns for complex scenarios
class AdvancedChangeStreamProcessor {
  constructor(db) {
    this.db = db;
    this.streamProcessors = new Map();
    this.changeBuffer = [];
    this.batchProcessor = null;
  }

  async setupMultiCollectionWorkflow() {
    // Coordinate changes across multiple related collections
    const collections = ['users', 'orders', 'inventory', 'payments'];
    const streams = [];

    for (const collectionName of collections) {
      const pipeline = [
        {
          $match: {
            'operationType': { $in: ['insert', 'update', 'delete'] }
          }
        },
        {
          $addFields: {
            sourceCollection: collectionName,
            changeId: { $toString: '$_id' },
            timestamp: '$clusterTime'
          }
        }
      ];

      const changeStream = this.db.collection(collectionName).watch(pipeline);

      changeStream.on('change', (change) => {
        this.processMultiCollectionChange(change);
      });

      streams.push({ collection: collectionName, stream: changeStream });
    }

    return streams;
  }

  async processMultiCollectionChange(change) {
    const { sourceCollection, operationType, documentKey, fullDocument } = change;

    // Implement cross-collection business logic
    switch (sourceCollection) {
      case 'users':
        if (operationType === 'insert') {
          await this.handleNewUserRegistration(fullDocument);
        } else if (operationType === 'update') {
          await this.handleUserProfileUpdate(documentKey._id, change.updateDescription);
        }
        break;

      case 'orders':
        await this.syncOrderRelatedData(change);
        break;

      case 'inventory':
        await this.propagateInventoryChanges(change);
        break;

      case 'payments':
        await this.handlePaymentEvents(change);
        break;
    }

    // Trigger cross-collection consistency checks
    await this.validateDataConsistency(change);
  }

  async syncOrderRelatedData(change) {
    const { operationType, fullDocument, documentKey } = change;

    if (operationType === 'insert' && fullDocument) {
      // New order created - sync with related systems
      const syncTasks = [
        this.updateCustomerOrderHistory(fullDocument.customer_id, fullDocument._id),
        this.reserveInventoryItems(fullDocument.items),
        this.createPaymentRecord(fullDocument._id, fullDocument.total_amount),
        this.updateSalesAnalytics(fullDocument)
      ];

      await Promise.allSettled(syncTasks);

    } else if (operationType === 'update') {
      const updatedFields = change.updateDescription?.updatedFields || {};

      // Sync specific field changes
      if ('status' in updatedFields) {
        await this.syncOrderStatusAcrossCollections(documentKey._id, updatedFields.status);
      }

      if ('items' in updatedFields) {
        await this.recalculateRelatedData(documentKey._id, updatedFields.items);
      }
    }
  }

  async setupBatchedChangeProcessing(options = {}) {
    // Process changes in batches for efficiency
    const batchSize = options.batchSize || 100;
    const flushInterval = options.flushIntervalMs || 5000;

    this.batchProcessor = setInterval(async () => {
      if (this.changeBuffer.length > 0) {
        const batch = this.changeBuffer.splice(0, batchSize);
        await this.processBatchedChanges(batch);
      }
    }, flushInterval);

    // Set up change streams to buffer changes
    const changeStream = this.db.collection('events').watch([
      {
        $match: {
          'operationType': { $in: ['insert', 'update'] }
        }
      }
    ]);

    changeStream.on('change', (change) => {
      this.changeBuffer.push({
        ...change,
        bufferedAt: new Date()
      });

      // Flush immediately if buffer is full
      if (this.changeBuffer.length >= batchSize) {
        this.flushChangeBuffer();
      }
    });
  }

  async processBatchedChanges(changes) {
    console.log(`Processing batch of ${changes.length} changes`);

    // Group changes by type for efficient processing
    const changeGroups = changes.reduce((groups, change) => {
      const key = `${change.operationType}_${change.ns?.coll || 'unknown'}`;
      groups[key] = groups[key] || [];
      groups[key].push(change);
      return groups;
    }, {});

    // Process each group
    for (const [groupKey, groupChanges] of Object.entries(changeGroups)) {
      await this.processChangeGroup(groupKey, groupChanges);
    }
  }

  async processChangeGroup(groupKey, changes) {
    const [operationType, collection] = groupKey.split('_');

    switch (collection) {
      case 'analytics_events':
        await this.updateAnalyticsDashboard(changes);
        break;

      case 'user_activities':
        await this.updateUserEngagementMetrics(changes);
        break;

      case 'system_logs':
        await this.processSystemLogBatch(changes);
        break;

      default:
        console.log(`Unhandled change group: ${groupKey}`);
    }
  }

  async setupChangeStreamWithDeduplication() {
    // Prevent duplicate processing of changes
    const processedChanges = new Set();
    const DEDUP_WINDOW_MS = 30000; // 30 seconds

    const changeStream = this.db.collection('critical_data').watch([
      {
        $match: {
          'operationType': { $in: ['insert', 'update', 'delete'] }
        }
      }
    ]);

    changeStream.on('change', async (change) => {
      const changeHash = this.generateChangeHash(change);

      if (processedChanges.has(changeHash)) {
        console.log('Duplicate change detected, skipping:', changeHash);
        return;
      }

      // Add to processed set
      processedChanges.add(changeHash);

      // Remove from set after dedup window
      setTimeout(() => {
        processedChanges.delete(changeHash);
      }, DEDUP_WINDOW_MS);

      // Process the change
      await this.processCriticalChange(change);
    });
  }

  generateChangeHash(change) {
    // Create hash from key change attributes
    const hashData = {
      operationType: change.operationType,
      documentKey: change.documentKey,
      clusterTime: change.clusterTime?.toString(),
      updateFields: change.updateDescription?.updatedFields ? 
        Object.keys(change.updateDescription.updatedFields).sort() : null
    };

    return JSON.stringify(hashData);
  }

  async setupResumableChangeStream(collectionName, pipeline = []) {
    // Implement resumable change streams with persistent resume tokens
    let resumeToken = await this.getStoredResumeToken(collectionName);

    const startChangeStream = () => {
      const options = { fullDocument: 'updateLookup' };

      if (resumeToken) {
        options.resumeAfter = resumeToken;
        console.log(`Resuming change stream for ${collectionName} from token:`, resumeToken);
      }

      const changeStream = this.db.collection(collectionName).watch(pipeline, options);

      changeStream.on('change', async (change) => {
        // Store resume token for recovery
        resumeToken = change._id;
        await this.storeResumeToken(collectionName, resumeToken);

        // Process the change
        await this.processResumeableChange(collectionName, change);
      });

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

        // Attempt to restart stream
        setTimeout(() => {
          console.log('Restarting change stream...');
          startChangeStream();
        }, 5000);
      });

      return changeStream;
    };

    return startChangeStream();
  }

  async storeResumeToken(collectionName, resumeToken) {
    await this.db.collection('change_stream_tokens').updateOne(
      { collection: collectionName },
      { 
        $set: { 
          resumeToken: resumeToken,
          updatedAt: new Date()
        }
      },
      { upsert: true }
    );
  }

  async getStoredResumeToken(collectionName) {
    const tokenDoc = await this.db.collection('change_stream_tokens').findOne({
      collection: collectionName
    });

    return tokenDoc?.resumeToken || null;
  }

  async setupChangeStreamWithFiltering(filterConfig) {
    // Dynamic filtering based on configuration
    const pipeline = [];

    // Operation type filter
    if (filterConfig.operationTypes) {
      pipeline.push({
        $match: {
          'operationType': { $in: filterConfig.operationTypes }
        }
      });
    }

    // Field-specific filters
    if (filterConfig.fieldFilters) {
      const fieldMatches = Object.entries(filterConfig.fieldFilters).map(([field, condition]) => {
        return { [`fullDocument.${field}`]: condition };
      });

      if (fieldMatches.length > 0) {
        pipeline.push({
          $match: { $and: fieldMatches }
        });
      }
    }

    // Custom filter functions
    if (filterConfig.customFilter) {
      pipeline.push({
        $match: {
          $expr: filterConfig.customFilter
        }
      });
    }

    // Projection for efficiency
    if (filterConfig.projection) {
      pipeline.push({
        $project: filterConfig.projection
      });
    }

    const changeStream = this.db.collection(filterConfig.collection).watch(pipeline);

    changeStream.on('change', (change) => {
      this.processFilteredChange(filterConfig.collection, change, filterConfig);
    });

    return changeStream;
  }

  // Placeholder methods
  async handleNewUserRegistration(user) { /* Implementation */ }
  async handleUserProfileUpdate(userId, changes) { /* Implementation */ }
  async propagateInventoryChanges(change) { /* Implementation */ }
  async handlePaymentEvents(change) { /* Implementation */ }
  async validateDataConsistency(change) { /* Implementation */ }
  async updateAnalyticsDashboard(changes) { /* Implementation */ }
  async updateUserEngagementMetrics(changes) { /* Implementation */ }
  async processSystemLogBatch(changes) { /* Implementation */ }
  async processCriticalChange(change) { /* Implementation */ }
  async processResumeableChange(collection, change) { /* Implementation */ }
  async processFilteredChange(collection, change, config) { /* Implementation */ }
}

// Real-time dashboard with Change Streams
class LiveDashboardService {
  constructor(db, websocketServer) {
    this.db = db;
    this.websockets = websocketServer;
    this.dashboardStreams = new Map();
    this.metricsCache = new Map();
  }

  async setupSalesDashboard() {
    // Real-time sales metrics dashboard
    const pipeline = [
      {
        $match: {
          $or: [
            // New sales
            { 
              'operationType': 'insert',
              'ns.coll': 'orders',
              'fullDocument.status': 'completed'
            },
            // Order updates affecting revenue
            {
              'operationType': 'update',
              'ns.coll': 'orders',
              'updateDescription.updatedFields.total_amount': { $exists: true }
            },
            // Refunds
            {
              'operationType': 'insert',
              'ns.coll': 'refunds'
            }
          ]
        }
      },
      {
        $addFields: {
          eventType: {
            $switch: {
              branches: [
                { 
                  case: { $eq: ['$operationType', 'insert'] },
                  then: {
                    $cond: {
                      if: { $eq: ['$ns.coll', 'refunds'] },
                      then: 'refund',
                      else: 'sale'
                    }
                  }
                },
                { case: { $eq: ['$operationType', 'update'] }, then: 'update' }
              ],
              default: 'unknown'
            }
          }
        }
      }
    ];

    const changeStream = this.db.watch(pipeline, {
      fullDocument: 'updateLookup'
    });

    changeStream.on('change', (change) => {
      this.processSalesChange(change);
    });

    this.dashboardStreams.set('sales', changeStream);
  }

  async processSalesChange(change) {
    const { eventType, fullDocument, operationType } = change;

    try {
      let metricsUpdate = {};

      switch (eventType) {
        case 'sale':
          metricsUpdate = await this.processSaleEvent(fullDocument);
          break;

        case 'refund':
          metricsUpdate = await this.processRefundEvent(fullDocument);
          break;

        case 'update':
          metricsUpdate = await this.processOrderUpdateEvent(change);
          break;
      }

      // Update cached metrics
      this.updateMetricsCache('sales', metricsUpdate);

      // Broadcast to connected clients
      this.broadcastMetricsUpdate('sales', metricsUpdate);

    } catch (error) {
      console.error('Error processing sales change:', error);
    }
  }

  async processSaleEvent(order) {
    const now = new Date();
    const today = now.toISOString().split('T')[0];

    // Calculate real-time metrics
    const dailyRevenue = await this.calculateDailyRevenue(today);
    const hourlyOrderCount = await this.calculateHourlyOrders(now);
    const topProducts = await this.getTopProductsToday(today);

    return {
      timestamp: now,
      newSale: {
        orderId: order._id,
        amount: order.total_amount,
        customerId: order.customer_id,
        items: order.items?.length || 0
      },
      aggregates: {
        dailyRevenue: dailyRevenue,
        hourlyOrderCount: hourlyOrderCount,
        totalOrdersToday: await this.getTotalOrdersToday(today),
        averageOrderValue: dailyRevenue / await this.getTotalOrdersToday(today)
      },
      topProducts: topProducts
    };
  }

  async setupInventoryDashboard() {
    // Real-time inventory monitoring
    const pipeline = [
      {
        $match: {
          'operationType': 'update',
          'ns.coll': 'inventory',
          'updateDescription.updatedFields.quantity': { $exists: true }
        }
      },
      {
        $addFields: {
          productId: '$fullDocument.product_id',
          newQuantity: '$fullDocument.quantity',
          quantityChange: '$updateDescription.updatedFields.quantity',
          threshold: '$fullDocument.reorder_threshold',
          category: '$fullDocument.category'
        }
      },
      {
        $match: {
          $or: [
            // Low stock alerts
            { $expr: { $lt: ['$newQuantity', '$threshold'] } },
            // Large quantity changes
            { $expr: { $gt: [{ $abs: '$quantityChange' }, 50] } },
            // Out of stock
            { newQuantity: 0 }
          ]
        }
      }
    ];

    const changeStream = this.db.collection('inventory').watch(pipeline, {
      fullDocument: 'updateLookup'
    });

    changeStream.on('change', (change) => {
      this.processInventoryChange(change);
    });

    this.dashboardStreams.set('inventory', changeStream);
  }

  async processInventoryChange(change) {
    const { productId, newQuantity, quantityChange, threshold, category } = change;

    const alertLevel = this.determineAlertLevel(newQuantity, threshold, quantityChange);
    const categoryMetrics = await this.getCategoryInventoryMetrics(category);

    const update = {
      timestamp: new Date(),
      inventory_alert: {
        productId: productId,
        quantity: newQuantity,
        change: quantityChange,
        alertLevel: alertLevel,
        category: category
      },
      category_metrics: categoryMetrics,
      low_stock_count: await this.getLowStockCount()
    };

    this.updateMetricsCache('inventory', update);
    this.broadcastMetricsUpdate('inventory', update);

    // Send critical alerts immediately
    if (alertLevel === 'critical') {
      this.sendCriticalInventoryAlert(productId, newQuantity);
    }
  }

  determineAlertLevel(quantity, threshold, change) {
    if (quantity === 0) return 'critical';
    if (quantity <= threshold * 0.5) return 'high';
    if (quantity <= threshold) return 'medium';
    if (Math.abs(change) > 100) return 'unusual';
    return 'normal';
  }

  async setupUserActivityDashboard() {
    // Real-time user activity tracking
    const pipeline = [
      {
        $match: {
          $or: [
            // New user registrations
            {
              'operationType': 'insert',
              'ns.coll': 'users'
            },
            // User login events
            {
              'operationType': 'insert',
              'ns.coll': 'user_sessions'
            },
            // User activity updates
            {
              'operationType': 'update',
              'ns.coll': 'users',
              'updateDescription.updatedFields.last_activity': { $exists: true }
            }
          ]
        }
      }
    ];

    const changeStream = this.db.watch(pipeline, {
      fullDocument: 'updateLookup'
    });

    changeStream.on('change', (change) => {
      this.processUserActivityChange(change);
    });

    this.dashboardStreams.set('user_activity', changeStream);
  }

  async processUserActivityChange(change) {
    const { operationType, ns, fullDocument } = change;

    let activityUpdate = {
      timestamp: new Date()
    };

    if (ns.coll === 'users' && operationType === 'insert') {
      // New user registration
      activityUpdate.new_user = {
        userId: fullDocument._id,
        email: fullDocument.email,
        registrationTime: fullDocument.created_at
      };

      activityUpdate.metrics = {
        dailyRegistrations: await this.getDailyRegistrations(),
        totalUsers: await this.getTotalUserCount(),
        activeUsersToday: await this.getActiveUsersToday()
      };

    } else if (ns.coll === 'user_sessions' && operationType === 'insert') {
      // New user session (login)
      activityUpdate.user_login = {
        userId: fullDocument.user_id,
        sessionId: fullDocument._id,
        loginTime: fullDocument.created_at,
        userAgent: fullDocument.user_agent
      };

      activityUpdate.metrics = {
        activeSessionsNow: await this.getActiveSessionCount(),
        loginsToday: await this.getDailyLogins()
      };
    }

    this.updateMetricsCache('user_activity', activityUpdate);
    this.broadcastMetricsUpdate('user_activity', activityUpdate);
  }

  updateMetricsCache(dashboardType, update) {
    const existing = this.metricsCache.get(dashboardType) || {};
    const merged = { ...existing, ...update };
    this.metricsCache.set(dashboardType, merged);
  }

  broadcastMetricsUpdate(dashboardType, update) {
    const message = {
      type: 'dashboard_update',
      dashboard: dashboardType,
      data: update
    };

    // Broadcast to all connected WebSocket clients
    this.websockets.emit('dashboard_update', message);
  }

  async sendCriticalInventoryAlert(productId, quantity) {
    const product = await this.db.collection('products').findOne({ _id: productId });

    const alert = {
      type: 'critical_inventory_alert',
      productId: productId,
      productName: product?.name || 'Unknown Product',
      quantity: quantity,
      timestamp: new Date(),
      severity: 'critical'
    };

    // Send to specific alert channels
    this.websockets.emit('critical_alert', alert);

    // Could also integrate with external alerting (email, Slack, etc.)
    await this.sendExternalAlert(alert);
  }

  async getCurrentMetrics(dashboardType) {
    // Get current cached metrics for dashboard initialization
    return this.metricsCache.get(dashboardType) || {};
  }

  async closeDashboard(dashboardType) {
    const stream = this.dashboardStreams.get(dashboardType);
    if (stream) {
      await stream.close();
      this.dashboardStreams.delete(dashboardType);
      this.metricsCache.delete(dashboardType);
    }
  }

  // Placeholder methods for metric calculations
  async calculateDailyRevenue(date) { /* Implementation */ }
  async calculateHourlyOrders(hour) { /* Implementation */ }
  async getTopProductsToday(date) { /* Implementation */ }
  async getTotalOrdersToday(date) { /* Implementation */ }
  async getCategoryInventoryMetrics(category) { /* Implementation */ }
  async getLowStockCount() { /* Implementation */ }
  async getDailyRegistrations() { /* Implementation */ }
  async getTotalUserCount() { /* Implementation */ }
  async getActiveUsersToday() { /* Implementation */ }
  async getActiveSessionCount() { /* Implementation */ }
  async getDailyLogins() { /* Implementation */ }
  async sendExternalAlert(alert) { /* Implementation */ }
}