maestro/crates/maestro-registry/src/client_api.rs

//! Client-facing HTTP API routes (NAP-215).
//!
//! Mounts under the same router as the daemon-facing `WS /daemon`. The cookie
//! middleware in [`crate::auth`] gates everything except the WS upgrade for
//! `/daemon` (daemons authenticate via the shared token in `Hello`).

use std::sync::Arc;
use std::time::Duration;

use axum::extract::{ws::WebSocketUpgrade, Path, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::{any, get};
use axum::{Json, Router};
use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, Session};
use uuid::Uuid;

use crate::events::events_ws;
use crate::jwt::issue_attach_jwt;
use crate::state::{Registry, RegistryError, RPC_TIMEOUT};
use crate::web_assets;

/// Shared state plumbed into client-facing handlers.
#[derive(Debug, Clone)]
pub struct ClientApiState {
    pub registry: Registry,
    pub token: Arc<String>,
    pub rpc_timeout: Duration,
}

/// Mount client-facing routes on the given router. Caller is responsible for
/// applying the cookie middleware (we do that in [`crate::server::router`]).
pub fn routes(state: ClientApiState) -> Router {
    Router::new()
        .route("/", get(index))
        .route("/assets/*file", get(asset))
        .route("/api/daemons", get(list_daemons))
        .route("/api/sessions", get(list_sessions).post(create_session))
        .route("/api/sessions/:id", axum::routing::delete(delete_session))
        .route("/api/sessions/:id/connect-info", get(session_connect_info))
        .route("/api/events", any(events_ws_handler))
        .with_state(state)
}

async fn index() -> Response {
    web_assets::serve_index()
}

async fn asset(Path(file): Path<String>) -> Response {
    web_assets::serve_asset(&file)
}

async fn list_daemons(State(s): State<ClientApiState>) -> Json<Vec<DaemonInfo>> {
    let infos: Vec<DaemonInfo> = s
        .registry
        .daemons()
        .await
        .into_iter()
        .map(Into::into)
        .collect();
    Json(infos)
}

async fn list_sessions(State(s): State<ClientApiState>) -> Json<Vec<Session>> {
    Json(s.registry.sessions().await)
}

async fn create_session(
    State(s): State<ClientApiState>,
    Json(body): Json<CreateSessionRequest>,
) -> Response {
    let daemon = body.daemon_id.clone();
    let outcome =
        tokio::time::timeout(s.rpc_timeout, s.registry.create_session(&daemon, body)).await;
    match outcome {
        Err(_) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(error_body("timeout", "daemon did not reply in time")),
        )
            .into_response(),
        Ok(Err(RegistryError::DaemonNotConnected(_))) => (
            StatusCode::NOT_FOUND,
            Json(error_body("unknown_daemon", "daemon is not connected")),
        )
            .into_response(),
        Ok(Err(RegistryError::Timeout(_))) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(error_body("timeout", "daemon did not reply in time")),
        )
            .into_response(),
        Ok(Err(e)) => (
            StatusCode::BAD_GATEWAY,
            Json(error_body("transport", &e.to_string())),
        )
            .into_response(),
        Ok(Ok(Err(rpc_err))) => (
            StatusCode::BAD_GATEWAY,
            Json(error_body(&rpc_err.0.code, &rpc_err.0.message)),
        )
            .into_response(),
        Ok(Ok(Ok(session))) => (StatusCode::CREATED, Json(session)).into_response(),
    }
}

async fn delete_session(State(s): State<ClientApiState>, Path(id): Path<Uuid>) -> Response {
    let Some(daemon) = s.registry.daemon_for_session(id).await else {
        return (
            StatusCode::NOT_FOUND,
            Json(error_body("unknown_session", "no such session")),
        )
            .into_response();
    };

    let outcome = tokio::time::timeout(s.rpc_timeout, s.registry.kill_session(&daemon, id)).await;
    match outcome {
        Err(_) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(error_body("timeout", "daemon did not reply in time")),
        )
            .into_response(),
        Ok(Err(RegistryError::DaemonNotConnected(_))) => (
            StatusCode::NOT_FOUND,
            Json(error_body("unknown_daemon", "daemon is not connected")),
        )
            .into_response(),
        Ok(Err(RegistryError::Timeout(_))) => (
            StatusCode::GATEWAY_TIMEOUT,
            Json(error_body("timeout", "daemon did not reply in time")),
        )
            .into_response(),
        Ok(Err(e)) => (
            StatusCode::BAD_GATEWAY,
            Json(error_body("transport", &e.to_string())),
        )
            .into_response(),
        Ok(Ok(Err(rpc_err))) => (
            StatusCode::BAD_GATEWAY,
            Json(error_body(&rpc_err.0.code, &rpc_err.0.message)),
        )
            .into_response(),
        Ok(Ok(Ok(()))) => StatusCode::NO_CONTENT.into_response(),
    }
}

async fn session_connect_info(State(s): State<ClientApiState>, Path(id): Path<Uuid>) -> Response {
    let Some(daemon) = s.registry.daemon_for_session(id).await else {
        return (
            StatusCode::NOT_FOUND,
            Json(error_body("unknown_session", "no such session")),
        )
            .into_response();
    };
    let Some(daemon_url) = s.registry.daemon_url(&daemon).await else {
        return (
            StatusCode::NOT_FOUND,
            Json(error_body("unknown_daemon", "owning daemon is offline")),
        )
            .into_response();
    };
    let jwt = match issue_attach_jwt(s.token.as_str(), &daemon, id) {
        Ok(t) => t,
        Err(e) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(error_body("jwt_failed", &e.to_string())),
            )
                .into_response();
        }
    };
    Json(ConnectInfo { daemon_url, jwt }).into_response()
}

async fn events_ws_handler(
    State(s): State<ClientApiState>,
    ws: WebSocketUpgrade,
) -> impl IntoResponse {
    let registry = s.registry.clone();
    ws.on_upgrade(move |socket| async move {
        let _ = events_ws(socket, registry).await;
    })
}

fn error_body(code: &str, message: &str) -> serde_json::Value {
    serde_json::json!({
        "error": {
            "code": code,
            "message": message,
        },
    })
}

/// Convenience: produce a `ClientApiState` from a `Registry` and shared token.
/// Uses the production [`RPC_TIMEOUT`].
pub fn state_for(registry: Registry, token: String) -> ClientApiState {
    ClientApiState {
        registry,
        token: Arc::new(token),
        rpc_timeout: RPC_TIMEOUT,
    }
}