call-center/services/asterisk_bridge_service/app.py

from __future__ import annotations

import os
from pathlib import Path
import threading
import time
from typing import Any, Callable
from datetime import datetime, timezone

import httpx
from fastapi import Depends, FastAPI

from services.shared.core import Role, utc_now_iso
from services.shared.db import DATABASE_URL, engine
from services.shared.models import (
    AsteriskBridgeStatusOut,
    AsteriskEventOut,
    BrowserSoftphoneConfigOut,
    EscalationOut,
    EscalationRequestIn,
    HealthResponse,
    VoiceAICallStateUpdateIn,
    VoiceAIHandoffRequestIn,
    VoiceAISummaryOut,
    VoiceCallActionOut,
    VoiceCallBlindTransferIn,
    VoiceCallClaimIn,
    VoiceLiveCallOut,
)
from services.shared.security import get_actor, issue_app_token, require_roles
from services.shared.sql_init import init_sql_schema
from services.asterisk_bridge_service import ami as bridge_ami
from services.asterisk_bridge_service import bridge_processing
from services.asterisk_bridge_service import call_control as bridge_call_control
from services.asterisk_bridge_service import config as bridge_config
from services.asterisk_bridge_service import http_client as bridge_http_client
from services.asterisk_bridge_service import ivr_fastagi as bridge_ivr_fastagi
from services.asterisk_bridge_service import presenters as bridge_presenters
from services.asterisk_bridge_service import recording_io as bridge_recording_io
from services.asterisk_bridge_service import reconcile as bridge_reconcile
from services.asterisk_bridge_service import routes as bridge_routes
from services.asterisk_bridge_service import runtime as bridge_runtime
from services.asterisk_bridge_service import voice_ai as bridge_voice_ai

app = FastAPI(title="asterisk-bridge-service", version="1.0.0")

init_sql_schema()


