feat: emersys initial commit

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

@@ -0,0 +1,26 @@
+const { getMappingConfig } = require('../../../../v0/util');
+
+// ref : https://learn.microsoft.com/en-us/linkedin/marketing/integrations/ads-reporting/conversions-api?view=li-lms-2024-02&tabs=http#adding-multiple-conversion-events-in-a-batch
+const BATCH_ENDPOINT = 'https://api.linkedin.com/rest/conversionEvents';
+const API_HEADER_METHOD = 'BATCH_CREATE';
+const API_VERSION = '202402'; // yyyymm format
+const API_PROTOCOL_VERSION = '2.0.0';
+
+const CONFIG_CATEGORIES = {
+  USER_INFO: {
+    name: 'linkedinUserInfoConfig',
+    type: 'user',
+  },
+};
+
+const MAPPING_CONFIG = getMappingConfig(CONFIG_CATEGORIES, __dirname);
+
+module.exports = {
+  MAX_BATCH_SIZE: 5000,
+  BATCH_ENDPOINT,
+  API_HEADER_METHOD,
+  API_VERSION,
+  API_PROTOCOL_VERSION,
+  CONFIG_CATEGORIES,
+  MAPPING_CONFIG,
+};

src/cdk/v2/destinations/emersys/procWorkflow.yaml

@@ -0,0 +1,57 @@
+bindings:
+  - name: EventType
+    path: ../../../../constants
+  - path: ../../bindings/jsontemplate
+    exportAll: true
+  - name: removeUndefinedValues
+    path: ../../../../v0/util
+  - name: defaultRequestConfig
+    path: ../../../../v0/util
+  - path: ./utils
+  - path: ./config
+  - path: lodash
+    name: cloneDeep
+
+steps:
+  - name: checkIfProcessed
+    condition: .message.statusCode
+    template: |
+      $.batchMode ? .message.body.JSON : .message
+    onComplete: return
+  - name: messageType
+    template: |
+      .message.type.toLowerCase()
+  - name: validateInput
+    template: |
+      let messageType = $.outputs.messageType;
+      $.assert(messageType, "Message type is not present. Aborting message.");
+      $.assert(messageType in {{$.EventType.([.TRACK, .IDENTIFY, .GROUP])}}, 
+        "message type " + messageType + " is not supported")
+      $.assertConfig(.destination.Config.emersysUsername, "Emersys user name is not configured. Aborting");
+      $.assertConfig(.destination.Config.emersysUserSecret, "Emersys user secret is not configured. Aborting");
+
+  - name: validateInputForTrack
+    description: Additional validation for Track events
+    condition: $.outputs.messageType === {{$.EventType.TRACK}}
+    template: |
+      $.assert(.message.event, "event could not be mapped to conversion rule. Aborting.")
+  - name: preparePayloadForIdentify
+    description: |
+      Builds identify payload.
+      ref: https://learn.microsoft.com/en-us/linkedin/marketing/integrations/ads-reporting/conversions-api?view=li-lms-2024-02&tabs=curl#adding-multiple-conversion-events-in-a-batch
+    template: |
+      $.context.payload = $.buildIdentifyPayload(.message, .destination.Config,);
+  - name: preparePayloadForGroup
+    description: |
+      Builds group payload.
+      ref: https://dev.emarsys.com/docs/core-api-reference/1m0m70hy3tuov-add-contacts-to-a-contact-list
+    template: |
+      $.context.payload = $.buildGroupPayload(.message, .destination.Config,);
+  - name: buildResponse
+    template: |
+      const response = $.defaultRequestConfig();
+      response.body.JSON = $.context.payload;
+      response.endpoint = $.deduceEndPoint(.message,.destination.Config);
+      response.method = "POST";
+      response.headers = $.buildHeader(.destination.Config)
+      response

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

@@ -0,0 +1,39 @@
+bindings:
+  - path: ./utils
+  - path: ./config
+  - name: handleRtTfSingleEventError
+    path: ../../../../v0/util/index
+
+steps:
+  - name: validateInput
+    template: |
+      $.assert(Array.isArray(^) && ^.length > 0, "Invalid event array")
+
+  - name: transform
+    externalWorkflow:
+      path: ./procWorkflow.yaml
+      bindings:
+        - name: batchMode
+          value: true
+    loopOverInput: true
+  - name: successfulEvents
+    template: |
+      $.outputs.transform#idx.output.({
+        "message": .[],
+        "destination": ^ [idx].destination,
+        "metadata": ^ [idx].metadata
+      })[]
+  - name: failedEvents
+    template: |
+      $.outputs.transform#idx.error.(
+        $.handleRtTfSingleEventError(^[idx], .originalError ?? ., {})
+      )[]
+
+  - name: batchSuccessfulEvents
+    description: Batches the successfulEvents
+    template: |
+      $.batchResponseBuilder($.outputs.successfulEvents);
+
+  - name: finalPayload
+    template: |
+      [...$.outputs.failedEvents, ...$.outputs.batchSuccessfulEvents]

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

