elasticsearch/ts/core/connection/connection-manager.ts

import { Client as ElasticClient } from '@elastic/elasticsearch';
import type { ElasticsearchConfig } from '../config/types.js';
import { HealthChecker, HealthStatus } from './health-check.js';
import type { HealthCheckResult } from './health-check.js';
import { CircuitBreaker } from './circuit-breaker.js';
import { Logger, defaultLogger } from '../observability/logger.js';
import { MetricsCollector, defaultMetricsCollector } from '../observability/metrics.js';
import { ConnectionError, ClusterUnavailableError } from '../errors/elasticsearch-error.js';

/**
 * Connection manager configuration
 */
export interface ConnectionManagerConfig extends ElasticsearchConfig {
  /** Enable health checks */
  enableHealthCheck?: boolean;

  /** Enable circuit breaker */
  enableCircuitBreaker?: boolean;

  /** Logger instance */
  logger?: Logger;

  /** Metrics collector */
  metricsCollector?: MetricsCollector;
}

/**
 * Connection manager for Elasticsearch client
 *
 * Provides:
 * - Singleton client instance
 * - Connection pooling
 * - Health monitoring
 * - Circuit breaker pattern
 * - Automatic reconnection
 *
 * @example
 * ```typescript
 * const manager = ElasticsearchConnectionManager.getInstance({
 *   nodes: ['http://localhost:9200'],
 *   auth: { type: 'basic', username: 'elastic', password: 'changeme' }
 * });
 *
 * const client = manager.getClient();
 * await client.search({ index: 'my-index', query: { match_all: {} } });
 * ```
 */
export class ElasticsearchConnectionManager {
  private static instance: ElasticsearchConnectionManager | null = null;

  private client: ElasticClient;
  private healthChecker: HealthChecker;
  private circuitBreaker: CircuitBreaker;
  private logger: Logger;
  private metrics: MetricsCollector;
  private config: ConnectionManagerConfig;
  private isInitialized = false;
  private connectionCount = 0;

  private constructor(config: ConnectionManagerConfig) {
    this.config = config;
    this.logger = config.logger || defaultLogger.child('connection-manager');
    this.metrics = config.metricsCollector || defaultMetricsCollector;

    // Initialize Elasticsearch client
    this.client = this.createClient(config);

    // Initialize health checker
    this.healthChecker = new HealthChecker(this.client, {
      interval: config.pool?.maxIdleTime || 30000,
      timeout: config.request?.timeout || 5000,
      checkClusterHealth: true,
    });

    // Initialize circuit breaker
    this.circuitBreaker = new CircuitBreaker('elasticsearch', {
      enabled: config.enableCircuitBreaker !== false,
      failureThreshold: 5,
      timeout: 60000,
    });

    this.logger.info('Elasticsearch connection manager created', {
      nodes: config.nodes,
      poolEnabled: !!config.pool,
      healthCheckEnabled: config.enableHealthCheck !== false,
      circuitBreakerEnabled: config.enableCircuitBreaker !== false,
    });
  }

  /**
   * Get singleton instance
   */
  static getInstance(config?: ConnectionManagerConfig): ElasticsearchConnectionManager {
    if (!ElasticsearchConnectionManager.instance) {
      if (!config) {
        throw new Error('Configuration required for first initialization');
      }
      ElasticsearchConnectionManager.instance = new ElasticsearchConnectionManager(config);
    }
    return ElasticsearchConnectionManager.instance;
  }

  /**
   * Reset singleton instance (useful for testing)
   */
  static resetInstance(): void {
    if (ElasticsearchConnectionManager.instance) {
      ElasticsearchConnectionManager.instance.destroy();
      ElasticsearchConnectionManager.instance = null;
    }
  }

  /**
   * Create Elasticsearch client with configuration
   */
  private createClient(config: ConnectionManagerConfig): ElasticClient {
    const nodes = Array.isArray(config.nodes) ? config.nodes : [config.nodes];

    const clientConfig: any = {
      nodes,
    };

    // Authentication
    if (config.auth) {
      switch (config.auth.type) {
        case 'basic':
          clientConfig.auth = {
            username: config.auth.username,
            password: config.auth.password,
          };
          break;
        case 'apiKey':
          clientConfig.auth = {
            apiKey: config.auth.apiKey,
          };
          break;
        case 'bearer':
          clientConfig.auth = {
            bearer: config.auth.token,
          };
          break;
        case 'cloud':
          clientConfig.cloud = {
            id: config.auth.id,
          };
          if (config.auth.apiKey) {
            clientConfig.auth = { apiKey: config.auth.apiKey };
          } else if (config.auth.username && config.auth.password) {
            clientConfig.auth = {
              username: config.auth.username,
              password: config.auth.password,
            };
          }
          break;
      }
    }

    // TLS configuration
    if (config.tls) {
      clientConfig.tls = config.tls;
    }

    // Request configuration
    if (config.request) {
      clientConfig.requestTimeout = config.request.timeout;
      clientConfig.maxRetries = config.request.maxRetries;
      clientConfig.compression = config.request.compression;
    }

    // Discovery/sniffing configuration
    if (config.discovery) {
      clientConfig.sniffOnStart = config.discovery.sniffOnStart;
      clientConfig.sniffInterval = config.discovery.interval;
      clientConfig.sniffOnConnectionFault = config.discovery.sniffOnConnectionFault;
    }

    // Proxy
    if (config.proxy) {
      clientConfig.proxy = config.proxy;
    }

    // Custom agent
    if (config.agent) {
      clientConfig.agent = config.agent;
    }

    return new ElasticClient(clientConfig);
  }

