[Linkedin Audiences] - Company Audience Sync Action

packages/destination-actions/src/destinations/linkedin-audiences/api.ts

@@ -1,9 +1,15 @@
 import type { RequestClient, ModifiedResponse } from '@segment/actions-core'
-
 import type { Settings } from './generated-types'
-import type { Payload } from './updateAudience/generated-types'
-import { BASE_URL, LINKEDIN_SOURCE_PLATFORM } from './constants'
-import type { ProfileAPIResponse, AdAccountUserResponse, LinkedInAudiencePayload } from './types'
+import { BASE_URL, LINKEDIN_SOURCE_PLATFORM, SEGMENT_TYPES } from './constants'
+import type { 
+  ProfileAPIResponse, 
+  AdAccountUserResponse, 
+  AudienceJSON, 
+  LinkedInBatchUpdateResponse, 
+  CreateDMPSegmentResponse, 
+  GetDMPSegmentResponse, 
+  SegmentType 
+} from './types'
 
 export class LinkedInAudiences {
   request: RequestClient
@@ -27,27 +33,27 @@ export class LinkedInAudiences {
     )
   }
 
-  async getDmpSegment(settings: Settings, payload: Payload): Promise<ModifiedResponse> {
+  async getDmpSegment(settings: Settings, sourceSegmentId = ''): Promise<ModifiedResponse<GetDMPSegmentResponse>> {
     return this.request(`${BASE_URL}/dmpSegments`, {
       method: 'GET',
       searchParams: {
         q: 'account',
         account: `urn:li:sponsoredAccount:${settings.ad_account_id}`,
-        sourceSegmentId: payload.personas_audience_key || '',
+        sourceSegmentId,
         sourcePlatform: LINKEDIN_SOURCE_PLATFORM
       }
     })
   }
 
-  async createDmpSegment(settings: Settings, payload: Payload): Promise<ModifiedResponse> {
+  async createDmpSegment(settings: Settings, sourceSegmentId: string, segmentName: string, segmentType: SegmentType): Promise<ModifiedResponse<CreateDMPSegmentResponse>> {
     return this.request(`${BASE_URL}/dmpSegments`, {
       method: 'POST',
       json: {
-        name: payload.dmp_segment_name,
+        name: segmentName,
         sourcePlatform: LINKEDIN_SOURCE_PLATFORM,
-        sourceSegmentId: payload.personas_audience_key,
+        sourceSegmentId,
         account: `urn:li:sponsoredAccount:${settings.ad_account_id}`,
-        type: 'USER',
+        type: segmentType,
         destinations: [
           {
             destination: 'LINKEDIN'
@@ -57,15 +63,14 @@ export class LinkedInAudiences {
     })
   }
 
-  async batchUpdate(dmpSegmentId: string, elements: LinkedInAudiencePayload[]): Promise<ModifiedResponse> {
-    return this.request(`${BASE_URL}/dmpSegments/${dmpSegmentId}/users`, {
+  async batchUpdate<E>(dmpSegmentId: string, json: AudienceJSON<E>, segmentType: SegmentType): Promise<ModifiedResponse<LinkedInBatchUpdateResponse>> {
+    const url = `${BASE_URL}/dmpSegments/${dmpSegmentId}/${segmentType === SEGMENT_TYPES.COMPANY ? 'companies' : 'users'}`
+    return this.request(url, {
       method: 'POST',
       headers: {
         'X-RestLi-Method': 'BATCH_CREATE'
       },
-      json: {
-        elements
-      },
+      json,
       throwHttpErrors: false
     })
   }

packages/destination-actions/src/destinations/linkedin-audiences/common-functions.ts

@@ -0,0 +1,145 @@
+import {StatsContext, MultiStatusResponse, RequestClient, PayloadValidationError, JSONLikeObject, RetryableError  } from '@segment/actions-core'
+import type { Settings } from './generated-types'
+import { LinkedInAudiences } from './api'
+import { SegmentType, AudienceJSON } from './types'
+import { SEGMENT_TYPES } from './constants'
+
+export async function send<P, E>(
+  request: RequestClient,
+  getSegmentSourceIdAndName: (payload: P) => { sourceSegmentId: string; segmentName: string },
+  buildJSON: (payloads: P[], settings: Settings) => AudienceJSON<E>,
+  validate: (payloads: P[], msResponse: MultiStatusResponse, isBatch: boolean, settings: Settings) => (P & { index: number })[],
+  settings: Settings,
+  payloads: P[],
+  segmentType: SegmentType,
+  isBatch: boolean,
+  statsContext: StatsContext | undefined
+) {
+  const msResponse = new MultiStatusResponse()
+  const validPayloads = validate(payloads, msResponse, isBatch, settings)
+
+  if(validPayloads.length === 0) {
+    if(isBatch){
+      return msResponse
+    } 
+    else {
+      throw new PayloadValidationError('No valid payloads to process after validation.')
+    }
+  }
+
+  const { sourceSegmentId, segmentName } = getSegmentSourceIdAndName(payloads[0])
+  const linkedinApiClient: LinkedInAudiences = new LinkedInAudiences(request)
+  const { id, type } = await getDmpSegmentIdAndType(linkedinApiClient, settings, sourceSegmentId, segmentName, segmentType, statsContext)
+
+  if(!id || typeof type != 'string'){
+    if(isBatch){
+      payloads.forEach((_, index) => {
+        msResponse.setErrorResponseAtIndex(index, {
+          status: 400,
+          errortype: 'PAYLOAD_VALIDATION_FAILED',
+          errormessage: `LinkedIn ${segmentType} Segment creation failed: segmentName: ${segmentName}, Source Segment Id ${sourceSegmentId}, type: ${type}, id: ${id}.`,
+          sent: payloads[index] as JSONLikeObject,
+          body: {segmentName, sourceSegmentId, type, segmentType} as JSONLikeObject
+        })
+      })
+      return msResponse
+    } 
+    else {
+      throw new PayloadValidationError(`LinkedIn ${segmentType} Segment creation failed: segmentName: ${segmentName}, Source Segment Id ${sourceSegmentId}, type: ${type}, id: ${id}.`)
+    }
+  }
+
+  if(typeof type == 'string' && type != segmentType) {
+    // reject all payloads if Segment Type mismatches
+    if(isBatch){
+      payloads.forEach((_, index) => {
+        msResponse.setErrorResponseAtIndex(index, {
+          status: 400,
+          errortype: 'PAYLOAD_VALIDATION_FAILED',
+          errormessage: `The existing DMP Segment with Source Segment Id ${sourceSegmentId} is of type ${type} and cannot be used to update a segment of type ${segmentType}.`,
+          sent: payloads[index] as JSONLikeObject,
+          body: {segmentName, sourceSegmentId, type} as JSONLikeObject
+        })
+      })
+      return msResponse
+    } 
+    else {
+      throw new PayloadValidationError(`The existing DMP Segment with Source Segment Id ${sourceSegmentId} is of type ${type} and cannot be used to update a segment of type ${segmentType}.`)
+    }
+  }
+
+  const json = buildJSON(validPayloads, settings)
+
+  statsContext?.statsClient?.incr('oauth_app_api_call', 1, [
+    ...statsContext?.tags,
+    `endpoint:add-or-remove-users-from-${segmentType === SEGMENT_TYPES.COMPANY ? 'abm-' : ''}dmpSegment`
+  ])
+
+  const response = await linkedinApiClient.batchUpdate(id, json, segmentType)
+
+  // At this point, if LinkedIn's API returns a 404 error, it's because the audience
+  // Segment just created isn't available yet for updates via this endpoint.
+  // Audiences are usually available to accept batches of data 1 - 2 minutes after
+  // they're created. Here, we'll throw an error and let Centrifuge handle the retry.
+  if (response.status !== 200) {
+    throw new RetryableError('Error while attempting to update LinkedIn DMP Segment. This batch will be retried.')
+  }
+  if(isBatch) {
+    const sentElements = json.elements
+
+    validPayloads.forEach((payload, index) => {
+      const e = response.data.elements[index]
+      if(e.status >= 200 && e.status < 300) {
+        msResponse.setSuccessResponseAtIndex(payload.index, {
+          status: e.status,
+          sent: payload as JSONLikeObject,
+          body: { elements: [sentElements[index]]} as JSONLikeObject
+        })
+      } 
+      else {
+        msResponse.setErrorResponseAtIndex(payload.index, {
+          status: e.status,
+          errortype: 'BAD_REQUEST',
+          errormessage: e.message || 'Failed to update LinkedIn Audience',
+          sent: payload as JSONLikeObject,
+          body: { elements: [sentElements[index]]} as JSONLikeObject
+        })
+      }
+    })
+    return msResponse
+  } 
+  else {
+    return response
+  }
+}
+
+async function getDmpSegmentIdAndType(
+  linkedinApiClient: LinkedInAudiences,
+  settings: Settings,
+  sourceSegmentId: string,
+  segmentName: string,
+  segmentType: SegmentType,
+  statsContext: StatsContext | undefined
+): Promise<{ id: string; type: SegmentType }> {
+  statsContext?.statsClient?.incr('oauth_app_api_call', 1, [...statsContext?.tags, `endpoint:get-${segmentType === SEGMENT_TYPES.COMPANY ? 'abm-' : ''}dmpSegment`])
+  const response = await linkedinApiClient.getDmpSegment(settings, sourceSegmentId)
+  const { id, type } = response.data?.elements?.[0] || {}
+  if (id && type) {
+    return { id, type }
+  }
+  return createDmpSegment(linkedinApiClient, settings, sourceSegmentId, segmentName, segmentType, statsContext)
+}
+
+async function createDmpSegment(
+  linkedinApiClient: LinkedInAudiences,
+  settings: Settings,
+  sourceSegmentId: string,
+  segmentName: string,
+  segmentType: SegmentType,
+  statsContext: StatsContext | undefined
+): Promise<{ id: string; type: SegmentType }> {
+  statsContext?.statsClient?.incr('oauth_app_api_call', 1, [...statsContext?.tags, `endpoint:create-${segmentType === SEGMENT_TYPES.COMPANY ? 'abm-' : ''}dmpSegment`])
+  const res = await linkedinApiClient.createDmpSegment(settings, sourceSegmentId, segmentName, segmentType)
+  const { id, type } = res.data
+  return { id, type }
+}

packages/destination-actions/src/destinations/linkedin-audiences/constants.ts

@@ -1,3 +1,12 @@
 export const LINKEDIN_API_VERSION = '202505'
 export const BASE_URL = 'https://api.linkedin.com/rest'
 export const LINKEDIN_SOURCE_PLATFORM = 'SEGMENT'
+export const SEGMENT_TYPES = {
+  USER: 'USER',
+  COMPANY: 'COMPANY'
+} as const
+
+export const AUDIENCE_ACTION = {
+  ADD: 'ADD',
+  REMOVE: 'REMOVE'
+} as const