MongoDB Time-Series Data Management and IoT Analytics: Advanced Optimization Patterns for Real-Time Sensor Data Processing

-- Traditional PostgreSQL approach for IoT sensor data - inefficient and problematic

-- Typical IoT sensor data table with poor time-series optimization
CREATE TABLE sensor_readings (
    reading_id BIGSERIAL PRIMARY KEY,
    device_id VARCHAR(100) NOT NULL,
    sensor_type VARCHAR(50) NOT NULL,
    location VARCHAR(200) NOT NULL,
    timestamp TIMESTAMP WITH TIME ZONE NOT NULL,

    -- Multiple sensor measurements (inefficient normalized structure)
    temperature DECIMAL(5,2),
    humidity DECIMAL(5,2),
    pressure DECIMAL(7,2),
    voltage DECIMAL(4,2),
    current DECIMAL(6,3),
    power_consumption DECIMAL(8,2),

    -- Device metadata (repetitive storage)
    device_model VARCHAR(100),
    firmware_version VARCHAR(50),
    installation_date DATE,

    -- Location data (redundant storage)
    building_id VARCHAR(50),
    floor_number INTEGER,
    room_number VARCHAR(20),
    latitude DECIMAL(10,8),
    longitude DECIMAL(11,8),

    -- Quality and status indicators
    signal_strength INTEGER,
    battery_level DECIMAL(5,2),
    reading_quality VARCHAR(20),
    data_source VARCHAR(50),

    -- Audit and tracking fields
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    processed_at TIMESTAMP,
    validation_status VARCHAR(20) DEFAULT 'pending'
);

-- Attempt at optimization with basic indexes (insufficient for time-series workloads)
CREATE INDEX sensor_readings_device_time_idx ON sensor_readings (device_id, timestamp DESC);
CREATE INDEX sensor_readings_timestamp_idx ON sensor_readings (timestamp DESC);
CREATE INDEX sensor_readings_sensor_type_idx ON sensor_readings (sensor_type, timestamp DESC);
CREATE INDEX sensor_readings_location_idx ON sensor_readings (building_id, floor_number, timestamp DESC);

-- Example of problematic IoT data insertion (individual row inserts are slow)
INSERT INTO sensor_readings (
    device_id, sensor_type, location, timestamp,
    temperature, humidity, pressure, voltage, current, power_consumption,
    device_model, firmware_version, installation_date,
    building_id, floor_number, room_number, latitude, longitude,
    signal_strength, battery_level, reading_quality, data_source
) VALUES (
    'TEMP_SENSOR_001_FLOOR_1_ROOM_A', 
    'Environmental Sensor',
    'Building A - Floor 1 - Room A - North Wall',
    CURRENT_TIMESTAMP,
    22.5, 65.3, 1013.25, 3.3, 0.125, 0.4125,
    'SensorTech ST-2000 Environmental Monitor',
    'v2.1.5',
    '2024-01-15',
    'BUILDING_A_MAIN_CAMPUS',
    1,
    'A101',
    40.7128,
    -74.0060,
    85,
    94.2,
    'excellent',
    'real_time_telemetry'
);

-- This approach requires individual inserts for each sensor reading:
-- - Device TEMP_SENSOR_001 sends reading every 30 seconds = 2,880 inserts/day
-- - Device HUMIDITY_SENSOR_002 sends reading every 60 seconds = 1,440 inserts/day  
-- - Device PRESSURE_SENSOR_003 sends reading every 15 seconds = 5,760 inserts/day
-- - For 1,000 devices = millions of individual INSERT operations per day

-- Example of inefficient time-series queries
-- Get hourly averages for temperature sensors (slow aggregation)
SELECT 
    device_id,
    DATE_TRUNC('hour', timestamp) as hour_bucket,

    -- Basic aggregations (computationally expensive)
    AVG(temperature) as avg_temperature,
    MIN(temperature) as min_temperature,
    MAX(temperature) as max_temperature,
    STDDEV(temperature) as temperature_variance,

    -- Count and quality metrics
    COUNT(*) as reading_count,
    COUNT(*) FILTER (WHERE reading_quality = 'excellent') as excellent_readings,
    AVG(signal_strength) as avg_signal_strength,

    -- Power consumption analysis
    AVG(power_consumption) as avg_power,
    SUM(power_consumption) as total_power

FROM sensor_readings
WHERE timestamp >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
  AND sensor_type = 'Environmental Sensor'
  AND temperature IS NOT NULL
GROUP BY device_id, DATE_TRUNC('hour', timestamp)
ORDER BY device_id, hour_bucket DESC;

-- Problems with traditional time-series queries:
-- 1. Full table scans despite indexes when dealing with large time ranges
-- 2. Expensive GROUP BY operations on timestamp truncation
-- 3. Multiple aggregation calculations are computationally intensive
-- 4. Poor performance with concurrent analytical queries
-- 5. No pre-aggregated rollups or materialized views for common patterns
-- 6. Inefficient storage with repeated device metadata in every row

