MongoDB Bulk Write Operations for High-Performance Data Processing: Enterprise-Scale Data Ingestion and Batch Processing with SQL-Compatible Patterns

-- Traditional PostgreSQL batch processing - inefficient row-by-row operations

-- Product catalog import with individual INSERT statements
-- This approach creates massive performance problems at scale

DO $$
DECLARE
    product_record RECORD;
    import_cursor CURSOR FOR 
        SELECT * FROM product_import_staging;
    total_processed INTEGER := 0;
    batch_size INTEGER := 1000;
    start_time TIMESTAMP;
    current_batch_time TIMESTAMP;

BEGIN
    start_time := CURRENT_TIMESTAMP;

    -- Process each record individually - extremely inefficient
    FOR product_record IN import_cursor LOOP
        -- Individual validation and processing
        BEGIN
            -- Product existence check (N+1 query problem)
            IF EXISTS (
                SELECT 1 FROM products 
                WHERE sku = product_record.sku
            ) THEN
                -- Update existing product
                UPDATE products 
                SET 
                    name = product_record.name,
                    description = product_record.description,
                    price = product_record.price,
                    category_id = (
                        SELECT category_id 
                        FROM categories 
                        WHERE category_name = product_record.category_name
                    ),
                    stock_quantity = product_record.stock_quantity,
                    weight_kg = product_record.weight_kg,
                    dimensions_json = product_record.dimensions_json::JSONB,
                    supplier_id = (
                        SELECT supplier_id 
                        FROM suppliers 
                        WHERE supplier_code = product_record.supplier_code
                    ),

                    -- Pricing and inventory details
                    cost_price = product_record.cost_price,
                    margin_percent = product_record.margin_percent,
                    tax_category = product_record.tax_category,
                    minimum_order_quantity = product_record.minimum_order_quantity,
                    lead_time_days = product_record.lead_time_days,

                    -- Status and lifecycle
                    status = product_record.status,
                    is_active = product_record.is_active,
                    availability_date = product_record.availability_date::DATE,
                    discontinuation_date = product_record.discontinuation_date::DATE,

                    -- SEO and marketing
                    seo_title = product_record.seo_title,
                    seo_description = product_record.seo_description,
                    keywords_array = string_to_array(product_record.keywords, ','),

                    -- Audit fields
                    updated_at = CURRENT_TIMESTAMP,
                    updated_by = 'bulk_import_system'

                WHERE sku = product_record.sku;

            ELSE
                -- Insert new product with complex validation
                INSERT INTO products (
                    sku, name, description, price, category_id,
                    stock_quantity, weight_kg, dimensions_json,
                    supplier_id, cost_price, margin_percent,
                    tax_category, minimum_order_quantity, lead_time_days,
                    status, is_active, availability_date, discontinuation_date,
                    seo_title, seo_description, keywords_array,
                    created_at, updated_at, created_by
                ) VALUES (
                    product_record.sku,
                    product_record.name,
                    product_record.description,
                    product_record.price,
                    (SELECT category_id FROM categories WHERE category_name = product_record.category_name),
                    product_record.stock_quantity,
                    product_record.weight_kg,
                    product_record.dimensions_json::JSONB,
                    (SELECT supplier_id FROM suppliers WHERE supplier_code = product_record.supplier_code),
                    product_record.cost_price,
                    product_record.margin_percent,
                    product_record.tax_category,
                    product_record.minimum_order_quantity,
                    product_record.lead_time_days,
                    product_record.status,
                    product_record.is_active,
                    product_record.availability_date::DATE,
                    product_record.discontinuation_date::DATE,
                    product_record.seo_title,
                    product_record.seo_description,
                    string_to_array(product_record.keywords, ','),
                    CURRENT_TIMESTAMP,
                    CURRENT_TIMESTAMP,
                    'bulk_import_system'
                );

            END IF;

            -- Process product variants (additional N+1 queries)
            IF product_record.variants_json IS NOT NULL THEN
                INSERT INTO product_variants (
                    product_sku,
                    variant_sku,
                    variant_attributes,
                    price_adjustment,
                    stock_quantity,
                    created_at
                )
                SELECT 
                    product_record.sku,
                    variant->>'sku',
                    variant->'attributes',
                    (variant->>'price_adjustment')::DECIMAL,
                    (variant->>'stock_quantity')::INTEGER,
                    CURRENT_TIMESTAMP
                FROM jsonb_array_elements(product_record.variants_json::JSONB) AS variant
                ON CONFLICT (variant_sku) DO UPDATE SET
                    variant_attributes = EXCLUDED.variant_attributes,
                    price_adjustment = EXCLUDED.price_adjustment,
                    stock_quantity = EXCLUDED.stock_quantity,
                    updated_at = CURRENT_TIMESTAMP;
            END IF;

            -- Update inventory tracking
            INSERT INTO inventory_transactions (
                product_sku,
                transaction_type,
                quantity_change,
                new_quantity,
                reason,
                created_at
            ) VALUES (
                product_record.sku,
                'bulk_import',
                product_record.stock_quantity,
                product_record.stock_quantity,
                'Product catalog import',
                CURRENT_TIMESTAMP
            );

            total_processed := total_processed + 1;

            -- Periodic progress reporting (every 1000 records)
            IF total_processed % batch_size = 0 THEN
                current_batch_time := CURRENT_TIMESTAMP;
                RAISE NOTICE 'Processed % records. Current batch time: % seconds', 
                    total_processed, 
                    EXTRACT(EPOCH FROM (current_batch_time - start_time));

                -- Commit intermediate results (but lose atomicity)
                COMMIT;
            END IF;

        EXCEPTION WHEN OTHERS THEN
            -- Log error and continue (poor error handling)
            INSERT INTO import_errors (
                record_data,
                error_message,
                created_at
            ) VALUES (
                row_to_json(product_record)::TEXT,
                SQLERRM,
                CURRENT_TIMESTAMP
            );

            RAISE NOTICE 'Error processing record %: %', product_record.sku, SQLERRM;
            CONTINUE;
        END;

    END LOOP;

    RAISE NOTICE 'Import completed. Total processed: % records in % seconds', 
        total_processed, 
        EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - start_time));

