fix: align usage event with MPS org-event convention (per-member fan-out)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

api/db/organization_client.py

@@ -8,6 +8,7 @@ from api.db.base_client import BaseDBClient
 from api.db.models import (
     APIKeyModel,
     OrganizationModel,
+    UserModel,
     organization_users_association,
 )
 from api.utils.api_key import generate_api_key
@@ -24,6 +25,22 @@ class OrganizationClient(BaseDBClient):
             )
             return result.scalars().first()
 
+    async def get_organization_users(self, organization_id: int) -> list[UserModel]:
+        """Get all users linked to an organization (many-to-many)."""
+        async with self.async_session() as session:
+            result = await session.execute(
+                select(UserModel)
+                .join(
+                    organization_users_association,
+                    organization_users_association.c.user_id == UserModel.id,
+                )
+                .where(
+                    organization_users_association.c.organization_id == organization_id
+                )
+                .order_by(UserModel.id)
+            )
+            return list(result.scalars().all())
+
     async def get_or_create_organization_by_provider_id(
         self, org_provider_id: str, user_id: int
     ) -> tuple[OrganizationModel, bool]:

api/services/call_concurrency.py

@@ -8,11 +8,7 @@ from api.constants import DEFAULT_ORG_CONCURRENCY_LIMIT
 from api.db import db_client
 from api.enums import OrganizationConfigurationKey, PostHogEvent
 from api.services.campaign.rate_limiter import rate_limiter
-from api.services.posthog_client import (
-    POSTHOG_ORGANIZATION_GROUP_TYPE,
-    POSTHOG_SERVER_EVENT_DISTINCT_ID,
-    capture_event,
-)
+from api.services.posthog_client import capture_event
 
 
 @dataclass(frozen=True)
