From 850a0fdea2d45bdd6908e2ab6eb6585b6f7e7063 Mon Sep 17 00:00:00 2001 From: martmull Date: Wed, 27 Aug 2025 12:01:33 +0200 Subject: [PATCH] Add command to update workflow version steps positions (#14098) Reopening https://github.com/twentyhq/twenty/pull/13721 after some updates --- .../components/WorkflowDiagramCanvasBase.tsx | 1 + ...flow-versions-and-workflow-runs.command.ts | 336 ++++++++++++++++++ .../1-5/1-5-upgrade-version-command.module.ts | 11 +- .../upgrade.command.ts | 7 +- 4 files changed, 352 insertions(+), 3 deletions(-) create mode 100644 packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-add-positions-to-workflow-versions-and-workflow-runs.command.ts diff --git a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowDiagramCanvasBase.tsx b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowDiagramCanvasBase.tsx index 07ed27be44..3474906f3f 100644 --- a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowDiagramCanvasBase.tsx +++ b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowDiagramCanvasBase.tsx @@ -193,6 +193,7 @@ export const WorkflowDiagramCanvasBase = ({ if (isWorkflowBranchEnabled) { return workflowDiagram; } + return getOrganizedDiagram(workflowDiagram); } return { nodes: [], edges: [] }; diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-add-positions-to-workflow-versions-and-workflow-runs.command.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-add-positions-to-workflow-versions-and-workflow-runs.command.ts new file mode 100644 index 0000000000..a8902c7374 --- /dev/null +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-add-positions-to-workflow-versions-and-workflow-runs.command.ts @@ -0,0 +1,336 @@ +import { InjectRepository } from '@nestjs/typeorm'; + +import { Command, Option } from 'nest-commander'; +import { Repository } from 'typeorm'; +import { isDefined } from 'twenty-shared/utils'; +import { v4 } from 'uuid'; +import Dagre from '@dagrejs/dagre'; + +import { + ActiveOrSuspendedWorkspacesMigrationCommandOptions, + ActiveOrSuspendedWorkspacesMigrationCommandRunner, + type RunOnWorkspaceArgs, +} from 'src/database/commands/command-runners/active-or-suspended-workspaces-migration.command-runner'; +import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; +import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; +import { WorkspaceDataSourceService } from 'src/engine/workspace-datasource/workspace-datasource.service'; +import { getWorkspaceSchemaName } from 'src/engine/workspace-datasource/utils/get-workspace-schema-name.util'; +import { type WorkflowTrigger } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type'; +import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type'; +import { type WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity'; + +type Node = { + id: string; + position: { x: number; y: number }; + size: number; + measured?: { width: number; height: number }; +}; + +type Edge = { id: string; source: string; target: string }; + +type Diagram = { + nodes: Node[]; + edges: Edge[]; +}; + +export type AddPositionsToWorkflowVersionsAndWorkflowRunsOptions = + ActiveOrSuspendedWorkspacesMigrationCommandOptions & { + processWorkflowRuns?: boolean; + }; + +@Command({ + name: 'upgrade:1-5:add-positions-to-workflow-versions-and-workflow-runs', + description: 'Add positions to workflow versions and workflow runs', +}) +export class AddPositionsToWorkflowVersionsAndWorkflowRuns extends ActiveOrSuspendedWorkspacesMigrationCommandRunner { + constructor( + @InjectRepository(Workspace, 'core') + protected readonly workspaceRepository: Repository, + private readonly workspaceDataSourceService: WorkspaceDataSourceService, + protected readonly twentyORMGlobalManager: TwentyORMGlobalManager, + ) { + super(workspaceRepository, twentyORMGlobalManager); + } + + @Option({ + flags: '--process-workflow-runs [process_workflow_runs]', + description: 'Process workflowRuns positions (default false)', + required: false, + }) + parseProcessWorkflowRuns(): boolean { + return true; + } + + override async runOnWorkspace({ + workspaceId, + options, + }: Omit & { + options: AddPositionsToWorkflowVersionsAndWorkflowRunsOptions; + }): Promise { + await this.addPositionsToWorkflowVersions({ workspaceId }); + + if (options.processWorkflowRuns) { + await this.addPositionsToWorkflowRuns({ workspaceId }); + } + } + + private async addPositionsToWorkflowVersions({ + workspaceId, + }: { + workspaceId: string; + }) { + const workflowVersionRepository = + await this.twentyORMGlobalManager.getRepositoryForWorkspace( + workspaceId, + 'workflowVersion', + { shouldBypassPermissionChecks: true }, + ); + + const workflowVersions = await workflowVersionRepository.find(); + + let count = 0; + + for (const workflowVersion of workflowVersions) { + try { + // We only update one branch workflow + if (this.isWorkflowMultiBranch(workflowVersion)) { + continue; + } + + const { updatedTrigger, updatedSteps } = + this.getOrganizedStepsAndTrigger({ + trigger: workflowVersion.trigger, + steps: workflowVersion.steps, + }); + + await workflowVersionRepository.update(workflowVersion.id, { + trigger: updatedTrigger, + steps: updatedSteps, + }); + count += 1; + } catch (error) { + this.logger.error( + `Error while adding positions to workflowVersion '${workflowVersion.id}'`, + error, + ); + } + } + + this.logger.log( + `Workflow versions updated count: ${count} out of ${workflowVersions.length}`, + ); + } + + private isWorkflowMultiBranch( + workflowVersion: WorkflowVersionWorkspaceEntity, + ) { + if ((workflowVersion.trigger?.nextStepIds ?? []).length > 1) { + return true; + } + + for (const step of workflowVersion.steps || []) { + if ((step.nextStepIds ?? []).length > 1) { + return true; + } + } + + return false; + } + + private async addPositionsToWorkflowRuns({ + workspaceId, + }: { + workspaceId: string; + }) { + const mainDataSource = + await this.workspaceDataSourceService.connectToMainDataSource(); + + const schemaName = getWorkspaceSchemaName(workspaceId); + + const workflowRuns = await mainDataSource.query( + `SELECT id, state FROM ${schemaName}."workflowRun"`, + ); + + for (const workflowRun of workflowRuns) { + try { + const { updatedTrigger, updatedSteps } = + this.getOrganizedStepsAndTrigger({ + trigger: workflowRun.state.flow.trigger, + steps: workflowRun.state.flow.steps, + }); + + const updatedState = { + ...workflowRun.state, + flow: { + ...workflowRun.state.flow, + trigger: updatedTrigger, + steps: updatedSteps, + }, + }; + + await mainDataSource.query( + `UPDATE ${schemaName}."workflowRun" SET state = $1::jsonb WHERE id = $2`, + [updatedState, workflowRun.id], + ); + } catch (error) { + this.logger.error( + `Error while adding positions to workflowRuns '${workflowRun.id}'`, + error, + ); + } + } + } + + private getNodePositionFromDiagram({ + stepId, + diagram, + }: { + stepId: string; + diagram: Diagram; + }) { + return diagram.nodes.find((node) => node.id === stepId)?.position; + } + + private createWorkflowDiagram({ + trigger, + steps, + }: { + trigger?: WorkflowTrigger | null; + steps: WorkflowAction[] | null; + }): Diagram { + const nodes: Node[] = []; + + const edges: Edge[] = []; + + if (!isDefined(trigger)) { + const triggerNextStepIds = isDefined(steps) + ? this.getRootSteps(steps).map((step) => step.id) + : []; + + triggerNextStepIds.forEach((stepId: string) => { + edges.push({ + id: v4(), + source: 'trigger', + target: stepId, + }); + }); + nodes.push({ id: 'trigger', size: 13, position: { x: 0, y: 0 } }); + } else { + nodes.push({ + id: 'trigger', + size: Math.min(trigger.name.length, 29), + position: { x: 0, y: 0 }, + }); + } + + for (const step of steps || []) { + nodes.push({ + id: step.id, + size: Math.min(step.name.length, 29), + position: { x: 0, y: 0 }, + }); + + step.nextStepIds?.forEach((nextStepId) => { + edges.push({ + id: v4(), + source: step.id, + target: nextStepId, + }); + }); + } + + for (const stepId of trigger?.nextStepIds || []) { + edges.push({ + id: v4(), + source: 'trigger', + target: stepId, + }); + } + + return { + nodes, + edges, + }; + } + + private getOrganizedStepsAndTrigger({ + trigger, + steps, + }: { + trigger?: WorkflowTrigger | null; + steps: WorkflowAction[] | null; + }) { + const workflowDiagram = this.createWorkflowDiagram({ steps, trigger }); + + const organizedDiagram = this.getOrganizedDiagram(workflowDiagram); + + const updatedTrigger: WorkflowTrigger | undefined = isDefined(trigger) + ? { + ...trigger, + position: this.getNodePositionFromDiagram({ + stepId: 'trigger', + diagram: organizedDiagram, + }), + } + : undefined; + + const updatedSteps: WorkflowAction[] = (steps || []).map( + (step: WorkflowAction) => ({ + ...step, + position: this.getNodePositionFromDiagram({ + stepId: step.id, + diagram: organizedDiagram, + }), + }), + ); + + return { + updatedTrigger, + updatedSteps, + }; + } + + private getRootSteps(steps: WorkflowAction[]): WorkflowAction[] { + const childIds = new Set(); + + for (const step of steps) { + step.nextStepIds?.forEach((id) => childIds.add(id)); + } + + return steps.filter((step) => !childIds.has(step.id)); + } + + private getOrganizedDiagram(diagram: Diagram): Diagram { + const graph = new Dagre.graphlib.Graph().setDefaultEdgeLabel(() => ({})); + + graph.setGraph({ + ranksep: 80, // Vertical distance between 2 nodes + nodesep: 200, // Horizontal distance between 2 nodes + rankdir: 'TB', + }); + + diagram.edges.forEach((edge) => graph.setEdge(edge.source, edge.target)); + diagram.nodes.forEach((node) => + graph.setNode(node.id, { + width: node.size * 6, + height: 50, + }), + ); + + Dagre.layout(graph); + + return { + nodes: diagram.nodes.map((node) => { + const position = graph.node(node.id); + + // We are shifting the dagre node position (anchor=center center) to the top left + // so it matches the React Flow node anchor point (top left). + const x = position.x - position.width / 2; + const y = position.y - position.height / 2; + + return { ...node, position: { x, y } }; + }), + edges: diagram.edges, + }; + } +} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-upgrade-version-command.module.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-upgrade-version-command.module.ts index 0cf2fe865c..149a75f0be 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-upgrade-version-command.module.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/1-5/1-5-upgrade-version-command.module.ts @@ -8,6 +8,7 @@ import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadat import { WorkspaceMetadataVersionModule } from 'src/engine/metadata-modules/workspace-metadata-version/workspace-metadata-version.module'; import { WorkspaceSchemaManagerModule } from 'src/engine/twenty-orm/workspace-schema-manager/workspace-schema-manager.module'; import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/workspace-datasource.module'; +import { AddPositionsToWorkflowVersionsAndWorkflowRuns } from 'src/database/commands/upgrade-version-command/1-5/1-5-add-positions-to-workflow-versions-and-workflow-runs.command'; @Module({ imports: [ @@ -19,7 +20,13 @@ import { WorkspaceDataSourceModule } from 'src/engine/workspace-datasource/works WorkspaceSchemaManagerModule, WorkspaceMetadataVersionModule, ], - providers: [RemoveFavoriteViewRelation], - exports: [RemoveFavoriteViewRelation], + providers: [ + RemoveFavoriteViewRelation, + AddPositionsToWorkflowVersionsAndWorkflowRuns, + ], + exports: [ + RemoveFavoriteViewRelation, + AddPositionsToWorkflowVersionsAndWorkflowRuns, + ], }) export class V1_5_UpgradeVersionCommandModule {} diff --git a/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts b/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts index cdaec72b56..46e3838501 100644 --- a/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts +++ b/packages/twenty-server/src/database/commands/upgrade-version-command/upgrade.command.ts @@ -37,6 +37,7 @@ import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity'; import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager'; import { SyncWorkspaceMetadataCommand } from 'src/engine/workspace-manager/workspace-sync-metadata/commands/sync-workspace-metadata.command'; import { compareVersionMajorAndMinor } from 'src/utils/version/compare-version-minor-and-major'; +import { AddPositionsToWorkflowVersionsAndWorkflowRuns } from 'src/database/commands/upgrade-version-command/1-5/1-5-add-positions-to-workflow-versions-and-workflow-runs.command'; const execPromise = promisify(exec); @@ -165,6 +166,7 @@ export class UpgradeCommand extends UpgradeCommandRunner { // 1.5 Commands protected readonly removeFavoriteViewRelation: RemoveFavoriteViewRelation, + protected readonly addPositionsToWorkflowVersionsAndWorkflowRuns: AddPositionsToWorkflowVersionsAndWorkflowRuns, ) { super( workspaceRepository, @@ -239,7 +241,10 @@ export class UpgradeCommand extends UpgradeCommandRunner { }; const commands_150: VersionCommands = { - beforeSyncMetadata: [this.removeFavoriteViewRelation], + beforeSyncMetadata: [ + this.removeFavoriteViewRelation, + this.addPositionsToWorkflowVersionsAndWorkflowRuns, + ], afterSyncMetadata: [], };