feat: regulation api support for Universal Analytics

regulation-worker/cmd/main.go

@@ -21,16 +21,22 @@ import (
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/destination"
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/initialize"
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/service"
+	oauth "github.com/rudderlabs/rudder-server/router/oauthResponseHandler"
 	"github.com/rudderlabs/rudder-server/services/filemanager"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 )
 
 var pkgLogger = logger.NewLogger().Child("regulation-worker")
 
-func main() {
+func InitAll() {
 	initialize.Init()
 	backendconfig.Init()
+	oauth.Init()
+}
+
+func main() {
+	InitAll()
 
 	pkgLogger.Info("starting regulation-worker")
 	ctx, cancel := context.WithCancel(context.Background())
@@ -58,6 +64,9 @@ func Run(ctx context.Context) {
 	if err != nil {
 		panic(fmt.Errorf("error while getting workspaceId: %w", err))
 	}
+	// setting up oauth
+	OAuth := oauth.NewOAuthErrorHandler(backendconfig.DefaultBackendConfig)
+
 	svc := service.JobSvc{
 		API: &client.JobAPI{
 			Client:         &http.Client{Timeout: config.GetDuration("HttpClient.regulationWorker.regulationManager.timeout", 60, time.Second)},
@@ -75,6 +84,7 @@ func Run(ctx context.Context) {
 			&api.APIManager{
 				Client:           &http.Client{Timeout: config.GetDuration("HttpClient.regulationWorker.transformer.timeout", 60, time.Second)},
 				DestTransformURL: config.MustGetString("DEST_TRANSFORM_URL"),
+				OAuth:            OAuth,
 			}),
 	}
 

regulation-worker/internal/delete/api/api.go

@@ -8,35 +8,40 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"os"
 	"strings"
 
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/model"
+	oauth "github.com/rudderlabs/rudder-server/router/oauthResponseHandler"
 	"github.com/rudderlabs/rudder-server/services/stats"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 )
 
 var (
 	pkgLogger             = logger.NewLogger().Child("api")
-	supportedDestinations = []string{"BRAZE", "AM", "INTERCOM", "CLEVERTAP", "AF", "MP"}
+	supportedDestinations = []string{"BRAZE", "AM", "INTERCOM", "CLEVERTAP", "AF", "MP", "GA"}
 )
 
 type APIManager struct {
 	Client           *http.Client
 	DestTransformURL string
+	OAuth            oauth.Authorizer
 }
 
 // prepares payload based on (job,destDetail) & make an API call to transformer.
 // gets (status, failure_reason) which is converted to appropriate model.Error & returned to caller.
-func (api *APIManager) Delete(ctx context.Context, job model.Job, destConfig map[string]interface{}, destName string) model.JobStatus {
+func (api *APIManager) Delete(ctx context.Context, job model.Job, destDetail model.Destination) model.JobStatus {
+	pkgLogger.Debugf("deleting: %v", job, " from API destination: %v", destDetail.Name)
 	method := http.MethodPost
 	endpoint := "/deleteUsers"
 	url := fmt.Sprint(api.DestTransformURL, endpoint)
 
-	bodySchema := mapJobToPayload(job, strings.ToLower(destName), destConfig)
+	bodySchema := mapJobToPayload(job, strings.ToLower(destDetail.Name), destDetail.Config)
+	pkgLogger.Debugf("payload: %#v", bodySchema)
 
 	reqBody, err := json.Marshal(bodySchema)
 	if err != nil {
@@ -49,11 +54,22 @@ func (api *APIManager) Delete(ctx context.Context, job model.Job, destConfig map
 	}
 	req.Header.Set("Content-Type", "application/json")
 
+	tokenInfo, getTokenInfoErr := api.getOAuthTokenInfo(job.WorkspaceID, &destDetail)
+	if getTokenInfoErr != nil {
+		// Problem in fetching token
+		return model.JobStatusFailed
+	}
+	setFailErr := setOAuthTokenInfo(*tokenInfo, req)
+	if setFailErr != nil {
+		// Marshal failure
+		return model.JobStatusFailed
+	}
+
 	fileCleaningTime := stats.Default.NewTaggedStat("file_cleaning_time", stats.TimerType, stats.Tags{
 		"jobId":       fmt.Sprintf("%d", job.ID),
 		"workspaceId": job.WorkspaceID,
 		"destType":    "api",
-		"destName":    strings.ToLower(destName),
+		"destName":    strings.ToLower(destDetail.Name),
 	})
 	fileCleaningTime.Start()
 	defer fileCleaningTime.End()
@@ -76,6 +92,13 @@ func (api *APIManager) Delete(ctx context.Context, job model.Job, destConfig map
 		return model.JobStatusFailed
 	}
 
+	// Refresh Flow Start
+	oauthDestRespStatus := api.getOAuthDestResponseStatus(tokenInfo, jobResp, job, destDetail)
+	if oauthDestRespStatus != model.JobStatusUndefined {
+		return oauthDestRespStatus
+	}
+	// Refresh Flow ends
+
 	switch resp.StatusCode {
 
 	case http.StatusOK:
@@ -126,3 +149,126 @@ func mapJobToPayload(job model.Job, destName string, destConfig map[string]inter
 		},
 	}
 }
+
+type handleRefreshFlowParams struct {
+	secret      json.RawMessage
+	destName    string
+	workspaceId string
+	accountId   string
+	isRefresh   bool
+}
+
+func shouldRefresh(jobResponses []JobRespSchema) bool {
+	var isRefresh bool
+	for _, jobResponse := range jobResponses {
+		isRefresh = jobResponse.AuthErrorCategory == oauth.REFRESH_TOKEN
+		if isRefresh {
+			break
+		}
+	}
+	return isRefresh
+}
+
+/*
+*
+This method handles the refresh flow for OAuth destinations
+When the status undefined(model.JobStatusUndefined) is returned, we can understand it is due to one of two things
+1. The error that was received was not an error that could trigger Refresh flow
+2. The destination itself is not OAuth one in the first place so it doesn't have errorCategory at all
+*
+*/
+func (api *APIManager) handleRefreshFlow(params *handleRefreshFlowParams) model.JobStatus {
+	if params.isRefresh {
+		pkgLogger.Debugf("Refresh flow triggered for %v\n", params.destName)
+		// Refresh OAuth flow
+		var refSecret *oauth.AuthResponse
+		var errCatStatusCode int
+		refTokenParams := &oauth.RefreshTokenParams{
+			Secret:          params.secret,
+			WorkspaceId:     params.workspaceId,
+			AccountId:       params.accountId,
+			DestDefName:     params.destName,
+			EventNamePrefix: "refresh_token",
+		}
+		errCatStatusCode, refSecret = api.OAuth.RefreshToken(refTokenParams)
+		refSec := *refSecret
+		if strings.TrimSpace(refSec.Err) != "" {
+			// There is an error occurring
+			pkgLogger.Warnf("Error: %v, Status: %v", refSec.Err, errCatStatusCode)
+		}
+		// Refresh is complete, the job has to be re-tried
+		return model.JobStatusFailed
+	}
+	// Indicates that OAuth refresh flow is not triggered
+	return model.JobStatusUndefined
+}
+
+func (api *APIManager) getOAuthDestResponseStatus(tokenInfo *OAuthTokenResult, jobResp []JobRespSchema, job model.Job, destDetail model.Destination) model.JobStatus {
+	status := model.JobStatusUndefined
+	if tokenInfo.IsOAuthEnabled {
+		shouldRefresh := shouldRefresh(jobResp)
+		status = api.handleRefreshFlow(&handleRefreshFlowParams{
+			isRefresh:   shouldRefresh,
+			secret:      tokenInfo.AccountSecretInfo.Account.Secret,
+			workspaceId: job.WorkspaceID,
+			accountId:   tokenInfo.DeleteAccountId,
+			destName:    destDetail.Name,
+		})
+	}
+	return status
+}
+
+type OAuthTokenResult struct {
+	AccountSecretInfo *oauth.AuthResponse
+	DeleteAccountId   string
+	IsOAuthEnabled    bool
+}
+
+func setOAuthTokenInfo(tokenInfo OAuthTokenResult, req *http.Request) error {
+	if tokenInfo.IsOAuthEnabled {
+		// setting oauth related information
+		payload, marshalErr := json.Marshal(tokenInfo.AccountSecretInfo.Account)
+		if marshalErr != nil {
+			marshalFailErr := fmt.Sprintf("error while marshalling account secret information: %v", marshalErr)
+			pkgLogger.Errorf(marshalFailErr)
+			return errors.New(marshalFailErr)
+		}
+		req.Header.Set("X-Rudder-Dest-Info", string(payload))
+	}
+	return nil
+}
+
+func (api *APIManager) getOAuthTokenInfo(workspaceId string, destDetail *model.Destination) (*OAuthTokenResult, error) {
+	var tokenStatusCode int
+	var accountSecretInfo *oauth.AuthResponse
+	// identifier to know if the destination supports OAuth for regulation-api
+	oAuthParamsResult := oauth.GetOAuthParams(oauth.OAuthParams{
+		DestConfig:    destDetail.Config,
+		DestDefConfig: destDetail.DestDefConfig,
+		IdKey:         "rudderDeleteAccountId",
+	})
+	if oAuthParamsResult.Enabled {
+		// Fetch Token call
+		// Get Access Token Information to send it as part of the event
+		tokenStatusCode, accountSecretInfo = api.OAuth.FetchToken(&oauth.RefreshTokenParams{
+			AccountId:       oAuthParamsResult.AccountId,
+			WorkspaceId:     workspaceId,
+			DestDefName:     destDetail.Name,
+			EventNamePrefix: "fetch_token",
+		})
+		pkgLogger.Debugf(`[%s][FetchToken] Token Fetch Method finished (statusCode, value): (%v, %+v)`, destDetail.Name, tokenStatusCode, accountSecretInfo)
+		if tokenStatusCode != http.StatusOK {
+			fetchFailStr := fmt.Sprintf("[%s][FetchToken] Error in Token Fetch statusCode: %d\t error: %s\n", destDetail.Name, tokenStatusCode, accountSecretInfo.Err)
+			pkgLogger.Errorf(fetchFailStr)
+			return nil, errors.New(fetchFailStr)
+		}
+
+	} else {
+		pkgLogger.Errorf("[%v] Destination probably doesn't support OAuth or some issue happened while doing OAuth for deletion [Enabled: %v]", destDetail.Name, oAuthParamsResult.Enabled)
+	}
+	return &OAuthTokenResult{
+		AccountSecretInfo: accountSecretInfo,
+		DeleteAccountId:   oAuthParamsResult.AccountId,
+		IsOAuthEnabled:    oAuthParamsResult.Enabled,
+	}, nil
+}