MongoDB Bulk Operations and High-Throughput Data Processing: Advanced Batch Processing Patterns for Enterprise Applications

-- Traditional PostgreSQL batch processing - inefficient and resource-intensive

-- Individual insert operations with poor performance characteristics
CREATE TABLE user_analytics (
    user_id BIGINT NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    event_data JSONB,
    session_id VARCHAR(100),
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    device_info JSONB,
    location_data JSONB,

    -- Performance indexes that become expensive with individual operations
    INDEX idx_user_timestamp (user_id, timestamp),
    INDEX idx_event_type (event_type),
    INDEX idx_session_id (session_id)
);

-- Slow row-by-row processing approach
DO $$
DECLARE
    event_record RECORD;
    total_records INTEGER := 0;
    batch_size INTEGER := 1000;
    start_time TIMESTAMP;
    processing_time INTERVAL;
BEGIN
    start_time := CURRENT_TIMESTAMP;

    -- Inefficient cursor-based processing
    FOR event_record IN 
        SELECT * FROM staging_events WHERE processed = false
        ORDER BY created_at
        LIMIT 100000
    LOOP
        BEGIN
            -- Individual INSERT operations are extremely slow
            INSERT INTO user_analytics (
                user_id, event_type, event_data, session_id, 
                device_info, location_data
            ) VALUES (
                event_record.user_id,
                event_record.event_type,
                event_record.event_data::JSONB,
                event_record.session_id,
                event_record.device_info::JSONB,
                event_record.location_data::JSONB
            );

            -- Update staging table (another slow operation)
            UPDATE staging_events 
            SET processed = true, processed_at = CURRENT_TIMESTAMP
            WHERE id = event_record.id;

            total_records := total_records + 1;

            -- Manual batching for commit control
            IF total_records % batch_size = 0 THEN
                COMMIT;
                RAISE NOTICE 'Processed % records', total_records;
            END IF;

        EXCEPTION WHEN OTHERS THEN
            -- Individual error handling is complex
            INSERT INTO error_log (
                operation, error_message, failed_data, timestamp
            ) VALUES (
                'user_analytics_insert',
                SQLERRM,
                row_to_json(event_record),
                CURRENT_TIMESTAMP
            );
            ROLLBACK;
        END;
    END LOOP;

    processing_time := CURRENT_TIMESTAMP - start_time;
    RAISE NOTICE 'Completed processing % records in %', total_records, processing_time;
END $$;

-- Traditional batch UPDATE operations with poor performance
UPDATE user_analytics ua
SET 
    location_data = COALESCE(loc.location_info, ua.location_data),
    device_info = device_info || COALESCE(dev.device_updates, '{}'::jsonb),
    last_updated = CURRENT_TIMESTAMP
FROM (
    -- Expensive JOIN operations on large datasets
    SELECT DISTINCT ON (user_id, session_id)
        user_id,
        session_id, 
        location_info,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY timestamp DESC) as rn
    FROM location_updates 
    WHERE processed = false
) loc,
(
    SELECT DISTINCT ON (user_id, session_id)
        user_id,
        session_id,
        device_updates
    FROM device_updates
    WHERE processed = false
) dev
WHERE ua.user_id = loc.user_id 
  AND ua.session_id = loc.session_id
  AND ua.user_id = dev.user_id
  AND ua.session_id = dev.session_id
  AND ua.timestamp >= CURRENT_DATE - INTERVAL '7 days';