  /**
   * Initialize connection manager
   */
  async initialize(): Promise<void> {
    if (this.isInitialized) {
      return;
    }

    try {
      this.logger.info('Initializing connection manager...');

      // Test connection
      await this.client.ping();
      this.logger.info('Successfully connected to Elasticsearch');

      // Start health checks if enabled
      if (this.config.enableHealthCheck !== false) {
        this.healthChecker.startPeriodicChecks((result) => {
          this.onHealthChange(result);
        });
        this.logger.info('Health checks started');
      }

      this.isInitialized = true;
      this.metrics.activeConnections.set(1);

      this.logger.info('Connection manager initialized successfully');
    } catch (error) {
      this.logger.error('Failed to initialize connection manager', error as Error);
      throw new ConnectionError(
        'Failed to connect to Elasticsearch cluster',
        {
          operation: 'initialize',
        },
        error as Error
      );
    }
  }

  /**
   * Get Elasticsearch client
   */
  getClient(): ElasticClient {
    if (!this.isInitialized) {
      throw new Error('Connection manager not initialized. Call initialize() first.');
    }

    // Check circuit breaker
    if (this.circuitBreaker.isOpen()) {
      const stats = this.circuitBreaker.getStats();
      throw new ClusterUnavailableError(
        `Elasticsearch cluster unavailable. Circuit breaker open until ${stats.nextAttemptTime?.toISOString()}`,
        {
          circuitState: stats.state,
        }
      );
    }

    this.connectionCount++;
    return this.client;
  }

  /**
   * Execute operation through circuit breaker
   */
  async execute<T>(operation: () => Promise<T>): Promise<T> {
    return this.circuitBreaker.execute(operation);
  }

  /**
   * Health check callback
   */
  private onHealthChange(result: HealthCheckResult): void {
    this.logger.info('Cluster health changed', {
      status: result.status,
      clusterHealth: result.clusterHealth,
      activeNodes: result.activeNodes,
      responseTimeMs: result.responseTimeMs,
    });

    // Open circuit breaker if unhealthy
    if (result.status === HealthStatus.UNHEALTHY) {
      this.logger.warn('Cluster unhealthy, opening circuit breaker');
      this.circuitBreaker.open();
    } else if (result.status === HealthStatus.HEALTHY && this.circuitBreaker.isOpen()) {
      this.logger.info('Cluster recovered, closing circuit breaker');
      this.circuitBreaker.close();
    }

    // Update metrics
    this.metrics.activeConnections.set(result.available ? 1 : 0);
  }

  /**
   * Perform health check
   */
  async healthCheck(): Promise<HealthCheckResult> {
    return this.healthChecker.check();
  }

  /**
   * Get current health status
   */
  getHealthStatus(): HealthStatus {
    return this.healthChecker.getStatus();
  }

  /**
   * Check if cluster is healthy
   */
  isHealthy(): boolean {
    return this.healthChecker.isHealthy();
  }

  /**
   * Check if cluster is available
   */
  isAvailable(): boolean {
    return this.healthChecker.isAvailable();
  }

  /**
   * Get circuit breaker state
   */
  getCircuitState(): string {
    return this.circuitBreaker.getState();
  }

  /**
   * Get connection statistics
   */
  getStats(): {
    initialized: boolean;
    connectionCount: number;
    healthStatus: HealthStatus;
    circuitState: string;
    lastHealthCheck?: HealthCheckResult;
  } {
    return {
      initialized: this.isInitialized,
      connectionCount: this.connectionCount,
      healthStatus: this.healthChecker.getStatus(),
      circuitState: this.circuitBreaker.getState(),
      lastHealthCheck: this.healthChecker.getLastCheckResult(),
    };
  }

  /**
   * Cleanup and close connections
   */
  async destroy(): Promise<void> {
    this.logger.info('Destroying connection manager...');

    // Stop health checks
    this.healthChecker.destroy();

    // Close Elasticsearch client
    try {
      await this.client.close();
      this.logger.info('Elasticsearch client closed');
    } catch (error) {
      this.logger.error('Error closing Elasticsearch client', error as Error);
    }

    this.isInitialized = false;
    this.metrics.activeConnections.set(0);

    this.logger.info('Connection manager destroyed');
  }
}