MongoDB Connection Pooling and Concurrency Management: High-Performance Database Scaling and Enterprise Connection Optimization

-- Traditional PostgreSQL connection management - manual connection handling with poor scalability

-- Basic connection configuration (limited flexibility)
CREATE DATABASE production_app;

-- Connection pool configuration in application.properties (static configuration)
-- spring.datasource.url=jdbc:postgresql://localhost:5432/production_app
-- spring.datasource.username=app_user
-- spring.datasource.password=secure_password
-- spring.datasource.driver-class-name=org.postgresql.Driver

-- HikariCP connection pool settings (manual tuning required)
-- spring.datasource.hikari.maximum-pool-size=20
-- spring.datasource.hikari.minimum-idle=5
-- spring.datasource.hikari.connection-timeout=30000
-- spring.datasource.hikari.idle-timeout=600000
-- spring.datasource.hikari.max-lifetime=1800000
-- spring.datasource.hikari.leak-detection-threshold=60000

-- Application layer connection management with manual pooling
CREATE TABLE connection_metrics (
    metric_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    metric_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

    -- Connection pool metrics
    pool_name VARCHAR(100),
    active_connections INTEGER,
    idle_connections INTEGER,
    total_connections INTEGER,
    max_pool_size INTEGER,

    -- Performance metrics
    connection_acquisition_time_ms INTEGER,
    connection_usage_time_ms INTEGER,
    query_execution_count INTEGER,
    failed_connection_attempts INTEGER,

    -- Resource utilization
    memory_usage_bytes BIGINT,
    cpu_usage_percent DECIMAL(5,2),
    connection_wait_count INTEGER,
    connection_timeout_count INTEGER,

    -- Error tracking
    connection_leak_count INTEGER,
    pool_exhaustion_count INTEGER,
    database_errors INTEGER
);

-- Manual connection monitoring with limited visibility
CREATE OR REPLACE FUNCTION monitor_connection_pool()
RETURNS TABLE(
    pool_status VARCHAR,
    active_count INTEGER,
    idle_count INTEGER,
    wait_count INTEGER,
    usage_percent DECIMAL
) AS $$
BEGIN
    -- Basic connection pool monitoring (limited capabilities)
    RETURN QUERY
    WITH pool_stats AS (
        SELECT 
            'main_pool' as pool_name,
            -- Simulated pool metrics (not real-time)
            15 as current_active,
            5 as current_idle,
            20 as pool_max_size,
            2 as current_waiting
    )
    SELECT 
        'operational'::VARCHAR as pool_status,
        ps.current_active,
        ps.current_idle,
        ps.current_waiting,
        ROUND((ps.current_active::DECIMAL / ps.pool_max_size::DECIMAL) * 100, 2) as usage_percent
    FROM pool_stats ps;
END;
$$ LANGUAGE plpgsql;

-- Inadequate connection handling in stored procedures
CREATE OR REPLACE FUNCTION process_high_volume_transactions()
RETURNS VOID AS $$
DECLARE
    batch_size INTEGER := 1000;
    processed_count INTEGER := 0;
    error_count INTEGER := 0;
    connection_failures INTEGER := 0;
    start_time TIMESTAMP := CURRENT_TIMESTAMP;

    -- Limited connection context
    transaction_cursor CURSOR FOR 
        SELECT transaction_id, amount, user_id, transaction_type
        FROM pending_transactions
        WHERE status = 'pending'
        ORDER BY created_at
        LIMIT 10000;

    transaction_record RECORD;

BEGIN
    RAISE NOTICE 'Starting high-volume transaction processing...';

    -- Manual transaction processing with connection overhead
    FOR transaction_record IN transaction_cursor LOOP
        BEGIN
            -- Each operation creates connection overhead and latency
            INSERT INTO processed_transactions (
                original_transaction_id, 
                amount, 
                user_id, 
                transaction_type,
                processed_at,
                processing_batch
            ) VALUES (
                transaction_record.transaction_id,
                transaction_record.amount,
                transaction_record.user_id,
                transaction_record.transaction_type,
                CURRENT_TIMESTAMP,
                'batch_' || EXTRACT(EPOCH FROM start_time)
            );

            -- Update original transaction status
            UPDATE pending_transactions 
            SET status = 'processed',
                processed_at = CURRENT_TIMESTAMP,
                processed_by = CURRENT_USER
            WHERE transaction_id = transaction_record.transaction_id;

            processed_count := processed_count + 1;

            -- Frequent commits create connection pressure
            IF processed_count % batch_size = 0 THEN
                COMMIT;
                RAISE NOTICE 'Processed % transactions', processed_count;

                -- Manual connection health check (limited effectiveness)
                PERFORM pg_stat_get_activity(NULL);
            END IF;

        EXCEPTION
            WHEN connection_exception THEN
                connection_failures := connection_failures + 1;
                RAISE WARNING 'Connection failure for transaction %: %', 
                    transaction_record.transaction_id, SQLERRM;

            WHEN OTHERS THEN
                error_count := error_count + 1;
                RAISE WARNING 'Processing error for transaction %: %', 
                    transaction_record.transaction_id, SQLERRM;
        END;
    END LOOP;

    RAISE NOTICE 'Transaction processing completed: % processed, % errors, % connection failures in %',
        processed_count, error_count, connection_failures, 
        CURRENT_TIMESTAMP - start_time;

    -- Limited connection pool reporting
    INSERT INTO connection_metrics (
        pool_name, active_connections, total_connections,
        query_execution_count, failed_connection_attempts,
        connection_timeout_count
    ) VALUES (
        'manual_pool', 
        -- Estimated metrics (not accurate)
        GREATEST(processed_count / 100, 1),
        20,
        processed_count,
        connection_failures,
        connection_failures
    );
