wulf-pulse/lib/services/postgres-client.ts

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<T extends QueryResultRow = any>(
    text: string,
    params?: any[]
  ): Promise<QueryResult<T>> {
    const start = Date.now();
    try {
      const result = await this.getPool().query<T>(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<PoolClient> {
    return await this.getPool().connect();
  }

  /**
   * Execute a transaction
   */
  async transaction<T>(
    callback: (client: PoolClient) => Promise<T>
  ): Promise<T> {
    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<T extends QueryResultRow = any>(
    table: string,
    data: Record<string, any>
  ): Promise<T> {
    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<T>(query, values);
    return result.rows[0];
  }

  /**
   * Update a record by ID
   */
  async update<T extends QueryResultRow = any>(
    table: string,
    id: number | string,
    data: Record<string, any>
  ): Promise<T> {
    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<T>(query, [...values, id]);
    return result.rows[0];
  }

  /**
   * Upsert (insert or update) a record
   */
  async upsert<T extends QueryResultRow = any>(
    table: string,
    data: Record<string, any>,
    conflictColumns: string[] = ['id']
  ): Promise<T> {
    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<T>(query, values);
    return result.rows[0];
  }

  /**
   * Bulk insert records
   */
  async bulkInsert(
    table: string,
    records: Record<string, any>[]
  ): Promise<number> {
    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
   * For nullable FK columns (assigned_resource_id etc.), uses COALESCE so that
   * a null incoming value does not overwrite an existing non-null DB value.
   */
  async bulkUpsert(
    table: string,
    records: Record<string, any>[],
    conflictColumns: string[] = ['id'],
    preserveExistingOnNull: string[] = []
  ): Promise<number> {
    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 =>
        preserveExistingOnNull.includes(key)
          ? `${key} = COALESCE(EXCLUDED.${key}, ${table}.${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<void> {
    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<number> {
    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<T extends QueryResultRow = any>(
    table: string,
    where: Record<string, any> = {},
    options: {
      limit?: number;
      offset?: number;
      orderBy?: string;
      includeDeleted?: boolean;
    } = {}
  ): Promise<T[]> {
    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<T>(query, values);
    return result.rows;
  }

  /**
   * Find a single record by ID
   */
  async findById<T extends QueryResultRow = any>(
    table: string,
    id: number | string,
    includeDeleted = false
  ): Promise<T | null> {
    const deletedClause = includeDeleted ? '' : 'AND is_deleted = false';
    const query = `
      SELECT * FROM ${table}
      WHERE id = $1 ${deletedClause}
      LIMIT 1
    `;
    const result = await this.query<T>(query, [id]);
    return result.rows[0] || null;
  }

  /**
   * Count records
   */
  async count(
    table: string,
    where: Record<string, any> = {},
    includeDeleted = false
  ): Promise<number> {
    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<boolean> {
    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<void> {
    if (this.pool) {
      await this.pool.end();
    }
  }
}

// Export singleton instance
export const postgresClient = PostgresClient.getInstance();
export default postgresClient;