END $$;

-- Problems with traditional row-by-row processing:
-- 1. Each operation requires a separate database round-trip
-- 2. N+1 query problems for lookups and validations
-- 3. No atomic batch operations - partial failures leave data inconsistent
-- 4. Poor performance - processing 100K records can take hours
-- 5. Resource intensive - high CPU and memory usage per operation
-- 6. Limited error handling - difficult to handle partial batch failures
-- 7. Complex transaction management across large datasets
-- 8. Manual progress tracking and monitoring implementation
-- 9. No built-in retry logic for transient failures
-- 10. Difficult to optimize - requires stored procedures or external tools

// MongoDB Bulk Write Operations - high-performance batch processing
const { MongoClient } = require('mongodb');

// Advanced MongoDB Bulk Operations Manager
class MongoBulkOperationsManager {
  constructor() {
    this.client = null;
    this.db = null;
    this.operationMetrics = new Map();
    this.errorHandlers = new Map();
    this.batchConfigurations = new Map();
  }

  async initialize() {
    console.log('Initializing MongoDB Bulk Operations Manager...');

    // Connect with optimized settings for bulk operations
    this.client = new MongoClient(process.env.MONGODB_URI || 'mongodb://localhost:27017', {
      // Connection pool optimized for bulk operations
      minPoolSize: 5,
      maxPoolSize: 20,
      maxIdleTimeMS: 30000,

      // Write concern optimized for throughput
      writeConcern: { w: 1, j: false }, // Faster for bulk imports
      readPreference: 'primary',

      // Batch operation optimizations
      maxBsonObjectSize: 16777216, // 16MB BSON limit
      compression: ['zlib'], // Reduce network overhead

      appName: 'BulkOperationsManager'
    });

    await this.client.connect();
    this.db = this.client.db('production');

    // Initialize batch configurations for different operation types
    await this.setupBatchConfigurations();

    console.log('✅ MongoDB Bulk Operations Manager initialized');
  }

  async setupBatchConfigurations() {
    console.log('Setting up batch operation configurations...');

    // Define optimized batch configurations for different scenarios
    const configurations = {
      // High-throughput product catalog imports
      'product_import': {
        batchSize: 1000,          // Optimal balance of memory and performance
        maxBatchSizeBytes: 10485760, // 10MB max batch size
        ordered: false,           // Allow parallel processing
        timeout: 300000,          // 5 minute timeout
        retryAttempts: 3,
        retryDelay: 1000,

        // Validation settings
        bypassDocumentValidation: false,
        validateDocuments: true,

        // Performance monitoring
        trackMetrics: true,
        logProgress: true,
        progressInterval: 5000
      },

      // Real-time order processing updates
      'order_updates': {
        batchSize: 500,
        maxBatchSizeBytes: 5242880, // 5MB max
        ordered: true,            // Maintain order for business logic
        timeout: 60000,           // 1 minute timeout
        retryAttempts: 5,
        retryDelay: 500,

        // Strict validation for financial data
        bypassDocumentValidation: false,
        validateDocuments: true,

        trackMetrics: true,
        logProgress: true,
        progressInterval: 1000
      },

      // Analytics data aggregation
      'analytics_batch': {
        batchSize: 2000,          // Larger batches for analytics
        maxBatchSizeBytes: 15728640, // 15MB max
        ordered: false,
        timeout: 600000,          // 10 minute timeout
        retryAttempts: 2,
        retryDelay: 2000,

        // Relaxed validation for analytical data
        bypassDocumentValidation: true,
        validateDocuments: false,

        trackMetrics: true,
        logProgress: true,
        progressInterval: 10000
      },

      // Log data ingestion (high volume, low latency)
      'log_ingestion': {
        batchSize: 5000,          // Very large batches
        maxBatchSizeBytes: 20971520, // 20MB max
        ordered: false,
        timeout: 120000,          // 2 minute timeout
        retryAttempts: 1,         // Minimal retries for logs
        retryDelay: 100,

        // Minimal validation for high throughput
        bypassDocumentValidation: true,
        validateDocuments: false,

        trackMetrics: false,      // Reduce overhead
        logProgress: false,
        progressInterval: 50000
      }
    };

    for (const [configName, config] of Object.entries(configurations)) {
      this.batchConfigurations.set(configName, config);
    }

    console.log('✅ Batch configurations initialized');
  }