-- Complex aggregation processing with performance issues
WITH hourly_metrics AS (
    SELECT 
        user_id,
        DATE_TRUNC('hour', timestamp) as hour_bucket,
        event_type,
        COUNT(*) as event_count,
        COUNT(DISTINCT session_id) as unique_sessions,

        -- Expensive JSON operations on large datasets
        AVG(CAST(event_data->>'duration' AS NUMERIC)) as avg_duration,
        SUM(CAST(event_data->>'value' AS NUMERIC)) as total_value,

        -- Complex device and location aggregations
        COUNT(DISTINCT device_info->>'device_id') as unique_devices,
        ARRAY_AGG(DISTINCT location_data->>'country') FILTER (WHERE location_data->>'country' IS NOT NULL) as countries

    FROM user_analytics
    WHERE timestamp >= CURRENT_DATE - INTERVAL '1 day'
    GROUP BY user_id, DATE_TRUNC('hour', timestamp), event_type
),
user_daily_summary AS (
    SELECT 
        user_id,
        DATE_TRUNC('day', hour_bucket) as date_bucket,

        -- Multiple aggregation levels cause performance problems
        SUM(event_count) as total_events,
        SUM(unique_sessions) as total_sessions,
        AVG(avg_duration) as overall_avg_duration,
        SUM(total_value) as daily_value,

        -- Array operations are expensive
        ARRAY_AGG(DISTINCT unnest(countries)) as all_countries,
        COUNT(DISTINCT event_type) as event_type_diversity

    FROM hourly_metrics
    GROUP BY user_id, DATE_TRUNC('day', hour_bucket)
)
-- Expensive UPSERT operations
INSERT INTO user_daily_summaries (
    user_id, summary_date, total_events, total_sessions,
    avg_duration, daily_value, countries_visited, event_diversity,
    computed_at
)
SELECT 
    user_id, date_bucket, total_events, total_sessions,
    overall_avg_duration, daily_value, all_countries, event_type_diversity,
    CURRENT_TIMESTAMP
FROM user_daily_summary
ON CONFLICT (user_id, summary_date) 
DO UPDATE SET
    total_events = EXCLUDED.total_events,
    total_sessions = EXCLUDED.total_sessions,
    avg_duration = EXCLUDED.avg_duration,
    daily_value = EXCLUDED.daily_value,
    countries_visited = EXCLUDED.countries_visited,
    event_diversity = EXCLUDED.event_diversity,
    computed_at = EXCLUDED.computed_at,
    updated_at = CURRENT_TIMESTAMP;

-- MySQL bulk processing limitations
-- MySQL has even more restrictive bulk operation capabilities
LOAD DATA INFILE '/tmp/user_events.csv'
INTO TABLE user_analytics
FIELDS TERMINATED BY ','
ENCLOSED BY '"'
LINES TERMINATED BY '\n'
IGNORE 1 ROWS
(user_id, event_type, @event_data, session_id, @device_info, @location_data)
SET 
    event_data = CAST(@event_data AS JSON),
    device_info = CAST(@device_info AS JSON),
    location_data = CAST(@location_data AS JSON),
    timestamp = CURRENT_TIMESTAMP;

-- Problems with traditional bulk processing approaches:
-- 1. Limited batch size handling leads to memory issues or poor performance
-- 2. Complex error handling with partial failure scenarios
-- 3. Manual transaction management and commit strategies
-- 4. Expensive index maintenance during large operations
-- 5. Limited parallelization and concurrency control
-- 6. Poor performance with large JSON/JSONB operations
-- 7. Complex rollback scenarios with partial batch failures
-- 8. Inefficient network utilization with individual operations
-- 9. Limited support for conditional updates and upserts
-- 10. Resource contention issues during high-volume processing

// MongoDB Bulk Operations - comprehensive high-throughput data processing system
const { MongoClient, ObjectId } = require('mongodb');
const EventEmitter = require('events');

const client = new MongoClient('mongodb://localhost:27017');
const db = client.db('analytics_platform');

// Advanced MongoDB bulk operations manager for enterprise applications
class MongoDBBulkOperationsManager extends EventEmitter {
  constructor(db) {
    super();
    this.db = db;
    this.collections = {
      userAnalytics: db.collection('user_analytics'),
      userSummaries: db.collection('user_daily_summaries'),
      deviceProfiles: db.collection('device_profiles'),
      locationData: db.collection('location_data'),
      sessionMetrics: db.collection('session_metrics'),
      errorLog: db.collection('bulk_operation_errors')
    };

    this.bulkOperationStats = new Map();
    this.processingQueue = [];
    this.maxBatchSize = 10000;
    this.concurrentOperations = 5;
    this.retryConfig = {
      maxRetries: 3,
      baseDelay: 1000,
      maxDelay: 10000
    };

    // Performance optimization settings
    this.bulkWriteOptions = {
      ordered: false, // Parallel processing for better performance
      writeConcern: { w: 1, j: true }, // Balance performance and durability
      bypassDocumentValidation: false
    };

    this.setupOperationHandlers();
  }

