diff options
Diffstat (limited to 'src/api/server.rs')
| -rw-r--r-- | src/api/server.rs | 769 |
1 files changed, 769 insertions, 0 deletions
diff --git a/src/api/server.rs b/src/api/server.rs new file mode 100644 index 00000000..2f9a423e --- /dev/null +++ b/src/api/server.rs @@ -0,0 +1,769 @@ +use std::fs; +use std::io::{BufRead, BufReader, Read, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use tracing::{debug, error, info, warn}; + +use crate::api::schema::{ + ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse, +}; +use crate::api::subscriptions::ActiveSubscription; +use crate::api::wait::wait_for_output; +use crate::api::{request_changes_ui, socket_path, ApiRequestMessage, ApiRequestSender, EventHub}; + +const SOCKET_PERMISSION_MODE: u32 = 0o600; +pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100); +pub(super) const APP_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5); +const INITIAL_REQUEST_TIMEOUT: Duration = Duration::from_secs(5); +const STREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(5); + +pub struct ServerHandle { + _thread: std::thread::JoinHandle<()>, + path: PathBuf, + running: Arc<AtomicBool>, +} + +impl Drop for ServerHandle { + fn drop(&mut self) { + self.running.store(false, Ordering::Relaxed); + + if let Err(err) = fs::remove_file(&self.path) { + if err.kind() != std::io::ErrorKind::NotFound { + warn!(path = %self.path.display(), err = %err, "failed to remove api socket on shutdown"); + } + } + } +} + +pub fn start_server( + api_tx: ApiRequestSender, + event_hub: EventHub, +) -> std::io::Result<ServerHandle> { + let path = socket_path(); + prepare_socket_path(&path)?; + + let listener = UnixListener::bind(&path)?; + restrict_socket_permissions(&path)?; + info!(path = %path.display(), "api server listening"); + + let running = Arc::new(AtomicBool::new(true)); + let listener_running = Arc::clone(&running); + let thread = std::thread::spawn(move || { + for stream in listener.incoming() { + match stream { + Ok(stream) => { + let api_tx = api_tx.clone(); + let event_hub = event_hub.clone(); + let connection_running = Arc::clone(&listener_running); + std::thread::spawn(move || { + if let Err(err) = + handle_connection(stream, &api_tx, &event_hub, &connection_running) + { + warn!(err = %err, "api connection failed"); + } + }); + } + Err(err) => { + error!(err = %err, "api listener accept failed"); + break; + } + } + } + debug!("api server thread exiting"); + }); + + Ok(ServerHandle { + _thread: thread, + path, + running, + }) +} + +fn prepare_socket_path(path: &Path) -> std::io::Result<()> { + crate::ipc::prepare_socket_path(path, |path| { + format!( + "herdr is already running (socket busy at {})", + path.display() + ) + }) +} + +fn restrict_socket_permissions(path: &Path) -> std::io::Result<()> { + crate::ipc::restrict_socket_permissions(path, SOCKET_PERMISSION_MODE) +} + +fn handle_connection( + mut stream: UnixStream, + api_tx: &ApiRequestSender, + event_hub: &EventHub, + running: &Arc<AtomicBool>, +) -> std::io::Result<()> { + stream.set_read_timeout(Some(INITIAL_REQUEST_TIMEOUT))?; + stream.set_write_timeout(Some(STREAM_WRITE_TIMEOUT))?; + + let mut line = String::new(); + { + let mut reader = BufReader::new(&stream); + let read = reader.read_line(&mut line)?; + if read == 0 { + return Ok(()); + } + } + + stream.set_read_timeout(None)?; + + let line = line.trim(); + if line.is_empty() { + return Ok(()); + } + + let request = match serde_json::from_str::<Request>(line) { + Ok(request) => request, + Err(err) => { + write_json_line_allow_disconnect( + &mut stream, + &ErrorResponse { + id: String::new(), + error: ErrorBody { + code: "invalid_request".into(), + message: format!("invalid request: {err}"), + }, + }, + )?; + return Ok(()); + } + }; + + let request_id = request.id.clone(); + let method = api_method_name(&request.method); + let changes_ui = request_changes_ui(&request); + crate::logging::api_request_started(&request_id, method, changes_ui); + + match request.method { + Method::EventsSubscribe(params) => { + let result = stream_subscriptions( + stream, + request_id.clone(), + params, + api_tx, + event_hub, + running, + ); + match &result { + Ok(()) => crate::logging::api_request_completed( + &request_id, + method, + "stream_closed", + changes_ui, + ), + Err(err) => { + crate::logging::api_request_failed(&request_id, method, &err.to_string()) + } + } + result + } + Method::PaneWaitForOutput(params) => { + let Some(response) = + wait_for_output(request_id.clone(), params, &mut stream, api_tx, running)? + else { + crate::logging::api_request_completed( + &request_id, + method, + "client_disconnected", + changes_ui, + ); + return Ok(()); + }; + let result = write_text_line_allow_disconnect(&mut stream, &response); + match &result { + Ok(()) => crate::logging::api_request_completed( + &request_id, + method, + api_response_outcome(&response), + changes_ui, + ), + Err(err) => { + crate::logging::api_request_failed(&request_id, method, &err.to_string()) + } + } + result + } + method_body => { + let response = handle_request( + Request { + id: request_id.clone(), + method: method_body, + }, + api_tx, + ); + let result = write_text_line_allow_disconnect(&mut stream, &response); + match &result { + Ok(()) => crate::logging::api_request_completed( + &request_id, + method, + api_response_outcome(&response), + changes_ui, + ), + Err(err) => { + crate::logging::api_request_failed(&request_id, method, &err.to_string()) + } + } + result + } + } +} + +fn handle_request(request: Request, api_tx: &ApiRequestSender) -> String { + match request.method { + Method::Ping(_) => serde_json::to_string(&SuccessResponse { + id: request.id, + result: ResponseResult::Pong { + version: env!("CARGO_PKG_VERSION").into(), + protocol: crate::protocol::PROTOCOL_VERSION, + }, + }) + .unwrap_or_else(|_| { + r#"{"id":"","error":{"code":"internal_error","message":"failed to encode response"}}"# + .to_string() + }), + _ => dispatch_to_app(request, api_tx), + } +} + +fn api_method_name(method: &Method) -> &'static str { + match method { + Method::Ping(_) => "ping", + Method::ServerStop(_) => "server.stop", + Method::ServerReloadConfig(_) => "server.reload_config", + Method::WorkspaceCreate(_) => "workspace.create", + Method::WorkspaceList(_) => "workspace.list", + Method::WorkspaceGet(_) => "workspace.get", + Method::WorkspaceFocus(_) => "workspace.focus", + Method::WorkspaceRename(_) => "workspace.rename", + Method::WorkspaceClose(_) => "workspace.close", + Method::TabCreate(_) => "tab.create", + Method::TabList(_) => "tab.list", + Method::TabGet(_) => "tab.get", + Method::TabFocus(_) => "tab.focus", + Method::TabRename(_) => "tab.rename", + Method::TabClose(_) => "tab.close", + Method::AgentList(_) => "agent.list", + Method::AgentGet(_) => "agent.get", + Method::AgentRead(_) => "agent.read", + Method::AgentSend(_) => "agent.send", + Method::AgentRename(_) => "agent.rename", + Method::AgentFocus(_) => "agent.focus", + Method::AgentStart(_) => "agent.start", + Method::PaneSplit(_) => "pane.split", + Method::PaneList(_) => "pane.list", + Method::PaneGet(_) => "pane.get", + Method::PaneRename(_) => "pane.rename", + Method::PaneSendText(_) => "pane.send_text", + Method::PaneSendKeys(_) => "pane.send_keys", + Method::PaneSendInput(_) => "pane.send_input", + Method::PaneRead(_) => "pane.read", + Method::PaneReportAgent(_) => "pane.report_agent", + Method::PaneClearAgentAuthority(_) => "pane.clear_agent_authority", + Method::PaneReleaseAgent(_) => "pane.release_agent", + Method::PaneClose(_) => "pane.close", + Method::EventsSubscribe(_) => "events.subscribe", + Method::EventsWait(_) => "events.wait", + Method::PaneWaitForOutput(_) => "pane.wait_for_output", + Method::IntegrationInstall(_) => "integration.install", + Method::IntegrationUninstall(_) => "integration.uninstall", + } +} + +fn api_response_outcome(response: &str) -> &'static str { + let Ok(value) = serde_json::from_str::<serde_json::Value>(response) else { + return "error"; + }; + + match value + .get("error") + .and_then(|error| error.get("code")) + .and_then(|code| code.as_str()) + { + Some("timeout") => "timeout", + Some(_) => "error", + None => "ok", + } +} + +fn stream_subscriptions( + mut stream: UnixStream, + request_id: String, + params: crate::api::schema::EventsSubscribeParams, + api_tx: &ApiRequestSender, + event_hub: &EventHub, + running: &Arc<AtomicBool>, +) -> std::io::Result<()> { + let mut subscriptions = Vec::with_capacity(params.subscriptions.len()); + for (index, subscription) in params.subscriptions.into_iter().enumerate() { + let active = + match ActiveSubscription::new(subscription, &request_id, index, api_tx, event_hub) { + Ok(active) => active, + Err(response) => { + if let Err(err) = write_json_line(&mut stream, &response) { + if is_connection_closed_error(&err) { + return Ok(()); + } + return Err(err); + } + return Ok(()); + } + }; + subscriptions.push(active); + } + + if let Err(err) = write_json_line( + &mut stream, + &SuccessResponse { + id: request_id, + result: ResponseResult::SubscriptionStarted {}, + }, + ) { + if is_connection_closed_error(&err) { + return Ok(()); + } + return Err(err); + } + + loop { + if should_stop_connection(&mut stream, running)? { + return Ok(()); + } + + for subscription in &mut subscriptions { + if let Some(event) = subscription.poll(api_tx, event_hub) { + if let Err(err) = write_json_line(&mut stream, &event) { + if is_connection_closed_error(&err) { + return Ok(()); + } + return Err(err); + } + } + } + std::thread::sleep(CONNECTION_POLL_INTERVAL); + } +} + +fn write_text_line(stream: &mut UnixStream, value: &str) -> std::io::Result<()> { + stream.write_all(value.as_bytes())?; + stream.write_all(b"\n")?; + stream.flush() +} + +fn write_text_line_allow_disconnect(stream: &mut UnixStream, value: &str) -> std::io::Result<()> { + match write_text_line(stream, value) { + Err(err) if is_connection_closed_error(&err) => Ok(()), + result => result, + } +} + +fn write_json_line<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> std::io::Result<()> { + let encoded = serde_json::to_string(value) + .map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?; + write_text_line(stream, &encoded) +} + +fn write_json_line_allow_disconnect<T: serde::Serialize>( + stream: &mut UnixStream, + value: &T, +) -> std::io::Result<()> { + let encoded = serde_json::to_string(value) + .map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?; + write_text_line_allow_disconnect(stream, &encoded) +} + +pub(super) fn should_stop_connection( + stream: &mut UnixStream, + running: &Arc<AtomicBool>, +) -> std::io::Result<bool> { + if !running.load(Ordering::Relaxed) { + return Ok(true); + } + + probe_stream_closed(stream) +} + +fn probe_stream_closed(stream: &mut UnixStream) -> std::io::Result<bool> { + stream.set_nonblocking(true)?; + let mut probe = [0u8; 1]; + let status = match stream.read(&mut probe) { + Ok(0) => Ok(true), + Ok(_) => Ok(true), + Err(err) + if matches!( + err.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted + ) => + { + Ok(false) + } + Err(err) if is_connection_closed_error(&err) => Ok(true), + Err(err) => Err(err), + }; + stream.set_nonblocking(false)?; + status +} + +fn is_connection_closed_error(err: &std::io::Error) -> bool { + matches!( + err.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::NotConnected + | std::io::ErrorKind::UnexpectedEof + | std::io::ErrorKind::WriteZero + ) +} + +fn dispatch_to_app(request: Request, api_tx: &ApiRequestSender) -> String { + dispatch_to_app_with_timeout(request, api_tx, None) +} + +pub(super) fn dispatch_to_app_with_timeout( + request: Request, + api_tx: &ApiRequestSender, + timeout: Option<Duration>, +) -> String { + let request_id = request.id.clone(); + let (respond_to, response_rx) = std::sync::mpsc::channel(); + if let Err(err) = api_tx.send(ApiRequestMessage { + request, + respond_to, + }) { + return error_response_json( + request_id, + "server_unavailable", + format!("failed to dispatch request: {err}"), + ); + } + + let response = match timeout { + Some(timeout) => response_rx.recv_timeout(timeout).map_err(|err| match err { + std::sync::mpsc::RecvTimeoutError::Timeout => std::io::Error::new( + std::io::ErrorKind::TimedOut, + format!( + "timed out waiting for app response after {} ms", + timeout.as_millis() + ), + ), + std::sync::mpsc::RecvTimeoutError::Disconnected => std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "app response channel closed", + ), + }), + None => response_rx + .recv() + .map_err(|err| std::io::Error::new(std::io::ErrorKind::BrokenPipe, err)), + }; + + match response { + Ok(response) => response, + Err(err) => error_response_json( + request_id, + "server_unavailable", + format!("request handling failed: {err}"), + ), + } +} + +fn error_response_json(id: String, code: &str, message: String) -> String { + serde_json::to_string(&ErrorResponse { + id, + error: ErrorBody { + code: code.into(), + message, + }, + }) + .unwrap_or_else(|_| { + r#"{"id":"","error":{"code":"internal_error","message":"failed to encode error response"}}"# + .to_string() + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::fs::PermissionsExt; + use std::sync::{Mutex, OnceLock}; + use tokio::sync::mpsc; + + fn env_lock() -> &'static Mutex<()> { + static LOCK: OnceLock<Mutex<()>> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) + } + + fn unique_test_path(name: &str) -> PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir().join(format!("herdr-{name}-{}-{nanos}", std::process::id())) + } + + fn read_line(stream: &mut UnixStream) -> String { + let mut reader = BufReader::new(stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + line + } + + #[test] + fn socket_path_prefers_explicit_env_override() { + let _guard = env_lock().lock().unwrap(); + let unique = format!("/tmp/herdr-test-{}.sock", std::process::id()); + std::env::remove_var(crate::session::SESSION_ENV_VAR); + crate::session::clear_explicit_session_for_test(); + std::env::set_var(crate::api::SOCKET_PATH_ENV_VAR, &unique); + assert_eq!(socket_path(), PathBuf::from(&unique)); + std::env::remove_var(crate::api::SOCKET_PATH_ENV_VAR); + } + + #[test] + fn socket_path_defaults_to_config_dir_even_when_xdg_runtime_dir_is_set() { + let _guard = env_lock().lock().unwrap(); + let config_home = unique_test_path("socket-default-config-home"); + let runtime_dir = unique_test_path("socket-default-runtime"); + std::env::remove_var(crate::api::SOCKET_PATH_ENV_VAR); + std::env::remove_var(crate::session::SESSION_ENV_VAR); + crate::session::clear_explicit_session_for_test(); + std::env::set_var("XDG_CONFIG_HOME", &config_home); + std::env::set_var("XDG_RUNTIME_DIR", &runtime_dir); + + let expected = config_home + .join(crate::config::app_dir_name()) + .join("herdr.sock"); + assert_eq!(socket_path(), expected); + + std::env::remove_var("XDG_CONFIG_HOME"); + std::env::remove_var("XDG_RUNTIME_DIR"); + } + + #[test] + fn socket_path_uses_named_session_dir() { + let _guard = env_lock().lock().unwrap(); + let config_home = unique_test_path("socket-named-config-home"); + std::env::remove_var(crate::api::SOCKET_PATH_ENV_VAR); + crate::session::clear_explicit_session_for_test(); + std::env::set_var(crate::session::SESSION_ENV_VAR, "work"); + std::env::set_var("XDG_CONFIG_HOME", &config_home); + + let expected = config_home + .join(crate::config::app_dir_name()) + .join("sessions") + .join("work") + .join("herdr.sock"); + assert_eq!(socket_path(), expected); + + std::env::remove_var(crate::session::SESSION_ENV_VAR); + std::env::remove_var("XDG_CONFIG_HOME"); + } + + #[test] + fn restrict_socket_permissions_sets_user_only_mode() { + let dir = unique_test_path("socket-perms"); + fs::create_dir_all(&dir).unwrap(); + let path = dir.join("api.sock"); + let _listener = UnixListener::bind(&path).unwrap(); + + restrict_socket_permissions(&path).unwrap(); + + let mode = fs::metadata(&path).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, SOCKET_PERMISSION_MODE); + + drop(_listener); + let _ = fs::remove_file(&path); + let _ = fs::remove_dir_all(&dir); + } + + #[test] + fn api_response_outcome_uses_top_level_error_shape() { + let ok_with_error_text = r#"{"id":"req","result":{"read":{"text":"user said \"error\": \"timeout\"","revision":1}}}"#; + assert_eq!(api_response_outcome(ok_with_error_text), "ok"); + + let timeout = r#"{"id":"req","error":{"code":"timeout","message":"timed out waiting for output match"}}"#; + assert_eq!(api_response_outcome(timeout), "timeout"); + + let generic_error = + r#"{"id":"req","error":{"code":"server_unavailable","message":"boom"}}"#; + assert_eq!(api_response_outcome(generic_error), "error"); + } + + #[test] + fn ping_request_returns_pong() { + let (tx, _rx) = mpsc::unbounded_channel(); + let response = handle_request( + Request { + id: "req_1".into(), + method: Method::Ping(crate::api::schema::PingParams::default()), + }, + &tx, + ); + + let parsed: SuccessResponse = serde_json::from_str(&response).unwrap(); + assert_eq!(parsed.id, "req_1"); + assert!(matches!(parsed.result, ResponseResult::Pong { .. })); + } + + #[test] + fn request_dispatches_to_app_channel() { + let (tx, mut rx) = mpsc::unbounded_channel(); + let request = Request { + id: "req_2".into(), + method: Method::WorkspaceList(crate::api::schema::EmptyParams::default()), + }; + + let request_for_thread = request.clone(); + let thread = std::thread::spawn(move || handle_request(request_for_thread, &tx)); + + let msg = rx.blocking_recv().unwrap(); + assert_eq!(msg.request.id, "req_2"); + msg.respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: "req_2".into(), + result: ResponseResult::Ok {}, + }) + .unwrap(), + ) + .unwrap(); + + let response = thread.join().unwrap(); + let parsed: SuccessResponse = serde_json::from_str(&response).unwrap(); + assert_eq!(parsed.id, "req_2"); + } + + #[test] + fn wait_for_output_stops_when_client_disconnects() { + let (api_tx, mut api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let (first_read_tx, first_read_rx) = std::sync::mpsc::channel(); + let responder = std::thread::spawn(move || { + let mut notified = false; + while let Some(msg) = api_rx.blocking_recv() { + assert!(matches!(msg.request.method, Method::PaneRead(_))); + if !notified { + first_read_tx.send(()).unwrap(); + notified = true; + } + msg.respond_to + .send( + serde_json::to_string(&SuccessResponse { + id: msg.request.id, + result: ResponseResult::PaneRead { + read: crate::api::schema::PaneReadResult { + pane_id: "pane_1".into(), + workspace_id: "ws_1".into(), + tab_id: "tab_1".into(), + source: crate::api::schema::ReadSource::RecentUnwrapped, + format: crate::api::schema::ReadFormat::Text, + text: String::new(), + revision: 0, + truncated: false, + }, + }, + }) + .unwrap(), + ) + .unwrap(); + } + }); + + let (mut client, server) = UnixStream::pair().unwrap(); + client + .write_all(br#"{"id":"req_wait","method":"pane.wait_for_output","params":{"pane_id":"pane_1","source":"recent","match":{"type":"substring","value":"never"}}}"#) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let event_hub = EventHub::default(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let server_thread = std::thread::spawn(move || { + let result = handle_connection(server, &api_tx, &event_hub, &server_running); + done_tx.send(result).unwrap(); + }); + + first_read_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + drop(client); + + let result = done_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + assert!(result.is_ok()); + + server_thread.join().unwrap(); + drop(running); + responder.join().unwrap(); + } + + #[test] + fn subscriptions_stop_when_client_disconnects() { + let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let (mut client, server) = UnixStream::pair().unwrap(); + client + .write_all( + br#"{"id":"sub_1","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#, + ) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let event_hub = EventHub::default(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let server_thread = std::thread::spawn(move || { + let result = handle_connection(server, &api_tx, &event_hub, &server_running); + done_tx.send(result).unwrap(); + }); + + let ack = read_line(&mut client); + let ack: serde_json::Value = serde_json::from_str(&ack).unwrap(); + assert_eq!(ack["result"]["type"], "subscription_started"); + + drop(client); + + let result = done_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + assert!(result.is_ok()); + server_thread.join().unwrap(); + } + + #[test] + fn subscriptions_stop_when_server_shuts_down() { + let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>(); + let (mut client, server) = UnixStream::pair().unwrap(); + client + .write_all( + br#"{"id":"sub_2","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#, + ) + .unwrap(); + client.write_all(b"\n").unwrap(); + client.flush().unwrap(); + + let running = Arc::new(AtomicBool::new(true)); + let server_running = Arc::clone(&running); + let event_hub = EventHub::default(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let server_thread = std::thread::spawn(move || { + let result = handle_connection(server, &api_tx, &event_hub, &server_running); + done_tx.send(result).unwrap(); + }); + + let ack = read_line(&mut client); + let ack: serde_json::Value = serde_json::from_str(&ack).unwrap(); + assert_eq!(ack["result"]["type"], "subscription_started"); + + running.store(false, Ordering::Relaxed); + + let result = done_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + assert!(result.is_ok()); + server_thread.join().unwrap(); + } +} |