feat: oauth access denied handling

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

@@ -15,8 +15,11 @@ import (
 	"os"
 	"strings"
 
+	"github.com/samber/lo"
+
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
+	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/model"
 	"github.com/rudderlabs/rudder-server/services/oauth"
 	"github.com/rudderlabs/rudder-server/utils/httputil"
@@ -109,17 +112,23 @@ func (api *APIManager) deleteWithRetry(ctx context.Context, job model.Job, desti
 	jobStatus := getJobStatus(resp.StatusCode, jobResp)
 	pkgLogger.Debugf("[%v] Job: %v, JobStatus: %v", destination.Name, job.ID, jobStatus)
 
-	if isOAuthEnabled && isTokenExpired(jobResp) && currentOauthRetryAttempt < api.MaxOAuthRefreshRetryAttempts {
-		err = api.refreshOAuthToken(destination.Name, job.WorkspaceID, oAuthDetail)
-		if err != nil {
-			pkgLogger.Error(err)
-			return model.JobStatus{Status: model.JobStatusFailed, Error: err}
+	oauthErrJob, oauthErrJobFound := getOAuthErrorJob(jobResp)
+
+	if oauthErrJobFound && isOAuthEnabled {
+		if oauthErrJob.AuthErrorCategory == oauth.AUTH_STATUS_INACTIVE {
+			return api.inactivateAuthStatus(&destination, job, oAuthDetail)
+		}
+		if oauthErrJob.AuthErrorCategory == oauth.REFRESH_TOKEN && currentOauthRetryAttempt < api.MaxOAuthRefreshRetryAttempts {
+			err = api.refreshOAuthToken(&destination, job, oAuthDetail)
+			if err != nil {
+				pkgLogger.Error(err)
+				return model.JobStatus{Status: model.JobStatusFailed, Error: err}
+			}
+			// retry the request
+			pkgLogger.Infof("[%v] Retrying deleteRequest job(id: %v) for the whole batch, RetryAttempt: %v", destination.Name, job.ID, currentOauthRetryAttempt+1)
+			return api.deleteWithRetry(ctx, job, destination, currentOauthRetryAttempt+1)
 		}
-		// retry the request
-		pkgLogger.Infof("[%v] Retrying deleteRequest job(id: %v) for the whole batch, RetryAttempt: %v", destination.Name, job.ID, currentOauthRetryAttempt+1)
-		return api.deleteWithRetry(ctx, job, destination, currentOauthRetryAttempt+1)
 	}
-
 	return jobStatus
 }
 
@@ -160,13 +169,10 @@ func mapJobToPayload(job model.Job, destName string, destConfig map[string]inter
 	}
 }
 
-func isTokenExpired(jobResponses []JobRespSchema) bool {
-	for _, jobResponse := range jobResponses {
-		if jobResponse.AuthErrorCategory == oauth.REFRESH_TOKEN {
-			return true
-		}
-	}
-	return false
+func getOAuthErrorJob(jobResponses []JobRespSchema) (JobRespSchema, bool) {
+	return lo.Find(jobResponses, func(item JobRespSchema) bool {
+		return lo.Contains([]string{oauth.AUTH_STATUS_INACTIVE, oauth.REFRESH_TOKEN}, item.AuthErrorCategory)
+	})
 }
 
 func setOAuthHeader(secretToken *oauth.AuthResponse, req *http.Request) error {
@@ -192,29 +198,55 @@ func (api *APIManager) getOAuthDetail(destDetail *model.Destination, workspaceId
 		EventNamePrefix: "fetch_token",
 	})
 	if tokenStatusCode != http.StatusOK {
-		return oauthDetail{}, fmt.Errorf("[%s][FetchToken] Error in Token Fetch statusCode: %d\t error: %s", destDetail.Name, tokenStatusCode, secretToken.Err)
+		return oauthDetail{}, fmt.Errorf("[%s][FetchToken] Error in Token Fetch statusCode: %d\t error: %s", destDetail.Name, tokenStatusCode, secretToken.ErrorMessage)
 	}
 	return oauthDetail{
 		id:          id,
 		secretToken: secretToken,
 	}, nil
 }
 
-func (api *APIManager) refreshOAuthToken(destName, workspaceId string, oAuthDetail oauthDetail) error {
+func (api *APIManager) inactivateAuthStatus(destination *model.Destination, job model.Job, oAuthDetail oauthDetail) (jobStatus model.JobStatus) {
+	dest := &backendconfig.DestinationT{
+		ID:     destination.DestinationID,
+		Config: destination.Config,
+		DestinationDefinition: backendconfig.DestinationDefinitionT{
+			Name:   destination.Name,
+			Config: destination.DestDefConfig,
+		},
+	}
+	_, resp := api.OAuth.AuthStatusToggle(&oauth.AuthStatusToggleParams{
+		Destination:     dest,
+		WorkspaceId:     job.WorkspaceID,
+		RudderAccountId: oAuthDetail.id,
+		AuthStatus:      oauth.AuthStatusInactive,
+	})
+	jobStatus.Status = model.JobStatusAborted
+	jobStatus.Error = fmt.Errorf(resp)
+	return jobStatus
+}
+
+func (api *APIManager) refreshOAuthToken(destination *model.Destination, job model.Job, oAuthDetail oauthDetail) error {
 	refTokenParams := &oauth.RefreshTokenParams{
 		Secret:          oAuthDetail.secretToken.Account.Secret,
-		WorkspaceId:     workspaceId,
+		WorkspaceId:     job.WorkspaceID,
 		AccountId:       oAuthDetail.id,
-		DestDefName:     destName,
+		DestDefName:     destination.Name,
 		EventNamePrefix: "refresh_token",
 	}
 	statusCode, refreshResponse := api.OAuth.RefreshToken(refTokenParams)
 	if statusCode != http.StatusOK {
+		if refreshResponse.Err == oauth.REF_TOKEN_INVALID_GRANT {
+			// authStatus should be made inactive
+			api.inactivateAuthStatus(destination, job, oAuthDetail)
+			return fmt.Errorf(refreshResponse.ErrorMessage)
+		}
+
 		var refreshRespErr string
 		if refreshResponse != nil {
-			refreshRespErr = refreshResponse.Err
+			refreshRespErr = refreshResponse.ErrorMessage
 		}
-		return fmt.Errorf("[%v] Failed to refresh token for destination in workspace(%v) & account(%v) with %v", destName, workspaceId, oAuthDetail.id, refreshRespErr)
+		return fmt.Errorf("[%v] Failed to refresh token for destination in workspace(%v) & account(%v) with %v", destination.Name, job.WorkspaceID, oAuthDetail.id, refreshRespErr)
 	}
 	return nil
 }