feat: update FBCA spec to support VDM v2 events

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/types/index.ts

@@ -160,6 +160,7 @@ type RouterTransformationRequestData = {
   message: object;
   metadata: Metadata;
   destination: Destination;
+  connection?: Record<string, unknown>;
 };
 
 type RouterTransformationRequest = {

src/v0/destinations/fb_custom_audience/config.js

@@ -4,6 +4,8 @@ function getEndPoint(audienceId) {
   return `${BASE_URL}/${audienceId}/users`;
 }
 
+const VDM_V2_SCHEMA_VERSION = '1.1';
+
 const schemaFields = [
   'EXTERN_ID',
   'EMAIL',
@@ -105,4 +107,5 @@ module.exports = {
   typeFields,
   subTypeFields,
   maxPayloadSize,
+  VDM_V2_SCHEMA_VERSION,
 };

src/v0/destinations/fb_custom_audience/recordTransform.js

@@ -2,7 +2,7 @@
 const lodash = require('lodash');
 const get = require('get-value');
 const { InstrumentationError, ConfigurationError } = require('@rudderstack/integrations-lib');
-const { schemaFields } = require('./config');
+const { schemaFields, VDM_V2_SCHEMA_VERSION } = require('./config');
 const { MappedToDestinationKey } = require('../../../constants');
 const stats = require('../../../util/stats');
 const {
@@ -17,6 +17,7 @@ const {
   ensureApplicableFormat,
   getUpdatedDataElement,
   getSchemaForEventMappedToDest,
+  getSchemaForEventMappedToDestForVDMv2,
   batchingWithPayloadSize,
   responseBuilderSimple,
   getDataSource,
@@ -100,18 +101,18 @@ const processRecordEventArray = (
 };
 
 async function processRecordInputs(groupedRecordInputs) {
-  const { destination } = groupedRecordInputs[0];
+  const { destination, connection } = groupedRecordInputs[0];
   const { message } = groupedRecordInputs[0];
-  const {
-    isHashRequired,
-    accessToken,
-    disableFormat,
-    type,
-    subType,
-    isRaw,
-    maxUserCount,
-    audienceId,
-  } = destination.Config;
+  const { accessToken, disableFormat, type, subType, isRaw, maxUserCount } = destination.Config;
+  let audienceId;
+  let isHashRequired;
+  if (connection?.config?.destination?.schemaVersion === VDM_V2_SCHEMA_VERSION) {
+    audienceId = connection?.config?.destination?.audienceId;
+    isHashRequired = connection?.config?.destination?.isHashRequired;
+  } else {
+    audienceId = destination.Config?.audienceId;
+    isHashRequired = destination.Config?.isHashRequired;
+  }
   const prepareParams = {
     access_token: accessToken,
   };
@@ -125,7 +126,7 @@ async function processRecordInputs(groupedRecordInputs) {
   // audience id validation
   let operationAudienceId = audienceId;
   const mappedToDestination = get(message, MappedToDestinationKey);
-  if (mappedToDestination) {
+  if (mappedToDestination && !operationAudienceId) {
     const { objectType } = getDestinationExternalIDInfoForRetl(message, 'FB_CUSTOM_AUDIENCE');
     operationAudienceId = objectType;
   }
@@ -136,7 +137,11 @@ async function processRecordInputs(groupedRecordInputs) {
   // user schema validation
   let { userSchema } = destination.Config;
   if (mappedToDestination) {
-    userSchema = getSchemaForEventMappedToDest(message);
+    if (connection?.config?.destination?.schemaVersion === VDM_V2_SCHEMA_VERSION) {
+      userSchema = getSchemaForEventMappedToDestForVDMv2(message);
+    } else {
+      userSchema = getSchemaForEventMappedToDest(message);
+    }
   }
   if (!Array.isArray(userSchema)) {
     userSchema = [userSchema];

src/v0/destinations/fb_custom_audience/transform.js

@@ -12,16 +12,16 @@ const {
 const {
   prepareDataField,
   getSchemaForEventMappedToDest,
+  getSchemaForEventMappedToDestForVDMv2,
   batchingWithPayloadSize,
   generateAppSecretProof,
   responseBuilderSimple,
   getDataSource,
 } = require('./util');
-const { schemaFields, USER_ADD, USER_DELETE } = require('./config');
+const { schemaFields, USER_ADD, USER_DELETE, VDM_V2_SCHEMA_VERSION } = require('./config');
 
 const { MappedToDestinationKey } = require('../../../constants');
 const { processRecordInputs } = require('./recordTransform');
-const logger = require('../../../logger');
 
 function checkForUnsupportedEventTypes(dictionary, keyList) {
   const unsupportedEventTypes = [];
@@ -154,11 +154,20 @@ const prepareToSendEvents = (
   });
   return toSendEvents;
 };
-const processEvent = (message, destination) => {
+const processEvent = (message, destination, connection) => {
   const respList = [];
   let toSendEvents = [];
   let { userSchema } = destination.Config;
-  const { isHashRequired, audienceId, maxUserCount } = destination.Config;
+  const { maxUserCount } = destination.Config;
+  let audienceId;
+  let isHashRequired;
+  if (connection?.config?.destination?.schemaVersion === VDM_V2_SCHEMA_VERSION) {
+    audienceId = connection?.config?.destination?.audienceId;
+    isHashRequired = connection?.config?.destination?.isHashRequired;
+  } else {
+    audienceId = destination.Config?.audienceId;
+    isHashRequired = destination.Config?.isHashRequired;
+  }
   if (!message.type) {
     throw new InstrumentationError('Message Type is not present. Aborting message.');
   }
@@ -182,7 +191,11 @@ const processEvent = (message, destination) => {
 
   // If mapped to destination, use the mapped fields instead of destination userschema
   if (mappedToDestination) {
-    userSchema = getSchemaForEventMappedToDest(message);
+    if (connection?.config?.destination?.schemaVersion === VDM_V2_SCHEMA_VERSION) {
+      userSchema = getSchemaForEventMappedToDestForVDMv2(message);
+    } else {
+      userSchema = getSchemaForEventMappedToDest(message);
+    }
   }
 
   // When one single schema field is added in the webapp, it does not appear to be an array
@@ -236,7 +249,7 @@ const processEvent = (message, destination) => {
   return respList;
 };
 
-const process = (event) => processEvent(event.message, event.destination);
+const process = (event) => processEvent(event.message, event.destination, event.connection);
 
 const processRouterDest = async (inputs, reqMetadata) => {
   const respList = [];
@@ -247,7 +260,6 @@ const processRouterDest = async (inputs, reqMetadata) => {
   const eventTypes = ['record', 'audiencelist'];
   const unsupportedEventList = checkForUnsupportedEventTypes(groupedInputs, eventTypes);
   if (unsupportedEventList.length > 0) {
-    logger.info(`unsupported events found ${unsupportedEventList}`);
     throw new ConfigurationError('unsupported events present in the event');
   }
 

src/v0/destinations/fb_custom_audience/util.js

@@ -61,6 +61,18 @@ const getSchemaForEventMappedToDest = (message) => {
   return userSchema;
 };
 
+const getSchemaForEventMappedToDestForVDMv2 = (message) => {
+  const mappedSchema = message.fields;
+  if (!mappedSchema) {
+    throw new InstrumentationError(
+      'event.fields is required property for events mapped to destination ',
+    );
+  }
+  let userSchema = Object.keys(mappedSchema);
+  userSchema = userSchema.map((field) => field.trim());
+  return userSchema;
+};
+
 // function responsible to ensure the user inputs are passed according to the allowed format
 const ensureApplicableFormat = (userProperty, userInformation) => {
   let updatedProperty;
@@ -261,6 +273,7 @@ const responseBuilderSimple = (payload, audienceId) => {
 module.exports = {
   prepareDataField,
   getSchemaForEventMappedToDest,
+  getSchemaForEventMappedToDestForVDMv2,
   batchingWithPayloadSize,
   ensureApplicableFormat,
   getUpdatedDataElement,

src/v0/util/index.js

@@ -2265,10 +2265,30 @@ 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);
 
 /**
+ * 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.metadata?.sourceCategory === 'warehouse' ? ev.metadata?.sourceId : 'default',
+    ]),
+  );
+
+/*
  * Gets url path omitting the hostname & protocol
  *
  * **Note**:
@@ -2342,6 +2362,7 @@ module.exports = {
   getValueFromMessage,
   getValueFromPropertiesOrTraits,
   getValuesAsArrayFromConfig,
+  groupRouterTransformEvents,
   handleSourceKeysOperation,
   hashToSha256,
   isAppleFamily,