active_call/

app.rs

use crate::{
    call::{ActiveCallRef, sip::Invitation},
    callrecord::{
        CallRecordFormatter, CallRecordManagerBuilder, CallRecordSender, DefaultCallRecordFormatter,
    },
    config::Config,
    locator::RewriteTargetLocator,
    useragent::{
        RegisterOption,
        invitation::{
            FnCreateInvitationHandler, PendingDialog, PendingDialogGuard,
            default_create_invite_handler,
        },
        public_address::{
            LearningMessageInspector, SharedPublicAddress, build_contact, build_public_contact_uri,
            find_local_addr_for_uri,
        },
        registration::{RegistrationHandle, UserCredential},
    },
};

use crate::media::{cache::set_cache_dir, engine::StreamEngine};
use anyhow::Result;
use arc_swap::ArcSwap;
use async_trait::async_trait;
use chrono::{DateTime, Local};
use futures::FutureExt;
use humantime::parse_duration;
use rsipstack::rsip::prelude::HeadersExt;
use rsipstack::rsip::{Accept, Method, typed};
use rsipstack::transaction::{
    Endpoint, TransactionReceiver,
    endpoint::{TargetLocator, TransportEventInspector},
};
use rsipstack::transport::transport_layer::DomainResolver;
use rsipstack::{dialog::dialog_layer::DialogLayer, transaction::endpoint::MessageInspector};
use rsipstack::{
    rsip::{Host, HostWithPort},
    transport::SipAddr,
};
use std::future::pending;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::str::FromStr;
use std::sync::{Arc, RwLock};
use std::time::Duration;
use std::{collections::HashMap, net::SocketAddr};
use std::{collections::HashSet, time::Instant};
use std::{
    path::Path,
    sync::atomic::{AtomicBool, AtomicU64, Ordering},
};
use tokio::select;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
use tracing::{info, warn};

/// Generates a short unique session id for incoming calls, e.g. `s.3f9a2b1c4d5e`,
/// instead of reusing the raw SIP dialog-id string. Collisions with live
/// sessions are retried (practically impossible with 48 bits of randomness).
fn generate_short_session_id(invitation: &Invitation) -> String {
    loop {
        let uuid = uuid::Uuid::new_v4().simple().to_string();
        let session_id = format!("s.{}", &uuid[..12]);
        if !invitation.session_exists(&session_id) {
            return session_id;
        }
    }
}

pub struct AppStateInner {
    pub config: Arc<Config>,
    pub token: CancellationToken,
    pub stream_engine: Arc<StreamEngine>,
    pub callrecord_sender: Option<CallRecordSender>,
    pub endpoint: Endpoint,
    pub registration_handles: Mutex<HashMap<String, CancellationToken>>,
    pub alive_users: Arc<RwLock<HashSet<String>>>,
    pub dialog_layer: Arc<DialogLayer>,
    pub create_invitation_handler: Option<FnCreateInvitationHandler>,
    pub invitation: Invitation,
    pub routing_state: Arc<crate::call::RoutingState>,
    pub pending_playbooks: Arc<Mutex<HashMap<String, (String, Instant)>>>,
    pub learned_public_address: SharedPublicAddress,

    pub active_calls: Arc<std::sync::Mutex<HashMap<String, ActiveCallRef>>>,
    pub total_calls: AtomicU64,
    pub total_failed_calls: AtomicU64,
    pub uptime: DateTime<Local>,
    pub shutting_down: Arc<AtomicBool>,
}

pub type AppState = Arc<AppStateInner>;

pub struct AppStateBuilder {
    pub config: Option<Config>,
    pub stream_engine: Option<Arc<StreamEngine>>,
    pub callrecord_sender: Option<CallRecordSender>,
    pub callrecord_formatter: Option<Arc<dyn CallRecordFormatter>>,
    pub cancel_token: Option<CancellationToken>,
    pub create_invitation_handler: Option<FnCreateInvitationHandler>,
    pub config_path: Option<String>,

    pub message_inspector: Option<Box<dyn MessageInspector>>,
    pub target_locator: Option<Box<dyn TargetLocator>>,
    pub transport_inspector: Option<Box<dyn TransportEventInspector>>,
}

impl AppStateInner {
    pub fn auto_learn_public_address_enabled(&self) -> bool {
        self.config.auto_learn_public_address.unwrap_or(false)
    }

    pub fn get_dump_events_file(&self, session_id: &String) -> String {
        let recorder_root = self.config.recorder_path();
        let root = Path::new(&recorder_root);
        if !root.exists() {
            match std::fs::create_dir_all(root) {
                Ok(_) => {
                    info!("created dump events root: {}", root.to_string_lossy());
                }
                Err(e) => {
                    warn!(
                        "Failed to create dump events root: {} {}",
                        e,
                        root.to_string_lossy()
                    );
                }
            }
        }
        root.join(format!("{}.events.jsonl", session_id))
            .to_string_lossy()
            .to_string()
    }

