`database/connection.js`

136 lines
/**
 * Database Connection Pool
 * Implements connection pooling, health checks, and automatic reconnection
 */

import pg from 'pg';
import dotenv from 'dotenv';

dotenv.config();

const { Pool } = pg;

// Pool configuration with production-grade settings
const poolConfig = {
    host: process.env.DB_HOST || 'localhost',
    port: parseInt(process.env.DB_PORT || '5432'),
    database: process.env.DB_NAME || 'omega_watches',
    user: process.env.DB_USER || 'postgres',
    password: process.env.DB_PASSWORD || 'postgres',

    // Connection pool settings
    max: parseInt(process.env.DB_POOL_MAX || '20'), // Maximum connections
    min: parseInt(process.env.DB_POOL_MIN || '5'),  // Minimum connections
    idleTimeoutMillis: 30000,                        // Close idle connections after 30s
    connectionTimeoutMillis: 2000,                   // Fail fast if can't connect in 2s
    maxUses: 7500,                                   // Recycle connections after 7500 uses

    // Statement timeout (10 seconds for queries)
    statement_timeout: parseInt(process.env.DB_STATEMENT_TIMEOUT || '10000'),

    // Application name for monitoring
    application_name: 'omega_watch_api'
};

// Create connection pool
const pool = new Pool(poolConfig);

// Connection pool event handlers
pool.on('connect', (client) => {
    console.log('📊 New database client connected');
});

pool.on('acquire', (client) => {
    // Client acquired from pool
});

pool.on('remove', (client) => {
    console.log('📊 Database client removed from pool');
});

pool.on('error', (err, client) => {
    console.error('❌ Unexpected database pool error:', err);
});

// Health check function
export async function checkDatabaseHealth() {
    try {
        const client = await pool.connect();
        try {
            const result = await client.query('SELECT NOW(), current_database(), version()');
            return {
                healthy: true,
                timestamp: result.rows[0].now,
                database: result.rows[0].current_database,
                version: result.rows[0].version.split(' ')[1],
                poolStats: {
                    total: pool.totalCount,
                    idle: pool.idleCount,
                    waiting: pool.waitingCount
                }
            };
        } finally {
            client.release();
        }
    } catch (error) {
        return {
            healthy: false,
            error: error.message
        };
    }
}

// Execute query with automatic retry
export async function executeQuery(query, params, options = {}) {
    const { retries = 3, retryDelay = 1000 } = options;

    for (let attempt = 1; attempt <= retries; attempt++) {
        try {
            const result = await pool.query(query, params);
            return result;
        } catch (error) {
            if (attempt === retries) {
                throw error;
            }

            // Retry on connection errors
            if (error.code === 'ECONNREFUSED' || error.code === 'ETIMEDOUT') {
                console.log(`⚠️ Query failed (attempt ${attempt}/${retries}), retrying...`);
                await new Promise(resolve => setTimeout(resolve, retryDelay * attempt));
            } else {
                throw error;
            }
        }
    }
}

// Transaction helper
export async function withTransaction(callback) {
    const client = await pool.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();
    }
}

// Graceful shutdown
export async function closePool() {
    console.log('📊 Closing database connection pool...');
    await pool.end();
    console.log('✅ Database connection pool closed');
}

// Process termination handlers
process.on('SIGTERM', closePool);
process.on('SIGINT', closePool);

export default pool;