From 41e887ec57f395aff7eda3e375c899c4369647a9 Mon Sep 17 00:00:00 2001 From: Abhishek Kumar Date: Thu, 9 Jul 2026 17:50:15 +0530 Subject: [PATCH] fix: align usage event with MPS org-event convention (per-member fan-out) Co-Authored-By: Claude Fable 5 --- api/db/organization_client.py | 17 +++++++++++ api/services/call_concurrency.py | 46 +++++++++++++++++++++++------- api/services/posthog_client.py | 5 ---- api/tests/test_call_concurrency.py | 35 ++++++++++++++++------- 4 files changed, 76 insertions(+), 27 deletions(-) diff --git a/api/db/organization_client.py b/api/db/organization_client.py index 6adc347a..ab4c290f 100644 --- a/api/db/organization_client.py +++ b/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]: diff --git a/api/services/call_concurrency.py b/api/services/call_concurrency.py index ec482c87..d3ad9f36 100644 --- a/api/services/call_concurrency.py +++ b/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: diff --git a/api/services/posthog_client.py b/api/services/posthog_client.py index 335fa1fb..610194e0 100644 --- a/api/services/posthog_client.py +++ b/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: diff --git a/api/tests/test_call_concurrency.py b/api/tests/test_call_concurrency.py index 43251450..eefb5f5b 100644 --- a/api/tests/test_call_concurrency.py +++ b/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