-- Attempting to get device performance trends (extremely slow)
WITH daily_device_metrics AS (
    SELECT 
        device_id,
        DATE_TRUNC('day', timestamp) as date_bucket,

        -- Performance indicators
        AVG(temperature) as avg_temp,
        AVG(humidity) as avg_humidity,
        AVG(pressure) as avg_pressure,
        AVG(power_consumption) as avg_power,
        AVG(battery_level) as avg_battery,
        AVG(signal_strength) as avg_signal,

        -- Operational metrics
        COUNT(*) as total_readings,
        COUNT(*) FILTER (WHERE reading_quality = 'poor') as poor_quality_readings,

        -- Calculate uptime percentage
        (COUNT(*)::DECIMAL / (24 * 60 * 2)) * 100 as uptime_percentage, -- Assuming 30-second intervals

        -- Data quality assessment
        (COUNT(*) FILTER (WHERE reading_quality IN ('excellent', 'good'))::DECIMAL / COUNT(*)) * 100 as quality_percentage

    FROM sensor_readings
    WHERE timestamp >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY device_id, DATE_TRUNC('day', timestamp)
),

device_trend_analysis AS (
    SELECT 
        device_id,

        -- Performance trends using window functions (very expensive)
        AVG(avg_temp) as overall_avg_temp,
        STDDEV(avg_temp) as temp_stability,

        -- Power consumption trends
        AVG(avg_power) as overall_avg_power,

        -- Calculate linear regression slope for temperature trend
        REGR_SLOPE(avg_temp, EXTRACT(EPOCH FROM date_bucket)) as temp_trend_slope,
        REGR_R2(avg_temp, EXTRACT(EPOCH FROM date_bucket)) as temp_trend_correlation,

        -- Battery degradation analysis
        REGR_SLOPE(avg_battery, EXTRACT(EPOCH FROM date_bucket)) as battery_degradation_slope,

        -- Reliability metrics
        AVG(uptime_percentage) as avg_uptime,
        MIN(quality_percentage) as min_quality_percentage,

        -- Trend classification
        CASE 
            WHEN REGR_SLOPE(avg_temp, EXTRACT(EPOCH FROM date_bucket)) > 0.01 THEN 'temperature_increasing'
            WHEN REGR_SLOPE(avg_temp, EXTRACT(EPOCH FROM date_bucket)) < -0.01 THEN 'temperature_decreasing'
            ELSE 'temperature_stable'
        END as temperature_trend,

        CASE 
            WHEN REGR_SLOPE(avg_battery, EXTRACT(EPOCH FROM date_bucket)) < -0.1 THEN 'battery_degrading'
            WHEN REGR_SLOPE(avg_battery, EXTRACT(EPOCH FROM date_bucket)) > 0.1 THEN 'battery_improving'
            ELSE 'battery_stable'
        END as battery_trend

    FROM daily_device_metrics
    GROUP BY device_id
)

-- This query would take minutes or hours on large datasets
SELECT 
    device_id,
    overall_avg_temp,
    temp_stability,
    overall_avg_power,
    temp_trend_slope,
    battery_degradation_slope,
    avg_uptime,
    temperature_trend,
    battery_trend,

    -- Device health scoring
    CASE 
        WHEN avg_uptime > 95 AND min_quality_percentage > 90 AND battery_degradation_slope > -0.05 THEN 'excellent'
        WHEN avg_uptime > 90 AND min_quality_percentage > 80 AND battery_degradation_slope > -0.1 THEN 'good'
        WHEN avg_uptime > 80 AND min_quality_percentage > 70 THEN 'fair'
        ELSE 'poor'
    END as device_health_grade

FROM device_trend_analysis
ORDER BY overall_avg_power DESC;

-- Storage analysis showing massive inefficiency
SELECT 
    schemaname,
    tablename,

    -- Table size metrics showing bloat
    pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as total_size,
    pg_size_pretty(pg_relation_size(schemaname||'.'||tablename)) as table_size,
    pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename) - pg_relation_size(schemaname||'.'||tablename)) as indexes_size,

    -- Row statistics
    n_tup_ins as total_rows,
    CASE 
        WHEN n_tup_ins > 0 THEN 
            pg_relation_size(schemaname||'.'||tablename) / n_tup_ins 
        ELSE 0 
    END as avg_row_size_bytes,

    -- Storage efficiency (poor due to repetitive data)
    CASE 
        WHEN n_tup_ins > 0 THEN
            ROUND((pg_relation_size(schemaname||'.'||tablename)::DECIMAL / 1024 / 1024 / 1024), 2)
        ELSE 0.0
    END as storage_size_gb

FROM pg_stat_user_tables 
WHERE tablename = 'sensor_readings'
ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC;

-- Traditional approach problems:
-- 1. Massive storage bloat due to repetitive device metadata in every row
-- 2. Poor compression due to mixed data types and variable-length fields
-- 3. Inefficient indexing strategies that don't leverage time-series patterns
-- 4. Slow aggregation queries that require full scans and expensive calculations
-- 5. No built-in support for time-based data retention and archival
-- 6. Complex partitioning strategies required for reasonable performance
-- 7. Poor concurrent read/write performance for mixed analytical and operational workloads
-- 8. Difficult real-time analytics due to lack of incremental aggregation support
-- 9. No specialized compression for time-series data patterns
-- 10. Manual optimization required for downsample and rollup operations

