pull from develop

src/constants/index.js

@@ -50,6 +50,10 @@ const MappedToDestinationKey = 'context.mappedToDestination';
 const GENERIC_TRUE_VALUES = ['true', 'True', 'TRUE', 't', 'T', '1'];
 const GENERIC_FALSE_VALUES = ['false', 'False', 'FALSE', 'f', 'F', '0'];
 
+const HTTP_CUSTOM_STATUS_CODES = {
+  FILTERED: 298,
+};
+
 module.exports = {
   EventType,
   GENERIC_TRUE_VALUES,
@@ -58,4 +62,5 @@ module.exports = {
   SpecedTraits,
   TraitsMapping,
   WhiteListedTraits,
+  HTTP_CUSTOM_STATUS_CODES,
 };

src/controllers/userTransform.ts

@@ -17,7 +17,7 @@ export default class UserTransformController {
     );
     const events = ctx.request.body as ProcessorTransformationRequest[];
     const processedRespone: UserTransformationServiceResponse =
-      await UserTransformService.transformRoutine(events);
+      await UserTransformService.transformRoutine(events, ctx.state.features);
     ctx.body = processedRespone.transformedEvents;
     ControllerUtility.postProcess(ctx, processedRespone.retryStatus);
     logger.debug(

src/middlewares/featureFlag.ts

@@ -0,0 +1,49 @@
+import { Context, Next } from 'koa';
+
+export interface FeatureFlags {
+  [key: string]: boolean | string;
+}
+
+export const FEATURE_FILTER_CODE = 'filter-code';
+
+export default class FeatureFlagMiddleware {
+  public static async handle(ctx: Context, next: Next): Promise<void> {
+    // Initialize ctx.state.features if it doesn't exist
+    ctx.state.features = (ctx.state.features || {}) as FeatureFlags;
+
+    // Get headers from the request
+    const { headers } = ctx.request;
+
+    // Filter headers that start with 'X-Feature-'
+    const featureHeaders = Object.keys(headers).filter((key) =>
+      key.toLowerCase().startsWith('x-feature-'),
+    );
+
+    // Convert feature headers to feature flags in ctx.state.features
+    featureHeaders.forEach((featureHeader) => {
+      // Get the feature name by removing the prefix, and convert to camelCase
+      const featureName = featureHeader
+        .substring(10)
+        .replace(/X-Feature-/g, '')
+        .toLowerCase();
+
+      let value: string | boolean | undefined;
+      const valueString = headers[featureHeader] as string;
+      if (valueString === 'true' || valueString === '?1') {
+        value = true;
+      } else if (valueString === 'false' || valueString === '?0') {
+        value = false;
+      } else {
+        value = valueString;
+      }
+
+      // Set the feature flag in ctx.state.features
+      if (value !== undefined) {
+        ctx.state.features[featureName] = value;
+      }
+    });
+
+    // Move to the next middleware
+    await next();
+  }
+}

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/routes/userTransform.ts

@@ -1,12 +1,14 @@
 import Router from '@koa/router';
 import RouteActivationController from '../middlewares/routeActivation';
+import FeatureFlagController from '../middlewares/featureFlag';
 import UserTransformController from '../controllers/userTransform';
 
 const router = new Router();
 
 router.post(
   '/customTransform',
   RouteActivationController.isUserTransformRouteActive,
+  FeatureFlagController.handle,
   UserTransformController.transform,
 );
 router.post(
@@ -31,4 +33,4 @@ router.post(
 );
 
 const userTransformRoutes = router.routes();
-export default userTransformRoutes;
+export default userTransformRoutes;

src/services/destination/nativeIntegration.ts

@@ -50,15 +50,15 @@ export default class NativeIntegrationDestinationService implements IntegrationD
     events: ProcessorTransformationRequest[],
     destinationType: string,
     version: string,
-    _requestMetadata: NonNullable<unknown>,
+    requestMetadata: NonNullable<unknown>,
   ): Promise<ProcessorTransformationResponse[]> {
     const destHandler = FetchHandler.getDestHandler(destinationType, version);
     const respList: ProcessorTransformationResponse[][] = await Promise.all(
       events.map(async (event) => {
         try {
           const transformedPayloads:
             | ProcessorTransformationOutput
-            | ProcessorTransformationOutput[] = await destHandler.process(event);
+            | ProcessorTransformationOutput[] = await destHandler.process(event, requestMetadata);
           return DestinationPostTransformationService.handleProcessorTransformSucessEvents(
             event,
             transformedPayloads,
@@ -88,7 +88,7 @@ export default class NativeIntegrationDestinationService implements IntegrationD
     events: RouterTransformationRequestData[],
     destinationType: string,
     version: string,
-    _requestMetadata: NonNullable<unknown>,
+    requestMetadata: NonNullable<unknown>,
   ): Promise<RouterTransformationResponse[]> {
     const destHandler = FetchHandler.getDestHandler(destinationType, version);
     const allDestEvents: NonNullable<unknown> = groupBy(
@@ -106,7 +106,7 @@ export default class NativeIntegrationDestinationService implements IntegrationD
         );
         try {
           const doRouterTransformationResponse: RouterTransformationResponse[] =
-            await destHandler.processRouterDest(cloneDeep(destInputArray));
+            await destHandler.processRouterDest(cloneDeep(destInputArray), requestMetadata);
           metaTO.metadata = destInputArray[0].metadata;
           return DestinationPostTransformationService.handleRouterTransformSuccessEvents(
             doRouterTransformationResponse,
@@ -132,7 +132,7 @@ export default class NativeIntegrationDestinationService implements IntegrationD
     events: RouterTransformationRequestData[],
     destinationType: string,
     version: any,
-    _requestMetadata: NonNullable<unknown>,
+    requestMetadata: NonNullable<unknown>,
   ): RouterTransformationResponse[] {
     const destHandler = FetchHandler.getDestHandler(destinationType, version);
     if (!destHandler.batch) {
@@ -145,7 +145,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/services/userTransform.ts

@@ -17,12 +17,16 @@ import { getMetadata, isNonFuncObject } from '../v0/util';
 import { SUPPORTED_FUNC_NAMES } from '../util/ivmFactory';
 import logger from '../logger';
 import stats from '../util/stats';
+import { CommonUtils } from '../util/common';
 // eslint-disable-next-line @typescript-eslint/no-unused-vars
 import { CatchErr, FixMe } from '../util/types';
+import { FeatureFlags, FEATURE_FILTER_CODE } from '../middlewares/featureFlag';
+import { HTTP_CUSTOM_STATUS_CODES } from '../constants';
 
 export default class UserTransformService {
   public static async transformRoutine(
     events: ProcessorTransformationRequest[],
+    features: FeatureFlags = {},
   ): Promise<UserTransformationServiceResponse> {
     let retryStatus = 200;
     const groupedEvents: NonNullable<unknown> = groupBy(
@@ -41,12 +45,13 @@ export default class UserTransformService {
       );
     }
     const responses = await Promise.all<FixMe>(
-      Object.entries(groupedEvents).map(async ([dest, destEvents]) => {
-        logger.debug(`dest: ${dest}`);
+      Object.entries(groupedEvents).map(async ([, destEvents]) => {
         const eventsToProcess = destEvents as ProcessorTransformationRequest[];
         const transformationVersionId =
           eventsToProcess[0]?.destination?.Transformations[0]?.VersionID;
         const messageIds = eventsToProcess.map((ev) => ev.metadata?.messageId);
+        const messageIdsSet = new Set<string>(messageIds);
+        const messageIdsInOutputSet = new Set<string>();
 
         const commonMetadata = {
           sourceId: eventsToProcess[0]?.metadata?.sourceId,
@@ -80,31 +85,55 @@ export default class UserTransformService {
             transformationVersionId,
             librariesVersionIDs,
           );
-          transformedEvents.push(
-            ...destTransformedEvents.map((ev) => {
-              if (ev.error) {
-                return {
-                  statusCode: 400,
-                  error: ev.error,
-                  metadata: isEmpty(ev.metadata) ? commonMetadata : ev.metadata,
-                } as unknown as ProcessorTransformationResponse;
-              }
-              if (!isNonFuncObject(ev.transformedEvent)) {
-                return {
-                  statusCode: 400,
-                  error: `returned event in events from user transformation is not an object. transformationVersionId:${transformationVersionId} and returned event: ${JSON.stringify(
-                    ev.transformedEvent,
-                  )}`,
-                  metadata: isEmpty(ev.metadata) ? commonMetadata : ev.metadata,
-                } as ProcessorTransformationResponse;
-              }
-              return {
-                output: ev.transformedEvent,
+
+          const transformedEventsWithMetadata: ProcessorTransformationResponse[] = [];
+          destTransformedEvents.forEach((ev) => {
+            if (ev.error) {
+              transformedEventsWithMetadata.push({
+                statusCode: 400,
+                error: ev.error,
                 metadata: isEmpty(ev.metadata) ? commonMetadata : ev.metadata,
-                statusCode: 200,
-              } as ProcessorTransformationResponse;
-            }),
-          );
+              } as unknown as ProcessorTransformationResponse);
+              return;
+            }
+            if (!isNonFuncObject(ev.transformedEvent)) {
+              transformedEventsWithMetadata.push({
+                statusCode: 400,
+                error: `returned event in events from user transformation is not an object. transformationVersionId:${transformationVersionId} and returned event: ${JSON.stringify(
+                  ev.transformedEvent,
+                )}`,
+                metadata: isEmpty(ev.metadata) ? commonMetadata : ev.metadata,
+              } as ProcessorTransformationResponse);
+              return;
+            }
+            // add messageId to output set
+            if (ev.metadata?.messageId) {
+              messageIdsInOutputSet.add(ev.metadata.messageId);
+            } else if (ev.metadata?.messageIds) {
+              ev.metadata.messageIds.forEach((id) => messageIdsInOutputSet.add(id));
+            }
+            transformedEventsWithMetadata.push({
+              output: ev.transformedEvent,
+              metadata: isEmpty(ev.metadata) ? commonMetadata : ev.metadata,
+              statusCode: 200,
+            } as ProcessorTransformationResponse);
+          });
+
+          if (features[FEATURE_FILTER_CODE]) {
+            // find difference between input and output messageIds
+            const messageIdsNotInOutput = CommonUtils.setDiff(messageIdsSet, messageIdsInOutputSet);
+            const droppedEvents = messageIdsNotInOutput.map((id) => ({
+              statusCode: HTTP_CUSTOM_STATUS_CODES.FILTERED,
+              metadata: {
+                ...commonMetadata,
+                messageId: id,
+                messageIds: null,
+              },
+            }));
+            transformedEvents.push(...droppedEvents);
+          }
+
+          transformedEvents.push(...transformedEventsWithMetadata);
         } catch (error: CatchErr) {
           logger.error(error);
           let status = 400;

src/util/common.js

@@ -20,6 +20,10 @@ const CommonUtils = {
     }
     return [obj];
   },
+
+  setDiff(mainSet, comparisionSet) {
+    return [...mainSet].filter((item) => !comparisionSet.has(item));
+  },
 };
 
 module.exports = {

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

@@ -22,8 +22,13 @@ const {
   isHttpStatusSuccess,
   simpleProcessRouterDestSync,
   simpleProcessRouterDest,
+  isNewStatusCodesAccepted,
 } = require('../../util');
-const { InstrumentationError, NetworkError } = require('../../util/errorTypes');
+const {
+  InstrumentationError,
+  NetworkError,
+  FilteredEventsError,
+} = require('../../util/errorTypes');
 const {
   ConfigCategory,
   mappingConfig,
@@ -223,7 +228,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 +247,10 @@ function processTrackWithUserAttributes(message, destination, mappingJson, proce
       );
       if (dedupedAttributePayload) {
         requestJson.attributes = [dedupedAttributePayload];
+      } else if (isNewStatusCodesAccepted(reqMetadata)) {
+        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 +459,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 +505,7 @@ async function process(event, processParams = { userStore: new Map() }) {
         destination,
         mappingConfig[category.name],
         processParams,
+        reqMetadata,
       );
       break;
     case EventType.GROUP: