chore: return 298 for filtered/dropped events in user transformer

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 @@
     );
     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

@@ -17,12 +17,16 @@
 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 @@
       );
     }
     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 @@
             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 = {