feat: onboard linkedin audience destination (#3857)

* feat: onboard linkedin audience destination

* chore: resolve conflicts

* chore: add test cases

* chore: fix lint errors

* refactor: linkedin audiences proc workflow

---------

Co-authored-by: Dilip Kola <[email protected]>

src/cdk/v2/destinations/linkedin_audience/config.ts

@@ -0,0 +1,10 @@
+export const SUPPORTED_EVENT_TYPE = 'record';
+export const ACTION_TYPES = ['insert', 'delete'];
+export const BASE_ENDPOINT = 'https://api.linkedin.com/rest';
+export const USER_ENDPOINT = '/dmpSegments/audienceId/users';
+export const COMPANY_ENDPOINT = '/dmpSegments/audienceId/companies';
+export const FIELD_MAP = {
+  sha256Email: 'SHA256_EMAIL',
+  sha512Email: 'SHA512_EMAIL',
+  googleAid: 'GOOGLE_AID',
+};

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

@@ -0,0 +1,89 @@
+bindings:
+  - path: ./config
+    exportAll: true
+  - path: ./utils
+    exportAll: true
+  - name: defaultRequestConfig
+    path: ../../../../v0/util
+
+steps:
+  - name: validateInput
+    description: Validate input, if all the required fields are available or not.
+    template: |
+      const config = .connection.config.destination;
+      const secret = .metadata.secret;
+      let messageType = .message.type;
+      $.assertConfig(config.audienceId, "Audience Id is not present. Aborting");
+      $.assertConfig(secret.accessToken, "Access Token is not present. Aborting");
+      $.assertConfig(config.audienceType, "audienceType is not present. Aborting");
+      $.assert(messageType, "Message Type is not present. Aborting message.");
+      $.assert(messageType.toLowerCase() === $.SUPPORTED_EVENT_TYPE, `Event type ${.message.type.toLowerCase()} is not supported. Aborting message.`);
+      $.assert(.message.fields, "`fields` is not present. Aborting message.");
+      $.assert(.message.identifiers, "`identifiers` is not present inside properties. Aborting message.");
+      $.assert($.containsAll([.message.action], $.ACTION_TYPES), "Unsupported action type. Aborting message.")
+
+  - name: getConfigs
+    description: This step fetches the configs from different places and combines them.
+    template: |
+      const config = .connection.config.destination;
+      {
+        audienceType: config.audienceType,
+        audienceId: config.audienceId,
+        accessToken: .metadata.secret.accessToken,
+        isHashRequired: config.isHashRequired,
+      }
+
+  - name: prepareUserTypeBasePayload
+    condition: $.outputs.getConfigs.audienceType === 'user'
+    steps:
+      - name: prepareUserIds
+        description: Prepare user ids for user audience type
+        template: |
+          const identifiers = $.outputs.getConfigs.isHashRequired === true ? 
+            $.hashIdentifiers(.message.identifiers) :
+            .message.identifiers;
+          $.prepareUserIds(identifiers)
+
+      - name: preparePayload
+        description: Prepare base payload for user audiences
+        template: |
+          const payload = {
+            'elements': [
+              {
+                'action': $.generateActionType(.message.action),
+                'userIds': $.outputs.prepareUserTypeBasePayload.prepareUserIds,
+                ....message.fields
+              }
+            ]
+          }
+          payload;
+
+  - name: prepareCompanyTypeBasePayload
+    description: Prepare base payload for company audiences
+    condition: $.outputs.getConfigs.audienceType === 'company'
+    template: |
+      const payload = {
+        'elements': [
+          {
+            'action': $.generateActionType(.message.action),
+            ....message.identifiers,
+            ....message.fields
+          }
+        ]
+      }
+      payload;
+
+  - name: buildResponseForProcessTransformation
+    description: build response depending upon batch size
+    template: |
+      const response = $.defaultRequestConfig();
+      response.body.JSON = {...$.outputs.prepareUserTypeBasePayload, ...$.outputs.prepareCompanyTypeBasePayload};
+      response.endpoint = $.generateEndpoint($.outputs.getConfigs.audienceType, $.outputs.getConfigs.audienceId);
+      response.headers = {
+          "Authorization": "Bearer " + $.outputs.getConfigs.accessToken,
+          "Content-Type": "application/json",
+          "X-RestLi-Method": "BATCH_CREATE",
+          "X-Restli-Protocol-Version": "2.0.0",
+          "LinkedIn-Version": "202409"
+        };
+      response;

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

@@ -0,0 +1,40 @@
+bindings:
+  - path: ./utils
+  - name: handleRtTfSingleEventError
+    path: ../../../../v0/util/index
+
+steps:
+  - name: validateInput
+    template: |
+      $.assert(Array.isArray(^) && ^.length > 0, "Invalid event array")
+
+  - name: transform
+    externalWorkflow:
+      path: ./procWorkflow.yaml
+      bindings:
+        - name: batchMode
+          value: true
+    loopOverInput: true
+
+  - name: successfulEvents
+    template: |
+      $.outputs.transform#idx.output.({
+        "message": .[],
+        "destination": ^ [idx].destination,
+        "metadata": ^ [idx].metadata
+      })[]
+
+  - name: failedEvents
+    template: |
+      $.outputs.transform#idx.error.(
+        $.handleRtTfSingleEventError(^[idx], .originalError ?? ., {})
+      )[]
+
+  - name: batchSuccessfulEvents
+    description: Batches the successfulEvents
+    template: |
+      $.batchResponseBuilder($.outputs.successfulEvents);
+
+  - name: finalPayload
+    template: |
+      [...$.outputs.batchSuccessfulEvents, ...$.outputs.failedEvents]

