feat(cdp): add discord template

posthog/cdp/templates/__init__.py

@@ -8,6 +8,7 @@
 from .clearbit.template_clearbit import template as clearbit
 from .posthog.template_posthog import template as posthog, TemplatePostHogMigrator
 from .aws_kinesis.template_aws_kinesis import template as aws_kinesis
+from .discord.template_discord import template as discord
 from .salesforce.template_salesforce import template_create as salesforce_create, template_update as salesforce_update
 from .mailjet.template_mailjet import (
     template_create_contact as mailjet_create_contact,
@@ -43,6 +44,7 @@
     braze,
     clearbit,
     customerio,
+    discord,
     engage,
     gleap,
     google_ads,

posthog/cdp/templates/discord/template_discord.py

@@ -0,0 +1,66 @@
+from posthog.cdp.templates.hog_function_template import HogFunctionTemplate, HogFunctionSubTemplate, SUB_TEMPLATE_COMMON
+
+template: HogFunctionTemplate = HogFunctionTemplate(
+    status="free",
+    id="template-discord",
+    name="Discord",
+    description="Sends a message to a discord channel",
+    icon_url="/static/services/discord.png",
+    category=["Customer Success"],
+    hog="""
+if (not match(inputs.webhookUrl, '^https://discord.com/api/webhooks/.*')) {
+    throw Error('Invalid url');
+}
+
+let res := fetch(inputs.webhookUrl, {
+    'body': {
+        'content': inputs.content
+    },
+    'method': 'POST',
+    'headers': {
+        'Content-Type': 'application/json'
+    }
+});
+
+if (res.status >= 400) {
+    throw Error(f'Failed to post message to Discord: {res.status}: {res.body}');
+}
+""".strip(),
+    inputs_schema=[
+        {
+            "key": "webhookUrl",
+            "type": "string",
+            "label": "Webhook URL",
+            "description": "See this page on how to generate a Webhook URL: https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks",
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "content",
+            "type": "string",
+            "label": "Content",
+            "description": "(see https://support.discord.com/hc/en-us/articles/210298617-Markdown-Text-101-Chat-Formatting-Bold-Italic-Underline)",
+            "default": "**{person.name}** triggered event: '{event.event}'",
+            "secret": False,
+            "required": True,
+        },
+    ],
+    sub_templates=[
+        HogFunctionSubTemplate(
+            id="early_access_feature_enrollment",
+            name="Post to Discord on feature enrollment",
+            description="Posts a message to Discord when a user enrolls or un-enrolls in an early access feature",
+            filters=SUB_TEMPLATE_COMMON["early_access_feature_enrollment"].filters,
+            inputs={
+                "content": "**{person.name}** {event.properties.$feature_enrollment ? 'enrolled in' : 'un-enrolled from'} the early access feature for '{event.properties.$feature_flag}'"
+            },
+        ),
+        HogFunctionSubTemplate(
+            id="survey_response",
+            name="Post to Discord on survey response",
+            description="Posts a message to Discord when a user responds to a survey",
+            filters=SUB_TEMPLATE_COMMON["survey_response"].filters,
+            inputs={"content": "**{person.name}** responded to survey **{event.properties.$survey_name}**"},
+        ),
+    ],
+)

posthog/cdp/templates/discord/test_template_discord.py

@@ -0,0 +1,50 @@
+import pytest
+from inline_snapshot import snapshot
+from posthog.cdp.templates.helpers import BaseHogFunctionTemplateTest
+from posthog.cdp.templates.discord.template_discord import template as template_discord
+
+
+class TestTemplateDiscord(BaseHogFunctionTemplateTest):
+    template = template_discord
+
+    def _inputs(self, **kwargs):
+        inputs = {
+            "webhookUrl": "https://discord.com/api/webhooks/00000000000000000/xxxxxxxxxxxxxx",
+            "content": "**[email protected]** triggered event: '$pageview'",
+        }
+        inputs.update(kwargs)
+        return inputs
+
+    def test_function_works(self):
+        self.run_function(inputs=self._inputs())
+
+        assert self.get_mock_fetch_calls()[0] == snapshot(
+            (
+                "https://discord.com/api/webhooks/00000000000000000/xxxxxxxxxxxxxx",
+                {
+                    "method": "POST",
+                    "headers": {
+                        "Content-Type": "application/json",
+                    },
+                    "body": {
+                        "content": "**[email protected]** triggered event: '$pageview'",
+                    },
+                },
+            )
+        )
+
+    def test_only_allow_teams_url(self):
+        for url, allowed in [
+            ["https://discord.com/api/webhooks/abc", True],
+            ["https://webhook.site/def", False],
+            ["https://webhook.site/def#https://discord.com/api/webhooks/abc", False],
+        ]:
+            if allowed:
+                self.run_function(inputs=self._inputs(webhookUrl=url))
+                assert len(self.get_mock_fetch_calls()) == 1
+            else:
+                with pytest.raises(Exception) as e:
+                    self.run_function(inputs=self._inputs(webhookUrl=url))
+                assert (
+                    e.value.message == "Invalid url"  # type: ignore[attr-defined]
+                )