heliosdb_proxy/
graphql_gateway.rs

1//! GraphQL-to-SQL gateway — HTTP listener.
2//!
3//! When `[graphql_gateway] enabled = true`, the proxy exposes an HTTP endpoint
4//! that accepts a GraphQL query (`POST` with `{"query": "..."}`), generates SQL
5//! from the configured schema, executes it over the backend PG-wire client, and
6//! returns a GraphQL JSON response (`{"data": {...}}`). Flat top-level
7//! selections are supported; nested-relationship shaping is a follow-on.
8
9use std::sync::Arc;
10use std::time::Duration;
11
12use serde_json::{json, Value};
13use tokio::io::{AsyncWriteExt, BufReader};
14use tokio::net::TcpListener;
15
16use crate::backend::{tls::default_client_config, BackendConfig, TlsMode};
17use crate::config::GraphqlGatewayConfig;
18use crate::graphql::introspector::{ColumnDefinition, TableDefinition};
19use crate::graphql::{GraphQLConfig, GraphQLEngine, GraphQLRequest, SchemaIntrospector};
20use crate::{ProxyError, Result};
21
22pub struct GraphqlGateway {
23    config: Arc<GraphqlGatewayConfig>,
24    engine: Arc<GraphQLEngine>,
25}
26
27impl GraphqlGateway {
28    pub fn new(config: GraphqlGatewayConfig) -> Self {
29        // Build the GraphQL schema from the configured tables.
30        let tabledefs: Vec<TableDefinition> = config
31            .tables
32            .iter()
33            .map(|t| TableDefinition {
34                name: t.name.clone(),
35                schema: "public".to_string(),
36                columns: t
37                    .columns
38                    .iter()
39                    .map(|c| ColumnDefinition {
40                        name: c.clone(),
41                        data_type: "text".to_string(),
42                        nullable: true,
43                        is_primary_key: c == "id",
44                        has_default: false,
45                    })
46                    .collect(),
47                foreign_keys: Vec::new(),
48            })
49            .collect();
50        let schema = SchemaIntrospector::new().build_schema(&tabledefs);
51
52        let bcfg = BackendConfig {
53            host: config.backend_host.clone(),
54            port: config.backend_port,
55            user: config.backend_user.clone(),
56            password: config.backend_password.clone(),
57            database: config.backend_database.clone(),
58            application_name: Some("heliosproxy-graphql".to_string()),
59            tls_mode: TlsMode::Disable,
60            connect_timeout: Duration::from_secs(5),
61            query_timeout: Duration::from_secs(30),
62            tls_config: default_client_config(),
63        };
64        let engine = GraphQLEngine::new(GraphQLConfig::default(), schema).with_backend(bcfg);
65
66        Self {
67            config: Arc::new(config),
68            engine: Arc::new(engine),
69        }
70    }
71
72    pub async fn run(self) -> Result<()> {
73        let listener = TcpListener::bind(&self.config.listen_address)
74            .await
75            .map_err(|e| {
76                ProxyError::Network(format!(
77                    "GraphQL gateway bind {}: {}",
78                    self.config.listen_address, e
79                ))
80            })?;
81        tracing::info!(addr = %self.config.listen_address, "GraphQL gateway listening");
82        let config = self.config.clone();
83        let engine = self.engine.clone();
84        loop {
85            let (stream, peer) = match listener.accept().await {
86                Ok(x) => x,
87                Err(e) => {
88                    tracing::warn!("GraphQL gateway accept error: {}", e);
89                    continue;
90                }
91            };
92            let config = config.clone();
93            let engine = engine.clone();
94            tokio::spawn(async move {
95                if let Err(e) = Self::handle(stream, config, engine).await {
96                    tracing::debug!(%peer, "GraphQL gateway error: {}", e);
97                }
98            });
99        }
100    }
101
102    async fn handle(
103        mut stream: tokio::net::TcpStream,
104        cfg: Arc<GraphqlGatewayConfig>,
105        engine: Arc<GraphQLEngine>,
106    ) -> Result<()> {
107        use crate::http_util;
108        let (reader, mut writer) = stream.split();
109        let mut reader = BufReader::new(reader);
110
111        // Bounded request read: overall deadline + header count/byte caps.
112        let deadline = tokio::time::Instant::now() + http_util::HTTP_READ_TIMEOUT;
113        let head = match http_util::read_head(&mut reader, deadline).await {
114            Ok(h) => h,
115            Err(_) => return Ok(()), // timeout / oversized headers / early close
116        };
117        let method = head.method.as_str();
118        let path = head.path.as_str();
119
120        if method == "GET" && (path == "/health" || path == "/") {
121            return Self::respond(&mut writer, 200, &json!({"status":"ok"})).await;
122        }
123        // Constant-time Bearer check.
124        let authorized = match cfg.auth_token.as_ref() {
125            None => true,
126            Some(tok) => head
127                .header("authorization")
128                .and_then(|v| v.strip_prefix("Bearer "))
129                .map(|got| http_util::constant_time_eq_str(got, tok))
130                .unwrap_or(false),
131        };
132        if !authorized {
133            return Self::respond(&mut writer, 401, &json!({"error":"unauthorized"})).await;
134        }
135        if method != "POST" {
136            return Self::respond(
137                &mut writer,
138                405,
139                &json!({"error":"use POST with a GraphQL query"}),
140            )
141            .await;
142        }
143        // Reject an oversized declared body BEFORE allocating for it.
144        if head.content_length > http_util::MAX_HTTP_BODY_BYTES {
145            return Self::respond(&mut writer, 413, &json!({"error":"request body too large"}))
146                .await;
147        }
148
149        let body_buf = match http_util::read_body(&mut reader, head.content_length, deadline).await
150        {
151            Ok(b) => b,
152            Err(_) => return Ok(()),
153        };
154        let req: Value = match serde_json::from_slice(&body_buf) {
155            Ok(v) => v,
156            Err(e) => {
157                return Self::respond(
158                    &mut writer,
159                    400,
160                    &json!({"errors":[{"message": format!("invalid JSON: {}", e)}]}),
161                )
162                .await
163            }
164        };
165        let query = req
166            .get("query")
167            .and_then(|q| q.as_str())
168            .unwrap_or("")
169            .trim();
170        if query.is_empty() {
171            return Self::respond(
172                &mut writer,
173                400,
174                &json!({"errors":[{"message":"missing 'query'"}]}),
175            )
176            .await;
177        }
178
179        let response = engine.execute(GraphQLRequest::new(query)).await;
180        let errors = response.errors.map(|errs| {
181            errs.iter()
182                .map(|e| json!({ "message": e.to_string() }))
183                .collect::<Vec<_>>()
184        });
185        let body = json!({ "data": response.data, "errors": errors });
186        Self::respond(&mut writer, 200, &body).await
187    }
188
189    async fn respond<W: AsyncWriteExt + Unpin>(
190        writer: &mut W,
191        status: u16,
192        body: &Value,
193    ) -> Result<()> {
194        let payload = serde_json::to_vec(body).unwrap_or_else(|_| b"{}".to_vec());
195        let reason = match status {
196            200 => "OK",
197            400 => "Bad Request",
198            401 => "Unauthorized",
199            405 => "Method Not Allowed",
200            _ => "OK",
201        };
202        let head = format!(
203            "HTTP/1.1 {} {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
204            status,
205            reason,
206            payload.len()
207        );
208        writer
209            .write_all(head.as_bytes())
210            .await
211            .map_err(|e| ProxyError::Network(format!("GraphQL gw write: {}", e)))?;
212        writer
213            .write_all(&payload)
214            .await
215            .map_err(|e| ProxyError::Network(format!("GraphQL gw write: {}", e)))?;
216        let _ = writer.flush().await;
217        Ok(())
218    }
219}
heliosdb_proxy/graphql_gateway.rs

heliosdb_proxy/
graphql_gateway.rs
