feat: add source id isolation for reverse etl

src/services/destination/cdkV2Integration.ts

@@ -1,7 +1,6 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 /* eslint-disable class-methods-use-this */
 import { TransformationError } from '@rudderstack/integrations-lib';
-import groupBy from 'lodash/groupBy';
 import { processCdkV2Workflow } from '../../cdk/v2/handler';
 import { DestinationService } from '../../interfaces/DestinationService';
 import {
@@ -20,6 +19,7 @@ import {
 } from '../../types/index';
 import stats from '../../util/stats';
 import { CatchErr } from '../../util/types';
+import { groupRouterTransformEvents } from '../../v0/util';
 import tags from '../../v0/util/tags';
 import { DestinationPostTransformationService } from './postTransformation';
 
@@ -112,46 +112,38 @@ export class CDKV2DestinationService implements DestinationService {
     _version: string,
     requestMetadata: NonNullable<unknown>,
   ): Promise<RouterTransformationResponse[]> {
-    const allDestEvents: object = groupBy(
-      events,
-      (ev: RouterTransformationRequestData) => ev.destination?.ID,
-    );
+    const groupedEvents: RouterTransformationRequestData[][] = groupRouterTransformEvents(events);
     const response: RouterTransformationResponse[][] = await Promise.all(
-      Object.values(allDestEvents).map(
-        async (destInputArray: RouterTransformationRequestData[]) => {
-          const metaTo = this.getTags(
-            destinationType,
-            destInputArray[0].metadata.destinationId,
-            destInputArray[0].metadata.workspaceId,
-            tags.FEATURES.ROUTER,
-          );
-          metaTo.metadata = destInputArray[0].metadata;
-          try {
-            const doRouterTransformationResponse: RouterTransformationResponse[] =
-              await processCdkV2Workflow(
-                destinationType,
-                destInputArray,
-                tags.FEATURES.ROUTER,
-                requestMetadata,
-              );
-            return DestinationPostTransformationService.handleRouterTransformSuccessEvents(
-              doRouterTransformationResponse,
-              undefined,
-              metaTo,
-              tags.IMPLEMENTATIONS.CDK_V2,
-              destinationType.toUpperCase(),
+      groupedEvents.map(async (destInputArray: RouterTransformationRequestData[]) => {
+        const metaTo = this.getTags(
+          destinationType,
+          destInputArray[0].metadata.destinationId,
+          destInputArray[0].metadata.workspaceId,
+          tags.FEATURES.ROUTER,
+        );
+        metaTo.metadata = destInputArray[0].metadata;
+        try {
+          const doRouterTransformationResponse: RouterTransformationResponse[] =
+            await processCdkV2Workflow(
+              destinationType,
+              destInputArray,
+              tags.FEATURES.ROUTER,
+              requestMetadata,
             );
-          } catch (error: CatchErr) {
-            metaTo.metadatas = destInputArray.map((input) => input.metadata);
-            const erroredResp =
-              DestinationPostTransformationService.handleRouterTransformFailureEvents(
-                error,
-                metaTo,
-              );
-            return [erroredResp];
-          }
-        },
-      ),
+          return DestinationPostTransformationService.handleRouterTransformSuccessEvents(
+            doRouterTransformationResponse,
+            undefined,
+            metaTo,
+            tags.IMPLEMENTATIONS.CDK_V2,
+            destinationType.toUpperCase(),
+          );
+        } catch (error: CatchErr) {
+          metaTo.metadatas = destInputArray.map((input) => input.metadata);
+          const erroredResp =
+            DestinationPostTransformationService.handleRouterTransformFailureEvents(error, metaTo);
+          return [erroredResp];
+        }
+      }),
     );
     return response.flat();
   }

src/services/destination/nativeIntegration.ts

@@ -26,6 +26,7 @@ import {
 import stats from '../../util/stats';
 import tags from '../../v0/util/tags';
 import { DestinationPostTransformationService } from './postTransformation';
+import { groupRouterTransformEvents } from '../../v0/util';
 
 export class NativeIntegrationDestinationService implements DestinationService {
   public init() {}
@@ -99,11 +100,7 @@ export class NativeIntegrationDestinationService implements DestinationService {
     requestMetadata: NonNullable<unknown>,
   ): Promise<RouterTransformationResponse[]> {
     const destHandler = FetchHandler.getDestHandler(destinationType, version);
-    const allDestEvents: NonNullable<unknown> = groupBy(
-      events,
-      (ev: RouterTransformationRequestData) => ev.destination?.ID,
-    );
-    const groupedEvents: RouterTransformationRequestData[][] = Object.values(allDestEvents);
+    const groupedEvents: RouterTransformationRequestData[][] = groupRouterTransformEvents(events);
     const response: RouterTransformationResponse[][] = await Promise.all(
       groupedEvents.map(async (destInputArray: RouterTransformationRequestData[]) => {
         const metaTO = this.getTags(

src/v0/util/index.js

@@ -2281,15 +2281,37 @@ const validateEventAndLowerCaseConversion = (event, isMandatory, convertToLowerC
   return convertToLowerCase ? event.toString().toLowerCase() : event.toString();
 };
 
-const applyCustomMappings = (message, mappings) =>
-  JsonTemplateEngine.createAsSync(mappings, { defaultPathType: PathType.JSON }).evaluate(message);
+/**
+ * This function applies custom mappings to the event.
+ * @param {*} event The event to be transformed.
+ * @param {*} mappings The custom mappings to be applied.
+ * @returns {object} The transformed event.
+ */
+const applyCustomMappings = (event, mappings) =>
+  JsonTemplateEngine.createAsSync(mappings, { defaultPathType: PathType.JSON }).evaluate(event);
 
 const applyJSONStringTemplate = (message, template) =>
   JsonTemplateEngine.createAsSync(template.replace(/{{/g, '${').replace(/}}/g, '}'), {
     defaultPathType: PathType.JSON,
   }).evaluate(message);
 
 /**
+ * This groups the events by destination ID and source ID.
+ * Note: sourceID is only used for rETL events.
+ * @param {*} events The events to be grouped.
+ * @returns {array} The array of grouped events.
+ */
+const groupRouterTransformEvents = (events) =>
+  Object.values(
+    lodash.groupBy(events, (ev) => [
+      ev.metadata?.destinationId || ev.destination?.ID,
+      ev.metadata?.sourceCategory === 'warehouse'
+        ? ev.metadata?.sourceId || ev.connection?.sourceId
+        : 'default',
+    ]),
+  );
+
+/*
  * Gets url path omitting the hostname & protocol
  *
  * **Note**:
@@ -2364,6 +2386,7 @@ module.exports = {
   getValueFromMessage,
   getValueFromPropertiesOrTraits,
   getValuesAsArrayFromConfig,
+  groupRouterTransformEvents,
   handleSourceKeysOperation,
   hashToSha256,
   isAppleFamily,