diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/client.rs | 245 | ||||
| -rw-r--r-- | src/api/event_hub.rs | 39 | ||||
| -rw-r--r-- | src/api/mod.rs | 1356 | ||||
| -rw-r--r-- | src/api/server.rs | 769 | ||||
| -rw-r--r-- | src/api/status.rs | 54 | ||||
| -rw-r--r-- | src/api/subscriptions.rs | 368 | ||||
| -rw-r--r-- | src/api/wait.rs | 124 |
7 files changed, 1610 insertions, 1345 deletions
diff --git a/src/api/client.rs b/src/api/client.rs new file mode 100644 index 00000000..7929f937 --- /dev/null +++ b/src/api/client.rs @@ -0,0 +1,245 @@ +use std::fmt; +use std::io::{self, BufRead, BufReader, Write}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::time::Duration; + +use serde::de::DeserializeOwned; + +use crate::api::schema::{ + ErrorResponse, EventsSubscribeParams, Method, PingParams, Request, ResponseResult, + SubscriptionEventEnvelope, SuccessResponse, +}; + +/// API connection target resolved by clients at the process edge. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ConnectionTarget { + LocalSession(Option<String>), + SocketPath(PathBuf), +} + +impl ConnectionTarget { + fn socket_path(&self) -> PathBuf { + match self { + Self::LocalSession(None) => crate::api::socket_path(), + Self::LocalSession(Some(name)) => crate::session::api_socket_path_for(Some(name)), + Self::SocketPath(path) => path.clone(), + } + } +} + +/// Reusable client for Herdr's newline-delimited JSON API. +#[derive(Debug, Clone)] +pub struct ApiClient { + target: ConnectionTarget, +} + +impl ApiClient { + pub fn local() -> Self { + Self::for_target(ConnectionTarget::LocalSession(None)) + } + + pub fn for_target(target: ConnectionTarget) -> Self { + Self { target } + } + + pub fn socket_path(&self) -> PathBuf { + self.target.socket_path() + } + + pub fn request(&self, request: Request) -> Result<SuccessResponse, ApiClientError> { + let value = self.request_value(&request)?; + parse_response_value(value) + } + + pub fn request_value(&self, request: &Request) -> Result<serde_json::Value, ApiClientError> { + let mut stream = self.connect()?; + write_request(&mut stream, request)?; + + let mut reader = BufReader::new(stream); + read_json_line(&mut reader) + } + + pub fn request_value_with_timeout( + &self, + request: &Request, + timeout: Duration, + ) -> Result<serde_json::Value, ApiClientError> { + let mut stream = self.connect()?; + stream.set_write_timeout(Some(timeout))?; + stream.set_read_timeout(Some(timeout))?; + write_request(&mut stream, request)?; + + let mut reader = BufReader::new(stream); + read_json_line(&mut reader) + } + + #[allow(dead_code)] // Kept as the typed subscription API; CLI wait paths use subscribe_value to preserve raw ack errors. + pub fn subscribe( + &self, + id: impl Into<String>, + params: EventsSubscribeParams, + read_timeout: Option<Duration>, + ) -> Result<(SuccessResponse, EventStream), ApiClientError> { + let request = Request { + id: id.into(), + method: Method::EventsSubscribe(params), + }; + let (ack, stream) = self.subscribe_value(&request, read_timeout)?; + Ok((parse_response_value(ack)?, stream)) + } + + pub fn subscribe_value( + &self, + request: &Request, + read_timeout: Option<Duration>, + ) -> Result<(serde_json::Value, EventStream), ApiClientError> { + let mut stream = self.connect()?; + write_request(&mut stream, request)?; + if let Some(timeout) = read_timeout { + stream.set_read_timeout(Some(timeout))?; + } + + let mut reader = BufReader::new(stream); + let ack = read_json_line(&mut reader)?; + Ok((ack, EventStream { reader })) + } + + pub fn status(&self) -> Result<crate::api::RuntimeStatus, ApiClientError> { + let response = self.request(Request { + id: "api-client:status".into(), + method: Method::Ping(PingParams::default()), + })?; + match response.result { + ResponseResult::Pong { version, protocol } => Ok(crate::api::RuntimeStatus { + version: Some(version), + protocol: Some(protocol), + }), + result => Err(ApiClientError::UnexpectedResult(format!("{result:?}"))), + } + } + + fn connect(&self) -> io::Result<UnixStream> { + UnixStream::connect(self.socket_path()) + } +} + +pub struct EventStream { + reader: BufReader<UnixStream>, +} + +impl EventStream { + pub fn next_value(&mut self) -> Result<Option<serde_json::Value>, ApiClientError> { + read_optional_json_line(&mut self.reader) + } + + pub fn next_event(&mut self) -> Result<Option<SubscriptionEventEnvelope>, ApiClientError> { + self.next_value()? + .map(serde_json::from_value) + .transpose() + .map_err(ApiClientError::Json) + } +} + +#[derive(Debug)] +pub enum ApiClientError { + Io(io::Error), + Json(serde_json::Error), + ErrorResponse(ErrorResponse), + EmptyResponse, + UnexpectedResult(String), +} + +impl fmt::Display for ApiClientError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Io(err) => write!(f, "{err}"), + Self::Json(err) => write!(f, "{err}"), + Self::ErrorResponse(response) => write!(f, "{}", response.error.message), + Self::EmptyResponse => write!(f, "empty api response"), + Self::UnexpectedResult(result) => write!(f, "unexpected api result: {result}"), + } + } +} + +impl std::error::Error for ApiClientError {} + +impl From<io::Error> for ApiClientError { + fn from(err: io::Error) -> Self { + Self::Io(err) + } +} + +impl From<serde_json::Error> for ApiClientError { + fn from(err: serde_json::Error) -> Self { + Self::Json(err) + } +} + +fn write_request(stream: &mut UnixStream, request: &Request) -> Result<(), ApiClientError> { + stream.write_all(serde_json::to_string(request)?.as_bytes())?; + stream.write_all(b"\n")?; + stream.flush()?; + Ok(()) +} + +fn read_json_line<T: DeserializeOwned>( + reader: &mut BufReader<UnixStream>, +) -> Result<T, ApiClientError> { + let mut line = String::new(); + let read = reader.read_line(&mut line)?; + if read == 0 || line.trim().is_empty() { + return Err(ApiClientError::EmptyResponse); + } + serde_json::from_str(&line).map_err(ApiClientError::Json) +} + +fn read_optional_json_line<T: DeserializeOwned>( + reader: &mut BufReader<UnixStream>, +) -> Result<Option<T>, ApiClientError> { + let mut line = String::new(); + let read = reader.read_line(&mut line)?; + if read == 0 { + return Ok(None); + } + if line.trim().is_empty() { + return Err(ApiClientError::EmptyResponse); + } + serde_json::from_str(&line) + .map(Some) + .map_err(ApiClientError::Json) +} + +#[derive(serde::Deserialize)] +#[serde(untagged)] +enum WireResponse { + Success(Box<SuccessResponse>), + Error(ErrorResponse), +} + +pub(crate) fn parse_response_value( + value: serde_json::Value, +) -> Result<SuccessResponse, ApiClientError> { + match serde_json::from_value(value)? { + WireResponse::Success(response) => Ok(*response), + WireResponse::Error(response) => Err(ApiClientError::ErrorResponse(response)), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn local_session_target_resolves_named_session_socket() { + let client = ApiClient::for_target(ConnectionTarget::LocalSession(Some("work".into()))); + assert!(client.socket_path().ends_with("sessions/work/herdr.sock")); + } + + #[test] + fn socket_path_target_uses_explicit_path() { + let path = PathBuf::from("/tmp/herdr-test.sock"); + let client = ApiClient::for_target(ConnectionTarget::SocketPath(path.clone())); + assert_eq!(client.socket_path(), path); + } +} diff --git a/src/api/event_hub.rs b/src/api/event_hub.rs new file mode 100644 index 00000000..b66a2157 --- /dev/null +++ b/src/api/event_hub.rs @@ -0,0 +1,39 @@ +#[derive(Clone, Default)] +pub struct EventHub { + inner: std::sync::Arc<std::sync::Mutex<EventHubState>>, +} + +#[derive(Default)] +struct EventHubState { + next_sequence: u64, + events: Vec<(u64, crate::api::schema::EventEnvelope)>, +} + +impl EventHub { + const MAX_EVENTS: usize = 512; + + pub fn push(&self, event: crate::api::schema::EventEnvelope) { + let Ok(mut state) = self.inner.lock() else { + return; + }; + state.next_sequence += 1; + let sequence = state.next_sequence; + state.events.push((sequence, event)); + let overflow = state.events.len().saturating_sub(Self::MAX_EVENTS); + if overflow > 0 { + state.events.drain(0..overflow); + } + } + + pub fn events_after(&self, sequence: u64) -> Vec<(u64, crate::api::schema::EventEnvelope)> { + let Ok(state) = self.inner.lock() else { + return Vec::new(); + }; + state + .events + .iter() + .filter(|(event_sequence, _)| *event_sequence > sequence) + .cloned() + .collect() + } +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 4893c295..34b37138 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,32 +1,23 @@ +pub mod client; +mod event_hub; pub mod schema; +mod server; +mod status; +mod subscriptions; +mod wait; -use std::fs; -use std::io::{self, 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; +pub use event_hub::EventHub; +pub use server::start_server; +pub use status::{read_runtime_status_at, RuntimeStatus}; -use tracing::{debug, error, info, warn}; +use std::path::PathBuf; -use regex::Regex; use tokio::sync::mpsc; -use crate::api::schema::{ - ErrorBody, ErrorResponse, Method, PaneAgentStatusChangedEvent, PaneOutputMatchedEvent, Request, - ResponseResult, Subscription, SubscriptionEventData, SubscriptionEventEnvelope, - SubscriptionEventKind, SuccessResponse, -}; +use crate::api::schema::{Method, Request}; pub const SOCKET_PATH_ENV_VAR: &str = "HERDR_SOCKET_PATH"; -const SOCKET_PERMISSION_MODE: u32 = 0o600; -const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100); -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(crate) fn request_changes_ui(request: &Request) -> bool { matches!( &request.method, @@ -58,1331 +49,6 @@ pub struct ApiRequestMessage { pub type ApiRequestSender = mpsc::UnboundedSender<ApiRequestMessage>; -#[derive(Clone, Default)] -pub struct EventHub { - inner: std::sync::Arc<std::sync::Mutex<EventHubState>>, -} - -#[derive(Default)] -struct EventHubState { - next_sequence: u64, - events: Vec<(u64, crate::api::schema::EventEnvelope)>, -} - -impl EventHub { - const MAX_EVENTS: usize = 512; - - pub fn push(&self, event: crate::api::schema::EventEnvelope) { - let Ok(mut state) = self.inner.lock() else { - return; - }; - state.next_sequence += 1; - let sequence = state.next_sequence; - state.events.push((sequence, event)); - let overflow = state.events.len().saturating_sub(Self::MAX_EVENTS); - if overflow > 0 { - state.events.drain(0..overflow); - } - } - - pub fn events_after(&self, sequence: u64) -> Vec<(u64, crate::api::schema::EventEnvelope)> { - let Ok(state) = self.inner.lock() else { - return Vec::new(); - }; - state - .events - .iter() - .filter(|(event_sequence, _)| *event_sequence > sequence) - .cloned() - .collect() - } -} - pub fn socket_path() -> PathBuf { crate::session::active_api_socket_path() } - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct RuntimeStatus { - pub version: Option<String>, - pub protocol: Option<u32>, -} - -pub fn read_runtime_status_at( - socket_path: &Path, - timeout: Duration, -) -> io::Result<Option<RuntimeStatus>> { - if !socket_path.exists() { - return Ok(None); - } - - let mut stream = match UnixStream::connect(socket_path) { - Ok(stream) => stream, - Err(err) - if matches!( - err.kind(), - io::ErrorKind::ConnectionRefused - | io::ErrorKind::NotFound - | io::ErrorKind::TimedOut - ) => - { - return Ok(None); - } - Err(err) => return Err(err), - }; - - stream.set_write_timeout(Some(timeout))?; - stream.set_read_timeout(Some(timeout))?; - - let request = Request { - id: "runtime:status".into(), - method: Method::Ping(crate::api::schema::PingParams::default()), - }; - stream.write_all(serde_json::to_string(&request)?.as_bytes())?; - stream.write_all(b"\n")?; - stream.flush()?; - - let mut reader = BufReader::new(stream); - let mut line = String::new(); - let read = reader.read_line(&mut line)?; - if read == 0 || line.trim().is_empty() { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "empty server status response", - )); - } - - let response: serde_json::Value = serde_json::from_str(&line).map_err(io::Error::other)?; - if response.get("error").is_some() { - return Err(io::Error::other(format!( - "server status request failed: {response}" - ))); - } - - let result = &response["result"]; - Ok(Some(RuntimeStatus { - version: result - .get("version") - .and_then(|value| value.as_str()) - .map(str::to_owned), - protocol: result - .get("protocol") - .and_then(|value| value.as_u64()) - .and_then(|value| u32::try_from(value).ok()), - })) -} - -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::server::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 output_match_read_source( - source: &crate::api::schema::ReadSource, -) -> crate::api::schema::ReadSource { - match source { - crate::api::schema::ReadSource::Recent => crate::api::schema::ReadSource::RecentUnwrapped, - other => *other, - } -} - -fn wait_for_output( - request_id: String, - params: crate::api::schema::PaneWaitForOutputParams, - stream: &mut UnixStream, - api_tx: &ApiRequestSender, - running: &Arc<AtomicBool>, -) -> std::io::Result<Option<String>> { - crate::logging::api_wait_started(&request_id, ¶ms.pane_id, params.timeout_ms); - let deadline = params - .timeout_ms - .map(|ms| std::time::Instant::now() + std::time::Duration::from_millis(ms)); - - let regex = match ¶ms.r#match { - crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) { - Ok(regex) => Some(regex), - Err(err) => { - return Ok(Some( - serde_json::to_string(&ErrorResponse { - id: request_id, - error: ErrorBody { - code: "invalid_regex".into(), - message: err.to_string(), - }, - }) - .unwrap(), - )); - } - }, - crate::api::schema::OutputMatch::Substring { .. } => None, - }; - - loop { - if should_stop_connection(stream, running)? { - crate::logging::api_wait_completed(&request_id, ¶ms.pane_id, "client_disconnected"); - return Ok(None); - } - - let read_request = Request { - id: format!("{request_id}:read"), - method: Method::PaneRead(crate::api::schema::PaneReadParams { - pane_id: params.pane_id.clone(), - source: output_match_read_source(¶ms.source), - lines: params.lines, - format: crate::api::schema::ReadFormat::Text, - strip_ansi: params.strip_ansi, - }), - }; - let response = - dispatch_to_app_with_timeout(read_request, api_tx, Some(APP_RESPONSE_TIMEOUT)); - let Ok(value) = serde_json::from_str::<serde_json::Value>(&response) else { - return Ok(Some(response)); - }; - if value.get("error").is_some() { - let mut value = value; - value["id"] = serde_json::Value::String(request_id.clone()); - return Ok(Some(serde_json::to_string(&value).unwrap())); - } - - let read_value = value["result"]["read"].clone(); - let Ok(read) = serde_json::from_value::<crate::api::schema::PaneReadResult>(read_value) - else { - return Ok(Some( - serde_json::to_string(&ErrorResponse { - id: request_id, - error: ErrorBody { - code: "internal_error".into(), - message: "failed to decode pane read result".into(), - }, - }) - .unwrap(), - )); - }; - - let matched_line = match_output(&read.text, ¶ms.r#match, regex.as_ref()); - if matched_line.is_some() { - let revision = read.revision; - crate::logging::api_wait_completed(&request_id, ¶ms.pane_id, "matched"); - return Ok(Some( - serde_json::to_string(&SuccessResponse { - id: request_id, - result: ResponseResult::OutputMatched { - pane_id: params.pane_id, - revision, - matched_line, - read, - }, - }) - .unwrap(), - )); - } - - if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) { - crate::logging::api_wait_timed_out(&request_id, ¶ms.pane_id); - return Ok(Some( - serde_json::to_string(&ErrorResponse { - id: request_id, - error: ErrorBody { - code: "timeout".into(), - message: "timed out waiting for output match".into(), - }, - }) - .unwrap(), - )); - } - - std::thread::sleep(CONNECTION_POLL_INTERVAL); - } -} - -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) -} - -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 match_output( - text: &str, - matcher: &crate::api::schema::OutputMatch, - regex: Option<&Regex>, -) -> Option<String> { - match matcher { - crate::api::schema::OutputMatch::Substring { value } => text - .lines() - .find(|line| line.contains(value)) - .map(|line| line.to_string()), - crate::api::schema::OutputMatch::Regex { .. } => regex.and_then(|re| { - text.lines() - .find(|line| re.is_match(line)) - .map(|line| line.to_string()) - }), - } -} - -struct ActiveOutputMatchedSubscription { - pane_id: String, - source: crate::api::schema::ReadSource, - lines: Option<u32>, - matcher: crate::api::schema::OutputMatch, - regex: Option<Regex>, - strip_ansi: bool, - currently_matching: bool, - request_prefix: String, -} - -struct ActiveAgentStatusChangedSubscription { - pane_id: String, - status_filter: Option<crate::api::schema::AgentStatus>, - last_status: Option<crate::api::schema::AgentStatus>, - request_prefix: String, -} - -struct ActiveEventSubscription { - event_kind: crate::api::schema::EventKind, - last_sequence: u64, -} - -enum ActiveSubscription { - Event(ActiveEventSubscription), - OutputMatched(ActiveOutputMatchedSubscription), - AgentStatusChanged(ActiveAgentStatusChangedSubscription), -} - -impl ActiveSubscription { - fn new( - subscription: Subscription, - request_id: &str, - index: usize, - api_tx: &ApiRequestSender, - _event_hub: &EventHub, - ) -> Result<Self, ErrorResponse> { - match subscription { - Subscription::WorkspaceCreated {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::WorkspaceCreated, - last_sequence: 0, - })), - Subscription::WorkspaceClosed {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::WorkspaceClosed, - last_sequence: 0, - })), - Subscription::WorkspaceFocused {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::WorkspaceFocused, - last_sequence: 0, - })), - Subscription::TabCreated {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::TabCreated, - last_sequence: 0, - })), - Subscription::TabClosed {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::TabClosed, - last_sequence: 0, - })), - Subscription::TabFocused {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::TabFocused, - last_sequence: 0, - })), - Subscription::TabRenamed {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::TabRenamed, - last_sequence: 0, - })), - Subscription::PaneCreated {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::PaneCreated, - last_sequence: 0, - })), - Subscription::PaneClosed {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::PaneClosed, - last_sequence: 0, - })), - Subscription::PaneFocused {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::PaneFocused, - last_sequence: 0, - })), - Subscription::PaneExited {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::PaneExited, - last_sequence: 0, - })), - Subscription::PaneAgentDetected {} => Ok(Self::Event(ActiveEventSubscription { - event_kind: crate::api::schema::EventKind::PaneAgentDetected, - last_sequence: 0, - })), - Subscription::PaneOutputMatched { - pane_id, - source, - lines, - r#match, - strip_ansi, - } => { - let regex = match &r#match { - crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) { - Ok(regex) => Some(regex), - Err(err) => { - return Err(ErrorResponse { - id: request_id.to_string(), - error: ErrorBody { - code: "invalid_regex".into(), - message: err.to_string(), - }, - }); - } - }, - crate::api::schema::OutputMatch::Substring { .. } => None, - }; - - let probe = pane_read( - format!("{request_id}:sub:{index}:probe"), - &pane_id, - source, - lines, - strip_ansi, - api_tx, - ); - probe?; - - Ok(Self::OutputMatched(ActiveOutputMatchedSubscription { - pane_id, - source, - lines, - matcher: r#match, - regex, - strip_ansi, - currently_matching: false, - request_prefix: format!("{request_id}:sub:{index}"), - })) - } - Subscription::PaneAgentStatusChanged { - pane_id, - agent_status, - } => { - let probe = pane_get(format!("{request_id}:sub:{index}:probe"), &pane_id, api_tx)?; - - Ok(Self::AgentStatusChanged( - ActiveAgentStatusChangedSubscription { - pane_id, - status_filter: agent_status, - last_status: Some(probe.agent_status), - request_prefix: format!("{request_id}:sub:{index}"), - }, - )) - } - } - } - - fn poll( - &mut self, - api_tx: &ApiRequestSender, - event_hub: &EventHub, - ) -> Option<serde_json::Value> { - match self { - Self::Event(subscription) => subscription.poll(event_hub), - Self::OutputMatched(subscription) => { - serde_json::to_value(subscription.poll(api_tx)?).ok() - } - Self::AgentStatusChanged(subscription) => { - serde_json::to_value(subscription.poll(api_tx)?).ok() - } - } - } -} - -impl ActiveEventSubscription { - fn poll(&mut self, event_hub: &EventHub) -> Option<serde_json::Value> { - for (sequence, event) in event_hub.events_after(self.last_sequence) { - self.last_sequence = sequence; - if event.event == self.event_kind { - return serde_json::to_value(event).ok(); - } - } - None - } -} - -impl ActiveOutputMatchedSubscription { - fn poll(&mut self, api_tx: &ApiRequestSender) -> Option<SubscriptionEventEnvelope> { - let read = pane_read( - format!("{}:read", self.request_prefix), - &self.pane_id, - output_match_read_source(&self.source), - self.lines, - self.strip_ansi, - api_tx, - ) - .ok()?; - - let matched_line = match_output(&read.text, &self.matcher, self.regex.as_ref()); - match matched_line { - Some(matched_line) => { - if self.currently_matching { - return None; - } - self.currently_matching = true; - Some(SubscriptionEventEnvelope { - event: SubscriptionEventKind::PaneOutputMatched, - data: SubscriptionEventData::PaneOutputMatched(PaneOutputMatchedEvent { - pane_id: self.pane_id.clone(), - matched_line, - read, - }), - }) - } - None => { - self.currently_matching = false; - None - } - } - } -} - -impl ActiveAgentStatusChangedSubscription { - fn poll(&mut self, api_tx: &ApiRequestSender) -> Option<SubscriptionEventEnvelope> { - let pane = pane_get( - format!("{}:pane", self.request_prefix), - &self.pane_id, - api_tx, - ) - .ok()?; - let current_status = pane.agent_status; - let previous_status = self.last_status.replace(current_status); - if previous_status.is_none() || previous_status == Some(current_status) { - return None; - } - if self - .status_filter - .is_some_and(|wanted| wanted != current_status) - { - return None; - } - - Some(SubscriptionEventEnvelope { - event: SubscriptionEventKind::PaneAgentStatusChanged, - data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent { - pane_id: pane.pane_id, - workspace_id: pane.workspace_id, - agent_status: current_status, - agent: pane.agent, - custom_status: pane.custom_status, - }), - }) - } -} - -fn pane_read( - request_id: String, - pane_id: &str, - source: crate::api::schema::ReadSource, - lines: Option<u32>, - strip_ansi: bool, - api_tx: &ApiRequestSender, -) -> Result<crate::api::schema::PaneReadResult, ErrorResponse> { - let response = dispatch_to_app_with_timeout( - Request { - id: request_id.clone(), - method: Method::PaneRead(crate::api::schema::PaneReadParams { - pane_id: pane_id.to_string(), - source, - lines, - format: crate::api::schema::ReadFormat::Text, - strip_ansi, - }), - }, - api_tx, - Some(APP_RESPONSE_TIMEOUT), - ); - let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse { - id: request_id.clone(), - error: ErrorBody { - code: "internal_error".into(), - message: "failed to decode pane read response".into(), - }, - })?; - if value.get("error").is_some() { - return serde_json::from_value(value).map_err(|_| ErrorResponse { - id: request_id, - error: ErrorBody { - code: "internal_error".into(), - message: "failed to decode pane read error".into(), - }, - }); - } - serde_json::from_value(value["result"]["read"].clone()).map_err(|_| ErrorResponse { - id: request_id, - error: ErrorBody { - code: "internal_error".into(), - message: "failed to decode pane read result".into(), - }, - }) -} - -fn pane_get( - request_id: String, - pane_id: &str, - api_tx: &ApiRequestSender, -) -> Result<crate::api::schema::PaneInfo, ErrorResponse> { - let response = dispatch_to_app_with_timeout( - Request { - id: request_id.clone(), - method: Method::PaneGet(crate::api::schema::PaneTarget { - pane_id: pane_id.to_string(), - }), - }, - api_tx, - Some(APP_RESPONSE_TIMEOUT), - ); - let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse { - id: request_id.clone(), - error: ErrorBody { - code: "internal_error".into(), - message: "failed to decode pane get response".into(), - }, - })?; - if value.get("error").is_some() { - return serde_json::from_value(value).map_err(|_| ErrorResponse { - id: request_id, - error: ErrorBody { - code: "internal_error".into(), - message: "failed to decode pane get error".into(), - }, - }); - } - serde_json::from_value(value["result"]["pane"].clone()).map_err(|_| ErrorResponse { - id: request_id, - error: ErrorBody { - code: "internal_error".into(), - message: "failed to decode pane get result".into(), - }, - }) -} - -fn dispatch_to_app(request: Request, api_tx: &ApiRequestSender) -> String { - dispatch_to_app_with_timeout(request, api_tx, None) -} - -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}; - - 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(SOCKET_PATH_ENV_VAR, &unique); - assert_eq!(socket_path(), PathBuf::from(&unique)); - std::env::remove_var(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(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(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(); - } -} 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(); + } +} diff --git a/src/api/status.rs b/src/api/status.rs new file mode 100644 index 00000000..f58e69c0 --- /dev/null +++ b/src/api/status.rs @@ -0,0 +1,54 @@ +use std::io; +use std::path::Path; +use std::time::Duration; + +use crate::api::schema::{Method, Request, ResponseResult}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RuntimeStatus { + pub version: Option<String>, + pub protocol: Option<u32>, +} + +pub fn read_runtime_status_at( + socket_path: &Path, + timeout: Duration, +) -> io::Result<Option<RuntimeStatus>> { + if !socket_path.exists() { + return Ok(None); + } + + let client = crate::api::client::ApiClient::for_target( + crate::api::client::ConnectionTarget::SocketPath(socket_path.to_path_buf()), + ); + let request = Request { + id: "runtime:status".into(), + method: Method::Ping(crate::api::schema::PingParams::default()), + }; + let response = client + .request_value_with_timeout(&request, timeout) + .and_then(crate::api::client::parse_response_value); + let response = match response { + Ok(response) => response, + Err(crate::api::client::ApiClientError::Io(err)) + if matches!( + err.kind(), + io::ErrorKind::ConnectionRefused + | io::ErrorKind::NotFound + | io::ErrorKind::TimedOut + ) => + { + return Ok(None); + } + Err(err) => return Err(io::Error::other(err)), + }; + match response.result { + ResponseResult::Pong { version, protocol } => Ok(Some(RuntimeStatus { + version: Some(version), + protocol: Some(protocol), + })), + result => Err(io::Error::other(format!( + "server status request returned unexpected result: {result:?}" + ))), + } +} diff --git a/src/api/subscriptions.rs b/src/api/subscriptions.rs new file mode 100644 index 00000000..891b743a --- /dev/null +++ b/src/api/subscriptions.rs @@ -0,0 +1,368 @@ +use regex::Regex; + +use crate::api::schema::{ + ErrorBody, ErrorResponse, Method, PaneAgentStatusChangedEvent, PaneOutputMatchedEvent, Request, + Subscription, SubscriptionEventData, SubscriptionEventEnvelope, SubscriptionEventKind, +}; +use crate::api::server::{dispatch_to_app_with_timeout, APP_RESPONSE_TIMEOUT}; +use crate::api::{ApiRequestSender, EventHub}; + +pub(super) fn output_match_read_source( + source: &crate::api::schema::ReadSource, +) -> crate::api::schema::ReadSource { + match source { + crate::api::schema::ReadSource::Recent => crate::api::schema::ReadSource::RecentUnwrapped, + other => *other, + } +} + +pub(super) fn match_output( + text: &str, + matcher: &crate::api::schema::OutputMatch, + regex: Option<&Regex>, +) -> Option<String> { + match matcher { + crate::api::schema::OutputMatch::Substring { value } => text + .lines() + .find(|line| line.contains(value)) + .map(|line| line.to_string()), + crate::api::schema::OutputMatch::Regex { .. } => regex.and_then(|re| { + text.lines() + .find(|line| re.is_match(line)) + .map(|line| line.to_string()) + }), + } +} + +pub(super) struct ActiveOutputMatchedSubscription { + pane_id: String, + source: crate::api::schema::ReadSource, + lines: Option<u32>, + matcher: crate::api::schema::OutputMatch, + regex: Option<Regex>, + strip_ansi: bool, + currently_matching: bool, + request_prefix: String, +} + +pub(super) struct ActiveAgentStatusChangedSubscription { + pane_id: String, + status_filter: Option<crate::api::schema::AgentStatus>, + last_status: Option<crate::api::schema::AgentStatus>, + request_prefix: String, +} + +pub(super) struct ActiveEventSubscription { + event_kind: crate::api::schema::EventKind, + last_sequence: u64, +} + +pub(super) enum ActiveSubscription { + Event(ActiveEventSubscription), + OutputMatched(ActiveOutputMatchedSubscription), + AgentStatusChanged(ActiveAgentStatusChangedSubscription), +} + +impl ActiveSubscription { + pub(super) fn new( + subscription: Subscription, + request_id: &str, + index: usize, + api_tx: &ApiRequestSender, + _event_hub: &EventHub, + ) -> Result<Self, ErrorResponse> { + match subscription { + Subscription::WorkspaceCreated {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::WorkspaceCreated, + last_sequence: 0, + })), + Subscription::WorkspaceClosed {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::WorkspaceClosed, + last_sequence: 0, + })), + Subscription::WorkspaceFocused {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::WorkspaceFocused, + last_sequence: 0, + })), + Subscription::TabCreated {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::TabCreated, + last_sequence: 0, + })), + Subscription::TabClosed {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::TabClosed, + last_sequence: 0, + })), + Subscription::TabFocused {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::TabFocused, + last_sequence: 0, + })), + Subscription::TabRenamed {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::TabRenamed, + last_sequence: 0, + })), + Subscription::PaneCreated {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::PaneCreated, + last_sequence: 0, + })), + Subscription::PaneClosed {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::PaneClosed, + last_sequence: 0, + })), + Subscription::PaneFocused {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::PaneFocused, + last_sequence: 0, + })), + Subscription::PaneExited {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::PaneExited, + last_sequence: 0, + })), + Subscription::PaneAgentDetected {} => Ok(Self::Event(ActiveEventSubscription { + event_kind: crate::api::schema::EventKind::PaneAgentDetected, + last_sequence: 0, + })), + Subscription::PaneOutputMatched { + pane_id, + source, + lines, + r#match, + strip_ansi, + } => { + let regex = match &r#match { + crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) { + Ok(regex) => Some(regex), + Err(err) => { + return Err(ErrorResponse { + id: request_id.to_string(), + error: ErrorBody { + code: "invalid_regex".into(), + message: err.to_string(), + }, + }); + } + }, + crate::api::schema::OutputMatch::Substring { .. } => None, + }; + + let probe = pane_read( + format!("{request_id}:sub:{index}:probe"), + &pane_id, + source, + lines, + strip_ansi, + api_tx, + ); + probe?; + + Ok(Self::OutputMatched(ActiveOutputMatchedSubscription { + pane_id, + source, + lines, + matcher: r#match, + regex, + strip_ansi, + currently_matching: false, + request_prefix: format!("{request_id}:sub:{index}"), + })) + } + Subscription::PaneAgentStatusChanged { + pane_id, + agent_status, + } => { + let probe = pane_get(format!("{request_id}:sub:{index}:probe"), &pane_id, api_tx)?; + + Ok(Self::AgentStatusChanged( + ActiveAgentStatusChangedSubscription { + pane_id, + status_filter: agent_status, + last_status: Some(probe.agent_status), + request_prefix: format!("{request_id}:sub:{index}"), + }, + )) + } + } + } + + pub(super) fn poll( + &mut self, + api_tx: &ApiRequestSender, + event_hub: &EventHub, + ) -> Option<serde_json::Value> { + match self { + Self::Event(subscription) => subscription.poll(event_hub), + Self::OutputMatched(subscription) => { + serde_json::to_value(subscription.poll(api_tx)?).ok() + } + Self::AgentStatusChanged(subscription) => { + serde_json::to_value(subscription.poll(api_tx)?).ok() + } + } + } +} + +impl ActiveEventSubscription { + fn poll(&mut self, event_hub: &EventHub) -> Option<serde_json::Value> { + for (sequence, event) in event_hub.events_after(self.last_sequence) { + self.last_sequence = sequence; + if event.event == self.event_kind { + return serde_json::to_value(event).ok(); + } + } + None + } +} + +impl ActiveOutputMatchedSubscription { + fn poll(&mut self, api_tx: &ApiRequestSender) -> Option<SubscriptionEventEnvelope> { + let read = pane_read( + format!("{}:read", self.request_prefix), + &self.pane_id, + output_match_read_source(&self.source), + self.lines, + self.strip_ansi, + api_tx, + ) + .ok()?; + + let matched_line = match_output(&read.text, &self.matcher, self.regex.as_ref()); + match matched_line { + Some(matched_line) => { + if self.currently_matching { + return None; + } + self.currently_matching = true; + Some(SubscriptionEventEnvelope { + event: SubscriptionEventKind::PaneOutputMatched, + data: SubscriptionEventData::PaneOutputMatched(PaneOutputMatchedEvent { + pane_id: self.pane_id.clone(), + matched_line, + read, + }), + }) + } + None => { + self.currently_matching = false; + None + } + } + } +} + +impl ActiveAgentStatusChangedSubscription { + fn poll(&mut self, api_tx: &ApiRequestSender) -> Option<SubscriptionEventEnvelope> { + let pane = pane_get( + format!("{}:pane", self.request_prefix), + &self.pane_id, + api_tx, + ) + .ok()?; + let current_status = pane.agent_status; + let previous_status = self.last_status.replace(current_status); + if previous_status.is_none() || previous_status == Some(current_status) { + return None; + } + if self + .status_filter + .is_some_and(|wanted| wanted != current_status) + { + return None; + } + + Some(SubscriptionEventEnvelope { + event: SubscriptionEventKind::PaneAgentStatusChanged, + data: SubscriptionEventData::PaneAgentStatusChanged(PaneAgentStatusChangedEvent { + pane_id: pane.pane_id, + workspace_id: pane.workspace_id, + agent_status: current_status, + agent: pane.agent, + custom_status: pane.custom_status, + }), + }) + } +} + +fn pane_read( + request_id: String, + pane_id: &str, + source: crate::api::schema::ReadSource, + lines: Option<u32>, + strip_ansi: bool, + api_tx: &ApiRequestSender, +) -> Result<crate::api::schema::PaneReadResult, ErrorResponse> { + let response = dispatch_to_app_with_timeout( + Request { + id: request_id.clone(), + method: Method::PaneRead(crate::api::schema::PaneReadParams { + pane_id: pane_id.to_string(), + source, + lines, + format: crate::api::schema::ReadFormat::Text, + strip_ansi, + }), + }, + api_tx, + Some(APP_RESPONSE_TIMEOUT), + ); + let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse { + id: request_id.clone(), + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode pane read response".into(), + }, + })?; + if value.get("error").is_some() { + return serde_json::from_value(value).map_err(|_| ErrorResponse { + id: request_id, + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode pane read error".into(), + }, + }); + } + serde_json::from_value(value["result"]["read"].clone()).map_err(|_| ErrorResponse { + id: request_id, + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode pane read result".into(), + }, + }) +} + +fn pane_get( + request_id: String, + pane_id: &str, + api_tx: &ApiRequestSender, +) -> Result<crate::api::schema::PaneInfo, ErrorResponse> { + let response = dispatch_to_app_with_timeout( + Request { + id: request_id.clone(), + method: Method::PaneGet(crate::api::schema::PaneTarget { + pane_id: pane_id.to_string(), + }), + }, + api_tx, + Some(APP_RESPONSE_TIMEOUT), + ); + let value: serde_json::Value = serde_json::from_str(&response).map_err(|_| ErrorResponse { + id: request_id.clone(), + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode pane get response".into(), + }, + })?; + if value.get("error").is_some() { + return serde_json::from_value(value).map_err(|_| ErrorResponse { + id: request_id, + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode pane get error".into(), + }, + }); + } + serde_json::from_value(value["result"]["pane"].clone()).map_err(|_| ErrorResponse { + id: request_id, + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode pane get result".into(), + }, + }) +} diff --git a/src/api/wait.rs b/src/api/wait.rs new file mode 100644 index 00000000..4bce14f2 --- /dev/null +++ b/src/api/wait.rs @@ -0,0 +1,124 @@ +use std::os::unix::net::UnixStream; +use std::sync::atomic::AtomicBool; +use std::sync::Arc; + +use regex::Regex; + +use crate::api::schema::{ + ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse, +}; +use crate::api::server::{ + dispatch_to_app_with_timeout, should_stop_connection, APP_RESPONSE_TIMEOUT, + CONNECTION_POLL_INTERVAL, +}; +use crate::api::subscriptions::{match_output, output_match_read_source}; +use crate::api::ApiRequestSender; + +pub(super) fn wait_for_output( + request_id: String, + params: crate::api::schema::PaneWaitForOutputParams, + stream: &mut UnixStream, + api_tx: &ApiRequestSender, + running: &Arc<AtomicBool>, +) -> std::io::Result<Option<String>> { + crate::logging::api_wait_started(&request_id, ¶ms.pane_id, params.timeout_ms); + let deadline = params + .timeout_ms + .map(|ms| std::time::Instant::now() + std::time::Duration::from_millis(ms)); + + let regex = match ¶ms.r#match { + crate::api::schema::OutputMatch::Regex { value } => match Regex::new(value) { + Ok(regex) => Some(regex), + Err(err) => { + return Ok(Some( + serde_json::to_string(&ErrorResponse { + id: request_id, + error: ErrorBody { + code: "invalid_regex".into(), + message: err.to_string(), + }, + }) + .unwrap(), + )); + } + }, + crate::api::schema::OutputMatch::Substring { .. } => None, + }; + + loop { + if should_stop_connection(stream, running)? { + crate::logging::api_wait_completed(&request_id, ¶ms.pane_id, "client_disconnected"); + return Ok(None); + } + + let read_request = Request { + id: format!("{request_id}:read"), + method: Method::PaneRead(crate::api::schema::PaneReadParams { + pane_id: params.pane_id.clone(), + source: output_match_read_source(¶ms.source), + lines: params.lines, + format: crate::api::schema::ReadFormat::Text, + strip_ansi: params.strip_ansi, + }), + }; + let response = + dispatch_to_app_with_timeout(read_request, api_tx, Some(APP_RESPONSE_TIMEOUT)); + let Ok(value) = serde_json::from_str::<serde_json::Value>(&response) else { + return Ok(Some(response)); + }; + if value.get("error").is_some() { + let mut value = value; + value["id"] = serde_json::Value::String(request_id.clone()); + return Ok(Some(serde_json::to_string(&value).unwrap())); + } + + let read_value = value["result"]["read"].clone(); + let Ok(read) = serde_json::from_value::<crate::api::schema::PaneReadResult>(read_value) + else { + return Ok(Some( + serde_json::to_string(&ErrorResponse { + id: request_id, + error: ErrorBody { + code: "internal_error".into(), + message: "failed to decode pane read result".into(), + }, + }) + .unwrap(), + )); + }; + + let matched_line = match_output(&read.text, ¶ms.r#match, regex.as_ref()); + if matched_line.is_some() { + let revision = read.revision; + crate::logging::api_wait_completed(&request_id, ¶ms.pane_id, "matched"); + return Ok(Some( + serde_json::to_string(&SuccessResponse { + id: request_id, + result: ResponseResult::OutputMatched { + pane_id: params.pane_id, + revision, + matched_line, + read, + }, + }) + .unwrap(), + )); + } + + if deadline.is_some_and(|deadline| std::time::Instant::now() >= deadline) { + crate::logging::api_wait_timed_out(&request_id, ¶ms.pane_id); + return Ok(Some( + serde_json::to_string(&ErrorResponse { + id: request_id, + error: ErrorBody { + code: "timeout".into(), + message: "timed out waiting for output match".into(), + }, + }) + .unwrap(), + )); + } + + std::thread::sleep(CONNECTION_POLL_INTERVAL); + } +} |