@@ -0,0 +1,125 @@
+import { EVENT_TYPE } from 'rudder-transformer-cdk/build/constants';
+
+const lodash = require('lodash');
+const crypto = require('crypto');
+const get = require('get-value');
+
+const {
+  InstrumentationError,
+  isDefinedAndNotNullAndNotEmpty,
+  removeUndefinedAndNullAndEmptyValues,
+  removeUndefinedAndNullValues,
+} = require('@rudderstack/integrations-lib');
+const { getValueFromMessage } = require('rudder-transformer-cdk/build/utils');
+const { getIntegrationsObj } = require('../../../../v0/util');
+const { EMAIL_FIELD_ID } = require('./config');
+
+function base64Sha(str) {
+  const hexDigest = crypto.createHash('sha1').update(str).digest('hex');
+  return Buffer.from(hexDigest).toString('base64');
+}
+
+function getWsseHeader(user, secret) {
+  const nonce = crypto.randomBytes(16).toString('hex');
+  const timestamp = new Date().toISOString();
+
+  const digest = base64Sha(nonce + timestamp + secret);
+  return `UsernameToken Username="${user}", PasswordDigest="${digest}", Nonce="${nonce}", Created="${timestamp}"`;
+}
+
+const buildHeader = (destConfig) => {
+  const { emersysUsername, emersysUserSecret } = destConfig;
+  return {
+    'Content-Type': 'application/json',
+    Accept: 'application/json',
+    'X-WSSE': getWsseHeader(emersysUsername, emersysUserSecret),
+  };
+};
+
+const buildIdentifyPayload = (message, destination) => {
+  let identifyPayload;
+  const { fieldMapping, emersysCustomIdentifier, discardEmptyProperties, defaultContactList } =
+    destination.Config;
+  const payload = {};
+  if (fieldMapping) {
+    fieldMapping.forEach((trait) => {
+      const { rudderProperty, emersysProperty } = trait;
+      const value = get(message, rudderProperty);
+      if (value) {
+        payload[emersysProperty] = value;
+      }
+    });
+  }
+
+  const emersysIdentifier = emersysCustomIdentifier || EMAIL_FIELD_ID;
+  const finalPayload =
+    discardEmptyProperties === true
+      ? removeUndefinedAndNullAndEmptyValues(payload) // empty property value has a significance in emersys
+      : removeUndefinedAndNullValues(payload);
+  const integrationObject = getIntegrationsObj(message, 'emersys');
+
+  // TODO: add validation for opt in field
+
+  if (isDefinedAndNotNullAndNotEmpty(payload[emersysIdentifier])) {
+    identifyPayload = {
+      key_id: integrationObject.customIdentifierId || emersysIdentifier,
+      contacts: [...finalPayload],
+      contact_list_id: integrationObject.contactListId || defaultContactList,
+    };
+  } else {
+    throw new InstrumentationError(
+      'Either configured custom contact identifier value or default identifier email value is missing',
+    );
+  }
+
+  return identifyPayload;
+};
+
+function findRudderPropertyByEmersysProperty(emersysProperty, fieldMapping) {
+  // Use lodash to find the object where the emersysProperty matches the input
+  const item = lodash.find(fieldMapping, { emersysProperty });
+  // Return the rudderProperty if the object is found, otherwise return null
+  return item ? item.rudderProperty : null;
+}
+
+const buildGroupPayload = (message, destination) => {
+  const { emersysCustomIdentifier, defaultContactList, fieldMapping } = destination.Config;
+  const integrationObject = getIntegrationsObj(message, 'emersys');
+  const emersysIdentifier = emersysCustomIdentifier || EMAIL_FIELD_ID;
+  const configuredPayloadProperty = findRudderPropertyByEmersysProperty(
+    emersysIdentifier,
+    fieldMapping,
+  );
+  const payload = {
+    key_id: integrationObject.customIdentifierId || emersysIdentifier,
+    external_ids: [getValueFromMessage(message.context.traits, configuredPayloadProperty)],
+  };
+  return {
+    payload,
+    contactListId: message.groupId || defaultContactList,
+  };
+};
+
+const deduceEndPoint = (message, destConfig) => {
+  let endPoint;
+  let contactListId;
+  const { type, groupId } = message;
+  switch (type) {
+    case EVENT_TYPE.IDENTIFY:
+      endPoint = 'https://api.emarsys.net/api/v2/contact/?create_if_not_exists=1';
+      break;
+    case EVENT_TYPE.GROUP:
+      contactListId = groupId || destConfig.defaultContactList;
+      endPoint = `https://api.emarsys.net/api/v2/contactlist/${contactListId}/add`;
+      break;
+    default:
+      break;
+  }
+  return endPoint;
+};
+module.exports = {
+  buildIdentifyPayload,
+  buildGroupPayload,
+  buildHeader,
+  deduceEndPoint,
+};