feat: garl record event support

src/v0/destinations/fb_custom_audience/recordTransform.js

@@ -1,11 +1,7 @@
 /* eslint-disable no-const-assign */
 const lodash = require('lodash');
 const get = require('get-value');
-const {
-  InstrumentationError,
-  ConfigurationError,
-  getErrorRespEvents,
-} = require('@rudderstack/integrations-lib');
+const { InstrumentationError, ConfigurationError } = require('@rudderstack/integrations-lib');
 const { schemaFields } = require('./config');
 const { MappedToDestinationKey } = require('../../../constants');
 const stats = require('../../../util/stats');
@@ -15,8 +11,8 @@ const {
   checkSubsetOfArray,
   returnArrayOfSubarrays,
   getSuccessRespEvents,
-  generateErrorObject,
 } = require('../../util');
+const { getErrorResponse, createFinalResponse } = require('../../util/recordUtils');
 const {
   ensureApplicableFormat,
   getUpdatedDataElement,
@@ -26,19 +22,6 @@ const {
   getDataSource,
 } = require('./util');
 
-function getErrorMetaData(inputs, acceptedOperations) {
-  const metadata = [];
-  // eslint-disable-next-line no-restricted-syntax
-  for (const key in inputs) {
-    if (!acceptedOperations.includes(key)) {
-      inputs[key].forEach((input) => {
-        metadata.push(input.metadata);
-      });
-    }
-  }
-  return metadata;
-}
-
 const processRecordEventArray = (
   recordChunksArray,
   userSchema,
@@ -177,8 +160,6 @@ async function processRecordInputs(groupedRecordInputs) {
     record.message.action?.toLowerCase(),
   );
 
-  const finalResponse = [];
-
   let insertResponse;
   let deleteResponse;
   let updateResponse;
@@ -238,32 +219,14 @@ async function processRecordInputs(groupedRecordInputs) {
     );
   }
 
-  const eventTypes = ['update', 'insert', 'delete'];
-  const errorMetaData = [];
-  const errorMetaDataObject = getErrorMetaData(groupedRecordsByAction, eventTypes);
-  if (errorMetaDataObject.length > 0) {
-    errorMetaData.push(errorMetaDataObject);
-  }
+  const errorResponse = getErrorResponse(groupedRecordsByAction);
 
-  const error = new InstrumentationError('Invalid action type in record event');
-  const errorObj = generateErrorObject(error);
-  const errorResponseList = errorMetaData.map((metadata) =>
-    getErrorRespEvents(metadata, errorObj.status, errorObj.message, errorObj.statTags),
+  const finalResponse = createFinalResponse(
+    deleteResponse,
+    insertResponse,
+    updateResponse,
+    errorResponse,
   );
-
-  if (deleteResponse && deleteResponse.batchedRequest.length > 0) {
-    finalResponse.push(deleteResponse);
-  }
-  if (insertResponse && insertResponse.batchedRequest.length > 0) {
-    finalResponse.push(insertResponse);
-  }
-  if (updateResponse && updateResponse.batchedRequest.length > 0) {
-    finalResponse.push(updateResponse);
-  }
-  if (errorResponseList.length > 0) {
-    finalResponse.push(...errorResponseList);
-  }
-
   if (finalResponse.length === 0) {
     throw new InstrumentationError(
       'Missing valid parameters, unable to generate transformed payload',

src/v0/destinations/google_adwords_remarketing_lists/config.js

@@ -1,6 +1,6 @@
 const { getMappingConfig } = require('../../util');
 
-const BASE_ENDPOINT = 'https://googleads.googleapis.com/v15/customers';
+const BASE_ENDPOINT = 'https://googleads.googleapis.com/v16/customers';
 const CONFIG_CATEGORIES = {
   AUDIENCE_LIST: { type: 'audienceList', name: 'offlineDataJobs' },
   ADDRESSINFO: { type: 'addressInfo', name: 'addressInfo' },

src/v0/destinations/google_adwords_remarketing_lists/recordTransform.js

@@ -0,0 +1,144 @@
+/* eslint-disable no-const-assign */
+const lodash = require('lodash');
+const { InstrumentationError } = require('@rudderstack/integrations-lib');
+const {
+  getValueFromMessage,
+  getAccessToken,
+  constructPayload,
+  returnArrayOfSubarrays,
+  getSuccessRespEvents,
+} = require('../../util');
+const { populateConsentFromConfig } = require('../../util/googleUtils');
+const { populateIdentifiers, responseBuilder } = require('./util');
+const { getErrorResponse, createFinalResponse } = require('../../util/recordUtils');
+const { offlineDataJobsMapping, consentConfigMap } = require('./config');
+
+const processRecordEventArray = (
+  records,
+  message,
+  destination,
+  accessToken,
+  developerToken,
+  operationType,
+) => {
+  let outputPayloads = {};
+  // ** only send it if identifier > 0
+
+  const fieldsArray = [];
+  const metadata = [];
+  records.forEach((record) => {
+    fieldsArray.push(record.message.fields);
+    metadata.push(record.metadata);
+  });
+
+  const userIdentifiersList = populateIdentifiers(fieldsArray, destination);
+
+  const outputPayload = constructPayload(message, offlineDataJobsMapping);
+  outputPayload.operations = [];
+  // breaking the userIdentiFier array in chunks of 20
+  const userIdentifierChunks = returnArrayOfSubarrays(userIdentifiersList, 20);
+  // putting each chunk in different create/remove operations
+  switch (operationType) {
+    case 'add':
+      // for add operation
+      userIdentifierChunks.forEach((element) => {
+        const operations = {
+          create: {},
+        };
+        operations.create.userIdentifiers = element;
+        outputPayload.operations.push(operations);
+      });
+      outputPayloads = { ...outputPayloads, create: outputPayload };
+      break;
+    case 'remove':
+      // for remove operation
+      userIdentifierChunks.forEach((element) => {
+        const operations = {
+          remove: {},
+        };
+        operations.remove.userIdentifiers = element;
+        outputPayload.operations.push(operations);
+      });
+      outputPayloads = { ...outputPayloads, remove: outputPayload };
+      break;
+    default:
+  }
+
+  const toSendEvents = [];
+  Object.values(outputPayloads).forEach((data) => {
+    const consentObj = populateConsentFromConfig(destination.Config, consentConfigMap);
+    toSendEvents.push(
+      responseBuilder(accessToken, developerToken, data, destination, message, consentObj),
+    );
+  });
+
+  const successResponse = getSuccessRespEvents(toSendEvents, metadata, destination, true);
+
+  return successResponse;
+};
+
+async function processRecordInputs(groupedRecordInputs) {
+  const { destination, message, metadata } = groupedRecordInputs[0];
+  const accessToken = getAccessToken(metadata, 'accessToken');
+  const developerToken = getValueFromMessage(metadata, 'secret.developer_token');
+
+  const groupedRecordsByAction = lodash.groupBy(groupedRecordInputs, (record) =>
+    record.message.action?.toLowerCase(),
+  );
+
+  let insertResponse;
+  let deleteResponse;
+  let updateResponse;
+
+  if (groupedRecordsByAction.delete) {
+    deleteResponse = processRecordEventArray(
+      groupedRecordsByAction.delete,
+      message,
+      destination,
+      accessToken,
+      developerToken,
+      'remove',
+    );
+  }
+
+  if (groupedRecordsByAction.insert) {
+    insertResponse = processRecordEventArray(
+      groupedRecordsByAction.insert,
+      message,
+      destination,
+      accessToken,
+      developerToken,
+      'add',
+    );
+  }
+
+  if (groupedRecordsByAction.update) {
+    updateResponse = processRecordEventArray(
+      groupedRecordsByAction.update,
+      message,
+      destination,
+      accessToken,
+      developerToken,
+      'add',
+    );
+  }
+
+  const errorResponse = getErrorResponse(groupedRecordsByAction);
+  const finalResponse = createFinalResponse(
+    deleteResponse,
+    insertResponse,
+    updateResponse,
+    errorResponse,
+  );
+  if (finalResponse.length === 0) {
+    throw new InstrumentationError(
+      'Missing valid parameters, unable to generate transformed payload',
+    );
+  }
+
+  return finalResponse;
+}
+
+module.exports = {
+  processRecordInputs,
+};