  async processBulkUserEvents(eventsData, options = {}) {
    console.log(`Processing bulk user events: ${eventsData.length} records`);

    const {
      batchSize = this.maxBatchSize,
      ordered = false,
      enableValidation = true,
      upsertMode = false
    } = options;

    const startTime = Date.now();
    const operationId = this.generateOperationId('bulk_user_events');

    try {
      // Initialize operation tracking
      this.initializeOperationStats(operationId, eventsData.length);

      // Process in optimized batches
      const batches = this.createOptimizedBatches(eventsData, batchSize);
      const results = [];

      // Execute batches with controlled concurrency
      for (let i = 0; i < batches.length; i += this.concurrentOperations) {
        const batchGroup = batches.slice(i, i + this.concurrentOperations);

        const batchPromises = batchGroup.map(batch => 
          this.executeBulkInsertBatch(batch, operationId, {
            ordered,
            enableValidation,
            upsertMode
          })
        );

        const batchResults = await Promise.allSettled(batchPromises);
        results.push(...batchResults);

        // Update progress
        const processedCount = Math.min((i + this.concurrentOperations) * batchSize, eventsData.length);
        this.updateOperationProgress(operationId, processedCount);

        console.log(`Processed ${processedCount}/${eventsData.length} events`);
      }

      // Aggregate results and handle errors
      const finalResult = this.aggregateBatchResults(results, operationId);
      const processingTime = Date.now() - startTime;

      console.log(`Bulk user events processing completed in ${processingTime}ms`);
      console.log(`Success: ${finalResult.totalInserted}, Errors: ${finalResult.totalErrors}`);

      // Store operation metrics
      await this.recordOperationMetrics(operationId, finalResult, processingTime);

      return finalResult;

    } catch (error) {
      console.error('Bulk user events processing failed:', error);
      await this.recordOperationError(operationId, error, eventsData.length);
      throw error;
    }
  }

  async executeBulkInsertBatch(batch, operationId, options) {
    const { ordered = false, enableValidation = true, upsertMode = false } = options;

    try {
      // Prepare bulk write operations
      const bulkOps = batch.map(event => {
        const document = {
          ...event,
          _id: event._id || new ObjectId(),
          processedAt: new Date(),
          batchId: operationId,

          // Enhanced metadata for analytics
          processingMetadata: {
            ingestionTime: new Date(),
            sourceSystem: event.sourceSystem || 'bulk_import',
            dataVersion: event.dataVersion || '1.0',
            validationPassed: enableValidation ? this.validateEventData(event) : true
          }
        };

        if (upsertMode) {
          return {
            updateOne: {
              filter: { 
                userId: event.userId, 
                sessionId: event.sessionId, 
                timestamp: event.timestamp 
              },
              update: { 
                $set: document,
                $setOnInsert: { createdAt: new Date() }
              },
              upsert: true
            }
          };
        } else {
          return {
            insertOne: {
              document: document
            }
          };
        }
      });

      // Execute bulk operation with optimized settings
      const bulkWriteResult = await this.collections.userAnalytics.bulkWrite(
        bulkOps, 
        {
          ...this.bulkWriteOptions,
          ordered: ordered
        }
      );

      return {
        success: true,
        result: bulkWriteResult,
        batchSize: batch.length,
        operationId: operationId
      };

    } catch (error) {
      console.error(`Batch operation failed for operation ${operationId}:`, error);

      // Log detailed error information
      await this.logBatchError(operationId, batch, error);

      return {
        success: false,
        error: error,
        batchSize: batch.length,
        operationId: operationId
      };
    }
  }

  async processBulkUserUpdates(updatesData, options = {}) {
    console.log(`Processing bulk user updates: ${updatesData.length} updates`);

    const {
      batchSize = this.maxBatchSize,
      arrayFilters = [],
      multi = false
    } = options;

    const operationId = this.generateOperationId('bulk_user_updates');
    const startTime = Date.now();

    try {
      this.initializeOperationStats(operationId, updatesData.length);

      // Group updates by operation type for optimization
      const updateGroups = this.groupUpdatesByType(updatesData);
      const results = [];

      for (const [updateType, updates] of Object.entries(updateGroups)) {
        console.log(`Processing ${updates.length} ${updateType} operations`);

        const batches = this.createOptimizedBatches(updates, batchSize);

        for (const batch of batches) {
          const bulkOps = batch.map(update => this.createUpdateOperation(update, {
            arrayFilters,
            multi,
            updateType
          }));

          try {
            const result = await this.collections.userAnalytics.bulkWrite(
              bulkOps,
              this.bulkWriteOptions
            );

            results.push({ success: true, result, updateType });

          } catch (error) {
            console.error(`Bulk update failed for type ${updateType}:`, error);
            results.push({ success: false, error, updateType, batchSize: batch.length });

            await this.logBatchError(operationId, batch, error);
          }
        }
      }

      const finalResult = this.aggregateUpdateResults(results, operationId);
      const processingTime = Date.now() - startTime;

      console.log(`Bulk updates completed in ${processingTime}ms`);
      await this.recordOperationMetrics(operationId, finalResult, processingTime);

      return finalResult;

    } catch (error) {
      console.error('Bulk updates processing failed:', error);
      throw error;
    }
  }

