feat: onboard algolia destination to cdk 2.0

__tests__/algolia-cdk.test.js

@@ -0,0 +1,62 @@
+const fs = require("fs");
+const path = require("path");
+const { TRANSFORMER_METRIC } = require("../v0/util/constant");
+const { getWorkflowEngine } = require("../cdk/v2/handler");
+
+const integration = "algolia";
+const name = "Algolia";
+
+const procWorkflowEnginePromise = getWorkflowEngine(integration, TRANSFORMER_METRIC.ERROR_AT.PROC);
+
+const inputDataFile = fs.readFileSync(
+  path.resolve(__dirname, `./data/${integration}_input.json`)
+);
+const outputDataFile = fs.readFileSync(
+  path.resolve(__dirname, `./data/${integration}_output.json`)
+);
+const inputData = JSON.parse(inputDataFile);
+const expectedData = JSON.parse(outputDataFile);
+
+describe(`${name} Tests`, () => {
+  describe("Processor Tests", () => {
+    inputData.forEach((input, index) => {
+      it(`${name} - payload: ${index}`, async () => {
+        const expected = expectedData[index];
+        try {
+          procWorkflowEngine = await procWorkflowEnginePromise;
+          const result = await procWorkflowEngine.execute(input);
+          expect(result.output).toEqual(expected);
+        } catch (error) {
+          // console.log(error);
+          expect(error.message).toEqual(expected.message);
+        }
+      });
+    });
+  });
+
+  const inputRouterDataFile = fs.readFileSync(
+    path.resolve(__dirname, `./data/${integration}_router_input.json`)
+  );
+  const outputRouterDataFile = fs.readFileSync(
+    path.resolve(__dirname, `./data/${integration}_router_output.json`)
+  );
+  const inputRouterData = JSON.parse(inputRouterDataFile);
+  const expectedRouterData = JSON.parse(outputRouterDataFile);
+
+  const rtWorkflowEnginePromise = getWorkflowEngine(integration, TRANSFORMER_METRIC.ERROR_AT.RT);
+  describe("Router Tests", () => {
+    inputRouterData.forEach((input, index) => {
+      it(`${name} - payload: ${index}`, async () => {
+        const expected = expectedRouterData[index];
+        try {
+          rtWorkflowEngine = await rtWorkflowEnginePromise;
+          const result = await rtWorkflowEngine.execute(input);
+          expect(result.output).toEqual(expected);
+        } catch (error) {
+          // console.log(error);
+          expect(error.message).toEqual(expected.message);
+        }
+      });
+    });
+  });
+});

__tests__/algolia.test.js

@@ -30,33 +30,28 @@ inputData.forEach((input, index) => {
   });
 });
 
