← back to Watches

database/pool-manager.js

465 lines

/**
 * Advanced Connection Pool Manager
 *
 * Features:
 * - Primary/Replica routing (read/write splitting)
 * - Connection pooling with PgBouncer integration
 * - Health monitoring
 * - Automatic failover
 * - Query performance tracking
 * - Circuit breaker pattern
 */

import pg from 'pg';
import EventEmitter from 'events';

const { Pool } = pg;

// ============================================================================
// CIRCUIT BREAKER - Fault Tolerance Pattern
// ============================================================================

class CircuitBreaker extends EventEmitter {
    constructor(options = {}) {
        super();
        this.failureThreshold = options.failureThreshold || 5;
        this.successThreshold = options.successThreshold || 2;
        this.timeout = options.timeout || 60000; // 60s
        this.failures = 0;
        this.successes = 0;
        this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
        this.nextAttempt = Date.now();
    }

    async execute(fn) {
        if (this.state === 'OPEN') {
            if (Date.now() < this.nextAttempt) {
                throw new Error('Circuit breaker is OPEN');
            }
            this.state = 'HALF_OPEN';
        }

        try {
            const result = await fn();
            this.onSuccess();
            return result;
        } catch (error) {
            this.onFailure();
            throw error;
        }
    }

    onSuccess() {
        this.failures = 0;

        if (this.state === 'HALF_OPEN') {
            this.successes++;
            if (this.successes >= this.successThreshold) {
                this.state = 'CLOSED';
                this.successes = 0;
                this.emit('closed');
            }
        }
    }

    onFailure() {
        this.failures++;
        this.successes = 0;

        if (this.failures >= this.failureThreshold) {
            this.state = 'OPEN';
            this.nextAttempt = Date.now() + this.timeout;
            this.emit('open', { failures: this.failures });
        }
    }

    getState() {
        return {
            state: this.state,
            failures: this.failures,
            successes: this.successes,
            nextAttempt: this.nextAttempt
        };
    }
}

// ============================================================================
// CONNECTION POOL MANAGER
// ============================================================================

class PoolManager extends EventEmitter {
    constructor(config = {}) {
        super();

        this.config = {
            // Primary (master) configuration
            primary: {
                host: config.primary?.host || process.env.DB_HOST || 'localhost',
                port: config.primary?.port || process.env.DB_PORT || 5432,
                database: config.primary?.database || process.env.DB_NAME || 'omega_watches',
                user: config.primary?.user || process.env.DB_USER || 'postgres',
                password: config.primary?.password || process.env.DB_PASSWORD || 'postgres',
                max: config.primary?.max || 20,
                min: config.primary?.min || 5,
                idleTimeoutMillis: 30000,
                connectionTimeoutMillis: 3000,
                statement_timeout: 10000,
                query_timeout: 10000,
                application_name: 'omega_primary'
            },

            // Replica (read-only) configuration
            replica: config.replica ? {
                host: config.replica.host,
                port: config.replica.port || 5432,
                database: config.replica.database || config.primary?.database || 'omega_watches',
                user: config.replica.user || config.primary?.user || 'postgres',
                password: config.replica.password || config.primary?.password || 'postgres',
                max: config.replica.max || 30, // More connections for read replicas
                min: config.replica.min || 10,
                idleTimeoutMillis: 30000,
                connectionTimeoutMillis: 3000,
                statement_timeout: 15000,
                query_timeout: 15000,
                application_name: 'omega_replica'
            } : null,

            // Advanced options
            useReplica: config.useReplica !== false, // Enable replica routing
            preferReplica: config.preferReplica || false, // Prefer replica for reads
            replicaFailover: config.replicaFailover !== false, // Fallback to primary
            enableMetrics: config.enableMetrics !== false,
            slowQueryThreshold: config.slowQueryThreshold || 1000 // 1s
        };

        this.primaryPool = null;
        this.replicaPool = null;
        this.primaryBreaker = new CircuitBreaker();
        this.replicaBreaker = new CircuitBreaker();

        this.metrics = {
            queries: {
                total: 0,
                read: 0,
                write: 0,
                slow: 0,
                errors: 0
            },
            connections: {
                primary: { acquired: 0, released: 0, errors: 0 },
                replica: { acquired: 0, released: 0, errors: 0 }
            },
            performance: {
                totalTime: 0,
                avgTime: 0,
                maxTime: 0,
                minTime: Infinity
            }
        };

        this.initialize();
    }