src/cdk/v2/destinations/linkedin_audience/utils.ts

@@ -0,0 +1,87 @@
+import lodash from 'lodash';
+import { hashToSha256 } from '@rudderstack/integrations-lib';
+import { createHash } from 'crypto';
+import { BASE_ENDPOINT, COMPANY_ENDPOINT, FIELD_MAP, USER_ENDPOINT } from './config';
+
+export function hashIdentifiers(identifiers: string[]): Record<string, string> {
+  const hashedIdentifiers = {};
+  Object.keys(identifiers).forEach((key) => {
+    if (key === 'sha256Email') {
+      hashedIdentifiers[key] = hashToSha256(identifiers[key]);
+    } else if (key === 'sha512Email') {
+      hashedIdentifiers[key] = createHash('sha512').update(identifiers[key]).digest('hex');
+    } else {
+      hashedIdentifiers[key] = identifiers[key];
+    }
+  });
+  return hashedIdentifiers;
+}
+
+export function prepareUserIds(
+  identifiers: Record<string, string>,
+): { idType: string; idValue: string }[] {
+  const userIds: { idType: string; idValue: string }[] = [];
+  Object.keys(identifiers).forEach((key) => {
+    userIds.push({ idType: FIELD_MAP[key], idValue: identifiers[key] });
+  });
+  return userIds;
+}
+
+export function generateEndpoint(audienceType: string, audienceId: string) {
+  if (audienceType === 'user') {
+    return BASE_ENDPOINT + USER_ENDPOINT.replace('audienceId', audienceId);
+  }
+  return BASE_ENDPOINT + COMPANY_ENDPOINT.replace('audienceId', audienceId);
+}
+
+export function batchResponseBuilder(successfulEvents) {
+  const chunkOnActionType = lodash.groupBy(
+    successfulEvents,
+    (event) => event.message[0].body.JSON.elements[0].action,
+  );
+  const result: any = [];
+  Object.keys(chunkOnActionType).forEach((actionType) => {
+    const firstEvent = chunkOnActionType[actionType][0];
+    const { method, endpoint, headers, type, version } = firstEvent.message[0];
+    const batchEvent = {
+      batchedRequest: {
+        body: {
+          JSON: { elements: firstEvent.message[0].body.JSON.elements },
+          JSON_ARRAY: {},
+          XML: {},
+          FORM: {},
+        },
+        version,
+        type,
+        method,
+        endpoint,
+        headers,
+        params: {},
+        files: {},
+      },
+      metadata: [firstEvent.metadata],
+      batched: true,
+      statusCode: 200,
+      destination: firstEvent.destination,
+    };
+    firstEvent.metadata = [firstEvent.metadata];
+    chunkOnActionType[actionType].forEach((element, index) => {
+      if (index !== 0) {
+        batchEvent.batchedRequest.body.JSON.elements.push(element.message[0].body.JSON.elements[0]);
+        batchEvent.metadata.push(element.metadata);
+      }
+    });
+    result.push(batchEvent);
+  });
+  return result;
+}
+
+export const generateActionType = (actionType: string): string => {
+  if (actionType === 'insert') {
+    return 'ADD';
+  }
+  if (actionType === 'delete') {
+    return 'REMOVE';
+  }
+  return actionType;
+};

src/features.ts

@@ -92,6 +92,7 @@ const defaultFeaturesConfig: FeaturesConfig = {
     HTTP: true,
     AMAZON_AUDIENCE: true,
     INTERCOM_V2: true,
+    LINKEDIN_AUDIENCE: true,
   },
   regulations: [
     'BRAZE',