  async performBulkProductImport(productData, options = {}) {
    console.log(`Starting bulk product import for ${productData.length} products...`);

    const config = this.batchConfigurations.get('product_import');
    const collection = this.db.collection('products');

    // Prepare bulk operations with sophisticated error handling
    const bulkOps = [];
    const processingMetrics = {
      startTime: Date.now(),
      totalRecords: productData.length,
      processedRecords: 0,
      successfulOperations: 0,
      failedOperations: 0,
      errors: [],
      batchTimes: []
    };

    try {
      // Prepare bulk write operations
      for (const product of productData) {
        const operation = await this.createProductOperation(product);
        if (operation) {
          bulkOps.push(operation);
        } else {
          processingMetrics.failedOperations++;
        }
      }

      // Execute bulk operations in optimized batches
      const results = await this.executeBulkOperations(
        collection, 
        bulkOps, 
        config, 
        processingMetrics
      );

      const totalTime = Date.now() - processingMetrics.startTime;

      console.log('✅ Bulk product import completed:', {
        totalRecords: processingMetrics.totalRecords,
        processedRecords: processingMetrics.processedRecords,
        successfulOperations: processingMetrics.successfulOperations,
        failedOperations: processingMetrics.failedOperations,
        totalTimeMs: totalTime,
        throughputPerSecond: Math.round((processingMetrics.successfulOperations / totalTime) * 1000),
        errorRate: ((processingMetrics.failedOperations / processingMetrics.totalRecords) * 100).toFixed(2) + '%'
      });

      return {
        success: true,
        results: results,
        metrics: processingMetrics,
        recommendations: this.generatePerformanceRecommendations(processingMetrics)
      };

    } catch (error) {
      console.error('Bulk product import failed:', error);
      return {
        success: false,
        error: error.message,
        metrics: processingMetrics,
        partialResults: processingMetrics.successfulOperations > 0
      };
    }
  }

