Connection Management

/**
 * Create database instance with connection configuration
 * @param cn - Connection string or configuration object
 * @param dc - Database context (optional user-defined data)
 * @returns Database instance with connection pool
 */
pgp(cn: string | IConnectionParameters, dc?: any): IDatabase

interface IConnectionParameters {
  // Connection details
  host?: string // Database host
  port?: number // Database port
  database?: string // Database name
  user?: string // Username
  password?: string // Password
  
  // SSL configuration
  ssl?: boolean | object // SSL settings
  
  // Pool configuration
  max?: number // Maximum pool connections
  min?: number // Minimum pool connections
  idleTimeoutMillis?: number // Idle connection timeout
  connectionTimeoutMillis?: number // Connection timeout
  maxUses?: number // Maximum connection uses before replacement
  
  // Other pg options
  application_name?: string // Application name
  keepAlive?: boolean // TCP keep-alive
  statement_timeout?: number // Statement timeout
  query_timeout?: number // Query timeout
  lock_timeout?: number // Lock timeout
  idle_in_transaction_session_timeout?: number // Idle in transaction timeout
}

// Connection string
const db = pgp('postgres://username:password@host:port/database');

// Connection object
const db2 = pgp({
  host: 'localhost',
  port: 5432,
  database: 'myapp',
  user: 'postgres',
  password: 'secret',
  max: 20, // Maximum connections in pool
  min: 5,  // Minimum connections in pool
  idleTimeoutMillis: 30000,
  ssl: false
});

// With database context
const db3 = pgp(connectionString, {
  userId: 123,
  sessionId: 'abc123',
  environment: 'production'
});

// SSL configuration
const dbSSL = pgp({
  host: 'secure-db.example.com',
  database: 'myapp',
  user: 'app_user',
  password: process.env.DB_PASSWORD,
  ssl: {
    rejectUnauthorized: false,
    ca: fs.readFileSync('server-ca.pem'),
    key: fs.readFileSync('client-key.pem'),
    cert: fs.readFileSync('client-cert.pem')
  }
});

/**
 * Database instance with automatic connection pooling
 */
interface IDatabase {
  // Pool access (read-only)
  readonly $pool: IPool // Connection pool instance
  readonly $cn: string | IConnectionParameters // Connection configuration
  readonly $dc: any // Database context
  readonly $config: ILibConfig // Library configuration
}

/**
 * Connection pool interface
 */
interface IPool {
  totalCount: number // Total connections in pool
  idleCount: number // Idle connections count
  waitingCount: number // Waiting clients count
  
  connect(): Promise<IClient> // Get connection from pool
  end(): Promise<void> // Close all connections
  
  on(event: string, listener: Function): IPool // Event handling
  removeListener(event: string, listener: Function): IPool
}

// Monitor pool status
console.log('Pool status:');
console.log('Total connections:', db.$pool.totalCount);
console.log('Idle connections:', db.$pool.idleCount);
console.log('Waiting clients:', db.$pool.waitingCount);

// Pool event handling
db.$pool.on('connect', client => {
  console.log('New client connected:', client.processID);
});

db.$pool.on('remove', client => {
  console.log('Client removed:', client.processID);
});

db.$pool.on('error', error => {
  console.error('Pool error:', error.message);
});

// Graceful shutdown
process.on('SIGINT', async () => {
  console.log('Closing database connections...');
  await db.$pool.end();
  process.exit(0);
});

/**
 * Establish direct connection (not from pool)
 * @param options - Connection options
 * @returns Promise resolving to connected client
 */
db.connect(options?: IConnectionOptions): Promise<IConnected>

interface IConnectionOptions {
  direct?: boolean // Use direct connection (bypass pool)
  onLost?(err: any, e: ILostContext): void // Connection lost handler
}

interface ILostContext {
  cn: string // Connection string
  dc: any // Database context
  start: Date // Connection start time
  client: IClient // Lost client instance
}

