feat: refactor router transformer - 4

src/v0/destinations/fb_custom_audience/transform.js

@@ -12,7 +12,8 @@ const {
   returnArrayOfSubarrays,
   isDefinedAndNotNull,
   flattenMap,
-  generateErrorObject
+  generateErrorObject,
+  getEventReqMetadata
 } = require("../../util");
 
 const {
@@ -31,6 +32,10 @@ const {
   ConfigurationError
 } = require("../../util/errorTypes");
 
+const {
+  client: errNotificationClient
+} = require("../../../util/errorNotifier");
+
 const getSchemaForEventMappedToDest = message => {
   const mappedSchema = get(message, "context.destinationFields");
   if (!mappedSchema) {
@@ -401,7 +406,7 @@ const process = event => {
   return processEvent(event.message, event.destination);
 };
 
-const processRouterDest = inputs => {
+const processRouterDest = (inputs, reqMetadata) => {
   if (!Array.isArray(inputs) || inputs.length <= 0) {
     const respEvents = getErrorRespEvents(null, 400, "Invalid event array");
     return [respEvents];
@@ -427,12 +432,21 @@ const processRouterDest = inputs => {
       return responseList;
     } catch (error) {
       const errObj = generateErrorObject(error);
-      return getErrorRespEvents(
+      const resp = getErrorRespEvents(
         [input.metadata],
-        400,
-        error.message || "Error occurred while processing the payload.",
+        errObj.status,
+        errObj.message,
         errObj.statTags
       );
+      errNotificationClient.notify(
+        error,
+        "Router Transformation (event level)",
+        {
+          ...resp,
+          ...reqMetadata,
+          ...getEventReqMetadata(input)
+        }
+      );
     }
   });
   return flattenMap(respList);

src/v0/destinations/googlepubsub/transform.js

@@ -1,4 +1,4 @@
-const { getSuccessRespEvents, getErrorRespEvents } = require("../../util");
+const { simpleProcessRouterDest } = require("../../util");
 const { ConfigurationError } = require("../../util/errorTypes");
 
 const { getTopic, createAttributesMetadata } = require("./util");
@@ -19,34 +19,13 @@ function process(event) {
   throw new ConfigurationError("No topic set for this event");
 }
 
-const processRouterDest = async inputs => {
-  if (!Array.isArray(inputs) || inputs.length <= 0) {
-    const respEvents = getErrorRespEvents(null, 400, "Invalid event array");
-    return [respEvents];
-  }
-
-  const respList = await Promise.all(
-    inputs.map(async input => {
-      try {
-        return getSuccessRespEvents(
-          await process(input),
-          [input.metadata],
-          input.destination
-        );
-      } catch (error) {
-        return getErrorRespEvents(
-          [input.metadata],
-          error.response
-            ? error.response.status
-            : error.code
-            ? error.code
-            : 400,
-          error.message || "Error occurred while processing payload."
-        );
-      }
-    })
+const processRouterDest = async (inputs, reqMetadata) => {
+  const respList = await simpleProcessRouterDest(
+    inputs,
+    "GOOGLEPUBSNUB",
+    process,
+    reqMetadata
   );
   return respList;
 };
-
 module.exports = { process, processRouterDest };

src/v0/destinations/googlesheets/transform.js

@@ -117,7 +117,7 @@ const process = event => {
 };
 
 // Router transform with batching by default
-const processRouterDest = async inputs => {
+const processRouterDest = async (inputs, reqMetadata) => {
   const successRespList = [];
   const errorRespList = [];
   const errorRespEvents = checkInvalidRtTfEvents(inputs, "GOOGLESHEETS");
@@ -137,7 +137,7 @@ const processRouterDest = async inputs => {
         const errRespEvent = handleRtTfSingleEventError(
           input,
           error,
-          "GOOGLESHEETS"
+          reqMetadata
         );
         errorRespList.push(errRespEvent);
       }

src/v0/destinations/heap/transform.js

@@ -69,8 +69,13 @@ const processEvent = (message, destination) => {
 const process = async event => {
   return processEvent(event.message, event.destination);
 };
-const processRouterDest = async inputs => {
-  const respList = await simpleProcessRouterDest(inputs, "HEAP", process);
+const processRouterDest = async (inputs, reqMetadata) => {
+  const respList = await simpleProcessRouterDest(
+    inputs,
+    "HEAP",
+    process,
+    reqMetadata
+  );
   return respList;
 };
 

src/v0/destinations/hs/transform.js

@@ -85,7 +85,7 @@ const process = async event => {
 };
 
 // we are batching by default at routerTransform
-const processRouterDest = async inputs => {
+const processRouterDest = async (inputs, reqMetadata) => {
   const errorRespEvents = checkInvalidRtTfEvents(inputs, DESTINATION);
   if (errorRespEvents.length > 0) {
     return errorRespEvents;
@@ -164,7 +164,7 @@ const processRouterDest = async inputs => {
         const errRespEvent = handleRtTfSingleEventError(
           input,
           error,
-          DESTINATION
+          reqMetadata
         );
         errorRespList.push(errRespEvent);
       }

src/v0/destinations/indicative/transform.js

@@ -6,9 +6,7 @@ const {
   getFieldValueFromMessage,
   constructPayload,
   defaultPostRequestConfig,
-  getSuccessRespEvents,
-  getErrorRespEvents,
-  generateErrorObject
+  simpleProcessRouterDest
 } = require("../../util");
 const {
   InstrumentationError,
@@ -199,43 +197,12 @@ const process = event => {
   return processEvent(event.message, event.destination);
 };
 
-const processRouterDest = async inputs => {
-  if (!Array.isArray(inputs) || inputs.length <= 0) {
-    const respEvents = getErrorRespEvents(null, 400, "Invalid event array");
-    return [respEvents];
-  }
-
-  const respList = await Promise.all(
-    inputs.map(async input => {
-      try {
-        if (input.message.statusCode) {
-          // already transformed event
-          return getSuccessRespEvents(
-            input.message,
-            [input.metadata],
-            input.destination
-          );
-        }
-        // if not transformed
-        return getSuccessRespEvents(
-          await process(input),
-          [input.metadata],
-          input.destination
-        );
-      } catch (error) {
-        const errObj = generateErrorObject(error);
-        return getErrorRespEvents(
-          [input.metadata],
-          error.response
-            ? error.response.status
-            : error.code
-            ? error.code
-            : 400,
-          error.message || "Error occurred while processing payload.",
-          errObj.statTags
-        );
-      }
-    })
+const processRouterDest = async (inputs, reqMetadata) => {
+  const respList = await simpleProcessRouterDest(
+    inputs,
+    "INDICATIVE",
+    process,
+    reqMetadata
   );
   return respList;
 };

src/v0/destinations/intercom/transform.js

@@ -13,15 +13,10 @@ const {
   defaultRequestConfig,
   defaultPostRequestConfig,
   getFieldValueFromMessage,
-  getSuccessRespEvents,
-  getErrorRespEvents,
   addExternalIdToTraits,
-  generateErrorObject
+  simpleProcessRouterDest
 } = require("../../util");
-const {
-  TransformationError,
-  InstrumentationError
-} = require("../../util/errorTypes");
+const { InstrumentationError } = require("../../util/errorTypes");
 
 function getCompanyAttribute(company) {
   const companiesList = [];
@@ -175,43 +170,12 @@ function process(event) {
   return response;
 }
 
-const processRouterDest = async inputs => {
-  if (!Array.isArray(inputs) || inputs.length <= 0) {
-    const respEvents = getErrorRespEvents(null, 400, "Invalid event array");
-    return [respEvents];
-  }
-
-  const respList = await Promise.all(
-    inputs.map(async input => {
-      try {
-        if (input.message.statusCode) {
-          // already transformed event
-          return getSuccessRespEvents(
-            input.message,
-            [input.metadata],
-            input.destination
-          );
-        }
-        // if not transformed
-        return getSuccessRespEvents(
-          await process(input),
-          [input.metadata],
-          input.destination
-        );
-      } catch (error) {
-        const erroObj = generateErrorObject(error);
-        return getErrorRespEvents(
-          [input.metadata],
-          error.response
-            ? error.response.status
-            : error.code
-            ? error.code
-            : 400,
-          error.message || "Error occurred while processing payload.",
-          erroObj.statTags
-        );
-      }
-    })
+const processRouterDest = async (inputs, reqMetadata) => {
+  const respList = await simpleProcessRouterDest(
+    inputs,
+    "INTERCOM",
+    process,
+    reqMetadata
   );
   return respList;
 };

src/v0/destinations/iterable/transform.js

@@ -490,7 +490,7 @@ function getEventChunks(
   }
 }
 
-const processRouterDest = async inputs => {
+const processRouterDest = async (inputs, reqMetadata) => {
   const errorRespEvents = checkInvalidRtTfEvents(inputs, "ITERABLE");
   if (errorRespEvents.length > 0) {
     return errorRespEvents;
@@ -528,7 +528,7 @@ const processRouterDest = async inputs => {
         const errRespEvent = handleRtTfSingleEventError(
           event,
           error,
-          "ITERABLE"
+          reqMetadata
         );
         errorRespList.push(errRespEvent);
       }

src/v0/destinations/june/transform.js

@@ -99,8 +99,13 @@ const process = event => {
   return processEvent(event.message, event.destination);
 };
 
-const processRouterDest = async inputs => {
-  const respList = await simpleProcessRouterDest(inputs, "JUNE", process);
+const processRouterDest = async (inputs, reqMetadata) => {
+  const respList = await simpleProcessRouterDest(
+    inputs,
+    "JUNE",
+    process,
+    reqMetadata
+  );
   return respList;
 };
 

src/v0/destinations/keen/transform.js

@@ -8,8 +8,7 @@ const {
   defaultRequestConfig,
   getParsedIP,
   getFieldValueFromMessage,
-  getSuccessRespEvents,
-  getErrorRespEvents
+  simpleProcessRouterDest
 } = require("../../util");
 const { InstrumentationError } = require("../../util/errorTypes");
 const { ENDPOINT } = require("./config");
@@ -148,37 +147,12 @@ function process(event) {
   return response;
 }
 
-const processRouterDest = async inputs => {
-  if (!Array.isArray(inputs) || inputs.length <= 0) {
-    const respEvents = getErrorRespEvents(null, 400, "Invalid event array");
-    return [respEvents];
-  }
-
-  const respList = await Promise.all(
-    inputs.map(async input => {
-      try {
-        if (input.message.statusCode) {
-          // already transformed event
-          return getSuccessRespEvents(
-            input.message,
-            [input.metadata],
-            input.destination
-          );
-        }
-        // if not transformed
-        return getSuccessRespEvents(
-          await process(input),
-          [input.metadata],
-          input.destination
-        );
-      } catch (error) {
-        return getErrorRespEvents(
-          [input.metadata],
-          error.response?.status || error.code || 400,
-          error.message || "Error occurred while processing payload."
-        );
-      }
-    })
+const processRouterDest = async (inputs, reqMetadata) => {
+  const respList = await simpleProcessRouterDest(
+    inputs,
+    "KEEN",
+    process,
+    reqMetadata
   );
   return respList;
 };

test/__tests__/data/clickup_router_output.json

@@ -227,6 +227,10 @@
         "jobId": 4
       }
     ],
+    "statTags": {
+      "errorCategory": "dataValidation",
+      "errorType": "instrumentation"
+    },
     "statusCode": 400
   }
 ]