← back to Wine Finder Next

lib/connectionPool.ts

384 lines

// Database Connection Pooling for Production
// Replace in-memory database with real database connections

/**
 * Connection Pool Manager
 *
 * In production, this would manage PostgreSQL/MongoDB connections
 * For now, provides the interface for future migration
 */

interface PoolConfig {
  min: number;
  max: number;
  acquireTimeout: number;
  idleTimeout: number;
}

interface PoolStats {
  total: number;
  active: number;
  idle: number;
  waiting: number;
}

class ConnectionPool {
  private config: PoolConfig;
  private connections: Set<any> = new Set();
  private available: any[] = [];
  private waitQueue: Array<(conn: any) => void> = [];
  private activeConnections = 0;
  private stats = {
    acquired: 0,
    released: 0,
    created: 0,
    destroyed: 0,
    timeouts: 0
  };

  constructor(config: Partial<PoolConfig> = {}) {
    this.config = {
      min: config.min || 2,
      max: config.max || 10,
      acquireTimeout: config.acquireTimeout || 30000,
      idleTimeout: config.idleTimeout || 300000
    };

    this.initialize();
  }

  private async initialize() {
    // Create minimum connections on startup
    for (let i = 0; i < this.config.min; i++) {
      await this.createConnection();
    }
  }

  private async createConnection(): Promise<any> {
    // In production, this would create actual database connection
    // For example: await pg.connect() or await mongoose.createConnection()

    const mockConnection = {
      id: crypto.randomUUID(),
      createdAt: Date.now(),
      lastUsed: Date.now(),
      inUse: false
    };

    this.connections.add(mockConnection);
    this.available.push(mockConnection);
    this.stats.created++;

    return mockConnection;
  }

  async acquire(): Promise<any> {
    // Try to get available connection
    if (this.available.length > 0) {
      const conn = this.available.pop()!;
      conn.inUse = true;
      conn.lastUsed = Date.now();
      this.activeConnections++;
      this.stats.acquired++;
      return conn;
    }

    // Create new connection if under max
    if (this.connections.size < this.config.max) {
      const conn = await this.createConnection();
      conn.inUse = true;
      this.activeConnections++;
      this.stats.acquired++;
      this.available.pop(); // Remove from available since we're using it
      return conn;
    }

    // Wait for connection to become available
    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        const index = this.waitQueue.indexOf(resolve);
        if (index > -1) {
          this.waitQueue.splice(index, 1);
        }
        this.stats.timeouts++;
        reject(new Error('Connection acquire timeout'));
      }, this.config.acquireTimeout);

      this.waitQueue.push((conn) => {
        clearTimeout(timeout);
        resolve(conn);
      });
    });
  }

  async release(conn: any): Promise<void> {
    if (!conn.inUse) {
      console.warn('Attempting to release connection that is not in use');
      return;
    }

    conn.inUse = false;
    conn.lastUsed = Date.now();
    this.activeConnections--;
    this.stats.released++;

    // If someone is waiting, give them the connection
    if (this.waitQueue.length > 0) {
      const waiter = this.waitQueue.shift()!;
      conn.inUse = true;
      this.activeConnections++;
      this.stats.acquired++;
      waiter(conn);
      return;
    }

    // Return to available pool
    this.available.push(conn);

    // Destroy excess idle connections
    await this.cleanupIdleConnections();
  }

  private async cleanupIdleConnections() {
    const now = Date.now();
    const toDestroy: any[] = [];

    // Keep minimum connections, destroy idle ones over minimum
    if (this.available.length > this.config.min) {
      for (const conn of this.available) {
        if (now - conn.lastUsed > this.config.idleTimeout) {
          toDestroy.push(conn);
        }
      }
    }

    for (const conn of toDestroy) {
      await this.destroyConnection(conn);
    }
  }

  private async destroyConnection(conn: any) {
    // In production: await conn.close()

    this.connections.delete(conn);
    const index = this.available.indexOf(conn);
    if (index > -1) {
      this.available.splice(index, 1);
    }
    this.stats.destroyed++;
  }

  getStats(): PoolStats {
    return {
      total: this.connections.size,
      active: this.activeConnections,
      idle: this.available.length,
      waiting: this.waitQueue.length
    };
  }

  getDetailedStats() {
    return {
      ...this.getStats(),
      ...this.stats,
      config: this.config
    };
  }

  async drain() {
    // Wait for all active connections to be released
    while (this.activeConnections > 0) {
      await new Promise(resolve => setTimeout(resolve, 100));
    }

    // Destroy all connections
    for (const conn of Array.from(this.connections)) {
      await this.destroyConnection(conn);
    }
  }
}