interface IConnected {
  readonly client: IClient // Raw database client
  
  // Connection management
  done(kill?: boolean): void | Promise<void> // Release connection
  
  // All database query methods available
  // (none, one, many, etc.)
  
  // Batch operations from spex
  batch(values: any[], options?: IBatchOptions): Promise<any[]>
  page(source: any, options?: IPageOptions, dest?: any): Promise<any>
  sequence(source: any, options?: ISequenceOptions, dest?: any): Promise<any>
}

// Basic connection
const connection = await db.connect();
try {
  const users = await connection.any('SELECT * FROM users');
  const user = await connection.one('SELECT * FROM users WHERE id = $1', [123]);
  
  // Use connection for multiple queries
  const profiles = await connection.any('SELECT * FROM user_profiles WHERE user_id = ANY($1)', 
    [users.map(u => u.id)]);
    
} finally {
  connection.done(); // Always release connection
}

// Direct connection (bypasses pool)
const directConn = await db.connect({ direct: true });
try {
  // Long-running operation that shouldn't tie up pool connection
  await directConn.any('SELECT * FROM large_table ORDER BY created_at');
} finally {
  await directConn.done(); // Direct connections return Promise
}

// Connection with loss monitoring
const monitoredConn = await db.connect({
  onLost: (err, context) => {
    console.error('Connection lost:', err.message);
    console.log('Connection was active for:', Date.now() - context.start.getTime(), 'ms');
    console.log('Client process ID:', context.client.processID);
  }
});

// Use connection...
monitoredConn.done();

// Connection with batch operations
const batchConn = await db.connect();
try {
  // Process data in batches
  const userIds = [1, 2, 3, 4, 5];
  const queries = userIds.map(id => 
    () => batchConn.one('SELECT * FROM users WHERE id = $1', [id])
  );
  
  const users = await batchConn.batch(queries, { concurrency: 3 });
  
} finally {
  batchConn.done();
}

/**
 * Connection event handlers in initialization options
 */
interface IInitOptions {
  // Connection events
  connect?(e: IConnectEvent): void // Client connected
  disconnect?(e: IDisconnectEvent): void // Client disconnected
  query?(e: IEventContext): void // Query executed
  receive?(e: IReceiveEvent): void // Data received
  task?(e: IEventContext): void // Task started
  transact?(e: IEventContext): void // Transaction started
  error?(err: any, e: IEventContext): void // Error occurred
  extend?(obj: IDatabase, dc: any): void // Database instance extended
}

interface IConnectEvent {
  client: IClient // Database client
  dc: any // Database context
  useCount: number // Connection use count
}

interface IDisconnectEvent {
  client: IClient // Database client
  dc: any // Database context
}

interface IReceiveEvent {
  data: any[] // Received data rows
  result: IResultExt | void // Full result object (undefined for streams)
  ctx: IEventContext // Event context
}

interface IEventContext {
  client: IClient // Database client
  cn: any // Connection configuration
  dc: any // Database context
  query: any // Query being executed
  params: any // Query parameters
  values: any // Parameter values
  queryFilePath?: string // Query file path (if applicable)
  ctx: ITaskContext // Task context
}

// Global connection monitoring
const pgp = require('pg-promise')({
  connect: (e) => {
    console.log('Connected to database:', {
      processId: e.client.processID,
      database: e.client.database,
      useCount: e.useCount
    });
  },
  
  disconnect: (e) => {
    console.log('Disconnected from database:', {
      processId: e.client.processID,
      database: e.client.database
    });
  },
  
  query: (e) => {
    console.log('Executing query:', {
      query: e.query,
      duration: Date.now() - e.ctx.start.getTime()
    });
  },
  
  receive: (e) => {
    console.log('Received data:', {
      rows: e.data?.length || 0,
      duration: e.result?.duration || 0
    });
  },
  
  error: (err, e) => {
    console.error('Database error:', {
      error: err.message,
      query: e.query,
      client: e.client.processID
    });
  }
});

