fix: handling partial error in a batch for reddit destination

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

@@ -36,7 +36,11 @@ steps:
     description: Batches the successfulEvents using endpoint
     condition: $.outputs.successfulEvents.length
     template: |
-      let batches = $.batchEvents($.outputs.successfulEvents);
+      const dontBatchTrueEvents =  $.outputs.successfulEvents{.metadata.dontBatch}[];
+      const dontBatchFalseEvents = $.outputs.successfulEvents{!.metadata.dontBatch}[];
+      const dontBatchTrueEventsChunks = $.chunk(dontBatchTrueEvents, 1);
+      
+      let batches = [...$.batchEvents(dontBatchFalseEvents), ...$.batchEventChunks(dontBatchTrueEventsChunks)];
       batches@batch.({
         "batchedRequest": {
           "body": {

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

@@ -81,6 +81,7 @@ const populateRevenueField = (eventType, properties) => {
 };
 module.exports = {
   batchEvents,
+  batchEventChunks,
   populateRevenueField,
   calculateDefaultRevenue,
 };

src/v0/destinations/reddit/networkHandler.js → src/v1/destinations/reddit/networkHandler.js

@@ -1,16 +1,31 @@
 const { RetryableError, TAG_NAMES, NetworkError } = require('@rudderstack/integrations-lib');
 const isString = require('lodash/isString');
 const { prepareProxyRequest, proxyRequest } = require('../../../adapters/network');
-const { isHttpStatusSuccess } = require('../../util/index');
+const { isHttpStatusSuccess } = require('../../../v0/util/index');
 const { REFRESH_TOKEN } = require('../../../adapters/networkhandler/authConstants');
 
 const {
   processAxiosResponse,
   getDynamicErrorType,
 } = require('../../../adapters/utils/networkUtils');
+const { TransformerProxyError } = require('../../../v0/util/errorTypes');
 
-const redditRespHandler = (destResponse) => {
-  const { status, response } = destResponse;
+const populateResponseWithDontBatch = (rudderJobMetadata, response) => {
+  const errorMessage = JSON.stringify(response);
+  return rudderJobMetadata.map(metadata => {
+    // eslint-disable-next-line no-param-reassign
+    metadata.dontBatch = true;
+    return {
+      statusCode: 500,
+      metadata,
+      error: errorMessage,
+    };
+  });
+};
+
+const redditRespHandler = (responseParams) => {
+  const { destinationResponse, rudderJobMetadata } = responseParams;
+  const { status, response } = destinationResponse;
 
   // to handle the case when authorization-token is invalid
   if (status === 401) {
@@ -28,38 +43,59 @@ const redditRespHandler = (destResponse) => {
     throw new RetryableError(
       `${errorMessage} during reddit response transformation`,
       status,
-      destResponse,
+      destinationResponse,
       authErrorCategory,
     );
   }
+  if (status === 400 && Array.isArray(rudderJobMetadata) && rudderJobMetadata.length > 1) {
+    // sending back 500 for retry only when events came in a batch
+    const responseWithDontBatch = populateResponseWithDontBatch(rudderJobMetadata, response);
+    throw new TransformerProxyError(
+      `REDDIT: Error transformer proxy during REDDIT response transformation`,
+      500,
+      {
+        [TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(500),
+      },
+      destinationResponse,
+      '',
+      responseWithDontBatch,
+    );
+  }
+
   throw new NetworkError(
     `${JSON.stringify(response)} during reddit response transformation`,
     status,
     {
       [TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
     },
-    destResponse,
+    destinationResponse,
   );
 };
 const responseHandler = (responseParams) => {
-  const { destinationResponse } = responseParams;
+  const { destinationResponse, rudderJobMetadata } = responseParams;
   const message = `Request Processed Successfully`;
   const { status } = destinationResponse;
   if (!isHttpStatusSuccess(status)) {
     // if error, successfully return status, message and original destination response
-    redditRespHandler(destinationResponse);
+    redditRespHandler(responseParams);
   }
   const { response } = destinationResponse;
   const errorMessage =
     response.invalid_events && Array.isArray(response.invalid_events)
       ? response?.invalid_events[0]?.error_message
       : null;
   const destResp = errorMessage || destinationResponse;
+  const responseWithIndividualEvents = rudderJobMetadata.map((metadata) => ({
+    statusCode: 200,
+    metadata,
+    error: 'success',
+  }));
   // Mostly any error will not have a status of 2xx
   return {
     status,
     message,
-    destResp,
+    destinationResponse: destResp,
+    response: responseWithIndividualEvents,
   };
 };
 // eslint-disable-next-line @typescript-eslint/naming-convention

test/integrations/destinations/reddit/dataDelivery/business.ts

@@ -39,6 +39,71 @@ const validRequestPayload = {
   ],
 };
 
+const validRequestMultipleEventsInPayload = {
+  events: [
+    {
+      event_at: '2019-10-14T09:03:17.562Z',
+      event_type: {
+        tracking_type: 'Purchase',
+      },
+      user: {
+        aaid: 'c12d34889302d3c656b5699fa9190b51c50d6f62fce57e13bd56b503d66c487a',
+        email: 'ac144532d9e4efeab19475d9253a879173ea12a3d2238d1cb8a332a7b3a105f2',
+        external_id: '7b023241a3132b792a5a33915a5afb3133cbb1e13d72879689bf6504de3b036d',
+        ip_address: 'e80bd55a3834b7c2a34ade23c7ecb54d2a49838227080f50716151e765a619db',
+        user_agent:
+          'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
+        screen_dimensions: {},
+      },
+      event_metadata: {
+        item_count: 3,
+        products: [
+          {
+            id: '123',
+            name: 'Monopoly',
+            category: 'Games',
+          },
+          {
+            id: '345',
+            name: 'UNO',
+            category: 'Games',
+          },
+        ],
+      },
+    },
+    {
+      event_at: '2018-10-14T09:03:17.562Z',
+      event_type: {
+        tracking_type: 'Purchase',
+      },
+      user: {
+        aaid: 'c12d34889302d3c656b5699fa9190b51c50d6f62fce57e13bd56b503d66c487a',
+        email: 'ac144532d9e4efeab19475d9253a879173ea12a3d2238d1cb8a332a7b3a105f2',
+        external_id: '7b023241a3132b792a5a33915a5afb3133cbb1e13d72879689bf6504de3b036d',
+        ip_address: 'e80bd55a3834b7c2a34ade23c7ecb54d2a49838227080f50716151e765a619db',
+        user_agent:
+          'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36',
+        screen_dimensions: {},
+      },
+      event_metadata: {
+        item_count: 3,
+        products: [
+          {
+            id: '123',
+            name: 'Monopoly',
+            category: 'Games',
+          },
+          {
+            id: '345',
+            name: 'UNO',
+            category: 'Games',
+          },
+        ],
+      },
+    },
+  ],
+};
+
 const commonHeaders = {
   Authorization: 'Bearer dummyAccessToken',
   'Content-Type': 'application/json',
@@ -73,13 +138,14 @@ export const testScenariosForV0API = [
         status: 200,
         body: {
           output: {
-            destResp: {
+            destinationResponse: {
               response: {
                 message: 'Successfully processed 1 conversion events.',
               },
               status: 200,
             },
-            message: 'Request Processed Successfully',
+            message:
+              '[Generic Response Handler] Request for destination: reddit Processed Successfully',
             status: 200,
           },
         },
@@ -116,8 +182,15 @@ export const testScenariosForV1API = [
         body: {
           output: {
             message: 'Request Processed Successfully',
+            destinationResponse: {
+              response: {
+                message: 'Successfully processed 1 conversion events.',
+              },
+              status: 200,
+            },
             response: [
               {
+                error: 'success',
                 metadata: generateMetadata(1),
                 statusCode: 200,
               },
@@ -180,4 +253,63 @@ export const testScenariosForV1API = [
       },
     },
   },
+  {
+    id: 'reddit_v1_scenario_3',
+    name: 'reddit',
+    description:
+      '[Proxy v1 API] :: Test for a valid request with a partial event failure from the destination',
+    scenario: 'PartialFailure',
+    feature: 'dataDelivery',
+    module: 'destination',
+    version: 'v1',
+    input: {
+      request: {
+        body: generateProxyV1Payload(
+          {
+            headers: commonHeaders,
+            JSON: validRequestMultipleEventsInPayload,
+            endpoint:
+              'https://ads-api.reddit.com/api/v2.0/conversions/events/partial_failed_events',
+          },
+          [generateMetadata(1), generateMetadata(2)],
+        ),
+        method: 'POST',
+      },
+    },
+    output: {
+      response: {
+        status: 200,
+        body: {
+          output: {
+            message: 'REDDIT: Error transformer proxy during REDDIT response transformation',
+            response: [
+              {
+                metadata: { ...generateMetadata(1), dontBatch: true },
+                statusCode: 500,
+                error:
+                  '{"message":"There were 1 invalid conversion events. None were processed.","invalid_events":[{"error_message":"event_at timestamp must be less than 168h0m0s old","event":{"event_at":"2018-10-14T09:03:17.562Z","event_type":{"tracking_type":"Purchase"},"event_metadata":{"item_count":0,"products":[{}],"conversion_id":"c054005afd85a4de74638a776eb8348d44ee875184d7a401830705b7a06e7df1"},"user":{"aaid":"c12d34889302d3c656b5699fa9190b51c50d6f62fce57e13bd56b503d66c487a","email":"ac144532d9e4efeab19475d9253a879173ea12a3d2238d1cb8a332a7b3a105f2","external_id":"7b023241a3132b792a5a33915a5afb3133cbb1e13d72879689bf6504de3b036d","ip_address":"e80bd55a3834b7c2a34ade23c7ecb54d2a49838227080f50716151e765a619db","user_agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36","screen_dimensions":{}}}}]}',
+              },
+              {
+                metadata: { ...generateMetadata(2), dontBatch: true },
+                statusCode: 500,
+                error:
+                  '{"message":"There were 1 invalid conversion events. None were processed.","invalid_events":[{"error_message":"event_at timestamp must be less than 168h0m0s old","event":{"event_at":"2018-10-14T09:03:17.562Z","event_type":{"tracking_type":"Purchase"},"event_metadata":{"item_count":0,"products":[{}],"conversion_id":"c054005afd85a4de74638a776eb8348d44ee875184d7a401830705b7a06e7df1"},"user":{"aaid":"c12d34889302d3c656b5699fa9190b51c50d6f62fce57e13bd56b503d66c487a","email":"ac144532d9e4efeab19475d9253a879173ea12a3d2238d1cb8a332a7b3a105f2","external_id":"7b023241a3132b792a5a33915a5afb3133cbb1e13d72879689bf6504de3b036d","ip_address":"e80bd55a3834b7c2a34ade23c7ecb54d2a49838227080f50716151e765a619db","user_agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.90 Safari/537.36","screen_dimensions":{}}}}]}',
+              },
+            ],
+            statTags: {
+              destType: 'REDDIT',
+              destinationId: 'default-destinationId',
+              errorCategory: 'network',
+              errorType: 'retryable',
+              feature: 'dataDelivery',
+              implementation: 'native',
+              module: 'destination',
+              workspaceId: 'default-workspaceId',
+            },
+            status: 500,
+          },
+        },
+      },
+    },
+  },
 ];

test/integrations/destinations/reddit/dataDelivery/oauth.ts

@@ -84,14 +84,13 @@ export const v0oauthScenarios = [
         status: 401,
         body: {
           output: {
-            authErrorCategory: 'REFRESH_TOKEN',
             destinationResponse: {
               response: 'Authorization Required',
               status: 401,
             },
             message:
-              'Request failed due to Authorization Required during reddit response transformation',
-            statTags: expectedStatTags,
+              '[Generic Response Handler] Request failed for destination reddit with status: 401',
+            statTags: { ...expectedStatTags, errorType: 'aborted' },
             status: 401,
           },
         },
@@ -121,7 +120,6 @@ export const v0oauthScenarios = [
         status: 401,
         body: {
           output: {
-            authErrorCategory: 'REFRESH_TOKEN',
             destinationResponse: {
               response: {
                 success: false,
@@ -134,8 +132,8 @@ export const v0oauthScenarios = [
               status: 401,
             },
             message:
-              'This server could not verify that you are authorized to access the document you requested. during reddit response transformation',
-            statTags: expectedStatTags,
+              '[Generic Response Handler] Request failed for destination reddit with status: 401',
+            statTags: { ...expectedStatTags, errorType: 'aborted' },
             status: 401,
           },
         },