refactor: changes ciHandler

Wire.go

@@ -144,6 +144,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/deploymentGroup"
 	"github.com/devtron-labs/devtron/pkg/dockerRegistry"
 	"github.com/devtron-labs/devtron/pkg/eventProcessor"
+	"github.com/devtron-labs/devtron/pkg/executor"
 	"github.com/devtron-labs/devtron/pkg/generateManifest"
 	"github.com/devtron-labs/devtron/pkg/gitops"
 	"github.com/devtron-labs/devtron/pkg/imageDigestPolicy"
@@ -219,7 +220,7 @@ func InitializeApp() (*App, error) {
 		userResource.UserResourceWireSet,
 		policyGovernance.PolicyGovernanceWireSet,
 		resourceScan.ScanningResultWireSet,
-
+		executor.ExecutorWireSet,
 		// -------wireset end ----------
 		// -------
 		gitSensor.GetConfig,
@@ -478,9 +479,6 @@ func InitializeApp() (*App, error) {
 		util.IntValidator,
 		types.GetCiCdConfig,
 
-		pipeline.NewWorkflowServiceImpl,
-		wire.Bind(new(pipeline.WorkflowService), new(*pipeline.WorkflowServiceImpl)),
-
 		pipeline.NewCiServiceImpl,
 		wire.Bind(new(pipeline.CiService), new(*pipeline.CiServiceImpl)),
 

api/argoApplication/wire_argoApplication.go

@@ -17,7 +17,7 @@
 package argoApplication
 
 import (
-	"github.com/devtron-labs/devtron/pkg/argoApplication"
+	argoApplication3 "github.com/devtron-labs/devtron/pkg/argoApplication"
 	"github.com/devtron-labs/devtron/pkg/argoApplication/read"
 	"github.com/devtron-labs/devtron/pkg/argoApplication/read/config"
 	"github.com/google/wire"
@@ -30,9 +30,9 @@ var ArgoApplicationWireSetFull = wire.NewSet(
 	config.NewArgoApplicationConfigServiceImpl,
 	wire.Bind(new(config.ArgoApplicationConfigService), new(*config.ArgoApplicationConfigServiceImpl)),
 
-	argoApplication.NewArgoApplicationServiceImpl,
-	argoApplication.NewArgoApplicationServiceExtendedServiceImpl,
-	wire.Bind(new(argoApplication.ArgoApplicationService), new(*argoApplication.ArgoApplicationServiceExtendedImpl)),
+	argoApplication3.NewArgoApplicationServiceImpl,
+	argoApplication3.NewArgoApplicationServiceExtendedServiceImpl,
+	wire.Bind(new(argoApplication3.ArgoApplicationService), new(*argoApplication3.ArgoApplicationServiceExtendedImpl)),
 
 	NewArgoApplicationRestHandlerImpl,
 	wire.Bind(new(ArgoApplicationRestHandler), new(*ArgoApplicationRestHandlerImpl)),
@@ -48,8 +48,8 @@ var ArgoApplicationWireSetEA = wire.NewSet(
 	config.NewArgoApplicationConfigServiceImpl,
 	wire.Bind(new(config.ArgoApplicationConfigService), new(*config.ArgoApplicationConfigServiceImpl)),
 
-	argoApplication.NewArgoApplicationServiceImpl,
-	wire.Bind(new(argoApplication.ArgoApplicationService), new(*argoApplication.ArgoApplicationServiceImpl)),
+	argoApplication3.NewArgoApplicationServiceImpl,
+	wire.Bind(new(argoApplication3.ArgoApplicationService), new(*argoApplication3.ArgoApplicationServiceImpl)),
 
 	NewArgoApplicationRestHandlerImpl,
 	wire.Bind(new(ArgoApplicationRestHandler), new(*ArgoApplicationRestHandlerImpl)),

api/restHandler/CoreAppRestHandler.go

@@ -29,6 +29,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/build/git/gitProvider"
 	"github.com/devtron-labs/devtron/pkg/build/git/gitProvider/read"
 	pipelineBean "github.com/devtron-labs/devtron/pkg/build/pipeline/bean"
+	common2 "github.com/devtron-labs/devtron/pkg/build/pipeline/bean/common"
 	bean3 "github.com/devtron-labs/devtron/pkg/chart/bean"
 	read5 "github.com/devtron-labs/devtron/pkg/chart/read"
 	"github.com/devtron-labs/devtron/pkg/cluster/environment/repository"
@@ -1688,7 +1689,7 @@ func (handler CoreAppRestHandlerImpl) createCiPipeline(appId int, userId int32,
 			ParentCiPipeline:         ciPipelineData.ParentCiPipeline,
 			ParentAppId:              ciPipelineData.ParentAppId,
 			LinkedCount:              ciPipelineData.LinkedCount,
-			PipelineType:             pipelineBean.PipelineType(ciPipelineData.PipelineType),
+			PipelineType:             common2.PipelineType(ciPipelineData.PipelineType),
 		},
 	}
 

api/restHandler/app/pipeline/configure/BuildPipelineRestHandler.go

@@ -740,7 +740,7 @@ func (handler *PipelineConfigRestHandlerImpl) TriggerCiPipeline(w http.ResponseW
 	ciTriggerRequest.TriggeredBy = userId
 	handler.Logger.Infow("request payload, TriggerCiPipeline", "payload", ciTriggerRequest)
 	response := make(map[string]string)
-	resp, err := handler.ciHandler.HandleCIManual(ciTriggerRequest)
+	resp, err := handler.ciHandlerService.HandleCIManual(ciTriggerRequest)
 	if errors.Is(err, bean1.ErrImagePathInUse) {
 		handler.Logger.Errorw("service err duplicate image tag, TriggerCiPipeline", "err", err, "payload", ciTriggerRequest)
 		common.WriteJsonResp(w, err, err, http.StatusConflict)
@@ -974,7 +974,7 @@ func (handler *PipelineConfigRestHandlerImpl) DownloadCiWorkflowArtifacts(w http
 	}
 	//RBAC
 
-	file, err := handler.ciHandler.DownloadCiWorkflowArtifacts(pipelineId, buildId)
+	file, err := handler.ciHandlerService.DownloadCiWorkflowArtifacts(pipelineId, buildId)
 	defer file.Close()
 	if err != nil {
 		handler.Logger.Errorw("service err, DownloadCiWorkflowArtifacts", "err", err, "pipelineId", pipelineId, "buildId", buildId)
@@ -1029,7 +1029,7 @@ func (handler *PipelineConfigRestHandlerImpl) GetHistoricBuildLogs(w http.Respon
 		return
 	}
 	//RBAC
-	resp, err := handler.ciHandler.GetHistoricBuildLogs(workflowId, nil)
+	resp, err := handler.ciHandlerService.GetHistoricBuildLogs(workflowId, nil)
 	if err != nil {
 		handler.Logger.Errorw("service err, GetHistoricBuildLogs", "err", err, "pipelineId", pipelineId, "workflowId", workflowId)
 		common.WriteJsonResp(w, err, resp, http.StatusInternalServerError)
@@ -1168,7 +1168,7 @@ func (handler *PipelineConfigRestHandlerImpl) GetBuildLogs(w http.ResponseWriter
 			return
 		}
 	}
-	logsReader, cleanUp, err := handler.ciHandler.GetRunningWorkflowLogs(workflowId)
+	logsReader, cleanUp, err := handler.ciHandlerService.GetRunningWorkflowLogs(workflowId)
 	if err != nil {
 		handler.Logger.Errorw("service err, GetBuildLogs", "err", err, "pipelineId", pipelineId, "workflowId", workflowId, "lastEventId", lastEventId)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
@@ -1611,7 +1611,7 @@ func (handler *PipelineConfigRestHandlerImpl) CancelWorkflow(w http.ResponseWrit
 
 	//RBAC
 
-	resp, err := handler.ciHandler.CancelBuild(workflowId, forceAbort)
+	resp, err := handler.ciHandlerService.CancelBuild(workflowId, forceAbort)
 	if err != nil {
 		handler.Logger.Errorw("service err, CancelWorkflow", "err", err, "workflowId", workflowId, "pipelineId", pipelineId)
 		if util.IsErrNoRows(err) {

api/restHandler/app/pipeline/configure/DeploymentPipelineRestHandler.go

@@ -1620,7 +1620,7 @@ func (handler *PipelineConfigRestHandlerImpl) GetPrePostDeploymentLogs(w http.Re
 	}
 	// RBAC CHECK
 
-	logsReader, cleanUp, err := handler.cdHandler.GetRunningWorkflowLogs(environmentId, pipelineId, workflowId)
+	logsReader, cleanUp, err := handler.cdHandlerService.GetRunningWorkflowLogs(environmentId, pipelineId, workflowId)
 	if err != nil {
 		handler.Logger.Errorw("service err, GetPrePostDeploymentLogs", "err", err, "appId", appId, "environmentId", environmentId, "pipelineId", pipelineId, "workflowId", workflowId)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
@@ -1745,7 +1745,7 @@ func (handler *PipelineConfigRestHandlerImpl) DownloadArtifacts(w http.ResponseW
 	}
 	// RBAC CHECK
 
-	file, err := handler.cdHandler.DownloadCdWorkflowArtifacts(buildId)
+	file, err := handler.cdHandlerService.DownloadCdWorkflowArtifacts(buildId)
 	defer file.Close()
 
 	if err != nil {
@@ -1929,7 +1929,7 @@ func (handler *PipelineConfigRestHandlerImpl) CancelStage(w http.ResponseWriter,
 	}
 	// RBAC
 
-	resp, err := handler.cdHandler.CancelStage(workflowRunnerId, forceAbort, userId)
+	resp, err := handler.cdHandlerService.CancelStage(workflowRunnerId, forceAbort, userId)
 	if err != nil {
 		handler.Logger.Errorw("service err, CancelStage", "err", err, "pipelineId", pipelineId, "workflowRunnerId", workflowRunnerId)
 		if util.IsErrNoRows(err) {

api/restHandler/app/pipeline/configure/PipelineConfigRestHandler.go

@@ -26,13 +26,15 @@ import (
 	read2 "github.com/devtron-labs/devtron/pkg/build/git/gitMaterial/read"
 	gitProviderRead "github.com/devtron-labs/devtron/pkg/build/git/gitProvider/read"
 	bean3 "github.com/devtron-labs/devtron/pkg/build/pipeline/bean"
+	"github.com/devtron-labs/devtron/pkg/build/trigger"
 	"github.com/devtron-labs/devtron/pkg/chart/gitOpsConfig"
 	read5 "github.com/devtron-labs/devtron/pkg/chart/read"
 	repository2 "github.com/devtron-labs/devtron/pkg/cluster/environment/repository"
 	"github.com/devtron-labs/devtron/pkg/deployment/manifest/deployedAppMetrics"
 	"github.com/devtron-labs/devtron/pkg/deployment/manifest/deploymentTemplate/chartRef"
-	"github.com/devtron-labs/devtron/pkg/pipeline/draftAwareConfigService"
 	validator2 "github.com/devtron-labs/devtron/pkg/deployment/manifest/deploymentTemplate/validator"
+	"github.com/devtron-labs/devtron/pkg/deployment/trigger/devtronApps"
+	"github.com/devtron-labs/devtron/pkg/pipeline/draftAwareConfigService"
 	security2 "github.com/devtron-labs/devtron/pkg/policyGovernance/security/imageScanning"
 	"github.com/devtron-labs/devtron/pkg/policyGovernance/security/imageScanning/read"
 	read3 "github.com/devtron-labs/devtron/pkg/team/read"
@@ -140,6 +142,8 @@ type PipelineConfigRestHandlerImpl struct {
 	environmentRepository               repository2.EnvironmentRepository
 	chartReadService                    read5.ChartReadService
 	draftAwareResourceService           draftAwareConfigService.DraftAwareConfigService
+	ciHandlerService                    trigger.HandlerService
+	cdHandlerService                    devtronApps.HandlerService
 }
 
 func NewPipelineRestHandlerImpl(pipelineBuilder pipeline.PipelineBuilder, Logger *zap.SugaredLogger,
@@ -175,6 +179,8 @@ func NewPipelineRestHandlerImpl(pipelineBuilder pipeline.PipelineBuilder, Logger
 	EnvironmentRepository repository2.EnvironmentRepository,
 	chartReadService read5.ChartReadService,
 	draftAwareResourceService draftAwareConfigService.DraftAwareConfigService,
+	ciHandlerService trigger.HandlerService,
+	cdHandlerService devtronApps.HandlerService,
 ) *PipelineConfigRestHandlerImpl {
 	envConfig := &PipelineRestHandlerEnvConfig{}
 	err := env.Parse(envConfig)
@@ -218,6 +224,8 @@ func NewPipelineRestHandlerImpl(pipelineBuilder pipeline.PipelineBuilder, Logger
 		environmentRepository:               EnvironmentRepository,
 		chartReadService:                    chartReadService,
 		draftAwareResourceService:           draftAwareResourceService,
+		ciHandlerService:                    ciHandlerService,
+		cdHandlerService:                    cdHandlerService,
 	}
 }
 

api/restHandler/app/pipeline/trigger/PipelineTriggerRestHandler.go

@@ -66,7 +66,7 @@ type PipelineTriggerRestHandlerImpl struct {
 	deploymentGroupService      deploymentGroup.DeploymentGroupService
 	deploymentConfigService     pipeline.PipelineDeploymentConfigService
 	deployedAppService          deployedApp.DeployedAppService
-	cdTriggerService            devtronApps.TriggerService
+	cdHandlerService            devtronApps.HandlerService
 	workflowEventPublishService out.WorkflowEventPublishService
 }
 
@@ -75,7 +75,7 @@ func NewPipelineRestHandler(appService app.AppService, userAuthService user.User
 	deploymentGroupService deploymentGroup.DeploymentGroupService,
 	deploymentConfigService pipeline.PipelineDeploymentConfigService,
 	deployedAppService deployedApp.DeployedAppService,
-	cdTriggerService devtronApps.TriggerService,
+	cdHandlerService devtronApps.HandlerService,
 	workflowEventPublishService out.WorkflowEventPublishService) *PipelineTriggerRestHandlerImpl {
 	pipelineHandler := &PipelineTriggerRestHandlerImpl{
 		appService:                  appService,
@@ -88,7 +88,7 @@ func NewPipelineRestHandler(appService app.AppService, userAuthService user.User
 		deploymentGroupService:      deploymentGroupService,
 		deploymentConfigService:     deploymentConfigService,
 		deployedAppService:          deployedAppService,
-		cdTriggerService:            cdTriggerService,
+		cdHandlerService:            cdHandlerService,
 		workflowEventPublishService: workflowEventPublishService,
 	}
 	return pipelineHandler
@@ -140,7 +140,7 @@ func (handler PipelineTriggerRestHandlerImpl) OverrideConfig(w http.ResponseWrit
 	triggerContext := bean3.TriggerContext{
 		Context: ctx,
 	}
-	mergeResp, helmPackageName, _, err := handler.cdTriggerService.ManualCdTrigger(triggerContext, &overrideRequest)
+	mergeResp, helmPackageName, _, err := handler.cdHandlerService.ManualCdTrigger(triggerContext, &overrideRequest)
 	span.End()
 	if err != nil {
 		handler.logger.Errorw("request err, OverrideConfig", "err", err, "payload", overrideRequest)

client/cron/CiStatusUpdateCron.go

@@ -21,7 +21,7 @@ import (
 	"github.com/caarlos0/env"
 	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
 	"github.com/devtron-labs/devtron/pkg/app"
-	"github.com/devtron-labs/devtron/pkg/pipeline"
+	"github.com/devtron-labs/devtron/pkg/workflow/dag"
 	cron2 "github.com/devtron-labs/devtron/util/cron"
 	"github.com/robfig/cron/v3"
 	"go.uber.org/zap"
@@ -38,12 +38,13 @@ type CiStatusUpdateCronImpl struct {
 	appService                   app.AppService
 	ciWorkflowStatusUpdateConfig *CiWorkflowStatusUpdateConfig
 	ciPipelineRepository         pipelineConfig.CiPipelineRepository
-	ciHandler                    pipeline.CiHandler
+	workflowDagExecutor          dag.WorkflowDagExecutor
 }
 
 func NewCiStatusUpdateCronImpl(logger *zap.SugaredLogger, appService app.AppService,
 	ciWorkflowStatusUpdateConfig *CiWorkflowStatusUpdateConfig, ciPipelineRepository pipelineConfig.CiPipelineRepository,
-	ciHandler pipeline.CiHandler, cronLogger *cron2.CronLoggerImpl) *CiStatusUpdateCronImpl {
+	cronLogger *cron2.CronLoggerImpl,
+	workflowDagExecutor dag.WorkflowDagExecutor) *CiStatusUpdateCronImpl {
 	cron := cron.New(
 		cron.WithChain(cron.Recover(cronLogger)))
 	cron.Start()
@@ -53,7 +54,7 @@ func NewCiStatusUpdateCronImpl(logger *zap.SugaredLogger, appService app.AppServ
 		appService:                   appService,
 		ciWorkflowStatusUpdateConfig: ciWorkflowStatusUpdateConfig,
 		ciPipelineRepository:         ciPipelineRepository,
-		ciHandler:                    ciHandler,
+		workflowDagExecutor:          workflowDagExecutor,
 	}
 
 	// execute periodically, update ci workflow status for failed process
@@ -87,7 +88,7 @@ func (impl *CiStatusUpdateCronImpl) UpdateCiWorkflowStatusFailedCron() {
 		impl.logger.Errorw("error in converting string to int", "err", err)
 		return
 	}
-	err = impl.ciHandler.UpdateCiWorkflowStatusFailure(timeoutForFailureCiBuild)
+	err = impl.workflowDagExecutor.UpdateCiWorkflowStatusFailure(timeoutForFailureCiBuild)
 	if err != nil {
 		impl.logger.Errorw("error in updating ci workflow status for failed workflows", "err", err)
 		return

client/cron/CiTriggerCron.go

@@ -22,8 +22,8 @@ import (
 	repository2 "github.com/devtron-labs/devtron/internal/sql/repository"
 	bean2 "github.com/devtron-labs/devtron/pkg/auth/user/bean"
 	"github.com/devtron-labs/devtron/pkg/bean"
-	pipelineConfigBean "github.com/devtron-labs/devtron/pkg/build/pipeline/bean"
-	"github.com/devtron-labs/devtron/pkg/pipeline"
+	"github.com/devtron-labs/devtron/pkg/build/pipeline/bean/common"
+	"github.com/devtron-labs/devtron/pkg/build/trigger"
 	"github.com/devtron-labs/devtron/pkg/pipeline/repository"
 	repository3 "github.com/devtron-labs/devtron/pkg/plugin/repository"
 	cron2 "github.com/devtron-labs/devtron/util/cron"
@@ -40,24 +40,25 @@ type CiTriggerCronImpl struct {
 	cron                    *cron.Cron
 	cfg                     *CiTriggerCronConfig
 	pipelineStageRepository repository.PipelineStageRepository
-	ciHandler               pipeline.CiHandler
 	ciArtifactRepository    repository2.CiArtifactRepository
 	globalPluginRepository  repository3.GlobalPluginRepository
+	ciHandlerService        trigger.HandlerService
 }
 
 func NewCiTriggerCronImpl(logger *zap.SugaredLogger, cfg *CiTriggerCronConfig, pipelineStageRepository repository.PipelineStageRepository,
-	ciHandler pipeline.CiHandler, ciArtifactRepository repository2.CiArtifactRepository, globalPluginRepository repository3.GlobalPluginRepository, cronLogger *cron2.CronLoggerImpl) *CiTriggerCronImpl {
+	ciArtifactRepository repository2.CiArtifactRepository, globalPluginRepository repository3.GlobalPluginRepository, cronLogger *cron2.CronLoggerImpl,
+	ciHandlerService trigger.HandlerService) *CiTriggerCronImpl {
 	cron := cron.New(
 		cron.WithChain(cron.Recover(cronLogger)))
 	cron.Start()
 	impl := &CiTriggerCronImpl{
 		logger:                  logger,
 		cron:                    cron,
 		pipelineStageRepository: pipelineStageRepository,
-		ciHandler:               ciHandler,
 		cfg:                     cfg,
 		ciArtifactRepository:    ciArtifactRepository,
 		globalPluginRepository:  globalPluginRepository,
+		ciHandlerService:        ciHandlerService,
 	}
 
 	_, err := cron.AddFunc(fmt.Sprintf("@every %dm", cfg.SourceControllerCronTime), impl.TriggerCiCron)
@@ -101,9 +102,9 @@ func (impl *CiTriggerCronImpl) TriggerCiCron() {
 			CiPipelineMaterial: ciPipelineMaterials,
 			TriggeredBy:        bean2.SYSTEM_USER_ID,
 			InvalidateCache:    false,
-			PipelineType:       string(pipelineConfigBean.CI_JOB),
+			PipelineType:       string(common.CI_JOB),
 		}
-		_, err = impl.ciHandler.HandleCIManual(ciTriggerRequest)
+		_, err = impl.ciHandlerService.HandleCIManual(ciTriggerRequest)
 		if err != nil {
 			return
 		}

client/events/EventBuilder.go

@@ -19,6 +19,7 @@ package client
 import (
 	"context"
 	"fmt"
+	buildBean "github.com/devtron-labs/devtron/pkg/build/pipeline/bean"
 	repository4 "github.com/devtron-labs/devtron/pkg/cluster/environment/repository"
 	"strings"
 	"time"
@@ -38,7 +39,7 @@ import (
 type EventFactory interface {
 	Build(eventType util.EventType, sourceId *int, appId int, envId *int, pipelineType util.PipelineType) (Event, error)
 	BuildExtraCDData(event Event, wfr *pipelineConfig.CdWorkflowRunner, pipelineOverrideId int, stage bean2.WorkflowType) Event
-	BuildExtraCIData(event Event, material *MaterialTriggerInfo) Event
+	BuildExtraCIData(event Event, material *buildBean.MaterialTriggerInfo) Event
 	//BuildFinalData(event Event) *Payload
 }
 
@@ -173,7 +174,7 @@ func (impl *EventSimpleFactoryImpl) BuildExtraCDData(event Event, wfr *pipelineC
 	return event
 }
 
-func (impl *EventSimpleFactoryImpl) BuildExtraCIData(event Event, material *MaterialTriggerInfo) Event {
+func (impl *EventSimpleFactoryImpl) BuildExtraCIData(event Event, material *buildBean.MaterialTriggerInfo) Event {
 	if material == nil {
 		materialInfo, err := impl.getCiMaterialInfo(event.PipelineId, event.CiArtifactId)
 		if err != nil {
@@ -207,22 +208,22 @@ func (impl *EventSimpleFactoryImpl) BuildExtraCIData(event Event, material *Mate
 	return event
 }
 
-func (impl *EventSimpleFactoryImpl) getCiMaterialInfo(ciPipelineId int, ciArtifactId int) (*MaterialTriggerInfo, error) {
-	materialTriggerInfo := &MaterialTriggerInfo{}
+func (impl *EventSimpleFactoryImpl) getCiMaterialInfo(ciPipelineId int, ciArtifactId int) (*buildBean.MaterialTriggerInfo, error) {
+	materialTriggerInfo := &buildBean.MaterialTriggerInfo{}
 	if ciPipelineId > 0 {
 		ciMaterials, err := impl.ciPipelineMaterialRepository.GetByPipelineId(ciPipelineId)
 		if err != nil {
 			impl.logger.Errorw("error on fetching materials for", "ciPipelineId", ciPipelineId, "err", err)
 			return nil, err
 		}
 
-		var ciMaterialsArr []CiPipelineMaterialResponse
+		var ciMaterialsArr []buildBean.CiPipelineMaterialResponse
 		for _, m := range ciMaterials {
 			if m.GitMaterial == nil {
 				impl.logger.Warnw("git material are empty", "material", m)
 				continue
 			}
-			res := CiPipelineMaterialResponse{
+			res := buildBean.CiPipelineMaterialResponse{
 				Id:              m.Id,
 				GitMaterialId:   m.GitMaterialId,
 				GitMaterialName: m.GitMaterial.Name[strings.Index(m.GitMaterial.Name, "-")+1:],