feat: cm360 router batching

src/services/destination/postTransformation.ts

@@ -16,6 +16,15 @@ import { ErrorReportingService } from '../errorReporting';
 import tags from '../../v0/util/tags';
 import stats from '../../util/stats';
 
+type ErrorResponse = {
+  status?: number;
+  message?: string;
+  destinationResponse?: object;
+  statTags?: object;
+  authErrorCategory?: string | undefined;
+  response?: object | undefined;
+};
+
 export class DestinationPostTransformationService {
   public static handleProcessorTransformSucessEvents(
     event: ProcessorTransformationRequest,
@@ -139,7 +148,7 @@ export class DestinationPostTransformationService {
   }
 
   public static handleDeliveryFailureEvents(
-    error: NonNullable<unknown>,
+    error: ErrorResponse,
     metaTo: MetaTransferObject,
   ): DeliveryResponse {
     const errObj = generateErrorObject(error, metaTo.errorDetails, false);
@@ -152,6 +161,12 @@ export class DestinationPostTransformationService {
         authErrorCategory: errObj.authErrorCategory,
       }),
     } as DeliveryResponse;
+
+    // for transformer-proxy to maintain contract
+    const { response } = error;
+    if (response) {
+      resp.response = response;
+    }
     ErrorReportingService.reportError(error, metaTo.errorContext, resp);
     return resp;
   }

src/types/index.ts

@@ -140,6 +140,7 @@ type DeliveryResponse = {
   destinationResponse: any;
   statTags: object;
   authErrorCategory?: string;
+  response?: object;
 };
 
 enum MessageType {

src/v0/destinations/am/transform.js

@@ -643,7 +643,7 @@ const processSingleMessage = (message, destination) => {
                       .trim()
                       .replaceAll(/{{([^{}]+)}}/g, get(message, getMessagePath));
         }
-        evType =useUserDefinedScreenEventName ? customScreenEv : eventType;
+        evType = customScreenEv || eventType;
         message.properties = updatedProperties;
         category = ConfigCategory.SCREEN;
       }

src/v0/destinations/campaign_manager/config.js

@@ -9,6 +9,8 @@ const ConfigCategories = {
   },
 };
 
