refactor(maestro): route every client through the shared DaemonClient
ci / rust (pull_request) Successful in 3m0s
ci / rust (pull_request) Successful in 3m0s
The CLI, TUI and end-to-end harness each carried their own client: three reqwest builders with three timeout policies, three copies of the success-or-body check, and route strings at every call site. Two of them discarded the daemon's structured error and reported only a status code. Deletes `maestro-cli/src/http.rs`, both `ws_base` copies in the TUI, the CLI's partial `DaemonInfo` (it dropped `version` and `presets`), and the e2e harness's `check_ok`. `maestro-web` gains a protocol dependency and stops building the error envelope with a format string and reading it back as untyped JSON; the daemon and the proxy now register routes from the same constants. `maestro_web::router` returns `Result` — an unusable daemon URL fails at startup instead of at first connect. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
01a1073943
commit
02356f84ab
Generated
+3
-1
@@ -1067,6 +1067,8 @@ dependencies = [
|
||||
"maestro-config",
|
||||
"maestro-protocol",
|
||||
"nix",
|
||||
"reqwest",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
"thiserror 1.0.69",
|
||||
@@ -1170,6 +1172,7 @@ dependencies = [
|
||||
"maestro-config",
|
||||
"maestro-protocol",
|
||||
"ratatui",
|
||||
"reqwest",
|
||||
"serde_json",
|
||||
"thiserror 1.0.69",
|
||||
"tokio",
|
||||
@@ -1190,7 +1193,6 @@ dependencies = [
|
||||
"futures-util",
|
||||
"include_dir",
|
||||
"maestro-config",
|
||||
"maestro-protocol",
|
||||
"mime_guess",
|
||||
"reqwest",
|
||||
"serde_json",
|
||||
|
||||
@@ -17,10 +17,8 @@ dirs.workspace = true
|
||||
futures-util.workspace = true
|
||||
libc.workspace = true
|
||||
maestro-config.workspace = true
|
||||
maestro-protocol.workspace = true
|
||||
maestro-protocol = { workspace = true, features = ["client"] }
|
||||
nix.workspace = true
|
||||
reqwest.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
thiserror.workspace = true
|
||||
tokio = { workspace = true, features = ["macros", "rt", "sync", "time", "io-util", "io-std", "signal"] }
|
||||
|
||||
@@ -13,7 +13,6 @@ use maestro_protocol::ClientControl;
|
||||
use nix::sys::termios::{cfmakeraw, tcgetattr, tcsetattr, SetArg, Termios};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
#[allow(clippy::result_large_err)]
|
||||
@@ -24,8 +23,8 @@ pub enum AttachError {
|
||||
Io(#[from] io::Error),
|
||||
#[error("termios: {0}")]
|
||||
Termios(#[from] nix::Error),
|
||||
#[error("invalid daemon_url: {0}")]
|
||||
InvalidDaemonUrl(String),
|
||||
#[error(transparent)]
|
||||
Url(#[from] maestro_protocol::routes::UrlError),
|
||||
#[error("invalid detach key: {0}")]
|
||||
InvalidDetachKey(String),
|
||||
#[error("stdin must be a TTY for attach")]
|
||||
@@ -63,26 +62,6 @@ pub fn parse_detach_key(spec: &str) -> Result<u8, AttachError> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the attach WS URL from the daemon's HTTP base.
|
||||
///
|
||||
/// The endpoint (`WS /api/attach/:id`) lives on the daemon's `bind_addr`, so we
|
||||
/// take the HTTP base (`http://127.0.0.1:7070`), swap the scheme to its WS
|
||||
/// equivalent, and append the route.
|
||||
#[allow(clippy::result_large_err)]
|
||||
pub fn attach_url(daemon_http_base: &str, session_id: Uuid) -> Result<String, AttachError> {
|
||||
let base = daemon_http_base.trim_end_matches('/');
|
||||
let ws_base = if let Some(rest) = base.strip_prefix("https://") {
|
||||
format!("wss://{rest}")
|
||||
} else if let Some(rest) = base.strip_prefix("http://") {
|
||||
format!("ws://{rest}")
|
||||
} else if base.starts_with("ws://") || base.starts_with("wss://") {
|
||||
base.to_string()
|
||||
} else {
|
||||
return Err(AttachError::InvalidDaemonUrl(daemon_http_base.to_string()));
|
||||
};
|
||||
Ok(format!("{ws_base}/api/attach/{session_id}"))
|
||||
}
|
||||
|
||||
/// RAII guard that puts the terminal into raw mode and restores on drop.
|
||||
#[derive(Debug)]
|
||||
struct RawTty {
|
||||
@@ -234,7 +213,6 @@ pub async fn run_attach_url(url: &str, detach_byte: u8) -> Result<(), AttachErro
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[test]
|
||||
fn parse_detach_key_handles_caret_form() {
|
||||
@@ -255,30 +233,4 @@ mod tests {
|
||||
Err(AttachError::InvalidDetachKey(_))
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attach_url_swaps_scheme_and_appends_route() {
|
||||
let id = Uuid::from_u128(0xabcd);
|
||||
assert_eq!(
|
||||
attach_url("http://127.0.0.1:7070", id).unwrap(),
|
||||
format!("ws://127.0.0.1:7070/api/attach/{id}")
|
||||
);
|
||||
assert_eq!(
|
||||
attach_url("https://laptop.lan:7070/", id).unwrap(),
|
||||
format!("wss://laptop.lan:7070/api/attach/{id}")
|
||||
);
|
||||
// already-ws bases pass through
|
||||
assert_eq!(
|
||||
attach_url("ws://10.0.0.1:7070", id).unwrap(),
|
||||
format!("ws://10.0.0.1:7070/api/attach/{id}")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn attach_url_rejects_unknown_scheme() {
|
||||
assert!(matches!(
|
||||
attach_url("ftp://x", Uuid::nil()),
|
||||
Err(AttachError::InvalidDaemonUrl(_))
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,10 +6,11 @@
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use crate::attach::{attach_url, parse_detach_key, run_attach_url};
|
||||
use maestro_protocol::DaemonClient;
|
||||
|
||||
use crate::attach::{parse_detach_key, run_attach_url};
|
||||
use crate::cli::{AttachArgs, Command, KillArgs, ListArgs, SpawnArgs};
|
||||
use crate::config::{default_config_path, CliConfig, ConfigError};
|
||||
use crate::http::DaemonClient;
|
||||
use crate::sessions::{run_kill, run_list, run_spawn};
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
@@ -18,7 +19,7 @@ pub enum CliError {
|
||||
#[error("config: {0}")]
|
||||
Config(#[from] ConfigError),
|
||||
#[error("{0}")]
|
||||
Http(#[from] crate::http::HttpError),
|
||||
Client(#[from] maestro_protocol::ClientError),
|
||||
#[error("{0}")]
|
||||
Attach(#[from] crate::attach::AttachError),
|
||||
#[error("{0}")]
|
||||
@@ -73,7 +74,7 @@ async fn spawn(
|
||||
println!("spawned {}", session.id);
|
||||
|
||||
if attach_after {
|
||||
let url = attach_url(daemon.base(), session.id)?;
|
||||
let url = daemon.attach_url(session.id)?;
|
||||
run_attach_url(&url, detach_key).await?;
|
||||
}
|
||||
Ok(())
|
||||
@@ -87,7 +88,7 @@ async fn attach(
|
||||
) -> Result<(), CliError> {
|
||||
let daemon = client(config_path, daemon_url_override)?;
|
||||
let detach = parse_detach_key(&args.detach_key)?;
|
||||
let url = attach_url(daemon.base(), args.session_id)?;
|
||||
let url = daemon.attach_url(args.session_id)?;
|
||||
run_attach_url(&url, detach).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -1,161 +0,0 @@
|
||||
//! Thin HTTP client wrapper around `reqwest` for the daemon's `/api/*` surface.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use reqwest::{Method, Response, StatusCode};
|
||||
use serde::de::DeserializeOwned;
|
||||
use serde::Serialize;
|
||||
|
||||
use crate::config::ConfigError;
|
||||
|
||||
/// Time budget for establishing a TCP connection to the daemon.
|
||||
const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
/// Total time budget for a single request (connect + send + receive). Bounds
|
||||
/// the case where the peer is TCP-alive but never makes forward progress.
|
||||
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
|
||||
/// Build a `reqwest::Client` with the CLI's standard timeouts and user-agent.
|
||||
///
|
||||
/// Centralizing construction here guarantees every client the CLI uses gets a
|
||||
/// connect *and* request timeout, so `maestro` never hangs forever on an
|
||||
/// unhealthy daemon (NAP-262).
|
||||
pub fn build_client() -> reqwest::Client {
|
||||
build_client_with_timeouts(CONNECT_TIMEOUT, REQUEST_TIMEOUT)
|
||||
}
|
||||
|
||||
fn build_client_with_timeouts(connect: Duration, request: Duration) -> reqwest::Client {
|
||||
reqwest::Client::builder()
|
||||
.user_agent(concat!("maestro-cli/", env!("CARGO_PKG_VERSION")))
|
||||
.connect_timeout(connect)
|
||||
.timeout(request)
|
||||
.build()
|
||||
.expect("build reqwest client")
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum HttpError {
|
||||
#[error("http: {0}")]
|
||||
Reqwest(#[from] reqwest::Error),
|
||||
#[error("server returned {status}: {body}")]
|
||||
ServerError { status: StatusCode, body: String },
|
||||
#[error("decode: {0}")]
|
||||
Json(#[from] serde_json::Error),
|
||||
#[error("config: {0}")]
|
||||
Config(#[from] ConfigError),
|
||||
}
|
||||
|
||||
impl HttpError {
|
||||
/// True when the error is a transport-level failure to reach the host
|
||||
/// (connection refused / connect error) rather than an HTTP-level
|
||||
/// rejection.
|
||||
pub fn is_unreachable(&self) -> bool {
|
||||
matches!(self, HttpError::Reqwest(e) if e.is_connect())
|
||||
}
|
||||
}
|
||||
|
||||
/// HTTP client bound to one daemon's `/api/*` base URL. The daemon's surface
|
||||
/// needs no auth — reaching its bind address is the authorization.
|
||||
#[derive(Debug)]
|
||||
pub struct DaemonClient {
|
||||
base: String,
|
||||
inner: reqwest::Client,
|
||||
}
|
||||
|
||||
impl DaemonClient {
|
||||
pub fn new(base: String) -> Self {
|
||||
Self {
|
||||
base: base.trim_end_matches('/').to_string(),
|
||||
inner: build_client(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn base(&self) -> &str {
|
||||
&self.base
|
||||
}
|
||||
|
||||
async fn send(
|
||||
&self,
|
||||
method: Method,
|
||||
path: &str,
|
||||
body: Option<serde_json::Value>,
|
||||
) -> Result<Response, HttpError> {
|
||||
let url = format!("{}{}", self.base, path);
|
||||
let mut req = self.inner.request(method, &url);
|
||||
if let Some(b) = body {
|
||||
req = req.json(&b);
|
||||
}
|
||||
Ok(req.send().await?)
|
||||
}
|
||||
|
||||
pub async fn get_json<T: DeserializeOwned>(&self, path: &str) -> Result<T, HttpError> {
|
||||
let resp = self.send(Method::GET, path, None).await?;
|
||||
ok_or_body(resp)
|
||||
.await?
|
||||
.json::<T>()
|
||||
.await
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
pub async fn post_json<B: Serialize, T: DeserializeOwned>(
|
||||
&self,
|
||||
path: &str,
|
||||
body: &B,
|
||||
) -> Result<T, HttpError> {
|
||||
let v = serde_json::to_value(body)?;
|
||||
let resp = self.send(Method::POST, path, Some(v)).await?;
|
||||
ok_or_body(resp)
|
||||
.await?
|
||||
.json::<T>()
|
||||
.await
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
pub async fn delete(&self, path: &str) -> Result<(), HttpError> {
|
||||
let resp = self.send(Method::DELETE, path, None).await?;
|
||||
ok_or_body(resp).await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
async fn ok_or_body(resp: Response) -> Result<Response, HttpError> {
|
||||
let status = resp.status();
|
||||
if status.is_success() {
|
||||
return Ok(resp);
|
||||
}
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
Err(HttpError::ServerError { status, body })
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// A peer that accepts the TCP connection but never sends a response must
|
||||
/// not hang the client forever: the request timeout has to fire (NAP-262).
|
||||
#[tokio::test]
|
||||
async fn request_timeout_fires_when_peer_never_responds() {
|
||||
use std::io::Read;
|
||||
use std::net::TcpListener;
|
||||
|
||||
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
// Accept one connection and sit on it without ever writing a response.
|
||||
let handle = std::thread::spawn(move || {
|
||||
if let Ok((mut sock, _)) = listener.accept() {
|
||||
let mut buf = [0u8; 1024];
|
||||
let _ = sock.read(&mut buf);
|
||||
std::thread::sleep(Duration::from_secs(2));
|
||||
}
|
||||
});
|
||||
|
||||
let client = build_client_with_timeouts(Duration::from_secs(5), Duration::from_millis(300));
|
||||
let err = client
|
||||
.get(format!("http://{addr}/"))
|
||||
.send()
|
||||
.await
|
||||
.expect_err("request should time out, not hang");
|
||||
assert!(err.is_timeout(), "expected timeout error, got: {err}");
|
||||
|
||||
let _ = handle.join();
|
||||
}
|
||||
}
|
||||
@@ -6,7 +6,6 @@ pub mod attach;
|
||||
pub mod cli;
|
||||
pub mod config;
|
||||
pub mod dispatch;
|
||||
pub mod http;
|
||||
pub mod sessions;
|
||||
|
||||
pub use dispatch::{dispatch, CliError};
|
||||
|
||||
@@ -3,31 +3,22 @@
|
||||
use std::collections::HashMap;
|
||||
use std::io::Write;
|
||||
|
||||
use maestro_protocol::{ActivityState, CreateSessionRequest, Session, SessionStatus, WorktreeSpec};
|
||||
use serde::Deserialize;
|
||||
use maestro_protocol::{
|
||||
ActivityState, ClientError, CreateSessionRequest, DaemonClient, DaemonInfo, Session,
|
||||
SessionStatus, WorktreeSpec,
|
||||
};
|
||||
use serde_json::Value;
|
||||
|
||||
use crate::cli::{KillArgs, ListArgs, SpawnArgs};
|
||||
use crate::http::{DaemonClient, HttpError};
|
||||
|
||||
/// The bits of `GET /api/info` the CLI renders. Deserialized structurally so
|
||||
/// the CLI doesn't depend on the daemon crate.
|
||||
#[derive(Debug, Clone, Default, Deserialize)]
|
||||
pub struct DaemonInfo {
|
||||
#[serde(default)]
|
||||
pub hostname: String,
|
||||
#[serde(default)]
|
||||
pub home_dir: String,
|
||||
}
|
||||
|
||||
/// `maestro list`
|
||||
pub async fn run_list(
|
||||
client: &DaemonClient,
|
||||
args: &ListArgs,
|
||||
out: &mut impl Write,
|
||||
) -> Result<(), HttpError> {
|
||||
let sessions: Vec<Session> = client.get_json("/api/sessions").await?;
|
||||
let info: DaemonInfo = client.get_json("/api/info").await.unwrap_or_default();
|
||||
) -> Result<(), ClientError> {
|
||||
let sessions = client.list_sessions().await?;
|
||||
let info = client.daemon_info().await.unwrap_or_default();
|
||||
|
||||
if args.json {
|
||||
let body = serde_json::json!({ "sessions": sessions });
|
||||
@@ -40,7 +31,7 @@ pub async fn run_list(
|
||||
}
|
||||
|
||||
/// `maestro spawn`
|
||||
pub async fn run_spawn(client: &DaemonClient, args: SpawnArgs) -> Result<Session, HttpError> {
|
||||
pub async fn run_spawn(client: &DaemonClient, args: SpawnArgs) -> Result<Session, ClientError> {
|
||||
let mut env: HashMap<String, String> = HashMap::new();
|
||||
for (k, v) in args.env.iter() {
|
||||
env.insert(k.clone(), v.clone());
|
||||
@@ -64,15 +55,12 @@ pub async fn run_spawn(client: &DaemonClient, args: SpawnArgs) -> Result<Session
|
||||
metadata,
|
||||
worktree,
|
||||
};
|
||||
let session: Session = client.post_json("/api/sessions", &body).await?;
|
||||
Ok(session)
|
||||
client.create_session(&body).await
|
||||
}
|
||||
|
||||
/// `maestro kill`
|
||||
pub async fn run_kill(client: &DaemonClient, args: &KillArgs) -> Result<(), HttpError> {
|
||||
client
|
||||
.delete(&format!("/api/sessions/{}", args.session_id))
|
||||
.await
|
||||
pub async fn run_kill(client: &DaemonClient, args: &KillArgs) -> Result<(), ClientError> {
|
||||
client.kill_session(args.session_id).await
|
||||
}
|
||||
|
||||
#[allow(clippy::write_literal)]
|
||||
@@ -167,6 +155,7 @@ mod tests {
|
||||
DaemonInfo {
|
||||
hostname: "naps-laptop".into(),
|
||||
home_dir: home.into(),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -19,31 +19,15 @@ use axum::routing::{any, delete, get};
|
||||
use axum::{Json, Router};
|
||||
use maestro_config::SpawnPreset;
|
||||
use maestro_protocol::{
|
||||
CatalogEvent, CreateSessionRequest, ErrorInfo, GitRoot, Session, PROTOCOL_VERSION,
|
||||
routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonInfo, GitRoot, Session,
|
||||
PROTOCOL_VERSION,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde::Deserialize;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::local_events::LocalEventBus;
|
||||
use crate::session::SessionManager;
|
||||
|
||||
/// Daemon identity returned by `GET /api/info`. Clients render this in the
|
||||
/// header ("you're connected to konishi"), plus the daemon's protocol version
|
||||
/// and its spawn presets (#26).
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct DaemonInfo {
|
||||
pub hostname: String,
|
||||
pub version: String,
|
||||
/// Home directory of the OS user this daemon runs as, so clients can
|
||||
/// prefill a spawn form's working directory. Empty if unresolved.
|
||||
pub home_dir: String,
|
||||
/// Spawn-form command presets resolved from `[spawn].presets` (#22), with
|
||||
/// the implicit `custom` tile appended. Clients render these as one-click
|
||||
/// tiles; it rides this existing payload so opening the form needs no extra
|
||||
/// round-trip (#26).
|
||||
pub presets: Vec<SpawnPreset>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct LocalApiState {
|
||||
pub manager: SessionManager,
|
||||
@@ -60,14 +44,14 @@ pub struct LocalApiState {
|
||||
|
||||
pub fn router(state: LocalApiState) -> Router {
|
||||
Router::new()
|
||||
.route("/api/sessions", get(list_sessions).post(create_session))
|
||||
.route("/api/sessions/:id", delete(kill_session))
|
||||
.route("/api/dirs", get(list_dir))
|
||||
.route("/api/recent-dirs", get(recent_dirs))
|
||||
.route("/api/gitroot", get(git_root))
|
||||
.route("/api/info", get(daemon_info))
|
||||
.route("/api/events", any(events_ws))
|
||||
.route("/api/attach/:id", any(attach_ws))
|
||||
.route(routes::SESSIONS, get(list_sessions).post(create_session))
|
||||
.route(routes::patterns::SESSION, delete(kill_session))
|
||||
.route(routes::DIRS, get(list_dir))
|
||||
.route(routes::RECENT_DIRS, get(recent_dirs))
|
||||
.route(routes::GITROOT, get(git_root))
|
||||
.route(routes::INFO, get(daemon_info))
|
||||
.route(routes::EVENTS, any(events_ws))
|
||||
.route(routes::patterns::ATTACH, any(attach_ws))
|
||||
.with_state(state)
|
||||
}
|
||||
|
||||
@@ -167,16 +151,7 @@ async fn kill_session(State(s): State<LocalApiState>, Path(id): Path<Uuid>) -> R
|
||||
}
|
||||
|
||||
fn error_response(status: StatusCode, code: &str, message: &str) -> Response {
|
||||
(
|
||||
status,
|
||||
Json(serde_json::json!({
|
||||
"error": ErrorInfo {
|
||||
code: code.into(),
|
||||
message: message.into(),
|
||||
},
|
||||
})),
|
||||
)
|
||||
.into_response()
|
||||
(status, Json(ApiError::new(code, message))).into_response()
|
||||
}
|
||||
|
||||
/// Convenience constructor — pulls the bits `LocalApiState` needs from
|
||||
|
||||
@@ -13,7 +13,7 @@ path = "src/lib.rs"
|
||||
[dependencies]
|
||||
chrono.workspace = true
|
||||
futures-util.workspace = true
|
||||
maestro-protocol.workspace = true
|
||||
maestro-protocol = { workspace = true, features = ["client"] }
|
||||
reqwest.workspace = true
|
||||
serde.workspace = true
|
||||
serde_json.workspace = true
|
||||
|
||||
@@ -1,13 +1,16 @@
|
||||
//! Thin HTTP+WS client used by tests to drive a daemon's `/api/*` surface.
|
||||
//! Thin test harness over [`maestro_protocol::DaemonClient`].
|
||||
//!
|
||||
//! HTTP calls delegate to the shared client so the harness exercises the same
|
||||
//! request building, timeouts and error decoding the CLI and TUI use. Only the
|
||||
//! WebSocket helpers are test-specific: they hand back a raw stream, or a
|
||||
//! handle that accumulates events in the background.
|
||||
|
||||
#![allow(clippy::result_large_err)]
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use maestro_protocol::{CatalogEvent, CreateSessionRequest, Session};
|
||||
use reqwest::Client as HttpClient;
|
||||
use maestro_protocol::{CatalogEvent, CreateSessionRequest, DaemonClient, Session};
|
||||
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
|
||||
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
||||
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
|
||||
@@ -16,73 +19,42 @@ use uuid::Uuid;
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
#[allow(clippy::result_large_err)]
|
||||
pub enum ClientError {
|
||||
#[error("http: {0}")]
|
||||
Http(#[from] reqwest::Error),
|
||||
#[error(transparent)]
|
||||
Api(#[from] maestro_protocol::ClientError),
|
||||
#[error("ws: {0}")]
|
||||
Ws(#[from] tokio_tungstenite::tungstenite::Error),
|
||||
#[error("json: {0}")]
|
||||
Json(#[from] serde_json::Error),
|
||||
#[error("unexpected status {status}: {body}")]
|
||||
Status { status: u16, body: String },
|
||||
}
|
||||
|
||||
/// HTTP client plus a few WS-aware helpers, bound to one daemon.
|
||||
/// Daemon client plus a few WS-aware helpers, bound to one daemon.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TestClient {
|
||||
http_base: String,
|
||||
ws_base: String,
|
||||
http: HttpClient,
|
||||
daemon: DaemonClient,
|
||||
}
|
||||
|
||||
impl TestClient {
|
||||
pub fn new(http_base: &str, ws_base: &str) -> Self {
|
||||
let http = HttpClient::builder()
|
||||
.timeout(Duration::from_secs(5))
|
||||
.build()
|
||||
.expect("reqwest client");
|
||||
pub fn new(http_base: &str) -> Self {
|
||||
Self {
|
||||
http_base: http_base.trim_end_matches('/').to_string(),
|
||||
ws_base: ws_base.trim_end_matches('/').to_string(),
|
||||
http,
|
||||
daemon: DaemonClient::new(http_base),
|
||||
}
|
||||
}
|
||||
|
||||
/// The shared client underneath, for calls this harness doesn't wrap.
|
||||
pub fn daemon(&self) -> &DaemonClient {
|
||||
&self.daemon
|
||||
}
|
||||
|
||||
pub async fn list_sessions(&self) -> Result<Vec<Session>, ClientError> {
|
||||
let r = self
|
||||
.http
|
||||
.get(format!("{}/api/sessions", self.http_base))
|
||||
.send()
|
||||
.await?;
|
||||
check_ok(&r)?;
|
||||
Ok(r.json().await?)
|
||||
Ok(self.daemon.list_sessions().await?)
|
||||
}
|
||||
|
||||
pub async fn spawn_session(&self, req: &CreateSessionRequest) -> Result<Session, ClientError> {
|
||||
let r = self
|
||||
.http
|
||||
.post(format!("{}/api/sessions", self.http_base))
|
||||
.json(req)
|
||||
.send()
|
||||
.await?;
|
||||
let status = r.status();
|
||||
if !status.is_success() {
|
||||
let body = r.text().await.unwrap_or_default();
|
||||
return Err(ClientError::Status {
|
||||
status: status.as_u16(),
|
||||
body,
|
||||
});
|
||||
}
|
||||
Ok(r.json().await?)
|
||||
Ok(self.daemon.create_session(req).await?)
|
||||
}
|
||||
|
||||
pub async fn kill_session(&self, id: Uuid) -> Result<(), ClientError> {
|
||||
let r = self
|
||||
.http
|
||||
.delete(format!("{}/api/sessions/{id}", self.http_base))
|
||||
.send()
|
||||
.await?;
|
||||
check_ok(&r)?;
|
||||
Ok(())
|
||||
Ok(self.daemon.kill_session(id).await?)
|
||||
}
|
||||
|
||||
/// Connect to the daemon's per-session WS and return the stream. The
|
||||
@@ -91,8 +63,7 @@ impl TestClient {
|
||||
&self,
|
||||
id: Uuid,
|
||||
) -> Result<WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>, ClientError> {
|
||||
let url = format!("{}/api/attach/{id}", self.ws_base);
|
||||
let request = url.into_client_request()?;
|
||||
let request = self.daemon.attach_url(id)?.into_client_request()?;
|
||||
let (ws, _resp) = connect_async(request).await?;
|
||||
Ok(ws)
|
||||
}
|
||||
@@ -100,7 +71,7 @@ impl TestClient {
|
||||
/// Subscribe to the daemon's `/api/events` WS. Returns a handle whose
|
||||
/// background task pushes parsed events into a shared vector.
|
||||
pub async fn subscribe_events(&self) -> EventStream {
|
||||
let url = format!("{}/api/events", self.ws_base);
|
||||
let url = self.daemon.events_url().expect("events url");
|
||||
let request = url.into_client_request().expect("ws req");
|
||||
let (ws, _resp) = connect_async(request).await.expect("events ws");
|
||||
|
||||
@@ -125,17 +96,6 @@ impl TestClient {
|
||||
}
|
||||
}
|
||||
|
||||
fn check_ok(r: &reqwest::Response) -> Result<(), ClientError> {
|
||||
let status = r.status();
|
||||
if !status.is_success() {
|
||||
return Err(ClientError::Status {
|
||||
status: status.as_u16(),
|
||||
body: format!("HTTP {status}"),
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Handle to a background task draining the events WS into a vector.
|
||||
#[derive(Debug)]
|
||||
pub struct EventStream {
|
||||
|
||||
@@ -50,9 +50,9 @@ impl Daemon {
|
||||
format!("http://{}", self.bind_addr)
|
||||
}
|
||||
|
||||
/// WS base URL of this daemon's API.
|
||||
pub fn ws_base(&self) -> String {
|
||||
format!("ws://{}", self.bind_addr)
|
||||
/// Shared client bound to this daemon.
|
||||
pub fn client(&self) -> maestro_protocol::DaemonClient {
|
||||
maestro_protocol::DaemonClient::new(self.http_base())
|
||||
}
|
||||
|
||||
/// Stop the underlying process, but keep its state dirs around so we can
|
||||
|
||||
@@ -21,7 +21,7 @@ use tokio_tungstenite::tungstenite::Message as WsMessage;
|
||||
|
||||
async fn bring_up() -> (Daemon, TestClient) {
|
||||
let daemon = Daemon::spawn().await;
|
||||
let client = TestClient::new(&daemon.http_base(), &daemon.ws_base());
|
||||
let client = TestClient::new(&daemon.http_base());
|
||||
(daemon, client)
|
||||
}
|
||||
|
||||
@@ -396,7 +396,7 @@ async fn daemon_events_ws_streams_snapshot_then_session_added() {
|
||||
let (daemon, client) = bring_up().await;
|
||||
let cwd = TempDir::new().unwrap();
|
||||
|
||||
let ws_url = format!("{}/api/events", daemon.ws_base());
|
||||
let ws_url = daemon.client().events_url().unwrap();
|
||||
let (mut ws, _) = tokio_tungstenite::connect_async(ws_url)
|
||||
.await
|
||||
.expect("events WS handshake");
|
||||
|
||||
@@ -38,7 +38,7 @@ where
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn catalog_bootstrap_against_real_daemon() {
|
||||
let daemon = Daemon::spawn().await;
|
||||
let ws_url = format!("{}/api/events", daemon.ws_base());
|
||||
let ws_url = daemon.client().events_url().unwrap();
|
||||
|
||||
let store = CatalogStore::default();
|
||||
let bg = store.clone();
|
||||
@@ -54,8 +54,7 @@ async fn catalog_bootstrap_against_real_daemon() {
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn post_session_propagates_through_events_ws() {
|
||||
let daemon = Daemon::spawn().await;
|
||||
let ws_url = format!("{}/api/events", daemon.ws_base());
|
||||
let http_base = daemon.http_base();
|
||||
let ws_url = daemon.client().events_url().unwrap();
|
||||
|
||||
let store = CatalogStore::default();
|
||||
let bg = store.clone();
|
||||
@@ -73,13 +72,11 @@ async fn post_session_propagates_through_events_ws() {
|
||||
metadata: Default::default(),
|
||||
worktree: None,
|
||||
};
|
||||
let res = reqwest::Client::new()
|
||||
.post(format!("{http_base}/api/sessions"))
|
||||
.json(&req)
|
||||
.send()
|
||||
daemon
|
||||
.client()
|
||||
.create_session(&req)
|
||||
.await
|
||||
.expect("post sessions");
|
||||
assert!(res.status().is_success(), "POST /api/sessions: {res:?}");
|
||||
.expect("create session");
|
||||
|
||||
wait_until(
|
||||
"session to surface in catalog",
|
||||
@@ -96,8 +93,7 @@ async fn post_session_propagates_through_events_ws() {
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn delete_session_propagates_removal() {
|
||||
let daemon = Daemon::spawn().await;
|
||||
let ws_url = format!("{}/api/events", daemon.ws_base());
|
||||
let http_base = daemon.http_base();
|
||||
let ws_url = daemon.client().events_url().unwrap();
|
||||
|
||||
let store = CatalogStore::default();
|
||||
let bg = store.clone();
|
||||
@@ -115,14 +111,8 @@ async fn delete_session_propagates_removal() {
|
||||
metadata: Default::default(),
|
||||
worktree: None,
|
||||
};
|
||||
let client = reqwest::Client::new();
|
||||
let res = client
|
||||
.post(format!("{http_base}/api/sessions"))
|
||||
.json(&req)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(res.status().is_success());
|
||||
let client = daemon.client();
|
||||
client.create_session(&req).await.expect("create session");
|
||||
|
||||
wait_until("creation", Duration::from_secs(2), || async {
|
||||
store.snapshot().sessions.len() == 1
|
||||
@@ -130,12 +120,7 @@ async fn delete_session_propagates_removal() {
|
||||
.await;
|
||||
|
||||
let sid = store.snapshot().sessions[0].id;
|
||||
let res = client
|
||||
.delete(format!("{http_base}/api/sessions/{sid}"))
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(res.status().is_success(), "DELETE /api/sessions: {res:?}");
|
||||
client.kill_session(sid).await.expect("kill session");
|
||||
|
||||
wait_until("removal", Duration::from_secs(2), || async {
|
||||
store.snapshot().sessions.is_empty()
|
||||
|
||||
@@ -12,9 +12,8 @@ clap.workspace = true
|
||||
crossterm.workspace = true
|
||||
futures-util.workspace = true
|
||||
maestro-config.workspace = true
|
||||
maestro-protocol.workspace = true
|
||||
maestro-protocol = { workspace = true, features = ["client"] }
|
||||
ratatui.workspace = true
|
||||
reqwest.workspace = true
|
||||
serde_json.workspace = true
|
||||
thiserror.workspace = true
|
||||
tokio = { workspace = true, features = ["sync", "time", "macros", "rt"] }
|
||||
|
||||
@@ -16,7 +16,7 @@ use std::time::Duration;
|
||||
|
||||
use crossterm::event::{Event as CtEvent, EventStream};
|
||||
use futures_util::StreamExt;
|
||||
use maestro_protocol::CreateSessionRequest;
|
||||
use maestro_protocol::{CreateSessionRequest, DaemonClient};
|
||||
use ratatui::backend::CrosstermBackend;
|
||||
use ratatui::Terminal;
|
||||
use uuid::Uuid;
|
||||
@@ -52,23 +52,22 @@ pub struct App {
|
||||
spawn: SpawnModal,
|
||||
help: HelpOverlay,
|
||||
chord: ChordState,
|
||||
daemon_base_url: String,
|
||||
daemon: DaemonClient,
|
||||
region: RegionMap,
|
||||
last_sent_size: HashMap<Uuid, (u16, u16)>,
|
||||
http: reqwest::Client,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for App {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("App")
|
||||
.field("daemon_base_url", &self.daemon_base_url)
|
||||
.field("daemon", &self.daemon.base())
|
||||
.field("state", &self.state)
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
impl App {
|
||||
pub fn new(cfg: TuiConfig, catalog: CatalogStore, daemon_base_url: String) -> Self {
|
||||
pub fn new(cfg: TuiConfig, catalog: CatalogStore, daemon: DaemonClient) -> Self {
|
||||
let chord_timeout = cfg.chord_timeout;
|
||||
let help = HelpOverlay {
|
||||
lines: render_help_lines(&cfg),
|
||||
@@ -81,13 +80,9 @@ impl App {
|
||||
spawn: SpawnModal::default(),
|
||||
help,
|
||||
chord: ChordState::new(chord_timeout),
|
||||
daemon_base_url,
|
||||
daemon,
|
||||
region: RegionMap::default(),
|
||||
last_sent_size: HashMap::new(),
|
||||
http: reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(5))
|
||||
.build()
|
||||
.expect("reqwest client"),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -225,10 +220,17 @@ impl App {
|
||||
Action::AttachHighlighted | Action::ToggleAttachHighlighted => {
|
||||
if let Some(sid) = self.pick_sidebar_target() {
|
||||
if self.emulators.get(sid).is_none() {
|
||||
let ws_url = format!("{}/api/attach/{sid}", ws_base(&self.daemon_base_url));
|
||||
match self.daemon.attach_url(sid) {
|
||||
Ok(ws_url) => {
|
||||
let handle = SessionTask::spawn(ws_url, 24, 80);
|
||||
self.emulators.insert(sid, handle);
|
||||
}
|
||||
Err(e) => {
|
||||
self.state.overlay =
|
||||
Some(OverlayKind::Toast(format!("attach failed: {e}")));
|
||||
}
|
||||
}
|
||||
}
|
||||
self.state.attach_to_focused(sid);
|
||||
}
|
||||
}
|
||||
@@ -294,16 +296,8 @@ impl App {
|
||||
metadata: Default::default(),
|
||||
worktree: None,
|
||||
};
|
||||
let url = format!("{}/api/sessions", self.daemon_base_url);
|
||||
match self.http.post(&url).json(&req).send().await {
|
||||
Ok(res) if res.status().is_success() => {
|
||||
self.state.overlay = None;
|
||||
}
|
||||
Ok(res) => {
|
||||
let status = res.status();
|
||||
self.state.overlay =
|
||||
Some(OverlayKind::Toast(format!("spawn failed: HTTP {status}")));
|
||||
}
|
||||
match self.daemon.create_session(&req).await {
|
||||
Ok(_) => self.state.overlay = None,
|
||||
Err(e) => {
|
||||
self.state.overlay = Some(OverlayKind::Toast(format!("spawn failed: {e}")));
|
||||
}
|
||||
@@ -311,17 +305,11 @@ impl App {
|
||||
}
|
||||
|
||||
async fn submit_kill(&mut self, sid: Uuid) {
|
||||
let url = format!("{}/api/sessions/{sid}", self.daemon_base_url);
|
||||
match self.http.delete(&url).send().await {
|
||||
Ok(res) if res.status().is_success() => {
|
||||
match self.daemon.kill_session(sid).await {
|
||||
Ok(()) => {
|
||||
self.emulators.remove(sid);
|
||||
self.last_sent_size.remove(&sid);
|
||||
}
|
||||
Ok(res) => {
|
||||
let status = res.status();
|
||||
self.state.overlay =
|
||||
Some(OverlayKind::Toast(format!("kill failed: HTTP {status}")));
|
||||
}
|
||||
Err(e) => {
|
||||
self.state.overlay = Some(OverlayKind::Toast(format!("kill failed: {e}")));
|
||||
}
|
||||
@@ -405,16 +393,6 @@ fn visible_sidebar_order(snap: &crate::catalog::CatalogSnapshot) -> Vec<Uuid> {
|
||||
snap.sessions.iter().map(|s| s.id).collect()
|
||||
}
|
||||
|
||||
fn ws_base(http: &str) -> String {
|
||||
if let Some(rest) = http.strip_prefix("http://") {
|
||||
format!("ws://{rest}")
|
||||
} else if let Some(rest) = http.strip_prefix("https://") {
|
||||
format!("wss://{rest}")
|
||||
} else {
|
||||
http.to_string()
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the lines shown by the help overlay from the live keymap, so the
|
||||
/// user always sees the bindings that are actually active (defaults + their
|
||||
/// own `[tui.keymap.*]` overrides).
|
||||
@@ -447,7 +425,11 @@ mod tests {
|
||||
|
||||
fn make_app() -> App {
|
||||
let cfg = TuiConfig::load(&TuiSection::default()).unwrap();
|
||||
App::new(cfg, CatalogStore::default(), "http://127.0.0.1:7070".into())
|
||||
App::new(
|
||||
cfg,
|
||||
CatalogStore::default(),
|
||||
DaemonClient::new("http://127.0.0.1:7070"),
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -22,6 +22,7 @@ pub mod terminal;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
use maestro_protocol::DaemonClient;
|
||||
use ratatui::backend::CrosstermBackend;
|
||||
use ratatui::Terminal;
|
||||
|
||||
@@ -61,8 +62,14 @@ pub async fn run(cfg: RunConfig) -> Result<(), TuiError> {
|
||||
|
||||
let base_url = cfg.daemon_url.trim_end_matches('/').to_string();
|
||||
|
||||
let daemon = DaemonClient::with_timeouts(
|
||||
&base_url,
|
||||
HEALTH_TIMEOUT,
|
||||
maestro_protocol::client::DEFAULT_REQUEST_TIMEOUT,
|
||||
);
|
||||
|
||||
// Health check — fail fast if no daemon is reachable.
|
||||
health_check(&base_url).await?;
|
||||
health_check(&daemon).await?;
|
||||
|
||||
// Load runtime config: explicit `--config <path>` wins, then
|
||||
// `$MAESTRO_CONFIG`, then the platform default. Missing file is fine —
|
||||
@@ -78,7 +85,9 @@ pub async fn run(cfg: RunConfig) -> Result<(), TuiError> {
|
||||
let tui_cfg = TuiConfig::load(&cfg_file.tui).map_err(|e| TuiError::Config(e.to_string()))?;
|
||||
|
||||
let catalog = CatalogStore::default();
|
||||
let ws_url = format!("{}/api/events", ws_base(&base_url));
|
||||
let ws_url = daemon
|
||||
.events_url()
|
||||
.map_err(|e| TuiError::Config(e.to_string()))?;
|
||||
let bg_catalog = catalog.clone();
|
||||
tokio::spawn(async move { bg_catalog.run_events_loop(ws_url).await });
|
||||
|
||||
@@ -92,33 +101,20 @@ pub async fn run(cfg: RunConfig) -> Result<(), TuiError> {
|
||||
let backend = CrosstermBackend::new(std::io::stdout());
|
||||
let mut terminal = Terminal::new(backend).map_err(|e| TuiError::Terminal(e.to_string()))?;
|
||||
|
||||
let app = App::new(tui_cfg, catalog, base_url);
|
||||
let app = App::new(tui_cfg, catalog, daemon);
|
||||
app.run(&mut terminal).await
|
||||
}
|
||||
|
||||
async fn health_check(base_url: &str) -> Result<(), TuiError> {
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(Duration::from_millis(500))
|
||||
.build()
|
||||
.expect("reqwest client");
|
||||
let url = format!("{base_url}/api/info");
|
||||
/// Startup probe budget. Deliberately far below the client default: the user is
|
||||
/// staring at a terminal, so an unreachable daemon must fail in under a second.
|
||||
const HEALTH_TIMEOUT: Duration = Duration::from_millis(500);
|
||||
|
||||
async fn health_check(daemon: &DaemonClient) -> Result<(), TuiError> {
|
||||
for _ in 0..3 {
|
||||
if let Ok(res) = client.get(&url).send().await {
|
||||
if res.status().is_success() {
|
||||
if daemon.daemon_info().await.is_ok() {
|
||||
return Ok(());
|
||||
}
|
||||
tokio::time::sleep(HEALTH_TIMEOUT).await;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
Err(TuiError::DaemonUnreachable(base_url.to_string()))
|
||||
}
|
||||
|
||||
fn ws_base(http: &str) -> String {
|
||||
if let Some(rest) = http.strip_prefix("http://") {
|
||||
format!("ws://{rest}")
|
||||
} else if let Some(rest) = http.strip_prefix("https://") {
|
||||
format!("wss://{rest}")
|
||||
} else {
|
||||
http.to_string()
|
||||
}
|
||||
Err(TuiError::DaemonUnreachable(daemon.base().to_string()))
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ dirs.workspace = true
|
||||
futures-util.workspace = true
|
||||
include_dir.workspace = true
|
||||
maestro-config.workspace = true
|
||||
maestro-protocol.workspace = true
|
||||
mime_guess.workspace = true
|
||||
reqwest.workspace = true
|
||||
serde_json.workspace = true
|
||||
|
||||
@@ -57,7 +57,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
|
||||
);
|
||||
}
|
||||
|
||||
let router = router(&cfg);
|
||||
let router = router(&cfg)?;
|
||||
let listener = TcpListener::bind(cfg.bind_addr).await?;
|
||||
tracing::info!(addr = %listener.local_addr()?, "serving");
|
||||
axum::serve(listener, router)
|
||||
|
||||
@@ -10,9 +10,11 @@ use std::sync::Arc;
|
||||
use axum::body::Body;
|
||||
use axum::extract::ws::{Message as AxMessage, WebSocket};
|
||||
use axum::extract::{Request, State};
|
||||
use axum::http::{header, HeaderMap, HeaderValue, StatusCode, Uri};
|
||||
use axum::http::{HeaderMap, HeaderValue, StatusCode, Uri};
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use maestro_protocol::routes::{self, UrlError};
|
||||
use maestro_protocol::ApiError;
|
||||
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -25,21 +27,13 @@ pub struct UpstreamConfig {
|
||||
}
|
||||
|
||||
impl UpstreamConfig {
|
||||
pub fn from_daemon_url(daemon_url: &str) -> Self {
|
||||
let http_base = daemon_url.trim_end_matches('/').to_string();
|
||||
let ws_base = if let Some(rest) = http_base.strip_prefix("https://") {
|
||||
format!("wss://{rest}")
|
||||
} else if let Some(rest) = http_base.strip_prefix("http://") {
|
||||
format!("ws://{rest}")
|
||||
} else {
|
||||
// Caller passed something already scheme-less or pre-wrapped; trust
|
||||
// it. Validation happens at config-load time.
|
||||
http_base.clone()
|
||||
};
|
||||
Self {
|
||||
pub fn from_daemon_url(daemon_url: &str) -> Result<Self, UrlError> {
|
||||
let http_base = routes::normalize_base(daemon_url).to_string();
|
||||
let ws_base = routes::ws_base(&http_base)?;
|
||||
Ok(Self {
|
||||
daemon_http_base: Arc::new(http_base),
|
||||
daemon_ws_base: Arc::new(ws_base),
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -170,29 +164,22 @@ fn filter_response_headers(headers: &reqwest::header::HeaderMap) -> HeaderMap {
|
||||
|
||||
fn bad_gateway(msg: &str) -> Response {
|
||||
tracing::warn!(error = msg, "proxy: bad gateway");
|
||||
// Plain JSON without a serde_json dep — the message is short and we control
|
||||
// every byte we write, so manual escaping of `"` + `\` is enough here.
|
||||
let escaped = msg.replace('\\', "\\\\").replace('"', "\\\"");
|
||||
let body = format!(r#"{{"error":{{"code":"proxy_failed","message":"{escaped}"}}}}"#);
|
||||
(
|
||||
StatusCode::BAD_GATEWAY,
|
||||
[(header::CONTENT_TYPE, "application/json")],
|
||||
body,
|
||||
)
|
||||
.into_response()
|
||||
api_error(StatusCode::BAD_GATEWAY, "proxy_failed", msg)
|
||||
}
|
||||
|
||||
fn payload_too_large() -> Response {
|
||||
tracing::warn!(limit = MAX_REQUEST_BODY, "proxy: request body exceeds cap");
|
||||
let body = format!(
|
||||
r#"{{"error":{{"code":"payload_too_large","message":"request body exceeds {MAX_REQUEST_BODY} bytes"}}}}"#
|
||||
);
|
||||
(
|
||||
api_error(
|
||||
StatusCode::PAYLOAD_TOO_LARGE,
|
||||
[(header::CONTENT_TYPE, "application/json")],
|
||||
body,
|
||||
"payload_too_large",
|
||||
&format!("request body exceeds {MAX_REQUEST_BODY} bytes"),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
|
||||
/// Answer in the same envelope the daemon uses, so a client cannot tell whether
|
||||
/// a failure came from the daemon or from this proxy.
|
||||
fn api_error(status: StatusCode, code: &str, message: &str) -> Response {
|
||||
(status, axum::Json(ApiError::new(code, message))).into_response()
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -292,7 +279,7 @@ const CLOSE_UPSTREAM_REJECTED: u16 = 4502;
|
||||
|
||||
/// Turn a failed upstream WS handshake into a browser-facing close code +
|
||||
/// reason. When the daemon rejected the upgrade with its structured JSON error
|
||||
/// body (`{"error":{"code","message"}}` — see `local_api::error_response`),
|
||||
/// body ([`ApiError`]),
|
||||
/// surface that message; otherwise fall back to a generic gateway reason. The
|
||||
/// reason rides a WS control frame, so it's truncated to the 123-byte limit.
|
||||
fn upstream_close_reason(e: &tokio_tungstenite::tungstenite::Error) -> (u16, String) {
|
||||
@@ -301,13 +288,8 @@ fn upstream_close_reason(e: &tokio_tungstenite::tungstenite::Error) -> (u16, Str
|
||||
WsError::Http(resp) => resp
|
||||
.body()
|
||||
.as_ref()
|
||||
.and_then(|b| serde_json::from_slice::<serde_json::Value>(b).ok())
|
||||
.and_then(|v| {
|
||||
v.get("error")
|
||||
.and_then(|err| err.get("message"))
|
||||
.and_then(|m| m.as_str())
|
||||
.map(str::to_owned)
|
||||
})
|
||||
.and_then(|b| serde_json::from_slice::<ApiError>(b).ok())
|
||||
.map(|e| e.error.message)
|
||||
.unwrap_or_else(|| "the session's daemon rejected the attach".to_owned()),
|
||||
_ => "could not reach the session's daemon".to_owned(),
|
||||
};
|
||||
@@ -346,14 +328,14 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn ws_base_swaps_http_scheme() {
|
||||
let u = UpstreamConfig::from_daemon_url("http://127.0.0.1:7070/");
|
||||
let u = UpstreamConfig::from_daemon_url("http://127.0.0.1:7070/").unwrap();
|
||||
assert_eq!(&*u.daemon_http_base, "http://127.0.0.1:7070");
|
||||
assert_eq!(&*u.daemon_ws_base, "ws://127.0.0.1:7070");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ws_base_swaps_https_scheme() {
|
||||
let u = UpstreamConfig::from_daemon_url("https://daemon.example/");
|
||||
let u = UpstreamConfig::from_daemon_url("https://daemon.example/").unwrap();
|
||||
assert_eq!(&*u.daemon_ws_base, "wss://daemon.example");
|
||||
}
|
||||
|
||||
@@ -362,7 +344,7 @@ mod tests {
|
||||
/// a bogus unreachable address; a 413 (not a 502) proves we bailed early.
|
||||
#[tokio::test]
|
||||
async fn oversized_request_body_is_rejected_with_413() {
|
||||
let state = ProxyState::new(UpstreamConfig::from_daemon_url("http://127.0.0.1:9"));
|
||||
let state = ProxyState::new(UpstreamConfig::from_daemon_url("http://127.0.0.1:9").unwrap());
|
||||
let big = vec![0u8; MAX_REQUEST_BODY + 1];
|
||||
let req = Request::builder()
|
||||
.method("POST")
|
||||
@@ -379,7 +361,7 @@ mod tests {
|
||||
/// limit, not off-by-one against legitimate traffic.
|
||||
#[tokio::test]
|
||||
async fn body_at_cap_is_not_rejected_by_the_size_check() {
|
||||
let state = ProxyState::new(UpstreamConfig::from_daemon_url("http://127.0.0.1:9"));
|
||||
let state = ProxyState::new(UpstreamConfig::from_daemon_url("http://127.0.0.1:9").unwrap());
|
||||
let body = vec![0u8; MAX_REQUEST_BODY];
|
||||
let req = Request::builder()
|
||||
.method("POST")
|
||||
|
||||
@@ -5,6 +5,7 @@ use axum::extract::{Request, State, WebSocketUpgrade};
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::routing::{any, get};
|
||||
use axum::Router;
|
||||
use maestro_protocol::routes::{self, UrlError};
|
||||
|
||||
use crate::config::WebConfig;
|
||||
use crate::proxy::{pump_ws, upstream_ws_url, ProxyState, UpstreamConfig};
|
||||
@@ -12,30 +13,37 @@ use crate::static_bundle;
|
||||
|
||||
/// Build the public router. The returned router is ready to hand to
|
||||
/// `axum::serve`.
|
||||
pub fn router(config: &WebConfig) -> Router {
|
||||
let upstream = UpstreamConfig::from_daemon_url(&config.daemon_url);
|
||||
pub fn router(config: &WebConfig) -> Result<Router, UrlError> {
|
||||
let upstream = UpstreamConfig::from_daemon_url(&config.daemon_url)?;
|
||||
let proxy_state = ProxyState::new(upstream);
|
||||
|
||||
// Mount each route at its real path (not under `nest`, which would strip
|
||||
// the `/api` prefix before the proxy could forward it). WS routes go
|
||||
// first so the HTTP fallback doesn't shadow them.
|
||||
Router::new()
|
||||
.route("/api/attach/:id", any(ws_proxy))
|
||||
.route("/api/events", any(ws_proxy))
|
||||
Ok(Router::new()
|
||||
.route(routes::patterns::ATTACH, any(ws_proxy))
|
||||
.route(routes::EVENTS, any(ws_proxy))
|
||||
.route("/", get(serve_index))
|
||||
.fallback(catch_all)
|
||||
.with_state(proxy_state)
|
||||
.with_state(proxy_state))
|
||||
}
|
||||
|
||||
/// Default handler: `/api/*` goes to the daemon, anything else falls through
|
||||
/// to the embedded SPA bundle (with SPA-style fallback to `index.html`).
|
||||
async fn catch_all(State(state): State<ProxyState>, req: Request) -> Response {
|
||||
if req.uri().path().starts_with("/api/") {
|
||||
if is_api_path(req.uri().path()) {
|
||||
return crate::proxy::proxy_http(State(state), req).await;
|
||||
}
|
||||
static_bundle::serve_path(req.uri().path())
|
||||
}
|
||||
|
||||
/// Split point between the proxied daemon surface and the SPA bundle. A daemon
|
||||
/// route added outside [`routes::API_PREFIX`] would silently be answered with
|
||||
/// `index.html` instead of being forwarded.
|
||||
fn is_api_path(path: &str) -> bool {
|
||||
path.starts_with(&format!("{}/", routes::API_PREFIX))
|
||||
}
|
||||
|
||||
async fn ws_proxy(State(state): State<ProxyState>, ws: WebSocketUpgrade, req: Request) -> Response {
|
||||
let upstream_url = upstream_ws_url(&state, req.uri());
|
||||
ws.on_upgrade(move |socket| async move {
|
||||
|
||||
@@ -66,7 +66,7 @@ async fn spawn_web(daemon_url: String) -> (SocketAddr, oneshot::Sender<()>) {
|
||||
daemon_url,
|
||||
dev_proxy_target: None,
|
||||
};
|
||||
let router = maestro_web::router(&cfg);
|
||||
let router = maestro_web::router(&cfg).expect("router");
|
||||
let listener = TcpListener::bind(cfg.bind_addr).await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let (tx, rx) = oneshot::channel::<()>();
|
||||
|
||||
+9
-11
@@ -1,3 +1,4 @@
|
||||
import * as routes from "./routes.js";
|
||||
import type {
|
||||
ApiError,
|
||||
CreateSessionRequest,
|
||||
@@ -25,12 +26,12 @@ async function jsonOrError<T>(res: Response): Promise<T> {
|
||||
}
|
||||
|
||||
export async function listSessions(): Promise<Session[]> {
|
||||
return jsonOrError(await fetch("/api/sessions", { credentials: "same-origin" }));
|
||||
return jsonOrError(await fetch(routes.SESSIONS, { credentials: "same-origin" }));
|
||||
}
|
||||
|
||||
export async function createSession(req: CreateSessionRequest): Promise<Session> {
|
||||
return jsonOrError(
|
||||
await fetch("/api/sessions", {
|
||||
await fetch(routes.SESSIONS, {
|
||||
method: "POST",
|
||||
credentials: "same-origin",
|
||||
headers: { "content-type": "application/json" },
|
||||
@@ -41,7 +42,7 @@ export async function createSession(req: CreateSessionRequest): Promise<Session>
|
||||
|
||||
export async function killSession(id: Uuid): Promise<void> {
|
||||
await jsonOrError(
|
||||
await fetch(`/api/sessions/${encodeURIComponent(id)}`, {
|
||||
await fetch(routes.session(id), {
|
||||
method: "DELETE",
|
||||
credentials: "same-origin",
|
||||
}),
|
||||
@@ -51,30 +52,27 @@ export async function killSession(id: Uuid): Promise<void> {
|
||||
/** List one directory's immediate subdirectories (NAP-329). `path` may be empty,
|
||||
* `~`, relative, or absolute — the daemon resolves it against its home dir. */
|
||||
export async function listDir(path: string): Promise<DirListing> {
|
||||
const url = `/api/dirs?path=${encodeURIComponent(path)}`;
|
||||
const url = routes.withPathQuery(routes.DIRS, path);
|
||||
return jsonOrError(await fetch(url, { credentials: "same-origin" }));
|
||||
}
|
||||
|
||||
/** Identity + spawn presets of the daemon this web app is connected to (#26). */
|
||||
export async function getDaemonInfo(): Promise<DaemonInfo> {
|
||||
return jsonOrError(await fetch("/api/info", { credentials: "same-origin" }));
|
||||
return jsonOrError(await fetch(routes.INFO, { credentials: "same-origin" }));
|
||||
}
|
||||
|
||||
/** Recently-used working dirs, newest-first and deduped by dir (#26). Drives the
|
||||
* spawn form's recents picker. */
|
||||
export async function recentDirs(): Promise<RecentDirs> {
|
||||
return jsonOrError(await fetch("/api/recent-dirs", { credentials: "same-origin" }));
|
||||
return jsonOrError(await fetch(routes.RECENT_DIRS, { credentials: "same-origin" }));
|
||||
}
|
||||
|
||||
/** Probe the enclosing git repo of `path` (#26). `root` is absent when the path
|
||||
* isn't inside a repo. `path` may be `~`/relative/absolute; the daemon resolves
|
||||
* it against its home dir. */
|
||||
export async function gitRoot(path: string): Promise<GitRoot> {
|
||||
const url = `/api/gitroot?path=${encodeURIComponent(path)}`;
|
||||
const url = routes.withPathQuery(routes.GITROOT, path);
|
||||
return jsonOrError(await fetch(url, { credentials: "same-origin" }));
|
||||
}
|
||||
|
||||
export function eventsWsUrl(): string {
|
||||
const proto = location.protocol === "https:" ? "wss:" : "ws:";
|
||||
return `${proto}//${location.host}/api/events`;
|
||||
}
|
||||
export { eventsWsUrl } from "./routes.js";
|
||||
|
||||
+2
-11
@@ -2,6 +2,7 @@ import { FitAddon } from "@xterm/addon-fit";
|
||||
import { Terminal } from "@xterm/xterm";
|
||||
import "@xterm/xterm/css/xterm.css";
|
||||
|
||||
import { attachWsUrl } from "./routes.js";
|
||||
import type { ClientControl, Uuid } from "./types.js";
|
||||
|
||||
/**
|
||||
@@ -51,16 +52,6 @@ function esc(s: string): string {
|
||||
.replace(/"/g, """);
|
||||
}
|
||||
|
||||
/**
|
||||
* Same-origin attach URL. The browser loads from `maestro-web`, which
|
||||
* reverse-proxies this WebSocket to the co-located daemon's `/api/attach/:id`
|
||||
* endpoint.
|
||||
*/
|
||||
function attachStreamUrl(sessionId: Uuid): string {
|
||||
const proto = location.protocol === "https:" ? "wss:" : "ws:";
|
||||
return `${proto}//${location.host}/api/attach/${encodeURIComponent(sessionId)}`;
|
||||
}
|
||||
|
||||
export function createAttachConnection(
|
||||
sessionId: Uuid,
|
||||
opts: AttachOptions = {},
|
||||
@@ -264,7 +255,7 @@ export function createAttachConnection(
|
||||
}
|
||||
};
|
||||
|
||||
const ws = new WebSocket(attachStreamUrl(sessionId));
|
||||
const ws = new WebSocket(attachWsUrl(sessionId));
|
||||
ws.binaryType = "arraybuffer";
|
||||
|
||||
const sendControl = (msg: ClientControl) => {
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
// Mirror of `maestro-protocol::routes`. Keep in sync with
|
||||
// crates/maestro-protocol/src/routes.rs.
|
||||
|
||||
import type { Uuid } from "./types.js";
|
||||
|
||||
export const API_PREFIX = "/api";
|
||||
|
||||
export const SESSIONS = "/api/sessions";
|
||||
export const INFO = "/api/info";
|
||||
export const DIRS = "/api/dirs";
|
||||
export const RECENT_DIRS = "/api/recent-dirs";
|
||||
export const GITROOT = "/api/gitroot";
|
||||
export const EVENTS = "/api/events";
|
||||
|
||||
/** Name of the query parameter `DIRS` and `GITROOT` both take. */
|
||||
export const PATH_QUERY = "path";
|
||||
|
||||
export function session(id: Uuid): string {
|
||||
return `${SESSIONS}/${encodeURIComponent(id)}`;
|
||||
}
|
||||
|
||||
export function attach(id: Uuid): string {
|
||||
return `${API_PREFIX}/attach/${encodeURIComponent(id)}`;
|
||||
}
|
||||
|
||||
/** Append the `path` query parameter `DIRS` and `GITROOT` accept. */
|
||||
export function withPathQuery(route: string, value: string): string {
|
||||
return `${route}?${PATH_QUERY}=${encodeURIComponent(value)}`;
|
||||
}
|
||||
|
||||
/** Same-origin WebSocket base — `wss:` under TLS, `ws:` otherwise. */
|
||||
export function wsBase(): string {
|
||||
const proto = location.protocol === "https:" ? "wss:" : "ws:";
|
||||
return `${proto}//${location.host}`;
|
||||
}
|
||||
|
||||
export function eventsWsUrl(): string {
|
||||
return `${wsBase()}${EVENTS}`;
|
||||
}
|
||||
|
||||
export function attachWsUrl(id: Uuid): string {
|
||||
return `${wsBase()}${attach(id)}`;
|
||||
}
|
||||
Reference in New Issue
Block a user