-- PostgreSQL time partitioning attempt (complex and limited)
-- Manual partitioning by month (operational overhead)
CREATE TABLE sensor_readings_2024_12 PARTITION OF sensor_readings
FOR VALUES FROM ('2024-12-01') TO ('2025-01-01');

CREATE TABLE sensor_readings_2024_11 PARTITION OF sensor_readings  
FOR VALUES FROM ('2024-11-01') TO ('2024-12-01');

-- Even with partitioning:
-- 1. Manual partition management overhead
-- 2. Limited optimization for time-series specific operations
-- 3. No automatic data retention and archival policies
-- 4. Complex cross-partition analytics and aggregation
-- 5. Poor storage compression compared to purpose-built time-series systems

// MongoDB Advanced Time-Series Data Management for IoT Analytics

const { MongoClient, ObjectId } = require('mongodb');

// Comprehensive IoT Time-Series Data Management System
class MongoDBIoTTimeSeriesManager {
  constructor(connectionString, config = {}) {
    this.client = new MongoClient(connectionString);
    this.db = null;
    this.timeSeriesCollections = new Map();

    // Advanced time-series configuration for IoT workloads
    this.config = {
      // Time-series collection optimization
      timeSeries: {
        metaField: 'device', // Device metadata field
        timeField: 'timestamp', // Time field
        granularity: 'seconds', // Data granularity: seconds, minutes, hours
        enableOptimization: true,
        enableCompression: true
      },

      // IoT-specific optimization settings
      iotOptimization: {
        enableDeviceBatching: config.enableDeviceBatching !== false,
        enableSensorGrouping: config.enableSensorGrouping !== false,
        enableDataValidation: config.enableDataValidation !== false,
        enableRealTimeAggregation: config.enableRealTimeAggregation !== false,
        enableAnomalyDetection: config.enableAnomalyDetection || false
      },

      // Data retention and archival
      dataRetention: {
        enableAutoArchival: config.enableAutoArchival || false,
        hotDataRetentionDays: config.hotDataRetentionDays || 30,
        warmDataRetentionDays: config.warmDataRetentionDays || 90,
        coldDataRetentionDays: config.coldDataRetentionDays || 365,
        enableCompression: config.enableRetentionCompression !== false
      },

      // Performance and monitoring
      performance: {
        enableMetrics: config.enableMetrics !== false,
        enableAlerts: config.enableAlerts || false,
        enablePerformanceOptimization: config.enablePerformanceOptimization !== false,
        maxConcurrentReads: config.maxConcurrentReads || 10,
        maxConcurrentWrites: config.maxConcurrentWrites || 5
      },

      // Real-time analytics
      analytics: {
        enableRealTimeAggregation: config.enableRealTimeAggregation !== false,
        enableTrendAnalysis: config.enableTrendAnalysis || false,
        enablePredictiveAnalytics: config.enablePredictiveAnalytics || false,
        aggregationWindowSizes: config.aggregationWindowSizes || ['1m', '5m', '15m', '1h', '1d']
      }
    };

    this.deviceRegistry = new Map();
    this.metricsCollector = null;
    this.alertManager = null;
  }

  async initialize() {
    console.log('Initializing MongoDB IoT Time-Series management system...');

    try {
      await this.client.connect();
      this.db = this.client.db('iot_time_series_platform');

      // Setup time-series collections for different sensor types
      await this.setupTimeSeriesCollections();

      // Initialize device registry and metadata management
      await this.setupDeviceManagement();

      // Setup real-time aggregation pipelines
      await this.setupRealTimeAggregation();

      // Initialize monitoring and alerting
      await this.setupMonitoringAndAlerting();

      console.log('IoT Time-Series system initialized successfully');

    } catch (error) {
      console.error('Error initializing IoT Time-Series system:', error);
      throw error;
    }
  }

  async setupTimeSeriesCollections() {
    console.log('Setting up optimized time-series collections for IoT data...');

    try {
      // Environmental sensor data collection
      await this.createTimeSeriesCollection('environmental_sensors', {
        metaField: 'device',
        timeField: 'timestamp',
        granularity: 'seconds'
      });

      // Power monitoring collection
      await this.createTimeSeriesCollection('power_monitoring', {
        metaField: 'device',
        timeField: 'timestamp',
        granularity: 'seconds'
      });

      // Network performance collection
      await this.createTimeSeriesCollection('network_performance', {
        metaField: 'device',
        timeField: 'timestamp',
        granularity: 'seconds'
      });

      // Device health and diagnostics collection
      await this.createTimeSeriesCollection('device_diagnostics', {
        metaField: 'device',
        timeField: 'timestamp',
        granularity: 'minutes'
      });

      // Security events collection
      await this.createTimeSeriesCollection('security_events', {
        metaField: 'device',
        timeField: 'timestamp',
        granularity: 'seconds'
      });

    } catch (error) {
      console.error('Error setting up time-series collections:', error);
      throw error;
    }
  }

