chore(release): pull release/v1.45.0 into main (#2712)

rudderlabs · Oct 12, 2023 · 1de7d94 · 1de7d94
2 parents f4c3a6b + a210657
commit 1de7d94
Show file tree

Hide file tree

Showing 77 changed files with 11,436 additions and 9,532 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,26 @@
 
 All notable changes to this project will be documented in this file. See [standard-version](https://github.com/conventional-changelog/standard-version) for commit guidelines.
 
+## [1.45.0](https://github.com/rudderlabs/rudder-transformer/compare/v1.44.2...v1.45.0) (2023-10-11)
+
+
+### Features
+
+* **integrations:** introduced new status codes to suppress or filter events ([#2611](https://github.com/rudderlabs/rudder-transformer/issues/2611)) ([6bdb01e](https://github.com/rudderlabs/rudder-transformer/commit/6bdb01e1152f52339dc6d1936cbe686e64faf070))
+* onboard braze stats ([#2703](https://github.com/rudderlabs/rudder-transformer/issues/2703)) ([f205325](https://github.com/rudderlabs/rudder-transformer/commit/f2053255bedac6017f909fe8b6de3775dd50bc93))
+* support page call in snapchat conversion ([#2688](https://github.com/rudderlabs/rudder-transformer/issues/2688)) ([d681d5e](https://github.com/rudderlabs/rudder-transformer/commit/d681d5e1a39c1914e3a237e1020553518c297416))
+
+
+### Bug Fixes
+
+* added matchId check and timestamp conversion ([#2709](https://github.com/rudderlabs/rudder-transformer/issues/2709)) ([f49244d](https://github.com/rudderlabs/rudder-transformer/commit/f49244dce6cf15812d894d22c43d2432b5a4756a))
+* bugsnag event sent as an object for track call ([#2681](https://github.com/rudderlabs/rudder-transformer/issues/2681)) ([b211840](https://github.com/rudderlabs/rudder-transformer/commit/b21184019691b472650901f6002ef81b579a1146))
+* correct handling of permission_denied error for GAEC & GARL ([#2699](https://github.com/rudderlabs/rudder-transformer/issues/2699)) ([eca3be1](https://github.com/rudderlabs/rudder-transformer/commit/eca3be1fdcb7ac0ebf9432e9686a75fad2d6dc78))
+* **garl:** partial failure handling ([#2666](https://github.com/rudderlabs/rudder-transformer/issues/2666)) ([d4cac26](https://github.com/rudderlabs/rudder-transformer/commit/d4cac268193e0db9540f95c7aab39157d70765c1))
+* **kafka:** add status code & batched value in response structure ([#2684](https://github.com/rudderlabs/rudder-transformer/issues/2684)) ([99f39f5](https://github.com/rudderlabs/rudder-transformer/commit/99f39f5cd62ceb82bd8f51d1de84442b1e59c004))
+* **klaviyo:** sending error field at root level for suppress events ([#2707](https://github.com/rudderlabs/rudder-transformer/issues/2707)) ([bc88f13](https://github.com/rudderlabs/rudder-transformer/commit/bc88f1303ec4f60a466492c72e8c3b57dd5070af))
+* snapchat conversion introduce missing fields ([#2704](https://github.com/rudderlabs/rudder-transformer/issues/2704)) ([d6488d5](https://github.com/rudderlabs/rudder-transformer/commit/d6488d5a189ea90d1142e4ecb34b2af0bb940fe9))
+
 ### [1.44.2](https://github.com/rudderlabs/rudder-transformer/compare/v1.44.1...v1.44.2) (2023-10-09)
 
 

diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "rudder-transformer",
-  "version": "1.44.2",
+  "version": "1.45.0",
   "description": "",
   "homepage": "https://github.com/rudderlabs/rudder-transformer#readme",
   "bugs": {

diff --git a/src/constants/index.js b/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,
 };
diff --git a/src/controllers/userTransform.ts b/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(

diff --git a/src/middlewares/featureFlag.ts b/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();
+  }
+}
diff --git a/src/routes/destination.ts b/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;
diff --git a/src/routes/userTransform.ts b/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;
diff --git a/src/services/destination/nativeIntegration.ts b/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(

diff --git a/src/services/misc.ts b/src/services/misc.ts
@@ -30,6 +30,7 @@ export default class MiscService {
     return {
       namespace: 'Unknown',
       cluster: 'Unknown',
+      features: ctx.state?.features || {},
     };
   }
 

diff --git a/src/services/userTransform.ts b/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;

diff --git a/src/util/common.js b/src/util/common.js
@@ -20,6 +20,10 @@ const CommonUtils = {
     }
     return [obj];
   },
+
+  setDiff(mainSet, comparisionSet) {
+    return [...mainSet].filter((item) => !comparisionSet.has(item));
+  },
 };
 
 module.exports = {

diff --git a/src/util/errorNotifier/bugsnag.js b/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 = [