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

src/v0/destinations/hs/transform.js

@@ -19,6 +19,7 @@
   fetchFinalSetOfTraits,
   getProperties,
   validateDestinationConfig,
+  convertToResponseFormat,
 } = require('./util');
 
 const processSingleMessage = async ({ message, destination, metadata }, propertyMap) => {
@@ -147,20 +148,38 @@
     }),
   );
 
-  if (successRespList.length > 0) {
+  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
+  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
+    dontBatchEvents: convertToResponseFormat(dontBatchTrueResponses),
+  };
 };
 // we are batching by default at routerTransform
 const processRouterDest = async (inputs, reqMetadata) => {
   const tempNewInputs = batchEventsInOrder(inputs);
   const batchedResponseList = [];
   const errorRespList = [];
+  const dontBatchEvents = [];
   const promises = tempNewInputs.map(async (inputEvents) => {
     const response = await processBatchRouter(inputEvents, reqMetadata);
     return response;
@@ -171,8 +190,10 @@
   results.forEach((response) => {
     errorRespList.push(...response.errorRespList);
     batchedResponseList.push(...response.batchedResponseList);
+    dontBatchEvents.push(...response.dontBatchEvents);
   });
-  return [...batchedResponseList, ...errorRespList];
+  console.log(JSON.stringify([...batchedResponseList, ...errorRespList, ...dontBatchEvents]));
+  return [...batchedResponseList, ...errorRespList, ...dontBatchEvents];
 };
 
 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,
@@ -844,6 +847,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,
@@ -863,4 +894,5 @@ module.exports = {
   getObjectAndIdentifierType,
   extractIDsForSearchAPI,
   getRequestData,
+  convertToResponseFormat,
 };

src/v1/destinations/hs/networkHandler.ts

@@ -0,0 +1,147 @@
+import { TransformerProxyError } from '../../../v0/util/errorTypes';
+import { prepareProxyRequest, proxyRequest } from '../../../adapters/network';
+import { isHttpStatusSuccess, getAuthErrCategoryFromStCode } from '../../../v0/util/index';
+import { DeliveryV1Response, DeliveryJobState } from '../../../types/index';
+
+import { processAxiosResponse, getDynamicErrorType } from '../../../adapters/utils/networkUtils';
+
+const tags = require('../../../v0/util/tags');
+
+/**
+ *
+ * @param results
+ * @param rudderJobMetadata
+ * @param destinationConfig
+ * @returns boolean
+ */
+
+const findFeatureandVersion = (response, rudderJobMetadata, destinationConfig) => {
+  const { results, errors } = response;
+  if (Array.isArray(rudderJobMetadata) && rudderJobMetadata.length === 1) {
+    return 'singleEvent';
+  }
+  if (destinationConfig?.apiVersion === 'legacyApi') {
+    return 'legacyApiWithMultipleEvents';
+  }
+  if (destinationConfig?.apiVersion === 'newApi') {
+    if (Array.isArray(results) && results.length === rudderJobMetadata.length)
+      return 'newApiWithMultipleEvents';
+
+    if (
+      Array.isArray(results) &&
+      results.length !== rudderJobMetadata.length &&
+      results.length + errors.length === rudderJobMetadata.length
+    )
+      return 'newApiWithMultipleEventsAndErrors';
+  }
+  return 'unableToFindVersionWithMultipleEvents';
+};
+
+const populateResponseWithDontBatch = (rudderJobMetadata, response) => {
+  const errorMessage = JSON.stringify(response);
+  const responseWithIndividualEvents: DeliveryJobState[] = [];
+
+  rudderJobMetadata.forEach((metadata) => {
+    responseWithIndividualEvents.push({
+      statusCode: 500,
+      metadata: { ...metadata, dontBatch: true },
+      error: errorMessage,
+    });
+  });
+  return responseWithIndividualEvents;
+};
+
+type Response = {
+  status?: string;
+  results?: Array<object>;
+  errors?: Array<object>;
+  startedAt?: Date;
+  completedAt?: Date;
+  message?: string;
+  correlationId?: string;
+  failureMessages?: Array<object>;
+};
+
+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: DeliveryJobState[] = [];
+  const { response, status } = destinationResponse;
+
+  if (isHttpStatusSuccess(status)) {
+    // populate different response for each event
+    const destResponse = response as Response;
+    let proxyOutputObj: DeliveryJobState;
+    const featureAndVersion = findFeatureandVersion(
+      destResponse,
+      rudderJobMetadata,
+      destinationRequest?.destinationConfig,
+    );
+    switch (featureAndVersion) {
+      case 'singleEvent':
+        proxyOutputObj = {
+          statusCode: status,
+          metadata: rudderJobMetadata[0],
+          error: JSON.stringify(destResponse),
+        };
+        responseWithIndividualEvents.push(proxyOutputObj);
+        break;
+      case 'newApiWithMultipleEvents':
+        rudderJobMetadata.forEach((metadata: any, index: string | number) => {
+          proxyOutputObj = {
+            statusCode: 200,
+            metadata,
+            error: JSON.stringify(destResponse.results?.[index]),
+          };
+          responseWithIndividualEvents.push(proxyOutputObj);
+        });
+        break;
+      default:
+        rudderJobMetadata.forEach((metadata) => {
+          proxyOutputObj = {
+            statusCode: 200,
+            metadata,
+            error: 'success',
+          };
+          responseWithIndividualEvents.push(proxyOutputObj);
+        });
+        break;
+    }
+    return {
+      status,
+      message: successMessage,
+      response: responseWithIndividualEvents,
+    } as DeliveryV1Response;
+  }
+
+  // At least one event in the batch is invalid.
+  if (status === 400 && Array.isArray(rudderJobMetadata) && rudderJobMetadata.length > 1) {
+    // sending back 500 for retry only when events came in a batch
+    return {
+      status: 500,
+      message: failureMessage,
+      response: populateResponseWithDontBatch(rudderJobMetadata, response),
+    } as DeliveryV1Response;
+  }
+  throw new TransformerProxyError(
+    failureMessage,
+    status,
+    {
+      [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
+    },
+    destinationResponse,
+    getAuthErrCategoryFromStCode(status),
+    response,
+  );
+};
+
+function networkHandler(this: any) {
+  this.prepareProxy = prepareProxyRequest;
+  this.proxy = proxyRequest;
+  this.processAxiosResponse = processAxiosResponse;
+  this.responseHandler = responseHandler;
+}
+
+module.exports = { networkHandler };