Parse Customer.io Track API batch multi-status responses#3657
Parse Customer.io Track API batch multi-status responses#3657jcpsimmons wants to merge 11 commits intosegmentio:mainfrom
Conversation
|
This draft is 1 of 3 upstream Customer.io PRs for CDP-4750. Scope here: only Track API batch multi-status parsing in Not in this PR:
Companion drafts:
This supersedes the Track API parsing portion of the older combined draft #3654. |
|
Hi @joe-ayoub-segment could I please have review on this and the other two PRs 🙏 |
joe-ayoub-segment
left a comment
joe-ayoub-segment
left a comment
There was a problem hiding this comment.
Hi @jcpsimmons,
Thanks for raising the PR for this change. It will really help customers with observability.
I left some comments for you to look at.
I didn't review all of the PR as i found it difficult to follow. Adding the types (see comments) will help me better understand the logic.
One final thing - please add proof of testing to the PR description.
- show batch payloads being delivered
- show error responses being returned correctly
Let me know if you have any questions or would like assistance.
best regards,
Joe
| 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 | ||
| } | ||
| } |
There was a problem hiding this comment.
nit: Can we reduce the redundant code here?
There was a problem hiding this comment.
Done, collapsed the two blocks into a single OR condition.
| 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`, { |
There was a problem hiding this comment.
Can we add an expected response type to this?
For example:
const response = await request<CustomerIOBatchResponse>(`${trackApiEndpoint(settings)}/api/${CUSTOMERIO_TRACK_API_VERSION}/batch`, {
method: 'POST',
json { batch }
})
There was a problem hiding this comment.
Also, consider wrapping this request in a try catch so that you can properly populate errors in the multistatusResponse when the entire payload fails.
try {
const response = await request(${trackApiEndpoint(settings)}/api/${CUSTOMERIO_TRACK_API_VERSION}/batch, {
method: 'POST',
json { batch }
})
}
catch(err) {
const error as CustomerIOErrorResponse
// for each payload in the batch, do setErrorResponseAtIndex
}
There was a problem hiding this comment.
Added. Renamed TrackApiResponse to CustomerIOBatchResponse and passed it as the generic: request<CustomerIOBatchResponse>(...). Also wrapped the whole thing in a try/catch per your other comment so failed requests populate every index with the error details.
There was a problem hiding this comment.
Added. The catch block extracts the status and message from the error, then iterates over every item in the batch and calls setErrorResponseAtIndex with the original payload and sent JSON. If the whole request blows up, every item in the multi-status response reflects the failure.
| } | ||
| }) | ||
|
|
||
| const responseBody = getResponseBody(response) |
There was a problem hiding this comment.
Can we add a type to this too please?
There was a problem hiding this comment.
Covered by the same rename. getResponseBody now returns CustomerIOBatchResponse | string | undefined and that flows through to parseTrackApiMultiStatusResponse.
| const error = errorMap.get(i) | ||
|
|
||
| if (!error) { | ||
| multiStatusResponse.setSuccessResponseAtIndex(i, { |
There was a problem hiding this comment.
The contents of each multiStatusResponse item is passed to the UI for observability purposes.
So if possible we should populate body and sent values correctly:
msResponse.setSuccessResponseAtIndex(index, {
status: 200,
body: // this should be the original payload object sent to the perform function,
sent: // this should be the actual JSON which was posted to your platform
})
There was a problem hiding this comment.
Good call. Threaded the original options array and the built batch array through to parseTrackApiErrors. Success responses now get body: options[i].payload (original payload from perform) and sent: batch[i] (the transformed JSON we posted).
| continue | ||
| } | ||
|
|
||
| multiStatusResponse.setErrorResponseAtIndex(i, { |
There was a problem hiding this comment.
Similarly to successful responses, we should try and populate error responses with correct details.
msResponse.setErrorResponseAtIndex(index, {
status: 400,
errortype,
errormessage,
body: // this should be the original payload object sent to the perform function,
sent: // this should be the actual JSON which was posted to your platform
})
There was a problem hiding this comment.
Same fix, error responses now also get body: options[i].payload and sent: batch[i].
| } | ||
| } | ||
|
|
||
| function getResponseBody(response: RequestResponse): unknown { |
There was a problem hiding this comment.
Please define a proper response type for this function.
There was a problem hiding this comment.
Done. Returns CustomerIOBatchResponse | string | undefined now.
… try/catch - Collapse duplicate 200/207 status blocks into single OR condition - Rename TrackApiResponse to CustomerIOBatchResponse and add as request generic - Add return type to getResponseBody - Thread original payloads and built batch through to populate body/sent in multi-status responses - Wrap batch request in try/catch to handle full payload failures
jcpsimmons
force-pushed
the
cdp-4750-cio-track-api-multistatus
branch
from
2244302 to
8be62ad
Compare
|
HI @joe-ayoub-segment thank you for that feedback - I've addressed could I please have a re-review? |
|
@joe-ayoub-segment friendly nudge - I've addressed all your feedback from the last review. Could you take another look? Happy to hop on a call if anything needs discussion. Thanks! |
There was a problem hiding this comment.
Pull request overview
This PR enhances the Customer.io destination’s batch sender to interpret Customer.io Track API batch responses (including HTTP 207 Multi-Status) into per-item MultiStatusResponse entries, so partial failures can be surfaced correctly by the Actions framework.
Changes:
- Updates
sendBatchto parse Track API batch responses and return aMultiStatusResponsewhen per-item errors are present. - Adds parsing utilities (
parseTrackApiErrors,parseTrackApiMultiStatusResponse) plus response-body extraction logic to support multiple response shapes. - Adds unit tests validating
207and200-with-errorsbatch behaviors and parser edge cases.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
packages/destination-actions/src/destinations/customerio/utils.ts |
Implements Track API batch multi-status parsing and updates sendBatch/types to use RequestClient. |
packages/destination-actions/src/destinations/customerio/__tests__/utils.test.ts |
Adds unit tests covering multi-status parsing and sendBatch behavior. |
| 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) | ||
| } | ||
| } |
There was a problem hiding this comment.
parseTrackApiErrors stores errors in a Map<number, TrackApiError>, so if the Track API returns multiple error entries for the same batch_index, earlier errors will be overwritten and lost. Consider grouping errors per index (e.g., Map<number, TrackApiError[]>) and combining messages/fields so the per-item response preserves all reported issues.
There was a problem hiding this comment.
@copilot apply changes based on this feedback
|
Hi @jcpsimmons, I was on PTO the last 2 weeks so just looking at this now. Thanks for applying the changes. Copilot is now also set up to review PRs. Can you respond to the comments it left please? Also I can see that CI is failing. Can you take a look please? Finally, would it be possible to include some proof of testing. This can be in the form of screenshots ro videos. The main things I'd be looking for are:
It looks like there are 2 types of responses which can come back from Customer.io. One with a string value which needds to be parsed into JSON, and another which is already JSON. Could you also demonstrate that you have tested these two different scenarios please? |
|
Hi @jcpsimmons - I took another look at this. I think it would be worth making a few changes. 1 Current code catches everything: This catches everything — HTTPError, RetryableError, TypeError, you name it — and silently converts it all into a MultiStatusResponse. The framework never sees a thrown error, so it never retries. How it should be done (following the Braze pattern at braze/utils.ts:625-658):
The getResponseBody function (in the diff) tries response.data, response.content, and response.body, then attempts JSON parse, then even base64 decode. But RequestClient (from @segment/actions-core) always returns a typed response where response.data is already parsed JSON when the response has a JSON content-type. The content/body fallbacks and the base64 decoding path are dead code — they'll never be hit with a RequestClient. This is speculative defensive coding that adds confusion. It should just use response.data.
If the API returns { errors: [] } (200 with an empty errors array), parseTrackApiMultiStatusResponse returns null, which means sendBatch falls through to returning the raw response. That's fine in practice (no errors = success), but it means a 207 with { errors: [] } also falls through to returning the raw response instead of a MultiStatusResponse with all successes. This is an inconsistence - a 207 with a valid errors array should probably always return a MultiStatusResponse, even if every item succeeded.
When the reason string from Customer.io doesn't match 'invalid' or 'required', errortype is set to undefined. This means the framework has no error code to work with. It would be safer to default to a generic error code like ErrorCodes.UNKNOWN_ERROR or ErrorCodes.INTEGRATION_ERROR rather than leaving it blank.
sendBatch can now return undefined (empty options), a MultiStatusResponse, or the raw Response object (when no errors are parsed). Can you just return MultiStatusResponse when a batch has been sent (via performBatch), and a raw response object when a single event is sent (via perform)? Everything else should be a thrown error. |
|
@sydneycollins-cio handing this PR to you along with CDP-4750 if you're able to take it over. Status from my side: latest review still has substantive follow-up. Joe requested proof-of-testing/screenshots plus the code-review follow-through, and the current batch-response behavior likely still needs one more correctness pass before this is ready. |
…lid responses, improve error message extraction, add 500 retry test
| multiStatusResponse.setErrorResponseAtIndex(i, { | ||
| status, | ||
| errortype: ErrorCodes.INTEGRATION_ERROR, | ||
| errormessage: message, | ||
| body: options[i].payload, | ||
| sent: batch[i] | ||
| }) |
There was a problem hiding this comment.
In the non-retryable HTTPError path, each per-item error response is forced to errortype: ErrorCodes.INTEGRATION_ERROR regardless of the HTTP status. This will misclassify authentication/authorization failures (401/403) and other status-derived categories, since the framework only auto-inferrs errortype when it’s omitted. Consider omitting errortype here (so it can be inferred) or mapping it based on the status code (e.g., invalid auth vs validation vs unknown).
| } catch (err) { | ||
| // Retryable HTTP errors (408 Request Timeout, 429 Too Many Requests, 5xx Server Errors) | ||
| // and unexpected non-HTTP errors should be rethrown so the framework's retry wrapper | ||
| // can handle them. Only convert to per-item errors for non-retryable HTTP failures. | ||
| if (err instanceof HTTPError) { | ||
| const status = err.response.status | ||
| if (status === 408 || status === 429 || status >= 500) { | ||
| throw err | ||
| } | ||
|
|
||
| const responseBody = err.response?.data as { message?: string } | undefined | ||
| const message = responseBody?.message ?? err.message ?? 'Unknown error' | ||
| const multiStatusResponse = new MultiStatusResponse() | ||
| for (let i = 0; i < options.length; i++) { | ||
| multiStatusResponse.setErrorResponseAtIndex(i, { | ||
| status, | ||
| errortype: ErrorCodes.INTEGRATION_ERROR, | ||
| errormessage: message, | ||
| body: options[i].payload, | ||
| sent: batch[i] | ||
| }) | ||
| } | ||
| return multiStatusResponse | ||
| } |
There was a problem hiding this comment.
sendBatch adds new behavior to convert non-retryable HTTPError responses into a per-item MultiStatusResponse, but the test suite only covers retryable HTTP errors (429/5xx) and unexpected response shapes. Add a unit test that simulates a non-retryable HTTPError (e.g., 400/401) and asserts the returned MultiStatusResponse contains per-item errors with the expected status/message (and desired error typing).
Testing EvidenceBatch of 3 events sent to Note on Track API response format: The Track API only returns entries for failed items in the Shared request payload: {
"batch": [
{
"type": "person", "action": "event",
"identifiers": { "id": "usr_abc123" },
"name": "Product Purchased",
"timestamp": "2026-04-21T18:00:00.000Z",
"data": { "product_id": "prod_999", "price": 49.99, "currency": "USD", "quantity": 2 }
},
{
"type": "person", "action": "event",
"identifiers": { "id": "usr_def456" },
"name": "Page Viewed",
"timestamp": "2026-04-21T18:00:01.000Z",
"data": { "page": "/pricing", "referrer": "https://google.com", "notes": "xxx... [1001 chars]" }
},
{
"type": "person", "action": "event",
"identifiers": { "id": "usr_ghi789" },
"name": "Email Opened",
"timestamp": "2026-04-21T18:00:02.000Z",
"data": { "campaign_id": "camp_123", "subject": "Your weekly digest" }
}
]
}Case 1 — HTTP 207 Multi-Status: partial failure
Case 2 — HTTP 200 with errors array: partial failure Track API returns 200 but includes per-item errors for items 0 and 2. Case 3 — HTTP 200 empty errors: all success Case 4 — HTTP 429: retryable Error rethrown — framework retries entire batch, no per-item conversion. Case 5 — HTTP 500: retryable Case 6 — HTTP 400: non-retryable flat error Flat error propagated to all items as Unit tests: 23/23 passing ( |
…vert UNKNOWN_ERROR default
| errors?: TrackApiError[] | ||
| } | ||
|
|
||
| function mapHttpStatusToErrorCode(status: number): string { |
There was a problem hiding this comment.
mapHttpStatusToErrorCode is typed to return string, but getErrorCodeFromHttpStatus returns keyof typeof ErrorCodes and ActionDestinationErrorResponseType.errortype expects that key type. As written, this can cause a TypeScript type error (or weaken typing) when setting errortype on the MultiStatus entries. Update the return type (or inline getErrorCodeFromHttpStatus) so errortype remains keyof typeof ErrorCodes.
| function mapHttpStatusToErrorCode(status: number): string { | |
| function mapHttpStatusToErrorCode(status: number): keyof typeof ErrorCodes { |
|
|
||
| for (const error of errors) { | ||
| if (typeof error.batch_index === 'number') { | ||
| const existing = errorMap.get(error.batch_index) | ||
| if (existing) { | ||
| existing.push(error) | ||
| } else { | ||
| errorMap.set(error.batch_index, [error]) | ||
| } | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
parseTrackApiErrors currently drops any error objects that don't include a numeric batch_index. If Customer.io returns an error entry without batch_index (or with an out-of-range index), the code will silently treat all items as success, which can hide real failures. Consider detecting unindexable errors and surfacing them (e.g., throw an IntegrationError, or apply a per-item error for the whole batch) rather than ignoring them.
| for (const error of errors) { | |
| if (typeof error.batch_index === 'number') { | |
| const existing = errorMap.get(error.batch_index) | |
| if (existing) { | |
| existing.push(error) | |
| } else { | |
| errorMap.set(error.batch_index, [error]) | |
| } | |
| } | |
| } | |
| const unindexableErrors: TrackApiError[] = [] | |
| for (const error of errors) { | |
| const batchIndex = error.batch_index | |
| if (!Number.isInteger(batchIndex) || batchIndex < 0 || batchIndex >= options.length) { | |
| unindexableErrors.push(error) | |
| continue | |
| } | |
| const existing = errorMap.get(batchIndex) | |
| if (existing) { | |
| existing.push(error) | |
| } else { | |
| errorMap.set(batchIndex, [error]) | |
| } | |
| } | |
| if (unindexableErrors.length > 0) { | |
| const errormessage = unindexableErrors | |
| .map((error) => { | |
| const indexDescription = | |
| typeof error.batch_index === 'number' ? `batch_index ${error.batch_index}` : 'missing batch_index' | |
| return error.message || `${error.reason || 'ERROR'}: ${error.field || 'unknown field'} (${indexDescription})` | |
| }) | |
| .join('; ') | |
| throw new IntegrationError(`Customer.io returned batch errors that could not be mapped to request items: ${errormessage}`) | |
| } |
| throw new IntegrationError( | ||
| 'Customer.io Track API batch response did not include an errors array', | ||
| 'INVALID_RESPONSE', | ||
| 400 |
There was a problem hiding this comment.
This IntegrationError is raised for an unexpected/invalid response shape from Customer.io, but it uses HTTP status 400 (client error). That can be misleading since it isn't caused by the input payload. Consider using a 5xx (commonly 500/502) status so the failure is classified as a dependency/partner response issue.
| 400 | |
| 502 |
…, fix return type
|
@joe-ayoub-segment ready for review when you have a chance! |
Parses Customer.io Track API batch responses into per-item
MultiStatusResponseresults so Segment can surface partial failures correctly.Testing
Verified locally under Node
22.13.1:TZ=UTC yarn cloud test --testPathPattern=src/destinations/customerio/__tests__/utils.test.ts✅./bin/run serve --destination=customerio -nboots successfully and exposes the Customer.io action routes ✅End-to-end testing using the local server was completed ✅
Added unit tests for new functionality
Tested end-to-end using the local server
[If destination is already live] Tested for backward compatibility of destination. Note: New required fields are a breaking change.
[Segmenters] Tested in the staging environment
[If applicable for this change] Tested for regression with Hadron.
Security Review
No field definitions or credential handling were changed in this PR.
passwordNew Destination Checklist
Not applicable.
verioning-info.tsfile. example