hold-slayer/core/pjsua_engine.py

"""
PJSUA2 SIP engine — call control *and* media in one library.

Why this exists
---------------
The gateway originally signalled with Sippy and expected PJSUA2 to carry
media. That cannot work: **PJSUA2 exposes no standalone RTP media object**.
Every ``AudioMedia`` subclass in the Python bindings is a file player,
recorder, tone generator or capture port, and RTP is reachable only through
``pj.Call.getAudioMedia()`` — on a dialog PJSUA2 itself owns. A design where
another stack owns the dialog can never obtain media from PJSUA2, so audio
never reached the classifier.

Owning the dialog is the price of owning the media, so this engine places the
call. See ``docs/architecture.md`` → "Media plane: why PJSUA2 places the call".

Safety
------
This engine is *only* reached through ``gateway.make_call``, which refuses
emergency numbers and enforces the concurrency cap **before** any SIP action.
Nothing here may be given a second dial path that bypasses those checks.

Threading
---------
Three execution contexts, as elsewhere in the codebase:

* the **asyncio loop** owns legs, the event bus, and the call manager;
* **PJSUA2 worker threads** run every ``on*`` callback below;
* (the Sippy ED thread is not involved — this engine replaces it.)

PJSUA2 callbacks cross to the loop through exactly one funnel,
``_post_from_pj`` → ``run_coroutine_threadsafe``. A callback must never touch
loop-owned state directly. Any thread PJSUA2 did not create must call
``libRegisterThread`` before touching a PJSUA2 object, which
``_ensure_registered`` handles.
"""

import asyncio
import gc
import logging
import threading
import uuid
from collections.abc import Callable

from core.sip_engine import SIPEngine
from models.device import Device

logger = logging.getLogger(__name__)


