feat: onboard Yandex Metrica Offline Events destination (#4534)

* chore: initial commit for ymoe

* feat: onboard yandex metrica offline events destination, initial draft commit

* chore: address comments, update csv generation updation logic

* feat: add oauthv2 support into yandex

* fix: add missing manager.go changes

* fix: oauthv2 panic & add success response

* chore: update request with destinationInfo in context

* fix: update csvfilepath and refactor code

* chore: add unit tests

* chore: refactore code

* chore: fix test case

* chore: fix upload config

* chore: address all formatting comments

* chore: address all formatting commentsx2

* chore: update to use strconv for Price

* chore: update error messages

* fix: error test-case

* chore: address comment, move id switch logic to function

* chore: address review comments

* chore: fix lint issue

---------

Co-authored-by: Sai Sankeerth <sanpj2292@github.com>
Co-authored-by: ItsSudip <sudip.paul1997@gmail.com>
Co-authored-by: Sudip Paul <67197965+ItsSudip@users.noreply.github.com>

router/batchrouter/asyncdestinationmanager/common/common.go

@@ -19,7 +19,7 @@ type AsyncDestinationManager interface {
 	GetUploadStats(UploadStatsInput GetUploadStatsInput) GetUploadStatsResponse
 }
 
-var AsyncDestinations = []string{"MARKETO_BULK_UPLOAD", "BING_ADS", "ELOQUA"}
+var AsyncDestinations = []string{"MARKETO_BULK_UPLOAD", "BING_ADS", "ELOQUA", "YANDEX_METRICA_OFFLINE_EVENTS"}
 
 type PollStatusResponse struct {
 	Complete       bool

router/batchrouter/asyncdestinationmanager/manager.go

@@ -10,6 +10,7 @@ import (
 	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/common"
 	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/eloqua"
 	marketobulkupload "github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/marketo-bulk-upload"
+	"github.com/rudderlabs/rudder-server/router/batchrouter/asyncdestinationmanager/yandexmetrica"
 )
 
 var json = jsoniter.ConfigCompatibleWithStandardLibrary
@@ -37,6 +38,8 @@ func NewManager(destination *backendconfig.DestinationT, backendConfig backendco
 		return marketobulkupload.NewManager(destination)
 	case "ELOQUA":
 		return eloqua.NewManager(destination)
+	case "YANDEX_METRICA_OFFLINE_EVENTS":
+		return yandexmetrica.NewManager(destination, backendConfig)
 	}
 	return nil, errors.New("invalid destination type")
 }

router/batchrouter/asyncdestinationmanager/yandexmetrica/augmenter/augmenter.go

@@ -0,0 +1,54 @@
+package augmenter
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/samber/lo"
+	"github.com/tidwall/gjson"
+
+	"github.com/rudderlabs/rudder-server/services/oauth/v2/common"
+)
+
+type yandexAugmenter struct{}
+
+var YandexReqAugmenter = &yandexAugmenter{}
+
+// Custom augmenter for Yandex which sets token to Authorization header
+func (y *yandexAugmenter) Augment(r *http.Request, body []byte, secret json.RawMessage) error {
+	if secret == nil {
+		return errors.New("secret is nil")
+	}
+	token := gjson.GetBytes(secret, "accessToken").String()
+	// format -> Authorization : OAuth <accessToken>
+	r.Header.Set("Authorization", fmt.Sprintf("OAuth %s", token))
+	r.Body = io.NopCloser(bytes.NewReader(body))
+	return nil
+}
+
+func GetAuthErrorCategoryForYandex(responseBody []byte) (string, error) {
+	/*
+		Sample response for Yandex
+		{
+		    "errors": [
+		        {
+		            "error_type": "invalid_token",
+		            "message": "Invalid oauth_token"
+		        }
+		    ],
+		    "code": 403,
+		    "message": "Invalid oauth_token"
+		}
+	*/
+	if len(lo.Filter(gjson.GetBytes(responseBody, "errors.#.error_type").Array(), func(errorTypeResult gjson.Result, _ int) bool {
+		return strings.Contains(errorTypeResult.String(), "invalid_token")
+	})) > 0 {
+		return common.CategoryRefreshToken, nil
+	}
+	return "", nil
+}

router/batchrouter/asyncdestinationmanager/yandexmetrica/testdata/uploadData.txt

@@ -0,0 +1,4 @@
+{"message":{"DateTime":"361519255","Price":894.7297191082301,"Target":"1234","UserId":"221312"},"metadata":{"job_id":5}}
+{"message":{"DateTime":"361789255","Price":894.729,"Target":"1234","UserId":"221232"},"metadata":{"job_id":6}}
+{"message":{"DateTime":"361511235","Price":23,"Target":"1234","UserId":"221652"},"metadata":{"job_id":7}}
+{"message":{"DateTime":"361545655","Price":65,"Target":"1234","UserId":"289712"},"metadata":{"job_id":8}}