feat(INT-305): onboard gladly destination

package.json

@@ -63,7 +63,7 @@
     "@koa/router": "^12.0.0",
     "@ndhoule/extend": "^2.0.0",
     "@pyroscope/nodejs": "^0.2.6",
-    "@rudderstack/workflow-engine": "^0.5.7",
+    "@rudderstack/workflow-engine": "^0.6.8",
     "ajv": "^8.12.0",
     "ajv-draft-04": "^1.0.0",
     "ajv-formats": "^2.1.1",

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

@@ -0,0 +1,75 @@
+bindings:
+  - name: EventType
+    path: ../../../../constants
+  - path: ./utils
+    exportAll: true
+  - name: defaultRequestConfig
+    path: ../../../../v0/util
+  - name: removeUndefinedAndNullValues
+    path: ../../../../v0/util
+  - name: getDestinationExternalID
+    path: ../../../../v0/util
+  - name: httpPOST
+    path: ../../../../adapters/network
+  - name: processAxiosResponse
+    path: ../../../../adapters/utils/networkUtils
+
+
+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.")
+      $.assert(messageType in {{$.EventType.([.IDENTIFY])}}, "message type " + messageType + " is not supported")
+      $.assertConfig(.destination.Config.apiToken, "API Token is not present. Aborting")
+      $.assertConfig(.destination.Config.domain, "Gladly domain is not present. Aborting")
+      $.assertConfig(.destination.Config.userName, "User Name is not present. Aborting")
+      $.assert(.message.userId, "userId is required")
+
+  - name: preparePayload
+    template: |
+      $.context.payload = {
+        name: .message.traits.name || .message.context.traits.name,
+        image: .message.traits.avatar || .message.context.traits.avatar,
+        address: .message.traits.address || .message.context.traits.address
+      }
+      $.context.payload.address && typeof $.context.payload.address === "object" ?  $.context.payload.address =  JSON.stringify($.context.payload.address)
+      $.context.payload.emails = $.formatField(.message, "email")
+      $.context.payload.phones = $.formatField(.message, "phone")
+      $.context.payload.customAttributes = $.getCustomAttributes(.message)
+      $.context.payload.externalCustomerId = $.getDestinationExternalID(.message, "gladlyExternalCustomerId")
+      $.context.payload = $.removeUndefinedAndNullValues($.context.payload)
+
+  - name: createUser
+    description: Prepare create user payload
+    template: |
+      const requestOptions = {
+        headers: $.getHeaders(.destination)
+      }
+      const requestPayload = {...$.context.payload, id: .message.userId}
+      const endpoint = $.getEndpoint(.destination)
+      const rawResponse = await $.httpPOST(endpoint, requestPayload, requestOptions)
+      const processedResponse = $.processAxiosResponse(rawResponse)
+      processedResponse.status == 400 ? $.assertHttpResp(processedResponse, "Unable to create or update user due to " + JSON.stringify(processedResponse.response));
+      processedResponse
+
+  - name: buildResponseForProcessTransformation
+    description: build response for updateUser
+    template: |
+      const response = $.defaultRequestConfig()
+      response.body.JSON = $.removeUndefinedAndNullValues($.context.payload)
+      response.endpoint = $.getEndpoint(.destination) + "/" + .message.userId
+      response.method = "PATCH"
+      response.headers = $.getHeaders(.destination)
+      response.status = $.outputs.createUser.status
+      response

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

@@ -0,0 +1,34 @@
+bindings:
+  - name: handleRtTfSingleEventError
+    path: ../../../../v0/util/index
+  - path: ./utils
+    exportAll: true
+
+steps:
+  - name: validateInput
+    template: |
+      $.assert(Array.isArray(^) && ^.length > 0, "Invalid event array")
+
+  - name: transform
+    externalWorkflow:
+      path: ./procWorkflow.yaml
+    loopOverInput: true
+
+  - name: successfulEvents
+    template: |
+      const status = $.getStatusCode($.requestMetadata, $.outputs.transform#idx.output.status)
+      $.outputs.transform#idx.output.({
+        "batchedRequest": .{~["status"]},
+        "batched": false,
+        "destination": ^[idx].destination,
+        "metadata": ^[idx].metadata[],
+        "statusCode": status
+      })[]
+  - name: failedEvents
+    template: |
+      $.outputs.transform#idx.error.(
+        $.handleRtTfSingleEventError(^[idx], .originalError ?? ., {})
+      )[]
+  - name: finalPayload
+    template: |
+      [...$.outputs.successfulEvents, ...$.outputs.failedEvents]

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