_bool_env = bridge_config.bool_env
_bridge_enabled = bridge_config.bridge_enabled
_ami_host = bridge_config.ami_host
_ami_port = bridge_config.ami_port
_ami_username = bridge_config.ami_username
_ami_secret = bridge_config.ami_secret
_ami_prefix = bridge_config.ami_prefix
_poll_interval = bridge_config.poll_interval
_recent_calls_window_seconds = bridge_config.recent_calls_window_seconds
_failed_retry_enabled = bridge_config.failed_retry_enabled
_failed_retry_interval_seconds = bridge_config.failed_retry_interval_seconds
_failed_retry_batch_size = bridge_config.failed_retry_batch_size
_reconcile_enabled = bridge_config.reconcile_enabled
_reconcile_stale_seconds = bridge_config.reconcile_stale_seconds
_reconcile_settle_seconds = bridge_config.reconcile_settle_seconds
_reconcile_scan_limit = bridge_config.reconcile_scan_limit
_forward_timeout_seconds = bridge_config.forward_timeout_seconds
_forward_max_attempts = bridge_config.forward_max_attempts
_forward_retry_backoff_seconds = bridge_config.forward_retry_backoff_seconds
_queue_map = bridge_config.queue_map
_sftp_enabled = bridge_config.sftp_enabled
_sftp_host = bridge_config.sftp_host
_sftp_port = bridge_config.sftp_port
_sftp_username = bridge_config.sftp_username
_sftp_password = bridge_config.sftp_password
_interaction_service_url = bridge_config.interaction_service_url
_voice_adapter_service_url = bridge_config.voice_adapter_service_url
_recording_service_url = bridge_config.recording_service_url
_ivr_service_url = bridge_config.ivr_service_url
_ai_voice_runtime_service_url = bridge_config.ai_voice_runtime_service_url
_ai_voice_enabled = bridge_config.ai_voice_enabled
_ai_voice_queue_config = bridge_config.ai_voice_queue_config
_ai_voice_config_for_queue = bridge_config.ai_voice_config_for_queue
_ivr_fastagi_enabled = bridge_config.ivr_fastagi_enabled
_ivr_fastagi_host = bridge_config.ivr_fastagi_host
_ivr_fastagi_port = bridge_config.ivr_fastagi_port
_ivr_dtmf_timeout_seconds = bridge_config.ivr_dtmf_timeout_seconds
_ivr_max_no_input_retries = bridge_config.ivr_max_no_input_retries
_ivr_max_invalid_retries = bridge_config.ivr_max_invalid_retries
_ivr_call_link_wait_seconds = bridge_config.ivr_call_link_wait_seconds
_callcontrol_enabled = bridge_config.callcontrol_enabled
_callcontrol_action_timeout_seconds = bridge_config.callcontrol_action_timeout_seconds
_webrtc_enabled = bridge_config.webrtc_enabled
_webrtc_ws_url = bridge_config.webrtc_ws_url
_webrtc_ice_servers = bridge_config.webrtc_ice_servers
_operator_extension_map = bridge_config.operator_extension_map
_user_for_operator_extension = bridge_config.user_for_operator_extension
_browser_sip_map = bridge_config.browser_sip_map
_default_browser_sip_domain = bridge_config.default_browser_sip_domain
_browser_softphone_config_for_actor = bridge_config.browser_softphone_config_for_actor
_transfer_target_map = bridge_config.transfer_target_map
_claim_context = bridge_config.claim_context
_transfer_context = bridge_config.transfer_context
_bridge_auth_mode = bridge_config.bridge_auth_mode
_bridge_auth_fallback_legacy = bridge_config.bridge_auth_fallback_legacy
_bridge_auth_user = bridge_config.bridge_auth_user
_bridge_auth_role = bridge_config.bridge_auth_role
_bridge_auth_subject = bridge_config.bridge_auth_subject
_bridge_auth_token_ttl_seconds = bridge_config.bridge_auth_token_ttl_seconds
_ai_voice_runtime_trusted_subjects = bridge_config.ai_voice_runtime_trusted_subjects
_routing_service_url = bridge_config.routing_service_url
_routing_level_for_queue_code = bridge_config.routing_level_for_queue_code
_routing_tenant_for_queue_code = bridge_config.routing_tenant_for_queue_code


def _legacy_headers() -> dict[str, str]:
    return bridge_http_client.legacy_headers(
        bridge_auth_user=_bridge_auth_user(),
        bridge_auth_role=_bridge_auth_role(),
    )


def _bearer_headers() -> dict[str, str]:
    return bridge_http_client.bearer_headers(
        issue_token=lambda: issue_app_token(
            subject=_bridge_auth_subject(),
            username=_bridge_auth_user(),
            role=_bridge_auth_role(),
            auth_source="service",
            provider="asterisk-bridge",
            ttl_seconds=_bridge_auth_token_ttl_seconds(),
        )
    )


def _request_with_bridge_auth(
    client: httpx.Client,
    *,
    method: str,
    url: str,
    retry_reset: Callable[[], None] | None = None,
    **kwargs: Any,
) -> httpx.Response:
    return bridge_http_client.request_with_bridge_auth(
        client=client,
        httpx_module=httpx,
        method=method,
        url=url,
        bridge_auth_mode=_bridge_auth_mode(),
        bridge_auth_fallback_legacy=_bridge_auth_fallback_legacy(),
        legacy_headers_payload=_legacy_headers(),
        bearer_headers_payload=_bearer_headers(),
        retry_reset=retry_reset,
        **kwargs,
    )


def _request_json(
    method: str,
    url: str,
    payload: dict[str, Any],
    *,
    timeout_seconds: float | None = None,
    max_attempts: int | None = None,
    retry_backoff_seconds: float | None = None,
) -> dict[str, Any]:
    return bridge_http_client.request_json(
        httpx_module=httpx,
        send_request=_request_with_bridge_auth,
        method=method,
        url=url,
        payload=payload,
        timeout_seconds=timeout_seconds,
        max_attempts=max_attempts,
        retry_backoff_seconds=retry_backoff_seconds,
        forward_timeout_seconds=_forward_timeout_seconds,
        forward_max_attempts=_forward_max_attempts,
        forward_retry_backoff_seconds=_forward_retry_backoff_seconds,
        sleep=time.sleep,
    )


