chore: Refactoring dag app services

App.go

@@ -22,6 +22,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"github.com/devtron-labs/common-lib/middlewares"
+	"github.com/devtron-labs/devtron/pkg/eventProcessor"
 	"log"
 	"net/http"
 	"os"
@@ -45,14 +46,15 @@ import (
 )
 
 type App struct {
-	MuxRouter     *router.MuxRouter
-	Logger        *zap.SugaredLogger
-	SSE           *sse.SSE
-	Enforcer      *casbin.SyncedEnforcer
-	server        *http.Server
-	db            *pg.DB
-	pubsubClient  *pubsub.PubSubClientServiceImpl
-	posthogClient *telemetry.PosthogClient
+	MuxRouter             *router.MuxRouter
+	Logger                *zap.SugaredLogger
+	SSE                   *sse.SSE
+	Enforcer              *casbin.SyncedEnforcer
+	server                *http.Server
+	db                    *pg.DB
+	pubsubClient          *pubsub.PubSubClientServiceImpl
+	posthogClient         *telemetry.PosthogClient
+	centralEventProcessor *eventProcessor.CentralEventProcessor
 	// used for local dev only
 	serveTls           bool
 	sessionManager2    *authMiddleware.SessionManager
@@ -69,21 +71,23 @@ func NewApp(router *router.MuxRouter,
 	sessionManager2 *authMiddleware.SessionManager,
 	posthogClient *telemetry.PosthogClient,
 	loggingMiddleware util.LoggingMiddleware,
+	centralEventProcessor *eventProcessor.CentralEventProcessor,
 ) *App {
 	//check argo connection
 	//todo - check argo-cd version on acd integration installation
 	app := &App{
-		MuxRouter:          router,
-		Logger:             Logger,
-		SSE:                sse,
-		Enforcer:           enforcer,
-		db:                 db,
-		pubsubClient:       pubsubClient,
-		serveTls:           false,
-		sessionManager2:    sessionManager2,
-		posthogClient:      posthogClient,
-		OtelTracingService: otel.NewOtelTracingServiceImpl(Logger),
-		loggingMiddleware:  loggingMiddleware,
+		MuxRouter:             router,
+		Logger:                Logger,
+		SSE:                   sse,
+		Enforcer:              enforcer,
+		db:                    db,
+		pubsubClient:          pubsubClient,
+		serveTls:              false,
+		sessionManager2:       sessionManager2,
+		posthogClient:         posthogClient,
+		OtelTracingService:    otel.NewOtelTracingServiceImpl(Logger),
+		loggingMiddleware:     loggingMiddleware,
+		centralEventProcessor: centralEventProcessor,
 	}
 	return app
 }

Wire.go

@@ -112,6 +112,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/appStore/installedApp/service/FullMode/resource"
 	"github.com/devtron-labs/devtron/pkg/appWorkflow"
 	"github.com/devtron-labs/devtron/pkg/attributes"
+	"github.com/devtron-labs/devtron/pkg/build"
 	"github.com/devtron-labs/devtron/pkg/bulkAction"
 	"github.com/devtron-labs/devtron/pkg/chart"
 	chartRepoRepository "github.com/devtron-labs/devtron/pkg/chartRepo/repository"
@@ -123,6 +124,7 @@ import (
 	"github.com/devtron-labs/devtron/pkg/devtronResource"
 	repository9 "github.com/devtron-labs/devtron/pkg/devtronResource/repository"
 	"github.com/devtron-labs/devtron/pkg/dockerRegistry"
+	"github.com/devtron-labs/devtron/pkg/eventProcessor"
 	"github.com/devtron-labs/devtron/pkg/generateManifest"
 	"github.com/devtron-labs/devtron/pkg/git"
 	"github.com/devtron-labs/devtron/pkg/gitops"
@@ -149,6 +151,8 @@ import (
 	"github.com/devtron-labs/devtron/pkg/variables"
 	"github.com/devtron-labs/devtron/pkg/variables/parsers"
 	repository10 "github.com/devtron-labs/devtron/pkg/variables/repository"
+	workflow3 "github.com/devtron-labs/devtron/pkg/workflow"
+	"github.com/devtron-labs/devtron/pkg/workflow/dag"
 	util2 "github.com/devtron-labs/devtron/util"
 	"github.com/devtron-labs/devtron/util/argo"
 	cron2 "github.com/devtron-labs/devtron/util/cron"
@@ -183,9 +187,12 @@ func InitializeApp() (*App, error) {
 		apiToken.ApiTokenWireSet,
 		webhookHelm.WebhookHelmWireSet,
 		terminal.TerminalWireSet,
+		build.BuildWireSet,
 		deployment2.DeploymentWireSet,
 		argoApplication.ArgoApplicationWireSet,
 
+		eventProcessor.EventProcessorWireSet,
+		workflow3.WorkflowWireSet,
 		// -------wireset end ----------
 		// -------
 		gitSensor.GetConfig,
@@ -483,21 +490,12 @@ func InitializeApp() (*App, error) {
 		pubsub.NewGitWebhookHandler,
 		wire.Bind(new(pubsub.GitWebhookHandler), new(*pubsub.GitWebhookHandlerImpl)),
 
-		pubsub.NewWorkflowStatusUpdateHandlerImpl,
-		wire.Bind(new(pubsub.WorkflowStatusUpdateHandler), new(*pubsub.WorkflowStatusUpdateHandlerImpl)),
-
 		pubsub.NewApplicationStatusHandlerImpl,
 		wire.Bind(new(pubsub.ApplicationStatusHandler), new(*pubsub.ApplicationStatusHandlerImpl)),
 
-		pubsub.GetCiEventConfig,
-		pubsub.NewCiEventHandlerImpl,
-		wire.Bind(new(pubsub.CiEventHandler), new(*pubsub.CiEventHandlerImpl)),
-
 		rbac.NewEnforcerUtilImpl,
 		wire.Bind(new(rbac.EnforcerUtil), new(*rbac.EnforcerUtilImpl)),
 
-		app.NewDeploymentEventHandlerImpl,
-		wire.Bind(new(app.DeploymentEventHandler), new(*app.DeploymentEventHandlerImpl)),
 		chartConfig.NewPipelineConfigRepository,
 		wire.Bind(new(chartConfig.PipelineConfigRepository), new(*chartConfig.PipelineConfigRepositoryImpl)),
 
@@ -582,8 +580,8 @@ func InitializeApp() (*App, error) {
 		pipeline.NewBlobStorageConfigServiceImpl,
 		wire.Bind(new(pipeline.BlobStorageConfigService), new(*pipeline.BlobStorageConfigServiceImpl)),
 
-		pipeline.NewWorkflowDagExecutorImpl,
-		wire.Bind(new(pipeline.WorkflowDagExecutor), new(*pipeline.WorkflowDagExecutorImpl)),
+		dag.NewWorkflowDagExecutorImpl,
+		wire.Bind(new(dag.WorkflowDagExecutor), new(*dag.WorkflowDagExecutorImpl)),
 		appClone.NewAppCloneServiceImpl,
 		wire.Bind(new(appClone.AppCloneService), new(*appClone.AppCloneServiceImpl)),
 

api/bean/ValuesOverrideRequest.go

@@ -21,7 +21,6 @@ import (
 	"encoding/json"
 	"github.com/devtron-labs/devtron/internal/sql/models"
 	"github.com/devtron-labs/devtron/pkg/pipeline/repository"
-	"time"
 )
 
 type WorkflowType string
@@ -81,27 +80,11 @@ type BulkCdDeployEvent struct {
 	UserId                int32                  `json:"userId"`
 }
 
-type AsyncCdDeployEvent struct {
-	ValuesOverrideRequest *ValuesOverrideRequest `json:"valuesOverrideRequest"`
-	TriggeredAt           time.Time              `json:"triggeredAt"`
-	TriggeredBy           int32                  `json:"triggeredBy"`
-}
-
 type ReleaseStatusUpdateRequest struct {
 	RequestId string             `json:"requestId"`
 	NewStatus models.ChartStatus `json:"newStatus"`
 }
 
-type TriggerEvent struct {
-	PerformChartPush           bool
-	PerformDeploymentOnCluster bool
-	GetManifestInResponse      bool
-	DeploymentAppType          string
-	ManifestStorageType        string
-	TriggeredBy                int32
-	TriggerdAt                 time.Time
-}
-
 type ArtifactsListFilterOptions struct {
 	//list filter data
 	Limit        int

api/restHandler/ExternalCiRestHandler.go

@@ -19,16 +19,15 @@ package restHandler
 
 import (
 	"encoding/json"
+	"github.com/devtron-labs/devtron/pkg/workflow/dag"
 	util3 "github.com/devtron-labs/devtron/api/util"
 	"net/http"
 	"strconv"
 
 	"github.com/devtron-labs/devtron/api/restHandler/common"
-	"github.com/devtron-labs/devtron/api/router/pubsub"
 	"github.com/devtron-labs/devtron/pkg/auth/authorisation/casbin"
 	"github.com/devtron-labs/devtron/pkg/auth/user"
 	"github.com/devtron-labs/devtron/pkg/pipeline"
-	"github.com/devtron-labs/devtron/util/rbac"
 	"github.com/gorilla/mux"
 	"go.uber.org/zap"
 	"gopkg.in/go-playground/validator.v9"
@@ -39,27 +38,22 @@ type ExternalCiRestHandler interface {
 }
 
 type ExternalCiRestHandlerImpl struct {
-	logger         *zap.SugaredLogger
-	webhookService pipeline.WebhookService
-	ciEventHandler pubsub.CiEventHandler
-	validator      *validator.Validate
-	userService    user.UserService
-	enforcer       casbin.Enforcer
-	enforcerUtil   rbac.EnforcerUtil
+	logger              *zap.SugaredLogger
+	validator           *validator.Validate
+	userService         user.UserService
+	enforcer            casbin.Enforcer
+	workflowDagExecutor dag.WorkflowDagExecutor
 }
 
-func NewExternalCiRestHandlerImpl(logger *zap.SugaredLogger, webhookService pipeline.WebhookService,
-	ciEventHandler pubsub.CiEventHandler, validator *validator.Validate, userService user.UserService,
-	enforcer casbin.Enforcer,
-	enforcerUtil rbac.EnforcerUtil) *ExternalCiRestHandlerImpl {
+func NewExternalCiRestHandlerImpl(logger *zap.SugaredLogger, validator *validator.Validate,
+	userService user.UserService, enforcer casbin.Enforcer,
+	workflowDagExecutor dag.WorkflowDagExecutor) *ExternalCiRestHandlerImpl {
 	return &ExternalCiRestHandlerImpl{
-		webhookService: webhookService,
-		logger:         logger,
-		ciEventHandler: ciEventHandler,
-		validator:      validator,
-		userService:    userService,
-		enforcer:       enforcer,
-		enforcerUtil:   enforcerUtil,
+		logger:              logger,
+		validator:           validator,
+		userService:         userService,
+		enforcer:            enforcer,
+		workflowDagExecutor: workflowDagExecutor,
 	}
 }
 
@@ -78,7 +72,7 @@ func (impl ExternalCiRestHandlerImpl) HandleExternalCiWebhook(w http.ResponseWri
 		return
 	}
 	decoder := json.NewDecoder(r.Body)
-	var req pubsub.CiCompleteEvent
+	var req pipeline.ExternalCiWebhookDto
 	err = decoder.Decode(&req)
 	if err != nil {
 		impl.logger.Errorw("request err, HandleExternalCiWebhook", "err", err, "payload", req)
@@ -95,7 +89,7 @@ func (impl ExternalCiRestHandlerImpl) HandleExternalCiWebhook(w http.ResponseWri
 		return
 	}
 	//fetching request
-	ciArtifactReq, err := impl.ciEventHandler.BuildCiArtifactRequestForWebhook(req)
+	ciArtifactReq, err := impl.workflowDagExecutor.BuildCiArtifactRequestForWebhook(req)
 	if err != nil {
 		impl.logger.Errorw("service err, HandleExternalCiWebhook", "err", err, "payload", req)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
@@ -108,7 +102,7 @@ func (impl ExternalCiRestHandlerImpl) HandleExternalCiWebhook(w http.ResponseWri
 		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
 		return
 	}
-	_, err = impl.webhookService.HandleExternalCiWebhook(externalCiId, ciArtifactReq, impl.checkExternalCiDeploymentAuth, token)
+	_, err = impl.workflowDagExecutor.HandleExternalCiWebhook(externalCiId, ciArtifactReq, impl.checkExternalCiDeploymentAuth, token)
 	if err != nil {
 		impl.logger.Errorw("service err, HandleExternalCiWebhook", "err", err, "payload", req)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)