  createUpdateOperation(update, options) {
    const { arrayFilters = [], multi = false, updateType } = options;

    const baseOperation = {
      filter: update.filter || { _id: update._id },
      update: {},
      ...(arrayFilters.length > 0 && { arrayFilters }),
      ...(multi && { multi: true })
    };

    switch (updateType) {
      case 'field_updates':
        return {
          updateMany: {
            ...baseOperation,
            update: {
              $set: {
                ...update.setFields,
                lastUpdated: new Date()
              },
              ...(update.unsetFields && { $unset: update.unsetFields }),
              ...(update.incFields && { $inc: update.incFields })
            }
          }
        };

      case 'array_operations':
        return {
          updateMany: {
            ...baseOperation,
            update: {
              ...(update.pushToArrays && { $push: update.pushToArrays }),
              ...(update.pullFromArrays && { $pull: update.pullFromArrays }),
              ...(update.addToSets && { $addToSet: update.addToSets }),
              $set: { lastUpdated: new Date() }
            }
          }
        };

      case 'nested_updates':
        return {
          updateMany: {
            ...baseOperation,
            update: {
              $set: {
                ...Object.entries(update.nestedFields || {}).reduce((acc, [path, value]) => {
                  acc[path] = value;
                  return acc;
                }, {}),
                lastUpdated: new Date()
              }
            }
          }
        };

      case 'conditional_updates':
        return {
          updateMany: {
            ...baseOperation,
            update: [
              {
                $set: {
                  ...update.conditionalFields,
                  lastUpdated: new Date(),
                  // Add conditional logic using aggregation pipeline
                  ...(update.computedFields && Object.entries(update.computedFields).reduce((acc, [field, expr]) => {
                    acc[field] = expr;
                    return acc;
                  }, {}))
                }
              }
            ]
          }
        };

      default:
        return {
          updateOne: {
            ...baseOperation,
            update: { $set: { ...update.fields, lastUpdated: new Date() } }
          }
        };
    }
  }

