call-center/services/asterisk_bridge_service/reconcile.py

from __future__ import annotations

from datetime import datetime, timezone

from sqlalchemy import select

from services.shared.db import get_session
from services.shared.sql_models import (
    AsteriskCallLinkRow,
    CallRecordingRow,
    VoiceAISessionRow,
    VoiceEventRow,
)


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

    return bridge_app


def parse_iso(value: str | None) -> datetime | None:
    text = str(value or "").strip()
    if not text:
        return None
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def has_voice_event(session, *, call_id: str, event_type: str) -> bool:
    row = session.execute(
        select(VoiceEventRow.id)
        .where(VoiceEventRow.call_id == call_id)
        .where(VoiceEventRow.event_type == event_type)
        .limit(1)
    ).scalar_one_or_none()
    return row is not None


def has_recording(session, *, call_id: str) -> bool:
    row = session.execute(
        select(CallRecordingRow.id).where(CallRecordingRow.call_id == call_id).limit(1)
    ).scalar_one_or_none()
    return row is not None


def latest_recording_for_call(session, *, call_id: str) -> CallRecordingRow | None:
    return session.execute(
        select(CallRecordingRow)
        .where(CallRecordingRow.call_id == call_id)
        .order_by(CallRecordingRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()


def _voice_session_recently_active(session, *, link: AsteriskCallLinkRow, now: datetime) -> bool:
    bridge = _bridge_app()
    non_terminal_states = {
        "queued",
        "greeting",
        "active",
        "thinking",
        "speaking",
        "listening",
        "handoff_requested",
        "handoff_required",
    }
    link_ai_state = str(link.ai_state or "").strip()
    voice_session_id = str(link.voice_session_id or "").strip()
    if not voice_session_id and link_ai_state not in non_terminal_states:
        return False
    voice_session = session.execute(
        select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == voice_session_id)
    ).scalar_one_or_none()
    if voice_session is None:
        return link_ai_state in non_terminal_states
    effective_ai_state = link_ai_state or str(voice_session.status or "").strip()
    if effective_ai_state not in non_terminal_states:
        return False
    recent_points = [
        bridge._parse_iso(voice_session.last_media_frame_at),
        bridge._parse_iso(voice_session.last_user_utterance_at),
        bridge._parse_iso(voice_session.last_ai_reply_at),
        bridge._parse_iso(voice_session.updated_at),
    ]
    recent_points = [point for point in recent_points if point is not None]
    if not recent_points:
        return False
    latest_point = max(recent_points)
    grace_seconds = max(bridge._reconcile_stale_seconds(), 12) + 8
    return max((now - latest_point).total_seconds(), 0) <= grace_seconds


def reconcile_stale_calls_once() -> None:
    bridge = _bridge_app()
    now = datetime.now(timezone.utc)
    session = get_session()
    try:
        unresolved_recording = (
            select(CallRecordingRow.id)
            .where(CallRecordingRow.call_id == AsteriskCallLinkRow.call_id)
            .correlate(AsteriskCallLinkRow)
            .exists()
        )
        active_ids = session.execute(
            select(AsteriskCallLinkRow.id)
            .where(AsteriskCallLinkRow.status == "active")
            .order_by(AsteriskCallLinkRow.id.asc())
            .limit(bridge._reconcile_scan_limit())
        ).scalars().all()
        ended_ids = session.execute(
            select(AsteriskCallLinkRow.id)
            .where(AsteriskCallLinkRow.status == "ended")
            .where(~unresolved_recording)
            .order_by(AsteriskCallLinkRow.id.asc())
            .limit(bridge._reconcile_scan_limit())
        ).scalars().all()
    finally:
        session.close()

    for link_id in active_ids:
        session = get_session()
        try:
            link = session.execute(
                select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.id == link_id)
            ).scalar_one_or_none()
            if link is None:
                continue

            started_dt = bridge._parse_iso(link.started_at) or bridge._parse_iso(link.updated_at) or now
            age_seconds = max((now - started_dt).total_seconds(), 0)
            if age_seconds < bridge._reconcile_stale_seconds():
                continue

            live_channels = bridge._list_channels_via_coreshowchannels(
                call_id=link.call_id,
                linked_id=link.linked_id,
            )
            if live_channels is None:
                continue
            if live_channels:
                preferred_channel = (
                    bridge._operator_channel_for_extension(live_channels, link.operator_extension)
                    or live_channels[0]
                )
                if preferred_channel != link.channel_name:
                    link.channel_name = preferred_channel
                link.updated_at = bridge.utc_now_iso()
                session.commit()
                continue

            if _voice_session_recently_active(session, link=link, now=now):
                link.updated_at = bridge.utc_now_iso()
                session.commit()
                continue

            ended_exists = bridge._has_voice_event(session, call_id=link.call_id, event_type="call.ended")
            if not ended_exists:
                bridge._emit_voice_event(
                    event_type="call.ended",
                    call_id=link.call_id,
                    interaction_id=link.interaction_id,
                    source_event_id=bridge._reconcile_source_event_id(link.call_id, "call.ended"),
                    payload={
                        "source": "asterisk",
                        "hangup_cause": "unknown",
                        "duration_seconds": None,
                        "linked_id": link.linked_id,
                        "reconciled": True,
                    },
                )
            now_iso = bridge.utc_now_iso()
            link.status = "ended"
            link.telephony_status = "ended"
            link.ended_at = link.ended_at or now_iso
            link.updated_at = now_iso
            session.commit()
        except Exception:
            session.rollback()
            continue
        finally:
            session.close()

    for link_id in ended_ids:
        session = get_session()
        try:
            link = session.execute(
                select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.id == link_id)
            ).scalar_one_or_none()
            if link is None:
                continue

            ready_exists = bridge._has_voice_event(session, call_id=link.call_id, event_type="recording.ready")
            recording_exists = bridge._has_recording(session, call_id=link.call_id)
            if recording_exists:
                continue

            candidate = bridge._find_remote_recording_for_call(link.call_id)
            if candidate is None:
                continue
            remote_path, file_name, mime_type, mtime_dt = candidate
            idle_seconds = max((now - mtime_dt).total_seconds(), 0)
            if idle_seconds < bridge._reconcile_settle_seconds():
                continue

            if not ready_exists:
                bridge._emit_voice_event(
                    event_type="recording.ready",
                    call_id=link.call_id,
                    interaction_id=link.interaction_id,
                    source_event_id=bridge._reconcile_source_event_id(link.call_id, "recording.ready"),
                    payload={
                        "source": "asterisk",
                        "remote_path": remote_path,
                        "file_name": file_name,
                        "mime_type": mime_type,
                        "duration_seconds": None,
                        "linked_id": link.linked_id,
                        "reconciled": True,
                    },
                )

            local_path, cleanup_required = bridge._fetch_recording_file(remote_path, file_name)
            try:
                bridge._upload_recording(
                    local_path=local_path,
                    call_id=link.call_id,
                    interaction_id=link.interaction_id,
                    source_event_id=bridge._reconcile_source_event_id(link.call_id, "recording.import"),
                    file_name=file_name,
                    mime_type=mime_type,
                    duration_seconds=None,
                )
            finally:
                if cleanup_required and local_path.exists():
                    local_path.unlink(missing_ok=True)

            link.updated_at = bridge.utc_now_iso()
            session.commit()
        except Exception:
            session.rollback()
            continue
        finally:
            session.close()