houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor workflow to use new functions (#17552)

Browse Source 
Removes the versioning system for logic functions (`latestVersion`,
`publishedVersions`) and simplifies the file storage structure.

### Changes
- Remove `publishOneLogicFunctionOrFail` and publishing logic from
workflow status updates
- Add `createLogicFunctionFromExistingLogicFunction` to duplicate logic
functions when creating draft workflow versions
- Update `createDraftStep` to create a new logic function copy instead
of referencing the same one
- Migrate file storage to v2 endpoints with
`applicationUniversalIdentifier`
- Unify path structure: source files at `source/workflow/{id}/`, built
files at `built-logic-function/workflow/{id}/`
- Store full paths in `sourceHandlerPath` and `builtHandlerPath` entity
fields
This commit is contained in:
Charles Bochet
2026-01-30 15:42:36 +01:00
committed by GitHub
parent d624652e36
commit 5996d0fc03
43 changed files with 1213 additions and 1604 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/modules/workflow/workflow-status/jobs/__tests__/workflow-statuses-update.job.spec.ts
-325
View File
@@ -1,325 +0,0 @@
import { Test, type TestingModule } from '@nestjs/testing';
import { getRepositoryToken } from '@nestjs/typeorm';
import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
import { LogicFunctionEntity } from 'src/engine/metadata-modules/logic-function/logic-function.entity';
import { LogicFunctionService } from 'src/engine/metadata-modules/logic-function/logic-function.service';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { WorkflowVersionStatus } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
import { WorkflowStatus } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
import {
WorkflowStatusesUpdateJob,
type WorkflowVersionBatchEvent,
WorkflowVersionEventType,
} from 'src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job';
describe('WorkflowStatusesUpdate', () => {
let job: WorkflowStatusesUpdateJob;
const mockWorkflowRepository = {
findOneOrFail: jest.fn(),
update: jest.fn(),
};
const mockWorkflowVersionRepository = {
findOneOrFail: jest.fn(),
find: jest.fn(),
update: jest.fn(),
};
const mockGlobalWorkspaceOrmManager = {
getRepository: jest
.fn()
.mockImplementation((_workspaceId, entity, options) => {
if (!options?.shouldBypassPermissionChecks) {
throw new Error(
'Permission check will fail because job runners dont have permissions',
);
}
if (entity === 'workflow') {
return Promise.resolve(mockWorkflowRepository);
}
if (entity === 'workflowVersion') {
return Promise.resolve(mockWorkflowVersionRepository);
}
return Promise.resolve(null);
}),
executeInWorkspaceContext: jest
.fn()
.mockImplementation((fn: () => any, _authContext?: any) => fn()),
};
const mockLogicFunctionService = {
publishOneLogicFunctionOrFail: jest.fn(),
findOneOrFail: jest.fn(),
};
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
providers: [
WorkflowStatusesUpdateJob,
{
provide: GlobalWorkspaceOrmManager,
useValue: mockGlobalWorkspaceOrmManager,
},
{
provide: LogicFunctionService,
useValue: mockLogicFunctionService,
},
{
provide: getRepositoryToken(ObjectMetadataEntity),
useValue: {
findOneOrFail: jest.fn().mockResolvedValue({
nameSingular: 'workflow',
}),
},
},
{
provide: getRepositoryToken(LogicFunctionEntity),
useValue: {
findOneOrFail: jest.fn().mockResolvedValue({
id: 'mock-logic-function-id',
}),
},
},
],
}).compile();
job = await module.resolve<WorkflowStatusesUpdateJob>(
WorkflowStatusesUpdateJob,
);
});
it('should be defined', () => {
expect(job).toBeDefined();
});
describe('handle', () => {
describe('when event type is CREATE', () => {
it('when already a draft, do not change anything', async () => {
const event: WorkflowVersionBatchEvent = {
workspaceId: '1',
type: WorkflowVersionEventType.CREATE,
workflowIds: ['1'],
};
const mockWorkflow = {
id: '1',
statuses: [WorkflowStatus.DRAFT],
};
mockWorkflowRepository.findOneOrFail.mockResolvedValue(mockWorkflow);
mockWorkflowVersionRepository.find.mockResolvedValue([
{ status: WorkflowVersionStatus.DRAFT },
]);
await job.handle(event);
expect(mockWorkflowRepository.findOneOrFail).toHaveBeenCalledTimes(1);
expect(mockWorkflowRepository.update).toHaveBeenCalledTimes(0);
});
it('when no draft yet, update statuses', async () => {
const event: WorkflowVersionBatchEvent = {
workspaceId: '1',
type: WorkflowVersionEventType.CREATE,
workflowIds: ['1'],
};
const mockWorkflow = {
id: '1',
statuses: [WorkflowStatus.ACTIVE],
};
mockWorkflowRepository.findOneOrFail.mockResolvedValue(mockWorkflow);
mockWorkflowVersionRepository.find.mockResolvedValue([
{ status: WorkflowVersionStatus.ACTIVE },
{ status: WorkflowVersionStatus.DRAFT },
]);
await job.handle(event);
expect(mockWorkflowRepository.findOneOrFail).toHaveBeenCalledTimes(1);
expect(mockWorkflowRepository.update).toHaveBeenCalledWith(
{ id: '1' },
{ statuses: [WorkflowStatus.DRAFT, WorkflowStatus.ACTIVE] },
);
});
});
describe('when event type is STATUS_UPDATE', () => {
test('when status is the same, should not do anything', async () => {
const event: WorkflowVersionBatchEvent = {
workspaceId: '1',
type: WorkflowVersionEventType.STATUS_UPDATE,
statusUpdates: [
{
workflowId: '1',
workflowVersionId: '1',
previousStatus: WorkflowVersionStatus.ACTIVE,
newStatus: WorkflowVersionStatus.ACTIVE,
},
],
};
const mockWorkflow = {
id: '1',
statuses: [WorkflowStatus.ACTIVE],
};
const mockWorkflowVersion = {
id: '1',
status: WorkflowVersionStatus.ACTIVE,
steps: [],
};
mockWorkflowRepository.findOneOrFail.mockResolvedValue(mockWorkflow);
mockWorkflowVersionRepository.findOneOrFail.mockResolvedValue(
mockWorkflowVersion,
);
mockWorkflowVersionRepository.find.mockResolvedValue([
{ status: WorkflowVersionStatus.ACTIVE },
]);
await job.handle(event);
expect(mockWorkflowRepository.findOneOrFail).toHaveBeenCalledTimes(1);
expect(
mockWorkflowVersionRepository.findOneOrFail,
).toHaveBeenCalledTimes(1);
expect(mockWorkflowRepository.update).toHaveBeenCalledTimes(0);
});
test('when WorkflowVersionStatus.DRAFT to WorkflowVersionStatus.ACTIVE, should activate and publish logic functions', async () => {
const event: WorkflowVersionBatchEvent = {
workspaceId: '1',
type: WorkflowVersionEventType.STATUS_UPDATE,
statusUpdates: [
{
workflowId: '1',
workflowVersionId: '1',
previousStatus: WorkflowVersionStatus.DRAFT,
newStatus: WorkflowVersionStatus.ACTIVE,
},
],
};
const mockWorkflow = {
id: '1',
statuses: [WorkflowStatus.DRAFT],
};
const mockWorkflowVersion = {
id: '1',
status: WorkflowVersionStatus.ACTIVE,
steps: [
{
type: 'CODE',
settings: {
input: {
logicFunctionId: 'logic-function-1',
},
},
},
],
};
const mockLogicFunction = {
id: 'logic-function-1',
};
mockWorkflowRepository.findOneOrFail.mockResolvedValue(mockWorkflow);
mockWorkflowVersionRepository.findOneOrFail.mockResolvedValue(
mockWorkflowVersion,
);
mockWorkflowVersionRepository.find.mockResolvedValue([
{ status: WorkflowVersionStatus.ACTIVE },
]);
mockLogicFunctionService.findOneOrFail.mockResolvedValue(
mockLogicFunction,
);
mockLogicFunctionService.publishOneLogicFunctionOrFail.mockResolvedValue(
mockLogicFunction,
);
await job.handle(event);
expect(mockWorkflowRepository.findOneOrFail).toHaveBeenCalledTimes(1);
expect(
mockWorkflowVersionRepository.findOneOrFail,
).toHaveBeenCalledTimes(1);
expect(
mockLogicFunctionService.publishOneLogicFunctionOrFail,
).toHaveBeenCalledWith('logic-function-1', '1');
expect(mockWorkflowVersionRepository.update).toHaveBeenCalledWith('1', {
steps: [
{
type: 'CODE',
settings: {
input: {
logicFunctionId: 'logic-function-1',
},
},
},
],
});
expect(mockWorkflowRepository.update).toHaveBeenCalledWith(
{ id: '1' },
{ statuses: [WorkflowStatus.ACTIVE] },
);
});
});
describe('when event type is DELETE', () => {
test('when status is not draft, should not do anything', async () => {
const event: WorkflowVersionBatchEvent = {
workspaceId: '1',
type: WorkflowVersionEventType.DELETE,
workflowIds: ['1'],
};
const mockWorkflow = {
id: '1',
statuses: [WorkflowStatus.ACTIVE],
};
mockWorkflowRepository.findOneOrFail.mockResolvedValue(mockWorkflow);
mockWorkflowVersionRepository.find.mockResolvedValue([
{ status: WorkflowVersionStatus.ACTIVE },
]);
await job.handle(event);
expect(mockWorkflowRepository.findOneOrFail).toHaveBeenCalledTimes(1);
expect(mockWorkflowRepository.update).toHaveBeenCalledTimes(0);
});
test('when status is draft, should delete', async () => {
const event: WorkflowVersionBatchEvent = {
workspaceId: '1',
type: WorkflowVersionEventType.DELETE,
workflowIds: ['1'],
};
const mockWorkflow = {
id: '1',
statuses: [WorkflowStatus.DRAFT],
};
mockWorkflowRepository.findOneOrFail.mockResolvedValue(mockWorkflow);
mockWorkflowVersionRepository.find.mockResolvedValue([]);
await job.handle(event);
expect(mockWorkflowRepository.findOneOrFail).toHaveBeenCalledTimes(1);
expect(mockWorkflowRepository.update).toHaveBeenCalledWith(
{ id: '1' },
{ statuses: [] },
);
});
});
});
});
packages/twenty-server/src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job.ts
-58
View File
@@ -1,13 +1,11 @@
import { Logger, Scope } from '@nestjs/common';
import isEqual from 'lodash.isequal';
import { isDefined } from 'twenty-shared/utils';
import { In } from 'typeorm';
import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { LogicFunctionService } from 'src/engine/metadata-modules/logic-function/logic-function.service';
import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager';
import { type WorkspaceRepository } from 'src/engine/twenty-orm/repository/workspace.repository';
import { buildSystemAuthContext } from 'src/engine/twenty-orm/utils/build-system-auth-context.util';
@@ -19,10 +17,6 @@ import {
WorkflowStatus,
type WorkflowWorkspaceEntity,
} from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
import {
type WorkflowAction,
WorkflowActionType,
} from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
export enum WorkflowVersionEventType {
CREATE = 'CREATE',
@@ -67,7 +61,6 @@ export class WorkflowStatusesUpdateJob {
constructor(
private readonly globalWorkspaceOrmManager: GlobalWorkspaceOrmManager,
private readonly logicFunctionService: LogicFunctionService,
) {}
@Process(WorkflowStatusesUpdateJob.name)
@@ -150,46 +143,6 @@ export class WorkflowStatusesUpdateJob {
);
}
private async handlePublishLogicFunction({
statusUpdate,
workspaceId,
workflowVersion,
workflowVersionRepository,
}: {
statusUpdate: WorkflowVersionStatusUpdate;
workspaceId: string;
workflowVersion: WorkflowVersionWorkspaceEntity;
workflowVersionRepository: WorkspaceRepository<WorkflowVersionWorkspaceEntity>;
}) {
const shouldComputeNewSteps =
statusUpdate.newStatus === WorkflowVersionStatus.ACTIVE &&
isDefined(workflowVersion.steps) &&
workflowVersion.steps.filter(
(step) => step.type === WorkflowActionType.CODE,
).length > 0;
if (shouldComputeNewSteps) {
const newSteps: WorkflowAction[] = [];
for (const step of workflowVersion.steps || []) {
const newStep = { ...step };
if (step.type === WorkflowActionType.CODE) {
await this.logicFunctionService.publishOneLogicFunctionOrFail(
step.settings.input.logicFunctionId,
workspaceId,
);
}
newSteps.push(newStep);
}
await workflowVersionRepository.update(statusUpdate.workflowVersionId, {
steps: newSteps,
});
}
}
private async handleWorkflowVersionStatusUpdated({
statusUpdate,
workspaceId,
@@ -217,17 +170,6 @@ export class WorkflowStatusesUpdateJob {
},
});
const workflowVersion = await workflowVersionRepository.findOneOrFail({
where: { id: statusUpdate.workflowVersionId },
});
await this.handlePublishLogicFunction({
workflowVersion,
workflowVersionRepository,
workspaceId,
statusUpdate,
});
const newWorkflowStatuses = await this.getWorkflowStatuses({
workflowId: statusUpdate.workflowId,
workflowVersionRepository,