feat: trade desk real time conversions (#3023)

* feat: initial commit

* refactor: added comments

* featL add payload builder utilities

* feat: add privacy settings and canonical names

* refactor: created separate transform.js for record and conversion use case

* feat: added getRevenue utility and unit testcases

* refactor: handle record/track type and added testcases

* fix: use map instead of forEach for async operations

* docs: add comments

* feat: review changes

* refactor: moved transform.js code logic to rtWorkflow

* feat: added generate exclusion list utility

* refactor: error message and function variable name

* refactor: import endpoint from config and added api links

* docs: add comments in response handler

* refactor: created separate validateConfig utility for both flows

* feat: updated category id mapping

* test: add generateExclusionList testcases

jest.default.config.js

@@ -27,7 +27,7 @@ module.exports = {
   coverageDirectory: 'reports/coverage',
 
   // An array of regexp pattern strings used to skip coverage collection
-  coveragePathIgnorePatterns: ['/node_modules/', '__tests__', 'warehouse/v0' ,'test'],
+  coveragePathIgnorePatterns: ['/node_modules/', '__tests__', 'warehouse/v0', 'test'],
 
   // A list of reporter names that Jest uses when writing coverage reports
   coverageReporters: ['json', 'text', 'lcov', 'clover'],

src/cdk/v2/destinations/the_trade_desk/config.js

@@ -1,8 +1,11 @@
-const SUPPORTED_EVENT_TYPE = 'record';
+const { getMappingConfig } = require('../../../../v0/util');
+
+const SUPPORTED_EVENT_TYPE = ['record', 'track'];
 const ACTION_TYPES = ['insert', 'delete'];
 const DATA_PROVIDER_ID = 'rudderstack';
 
 // ref:- https://partner.thetradedesk.com/v3/portal/data/doc/DataEnvironments
+// api ref:- https://partner.thetradedesk.com/v3/portal/data/doc/post-data-advertiser-external
 const DATA_SERVERS_BASE_ENDPOINTS_MAP = {
   apac: 'https://sin-data.adsrvr.org',
   tokyo: 'https://tok-data.adsrvr.org',
@@ -12,10 +15,66 @@ const DATA_SERVERS_BASE_ENDPOINTS_MAP = {
   china: 'https://data-cn2.adsrvr.cn',
 };
 
+// ref:- https://partner.thetradedesk.com/v3/portal/data/doc/DataConversionEventsApi
+const REAL_TIME_CONVERSION_ENDPOINT = 'https://insight.adsrvr.org/track/realtimeconversion';
+
+const CONVERSION_SUPPORTED_ID_TYPES = [
+  'TDID',
+  'IDFA',
+  'AAID',
+  'DAID',
+  'NAID',
+  'IDL',
+  'EUID',
+  'UID2',
+];
+
+const ECOMM_EVENT_MAP = {
+  'product added': {
+    event: 'addtocart',
+    rootLevelPriceSupported: true,
+  },
+  'order completed': {
+    event: 'purchase',
+    itemsArray: true,
+    revenueFieldSupported: true,
+  },
+  'product viewed': {
+    event: 'viewitem',
+    rootLevelPriceSupported: true,
+  },
+  'checkout started': {
+    event: 'startcheckout',
+    itemsArray: true,
+    revenueFieldSupported: true,
+  },
+  'cart viewed': {
+    event: 'viewcart',
+    itemsArray: true,
+  },
+  'product added to wishlist': {
+    event: 'wishlistitem',
+    rootLevelPriceSupported: true,
+  },
+};
+
+const CONFIG_CATEGORIES = {
+  COMMON_CONFIGS: { name: 'TTDCommonConfig' },
+  ITEM_CONFIGS: { name: 'TTDItemConfig' },
+};
+
+const MAPPING_CONFIG = getMappingConfig(CONFIG_CATEGORIES, __dirname);
+
 module.exports = {
   SUPPORTED_EVENT_TYPE,
   ACTION_TYPES,
   DATA_PROVIDER_ID,
   MAX_REQUEST_SIZE_IN_BYTES: 2500000,
   DATA_SERVERS_BASE_ENDPOINTS_MAP,
+  CONVERSION_SUPPORTED_ID_TYPES,
+  CONFIG_CATEGORIES,
+  COMMON_CONFIGS: MAPPING_CONFIG[CONFIG_CATEGORIES.COMMON_CONFIGS.name],
+  ITEM_CONFIGS: MAPPING_CONFIG[CONFIG_CATEGORIES.ITEM_CONFIGS.name],
+  ECOMM_EVENT_MAP,
+  REAL_TIME_CONVERSION_ENDPOINT,
 };

src/cdk/v2/destinations/the_trade_desk/data/TTDCommonConfig.json

@@ -0,0 +1,22 @@
+[
+  {
+    "destKey": "currency",
+    "sourceKeys": "properties.currency"
+  },
+  {
+    "destKey": "client_ip",
+    "sourceKeys": ["context.ip", "request_ip"]
+  },
+  {
+    "destKey": "referrer_url",
+    "sourceKeys": "context.page.referrer"
+  },
+  {
+    "destKey": "imp",
+    "sourceKeys": "messageId"
+  },
+  {
+    "destKey": "order_id",
+    "sourceKeys": "properties.order_id"
+  }
+]

src/cdk/v2/destinations/the_trade_desk/data/TTDItemConfig.json

@@ -0,0 +1,22 @@
+[
+  {
+    "destKey": "item_code",
+    "sourceKeys": ["product_id", "sku"]
+  },
+  {
+    "destKey": "name",
+    "sourceKeys": "name"
+  },
+  {
+    "destKey": "qty",
+    "sourceKeys": "quantity"
+  },
+  {
+    "destKey": "price",
+    "sourceKeys": "price"
+  },
+  {
+    "destKey": "cat",
+    "sourceKeys": "category_id"
+  }
+]

src/cdk/v2/destinations/the_trade_desk/rtWorkflow.yaml

@@ -1,17 +1,47 @@
 bindings:
-  - name: processRouterDest
-    path: ./utils
+  - name: EventType
+    path: ../../../../constants
+  - name: processRecordInputs
+    path: ./transformRecord
+  - name: processConversionInputs
+    path: ./transformConversion
+  - name: handleRtTfSingleEventError
+    path: ../../../../v0/util/index
+  - name: InstrumentationError
+    path: '@rudderstack/integrations-lib'
 
 steps:
-  - name: validateInput
+  - name: validateCommonConfig
+    description: |
+      validate common config for first party data and realtime conversion flow
     template: |
-      $.assert(Array.isArray(^) && ^.length > 0, "Invalid event array")
       const config = ^[0].destination.Config
-      $.assertConfig(config.audienceId, "Segment name is not present. Aborting")
       $.assertConfig(config.advertiserId, "Advertiser ID is not present. Aborting")
-      $.assertConfig(config.advertiserSecretKey, "Advertiser Secret Key is not present. Aborting")
-      config.ttlInDays ?  $.assertConfig(config.ttlInDays >=0 && config.ttlInDays <= 180, "TTL is out of range. Allowed values are 0 to 180 days")
 
-  - name: processRouterDest
+  - name: validateInput
+    template: |
+      $.assert(Array.isArray(^) && ^.length > 0, "Invalid event array")
+
+  - name: processRecordEvents
+    template: |
+      $.processRecordInputs(^.{.message.type === $.EventType.RECORD}[], ^[0].destination)
+
+  - name: processConversionEvents
+    template: |
+      $.processConversionInputs(^.{.message.type === $.EventType.TRACK}[])
+
+  - name: failOtherEvents
+    template: |
+      const otherEvents = ^.{.message.type !== $.EventType.TRACK && .message.type !== $.EventType.RECORD}[]
+      let failedEvents = otherEvents.map(
+        function(event) {
+        const error = new $.InstrumentationError("Event type " + event.message.type + " is not supported");
+          $.handleRtTfSingleEventError(event, error, {})
+        }
+      )
+
+      failedEvents ?? []
+
+  - name: finalPayload
     template: |
-      $.processRouterDest(^)
+      [...$.outputs.processRecordEvents, ...$.outputs.processConversionEvents, ...$.outputs.failOtherEvents]

src/cdk/v2/destinations/the_trade_desk/transformConversion.js

@@ -0,0 +1,98 @@
+const { InstrumentationError, ConfigurationError } = require('@rudderstack/integrations-lib');
+const {
+  defaultRequestConfig,
+  simpleProcessRouterDest,
+  defaultPostRequestConfig,
+  removeUndefinedAndNullValues,
+} = require('../../../../v0/util');
+const { EventType } = require('../../../../constants');
+const { REAL_TIME_CONVERSION_ENDPOINT } = require('./config');
+const {
+  prepareFromConfig,
+  prepareCommonPayload,
+  getRevenue,
+  prepareItemsPayload,
+  getAdvertisingId,
+  prepareCustomProperties,
+  populateEventName,
+  getDataProcessingOptions,
+  getPrivacySetting,
+  enrichTrackPayload,
+} = require('./utils');
+
+const responseBuilder = (payload) => {
+  const response = defaultRequestConfig();
+  response.endpoint = REAL_TIME_CONVERSION_ENDPOINT;
+  response.method = defaultPostRequestConfig.requestMethod;
+  response.body.JSON = payload;
+  return response;
+};
+
+const validateInputAndConfig = (message, destination) => {
+  const { Config } = destination;
+  if (!Config.trackerId) {
+    throw new ConfigurationError('Tracking Tag ID is not present. Aborting');
+  }
+
+  if (!message.type) {
+    throw new InstrumentationError('Event type is required');
+  }
+
+  const messageType = message.type.toLowerCase();
+  if (messageType !== EventType.TRACK) {
+    throw new InstrumentationError(`Event type "${messageType}" is not supported`);
+  }
+
+  if (!message.event) {
+    throw new InstrumentationError('Event name is not present. Aborting.');
+  }
+};
+
+const prepareTrackPayload = (message, destination) => {
+  const configPayload = prepareFromConfig(destination);
+  const commonPayload = prepareCommonPayload(message);
+  // prepare items array
+  const items = prepareItemsPayload(message);
+  const { id, type } = getAdvertisingId(message);
+  // get td1-td10 custom properties
+  const customProperties = prepareCustomProperties(message, destination);
+  const eventName = populateEventName(message, destination);
+  const value = getRevenue(message);
+  let payload = {
+    ...configPayload,
+    ...commonPayload,
+    event_name: eventName,
+    value,
+    items,
+    adid: id,
+    adid_type: type,
+    ...customProperties,
+    data_processing_option: getDataProcessingOptions(message),
+    privacy_settings: getPrivacySetting(message),
+  };
+
+  payload = enrichTrackPayload(message, payload);
+  return { data: [removeUndefinedAndNullValues(payload)] };
+};
+
+const trackResponseBuilder = (message, destination) => {
+  const payload = prepareTrackPayload(message, destination);
+  return responseBuilder(payload);
+};
+
+const processEvent = (message, destination) => {
+  validateInputAndConfig(message, destination);
+  return trackResponseBuilder(message, destination);
+};
+
+const process = (event) => processEvent(event.message, event.destination);
+
+const processConversionInputs = async (inputs, reqMetadata) => {
+  if (!inputs || inputs.length === 0) {
+    return [];
+  }
+  const respList = await simpleProcessRouterDest(inputs, process, reqMetadata);
+  return respList;
+};
+
+module.exports = { processConversionInputs };