feat: notifier behind nats (#5185)

* feat: notifier behind nats

* chore: wire

* chore: run test case again after updating common lib by taking merge

* fix: update common lib and wire

* fix : add argo cd assets

* chore: update branch

* chore: merge main

* chore: update common lib

Author: komalreddy3

client/events/EventClient.go

@@ -37,8 +37,12 @@ import (
 )
 
 type EventClientConfig struct {
-	DestinationURL string `env:"EVENT_URL" envDefault:"http://localhost:3000/notify"`
+	DestinationURL     string             `env:"EVENT_URL" envDefault:"http://localhost:3000/notify"`
+	NotificationMedium NotificationMedium `env:"NOTIFICATION_MEDIUM" envDefault:"rest"`
 }
+type NotificationMedium string
+
+const PUB_SUB NotificationMedium = "nats"
 
 func GetEventClientConfig() (*EventClientConfig, error) {
 	cfg := &EventClientConfig{}
@@ -238,6 +242,16 @@ func (impl *EventRESTClientImpl) WriteNotificationEvent(event Event) (bool, erro
 	}
 	return true, err
 }
+func (impl *EventRESTClientImpl) sendEventsOnNats(body []byte) error {
+
+	err := impl.pubsubClient.Publish(pubsub.NOTIFICATION_EVENT_TOPIC, string(body))
+	if err != nil {
+		impl.logger.Errorw("err while publishing msg for testing topic", "msg", body, "err", err)
+		return err
+	}
+	return nil
+
+}
 
 // do not call this method if notification module is not installed
 func (impl *EventRESTClientImpl) sendEvent(event Event) (bool, error) {
@@ -247,6 +261,14 @@ func (impl *EventRESTClientImpl) sendEvent(event Event) (bool, error) {
 		impl.logger.Errorw("error while marshaling event request ", "err", err)
 		return false, err
 	}
+	if impl.config.NotificationMedium == PUB_SUB {
+		err = impl.sendEventsOnNats(body)
+		if 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))
 	if err != nil {

client/events/event_test.go

@@ -0,0 +1,37 @@
+package client
+
+import (
+	"fmt"
+	pubsub_lib "github.com/devtron-labs/common-lib/pubsub-lib"
+	"github.com/devtron-labs/devtron/internal/sql/repository"
+	"github.com/devtron-labs/devtron/internal/sql/repository/pipelineConfig"
+	"github.com/devtron-labs/devtron/internal/util"
+	"github.com/devtron-labs/devtron/pkg/sql"
+	"testing"
+)
+
+func TestSendEventsOnNats(t *testing.T) {
+	logger, err := util.NewSugardLogger()
+	//nats, err := pubsub_lib.NewNatsClient(logger)
+	//mockPubsubClient := NewPubSubClientServiceImpl(logger)
+	mockPubsubClient, err := pubsub_lib.NewPubSubClientServiceImpl(logger)
+	client := util.NewHttpClient()
+	config := sql.Config{}
+	db, err := sql.NewDbConnection(&config, logger)
+	trans := sql.NewTransactionUtilImpl(db)
+	impl := &EventRESTClientImpl{
+		logger:               logger,
+		pubsubClient:         mockPubsubClient,
+		client:               client,
+		config:               &EventClientConfig{DestinationURL: "localhost:3000/notify", NotificationMedium: PUB_SUB},
+		ciPipelineRepository: pipelineConfig.NewCiPipelineRepositoryImpl(db, logger, trans),
+		pipelineRepository:   pipelineConfig.NewPipelineRepositoryImpl(db, logger),
+		attributesRepository: repository.NewAttributesRepositoryImpl(db),
+	}
+	//xpectedTopic := "NOTIFICATION_EVENT_TOPIC"
+	expectedMsg := "'{\"eventTypeId\":1,\"pipelineId\":123,\"payload\":{\"key\":\"value\"},\"eventTime\":\"2024-05-09T12:00:00Z\",\"appId\":456,\"envId\":789,\"teamId\":101}'"
+
+	err = impl.sendEventsOnNats([]byte(expectedMsg))
+	fmt.Println(err)
+
+}

env_gen.md

@@ -177,6 +177,7 @@
  | NATS_MSG_MAX_AGE | 86400 |  | 
  | NATS_MSG_PROCESSING_BATCH_SIZE | 1 |  | 
  | NATS_SERVER_HOST | nats://devtron-nats.devtroncd:4222 |  | 
+ | NOTIFICATION_MEDIUM | rest |  | 
  | ORCH_HOST | http://devtroncd-orchestrator-service-prod.devtroncd/webhook/msg/nats |  | 
  | ORCH_TOKEN |  |  | 
  | OTEL_COLLECTOR_URL |  |  | 

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/deckarep/golang-set v1.8.0
 	github.com/devtron-labs/authenticator v0.4.35-0.20240405091826-a91813c53470
-	github.com/devtron-labs/common-lib v0.0.18-0.20240524141543-f4ed1281e694
+	github.com/devtron-labs/common-lib v0.0.19-0.20240607054959-82c79c23b046
 	github.com/devtron-labs/protos v0.0.3-0.20240326053929-48e42d9d4534
 	github.com/evanphx/json-patch v5.6.0+incompatible
 	github.com/gammazero/workerpool v1.1.3

go.sum

@@ -207,8 +207,8 @@ github.com/denisenkom/go-mssqldb v0.0.0-20190707035753-2be1aa521ff4 h1:YcpmyvADG
 github.com/denisenkom/go-mssqldb v0.0.0-20190707035753-2be1aa521ff4/go.mod h1:zAg7JM8CkOJ43xKXIj7eRO9kmWm/TW578qo+oDO6tuM=
 github.com/devtron-labs/authenticator v0.4.35-0.20240405091826-a91813c53470 h1:AUTYcDnL6w6Ux+264VldYaOUQAP6pDZ5Tq8wCKJyiEg=
 github.com/devtron-labs/authenticator v0.4.35-0.20240405091826-a91813c53470/go.mod h1:JQxTCMmQisrpjzETJr0tzVadV+wW23rHEZAY7JVyK3s=
-github.com/devtron-labs/common-lib v0.0.18-0.20240524141543-f4ed1281e694 h1:lUcMarRvAKzsLpmuYwFgOsKLJQpHsJuvbKG+we/dI58=
-github.com/devtron-labs/common-lib v0.0.18-0.20240524141543-f4ed1281e694/go.mod h1:deAcJ5IjUjM6ozZQLJEgPWDUA0mKa632LBsKx8uM9TE=
+github.com/devtron-labs/common-lib v0.0.19-0.20240607054959-82c79c23b046 h1:hOyqkgILg+eDttLV6X7OAAo9PKEHzInUmBTVy/EY/iI=
+github.com/devtron-labs/common-lib v0.0.19-0.20240607054959-82c79c23b046/go.mod h1:deAcJ5IjUjM6ozZQLJEgPWDUA0mKa632LBsKx8uM9TE=
 github.com/devtron-labs/protos v0.0.3-0.20240326053929-48e42d9d4534 h1:TElPRU69QedW7DIQiiQxtjwSQ6cK0fCTAMGvSLhP0ac=
 github.com/devtron-labs/protos v0.0.3-0.20240326053929-48e42d9d4534/go.mod h1:ypUknVph8Ph4dxSlrFoouf7wLedQxHku2LQwgRrdgS4=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=