END;
$$ LANGUAGE plpgsql;

-- Complex manual connection management for concurrent operations
CREATE OR REPLACE FUNCTION concurrent_data_processing()
RETURNS TABLE(
    worker_id INTEGER,
    records_processed INTEGER,
    processing_time INTERVAL,
    connection_efficiency DECIMAL
) AS $$
DECLARE
    worker_count INTEGER := 5;
    records_per_worker INTEGER := 2000;
    worker_index INTEGER;
    processing_start TIMESTAMP;
    processing_end TIMESTAMP;

BEGIN
    processing_start := CURRENT_TIMESTAMP;

    -- Simulate concurrent workers (limited parallelization in PostgreSQL)
    FOR worker_index IN 1..worker_count LOOP
        BEGIN
            -- Each worker creates separate connection overhead
            PERFORM process_worker_batch(worker_index, records_per_worker);

            processing_end := CURRENT_TIMESTAMP;

            RETURN QUERY 
            SELECT 
                worker_index,
                records_per_worker,
                processing_end - processing_start,
                ROUND(
                    records_per_worker::DECIMAL / 
                    EXTRACT(EPOCH FROM processing_end - processing_start)::DECIMAL, 
                    2
                ) as efficiency;

        EXCEPTION
            WHEN connection_exception THEN
                RAISE WARNING 'Worker % failed due to connection issues', worker_index;

                RETURN QUERY 
                SELECT worker_index, 0, INTERVAL '0', 0.0::DECIMAL;

            WHEN OTHERS THEN
                RAISE WARNING 'Worker % failed: %', worker_index, SQLERRM;

                RETURN QUERY 
                SELECT worker_index, 0, INTERVAL '0', 0.0::DECIMAL;
        END;
    END LOOP;

    RETURN;
END;
$$ LANGUAGE plpgsql;

-- Helper function for worker batch processing
CREATE OR REPLACE FUNCTION process_worker_batch(
    p_worker_id INTEGER,
    p_batch_size INTEGER
) RETURNS VOID AS $$
DECLARE
    processed INTEGER := 0;
    batch_start TIMESTAMP := CURRENT_TIMESTAMP;
BEGIN
    -- Simulated batch processing with connection overhead
    WHILE processed < p_batch_size LOOP
        -- Each operation has connection acquisition overhead
        INSERT INTO worker_results (
            worker_id,
            batch_item,
            processed_at,
            processing_order
        ) VALUES (
            p_worker_id,
            processed + 1,
            CURRENT_TIMESTAMP,
            processed
        );

        processed := processed + 1;

        -- Frequent connection status checks
        IF processed % 100 = 0 THEN
            PERFORM pg_stat_get_activity(NULL);
        END IF;
    END LOOP;

    RAISE NOTICE 'Worker % completed % records in %',
        p_worker_id, processed, CURRENT_TIMESTAMP - batch_start;
END;
$$ LANGUAGE plpgsql;

-- Limited connection pool analysis and optimization
WITH connection_analysis AS (
    SELECT 
        pool_name,
        AVG(active_connections) as avg_active,
        MAX(active_connections) as peak_active,
        AVG(connection_acquisition_time_ms) as avg_acquisition_time,
        COUNT(*) FILTER (WHERE connection_timeout_count > 0) as timeout_incidents,
        COUNT(*) FILTER (WHERE pool_exhaustion_count > 0) as exhaustion_incidents,

        -- Basic utilization calculation
        AVG(active_connections::DECIMAL / total_connections::DECIMAL) as avg_utilization,

        -- Simple performance metrics
        AVG(query_execution_count) as avg_query_throughput,
        SUM(failed_connection_attempts) as total_failures

    FROM connection_metrics
    WHERE metric_timestamp >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
    GROUP BY pool_name
),