-const batchInputDataFile = fs.readFileSync(
-  path.resolve(__dirname, `./data/${integration}_batch_input.json`)
+const routerInputDataFile = fs.readFileSync(
+  path.resolve(__dirname, `./data/${integration}_router_input.json`)
 );
-const batchOutputDataFile = fs.readFileSync(
-  path.resolve(__dirname, `./data/${integration}_batch_output.json`)
+const routerInputData = JSON.parse(routerInputDataFile);
+const routerOutputDataFile = fs.readFileSync(
+  path.resolve(__dirname, `./data/${integration}_router_output.json`)
 );
+const routerOutputData = JSON.parse(routerOutputDataFile);
 
-// const batchInputData = JSON.parse(batchInputDataFile);
-// const batchExpectedData = JSON.parse(batchOutputDataFile);
-
-// batchInputData.forEach((input, index) => {
-//   test(`${name} Batching ${index}`, () => {
-//     const output = transformer.batch(input);
-//     //console.log(output);
-//     expect(Array.isArray(output)).toEqual(true);
-//     expect(output.length).toEqual(batchExpectedData[index].length);
-//     output.forEach((input, indexInner) => {
-//       expect(output[indexInner]).toEqual(batchExpectedData[index][indexInner]);
-//     })
-//   });
-// });
-
-// Batching using routerTransform
-test('Batching', async () => {
-  const batchInputData = JSON.parse(batchInputDataFile);
-  const batchExpectedData = JSON.parse(batchOutputDataFile);
-  const output = await transformer.processRouterDest(batchInputData);
-  expect(Array.isArray(output)).toEqual(true);
-  expect(output).toEqual(batchExpectedData);
+describe('Router Tests', () => {
+  routerInputData.forEach((input, index) => {
+    it(`${name} Tests: payload - ${index}`, async () => {
+      let output, expected;
+      try {
+        output = await transformer.processRouterDest(input);
+        expected = routerOutputData[index]
+      } catch (error) {
+        output = error.message;
+        // console.log(output);
+        expected = routerOutputData[index].message;
+      }
+      expect(output).toEqual(expected);
+    });
+  });
 });

__tests__/data/algolia_batch_input.json → __tests__/data/algolia_router_input.json

@@ -1,4 +1,5 @@
 [
+  [
     {
       "message": {
         "type": "track",
@@ -251,5 +252,8 @@
         "IsProcessorEnabled": true
       }
     }
-  ]
+  ],
+  [],
+  {}
+]
 

__tests__/data/algolia_batch_output.json → __tests__/data/algolia_router_output.json

@@ -1,4 +1,5 @@
 [
+  [
     {
       "batchedRequest": {
         "version": "1",
@@ -97,5 +98,14 @@
         "IsProcessorEnabled": true
       }
     }
-  ]
+  ],
+  {
+      "message": "Invalid event array",
+      "statusCode": 400
+  },
+  {
+    "message": "Invalid event array",
+    "statusCode": 400
+  }
+]
 

cdk/v2/destinations/algolia/procWorkflow.yaml

@@ -0,0 +1,19 @@
+bindings:
+- name: EventType
+  path: ../../../../constants
+
+steps:
+  - name: validateInput
+    template: |
+      (
+        $assert($exists(message.type), "message Type is not present. Aborting message.");
+        $assert($exists(destination.Config.apiKey), "Invalid Api Key");
+        $assert($exists(destination.Config.applicationId), "Invalid Application Id");
+        $assert(message.type in [$EventType.TRACK], 
+          "message type " & message.type & " not supported")
+      )
+  - name: trackEventWorkflow
+    condition: message.type = $EventType.TRACK
+    externalWorkflow:
+      path: ./trackEventWorkflow.yaml
+

cdk/v2/destinations/algolia/rtWorkflow.yaml

@@ -0,0 +1,68 @@
+bindings:
+- path: ../../../../v0/destinations/algolia/config
+
+steps:
+  - name: validateInput
+    template: |
+      (
+        $assert($type($) = "array" and $count($) > 0, "Invalid event array")
+      )
+
+  - name: prepareContext
+    template:
+      $setContext("batchMode", true)
+
+  - name: transform
+    externalWorkflow:
+      path: ./procWorkflow.yaml
+    loopOverInput: true
+
+  - name: successfulEvents
+    template: |
+      $outputs.transform#$i.output.{
+          "output": $,
+          /* $$ refers to input (root of the document) */
+          "destination": $$[$i].destination,
+          "metadata": $$[$i].metadata
+      }
+  - name: failedEvents
+    template: |
+      $outputs.transform#$i.error.{
+        "metadata": $toArray($$[$i].metadata),
+        "batched": false,
+        "statusCode": status,
+        "error": message
+      }
+  - name: batchSuccessfulEvents
+    description: Batches the successfulEvents
+    template: |
+      (
+        $batches := $chunk($outputs.successfulEvents, $MAX_BATCH_SIZE);
+        $batches.{
+          "batchedRequest": {
+            "body": {
+              "JSON": {"events": output},
+              "JSON_ARRAY": {},
+              "XML": {},
+              "FORM": {}
+            },
+            "version": "1",
+            "type": "REST",
+            "method": "POST",
+            "endpoint": $ENDPOINT,
+          "headers": $[0].destination.Config.{
+            "X-Algolia-Application-Id": applicationId,
+            "X-Algolia-API-Key": apiKey
+          },
+            "params": {},
+            "files": {}
+          },
+          "metadata": metadata,
+          "batched": true,
+          "statusCode": 200,
+          "destination": $[0].destination
+        } ~> $toArray;
+      )
+  - name: finalPayload
+    template: |
+      $outputs.($append(batchSuccessfulEvents, failedEvents))

cdk/v2/destinations/algolia/trackEventWorkflow.yaml