class PJSUAEngine(SIPEngine):
    """SIP engine backed by PJSUA2 for both signalling and media."""

    def __init__(
        self,
        sip_address: str = "0.0.0.0",
        sip_port: int = 5060,
        trunk_host: str = "",
        trunk_port: int = 5060,
        trunk_username: str = "",
        trunk_password: str = "",
        trunk_transport: str = "udp",
        domain: str = "gateway.local",
        did: str = "",
        media_pipeline=None,
        on_leg_state_change: Callable | None = None,
        on_device_registered: Callable | None = None,
        on_incoming_call: Callable | None = None,
    ):
        self._sip_address = sip_address
        self._sip_port = sip_port
        self._trunk_host = trunk_host
        self._trunk_port = trunk_port
        self._trunk_username = trunk_username
        self._trunk_password = trunk_password
        self._trunk_transport = trunk_transport
        self._domain = domain
        self._did = did

        # The media pipeline owns the PJSUA2 Endpoint; this engine borrows it
        # rather than creating a second one (PJSUA2 permits only one).
        self.media_pipeline = media_pipeline

        self._on_leg_state_change = on_leg_state_change
        self._on_device_registered = on_device_registered
        self._on_incoming_call = on_incoming_call

        self._loop: asyncio.AbstractEventLoop | None = None
        self._ready = False
        self._account = None
        self._trunk_registered = False
        self._trunk_reason = "not started"

        # PJSUA2-thread-owned: maps leg_id → pj.Call. Only touched from a
        # PJSUA2 callback or a method that has registered itself first.
        self._calls: dict[str, object] = {}
        self._lock = threading.Lock()

    # ================================================================
    # Thread boundary
    # ================================================================

    def _post_from_pj(self, coro) -> None:
        """Schedule loop work from a PJSUA2 worker thread. The one funnel."""
        if self._loop is None:
            return
        asyncio.run_coroutine_threadsafe(coro, self._loop)

    def _ensure_registered(self) -> None:
        """Register the calling thread with PJSUA2 if it isn't already.

        PJSUA2 aborts when a thread it does not know touches its objects.
        Calls made from the asyncio loop (hangup, DTMF) hit this.
        """
        try:
            import pjsua2 as pj

            ep = pj.Endpoint.instance()
            if not ep.libIsThreadRegistered():
                ep.libRegisterThread(threading.current_thread().name)
        except Exception as e:  # pragma: no cover - defensive
            logger.debug(f"  thread registration skipped: {e}")

    async def _emit_leg_state(self, leg_id: str, state: str) -> None:
        """Deliver a leg-state change on the loop."""
        if self._on_leg_state_change is None:
            return
        result = self._on_leg_state_change(leg_id, state)
        if asyncio.iscoroutine(result):
            await result

    # ================================================================
    # Lifecycle
    # ================================================================

    async def start(self) -> None:
        """Create the SIP transport and register with the trunk."""
        self._loop = asyncio.get_running_loop()
        logger.info("🔌 Starting PJSUA2 SIP engine...")

        if self.media_pipeline is None or not self.media_pipeline.endpoint:
            raise RuntimeError(
                "PJSUAEngine requires a started MediaPipeline — PJSUA2 allows "
                "only one Endpoint, so the pipeline owns it and the engine "
                "borrows it."
            )

        import pjsua2 as pj

        ep = self.media_pipeline.endpoint

        transport_cfg = pj.TransportConfig()
        transport_cfg.port = self._sip_port
        if self._sip_address and self._sip_address != "0.0.0.0":
            transport_cfg.boundAddress = self._sip_address

        tp_type = (
            pj.PJSIP_TRANSPORT_TCP
            if self._trunk_transport.lower() == "tcp"
            else pj.PJSIP_TRANSPORT_UDP
        )
        ep.transportCreate(tp_type, transport_cfg)

        self._create_account(ep)

        self._ready = True
        logger.info(f"🔌 PJSUA2 SIP engine ready on {self._sip_address}:{self._sip_port}")

    def _create_account(self, ep) -> None:
        """Build the account — registered to the trunk, or local-only."""
        import pjsua2 as pj

        engine = self

        class _Account(pj.Account):
            def onRegState(self, prm):  # noqa: N802 — PJSUA2 callback name
                try:
                    info = self.getInfo()
                    engine._trunk_registered = bool(info.regIsActive)
                    engine._trunk_reason = f"{prm.code} {prm.reason}".strip()
                    if info.regIsActive:
                        logger.info("  ✅ Trunk registration accepted")
                    else:
                        # A rejected REGISTER must not read as "registered":
                        # /health treats a registered trunk as a condition of
                        # being healthy.
                        logger.error(
                            f"  ❌ Trunk registration failed: {engine._trunk_reason}"
                        )
                except Exception as e:
                    logger.error(f"  onRegState error: {e}", exc_info=True)

            def onIncomingCall(self, prm):  # noqa: N802 — PJSUA2 callback name
                try:
                    engine._handle_incoming(self, prm.callId)
                except Exception as e:
                    logger.error(f"  onIncomingCall error: {e}", exc_info=True)

        acc_cfg = pj.AccountConfig()

        if self._trunk_host:
            acc_cfg.idUri = f"sip:{self._trunk_username}@{self._trunk_host}"
            acc_cfg.regConfig.registrarUri = f"sip:{self._trunk_host}:{self._trunk_port}"
            cred = pj.AuthCredInfo(
                "digest", "*", self._trunk_username, 0, self._trunk_password
            )
            acc_cfg.sipConfig.authCreds.append(cred)
        else:
            # No trunk configured: a local-only account still lets devices
            # register and inbound calls arrive.
            acc_cfg.idUri = f"sip:gateway@{self._domain}"
            self._trunk_reason = "No SIP trunk configured"

        self._account = _Account()
        self._account.create(acc_cfg)

        if self._trunk_host:
            logger.info(f"  Registering with trunk: {self._trunk_host}:{self._trunk_port}")

    async def stop(self) -> None:
        """Hang up everything and drop the account."""
        logger.info("🔌 Stopping PJSUA2 SIP engine...")
        self._ready = False

        self._ensure_registered()
        had_calls = bool(self._calls)
        for leg_id in list(self._calls.keys()):
            try:
                await self.hangup(leg_id)
            except Exception as e:
                logger.debug(f"  hangup during shutdown failed for {leg_id}: {e}")

        # hangup() only queues the BYE. Give PJSUA2 a moment to send it and
        # tear the media down, or the account is deleted with a call still
        # active ("deleting account 0 while call 0 is still active") and the
        # far end is left waiting on a dialog nobody closed.
        if had_calls:
            await asyncio.sleep(0.5)

        # Drop every PJSUA2 object before the pipeline destroys the endpoint.
        # A Call or Account finalised after libDestroy() aborts the process on
        # a native assertion, exactly as a stray media port does — and a Call
        # still alive keeps delivering callbacks into a half-torn-down
        # interpreter. Dropping the last reference is not enough on its own,
        # so force the collection here.
        with self._lock:
            self._calls.clear()
        self._account = None
        gc.collect()
        logger.info("🔌 PJSUA2 SIP engine stopped")

    async def is_ready(self) -> bool:
        return self._ready

    # ================================================================
    # Calls
    # ================================================================

    def _make_call_class(self):
        """Build the pj.Call subclass bound to this engine."""
        import pjsua2 as pj

        engine = self

        class _Call(pj.Call):
            def __init__(self, acc, leg_id: str, call_id=pj.PJSUA_INVALID_ID):
                super().__init__(acc, call_id)
                self.leg_id = leg_id

            def onCallState(self, prm):  # noqa: N802 — PJSUA2 callback name
                # PJSUA2 keeps delivering callbacks while the interpreter is
                # tearing down, when module globals may already be cleared —
                # hence the local alias and the bare except. A raise here
                # escapes into C++ and takes the worker thread with it.
                state_map = _STATE_MAP
                try:
                    info = self.getInfo()
                    state = state_map.get(info.state)
                    if state is None:
                        return
                    if state == "terminated":
                        engine._forget_call(self.leg_id)
                    engine._post_from_pj(engine._emit_leg_state(self.leg_id, state))
                except Exception:
                    try:
                        logger.error("  onCallState error", exc_info=True)
                    except Exception:
                        pass

            def onCallMediaState(self, prm):  # noqa: N802 — PJSUA2 callback name
                """Media is up — hand the audio to the pipeline.

                This is the callback the whole refactor exists for: it is the
                only place PJSUA2 surfaces an RTP-backed AudioMedia.
                """
                try:
                    info = self.getInfo()
                    for i, mi in enumerate(info.media):
                        if (
                            mi.type == pj.PJMEDIA_TYPE_AUDIO
                            and mi.status == pj.PJSUA_CALL_MEDIA_ACTIVE
                        ):
                            engine._attach_media(self.leg_id, self.getAudioMedia(i))
                            break
                except Exception:
                    try:
                        logger.error("  onCallMediaState error", exc_info=True)
                    except Exception:
                        pass

        return _Call

    def _attach_media(self, leg_id: str, audio_media) -> None:
        """Register a live AudioMedia with the pipeline (PJSUA2 thread)."""
        if self.media_pipeline is None:
            return
        try:
            self.media_pipeline.attach_call_media(leg_id, audio_media)
            logger.info(f"  🎵 Media active for {leg_id}")
        except Exception as e:
            logger.error(f"  Failed to attach media for {leg_id}: {e}", exc_info=True)

    def _forget_call(self, leg_id: str) -> None:
        with self._lock:
            self._calls.pop(leg_id, None)
        if self.media_pipeline is not None:
            try:
                self.media_pipeline.remove_stream(leg_id)
            except Exception as e:
                logger.debug(f"  stream cleanup failed for {leg_id}: {e}")

    async def make_call(self, number: str, caller_id: str | None = None) -> str:
        """Place an outbound call. Reached only via gateway.make_call."""
        if not self._ready:
            raise RuntimeError("SIP engine not ready")

        import pjsua2 as pj

        leg_id = f"leg_{uuid.uuid4().hex[:12]}"
        target = (
            f"sip:{number}@{self._trunk_host}:{self._trunk_port}"
            if self._trunk_host
            else f"sip:{number}@{self._domain}"
        )
        logger.info(f"📞 Placing call to {target} (leg: {leg_id})")

        self._ensure_registered()
        call_cls = self._make_call_class()
        call = call_cls(self._account, leg_id)

        prm = pj.CallOpParam(True)
        call.makeCall(target, prm)

        with self._lock:
            self._calls[leg_id] = call
        return leg_id

    async def hangup(self, call_leg_id: str) -> None:
        import pjsua2 as pj

        with self._lock:
            call = self._calls.get(call_leg_id)
        if call is None:
            return
        self._ensure_registered()
        try:
            call.hangup(pj.CallOpParam(True))
        except Exception as e:
            logger.debug(f"  hangup failed for {call_leg_id}: {e}")
        self._forget_call(call_leg_id)

    async def send_dtmf(self, call_leg_id: str, digits: str) -> None:
        """Send DTMF as RFC 2833 — the in-band path a real IVR expects."""
        with self._lock:
            call = self._calls.get(call_leg_id)
        if call is None:
            logger.warning(f"  send_dtmf: no call for {call_leg_id}")
            return
        self._ensure_registered()
        call.dialDtmf(digits)
        logger.info(f"  Sent DTMF '{digits}' on {call_leg_id}")

    async def call_device(self, device: Device) -> str:
        """Ring a registered device (transfer target)."""
        if not self._ready:
            raise RuntimeError("SIP engine not ready")

        import pjsua2 as pj

        leg_id = f"leg_{uuid.uuid4().hex[:12]}"
        target = device.sip_uri or f"sip:{device.id}@{self._domain}"
        logger.info(f"📞 Ringing device {device.id} at {target} (leg: {leg_id})")

        self._ensure_registered()
        call_cls = self._make_call_class()
        call = call_cls(self._account, leg_id)
        call.makeCall(target, pj.CallOpParam(True))

        with self._lock:
            self._calls[leg_id] = call
        return leg_id

    def _handle_incoming(self, account, call_id) -> None:
        """Inbound INVITE (PJSUA2 thread) — answer and hand to the receptionist."""
        import pjsua2 as pj

        leg_id = f"leg_{uuid.uuid4().hex[:12]}"
        call_cls = self._make_call_class()
        call = call_cls(account, leg_id, call_id)

        try:
            info = call.getInfo()
            remote = info.remoteUri
        except Exception:
            remote = "unknown"

        with self._lock:
            self._calls[leg_id] = call

        call.answer(pj.CallOpParam(True))
        logger.info(f"📞 Inbound call {leg_id} from {remote}")

        if self._on_incoming_call is not None:
            result = self._on_incoming_call(leg_id, remote)
            if asyncio.iscoroutine(result):
                self._post_from_pj(result)

    # ================================================================
    # Bridging
    # ================================================================

    async def bridge_calls(self, leg_a: str, leg_b: str) -> str:
        """Join two legs in the conference bridge."""
        bridge_id = f"bridge_{uuid.uuid4().hex[:8]}"
        if self.media_pipeline is not None:
            self.media_pipeline.bridge_streams(leg_a, leg_b)
        logger.info(f"  🌉 Bridged {leg_a} ↔ {leg_b} ({bridge_id})")
        return bridge_id

    async def unbridge(self, bridge_id: str) -> None:
        logger.info(f"  Unbridged {bridge_id}")

    def get_audio_stream(self, call_leg_id: str):
        if self.media_pipeline is not None:
            return self.media_pipeline.get_audio_tap(call_leg_id)
        return None

    # ================================================================
    # Status
    # ================================================================

    async def get_registered_devices(self) -> list[dict]:
        return []

    async def get_trunk_status(self) -> dict:
        return {
            "registered": self._trunk_registered,
            "host": self._trunk_host or "not configured",
            "port": self._trunk_port,
            "transport": self._trunk_transport,
            "username": self._trunk_username,
            "reason": None if self._trunk_registered else self._trunk_reason,
        }


# Populated lazily: the pjsua2 constants are unavailable until import, and
# the module must import cleanly in stub mode.
_STATE_MAP: dict = {}


def _init_state_map() -> None:
    global _STATE_MAP
    if _STATE_MAP:
        return
    try:
        import pjsua2 as pj
    except ImportError:
        return
    _STATE_MAP = {
        pj.PJSIP_INV_STATE_CALLING: "trying",
        pj.PJSIP_INV_STATE_EARLY: "ringing",
        pj.PJSIP_INV_STATE_CONNECTING: "trying",
        pj.PJSIP_INV_STATE_CONFIRMED: "connected",
        pj.PJSIP_INV_STATE_DISCONNECTED: "terminated",
    }


_init_state_map()