    initialize() {
        // Create primary pool
        this.primaryPool = new Pool(this.config.primary);
        this.setupPoolEvents(this.primaryPool, 'primary');

        // Create replica pool if configured
        if (this.config.replica) {
            this.replicaPool = new Pool(this.config.replica);
            this.setupPoolEvents(this.replicaPool, 'replica');
        }

        // Circuit breaker event handlers
        this.primaryBreaker.on('open', (data) => {
            console.error('❌ Primary circuit breaker OPEN:', data);
            this.emit('primary:circuit:open', data);
        });

        this.primaryBreaker.on('closed', () => {
            console.log('✅ Primary circuit breaker CLOSED');
            this.emit('primary:circuit:closed');
        });

        if (this.replicaBreaker) {
            this.replicaBreaker.on('open', (data) => {
                console.warn('⚠️ Replica circuit breaker OPEN:', data);
                this.emit('replica:circuit:open', data);
            });

            this.replicaBreaker.on('closed', () => {
                console.log('✅ Replica circuit breaker CLOSED');
                this.emit('replica:circuit:closed');
            });
        }
    }

    setupPoolEvents(pool, name) {
        pool.on('connect', (client) => {
            this.metrics.connections[name].acquired++;
            this.emit(`${name}:connect`, client);
        });

        pool.on('acquire', (client) => {
            this.emit(`${name}:acquire`, client);
        });

        pool.on('remove', (client) => {
            this.metrics.connections[name].released++;
            this.emit(`${name}:remove`, client);
        });

        pool.on('error', (err, client) => {
            this.metrics.connections[name].errors++;
            console.error(`❌ Pool error (${name}):`, err.message);
            this.emit(`${name}:error`, err);
        });
    }

    /**
     * Determine if query is read-only
     */
    isReadQuery(sql) {
        const upperSQL = sql.trim().toUpperCase();
        const readPatterns = /^(SELECT|SHOW|DESCRIBE|EXPLAIN|WITH.*SELECT)/;
        const writePatterns = /^(INSERT|UPDATE|DELETE|CREATE|ALTER|DROP|TRUNCATE|GRANT|REVOKE)/;

        if (writePatterns.test(upperSQL)) return false;
        if (readPatterns.test(upperSQL)) return true;

        // Default to safe (primary) for unknown patterns
        return false;
    }

    /**
     * Select appropriate pool for query
     */
    selectPool(options = {}) {
        const forcePool = options.pool; // 'primary' or 'replica'
        const sql = options.sql || '';

        // Force specific pool
        if (forcePool === 'primary') {
            return { pool: this.primaryPool, breaker: this.primaryBreaker, name: 'primary' };
        }
        if (forcePool === 'replica' && this.replicaPool) {
            return { pool: this.replicaPool, breaker: this.replicaBreaker, name: 'replica' };
        }

        // No replica configured, use primary
        if (!this.replicaPool || !this.config.useReplica) {
            return { pool: this.primaryPool, breaker: this.primaryBreaker, name: 'primary' };
        }

        // Check if replica circuit is open
        if (this.replicaBreaker.state === 'OPEN' && !this.config.replicaFailover) {
            return { pool: this.primaryPool, breaker: this.primaryBreaker, name: 'primary' };
        }

        // Route based on query type
        const isRead = this.isReadQuery(sql);

        if (isRead && this.config.preferReplica) {
            return { pool: this.replicaPool, breaker: this.replicaBreaker, name: 'replica' };
        }

        if (isRead) {
            // Round-robin or load balance (simplified version uses replica)
            return { pool: this.replicaPool, breaker: this.replicaBreaker, name: 'replica' };
        }

        // Write queries always go to primary
        return { pool: this.primaryPool, breaker: this.primaryBreaker, name: 'primary' };
    }

