tawasuyu/brahman

1//! `shuma-gateway` — HTTP/JSON adapter para el daemon.
2//!
3//! Acepta `POST /rpc` con body JSON serializado como `shuma_protocol::Request`,
4//! hace round-trip al admin socket via postcard, devuelve `Response` como JSON.
5//!
6//! Diseñado para clients no-Rust (curl, Python, web app) que no pueden
7//! hablar postcard directo. NO es un proxy completo — sólo translation
8//! layer del protocolo.
9//!
10//! Sin dep de axum/hyper: HTTP parser ad-hoc, suficiente para 1 endpoint.
11
12use shuma_protocol::{default_socket_path, read_frame, write_frame, Request, Response};
13use std::sync::Arc;
14use tokio::io::{AsyncReadExt, AsyncWriteExt};
15use tokio::net::{TcpListener, TcpStream, UnixStream};
16use tracing::{info, warn};
17
18const DEFAULT_LISTEN: &str = "127.0.0.1:7378";
19
20#[tokio::main]
21async fn main() -> anyhow::Result<()> {
22    init_tracing();
23    let listen = std::env::var("SHIPOTE_GATEWAY_LISTEN").unwrap_or_else(|_| DEFAULT_LISTEN.into());
24    let daemon_sock = Arc::new(default_socket_path());
25    let listener = TcpListener::bind(&listen).await?;
26    info!(listen = %listen, daemon = %daemon_sock.display(), "shuma-gateway listening");
27
28    loop {
29        match listener.accept().await {
30            Ok((stream, peer)) => {
31                let sock = daemon_sock.clone();
32                tokio::spawn(async move {
33                    if let Err(e) = handle_http(stream, sock).await {
34                        warn!(?e, ?peer, "request error");
35                    }
36                });
37            }
38            Err(e) => {
39                warn!(?e, "accept failed");
40                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
41            }
42        }
43    }
44}
45
46async fn handle_http(mut stream: TcpStream, daemon_sock: Arc<std::path::PathBuf>) -> anyhow::Result<()> {
47    // Parser HTTP mínimo: read hasta `\r\n\r\n`, parsear request line +
48    // Content-Length, después leer body exacto.
49    let mut buf = Vec::with_capacity(4096);
50    let mut tmp = [0u8; 4096];
51    let header_end;
52    loop {
53        let n = stream.read(&mut tmp).await?;
54        if n == 0 {
55            return Ok(()); // closed
56        }
57        buf.extend_from_slice(&tmp[..n]);
58        if let Some(pos) = find_double_crlf(&buf) {
59            header_end = pos + 4;
60            break;
61        }
62        if buf.len() > 64 * 1024 {
63            return write_error(&mut stream, 413, "headers too large").await;
64        }
65    }
66
67    let header_str = std::str::from_utf8(&buf[..header_end - 4])?;
68    let mut lines = header_str.lines();
69    let request_line = lines.next().unwrap_or("");
70    let mut parts = request_line.split_whitespace();
71    let method = parts.next().unwrap_or("");
72    let path = parts.next().unwrap_or("");
73    let mut content_length: usize = 0;
74    for line in lines {
75        if let Some(v) = line.strip_prefix("Content-Length:").or_else(|| line.strip_prefix("content-length:")) {
76            content_length = v.trim().parse().unwrap_or(0);
77        }
78    }
79
80    // Rutas:
81    if method == "GET" && (path == "/" || path == "/health") {
82        return write_text(&mut stream, 200, "shuma-gateway ok\n").await;
83    }
84    if method != "POST" || path != "/rpc" {
85        return write_error(&mut stream, 404, "use POST /rpc").await;
86    }
87
88    // Leer body.
89    let mut body = buf[header_end..].to_vec();
90    while body.len() < content_length {
91        let n = stream.read(&mut tmp).await?;
92        if n == 0 {
93            break;
94        }
95        body.extend_from_slice(&tmp[..n]);
96    }
97    body.truncate(content_length);
98
99    // Parsear JSON → Request.
100    let req: Request = match serde_json::from_slice(&body) {
101        Ok(r) => r,
102        Err(e) => return write_error(&mut stream, 400, &format!("bad json: {e}")).await,
103    };
104
105    // Round-trip al daemon.
106    let resp = match round_trip_daemon(&daemon_sock, &req).await {
107        Ok(r) => r,
108        Err(e) => return write_error(&mut stream, 502, &format!("daemon: {e}")).await,
109    };
110
111    // Serializar Response → JSON.
112    let body_json = serde_json::to_vec(&resp)?;
113    write_response(&mut stream, 200, "application/json", &body_json).await
114}
115
116async fn round_trip_daemon(sock: &std::path::Path, req: &Request) -> anyhow::Result<Response> {
117    let mut stream = UnixStream::connect(sock).await?;
118    write_frame(&mut stream, req).await?;
119    let resp: Response = read_frame(&mut stream).await?;
120    Ok(resp)
121}
122
123fn find_double_crlf(buf: &[u8]) -> Option<usize> {
124    buf.windows(4).position(|w| w == b"\r\n\r\n")
125}
126
127async fn write_response(
128    stream: &mut TcpStream,
129    code: u16,
130    content_type: &str,
131    body: &[u8],
132) -> anyhow::Result<()> {
133    let status = match code {
134        200 => "OK",
135        400 => "Bad Request",
136        404 => "Not Found",
137        413 => "Payload Too Large",
138        502 => "Bad Gateway",
139        _ => "Unknown",
140    };
141    let head = format!(
142        "HTTP/1.1 {code} {status}\r\n\
143         Content-Type: {content_type}\r\n\
144         Content-Length: {}\r\n\
145         Connection: close\r\n\
146         \r\n",
147        body.len()
148    );
149    stream.write_all(head.as_bytes()).await?;
150    stream.write_all(body).await?;
151    stream.flush().await?;
152    Ok(())
153}
154
155async fn write_text(stream: &mut TcpStream, code: u16, body: &str) -> anyhow::Result<()> {
156    write_response(stream, code, "text/plain", body.as_bytes()).await
157}
158
159async fn write_error(stream: &mut TcpStream, code: u16, msg: &str) -> anyhow::Result<()> {
160    let body = serde_json::json!({ "error": msg }).to_string();
161    write_response(stream, code, "application/json", body.as_bytes()).await
162}
163
164fn init_tracing() {
165    use tracing_subscriber::{fmt, EnvFilter};
166    let filter = EnvFilter::try_from_env("SHIPOTE_GATEWAY_LOG").unwrap_or_else(|_| EnvFilter::new("info"));
167    fmt().with_env_filter(filter).init();
168}