const db = pgp(connectionString);

/**
 * Library termination and cleanup
 */
pgp.end(): void // Close all connection pools

/**
 * Individual pool termination
 */
db.$pool.end(): Promise<void> // Close specific connection pool

// Application shutdown
process.on('SIGTERM', () => {
  console.log('Shutting down gracefully...');
  
  // Close all pg-promise connection pools
  pgp.end();
  
  // Or close specific pool
  // await db.$pool.end();
});

// Graceful server shutdown
async function gracefulShutdown() {
  try {
    console.log('Closing database connections...');
    
    // Wait for active connections to finish
    while (db.$pool.totalCount > db.$pool.idleCount) {
      console.log('Waiting for connections to finish...');
      await new Promise(resolve => setTimeout(resolve, 100));
    }
    
    // Close the pool
    await db.$pool.end();
    console.log('Database connections closed');
    
  } catch (error) {
    console.error('Error during shutdown:', error);
  } finally {
    process.exit(0);
  }
}

// Multiple database instances
const userDB = pgp(userConnectionString);
const analyticsDB = pgp(analyticsConnectionString);

// Shutdown both
async function shutdownAll() {
  await Promise.all([
    userDB.$pool.end(),
    analyticsDB.$pool.end()
  ]);
  
  // Or close all at once
  pgp.end();
}

// Connection health check
async function checkDatabaseHealth() {
  try {
    await db.one('SELECT 1 as alive');
    return { status: 'healthy', timestamp: new Date() };
  } catch (error) {
    return { 
      status: 'unhealthy', 
      error: error.message, 
      timestamp: new Date() 
    };
  }
}

// Retry wrapper for connection issues
async function withRetry(operation, maxRetries = 3, delay = 1000) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await operation();
    } catch (error) {
      if (i === maxRetries - 1) throw error;
      
      if (error.code === 'ECONNRESET' || error.code === 'ECONNREFUSED') {
        console.log(`Connection failed, retrying in ${delay}ms... (${i + 1}/${maxRetries})`);
        await new Promise(resolve => setTimeout(resolve, delay));
        delay *= 2; // Exponential backoff
      } else {
        throw error; // Re-throw non-connection errors
      }
    }
  }
}

// Usage
const users = await withRetry(() => db.any('SELECT * FROM users'));

// Client interface (from node-postgres)
interface IClient {
  processID: number // PostgreSQL backend process ID
  secretKey: number // Secret key for cancellation
  database: string // Connected database name
  user: string // Connected user
  host: string // Database host
  port: number // Database port
  
  // Connection state
  connection: IConnection // Underlying connection
  
  // Query execution
  query(text: string, values?: any[]): Promise<IResult>
  query(config: IQueryConfig): Promise<IResult>
  
  // Connection management
  connect(): Promise<void>
  end(): Promise<void>
  
  // Event handling
  on(event: string, listener: Function): IClient
  removeListener(event: string, listener: Function): IClient
}

interface IConnection {
  stream: any // Network stream
  
  // Event handling
  on(event: string, listener: Function): IConnection
  removeListener(event: string, listener: Function): IConnection
}

// Query configuration
interface IQueryConfig {
  text: string // SQL query
  values?: any[] // Parameter values
  name?: string // Prepared statement name
  binary?: boolean // Binary result format
  rowMode?: 'array' // Row mode
  types?: ITypes // Type parsers
}

// Library configuration
interface ILibConfig {
  version: string // pg-promise version
  promise: IGenericPromise // Promise library
  options: IInitOptions // Initialization options
  pgp: IMain // Main pg-promise instance
  $npm: any // Internal npm modules
}

// Generic promise interface
interface IGenericPromise {
  (cb: (resolve: Function, reject: Function) => void): Promise<any>
  resolve(value?: any): Promise<any>
  reject(reason?: any): Promise<any>
  all(iterable: any): Promise<any>
}