feat(cdp): segment destination mapping #31336

MarconLP · 2025-04-17T09:13:16Z

Important

👉 Stay up-to-date with PostHog coding conventions for a smoother review.

Changes

PostHog / Segment view

PostHog

Segment

attempt to map a segment destination to posthog
- most interesting is probably the template.ts: https://github.com/PostHog/posthog/blob/external-destinations/plugin-server/src/cdp/legacy-plugins/_segmentDestinations/amplitude/template.ts

Todos:

translate FQL to posthog filters
~~ability to select an entire object (we can technically do this using a {event.properties} string field)~~ convert to string and use {event.properties} for now
execute different actions depending on the mapping

TODOs:

Refer to this PR for the followup status #32651

Follow-up:

default values defined here are broken
debug log option for all (segment?) destinations
select entire object / individual keys switcher
make hog_function_template endpoint paginated
do not use inline snapshots for bigger results
implement retry to cyclotron

Does this work well for both Cloud and self-hosted?

How did you test this code?

greptile-apps

PR Summary

Here's my review of the changes for the Segment destination mapping functionality:

Added comprehensive Amplitude destination integration with extensive test coverage and utility functions for PostHog's CDP.

Key changes:

Implemented core Amplitude actions (logEvent, logPurchase, identifyUser, groupIdentifyUser) with proper type definitions and field mappings
Added robust user property handling with support for UTM tracking, referrer data, and user agent parsing
Created regional endpoint support for both North America and Europe with configurable batch/single event processing
Built timestamp conversion utilities and session ID formatting for proper event timing
Introduced comprehensive test coverage across all major components with edge case handling

Note: The template.ts file needs significant revision as it contains deprecated code, hardcoded values and inadequate error handling.

_{34 file(s) reviewed, 45 comment(s)}
_{Edit PR Review Bot Settings | Greptile}

...er/src/cdp/legacy-plugins/_segmentDestinations/amplitude/__tests__/convert-timestamp.test.ts

...rc/cdp/legacy-plugins/_segmentDestinations/amplitude/__tests__/merge-user-properties.test.ts

plugin-server/src/cdp/legacy-plugins/_segmentDestinations/amplitude/__tests__/amplitude.test.ts

plugin-server/src/cdp/legacy-plugins/_segmentDestinations/amplitude/referrer.ts

plugin-server/src/cdp/legacy-plugins/_segmentDestinations/amplitude/properties.ts

plugin-server/src/cdp/legacy-plugins/_segmentDestinations/amplitude/regional-endpoints.ts

plugin-server/src/cdp/legacy-plugins/_segmentDestinations/amplitude/utm.ts

github-actions · 2025-04-17T09:25:32Z

Size Change: +10 B (0%)

Total Size: 3.71 MB

ℹ️ View Unchanged

Filename	Size	Change
`frontend/dist/toolbar.js`	3.71 MB	+10 B (0%)

_{compressed-size-action}

plugin-server/src/cdp/services/legacy-plugin-executor.service.ts

graphite-app · 2025-04-24T13:18:16Z

plugin-server/src/cdp/legacy-plugins/template.ts

+        default: field.type !== 'object' || typeof field.default !== 'undefined' && ('@path' in field.default) ? translateInputs(field.default) : Object.fromEntries(Object.entries(field.properties ?? {}).map(([key, _]) => {
+            const defaultVal = field.default as Record<string, object> ?? {}
+            return [key, translateInputs(defaultVal[key])]
+        })),


The default value logic has two potential issues:

The condition typeof field.default !== 'undefined' && ('@path' in field.default) may cause runtime errors if field.default is undefined, as the second part would attempt to check a property on an undefined value.

The type assertion field.default as Record<string, object> could be unsafe when field.default is undefined.

Consider using optional chaining and nullish coalescing to handle these cases more safely:

default: field.type !== 'object' || (field.default && '@path' in field.default) ? translateInputs(field.default) : Object.fromEntries(Object.entries(field.properties ?? {}).map(([key, _]) => { const defaultVal = field.default ? (field.default as Record<string, object>) : {}; return [key, translateInputs(defaultVal[key])]; })),

Suggested change

default: field.type !== 'object' || typeof field.default !== 'undefined' && ('@path' in field.default) ? translateInputs(field.default) : Object.fromEntries(Object.entries(field.properties ?? {}).map(([key, _]) => {

const defaultVal = field.default as Record<string, object> ?? {}

return [key, translateInputs(defaultVal[key])]

})),