  async createProductOperation(productData) {
    try {
      // Comprehensive data transformation and validation
      const transformedProduct = {
        // Core product information
        sku: productData.sku,
        name: productData.name,
        description: productData.description,

        // Pricing and financial data
        pricing: {
          basePrice: parseFloat(productData.price) || 0,
          costPrice: parseFloat(productData.cost_price) || 0,
          marginPercent: parseFloat(productData.margin_percent) || 0,
          currency: productData.currency || 'USD',
          taxCategory: productData.tax_category || 'standard'
        },

        // Inventory management
        inventory: {
          stockQuantity: parseInt(productData.stock_quantity) || 0,
          minimumOrderQuantity: parseInt(productData.minimum_order_quantity) || 1,
          leadTimeDays: parseInt(productData.lead_time_days) || 0,
          trackInventory: productData.track_inventory !== 'false'
        },

        // Product specifications
        specifications: {
          weight: {
            value: parseFloat(productData.weight_kg) || 0,
            unit: 'kg'
          },
          dimensions: productData.dimensions_json ? 
            JSON.parse(productData.dimensions_json) : null,
          attributes: productData.attributes_json ? 
            JSON.parse(productData.attributes_json) : {}
        },

        // Categorization and classification
        classification: {
          categoryId: productData.category_id,
          categoryPath: productData.category_path ? 
            productData.category_path.split('/') : [],
          tags: productData.tags ? 
            productData.tags.split(',').map(tag => tag.trim()) : [],
          brand: productData.brand || null
        },

        // Supplier and sourcing information
        supplier: {
          supplierId: productData.supplier_id,
          supplierCode: productData.supplier_code,
          supplierProductCode: productData.supplier_product_code,
          leadTimeFromSupplier: parseInt(productData.supplier_lead_time) || 0
        },

        // Product lifecycle and status
        lifecycle: {
          status: productData.status || 'active',
          isActive: productData.is_active !== 'false',
          availabilityDate: productData.availability_date ? 
            new Date(productData.availability_date) : new Date(),
          discontinuationDate: productData.discontinuation_date ? 
            new Date(productData.discontinuation_date) : null,
          seasonality: productData.seasonality || null
        },

        // SEO and marketing data
        marketing: {
          seoTitle: productData.seo_title || productData.name,
          seoDescription: productData.seo_description || productData.description,
          keywords: productData.keywords ? 
            productData.keywords.split(',').map(kw => kw.trim()) : [],
          promotionalText: productData.promotional_text || null,
          featuredProduct: productData.featured_product === 'true'
        },

        // Product variants and options
        variants: productData.variants_json ? 
          JSON.parse(productData.variants_json).map(variant => ({
            variantId: variant.variant_id || this.generateVariantId(),
            variantSku: variant.sku,
            attributes: variant.attributes || {},
            priceAdjustment: parseFloat(variant.price_adjustment) || 0,
            stockQuantity: parseInt(variant.stock_quantity) || 0,
            isActive: variant.is_active !== 'false'
          })) : [],

        // Media and assets
        media: {
          images: productData.images_json ? 
            JSON.parse(productData.images_json) : [],
          documents: productData.documents_json ? 
            JSON.parse(productData.documents_json) : [],
          videos: productData.videos_json ? 
            JSON.parse(productData.videos_json) : []
        },

        // Compliance and regulatory
        compliance: {
          regulatoryInfo: productData.regulatory_info_json ? 
            JSON.parse(productData.regulatory_info_json) : {},
          certifications: productData.certifications ? 
            productData.certifications.split(',').map(cert => cert.trim()) : [],
          restrictions: productData.restrictions_json ? 
            JSON.parse(productData.restrictions_json) : {}
        },

        // Audit and tracking
        audit: {
          createdAt: new Date(),
          updatedAt: new Date(),
          createdBy: 'bulk_import_system',
          importBatch: options.batchId || this.generateBatchId(),
          dataSource: options.dataSource || 'csv_import',
          version: 1
        }
      };

      // Determine operation type (insert vs update)
      const existingProduct = await this.db.collection('products')
        .findOne({ sku: transformedProduct.sku }, { projection: { _id: 1, audit: 1 } });

      if (existingProduct) {
        // Update existing product
        transformedProduct.audit.updatedAt = new Date();
        transformedProduct.audit.version = (existingProduct.audit?.version || 1) + 1;

        return {
          updateOne: {
            filter: { sku: transformedProduct.sku },
            update: { $set: transformedProduct },
            upsert: false
          }
        };
      } else {
        // Insert new product
        return {
          insertOne: {
            document: transformedProduct
          }
        };
      }

    } catch (error) {
      console.error(`Error creating operation for product ${productData.sku}:`, error);
      return null;
    }
  }

  async executeBulkOperations(collection, operations, config, metrics) {
    const results = [];
    const totalBatches = Math.ceil(operations.length / config.batchSize);

    console.log(`Executing ${operations.length} operations in ${totalBatches} batches...`);

    for (let i = 0; i < operations.length; i += config.batchSize) {
      const batchStart = Date.now();
      const batch = operations.slice(i, i + config.batchSize);
      const batchNumber = Math.floor(i / config.batchSize) + 1;

      try {
        // Execute bulk write with optimized options
        const result = await collection.bulkWrite(batch, {
          ordered: config.ordered,
          bypassDocumentValidation: config.bypassDocumentValidation,
          writeConcern: { w: 1, j: false }, // Optimized for throughput
        });

        // Track successful operations
        metrics.successfulOperations += result.insertedCount + result.modifiedCount + result.upsertedCount;
        metrics.processedRecords += batch.length;

        const batchTime = Date.now() - batchStart;
        metrics.batchTimes.push(batchTime);

        results.push({
          batchNumber: batchNumber,
          batchSize: batch.length,
          result: result,
          processingTimeMs: batchTime,
          throughputPerSecond: Math.round((batch.length / batchTime) * 1000)
        });

        // Progress logging
        if (config.logProgress && batchNumber % Math.ceil(totalBatches / 10) === 0) {
          const progressPercent = ((i + batch.length) / operations.length * 100).toFixed(1);
          console.log(`Batch ${batchNumber}/${totalBatches} completed (${progressPercent}%) - ${Math.round((batch.length / batchTime) * 1000)} ops/sec`);
        }

      } catch (error) {
        console.error(`Batch ${batchNumber} failed:`, error);

        metrics.failedOperations += batch.length;
        metrics.errors.push({
          batchNumber: batchNumber,
          error: error.message,
          batchSize: batch.length,
          timestamp: new Date()
        });

        // Implement retry logic for transient errors
        if (config.retryAttempts > 0 && this.isRetryableError(error)) {
          console.log(`Retrying batch ${batchNumber} in ${config.retryDelay}ms...`);
          await new Promise(resolve => setTimeout(resolve, config.retryDelay));

          try {
            const retryResult = await collection.bulkWrite(batch, {
              ordered: config.ordered,
              bypassDocumentValidation: config.bypassDocumentValidation,
              writeConcern: { w: 1, j: false }
            });

            metrics.successfulOperations += retryResult.insertedCount + retryResult.modifiedCount + retryResult.upsertedCount;
            metrics.processedRecords += batch.length;

            results.push({
              batchNumber: batchNumber,
              batchSize: batch.length,
              result: retryResult,
              processingTimeMs: Date.now() - batchStart,
              retryAttempt: true
            });

            console.log(`✅ Retry successful for batch ${batchNumber}`);

          } catch (retryError) {
            console.error(`Retry failed for batch ${batchNumber}:`, retryError);
            metrics.errors.push({
              batchNumber: batchNumber,
              error: `Retry failed: ${retryError.message}`,
              batchSize: batch.length,
              timestamp: new Date()
            });
          }
        }
      }
    }

    return results;
  }

