import { Pool , PoolClient } from "pg" ; import { Job , JobHelpers , SharedOptions , TaskSpec , WithPgClient , WorkerSharedOptions , } from "./interfaces" ; import { processSharedOptions } from "./lib" ; import { Logger } from "./logger" ; export function makeAddJob ( options , withPgClient ) { const { escapedWorkerSchema , useNodeTime } = processSharedOptions ( options ) ; return ( identifier , payload = { } , spec = { } ) => { return withPgClient ( async ( pgClient ) => { const { rows } = await pgClient . query ( `
        select * from ${ escapedWorkerSchema }.add_job(
          identifier => $1::text,
          payload => $2::json,
          queue_name => $3::text,
          run_at => $4::timestamptz,
          max_attempts => $5::int,
          job_key => $6::text,
          priority => $7::int,
          flags => $8::text[],
          job_key_mode => $9::text
        );
        ` , [ identifier , JSON . stringify ( payload ) , spec . queueName || null , spec . runAt ? spec . runAt . toISOString ( ) : useNodeTime ? new Date ( ) . toISOString ( ) : null , spec . maxAttempts || null , spec . jobKey || null , spec . priority || null , spec . flags || null , spec . jobKeyMode || null , ] ) ; const job = rows [ 0 ] ; return job ; } ) ; } ; } export function makeJobHelpers ( options , job , { withPgClient , logger : overrideLogger , } ) { const baseLogger = overrideLogger || processSharedOptions ( options ) . logger ; const logger = baseLogger . scope ( { label : "job" , taskIdentifier : job . task_identifier , jobId : job . id , } ) ; const helpers = { job , logger , withPgClient , query : ( queryText , values ) => withPgClient ( ( pgClient ) => pgClient . query ( queryText , values ) ) , addJob : makeAddJob ( options , withPgClient ) , } ; Object . assign ( helpers , { debug ( format , ... parameters ) { logger . error ( "REMOVED: `helpers.debug` has been replaced with `helpers.logger.debug`; please do not use `helpers.debug`" ) ; logger . debug ( format , { parameters } ) ; } , } as any ) ; return helpers ; } export function makeWithPgClientFromPool ( pgPool ) { return async < T > ( callback ) => { const client = await pgPool . connect ( ) ; try { return await callback ( client ) ; } finally { await client . release ( ) ; } } ; } export function makeWithPgClientFromClient ( pgClient ) { return async < T > ( callback ) => { return callback ( pgClient ) ; } ; }