default: field.type !== 'object' || (field.default && '@path' in field.default) ? translateInputs(field.default) : Object.fromEntries(Object.entries(field.properties ?? {}).map(([key, _]) => {

const defaultVal = field.default ? (field.default as Record<string, object>) : {}

return [key, translateInputs(defaultVal[key])]

})),

Spotted by Diamond

Is this helpful? React 👍 or 👎 to let us know.

posthog-bot · 2025-04-24T13:57:35Z

📸 UI snapshots have been updated

1 snapshot changes in total. 0 added, 1 modified, 0 deleted:

chromium: 0 added, 1 modified, 0 deleted (diff for shard 7)
webkit: 0 added, 0 modified, 0 deleted

Triggered by this commit.

👉 Review this PR's diff of snapshots.

posthog-bot · 2025-04-24T14:07:05Z

📸 UI snapshots have been updated

1 snapshot changes in total. 0 added, 1 modified, 0 deleted:

chromium: 0 added, 1 modified, 0 deleted (diff for shard 7)
webkit: 0 added, 0 modified, 0 deleted

Triggered by this commit.

👉 Review this PR's diff of snapshots.

rust/Cargo.lock

plugin-server/src/cdp/types.ts

plugin-server/src/cdp/services/segment-destination-executor.service.ts

meikelmosby · 2025-05-21T08:06:18Z

plugin-server/src/cdp/services/segment-destination-executor.service.ts

all of this code is untested. please add tests.

added tests

meikelmosby · 2025-05-21T08:09:10Z

plugin-server/src/cdp/consumers/cdp-cyclotron-segment-worker.consumer.ts

