fix: allow only enabled dest in backendSubscriber, fix klaviyo bulk (#…

…5309)
rudderlabs · Nov 20, 2024 · 8808245 · 8808245
1 parent 1c78429
commit 8808245
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 5 deletions.
diff --git a/router/batchrouter/asyncdestinationmanager/klaviyobulkupload/apiService.go b/router/batchrouter/asyncdestinationmanager/klaviyobulkupload/apiService.go
@@ -116,16 +116,20 @@ func (k *KlaviyoAPIServiceImpl) GetUploadErrors(importId string) (*UploadStatusR
 	return &importErrorResp, importErrorRespErr
 }
 
-func NewKlaviyoAPIService(destination *backendconfig.DestinationT, logger logger.Logger, statsFactory stats.Stats) KlaviyoAPIService {
+func NewKlaviyoAPIService(destination *backendconfig.DestinationT, logger logger.Logger, statsFactory stats.Stats) (KlaviyoAPIService, error) {
+	privateApiKey, ok := destination.Config["privateApiKey"].(string)
+	if !ok {
+		return nil, fmt.Errorf("privateApiKey not found or not a string")
+	}
 	return &KlaviyoAPIServiceImpl{
 		client:        http.DefaultClient,
-		PrivateAPIKey: destination.Config["privateApiKey"].(string),
+		PrivateAPIKey: privateApiKey,
 		logger:        logger,
 		statsFactory:  statsFactory,
 		statLabels: stats.Tags{
 			"module":   "batch_router",
 			"destType": destination.Name,
 			"destID":   destination.ID,
 		},
-	}
+	}, nil
 }
diff --git a/router/batchrouter/asyncdestinationmanager/klaviyobulkupload/klaviyobulkupload.go b/router/batchrouter/asyncdestinationmanager/klaviyobulkupload/klaviyobulkupload.go
@@ -58,12 +58,16 @@ func createFinalPayload(combinedProfiles []Profile, listId string) Payload {
 
 func NewManager(logger logger.Logger, StatsFactory stats.Stats, destination *backendconfig.DestinationT) (*KlaviyoBulkUploader, error) {
 	klaviyoLogger := logger.Child("KlaviyoBulkUpload").Child("KlaviyoBulkUploader")
+	apiService, err := NewKlaviyoAPIService(destination, klaviyoLogger, StatsFactory)
+	if err != nil {
+		return nil, err
+	}
 	return &KlaviyoBulkUploader{
 		DestName:          destination.DestinationDefinition.Name,
 		DestinationConfig: destination.Config,
 		Logger:            klaviyoLogger,
 		StatsFactory:      StatsFactory,
-		KlaviyoAPIService: NewKlaviyoAPIService(destination, klaviyoLogger, StatsFactory),
+		KlaviyoAPIService: apiService,
 	}, nil
 }
 

diff --git a/router/batchrouter/handle_lifecycle.go b/router/batchrouter/handle_lifecycle.go
@@ -388,7 +388,7 @@ func (brt *Handle) backendConfigSubscriber() {
 			for _, source := range wConfig.Sources {
 				if len(source.Destinations) > 0 {
 					for _, destination := range source.Destinations {
-						if destination.DestinationDefinition.Name == brt.destType {
+						if destination.DestinationDefinition.Name == brt.destType && destination.Enabled {
 							if _, ok := destinationsMap[destination.ID]; !ok {
 								destinationsMap[destination.ID] = &routerutils.DestinationWithSources{Destination: destination, Sources: []backendconfig.SourceT{}}
 								uploadIntervalMap[destination.ID] = brt.uploadInterval(destination.Config)