Merge branch 'main' into feat/vici-dial

api/services/telephony/ari_manager.py

@@ -26,6 +26,10 @@ from loguru import logger
 from api.constants import REDIS_URL
 from api.db import db_client
 from api.enums import CallType, WorkflowRunMode
+from api.services.call_concurrency import (
+    CallConcurrencyLimitError,
+    call_concurrency,
+)
 from api.services.quota_service import authorize_workflow_run_start
 from api.services.telephony.call_transfer_manager import get_call_transfer_manager
 from api.services.telephony.transfer_event_protocol import (
@@ -119,11 +123,20 @@ class ARIConnection:
         r = await self._get_redis()
         return await r.exists(f"{_EXT_CHANNEL_KEY_PREFIX}{channel_id}") > 0
 
-    async def _delete_ext_channel(self, channel_id: str):
+    async def _delete_ext_channel(self, channel_id: Optional[str]):
         """Remove the external media channel marker."""
+        if not channel_id:
+            return
         r = await self._get_redis()
         await r.delete(f"{_EXT_CHANNEL_KEY_PREFIX}{channel_id}")
 
+    async def _delete_transfer_channel_mapping(self, channel_id: Optional[str]):
+        """Remove transfer destination channel correlation marker."""
+        if not channel_id:
+            return
+        r = await self._get_redis()
+        await r.delete(f"ari:transfer_channel:{channel_id}")
+
     async def _set_pending_bridge(
         self,
         ext_channel_id: str,
@@ -340,19 +353,18 @@ class ARIConnection:
 
                 workflow_run_id = args_dict.get("workflow_run_id")
                 workflow_id = args_dict.get("workflow_id")
-                user_id = args_dict.get("user_id")
 
-                if not workflow_run_id or not workflow_id or not user_id:
+                if not workflow_run_id or not workflow_id:
                     logger.warning(
                         f"[ARI org={self.organization_id}] StasisStart missing required args: "
-                        f"workflow_run_id={workflow_run_id}, workflow_id={workflow_id}, user_id={user_id}"
+                        f"workflow_run_id={workflow_run_id}, workflow_id={workflow_id}"
                     )
                     return
 
                 # Start pipeline connection in background task
                 asyncio.create_task(
                     self._handle_stasis_start(
-                        channel_id, channel_state, workflow_run_id, workflow_id, user_id
+                        channel_id, channel_state, workflow_run_id, workflow_id
                     )
                 )
 
@@ -521,7 +533,6 @@ class ARIConnection:
     async def _create_external_media(
         self,
         workflow_id: str,
-        user_id: str,
         workflow_run_id: str,
         channel_id: Optional[str] = None,
     ) -> str:
@@ -537,10 +548,10 @@ class ARIConnection:
         the POST and avoid racing against the StasisStart event.
         """
         # v() appends URI query params to the websocket_client.conf URL
-        # e.g. wss://api.dograh.com/ws/ari?workflow_id=1&user_id=2&workflow_run_id=3
+        # e.g. wss://api.dograh.com/ws/ari?workflow_id=1&organization_id=2&workflow_run_id=3
         transport_data = (
             f"v(workflow_id={workflow_id},"
-            f"user_id={user_id},"
+            f"organization_id={self.organization_id},"
             f"workflow_run_id={workflow_run_id})"
         )
 
@@ -604,6 +615,8 @@ class ARIConnection:
         channel = event.get("channel", {})
         caller_number = channel.get("caller", {}).get("number", "unknown")
         called_number = channel.get("dialplan", {}).get("exten", "unknown")
+        concurrency_slot = None
+        workflow_run = None
 
         try:
             # 1. Resolve the workflow from the called extension via the
@@ -650,6 +663,20 @@ class ARIConnection:
 
             user_id = workflow.user_id
 
+            try:
+                concurrency_slot = await call_concurrency.acquire_org_slot(
+                    self.organization_id,
+                    source="ari_inbound",
+                    timeout=0,
+                )
+            except CallConcurrencyLimitError:
+                logger.warning(
+                    f"[ARI org={self.organization_id}] Concurrent call limit "
+                    f"reached; hanging up inbound channel {channel_id}"
+                )
+                await self._delete_channel(channel_id)
+                return
+
             # 3. Create workflow run
             call_id = channel_id
             # Capture the upstream-PBX (VICIdial) identity off the SIP headers so
@@ -675,7 +702,9 @@ class ARIConnection:
                 gathered_context={
                     "call_id": call_id,
                 },
+                organization_id=self.organization_id,
             )
+            await call_concurrency.bind_workflow_run(concurrency_slot, workflow_run.id)
 
             logger.info(
                 f"[ARI org={self.organization_id}] Created inbound workflow run "
@@ -687,6 +716,7 @@ class ARIConnection:
             # store the MPS correlation id before the pipeline starts.
             quota_result = await authorize_workflow_run_start(
                 workflow_id=inbound_workflow_id,
+                organization_id=self.organization_id,
                 workflow_run_id=workflow_run.id,
             )
             if not quota_result.has_quota:
@@ -694,6 +724,7 @@ class ARIConnection:
                     f"[ARI org={self.organization_id}] Quota exceeded for user {user_id} "
                     f"— hanging up inbound call {channel_id}"
                 )
+                await call_concurrency.release_workflow_run_slot(workflow_run.id)
                 await self._delete_channel(channel_id)
                 return
 
@@ -706,9 +737,12 @@ class ARIConnection:
                 channel_state,
                 str(workflow_run.id),
                 str(inbound_workflow_id),
-                str(user_id),
             )
         except Exception as e:
+            if workflow_run:
+                await call_concurrency.release_workflow_run_slot(workflow_run.id)
+            elif concurrency_slot:
+                await call_concurrency.release_slot(concurrency_slot)
             logger.error(
                 f"[ARI org={self.organization_id}] Error handling inbound StasisStart "
                 f"for channel {channel_id}: {e}"
@@ -724,7 +758,6 @@ class ARIConnection:
         channel_state: str,
         workflow_run_id: str,
         workflow_id: str,
-        user_id: str,
     ):
         """Set up external media for a caller channel that has entered Stasis.
 
@@ -768,7 +801,6 @@ class ARIConnection:
             # 3. Create the ext media channel with the id we just registered.
             created_id = await self._create_external_media(
                 workflow_id,
-                user_id,
                 workflow_run_id,
                 channel_id=ext_channel_id,
             )
@@ -844,6 +876,14 @@ class ARIConnection:
         the bridge and both channels — like endConferenceOnExit.
         """
         try:
+            # Release the org concurrency slot. Normally the pipeline's own
+            # teardown does this when the ext media websocket closes, but if
+            # the pipeline never started (caller hung up before external
+            # media connected, ext media creation failed, ...) this is the
+            # only cleanup that runs before the Redis stale timeout. No-op
+            # when the slot was already released.
+            await call_concurrency.unregister_active_call(int(workflow_run_id))
+
             workflow_run = await db_client.get_workflow_run_by_id(int(workflow_run_id))
             if not workflow_run or not workflow_run.gathered_context:
                 logger.warning(
@@ -859,6 +899,11 @@ class ARIConnection:
             ext_channel_id = ctx.get("ext_channel_id")
             bridge_id = ctx.get("bridge_id")
             transfer_state = ctx.get("transfer_state")
+            transfer_bridge_id = ctx.get("transfer_bridge_id") or bridge_id
+            transfer_caller_channel_id = (
+                ctx.get("transfer_caller_channel_id") or call_id
+            )
+            transfer_destination_channel_id = ctx.get("transfer_destination_channel_id")
 
             # Check if this is a call transfer scenario external channel. Skip full teardown if
             # transfer is in progress and this is the external media channel
@@ -889,6 +934,64 @@ class ARIConnection:
                 )
                 return
 
+            if (
+                transfer_state == "complete"
+                and transfer_bridge_id
+                and transfer_caller_channel_id
+                and transfer_destination_channel_id
+                and channel_id
+                in (
+                    transfer_caller_channel_id,
+                    transfer_destination_channel_id,
+                )
+            ):
+                peer_channel_id = (
+                    transfer_destination_channel_id
+                    if channel_id == transfer_caller_channel_id
+                    else transfer_caller_channel_id
+                )
+                logger.info(
+                    f"[ARI org={self.organization_id}] Completed transfer participant "
+                    f"{channel_id} left Stasis; tearing down peer {peer_channel_id} "
+                    f"and bridge {transfer_bridge_id}"
+                )
+
+                # Mark terminal state before issuing ARI deletes so duplicate
+                # StasisEnd events run through the idempotent normal cleanup path.
+                ctx["transfer_state"] = "terminated"
+                await db_client.update_workflow_run(
+                    run_id=int(workflow_run_id), gathered_context=ctx
+                )
+
+                await self._delete_bridge(transfer_bridge_id)
+                if peer_channel_id and peer_channel_id != channel_id:
+                    await self._delete_channel(peer_channel_id)
+
+                keys_to_delete = [
+                    cid
+                    for cid in (
+                        transfer_caller_channel_id,
+                        transfer_destination_channel_id,
+                        ext_channel_id,
+                        channel_id,
+                    )
+                    if cid
+                ]
+                if keys_to_delete:
+                    await self._delete_channel_run(*keys_to_delete)
+
+                await self._delete_ext_channel(ext_channel_id)
+                await self._delete_transfer_channel_mapping(
+                    transfer_destination_channel_id
+                )
+
+                logger.info(
+                    f"[ARI org={self.organization_id}] Completed transfer teardown "
+                    f"finished for channel={channel_id}, peer={peer_channel_id}, "
+                    f"bridge={transfer_bridge_id}"
+                )
+                return
+
             # Normal full teardown for non-transfer scenarios (transfer_state is None or not in-progress)
             # Delete the bridge first (removes channels from it)
             if bridge_id:
@@ -908,6 +1011,7 @@ class ARIConnection:
 
             # Clean up the Redis marker for external channel
             await self._delete_ext_channel(ext_channel_id)
+            await self._delete_transfer_channel_mapping(transfer_destination_channel_id)
 
             logger.info(
                 f"[ARI org={self.organization_id}] StasisEnd full teardown for "