diff options
Diffstat (limited to 'src/api/mod.rs')
| -rw-r--r-- | src/api/mod.rs | 504 |
1 files changed, 474 insertions, 30 deletions
diff --git a/src/api/mod.rs b/src/api/mod.rs index 642dc70d..e2c29490 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -4,13 +4,16 @@ use std::fs; use std::io::{BufRead, BufReader, Write}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; +use std::time::Duration; use tracing::{debug, error, info, warn}; use regex::Regex; use crate::api::schema::{ - ErrorBody, ErrorResponse, Method, Request, ResponseResult, SuccessResponse, + ErrorBody, ErrorResponse, Method, PaneAgentStateChangedEvent, PaneOutputMatchedEvent, Request, + ResponseResult, Subscription, SubscriptionEventData, SubscriptionEventEnvelope, + SubscriptionEventKind, SuccessResponse, }; pub const SOCKET_PATH_ENV_VAR: &str = "HERDR_SOCKET_PATH"; @@ -20,6 +23,46 @@ pub struct ApiRequestMessage { pub respond_to: std::sync::mpsc::Sender<String>, } +#[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 { if let Ok(path) = std::env::var(SOCKET_PATH_ENV_VAR) { return PathBuf::from(path); @@ -57,6 +100,7 @@ impl Drop for ServerHandle { pub fn start_server( api_tx: std::sync::mpsc::Sender<ApiRequestMessage>, + event_hub: EventHub, ) -> std::io::Result<ServerHandle> { let path = socket_path(); prepare_socket_path(&path)?; @@ -68,9 +112,13 @@ pub fn start_server( for stream in listener.incoming() { match stream { Ok(stream) => { - if let Err(err) = handle_connection(stream, &api_tx) { - warn!(err = %err, "api connection failed"); - } + let api_tx = api_tx.clone(); + let event_hub = event_hub.clone(); + std::thread::spawn(move || { + if let Err(err) = handle_connection(stream, &api_tx, &event_hub) { + warn!(err = %err, "api connection failed"); + } + }); } Err(err) => { error!(err = %err, "api listener accept failed"); @@ -104,6 +152,7 @@ fn prepare_socket_path(path: &Path) -> std::io::Result<()> { fn handle_connection( mut stream: UnixStream, api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>, + event_hub: &EventHub, ) -> std::io::Result<()> { let mut line = String::new(); { @@ -119,21 +168,41 @@ fn handle_connection( return Ok(()); } - let response = match serde_json::from_str::<Request>(line) { - Ok(request) => handle_request(request, api_tx), - Err(err) => serde_json::to_string(&ErrorResponse { - id: String::new(), - error: ErrorBody { - code: "invalid_request".into(), - message: format!("invalid request: {err}"), - }, - })?, + let request = match serde_json::from_str::<Request>(line) { + Ok(request) => request, + Err(err) => { + write_json_line( + &mut stream, + &ErrorResponse { + id: String::new(), + error: ErrorBody { + code: "invalid_request".into(), + message: format!("invalid request: {err}"), + }, + }, + )?; + return Ok(()); + } }; - stream.write_all(response.as_bytes())?; - stream.write_all(b"\n")?; - stream.flush()?; - Ok(()) + match request.method { + Method::EventsSubscribe(params) => { + stream_subscriptions(stream, request.id, params, api_tx, event_hub) + } + method => { + let response = handle_request( + Request { + id: request.id, + method, + }, + api_tx, + ); + stream.write_all(response.as_bytes())?; + stream.write_all(b"\n")?; + stream.flush()?; + Ok(()) + } + } } fn handle_request(request: Request, api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>) -> String { @@ -213,19 +282,7 @@ fn wait_for_output( .unwrap(); }; - let matched_line = match ¶ms.r#match { - crate::api::schema::OutputMatch::Substring { value } => read - .text - .lines() - .find(|line| line.contains(value)) - .map(|line| line.to_string()), - crate::api::schema::OutputMatch::Regex { .. } => regex.as_ref().and_then(|re| { - read.text - .lines() - .find(|line| re.is_match(line)) - .map(|line| line.to_string()) - }), - }; + let matched_line = match_output(&read.text, ¶ms.r#match, regex.as_ref()); if matched_line.is_some() { let revision = read.revision; return serde_json::to_string(&SuccessResponse { @@ -255,6 +312,393 @@ fn wait_for_output( } } +fn stream_subscriptions( + mut stream: UnixStream, + request_id: String, + params: crate::api::schema::EventsSubscribeParams, + api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>, + event_hub: &EventHub, +) -> 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) => { + write_json_line(&mut stream, &response)?; + return Ok(()); + } + }; + subscriptions.push(active); + } + + write_json_line( + &mut stream, + &SuccessResponse { + id: request_id, + result: ResponseResult::SubscriptionStarted {}, + }, + )?; + + loop { + for subscription in &mut subscriptions { + if let Some(event) = subscription.poll(api_tx, event_hub) { + write_json_line(&mut stream, &event)?; + } + } + std::thread::sleep(Duration::from_millis(100)); + } +} + +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}")))?; + stream.write_all(encoded.as_bytes())?; + stream.write_all(b"\n")?; + stream.flush()?; + Ok(()) +} + +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 ActiveAgentStateChangedSubscription { + pane_id: String, + state_filter: Option<crate::api::schema::PaneAgentState>, + last_state: Option<crate::api::schema::PaneAgentState>, + request_prefix: String, +} + +struct ActiveEventSubscription { + event_kind: crate::api::schema::EventKind, + last_sequence: u64, +} + +enum ActiveSubscription { + Event(ActiveEventSubscription), + OutputMatched(ActiveOutputMatchedSubscription), + AgentStateChanged(ActiveAgentStateChangedSubscription), +} + +impl ActiveSubscription { + fn new( + subscription: Subscription, + request_id: &str, + index: usize, + api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>, + _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::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.clone(), + lines, + strip_ansi, + api_tx, + ); + if let Err(error) = probe { + return Err(error); + } + + 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::PaneAgentStateChanged { pane_id, state } => { + let probe = + match pane_get(format!("{request_id}:sub:{index}:probe"), &pane_id, api_tx) { + Ok(probe) => probe, + Err(error) => return Err(error), + }; + + Ok(Self::AgentStateChanged( + ActiveAgentStateChangedSubscription { + pane_id, + state_filter: state, + last_state: Some(probe.agent_state), + request_prefix: format!("{request_id}:sub:{index}"), + }, + )) + } + } + } + + fn poll( + &mut self, + api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>, + 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::AgentStateChanged(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: &std::sync::mpsc::Sender<ApiRequestMessage>, + ) -> Option<SubscriptionEventEnvelope> { + let read = pane_read( + format!("{}:read", self.request_prefix), + &self.pane_id, + self.source.clone(), + 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 ActiveAgentStateChangedSubscription { + fn poll( + &mut self, + api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>, + ) -> Option<SubscriptionEventEnvelope> { + let pane = pane_get( + format!("{}:pane", self.request_prefix), + &self.pane_id, + api_tx, + ) + .ok()?; + let current_state = pane.agent_state; + let previous_state = self.last_state.replace(current_state); + if previous_state.is_none() || previous_state == Some(current_state) { + return None; + } + if self + .state_filter + .is_some_and(|wanted| wanted != current_state) + { + return None; + } + + Some(SubscriptionEventEnvelope { + event: SubscriptionEventKind::PaneAgentStateChanged, + data: SubscriptionEventData::PaneAgentStateChanged(PaneAgentStateChangedEvent { + pane_id: pane.pane_id, + workspace_id: pane.workspace_id, + state: current_state, + agent: pane.agent, + }), + }) + } +} + +fn pane_read( + request_id: String, + pane_id: &str, + source: crate::api::schema::ReadSource, + lines: Option<u32>, + strip_ansi: bool, + api_tx: &std::sync::mpsc::Sender<ApiRequestMessage>, +) -> Result<crate::api::schema::PaneReadResult, ErrorResponse> { + let response = dispatch_to_app( + Request { + id: request_id.clone(), + method: Method::PaneRead(crate::api::schema::PaneReadParams { + pane_id: pane_id.to_string(), + source, + lines, + strip_ansi, + }), + }, + api_tx, + ); + 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: &std::sync::mpsc::Sender<ApiRequestMessage>, +) -> Result<crate::api::schema::PaneInfo, ErrorResponse> { + let response = dispatch_to_app( + Request { + id: request_id.clone(), + method: Method::PaneGet(crate::api::schema::PaneTarget { + pane_id: pane_id.to_string(), + }), + }, + api_tx, + ); + 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: &std::sync::mpsc::Sender<ApiRequestMessage>, |