    /**
     * Execute query with automatic routing and circuit breaking
     */
    async query(sql, params = [], options = {}) {
        const startTime = Date.now();
        const { pool, breaker, name } = this.selectPool({ sql, ...options });

        this.metrics.queries.total++;
        if (this.isReadQuery(sql)) {
            this.metrics.queries.read++;
        } else {
            this.metrics.queries.write++;
        }

        try {
            const result = await breaker.execute(async () => {
                return await pool.query(sql, params);
            });

            const duration = Date.now() - startTime;
            this.recordMetrics(duration);

            if (duration > this.config.slowQueryThreshold) {
                this.metrics.queries.slow++;
                console.warn(`🐌 Slow query (${duration}ms) on ${name}:`, sql.substring(0, 100));
                this.emit('slow:query', { sql, duration, pool: name });
            }

            this.emit('query:success', { sql, duration, pool: name });
            return result;

        } catch (error) {
            this.metrics.queries.errors++;
            const duration = Date.now() - startTime;

            console.error(`❌ Query error on ${name}:`, error.message);
            this.emit('query:error', { sql, error, duration, pool: name });

            // Attempt failover to primary if replica failed
            if (name === 'replica' && this.config.replicaFailover) {
                console.warn('⚠️ Failing over to primary...');
                return await this.primaryPool.query(sql, params);
            }

            throw error;
        }
    }

    /**
     * Execute transaction on primary
     */
    async transaction(callback) {
        const client = await this.primaryPool.connect();

        try {
            await client.query('BEGIN');
            const result = await callback(client);
            await client.query('COMMIT');
            return result;
        } catch (error) {
            await client.query('ROLLBACK');
            throw error;
        } finally {
            client.release();
        }
    }

    /**
     * Get connection from specific pool
     */
    async getConnection(poolName = 'primary') {
        const pool = poolName === 'replica' && this.replicaPool ?
            this.replicaPool : this.primaryPool;
        return await pool.connect();
    }

    /**
     * Health check for all pools
     */
    async healthCheck() {
        const health = {
            timestamp: new Date().toISOString(),
            primary: { healthy: false, latency: null, error: null },
            replica: { healthy: false, latency: null, error: null }
        };

        // Check primary
        try {
            const start = Date.now();
            await this.primaryPool.query('SELECT 1');
            health.primary.healthy = true;
            health.primary.latency = Date.now() - start;
        } catch (error) {
            health.primary.error = error.message;
        }

        // Check replica
        if (this.replicaPool) {
            try {
                const start = Date.now();
                await this.replicaPool.query('SELECT 1');
                health.replica.healthy = true;
                health.replica.latency = Date.now() - start;
            } catch (error) {
                health.replica.error = error.message;
            }
        }

        return health;
    }

    /**
     * Get pool statistics
     */
    getStats() {
        return {
            metrics: this.metrics,
            primary: {
                total: this.primaryPool.totalCount,
                idle: this.primaryPool.idleCount,
                waiting: this.primaryPool.waitingCount,
                circuitBreaker: this.primaryBreaker.getState()
            },
            replica: this.replicaPool ? {
                total: this.replicaPool.totalCount,
                idle: this.replicaPool.idleCount,
                waiting: this.replicaPool.waitingCount,
                circuitBreaker: this.replicaBreaker.getState()
            } : null
        };
    }

    recordMetrics(duration) {
        this.metrics.performance.totalTime += duration;
        this.metrics.performance.avgTime =
            this.metrics.performance.totalTime / this.metrics.queries.total;
        this.metrics.performance.maxTime =
            Math.max(this.metrics.performance.maxTime, duration);
        this.metrics.performance.minTime =
            Math.min(this.metrics.performance.minTime, duration);
    }

    /**
     * Graceful shutdown
     */
    async close() {
        console.log('📊 Closing database connection pools...');

        const promises = [];

        if (this.primaryPool) {
            promises.push(this.primaryPool.end());
        }

        if (this.replicaPool) {
            promises.push(this.replicaPool.end());
        }

        await Promise.all(promises);
        console.log('✅ All database pools closed');
    }
}

// ============================================================================
// SINGLETON INSTANCE
// ============================================================================

let poolManager = null;

export function getPoolManager(config) {
    if (!poolManager) {
        poolManager = new PoolManager(config);
    }
    return poolManager;
}

export async function closePoolManager() {
    if (poolManager) {
        await poolManager.close();
        poolManager = null;
    }
}

// Process cleanup
process.on('SIGTERM', closePoolManager);
process.on('SIGINT', closePoolManager);

export { PoolManager, CircuitBreaker };
export default PoolManager;