pool_health_assessment AS (
    SELECT 
        ca.*,

        -- Basic health scoring (limited insight)
        CASE 
            WHEN ca.avg_utilization > 0.9 THEN 'overloaded'
            WHEN ca.avg_utilization > 0.7 THEN 'high_usage'
            WHEN ca.avg_utilization > 0.5 THEN 'normal'
            ELSE 'underutilized'
        END as pool_health,

        -- Simple recommendations
        CASE 
            WHEN ca.timeout_incidents > 5 THEN 'increase_pool_size'
            WHEN ca.avg_acquisition_time > 5000 THEN 'optimize_connection_creation'
            WHEN ca.exhaustion_incidents > 0 THEN 'review_connection_limits'
            ELSE 'monitor_trends'
        END as recommendation,

        -- Limited optimization suggestions
        CASE 
            WHEN ca.avg_utilization < 0.3 THEN 'reduce_pool_size_for_efficiency'
            WHEN ca.total_failures > 100 THEN 'investigate_connection_failures'
            ELSE 'maintain_current_configuration'
        END as optimization_advice

    FROM connection_analysis ca
)

SELECT 
    pha.pool_name,
    pha.avg_active,
    pha.peak_active,
    ROUND(pha.avg_utilization * 100, 1) as utilization_percent,
    pha.avg_acquisition_time || 'ms' as avg_connection_time,
    pha.pool_health,
    pha.recommendation,
    pha.optimization_advice,

    -- Basic performance assessment
    CASE 
        WHEN pha.avg_query_throughput > 1000 THEN 'high_performance'
        WHEN pha.avg_query_throughput > 500 THEN 'moderate_performance'
        ELSE 'low_performance'
    END as performance_assessment

FROM pool_health_assessment pha
ORDER BY pha.avg_utilization DESC;

-- Problems with traditional connection management:
-- 1. Manual configuration and tuning required for different workloads
-- 2. Limited visibility into connection usage patterns and performance
-- 3. Poor handling of connection spikes and variable load scenarios
-- 4. Rigid pooling strategies that don't adapt to application patterns
-- 5. Complex error handling for connection failures and timeouts
-- 6. Inefficient resource utilization with static pool configurations
-- 7. Difficult monitoring and debugging of connection-related issues
-- 8. Poor integration with modern microservices and cloud-native architectures
-- 9. Limited scalability with concurrent operations and high-throughput scenarios
-- 10. Complex optimization requiring deep database and application expertise

// MongoDB Advanced Connection Pooling - enterprise-grade connection management and optimization
const { MongoClient, MongoServerError, MongoNetworkError } = require('mongodb');
const { EventEmitter } = require('events');

// Advanced MongoDB connection pool manager with intelligent optimization
class AdvancedConnectionPoolManager extends EventEmitter {
  constructor(config = {}) {
    super();

    this.config = {
      // Connection configuration
      uri: config.uri || 'mongodb://localhost:27017',
      database: config.database || 'production_app',

      // Connection pool configuration
      minPoolSize: config.minPoolSize || 5,
      maxPoolSize: config.maxPoolSize || 100,
      maxIdleTimeMS: config.maxIdleTimeMS || 30000,
      waitQueueTimeoutMS: config.waitQueueTimeoutMS || 5000,

      // Advanced pooling features
      enableConnectionPooling: config.enableConnectionPooling !== false,
      enableReadPreference: config.enableReadPreference !== false,
      enableWriteConcern: config.enableWriteConcern !== false,

      // Performance optimization
      maxConnecting: config.maxConnecting || 2,
      heartbeatFrequencyMS: config.heartbeatFrequencyMS || 10000,
      serverSelectionTimeoutMS: config.serverSelectionTimeoutMS || 30000,
      socketTimeoutMS: config.socketTimeoutMS || 0,

      // Connection management
      retryWrites: config.retryWrites !== false,
      retryReads: config.retryReads !== false,
      compressors: config.compressors || ['snappy', 'zlib'],

      // Monitoring and analytics
      enableConnectionPoolMonitoring: config.enableConnectionPoolMonitoring !== false,
      enablePerformanceAnalytics: config.enablePerformanceAnalytics !== false,
      enableAdaptivePooling: config.enableAdaptivePooling !== false,

      // Application-specific optimization
      applicationName: config.applicationName || 'enterprise-mongodb-app',
      loadBalanced: config.loadBalanced || false,
      directConnection: config.directConnection || false
    };

    // Connection pool state
    this.connectionState = {
      isInitialized: false,
      client: null,
      database: null,
      connectionStats: {
        totalConnections: 0,
        activeConnections: 0,
        availableConnections: 0,
        connectionRequests: 0,
        failedConnections: 0,
        pooledConnections: 0
      }
    };

    // Performance monitoring
    this.performanceMetrics = {
      connectionAcquisitionTimes: [],
      operationLatencies: [],
      throughputMeasurements: [],
      errorRates: [],
      resourceUtilization: []
    };

    // Connection pool event handlers
    this.poolEventHandlers = new Map();

    // Adaptive pooling algorithm
    this.adaptivePooling = {
      enabled: this.config.enableAdaptivePooling,
      learningPeriodMS: 300000, // 5 minutes
      adjustmentThreshold: 0.15,
      lastAdjustment: Date.now(),
      performanceBaseline: null
    };

    this.initializeConnectionPool();
  }

