feat: cm360 batching support

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/data/CampaignManagerTrackConfig.json

@@ -13,10 +13,7 @@
     "destKey": "timestampMicros",
     "sourceKeys": "timestamp",
     "sourceFromGenericMap": true,
-    "required": true,
-    "metadata": {
-      "type": "microSecondTimestamp"
-    }
+    "required": true
   },
   {
     "destKey": "floodlightActivityId",

src/v0/destinations/campaign_manager/networkHandler.js

@@ -9,14 +9,36 @@
 const tags = require('../../util/tags');
 
 function checkIfFailuresAreRetryable(response) {
+  const { status } = response;
+  let isRetryable = true;
   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) {
+        st.errors.forEach(err => {
+          // if code is any of these, event is not retryable
+          if (
+            err.code === 'PERMISSION_DENIED' ||
+            err.code === 'INVALID_ARGUMENT' ||
+            err.code === 'NOT_FOUND'
+          ) {
+            isRetryable = false;
+          }
+        })
+      }
     }
-    return true;
+    return isRetryable;
   } catch (e) {
     return true;
   }

src/v0/destinations/campaign_manager/transform.js

@@ -1,12 +1,15 @@
+const lodash = require('lodash');
 const { EventType } = require('../../../constants');
-
 const {
   constructPayload,
   defaultRequestConfig,
   defaultPostRequestConfig,
+  defaultBatchRequestConfig,
   removeUndefinedAndNullValues,
+  getSuccessRespEvents,
   isDefinedAndNotNull,
-  simpleProcessRouterDest,
+  checkInvalidRtTfEvents,
+  handleRtTfSingleEventError,
   getAccessToken,
 } = require('../../util');
 
@@ -16,6 +19,7 @@
   BASE_URL,
   EncryptionEntityType,
   EncryptionSource,
+  MAX_BATCH_CONVERSATIONS_SIZE,
 } = require('./config');
 
 const { InstrumentationError } = require('../../util/errorTypes');
@@ -72,7 +76,28 @@
     delete requestJson.childDirectedTreatment;
     delete requestJson.limitAdTracking;
   }
-  requestJson.timestampMicros = requestJson.timestampMicros.toString();
+
+  // for handling when input is timestamp as string
+  const numTimestamp = /^\d+$/.test(requestJson.timestampMicros);
+  if (numTimestamp) {
+    // is digit only, below convert string timestamp to numeric
+    requestJson.timestampMicros *= 1;
+  }
+
+  // 2022-10-11T05:453:90.ZZ
+  // 16483423423423423
+  const date = new Date(requestJson.timestampMicros);
+  let unixTimestamp = date.getTime();
+  // Date, moment both are not able to distinguish input if it is second,millisecond or microsecond unix timestamp
+  // Using count of digits to distinguish between these 3, 9999999999999 (13 digits) means Nov 20 2286 which is long far in future
+  if (unixTimestamp.toString().length === 13) {
+    // milliseconds
+    unixTimestamp *= 1000;
+  } else if (unixTimestamp.toString().length === 10) {
+    // seconds
+    unixTimestamp *= 1000000;
+  }
+  requestJson.timestampMicros = unixTimestamp.toString();
 
   const encryptionInfo = {};
   // prepare encrptionInfo if encryptedUserId or encryptedUserIdCandidates is given
@@ -138,35 +163,17 @@
     );
   }
 
-  let count = 0;
-
-  if (response.body.JSON.conversions[0].gclid) {
-    count += 1;
-  }
-
-  if (response.body.JSON.conversions[0].dclid) {
-    count += 1;
-  }
-
-  if (response.body.JSON.conversions[0].encryptedUserId) {
-    count += 1;
-  }
-
-  if (response.body.JSON.conversions[0].encryptedUserIdCandidates) {
-    count += 1;
-  }
-
-  if (response.body.JSON.conversions[0].mobileDeviceId) {
-    count += 1;
-  }
-
-  if (response.body.JSON.conversions[0].impressionId) {
-    count += 1;
-  }
-
-  if (count !== 1) {
+  if (
+    !response.body.JSON.conversions[0].gclid &&
+    !response.body.JSON.conversions[0].matchId &&
+    !response.body.JSON.conversions[0].dclid &&
+    !response.body.JSON.conversions[0].encryptedUserId &&
+    !response.body.JSON.conversions[0].encryptedUserIdCandidates &&
+    !response.body.JSON.conversions[0].mobileDeviceId &&
+    !response.body.JSON.conversions[0].impressionId
+  ) {
     throw new InstrumentationError(
-      '[CAMPAIGN MANAGER (DCM)]: For CM360 we need one of encryptedUserId,encryptedUserIdCandidates, matchId, mobileDeviceId, gclid, dclid, impressionId.',
+      '[CAMPAIGN MANAGER (DCM)]: Atleast one of encryptedUserId,encryptedUserIdCandidates, matchId, mobileDeviceId, gclid, dclid, impressionId.',
     );
   }
 }
@@ -194,9 +201,110 @@
   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 };

test/__tests__/campaign_manager.test.js

@@ -28,16 +28,15 @@ describe(`${name} Tests`, () => {
     testData.forEach((dataPoint, index) => {
       it(`${index}. ${integration} - ${dataPoint.description}`, async () => {
         try {
-          let output = await transformer.process(dataPoint.input);
-          delete output.body.JSON.idempotency;
+          const output = await transformer.process(dataPoint.input);
           expect(output).toEqual(dataPoint.output);
         } catch (error) {
           expect(error.message).toEqual(dataPoint.output.error);
         }
       });
     });
   });
-
+  
   describe("Router Tests", () => {
     it("Payload", async () => {
       const routerOutput = await transformer.processRouterDest(inputRouterData);