Merge pull request #6526 from devtron-labs/notifier-refac

feat: Enable selection of all CI pipelines at once when the Environment filter is applied in Notifications

Author: prkhrkat

api/restHandler/NotificationRestHandler.go

@@ -51,7 +51,6 @@ const (
 )
 
 type NotificationRestHandler interface {
-	SaveNotificationSettings(w http.ResponseWriter, r *http.Request)
 	SaveNotificationSettingsV2(w http.ResponseWriter, r *http.Request)
 	UpdateNotificationSettings(w http.ResponseWriter, r *http.Request)
 	SaveNotificationChannelConfig(w http.ResponseWriter, r *http.Request)
@@ -118,66 +117,6 @@ func NewNotificationRestHandlerImpl(dockerRegistryConfig pipeline.DockerRegistry
 	}
 }
 
-// SaveNotificationSettings will be deprecated in future
-func (impl NotificationRestHandlerImpl) SaveNotificationSettings(w http.ResponseWriter, r *http.Request) {
-	userId, err := impl.userAuthService.GetLoggedInUser(r)
-	if userId == 0 || err != nil {
-		common.WriteJsonResp(w, err, "Unauthorized User", http.StatusUnauthorized)
-		return
-	}
-	var notificationSetting beans.NotificationRequest
-	err = json.NewDecoder(r.Body).Decode(&notificationSetting)
-	if err != nil {
-		impl.logger.Errorw("request err, SaveNotificationSettings", "err", err, "payload", notificationSetting)
-		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
-		return
-	}
-	impl.logger.Infow("request payload, SaveNotificationSettings", "err", err, "payload", notificationSetting)
-	err = impl.validator.Struct(notificationSetting)
-	if err != nil {
-		impl.logger.Errorw("validation err, SaveNotificationSettings", "err", err, "payload", notificationSetting)
-		common.WriteJsonResp(w, err, nil, http.StatusBadRequest)
-		return
-	}
-
-	//RBAC
-	token := r.Header.Get("token")
-	if isSuperAdmin := impl.enforcer.Enforce(token, casbin.ResourceGlobal, casbin.ActionGet, "*"); !isSuperAdmin {
-		common.WriteJsonResp(w, err, nil, http.StatusForbidden)
-		return
-	}
-	//RBAC
-
-	providers := notificationSetting.Providers
-
-	if len(providers) != 0 {
-		for _, provider := range providers {
-			if provider.Destination == util.SMTP || provider.Destination == util.SES {
-				if provider.Recipient == "" {
-					userEmail, err := impl.userAuthService.GetEmailById(int32(provider.ConfigId))
-					if err != nil {
-						impl.logger.Errorw("service err, SaveNotificationSettings", "err", err, "payload", notificationSetting)
-						common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
-						return
-					}
-					provider.Recipient = userEmail
-				}
-				// get default configID for SES and SMTP
-				provider.ConfigId = notificationSetting.SesConfigId
-			}
-		}
-	}
-
-	res, err := impl.notificationService.CreateOrUpdateNotificationSettings(&notificationSetting, userId)
-	if err != nil {
-		impl.logger.Errorw("service err, SaveNotificationSettings", "err", err, "payload", notificationSetting)
-		common.WriteJsonResp(w, err, nil, http.StatusInternalServerError)
-		return
-	}
-	w.Header().Set("Content-Type", "application/json")
-	common.WriteJsonResp(w, nil, res, http.StatusOK)
-}
-
 func (impl NotificationRestHandlerImpl) SaveNotificationSettingsV2(w http.ResponseWriter, r *http.Request) {
 	userId, err := impl.userAuthService.GetLoggedInUser(r)
 	if userId == 0 || err != nil {

api/router/NotificationRouter.go

@@ -32,9 +32,9 @@ func NewNotificationRouterImpl(notificationRestHandler restHandler.NotificationR
 	return &NotificationRouterImpl{notificationRestHandler: notificationRestHandler}
 }
 func (impl NotificationRouterImpl) InitNotificationRegRouter(configRouter *mux.Router) {
-	// to maintain backward compatibility, will be deprecated in future
+	// do not remove it, this path is used in devtcl cli
 	configRouter.Path("").
-		HandlerFunc(impl.notificationRestHandler.SaveNotificationSettings).
+		HandlerFunc(impl.notificationRestHandler.SaveNotificationSettingsV2).
 		Methods("POST")
 	// new router to save notification settings
 	configRouter.Path("/v2").

client/events/EventBuilder.go

@@ -206,6 +206,26 @@ func (impl *EventSimpleFactoryImpl) BuildExtraCIData(event Event, material *buil
 		payload.TriggeredBy = user.EmailId
 		event.Payload = payload
 	}
+
+	// fetching all the envs which are directly or indirectly linked with the ci pipeline
+	if event.PipelineId > 0 {
+		// Get the pipeline to check if it's external
+		ciPipeline, err := impl.ciPipelineRepository.FindById(event.PipelineId)
+		if err != nil {
+			impl.logger.Errorw("error in getting ci pipeline", "pipelineId", event.PipelineId, "err", err)
+		} else {
+			envs, err := impl.envRepository.FindEnvLinkedWithCiPipelines(ciPipeline.IsExternal, []int{event.PipelineId})
+			if err != nil {
+				impl.logger.Errorw("error in finding environments linked with ci pipeline", "pipelineId", event.PipelineId, "err", err)
+			} else {
+				event.EnvIdsForCiPipeline = make([]int, 0, len(envs))
+				for _, env := range envs {
+					event.EnvIdsForCiPipeline = append(event.EnvIdsForCiPipeline, env.Id)
+				}
+			}
+		}
+	}
+
 	return event
 }
 

client/events/EventClient.go

@@ -18,6 +18,7 @@ package client
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -38,6 +39,7 @@ import (
 type EventClientConfig struct {
 	DestinationURL     string             `env:"EVENT_URL" envDefault:"http://localhost:3000/notify" description:"Notifier service url"`
 	NotificationMedium NotificationMedium `env:"NOTIFICATION_MEDIUM" envDefault:"rest" description:"notification medium"`
+	EnableNotifierV2   bool               `env:"ENABLE_NOTIFIER_V2" envDefault:"false" description:"enable notifier v2"`
 }
 type NotificationMedium string
 
@@ -58,24 +60,25 @@ type EventClient interface {
 }
 
 type Event struct {
-	EventTypeId        int               `json:"eventTypeId"`
-	EventName          string            `json:"eventName"`
-	PipelineId         int               `json:"pipelineId"`
-	PipelineType       string            `json:"pipelineType"`
-	CorrelationId      string            `json:"correlationId"`
-	Payload            *Payload          `json:"payload"`
-	EventTime          string            `json:"eventTime"`
-	TeamId             int               `json:"teamId"`
-	AppId              int               `json:"appId"`
-	EnvId              int               `json:"envId"`
-	IsProdEnv          bool              `json:"isProdEnv"`
-	ClusterId          int               `json:"clusterId"`
-	CdWorkflowType     bean.WorkflowType `json:"cdWorkflowType,omitempty"`
-	CdWorkflowRunnerId int               `json:"cdWorkflowRunnerId"`
-	CiWorkflowRunnerId int               `json:"ciWorkflowRunnerId"`
-	CiArtifactId       int               `json:"ciArtifactId"`
-	BaseUrl            string            `json:"baseUrl"`
-	UserId             int               `json:"-"`
+	EventTypeId         int               `json:"eventTypeId"`
+	EventName           string            `json:"eventName"`
+	PipelineId          int               `json:"pipelineId"`
+	PipelineType        string            `json:"pipelineType"`
+	CorrelationId       string            `json:"correlationId"`
+	Payload             *Payload          `json:"payload"`
+	EventTime           string            `json:"eventTime"`
+	TeamId              int               `json:"teamId"`
+	AppId               int               `json:"appId"`
+	EnvId               int               `json:"envId"`
+	IsProdEnv           bool              `json:"isProdEnv"`
+	ClusterId           int               `json:"clusterId"`
+	CdWorkflowType      bean.WorkflowType `json:"cdWorkflowType,omitempty"`
+	CdWorkflowRunnerId  int               `json:"cdWorkflowRunnerId"`
+	CiWorkflowRunnerId  int               `json:"ciWorkflowRunnerId"`
+	CiArtifactId        int               `json:"ciArtifactId"`
+	EnvIdsForCiPipeline []int             `json:"envIdsForCiPipeline"`
+	BaseUrl             string            `json:"baseUrl"`
+	UserId              int               `json:"-"`
 }
 
 type Payload struct {
@@ -95,22 +98,25 @@ type Payload struct {
 }
 
 type EventRESTClientImpl struct {
-	logger               *zap.SugaredLogger
-	client               *http.Client
-	config               *EventClientConfig
-	pubsubClient         *pubsub.PubSubClientServiceImpl
-	ciPipelineRepository pipelineConfig.CiPipelineRepository
-	pipelineRepository   pipelineConfig.PipelineRepository
-	attributesRepository repository.AttributesRepository
-	moduleService        module.ModuleService
+	logger                         *zap.SugaredLogger
+	client                         *http.Client
+	config                         *EventClientConfig
+	pubsubClient                   *pubsub.PubSubClientServiceImpl
+	ciPipelineRepository           pipelineConfig.CiPipelineRepository
+	pipelineRepository             pipelineConfig.PipelineRepository
+	attributesRepository           repository.AttributesRepository
+	moduleService                  module.ModuleService
+	notificationSettingsRepository repository.NotificationSettingsRepository
 }
 
 func NewEventRESTClientImpl(logger *zap.SugaredLogger, client *http.Client, config *EventClientConfig, pubsubClient *pubsub.PubSubClientServiceImpl,
 	ciPipelineRepository pipelineConfig.CiPipelineRepository, pipelineRepository pipelineConfig.PipelineRepository,
-	attributesRepository repository.AttributesRepository, moduleService module.ModuleService) *EventRESTClientImpl {
+	attributesRepository repository.AttributesRepository, moduleService module.ModuleService,
+	notificationSettingsRepository repository.NotificationSettingsRepository) *EventRESTClientImpl {
 	return &EventRESTClientImpl{logger: logger, client: client, config: config, pubsubClient: pubsubClient,
 		ciPipelineRepository: ciPipelineRepository, pipelineRepository: pipelineRepository,
-		attributesRepository: attributesRepository, moduleService: moduleService}
+		attributesRepository: attributesRepository, moduleService: moduleService,
+		notificationSettingsRepository: notificationSettingsRepository}
 }
 
 func (impl *EventRESTClientImpl) buildFinalPayload(event Event, cdPipeline *pipelineConfig.Pipeline, ciPipeline *pipelineConfig.CiPipeline) *Payload {
@@ -235,34 +241,131 @@ func (impl *EventRESTClientImpl) sendEventsOnNats(body []byte) error {
 // do not call this method if notification module is not installed
 func (impl *EventRESTClientImpl) sendEvent(event Event) (bool, error) {
 	impl.logger.Debugw("event before send", "event", event)
-	body, err := json.Marshal(event)
+
+	// Step 1: Create payload and destination URL based on config
+	bodyBytes, destinationUrl, err := impl.createPayloadAndDestination(event)
 	if err != nil {
-		impl.logger.Errorw("error while marshaling event request ", "err", err)
 		return false, err
 	}
+
+	// Step 2: Send via appropriate medium (NATS or REST)
+	return impl.deliverEvent(bodyBytes, destinationUrl)
+}
+
+func (impl *EventRESTClientImpl) createPayloadAndDestination(event Event) ([]byte, string, error) {
+	if impl.config.EnableNotifierV2 {
+		return impl.createV2PayloadAndDestination(event)
+	}
+	return impl.createDefaultPayloadAndDestination(event)
+}
+
+func (impl *EventRESTClientImpl) createV2PayloadAndDestination(event Event) ([]byte, string, error) {
+	destinationUrl := impl.config.DestinationURL + "/v2"
+
+	// Fetch notification settings
+	req := repository.GetRulesRequest{
+		TeamId:              event.TeamId,
+		EnvId:               event.EnvId,
+		AppId:               event.AppId,
+		PipelineId:          event.PipelineId,
+		PipelineType:        event.PipelineType,
+		IsProdEnv:           &event.IsProdEnv,
+		ClusterId:           event.ClusterId,
+		EnvIdsForCiPipeline: event.EnvIdsForCiPipeline,
+	}
+	notificationSettings, err := impl.notificationSettingsRepository.FindNotificationSettingsWithRules(
+		context.Background(), event.EventTypeId, req,
+	)
+	if err != nil {
+		impl.logger.Errorw("error while fetching notification settings", "err", err)
+		return nil, "", err
+	}
+
+	// Process notification settings into beans
+	notificationSettingsBean, err := impl.processNotificationSettings(notificationSettings)
+	if err != nil {
+		return nil, "", err
+	}
+
+	// Create combined payload
+	combinedPayload := map[string]interface{}{
+		"event":                event,
+		"notificationSettings": notificationSettingsBean,
+	}
+
+	bodyBytes, err := json.Marshal(combinedPayload)
+	if err != nil {
+		impl.logger.Errorw("error while marshaling combined event request", "err", err)
+		return nil, "", err
+	}
+
+	return bodyBytes, destinationUrl, nil
+}
+
+func (impl *EventRESTClientImpl) createDefaultPayloadAndDestination(event Event) ([]byte, string, error) {
+	bodyBytes, err := json.Marshal(event)
+	if err != nil {
+		impl.logger.Errorw("error while marshaling event request", "err", err)
+		return nil, "", err
+	}
+	return bodyBytes, impl.config.DestinationURL, nil
+}
+
+func (impl *EventRESTClientImpl) processNotificationSettings(notificationSettings []repository.NotificationSettings) ([]*repository.NotificationSettingsBean, error) {
+	notificationSettingsBean := make([]*repository.NotificationSettingsBean, 0)
+	for _, item := range notificationSettings {
+		config := make([]repository.ConfigEntry, 0)
+		if item.Config != "" {
+			if err := json.Unmarshal([]byte(item.Config), &config); err != nil {
+				impl.logger.Errorw("error while unmarshaling config", "err", err)
+				return nil, err
+			}
+		}
+		notificationSettingsBean = append(notificationSettingsBean, &repository.NotificationSettingsBean{
+			Id:           item.Id,
+			TeamId:       item.TeamId,
+			AppId:        item.AppId,
+			EnvId:        item.EnvId,
+			PipelineId:   item.PipelineId,
+			PipelineType: item.PipelineType,
+			EventTypeId:  item.EventTypeId,
+			Config:       config,
+			ViewId:       item.ViewId,
+		})
+	}
+	return notificationSettingsBean, nil
+}
+
+func (impl *EventRESTClientImpl) deliverEvent(bodyBytes []byte, destinationUrl string) (bool, error) {
 	if impl.config.NotificationMedium == PUB_SUB {
-		err = impl.sendEventsOnNats(body)
-		if err != nil {
-			impl.logger.Errorw("error while publishing event  ", "err", err)
+		if err := impl.sendEventsOnNats(bodyBytes); err != nil {
+			impl.logger.Errorw("error while publishing event", "err", err)
 			return false, err
 		}
 		return true, nil
 	}
-	var reqBody = []byte(body)
-	req, err := http.NewRequest(http.MethodPost, impl.config.DestinationURL, bytes.NewBuffer(reqBody))
+
+	req, err := http.NewRequest(http.MethodPost, destinationUrl, bytes.NewBuffer(bodyBytes))
 	if err != nil {
-		impl.logger.Errorw("error while writing event", "err", err)
+		impl.logger.Errorw("error while creating HTTP request", "err", err)
 		return false, err
 	}
 	req.Header.Set("Content-Type", "application/json")
+
 	resp, err := impl.client.Do(req)
 	if err != nil {
-		impl.logger.Errorw("error while UpdateJiraTransition request ", "err", err)
+		impl.logger.Errorw("error while sending HTTP request", "err", err)
 		return false, err
 	}
 	defer resp.Body.Close()
-	impl.logger.Debugw("event completed", "event resp", resp)
-	return true, err
+
+	if resp.StatusCode >= 300 {
+		impl.logger.Errorw("unexpected response from notifier", "status", resp.StatusCode)
+		return false, fmt.Errorf("unexpected response code: %d", resp.StatusCode)
+	}
+
+	impl.logger.Debugw("event successfully delivered", "status", resp.StatusCode)
+	return true, nil
 }
 
 func (impl *EventRESTClientImpl) WriteNatsEvent(topic string, payload interface{}) error {

env_gen.json

@@ -174,6 +174,7 @@
  | ECR_REPO_NAME_PREFIX | string |test/ | Prefix for ECR repo to be created in does not exist |  | false |
  | ENABLE_ASYNC_ARGO_CD_INSTALL_DEVTRON_CHART | bool |false | To enable async installation of gitops application |  | false |
  | ENABLE_ASYNC_INSTALL_DEVTRON_CHART | bool |false | To enable async installation of no-gitops application |  | false |
+ | ENABLE_NOTIFIER_V2 | bool |false | enable notifier v2 |  | false |
  | EPHEMERAL_SERVER_VERSION_REGEX | string |v[1-9]\.\b(2[3-9]\|[3-9][0-9])\b.* | ephemeral containers support version regex that is compared with k8sServerVersion |  | false |
  | EVENT_URL | string |http://localhost:3000/notify | Notifier service url |  | false |
  | EXECUTE_WIRE_NIL_CHECKER | bool |false | checks for any nil pointer in wire.go |  | false |