feat: onboard new destination the trade desk

src/adapters/network.js

@@ -287,7 +287,7 @@ function getFormData(payload) {
  * @returns
  */
 const prepareProxyRequest = (request) => {
-  const { body, method, params, endpoint, headers } = request;
+  const { body, method, params, endpoint, headers, destinationConfig: config } = request;
   const { payload, payloadFormat } = getPayloadData(body);
   let data;
 
@@ -313,7 +313,7 @@ const prepareProxyRequest = (request) => {
   }
   // Ref: https://github.com/rudderlabs/rudder-server/blob/master/router/network.go#L164
   headers['User-Agent'] = 'RudderLabs';
-  return removeUndefinedValues({ endpoint, data, params, headers, method });
+  return removeUndefinedValues({ endpoint, data, params, headers, method, config });
 };
 
 /**

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

@@ -0,0 +1,21 @@
+const SUPPORTED_EVENT_TYPE = 'record';
+const ACTION_TYPES = ['insert', 'delete'];
+const DATA_PROVIDER_ID = 'rudderstack';
+
+// ref:- https://partner.thetradedesk.com/v3/portal/data/doc/DataEnvironments
+const DATA_SERVERS_BASE_ENDPOINTS_MAP = {
+  apac: 'https://sin-data.adsrvr.org',
+  tokyo: 'https://tok-data.adsrvr.org',
+  usEastCoast: 'https://use-data.adsrvr.org',
+  usWestCoast: 'https://usw-data.adsrvr.org',
+  ukEu: 'https://euw-data.adsrvr.org',
+  china: 'https://data-cn2.adsrvr.cn',
+};
+
+module.exports = {
+  SUPPORTED_EVENT_TYPE,
+  ACTION_TYPES,
+  DATA_PROVIDER_ID,
+  MAX_REQUEST_SIZE_IN_BYTES: 2500000,
+  DATA_SERVERS_BASE_ENDPOINTS_MAP,
+};

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

@@ -0,0 +1,17 @@
+bindings:
+  - name: processRouterDest
+    path: ./utils
+
+steps:
+  - name: validateInput
+    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
+    template: |
+      $.processRouterDest(^)

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

@@ -0,0 +1,107 @@
+const lodash = require('lodash');
+const CryptoJS = require('crypto-js');
+const { InstrumentationError, AbortedError } = require('@rudderstack/integrations-lib');
+const { BatchUtils } = require('@rudderstack/workflow-engine');
+const {
+  defaultPostRequestConfig,
+  defaultRequestConfig,
+  getSuccessRespEvents,
+  removeUndefinedAndNullValues,
+  handleRtTfSingleEventError,
+} = require('../../../../v0/util');
+const tradeDeskConfig = require('./config');
+
+const { DATA_PROVIDER_ID, DATA_SERVERS_BASE_ENDPOINTS_MAP } = tradeDeskConfig;
+
+const ttlInMin = (ttl) => parseInt(ttl, 10) * 1440;
+const getBaseEndpoint = (dataServer) => DATA_SERVERS_BASE_ENDPOINTS_MAP[dataServer];
+const getFirstPartyEndpoint = (dataServer) => `${getBaseEndpoint(dataServer)}/data/advertiser`;
+
+const getSignatureHeader = (request, secretKey) => {
+  if (!secretKey) {
+    throw new AbortedError('Secret key is missing. Aborting');
+  }
+  const sha1 = CryptoJS.HmacSHA1(JSON.stringify(request), secretKey);
+  const base = CryptoJS.enc.Base64.stringify(sha1);
+  return base;
+};
+
+const responseBuilder = (items, config) => {
+  const { advertiserId, dataServer } = config;
+
+  const payload = { DataProviderId: DATA_PROVIDER_ID, AdvertiserId: advertiserId, Items: items };
+
+  const response = defaultRequestConfig();
+  response.endpoint = getFirstPartyEndpoint(dataServer);
+  response.method = defaultPostRequestConfig.requestMethod;
+  response.body.JSON = removeUndefinedAndNullValues(payload);
+  return response;
+};
+
+const batchResponseBuilder = (items, config) => {
+  const response = [];
+  const itemsChunks = BatchUtils.chunkArrayBySizeAndLength(items, {
+    // TODO: use destructuring at the top of file once proper 'mocking' is implemented.
+    // eslint-disable-next-line unicorn/consistent-destructuring
+    maxSizeInBytes: tradeDeskConfig.MAX_REQUEST_SIZE_IN_BYTES,
+  });
+
+  itemsChunks.items.forEach((chunk) => {
+    response.push(responseBuilder(chunk, config));
+  });
+
+  return response;
+};
+
+const processRecordInputs = (inputs, destination) => {
+  const { Config } = destination;
+  const items = [];
+  const successMetadata = [];
+  const errorResponseList = [];
+
+  const error = new InstrumentationError('Invalid action type');
+
+  inputs.forEach((input) => {
+    const { fields, action } = input.message;
+    const isInsertOrDelete = action === 'insert' || action === 'delete';
+
+    if (isInsertOrDelete) {
+      successMetadata.push(input.metadata);
+      const data = [
+        {
+          Name: Config.audienceId,
+          TTLInMinutes: action === 'insert' ? ttlInMin(Config.ttlInDays) : 0,
+        },
+      ];
+
+      Object.keys(fields).forEach((id) => {
+        const value = fields[id];
+        if (value) {
+          // adding only non empty ID's
+          items.push({ [id]: value, Data: data });
+        }
+      });
+    } else {
+      errorResponseList.push(handleRtTfSingleEventError(input, error, {}));
+    }
+  });
+
+  const payloads = batchResponseBuilder(items, Config);
+
+  const response = getSuccessRespEvents(payloads, successMetadata, destination, true);
+  return [response, ...errorResponseList];
+};
+
+const processRouterDest = (inputs) => {
+  const respList = [];
+  const { destination } = inputs[0];
+  const groupedInputs = lodash.groupBy(inputs, (input) => input.message.type);
+  if (groupedInputs.record) {
+    const transformedRecordEvent = processRecordInputs(groupedInputs.record, destination);
+    respList.push(...transformedRecordEvent);
+  }
+
+  return respList;
+};
+
+module.exports = { getSignatureHeader, processRouterDest };

src/cdk/v2/destinations/the_trade_desk/utils.test.js

@@ -0,0 +1,49 @@
+const { AbortedError } = require('@rudderstack/integrations-lib');
+const { getSignatureHeader } = require('./utils');
+
+describe('getSignatureHeader', () => {
+  it('should calculate the signature header for a valid request and secret key', () => {
+    const request = { data: 'example' };
+    const secretKey = 'secret';
+    const expected = 'rvxETQ7kIU5Cko3GddD2AeFpz8E=';
+
+    const result = getSignatureHeader(request, secretKey);
+
+    expect(result).toBe(expected);
+  });
+
+  it('should handle requests with different data types and secret key', () => {
+    const request1 = { data: 'example' };
+    const secretKey1 = 'secret';
+    const expected1 = 'rvxETQ7kIU5Cko3GddD2AeFpz8E=';
+
+    const result1 = getSignatureHeader(request1, secretKey1);
+
+    expect(result1).toBe(expected1);
+
+    const request2 = { data: 123 };
+    const secretKey2 = 'secret';
+    const expected2 = 'V5RSVwxqHRLkZftZ0+IrZAp4L4s=';
+
+    const result2 = getSignatureHeader(request2, secretKey2);
+
+    expect(result2).toBe(expected2);
+
+    const request3 = { data: true };
+    const secretKey3 = 'secret';
+    const expected3 = 'oZ28NtyMYDGxRV0E+Tgvz7B1jds=';
+
+    const result3 = getSignatureHeader(request3, secretKey3);
+
+    expect(result3).toBe(expected3);
+  });
+
+  it('should throw an AbortedError when secret key is missing', () => {
+    const request = { data: 'example' };
+    const secretKey = null;
+
+    expect(() => {
+      getSignatureHeader(request, secretKey);
+    }).toThrow(AbortedError);
+  });
+});

src/features.json

@@ -64,6 +64,7 @@
     "ONE_SIGNAL": true,
     "TIKTOK_AUDIENCE": true,
     "REDDIT": true,
+    "THE_TRADE_DESK": true,
     "INTERCOM": true
   },
   "supportSourceTransformV1": true,

src/v0/destinations/the_trade_desk/networkHandler.js

@@ -0,0 +1,86 @@
+const { NetworkError, AbortedError } = require('@rudderstack/integrations-lib');
+const { httpSend, prepareProxyRequest } = require('../../../adapters/network');
+const {
+  processAxiosResponse,
+  getDynamicErrorType,
+} = require('../../../adapters/utils/networkUtils');
+const { getSignatureHeader } = require('../../../cdk/v2/destinations/the_trade_desk/utils');
+const { isHttpStatusSuccess } = require('../../util/index');
+const tags = require('../../util/tags');
+const { JSON_MIME_TYPE } = require('../../util/constant');
+
+const proxyRequest = async (request) => {
+  const { endpoint, data, method, params, headers, config } = prepareProxyRequest(request);
+
+  if (!config?.advertiserSecretKey) {
+    throw new AbortedError('Advertiser secret key is missing in destination config. Aborting');
+  }
+
+  if (!process.env.THE_TRADE_DESK_DATA_PROVIDER_SECRET_KEY) {
+    throw new AbortedError('Data provider secret key is missing. Aborting');
+  }
+
+  const ProxyHeaders = {
+    ...headers,
+    TtdSignature: getSignatureHeader(data, config.advertiserSecretKey),
+    'TtdSignature-dp': getSignatureHeader(
+      data,
+      process.env.THE_TRADE_DESK_DATA_PROVIDER_SECRET_KEY,
+    ),
+    'Content-Type': JSON_MIME_TYPE,
+  };
+
+  const requestOptions = {
+    url: endpoint,
+    data,
+    params,
+    headers: ProxyHeaders,
+    method,
+  };
+  const response = await httpSend(requestOptions, { feature: 'proxy', destType: 'the_trade_desk' });
+  return response;
+};
+
+const responseHandler = (destinationResponse) => {
+  const message = 'Request Processed Successfully';
+  const { response, status } = destinationResponse;
+
+  // if the response from destination is not a success case build an explicit error
+  if (!isHttpStatusSuccess(status)) {
+    throw new NetworkError(
+      `Request failed with status: ${status} due to ${response}`,
+      status,
+      {
+        [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
+      },
+      destinationResponse,
+    );
+  }
+
+  // Trade desk returns 200 with an error in case of "Failed to parse TDID, DAID, UID2, IDL, EUID, or failed to decrypt UID2Token or EUIDToken"
+  // https://partner.thetradedesk.com/v3/portal/data/doc/post-data-advertiser-external
+  // {"FailedLines":[{"ErrorCode":"MissingUserId","Message":"Invalid DAID, item #1"}]}
+  if ('FailedLines' in response && response.FailedLines.length > 0) {
+    throw new AbortedError(
+      `Request failed with status: ${status} due to ${JSON.stringify(response)}`,
+      400,
+      destinationResponse,
+    );
+  }
+
+  // else successfully return status, message and original destination response
+  // Trade desk returns 200 with empty object '{}' in response if all the events are processed successfully
+  return {
+    status,
+    message,
+    destinationResponse,
+  };
+};
+
+function networkHandler() {
+  this.proxy = proxyRequest;
+  this.processAxiosResponse = processAxiosResponse;
+  this.prepareProxy = prepareProxyRequest;
+  this.responseHandler = responseHandler;
+}
+module.exports = { networkHandler };

test/integrations/destinations/the_trade_desk/common.ts

@@ -0,0 +1,44 @@
+const destType = 'the_trade_desk';
+const destTypeInUpperCase = 'THE_TRADE_DESK';
+const advertiserId = 'test-advertiser-id';
+const dataProviderId = 'rudderstack';
+const segmentName = 'test-segment';
+const sampleDestination = {
+  Config: {
+    advertiserId,
+    advertiserSecretKey: 'test-advertiser-secret-key',
+    dataServer: 'apac',
+    ttlInDays: 30,
+    audienceId: segmentName,
+  },
+  DestinationDefinition: { Config: { cdkV2Enabled: true } },
+};
+
+const sampleSource = {
+  job_id: 'test-job-id',
+  job_run_id: 'test-job-run-id',
+  task_run_id: 'test-task-run-id',
+  version: 'v1.40.4',
+};
+
+const sampleContext = {
+  destinationFields: 'daid, uid2',
+  externalId: [
+    {
+      identifierType: 'tdid',
+      type: 'THE_TRADE_DESK-test-segment',
+    },
+  ],
+  mappedToDestination: 'true',
+  sources: sampleSource,
+};
+
+export {
+  destType,
+  destTypeInUpperCase,
+  advertiserId,
+  dataProviderId,
+  segmentName,
+  sampleDestination,
+  sampleContext,
+};