diff --git a/.npmrc b/.npmrc new file mode 100644 index 000000000..e9ee3cb4d --- /dev/null +++ b/.npmrc @@ -0,0 +1 @@ +legacy-peer-deps=true \ No newline at end of file diff --git a/docker/images/n8n-custom/docker-entrypoint.sh b/docker/images/n8n-custom/docker-entrypoint.sh index 2dd4dae10..acd6a6019 100755 --- a/docker/images/n8n-custom/docker-entrypoint.sh +++ b/docker/images/n8n-custom/docker-entrypoint.sh @@ -6,6 +6,8 @@ if [ -d /root/.n8n ] ; then ln -s /root/.n8n /home/node/ fi +chown -R node /home/node + if [ "$#" -gt 0 ]; then # Got started with arguments COMMAND=$1; diff --git a/packages/cli/commands/execute.ts b/packages/cli/commands/execute.ts index b6641628d..b74eb7397 100644 --- a/packages/cli/commands/execute.ts +++ b/packages/cli/commands/execute.ts @@ -11,6 +11,7 @@ import { CredentialTypes, Db, ExternalHooks, + InternalHooksManager, IWorkflowBase, IWorkflowExecutionDataProcess, LoadNodesAndCredentials, @@ -123,6 +124,9 @@ export class Execute extends Command { const externalHooks = ExternalHooks(); await externalHooks.init(); + const instanceId = await UserSettings.getInstanceId(); + InternalHooksManager.init(instanceId); + // Add the found types to an instance other parts of the application can use const nodeTypes = NodeTypes(); await nodeTypes.init(loadNodesAndCredentials.nodeTypes); diff --git a/packages/cli/commands/executeBatch.ts b/packages/cli/commands/executeBatch.ts index d4489c38d..587f91c2a 100644 --- a/packages/cli/commands/executeBatch.ts +++ b/packages/cli/commands/executeBatch.ts @@ -28,6 +28,7 @@ import { CredentialTypes, Db, ExternalHooks, + InternalHooksManager, IWorkflowDb, IWorkflowExecutionDataProcess, LoadNodesAndCredentials, @@ -55,12 +56,12 @@ export class ExecuteBatch extends Command { static executionTimeout = 3 * 60 * 1000; static examples = [ - `$ n8n executeAll`, - `$ n8n executeAll --concurrency=10 --skipList=/data/skipList.txt`, - `$ n8n executeAll --debug --output=/data/output.json`, - `$ n8n executeAll --ids=10,13,15 --shortOutput`, - `$ n8n executeAll --snapshot=/data/snapshots --shallow`, - `$ n8n executeAll --compare=/data/previousExecutionData --retries=2`, + `$ n8n executeBatch`, + `$ n8n executeBatch --concurrency=10 --skipList=/data/skipList.txt`, + `$ n8n executeBatch --debug --output=/data/output.json`, + `$ n8n executeBatch --ids=10,13,15 --shortOutput`, + `$ n8n executeBatch --snapshot=/data/snapshots --shallow`, + `$ n8n executeBatch --compare=/data/previousExecutionData --retries=2`, ]; static flags = { @@ -303,6 +304,9 @@ export class ExecuteBatch extends Command { const externalHooks = ExternalHooks(); await externalHooks.init(); + const instanceId = await UserSettings.getInstanceId(); + InternalHooksManager.init(instanceId); + // Add the found types to an instance other parts of the application can use const nodeTypes = NodeTypes(); await nodeTypes.init(loadNodesAndCredentials.nodeTypes); @@ -813,10 +817,22 @@ export class ExecuteBatch extends Command { const changes = diff(JSON.parse(contents), data, { keysOnly: true }); if (changes !== undefined) { - // we have structural changes. Report them. - executionResult.error = `Workflow may contain breaking changes`; - executionResult.changes = changes; - executionResult.executionStatus = 'error'; + // If we had only additions with no removals + // Then we treat as a warning and not an error. + // To find this, we convert the object to JSON + // and search for the `__deleted` string + const changesJson = JSON.stringify(changes); + if (changesJson.includes('__deleted')) { + // we have structural changes. Report them. + executionResult.error = 'Workflow may contain breaking changes'; + executionResult.changes = changes; + executionResult.executionStatus = 'error'; + } else { + executionResult.error = + 'Workflow contains new data that previously did not exist.'; + executionResult.changes = changes; + executionResult.executionStatus = 'warning'; + } } else { executionResult.executionStatus = 'success'; } diff --git a/packages/cli/commands/worker.ts b/packages/cli/commands/worker.ts index 9a06868f3..28a02b9b3 100644 --- a/packages/cli/commands/worker.ts +++ b/packages/cli/commands/worker.ts @@ -12,7 +12,7 @@ import * as PCancelable from 'p-cancelable'; import { Command, flags } from '@oclif/command'; import { UserSettings, WorkflowExecute } from 'n8n-core'; -import { INodeTypes, IRun, Workflow, LoggerProxy } from 'n8n-workflow'; +import { IExecuteResponsePromiseData, INodeTypes, IRun, Workflow, LoggerProxy } from 'n8n-workflow'; import { FindOneOptions } from 'typeorm'; @@ -25,11 +25,13 @@ import { GenericHelpers, IBullJobData, IBullJobResponse, + IBullWebhookResponse, IExecutionFlattedDb, InternalHooksManager, LoadNodesAndCredentials, NodeTypes, ResponseHelper, + WebhookHelpers, WorkflowExecuteAdditionalData, } from '../src'; @@ -172,6 +174,16 @@ export class Worker extends Command { currentExecutionDb.workflowData, { retryOf: currentExecutionDb.retryOf as string }, ); + + additionalData.hooks.hookFunctions.sendResponse = [ + async (response: IExecuteResponsePromiseData): Promise => { + await job.progress({ + executionId: job.data.executionId as string, + response: WebhookHelpers.encodeWebhookResponse(response), + } as IBullWebhookResponse); + }, + ]; + additionalData.executionId = jobData.executionId; let workflowExecute: WorkflowExecute; diff --git a/packages/cli/package.json b/packages/cli/package.json index 2ab8191e7..8dc2ef1da 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -1,6 +1,6 @@ { "name": "n8n", - "version": "0.145.0", + "version": "0.149.0", "description": "n8n Workflow Automation Tool", "license": "SEE LICENSE IN LICENSE.md", "homepage": "https://n8n.io", @@ -83,7 +83,7 @@ "dependencies": { "@oclif/command": "^1.5.18", "@oclif/errors": "^1.2.2", - "@rudderstack/rudder-sdk-node": "^1.0.2", + "@rudderstack/rudder-sdk-node": "1.0.6", "@types/json-diff": "^0.5.1", "@types/jsonwebtoken": "^8.5.2", "basic-auth": "^2.0.1", @@ -110,10 +110,10 @@ "localtunnel": "^2.0.0", "lodash.get": "^4.4.2", "mysql2": "~2.3.0", - "n8n-core": "~0.90.0", - "n8n-editor-ui": "~0.113.0", - "n8n-nodes-base": "~0.142.0", - "n8n-workflow": "~0.73.0", + "n8n-core": "~0.93.0", + "n8n-editor-ui": "~0.116.0", + "n8n-nodes-base": "~0.146.0", + "n8n-workflow": "~0.76.0", "oauth-1.0a": "^2.2.6", "open": "^7.0.0", "pg": "^8.3.0", diff --git a/packages/cli/src/ActiveExecutions.ts b/packages/cli/src/ActiveExecutions.ts index dac67322c..cd02ebe67 100644 --- a/packages/cli/src/ActiveExecutions.ts +++ b/packages/cli/src/ActiveExecutions.ts @@ -5,9 +5,12 @@ /* eslint-disable @typescript-eslint/no-unsafe-call */ /* eslint-disable @typescript-eslint/no-non-null-assertion */ /* eslint-disable @typescript-eslint/no-unsafe-assignment */ -import { IRun } from 'n8n-workflow'; - -import { createDeferredPromise } from 'n8n-core'; +import { + createDeferredPromise, + IDeferredPromise, + IExecuteResponsePromiseData, + IRun, +} from 'n8n-workflow'; import { ChildProcess } from 'child_process'; // eslint-disable-next-line import/no-extraneous-dependencies @@ -116,6 +119,28 @@ export class ActiveExecutions { this.activeExecutions[executionId].workflowExecution = workflowExecution; } + attachResponsePromise( + executionId: string, + responsePromise: IDeferredPromise, + ): void { + if (this.activeExecutions[executionId] === undefined) { + throw new Error( + `No active execution with id "${executionId}" got found to attach to workflowExecution to!`, + ); + } + + this.activeExecutions[executionId].responsePromise = responsePromise; + } + + resolveResponsePromise(executionId: string, response: IExecuteResponsePromiseData): void { + if (this.activeExecutions[executionId] === undefined) { + return; + } + + // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access + this.activeExecutions[executionId].responsePromise?.resolve(response); + } + /** * Remove an active execution * @@ -193,6 +218,7 @@ export class ActiveExecutions { this.activeExecutions[executionId].postExecutePromises.push(waitPromise); + // eslint-disable-next-line @typescript-eslint/no-unsafe-return, @typescript-eslint/no-unsafe-member-access return waitPromise.promise(); } diff --git a/packages/cli/src/ActiveWorkflowRunner.ts b/packages/cli/src/ActiveWorkflowRunner.ts index 181671c4f..dd8ac09c3 100644 --- a/packages/cli/src/ActiveWorkflowRunner.ts +++ b/packages/cli/src/ActiveWorkflowRunner.ts @@ -12,7 +12,9 @@ import { ActiveWorkflows, NodeExecuteFunctions } from 'n8n-core'; import { + IDeferredPromise, IExecuteData, + IExecuteResponsePromiseData, IGetExecutePollFunctions, IGetExecuteTriggerFunctions, INode, @@ -40,8 +42,6 @@ import { NodeTypes, ResponseHelper, WebhookHelpers, - // eslint-disable-next-line @typescript-eslint/no-unused-vars - WorkflowCredentials, WorkflowExecuteAdditionalData, WorkflowHelpers, WorkflowRunner, @@ -550,6 +550,7 @@ export class ActiveWorkflowRunner { data: INodeExecutionData[][], additionalData: IWorkflowExecuteAdditionalDataWorkflow, mode: WorkflowExecuteMode, + responsePromise?: IDeferredPromise, ) { const nodeExecutionStack: IExecuteData[] = [ { @@ -580,7 +581,7 @@ export class ActiveWorkflowRunner { }; const workflowRunner = new WorkflowRunner(); - return workflowRunner.run(runData, true); + return workflowRunner.run(runData, true, undefined, undefined, responsePromise); } /** @@ -641,13 +642,16 @@ export class ActiveWorkflowRunner { mode, activation, ); - returnFunctions.emit = (data: INodeExecutionData[][]): void => { + returnFunctions.emit = ( + data: INodeExecutionData[][], + responsePromise?: IDeferredPromise, + ): void => { // eslint-disable-next-line @typescript-eslint/restrict-template-expressions Logger.debug(`Received trigger for workflow "${workflow.name}"`); WorkflowHelpers.saveStaticData(workflow); // eslint-disable-next-line id-denylist - this.runWorkflow(workflowData, node, data, additionalData, mode).catch((err) => - console.error(err), + this.runWorkflow(workflowData, node, data, additionalData, mode, responsePromise).catch( + (error) => console.error(error), ); }; return returnFunctions; diff --git a/packages/cli/src/Interfaces.ts b/packages/cli/src/Interfaces.ts index d5c11a8f3..556aa7449 100644 --- a/packages/cli/src/Interfaces.ts +++ b/packages/cli/src/Interfaces.ts @@ -7,19 +7,19 @@ import { ICredentialsEncrypted, ICredentialType, IDataObject, + IDeferredPromise, + IExecuteResponsePromiseData, IRun, IRunData, IRunExecutionData, ITaskData, ITelemetrySettings, IWorkflowBase as IWorkflowBaseWorkflow, - // eslint-disable-next-line @typescript-eslint/no-unused-vars - IWorkflowCredentials, Workflow, WorkflowExecuteMode, } from 'n8n-workflow'; -import { IDeferredPromise, WorkflowExecute } from 'n8n-core'; +import { WorkflowExecute } from 'n8n-core'; // eslint-disable-next-line import/no-extraneous-dependencies import * as PCancelable from 'p-cancelable'; @@ -47,6 +47,11 @@ export interface IBullJobResponse { success: boolean; } +export interface IBullWebhookResponse { + executionId: string; + response: IExecuteResponsePromiseData; +} + export interface ICustomRequest extends Request { parsedUrl: Url | undefined; } @@ -237,6 +242,7 @@ export interface IExecutingWorkflowData { process?: ChildProcess; startedAt: Date; postExecutePromises: Array>; + responsePromise?: IDeferredPromise; workflowExecution?: PCancelable; } @@ -490,6 +496,7 @@ export interface IPushDataConsoleMessage { export interface IResponseCallbackData { data?: IDataObject | IDataObject[]; + headers?: object; noWebhookResponse?: boolean; responseCode?: number; } diff --git a/packages/cli/src/NodeTypes.ts b/packages/cli/src/NodeTypes.ts index b6ed97511..ff4e8c027 100644 --- a/packages/cli/src/NodeTypes.ts +++ b/packages/cli/src/NodeTypes.ts @@ -40,6 +40,9 @@ class NodeTypesClass implements INodeTypes { } getByNameAndVersion(nodeType: string, version?: number): INodeType { + if (this.nodeTypes[nodeType] === undefined) { + throw new Error(`The node-type "${nodeType}" is not known!`); + } return NodeHelpers.getVersionedTypeNode(this.nodeTypes[nodeType].type, version); } } diff --git a/packages/cli/src/Queue.ts b/packages/cli/src/Queue.ts index 9143c59ee..5d215a2bd 100644 --- a/packages/cli/src/Queue.ts +++ b/packages/cli/src/Queue.ts @@ -1,12 +1,21 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access */ import * as Bull from 'bull'; import * as config from '../config'; // eslint-disable-next-line import/no-cycle -import { IBullJobData } from './Interfaces'; +import { IBullJobData, IBullWebhookResponse } from './Interfaces'; +// eslint-disable-next-line import/no-cycle +import * as ActiveExecutions from './ActiveExecutions'; +// eslint-disable-next-line import/no-cycle +import * as WebhookHelpers from './WebhookHelpers'; export class Queue { + private activeExecutions: ActiveExecutions.ActiveExecutions; + private jobQueue: Bull.Queue; constructor() { + this.activeExecutions = ActiveExecutions.getInstance(); + const prefix = config.get('queue.bull.prefix') as string; const redisOptions = config.get('queue.bull.redis') as object; // Disabling ready check is necessary as it allows worker to @@ -16,6 +25,14 @@ export class Queue { // More here: https://github.com/OptimalBits/bull/issues/890 // @ts-ignore this.jobQueue = new Bull('jobs', { prefix, redis: redisOptions, enableReadyCheck: false }); + + this.jobQueue.on('global:progress', (jobId, progress: IBullWebhookResponse) => { + this.activeExecutions.resolveResponsePromise( + // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access + progress.executionId, + WebhookHelpers.decodeWebhookResponse(progress.response), + ); + }); } async add(jobData: IBullJobData, jobOptions: object): Promise { diff --git a/packages/cli/src/ResponseHelper.ts b/packages/cli/src/ResponseHelper.ts index f6deb551b..e8430c695 100644 --- a/packages/cli/src/ResponseHelper.ts +++ b/packages/cli/src/ResponseHelper.ts @@ -72,11 +72,16 @@ export function sendSuccessResponse( data: any, raw?: boolean, responseCode?: number, + responseHeader?: object, ) { if (responseCode !== undefined) { res.status(responseCode); } + if (responseHeader) { + res.header(responseHeader); + } + if (raw === true) { if (typeof data === 'string') { res.send(data); diff --git a/packages/cli/src/Server.ts b/packages/cli/src/Server.ts index e9de494f4..9cf50c40b 100644 --- a/packages/cli/src/Server.ts +++ b/packages/cli/src/Server.ts @@ -679,6 +679,7 @@ class App { // @ts-ignore savedWorkflow.id = savedWorkflow.id.toString(); + await this.externalHooks.run('workflow.afterCreate', [savedWorkflow]); void InternalHooksManager.getInstance().onWorkflowCreated(newWorkflow as IWorkflowBase); return savedWorkflow; }, @@ -1579,11 +1580,11 @@ class App { const findQuery = {} as FindManyOptions; if (req.query.filter) { findQuery.where = JSON.parse(req.query.filter as string); - if ((findQuery.where! as IDataObject).id !== undefined) { + if (findQuery.where.id !== undefined) { // No idea if multiple where parameters make db search // slower but to be sure that that is not the case we // remove all unnecessary fields in case the id is defined. - findQuery.where = { id: (findQuery.where! as IDataObject).id }; + findQuery.where = { id: findQuery.where.id }; } } @@ -2668,7 +2669,13 @@ class App { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); @@ -2719,7 +2726,13 @@ class App { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); @@ -2745,7 +2758,13 @@ class App { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); diff --git a/packages/cli/src/WebhookHelpers.ts b/packages/cli/src/WebhookHelpers.ts index ef0c47ac2..203bf20b0 100644 --- a/packages/cli/src/WebhookHelpers.ts +++ b/packages/cli/src/WebhookHelpers.ts @@ -1,3 +1,4 @@ +/* eslint-disable @typescript-eslint/no-unsafe-call */ /* eslint-disable no-param-reassign */ /* eslint-disable @typescript-eslint/prefer-optional-chain */ /* eslint-disable @typescript-eslint/no-shadow */ @@ -18,9 +19,13 @@ import { get } from 'lodash'; import { BINARY_ENCODING, NodeExecuteFunctions } from 'n8n-core'; import { + createDeferredPromise, IBinaryKeyData, IDataObject, + IDeferredPromise, IExecuteData, + IExecuteResponsePromiseData, + IN8nHttpFullResponse, INode, IRunExecutionData, IWebhookData, @@ -34,20 +39,20 @@ import { } from 'n8n-workflow'; // eslint-disable-next-line import/no-cycle import { - ActiveExecutions, GenericHelpers, IExecutionDb, IResponseCallbackData, IWorkflowDb, IWorkflowExecutionDataProcess, ResponseHelper, - // eslint-disable-next-line @typescript-eslint/no-unused-vars - WorkflowCredentials, WorkflowExecuteAdditionalData, WorkflowHelpers, WorkflowRunner, } from '.'; +// eslint-disable-next-line import/no-cycle +import * as ActiveExecutions from './ActiveExecutions'; + const activeExecutions = ActiveExecutions.getInstance(); /** @@ -91,6 +96,35 @@ export function getWorkflowWebhooks( return returnData; } +export function decodeWebhookResponse( + response: IExecuteResponsePromiseData, +): IExecuteResponsePromiseData { + if ( + typeof response === 'object' && + typeof response.body === 'object' && + (response.body as IDataObject)['__@N8nEncodedBuffer@__'] + ) { + response.body = Buffer.from( + (response.body as IDataObject)['__@N8nEncodedBuffer@__'] as string, + BINARY_ENCODING, + ); + } + + return response; +} + +export function encodeWebhookResponse( + response: IExecuteResponsePromiseData, +): IExecuteResponsePromiseData { + if (typeof response === 'object' && Buffer.isBuffer(response.body)) { + response.body = { + '__@N8nEncodedBuffer@__': response.body.toString(BINARY_ENCODING), + }; + } + + return response; +} + /** * Returns all the webhooks which should be created for the give workflow * @@ -169,7 +203,7 @@ export async function executeWebhook( 200, ) as number; - if (!['onReceived', 'lastNode'].includes(responseMode as string)) { + if (!['onReceived', 'lastNode', 'responseNode'].includes(responseMode as string)) { // If the mode is not known we error. Is probably best like that instead of using // the default that people know as early as possible (probably already testing phase) // that something does not resolve properly. @@ -356,9 +390,52 @@ export async function executeWebhook( workflowData, }; + let responsePromise: IDeferredPromise | undefined; + if (responseMode === 'responseNode') { + responsePromise = await createDeferredPromise(); + responsePromise + .promise() + .then((response: IN8nHttpFullResponse) => { + if (didSendResponse) { + return; + } + + if (Buffer.isBuffer(response.body)) { + res.header(response.headers); + res.end(response.body); + + responseCallback(null, { + noWebhookResponse: true, + }); + } else { + // TODO: This probably needs some more changes depending on the options on the + // Webhook Response node + responseCallback(null, { + data: response.body as IDataObject, + headers: response.headers, + responseCode: response.statusCode, + }); + } + + didSendResponse = true; + }) + .catch(async (error) => { + Logger.error( + `Error with Webhook-Response for execution "${executionId}": "${error.message}"`, + { executionId, workflowId: workflow.id }, + ); + }); + } + // Start now to run the workflow const workflowRunner = new WorkflowRunner(); - executionId = await workflowRunner.run(runData, true, !didSendResponse, executionId); + executionId = await workflowRunner.run( + runData, + true, + !didSendResponse, + executionId, + responsePromise, + ); Logger.verbose( `Started execution of workflow "${workflow.name}" from webhook with execution ID ${executionId}`, @@ -398,6 +475,20 @@ export async function executeWebhook( return data; } + if (responseMode === 'responseNode') { + if (!didSendResponse) { + // Return an error if no Webhook-Response node did send any data + responseCallback(null, { + data: { + message: 'Workflow executed sucessfully.', + }, + responseCode, + }); + didSendResponse = true; + } + return undefined; + } + if (returnData === undefined) { if (!didSendResponse) { responseCallback(null, { diff --git a/packages/cli/src/WebhookServer.ts b/packages/cli/src/WebhookServer.ts index 4cf3afc7b..c63526bfc 100644 --- a/packages/cli/src/WebhookServer.ts +++ b/packages/cli/src/WebhookServer.ts @@ -64,7 +64,13 @@ export function registerProductionWebhooks() { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); @@ -115,7 +121,13 @@ export function registerProductionWebhooks() { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); @@ -141,7 +153,13 @@ export function registerProductionWebhooks() { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); @@ -173,7 +191,13 @@ export function registerProductionWebhooks() { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); @@ -199,7 +223,13 @@ export function registerProductionWebhooks() { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); @@ -225,7 +255,13 @@ export function registerProductionWebhooks() { return; } - ResponseHelper.sendSuccessResponse(res, response.data, true, response.responseCode); + ResponseHelper.sendSuccessResponse( + res, + response.data, + true, + response.responseCode, + response.headers, + ); }, ); } diff --git a/packages/cli/src/WorkflowRunner.ts b/packages/cli/src/WorkflowRunner.ts index 8984384aa..fd18ff3d0 100644 --- a/packages/cli/src/WorkflowRunner.ts +++ b/packages/cli/src/WorkflowRunner.ts @@ -15,6 +15,8 @@ import { IProcessMessage, WorkflowExecute } from 'n8n-core'; import { ExecutionError, + IDeferredPromise, + IExecuteResponsePromiseData, IRun, LoggerProxy as Logger, Workflow, @@ -41,9 +43,7 @@ import { IBullJobResponse, ICredentialsOverwrite, ICredentialsTypeData, - IExecutionDb, IExecutionFlattedDb, - IExecutionResponse, IProcessMessageDataHook, ITransferNodeTypes, IWorkflowExecutionDataProcess, @@ -51,6 +51,7 @@ import { NodeTypes, Push, ResponseHelper, + WebhookHelpers, WorkflowExecuteAdditionalData, WorkflowHelpers, } from '.'; @@ -146,6 +147,7 @@ export class WorkflowRunner { loadStaticData?: boolean, realtime?: boolean, executionId?: string, + responsePromise?: IDeferredPromise, ): Promise { const executionsProcess = config.get('executions.process') as string; const executionsMode = config.get('executions.mode') as string; @@ -153,11 +155,17 @@ export class WorkflowRunner { if (executionsMode === 'queue' && data.executionMode !== 'manual') { // Do not run "manual" executions in bull because sending events to the // frontend would not be possible - executionId = await this.runBull(data, loadStaticData, realtime, executionId); + executionId = await this.runBull( + data, + loadStaticData, + realtime, + executionId, + responsePromise, + ); } else if (executionsProcess === 'main') { - executionId = await this.runMainProcess(data, loadStaticData, executionId); + executionId = await this.runMainProcess(data, loadStaticData, executionId, responsePromise); } else { - executionId = await this.runSubprocess(data, loadStaticData, executionId); + executionId = await this.runSubprocess(data, loadStaticData, executionId, responsePromise); } const postExecutePromise = this.activeExecutions.getPostExecutePromise(executionId); @@ -200,6 +208,7 @@ export class WorkflowRunner { data: IWorkflowExecutionDataProcess, loadStaticData?: boolean, restartExecutionId?: string, + responsePromise?: IDeferredPromise, ): Promise { if (loadStaticData === true && data.workflowData.id) { data.workflowData.staticData = await WorkflowHelpers.getStaticDataById( @@ -256,6 +265,15 @@ export class WorkflowRunner { executionId, true, ); + + additionalData.hooks.hookFunctions.sendResponse = [ + async (response: IExecuteResponsePromiseData): Promise => { + if (responsePromise) { + responsePromise.resolve(response); + } + }, + ]; + additionalData.sendMessageToUI = WorkflowExecuteAdditionalData.sendMessageToUI.bind({ sessionId: data.sessionId, }); @@ -341,11 +359,15 @@ export class WorkflowRunner { loadStaticData?: boolean, realtime?: boolean, restartExecutionId?: string, + responsePromise?: IDeferredPromise, ): Promise { // TODO: If "loadStaticData" is set to true it has to load data new on worker // Register the active execution const executionId = await this.activeExecutions.add(data, undefined, restartExecutionId); + if (responsePromise) { + this.activeExecutions.attachResponsePromise(executionId, responsePromise); + } const jobData: IBullJobData = { executionId, @@ -545,6 +567,7 @@ export class WorkflowRunner { data: IWorkflowExecutionDataProcess, loadStaticData?: boolean, restartExecutionId?: string, + responsePromise?: IDeferredPromise, ): Promise { let startedAt = new Date(); const subprocess = fork(pathJoin(__dirname, 'WorkflowRunnerProcess.js')); @@ -653,6 +676,10 @@ export class WorkflowRunner { } else if (message.type === 'end') { clearTimeout(executionTimeout); this.activeExecutions.remove(executionId, message.data.runData); + } else if (message.type === 'sendResponse') { + if (responsePromise) { + responsePromise.resolve(WebhookHelpers.decodeWebhookResponse(message.data.response)); + } } else if (message.type === 'sendMessageToUI') { // eslint-disable-next-line @typescript-eslint/no-unsafe-call WorkflowExecuteAdditionalData.sendMessageToUI.bind({ sessionId: data.sessionId })( diff --git a/packages/cli/src/WorkflowRunnerProcess.ts b/packages/cli/src/WorkflowRunnerProcess.ts index d7039d69a..e8b8274c9 100644 --- a/packages/cli/src/WorkflowRunnerProcess.ts +++ b/packages/cli/src/WorkflowRunnerProcess.ts @@ -10,6 +10,7 @@ import { IProcessMessage, UserSettings, WorkflowExecute } from 'n8n-core'; import { ExecutionError, IDataObject, + IExecuteResponsePromiseData, IExecuteWorkflowInfo, ILogger, INodeExecutionData, @@ -33,6 +34,7 @@ import { IWorkflowExecuteProcess, IWorkflowExecutionDataProcessWithExecution, NodeTypes, + WebhookHelpers, WorkflowExecuteAdditionalData, WorkflowHelpers, } from '.'; @@ -200,6 +202,15 @@ export class WorkflowRunnerProcess { workflowTimeout <= 0 ? undefined : Date.now() + workflowTimeout * 1000, ); additionalData.hooks = this.getProcessForwardHooks(); + + additionalData.hooks.hookFunctions.sendResponse = [ + async (response: IExecuteResponsePromiseData): Promise => { + await sendToParentProcess('sendResponse', { + response: WebhookHelpers.encodeWebhookResponse(response), + }); + }, + ]; + additionalData.executionId = inputData.executionId; // eslint-disable-next-line @typescript-eslint/no-explicit-any diff --git a/packages/cli/src/databases/MigrationHelpers.ts b/packages/cli/src/databases/MigrationHelpers.ts new file mode 100644 index 000000000..7db121bd8 --- /dev/null +++ b/packages/cli/src/databases/MigrationHelpers.ts @@ -0,0 +1,39 @@ +import { QueryRunner } from 'typeorm'; + +export class MigrationHelpers { + queryRunner: QueryRunner; + + constructor(queryRunner: QueryRunner) { + this.queryRunner = queryRunner; + } + + // runs an operation sequential on chunks of a query that returns a potentially large Array. + /* eslint-disable no-await-in-loop */ + async runChunked( + query: string, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + operation: (results: any[]) => Promise, + limit = 100, + ): Promise { + let offset = 0; + let chunkedQuery: string; + let chunkedQueryResults: unknown[]; + + do { + chunkedQuery = this.chunkQuery(query, limit, offset); + chunkedQueryResults = (await this.queryRunner.query(chunkedQuery)) as unknown[]; + // pass a copy to prevent errors from mutation + await operation([...chunkedQueryResults]); + offset += limit; + } while (chunkedQueryResults.length === limit); + } + /* eslint-enable no-await-in-loop */ + + private chunkQuery(query: string, limit: number, offset = 0): string { + return ` + ${query} + LIMIT ${limit} + OFFSET ${offset} + `; + } +} diff --git a/packages/cli/src/databases/mysqldb/migrations/1630451444017-UpdateWorkflowCredentials.ts b/packages/cli/src/databases/mysqldb/migrations/1630451444017-UpdateWorkflowCredentials.ts index 0012ee0aa..0061052c2 100644 --- a/packages/cli/src/databases/mysqldb/migrations/1630451444017-UpdateWorkflowCredentials.ts +++ b/packages/cli/src/databases/mysqldb/migrations/1630451444017-UpdateWorkflowCredentials.ts @@ -1,5 +1,6 @@ import { MigrationInterface, QueryRunner } from 'typeorm'; import config = require('../../../../config'); +import { MigrationHelpers } from '../../MigrationHelpers'; // replacing the credentials in workflows and execution // `nodeType: name` changes to `nodeType: { id, name }` @@ -8,58 +9,100 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac name = 'UpdateWorkflowCredentials1630451444017'; public async up(queryRunner: QueryRunner): Promise { + console.log('Start migration', this.name); + console.time(this.name); const tablePrefix = config.get('database.tablePrefix'); + const helpers = new MigrationHelpers(queryRunner); const credentialsEntities = await queryRunner.query(` SELECT id, name, type FROM ${tablePrefix}credentials_entity `); - const workflows = await queryRunner.query(` + const workflowsQuery = ` SELECT id, nodes FROM ${tablePrefix}workflow_entity - `); + `; // @ts-ignore - workflows.forEach(async (workflow) => { - const nodes = workflow.nodes; - let credentialsUpdated = false; - // @ts-ignore - nodes.forEach((node) => { - if (node.credentials) { - const allNodeCredentials = Object.entries(node.credentials); - for (const [type, name] of allNodeCredentials) { - if (typeof name === 'string') { - // @ts-ignore - const matchingCredentials = credentialsEntities.find( - // @ts-ignore - (credentials) => credentials.name === name && credentials.type === type, - ); - node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; - credentialsUpdated = true; + await helpers.runChunked(workflowsQuery, (workflows) => { + workflows.forEach(async (workflow) => { + const nodes = workflow.nodes; + let credentialsUpdated = false; + // @ts-ignore + nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, name] of allNodeCredentials) { + if (typeof name === 'string') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.name === name && credentials.type === type, + ); + node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; + credentialsUpdated = true; + } } } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}workflow_entity + SET nodes = :nodes + WHERE id = '${workflow.id}' + `, + { nodes: JSON.stringify(nodes) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); } }); - if (credentialsUpdated) { - const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( - ` - UPDATE ${tablePrefix}workflow_entity - SET nodes = :nodes - WHERE id = '${workflow.id}' - `, - { nodes: JSON.stringify(nodes) }, - {}, - ); - - await queryRunner.query(updateQuery, updateParams); - } }); - const waitingExecutions = await queryRunner.query(` + const waitingExecutionsQuery = ` SELECT id, workflowData FROM ${tablePrefix}execution_entity WHERE waitTill IS NOT NULL AND finished = 0 - `); + `; + // @ts-ignore + await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => { + waitingExecutions.forEach(async (execution) => { + const data = execution.workflowData; + let credentialsUpdated = false; + // @ts-ignore + data.nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, name] of allNodeCredentials) { + if (typeof name === 'string') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.name === name && credentials.type === type, + ); + node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; + credentialsUpdated = true; + } + } + } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}execution_entity + SET workflowData = :data + WHERE id = '${execution.id}' + `, + { data: JSON.stringify(data) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); + } + }); + }); const retryableExecutions = await queryRunner.query(` SELECT id, workflowData @@ -68,8 +111,8 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac ORDER BY startedAt DESC LIMIT 200 `); - - [...waitingExecutions, ...retryableExecutions].forEach(async (execution) => { + // @ts-ignore + retryableExecutions.forEach(async (execution) => { const data = execution.workflowData; let credentialsUpdated = false; // @ts-ignore @@ -78,7 +121,6 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac const allNodeCredentials = Object.entries(node.credentials); for (const [type, name] of allNodeCredentials) { if (typeof name === 'string') { - // @ts-ignore const matchingCredentials = credentialsEntities.find( // @ts-ignore (credentials) => credentials.name === name && credentials.type === type, @@ -92,77 +134,124 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac if (credentialsUpdated) { const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( ` - UPDATE ${tablePrefix}execution_entity - SET workflowData = :data - WHERE id = '${execution.id}' - `, + UPDATE ${tablePrefix}execution_entity + SET workflowData = :data + WHERE id = '${execution.id}' + `, { data: JSON.stringify(data) }, {}, ); - await queryRunner.query(updateQuery, updateParams); + queryRunner.query(updateQuery, updateParams); } }); + console.timeEnd(this.name); } public async down(queryRunner: QueryRunner): Promise { const tablePrefix = config.get('database.tablePrefix'); + const helpers = new MigrationHelpers(queryRunner); const credentialsEntities = await queryRunner.query(` SELECT id, name, type FROM ${tablePrefix}credentials_entity `); - const workflows = await queryRunner.query(` + const workflowsQuery = ` SELECT id, nodes FROM ${tablePrefix}workflow_entity - `); + `; // @ts-ignore - workflows.forEach(async (workflow) => { - const nodes = workflow.nodes; - let credentialsUpdated = false; - // @ts-ignore - nodes.forEach((node) => { - if (node.credentials) { - const allNodeCredentials = Object.entries(node.credentials); - for (const [type, creds] of allNodeCredentials) { - if (typeof creds === 'object') { - // @ts-ignore - const matchingCredentials = credentialsEntities.find( - // @ts-ignore - (credentials) => credentials.id === creds.id && credentials.type === type, - ); - if (matchingCredentials) { - node.credentials[type] = matchingCredentials.name; - } else { - // @ts-ignore - node.credentials[type] = creds.name; + await helpers.runChunked(workflowsQuery, (workflows) => { + workflows.forEach(async (workflow) => { + const nodes = workflow.nodes; + let credentialsUpdated = false; + // @ts-ignore + nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, creds] of allNodeCredentials) { + if (typeof creds === 'object') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.id === creds.id && credentials.type === type, + ); + if (matchingCredentials) { + node.credentials[type] = matchingCredentials.name; + } else { + // @ts-ignore + node.credentials[type] = creds.name; + } + credentialsUpdated = true; } - credentialsUpdated = true; } } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}workflow_entity + SET nodes = :nodes + WHERE id = '${workflow.id}' + `, + { nodes: JSON.stringify(nodes) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); } }); - if (credentialsUpdated) { - const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( - ` - UPDATE ${tablePrefix}workflow_entity - SET nodes = :nodes - WHERE id = '${workflow.id}' - `, - { nodes: JSON.stringify(nodes) }, - {}, - ); - - await queryRunner.query(updateQuery, updateParams); - } }); - const waitingExecutions = await queryRunner.query(` + const waitingExecutionsQuery = ` SELECT id, workflowData FROM ${tablePrefix}execution_entity WHERE waitTill IS NOT NULL AND finished = 0 - `); + `; + // @ts-ignore + await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => { + waitingExecutions.forEach(async (execution) => { + const data = execution.workflowData; + let credentialsUpdated = false; + // @ts-ignore + data.nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, creds] of allNodeCredentials) { + if (typeof creds === 'object') { + // @ts-ignore + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.id === creds.id && credentials.type === type, + ); + if (matchingCredentials) { + node.credentials[type] = matchingCredentials.name; + } else { + // @ts-ignore + node.credentials[type] = creds.name; + } + credentialsUpdated = true; + } + } + } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}execution_entity + SET workflowData = :data + WHERE id = '${execution.id}' + `, + { data: JSON.stringify(data) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); + } + }); + }); const retryableExecutions = await queryRunner.query(` SELECT id, workflowData @@ -171,8 +260,8 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac ORDER BY startedAt DESC LIMIT 200 `); - - [...waitingExecutions, ...retryableExecutions].forEach(async (execution) => { + // @ts-ignore + retryableExecutions.forEach(async (execution) => { const data = execution.workflowData; let credentialsUpdated = false; // @ts-ignore @@ -200,15 +289,15 @@ export class UpdateWorkflowCredentials1630451444017 implements MigrationInterfac if (credentialsUpdated) { const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( ` - UPDATE ${tablePrefix}execution_entity - SET workflowData = :data - WHERE id = '${execution.id}' - `, + UPDATE ${tablePrefix}execution_entity + SET workflowData = :data + WHERE id = '${execution.id}' + `, { data: JSON.stringify(data) }, {}, ); - await queryRunner.query(updateQuery, updateParams); + queryRunner.query(updateQuery, updateParams); } }); } diff --git a/packages/cli/src/databases/postgresdb/migrations/1630419189837-UpdateWorkflowCredentials.ts b/packages/cli/src/databases/postgresdb/migrations/1630419189837-UpdateWorkflowCredentials.ts index 357d7c297..ad3e44f0e 100644 --- a/packages/cli/src/databases/postgresdb/migrations/1630419189837-UpdateWorkflowCredentials.ts +++ b/packages/cli/src/databases/postgresdb/migrations/1630419189837-UpdateWorkflowCredentials.ts @@ -1,5 +1,6 @@ import { MigrationInterface, QueryRunner } from 'typeorm'; import config = require('../../../../config'); +import { MigrationHelpers } from '../../MigrationHelpers'; // replacing the credentials in workflows and execution // `nodeType: name` changes to `nodeType: { id, name }` @@ -8,62 +9,104 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac name = 'UpdateWorkflowCredentials1630419189837'; public async up(queryRunner: QueryRunner): Promise { + console.log('Start migration', this.name); + console.time(this.name); let tablePrefix = config.get('database.tablePrefix'); const schema = config.get('database.postgresdb.schema'); if (schema) { tablePrefix = schema + '.' + tablePrefix; } + const helpers = new MigrationHelpers(queryRunner); const credentialsEntities = await queryRunner.query(` SELECT id, name, type FROM ${tablePrefix}credentials_entity `); - const workflows = await queryRunner.query(` + const workflowsQuery = ` SELECT id, nodes FROM ${tablePrefix}workflow_entity - `); + `; // @ts-ignore - workflows.forEach(async (workflow) => { - const nodes = workflow.nodes; - let credentialsUpdated = false; - // @ts-ignore - nodes.forEach((node) => { - if (node.credentials) { - const allNodeCredentials = Object.entries(node.credentials); - for (const [type, name] of allNodeCredentials) { - if (typeof name === 'string') { - // @ts-ignore - const matchingCredentials = credentialsEntities.find( - // @ts-ignore - (credentials) => credentials.name === name && credentials.type === type, - ); - node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; - credentialsUpdated = true; + await helpers.runChunked(workflowsQuery, (workflows) => { + workflows.forEach(async (workflow) => { + const nodes = workflow.nodes; + let credentialsUpdated = false; + // @ts-ignore + nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, name] of allNodeCredentials) { + if (typeof name === 'string') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.name === name && credentials.type === type, + ); + node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; + credentialsUpdated = true; + } } } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}workflow_entity + SET nodes = :nodes + WHERE id = '${workflow.id}' + `, + { nodes: JSON.stringify(nodes) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); } }); - if (credentialsUpdated) { - const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( - ` - UPDATE ${tablePrefix}workflow_entity - SET nodes = :nodes - WHERE id = '${workflow.id}' - `, - { nodes: JSON.stringify(nodes) }, - {}, - ); - - await queryRunner.query(updateQuery, updateParams); - } }); - const waitingExecutions = await queryRunner.query(` + const waitingExecutionsQuery = ` SELECT id, "workflowData" FROM ${tablePrefix}execution_entity WHERE "waitTill" IS NOT NULL AND finished = FALSE - `); + `; + // @ts-ignore + await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => { + waitingExecutions.forEach(async (execution) => { + const data = execution.workflowData; + let credentialsUpdated = false; + // @ts-ignore + data.nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, name] of allNodeCredentials) { + if (typeof name === 'string') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.name === name && credentials.type === type, + ); + node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; + credentialsUpdated = true; + } + } + } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}execution_entity + SET "workflowData" = :data + WHERE id = '${execution.id}' + `, + { data: JSON.stringify(data) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); + } + }); + }); const retryableExecutions = await queryRunner.query(` SELECT id, "workflowData" @@ -73,7 +116,8 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac LIMIT 200 `); - [...waitingExecutions, ...retryableExecutions].forEach(async (execution) => { + // @ts-ignore + retryableExecutions.forEach(async (execution) => { const data = execution.workflowData; let credentialsUpdated = false; // @ts-ignore @@ -104,9 +148,10 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac {}, ); - await queryRunner.query(updateQuery, updateParams); + queryRunner.query(updateQuery, updateParams); } }); + console.timeEnd(this.name); } public async down(queryRunner: QueryRunner): Promise { @@ -115,62 +160,109 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac if (schema) { tablePrefix = schema + '.' + tablePrefix; } + const helpers = new MigrationHelpers(queryRunner); const credentialsEntities = await queryRunner.query(` SELECT id, name, type FROM ${tablePrefix}credentials_entity `); - const workflows = await queryRunner.query(` + const workflowsQuery = ` SELECT id, nodes FROM ${tablePrefix}workflow_entity - `); + `; // @ts-ignore - workflows.forEach(async (workflow) => { - const nodes = workflow.nodes; - let credentialsUpdated = false; - // @ts-ignore - nodes.forEach((node) => { - if (node.credentials) { - const allNodeCredentials = Object.entries(node.credentials); - for (const [type, creds] of allNodeCredentials) { - if (typeof creds === 'object') { - // @ts-ignore - const matchingCredentials = credentialsEntities.find( + await helpers.runChunked(workflowsQuery, (workflows) => { + workflows.forEach(async (workflow) => { + const nodes = workflow.nodes; + let credentialsUpdated = false; + // @ts-ignore + nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, creds] of allNodeCredentials) { + if (typeof creds === 'object') { // @ts-ignore - (credentials) => credentials.id === creds.id && credentials.type === type, - ); - if (matchingCredentials) { - node.credentials[type] = matchingCredentials.name; - } else { - // @ts-ignore - node.credentials[type] = creds.name; + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.id === creds.id && credentials.type === type, + ); + if (matchingCredentials) { + node.credentials[type] = matchingCredentials.name; + } else { + // @ts-ignore + node.credentials[type] = creds.name; + } + credentialsUpdated = true; } - credentialsUpdated = true; } } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}workflow_entity + SET nodes = :nodes + WHERE id = '${workflow.id}' + `, + { nodes: JSON.stringify(nodes) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); } }); - if (credentialsUpdated) { - const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( - ` - UPDATE ${tablePrefix}workflow_entity - SET nodes = :nodes - WHERE id = '${workflow.id}' - `, - { nodes: JSON.stringify(nodes) }, - {}, - ); - - await queryRunner.query(updateQuery, updateParams); - } }); - const waitingExecutions = await queryRunner.query(` + const waitingExecutionsQuery = ` SELECT id, "workflowData" FROM ${tablePrefix}execution_entity WHERE "waitTill" IS NOT NULL AND finished = FALSE - `); + `; + // @ts-ignore + await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => { + waitingExecutions.forEach(async (execution) => { + const data = execution.workflowData; + let credentialsUpdated = false; + // @ts-ignore + data.nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, creds] of allNodeCredentials) { + if (typeof creds === 'object') { + // @ts-ignore + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.id === creds.id && credentials.type === type, + ); + if (matchingCredentials) { + node.credentials[type] = matchingCredentials.name; + } else { + // @ts-ignore + node.credentials[type] = creds.name; + } + credentialsUpdated = true; + } + } + } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE ${tablePrefix}execution_entity + SET "workflowData" = :data + WHERE id = '${execution.id}' + `, + { data: JSON.stringify(data) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); + } + }); + }); const retryableExecutions = await queryRunner.query(` SELECT id, "workflowData" @@ -179,8 +271,8 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac ORDER BY "startedAt" DESC LIMIT 200 `); - - [...waitingExecutions, ...retryableExecutions].forEach(async (execution) => { + // @ts-ignore + retryableExecutions.forEach(async (execution) => { const data = execution.workflowData; let credentialsUpdated = false; // @ts-ignore @@ -208,15 +300,15 @@ export class UpdateWorkflowCredentials1630419189837 implements MigrationInterfac if (credentialsUpdated) { const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( ` - UPDATE ${tablePrefix}execution_entity - SET "workflowData" = :data - WHERE id = '${execution.id}' - `, + UPDATE ${tablePrefix}execution_entity + SET "workflowData" = :data + WHERE id = '${execution.id}' + `, { data: JSON.stringify(data) }, {}, ); - await queryRunner.query(updateQuery, updateParams); + queryRunner.query(updateQuery, updateParams); } }); } diff --git a/packages/cli/src/databases/sqlite/migrations/1630330987096-UpdateWorkflowCredentials.ts b/packages/cli/src/databases/sqlite/migrations/1630330987096-UpdateWorkflowCredentials.ts index f2a6f0a19..273f644e4 100644 --- a/packages/cli/src/databases/sqlite/migrations/1630330987096-UpdateWorkflowCredentials.ts +++ b/packages/cli/src/databases/sqlite/migrations/1630330987096-UpdateWorkflowCredentials.ts @@ -1,5 +1,6 @@ import { MigrationInterface, QueryRunner } from 'typeorm'; import config = require('../../../../config'); +import { MigrationHelpers } from '../../MigrationHelpers'; // replacing the credentials in workflows and execution // `nodeType: name` changes to `nodeType: { id, name }` @@ -8,58 +9,101 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac name = 'UpdateWorkflowCredentials1630330987096'; public async up(queryRunner: QueryRunner): Promise { + console.log('Start migration', this.name); + console.time(this.name); const tablePrefix = config.get('database.tablePrefix'); + const helpers = new MigrationHelpers(queryRunner); const credentialsEntities = await queryRunner.query(` SELECT id, name, type FROM "${tablePrefix}credentials_entity" `); - const workflows = await queryRunner.query(` + const workflowsQuery = ` SELECT id, nodes FROM "${tablePrefix}workflow_entity" - `); + `; + // @ts-ignore - workflows.forEach(async (workflow) => { - const nodes = JSON.parse(workflow.nodes); - let credentialsUpdated = false; - // @ts-ignore - nodes.forEach((node) => { - if (node.credentials) { - const allNodeCredentials = Object.entries(node.credentials); - for (const [type, name] of allNodeCredentials) { - if (typeof name === 'string') { - // @ts-ignore - const matchingCredentials = credentialsEntities.find( - // @ts-ignore - (credentials) => credentials.name === name && credentials.type === type, - ); - node.credentials[type] = { id: matchingCredentials?.id || null, name }; - credentialsUpdated = true; + await helpers.runChunked(workflowsQuery, (workflows) => { + workflows.forEach(async (workflow) => { + const nodes = JSON.parse(workflow.nodes); + let credentialsUpdated = false; + // @ts-ignore + nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, name] of allNodeCredentials) { + if (typeof name === 'string') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.name === name && credentials.type === type, + ); + node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; + credentialsUpdated = true; + } } } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE "${tablePrefix}workflow_entity" + SET nodes = :nodes + WHERE id = '${workflow.id}' + `, + { nodes: JSON.stringify(nodes) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); } }); - if (credentialsUpdated) { - const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( - ` - UPDATE "${tablePrefix}workflow_entity" - SET nodes = :nodes - WHERE id = '${workflow.id}' - `, - { nodes: JSON.stringify(nodes) }, - {}, - ); - - await queryRunner.query(updateQuery, updateParams); - } }); - const waitingExecutions = await queryRunner.query(` + const waitingExecutionsQuery = ` SELECT id, "workflowData" FROM "${tablePrefix}execution_entity" WHERE "waitTill" IS NOT NULL AND finished = 0 - `); + `; + // @ts-ignore + await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => { + waitingExecutions.forEach(async (execution) => { + const data = JSON.parse(execution.workflowData); + let credentialsUpdated = false; + // @ts-ignore + data.nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, name] of allNodeCredentials) { + if (typeof name === 'string') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore + (credentials) => credentials.name === name && credentials.type === type, + ); + node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; + credentialsUpdated = true; + } + } + } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE "${tablePrefix}execution_entity" + SET "workflowData" = :data + WHERE id = '${execution.id}' + `, + { data: JSON.stringify(data) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); + } + }); + }); const retryableExecutions = await queryRunner.query(` SELECT id, "workflowData" @@ -68,8 +112,8 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac ORDER BY "startedAt" DESC LIMIT 200 `); - - [...waitingExecutions, ...retryableExecutions].forEach(async (execution) => { + // @ts-ignore + retryableExecutions.forEach(async (execution) => { const data = JSON.parse(execution.workflowData); let credentialsUpdated = false; // @ts-ignore @@ -78,12 +122,11 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac const allNodeCredentials = Object.entries(node.credentials); for (const [type, name] of allNodeCredentials) { if (typeof name === 'string') { - // @ts-ignore const matchingCredentials = credentialsEntities.find( // @ts-ignore (credentials) => credentials.name === name && credentials.type === type, ); - node.credentials[type] = { id: matchingCredentials?.id || null, name }; + node.credentials[type] = { id: matchingCredentials?.id.toString() || null, name }; credentialsUpdated = true; } } @@ -92,77 +135,127 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac if (credentialsUpdated) { const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( ` - UPDATE "${tablePrefix}execution_entity" - SET "workflowData" = :data - WHERE id = '${execution.id}' - `, + UPDATE "${tablePrefix}execution_entity" + SET "workflowData" = :data + WHERE id = '${execution.id}' + `, { data: JSON.stringify(data) }, {}, ); - await queryRunner.query(updateQuery, updateParams); + queryRunner.query(updateQuery, updateParams); } }); + console.timeEnd(this.name); } public async down(queryRunner: QueryRunner): Promise { const tablePrefix = config.get('database.tablePrefix'); + const helpers = new MigrationHelpers(queryRunner); const credentialsEntities = await queryRunner.query(` SELECT id, name, type FROM "${tablePrefix}credentials_entity" `); - const workflows = await queryRunner.query(` + const workflowsQuery = ` SELECT id, nodes FROM "${tablePrefix}workflow_entity" - `); + `; + // @ts-ignore - workflows.forEach(async (workflow) => { - const nodes = JSON.parse(workflow.nodes); - let credentialsUpdated = false; + await helpers.runChunked(workflowsQuery, (workflows) => { // @ts-ignore - nodes.forEach((node) => { - if (node.credentials) { - const allNodeCredentials = Object.entries(node.credentials); - for (const [type, creds] of allNodeCredentials) { - if (typeof creds === 'object') { - // @ts-ignore - const matchingCredentials = credentialsEntities.find( - // @ts-ignore - (credentials) => credentials.id === creds.id && credentials.type === type, - ); - if (matchingCredentials) { - node.credentials[type] = matchingCredentials.name; - } else { - // @ts-ignore - node.credentials[type] = creds.name; + workflows.forEach(async (workflow) => { + const nodes = JSON.parse(workflow.nodes); + let credentialsUpdated = false; + // @ts-ignore + nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, creds] of allNodeCredentials) { + if (typeof creds === 'object') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore double-equals because creds.id can be string or number + (credentials) => credentials.id == creds.id && credentials.type === type, + ); + if (matchingCredentials) { + node.credentials[type] = matchingCredentials.name; + } else { + // @ts-ignore + node.credentials[type] = creds.name; + } + credentialsUpdated = true; } - credentialsUpdated = true; } } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE "${tablePrefix}workflow_entity" + SET nodes = :nodes + WHERE id = '${workflow.id}' + `, + { nodes: JSON.stringify(nodes) }, + {}, + ); + + queryRunner.query(updateQuery, updateParams); } }); - if (credentialsUpdated) { - const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( - ` - UPDATE "${tablePrefix}workflow_entity" - SET nodes = :nodes - WHERE id = '${workflow.id}' - `, - { nodes: JSON.stringify(nodes) }, - {}, - ); - - await queryRunner.query(updateQuery, updateParams); - } }); - const waitingExecutions = await queryRunner.query(` + const waitingExecutionsQuery = ` SELECT id, "workflowData" FROM "${tablePrefix}execution_entity" WHERE "waitTill" IS NOT NULL AND finished = 0 - `); + `; + + // @ts-ignore + await helpers.runChunked(waitingExecutionsQuery, (waitingExecutions) => { + // @ts-ignore + waitingExecutions.forEach(async (execution) => { + const data = JSON.parse(execution.workflowData); + let credentialsUpdated = false; + // @ts-ignore + data.nodes.forEach((node) => { + if (node.credentials) { + const allNodeCredentials = Object.entries(node.credentials); + for (const [type, creds] of allNodeCredentials) { + if (typeof creds === 'object') { + const matchingCredentials = credentialsEntities.find( + // @ts-ignore double-equals because creds.id can be string or number + (credentials) => credentials.id == creds.id && credentials.type === type, + ); + if (matchingCredentials) { + node.credentials[type] = matchingCredentials.name; + } else { + // @ts-ignore + node.credentials[type] = creds.name; + } + credentialsUpdated = true; + } + } + } + }); + if (credentialsUpdated) { + const [updateQuery, updateParams] = + queryRunner.connection.driver.escapeQueryWithParameters( + ` + UPDATE "${tablePrefix}execution_entity" + SET "workflowData" = :data + WHERE id = '${execution.id}' + `, + { data: JSON.stringify(data) }, + {}, + ); + + await queryRunner.query(updateQuery, updateParams); + } + }); + }); const retryableExecutions = await queryRunner.query(` SELECT id, "workflowData" @@ -172,7 +265,8 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac LIMIT 200 `); - [...waitingExecutions, ...retryableExecutions].forEach(async (execution) => { + // @ts-ignore + retryableExecutions.forEach(async (execution) => { const data = JSON.parse(execution.workflowData); let credentialsUpdated = false; // @ts-ignore @@ -181,10 +275,9 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac const allNodeCredentials = Object.entries(node.credentials); for (const [type, creds] of allNodeCredentials) { if (typeof creds === 'object') { - // @ts-ignore const matchingCredentials = credentialsEntities.find( - // @ts-ignore - (credentials) => credentials.id === creds.id && credentials.type === type, + // @ts-ignore double-equals because creds.id can be string or number + (credentials) => credentials.id == creds.id && credentials.type === type, ); if (matchingCredentials) { node.credentials[type] = matchingCredentials.name; @@ -200,15 +293,15 @@ export class UpdateWorkflowCredentials1630330987096 implements MigrationInterfac if (credentialsUpdated) { const [updateQuery, updateParams] = queryRunner.connection.driver.escapeQueryWithParameters( ` - UPDATE "${tablePrefix}execution_entity" - SET "workflowData" = :data - WHERE id = '${execution.id}' - `, + UPDATE "${tablePrefix}execution_entity" + SET "workflowData" = :data + WHERE id = '${execution.id}' + `, { data: JSON.stringify(data) }, {}, ); - await queryRunner.query(updateQuery, updateParams); + queryRunner.query(updateQuery, updateParams); } }); } diff --git a/packages/cli/src/telemetry/index.ts b/packages/cli/src/telemetry/index.ts index fb38ed257..d350c6d8b 100644 --- a/packages/cli/src/telemetry/index.ts +++ b/packages/cli/src/telemetry/index.ts @@ -119,6 +119,7 @@ export class Telemetry { this.client.identify( { userId: this.instanceId, + anonymousId: '000000000000', traits: { ...traits, instanceId: this.instanceId, @@ -138,6 +139,7 @@ export class Telemetry { this.client.track( { userId: this.instanceId, + anonymousId: '000000000000', event: eventName, properties, }, diff --git a/packages/core/package.json b/packages/core/package.json index 88bfc1170..dfc9c18a5 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -1,6 +1,6 @@ { "name": "n8n-core", - "version": "0.90.0", + "version": "0.93.0", "description": "Core functionality of n8n", "license": "SEE LICENSE IN LICENSE.md", "homepage": "https://n8n.io", @@ -50,7 +50,7 @@ "form-data": "^4.0.0", "lodash.get": "^4.4.2", "mime-types": "^2.1.27", - "n8n-workflow": "~0.73.0", + "n8n-workflow": "~0.76.0", "oauth-1.0a": "^2.2.6", "p-cancelable": "^2.0.0", "qs": "^6.10.1", diff --git a/packages/core/src/NodeExecuteFunctions.ts b/packages/core/src/NodeExecuteFunctions.ts index 6ef54f37a..fa1894a1d 100644 --- a/packages/core/src/NodeExecuteFunctions.ts +++ b/packages/core/src/NodeExecuteFunctions.ts @@ -22,6 +22,7 @@ import { ICredentialsExpressionResolveValues, IDataObject, IExecuteFunctions, + IExecuteResponsePromiseData, IExecuteSingleFunctions, IExecuteWorkflowInfo, IHttpRequestOptions, @@ -71,7 +72,7 @@ import { fromBuffer } from 'file-type'; import { lookup } from 'mime-types'; import axios, { AxiosProxyConfig, AxiosRequestConfig, Method } from 'axios'; -import { URLSearchParams } from 'url'; +import { URL, URLSearchParams } from 'url'; // eslint-disable-next-line import/no-cycle import { BINARY_ENCODING, @@ -86,6 +87,14 @@ import { axios.defaults.timeout = 300000; // Prevent axios from adding x-form-www-urlencoded headers by default axios.defaults.headers.post = {}; +axios.defaults.headers.put = {}; +axios.defaults.headers.patch = {}; +axios.defaults.paramsSerializer = (params) => { + if (params instanceof URLSearchParams) { + return params.toString(); + } + return stringify(params, { arrayFormat: 'indices' }); +}; const requestPromiseWithDefaults = requestPromise.defaults({ timeout: 300000, // 5 minutes @@ -128,6 +137,28 @@ function searchForHeader(headers: IDataObject, headerName: string) { return headerNames.find((thisHeader) => thisHeader.toLowerCase() === headerName); } +async function generateContentLengthHeader(formData: FormData, headers: IDataObject) { + if (!formData || !formData.getLength) { + return; + } + try { + const length = await new Promise((res, rej) => { + formData.getLength((error: Error | null, length: number) => { + if (error) { + rej(error); + return; + } + res(length); + }); + }); + headers = Object.assign(headers, { + 'content-length': length, + }); + } catch (error) { + Logger.error('Unable to calculate form data length', { error }); + } +} + async function parseRequestObject(requestObject: IDataObject) { // This function is a temporary implementation // That translates all http requests done via @@ -192,6 +223,7 @@ async function parseRequestObject(requestObject: IDataObject) { delete axiosConfig.headers[contentTypeHeaderKeyName]; const headers = axiosConfig.data.getHeaders(); axiosConfig.headers = Object.assign(axiosConfig.headers || {}, headers); + await generateContentLengthHeader(axiosConfig.data, axiosConfig.headers); } else { // When using the `form` property it means the content should be x-www-form-urlencoded. if (requestObject.form !== undefined && requestObject.body === undefined) { @@ -228,6 +260,7 @@ async function parseRequestObject(requestObject: IDataObject) { // Mix in headers as FormData creates the boundary. const headers = axiosConfig.data.getHeaders(); axiosConfig.headers = Object.assign(axiosConfig.headers || {}, headers); + await generateContentLengthHeader(axiosConfig.data, axiosConfig.headers); } else if (requestObject.body !== undefined) { // If we have body and possibly form if (requestObject.form !== undefined) { @@ -338,7 +371,63 @@ async function parseRequestObject(requestObject: IDataObject) { } if (requestObject.proxy !== undefined) { - axiosConfig.proxy = requestObject.proxy as AxiosProxyConfig; + // try our best to parse the url provided. + if (typeof requestObject.proxy === 'string') { + try { + const url = new URL(requestObject.proxy); + axiosConfig.proxy = { + host: url.hostname, + port: parseInt(url.port, 10), + protocol: url.protocol, + }; + if (!url.port) { + // Sets port to a default if not informed + if (url.protocol === 'http') { + axiosConfig.proxy.port = 80; + } else if (url.protocol === 'https') { + axiosConfig.proxy.port = 443; + } + } + if (url.username || url.password) { + axiosConfig.proxy.auth = { + username: url.username, + password: url.password, + }; + } + } catch (error) { + // Not a valid URL. We will try to simply parse stuff + // such as user:pass@host:port without protocol (we'll assume http) + if (requestObject.proxy.includes('@')) { + const [userpass, hostport] = requestObject.proxy.split('@'); + const [username, password] = userpass.split(':'); + const [hostname, port] = hostport.split(':'); + axiosConfig.proxy = { + host: hostname, + port: parseInt(port, 10), + protocol: 'http', + auth: { + username, + password, + }, + }; + } else if (requestObject.proxy.includes(':')) { + const [hostname, port] = requestObject.proxy.split(':'); + axiosConfig.proxy = { + host: hostname, + port: parseInt(port, 10), + protocol: 'http', + }; + } else { + axiosConfig.proxy = { + host: requestObject.proxy, + port: 80, + protocol: 'http', + }; + } + } + } else { + axiosConfig.proxy = requestObject.proxy as AxiosProxyConfig; + } } if (requestObject.encoding === null) { @@ -357,6 +446,7 @@ async function parseRequestObject(requestObject: IDataObject) { if ( requestObject.json !== false && axiosConfig.data !== undefined && + axiosConfig.data !== '' && !(axiosConfig.data instanceof Buffer) && !allHeaders.some((headerKey) => headerKey.toLowerCase() === 'content-type') ) { @@ -406,6 +496,11 @@ async function proxyRequestToAxios( axiosConfig = Object.assign(axiosConfig, await parseRequestObject(configObject)); + Logger.debug('Proxying request to axios', { + originalConfig: configObject, + parsedConfig: axiosConfig, + }); + return new Promise((resolve, reject) => { axios(axiosConfig) .then((response) => { @@ -1567,6 +1662,9 @@ export function getExecuteFunctions( Logger.warn(`There was a problem sending messsage to UI: ${error.message}`); } }, + async sendResponse(response: IExecuteResponsePromiseData): Promise { + await additionalData.hooks?.executeHookFunctions('sendResponse', [response]); + }, helpers: { httpRequest, prepareBinaryData, diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 15d1cce2b..b0c6167aa 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -12,7 +12,6 @@ export * from './ActiveWorkflows'; export * from './ActiveWebhooks'; export * from './Constants'; export * from './Credentials'; -export * from './DeferredPromise'; export * from './Interfaces'; export * from './LoadNodeParameterOptions'; export * from './NodeExecuteFunctions'; diff --git a/packages/core/test/Helpers.ts b/packages/core/test/Helpers.ts index 387ac67a8..eb5920182 100644 --- a/packages/core/test/Helpers.ts +++ b/packages/core/test/Helpers.ts @@ -4,6 +4,7 @@ import { ICredentialDataDecryptedObject, ICredentialsHelper, IDataObject, + IDeferredPromise, IExecuteWorkflowInfo, INodeCredentialsDetails, INodeExecutionData, @@ -20,7 +21,7 @@ import { WorkflowHooks, } from 'n8n-workflow'; -import { Credentials, IDeferredPromise, IExecuteFunctions } from '../src'; +import { Credentials, IExecuteFunctions } from '../src'; export class CredentialsHelper extends ICredentialsHelper { getDecrypted( diff --git a/packages/core/test/WorkflowExecute.test.ts b/packages/core/test/WorkflowExecute.test.ts index 364fb23d5..b1ac658dc 100644 --- a/packages/core/test/WorkflowExecute.test.ts +++ b/packages/core/test/WorkflowExecute.test.ts @@ -1,6 +1,14 @@ -import { IConnections, ILogger, INode, IRun, LoggerProxy, Workflow } from 'n8n-workflow'; +import { + createDeferredPromise, + IConnections, + ILogger, + INode, + IRun, + LoggerProxy, + Workflow, +} from 'n8n-workflow'; -import { createDeferredPromise, WorkflowExecute } from '../src'; +import { WorkflowExecute } from '../src'; import * as Helpers from './Helpers'; diff --git a/packages/design-system/package.json b/packages/design-system/package.json index 13fff0210..7a1b8c79a 100644 --- a/packages/design-system/package.json +++ b/packages/design-system/package.json @@ -1,6 +1,6 @@ { "name": "n8n-design-system", - "version": "0.5.0", + "version": "0.6.0", "license": "SEE LICENSE IN LICENSE.md", "homepage": "https://n8n.io", "author": { diff --git a/packages/design-system/src/styleguide/border.stories.mdx b/packages/design-system/src/styleguide/border.stories.mdx index d5a83e064..b7f0785f3 100644 --- a/packages/design-system/src/styleguide/border.stories.mdx +++ b/packages/design-system/src/styleguide/border.stories.mdx @@ -16,7 +16,7 @@ import VariableTable from './VariableTable.vue'; {{ - template: ``, + template: ``, components: { VariableTable, }, diff --git a/packages/design-system/src/styleguide/colors.stories.mdx b/packages/design-system/src/styleguide/colors.stories.mdx index 16e372bcd..7090bbde3 100644 --- a/packages/design-system/src/styleguide/colors.stories.mdx +++ b/packages/design-system/src/styleguide/colors.stories.mdx @@ -44,7 +44,7 @@ import ColorCircles from './ColorCircles.vue'; {{ - template: ``, + template: ``, components: { ColorCircles, }, @@ -109,7 +109,7 @@ import ColorCircles from './ColorCircles.vue'; {{ - template: ``, + template: ``, components: { ColorCircles, }, @@ -129,3 +129,16 @@ import ColorCircles from './ColorCircles.vue'; }} + +## Canvas + + + + {{ + template: ``, + components: { + ColorCircles, + }, + }} + + diff --git a/packages/design-system/theme/src/_tokens.scss b/packages/design-system/theme/src/_tokens.scss index c2c836db9..24e245a34 100644 --- a/packages/design-system/theme/src/_tokens.scss +++ b/packages/design-system/theme/src/_tokens.scss @@ -75,6 +75,15 @@ var(--color-success-tint-2-l) ); + --color-success-light-h: 150; + --color-success-light-s: 54%; + --color-success-light-l: 70%; + --color-success-light: hsl( + var(--color-success-light-h), + var(--color-success-light-s), + var(--color-success-light-l) + ); + --color-warning-h: 36; --color-warning-s: 77%; --color-warning-l: 57%; @@ -187,6 +196,24 @@ var(--color-text-xlight-l) ); + --color-foreground-xdark-h: 220; + --color-foreground-xdark-s: 7.4%; + --color-foreground-xdark-l: 52.5%; + --color-foreground-xdark: hsl( + var(--color-foreground-xdark-h), + var(--color-foreground-xdark-s), + var(--color-foreground-xdark-l) + ); + + --color-foreground-dark-h: 228; + --color-foreground-dark-s: 9.6%; + --color-foreground-dark-l: 79.6%; + --color-foreground-dark: hsl( + var(--color-foreground-dark-h), + var(--color-foreground-dark-s), + var(--color-foreground-dark-l) + ); + --color-foreground-base-h: 220; --color-foreground-base-s: 20%; --color-foreground-base-l: 88.2%; @@ -259,6 +286,25 @@ var(--color-background-xlight-l) ); + --color-canvas-dot-h: 204; + --color-canvas-dot-s: 15.6%; + --color-canvas-dot-l: 87.5%; + --color-canvas-dot: hsl( + var(--color-canvas-dot-h), + var(--color-canvas-dot-s), + var(--color-canvas-dot-l) + ); + + --color-canvas-background-h: 260; + --color-canvas-background-s: 100%; + --color-canvas-background-l: 99.4%; + --color-canvas-background: hsl( + var(--color-canvas-background-h), + var(--color-canvas-background-s), + var(--color-canvas-background-l) + ); + + --border-radius-xlarge: 12px; --border-radius-large: 8px; --border-radius-base: 4px; --border-radius-small: 2px; diff --git a/packages/editor-ui/package.json b/packages/editor-ui/package.json index 2f381b574..93be8548e 100644 --- a/packages/editor-ui/package.json +++ b/packages/editor-ui/package.json @@ -1,6 +1,6 @@ { "name": "n8n-editor-ui", - "version": "0.113.0", + "version": "0.116.0", "description": "Workflow Editor UI for n8n", "license": "SEE LICENSE IN LICENSE.md", "homepage": "https://n8n.io", @@ -26,7 +26,7 @@ }, "dependencies": { "@fontsource/open-sans": "^4.5.0", - "n8n-design-system": "~0.5.0", + "n8n-design-system": "~0.6.0", "timeago.js": "^4.0.2", "v-click-outside": "^3.1.2", "vue-fragment": "^1.5.2" @@ -71,7 +71,7 @@ "lodash.debounce": "^4.0.8", "lodash.get": "^4.4.2", "lodash.set": "^4.3.2", - "n8n-workflow": "~0.73.0", + "n8n-workflow": "~0.76.0", "sass": "^1.26.5", "normalize-wheel": "^1.0.1", "prismjs": "^1.17.1", diff --git a/packages/editor-ui/src/Interface.ts b/packages/editor-ui/src/Interface.ts index 02fe67ec4..7332f3548 100644 --- a/packages/editor-ui/src/Interface.ts +++ b/packages/editor-ui/src/Interface.ts @@ -22,32 +22,61 @@ import { WorkflowExecuteMode, } from 'n8n-workflow'; -import { - PaintStyle, -} from 'jsplumb'; - declare module 'jsplumb' { + interface PaintStyle { + stroke?: string; + fill?: string; + strokeWidth?: number; + outlineStroke?: string; + outlineWidth?: number; + } + interface Anchor { lastReturnValue: number[]; } interface Connection { + __meta?: { + sourceNodeName: string, + sourceOutputIndex: number, + targetNodeName: string, + targetOutputIndex: number, + }; + canvas?: HTMLElement; + connector?: { + setTargetEndpoint: (endpoint: Endpoint) => void; + resetTargetEndpoint: () => void; + bounds: { + minX: number; + maxX: number; + minY: number; + maxY: number; + } + }; + // bind(event: string, (connection: Connection): void;): void; // tslint:disable-line:no-any - bind(event: string, callback: Function): void; // tslint:disable-line:no-any + bind(event: string, callback: Function): void; removeOverlay(name: string): void; removeOverlays(): void; setParameter(name: string, value: any): void; // tslint:disable-line:no-any setPaintStyle(arg0: PaintStyle): void; addOverlay(arg0: any[]): void; // tslint:disable-line:no-any setConnector(arg0: any[]): void; // tslint:disable-line:no-any + getUuids(): [string, string]; } interface Endpoint { + __meta?: { + nodeName: string, + index: number, + }; getOverlay(name: string): any; // tslint:disable-line:no-any } interface Overlay { setVisible(visible: boolean): void; + setLocation(location: number): void; + canvas?: HTMLElement; } interface OnConnectionBindInfo { @@ -66,18 +95,14 @@ export interface IEndpointOptions { dragProxy?: any; // tslint:disable-line:no-any endpoint?: string; endpointStyle?: object; + endpointHoverStyle?: object; isSource?: boolean; isTarget?: boolean; maxConnections?: number; overlays?: any; // tslint:disable-line:no-any parameters?: any; // tslint:disable-line:no-any uuid?: string; -} - -export interface IConnectionsUi { - [key: string]: { - [key: string]: IEndpointOptions; - }; + enabled?: boolean; } export interface IUpdateInformation { @@ -95,20 +120,16 @@ export interface INodeUpdatePropertiesInformation { }; } -export type XYPositon = [number, number]; +export type XYPosition = [number, number]; export type MessageType = 'success' | 'warning' | 'info' | 'error'; export interface INodeUi extends INode { - position: XYPositon; + position: XYPosition; color?: string; notes?: string; issues?: INodeIssues; - _jsPlumb?: { - endpoints?: { - [key: string]: IEndpointOptions[]; - }; - }; + name: string; } export interface INodeTypesMaxCount { @@ -604,7 +625,7 @@ export interface IRootState { lastSelectedNodeOutputIndex: number | null; nodeIndex: Array; nodeTypes: INodeTypeDescription[]; - nodeViewOffsetPosition: XYPositon; + nodeViewOffsetPosition: XYPosition; nodeViewMoveInProgress: boolean; selectedNodes: INodeUi[]; sessionId: string; @@ -670,5 +691,13 @@ export interface IRestApiContext { export interface IZoomConfig { scale: number; - offset: XYPositon; + offset: XYPosition; } + +export interface IBounds { + minX: number; + minY: number; + maxX: number; + maxY: number; +} + diff --git a/packages/editor-ui/src/components/Node.vue b/packages/editor-ui/src/components/Node.vue index adabe3b3f..3f806432c 100644 --- a/packages/editor-ui/src/components/Node.vue +++ b/packages/editor-ui/src/components/Node.vue @@ -1,25 +1,35 @@