netget/server/graphql/
mod.rs

1//! GraphQL over HTTP (GraphQL-over-HTTP draft) on hyper. Rust owns request parsing, media-type
2//! negotiation, validation against the schema, introspection and execution; the handler owns
3//! the data behind every query and mutation.
4pub mod actions;
5pub mod engine;
6pub mod ws;
7
8use crate::llm::action_helper::call_llm;
9use crate::llm::actions::protocol_trait::ActionResult;
10use crate::logging::emit::Log;
11use crate::protocol::{Event, SpawnContext};
12use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
13use crate::server::connection::ConnectionId;
14use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
15use apollo_compiler::{executable::OperationType, validation::Valid, Schema};
16use bytes::Bytes;
17use engine::RequestErrors;
18use http_body_util::{BodyExt, Full, Limited};
19use hyper::{
20    body::Incoming, header, server::conn::http1, service::service_fn, Method, Request, Response,
21    StatusCode,
22};
23use hyper_util::rt::{TokioIo, TokioTimer};
24use serde_json::{json, Map, Value};
25use std::{convert::Infallible, net::SocketAddr, sync::Arc, time::Duration};
26
27pub const DEFAULT_SCHEMA: &str = "type Query { hello(name: String): String! }";
28pub const DEFAULT_ENDPOINT: &str = "/graphql";
29pub const DEFAULT_INTROSPECTION: bool = true;
30const HEADER_TIMEOUT: Duration = Duration::from_secs(30);
31const BODY_TIMEOUT: Duration = Duration::from_secs(30);
32pub const GRAPHQL_RESPONSE: &str = "application/graphql-response+json";
33
34type UpgradeSlot = Arc<std::sync::Mutex<Option<hyper::upgrade::OnUpgrade>>>;
35
36struct Shared {
37    ctx: SpawnContext,
38    schema: Valid<Schema>,
39    endpoint: String,
40    introspection: bool,
41    init_timeout: Duration,
42}
43
44pub async fn spawn(ctx: SpawnContext) -> anyhow::Result<SocketAddr> {
45    let p = ctx.startup_params.as_ref();
46    let sdl = p
47        .map(|p| p.get_optional_string("schema"))
48        .transpose()?
49        .flatten()
50        .unwrap_or_else(|| DEFAULT_SCHEMA.to_owned());
51    let schema = engine::load_schema(&sdl)?;
52    let endpoint = p
53        .map(|p| p.get_optional_string("endpoint"))
54        .transpose()?
55        .flatten()
56        .unwrap_or_else(|| DEFAULT_ENDPOINT.to_owned());
57    anyhow::ensure!(
58        endpoint.starts_with('/')
59            && endpoint.len() <= 256
60            && !endpoint.contains(['?', '#', ' '])
61            && !crate::utils::sanitize::has_controls(&endpoint),
62        "endpoint must be an absolute path without query or fragment"
63    );
64    let introspection = p
65        .map(|p| p.get_optional_bool("introspection"))
66        .transpose()?
67        .flatten()
68        .unwrap_or(DEFAULT_INTROSPECTION);
69    let init_timeout = p
70        .map(|p| p.get_optional_u64("connection_init_timeout_secs"))
71        .transpose()?
72        .flatten()
73        .unwrap_or(ws::CONNECTION_INIT_TIMEOUT.as_secs());
74    anyhow::ensure!(
75        (1..=300).contains(&init_timeout),
76        "connection_init_timeout_secs must be 1..=300"
77    );
78    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
79    let local = listener.local_addr()?;
80    Log::new(Some(&ctx.status_tx)).info(format!(
81        "GraphQL server at http://{local}{endpoint} (introspection {})",
82        if introspection { "on" } else { "off" }
83    ));
84    let shared = Arc::new(Shared {
85        ctx: ctx.clone(),
86        schema,
87        endpoint,
88        introspection,
89        init_timeout: Duration::from_secs(init_timeout),
90    });
91    let server_id = ctx.server_id;
92    let accept = tokio::spawn(async move {
93        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
94        loop {
95            let (stream, peer, permit) = match accept_bounded(&listener, &limiter, b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", "GraphQL", Some(&shared.ctx.status_tx)).await {
96                Ok(v) => v,
97                Err(_) => break,
98            };
99            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
100            let now = crate::utils::clock::Instant::now();
101            shared
102                .ctx
103                .state
104                .add_connection_to_server(
105                    server_id,
106                    ConnectionState {
107                        id,
108                        remote_addr: peer,
109                        local_addr: local,
110                        bytes_sent: 0,
111                        bytes_received: 0,
112                        packets_sent: 0,
113                        packets_received: 0,
114                        last_activity: now,
115                        status: ConnectionStatus::Active,
116                        status_changed_at: now,
117                        protocol_info: ProtocolConnectionInfo::empty(),
118                    },
119                )
120                .await;
121            let child = shared.clone();
122            shared
123                .ctx
124                .state
125                .spawn_server_task(server_id, async move {
126                    let _permit = permit;
127                    let svc_shared = child.clone();
128                    let upgrade: UpgradeSlot = Arc::default();
129                    let svc_upgrade = upgrade.clone();
130                    let service = service_fn(move |req| {
131                        let shared = svc_shared.clone();
132                        let upgrade = svc_upgrade.clone();
133                        async move { Ok::<_, Infallible>(handle(&shared, id, req, &upgrade).await) }
134                    });
135                    let mut builder = http1::Builder::new();
136                    builder
137                        .timer(TokioTimer::new())
138                        .header_read_timeout(HEADER_TIMEOUT)
139                        .max_headers(64)
140                        .max_buf_size(32 * 1024);
141                    if let Err(e) = builder
142                        .serve_connection(TokioIo::new(stream), service)
143                        .with_upgrades()
144                        .await
145                    {
146                        Log::new(Some(&child.ctx.status_tx))
147                            .debug(format!("GraphQL connection {id}: {e}"));
148                    }
149                    // An accepted WebSocket upgrade continues in this task, so it keeps the
150                    // connection's permit and its place in the server's task registry.
151                    let pending = upgrade.lock().ok().and_then(|mut u| u.take());
152                    if let Some(on_upgrade) = pending {
153                        if let Ok(upgraded) = on_upgrade.await {
154                            let socket = tokio_tungstenite::WebSocketStream::from_raw_socket(
155                                TokioIo::new(upgraded),
156                                tokio_tungstenite::tungstenite::protocol::Role::Server,
157                                Some(ws::ws_config()),
158                            )
159                            .await;
160                            ws::session(child.clone(), id, socket).await;
161                        }
162                    }
163                    child
164                        .ctx
165                        .state
166                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
167                        .await;
168                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
169                })
170                .await;
171        }
172    });
173    ctx.state.register_server_task(server_id, accept).await;
174    Ok(local)
175}
176
177#[derive(Clone, Copy, PartialEq, Debug)]
178pub enum Media {
179    GraphqlResponse,
180    Json,
181}
182
183impl Media {
184    fn content_type(self) -> &'static str {
185        match self {
186            Media::GraphqlResponse => "application/graphql-response+json; charset=utf-8",
187            Media::Json => "application/json; charset=utf-8",
188        }
189    }
190}
191
192/// Pick the response media type from `Accept`: no header means `application/json` (the spec's
193/// legacy rule); otherwise the acceptable type with the highest q, an exact type beating a
194/// wildcard, and `None` (406) when neither type is acceptable.
195pub fn negotiate(accept: Option<&str>) -> Option<Media> {
196    let Some(accept) = accept.filter(|a| !a.trim().is_empty()) else {
197        return Some(Media::Json);
198    };
199    let mut best: Option<(f32, u8, Media)> = None;
200    for item in accept.split(',').take(32) {
201        let mut parts = item.split(';');
202        let ty = parts.next().unwrap_or("").trim().to_ascii_lowercase();
203        let q = parts
204            .filter_map(|p| p.trim().strip_prefix("q="))
205            .next()
206            .and_then(|q| q.trim().parse::<f32>().ok())
207            .unwrap_or(1.0);
208        if q <= 0.0 {
209            continue;
210        }
211        let (media, specificity) = match ty.as_str() {
212            GRAPHQL_RESPONSE => (Media::GraphqlResponse, 2),
213            "application/json" => (Media::Json, 2),
214            "application/*" | "*/*" => (Media::GraphqlResponse, 1),
215            _ => continue,
216        };
217        let better = match best {
218            None => true,
219            Some((bq, bs, bm)) => {
220                q > bq
221                    || (q == bq && specificity > bs)
222                    || (q == bq
223                        && specificity == bs
224                        && media == Media::GraphqlResponse
225                        && bm == Media::Json)
226            }
227        };
228        if better {
229            best = Some((q, specificity, media));
230        }
231    }
232    best.map(|(_, _, m)| m)
233}
234
235fn respond(status: u16, content_type: &'static str, body: Vec<u8>) -> Response<Full<Bytes>> {
236    let mut r = Response::new(Full::new(Bytes::from(body)));
237    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
238    r.headers_mut().insert(
239        header::CONTENT_TYPE,
240        header::HeaderValue::from_static(content_type),
241    );
242    r
243}
244
245fn graphql(status: u16, media: Media, body: &Value) -> Response<Full<Bytes>> {
246    respond(
247        status,
248        media.content_type(),
249        serde_json::to_vec(body).unwrap_or_default(),
250    )
251}
252
253/// A request error: 400 under `application/graphql-response+json`, 200 under legacy
254/// `application/json` for a well-formed request (the spec's compatibility rule).
255fn request_error(media: Media, well_formed: bool, errors: &RequestErrors) -> Response<Full<Bytes>> {
256    let status = if media == Media::Json && well_formed {
257        200
258    } else {
259        400
260    };
261    graphql(status, media, &errors.body())
262}
263
264fn not_allowed(allow: &'static str) -> Response<Full<Bytes>> {
265    let mut r = respond(405, "text/plain", b"method not allowed".to_vec());
266    r.headers_mut()
267        .insert(header::ALLOW, header::HeaderValue::from_static(allow));
268    r
269}
270
271fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
272    let summary = format!("GraphQL connection {id} operation={operation} decision={decision}");
273    let log = Log::new(Some(&ctx.status_tx));
274    if decision.starts_with("fail_closed_") || decision == "model_silent" {
275        log.error(summary);
276    } else {
277        log.info(summary);
278    }
279}
280
281async fn handle(
282    shared: &Shared,
283    id: ConnectionId,
284    req: Request<Incoming>,
285    upgrade: &UpgradeSlot,
286) -> Response<Full<Bytes>> {
287    let ctx = &shared.ctx;
288    ctx.state
289        .update_connection_stats(
290            ctx.server_id,
291            id,
292            Some(req.uri().to_string().len() as u64),
293            None,
294            Some(1),
295            None,
296        )
297        .await;
298    let response = route(shared, id, req, upgrade).await;
299    let sent = hyper::body::Body::size_hint(response.body())
300        .exact()
301        .unwrap_or(0);
302    ctx.state
303        .update_connection_stats(ctx.server_id, id, None, Some(sent), None, Some(1))
304        .await;
305    response
306}
307
308struct GraphqlRequest {
309    query: String,
310    operation_name: Option<String>,
311    variables: Map<String, Value>,
312}
313
314/// Decode a query-string component (`+` is a space, `%XX` a byte).
315fn percent_decode(s: &str) -> Option<String> {
316    let bytes = s.as_bytes();
317    let mut out = Vec::with_capacity(bytes.len());
318    let mut i = 0;
319    while i < bytes.len() {
320        match bytes[i] {
321            b'+' => out.push(b' '),
322            b'%' => {
323                let hex = s.get(i + 1..i + 3)?;
324                out.push(u8::from_str_radix(hex, 16).ok()?);
325                i += 2;
326            }
327            b => out.push(b),
328        }
329        i += 1;
330    }
331    String::from_utf8(out).ok()
332}
333
334fn decode_params(
335    query: Option<Value>,
336    operation_name: Option<Value>,
337    variables: Option<Value>,
338) -> Result<GraphqlRequest, &'static str> {
339    let query = match query {
340        Some(Value::String(q)) if !q.trim().is_empty() => q,
341        _ => return Err("query is required and must be a non-empty string"),
342    };
343    let operation_name = match operation_name {
344        None | Some(Value::Null) => None,
345        Some(Value::String(n)) if n.len() <= 256 => Some(n),
346        _ => return Err("operationName must be a string"),
347    };
348    let variables = match variables {
349        None | Some(Value::Null) => Map::new(),
350        Some(Value::Object(m)) => m,
351        _ => return Err("variables must be an object"),
352    };
353    Ok(GraphqlRequest {
354        query,
355        operation_name,
356        variables,
357    })
358}
359
360fn from_query_string(qs: &str) -> Result<GraphqlRequest, &'static str> {
361    let (mut query, mut name, mut vars) = (None, None, None);
362    for pair in qs.split('&').filter(|p| !p.is_empty()).take(16) {
363        let (k, v) = pair.split_once('=').unwrap_or((pair, ""));
364        let v = percent_decode(v).ok_or("query string is not valid UTF-8")?;
365        match k {
366            "query" => query = Some(Value::String(v)),
367            "operationName" if !v.is_empty() => name = Some(Value::String(v)),
368            "variables" if !v.is_empty() => {
369                vars = Some(serde_json::from_str(&v).map_err(|_| "variables is not JSON")?)
370            }
371            _ => {}
372        }
373    }
374    decode_params(query, name, vars)
375}
376
377async fn route(
378    shared: &Shared,
379    id: ConnectionId,
380    mut req: Request<Incoming>,
381    upgrade: &UpgradeSlot,
382) -> Response<Full<Bytes>> {
383    if req.uri().path() != shared.endpoint {
384        return respond(404, "text/plain", b"not found".to_vec());
385    }
386    if ws::is_upgrade(&req) {
387        return match ws::handshake(&req) {
388            Ok(accepted) => {
389                if let Ok(mut slot) = upgrade.lock() {
390                    *slot = Some(hyper::upgrade::on(&mut req));
391                }
392                accepted.map(|()| Full::new(Bytes::new()))
393            }
394            Err((status, why)) => respond(status, "text/plain", why.as_bytes().to_vec()),
395        };
396    }
397    let method = req.method().clone();
398    if method != Method::GET && method != Method::POST {
399        return not_allowed("GET, POST");
400    }
401    let accept = req
402        .headers()
403        .get(header::ACCEPT)
404        .and_then(|v| v.to_str().ok())
405        .map(str::to_owned);
406    let Some(media) = negotiate(accept.as_deref()) else {
407        return respond(
408            406,
409            "text/plain",
410            b"accept application/graphql-response+json or application/json".to_vec(),
411        );
412    };
413    let parsed = if method == Method::GET {
414        from_query_string(req.uri().query().unwrap_or(""))
415    } else {
416        let content_type = req
417            .headers()
418            .get(header::CONTENT_TYPE)
419            .and_then(|v| v.to_str().ok())
420            .unwrap_or("")
421            .split(';')
422            .next()
423            .unwrap_or("")
424            .trim()
425            .to_ascii_lowercase();
426        if content_type != "application/json" {
427            return respond(
428                415,
429                "text/plain",
430                b"POST bodies must be application/json".to_vec(),
431            );
432        }
433        let body = match tokio::time::timeout(
434            BODY_TIMEOUT,
435            Limited::new(req.into_body(), engine::MAX_BODY_BYTES).collect(),
436        )
437        .await
438        {
439            Ok(Ok(b)) => b.to_bytes(),
440            _ => {
441                return request_error(
442                    media,
443                    false,
444                    &RequestErrors::one("request body unreadable or over 1 MiB"),
445                )
446            }
447        };
448        match serde_json::from_slice::<Value>(&body) {
449            Ok(Value::Object(mut m)) if engine::budget_ok(&Value::Object(m.clone())) => {
450                decode_params(
451                    m.remove("query"),
452                    m.remove("operationName"),
453                    m.remove("variables"),
454                )
455            }
456            _ => Err("body must be a JSON object"),
457        }
458    };
459    let request = match parsed {
460        Ok(r) => r,
461        Err(e) => return request_error(media, false, &RequestErrors::one(e)),
462    };
463    let prepared = match engine::prepare(
464        &shared.schema,
465        &request.query,
466        request.operation_name.as_deref(),
467        &request.variables,
468    ) {
469        Ok(p) => p,
470        Err(errors) => {
471            outcome(&shared.ctx, id, "request", "protocol_refusal");
472            return request_error(media, true, &errors);
473        }
474    };
475    match prepared.operation_type {
476        OperationType::Mutation if method == Method::GET => return not_allowed("POST"),
477        OperationType::Subscription => {
478            return request_error(
479                media,
480                true,
481                &RequestErrors::one(
482                    "subscriptions need a WebSocket using the graphql-transport-ws subprotocol",
483                ),
484            )
485        }
486        _ => {}
487    }
488    match answer_operation(shared, id, &prepared, &request.query, method.as_str()).await {
489        Ok(body) => graphql(200, media, &body),
490        Err(e) => unavailable(media, &e),
491    }
492}
493
494/// Collect the handler's answers: custom results named in `allowed`, and a close request as
495/// `{"type": "disconnect"}`, in the order the handler gave them.
496fn collect_answers(results: Vec<ActionResult>, allowed: &[&str]) -> Vec<Value> {
497    let mut out = Vec::new();
498    for r in results {
499        match r {
500            ActionResult::Custom { name, data } if allowed.contains(&name.as_str()) => {
501                out.push(data)
502            }
503            ActionResult::CloseConnection => out.push(json!({"type": "disconnect"})),
504            ActionResult::Multiple(items) => out.extend(collect_answers(items, allowed)),
505            _ => {}
506        }
507    }
508    out
509}
510
511/// Ask the handler about `event`. Every failure is logged with its decision tag and returned
512/// as `Err`, which the caller turns into a category error — never into data.
513async fn ask(
514    shared: &Shared,
515    id: ConnectionId,
516    event: Event,
517    op_type: &str,
518    allowed: &[&str],
519) -> anyhow::Result<Vec<Value>> {
520    let ctx = &shared.ctx;
521    let result = match call_llm(
522        &ctx.llm_client,
523        &ctx.state,
524        ctx.server_id,
525        Some(id),
526        &event,
527        &actions::GraphqlProtocol,
528    )
529    .await
530    {
531        Ok(r) => r,
532        Err(e) => {
533            outcome(ctx, id, op_type, "fail_closed_llm_error");
534            return Err(e);
535        }
536    };
537    if !result.failures.is_empty() {
538        outcome(ctx, id, op_type, "fail_closed_invalid_reply");
539        anyhow::bail!("invalid handler answer");
540    }
541    Ok(collect_answers(result.protocol_results, allowed))
542}
543
544/// Answer a query or mutation: introspection by Rust alone, everything else by executing the
545/// handler's data. `Ok` is a complete GraphQL response body.
546async fn answer_operation(
547    shared: &Shared,
548    id: ConnectionId,
549    prepared: &engine::Prepared,
550    query: &str,
551    method: &str,
552) -> anyhow::Result<Value> {
553    if prepared.is_introspection() {
554        let r = prepared.execute(&shared.schema, &json!({}), shared.introspection)?;
555        return Ok(serde_json::to_value(r)?);
556    }
557    let op_type = engine::operation_type_name(prepared.operation_type);
558    let event = Event::new(
559        &actions::OPERATION_EVENT,
560        json!({
561            "operation_type": op_type,
562            "operation_name": prepared.operation_name,
563            "query": query,
564            "variables": prepared.variables_json(),
565            "root_fields": prepared.root_fields(),
566            "shape": prepared.shape(&shared.schema),
567            "method": method,
568        }),
569    );
570    let ctx = &shared.ctx;
571    let mut answers = ask(
572        shared,
573        id,
574        event,
575        op_type,
576        &["graphql_result", "graphql_error"],
577    )
578    .await?;
579    let answer = match answers.len() {
580        1 => answers.remove(0),
581        0 => {
582            outcome(ctx, id, op_type, "model_silent");
583            anyhow::bail!("no handler answer");
584        }
585        _ => {
586            outcome(ctx, id, op_type, "fail_closed_invalid_reply");
587            anyhow::bail!("more than one handler answer");
588        }
589    };
590    if answer["type"] == "graphql_error" {
591        outcome(ctx, id, op_type, "model_reject");
592        return Ok(json!({"errors": [refusal(&answer)], "data": null}));
593    }
594    match execute_answer(shared, prepared, &answer) {
595        Ok(body) => {
596            outcome(ctx, id, op_type, "model_answer");
597            Ok(body)
598        }
599        Err(e) => {
600            outcome(ctx, id, op_type, "fail_closed_invalid_reply");
601            Err(e)
602        }
603    }
604}
605
606/// A `graphql_error` answer as one GraphQL error.
607fn refusal(answer: &Value) -> Value {
608    let mut error = json!({"message": answer["message"]});
609    if let Some(x) = answer.get("extensions").filter(|x| x.is_object()) {
610        error["extensions"] = x.clone();
611    }
612    error
613}
614
615/// Execute a `graphql_result` or `graphql_event` answer over the operation, appending the
616/// handler's own errors.
617fn execute_answer(
618    shared: &Shared,
619    prepared: &engine::Prepared,
620    answer: &Value,
621) -> anyhow::Result<Value> {
622    let mut r = prepared.execute(&shared.schema, &answer["data"], shared.introspection)?;
623    r.errors
624        .extend(engine::handler_errors(answer.get("errors"))?);
625    Ok(serde_json::to_value(r)?)
626}
627
628/// The handler could not answer: 503 when overloaded, 500 otherwise, with a category message —
629/// never the error itself, and never data the handler did not give.
630fn unavailable(media: Media, e: &anyhow::Error) -> Response<Full<Bytes>> {
631    let failure = crate::utils::wire_failure::WireFailure::classify(e);
632    graphql(
633        if failure.is_overloaded() { 503 } else { 500 },
634        media,
635        &json!({"errors": [{"message": failure.text()}]}),
636    )
637}
netget/server/graphql/mod.rs

netget/server/graphql/
mod.rs
