feat(integrations): introduced new status codes to suppress or filter events

src/routes/destination.ts

@@ -1,6 +1,7 @@
 import Router from '@koa/router';
 import DestinationController from '../controllers/destination';
 import RegulationController from '../controllers/regulation';
+import FeatureFlagController from '../middlewares/featureFlag';
 import RouteActivationController from '../middlewares/routeActivation';
 
 const router = new Router();
@@ -9,22 +10,25 @@ router.post(
   '/:version/destinations/:destination',
   RouteActivationController.isDestinationRouteActive,
   RouteActivationController.destinationProcFilter,
+  FeatureFlagController.handle,
   DestinationController.destinationTransformAtProcessor,
 );
 router.post(
   '/routerTransform',
   RouteActivationController.isDestinationRouteActive,
   RouteActivationController.destinationRtFilter,
+  FeatureFlagController.handle,
   DestinationController.destinationTransformAtRouter,
 );
 router.post(
   '/batch',
   RouteActivationController.isDestinationRouteActive,
   RouteActivationController.destinationBatchFilter,
+  FeatureFlagController.handle,
   DestinationController.batchProcess,
 );
 
 router.post('/deleteUsers', RegulationController.deleteUsers);
 
 const destinationRoutes = router.routes();
-export default destinationRoutes;
+export default destinationRoutes;

src/services/destination/nativeIntegration.ts

