refactor(core): Inject dependencies into workflow services (no-changelog) (#8066)

Inject dependencies into workflow services (no-changelog)

Up next:

- ~~Make workflow services injectable~~ #8033
- ~~Inject dependencies into workflow services~~ (current)
- Consolidate workflow controllers into one
- Make workflow controller injectable
- Inject dependencies into workflow controller
This commit is contained in:
Iván Ovejero 2023-12-18 16:10:30 +01:00 committed by GitHub
parent a651089a10
commit 73d400a1bf
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 127 additions and 80 deletions

View file

@ -6,6 +6,7 @@ import type { IExecutionResponse, IExecutionFlattedResponse } from '@/Interfaces
import { EnterpriseWorkflowService } from '../workflows/workflow.service.ee';
import type { WorkflowWithSharingsAndCredentials } from '@/workflows/workflows.types';
import Container from 'typedi';
import { WorkflowService } from '@/workflows/workflow.service';
export class EEExecutionsService extends ExecutionsService {
/**
@ -24,14 +25,15 @@ export class EEExecutionsService extends ExecutionsService {
if (!execution) return;
const relations = ['shared', 'shared.user', 'shared.role'];
const enterpriseWorkflowService = Container.get(EnterpriseWorkflowService);
const workflow = (await enterpriseWorkflowService.get(
const workflow = (await Container.get(WorkflowService).get(
{ id: execution.workflowId },
{ relations },
)) as WorkflowWithSharingsAndCredentials;
if (!workflow) return;
const enterpriseWorkflowService = Container.get(EnterpriseWorkflowService);
enterpriseWorkflowService.addOwnerAndSharings(workflow);
await enterpriseWorkflowService.addCredentialsToWorkflow(workflow, req.user);

View file

@ -13,22 +13,31 @@ import type {
import { CredentialsService } from '@/credentials/credentials.service';
import { ApplicationError, NodeOperationError } from 'n8n-workflow';
import { RoleService } from '@/services/role.service';
import Container, { Service } from 'typedi';
import { Service } from 'typedi';
import type { CredentialsEntity } from '@db/entities/CredentialsEntity';
import { SharedWorkflowRepository } from '@db/repositories/sharedWorkflow.repository';
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
import { NotFoundError } from '@/errors/response-errors/not-found.error';
@Service()
export class EnterpriseWorkflowService extends WorkflowService {
export class EnterpriseWorkflowService {
constructor(
private readonly workflowService: WorkflowService,
private readonly userService: UserService,
private readonly roleService: RoleService,
private readonly sharedWorkflowRepository: SharedWorkflowRepository,
) {}
async isOwned(
user: User,
workflowId: string,
): Promise<{ ownsWorkflow: boolean; workflow?: WorkflowEntity }> {
const sharing = await this.getSharing(user, workflowId, { allowGlobalScope: false }, [
'workflow',
'role',
]);
const sharing = await this.workflowService.getSharing(
user,
workflowId,
{ allowGlobalScope: false },
['workflow', 'role'],
);
if (!sharing || sharing.role.name !== 'owner') return { ownsWorkflow: false };
@ -65,8 +74,8 @@ export class EnterpriseWorkflowService extends WorkflowService {
workflow: WorkflowEntity,
shareWithIds: string[],
): Promise<SharedWorkflow[]> {
const users = await Container.get(UserService).getByIds(transaction, shareWithIds);
const role = await Container.get(RoleService).findWorkflowEditorRole();
const users = await this.userService.getByIds(transaction, shareWithIds);
const role = await this.roleService.findWorkflowEditorRole();
const newSharedWorkflows = users.reduce<SharedWorkflow[]>((acc, user) => {
if (user.isPending) {
@ -77,7 +86,7 @@ export class EnterpriseWorkflowService extends WorkflowService {
userId: user.id,
roleId: role?.id,
};
acc.push(Container.get(SharedWorkflowRepository).create(entity));
acc.push(this.sharedWorkflowRepository.create(entity));
return acc;
}, []);
@ -173,7 +182,7 @@ export class EnterpriseWorkflowService extends WorkflowService {
}
async preventTampering(workflow: WorkflowEntity, workflowId: string, user: User) {
const previousVersion = await this.get({ id: workflowId });
const previousVersion = await this.workflowService.get({ id: workflowId });
if (!previousVersion) {
throw new NotFoundError('Workflow not found');

View file

@ -1,4 +1,4 @@
import { Container, Service } from 'typedi';
import Container, { Service } from 'typedi';
import type { IDataObject, INode, IPinData } from 'n8n-workflow';
import { NodeApiError, ErrorReporterProxy as ErrorReporter, Workflow } from 'n8n-workflow';
import type { FindManyOptions, FindOptionsSelect, FindOptionsWhere, UpdateResult } from 'typeorm';
@ -43,6 +43,23 @@ export type WorkflowsGetSharedOptions =
@Service()
export class WorkflowService {
constructor(
private readonly logger: Logger,
private readonly executionRepository: ExecutionRepository,
private readonly sharedWorkflowRepository: SharedWorkflowRepository,
private readonly workflowRepository: WorkflowRepository,
private readonly workflowTagMappingRepository: WorkflowTagMappingRepository,
private readonly binaryDataService: BinaryDataService,
private readonly ownershipService: OwnershipService,
private readonly tagService: TagService,
private readonly workflowHistoryService: WorkflowHistoryService,
private readonly multiMainSetup: MultiMainSetup,
private readonly nodeTypes: NodeTypes,
private readonly testWebhooks: TestWebhooks,
private readonly externalHooks: ExternalHooks,
private readonly activeWorkflowRunner: ActiveWorkflowRunner,
) {}
async getSharing(
user: User,
workflowId: string,
@ -58,7 +75,7 @@ export class WorkflowService {
where.userId = user.id;
}
return Container.get(SharedWorkflowRepository).findOne({ where, relations });
return this.sharedWorkflowRepository.findOne({ where, relations });
}
/**
@ -89,7 +106,7 @@ export class WorkflowService {
nodes: workflow.nodes,
connections: workflow.connections,
active: workflow.active,
nodeTypes: Container.get(NodeTypes),
nodeTypes: this.nodeTypes,
}).getParentNodes(startNodeName);
let checkNodeName = '';
@ -104,7 +121,7 @@ export class WorkflowService {
}
async get(workflow: FindOptionsWhere<WorkflowEntity>, options?: { relations: string[] }) {
return Container.get(WorkflowRepository).findOne({
return this.workflowRepository.findOne({
where: workflow,
relations: options?.relations,
});
@ -175,15 +192,14 @@ export class WorkflowService {
findManyOptions.take = options.take;
}
const [workflows, count] = (await Container.get(WorkflowRepository).findAndCount(
findManyOptions,
)) as [ListQuery.Workflow.Plain[] | ListQuery.Workflow.WithSharing[], number];
const [workflows, count] = (await this.workflowRepository.findAndCount(findManyOptions)) as [
ListQuery.Workflow.Plain[] | ListQuery.Workflow.WithSharing[],
number,
];
return hasSharing(workflows)
? {
workflows: workflows.map((w) =>
Container.get(OwnershipService).addOwnedByAndSharedWith(w),
),
workflows: workflows.map((w) => this.ownershipService.addOwnedByAndSharedWith(w)),
count,
}
: { workflows, count };
@ -197,7 +213,7 @@ export class WorkflowService {
forceSave?: boolean,
roles?: string[],
): Promise<WorkflowEntity> {
const shared = await Container.get(SharedWorkflowRepository).findOne({
const shared = await this.sharedWorkflowRepository.findOne({
relations: ['workflow', 'role'],
where: await whereClause({
user,
@ -208,9 +224,8 @@ export class WorkflowService {
}),
});
const logger = Container.get(Logger);
if (!shared) {
logger.verbose('User attempted to update a workflow without permissions', {
this.logger.verbose('User attempted to update a workflow without permissions', {
workflowId,
userId: user.id,
});
@ -236,7 +251,7 @@ export class WorkflowService {
// Update the workflow's version when changing properties such as
// `name`, `pinData`, `nodes`, `connections`, `settings` or `tags`
workflow.versionId = uuid();
logger.verbose(
this.logger.verbose(
`Updating versionId for workflow ${workflowId} for user ${user.id} after saving`,
{
previousVersionId: shared.workflow.versionId,
@ -250,7 +265,7 @@ export class WorkflowService {
WorkflowHelpers.addNodeIds(workflow);
await Container.get(ExternalHooks).run('workflow.update', [workflow]);
await this.externalHooks.run('workflow.update', [workflow]);
/**
* If the workflow being updated is stored as `active`, remove it from
@ -260,7 +275,7 @@ export class WorkflowService {
* will take effect only on removing and re-adding.
*/
if (shared.workflow.active) {
await Container.get(ActiveWorkflowRunner).remove(workflowId);
await this.activeWorkflowRunner.remove(workflowId);
}
const workflowSettings = workflow.settings ?? {};
@ -289,7 +304,7 @@ export class WorkflowService {
await validateEntity(workflow);
}
await Container.get(WorkflowRepository).update(
await this.workflowRepository.update(
workflowId,
pick(workflow, [
'name',
@ -304,21 +319,21 @@ export class WorkflowService {
);
if (tagIds && !config.getEnv('workflowTagsDisabled')) {
await Container.get(WorkflowTagMappingRepository).delete({ workflowId });
await Container.get(WorkflowTagMappingRepository).insert(
await this.workflowTagMappingRepository.delete({ workflowId });
await this.workflowTagMappingRepository.insert(
tagIds.map((tagId) => ({ tagId, workflowId })),
);
}
if (workflow.versionId !== shared.workflow.versionId) {
await Container.get(WorkflowHistoryService).saveVersion(user, workflow, workflowId);
await this.workflowHistoryService.saveVersion(user, workflow, workflowId);
}
const relations = config.getEnv('workflowTagsDisabled') ? [] : ['tags'];
// We sadly get nothing back from "update". Neither if it updated a record
// nor the new value. So query now the hopefully updated entry.
const updatedWorkflow = await Container.get(WorkflowRepository).findOne({
const updatedWorkflow = await this.workflowRepository.findOne({
where: { id: workflowId },
relations,
});
@ -330,26 +345,26 @@ export class WorkflowService {
}
if (updatedWorkflow.tags?.length && tagIds?.length) {
updatedWorkflow.tags = Container.get(TagService).sortByRequestOrder(updatedWorkflow.tags, {
updatedWorkflow.tags = this.tagService.sortByRequestOrder(updatedWorkflow.tags, {
requestOrder: tagIds,
});
}
await Container.get(ExternalHooks).run('workflow.afterUpdate', [updatedWorkflow]);
await this.externalHooks.run('workflow.afterUpdate', [updatedWorkflow]);
void Container.get(InternalHooks).onWorkflowSaved(user, updatedWorkflow, false);
if (updatedWorkflow.active) {
// When the workflow is supposed to be active add it again
try {
await Container.get(ExternalHooks).run('workflow.activate', [updatedWorkflow]);
await Container.get(ActiveWorkflowRunner).add(
await this.externalHooks.run('workflow.activate', [updatedWorkflow]);
await this.activeWorkflowRunner.add(
workflowId,
shared.workflow.active ? 'update' : 'activate',
);
} catch (error) {
// If workflow could not be activated set it again to inactive
// and revert the versionId change so UI remains consistent
await Container.get(WorkflowRepository).update(workflowId, {
await this.workflowRepository.update(workflowId, {
active: false,
versionId: shared.workflow.versionId,
});
@ -366,12 +381,10 @@ export class WorkflowService {
}
}
const multiMainSetup = Container.get(MultiMainSetup);
await this.multiMainSetup.init();
await multiMainSetup.init();
if (multiMainSetup.isEnabled) {
await Container.get(MultiMainSetup).broadcastWorkflowActiveStateChanged({
if (this.multiMainSetup.isEnabled) {
await this.multiMainSetup.broadcastWorkflowActiveStateChanged({
workflowId,
oldState,
newState: updatedWorkflow.active,
@ -412,14 +425,14 @@ export class WorkflowService {
nodes: workflowData.nodes,
connections: workflowData.connections,
active: false,
nodeTypes: Container.get(NodeTypes),
nodeTypes: this.nodeTypes,
staticData: undefined,
settings: workflowData.settings,
});
const additionalData = await WorkflowExecuteAdditionalData.getBase(user.id);
const needsWebhook = await Container.get(TestWebhooks).needsWebhookData(
const needsWebhook = await this.testWebhooks.needsWebhookData(
workflowData,
workflow,
additionalData,
@ -465,9 +478,9 @@ export class WorkflowService {
}
async delete(user: User, workflowId: string): Promise<WorkflowEntity | undefined> {
await Container.get(ExternalHooks).run('workflow.delete', [workflowId]);
await this.externalHooks.run('workflow.delete', [workflowId]);
const sharedWorkflow = await Container.get(SharedWorkflowRepository).findOne({
const sharedWorkflow = await this.sharedWorkflowRepository.findOne({
relations: ['workflow', 'role'],
where: await whereClause({
user,
@ -484,27 +497,27 @@ export class WorkflowService {
if (sharedWorkflow.workflow.active) {
// deactivate before deleting
await Container.get(ActiveWorkflowRunner).remove(workflowId);
await this.activeWorkflowRunner.remove(workflowId);
}
const idsForDeletion = await Container.get(ExecutionRepository)
const idsForDeletion = await this.executionRepository
.find({
select: ['id'],
where: { workflowId },
})
.then((rows) => rows.map(({ id: executionId }) => ({ workflowId, executionId })));
await Container.get(WorkflowRepository).delete(workflowId);
await Container.get(BinaryDataService).deleteMany(idsForDeletion);
await this.workflowRepository.delete(workflowId);
await this.binaryDataService.deleteMany(idsForDeletion);
void Container.get(InternalHooks).onWorkflowDeleted(user, workflowId, false);
await Container.get(ExternalHooks).run('workflow.afterDelete', [workflowId]);
await this.externalHooks.run('workflow.afterDelete', [workflowId]);
return sharedWorkflow.workflow;
}
async updateWorkflowTriggerCount(id: string, triggerCount: number): Promise<UpdateResult> {
const qb = Container.get(WorkflowRepository).createQueryBuilder('workflow');
const qb = this.workflowRepository.createQueryBuilder('workflow');
return qb
.update()
.set({
@ -533,7 +546,7 @@ export class WorkflowService {
workflow.staticData.__dataChanged = false;
} catch (error) {
ErrorReporter.error(error);
Container.get(Logger).error(
this.logger.error(
// eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
`There was a problem saving the workflow with id "${workflow.id}" to save changed Data: "${error.message}"`,
{ workflowId: workflow.id },
@ -550,7 +563,7 @@ export class WorkflowService {
* @param {IDataObject} newStaticData The static data to save
*/
async saveStaticDataById(workflowId: string, newStaticData: IDataObject): Promise<void> {
await Container.get(WorkflowRepository).update(workflowId, {
await this.workflowRepository.update(workflowId, {
staticData: newStaticData,
});
}

View file

@ -27,6 +27,7 @@ import { UnauthorizedError } from '@/errors/response-errors/unauthorized.error';
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
import { NotFoundError } from '@/errors/response-errors/not-found.error';
import { InternalServerError } from '@/errors/response-errors/internal-server.error';
import { WorkflowService } from './workflow.service';
export const EEWorkflowController = express.Router();
@ -67,14 +68,10 @@ EEWorkflowController.put(
workflow = undefined;
// Allow owners/admins to share
if (await req.user.hasGlobalScope('workflow:share')) {
const sharedRes = await Container.get(EnterpriseWorkflowService).getSharing(
req.user,
workflowId,
{
allowGlobalScope: true,
globalScope: 'workflow:share',
},
);
const sharedRes = await Container.get(WorkflowService).getSharing(req.user, workflowId, {
allowGlobalScope: true,
globalScope: 'workflow:share',
});
workflow = sharedRes?.workflow;
}
if (!workflow) {
@ -132,9 +129,7 @@ EEWorkflowController.get(
relations.push('tags');
}
const enterpriseWorkflowService = Container.get(EnterpriseWorkflowService);
const workflow = await enterpriseWorkflowService.get({ id: workflowId }, { relations });
const workflow = await Container.get(WorkflowService).get({ id: workflowId }, { relations });
if (!workflow) {
throw new NotFoundError(`Workflow with ID "${workflowId}" does not exist`);
@ -147,6 +142,8 @@ EEWorkflowController.get(
);
}
const enterpriseWorkflowService = Container.get(EnterpriseWorkflowService);
enterpriseWorkflowService.addOwnerAndSharings(workflow);
await enterpriseWorkflowService.addCredentialsToWorkflow(workflow, req.user);
return workflow;
@ -253,7 +250,7 @@ EEWorkflowController.get(
try {
const sharedWorkflowIds = await WorkflowHelpers.getSharedWorkflowIds(req.user);
const { workflows: data, count } = await Container.get(EnterpriseWorkflowService).getMany(
const { workflows: data, count } = await Container.get(WorkflowService).getMany(
sharedWorkflowIds,
req.listQueryOptions,
);
@ -283,7 +280,7 @@ EEWorkflowController.patch(
req.user,
);
const updatedWorkflow = await Container.get(EnterpriseWorkflowService).update(
const updatedWorkflow = await Container.get(WorkflowService).update(
req.user,
safeWorkflow,
workflowId,
@ -313,7 +310,7 @@ EEWorkflowController.post(
req.body.workflowData.nodes = safeWorkflow.nodes;
}
return Container.get(EnterpriseWorkflowService).runManually(
return Container.get(WorkflowService).runManually(
req.body,
req.user,
GenericHelpers.getSessionId(req),

View file

@ -16,6 +16,8 @@ import { getAllRoles } from '../shared/db/roles';
import { createUser } from '../shared/db/users';
import { createWorkflow, createWorkflowWithTrigger } from '../shared/db/workflows';
import { createTag } from '../shared/db/tags';
import { mockInstance } from '../../shared/mocking';
import { Push } from '@/push';
let workflowOwnerRole: Role;
let owner: User;
@ -27,6 +29,8 @@ let workflowRunner: ActiveWorkflowRunner;
const testServer = utils.setupTestServer({ endpointGroups: ['publicApi'] });
const license = testServer.license;
mockInstance(Push);
beforeAll(async () => {
const [globalOwnerRole, globalMemberRole, fetchedWorkflowOwnerRole] = await getAllRoles();

View file

@ -3,16 +3,38 @@ import { ActiveWorkflowRunner } from '@/ActiveWorkflowRunner';
import * as testDb from './shared/testDb';
import { WorkflowService } from '@/workflows/workflow.service';
import { mockInstance } from '../shared/mocking';
import { Telemetry } from '@/telemetry';
import { createOwner } from './shared/db/users';
import { createWorkflow } from './shared/db/workflows';
import { SharedWorkflowRepository } from '@/databases/repositories/sharedWorkflow.repository';
import { mock } from 'jest-mock-extended';
import { WorkflowRepository } from '@/databases/repositories/workflow.repository';
import { Telemetry } from '@/telemetry';
mockInstance(Telemetry);
const activeWorkflowRunner = mockInstance(ActiveWorkflowRunner);
let workflowService: WorkflowService;
let activeWorkflowRunner: ActiveWorkflowRunner;
beforeAll(async () => {
await testDb.init();
activeWorkflowRunner = mockInstance(ActiveWorkflowRunner);
mockInstance(Telemetry);
workflowService = new WorkflowService(
mock(),
mock(),
Container.get(SharedWorkflowRepository),
Container.get(WorkflowRepository),
mock(),
mock(),
mock(),
mock(),
mock(),
mock(),
mock(),
mock(),
mock(),
activeWorkflowRunner,
);
});
afterEach(async () => {
@ -32,7 +54,7 @@ describe('update()', () => {
const removeSpy = jest.spyOn(activeWorkflowRunner, 'remove');
const addSpy = jest.spyOn(activeWorkflowRunner, 'add');
await Container.get(WorkflowService).update(owner, workflow, workflow.id);
await workflowService.update(owner, workflow, workflow.id);
expect(removeSpy).toHaveBeenCalledTimes(1);
const [removedWorkflowId] = removeSpy.mock.calls[0];
@ -52,7 +74,7 @@ describe('update()', () => {
const addSpy = jest.spyOn(activeWorkflowRunner, 'add');
workflow.active = false;
await Container.get(WorkflowService).update(owner, workflow, workflow.id);
await workflowService.update(owner, workflow, workflow.id);
expect(removeSpy).toHaveBeenCalledTimes(1);
const [removedWorkflowId] = removeSpy.mock.calls[0];

View file

@ -20,8 +20,7 @@ import { getCredentialOwnerRole, getGlobalMemberRole, getGlobalOwnerRole } from
import { createUser } from './shared/db/users';
import { createWorkflow, getWorkflowSharing, shareWorkflowWithUsers } from './shared/db/workflows';
import type { Role } from '@/databases/entities/Role';
import { EnterpriseWorkflowService } from '@/workflows/workflow.service.ee';
import { WorkflowService } from '@/workflows/workflow.service';
import { Push } from '@/push';
let globalMemberRole: Role;
let owner: User;
@ -33,6 +32,7 @@ let authAnotherMemberAgent: SuperAgentTest;
let saveCredential: SaveCredentialFunction;
const activeWorkflowRunnerLike = mockInstance(ActiveWorkflowRunner);
mockInstance(Push);
const sharingSpy = jest.spyOn(UserManagementHelpers, 'isSharingEnabled').mockReturnValue(true);
const testServer = utils.setupTestServer({
@ -57,9 +57,6 @@ beforeAll(async () => {
saveCredential = affixRoleToSaveCredential(credentialOwnerRole);
await utils.initNodeTypes();
Container.set(WorkflowService, new WorkflowService());
Container.set(EnterpriseWorkflowService, new EnterpriseWorkflowService());
});
beforeEach(async () => {

View file

@ -18,17 +18,20 @@ import { saveCredential } from './shared/db/credentials';
import { createOwner } from './shared/db/users';
import { createWorkflow } from './shared/db/workflows';
import { createTag } from './shared/db/tags';
import { Push } from '@/push';
let owner: User;
let authOwnerAgent: SuperAgentTest;
jest.spyOn(UserManagementHelpers, 'isSharingEnabled').mockReturnValue(false);
const testServer = utils.setupTestServer({ endpointGroups: ['workflows'] });
const license = testServer.license;
const { objectContaining, arrayContaining, any } = expect;
const activeWorkflowRunnerLike = mockInstance(ActiveWorkflowRunner);
mockInstance(Push);
beforeAll(async () => {
owner = await createOwner();