From ae1fb77203ed58220eed5e400a0321d9f56e7018 Mon Sep 17 00:00:00 2001 From: Marc Littlemore Date: Mon, 12 Jan 2026 08:59:07 +0000 Subject: [PATCH] feat(core): Add execution source to workflow-executed event (#24058) --- .../__tests__/active-workflow-manager.test.ts | 1 + packages/cli/src/active-workflow-manager.ts | 11 ++ packages/cli/src/commands/execute.ts | 9 + .../__tests__/test-runner.service.ee.test.ts | 7 + .../test-runner/test-runner.service.ee.ts | 10 ++ .../log-streaming-event-relay.test.ts | 163 ++++++++++++++++++ .../cli/src/events/maps/relay.event-map.ts | 20 +-- .../relays/log-streaming.event-relay.ts | 40 ++++- packages/cli/src/webhooks/webhook-helpers.ts | 8 + .../src/workflow-execute-additional-data.ts | 8 + .../workflow-execution.service.test.ts | 4 + .../workflows/workflow-execution.service.ts | 19 +- 12 files changed, 285 insertions(+), 15 deletions(-) diff --git a/packages/cli/src/__tests__/active-workflow-manager.test.ts b/packages/cli/src/__tests__/active-workflow-manager.test.ts index bebddf12fc1..10d5bf703c2 100644 --- a/packages/cli/src/__tests__/active-workflow-manager.test.ts +++ b/packages/cli/src/__tests__/active-workflow-manager.test.ts @@ -33,6 +33,7 @@ describe('ActiveWorkflowManager', () => { mock(), mock(), mock(), + mock(), ); }); diff --git a/packages/cli/src/active-workflow-manager.ts b/packages/cli/src/active-workflow-manager.ts index 3610890f8de..c602bcc18bc 100644 --- a/packages/cli/src/active-workflow-manager.ts +++ b/packages/cli/src/active-workflow-manager.ts @@ -48,6 +48,7 @@ import { strict } from 'node:assert'; import { ActivationErrorsService } from '@/activation-errors.service'; import { ActiveExecutions } from '@/active-executions'; +import { EventService } from '@/events/event.service'; import { executeErrorWorkflow } from '@/execution-lifecycle/execute-error-workflow'; import { ExecutionService } from '@/executions/execution.service'; import { ExternalHooks } from '@/external-hooks'; @@ -92,6 +93,7 @@ export class ActiveWorkflowManager { private readonly publisher: Publisher, private readonly workflowsConfig: WorkflowsConfig, private readonly push: Push, + private readonly eventService: EventService, ) { this.logger = this.logger.scoped(['workflow-activation']); } @@ -360,6 +362,15 @@ export class ActiveWorkflowManager { responsePromise, ); + void executePromise.then((executionId) => { + this.eventService.emit('workflow-executed', { + workflowId: workflowData.id, + workflowName: workflowData.name, + executionId, + source: 'trigger', + }); + }); + if (donePromise) { void executePromise.then((executionId) => { this.activeExecutions diff --git a/packages/cli/src/commands/execute.ts b/packages/cli/src/commands/execute.ts index 77a4beaf0f8..22be718f5d5 100644 --- a/packages/cli/src/commands/execute.ts +++ b/packages/cli/src/commands/execute.ts @@ -6,6 +6,7 @@ import { ExecutionBaseError, UnexpectedError, UserError } from 'n8n-workflow'; import { z } from 'zod'; import { ActiveExecutions } from '@/active-executions'; +import { EventService } from '@/events/event.service'; import { OwnershipService } from '@/services/ownership.service'; import { findCliWorkflowStart, isWorkflowIdValid } from '@/utils'; import { WorkflowRunner } from '@/workflow-runner'; @@ -94,6 +95,14 @@ export class Execute extends BaseCommand> { const executionId = await workflowRunner.run(runData); + Container.get(EventService).emit('workflow-executed', { + user: { id: user.id }, + workflowId: workflowData.id, + workflowName: workflowData.name, + executionId, + source: 'cli', + }); + const activeExecutions = Container.get(ActiveExecutions); const data = await activeExecutions.getPostExecutePromise(executionId); diff --git a/packages/cli/src/evaluation.ee/test-runner/__tests__/test-runner.service.ee.test.ts b/packages/cli/src/evaluation.ee/test-runner/__tests__/test-runner.service.ee.test.ts index 2409b419fa3..26114a9bf48 100644 --- a/packages/cli/src/evaluation.ee/test-runner/__tests__/test-runner.service.ee.test.ts +++ b/packages/cli/src/evaluation.ee/test-runner/__tests__/test-runner.service.ee.test.ts @@ -59,6 +59,7 @@ describe('TestRunnerService', () => { testCaseExecutionRepository, errorReporter, executionsConfig, + mock(), ); testRunRepository.createTestRun.mockResolvedValue(mock({ id: 'test-run-id' })); @@ -497,6 +498,7 @@ describe('TestRunnerService', () => { testCaseExecutionRepository, errorReporter, queueModeConfig, + mock(), ); process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS = 'true'; @@ -679,6 +681,8 @@ describe('TestRunnerService', () => { // Setup test data const triggerNodeName = 'TriggerNode'; const workflow = mock({ + id: 'workflow-id', + name: 'Test Workflow', nodes: [ { id: 'node1', @@ -807,6 +811,7 @@ describe('TestRunnerService', () => { testCaseExecutionRepository, errorReporter, queueModeConfig, + mock(), ); }); @@ -814,6 +819,8 @@ describe('TestRunnerService', () => { // Setup test data const triggerNodeName = 'TriggerNode'; const workflow = mock({ + id: 'workflow-id', + name: 'Test Workflow', nodes: [ { id: 'node1', diff --git a/packages/cli/src/evaluation.ee/test-runner/test-runner.service.ee.ts b/packages/cli/src/evaluation.ee/test-runner/test-runner.service.ee.ts index fd274218a85..16978657f3b 100644 --- a/packages/cli/src/evaluation.ee/test-runner/test-runner.service.ee.ts +++ b/packages/cli/src/evaluation.ee/test-runner/test-runner.service.ee.ts @@ -27,6 +27,7 @@ import assert from 'node:assert'; import { JsonObject } from 'openid-client'; import { ActiveExecutions } from '@/active-executions'; +import { EventService } from '@/events/event.service'; import { TestCaseExecutionError, TestRunError } from '@/evaluation.ee/test-runner/errors.ee'; import { checkNodeParameterNotEmpty, @@ -70,6 +71,7 @@ export class TestRunnerService { private readonly testCaseExecutionRepository: TestCaseExecutionRepository, private readonly errorReporter: ErrorReporter, private readonly executionsConfig: ExecutionsConfig, + private readonly eventService: EventService, ) {} /** @@ -280,6 +282,14 @@ export class TestRunnerService { const executionId = await this.workflowRunner.run(data); assert(executionId); + this.eventService.emit('workflow-executed', { + user: metadata.userId ? { id: metadata.userId } : undefined, + workflowId: workflow.id, + workflowName: workflow.name, + executionId, + source: 'evaluation', + }); + // Listen to the abort signal to stop the execution in case test run is cancelled abortSignal.addEventListener('abort', () => { this.activeExecutions.stopExecution( diff --git a/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts b/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts index f02d550cf26..83106741f0c 100644 --- a/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts +++ b/packages/cli/src/events/__tests__/log-streaming-event-relay.test.ts @@ -1732,6 +1732,169 @@ describe('LogStreamingEventRelay', () => { }, }); }); + + it('should log on `workflow-executed` event for webhook execution', () => { + const event: RelayEventMap['workflow-executed'] = { + workflowId: 'wf-webhook', + workflowName: 'Webhook Test Workflow', + executionId: 'exec-webhook-123', + source: 'webhook', + }; + + eventService.emit('workflow-executed', event); + + expect(eventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.workflow.executed', + payload: { + workflowId: 'wf-webhook', + workflowName: 'Webhook Test Workflow', + executionId: 'exec-webhook-123', + source: 'webhook', + }, + }); + }); + + it('should log on `workflow-executed` event for trigger execution', () => { + const event: RelayEventMap['workflow-executed'] = { + workflowId: 'wf-trigger', + workflowName: 'Trigger Test Workflow', + executionId: 'exec-trigger-123', + source: 'trigger', + }; + + eventService.emit('workflow-executed', event); + + expect(eventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.workflow.executed', + payload: { + workflowId: 'wf-trigger', + workflowName: 'Trigger Test Workflow', + executionId: 'exec-trigger-123', + source: 'trigger', + }, + }); + }); + + it('should log on `workflow-executed` event for error workflow execution', () => { + const event: RelayEventMap['workflow-executed'] = { + workflowId: 'wf-error', + workflowName: 'Error Test Workflow', + executionId: 'exec-error-123', + source: 'error', + }; + + eventService.emit('workflow-executed', event); + + expect(eventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.workflow.executed', + payload: { + workflowId: 'wf-error', + workflowName: 'Error Test Workflow', + executionId: 'exec-error-123', + source: 'error', + }, + }); + }); + + it('should log on `workflow-executed` event for CLI execution', () => { + const event: RelayEventMap['workflow-executed'] = { + user: { + id: 'user-cli', + }, + workflowId: 'wf-cli', + workflowName: 'CLI Test Workflow', + executionId: 'exec-cli-123', + source: 'cli', + }; + + eventService.emit('workflow-executed', event); + + expect(eventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.workflow.executed', + payload: { + userId: 'user-cli', + workflowId: 'wf-cli', + workflowName: 'CLI Test Workflow', + executionId: 'exec-cli-123', + source: 'cli', + }, + }); + }); + + it('should log on `workflow-executed` event for integrated/subworkflow execution', () => { + const event: RelayEventMap['workflow-executed'] = { + user: { + id: 'user-integrated', + }, + workflowId: 'wf-integrated', + workflowName: 'Integrated Test Workflow', + executionId: 'exec-integrated-123', + source: 'integrated', + }; + + eventService.emit('workflow-executed', event); + + expect(eventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.workflow.executed', + payload: { + userId: 'user-integrated', + workflowId: 'wf-integrated', + workflowName: 'Integrated Test Workflow', + executionId: 'exec-integrated-123', + source: 'integrated', + }, + }); + }); + + it('should log on `workflow-executed` event for evaluation execution', () => { + const event: RelayEventMap['workflow-executed'] = { + user: { + id: 'user-eval', + }, + workflowId: 'wf-evaluation', + workflowName: 'Evaluation Test Workflow', + executionId: 'exec-evaluation-123', + source: 'evaluation', + }; + + eventService.emit('workflow-executed', event); + + expect(eventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.workflow.executed', + payload: { + userId: 'user-eval', + workflowId: 'wf-evaluation', + workflowName: 'Evaluation Test Workflow', + executionId: 'exec-evaluation-123', + source: 'evaluation', + }, + }); + }); + + it('should log on `workflow-executed` event for chat execution', () => { + const event: RelayEventMap['workflow-executed'] = { + user: { + id: 'user-chat', + }, + workflowId: 'wf-chat', + workflowName: 'Chat Test Workflow', + executionId: 'exec-chat-123', + source: 'chat', + }; + + eventService.emit('workflow-executed', event); + + expect(eventBus.sendAuditEvent).toHaveBeenCalledWith({ + eventName: 'n8n.audit.workflow.executed', + payload: { + userId: 'user-chat', + workflowId: 'wf-chat', + workflowName: 'Chat Test Workflow', + executionId: 'exec-chat-123', + source: 'chat', + }, + }); + }); }); describe('AI events', () => { diff --git a/packages/cli/src/events/maps/relay.event-map.ts b/packages/cli/src/events/maps/relay.event-map.ts index faf14cfb6e0..f47a60aa1ed 100644 --- a/packages/cli/src/events/maps/relay.event-map.ts +++ b/packages/cli/src/events/maps/relay.event-map.ts @@ -130,16 +130,16 @@ export type RelayEventMap = { workflowId: string; workflowName: string; executionId: string; - source: 'user-manual' | 'user-retry'; - // TODO: To be added in future PR - // | 'webhook' - // | 'trigger' - // | 'error' - // | 'cli' - // | 'integrated' - // | 'internal' - // | 'evaluation' - // | 'chat'; + source: + | 'user-manual' + | 'user-retry' + | 'webhook' + | 'trigger' + | 'error' + | 'cli' + | 'integrated' + | 'evaluation' + | 'chat'; }; // #endregion diff --git a/packages/cli/src/events/relays/log-streaming.event-relay.ts b/packages/cli/src/events/relays/log-streaming.event-relay.ts index 77281f5a5df..360597500df 100644 --- a/packages/cli/src/events/relays/log-streaming.event-relay.ts +++ b/packages/cli/src/events/relays/log-streaming.event-relay.ts @@ -5,9 +5,16 @@ import type { IWorkflowBase } from 'n8n-workflow'; import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus'; import { EventService } from '@/events/event.service'; -import type { RelayEventMap } from '@/events/maps/relay.event-map'; +import type { RelayEventMap, UserLike } from '@/events/maps/relay.event-map'; import { EventRelay } from '@/events/relays/event-relay'; +type WorkflowExecutedEvent = RelayEventMap['workflow-executed']; +type WorkflowExecutedEventWithUser = WorkflowExecutedEvent & { user: UserLike }; + +function hasUser(event: WorkflowExecutedEvent): event is WorkflowExecutedEventWithUser { + return event.user !== undefined; +} + @Service() export class LogStreamingEventRelay extends EventRelay { constructor( @@ -251,17 +258,17 @@ export class LogStreamingEventRelay extends EventRelay { } @Redactable() - private workflowExecuted({ + private workflowExecutedWithUser({ user, workflowId, workflowName, executionId, source, - }: RelayEventMap['workflow-executed']) { + }: WorkflowExecutedEventWithUser) { void this.eventBus.sendAuditEvent({ eventName: 'n8n.audit.workflow.executed', payload: { - ...(user && { ...user }), + ...user, workflowId, workflowName, executionId, @@ -270,6 +277,31 @@ export class LogStreamingEventRelay extends EventRelay { }); } + private workflowExecutedWithoutUser({ + workflowId, + workflowName, + executionId, + source, + }: WorkflowExecutedEvent) { + void this.eventBus.sendAuditEvent({ + eventName: 'n8n.audit.workflow.executed', + payload: { + workflowId, + workflowName, + executionId, + source, + }, + }); + } + + private workflowExecuted(event: WorkflowExecutedEvent) { + if (hasUser(event)) { + this.workflowExecutedWithUser(event); + } else { + this.workflowExecutedWithoutUser(event); + } + } + // #endregion // #region Node diff --git a/packages/cli/src/webhooks/webhook-helpers.ts b/packages/cli/src/webhooks/webhook-helpers.ts index 08e5fcf3fd9..6c42ef54a79 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/packages/cli/src/webhooks/webhook-helpers.ts @@ -58,6 +58,7 @@ import type { import { ActiveExecutions } from '@/active-executions'; import { MCP_TRIGGER_NODE_TYPE } from '@/constants'; +import { EventService } from '@/events/event.service'; import { InternalServerError } from '@/errors/response-errors/internal-server.error'; import { NotFoundError } from '@/errors/response-errors/not-found.error'; import { UnprocessableRequestError } from '@/errors/response-errors/unprocessable.error'; @@ -635,6 +636,13 @@ export async function executeWebhook( responsePromise, ); + Container.get(EventService).emit('workflow-executed', { + workflowId: workflowData.id, + workflowName: workflowData.name, + executionId, + source: 'webhook', + }); + if (responseMode === 'formPage' && !didSendResponse) { res.send({ formWaitingUrl: `${additionalData.formWaitingBaseUrl}/${executionId}` }); process.nextTick(() => res.end()); diff --git a/packages/cli/src/workflow-execute-additional-data.ts b/packages/cli/src/workflow-execute-additional-data.ts index 5cc7dce062c..9f33e34dc55 100644 --- a/packages/cli/src/workflow-execute-additional-data.ts +++ b/packages/cli/src/workflow-execute-additional-data.ts @@ -224,6 +224,14 @@ export async function executeWorkflow( const executionId = await activeExecutions.add(runData); + Container.get(EventService).emit('workflow-executed', { + user: additionalData.userId ? { id: additionalData.userId } : undefined, + workflowId: workflowData.id, + workflowName: workflowData.name, + executionId, + source: 'integrated', + }); + const executionPromise = startExecution( additionalData, options, diff --git a/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts b/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts index 29b5cec72f3..a115aab242d 100644 --- a/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts +++ b/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts @@ -89,6 +89,7 @@ describe('WorkflowExecutionService', () => { mock(), mock(), mock(), + mock(), ); const additionalData = mock({}); @@ -374,6 +375,7 @@ describe('WorkflowExecutionService', () => { mock(), mock(), mock(), + mock(), ); const runPayload: WorkflowRequest.FullManualExecutionFromKnownTriggerPayload = { @@ -554,6 +556,7 @@ describe('WorkflowExecutionService', () => { globalConfigMock, mock(), mock(), + mock(), ); }); @@ -698,6 +701,7 @@ describe('WorkflowExecutionService', () => { globalConfig, mock(), mock(), + mock(), ); await service.executeErrorWorkflow( diff --git a/packages/cli/src/workflows/workflow-execution.service.ts b/packages/cli/src/workflows/workflow-execution.service.ts index aaaf6b9c3cc..bc2c0016a13 100644 --- a/packages/cli/src/workflows/workflow-execution.service.ts +++ b/packages/cli/src/workflows/workflow-execution.service.ts @@ -25,6 +25,7 @@ import { createRunExecutionData, } from 'n8n-workflow'; +import { EventService } from '@/events/event.service'; import { ExecutionDataService } from '@/executions/execution-data.service'; import { SubworkflowPolicyChecker } from '@/executions/pre-execution-checks'; import type { IWorkflowErrorData } from '@/interfaces'; @@ -47,6 +48,7 @@ export class WorkflowExecutionService { private readonly globalConfig: GlobalConfig, private readonly subworkflowPolicyChecker: SubworkflowPolicyChecker, private readonly executionDataService: ExecutionDataService, + private readonly eventService: EventService, ) {} async runWorkflow( @@ -267,6 +269,14 @@ export class WorkflowExecutionService { const executionId = await this.workflowRunner.run(data, undefined, true); + this.eventService.emit('workflow-executed', { + user: { id: user.id }, + workflowId: workflowData.id, + workflowName: workflowData.name, + executionId, + source: 'chat', + }); + return { executionId, }; @@ -419,7 +429,14 @@ export class WorkflowExecutionService { projectId: runningProject.id, }; - await this.workflowRunner.run(runData); + const executionId = await this.workflowRunner.run(runData); + + this.eventService.emit('workflow-executed', { + workflowId, + workflowName: workflowData.name, + executionId, + source: 'error', + }); } catch (error) { this.errorReporter.error(error); this.logger.error(