call-center/services/asterisk_bridge_service/routes.py

from __future__ import annotations

from fastapi import HTTPException
from sqlalchemy import select

from services.shared.core import utc_now_iso
from services.shared.db import get_session
from services.shared.models import (
    EscalationOut,
    EscalationRequestIn,
    HealthResponse,
    VoiceAICallStateUpdateIn,
    VoiceAIHandoffRequestIn,
    VoiceAISummaryOut,
    VoiceCallBlindTransferIn,
    VoiceCallClaimIn,
    VoiceLiveCallOut,
)
from services.shared.sql_models import AsteriskCallLinkRow, AsteriskEventLogRow

from services.asterisk_bridge_service import call_control


def _bridge_app():
    import services.asterisk_bridge_service.app as bridge_app

    return bridge_app


def list_live_calls(include_ended: bool = False, limit: int = 100) -> list[VoiceLiveCallOut]:
    bridge = _bridge_app()
    session = get_session()
    try:
        stmt = select(AsteriskCallLinkRow).order_by(AsteriskCallLinkRow.id.desc())
        if not include_ended:
            stmt = stmt.where(AsteriskCallLinkRow.status != "ended")
        rows = session.execute(stmt).scalars().all()
        result: list[VoiceLiveCallOut] = []
        for row in rows[: max(limit, 1)]:
            if not row.interaction_id:
                continue
            if not include_ended and bridge._call_is_ended(row):
                continue
            result.append(bridge._live_call_to_out(session, row))
        return result
    finally:
        session.close()


def list_recent_calls(limit: int = 20) -> list[VoiceLiveCallOut]:
    bridge = _bridge_app()
    session = get_session()
    try:
        rows = session.execute(
            select(AsteriskCallLinkRow)
            .where(AsteriskCallLinkRow.status == "ended")
            .order_by(AsteriskCallLinkRow.id.desc())
            .limit(max(limit * 4, limit, 20))
        ).scalars().all()
        now = bridge.datetime.now(bridge.timezone.utc)
        window_seconds = bridge._recent_calls_window_seconds()
        result: list[tuple[bridge.datetime, VoiceLiveCallOut]] = []
        for row in rows:
            if not row.interaction_id:
                continue
            payload = bridge._live_call_to_out(session, row)
            transition_dt = bridge._parse_iso(payload.last_transition_at or row.ended_at or row.updated_at)
            if transition_dt is None:
                continue
            if (now - transition_dt).total_seconds() > window_seconds:
                continue
            result.append((transition_dt, payload))
        result.sort(key=lambda item: item[0], reverse=True)
        return [item for _, item in result[: max(limit, 1)]]
    finally:
        session.close()


def claim_live_call(call_id: str, body: VoiceCallClaimIn, actor: dict):
    return call_control.claim_live_call(call_id, body, actor)


def hangup_live_call(call_id: str, actor: dict):
    return call_control.hangup_live_call(call_id, actor)


def blind_transfer_live_call(call_id: str, body: VoiceCallBlindTransferIn, actor: dict):
    return call_control.blind_transfer_live_call(call_id, body, actor)


def list_live_call_actions(call_id: str, limit: int, actor: dict):
    return call_control.list_live_call_actions(call_id, limit, actor)


def get_live_call_ai_summary(call_id: str) -> VoiceAISummaryOut | None:
    bridge = _bridge_app()
    return bridge._voice_ai_summary_for_call(call_id)


def escalate_live_call(call_id: str, body: EscalationRequestIn, actor: dict) -> EscalationOut:
    bridge = _bridge_app()
    return bridge._create_escalation(call_id, body, actor)


def voice_ai_handoff_call(call_id: str, body: VoiceAIHandoffRequestIn, actor: dict) -> VoiceLiveCallOut:
    bridge = _bridge_app()
    return bridge._request_voice_ai_handoff(call_id, body, actor)


def voice_ai_update_call_state(call_id: str, body: VoiceAICallStateUpdateIn, actor: dict) -> VoiceLiveCallOut:
    bridge = _bridge_app()
    return bridge._update_voice_ai_call_state(call_id, body, actor)


def health() -> HealthResponse:
    return HealthResponse(status="ok", service="asterisk-bridge-service")


def asterisk_status():
    bridge = _bridge_app()
    snap = bridge._STATE.snapshot()
    status = "ok" if bridge._bridge_enabled() else "disabled"
    return bridge.AsteriskBridgeStatusOut(
        status=status,
        ami_connected=bool(snap["ami_connected"]),
        ami_host=bridge._ami_host() or None,
        last_event_at=snap["last_event_at"],
        queue_codes_loaded=sorted(bridge._queue_map().keys()),
        sftp_enabled=bridge._sftp_enabled(),
        bridge_auth_mode=bridge._bridge_auth_mode(),
        callcontrol_enabled=bridge._callcontrol_enabled(),
        webrtc_enabled=bridge._webrtc_enabled(),
        webrtc_ws_url=bridge._webrtc_ws_url() or None,
    )


def browser_softphone_config(actor: dict):
    bridge = _bridge_app()
    return bridge._browser_softphone_config_for_actor(actor)


def list_asterisk_events(status: str | None = None, limit: int = 100):
    bridge = _bridge_app()
    session = get_session()
    try:
        stmt = select(AsteriskEventLogRow).order_by(AsteriskEventLogRow.id.desc())
        if status:
            stmt = stmt.where(AsteriskEventLogRow.forward_status == status)
        rows = session.execute(stmt).scalars().all()
        return [bridge._to_out(row) for row in rows[:limit]]
    finally:
        session.close()


def get_asterisk_event(bridge_event_id: str):
    bridge = _bridge_app()
    session = get_session()
    try:
        row = session.execute(
            select(AsteriskEventLogRow)
            .where(AsteriskEventLogRow.bridge_event_id == bridge_event_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Asterisk bridge event not found")
        return bridge._to_out(row)
    finally:
        session.close()


def retry_asterisk_event(bridge_event_id: str):
    bridge = _bridge_app()
    if not bridge._claim_bridge_event_for_processing(bridge_event_id):
        session = get_session()
        try:
            row = bridge._load_bridge_event_row(session, bridge_event_id)
            if row is None:
                raise HTTPException(status_code=404, detail="Asterisk bridge event not found")
            return bridge._to_out(row)
        finally:
            session.close()

    row = bridge._process_claimed_bridge_event(bridge_event_id)
    if row is None:
        raise HTTPException(status_code=404, detail="Asterisk bridge event not found")
    return bridge._to_out(row)


def reconnect_asterisk():
    bridge = _bridge_app()
    bridge._STATE.request_reconnect()
    return asterisk_status()