  async processBulkAggregationUpdates(aggregationConfig) {
    console.log('Processing bulk aggregation updates...');

    const operationId = this.generateOperationId('bulk_aggregation');
    const startTime = Date.now();

    try {
      // Execute aggregation pipeline to compute updates
      const aggregationPipeline = [
        // Stage 1: Match and filter data
        {
          $match: {
            timestamp: { 
              $gte: aggregationConfig.dateRange?.start || new Date(Date.now() - 24*60*60*1000),
              $lte: aggregationConfig.dateRange?.end || new Date()
            },
            ...(aggregationConfig.additionalFilters || {})
          }
        },

        // Stage 2: Group and aggregate metrics
        {
          $group: {
            _id: aggregationConfig.groupBy || { 
              userId: '$userId', 
              date: { $dateToString: { format: '%Y-%m-%d', date: '$timestamp' } }
            },

            // Event metrics
            totalEvents: { $sum: 1 },
            uniqueSessions: { $addToSet: '$sessionId' },
            eventTypes: { $addToSet: '$eventType' },

            // Duration and value metrics
            totalDuration: { 
              $sum: { 
                $toDouble: { $ifNull: ['$eventData.duration', 0] } 
              } 
            },
            totalValue: { 
              $sum: { 
                $toDouble: { $ifNull: ['$eventData.value', 0] } 
              } 
            },

            // Device and location aggregations
            uniqueDevices: { $addToSet: '$deviceInfo.deviceId' },
            countries: { $addToSet: '$locationData.country' },

            // Time-based metrics
            firstEvent: { $min: '$timestamp' },
            lastEvent: { $max: '$timestamp' },

            // Advanced metrics
            bounceRate: {
              $avg: {
                $cond: [
                  { $eq: [{ $size: '$uniqueSessions' }, 1] },
                  1, 0
                ]
              }
            }
          }
        },

        // Stage 3: Compute derived metrics
        {
          $addFields: {
            sessionCount: { $size: '$uniqueSessions' },
            eventTypeCount: { $size: '$eventTypes' },
            deviceCount: { $size: '$uniqueDevices' },
            countryCount: { $size: '$countries' },

            // Calculate averages
            avgDuration: { 
              $divide: ['$totalDuration', '$totalEvents'] 
            },
            avgValue: { 
              $divide: ['$totalValue', '$totalEvents'] 
            },
            avgEventsPerSession: { 
              $divide: ['$totalEvents', { $size: '$uniqueSessions' }] 
            },

            // Session duration
            sessionDuration: { 
              $subtract: ['$lastEvent', '$firstEvent'] 
            },

            // User engagement score
            engagementScore: {
              $add: [
                { $multiply: ['$totalEvents', 1] },
                { $multiply: [{ $size: '$uniqueSessions' }, 2] },
                { $multiply: ['$totalValue', 0.1] }
              ]
            },

            computedAt: new Date()
          }
        }
      ];

      // Execute aggregation
      const aggregationResults = await this.collections.userAnalytics
        .aggregate(aggregationPipeline)
        .toArray();

      console.log(`Aggregation computed ${aggregationResults.length} summary records`);

      // Convert aggregation results to bulk upsert operations
      const upsertOps = aggregationResults.map(result => ({
        updateOne: {
          filter: {
            userId: result._id.userId,
            summaryDate: result._id.date
          },
          update: {
            $set: {
              ...result,
              _id: undefined, // Remove the grouped _id
              userId: result._id.userId,
              summaryDate: result._id.date,
              lastUpdated: new Date()
            }
          },
          upsert: true
        }
      }));

      // Execute bulk upsert operations
      const bulkResult = await this.collections.userSummaries.bulkWrite(
        upsertOps,
        this.bulkWriteOptions
      );

      const processingTime = Date.now() - startTime;

      console.log(`Bulk aggregation completed in ${processingTime}ms`);
      console.log(`Upserted: ${bulkResult.upsertedCount}, Modified: ${bulkResult.modifiedCount}`);

      await this.recordOperationMetrics(operationId, {
        aggregationResults: aggregationResults.length,
        upsertedCount: bulkResult.upsertedCount,
        modifiedCount: bulkResult.modifiedCount
      }, processingTime);

      return {
        success: true,
        aggregationResults: aggregationResults.length,
        bulkResult: bulkResult,
        processingTime: processingTime
      };

    } catch (error) {
      console.error('Bulk aggregation processing failed:', error);
      await this.recordOperationError(operationId, error, 0);
      throw error;
    }
  }

  async processBulkDeleteOperations(deleteConfig) {
    console.log('Processing bulk delete operations...');

    const operationId = this.generateOperationId('bulk_delete');
    const startTime = Date.now();

    try {
      const {
        conditions = [],
        archiveBeforeDelete = true,
        batchSize = 5000,
        dryRun = false
      } = deleteConfig;

      let totalDeleted = 0;
      let totalArchived = 0;

      for (const condition of conditions) {
        console.log(`Processing delete condition: ${JSON.stringify(condition.filter)}`);

        if (dryRun) {
          // Count documents that would be deleted
          const count = await this.collections.userAnalytics.countDocuments(condition.filter);
          console.log(`[DRY RUN] Would delete ${count} documents`);
          continue;
        }

        // Archive before delete if requested
        if (archiveBeforeDelete) {
          console.log('Archiving documents before deletion...');

          const documentsToArchive = await this.collections.userAnalytics
            .find(condition.filter)
            .limit(condition.limit || 100000)
            .toArray();

          if (documentsToArchive.length > 0) {
            // Add archive metadata
            const archiveDocuments = documentsToArchive.map(doc => ({
              ...doc,
              archivedAt: new Date(),
              archiveReason: condition.reason || 'bulk_delete_operation',
              originalCollection: 'user_analytics'
            }));

            await this.db.collection('archived_user_analytics')
              .insertMany(archiveDocuments);

            totalArchived += documentsToArchive.length;
            console.log(`Archived ${documentsToArchive.length} documents`);
          }
        }

        // Execute bulk delete
        const deleteResult = await this.collections.userAnalytics.deleteMany(
          condition.filter
        );

        totalDeleted += deleteResult.deletedCount;
        console.log(`Deleted ${deleteResult.deletedCount} documents`);

        // Add delay between operations to reduce system load
        if (condition.delayMs) {
          await new Promise(resolve => setTimeout(resolve, condition.delayMs));
        }
      }

      const processingTime = Date.now() - startTime;

      console.log(`Bulk delete completed in ${processingTime}ms`);
      console.log(`Total deleted: ${totalDeleted}, Total archived: ${totalArchived}`);

      await this.recordOperationMetrics(operationId, {
        totalDeleted,
        totalArchived,
        dryRun
      }, processingTime);

      return {
        success: true,
        totalDeleted,
        totalArchived,
        processingTime
      };

    } catch (error) {
      console.error('Bulk delete processing failed:', error);
      await this.recordOperationError(operationId, error, 0);
      throw error;
    }
  }

