use super::*;
impl GraphqlEngine {
pub async fn execute(&self, session: &Session, mut request: Request) -> Response {
let authority = match resolve_execution_authority(&self.inner, session, &request) {
Ok(authority) => authority,
Err(()) => {
return Response::from_errors(vec![ServerError::new(
"GraphQL execution requires a named application surface for multi-role principals, a membership-checked role surface, or an anonymous session",
None,
)]);
}
};
let privilege = authority.privilege_role.clone();
let introspection = self.inner.graphiql && is_pure_introspection_request(&mut request);
let schema = if introspection {
self.inner
.graphiql_schemas
.get(&privilege)
.or_else(|| self.inner.schemas.get(&privilege))
} else {
self.inner.schemas.get(&privilege)
};
let Some(schema) = schema else {
return Response::from_errors(vec![ServerError::new(
format!("privilege pack `{privilege}` is not configured for GraphQL"),
None,
)]);
};
if has_multiple_protocol_query_roots(&self.inner, &privilege, &mut request) {
return protocol_multi_root_error_response();
}
let accumulator = match self.protocol_accumulator(&authority, session, &request) {
Ok(accumulator) => accumulator,
Err(()) => return protocol_internal_error_response(),
};
if introspection {
request = request.only_introspection();
}
let mut request = request
.data(session.clone())
.data(authority)
.data(Arc::clone(&self.inner));
if let Some(accumulator) = &accumulator {
request = request.data(accumulator.clone());
}
let start = std::time::Instant::now();
let response =
attach_protocol_response(schema.execute(request).await, accumulator.as_ref());
let status = metrics_status_for_response(&response);
let root_field = match &response.data {
Value::Object(map) => map.keys().next().map(|s| s.as_str()).unwrap_or("_"),
_ => "_",
};
record_metrics(session, root_field, status, start.elapsed());
response
}
pub fn execute_stream(
&self,
session: &Session,
mut request: Request,
) -> BoxStream<'static, async_graphql::Response> {
let authority = match resolve_execution_authority(&self.inner, session, &request) {
Ok(authority) => authority,
Err(()) => {
return stream::once(async {
Response::from_errors(vec![ServerError::new(
"GraphQL execution requires a named application surface for multi-role principals, a membership-checked role surface, or an anonymous session",
None,
)])
})
.boxed();
}
};
let privilege = authority.privilege_role.clone();
let introspection = self.inner.graphiql && is_pure_introspection_request(&mut request);
let schema = if introspection {
self.inner
.graphiql_schemas
.get(&privilege)
.or_else(|| self.inner.schemas.get(&privilege))
} else {
self.inner.schemas.get(&privilege)
};
let Some(schema) = schema.cloned() else {
return stream::once(async move {
Response::from_errors(vec![ServerError::new(
format!("privilege pack `{privilege}` is not configured for GraphQL"),
None,
)])
})
.boxed();
};
if has_multiple_protocol_query_roots(&self.inner, &privilege, &mut request) {
return stream::once(async { protocol_multi_root_error_response() }).boxed();
}
let accumulator = match self.protocol_accumulator(&authority, session, &request) {
Ok(accumulator) => accumulator,
Err(()) => {
return stream::once(async { protocol_internal_error_response() }).boxed();
}
};
if accumulator
.as_ref()
.is_some_and(|accumulator| accumulator.begin_stream().is_err())
{
return stream::once(async { protocol_internal_error_response() }).boxed();
}
if introspection {
request = request.only_introspection();
}
let mut request = request
.data(session.clone())
.data(authority)
.data(std::sync::Arc::clone(&self.inner));
if let Some(accumulator) = &accumulator {
request = request.data(accumulator.clone());
}
schema
.execute_stream(request)
.map(move |response| attach_protocol_response(response, accumulator.as_ref()))
.boxed()
}
fn protocol_accumulator(
&self,
authority: &ExecutionAuthority,
session: &Session,
request: &Request,
) -> Result<Option<ProtocolResponseAccumulator>, ()> {
let Some(runtime) = &self.inner.protocol else {
return Ok(None);
};
let (surface_identity, surface_info, authorization_fingerprint, claim_keys) =
select_protocol_surface(runtime, authority)?;
let trusted_presets = surface_info
.trusted_presets
.iter()
.map(|descriptor| resolve_protocol_preset(session, descriptor).ok_or(()))
.collect::<Result<Vec<_>, _>>()?;
let principal = request
.data
.get(&TypeId::of::<VerifiedPrincipal>())
.and_then(|principal| principal.downcast_ref::<VerifiedPrincipal>());
let principal_partition =
principal.map(|principal| principal.partition_for_service(&runtime.service_id));
let projection_principal = principal_partition
.as_deref()
.map(crate::command_ledger::PrincipalPartitionId::new)
.transpose()
.map_err(|_| ())?;
let session_authorization_context = claim_keys
.iter()
.map(|key| (key.as_str(), session.get(key)))
.collect::<Vec<_>>();
#[derive(Serialize)]
struct CacheScopeMaterial<'a> {
domain: &'static str,
version: u32,
namespace: &'a str,
service_id: &'a str,
privilege: &'a str,
asserted_roles: &'a [String],
surface: &'a ClientSurfaceIdentity,
schema_fingerprint: &'a str,
protocol_fingerprint: &'a str,
authorization_surface_fingerprint: &'a str,
identity_mode: &'static str,
verified_principal_partition: Option<&'a str>,
session_authorization_context: Vec<(&'a str, Option<&'a str>)>,
trusted_presets: &'a [DistributedTrustedPreset],
}
let material = CacheScopeMaterial {
domain: "distributed.graphql.cache-scope",
version: 2,
namespace: &runtime.namespace,
service_id: &runtime.service_id,
privilege: &authority.privilege_role,
asserted_roles: &authority.asserted_roles,
surface: &surface_identity,
schema_fingerprint: &surface_info.schema_fingerprint,
protocol_fingerprint: &surface_info.protocol_fingerprint,
authorization_surface_fingerprint: authorization_fingerprint,
identity_mode: identity_mode_label(self.inner.identity.mode),
verified_principal_partition: principal_partition.as_deref(),
session_authorization_context,
trusted_presets: &trusted_presets,
};
let cache_scope = runtime
.codec
.issue(ProtocolTokenPurpose::CacheScope, &material)
.map_err(|_| ())?;
let envelope = DistributedEnvelopeV1::new(
surface_info.schema_fingerprint.clone(),
authorization_fingerprint.to_string(),
cache_scope,
Some(operation_fingerprint(&request.query)),
)
.with_trusted_presets(trusted_presets.clone());
let accumulator = ProtocolResponseAccumulator::new(envelope, runtime.codec.clone());
if let Some(principal_scope) = projection_principal {
let visibility_surface = self
.inner
.role_surfaces
.get(&authority.privilege_role)
.cloned()
.ok_or(())?;
let selected_surface = match &surface_identity {
ClientSurfaceIdentity::Role { name } => self.inner.role_surfaces.get(name),
ClientSurfaceIdentity::Application { name, .. } => {
self.inner.application_surfaces.get(name)
}
}
.cloned()
.ok_or(())?;
let export = DistributedClientSurfaceExport::from_selected_with_execution(
&runtime.service_id,
selected_surface,
ClientExecutionLimits::from_runtime(
self.inner.max_depth,
self.inner.max_complexity,
self.inner.max_bool_width,
self.inner.max_in_list,
)
.map_err(|_| ())?,
)
.map_err(|_| ())?;
let issued_at_unix_ms = crate::time::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(|_| ())?
.as_millis()
.try_into()
.map_err(|_| ())?;
let projection_request =
crate::graphql::projection_delta::runtime::ProtocolProjectionRequestSeed::new(
export,
Arc::clone(&runtime.projection_programs),
principal_scope,
authorization_fingerprint.to_string(),
trusted_presets,
issued_at_unix_ms,
)
.map_err(|_| ())?
.with_visibility_surface(visibility_surface)
.map_err(|_| ())?;
accumulator
.bind_projection_request(projection_request)
.map_err(|_| ())?;
}
accumulator
.set_requested_live_resume(parse_requested_live_resume(request))
.map_err(|_| ())?;
Ok(Some(accumulator))
}
}