  async createTimeSeriesCollection(collectionName, timeSeriesOptions) {
    console.log(`Creating optimized time-series collection: ${collectionName}`);

    try {
      // Create time-series collection with MongoDB native optimization
      await this.db.createCollection(collectionName, {
        timeseries: {
          timeField: timeSeriesOptions.timeField,
          metaField: timeSeriesOptions.metaField,
          granularity: timeSeriesOptions.granularity
        },
        // Enable compression for storage efficiency
        storageEngine: {
          wiredTiger: {
            configString: 'block_compressor=zstd'
          }
        }
      });

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

      // Create optimized indexes for time-series access patterns
      await collection.createIndex({ 
        [timeSeriesOptions.metaField + '.device_id']: 1, 
        [timeSeriesOptions.timeField]: -1 
      });

      await collection.createIndex({ 
        [timeSeriesOptions.metaField + '.location']: 1, 
        [timeSeriesOptions.timeField]: -1 
      });

      await collection.createIndex({ 
        [timeSeriesOptions.metaField + '.sensor_type']: 1, 
        [timeSeriesOptions.timeField]: -1 
      });

      // Store collection reference
      this.timeSeriesCollections.set(collectionName, {
        collection: collection,
        options: timeSeriesOptions
      });

      console.log(`Time-series collection ${collectionName} created successfully`);

    } catch (error) {
      console.error(`Error creating time-series collection ${collectionName}:`, error);
      throw error;
    }
  }

  async setupDeviceManagement() {
    console.log('Setting up device registry and metadata management...');

    const deviceRegistry = this.db.collection('device_registry');

    // Create indexes for device management
    await deviceRegistry.createIndex({ device_id: 1 }, { unique: true });
    await deviceRegistry.createIndex({ location: 1 });
    await deviceRegistry.createIndex({ sensor_type: 1 });
    await deviceRegistry.createIndex({ status: 1, last_seen: -1 });

    // Create device groups collection for hierarchical organization
    const deviceGroups = this.db.collection('device_groups');
    await deviceGroups.createIndex({ group_id: 1 }, { unique: true });
    await deviceGroups.createIndex({ parent_group: 1 });
  }

  async registerIoTDevice(deviceData) {
    console.log(`Registering IoT device: ${deviceData.device_id}`);

    try {
      const deviceRegistry = this.db.collection('device_registry');

      // Check for existing device
      const existingDevice = await deviceRegistry.findOne({ device_id: deviceData.device_id });
      if (existingDevice) {
        throw new Error(`Device already registered: ${deviceData.device_id}`);
      }

      const deviceRegistration = {
        device_id: deviceData.device_id,
        device_name: deviceData.device_name,
        sensor_type: deviceData.sensor_type,

        // Location and installation information
        location: {
          building: deviceData.building,
          floor: deviceData.floor,
          room: deviceData.room,
          coordinates: {
            latitude: deviceData.latitude || null,
            longitude: deviceData.longitude || null
          },
          zone: deviceData.zone || null
        },

        // Technical specifications
        specifications: {
          model: deviceData.model,
          manufacturer: deviceData.manufacturer,
          firmware_version: deviceData.firmware_version,
          hardware_revision: deviceData.hardware_revision,
          communication_protocol: deviceData.communication_protocol || 'mqtt',
          power_source: deviceData.power_source || 'battery',
          sampling_rate: deviceData.sampling_rate || 60 // seconds
        },

        // Operational configuration
        configuration: {
          data_retention_days: deviceData.data_retention_days || 365,
          alert_thresholds: deviceData.alert_thresholds || {},
          measurement_units: deviceData.measurement_units || {},
          calibration_data: deviceData.calibration_data || {},
          quality_settings: deviceData.quality_settings || {}
        },

        // Status and monitoring
        status: 'active',
        health_status: 'unknown',
        last_seen: null,
        last_reading: null,

        // Registration metadata
        registered_at: new Date(),
        updated_at: new Date(),
        registered_by: deviceData.registered_by || 'system'
      };

      const result = await deviceRegistry.insertOne(deviceRegistration);

      // Add to in-memory registry for quick access
      this.deviceRegistry.set(deviceData.device_id, deviceRegistration);

      console.log(`Device ${deviceData.device_id} registered successfully`);
      return {
        success: true,
        device_id: deviceData.device_id,
        registration_id: result.insertedId
      };

    } catch (error) {
      console.error(`Error registering device ${deviceData.device_id}:`, error);
      throw error;
    }
  }