    pub fn get_recorder_file(&self, session_id: &String) -> String {
        let recorder_root = self.config.recorder_path();
        let root = Path::new(&recorder_root);
        if !root.exists() {
            match std::fs::create_dir_all(root) {
                Ok(_) => {
                    info!("created recorder root: {}", root.to_string_lossy());
                }
                Err(e) => {
                    warn!(
                        "Failed to create recorder root: {} {}",
                        e,
                        root.to_string_lossy()
                    );
                }
            }
        }
        let desired_ext = self.config.recorder_format().extension();
        let mut filename = session_id.clone();
        if !filename
            .to_lowercase()
            .ends_with(&format!(".{}", desired_ext.to_lowercase()))
        {
            filename = format!("{}.{}", filename, desired_ext);
        }
        root.join(filename).to_string_lossy().to_string()
    }

    pub async fn serve(self: Arc<Self>) -> Result<()> {
        let incoming_txs = self.endpoint.incoming_transactions()?;
        let token = self.token.child_token();
        let endpoint_inner = self.endpoint.inner.clone();
        let dialog_layer = self.dialog_layer.clone();
        let app_state_clone = self.clone();

        match self.start_registration().await {
            Ok(count) => {
                info!("registration started, count: {}", count);
            }
            Err(e) => {
                warn!("failed to start registration: {:?}", e);
            }
        }

        let pending_cleanup_state = self.clone();
        let pending_cleanup_token = token.clone();
        crate::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(60));
            let ttl = Duration::from_secs(300);
            loop {
                tokio::select! {
                    _ = pending_cleanup_token.cancelled() => break,
                    _ = interval.tick() => {
                        let mut pending = pending_cleanup_state.pending_playbooks.lock().await;
                        let before = pending.len();
                        pending.retain(|_, (_, created_at)| created_at.elapsed() < ttl);
                        let removed = before - pending.len();
                        if removed > 0 {
                            info!(removed, remaining = pending.len(), "cleaned up stale pending_playbooks entries");
                        }
                    }
                }
            }
        });

        tokio::select! {
            _ = token.cancelled() => {
                info!("cancelled");
            }
            result = endpoint_inner.serve() => {
                if let Err(e) = result {
                    info!("endpoint serve error: {:?}", e);
                }
            }
            result = app_state_clone.process_incoming_request(dialog_layer.clone(), incoming_txs) => {
                if let Err(e) = result {
                    info!("process incoming request error: {:?}", e);
                }
            },
        }

        let total_secs = self.config.graceful_shutdown_timeout.unwrap_or(30);
        let timeout = self
            .config
            .graceful_shutdown
            .map(|_| Duration::from_secs(total_secs));

        match self.stop_registration(timeout).await {
            Ok(_) => {
                info!("registration stopped, waiting for clear");
            }
            Err(e) => {
                warn!("failed to stop registration: {:?}", e);
            }
        }
        info!("stopping");
        Ok(())
    }

    async fn process_incoming_request(
        self: Arc<Self>,
        dialog_layer: Arc<DialogLayer>,
        mut incoming: TransactionReceiver,
    ) -> Result<()> {
        while let Some(mut tx) = incoming.recv().await {
            let key: &rsipstack::transaction::key::TransactionKey = &tx.key;
            info!(?key, "received transaction");
            if tx.original.to_header()?.tag()?.as_ref().is_some() {
                match dialog_layer.match_dialog(&tx) {
                    Some(mut d) => {
                        crate::spawn(async move {
                            match d.handle(&mut tx).await {
                                Ok(_) => (),
                                Err(e) => {
                                    info!("error handling transaction: {:?}", e);
                                }
                            }
                        });
                        continue;
                    }
                    None => {
                        info!("dialog not found: {}", tx.original);
                        match tx
                            .reply(rsipstack::rsip::StatusCode::CallTransactionDoesNotExist)
                            .await
                        {
                            Ok(_) => (),
                            Err(e) => {
                                info!("error replying to request: {:?}", e);
                            }
                        }
                        continue;
                    }
                }
            }
            // out dialog, new server dialog
            let (state_sender, state_receiver) = dialog_layer.new_dialog_state_channel();
            match tx.original.method {
                rsipstack::rsip::Method::Invite | rsipstack::rsip::Method::Ack => {
                    // Reject new INVITEs during graceful shutdown
                    if self.shutting_down.load(Ordering::Relaxed) {
                        info!(?key, "rejecting INVITE during graceful shutdown");
                        match tx
                            .reply_with(
                                rsipstack::rsip::StatusCode::ServiceUnavailable,
                                vec![rsipstack::rsip::Header::Other(
                                    "Reason".into(),
                                    "SIP;cause=503;text=\"Server shutting down\"".into(),
                                )],
                                None,
                            )
                            .await
                        {
                            Ok(_) => (),
                            Err(e) => {
                                info!("error replying to request: {:?}", e);
                            }
                        }
                        continue;
                    }

                    let invitation_handler = match self.create_invitation_handler {
                        Some(ref create_invitation_handler) => {
                            create_invitation_handler(self.config.handler.as_ref()).ok()
                        }
                        _ => default_create_invite_handler(
                            self.config.handler.as_ref(),
                            Some(self.clone()),
                        ),
                    };
                    let invitation_handler = match invitation_handler {
                        Some(h) => h,
                        None => {
                            info!(?key, "no invite handler configured, rejecting INVITE");
                            match tx
                                .reply_with(
                                    rsipstack::rsip::StatusCode::ServiceUnavailable,
                                    vec![rsipstack::rsip::Header::Other(
                                        "Reason".into(),
                                        "SIP;cause=503;text=\"No invite handler configured\""
                                            .into(),
                                    )],
                                    None,
                                )
                                .await
                            {
                                Ok(_) => (),
                                Err(e) => {
                                    info!("error replying to request: {:?}", e);
                                }
                            }
                            continue;
                        }
                    };
                    let local_addr = tx
                        .connection
                        .as_ref()
                        .map(|connection| connection.get_addr().clone())
                        .or_else(|| dialog_layer.endpoint.get_addrs().first().cloned());
                    let contact_username =
                        tx.original.uri.auth.as_ref().map(|auth| auth.user.as_str());
                    let contact = local_addr.as_ref().map(|addr| {
                        build_public_contact_uri(
                            &self.learned_public_address,
                            self.auto_learn_public_address_enabled(),
                            addr,
                            contact_username,
                            None,
                        )
                    });

                    let dialog = match dialog_layer.get_or_create_server_invite(
                        &tx,
                        state_sender,
                        None,
                        contact,
                    ) {
                        Ok(d) => d,
                        Err(e) => {
                            // 481 Dialog/Transaction Does Not Exist
                            info!("failed to obtain dialog: {:?}", e);
                            match tx
                                .reply(rsipstack::rsip::StatusCode::CallTransactionDoesNotExist)
                                .await
                            {
                                Ok(_) => (),
                                Err(e) => {
                                    info!("error replying to request: {:?}", e);
                                }
                            }
                            continue;
                        }
                    };

                    let dialog_id = dialog.id();
                    let dialog_id_str = dialog_id.to_string();
                    // Incoming calls get a short public session id instead of
                    // the raw dialog-id string; the guard registers the mapping
                    // below so accept/hangup/message lookups can resolve it.
                    let session_id = generate_short_session_id(&self.invitation);
                    let dialog_id_for_cleanup = dialog_id.clone();
                    let token = self.token.child_token();
                    let pending_dialog = PendingDialog {
                        token: token.clone(),
                        dialog: dialog.clone(),
                        state_receiver,
                    };

                    let guard = Arc::new(PendingDialogGuard::new_with_session(
                        self.invitation.clone(),
                        dialog_id,
                        session_id.clone(),
                        pending_dialog,
                    ));

                    let accept_timeout = self
                        .config
                        .accept_timeout
                        .as_ref()
                        .and_then(|t| parse_duration(t).ok())
                        .unwrap_or_else(|| Duration::from_secs(60));

                    let mut dialog_ref = dialog.clone();
                    let routing_state = self.routing_state.clone();
                    let dialog_for_reject = dialog.clone();
                    let invitation_for_cleanup = self.invitation.clone();
                    let session_id_for_task = session_id.clone();
                    crate::spawn(async move {
                        info!(session_id = session_id_for_task, id = dialog_id_str, "incoming invite task started");
                        let _pending_guard = guard;
                        let token_ref = token.clone();
                        let accept_timeout_sleep = tokio::time::sleep(accept_timeout);
                        let invite_handler = invitation_handler.on_invite(
                            session_id_for_task.clone(),
                            token.clone(),
                            dialog.clone(),
                            routing_state,
                        );
                        let dialog_handle = dialog_ref.handle(&mut tx);
                        tokio::pin!(accept_timeout_sleep);
                        tokio::pin!(invite_handler);
                        tokio::pin!(dialog_handle);

                        let mut cancel_done = false;
                        let mut accept_timeout_done = false;
                        let mut invite_done = false;
                        loop {
                            let mut reject_request = None;

                            select! {
                                _ = token_ref.cancelled(), if !cancel_done => {
                                    cancel_done = true;
                                    reject_request = Some((
                                        rsipstack::rsip::StatusCode::ServiceUnavailable,
                                        "invite cancelled".to_string(),
                                        "cancelled",
                                    ));
                                }
                                _ = &mut accept_timeout_sleep, if !accept_timeout_done
                                    && dialog_for_reject.state().can_cancel() => {
                                    accept_timeout_done = true;
                                    reject_request = Some((
                                        rsipstack::rsip::StatusCode::RequestTimeout,
                                        "accept timeout".to_string(),
                                        "accept timeout",
                                    ));
                                }
                                result = &mut invite_handler, if !invite_done => {
                                    invite_done = true;
                                    match result {
                                        Ok(_) => {
                                            info!(id = dialog_id_str, "invite handler completed");
                                        }
                                        Err(e) => {
                                            info!(id = dialog_id_str, "error handling invite: {:?}", e);
                                            reject_request = Some((
                                                rsipstack::rsip::StatusCode::ServiceUnavailable,
                                                format!("Failed to process invite: {}", e),
                                                "invite handler error",
                                            ));
                                        }
                                    }
                                }
                                result = &mut dialog_handle => {
                                    match result {
                                        Ok(_) => {
                                            info!(id = dialog_id_str, "dialog handling finished");
                                        }
                                        Err(e) => {
                                            info!(
                                                id = dialog_id_str,
                                                "dialog handling ended with error: {:?}", e
                                            );
                                        }
                                    }
                                    if matches!(
                                        dialog_for_reject.state(),
                                        rsipstack::dialog::dialog::DialogState::Terminated(_, _)
                                    ) {
                                        info!(
                                            id = dialog_id_str,
                                            "terminated invite dialog finished, cancelling invite token"
                                        );
                                        token_ref.cancel();
                                    }
                                    info!(id = dialog_id_str, "incoming invite task finished");
                                    break;
                                }
                            }

                            if let Some((code, reason, source)) = reject_request {
                                if dialog_for_reject.state().can_cancel() {
                                    info!(
                                        id = dialog_id_str,
                                        ?code,
                                        %reason,
                                        source,
                                        "rejecting invite"
                                    );
                                    if let Err(e) =
                                        dialog_for_reject.reject(Some(code), Some(reason))
                                    {
                                        info!(
                                            id = dialog_id_str,
                                            "error rejecting invite: {:?}", e
                                        );
                                    }
                                    invitation_for_cleanup.get_pending_call(&dialog_id_for_cleanup);
                                    invitation_for_cleanup
                                        .dialog_layer
                                        .remove_dialog(&dialog_id_for_cleanup);
                                }
                            }
                        }
                    });
                }
                rsipstack::rsip::Method::Options => {
                    if self.config.enable_options_response.unwrap_or(true) {
                        info!(?key, "responding to out-of-dialog OPTIONS request");
                        let allow_header: rsipstack::rsip::Header =
                            typed::Allow::from(Method::all()).into();
                        let accept_header =
                            rsipstack::rsip::Header::Accept(Accept::new("application/sdp"));
                        match tx
                            .reply_with(
                                rsipstack::rsip::StatusCode::OK,
                                vec![allow_header, accept_header],
                                None,
                            )
                            .await
                        {
                            Ok(_) => (),
                            Err(e) => {
                                info!("error replying to OPTIONS: {:?}", e);
                            }
                        }
                    } else {
                        info!(?key, "ignoring out-of-dialog OPTIONS request");
                    }
                    continue;
                }
                rsipstack::rsip::Method::Refer => {
                    info!(?key, "ignoring out-of-dialog REFER");
                    match tx.reply(rsipstack::rsip::StatusCode::BadRequest).await {
                        Ok(_) => (),
                        Err(e) => {
                            info!("error replying to out-of-dialog REFER: {:?}", e);
                        }
                    }
                    continue;
                }
                _ => {
                    info!(?key, "received request: {:?}", tx.original.method);
                    match tx.reply(rsipstack::rsip::StatusCode::OK).await {
                        Ok(_) => (),
                        Err(e) => {
                            info!("error replying to request: {:?}", e);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    pub fn stop(&self) {
        if self.shutting_down.swap(true, Ordering::Relaxed) {
            return;
        }
        info!("stopping, marking as shutting down");
        self.token.cancel();
    }

    pub async fn graceful_stop(&self, total_timeout_secs: u64) -> Result<()> {
        if self.shutting_down.swap(true, Ordering::Relaxed) {
            return Ok(());
        }

        info!("graceful stopping, marking as shutting down");
        let timeout = Duration::from_secs(total_timeout_secs);

        let (reg_result, ()) = tokio::join!(
            self.stop_registration(Some(timeout)),
            self.wait_for_active_calls(timeout),
        );
        if let Err(e) = reg_result {
            warn!("stop_registration error: {}", e);
        }
        self.token.cancel();
        Ok(())
    }

    async fn wait_for_active_calls(&self, timeout: Duration) {
        let active_calls = self.active_calls.clone();
        let check_loop = async move {
            let mut last_count = usize::MAX;
            loop {
                let count = active_calls.lock().unwrap().len();
                if count == 0 {
                    break;
                }
                if count != last_count {
                    info!(active_calls = count, "waiting for active calls to finish");
                    last_count = count;
                }
                tokio::time::sleep(Duration::from_millis(500)).await;
            }
        };
        match tokio::time::timeout(timeout, check_loop).await {
            Ok(()) => info!("all active calls finished"),
            Err(_) => warn!("timed out waiting for active calls to finish, forcing shutdown"),
        }
    }

    pub async fn start_registration(&self) -> Result<usize> {
        let mut count = 0;
        if let Some(register_users) = &self.config.register_users {
            for option in register_users.iter() {
                match self.register(option.clone()).await {
                    Ok(_) => {
                        count += 1;
                    }
                    Err(e) => {
                        warn!("failed to register user: {:?} {:?}", e, option);
                    }
                }
            }
        }
        Ok(count)
    }

    pub fn find_credentials_for_callee(&self, callee: &str) -> Option<UserCredential> {
        let callee_uri = crate::sip_util::ensure_sip_scheme(callee.to_string());

        let parsed_callee = match rsipstack::rsip::Uri::try_from(callee_uri.as_str()) {
            Ok(uri) => uri,
            Err(e) => {
                warn!("failed to parse callee URI: {} {:?}", callee, e);
                return None;
            }
        };

        // Matching is host-kind-homogeneous: a domain callee matches domain
        // servers, an IP callee matches IP servers.
        match &parsed_callee.host_with_port.host {
            rsipstack::rsip::Host::Domain(domain) => {
                let domain = domain.0.clone();
                self.find_credentials(
                    callee,
                    &format!("domain {}", domain),
                    |host| matches!(host, rsipstack::rsip::Host::Domain(d) if d.0 == domain),
                )
            }
            rsipstack::rsip::Host::IpAddr(ip) => {
                let ip = *ip;
                self.find_credentials(callee, "IP match", move |host| {
                    matches!(host, rsipstack::rsip::Host::IpAddr(server_ip) if *server_ip == ip)
                })
            }
        }
    }

    /// Scan registered users for the first server matching `pred` that has a
    /// credential.
    fn find_credentials(
        &self,
        callee: &str,
        match_type: &str,
        pred: impl Fn(&rsipstack::rsip::Host) -> bool,
    ) -> Option<UserCredential> {
        let register_users = self.config.register_users.as_ref()?;
        for option in register_users.iter() {
            let server = crate::sip_util::ensure_sip_scheme(option.server.clone());

            let parsed_server = match rsipstack::rsip::Uri::try_from(server.as_str()) {
                Ok(uri) => uri,
                Err(e) => {
                    warn!("failed to parse server URI: {} {:?}", option.server, e);
                    continue;
                }
            };

            if pred(&parsed_server.host_with_port.host) {
                if let Some(cred) = &option.credential {
                    info!(
                        callee,
                        username = cred.username,
                        server = option.server,
                        match_type,
                        "Auto-injecting credentials from registered user for outbound call"
                    );
                    return Some(cred.clone());
                }
            }
        }
        None
    }

    pub async fn stop_registration(&self, wait_for_clear: Option<Duration>) -> Result<()> {
        {
            let mut handles = self.registration_handles.lock().await;
            for (_, cancel_token) in handles.drain() {
                cancel_token.cancel();
            }
        }

        if let Some(duration) = wait_for_clear {
            let live_users = self.alive_users.clone();
            let check_loop = async move {
                loop {
                    let is_empty = {
                        let users = live_users
                            .read()
                            .map_err(|_| anyhow::anyhow!("Lock poisoned"))?;
                        users.is_empty()
                    };
                    if is_empty {
                        break;
                    }
                    tokio::time::sleep(Duration::from_millis(50)).await;
                }
                Ok::<(), anyhow::Error>(())
            };
            match tokio::time::timeout(duration, check_loop).await {
                Ok(_) => {}
                Err(e) => {
                    warn!("failed to wait for clear: {}", e);
                    return Err(anyhow::anyhow!("failed to wait for clear: {}", e));
                }
            }
        }
        Ok(())
    }

    pub async fn register(&self, option: RegisterOption) -> Result<()> {
        let user = option.aor();
        let server = crate::sip_util::ensure_sip_scheme(option.server.clone());
        let sip_server = match rsipstack::rsip::Uri::try_from(server) {
            Ok(uri) => uri,
            Err(e) => {
                warn!("failed to parse server: {} {:?}", e, option.server);
                return Err(anyhow::anyhow!("failed to parse server: {}", e));
            }
        };
        let cancel_token = self.token.child_token();
        let credential = option.credential.clone().map(|c| c.into());
        let registration = rsipstack::dialog::registration::Registration::new(
            self.endpoint.inner.clone(),
            credential,
        );
        let mut handle = RegistrationHandle {
            registration,
            option,
            cancel_token: cancel_token.clone(),
            start_time: Instant::now(),
            last_update: Instant::now(),
            last_response: None,
        };
        self.registration_handles
            .lock()
            .await
            .insert(user.clone(), cancel_token);
        tracing::debug!(user = user.as_str(), "starting registration task");
        let alive_users = self.alive_users.clone();

        crate::spawn(async move {
            handle.start_time = Instant::now();
            let cancel_token = handle.cancel_token.clone();
            let addrs = handle.registration.endpoint.get_addrs();
            let local_bind_addr = if let Some(addr) = find_local_addr_for_uri(&addrs, &sip_server) {
                addr
            } else {
                warn!(
                    user = user.as_str(),
                    server = %sip_server,
                    "failed to get local bind address for registration transport"
                );
                alive_users.write().unwrap().remove(&user);
                return;
            };
            let user = handle.option.aor();
            alive_users.write().unwrap().remove(&user);
            let mut contact_address = local_bind_addr.addr.clone();
            let mut contact = build_contact(
                &local_bind_addr,
                Some(contact_address.clone()),
                Some(handle.option.username.as_str()),
                None,
            );
            let mut should_register = true;
            let mut timer = pending().boxed();

            loop {
                select! {
                    _ = cancel_token.cancelled() => {
                        break;
                    }
                    _ = timer.as_mut(), if !should_register => {
                        should_register = true;
                        timer = Box::pin(pending());
                    }
                    result = handle.do_register(&sip_server, None, &contact), if should_register => {
                        match result {
                            Ok((expires, new_addr)) => {
                                if handle
                                    .should_retry_registration_now(
                                        &local_bind_addr,
                                        &contact_address,
                                        new_addr.as_ref(),
                                    )
                                {
                                    if let Some(next_contact_address) = new_addr {
                                        info!(
                                            user = user.as_str(),
                                            current_contact = %contact_address,
                                            next_contact = %next_contact_address,
                                            "public address changed, retrying registration immediately",
                                        );
                                        contact_address = next_contact_address;
                                        contact = build_contact(
                                            &local_bind_addr,
                                            Some(contact_address.clone()),
                                            Some(handle.option.username.as_str()),
                                            None,
                                        );
                                        continue;
                                    }
                                }
                                info!(
                                    user = user.as_str(),
                                    expires = expires,
                                    contact = %contact_address,
                                    alive_users = alive_users.read().unwrap().len(),
                                    "registration refreshed",
                                );
                                alive_users.write().unwrap().insert(user.clone());
                                should_register = false;
                                timer = Box::pin(tokio::time::sleep(Duration::from_secs(
                                    (expires * 3 / 4) as u64,
                                )));
                            }
                            Err(e) => {
                                warn!(
                                    user = user.as_str(),
                                    alive_users = alive_users.read().unwrap().len(),
                                    "registration failed: {:?}", e
                                );
                                should_register = false;
                                timer = Box::pin(tokio::time::sleep(Duration::from_secs(60)));
                            }
                        }
                    }
                }
            }
            handle
                .do_register(&sip_server, Some(0), &contact)
                .await
                .ok();
            alive_users.write().unwrap().remove(&user);
        });
        Ok(())
    }
}

impl Drop for AppStateInner {
    fn drop(&mut self) {
        self.stop();
    }
}

struct SimpleDomainResolver;

#[async_trait]
impl DomainResolver for SimpleDomainResolver {
    async fn resolve(&self, target: &SipAddr) -> rsipstack::Result<SipAddr> {
        match &target.addr.host {
            Host::Domain(domain) => {
                let port: u16 = target.addr.port.map(|p| p.value()).unwrap_or(5060);
                let addr_str = format!("{}:{}", domain, port);
                match tokio::net::lookup_host(&addr_str).await {
                    Ok(mut addrs) => {
                        if let Some(addr) = addrs.next() {
                            Ok(SipAddr {
                                r#type: target.r#type,
                                addr: HostWithPort {
                                    host: Host::IpAddr(addr.ip()),
                                    port: Some(rsipstack::rsip::Port(addr.port())),
                                },
                            })
                        } else {
                            Err(rsipstack::Error::DnsResolutionError(format!(
                                "no addresses found for {}",
                                domain
                            )))
                        }
                    }
                    Err(e) => Err(rsipstack::Error::DnsResolutionError(format!(
                        "DNS resolution failed for {}: {}",
                        domain, e
                    ))),
                }
            }
            _ => Ok(target.clone()),
        }
    }
}

impl AppStateBuilder {
    pub fn new() -> Self {
        Self {
            config: None,
            stream_engine: None,
            callrecord_sender: None,
            callrecord_formatter: None,
            cancel_token: None,
            create_invitation_handler: None,
            config_path: None,
            message_inspector: None,
            target_locator: None,
            transport_inspector: None,
        }
    }

    pub fn with_config(mut self, config: Config) -> Self {
        self.config = Some(config);
        self
    }

    pub fn with_stream_engine(mut self, stream_engine: Arc<StreamEngine>) -> Self {
        self.stream_engine = Some(stream_engine);
        self
    }

    pub fn with_callrecord_sender(mut self, sender: CallRecordSender) -> Self {
        self.callrecord_sender = Some(sender);
        self
    }

    pub fn with_cancel_token(mut self, token: CancellationToken) -> Self {
        self.cancel_token = Some(token);
        self
    }

    pub fn with_config_metadata(mut self, path: Option<String>) -> Self {
        self.config_path = path;
        self
    }

    pub fn with_inspector(&mut self, inspector: Box<dyn MessageInspector>) -> &mut Self {
        self.message_inspector = Some(inspector);
        self
    }
    pub fn with_target_locator(&mut self, locator: Box<dyn TargetLocator>) -> &mut Self {
        self.target_locator = Some(locator);
        self
    }

    pub fn with_transport_inspector(
        &mut self,
        inspector: Box<dyn TransportEventInspector>,
    ) -> &mut Self {
        self.transport_inspector = Some(inspector);
        self
    }

    pub async fn build(self) -> Result<AppState> {
        let config: Arc<Config> = Arc::new(self.config.unwrap_or_default());
        let token = self
            .cancel_token
            .unwrap_or_else(|| CancellationToken::new());
        let _ = set_cache_dir(&config.media_cache_path);
        let local_ip = if !config.addr.is_empty() {
            std::net::IpAddr::from_str(config.addr.as_str())?
        } else {
            crate::net_tool::get_first_non_loopback_interface()?
        };
        let transport_layer = match catch_unwind(AssertUnwindSafe(|| {
            rsipstack::transport::TransportLayer::new(token.clone())
        })) {
            Ok(tl) => tl,
            Err(_) => {
                warn!(
                    "failed to initialize default DNS resolver with hickory-resolver, falling back to simple resolver via tokio::net::lookup_host"
                );
                rsipstack::transport::TransportLayer::new_with_domain_resolver(
                    token.clone(),
                    Box::new(SimpleDomainResolver),
                )
            }
        };
        let local_addr: SocketAddr = format!("{}:{}", local_ip, config.udp_port).parse()?;

        // Create UDP socket with SO_REUSEPORT for graceful restarts
        #[cfg(unix)]
        let std_socket = {
            use socket2::{Domain, Protocol, SockAddr, Socket, Type};

            let domain = if local_addr.is_ipv4() {
                Domain::IPV4
            } else {
                Domain::IPV6
            };
            let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))
                .map_err(|err| anyhow::anyhow!("Failed to create UDP socket: {}", err))?;

            socket
                .set_reuse_address(true)
                .map_err(|err| anyhow::anyhow!("Failed to set SO_REUSEADDR: {}", err))?;

            // SO_REUSEPORT (Linux/BSD)
            #[cfg(not(any(target_os = "solaris", target_os = "illumos", target_os = "cygwin")))]
            socket
                .set_reuse_port(true)
                .map_err(|err| anyhow::anyhow!("Failed to set SO_REUSEPORT: {}", err))?;

            socket
                .bind(&SockAddr::from(local_addr))
                .map_err(|err| anyhow::anyhow!("Failed to bind UDP socket: {}", err))?;

            let std_socket: std::net::UdpSocket = socket.into();
            std_socket
        };

        #[cfg(not(unix))]
        let std_socket = std::net::UdpSocket::bind(local_addr)?;

        std_socket.set_nonblocking(true)?;
        let tokio_socket = tokio::net::UdpSocket::from_std(std_socket)?;
        // Use the actual bound address (important when port=0 lets OS assign a port)
        let actual_addr = tokio_socket.local_addr()?;
        let bind_addr = rsipstack::transport::SipConnection::resolve_bind_address(actual_addr);
        let mut learned_public_address: SharedPublicAddress =
            Arc::new(ArcSwap::from_pointee(bind_addr.into()));

        let udp_inner = rsipstack::transport::udp::UdpInner {
            conn: tokio_socket,
            addr: rsipstack::transport::SipAddr {
                r#type: Some(rsipstack::rsip::transport::Transport::Udp),
                addr: bind_addr.into(),
            },
        };

        let external = config
            .external_ip
            .as_ref()
            .map(|ip| {
                format!("{}:{}", ip, actual_addr.port())
                    .parse()
                    .map_err(|e| anyhow::anyhow!("Failed to parse external address: {}", e))
            })
            .transpose()?;

        let udp_conn = rsipstack::transport::udp::UdpConnection::attach(
            udp_inner,
            external,
            Some(token.child_token()),
        )
        .await;

        info!(
            "start useragent, addr: {} (SO_REUSEPORT enabled)",
            udp_conn.get_addr()
        );

        transport_layer.add_transport(udp_conn.into());

        // Optional SIP over TLS transport
        if let Some(tls_port) = config.tls_port {
            let tls_addr: std::net::SocketAddr = format!("{}:{}", local_ip, tls_port).parse()?;
            let mut tls_cfg = rsipstack::transport::tls::TlsConfig::default();
            if let Some(ref cert_path) = config.tls_cert_file {
                tls_cfg.cert = Some(
                    std::fs::read(cert_path)
                        .map_err(|e| anyhow::anyhow!("tls_cert_file: {}", e))?,
                );
            }
            if let Some(ref key_path) = config.tls_key_file {
                tls_cfg.key = Some(
                    std::fs::read(key_path).map_err(|e| anyhow::anyhow!("tls_key_file: {}", e))?,
                );
            }
            let external_tls_addr = config
                .external_ip
                .as_ref()
                .and_then(|ip| format!("{}:{}", ip, tls_port).parse().ok());
            match rsipstack::transport::tls::TlsListenerConnection::new(
                tls_addr,
                external_tls_addr,
                tls_cfg,
            )
            .await
            {
                Ok(tls_conn) => {
                    transport_layer.add_transport(tls_conn.into());
                    info!("TLS SIP transport started on {}:{}", local_ip, tls_port);
                }
                Err(e) => {
                    return Err(anyhow::anyhow!("Failed to start TLS SIP transport: {}", e));
                }
            }
        }

        let endpoint_option = rsipstack::transaction::endpoint::EndpointOption::default();
        let mut endpoint_builder = rsipstack::EndpointBuilder::new();
        if let Some(ref user_agent) = config.useragent {
            endpoint_builder.with_user_agent(user_agent.as_str());
        }

        let mut endpoint_builder = endpoint_builder
            .with_cancel_token(token.child_token())
            .with_transport_layer(transport_layer)
            .with_option(endpoint_option);

        if config.auto_learn_public_address.unwrap_or_default() {
            let inspector = LearningMessageInspector::new(bind_addr.into(), self.message_inspector);
            learned_public_address = inspector.shared_public_address();
            endpoint_builder = endpoint_builder.with_inspector(Box::new(inspector));
        } else if let Some(inspector) = self.message_inspector {
            endpoint_builder = endpoint_builder.with_inspector(inspector);
        }

        if let Some(locator) = self.target_locator {
            endpoint_builder.with_target_locator(locator);
        } else if let Some(ref rules) = config.rewrites {
            endpoint_builder
                .with_target_locator(Box::new(RewriteTargetLocator::new(rules.clone())));
        }

        if let Some(inspector) = self.transport_inspector {
            endpoint_builder = endpoint_builder.with_transport_inspector(inspector);
        }

        let endpoint = endpoint_builder.build();
        let dialog_layer = Arc::new(DialogLayer::new(endpoint.inner.clone()));

        let stream_engine = self.stream_engine.unwrap_or_default();

        let callrecord_formatter = if let Some(formatter) = self.callrecord_formatter {
            formatter
        } else {
            let formatter = if let Some(ref callrecord) = config.callrecord {
                DefaultCallRecordFormatter::new_with_config(callrecord)
            } else {
                DefaultCallRecordFormatter::default()
            };
            Arc::new(formatter)
        };

        let callrecord_sender = if let Some(sender) = self.callrecord_sender {
            Some(sender)
        } else if let Some(ref callrecord) = config.callrecord {
            let builder = CallRecordManagerBuilder::new()
                .with_cancel_token(token.child_token())
                .with_config(callrecord.clone())
                .with_max_concurrent(32)
                .with_formatter(callrecord_formatter.clone());

            let mut callrecord_manager = builder.build();
            let sender = callrecord_manager.sender.clone();
            crate::spawn(async move {
                callrecord_manager.serve().await;
            });
            Some(sender)
        } else {
            None
        };

        let app_state = Arc::new(AppStateInner {
            config,
            token,
            stream_engine,
            callrecord_sender,
            endpoint,
            registration_handles: Mutex::new(HashMap::new()),
            alive_users: Arc::new(RwLock::new(HashSet::new())),
            dialog_layer: dialog_layer.clone(),
            create_invitation_handler: self.create_invitation_handler,
            invitation: Invitation::new(dialog_layer),
            routing_state: Arc::new(crate::call::RoutingState::new()),
            pending_playbooks: Arc::new(Mutex::new(HashMap::new())),
            learned_public_address,
            active_calls: Arc::new(std::sync::Mutex::new(HashMap::new())),
            total_calls: AtomicU64::new(0),
            total_failed_calls: AtomicU64::new(0),
            uptime: Local::now(),
            shutting_down: Arc::new(AtomicBool::new(false)),
        });

        Ok(app_state)
    }
}