refactor: Refactor ci and multi cluster cicd (#6505)

* changes around ciHandler

* common constants create

* ci service and trigger service creation

* wire file missed

* trigger service moved completely

* feat: added ucid validation for ci/cd workflow status update event

* telemetry svc refactoring

* updated wire gen files

* fix: CheckIfReTriggerRequired condition and CdHandlerImpl.UpdateWorkflow method

* fix: workflow message update

* updated import alias

* fix: ci/ cd status update inconsistencies

* fix: checkForOptOut method error handling

* fix: panic handling

* fix: updated DevtronAdministratorInstanceLabelKey for system executor

* fix: reverted base64 encoding

* fix: ci/ cd workflow msg update

* fix: ci/ cd workflow msg update

* matrics for retrigger failed event

* updated logger

* ci service ent oss completed

* added app to subdirectory for categorising entity

* cd handler and ci handle extracted, executor workflow service executed

* workflow service ent change

* workflow service ent change

* Redundant import removed

* rename pkg/

* reverted pkg categorisation

* chore: posthug refactoring and const renamed

* chore: kubewatch const renamed

* chore: removed posthug module

* chore: ci/cd workflow message datatype updated

* wip

* file renaming

* deployStage unstacked file

* file renamed

* updated BuildxCacheFlags description

---------

Co-authored-by: Ash-exp <asutosh2000ad@gmail.com>

Author: kartik-579

App.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"github.com/devtron-labs/common-lib/middlewares"
 	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
+	posthogTelemetry "github.com/devtron-labs/common-lib/telemetry"
 	"github.com/devtron-labs/devtron/pkg/eventProcessor"
 	"github.com/devtron-labs/devtron/pkg/eventProcessor/in"
 	"log"
@@ -31,7 +32,6 @@ import (
 	"time"
 
 	"github.com/devtron-labs/devtron/api/util"
-	"github.com/devtron-labs/devtron/client/telemetry"
 	"github.com/devtron-labs/devtron/otel"
 	"github.com/devtron-labs/devtron/pkg/auth/user"
 
@@ -55,7 +55,7 @@ type App struct {
 	EnforcerV2    *casbinv2.SyncedEnforcer
 	server        *http.Server
 	db            *pg.DB
-	posthogClient *telemetry.PosthogClient
+	posthogClient *posthogTelemetry.PosthogClient
 	// eventProcessor.CentralEventProcessor is used to register event processors
 	centralEventProcessor *eventProcessor.CentralEventProcessor // do not remove this.
 	// used for local dev only
@@ -73,7 +73,7 @@ func NewApp(router *router.MuxRouter,
 	enforcer *casbin.SyncedEnforcer,
 	db *pg.DB,
 	sessionManager2 *authMiddleware.SessionManager,
-	posthogClient *telemetry.PosthogClient,
+	posthogClient *posthogTelemetry.PosthogClient,
 	loggingMiddleware util.LoggingMiddleware,
 	centralEventProcessor *eventProcessor.CentralEventProcessor,
 	pubSubClient *pubsub.PubSubClientServiceImpl,

Wire.go

@@ -22,7 +22,8 @@ package main
 import (
 	"github.com/devtron-labs/authenticator/middleware"
 	cloudProviderIdentifier "github.com/devtron-labs/common-lib/cloud-provider-identifier"
-	pubsub1 "github.com/devtron-labs/common-lib/pubsub-lib"
+	pubSub "github.com/devtron-labs/common-lib/pubsub-lib"
+	posthogTelemetry "github.com/devtron-labs/common-lib/telemetry"
 	util4 "github.com/devtron-labs/common-lib/utils/k8s"
 	"github.com/devtron-labs/devtron/api/apiToken"
 	appStoreRestHandler "github.com/devtron-labs/devtron/api/appStore"
@@ -144,6 +145,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"
@@ -165,6 +167,7 @@ import (
 	resourceGroup2 "github.com/devtron-labs/devtron/pkg/resourceGroup"
 	"github.com/devtron-labs/devtron/pkg/resourceQualifiers"
 	"github.com/devtron-labs/devtron/pkg/sql"
+	"github.com/devtron-labs/devtron/pkg/ucid"
 	util3 "github.com/devtron-labs/devtron/pkg/util"
 	"github.com/devtron-labs/devtron/pkg/variables"
 	"github.com/devtron-labs/devtron/pkg/variables/parsers"
@@ -219,7 +222,7 @@ func InitializeApp() (*App, error) {
 		userResource.UserResourceWireSet,
 		policyGovernance.PolicyGovernanceWireSet,
 		resourceScan.ScanningResultWireSet,
-
+		executor.ExecutorWireSet,
 		// -------wireset end ----------
 		// -------
 		gitSensor.GetConfig,
@@ -478,9 +481,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)),
 
@@ -501,7 +501,7 @@ func InitializeApp() (*App, error) {
 		pipeline.NewCiLogServiceImpl,
 		wire.Bind(new(pipeline.CiLogService), new(*pipeline.CiLogServiceImpl)),
 
-		pubsub1.NewPubSubClientServiceImpl,
+		pubSub.NewPubSubClientServiceImpl,
 
 		rbac.NewEnforcerUtilImpl,
 		wire.Bind(new(rbac.EnforcerUtil), new(*rbac.EnforcerUtilImpl)),
@@ -696,7 +696,8 @@ func InitializeApp() (*App, error) {
 		wire.Bind(new(router.TelemetryRouter), new(*router.TelemetryRouterImpl)),
 		restHandler.NewTelemetryRestHandlerImpl,
 		wire.Bind(new(restHandler.TelemetryRestHandler), new(*restHandler.TelemetryRestHandlerImpl)),
-		telemetry.NewPosthogClient,
+		posthogTelemetry.NewPosthogClient,
+		ucid.WireSet,
 
 		cloudProviderIdentifier.NewProviderIdentifierServiceImpl,
 		wire.Bind(new(cloudProviderIdentifier.ProviderIdentifierService), new(*cloudProviderIdentifier.ProviderIdentifierServiceImpl)),

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

@@ -25,6 +25,7 @@ import (
 	"github.com/devtron-labs/devtron/internal/sql/constants"
 	"github.com/devtron-labs/devtron/pkg/build/artifacts/imageTagging"
 	bean2 "github.com/devtron-labs/devtron/pkg/build/pipeline/bean"
+	eventProcessorBean "github.com/devtron-labs/devtron/pkg/eventProcessor/bean"
 	constants2 "github.com/devtron-labs/devtron/pkg/pipeline/constants"
 	"github.com/devtron-labs/devtron/util/stringsUtil"
 	"golang.org/x/exp/maps"
@@ -37,7 +38,6 @@ import (
 	"github.com/devtron-labs/devtron/util/response/pagination"
 	"github.com/gorilla/schema"
 
-	"github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	"github.com/devtron-labs/devtron/api/restHandler/common"
 	"github.com/devtron-labs/devtron/client/gitSensor"
 	"github.com/devtron-labs/devtron/internal/sql/repository"
@@ -701,8 +701,7 @@ func (handler *PipelineConfigRestHandlerImpl) TriggerCiPipeline(w http.ResponseW
 	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)
@@ -918,7 +917,8 @@ func (handler *PipelineConfigRestHandlerImpl) DownloadCiWorkflowArtifacts(w http
 		return
 	}
 
-	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)
 		if util.IsErrNoRows(err) {
@@ -974,8 +974,8 @@ func (handler *PipelineConfigRestHandlerImpl) GetHistoricBuildLogs(w http.Respon
 		common.WriteJsonResp(w, nil, "Unauthorized User", http.StatusForbidden)
 		return
 	}
-
-	resp, err := handler.ciHandler.GetHistoricBuildLogs(workflowId, nil)
+	// RBAC
+	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)
@@ -1114,7 +1114,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)
@@ -1446,15 +1446,15 @@ func (handler *PipelineConfigRestHandlerImpl) DeleteMaterial(w http.ResponseWrit
 
 func (handler *PipelineConfigRestHandlerImpl) HandleWorkflowWebhook(w http.ResponseWriter, r *http.Request) {
 	decoder := json.NewDecoder(r.Body)
-	var wfUpdateReq v1alpha1.WorkflowStatus
+	var wfUpdateReq eventProcessorBean.CiCdStatus
 	err := decoder.Decode(&wfUpdateReq)
 	if err != nil {
 		handler.Logger.Errorw("request err, HandleWorkflowWebhook", "err", err, "payload", wfUpdateReq)
 		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
 		return
 	}
 	handler.Logger.Infow("request payload, HandleWorkflowWebhook", "payload", wfUpdateReq)
-	resp, err := handler.ciHandler.UpdateWorkflow(wfUpdateReq)
+	resp, _, err := handler.ciHandler.UpdateWorkflow(wfUpdateReq)
 	if err != nil {
 		handler.Logger.Errorw("service err, HandleWorkflowWebhook", "err", err, "payload", wfUpdateReq)
 		common.WriteJsonResp(w, err, resp, http.StatusInternalServerError)
@@ -1559,7 +1559,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,12 +26,14 @@ 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"
 	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"
@@ -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/argocdServer/config/Config.go

@@ -71,7 +71,7 @@ func (impl *ArgoCDConfigGetterImpl) GetGRPCConfig() (*bean.ArgoGRPCConfig, error
 }
 
 func (impl *ArgoCDConfigGetterImpl) GetK8sConfig() (*bean.ArgoK8sConfig, error) {
-	clusterBean, err := impl.clusterReadService.FindOne(bean2.DEFAULT_CLUSTER)
+	clusterBean, err := impl.clusterReadService.FindOne(bean2.DefaultCluster)
 	if err != nil {
 		impl.logger.Errorw("error in fetching cluster bean from db", "err", err)
 		return nil, err

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