  async insertSensorData(collectionName, deviceId, sensorData, options = {}) {
    console.log(`Inserting sensor data for device ${deviceId} into ${collectionName}`);

    try {
      const timeSeriesInfo = this.timeSeriesCollections.get(collectionName);
      if (!timeSeriesInfo) {
        throw new Error(`Time-series collection not found: ${collectionName}`);
      }

      const collection = timeSeriesInfo.collection;

      // Get device metadata
      const deviceInfo = await this.getDeviceInfo(deviceId);
      if (!deviceInfo) {
        throw new Error(`Device not registered: ${deviceId}`);
      }

      // Prepare optimized time-series document
      const timeSeriesDocument = {
        timestamp: sensorData.timestamp || new Date(),

        // Device metadata (stored efficiently in metaField)
        device: {
          device_id: deviceId,
          sensor_type: deviceInfo.sensor_type,
          location: deviceInfo.location,
          model: deviceInfo.specifications.model
        },

        // Sensor measurements (optimized structure)
        measurements: this.optimizeMeasurements(sensorData.measurements),

        // Data quality indicators
        quality: {
          signal_strength: sensorData.signal_strength,
          battery_level: sensorData.battery_level,
          reading_quality: sensorData.reading_quality || 'good',
          data_source: sensorData.data_source || 'sensor'
        },

        // Operational metadata
        metadata: {
          firmware_version: deviceInfo.specifications.firmware_version,
          sampling_interval: deviceInfo.specifications.sampling_rate,
          processing_flags: sensorData.processing_flags || []
        }
      };

      // Apply data validation if enabled
      if (this.config.iotOptimization.enableDataValidation) {
        await this.validateSensorData(deviceId, timeSeriesDocument);
      }

      // Insert with time-series optimization
      const result = await collection.insertOne(timeSeriesDocument, {
        writeConcern: { w: 'majority', j: true }
      });

      // Update device last seen timestamp
      await this.updateDeviceLastSeen(deviceId);

      // Trigger real-time aggregation if enabled
      if (this.config.analytics.enableRealTimeAggregation) {
        await this.triggerRealTimeAggregation(collectionName, deviceId, timeSeriesDocument);
      }

      return {
        success: true,
        inserted_id: result.insertedId,
        collection: collectionName,
        device_id: deviceId,
        timestamp: timeSeriesDocument.timestamp
      };

    } catch (error) {
      console.error(`Error inserting sensor data for device ${deviceId}:`, error);
      throw error;
    }
  }

  optimizeMeasurements(measurements) {
    if (!measurements || typeof measurements !== 'object') {
      return measurements;
    }

    // Optimize measurement structure for compression and query performance
    const optimized = {};

    Object.entries(measurements).forEach(([key, value]) => {
      if (typeof value === 'number') {
        // Round to reasonable precision to improve compression
        optimized[key] = Math.round(value * 100) / 100;
      } else {
        optimized[key] = value;
      }
    });

    return optimized;
  }

  async validateSensorData(deviceId, document) {
    const deviceInfo = this.deviceRegistry.get(deviceId);
    if (!deviceInfo) return;

    // Apply device-specific validation rules
    const thresholds = deviceInfo.configuration.alert_thresholds;

    Object.entries(document.measurements).forEach(([measurement, value]) => {
      if (thresholds[measurement]) {
        const threshold = thresholds[measurement];

        if (value < threshold.min || value > threshold.max) {
          // Log anomaly but don't reject data
          console.warn(`Measurement ${measurement} out of range for device ${deviceId}: ${value}`);
          document.quality.anomaly_detected = true;
          document.quality.anomaly_type = `${measurement}_out_of_range`;
        }
      }
    });
  }

  async getDeviceInfo(deviceId) {
    if (this.deviceRegistry.has(deviceId)) {
      return this.deviceRegistry.get(deviceId);
    }

    const deviceRegistry = this.db.collection('device_registry');
    const device = await deviceRegistry.findOne({ device_id: deviceId });

    if (device) {
      this.deviceRegistry.set(deviceId, device);
    }

    return device;
  }

  async updateDeviceLastSeen(deviceId) {
    try {
      const deviceRegistry = this.db.collection('device_registry');

      await deviceRegistry.updateOne(
        { device_id: deviceId },
        { 
          $set: { 
            last_seen: new Date(),
            updated_at: new Date()
          }
        }
      );

      // Update in-memory cache
      if (this.deviceRegistry.has(deviceId)) {
        const device = this.deviceRegistry.get(deviceId);
        device.last_seen = new Date();
        device.updated_at = new Date();
      }

    } catch (error) {
      console.error(`Error updating last seen for device ${deviceId}:`, error);
      // Don't throw - last seen updates shouldn't break data insertion
    }
  }

  async queryTimeSeriesData(collectionName, query = {}, options = {}) {
    console.log(`Querying time-series data from ${collectionName}`);

    try {
      const timeSeriesInfo = this.timeSeriesCollections.get(collectionName);
      if (!timeSeriesInfo) {
        throw new Error(`Time-series collection not found: ${collectionName}`);
      }

      const collection = timeSeriesInfo.collection;
      const startTime = Date.now();

      // Apply time-series optimized query patterns
      const optimizedQuery = this.optimizeTimeSeriesQuery(query);
      const optimizedOptions = this.optimizeQueryOptions(options);

      const results = await collection.find(optimizedQuery, optimizedOptions).toArray();
      const queryTime = Date.now() - startTime;

      console.log(`Time-series query completed in ${queryTime}ms, returned ${results.length} documents`);

      return {
        data: results,
        metadata: {
          collection: collectionName,
          query_time_ms: queryTime,
          document_count: results.length,
          query: optimizedQuery
        }
      };

    } catch (error) {
      console.error(`Error querying time-series data from ${collectionName}:`, error);
      throw error;
    }
  }