  // Utility methods
  createOptimizedBatches(data, batchSize) {
    const batches = [];

    // Sort data for better insertion performance (optional)
    const sortedData = data.sort((a, b) => {
      if (a.userId !== b.userId) return a.userId.localeCompare(b.userId);
      if (a.timestamp !== b.timestamp) return new Date(a.timestamp) - new Date(b.timestamp);
      return 0;
    });

    for (let i = 0; i < sortedData.length; i += batchSize) {
      batches.push(sortedData.slice(i, i + batchSize));
    }

    return batches;
  }

  groupUpdatesByType(updates) {
    const groups = {
      field_updates: [],
      array_operations: [],
      nested_updates: [],
      conditional_updates: []
    };

    for (const update of updates) {
      if (update.setFields || update.unsetFields || update.incFields) {
        groups.field_updates.push(update);
      } else if (update.pushToArrays || update.pullFromArrays || update.addToSets) {
        groups.array_operations.push(update);
      } else if (update.nestedFields) {
        groups.nested_updates.push(update);
      } else if (update.conditionalFields || update.computedFields) {
        groups.conditional_updates.push(update);
      } else {
        groups.field_updates.push(update); // Default category
      }
    }

    return groups;
  }

  validateEventData(event) {
    // Implement validation logic
    const required = ['userId', 'eventType', 'timestamp'];
    return required.every(field => event[field] != null);
  }

