feat: onboard klaviyo bulk upload destination

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

@@ -0,0 +1,5 @@
+const KLAVIYO_BULK_UPLOAD = 'klaviyo_bulk_upload';
+
+module.exports = {
+  KLAVIYO_BULK_UPLOAD,
+};

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

@@ -0,0 +1,35 @@
+bindings:
+  - name: EventType
+    path: ../../../../constants
+  - path: ../../bindings/jsontemplate
+  - name: defaultRequestConfig
+    path: ../../../../v0/util
+  - name: removeUndefinedAndNullValues
+    path: ../../../../v0/util
+  - path: ./utils
+  - path: ../../bindings/jsontemplate
+    exportAll: true
+  - path: ./config
+
+steps:
+  - name: messageType
+    template: |
+      .message.type.toLowerCase();
+  - name: validateInput
+    template: |
+      let messageType = .message.type;
+      $.assert(messageType, "message Type is not present. Aborting");
+      $.assert(.message.type.toLowerCase() ==='identify', "Event type " + .message.type.toLowerCase() + " is not supported. Aborting message.");
+      $.assertConfig(.destination.Config.privateApiKey, "Private Api Key is not present. Aborting");
+
+  - name: generatePayload
+    template: |
+      const transformedPayload = $.combinePayloads(^.{.message.type === $.EventType.IDENTIFY}[], ^[0].destination)
+      transformedPayload
+
+  - name: buildResponseForProcessTransformation
+    description: build response
+    template: |
+      const response = $.defaultRequestConfig();
+      response.body.JSON = $.outputs.generatePayload
+      response

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

@@ -0,0 +1,145 @@
+/* eslint-disable no-param-reassign */
+const { removeUndefinedValues } = require('../../../../v0/util');
+
+const locationAttributes = [
+  'address1',
+  'address2',
+  'city',
+  'country',
+  'latitude',
+  'longitude',
+  'region',
+  'zip',
+  'ip',
+];
+
+const standardTraits = [
+  'email',
+  'phone_number',
+  'external_id',
+  'anonymous_id',
+  '_kx',
+  'first_name',
+  'last_name',
+  'organization',
+  'title',
+  'image',
+];
+
+function setStandardAndCustomTraits(traits) {
+  const standardAttributes = {};
+  const customAttributes = {};
+  return Object.keys(traits).reduce(
+    (result, key) => {
+      if (!locationAttributes.includes(key) && standardTraits.includes(key)) {
+        result.standardAttributes[key] = traits[key];
+      } else if (!locationAttributes.includes(key)) {
+        result.customAttributes[key] = traits[key];
+      }
+      return result;
+    },
+    { standardAttributes, customAttributes },
+  );
+}
+
+function generateLocationObject({
+  traits: { address1, address2, city, country, latitude, longitude, region, zip, ip },
+}) {
+  const locationObject = {
+    address1,
+    address2,
+    city,
+    country,
+    latitude,
+    longitude,
+    region,
+    zip,
+    ip,
+  };
+
+  return removeUndefinedValues(locationObject);
+}
+
+function transformSingleMessage(data, metadata) {
+  const { context, traits } = data;
+  const location = generateLocationObject(data);
+  const { jobId } = metadata;
+  const { standardAttributes, customAttributes } = setStandardAndCustomTraits(traits);
+  const transformedSinglePayload = {
+    type: 'profile',
+    attributes: {
+      ...standardAttributes,
+      location,
+      properties: customAttributes,
+      anonymous_id: context.externalId[0].id,
+      jobIdentifier: `${context.externalId[0].id}:${jobId}`,
+    },
+  };
+  if (context.externalId[0].identifierType === 'id') {
+    transformedSinglePayload.id = context.externalId[0].id || traits.id;
+    transformedSinglePayload.attributes.anonymous_id = context.externalId[0].id;
+  } else if (context.externalId[0].identifierType === 'email') {
+    transformedSinglePayload.attributes.email = context.externalId[0].id;
+  }
+  if (Object.keys(transformedSinglePayload.attributes.location).length === 0) {
+    delete transformedSinglePayload.attributes.location;
+  }
+  if (Object.keys(transformedSinglePayload.attributes.properties).length === 0) {
+    delete transformedSinglePayload.attributes.properties;
+  }
+  return removeUndefinedValues(transformedSinglePayload);
+}
+
+function wrapCombinePayloads(transformedInputs, destinationObj) {
+  if (destinationObj.Config.listId) {
+    return {
+      payload: {
+        data: {
+          type: 'profile-bulk-import-job',
+          attributes: {
+            profiles: {
+              data: transformedInputs,
+            },
+          },
+          relationships: {
+            lists: {
+              data: [
+                {
+                  type: 'list',
+                  id: destinationObj.Config.listId,
+                },
+              ],
+            },
+          },
+        },
+      },
+      destination: destinationObj,
+    };
+  }
+  return {
+    payload: {
+      data: {
+        type: 'profile-bulk-import-job',
+        attributes: {
+          profiles: {
+            data: transformedInputs,
+          },
+        },
+      },
+    },
+    destination: destinationObj,
+  };
+}
+
+function combinePayloads(inputs) {
+  const transformedInputs = inputs.map((input) => {
+    const { message, metadata } = input;
+    return transformSingleMessage(message, metadata);
+  });
+  const destinationObj = inputs[inputs.length - 1].destination;
+
+  const { payload } = wrapCombinePayloads(transformedInputs, destinationObj);
+  return { ...payload };
+}
+
+module.exports = { transformSingleMessage, combinePayloads };

src/constants/destinationCanonicalNames.js

@@ -166,6 +166,13 @@ const DestCanonicalNames = {
   ],
   koala: ['Koala', 'koala', 'KOALA'],
   bloomreach: ['Bloomreach', 'bloomreach', 'BLOOMREACH'],
+  KLAVIYO_BULK_UPLOAD: [
+    'klaviyo bulk upload',
+    'klaviyo_bulk_upload',
+    'klaviyoBulkUpload',
+    'Klaviyo Bulk Upload',
+    'klaviyobulkupload',
+  ],
   emarsys: ['EMARSYS', 'Emarsys', 'emarsys'],
 };