feat: add support for hash to bingads offline conversions

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/bingads_test.go

@@ -70,11 +70,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{
-				URL:    "http://localhost/upload1",
-				client: &http.Client{},
-			}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 			bingAdsService.EXPECT().GetBulkUploadUrl().Return(&mockbingads.GetBulkUploadUrlResponse{
 				UploadUrl: "http://localhost/upload1",
 				RequestId: misc.FastUUID().URN(),
@@ -138,8 +134,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 			errorMsg := "Error in getting bulk upload url"
 			bingAdsService.EXPECT().GetBulkUploadUrl().Return(nil, errors.New(errorMsg))
 			bingAdsService.EXPECT().GetBulkUploadUrl().Return(nil, errors.New(errorMsg))
@@ -181,8 +176,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			ClientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &ClientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 			errMsg := "unable to get bulk upload url, check your credentials"
 			bingAdsService.EXPECT().GetBulkUploadUrl().Return(nil, errors.New(errMsg))
 
@@ -227,8 +221,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 			bingAdsService.EXPECT().GetBulkUploadUrl().Return(&mockbingads.GetBulkUploadUrlResponse{
 				UploadUrl: "http://localhost/upload1",
 				RequestId: misc.FastUUID().URN(),
@@ -285,8 +278,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 
 			bingAdsService.EXPECT().GetBulkUploadStatus("dummyRequestId123").Return(&mockbingads.GetBulkUploadStatusResponse{
 				PercentComplete: int64(100),
@@ -308,8 +300,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 
 			bingAdsService.EXPECT().GetBulkUploadStatus("dummyRequestId123").Return(nil, fmt.Errorf("failed to get bulk upload status:"))
 			pollInput := common.AsyncPoll{
@@ -327,8 +318,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 
 			bingAdsService.EXPECT().GetBulkUploadStatus("dummyRequestId123").Return(&mockbingads.GetBulkUploadStatusResponse{
 				PercentComplete: int64(100),
@@ -356,8 +346,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 
 			bingAdsService.EXPECT().GetBulkUploadStatus("dummyRequestId123").Return(&mockbingads.GetBulkUploadStatusResponse{
 				PercentComplete: int64(0),
@@ -383,8 +372,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 
 			bingAdsService.EXPECT().GetBulkUploadStatus("dummyRequestId123").Return(&mockbingads.GetBulkUploadStatusResponse{
 				PercentComplete: int64(0),
@@ -410,8 +398,7 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 			initBingads()
 			ctrl := gomock.NewController(GinkgoT())
 			bingAdsService := mockbulkservice.NewMockBulkServiceI(ctrl)
-			clientI := Client{}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 
 			bingAdsService.EXPECT().GetBulkUploadStatus("dummyRequestId456").Return(&mockbingads.GetBulkUploadStatusResponse{
 				PercentComplete: int64(100),
@@ -454,10 +441,8 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 				http.ServeFile(w, r, errorsTemplateFilePath)
 			}))
 			defer ts.Close()
-			client := ts.Client()
 			modifiedURL := ts.URL // Use the test server URL
-			clientI := Client{client: client, URL: modifiedURL}
-			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, &clientI)
+			bulkUploader := NewBingAdsBulkUploader(logger.NOP, stats.NOP, "BING_ADS", bingAdsService, true)
 
 			UploadStatsInput := common.GetUploadStatsInput{
 				FailedJobParameters: modifiedURL,
@@ -531,11 +516,15 @@ var _ = Describe("Bing ads Offline Conversions", func() {
 		})
 		It("Transform() Test -> successful ", func() {
 			job := &jobsdb.JobT{
-				EventPayload: []byte("{\"type\": \"record\", \"action\": \"insert\", \"fields\": {\"conversionName\": \"Test-Integration\", \"conversionTime\": \"5/22/2023 6:27:54 AM\", \"conversionValue\": \"100\", \"microsoftClickId\": \"click_id\", \"conversionCurrencyCode\": \"USD\"}}"),
+				EventPayload: []byte("{\n  \"type\": \"record\",\n  \"action\": \"insert\",\n  \"fields\": {\n    \"conversionName\": \"Test-Integration\",\n    \"conversionTime\": \"5/22/2023 6:27:54 AM\",\n    \"conversionValue\": \"100\",\n    \"microsoftClickId\": \"click_id\",\n    \"conversionCurrencyCode\": \"USD\",\n    \"email\":\"[email protected]\",\n    \"phone\":\"+911234567890\"\n  }\n}"),
 			}
-			uploader := &BingAdsBulkUploader{}
+			uploader := &BingAdsBulkUploader{
+				isHashRequired: true,
+			}
+			expectedResp := `{"message":{"fields":{"conversionCurrencyCode":"USD","conversionName":"Test-Integration","conversionTime":"5/22/2023 6:27:54 AM","conversionValue":"100","email":"28a4da98f8812110001ab8ffacde3b38b4725a9e3570c39299fbf2d12c5aa70e","microsoftClickId":"click_id","phone":"8c229df83de8ab269e90918846e326c4008c86481393223d17a30ff5a407b08e"},"action":"insert"},"metadata":{"jobId":0}}`
 			// Execute
-			_, err := uploader.Transform(job)
+			resp, err := uploader.Transform(job)
+			Expect(resp).To(Equal(expectedResp))
 			Expect(err).To(BeNil())
 		})
 

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/bulk_uploader.go

@@ -19,15 +19,15 @@
 	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/common"
 )
 
-func NewBingAdsBulkUploader(logger logger.Logger, statsFactory stats.Stats, destName string, service bingads.BulkServiceI, client *Client) *BingAdsBulkUploader {
+func NewBingAdsBulkUploader(logger logger.Logger, statsFactory stats.Stats, destName string, service bingads.BulkServiceI, isHashRequired bool) *BingAdsBulkUploader {
 	return &BingAdsBulkUploader{
-		destName:      destName,
-		service:       service,
-		logger:        logger.Child("BingAds").Child("BingAdsBulkUploader"),
-		statsFactory:  statsFactory,
-		client:        *client,
-		fileSizeLimit: common.GetBatchRouterConfigInt64("MaxUploadLimit", destName, 100*bytesize.MB),
-		eventsLimit:   common.GetBatchRouterConfigInt64("MaxEventsLimit", destName, 1000),
+		isHashRequired: isHashRequired,
+		destName:       destName,
+		service:        service,
+		logger:         logger.Child("BingAds").Child("BingAdsBulkUploader"),
+		statsFactory:   statsFactory,
+		fileSizeLimit:  common.GetBatchRouterConfigInt64("MaxUploadLimit", destName, 100*bytesize.MB),
+		eventsLimit:    common.GetBatchRouterConfigInt64("MaxEventsLimit", destName, 1000),
 	}
 }
 
@@ -81,6 +81,12 @@
 			}
 		}
 	}
+	if b.isHashRequired {
+		event.Fields, err = hashFields(fields)
+		if err != nil {
+			return payload, fmt.Errorf("Unable to hash fields.%w", err)
+		}
+	}
 	data := Data{
 		Message: Message{
 			Fields: event.Fields,

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/manager.go

@@ -49,9 +49,7 @@ func newManagerInternal(logger logger.Logger, statsFactory stats.Stats, destinat
 		TokenSource:    &token,
 	}
 	session := bingads.NewSession(sessionConfig)
-
-	clientNew := Client{}
-	bingUploader := NewBingAdsBulkUploader(logger, statsFactory, destination.DestinationDefinition.Name, bingads.NewBulkService(session), &clientNew)
+	bingUploader := NewBingAdsBulkUploader(logger, statsFactory, destination.DestinationDefinition.Name, bingads.NewBulkService(session), destConfig.IsHashRequired)
 	return bingUploader, nil
 }
 

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/types.go

@@ -3,25 +3,20 @@ package offline_conversions
 import (
 	"encoding/csv"
 	"encoding/json"
-	"net/http"
 
 	"github.com/rudderlabs/bing-ads-go-sdk/bingads"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 )
 
-type Client struct {
-	URL    string
-	client *http.Client
-}
 type BingAdsBulkUploader struct {
-	destName      string
-	service       bingads.BulkServiceI
-	logger        logger.Logger
-	statsFactory  stats.Stats
-	client        Client
-	fileSizeLimit int64
-	eventsLimit   int64
+	destName       string
+	service        bingads.BulkServiceI
+	logger         logger.Logger
+	statsFactory   stats.Stats
+	fileSizeLimit  int64
+	eventsLimit    int64
+	isHashRequired bool
 }
 type Message struct {
 	Fields json.RawMessage `json:"fields"`
@@ -41,6 +36,7 @@ type DestinationConfig struct {
 	CustomerAccountID string `json:"customerAccountId"`
 	CustomerID        string `json:"customerId"`
 	RudderAccountID   string `json:"rudderAccountId"`
+	IsHashRequired    bool   `json:"isHashRequired"`
 }
 
 type ActionFileInfo struct {

router/batchrouter/asyncdestinationmanager/bing-ads/offline-conversions/util.go

@@ -3,8 +3,10 @@
 import (
 	"archive/zip"
 	"bufio"
+	"crypto/sha256"
 	"encoding/csv"
 	"encoding/json"
+	stdjson "encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -466,3 +468,43 @@
 	}
 	return nil
 }
+
+func calculateHashCode(data string) string {
+	// Join the strings into a single string with a separator
+	hash := sha256.New()
+	hash.Write([]byte(data))
+	hashBytes := hash.Sum(nil)
+	hashCode := fmt.Sprintf("%x", hashBytes)
+
+	return hashCode
+}
+
+func hashFields(input map[string]interface{}) (stdjson.RawMessage, error) {
+	// Create a new map to hold the hashed fields
+	hashedMap := make(map[string]interface{})
+
+	// Iterate over the input map
+	for key, value := range input {
+		// Check if the key is "email" or "phone"
+		if key == "email" || key == "phone" {
+			// Ensure the value is a string before hashing
+			if strVal, ok := value.(string); ok {
+				hashedMap[key] = calculateHashCode(strVal)
+			} else {
+				// If not a string, preserve the original value
+				hashedMap[key] = value
+			}
+		} else {
+			// Preserve other fields unchanged
+			hashedMap[key] = value
+		}
+	}
+
+	// Convert the resulting map to JSON RawMessage
+	result, err := json.Marshal(hashedMap)
+	if err != nil {
+		return nil, err
+	}
+
+	return json.RawMessage(result), nil
+}