  isRetryableError(error) {
    // Define retryable error conditions
    const retryableErrors = [
      'network timeout',
      'connection pool timeout',
      'temporary failure',
      'server selection timeout',
      'connection interrupted'
    ];

    return retryableErrors.some(retryableError => 
      error.message.toLowerCase().includes(retryableError)
    );
  }

  generateVariantId() {
    return 'var_' + Date.now() + '_' + Math.random().toString(36).substr(2, 9);
  }

  generateBatchId() {
    return 'batch_' + new Date().toISOString().replace(/[:.]/g, '') + '_' + Math.random().toString(36).substr(2, 6);
  }

  generatePerformanceRecommendations(metrics) {
    const recommendations = [];
    const totalTime = Date.now() - metrics.startTime;
    const overallThroughput = (metrics.successfulOperations / totalTime) * 1000;

    // Throughput analysis
    if (overallThroughput < 100) {
      recommendations.push({
        type: 'performance',
        priority: 'high',
        message: 'Low throughput detected. Consider increasing batch size or optimizing document structure.',
        currentValue: Math.round(overallThroughput),
        targetValue: 500
      });
    }

    // Error rate analysis
    const errorRate = (metrics.failedOperations / metrics.totalRecords) * 100;
    if (errorRate > 5) {
      recommendations.push({
        type: 'reliability',
        priority: 'high',
        message: 'High error rate. Review data quality and validation rules.',
        currentValue: errorRate.toFixed(2) + '%',
        targetValue: '< 1%'
      });
    }

    // Batch timing analysis
    if (metrics.batchTimes.length > 0) {
      const avgBatchTime = metrics.batchTimes.reduce((a, b) => a + b, 0) / metrics.batchTimes.length;
      const maxBatchTime = Math.max(...metrics.batchTimes);

      if (maxBatchTime > avgBatchTime * 3) {
        recommendations.push({
          type: 'optimization',
          priority: 'medium',
          message: 'Inconsistent batch processing times. Consider connection pool tuning.',
          currentValue: `${Math.round(maxBatchTime)}ms max, ${Math.round(avgBatchTime)}ms avg`
        });
      }
    }

    return recommendations.length > 0 ? recommendations : [
      { type: 'status', priority: 'info', message: 'Bulk operations performing optimally.' }
    ];
  }

  async performBulkOrderUpdates(orderUpdates) {
    console.log(`Processing ${orderUpdates.length} order updates...`);

    const config = this.batchConfigurations.get('order_updates');
    const collection = this.db.collection('orders');
    const bulkOps = [];

    // Create update operations for orders
    for (const update of orderUpdates) {
      const operation = {
        updateOne: {
          filter: { orderId: update.orderId },
          update: {
            $set: {
              status: update.status,
              updatedAt: new Date(),
              updatedBy: update.updatedBy || 'system',

              // Track status history
              $push: {
                statusHistory: {
                  status: update.status,
                  timestamp: new Date(),
                  updatedBy: update.updatedBy || 'system',
                  reason: update.reason || 'bulk_update'
                }
              }
            }
          },
          upsert: false
        }
      };

      // Add conditional updates based on status
      if (update.status === 'shipped') {
        operation.updateOne.update.$set.shippedAt = new Date();
        operation.updateOne.update.$set.trackingNumber = update.trackingNumber;
        operation.updateOne.update.$set.carrier = update.carrier;
      } else if (update.status === 'delivered') {
        operation.updateOne.update.$set.deliveredAt = new Date();
        operation.updateOne.update.$set.deliveryConfirmation = update.deliveryConfirmation;
      }

      bulkOps.push(operation);
    }

    return await this.executeBulkOperations(collection, bulkOps, config, {
      startTime: Date.now(),
      totalRecords: orderUpdates.length,
      processedRecords: 0,
      successfulOperations: 0,
      failedOperations: 0,
      errors: [],
      batchTimes: []
    });
  }

