n8n/packages/cli/src/WorkflowHelpers.ts

302 lines
7.2 KiB
TypeScript
Raw Normal View History

2019-06-23 03:35:23 -07:00
import {
Db,
2019-12-19 14:07:55 -08:00
ITransferNodeTypes,
IWorkflowExecutionDataProcess,
2019-06-23 03:35:23 -07:00
IWorkflowErrorData,
NodeTypes,
WorkflowCredentials,
WorkflowRunner,
2019-06-23 03:35:23 -07:00
} from './';
import {
IDataObject,
2019-06-23 03:35:23 -07:00
IExecuteData,
INode,
2019-12-19 14:07:55 -08:00
IRun,
2019-06-23 03:35:23 -07:00
IRunExecutionData,
2019-12-19 14:07:55 -08:00
ITaskData,
2019-06-23 03:35:23 -07:00
Workflow,
} from 'n8n-workflow';
import * as config from '../config';
2019-06-23 03:35:23 -07:00
const ERROR_TRIGGER_TYPE = config.get('nodes.errorTriggerType') as string;
2019-12-19 14:07:55 -08:00
/**
* Returns the data of the last executed node
*
* @export
* @param {IRun} inputData
* @returns {(ITaskData | undefined)}
*/
export function getDataLastExecutedNodeData(inputData: IRun): ITaskData | undefined {
const runData = inputData.data.resultData.runData;
const lastNodeExecuted = inputData.data.resultData.lastNodeExecuted;
if (lastNodeExecuted === undefined) {
return undefined;
}
if (runData[lastNodeExecuted] === undefined) {
return undefined;
}
return runData[lastNodeExecuted][runData[lastNodeExecuted].length - 1];
}
2019-06-23 03:35:23 -07:00
/**
* Returns if the given id is a valid workflow id
*
* @param {(string | null | undefined)} id The id to check
* @returns {boolean}
* @memberof App
*/
export function isWorkflowIdValid (id: string | null | undefined | number): boolean {
if (typeof id === 'string') {
id = parseInt(id, 10);
}
if (isNaN(id as number)) {
return false;
}
return true;
}
/**
* Executes the error workflow
*
* @export
* @param {string} workflowId The id of the error workflow
* @param {IWorkflowErrorData} workflowErrorData The error data
* @returns {Promise<void>}
*/
export async function executeErrorWorkflow(workflowId: string, workflowErrorData: IWorkflowErrorData): Promise<void> {
// Wrap everything in try/catch to make sure that no errors bubble up and all get caught here
try {
const workflowData = await Db.collections.Workflow!.findOne({ id: workflowId });
if (workflowData === undefined) {
// The error workflow could not be found
console.error(`ERROR: Calling Error Workflow for "${workflowErrorData.workflow.id}". Could not find error workflow "${workflowId}"`);
return;
}
const executionMode = 'error';
const nodeTypes = NodeTypes();
const workflowInstance = new Workflow(workflowId, workflowData.nodes, workflowData.connections, workflowData.active, nodeTypes, workflowData.staticData, workflowData.settings);
2019-06-23 03:35:23 -07:00
let node: INode;
let workflowStartNode: INode | undefined;
for (const nodeName of Object.keys(workflowInstance.nodes)) {
node = workflowInstance.nodes[nodeName];
if (node.type === ERROR_TRIGGER_TYPE) {
workflowStartNode = node;
}
}
if (workflowStartNode === undefined) {
console.error(`ERROR: Calling Error Workflow for "${workflowErrorData.workflow.id}". Could not find "${ERROR_TRIGGER_TYPE}" in workflow "${workflowId}"`);
return;
}
// Can execute without webhook so go on
// Initialize the data of the webhook node
const nodeExecutionStack: IExecuteData[] = [];
nodeExecutionStack.push(
{
node: workflowStartNode,
data: {
main: [
[
{
json: workflowErrorData
}
]
],
},
},
);
const runExecutionData: IRunExecutionData = {
startData: {
},
resultData: {
runData: {},
},
executionData: {
contextData: {},
nodeExecutionStack,
waitingExecution: {},
},
};
const credentials = await WorkflowCredentials(workflowData.nodes);
2019-06-23 03:35:23 -07:00
const runData: IWorkflowExecutionDataProcess = {
credentials,
executionMode,
executionData: runExecutionData,
workflowData,
};
const workflowRunner = new WorkflowRunner();
await workflowRunner.run(runData);
2019-06-23 03:35:23 -07:00
} catch (error) {
console.error(`ERROR: Calling Error Workflow for "${workflowErrorData.workflow.id}": ${error.message}`);
}
}
2019-12-19 14:07:55 -08:00
/**
* Returns all the defined NodeTypes
*
* @export
* @returns {ITransferNodeTypes}
*/
export function getAllNodeTypeData(): ITransferNodeTypes {
const nodeTypes = NodeTypes();
// Get the data of all thenode types that they
// can be loaded again in the process
const returnData: ITransferNodeTypes = {};
for (const nodeTypeName of Object.keys(nodeTypes.nodeTypes)) {
if (nodeTypes.nodeTypes[nodeTypeName] === undefined) {
throw new Error(`The NodeType "${nodeTypeName}" could not be found!`);
}
returnData[nodeTypeName] = {
className: nodeTypes.nodeTypes[nodeTypeName].type.constructor.name,
sourcePath: nodeTypes.nodeTypes[nodeTypeName].sourcePath,
};
}
return returnData;
}
/**
* Returns the data of the node types that are needed
* to execute the given nodes
*
* @export
* @param {INode[]} nodes
* @returns {ITransferNodeTypes}
*/
export function getNodeTypeData(nodes: INode[]): ITransferNodeTypes {
const nodeTypes = NodeTypes();
// Check which node-types have to be loaded
const neededNodeTypes = getNeededNodeTypes(nodes);
// Get all the data of the needed node types that they
// can be loaded again in the process
const returnData: ITransferNodeTypes = {};
for (const nodeTypeName of neededNodeTypes) {
if (nodeTypes.nodeTypes[nodeTypeName] === undefined) {
throw new Error(`The NodeType "${nodeTypeName}" could not be found!`);
}
returnData[nodeTypeName] = {
className: nodeTypes.nodeTypes[nodeTypeName].type.constructor.name,
sourcePath: nodeTypes.nodeTypes[nodeTypeName].sourcePath,
};
}
return returnData;
}
/**
* Returns the names of the NodeTypes which are are needed
* to execute the gives nodes
*
* @export
* @param {INode[]} nodes
* @returns {string[]}
*/
export function getNeededNodeTypes(nodes: INode[]): string[] {
// Check which node-types have to be loaded
const neededNodeTypes: string[] = [];
for (const node of nodes) {
if (!neededNodeTypes.includes(node.type)) {
neededNodeTypes.push(node.type);
}
}
return neededNodeTypes;
}
2019-06-23 03:35:23 -07:00
/**
* Saves the static data if it changed
*
* @export
* @param {Workflow} workflow
* @returns {Promise <void>}
*/
export async function saveStaticData(workflow: Workflow): Promise <void> {
if (workflow.staticData.__dataChanged === true) {
// Static data of workflow changed and so has to be saved
if (isWorkflowIdValid(workflow.id) === true) {
// Workflow is saved so update in database
try {
await saveStaticDataById(workflow.id!, workflow.staticData);
2019-06-23 03:35:23 -07:00
workflow.staticData.__dataChanged = false;
} catch (e) {
// TODO: Add proper logging!
console.error(`There was a problem saving the workflow with id "${workflow.id}" to save changed staticData: ${e.message}`);
}
}
}
}
/**
* Saves the given static data on workflow
*
* @export
* @param {(string | number)} workflowId The id of the workflow to save data on
* @param {IDataObject} newStaticData The static data to save
* @returns {Promise<void>}
*/
export async function saveStaticDataById(workflowId: string | number, newStaticData: IDataObject): Promise<void> {
await Db.collections.Workflow!
.update(workflowId, {
staticData: newStaticData,
});
}
/**
* Returns the static data of workflow
*
* @export
* @param {(string | number)} workflowId The id of the workflow to get static data of
* @returns
*/
export async function getStaticDataById(workflowId: string | number) {
const workflowData = await Db.collections.Workflow!
.findOne(workflowId, { select: ['staticData']});
if (workflowData === undefined) {
return {};
}
return workflowData.staticData || {};
}