MongoDB Capped Collections: High-Performance Circular Buffers with SQL-Style Fixed-Size Data Management

-- Traditional SQL log table - grows indefinitely
CREATE TABLE application_logs (
    log_id BIGSERIAL PRIMARY KEY,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    level VARCHAR(10) NOT NULL,
    service_name VARCHAR(50) NOT NULL,
    message TEXT NOT NULL,
    metadata JSONB,
    request_id UUID,
    user_id INTEGER,
    session_id VARCHAR(100),

    INDEX idx_timestamp (timestamp),
    INDEX idx_level (level),
    INDEX idx_service (service_name)
);

-- High-volume insertions
INSERT INTO application_logs (level, service_name, message, metadata, request_id, user_id)
VALUES 
    ('INFO', 'auth-service', 'User login successful', '{"ip": "192.168.1.100", "browser": "Chrome"}', uuid_generate_v4(), 12345),
    ('ERROR', 'payment-service', 'Payment processing failed', '{"amount": 99.99, "currency": "USD", "error_code": "CARD_DECLINED"}', uuid_generate_v4(), 67890),
    ('DEBUG', 'api-gateway', 'Request routed to microservice', '{"path": "/api/v1/users", "method": "GET", "response_time": 45}', uuid_generate_v4(), 11111);

-- Problems with unlimited growth:
-- 1. Table size grows indefinitely requiring manual cleanup
-- 2. Performance degrades as table size increases
-- 3. Index maintenance overhead scales with data volume
-- 4. Complex retention policies need external scheduling
-- 5. Storage costs increase without bounds
-- 6. Backup and maintenance times increase linearly

-- Manual cleanup required with complex scheduling
DELETE FROM application_logs 
WHERE timestamp < NOW() - INTERVAL '30 days';

-- Problems with manual cleanup:
-- - Requires scheduled maintenance scripts
-- - DELETE operations can cause table locks
-- - Index fragmentation after large deletions
-- - Uneven performance during cleanup windows
-- - Risk of accidentally deleting important data
-- - Complex retention rules difficult to implement

// MongoDB capped collection - automatic size management
// Create capped collection with automatic circular buffer behavior
db.createCollection("application_logs", {
  capped: true,
  size: 100 * 1024 * 1024, // 100MB maximum size
  max: 50000,              // Maximum 50,000 documents
  autoIndexId: false       // Optimize for insert performance
});

// High-performance insertions with guaranteed order preservation
db.application_logs.insertMany([
  {
    timestamp: new Date(),
    level: "INFO",
    serviceName: "auth-service",
    message: "User login successful",
    metadata: {
      ip: "192.168.1.100",
      browser: "Chrome",
      responseTime: 23
    },
    requestId: "req_001",
    userId: 12345,
    sessionId: "sess_abc123"
  },
  {
    timestamp: new Date(),
    level: "ERROR", 
    serviceName: "payment-service",
    message: "Payment processing failed",
    metadata: {
      amount: 99.99,
      currency: "USD",
      errorCode: "CARD_DECLINED",
      attemptNumber: 2
    },
    requestId: "req_002",
    userId: 67890
  },
  {
    timestamp: new Date(),
    level: "DEBUG",
    serviceName: "api-gateway", 
    message: "Request routed to microservice",
    metadata: {
      path: "/api/v1/users",
      method: "GET",
      responseTime: 45,
      upstreamService: "user-service"
    },
    requestId: "req_003",
    userId: 11111
  }
]);

// Benefits of capped collections:
// - Fixed size with automatic circular buffer behavior
// - Guaranteed insert order preservation (natural order)
// - High-performance insertions (no index maintenance overhead)
// - Automatic data lifecycle management (oldest data removed automatically)
// - Predictable performance characteristics regardless of data volume
// - No manual cleanup or maintenance required
// - Optimized for append-only workloads
// - Built-in tailable cursor support for real-time streaming

// Comprehensive capped collection management system
class CappedCollectionManager {
  constructor(db) {
    this.db = db;
    this.collections = new Map();
    this.tailableCursors = new Map();
  }