  async performAnalyticsBatchProcessing(analyticsData) {
    console.log(`Processing ${analyticsData.length} analytics records...`);

    const config = this.batchConfigurations.get('analytics_batch');
    const collection = this.db.collection('analytics_events');

    // Transform analytics data for bulk insert
    const bulkOps = analyticsData.map(event => ({
      insertOne: {
        document: {
          eventType: event.type,
          userId: event.userId,
          sessionId: event.sessionId,
          timestamp: new Date(event.timestamp),

          // Event properties
          properties: event.properties || {},

          // User context
          userAgent: event.userAgent,
          ipAddress: event.ipAddress,

          // Page/app context
          page: {
            url: event.pageUrl,
            title: event.pageTitle,
            referrer: event.referrer
          },

          // Device and browser info
          device: event.device || {},
          browser: event.browser || {},

          // Geographic data
          geo: event.geo || {},

          // Processing metadata
          processedAt: new Date(),
          batchId: this.generateBatchId()
        }
      }
    }));

    return await this.executeBulkOperations(collection, bulkOps, config, {
      startTime: Date.now(),
      totalRecords: analyticsData.length,
      processedRecords: 0,
      successfulOperations: 0,
      failedOperations: 0,
      errors: [],
      batchTimes: []
    });
  }

  async getBulkOperationStats() {
    const collections = ['products', 'orders', 'analytics_events'];
    const stats = {};

    for (const collectionName of collections) {
      const collection = this.db.collection(collectionName);

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

      // Get recent bulk operation metrics
      const recentBulkOps = await collection.aggregate([
        {
          $match: {
            'audit.importBatch': { $exists: true },
            'audit.createdAt': { 
              $gte: new Date(Date.now() - 24 * 60 * 60 * 1000) // Last 24 hours
            }
          }
        },
        {
          $group: {
            _id: '$audit.importBatch',
            count: { $sum: 1 },
            earliestRecord: { $min: '$audit.createdAt' },
            latestRecord: { $max: '$audit.createdAt' },
            dataSource: { $first: '$audit.dataSource' }
          }
        },
        { $sort: { latestRecord: -1 } },
        { $limit: 10 }
      ]).toArray();

      stats[collectionName] = {
        documentCount: collStats.count,
        storageSize: collStats.storageSize,
        indexSize: collStats.totalIndexSize,
        avgDocumentSize: collStats.avgObjSize,
        recentBulkOperations: recentBulkOps
      };
    }

    return {
      timestamp: new Date(),
      collections: stats,
      summary: this.generateStatsummary(stats)
    };
  }

  generateStatsummary(stats) {
    let totalDocuments = 0;
    let totalStorageSize = 0;
    let recentBulkOperations = 0;

    for (const [collectionName, collectionStats] of Object.entries(stats)) {
      totalDocuments += collectionStats.documentCount;
      totalStorageSize += collectionStats.storageSize;
      recentBulkOperations += collectionStats.recentBulkOperations.length;
    }

    return {
      totalDocuments,
      totalStorageSize,
      recentBulkOperations,
      averageDocumentSize: totalDocuments > 0 ? Math.round(totalStorageSize / totalDocuments) : 0,
      performanceIndicators: {
        storageEfficiency: totalStorageSize < (totalDocuments * 1000) ? 'excellent' : 'review',
        bulkOperationActivity: recentBulkOperations > 5 ? 'high' : 'normal'
      }
    };
  }

  async shutdown() {
    console.log('Shutting down MongoDB Bulk Operations Manager...');

    if (this.client) {
      await this.client.close();
      console.log('✅ MongoDB connection closed');
    }

    this.operationMetrics.clear();
    this.errorHandlers.clear();
    this.batchConfigurations.clear();
  }
}

// Export the bulk operations manager
module.exports = { MongoBulkOperationsManager };

// Benefits of MongoDB Bulk Operations:
// - Native batch processing eliminates individual operation overhead
// - Intelligent batching with configurable size and ordering options
// - Built-in error handling with detailed failure reporting and retry logic
// - High-performance throughput optimization for large-scale data processing
// - Atomic batch operations with comprehensive transaction support
// - Advanced monitoring and metrics tracking for performance analysis
// - Flexible operation types supporting mixed insert/update/delete operations
// - Production-ready error recovery and partial failure handling
// - Comprehensive performance recommendations and optimization guidance
// - SQL-compatible batch processing patterns through QueryLeaf integration

// Advanced bulk operations patterns for enterprise data processing
class EnterpriseBulkProcessor {
  constructor() {
    this.processingQueues = new Map();
    this.operationStrategies = new Map();
    this.performanceProfiler = new Map();
    this.errorRecoveryHandlers = new Map();
  }