  async initializeConnectionPool() {
    console.log('Initializing advanced MongoDB connection pool...');

    try {
      // Create MongoDB client with optimized connection pool settings
      this.connectionState.client = new MongoClient(this.config.uri, {
        // Connection pool configuration
        minPoolSize: this.config.minPoolSize,
        maxPoolSize: this.config.maxPoolSize,
        maxIdleTimeMS: this.config.maxIdleTimeMS,
        waitQueueTimeoutMS: this.config.waitQueueTimeoutMS,
        maxConnecting: this.config.maxConnecting,

        // Server selection and timeouts
        serverSelectionTimeoutMS: this.config.serverSelectionTimeoutMS,
        heartbeatFrequencyMS: this.config.heartbeatFrequencyMS,
        socketTimeoutMS: this.config.socketTimeoutMS,
        connectTimeoutMS: 10000,

        // Connection optimization
        retryWrites: this.config.retryWrites,
        retryReads: this.config.retryReads,
        compressors: this.config.compressors,

        // Application configuration
        appName: this.config.applicationName,
        loadBalanced: this.config.loadBalanced,
        directConnection: this.config.directConnection,

        // Read and write preferences
        readPreference: 'secondaryPreferred',
        writeConcern: { w: 'majority', j: true },
        readConcern: { level: 'majority' },

        // Monitoring configuration
        monitorCommands: this.config.enableConnectionPoolMonitoring,
        loggerLevel: 'info'
      });

      // Setup connection pool event monitoring
      if (this.config.enableConnectionPoolMonitoring) {
        this.setupConnectionPoolMonitoring();
      }

      // Connect to MongoDB
      await this.connectionState.client.connect();
      this.connectionState.database = this.connectionState.client.db(this.config.database);
      this.connectionState.isInitialized = true;

      // Initialize performance monitoring
      if (this.config.enablePerformanceAnalytics) {
        await this.initializePerformanceMonitoring();
      }

      // Setup adaptive pooling if enabled
      if (this.config.enableAdaptivePooling) {
        this.setupAdaptivePooling();
      }

      console.log('MongoDB connection pool initialized successfully', {
        database: this.config.database,
        minPoolSize: this.config.minPoolSize,
        maxPoolSize: this.config.maxPoolSize,
        adaptivePooling: this.config.enableAdaptivePooling
      });

      this.emit('connectionPoolReady', this.getConnectionStats());

      return this.connectionState.database;

    } catch (error) {
      console.error('Failed to initialize connection pool:', error);
      this.emit('connectionPoolError', error);
      throw error;
    }
  }

  setupConnectionPoolMonitoring() {
    console.log('Setting up comprehensive connection pool monitoring...');

    // Connection pool opened
    this.connectionState.client.on('connectionPoolCreated', (event) => {
      console.log(`Connection pool created: ${event.address}`, {
        maxPoolSize: event.options?.maxPoolSize,
        minPoolSize: event.options?.minPoolSize
      });

      this.emit('poolCreated', event);
    });

    // Connection created
    this.connectionState.client.on('connectionCreated', (event) => {
      this.connectionState.connectionStats.totalConnections++;
      this.connectionState.connectionStats.availableConnections++;

      console.log(`Connection created: ${event.connectionId}`, {
        totalConnections: this.connectionState.connectionStats.totalConnections
      });

      this.emit('connectionCreated', event);
    });

    // Connection ready
    this.connectionState.client.on('connectionReady', (event) => {
      console.log(`Connection ready: ${event.connectionId}`);
      this.emit('connectionReady', event);
    });

    // Connection checked out
    this.connectionState.client.on('connectionCheckedOut', (event) => {
      this.connectionState.connectionStats.activeConnections++;
      this.connectionState.connectionStats.availableConnections--;

      const checkoutTime = Date.now();
      this.recordConnectionAcquisitionTime(checkoutTime);

      this.emit('connectionCheckedOut', event);
    });

    // Connection checked in
    this.connectionState.client.on('connectionCheckedIn', (event) => {
      this.connectionState.connectionStats.activeConnections--;
      this.connectionState.connectionStats.availableConnections++;

      this.emit('connectionCheckedIn', event);
    });

    // Connection pool closed
    this.connectionState.client.on('connectionPoolClosed', (event) => {
      console.log(`Connection pool closed: ${event.address}`);
      this.emit('connectionPoolClosed', event);
    });

    // Connection check out failed
    this.connectionState.client.on('connectionCheckOutFailed', (event) => {
      this.connectionState.connectionStats.failedConnections++;

      console.warn(`Connection checkout failed: ${event.reason}`, {
        failedConnections: this.connectionState.connectionStats.failedConnections
      });

      this.emit('connectionCheckoutFailed', event);

      // Trigger adaptive pooling adjustment if enabled
      if (this.config.enableAdaptivePooling) {
        this.evaluatePoolingAdjustment('checkout_failure');
      }
    });

    // Connection closed
    this.connectionState.client.on('connectionClosed', (event) => {
      this.connectionState.connectionStats.totalConnections--;

      console.log(`Connection closed: ${event.connectionId}`, {
        reason: event.reason,
        totalConnections: this.connectionState.connectionStats.totalConnections
      });

      this.emit('connectionClosed', event);
    });
  }

