diff --git a/packages/@n8n/task-runner/package.json b/packages/@n8n/task-runner/package.json index cb47095c15..4b5478a97d 100644 --- a/packages/@n8n/task-runner/package.json +++ b/packages/@n8n/task-runner/package.json @@ -17,10 +17,22 @@ }, "main": "dist/start.js", "module": "src/start.ts", - "types": "dist/start.d.ts", + "types": "dist/index.d.ts", "files": [ "dist/**/*" ], + "exports": { + "./start": { + "require": "./dist/start.js", + "import": "./src/start.ts", + "types": "./dist/start.d.ts" + }, + ".": { + "require": "./dist/index.js", + "import": "./src/index.ts", + "types": "./dist/index.d.ts" + } + }, "dependencies": { "@n8n/config": "workspace:*", "acorn": "8.14.0", diff --git a/packages/@n8n/task-runner/src/index.ts b/packages/@n8n/task-runner/src/index.ts index 59e6f6d288..bc770ea08e 100644 --- a/packages/@n8n/task-runner/src/index.ts +++ b/packages/@n8n/task-runner/src/index.ts @@ -1,2 +1,3 @@ export * from './task-runner'; export * from './runner-types'; +export * from './message-types'; diff --git a/packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/built-ins-parser-state.ts b/packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/built-ins-parser-state.ts index 112c97ccda..064a45df67 100644 --- a/packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/built-ins-parser-state.ts +++ b/packages/@n8n/task-runner/src/js-task-runner/built-ins-parser/built-ins-parser-state.ts @@ -1,4 +1,4 @@ -import type { N8nMessage } from '../../runner-types'; +import type { BrokerMessage } from '@/message-types'; /** * Class to keep track of which built-in variables are accessed in the code @@ -53,7 +53,7 @@ export class BuiltInsParserState { this.needs$prevNode = true; } - toDataRequestParams(): N8nMessage.ToRequester.TaskDataRequest['requestParams'] { + toDataRequestParams(): BrokerMessage.ToRequester.TaskDataRequest['requestParams'] { return { dataOfNodes: this.needsAllNodes ? 'all' : Array.from(this.neededNodeNames), env: this.needs$env, diff --git a/packages/@n8n/task-runner/src/message-types.ts b/packages/@n8n/task-runner/src/message-types.ts new file mode 100644 index 0000000000..95c5f5b72f --- /dev/null +++ b/packages/@n8n/task-runner/src/message-types.ts @@ -0,0 +1,204 @@ +import type { INodeTypeBaseDescription } from 'n8n-workflow'; + +import type { RPC_ALLOW_LIST, TaskDataRequestParams, TaskResultData } from './runner-types'; + +export namespace BrokerMessage { + export namespace ToRunner { + export interface InfoRequest { + type: 'broker:inforequest'; + } + + export interface RunnerRegistered { + type: 'broker:runnerregistered'; + } + + export interface TaskOfferAccept { + type: 'broker:taskofferaccept'; + taskId: string; + offerId: string; + } + + export interface TaskCancel { + type: 'broker:taskcancel'; + taskId: string; + reason: string; + } + + export interface TaskSettings { + type: 'broker:tasksettings'; + taskId: string; + settings: unknown; + } + + export interface RPCResponse { + type: 'broker:rpcresponse'; + callId: string; + taskId: string; + status: 'success' | 'error'; + data: unknown; + } + + export interface TaskDataResponse { + type: 'broker:taskdataresponse'; + taskId: string; + requestId: string; + data: unknown; + } + + export interface NodeTypes { + type: 'broker:nodetypes'; + nodeTypes: INodeTypeBaseDescription[]; + } + + export type All = + | InfoRequest + | TaskOfferAccept + | TaskCancel + | TaskSettings + | RunnerRegistered + | RPCResponse + | TaskDataResponse + | NodeTypes; + } + + export namespace ToRequester { + export interface TaskReady { + type: 'broker:taskready'; + requestId: string; + taskId: string; + } + + export interface TaskDone { + type: 'broker:taskdone'; + taskId: string; + data: TaskResultData; + } + + export interface TaskError { + type: 'broker:taskerror'; + taskId: string; + error: unknown; + } + + export interface TaskDataRequest { + type: 'broker:taskdatarequest'; + taskId: string; + requestId: string; + requestParams: TaskDataRequestParams; + } + + export interface RPC { + type: 'broker:rpc'; + callId: string; + taskId: string; + name: (typeof RPC_ALLOW_LIST)[number]; + params: unknown[]; + } + + export type All = TaskReady | TaskDone | TaskError | TaskDataRequest | RPC; + } +} + +export namespace RequesterMessage { + export namespace ToBroker { + export interface TaskSettings { + type: 'requester:tasksettings'; + taskId: string; + settings: unknown; + } + + export interface TaskCancel { + type: 'requester:taskcancel'; + taskId: string; + reason: string; + } + + export interface TaskDataResponse { + type: 'requester:taskdataresponse'; + taskId: string; + requestId: string; + data: unknown; + } + + export interface RPCResponse { + type: 'requester:rpcresponse'; + taskId: string; + callId: string; + status: 'success' | 'error'; + data: unknown; + } + + export interface TaskRequest { + type: 'requester:taskrequest'; + requestId: string; + taskType: string; + } + + export type All = TaskSettings | TaskCancel | RPCResponse | TaskDataResponse | TaskRequest; + } +} + +export namespace RunnerMessage { + export namespace ToBroker { + export interface Info { + type: 'runner:info'; + name: string; + types: string[]; + } + + export interface TaskAccepted { + type: 'runner:taskaccepted'; + taskId: string; + } + + export interface TaskRejected { + type: 'runner:taskrejected'; + taskId: string; + reason: string; + } + + export interface TaskDone { + type: 'runner:taskdone'; + taskId: string; + data: TaskResultData; + } + + export interface TaskError { + type: 'runner:taskerror'; + taskId: string; + error: unknown; + } + + export interface TaskOffer { + type: 'runner:taskoffer'; + offerId: string; + taskType: string; + validFor: number; + } + + export interface TaskDataRequest { + type: 'runner:taskdatarequest'; + taskId: string; + requestId: string; + requestParams: TaskDataRequestParams; + } + + export interface RPC { + type: 'runner:rpc'; + callId: string; + taskId: string; + name: (typeof RPC_ALLOW_LIST)[number]; + params: unknown[]; + } + + export type All = + | Info + | TaskDone + | TaskError + | TaskAccepted + | TaskRejected + | TaskOffer + | RPC + | TaskDataRequest; + } +} diff --git a/packages/@n8n/task-runner/src/runner-types.ts b/packages/@n8n/task-runner/src/runner-types.ts index 898279feac..c55b50bf4c 100644 --- a/packages/@n8n/task-runner/src/runner-types.ts +++ b/packages/@n8n/task-runner/src/runner-types.ts @@ -1,216 +1,90 @@ -import type { INodeExecutionData, INodeTypeBaseDescription } from 'n8n-workflow'; +import type { + EnvProviderState, + IDataObject, + IExecuteData, + IExecuteFunctions, + INode, + INodeExecutionData, + INodeParameters, + IRunExecutionData, + ITaskDataConnections, + IWorkflowExecuteAdditionalData, + Workflow, + WorkflowExecuteMode, + WorkflowParameters, +} from 'n8n-workflow'; +/** + * 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 DataRequestResponse { + workflow: Omit; + inputData: ITaskDataConnections; + 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; + additionalData: PartialAdditionalData; +} + export interface TaskResultData { result: INodeExecutionData[]; customData?: Record; } -export namespace N8nMessage { - export namespace ToRunner { - export interface InfoRequest { - type: 'broker:inforequest'; - } +export interface TaskData { + executeFunctions: IExecuteFunctions; + inputData: ITaskDataConnections; + node: INode; - export interface RunnerRegistered { - type: 'broker:runnerregistered'; - } - - export interface TaskOfferAccept { - type: 'broker:taskofferaccept'; - taskId: string; - offerId: string; - } - - export interface TaskCancel { - type: 'broker:taskcancel'; - taskId: string; - reason: string; - } - - export interface TaskSettings { - type: 'broker:tasksettings'; - taskId: string; - settings: unknown; - } - - export interface RPCResponse { - type: 'broker:rpcresponse'; - callId: string; - taskId: string; - status: 'success' | 'error'; - data: unknown; - } - - export interface TaskDataResponse { - type: 'broker:taskdataresponse'; - taskId: string; - requestId: string; - data: unknown; - } - - export interface NodeTypes { - type: 'broker:nodetypes'; - nodeTypes: INodeTypeBaseDescription[]; - } - - export type All = - | InfoRequest - | TaskOfferAccept - | TaskCancel - | TaskSettings - | RunnerRegistered - | RPCResponse - | TaskDataResponse - | NodeTypes; - } - - export namespace ToRequester { - export interface TaskReady { - type: 'broker:taskready'; - requestId: string; - taskId: string; - } - - export interface TaskDone { - type: 'broker:taskdone'; - taskId: string; - data: TaskResultData; - } - - export interface TaskError { - type: 'broker:taskerror'; - taskId: string; - error: unknown; - } - - export interface TaskDataRequest { - type: 'broker:taskdatarequest'; - taskId: string; - requestId: string; - requestParams: TaskDataRequestParams; - } - - export interface RPC { - type: 'broker:rpc'; - callId: string; - taskId: string; - name: (typeof RPC_ALLOW_LIST)[number]; - params: unknown[]; - } - - export type All = TaskReady | TaskDone | TaskError | TaskDataRequest | RPC; - } + workflow: Workflow; + runExecutionData: IRunExecutionData; + runIndex: number; + itemIndex: number; + activeNodeName: string; + connectionInputData: INodeExecutionData[]; + siblingParameters: INodeParameters; + mode: WorkflowExecuteMode; + envProviderState: EnvProviderState; + executeData?: IExecuteData; + defaultReturnRunIndex: number; + selfData: IDataObject; + contextNodeName: string; + additionalData: IWorkflowExecuteAdditionalData; } -export namespace RequesterMessage { - export namespace ToN8n { - export interface TaskSettings { - type: 'requester:tasksettings'; - taskId: string; - settings: unknown; - } - - export interface TaskCancel { - type: 'requester:taskcancel'; - taskId: string; - reason: string; - } - - export interface TaskDataResponse { - type: 'requester:taskdataresponse'; - taskId: string; - requestId: string; - data: unknown; - } - - export interface RPCResponse { - type: 'requester:rpcresponse'; - taskId: string; - callId: string; - status: 'success' | 'error'; - data: unknown; - } - - export interface TaskRequest { - type: 'requester:taskrequest'; - requestId: string; - taskType: string; - } - - export type All = TaskSettings | TaskCancel | RPCResponse | TaskDataResponse | TaskRequest; - } -} - -export namespace RunnerMessage { - export namespace ToN8n { - export interface Info { - type: 'runner:info'; - name: string; - types: string[]; - } - - export interface TaskAccepted { - type: 'runner:taskaccepted'; - taskId: string; - } - - export interface TaskRejected { - type: 'runner:taskrejected'; - taskId: string; - reason: string; - } - - export interface TaskDone { - type: 'runner:taskdone'; - taskId: string; - data: TaskResultData; - } - - export interface TaskError { - type: 'runner:taskerror'; - taskId: string; - error: unknown; - } - - export interface TaskOffer { - type: 'runner:taskoffer'; - offerId: string; - taskType: string; - validFor: number; - } - - export interface TaskDataRequest { - type: 'runner:taskdatarequest'; - taskId: string; - requestId: string; - requestParams: TaskDataRequestParams; - } - - export interface RPC { - type: 'runner:rpc'; - callId: string; - taskId: string; - name: (typeof RPC_ALLOW_LIST)[number]; - params: unknown[]; - } - - export type All = - | Info - | TaskDone - | TaskError - | TaskAccepted - | TaskRejected - | TaskOffer - | RPC - | TaskDataRequest; - } +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 const RPC_ALLOW_LIST = [ diff --git a/packages/@n8n/task-runner/src/task-runner.ts b/packages/@n8n/task-runner/src/task-runner.ts index b292bd4413..81de93e6b0 100644 --- a/packages/@n8n/task-runner/src/task-runner.ts +++ b/packages/@n8n/task-runner/src/task-runner.ts @@ -2,14 +2,10 @@ import { ApplicationError, type INodeTypeDescription } from 'n8n-workflow'; import { nanoid } from 'nanoid'; import { type MessageEvent, WebSocket } from 'ws'; -import type { BaseRunnerConfig } from './config/base-runner-config'; -import { TaskRunnerNodeTypes } from './node-types'; -import { - RPC_ALLOW_LIST, - type RunnerMessage, - type N8nMessage, - type TaskResultData, -} from './runner-types'; +import type { BaseRunnerConfig } from '@/config/base-runner-config'; +import type { BrokerMessage, RunnerMessage } from '@/message-types'; +import { TaskRunnerNodeTypes } from '@/node-types'; +import { RPC_ALLOW_LIST, type TaskResultData } from '@/runner-types'; export interface Task { taskId: string; @@ -90,7 +86,7 @@ export abstract class TaskRunner { private receiveMessage = (message: MessageEvent) => { // eslint-disable-next-line n8n-local-rules/no-uncaught-json-parse - const data = JSON.parse(message.data as string) as N8nMessage.ToRunner.All; + const data = JSON.parse(message.data as string) as BrokerMessage.ToRunner.All; void this.onMessage(data); }; @@ -140,11 +136,11 @@ export abstract class TaskRunner { } } - send(message: RunnerMessage.ToN8n.All) { + send(message: RunnerMessage.ToBroker.All) { this.ws.send(JSON.stringify(message)); } - onMessage(message: N8nMessage.ToRunner.All) { + onMessage(message: BrokerMessage.ToRunner.All) { switch (message.type) { case 'broker:inforequest': this.send({ @@ -252,7 +248,7 @@ export abstract class TaskRunner { this.sendOffers(); } - taskDone(taskId: string, data: RunnerMessage.ToN8n.TaskDone['data']) { + taskDone(taskId: string, data: RunnerMessage.ToBroker.TaskDone['data']) { this.send({ type: 'runner:taskdone', taskId, @@ -288,7 +284,7 @@ export abstract class TaskRunner { async requestData( taskId: Task['taskId'], - requestParams: RunnerMessage.ToN8n.TaskDataRequest['requestParams'], + requestParams: RunnerMessage.ToBroker.TaskDataRequest['requestParams'], ): Promise { const requestId = nanoid(); @@ -314,7 +310,7 @@ export abstract class TaskRunner { } } - async makeRpcCall(taskId: string, name: RunnerMessage.ToN8n.RPC['name'], params: unknown[]) { + async makeRpcCall(taskId: string, name: RunnerMessage.ToBroker.RPC['name'], params: unknown[]) { const callId = nanoid(); const dataPromise = new Promise((resolve, reject) => { @@ -342,7 +338,7 @@ export abstract class TaskRunner { handleRpcResponse( callId: string, - status: N8nMessage.ToRunner.RPCResponse['status'], + status: BrokerMessage.ToRunner.RPCResponse['status'], data: unknown, ) { const call = this.rpcCalls.get(callId); diff --git a/packages/cli/src/runners/__tests__/task-broker.test.ts b/packages/cli/src/runners/__tests__/task-broker.test.ts index 8787ba5955..715c5d8eb8 100644 --- a/packages/cli/src/runners/__tests__/task-broker.test.ts +++ b/packages/cli/src/runners/__tests__/task-broker.test.ts @@ -1,7 +1,7 @@ +import type { RunnerMessage, TaskResultData } from '@n8n/task-runner'; import { mock } from 'jest-mock-extended'; import { TaskRejectError } from '../errors'; -import type { RunnerMessage, TaskResultData } from '../runner-types'; import { TaskBroker } from '../task-broker.service'; import type { TaskOffer, TaskRequest, TaskRunner } from '../task-broker.service'; @@ -381,7 +381,7 @@ describe('TaskBroker', () => { const runnerId = 'runner1'; const taskId = 'task1'; - const message: RunnerMessage.ToN8n.TaskAccepted = { + const message: RunnerMessage.ToBroker.TaskAccepted = { type: 'runner:taskaccepted', taskId, }; @@ -406,7 +406,7 @@ describe('TaskBroker', () => { const taskId = 'task1'; const rejectionReason = 'Task execution failed'; - const message: RunnerMessage.ToN8n.TaskRejected = { + const message: RunnerMessage.ToBroker.TaskRejected = { type: 'runner:taskrejected', taskId, reason: rejectionReason, @@ -433,7 +433,7 @@ describe('TaskBroker', () => { const requesterId = 'requester1'; const data = mock(); - const message: RunnerMessage.ToN8n.TaskDone = { + const message: RunnerMessage.ToBroker.TaskDone = { type: 'runner:taskdone', taskId, data, @@ -464,7 +464,7 @@ describe('TaskBroker', () => { const requesterId = 'requester1'; const errorMessage = 'Task execution failed'; - const message: RunnerMessage.ToN8n.TaskError = { + const message: RunnerMessage.ToBroker.TaskError = { type: 'runner:taskerror', taskId, error: errorMessage, @@ -494,14 +494,14 @@ describe('TaskBroker', () => { const taskId = 'task1'; const requesterId = 'requester1'; const requestId = 'request1'; - const requestParams: RunnerMessage.ToN8n.TaskDataRequest['requestParams'] = { + const requestParams: RunnerMessage.ToBroker.TaskDataRequest['requestParams'] = { dataOfNodes: 'all', env: true, input: true, prevNode: true, }; - const message: RunnerMessage.ToN8n.TaskDataRequest = { + const message: RunnerMessage.ToBroker.TaskDataRequest = { type: 'runner:taskdatarequest', taskId, requestId, @@ -534,7 +534,7 @@ describe('TaskBroker', () => { const rpcName = 'helpers.httpRequestWithAuthentication'; const rpcParams = ['param1', 'param2']; - const message: RunnerMessage.ToN8n.RPC = { + const message: RunnerMessage.ToBroker.RPC = { type: 'runner:rpc', taskId, callId, diff --git a/packages/cli/src/runners/runner-types.ts b/packages/cli/src/runners/runner-types.ts index c5d4eb81c3..8fcfe968d3 100644 --- a/packages/cli/src/runners/runner-types.ts +++ b/packages/cli/src/runners/runner-types.ts @@ -1,5 +1,5 @@ import type { Response } from 'express'; -import type { INodeExecutionData, INodeTypeBaseDescription } from 'n8n-workflow'; +import type { INodeExecutionData } from 'n8n-workflow'; import type WebSocket from 'ws'; import type { TaskRunner } from './task-broker.service'; @@ -34,230 +34,3 @@ export interface TaskRunnerServerInitRequest } export type TaskRunnerServerInitResponse = Response & { req: TaskRunnerServerInitRequest }; - -export namespace N8nMessage { - export namespace ToRunner { - export interface InfoRequest { - type: 'broker:inforequest'; - } - - export interface RunnerRegistered { - type: 'broker:runnerregistered'; - } - - export interface TaskOfferAccept { - type: 'broker:taskofferaccept'; - taskId: string; - offerId: string; - } - - export interface TaskCancel { - type: 'broker:taskcancel'; - taskId: string; - reason: string; - } - - export interface TaskSettings { - type: 'broker:tasksettings'; - taskId: string; - settings: unknown; - } - - export interface RPCResponse { - type: 'broker:rpcresponse'; - callId: string; - taskId: string; - status: 'success' | 'error'; - data: unknown; - } - - export interface TaskDataResponse { - type: 'broker:taskdataresponse'; - taskId: string; - requestId: string; - data: unknown; - } - - export interface NodeTypes { - type: 'broker:nodetypes'; - nodeTypes: INodeTypeBaseDescription[]; - } - - export type All = - | InfoRequest - | TaskOfferAccept - | TaskCancel - | TaskSettings - | RunnerRegistered - | RPCResponse - | TaskDataResponse - | NodeTypes; - } - - export namespace ToRequester { - export interface TaskReady { - type: 'broker:taskready'; - requestId: string; - taskId: string; - } - - export interface TaskDone { - type: 'broker:taskdone'; - taskId: string; - data: TaskResultData; - } - - export interface TaskError { - type: 'broker:taskerror'; - taskId: string; - error: unknown; - } - - export interface TaskDataRequest { - type: 'broker:taskdatarequest'; - taskId: string; - requestId: string; - requestParams: TaskDataRequestParams; - } - - export interface RPC { - type: 'broker:rpc'; - callId: string; - taskId: string; - name: (typeof RPC_ALLOW_LIST)[number]; - params: unknown[]; - } - - export type All = TaskReady | TaskDone | TaskError | TaskDataRequest | RPC; - } -} - -export namespace RequesterMessage { - export namespace ToN8n { - export interface TaskSettings { - type: 'requester:tasksettings'; - taskId: string; - settings: unknown; - } - - export interface TaskCancel { - type: 'requester:taskcancel'; - taskId: string; - reason: string; - } - - export interface TaskDataResponse { - type: 'requester:taskdataresponse'; - taskId: string; - requestId: string; - data: unknown; - } - - export interface RPCResponse { - type: 'requester:rpcresponse'; - taskId: string; - callId: string; - status: 'success' | 'error'; - data: unknown; - } - - export interface TaskRequest { - type: 'requester:taskrequest'; - requestId: string; - taskType: string; - } - - export type All = TaskSettings | TaskCancel | RPCResponse | TaskDataResponse | TaskRequest; - } -} - -export namespace RunnerMessage { - export namespace ToN8n { - export interface Info { - type: 'runner:info'; - name: string; - types: string[]; - } - - export interface TaskAccepted { - type: 'runner:taskaccepted'; - taskId: string; - } - - export interface TaskRejected { - type: 'runner:taskrejected'; - taskId: string; - reason: string; - } - - export interface TaskDone { - type: 'runner:taskdone'; - taskId: string; - data: TaskResultData; - } - - export interface TaskError { - type: 'runner:taskerror'; - taskId: string; - error: unknown; - } - - export interface TaskOffer { - type: 'runner:taskoffer'; - offerId: string; - taskType: string; - validFor: number; - } - - export interface TaskDataRequest { - type: 'runner:taskdatarequest'; - taskId: string; - requestId: string; - requestParams: TaskDataRequestParams; - } - - export interface RPC { - type: 'runner:rpc'; - callId: string; - taskId: string; - name: (typeof RPC_ALLOW_LIST)[number]; - params: unknown[]; - } - - export type All = - | Info - | TaskDone - | TaskError - | TaskAccepted - | TaskRejected - | TaskOffer - | RPC - | TaskDataRequest; - } -} - -export const RPC_ALLOW_LIST = [ - 'logNodeOutput', - 'helpers.httpRequestWithAuthentication', - 'helpers.requestWithAuthenticationPaginated', - // "helpers.normalizeItems" - // "helpers.constructExecutionMetaData" - // "helpers.assertBinaryData" - 'helpers.getBinaryDataBuffer', - // "helpers.copyInputItems" - // "helpers.returnJsonArray" - 'helpers.getSSHClient', - 'helpers.createReadStream', - // "helpers.getStoragePath" - 'helpers.writeContentToFile', - 'helpers.prepareBinaryData', - 'helpers.setBinaryDataBuffer', - 'helpers.copyBinaryFile', - 'helpers.binaryToBuffer', - // "helpers.binaryToString" - // "helpers.getBinaryPath" - 'helpers.getBinaryStream', - 'helpers.getBinaryMetadata', - 'helpers.createDeferredPromise', - 'helpers.httpRequest', -] as const; diff --git a/packages/cli/src/runners/runner-ws-server.ts b/packages/cli/src/runners/runner-ws-server.ts index 5d4c4e9607..baaac82bf3 100644 --- a/packages/cli/src/runners/runner-ws-server.ts +++ b/packages/cli/src/runners/runner-ws-server.ts @@ -1,3 +1,4 @@ +import type { BrokerMessage, RunnerMessage } from '@n8n/task-runner'; import { Service } from 'typedi'; import type WebSocket from 'ws'; @@ -5,11 +6,9 @@ import { Logger } from '@/logging/logger.service'; import { DefaultTaskRunnerDisconnectAnalyzer } from './default-task-runner-disconnect-analyzer'; import type { - RunnerMessage, - N8nMessage, + DisconnectAnalyzer, TaskRunnerServerInitRequest, TaskRunnerServerInitResponse, - DisconnectAnalyzer, } from './runner-types'; import { TaskBroker, type MessageCallback, type TaskRunner } from './task-broker.service'; @@ -35,7 +34,7 @@ export class TaskRunnerWsServer { return this.disconnectAnalyzer; } - sendMessage(id: TaskRunner['id'], message: N8nMessage.ToRunner.All) { + sendMessage(id: TaskRunner['id'], message: BrokerMessage.ToRunner.All) { this.runnerConnections.get(id)?.send(JSON.stringify(message)); } @@ -49,9 +48,9 @@ export class TaskRunnerWsServer { try { const buffer = Array.isArray(data) ? Buffer.concat(data) : Buffer.from(data); - const message: RunnerMessage.ToN8n.All = JSON.parse( + const message: RunnerMessage.ToBroker.All = JSON.parse( buffer.toString('utf8'), - ) as RunnerMessage.ToN8n.All; + ) as RunnerMessage.ToBroker.All; if (!isConnected && message.type !== 'runner:info') { return; @@ -94,7 +93,7 @@ export class TaskRunnerWsServer { connection.on('message', onMessage); connection.send( - JSON.stringify({ type: 'broker:inforequest' } as N8nMessage.ToRunner.InfoRequest), + JSON.stringify({ type: 'broker:inforequest' } as BrokerMessage.ToRunner.InfoRequest), ); } diff --git a/packages/cli/src/runners/task-broker.service.ts b/packages/cli/src/runners/task-broker.service.ts index 754d99ef32..7d00cf232e 100644 --- a/packages/cli/src/runners/task-broker.service.ts +++ b/packages/cli/src/runners/task-broker.service.ts @@ -1,3 +1,9 @@ +import type { + BrokerMessage, + RequesterMessage, + RunnerMessage, + TaskResultData, +} from '@n8n/task-runner'; import { ApplicationError } from 'n8n-workflow'; import { nanoid } from 'nanoid'; import { Service } from 'typedi'; @@ -6,7 +12,6 @@ import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials'; import { Logger } from '@/logging/logger.service'; import { TaskRejectError } from './errors'; -import type { N8nMessage, RunnerMessage, RequesterMessage, TaskResultData } from './runner-types'; export interface TaskRunner { id: string; @@ -38,13 +43,15 @@ export interface TaskRequest { acceptInProgress?: boolean; } -export type MessageCallback = (message: N8nMessage.ToRunner.All) => Promise | void; +export type MessageCallback = (message: BrokerMessage.ToRunner.All) => Promise | void; export type RequesterMessageCallback = ( - message: N8nMessage.ToRequester.All, + message: BrokerMessage.ToRequester.All, ) => Promise | void; type RunnerAcceptCallback = () => void; -type RequesterAcceptCallback = (settings: RequesterMessage.ToN8n.TaskSettings['settings']) => void; +type RequesterAcceptCallback = ( + settings: RequesterMessage.ToBroker.TaskSettings['settings'], +) => void; type TaskRejectCallback = (reason: TaskRejectError) => void; @Service() @@ -134,11 +141,11 @@ export class TaskBroker { this.requesters.delete(requesterId); } - private async messageRunner(runnerId: TaskRunner['id'], message: N8nMessage.ToRunner.All) { + private async messageRunner(runnerId: TaskRunner['id'], message: BrokerMessage.ToRunner.All) { await this.knownRunners.get(runnerId)?.messageCallback(message); } - private async messageAllRunners(message: N8nMessage.ToRunner.All) { + private async messageAllRunners(message: BrokerMessage.ToRunner.All) { await Promise.allSettled( [...this.knownRunners.values()].map(async (runner) => { await runner.messageCallback(message); @@ -146,11 +153,11 @@ export class TaskBroker { ); } - private async messageRequester(requesterId: string, message: N8nMessage.ToRequester.All) { + private async messageRequester(requesterId: string, message: BrokerMessage.ToRequester.All) { await this.requesters.get(requesterId)?.(message); } - async onRunnerMessage(runnerId: TaskRunner['id'], message: RunnerMessage.ToN8n.All) { + async onRunnerMessage(runnerId: TaskRunner['id'], message: RunnerMessage.ToBroker.All) { const runner = this.knownRunners.get(runnerId); if (!runner) { return; @@ -193,7 +200,7 @@ export class TaskBroker { async handleRpcRequest( taskId: Task['id'], callId: string, - name: RunnerMessage.ToN8n.RPC['name'], + name: RunnerMessage.ToBroker.RPC['name'], params: unknown[], ) { const task = this.tasks.get(taskId); @@ -227,8 +234,8 @@ export class TaskBroker { async handleDataRequest( taskId: Task['id'], - requestId: RunnerMessage.ToN8n.TaskDataRequest['requestId'], - requestParams: RunnerMessage.ToN8n.TaskDataRequest['requestParams'], + requestId: RunnerMessage.ToBroker.TaskDataRequest['requestId'], + requestParams: RunnerMessage.ToBroker.TaskDataRequest['requestParams'], ) { const task = this.tasks.get(taskId); if (!task) { @@ -244,7 +251,7 @@ export class TaskBroker { async handleResponse( taskId: Task['id'], - requestId: RunnerMessage.ToN8n.TaskDataRequest['requestId'], + requestId: RunnerMessage.ToBroker.TaskDataRequest['requestId'], data: unknown, ) { const task = this.tasks.get(taskId); @@ -259,7 +266,7 @@ export class TaskBroker { }); } - async onRequesterMessage(requesterId: string, message: RequesterMessage.ToN8n.All) { + async onRequesterMessage(requesterId: string, message: RequesterMessage.ToBroker.All) { switch (message.type) { case 'requester:tasksettings': this.handleRequesterAccept(message.taskId, message.settings); @@ -291,7 +298,7 @@ export class TaskBroker { async handleRequesterRpcResponse( taskId: string, callId: string, - status: RequesterMessage.ToN8n.RPCResponse['status'], + status: RequesterMessage.ToBroker.RPCResponse['status'], data: unknown, ) { const runner = await this.getRunnerOrFailTask(taskId); @@ -317,7 +324,7 @@ export class TaskBroker { handleRequesterAccept( taskId: Task['id'], - settings: RequesterMessage.ToN8n.TaskSettings['settings'], + settings: RequesterMessage.ToBroker.TaskSettings['settings'], ) { const acceptReject = this.requesterAcceptRejects.get(taskId); if (acceptReject) { @@ -467,10 +474,12 @@ export class TaskBroker { this.pendingTaskRequests.splice(requestIndex, 1); try { - const acceptPromise = new Promise( + const acceptPromise = new Promise( (resolve, reject) => { this.requesterAcceptRejects.set(taskId, { - accept: resolve as (settings: RequesterMessage.ToN8n.TaskSettings['settings']) => void, + accept: resolve as ( + settings: RequesterMessage.ToBroker.TaskSettings['settings'], + ) => void, reject, }); diff --git a/packages/cli/src/runners/task-managers/__tests__/data-request-response-builder.test.ts b/packages/cli/src/runners/task-managers/__tests__/data-request-response-builder.test.ts index 8fc0198488..ea1492f64e 100644 --- a/packages/cli/src/runners/task-managers/__tests__/data-request-response-builder.test.ts +++ b/packages/cli/src/runners/task-managers/__tests__/data-request-response-builder.test.ts @@ -1,9 +1,9 @@ +import type { 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 { DataRequestResponseBuilder } from '../data-request-response-builder'; -import type { TaskData } from '../task-manager'; const triggerNode: INode = mock({ name: 'Trigger', diff --git a/packages/cli/src/runners/task-managers/data-request-response-builder.ts b/packages/cli/src/runners/task-managers/data-request-response-builder.ts index 6f49743aeb..bc498c33a7 100644 --- a/packages/cli/src/runners/task-managers/data-request-response-builder.ts +++ b/packages/cli/src/runners/task-managers/data-request-response-builder.ts @@ -1,3 +1,9 @@ +import type { + DataRequestResponse, + BrokerMessage, + PartialAdditionalData, + TaskData, +} from '@n8n/task-runner'; import type { EnvProviderState, IExecuteData, @@ -11,9 +17,6 @@ import type { WorkflowParameters, } from 'n8n-workflow'; -import type { DataRequestResponse, PartialAdditionalData, TaskData } from './task-manager'; -import type { N8nMessage } from '../runner-types'; - /** * 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. @@ -23,7 +26,7 @@ export class DataRequestResponseBuilder { constructor( private readonly taskData: TaskData, - private readonly requestParams: N8nMessage.ToRequester.TaskDataRequest['requestParams'], + private readonly requestParams: BrokerMessage.ToRequester.TaskDataRequest['requestParams'], ) { this.requestedNodeNames = new Set(requestParams.dataOfNodes); diff --git a/packages/cli/src/runners/task-managers/local-task-manager.ts b/packages/cli/src/runners/task-managers/local-task-manager.ts index a8fca01b2c..623b9c912e 100644 --- a/packages/cli/src/runners/task-managers/local-task-manager.ts +++ b/packages/cli/src/runners/task-managers/local-task-manager.ts @@ -1,7 +1,7 @@ +import type { RequesterMessage } from '@n8n/task-runner'; import Container from 'typedi'; import { TaskManager } from './task-manager'; -import type { RequesterMessage } from '../runner-types'; import type { RequesterMessageCallback } from '../task-broker.service'; import { TaskBroker } from '../task-broker.service'; @@ -24,7 +24,7 @@ export class LocalTaskManager extends TaskManager { ); } - sendMessage(message: RequesterMessage.ToN8n.All) { + sendMessage(message: RequesterMessage.ToBroker.All) { void this.taskBroker.onRequesterMessage(this.id, message); } } diff --git a/packages/cli/src/runners/task-managers/task-manager.ts b/packages/cli/src/runners/task-managers/task-manager.ts index d1ef30665a..617008a450 100644 --- a/packages/cli/src/runners/task-managers/task-manager.ts +++ b/packages/cli/src/runners/task-managers/task-manager.ts @@ -1,30 +1,24 @@ -import { - type EnvProviderState, - type IExecuteFunctions, - type Workflow, - type IRunExecutionData, - type INodeExecutionData, - type ITaskDataConnections, - type INode, - type WorkflowParameters, - type INodeParameters, - type WorkflowExecuteMode, - type IExecuteData, - type IDataObject, - type IWorkflowExecuteAdditionalData, - type Result, - createResultOk, - createResultError, +import type { TaskResultData, RequesterMessage, BrokerMessage, TaskData } from '@n8n/task-runner'; +import { RPC_ALLOW_LIST } from '@n8n/task-runner'; +import type { + EnvProviderState, + IExecuteFunctions, + Workflow, + IRunExecutionData, + INodeExecutionData, + ITaskDataConnections, + INode, + INodeParameters, + WorkflowExecuteMode, + IExecuteData, + IDataObject, + IWorkflowExecuteAdditionalData, + Result, } from 'n8n-workflow'; +import { createResultOk, createResultError } from 'n8n-workflow'; import { nanoid } from 'nanoid'; import { DataRequestResponseBuilder } from './data-request-response-builder'; -import { - RPC_ALLOW_LIST, - type TaskResultData, - type N8nMessage, - type RequesterMessage, -} from '../runner-types'; export type RequestAccept = (jobId: string) => void; export type RequestReject = (reason: string) => void; @@ -32,62 +26,6 @@ export type RequestReject = (reason: string) => void; export type TaskAccept = (data: TaskResultData) => void; export type TaskReject = (error: unknown) => void; -export interface TaskData { - executeFunctions: IExecuteFunctions; - inputData: ITaskDataConnections; - node: INode; - - workflow: Workflow; - runExecutionData: IRunExecutionData; - runIndex: number; - itemIndex: number; - activeNodeName: string; - connectionInputData: INodeExecutionData[]; - siblingParameters: INodeParameters; - mode: WorkflowExecuteMode; - envProviderState: EnvProviderState; - executeData?: IExecuteData; - defaultReturnRunIndex: number; - selfData: IDataObject; - contextNodeName: string; - additionalData: IWorkflowExecuteAdditionalData; -} - -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 { - workflow: Omit; - inputData: ITaskDataConnections; - 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; - additionalData: PartialAdditionalData; -} - export interface TaskRequest { requestId: string; taskType: string; @@ -219,9 +157,9 @@ export class TaskManager { } } - sendMessage(_message: RequesterMessage.ToN8n.All) {} + sendMessage(_message: RequesterMessage.ToBroker.All) {} - onMessage(message: N8nMessage.ToRequester.All) { + onMessage(message: BrokerMessage.ToRequester.All) { switch (message.type) { case 'broker:taskready': this.taskReady(message.requestId, message.taskId); @@ -282,7 +220,7 @@ export class TaskManager { sendTaskData( taskId: string, requestId: string, - requestParams: N8nMessage.ToRequester.TaskDataRequest['requestParams'], + requestParams: BrokerMessage.ToRequester.TaskDataRequest['requestParams'], ) { const job = this.tasks.get(taskId); if (!job) { @@ -304,7 +242,7 @@ export class TaskManager { async handleRpc( taskId: string, callId: string, - name: N8nMessage.ToRequester.RPC['name'], + name: BrokerMessage.ToRequester.RPC['name'], params: unknown[], ) { const job = this.tasks.get(taskId); diff --git a/packages/cli/src/runners/task-runner-process.ts b/packages/cli/src/runners/task-runner-process.ts index 2eaa62621c..9e731a99c5 100644 --- a/packages/cli/src/runners/task-runner-process.ts +++ b/packages/cli/src/runners/task-runner-process.ts @@ -92,7 +92,7 @@ export class TaskRunnerProcess extends TypedEmitter { } startNode(grantToken: string, n8nUri: string) { - const startScript = require.resolve('@n8n/task-runner'); + const startScript = require.resolve('@n8n/task-runner/start'); return spawn('node', [startScript], { env: this.getProcessEnvVars(grantToken, n8nUri), diff --git a/packages/cli/test/integration/runners/task-runner-module.internal.test.ts b/packages/cli/test/integration/runners/task-runner-module.internal.test.ts index f53adde5e7..922f7fee4b 100644 --- a/packages/cli/test/integration/runners/task-runner-module.internal.test.ts +++ b/packages/cli/test/integration/runners/task-runner-module.internal.test.ts @@ -8,6 +8,7 @@ import { TaskRunnerWsServer } from '../../../src/runners/runner-ws-server'; describe('TaskRunnerModule in internal_childprocess mode', () => { const runnerConfig = Container.get(TaskRunnersConfig); + runnerConfig.port = 0; // Random port runnerConfig.mode = 'internal_childprocess'; const module = Container.get(TaskRunnerModule);