  async createLogCollection(serviceName, options = {}) {
    // Create service-specific log collection
    const collectionName = `${serviceName}_logs`;
    const defaultOptions = {
      size: 50 * 1024 * 1024,  // 50MB default size
      max: 25000,              // 25K documents default
      autoIndexId: false       // Optimize for pure append workload
    };

    const cappedOptions = {
      capped: true,
      ...defaultOptions,
      ...options
    };

    try {
      // Create the capped collection
      await this.db.createCollection(collectionName, cappedOptions);

      // Store collection configuration
      this.collections.set(collectionName, {
        type: 'logs',
        service: serviceName,
        options: cappedOptions,
        createdAt: new Date(),
        totalInserts: 0,
        lastActivity: new Date()
      });

      console.log(`Created capped log collection: ${collectionName}`, cappedOptions);
      return this.db.collection(collectionName);

    } catch (error) {
      if (error.code === 48) { // Collection already exists
        console.log(`Capped collection ${collectionName} already exists`);
        return this.db.collection(collectionName);
      }
      throw error;
    }
  }

  async createMetricsCollection(metricType, options = {}) {
    // Create high-frequency metrics collection
    const collectionName = `metrics_${metricType}`;
    const metricsOptions = {
      capped: true,
      size: 200 * 1024 * 1024, // 200MB for metrics data
      max: 100000,             // 100K metric documents
      autoIndexId: false
    };

    const collection = await this.db.createCollection(collectionName, {
      ...metricsOptions,
      ...options
    });

    this.collections.set(collectionName, {
      type: 'metrics',
      metricType: metricType,
      options: metricsOptions,
      createdAt: new Date(),
      totalInserts: 0,
      lastActivity: new Date()
    });

    return collection;
  }

  async createEventStreamCollection(streamName, options = {}) {
    // Create event streaming collection for real-time processing
    const collectionName = `events_${streamName}`;
    const eventOptions = {
      capped: true,
      size: 100 * 1024 * 1024, // 100MB for event stream
      max: 50000,              // 50K events
      autoIndexId: false
    };

    const collection = await this.db.createCollection(collectionName, {
      ...eventOptions,
      ...options
    });

    this.collections.set(collectionName, {
      type: 'events',
      streamName: streamName,
      options: eventOptions,
      createdAt: new Date(),
      totalInserts: 0,
      lastActivity: new Date()
    });

    return collection;
  }

  async logMessage(serviceName, logData) {
    // High-performance logging with automatic batching
    const collectionName = `${serviceName}_logs`;
    let collection = this.db.collection(collectionName);

    // Create collection if it doesn't exist
    if (!this.collections.has(collectionName)) {
      collection = await this.createLogCollection(serviceName);
    }

    // Prepare log document with required fields
    const logDocument = {
      timestamp: logData.timestamp || new Date(),
      level: logData.level || 'INFO',
      serviceName: serviceName,
      message: logData.message,

      // Optional structured data
      metadata: logData.metadata || {},
      requestId: logData.requestId || null,
      userId: logData.userId || null,
      sessionId: logData.sessionId || null,
      traceId: logData.traceId || null,
      spanId: logData.spanId || null,

      // Performance tracking
      hostname: logData.hostname || require('os').hostname(),
      processId: process.pid,
      threadId: logData.threadId || 0,

      // Categorization
      category: logData.category || 'general',
      tags: logData.tags || []
    };

    // Insert with fire-and-forget for maximum performance
    await collection.insertOne(logDocument, { 
      writeConcern: { w: 0 } // Fire-and-forget for logs
    });

    // Update collection statistics
    const collectionInfo = this.collections.get(collectionName);
    if (collectionInfo) {
      collectionInfo.totalInserts++;
      collectionInfo.lastActivity = new Date();
    }

    return logDocument._id;
  }

  async writeMetrics(metricType, metricsData) {
    // High-throughput metrics writing
    const collectionName = `metrics_${metricType}`;
    let collection = this.db.collection(collectionName);

    if (!this.collections.has(collectionName)) {
      collection = await this.createMetricsCollection(metricType);
    }

    // Prepare metrics document
    const metricsDocument = {
      timestamp: metricsData.timestamp || new Date(),
      metricType: metricType,

      // Metric values
      values: metricsData.values || {},

      // Dimensions for grouping and filtering
      dimensions: {
        service: metricsData.service,
        environment: metricsData.environment || 'production',
        region: metricsData.region || 'us-east-1',
        version: metricsData.version || '1.0.0',
        ...metricsData.dimensions
      },

      // Aggregation-friendly structure
      counters: metricsData.counters || {},
      gauges: metricsData.gauges || {},
      histograms: metricsData.histograms || {},
      timers: metricsData.timers || {},

      // Source information
      source: {
        hostname: metricsData.hostname || require('os').hostname(),
        processId: process.pid,
        collectionId: metricsData.collectionId || null
      }
    };

    // Batch insertion for metrics (multiple metrics per call)
    if (Array.isArray(metricsData)) {
      const documents = metricsData.map(data => ({
        timestamp: data.timestamp || new Date(),
        metricType: metricType,
        values: data.values || {},
        dimensions: { ...data.dimensions },
        counters: data.counters || {},
        gauges: data.gauges || {},
        histograms: data.histograms || {},
        timers: data.timers || {},
        source: {
          hostname: data.hostname || require('os').hostname(),
          processId: process.pid,
          collectionId: data.collectionId || null
        }
      }));

      await collection.insertMany(documents, { 
        ordered: false, // Allow partial success
        writeConcern: { w: 0 }
      });

      return documents.length;
    } else {
      await collection.insertOne(metricsDocument, { 
        writeConcern: { w: 0 }
      });

      return 1;
    }
  }

