refactor(core): Cache workflow ownership (#6738)

* refactor: Set up ownership service

* refactor: Specify cache keys and values

* refactor: Replace util with service calls

* test: Mock service in tests

* refactor: Use dependency injection

* test: Write tests

* refactor: Apply feedback from Omar and Micha

* test: Fix tests

* test: Fix missing spot

* refactor: Return user entity from cache

* refactor: More dependency injection!

packages/cli/src/UserManagement/PermissionChecker.ts

@@ -9,9 +9,11 @@ import { In } from 'typeorm';
 import * as Db from '@/Db';
 import config from '@/config';
 import type { SharedCredentials } from '@db/entities/SharedCredentials';
-import { getRoleId, getWorkflowOwner, isSharingEnabled } from './UserManagementHelper';
+import { getRoleId, isSharingEnabled } from './UserManagementHelper';
 import { WorkflowsService } from '@/workflows/workflows.services';
 import { UserService } from '@/user/user.service';
+import { OwnershipService } from '@/services/ownership.service';
+import Container from 'typedi';
 
 export class PermissionChecker {
 	/**
@@ -101,7 +103,9 @@ export class PermissionChecker {
 			policy = 'workflowsFromSameOwner';
 		}
 
-		const subworkflowOwner = await getWorkflowOwner(subworkflow.id);
+		const subworkflowOwner = await Container.get(OwnershipService).getWorkflowOwnerCached(
+			subworkflow.id,
+		);
 
 		const errorToThrow = new SubworkflowOperationError(
 			`Target workflow ID ${subworkflow.id ?? ''} may not be called`,

packages/cli/src/UserManagement/UserManagementHelper.ts

@@ -14,17 +14,6 @@ import { License } from '@/License';
 import { getWebhookBaseUrl } from '@/WebhookHelpers';
 import type { PostHogClient } from '@/posthog';
 
-export async function getWorkflowOwner(workflowId: string): Promise<User> {
-	const workflowOwnerRole = await Container.get(RoleRepository).findWorkflowOwnerRole();
-
-	const sharedWorkflow = await Db.collections.SharedWorkflow.findOneOrFail({
-		where: { workflowId, roleId: workflowOwnerRole?.id ?? undefined },
-		relations: ['user', 'user.globalRole'],
-	});
-
-	return sharedWorkflow.user;
-}
-
 export function isEmailSetUp(): boolean {
 	const smtp = config.getEnv('userManagement.emails.mode') === 'smtp';
 	const host = !!config.getEnv('userManagement.emails.smtp.host');

packages/cli/src/WaitTracker.ts

@@ -16,10 +16,10 @@ import type {
 	IWorkflowExecutionDataProcess,
 } from '@/Interfaces';
 import { WorkflowRunner } from '@/WorkflowRunner';
-import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
 import { recoverExecutionDataFromEventLogMessages } from './eventbus/MessageEventBus/recoverEvents';
 import { ExecutionRepository } from '@db/repositories';
 import type { ExecutionEntity } from '@db/entities/ExecutionEntity';
+import { OwnershipService } from './services/ownership.service';
 
 @Service()
 export class WaitTracker {
@@ -32,7 +32,10 @@ export class WaitTracker {
 
 	mainTimer: NodeJS.Timeout;
 
-	constructor(private executionRepository: ExecutionRepository) {
+	constructor(
+		private executionRepository: ExecutionRepository,
+		private ownershipService: OwnershipService,
+	) {
 		// Poll every 60 seconds a list of upcoming executions
 		this.mainTimer = setInterval(() => {
 			void this.getWaitingExecutions();
@@ -180,7 +183,8 @@ export class WaitTracker {
 			if (!fullExecutionData.workflowData.id) {
 				throw new Error('Only saved workflows can be resumed.');
 			}
-			const user = await getWorkflowOwner(fullExecutionData.workflowData.id);
+			const workflowId = fullExecutionData.workflowData.id;
+			const user = await this.ownershipService.getWorkflowOwnerCached(workflowId);
 
 			const data: IWorkflowExecutionDataProcess = {
 				executionMode: fullExecutionData.mode,

packages/cli/src/WaitingWebhooks.ts

@@ -8,14 +8,15 @@ import * as WebhookHelpers from '@/WebhookHelpers';
 import { NodeTypes } from '@/NodeTypes';
 import type { IExecutionResponse, IResponseCallbackData, IWorkflowDb } from '@/Interfaces';
 import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData';
-import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
 import { ExecutionRepository } from '@db/repositories';
+import { OwnershipService } from './services/ownership.service';
 
 @Service()
 export class WaitingWebhooks {
 	constructor(
 		private nodeTypes: NodeTypes,
 		private executionRepository: ExecutionRepository,
+		private ownershipService: OwnershipService,
 	) {}
 
 	async executeWebhook(
@@ -83,7 +84,7 @@ export class WaitingWebhooks {
 		const { workflowData } = fullExecutionData;
 
 		const workflow = new Workflow({
-			id: workflowData.id!.toString(),
+			id: workflowData.id!,
 			name: workflowData.name,
 			nodes: workflowData.nodes,
 			connections: workflowData.connections,
@@ -95,7 +96,7 @@ export class WaitingWebhooks {
 
 		let workflowOwner;
 		try {
-			workflowOwner = await getWorkflowOwner(workflowData.id!.toString());
+			workflowOwner = await this.ownershipService.getWorkflowOwnerCached(workflowData.id!);
 		} catch (error) {
 			throw new ResponseHelper.NotFoundError('Could not find workflow');
 		}

packages/cli/src/WebhookHelpers.ts

@@ -58,8 +58,8 @@ import * as WorkflowExecuteAdditionalData from '@/WorkflowExecuteAdditionalData'
 import { ActiveExecutions } from '@/ActiveExecutions';
 import type { User } from '@db/entities/User';
 import type { WorkflowEntity } from '@db/entities/WorkflowEntity';
-import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
 import { EventsService } from '@/services/events.service';
+import { OwnershipService } from './services/ownership.service';
 
 const pipeline = promisify(stream.pipeline);
 
@@ -175,7 +175,7 @@ export async function executeWebhook(
 		user = (workflowData as WorkflowEntity).shared[0].user;
 	} else {
 		try {
-			user = await getWorkflowOwner(workflowData.id);
+			user = await Container.get(OwnershipService).getWorkflowOwnerCached(workflowData.id);
 		} catch (error) {
 			throw new ResponseHelper.NotFoundError('Cannot find workflow');
 		}

packages/cli/src/WorkflowExecuteAdditionalData.ts

@@ -58,7 +58,6 @@ import { NodeTypes } from '@/NodeTypes';
 import { Push } from '@/push';
 import * as WebhookHelpers from '@/WebhookHelpers';
 import * as WorkflowHelpers from '@/WorkflowHelpers';
-import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
 import { findSubworkflowStart, isWorkflowIdValid } from '@/utils';
 import { PermissionChecker } from './UserManagement/PermissionChecker';
 import { WorkflowsService } from './workflows/workflows.services';
@@ -66,6 +65,7 @@ import { InternalHooks } from '@/InternalHooks';
 import type { ExecutionMetadata } from '@db/entities/ExecutionMetadata';
 import { ExecutionRepository } from '@db/repositories';
 import { EventsService } from '@/services/events.service';
+import { OwnershipService } from './services/ownership.service';
 
 const ERROR_TRIGGER_TYPE = config.getEnv('nodes.errorTriggerType');
 
@@ -146,7 +146,9 @@ export function executeErrorWorkflow(
 				// make sure there are no possible security gaps
 				return;
 			}
-			getWorkflowOwner(workflowId)
+
+			Container.get(OwnershipService)
+				.getWorkflowOwnerCached(workflowId)
 				.then((user) => {
 					void WorkflowHelpers.executeErrorWorkflow(errorWorkflow, workflowErrorData, user);
 				})
@@ -169,7 +171,9 @@ export function executeErrorWorkflow(
 			workflowData.nodes.some((node) => node.type === ERROR_TRIGGER_TYPE)
 		) {
 			Logger.verbose('Start internal error workflow', { executionId, workflowId });
-			void getWorkflowOwner(workflowId).then((user) => {
+			void Container.get(OwnershipService)
+				.getWorkflowOwnerCached(workflowId)
+				.then((user) => {
 					void WorkflowHelpers.executeErrorWorkflow(workflowId, workflowErrorData, user);
 				});
 		}

packages/cli/src/commands/worker.ts

@@ -18,11 +18,11 @@ import { PermissionChecker } from '@/UserManagement/PermissionChecker';
 import config from '@/config';
 import type { Job, JobId, JobQueue, JobResponse, WebhookResponse } from '@/Queue';
 import { Queue } from '@/Queue';
-import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
 import { generateFailedExecutionFromError } from '@/WorkflowHelpers';
 import { N8N_VERSION } from '@/constants';
 import { BaseCommand } from './BaseCommand';
 import { ExecutionRepository } from '@db/repositories';
+import { OwnershipService } from '@/services/ownership.service';
 
 export class Worker extends BaseCommand {
 	static description = '\nStarts a n8n worker';
@@ -112,7 +112,7 @@ export class Worker extends BaseCommand {
 			`Start job: ${job.id} (Workflow ID: ${workflowId} | Execution: ${executionId})`,
 		);
 
-		const workflowOwner = await getWorkflowOwner(workflowId);
+		const workflowOwner = await Container.get(OwnershipService).getWorkflowOwnerCached(workflowId);
 
 		let { staticData } = fullExecutionData.workflowData;
 		if (loadStaticData) {

packages/cli/src/services/cache.service.ts

@@ -9,6 +9,10 @@ import { LoggerProxy, jsonStringify } from 'n8n-workflow';
 
 @Service()
 export class CacheService {
+	/**
+	 * Keys and values:
+	 * - `'cache:workflow-owner:${workflowId}'`: `User`
+	 */
 	private cache: RedisCache | MemoryCache | undefined;
 
 	async init() {

packages/cli/src/services/events.service.ts

@@ -1,15 +1,18 @@
 import { EventEmitter } from 'events';
-import { Service } from 'typedi';
+import Container, { Service } from 'typedi';
 import type { INode, IRun, IWorkflowBase } from 'n8n-workflow';
 import { LoggerProxy } from 'n8n-workflow';
 import { StatisticsNames } from '@db/entities/WorkflowStatistics';
 import { WorkflowStatisticsRepository } from '@db/repositories';
-import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
 import { UserService } from '@/user/user.service';
+import { OwnershipService } from './ownership.service';
 
 @Service()
 export class EventsService extends EventEmitter {
-	constructor(private repository: WorkflowStatisticsRepository) {
+	constructor(
+		private repository: WorkflowStatisticsRepository,
+		private ownershipService: OwnershipService,
+	) {
 		super({ captureRejections: true });
 		if ('SKIP_STATISTICS_EVENTS' in process.env) return;
 
@@ -41,7 +44,7 @@ export class EventsService extends EventEmitter {
 			const upsertResult = await this.repository.upsertWorkflowStatistics(name, workflowId);
 
 			if (name === 'production_success' && upsertResult === 'insert') {
-				const owner = await getWorkflowOwner(workflowId);
+				const owner = await Container.get(OwnershipService).getWorkflowOwnerCached(workflowId);
 				const metrics = {
 					user_id: owner.id,
 					workflow_id: workflowId,
@@ -72,7 +75,7 @@ export class EventsService extends EventEmitter {
 		if (insertResult === 'failed') return;
 
 		// Compile the metrics since this was a new data loaded event
-		const owner = await getWorkflowOwner(workflowId);
+		const owner = await this.ownershipService.getWorkflowOwnerCached(workflowId);
 
 		let metrics = {
 			user_id: owner.id,

packages/cli/src/services/ownership.service.ts

@@ -0,0 +1,36 @@
+import { Service } from 'typedi';
+import { CacheService } from './cache.service';
+import { RoleRepository, SharedWorkflowRepository, UserRepository } from '@/databases/repositories';
+import type { User } from '@/databases/entities/User';
+
+@Service()
+export class OwnershipService {
+	constructor(
+		private cacheService: CacheService,
+		private userRepository: UserRepository,
+		private roleRepository: RoleRepository,
+		private sharedWorkflowRepository: SharedWorkflowRepository,
+	) {}
+
+	/**
+	 * Retrieve the user who owns the workflow. Note that workflow ownership is **immutable**.
+	 */
+	async getWorkflowOwnerCached(workflowId: string) {
+		const cachedValue = await this.cacheService.get<User>(`cache:workflow-owner:${workflowId}`);
+
+		if (cachedValue) return this.userRepository.create(cachedValue);
+
+		const workflowOwnerRole = await this.roleRepository.findWorkflowOwnerRole();
+
+		if (!workflowOwnerRole) throw new Error('Failed to find workflow owner role');
+
+		const sharedWorkflow = await this.sharedWorkflowRepository.findOneOrFail({
+			where: { workflowId, roleId: workflowOwnerRole.id },
+			relations: ['user', 'user.globalRole'],
+		});
+
+		void this.cacheService.set(`cache:workflow-owner:${workflowId}`, sharedWorkflow.user);
+
+		return sharedWorkflow.user;
+	}
+}

packages/cli/test/unit/PermissionChecker.test.ts

@@ -23,6 +23,7 @@ import * as testDb from '../integration/shared/testDb';
 import { mockNodeTypesData } from './Helpers';
 import type { SaveCredentialFunction } from '../integration/shared/types';
 import { mockInstance } from '../integration/shared/utils/';
+import { OwnershipService } from '@/services/ownership.service';
 
 let mockNodeTypes: INodeTypes;
 let credentialOwnerRole: Role;
@@ -221,6 +222,7 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
 	const userId = uuid();
 	const fakeUser = new User();
 	fakeUser.id = userId;
+	const ownershipService = mockInstance(OwnershipService);
 
 	const ownerMockRole = new Role();
 	ownerMockRole.name = 'owner';
@@ -234,7 +236,9 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
 
 	test('sets default policy from environment when subworkflow has none', async () => {
 		config.set('workflows.callerPolicyDefaultOption', 'none');
-		jest.spyOn(UserManagementHelper, 'getWorkflowOwner').mockImplementation(async (workflowId) => {
+		jest
+			.spyOn(ownershipService, 'getWorkflowOwnerCached')
+			.mockImplementation(async (workflowId) => {
 				return fakeUser;
 			});
 		jest.spyOn(UserManagementHelper, 'isSharingEnabled').mockReturnValue(true);
@@ -253,7 +257,7 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
 
 	test('if sharing is disabled, ensures that workflows are owner by same user', async () => {
 		jest
-			.spyOn(UserManagementHelper, 'getWorkflowOwner')
+			.spyOn(ownershipService, 'getWorkflowOwnerCached')
 			.mockImplementation(async (workflowId) => fakeUser);
 		jest.spyOn(UserManagementHelper, 'isSharingEnabled').mockReturnValue(false);
 		jest.spyOn(UserService, 'get').mockImplementation(async () => fakeUser);
@@ -287,7 +291,7 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
 	test('list of ids must include the parent workflow id', async () => {
 		const invalidParentWorkflowId = uuid();
 		jest
-			.spyOn(UserManagementHelper, 'getWorkflowOwner')
+			.spyOn(ownershipService, 'getWorkflowOwnerCached')
 			.mockImplementation(async (workflowId) => fakeUser);
 		jest.spyOn(UserManagementHelper, 'isSharingEnabled').mockReturnValue(true);
 		jest.spyOn(UserService, 'get').mockImplementation(async () => fakeUser);
@@ -313,7 +317,7 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
 
 	test('sameOwner passes when both workflows are owned by the same user', async () => {
 		jest
-			.spyOn(UserManagementHelper, 'getWorkflowOwner')
+			.spyOn(ownershipService, 'getWorkflowOwnerCached')
 			.mockImplementation(async (workflowId) => fakeUser);
 		jest.spyOn(UserManagementHelper, 'isSharingEnabled').mockReturnValue(false);
 		jest.spyOn(UserService, 'get').mockImplementation(async () => fakeUser);
@@ -336,7 +340,7 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
 	test('workflowsFromAList works when the list contains the parent id', async () => {
 		const workflowId = uuid();
 		jest
-			.spyOn(UserManagementHelper, 'getWorkflowOwner')
+			.spyOn(ownershipService, 'getWorkflowOwnerCached')
 			.mockImplementation(async (workflowId) => fakeUser);
 		jest.spyOn(UserManagementHelper, 'isSharingEnabled').mockReturnValue(true);
 		jest.spyOn(UserService, 'get').mockImplementation(async () => fakeUser);
@@ -362,7 +366,7 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
 
 	test('should not throw when workflow policy is set to any', async () => {
 		jest
-			.spyOn(UserManagementHelper, 'getWorkflowOwner')
+			.spyOn(ownershipService, 'getWorkflowOwnerCached')
 			.mockImplementation(async (workflowId) => fakeUser);
 		jest.spyOn(UserManagementHelper, 'isSharingEnabled').mockReturnValue(true);
 		jest.spyOn(UserService, 'get').mockImplementation(async () => fakeUser);

packages/cli/test/unit/services/events.service.test.ts

@@ -15,13 +15,15 @@ import type { WorkflowStatistics } from '@db/entities/WorkflowStatistics';
 import { WorkflowStatisticsRepository } from '@db/repositories';
 import { EventsService } from '@/services/events.service';
 import { UserService } from '@/user/user.service';
-import { getWorkflowOwner } from '@/UserManagement/UserManagementHelper';
+import { OwnershipService } from '@/services/ownership.service';
+import { mockInstance } from '../../integration/shared/utils';
 
 jest.mock('@/UserManagement/UserManagementHelper', () => ({ getWorkflowOwner: jest.fn() }));
 
 describe('EventsService', () => {
 	const dbType = config.getEnv('database.type');
 	const fakeUser = mock<User>({ id: 'abcde-fghij' });
+	const ownershipService = mockInstance(OwnershipService);
 
 	const entityManager = mock<EntityManager>();
 	const dataSource = mock<DataSource>({
@@ -36,10 +38,13 @@ describe('EventsService', () => {
 	LoggerProxy.init(mock<ILogger>());
 	config.set('diagnostics.enabled', true);
 	config.set('deployment.type', 'n8n-testing');
-	mocked(getWorkflowOwner).mockResolvedValue(fakeUser);
+	mocked(ownershipService.getWorkflowOwnerCached).mockResolvedValue(fakeUser);
 	const updateUserSettingsMock = jest.spyOn(UserService, 'updateUserSettings').mockImplementation();
 
-	const eventsService = new EventsService(new WorkflowStatisticsRepository(dataSource));
+	const eventsService = new EventsService(
+		new WorkflowStatisticsRepository(dataSource),
+		ownershipService,
+	);
 
 	const onFirstProductionWorkflowSuccess = jest.fn();
 	const onFirstWorkflowDataLoad = jest.fn();

packages/cli/test/unit/services/ownership.service.test.ts

@@ -0,0 +1,68 @@
+import { OwnershipService } from '@/services/ownership.service';
+import { RoleRepository, SharedWorkflowRepository, UserRepository } from '@/databases/repositories';
+import { mockInstance } from '../../integration/shared/utils';
+import { Role } from '@/databases/entities/Role';
+import { randomInteger } from '../../integration/shared/random';
+import { SharedWorkflow } from '@/databases/entities/SharedWorkflow';
+import { CacheService } from '@/services/cache.service';
+import { User } from '@/databases/entities/User';
+
+const wfOwnerRole = () =>
+	Object.assign(new Role(), {
+		scope: 'workflow',
+		name: 'owner',
+		id: randomInteger(),
+	});
+
+describe('OwnershipService', () => {
+	const cacheService = mockInstance(CacheService);
+	const roleRepository = mockInstance(RoleRepository);
+	const userRepository = mockInstance(UserRepository);
+	const sharedWorkflowRepository = mockInstance(SharedWorkflowRepository);
+
+	const ownershipService = new OwnershipService(
+		cacheService,
+		userRepository,
+		roleRepository,
+		sharedWorkflowRepository,
+	);
+
+	beforeEach(() => {
+		jest.clearAllMocks();
+	});
+
+	describe('getWorkflowOwner()', () => {
+		test('should retrieve a workflow owner', async () => {
+			roleRepository.findWorkflowOwnerRole.mockResolvedValueOnce(wfOwnerRole());
+
+			const mockOwner = new User();
+			const mockNonOwner = new User();
+
+			const sharedWorkflow = Object.assign(new SharedWorkflow(), {
+				role: new Role(),
+				user: mockOwner,
+			});
+
+			sharedWorkflowRepository.findOneOrFail.mockResolvedValueOnce(sharedWorkflow);
+
+			const returnedOwner = await ownershipService.getWorkflowOwnerCached('some-workflow-id');
+
+			expect(returnedOwner).toBe(mockOwner);
+			expect(returnedOwner).not.toBe(mockNonOwner);
+		});
+
+		test('should throw if no workflow owner role found', async () => {
+			roleRepository.findWorkflowOwnerRole.mockRejectedValueOnce(new Error());
+
+			await expect(ownershipService.getWorkflowOwnerCached('some-workflow-id')).rejects.toThrow();
+		});
+
+		test('should throw if no workflow owner found', async () => {
+			roleRepository.findWorkflowOwnerRole.mockResolvedValueOnce(wfOwnerRole());
+
+			sharedWorkflowRepository.findOneOrFail.mockRejectedValue(new Error());
+
+			await expect(ownershipService.getWorkflowOwnerCached('some-workflow-id')).rejects.toThrow();
+		});
+	});
+});