chore: add feature flag support for filtered status code

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

@@ -20,10 +20,13 @@ 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(
@@ -42,8 +45,7 @@ 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;
@@ -117,18 +119,19 @@ export default class UserTransformService {
             } as ProcessorTransformationResponse);
           });
 
-          // TODO: add feature flag based on rudder-server version to do this
-          // find difference between input and output messageIds
-          const messageIdsNotInOutput = CommonUtils.setDiff(messageIdsSet, messageIdsInOutputSet);
-          const droppedEvents = messageIdsNotInOutput.map((id) => ({
-            statusCode: 298,
-            metadata: {
-              ...commonMetadata,
-              messageId: id,
-              messageIds: null,
-            },
-          }));
-          transformedEvents.push(...droppedEvents);
+          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) {

test/__tests__/user_transformation_ts.test.ts

@@ -1,5 +1,6 @@
-import UserTransformService from '../../src/services/userTransform';
 import fetch from 'node-fetch';
+import UserTransformService from '../../src/services/userTransform';
+import { FeatureFlags, FEATURE_FILTER_CODE } from '../../src/middlewares/featureFlag';
 jest.mock('node-fetch', () => jest.fn());
 
 const integration = 'user_transformation_service';
@@ -29,7 +30,9 @@ describe('User Transform Service', () => {
       status: 200,
       json: jest.fn().mockResolvedValue(respBody),
     });
-    const output = await UserTransformService.transformRoutine(inputData);
+    const output = await UserTransformService.transformRoutine(inputData, {
+      [FEATURE_FILTER_CODE]: true,
+    } as FeatureFlags);
     expect(fetch).toHaveBeenCalledWith(
       `https://api.rudderlabs.com/transformation/getByVersionId?versionId=${versionId}`,
     );