feat(workflows): report every committed workflow creation

products/workflows/backend/models/hog_flow/hog_flow.py

@@ -8,11 +8,14 @@
 
 import structlog
 
+from posthog.event_usage import groups
 from posthog.helpers.encrypted_fields import EncryptedJSONStringField
 from posthog.models.tagged_items_relation import Taggable
 from posthog.models.team.team import Team
 from posthog.models.utils import UUIDTModel
+from posthog.ph_client import ph_scoped_capture
 from posthog.plugins.plugin_server_api import reload_hog_flows_on_workers
+from posthog.utils import get_instance_region
 
 from products.actions.backend.models.action import Action
 from products.workflows.backend.facade.enums import (
@@ -299,8 +302,34 @@ def __str__(self):
         return f"HogFlow {self.id}/{self.version}: {self.name}"
 
 
+def report_workflow_created(team_id: int, workflow_id: str, origin_product: str | None, actions_count: int) -> None:
+    try:
+        team = Team.objects.only("uuid", "organization_id").get(id=team_id)
+        with ph_scoped_capture(event_region=get_instance_region()) as capture:
+            capture(
+                distinct_id=str(team.uuid),
+                event="workflows workflow created",
+                properties={
+                    "workflow_id": workflow_id,
+                    "origin_product": origin_product,
+                    "actions_count": actions_count,
+                },
+                groups=groups(team=team),
+            )
+    except Exception:
+        logger.exception("Failed to report workflow creation", team_id=team_id, workflow_id=workflow_id)
+
+
 @receiver(post_save, sender=HogFlow)
-def hog_flow_saved(sender, instance: HogFlow, created, update_fields=None, **kwargs):
+def hog_flow_saved(sender, instance: HogFlow, created, update_fields=None, using="default", **kwargs):
+    if created:
+        team_id = instance.team_id
+        workflow_id = str(instance.id)
+        origin_product = instance.origin_product
+        actions_count = len(instance.actions or [])
+        transaction.on_commit(
+            lambda: report_workflow_created(team_id, workflow_id, origin_product, actions_count), using=using
+        )
     # Draft columns don't affect live execution, so workers don't need a config reload for them.
     if update_fields and set(update_fields) <= {"draft", "draft_updated_at"}:
         return

products/workflows/backend/test/test_hog_flow.py

@@ -1,14 +1,22 @@
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from django.core.management import call_command
-from django.test import TestCase
+from django.db import transaction
+from django.test import TestCase, override_settings
 from django.utils import timezone
 
 from posthog.models.user import User
 
 from products.actions.backend.models.action import Action
 from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
 
+SCOPED_CAPTURE = "products.workflows.backend.models.hog_flow.hog_flow.ph_scoped_capture"
+
+
+def captured_events(scoped_capture: MagicMock) -> list[dict]:
+    capture = scoped_capture.return_value.__enter__.return_value
+    return [call.kwargs for call in capture.call_args_list]
+
 
 class TestHogFlow(TestCase):
     def setUp(self):
@@ -18,6 +26,59 @@ def setUp(self):
         self.user = user
         self.org = org
 
+    @patch(SCOPED_CAPTURE)
+    def test_creation_reports_once_after_commit(self, scoped_capture: MagicMock) -> None:
+        with self.captureOnCommitCallbacks(execute=True):
+            flow = HogFlow.objects.create(
+                name="Welcome",
+                team=self.team,
+                created_by=self.user,
+                origin_product="broadcasts",
+                actions=[{"type": "trigger"}],
+            )
+            flow.name = "Renamed"
+            flow.origin_product = "loops"
+            flow.actions = []
+            flow.save()
+            scoped_capture.assert_not_called()
+
+        events = captured_events(scoped_capture)
+        assert len(events) == 1
+        assert events[0]["event"] == "workflows workflow created"
+        assert events[0]["distinct_id"] == str(self.team.uuid)
+        assert events[0]["groups"]["project"] == str(self.team.uuid)
+        assert events[0]["groups"]["organization"] == str(self.org.id)
+        assert events[0]["properties"] == {
+            "workflow_id": str(flow.id),
+            "origin_product": "broadcasts",
+            "actions_count": 1,
+        }
+
+    @override_settings(CLOUD_DEPLOYMENT="EU")
+    @patch(SCOPED_CAPTURE)
+    def test_creation_reports_from_the_instance_region(self, scoped_capture: MagicMock) -> None:
+        with self.captureOnCommitCallbacks(execute=True):
+            HogFlow.objects.create(name="Welcome", team=self.team)
+        scoped_capture.assert_called_once_with(event_region="EU")
+
+    @patch(SCOPED_CAPTURE, side_effect=RuntimeError("Analytics unavailable"))
+    def test_analytics_failure_does_not_fail_creation(self, scoped_capture: MagicMock) -> None:
+        with self.captureOnCommitCallbacks(execute=True):
+            flow = HogFlow.objects.create(name="Welcome", team=self.team)
+        scoped_capture.assert_called_once()
+        assert HogFlow.objects.filter(id=flow.id, team=self.team).exists()
+
+    @patch(SCOPED_CAPTURE)
+    def test_rolled_back_creation_does_not_report(self, scoped_capture: MagicMock) -> None:
+        with self.captureOnCommitCallbacks(execute=True):
+            try:
+                with transaction.atomic():
+                    HogFlow.objects.create(name="Rolled back", team=self.team)
+                    raise ValueError("Abort")
+            except ValueError:
+                pass
+        scoped_capture.assert_not_called()
+
     @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
     def test_hog_flow_saved_receiver(self, mock_reload):
         hog_flow = HogFlow.objects.create(name="Test Flow", team=self.team)