  async publishEvent(streamName, eventData) {
    // Event streaming with guaranteed order preservation
    const collectionName = `events_${streamName}`;
    let collection = this.db.collection(collectionName);

    if (!this.collections.has(collectionName)) {
      collection = await this.createEventStreamCollection(streamName);
    }

    const eventDocument = {
      timestamp: eventData.timestamp || new Date(),
      eventId: eventData.eventId || new ObjectId(),
      eventType: eventData.eventType,
      streamName: streamName,

      // Event payload
      data: eventData.data || {},

      // Event metadata
      metadata: {
        version: eventData.version || '1.0',
        source: eventData.source || 'unknown',
        correlationId: eventData.correlationId || null,
        causationId: eventData.causationId || null,
        userId: eventData.userId || null,
        sessionId: eventData.sessionId || null,
        ...eventData.metadata
      },

      // Event context
      context: {
        service: eventData.service || 'unknown',
        environment: eventData.environment || 'production',
        hostname: require('os').hostname(),
        processId: process.pid,
        requestId: eventData.requestId || null
      }
    };

    // Events may need acknowledgment
    const result = await collection.insertOne(eventDocument, {
      writeConcern: { w: 1, j: true } // Ensure durability for events
    });

    return {
      eventId: eventDocument.eventId,
      insertedId: result.insertedId,
      timestamp: eventDocument.timestamp
    };
  }

  async queryRecentLogs(serviceName, options = {}) {
    // Query recent logs with natural ordering (insertion order)
    const collectionName = `${serviceName}_logs`;
    const collection = this.db.collection(collectionName);

    const query = {};

    // Add filters
    if (options.level) {
      query.level = options.level;
    }

    if (options.since) {
      query.timestamp = { $gte: options.since };
    }

    if (options.userId) {
      query.userId = options.userId;
    }

    if (options.category) {
      query.category = options.category;
    }

    // Use natural ordering for efficiency (no index needed)
    const cursor = collection.find(query);

    if (options.reverse) {
      // Get most recent first (reverse natural order)
      cursor.sort({ $natural: -1 });
    }

    if (options.limit) {
      cursor.limit(options.limit);
    }

    const logs = await cursor.toArray();

    return {
      logs: logs,
      count: logs.length,
      service: serviceName,
      query: query,
      options: options
    };
  }

  async getMetricsAggregation(metricType, timeRange, aggregationType = 'avg') {
    // Efficient metrics aggregation over time ranges
    const collectionName = `metrics_${metricType}`;
    const collection = this.db.collection(collectionName);

    const pipeline = [
      {
        $match: {
          timestamp: {
            $gte: timeRange.start,
            $lte: timeRange.end
          }
        }
      },
      {
        $group: {
          _id: {
            service: '$dimensions.service',
            environment: '$dimensions.environment',
            // Group by time bucket for time-series analysis
            timeBucket: {
              $dateTrunc: {
                date: '$timestamp',
                unit: timeRange.bucketSize || 'minute',
                binSize: timeRange.binSize || 1
              }
            }
          },

          // Aggregate different metric types
          avgValues: { $avg: '$values' },
          maxValues: { $max: '$values' },
          minValues: { $min: '$values' },
          sumCounters: { $sum: '$counters' },

          count: { $sum: 1 },

          firstTimestamp: { $min: '$timestamp' },
          lastTimestamp: { $max: '$timestamp' }
        }
      },
      {
        $sort: {
          '_id.timeBucket': 1,
          '_id.service': 1
        }
      },
      {
        $project: {
          service: '$_id.service',
          environment: '$_id.environment',
          timeBucket: '$_id.timeBucket',

          aggregatedValue: {
            $switch: {
              branches: [
                { case: { $eq: [aggregationType, 'avg'] }, then: '$avgValues' },
                { case: { $eq: [aggregationType, 'max'] }, then: '$maxValues' },
                { case: { $eq: [aggregationType, 'min'] }, then: '$minValues' },
                { case: { $eq: [aggregationType, 'sum'] }, then: '$sumCounters' }
              ],
              default: '$avgValues'
            }
          },

          dataPoints: '$count',
          timeRange: {
            start: '$firstTimestamp',
            end: '$lastTimestamp'
          },

          _id: 0
        }
      }
    ];

    const results = await collection.aggregate(pipeline).toArray();

    return {
      metricType: metricType,
      aggregationType: aggregationType,
      timeRange: timeRange,
      results: results,
      totalDataPoints: results.reduce((sum, r) => sum + r.dataPoints, 0)
    };
  }