+const MAX_BATCH_CONVERSATIONS_SIZE = 1000;
+
 const EncryptionEntityType = [
   'ENCRYPTION_ENTITY_TYPE_UNKNOWN',
   'DCM_ACCOUNT',
@@ -28,4 +30,5 @@ module.exports = {
   BASE_URL,
   EncryptionEntityType,
   EncryptionSource,
+  MAX_BATCH_CONVERSATIONS_SIZE,
 };

src/v0/destinations/campaign_manager/networkHandler.js

@@ -1,30 +1,150 @@
 const { AbortedError, RetryableError, NetworkError } = require('@rudderstack/integrations-lib');
+/* eslint-disable no-param-reassign */
+/* eslint-disable no-restricted-syntax */
 const { prepareProxyRequest, proxyRequest } = require('../../../adapters/network');
 const { isHttpStatusSuccess, getAuthErrCategoryFromStCode } = require('../../util/index');
 
 const {
   processAxiosResponse,
   getDynamicErrorType,
 } = require('../../../adapters/utils/networkUtils');
+const { TransformerProxyError } = require('../../util/errorTypes');
 const tags = require('../../util/tags');
 
-function checkIfFailuresAreRetryable(response) {
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+function checkIfFailuresAreRetryable(response, proxyOutputObj) {
+  const { status } = response;
   try {
-    if (Array.isArray(response.status) && Array.isArray(response.status[0].errors)) {
-      return (
-        response.status[0].errors[0].code !== 'PERMISSION_DENIED' &&
-        response.status[0].errors[0].code !== 'INVALID_ARGUMENT'
-      );
+    if (Array.isArray(status)) {
+      // iterate over each status, and if found retryable in conversations ..retry else discard
+      /* status : [{
+        "conversion": {
+          object (Conversion)
+        },
+        "errors": [
+          {
+            object (ConversionError)
+          }
+        ],
+        "kind": string
+      }] */
+      for (const st of status) {
+        for(const err of st.errors) {
+          // if code is any of these, event is not retryable
+          if (
+            err.code === 'PERMISSION_DENIED' ||
+            err.code === 'INVALID_ARGUMENT' ||
+            err.code === 'NOT_FOUND'
+          ) {
+            return false;
+          }
+        }
+      }
     }
     return true;
   } catch (e) {
-    return true;
+    return false;
+  }
+}
+
+function isEventRetryable(element, proxyOutputObj) {
+  let flag = false;
+  let errorMsg = "";
+  // success event
+  if (!element.errors) {
+    return flag;
+  }
+  for(const err of element.errors) {
+    errorMsg += `${err.message}, `;
+    if (err.code === 'INTERNAL') {
+      flag = true;
+    }
+  }
+  if (errorMsg) {
+    proxyOutputObj.error = errorMsg;
   }
+  return flag;
+}
+
+function isEventAbortable(element, proxyOutputObj) {
+  let flag = false;
+  let errorMsg = "";
+  // success event
+  if (!element.errors) {
+    return flag;
+  }
+  for(const err of element.errors) {
+    errorMsg += `${err.message}, `;
+    // if code is any of these, event is not retryable
+    if (
+      err.code === 'PERMISSION_DENIED' ||
+      err.code === 'INVALID_ARGUMENT' ||
+      err.code === 'NOT_FOUND'
+    ) {
+      flag = true;
+    }
+  }
+  if (errorMsg) {
+    proxyOutputObj.error = errorMsg;
+  }
+  return flag;
 }
 
 const responseHandler = (destinationResponse) => {
   const message = `[CAMPAIGN_MANAGER Response Handler] - Request Processed Successfully`;
-  const { response, status } = destinationResponse;
+  const responseWithIndividualEvents = [];
+  const { response, status, rudderJobMetadata } = destinationResponse;
+
+  if (Array.isArray(rudderJobMetadata)) {
+    if (isHttpStatusSuccess(status)) {
+      // check for Partial Event failures and Successes 
+      const destPartialStatus = response.status;
+
+      for (const [idx, element] of destPartialStatus.entries()) {
+        const proxyOutputObj = {
+          statusCode: 200,
+          metadata: rudderJobMetadata[idx],
+          error: "success"
+        };
+        // update status of partial event as per retriable or abortable
+        if (isEventRetryable(element, proxyOutputObj)) {
+          proxyOutputObj.statusCode = 500;
+        } else if (isEventAbortable(element, proxyOutputObj)) { 
+          proxyOutputObj.statusCode = 400;
+        }
+        responseWithIndividualEvents.push(proxyOutputObj);
+      }
+
+      return {
+        status,
+        message,
+        destinationResponse,
+        response: responseWithIndividualEvents
+      }
+    }
+
+    // in case of failure status, populate response to maintain len(metadata)=len(response)
+    const errorMessage = response.error?.message || 'unknown error format';
+    for (const metadata of rudderJobMetadata) {
+      responseWithIndividualEvents.push({
+        statusCode: 500,
+        metadata,
+        error: errorMessage
+      });
+    }
+
+    throw new TransformerProxyError(
+      `Campaign Manager: Error proxy during CAMPAIGN_MANAGER response transformation`,
+      500,
+      {
+        [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
+      },
+      destinationResponse,
+      getAuthErrCategoryFromStCode(status),
+      responseWithIndividualEvents
+    );
+  }
+
   if (isHttpStatusSuccess(status)) {
     // check for Failures
     if (response.hasFailures === true) {
@@ -60,6 +180,7 @@ const responseHandler = (destinationResponse) => {
     destinationResponse,
     getAuthErrCategoryFromStCode(status),
   );
+
 };
 
 function networkHandler() {

src/v0/destinations/campaign_manager/transform.js

@@ -1,13 +1,16 @@
 const { InstrumentationError } = require('@rudderstack/integrations-lib');
+const lodash = require('lodash');
 const { EventType } = require('../../../constants');
-
 const {
   constructPayload,
   defaultRequestConfig,
   defaultPostRequestConfig,
+  defaultBatchRequestConfig,
   removeUndefinedAndNullValues,
+  getSuccessRespEvents,
   isDefinedAndNotNull,
-  simpleProcessRouterDest,
+  checkInvalidRtTfEvents,
+  handleRtTfSingleEventError,
   getAccessToken,
 } = require('../../util');
 
@@ -17,6 +20,7 @@ const {
   BASE_URL,
   EncryptionEntityType,
   EncryptionSource,
+  MAX_BATCH_CONVERSATIONS_SIZE,
 } = require('./config');
 
 const { convertToMicroseconds } = require('./util');
@@ -178,9 +182,110 @@ function process(event) {
   return response;
 }
 
+const generateBatch = (eventKind, events) => {
+  const batchRequestObject = defaultBatchRequestConfig();
+  const conversions = [];
+  let encryptionInfo = {};
+  const metadata = [];
+  // extracting destination, message from the first event in a batch
+  const { destination, message } = events[0];
+  // Batch event into dest batch structure
+  events.forEach((ev) => {
+    conversions.push(...ev.message.body.JSON.conversions);
+    metadata.push(ev.metadata);
+    if (ev.message.body.JSON.encryptionInfo) {
+      encryptionInfo = ev.message.body.JSON.encryptionInfo;
+    }
+  });
+
+  batchRequestObject.batchedRequest.body.JSON = {
+    kind: eventKind,
+    conversions,
+  };
+
+  if (Object.keys(encryptionInfo).length > 0) {
+    batchRequestObject.batchedRequest.body.JSON.encryptionInfo = encryptionInfo;
+  }
+
+  batchRequestObject.batchedRequest.endpoint = message.endpoint;
+
+  batchRequestObject.batchedRequest.headers = message.headers;
+
+  return {
+    ...batchRequestObject,
+    metadata,
+    destination,
+  };
+};
+
+const batchEvents = (eventChunksArray) => {
+  const batchedResponseList = [];
+
+  // group batchInsert and batchUpdate payloads
+  const groupedEventChunks = lodash.groupBy(
+    eventChunksArray,
+    (event) => event.message.body.JSON.kind,
+  );
+  Object.keys(groupedEventChunks).forEach((eventKind) => {
+    // eventChunks = [[e1,e2,e3,..batchSize],[e1,e2,e3,..batchSize]..]
+    const eventChunks = lodash.chunk(groupedEventChunks[eventKind], MAX_BATCH_CONVERSATIONS_SIZE);
+    eventChunks.forEach((chunk) => {
+      const batchEventResponse = generateBatch(eventKind, chunk);
+      batchedResponseList.push(
+        getSuccessRespEvents(
+          batchEventResponse.batchedRequest,
+          batchEventResponse.metadata,
+          batchEventResponse.destination,
+          true,
+        ),
+      );
+    });
+  });
+  return batchedResponseList;
+};
+
 const processRouterDest = async (inputs, reqMetadata) => {
-  const respList = await simpleProcessRouterDest(inputs, process, reqMetadata);
-  return respList;
+  const errorRespEvents = checkInvalidRtTfEvents(inputs);
+  if (errorRespEvents.length > 0) {
+    return errorRespEvents;
+  }
+
+  const batchErrorRespList = [];
+  const eventChunksArray = [];
+  const { destination } = inputs[0];
+  await Promise.all(
+    inputs.map(async (event) => {
+      try {
+        if (event.message.statusCode) {
+          // already transformed event
+          eventChunksArray.push({
+            message: event.message,
+            metadata: event.metadata,
+            destination,
+          });
+        } else {
+          // if not transformed
+          const proccessedRespList = process(event);
+          const transformedPayload = {
+            message: proccessedRespList,
+            metadata: event.metadata,
+            destination,
+          };
+          eventChunksArray.push(transformedPayload);
+        }
+      } catch (error) {
+        const errRespEvent = handleRtTfSingleEventError(event, error, reqMetadata);
+        batchErrorRespList.push(errRespEvent);
+      }
+    }),
+  );
+
+  let batchResponseList = [];
+  if (eventChunksArray.length > 0) {
+    batchResponseList = batchEvents(eventChunksArray);
+  }
+
+  return [...batchResponseList, ...batchErrorRespList];
 };
 
 module.exports = { process, processRouterDest };

src/v0/util/errorTypes/index.js

@@ -1,5 +1,7 @@
+const TransformerProxyError = require('./transformerProxyError');
 const FilteredEventsError = require('./filteredEventsError');
 
 module.exports = {
+  TransformerProxyError,
   FilteredEventsError,
 };