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 @@
   SEARCH_LIMIT_VALUE,
   hsCommonConfigJson,
   DESTINATION,
+  MAX_CONTACTS_PER_REQUEST,
 } = require('./config');
 
 const tags = require('../../util/tags');
@@ -464,111 +467,164 @@
 };
 
 /**
- * DOC: https://developers.hubspot.com/docs/api/crm/search
- * @param {*} inputs
- * @param {*} destination
+ * Validates object and identifier type is present in message
+ * @param {*} firstMessage
+ * @returns
  */
-const getExistingData = async (inputs, destination) => {
-  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.');
-    }
-  } else {
-    throw new InstrumentationError('rETL - objectType or identifier type not found. ');
+const getObjectAndIdentifierType = (firstMessage) => {
+  const { objectType, identifierType } = getDestinationExternalIDInfoForRetl(
+    firstMessage,
+    DESTINATION,
+  );
+  if (!objectType || !identifierType) {
+    throw new InstrumentationError('rETL - external Id not found.');
   }
-  inputs.map(async (input) => {
+  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);
-    values.push(destinationExternalId.toString().toLowerCase());
+    return destinationExternalId.toString().toLowerCase();
   });
 
-  values = Array.from(new Set(values));
-  const requestData = {
-    filterGroups: [
-      {
-        filters: [
-          {
-            propertyName: identifierType,
-            values,
-            operator: 'IN',
-          },
-        ],
-      },
-    ],
-    properties: [identifierType],
-    limit: SEARCH_LIMIT_VALUE,
-    after: 0,
-  };
+  return Array.from(new Set(values));
+};
 
-  const requestOptions = {
-    headers: {
-      'Content-Type': JSON_MIME_TYPE,
-      Authorization: `Bearer ${Config.accessToken}`,
-    },
-  };
-  let checkAfter = 1; // variable to keep checking if we have more results
+/**
+ * Returns hubspot records
+ * Ref : https://developers.hubspot.com/docs/api/crm/search
+ * @param {*} data
+ * @param {*} requestOptions
+ * @param {*} objectType
+ * @param {*} identifierType
+ * @param {*} destination
+ * @returns
+ */
+const performHubSpotSearch = async (
+  reqdata,
+  reqOptions,
+  objectType,
+  identifierType,
+  destination,
+) => {
+  let checkAfter = 1;
+  const searchResults = [];
+  const requestData = reqdata;
+  const { Config } = destination;
 
-  /* eslint-disable no-await-in-loop */
+  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 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) {
+    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. ${searchResponse.response?.message}`,
-        searchResponse.status,
+        `rETL - Error during searching object record. ${JSON.stringify(processedResponse.response?.message)}`,
+        processedResponse.status,
         {
-          [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(searchResponse.status),
+          [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(processedResponse.status),
         },
-        searchResponse,
+        processedResponse,
       );
     }
 
-    const after = searchResponse.response?.paging?.next?.after || 0;
-
+    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 = searchResponse.response?.results;
+    const results = processedResponse.response?.results;
     if (results) {
-      updateHubspotIds = results.map((result) => {
-        const propertyValue = result.properties[identifierType];
-        return { id: result.id, property: propertyValue };
-      });
+      searchResults.push(
+        ...results.map((result) => ({
+          id: result.id,
+          property: result.properties[identifierType],
+        })),
+      );
+    }
+  }
+
+  return searchResults;
+};
+
+/**
+ * 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);
+
+  // eslint-disable-next-line no-restricted-syntax
+  for (const chunk of valuesChunk) {
+    const requestData = {
+      filterGroups: [
+        {
+          filters: [
+            {
+              propertyName: identifierType,
+              values: chunk,
+              operator: 'IN',
+            },
+          ],
+        },
+      ],
+      properties: [identifierType],
+      limit: SEARCH_LIMIT_VALUE,
+      after: 0,
+    };
+
+    const requestOptions = {
+      headers: {
+        'Content-Type': JSON_MIME_TYPE,
+        Authorization: `Bearer ${Config.accessToken}`,
+      },
+    };
+    const searchResults = await performHubSpotSearch(
+      requestData,
+      requestOptions,
+      objectType,
+      identifierType,
+      destination,
+    );
+    if (searchResults.length > 0) {
+      updateHubspotIds.push(...searchResults);
     }
   }
   return updateHubspotIds;
@@ -601,7 +657,7 @@
 
 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 +736,6 @@
   validatePayloadDataTypes,
   getUTCMidnightTimeStampValue,
   populateTraits,
+  getObjectAndIdentifierType,
+  extractIDsForSearchAPI,
 };