use std::{
|
collections::HashMap,
|
env,
|
io::{BufRead, BufReader, Read, Write},
|
net::{TcpListener, TcpStream},
|
process::{Child, Command, Stdio},
|
sync::{Arc, Mutex},
|
thread,
|
time::{Duration, SystemTime, UNIX_EPOCH},
|
};
|
|
use anyhow::{Context, Result, anyhow};
|
use serde::Deserialize;
|
use serde_json::{Value, json};
|
use tracing::{info, warn};
|
|
const DEFAULT_BIND_ADDR: &str = "127.0.0.1:18080";
|
const HTTP_READ_LIMIT_BYTES: usize = 1024 * 1024;
|
const DEFAULT_START_READY_TIMEOUT_MS: u64 = 10_000;
|
|
pub fn service_mode_enabled() -> bool {
|
env::args().any(|arg| arg == "service" || arg == "--service")
|
|| bool_env("CV_HELPER_SERVICE_ENABLED", false)
|
}
|
|
pub async fn run_service() -> Result<()> {
|
let config = ServiceConfig::from_env()?;
|
let listener = TcpListener::bind(&config.bind_addr)
|
.with_context(|| format!("failed to bind helper service {}", config.bind_addr))?;
|
let state = Arc::new(ServiceState {
|
sessions: Mutex::new(HashMap::new()),
|
config: config.clone(),
|
});
|
info!(
|
bind_addr = %config.bind_addr,
|
auth_profiles = ?config.auth_profiles.keys().collect::<Vec<_>>(),
|
"combrabo voice helper service started"
|
);
|
|
for stream in listener.incoming() {
|
match stream {
|
Ok(stream) => {
|
let state = state.clone();
|
thread::spawn(move || {
|
if let Err(error) = handle_connection(stream, state) {
|
warn!(error = %error, "helper service connection failed");
|
}
|
});
|
}
|
Err(error) => warn!(error = %error, "helper service accept failed"),
|
}
|
}
|
Ok(())
|
}
|
|
fn handle_connection(mut stream: TcpStream, state: Arc<ServiceState>) -> Result<()> {
|
stream.set_read_timeout(Some(Duration::from_secs(5)))?;
|
let request = HttpRequest::read(&mut stream)?;
|
let response = route(request, state);
|
response.write(&mut stream)?;
|
Ok(())
|
}
|
|
fn route(request: HttpRequest, state: Arc<ServiceState>) -> HttpResponse {
|
if request.method == "GET"
|
&& (request.path == "/health" || request.path == "/internal/combrabo-voice/health")
|
{
|
return ok(json!({
|
"status": "UP",
|
"service": "lm-livekit-helper",
|
"version": env!("CARGO_PKG_VERSION"),
|
"mode": "service"
|
}));
|
}
|
if !authorized(&request, &state.config) {
|
return error(
|
401,
|
2001003404,
|
"HELPER_AUTH_FAILED",
|
"auth",
|
false,
|
trace_id(&request),
|
);
|
}
|
if request.method == "POST" && request.path == "/internal/combrabo-voice/sessions/start" {
|
return start_session(&request, state);
|
}
|
if request.method == "GET"
|
&& request
|
.path
|
.starts_with("/internal/combrabo-voice/sessions/")
|
{
|
let call_id = request
|
.path
|
.trim_start_matches("/internal/combrabo-voice/sessions/")
|
.trim_matches('/');
|
return get_session(call_id, state);
|
}
|
if request.method == "POST"
|
&& request
|
.path
|
.starts_with("/internal/combrabo-voice/sessions/")
|
&& request.path.ends_with("/stop")
|
{
|
let call_id = request
|
.path
|
.trim_start_matches("/internal/combrabo-voice/sessions/")
|
.trim_end_matches("/stop")
|
.trim_matches('/');
|
return stop_session(call_id, &request, state);
|
}
|
error(
|
404,
|
2001003404,
|
"HELPER_ROUTE_NOT_FOUND",
|
"helper_http",
|
false,
|
trace_id(&request),
|
)
|
}
|
|
fn start_session(request: &HttpRequest, state: Arc<ServiceState>) -> HttpResponse {
|
let start_req: SessionStartRequest = match serde_json::from_slice(&request.body) {
|
Ok(value) => value,
|
Err(_) => {
|
return error(
|
400,
|
2001003404,
|
"HELPER_REQUEST_INVALID",
|
"validate",
|
false,
|
trace_id(request),
|
);
|
}
|
};
|
let trace_id = start_req.trace_id.clone().or_else(|| trace_id(request));
|
if start_req.call_id.trim().is_empty() || start_req.runtime_session_nonce.trim().is_empty() {
|
return error(
|
400,
|
2001003404,
|
"HELPER_REQUEST_INVALID",
|
"validate",
|
false,
|
trace_id,
|
);
|
}
|
let auth_profile = start_req
|
.auth_profile
|
.as_deref()
|
.filter(|value| !value.trim().is_empty())
|
.unwrap_or("default")
|
.to_string();
|
let profile = match state.config.auth_profiles.get(&auth_profile) {
|
Some(value) => value.clone(),
|
None => {
|
return error(
|
200,
|
2001003404,
|
"HELPER_AUTH_PROFILE_NOT_FOUND",
|
"auth",
|
true,
|
trace_id,
|
);
|
}
|
};
|
let mut sessions = match state.sessions.lock() {
|
Ok(value) => value,
|
Err(_) => {
|
return error(
|
200,
|
2001003404,
|
"HELPER_SESSION_LOCK_FAILED",
|
"runtime_ready",
|
true,
|
trace_id,
|
);
|
}
|
};
|
if let Some(existing) = sessions.get_mut(&start_req.call_id) {
|
let _ = refresh_child_status(existing);
|
if existing.runtime_session_nonce == start_req.runtime_session_nonce
|
&& existing.status != "STOPPED"
|
{
|
return ok(session_start_data(existing));
|
}
|
return error(
|
200,
|
2001003404,
|
"RUNTIME_SESSION_CONFLICT",
|
"runtime_ready",
|
true,
|
trace_id,
|
);
|
}
|
let child = match spawn_worker(&start_req, &profile, &state.config) {
|
Ok(value) => value,
|
Err(spawn_error) => {
|
warn!(
|
call_id = %start_req.call_id,
|
trace_id = ?trace_id,
|
error = %spawn_error,
|
"helper service failed to spawn worker"
|
);
|
return error(
|
200,
|
2001003404,
|
"RUNTIME_START_FAILED",
|
"runtime_ready",
|
true,
|
trace_id,
|
);
|
}
|
};
|
let runtime_session_id = format!("rt_{}", start_req.call_id);
|
let mut child = child;
|
let child_stdout = child.stdout.take();
|
let session = HelperSession {
|
call_id: start_req.call_id.clone(),
|
trace_id: trace_id.clone(),
|
runtime_session_nonce: start_req.runtime_session_nonce.clone(),
|
runtime_session_id,
|
event_callback_url: start_req
|
.event_callback
|
.as_ref()
|
.and_then(|value| value.url.clone())
|
.filter(|value| !value.trim().is_empty()),
|
event_callback_token: Some(profile.event_callback_token.clone()),
|
status: "STARTING".to_string(),
|
bot_participant_joined: false,
|
bot_track_ready: false,
|
first_audio_source: start_req
|
.audio
|
.as_ref()
|
.and_then(|value| value.first_audio_source.clone())
|
.unwrap_or_else(|| "fixed_greeting_tts".to_string()),
|
active_turn_id: None,
|
last_event_type: Some("helper_worker_started".to_string()),
|
last_event_at: now_millis(),
|
child: Some(child),
|
};
|
sessions.insert(start_req.call_id.clone(), session);
|
drop(sessions);
|
|
if let Some(stdout) = child_stdout {
|
spawn_worker_stdout_monitor(start_req.call_id.clone(), state.clone(), stdout);
|
}
|
|
match wait_for_session_ready(&start_req.call_id, state, &trace_id) {
|
WaitReadyResult::Ready(response) => ok(response),
|
WaitReadyResult::Failed(response) => response,
|
}
|
}
|
|
fn get_session(call_id: &str, state: Arc<ServiceState>) -> HttpResponse {
|
let mut sessions = match state.sessions.lock() {
|
Ok(value) => value,
|
Err(_) => {
|
return error(
|
200,
|
2001003404,
|
"HELPER_SESSION_LOCK_FAILED",
|
"runtime_ready",
|
true,
|
None,
|
);
|
}
|
};
|
let Some(session) = sessions.get_mut(call_id) else {
|
return error(
|
200,
|
2001003404,
|
"CALL_NOT_FOUND",
|
"runtime_ready",
|
false,
|
None,
|
);
|
};
|
let _ = refresh_child_status(session);
|
ok(json!({
|
"callId": &session.call_id,
|
"status": &session.status,
|
"botParticipantJoined": session.bot_participant_joined,
|
"botTrackReady": session.bot_track_ready,
|
"activeTurnId": session.active_turn_id.as_ref(),
|
"lastEventType": session.last_event_type.as_ref(),
|
"lastEventAt": session.last_event_at.to_string(),
|
}))
|
}
|
|
fn stop_session(call_id: &str, request: &HttpRequest, state: Arc<ServiceState>) -> HttpResponse {
|
let stop_req: SessionStopRequest = serde_json::from_slice(&request.body).unwrap_or_default();
|
let mut sessions = match state.sessions.lock() {
|
Ok(value) => value,
|
Err(_) => {
|
return error(
|
200,
|
2001003404,
|
"HELPER_SESSION_LOCK_FAILED",
|
"runtime_stop_requested",
|
true,
|
trace_id(request),
|
);
|
}
|
};
|
let Some(mut session) = sessions.remove(call_id) else {
|
return ok(json!({
|
"callId": call_id,
|
"status": "STOPPED",
|
"alreadyStopped": true,
|
}));
|
};
|
if stop_req
|
.runtime_session_nonce
|
.as_ref()
|
.is_some_and(|value| *value != session.runtime_session_nonce)
|
{
|
sessions.insert(call_id.to_string(), session);
|
return error(
|
200,
|
2001003404,
|
"RUNTIME_SESSION_MISMATCH",
|
"runtime_stop_requested",
|
true,
|
trace_id(request),
|
);
|
}
|
if let Some(child) = session.child.as_mut() {
|
let _ = child.kill();
|
let _ = child.wait();
|
}
|
session.status = "STOPPED".to_string();
|
ok(json!({
|
"callId": session.call_id,
|
"status": "STOPPED",
|
"alreadyStopped": false,
|
}))
|
}
|
|
fn spawn_worker(
|
req: &SessionStartRequest,
|
profile: &AuthProfile,
|
config: &ServiceConfig,
|
) -> Result<Child> {
|
let exe = env::current_exe().context("failed to resolve helper executable")?;
|
let mut command = Command::new(exe);
|
command
|
.env_remove("CV_HELPER_SERVICE_ENABLED")
|
.env("CV_CALL_ID", &req.call_id)
|
.env(
|
"CV_TRACE_ID",
|
req.trace_id
|
.clone()
|
.unwrap_or_else(|| format!("trace_{}", req.call_id)),
|
)
|
.env("CV_RUNTIME_CALL_ID", &req.call_id)
|
.env(
|
"CV_RUNTIME_TRACE_ID",
|
req.trace_id
|
.clone()
|
.unwrap_or_else(|| format!("trace_{}", req.call_id)),
|
)
|
.env("CV_RUNTIME_SESSION_NONCE", &req.runtime_session_nonce)
|
.env("CV_LIVEKIT_URL", &req.livekit.url)
|
.env("CV_LIVEKIT_ROOM_ID", &req.livekit.room_id)
|
.env("CV_LIVEKIT_BOT_TOKEN", &req.livekit.bot_token)
|
.env(
|
"CV_LIVEKIT_BOT_PARTICIPANT_IDENTITY",
|
&req.livekit.bot_participant_identity,
|
)
|
.env("CV_RUNTIME_TURN_BRIDGE_TOKEN", &profile.turn_bridge_token)
|
.stdout(Stdio::piped())
|
.stderr(Stdio::inherit());
|
if let Some(value) = &req.livekit.user_participant_identity {
|
command.env("CV_LIVEKIT_USER_PARTICIPANT_IDENTITY", value);
|
}
|
if let Some(turn_bridge) = &req.turn_bridge {
|
command.env("CV_RUNTIME_TURN_BRIDGE_URL", &turn_bridge.url);
|
}
|
if let Some(runtime) = &req.runtime {
|
if let Some(value) = &runtime.turn_artifact_dir {
|
command.env("CV_RUNTIME_TURN_ARTIFACT_DIR", value);
|
}
|
if let Some(value) = runtime.vad_rms_threshold {
|
command.env("CV_VAD_RMS_THRESHOLD", value.to_string());
|
}
|
if let Some(value) = runtime.vad_peak_threshold {
|
command.env("CV_VAD_PEAK_THRESHOLD", value.to_string());
|
}
|
if let Some(value) = runtime.vad_start_frames {
|
command.env("CV_VAD_START_FRAMES", value.to_string());
|
}
|
if let Some(value) = runtime.vad_end_silence_ms {
|
command.env("CV_VAD_END_SILENCE_MS", value.to_string());
|
}
|
if let Some(value) = runtime.vad_min_speech_ms {
|
command.env("CV_VAD_MIN_SPEECH_MS", value.to_string());
|
}
|
if let Some(value) = runtime.vad_max_turn_ms {
|
command.env("CV_VAD_MAX_TURN_MS", value.to_string());
|
}
|
if let Some(value) = runtime.vad_initial_ignore_ms {
|
command.env("CV_VAD_INITIAL_IGNORE_MS", value.to_string());
|
}
|
if let Some(value) = runtime.vad_post_greeting_delay_ms {
|
command.env("CV_VAD_POST_GREETING_DELAY_MS", value.to_string());
|
}
|
if let Some(value) = runtime.asr_streaming_enabled {
|
command.env("CV_RUNTIME_ASR_STREAM_ENABLED", value.to_string());
|
}
|
if let Some(value) = &runtime.asr_stream_url {
|
command.env("CV_RUNTIME_ASR_STREAM_URL", value);
|
}
|
if let Some(true) = runtime.audio_debug_dump_enabled {
|
if let Some(value) = &config.audio_debug_dump_dir {
|
command.env("CV_AUDIO_DEBUG_DUMP_DIR", value);
|
}
|
}
|
}
|
if let Some(audio) = &req.audio {
|
if let Some(first_audio_source) = &audio.first_audio_source {
|
command.env("CV_FIRST_AUDIO_SOURCE", first_audio_source);
|
command.env("CV_GREETING_SOURCE", first_audio_source);
|
}
|
if let Some(greeting_audio) = &audio.greeting_audio {
|
match greeting_audio.r#type.as_deref() {
|
Some("remote_url") => {
|
if let Some(value) = &greeting_audio.path_ref {
|
command.env("CV_GREETING_AUDIO_URL", value);
|
}
|
}
|
_ => {
|
if let Some(value) = &greeting_audio.path_ref {
|
command.env("CV_GREETING_AUDIO_FILE", value);
|
}
|
}
|
}
|
}
|
}
|
command.spawn().map_err(|error| anyhow!(error))
|
}
|
|
fn spawn_worker_stdout_monitor(
|
call_id: String,
|
state: Arc<ServiceState>,
|
stdout: std::process::ChildStdout,
|
) {
|
thread::spawn(move || {
|
let reader = BufReader::new(stdout);
|
for line in reader.lines() {
|
let Ok(line) = line else {
|
continue;
|
};
|
println!("{line}");
|
if let Ok(value) = serde_json::from_str::<Value>(&line) {
|
apply_worker_event(&call_id, &state, &value);
|
}
|
}
|
});
|
}
|
|
fn apply_worker_event(call_id: &str, state: &Arc<ServiceState>, value: &Value) {
|
if value.get("type").and_then(Value::as_str) != Some("cv_activity") {
|
return;
|
}
|
if value.get("callId").and_then(Value::as_str) != Some(call_id) {
|
return;
|
}
|
let event_name = value
|
.get("eventName")
|
.and_then(Value::as_str)
|
.unwrap_or("worker_event");
|
let result = value.get("result").and_then(Value::as_str).unwrap_or("ok");
|
let callback = {
|
let Ok(mut sessions) = state.sessions.lock() else {
|
return;
|
};
|
let Some(session) = sessions.get_mut(call_id) else {
|
return;
|
};
|
session.last_event_type = Some(event_name.to_string());
|
session.last_event_at = now_millis();
|
let callback = build_event_callback_dispatch(session, value);
|
if result != "ok" {
|
session.status = "FAILED".to_string();
|
session.bot_participant_joined = false;
|
session.bot_track_ready = false;
|
} else {
|
match event_name {
|
"helper_worker_started" => {
|
session.status = "STARTING".to_string();
|
}
|
"bot_participant_joined" => {
|
session.bot_participant_joined = true;
|
}
|
"bot_track_ready" => {
|
session.bot_participant_joined = true;
|
session.bot_track_ready = true;
|
session.status = "STARTED".to_string();
|
}
|
_ => {}
|
}
|
}
|
callback
|
};
|
if let Some(callback) = callback {
|
dispatch_event_callback(callback);
|
}
|
}
|
|
fn build_event_callback_dispatch(
|
session: &HelperSession,
|
value: &Value,
|
) -> Option<EventCallbackDispatch> {
|
let url = session
|
.event_callback_url
|
.as_ref()
|
.filter(|value| !value.trim().is_empty())?
|
.clone();
|
let token = session
|
.event_callback_token
|
.as_ref()
|
.filter(|value| !value.trim().is_empty())?
|
.clone();
|
Some(EventCallbackDispatch {
|
url,
|
token,
|
call_id: session.call_id.clone(),
|
trace_id: session.trace_id.clone(),
|
runtime_session_nonce: session.runtime_session_nonce.clone(),
|
payload: value.clone(),
|
})
|
}
|
|
fn dispatch_event_callback(callback: EventCallbackDispatch) {
|
thread::spawn(move || {
|
let event_name = callback
|
.payload
|
.get("eventName")
|
.and_then(Value::as_str)
|
.unwrap_or("worker_event")
|
.to_string();
|
match post_event_callback(&callback) {
|
Ok(status) if (200..300).contains(&status) => {
|
info!(
|
call_id = %callback.call_id,
|
trace_id = ?callback.trace_id,
|
event_name = %event_name,
|
status = status,
|
"helper event callback delivered"
|
);
|
}
|
Ok(status) => {
|
warn!(
|
call_id = %callback.call_id,
|
trace_id = ?callback.trace_id,
|
event_name = %event_name,
|
status = status,
|
"helper event callback rejected"
|
);
|
}
|
Err(error) => {
|
warn!(
|
call_id = %callback.call_id,
|
trace_id = ?callback.trace_id,
|
event_name = %event_name,
|
error = %error,
|
"helper event callback failed"
|
);
|
}
|
}
|
});
|
}
|
|
fn post_event_callback(callback: &EventCallbackDispatch) -> Result<u16> {
|
let (host, path) = parse_http_url(&callback.url)?;
|
let body = serde_json::to_vec(&callback.payload)?;
|
let mut stream = TcpStream::connect(&host)
|
.with_context(|| format!("failed to connect callback host {host}"))?;
|
stream.set_read_timeout(Some(Duration::from_secs(3)))?;
|
stream.set_write_timeout(Some(Duration::from_secs(3)))?;
|
let trace_header = callback.trace_id.clone().unwrap_or_default();
|
let request = format!(
|
"POST {path} HTTP/1.1\r\n\
|
Host: {host}\r\n\
|
Authorization: Bearer {}\r\n\
|
X-CV-Runtime-Token: {}\r\n\
|
X-CV-Call-Id: {}\r\n\
|
X-CV-Trace-Id: {}\r\n\
|
X-CV-Runtime-Session-Nonce: {}\r\n\
|
Content-Type: application/json\r\n\
|
Content-Length: {}\r\n\
|
Connection: close\r\n\
|
\r\n",
|
callback.token,
|
callback.token,
|
callback.call_id,
|
trace_header,
|
callback.runtime_session_nonce,
|
body.len()
|
);
|
stream.write_all(request.as_bytes())?;
|
stream.write_all(&body)?;
|
stream.flush()?;
|
let mut response = String::new();
|
stream.read_to_string(&mut response)?;
|
response
|
.lines()
|
.next()
|
.and_then(|line| line.split_whitespace().nth(1))
|
.and_then(|value| value.parse::<u16>().ok())
|
.ok_or_else(|| anyhow!("callback response status missing"))
|
}
|
|
fn parse_http_url(url: &str) -> Result<(String, String)> {
|
let Some(rest) = url.strip_prefix("http://") else {
|
return Err(anyhow!("only http callback url is supported"));
|
};
|
let (host, path) = match rest.split_once('/') {
|
Some((host, path)) => (host, format!("/{path}")),
|
None => (rest, "/".to_string()),
|
};
|
if host.trim().is_empty() {
|
return Err(anyhow!("callback host missing"));
|
}
|
Ok((host.to_string(), path))
|
}
|
|
enum WaitReadyResult {
|
Ready(Value),
|
Failed(HttpResponse),
|
}
|
|
fn wait_for_session_ready(
|
call_id: &str,
|
state: Arc<ServiceState>,
|
trace_id: &Option<String>,
|
) -> WaitReadyResult {
|
let deadline = SystemTime::now()
|
.checked_add(Duration::from_millis(state.config.start_ready_timeout_ms))
|
.unwrap_or_else(SystemTime::now);
|
loop {
|
let snapshot = {
|
let mut sessions = match state.sessions.lock() {
|
Ok(value) => value,
|
Err(_) => {
|
return WaitReadyResult::Failed(error(
|
200,
|
2001003404,
|
"HELPER_SESSION_LOCK_FAILED",
|
"runtime_ready",
|
true,
|
trace_id.clone(),
|
));
|
}
|
};
|
let Some(session) = sessions.get_mut(call_id) else {
|
return WaitReadyResult::Failed(error(
|
200,
|
2001003404,
|
"CALL_NOT_FOUND",
|
"runtime_ready",
|
false,
|
trace_id.clone(),
|
));
|
};
|
let _ = refresh_child_status(session);
|
(
|
session.status.clone(),
|
session.bot_participant_joined,
|
session.bot_track_ready,
|
session_start_data(session),
|
)
|
};
|
if snapshot.0 == "STARTED" && snapshot.1 && snapshot.2 {
|
return WaitReadyResult::Ready(snapshot.3);
|
}
|
if snapshot.0 == "FAILED" || snapshot.0 == "STOPPED" {
|
return WaitReadyResult::Failed(error(
|
200,
|
2001003404,
|
"RUNTIME_START_FAILED",
|
"runtime_ready",
|
true,
|
trace_id.clone(),
|
));
|
}
|
if SystemTime::now() >= deadline {
|
mark_session_start_timeout(call_id, &state);
|
return WaitReadyResult::Failed(error(
|
200,
|
2001003404,
|
"RUNTIME_START_TIMEOUT",
|
"runtime_ready",
|
true,
|
trace_id.clone(),
|
));
|
}
|
thread::sleep(Duration::from_millis(50));
|
}
|
}
|
|
fn mark_session_start_timeout(call_id: &str, state: &Arc<ServiceState>) {
|
let Ok(mut sessions) = state.sessions.lock() else {
|
return;
|
};
|
let Some(session) = sessions.get_mut(call_id) else {
|
return;
|
};
|
if let Some(child) = session.child.as_mut() {
|
let _ = child.kill();
|
let _ = child.wait();
|
}
|
session.status = "FAILED".to_string();
|
session.bot_participant_joined = false;
|
session.bot_track_ready = false;
|
session.last_event_type = Some("worker_ready_timeout".to_string());
|
session.last_event_at = now_millis();
|
}
|
|
fn refresh_child_status(session: &mut HelperSession) -> Result<()> {
|
if let Some(child) = session.child.as_mut() {
|
if let Some(status) = child.try_wait()? {
|
session.status = if status.success() {
|
"STOPPED".to_string()
|
} else {
|
"FAILED".to_string()
|
};
|
session.bot_participant_joined = false;
|
session.bot_track_ready = false;
|
session.last_event_type = Some("worker_exited".to_string());
|
session.last_event_at = now_millis();
|
}
|
}
|
Ok(())
|
}
|
|
fn session_start_data(session: &HelperSession) -> Value {
|
json!({
|
"callId": &session.call_id,
|
"status": &session.status,
|
"runtimeSessionId": &session.runtime_session_id,
|
"botParticipantJoined": session.bot_participant_joined,
|
"botTrackReady": session.bot_track_ready,
|
"firstAudioSource": &session.first_audio_source,
|
})
|
}
|
|
fn authorized(request: &HttpRequest, config: &ServiceConfig) -> bool {
|
let Some(authorization) = request.headers.get("authorization") else {
|
return false;
|
};
|
let Some(token) = authorization.strip_prefix("Bearer ") else {
|
return false;
|
};
|
config
|
.auth_profiles
|
.values()
|
.any(|profile| constant_time_eq(token.as_bytes(), profile.helper_auth_token.as_bytes()))
|
}
|
|
fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
|
if left.len() != right.len() {
|
return false;
|
}
|
left.iter()
|
.zip(right.iter())
|
.fold(0u8, |acc, (l, r)| acc | (l ^ r))
|
== 0
|
}
|
|
fn trace_id(request: &HttpRequest) -> Option<String> {
|
request.headers.get("x-voice-trace-id").cloned()
|
}
|
|
fn ok(data: Value) -> HttpResponse {
|
HttpResponse::json(
|
200,
|
json!({
|
"code": 0,
|
"msg": "",
|
"data": data
|
}),
|
)
|
}
|
|
fn error(
|
http_status: u16,
|
code: i64,
|
reason_code: &str,
|
stage: &str,
|
retryable: bool,
|
trace_id: Option<String>,
|
) -> HttpResponse {
|
HttpResponse::json(
|
http_status,
|
json!({
|
"code": code,
|
"msg": reason_code,
|
"data": {
|
"reasonCode": reason_code,
|
"stage": stage,
|
"retryable": retryable,
|
"traceId": trace_id
|
}
|
}),
|
)
|
}
|
|
fn now_millis() -> u128 {
|
SystemTime::now()
|
.duration_since(UNIX_EPOCH)
|
.unwrap_or_default()
|
.as_millis()
|
}
|
|
#[derive(Clone)]
|
struct ServiceConfig {
|
bind_addr: String,
|
audio_debug_dump_dir: Option<String>,
|
start_ready_timeout_ms: u64,
|
auth_profiles: HashMap<String, AuthProfile>,
|
}
|
|
impl ServiceConfig {
|
fn from_env() -> Result<Self> {
|
let bind_addr = env::var("CV_HELPER_SERVICE_BIND")
|
.ok()
|
.filter(|value| !value.trim().is_empty())
|
.unwrap_or_else(|| DEFAULT_BIND_ADDR.to_string());
|
let mut auth_profiles = HashMap::new();
|
for profile in ["default", "local-dev", "dev"] {
|
if let Some(auth_profile) = AuthProfile::from_env(profile) {
|
auth_profiles.insert(profile.to_string(), auth_profile);
|
}
|
}
|
if auth_profiles.is_empty() {
|
return Err(anyhow!("missing helper auth profile config"));
|
}
|
Ok(Self {
|
bind_addr,
|
audio_debug_dump_dir: env::var("CV_AUDIO_DEBUG_DUMP_DIR").ok(),
|
start_ready_timeout_ms: env::var("CV_HELPER_SERVICE_START_READY_TIMEOUT_MS")
|
.ok()
|
.and_then(|value| value.trim().parse::<u64>().ok())
|
.filter(|value| *value > 0)
|
.unwrap_or(DEFAULT_START_READY_TIMEOUT_MS),
|
auth_profiles,
|
})
|
}
|
}
|
|
#[derive(Clone)]
|
struct AuthProfile {
|
helper_auth_token: String,
|
turn_bridge_token: String,
|
event_callback_token: String,
|
}
|
|
impl AuthProfile {
|
fn from_env(profile: &str) -> Option<Self> {
|
let suffix = profile
|
.chars()
|
.map(|ch| if ch == '-' { '_' } else { ch })
|
.collect::<String>()
|
.to_ascii_uppercase();
|
let helper_auth_token = env::var(format!("CV_HELPER_AUTH_TOKEN_{suffix}"))
|
.or_else(|_| env::var("CV_HELPER_AUTH_TOKEN"))
|
.ok()?;
|
let turn_bridge_token = env::var(format!("CV_TURN_BRIDGE_TOKEN_{suffix}"))
|
.or_else(|_| env::var("CV_RUNTIME_TURN_BRIDGE_TOKEN"))
|
.ok()?;
|
let event_callback_token = env::var(format!("CV_EVENT_CALLBACK_TOKEN_{suffix}"))
|
.or_else(|_| env::var("CV_RUNTIME_EVENT_CALLBACK_TOKEN"))
|
.unwrap_or_else(|_| turn_bridge_token.clone());
|
Some(Self {
|
helper_auth_token,
|
turn_bridge_token,
|
event_callback_token,
|
})
|
}
|
}
|
|
struct EventCallbackDispatch {
|
url: String,
|
token: String,
|
call_id: String,
|
trace_id: Option<String>,
|
runtime_session_nonce: String,
|
payload: Value,
|
}
|
|
struct ServiceState {
|
sessions: Mutex<HashMap<String, HelperSession>>,
|
config: ServiceConfig,
|
}
|
|
struct HelperSession {
|
call_id: String,
|
trace_id: Option<String>,
|
runtime_session_nonce: String,
|
runtime_session_id: String,
|
event_callback_url: Option<String>,
|
event_callback_token: Option<String>,
|
status: String,
|
bot_participant_joined: bool,
|
bot_track_ready: bool,
|
first_audio_source: String,
|
active_turn_id: Option<String>,
|
last_event_type: Option<String>,
|
last_event_at: u128,
|
child: Option<Child>,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct SessionStartRequest {
|
call_id: String,
|
trace_id: Option<String>,
|
runtime_session_nonce: String,
|
livekit: LiveKitStart,
|
turn_bridge: Option<TurnBridgeStart>,
|
event_callback: Option<EventCallbackStart>,
|
auth_profile: Option<String>,
|
audio: Option<AudioStart>,
|
runtime: Option<RuntimeStart>,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct LiveKitStart {
|
url: String,
|
room_id: String,
|
bot_token: String,
|
bot_participant_identity: String,
|
user_participant_identity: Option<String>,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct TurnBridgeStart {
|
url: String,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct EventCallbackStart {
|
url: Option<String>,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct AudioStart {
|
first_audio_source: Option<String>,
|
greeting_audio: Option<GreetingAudioStart>,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct GreetingAudioStart {
|
r#type: Option<String>,
|
path_ref: Option<String>,
|
}
|
|
#[derive(Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct RuntimeStart {
|
turn_artifact_dir: Option<String>,
|
vad_rms_threshold: Option<f64>,
|
vad_peak_threshold: Option<f64>,
|
vad_start_frames: Option<u32>,
|
vad_end_silence_ms: Option<u64>,
|
vad_min_speech_ms: Option<u64>,
|
vad_max_turn_ms: Option<u64>,
|
vad_initial_ignore_ms: Option<u64>,
|
vad_post_greeting_delay_ms: Option<u64>,
|
audio_debug_dump_enabled: Option<bool>,
|
asr_streaming_enabled: Option<bool>,
|
asr_stream_url: Option<String>,
|
}
|
|
#[derive(Default, Deserialize)]
|
#[serde(rename_all = "camelCase")]
|
struct SessionStopRequest {
|
runtime_session_nonce: Option<String>,
|
}
|
|
struct HttpRequest {
|
method: String,
|
path: String,
|
headers: HashMap<String, String>,
|
body: Vec<u8>,
|
}
|
|
impl HttpRequest {
|
fn read(stream: &mut TcpStream) -> Result<Self> {
|
let mut buffer = Vec::new();
|
let mut temp = [0u8; 4096];
|
let header_end;
|
loop {
|
let read = stream.read(&mut temp)?;
|
if read == 0 {
|
return Err(anyhow!("connection closed before request header"));
|
}
|
buffer.extend_from_slice(&temp[..read]);
|
if buffer.len() > HTTP_READ_LIMIT_BYTES {
|
return Err(anyhow!("http request too large"));
|
}
|
if let Some(index) = find_header_end(&buffer) {
|
header_end = index;
|
break;
|
}
|
}
|
let header_text = String::from_utf8_lossy(&buffer[..header_end]).to_string();
|
let mut lines = header_text.split("\r\n");
|
let request_line = lines
|
.next()
|
.ok_or_else(|| anyhow!("missing request line"))?;
|
let parts = request_line.split_whitespace().collect::<Vec<_>>();
|
if parts.len() < 2 {
|
return Err(anyhow!("invalid request line"));
|
}
|
let method = parts[0].to_string();
|
let path = parts[1].split('?').next().unwrap_or(parts[1]).to_string();
|
let mut headers = HashMap::new();
|
for line in lines {
|
if let Some((key, value)) = line.split_once(':') {
|
headers.insert(key.trim().to_ascii_lowercase(), value.trim().to_string());
|
}
|
}
|
let content_length = headers
|
.get("content-length")
|
.and_then(|value| value.parse::<usize>().ok())
|
.unwrap_or(0);
|
let body_start = header_end + 4;
|
let mut body = buffer[body_start..].to_vec();
|
while body.len() < content_length {
|
let read = stream.read(&mut temp)?;
|
if read == 0 {
|
break;
|
}
|
body.extend_from_slice(&temp[..read]);
|
if body.len() > HTTP_READ_LIMIT_BYTES {
|
return Err(anyhow!("http request body too large"));
|
}
|
}
|
body.truncate(content_length);
|
Ok(Self {
|
method,
|
path,
|
headers,
|
body,
|
})
|
}
|
}
|
|
struct HttpResponse {
|
status: u16,
|
body: Vec<u8>,
|
}
|
|
impl HttpResponse {
|
fn json(status: u16, body: Value) -> Self {
|
Self {
|
status,
|
body: serde_json::to_vec(&body).unwrap_or_else(|_| b"{}".to_vec()),
|
}
|
}
|
|
fn write(self, stream: &mut TcpStream) -> Result<()> {
|
let reason = match self.status {
|
200 => "OK",
|
400 => "Bad Request",
|
401 => "Unauthorized",
|
404 => "Not Found",
|
_ => "Internal Server Error",
|
};
|
write!(
|
stream,
|
"HTTP/1.1 {} {}\r\nContent-Type: application/json;charset=utf-8\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
|
self.status,
|
reason,
|
self.body.len()
|
)?;
|
stream.write_all(&self.body)?;
|
Ok(())
|
}
|
}
|
|
fn find_header_end(buffer: &[u8]) -> Option<usize> {
|
buffer.windows(4).position(|window| window == b"\r\n\r\n")
|
}
|
|
fn bool_env(key: &str, default_value: bool) -> bool {
|
match env::var(key) {
|
Ok(value) => match value.trim().to_ascii_lowercase().as_str() {
|
"1" | "true" | "yes" | "on" => true,
|
"0" | "false" | "no" | "off" => false,
|
_ => default_value,
|
},
|
Err(_) => default_value,
|
}
|
}
|