  async createTailableCursor(collectionName, options = {}) {
    // Create tailable cursor for real-time streaming
    const collection = this.db.collection(collectionName);

    // Verify collection is capped
    const collectionInfo = await this.db.command({
      collStats: collectionName
    });

    if (!collectionInfo.capped) {
      throw new Error(`Collection ${collectionName} is not capped - tailable cursors require capped collections`);
    }

    const query = options.filter || {};
    const cursorOptions = {
      tailable: true,      // Don't close cursor when reaching end
      awaitData: true,     // Block briefly when no data available
      noCursorTimeout: true, // Don't timeout cursor
      maxTimeMS: options.maxTimeMS || 1000,
      batchSize: options.batchSize || 100
    };

    const cursor = collection.find(query, cursorOptions);

    // Store cursor reference for management
    const cursorId = `${collectionName}_${Date.now()}`;
    this.tailableCursors.set(cursorId, {
      cursor: cursor,
      collection: collectionName,
      filter: query,
      createdAt: new Date(),
      lastActivity: new Date()
    });

    return {
      cursorId: cursorId,
      cursor: cursor
    };
  }

  async streamData(collectionName, callback, options = {}) {
    // High-level streaming interface with automatic reconnection
    const { cursor, cursorId } = await this.createTailableCursor(collectionName, options);

    console.log(`Starting data stream from ${collectionName}`);

    try {
      while (await cursor.hasNext()) {
        const document = await cursor.next();

        if (document) {
          // Update last activity
          const cursorInfo = this.tailableCursors.get(cursorId);
          if (cursorInfo) {
            cursorInfo.lastActivity = new Date();
          }

          // Process document through callback
          try {
            await callback(document, { 
              collection: collectionName,
              cursorId: cursorId 
            });
          } catch (callbackError) {
            console.error('Stream callback error:', callbackError);
            // Continue streaming despite callback errors
          }
        }
      }
    } catch (streamError) {
      console.error(`Stream error for ${collectionName}:`, streamError);

      // Cleanup cursor reference
      this.tailableCursors.delete(cursorId);

      // Auto-reconnect for network errors
      if (streamError.name === 'MongoNetworkError' && options.autoReconnect !== false) {
        console.log(`Attempting to reconnect stream for ${collectionName}...`);
        setTimeout(() => {
          this.streamData(collectionName, callback, options);
        }, options.reconnectDelay || 5000);
      }

      throw streamError;
    }
  }

  async getCappedCollectionStats(collectionName) {
    // Get comprehensive statistics for capped collection
    const stats = await this.db.command({
      collStats: collectionName,
      indexDetails: true
    });

    const collection = this.db.collection(collectionName);

    // Get document count and size information
    const documentCount = await collection.estimatedDocumentCount();
    const newestDoc = await collection.findOne({}, { sort: { $natural: -1 } });
    const oldestDoc = await collection.findOne({}, { sort: { $natural: 1 } });

    return {
      collection: collectionName,
      capped: stats.capped,

      // Size information
      maxSize: stats.maxSize,
      size: stats.size,
      storageSize: stats.storageSize,
      sizeUtilization: stats.size / stats.maxSize,

      // Document information
      maxDocuments: stats.max,
      documentCount: documentCount,
      avgDocumentSize: documentCount > 0 ? stats.size / documentCount : 0,
      documentUtilization: stats.max ? documentCount / stats.max : null,

      // Time range information
      timespan: newestDoc && oldestDoc ? {
        oldest: oldestDoc.timestamp || oldestDoc._id.getTimestamp(),
        newest: newestDoc.timestamp || newestDoc._id.getTimestamp(),
        spanMs: newestDoc && oldestDoc ? 
          (newestDoc.timestamp || newestDoc._id.getTimestamp()).getTime() - 
          (oldestDoc.timestamp || oldestDoc._id.getTimestamp()).getTime() : 0
      } : null,

      // Performance information
      indexes: stats.indexSizes,
      totalIndexSize: Object.values(stats.indexSizes).reduce((sum, size) => sum + size, 0),

      // Collection metadata
      collectionInfo: this.collections.get(collectionName) || null,

      analyzedAt: new Date()
    };
  }

