diff --git a/api/enums.py b/api/enums.py index 3d43e1b8..4dda8441 100644 --- a/api/enums.py +++ b/api/enums.py @@ -200,3 +200,5 @@ class PostHogEvent(str, Enum): SIGNED_IN = "signed_in" ORGANIZATION_CREATED = "organization_created" ORGANIZATION_USER_ASSOCIATED = "organization_user_associated" + # usage_* events track orgs hitting capacity/limit boundaries + USAGE_CONCURRENT_CALL_LIMIT_REACHED = "usage_concurrent_call_limit_reached" diff --git a/api/services/auth/depends.py b/api/services/auth/depends.py index 488d45a4..52ee5a56 100644 --- a/api/services/auth/depends.py +++ b/api/services/auth/depends.py @@ -14,6 +14,7 @@ from api.services.auth.stack_auth import stackauth from api.services.configuration.registry import ServiceProviders from api.services.mps_billing import ensure_hosted_mps_billing_account_v2 from api.services.posthog_client import ( + POSTHOG_ORGANIZATION_GROUP_TYPE, capture_event, group_identify, set_person_properties, @@ -33,7 +34,6 @@ async def require_local_auth() -> None: raise HTTPException(status_code=404, detail="Not found") -POSTHOG_ORGANIZATION_GROUP_TYPE = "organization" POSTHOG_ORGANIZATION_USES_MPS_BILLING_V2_PROPERTY = "uses_mps_billing_v2" diff --git a/api/services/call_concurrency.py b/api/services/call_concurrency.py index b499a29d..ec482c87 100644 --- a/api/services/call_concurrency.py +++ b/api/services/call_concurrency.py @@ -6,8 +6,13 @@ from loguru import logger from api.constants import DEFAULT_ORG_CONCURRENCY_LIMIT from api.db import db_client -from api.enums import OrganizationConfigurationKey +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, +) @dataclass(frozen=True) @@ -128,6 +133,22 @@ class CallConcurrencyService: f"source={source}, active_calls={current_count}/{max_concurrent}" f"{scope_note}, waited={wait_time:.1f}s" ) + properties = { + "organization_id": organization_id, + "source": source, + "max_concurrent": max_concurrent, + "active_calls": current_count, + "waited_seconds": round(wait_time, 1), + } + 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)}, + ) raise CallConcurrencyLimitError( organization_id=organization_id, source=source, diff --git a/api/services/posthog_client.py b/api/services/posthog_client.py index 15e3a4ac..335fa1fb 100644 --- a/api/services/posthog_client.py +++ b/api/services/posthog_client.py @@ -7,6 +7,12 @@ 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 485e2741..43251450 100644 --- a/api/tests/test_call_concurrency.py +++ b/api/tests/test_call_concurrency.py @@ -69,6 +69,38 @@ async def test_acquire_org_slot_logs_warning_when_limit_reached(): assert "active_calls=12/10" in log_message +@pytest.mark.asyncio +async def test_acquire_org_slot_fires_usage_event_when_limit_reached(): + from api.enums import PostHogEvent + + service = CallConcurrencyService() + + with ( + patch("api.services.call_concurrency.db_client") as mock_db, + patch("api.services.call_concurrency.rate_limiter") as mock_rate_limiter, + patch("api.services.call_concurrency.capture_event") as mock_capture, + ): + mock_db.get_configuration = AsyncMock(return_value=None) + mock_rate_limiter.try_acquire_concurrent_slot_details = AsyncMock( + return_value=None + ) + mock_rate_limiter.get_concurrent_count = AsyncMock(return_value=10) + + 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"] + + @pytest.mark.asyncio async def test_acquire_org_slot_passes_scope_to_rate_limiter(): service = CallConcurrencyService()