  async executeWithPoolManagement(operation, options = {}) {
    console.log('Executing operation with advanced pool management...');
    const startTime = Date.now();

    try {
      if (!this.connectionState.isInitialized) {
        throw new Error('Connection pool not initialized');
      }

      // Record connection request
      this.connectionState.connectionStats.connectionRequests++;

      // Check pool health before operation
      const poolHealth = await this.assessPoolHealth();
      if (poolHealth.status === 'critical') {
        console.warn('Pool in critical state, applying emergency measures...');
        await this.applyEmergencyPoolMeasures(poolHealth);
      }

      // Execute operation with connection management
      const result = await this.executeOperationWithRetry(operation, options);

      // Record successful operation
      const executionTime = Date.now() - startTime;
      this.recordOperationLatency(executionTime);

      // Update performance metrics
      if (this.config.enablePerformanceAnalytics) {
        this.updatePerformanceMetrics(executionTime, 'success');
      }

      return result;

    } catch (error) {
      const executionTime = Date.now() - startTime;

      console.error('Operation failed with connection pool:', error.message);

      // Record failed operation
      this.recordOperationLatency(executionTime, 'error');

      // Handle connection-specific errors
      if (this.isConnectionError(error)) {
        await this.handleConnectionError(error, options);
      }

      // Update error metrics
      if (this.config.enablePerformanceAnalytics) {
        this.updatePerformanceMetrics(executionTime, 'error');
      }

      throw error;
    }
  }