  optimizeTimeSeriesQuery(query) {
    const optimized = { ...query };

    // Optimize time range queries for time-series performance
    if (query.timestamp) {
      optimized.timestamp = query.timestamp;
    } else if (query.timeRange) {
      optimized.timestamp = {
        $gte: new Date(query.timeRange.start),
        $lte: new Date(query.timeRange.end)
      };
      delete optimized.timeRange;
    }

    // Optimize device filtering for metadata field
    if (query.device_id) {
      optimized['device.device_id'] = query.device_id;
      delete optimized.device_id;
    }

    if (query.sensor_type) {
      optimized['device.sensor_type'] = query.sensor_type;
      delete optimized.sensor_type;
    }

    return optimized;
  }

  optimizeQueryOptions(options) {
    const optimized = {
      ...options,
      // Default sort by timestamp descending for time-series efficiency
      sort: options.sort || { timestamp: -1 },
      // Limit results for performance unless explicitly requested
      limit: options.limit || 1000
    };

    return optimized;
  }

  // Advanced time-series aggregation operations
  async generateDeviceAnalytics(deviceId, timeRange, granularity = '1h') {
    console.log(`Generating analytics for device ${deviceId} with ${granularity} granularity`);

    try {
      const pipeline = [
        {
          $match: {
            'device.device_id': deviceId,
            timestamp: {
              $gte: new Date(timeRange.start),
              $lte: new Date(timeRange.end)
            }
          }
        },
        {
          $group: {
            _id: {
              $dateTrunc: {
                date: '$timestamp',
                unit: this.getTimeUnit(granularity),
                binSize: this.getBinSize(granularity)
              }
            },

            // Basic statistics
            reading_count: { $sum: 1 },
            avg_signal_strength: { $avg: '$quality.signal_strength' },
            avg_battery_level: { $avg: '$quality.battery_level' },

            // Measurement aggregations (dynamic based on sensor type)
            measurements: {
              $push: '$measurements'
            }
          }
        },
        {
          $addFields: {
            timestamp: '$_id',
            // Calculate measurement statistics
            measurement_stats: {
              $reduce: {
                input: '$measurements',
                initialValue: {},
                in: this.createMeasurementReducer()
              }
            }
          }
        },
        {
          $project: {
            _id: 0,
            timestamp: 1,
            reading_count: 1,
            avg_signal_strength: 1,
            avg_battery_level: 1,
            measurement_stats: 1
          }
        },
        {
          $sort: { timestamp: 1 }
        }
      ];

      // Execute aggregation on appropriate collection
      const collection = this.getCollectionForDevice(deviceId);
      const results = await collection.aggregate(pipeline).toArray();

      return {
        device_id: deviceId,
        time_range: timeRange,
        granularity: granularity,
        analytics: results
      };

    } catch (error) {
      console.error(`Error generating analytics for device ${deviceId}:`, error);
      throw error;
    }
  }

  async generateFleetAnalytics(fleetQuery, timeRange, granularity = '1h') {
    console.log(`Generating fleet analytics with ${granularity} granularity`);

    try {
      const pipeline = [
        {
          $match: {
            ...this.optimizeTimeSeriesQuery(fleetQuery),
            timestamp: {
              $gte: new Date(timeRange.start),
              $lte: new Date(timeRange.end)
            }
          }
        },
        {
          $group: {
            _id: {
              time_bucket: {
                $dateTrunc: {
                  date: '$timestamp',
                  unit: this.getTimeUnit(granularity),
                  binSize: this.getBinSize(granularity)
                }
              },
              device_id: '$device.device_id',
              location: '$device.location.building'
            },

            // Device-level aggregations
            reading_count: { $sum: 1 },
            avg_battery: { $avg: '$quality.battery_level' },
            avg_signal: { $avg: '$quality.signal_strength' },

            // Measurement aggregations
            measurements: { $push: '$measurements' }
          }
        },
        {
          $group: {
            _id: '$_id.time_bucket',

            // Fleet-level metrics
            total_devices: { $sum: 1 },
            total_readings: { $sum: '$reading_count' },
            avg_fleet_battery: { $avg: '$avg_battery' },
            avg_fleet_signal: { $avg: '$avg_signal' },

            // Device health distribution
            healthy_devices: {
              $sum: {
                $cond: [
                  { $and: [
                    { $gte: ['$avg_battery', 20] },
                    { $gte: ['$avg_signal', 50] }
                  ]},
                  1,
                  0
                ]
              }
            },

            // Location distribution
            locations: {
              $addToSet: '$_id.location'
            }
          }
        },
        {
          $addFields: {
            timestamp: '$_id',
            device_health_percentage: {
              $multiply: [
                { $divide: ['$healthy_devices', '$total_devices'] },
                100
              ]
            }
          }
        },
        {
          $project: {
            _id: 0,
            timestamp: 1,
            total_devices: 1,
            total_readings: 1,
            avg_fleet_battery: { $round: ['$avg_fleet_battery', 2] },
            avg_fleet_signal: { $round: ['$avg_fleet_signal', 2] },
            device_health_percentage: { $round: ['$device_health_percentage', 2] },
            healthy_devices: 1,
            unique_locations: { $size: '$locations' }
          }
        },
        {
          $sort: { timestamp: 1 }
        }
      ];

      const collection = this.db.collection('environmental_sensors'); // Default collection
      const results = await collection.aggregate(pipeline).toArray();

      return {
        fleet_query: fleetQuery,
        time_range: timeRange,
        granularity: granularity,
        analytics: results
      };

    } catch (error) {
      console.error(`Error generating fleet analytics:`, error);
      throw error;
    }
  }

