import { Pool , PoolClient , PoolConfig } from 'pg' ; import { DatabaseConnection , QueryResult , } from '../../driver/database-connection.js' ; import { Driver , TransactionSettings } from '../../driver/driver.js' ; import { CompiledQuery } from '../../query-compiler/compiled-query.js' ; import { isFunction , freeze } from '../../util/object-utils.js' ; import { PostgresDialectConfig } from './postgres-dialect-config.js' ; const PRIVATE_RELEASE_METHOD = Symbol ( ) ; export class PostgresDriver implements Driver { readonly #config : PostgresDialectConfig ; readonly #connections = new WeakMap < PoolClient , DatabaseConnection > ( ) ; #pool : Pool | null = null ; constructor ( config ) { this . #config = freeze ( { ... config } ) ; } async init ( ) { const PoolConstructor = await importPgPool ( ) ; this . #pool = new PoolConstructor ( this . #config ) ; } async acquireConnection ( ) { const client = await this . #pool ! . connect ( ) ; let connection = this . #connections . get ( client ) ; if ( ! connection ) { connection = new PostgresConnection ( client ) ; this . #connections . set ( client , connection ) ; if ( this . #config . onCreateConnection ) { await this . #config . onCreateConnection ( connection ) ; } } return connection ; } async beginTransaction ( connection , settings ) { if ( settings . isolationLevel ) { await connection . executeQuery ( CompiledQuery . raw ( `start transaction isolation level ${ settings . isolationLevel }` ) ) ; } else { await connection . executeQuery ( CompiledQuery . raw ( 'begin' ) ) ; } } async commitTransaction ( connection ) { await connection . executeQuery ( CompiledQuery . raw ( 'commit' ) ) ; } async rollbackTransaction ( connection ) { await connection . executeQuery ( CompiledQuery . raw ( 'rollback' ) ) ; } async releaseConnection ( connection ) { const pgConnection = connection as PostgresConnection ; pgConnection [ PRIVATE_RELEASE_METHOD ] ( ) ; } async destroy ( ) { if ( this . #pool ) { const pool = this . #pool ; this . #pool = null ; await pool . end ( ) ; } } } type PoolConstructor = new ( config ) => Pool ; async function importPgPool ( ) { try { const pgModule = await import ( 'pg' ) ; if ( isFunction ( pgModule . Pool ) ) { return pgModule . Pool ; } else { return pgModule . default . Pool ; } } catch ( error ) { throw new Error ( 'Postgres client not installed. Please run `npm install pg`' ) ; } } class PostgresConnection implements DatabaseConnection { #client : PoolClient ; constructor ( client ) { this . #client = client ; } async executeQuery < O > ( compiledQuery ) { const result = await this . #client . query < O > ( compiledQuery . sql , [ ... compiledQuery . parameters , ] ) ; if ( result . command === 'UPDATE' || result . command === 'DELETE' ) { return { numUpdatedOrDeletedRows : BigInt ( result . rowCount ) , rows : result . rows ?? [ ] , } ; } return { rows : result . rows ?? [ ] , } ; } [ PRIVATE_RELEASE_METHOD ] ( ) { this . #client . release ( ) ; } }