@@ -0,0 +1,60 @@
+const { getFieldValueFromMessage, base64Convertor, isNewStatusCodesAccepted } = require('../../../../v0/util');
+const { HTTP_STATUS_CODES } = require('../../../../v0/util/constant');
+
+const reservedCustomAttributes = [
+  'email',
+  'phone',
+  'address',
+  'name',
+  'avatar',
+  'firstName',
+  'lstName',
+  'userId',
+];
+
+const getHeaders = (destination) => {
+  const { apiToken, userName } = destination.Config;
+  return {
+    'Content-Type': 'application/json',
+    Authorization: `Basic ${base64Convertor(`${userName}:${apiToken}`)}`,
+  };
+};
+
+const getEndpoint = (destination) => {
+  const { domain } = destination.Config;
+  return `https://${domain}/api/v1/customer-profiles`;
+};
+
+const formatField = (message, fieldName) => {
+  const field = getFieldValueFromMessage(message, fieldName);
+  if (field) {
+    if (Array.isArray(field)) {
+      return field.map((item) => ({ original: item }));
+    }
+    return [{ original: field }];
+  }
+  return undefined;
+};
+
+const getCustomAttributes = (message) => {
+  if (message.traits?.customAttributes && typeof message.traits?.customAttributes === 'object'){
+    return message.traits?.customAttributes;
+  }
+
+  const customAttributes = message.context.traits;
+  reservedCustomAttributes.forEach((customAttribute) => {
+    if (customAttributes[customAttribute]) {
+      delete customAttributes[customAttribute];
+    }
+  });
+  return customAttributes;
+};
+
+const getStatusCode = (requestMetadata, statusCode) => {
+  if(isNewStatusCodesAccepted(requestMetadata) && statusCode === 200){
+    return HTTP_STATUS_CODES.SUPPRESS_EVENTS;
+  }
+  return statusCode;
+}
+
+module.exports = { formatField, getCustomAttributes, getEndpoint, getHeaders, getStatusCode };

src/cdk/v2/handler.ts

@@ -64,9 +64,9 @@ export function getCachedWorkflowEngine(
   return workflowEnginePromiseMap[destName][feature];
 }
 
-export async function executeWorkflow(workflowEngine: WorkflowEngine, parsedEvent: FixMe) {
+export async function executeWorkflow(workflowEngine: WorkflowEngine, parsedEvent: FixMe, requestMetadata:  NonNullable<unknown>){
   try {
-    const result = await workflowEngine.execute(parsedEvent);
+    const result = await workflowEngine.execute(parsedEvent, {requestMetadata});
     // TODO: Handle remaining output scenarios
     return result.output;
   } catch (error) {
@@ -78,11 +78,12 @@ export async function processCdkV2Workflow(
   destType: string,
   parsedEvent: FixMe,
   feature: string,
+  requestMetadata:  NonNullable<unknown> = {},
   bindings: Record<string, FixMe> = {},
 ) {
   try {
     const workflowEngine = await getCachedWorkflowEngine(destType, feature, bindings);
-    return await executeWorkflow(workflowEngine, parsedEvent);
+    return await executeWorkflow(workflowEngine, parsedEvent, requestMetadata);
   } catch (error) {
     throw getErrorInfo(error, isCdkV2Destination(parsedEvent), defTags);
   }

src/constants/destinationCanonicalNames.js

@@ -139,7 +139,7 @@ const DestCanonicalNames = {
     'twitter_ads',
     'TWITTER_ADS',
   ],
-  BRAZE: ['BRAZE', 'Braze', 'braze'],
+  BRAZE: ['BRAZE', 'Braze', 'braze']
 };
 
 module.exports = { DestHandlerMap, DestCanonicalNames };

src/features.json

@@ -59,6 +59,7 @@
     "TWITTER_ADS": true,
     "CLEVERTAP": true,
     "ORTTO": true,
+    "GLADLY": true,
     "ONE_SIGNAL": true,
     "TIKTOK_AUDIENCE": true
   }

src/services/destination/cdkV2Integration.ts

@@ -52,7 +52,7 @@ export default class CDKV2DestinationService implements IntegrationDestinationSe
     events: ProcessorTransformationRequest[],
     destinationType: string,
     _version: string,
-    _requestMetadata: NonNullable<unknown>,
+    requestMetadata: NonNullable<unknown>,
   ): Promise<ProcessorTransformationResponse[]> {
     // TODO: Change the promise type
     const respList: ProcessorTransformationResponse[][] = await Promise.all(
@@ -64,6 +64,7 @@ export default class CDKV2DestinationService implements IntegrationDestinationSe
             destinationType,
             event,
             tags.FEATURES.PROCESSOR,
+            requestMetadata
           );
 
           stats.increment('event_transform_success', {
@@ -108,7 +109,7 @@ export default class CDKV2DestinationService implements IntegrationDestinationSe
     events: RouterTransformationRequestData[],
     destinationType: string,
     _version: string,
-    _requestMetadata: NonNullable<unknown>,
+    requestMetadata: NonNullable<unknown>,
   ): Promise<RouterTransformationResponse[]> {
     const allDestEvents: object = groupBy(
       events,
@@ -126,7 +127,7 @@ export default class CDKV2DestinationService implements IntegrationDestinationSe
           metaTo.metadata = destInputArray[0].metadata;
           try {
             const doRouterTransformationResponse: RouterTransformationResponse[] =
-              await processCdkV2Workflow(destinationType, destInputArray, tags.FEATURES.ROUTER);
+              await processCdkV2Workflow(destinationType, destInputArray, tags.FEATURES.ROUTER, requestMetadata);
             return DestinationPostTransformationService.handleRouterTransformSuccessEvents(
               doRouterTransformationResponse,
               undefined,