active_call/
main_builder.rs

1use crate::media::engine::StreamEngine;
2use anyhow::Result;
3use axum::Router;
4use axum::response::IntoResponse;
5use axum::routing::get;
6use clap::Parser;
7use dotenvy::dotenv;
8use futures::{FutureExt, future};
9use reqwest::StatusCode;
10use std::sync::Arc;
11use tokio::signal;
12use tower_http::services::ServeDir;
13use tracing::level_filters::LevelFilter;
14use tracing::{info, warn};
15use tracing_appender::non_blocking::WorkerGuard;
16use tracing_subscriber::EnvFilter;
17use tracing_subscriber::fmt::time::LocalTime;
18use tracing_subscriber::layer::SubscriberExt;
19use tracing_subscriber::util::SubscriberInitExt;
20
21use crate::app::{AppStateBuilder, AppStateInner};
22use crate::config::{Cli, Config};
23use uuid::Uuid;
24
25pub struct MainBuilder {
26    pub cli: Cli,
27    pub config: Config,
28    /// Default active-call routes
29    pub router: Router<Arc<AppStateInner>>,
30    /// If None - default will be used
31    pub stream_engine: Option<Arc<StreamEngine>>,
32    /// To keep the logging guard alive
33    pub guard_holder: Option<WorkerGuard>,
34}
35
36impl MainBuilder {
37    pub fn from_cli(cli: Cli) -> Self {
38        let config = if let Some(path) = &cli.conf {
39            Config::load(&path).unwrap_or_else(|e| {
40                println!("Failed to load config from {}: {}, using defaults", path, e);
41                Config::default()
42            })
43        } else {
44            Config::default()
45        };
46        Self::new(cli, config)
47    }
48
49    pub fn new(cli: Cli, config: Config) -> Self {
50        let config = Self::apply_cli_overrides(&cli, config);
51        MainBuilder {
52            cli,
53            config,
54            stream_engine: None,
55            router: Self::default_router(),
56            guard_holder: None,
57        }
58    }
59
60    fn apply_cli_overrides(cli: &Cli, mut config: Config) -> Config {
61        if let Some(ref http) = cli.http {
62            config.http_addr = http.clone();
63        }
64
65        if let Some(ref sip) = cli.sip {
66            if let Ok(port) = sip.parse::<u16>() {
67                config.udp_port = port;
68            } else if let Ok(socket_addr) = sip.parse::<std::net::SocketAddr>() {
69                config.addr = socket_addr.ip().to_string();
70                config.udp_port = socket_addr.port();
71            } else {
72                config.addr = sip.clone();
73            }
74        }
75
76        // Auto-configure handler from CLI parameter
77        if let Some(handler_str) = &cli.handler {
78            use crate::config::InviteHandlerConfig;
79
80            if handler_str.starts_with("http://") || handler_str.starts_with("https://") {
81                // Webhook handler
82                config.handler = Some(InviteHandlerConfig::Webhook {
83                    url: Some(handler_str.clone()),
84                    urls: None,
85                    method: None,
86                    headers: None,
87                });
88                info!("CLI handler configured as webhook: {}", handler_str);
89            } else if handler_str.ends_with(".md") {
90                // Playbook handler with default playbook
91                config.handler = Some(InviteHandlerConfig::Playbook {
92                    rules: None,
93                    default: Some(handler_str.clone()),
94                });
95                info!(
96                    "CLI handler configured as playbook default: {}",
97                    handler_str
98                );
99            } else {
100                warn!(
101                    "Invalid handler format: {}. Should be http(s):// URL or .md file",
102                    handler_str
103                );
104            }
105        }
106
107        if let Some(ref external_ip) = cli.external_ip {
108            config.external_ip = Some(external_ip.clone());
109        }
110
111        if let Some(ref codecs) = cli.codecs {
112            config.codecs = Some(codecs.clone());
113        }
114
115        config
116    }
117
118    pub async fn run(mut self) -> Result<()> {
119        Self::init();
120        #[cfg(feature = "offline")]
121        if self.handle_offline()? {
122            return Ok(());
123        }
124        self.setup_logging()?;
125        info!("Starting active-call service...");
126
127        let app_state = self.build_app_state().await?;
128        self.handle_cli_direct_call(app_state.clone()).await;
129        let listener = self.build_tcp_listener()?;
130        let router = self.router.clone().with_state(app_state.clone());
131        self.serve(router, app_state, listener).await
132    }
133
134    fn init() {
135        rustls::crypto::aws_lc_rs::default_provider()
136            .install_default()
137            .expect("Failed to install rustls crypto provider");
138        dotenv().ok();
139    }
140
141    #[cfg(feature = "offline")]
142    fn handle_offline(&self) -> Result<bool> {
143        use crate::offline::{ModelDownloader, ModelType, OfflineConfig, init_offline_models};
144        use std::path::PathBuf;
145
146        // Handle model download if requested
147        if let Some(model_type) = &self.cli.download_models {
148            let models_dir = PathBuf::from(&self.cli.models_dir);
149            let downloader = ModelDownloader::new()?;
150
151            let model = ModelType::from_str(model_type).ok_or_else(|| {
152                anyhow::anyhow!(
153                    "Unknown model type: {}. Use: sensevoice, supertonic, or all",
154                    model_type
155                )
156            })?;
157
158            downloader.download(model, &models_dir)?;
159            println!("✓ Models downloaded to: {}", models_dir.display());
160
161            if self.cli.exit_after_download {
162                return Ok(true);
163            }
164        }
165
166        // Initialize offline models
167        let offline_config =
168            OfflineConfig::new(PathBuf::from(&self.cli.models_dir), num_cpus::get().min(4));
169
170        // Only initialize if models directory exists
171        if offline_config.models_dir.exists() {
172            init_offline_models(offline_config)?;
173            println!("Offline models initialized from: {}", self.cli.models_dir);
174        } else {
175            println!(
176                "Models directory not found: {}. Offline features will not be available. Run with --download-models to download.",
177                self.cli.models_dir
178            );
179        }
180
181        Ok(false)
182    }
183
184    fn setup_logging(&mut self) -> Result<()> {
185        let mut env_filter = EnvFilter::from_default_env();
186        if let Some(Ok(level)) = self
187            .config
188            .log_level
189            .as_ref()
190            .map(|level| level.parse::<LevelFilter>())
191        {
192            env_filter = env_filter.add_directive(level.into());
193        }
194        env_filter = env_filter.add_directive("ort=warn".parse()?);
195        let mut file_layer = None;
196        let mut fmt_layer = None;
197        if let Some(ref log_file) = self.config.log_file {
198            let file = std::fs::OpenOptions::new()
199                .create(true)
200                .append(true)
201                .open(log_file)
202                .expect("Failed to open log file");
203            let (non_blocking, guard) = tracing_appender::non_blocking(file);
204            self.guard_holder = Some(guard);
205            file_layer = Some(
206                tracing_subscriber::fmt::layer()
207                    .with_timer(LocalTime::rfc_3339())
208                    .with_ansi(false)
209                    .with_writer(non_blocking),
210            );
211        } else {
212            fmt_layer = Some(tracing_subscriber::fmt::layer().with_timer(LocalTime::rfc_3339()));
213        }
214
215        if let Some(file_layer) = file_layer {
216            tracing_subscriber::registry()
217                .with(env_filter)
218                .with(file_layer)
219                .try_init()?;
220        } else if let Some(fmt_layer) = fmt_layer {
221            tracing_subscriber::registry()
222                .with(env_filter)
223                .with(fmt_layer)
224                .try_init()?;
225        }
226
227        Ok(())
228    }
229
230    async fn build_app_state(&self) -> Result<Arc<AppStateInner>> {
231        let stream_engine = self
232            .stream_engine
233            .clone()
234            .unwrap_or_else(|| Arc::new(StreamEngine::default()));
235
236        let result = AppStateBuilder::new()
237            .with_config(self.config.clone())
238            .with_stream_engine(stream_engine)
239            .with_config_metadata(self.cli.conf.clone())
240            .build()
241            .await?;
242        info!("AppState started");
243        Ok(result)
244    }
245
246    async fn handle_cli_direct_call(&self, app_state: Arc<AppStateInner>) {
247        if let Some(ref callee) = self.cli.call {
248            let callee = callee.clone();
249            let app_state_clone = app_state.clone();
250            let playbook = if let Some(h) = &self.cli.handler {
251                if h.ends_with(".md") {
252                    Some(h.clone())
253                } else {
254                    None
255                }
256            } else {
257                None
258            };
259
260            tokio::spawn(async move {
261                // Wait a bit for the SIP stack to initialize
262                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
263
264                let session_id = format!("c.{}", Uuid::new_v4());
265                info!(session_id, "Starting CLI outgoing call to: {}", callee);
266
267                let (command_sender, command_receiver) = tokio::sync::mpsc::unbounded_channel();
268                let (event_sender, _event_receiver) = tokio::sync::mpsc::unbounded_channel();
269                let (_audio_tx, audio_rx) = tokio::sync::mpsc::unbounded_channel();
270
271                use crate::CallOption;
272                use crate::call::{ActiveCallType, Command};
273
274                let invite_cmd = Command::Invite {
275                    option: CallOption {
276                        callee: Some(callee.clone()),
277                        ..Default::default()
278                    },
279                };
280
281                let _ = command_sender.send(invite_cmd);
282
283                crate::handler::handler::call_handler_core(
284                    ActiveCallType::Sip,
285                    session_id,
286                    app_state_clone,
287                    tokio_util::sync::CancellationToken::new(),
288                    audio_rx,
289                    None,
290                    false,
291                    0,
292                    command_receiver,
293                    event_sender,
294                    None,     // extras
295                    playbook, // playbook_name — passed directly
296                )
297                .await;
298            });
299        }
300    }
301
302    fn build_tcp_listener(&self) -> Result<tokio::net::TcpListener> {
303        let http_addr = self.config.http_addr.clone();
304
305        // Create TCP listener with SO_REUSEPORT for graceful restarts
306        let addr: std::net::SocketAddr = http_addr.parse()?;
307        // Create socket manually to set SO_REUSEPORT before bind
308        let std_listener = {
309            use socket2::{Domain, Protocol, Socket, Type};
310
311            let domain = if addr.is_ipv4() {
312                Domain::IPV4
313            } else {
314                Domain::IPV6
315            };
316            let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))?;
317            socket.set_reuse_address(true)?;
318            #[cfg(all(unix, not(any(target_os = "solaris", target_os = "illumos"))))]
319            socket.set_reuse_port(true)?;
320            socket.bind(&addr.into())?;
321            socket.listen(1024)?;
322            socket.set_nonblocking(true)?;
323            std::net::TcpListener::from(socket)
324        };
325
326        let listener = tokio::net::TcpListener::from_std(std_listener)?;
327        info!("listening on http://{} (SO_REUSEPORT enabled)", http_addr);
328        Ok(listener)
329    }
330
331    fn default_router() -> Router<Arc<AppStateInner>> {
332        let router = crate::handler::call_router()
333            .merge(crate::handler::playbook_router())
334            .merge(crate::handler::iceservers_router())
335            .route("/", get(index))
336            .nest_service("/static", ServeDir::new("static"));
337        router
338    }
339
340    async fn serve(
341        &self,
342        router: Router,
343        app_state: Arc<AppStateInner>,
344        listener: tokio::net::TcpListener,
345    ) -> Result<()> {
346        let app_state_clone = app_state.clone();
347        let graceful_shutdown = self.config.graceful_shutdown.unwrap_or_default();
348        let graceful_shutdown_timeout = self.config.graceful_shutdown_timeout.unwrap_or(30);
349
350        let axum_serving = axum::serve(listener, router).into_future();
351        let app_state_serving = app_state_clone.serve();
352        let mut canceled = false;
353        let cancel_timeout = future::pending().boxed();
354        let shutdown_task = future::pending::<anyhow::Result<()>>().boxed();
355        let shutdown_signal = shutdown_signal().boxed();
356
357        tokio::pin!(axum_serving);
358        tokio::pin!(app_state_serving);
359        tokio::pin!(cancel_timeout);
360        tokio::pin!(shutdown_task);
361        tokio::pin!(shutdown_signal);
362
363        loop {
364            tokio::select! {
365                result = &mut axum_serving => {
366                    if let Err(e) = result {
367                        warn!("axum serve error: {:?}", e);
368                    }
369                    break;
370                }
371                res = &mut app_state_serving => {
372                    if let Err(e) = res {
373                        warn!("AppState server error: {}", e);
374                    }
375                    break;
376                }
377                res = &mut shutdown_task, if canceled => {
378                    match res {
379                        Ok(()) => {
380                            info!("Graceful AppState shutdown completed");
381                        }
382                        Err(e) => {
383                            warn!("Graceful AppState shutdown failed: {}", e);
384                            app_state.stop();
385                        }
386                    }
387                    break;
388                }
389                signal = &mut shutdown_signal, if !canceled => {
390                    match signal {
391                        Ok(ShutdownSignal::CtrlC) => info!("SIGINT (Ctrl-C) received"),
392                        #[cfg(unix)]
393                        Ok(ShutdownSignal::SigTerm) => info!("SIGTERM received"),
394                        Err(e) => {
395                            warn!("Shutdown signal handler failed: {}", e);
396                            break;
397                        }
398                    }
399                    if graceful_shutdown {
400                        let app_state = app_state.clone();
401                        shutdown_task.set(async move { app_state.graceful_stop(graceful_shutdown_timeout).await }.boxed());
402                        *cancel_timeout = tokio::time::sleep(tokio::time::Duration::from_secs(graceful_shutdown_timeout)).boxed();
403                        canceled = true;
404                    } else {
405                        break;
406                    }
407                }
408                _ = &mut cancel_timeout => {
409                    warn!("Shutdown timeout reached, forcing exit");
410                    break;
411                }
412            }
413        }
414        Ok(())
415    }
416}
417
418impl Default for MainBuilder {
419    fn default() -> Self {
420        Self::from_cli(Cli::parse())
421    }
422}
423
424pub async fn index() -> impl IntoResponse {
425    match std::fs::read_to_string("static/index.html") {
426        Ok(content) => (StatusCode::OK, [("content-type", "text/html")], content).into_response(),
427        Err(_) => (StatusCode::NOT_FOUND, "Index not found").into_response(),
428    }
429}
430
431enum ShutdownSignal {
432    CtrlC,
433    #[cfg(unix)]
434    SigTerm,
435}
436
437async fn shutdown_signal() -> Result<ShutdownSignal> {
438    #[cfg(unix)]
439    {
440        let mut sigterm =
441            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
442        tokio::select! {
443            result = signal::ctrl_c() => {
444                result?;
445                Ok(ShutdownSignal::CtrlC)
446            }
447            _ = sigterm.recv() => Ok(ShutdownSignal::SigTerm),
448        }
449    }
450    #[cfg(not(unix))]
451    {
452        signal::ctrl_c().await?;
453        Ok(ShutdownSignal::CtrlC)
454    }
455}
active_call/main_builder.rs

active_call/
main_builder.rs
