feat: changes for supporting record event in FB audience

src/v0/destinations/fb_custom_audience/transform.js

@@ -1,14 +1,7 @@
 const lodash = require('lodash');
 const get = require('get-value');
+const { InstrumentationError, ConfigurationError } = require('@rudderstack/integrations-lib');
 const {
-  InstrumentationError,
-  TransformationError,
-  ConfigurationError,
-} = require('@rudderstack/integrations-lib');
-const {
-  defaultRequestConfig,
-  defaultPostRequestConfig,
-  defaultDeleteRequestConfig,
   checkSubsetOfArray,
   isDefinedAndNotNullAndNotEmpty,
   returnArrayOfSubarrays,
@@ -21,36 +14,10 @@
   getSchemaForEventMappedToDest,
   batchingWithPayloadSize,
 } = require('./util');
-const {
-  getEndPoint,
-  schemaFields,
-  USER_ADD,
-  USER_DELETE,
-  typeFields,
-  subTypeFields,
-} = require('./config');
+const { schemaFields, USER_ADD, USER_DELETE, typeFields, subTypeFields } = require('./config');
 
 const { MappedToDestinationKey } = require('../../../constants');
-
-const responseBuilderSimple = (payload, audienceId) => {
-  if (payload) {
-    const responseParams = payload.responseField;
-    const response = defaultRequestConfig();
-    response.endpoint = getEndPoint(audienceId);
-
-    if (payload.operationCategory === 'add') {
-      response.method = defaultPostRequestConfig.requestMethod;
-    }
-    if (payload.operationCategory === 'remove') {
-      response.method = defaultDeleteRequestConfig.requestMethod;
-    }
-
-    response.params = responseParams;
-    return response;
-  }
-  // fail-safety for developer error
-  throw new TransformationError(`Payload could not be constructed`);
-};
+const { processRecordInputs, responseBuilderSimple } = require('./transformV2');
 
 // Function responsible prepare the payload field of every event parameter
 
@@ -243,6 +210,7 @@
       ),
     );
   }
+
   toSendEvents.forEach((sendEvent) => {
     respList.push(responseBuilderSimple(sendEvent, operationAudienceId));
   });
@@ -258,7 +226,24 @@
 const process = (event) => processEvent(event.message, event.destination);
 
 const processRouterDest = async (inputs, reqMetadata) => {
-  const respList = await simpleProcessRouterDest(inputs, process, reqMetadata);
+  const respList = [];
+  const groupedInputs = lodash.groupBy(inputs, (input) => input.message.type?.toLowerCase());
+  let transformedRecordEvent = [];
+  let transformedAudienceEvent = [];
+
+  if (groupedInputs.record) {
+    transformedRecordEvent = await processRecordInputs(groupedInputs.record, reqMetadata);
+  }
+
+  if (groupedInputs.audiencelist) {
+    transformedAudienceEvent = await simpleProcessRouterDest(
+      groupedInputs.audiencelist,
+      process,
+      reqMetadata,
+    );
+  }
+
+  respList.push(...transformedRecordEvent, ...transformedAudienceEvent);
   return flattenMap(respList);
 };
 

src/v0/destinations/fb_custom_audience/transformV2.js