  async optimizeCappedCollection(collectionName, analysisOptions = {}) {
    // Analyze and provide optimization recommendations
    const stats = await this.getCappedCollectionStats(collectionName);
    const recommendations = [];

    // Size utilization analysis
    if (stats.sizeUtilization < 0.5) {
      recommendations.push({
        type: 'size_optimization',
        priority: 'medium',
        message: `Collection is only ${(stats.sizeUtilization * 100).toFixed(1)}% full. Consider reducing maxSize to save storage.`,
        suggestedMaxSize: Math.ceil(stats.size * 1.2) // 20% headroom
      });
    }

    if (stats.sizeUtilization > 0.9) {
      recommendations.push({
        type: 'size_warning',
        priority: 'high',
        message: `Collection is ${(stats.sizeUtilization * 100).toFixed(1)}% full. Consider increasing maxSize to prevent data loss.`,
        suggestedMaxSize: Math.ceil(stats.maxSize * 1.5) // 50% increase
      });
    }

    // Document count analysis
    if (stats.documentUtilization && stats.documentUtilization < 0.5) {
      recommendations.push({
        type: 'document_optimization',
        priority: 'low',
        message: `Only ${(stats.documentUtilization * 100).toFixed(1)}% of max documents used. Consider reducing max document limit.`,
        suggestedMaxDocs: Math.ceil(stats.documentCount * 1.2)
      });
    }

    // Document size analysis
    if (stats.avgDocumentSize > 10 * 1024) { // 10KB average
      recommendations.push({
        type: 'document_size_warning',
        priority: 'medium',
        message: `Average document size is ${(stats.avgDocumentSize / 1024).toFixed(1)}KB. Large documents may impact performance in capped collections.`
      });
    }

    // Index analysis
    if (stats.totalIndexSize > stats.size * 0.2) { // Indexes > 20% of data size
      recommendations.push({
        type: 'index_optimization',
        priority: 'medium',
        message: `Index size (${(stats.totalIndexSize / 1024 / 1024).toFixed(1)}MB) is large relative to data size. Consider if all indexes are necessary for capped collection workload.`
      });
    }

    // Time span analysis
    if (stats.timespan && stats.timespan.spanMs < 60 * 60 * 1000) { // Less than 1 hour
      recommendations.push({
        type: 'retention_warning',
        priority: 'high',
        message: `Data retention span is only ${(stats.timespan.spanMs / (60 * 1000)).toFixed(1)} minutes. Consider increasing collection size for longer data retention.`
      });
    }

    return {
      collectionStats: stats,
      recommendations: recommendations,
      optimizationScore: this.calculateOptimizationScore(stats, recommendations),
      analyzedAt: new Date()
    };
  }

  calculateOptimizationScore(stats, recommendations) {
    // Calculate optimization score (0-100, higher is better)
    let score = 100;

    // Deduct points for each recommendation based on priority
    recommendations.forEach(rec => {
      switch (rec.priority) {
        case 'high':
          score -= 30;
          break;
        case 'medium':
          score -= 15;
          break;
        case 'low':
          score -= 5;
          break;
      }
    });

    // Bonus points for good utilization
    if (stats.sizeUtilization >= 0.6 && stats.sizeUtilization <= 0.8) {
      score += 10; // Good size utilization
    }

    if (stats.avgDocumentSize < 5 * 1024) { // < 5KB average
      score += 5; // Good document size
    }

    return Math.max(0, Math.min(100, score));
  }

  async closeTailableCursor(cursorId) {
    // Safely close tailable cursor
    const cursorInfo = this.tailableCursors.get(cursorId);

    if (cursorInfo) {
      try {
        await cursorInfo.cursor.close();
      } catch (error) {
        console.error(`Error closing cursor ${cursorId}:`, error);
      }

      this.tailableCursors.delete(cursorId);
      console.log(`Closed tailable cursor: ${cursorId}`);
    }
  }

  async cleanup() {
    // Cleanup all tailable cursors
    const cursors = Array.from(this.tailableCursors.keys());

    for (const cursorId of cursors) {
      await this.closeTailableCursor(cursorId);
    }

    console.log(`Cleaned up ${cursors.length} tailable cursors`);
  }
}

// Real-time streaming and event processing with tailable cursors
class RealTimeStreamProcessor {
  constructor(db) {
    this.db = db;
    this.cappedManager = new CappedCollectionManager(db);
    this.activeStreams = new Map();
    this.eventHandlers = new Map();
  }

