import { BinaryDataManager , IBinaryDataConfig , UserSettings } from 'n8n-core' ; import { Command , flags } from '@oclif/command' ; import * as Redis from 'ioredis' ; import { IDataObject , LoggerProxy } from 'n8n-workflow' ; import * as config from '../config' ; import { ActiveExecutions , ActiveWorkflowRunner , CredentialsOverwrites , CredentialTypes , Db , ExternalHooks , GenericHelpers , InternalHooksManager , LoadNodesAndCredentials , NodeTypes , WebhookServer , } from '../src' ; import { getLogger } from '../src/Logger' ; let activeWorkflowRunner ; let processExistCode = 0 ; export class Webhook extends Command { static description = 'Starts n8n webhook process. Intercepts only production URLs.' ; static examples = [ `$ n8n webhook` ] ; static flags = { help : flags . help ( { char : 'h' } ) , } ; /**
     * Stops the n8n in a graceful way.
     * Make for example sure that all the webhooks from third party services
     * get removed.
     */ static async stopProcess ( ) { LoggerProxy . info ( `\nStopping n8n...` ) ; try { const externalHooks = ExternalHooks ( ) ; await externalHooks . run ( 'n8n.stop' , [ ] ) ; setTimeout ( ( ) => { process . exit ( processExistCode ) ; } , 30000 ) ; const activeExecutionsInstance = ActiveExecutions . getInstance ( ) ; let executingWorkflows = activeExecutionsInstance . getActiveExecutions ( ) ; let count = 0 ; while ( executingWorkflows . length !== 0 ) { if ( count ++ % 4 === 0 ) { LoggerProxy . info ( `Waiting for ${ executingWorkflows . length } active executions to finish...` ) ; } await new Promise ( ( resolve ) => { setTimeout ( resolve , 500 ) ; } ) ; executingWorkflows = activeExecutionsInstance . getActiveExecutions ( ) ; } } catch ( error ) { LoggerProxy . error ( 'There was an error shutting down n8n.' , error ) ; } process . exit ( processExistCode ) ; } async run ( ) { const logger = getLogger ( ) ; LoggerProxy . init ( logger ) ; process . on ( 'SIGTERM' , Webhook . stopProcess ) ; process . on ( 'SIGINT' , Webhook . stopProcess ) ; const { flags } = this . parse ( Webhook ) ; await ( async ( ) => { if ( config . get ( 'executions.mode' ) !== 'queue' ) { /**
                 * It is technically possible to run without queues but
                 * there are 2 known bugs when running in this mode:
                 * - Executions list will be problematic as the main process
                 * is not aware of current executions in the webhook processes
                 * and therefore will display all current executions as error
                 * as it is unable to determine if it is still running or crashed
                 * - You cannot stop currently executing jobs from webhook processes
                 * when running without queues as the main process cannot talk to
                 * the wehbook processes to communicate workflow execution interruption.
                 */ this . error ( 'Webhook processes can only run with execution mode as queue.' ) ; } try { const startDbInitPromise = Db . init ( ) . catch ( ( error ) => { logger . error ( `There was an error initializing DB: "${ error . message }"` ) ; processExistCode = 1 ; process . emit ( 'SIGINT' ) ; process . exit ( 1 ) ; } ) ; const userSettings = await UserSettings . prepareUserSettings ( ) ; const loadNodesAndCredentials = LoadNodesAndCredentials ( ) ; await loadNodesAndCredentials . init ( ) ; const credentialsOverwrites = CredentialsOverwrites ( ) ; await credentialsOverwrites . init ( ) ; const externalHooks = ExternalHooks ( ) ; await externalHooks . init ( ) ; const nodeTypes = NodeTypes ( ) ; await nodeTypes . init ( loadNodesAndCredentials . nodeTypes ) ; const credentialTypes = CredentialTypes ( ) ; await credentialTypes . init ( loadNodesAndCredentials . credentialTypes ) ; await startDbInitPromise ; const instanceId = await UserSettings . getInstanceId ( ) ; const { cli } = await GenericHelpers . getVersions ( ) ; InternalHooksManager . init ( instanceId , cli , nodeTypes ) ; const binaryDataConfig = config . get ( 'binaryDataManager' ) as IBinaryDataConfig ; await BinaryDataManager . init ( binaryDataConfig ) ; if ( config . get ( 'executions.mode' ) === 'queue' ) { const redisHost = config . get ( 'queue.bull.redis.host' ) ; const redisPassword = config . get ( 'queue.bull.redis.password' ) ; const redisPort = config . get ( 'queue.bull.redis.port' ) ; const redisDB = config . get ( 'queue.bull.redis.db' ) ; const redisConnectionTimeoutLimit = config . get ( 'queue.bull.redis.timeoutThreshold' ) ; let lastTimer = 0 ; let cumulativeTimeout = 0 ; const settings = { retryStrategy : ( times ) : number | null => { const now = Date . now ( ) ; if ( now - lastTimer > 30000 ) { lastTimer = now ; cumulativeTimeout = 0 ; } else { cumulativeTimeout += now - lastTimer ; lastTimer = now ; if ( cumulativeTimeout > redisConnectionTimeoutLimit ) { logger . error ( `Unable to connect to Redis after ${ redisConnectionTimeoutLimit }. Exiting process.` ) ; process . exit ( 1 ) ; } } return 500 ; } , } as IDataObject ; if ( redisHost ) { settings . host = redisHost ; } if ( redisPassword ) { settings . password = redisPassword ; } if ( redisPort ) { settings . port = redisPort ; } if ( redisDB ) { settings . db = redisDB ; } const redis = new Redis ( settings ) ; redis . on ( 'error' , ( error ) => { if ( error . toString ( ) . includes ( 'ECONNREFUSED' ) === true ) { logger . warn ( 'Redis unavailable - trying to reconnect...' ) ; } else { logger . warn ( 'Error with Redis: ' , error ) ; } } ) ; } await WebhookServer . start ( ) ; activeWorkflowRunner = ActiveWorkflowRunner . getInstance ( ) ; await activeWorkflowRunner . initWebhooks ( ) ; const editorUrl = GenericHelpers . getBaseUrl ( ) ; console . info ( 'Webhook listener waiting for requests.' ) ; } catch ( error ) { console . error ( 'Exiting due to error. See log message for details.' ) ; logger . error ( `Webhook process cannot continue. "${ error . message }"` ) ; processExistCode = 1 ; process . emit ( 'SIGINT' ) ; process . exit ( 1 ) ; } } ) ( ) ; } }