Add multistatus in linkedin conversions' stream conversion event

Author: harsh-joshi99

packages/destination-actions/src/destinations/linkedin-conversions/api/index.ts

@@ -3,7 +3,12 @@ import {
   ModifiedResponse,
   DynamicFieldResponse,
   ActionHookResponse,
-  PayloadValidationError
+  PayloadValidationError,
+  JSONLikeObject,
+  MultiStatusResponse,
+  HTTPError,
+  InvalidAuthenticationError,
+  ErrorCodes
 } from '@segment/actions-core'
 import { BASE_URL, DEFAULT_POST_CLICK_LOOKBACK_WINDOW, DEFAULT_VIEW_THROUGH_LOOKBACK_WINDOW } from '../constants'
 import type {
@@ -471,39 +476,95 @@ export class LinkedInConversions {
     })
   }
 
-  async batchConversionAdd(payloads: Payload[]): Promise<ModifiedResponse> {
-    return this.request(`${BASE_URL}/conversionEvents`, {
-      method: 'post',
-      headers: {
-        'X-RestLi-Method': 'BATCH_CREATE'
-      },
-      json: {
-        elements: [
-          ...payloads.map((payload) => {
-            const conversionTime = isNotEpochTimestampInMilliseconds(payload.conversionHappenedAt)
-              ? convertToEpochMillis(payload.conversionHappenedAt)
-              : Number(payload.conversionHappenedAt)
-            validate(payload, conversionTime)
-
-            const userIds = this.buildUserIdsArray(payload)
-            return {
-              conversion: `urn:lla:llaPartnerConversion:${this.conversionRuleId}`,
-              conversionHappenedAt: conversionTime,
-              conversionValue: payload.conversionValue,
-              eventId: payload.eventId,
-              user: {
-                userIds,
-                userInfo: payload.userInfo,
-                // only 1 externalId value allowed currently in the externalIds array by LinkedIn currently Oct 2025
-                ...(Array.isArray(payload?.externalIds) && payload.externalIds.length > 0
-                  ? { externalIds: [payload.externalIds[0]] }
-                  : {})
-              }
-            }
-          })
-        ]
+  async batchConversionAdd(payloads: Payload[]): Promise<MultiStatusResponse> {
+    const multiStatusResponse = new MultiStatusResponse()
+    const validElements: ReturnType<typeof this.buildConversionElement>[] = []
+    const validResponseIndices: number[] = []
+
+    payloads.forEach((payload, i) => {
+      const conversionTime = isNotEpochTimestampInMilliseconds(payload.conversionHappenedAt)
+        ? convertToEpochMillis(payload.conversionHappenedAt)
+        : Number(payload.conversionHappenedAt)
+
+      try {
+        validate(payload, conversionTime)
+        validElements.push(this.buildConversionElement(payload, conversionTime))
+        validResponseIndices.push(i)
+      } catch (e) {
+        multiStatusResponse.setErrorResponseAtIndex(i, {
+          status: 400,
+          errortype: 'PAYLOAD_VALIDATION_FAILED',
+          errormessage: (e as Error).message
+        })
       }
     })
+
+    if (validElements.length === 0) {
+      return multiStatusResponse
+    }
+
+    try {
+      const response = await this.request(`${BASE_URL}/conversionEvents`, {
+        method: 'post',
+        headers: {
+          'X-RestLi-Method': 'BATCH_CREATE'
+        },
+        json: {
+          elements: validElements
+        }
+      })
+
+      validResponseIndices.forEach((originalIndex, filteredIndex) => {
+        multiStatusResponse.setSuccessResponseAtIndex(originalIndex, {
+          status: 201,
+          sent: validElements[filteredIndex] as unknown as JSONLikeObject,
+          body: (response.content ?? response.data ?? '') as unknown as JSONLikeObject
+        })
+      })
+    } catch (error) {
+      if (error instanceof HTTPError) {
+        const status = error.response.status
+
+        // 401 means the OAuth token is expired — re-throw so the platform triggers a token refresh
+        if (status === 401) {
+          throw new InvalidAuthenticationError(
+            'LinkedIn OAuth token is expired or invalid. Please re-authenticate.',
+            ErrorCodes.INVALID_AUTHENTICATION
+          )
+        }
+
+        // For all other API errors, mark every valid event as failed with the actual status
+        validResponseIndices.forEach((originalIndex, filteredIndex) => {
+          multiStatusResponse.setErrorResponseAtIndex(originalIndex, {
+            status,
+            errormessage: error.message,
+            sent: validElements[filteredIndex] as unknown as JSONLikeObject
+          })
+        })
+      } else {
+        throw error
+      }
+    }
+
+    return multiStatusResponse
+  }
+
+  private buildConversionElement(payload: Payload, conversionTime: number) {
+    const userIds = this.buildUserIdsArray(payload)
+    return {
+      conversion: `urn:lla:llaPartnerConversion:${this.conversionRuleId}`,
+      conversionHappenedAt: conversionTime,
+      conversionValue: payload.conversionValue,
+      eventId: payload.eventId,
+      user: {
+        userIds,
+        userInfo: payload.userInfo,
+        // only 1 externalId value allowed currently in the externalIds array by LinkedIn currently Oct 2025
+        ...(Array.isArray(payload?.externalIds) && payload.externalIds.length > 0
+          ? { externalIds: [payload.externalIds[0]] }
+          : {})
+      }
+    }
   }
 
   async bulkAssociateCampaignToConversion(campaignIds?: string[]): Promise<ModifiedResponse | void> {

packages/destination-actions/src/destinations/linkedin-conversions/streamConversion/__tests__/index.test.ts

@@ -1211,3 +1211,177 @@ describe('LinkedinConversions.onMappingSave - performHook', () => {
     })
   })
 })
+
+describe('LinkedinConversions.multistatus', () => {
+  const staleTimestamp = new Date(Date.now() - 100 * 24 * 60 * 60 * 1000).toISOString()
+
+  const staleEvent = createTestEvent({
+    messageId: 'stale-event-id',
+    event: 'Funding Application Submitted',
+    type: 'track',
+    timestamp: staleTimestamp,
+    context: { traits: { email: '[email protected]' } },
+    properties: {}
+  })
+
+  const batchMapping = {
+    email: { '@path': '$.context.traits.email' },
+    conversionHappenedAt: { '@path': '$.timestamp' },
+    onMappingSave: {
+      inputs: {},
+      outputs: { id: payload.conversionId }
+    },
+    enable_batching: true,
+    batch_size: 5000
+  }
+
+  it('should mark stale event as PAYLOAD_VALIDATION_FAILED and still send valid events', async () => {
+    nock(`${BASE_URL}/conversionEvents`)
+      .post('', {
+        elements: [
+          {
+            conversion: `urn:lla:llaPartnerConversion:${payload.conversionId}`,
+            conversionHappenedAt: currentTimestamp,
+            user: {
+              userIds: [
+                { idType: 'SHA256_EMAIL', idValue: '584c4423c421df49955759498a71495aba49b8780eb9387dff333b6f0982c777' }
+              ]
+            }
+          }
+        ]
+      })
+      .reply(201)
+
+    const response = await testDestination.executeBatch('streamConversion', {
+      events: [event, staleEvent],
+      settings,
+      mapping: batchMapping
+    })
+
+    // valid event succeeds
+    expect(response[0]).toMatchObject({ status: 201 })
+    // stale event fails with PAYLOAD_VALIDATION_FAILED
+    expect(response[1]).toMatchObject({
+      status: 400,
+      errortype: 'PAYLOAD_VALIDATION_FAILED',
+      errormessage: 'Timestamp should be within the past 90 days.',
+      errorreporter: 'INTEGRATIONS'
+    })
+  })
+
+  it('should not make an API call when all events in the batch are invalid', async () => {
+    const staleEvent2 = createTestEvent({
+      messageId: 'stale-event-id-2',
+      event: 'Funding Application Submitted',
+      type: 'track',
+      timestamp: staleTimestamp,
+      context: { traits: { email: '[email protected]' } },
+      properties: {}
+    })
+
+    const scope = nock(`${BASE_URL}/conversionEvents`).post('').reply(201)
+
+    const response = await testDestination.executeBatch('streamConversion', {
+      events: [staleEvent, staleEvent2],
+      settings,
+      mapping: batchMapping
+    })
+
+    expect(scope.isDone()).toBe(false)
+    expect(response[0]).toMatchObject({ status: 400, errortype: 'PAYLOAD_VALIDATION_FAILED' })
+    expect(response[1]).toMatchObject({ status: 400, errortype: 'PAYLOAD_VALIDATION_FAILED' })
+  })
+
+  it('should mark event as PAYLOAD_VALIDATION_FAILED when user ID is missing', async () => {
+    const noUserIdEvent = createTestEvent({
+      messageId: 'no-user-id-event',
+      event: 'Funding Application Submitted',
+      type: 'track',
+      timestamp: currentTimestamp.toString(),
+      context: { traits: {} },
+      properties: {}
+    })
+
+    nock(`${BASE_URL}/conversionEvents`)
+      .post('', {
+        elements: [
+          {
+            conversion: `urn:lla:llaPartnerConversion:${payload.conversionId}`,
+            conversionHappenedAt: currentTimestamp,
+            user: {
+              userIds: [
+                { idType: 'SHA256_EMAIL', idValue: '584c4423c421df49955759498a71495aba49b8780eb9387dff333b6f0982c777' }
+              ]
+            }
+          }
+        ]
+      })
+      .reply(201)
+
+    const response = await testDestination.executeBatch('streamConversion', {
+      events: [event, noUserIdEvent],
+      settings,
+      mapping: batchMapping
+    })
+
+    expect(response[0]).toMatchObject({ status: 201 })
+    expect(response[1]).toMatchObject({
+      status: 400,
+      errortype: 'PAYLOAD_VALIDATION_FAILED',
+      errormessage: 'One of email or LinkedIn UUID or Axciom ID or Oracle ID is required.',
+      errorreporter: 'INTEGRATIONS'
+    })
+  })
+
+  it('should correctly map responses to original batch indices', async () => {
+    nock(`${BASE_URL}/conversionEvents`).post('').reply(201)
+
+    const response = await testDestination.executeBatch('streamConversion', {
+      events: [staleEvent, event, staleEvent],
+      settings,
+      mapping: batchMapping
+    })
+
+    expect(response[0]).toMatchObject({ status: 400, errortype: 'PAYLOAD_VALIDATION_FAILED' })
+    expect(response[1]).toMatchObject({ status: 201 })
+    expect(response[2]).toMatchObject({ status: 400, errortype: 'PAYLOAD_VALIDATION_FAILED' })
+  })
+
+  it('should throw InvalidAuthenticationError when LinkedIn returns 401', async () => {
+    nock(`${BASE_URL}/conversionEvents`).post('').reply(401, { message: 'Unauthorized' })
+
+    await expect(
+      testDestination.executeBatch('streamConversion', {
+        events: [event],
+        settings,
+        mapping: batchMapping
+      })
+    ).rejects.toThrow('LinkedIn OAuth token is expired or invalid. Please re-authenticate.')
+  })
+
+  it('should mark all valid events as failed when LinkedIn returns 429', async () => {
+    nock(`${BASE_URL}/conversionEvents`).post('').reply(429, { message: 'Too Many Requests' })
+
+    const response = await testDestination.executeBatch('streamConversion', {
+      events: [event, secondEvent],
+      settings,
+      mapping: batchMapping
+    })
+
+    expect(response[0]).toMatchObject({ status: 429 })
+    expect(response[1]).toMatchObject({ status: 429 })
+  })
+
+  it('should mark all valid events as failed when LinkedIn returns 500', async () => {
+    nock(`${BASE_URL}/conversionEvents`).post('').reply(500, { message: 'Internal Server Error' })
+
+    const response = await testDestination.executeBatch('streamConversion', {
+      events: [event, secondEvent],
+      settings,
+      mapping: batchMapping
+    })
+
+    expect(response[0]).toMatchObject({ status: 500 })
+    expect(response[1]).toMatchObject({ status: 500 })
+  })
+})

packages/destination-actions/src/destinations/linkedin-conversions/streamConversion/index.ts

@@ -355,12 +355,7 @@ const action: ActionDefinition<Settings, Payload, undefined, OnMappingSaveInputs
     }
 
     linkedinApiClient.setConversionRuleId(conversionRuleId)
-
-    try {
-      return linkedinApiClient.batchConversionAdd(payloads)
-    } catch (error) {
-      throw handleRequestError(error)
-    }
+    return linkedinApiClient.batchConversionAdd(payloads)
   }
 }