  async setupLogStreaming(services = []) {
    // Setup real-time log streaming for multiple services
    for (const service of services) {
      await this.cappedManager.createLogCollection(service, {
        size: 100 * 1024 * 1024, // 100MB per service
        max: 50000
      });

      // Start streaming logs for this service
      this.startLogStream(service);
    }
  }

  async startLogStream(serviceName) {
    const collectionName = `${serviceName}_logs`;

    console.log(`Starting log stream for ${serviceName}...`);

    // Create stream processor
    const streamProcessor = async (logDocument, streamContext) => {
      try {
        // Process log based on level
        await this.processLogMessage(logDocument, streamContext);

        // Trigger alerts for critical logs
        if (logDocument.level === 'ERROR' || logDocument.level === 'FATAL') {
          await this.handleCriticalLog(logDocument);
        }

        // Update real-time metrics
        await this.updateLogMetrics(serviceName, logDocument);

        // Forward to external systems if needed
        if (this.eventHandlers.has('log_processed')) {
          await this.eventHandlers.get('log_processed')(logDocument);
        }

      } catch (processingError) {
        console.error('Log processing error:', processingError);
      }
    };

    // Start the stream
    const streamPromise = this.cappedManager.streamData(
      collectionName,
      streamProcessor,
      {
        autoReconnect: true,
        reconnectDelay: 5000,
        batchSize: 50
      }
    );

    this.activeStreams.set(serviceName, streamPromise);
  }

  async processLogMessage(logDocument, streamContext) {
    // Real-time log message processing
    const processing = {
      timestamp: new Date(),
      service: logDocument.serviceName,
      level: logDocument.level,
      messageLength: logDocument.message.length,
      hasMetadata: Object.keys(logDocument.metadata || {}).length > 0,
      processingLatency: Date.now() - logDocument.timestamp.getTime()
    };

    // Pattern matching for specific log types
    if (logDocument.message.includes('OutOfMemoryError')) {
      await this.handleOutOfMemoryAlert(logDocument);
    }

    if (logDocument.message.includes('Connection timeout')) {
      await this.handleConnectionIssue(logDocument);
    }

    if (logDocument.requestId && logDocument.level === 'ERROR') {
      await this.trackRequestError(logDocument);
    }

    // Store processing metadata for analytics
    await this.db.collection('log_processing_stats').insertOne({
      ...processing,
      logId: logDocument._id
    });
  }

  async handleCriticalLog(logDocument) {
    // Handle critical log events
    const alert = {
      timestamp: new Date(),
      alertType: 'critical_log',
      severity: logDocument.level,
      service: logDocument.serviceName,
      message: logDocument.message,
      metadata: logDocument.metadata,

      // Context information
      requestId: logDocument.requestId,
      userId: logDocument.userId,
      sessionId: logDocument.sessionId,

      // Alert details
      alertId: new ObjectId(),
      acknowledged: false,
      escalated: false
    };

    // Store alert
    await this.db.collection('critical_alerts').insertOne(alert);

    // Send notifications (implement based on your notification system)
    await this.sendAlertNotification(alert);

    // Auto-escalate if needed
    if (logDocument.level === 'FATAL') {
      setTimeout(async () => {
        await this.escalateAlert(alert.alertId);
      }, 5 * 60 * 1000); // Escalate after 5 minutes if not acknowledged
    }
  }

  async setupMetricsStreaming(metricTypes = []) {
    // Setup real-time metrics streaming
    for (const metricType of metricTypes) {
      await this.cappedManager.createMetricsCollection(metricType, {
        size: 200 * 1024 * 1024, // 200MB per metric type
        max: 100000
      });

      this.startMetricsStream(metricType);
    }
  }

  async startMetricsStream(metricType) {
    const collectionName = `metrics_${metricType}`;

    const metricsProcessor = async (metricsDocument, streamContext) => {
      try {
        // Real-time metrics processing
        await this.processMetricsData(metricsDocument, streamContext);

        // Check for threshold violations
        await this.checkMetricsThresholds(metricsDocument);

        // Update real-time dashboards
        if (this.eventHandlers.has('metrics_updated')) {
          await this.eventHandlers.get('metrics_updated')(metricsDocument);
        }

        // Aggregate into time-series buckets
        await this.aggregateMetricsData(metricsDocument);

      } catch (processingError) {
        console.error('Metrics processing error:', processingError);
      }
    };

    const streamPromise = this.cappedManager.streamData(
      collectionName,
      metricsProcessor,
      {
        autoReconnect: true,
        batchSize: 100,
        filter: { 
          // Only process metrics from last 5 minutes to avoid historical data on restart
          timestamp: { $gte: new Date(Date.now() - 5 * 60 * 1000) }
        }
      }
    );

    this.activeStreams.set(`metrics_${metricType}`, streamPromise);
  }