@@ -0,0 +1,86 @@
+bindings:
+  - path: ../../../../v0/destinations/algolia/util
+  - path: ../../../../v0/destinations/algolia/config
+  - path: ../../../../v0/util
+
+steps:
+  - name: validateInput
+    template: |
+      (
+        $assert($exists(message.event), "event is required for track call");
+      )
+
+  - name: preparePayload
+    template: |
+      (
+        $eventTypeMap := $eventTypeMapping(destination.Config);
+        $event := message.event.$trim().$lowercase();
+        $eventType := $getValueFromMessage(message, "properties.eventType") ? $getValueFromMessage(message, "properties.eventType") : $lookup($eventTypeMap, $event);
+        $assert($exists($eventType), "eventType is mandatory for track call");
+        $setContext("payload", $constructPayload(message, $trackMapping));
+        $newPayload := $merge([$context.payload, { "eventName": $event, "eventType": $eventType }]);
+        $setContext("payload", $newPayload);
+        $setContext("payload", $genericpayloadValidator($context.payload));
+      )
+
+  - name: populateProductsData
+    condition: $context.payload.eventName in ["product list viewed", "order completed"] and $exists(message.properties.products)
+    steps:
+    - name: populateForClickEvent
+      condition: $context.payload.eventType = "click"
+      template: |
+        (
+          $validProducts := message.properties.products[$exists(objectId) and $exists(position)]#$i[$i<20];
+          $objectAndPositionIds := {
+            "objectIDs": $validProducts.objectId,
+            "positions": $validProducts.position
+          };
+          $objLen := $count($validProducts.objectId);
+          $posLen := $count($validProducts.position);
+          $newPayload := $merge([$context.payload, $objectAndPositionIds]);
+          $setContext("payload", $clickPayloadValidator($newPayload))
+        )
+    - name: populateForOtherEvents
+      condition: $context.payload.eventType != "click"
+      template: |
+        (
+          $objectIDs := {
+            "objectIDs": message.properties.products.objectId#$i[$i<20]
+          };
+          $setContext("payload", $merge([$context.payload, $objectIDs]));
+        )
+
+  - name: validateDestPayload
+    template: |
+      (
+        $assert($context.payload.($not(filters and objectIDs)), "event can't have both objectIds and filters at the same time.");
+        $assert($context.payload.(filters or objectIDs), "Either filters or  objectIds is required.");
+      )
+
+  - name: prepareResponsePayloadForBatch
+    condition: $exists($context.batchMode)
+    template: |
+      $context.payload
+    onComplete: return
+
+  - name: prepareResponsePayload
+    condition: $not($exists($context.batchMode))
+    template: |
+      {
+        "body": {
+          "JSON": {"events": [$context.payload]},
+          "JSON_ARRAY": {},
+          "XML": {},
+          "FORM": {}
+        },
+        "version": "1",
+        "type": "REST",
+        "method": "POST",
+        "endpoint": $ENDPOINT,
+        "headers": {
+          "X-Algolia-Application-Id": destination.Config.applicationId,
+          "X-Algolia-API-Key": destination.Config.apiKey
+        },
+        "params": {},
+        "files": {}
+      }

cdk/v2/pinterest_tag/procWorkflow.yaml → cdk/v2/destinations/pinterest_tag/procWorkflow.yaml

@@ -1,12 +1,12 @@
 bindings:
   - name: EventType
-    path: ../../../constants
+    path: ../../../../constants
   - name: SourceKeys
-    path: ../../../v0/util/data/GenericFieldMapping.json
+    path: ../../../../v0/util/data/GenericFieldMapping.json
     exportAll: true
-  - path: ../../../v0/destinations/pinterest_tag/utils
+  - path: ../../../../v0/destinations/pinterest_tag/utils
   - name: ENDPOINT
-    path: ../../../v0/destinations/pinterest_tag/config
+    path: ../../../../v0/destinations/pinterest_tag/config
 steps:
   - name: validateInput
     template: |

cdk/v2/pinterest_tag/rtWorkflow.yaml → cdk/v2/destinations/pinterest_tag/rtWorkflow.yaml

@@ -1,5 +1,5 @@
 bindings:
-  - path: ../../../v0/destinations/pinterest_tag/config
+  - path: ../../../../v0/destinations/pinterest_tag/config
 steps:
   - name: prepareContext
     template: $setContext("batchMode", true)

cdk/v2/utils.js

@@ -49,7 +49,7 @@ async function getWorkflowPath(
 function getRootPathForDestination(destName) {
   // TODO: Resolve the CDK v2 destination directory
   // path from the root directory
-  return path.join(CDK_V2_ROOT_DIR, destName);
+  return path.join(CDK_V2_ROOT_DIR, "destinations", destName);
 }
 
 async function getPlatformBindingsPaths() {

v0/destinations/algolia/transform.js

@@ -118,9 +118,8 @@ const process = event => {
 };
 
 const processRouterDest = async inputs => {
-  const errorRespEvents = checkInvalidRtTfEvents(inputs, "ALGOLIA");
-  if (errorRespEvents.length > 0) {
-    return errorRespEvents;
+  if (!Array.isArray(inputs) || inputs.length === 0) {
+    throw new CustomError("Invalid event array", 400);
   }
 
   const inputChunks = returnArrayOfSubarrays(inputs, MAX_BATCH_SIZE);