feat(hs): chunking data based on batch limit

src/v0/destinations/hs/config.js

@@ -64,6 +64,8 @@ const API_VERSION = {
   v3: 'newApi',
 };
 
+const MAX_CONTACTS_PER_REQUEST = 100;
+
 const ConfigCategory = {
   COMMON: {
     name: 'HSCommonConfig',
@@ -109,5 +111,6 @@ module.exports = {
   SEARCH_LIMIT_VALUE,
   RETL_SOURCE,
   RETL_CREATE_ASSOCIATION_OPERATION,
+  MAX_CONTACTS_PER_REQUEST,
   DESTINATION: 'HS',
 };

src/v0/destinations/hs/util.js

@@ -1,3 +1,5 @@
+/* eslint-disable no-await-in-loop */
+const lodash = require('lodash');
 const get = require('get-value');
 const {
   NetworkInstrumentationError,
@@ -25,6 +27,7 @@ const {
   SEARCH_LIMIT_VALUE,
   hsCommonConfigJson,
   DESTINATION,
+  MAX_CONTACTS_PER_REQUEST,
 } = require('./config');
 
 const tags = require('../../util/tags');
@@ -160,16 +163,14 @@ const validatePayloadDataTypes = (propertyMap, hsSupportedKey, value, traitsKey)
 
   if (propertyMap[hsSupportedKey] === 'bool' && typeof propValue === 'object') {
     throw new InstrumentationError(
-      `Property ${traitsKey} data type ${typeof propValue} is not matching with Hubspot property data type ${
-        propertyMap[hsSupportedKey]
+      `Property ${traitsKey} data type ${typeof propValue} is not matching with Hubspot property data type ${propertyMap[hsSupportedKey]
       }`,
     );
   }
 
   if (propertyMap[hsSupportedKey] === 'number' && typeof propValue !== 'number') {
     throw new InstrumentationError(
-      `Property ${traitsKey} data type ${typeof propValue} is not matching with Hubspot property data type ${
-        propertyMap[hsSupportedKey]
+      `Property ${traitsKey} data type ${typeof propValue} is not matching with Hubspot property data type ${propertyMap[hsSupportedKey]
       }`,
     );
   }
@@ -464,42 +465,128 @@ const getEventAndPropertiesFromConfig = (message, destination, payload) => {
 };
 
 /**
- * DOC: https://developers.hubspot.com/docs/api/crm/search
+ * Validates object and identifier type is present in message
+ * @param {*} firstMessage
+ * @returns
+ */
+const getObjectAndIdentifierType = (firstMessage) => {
+  const { objectType, identifierType } = getDestinationExternalIDInfoForRetl(
+    firstMessage,
+    DESTINATION,
+  );
+  if (!objectType || !identifierType) {
+    throw new InstrumentationError('rETL - external Id not found.');
+  }
+  return { objectType, identifierType };
+};
+
+/**
+ * Returns values for search api call
  * @param {*} inputs
+ * @returns
+ */
+const extractIDsForSearchAPI = (inputs) => {
+  const values = inputs.map((input) => {
+    const { message } = input;
+    const { destinationExternalId } = getDestinationExternalIDInfoForRetl(message, DESTINATION);
+    return destinationExternalId.toString().toLowerCase();
+  });
+
+  return Array.from(new Set(values));
+};
+
+/**
+ * Returns hubspot records
+ * Ref : https://developers.hubspot.com/docs/api/crm/search
+ * @param {*} data
+ * @param {*} requestOptions
+ * @param {*} objectType
+ * @param {*} identifierType
  * @param {*} destination
+ * @returns
  */
-const getExistingData = async (inputs, destination) => {
+const performHubSpotSearch = async (
+  reqdata,
+  reqOptions,
+  objectType,
+  identifierType,
+  destination,
+) => {
+  let checkAfter = 1;
+  const searchResults = [];
+  const requestData = reqdata;
   const { Config } = destination;
-  let values = [];
-  let searchResponse;
-  let updateHubspotIds = [];
-  const firstMessage = inputs[0].message;
-  let objectType = null;
-  let identifierType = null;
-
-  if (firstMessage) {
-    objectType = getDestinationExternalIDInfoForRetl(firstMessage, DESTINATION).objectType;
-    identifierType = getDestinationExternalIDInfoForRetl(firstMessage, DESTINATION).identifierType;
-    if (!objectType || !identifierType) {
-      throw new InstrumentationError('rETL - external Id not found.');
+
+  const endpoint = IDENTIFY_CRM_SEARCH_ALL_OBJECTS.replace(':objectType', objectType);
+  const endpointPath = `objects/:objectType/search`;
+
+  const url =
+    Config.authorizationType === 'newPrivateAppApi'
+      ? endpoint
+      : `${endpoint}?hapikey=${Config.apiKey}`;
+
+  const requestOptions = Config.authorizationType === 'newPrivateAppApi' ? reqOptions : {};
+
+  /* *
+   * This is needed for processing paginated response when searching hubspot.
+   * we can't avoid await in loop as response to the request contains the pagination details
+   * */
+
+  while (checkAfter) {
+    const searchResponse = await httpPOST(url, requestData, requestOptions, {
+      destType: 'hs',
+      feature: 'transformation',
+      endpointPath,
+    });
+
+    const processedResponse = processAxiosResponse(searchResponse);
+
+    if (processedResponse.status !== 200) {
+      throw new NetworkError(
+        `rETL - Error during searching object record. ${JSON.stringify(
+          processedResponse.response?.message,
+        )}`,
+        processedResponse.status,
+        {
+          [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(processedResponse.status),
+        },
+        processedResponse,
+      );
+    }
+
+    const after = processedResponse.response?.paging?.next?.after || 0;
+    requestData.after = after; // assigning to the new value of after
+    checkAfter = after; // assigning to the new value if no after we assign it to 0 and no more calls will take place
+
+    const results = processedResponse.response?.results;
+    if (results) {
+      searchResults.push(
+        ...results.map((result) => ({
+          id: result.id,
+          property: result.properties[identifierType],
+        })),
+      );
     }
-  } else {
-    throw new InstrumentationError('rETL - objectType or identifier type not found. ');
   }
-  inputs.map(async (input) => {
-    const { message } = input;
-    const { destinationExternalId } = getDestinationExternalIDInfoForRetl(message, DESTINATION);
-    values.push(destinationExternalId.toString().toLowerCase());
-  });
 
-  values = Array.from(new Set(values));
+  return searchResults;
+};
+
+/**
+ * Returns requestData and requestOptions
+ * @param {*} identifierType
+ * @param {*} chunk
+ * @param {*} accessToken
+ * @returns
+ */
+const getRequestDataAndRequestOptions = (identifierType, chunk) => {
   const requestData = {
     filterGroups: [
       {
         filters: [
           {
             propertyName: identifierType,
-            values,
+            values: chunk,
             operator: 'IN',
           },
         ],
@@ -510,65 +597,48 @@ const getExistingData = async (inputs, destination) => {
     after: 0,
   };
 
+  return { requestData };
+};
+
+/**
+ * DOC: https://developers.hubspot.com/docs/api/crm/search
+ * @param {*} inputs
+ * @param {*} destination
+ */
+const getExistingContactsData = async (inputs, destination) => {
+  const { Config } = destination;
+  const updateHubspotIds = [];
+  const firstMessage = inputs[0].message;
+
+  if (!firstMessage) {
+    throw new InstrumentationError('rETL - objectType or identifier type not found.');
+  }
+
+  const { objectType, identifierType } = getObjectAndIdentifierType(firstMessage);
+
+  const values = extractIDsForSearchAPI(inputs);
+  const valuesChunk = lodash.chunk(values, MAX_CONTACTS_PER_REQUEST);
   const requestOptions = {
     headers: {
       'Content-Type': JSON_MIME_TYPE,
       Authorization: `Bearer ${Config.accessToken}`,
     },
   };
-  let checkAfter = 1; // variable to keep checking if we have more results
-
-  /* eslint-disable no-await-in-loop */
-
-  /* *
-   * This is needed for processing paginated response when searching hubspot.
-   * we can't avoid await in loop as response to the request contains the pagination details
-   * */
-
-  while (checkAfter) {
-    const endpoint = IDENTIFY_CRM_SEARCH_ALL_OBJECTS.replace(':objectType', objectType);
-    const endpointPath = `objects/:objectType/search`;
-
-    const url =
-      Config.authorizationType === 'newPrivateAppApi'
-        ? endpoint
-        : `${endpoint}?hapikey=${Config.apiKey}`;
-    searchResponse =
-      Config.authorizationType === 'newPrivateAppApi'
-        ? await httpPOST(url, requestData, requestOptions, {
-            destType: 'hs',
-            feature: 'transformation',
-            endpointPath,
-          })
-        : await httpPOST(url, requestData, {
-            destType: 'hs',
-            feature: 'transformation',
-            endpointPath,
-          });
-    searchResponse = processAxiosResponse(searchResponse);
-
-    if (searchResponse.status !== 200) {
-      throw new NetworkError(
-        `rETL - Error during searching object record. ${searchResponse.response?.message}`,
-        searchResponse.status,
-        {
-          [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(searchResponse.status),
-        },
-        searchResponse,
-      );
-    }
-
-    const after = searchResponse.response?.paging?.next?.after || 0;
-
-    requestData.after = after; // assigning to the new value of after
-    checkAfter = after; // assigning to the new value if no after we assign it to 0 and no more calls will take place
-
-    const results = searchResponse.response?.results;
-    if (results) {
-      updateHubspotIds = results.map((result) => {
-        const propertyValue = result.properties[identifierType];
-        return { id: result.id, property: propertyValue };
-      });
+  // eslint-disable-next-line no-restricted-syntax
+  for (const chunk of valuesChunk) {
+    const { requestData } = getRequestDataAndRequestOptions(
+      identifierType,
+      chunk,
+    );
+    const searchResults = await performHubSpotSearch(
+      requestData,
+      requestOptions,
+      objectType,
+      identifierType,
+      destination,
+    );
+    if (searchResults.length > 0) {
+      updateHubspotIds.push(...searchResults);
     }
   }
   return updateHubspotIds;
@@ -601,7 +671,7 @@ const setHsSearchId = (input, id) => {
 
 const splitEventsForCreateUpdate = async (inputs, destination) => {
   // get all the id and properties of already existing objects needed for update.
-  const updateHubspotIds = await getExistingData(inputs, destination);
+  const updateHubspotIds = await getExistingContactsData(inputs, destination);
 
   const resultInput = inputs.map((input) => {
     const { message } = input;
@@ -680,4 +750,7 @@ module.exports = {
   validatePayloadDataTypes,
   getUTCMidnightTimeStampValue,
   populateTraits,
+  getObjectAndIdentifierType,
+  extractIDsForSearchAPI,
+  getRequestDataAndRequestOptions,
 };