import { Pool, PoolClient, QueryResult, QueryResultRow } from 'pg'; /** * PostgreSQL Client Service * Manages database connection pool and provides query methods */ class PostgresClient { private static instance: PostgresClient | null = null; private pool: Pool | null = null; private constructor() { // Pool will be initialized lazily on first use } /** * Initialize the connection pool (lazy initialization) */ private initializePool(): void { if (this.pool) { return; // Already initialized } // Use the configured host (defaults to 'postgres' for Docker network) const host = process.env.POSTGRES_HOST || 'localhost'; this.pool = new Pool({ host, port: parseInt(process.env.POSTGRES_PORT || '5432'), database: process.env.POSTGRES_DB || 'pulse_autotask', user: process.env.POSTGRES_USER || 'pulse_user', password: process.env.POSTGRES_PASSWORD, max: 10, // Maximum number of clients in the pool idleTimeoutMillis: 30000, // Close idle clients after 30 seconds connectionTimeoutMillis: 2000, // Return an error after 2 seconds if connection could not be established }); // Handle pool errors this.pool.on('error', (err: Error) => { console.error('Unexpected error on idle PostgreSQL client', err); }); } /** * Get the pool, initializing if necessary */ private getPool(): Pool { this.initializePool(); return this.pool!; } /** * Get singleton instance of PostgresClient */ public static getInstance(): PostgresClient { if (!PostgresClient.instance) { PostgresClient.instance = new PostgresClient(); } return PostgresClient.instance; } /** * Execute a query with parameters */ async query( text: string, params?: any[] ): Promise> { const start = Date.now(); try { const result = await this.getPool().query(text, params); const duration = Date.now() - start; if (duration > 1000) { console.warn(`Slow query (${duration}ms):`, text.substring(0, 100)); } return result; } catch (error) { console.error('Database query error:', error); console.error('Query:', text); console.error('Params:', params); throw error; } } /** * Get a client from the pool for transactions */ async getClient(): Promise { return await this.getPool().connect(); } /** * Execute a transaction */ async transaction( callback: (client: PoolClient) => Promise ): Promise { const client = await this.getClient(); 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(); } } /** * Insert a single record */ async insert( table: string, data: Record ): Promise { const keys = Object.keys(data); const values = Object.values(data); const placeholders = keys.map((_, i) => `$${i + 1}`).join(', '); const query = ` INSERT INTO ${table} (${keys.join(', ')}) VALUES (${placeholders}) RETURNING * `; const result = await this.query(query, values); return result.rows[0]; } /** * Update a record by ID */ async update( table: string, id: number | string, data: Record ): Promise { const keys = Object.keys(data); const values = Object.values(data); const setClause = keys.map((key, i) => `${key} = $${i + 1}`).join(', '); const query = ` UPDATE ${table} SET ${setClause}, updated_at = CURRENT_TIMESTAMP WHERE id = $${keys.length + 1} RETURNING * `; const result = await this.query(query, [...values, id]); return result.rows[0]; } /** * Upsert (insert or update) a record */ async upsert( table: string, data: Record, conflictColumns: string[] = ['id'] ): Promise { const keys = Object.keys(data); const values = Object.values(data); const placeholders = keys.map((_, i) => `$${i + 1}`).join(', '); // Build UPDATE clause for conflict resolution const updateKeys = keys.filter(k => !conflictColumns.includes(k)); const updateClause = updateKeys .map(key => `${key} = EXCLUDED.${key}`) .join(', '); const query = ` INSERT INTO ${table} (${keys.join(', ')}) VALUES (${placeholders}) ON CONFLICT (${conflictColumns.join(', ')}) DO UPDATE SET ${updateClause}, updated_at = CURRENT_TIMESTAMP RETURNING * `; const result = await this.query(query, values); return result.rows[0]; } /** * Bulk insert records */ async bulkInsert( table: string, records: Record[] ): Promise { if (records.length === 0) return 0; const keys = Object.keys(records[0]); const placeholders: string[] = []; const values: any[] = []; records.forEach((record, recordIndex) => { const recordPlaceholders = keys.map( (_, keyIndex) => `$${recordIndex * keys.length + keyIndex + 1}` ); placeholders.push(`(${recordPlaceholders.join(', ')})`); values.push(...keys.map(key => record[key])); }); const query = ` INSERT INTO ${table} (${keys.join(', ')}) VALUES ${placeholders.join(', ')} ON CONFLICT (id) DO NOTHING `; const result = await this.query(query, values); return result.rowCount || 0; } /** * Bulk upsert records */ async bulkUpsert( table: string, records: Record[], conflictColumns: string[] = ['id'] ): Promise { if (records.length === 0) return 0; const keys = Object.keys(records[0]); const placeholders: string[] = []; const values: any[] = []; records.forEach((record, recordIndex) => { const recordPlaceholders = keys.map( (_, keyIndex) => `$${recordIndex * keys.length + keyIndex + 1}` ); placeholders.push(`(${recordPlaceholders.join(', ')})`); values.push(...keys.map(key => record[key])); }); // Build UPDATE clause for conflict resolution const updateKeys = keys.filter(k => !conflictColumns.includes(k)); const updateClause = updateKeys .map(key => `${key} = EXCLUDED.${key}`) .join(', '); const query = ` INSERT INTO ${table} (${keys.join(', ')}) VALUES ${placeholders.join(', ')} ON CONFLICT (${conflictColumns.join(', ')}) DO UPDATE SET ${updateClause}, updated_at = CURRENT_TIMESTAMP `; const result = await this.query(query, values); return result.rowCount || 0; } /** * Soft delete a record */ async softDelete( table: string, id: number | string ): Promise { const query = ` UPDATE ${table} SET is_deleted = true, deleted_at = CURRENT_TIMESTAMP WHERE id = $1 `; await this.query(query, [id]); } /** * Soft delete multiple records */ async softDeleteMany( table: string, ids: (number | string)[] ): Promise { if (ids.length === 0) return 0; const query = ` UPDATE ${table} SET is_deleted = true, deleted_at = CURRENT_TIMESTAMP WHERE id = ANY($1::bigint[]) `; const result = await this.query(query, [ids]); return result.rowCount || 0; } /** * Find records by criteria */ async find( table: string, where: Record = {}, options: { limit?: number; offset?: number; orderBy?: string; includeDeleted?: boolean; } = {} ): Promise { const conditions: string[] = []; const values: any[] = []; let paramIndex = 1; // Add where conditions Object.entries(where).forEach(([key, value]) => { conditions.push(`${key} = $${paramIndex}`); values.push(value); paramIndex++; }); // Exclude deleted records by default if (!options.includeDeleted) { conditions.push('is_deleted = false'); } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; const orderByClause = options.orderBy ? `ORDER BY ${options.orderBy}` : ''; const limitClause = options.limit ? `LIMIT ${options.limit}` : ''; const offsetClause = options.offset ? `OFFSET ${options.offset}` : ''; const query = ` SELECT * FROM ${table} ${whereClause} ${orderByClause} ${limitClause} ${offsetClause} `; const result = await this.query(query, values); return result.rows; } /** * Find a single record by ID */ async findById( table: string, id: number | string, includeDeleted = false ): Promise { const deletedClause = includeDeleted ? '' : 'AND is_deleted = false'; const query = ` SELECT * FROM ${table} WHERE id = $1 ${deletedClause} LIMIT 1 `; const result = await this.query(query, [id]); return result.rows[0] || null; } /** * Count records */ async count( table: string, where: Record = {}, includeDeleted = false ): Promise { const conditions: string[] = []; const values: any[] = []; let paramIndex = 1; Object.entries(where).forEach(([key, value]) => { conditions.push(`${key} = $${paramIndex}`); values.push(value); paramIndex++; }); if (!includeDeleted) { conditions.push('is_deleted = false'); } const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''; const query = `SELECT COUNT(*) as count FROM ${table} ${whereClause}`; const result = await this.query<{ count: string }>(query, values); return parseInt(result.rows[0].count); } /** * Test database connection */ async testConnection(): Promise { try { await this.query('SELECT 1'); return true; } catch (error) { console.error('Database connection test failed:', error); return false; } } /** * Close all connections in the pool */ async close(): Promise { if (this.pool) { await this.pool.end(); } } } // Export singleton instance export const postgresClient = PostgresClient.getInstance(); export default postgresClient;