+
+    public async processInvocations(invocations: HogFunctionInvocation[]): Promise<HogFunctionInvocationResult[]> {
+        // Segment plugins fire fetch requests and so need to be run in true parallel
+        return await Promise.all(


Fail-fast or fail-safe?
If one plugin throws an error, Promise.all will reject immediately. Ask yourself:
Do you want all or nothing (fail-fast)? → Promise.all is fine.
Do you want to collect all results/errors? → Use Promise.allSettled.

@meikelmosby not fully sure on this one - technically, we want to capture all results & errors from all invocations, but wouldn't the same apply to the plugin worker?

going to keep as is for now - https://posthog.slack.com/archives/C06GG249PR6/p1747918574348349

Jumping in - the case here is basically that the executor code should only throw if there is a genuine programming error. If so we essentially have no choice but to crash out so there isn't much of a difference between all and allSettled, as we have no course of action to partially handle a batch.

Generally it would be good if we could partially handle a batch, but thats not how kafka works so this is an issue with out architecture moreso than this specific code path

meikelmosby · 2025-05-21T08:10:09Z

plugin-server/src/cdp/consumers/cdp-cyclotron-segment-worker.consumer.ts

adding some tests would be good here as well..

I think the tests for the executor cover this one as well

frontend/src/lib/api.ts

frontend/src/scenes/pipeline/hogfunctions/HogFunctionConfiguration.tsx

plugin-server/src/cdp/legacy-plugins/segmentTemplates.test.ts

plugin-server/src/cdp/legacy-plugins/segmentTemplates.ts

benjackwhite · 2025-05-22T13:06:10Z

plugin-server/src/cdp/consumers/cdp-cyclotron-segment-worker.consumer.ts

+
+    public async processInvocations(invocations: HogFunctionInvocation[]): Promise<HogFunctionInvocationResult[]> {
+        // Segment plugins fire fetch requests and so need to be run in true parallel
+        return await Promise.all(


Jumping in - the case here is basically that the executor code should only throw if there is a genuine programming error. If so we essentially have no choice but to crash out so there isn't much of a difference between all and allSettled, as we have no course of action to partially handle a batch.

Generally it would be good if we could partially handle a batch, but thats not how kafka works so this is an issue with out architecture moreso than this specific code path

MarconLP · 2025-05-22T15:51:08Z

Want to get this merged with all destinations hidden for now. https://github.com/PostHog/posthog/blob/external-destinations/plugin-server/src/cdp/segment/segment-templates.ts#L321-L331

benjackwhite

I think this is good to go. Still some follow up needed for sure but lets iterate

benjackwhite · 2025-05-23T07:48:48Z

plugin-server/src/cdp/consumers/cdp-cyclotron-segment-worker.test.ts

+                }
+            })
+
+            expect(invocationResults[0].logs).toMatchInlineSnapshot(`


When you have really big snapshots like this we should just use toMatchSnapshot - inline ones are good when there is a limited number of options. Now the file is so huge its virtually unreadable.

benjackwhite · 2025-05-23T07:50:15Z

plugin-server/src/cdp/services/segment-destination-executor.service.ts

+            pluginExecutionDuration.observe(performance.now() - start)
+        } catch (e) {
+            if (e instanceof RetryError) {
+                // NOTE: Schedule as a retry to cyclotron?


This still needs to be implemented?

Can be left out for this PR but needs to be done as follow up asap

adjusted requested changes

MarconLP added 5 commits April 16, 2025 09:22

add amplitude destination

3b37c1d

install dependencies

896962f

add dayjs

ce7f3bb

add plugin to list

d2802d2

map mapping templates

03dfe82

greptile-apps bot reviewed Apr 17, 2025

View reviewed changes

MarconLP added 12 commits April 18, 2025 11:26

add logic to translate filters and inputs

11e010b

fix

fab5861

fixes

21dab90

fixes

734ddd1

increase pagination limit

758e2e6

fixes

544a9ee

fixes

fc8e117

remove copied code

5d7bb87

move template.ts somewhere else

057d1a1

remove old code

023b3da

fixes

eb4a7a5

fixes

22d8e76

graphite-app bot reviewed Apr 24, 2025

View reviewed changes

plugin-server/src/cdp/services/legacy-plugin-executor.service.ts Outdated Show resolved Hide resolved

fixes

28b3a09

graphite-app bot reviewed Apr 24, 2025

View reviewed changes

MarconLP and others added 5 commits April 24, 2025 15:24

rename

2bd4168

add tests

631f7c0

hide partner action field

d4ab5fa

ignore useless async keyword

7cf89cc

Update UI snapshots for chromium (7)

2a4b1a5

Update UI snapshots for chromium (7)

10fd4d3

update snapshot tests

7ac43b4

MarconLP enabled auto-merge (squash) May 19, 2025 22:04

MarconLP added 2 commits May 20, 2025 21:52

fix version in snapshot

07d10b5

Merge branch 'master' into external-destinations

aab5186

meikelmosby previously requested changes May 21, 2025

View reviewed changes

MarconLP added 11 commits May 21, 2025 10:38

revert make id nullable

a23c5f6

Discard changes to rust/Cargo.lock

13c8579

change log type to debug

84ba4f7

fix ts issue

3ed1bb6

add id

5aed870

update snapshot

f13f5b6

hide all destinations at the moment

73c12dc

add tests

6f5f289

remove note

1bd318f

fix timezone issue in test

7c32599

fix?

fe348c3

benjackwhite reviewed May 22, 2025

View reviewed changes

MarconLP added 6 commits May 22, 2025 15:16

update template to hide all destinations

cfac0f4

another timezone fix

deffd5b

fix

bea8db6

Merge branch 'master' into external-destinations

a07f0bf

fixes

da594c1

increase timeout for test

3eb7584

benjackwhite approved these changes May 23, 2025

View reviewed changes

MarconLP disabled auto-merge May 23, 2025 07:54

MarconLP merged commit c3944dc into master May 23, 2025
93 of 96 checks passed

MarconLP deleted the external-destinations branch May 23, 2025 08:03

MarconLP mentioned this pull request May 26, 2025

feat(cdp): segment follow up pr #32651

Merged

11 tasks

feat(cdp): segment destination mapping #31336

feat(cdp): segment destination mapping #31336

Uh oh!

Conversation

MarconLP commented Apr 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

PostHog / Segment view

PostHog

Segment

Refer to this PR for the followup status #32651

Does this work well for both Cloud and self-hosted?

How did you test this code?

Uh oh!

greptile-apps bot left a comment

Choose a reason for hiding this comment

PR Summary

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Apr 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

graphite-app bot Apr 24, 2025

Choose a reason for hiding this comment

Uh oh!

posthog-bot commented Apr 24, 2025

📸 UI snapshots have been updated

Uh oh!

posthog-bot commented Apr 24, 2025

📸 UI snapshots have been updated

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MarconLP May 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MarconLP commented May 22, 2025

Uh oh!

benjackwhite left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MarconLP commented Apr 17, 2025 •

edited

Loading

github-actions bot commented Apr 17, 2025 •

edited

Loading

MarconLP May 22, 2025 •

edited

Loading