  async processMetricsData(metricsDocument, streamContext) {
    // Process individual metrics document
    const metricType = metricsDocument.metricType;
    const values = metricsDocument.values || {};
    const counters = metricsDocument.counters || {};
    const gauges = metricsDocument.gauges || {};

    // Calculate derived metrics
    const derivedMetrics = {
      timestamp: metricsDocument.timestamp,
      metricType: metricType,
      service: metricsDocument.dimensions?.service,

      // Calculate rates and percentages
      rates: {},
      percentages: {},
      health: {}
    };

    // Calculate request rate if applicable
    if (counters.requests) {
      const timeWindow = 60; // 1 minute window
      const requestRate = counters.requests / timeWindow;
      derivedMetrics.rates.requestsPerSecond = requestRate;
    }

    // Calculate error percentage
    if (counters.requests && counters.errors) {
      derivedMetrics.percentages.errorRate = (counters.errors / counters.requests) * 100;
    }

    // Calculate response time percentiles if histogram data available
    if (metricsDocument.histograms?.response_time) {
      derivedMetrics.responseTime = this.calculatePercentiles(
        metricsDocument.histograms.response_time
      );
    }

    // Health scoring
    derivedMetrics.health.score = this.calculateHealthScore(metricsDocument);
    derivedMetrics.health.status = this.getHealthStatus(derivedMetrics.health.score);

    // Store derived metrics
    await this.db.collection('derived_metrics').insertOne(derivedMetrics);
  }

  async checkMetricsThresholds(metricsDocument) {
    // Check metrics against defined thresholds
    const thresholds = await this.getThresholdsForService(
      metricsDocument.dimensions?.service
    );

    const violations = [];

    // Check various threshold types
    Object.entries(thresholds.counters || {}).forEach(([metric, threshold]) => {
      const value = metricsDocument.counters?.[metric];
      if (value !== undefined && value > threshold.max) {
        violations.push({
          type: 'counter',
          metric: metric,
          value: value,
          threshold: threshold.max,
          severity: threshold.severity || 'warning'
        });
      }
    });

    Object.entries(thresholds.gauges || {}).forEach(([metric, threshold]) => {
      const value = metricsDocument.gauges?.[metric];
      if (value !== undefined) {
        if (threshold.max && value > threshold.max) {
          violations.push({
            type: 'gauge_high',
            metric: metric,
            value: value,
            threshold: threshold.max,
            severity: threshold.severity || 'warning'
          });
        }
        if (threshold.min && value < threshold.min) {
          violations.push({
            type: 'gauge_low',
            metric: metric,
            value: value,
            threshold: threshold.min,
            severity: threshold.severity || 'warning'
          });
        }
      }
    });

    // Handle threshold violations
    for (const violation of violations) {
      await this.handleThresholdViolation(violation, metricsDocument);
    }
  }

  async setupEventStreaming(streamNames = []) {
    // Setup event streaming for event-driven architectures
    for (const streamName of streamNames) {
      await this.cappedManager.createEventStreamCollection(streamName, {
        size: 100 * 1024 * 1024,
        max: 50000
      });

      this.startEventStream(streamName);
    }
  }

  async startEventStream(streamName) {
    const collectionName = `events_${streamName}`;

    const eventProcessor = async (eventDocument, streamContext) => {
      try {
        // Process event based on type
        await this.processEvent(eventDocument, streamContext);

        // Trigger event handlers
        const eventType = eventDocument.eventType;
        if (this.eventHandlers.has(eventType)) {
          await this.eventHandlers.get(eventType)(eventDocument);
        }

        // Update event processing metrics
        await this.updateEventMetrics(streamName, eventDocument);

      } catch (processingError) {
        console.error('Event processing error:', processingError);

        // Handle event processing failure
        await this.handleEventProcessingFailure(eventDocument, processingError);
      }
    };

    const streamPromise = this.cappedManager.streamData(
      collectionName,
      eventProcessor,
      {
        autoReconnect: true,
        batchSize: 25 // Smaller batches for events to reduce latency
      }
    );

    this.activeStreams.set(`events_${streamName}`, streamPromise);
  }