  async initializeProcessingStrategies() {
    console.log('Initializing enterprise bulk processing strategies...');

    // Define processing strategies for different data types
    const strategies = {
      // High-frequency transactional updates
      'financial_transactions': {
        processingMode: 'ordered_atomic',
        batchSize: 100,
        maxConcurrency: 1,
        errorTolerance: 'zero',
        consistencyLevel: 'strong',

        validation: {
          strict: true,
          schemaValidation: true,
          businessRules: true,
          auditLogging: true
        },

        performance: {
          priorityLevel: 'critical',
          maxLatencyMs: 1000,
          throughputTarget: 500
        }
      },

      // Bulk inventory synchronization
      'inventory_sync': {
        processingMode: 'unordered_parallel',
        batchSize: 1000,
        maxConcurrency: 5,
        errorTolerance: 'partial',
        consistencyLevel: 'eventual',

        validation: {
          strict: false,
          schemaValidation: true,
          businessRules: false,
          auditLogging: false
        },

        performance: {
          priorityLevel: 'high',
          maxLatencyMs: 5000,
          throughputTarget: 2000
        }
      },

      // Analytics data ingestion
      'analytics_ingestion': {
        processingMode: 'unordered_parallel',
        batchSize: 5000,
        maxConcurrency: 10,
        errorTolerance: 'high',
        consistencyLevel: 'relaxed',

        validation: {
          strict: false,
          schemaValidation: false,
          businessRules: false,
          auditLogging: false
        },

        performance: {
          priorityLevel: 'normal',
          maxLatencyMs: 30000,
          throughputTarget: 10000
        }
      },

      // Customer data migration
      'customer_migration': {
        processingMode: 'ordered_atomic',
        batchSize: 200,
        maxConcurrency: 2,
        errorTolerance: 'low',
        consistencyLevel: 'strong',

        validation: {
          strict: true,
          schemaValidation: true,
          businessRules: true,
          auditLogging: true
        },

        performance: {
          priorityLevel: 'high',
          maxLatencyMs: 10000,
          throughputTarget: 100
        }
      }
    };

    for (const [strategyName, strategy] of Object.entries(strategies)) {
      this.operationStrategies.set(strategyName, strategy);
    }

    console.log('✅ Processing strategies initialized');
  }

  async processDataWithStrategy(strategyName, data, options = {}) {
    const strategy = this.operationStrategies.get(strategyName);
    if (!strategy) {
      throw new Error(`Unknown processing strategy: ${strategyName}`);
    }

    console.log(`Processing ${data.length} records with ${strategyName} strategy...`);

    const processingContext = {
      strategyName,
      strategy,
      startTime: Date.now(),
      totalRecords: data.length,
      processedRecords: 0,
      successfulRecords: 0,
      failedRecords: 0,
      errors: [],
      performanceMetrics: {
        batchTimes: [],
        throughputMeasurements: [],
        memoryUsage: []
      }
    };

    try {
      // Apply strategy-specific processing
      switch (strategy.processingMode) {
        case 'ordered_atomic':
          return await this.processOrderedAtomic(data, strategy, processingContext);
        case 'unordered_parallel':
          return await this.processUnorderedParallel(data, strategy, processingContext);
        default:
          throw new Error(`Unknown processing mode: ${strategy.processingMode}`);
      }

    } catch (error) {
      console.error(`Processing failed for strategy ${strategyName}:`, error);
      return {
        success: false,
        error: error.message,
        context: processingContext
      };
    }
  }

  async processOrderedAtomic(data, strategy, context) {
    console.log('Processing with ordered atomic mode...');

    const collection = this.db.collection(context.collectionName || 'bulk_operations');
    const results = [];

    // Process in sequential batches to maintain order
    for (let i = 0; i < data.length; i += strategy.batchSize) {
      const batchStart = Date.now();
      const batch = data.slice(i, i + strategy.batchSize);

      try {
        // Start transaction for atomic processing
        const session = this.client.startSession();

        await session.withTransaction(async () => {
          const bulkOps = batch.map(record => this.createBulkOperation(record, strategy));

          const result = await collection.bulkWrite(bulkOps, {
            ordered: true,
            session: session,
            writeConcern: { w: 'majority', j: true } // Strong consistency
          });

          context.successfulRecords += result.insertedCount + result.modifiedCount;
          context.processedRecords += batch.length;

          results.push({
            batchIndex: Math.floor(i / strategy.batchSize),
            result: result,
            processingTimeMs: Date.now() - batchStart
          });
        });

        await session.endSession();

        // Performance monitoring for ordered processing
        const batchTime = Date.now() - batchStart;
        context.performanceMetrics.batchTimes.push(batchTime);

        if (batchTime > strategy.performance.maxLatencyMs) {
          console.warn(`Batch latency ${batchTime}ms exceeds target ${strategy.performance.maxLatencyMs}ms`);
        }

      } catch (error) {
        console.error(`Ordered batch failed at index ${i}:`, error);
        context.failedRecords += batch.length;
        context.errors.push({
          batchIndex: Math.floor(i / strategy.batchSize),
          error: error.message,
          recordCount: batch.length
        });

        // For zero error tolerance, stop processing
        if (strategy.errorTolerance === 'zero') {
          throw new Error(`Processing stopped due to zero error tolerance: ${error.message}`);
        }
      }
    }

    return {
      success: true,
      results: results,
      context: context
    };
  }

