asterisk_shim_server.py

"""
src/servers/asterisk_shim_server.py

FastAPI server for the Asterisk shim backend. Intended to be run on the same machine as Asterisk server.

Bridges Asterisk ExternalMedia (RTP μ-law) to a single WebSocket connected to ECS at /voice/voice (internal ALB).
The shim forwards audio frames and simple call events between Asterisk and ECS.
ECS owns the AI voice agent session and recordings.

Topology:
- Asterisk PSTN channel ↔ ARI mixing bridge ↔ ExternalMedia channel
- ExternalMedia UDP/RTP ↔ shim (this process)
- Shim ↔ WSS to ECS /voice/voice (internal ALB)
- ECS ↔ AI voice agent server
"""

import asyncio

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from ari.asterisk_ari_supervisor import ARISupervisor
from env_config import EnvConfig
from utils.logger import get_logger


# Constants _______________________________________________________________________________________

app = FastAPI(
    title=__name__,
    version="1.0.0",
    debug=True,
    docs_url="/docs",
)

logger = get_logger(name=__name__)


# CORS ____________________________________________________________________________________________

# sample CORS (very permissive)
app.add_middleware(
    middleware_class=CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)


# Global supervisor instance ______________________________________________________________________

_supervisor: ARISupervisor | None = None
_sup_task: asyncio.Task | None = None


# Helper functions _______________________________________________________________________________)

async def _start_supervisor() -> None:
    """
    Start or restart the ARI supervisor.
    """
    global _supervisor, _sup_task

    # stop existing supervisor if running
    if _supervisor:
        await _supervisor.stop()
    if _sup_task and not _sup_task.done():
        _sup_task.cancel()
        try:
            await _sup_task
        except asyncio.CancelledError:
            pass

    # start new supervisor
    logger.info("Starting ARI supervisor")
    _supervisor = ARISupervisor()
    _sup_task = asyncio.create_task(
        coro=_supervisor.run(),
        name="ari-supervisor"
    )

    # monitor supervisor task and restart if it dies
    async def monitor_supervisor() -> None:
        """
        Monitor the supervisor task and restart it if it dies.
        """
        try:
            while True:
                if _sup_task and _sup_task.done():
                    try:
                        exception = _sup_task.exception()
                        if exception:
                            logger.error("ARI supervisor died with exception: %s", exception)
                        else:
                            logger.warning("ARI supervisor exited unexpectedly")
                    except asyncio.CancelledError:
                        logger.warning("ARI supervisor was cancelled")

                    logger.info("Restarting ARI supervisor in 2 seconds...")
                    await asyncio.sleep(2.0)
                    await _start_supervisor()
                    break
                await asyncio.sleep(1.0)
        except Exception as e:
            logger.error("Error in supervisor monitor: %s", e, exc_info=True)

    # create and start the supervisor monitor task
    asyncio.create_task(
        coro=monitor_supervisor(),
        name="supervisor-monitor"
    )


# Startup/Shutdown ________________________________________________________________________________

@app.on_event("startup")
async def startup_event() -> None:
    """
    Initialize and start the ARI supervisor background task.
    """
    logger.info("Starting ARI supervisor")
    await _start_supervisor()


@app.on_event("shutdown")
async def shutdown_event() -> None:
    """
    Cancel the ARI supervisor task on application shutdown.
    """
    logger.info("Shutting down ARI supervisor on application shutdown")
    if _supervisor:
        await _supervisor.stop()
    if _sup_task:
        _sup_task.cancel()
        try:
            await _sup_task
        except asyncio.CancelledError:
            pass


# HTTP endpoints __________________________________________________________________________________

