feat(cdp): Working customer.io destination (#24134)

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

@@ -97,7 +97,7 @@ function DictionaryField({ onChange, value }: { onChange?: (value: any) => void;
                     />
 
                     <HogFunctionTemplateInput
-                        className="flex-2 max-w-full"
+                        className="flex-2 overflow-hidden"
                         value={val}
                         language="hogTemplate"
                         onChange={(val) => {

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

@@ -145,7 +145,7 @@ export const hogFunctionConfigurationLogic = kea<hogFunctionConfigurationLogicTy
         upsertHogFunction: (configuration: HogFunctionConfigurationType) => ({ configuration }),
         duplicate: true,
         duplicateFromTemplate: true,
-        resetToTemplate: true,
+        resetToTemplate: (keepInputs = true) => ({ keepInputs }),
         deleteHogFunction: true,
         sparklineQueryChanged: (sparklineQuery: TrendsQuery) => ({ sparklineQuery } as { sparklineQuery: TrendsQuery }),
     }),
@@ -156,6 +156,13 @@ export const hogFunctionConfigurationLogic = kea<hogFunctionConfigurationLogicTy
                 setShowSource: (_, { showSource }) => showSource,
             },
         ],
+
+        hasHadSubmissionErrors: [
+            false,
+            {
+                upsertHogFunctionFailure: () => true,
+            },
+        ],
     }),
     loaders(({ props, values }) => ({
         template: [
@@ -576,15 +583,15 @@ export const hogFunctionConfigurationLogic = kea<hogFunctionConfigurationLogicTy
                 )
             }
         },
-        resetToTemplate: async () => {
+        resetToTemplate: async ({ keepInputs }) => {
             if (values.hogFunction?.template) {
                 const template = values.hogFunction.template
                 // Fill defaults from template
                 const inputs: Record<string, HogFunctionInputType> = {}
 
                 template.inputs_schema?.forEach((schema) => {
-                    if (schema.default) {
-                        inputs[schema.key] = { value: schema.default }
+                    inputs[schema.key] = (keepInputs ? values.configuration.inputs?.[schema.key] : undefined) ?? {
+                        value: schema.default,
                     }
                 })
 
@@ -600,8 +607,10 @@ export const hogFunctionConfigurationLogic = kea<hogFunctionConfigurationLogicTy
             }
         },
         setConfigurationValue: () => {
-            // Clear the manually set errors otherwise the submission won't work
-            actions.setConfigurationManualErrors({})
+            if (values.hasHadSubmissionErrors) {
+                // Clear the manually set errors otherwise the submission won't work
+                actions.setConfigurationManualErrors({})
+            }
         },
 
         deleteHogFunction: async () => {

posthog/cdp/templates/customerio/template_customerio.py

@@ -3,42 +3,59 @@
 # Based off of https://customer.io/docs/api/track/#operation/entity
 
 template: HogFunctionTemplate = HogFunctionTemplate(
-    status="alpha",
+    status="beta",
     id="template-customerio",
-    name="Update persons in Customer.io",
-    description="Updates persons in Customer.io",
+    name="Send events to Customer.io",
+    description="Identify or track events against customers in Customer.io",
     icon_url="/static/services/customerio.png",
     hog="""
-fn callCustomerIoApi(method, path, body) {
-    // TODO: Base64 encode the site_id and token
-    fetch(f'https://{inputs.host}{path}', {
-        'method': 'POST',
-        'headers': {
-            'User-Agent': 'PostHog Customer.io App',
-            'Authorization': f'Basic {base64Encode(f'{inputs.site_id}:{inputs.token}')}',
-            'Content-Type': 'application/json'
-        },
-        'body': body
-    })
+let action := inputs.action
+let name := event.name
+
+if (action == 'automatic') {
+    if (event.name == '$identify') {
+        action := 'identify'
+        name := null
+    } else if (event.name == '$pageview') {
+        action := 'page'
+        name := event.properties.$current_url
+    } else if (event.name == '$screen') {
+        action := 'screen'
+        name := event.properties.$screen_name
+    } else {
+        action := 'event'
+    }
+}
+
+let attributes := inputs.include_all_properties ? event.properties : {}
+
+let timestamp := toInt(toUnixTimestamp(toDateTime(event.timestamp)))
+
+for (let key, value in inputs.attributes) {
+    attributes[key] := value
 }
 
-fn trackIdentify() {
-    // Upsert the customer
-    let payload := {
+let res := fetch(f'https://{inputs.host}/api/v2/entity', {
+    'method': 'POST',
+    'headers': {
+        'User-Agent': 'PostHog Customer.io App',
+        'Authorization': f'Basic {base64Encode(f'{inputs.site_id}:{inputs.token}')}',
+        'Content-Type': 'application/json'
+    },
+    'body': {
         'type': 'person',
-        'identifiers': {
-            // TODO: Make the id input configurable
-            'id': inputs.identifier
-        },
-        'action': 'identify',
-        'attributes': inputs.properties
+        'action': action,
+        'name': name,
+        'identifiers': inputs.identifiers,
+        'attributes': attributes,
+        'timestamp': timestamp
     }
+})
 
-    await callCustomerIoApi('POST', f'/api/v2/entity', payload)
+if (res.status >= 400) {
+    print('Error from customer.io api:', res.status, res.body)
 }
 
-trackIdentify()
-
 """.strip(),
     inputs_schema=[
         {
@@ -56,15 +73,6 @@
             "secret": True,
             "required": True,
         },
-        {
-            "key": "identifier",
-            "type": "string",
-            "label": "The ID that should be used for the user",
-            "description": "You can choose to fill this from an email property or an ID property. If the value is empty nothing will be sent.",
-            "default": "{person.properties.email}",
-            "secret": False,
-            "required": True,
-        },
         {
             "key": "host",
             "type": "choice",
@@ -85,21 +93,79 @@
             "required": True,
         },
         {
-            "key": "properties",
+            "key": "identifiers",
             "type": "dictionary",
-            "label": "Property mapping",
-            "description": "Map of Customer.io person properties and their values. You can use the filters section to filter out unwanted events.",
+            "label": "Identifiers",
+            "description": "You can choose to fill this from an `email` property or an `id` property. If the value is empty nothing will be sent. See here for more information: https://customer.io/docs/api/track/#operation/entity",
+            "default": {
+                "email": "{person.properties.email}",
+            },
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "action",
+            "type": "choice",
+            "label": "Action",
+            "description": "Choose the action to be tracked. Automatic will convert $identify, $pageview and $screen to identify, page and screen automatically - otherwise defaulting to event",
+            "default": "automatic",
+            "choices": [
+                {
+                    "label": "Automatic",
+                    "value": "automatic",
+                },
+                {
+                    "label": "Identify",
+                    "value": "identify",
+                },
+                {
+                    "label": "Event",
+                    "value": "event",
+                },
+                {
+                    "label": "Page",
+                    "value": "page",
+                },
+                {
+                    "label": "Screen",
+                    "value": "screen",
+                },
+                {
+                    "label": "Delete",
+                    "value": "delete",
+                },
+            ],
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "include_all_properties",
+            "type": "boolean",
+            "label": "Include all properties as attributes",
+            "description": "If set, all event properties will be included as attributes. Individual attributes can be overridden below.",
+            "default": False,
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "attributes",
+            "type": "dictionary",
+            "label": "Attribute mapping",
+            "description": "Map of Customer.io attributes and their values. You can use the filters section to filter out unwanted events.",
             "default": {
                 "email": "{person.properties.email}",
                 "lastname": "{person.properties.lastname}",
                 "firstname": "{person.properties.firstname}",
             },
             "secret": False,
-            "required": True,
+            "required": False,
         },
     ],
     filters={
-        "events": [{"id": "$identify", "name": "$identify", "type": "events", "order": 0}],
+        "events": [
+            {"id": "$identify", "name": "$identify", "type": "events", "order": 0},
+            {"id": "$pageview", "name": "$pageview", "type": "events", "order": 0},
+        ],
         "actions": [],
         "filter_test_accounts": True,
     },

posthog/cdp/templates/customerio/test_template_customerio.py

@@ -1,42 +1,98 @@
+from inline_snapshot import snapshot
 from posthog.cdp.templates.helpers import BaseHogFunctionTemplateTest
 from posthog.cdp.templates.customerio.template_customerio import template as template_customerio
 
 
+def create_inputs(**kwargs):
+    inputs = {
+        "site_id": "SITE_ID",
+        "token": "TOKEN",
+        "host": "track.customer.io",
+        "action": "automatic",
+        "include_all_properties": False,
+        "identifiers": {"email": "[email protected]"},
+        "attributes": {"name": "example"},
+    }
+    inputs.update(kwargs)
+
+    return inputs
+
+
 class TestTemplateCustomerio(BaseHogFunctionTemplateTest):
     template = template_customerio
 
-    def _inputs(self, **kwargs):
-        inputs = {
-            "site_id": "SITE_ID",
-            "token": "TOKEN",
-            "identifier": "[email protected]",
-            "host": "track.customer.io",
-            "properties": {"name": "example"},
-        }
-        inputs.update(kwargs)
-        return inputs
-
     def test_function_works(self):
-        res = self.run_function(inputs=self._inputs())
-
-        assert res.result is None
-
-        assert self.get_mock_fetch_calls()[0] == (
-            "https://track.customer.io/api/v2/entity",
-            {
-                "method": "POST",
-                "headers": {
-                    "User-Agent": "PostHog Customer.io App",
-                    "Authorization": "Basic U0lURV9JRDpUT0tFTg==",
-                    "Content-Type": "application/json",
-                },
-                "body": {
-                    "type": "person",
-                    "identifiers": {"id": "[email protected]"},
-                    "action": "identify",
-                    "attributes": {
-                        "name": "example",
+        self.run_function(
+            inputs=create_inputs(),
+            globals={
+                "event": {"name": "$pageview", "properties": {"url": "https://example.com"}},
+            },
+        )
+
+        assert self.get_mock_fetch_calls()[0] == snapshot(
+            (
+                "https://track.customer.io/api/v2/entity",
+                {
+                    "method": "POST",
+                    "headers": {
+                        "User-Agent": "PostHog Customer.io App",
+                        "Authorization": "Basic U0lURV9JRDpUT0tFTg==",
+                        "Content-Type": "application/json",
+                    },
+                    "body": {
+                        "type": "person",
+                        "action": "page",
+                        "name": None,
+                        "identifiers": {"email": "[email protected]"},
+                        "attributes": {"name": "example"},
+                        "timestamp": 1704067200,
                     },
                 },
-            },
+            )
         )
+
+    def test_body_includes_all_properties_if_set(self):
+        self.run_function(inputs=create_inputs(include_all_properties=False))
+
+        assert self.get_mock_fetch_calls()[0][1]["body"]["attributes"] == snapshot({"name": "example"})
+
+        self.run_function(inputs=create_inputs(include_all_properties=True))
+
+        assert self.get_mock_fetch_calls()[0][1]["body"]["attributes"] == snapshot(
+            {"$current_url": "https://example.com", "name": "example"}
+        )
+
+    def test_automatic_action_mapping(self):
+        for event_name, expected_action in [
+            ("$identify", "identify"),
+            ("$pageview", "page"),
+            ("$screen", "screen"),
+            ("$autocapture", "event"),
+            ("custom", "event"),
+        ]:
+            self.run_function(
+                inputs=create_inputs(),
+                globals={
+                    "event": {"name": event_name, "properties": {"url": "https://example.com"}},
+                },
+            )
+
+            assert self.get_mock_fetch_calls()[0][1]["body"]["action"] == expected_action
+
+    def test_enforced_action(self):
+        for event_name in [
+            "$identify",
+            "$pageview",
+            "$screen",
+            "$autocapture",
+            "custom",
+        ]:
+            self.run_function(
+                inputs=create_inputs(action="event"),
+                globals={
+                    "event": {"name": event_name, "properties": {"url": "https://example.com"}},
+                },
+            )
+
+            assert self.get_mock_fetch_calls()[0][1]["body"]["action"] == "event"
+            assert self.get_mock_fetch_calls()[0][1]["body"]["name"] == event_name

posthog/cdp/templates/helpers.py

@@ -63,6 +63,8 @@ def createHogGlobals(self, globals=None) -> dict:
         return data
 
     def run_function(self, inputs: dict, globals=None):
+        self.mock_fetch.reset_mock()
+        self.mock_print.reset_mock()
         # Create the globals object
         globals = self.createHogGlobals(globals)
         globals["inputs"] = inputs

posthog/cdp/validation.py

@@ -60,7 +60,7 @@ def validate(self, attrs):
         name: str = schema["key"]
         item_type = schema["type"]
 
-        if schema.get("required") and not value:
+        if schema.get("required") and (value is None or value == ""):
             raise serializers.ValidationError({"inputs": {name: f"This field is required."}})
 
         if not value: