From 704511b9c186ea36add599c653b7ad7e9c994c25 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Mon, 12 Oct 2026 10:25:26 +0000 Subject: [PATCH 1/3] feat(workflows): report every committed workflow creation Co-Authored-By: Claude Opus 5.5 --- .../backend/models/hog_flow/hog_flow.py | 23 +++++++- .../workflows/backend/test/test_hog_flow.py | 53 +++++++++++++++++++ 2 files changed, 75 insertions(+), 1 deletion(-) diff --git a/products/workflows/backend/models/hog_flow/hog_flow.py b/products/workflows/backend/models/hog_flow/hog_flow.py index 0006761dd448..0ffa7dd235f0 100644 --- a/products/workflows/backend/models/hog_flow/hog_flow.py +++ b/products/workflows/backend/models/hog_flow/hog_flow.py @@ -8,6 +8,7 @@ import structlog +from posthog.event_usage import report_team_action from posthog.helpers.encrypted_fields import EncryptedJSONStringField from posthog.models.tagged_items_relation import Taggable from posthog.models.team.team import Team @@ -299,8 +300,28 @@ 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) + report_team_action( + team, + "workflows workflow created", + {"workflow_id": workflow_id, "origin_product": origin_product, "actions_count": actions_count}, + ) + 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 diff --git a/products/workflows/backend/test/test_hog_flow.py b/products/workflows/backend/test/test_hog_flow.py index a3f6e1066da7..fc46347a8d34 100644 --- a/products/workflows/backend/test/test_hog_flow.py +++ b/products/workflows/backend/test/test_hog_flow.py @@ -1,6 +1,9 @@ +from typing import TYPE_CHECKING + from unittest.mock import patch from django.core.management import call_command +from django.db import transaction from django.test import TestCase from django.utils import timezone @@ -9,6 +12,9 @@ from products.actions.backend.models.action import Action from products.workflows.backend.models.hog_flow.hog_flow import HogFlow +if TYPE_CHECKING: + from unittest.mock import MagicMock + class TestHogFlow(TestCase): def setUp(self): @@ -18,6 +24,53 @@ def setUp(self): self.user = user self.org = org + @patch("posthoganalytics.capture") + def test_creation_reports_once_after_commit(self, 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() + capture.assert_not_called() + + events = [ + call.kwargs for call in capture.call_args_list if call.kwargs.get("event") == "workflows workflow created" + ] + assert len(events) == 1 + 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, + } + + @patch("posthoganalytics.capture", side_effect=RuntimeError("Analytics unavailable")) + def test_analytics_failure_does_not_fail_creation(self, capture: MagicMock) -> None: + with self.captureOnCommitCallbacks(execute=True): + flow = HogFlow.objects.create(name="Welcome", team=self.team) + capture.assert_called_once() + assert HogFlow.objects.filter(id=flow.id, team=self.team).exists() + + @patch("posthoganalytics.capture") + def test_rolled_back_creation_does_not_report(self, 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 + 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) From a006eb466b83a5a6743e100533ceb590097b236f Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:12:19 +0000 Subject: [PATCH 2/3] fix(workflows): deliver the creation event from celery workers too A data suggestion saved by a Celery task creates its workflow inside the worker, where the global analytics client can lose queued events when the child exits. Capture through the scoped client, which flushes on exit. Co-Authored-By: Claude Opus 5.5 --- .../backend/models/hog_flow/hog_flow.py | 20 +++++++---- .../workflows/backend/test/test_hog_flow.py | 35 ++++++++++--------- 2 files changed, 32 insertions(+), 23 deletions(-) diff --git a/products/workflows/backend/models/hog_flow/hog_flow.py b/products/workflows/backend/models/hog_flow/hog_flow.py index 0ffa7dd235f0..dc723bd04c03 100644 --- a/products/workflows/backend/models/hog_flow/hog_flow.py +++ b/products/workflows/backend/models/hog_flow/hog_flow.py @@ -8,12 +8,14 @@ import structlog -from posthog.event_usage import report_team_action +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 ( @@ -303,11 +305,17 @@ def __str__(self): 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) - report_team_action( - team, - "workflows workflow created", - {"workflow_id": workflow_id, "origin_product": origin_product, "actions_count": actions_count}, - ) + 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) diff --git a/products/workflows/backend/test/test_hog_flow.py b/products/workflows/backend/test/test_hog_flow.py index fc46347a8d34..f0e903ea19aa 100644 --- a/products/workflows/backend/test/test_hog_flow.py +++ b/products/workflows/backend/test/test_hog_flow.py @@ -1,6 +1,4 @@ -from typing import TYPE_CHECKING - -from unittest.mock import patch +from unittest.mock import MagicMock, patch from django.core.management import call_command from django.db import transaction @@ -12,8 +10,12 @@ from products.actions.backend.models.action import Action from products.workflows.backend.models.hog_flow.hog_flow import HogFlow -if TYPE_CHECKING: - from unittest.mock import MagicMock +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): @@ -24,8 +26,8 @@ def setUp(self): self.user = user self.org = org - @patch("posthoganalytics.capture") - def test_creation_reports_once_after_commit(self, capture: MagicMock) -> None: + @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", @@ -38,12 +40,11 @@ def test_creation_reports_once_after_commit(self, capture: MagicMock) -> None: flow.origin_product = "loops" flow.actions = [] flow.save() - capture.assert_not_called() + scoped_capture.assert_not_called() - events = [ - call.kwargs for call in capture.call_args_list if call.kwargs.get("event") == "workflows workflow created" - ] + 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) @@ -53,15 +54,15 @@ def test_creation_reports_once_after_commit(self, capture: MagicMock) -> None: "actions_count": 1, } - @patch("posthoganalytics.capture", side_effect=RuntimeError("Analytics unavailable")) - def test_analytics_failure_does_not_fail_creation(self, capture: MagicMock) -> None: + @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) - capture.assert_called_once() + scoped_capture.assert_called_once() assert HogFlow.objects.filter(id=flow.id, team=self.team).exists() - @patch("posthoganalytics.capture") - def test_rolled_back_creation_does_not_report(self, capture: MagicMock) -> None: + @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(): @@ -69,7 +70,7 @@ def test_rolled_back_creation_does_not_report(self, capture: MagicMock) -> None: raise ValueError("Abort") except ValueError: pass - capture.assert_not_called() + 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): From b170c9c1dd5e1cc7e3c1bbe29160037e597e9490 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:19:00 +0000 Subject: [PATCH 3/3] test(workflows): pin the creation event to the instance region Co-Authored-By: Claude Opus 5.5 --- products/workflows/backend/test/test_hog_flow.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/products/workflows/backend/test/test_hog_flow.py b/products/workflows/backend/test/test_hog_flow.py index f0e903ea19aa..65b69fa26170 100644 --- a/products/workflows/backend/test/test_hog_flow.py +++ b/products/workflows/backend/test/test_hog_flow.py @@ -2,7 +2,7 @@ from django.core.management import call_command from django.db import transaction -from django.test import TestCase +from django.test import TestCase, override_settings from django.utils import timezone from posthog.models.user import User @@ -54,6 +54,13 @@ def test_creation_reports_once_after_commit(self, scoped_capture: MagicMock) -> "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):