feat: adding support for future oauth facility in salesforce (#2746)

* feat: oauth in salesforce * fix: backward compatibility initial * fix: editing error class and proxy test case * fix: access token call for legacy auth * Update src/v0/destinations/salesforce/utils.js Co-authored-by: Sudip Paul <[email protected]> * feat: added missing authorization logic for axios calls * fix: adding new dest salesforce oauth * fix: review comment addressed * feat: review comments addressed * feat: small edit * fix: review comments addressed * Update src/v0/destinations/salesforce/transform.js Co-authored-by: Sankeerth <[email protected]> * fix: review comments addressed * fix: review comments addressed --------- Co-authored-by: Sudip Paul <[email protected]> Co-authored-by: Sankeerth <[email protected]>
rudderlabs · Nov 13, 2023 · 916ea4c · 916ea4c
1 parent e82cef8
commit 916ea4c
Show file tree

Hide file tree

Showing 7 changed files with 236 additions and 29 deletions.
diff --git a/src/constants/destinationCanonicalNames.js b/src/constants/destinationCanonicalNames.js
@@ -1,5 +1,6 @@
 const DestHandlerMap = {
   ga360: 'ga',
+  salesforce_oauth: 'salesforce',
 };
 
 const DestCanonicalNames = {

diff --git a/src/features.json b/src/features.json
@@ -25,6 +25,7 @@
     "PINTEREST_TAG": true,
     "PROFITWELL": true,
     "SALESFORCE": true,
+    "SALESFORCE_OAUTH": true,
     "SFMC": true,
     "SNAPCHAT_CONVERSION": true,
     "TIKTOK_ADS": true,

diff --git a/src/v0/destinations/salesforce/config.js b/src/v0/destinations/salesforce/config.js
@@ -24,6 +24,8 @@ const SF_TOKEN_REQUEST_URL = 'https://login.salesforce.com/services/oauth2/token
 const SF_TOKEN_REQUEST_URL_SANDBOX = 'https://test.salesforce.com/services/oauth2/token';
 
 const DESTINATION = 'Salesforce';
+const OAUTH = 'oauth';
+const LEGACY = 'legacy';
 
 const mappingConfig = getMappingConfig(ConfigCategory, __dirname);
 
@@ -37,4 +39,6 @@ module.exports = {
   ignoredContactTraits: mappingConfig[ConfigCategory.IGNORE_CONTACT.name],
   ACCESS_TOKEN_CACHE_TTL,
   DESTINATION,
+  OAUTH,
+  LEGACY,
 };
diff --git a/src/v0/destinations/salesforce/transform.js b/src/v0/destinations/salesforce/transform.js
@@ -26,8 +26,9 @@ const {
   checkInvalidRtTfEvents,
   handleRtTfSingleEventError,
   generateErrorObject,
+  isHttpStatusSuccess,
 } = require('../../util');
-const { getAccessToken, salesforceResponseHandler } = require('./utils');
+const { salesforceResponseHandler, collectAuthorizationInfo, getAuthHeader } = require('./utils');
 const { handleHttpRequest } = require('../../../adapters/network');
 const { JSON_MIME_TYPE } = require('../../util/constant');
 
@@ -41,6 +42,7 @@ function responseBuilderSimple(
   authorizationData,
   mapProperty,
   mappedToDestination,
+  authorizationFlow,
 ) {
   const { salesforceType, salesforceId } = salesforceMap;
 
@@ -87,12 +89,12 @@ function responseBuilderSimple(
   }
 
   const response = defaultRequestConfig();
-  const header = {
+
+  response.method = defaultPostRequestConfig.requestMethod;
+  response.headers = {
     'Content-Type': JSON_MIME_TYPE,
-    Authorization: authorizationData.token,
+    ...getAuthHeader({ authorizationFlow, authorizationData }),
   };
-  response.method = defaultPostRequestConfig.requestMethod;
-  response.headers = header;
   response.body.JSON = removeUndefinedValues(rawPayload);
   response.endpoint = targetEndpoint;
 
@@ -106,24 +108,26 @@ async function getSaleforceIdForRecord(
   identifierType,
   identifierValue,
   destination,
+  authorizationFlow,
 ) {
   const objSearchUrl = `${authorizationData.instanceUrl}/services/data/v${SF_API_VERSION}/parameterizedSearch/?q=${identifierValue}&sobject=${objectType}&in=${identifierType}&${objectType}.fields=id,${identifierType}`;
   const { processedResponse: processedsfSearchResponse } = await handleHttpRequest(
     'get',
     objSearchUrl,
     {
-      headers: { Authorization: authorizationData.token },
+      headers: getAuthHeader({ authorizationFlow, authorizationData }),
     },
     {
       destType: 'salesforce',
       feature: 'transformation',
     },
   );
-  if (processedsfSearchResponse.status !== 200) {
+  if (!isHttpStatusSuccess(processedsfSearchResponse.status)) {
     salesforceResponseHandler(
       processedsfSearchResponse,
       `:- SALESFORCE SEARCH BY ID`,
       destination.ID,
+      authorizationFlow,
     );
   }
   const searchRecord = processedsfSearchResponse.response?.searchRecords?.find(
@@ -149,7 +153,12 @@ async function getSaleforceIdForRecord(
 // We'll use the Salesforce Object names by removing "Salesforce-" string from the type field
 //
 // Default Object type will be "Lead" for backward compatibility
-async function getSalesforceIdFromPayload(message, authorizationData, destination) {
+async function getSalesforceIdFromPayload(
+  message,
+  authorizationData,
+  destination,
+  authorizationFlow,
+) {
   // define default map
   const salesforceMaps = [];
 
@@ -191,6 +200,7 @@ async function getSalesforceIdFromPayload(message, authorizationData, destinatio
         identifierType,
         id,
         destination,
+        authorizationFlow,
       );
     }
 
@@ -212,21 +222,27 @@ async function getSalesforceIdFromPayload(message, authorizationData, destinatio
       throw new InstrumentationError('Invalid Email address for Lead Objet');
     }
     const leadQueryUrl = `${authorizationData.instanceUrl}/services/data/v${SF_API_VERSION}/parameterizedSearch/?q=${email}&sobject=Lead&Lead.fields=id,IsConverted,ConvertedContactId,IsDeleted`;
+
     // request configuration will be conditional
     const { processedResponse: processedLeadQueryResponse } = await handleHttpRequest(
       'get',
       leadQueryUrl,
       {
-        headers: { Authorization: authorizationData.token },
+        headers: getAuthHeader({ authorizationFlow, authorizationData }),
       },
       {
         destType: 'salesforce',
         feature: 'transformation',
       },
     );
 
-    if (processedLeadQueryResponse.status !== 200) {
-      salesforceResponseHandler(processedLeadQueryResponse, `:- during Lead Query`, destination.ID);
+    if (!isHttpStatusSuccess(processedLeadQueryResponse.status)) {
+      salesforceResponseHandler(
+        processedLeadQueryResponse,
+        `:- during Lead Query`,
+        destination.ID,
+        authorizationFlow,
+      );
     }
 
     if (processedLeadQueryResponse.response.searchRecords.length > 0) {
@@ -262,7 +278,7 @@ async function getSalesforceIdFromPayload(message, authorizationData, destinatio
 }
 
 // Function for handling identify events
-async function processIdentify(message, authorizationData, destination) {
+async function processIdentify(message, authorizationData, destination, authorizationFlow) {
   const mapProperty =
     destination.Config.mapProperty === undefined ? true : destination.Config.mapProperty;
   // check the traits before hand
@@ -283,7 +299,12 @@ async function processIdentify(message, authorizationData, destination) {
   const responseData = [];
 
   // get salesforce object map
-  const salesforceMaps = await getSalesforceIdFromPayload(message, authorizationData, destination);
+  const salesforceMaps = await getSalesforceIdFromPayload(
+    message,
+    authorizationData,
+    destination,
+    authorizationFlow,
+  );
 
   // iterate over the object types found
   salesforceMaps.forEach((salesforceMap) => {
@@ -295,6 +316,7 @@ async function processIdentify(message, authorizationData, destination) {
         authorizationData,
         mapProperty,
         mappedToDestination,
+        authorizationFlow,
       ),
     );
   });
@@ -304,20 +326,24 @@ async function processIdentify(message, authorizationData, destination) {
 
 // Generic process function which invokes specific handler functions depending on message type
 // and event type where applicable
-async function processSingleMessage(message, authorizationData, destination) {
+async function processSingleMessage(message, authorizationData, destination, authorizationFlow) {
   let response;
   if (message.type === EventType.IDENTIFY) {
-    response = await processIdentify(message, authorizationData, destination);
+    response = await processIdentify(message, authorizationData, destination, authorizationFlow);
   } else {
     throw new InstrumentationError(`message type ${message.type} is not supported`);
   }
   return response;
 }
 
 async function process(event) {
-  // Get the authorization header if not available
-  const authorizationData = await getAccessToken(event.destination);
-  const response = await processSingleMessage(event.message, authorizationData, event.destination);
+  const authInfo = await collectAuthorizationInfo(event);
+  const response = await processSingleMessage(
+    event.message,
+    authInfo.authorizationData,
+    event.destination,
+    authInfo.authorizationFlow,
+  );
   return response;
 }
 
@@ -326,10 +352,9 @@ const processRouterDest = async (inputs, reqMetadata) => {
   if (errorRespEvents.length > 0) {
     return errorRespEvents;
   }
-
-  let authorizationData;
+  let authInfo;
   try {
-    authorizationData = await getAccessToken(inputs[0].destination);
+    authInfo = await collectAuthorizationInfo(inputs[0]);
   } catch (error) {
     const errObj = generateErrorObject(error);
     const respEvents = getErrorRespEvents(
@@ -351,7 +376,12 @@ const processRouterDest = async (inputs, reqMetadata) => {
 
         // unprocessed payload
         return getSuccessRespEvents(
-          await processSingleMessage(input.message, authorizationData, input.destination),
+          await processSingleMessage(
+            input.message,
+            authInfo.authorizationData,
+            input.destination,
+            authInfo.authorizationFlow,
+          ),
           [input.metadata],
           input.destination,
         );

diff --git a/src/v0/destinations/salesforce/utils.js b/src/v0/destinations/salesforce/utils.js
@@ -1,12 +1,18 @@
 const { RetryableError, ThrottledError, AbortedError } = require('@rudderstack/integrations-lib');
 const { handleHttpRequest } = require('../../../adapters/network');
-const { isHttpStatusSuccess } = require('../../util');
+const {
+  isHttpStatusSuccess,
+  getAuthErrCategoryFromStCode,
+  isDefinedAndNotNull,
+} = require('../../util');
 const Cache = require('../../util/cache');
 const {
   ACCESS_TOKEN_CACHE_TTL,
   SF_TOKEN_REQUEST_URL_SANDBOX,
   SF_TOKEN_REQUEST_URL,
   DESTINATION,
+  LEGACY,
+  OAUTH,
 } = require('./config');
 
 const ACCESS_TOKEN_CACHE = new Cache(ACCESS_TOKEN_CACHE_TTL);
@@ -19,21 +25,24 @@ const ACCESS_TOKEN_CACHE = new Cache(ACCESS_TOKEN_CACHE_TTL);
  * @param {*} stage
  * @param {String} authKey
  */
-const salesforceResponseHandler = (destResponse, sourceMessage, authKey) => {
+const salesforceResponseHandler = (destResponse, sourceMessage, authKey, authorizationFlow) => {
   const { status, response } = destResponse;
 
   // if the response from destination is not a success case build an explicit error
   if (!isHttpStatusSuccess(status) && status >= 400) {
     const matchErrorCode = (errorCode) =>
       response && Array.isArray(response) && response.some((resp) => resp?.errorCode === errorCode);
     if (status === 401 && authKey && matchErrorCode('INVALID_SESSION_ID')) {
-      // checking for invalid/expired token errors and evicting cache in that case
-      // rudderJobMetadata contains some destination info which is being used to evict the cache
-      ACCESS_TOKEN_CACHE.del(authKey);
+      if (authorizationFlow === LEGACY) {
+        // checking for invalid/expired token errors and evicting cache in that case
+        // rudderJobMetadata contains some destination info which is being used to evict the cache
+        ACCESS_TOKEN_CACHE.del(authKey);
+      }
       throw new RetryableError(
         `${DESTINATION} Request Failed - due to "INVALID_SESSION_ID", (Retryable) ${sourceMessage}`,
         500,
         destResponse,
+        authorizationFlow === LEGACY ? '' : getAuthErrCategoryFromStCode(status),
       );
     } else if (status === 403 && matchErrorCode('REQUEST_LIMIT_EXCEEDED')) {
       // If the error code is REQUEST_LIMIT_EXCEEDED, you’ve exceeded API request limits in your org.
@@ -89,6 +98,11 @@ const salesforceResponseHandler = (destResponse, sourceMessage, authKey) => {
  * @param {*} destination
  * @returns
  */
+const getAccessTokenOauth = (metadata) => ({
+  token: metadata.secret?.access_token,
+  instanceUrl: metadata.secret?.instance_url,
+});
+
 const getAccessToken = async (destination) => {
   const accessTokenKey = destination.ID;
 
@@ -122,6 +136,7 @@ const getAccessToken = async (destination) => {
         processedResponse,
         `:- authentication failed during fetching access token.`,
         accessTokenKey,
+        LEGACY,
       );
     }
     const token = httpResponse.response.data;
@@ -131,6 +146,7 @@ const getAccessToken = async (destination) => {
         processedResponse,
         `:- authentication failed could not retrieve authorization token.`,
         accessTokenKey,
+        LEGACY,
       );
     }
     return {
@@ -140,4 +156,30 @@ const getAccessToken = async (destination) => {
   });
 };
 
-module.exports = { getAccessToken, salesforceResponseHandler };
+const collectAuthorizationInfo = async (event) => {
+  let authorizationFlow;
+  let authorizationData;
+  if (isDefinedAndNotNull(event.metadata?.secret)) {
+    authorizationFlow = OAUTH;
+    authorizationData = getAccessTokenOauth(event.metadata);
+  } else {
+    authorizationFlow = LEGACY;
+    authorizationData = await getAccessToken(event.destination);
+  }
+  return { authorizationFlow, authorizationData };
+};
+
+const getAuthHeader = (authInfo) => {
+  const { authorizationFlow, authorizationData } = authInfo;
+  return authorizationFlow === OAUTH
+    ? { Authorization: `Bearer ${authorizationData.token}` }
+    : { Authorization: authorizationData.token };
+};
+
+module.exports = {
+  getAccessTokenOauth,
+  salesforceResponseHandler,
+  getAccessToken,
+  collectAuthorizationInfo,
+  getAuthHeader,
+};
diff --git a/test/integrations/destinations/salesforce/dataDelivery/data.ts b/test/integrations/destinations/salesforce/dataDelivery/data.ts
@@ -119,6 +119,7 @@ export const data = [
         body: {
           output: {
             status: 500,
+            authErrorCategory: 'REFRESH_TOKEN',
             message:
               'Salesforce Request Failed - due to "INVALID_SESSION_ID", (Retryable) during Salesforce Response Handling',
             destinationResponse: {
@@ -128,12 +129,12 @@ export const data = [
                   errorCode: 'INVALID_SESSION_ID',
                 },
               ],
-              status: 401,
               rudderJobMetadata: {
                 destInfo: {
                   authKey: '2HezPl1w11opbFSxnLDEgZ7kWTf',
                 },
               },
+              status: 401,
             },
             statTags: {
               destType: 'SALESFORCE',