  generateOperationId(operationType) {
    return `${operationType}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
  }

  initializeOperationStats(operationId, totalRecords) {
    this.bulkOperationStats.set(operationId, {
      startTime: Date.now(),
      totalRecords: totalRecords,
      processedRecords: 0,
      errors: 0,
      status: 'running'
    });
  }

  updateOperationProgress(operationId, processedCount) {
    const stats = this.bulkOperationStats.get(operationId);
    if (stats) {
      stats.processedRecords = processedCount;
      stats.progress = (processedCount / stats.totalRecords) * 100;
    }
  }

  aggregateBatchResults(results, operationId) {
    let totalInserted = 0;
    let totalModified = 0;
    let totalUpserted = 0;
    let totalErrors = 0;

    for (const result of results) {
      if (result.status === 'fulfilled' && result.value.success) {
        const bulkResult = result.value.result;
        totalInserted += bulkResult.insertedCount || 0;
        totalModified += bulkResult.modifiedCount || 0;
        totalUpserted += bulkResult.upsertedCount || 0;
      } else {
        totalErrors += result.value?.batchSize || 1;
      }
    }

    return {
      operationId,
      totalInserted,
      totalModified,
      totalUpserted,
      totalErrors,
      successRate: ((totalInserted + totalModified + totalUpserted) / 
                   (totalInserted + totalModified + totalUpserted + totalErrors)) * 100
    };
  }

  aggregateUpdateResults(results, operationId) {
    let totalMatched = 0;
    let totalModified = 0;
    let totalUpserted = 0;
    let totalErrors = 0;

    for (const result of results) {
      if (result.success) {
        const bulkResult = result.result;
        totalMatched += bulkResult.matchedCount || 0;
        totalModified += bulkResult.modifiedCount || 0;
        totalUpserted += bulkResult.upsertedCount || 0;
      } else {
        totalErrors += result.batchSize || 1;
      }
    }

    return {
      operationId,
      totalMatched,
      totalModified,
      totalUpserted,
      totalErrors,
      successRate: ((totalMatched + totalUpserted) / 
                   (totalMatched + totalUpserted + totalErrors)) * 100
    };
  }

  async logBatchError(operationId, batch, error) {
    const errorDoc = {
      operationId: operationId,
      timestamp: new Date(),
      errorMessage: error.message,
      errorCode: error.code,
      batchSize: batch.length,
      sampleDocument: batch[0], // First document for debugging
      stackTrace: error.stack
    };

    try {
      await this.collections.errorLog.insertOne(errorDoc);
    } catch (logError) {
      console.error('Failed to log batch error:', logError);
    }
  }

  async recordOperationMetrics(operationId, result, processingTime) {
    const metricsDoc = {
      operationId: operationId,
      timestamp: new Date(),
      processingTime: processingTime,
      result: result,

      // Performance metrics
      throughput: result.totalInserted ? (result.totalInserted / processingTime) * 1000 : 0, // docs per second
      errorRate: result.totalErrors ? (result.totalErrors / (result.totalInserted + result.totalErrors)) : 0
    };

    try {
      await this.collections.operationMetrics.insertOne(metricsDoc);
    } catch (error) {
      console.error('Failed to record operation metrics:', error);
    }
  }

  async recordOperationError(operationId, error, recordCount) {
    const errorDoc = {
      operationId: operationId,
      timestamp: new Date(),
      errorMessage: error.message,
      errorCode: error.code,
      recordCount: recordCount,
      stackTrace: error.stack
    };

    try {
      await this.collections.errorLog.insertOne(errorDoc);
    } catch (logError) {
      console.error('Failed to record operation error:', logError);
    }
  }

  setupOperationHandlers() {
    // Handle operation completion events
    this.on('operation_complete', (result) => {
      console.log(`Operation ${result.operationId} completed with ${result.successRate}% success rate`);
    });

    this.on('operation_error', (error) => {
      console.error(`Operation failed:`, error);
    });
  }

  getOperationStats(operationId) {
    return this.bulkOperationStats.get(operationId);
  }

  getAllOperationStats() {
    return Object.fromEntries(this.bulkOperationStats);
  }
}

// Example usage: Complete enterprise bulk processing system
async function setupEnterpriseDataProcessing() {
  console.log('Setting up enterprise bulk data processing system...');

  const bulkManager = new MongoDBBulkOperationsManager(db);

  // Process large user event dataset
  const userEventsData = [
    {
      userId: 'user_001',
      eventType: 'page_view',
      timestamp: new Date(),
      sessionId: 'session_123',
      eventData: { page: '/dashboard', duration: 1500 },
      deviceInfo: { deviceId: 'device_456', type: 'desktop' },
      locationData: { country: 'US', city: 'San Francisco' }
    },
    // ... thousands more events
  ];

  // Execute bulk insert with performance optimization
  const insertResult = await bulkManager.processBulkUserEvents(userEventsData, {
    batchSize: 5000,
    ordered: false,
    enableValidation: true,
    upsertMode: false
  });

  console.log('Bulk insert result:', insertResult);

  // Process bulk updates for user enrichment
  const updateOperations = [
    {
      filter: { userId: 'user_001' },
      setFields: {
        'profile.lastActivity': new Date(),
        'profile.totalSessions': 25
      },
      incFields: {
        'metrics.totalEvents': 1,
        'metrics.totalValue': 10.50
      }
    }
    // ... more update operations
  ];

  const updateResult = await bulkManager.processBulkUserUpdates(updateOperations);
  console.log('Bulk update result:', updateResult);

  // Execute aggregation-based summary updates
  const aggregationResult = await bulkManager.processBulkAggregationUpdates({
    dateRange: {
      start: new Date(Date.now() - 24*60*60*1000), // Last 24 hours
      end: new Date()
    },
    groupBy: {
      userId: '$userId',
      date: { $dateToString: { format: '%Y-%m-%d', date: '$timestamp' } }
    }
  });

  console.log('Aggregation result:', aggregationResult);

  return bulkManager;
}

// Benefits of MongoDB Bulk Operations:
// - Exceptional performance with batch processing eliminating network round-trips
// - Flexible ordered and unordered execution modes for different consistency requirements  
// - Comprehensive error handling with detailed failure reporting and partial success tracking
// - Advanced filtering and transformation capabilities during bulk operations
// - Built-in support for upserts, array operations, and complex document updates
// - Optimized resource utilization with configurable batch sizes and concurrency control
// - Integrated monitoring and metrics collection for operation performance analysis
// - Native support for complex aggregation-based bulk updates and data transformations
// - Sophisticated retry mechanisms and dead letter queue patterns for reliability
// - SQL-compatible bulk operation patterns through QueryLeaf integration

module.exports = {
  MongoDBBulkOperationsManager,
  setupEnterpriseDataProcessing
};

// Production-grade bulk operations with advanced performance patterns
class EnterpriseBulkProcessor extends MongoDBBulkOperationsManager {
  constructor(db, enterpriseConfig) {
    super(db);

    this.enterpriseConfig = {
      distributedProcessing: enterpriseConfig.distributedProcessing || false,
      shardingAware: enterpriseConfig.shardingAware || false,
      compressionEnabled: enterpriseConfig.compressionEnabled || true,
      memoryOptimization: enterpriseConfig.memoryOptimization || true,
      performanceMonitoring: enterpriseConfig.performanceMonitoring || true
    };

    this.setupEnterpriseFeatures();
  }

  async processDistributedBulkOperations(operationConfig) {
    console.log('Processing distributed bulk operations across cluster...');

    const {
      collections,
      shardKey,
      parallelism = 10,
      replicationFactor = 1
    } = operationConfig;

    // Distribute operations based on shard key for optimal performance
    const shardDistribution = await this.analyzeShardDistribution(shardKey);

    const distributedTasks = collections.map(async (collectionConfig) => {
      const optimizedBatches = this.createShardAwareBatches(
        collectionConfig.data,
        shardKey,
        shardDistribution
      );

      return this.processShardOptimizedBatches(optimizedBatches, collectionConfig);
    });

    const results = await Promise.allSettled(distributedTasks);
    return this.aggregateDistributedResults(results);
  }

  async implementStreamingBulkOperations(streamConfig) {
    console.log('Setting up streaming bulk operations for continuous processing...');

    const {
      sourceStream,
      batchSize = 1000,
      flushInterval = 5000,
      backpressureThreshold = 10000
    } = streamConfig;

    let batchBuffer = [];
    let lastFlush = Date.now();

    return new Promise((resolve, reject) => {
      sourceStream.on('data', async (data) => {
        batchBuffer.push(data);

        // Check for batch completion or time-based flush
        const shouldFlush = batchBuffer.length >= batchSize || 
                           (Date.now() - lastFlush) >= flushInterval;

        if (shouldFlush) {
          try {
            await this.flushBatchBuffer(batchBuffer);
            batchBuffer = [];
            lastFlush = Date.now();
          } catch (error) {
            reject(error);
          }
        }

        // Implement backpressure control
        if (batchBuffer.length >= backpressureThreshold) {
          sourceStream.pause();
          await this.flushBatchBuffer(batchBuffer);
          batchBuffer = [];
          sourceStream.resume();
        }
      });

      sourceStream.on('end', async () => {
        if (batchBuffer.length > 0) {
          await this.flushBatchBuffer(batchBuffer);
        }
        resolve();
      });

      sourceStream.on('error', reject);
    });
  }

  async optimizeForTimeSeriesData(timeSeriesConfig) {
    console.log('Optimizing bulk operations for time series data...');

    const {
      timeField = 'timestamp',
      metaField = 'metadata',
      granularity = 'hours',
      compressionEnabled = true
    } = timeSeriesConfig;

    // Create time-based bucketing for optimal insertion performance
    const bucketedOperations = this.createTimeBuckets(
      timeSeriesConfig.data,
      timeField,
      granularity
    );

    const bulkOps = bucketedOperations.map(bucket => ({
      insertOne: {
        document: {
          _id: new ObjectId(),
          [timeField]: bucket.bucketTime,
          [metaField]: bucket.metadata,
          measurements: bucket.measurements,

          // Time series optimization metadata
          bucketInfo: {
            count: bucket.measurements.length,
            min: bucket.min,
            max: bucket.max,
            granularity: granularity
          }
        }
      }
    }));

    return await this.collections.timeSeries.bulkWrite(bulkOps, {
      ...this.bulkWriteOptions,
      ordered: true // Maintain time order for time series
    });
  }
}