  async processUnorderedParallel(data, strategy, context) {
    console.log(`Processing with unordered parallel mode (${strategy.maxConcurrency} concurrent batches)...`);

    const collection = this.db.collection(context.collectionName || 'bulk_operations');
    const results = [];
    const concurrentPromises = [];

    // Create batches for parallel processing
    const batches = [];
    for (let i = 0; i < data.length; i += strategy.batchSize) {
      batches.push({
        index: Math.floor(i / strategy.batchSize),
        data: data.slice(i, i + strategy.batchSize)
      });
    }

    // Process batches with controlled concurrency
    for (let i = 0; i < batches.length; i += strategy.maxConcurrency) {
      const concurrentBatches = batches.slice(i, i + strategy.maxConcurrency);

      const batchPromises = concurrentBatches.map(async (batch) => {
        const batchStart = Date.now();

        try {
          const bulkOps = batch.data.map(record => this.createBulkOperation(record, strategy));

          const result = await collection.bulkWrite(bulkOps, {
            ordered: false, // Allow parallel processing within batch
            writeConcern: { w: 1, j: false } // Optimized for throughput
          });

          context.successfulRecords += result.insertedCount + result.modifiedCount;
          context.processedRecords += batch.data.length;

          return {
            batchIndex: batch.index,
            result: result,
            processingTimeMs: Date.now() - batchStart,
            throughputPerSecond: Math.round((batch.data.length / (Date.now() - batchStart)) * 1000)
          };

        } catch (error) {
          context.failedRecords += batch.data.length;
          context.errors.push({
            batchIndex: batch.index,
            error: error.message,
            recordCount: batch.data.length
          });

          return {
            batchIndex: batch.index,
            error: error.message,
            recordCount: batch.data.length
          };
        }
      });

      // Wait for current batch of concurrent operations
      const batchResults = await Promise.all(batchPromises);
      results.push(...batchResults);

      // Track performance metrics
      const successfulBatches = batchResults.filter(r => !r.error);
      if (successfulBatches.length > 0) {
        const avgThroughput = successfulBatches.reduce((sum, r) => sum + r.throughputPerSecond, 0) / successfulBatches.length;
        context.performanceMetrics.throughputMeasurements.push(avgThroughput);
      }
    }

    return {
      success: true,
      results: results,
      context: context
    };
  }

  createBulkOperation(record, strategy) {
    // Apply validation based on strategy
    if (strategy.validation.strict) {
      this.validateRecord(record, strategy);
    }

    // Transform record based on operation type
    const transformedRecord = this.transformRecord(record, strategy);

    // Return appropriate bulk operation
    if (record.operationType === 'update') {
      return {
        updateOne: {
          filter: { _id: record._id },
          update: { $set: transformedRecord },
          upsert: record.upsert || false
        }
      };
    } else {
      return {
        insertOne: {
          document: transformedRecord
        }
      };
    }
  }

  validateRecord(record, strategy) {
    // Implement validation logic based on strategy
    if (strategy.validation.schemaValidation) {
      // Schema validation logic
    }

    if (strategy.validation.businessRules) {
      // Business rules validation
    }

    return true;
  }

  transformRecord(record, strategy) {
    // Apply transformations based on strategy
    const transformed = { ...record };

    // Add audit fields based on strategy requirements
    if (strategy.validation.auditLogging) {
      transformed.audit = {
        processedAt: new Date(),
        strategy: strategy,
        version: 1
      };
    }

    return transformed;
  }

  async getProcessingMetrics() {
    const metrics = {
      timestamp: new Date(),
      strategies: {}
    };

    for (const [strategyName, strategy] of this.operationStrategies) {
      const performanceProfile = this.performanceProfiler.get(strategyName);

      metrics.strategies[strategyName] = {
        configuration: strategy,
        performance: performanceProfile || {
          avgThroughput: 0,
          avgLatency: 0,
          errorRate: 0,
          lastUsed: null
        }
      };
    }

    return metrics;
  }
}

// Export the enterprise bulk processor
module.exports = { EnterpriseBulkProcessor };