urmi-ai/backend/app/voice/call_service.py

import secrets
import uuid
from datetime import datetime, timezone

from sqlalchemy.orm import Session

from ..config import settings
from ..models import PhoneNumber, UserAccount, VoiceCall
from .bonvoice_client import BonvoiceError, initiate_outbound_call
from .public_url import unified_voicebot_url
from .session_store import delete_session, get_session, save_session, update_session
from .validators import format_bonvoice_destination, require_owned_did

TERMINAL_STATUSES = frozenset({"ended", "failed", "no_answer"})


def media_url_for_session(session_token: str) -> str:
    return unified_voicebot_url()


def voice_media_ws_url() -> str:
    return unified_voicebot_url()


def serialize_call(call: VoiceCall) -> dict:
    return {
        "id": str(call.id),
        "direction": call.direction,
        "from": call.from_did,
        "to": call.to_number,
        "status": call.status,
        "stream_id": call.stream_id,
        "call_id": call.call_id,
        "bridge_mode": call.bridge_mode,
        "upstream": call.upstream_ws,
        "error_message": call.error_message,
        "answered_at": call.answered_at,
        "ended_at": call.ended_at,
        "created_at": call.created_at,
    }


async def create_outbound_call(
    db: Session,
    *,
    user: UserAccount,
    from_value: str,
    to_value: str,
    upstream: str | None,
    bridge_mode: str,
    metadata: dict | None = None,
) -> VoiceCall:
    owned_number = require_owned_did(db, user, from_value)
    destination = format_bonvoice_destination(to_value)
    if len(destination) < 10:
        raise ValueError("to must be a valid phone number")

    session_token = secrets.token_urlsafe(24)
    provider_event_id = uuid.uuid4().hex.upper()
    call = VoiceCall(
        user_id=user.id,
        phone_number_id=owned_number.id,
        direction="outbound",
        from_did=owned_number.phone_number,
        to_number=destination,
        status="initiated",
        provider_event_id=provider_event_id,
        session_token=session_token,
        bridge_mode=bridge_mode,
        upstream_ws=upstream,
    )
    db.add(call)
    db.commit()
    db.refresh(call)

    await save_session(
        session_token,
        {
            "call_id": str(call.id),
            "user_id": str(user.id),
            "direction": "outbound",
            "from": owned_number.phone_number,
            "to": destination,
            "bridge_mode": bridge_mode,
            "upstream_ws": upstream,
            "status": "pending",
            "metadata": metadata or {},
        },
    )

    voicebot_url = unified_voicebot_url()
    try:
        await initiate_outbound_call(
            destination=destination,
            leg_a_caller_id=owned_number.phone_number,
            voicebot_url=voicebot_url,
            event_id=provider_event_id,
        )
    except BonvoiceError as exc:
        call.status = "failed"
        call.error_message = str(exc)
        call.ended_at = datetime.now(timezone.utc)
        db.commit()
        raise

    return call


def create_inbound_call_record(
    db: Session,
    *,
    phone_number: PhoneNumber,
    caller: str,
    upstream: str,
    stream_id: str | None,
    provider_call_id: str | None,
) -> VoiceCall:
    session_token = secrets.token_urlsafe(24)
    call = VoiceCall(
        user_id=phone_number.user_id,
        phone_number_id=phone_number.id,
        direction="inbound",
        from_did=phone_number.phone_number,
        to_number=caller,
        status="active",
        session_token=session_token,
        stream_id=stream_id,
        call_id=provider_call_id,
        bridge_mode="upstream",
        upstream_ws=upstream,
        answered_at=datetime.now(timezone.utc),
    )
    db.add(call)
    db.commit()
    db.refresh(call)
    return call


def finalize_call(db: Session, call: VoiceCall | None, *, status: str = "ended") -> None:
    if not call:
        return
    db.refresh(call)
    if call.status in TERMINAL_STATUSES:
        return
    call.status = status
    if not call.ended_at:
        call.ended_at = datetime.now(timezone.utc)
    db.commit()


async def finalize_call_async(db: Session, call: VoiceCall | None, *, status: str = "ended") -> None:
    if not call:
        return
    finalize_call(db, call, status=status)
    if call.session_token:
        await delete_session(call.session_token)


async def refresh_call_lifecycle(db: Session, call: VoiceCall) -> VoiceCall:
    if call.status in TERMINAL_STATUSES:
        return call

    now = datetime.now(timezone.utc)
    age_seconds = (now - call.created_at).total_seconds()

    if call.status == "initiated":
        if age_seconds >= 45:
            call.status = "no_answer"
            call.ended_at = now
            if not call.error_message:
                call.error_message = "Call ended before media session connected"
            db.commit()
            db.refresh(call)
            await delete_session(call.session_token)

    if call.status == "active":
        session = await get_session(call.session_token)
        if not session:
            call.status = "ended"
            call.ended_at = now
            db.commit()
            db.refresh(call)
        elif session.get("status") == "ended":
            call.status = "ended"
            if not call.ended_at:
                call.ended_at = now
            db.commit()
            db.refresh(call)
            await delete_session(call.session_token)
        elif age_seconds >= 600:
            call.status = "ended"
            call.ended_at = now
            if not call.error_message:
                call.error_message = "Call exceeded maximum duration"
            db.commit()
            db.refresh(call)
            await delete_session(call.session_token)

    return call