  async processEvent(eventDocument, streamContext) {
    // Process individual event
    const eventType = eventDocument.eventType;
    const eventData = eventDocument.data;
    const eventMetadata = eventDocument.metadata;

    // Event processing based on type
    switch (eventType) {
      case 'user_action':
        await this.processUserActionEvent(eventDocument);
        break;

      case 'system_state_change':
        await this.processSystemStateEvent(eventDocument);
        break;

      case 'transaction_completed':
        await this.processTransactionEvent(eventDocument);
        break;

      case 'alert_triggered':
        await this.processAlertEvent(eventDocument);
        break;

      default:
        await this.processGenericEvent(eventDocument);
    }

    // Store event processing record
    await this.db.collection('event_processing_log').insertOne({
      eventId: eventDocument.eventId,
      eventType: eventType,
      streamName: eventDocument.streamName,
      processedAt: new Date(),
      processingLatency: Date.now() - eventDocument.timestamp.getTime(),
      success: true
    });
  }

  // Event handler registration
  registerEventHandler(eventType, handler) {
    this.eventHandlers.set(eventType, handler);
  }

  unregisterEventHandler(eventType) {
    this.eventHandlers.delete(eventType);
  }

  // Utility methods
  async getThresholdsForService(serviceName) {
    // Get threshold configuration for service
    const config = await this.db.collection('service_thresholds').findOne({
      service: serviceName
    });

    return config?.thresholds || {
      counters: {},
      gauges: {},
      histograms: {}
    };
  }

  calculatePercentiles(histogramData) {
    // Calculate percentiles from histogram data
    // Implementation depends on histogram format
    return {
      p50: 0,
      p90: 0,
      p95: 0,
      p99: 0
    };
  }

  calculateHealthScore(metricsDocument) {
    // Calculate overall health score from metrics
    let score = 100;

    // Deduct based on error rates, response times, etc.
    const errorRate = metricsDocument.counters?.errors / metricsDocument.counters?.requests;
    if (errorRate > 0.05) score -= 30; // > 5% error rate
    if (errorRate > 0.01) score -= 15; // > 1% error rate

    return Math.max(0, score);
  }

  getHealthStatus(score) {
    if (score >= 90) return 'healthy';
    if (score >= 70) return 'warning';
    if (score >= 50) return 'critical';
    return 'unhealthy';
  }

  async handleThresholdViolation(violation, metricsDocument) {
    // Handle metrics threshold violations
    console.log(`Threshold violation: ${violation.metric} = ${violation.value} (threshold: ${violation.threshold})`);

    // Store violation record
    await this.db.collection('threshold_violations').insertOne({
      ...violation,
      timestamp: new Date(),
      service: metricsDocument.dimensions?.service,
      environment: metricsDocument.dimensions?.environment,
      metricsDocument: metricsDocument._id
    });
  }

  async handleEventProcessingFailure(eventDocument, error) {
    // Handle event processing failures
    await this.db.collection('event_processing_errors').insertOne({
      eventId: eventDocument.eventId,
      eventType: eventDocument.eventType,
      streamName: eventDocument.streamName,
      error: error.message,
      errorStack: error.stack,
      failedAt: new Date(),
      retryCount: 0
    });
  }

  // Cleanup and shutdown
  async stopAllStreams() {
    const streamPromises = Array.from(this.activeStreams.values());

    // Stop all active streams
    for (const [streamName, streamPromise] of this.activeStreams.entries()) {
      console.log(`Stopping stream: ${streamName}`);
      // Streams will stop when cursors are closed
    }

    await this.cappedManager.cleanup();
    this.activeStreams.clear();

    console.log(`Stopped ${streamPromises.length} streams`);
  }

  // Placeholder methods for event processing
  async processUserActionEvent(eventDocument) { /* Implementation */ }
  async processSystemStateEvent(eventDocument) { /* Implementation */ }
  async processTransactionEvent(eventDocument) { /* Implementation */ }
  async processAlertEvent(eventDocument) { /* Implementation */ }
  async processGenericEvent(eventDocument) { /* Implementation */ }
  async updateLogMetrics(serviceName, logDocument) { /* Implementation */ }
  async updateEventMetrics(streamName, eventDocument) { /* Implementation */ }
  async sendAlertNotification(alert) { /* Implementation */ }
  async escalateAlert(alertId) { /* Implementation */ }
  async handleOutOfMemoryAlert(logDocument) { /* Implementation */ }
  async handleConnectionIssue(logDocument) { /* Implementation */ }
  async trackRequestError(logDocument) { /* Implementation */ }
  async aggregateMetricsData(metricsDocument) { /* Implementation */ }
}