// Singleton instance
let pool: ConnectionPool | null = null;

export function getConnectionPool(config?: Partial<PoolConfig>): ConnectionPool {
  if (!pool) {
    pool = new ConnectionPool(config);
  }
  return pool;
}

export async function withConnection<T>(
  operation: (conn: any) => Promise<T>
): Promise<T> {
  const pool = getConnectionPool();
  const conn = await pool.acquire();

  try {
    return await operation(conn);
  } finally {
    await pool.release(conn);
  }
}

// ============= TRANSACTION SUPPORT =============

export async function withTransaction<T>(
  operation: (conn: any, tx: any) => Promise<T>
): Promise<T> {
  const pool = getConnectionPool();
  const conn = await pool.acquire();

  // Mock transaction object
  // In production: const tx = await conn.beginTransaction()
  const tx = {
    commit: async () => { /* commit logic */ },
    rollback: async () => { /* rollback logic */ }
  };

  try {
    const result = await operation(conn, tx);
    await tx.commit();
    return result;
  } catch (error) {
    await tx.rollback();
    throw error;
  } finally {
    await pool.release(conn);
  }
}

// ============= QUERY HELPERS =============

export interface QueryOptions {
  timeout?: number;
  retries?: number;
  cache?: boolean;
  cacheTTL?: number;
}

const queryCache = new Map<string, { data: any; timestamp: number }>();

export async function query<T>(
  sql: string,
  params: any[] = [],
  options: QueryOptions = {}
): Promise<T> {
  const {
    timeout = 10000,
    retries = 3,
    cache = false,
    cacheTTL = 60000
  } = options;

  // Check cache
  if (cache) {
    const cacheKey = `${sql}:${JSON.stringify(params)}`;
    const cached = queryCache.get(cacheKey);

    if (cached && Date.now() - cached.timestamp < cacheTTL) {
      return cached.data as T;
    }
  }

  return withConnection(async (conn) => {
    let lastError: Error | null = null;

    for (let attempt = 0; attempt < retries; attempt++) {
      try {
        // In production: const result = await conn.query(sql, params, { timeout })
        const result = { rows: [], rowCount: 0 } as any;

        // Cache result
        if (cache) {
          const cacheKey = `${sql}:${JSON.stringify(params)}`;
          queryCache.set(cacheKey, { data: result, timestamp: Date.now() });
        }

        return result as T;
      } catch (error) {
        lastError = error as Error;

        // Don't retry on certain errors
        if (error instanceof Error && error.message.includes('syntax')) {
          break;
        }

        // Wait before retry
        if (attempt < retries - 1) {
          await new Promise(resolve => setTimeout(resolve, Math.pow(2, attempt) * 100));
        }
      }
    }

    throw lastError || new Error('Query failed');
  });
}

// ============= BATCH OPERATIONS =============

export async function batchInsert<T>(
  table: string,
  records: T[],
  batchSize: number = 100
): Promise<void> {
  return withConnection(async (conn) => {
    for (let i = 0; i < records.length; i += batchSize) {
      const batch = records.slice(i, i + batchSize);

      // In production: build and execute batch insert query
      // await conn.query(`INSERT INTO ${table} VALUES ...`, batch);

      console.log(`Inserted batch ${i / batchSize + 1} with ${batch.length} records`);
    }
  });
}

// ============= HEALTH CHECK =============

export async function healthCheck(): Promise<{
  healthy: boolean;
  stats: PoolStats;
  latency: number;
}> {
  const start = Date.now();
  const pool = getConnectionPool();

  try {
    await withConnection(async (conn) => {
      // In production: await conn.query('SELECT 1')
      return true;
    });

    return {
      healthy: true,
      stats: pool.getStats(),
      latency: Date.now() - start
    };
  } catch (error) {
    return {
      healthy: false,
      stats: pool.getStats(),
      latency: Date.now() - start
    };
  }
}

// ============= CLEANUP ON SHUTDOWN =============

process.on('SIGTERM', async () => {
  console.log('SIGTERM received, draining connection pool...');
  if (pool) {
    await pool.drain();
  }
  process.exit(0);
});

process.on('SIGINT', async () => {
  console.log('SIGINT received, draining connection pool...');
  if (pool) {
    await pool.drain();
  }
  process.exit(0);
});