chore: refactoring v4

App.go

@@ -35,7 +35,6 @@ import (
 
 	"github.com/casbin/casbin"
 	authMiddleware "github.com/devtron-labs/authenticator/middleware"
-	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
 	"github.com/devtron-labs/devtron/api/router"
 	"github.com/devtron-labs/devtron/api/sse"
 	"github.com/devtron-labs/devtron/internal/middleware"
@@ -52,7 +51,6 @@ type App struct {
 	Enforcer              *casbin.SyncedEnforcer
 	server                *http.Server
 	db                    *pg.DB
-	pubsubClient          *pubsub.PubSubClientServiceImpl
 	posthogClient         *telemetry.PosthogClient
 	centralEventProcessor *eventProcessor.CentralEventProcessor
 	// used for local dev only
@@ -67,7 +65,6 @@ func NewApp(router *router.MuxRouter,
 	sse *sse.SSE,
 	enforcer *casbin.SyncedEnforcer,
 	db *pg.DB,
-	pubsubClient *pubsub.PubSubClientServiceImpl,
 	sessionManager2 *authMiddleware.SessionManager,
 	posthogClient *telemetry.PosthogClient,
 	loggingMiddleware util.LoggingMiddleware,
@@ -81,7 +78,6 @@ func NewApp(router *router.MuxRouter,
 		SSE:                   sse,
 		Enforcer:              enforcer,
 		db:                    db,
-		pubsubClient:          pubsubClient,
 		serveTls:              false,
 		sessionManager2:       sessionManager2,
 		posthogClient:         posthogClient,

Wire.go

@@ -66,7 +66,6 @@ import (
 	status3 "github.com/devtron-labs/devtron/api/router/app/pipeline/status"
 	trigger2 "github.com/devtron-labs/devtron/api/router/app/pipeline/trigger"
 	workflow2 "github.com/devtron-labs/devtron/api/router/app/workflow"
-	"github.com/devtron-labs/devtron/api/router/pubsub"
 	"github.com/devtron-labs/devtron/api/server"
 	"github.com/devtron-labs/devtron/api/sse"
 	"github.com/devtron-labs/devtron/api/team"
@@ -491,12 +490,6 @@ func InitializeApp() (*App, error) {
 
 		pubsub1.NewPubSubClientServiceImpl,
 
-		pubsub.NewGitWebhookHandler,
-		wire.Bind(new(pubsub.GitWebhookHandler), new(*pubsub.GitWebhookHandlerImpl)),
-
-		pubsub.NewApplicationStatusHandlerImpl,
-		wire.Bind(new(pubsub.ApplicationStatusHandler), new(*pubsub.ApplicationStatusHandlerImpl)),
-
 		rbac.NewEnforcerUtilImpl,
 		wire.Bind(new(rbac.EnforcerUtil), new(*rbac.EnforcerUtilImpl)),
 
@@ -966,8 +959,6 @@ func InitializeApp() (*App, error) {
 		pipeline.NewPluginInputVariableParserImpl,
 		wire.Bind(new(pipeline.PluginInputVariableParser), new(*pipeline.PluginInputVariableParserImpl)),
 
-		pipeline.NewPipelineConfigListenerServiceImpl,
-		wire.Bind(new(pipeline.PipelineConfigListenerService), new(*pipeline.PipelineConfigListenerServiceImpl)),
 		cron2.NewCronLoggerImpl,
 
 		imageDigestPolicy.NewImageDigestPolicyServiceImpl,

api/restHandler/ExternalCiRestHandler.go

@@ -19,8 +19,8 @@ package restHandler
 
 import (
 	"encoding/json"
-	util3 "github.com/devtron-labs/devtron/api/util"
 	"github.com/devtron-labs/devtron/pkg/workflow/dag"
+	util3 "github.com/devtron-labs/devtron/api/util"
 	"net/http"
 	"strconv"
 

api/restHandler/WebhookEventHandler.go

@@ -18,7 +18,7 @@
 package restHandler
 
 import (
-	pubsub "github.com/devtron-labs/common-lib/pubsub-lib"
+	"github.com/devtron-labs/devtron/pkg/eventProcessor/out"
 	"io/ioutil"
 	"net/http"
 	"strconv"
@@ -36,21 +36,24 @@ type WebhookEventHandler interface {
 }
 
 type WebhookEventHandlerImpl struct {
-	logger                 *zap.SugaredLogger
-	gitHostConfig          pipeline.GitHostConfig
-	eventClient            client.EventClient
-	webhookSecretValidator git.WebhookSecretValidator
-	webhookEventDataConfig pipeline.WebhookEventDataConfig
+	logger                        *zap.SugaredLogger
+	gitHostConfig                 pipeline.GitHostConfig
+	eventClient                   client.EventClient
+	webhookSecretValidator        git.WebhookSecretValidator
+	webhookEventDataConfig        pipeline.WebhookEventDataConfig
+	ciPipelineEventPublishService out.CIPipelineEventPublishService
 }
 
 func NewWebhookEventHandlerImpl(logger *zap.SugaredLogger, gitHostConfig pipeline.GitHostConfig, eventClient client.EventClient,
-	webhookSecretValidator git.WebhookSecretValidator, webhookEventDataConfig pipeline.WebhookEventDataConfig) *WebhookEventHandlerImpl {
+	webhookSecretValidator git.WebhookSecretValidator, webhookEventDataConfig pipeline.WebhookEventDataConfig,
+	ciPipelineEventPublishService out.CIPipelineEventPublishService) *WebhookEventHandlerImpl {
 	return &WebhookEventHandlerImpl{
-		logger:                 logger,
-		gitHostConfig:          gitHostConfig,
-		eventClient:            eventClient,
-		webhookSecretValidator: webhookSecretValidator,
-		webhookEventDataConfig: webhookEventDataConfig,
+		logger:                        logger,
+		gitHostConfig:                 gitHostConfig,
+		eventClient:                   eventClient,
+		webhookSecretValidator:        webhookSecretValidator,
+		webhookEventDataConfig:        webhookEventDataConfig,
+		ciPipelineEventPublishService: ciPipelineEventPublishService,
 	}
 }
 
@@ -121,7 +124,7 @@ func (impl WebhookEventHandlerImpl) OnWebhookEvent(w http.ResponseWriter, r *htt
 	}
 
 	// write event
-	err = impl.eventClient.WriteNatsEvent(pubsub.WEBHOOK_EVENT_TOPIC, webhookEvent)
+	err = impl.ciPipelineEventPublishService.PublishGitWebhookEvent(gitHostId, eventType, string(requestBodyBytes))
 	if err != nil {
 		impl.logger.Errorw("Error while handling webhook in git-sensor", "err", err)
 		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)

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

@@ -27,7 +27,6 @@ import (
 	bean3 "github.com/devtron-labs/devtron/pkg/deployment/trigger/devtronApps/bean"
 	"github.com/devtron-labs/devtron/pkg/eventProcessor/out"
 	bean4 "github.com/devtron-labs/devtron/pkg/eventProcessor/out/bean"
-	"github.com/devtron-labs/devtron/pkg/workflow/dag"
 	"net/http"
 	"strconv"
 
@@ -65,7 +64,6 @@ type PipelineTriggerRestHandlerImpl struct {
 	enforcer                    casbin.Enforcer
 	teamService                 team.TeamService
 	logger                      *zap.SugaredLogger
-	workflowDagExecutor         dag.WorkflowDagExecutor
 	enforcerUtil                rbac.EnforcerUtil
 	deploymentGroupService      deploymentGroup.DeploymentGroupService
 	argoUserService             argo.ArgoUserService
@@ -77,7 +75,7 @@ type PipelineTriggerRestHandlerImpl struct {
 
 func NewPipelineRestHandler(appService app.AppService, userAuthService user.UserService, validator *validator.Validate,
 	enforcer casbin.Enforcer, teamService team.TeamService, logger *zap.SugaredLogger, enforcerUtil rbac.EnforcerUtil,
-	workflowDagExecutor dag.WorkflowDagExecutor, deploymentGroupService deploymentGroup.DeploymentGroupService,
+	deploymentGroupService deploymentGroup.DeploymentGroupService,
 	argoUserService argo.ArgoUserService, deploymentConfigService pipeline.DeploymentConfigService,
 	deployedAppService deployedApp.DeployedAppService,
 	cdTriggerService devtronApps.TriggerService,
@@ -89,7 +87,6 @@ func NewPipelineRestHandler(appService app.AppService, userAuthService user.User
 		enforcer:                    enforcer,
 		teamService:                 teamService,
 		logger:                      logger,
-		workflowDagExecutor:         workflowDagExecutor,
 		enforcerUtil:                enforcerUtil,
 		deploymentGroupService:      deploymentGroupService,
 		argoUserService:             argoUserService,

api/router/router.go

@@ -19,7 +19,6 @@ package router
 
 import (
 	"encoding/json"
-	pubsub2 "github.com/devtron-labs/common-lib/pubsub-lib"
 	"github.com/devtron-labs/devtron/api/apiToken"
 	"github.com/devtron-labs/devtron/api/appStore"
 	"github.com/devtron-labs/devtron/api/appStore/chartGroup"
@@ -39,7 +38,6 @@ import (
 	"github.com/devtron-labs/devtron/api/module"
 	"github.com/devtron-labs/devtron/api/restHandler/common"
 	"github.com/devtron-labs/devtron/api/router/app"
-	"github.com/devtron-labs/devtron/api/router/pubsub"
 	"github.com/devtron-labs/devtron/api/server"
 	"github.com/devtron-labs/devtron/api/team"
 	terminal2 "github.com/devtron-labs/devtron/api/terminal"
@@ -69,10 +67,7 @@ type MuxRouter struct {
 	DockerRegRouter                    DockerRegRouter
 	NotificationRouter                 NotificationRouter
 	TeamRouter                         team.TeamRouter
-	pubsubClient                       *pubsub2.PubSubClientServiceImpl
 	UserRouter                         user.UserRouter
-	gitWebhookHandler                  pubsub.GitWebhookHandler
-	appUpdateHandler                   pubsub.ApplicationStatusHandler
 	ChartRefRouter                     ChartRefRouter
 	ConfigMapRouter                    ConfigMapRouter
 	AppStoreRouter                     appStore.AppStoreRouter
@@ -129,9 +124,7 @@ func NewMuxRouter(logger *zap.SugaredLogger,
 	DockerRegRouter DockerRegRouter,
 	NotificationRouter NotificationRouter,
 	TeamRouter team.TeamRouter,
-	gitWebhookHandler pubsub.GitWebhookHandler,
-	appUpdateHandler pubsub.ApplicationStatusHandler,
-	pubsubClient *pubsub2.PubSubClientServiceImpl, UserRouter user.UserRouter,
+	UserRouter user.UserRouter,
 	ChartRefRouter ChartRefRouter, ConfigMapRouter ConfigMapRouter, AppStoreRouter appStore.AppStoreRouter, chartRepositoryRouter chartRepo.ChartRepositoryRouter,
 	ReleaseMetricsRouter ReleaseMetricsRouter, deploymentGroupRouter DeploymentGroupRouter, batchOperationRouter BatchOperationRouter,
 	chartGroupRouter chartGroup.ChartGroupRouter, imageScanRouter ImageScanRouter,
@@ -164,9 +157,6 @@ func NewMuxRouter(logger *zap.SugaredLogger,
 		NotificationRouter:                 NotificationRouter,
 		TeamRouter:                         TeamRouter,
 		logger:                             logger,
-		gitWebhookHandler:                  gitWebhookHandler,
-		appUpdateHandler:                   appUpdateHandler,
-		pubsubClient:                       pubsubClient,
 		UserRouter:                         UserRouter,
 		ChartRefRouter:                     ChartRefRouter,
 		ConfigMapRouter:                    ConfigMapRouter,