pulpod 0.0.42

Pulpo daemon — manages agent sessions via tmux/Docker
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

use std::sync::Arc;

use axum::{Json, http::StatusCode};
use pulpo_common::api::ErrorResponse;
use pulpo_common::peer::PeerInfo;
use serde::de::DeserializeOwned;

use crate::remote::{
    RemoteNodeTarget, apply_remote_auth, normalize_http_base, resolve_peer_target,
};

use super::AppState;

pub(super) type ApiError = (StatusCode, Json<ErrorResponse>);

#[derive(Debug, Clone)]
pub(super) struct RemoteSessionNodeTarget {
    pub session_id: String,
    pub node_name: String,
    pub base_url: String,
    pub token: Option<String>,
}

pub(super) fn internal_error(msg: &str) -> ApiError {
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(ErrorResponse {
            error: msg.to_owned(),
        }),
    )
}

pub(super) fn bad_gateway(msg: &str) -> ApiError {
    (
        StatusCode::BAD_GATEWAY,
        Json(ErrorResponse {
            error: msg.to_owned(),
        }),
    )
}

fn reqwest_status_to_axum(status: reqwest::StatusCode) -> StatusCode {
    StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY)
}

pub(super) async fn reqwest_error_response(resp: reqwest::Response, fallback: &str) -> ApiError {
    let status = reqwest_status_to_axum(resp.status());
    let error = match resp.json::<ErrorResponse>().await {
        Ok(body) => body.error,
        Err(_) => fallback.to_owned(),
    };
    (status, Json(ErrorResponse { error }))
}

pub(super) async fn send_remote_request(
    request: reqwest::RequestBuilder,
    failure: String,
) -> Result<reqwest::Response, ApiError> {
    request
        .send()
        .await
        .map_err(|e| bad_gateway(&format!("{failure}: {e}")))
}

pub(super) fn remote_json_request(
    target: &RemoteSessionNodeTarget,
    request: reqwest::RequestBuilder,
) -> reqwest::RequestBuilder {
    apply_remote_auth(request, target.token.as_deref())
}

pub(super) async fn parse_remote_json<T: DeserializeOwned>(
    resp: reqwest::Response,
    fallback: &str,
    parse_error: &str,
) -> Result<T, ApiError> {
    if !resp.status().is_success() {
        return Err(reqwest_error_response(resp, fallback).await);
    }

    resp.json::<T>()
        .await
        .map_err(|e| internal_error(&format!("{parse_error}: {e}")))
}

pub(super) async fn expect_remote_no_content(
    resp: reqwest::Response,
    fallback: &str,
) -> Result<StatusCode, ApiError> {
    if !resp.status().is_success() {
        return Err(reqwest_error_response(resp, fallback).await);
    }
    Ok(StatusCode::NO_CONTENT)
}

pub(super) async fn resolve_remote_session_node_target(
    state: &Arc<AppState>,
    id: &str,
) -> Result<Option<RemoteSessionNodeTarget>, ApiError> {
    let Some(session_index) = &state.session_index else {
        return Ok(None);
    };
    let Some(entry) = session_index.get(id).await else {
        return Ok(None);
    };

    let peer: Option<PeerInfo> = state.peer_registry.get(&entry.node_name).await;
    let address = peer
        .as_ref()
        .map(|p| p.address.clone())
        .or_else(|| entry.node_address.clone());

    let Some(address) = address else {
        return Err(bad_gateway(&format!(
            "node address unknown for remote session {id} on node {}",
            entry.node_name
        )));
    };

    let token = state.peer_registry.get_token(&entry.node_name).await;
    Ok(Some(RemoteSessionNodeTarget {
        session_id: entry.session_id,
        node_name: entry.node_name,
        base_url: normalize_http_base(&address),
        token,
    }))
}

pub(super) async fn resolve_remote_node_target(
    state: &Arc<AppState>,
    target_node: &str,
) -> Result<Option<RemoteNodeTarget>, ApiError> {
    let local_name = state.config.read().await.node.name.clone();
    if target_node == local_name {
        return Ok(None);
    }

    let Some(target) = resolve_peer_target(&state.peer_registry, target_node).await else {
        return Err((
            StatusCode::NOT_FOUND,
            Json(ErrorResponse {
                error: format!("target node not found: {target_node}"),
            }),
        ));
    };
    Ok(Some(target))
}