  getTimeUnit(granularity) {
    const unitMap = {
      '1s': 'second',
      '1m': 'minute',
      '5m': 'minute',
      '15m': 'minute',
      '1h': 'hour',
      '1d': 'day'
    };
    return unitMap[granularity] || 'hour';
  }

  getBinSize(granularity) {
    const binMap = {
      '1s': 1,
      '1m': 1,
      '5m': 5,
      '15m': 15,
      '1h': 1,
      '1d': 1
    };
    return binMap[granularity] || 1;
  }

  createMeasurementReducer() {
    return {
      $function: {
        body: `function(accumulator, measurement) {
          for (let key in measurement) {
            if (typeof measurement[key] === 'number') {
              if (!accumulator[key]) {
                accumulator[key] = { sum: 0, count: 0, min: measurement[key], max: measurement[key] };
              }
              accumulator[key].sum += measurement[key];
              accumulator[key].count++;
              accumulator[key].min = Math.min(accumulator[key].min, measurement[key]);
              accumulator[key].max = Math.max(accumulator[key].max, measurement[key]);
              accumulator[key].avg = accumulator[key].sum / accumulator[key].count;
            }
          }
          return accumulator;
        }`,
        args: ['$$value', '$$this'],
        lang: 'js'
      }
    };
  }

  getCollectionForDevice(deviceId) {
    // Simple mapping - in production, this would use device registry
    return this.timeSeriesCollections.get('environmental_sensors').collection;
  }

  async setupRealTimeAggregation() {
    if (!this.config.analytics.enableRealTimeAggregation) {
      return;
    }

    console.log('Setting up real-time aggregation pipelines...');

    // Setup materialized views for common aggregations
    await this.createMaterializedViews();

    // Setup change streams for real-time updates
    await this.setupChangeStreams();
  }

  async createMaterializedViews() {
    console.log('Creating materialized views for real-time analytics...');

    // Hourly device summaries
    await this.db.createCollection('device_hourly_summary');

    // Daily fleet metrics  
    await this.db.createCollection('fleet_daily_metrics');

    // Real-time alerts collection
    await this.db.createCollection('real_time_alerts');
  }

  async setupChangeStreams() {
    console.log('Setting up change streams for real-time processing...');

    for (const [collectionName, collectionInfo] of this.timeSeriesCollections) {
      const collection = collectionInfo.collection;

      const changeStream = collection.watch([], { fullDocument: 'updateLookup' });

      changeStream.on('change', async (change) => {
        if (change.operationType === 'insert') {
          await this.processRealTimeInsert(collectionName, change.fullDocument);
        }
      });
    }
  }

  async processRealTimeInsert(collectionName, document) {
    try {
      // Update real-time aggregations
      await this.updateRealTimeAggregations(collectionName, document);

      // Check for alerts
      if (this.config.performance.enableAlerts) {
        await this.checkAlertConditions(document);
      }

    } catch (error) {
      console.error(`Error processing real-time insert:`, error);
      // Don't throw - real-time processing shouldn't break inserts
    }
  }

  async updateRealTimeAggregations(collectionName, document) {
    const deviceId = document.device.device_id;
    const timestamp = document.timestamp;
    const hourBucket = new Date(timestamp.getFullYear(), timestamp.getMonth(), timestamp.getDate(), timestamp.getHours());

    // Update hourly device summary
    const deviceHourlySummary = this.db.collection('device_hourly_summary');

    await deviceHourlySummary.updateOne(
      {
        device_id: deviceId,
        hour_bucket: hourBucket
      },
      {
        $inc: {
          reading_count: 1,
          'quality.signal_strength_sum': document.quality.signal_strength || 0,
          'quality.battery_level_sum': document.quality.battery_level || 0
        },
        $min: {
          'timestamp.first_reading': timestamp
        },
        $max: {
          'timestamp.last_reading': timestamp
        },
        $set: {
          device: document.device,
          updated_at: new Date()
        }
      },
      { upsert: true }
    );
  }

  async checkAlertConditions(document) {
    const deviceId = document.device.device_id;
    const deviceInfo = await this.getDeviceInfo(deviceId);

    if (!deviceInfo || !deviceInfo.configuration.alert_thresholds) {
      return;
    }

    const thresholds = deviceInfo.configuration.alert_thresholds;
    const alerts = [];

    // Check measurement thresholds
    Object.entries(document.measurements).forEach(([measurement, value]) => {
      const threshold = thresholds[measurement];
      if (threshold) {
        if (value < threshold.critical_min || value > threshold.critical_max) {
          alerts.push({
            type: 'critical_threshold',
            measurement: measurement,
            value: value,
            threshold: threshold,
            severity: 'critical'
          });
        } else if (value < threshold.warning_min || value > threshold.warning_max) {
          alerts.push({
            type: 'warning_threshold',
            measurement: measurement,
            value: value,
            threshold: threshold,
            severity: 'warning'
          });
        }
      }
    });

    // Check device health
    if (document.quality.battery_level < 20) {
      alerts.push({
        type: 'low_battery',
        value: document.quality.battery_level,
        severity: 'warning'
      });
    }

    if (document.quality.signal_strength < 30) {
      alerts.push({
        type: 'poor_signal',
        value: document.quality.signal_strength,
        severity: 'warning'
      });
    }

    // Store alerts if any
    if (alerts.length > 0) {
      await this.storeAlerts(deviceId, document.timestamp, alerts);
    }
  }

