feat: move hubspot to transformer proxy to enable partial batch handling

src/v0/destinations/hs/transform.js

@@ -15,6 +15,7 @@ const {
   fetchFinalSetOfTraits,
   getProperties,
   validateDestinationConfig,
+  convertToResponseFormat,
 } = require('./util');
 
 const processSingleMessage = async (message, destination, propertyMap) => {
@@ -137,16 +138,32 @@ const processRouterDest = async (inputs, reqMetadata) => {
     }),
   );
 
+  const dontBatchTrueResponses = [];
+  const dontBatchFalseOrUndefinedResponses = [];
+  // segregating successRepList depending on dontbatch value
+  successRespList.forEach((successResp) => {
+    if (successResp.metadata?.dontBatch) {
+      dontBatchTrueResponses.push(successResp);
+    } else {
+      dontBatchFalseOrUndefinedResponses.push(successResp);
+    }
+  });
+
   // batch implementation
   let batchedResponseList = [];
-  if (successRespList.length > 0) {
+  if (dontBatchFalseOrUndefinedResponses.length > 0) {
     if (destination.Config.apiVersion === API_VERSION.v3) {
-      batchedResponseList = batchEvents(successRespList);
+      batchedResponseList = batchEvents(dontBatchFalseOrUndefinedResponses);
     } else {
-      batchedResponseList = legacyBatchEvents(successRespList);
+      batchedResponseList = legacyBatchEvents(dontBatchFalseOrUndefinedResponses);
     }
   }
-  return [...batchedResponseList, ...errorRespList];
+  return [
+    ...batchedResponseList,
+    ...errorRespList,
+    // if there are any events where dontbatch set to true we need to update them according to the response format
+    ...convertToResponseFormat(dontBatchTrueResponses),
+  ];
 };
 
 module.exports = { process, processRouterDest };

src/v0/destinations/hs/util.js

@@ -22,6 +22,9 @@ const {
   getValueFromMessage,
   isNull,
   validateEventName,
+  defaultBatchRequestConfig,
+  defaultPostRequestConfig,
+  getSuccessRespEvents,
 } = require('../../util');
 const {
   CONTACT_PROPERTY_MAP_ENDPOINT,
@@ -837,6 +840,34 @@ const addExternalIdToHSTraits = (message) => {
   set(getFieldValueFromMessage(message, 'traits'), externalIdObj.identifierType, externalIdObj.id);
 };
 
+const convertToResponseFormat = (successRespListWithDontBatchTrue) => {
+  const response = [];
+  if (Array.isArray(successRespListWithDontBatchTrue)) {
+    successRespListWithDontBatchTrue.forEach((event) => {
+      const { message, metadata, destination } = event;
+      const endpoint = get(message, 'endpoint');
+
+      const batchedResponse = defaultBatchRequestConfig();
+      batchedResponse.batchedRequest.headers = message.headers;
+      batchedResponse.batchedRequest.endpoint = endpoint;
+      batchedResponse.batchedRequest.body = message.body;
+      batchedResponse.batchedRequest.params = message.params;
+      batchedResponse.batchedRequest.method = defaultPostRequestConfig.requestMethod;
+      batchedResponse.metadata = [metadata];
+      batchedResponse.destination = destination;
+
+      response.push(
+        getSuccessRespEvents(
+          batchedResponse.batchedRequest,
+          batchedResponse.metadata,
+          batchedResponse.destination,
+        ),
+      );
+    });
+  }
+  return response;
+};
+
 module.exports = {
   validateDestinationConfig,
   addExternalIdToHSTraits,
@@ -856,4 +887,5 @@ module.exports = {
   getObjectAndIdentifierType,
   extractIDsForSearchAPI,
   getRequestData,
+  convertToResponseFormat,
 };

src/v1/destinations/hs/networkHandler.js

@@ -0,0 +1,104 @@
+/* eslint-disable no-param-reassign */
+/* eslint-disable no-restricted-syntax */
+const { TransformerProxyError } = require('../../../v0/util/errorTypes');
+const { prepareProxyRequest, proxyRequest } = require('../../../adapters/network');
+const { isHttpStatusSuccess, getAuthErrCategoryFromStCode } = require('../../../v0/util/index');
+
+const {
+  processAxiosResponse,
+  getDynamicErrorType,
+} = require('../../../adapters/utils/networkUtils');
+const tags = require('../../../v0/util/tags');
+
+const verify = (results, rudderJobMetadata, destinationConfig) => {
+  if (destinationConfig?.apiVersion === undefined) {
+    return false;
+  }
+  if (destinationConfig?.apiVersion === 'newApi') {
+    return Array.isArray(results) && results.length === rudderJobMetadata.length;
+  }
+  return true;
+};
+
+const populateResponseWithDontBatch = (rudderJobMetadata, status, response) => {
+  const errorMessage = JSON.stringify(response);
+  const responseWithIndividualEvents = [];
+  for (const metadata of rudderJobMetadata) {
+    metadata.dontBatch = true;
+    responseWithIndividualEvents.push({
+      statusCode: 500,
+      metadata,
+      error: errorMessage,
+    });
+  }
+  return responseWithIndividualEvents;
+};
+const responseHandler = (responseParams) => {
+  const { destinationResponse, rudderJobMetadata, destinationRequest } = responseParams;
+  const successMessage = `[HUBSPOT Response V1 Handler] - Request Processed Successfully`;
+  const failureMessage =
+    'HUBSPOT: Error in transformer proxy v1 during HUBSPOT response transformation';
+  const responseWithIndividualEvents = [];
+  const { response, status } = destinationResponse;
+
+  if (isHttpStatusSuccess(status)) {
+    // populate different response for each event
+    const results = response?.results;
+    if (verify(results, rudderJobMetadata, destinationRequest?.destinationConfig)) {
+      for (const metadata of rudderJobMetadata) {
+        const proxyOutputObj = {
+          statusCode: 200,
+          metadata,
+          error: 'success',
+        };
+        responseWithIndividualEvents.push(proxyOutputObj);
+      }
+      return {
+        status,
+        message: successMessage,
+        destinationResponse,
+        response: responseWithIndividualEvents,
+      };
+    }
+    // return the destiantionResponse as it is when the response is not in expected format
+    throw new TransformerProxyError(
+      failureMessage,
+      status,
+      {
+        [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
+      },
+      destinationResponse,
+      getAuthErrCategoryFromStCode(status),
+    );
+  }
+
+  // At least one event in the batch is invalid.
+  if (status === 400 && rudderJobMetadata.length > 1) {
+    // sending back 500 for retry only when events came in a batch
+    return {
+      status: 500,
+      message: failureMessage,
+      destinationResponse,
+      response: populateResponseWithDontBatch(rudderJobMetadata, status, response),
+    };
+  }
+  throw new TransformerProxyError(
+    failureMessage,
+    status,
+    {
+      [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
+    },
+    destinationResponse,
+    getAuthErrCategoryFromStCode(status),
+    responseWithIndividualEvents,
+  );
+};
+
+function networkHandler() {
+  this.prepareProxy = prepareProxyRequest;
+  this.proxy = proxyRequest;
+  this.processAxiosResponse = processAxiosResponse;
+  this.responseHandler = responseHandler;
+}
+
+module.exports = { networkHandler };