@app.get("/health")
async def health() -> JSONResponse:
    """
    Return health and number of active call sessions.
    Returns:
        JSONResponse: A JSON response containing the health and number of active call sessions.
    """
    logger.debug("Health endpoint called")
    if not _supervisor:
        logger.error("Supervisor not initialized")
        return JSONResponse({
            "status": "error",
            "message": "Supervisor not initialized"
        })

    status = _supervisor.get_status()
    task_status = "unknown"
    if _sup_task:
        if _sup_task.cancelled():
            task_status = "cancelled"
        elif _sup_task.done():
            try:
                exception = _sup_task.exception()
                if exception:
                    task_status = f"failed: {exception}"
                else:
                    task_status = "completed"
            except asyncio.CancelledError:
                task_status = "cancelled"
        else:
            task_status = "running"

    # return the health and number of active call sessions
    return JSONResponse({
        "status": "ok",
        "supervisor_task_status": task_status,
        "config": {
            "ari_base": EnvConfig.ARI_BASE,
            "ari_app": EnvConfig.ARI_APP,
            "external_media_host": EnvConfig.EXTERNAL_MEDIA_HOST,
            "ecs_media_wss_url": EnvConfig.ECS_MEDIA_WSS_URL,
        },
        **status
    })


@app.post("/reconnect")
async def force_reconnect() -> JSONResponse:
    """
    Force a reconnection to ARI (for debugging).
    Returns:
        JSONResponse: A JSON response containing the status and message.
    """
    if not _supervisor:
        logger.error("Supervisor not initialized")
        raise HTTPException(
            status_code=500,
            detail="Supervisor not initialized"
        )

    logger.info("Supervisor reconnection triggered")
    await _supervisor.force_reconnect()
    logger.info("Supervisor reconnection completed")
    return JSONResponse({
        "status": "ok",
        "message": "Reconnection triggered"
    })


@app.post("/calls/{call_id}/hangup")
async def hangup_call(call_id: str) -> JSONResponse:
    """
    Hang up a call by deleting the PSTN channel via ARI.
    Args:
        call_id: The ID of the call to hang up.
    Returns:
        JSONResponse: A JSON response containing the status and message.
    """
    if not _supervisor or not _supervisor._http:
        logger.error("Supervisor not initialized")
        raise HTTPException(
            status_code=500,
            detail="Supervisor not initialized"
        )
    try:
        async with _supervisor._http.delete(
            url=f"{EnvConfig.ARI_BASE}/channels/{call_id}"
        ) as r:
            if r.status == 404:
                logger.error("Channel not found")
                raise HTTPException(
                    status_code=404,
                    detail="Channel not found"
                )
            r.raise_for_status()

        logger.info("Hung up call %s via ARI", call_id)
        return JSONResponse({
            "status": "ok",
            "message": f"Call {call_id} hung up"
        })
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Error hanging up call %s: %s", call_id, e)
        raise HTTPException(
            status_code=500,
            detail=f"Error hanging up call: {e}"
        )


@app.post("/calls/originate")
async def originate_call(
    endpoint: str,
    context: str = "default",
    extension: str = "s",
    priority: int = 1,
    timeout: int = 30,
) -> JSONResponse:
    """
    Originate a new call via ARI (optional endpoint).
    Args:
        endpoint: The endpoint to originate the call to.
        context: The context to originate the call to.
        extension: The extension to originate the call to.
        priority: The priority to originate the call to.
        timeout: The timeout to originate the call to.
    Returns:
        JSONResponse: A JSON response containing the status, call_id, and message.
    """
    if not _supervisor or not _supervisor._http:
        logger.error("Supervisor not initialized")
        raise HTTPException(
            status_code=500,
            detail="Supervisor not initialized"
        )

    try:
        params = {
            "endpoint": endpoint,
            "context": context,
            "extension": extension,
            "priority": priority,
            "timeout": timeout,
            "app": EnvConfig.ARI_APP,
        }

        async with _supervisor._http.post(
            url=f"{EnvConfig.ARI_BASE}/channels",
            params=params
        ) as r:
            r.raise_for_status()
            channel = await r.json()

        call_id = channel["id"]
        logger.info("Originated call %s to %s", call_id, endpoint)
        return JSONResponse({
            "status": "ok", 
            "call_id": call_id,
            "message": f"Call originated to {endpoint}"
        })
    except Exception as e:
        logger.error("Error originating call to %s: %s", endpoint, e)
        raise HTTPException(
            status_code=500,
            detail=f"Error originating call: {e}"
        )