@@ -134,6 +130,7 @@ class CallConcurrencyService:
                     f"{scope_note}, waited={wait_time:.1f}s"
                 )
                 properties = {
+                    "event_source": "dograh",
                     "organization_id": organization_id,
                     "source": source,
                     "max_concurrent": max_concurrent,
@@ -143,12 +140,7 @@ class CallConcurrencyService:
                 if scope_key:
                     properties["scope_key"] = scope_key
                     properties["scope_max_concurrent"] = scope_max_concurrent
-                capture_event(
-                    distinct_id=POSTHOG_SERVER_EVENT_DISTINCT_ID,
-                    event=PostHogEvent.USAGE_CONCURRENT_CALL_LIMIT_REACHED,
-                    properties=properties,
-                    groups={POSTHOG_ORGANIZATION_GROUP_TYPE: str(organization_id)},
-                )
+                await self._notify_limit_reached(organization_id, properties)
                 raise CallConcurrencyLimitError(
                     organization_id=organization_id,
                     source=source,
@@ -162,6 +154,38 @@ class CallConcurrencyService:
             )
             await asyncio.sleep(min(retry_interval, max(0, timeout - wait_time)))
 
+    async def _notify_limit_reached(
+        self, organization_id: int, properties: dict
+    ) -> None:
+        """Fan the usage event out to every org member's provider_id, matching
+        how MPS emits org-scoped billing events (billing_posthog_service.py)
+        into the shared PostHog project. Never raises.
+
+        NOTE: intentionally NOT attaching ``$groups`` (organization) to this
+        event. PostHog evaluates a $groups event at both person and group
+        scope, which double-triggers person-scoped workflows enrolled on the
+        event. The org is still available as the ``organization_id`` property.
+        """
+        try:
+            members = await db_client.get_organization_users(organization_id)
+            if not members:
+                logger.debug(
+                    f"No users found for org {organization_id}; skipping "
+                    "concurrent-call-limit PostHog event"
+                )
+                return
+            for member in members:
+                capture_event(
+                    distinct_id=str(member.provider_id),
+                    event=PostHogEvent.USAGE_CONCURRENT_CALL_LIMIT_REACHED,
+                    properties=properties,
+                )
+        except Exception:
+            logger.exception(
+                "Failed to send concurrent-call-limit PostHog event for org "
+                f"{organization_id}"
+            )
+
     async def bind_workflow_run(
         self, slot: CallConcurrencySlot, workflow_run_id: int
     ) -> None:

api/services/posthog_client.py

@@ -8,11 +8,6 @@ from api.constants import POSTHOG_API_KEY, POSTHOG_HOST
 _posthog_client: Posthog | None = None
 POSTHOG_SERVER_GROUP_IDENTIFY_DISTINCT_ID = "server-group-identify"
 POSTHOG_ORGANIZATION_GROUP_TYPE = "organization"
-# Stable distinct_id for server-originated events with no acting user.
-# Group-linked events must stay identified (setting $process_person_profile
-# to False would unlink them from the organization group), so a single shared
-# "server" person absorbs them instead of minting one person per org.
-POSTHOG_SERVER_EVENT_DISTINCT_ID = "server"
 
 
 def get_posthog() -> Posthog | None:

api/tests/test_call_concurrency.py

@@ -70,10 +70,18 @@ async def test_acquire_org_slot_logs_warning_when_limit_reached():
 
 
 @pytest.mark.asyncio
-async def test_acquire_org_slot_fires_usage_event_when_limit_reached():
+async def test_acquire_org_slot_fires_usage_event_per_org_member_when_limit_reached():
+    """Mirrors the MPS org-event convention: one event per org member with the
+    member's provider_id as distinct_id, event_source property, no $groups."""
+    from types import SimpleNamespace
+
     from api.enums import PostHogEvent
 
     service = CallConcurrencyService()
+    members = [
+        SimpleNamespace(provider_id="user-a"),
+        SimpleNamespace(provider_id="user-b"),
+    ]
 
     with (
         patch("api.services.call_concurrency.db_client") as mock_db,
@@ -81,6 +89,7 @@ async def test_acquire_org_slot_fires_usage_event_when_limit_reached():
         patch("api.services.call_concurrency.capture_event") as mock_capture,
     ):
         mock_db.get_configuration = AsyncMock(return_value=None)
+        mock_db.get_organization_users = AsyncMock(return_value=members)
         mock_rate_limiter.try_acquire_concurrent_slot_details = AsyncMock(
             return_value=None
         )
@@ -89,16 +98,20 @@ async def test_acquire_org_slot_fires_usage_event_when_limit_reached():
         with pytest.raises(CallConcurrencyLimitError):
             await service.acquire_org_slot(199, source="webrtc", timeout=0)
 
-    mock_capture.assert_called_once()
-    kwargs = mock_capture.call_args.kwargs
-    assert kwargs["event"] == PostHogEvent.USAGE_CONCURRENT_CALL_LIMIT_REACHED
-    assert kwargs["distinct_id"] == "server"
-    assert kwargs["groups"] == {"organization": "199"}
-    assert kwargs["properties"]["organization_id"] == 199
-    assert kwargs["properties"]["source"] == "webrtc"
-    assert kwargs["properties"]["active_calls"] == 10
-    assert kwargs["properties"]["max_concurrent"] == 10
-    assert "scope_key" not in kwargs["properties"]
+    mock_db.get_organization_users.assert_awaited_once_with(199)
+    assert mock_capture.call_count == 2
+    distinct_ids = [c.kwargs["distinct_id"] for c in mock_capture.call_args_list]
+    assert distinct_ids == ["user-a", "user-b"]
+    for call in mock_capture.call_args_list:
+        kwargs = call.kwargs
+        assert kwargs["event"] == PostHogEvent.USAGE_CONCURRENT_CALL_LIMIT_REACHED
+        assert "groups" not in kwargs
+        assert kwargs["properties"]["event_source"] == "dograh"
+        assert kwargs["properties"]["organization_id"] == 199
+        assert kwargs["properties"]["source"] == "webrtc"
+        assert kwargs["properties"]["active_calls"] == 10
+        assert kwargs["properties"]["max_concurrent"] == 10
+        assert "scope_key" not in kwargs["properties"]
 
 
 @pytest.mark.asyncio