Parse Customer.io Track API batch multi-status responses

packages/destination-actions/src/destinations/customerio/__tests__/utils.test.ts

@@ -1,4 +1,5 @@
-import { resolveIdentifiers, isIsoDate } from '../utils'
+import { MultiStatusResponse } from '@segment/actions-core'
+import { isIsoDate, parseTrackApiErrors, parseTrackApiMultiStatusResponse, resolveIdentifiers, sendBatch } from '../utils'
 
 describe('isIsoDate', () => {
   it('should return true for valid ISO date with fractional seconds from 1-9 digits', () => {
@@ -79,3 +80,129 @@ describe('resolveIdentifiers', () => {
     expect(resolveIdentifiers({})).toBeUndefined()
   })
 })
+
+describe('sendBatch', () => {
+  it('should parse 207 multi-status Track API responses', async () => {
+    const request = jest.fn().mockResolvedValue({
+      status: 207,
+      data: {
+        errors: [
+          {
+            batch_index: 1,
+            reason: 'invalid',
+            message: 'Attribute value too long'
+          }
+        ]
+      }
+    })
+
+    const response = await sendBatch(request, [
+      {
+        type: 'person',
+        action: 'event',
+        settings: {},
+        payload: { person_id: 'user-1', name: 'First' }
+      },
+      {
+        type: 'person',
+        action: 'event',
+        settings: {},
+        payload: { person_id: 'user-2', name: 'Second' }
+      }
+    ])
+
+    expect(response).toBeInstanceOf(MultiStatusResponse)
+    expect((response as MultiStatusResponse).length()).toBe(2)
+    expect((response as MultiStatusResponse).getResponseAtIndex(0).value()).toEqual({
+      status: 200,
+      body: {},
+      sent: {}
+    })
+    expect((response as MultiStatusResponse).getResponseAtIndex(1).value()).toEqual({
+      status: 400,
+      errormessage: 'Attribute value too long',
+      errortype: 'PAYLOAD_VALIDATION_FAILED',
+      body: {
+        batch_index: 1,
+        reason: 'invalid',
+        message: 'Attribute value too long'
+      }
+    })
+  })
+
+  it('should parse 200 Track API responses that still contain batch errors', async () => {
+    const request = jest.fn().mockResolvedValue({
+      status: 200,
+      data: {
+        errors: [
+          {
+            batch_index: 0,
+            reason: 'required',
+            field: 'name',
+            message: 'Name is required'
+          }
+        ]
+      }
+    })
+
+    const response = await sendBatch(request, [
+      {
+        type: 'person',
+        action: 'event',
+        settings: {},
+        payload: { person_id: 'user-1', name: 'First' }
+      }
+    ])
+
+    expect(response).toBeInstanceOf(MultiStatusResponse)
+    expect((response as MultiStatusResponse).getResponseAtIndex(0).value()).toEqual({
+      status: 400,
+      errormessage: 'Name is required',
+      errortype: 'PAYLOAD_VALIDATION_FAILED',
+      body: {
+        batch_index: 0,
+        reason: 'required',
+        field: 'name',
+        message: 'Name is required'
+      }
+    })
+  })
+})
+
+describe('parseTrackApiErrors', () => {
+  it('should fill success entries for items without errors', () => {
+    const response = parseTrackApiErrors(
+      [
+        {
+          batch_index: 1,
+          reason: 'required',
+          field: 'name',
+          message: 'Name is required'
+        }
+      ],
+      3
+    )
+
+    expect(response.getAllResponses().map((result) => result.value())).toEqual([
+      { status: 200, body: {}, sent: {} },
+      {
+        status: 400,
+        errormessage: 'Name is required',
+        errortype: 'PAYLOAD_VALIDATION_FAILED',
+        body: {
+          batch_index: 1,
+          reason: 'required',
+          field: 'name',
+          message: 'Name is required'
+        }
+      },
+      { status: 200, body: {}, sent: {} }
+    ])
+  })
+})
+
+describe('parseTrackApiMultiStatusResponse', () => {
+  it('should return null for non-Track API response bodies', () => {
+    expect(parseTrackApiMultiStatusResponse({ ok: true }, 1)).toBeNull()
+  })
+})

packages/destination-actions/src/destinations/customerio/utils.ts

@@ -1,6 +1,7 @@
 import dayjs from '../../lib/dayjs'
 import isPlainObject from 'lodash/isPlainObject'
 import { fullFormats } from 'ajv-formats/dist/formats'
+import { ErrorCodes, MultiStatusResponse, type RequestClient } from '@segment/actions-core'
 import { CUSTOMERIO_TRACK_API_VERSION } from './versioning-info'
 
 const isEmail = (value: string): boolean => {
@@ -196,23 +197,144 @@ export const resolveIdentifiers = ({
   }
 }
 
-export const sendBatch = <Payload extends BasePayload>(request: Function, options: RequestPayload<Payload>[]) => {
+export const sendBatch = async <Payload extends BasePayload>(
+  request: RequestClient,
+  options: RequestPayload<Payload>[]
+) => {
   if (!options?.length) {
     return
   }
 
   const [{ settings }] = options
   const batch = options.map((opts) => buildPayload(opts))
 
-  return request(`${trackApiEndpoint(settings)}/api/${CUSTOMERIO_TRACK_API_VERSION}/batch`, {
+  const response = await request(`${trackApiEndpoint(settings)}/api/${CUSTOMERIO_TRACK_API_VERSION}/batch`, {
     method: 'post',
     json: {
       batch
     }
   })
+
+  const responseBody = getResponseBody(response)
+
+  if (response?.status === 207 && responseBody) {
+    const parsedResults = parseTrackApiMultiStatusResponse(responseBody, batch.length)
+    if (parsedResults) {
+      return parsedResults
+    }
+  }
+
+  if (response?.status === 200 && responseBody) {
+    const parsedResults = parseTrackApiMultiStatusResponse(responseBody, batch.length)
+    if (parsedResults) {
+      return parsedResults
+    }
+  }
+
+  return response
+}
+
+interface TrackApiError {
+  batch_index?: number
+  reason?: string
+  field?: string
+  message?: string
+}
+
+interface TrackApiResponse {
+  errors?: TrackApiError[]
+}
+
+interface RequestResponse {
+  status?: number
+  data?: unknown
+  content?: unknown
+  body?: unknown
+}
+
+function mapTrackApiReasonToErrorCode(reason: string | undefined) {
+  if (!reason) {
+    return undefined
+  }
+
+  switch (reason.toLowerCase()) {
+    case 'invalid':
+    case 'required':
+      return ErrorCodes.PAYLOAD_VALIDATION_FAILED
+    default:
+      return undefined
+  }
+}
+
+function getResponseBody(response: RequestResponse): unknown {
+  const body = response.data ?? response.content ?? response.body
+
+  if (typeof body !== 'string') {
+    return body
+  }
+
+  try {
+    return JSON.parse(body)
+  } catch {
+    try {
+      const decoded = Buffer.from(body, 'base64').toString('utf-8')
+      return JSON.parse(decoded)
+    } catch {
+      return body
+    }
+  }
+}
+
+export function parseTrackApiErrors(errors: TrackApiError[], totalItems: number): MultiStatusResponse {
+  const multiStatusResponse = new MultiStatusResponse()
+  const errorMap = new Map<number, TrackApiError>()
+
+  for (const error of errors) {
+    if (typeof error.batch_index === 'number') {
+      errorMap.set(error.batch_index, error)
+    }
+  }
+
+  for (let i = 0; i < totalItems; i++) {
+    const error = errorMap.get(i)
+
+    if (!error) {
+      multiStatusResponse.setSuccessResponseAtIndex(i, {
+        status: 200,
+        body: {},
+        sent: {}
+      })
+      continue
+    }
+
+    multiStatusResponse.setErrorResponseAtIndex(i, {
+      status: 400,
+      errormessage: error.message || `${error.reason || 'ERROR'}: ${error.field || 'unknown field'}`,
+      errortype: mapTrackApiReasonToErrorCode(error.reason),
+      body: error
+    })
+  }
+
+  return multiStatusResponse
+}
+
+export function parseTrackApiMultiStatusResponse(
+  responseBody: unknown,
+  totalItems: number
+): MultiStatusResponse | null {
+  if (!isRecord(responseBody)) {
+    return null
+  }
+
+  const { errors } = responseBody as TrackApiResponse
+  if (!Array.isArray(errors) || errors.length === 0) {
+    return null
+  }
+
+  return parseTrackApiErrors(errors, totalItems)
 }
 
-export const sendSingle = <Payload extends BasePayload>(request: Function, options: RequestPayload<Payload>) => {
+export const sendSingle = <Payload extends BasePayload>(request: RequestClient, options: RequestPayload<Payload>) => {
   const json = buildPayload(options)
   return request(`${trackApiEndpoint(options.settings)}/api/${CUSTOMERIO_TRACK_API_VERSION}/entity`, {
     method: 'post',