use crate::graphql_plan::QueryPlan;
use serde_json::{json, Map, Value};
#[async_trait::async_trait]
pub(crate) trait SubgraphFetcher: Sync {
async fn fetch(&self, subgraph: &str, query: &str, variables: Value) -> Value;
}
pub(crate) async fn execute(plan: &QueryPlan, fetcher: &dyn SubgraphFetcher) -> Value {
let mut data = json!({});
for fetch in &plan.fetches {
match &fetch.requires {
None => {
let resp = fetcher
.fetch(&fetch.subgraph, &fetch.query, json!({}))
.await;
merge(&mut data, fetch_data(&resp));
}
Some(req) => {
let reprs = representations(&data, &req.path, &req.type_name, &req.key);
let resp = fetcher
.fetch(
&fetch.subgraph,
&fetch.query,
json!({ "representations": reprs }),
)
.await;
let entities = resp
.pointer("/data/_entities")
.or_else(|| resp.pointer("/_entities"))
.cloned()
.unwrap_or_else(|| json!([]));
stitch(&mut data, &req.path, &entities);
}
}
}
json!({ "data": data })
}
fn fetch_data(resp: &Value) -> &Value {
resp.get("data").unwrap_or(resp)
}
async fn invoke_subgraph(
invoker: &dyn boatramp_handlers::Invoker,
subgraph: &str,
query: &str,
variables: Value,
bearer: Option<&str>,
depth: u32,
) -> Value {
let body = json!({ "query": query, "variables": variables })
.to_string()
.into_bytes();
let mut headers = vec![("content-type".to_string(), b"application/json".to_vec())];
if let Some(token) = bearer {
headers.push((
"authorization".to_string(),
format!("Bearer {token}").into_bytes(),
));
}
let request = boatramp_handlers::InvokeRequest {
method: "POST".to_string(),
path: "/".to_string(),
headers,
body,
};
match invoker.invoke(subgraph, request, depth).await {
Ok(resp) => serde_json::from_slice(&resp.body).unwrap_or_else(|_| {
json!({ "errors": [{ "message": format!("subgraph `{subgraph}` returned invalid JSON") }] })
}),
Err(boatramp_handlers::InvokeError::NotFound) => json!({ "errors": [{
"message": format!(
"subgraph `{subgraph}` is registered but no function named `{subgraph}` is deployed"
)
}] }),
Err(boatramp_handlers::InvokeError::Failed(msg)) => json!({ "errors": [{
"message": format!("subgraph `{subgraph}` failed: {msg}")
}] }),
}
}
pub(crate) struct BackendRouter {
invoker: std::sync::Arc<dyn boatramp_handlers::Invoker>,
project: String,
sql_provider: Option<std::sync::Arc<dyn boatramp_core::sql::SqlBackends>>,
sql_subgraphs: std::collections::BTreeMap<
String,
(String, boatramp_core::config::HandlerGraphqlDataConfig),
>,
bearer: Option<String>,
depth: u32,
}
impl BackendRouter {
pub(crate) fn new(
invoker: std::sync::Arc<dyn boatramp_handlers::Invoker>,
project: String,
sql_provider: Option<std::sync::Arc<dyn boatramp_core::sql::SqlBackends>>,
sql_subgraphs: std::collections::BTreeMap<
String,
(String, boatramp_core::config::HandlerGraphqlDataConfig),
>,
bearer: Option<String>,
) -> Self {
Self {
invoker,
project,
sql_provider,
sql_subgraphs,
bearer,
depth: 0,
}
}
pub(crate) fn at_depth(mut self, depth: u32) -> Self {
self.depth = depth;
self
}
async fn run_sql(
&self,
subgraph: &str,
site: &str,
config: &boatramp_core::config::HandlerGraphqlDataConfig,
query: &str,
variables: Value,
) -> Value {
let Some(provider) = &self.sql_provider else {
return json!({ "errors": [{ "message": "the federation gateway has no SQL backend configured" }] });
};
let backend = match provider.database(&self.project, site, &config.source).await {
Ok(backend) => backend,
Err(err) => {
return json!({ "errors": [{ "message": format!("subgraph `{subgraph}` database unavailable: {err}") }] })
}
};
let schema = match crate::graphql_data::introspect::introspect_sqlite(backend.as_ref())
.await
{
Ok(schema) => schema,
Err(err) => {
return json!({ "errors": [{ "message": format!("subgraph `{subgraph}` introspection failed: {err}") }] })
}
};
let policy = crate::graphql_data::policy_from_config(config);
let claims =
crate::graphql_data::request_claims(&self.project, self.bearer.as_deref(), config)
.await;
let dialect = crate::graphql_data::dialect::Sqlite;
let invoker = Some(self.invoker.as_ref());
if crate::graphql_data::compile::is_entities_query(query) {
crate::graphql_data::runner::execute_entities(
backend.as_ref(),
&dialect,
&schema,
&policy,
&claims,
query,
&variables,
invoker,
self.bearer.as_deref(),
self.depth,
)
.await
} else {
crate::graphql_data::runner::execute(
backend.as_ref(),
&dialect,
&schema,
&policy,
&claims,
query,
&variables,
invoker,
self.bearer.as_deref(),
self.depth,
)
.await
}
}
}
#[async_trait::async_trait]
impl SubgraphFetcher for BackendRouter {
async fn fetch(&self, subgraph: &str, query: &str, variables: Value) -> Value {
if let Some((site, config)) = self.sql_subgraphs.get(subgraph) {
return self.run_sql(subgraph, site, config, query, variables).await;
}
invoke_subgraph(
self.invoker.as_ref(),
subgraph,
query,
variables,
self.bearer.as_deref(),
self.depth,
)
.await
}
}
pub(crate) struct FederationRunner {
runtime: std::sync::Weak<crate::HandlerRuntimeInner>,
project: String,
}
impl FederationRunner {
pub(crate) fn new(runtime: std::sync::Weak<crate::HandlerRuntimeInner>) -> Self {
Self {
runtime,
project: boatramp_core::project::DEFAULT_PROJECT.to_string(),
}
}
pub(crate) fn scoped(
&self,
project: boatramp_core::project::ProjectRef<'_>,
) -> std::sync::Arc<dyn boatramp_handlers::SupergraphRunner> {
std::sync::Arc::new(Self {
runtime: self.runtime.clone(),
project: project.as_str().to_string(),
})
}
}
fn strip_bearer(raw: &str) -> &str {
raw.strip_prefix("Bearer ")
.or_else(|| raw.strip_prefix("bearer "))
.unwrap_or(raw)
}
#[async_trait::async_trait]
impl boatramp_handlers::SupergraphRunner for FederationRunner {
async fn run(
&self,
request: boatramp_handlers::GraphqlRequest,
depth: u32,
) -> Result<Vec<u8>, boatramp_handlers::SupergraphRunError> {
use boatramp_handlers::SupergraphRunError;
let Some(inner) = self.runtime.upgrade() else {
return Err(SupergraphRunError::Failed(
"handler runtime is shutting down".into(),
));
};
let kv = inner.kv.as_ref();
let project = self.project.as_str();
let hash = match (&request.query, &request.persisted_hash) {
(Some(query), _) => crate::graphql_apq::sha256_hex(query),
(None, Some(hash)) => hash.clone(),
(None, None) => {
return Err(SupergraphRunError::PlanFailed(
"no query or persisted hash supplied".into(),
))
}
};
let Some(query) = crate::graphql_apq::safelisted_query(kv, project, &hash).await else {
return Err(SupergraphRunError::NotSafelisted);
};
let limits = crate::graphql_guard::limits_from(
&boatramp_core::config::HandlerGraphqlConfig::default(),
);
if let crate::graphql_guard::GuardVerdict::Reject(reason) =
crate::graphql_guard::guard_query(&query, &limits)
{
return Err(SupergraphRunError::PlanFailed(reason));
}
let supergraph = crate::graphql_registry::supergraph(kv, project)
.await
.map_err(|e| {
SupergraphRunError::Failed(format!("supergraph composition failed: {e}"))
})?;
let plan = crate::graphql_plan::plan(&query, &supergraph)
.map_err(|_| SupergraphRunError::PlanFailed("the query cannot be planned".into()))?;
let Some(invoker) = inner.invoker.get() else {
return Err(SupergraphRunError::Failed("no invoker configured".into()));
};
let sql_subgraphs = crate::graphql_registry::sql_subgraphs(kv, project).await;
let bearer = request
.authorization
.as_deref()
.map(|raw| strip_bearer(raw).to_string());
let router = BackendRouter::new(
invoker.scoped(boatramp_core::project::ProjectRef::new(project)),
project.to_string(),
inner.sql.clone(),
sql_subgraphs,
bearer,
)
.at_depth(depth);
let response = execute(&plan, &router).await;
serde_json::to_vec(&response)
.map_err(|e| SupergraphRunError::Failed(format!("serializing response: {e}")))
}
}
fn merge(dst: &mut Value, src: &Value) {
match (dst, src) {
(Value::Object(d), Value::Object(s)) => {
for (k, v) in s {
merge(d.entry(k.clone()).or_insert(Value::Null), v);
}
}
(d, s) => *d = s.clone(),
}
}
fn navigate<'a>(data: &'a Value, path: &[String]) -> Option<&'a Value> {
let mut cur = data;
for seg in path {
cur = cur.get(seg)?;
}
Some(cur)
}
fn navigate_mut<'a>(data: &'a mut Value, path: &[String]) -> Option<&'a mut Value> {
let mut cur = data;
for seg in path {
cur = cur.get_mut(seg)?;
}
Some(cur)
}
fn representations(data: &Value, path: &[String], type_name: &str, key: &[String]) -> Value {
let mut out = Vec::new();
if let Some(node) = navigate(data, path) {
collect_reprs(node, type_name, key, &mut out);
}
Value::Array(out)
}
fn collect_reprs(node: &Value, type_name: &str, key: &[String], out: &mut Vec<Value>) {
match node {
Value::Array(items) => {
for item in items {
collect_reprs(item, type_name, key, out);
}
}
Value::Object(_) => {
let mut repr = Map::new();
repr.insert("__typename".to_string(), json!(type_name));
for k in key {
if let Some(v) = node.get(k) {
repr.insert(k.clone(), v.clone());
}
}
out.push(Value::Object(repr));
}
_ => {}
}
}
fn stitch(data: &mut Value, path: &[String], entities: &Value) {
let Some(node) = navigate_mut(data, path) else {
return;
};
let ents = entities.as_array().cloned().unwrap_or_default();
match node {
Value::Array(items) => {
for (item, ent) in items.iter_mut().zip(ents.iter()) {
merge(item, ent);
}
}
Value::Object(_) => {
if let Some(ent) = ents.first() {
merge(node, ent);
}
}
_ => {}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::graphql_federation::compose;
use crate::graphql_plan::plan;
use std::collections::HashMap;
const ACCOUNTS: &str = r#"
type Query { me: User }
type User @key(fields: "id") { id: ID! name: String }
"#;
const ACCOUNTS_LIST: &str = r#"
type Query { users: [User] }
type User @key(fields: "id") { id: ID! name: String }
"#;
const REVIEWS: &str = r#"
type Query { topReviews: [Review] }
type Review { id: ID! body: String }
extend type User @key(fields: "id") { id: ID! @external reviews: [Review] }
"#;
struct Mock(HashMap<&'static str, Value>);
#[async_trait::async_trait]
impl SubgraphFetcher for Mock {
async fn fetch(&self, subgraph: &str, _query: &str, _variables: Value) -> Value {
self.0.get(subgraph).cloned().unwrap_or_else(|| json!({}))
}
}
struct ContractRunner;
#[async_trait::async_trait]
impl SubgraphFetcher for ContractRunner {
async fn fetch(&self, subgraph: &str, query: &str, variables: Value) -> Value {
match subgraph {
"accounts" => json!({ "data": { "users": [
{ "__typename": "User", "id": "1", "name": "Alice" },
{ "__typename": "User", "id": "2", "name": "Bob" },
] } }),
"reviews" => {
assert!(
query.contains("_entities"),
"entity fetch must use _entities"
);
let reprs = variables
.get("representations")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
let entities: Vec<Value> = reprs
.iter()
.map(|r| {
let id = r.get("id").and_then(|v| v.as_str()).unwrap_or("");
json!({ "reviews": [ { "body": format!("review for {id}") } ] })
})
.collect();
json!({ "data": { "_entities": entities } })
}
other => json!({ "errors": [{ "message": format!("unknown subgraph {other}") }] }),
}
}
}
struct MissingInvoker;
#[async_trait::async_trait]
impl boatramp_handlers::Invoker for MissingInvoker {
async fn invoke(
&self,
_target: &str,
_request: boatramp_handlers::InvokeRequest,
_depth: u32,
) -> Result<boatramp_handlers::InvokeResponse, boatramp_handlers::InvokeError> {
Err(boatramp_handlers::InvokeError::NotFound)
}
}
struct AuthEchoInvoker;
#[async_trait::async_trait]
impl boatramp_handlers::Invoker for AuthEchoInvoker {
async fn invoke(
&self,
_target: &str,
request: boatramp_handlers::InvokeRequest,
_depth: u32,
) -> Result<boatramp_handlers::InvokeResponse, boatramp_handlers::InvokeError> {
let authz = request
.headers
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case("authorization"))
.map(|(_, v)| String::from_utf8_lossy(v).into_owned());
let body = match authz {
Some(value) => json!({ "data": { "identity": value } }),
None => json!({ "errors": [
{ "message": "unauthenticated", "extensions": { "code": "UNAUTHENTICATED" } }
] }),
};
Ok(boatramp_handlers::InvokeResponse {
status: 200,
headers: vec![("content-type".to_string(), b"application/json".to_vec())],
body: serde_json::to_vec(&body).unwrap(),
})
}
}
#[tokio::test]
async fn merges_root_fetches_from_distinct_subgraphs() {
let sg = compose(&[
("accounts".into(), ACCOUNTS.into()),
("reviews".into(), REVIEWS.into()),
])
.unwrap();
let plan = plan("{ me { name } topReviews { body } }", &sg).unwrap();
let mock = Mock(HashMap::from([
("accounts", json!({ "data": { "me": { "name": "Alice" } } })),
(
"reviews",
json!({ "data": { "topReviews": [{ "body": "ok" }] } }),
),
]));
let out = execute(&plan, &mock).await;
assert_eq!(out["data"]["me"]["name"], json!("Alice"));
assert_eq!(out["data"]["topReviews"][0]["body"], json!("ok"));
}
#[tokio::test]
async fn stitches_a_cross_subgraph_entity_field() {
let sg = compose(&[
("accounts".into(), ACCOUNTS.into()),
("reviews".into(), REVIEWS.into()),
])
.unwrap();
let plan = plan("{ me { name reviews { body } } }", &sg).unwrap();
let mock = Mock(HashMap::from([
(
"accounts",
json!({ "data": { "me": { "name": "Alice", "__typename": "User", "id": "1" } } }),
),
(
"reviews",
json!({ "data": { "_entities": [{ "reviews": [{ "body": "great" }] }] } }),
),
]));
let out = execute(&plan, &mock).await;
assert_eq!(out["data"]["me"]["name"], json!("Alice"));
assert_eq!(out["data"]["me"]["reviews"][0]["body"], json!("great"));
}
#[tokio::test]
async fn executes_a_list_entity_fetch_joining_each_element_by_its_key() {
let sg = compose(&[
("accounts".into(), ACCOUNTS_LIST.into()),
("reviews".into(), REVIEWS.into()),
])
.unwrap();
let plan = plan("{ users { name reviews { body } } }", &sg).unwrap();
let out = execute(&plan, &ContractRunner).await;
assert_eq!(out["data"]["users"][0]["name"], json!("Alice"));
assert_eq!(
out["data"]["users"][0]["reviews"][0]["body"],
json!("review for 1")
);
assert_eq!(out["data"]["users"][1]["name"], json!("Bob"));
assert_eq!(
out["data"]["users"][1]["reviews"][0]["body"],
json!("review for 2")
);
}
#[tokio::test]
async fn a_function_subgraph_that_is_not_deployed_is_reported_precisely() {
let router = BackendRouter::new(
std::sync::Arc::new(MissingInvoker),
"default".to_string(),
None,
std::collections::BTreeMap::new(),
None,
);
let resp = router.fetch("accounts", "{ me { id } }", json!({})).await;
let msg = resp["errors"][0]["message"].as_str().unwrap_or_default();
assert!(
msg.contains("no function named `accounts` is deployed"),
"unexpected error: {msg}"
);
}
#[tokio::test]
async fn forwards_the_callers_verified_bearer_to_a_function_subgraph() {
let router = BackendRouter::new(
std::sync::Arc::new(AuthEchoInvoker),
"default".to_string(),
None,
std::collections::BTreeMap::new(),
Some("t-acme".to_string()),
);
let root = router.fetch("orders", "{ me { id } }", json!({})).await;
assert_eq!(root["data"]["identity"], json!("Bearer t-acme"));
let entity = router
.fetch(
"orders",
"query($r: [_Any!]!) { _entities(representations: $r) { id } }",
json!({ "representations": [{ "__typename": "Order", "id": "1" }] }),
)
.await;
assert_eq!(
entity["data"]["identity"],
json!("Bearer t-acme"),
"the bearer must ride the _entities fetch too, or a stitched field would go anonymous"
);
}
#[tokio::test]
async fn an_anonymous_gateway_call_forwards_no_bearer_so_an_authed_field_is_refused() {
let router = BackendRouter::new(
std::sync::Arc::new(AuthEchoInvoker),
"default".to_string(),
None,
std::collections::BTreeMap::new(),
None,
);
let resp = router.fetch("orders", "{ me { id } }", json!({})).await;
assert_eq!(
resp["errors"][0]["extensions"]["code"],
json!("UNAUTHENTICATED"),
"with no forwarded identity a subgraph's authed field must refuse, not resolve anonymously"
);
}
#[test]
fn merge_is_a_deep_object_merge() {
let mut a = json!({ "me": { "name": "x" } });
merge(&mut a, &json!({ "me": { "age": 3 }, "other": 1 }));
assert_eq!(a, json!({ "me": { "name": "x", "age": 3 }, "other": 1 }));
}
struct DepthEchoInvoker;
#[async_trait::async_trait]
impl boatramp_handlers::Invoker for DepthEchoInvoker {
async fn invoke(
&self,
_target: &str,
_request: boatramp_handlers::InvokeRequest,
depth: u32,
) -> Result<boatramp_handlers::InvokeResponse, boatramp_handlers::InvokeError> {
Ok(boatramp_handlers::InvokeResponse {
status: 200,
headers: vec![("content-type".to_string(), b"application/json".to_vec())],
body: serde_json::to_vec(&json!({ "data": { "depth": depth } })).unwrap(),
})
}
}
#[tokio::test]
async fn at_depth_dispatches_function_fetches_at_that_depth() {
let root = BackendRouter::new(
std::sync::Arc::new(DepthEchoInvoker),
"default".to_string(),
None,
std::collections::BTreeMap::new(),
None,
);
assert_eq!(
root.fetch("s", "{ x }", json!({})).await["data"]["depth"],
json!(0)
);
let scoped = BackendRouter::new(
std::sync::Arc::new(DepthEchoInvoker),
"default".to_string(),
None,
std::collections::BTreeMap::new(),
None,
)
.at_depth(4);
assert_eq!(
scoped.fetch("s", "{ x }", json!({})).await["data"]["depth"],
json!(4)
);
}
}