feat: cm360 transformerproxy V1 features.json update

src/adapters/networkHandlerFactory.js

@@ -8,6 +8,8 @@ const { getIntegrations } = require('../routes/utils');
 
 const handlers = {
   generic: GenericNetworkHandler,
+  v0: {},
+  v1: {},
 };
 
 // Dynamically import the network handlers for all
@@ -16,16 +18,26 @@ SUPPORTED_VERSIONS.forEach((version) => {
   const destinations = getIntegrations(path.resolve(__dirname, `../${version}/destinations`));
   destinations.forEach((dest) => {
     try {
-      handlers[dest] = require(`../${version}/destinations/${dest}/networkHandler`).networkHandler;
+      // handles = {
+      //   v0: {
+      //     dest: handler
+      //   },
+      //   v1: {
+      //     dest: handler
+      //   },
+      // generic: GenericNetworkHandler,
+      // }
+      handlers[version][dest] =
+        require(`../${version}/destinations/${dest}/networkHandler`).networkHandler;
     } catch {
       // Do nothing as exception indicates
       // network handler is not defined for that destination
     }
   });
 });
 
-const getNetworkHandler = (type) => {
-  const NetworkHandler = handlers[type] || handlers.generic;
+const getNetworkHandler = (type, version) => {
+  const NetworkHandler = handlers[version][type] || handlers.generic;
   return new NetworkHandler();
 };
 

src/adapters/networkHandlerFactory.test.js

@@ -0,0 +1,25 @@
+const { getNetworkHandler } = require('./networkHandlerFactory');
+const { networkHandler: GenericNetworkHandler } = require('./networkhandler/genericNetworkHandler');
+
+describe(`Network Handler Tests`, () => {
+  it('Should return v0 networkhandler', () => {
+    let proxyHandler = getNetworkHandler('campaign_manager', `v0`);
+    const cmProxy = require(`../v0/destinations/campaign_manager/networkHandler`).networkHandler;
+    expect(proxyHandler).toEqual(new cmProxy());
+
+    proxyHandler = getNetworkHandler('braze', `v0`);
+    const brazeProxy = require(`../v0/destinations/braze/networkHandler`).networkHandler;
+    expect(proxyHandler).toEqual(new brazeProxy());
+  });
+
+  it('Should return v1 networkhandler', () => {
+    let proxyHandler = getNetworkHandler('campaign_manager', `v1`);
+    const cmProxy = require(`../v1/destinations/campaign_manager/networkHandler`).networkHandler;
+    expect(proxyHandler).toEqual(new cmProxy());
+  });
+
+  it('Should return genericHandler if v1 proxy and handler is not present for destination', () => {
+    let proxyHandler = getNetworkHandler('braze', `v1`);
+    expect(proxyHandler).toEqual(new GenericNetworkHandler());
+  });
+});

src/controllers/delivery.ts

@@ -16,9 +16,15 @@ export class DeliveryController {
     const requestMetadata = MiscService.getRequestMetadata(ctx);
     const event = ctx.request.body as ProcessorTransformationOutput;
     const { destination }: { destination: string } = ctx.params;
+    const { version }: { version: string } = ctx.params;
     const integrationService = ServiceSelector.getNativeDestinationService();
     try {
-      deliveryResponse = await integrationService.deliver(event, destination, requestMetadata);
+      deliveryResponse = await integrationService.deliver(
+        event,
+        destination,
+        requestMetadata,
+        version,
+      );
     } catch (error: any) {
       const metaTO = integrationService.getTags(
         destination,

src/features.json

@@ -64,5 +64,6 @@
     "ONE_SIGNAL": true,
     "TIKTOK_AUDIENCE": true
   },
-  "supportSourceTransformV1": true
+  "supportSourceTransformV1": true,
+  "transformerProxyV1": true
 }

src/interfaces/DestinationService.ts

@@ -47,6 +47,7 @@ export interface DestinationService {
     event: ProcessorTransformationOutput,
     destinationType: string,
     requestMetadata: NonNullable<unknown>,
+    version: string,
   ): Promise<DeliveryResponse>;
 
   processUserDeletion(

src/routes/utils/constants.js

@@ -1,4 +1,4 @@
-const SUPPORTED_VERSIONS = ['v0'];
+const SUPPORTED_VERSIONS = ['v0', 'v1'];
 const API_VERSION = '2';
 const INTEGRATION_SERVICE = {
   COMPARATOR: 'comparator',
@@ -7,8 +7,8 @@ const INTEGRATION_SERVICE = {
   NATIVE_DEST: 'native_dest',
   NATIVE_SOURCE: 'native_source',
 };
-const CHANNELS= {
-  sources: 'sources'
+const CHANNELS = {
+  sources: 'sources',
 };
 
 const RETL_TIMESTAMP = 'timestamp';

src/services/comparator.ts

@@ -368,11 +368,13 @@
     event: ProcessorTransformationOutput,
     destinationType: string,
     requestMetadata: NonNullable<unknown>,
+    version: string,
   ): Promise<DeliveryResponse> {
     const primaryResplist = await this.primaryService.deliver(
       event,
       destinationType,
       requestMetadata,
+      version,
     );
     logger.error('[LIVE_COMPARE_TEST] not implemented for delivery routine');
 

src/services/destination/nativeIntegration.ts

@@ -172,9 +172,10 @@ export class NativeIntegrationDestinationService implements DestinationService {
     destinationRequest: ProcessorTransformationOutput,
     destinationType: string,
     _requestMetadata: NonNullable<unknown>,
+    version: string,
   ): Promise<DeliveryResponse> {
     try {
-      const networkHandler = networkHandlerFactory.getNetworkHandler(destinationType);
+      const networkHandler = networkHandlerFactory.getNetworkHandler(destinationType, version);
       const rawProxyResponse = await networkHandler.proxy(destinationRequest, destinationType);
       const processedProxyResponse = networkHandler.processAxiosResponse(rawProxyResponse);
       return networkHandler.responseHandler(

src/v1/destinations/campaign_manager/networkHandler.js

@@ -0,0 +1,117 @@
+/* eslint-disable no-param-reassign */
+/* eslint-disable no-restricted-syntax */
+const { NetworkError } = require('@rudderstack/integrations-lib');
+const { prepareProxyRequest, proxyRequest } = require('../../../adapters/network');
+const { isHttpStatusSuccess, getAuthErrCategoryFromStCode } = require('../../../v0/util/index');
+
+const {
+  processAxiosResponse,
+  getDynamicErrorType,
+} = require('../../../adapters/utils/networkUtils');
+const tags = require('../../../v0/util/tags');
+
+function isEventRetryableAndExtractErrMsg(element, proxyOutputObj) {
+  let isRetryable = false;
+  let errorMsg = '';
+  // success event
+  if (!element.errors) {
+    return isRetryable;
+  }
+  for (const err of element.errors) {
+    errorMsg += `${err.message}, `;
+    if (err.code === 'INTERNAL') {
+      isRetryable = true;
+    }
+  }
+  if (errorMsg) {
+    proxyOutputObj.error = errorMsg;
+  }
+  return isRetryable;
+}
+
+function isEventAbortableAndExtractErrMsg(element, proxyOutputObj) {
+  let isAbortable = false;
+  let errorMsg = '';
+  // success event
+  if (!element.errors) {
+    return isAbortable;
+  }
+  for (const err of element.errors) {
+    errorMsg += `${err.message}, `;
+    // if code is any of these, event is not retryable
+    if (
+      err.code === 'PERMISSION_DENIED' ||
+      err.code === 'INVALID_ARGUMENT' ||
+      err.code === 'NOT_FOUND'
+    ) {
+      isAbortable = true;
+    }
+  }
+  if (errorMsg) {
+    proxyOutputObj.error = errorMsg;
+  }
+  return isAbortable;
+}
+
+const responseHandler = (destinationResponse) => {
+  const message = `[CAMPAIGN_MANAGER Response V1 Handler] - Request Processed Successfully`;
+  const responseWithIndividualEvents = [];
+  const { response, status, rudderJobMetadata } = destinationResponse;
+
+  if (isHttpStatusSuccess(status)) {
+    // check for Partial Event failures and Successes
+    const destPartialStatus = response.status;
+
+    for (const [idx, element] of destPartialStatus.entries()) {
+      const proxyOutputObj = {
+        statusCode: 200,
+        metadata: rudderJobMetadata[idx],
+        error: 'success',
+      };
+      // update status of partial event as per retriable or abortable
+      if (isEventRetryableAndExtractErrMsg(element, proxyOutputObj)) {
+        proxyOutputObj.statusCode = 500;
+      } else if (isEventAbortableAndExtractErrMsg(element, proxyOutputObj)) {
+        proxyOutputObj.statusCode = 400;
+      }
+      responseWithIndividualEvents.push(proxyOutputObj);
+    }
+
+    return {
+      status,
+      message,
+      destinationResponse,
+      response: responseWithIndividualEvents,
+    };
+  }
+
+  // in case of failure status, populate response to maintain len(metadata)=len(response)
+  const errorMessage = response.error?.message || 'unknown error format';
+  for (const metadata of rudderJobMetadata) {
+    responseWithIndividualEvents.push({
+      statusCode: 500,
+      metadata,
+      error: errorMessage,
+    });
+  }
+
+  throw new NetworkError(
+    `Campaign Manager: Error proxy v1 during CAMPAIGN_MANAGER response transformation`,
+    500,
+    {
+      [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
+    },
+    destinationResponse,
+    getAuthErrCategoryFromStCode(status),
+    responseWithIndividualEvents,
+  );
+};
+
+function networkHandler() {
+  this.prepareProxy = prepareProxyRequest;
+  this.proxy = proxyRequest;
+  this.processAxiosResponse = processAxiosResponse;
+  this.responseHandler = responseHandler;
+}
+
+module.exports = { networkHandler };