import * as assert from "assert" ; import { EventEmitter } from "events" ; import { Client , Pool , PoolClient } from "pg" ; import { defaults } from "./config" ; import { makeAddJob , makeWithPgClientFromPool } from "./helpers" ; import { AddJobFunction , RunnerOptions , SharedOptions , WithPgClient , WorkerEvents , } from "./interfaces" ; import { defaultLogger , Logger , LogScope } from "./logger" ; import { migrate } from "./migrate" ; interface CompiledSharedOptions { events : WorkerEvents ; logger : Logger ; workerSchema : string ; escapedWorkerSchema : string ; maxContiguousErrors : number ; useNodeTime : boolean ; } interface ProcessSharedOptionsSettings { scope ? : LogScope ; } const _sharedOptionsCache = new WeakMap < SharedOptions , CompiledSharedOptions > ( ) ; export function processSharedOptions ( options , { scope } = { } ) { let compiled = _sharedOptionsCache . get ( options ) ; if ( ! compiled ) { const { logger = defaultLogger , schema : workerSchema = defaults . schema , events = new EventEmitter ( ) , useNodeTime = false , } = options ; const escapedWorkerSchema = Client . prototype . escapeIdentifier ( workerSchema ) ; compiled = { events , logger , workerSchema , escapedWorkerSchema , maxContiguousErrors : defaults . maxContiguousErrors , useNodeTime , } ; _sharedOptionsCache . set ( options , compiled ) ; } if ( scope ) { return { ... compiled , logger : compiled . logger . scope ( scope ) , } ; } else { return compiled ; } } export type Releasers = Array < ( ) => void | Promise < void > > ; export async function assertPool ( options , releasers ) { const { logger } = processSharedOptions ( options ) ; assert ( ! options . pgPool || ! options . connectionString , "Both `pgPool` and `connectionString` are set, at most one of these options should be provided" ) ; let pgPool ; const connectionString = options . connectionString || process . env . DATABASE_URL ; if ( options . pgPool ) { pgPool = options . pgPool ; } else if ( connectionString ) { pgPool = new Pool ( { connectionString , max : options . maxPoolSize , } ) ; releasers . push ( ( ) => { pgPool . end ( ) ; } ) ; } else if ( process . env . PGDATABASE ) { pgPool = new Pool ( { max : options . maxPoolSize , } ) ; releasers . push ( ( ) => { pgPool . end ( ) ; } ) ; } else { throw new Error ( "You must either specify `pgPool` or `connectionString`, or you must make the `DATABASE_URL` or `PG*` environmental variables available." ) ; } const handlePoolError = ( err ) => { logger . error ( `PostgreSQL idle client generated error: ${ err . message }` , { error : err , } ) ; } ; const handleClientError = ( err ) => { logger . error ( `PostgreSQL active client generated error: ${ err . message }` , { error : err , } ) ; } ; pgPool . on ( "error" , handlePoolError ) ; const handlePoolConnect = ( client ) => { client . on ( "error" , handleClientError ) ; } ; pgPool . on ( "connect" , handlePoolConnect ) ; releasers . push ( ( ) => { pgPool . removeListener ( "error" , handlePoolError ) ; pgPool . removeListener ( "connect" , handlePoolConnect ) ; } ) ; return pgPool ; } export type Release = ( ) => Promise < void > ; export async function withReleasers < T > ( callback ) { const releasers = [ ] ; const release = async ( ) => { let firstError = null ; for ( let i = releasers . length - 1 ; i >= 0 ; i -- ) { try { await releasers [ i ] ( ) ; } catch ( e ) { firstError = firstError || e ; } } if ( firstError ) { throw firstError ; } } ; try { return await callback ( releasers , release ) ; } catch ( e ) { try { await release ( ) ; } catch ( e2 ) { } throw e ; } } interface ProcessOptionsExtensions { pgPool : Pool ; withPgClient : WithPgClient ; addJob : AddJobFunction ; release : Release ; releasers : Releasers ; } export interface CompiledOptions extends CompiledSharedOptions , ProcessOptionsExtensions { } export const getUtilsAndReleasersFromOptions = async ( options , settings = { } ) : Promise < CompiledOptions > => { const shared = processSharedOptions ( options , settings ) ; const { concurrency = defaults . concurrentJobs } = options ; return withReleasers ( async ( releasers , release ) : Promise < CompiledOptions > => { const pgPool = await assertPool ( options , releasers ) ; const max = pgPool ?. options ?. max || 10 ; if ( max < concurrency ) { console . warn ( `WARNING: having maxPoolSize (${ max }) smaller than concurrency (${ concurrency }) may lead to non-optimal performance.` ) ; } const withPgClient = makeWithPgClientFromPool ( pgPool ) ; await withPgClient ( ( client ) => migrate ( options , client ) ) ; const addJob = makeAddJob ( options , withPgClient ) ; return { ... shared , pgPool , withPgClient , addJob , release , releasers , } ; } ) ; } ;