feat: add batching support

rudderlabs · Mar 11, 2024 · 4199b02 · 4199b02
1 parent b5bcf3e
commit 4199b02
Show file tree

Hide file tree

Showing 3 changed files with 81 additions and 55 deletions.
diff --git a/src/cdk/v2/destinations/movable_ink/procWorkflow.yaml b/src/cdk/v2/destinations/movable_ink/procWorkflow.yaml
@@ -53,13 +53,19 @@ steps:
       }
 
   - name: buildResponse
+    description: In batchMode we return payload directly
+    condition: $.batchMode
     template: |
-      const response = $.defaultRequestConfig();
-      response.body.JSON = $.context.payload;
-      response.endpoint = .destination.Config.endpoint;
-      response.method = "POST";
-      response.headers = {
-        "Content-Type": "application/json",
-        "Authorization": "Basic " + $.base64Convertor(.destination.Config.accessKey + ":" + .destination.Config.accessSecret)
-      }
-      response;
+      $.context.payload
+    else:
+      name: buildResponseForProcessTransformation
+      template: |
+        const response = $.defaultRequestConfig();
+        response.body.JSON = $.context.payload;
+        response.endpoint = .destination.Config.endpoint;
+        response.method = "POST";
+        response.headers = {
+          "Content-Type": "application/json",
+          "Authorization": "Basic " + $.base64Convertor(.destination.Config.accessKey + ":" + .destination.Config.accessSecret)
+        }
+        response;
diff --git a/src/cdk/v2/destinations/movable_ink/rtWorkflow.yaml b/src/cdk/v2/destinations/movable_ink/rtWorkflow.yaml
@@ -3,6 +3,8 @@ bindings:
     path: ../../../../v0/util/index
   - path: ./utils
     exportAll: true
+  - name: base64Convertor
+    path: ../../../../v0/util
 
 steps:
   - name: validateInput
@@ -12,6 +14,9 @@ steps:
   - name: transform
     externalWorkflow:
       path: ./procWorkflow.yaml
+      bindings:
+        - name: batchMode
+          value: true
     loopOverInput: true
 
   - name: successfulEvents
@@ -20,14 +25,45 @@ steps:
         "batchedRequest": .,
         "batched": false,
         "destination": ^[idx].destination,
-        "metadata": ^[idx].metadata[],
+        "metadata": ^[idx].metadata,
         "statusCode": 200
       })[]
+
   - name: failedEvents
     template: |
       $.outputs.transform#idx.error.(
         $.handleRtTfSingleEventError(^[idx], .originalError ?? ., {})
       )[]
+
+  - name: batchSuccessfulEvents
+    description: Batches the successfulEvents
+    template: |
+      let batches = [$.outputs.successfulEvents];
+      batches@batch.({
+        "batchedRequest": {
+          "body": {
+            "JSON": {"events": ~r batch.batchedRequest[]},
+            "JSON_ARRAY": {},
+            "XML": {},
+            "FORM": {}
+          },
+          "version": "1",
+          "type": "REST",
+          "method": "POST",
+          "endpoint": batch[0].destination.Config.().(.endpoint),
+          "headers": batch[0].destination.Config.().({
+            "Content-Type": "application/json",
+            "Authorization": "Basic " + $.base64Convertor(.accessKey + ":" + .accessSecret)
+          }),
+          "params": {},
+          "files": {}
+        },
+        "metadata": ~r batch.metadata[],
+        "batched": true,
+        "statusCode": 200,
+        "destination": batch[0].destination
+      })[];
+
   - name: finalPayload
     template: |
-      [...$.outputs.successfulEvents, ...$.outputs.failedEvents]
+      [...$.outputs.batchSuccessfulEvents, ...$.outputs.failedEvents]
diff --git a/test/integrations/destinations/movable_ink/router/data.ts b/test/integrations/destinations/movable_ink/router/data.ts
@@ -1,6 +1,6 @@
 import { RouterTestData } from '../../../testTypes';
 import { RouterTransformationRequest } from '../../../../../src/types';
-import { generateMetadata, transformResultBuilder } from '../../../testUtils';
+import { generateMetadata } from '../../../testUtils';
 import {
   destType,
   channel,
@@ -101,56 +101,40 @@ export const data: RouterTestData[] = [
                 params: {},
                 body: {
                   JSON: {
-                    type: 'identify',
-                    userId: traits.email,
-                    anonymousId: 'anonId123',
-                    traits,
-                    integrations: {
-                      All: true,
-                    },
-                    originalTimestamp: '2024-03-04T15:32:56.409Z',
-                    timestamp: 1709566376409,
+                    events: [
+                      {
+                        type: 'identify',
+                        userId: traits.email,
+                        anonymousId: 'anonId123',
+                        traits,
+                        integrations: {
+                          All: true,
+                        },
+                        originalTimestamp: '2024-03-04T15:32:56.409Z',
+                        timestamp: 1709566376409,
+                      },
+                      {
+                        type: 'track',
+                        channel,
+                        userId: 'userId123',
+                        anonymousId: 'anonId123',
+                        properties: trackTestProperties['Product Added'],
+                        integrations: {
+                          All: true,
+                        },
+                        originalTimestamp: '2024-03-04T15:32:56.409Z',
+                        timestamp: 1709566376409,
+                      },
+                    ],
                   },
                   JSON_ARRAY: {},
                   XML: {},
                   FORM: {},
                 },
                 files: {},
               },
-              metadata: [generateMetadata(1)],
-              batched: false,
-              statusCode: 200,
-              destination,
-            },
-            {
-              batchedRequest: {
-                version: '1',
-                type: 'REST',
-                method: 'POST',
-                endpoint: destination.Config.endpoint,
-                headers,
-                params: {},
-                body: {
-                  JSON: {
-                    type: 'track',
-                    channel,
-                    userId: 'userId123',
-                    anonymousId: 'anonId123',
-                    properties: trackTestProperties['Product Added'],
-                    integrations: {
-                      All: true,
-                    },
-                    originalTimestamp: '2024-03-04T15:32:56.409Z',
-                    timestamp: 1709566376409,
-                  },
-                  JSON_ARRAY: {},
-                  XML: {},
-                  FORM: {},
-                },
-                files: {},
-              },
-              metadata: [generateMetadata(3)],
-              batched: false,
+              metadata: [generateMetadata(1), generateMetadata(3)],
+              batched: true,
               statusCode: 200,
               destination,
             },