admin/n8n-enterprise-unlocked
1
0
Fork 0
mirror of https://github.com/Abdulazizzn/n8n-enterprise-unlocked.git synced 2025-12-18 02:21:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): Move execution engine code out of n8n-workflow (no-changelog) (#12147)

Browse Source
This commit is contained in:
कारतोफ्फेलस्क्रिप्ट™
2024-12-12 13:54:44 +01:00
committed by GitHub
parent 73f0c4cca9
commit 5a055ed526
44 changed files with 1995 additions and 1795 deletions
Download Patch File Download Diff File Expand all files Collapse all files

565
packages/workflow/src/Workflow.ts
View File

@@ -1,62 +1,36 @@
/* eslint-disable @typescript-eslint/no-use-before-define */
/* eslint-disable @typescript-eslint/no-unsafe-member-access */
/* eslint-disable @typescript-eslint/no-unsafe-return */
/* eslint-disable @typescript-eslint/no-for-in-array */
/* eslint-disable @typescript-eslint/prefer-nullish-coalescing */
import {
MANUAL_CHAT_TRIGGER_LANGCHAIN_NODE_TYPE,
NODES_WITH_RENAMABLE_CONTENT,
STARTING_NODE_TYPES,
} from './Constants';
import type { IDeferredPromise } from './DeferredPromise';
import { ApplicationError } from './errors/application.error';
import { Expression } from './Expression';
import { getGlobalState } from './GlobalState';
import type {
IConnections,
IExecuteResponsePromiseData,
IGetExecuteTriggerFunctions,
INode,
INodeExecuteFunctions,
INodeExecutionData,
INodeIssues,
INodeParameters,
INodes,
INodeType,
INodeTypes,
IPinData,
IPollFunctions,
IRunExecutionData,
ITaskDataConnections,
ITriggerResponse,
IWebhookData,
IWebhookResponseData,
IWorkflowIssues,
IWorkflowExecuteAdditionalData,
IWorkflowSettings,
WebhookSetupMethodNames,
WorkflowActivateMode,
WorkflowExecuteMode,
IConnection,
IConnectedNode,
IDataObject,
IExecuteData,
INodeConnection,
IObservableObject,
IRun,
IRunNodeResponse,
NodeParameterValueType,
CloseFunction,
INodeOutputConfiguration,
} from './Interfaces';
import { Node, NodeConnectionType } from './Interfaces';
import { NodeConnectionType } from './Interfaces';
import * as NodeHelpers from './NodeHelpers';
import * as ObservableObject from './ObservableObject';
import { RoutingNode } from './RoutingNode';
function dedupe<T>(arr: T[]): T[] {
return [...new Set(arr)];
@@ -214,112 +188,6 @@ export class Workflow {
return returnConnection;
}
/**
* A workflow can only be activated if it has a node which has either triggers
* or webhooks defined.
*
* @param {string[]} [ignoreNodeTypes] Node-types to ignore in the check
*/
checkIfWorkflowCanBeActivated(ignoreNodeTypes?: string[]): boolean {
let node: INode;
let nodeType: INodeType | undefined;
for (const nodeName of Object.keys(this.nodes)) {
node = this.nodes[nodeName];
if (node.disabled === true) {
// Deactivated nodes can not trigger a run so ignore
continue;
}
if (ignoreNodeTypes !== undefined && ignoreNodeTypes.includes(node.type)) {
continue;
}
nodeType = this.nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
if (nodeType === undefined) {
// Type is not known so check is not possible
continue;
}
if (
nodeType.poll !== undefined ||
nodeType.trigger !== undefined ||
nodeType.webhook !== undefined
) {
// Is a trigger node. So workflow can be activated.
return true;
}
}
return false;
}
/**
* Checks if everything in the workflow is complete
* and ready to be executed. If it returns null everything
* is fine. If there are issues it returns the issues
* which have been found for the different nodes.
* TODO: Does currently not check for credential issues!
*/
checkReadyForExecution(
inputData: {
startNode?: string;
destinationNode?: string;
pinDataNodeNames?: string[];
} = {},
): IWorkflowIssues | null {
const workflowIssues: IWorkflowIssues = {};
let checkNodes: string[] = [];
if (inputData.destinationNode) {
// If a destination node is given we have to check all the nodes
// leading up to it
checkNodes = this.getParentNodes(inputData.destinationNode);
checkNodes.push(inputData.destinationNode);
} else if (inputData.startNode) {
// If a start node is given we have to check all nodes which
// come after it
checkNodes = this.getChildNodes(inputData.startNode);
checkNodes.push(inputData.startNode);
}
for (const nodeName of checkNodes) {
let nodeIssues: INodeIssues | null = null;
const node = this.nodes[nodeName];
if (node.disabled === true) {
continue;
}
const nodeType = this.nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
if (nodeType === undefined) {
// Node type is not known
nodeIssues = {
typeUnknown: true,
};
} else {
nodeIssues = NodeHelpers.getNodeParametersIssues(
nodeType.description.properties,
node,
inputData.pinDataNodeNames,
);
}
if (nodeIssues !== null) {
workflowIssues[node.name] = nodeIssues;
}
}
if (Object.keys(workflowIssues).length === 0) {
return null;
}
return workflowIssues;
}
/**
* Returns the static data of the workflow.
* It gets saved with the workflow and will be the same for
@@ -1065,437 +933,6 @@ export class Workflow {
return this.__getStartNode(Object.keys(this.nodes));
}
async createWebhookIfNotExists(
webhookData: IWebhookData,
nodeExecuteFunctions: INodeExecuteFunctions,
mode: WorkflowExecuteMode,
activation: WorkflowActivateMode,
): Promise<void> {
const webhookExists = await this.runWebhookMethod(
'checkExists',
webhookData,
nodeExecuteFunctions,
mode,
activation,
);
if (!webhookExists) {
// If webhook does not exist yet create it
await this.runWebhookMethod('create', webhookData, nodeExecuteFunctions, mode, activation);
}
}
async deleteWebhook(
webhookData: IWebhookData,
nodeExecuteFunctions: INodeExecuteFunctions,
mode: WorkflowExecuteMode,
activation: WorkflowActivateMode,
) {
await this.runWebhookMethod('delete', webhookData, nodeExecuteFunctions, mode, activation);
}
private async runWebhookMethod(
method: WebhookSetupMethodNames,
webhookData: IWebhookData,
nodeExecuteFunctions: INodeExecuteFunctions,
mode: WorkflowExecuteMode,
activation: WorkflowActivateMode,
): Promise<boolean | undefined> {
const node = this.getNode(webhookData.node);
if (!node) return;
const nodeType = this.nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
const webhookFn = nodeType.webhookMethods?.[webhookData.webhookDescription.name]?.[method];
if (webhookFn === undefined) return;
const thisArgs = nodeExecuteFunctions.getExecuteHookFunctions(
this,
node,
webhookData.workflowExecuteAdditionalData,
mode,
activation,
webhookData,
);
return await webhookFn.call(thisArgs);
}
/**
* Runs the given trigger node so that it can trigger the workflow
* when the node has data.
*
*/
async runTrigger(
node: INode,
getTriggerFunctions: IGetExecuteTriggerFunctions,
additionalData: IWorkflowExecuteAdditionalData,
mode: WorkflowExecuteMode,
activation: WorkflowActivateMode,
): Promise<ITriggerResponse | undefined> {
const triggerFunctions = getTriggerFunctions(this, node, additionalData, mode, activation);
const nodeType = this.nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
if (nodeType === undefined) {
throw new ApplicationError('Node with unknown node type', {
extra: { nodeName: node.name },
tags: { nodeType: node.type },
});
}
if (!nodeType.trigger) {
throw new ApplicationError('Node type does not have a trigger function defined', {
extra: { nodeName: node.name },
tags: { nodeType: node.type },
});
}
if (mode === 'manual') {
// In manual mode we do not just start the trigger function we also
// want to be able to get informed as soon as the first data got emitted
const triggerResponse = await nodeType.trigger.call(triggerFunctions);
// Add the manual trigger response which resolves when the first time data got emitted
triggerResponse!.manualTriggerResponse = new Promise((resolve, reject) => {
triggerFunctions.emit = (
(resolveEmit) =>
(
data: INodeExecutionData[][],
responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
donePromise?: IDeferredPromise<IRun>,
) => {
additionalData.hooks!.hookFunctions.sendResponse = [
async (response: IExecuteResponsePromiseData): Promise<void> => {
if (responsePromise) {
responsePromise.resolve(response);
}
},
];
if (donePromise) {
additionalData.hooks!.hookFunctions.workflowExecuteAfter?.unshift(
async (runData: IRun): Promise<void> => {
return donePromise.resolve(runData);
},
);
}
resolveEmit(data);
}
)(resolve);
triggerFunctions.emitError = (
(rejectEmit) =>
(error: Error, responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>) => {
additionalData.hooks!.hookFunctions.sendResponse = [
async (): Promise<void> => {
if (responsePromise) {
responsePromise.reject(error);
}
},
];
rejectEmit(error);
}
)(reject);
});
return triggerResponse;
}
// In all other modes simply start the trigger
return await nodeType.trigger.call(triggerFunctions);
}
/**
* Runs the given trigger node so that it can trigger the workflow
* when the node has data.
*
*/
async runPoll(
node: INode,
pollFunctions: IPollFunctions,
): Promise<INodeExecutionData[][] | null> {
const nodeType = this.nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
if (nodeType === undefined) {
throw new ApplicationError('Node with unknown node type', {
extra: { nodeName: node.name },
tags: { nodeType: node.type },
});
}
if (!nodeType.poll) {
throw new ApplicationError('Node type does not have a poll function defined', {
extra: { nodeName: node.name },
tags: { nodeType: node.type },
});
}
return await nodeType.poll.call(pollFunctions);
}
/**
* Executes the webhook data to see what it should return and if the
* workflow should be started or not
*
*/
async runWebhook(
webhookData: IWebhookData,
node: INode,
additionalData: IWorkflowExecuteAdditionalData,
nodeExecuteFunctions: INodeExecuteFunctions,
mode: WorkflowExecuteMode,
runExecutionData: IRunExecutionData | null,
): Promise<IWebhookResponseData> {
const nodeType = this.nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
if (nodeType === undefined) {
throw new ApplicationError('Unknown node type of webhook node', {
extra: { nodeName: node.name },
});
} else if (nodeType.webhook === undefined) {
throw new ApplicationError('Node does not have any webhooks defined', {
extra: { nodeName: node.name },
});
}
const closeFunctions: CloseFunction[] = [];
const context = nodeExecuteFunctions.getExecuteWebhookFunctions(
this,
node,
additionalData,
mode,
webhookData,
closeFunctions,
runExecutionData,
);
return nodeType instanceof Node
? await nodeType.webhook(context)
: await nodeType.webhook.call(context);
}
/**
* Executes the given node.
*
*/
// eslint-disable-next-line complexity
async runNode(
executionData: IExecuteData,
runExecutionData: IRunExecutionData,
runIndex: number,
additionalData: IWorkflowExecuteAdditionalData,
nodeExecuteFunctions: INodeExecuteFunctions,
mode: WorkflowExecuteMode,
abortSignal?: AbortSignal,
): Promise<IRunNodeResponse> {
const { node } = executionData;
let inputData = executionData.data;
if (node.disabled === true) {
// If node is disabled simply pass the data through
// return NodeRunHelpers.
if (inputData.hasOwnProperty('main') && inputData.main.length > 0) {
// If the node is disabled simply return the data from the first main input
if (inputData.main[0] === null) {
return { data: undefined };
}
return { data: [inputData.main[0]] };
}
return { data: undefined };
}
const nodeType = this.nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
if (nodeType === undefined) {
throw new ApplicationError('Node type is unknown so cannot run it', {
tags: { nodeType: node.type },
});
}
let connectionInputData: INodeExecutionData[] = [];
if (nodeType.execute || (!nodeType.poll && !nodeType.trigger && !nodeType.webhook)) {
// Only stop if first input is empty for execute runs. For all others run anyways
// because then it is a trigger node. As they only pass data through and so the input-data
// becomes output-data it has to be possible.
if (inputData.main?.length > 0) {
// We always use the data of main input and the first input for execute
connectionInputData = inputData.main[0] as INodeExecutionData[];
}
const forceInputNodeExecution = this.settings.executionOrder !== 'v1';
if (!forceInputNodeExecution) {
// If the nodes do not get force executed data of some inputs may be missing
// for that reason do we use the data of the first one that contains any
for (const mainData of inputData.main) {
if (mainData?.length) {
connectionInputData = mainData;
break;
}
}
}
if (connectionInputData.length === 0) {
// No data for node so return
return { data: undefined };
}
}
if (
runExecutionData.resultData.lastNodeExecuted === node.name &&
runExecutionData.resultData.error !== undefined
) {
// The node did already fail. So throw an error here that it displays and logs it correctly.
// Does get used by webhook and trigger nodes in case they throw an error that it is possible
// to log the error and display in Editor-UI.
if (
runExecutionData.resultData.error.name === 'NodeOperationError' ||
runExecutionData.resultData.error.name === 'NodeApiError'
) {
throw runExecutionData.resultData.error;
}
const error = new Error(runExecutionData.resultData.error.message);
error.stack = runExecutionData.resultData.error.stack;
throw error;
}
if (node.executeOnce === true) {
// If node should be executed only once so use only the first input item
const newInputData: ITaskDataConnections = {};
for (const connectionType of Object.keys(inputData)) {
newInputData[connectionType] = inputData[connectionType].map((input) => {
// eslint-disable-next-line @typescript-eslint/prefer-optional-chain
return input && input.slice(0, 1);
});
}
inputData = newInputData;
}
if (nodeType.execute) {
const closeFunctions: CloseFunction[] = [];
const context = nodeExecuteFunctions.getExecuteFunctions(
this,
runExecutionData,
runIndex,
connectionInputData,
inputData,
node,
additionalData,
executionData,
mode,
closeFunctions,
abortSignal,
);
const data =
nodeType instanceof Node
? await nodeType.execute(context)
: await nodeType.execute.call(context);
const closeFunctionsResults = await Promise.allSettled(
closeFunctions.map(async (fn) => await fn()),
);
const closingErrors = closeFunctionsResults
.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
.map((result) => result.reason);
if (closingErrors.length > 0) {
if (closingErrors[0] instanceof Error) throw closingErrors[0];
throw new ApplicationError("Error on execution node's close function(s)", {
extra: { nodeName: node.name },
tags: { nodeType: node.type },
cause: closingErrors,
});
}
return { data };
} else if (nodeType.poll) {
if (mode === 'manual') {
// In manual mode run the poll function
const thisArgs = nodeExecuteFunctions.getExecutePollFunctions(
this,
node,
additionalData,
mode,
'manual',
);
return { data: await nodeType.poll.call(thisArgs) };
}
// In any other mode pass data through as it already contains the result of the poll
return { data: inputData.main as INodeExecutionData[][] };
} else if (nodeType.trigger) {
if (mode === 'manual') {
// In manual mode start the trigger
const triggerResponse = await this.runTrigger(
node,
nodeExecuteFunctions.getExecuteTriggerFunctions,
additionalData,
mode,
'manual',
);
if (triggerResponse === undefined) {
return { data: null };
}
let closeFunction;
if (triggerResponse.closeFunction) {
// In manual mode we return the trigger closeFunction. That allows it to be called directly
// but we do not have to wait for it to finish. That is important for things like queue-nodes.
// There the full close will may be delayed till a message gets acknowledged after the execution.
// If we would not be able to wait for it to close would it cause problems with "own" mode as the
// process would be killed directly after it and so the acknowledge would not have been finished yet.
closeFunction = triggerResponse.closeFunction;
// Manual testing of Trigger nodes creates an execution. If the execution is cancelled, `closeFunction` should be called to cleanup any open connections/consumers
abortSignal?.addEventListener('abort', closeFunction);
}
if (triggerResponse.manualTriggerFunction !== undefined) {
// If a manual trigger function is defined call it and wait till it did run
await triggerResponse.manualTriggerFunction();
}
const response = await triggerResponse.manualTriggerResponse!;
if (response.length === 0) {
return { data: null, closeFunction };
}
return { data: response, closeFunction };
}
// For trigger nodes in any mode except "manual" do we simply pass the data through
return { data: inputData.main as INodeExecutionData[][] };
} else if (nodeType.webhook) {
// For webhook nodes always simply pass the data through
return { data: inputData.main as INodeExecutionData[][] };
} else {
// For nodes which have routing information on properties
const routingNode = new RoutingNode(
this,
node,
connectionInputData,
runExecutionData ?? null,
additionalData,
mode,
);
return {
data: await routingNode.runNode(
inputData,
runIndex,
nodeType,
executionData,
nodeExecuteFunctions,
undefined,
abortSignal,
),
};
}
}
}
function hasDotNotationBannedChar(nodeName: string) {