  async executeOperationWithRetry(operation, options) {
    const maxRetries = options.maxRetries || 3;
    const retryDelayMs = options.retryDelayMs || 1000;
    let lastError = null;

    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        // Execute the operation
        const result = await operation(this.connectionState.database);

        if (attempt > 1) {
          console.log(`Operation succeeded on retry attempt ${attempt}`);
        }

        return result;

      } catch (error) {
        lastError = error;

        // Check if error is retryable
        if (!this.isRetryableError(error) || attempt === maxRetries) {
          throw error;
        }

        console.warn(`Operation failed (attempt ${attempt}/${maxRetries}): ${error.message}`);

        // Wait before retry with exponential backoff
        const delay = retryDelayMs * Math.pow(2, attempt - 1);
        await this.sleep(delay);
      }
    }

    throw lastError;
  }

  async performBulkOperationsWithPoolOptimization(collectionName, operations, options = {}) {
    console.log(`Executing bulk operations with pool optimization: ${operations.length} operations...`);
    const startTime = Date.now();

    try {
      // Optimize pool for bulk operations
      await this.optimizePoolForBulkOperations(operations.length);

      const collection = this.connectionState.database.collection(collectionName);
      const batchSize = options.batchSize || 1000;
      const results = {
        totalOperations: operations.length,
        successfulOperations: 0,
        failedOperations: 0,
        batches: [],
        totalTime: 0,
        averageLatency: 0
      };

      // Process operations in optimized batches
      const batches = this.createOptimizedBatches(operations, batchSize);

      for (let batchIndex = 0; batchIndex < batches.length; batchIndex++) {
        const batch = batches[batchIndex];
        const batchStartTime = Date.now();

        try {
          const batchResult = await this.executeWithPoolManagement(async (db) => {
            return await collection.bulkWrite(batch, {
              ordered: options.ordered !== false,
              writeConcern: { w: 'majority', j: true }
            });
          });

          const batchTime = Date.now() - batchStartTime;
          results.successfulOperations += batchResult.insertedCount + batchResult.modifiedCount;
          results.batches.push({
            batchIndex,
            batchSize: batch.length,
            executionTime: batchTime,
            insertedCount: batchResult.insertedCount,
            modifiedCount: batchResult.modifiedCount,
            deletedCount: batchResult.deletedCount
          });

          console.log(`Batch ${batchIndex + 1}/${batches.length} completed: ${batch.length} operations in ${batchTime}ms`);

        } catch (batchError) {
          console.error(`Batch ${batchIndex + 1} failed:`, batchError.message);
          results.failedOperations += batch.length;

          if (!options.continueOnError) {
            throw batchError;
          }
        }
      }

      // Calculate final statistics
      results.totalTime = Date.now() - startTime;
      results.averageLatency = results.totalTime / results.batches.length;

      console.log(`Bulk operations completed: ${results.successfulOperations}/${results.totalOperations} successful in ${results.totalTime}ms`);

      return results;

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

  async handleConcurrentOperations(concurrentTasks, options = {}) {
    console.log(`Managing ${concurrentTasks.length} concurrent operations with pool optimization...`);
    const startTime = Date.now();

    try {
      // Optimize pool for concurrent operations
      await this.optimizePoolForConcurrency(concurrentTasks.length);

      const maxConcurrency = options.maxConcurrency || Math.min(concurrentTasks.length, this.config.maxPoolSize * 0.8);
      const results = [];
      const errors = [];

      // Execute tasks with controlled concurrency
      const taskPromises = [];
      const semaphore = { count: maxConcurrency };

      for (let i = 0; i < concurrentTasks.length; i++) {
        const task = concurrentTasks[i];
        const taskPromise = this.executeConcurrentTask(task, i, semaphore, options);
        taskPromises.push(taskPromise);
      }

      // Wait for all tasks to complete
      const taskResults = await Promise.allSettled(taskPromises);

      // Process results
      taskResults.forEach((result, index) => {
        if (result.status === 'fulfilled') {
          results.push({
            taskIndex: index,
            result: result.value,
            success: true
          });
        } else {
          errors.push({
            taskIndex: index,
            error: result.reason.message,
            success: false
          });
        }
      });

      const totalTime = Date.now() - startTime;

      console.log(`Concurrent operations completed: ${results.length} successful, ${errors.length} failed in ${totalTime}ms`);

      return {
        totalTasks: concurrentTasks.length,
        successfulTasks: results.length,
        failedTasks: errors.length,
        totalTime,
        results,
        errors,
        averageConcurrency: maxConcurrency
      };

    } catch (error) {
      console.error('Concurrent operations management failed:', error);
      throw error;
    }
  }

  async executeConcurrentTask(task, taskIndex, semaphore, options) {
    // Wait for semaphore (connection availability)
    await this.acquireSemaphore(semaphore);

    try {
      const taskStartTime = Date.now();

      const result = await this.executeWithPoolManagement(async (db) => {
        return await task(db, taskIndex);
      }, options);

      const taskTime = Date.now() - taskStartTime;

      return {
        taskIndex,
        executionTime: taskTime,
        result
      };

    } finally {
      this.releaseSemaphore(semaphore);
    }
  }

  async optimizePoolForBulkOperations(operationCount) {
    console.log(`Optimizing connection pool for ${operationCount} bulk operations...`);

    // Calculate optimal pool size for bulk operations
    const estimatedConnections = Math.min(
      Math.ceil(operationCount / 1000) + 2, // Base estimate plus buffer
      this.config.maxPoolSize
    );

    // Temporarily adjust pool if needed
    if (estimatedConnections > this.config.minPoolSize) {
      console.log(`Temporarily increasing pool size to ${estimatedConnections} for bulk operations`);
      // Note: In production, this would adjust pool configuration dynamically
    }
  }

  async optimizePoolForConcurrency(concurrentTaskCount) {
    console.log(`Optimizing connection pool for ${concurrentTaskCount} concurrent operations...`);

    // Ensure sufficient connections for concurrency
    const requiredConnections = Math.min(concurrentTaskCount + 2, this.config.maxPoolSize);

    if (requiredConnections > this.connectionState.connectionStats.totalConnections) {
      console.log(`Pool optimization: ensuring ${requiredConnections} connections are available`);
      // Note: MongoDB driver automatically manages this, but we can provide hints
    }
  }

  async assessPoolHealth() {
    const stats = this.getConnectionStats();
    const utilizationRatio = stats.activeConnections / this.config.maxPoolSize;
    const failureRate = stats.failedConnections / Math.max(stats.connectionRequests, 1);

    let status = 'healthy';
    const issues = [];

    if (utilizationRatio > 0.9) {
      status = 'critical';
      issues.push('high_utilization');
    } else if (utilizationRatio > 0.7) {
      status = 'warning';
      issues.push('moderate_utilization');
    }

    if (failureRate > 0.1) {
      status = status === 'healthy' ? 'warning' : 'critical';
      issues.push('high_failure_rate');
    }

    if (stats.availableConnections === 0) {
      status = 'critical';
      issues.push('no_available_connections');
    }

    return {
      status,
      utilizationRatio,
      failureRate,
      issues,
      recommendations: this.generateHealthRecommendations(issues)
    };
  }

  generateHealthRecommendations(issues) {
    const recommendations = [];

    if (issues.includes('high_utilization')) {
      recommendations.push('Consider increasing maxPoolSize');
    }

    if (issues.includes('high_failure_rate')) {
      recommendations.push('Check network connectivity and server health');
    }

    if (issues.includes('no_available_connections')) {
      recommendations.push('Investigate connection leaks and optimize operation duration');
    }

    return recommendations;
  }

  async applyEmergencyPoolMeasures(poolHealth) {
    console.log('Applying emergency pool measures:', poolHealth.issues);

    if (poolHealth.issues.includes('no_available_connections')) {
      console.log('Force closing idle connections to recover pool capacity...');
      // In production, this would implement connection cleanup
    }

    if (poolHealth.issues.includes('high_failure_rate')) {
      console.log('Implementing circuit breaker for connection failures...');
      // In production, this would implement circuit breaker pattern
    }
  }

  setupAdaptivePooling() {
    console.log('Setting up adaptive connection pooling algorithm...');

    setInterval(() => {
      this.evaluateAndAdjustPool();
    }, this.adaptivePooling.learningPeriodMS);
  }

  async evaluateAndAdjustPool() {
    if (!this.adaptivePooling.enabled) return;

    console.log('Evaluating pool performance for adaptive adjustment...');

    const currentMetrics = this.calculatePerformanceMetrics();

    if (this.adaptivePooling.performanceBaseline === null) {
      this.adaptivePooling.performanceBaseline = currentMetrics;
      return;
    }

    const performanceChange = this.comparePerformanceMetrics(
      currentMetrics,
      this.adaptivePooling.performanceBaseline
    );

    if (Math.abs(performanceChange) > this.adaptivePooling.adjustmentThreshold) {
      await this.adjustPoolConfiguration(performanceChange, currentMetrics);
      this.adaptivePooling.performanceBaseline = currentMetrics;
    }
  }

  async adjustPoolConfiguration(performanceChange, metrics) {
    console.log(`Adaptive pooling: adjusting configuration based on ${performanceChange > 0 ? 'improved' : 'degraded'} performance`);

    if (performanceChange < -this.adaptivePooling.adjustmentThreshold) {
      // Performance degraded, try to optimize
      if (metrics.utilizationRatio > 0.8) {
        console.log('Increasing pool size due to high utilization');
        // In production, would adjust pool size
      }
    } else if (performanceChange > this.adaptivePooling.adjustmentThreshold) {
      // Performance improved, maintain or optimize further
      console.log('Performance improved, maintaining current pool configuration');
    }
  }

  // Utility methods for connection pool management

  recordConnectionAcquisitionTime(checkoutTime) {
    const acquisitionTime = Date.now() - checkoutTime;
    this.performanceMetrics.connectionAcquisitionTimes.push(acquisitionTime);

    // Keep only recent measurements
    if (this.performanceMetrics.connectionAcquisitionTimes.length > 1000) {
      this.performanceMetrics.connectionAcquisitionTimes = 
        this.performanceMetrics.connectionAcquisitionTimes.slice(-500);
    }
  }

  recordOperationLatency(latency, status = 'success') {
    this.performanceMetrics.operationLatencies.push({
      latency,
      status,
      timestamp: Date.now()
    });

    // Keep only recent measurements
    if (this.performanceMetrics.operationLatencies.length > 1000) {
      this.performanceMetrics.operationLatencies = 
        this.performanceMetrics.operationLatencies.slice(-500);
    }
  }

  isConnectionError(error) {
    return error instanceof MongoNetworkError || 
           error instanceof MongoServerError ||
           error.message.includes('connection') ||
           error.message.includes('timeout');
  }

  isRetryableError(error) {
    if (error instanceof MongoNetworkError) return true;
    if (error.code === 11000) return false; // Duplicate key error
    if (error.message.includes('timeout')) return true;
    return false;
  }

  async handleConnectionError(error, options) {
    console.warn('Handling connection error:', error.message);

    if (error instanceof MongoNetworkError) {
      console.log('Network error detected, checking pool health...');
      const poolHealth = await this.assessPoolHealth();
      if (poolHealth.status === 'critical') {
        await this.applyEmergencyPoolMeasures(poolHealth);
      }
    }
  }

  createOptimizedBatches(operations, batchSize) {
    const batches = [];
    for (let i = 0; i < operations.length; i += batchSize) {
      batches.push(operations.slice(i, i + batchSize));
    }
    return batches;
  }

  async acquireSemaphore(semaphore) {
    while (semaphore.count <= 0) {
      await this.sleep(10);
    }
    semaphore.count--;
  }

  releaseSemaphore(semaphore) {
    semaphore.count++;
  }

  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  getConnectionStats() {
    return {
      ...this.connectionState.connectionStats,
      poolSize: this.config.maxPoolSize,
      utilizationRatio: this.connectionState.connectionStats.activeConnections / this.config.maxPoolSize,
      timestamp: new Date()
    };
  }

  calculatePerformanceMetrics() {
    const recent = this.performanceMetrics.operationLatencies.slice(-100);
    const avgLatency = recent.reduce((sum, op) => sum + op.latency, 0) / recent.length || 0;
    const successRate = recent.filter(op => op.status === 'success').length / recent.length || 0;
    const utilizationRatio = this.connectionState.connectionStats.activeConnections / this.config.maxPoolSize;

    return {
      avgLatency,
      successRate,
      utilizationRatio,
      throughput: recent.length / 5 // Operations per second estimate
    };
  }

  comparePerformanceMetrics(current, baseline) {
    const latencyChange = (baseline.avgLatency - current.avgLatency) / baseline.avgLatency;
    const successRateChange = current.successRate - baseline.successRate;
    const throughputChange = (current.throughput - baseline.throughput) / baseline.throughput;

    // Weighted performance score
    return (latencyChange * 0.4) + (successRateChange * 0.3) + (throughputChange * 0.3);
  }

  async getDetailedPoolAnalytics() {
    const stats = this.getConnectionStats();
    const metrics = this.calculatePerformanceMetrics();
    const poolHealth = await this.assessPoolHealth();

    return {
      connectionStats: stats,
      performanceMetrics: metrics,
      poolHealth: poolHealth,
      configuration: {
        minPoolSize: this.config.minPoolSize,
        maxPoolSize: this.config.maxPoolSize,
        maxIdleTimeMS: this.config.maxIdleTimeMS,
        adaptivePoolingEnabled: this.config.enableAdaptivePooling
      },
      recommendations: poolHealth.recommendations
    };
  }

  async closeConnectionPool() {
    console.log('Closing MongoDB connection pool...');

    if (this.connectionState.client) {
      await this.connectionState.client.close();
      this.connectionState.isInitialized = false;
      console.log('Connection pool closed successfully');
    }
  }
}

// Example usage for enterprise-scale applications
async function demonstrateAdvancedConnectionPooling() {
  const poolManager = new AdvancedConnectionPoolManager({
    uri: 'mongodb://localhost:27017',
    database: 'production_analytics',
    minPoolSize: 10,
    maxPoolSize: 50,
    enableAdaptivePooling: true,
    enablePerformanceAnalytics: true,
    applicationName: 'enterprise-data-processor'
  });

  try {
    // Wait for pool initialization
    await poolManager.initializeConnectionPool();

    // Demonstrate bulk operations with pool optimization
    const bulkOperations = Array.from({ length: 5000 }, (_, index) => ({
      insertOne: {
        document: {
          userId: `user_${index}`,
          eventType: 'page_view',
          timestamp: new Date(),
          sessionId: `session_${Math.floor(index / 100)}`,
          data: {
            page: `/page_${index % 50}`,
            duration: Math.floor(Math.random() * 300),
            source: 'web'
          }
        }
      }
    }));

    console.log('Executing bulk operations with pool optimization...');
    const bulkResults = await poolManager.performBulkOperationsWithPoolOptimization(
      'user_events',
      bulkOperations,
      {
        batchSize: 1000,
        continueOnError: true
      }
    );

    // Demonstrate concurrent operations
    const concurrentTasks = Array.from({ length: 20 }, (_, index) => 
      async (db, taskIndex) => {
        const collection = db.collection('analytics_data');

        // Simulate complex aggregation
        const result = await collection.aggregate([
          { $match: { userId: { $regex: `user_${taskIndex}` } } },
          { $group: {
            _id: '$eventType',
            count: { $sum: 1 },
            avgDuration: { $avg: '$data.duration' }
          }},
          { $sort: { count: -1 } }
        ]).toArray();

        return { taskIndex, resultCount: result.length };
      }
    );

    console.log('Executing concurrent operations with pool management...');
    const concurrentResults = await poolManager.handleConcurrentOperations(concurrentTasks, {
      maxConcurrency: 15
    });

    // Get detailed analytics
    const poolAnalytics = await poolManager.getDetailedPoolAnalytics();
    console.log('Connection Pool Analytics:', JSON.stringify(poolAnalytics, null, 2));

    return {
      bulkResults,
      concurrentResults,
      poolAnalytics
    };

  } catch (error) {
    console.error('Advanced connection pooling demonstration failed:', error);
    throw error;
  } finally {
    await poolManager.closeConnectionPool();
  }
}

// Benefits of MongoDB Advanced Connection Pooling:
// - Intelligent connection management with automatic optimization and resource management
// - Comprehensive monitoring with real-time pool health assessment and performance analytics
// - Adaptive pooling algorithms that adjust to application patterns and workload changes
// - Advanced error handling with retry mechanisms and circuit breaker patterns
// - Support for concurrent operations with intelligent connection allocation and management
// - Production-ready scalability with distributed connection management and optimization
// - Comprehensive analytics and monitoring for operational insight and troubleshooting
// - Seamless integration with MongoDB's native connection pooling and cluster management

module.exports = {
  AdvancedConnectionPoolManager,
  demonstrateAdvancedConnectionPooling
};