Parse Customer.io Track API batch multi-status responses

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

@@ -1,4 +1,11 @@
-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 +86,140 @@ 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: { person_id: 'user-1', name: 'First' },
+      sent: { type: 'person', action: 'event', identifiers: { id: 'user-1' }, name: 'First' }
+    })
+    expect((response as MultiStatusResponse).getResponseAtIndex(1).value()).toEqual({
+      status: 400,
+      errormessage: 'Attribute value too long',
+      errortype: 'PAYLOAD_VALIDATION_FAILED',
+      body: { person_id: 'user-2', name: 'Second' },
+      sent: { type: 'person', action: 'event', identifiers: { id: 'user-2' }, name: 'Second' }
+    })
+  })
+
+  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: { person_id: 'user-1', name: 'First' },
+      sent: { type: 'person', action: 'event', identifiers: { id: 'user-1' }, name: 'First' }
+    })
+  })
+})
+
+describe('parseTrackApiErrors', () => {
+  it('should fill success entries for items without errors', () => {
+    const options = [
+      { type: 'person', action: 'event', settings: {}, payload: { person_id: 'user-0' } },
+      { type: 'person', action: 'event', settings: {}, payload: { person_id: 'user-1' } },
+      { type: 'person', action: 'event', settings: {}, payload: { person_id: 'user-2' } }
+    ]
+    const batch = [
+      { type: 'person', action: 'event', identifiers: { id: 'user-0' } },
+      { type: 'person', action: 'event', identifiers: { id: 'user-1' } },
+      { type: 'person', action: 'event', identifiers: { id: 'user-2' } }
+    ]
+
+    const response = parseTrackApiErrors(
+      [
+        {
+          batch_index: 1,
+          reason: 'required',
+          field: 'name',
+          message: 'Name is required'
+        }
+      ],
+      options,
+      batch
+    )
+
+    expect(response.getAllResponses().map((result) => result.value())).toEqual([
+      {
+        status: 200,
+        body: { person_id: 'user-0' },
+        sent: { type: 'person', action: 'event', identifiers: { id: 'user-0' } }
+      },
+      {
+        status: 400,
+        errormessage: 'Name is required',
+        errortype: 'PAYLOAD_VALIDATION_FAILED',
+        body: { person_id: 'user-1' },
+        sent: { type: 'person', action: 'event', identifiers: { id: 'user-1' } }
+      },
+      {
+        status: 200,
+        body: { person_id: 'user-2' },
+        sent: { type: 'person', action: 'event', identifiers: { id: 'user-2' } }
+      }
+    ])
+  })
+})
+
+describe('parseTrackApiMultiStatusResponse', () => {
+  it('should return null for non-Track API response bodies', () => {
+    const options = [{ type: 'person', action: 'event', settings: {}, payload: { person_id: 'user-0' } }]
+    const batch = [{ type: 'person', action: 'event', identifiers: { id: 'user-0' } }]
+    expect(parseTrackApiMultiStatusResponse({ ok: true }, options, batch)).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, RequestClient } from '@segment/actions-core'
 import { CUSTOMERIO_TRACK_API_VERSION } from './versioning-info'
 
 const isEmail = (value: string): boolean => {
@@ -196,23 +197,163 @@ 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`, {
-    method: 'post',
-    json: {
-      batch
+  try {
+    const response = await request<CustomerIOBatchResponse>(
+      `${trackApiEndpoint(settings)}/api/${CUSTOMERIO_TRACK_API_VERSION}/batch`,
+      {
+        method: 'post',
+        json: {
+          batch
+        }
+      }
+    )
+
+    const responseBody = getResponseBody(response)
+
+    if ((response?.status === 200 || response?.status === 207) && responseBody) {
+      const parsedResults = parseTrackApiMultiStatusResponse(responseBody, options, batch)
+      if (parsedResults) {
+        return parsedResults
+      }
     }
-  })
+
+    return response
+  } catch (err) {
+    const error = err as { message?: string; response?: { status?: number } }
+    const status = error.response?.status ?? 500
+    const message = error.message ?? 'Unknown error'
+
+    const multiStatusResponse = new MultiStatusResponse()
+    for (let i = 0; i < options.length; i++) {
+      multiStatusResponse.setErrorResponseAtIndex(i, {
+        status,
+        errormessage: message,
+        body: options[i].payload,
+        sent: batch[i]
+      })
+    }
+    return multiStatusResponse
+  }
+}
+
+interface TrackApiError {
+  batch_index?: number
+  reason?: string
+  field?: string
+  message?: string
+}
+
+interface CustomerIOBatchResponse {
+  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): CustomerIOBatchResponse | string | undefined {
+  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<Payload extends BasePayload>(
+  errors: TrackApiError[],
+  options: RequestPayload<Payload>[],
+  batch: Record<string, unknown>[]
+): 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 < options.length; i++) {
+    const error = errorMap.get(i)
+
+    if (!error) {
+      multiStatusResponse.setSuccessResponseAtIndex(i, {
+        status: 200,
+        body: options[i].payload,
+        sent: batch[i]
+      })
+      continue
+    }
+
+    multiStatusResponse.setErrorResponseAtIndex(i, {
+      status: 400,
+      errormessage: error.message || `${error.reason || 'ERROR'}: ${error.field || 'unknown field'}`,
+      errortype: mapTrackApiReasonToErrorCode(error.reason),
+      body: options[i].payload,
+      sent: batch[i]
+    })
+  }
+
+  return multiStatusResponse
+}
+
+export function parseTrackApiMultiStatusResponse<Payload extends BasePayload>(
+  responseBody: CustomerIOBatchResponse | string | undefined,
+  options: RequestPayload<Payload>[],
+  batch: Record<string, unknown>[]
+): MultiStatusResponse | null {
+  if (!isRecord(responseBody)) {
+    return null
+  }
+
+  const { errors } = responseBody as CustomerIOBatchResponse
+  if (!Array.isArray(errors) || errors.length === 0) {
+    return null
+  }
+
+  return parseTrackApiErrors(errors, options, batch)
 }
 
-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',