ringo-flow 0.11.0

//! A live agent session: one headless backend instance plus its event stream.
//! Incoming events are folded into an [`AgentState`] (published over a `watch`
//! channel) that the runner asserts against.

use super::state::{AgentState, reduce};
use anyhow::{Context, Result};
use ringo_core::account::{Account, BackendOptions};
use ringo_core::backend::Backend;
use ringo_core::backend::BaresipBackend;
use ringo_core::event::AppEvent;
use ringo_core::event::InviteHeaders;
use ringo_core::phone::Phone;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::watch;

/// How often the reader task polls for inbound INVITE headers.
const TRACE_POLL_INTERVAL: Duration = Duration::from_millis(150);

pub struct AgentSession {
    pub aor: String,
    pub regint: u32,
    /// Backend audio key for this agent (the account username) — identifies its
    /// in-process received-audio buffer.
    audio_key: String,
    phone: Box<dyn Phone>,
    state_rx: watch::Receiver<AgentState>,
    _handle: Box<dyn Send>,
}

impl AgentSession {
    /// Spawn a backend for an account, connect, and fold events into shared
    /// state. `name` labels the instance/logs.
    pub async fn connect(name: &str, account: Account, options: &BackendOptions) -> Result<Self> {
        let rt = tokio::runtime::Handle::current();
        let session = BaresipBackend
            .spawn_session(&rt, name, &account, options)
            .with_context(|| format!("spawn backend for `{name}`"))?;

        let (state_tx, state_rx) = watch::channel(AgentState::default());
        let state_tx = Arc::new(state_tx);

        // Bridge the sync mpsc events into an async channel so the reader task
        // can consume them without blocking the tokio runtime.
        let (async_event_tx, mut async_event_rx) = tokio::sync::mpsc::channel::<AppEvent>(64);
        let events_rx = session.events;
        tokio::task::spawn_blocking(move || {
            while let Ok(event) = events_rx.recv() {
                if async_event_tx.blocking_send(event).is_err() {
                    break;
                }
            }
        });

        // Reader: events → state.
        let reader_tx = Arc::clone(&state_tx);
        tokio::spawn(async move {
            while let Some(event) = async_event_rx.recv().await {
                reader_tx.send_modify(|s| reduce(s, &event));
            }
        });

        // Trace poll: inbound INVITE headers → state (the events don't carry
        // them). Stops once the session's receivers are gone.
        if let Some(header_poll) = session.header_poll {
            let trace_tx = Arc::clone(&state_tx);
            tokio::spawn(async move {
                let poll = Arc::new(header_poll);
                let mut ticker = tokio::time::interval(TRACE_POLL_INTERVAL);
                ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
                loop {
                    ticker.tick().await;
                    if trace_tx.is_closed() {
                        break;
                    }
                    let poll = Arc::clone(&poll);
                    match tokio::task::spawn_blocking(move || poll()).await {
                        Ok(Some(invites)) => merge_received_headers(&trace_tx, invites),
                        Ok(None) => {}
                        Err(_) => break,
                    }
                }
            });
        }

        Ok(Self {
            aor: format!("sip:{}@{}", account.username, account.domain),
            regint: account.regint.unwrap_or(3600),
            audio_key: account.username.clone(),
            phone: session.phone,
            state_rx,
            _handle: session.handle,
        })
    }

    /// This agent's SIP domain (the host part of its AOR), used to build a full
    /// request URI when dialing a bare number/extension.
    pub fn domain(&self) -> &str {
        self.aor.rsplit('@').next().unwrap_or("")
    }

    /// A fresh receiver on this agent's state (used by `wait` to guard calls).
    pub fn state(&self) -> watch::Receiver<AgentState> {
        self.state_rx.clone()
    }

    /// Switch the agent's audio source on its active call.
    pub fn set_audio_source(&self, spec: &str) {
        self.phone.set_audio_source(spec);
    }

    /// Recently received (decoded) mono audio + sample rate, captured in-process
    /// by the backend. Used by `verify-audio` instead of reading WAV recordings.
    pub fn received_audio(&self) -> Option<(Vec<i16>, u32)> {
        ringo_core::received_audio(&self.audio_key)
    }

    /// Recently sent (rendered) mono audio + sample rate, captured in-process.
    /// Populated only with `--save-audio` (full capture); used to save the sent
    /// recording.
    pub fn sent_audio(&self) -> Option<(Vec<i16>, u32)> {
        ringo_core::sent_audio(&self.audio_key)
    }

    // ── Commands ────────────────────────────────────────────────────────────

    pub fn register(&self) {
        self.phone.register(&self.aor, self.regint);
    }
    pub fn dial(&self, target: &str) {
        self.phone.dial(target);
    }
    pub fn accept(&self) {
        self.phone.accept();
    }
    pub fn hold(&self) {
        self.phone.hold();
    }
    pub fn resume(&self) {
        self.phone.resume();
    }
    pub fn mute(&self) {
        self.phone.mute();
    }
    pub fn send_dtmf(&self, digit: char) {
        self.phone.send_dtmf(digit);
    }
    pub fn add_header(&self, key: &str, value: &str) {
        self.phone.add_header(key, value);
    }
    pub fn hangup(&self) {
        self.phone.hangup();
    }
    pub fn hangup_all(&self) {
        self.phone.hangup_all();
    }
    pub fn transfer(&self, uri: &str) {
        self.phone.transfer(uri);
    }
    pub fn attended_transfer_start(&self, uri: &str) {
        self.phone.attended_transfer_start(uri);
    }
    pub fn attended_transfer_exec(&self) {
        self.phone.attended_transfer_exec();
    }
    pub fn attended_transfer_abort(&self) {
        self.phone.attended_transfer_abort();
    }
    pub fn deflect_incoming(&self, contact: &str, diversion: Option<&str>) {
        self.phone.deflect_incoming(contact, diversion);
    }
    pub fn arm_invite_response(&self, scode: u16, reason: &str, headers: Vec<String>) {
        self.phone.arm_invite_response(scode, reason, headers);
    }
    pub fn disarm_invite_response(&self) {
        self.phone.disarm_invite_response();
    }
}

/// Merge newly-parsed inbound INVITE headers into the agent state, notifying
/// watchers only when a new Call-ID appears (so the `await_until` wait loops
/// aren't woken on every poll).
fn merge_received_headers(state_tx: &watch::Sender<AgentState>, invites: InviteHeaders) {
    if invites.is_empty() {
        return;
    }
    let has_new = {
        let cur = state_tx.borrow();
        invites
            .keys()
            .any(|k| !cur.received_headers.contains_key(k))
    };
    if has_new {
        state_tx.send_modify(|s| {
            for (call_id, headers) in invites {
                s.received_headers.entry(call_id).or_insert(headers);
            }
        });
    }
}