def _post_json(url: str, payload: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
    return _request_json("POST", url, payload, **kwargs)


def _patch_json(url: str, payload: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
    return _request_json("PATCH", url, payload, **kwargs)


_BridgeState = bridge_runtime.BridgeState
_STATE = _BridgeState()
_BRIDGE_SINGLETON_LOCK_KEY = 417_352_001
_BRIDGE_SINGLETON_LOCK_MESSAGE = (
    "Asterisk bridge background workers require singleton deployment "
    "(replicas=1, workers=1)"
)
_BRIDGE_SINGLETON_DB_CONNECTION = None
_BRIDGE_SINGLETON_FILE_HANDLE = None


def _bridge_singleton_lock_path() -> Path:
    return bridge_runtime.bridge_singleton_lock_path(
        database_url=DATABASE_URL,
        cc_data_dir=os.getenv("CC_DATA_DIR", ".data"),
    )


def _try_lock_file(handle) -> bool:
    return bridge_runtime.try_lock_file(handle)


def _unlock_file(handle) -> None:
    bridge_runtime.unlock_file(handle)


def _try_acquire_bridge_singleton_guard() -> bool:
    global _BRIDGE_SINGLETON_DB_CONNECTION, _BRIDGE_SINGLETON_FILE_HANDLE

    acquired = bridge_runtime.try_acquire_bridge_singleton_guard(
        database_url=DATABASE_URL,
        engine=engine,
        lock_key=_BRIDGE_SINGLETON_LOCK_KEY,
        cc_data_dir=os.getenv("CC_DATA_DIR", ".data"),
    )
    _BRIDGE_SINGLETON_DB_CONNECTION = bridge_runtime.bridge_singleton_db_connection()
    _BRIDGE_SINGLETON_FILE_HANDLE = bridge_runtime.bridge_singleton_file_handle()
    return acquired


def _ensure_bridge_singleton_guard() -> None:
    if not _bridge_enabled():
        return
    if not _try_acquire_bridge_singleton_guard():
        raise RuntimeError(_BRIDGE_SINGLETON_LOCK_MESSAGE)


def _release_bridge_singleton_guard() -> None:
    global _BRIDGE_SINGLETON_DB_CONNECTION, _BRIDGE_SINGLETON_FILE_HANDLE

    bridge_runtime.release_bridge_singleton_guard(lock_key=_BRIDGE_SINGLETON_LOCK_KEY)
    _BRIDGE_SINGLETON_DB_CONNECTION = None
    _BRIDGE_SINGLETON_FILE_HANDLE = None


def _background_threads_alive() -> list[threading.Thread]:
    return bridge_runtime.background_threads_alive()


def _background_stop_event() -> threading.Event:
    return bridge_runtime.background_stop_event()


def _background_shutdown_timeout_seconds() -> float:
    return bridge_runtime.background_shutdown_timeout_seconds(
        poll_interval=_poll_interval(),
        failed_retry_interval_seconds=_failed_retry_interval_seconds(),
    )


def _start_background_threads() -> None:
    try:
        extra_loops: list[tuple[str, Callable[..., Any]]] = []
        if _ivr_fastagi_enabled():
            extra_loops.append(("asterisk-ivr-fastagi-loop", _ivr_fastagi_loop))
        bridge_runtime.start_background_threads(
            ami_loop=_ami_loop,
            failed_retry_loop=_failed_retry_loop,
            ensure_guard=_ensure_bridge_singleton_guard,
            extra_loops=extra_loops,
        )
    except Exception:
        _release_bridge_singleton_guard()
        raise


def _stop_background_threads() -> None:
    bridge_runtime.stop_background_threads(
        state=_STATE,
        release_guard=_release_bridge_singleton_guard,
        shutdown_timeout_seconds=_background_shutdown_timeout_seconds(),
    )


# Final compatibility re-exports. Keep these in app.py so tests and extracted
# modules still share the same compatibility surface during the transition.
_to_out = bridge_presenters.to_out
_action_to_out = bridge_presenters.action_to_out
_safe_json_loads = bridge_presenters.safe_json_loads
_latest_voice_event_row = bridge_presenters.latest_voice_event_row
_latest_successful_action_row = bridge_presenters.latest_successful_action_row
_latest_iso = bridge_presenters.latest_iso
_hangup_cause_for_call = bridge_presenters.hangup_cause_for_call
_terminal_action_for_call = bridge_presenters.terminal_action_for_call
_live_call_to_out = bridge_presenters.live_call_to_out

_append_interaction_timeline = bridge_voice_ai.append_interaction_timeline
_start_voice_ai_session = bridge_voice_ai.start_voice_ai_session
_notify_voice_ai_telephony_event = bridge_voice_ai.notify_voice_ai_telephony_event
_register_voice_ai_media_bridge = bridge_voice_ai.register_voice_ai_media_bridge
_trusted_voice_runtime_actor = bridge_voice_ai.trusted_voice_runtime_actor
_assert_trusted_voice_runtime_actor = bridge_voice_ai.assert_trusted_voice_runtime_actor
_queue_code_for_queue_id = bridge_voice_ai._queue_code_for_queue_id
_request_voice_ai_handoff = bridge_voice_ai.request_handoff
_update_voice_ai_call_state = bridge_voice_ai.update_call_ai_state
_voice_ai_summary_for_call = bridge_voice_ai.voice_ai_summary_for_call
_create_escalation = bridge_voice_ai.create_escalation
_release_routing_agent = bridge_voice_ai.release_routing_agent

_first_non_empty = bridge_ami.first_non_empty
_extract_call_id = bridge_ami.extract_call_id
_extract_linked_id = bridge_ami.extract_linked_id
_resolve_channel_from_payload = bridge_ami.resolve_channel_from_payload
_list_channels_via_coreshowchannels = bridge_ami.list_channels_via_coreshowchannels
_resolve_channel_via_coreshowchannels = bridge_ami.resolve_channel_via_coreshowchannels
_extract_extension_from_channel = bridge_ami.extract_extension_from_channel
_operator_channel_for_extension = bridge_ami.operator_channel_for_extension
_CallChannelSnapshot = bridge_ami.CallChannelSnapshot
_latest_event_channel = bridge_ami.latest_event_channel
_build_call_channel_snapshot = bridge_ami.build_call_channel_snapshot
_candidate_channels_for_call = bridge_ami.candidate_channels_for_call
_resolve_channel_name = bridge_ami.resolve_channel_name
_ami_action = bridge_ami.ami_action
_read_ami_frame = bridge_ami.read_ami_frame
_send_ami_action = bridge_ami.send_ami_action
_ami_login = bridge_ami.ami_login
_ami_loop = bridge_ami.ami_loop

_fetch_recording_file = bridge_recording_io.fetch_recording_file
_recording_search_directories = bridge_recording_io.recording_search_directories
_guess_mime_type = bridge_recording_io.guess_mime_type
_find_remote_recording_for_call = bridge_recording_io.find_remote_recording_for_call

_AgiRequest = bridge_ivr_fastagi.AgiRequest
_AgiHangup = bridge_ivr_fastagi.AgiHangup
_SimpleAgiChannel = bridge_ivr_fastagi.SimpleAgiChannel
_parse_agi_request = bridge_ivr_fastagi.parse_agi_request
_run_ivr_fastagi_call = bridge_ivr_fastagi.run_ivr_call
_ivr_fastagi_loop = bridge_ivr_fastagi.fastagi_loop

_parse_iso = bridge_reconcile.parse_iso
_has_voice_event = bridge_reconcile.has_voice_event
_has_recording = bridge_reconcile.has_recording
_latest_recording_for_call = bridge_reconcile.latest_recording_for_call
_reconcile_stale_calls_once = bridge_reconcile.reconcile_stale_calls_once

_create_bridge_log = bridge_processing.create_bridge_log
_mark_log_forwarded = bridge_processing.mark_log_forwarded
_mark_log_failed = bridge_processing.mark_log_failed
_stable_source_event_id = bridge_processing.stable_source_event_id
_bridge_source_event_id = bridge_processing.bridge_source_event_id
_reconcile_source_event_id = bridge_processing.reconcile_source_event_id
_load_bridge_event_row = bridge_processing.load_bridge_event_row
_claim_bridge_event_for_processing = bridge_processing.claim_bridge_event_for_processing
_process_claimed_bridge_event = bridge_processing.process_claimed_bridge_event
_create_action_log = bridge_processing.create_action_log
_set_action_result = bridge_processing.set_action_result
_find_call_link = bridge_processing.find_call_link
_find_interaction_in_db_by_call_id = bridge_processing.find_interaction_in_db_by_call_id
_recover_call_link_from_started_event = bridge_processing.recover_call_link_from_started_event
_find_forwarded_log = bridge_processing.find_forwarded_log
_create_interaction = bridge_processing.create_interaction
_find_interaction_by_call_id = bridge_processing.find_interaction_by_call_id
_resolve_started_interaction_id = bridge_processing.resolve_started_interaction_id
_callcontrol_side_effect_timeout_seconds = bridge_processing.callcontrol_side_effect_timeout_seconds
_callcontrol_side_effect_max_attempts = bridge_processing.callcontrol_side_effect_max_attempts
_emit_voice_event = bridge_processing.emit_voice_event
_assign_interaction = bridge_processing.assign_interaction
_upload_recording = bridge_processing.upload_recording
_normalize_duration = bridge_processing.normalize_duration
_process_call_started = bridge_processing.process_call_started
_process_audio_bridge_requested = bridge_processing.process_audio_bridge_requested
_process_audio_bridge_ended = bridge_processing.process_audio_bridge_ended
_process_call_ended = bridge_processing.process_call_ended
_process_operator_connected = bridge_processing.process_operator_connected
_process_recording_ready = bridge_processing.process_recording_ready
_process_bridge_row = bridge_processing.process_bridge_row
_record_ami_payload = bridge_processing.record_ami_payload
_retry_failed_events_once = bridge_processing.retry_failed_events_once
_failed_retry_loop = bridge_processing.failed_retry_loop

_load_live_call_or_404 = bridge_call_control.load_live_call_or_404
_call_is_ended = bridge_call_control.call_is_ended
_resolve_operator_extension = bridge_call_control.resolve_operator_extension
_resolve_transfer_extension = bridge_call_control.resolve_transfer_extension
_assert_callcontrol_enabled = bridge_call_control.assert_callcontrol_enabled
_assert_call_control_permissions = bridge_call_control.assert_call_control_permissions


@app.get("/asterisk/live-calls", response_model=list[VoiceLiveCallOut])
def list_live_calls(
    include_ended: bool = False,
    limit: int = 100,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> list[VoiceLiveCallOut]:
    return bridge_routes.list_live_calls(include_ended=include_ended, limit=limit)


@app.get("/asterisk/recent-calls", response_model=list[VoiceLiveCallOut])
def list_recent_calls(
    limit: int = 20,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> list[VoiceLiveCallOut]:
    return bridge_routes.list_recent_calls(limit=limit)


@app.post("/asterisk/live-calls/{call_id}/claim", response_model=VoiceLiveCallOut)
def claim_live_call(
    call_id: str,
    body: VoiceCallClaimIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> VoiceLiveCallOut:
    return bridge_routes.claim_live_call(call_id=call_id, body=body, actor=actor)


@app.post("/asterisk/live-calls/{call_id}/hangup", response_model=VoiceLiveCallOut)
def hangup_live_call(
    call_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> VoiceLiveCallOut:
    return bridge_routes.hangup_live_call(call_id=call_id, actor=actor)


@app.post("/asterisk/live-calls/{call_id}/blind-transfer", response_model=VoiceLiveCallOut)
def blind_transfer_live_call(
    call_id: str,
    body: VoiceCallBlindTransferIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> VoiceLiveCallOut:
    return bridge_routes.blind_transfer_live_call(call_id=call_id, body=body, actor=actor)


@app.post("/asterisk/live-calls/{call_id}/escalations", response_model=EscalationOut)
def escalate_live_call(
    call_id: str,
    body: EscalationRequestIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> EscalationOut:
    return bridge_routes.escalate_live_call(call_id=call_id, body=body, actor=actor)


@app.get("/asterisk/live-calls/{call_id}/actions", response_model=list[VoiceCallActionOut])
def list_live_call_actions(
    call_id: str,
    limit: int = 100,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> list[VoiceCallActionOut]:
    return bridge_routes.list_live_call_actions(call_id=call_id, limit=limit, actor=actor)


@app.get("/asterisk/live-calls/{call_id}/ai-summary", response_model=VoiceAISummaryOut | None)
def get_live_call_ai_summary(
    call_id: str,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> VoiceAISummaryOut | None:
    return bridge_routes.get_live_call_ai_summary(call_id)


@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
    return bridge_routes.health()


@app.get("/asterisk/status", response_model=AsteriskBridgeStatusOut)
def asterisk_status(_: dict = Depends(require_roles(Role.ADMIN))) -> AsteriskBridgeStatusOut:
    return bridge_routes.asterisk_status()


@app.get("/asterisk/browser-softphone/config", response_model=BrowserSoftphoneConfigOut)
def browser_softphone_config(
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> BrowserSoftphoneConfigOut:
    return bridge_routes.browser_softphone_config(actor)


@app.get("/asterisk/events", response_model=list[AsteriskEventOut])
def list_asterisk_events(
    status: str | None = None,
    limit: int = 100,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> list[AsteriskEventOut]:
    return bridge_routes.list_asterisk_events(status=status, limit=limit)


@app.get("/asterisk/events/{bridge_event_id}", response_model=AsteriskEventOut)
def get_asterisk_event(
    bridge_event_id: str,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> AsteriskEventOut:
    return bridge_routes.get_asterisk_event(bridge_event_id)


@app.post("/asterisk/events/{bridge_event_id}/retry", response_model=AsteriskEventOut)
def retry_asterisk_event(
    bridge_event_id: str,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> AsteriskEventOut:
    return bridge_routes.retry_asterisk_event(bridge_event_id)


@app.post("/internal/voice-ai/calls/{call_id}/handoff", response_model=VoiceLiveCallOut)
def voice_ai_handoff_call(
    call_id: str,
    body: VoiceAIHandoffRequestIn,
    actor: dict = Depends(get_actor),
) -> VoiceLiveCallOut:
    return bridge_routes.voice_ai_handoff_call(call_id=call_id, body=body, actor=actor)


@app.post("/internal/voice-ai/calls/{call_id}/state", response_model=VoiceLiveCallOut)
def voice_ai_update_call_state(
    call_id: str,
    body: VoiceAICallStateUpdateIn,
    actor: dict = Depends(get_actor),
) -> VoiceLiveCallOut:
    return bridge_routes.voice_ai_update_call_state(call_id=call_id, body=body, actor=actor)


@app.post("/asterisk/reconnect", response_model=AsteriskBridgeStatusOut)
def reconnect_asterisk(_: dict = Depends(require_roles(Role.ADMIN))) -> AsteriskBridgeStatusOut:
    return bridge_routes.reconnect_asterisk()


@app.on_event("startup")
def _startup() -> None:
    _start_background_threads()


@app.on_event("shutdown")
def _shutdown() -> None:
    _stop_background_threads()