  async storeAlerts(deviceId, timestamp, alerts) {
    const realTimeAlerts = this.db.collection('real_time_alerts');

    const alertDocument = {
      device_id: deviceId,
      timestamp: timestamp,
      alerts: alerts,
      status: 'active',
      created_at: new Date()
    };

    await realTimeAlerts.insertOne(alertDocument);
    console.log(`Stored ${alerts.length} alerts for device ${deviceId}`);
  }

  async setupMonitoringAndAlerting() {
    if (this.config.performance.enableMetrics) {
      console.log('Setting up performance monitoring and alerting...');

      // Create metrics collection
      const metricsCollection = this.db.collection('system_metrics');
      await metricsCollection.createIndex({ timestamp: -1 });
      await metricsCollection.createIndex({ metric_type: 1, timestamp: -1 });

      // Start metrics collection
      this.startMetricsCollection();
    }
  }

  startMetricsCollection() {
    setInterval(async () => {
      try {
        await this.collectSystemMetrics();
      } catch (error) {
        console.error('Error collecting system metrics:', error);
      }
    }, 60000); // Collect metrics every minute
  }

  async collectSystemMetrics() {
    const timestamp = new Date();

    // Collect performance metrics for each time-series collection
    for (const [collectionName, collectionInfo] of this.timeSeriesCollections) {
      const collection = collectionInfo.collection;

      // Get collection stats
      const stats = await this.db.runCommand({ collStats: collectionName });

      // Calculate recent insertion rate
      const oneMinuteAgo = new Date(timestamp.getTime() - 60000);
      const recentInserts = await collection.countDocuments({
        timestamp: { $gte: oneMinuteAgo }
      });

      const metrics = {
        timestamp: timestamp,
        metric_type: 'collection_performance',
        collection_name: collectionName,
        metrics: {
          document_count: stats.count,
          storage_size: stats.storageSize,
          index_size: stats.totalIndexSize,
          recent_inserts_per_minute: recentInserts,
          avg_object_size: stats.avgObjSize
        }
      };

      await this.db.collection('system_metrics').insertOne(metrics);
    }
  }
}

// Example usage: Comprehensive IoT time-series platform
async function demonstrateIoTTimeSeriesManagement() {
  const iotManager = new MongoDBIoTTimeSeriesManager('mongodb://localhost:27017', {
    enableRealTimeAggregation: true,
    enableAnomalyDetection: true,
    enableMetrics: true,
    hotDataRetentionDays: 30,
    warmDataRetentionDays: 90
  });

  await iotManager.initialize();

  // Register IoT devices
  await iotManager.registerIoTDevice({
    device_id: 'ENV_SENSOR_001',
    device_name: 'Environmental Sensor - Conference Room A',
    sensor_type: 'environmental',
    building: 'Main Office',
    floor: 2,
    room: 'Conference Room A',
    model: 'SensorTech ST-ENV-2000',
    manufacturer: 'SensorTech Corp',
    firmware_version: 'v2.1.3',
    communication_protocol: 'mqtt',
    sampling_rate: 60,
    alert_thresholds: {
      temperature: { min: 18, max: 26, warning_min: 20, warning_max: 24, critical_min: 15, critical_max: 30 },
      humidity: { min: 30, max: 70, warning_min: 40, warning_max: 60, critical_min: 20, critical_max: 80 }
    }
  });

  // Insert time-series sensor data
  await iotManager.insertSensorData('environmental_sensors', 'ENV_SENSOR_001', {
    measurements: {
      temperature: 22.5,
      humidity: 55.2,
      pressure: 1013.25,
      light_level: 450
    },
    signal_strength: 85,
    battery_level: 92,
    reading_quality: 'excellent'
  });

  // Generate device analytics
  const deviceAnalytics = await iotManager.generateDeviceAnalytics(
    'ENV_SENSOR_001',
    {
      start: new Date(Date.now() - 24 * 60 * 60 * 1000), // 24 hours ago
      end: new Date()
    },
    '1h'
  );

  console.log('Device Analytics:', deviceAnalytics);

  // Generate fleet analytics
  const fleetAnalytics = await iotManager.generateFleetAnalytics(
    { 'device.sensor_type': 'environmental' },
    {
      start: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000), // 7 days ago
      end: new Date()
    },
    '1d'
  );

  console.log('Fleet Analytics:', fleetAnalytics);
}

module.exports = {
  MongoDBIoTTimeSeriesManager
};