@@ -0,0 +1,249 @@
+/* eslint-disable no-const-assign */
+const lodash = require('lodash');
+const get = require('get-value');
+const {
+  InstrumentationError,
+  ConfigurationError,
+  TransformationError,
+} = require('@rudderstack/integrations-lib');
+const { schemaFields, typeFields, subTypeFields, getEndPoint } = require('./config');
+const { MappedToDestinationKey } = require('../../../constants');
+const stats = require('../../../util/stats');
+const {
+  getDestinationExternalIDInfoForRetl,
+  isDefinedAndNotNullAndNotEmpty,
+  checkSubsetOfArray,
+  defaultRequestConfig,
+  returnArrayOfSubarrays,
+  defaultPostRequestConfig,
+  defaultDeleteRequestConfig,
+} = require('../../util');
+const {
+  ensureApplicableFormat,
+  getUpdatedDataElement,
+  getSchemaForEventMappedToDest,
+  batchingWithPayloadSize,
+} = require('./util');
+
+const responseBuilderSimple = (payload, audienceId) => {
+  if (payload) {
+    const responseParams = payload.responseField;
+    const response = defaultRequestConfig();
+    response.endpoint = getEndPoint(audienceId);
+
+    if (payload.operationCategory === 'add') {
+      response.method = defaultPostRequestConfig.requestMethod;
+    }
+    if (payload.operationCategory === 'remove') {
+      response.method = defaultDeleteRequestConfig.requestMethod;
+    }
+
+    response.params = responseParams;
+    return response;
+  }
+  // fail-safety for developer error
+  throw new TransformationError(`Payload could not be constructed`);
+};
+
+const processRecordEventArray = (
+  recordChunksArray,
+  userSchema,
+  isHashRequired,
+  disableFormat,
+  paramsPayload,
+  prepareParams,
+  destination,
+  operation,
+) => {
+  const toSendEvents = [];
+  recordChunksArray.forEach((recordArray) => {
+    const data = [];
+    recordArray.forEach((input) => {
+      const { fields } = input.message;
+      let dataElement = [];
+      let nullUserData = true;
+
+      userSchema.forEach((eachProperty) => {
+        const userProperty = fields[eachProperty];
+        let updatedProperty = userProperty;
+
+        if (isHashRequired && !disableFormat) {
+          updatedProperty = ensureApplicableFormat(eachProperty, userProperty);
+        }
+
+        dataElement = getUpdatedDataElement(
+          dataElement,
+          isHashRequired,
+          eachProperty,
+          updatedProperty,
+        );
+
+        if (dataElement[dataElement.length - 1]) {
+          nullUserData = false;
+        }
+      });
+
+      if (nullUserData) {
+        stats.increment('fb_custom_audience_event_having_all_null_field_values_for_a_user', {
+          destinationId: destination.ID,
+          nullFields: userSchema,
+        });
+      }
+
+      data.push(dataElement);
+    });
+
+    const prepareFinalPayload = lodash.cloneDeep(paramsPayload);
+    prepareFinalPayload.schema = userSchema;
+    prepareFinalPayload.data = data;
+    const payloadBatches = batchingWithPayloadSize(prepareFinalPayload);
+
+    payloadBatches.forEach((payloadBatch) => {
+      const response = {
+        ...prepareParams,
+        payload: payloadBatch,
+      };
+      const wrappedResponse = {
+        responseField: response,
+        operationCategory: operation,
+      };
+      toSendEvents.push(wrappedResponse);
+    });
+  });
+
+  return toSendEvents;
+};
+
+async function processRecordInputs(groupedRecordInputs) {
+  const { destination } = groupedRecordInputs[0];
+  const { message } = groupedRecordInputs[0];
+  const { isHashRequired, accessToken, disableFormat, type, subType, isRaw, maxUserCount } =
+    destination.Config;
+  const prepareParams = {
+    access_token: accessToken,
+  };
+
+  // maxUserCount validation
+  const maxUserCountNumber = parseInt(maxUserCount, 10);
+  if (Number.isNaN(maxUserCountNumber)) {
+    throw new ConfigurationError('Batch size must be an Integer.');
+  }
+
+  // audience id validation
+  const { audienceId } = destination.Config;
+  let operationAudienceId = audienceId;
+  const mappedToDestination = get(message, MappedToDestinationKey);
+  if (!operationAudienceId && mappedToDestination) {
+    const { objectType } = getDestinationExternalIDInfoForRetl(message, 'FB_CUSTOM_AUDIENCE');
+    operationAudienceId = objectType;
+  }
+  if (!isDefinedAndNotNullAndNotEmpty(operationAudienceId)) {
+    throw new ConfigurationError('Audience ID is a mandatory field');
+  }
+
+  // user schema validation
+  let { userSchema } = destination.Config;
+  if (mappedToDestination) {
+    userSchema = getSchemaForEventMappedToDest(message);
+  }
+  if (!Array.isArray(userSchema)) {
+    userSchema = [userSchema];
+  }
+  if (!checkSubsetOfArray(schemaFields, userSchema)) {
+    throw new ConfigurationError('One or more of the schema fields are not supported');
+  }
+
+  const paramsPayload = {};
+  const dataSource = {};
+  if (isRaw) {
+    paramsPayload.is_raw = isRaw;
+  }
+  if (type && type !== 'NA' && typeFields.includes(type)) {
+    dataSource.type = type;
+  }
+  if (subType && subType !== 'NA' && subTypeFields.includes(subType)) {
+    dataSource.sub_type = subType;
+  }
+  if (Object.keys(dataSource).length > 0) {
+    paramsPayload.data_source = dataSource;
+  }
+
+  const groupedRecordsByAction = lodash.groupBy(groupedRecordInputs, (record) =>
+    record.message.action?.toLowerCase(),
+  );
+
+  const toSendEvents = [];
+
+  if (groupedRecordsByAction.delete) {
+    const deleteRecordChunksArray = returnArrayOfSubarrays(
+      groupedRecordsByAction.delete,
+      maxUserCountNumber,
+    );
+    toSendEvents.push(
+      ...processRecordEventArray(
+        deleteRecordChunksArray,
+        userSchema,
+        isHashRequired,
+        disableFormat,
+        paramsPayload,
+        prepareParams,
+        destination,
+        'remove',
+      ),
+    );
+  }
+
+  if (groupedRecordsByAction.insert) {
+    const insertRecordChunksArray = returnArrayOfSubarrays(
+      groupedRecordsByAction.insert,
+      maxUserCountNumber,
+    );
+    toSendEvents.push(
+      ...processRecordEventArray(
+        insertRecordChunksArray,
+        userSchema,
+        isHashRequired,
+        disableFormat,
+        paramsPayload,
+        prepareParams,
+        destination,
+        'add',
+      ),
+    );
+  }
+
+  if (groupedRecordsByAction.update) {
+    const updateRecordChunksArray = returnArrayOfSubarrays(
+      groupedRecordsByAction.update,
+      maxUserCountNumber,
+    );
+    toSendEvents.push(
+      ...processRecordEventArray(
+        updateRecordChunksArray,
+        userSchema,
+        isHashRequired,
+        disableFormat,
+        paramsPayload,
+        prepareParams,
+        destination,
+        'add',
+      ),
+    );
+  }
+
+  const respList = [];
+  toSendEvents.forEach((sendEvent) => {
+    respList.push(responseBuilderSimple(sendEvent, operationAudienceId));
+  });
+  if (respList.length === 0) {
+    throw new InstrumentationError(
+      'Missing valid parameters, unable to generate transformed payload',
+    );
+  }
+  return respList;
+}
+
+module.exports = {
+  processRecordInputs,
+  responseBuilderSimple,
+};

src/v0/destinations/fb_custom_audience/util.js

@@ -206,4 +206,10 @@ const prepareDataField = (
   return data;
 };
 
-module.exports = { prepareDataField, getSchemaForEventMappedToDest, batchingWithPayloadSize };
+module.exports = {
+  prepareDataField,
+  getSchemaForEventMappedToDest,
+  batchingWithPayloadSize,
+  ensureApplicableFormat,
+  getUpdatedDataElement,
+};