mudhorn/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2024-11-14 00:24:07 -08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'master' into cat-301-runner-idle-shutdown

Browse source
This commit is contained in:
Iván Ovejero 2024-11-07 13:24:23 +01:00
parent b18a3e269d d25ae8e0d9
commit 5b139e49f6
No known key found for this signature in database
53 changed files with 1678 additions and 1872 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.github/workflows/release-publish.yml vendored
View file

@ -42,7 +42,7 @@ jobs:
uses: actions/cache/save@v4.0.0
with:
path: ./packages/**/dist
key: ${{ github.sha }}-base:build
key: ${{ github.sha }}-release:build
- name: Dry-run publishing
run: pnpm publish -r --no-git-checks --dry-run
@ -141,7 +141,7 @@ jobs:
uses: actions/cache/restore@v4.0.0
with:
path: ./packages/**/dist
key: ${{ github.sha }}:db-tests
key: ${{ github.sha }}-release:build
- name: Create a frontend release
uses: getsentry/action-release@v1.7.0

9
packages/@n8n/config/src/configs/runners.config.ts
View file

@ -10,9 +10,8 @@ export type TaskRunnerMode = 'internal_childprocess' | 'internal_launcher' | 'ex
@Config
export class TaskRunnersConfig {
// Defaults to true for now
@Env('N8N_RUNNERS_DISABLED')
disabled: boolean = true;
@Env('N8N_RUNNERS_ENABLED')
enabled: boolean = false;
// Defaults to true for now
@Env('N8N_RUNNERS_MODE')
@ -51,6 +50,10 @@ export class TaskRunnersConfig {
@Env('N8N_RUNNERS_MAX_CONCURRENCY')
maxConcurrency: number = 5;
/** Should the output of deduplication be asserted for correctness */
@Env('N8N_RUNNERS_ASSERT_DEDUPLICATION_OUTPUT')
assertDeduplicationOutput: boolean = false;
/** How long (in minutes) until shutting down an idle runner. */
@Env('N8N_RUNNERS_IDLE_TIMEOUT')
idleTimeout: number = 5;

3
packages/@n8n/config/test/config.test.ts
View file

@ -222,7 +222,7 @@ describe('GlobalConfig', () => {
},
},
taskRunners: {
disabled: true,
enabled: false,
mode: 'internal_childprocess',
path: '/runners',
authToken: '',
@ -233,6 +233,7 @@ describe('GlobalConfig', () => {
launcherRunner: 'javascript',
maxOldSpaceSize: '',
maxConcurrency: 5,
assertDeduplicationOutput: false,
},
sentry: {
backendDsn: '',

29
packages/@n8n/task-runner/src/data-request/data-request-response-reconstruct.ts Normal file
View file

@ -0,0 +1,29 @@
import type { IExecuteData, INodeExecutionData } from 'n8n-workflow';
import type { DataRequestResponse } from '@/runner-types';
/**
* Reconstructs data from a DataRequestResponse to the initial
* data structures.
*/
export class DataRequestResponseReconstruct {
/**
* Reconstructs `connectionInputData` from a DataRequestResponse
*/
reconstructConnectionInputData(
inputData: DataRequestResponse['inputData'],
): INodeExecutionData[] {
return inputData?.main?.[0] ?? [];
}
/**
* Reconstruct `executeData` from a DataRequestResponse
*/
reconstructExecuteData(response: DataRequestResponse): IExecuteData {
return {
data: response.inputData,
node: response.node,
source: response.connectionInputSource,
};
}
}

1
packages/@n8n/task-runner/src/index.ts
View file

@ -1,3 +1,4 @@
export * from './task-runner';
export * from './runner-types';
export * from './message-types';
export * from './data-request/data-request-response-reconstruct';

38
packages/@n8n/task-runner/src/js-task-runner/__tests__/js-task-runner.test.ts
View file

@ -3,15 +3,21 @@ import type { CodeExecutionMode, IDataObject } from 'n8n-workflow';
import fs from 'node:fs';
import { builtinModules } from 'node:module';
import type { JsRunnerConfig } from '@/config/js-runner-config';
import { MainConfig } from '@/config/main-config';
import { ExecutionError } from '@/js-task-runner/errors/execution-error';
import { ValidationError } from '@/js-task-runner/errors/validation-error';
import type { DataRequestResponse, JSExecSettings } from '@/js-task-runner/js-task-runner';
import type { JSExecSettings } from '@/js-task-runner/js-task-runner';
import { JsTaskRunner } from '@/js-task-runner/js-task-runner';
import type { DataRequestResponse } from '@/runner-types';
import type { Task } from '@/task-runner';
import { newCodeTaskData, newTaskWithSettings, withPairedItem, wrapIntoJson } from './test-data';
import type { JsRunnerConfig } from '../../config/js-runner-config';
import { MainConfig } from '../../config/main-config';
import { ExecutionError } from '../errors/execution-error';
import {
newDataRequestResponse,
newTaskWithSettings,
withPairedItem,
wrapIntoJson,
} from './test-data';
jest.mock('ws');
@ -68,7 +74,7 @@ describe('JsTaskRunner', () => {
nodeMode: 'runOnceForAllItems',
...settings,
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson)),
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson)),
runner,
});
};
@ -91,7 +97,7 @@ describe('JsTaskRunner', () => {
nodeMode: 'runOnceForEachItem',
...settings,
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson)),
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson)),
runner,
});
};
@ -108,7 +114,7 @@ describe('JsTaskRunner', () => {
await execTaskWithParams({
task,
taskData: newCodeTaskData([wrapIntoJson({})]),
taskData: newDataRequestResponse([wrapIntoJson({})]),
});
expect(defaultTaskRunner.makeRpcCall).toHaveBeenCalledWith(task.taskId, 'logNodeOutput', [
@ -243,7 +249,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: $env.VAR1 }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: {
isEnvAccessBlocked: false,
isProcessAvailable: true,
@ -262,7 +268,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: $env.VAR1 }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: {
isEnvAccessBlocked: true,
isProcessAvailable: true,
@ -279,7 +285,7 @@ describe('JsTaskRunner', () => {
code: 'return Object.values($env).concat(Object.keys($env))',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: {
isEnvAccessBlocked: false,
isProcessAvailable: true,
@ -298,7 +304,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: $env.N8N_RUNNERS_N8N_URI }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: undefined,
}),
});
@ -313,7 +319,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: Buffer.from("test-buffer").toString() }',
nodeMode: 'runOnceForAllItems',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: undefined,
}),
});
@ -325,7 +331,7 @@ describe('JsTaskRunner', () => {
code: 'return { val: Buffer.from("test-buffer").toString() }',
nodeMode: 'runOnceForEachItem',
}),
taskData: newCodeTaskData(inputItems.map(wrapIntoJson), {
taskData: newDataRequestResponse(inputItems.map(wrapIntoJson), {
envProviderState: undefined,
}),
});
@ -771,7 +777,7 @@ describe('JsTaskRunner', () => {
code: 'unknown',
nodeMode,
}),
taskData: newCodeTaskData([wrapIntoJson({ a: 1 })]),
taskData: newDataRequestResponse([wrapIntoJson({ a: 1 })]),
}),
).rejects.toThrow(ExecutionError);
},
@ -793,7 +799,7 @@ describe('JsTaskRunner', () => {
jest.spyOn(runner, 'sendOffers').mockImplementation(() => {});
jest
.spyOn(runner, 'requestData')
.mockResolvedValue(newCodeTaskData([wrapIntoJson({ a: 1 })]));
.mockResolvedValue(newDataRequestResponse([wrapIntoJson({ a: 1 })]));
await runner.receivedSettings(taskId, task.settings);

29
packages/@n8n/task-runner/src/js-task-runner/__tests__/test-data.ts
View file

@ -2,7 +2,8 @@ import type { IDataObject, INode, INodeExecutionData, ITaskData } from 'n8n-work
import { NodeConnectionType } from 'n8n-workflow';
import { nanoid } from 'nanoid';
import type { DataRequestResponse, JSExecSettings } from '@/js-task-runner/js-task-runner';
import type { JSExecSettings } from '@/js-task-runner/js-task-runner';
import type { DataRequestResponse } from '@/runner-types';
import type { Task } from '@/task-runner';
/**
@ -46,10 +47,10 @@ export const newTaskData = (opts: Partial<ITaskData> & Pick<ITaskData, 'source'>
});
/**
* Creates a new all code task data with the given options
* Creates a new data request response with the given options
*/
export const newCodeTaskData = (
codeNodeInputData: INodeExecutionData[],
export const newDataRequestResponse = (
inputData: INodeExecutionData[],
opts: Partial<DataRequestResponse> = {},
): DataRequestResponse => {
const codeNode = newNode({
@ -83,9 +84,8 @@ export const newCodeTaskData = (
nodes: [manualTriggerNode, codeNode],
},
inputData: {
main: [codeNodeInputData],
main: [inputData],
},
connectionInputData: codeNodeInputData,
node: codeNode,
runExecutionData: {
startData: {},
@ -95,7 +95,7 @@ export const newCodeTaskData = (
newTaskData({
source: [],
data: {
main: [codeNodeInputData],
main: [inputData],
},
}),
],
@ -137,14 +137,13 @@ export const newCodeTaskData = (
var: 'value',
},
},
executeData: {
node: codeNode,
data: {
main: [codeNodeInputData],
},
source: {
main: [{ previousNode: manualTriggerNode.name }],
},
connectionInputSource: {
main: [
{
previousNode: 'Trigger',
previousNodeOutput: 0,
},
],
},
...opts,
};

20
packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/__tests__/built-ins-parser.test.ts
View file

@ -1,8 +1,13 @@
import { getAdditionalKeys } from 'n8n-core';
import type { IDataObject, INodeType, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
import type {
IDataObject,
IExecuteData,
INodeType,
IWorkflowExecuteAdditionalData,
} from 'n8n-workflow';
import { Workflow, WorkflowDataProxy } from 'n8n-workflow';
import { newCodeTaskData } from '../../__tests__/test-data';
import { newDataRequestResponse } from '../../__tests__/test-data';
import { BuiltInsParser } from '../built-ins-parser';
import { BuiltInsParserState } from '../built-ins-parser-state';
@ -159,7 +164,12 @@ describe('BuiltInsParser', () => {
describe('WorkflowDataProxy built-ins', () => {
it('should have a known list of built-ins', () => {
const data = newCodeTaskData([]);
const data = newDataRequestResponse([]);
const executeData: IExecuteData = {
data: {},
node: data.node,
source: data.connectionInputSource,
};
const dataProxy = new WorkflowDataProxy(
new Workflow({
...data.workflow,
@ -179,7 +189,7 @@ describe('BuiltInsParser', () => {
data.runIndex,
0,
data.activeNodeName,
data.connectionInputData,
[],
data.siblingParameters,
data.mode,
getAdditionalKeys(
@ -187,7 +197,7 @@ describe('BuiltInsParser', () => {
data.mode,
data.runExecutionData,
),
data.executeData,
executeData,
data.defaultReturnRunIndex,
data.selfData,
data.contextNodeName,

114
packages/@n8n/task-runner/src/js-task-runner/js-task-runner.ts
View file

@ -1,28 +1,25 @@
import { getAdditionalKeys } from 'n8n-core';
import {
WorkflowDataProxy,
// type IWorkflowDataProxyAdditionalKeys,
Workflow,
} from 'n8n-workflow';
import { WorkflowDataProxy, Workflow } from 'n8n-workflow';
import type {
CodeExecutionMode,
INode,
ITaskDataConnections,
IWorkflowExecuteAdditionalData,
WorkflowParameters,
IDataObject,
IExecuteData,
INodeExecutionData,
INodeParameters,
IRunExecutionData,
WorkflowExecuteMode,
WorkflowParameters,
ITaskDataConnections,
INode,
IRunExecutionData,
EnvProviderState,
IExecuteData,
INodeTypeDescription,
} from 'n8n-workflow';
import * as a from 'node:assert';
import { runInNewContext, type Context } from 'node:vm';
import type { TaskResultData } from '@/runner-types';
import type { MainConfig } from '@/config/main-config';
import type { DataRequestResponse, PartialAdditionalData, TaskResultData } from '@/runner-types';
import { type Task, TaskRunner } from '@/task-runner';
import { BuiltInsParser } from './built-ins-parser/built-ins-parser';
@ -33,7 +30,7 @@ import { makeSerializable } from './errors/serializable-error';
import type { RequireResolver } from './require-resolver';
import { createRequireResolver } from './require-resolver';
import { validateRunForAllItemsOutput, validateRunForEachItemOutput } from './result-validation';
import type { MainConfig } from '../config/main-config';
import { DataRequestResponseReconstruct } from '../data-request/data-request-response-reconstruct';
export interface JSExecSettings {
code: string;
@ -45,34 +42,19 @@ export interface JSExecSettings {
mode: WorkflowExecuteMode;
}
export interface PartialAdditionalData {
executionId?: string;
restartExecutionId?: string;
restApiUrl: string;
instanceBaseUrl: string;
formWaitingBaseUrl: string;
webhookBaseUrl: string;
webhookWaitingBaseUrl: string;
webhookTestBaseUrl: string;
currentNodeParameters?: INodeParameters;
executionTimeoutTimestamp?: number;
userId?: string;
variables: IDataObject;
}
export interface DataRequestResponse {
export interface JsTaskData {
workflow: Omit<WorkflowParameters, 'nodeTypes'>;
inputData: ITaskDataConnections;
connectionInputData: INodeExecutionData[];
node: INode;
runExecutionData: IRunExecutionData;
runIndex: number;
itemIndex: number;
activeNodeName: string;
connectionInputData: INodeExecutionData[];
siblingParameters: INodeParameters;
mode: WorkflowExecuteMode;
envProviderState?: EnvProviderState;
envProviderState: EnvProviderState;
executeData?: IExecuteData;
defaultReturnRunIndex: number;
selfData: IDataObject;
@ -89,6 +71,8 @@ export class JsTaskRunner extends TaskRunner {
private readonly builtInsParser = new BuiltInsParser();
private readonly taskDataReconstruct = new DataRequestResponseReconstruct();
constructor(config: MainConfig, name = 'JS Task Runner') {
super({
taskType: 'javascript',
@ -115,33 +99,14 @@ export class JsTaskRunner extends TaskRunner {
? neededBuiltInsResult.result
: BuiltInsParserState.newNeedsAllDataState();
const data = await this.requestData<DataRequestResponse>(
const dataResponse = await this.requestData<DataRequestResponse>(
task.taskId,
neededBuiltIns.toDataRequestParams(),
);
/**
* We request node types only when we know a task needs all nodes, because
* needing all nodes means that the task relies on paired item functionality,
* which is the same requirement for needing node types.
*/
if (neededBuiltIns.needsAllNodes) {
const uniqueNodeTypes = new Map(
data.workflow.nodes.map((node) => [
`${node.type}|${node.typeVersion}`,
{ name: node.type, version: node.typeVersion },
]),
);
const data = this.reconstructTaskData(dataResponse);
const unknownNodeTypes = this.nodeTypes.onlyUnknown([...uniqueNodeTypes.values()]);
const nodeTypes = await this.requestNodeTypes<INodeTypeDescription[]>(
task.taskId,
unknownNodeTypes,
);
this.nodeTypes.addNodeTypeDescriptions(nodeTypes);
}
await this.requestNodeTypeIfNeeded(neededBuiltIns, data.workflow, task.taskId);
const workflowParams = data.workflow;
const workflow = new Workflow({
@ -201,7 +166,7 @@ export class JsTaskRunner extends TaskRunner {
private async runForAllItems(
taskId: string,
settings: JSExecSettings,
data: DataRequestResponse,
data: JsTaskData,
workflow: Workflow,
customConsole: CustomConsole,
): Promise<INodeExecutionData[]> {
@ -248,7 +213,7 @@ export class JsTaskRunner extends TaskRunner {
private async runForEachItem(
taskId: string,
settings: JSExecSettings,
data: DataRequestResponse,
data: JsTaskData,
workflow: Workflow,
customConsole: CustomConsole,
): Promise<INodeExecutionData[]> {
@ -315,7 +280,7 @@ export class JsTaskRunner extends TaskRunner {
return returnData;
}
private createDataProxy(data: DataRequestResponse, workflow: Workflow, itemIndex: number) {
private createDataProxy(data: JsTaskData, workflow: Workflow, itemIndex: number) {
return new WorkflowDataProxy(
workflow,
data.runExecutionData,
@ -359,4 +324,43 @@ export class JsTaskRunner extends TaskRunner {
return new ExecutionError({ message: JSON.stringify(error) });
}
private reconstructTaskData(response: DataRequestResponse): JsTaskData {
return {
...response,
connectionInputData: this.taskDataReconstruct.reconstructConnectionInputData(
response.inputData,
),
executeData: this.taskDataReconstruct.reconstructExecuteData(response),
};
}
private async requestNodeTypeIfNeeded(
neededBuiltIns: BuiltInsParserState,
workflow: JsTaskData['workflow'],
taskId: string,
) {
/**
* We request node types only when we know a task needs all nodes, because
* needing all nodes means that the task relies on paired item functionality,
* which is the same requirement for needing node types.
*/
if (neededBuiltIns.needsAllNodes) {
const uniqueNodeTypes = new Map(
workflow.nodes.map((node) => [
`${node.type}|${node.typeVersion}`,
{ name: node.type, version: node.typeVersion },
]),
);
const unknownNodeTypes = this.nodeTypes.onlyUnknown([...uniqueNodeTypes.values()]);
const nodeTypes = await this.requestNodeTypes<INodeTypeDescription[]>(
taskId,
unknownNodeTypes,
);
this.nodeTypes.addNodeTypeDescriptions(nodeTypes);
}
}
}

4
packages/@n8n/task-runner/src/runner-types.ts
View file

@ -8,6 +8,7 @@ import type {
INodeParameters,
IRunExecutionData,
ITaskDataConnections,
ITaskDataConnectionsSource,
IWorkflowExecuteAdditionalData,
Workflow,
WorkflowExecuteMode,
@ -29,17 +30,16 @@ export interface TaskDataRequestParams {
export interface DataRequestResponse {
workflow: Omit<WorkflowParameters, 'nodeTypes'>;
inputData: ITaskDataConnections;
connectionInputSource: ITaskDataConnectionsSource | null;
node: INode;
runExecutionData: IRunExecutionData;
runIndex: number;
itemIndex: number;
activeNodeName: string;
connectionInputData: INodeExecutionData[];
siblingParameters: INodeParameters;
mode: WorkflowExecuteMode;
envProviderState: EnvProviderState;
executeData?: IExecuteData;
defaultReturnRunIndex: number;
selfData: IDataObject;
contextNodeName: string;

2
packages/cli/src/commands/start.ts
View file

@ -221,7 +221,7 @@ export class Start extends BaseCommand {
}
const { taskRunners: taskRunnerConfig } = this.globalConfig;
if (!taskRunnerConfig.disabled) {
if (taskRunnerConfig.enabled) {
const { TaskRunnerModule } = await import('@/runners/task-runner-module');
const taskRunnerModule = Container.get(TaskRunnerModule);
await taskRunnerModule.start();

2
packages/cli/src/commands/worker.ts
View file

@ -113,7 +113,7 @@ export class Worker extends BaseCommand {
);
const { taskRunners: taskRunnerConfig } = this.globalConfig;
if (!taskRunnerConfig.disabled) {
if (taskRunnerConfig.enabled) {
const { TaskRunnerModule } = await import('@/runners/task-runner-module');
const taskRunnerModule = Container.get(TaskRunnerModule);
await taskRunnerModule.start();

2
packages/cli/src/runners/__tests__/task-runner-process.test.ts
View file

@ -22,7 +22,7 @@ require('child_process').spawn = spawnMock;
describe('TaskRunnerProcess', () => {
const logger = mockInstance(Logger);
const runnerConfig = mockInstance(TaskRunnersConfig);
runnerConfig.disabled = false;
runnerConfig.enabled = true;
runnerConfig.mode = 'internal_childprocess';
const authService = mock<TaskRunnerAuthService>();
let taskRunnerProcess = new TaskRunnerProcess(logger, runnerConfig, authService);

12
packages/cli/src/runners/runner-types.ts
View file

@ -5,18 +5,6 @@ import type WebSocket from 'ws';
import type { TaskRunner } from './task-broker.service';
import type { AuthlessRequest } from '../requests';
/**
* Specifies what data should be included for a task data request.
*/
export interface TaskDataRequestParams {
dataOfNodes: string[] | 'all';
prevNode: boolean;
/** Whether input data for the node should be included */
input: boolean;
/** Whether env provider's state should be included */
env: boolean;
}
export interface DisconnectAnalyzer {
determineDisconnectReason(runnerId: TaskRunner['id']): Promise<Error>;
}

338
packages/cli/src/runners/task-managers/__tests__/data-request-response-builder.test.ts
View file

@ -1,42 +1,10 @@
import type { TaskData } from '@n8n/task-runner';
import type { PartialAdditionalData, TaskData } from '@n8n/task-runner';
import { mock } from 'jest-mock-extended';
import type { IExecuteFunctions, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
import { type INode, type INodeExecutionData, type Workflow } from 'n8n-workflow';
import type { Workflow } from 'n8n-workflow';
import { DataRequestResponseBuilder } from '../data-request-response-builder';
const triggerNode: INode = mock<INode>({
name: 'Trigger',
});
const debugHelperNode: INode = mock<INode>({
name: 'DebugHelper',
});
const codeNode: INode = mock<INode>({
name: 'Code',
});
const workflow: TaskData['workflow'] = mock<Workflow>();
const debugHelperNodeOutItems: INodeExecutionData[] = [
{
json: {
uid: 'abb74fd4-bef2-4fae-9d53-ea24e9eb3032',
email: 'Dan.Schmidt31@yahoo.com',
firstname: 'Toni',
lastname: 'Schuster',
password: 'Q!D6C2',
},
pairedItem: {
item: 0,
},
},
];
const codeNodeInputItems: INodeExecutionData[] = debugHelperNodeOutItems;
const connectionInputData: TaskData['connectionInputData'] = codeNodeInputItems;
const envProviderState: TaskData['envProviderState'] = mock<TaskData['envProviderState']>({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
const additionalData = mock<IWorkflowExecuteAdditionalData>({
const additionalData = mock<PartialAdditionalData>({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
@ -50,275 +18,57 @@ const additionalData = mock<IWorkflowExecuteAdditionalData>({
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
});
const executeFunctions = mock<IExecuteFunctions>();
/**
* Drawn with https://asciiflow.com/#/
* Task data for an execution of the following WF:
* where ►► denotes the currently being executing node.
* ►►
* ┌───────────┐ ┌─────────────┐ ┌────────┐
* │ Trigger ├──►│ DebugHelper ├───►│ Code │
* └───────────┘ └─────────────┘ └────────┘
*/
const taskData: TaskData = {
executeFunctions,
workflow,
connectionInputData,
inputData: {
main: [codeNodeInputItems],
},
itemIndex: 0,
activeNodeName: codeNode.name,
contextNodeName: codeNode.name,
defaultReturnRunIndex: -1,
mode: 'manual',
envProviderState,
node: codeNode,
runExecutionData: {
startData: {
destinationNode: codeNode.name,
runNodeFilter: [triggerNode.name, debugHelperNode.name, codeNode.name],
},
resultData: {
runData: {
[triggerNode.name]: [
{
hints: [],
startTime: 1730313407328,
executionTime: 1,
source: [],
executionStatus: 'success',
data: {
main: [[]],
},
},
],
[debugHelperNode.name]: [
{
hints: [],
startTime: 1730313407330,
executionTime: 1,
source: [
{
previousNode: triggerNode.name,
},
],
executionStatus: 'success',
data: {
main: [debugHelperNodeOutItems],
},
},
],
},
pinData: {},
},
executionData: {
contextData: {},
nodeExecutionStack: [],
metadata: {},
waitingExecution: {
[codeNode.name]: {
'0': {
main: [codeNodeInputItems],
},
},
},
waitingExecutionSource: {
[codeNode.name]: {
'0': {
main: [
{
previousNode: debugHelperNode.name,
},
],
},
},
},
},
},
runIndex: 0,
selfData: {},
siblingParameters: {},
executeData: {
node: codeNode,
data: {
main: [codeNodeInputItems],
},
source: {
main: [
{
previousNode: debugHelperNode.name,
previousNodeOutput: 0,
},
],
},
},
const workflow: TaskData['workflow'] = mock<Workflow>({
id: '1',
name: 'Test Workflow',
active: true,
connectionsBySourceNode: {},
nodes: {},
pinData: {},
settings: {},
staticData: {},
});
const taskData = mock<TaskData>({
additionalData,
} as const;
workflow,
});
describe('DataRequestResponseBuilder', () => {
const allDataParam: DataRequestResponseBuilder['requestParams'] = {
dataOfNodes: 'all',
env: true,
input: true,
prevNode: true,
};
const builder = new DataRequestResponseBuilder();
const newRequestParam = (opts: Partial<DataRequestResponseBuilder['requestParams']>) => ({
...allDataParam,
...opts,
});
it('picks only specific properties for additional data', () => {
const result = builder.buildFromTaskData(taskData);
describe('all data', () => {
it('should build the runExecutionData as is when everything is requested', () => {
const dataRequestResponseBuilder = new DataRequestResponseBuilder(taskData, allDataParam);
const { runExecutionData } = dataRequestResponseBuilder.build();
expect(runExecutionData).toStrictEqual(taskData.runExecutionData);
expect(result.additionalData).toStrictEqual({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
variables: additionalData.variables,
webhookBaseUrl: 'http://localhost:5678/webhook',
webhookTestBaseUrl: 'http://localhost:5678/webhook-test',
webhookWaitingBaseUrl: 'http://localhost:5678/webhook-waiting',
executionId: '45844',
userId: '114984bc-44b3-4dd4-9b54-a4a8d34d51d5',
currentNodeParameters: undefined,
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
});
});
describe('envProviderState', () => {
it("should filter out envProviderState when it's not requested", () => {
const dataRequestResponseBuilder = new DataRequestResponseBuilder(
taskData,
newRequestParam({
env: false,
}),
);
it('picks only specific properties for workflow', () => {
const result = builder.buildFromTaskData(taskData);
const result = dataRequestResponseBuilder.build();
expect(result.envProviderState).toStrictEqual({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
});
});
describe('additionalData', () => {
it('picks only specific properties for additional data', () => {
const dataRequestResponseBuilder = new DataRequestResponseBuilder(taskData, allDataParam);
const result = dataRequestResponseBuilder.build();
expect(result.additionalData).toStrictEqual({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
webhookBaseUrl: 'http://localhost:5678/webhook',
webhookTestBaseUrl: 'http://localhost:5678/webhook-test',
webhookWaitingBaseUrl: 'http://localhost:5678/webhook-waiting',
executionId: '45844',
userId: '114984bc-44b3-4dd4-9b54-a4a8d34d51d5',
currentNodeParameters: undefined,
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
variables: additionalData.variables,
});
});
});
describe('input data', () => {
const allExceptInputParam = newRequestParam({
input: false,
});
it('drops input data from executeData', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.executeData).toStrictEqual({
node: taskData.executeData!.node,
source: taskData.executeData!.source,
data: {},
});
});
it('drops input data from result', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.inputData).toStrictEqual({});
});
it('drops input data from result', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.inputData).toStrictEqual({});
});
it('drops input data from connectionInputData', () => {
const result = new DataRequestResponseBuilder(taskData, allExceptInputParam).build();
expect(result.connectionInputData).toStrictEqual([]);
});
});
describe('nodes', () => {
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: ['Code'], prevNode: false }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: [codeNode.name], prevNode: false }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return only DebugHelper's data when only DebugHelper node is requested", () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: [debugHelperNode.name], prevNode: false }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return DebugHelper's data when only prevNode node is requested", () => {
const result = new DataRequestResponseBuilder(
taskData,
newRequestParam({ dataOfNodes: [], prevNode: true }),
).build();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
expect(result.workflow).toStrictEqual({
id: '1',
name: 'Test Workflow',
active: true,
connections: workflow.connectionsBySourceNode,
nodes: [],
pinData: workflow.pinData,
settings: workflow.settings,
staticData: workflow.staticData,
});
});
});

300
packages/cli/src/runners/task-managers/__tests__/data-request-response-stripper.test.ts Normal file
View file

@ -0,0 +1,300 @@
import type { DataRequestResponse, TaskDataRequestParams } from '@n8n/task-runner';
import { mock } from 'jest-mock-extended';
import type { IWorkflowExecuteAdditionalData } from 'n8n-workflow';
import { type INode, type INodeExecutionData } from 'n8n-workflow';
import { DataRequestResponseStripper } from '../data-request-response-stripper';
const triggerNode: INode = mock<INode>({
name: 'Trigger',
});
const debugHelperNode: INode = mock<INode>({
name: 'DebugHelper',
});
const codeNode: INode = mock<INode>({
name: 'Code',
});
const workflow: DataRequestResponse['workflow'] = mock<DataRequestResponse['workflow']>();
const debugHelperNodeOutItems: INodeExecutionData[] = [
{
json: {
uid: 'abb74fd4-bef2-4fae-9d53-ea24e9eb3032',
email: 'Dan.Schmidt31@yahoo.com',
firstname: 'Toni',
lastname: 'Schuster',
password: 'Q!D6C2',
},
pairedItem: {
item: 0,
},
},
];
const codeNodeInputItems: INodeExecutionData[] = debugHelperNodeOutItems;
const envProviderState: DataRequestResponse['envProviderState'] = mock<
DataRequestResponse['envProviderState']
>({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
const additionalData = mock<IWorkflowExecuteAdditionalData>({
formWaitingBaseUrl: 'http://localhost:5678/form-waiting',
instanceBaseUrl: 'http://localhost:5678/',
restApiUrl: 'http://localhost:5678/rest',
variables: {},
webhookBaseUrl: 'http://localhost:5678/webhook',
webhookTestBaseUrl: 'http://localhost:5678/webhook-test',
webhookWaitingBaseUrl: 'http://localhost:5678/webhook-waiting',
executionId: '45844',
userId: '114984bc-44b3-4dd4-9b54-a4a8d34d51d5',
currentNodeParameters: undefined,
executionTimeoutTimestamp: undefined,
restartExecutionId: undefined,
});
/**
* Drawn with https://asciiflow.com/#/
* Task data for an execution of the following WF:
* where ►► denotes the currently being executing node.
* ►►
* ┌───────────┐ ┌─────────────┐ ┌────────┐
* │ Trigger ├──►│ DebugHelper ├───►│ Code │
* └───────────┘ └─────────────┘ └────────┘
*/
const taskData: DataRequestResponse = {
workflow,
inputData: {
main: [codeNodeInputItems],
},
itemIndex: 0,
activeNodeName: codeNode.name,
contextNodeName: codeNode.name,
defaultReturnRunIndex: -1,
mode: 'manual',
envProviderState,
node: codeNode,
runExecutionData: {
startData: {
destinationNode: codeNode.name,
runNodeFilter: [triggerNode.name, debugHelperNode.name, codeNode.name],
},
resultData: {
runData: {
[triggerNode.name]: [
{
hints: [],
startTime: 1730313407328,
executionTime: 1,
source: [],
executionStatus: 'success',
data: {
main: [[]],
},
},
],
[debugHelperNode.name]: [
{
hints: [],
startTime: 1730313407330,
executionTime: 1,
source: [
{
previousNode: triggerNode.name,
},
],
executionStatus: 'success',
data: {
main: [debugHelperNodeOutItems],
},
},
],
},
pinData: {},
},
executionData: {
contextData: {},
nodeExecutionStack: [],
metadata: {},
waitingExecution: {
[codeNode.name]: {
'0': {
main: [codeNodeInputItems],
},
},
},
waitingExecutionSource: {
[codeNode.name]: {
'0': {
main: [
{
previousNode: debugHelperNode.name,
},
],
},
},
},
},
},
runIndex: 0,
selfData: {},
siblingParameters: {},
connectionInputSource: {
main: [
{
previousNode: debugHelperNode.name,
previousNodeOutput: 0,
},
],
},
additionalData,
} as const;
describe('DataRequestResponseStripper', () => {
const allDataParam: TaskDataRequestParams = {
dataOfNodes: 'all',
env: true,
input: true,
prevNode: true,
};
const newRequestParam = (opts: Partial<TaskDataRequestParams>) => ({
...allDataParam,
...opts,
});
describe('all data', () => {
it('should build the runExecutionData as is when everything is requested', () => {
const dataRequestResponseBuilder = new DataRequestResponseStripper(taskData, allDataParam);
const { runExecutionData } = dataRequestResponseBuilder.strip();
expect(runExecutionData).toStrictEqual(taskData.runExecutionData);
});
});
describe('envProviderState', () => {
it("should filter out envProviderState when it's not requested", () => {
const dataRequestResponseBuilder = new DataRequestResponseStripper(
taskData,
newRequestParam({
env: false,
}),
);
const result = dataRequestResponseBuilder.strip();
expect(result.envProviderState).toStrictEqual({
env: {},
isEnvAccessBlocked: false,
isProcessAvailable: true,
});
});
});
describe('input data', () => {
const allExceptInputParam = newRequestParam({
input: false,
});
it('drops input data from result', () => {
const result = new DataRequestResponseStripper(taskData, allExceptInputParam).strip();
expect(result.inputData).toStrictEqual({});
});
it('drops input data from result', () => {
const result = new DataRequestResponseStripper(taskData, allExceptInputParam).strip();
expect(result.inputData).toStrictEqual({});
});
});
describe('nodes', () => {
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: ['Code'], prevNode: false }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it('should return empty run data when only Code node is requested', () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: [codeNode.name], prevNode: false }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return only DebugHelper's data when only DebugHelper node is requested", () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: [debugHelperNode.name], prevNode: false }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
it("should return DebugHelper's data when only prevNode node is requested", () => {
const result = new DataRequestResponseStripper(
taskData,
newRequestParam({ dataOfNodes: [], prevNode: true }),
).strip();
expect(result.runExecutionData.resultData.runData).toStrictEqual({
[debugHelperNode.name]: taskData.runExecutionData.resultData.runData[debugHelperNode.name],
});
expect(result.runExecutionData.resultData.pinData).toStrictEqual({});
// executionData & startData contain only metadata --> returned as is
expect(result.runExecutionData.startData).toStrictEqual(taskData.runExecutionData.startData);
expect(result.runExecutionData.executionData).toStrictEqual(
taskData.runExecutionData.executionData,
);
});
});
describe('passthrough properties', () => {
test.each<Array<keyof DataRequestResponse>>([
['workflow'],
['connectionInputSource'],
['node'],
['runIndex'],
['itemIndex'],
['activeNodeName'],
['siblingParameters'],
['mode'],
['defaultReturnRunIndex'],
['selfData'],
['contextNodeName'],
['additionalData'],
])("it doesn't change %s", (propertyName) => {
const dataRequestResponseBuilder = new DataRequestResponseStripper(taskData, allDataParam);
const result = dataRequestResponseBuilder.strip();
expect(result[propertyName]).toBe(taskData[propertyName]);
});
});
});

190
packages/cli/src/runners/task-managers/data-request-response-builder.ts
View file

@ -1,63 +1,30 @@
import type {
DataRequestResponse,
BrokerMessage,
PartialAdditionalData,
TaskData,
} from '@n8n/task-runner';
import type {
EnvProviderState,
IExecuteData,
INodeExecutionData,
IPinData,
IRunData,
IRunExecutionData,
ITaskDataConnections,
IWorkflowExecuteAdditionalData,
Workflow,
WorkflowParameters,
} from 'n8n-workflow';
import type { DataRequestResponse, PartialAdditionalData, TaskData } from '@n8n/task-runner';
import type { IWorkflowExecuteAdditionalData, Workflow, WorkflowParameters } from 'n8n-workflow';
/**
* Builds the response to a data request coming from a Task Runner. Tries to minimize
* the amount of data that is sent to the runner by only providing what is requested.
* Transforms TaskData to DataRequestResponse. The main purpose of the
* transformation is to make sure there is no duplication in the data
* (e.g. connectionInputData and executeData.data can be derived from
* inputData).
*/
export class DataRequestResponseBuilder {
private requestedNodeNames = new Set<string>();
constructor(
private readonly taskData: TaskData,
private readonly requestParams: BrokerMessage.ToRequester.TaskDataRequest['requestParams'],
) {
this.requestedNodeNames = new Set(requestParams.dataOfNodes);
if (this.requestParams.prevNode && this.requestParams.dataOfNodes !== 'all') {
this.requestedNodeNames.add(this.determinePrevNodeName());
}
}
/**
* Builds a response to the data request
*/
build(): DataRequestResponse {
const { taskData: td } = this;
buildFromTaskData(taskData: TaskData): DataRequestResponse {
return {
workflow: this.buildWorkflow(td.workflow),
connectionInputData: this.buildConnectionInputData(td.connectionInputData),
inputData: this.buildInputData(td.inputData),
itemIndex: td.itemIndex,
activeNodeName: td.activeNodeName,
contextNodeName: td.contextNodeName,
defaultReturnRunIndex: td.defaultReturnRunIndex,
mode: td.mode,
envProviderState: this.buildEnvProviderState(td.envProviderState),
node: td.node, // The current node being executed
runExecutionData: this.buildRunExecutionData(td.runExecutionData),
runIndex: td.runIndex,
selfData: td.selfData,
siblingParameters: td.siblingParameters,
executeData: this.buildExecuteData(td.executeData),
additionalData: this.buildAdditionalData(td.additionalData),
workflow: this.buildWorkflow(taskData.workflow),
inputData: taskData.inputData,
connectionInputSource: taskData.executeData?.source ?? null,
itemIndex: taskData.itemIndex,
activeNodeName: taskData.activeNodeName,
contextNodeName: taskData.contextNodeName,
defaultReturnRunIndex: taskData.defaultReturnRunIndex,
mode: taskData.mode,
envProviderState: taskData.envProviderState,
node: taskData.node,
runExecutionData: taskData.runExecutionData,
runIndex: taskData.runIndex,
selfData: taskData.selfData,
siblingParameters: taskData.siblingParameters,
additionalData: this.buildAdditionalData(taskData.additionalData),
};
}
@ -80,86 +47,6 @@ export class DataRequestResponseBuilder {
};
}
private buildExecuteData(executeData: IExecuteData | undefined): IExecuteData | undefined {
if (executeData === undefined) {
return undefined;
}
return {
node: executeData.node, // The current node being executed
data: this.requestParams.input ? executeData.data : {},
source: executeData.source,
};
}
private buildRunExecutionData(runExecutionData: IRunExecutionData): IRunExecutionData {
if (this.requestParams.dataOfNodes === 'all') {
return runExecutionData;
}
return {
startData: runExecutionData.startData,
resultData: {
error: runExecutionData.resultData.error,
lastNodeExecuted: runExecutionData.resultData.lastNodeExecuted,
metadata: runExecutionData.resultData.metadata,
runData: this.buildRunData(runExecutionData.resultData.runData),
pinData: this.buildPinData(runExecutionData.resultData.pinData),
},
executionData: runExecutionData.executionData
? {
// TODO: Figure out what these two are and can they be filtered
contextData: runExecutionData.executionData?.contextData,
nodeExecutionStack: runExecutionData.executionData.nodeExecutionStack,
metadata: runExecutionData.executionData.metadata,
waitingExecution: runExecutionData.executionData.waitingExecution,
waitingExecutionSource: runExecutionData.executionData.waitingExecutionSource,
}
: undefined,
};
}
private buildRunData(runData: IRunData): IRunData {
return this.filterObjectByNodeNames(runData);
}
private buildPinData(pinData: IPinData | undefined): IPinData | undefined {
return pinData ? this.filterObjectByNodeNames(pinData) : undefined;
}
private buildEnvProviderState(envProviderState: EnvProviderState): EnvProviderState {
if (this.requestParams.env) {
// In case `isEnvAccessBlocked` = true, the provider state has already sanitized
// the environment variables and we can return it as is.
return envProviderState;
}
return {
env: {},
isEnvAccessBlocked: envProviderState.isEnvAccessBlocked,
isProcessAvailable: envProviderState.isProcessAvailable,
};
}
private buildInputData(inputData: ITaskDataConnections): ITaskDataConnections {
if (this.requestParams.input) {
return inputData;
}
return {};
}
private buildConnectionInputData(
connectionInputData: INodeExecutionData[],
): INodeExecutionData[] {
if (this.requestParams.input) {
return connectionInputData;
}
return [];
}
private buildWorkflow(workflow: Workflow): Omit<WorkflowParameters, 'nodeTypes'> {
return {
id: workflow.id,
@ -172,37 +59,4 @@ export class DataRequestResponseBuilder {
staticData: workflow.staticData,
};
}
/**
* Assuming the given `obj` is an object where the keys are node names,
* filters the object to only include the node names that are requested.
*/
private filterObjectByNodeNames<T extends Record<string, unknown>>(obj: T): T {
if (this.requestParams.dataOfNodes === 'all') {
return obj;
}
const filteredObj: T = {} as T;
for (const nodeName in obj) {
if (!Object.prototype.hasOwnProperty.call(obj, nodeName)) {
continue;
}
if (this.requestedNodeNames.has(nodeName)) {
filteredObj[nodeName] = obj[nodeName];
}
}
return filteredObj;
}
private determinePrevNodeName(): string {
const sourceData = this.taskData.executeData?.source?.main?.[0];
if (!sourceData) {
return '';
}
return sourceData.previousNode;
}
}

131
packages/cli/src/runners/task-managers/data-request-response-stripper.ts Normal file
View file

@ -0,0 +1,131 @@
import type { DataRequestResponse, BrokerMessage } from '@n8n/task-runner';
import type {
EnvProviderState,
IPinData,
IRunData,
IRunExecutionData,
ITaskDataConnections,
} from 'n8n-workflow';
/**
* Strips data from data request response based on the specified parameters
*/
export class DataRequestResponseStripper {
private requestedNodeNames = new Set<string>();
constructor(
private readonly dataResponse: DataRequestResponse,
private readonly stripParams: BrokerMessage.ToRequester.TaskDataRequest['requestParams'],
) {
this.requestedNodeNames = new Set(stripParams.dataOfNodes);
if (this.stripParams.prevNode && this.stripParams.dataOfNodes !== 'all') {
this.requestedNodeNames.add(this.determinePrevNodeName());
}
}
/**
* Builds a response to the data request
*/
strip(): DataRequestResponse {
const { dataResponse: dr } = this;
return {
...dr,
inputData: this.stripInputData(dr.inputData),
envProviderState: this.stripEnvProviderState(dr.envProviderState),
runExecutionData: this.stripRunExecutionData(dr.runExecutionData),
};
}
private stripRunExecutionData(runExecutionData: IRunExecutionData): IRunExecutionData {
if (this.stripParams.dataOfNodes === 'all') {
return runExecutionData;
}
return {
startData: runExecutionData.startData,
resultData: {
error: runExecutionData.resultData.error,
lastNodeExecuted: runExecutionData.resultData.lastNodeExecuted,
metadata: runExecutionData.resultData.metadata,
runData: this.stripRunData(runExecutionData.resultData.runData),
pinData: this.stripPinData(runExecutionData.resultData.pinData),
},
executionData: runExecutionData.executionData
? {
// TODO: Figure out what these two are and can they be stripped
contextData: runExecutionData.executionData?.contextData,
nodeExecutionStack: runExecutionData.executionData.nodeExecutionStack,
metadata: runExecutionData.executionData.metadata,
waitingExecution: runExecutionData.executionData.waitingExecution,
waitingExecutionSource: runExecutionData.executionData.waitingExecutionSource,
}
: undefined,
};
}
private stripRunData(runData: IRunData): IRunData {
return this.filterObjectByNodeNames(runData);
}
private stripPinData(pinData: IPinData | undefined): IPinData | undefined {
return pinData ? this.filterObjectByNodeNames(pinData) : undefined;
}
private stripEnvProviderState(envProviderState: EnvProviderState): EnvProviderState {
if (this.stripParams.env) {
// In case `isEnvAccessBlocked` = true, the provider state has already sanitized
// the environment variables and we can return it as is.
return envProviderState;
}
return {
env: {},
isEnvAccessBlocked: envProviderState.isEnvAccessBlocked,
isProcessAvailable: envProviderState.isProcessAvailable,
};
}
private stripInputData(inputData: ITaskDataConnections): ITaskDataConnections {
if (this.stripParams.input) {
return inputData;
}
return {};
}
/**
* Assuming the given `obj` is an object where the keys are node names,
* filters the object to only include the node names that are requested.
*/
private filterObjectByNodeNames<T extends Record<string, unknown>>(obj: T): T {
if (this.stripParams.dataOfNodes === 'all') {
return obj;
}
const filteredObj: T = {} as T;
for (const nodeName in obj) {
if (!Object.prototype.hasOwnProperty.call(obj, nodeName)) {
continue;
}
if (this.requestedNodeNames.has(nodeName)) {
filteredObj[nodeName] = obj[nodeName];
}
}
return filteredObj;
}
private determinePrevNodeName(): string {
const sourceData = this.dataResponse.connectionInputSource?.main?.[0];
if (!sourceData) {
return '';
}
return sourceData.previousNode;
}
}

33
packages/cli/src/runners/task-managers/task-manager.ts
View file

@ -1,5 +1,6 @@
import { TaskRunnersConfig } from '@n8n/config';
import type { TaskResultData, RequesterMessage, BrokerMessage, TaskData } from '@n8n/task-runner';
import { RPC_ALLOW_LIST } from '@n8n/task-runner';
import { DataRequestResponseReconstruct, RPC_ALLOW_LIST } from '@n8n/task-runner';
import type {
EnvProviderState,
IExecuteFunctions,
@ -17,11 +18,13 @@ import type {
} from 'n8n-workflow';
import { createResultOk, createResultError } from 'n8n-workflow';
import { nanoid } from 'nanoid';
import { Service } from 'typedi';
import * as a from 'node:assert/strict';
import Container, { Service } from 'typedi';
import { NodeTypes } from '@/node-types';
import { DataRequestResponseBuilder } from './data-request-response-builder';
import { DataRequestResponseStripper } from './data-request-response-stripper';
export type RequestAccept = (jobId: string) => void;
export type RequestReject = (reason: string) => void;
@ -56,6 +59,10 @@ export abstract class TaskManager {
tasks: Map<string, Task> = new Map();
private readonly runnerConfig = Container.get(TaskRunnersConfig);
private readonly dataResponseBuilder = new DataRequestResponseBuilder();
constructor(private readonly nodeTypes: NodeTypes) {}
async startTask<TData, TError>(
@ -237,14 +244,30 @@ export abstract class TaskManager {
return;
}
const dataRequestResponseBuilder = new DataRequestResponseBuilder(job.data, requestParams);
const requestedData = dataRequestResponseBuilder.build();
const dataRequestResponse = this.dataResponseBuilder.buildFromTaskData(job.data);
if (this.runnerConfig.assertDeduplicationOutput) {
const reconstruct = new DataRequestResponseReconstruct();
a.deepStrictEqual(
reconstruct.reconstructConnectionInputData(dataRequestResponse.inputData),
job.data.connectionInputData,
);
a.deepStrictEqual(
reconstruct.reconstructExecuteData(dataRequestResponse),
job.data.executeData,
);
}
const strippedData = new DataRequestResponseStripper(
dataRequestResponse,
requestParams,
).strip();
this.sendMessage({
type: 'requester:taskdataresponse',
taskId,
requestId,
data: requestedData,
data: strippedData,
});
}

2
packages/cli/src/runners/task-runner-module.ts
View file

@ -43,7 +43,7 @@ export class TaskRunnerModule {
}
async start() {
a.ok(!this.runnerConfig.disabled, 'Task runner is disabled');
a.ok(this.runnerConfig.enabled, 'Task runner is disabled');
await this.loadTaskManager();
await this.loadTaskRunnerServer();

2
packages/cli/test/integration/commands/worker.cmd.test.ts
View file

@ -26,7 +26,7 @@ import { mockInstance } from '../../shared/mocking';
config.set('executions.mode', 'queue');
config.set('binaryDataManager.availableModes', 'filesystem');
Container.get(TaskRunnersConfig).disabled = false;
Container.get(TaskRunnersConfig).enabled = true;
mockInstance(LoadNodesAndCredentials);
const binaryDataService = mockInstance(BinaryDataService);
const externalHooks = mockInstance(ExternalHooks);

4
packages/cli/test/integration/runners/task-runner-module.external.test.ts
View file

@ -18,14 +18,14 @@ describe('TaskRunnerModule in external mode', () => {
describe('start', () => {
it('should throw if the task runner is disabled', async () => {
runnerConfig.disabled = true;
runnerConfig.enabled = false;
// Act
await expect(module.start()).rejects.toThrow('Task runner is disabled');
});
it('should start the task runner', async () => {
runnerConfig.disabled = false;
runnerConfig.enabled = true;
// Act
await module.start();

4
packages/cli/test/integration/runners/task-runner-module.internal.test.ts
View file

@ -18,14 +18,14 @@ describe('TaskRunnerModule in internal_childprocess mode', () => {
describe('start', () => {
it('should throw if the task runner is disabled', async () => {
runnerConfig.disabled = true;
runnerConfig.enabled = false;
// Act
await expect(module.start()).rejects.toThrow('Task runner is disabled');
});
it('should start the task runner', async () => {
runnerConfig.disabled = false;
runnerConfig.enabled = true;
// Act
await module.start();

2
packages/cli/test/integration/runners/task-runner-process.test.ts
View file

@ -10,7 +10,7 @@ import { retryUntil } from '@test-integration/retry-until';
describe('TaskRunnerProcess', () => {
const authToken = 'token';
const runnerConfig = Container.get(TaskRunnersConfig);
runnerConfig.disabled = false;
runnerConfig.enabled = true;
runnerConfig.mode = 'internal_childprocess';
runnerConfig.authToken = authToken;
runnerConfig.port = 0; // Use any port

22
packages/core/src/NodeExecuteFunctions.ts
View file

@ -108,6 +108,7 @@ import type {
AiEvent,
ISupplyDataFunctions,
WebhookType,
SchedulingFunctions,
} from 'n8n-workflow';
import {
NodeConnectionType,
@ -172,6 +173,7 @@ import {
TriggerContext,
WebhookContext,
} from './node-execution-context';
import { ScheduledTaskManager } from './ScheduledTaskManager';
import { getSecretsProxy } from './Secrets';
import { SSHClientsManager } from './SSHClientsManager';
@ -3023,7 +3025,7 @@ const executionCancellationFunctions = (
},
});
const getRequestHelperFunctions = (
export const getRequestHelperFunctions = (
workflow: Workflow,
node: INode,
additionalData: IWorkflowExecuteAdditionalData,
@ -3343,11 +3345,19 @@ const getRequestHelperFunctions = (
};
};
const getSSHTunnelFunctions = (): SSHTunnelFunctions => ({
export const getSSHTunnelFunctions = (): SSHTunnelFunctions => ({
getSSHClient: async (credentials) =>
await Container.get(SSHClientsManager).getClient(credentials),
});
export const getSchedulingFunctions = (workflow: Workflow): SchedulingFunctions => {
const scheduledTaskManager = Container.get(ScheduledTaskManager);
return {
registerCron: (cronExpression, onTick) =>
scheduledTaskManager.registerCron(workflow, cronExpression, onTick),
};
};
const getAllowedPaths = () => {
const restrictFileAccessTo = process.env[RESTRICT_FILE_ACCESS_TO];
if (!restrictFileAccessTo) {
@ -3414,7 +3424,7 @@ export function isFilePathBlocked(filePath: string): boolean {
return false;
}
const getFileSystemHelperFunctions = (node: INode): FileSystemHelperFunctions => ({
export const getFileSystemHelperFunctions = (node: INode): FileSystemHelperFunctions => ({
async createReadStream(filePath) {
try {
await fsAccess(filePath);
@ -3450,7 +3460,7 @@ const getFileSystemHelperFunctions = (node: INode): FileSystemHelperFunctions =>
},
});
const getNodeHelperFunctions = (
export const getNodeHelperFunctions = (
{ executionId }: IWorkflowExecuteAdditionalData,
workflowId: string,
): NodeHelperFunctions => ({
@ -3458,7 +3468,7 @@ const getNodeHelperFunctions = (
await copyBinaryFile(workflowId, executionId!, filePath, fileName, mimeType),
});
const getBinaryHelperFunctions = (
export const getBinaryHelperFunctions = (
{ executionId }: IWorkflowExecuteAdditionalData,
workflowId: string,
): BinaryHelperFunctions => ({
@ -3476,7 +3486,7 @@ const getBinaryHelperFunctions = (
},
});
const getCheckProcessedHelperFunctions = (
export const getCheckProcessedHelperFunctions = (
workflow: Workflow,
node: INode,
): DeduplicationHelperFunctions => ({

14
packages/core/src/node-execution-context/execute-single-context.ts
View file

@ -27,13 +27,13 @@ import {
continueOnFail,
getAdditionalKeys,
getBinaryDataBuffer,
getBinaryHelperFunctions,
getCredentials,
getNodeParameter,
getRequestHelperFunctions,
returnJsonArray,
} from '@/NodeExecuteFunctions';
import { BinaryHelpers } from './helpers/binary-helpers';
import { RequestHelpers } from './helpers/request-helpers';
import { NodeExecutionContext } from './node-execution-context';
export class ExecuteSingleContext extends NodeExecutionContext implements IExecuteSingleFunctions {
@ -57,8 +57,14 @@ export class ExecuteSingleContext extends NodeExecutionContext implements IExecu
this.helpers = {
createDeferredPromise,
returnJsonArray,
...new BinaryHelpers(workflow, additionalData).exported,
...new RequestHelpers(this, workflow, node, additionalData).exported,
...getRequestHelperFunctions(
workflow,
node,
additionalData,
runExecutionData,
connectionInputData,
),
...getBinaryHelperFunctions(additionalData, workflow.id),
assertBinaryData: (propertyName, inputIndex = 0) =>
assertBinaryData(inputData, node, itemIndex, propertyName, inputIndex),

136
packages/core/src/node-execution-context/helpers/__tests__/binary-helpers.test.ts
View file

@ -1,136 +0,0 @@
import FileType from 'file-type';
import { IncomingMessage, type ClientRequest } from 'http';
import { mock } from 'jest-mock-extended';
import type { Workflow, IWorkflowExecuteAdditionalData, IBinaryData } from 'n8n-workflow';
import type { Socket } from 'net';
import { Container } from 'typedi';
import { BinaryDataService } from '@/BinaryData/BinaryData.service';
import { BinaryHelpers } from '../binary-helpers';
jest.mock('file-type');
describe('BinaryHelpers', () => {
let binaryDataService = mock<BinaryDataService>();
Container.set(BinaryDataService, binaryDataService);
const workflow = mock<Workflow>({ id: '123' });
const additionalData = mock<IWorkflowExecuteAdditionalData>({ executionId: '456' });
const binaryHelpers = new BinaryHelpers(workflow, additionalData);
beforeEach(() => {
jest.clearAllMocks();
binaryDataService.store.mockImplementation(
async (_workflowId, _executionId, _buffer, value) => value,
);
});
describe('getBinaryPath', () => {
it('should call getPath method of BinaryDataService', () => {
binaryHelpers.getBinaryPath('mock-binary-data-id');
expect(binaryDataService.getPath).toHaveBeenCalledWith('mock-binary-data-id');
});
});
describe('getBinaryMetadata', () => {
it('should call getMetadata method of BinaryDataService', async () => {
await binaryHelpers.getBinaryMetadata('mock-binary-data-id');
expect(binaryDataService.getMetadata).toHaveBeenCalledWith('mock-binary-data-id');
});
});
describe('getBinaryStream', () => {
it('should call getStream method of BinaryDataService', async () => {
await binaryHelpers.getBinaryStream('mock-binary-data-id');
expect(binaryDataService.getAsStream).toHaveBeenCalledWith('mock-binary-data-id', undefined);
});
});
describe('prepareBinaryData', () => {
it('should guess the mime type and file extension if not provided', async () => {
const buffer = Buffer.from('test');
const fileTypeData = { mime: 'application/pdf', ext: 'pdf' };
(FileType.fromBuffer as jest.Mock).mockResolvedValue(fileTypeData);
const binaryData = await binaryHelpers.prepareBinaryData(buffer);
expect(binaryData.mimeType).toEqual('application/pdf');
expect(binaryData.fileExtension).toEqual('pdf');
expect(binaryData.fileType).toEqual('pdf');
expect(binaryData.fileName).toBeUndefined();
expect(binaryData.directory).toBeUndefined();
expect(binaryDataService.store).toHaveBeenCalledWith(
workflow.id,
additionalData.executionId!,
buffer,
binaryData,
);
});
it('should use the provided mime type and file extension if provided', async () => {
const buffer = Buffer.from('test');
const mimeType = 'application/octet-stream';
const binaryData = await binaryHelpers.prepareBinaryData(buffer, undefined, mimeType);
expect(binaryData.mimeType).toEqual(mimeType);
expect(binaryData.fileExtension).toEqual('bin');
expect(binaryData.fileType).toBeUndefined();
expect(binaryData.fileName).toBeUndefined();
expect(binaryData.directory).toBeUndefined();
expect(binaryDataService.store).toHaveBeenCalledWith(
workflow.id,
additionalData.executionId!,
buffer,
binaryData,
);
});
const mockSocket = mock<Socket>({ readableHighWaterMark: 0 });
it('should use the contentDisposition.filename, responseUrl, and contentType properties to set the fileName, directory, and mimeType properties of the binaryData object', async () => {
const incomingMessage = new IncomingMessage(mockSocket);
incomingMessage.contentDisposition = { filename: 'test.txt', type: 'attachment' };
incomingMessage.contentType = 'text/plain';
incomingMessage.responseUrl = 'https://example.com/test.txt';
const binaryData = await binaryHelpers.prepareBinaryData(incomingMessage);
expect(binaryData.fileName).toEqual('test.txt');
expect(binaryData.fileType).toEqual('text');
expect(binaryData.directory).toBeUndefined();
expect(binaryData.mimeType).toEqual('text/plain');
expect(binaryData.fileExtension).toEqual('txt');
});
it('should use the req.path property to set the fileName property of the binaryData object if contentDisposition.filename and responseUrl are not provided', async () => {
const incomingMessage = new IncomingMessage(mockSocket);
incomingMessage.contentType = 'text/plain';
incomingMessage.req = mock<ClientRequest>({ path: '/test.txt' });
const binaryData = await binaryHelpers.prepareBinaryData(incomingMessage);
expect(binaryData.fileName).toEqual('test.txt');
expect(binaryData.directory).toBeUndefined();
expect(binaryData.mimeType).toEqual('text/plain');
expect(binaryData.fileExtension).toEqual('txt');
});
});
describe('setBinaryDataBuffer', () => {
it('should call store method of BinaryDataService', async () => {
const binaryData = mock<IBinaryData>();
const bufferOrStream = mock<Buffer>();
await binaryHelpers.setBinaryDataBuffer(binaryData, bufferOrStream);
expect(binaryDataService.store).toHaveBeenCalledWith(
workflow.id,
additionalData.executionId,
bufferOrStream,
binaryData,
);
});
});
});

33
packages/core/src/node-execution-context/helpers/__tests__/scheduling-helpers.test.ts
View file

@ -1,33 +0,0 @@
import { mock } from 'jest-mock-extended';
import type { Workflow } from 'n8n-workflow';
import { Container } from 'typedi';
import { ScheduledTaskManager } from '@/ScheduledTaskManager';
import { SchedulingHelpers } from '../scheduling-helpers';
describe('SchedulingHelpers', () => {
const scheduledTaskManager = mock<ScheduledTaskManager>();
Container.set(ScheduledTaskManager, scheduledTaskManager);
const workflow = mock<Workflow>();
const schedulingHelpers = new SchedulingHelpers(workflow);
beforeEach(() => {
jest.clearAllMocks();
});
describe('registerCron', () => {
it('should call registerCron method of ScheduledTaskManager', () => {
const cronExpression = '* * * * * *';
const onTick = jest.fn();
schedulingHelpers.registerCron(cronExpression, onTick);
expect(scheduledTaskManager.registerCron).toHaveBeenCalledWith(
workflow,
cronExpression,
onTick,
);
});
});
});

32
packages/core/src/node-execution-context/helpers/__tests__/ssh-tunnel-helpers.test.ts
View file

@ -1,32 +0,0 @@
import { mock } from 'jest-mock-extended';
import type { SSHCredentials } from 'n8n-workflow';
import type { Client } from 'ssh2';
import { Container } from 'typedi';
import { SSHClientsManager } from '@/SSHClientsManager';
import { SSHTunnelHelpers } from '../ssh-tunnel-helpers';
describe('SSHTunnelHelpers', () => {
const sshClientsManager = mock<SSHClientsManager>();
Container.set(SSHClientsManager, sshClientsManager);
const sshTunnelHelpers = new SSHTunnelHelpers();
beforeEach(() => {
jest.clearAllMocks();
});
describe('getSSHClient', () => {
const credentials = mock<SSHCredentials>();
it('should call SSHClientsManager.getClient with the given credentials', async () => {
const mockClient = mock<Client>();
sshClientsManager.getClient.mockResolvedValue(mockClient);
const client = await sshTunnelHelpers.getSSHClient(credentials);
expect(sshClientsManager.getClient).toHaveBeenCalledWith(credentials);
expect(client).toBe(mockClient);
});
});
});

148
packages/core/src/node-execution-context/helpers/binary-helpers.ts
View file

@ -1,148 +0,0 @@
import FileType from 'file-type';
import { IncomingMessage } from 'http';
import MimeTypes from 'mime-types';
import { ApplicationError, fileTypeFromMimeType } from 'n8n-workflow';
import type {
BinaryHelperFunctions,
IWorkflowExecuteAdditionalData,
Workflow,
IBinaryData,
} from 'n8n-workflow';
import path from 'path';
import type { Readable } from 'stream';
import Container from 'typedi';
import { BinaryDataService } from '@/BinaryData/BinaryData.service';
import { binaryToBuffer } from '@/BinaryData/utils';
// eslint-disable-next-line import/no-cycle
import { binaryToString } from '@/NodeExecuteFunctions';
export class BinaryHelpers {
private readonly binaryDataService = Container.get(BinaryDataService);
constructor(
private readonly workflow: Workflow,
private readonly additionalData: IWorkflowExecuteAdditionalData,
) {}
get exported(): BinaryHelperFunctions {
return {
getBinaryPath: this.getBinaryPath.bind(this),
getBinaryMetadata: this.getBinaryMetadata.bind(this),
getBinaryStream: this.getBinaryStream.bind(this),
binaryToBuffer,
binaryToString,
prepareBinaryData: this.prepareBinaryData.bind(this),
setBinaryDataBuffer: this.setBinaryDataBuffer.bind(this),
copyBinaryFile: this.copyBinaryFile.bind(this),
};
}
getBinaryPath(binaryDataId: string) {
return this.binaryDataService.getPath(binaryDataId);
}
async getBinaryMetadata(binaryDataId: string) {
return await this.binaryDataService.getMetadata(binaryDataId);
}
async getBinaryStream(binaryDataId: string, chunkSize?: number) {
return await this.binaryDataService.getAsStream(binaryDataId, chunkSize);
}
// eslint-disable-next-line complexity
async prepareBinaryData(binaryData: Buffer | Readable, filePath?: string, mimeType?: string) {
let fileExtension: string | undefined;
if (binaryData instanceof IncomingMessage) {
if (!filePath) {
try {
const { responseUrl } = binaryData;
filePath =
binaryData.contentDisposition?.filename ??
((responseUrl && new URL(responseUrl).pathname) ?? binaryData.req?.path)?.slice(1);
} catch {}
}
if (!mimeType) {
mimeType = binaryData.contentType;
}
}
if (!mimeType) {
// If no mime type is given figure it out
if (filePath) {
// Use file path to guess mime type
const mimeTypeLookup = MimeTypes.lookup(filePath);
if (mimeTypeLookup) {
mimeType = mimeTypeLookup;
}
}
if (!mimeType) {
if (Buffer.isBuffer(binaryData)) {
// Use buffer to guess mime type
const fileTypeData = await FileType.fromBuffer(binaryData);
if (fileTypeData) {
mimeType = fileTypeData.mime;
fileExtension = fileTypeData.ext;
}
} else if (binaryData instanceof IncomingMessage) {
mimeType = binaryData.headers['content-type'];
} else {
// TODO: detect filetype from other kind of streams
}
}
}
if (!fileExtension && mimeType) {
fileExtension = MimeTypes.extension(mimeType) || undefined;
}
if (!mimeType) {
// Fall back to text
mimeType = 'text/plain';
}
const returnData: IBinaryData = {
mimeType,
fileType: fileTypeFromMimeType(mimeType),
fileExtension,
data: '',
};
if (filePath) {
if (filePath.includes('?')) {
// Remove maybe present query parameters
filePath = filePath.split('?').shift();
}
const filePathParts = path.parse(filePath as string);
if (filePathParts.dir !== '') {
returnData.directory = filePathParts.dir;
}
returnData.fileName = filePathParts.base;
// Remove the dot
const extractedFileExtension = filePathParts.ext.slice(1);
if (extractedFileExtension) {
returnData.fileExtension = extractedFileExtension;
}
}
return await this.setBinaryDataBuffer(returnData, binaryData);
}
async setBinaryDataBuffer(binaryData: IBinaryData, bufferOrStream: Buffer | Readable) {
return await this.binaryDataService.store(
this.workflow.id,
this.additionalData.executionId!,
bufferOrStream,
binaryData,
);
}
async copyBinaryFile(): Promise<never> {
throw new ApplicationError('`copyBinaryFile` has been removed. Please upgrade this node.');
}
}

381
packages/core/src/node-execution-context/helpers/request-helpers.ts
View file

@ -1,381 +0,0 @@
import { createHash } from 'crypto';
import { pick } from 'lodash';
import { jsonParse, NodeOperationError, sleep } from 'n8n-workflow';
import type {
RequestHelperFunctions,
IAdditionalCredentialOptions,
IAllExecuteFunctions,
IExecuteData,
IHttpRequestOptions,
IN8nHttpFullResponse,
IN8nHttpResponse,
INode,
INodeExecutionData,
IOAuth2Options,
IRequestOptions,
IRunExecutionData,
IWorkflowDataProxyAdditionalKeys,
IWorkflowExecuteAdditionalData,
NodeParameterValueType,
PaginationOptions,
Workflow,
WorkflowExecuteMode,
} from 'n8n-workflow';
import { Readable } from 'stream';
// eslint-disable-next-line import/no-cycle
import {
applyPaginationRequestData,
binaryToString,
httpRequest,
httpRequestWithAuthentication,
proxyRequestToAxios,
requestOAuth1,
requestOAuth2,
requestWithAuthentication,
validateUrl,
} from '@/NodeExecuteFunctions';
export class RequestHelpers {
constructor(
private readonly context: IAllExecuteFunctions,
private readonly workflow: Workflow,
private readonly node: INode,
private readonly additionalData: IWorkflowExecuteAdditionalData,
private readonly runExecutionData: IRunExecutionData | null = null,
private readonly connectionInputData: INodeExecutionData[] = [],
) {}
get exported(): RequestHelperFunctions {
return {
httpRequest,
httpRequestWithAuthentication: this.httpRequestWithAuthentication.bind(this),
requestWithAuthenticationPaginated: this.requestWithAuthenticationPaginated.bind(this),
request: this.request.bind(this),
requestWithAuthentication: this.requestWithAuthentication.bind(this),
requestOAuth1: this.requestOAuth1.bind(this),
requestOAuth2: this.requestOAuth2.bind(this),
};
}
get httpRequest() {
return httpRequest;
}
async httpRequestWithAuthentication(
credentialsType: string,
requestOptions: IHttpRequestOptions,
additionalCredentialOptions?: IAdditionalCredentialOptions,
) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-return
return await httpRequestWithAuthentication.call(
this.context,
credentialsType,
requestOptions,
this.workflow,
this.node,
this.additionalData,
additionalCredentialOptions,
);
}
// eslint-disable-next-line complexity
async requestWithAuthenticationPaginated(
requestOptions: IRequestOptions,
itemIndex: number,
paginationOptions: PaginationOptions,
credentialsType?: string,
additionalCredentialOptions?: IAdditionalCredentialOptions,
): Promise<unknown[]> {
const responseData = [];
if (!requestOptions.qs) {
requestOptions.qs = {};
}
requestOptions.resolveWithFullResponse = true;
requestOptions.simple = false;
let tempResponseData: IN8nHttpFullResponse;
let makeAdditionalRequest: boolean;
let paginateRequestData: PaginationOptions['request'];
const runIndex = 0;
const additionalKeys = {
$request: requestOptions,
$response: {} as IN8nHttpFullResponse,
$version: this.node.typeVersion,
$pageCount: 0,
};
const executeData: IExecuteData = {
data: {},
node: this.node,
source: null,
};
const hashData = {
identicalCount: 0,
previousLength: 0,
previousHash: '',
};
do {
paginateRequestData = this.getResolvedValue(
paginationOptions.request as unknown as NodeParameterValueType,
itemIndex,
runIndex,
executeData,
additionalKeys,
false,
) as object as PaginationOptions['request'];
const tempRequestOptions = applyPaginationRequestData(requestOptions, paginateRequestData);
if (!validateUrl(tempRequestOptions.uri as string)) {
throw new NodeOperationError(
this.node,
`'${paginateRequestData.url}' is not a valid URL.`,
{
itemIndex,
runIndex,
type: 'invalid_url',
},
);
}
if (credentialsType) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
tempResponseData = await this.requestWithAuthentication(
credentialsType,
tempRequestOptions,
additionalCredentialOptions,
);
} else {
// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
tempResponseData = await this.request(tempRequestOptions);
}
const newResponse: IN8nHttpFullResponse = Object.assign(
{
body: {},
headers: {},
statusCode: 0,
},
pick(tempResponseData, ['body', 'headers', 'statusCode']),
);
let contentBody: Exclude<IN8nHttpResponse, Buffer>;
if (newResponse.body instanceof Readable && paginationOptions.binaryResult !== true) {
// Keep the original string version that we can use it to hash if needed
contentBody = await binaryToString(newResponse.body as Buffer | Readable);
const responseContentType = newResponse.headers['content-type']?.toString() ?? '';
if (responseContentType.includes('application/json')) {
newResponse.body = jsonParse(contentBody, { fallbackValue: {} });
} else {
newResponse.body = contentBody;
}
tempResponseData.__bodyResolved = true;
tempResponseData.body = newResponse.body;
} else {
contentBody = newResponse.body;
}
if (paginationOptions.binaryResult !== true || tempResponseData.headers.etag) {
// If the data is not binary (and so not a stream), or an etag is present,
// we check via etag or hash if identical data is received
let contentLength = 0;
if ('content-length' in tempResponseData.headers) {
contentLength = parseInt(tempResponseData.headers['content-length'] as string) || 0;
}
if (hashData.previousLength === contentLength) {
let hash: string;
if (tempResponseData.headers.etag) {
// If an etag is provided, we use it as "hash"
hash = tempResponseData.headers.etag as string;
} else {
// If there is no etag, we calculate a hash from the data in the body
if (typeof contentBody !== 'string') {
contentBody = JSON.stringify(contentBody);
}
hash = createHash('md5').update(contentBody).digest('base64');
}
if (hashData.previousHash === hash) {
hashData.identicalCount += 1;
if (hashData.identicalCount > 2) {
// Length was identical 5x and hash 3x
throw new NodeOperationError(
this.node,
'The returned response was identical 5x, so requests got stopped',
{
itemIndex,
description:
'Check if "Pagination Completed When" has been configured correctly.',
},
);
}
} else {
hashData.identicalCount = 0;
}
hashData.previousHash = hash;
} else {
hashData.identicalCount = 0;
}
hashData.previousLength = contentLength;
}
responseData.push(tempResponseData);
additionalKeys.$response = newResponse;
additionalKeys.$pageCount = additionalKeys.$pageCount + 1;
const maxRequests = this.getResolvedValue(
paginationOptions.maxRequests,
itemIndex,
runIndex,
executeData,
additionalKeys,
false,
) as number;
if (maxRequests && additionalKeys.$pageCount >= maxRequests) {
break;
}
makeAdditionalRequest = this.getResolvedValue(
paginationOptions.continue,
itemIndex,
runIndex,
executeData,
additionalKeys,
false,
) as boolean;
if (makeAdditionalRequest) {
if (paginationOptions.requestInterval) {
const requestInterval = this.getResolvedValue(
paginationOptions.requestInterval,
itemIndex,
runIndex,
executeData,
additionalKeys,
false,
) as number;
await sleep(requestInterval);
}
if (tempResponseData.statusCode < 200 || tempResponseData.statusCode >= 300) {
// We have it configured to let all requests pass no matter the response code
// via "requestOptions.simple = false" to not by default fail if it is for example
// configured to stop on 404 response codes. For that reason we have to throw here
// now an error manually if the response code is not a success one.
let data = tempResponseData.body;
if (data instanceof Readable && paginationOptions.binaryResult !== true) {
data = await binaryToString(data as Buffer | Readable);
} else if (typeof data === 'object') {
data = JSON.stringify(data);
}
throw Object.assign(new Error(`${tempResponseData.statusCode} - "${data?.toString()}"`), {
statusCode: tempResponseData.statusCode,
error: data,
isAxiosError: true,
response: {
headers: tempResponseData.headers,
status: tempResponseData.statusCode,
statusText: tempResponseData.statusMessage,
},
});
}
}
} while (makeAdditionalRequest);
return responseData;
}
async request(uriOrObject: string | IRequestOptions, options?: IRequestOptions) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-return
return await proxyRequestToAxios(
this.workflow,
this.additionalData,
this.node,
uriOrObject,
options,
);
}
async requestWithAuthentication(
credentialsType: string,
requestOptions: IRequestOptions,
additionalCredentialOptions?: IAdditionalCredentialOptions,
itemIndex?: number,
) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-return
return await requestWithAuthentication.call(
this.context,
credentialsType,
requestOptions,
this.workflow,
this.node,
this.additionalData,
additionalCredentialOptions,
itemIndex,
);
}
async requestOAuth1(credentialsType: string, requestOptions: IRequestOptions) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-return
return await requestOAuth1.call(this.context, credentialsType, requestOptions);
}
async requestOAuth2(
credentialsType: string,
requestOptions: IRequestOptions,
oAuth2Options?: IOAuth2Options,
) {
// eslint-disable-next-line @typescript-eslint/no-unsafe-return
return await requestOAuth2.call(
this.context,
credentialsType,
requestOptions,
this.node,
this.additionalData,
oAuth2Options,
);
}
private getResolvedValue(
parameterValue: NodeParameterValueType,
itemIndex: number,
runIndex: number,
executeData: IExecuteData,
additionalKeys?: IWorkflowDataProxyAdditionalKeys,
returnObjectAsString = false,
): NodeParameterValueType {
const mode: WorkflowExecuteMode = 'internal';
if (
typeof parameterValue === 'object' ||
(typeof parameterValue === 'string' && parameterValue.charAt(0) === '=')
) {
return this.workflow.expression.getParameterValue(
parameterValue,
this.runExecutionData,
runIndex,
itemIndex,
this.node.name,
this.connectionInputData,
mode,
additionalKeys ?? {},
executeData,
returnObjectAsString,
);
}
return parameterValue;
}
}

20
packages/core/src/node-execution-context/helpers/scheduling-helpers.ts
View file

@ -1,20 +0,0 @@
import type { CronExpression, Workflow, SchedulingFunctions } from 'n8n-workflow';
import { Container } from 'typedi';
import { ScheduledTaskManager } from '@/ScheduledTaskManager';
export class SchedulingHelpers {
private readonly scheduledTaskManager = Container.get(ScheduledTaskManager);
constructor(private readonly workflow: Workflow) {}
get exported(): SchedulingFunctions {
return {
registerCron: this.registerCron.bind(this),
};
}
registerCron(cronExpression: CronExpression, onTick: () => void) {
this.scheduledTaskManager.registerCron(this.workflow, cronExpression, onTick);
}
}

18
packages/core/src/node-execution-context/helpers/ssh-tunnel-helpers.ts
View file

@ -1,18 +0,0 @@
import type { SSHCredentials, SSHTunnelFunctions } from 'n8n-workflow';
import { Container } from 'typedi';
import { SSHClientsManager } from '@/SSHClientsManager';
export class SSHTunnelHelpers {
private readonly sshClientsManager = Container.get(SSHClientsManager);
get exported(): SSHTunnelFunctions {
return {
getSSHClient: this.getSSHClient.bind(this),
};
}
async getSSHClient(credentials: SSHCredentials) {
return await this.sshClientsManager.getClient(credentials);
}
}

4
packages/core/src/node-execution-context/hook-context.ts
View file

@ -21,10 +21,10 @@ import {
getCredentials,
getNodeParameter,
getNodeWebhookUrl,
getRequestHelperFunctions,
getWebhookDescription,
} from '@/NodeExecuteFunctions';
import { RequestHelpers } from './helpers/request-helpers';
import { NodeExecutionContext } from './node-execution-context';
export class HookContext extends NodeExecutionContext implements IHookFunctions {
@ -40,7 +40,7 @@ export class HookContext extends NodeExecutionContext implements IHookFunctions
) {
super(workflow, node, additionalData, mode);
this.helpers = new RequestHelpers(this, workflow, node, additionalData);
this.helpers = getRequestHelperFunctions(workflow, node, additionalData);
}
getActivationMode() {

14
packages/core/src/node-execution-context/load-options-context.ts
View file

@ -13,10 +13,14 @@ import type {
import { extractValue } from '@/ExtractValue';
// eslint-disable-next-line import/no-cycle
import { getAdditionalKeys, getCredentials, getNodeParameter } from '@/NodeExecuteFunctions';
import {
getAdditionalKeys,
getCredentials,
getNodeParameter,
getRequestHelperFunctions,
getSSHTunnelFunctions,
} from '@/NodeExecuteFunctions';
import { RequestHelpers } from './helpers/request-helpers';
import { SSHTunnelHelpers } from './helpers/ssh-tunnel-helpers';
import { NodeExecutionContext } from './node-execution-context';
export class LoadOptionsContext extends NodeExecutionContext implements ILoadOptionsFunctions {
@ -31,8 +35,8 @@ export class LoadOptionsContext extends NodeExecutionContext implements ILoadOpt
super(workflow, node, additionalData, 'internal');
this.helpers = {
...new RequestHelpers(this, workflow, node, additionalData).exported,
...new SSHTunnelHelpers().exported,
...getSSHTunnelFunctions(),
...getRequestHelperFunctions(workflow, node, additionalData),
};
}

12
packages/core/src/node-execution-context/poll-context.ts
View file

@ -16,14 +16,14 @@ import { ApplicationError, createDeferredPromise } from 'n8n-workflow';
// eslint-disable-next-line import/no-cycle
import {
getAdditionalKeys,
getBinaryHelperFunctions,
getCredentials,
getNodeParameter,
getRequestHelperFunctions,
getSchedulingFunctions,
returnJsonArray,
} from '@/NodeExecuteFunctions';
import { BinaryHelpers } from './helpers/binary-helpers';
import { RequestHelpers } from './helpers/request-helpers';
import { SchedulingHelpers } from './helpers/scheduling-helpers';
import { NodeExecutionContext } from './node-execution-context';
const throwOnEmit = () => {
@ -51,9 +51,9 @@ export class PollContext extends NodeExecutionContext implements IPollFunctions
this.helpers = {
createDeferredPromise,
returnJsonArray,
...new BinaryHelpers(workflow, additionalData).exported,
...new RequestHelpers(this, workflow, node, additionalData).exported,
...new SchedulingHelpers(workflow).exported,
...getRequestHelperFunctions(workflow, node, additionalData),
...getBinaryHelperFunctions(additionalData, workflow.id),
...getSchedulingFunctions(workflow),
};
}

16
packages/core/src/node-execution-context/trigger-context.ts
View file

@ -16,15 +16,15 @@ import { ApplicationError, createDeferredPromise } from 'n8n-workflow';
// eslint-disable-next-line import/no-cycle
import {
getAdditionalKeys,
getBinaryHelperFunctions,
getCredentials,
getNodeParameter,
getRequestHelperFunctions,
getSchedulingFunctions,
getSSHTunnelFunctions,
returnJsonArray,
} from '@/NodeExecuteFunctions';
import { BinaryHelpers } from './helpers/binary-helpers';
import { RequestHelpers } from './helpers/request-helpers';
import { SchedulingHelpers } from './helpers/scheduling-helpers';
import { SSHTunnelHelpers } from './helpers/ssh-tunnel-helpers';
import { NodeExecutionContext } from './node-execution-context';
const throwOnEmit = () => {
@ -52,10 +52,10 @@ export class TriggerContext extends NodeExecutionContext implements ITriggerFunc
this.helpers = {
createDeferredPromise,
returnJsonArray,
...new BinaryHelpers(workflow, additionalData).exported,
...new RequestHelpers(this, workflow, node, additionalData).exported,
...new SchedulingHelpers(workflow).exported,
...new SSHTunnelHelpers().exported,
...getSSHTunnelFunctions(),
...getRequestHelperFunctions(workflow, node, additionalData),
...getBinaryHelperFunctions(additionalData, workflow.id),
...getSchedulingFunctions(workflow),
};
}

8
packages/core/src/node-execution-context/webhook-context.ts
View file

@ -24,15 +24,15 @@ import { ApplicationError, createDeferredPromise } from 'n8n-workflow';
import {
copyBinaryFile,
getAdditionalKeys,
getBinaryHelperFunctions,
getCredentials,
getInputConnectionData,
getNodeParameter,
getNodeWebhookUrl,
getRequestHelperFunctions,
returnJsonArray,
} from '@/NodeExecuteFunctions';
import { BinaryHelpers } from './helpers/binary-helpers';
import { RequestHelpers } from './helpers/request-helpers';
import { NodeExecutionContext } from './node-execution-context';
export class WebhookContext extends NodeExecutionContext implements IWebhookFunctions {
@ -54,8 +54,8 @@ export class WebhookContext extends NodeExecutionContext implements IWebhookFunc
this.helpers = {
createDeferredPromise,
returnJsonArray,
...new BinaryHelpers(workflow, additionalData).exported,
...new RequestHelpers(this, workflow, node, additionalData).exported,
...getRequestHelperFunctions(workflow, node, additionalData),
...getBinaryHelperFunctions(additionalData, workflow.id),
};
this.nodeHelpers = {

154
packages/editor-ui/src/components/Error/NodeErrorView.test.ts
View file

@ -1,38 +1,61 @@
import { createComponentRenderer } from '@/__tests__/render';
import { SETTINGS_STORE_DEFAULT_STATE } from '@/__tests__/utils';
import NodeErrorView from '@/components/Error/NodeErrorView.vue';
import { STORES } from '@/constants';
import { createTestingPinia } from '@pinia/testing';
import { type INode } from 'n8n-workflow';
import type { NodeError } from 'n8n-workflow';
import { useAssistantStore } from '@/stores/assistant.store';
import { useNodeTypesStore } from '@/stores/nodeTypes.store';
import { mockedStore } from '@/__tests__/utils';
import userEvent from '@testing-library/user-event';
import { useNDVStore } from '@/stores/ndv.store';
const DEFAULT_SETUP = {
pinia: createTestingPinia({
initialState: {
[STORES.SETTINGS]: SETTINGS_STORE_DEFAULT_STATE,
},
}),
};
const renderComponent = createComponentRenderer(NodeErrorView);
const renderComponent = createComponentRenderer(NodeErrorView, DEFAULT_SETUP);
let mockAiAssistantStore: ReturnType<typeof mockedStore<typeof useAssistantStore>>;
let mockNodeTypeStore: ReturnType<typeof mockedStore<typeof useNodeTypesStore>>;
let mockNdvStore: ReturnType<typeof mockedStore<typeof useNDVStore>>;
describe('NodeErrorView.vue', () => {
let mockNode: INode;
afterEach(() => {
mockNode = {
parameters: {
mode: 'runOnceForAllItems',
language: 'javaScript',
jsCode: 'cons error = 9;',
notice: '',
let error: NodeError;
beforeEach(() => {
createTestingPinia();
mockAiAssistantStore = mockedStore(useAssistantStore);
mockNodeTypeStore = mockedStore(useNodeTypesStore);
mockNdvStore = mockedStore(useNDVStore);
//@ts-expect-error
error = {
name: 'NodeOperationError',
message: 'Test error message',
description: 'Test error description',
context: {
descriptionKey: 'noInputConnection',
nodeCause: 'Test node cause',
runIndex: '1',
itemIndex: '2',
parameter: 'testParameter',
data: { key: 'value' },
causeDetailed: 'Detailed cause',
},
id: 'd1ce5dc9-f9ae-4ac6-84e5-0696ba175dd9',
name: 'Code',
type: 'n8n-nodes-base.code',
typeVersion: 2,
position: [940, 240],
node: {
parameters: {
mode: 'runOnceForAllItems',
language: 'javaScript',
jsCode: 'cons error = 9;',
notice: '',
},
id: 'd1ce5dc9-f9ae-4ac6-84e5-0696ba175dd9',
name: 'ErrorCode',
type: 'n8n-nodes-base.code',
typeVersion: 2,
position: [940, 240],
},
stack: 'Test stack trace',
};
});
afterEach(() => {
vi.clearAllMocks();
});
@ -40,7 +63,7 @@ describe('NodeErrorView.vue', () => {
const { getByTestId } = renderComponent({
props: {
error: {
node: mockNode,
node: error.node,
messages: ['Unexpected identifier [line 1]'],
},
},
@ -55,7 +78,7 @@ describe('NodeErrorView.vue', () => {
const { getByTestId } = renderComponent({
props: {
error: {
node: mockNode,
node: error.node,
message: 'Unexpected identifier [line 1]',
},
},
@ -67,24 +90,20 @@ describe('NodeErrorView.vue', () => {
});
it('should not render AI assistant button when error happens in deprecated function node', async () => {
const aiAssistantStore = useAssistantStore(DEFAULT_SETUP.pinia);
const nodeTypeStore = useNodeTypesStore(DEFAULT_SETUP.pinia);
//@ts-expect-error
nodeTypeStore.getNodeType = vi.fn(() => ({
mockNodeTypeStore.getNodeType = vi.fn(() => ({
type: 'n8n-nodes-base.function',
typeVersion: 1,
hidden: true,
}));
//@ts-expect-error
aiAssistantStore.canShowAssistantButtonsOnCanvas = true;
mockAiAssistantStore.canShowAssistantButtonsOnCanvas = true;
const { queryByTestId } = renderComponent({
props: {
error: {
node: {
...mockNode,
...error.node,
type: 'n8n-nodes-base.function',
typeVersion: 1,
},
@ -96,4 +115,73 @@ describe('NodeErrorView.vue', () => {
expect(aiAssistantButton).toBeNull();
});
it('renders error message', () => {
const { getByTestId } = renderComponent({
props: { error },
});
expect(getByTestId('node-error-message').textContent).toContain('Test error message');
});
it('renders error description', () => {
const { getByTestId } = renderComponent({
props: { error },
});
expect(getByTestId('node-error-description').innerHTML).toContain(
'This node has no input data. Please make sure this node is connected to another node.',
);
});
it('renders stack trace', () => {
const { getByText } = renderComponent({
props: { error },
});
expect(getByText('Test stack trace')).toBeTruthy();
});
it('renders open node button when the error is in sub node', () => {
const { getByTestId, queryByTestId } = renderComponent({
props: {
error: {
...error,
name: 'NodeOperationError',
functionality: 'configuration-node',
},
},
});
expect(getByTestId('node-error-view-open-node-button')).toHaveTextContent('Open errored node');
expect(queryByTestId('ask-assistant-button')).not.toBeInTheDocument();
});
it('does not renders open node button when the error is in sub node', () => {
mockAiAssistantStore.canShowAssistantButtonsOnCanvas = true;
const { getByTestId, queryByTestId } = renderComponent({
props: {
error,
},
});
expect(queryByTestId('node-error-view-open-node-button')).not.toBeInTheDocument();
expect(getByTestId('ask-assistant-button')).toBeInTheDocument();
});
it('open error node details when open error node is clicked', async () => {
const { getByTestId, emitted } = renderComponent({
props: {
error: {
...error,
name: 'NodeOperationError',
functionality: 'configuration-node',
},
},
});
await userEvent.click(getByTestId('node-error-view-open-node-button'));
expect(emitted().click).toHaveLength(1);
expect(mockNdvStore.activeNodeName).toBe(error.node.name);
});
});

55
packages/editor-ui/src/components/Error/NodeErrorView.vue
View file

@ -117,7 +117,7 @@ const prepareRawMessages = computed(() => {
});
const isAskAssistantAvailable = computed(() => {
if (!node.value) {
if (!node.value || isSubNodeError.value) {
return false;
}
const isCustomNode = node.value.type === undefined || isCommunityPackageName(node.value.type);
@ -132,6 +132,13 @@ const assistantAlreadyAsked = computed(() => {
});
});
const isSubNodeError = computed(() => {
return (
props.error.name === 'NodeOperationError' &&
(props.error as NodeOperationError).functionality === 'configuration-node'
);
});
function nodeVersionTag(nodeType: NodeError['node']): string {
if (!nodeType || ('hidden' in nodeType && nodeType.hidden)) {
return i18n.baseText('nodeSettings.deprecated');
@ -153,19 +160,6 @@ function prepareDescription(description: string): string {
}
function getErrorDescription(): string {
const isSubNodeError =
props.error.name === 'NodeOperationError' &&
(props.error as NodeOperationError).functionality === 'configuration-node';
if (isSubNodeError) {
return prepareDescription(
props.error.description +
i18n.baseText('pushConnection.executionError.openNode', {
interpolate: { node: props.error.node.name },
}),
);
}
if (props.error.context?.descriptionKey) {
const interpolate = {
nodeCause: props.error.context.nodeCause as string,
@ -205,13 +199,10 @@ function addItemIndexSuffix(message: string): string {
function getErrorMessage(): string {
let message = '';
const isSubNodeError =
props.error.name === 'NodeOperationError' &&
(props.error as NodeOperationError).functionality === 'configuration-node';
const isNonEmptyString = (value?: unknown): value is string =>
!!value && typeof value === 'string';
if (isSubNodeError) {
if (isSubNodeError.value) {
message = i18n.baseText('nodeErrorView.errorSubNode', {
interpolate: { node: props.error.node.name },
});
@ -390,6 +381,10 @@ function nodeIsHidden() {
return nodeType?.hidden ?? false;
}
const onOpenErrorNodeDetailClick = () => {
ndvStore.activeNodeName = props.error.node.name;
};
async function onAskAssistantClick() {
const { message, lineNumber, description } = props.error;
const sessionInProgress = !assistantStore.isSessionEnded;
@ -428,14 +423,25 @@ async function onAskAssistantClick() {
</div>
</div>
<div
v-if="error.description || error.context?.descriptionKey"
v-if="(error.description || error.context?.descriptionKey) && !isSubNodeError"
data-test-id="node-error-description"
class="node-error-view__header-description"
v-n8n-html="getErrorDescription()"
></div>
<div v-if="isSubNodeError">
<n8n-button
icon="arrow-right"
type="secondary"
:label="i18n.baseText('pushConnection.executionError.openNode')"
class="node-error-view__button"
data-test-id="node-error-view-open-node-button"
@click="onOpenErrorNodeDetailClick"
/>
</div>
<div
v-if="isAskAssistantAvailable"
class="node-error-view__assistant-button"
class="node-error-view__button"
data-test-id="node-error-view-ask-assistant-button"
>
<InlineAskAssistantButton :asked="assistantAlreadyAsked" @click="onAskAssistantClick" />
@ -696,9 +702,14 @@ async function onAskAssistantClick() {
}
}
&__assistant-button {
&__button {
margin-left: var(--spacing-s);
margin-bottom: var(--spacing-xs);
flex-direction: row-reverse;
span {
margin-right: var(--spacing-5xs);
margin-left: var(--spacing-5xs);
}
}
&__debugging {
@ -831,7 +842,7 @@ async function onAskAssistantClick() {
}
}
.node-error-view__assistant-button {
.node-error-view__button {
margin-top: var(--spacing-xs);
}
</style>

6
packages/editor-ui/src/components/RunData.vue
View file

@ -289,7 +289,7 @@ export default defineComponent({
return false;
}
const canPinNode = usePinnedData(this.node).canPinNode(false);
const canPinNode = usePinnedData(this.node).canPinNode(false, this.currentOutputIndex);
return (
canPinNode &&
@ -1214,9 +1214,7 @@ export default defineComponent({
<template>
<div :class="['run-data', $style.container]" @mouseover="activatePane">
<n8n-callout
v-if="
canPinData && pinnedData.hasData.value && !editMode.enabled && !isProductionExecutionPreview
"
v-if="pinnedData.hasData.value && !editMode.enabled && !isProductionExecutionPreview"
theme="secondary"
icon="thumbtack"
:class="$style.pinnedDataCallout"

657
packages/editor-ui/src/components/SettingsLogStreaming/EventDestinationSettingsModal.ee.vue
View file

@ -1,17 +1,15 @@
<script lang="ts">
<script setup lang="ts">
import { computed, onMounted, ref } from 'vue';
import { get, set, unset } from 'lodash-es';
import { mapStores } from 'pinia';
import { useLogStreamingStore } from '@/stores/logStreaming.store';
import { useNDVStore } from '@/stores/ndv.store';
import { useWorkflowsStore } from '@/stores/workflows.store';
import ParameterInputList from '@/components/ParameterInputList.vue';
import type { IMenuItem, INodeUi, IUpdateInformation, ModalKey } from '@/Interface';
import type {
IDataObject,
NodeParameterValue,
MessageEventBusDestinationOptions,
INodeParameters,
NodeParameterValueType,
MessageEventBusDestinationSentryOptions,
MessageEventBusDestinationSyslogOptions,
MessageEventBusDestinationWebhookOptions,
} from 'n8n-workflow';
import {
deepCopy,
@ -22,338 +20,337 @@ import {
defaultMessageEventBusDestinationSyslogOptions,
defaultMessageEventBusDestinationSentryOptions,
} from 'n8n-workflow';
import type { PropType } from 'vue';
import { defineComponent } from 'vue';
import type { EventBus } from 'n8n-design-system';
import { createEventBus } from 'n8n-design-system/utils';
import { useLogStreamingStore } from '@/stores/logStreaming.store';
import { useNDVStore } from '@/stores/ndv.store';
import { useWorkflowsStore } from '@/stores/workflows.store';
import ParameterInputList from '@/components/ParameterInputList.vue';
import type { IMenuItem, IUpdateInformation, ModalKey } from '@/Interface';
import { LOG_STREAM_MODAL_KEY, MODAL_CONFIRM } from '@/constants';
import Modal from '@/components/Modal.vue';
import { useI18n } from '@/composables/useI18n';
import { useMessage } from '@/composables/useMessage';
import { useUIStore } from '@/stores/ui.store';
import { hasPermission } from '@/utils/rbac/permissions';
import { destinationToFakeINodeUi } from '@/components/SettingsLogStreaming/Helpers.ee';
import type { BaseTextKey } from '@/plugins/i18n';
import InlineNameEdit from '@/components/InlineNameEdit.vue';
import SaveButton from '@/components/SaveButton.vue';
import EventSelection from '@/components/SettingsLogStreaming/EventSelection.ee.vue';
import { useTelemetry } from '@/composables/useTelemetry';
import { useRootStore } from '@/stores/root.store';
import {
webhookModalDescription,
sentryModalDescription,
syslogModalDescription,
} from './descriptions.ee';
import type { BaseTextKey } from '@/plugins/i18n';
import InlineNameEdit from '@/components/InlineNameEdit.vue';
import SaveButton from '@/components/SaveButton.vue';
import EventSelection from '@/components/SettingsLogStreaming/EventSelection.ee.vue';
import type { EventBus } from 'n8n-design-system';
import { createEventBus } from 'n8n-design-system/utils';
import { useTelemetry } from '@/composables/useTelemetry';
import { useRootStore } from '@/stores/root.store';
export default defineComponent({
name: 'EventDestinationSettingsModal',
components: {
Modal,
ParameterInputList,
InlineNameEdit,
SaveButton,
EventSelection,
defineOptions({ name: 'EventDestinationSettingsModal' });
const props = withDefaults(
defineProps<{
modalName: ModalKey;
destination?: MessageEventBusDestinationOptions;
isNew?: boolean;
eventBus?: EventBus;
}>(),
{
destination: () => deepCopy(defaultMessageEventBusDestinationOptions),
isNew: false,
},
props: {
modalName: {
type: String as PropType<ModalKey>,
required: true,
},
destination: {
type: Object,
default: () => deepCopy(defaultMessageEventBusDestinationOptions),
},
isNew: Boolean,
eventBus: {
type: Object as PropType<EventBus>,
},
},
setup() {
return {
...useMessage(),
};
},
data() {
return {
unchanged: !this.isNew,
activeTab: 'settings',
hasOnceBeenSaved: !this.isNew,
isSaving: false,
isDeleting: false,
loading: false,
showRemoveConfirm: false,
typeSelectValue: '',
typeSelectPlaceholder: 'Destination Type',
nodeParameters: deepCopy(defaultMessageEventBusDestinationOptions) as INodeParameters,
webhookDescription: webhookModalDescription,
sentryDescription: sentryModalDescription,
syslogDescription: syslogModalDescription,
modalBus: createEventBus(),
headerLabel: this.destination.label,
testMessageSent: false,
testMessageResult: false,
LOG_STREAM_MODAL_KEY,
};
},
computed: {
...mapStores(useUIStore, useLogStreamingStore, useNDVStore, useWorkflowsStore),
typeSelectOptions(): Array<{ value: string; label: BaseTextKey }> {
const options: Array<{ value: string; label: BaseTextKey }> = [];
for (const t of messageEventBusDestinationTypeNames) {
if (t === MessageEventBusDestinationTypeNames.abstract) {
continue;
}
options.push({
value: t,
label: `settings.log-streaming.${t}` as BaseTextKey,
});
}
return options;
},
isTypeAbstract(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.abstract;
},
isTypeWebhook(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.webhook;
},
isTypeSyslog(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.syslog;
},
isTypeSentry(): boolean {
return this.nodeParameters.__type === MessageEventBusDestinationTypeNames.sentry;
},
node(): INodeUi {
return destinationToFakeINodeUi(this.nodeParameters);
},
typeLabelName(): BaseTextKey {
return `settings.log-streaming.${this.nodeParameters.__type}` as BaseTextKey;
},
sidebarItems(): IMenuItem[] {
const items: IMenuItem[] = [
{
id: 'settings',
label: this.$locale.baseText('settings.log-streaming.tab.settings'),
position: 'top',
},
];
if (!this.isTypeAbstract) {
items.push({
id: 'events',
label: this.$locale.baseText('settings.log-streaming.tab.events'),
position: 'top',
});
}
return items;
},
canManageLogStreaming(): boolean {
return hasPermission(['rbac'], { rbac: { scope: 'logStreaming:manage' } });
},
},
mounted() {
this.setupNode(
Object.assign(deepCopy(defaultMessageEventBusDestinationOptions), this.destination),
);
this.workflowsStore.$onAction(
({
name, // name of the action
args, // array of parameters passed to the action
}) => {
if (name === 'updateNodeProperties') {
for (const arg of args) {
if (arg.name === this.destination.id) {
if ('credentials' in arg.properties) {
this.unchanged = false;
this.nodeParameters.credentials = arg.properties
.credentials as NodeParameterValueType;
}
}
}
}
},
);
},
methods: {
onInput() {
this.unchanged = false;
this.testMessageSent = false;
},
onTabSelect(tab: string) {
this.activeTab = tab;
},
onLabelChange(value: string) {
this.onInput();
this.headerLabel = value;
this.nodeParameters.label = value;
},
setupNode(options: MessageEventBusDestinationOptions) {
this.workflowsStore.removeNode(this.node);
this.ndvStore.activeNodeName = options.id ?? 'thisshouldnothappen';
this.workflowsStore.addNode(destinationToFakeINodeUi(options));
this.nodeParameters = options as INodeParameters;
this.logStreamingStore.items[this.destination.id].destination = options;
},
onTypeSelectInput(destinationType: MessageEventBusDestinationTypeNames) {
this.typeSelectValue = destinationType;
},
async onContinueAddClicked() {
let newDestination;
switch (this.typeSelectValue) {
case MessageEventBusDestinationTypeNames.syslog:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSyslogOptions), {
id: this.destination.id,
});
break;
case MessageEventBusDestinationTypeNames.sentry:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSentryOptions), {
id: this.destination.id,
});
break;
case MessageEventBusDestinationTypeNames.webhook:
newDestination = Object.assign(
deepCopy(defaultMessageEventBusDestinationWebhookOptions),
{ id: this.destination.id },
);
break;
}
);
const { modalName, destination, isNew, eventBus } = props;
if (newDestination) {
this.headerLabel = newDestination?.label ?? this.headerLabel;
this.setupNode(newDestination);
}
},
valueChanged(parameterData: IUpdateInformation) {
this.unchanged = false;
this.testMessageSent = false;
const newValue: NodeParameterValue = parameterData.value as string | number;
const parameterPath = parameterData.name?.startsWith('parameters.')
? parameterData.name.split('.').slice(1).join('.')
: parameterData.name || '';
const i18n = useI18n();
const { confirm } = useMessage();
const telemetry = useTelemetry();
const logStreamingStore = useLogStreamingStore();
const ndvStore = useNDVStore();
const workflowsStore = useWorkflowsStore();
const uiStore = useUIStore();
const nodeParameters = deepCopy(this.nodeParameters);
const unchanged = ref(!isNew);
const activeTab = ref('settings');
const hasOnceBeenSaved = ref(!isNew);
const isSaving = ref(false);
const isDeleting = ref(false);
const loading = ref(false);
const typeSelectValue = ref('');
const typeSelectPlaceholder = ref('Destination Type');
const nodeParameters = ref(deepCopy(defaultMessageEventBusDestinationOptions) as INodeParameters);
const webhookDescription = ref(webhookModalDescription);
const sentryDescription = ref(sentryModalDescription);
const syslogDescription = ref(syslogModalDescription);
const modalBus = ref(createEventBus());
const headerLabel = ref(destination.label!);
const testMessageSent = ref(false);
const testMessageResult = ref(false);
// Check if the path is supposed to change an array and if so get
// the needed data like path and index
const parameterPathArray = parameterPath.match(/(.*)\[(\d+)\]$/);
// Apply the new value
if (parameterData.value === undefined && parameterPathArray !== null) {
// Delete array item
const path = parameterPathArray[1] as keyof MessageEventBusDestinationOptions;
const index = parameterPathArray[2];
const data = get(nodeParameters, path);
if (Array.isArray(data)) {
data.splice(parseInt(index, 10), 1);
nodeParameters[path] = data as never;
}
} else {
if (newValue === undefined) {
unset(nodeParameters, parameterPath);
} else {
set(nodeParameters, parameterPath, newValue);
}
}
this.nodeParameters = deepCopy(nodeParameters);
this.workflowsStore.updateNodeProperties({
name: this.node.name,
properties: { parameters: this.nodeParameters as unknown as IDataObject, position: [0, 0] },
});
if (this.hasOnceBeenSaved) {
this.logStreamingStore.updateDestination(this.nodeParameters);
}
},
async sendTestEvent() {
this.testMessageResult = await this.logStreamingStore.sendTestMessage(this.nodeParameters);
this.testMessageSent = true;
},
async removeThis() {
const deleteConfirmed = await this.confirm(
this.$locale.baseText('settings.log-streaming.destinationDelete.message', {
interpolate: { destinationName: this.destination.label },
}),
this.$locale.baseText('settings.log-streaming.destinationDelete.headline'),
{
type: 'warning',
confirmButtonText: this.$locale.baseText(
'settings.log-streaming.destinationDelete.confirmButtonText',
),
cancelButtonText: this.$locale.baseText(
'settings.log-streaming.destinationDelete.cancelButtonText',
),
},
);
if (deleteConfirmed !== MODAL_CONFIRM) {
return;
} else {
this.callEventBus('remove', this.destination.id);
this.uiStore.closeModal(LOG_STREAM_MODAL_KEY);
this.uiStore.stateIsDirty = false;
}
},
onModalClose() {
if (!this.hasOnceBeenSaved) {
this.workflowsStore.removeNode(this.node);
if (this.nodeParameters.id && typeof this.nodeParameters.id !== 'object') {
this.logStreamingStore.removeDestination(this.nodeParameters.id.toString());
}
}
this.ndvStore.activeNodeName = null;
this.callEventBus('closing', this.destination.id);
this.uiStore.stateIsDirty = false;
},
async saveDestination() {
if (this.unchanged || !this.destination.id) {
return;
}
const saveResult = await this.logStreamingStore.saveDestination(this.nodeParameters);
if (saveResult) {
this.hasOnceBeenSaved = true;
this.testMessageSent = false;
this.unchanged = true;
this.callEventBus('destinationWasSaved', this.destination.id);
this.uiStore.stateIsDirty = false;
const destinationType = (
this.nodeParameters.__type && typeof this.nodeParameters.__type !== 'object'
? `${this.nodeParameters.__type}`
: 'unknown'
)
.replace('$$MessageEventBusDestination', '')
.toLowerCase();
const isComplete = () => {
if (this.isTypeWebhook) {
return this.destination.host !== '';
} else if (this.isTypeSentry) {
return this.destination.dsn !== '';
} else if (this.isTypeSyslog) {
return (
this.destination.host !== '' &&
this.destination.port !== undefined &&
this.destination.protocol !== '' &&
this.destination.facility !== undefined &&
this.destination.app_name !== ''
);
}
return false;
};
useTelemetry().track('User updated log streaming destination', {
instance_id: useRootStore().instanceId,
destination_type: destinationType,
is_complete: isComplete(),
is_active: this.destination.enabled,
});
}
},
callEventBus(event: string, data: unknown) {
if (this.eventBus) {
this.eventBus.emit(event, data);
}
},
},
const typeSelectOptions = computed(() => {
const options: Array<{ value: string; label: BaseTextKey }> = [];
for (const t of messageEventBusDestinationTypeNames) {
if (t === MessageEventBusDestinationTypeNames.abstract) {
continue;
}
options.push({
value: t,
label: `settings.log-streaming.${t}` as BaseTextKey,
});
}
return options;
});
const isTypeAbstract = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.abstract,
);
const isTypeWebhook = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.webhook,
);
const isTypeSyslog = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.syslog,
);
const isTypeSentry = computed(
() => nodeParameters.value.__type === MessageEventBusDestinationTypeNames.sentry,
);
const node = computed(() => destinationToFakeINodeUi(nodeParameters.value));
const typeLabelName = computed(
() => `settings.log-streaming.${nodeParameters.value.__type}` as BaseTextKey,
);
const sidebarItems = computed(() => {
const items: IMenuItem[] = [
{
id: 'settings',
label: i18n.baseText('settings.log-streaming.tab.settings'),
position: 'top',
},
];
if (!isTypeAbstract.value) {
items.push({
id: 'events',
label: i18n.baseText('settings.log-streaming.tab.events'),
position: 'top',
});
}
return items;
});
const canManageLogStreaming = computed(() =>
hasPermission(['rbac'], { rbac: { scope: 'logStreaming:manage' } }),
);
onMounted(() => {
setupNode(Object.assign(deepCopy(defaultMessageEventBusDestinationOptions), destination));
workflowsStore.$onAction(({ name, args }) => {
if (name === 'updateNodeProperties') {
for (const arg of args) {
if (arg.name === destination.id) {
if ('credentials' in arg.properties) {
unchanged.value = false;
nodeParameters.value.credentials = arg.properties.credentials as NodeParameterValueType;
}
}
}
}
});
});
function onInput() {
unchanged.value = false;
testMessageSent.value = false;
}
function onTabSelect(tab: string) {
activeTab.value = tab;
}
function onLabelChange(value: string) {
onInput();
headerLabel.value = value;
nodeParameters.value.label = value;
}
function setupNode(options: MessageEventBusDestinationOptions) {
workflowsStore.removeNode(node.value);
ndvStore.activeNodeName = options.id ?? 'thisshouldnothappen';
workflowsStore.addNode(destinationToFakeINodeUi(options));
nodeParameters.value = options as INodeParameters;
logStreamingStore.items[destination.id!].destination = options;
}
function onTypeSelectInput(destinationType: MessageEventBusDestinationTypeNames) {
typeSelectValue.value = destinationType;
}
async function onContinueAddClicked() {
let newDestination;
switch (typeSelectValue.value) {
case MessageEventBusDestinationTypeNames.syslog:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSyslogOptions), {
id: destination.id,
});
break;
case MessageEventBusDestinationTypeNames.sentry:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationSentryOptions), {
id: destination.id,
});
break;
case MessageEventBusDestinationTypeNames.webhook:
newDestination = Object.assign(deepCopy(defaultMessageEventBusDestinationWebhookOptions), {
id: destination.id,
});
break;
}
if (newDestination) {
headerLabel.value = newDestination?.label ?? headerLabel.value;
setupNode(newDestination);
}
}
function valueChanged(parameterData: IUpdateInformation) {
unchanged.value = false;
testMessageSent.value = false;
const newValue: NodeParameterValue = parameterData.value as string | number;
const parameterPath = parameterData.name?.startsWith('parameters.')
? parameterData.name.split('.').slice(1).join('.')
: parameterData.name || '';
const nodeParametersCopy = deepCopy(nodeParameters.value);
if (parameterData.value === undefined && parameterPath.match(/(.*)\[(\d+)\]$/)) {
const path = parameterPath.match(
/(.*)\[(\d+)\]$/,
)?.[1] as keyof MessageEventBusDestinationOptions;
const index = parseInt(parameterPath.match(/(.*)\[(\d+)\]$/)?.[2] ?? '0', 10);
const data = get(nodeParametersCopy, path);
if (Array.isArray(data)) {
data.splice(index, 1);
nodeParametersCopy[path] = data as never;
}
} else {
if (newValue === undefined) {
unset(nodeParametersCopy, parameterPath);
} else {
set(nodeParametersCopy, parameterPath, newValue);
}
}
nodeParameters.value = deepCopy(nodeParametersCopy);
workflowsStore.updateNodeProperties({
name: node.value.name,
properties: { parameters: nodeParameters.value as unknown as IDataObject, position: [0, 0] },
});
if (hasOnceBeenSaved.value) {
logStreamingStore.updateDestination(nodeParameters.value);
}
}
async function sendTestEvent() {
testMessageResult.value = await logStreamingStore.sendTestMessage(nodeParameters.value);
testMessageSent.value = true;
}
async function removeThis() {
const deleteConfirmed = await confirm(
i18n.baseText('settings.log-streaming.destinationDelete.message', {
interpolate: { destinationName: destination.label! },
}),
i18n.baseText('settings.log-streaming.destinationDelete.headline'),
{
type: 'warning',
confirmButtonText: i18n.baseText(
'settings.log-streaming.destinationDelete.confirmButtonText',
),
cancelButtonText: i18n.baseText('settings.log-streaming.destinationDelete.cancelButtonText'),
},
);
if (deleteConfirmed !== MODAL_CONFIRM) {
return;
} else {
callEventBus('remove', destination.id);
uiStore.closeModal(LOG_STREAM_MODAL_KEY);
uiStore.stateIsDirty = false;
}
}
function onModalClose() {
if (!hasOnceBeenSaved.value) {
workflowsStore.removeNode(node.value);
if (nodeParameters.value.id && typeof nodeParameters.value.id !== 'object') {
logStreamingStore.removeDestination(nodeParameters.value.id.toString());
}
}
ndvStore.activeNodeName = null;
callEventBus('closing', destination.id);
uiStore.stateIsDirty = false;
}
async function saveDestination() {
if (unchanged.value || !destination.id) {
return;
}
const saveResult = await logStreamingStore.saveDestination(nodeParameters.value);
if (saveResult) {
hasOnceBeenSaved.value = true;
testMessageSent.value = false;
unchanged.value = true;
callEventBus('destinationWasSaved', destination.id);
uiStore.stateIsDirty = false;
const destinationType = (
nodeParameters.value.__type && typeof nodeParameters.value.__type !== 'object'
? `${nodeParameters.value.__type}`
: 'unknown'
)
.replace('$$MessageEventBusDestination', '')
.toLowerCase();
const isComplete = () => {
if (isTypeWebhook.value) {
const webhookDestination = destination as MessageEventBusDestinationWebhookOptions;
return webhookDestination.url !== '';
} else if (isTypeSentry.value) {
const sentryDestination = destination as MessageEventBusDestinationSentryOptions;
return sentryDestination.dsn !== '';
} else if (isTypeSyslog.value) {
const syslogDestination = destination as MessageEventBusDestinationSyslogOptions;
return (
syslogDestination.host !== '' &&
syslogDestination.port !== undefined &&
// @ts-expect-error TODO: fix this typing
syslogDestination.protocol !== '' &&
syslogDestination.facility !== undefined &&
syslogDestination.app_name !== ''
);
}
return false;
};
telemetry.track('User updated log streaming destination', {
instance_id: useRootStore().instanceId,
destination_type: destinationType,
is_complete: isComplete(),
is_active: destination.enabled,
});
}
}
function callEventBus(event: string, data: unknown) {
if (eventBus) {
eventBus.emit(event, data);
}
}
</script>
<template>
@ -381,7 +378,7 @@ export default defineComponent({
<div :class="$style.destinationInfo">
<InlineNameEdit
:model-value="headerLabel"
:subtitle="!isTypeAbstract ? $locale.baseText(typeLabelName) : 'Select type'"
:subtitle="!isTypeAbstract ? i18n.baseText(typeLabelName) : 'Select type'"
:readonly="isTypeAbstract"
type="Credential"
data-test-id="subtitle-showing-type"
@ -406,7 +403,7 @@ export default defineComponent({
<template v-if="canManageLogStreaming">
<n8n-icon-button
v-if="nodeParameters && hasOnceBeenSaved"
:title="$locale.baseText('settings.log-streaming.delete')"
:title="i18n.baseText('settings.log-streaming.delete')"
icon="trash"
type="tertiary"
:disabled="isSaving"
@ -417,7 +414,7 @@ export default defineComponent({
<SaveButton
:saved="unchanged && hasOnceBeenSaved"
:disabled="isTypeAbstract || unchanged"
:saving-label="$locale.baseText('settings.log-streaming.saving')"
:saving-label="i18n.baseText('settings.log-streaming.saving')"
data-test-id="destination-save-button"
@click="saveDestination"
/>
@ -432,8 +429,8 @@ export default defineComponent({
<template v-if="isTypeAbstract">
<n8n-input-label
:class="$style.typeSelector"
:label="$locale.baseText('settings.log-streaming.selecttype')"
:tooltip-text="$locale.baseText('settings.log-streaming.selecttypehint')"
:label="i18n.baseText('settings.log-streaming.selecttype')"
:tooltip-text="i18n.baseText('settings.log-streaming.selecttypehint')"
:bold="false"
size="medium"
:underline="false"
@ -450,7 +447,7 @@ export default defineComponent({
v-for="option in typeSelectOptions || []"
:key="option.value"
:value="option.value"
:label="$locale.baseText(option.label)"
:label="i18n.baseText(option.label)"
/>
</n8n-select>
<div class="mt-m text-right">
@ -460,7 +457,7 @@ export default defineComponent({
:disabled="!typeSelectValue"
@click="onContinueAddClicked"
>
{{ $locale.baseText(`settings.log-streaming.continue`) }}
{{ i18n.baseText(`settings.log-streaming.continue`) }}
</n8n-button>
</div>
</n8n-input-label>
@ -505,7 +502,7 @@ export default defineComponent({
<div class="">
<n8n-input-label
class="mb-m mt-m"
:label="$locale.baseText('settings.log-streaming.tab.events.title')"
:label="i18n.baseText('settings.log-streaming.tab.events.title')"
:bold="true"
size="medium"
:underline="false"

133
packages/editor-ui/src/composables/usePinnedData.test.ts
View file

@ -3,9 +3,10 @@ import { setActivePinia, createPinia } from 'pinia';
import { ref } from 'vue';
import { usePinnedData } from '@/composables/usePinnedData';
import type { INodeUi } from '@/Interface';
import { MAX_PINNED_DATA_SIZE } from '@/constants';
import { HTTP_REQUEST_NODE_TYPE, IF_NODE_TYPE, MAX_PINNED_DATA_SIZE } from '@/constants';
import { useWorkflowsStore } from '@/stores/workflows.store';
import { useTelemetry } from '@/composables/useTelemetry';
import { NodeConnectionType, STICKY_NODE_TYPE, type INodeTypeDescription } from 'n8n-workflow';
vi.mock('@/composables/useToast', () => ({ useToast: vi.fn(() => ({ showError: vi.fn() })) }));
vi.mock('@/composables/useI18n', () => ({
@ -17,6 +18,13 @@ vi.mock('@/composables/useExternalHooks', () => ({
})),
}));
const getNodeType = vi.fn();
vi.mock('@/stores/nodeTypes.store', () => ({
useNodeTypesStore: vi.fn(() => ({
getNodeType,
})),
}));
describe('usePinnedData', () => {
beforeEach(() => {
setActivePinia(createPinia());
@ -133,4 +141,127 @@ describe('usePinnedData', () => {
expect(spy).toHaveBeenCalled();
});
});
describe('canPinData()', () => {
afterEach(() => {
vi.clearAllMocks();
});
it('allows pin on single output', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: HTTP_REQUEST_NODE_TYPE,
parameters: {},
onError: 'stopWorkflow',
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([NodeConnectionType.Main], HTTP_REQUEST_NODE_TYPE));
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(true);
expect(canPinNode(false, 0)).toBe(true);
// validate out of range index
expect(canPinNode(false, 1)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
});
it('allows pin on one main and one error output', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: HTTP_REQUEST_NODE_TYPE,
parameters: {},
onError: 'continueErrorOutput',
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([NodeConnectionType.Main], HTTP_REQUEST_NODE_TYPE));
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(true);
expect(canPinNode(false, 0)).toBe(true);
expect(canPinNode(false, 1)).toBe(false);
// validate out of range index
expect(canPinNode(false, 2)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
});
it('does not allow pin on two main outputs', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: IF_NODE_TYPE,
parameters: {},
onError: 'stopWorkflow',
} as INodeUi);
getNodeType.mockReturnValue(
makeNodeType([NodeConnectionType.Main, NodeConnectionType.Main], IF_NODE_TYPE),
);
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(false);
expect(canPinNode(false, 0)).toBe(false);
expect(canPinNode(false, 1)).toBe(false);
// validate out of range index
expect(canPinNode(false, 2)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
});
it('does not allow pin on denylisted node', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: STICKY_NODE_TYPE,
} as INodeUi);
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(false);
expect(canPinNode(false, 0)).toBe(false);
});
it('does not allow pin with checkDataEmpty and no pin', async () => {
const node = ref({
name: 'single output node',
typeVersion: 1,
type: HTTP_REQUEST_NODE_TYPE,
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([NodeConnectionType.Main], HTTP_REQUEST_NODE_TYPE));
const { canPinNode } = usePinnedData(node);
expect(canPinNode(true)).toBe(false);
expect(canPinNode(true, 0)).toBe(false);
});
it('does not allow pin without output', async () => {
const node = ref({
name: 'zero output node',
typeVersion: 1,
type: 'n8n-nodes-base.stopAndError',
} as INodeUi);
getNodeType.mockReturnValue(makeNodeType([], 'n8n-nodes-base.stopAndError'));
const { canPinNode } = usePinnedData(node);
expect(canPinNode()).toBe(false);
expect(canPinNode(false, 0)).toBe(false);
expect(canPinNode(false, -1)).toBe(false);
expect(canPinNode(false, 1)).toBe(false);
});
});
});
const makeNodeType = (outputs: NodeConnectionType[], name: string) =>
({
displayName: name,
name,
version: [1],
inputs: [],
outputs,
properties: [],
defaults: { color: '', name: '' },
group: [],
description: '',
}) as INodeTypeDescription;

27
packages/editor-ui/src/composables/usePinnedData.ts
View file

@ -75,9 +75,9 @@ export function usePinnedData(
);
});
function canPinNode(checkDataEmpty = false) {
function canPinNode(checkDataEmpty = false, outputIndex?: number) {
const targetNode = unref(node);
if (targetNode === null) return false;
if (targetNode === null || PIN_DATA_NODE_TYPES_DENYLIST.includes(targetNode.type)) return false;
const nodeType = useNodeTypesStore().getNodeType(targetNode.type, targetNode.typeVersion);
const dataToPin = getInputDataWithPinned(targetNode);
@ -85,14 +85,25 @@ export function usePinnedData(
if (!nodeType || (checkDataEmpty && dataToPin.length === 0)) return false;
const workflow = workflowsStore.getCurrentWorkflow();
const outputs = NodeHelpers.getNodeOutputs(workflow, targetNode, nodeType);
const mainOutputs = outputs.filter((output) =>
typeof output === 'string'
? output === NodeConnectionType.Main
: output.type === NodeConnectionType.Main,
const outputs = NodeHelpers.getNodeOutputs(workflow, targetNode, nodeType).map((output) =>
typeof output === 'string' ? { type: output } : output,
);
return mainOutputs.length === 1 && !PIN_DATA_NODE_TYPES_DENYLIST.includes(targetNode.type);
const mainOutputs = outputs.filter(
(output) => output.type === NodeConnectionType.Main && output.category !== 'error',
);
let indexAcceptable = true;
if (outputIndex !== undefined) {
const output = outputs[outputIndex];
if (outputs[outputIndex] === undefined) return false;
indexAcceptable = output.type === NodeConnectionType.Main && output.category !== 'error';
}
return mainOutputs.length === 1 && indexAcceptable;
}
function isValidJSON(data: string): boolean {

2
packages/editor-ui/src/plugins/i18n/locales/en.json
View file

@ -1498,7 +1498,7 @@
"pushConnection.executionFailed": "Execution failed",
"pushConnection.executionFailed.message": "There might not be enough memory to finish the execution. Tips for avoiding this <a target=\"_blank\" href=\"https://docs.n8n.io/flow-logic/error-handling/memory-errors/\">here</a>",
"pushConnection.executionError": "There was a problem executing the workflow{error}",
"pushConnection.executionError.openNode": " <a data-action='openNodeDetail' data-action-parameter-node='{node}'>Open node</a>",
"pushConnection.executionError.openNode": "Open errored node",
"pushConnection.executionError.details": "<br /><strong>{details}</strong>",
"prompts.productTeamMessage": "Our product team will get in touch personally",
"prompts.npsSurvey.recommendationQuestion": "How likely are you to recommend n8n to a friend or colleague?",

23
packages/nodes-base/credentials/OuraApi.credentials.ts
View file

@ -1,4 +1,9 @@
import type { ICredentialType, INodeProperties } from 'n8n-workflow';
import type {
IAuthenticateGeneric,
ICredentialTestRequest,
ICredentialType,
INodeProperties,
} from 'n8n-workflow';
export class OuraApi implements ICredentialType {
name = 'ouraApi';
@ -16,4 +21,20 @@ export class OuraApi implements ICredentialType {
default: '',
},
];
authenticate: IAuthenticateGeneric = {
type: 'generic',
properties: {
headers: {
Authorization: '=Bearer {{$credentials.accessToken}}',
},
},
};
test: ICredentialTestRequest = {
request: {
baseURL: 'https://api.ouraring.com',
url: '/v2/usercollection/personal_info',
},
};
}

2
packages/nodes-base/nodes/Code/Code.node.ts
View file

@ -108,7 +108,7 @@ export class Code implements INodeType {
: 'javaScript';
const codeParameterName = language === 'python' ? 'pythonCode' : 'jsCode';
if (!runnersConfig.disabled && language === 'javaScript') {
if (runnersConfig.enabled && language === 'javaScript') {
const code = this.getNodeParameter(codeParameterName, 0) as string;
const sandbox = new JsTaskRunnerSandbox(code, nodeMode, workflowMode, this);

2
packages/nodes-base/nodes/Google/CloudStorage/BucketDescription.ts
View file

@ -235,7 +235,7 @@ export const bucketOperations: INodeProperties[] = [
preSend: [parseJSONBody],
},
},
action: 'Create a new Bucket',
action: 'Update the metadata of a Bucket',
},
],
default: 'getAll',

12
packages/nodes-base/nodes/Oura/GenericFunctions.ts
View file

@ -4,7 +4,7 @@ import type {
IHookFunctions,
ILoadOptionsFunctions,
JsonObject,
IRequestOptions,
IHttpRequestOptions,
IHttpRequestMethods,
} from 'n8n-workflow';
import { NodeApiError } from 'n8n-workflow';
@ -18,15 +18,11 @@ export async function ouraApiRequest(
uri?: string,
option: IDataObject = {},
) {
const credentials = await this.getCredentials('ouraApi');
let options: IRequestOptions = {
headers: {
Authorization: `Bearer ${credentials.accessToken}`,
},
let options: IHttpRequestOptions = {
method,
qs,
body,
uri: uri || `https://api.ouraring.com/v1${resource}`,
url: uri ?? `https://api.ouraring.com/v2${resource}`,
json: true,
};
@ -41,7 +37,7 @@ export async function ouraApiRequest(
options = Object.assign({}, options, option);
try {
return await this.helpers.request(options);
return await this.helpers.httpRequestWithAuthentication.call(this, 'ouraApi', options);
} catch (error) {
throw new NodeApiError(this.getNode(), error as JsonObject);
}

174
packages/nodes-base/nodes/Oura/Oura.node.ts
View file

@ -63,94 +63,126 @@ export class Oura implements INodeType {
const length = items.length;
let responseData;
const returnData: IDataObject[] = [];
const returnData: INodeExecutionData[] = [];
const resource = this.getNodeParameter('resource', 0);
const operation = this.getNodeParameter('operation', 0);
for (let i = 0; i < length; i++) {
if (resource === 'profile') {
// *********************************************************************
// profile
// *********************************************************************
try {
if (resource === 'profile') {
// *********************************************************************
// profile
// *********************************************************************
// https://cloud.ouraring.com/docs/personal-info
// https://cloud.ouraring.com/docs/personal-info
if (operation === 'get') {
// ----------------------------------
// profile: get
// ----------------------------------
if (operation === 'get') {
// ----------------------------------
// profile: get
// ----------------------------------
responseData = await ouraApiRequest.call(this, 'GET', '/userinfo');
}
} else if (resource === 'summary') {
// *********************************************************************
// summary
// *********************************************************************
// https://cloud.ouraring.com/docs/daily-summaries
const qs: IDataObject = {};
const { start, end } = this.getNodeParameter('filters', i) as {
start: string;
end: string;
};
const returnAll = this.getNodeParameter('returnAll', 0);
if (start) {
qs.start = moment(start).format('YYYY-MM-DD');
}
if (end) {
qs.end = moment(end).format('YYYY-MM-DD');
}
if (operation === 'getActivity') {
// ----------------------------------
// profile: getActivity
// ----------------------------------
responseData = await ouraApiRequest.call(this, 'GET', '/activity', {}, qs);
responseData = responseData.activity;
if (!returnAll) {
const limit = this.getNodeParameter('limit', 0);
responseData = responseData.splice(0, limit);
responseData = await ouraApiRequest.call(this, 'GET', '/usercollection/personal_info');
}
} else if (operation === 'getReadiness') {
// ----------------------------------
// profile: getReadiness
// ----------------------------------
} else if (resource === 'summary') {
// *********************************************************************
// summary
// *********************************************************************
responseData = await ouraApiRequest.call(this, 'GET', '/readiness', {}, qs);
responseData = responseData.readiness;
// https://cloud.ouraring.com/docs/daily-summaries
if (!returnAll) {
const limit = this.getNodeParameter('limit', 0);
responseData = responseData.splice(0, limit);
const qs: IDataObject = {};
const { start, end } = this.getNodeParameter('filters', i) as {
start: string;
end: string;
};
const returnAll = this.getNodeParameter('returnAll', 0);
if (start) {
qs.start_date = moment(start).format('YYYY-MM-DD');
}
} else if (operation === 'getSleep') {
// ----------------------------------
// profile: getSleep
// ----------------------------------
responseData = await ouraApiRequest.call(this, 'GET', '/sleep', {}, qs);
responseData = responseData.sleep;
if (end) {
qs.end_date = moment(end).format('YYYY-MM-DD');
}
if (!returnAll) {
const limit = this.getNodeParameter('limit', 0);
responseData = responseData.splice(0, limit);
if (operation === 'getActivity') {
// ----------------------------------
// profile: getActivity
// ----------------------------------
responseData = await ouraApiRequest.call(
this,
'GET',
'/usercollection/daily_activity',
{},
qs,
);
responseData = responseData.data;
if (!returnAll) {
const limit = this.getNodeParameter('limit', 0);
responseData = responseData.splice(0, limit);
}
} else if (operation === 'getReadiness') {
// ----------------------------------
// profile: getReadiness
// ----------------------------------
responseData = await ouraApiRequest.call(
this,
'GET',
'/usercollection/daily_readiness',
{},
qs,
);
responseData = responseData.data;
if (!returnAll) {
const limit = this.getNodeParameter('limit', 0);
responseData = responseData.splice(0, limit);
}
} else if (operation === 'getSleep') {
// ----------------------------------
// profile: getSleep
// ----------------------------------
responseData = await ouraApiRequest.call(
this,
'GET',
'/usercollection/daily_sleep',
{},
qs,
);
responseData = responseData.data;
if (!returnAll) {
const limit = this.getNodeParameter('limit', 0);
responseData = responseData.splice(0, limit);
}
}
}
const executionData = this.helpers.constructExecutionMetaData(
this.helpers.returnJsonArray(responseData as IDataObject[]),
{ itemData: { item: i } },
);
returnData.push(...executionData);
} catch (error) {
if (this.continueOnFail()) {
const executionErrorData = this.helpers.constructExecutionMetaData(
this.helpers.returnJsonArray({ error: error.message }),
{ itemData: { item: i } },
);
returnData.push(...executionErrorData);
continue;
}
throw error;
}
Array.isArray(responseData)
? returnData.push(...(responseData as IDataObject[]))
: returnData.push(responseData as IDataObject);
}
return [this.helpers.returnJsonArray(returnData)];
return [returnData];
}
}

8
packages/nodes-base/nodes/Oura/test/apiResponses.ts Normal file
View file

@ -0,0 +1,8 @@
export const profileResponse = {
id: 'some-id',
age: 30,
weight: 168,
height: 80,
biological_sex: 'male',
email: 'nathan@n8n.io',
};

76
packages/nodes-base/nodes/Oura/test/oura.node.test.ts Normal file
View file

@ -0,0 +1,76 @@
import type {
IExecuteFunctions,
IHookFunctions,
ILoadOptionsFunctions,
IHttpRequestMethods,
INode,
} from 'n8n-workflow';
import nock from 'nock';
import { setup, equalityTest, workflowToTests, getWorkflowFilenames } from '@test/nodes/Helpers';
import { profileResponse } from './apiResponses';
import { ouraApiRequest } from '../GenericFunctions';
const node: INode = {
id: '2cdb46cf-b561-4537-a982-b8d26dd7718b',
name: 'Oura',
type: 'n8n-nodes-base.oura',
typeVersion: 1,
position: [0, 0],
parameters: {
resource: 'profile',
operation: 'get',
},
};
const mockThis = {
helpers: {
httpRequestWithAuthentication: jest
.fn()
.mockResolvedValue({ statusCode: 200, data: profileResponse }),
},
getNode() {
return node;
},
getNodeParameter: jest.fn(),
} as unknown as IHookFunctions | IExecuteFunctions | ILoadOptionsFunctions;
describe('Oura', () => {
describe('ouraApiRequest', () => {
it('should make an authenticated API request to Oura', async () => {
const method: IHttpRequestMethods = 'GET';
const resource = '/usercollection/personal_info';
await ouraApiRequest.call(mockThis, method, resource);
expect(mockThis.helpers.httpRequestWithAuthentication).toHaveBeenCalledWith('ouraApi', {
method: 'GET',
url: 'https://api.ouraring.com/v2/usercollection/personal_info',
json: true,
});
});
});
describe('Run Oura workflow', () => {
const workflows = getWorkflowFilenames(__dirname);
const tests = workflowToTests(workflows);
beforeAll(() => {
nock.disableNetConnect();
nock('https://api.ouraring.com/v2')
.get('/usercollection/personal_info')
.reply(200, profileResponse);
});
afterAll(() => {
nock.restore();
});
const nodeTypes = setup(tests);
for (const testData of tests) {
test(testData.description, async () => await equalityTest(testData, nodeTypes));
}
});
});

86
packages/nodes-base/nodes/Oura/test/oura_test_workflow.json Normal file
View file

@ -0,0 +1,86 @@
{
"name": "Oura Test Workflow",
"nodes": [
{
"parameters": {},
"id": "c1e3b825-a9a8-4def-986b-9108d9441992",
"name": "When clicking ‘Test workflow’",
"type": "n8n-nodes-base.manualTrigger",
"position": [720, 400],
"typeVersion": 1
},
{
"parameters": {
"resource": "profile"
},
"id": "7969bf78-9343-4f81-8f79-dc415a60e168",
"name": "Oura",
"type": "n8n-nodes-base.oura",
"typeVersion": 1,
"position": [940, 400],
"credentials": {
"ouraApi": {
"id": "r083EOdhFatkVvFy",
"name": "Oura account"
}
}
},
{
"parameters": {},
"id": "9b97fa0e-51a6-41d3-8a7d-cff0531e5527",
"name": "No Operation, do nothing",
"type": "n8n-nodes-base.noOp",
"typeVersion": 1,
"position": [1140, 400]
}
],
"pinData": {
"No Operation, do nothing": [
{
"json": {
"id": "some-id",
"age": 30,
"weight": 168,
"height": 80,
"biological_sex": "male",
"email": "nathan@n8n.io"
}
}
]
},
"connections": {
"When clicking ‘Test workflow’": {
"main": [
[
{
"node": "Oura",
"type": "main",
"index": 0
}
]
]
},
"Oura": {
"main": [
[
{
"node": "No Operation, do nothing",
"type": "main",
"index": 0
}
]
]
}
},
"active": false,
"settings": {
"executionOrder": "v1"
},
"versionId": "bd108f46-f6fc-4c22-8655-ade2f51c4b33",
"meta": {
"templateCredsSetupCompleted": true,
"instanceId": "0fa937d34dcabeff4bd6480d3b42cc95edf3bc20e6810819086ef1ce2623639d"
},
"id": "SrUileWU90mQeo02",
"tags": []
}