@@ -58,7 +58,7 @@ export default class NativeIntegrationDestinationService implements IntegrationD
         try {
           const transformedPayloads:
             | ProcessorTransformationOutput
-            | ProcessorTransformationOutput[] = await destHandler.process(event);
+            | ProcessorTransformationOutput[] = await destHandler.process(event, _requestMetadata);
           return DestinationPostTransformationService.handleProcessorTransformSucessEvents(
             event,
             transformedPayloads,
@@ -106,7 +106,10 @@ export default class NativeIntegrationDestinationService implements IntegrationD
         );
         try {
           const doRouterTransformationResponse: RouterTransformationResponse[] =
-            await destHandler.processRouterDest(cloneDeep(destInputArray));
+            await destHandler.processRouterDest(
+              cloneDeep(destInputArray),
+              cloneDeep(_requestMetadata),
+            );
           metaTO.metadata = destInputArray[0].metadata;
           return DestinationPostTransformationService.handleRouterTransformSuccessEvents(
             doRouterTransformationResponse,
@@ -145,7 +148,10 @@ export default class NativeIntegrationDestinationService implements IntegrationD
     const groupedEvents: RouterTransformationRequestData[][] = Object.values(allDestEvents);
     const response = groupedEvents.map((destEvents) => {
       try {
-        const destBatchedRequests: RouterTransformationResponse[] = destHandler.batch(destEvents);
+        const destBatchedRequests: RouterTransformationResponse[] = destHandler.batch(
+          destEvents,
+          _requestMetadata,
+        );
         return destBatchedRequests;
       } catch (error: any) {
         const metaTO = this.getTags(

src/services/misc.ts

@@ -30,6 +30,7 @@ export default class MiscService {
     return {
       namespace: 'Unknown',
       cluster: 'Unknown',
+      features: ctx?.state?.features || {},
     };
   }
 

src/util/errorNotifier/bugsnag.js

@@ -23,6 +23,7 @@ const {
   UnhandledStatusCodeError,
   UnauthorizedError,
   NetworkInstrumentationError,
+  FilteredEventsError,
 } = require('../../v0/util/errorTypes');
 
 const {
@@ -48,6 +49,7 @@ const errorTypesDenyList = [
   NetworkInstrumentationError,
   CDKCustomError,
   DataValidationError,
+  FilteredEventsError,
 ];
 
 const pathsDenyList = [

src/v0/destinations/braze/transform.js

@@ -23,7 +23,11 @@ const {
   simpleProcessRouterDestSync,
   simpleProcessRouterDest,
 } = require('../../util');
-const { InstrumentationError, NetworkError } = require('../../util/errorTypes');
+const {
+  InstrumentationError,
+  NetworkError,
+  FilteredEventsError,
+} = require('../../util/errorTypes');
 const {
   ConfigCategory,
   mappingConfig,
@@ -39,7 +43,7 @@ const logger = require('../../../logger');
 const { getEndpointFromConfig } = require('./util');
 const { handleHttpRequest } = require('../../../adapters/network');
 const { getDynamicErrorType } = require('../../../adapters/utils/networkUtils');
-const { JSON_MIME_TYPE } = require('../../util/constant');
+const { JSON_MIME_TYPE, FEATURE_FILTER_CODE } = require('../../util/constant');
 
 function formatGender(gender) {
   // few possible cases of woman
@@ -223,7 +227,13 @@ async function processIdentify(message, destination) {
   }
 }
 
-function processTrackWithUserAttributes(message, destination, mappingJson, processParams) {
+function processTrackWithUserAttributes(
+  message,
+  destination,
+  mappingJson,
+  processParams,
+  reqMetadata,
+) {
   let payload = getUserAttributesObject(message, mappingJson);
   if (payload && Object.keys(payload).length > 0) {
     payload = setExternalIdOrAliasObject(payload, message);
@@ -236,6 +246,10 @@ function processTrackWithUserAttributes(message, destination, mappingJson, proce
       );
       if (dedupedAttributePayload) {
         requestJson.attributes = [dedupedAttributePayload];
+      } else if (reqMetadata?.features && reqMetadata.features[FEATURE_FILTER_CODE]) {
+        throw new FilteredEventsError(
+          '[Braze Deduplication]: Duplicate user detected, the user is dropped',
+        );
       } else {
         throw new InstrumentationError(
           '[Braze Deduplication]: Duplicate user detected, the user is dropped',
@@ -444,7 +458,7 @@ function processAlias(message, destination) {
   );
 }
 
-async function process(event, processParams = { userStore: new Map() }) {
+async function process(event, processParams = { userStore: new Map() }, reqMetadata = {}) {
   let response;
   const { message, destination } = event;
   const messageType = message.type.toLowerCase();
@@ -490,6 +504,7 @@ async function process(event, processParams = { userStore: new Map() }) {
         destination,
         mappingConfig[category.name],
         processParams,
+        reqMetadata,
       );
       break;
     case EventType.GROUP:

src/v0/destinations/braze/util.js

@@ -22,7 +22,7 @@ const {
   ALIAS_BRAZE_MAX_REQ_COUNT,
   TRACK_BRAZE_MAX_REQ_COUNT,
 } = require('./config');
-const { JSON_MIME_TYPE } = require('../../util/constant');
+const { JSON_MIME_TYPE, HTTP_STATUS_CODES } = require('../../util/constant');
 const { isObject } = require('../../util');
 const { removeUndefinedValues, getIntegrationsObj } = require('../../util');
 const { InstrumentationError } = require('../../util/errorTypes');
@@ -363,11 +363,14 @@ const processBatch = (transformedEvents) => {
   const purchaseArray = [];
   const successMetadata = [];
   const failureResponses = [];
+  const filteredResponses = [];
   const subscriptionsArray = [];
   const mergeUsersArray = [];
   for (const transformedEvent of transformedEvents) {
     if (!isHttpStatusSuccess(transformedEvent?.statusCode)) {
       failureResponses.push(transformedEvent);
+    } else if (transformedEvent?.statusCode === HTTP_STATUS_CODES.FILTER_EVENTS) {
+      filteredResponses.push(transformedEvent);
     } else if (transformedEvent?.batchedRequest?.body?.JSON) {
       const { attributes, events, purchases, subscription_groups, merge_updates } =
         transformedEvent.batchedRequest.body.JSON;
@@ -446,6 +449,10 @@ const processBatch = (transformedEvents) => {
     finalResponse.push(...failureResponses);
   }
 
+  if (filteredResponses.length > 0) {
+    finalResponse.push(...filteredResponses);
+  }
+
   return finalResponse;
 };
 

src/v0/destinations/klaviyo/transform.js

@@ -35,9 +35,8 @@
   handleRtTfSingleEventError,
   flattenJson,
 } = require('../../util');
-
 const { ConfigurationError, InstrumentationError } = require('../../util/errorTypes');
-const { JSON_MIME_TYPE } = require('../../util/constant');
+const { JSON_MIME_TYPE, HTTP_STATUS_CODES, FEATURE_FILTER_CODE } = require('../../util/constant');
 
 /**
  * Main Identify request handler func
@@ -105,7 +104,11 @@
     },
   };
 
-  const profileId = await getIdFromNewOrExistingProfile(endpoint, payload, requestOptions);
+  const { profileId, response } = await getIdFromNewOrExistingProfile(
+    endpoint,
+    payload,
+    requestOptions,
+  );
 
   // Update Profile
   const responseArray = [profileUpdateResponseBuilder(payload, profileId, category, privateApiKey)];
@@ -115,7 +118,8 @@
     responseArray.push(subscribeUserToList(message, traitsInfo, destination));
     return responseArray;
   }
-  return responseArray[0];
+
+  return { ...responseArray[0], error: JSON.stringify(response) };
 };
 
 // ----------------------
@@ -326,13 +330,35 @@
       }
     }),
   );
-  let batchedSubscribeResponseList = [];
+  const batchedSubscribeResponseList = [];
   if (subscribeRespList.length > 0) {
-    batchedSubscribeResponseList = batchSubscribeEvents(subscribeRespList);
+    const batchedResponseList = batchSubscribeEvents(subscribeRespList, reqMetadata);
+    batchedSubscribeResponseList.push(...batchedResponseList);
   }
-  const nonSubscribeSuccessList = nonSubscribeRespList.map((resp) =>
-    getSuccessRespEvents(resp.message, [resp.metadata], resp.destination),
-  );
+  const nonSubscribeSuccessList = nonSubscribeRespList.map((resp) => {
+    const response = resp;
+    if (
+      reqMetadata?.features &&
+      reqMetadata.features[FEATURE_FILTER_CODE] &&
+      response.message?.error
+    ) {
+      return getSuccessRespEvents(
+        response.message,
+        [response.metadata],
+        response.destination,
+        false,
+        HTTP_STATUS_CODES.SUPPRESS_EVENTS,
+      );
+    }
+
+    if (response.message?.error) {
+      delete response.message?.error;
+      return getSuccessRespEvents(response.message, [response.metadata], response.destination);
+    }
+
+    return getSuccessRespEvents(response.message, [response.metadata], response.destination);
+  });
+
   batchResponseList = [...batchedSubscribeResponseList, ...nonSubscribeSuccessList];
 
   return [...batchResponseList, ...batchErrorRespList];

src/v0/destinations/klaviyo/util.js

@@ -12,14 +12,13 @@
   getSuccessRespEvents,
   defaultPatchRequestConfig,
 } = require('../../util');
-
-const { BASE_ENDPOINT, MAPPING_CONFIG, CONFIG_CATEGORIES, MAX_BATCH_SIZE } = require('./config');
-const { JSON_MIME_TYPE } = require('../../util/constant');
-const { NetworkError, InstrumentationError } = require('../../util/errorTypes');
-const { getDynamicErrorType } = require('../../../adapters/utils/networkUtils');
 const tags = require('../../util/tags');
 const { handleHttpRequest } = require('../../../adapters/network');
+const { JSON_MIME_TYPE, FEATURE_FILTER_CODE } = require('../../util/constant');
+const { NetworkError, InstrumentationError } = require('../../util/errorTypes');
+const { getDynamicErrorType } = require('../../../adapters/utils/networkUtils');
 const { client: errNotificationClient } = require('../../../util/errorNotifier');
+const { BASE_ENDPOINT, MAPPING_CONFIG, CONFIG_CATEGORIES, MAX_BATCH_SIZE } = require('./config');
 
 const REVISION_CONSTANT = '2023-02-22';
 
@@ -35,6 +34,7 @@
  * @returns
  */
 const getIdFromNewOrExistingProfile = async (endpoint, payload, requestOptions) => {
+  let response;
   let profileId;
   const endpointPath = '/api/profiles';
   const { processedResponse: resp } = await handleHttpRequest(
@@ -48,15 +48,19 @@
       endpointPath,
     },
   );
+
   if (resp.status === 201) {
     profileId = resp.response?.data?.id;
+    const { data } = resp.response;
+    response = { id: data.id, attributes: data.attributes };
   } else if (resp.status === 409) {
     const { errors } = resp.response;
     profileId = errors?.[0]?.meta?.duplicate_profile_id;
+    response = errors;
   }
 
   if (profileId) {
-    return profileId;
+    return { profileId, response };
   }
 
   let statusCode = resp.status;
@@ -80,6 +84,14 @@
   );
 };
 
+/**
+ * Update profile response builder
+ * @param {*} payload
+ * @param {*} profileId
+ * @param {*} category
+ * @param {*} privateApiKey
+ * @returns
+ */
 const profileUpdateResponseBuilder = (payload, profileId, category, privateApiKey) => {
   const updatedPayload = payload;
   const identifyResponse = defaultRequestConfig();
@@ -238,15 +250,15 @@
  * @param {*} subscribeResponseList
  * @returns
  */
-const groupSubsribeResponsesUsingListId = (subscribeResponseList) => {
+const groupSubscribeResponsesUsingListId = (subscribeResponseList) => {
   const subscribeEventGroups = lodash.groupBy(
     subscribeResponseList,
     (event) => event.message.body.JSON.data.attributes.list_id,
   );
   return subscribeEventGroups;
 };
 
-const getBatchedResponseList = (subscribeEventGroups, identifyResponseList) => {
+const getBatchedResponseList = (subscribeEventGroups, identifyResponseList, reqMetadata) => {
   let batchedResponseList = [];
   Object.keys(subscribeEventGroups).forEach((listId) => {
     // eventChunks = [[e1,e2,e3,..batchSize],[e1,e2,e3,..batchSize]..]
@@ -262,16 +274,21 @@
     });
     batchedResponseList = [...batchedResponseList, ...batchedResponse];
   });
-  identifyResponseList.forEach((response) => {
-    batchedResponseList[0].batchedRequest.push(response);
-  });
+
+  if (!(reqMetadata?.features && reqMetadata.features[FEATURE_FILTER_CODE])) {
+    identifyResponseList.forEach((response) => {
+      batchedResponseList[0].batchedRequest.push(response);
+    });
+  }
+
   return batchedResponseList;
 };
 
-const batchSubscribeEvents = (subscribeRespList) => {
+const batchSubscribeEvents = (subscribeRespList, reqMetadata) => {
   const identifyResponseList = [];
   subscribeRespList.forEach((event) => {
     const processedEvent = event;
+    // for group and identify events (it will contain only subscribe response)
     if (processedEvent.message.length === 2) {
       // the array will contain one update profile reponse and one subscribe reponse
       identifyResponseList.push(event.message[0]);
@@ -282,9 +299,13 @@
     }
   });
 
-  const subscribeEventGroups = groupSubsribeResponsesUsingListId(subscribeRespList);
+  const subscribeEventGroups = groupSubscribeResponsesUsingListId(subscribeRespList);
 
-  const batchedResponseList = getBatchedResponseList(subscribeEventGroups, identifyResponseList);
+  const batchedResponseList = getBatchedResponseList(
+    subscribeEventGroups,
+    identifyResponseList,
+    reqMetadata,
+  );
 
   return batchedResponseList;
 };
@@ -295,6 +316,6 @@
   populateCustomFieldsFromTraits,
   generateBatchedPaylaodForArray,
   batchSubscribeEvents,
-  getIdFromNewOrExistingProfile,
   profileUpdateResponseBuilder,
+  getIdFromNewOrExistingProfile,
 };