Inject pindata into items flow (#3420)

* ⚡ Inject pin data - Second approach * 🔥 Remove unneeded lint exception
n8n-io · Jun 29, 2022 · aa34753 · aa34753
1 parent 07759e0
commit aa34753
Show file tree

Hide file tree

Showing 9 changed files with 72 additions and 7 deletions.
diff --git a/packages/cli/src/Interfaces.ts b/packages/cli/src/Interfaces.ts
@@ -14,6 +14,7 @@ import {
 	ITaskData,
 	ITelemetrySettings,
 	IWorkflowBase as IWorkflowBaseWorkflow,
+	PinDataPayload,
 	Workflow,
 	WorkflowExecuteMode,
 } from 'n8n-workflow';
@@ -645,6 +646,7 @@ export interface IWorkflowExecutionDataProcess {
 	executionMode: WorkflowExecuteMode;
 	executionData?: IRunExecutionData;
 	runData?: IRunData;
+	pinData?: PinDataPayload;
 	retryOf?: number | string;
 	sessionId?: string;
 	startNodes?: string[];

diff --git a/packages/cli/src/Server.ts b/packages/cli/src/Server.ts
@@ -1144,6 +1144,7 @@ class App {
 				): Promise<IExecutionPushResponse> => {
 					const { workflowData } = req.body;
 					const { runData } = req.body;
+					const { pinData } = req.body;
 					const { startNodes } = req.body;
 					const { destinationNode } = req.body;
 					const executionMode = 'manual';
@@ -1194,6 +1195,7 @@ class App {
 						destinationNode,
 						executionMode,
 						runData,
+						pinData,
 						sessionId,
 						startNodes,
 						workflowData,

diff --git a/packages/cli/src/WorkflowRunner.ts b/packages/cli/src/WorkflowRunner.ts
@@ -310,7 +310,12 @@ export class WorkflowRunner {
 
 				// Can execute without webhook so go on
 				const workflowExecute = new WorkflowExecute(additionalData, data.executionMode);
-				workflowExecution = workflowExecute.run(workflow, undefined, data.destinationNode);
+				workflowExecution = workflowExecute.run(
+					workflow,
+					undefined,
+					data.destinationNode,
+					data.pinData,
+				);
 			} else {
 				Logger.debug(`Execution ID ${executionId} is a partial execution.`, { executionId });
 				// Execute only the nodes between start and destination nodes
@@ -320,6 +325,7 @@ export class WorkflowRunner {
 					data.runData,
 					data.startNodes,
 					data.destinationNode,
+					data.pinData,
 				);
 			}
 

diff --git a/packages/cli/src/WorkflowRunnerProcess.ts b/packages/cli/src/WorkflowRunnerProcess.ts
@@ -347,7 +347,12 @@ export class WorkflowRunnerProcess {
 
 			// Can execute without webhook so go on
 			this.workflowExecute = new WorkflowExecute(additionalData, this.data.executionMode);
-			return this.workflowExecute.run(this.workflow, undefined, this.data.destinationNode);
+			return this.workflowExecute.run(
+				this.workflow,
+				undefined,
+				this.data.destinationNode,
+				this.data.pinData,
+			);
 		}
 		// Execute only the nodes between start and destination nodes
 		this.workflowExecute = new WorkflowExecute(additionalData, this.data.executionMode);
@@ -356,6 +361,7 @@ export class WorkflowRunnerProcess {
 			this.data.runData,
 			this.data.startNodes,
 			this.data.destinationNode,
+			this.data.pinData,
 		);
 	}
 

diff --git a/packages/cli/src/requests.d.ts b/packages/cli/src/requests.d.ts
@@ -8,6 +8,7 @@ import {
 	INodeCredentialTestRequest,
 	IRunData,
 	IWorkflowSettings,
+	PinDataPayload,
 } from 'n8n-workflow';
 
 import { User } from './databases/entities/User';
@@ -71,6 +72,7 @@ export declare namespace WorkflowRequest {
 		{
 			workflowData: IWorkflowDb;
 			runData: IRunData;
+			pinData: PinDataPayload;
 			startNodes?: string[];
 			destinationNode?: string;
 		}

diff --git a/packages/core/src/WorkflowExecute.ts b/packages/core/src/WorkflowExecute.ts
@@ -32,6 +32,7 @@ import {
 	LoggerProxy as Logger,
 	NodeApiError,
 	NodeOperationError,
+	PinDataPayload,
 	Workflow,
 	WorkflowExecuteMode,
 	WorkflowOperationError,
@@ -59,6 +60,7 @@ export class WorkflowExecute {
 			startData: {},
 			resultData: {
 				runData: {},
+				pinData: {},
 			},
 			executionData: {
 				contextData: {},
@@ -82,7 +84,12 @@ export class WorkflowExecute {
 	//            PCancelable to a regular Promise and does so not allow canceling
 	//            active executions anymore
 	// eslint-disable-next-line @typescript-eslint/promise-function-async
-	run(workflow: Workflow, startNode?: INode, destinationNode?: string): PCancelable<IRun> {
+	run(
+		workflow: Workflow,
+		startNode?: INode,
+		destinationNode?: string,
+		pinData?: PinDataPayload,
+	): PCancelable<IRun> {
 		// Get the nodes to start workflow execution from
 		startNode = startNode || workflow.getStartNode(destinationNode);
 
@@ -121,6 +128,7 @@ export class WorkflowExecute {
 			},
 			resultData: {
 				runData: {},
+				pinData,
 			},
 			executionData: {
 				contextData: {},
@@ -152,6 +160,7 @@ export class WorkflowExecute {
 		runData: IRunData,
 		startNodes: string[],
 		destinationNode: string,
+		pinData?: PinDataPayload,
 		// @ts-ignore
 	): PCancelable<IRun> {
 		let incomingNodeConnections: INodeConnections | undefined;
@@ -258,6 +267,7 @@ export class WorkflowExecute {
 			},
 			resultData: {
 				runData,
+				pinData,
 			},
 			executionData: {
 				contextData: {},
@@ -927,11 +937,32 @@ export class WorkflowExecute {
 								this.mode,
 							);
 							nodeSuccessData = runNodeData.data;
+							const { pinData } = this.runExecutionData.resultData;
 
-							if (runNodeData.closeFunction) {
-								// Explanation why we do this can be found in n8n-workflow/Workflow.ts -> runNode
-								// eslint-disable-next-line @typescript-eslint/no-unsafe-call
-								closeFunction = runNodeData.closeFunction();
+							if (pinData && pinData[executionNode.name] !== undefined) {
+								const nodePinData = pinData[executionNode.name][runIndex];
+								nodeSuccessData = [[{ json: nodePinData }]];
+							} else {
+								Logger.debug(`Running node "${executionNode.name}" started`, {
+									node: executionNode.name,
+									workflowId: workflow.id,
+								});
+								const runNodeData = await workflow.runNode(
+									executionData.node,
+									executionData.data,
+									this.runExecutionData,
+									runIndex,
+									this.additionalData,
+									NodeExecuteFunctions,
+									this.mode,
+								);
+								nodeSuccessData = runNodeData.data;
+
+								if (runNodeData.closeFunction) {
+									// Explanation why we do this can be found in n8n-workflow/Workflow.ts -> runNode
+									// eslint-disable-next-line @typescript-eslint/no-unsafe-call
+									closeFunction = runNodeData.closeFunction();
+								}
 							}
 
 							Logger.debug(`Running node "${executionNode.name}" finished successfully`, {

diff --git a/packages/editor-ui/src/Interface.ts b/packages/editor-ui/src/Interface.ts
@@ -21,6 +21,7 @@ import {
 	ITelemetrySettings,
 	IWorkflowSettings as IWorkflowSettingsWorkflow,
 	WorkflowExecuteMode,
+	PinDataPayload,
 } from 'n8n-workflow';
 
 export * from 'n8n-design-system/src/types';
@@ -211,6 +212,7 @@ export interface IStartRunData {
 	startNodes?: string[];
 	destinationNode?: string;
 	runData?: IRunData;
+	pinData?: PinDataPayload;
 }
 
 export interface IRunDataUi {

diff --git a/packages/editor-ui/src/components/mixins/workflowRun.ts b/packages/editor-ui/src/components/mixins/workflowRun.ts
@@ -8,6 +8,7 @@ import {
 	IRunData,
 	IRunExecutionData,
 	NodeHelpers,
+	PinDataPayload,
 } from 'n8n-workflow';
 
 import { externalHooks } from '@/components/mixins/externalHooks';
@@ -151,9 +152,16 @@ export const workflowRun = mixins(
 
 				const workflowData = await this.getWorkflowDataToSave();
 
+				const pinData = Object.values(workflow.nodes).reduce<PinDataPayload>((acc, node) => {
+					if (node.pinData) acc[node.name] = [node.pinData];
+
+					return acc;
+				}, {});
+
 				const startRunData: IStartRunData = {
 					workflowData,
 					runData: newRunData,
+					pinData,
 					startNodes,
 				};
 				if (nodeName) {
@@ -174,6 +182,7 @@ export const workflowRun = mixins(
 					data: {
 						resultData: {
 							runData: newRunData || {},
+							pinData,
 							startNodes,
 							workflowData,
 						},

diff --git a/packages/workflow/src/Interfaces.ts b/packages/workflow/src/Interfaces.ts
@@ -795,6 +795,10 @@ export interface INode {
 	pinData?: IDataObject;
 }
 
+export interface PinDataPayload {
+	[nodeName: string]: IDataObject[];
+}
+
 export interface INodes {
 	[key: string]: INode;
 }
@@ -1293,6 +1297,7 @@ export interface IRunExecutionData {
 	resultData: {
 		error?: ExecutionError;
 		runData: IRunData;
+		pinData?: PinDataPayload;
 		lastNodeExecuted?: string;
 	};
 	executionData?: {