Data pinning

packages/cli/src/Interfaces.ts

@@ -15,6 +15,7 @@ import {
 	ITelemetrySettings,
 	ITelemetryTrackProperties,
 	IWorkflowBase as IWorkflowBaseWorkflow,
+	PinData,
 	Workflow,
 	WorkflowExecuteMode,
 } from 'n8n-workflow';
@@ -646,6 +647,7 @@ export interface IWorkflowExecutionDataProcess {
 	executionMode: WorkflowExecuteMode;
 	executionData?: IRunExecutionData;
 	runData?: IRunData;
+	pinData?: PinData;
 	retryOf?: number | string;
 	sessionId?: string;
 	startNodes?: string[];

packages/cli/src/Server.ts

@@ -74,6 +74,7 @@ import {
 	IWorkflowBase,
 	LoggerProxy,
 	NodeHelpers,
+	PinData,
 	WebhookHttpMethod,
 	Workflow,
 	WorkflowExecuteMode,
@@ -129,6 +130,7 @@ import {
 	WorkflowRunner,
 	getCredentialForUser,
 	getCredentialWithoutUser,
+	IWorkflowDb,
 } from '.';
 
 import config from '../config';
@@ -156,16 +158,17 @@ import type {
 } from './requests';
 import { DEFAULT_EXECUTIONS_GET_ALL_LIMIT, validateEntity } from './GenericHelpers';
 import { ExecutionEntity } from './databases/entities/ExecutionEntity';
-import { SharedWorkflow } from './databases/entities/SharedWorkflow';
 import { AUTH_COOKIE_NAME, RESPONSE_ERROR_MESSAGES } from './constants';
 import { credentialsController } from './api/credentials.api';
+import { workflowsController } from './api/workflows.api';
 import { oauth2CredentialController } from './api/oauth2Credential.api';
 import {
 	getInstanceBaseUrl,
 	isEmailSetUp,
 	isUserManagementEnabled,
 } from './UserManagement/UserManagementHelper';
 import { loadPublicApiVersions } from './PublicApi';
+import { SharedWorkflow } from './databases/entities/SharedWorkflow';
 
 require('body-parser-xml')(bodyParser);
 
@@ -754,74 +757,6 @@ class App {
 		// Workflow
 		// ----------------------------------------
 
-		// Creates a new workflow
-		this.app.post(
-			`/${this.restEndpoint}/workflows`,
-			ResponseHelper.send(async (req: WorkflowRequest.Create) => {
-				delete req.body.id; // delete if sent
-
-				const newWorkflow = new WorkflowEntity();
-
-				Object.assign(newWorkflow, req.body);
-
-				await validateEntity(newWorkflow);
-
-				await this.externalHooks.run('workflow.create', [newWorkflow]);
-
-				const { tags: tagIds } = req.body;
-
-				if (tagIds?.length && !config.getEnv('workflowTagsDisabled')) {
-					newWorkflow.tags = await Db.collections.Tag.findByIds(tagIds, {
-						select: ['id', 'name'],
-					});
-				}
-
-				await WorkflowHelpers.replaceInvalidCredentials(newWorkflow);
-
-				let savedWorkflow: undefined | WorkflowEntity;
-
-				await getConnection().transaction(async (transactionManager) => {
-					savedWorkflow = await transactionManager.save<WorkflowEntity>(newWorkflow);
-
-					const role = await Db.collections.Role.findOneOrFail({
-						name: 'owner',
-						scope: 'workflow',
-					});
-
-					const newSharedWorkflow = new SharedWorkflow();
-
-					Object.assign(newSharedWorkflow, {
-						role,
-						user: req.user,
-						workflow: savedWorkflow,
-					});
-
-					await transactionManager.save<SharedWorkflow>(newSharedWorkflow);
-				});
-
-				if (!savedWorkflow) {
-					LoggerProxy.error('Failed to create workflow', { userId: req.user.id });
-					throw new ResponseHelper.ResponseError('Failed to save workflow');
-				}
-
-				if (tagIds && !config.getEnv('workflowTagsDisabled')) {
-					savedWorkflow.tags = TagHelpers.sortByRequestOrder(savedWorkflow.tags, {
-						requestOrder: tagIds,
-					});
-				}
-
-				await this.externalHooks.run('workflow.afterCreate', [savedWorkflow]);
-				void InternalHooksManager.getInstance().onWorkflowCreated(req.user.id, newWorkflow, false);
-
-				const { id, ...rest } = savedWorkflow;
-
-				return {
-					id: id.toString(),
-					...rest,
-				};
-			}),
-		);
-
 		// Reads and returns workflow data from an URL
 		this.app.get(
 			`/${this.restEndpoint}/workflows/from-url`,
@@ -947,50 +882,6 @@ class App {
 			}),
 		);
 
-		// Returns a specific workflow
-		this.app.get(
-			`/${this.restEndpoint}/workflows/:id`,
-			ResponseHelper.send(async (req: WorkflowRequest.Get) => {
-				const { id: workflowId } = req.params;
-
-				let relations = ['workflow', 'workflow.tags'];
-
-				if (config.getEnv('workflowTagsDisabled')) {
-					relations = relations.filter((relation) => relation !== 'workflow.tags');
-				}
-
-				const shared = await Db.collections.SharedWorkflow.findOne({
-					relations,
-					where: whereClause({
-						user: req.user,
-						entityType: 'workflow',
-						entityId: workflowId,
-					}),
-				});
-
-				if (!shared) {
-					LoggerProxy.info('User attempted to access a workflow without permissions', {
-						workflowId,
-						userId: req.user.id,
-					});
-					throw new ResponseHelper.ResponseError(
-						`Workflow with ID "${workflowId}" could not be found.`,
-						undefined,
-						404,
-					);
-				}
-
-				const {
-					workflow: { id, ...rest },
-				} = shared;
-
-				return {
-					id: id.toString(),
-					...rest,
-				};
-			}),
-		);
-
 		// Updates an existing workflow
 		this.app.patch(
 			`/${this.restEndpoint}/workflows/:id`,
@@ -1189,19 +1080,23 @@ 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';
 					const activationMode = 'manual';
 
 					const sessionId = GenericHelpers.getSessionId(req);
 
+					const pinnedTrigger = findFirstPinnedTrigger(workflowData, pinData);
+
 					// If webhooks nodes exist and are active we have to wait for till we receive a call
 					if (
-						runData === undefined ||
-						startNodes === undefined ||
-						startNodes.length === 0 ||
-						destinationNode === undefined
+						pinnedTrigger === undefined &&
+						(runData === undefined ||
+							startNodes === undefined ||
+							startNodes.length === 0 ||
+							destinationNode === undefined)
 					) {
 						const additionalData = await WorkflowExecuteAdditionalData.getBase(req.user.id);
 						const nodeTypes = NodeTypes();
@@ -1239,11 +1134,17 @@ class App {
 						destinationNode,
 						executionMode,
 						runData,
+						pinData,
 						sessionId,
 						startNodes,
 						workflowData,
 						userId: req.user.id,
 					};
+
+					if (pinnedTrigger) {
+						data.startNodes = [pinnedTrigger.name];
+					}
+
 					const workflowRunner = new WorkflowRunner();
 					const executionId = await workflowRunner.run(data);
 
@@ -1254,6 +1155,8 @@ class App {
 			),
 		);
 
+		this.app.use(`/${this.restEndpoint}/workflows`, workflowsController);
+
 		// Retrieves all tags, with or without usage count
 		this.app.get(
 			`/${this.restEndpoint}/tags`,
@@ -2912,3 +2815,20 @@ function isOAuth(credType: ICredentialType) {
 		)
 	);
 }
+
+const TRIGGER_NODE_SUFFIXES = ['trigger', 'webhook'];
+
+const isTrigger = (str: string) =>
+	TRIGGER_NODE_SUFFIXES.some((suffix) => str.toLowerCase().includes(suffix));
+
+function findFirstPinnedTrigger(workflow: IWorkflowDb, pinData?: PinData) {
+	if (!pinData) return;
+
+	const firstPinnedTriggerName = Object.keys(pinData).find(isTrigger);
+
+	if (!firstPinnedTriggerName) return;
+
+	return workflow.nodes.find(
+		({ type, name }) => isTrigger(type) && name === firstPinnedTriggerName,
+	);
+}

packages/cli/src/WebhookHelpers.ts

@@ -485,6 +485,10 @@ export async function executeWebhook(
 					return undefined;
 				}
 
+				if (workflowData.pinData) {
+					data.data.resultData.pinData = workflowData.pinData;
+				}
+
 				const returnData = WorkflowHelpers.getDataLastExecutedNodeData(data);
 				if (data.data.resultData.error || returnData?.error !== undefined) {
 					if (!didSendResponse) {

packages/cli/src/WorkflowHelpers.ts

@@ -50,7 +50,7 @@ const ERROR_TRIGGER_TYPE = config.getEnv('nodes.errorTriggerType');
  * @returns {(ITaskData | undefined)}
  */
 export function getDataLastExecutedNodeData(inputData: IRun): ITaskData | undefined {
-	const { runData } = inputData.data.resultData;
+	const { runData, pinData = {} } = inputData.data.resultData;
 	const { lastNodeExecuted } = inputData.data.resultData;
 
 	if (lastNodeExecuted === undefined) {
@@ -61,7 +61,26 @@ export function getDataLastExecutedNodeData(inputData: IRun): ITaskData | undefi
 		return undefined;
 	}
 
-	return runData[lastNodeExecuted][runData[lastNodeExecuted].length - 1];
+	const lastNodeRunData = runData[lastNodeExecuted][runData[lastNodeExecuted].length - 1];
+
+	let lastNodePinData = pinData[lastNodeExecuted];
+
+	if (lastNodePinData) {
+		if (!Array.isArray(lastNodePinData)) lastNodePinData = [lastNodePinData];
+
+		const itemsPerRun = lastNodePinData.map((item, index) => {
+			return { json: item, pairedItem: { item: index } };
+		});
+
+		return {
+			startTime: 0,
+			executionTime: 0,
+			data: { main: [itemsPerRun] },
+			source: lastNodeRunData.source,
+		};
+	}
+
+	return lastNodeRunData;
 }
 
 /**

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,
 				);
 			}
 

packages/cli/src/WorkflowRunnerProcess.ts

@@ -345,9 +345,22 @@ export class WorkflowRunnerProcess {
 		) {
 			// Execute all nodes
 
+			let startNode;
+			if (
+				this.data.startNodes?.length === 1 &&
+				Object.keys(this.data.pinData ?? {}).includes(this.data.startNodes[0])
+			) {
+				startNode = this.workflow.getNode(this.data.startNodes[0]) ?? undefined;
+			}
+
 			// 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,
+				startNode,
+				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 +369,7 @@ export class WorkflowRunnerProcess {
 			this.data.runData,
 			this.data.startNodes,
 			this.data.destinationNode,
+			this.data.pinData,
 		);
 	}