use crate::context::GraphQLContext;
use crate::error::GraphQLError;
use crate::schema::object::TableObjectType;
use crate::schema::relationship::RelationshipField;
use crate::schema::{build_schema, FederationEntity, GeneratedSchema, MutationType, SchemaConfig};
use crate::subscription::{
generate_subscription_fields, NotifyBroker, SubscriptionField as SubField,
};
use async_graphql::dynamic::*;
use async_graphql::extensions::{
Extension, ExtensionContext, ExtensionFactory, NextResolve, ResolveInfo,
};
use async_graphql::parser::types::{FragmentDefinition, Selection, SelectionSet, TypeCondition};
use async_graphql::{Name, Positioned, SelectionField, ServerResult, Value};
use async_graphql_axum::{GraphQLRequest, GraphQLResponse};
use axum::extract::State;
use axum::response::IntoResponse;
use postrust_core::schema_cache::SchemaCache;
use sqlx::{AssertSqlSafe, PgPool};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, info, trace};
pub struct GraphQLState {
pub pool: PgPool,
pub schema_cache: Arc<SchemaCache>,
pub generated_schema: GeneratedSchema,
pub schema: Schema,
pub role_schemas: HashMap<(String, bool), Schema>,
pub ungranted_schema: Option<Schema>,
pub config: SchemaConfig,
pub subscription_fields: Vec<SubField>,
pub broker: Arc<RwLock<Option<NotifyBroker>>>,
}
struct NullableFederationEntity;
impl ExtensionFactory for NullableFederationEntity {
fn create(&self) -> Arc<dyn Extension> {
Arc::new(Self)
}
}
#[async_graphql::async_trait::async_trait]
impl Extension for NullableFederationEntity {
async fn resolve(
&self,
ctx: &ExtensionContext<'_>,
info: ResolveInfo<'_>,
next: NextResolve<'_>,
) -> ServerResult<Option<Value>> {
let parent_type = info.parent_type;
let return_type = info.return_type;
match next.run(ctx, info).await {
Err(error)
if error.message
== "internal: invalid value for union \"_Entity\", expected \"FieldValue::WithType\""
&& parent_type == "[_Entity]"
&& return_type == "_Entity" =>
{
Ok(Some(Value::Null))
}
result => result,
}
}
}
fn build_role_schemas(
schema_cache: &Arc<SchemaCache>,
config: &SchemaConfig,
) -> Result<HashMap<(String, bool), Schema>, GraphQLError> {
if !config.names.has_permissions() {
return Ok(HashMap::new());
}
let mut schemas = HashMap::new();
for role in config.names.roles() {
if role == postrust_auth::hasura::ADMIN_ROLE {
continue;
}
let variants: &[bool] = match crate::role::has_backend_only(&config.names, role) {
true => &[false, true],
false => &[false],
};
for &backend in variants {
let view = Arc::new(crate::role::cache_for_role(
schema_cache,
&config.names,
role,
backend,
));
let mut role_config = config.clone();
role_config.role = Some(role.to_string());
let generated = build_schema(&view, &role_config);
let built = build_dynamic_schema(
&generated,
&view,
None,
role_config.max_rows,
Arc::new(role_config.names.clone()),
role_config.subscription_refresh(),
Some(role),
);
match built {
Ok(schema) => {
schemas.insert((role.to_string(), backend), schema);
}
Err(e) => tracing::error!(
"GraphQL schema for role \"{}\" could not be built, \
so that role is refused: {}",
role,
e
),
}
}
}
tracing::info!("GraphQL schemas built for {} roles", schemas.len());
Ok(schemas)
}
fn build_ungranted_schema(
schema_cache: &Arc<SchemaCache>,
config: &SchemaConfig,
) -> Option<Schema> {
if !config.names.has_permissions() {
return None;
}
const NOBODY: &str = "\u{0}ungranted";
let view = Arc::new(crate::role::cache_for_role(
schema_cache,
&config.names,
NOBODY,
false,
));
let mut role_config = config.clone();
role_config.role = Some(NOBODY.to_string());
let generated = build_schema(&view, &role_config);
build_dynamic_schema(
&generated,
&view,
None,
role_config.max_rows,
Arc::new(role_config.names.clone()),
role_config.subscription_refresh(),
Some(NOBODY),
)
.map_err(|e| {
tracing::error!(
"the schema for a role with nothing granted could not be built, so such a role is refused instead: {}",
e
)
})
.ok()
}
impl GraphQLState {
pub fn new(
pool: PgPool,
schema_cache: Arc<SchemaCache>,
config: SchemaConfig,
) -> Result<Self, GraphQLError> {
let generated_schema = build_schema(&schema_cache, &config);
let subscription_fields = if config.enable_subscriptions {
generate_subscription_fields(&schema_cache, &generated_schema)
} else {
Vec::new()
};
let schema = build_dynamic_schema(
&generated_schema,
&schema_cache,
if config.enable_subscriptions {
Some(subscription_fields.as_slice())
} else {
None
},
config.max_rows,
Arc::new(config.names.clone()),
config.subscription_refresh(),
None,
)?;
let role_schemas = build_role_schemas(&schema_cache, &config)?;
let ungranted_schema = build_ungranted_schema(&schema_cache, &config);
Ok(Self {
pool: pool.clone(),
schema_cache,
generated_schema,
schema,
role_schemas,
ungranted_schema,
config,
subscription_fields,
broker: Arc::new(RwLock::new(None)),
})
}
pub fn schema_for(&self, role: Option<&str>, backend: bool) -> Option<&Schema> {
if !self.config.names.has_permissions() {
return Some(&self.schema);
}
match role {
None | Some(postrust_auth::hasura::ADMIN_ROLE) => Some(&self.schema),
Some(role) => self
.role_schemas
.get(&(role.to_string(), backend))
.or_else(|| self.role_schemas.get(&(role.to_string(), false)))
.or(self.ungranted_schema.as_ref()),
}
}
#[allow(clippy::needless_update)]
pub fn rebuild(&mut self) -> Result<(), GraphQLError> {
self.role_schemas = build_role_schemas(&self.schema_cache, &self.config)?;
self.ungranted_schema = build_ungranted_schema(&self.schema_cache, &self.config);
self.generated_schema = build_schema(&self.schema_cache, &self.config);
self.subscription_fields = if self.config.enable_subscriptions {
generate_subscription_fields(&self.schema_cache, &self.generated_schema)
} else {
Vec::new()
};
self.schema = build_dynamic_schema(
&self.generated_schema,
&self.schema_cache,
if self.config.enable_subscriptions {
Some(self.subscription_fields.as_slice())
} else {
None
},
self.config.max_rows,
Arc::new(self.config.names.clone()),
self.config.subscription_refresh(),
None,
)?;
Ok(())
}
pub async fn init_subscriptions(&self) -> Result<(), crate::subscription::BrokerError> {
if !self.config.enable_subscriptions {
return Ok(());
}
let broker = NotifyBroker::new(self.pool.clone());
let channels: Vec<String> = self
.subscription_fields
.iter()
.map(|f| f.channel_name())
.collect();
if !channels.is_empty() {
broker.start(channels).await?;
info!(
"Subscription broker started with {} channels",
self.subscription_fields.len()
);
}
let mut broker_guard = self.broker.write().await;
*broker_guard = Some(broker);
Ok(())
}
pub async fn stop_subscriptions(&self) {
let broker_guard = self.broker.read().await;
if let Some(broker) = broker_guard.as_ref() {
broker.stop().await;
}
}
pub async fn continue_subscriptions(
&mut self,
previous: Option<&GraphQLState>,
) -> Result<(), crate::subscription::BrokerError> {
if let Some(previous) = previous {
self.broker = Arc::clone(&previous.broker);
}
if !self.config.enable_subscriptions {
return Ok(());
}
if let Some(broker) = self.broker.read().await.as_ref() {
let channels = self
.subscription_fields
.iter()
.map(|f| f.channel_name())
.collect();
return broker.listen(channels).await;
}
self.init_subscriptions().await
}
pub async fn get_broker(&self) -> Option<Arc<RwLock<Option<NotifyBroker>>>> {
Some(Arc::clone(&self.broker))
}
}
pub async fn graphql_handler(
State(state): State<Arc<GraphQLState>>,
ctx: GraphQLContext,
req: GraphQLRequest,
) -> GraphQLResponse {
let request = req
.into_inner()
.data(ctx)
.data(state.pool.clone())
.data(Arc::clone(&state.broker));
state.schema.execute(request).await.into()
}
pub async fn graphql_ws_handler(
State(state): State<Arc<GraphQLState>>,
protocol: async_graphql_axum::GraphQLProtocol,
ws: axum::extract::WebSocketUpgrade,
) -> impl IntoResponse {
let schema = state.schema.clone();
let pool = state.pool.clone();
let broker = Arc::clone(&state.broker);
ws.protocols(["graphql-transport-ws", "graphql-ws"])
.on_upgrade(move |socket| async move {
let mut data = async_graphql::Data::default();
data.insert(pool);
data.insert(broker);
async_graphql_axum::GraphQLWebSocket::new(socket, schema, protocol)
.with_data(data)
.serve()
.await
})
}
pub async fn graphql_playground() -> impl axum::response::IntoResponse {
axum::response::Html(async_graphql::http::playground_source(
async_graphql::http::GraphQLPlaygroundConfig::new("/graphql")
.subscription_endpoint("/graphql/ws"),
))
}
fn column_for<'a>(
names: &'a crate::names::NameOverrides,
schema: &str,
table: &str,
field: &'a str,
) -> &'a str {
names.column_source(schema, table, field).unwrap_or(field)
}
fn table_column_for<'a>(
names: &'a crate::names::NameOverrides,
table: &postrust_core::schema_cache::Table,
field: &'a str,
) -> &'a str {
column_for(names, &table.schema, &table.name, field)
}
fn rename_projection(
table: &postrust_core::schema_cache::Table,
alias: &str,
names: &crate::names::NameOverrides,
) -> Option<String> {
if !names.renames_columns(&table.schema, &table.name) {
return None;
}
let parts: Vec<String> = table
.columns
.values()
.map(|column| {
let exposed = names
.column(&table.schema, &table.name, &column.name)
.unwrap_or(&column.name);
format!(
"{}.{} AS {}",
postrust_sql::escape_ident(alias),
postrust_sql::escape_ident(&column.name),
postrust_sql::escape_ident(exposed)
)
})
.collect();
match parts.is_empty() {
true => None,
false => Some(parts.join(", ")),
}
}
fn exposed_column_types(
table: &postrust_core::schema_cache::Table,
names: &crate::names::NameOverrides,
) -> HashMap<String, String> {
table
.columns
.values()
.map(|column| {
let exposed = names
.column(&table.schema, &table.name, &column.name)
.unwrap_or(&column.name);
(exposed.to_string(), column.nominal_type.clone())
})
.collect()
}
fn write_type_ref(graphql_type: &crate::types::GraphQLType) -> TypeRef {
match graphql_type {
crate::types::GraphQLType::List(inner) => TypeRef::named_list(leaf_scalar_name(inner)),
other => TypeRef::named(leaf_scalar_name(other)),
}
}
fn leaf_scalar_name(graphql_type: &crate::types::GraphQLType) -> String {
match graphql_type {
crate::types::GraphQLType::List(inner) => leaf_scalar_name(inner),
other => other.to_string(),
}
}
fn build_dynamic_schema(
generated: &GeneratedSchema,
_schema_cache: &SchemaCache,
subscription_fields: Option<&[SubField]>,
max_rows: Option<i64>,
names: Arc<crate::names::NameOverrides>,
subscription_refresh: std::time::Duration,
role: Option<&str>,
) -> Result<Schema, GraphQLError> {
let mut object_types: HashMap<String, Object> = HashMap::new();
for (type_name, obj) in &generated.object_types {
if obj.fields.is_empty() {
continue;
}
let relationships = generated
.relationship_fields
.get(type_name)
.map(|r| r.as_slice())
.unwrap_or(&[]);
let table_obj = create_object_type(
obj,
relationships,
generated.federation_entities.get(type_name),
);
object_types.insert(type_name.clone(), table_obj);
}
let countable: HashSet<&str> = generated
.query_fields
.iter()
.filter(|field| field.aggregates && !field.is_by_pk)
.map(|field| field.type_name.as_str())
.collect();
for (type_name, obj) in &generated.object_types {
if obj.fields.is_empty() && !countable.contains(type_name.as_str()) {
continue;
}
for aggregate_type in create_aggregate_types(&obj.local_name, obj) {
object_types.insert(aggregate_type.type_name().to_string(), aggregate_type);
}
}
let mutable: HashMap<String, String> = generated
.mutation_fields
.iter()
.filter_map(|f| {
let response = f
.return_type
.trim_matches(|c| c == '[' || c == ']' || c == '!')
.strip_suffix("_mutation_response")?;
Some((response.to_string(), f.type_name.clone()))
})
.collect();
for (base_name, row_type_name) in mutable {
let returning = generated
.object_types
.get(&row_type_name)
.is_some_and(|object| !object.fields.is_empty());
object_types.insert(
mutation_response_type_name(&base_name),
create_mutation_response_type(&base_name, &row_type_name, returning),
);
}
let query_root = if generated.enable_federation {
"Query"
} else {
"query_root"
};
let mutation_root = if generated.enable_federation {
"Mutation"
} else {
"mutation_root"
};
let subscription_root = if generated.enable_federation {
"Subscription"
} else {
"subscription_root"
};
let relationships = Arc::new(generated.relationship_fields.clone());
let type_names: Arc<HashMap<(String, String), String>> = Arc::new(
generated
.object_types
.iter()
.map(|(type_name, object)| {
(
(object.table.schema.clone(), object.table.name.clone()),
type_name.clone(),
)
})
.collect(),
);
let query = add_function_fields(
create_query_type(
generated,
query_root,
max_rows,
Arc::clone(&relationships),
Arc::clone(&names),
),
generated,
false,
max_rows,
Arc::clone(&relationships),
Arc::clone(&names),
);
let mutation = if !generated.mutation_fields.is_empty()
|| generated.function_fields.iter().any(|f| f.volatile)
{
Some(add_function_fields(
create_mutation_type(
generated,
mutation_root,
Arc::clone(&relationships),
Arc::clone(&type_names),
Arc::clone(&names),
max_rows,
role,
),
generated,
true,
max_rows,
Arc::clone(&relationships),
Arc::clone(&names),
))
} else {
None
};
let subscription = subscription_fields.map(|_| {
create_subscription_type(
generated,
subscription_root,
max_rows,
Arc::clone(&relationships),
Arc::clone(&names),
subscription_refresh,
)
});
let mut builder = Schema::build(
query_root,
mutation.as_ref().map(|_| mutation_root),
subscription.as_ref().map(|_| subscription_root),
);
if generated.enable_federation {
builder = builder
.enable_federation()
.extension(NullableFederationEntity);
let entity_state = Arc::new(EntityResolverState::new(
generated,
Arc::clone(&relationships),
Arc::clone(&names),
max_rows,
));
builder = builder.entity_resolver(move |ctx| {
let state = Arc::clone(&entity_state);
FieldFuture::new(async move { resolve_entities(&ctx, state.as_ref()).await })
});
}
for (_, obj) in object_types {
builder = builder.register(obj);
}
builder = builder.register(query);
if let Some(mutation) = mutation {
builder = builder.register(mutation);
}
if let Some(subscription) = subscription {
builder = builder.register(subscription);
}
let (bool_exp_inputs, bool_exp_scalars) = crate::input::bool_exp::build_inputs(
&generated.object_types,
&generated.relationship_fields,
);
for input in bool_exp_inputs {
builder = builder.register(input);
}
let mut scalar_names: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
for object in generated.object_types.values() {
for field in object.fields.iter().chain(&object.writable_fields) {
scalar_names.insert(leaf_scalar_name(&field.graphql_type));
}
}
scalar_names.insert("JSON".to_string());
for function in &generated.function_fields {
for (_, pg_type, _) in &function.arguments {
scalar_names.insert(leaf_scalar_name(&crate::types::pg_type_to_graphql(pg_type)));
}
}
for object in generated.object_types.values() {
for field in &object.fields {
for (_, pg_type, _) in &field.arguments {
scalar_names.insert(leaf_scalar_name(&crate::types::pg_type_to_graphql(pg_type)));
}
}
}
for relationships in generated.relationship_fields.values() {
for relationship in relationships {
for (_, pg_type, _) in &relationship.arguments {
scalar_names.insert(leaf_scalar_name(&crate::types::pg_type_to_graphql(pg_type)));
}
}
}
scalar_names.extend(bool_exp_scalars);
for name in scalar_names {
if matches!(name.as_str(), "Int" | "Float" | "String" | "Boolean" | "ID") {
continue;
}
if generated.enum_types.contains_key(&name) {
continue;
}
builder = builder
.register(Scalar::new(&name).description(format!("The PostgreSQL `{}` type.", name)));
}
let mut key_inputs: HashMap<String, InputObject> = HashMap::new();
for field in &generated.mutation_fields {
if field.mutation_type != MutationType::UpdateByPk || field.pk_columns.is_empty() {
continue;
}
let base_name = field.local_type_name.as_str();
let type_name = format!("{}_pk_columns_input", base_name);
let mut input = InputObject::new(&type_name)
.description(format!("The primary key of one {} row.", base_name));
for (column, pg_type) in &field.pk_columns {
input = input.field(InputValue::new(
names
.column(&field.schema_name, &field.table_name, column)
.unwrap_or(column),
TypeRef::named_nn(exposed_pk_argument_type(
names.as_ref(),
&field.schema_name,
&field.table_name,
column,
pg_type,
)),
));
}
key_inputs.insert(type_name, input);
}
for (_, input) in key_inputs {
builder = builder.register(input);
}
let insertable_types: HashSet<&str> = generated
.object_types
.iter()
.filter(|(_, object)| crate::input::mutation::is_insertable(&object.table))
.map(|(type_name, _)| type_name.as_str())
.collect();
for (type_name, object) in &generated.object_types {
let local_type_name = object.local_name.as_str();
let table = &object.table;
let writable_named = |allowed: Option<&crate::names::ColumnSet>| {
object
.writable_fields
.iter()
.filter(|field| {
let column = table_column_for(&names, table, &field.name);
table.get_column(column).is_some()
&& allowed.is_none_or(|set| set.allows(column))
})
.collect::<Vec<&crate::schema::object::GraphQLField>>()
};
let writable = writable_named(None);
let insert_fields = write_fields(object, &names, role, crate::role::Verb::Insert);
let update_fields = write_fields(object, &names, role, crate::role::Verb::Update);
if writable.is_empty() {
continue;
}
let relationships = generated
.relationship_fields
.get(type_name)
.map(|r| r.as_slice())
.unwrap_or(&[]);
let conflict_type = match table.unique_constraints.is_empty() {
true => None,
false => Some(format!("{}_on_conflict", local_type_name)),
};
if crate::input::mutation::is_insertable(table) {
let mut insert = InputObject::new(format!("{}_insert_input", local_type_name))
.description(format!("The columns of a new {} row.", local_type_name));
let mut taken: HashSet<&str> = HashSet::new();
for field in &insert_fields {
taken.insert(field.name.as_str());
insert = insert.field(InputValue::new(
&field.name,
write_type_ref(&field.graphql_type),
));
}
for relationship in relationships {
if !insertable_types.contains(relationship.target_type.as_str()) {
continue;
}
if !taken.insert(relationship.name.as_str()) {
continue;
}
let suffix = match relationship.is_list {
true => "arr_rel_insert_input",
false => "obj_rel_insert_input",
};
insert = insert.field(InputValue::new(
&relationship.name,
TypeRef::named(format!("{}_{}", relationship.target_local_name, suffix)),
));
}
builder = builder.register(insert);
let data_type = format!("{}_insert_input", local_type_name);
let mut object_rel =
InputObject::new(format!("{}_obj_rel_insert_input", local_type_name))
.description(format!(
"One {} row written beside its parent.",
local_type_name
))
.field(InputValue::new("data", TypeRef::named_nn(&data_type)));
let mut array_rel =
InputObject::new(format!("{}_arr_rel_insert_input", local_type_name))
.description(format!(
"{} rows written beside their parent.",
local_type_name
))
.field(InputValue::new(
"data",
TypeRef::named_nn_list_nn(&data_type),
));
if let Some(conflict) = &conflict_type {
object_rel =
object_rel.field(InputValue::new("on_conflict", TypeRef::named(conflict)));
array_rel =
array_rel.field(InputValue::new("on_conflict", TypeRef::named(conflict)));
}
builder = builder.register(object_rel).register(array_rel);
}
if crate::input::mutation::is_updatable(table) {
let mut set = InputObject::new(format!("{}_set_input", local_type_name))
.description(format!("Columns of {} to replace.", local_type_name));
let mut numeric = InputObject::new(format!("{}_inc_input", local_type_name))
.description(format!("Columns of {} to add to.", local_type_name));
let mut any_numeric = false;
let mut any_jsonb = false;
for field in &update_fields {
if matches!(&field.graphql_type, crate::types::GraphQLType::Json) {
any_jsonb = true;
}
set = set.field(InputValue::new(
&field.name,
write_type_ref(&field.graphql_type),
));
if crate::schema::aggregate::is_incrementable(&field.graphql_type) {
any_numeric = true;
numeric = numeric.field(InputValue::new(
&field.name,
TypeRef::named(leaf_scalar_name(&field.graphql_type)),
));
}
}
builder = builder.register(set);
if any_numeric {
builder = builder.register(numeric);
}
let mut updates = InputObject::new(format!("{}_updates", local_type_name))
.description(format!(
"One update to {}: which rows, and what to write.",
local_type_name
))
.field(InputValue::new(
"where",
TypeRef::named_nn(crate::input::bool_exp::bool_exp_type_name(local_type_name)),
))
.field(InputValue::new(
"_set",
TypeRef::named(format!("{}_set_input", local_type_name)),
));
if any_numeric {
updates = updates.field(InputValue::new(
"_inc",
TypeRef::named(format!("{}_inc_input", local_type_name)),
));
}
for (operator, _) in JSONB_OPERATORS {
if !any_jsonb {
break;
}
updates = updates.field(InputValue::new(
*operator,
TypeRef::named(jsonb_operator_input(local_type_name, operator)),
));
}
builder = builder.register(updates);
if any_jsonb {
for (operator, item) in JSONB_OPERATORS {
let mut input =
InputObject::new(jsonb_operator_input(local_type_name, operator))
.description(format!(
"Columns of {} to apply `{}` to.",
local_type_name, operator
));
for field in &update_fields {
if !matches!(&field.graphql_type, crate::types::GraphQLType::Json) {
continue;
}
input = input.field(InputValue::new(&field.name, item()));
}
builder = builder.register(input);
}
}
}
}
for function in &generated.function_fields {
if function.arguments.is_empty() {
continue;
}
let mut args = InputObject::new(format!("{}_args", function.name))
.description(format!("Arguments to {}.", function.name));
for (name, pg_type, _) in &function.arguments {
let scalar = crate::types::pg_type_to_graphql(pg_type).to_string();
args = args.field(InputValue::new(name, TypeRef::named(scalar)));
}
builder = builder.register(args);
}
for (type_name, relationships) in &generated.relationship_fields {
let local_type_name = generated
.object_types
.get(type_name)
.map(|object| object.local_name.as_str())
.unwrap_or(type_name);
for relationship in relationships {
if relationship.arguments.is_empty() {
continue;
}
let mut args =
InputObject::new(computed_args_type_name(local_type_name, &relationship.name))
.description(format!(
"Arguments to {}, beside the row it is asked of.",
relationship.name
));
for (name, pg_type, _) in &relationship.arguments {
let scalar = crate::types::pg_type_to_graphql(pg_type).to_string();
args = args.field(InputValue::new(name, TypeRef::named(scalar)));
}
builder = builder.register(args);
}
}
for object in generated.object_types.values() {
let local_type_name = object.local_name.as_str();
for field in &object.fields {
if field.arguments.is_empty() {
continue;
}
let mut args = InputObject::new(computed_args_type_name(local_type_name, &field.name))
.description(format!(
"Arguments to {}, beside the row it is asked of.",
field.name
));
for (name, pg_type, _) in &field.arguments {
let scalar = crate::types::pg_type_to_graphql(pg_type).to_string();
args = args.field(InputValue::new(name, TypeRef::named(scalar)));
}
builder = builder.register(args);
}
}
{
use crate::input::bool_exp as be;
let mut targets: BTreeMap<&str, &str> = BTreeMap::new();
for relationships in generated.relationship_fields.values() {
for relationship in relationships {
if relationship.is_list {
targets.insert(
relationship.target_local_name.as_str(),
relationship.target_type.as_str(),
);
}
}
}
for (target, target_type) in targets {
let Some(object) = generated.object_types.get(target_type) else {
continue;
};
let booleans: Vec<&str> = object
.fields
.iter()
.filter(|field| {
object.table.get_column(&field.name).is_some()
&& matches!(field.graphql_type, crate::types::GraphQLType::Boolean)
})
.map(|field| field.name.as_str())
.collect();
let mut over = InputObject::new(be::aggregate_bool_exp_type_name(target)).description(
format!("A question about the whole set of related {} rows.", target),
);
for function in be::AGGREGATE_PREDICATES {
if function != "count" && booleans.is_empty() {
continue;
}
over = over.field(InputValue::new(
function,
TypeRef::named(be::aggregate_bool_exp_function(target, function)),
));
let (arguments, predicate) = match function {
"count" => (
TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
target,
)),
be::comparison_type_name("Int"),
),
_ => (
TypeRef::named_nn(be::aggregate_bool_exp_columns(target, function)),
be::comparison_type_name("Boolean"),
),
};
let mut input = InputObject::new(be::aggregate_bool_exp_function(target, function))
.description(format!(
"`{}` over the related {} rows, and what it has to be.",
function, target
))
.field(InputValue::new("arguments", arguments))
.field(InputValue::new("distinct", TypeRef::named("Boolean")))
.field(InputValue::new(
"filter",
TypeRef::named(be::bool_exp_type_name(target)),
))
.field(InputValue::new("predicate", TypeRef::named_nn(predicate)));
if function == "count" {
input = input.description(format!(
"How many related {} rows there have to be.",
target
));
}
builder = builder.register(input);
if function != "count" {
let mut columns = Enum::new(be::aggregate_bool_exp_columns(target, function))
.description(format!("A boolean column of {}.", target));
for name in &booleans {
columns = columns.item(EnumItem::new(*name));
}
builder = builder.register(columns);
}
}
builder = builder.register(over);
}
}
for object in generated.object_types.values() {
let local_type_name = object.local_name.as_str();
let updatable_columns = write_fields(object, &names, role, crate::role::Verb::Update);
if object.table.unique_constraints.is_empty() {
continue;
}
let mut constraints =
Enum::new(format!("{}_constraint", local_type_name)).description(format!(
"A uniqueness of {} that an insert may conflict with.",
local_type_name
));
for (name, columns) in &object.table.unique_constraints {
constraints = constraints
.item(EnumItem::new(name).description(format!("unique ({})", columns.join(", "))));
}
let mut updatable = Enum::new(format!("{}_update_column", local_type_name)).description(
format!("A column of {} an upsert may write.", local_type_name),
);
for field in &updatable_columns {
updatable = updatable.item(EnumItem::new(&field.name));
}
if updatable_columns.is_empty() {
updatable = updatable.item(
EnumItem::new(PLACEHOLDER_COLUMN)
.description("No column may be written by this upsert."),
);
}
let on_conflict = InputObject::new(format!("{}_on_conflict", local_type_name))
.description(format!(
"What to do when an insert into {} conflicts.",
local_type_name
))
.field(InputValue::new(
"constraint",
TypeRef::named_nn(format!("{}_constraint", local_type_name)),
))
.field(
InputValue::new(
"update_columns",
TypeRef::named_nn_list_nn(format!("{}_update_column", local_type_name)),
)
.default_value(Value::List(Vec::new())),
)
.field(InputValue::new(
"where",
TypeRef::named(crate::input::bool_exp::bool_exp_type_name(local_type_name)),
));
builder = builder
.register(constraints)
.register(updatable)
.register(on_conflict);
}
for (type_name, members) in &generated.enum_types {
let mut generated_enum =
Enum::new(type_name).description(format!("The values {} allows.", type_name));
for (value, comment) in members {
let item = EnumItem::new(value);
generated_enum = generated_enum.item(match comment {
Some(description) => item.description(description),
None => item,
});
}
builder = builder.register(generated_enum);
}
let (order_inputs, order_enums) = crate::input::order_by::build_inputs(
&generated.object_types,
&generated.relationship_fields,
);
for input in order_inputs {
builder = builder.register(input);
}
for enum_type in order_enums {
builder = builder.register(enum_type);
}
builder
.finish()
.map_err(|e| GraphQLError::SchemaError(e.to_string()))
}
const PLACEHOLDER_COLUMN: &str = "_PLACEHOLDER";
pub fn mutation_response_type_name(base_name: &str) -> String {
format!("{}_mutation_response", base_name)
}
fn create_mutation_response_type(base_name: &str, row_type: &str, returning: bool) -> Object {
let response_name = mutation_response_type_name(base_name);
let response = Object::new(&response_name)
.description(format!("The rows {} changed, and how many.", base_name))
.field(Field::new(
"affected_rows",
TypeRef::named_nn(TypeRef::INT),
|ctx| {
FieldFuture::new(async move {
if let Some(Value::Object(map)) = ctx.parent_value.as_value() {
if let Some(value) = map.get(&async_graphql::Name::new("affected_rows")) {
return Ok(Some(FieldValue::value(value.clone())));
}
}
Ok(Some(FieldValue::value(Value::from(0))))
})
},
));
if !returning {
return response;
}
response.field(Field::new(
"returning",
TypeRef::named_nn_list_nn(row_type.to_string()),
|ctx| {
FieldFuture::new(async move {
let rows = match ctx.parent_value.as_value() {
Some(Value::Object(map)) => {
match map.get(&async_graphql::Name::new("returning")) {
Some(Value::List(items)) => items.clone(),
_ => Vec::new(),
}
}
_ => Vec::new(),
};
Ok(Some(FieldValue::list(
rows.into_iter().map(FieldValue::value),
)))
})
},
))
}
fn create_aggregate_types(base_name: &str, object: &TableObjectType) -> Vec<Object> {
use crate::schema::aggregate as agg;
let mut types = Vec::new();
let fields_type = agg::aggregate_fields_type_name(base_name);
let has_rows = !object.fields.is_empty();
let mut over = Object::new(agg::aggregate_type_name(base_name))
.description(match has_rows {
true => format!("Aggregates over {}, with the rows themselves.", base_name),
false => format!("Aggregates over {}.", base_name),
})
.field(Field::new(
"aggregate",
TypeRef::named(&fields_type),
|ctx| FieldFuture::new(async move { Ok(child_of(&ctx, "aggregate")) }),
));
if has_rows {
over = over.field(Field::new(
"nodes",
TypeRef::named_nn_list_nn(object.name.clone()),
|ctx| {
FieldFuture::new(async move {
let rows = match child_value(&ctx, "nodes") {
Some(Value::List(items)) => items,
_ => Vec::new(),
};
Ok(Some(FieldValue::list(
rows.into_iter().map(FieldValue::value),
)))
})
},
));
}
types.push(over);
let mut count = Field::new("count", TypeRef::named_nn(TypeRef::INT), |ctx| {
let key = ctx.ctx.field().alias().unwrap_or("count").to_string();
FieldFuture::new(async move {
Ok(Some(FieldValue::value(
child_value(&ctx, &key).unwrap_or(Value::from(0)),
)))
})
});
if has_rows {
count = count
.argument(InputValue::new(
"columns",
TypeRef::named_nn_list(agg_select_column(base_name)),
))
.argument(InputValue::new("distinct", TypeRef::named("Boolean")));
}
let mut aggregate_fields = Object::new(&fields_type)
.description(format!("Aggregate functions over {}.", base_name))
.field(count);
for (function, returns, columns) in agg::functions_for(object) {
let function_type = agg::function_fields_type_name(base_name, function);
let mut per_column = Object::new(&function_type).description(format!(
"`{}` of each {} column it applies to.",
function, base_name
));
for column in &columns {
let column_name = column.clone();
let type_name = agg::field_type_for(object, column, returns);
per_column =
per_column.field(Field::new(column, TypeRef::named(type_name), move |ctx| {
let column_name = column_name.clone();
FieldFuture::new(async move {
Ok(child_value(&ctx, &column_name).map(FieldValue::value))
})
}));
}
types.push(per_column);
let owned = function.to_string();
aggregate_fields = aggregate_fields.field(Field::new(
function,
TypeRef::named(&function_type),
move |ctx| {
let owned = owned.clone();
FieldFuture::new(async move { Ok(child_of(&ctx, &owned)) })
},
));
}
types.push(aggregate_fields);
types
}
fn agg_select_column(base_name: &str) -> String {
crate::input::order_by::select_column_type_name(base_name)
}
fn child_value(ctx: &ResolverContext<'_>, key: &str) -> Option<Value> {
match ctx.parent_value.as_value() {
Some(Value::Object(map)) => map.get(&async_graphql::Name::new(key)).cloned(),
_ => None,
}
}
fn child_of<'a>(ctx: &ResolverContext<'_>, key: &str) -> Option<FieldValue<'a>> {
match child_value(ctx, key) {
None | Some(Value::Null) => None,
Some(value) => Some(FieldValue::value(value)),
}
}
fn coded_error(code: &'static str, message: impl Into<String>) -> async_graphql::Error {
let mut error = async_graphql::Error::new(message);
let mut extensions = async_graphql::ErrorExtensionValues::default();
extensions.set("code", code);
error.extensions = Some(extensions);
error
}
fn validation_error(message: impl Into<String>) -> async_graphql::Error {
coded_error("validation-failed", message)
}
fn database_error(error: sqlx::Error) -> async_graphql::Error {
let sqlx::Error::Database(db) = &error else {
return async_graphql::Error::new(error.to_string());
};
let described = match db.code().as_deref() {
Some("23505") => "Uniqueness violation. ",
Some("23502") => "Not-NULL violation. ",
Some("23503") => "Foreign key violation. ",
Some("23514") => "Check constraint violation. ",
_ => "",
};
let error = async_graphql::Error::new(format!("{}{}", described, db.message()));
let coded = match db.code().as_deref() {
Some(code) if code.starts_with("23") => Some("constraint-violation"),
Some("22025") => Some("bad-request"),
Some(code) if code.starts_with("22") => Some("data-exception"),
_ => None,
};
match coded {
Some(code) => at_code(error, code),
None => error,
}
}
fn at_code(mut error: async_graphql::Error, code: &'static str) -> async_graphql::Error {
let mut extensions = error.extensions.take().unwrap_or_default();
extensions.set("code", code);
error.extensions = Some(extensions);
error
}
fn at_path(mut error: async_graphql::Error, path: &str) -> async_graphql::Error {
let mut extensions = error.extensions.take().unwrap_or_default();
extensions.set("path", path);
error.extensions = Some(extensions);
error
}
fn create_object_type(
obj: &TableObjectType,
relationships: &[RelationshipField],
federation_entity: Option<&FederationEntity>,
) -> Object {
let mut object = Object::new(&obj.name);
let entity_is_shared = federation_entity.is_some_and(|entity| entity.shared);
if let Some(entity) = federation_entity {
object = object.key(entity.key_fields_sdl());
}
object = match obj.description() {
Some("") => object,
Some(desc) => object.description(desc),
None => object.description(format!(
"columns and relationships of \"{}\"",
obj.table_name()
)),
};
let mut taken: HashSet<String> = obj.fields.iter().map(|f| f.name.clone()).collect();
let mut fields: Vec<(String, Field)> = Vec::new();
for field in &obj.fields {
let field_name = field.name.clone();
let field_type = graphql_type_ref(&field.type_string());
let takes_path = matches!(&field.graphql_type, crate::types::GraphQLType::Json)
|| matches!(
&field.graphql_type,
crate::types::GraphQLType::Custom(name) if name == "json"
);
let gql_field = Field::new(&field.name, field_type, move |ctx| {
let field_name = field_name.clone();
FieldFuture::new(async move {
if let Some(Value::Object(map)) = ctx.parent_value.as_value() {
let key = async_graphql::Name::new(&field_name);
let path = match takes_path {
true => ctx.args.get("path").and_then(|v| v.string().ok()),
false => None,
};
match map.get(&key) {
Some(Value::Null) => return Ok(None),
Some(val) => {
let val = match path {
Some(path) => walk_json_path(val, path)?,
None => val.clone(),
};
return match val {
Value::Null => Ok(None),
val => Ok(Some(FieldValue::value(val))),
};
}
None => {}
}
}
Ok(None)
})
});
let gql_field = match takes_path {
true => gql_field.argument(
InputValue::new("path", TypeRef::named("String")).description("JSON select path"),
),
false => gql_field,
};
let gql_field = match field.arguments.is_empty() {
true => gql_field,
false => gql_field.argument(InputValue::new(
"args",
TypeRef::named_nn(computed_args_type_name(&obj.local_name, &field.name)),
)),
};
let gql_field = if let Some(desc) = &field.description {
gql_field.description(desc)
} else {
gql_field
};
let gql_field = if entity_is_shared && !field.is_pk {
gql_field.shareable()
} else {
gql_field
};
fields.push((field.name.clone(), gql_field));
}
for rel in relationships {
if !taken.insert(rel.name.clone()) {
tracing::warn!(
"{}: not exposing the relationship to \"{}\" as `{}` -- the table \
already has a column of that name",
obj.name,
rel.target_type,
rel.name
);
continue;
}
let field_name = rel.name.clone();
let field_type = if rel.is_list {
TypeRef::named_nn_list_nn(&rel.target_type)
} else if always_present(rel, &obj.table) {
TypeRef::named_nn(&rel.target_type)
} else {
TypeRef::named(&rel.target_type)
};
let is_list = rel.is_list;
let mut gql_field = Field::new(&rel.name, field_type, move |ctx| {
let field_name = field_name.clone();
FieldFuture::new(async move {
if let Some(Value::Object(map)) = ctx.parent_value.as_value() {
let key = async_graphql::Name::new(&field_name);
match map.get(&key) {
Some(Value::Null) if !is_list => return Ok(None),
Some(val) => return Ok(Some(FieldValue::value(val.clone()))),
None => {}
}
}
Ok(None)
})
});
if !rel.arguments.is_empty() {
gql_field = gql_field.argument(InputValue::new(
"args",
TypeRef::named_nn(computed_args_type_name(&obj.local_name, &rel.name)),
));
}
if rel.is_list {
gql_field = gql_field
.argument(InputValue::new(
"distinct_on",
TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
&rel.target_local_name,
)),
))
.argument(InputValue::new(
"where",
TypeRef::named(crate::input::bool_exp::bool_exp_type_name(
&rel.target_local_name,
)),
))
.argument(InputValue::new(
"order_by",
TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(
&rel.target_local_name,
)),
))
.argument(InputValue::new("limit", TypeRef::named("Int")))
.argument(InputValue::new("offset", TypeRef::named("Int")));
}
let gql_field = if let Some(desc) = &rel.description {
gql_field.description(desc)
} else {
gql_field
};
let gql_field = if entity_is_shared {
gql_field.shareable()
} else {
gql_field
};
fields.push((rel.name.clone(), gql_field));
if rel.is_list {
let aggregate_field = format!("{}_aggregate", rel.name);
if taken.insert(aggregate_field.clone()) {
let key = aggregate_field.clone();
fields.push((
aggregate_field.clone(),
Field::new(
&aggregate_field,
TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
&rel.target_local_name,
)),
move |ctx| {
let key = key.clone();
FieldFuture::new(async move {
Ok(Some(FieldValue::value(
child_value(&ctx, &key).unwrap_or(Value::Null),
)))
})
},
)
.argument(InputValue::new(
"distinct_on",
TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
&rel.target_local_name,
)),
))
.argument(InputValue::new(
"where",
TypeRef::named(crate::input::bool_exp::bool_exp_type_name(
&rel.target_local_name,
)),
))
.argument(InputValue::new(
"order_by",
TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(
&rel.target_local_name,
)),
))
.argument(InputValue::new("limit", TypeRef::named("Int")))
.argument(InputValue::new("offset", TypeRef::named("Int")))
.description(format!("Aggregates over {}.", rel.name)),
));
if !rel.arguments.is_empty() {
let last = fields.len() - 1;
let (name, field) = fields.remove(last);
fields.push((
name,
field.argument(InputValue::new(
"args",
TypeRef::named_nn(computed_args_type_name(&obj.local_name, &rel.name)),
)),
));
}
if entity_is_shared {
let last = fields.len() - 1;
let (name, field) = fields.remove(last);
fields.push((name, field.shareable()));
}
}
}
}
fields.sort_by(|(a, _), (b, _)| a.cmp(b));
for (_, field) in fields {
object = object.field(field);
}
object
}
trait RootField: Sized {
fn with_argument(self, value: InputValue) -> Self;
}
impl RootField for Field {
fn with_argument(self, value: InputValue) -> Self {
self.argument(value)
}
}
impl RootField for SubscriptionField {
fn with_argument(self, value: InputValue) -> Self {
self.argument(value)
}
}
fn with_row_arguments<F: RootField>(field: F, type_name: &str) -> F {
with_row_arguments_named(field, type_name, true)
}
fn with_row_arguments_named<F: RootField>(field: F, type_name: &str, columns: bool) -> F {
let mut field = field;
if columns {
field = field.with_argument(InputValue::new(
"distinct_on",
TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(type_name)),
));
}
field = field
.with_argument(InputValue::new("limit", TypeRef::named("Int")))
.with_argument(InputValue::new("offset", TypeRef::named("Int")));
if columns {
field = field.with_argument(InputValue::new(
"order_by",
TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(type_name)),
));
}
field.with_argument(InputValue::new(
"where",
TypeRef::named(crate::input::bool_exp::bool_exp_type_name(type_name)),
))
}
fn with_key_arguments<F: RootField>(
field: F,
schema_name: &str,
table_name: &str,
pk_columns: &[(String, String)],
names: &crate::names::NameOverrides,
) -> F {
let mut field = field;
for (column, pg_type) in pk_columns {
field = field.with_argument(InputValue::new(
names
.column(schema_name, table_name, column)
.unwrap_or(column),
TypeRef::named_nn(exposed_pk_argument_type(
names,
schema_name,
table_name,
column,
pg_type,
)),
));
}
field
}
fn create_query_type(
generated: &GeneratedSchema,
root_name: &str,
max_rows: Option<i64>,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
names: Arc<crate::names::NameOverrides>,
) -> Object {
let mut roots: Vec<(String, Field)> = Vec::new();
for field in &generated.query_fields {
let table_name = field.table_name.clone();
let schema_name = field.schema_name.clone();
let type_name = field.type_name.clone();
let local_type_name = field.local_type_name.clone();
let is_by_pk = field.is_by_pk;
let pk_columns = field.pk_columns.clone();
let return_type = graphql_type_ref(&field.return_type);
let spec_type_name = local_type_name.clone();
let spec = Arc::new(QueryFieldSpec {
schema_name,
table_name,
type_name,
is_by_pk,
pk_columns: pk_columns.clone(),
max_rows,
relationships: Arc::clone(&relationships),
names: Arc::clone(&names),
call: None,
});
let mut gql_field = Field::new(&field.name, return_type, move |ctx| {
let spec = Arc::clone(&spec);
FieldFuture::new(async move { resolve_query(&ctx, &spec).await })
});
gql_field = match is_by_pk {
false => with_row_arguments(gql_field, &spec_type_name),
true => with_key_arguments(
gql_field,
&field.schema_name,
&field.table_name,
&pk_columns,
names.as_ref(),
),
};
if let Some(desc) = &field.description {
gql_field = gql_field.description(desc);
}
if field.rows {
roots.push((field.name.clone(), gql_field));
}
if !is_by_pk && field.aggregates {
let agg_spec = Arc::new(AggregateSpec {
schema_name: field.schema_name.clone(),
table_name: field.table_name.clone(),
type_name: field.type_name.clone(),
max_rows,
relationships: Arc::clone(&relationships),
names: Arc::clone(&names),
call: None,
});
let aggregate_field_name = field.aggregate_name.clone().unwrap_or_else(|| {
crate::schema::aggregate::aggregate_type_name(&field.local_type_name)
});
let aggregate_field_name_for_sorting = aggregate_field_name.clone();
let mut agg_field = Field::new(
aggregate_field_name,
TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
&field.local_type_name,
)),
move |ctx| {
let agg_spec = Arc::clone(&agg_spec);
FieldFuture::new(async move { resolve_aggregate(&ctx, &agg_spec).await })
},
);
agg_field = with_row_arguments_named(agg_field, &spec_type_name, field.rows);
agg_field = match field.aggregate_description.as_deref() {
Some("") => agg_field,
Some(given) => agg_field.description(given),
None => agg_field.description(format!(
"fetch aggregated fields from the table: \"{}\"",
field.table_name
)),
};
roots.push((aggregate_field_name_for_sorting, agg_field));
}
}
roots.sort_by(|(a, _), (b, _)| a.cmp(b));
let mut query = Object::new(root_name);
if roots.is_empty() {
return query.field(
Field::new("no_queries_available", TypeRef::named("String"), |_| {
FieldFuture::new(async move { Ok(None::<Value>) })
})
.description("There are no queries available to the current role."),
);
}
for (_, field) in roots {
query = query.field(field);
}
query
}
type JsonbOperator = (&'static str, fn() -> TypeRef);
const JSONB_OPERATORS: &[JsonbOperator] = &[
("_append", || TypeRef::named("jsonb")),
("_delete_at_path", || TypeRef::named_nn_list("String")),
("_delete_elem", || TypeRef::named("Int")),
("_delete_key", || TypeRef::named("String")),
("_prepend", || TypeRef::named("jsonb")),
];
fn jsonb_operator_input(type_name: &str, operator: &str) -> String {
format!("{}{}_input", type_name, operator)
}
fn with_update_operators(
field: Field,
base_name: &str,
has_numeric: bool,
has_jsonb: bool,
) -> Field {
let mut field = field;
if has_jsonb {
for (operator, _) in JSONB_OPERATORS {
field = field.argument(InputValue::new(
*operator,
TypeRef::named(jsonb_operator_input(base_name, operator)),
));
}
}
if has_numeric {
field = field.argument(InputValue::new(
"_inc",
TypeRef::named(format!("{}_inc_input", base_name)),
));
}
field.argument(InputValue::new(
"_set",
TypeRef::named(format!("{}_set_input", base_name)),
))
}
fn write_fields<'a>(
object: &'a crate::schema::object::TableObjectType,
names: &crate::names::NameOverrides,
role: Option<&str>,
verb: crate::role::Verb,
) -> Vec<&'a crate::schema::object::GraphQLField> {
let allowed = role.and_then(|role| {
names
.permissions(&object.table.schema, &object.table.name, role)
.and_then(|granted| granted.write_columns(verb))
});
object
.writable_fields
.iter()
.filter(|field| {
let column = table_column_for(names, &object.table, &field.name);
object
.table
.get_column(column)
.is_some_and(|c| !c.always_generated)
&& allowed.is_none_or(|set| set.allows(column))
})
.collect()
}
fn update_inputs(
object: &crate::schema::object::TableObjectType,
names: &crate::names::NameOverrides,
role: Option<&str>,
) -> (bool, bool) {
let fields = write_fields(object, names, role, crate::role::Verb::Update);
(
fields
.iter()
.any(|f| crate::schema::aggregate::is_incrementable(&f.graphql_type)),
fields
.iter()
.any(|f| matches!(&f.graphql_type, crate::types::GraphQLType::Json)),
)
}
fn create_mutation_type(
generated: &GeneratedSchema,
root_name: &str,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
type_names: Arc<HashMap<(String, String), String>>,
names: Arc<crate::names::NameOverrides>,
max_rows: Option<i64>,
role: Option<&str>,
) -> Object {
let mut mutation = Object::new(root_name);
let mut roots: Vec<(String, Field)> = Vec::new();
let mut has_numeric_column: HashSet<String> = HashSet::new();
let mut has_jsonb_column: HashSet<String> = HashSet::new();
for object in generated.object_types.values() {
let (numeric, jsonb) = update_inputs(object, &names, role);
if numeric {
has_numeric_column.insert(object.local_name.clone());
}
if jsonb {
has_jsonb_column.insert(object.local_name.clone());
}
}
let has_conflict_target: HashSet<String> = generated
.object_types
.iter()
.filter(|(_, object)| !object.table.unique_constraints.is_empty())
.map(|(_, object)| object.local_name.clone())
.collect();
for field in &generated.mutation_fields {
let table_name = field.table_name.clone();
let schema_name = field.schema_name.clone();
let mutation_type = field.mutation_type;
let pk_columns = field.pk_columns.clone();
let return_type = graphql_type_ref(&field.return_type);
let where_type = field.local_type_name.clone();
let resolver_pk_columns = pk_columns.clone();
let field_relationships = Arc::clone(&relationships);
let field_type_names = Arc::clone(&type_names);
let field_names = Arc::clone(&names);
let mut gql_field = Field::new(&field.name, return_type, move |ctx| {
let table_name = table_name.clone();
let schema_name = schema_name.clone();
let pk_columns = resolver_pk_columns.clone();
let relationships = Arc::clone(&field_relationships);
let type_names = Arc::clone(&field_type_names);
let names = Arc::clone(&field_names);
FieldFuture::new(async move {
resolve_mutation(
&ctx,
&schema_name,
&table_name,
mutation_type,
&pk_columns,
&relationships,
&type_names,
&names,
max_rows,
)
.await
})
});
match mutation_type {
MutationType::Insert | MutationType::InsertOne => {
let insert_input = format!("{}_insert_input", where_type);
gql_field = if mutation_type == MutationType::Insert {
gql_field.argument(InputValue::new(
"objects",
TypeRef::named_nn_list_nn(&insert_input),
))
} else {
gql_field.argument(InputValue::new("object", TypeRef::named_nn(&insert_input)))
};
if has_conflict_target.contains(&where_type) {
gql_field = gql_field.argument(InputValue::new(
"on_conflict",
TypeRef::named(format!("{}_on_conflict", where_type)),
));
}
}
MutationType::UpdateByPk => {
gql_field = gql_field.argument(InputValue::new(
"pk_columns",
TypeRef::named_nn(format!("{}_pk_columns_input", where_type)),
));
gql_field = with_update_operators(
gql_field,
&where_type,
has_numeric_column.contains(&where_type),
has_jsonb_column.contains(&where_type),
);
}
MutationType::Update => {
gql_field = gql_field.argument(InputValue::new(
"where",
TypeRef::named_nn(crate::input::bool_exp::bool_exp_type_name(&where_type)),
));
gql_field = with_update_operators(
gql_field,
&where_type,
has_numeric_column.contains(&where_type),
has_jsonb_column.contains(&where_type),
);
}
MutationType::UpdateMany => {
gql_field = gql_field.argument(InputValue::new(
"updates",
TypeRef::named_nn_list_nn(format!("{}_updates", where_type)),
));
}
MutationType::DeleteByPk => {
for (col_name, pg_type) in &pk_columns {
gql_field = gql_field.argument(InputValue::new(
names
.column(&field.schema_name, &field.table_name, col_name)
.unwrap_or(col_name),
TypeRef::named_nn(exposed_pk_argument_type(
names.as_ref(),
&field.schema_name,
&field.table_name,
col_name,
pg_type,
)),
));
}
}
MutationType::Delete => {
gql_field = gql_field.argument(InputValue::new(
"where",
TypeRef::named_nn(crate::input::bool_exp::bool_exp_type_name(&where_type)),
));
}
}
if let Some(desc) = &field.description {
gql_field = gql_field.description(desc);
}
roots.push((field.name.clone(), gql_field));
}
roots.sort_by(|(a, _), (b, _)| a.cmp(b));
for (_, field) in roots {
mutation = mutation.field(field);
}
mutation
}
fn create_subscription_type(
generated: &GeneratedSchema,
root_name: &str,
max_rows: Option<i64>,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
names: Arc<crate::names::NameOverrides>,
refresh: std::time::Duration,
) -> Subscription {
let mut roots: Vec<(String, SubscriptionField)> = Vec::new();
for field in &generated.query_fields {
let spec_type_name = field.local_type_name.clone();
let pk_columns = field.pk_columns.clone();
let is_by_pk = field.is_by_pk;
let return_type = graphql_type_ref(&field.return_type);
let watched = vec![crate::subscription::table_channel_name(
&field.schema_name,
&field.table_name,
)];
let spec = Arc::new(QueryFieldSpec {
schema_name: field.schema_name.clone(),
table_name: field.table_name.clone(),
type_name: field.type_name.clone(),
is_by_pk,
pk_columns: pk_columns.clone(),
max_rows,
relationships: Arc::clone(&relationships),
names: Arc::clone(&names),
call: None,
});
let mut gql_field = SubscriptionField::new(&field.name, return_type, move |ctx| {
let spec = Arc::clone(&spec);
let watched = watched.clone();
SubscriptionFieldFuture::new(async move {
let wake = wake_on(&ctx, &watched, refresh).await;
Ok(live(ctx, LiveSource::Rows(spec), wake))
})
});
gql_field = match is_by_pk {
false => with_row_arguments(gql_field, &spec_type_name),
true => with_key_arguments(
gql_field,
&field.schema_name,
&field.table_name,
&pk_columns,
names.as_ref(),
),
};
if let Some(desc) = &field.description {
gql_field = gql_field.description(desc);
}
if field.rows {
roots.push((field.name.clone(), gql_field));
}
if is_by_pk || !field.aggregates {
continue;
}
let agg_spec = Arc::new(AggregateSpec {
schema_name: field.schema_name.clone(),
table_name: field.table_name.clone(),
type_name: field.type_name.clone(),
max_rows,
relationships: Arc::clone(&relationships),
names: Arc::clone(&names),
call: None,
});
let agg_watched = vec![crate::subscription::table_channel_name(
&field.schema_name,
&field.table_name,
)];
let agg_name = field.aggregate_name.clone().unwrap_or_else(|| {
crate::schema::aggregate::aggregate_type_name(&field.local_type_name)
});
let mut agg_field = SubscriptionField::new(
agg_name.clone(),
TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
&field.local_type_name,
)),
move |ctx| {
let agg_spec = Arc::clone(&agg_spec);
let watched = agg_watched.clone();
SubscriptionFieldFuture::new(async move {
let wake = wake_on(&ctx, &watched, refresh).await;
Ok(live(ctx, LiveSource::Aggregate(agg_spec), wake))
})
},
);
agg_field = with_row_arguments_named(agg_field, &spec_type_name, field.rows);
agg_field = match field.aggregate_description.as_deref() {
Some("") => agg_field,
Some(given) => agg_field.description(given),
None => agg_field.description(format!(
"fetch aggregated fields from the table: \"{}\"",
field.table_name
)),
};
roots.push((agg_name, agg_field));
}
roots.sort_by(|(a, _), (b, _)| a.cmp(b));
let mut subscription = Subscription::new(root_name);
if roots.is_empty() {
return subscription.field(SubscriptionField::new(
"no_subscriptions_available",
TypeRef::named("String"),
|_| {
SubscriptionFieldFuture::new(async move {
Ok(futures::stream::empty::<
Result<FieldValue, async_graphql::Error>,
>())
})
},
));
}
for (_, field) in roots {
subscription = subscription.field(field);
}
subscription
}
async fn wake_on(
ctx: &ResolverContext<'_>,
channels: &[String],
refresh: std::time::Duration,
) -> futures::stream::BoxStream<'static, ()> {
use futures::StreamExt;
let mut streams: Vec<futures::stream::BoxStream<'static, ()>> =
Vec::with_capacity(channels.len() + 1);
streams.push(
tokio_stream::wrappers::IntervalStream::new({
let mut interval = tokio::time::interval(refresh);
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
interval
})
.map(|_| ())
.boxed(),
);
if let Ok(broker_arc) = ctx.data::<Arc<RwLock<Option<NotifyBroker>>>>() {
let guard = broker_arc.read().await;
if let Some(broker) = guard.as_ref() {
for channel in channels {
streams.push(
broker
.subscribe_or_create(channel)
.await
.map(|_| ())
.boxed(),
);
}
}
}
futures::stream::select_all(streams).boxed()
}
enum LiveSource {
Rows(Arc<QueryFieldSpec>),
Aggregate(Arc<AggregateSpec>),
}
async fn read_live<'a>(
ctx: &ResolverContext<'a>,
source: &LiveSource,
) -> Result<(serde_json::Value, Option<FieldValue<'a>>), async_graphql::Error> {
match source {
LiveSource::Rows(spec) => {
let rows = query_rows(ctx, spec).await?;
Ok((
serde_json::Value::Array(rows.clone()),
rows_as_field_value(rows, spec.is_by_pk),
))
}
LiveSource::Aggregate(spec) => {
let value = aggregate_value(ctx, spec).await?;
let seen = serde_json::to_value(&value).unwrap_or_default();
Ok((seen, Some(FieldValue::value(value))))
}
}
}
fn live<'a>(
ctx: ResolverContext<'a>,
source: LiveSource,
wake: futures::stream::BoxStream<'static, ()>,
) -> impl futures::Stream<Item = Result<FieldValue<'a>, async_graphql::Error>> + Send + 'a {
use futures::StreamExt;
struct Live<'a> {
ctx: ResolverContext<'a>,
source: LiveSource,
wake: futures::stream::BoxStream<'static, ()>,
sent: Option<serde_json::Value>,
started: bool,
failed: bool,
_active: crate::subscription::ActiveSubscription,
}
futures::stream::unfold(
Live {
ctx,
source,
wake,
sent: None,
started: false,
failed: false,
_active: crate::subscription::ActiveSubscription::open(),
},
|mut live: Live<'a>| async move {
if live.failed {
return None;
}
loop {
if live.started {
live.wake.next().await?;
} else {
live.started = true;
}
let read = read_live(&live.ctx, &live.source).await;
match read {
Err(e) => {
live.failed = true;
return Some((Err(e), live));
}
Ok((seen, value)) => {
if live.sent.as_ref() == Some(&seen) {
continue;
}
live.sent = Some(seen);
let value = value.unwrap_or_else(|| FieldValue::value(Value::Null));
return Some((Ok(value), live));
}
}
}
},
)
}
struct QueryFieldSpec {
schema_name: String,
table_name: String,
type_name: String,
is_by_pk: bool,
pk_columns: Vec<(String, String)>,
max_rows: Option<i64>,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
names: Arc<crate::names::NameOverrides>,
call: Option<Arc<FunctionCall>>,
}
#[derive(Clone)]
struct EntityLookup {
schema_name: String,
table_name: String,
type_name: String,
key_columns: Vec<EntityKeyColumn>,
row_column_types: HashMap<String, String>,
}
#[derive(Clone)]
struct EntityKeyColumn {
field_name: String,
column_name: String,
pg_type: String,
explicit_id: bool,
}
struct EntityResolverState {
entities: HashMap<String, EntityLookup>,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
names: Arc<crate::names::NameOverrides>,
max_rows: Option<i64>,
}
impl EntityResolverState {
fn new(
generated: &GeneratedSchema,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
names: Arc<crate::names::NameOverrides>,
max_rows: Option<i64>,
) -> Self {
let mut entities = HashMap::new();
for (type_name, entity) in &generated.federation_entities {
let Some(object) = generated.object_types.get(type_name) else {
continue;
};
let mut key_columns = Vec::with_capacity(entity.key_fields.len());
for field_name in &entity.key_fields {
let column_name = names
.column_source(&object.table.schema, &object.table.name, field_name)
.unwrap_or(field_name);
let Some(column) = object.table.get_column(column_name) else {
continue;
};
key_columns.push(EntityKeyColumn {
field_name: field_name.clone(),
column_name: column.name.clone(),
pg_type: column.nominal_type.clone(),
explicit_id: matches!(
names.column_type(&object.table.schema, &object.table.name, &column.name),
Some(crate::types::GraphQLType::Id)
),
});
}
if key_columns.len() != entity.key_fields.len() {
continue;
}
entities.insert(
type_name.clone(),
EntityLookup {
schema_name: object.table.schema.clone(),
table_name: object.table.name.clone(),
type_name: type_name.clone(),
key_columns,
row_column_types: exposed_column_types(&object.table, names.as_ref()),
},
);
}
Self {
entities,
relationships,
names,
max_rows,
}
}
}
struct EntityRepresentation {
index: usize,
type_name: String,
values: serde_json::Map<String, serde_json::Value>,
}
fn entity_field_applicability(
selection_set: &SelectionSet,
fragments: &HashMap<Name, Positioned<FragmentDefinition>>,
type_name: &str,
) -> Vec<bool> {
fn condition_matches(condition: Option<&Positioned<TypeCondition>>, type_name: &str) -> bool {
condition.is_none_or(|condition| {
let condition = condition.node.on.node.as_str();
condition == type_name || condition == "_Entity"
})
}
fn collect(
selection_set: &SelectionSet,
fragments: &HashMap<Name, Positioned<FragmentDefinition>>,
type_name: &str,
parent_matches: bool,
applicability: &mut Vec<bool>,
) {
for selection in &selection_set.items {
match &selection.node {
Selection::Field(_) => applicability.push(parent_matches),
Selection::FragmentSpread(spread) => {
let Some(fragment) = fragments.get(&spread.node.fragment_name.node) else {
continue;
};
collect(
&fragment.node.selection_set.node,
fragments,
type_name,
parent_matches
&& condition_matches(Some(&fragment.node.type_condition), type_name),
applicability,
);
}
Selection::InlineFragment(fragment) => collect(
&fragment.node.selection_set.node,
fragments,
type_name,
parent_matches
&& condition_matches(fragment.node.type_condition.as_ref(), type_name),
applicability,
),
}
}
}
let mut applicability = Vec::new();
collect(
selection_set,
fragments,
type_name,
true,
&mut applicability,
);
applicability
}
fn entity_selection_fields<'ctx, 'a>(
ctx: &'ctx ResolverContext<'a>,
type_name: &str,
) -> Vec<SelectionField<'ctx>> {
let applicability = entity_field_applicability(
&ctx.ctx.item.node.selection_set.node,
&ctx.ctx.query_env.fragments,
type_name,
);
let fields: Vec<_> = ctx.field().selection_set().collect();
debug_assert_eq!(fields.len(), applicability.len());
fields
.into_iter()
.zip(applicability)
.filter_map(|(field, applies)| applies.then_some(field))
.collect()
}
async fn resolve_entities<'a>(
ctx: &ResolverContext<'a>,
state: &EntityResolverState,
) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
let representations = ctx.args.try_get("representations")?.list()?;
let mut grouped: HashMap<String, Vec<EntityRepresentation>> = HashMap::new();
let mut count = 0;
for (index, representation) in representations.iter().enumerate() {
count += 1;
let value = accessor_to_json(&representation);
let serde_json::Value::Object(values) = value else {
return Err(async_graphql::Error::new(
"entity representation must be an object",
));
};
let type_name = values
.get("__typename")
.and_then(serde_json::Value::as_str)
.ok_or_else(|| {
async_graphql::Error::new("entity representation is missing __typename")
})?
.to_string();
grouped
.entry(type_name.clone())
.or_default()
.push(EntityRepresentation {
index,
type_name,
values,
});
}
let mut resolved: Vec<Option<FieldValue<'a>>> =
std::iter::repeat_with(|| None).take(count).collect();
for (type_name, representations) in grouped {
let lookup = state.entities.get(&type_name).ok_or_else(|| {
async_graphql::Error::new(format!("unknown federation entity \"{}\"", type_name))
})?;
let rows = resolve_entity_group(ctx, lookup, &representations, state).await?;
for (ordinality, representation) in representations.into_iter().enumerate() {
resolved[representation.index] = Some(match &rows[ordinality] {
Some(row) => FieldValue::value(json_to_value(row.clone()))
.with_type(representation.type_name),
None => FieldValue::NULL,
});
}
}
Ok(Some(FieldValue::list(resolved.into_iter().map(|value| {
value.expect("every representation was resolved or errored")
}))))
}
async fn resolve_entity_group(
ctx: &ResolverContext<'_>,
lookup: &EntityLookup,
representations: &[EntityRepresentation],
state: &EntityResolverState,
) -> Result<Vec<Option<serde_json::Value>>, async_graphql::Error> {
if representations.is_empty() {
return Ok(Vec::new());
}
let pool = ctx.data::<PgPool>()?;
let gql_ctx = ctx.data::<GraphQLContext>()?;
let mut bound_values = Vec::new();
let mut value_rows = Vec::with_capacity(representations.len());
for (ordinality, representation) in representations.iter().enumerate() {
let normalized_values = normalized_entity_key_values(
&representation.values,
&lookup.key_columns,
&lookup.type_name,
)?;
let mut values = Vec::with_capacity(lookup.key_columns.len() + 1);
for (key_column, value) in lookup.key_columns.iter().zip(normalized_values) {
values.push(format!(
"${}::{}",
bound_values.len() + 1,
key_column.pg_type
));
bound_values.push(value);
}
values.push(ordinality.to_string());
value_rows.push(format!("({})", values.join(", ")));
}
let representation_alias = "pgrst_entity_representation";
let ordinality_column = "pgrst_entity_ordinality";
let key_aliases: Vec<String> = (0..lookup.key_columns.len())
.map(|index| format!("pgrst_entity_key_{index}"))
.collect();
let mut representation_columns: Vec<String> = key_aliases
.iter()
.map(|column| postrust_sql::escape_ident(column))
.collect();
representation_columns.push(postrust_sql::escape_ident(ordinality_column));
let representations_sql = format!(
"(VALUES {}) AS {} ({})",
value_rows.join(", "),
postrust_sql::escape_ident(representation_alias),
representation_columns.join(", ")
);
let key_predicate = lookup
.key_columns
.iter()
.zip(&key_aliases)
.map(|(key_column, key_alias)| {
format!(
"{}.{} = {}.{}",
postrust_sql::escape_ident(READ_ROW),
postrust_sql::escape_ident(&key_column.column_name),
postrust_sql::escape_ident(representation_alias),
postrust_sql::escape_ident(key_alias)
)
})
.collect::<Vec<_>>()
.join(" AND ");
let mut where_sql = format!(" WHERE ({key_predicate})");
let permission = permission_predicate(
&gql_ctx.caller(),
state.names.as_ref(),
&lookup.schema_name,
&lookup.table_name,
)?;
if let Some(predicate) = permission {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let scope = WhereScope::table(
&lookup.schema_name,
&lookup.table_name,
&lookup.type_name,
state.names.as_ref(),
)
.under_alias(READ_ROW)
.with_resolution(cache, state.relationships.as_ref())
.for_caller(gql_ctx.caller());
let (filter_sql, filter_values) =
build_where_clause(Some(&predicate), bound_values.len() + 1, &scope)?;
if !filter_sql.is_empty() {
where_sql = format!(
"{} AND ({})",
where_sql,
filter_sql.trim_start_matches("WHERE ")
);
bound_values.extend(filter_values);
}
}
let source = format!(
"{}.{} AS {}",
postrust_sql::escape_ident(&lookup.schema_name),
postrust_sql::escape_ident(&lookup.table_name),
postrust_sql::escape_ident(READ_ROW)
);
let source_rows = format!(
"SELECT {}.* FROM {}{}",
postrust_sql::escape_ident(READ_ROW),
source,
where_sql
);
let mut projection = {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
guard
.as_ref()
.and_then(|cache| {
cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
&lookup.schema_name,
&lookup.table_name,
))
})
.and_then(|table| rename_projection(table, "src", state.names.as_ref()))
.unwrap_or_else(|| "src.*".to_string())
};
let projected_rows = {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let table = cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
&lookup.schema_name,
&lookup.table_name,
));
let selected_fields = entity_selection_fields(ctx, &lookup.type_name);
let mut param_idx = bound_values.len() + 1;
let computed = match table {
Some(table) => computed_projections(
table,
selected_fields.iter().copied(),
"src",
state.names.as_ref(),
cache,
&mut param_idx,
&mut bound_values,
)?,
None => Vec::new(),
};
let embeds = build_embed_expressions(
&gql_ctx.caller(),
cache,
state.relationships.as_ref(),
&lookup.type_name,
"src",
selected_fields,
state.max_rows,
&mut 0,
&mut param_idx,
&mut bound_values,
state.names.as_ref(),
)?;
for expression in &computed {
projection.push_str(", ");
projection.push_str(expression);
}
for (field_name, expression) in &embeds {
projection.push_str(", ");
projection.push_str(expression);
projection.push_str(" AS ");
projection.push_str(&postrust_sql::escape_ident(field_name));
}
format!("SELECT {} FROM ({}) AS src", projection, source_rows)
};
let sql = format!(
"SELECT json_build_array({}.{}, {}) FROM {} CROSS JOIN LATERAL ({}) AS t",
postrust_sql::escape_ident(representation_alias),
postrust_sql::escape_ident(ordinality_column),
row_json("t", &lookup.row_column_types),
representations_sql,
projected_rows,
);
let mut tx = begin_with_session(pool, gql_ctx).await?;
let rows = execute_query_on(&mut tx, &sql, &bound_values).await?;
tx.commit().await?;
let mut by_ordinal = std::iter::repeat_with(|| None)
.take(representations.len())
.collect::<Vec<_>>();
for result in rows {
let serde_json::Value::Array(mut pair) = result else {
return Err(async_graphql::Error::new(
"entity query returned an invalid result",
));
};
if pair.len() != 2 {
return Err(async_graphql::Error::new(
"entity query returned an invalid result",
));
}
let row = pair.pop().expect("the result pair has two values");
let ordinality = pair
.pop()
.and_then(|value| value.as_u64())
.and_then(|value| usize::try_from(value).ok())
.filter(|value| *value < by_ordinal.len())
.ok_or_else(|| async_graphql::Error::new("entity query returned an invalid ordinal"))?;
by_ordinal[ordinality] = Some(row);
}
Ok(by_ordinal)
}
fn normalized_entity_key_values(
values: &serde_json::Map<String, serde_json::Value>,
key_columns: &[EntityKeyColumn],
type_name: &str,
) -> Result<Vec<serde_json::Value>, async_graphql::Error> {
let mut key = Vec::with_capacity(key_columns.len());
for key_column in key_columns {
let value = values.get(&key_column.field_name).ok_or_else(|| {
async_graphql::Error::new(format!(
"entity representation for \"{}\" is missing key field \"{}\"",
type_name, key_column.field_name
))
})?;
key.push(normalize_entity_key_value(type_name, key_column, value)?);
}
Ok(key)
}
fn normalize_entity_key_value(
type_name: &str,
key_column: &EntityKeyColumn,
value: &serde_json::Value,
) -> Result<serde_json::Value, async_graphql::Error> {
let invalid = |expected: &str| {
async_graphql::Error::new(format!(
"entity representation for \"{}\" has invalid key field \"{}\": expected {}, found {}",
type_name,
key_column.field_name,
expected,
json_value_kind(value)
))
};
if key_column.explicit_id {
return normalize_graphql_id(value, &key_column.pg_type).ok_or_else(|| {
invalid(&format!(
"an ID compatible with PostgreSQL type \"{}\"",
key_column.pg_type
))
});
}
let pg_type = key_column.pg_type.trim().to_ascii_lowercase();
let normalized = if matches!(pg_type.as_str(), "int8" | "bigint") {
match value {
serde_json::Value::Number(number) => number.as_i64(),
serde_json::Value::String(text) => text.parse::<i64>().ok(),
_ => None,
}
.map(serde_json::Value::from)
} else if integer_bounds(&pg_type).is_some() {
value.as_i64().and_then(|integer| {
let (min, max) = integer_bounds(&pg_type)?;
(min..=max)
.contains(&integer)
.then(|| serde_json::Value::from(integer))
})
} else if matches!(
pg_type.as_str(),
"float4" | "float8" | "real" | "double precision"
) {
value
.as_f64()
.and_then(serde_json::Number::from_f64)
.map(serde_json::Value::Number)
} else if matches!(pg_type.as_str(), "numeric" | "decimal") {
match value {
serde_json::Value::Number(number) => {
Some(serde_json::Value::String(number.to_string()))
}
serde_json::Value::String(text) if text.parse::<sqlx::types::BigDecimal>().is_ok() => {
Some(serde_json::Value::String(text.clone()))
}
_ => None,
}
} else if matches!(pg_type.as_str(), "bool" | "boolean") {
value.as_bool().map(serde_json::Value::Bool)
} else if matches!(pg_type.as_str(), "json" | "jsonb") {
Some(value.clone())
} else if is_array_type(&pg_type) {
value.is_array().then(|| value.clone())
} else {
value
.as_str()
.map(|text| serde_json::Value::String(text.to_string()))
};
normalized.ok_or_else(|| {
invalid(&format!(
"a value compatible with PostgreSQL type \"{}\"",
key_column.pg_type
))
})
}
fn normalize_graphql_id(value: &serde_json::Value, pg_type: &str) -> Option<serde_json::Value> {
if integer_bounds(pg_type).is_some() {
let integer = match value {
serde_json::Value::Number(number) => number.as_i64(),
serde_json::Value::String(text) => text.parse::<i64>().ok(),
_ => None,
}?;
let (min, max) = integer_bounds(pg_type)?;
return (min..=max)
.contains(&integer)
.then(|| serde_json::Value::from(integer));
}
if pg_type.eq_ignore_ascii_case("uuid") {
return normalize_uuid(value).map(serde_json::Value::String);
}
if is_text_type(pg_type) {
return match value {
serde_json::Value::String(s) => Some(serde_json::Value::String(s.clone())),
serde_json::Value::Number(n) if n.is_i64() || n.is_u64() => {
Some(serde_json::Value::String(n.to_string()))
}
_ => None,
};
}
None
}
fn integer_bounds(pg_type: &str) -> Option<(i64, i64)> {
match pg_type.trim().to_ascii_lowercase().as_str() {
"int2" | "smallint" => Some((i16::MIN.into(), i16::MAX.into())),
"int4" | "int" | "integer" => Some((i32::MIN.into(), i32::MAX.into())),
"int8" | "bigint" => Some((i64::MIN, i64::MAX)),
_ => None,
}
}
fn is_text_type(pg_type: &str) -> bool {
matches!(
pg_type.trim().to_ascii_lowercase().as_str(),
"text" | "varchar" | "character varying" | "char" | "character" | "bpchar" | "citext"
)
}
fn normalize_uuid(value: &serde_json::Value) -> Option<String> {
sqlx::types::Uuid::parse_str(value.as_str()?)
.ok()
.map(|uuid| uuid.to_string())
}
fn json_value_kind(value: &serde_json::Value) -> &'static str {
match value {
serde_json::Value::Null => "null",
serde_json::Value::Bool(_) => "a boolean",
serde_json::Value::Number(_) => "a number",
serde_json::Value::String(_) => "a string",
serde_json::Value::Array(_) => "an array",
serde_json::Value::Object(_) => "an object",
}
}
fn add_function_fields(
mut root: Object,
generated: &GeneratedSchema,
volatile: bool,
max_rows: Option<i64>,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
names: Arc<crate::names::NameOverrides>,
) -> Object {
let taken: HashSet<&str> = if volatile {
generated
.mutation_fields
.iter()
.map(|f| f.name.as_str())
.collect()
} else {
generated
.query_fields
.iter()
.map(|f| f.name.as_str())
.collect()
};
for function in &generated.function_fields {
if function.volatile != volatile {
continue;
}
if taken.contains(function.name.as_str()) {
tracing::warn!(
"not exposing {}.{}({}) -- a table already has a root field \
called {}",
function.schema_name,
function.function_name,
function.argument_signature(),
function.name,
);
continue;
}
let spec = Arc::new(QueryFieldSpec {
schema_name: function.returns_table.0.clone(),
table_name: function.returns_table.1.clone(),
type_name: function.returns.clone(),
is_by_pk: false,
pk_columns: Vec::new(),
max_rows,
relationships: Arc::clone(&relationships),
names: Arc::clone(&names),
call: Some(Arc::new(FunctionCall {
schema: function.schema_name.clone(),
name: function.function_name.clone(),
arguments: function.arguments.clone(),
session_argument: function.session_argument.clone(),
})),
});
let mut field = Field::new(
&function.name,
TypeRef::named_nn_list_nn(&function.returns),
move |ctx| {
let spec = Arc::clone(&spec);
FieldFuture::new(async move { resolve_query(&ctx, &spec).await })
},
);
if !function.arguments.is_empty() {
field = field.argument(InputValue::new(
"args",
TypeRef::named_nn(format!("{}_args", function.name)),
));
}
field = field
.argument(InputValue::new(
"where",
TypeRef::named(crate::input::bool_exp::bool_exp_type_name(
&function.local_returns,
)),
))
.argument(InputValue::new(
"order_by",
TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(
&function.local_returns,
)),
))
.argument(InputValue::new(
"distinct_on",
TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
&function.local_returns,
)),
))
.argument(InputValue::new("limit", TypeRef::named("Int")))
.argument(InputValue::new("offset", TypeRef::named("Int")));
if let Some(description) = &function.description {
field = field.description(description);
}
root = root.field(field);
let agg_spec = Arc::new(AggregateSpec {
schema_name: function.returns_table.0.clone(),
table_name: function.returns_table.1.clone(),
type_name: function.returns.clone(),
max_rows,
relationships: Arc::clone(&relationships),
names: Arc::clone(&names),
call: Some(Arc::new(FunctionCall {
schema: function.schema_name.clone(),
name: function.function_name.clone(),
arguments: function.arguments.clone(),
session_argument: function.session_argument.clone(),
})),
});
let mut aggregate = Field::new(
format!("{}_aggregate", function.name),
TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
&function.local_returns,
)),
move |ctx| {
let agg_spec = Arc::clone(&agg_spec);
FieldFuture::new(async move { resolve_aggregate(&ctx, &agg_spec).await })
},
);
if !function.arguments.is_empty() {
aggregate = aggregate.argument(InputValue::new(
"args",
TypeRef::named_nn(format!("{}_args", function.name)),
));
}
aggregate = with_row_arguments(aggregate, &function.local_returns).description(format!(
"fetch aggregated fields from the table: \"{}\"",
function.returns_table.1
));
root = root.field(aggregate);
}
root
}
fn function_source(
ctx: &ResolverContext<'_>,
call: &FunctionCall,
alias: &str,
bound_values: &mut Vec<serde_json::Value>,
) -> Result<String, async_graphql::Error> {
let given = ctx
.args
.try_get("args")
.ok()
.map(|v| accessor_to_json(&v))
.unwrap_or_else(|| serde_json::json!({}));
let mut passed = Vec::new();
for (name, _, required) in &call.arguments {
match given.get(name) {
None if *required => {
return Err(async_graphql::Error::new(format!(
"{} needs the argument \"{}\"",
call.name, name
)))
}
None => continue,
Some(value) => {
let (_, pg_type, _) = call
.arguments
.iter()
.find(|(argument, _, _)| argument == name)
.expect("iterating the arguments");
passed.push(format!(
"{} => ${}::{}",
postrust_sql::escape_ident(name),
bound_values.len() + 1,
pg_type
));
bound_values.push(value.clone());
}
}
}
if let Some(session) = &call.session_argument {
passed.push(format!(
"{} => {}",
postrust_sql::escape_ident(session),
SESSION_ARGUMENT
));
}
Ok(format!(
"{}.{}({}) AS {}",
postrust_sql::escape_ident(&call.schema),
postrust_sql::escape_ident(&call.name),
passed.join(", "),
postrust_sql::escape_ident(alias)
))
}
struct FunctionCall {
schema: String,
name: String,
arguments: Vec<(String, String, bool)>,
session_argument: Option<String>,
}
struct AggregateSpec {
schema_name: String,
table_name: String,
type_name: String,
max_rows: Option<i64>,
relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
names: Arc<crate::names::NameOverrides>,
call: Option<Arc<FunctionCall>>,
}
async fn resolve_aggregate<'a>(
ctx: &ResolverContext<'a>,
spec: &AggregateSpec,
) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
Ok(Some(FieldValue::value(aggregate_value(ctx, spec).await?)))
}
async fn aggregate_value(
ctx: &ResolverContext<'_>,
spec: &AggregateSpec,
) -> Result<Value, async_graphql::Error> {
use crate::schema::aggregate as agg;
let pool = ctx.data::<PgPool>()?;
let gql_ctx = ctx.data::<GraphQLContext>()?;
let mut bound_values: Vec<serde_json::Value> = Vec::new();
let source = match &spec.call {
None => format!(
"{}.{}",
postrust_sql::escape_ident(&spec.schema_name),
postrust_sql::escape_ident(&spec.table_name)
),
Some(call) => function_source(ctx, call, &spec.table_name, &mut bound_values)?,
};
let mut where_sql = String::new();
{
let requested = ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v));
let permission = permission_predicate(
&gql_ctx.caller(),
spec.names.as_ref(),
&spec.schema_name,
&spec.table_name,
)?;
if let Some(predicate) = and_predicates(requested, permission) {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let scope = match &spec.call {
None => WhereScope::table(
&spec.schema_name,
&spec.table_name,
&spec.type_name,
spec.names.as_ref(),
)
.with_resolution(cache, spec.relationships.as_ref()),
Some(_) => WhereScope::for_alias(
&spec.schema_name,
&spec.table_name,
&spec.table_name,
&spec.type_name,
cache,
spec.relationships.as_ref(),
spec.names.as_ref(),
),
}
.for_caller(gql_ctx.caller());
let (sql, values) =
build_where_clause(Some(&predicate), bound_values.len() + 1, &scope)?;
if !sql.is_empty() {
where_sql = format!(" {}", sql);
bound_values.extend(values);
}
}
}
let order_sql = build_order_by_clause(
ctx,
&gql_ctx.schema_cache,
&spec.schema_name,
&spec.table_name,
&spec.type_name,
spec.relationships.as_ref(),
&format!(
"{}.{}",
postrust_sql::escape_ident(&spec.schema_name),
postrust_sql::escape_ident(&spec.table_name)
),
spec.names.as_ref(),
)
.await?;
let requested_limit = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok());
let offset = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok());
let ceiling = match (
spec.max_rows,
permission_limit(
&gql_ctx.caller(),
spec.names.as_ref(),
&spec.schema_name,
&spec.table_name,
),
) {
(Some(configured), Some(granted)) => Some(configured.min(granted)),
(only, None) | (None, only) => only,
};
let limit = match (requested_limit, ceiling) {
(Some(requested), Some(ceiling)) => Some(requested.min(ceiling)),
(Some(requested), None) => Some(requested),
(None, ceiling) => ceiling,
};
let rows = format!("SELECT * FROM {}{}{}", source, where_sql, order_sql);
let mut inner = rows.clone();
if let Some(limit) = limit {
inner.push_str(&format!(" LIMIT {}", limit));
}
if let Some(offset) = offset {
inner.push_str(&format!(" OFFSET {}", offset));
}
let mut counted = rows;
if let Some(limit) = requested_limit {
counted.push_str(&format!(" LIMIT {}", limit));
}
if let Some(offset) = offset {
counted.push_str(&format!(" OFFSET {}", offset));
}
let mut wants_nodes = false;
let mut wants_aggregate = false;
let mut wanted: BTreeMap<String, Vec<String>> = BTreeMap::new();
let mut counts: Vec<(String, String)> = Vec::new();
for selection in ctx.field().selection_set() {
match selection.name() {
"nodes" => wants_nodes = true,
"aggregate" => {
wants_aggregate = true;
for function in selection.selection_set() {
if function.name() == "__typename" {
continue;
}
if function.name() == "count" {
counts.push((
function.alias().unwrap_or("count").to_string(),
count_expression(
function,
(&spec.schema_name, &spec.table_name),
None,
spec.names.as_ref(),
),
));
continue;
}
let columns = wanted.entry(function.name().to_string()).or_default();
for column in function.selection_set() {
let name = column.name().to_string();
if name != "__typename" && !columns.contains(&name) {
columns.push(name);
}
}
}
}
_ => {}
}
}
let mut result = async_graphql::indexmap::IndexMap::new();
if wants_aggregate {
let mut parts = match counts.iter().any(|(key, _)| key == "count") {
true => Vec::new(),
false => vec!["'count', count(*)".to_string()],
};
for (key, expression) in &counts {
parts.push(format!(
"{}, {}",
postrust_sql::quote_literal(key),
expression
));
}
for (function, columns) in &wanted {
let sql_function = agg::NUMERIC_AGGREGATES
.iter()
.chain(agg::ORDERED_AGGREGATES.iter())
.find(|(name, _)| name == function)
.map(|(name, _)| *name)
.ok_or_else(|| {
async_graphql::Error::new(format!("unknown aggregate \"{}\"", function))
})?;
let per_column: Vec<String> = columns
.iter()
.map(|field| {
let source = spec
.names
.column_source(&spec.schema_name, &spec.table_name, field)
.unwrap_or(field);
format!(
"{}, {}({})",
postrust_sql::quote_literal(field),
sql_function,
postrust_sql::escape_ident(source)
)
})
.collect();
parts.push(format!(
"'{}', json_build_object({})",
sql_function,
per_column.join(", ")
));
}
let sql = format!(
"SELECT json_build_object({}) FROM ({}) AS pgrst_agg",
parts.join(", "),
counted
);
let mut conn = begin_with_session(pool, gql_ctx).await?;
let rows = execute_query_on(&mut conn, &sql, &bound_values).await?;
conn.commit().await?;
if let Some(first) = rows.into_iter().next() {
if let Value::Object(map) = json_to_value(first) {
result.insert(async_graphql::Name::new("aggregate"), Value::Object(map));
}
}
}
if wants_nodes {
let nodes = ctx
.field()
.selection_set()
.find(|selection| selection.name() == "nodes")
.ok_or_else(|| async_graphql::Error::new("nodes was asked for and then was not"))?;
let mut node_values = bound_values.clone();
let (projection, row_column_types) = {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let qi = postrust_core::api_request::QualifiedIdentifier::new(
&spec.schema_name,
&spec.table_name,
);
let table = cache.get_table(&qi);
let mut param_idx = node_values.len() + 1;
let computed = match table {
Some(table) => computed_projections(
table,
nodes.selection_set(),
"src",
spec.names.as_ref(),
cache,
&mut param_idx,
&mut node_values,
)?,
None => Vec::new(),
};
let embeds = build_embed_expressions(
&gql_ctx.caller(),
cache,
spec.relationships.as_ref(),
&spec.type_name,
"src",
nodes.selection_set(),
spec.max_rows,
&mut 0,
&mut param_idx,
&mut node_values,
spec.names.as_ref(),
)?;
let mut projection = table
.and_then(|table| rename_projection(table, "src", spec.names.as_ref()))
.unwrap_or_else(|| "src.*".to_string());
for expression in &computed {
projection.push_str(", ");
projection.push_str(expression);
}
for (field_name, expression) in &embeds {
projection.push_str(", ");
projection.push_str(expression);
projection.push_str(" AS ");
projection.push_str(&postrust_sql::escape_ident(field_name));
}
(
projection,
table
.map(|table| exposed_column_types(table, spec.names.as_ref()))
.unwrap_or_default(),
)
};
let sql = format!(
"SELECT {} FROM (SELECT {} FROM ({}) AS src) AS pgrst_nodes",
row_json("pgrst_nodes", &row_column_types),
projection,
inner
);
let mut conn = begin_with_session(pool, gql_ctx).await?;
let rows = execute_query_on(&mut conn, &sql, &node_values).await?;
conn.commit().await?;
result.insert(
async_graphql::Name::new("nodes"),
Value::List(rows.into_iter().map(json_to_value).collect()),
);
}
let _ = &spec.type_name;
Ok(Value::Object(result))
}
async fn resolve_query<'a>(
ctx: &ResolverContext<'a>,
spec: &QueryFieldSpec,
) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
Ok(rows_as_field_value(
query_rows(ctx, spec).await?,
spec.is_by_pk,
))
}
async fn query_rows(
ctx: &ResolverContext<'_>,
spec: &QueryFieldSpec,
) -> Result<Vec<serde_json::Value>, async_graphql::Error> {
let schema_name = spec.schema_name.as_str();
let table_name = spec.table_name.as_str();
let type_name = spec.type_name.as_str();
let is_by_pk = spec.is_by_pk;
let pk_columns = spec.pk_columns.as_slice();
let max_rows = spec.max_rows;
let relationships = spec.relationships.as_ref();
let pool = ctx.data::<PgPool>()?;
let gql_ctx = ctx.data::<GraphQLContext>()?;
debug!("Resolving query for table: {}", table_name);
let requested_limit: Option<i64> = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok());
let offset: Option<i64> = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok());
let limit: Option<i64> = if is_by_pk {
Some(1)
} else {
let ceiling = match (
max_rows,
permission_limit(
&gql_ctx.caller(),
spec.names.as_ref(),
schema_name,
table_name,
),
) {
(Some(configured), Some(granted)) => Some(configured.min(granted)),
(only, None) | (None, only) => only,
};
match (requested_limit, ceiling) {
(Some(requested), Some(ceiling)) => Some(requested.min(ceiling)),
(Some(requested), None) => Some(requested),
(None, ceiling) => ceiling,
}
};
let mut where_sql = String::new();
let mut bound_values: Vec<serde_json::Value> = Vec::new();
let source = match &spec.call {
None => format!(
"{}.{} AS {}",
postrust_sql::escape_ident(schema_name),
postrust_sql::escape_ident(table_name),
postrust_sql::escape_ident(READ_ROW)
),
Some(call) => function_source(ctx, call, table_name, &mut bound_values)?,
};
let source_ref = match &spec.call {
None => postrust_sql::escape_ident(READ_ROW),
Some(_) => postrust_sql::escape_ident(table_name),
};
if is_by_pk {
if pk_columns.is_empty() {
return Err(async_graphql::Error::new(format!(
"\"{}\" has no primary key, so it cannot be queried by key",
table_name
)));
}
let mut conditions = Vec::with_capacity(pk_columns.len());
for (idx, (col_name, pg_type)) in pk_columns.iter().enumerate() {
let field = spec
.names
.column(schema_name, table_name, col_name)
.unwrap_or(col_name);
let value = ctx.args.try_get(field).map_err(|_| {
async_graphql::Error::new(format!(
"missing required primary key argument \"{}\"",
field
))
})?;
conditions.push(format!(
"{} = ${}::{}",
postrust_sql::escape_ident(col_name),
idx + 1,
pg_type
));
bound_values.push(accessor_to_json(&value));
}
where_sql = format!(" WHERE {}", conditions.join(" AND "));
}
{
let requested = match is_by_pk {
true => None,
false => ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v)),
};
let permission = permission_predicate(
&gql_ctx.caller(),
spec.names.as_ref(),
schema_name,
table_name,
)?;
if let Some(predicate) = and_predicates(requested, permission) {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let scope = match &spec.call {
None => WhereScope::table(schema_name, table_name, type_name, spec.names.as_ref())
.under_alias(READ_ROW)
.with_resolution(cache, relationships)
.for_caller(gql_ctx.caller()),
Some(_) => WhereScope::for_alias(
schema_name,
table_name,
table_name,
type_name,
cache,
relationships,
spec.names.as_ref(),
)
.for_caller(gql_ctx.caller()),
};
let (filter_sql, filter_values) =
build_where_clause(Some(&predicate), bound_values.len() + 1, &scope)?;
if !filter_sql.is_empty() {
where_sql = match where_sql.is_empty() {
true => format!(" {}", filter_sql),
false => format!(
"{} AND ({})",
where_sql,
filter_sql.trim_start_matches("WHERE ")
),
};
bound_values.extend(filter_values);
}
}
}
let (order_sql, distinct_on) = if is_by_pk {
(String::new(), Vec::new())
} else {
(
build_order_by_clause(
ctx,
&gql_ctx.schema_cache,
schema_name,
table_name,
type_name,
relationships,
&source_ref,
spec.names.as_ref(),
)
.await?,
build_distinct_on(
ctx,
&gql_ctx.schema_cache,
schema_name,
table_name,
spec.names.as_ref(),
)
.await?,
)
};
let (distinct_sql, order_sql) = {
let written = order_sql.strip_prefix(" ORDER BY ");
let (prefix, order) = distinct_on_clause(&distinct_on, written).map_err(|error| {
at_path(
error,
&format!(
"$.selectionSet.{}.args",
ctx.ctx.field().alias().unwrap_or(ctx.ctx.field().name())
),
)
})?;
(
prefix,
match order {
Some(order) => format!(" ORDER BY {}", order),
None => String::new(),
},
)
};
let (computed, row_column_types) = {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
match guard.as_ref() {
Some(cache) => {
let qi =
postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
match cache.get_table(&qi) {
Some(table) => {
let mut param_idx = bound_values.len() + 1;
(
computed_projections(
table,
ctx.field().selection_set(),
"src",
spec.names.as_ref(),
cache,
&mut param_idx,
&mut bound_values,
)?,
exposed_column_types(table, spec.names.as_ref()),
)
}
None => (Vec::new(), HashMap::new()),
}
}
None => (Vec::new(), HashMap::new()),
}
};
let mut inner = format!(
"SELECT {}* FROM {}{}{}",
distinct_sql, source, where_sql, order_sql
);
if let Some(limit) = limit {
inner.push_str(&format!(" LIMIT {}", limit));
}
if let Some(offset) = offset {
inner.push_str(&format!(" OFFSET {}", offset));
}
let embed_expressions = {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
match guard.as_ref() {
Some(cache) => {
let mut param_idx = bound_values.len() + 1;
build_embed_expressions(
&gql_ctx.caller(),
cache,
relationships,
type_name,
"src",
ctx.field().selection_set(),
max_rows,
&mut 0,
&mut param_idx,
&mut bound_values,
spec.names.as_ref(),
)?
}
None => Vec::new(),
}
};
let renamed = {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
guard
.as_ref()
.and_then(|cache| {
cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
schema_name,
table_name,
))
})
.and_then(|table| rename_projection(table, "src", spec.names.as_ref()))
};
let inner = if embed_expressions.is_empty() && computed.is_empty() && renamed.is_none() {
inner
} else {
let mut projection = renamed.unwrap_or_else(|| "src.*".to_string());
for expression in &computed {
projection.push_str(", ");
projection.push_str(expression);
}
for (field_name, expression) in &embed_expressions {
projection.push_str(", ");
projection.push_str(expression);
projection.push_str(" AS ");
projection.push_str(&postrust_sql::escape_ident(field_name));
}
format!("SELECT {} FROM ({}) AS src", projection, inner)
};
let sql = format!(
"SELECT {} FROM ({}) t",
row_json("t", &row_column_types),
inner
);
let mut tx = begin_with_session(pool, gql_ctx).await?;
let mut result = execute_query_on(&mut tx, &sql, &bound_values).await?;
if embed_expressions.is_empty() {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let embed_ctx = EmbedContext {
schema_cache: cache,
relationships,
max_rows,
};
embed_relationships(&mut tx, &embed_ctx, type_name, ctx.field(), &mut result).await?;
}
tx.commit().await?;
Ok(result)
}
fn rows_as_field_value<'a>(rows: Vec<serde_json::Value>, is_by_pk: bool) -> Option<FieldValue<'a>> {
match is_by_pk {
true => rows
.into_iter()
.next()
.map(|v| FieldValue::value(json_to_value(v))),
false => Some(FieldValue::list(
rows.into_iter()
.map(|v| FieldValue::value(json_to_value(v))),
)),
}
}
#[allow(clippy::too_many_arguments)]
async fn resolve_mutation<'a>(
ctx: &ResolverContext<'a>,
schema_name: &str,
table_name: &str,
mutation_type: MutationType,
pk_columns: &[(String, String)],
relationships: &Arc<HashMap<String, Vec<RelationshipField>>>,
type_names: &Arc<HashMap<(String, String), String>>,
names: &Arc<crate::names::NameOverrides>,
max_rows: Option<i64>,
) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
let pool = ctx.data::<PgPool>()?;
let gql_ctx = ctx.data::<GraphQLContext>()?;
debug!(
"Resolving mutation for table: {} type: {:?}",
table_name, mutation_type
);
let type_name = type_names
.get(&(schema_name.to_string(), table_name.to_string()))
.cloned()
.unwrap_or_else(|| table_name.to_string());
let by_key = matches!(
mutation_type,
MutationType::InsertOne | MutationType::UpdateByPk | MutationType::DeleteByPk
);
let returning = if by_key {
Some(ctx.field())
} else {
ctx.field()
.selection_set()
.find(|selection| selection.name() == "returning")
};
if mutation_type == MutationType::UpdateMany {
let updates = ctx
.args
.try_get("updates")
.ok()
.map(|v| accessor_to_json(&v))
.unwrap_or(serde_json::Value::Null);
let serde_json::Value::Array(entries) = updates else {
return Err(async_graphql::Error::new(
"\"updates\" takes a list of {where, _set, …} objects",
));
};
let column_types = column_types_of(&gql_ctx.schema_cache, schema_name, table_name).await;
let mut answers: Vec<FieldValue<'_>> = Vec::with_capacity(entries.len());
for entry in entries {
let serde_json::Value::Object(entry) = entry else {
return Err(async_graphql::Error::new(
"each update is an object of {where, _set, …}",
));
};
let operators: Vec<(&'static str, serde_json::Value)> = UPDATE_OPERATORS
.iter()
.filter_map(|name| entry.get(*name).map(|value| (*name, value.clone())))
.filter(|(_, value)| !value.is_null())
.collect();
let rows = execute_update(
pool,
gql_ctx,
schema_name,
table_name,
&type_name,
operators,
column_types.clone(),
entry.get("where").cloned(),
relationships.as_ref(),
names.as_ref(),
)
.await?;
let affected = rows.len();
let rows = match returning {
Some(returning) if !rows.is_empty() => {
reread_returning(
pool,
gql_ctx,
schema_name,
table_name,
&type_name,
rows,
returning,
relationships.as_ref(),
names.as_ref(),
max_rows,
)
.await?
}
_ => rows,
};
answers.extend(mutation_result(rows, affected, false));
}
return Ok(Some(FieldValue::list(answers)));
}
let (result, affected) = match mutation_type {
MutationType::Insert | MutationType::InsertOne => {
let objects = ctx
.args
.try_get("objects")
.ok()
.map(|v| accessor_to_json(&v))
.or_else(|| {
ctx.args
.try_get("object")
.ok()
.map(|v| serde_json::Value::Array(vec![accessor_to_json(&v)]))
})
.unwrap_or_else(|| serde_json::Value::Array(vec![]));
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let here = format!(
"$.selectionSet.{}",
ctx.ctx.field().alias().unwrap_or(ctx.ctx.field().name())
);
let named = match mutation_type {
MutationType::InsertOne => "object",
_ => "objects",
};
let context = InsertContext {
on_conflict: ctx
.args
.try_get("on_conflict")
.ok()
.map(|v| accessor_to_json(&v))
.filter(|v| !v.is_null()),
cache,
relationships: relationships.as_ref(),
type_names: type_names.as_ref(),
names: names.as_ref(),
caller: gql_ctx.caller(),
row: String::new(),
objects: format!("{}.args.{}", here, named),
conflict: format!("{}.args.on_conflict", here),
};
execute_insert(
pool,
gql_ctx,
schema_name,
table_name,
objects,
&context,
mutation_type == MutationType::InsertOne,
)
.await?
}
MutationType::UpdateMany | MutationType::Update | MutationType::UpdateByPk => {
let operators: Vec<(&'static str, serde_json::Value)> = UPDATE_OPERATORS
.iter()
.filter_map(|name| {
ctx.args
.try_get(name)
.ok()
.map(|v| (*name, accessor_to_json(&v)))
})
.filter(|(_, value)| !value.is_null())
.collect();
let where_clause = if mutation_type == MutationType::UpdateByPk {
Some(pk_where_from_args(
ctx,
schema_name,
table_name,
pk_columns,
names.as_ref(),
)?)
} else {
ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v))
};
let where_clause = and_predicates(
where_clause,
permission_filter(
&gql_ctx.caller(),
names.as_ref(),
schema_name,
table_name,
crate::role::Verb::Update,
)?,
);
execute_update(
pool,
gql_ctx,
schema_name,
table_name,
&type_name,
operators,
column_types_of(&gql_ctx.schema_cache, schema_name, table_name).await,
where_clause,
relationships.as_ref(),
names.as_ref(),
)
.await
.map(|rows| {
let count = rows.len();
(rows, count)
})?
}
MutationType::Delete | MutationType::DeleteByPk => {
let where_clause = if mutation_type == MutationType::DeleteByPk {
Some(pk_where_from_args(
ctx,
schema_name,
table_name,
pk_columns,
names.as_ref(),
)?)
} else {
ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v))
};
let where_clause = and_predicates(
where_clause,
permission_filter(
&gql_ctx.caller(),
names.as_ref(),
schema_name,
table_name,
crate::role::Verb::Delete,
)?,
);
execute_delete(
pool,
gql_ctx,
schema_name,
table_name,
&type_name,
where_clause,
returning,
relationships.as_ref(),
names.as_ref(),
max_rows,
)
.await
.map(|rows| {
let count = rows.len();
(rows, count)
})?
}
};
let is_delete = matches!(
mutation_type,
MutationType::Delete | MutationType::DeleteByPk
);
let result = match returning {
Some(returning) if !result.is_empty() && !is_delete => {
reread_returning(
pool,
gql_ctx,
schema_name,
table_name,
&type_name,
result,
returning,
relationships.as_ref(),
names.as_ref(),
max_rows,
)
.await?
}
_ => result,
};
Ok(mutation_result(result, affected, by_key))
}
async fn write_tx<'a>(
gql_ctx: &'a GraphQLContext,
pool: &PgPool,
) -> Result<
tokio::sync::MutexGuard<'a, Option<sqlx::Transaction<'static, sqlx::Postgres>>>,
async_graphql::Error,
> {
let mut guard = gql_ctx.write.lock().await;
if guard.is_none() {
*guard = Some(begin_with_session(pool, gql_ctx).await?);
}
Ok(guard)
}
async fn begin_with_session(
pool: &PgPool,
gql_ctx: &GraphQLContext,
) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, async_graphql::Error> {
let mut tx = pool.begin().await?;
gql_ctx.session().apply_sqlx(&mut tx).await?;
Ok(tx)
}
fn row_json(expr: &str, column_types: &HashMap<String, String>) -> String {
let mut shapes: Vec<(&String, &String)> = column_types
.iter()
.filter(|(_, pg_type)| matches!(pg_type.as_str(), "geometry" | "geography"))
.collect();
if shapes.is_empty() {
return format!("row_to_json({})", expr);
}
shapes.sort();
let overrides: Vec<String> = shapes
.iter()
.map(|(name, _)| {
format!(
"{}, ST_AsGeoJSON({}.{}, 9, 4)::jsonb",
postrust_sql::quote_literal(name),
expr,
postrust_sql::escape_ident(name)
)
})
.collect();
format!(
"(to_jsonb({}) || jsonb_build_object({}))",
expr,
overrides.join(", ")
)
}
async fn execute_query_on(
conn: &mut sqlx::PgConnection,
sql: &str,
params: &[serde_json::Value],
) -> Result<Vec<serde_json::Value>, async_graphql::Error> {
use sqlx::Row;
trace!("Executing SQL: {}", sql);
let mut query = sqlx::query(AssertSqlSafe(sql));
for param in params {
query = bind_json_value(query, param);
}
let rows = query.fetch_all(&mut *conn).await.map_err(database_error)?;
let results: Vec<serde_json::Value> = rows
.into_iter()
.filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
.collect();
Ok(results)
}
fn check_geojson(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
fn refuse<T>(message: &str) -> Result<T, async_graphql::Error> {
Err(coded_error("parse-failed", message))
}
fn position(
value: &serde_json::Value,
) -> Result<&Vec<serde_json::Value>, async_graphql::Error> {
match value.as_array() {
Some(items) if items.len() >= 2 && items.iter().all(|i| i.is_number()) => Ok(items),
_ => refuse("A Position needs at least 2 elements"),
}
}
fn positions(
value: Option<&serde_json::Value>,
least: usize,
what: &str,
) -> Result<Vec<Vec<serde_json::Value>>, async_graphql::Error> {
let items = match value.and_then(|v| v.as_array()) {
Some(items) => items,
None => return refuse(&format!("A {} needs at least {} Positions", what, least)),
};
if items.len() < least {
return refuse(&format!("A {} needs at least {} Positions", what, least));
}
items.iter().map(|item| position(item).cloned()).collect()
}
fn ring(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
let points = positions(Some(value), 4, "LinearRing")?;
match (points.first(), points.last()) {
(Some(first), Some(last)) if first == last => Ok(()),
_ => refuse("the first and last locations have to be equal for a LinearRing"),
}
}
fn polygon(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
let Some(rings) = value.as_array() else {
return refuse("A LinearRing needs at least 4 Positions");
};
rings.iter().try_for_each(ring)
}
let serde_json::Value::Object(map) = value else {
return Ok(());
};
let Some(kind) = map.get("type").and_then(|v| v.as_str()) else {
return Ok(());
};
let coordinates = map.get("coordinates");
match kind {
"Point" => {
position(coordinates.unwrap_or(&serde_json::Value::Null))?;
}
"MultiPoint" => {
positions(coordinates, 0, "MultiPoint")?;
}
"LineString" => {
positions(coordinates, 2, "LineString")?;
}
"MultiLineString" => {
for line in coordinates.and_then(|v| v.as_array()).into_iter().flatten() {
positions(Some(line), 2, "LineString")?;
}
}
"Polygon" => polygon(coordinates.unwrap_or(&serde_json::Value::Null))?,
"MultiPolygon" => {
for one in coordinates.and_then(|v| v.as_array()).into_iter().flatten() {
polygon(one)?;
}
}
"GeometryCollection" => {
for geometry in map
.get("geometries")
.and_then(|v| v.as_array())
.into_iter()
.flatten()
{
check_geojson(geometry)?;
}
}
other => return refuse(&format!("unexpected geometry type: {}", other)),
}
Ok(())
}
fn computed_call(
definition: &postrust_core::schema_cache::ComputedColumn,
row: &str,
row_by_name: &str,
) -> String {
let function = format!(
"{}.{}",
postrust_sql::escape_ident(&definition.function.schema),
postrust_sql::escape_ident(&definition.function.name)
);
match (&definition.session_argument, &definition.row_argument) {
(Some(session), Some(row_argument)) => format!(
"{}({} => {}, {} => {})",
function,
postrust_sql::escape_ident(row_argument),
row_by_name,
postrust_sql::escape_ident(session),
SESSION_ARGUMENT
),
_ => format!("{}({})", function, row),
}
}
const SESSION_ARGUMENT: &str = "coalesce(current_setting('hasura.session', true), '{}')::json";
const UPDATE_OPERATORS: [&str; 7] = [
"_set",
"_inc",
"_append",
"_prepend",
"_delete_key",
"_delete_elem",
"_delete_at_path",
];
const READ_ROW: &str = "pgrst_src";
const WRITTEN_ROW: &str = "pgrst_row";
fn write_cast(column_types: &HashMap<String, String>, column: &str) -> String {
match column_types.get(column) {
Some(pg_type) => format!("::{}", pg_type),
None => String::new(),
}
}
fn write_operand(
column_types: &HashMap<String, String>,
column: &str,
value: &serde_json::Value,
) -> serde_json::Value {
match column_types.get(column).map(String::as_str) {
Some("json") | Some("jsonb") => match value {
serde_json::Value::String(_)
| serde_json::Value::Number(_)
| serde_json::Value::Bool(_) => serde_json::Value::String(value.to_string()),
other => other.clone(),
},
Some(pg_type) if is_array_type(pg_type) => match value {
serde_json::Value::Array(_) => serde_json::Value::String(array_literal(value)),
other => other.clone(),
},
_ => value.clone(),
}
}
fn is_array_type(pg_type: &str) -> bool {
pg_type.ends_with("[]") || pg_type.starts_with('_')
}
fn array_literal(value: &serde_json::Value) -> String {
fn element(value: &serde_json::Value) -> String {
match value {
serde_json::Value::Null => "NULL".to_string(),
serde_json::Value::Array(_) => array_literal(value),
serde_json::Value::String(text) => {
format!("\"{}\"", text.replace('\\', "\\\\").replace('"', "\\\""))
}
other => format!("\"{}\"", other.to_string().replace('"', "\\\"")),
}
}
let items = match value {
serde_json::Value::Array(items) => items,
other => return element(other),
};
format!(
"{{{}}}",
items.iter().map(element).collect::<Vec<_>>().join(",")
)
}
fn write_expression(
column_types: &HashMap<String, String>,
column: &str,
value: &serde_json::Value,
placeholder: &str,
) -> String {
let pg_type = column_types.get(column).map(String::as_str);
if value.is_object() && matches!(pg_type, Some("geometry") | Some("geography")) {
return match pg_type {
Some("geography") => format!("ST_GeomFromGeoJSON({})::geography", placeholder),
_ => format!("ST_GeomFromGeoJSON({})", placeholder),
};
}
if pg_type == Some("money") && value.is_number() {
return format!("{}::numeric::money", placeholder);
}
format!("{}{}", placeholder, write_cast(column_types, column))
}
async fn column_types_of(
schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
schema_name: &str,
table_name: &str,
) -> HashMap<String, String> {
let Ok(guard) = schema_cache.get().await else {
return HashMap::new();
};
let Some(cache) = guard.as_ref() else {
return HashMap::new();
};
let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
match cache.get_table(&qi) {
Some(table) => table
.columns
.values()
.map(|c| (c.name.clone(), c.nominal_type.clone()))
.collect(),
None => HashMap::new(),
}
}
fn insert_row<'life>(
conn: &'life mut sqlx::PgConnection,
schema_name: &'life str,
table_name: &'life str,
object: serde_json::Map<String, serde_json::Value>,
context: &'life InsertContext<'life>,
written_count: &'life mut usize,
) -> std::pin::Pin<
Box<
dyn std::future::Future<Output = Result<serde_json::Value, async_graphql::Error>>
+ Send
+ 'life,
>,
> {
Box::pin(async move {
use sqlx::Row;
let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
let table = context.cache.get_table(&qi).ok_or_else(|| {
async_graphql::Error::new(format!("unknown table \"{}\"", table_name))
})?;
let column_types: HashMap<String, String> = table
.columns
.values()
.map(|c| (c.name.clone(), c.nominal_type.clone()))
.collect();
let type_name = context
.type_names
.get(&(schema_name.to_string(), table_name.to_string()));
let relationships: &[RelationshipField] = type_name
.and_then(|name| context.relationships.get(name))
.map(|r| r.as_slice())
.unwrap_or(&[]);
let mut columns = serde_json::Map::new();
type Nested<'r> = (
&'r RelationshipField,
serde_json::Value,
Option<serde_json::Value>,
);
let mut to_one: Vec<Nested> = Vec::new();
let mut to_many: Vec<Nested> = Vec::new();
for (key, value) in object {
match relationships.iter().find(|r| r.name == key) {
None => {
columns.insert(
table_column_for(context.names, table, &key).to_string(),
value,
);
}
Some(_) if value.is_null() => {
continue;
}
Some(relationship) => {
let (data, conflict) = match &value {
serde_json::Value::Object(map) => (
map.get("data").cloned().unwrap_or(value.clone()),
map.get("on_conflict").cloned().filter(|v| !v.is_null()),
),
other => (other.clone(), None),
};
if child_holds_the_key(&relationship.relationship) {
to_many.push((relationship, data, conflict));
} else {
to_one.push((relationship, data, conflict));
}
}
}
}
for (column, value) in &columns {
if matches!(
column_types.get(column).map(String::as_str),
Some("geometry") | Some("geography")
) {
check_geojson(value).map_err(|error| {
let field = context
.names
.column(schema_name, table_name, column)
.unwrap_or(column);
at_path(error, &format!("{}.{}", context.row, field))
})?;
}
}
for (relationship, data, conflict) in to_one {
let plan =
postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, context.cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let serde_json::Value::Object(child) = data else {
return Err(async_graphql::Error::new(format!(
"\"{}\" takes an object to insert",
relationship.name
)));
};
for (local, _) in &plan.columns {
if columns.get(local).is_some_and(|v| !v.is_null()) {
return Err(at_path(
validation_error(format!(
"cannot insert object relationship \"{}\" as \"{}\" column \
values are already determined",
relationship.name, local
)),
&format!("{}.{}", context.row, relationship.name),
));
}
}
let under = format!("{}.{}", context.row, relationship.name);
let nested = InsertContext {
on_conflict: conflict,
cache: context.cache,
relationships: context.relationships,
type_names: context.type_names,
names: context.names,
caller: context.caller,
row: format!("{}.data", under),
objects: format!("{}.data", under),
conflict: format!("{}.on_conflict", under),
};
let written = insert_row(
conn,
&plan.foreign_schema,
&plan.foreign_table,
child,
&nested,
written_count,
)
.await?;
for (local, foreign) in &plan.columns {
if let Some(value) = written.get(foreign) {
columns.insert(local.clone(), value.clone());
}
}
}
for (column, value) in crate::role::presets(
&context.caller,
context.names,
schema_name,
table_name,
crate::role::Verb::Insert,
)
.map_err(|fault| coded_error(fault.code(), fault.to_string()))?
{
columns.insert(column, value);
}
let column_names: Vec<&str> = columns.keys().map(|k| k.as_str()).collect();
let mut conflict_values: Vec<serde_json::Value> = Vec::new();
let conflict_sql = match &context.on_conflict {
Some(serde_json::Value::Object(spec)) => {
let constraint = spec.get("constraint").and_then(|v| v.as_str());
let Some(constraint) = constraint else {
return Err(async_graphql::Error::new(
"on_conflict needs a constraint to resolve against",
));
};
if !table
.unique_constraints
.iter()
.any(|(name, _)| name == constraint)
{
return Err(async_graphql::Error::new(format!(
"\"{}\" is not a unique constraint of \"{}\"",
constraint, table_name
)));
}
let updates: Vec<&str> = spec
.get("update_columns")
.and_then(|v| v.as_array())
.map(|items| items.iter().filter_map(|i| i.as_str()).collect())
.unwrap_or_default();
if updates.contains(&PLACEHOLDER_COLUMN) {
return Err(at_path(
coded_error("validation-failed", "erroneous column name"),
&context.conflict,
));
}
if updates.is_empty() {
format!(
" ON CONFLICT ON CONSTRAINT {} DO NOTHING",
postrust_sql::escape_ident(constraint)
)
} else {
let preset = crate::role::presets(
&context.caller,
context.names,
schema_name,
table_name,
crate::role::Verb::Update,
)
.map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
let mut param_idx = column_names.len() + 1;
let mut assignments: Vec<String> =
Vec::with_capacity(preset.len() + updates.len());
for (column, value) in &preset {
assignments.push(format!(
"{} = {}",
postrust_sql::escape_ident(column),
write_expression(
&column_types,
column,
value,
&format!("${}", param_idx)
)
));
conflict_values.push(write_operand(&column_types, column, value));
param_idx += 1;
}
for field in &updates {
let column = table_column_for(context.names, table, field);
if preset.iter().any(|(written, _)| written == column) {
continue;
}
assignments.push(format!(
"{} = EXCLUDED.{}",
postrust_sql::escape_ident(column),
postrust_sql::escape_ident(column)
));
}
let granted = permission_filter(
&context.caller,
context.names,
schema_name,
table_name,
crate::role::Verb::Update,
)?;
let requested = spec.get("where").filter(|f| !f.is_null()).cloned();
let condition = match and_predicates(requested, granted) {
Some(filter) => {
let filter = &filter;
let type_name = context
.type_names
.get(&(schema_name.to_string(), table_name.to_string()))
.map(String::as_str)
.unwrap_or(table_name);
let scope = WhereScope::table(
schema_name,
table_name,
type_name,
context.names,
)
.under_alias(WRITTEN_ROW)
.with_resolution(context.cache, context.relationships);
let mut alias_counter = 0usize;
build_condition(
filter,
&scope,
&mut param_idx,
&mut conflict_values,
&mut alias_counter,
)?
.map(|sql| format!(" WHERE {}", sql))
.unwrap_or_default()
}
None => String::new(),
};
format!(
" ON CONFLICT ON CONSTRAINT {} DO UPDATE SET {}{}",
postrust_sql::escape_ident(constraint),
assignments.join(", "),
condition
)
}
}
_ => String::new(),
};
let qualified_table = postrust_sql::escape_ident(WRITTEN_ROW);
let check_scope = WhereScope::for_alias(
schema_name,
table_name,
WRITTEN_ROW,
context
.type_names
.get(&(schema_name.to_string(), table_name.to_string()))
.map(String::as_str)
.unwrap_or(table_name),
context.cache,
context.relationships,
context.names,
);
let (check_sql, check_values) = {
let check = crate::role::write_check(
&context.caller,
context.names,
schema_name,
table_name,
crate::role::Verb::Insert,
)
.map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
match check {
None => (String::new(), Vec::new()),
Some(check) => {
let mut values = Vec::new();
let mut param_idx = column_names.len() + conflict_values.len() + 1;
let mut alias_counter = 0usize;
match build_condition(
&check,
&check_scope,
&mut param_idx,
&mut values,
&mut alias_counter,
)? {
None => (String::new(), Vec::new()),
Some(sql) => (
format!(
", ({}) AS {}",
sql,
postrust_sql::escape_ident(CHECK_COLUMN)
),
values,
),
}
}
}
};
let names = column_names;
let written = if names.is_empty() {
let sql = format!(
"INSERT INTO {}.{} AS {} DEFAULT VALUES{} RETURNING {}{}",
postrust_sql::escape_ident(schema_name),
postrust_sql::escape_ident(table_name),
qualified_table,
conflict_sql,
row_json(&qualified_table, &column_types),
check_sql
);
let mut query = sqlx::query(AssertSqlSafe(sql));
for value in &conflict_values {
query = bind_json_value(query, value);
}
for value in &check_values {
query = bind_json_value(query, value);
}
query
.fetch_optional(&mut *conn)
.await
.map_err(|error| at_path(database_error(error), &context.objects))?
} else {
let placeholders: Vec<String> = names
.iter()
.enumerate()
.map(|(i, column)| {
let placeholder = format!("${}", i + 1);
match columns.get(*column) {
Some(value) => write_expression(&column_types, column, value, &placeholder),
None => placeholder,
}
})
.collect();
let sql = format!(
"INSERT INTO {}.{} AS {} ({}) VALUES ({}){} RETURNING {}{}",
postrust_sql::escape_ident(schema_name),
postrust_sql::escape_ident(table_name),
qualified_table,
names
.iter()
.map(|c| postrust_sql::escape_ident(c))
.collect::<Vec<_>>()
.join(", "),
placeholders.join(", "),
conflict_sql,
row_json(&qualified_table, &column_types),
check_sql
);
trace!("Executing INSERT SQL: {}", sql);
let mut query = sqlx::query(AssertSqlSafe(sql));
for column in &names {
if let Some(value) = columns.get(*column) {
query = bind_json_value(query, &write_operand(&column_types, column, value));
}
}
for value in &conflict_values {
query = bind_json_value(query, value);
}
for value in &check_values {
query = bind_json_value(query, value);
}
query
.fetch_optional(&mut *conn)
.await
.map_err(|error| at_path(database_error(error), &context.objects))?
};
let Some(written) = written else {
return Ok(serde_json::Value::Null);
};
refuse_unchecked_rows(std::slice::from_ref(&written), !check_sql.is_empty())
.map_err(|error| at_path(error, &context.objects))?;
let row: serde_json::Value = written.try_get(0).unwrap_or(serde_json::Value::Null);
*written_count += 1;
for (relationship, data, conflict) in to_many {
let plan =
postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, context.cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let children = match data {
serde_json::Value::Array(items) => items,
serde_json::Value::Object(map) => vec![serde_json::Value::Object(map)],
serde_json::Value::Null => Vec::new(),
_ => {
return Err(async_graphql::Error::new(format!(
"\"{}\" takes objects to insert",
relationship.name
)))
}
};
for (child_index, child) in children.into_iter().enumerate() {
let serde_json::Value::Object(mut child) = child else {
continue;
};
for (local, foreign) in &plan.columns {
if child.get(foreign).is_some_and(|v| !v.is_null()) {
return Err(at_path(
validation_error(format!(
"cannot insert \"{}\" columns as their values are already \
being determined by parent insert",
foreign
)),
&format!(
"{}.{}{}",
context.row,
relationship.name,
match relationship.is_list {
true => format!(".data[{}]", child_index),
false => ".data".to_string(),
}
),
));
}
if let Some(value) = row.get(local) {
child.insert(foreign.clone(), value.clone());
}
}
let at = match relationship.is_list {
true => format!(".data[{}]", child_index),
false => ".data".to_string(),
};
let under = format!("{}.{}", context.row, relationship.name);
let nested = InsertContext {
on_conflict: conflict.clone(),
cache: context.cache,
relationships: context.relationships,
type_names: context.type_names,
names: context.names,
caller: context.caller,
row: format!("{}{}", under, at),
objects: format!("{}.data", under),
conflict: format!("{}.on_conflict", under),
};
insert_row(
conn,
&plan.foreign_schema,
&plan.foreign_table,
child,
&nested,
written_count,
)
.await?;
}
}
Ok(row)
})
}
#[derive(Clone)]
struct InsertContext<'a> {
on_conflict: Option<serde_json::Value>,
cache: &'a SchemaCache,
relationships: &'a HashMap<String, Vec<RelationshipField>>,
type_names: &'a HashMap<(String, String), String>,
names: &'a crate::names::NameOverrides,
caller: crate::role::Caller<'a>,
row: String,
objects: String,
conflict: String,
}
#[allow(clippy::too_many_arguments)]
async fn reread_returning(
pool: &PgPool,
gql_ctx: &GraphQLContext,
schema_name: &str,
table_name: &str,
type_name: &str,
rows: Vec<Value>,
returning: async_graphql::SelectionField<'_>,
relationships: &HashMap<String, Vec<RelationshipField>>,
names: &crate::names::NameOverrides,
max_rows: Option<i64>,
) -> Result<Vec<Value>, async_graphql::Error> {
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let Some(cache) = guard.as_ref() else {
return Ok(rows);
};
let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
let Some(table) = cache.get_table(&qi) else {
return Ok(rows);
};
if table.pk_cols.is_empty() {
return Ok(rows);
}
let mut param_idx = 1usize;
let mut values: Vec<serde_json::Value> = Vec::new();
let mut alias_counter = 0usize;
let embeds = build_embed_expressions(
&gql_ctx.caller(),
cache,
relationships,
type_name,
"src",
returning.selection_set(),
max_rows,
&mut alias_counter,
&mut param_idx,
&mut values,
names,
)?;
let computed = computed_projections(
table,
returning.selection_set(),
"src",
names,
cache,
&mut param_idx,
&mut values,
)?;
let renames = names.renames_columns(&table.schema, &table.name);
if embeds.is_empty() && computed.is_empty() && !renames {
return Ok(rows);
}
let mut conditions = Vec::with_capacity(rows.len());
for row in &rows {
let mut parts = Vec::with_capacity(table.pk_cols.len());
for column in &table.pk_cols {
let Value::Object(map) = row else {
return Ok(rows);
};
let Some(value) = map.get(&async_graphql::Name::new(column)) else {
return Ok(rows);
};
parts.push(format!(
"{} = ${}",
postrust_sql::escape_ident(column),
param_idx
));
values.push(value_to_json(value));
param_idx += 1;
}
conditions.push(format!("({})", parts.join(" AND ")));
}
if conditions.is_empty() {
return Ok(rows);
}
let mut projection =
rename_projection(table, "src", names).unwrap_or_else(|| "src.*".to_string());
for expression in &computed {
projection.push_str(", ");
projection.push_str(expression);
}
for (name, expression) in &embeds {
projection.push_str(", ");
projection.push_str(expression);
projection.push_str(" AS ");
projection.push_str(&postrust_sql::escape_ident(name));
}
let sql = format!(
"SELECT {} FROM (SELECT {} FROM (SELECT * FROM {}.{} WHERE {}) AS src) AS pgrst_r",
row_json("pgrst_r", &exposed_column_types(table, names)),
projection,
postrust_sql::escape_ident(schema_name),
postrust_sql::escape_ident(table_name),
conditions.join(" OR ")
);
let mut guard = write_tx(gql_ctx, pool).await?;
let conn = guard.as_mut().expect("write_tx opens one");
let reread = execute_query_on(conn, &sql, &values).await?;
drop(guard);
if reread.is_empty() {
return Ok(rows);
}
Ok(reread.into_iter().map(json_to_value).collect())
}
async fn execute_insert(
pool: &PgPool,
gql_ctx: &GraphQLContext,
schema_name: &str,
table_name: &str,
objects: serde_json::Value,
context: &InsertContext<'_>,
single: bool,
) -> Result<(Vec<Value>, usize), async_graphql::Error> {
trace!("Insert mutation for {}: {:?}", table_name, objects);
let objects_array = match objects {
serde_json::Value::Array(arr) => arr,
serde_json::Value::Object(obj) => vec![serde_json::Value::Object(obj)],
_ => {
return Err(async_graphql::Error::new(
"objects must be an array or object",
))
}
};
if objects_array.is_empty() {
return Err(async_graphql::Error::new("objects cannot be empty"));
}
let mut guard = write_tx(gql_ctx, pool).await?;
let conn = guard.as_mut().expect("write_tx opens one");
let mut inserted: Vec<Value> = Vec::new();
let mut written = 0usize;
for (index, object) in objects_array.into_iter().enumerate() {
let serde_json::Value::Object(map) = object else {
return Err(async_graphql::Error::new(
"each object to insert is an object",
));
};
let context = InsertContext {
row: match single {
true => context.objects.clone(),
false => format!("{}[{}]", context.objects, index),
},
..context.clone()
};
let row = insert_row(conn, schema_name, table_name, map, &context, &mut written).await?;
if !row.is_null() {
inserted.push(json_to_value(row));
}
}
Ok((inserted, written))
}
fn mutation_result<'a>(rows: Vec<Value>, affected: usize, by_key: bool) -> Option<FieldValue<'a>> {
if by_key {
return rows.into_iter().next().map(FieldValue::value);
}
let mut response = async_graphql::indexmap::IndexMap::new();
response.insert(
async_graphql::Name::new("affected_rows"),
Value::from(affected),
);
response.insert(async_graphql::Name::new("returning"), Value::List(rows));
Some(FieldValue::value(Value::Object(response)))
}
fn bind_json_value<'q>(
query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
value: &serde_json::Value,
) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> {
match value {
serde_json::Value::Null => query.bind(None::<String>),
serde_json::Value::Bool(b) => query.bind(*b),
serde_json::Value::Number(n) => {
if let Some(i) = n.as_i64() {
query.bind(i)
} else if let Some(f) = n.as_f64() {
query.bind(f)
} else {
query.bind(n.to_string())
}
}
serde_json::Value::String(s) => query.bind(s.clone()),
_ => query.bind(value.to_string()),
}
}
#[allow(clippy::too_many_arguments)] async fn execute_update(
pool: &PgPool,
gql_ctx: &GraphQLContext,
schema_name: &str,
table_name: &str,
type_name: &str,
operators: Vec<(&'static str, serde_json::Value)>,
column_types: HashMap<String, String>,
where_clause: Option<serde_json::Value>,
relationships: &HashMap<String, Vec<RelationshipField>>,
names: &crate::names::NameOverrides,
) -> Result<Vec<Value>, async_graphql::Error> {
use sqlx::Row;
trace!("Update mutation for {}: {:?}", table_name, operators);
let mut set_parts: Vec<String> = Vec::new();
let mut set_values: Vec<serde_json::Value> = Vec::new();
let mut param_idx = 1;
let mut written: HashSet<String> = HashSet::new();
let preset = crate::role::presets(
&gql_ctx.caller(),
names,
schema_name,
table_name,
crate::role::Verb::Update,
)
.map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
let preset_columns: HashSet<&str> = preset.iter().map(|(name, _)| name.as_str()).collect();
for (column, value) in &preset {
let placeholder =
write_expression(&column_types, column, value, &format!("${}", param_idx));
set_parts.push(format!(
"{} = {}",
postrust_sql::escape_ident(column),
placeholder
));
set_values.push(write_operand(&column_types, column, value));
param_idx += 1;
}
for (operator, payload) in &operators {
let serde_json::Value::Object(map) = payload else {
return Err(async_graphql::Error::new(format!(
"\"{}\" takes an object mapping columns to values",
operator
)));
};
for (field, value) in map {
let column = &names
.column_source(schema_name, table_name, field)
.unwrap_or(field)
.to_string();
if preset_columns.contains(column.as_str()) {
continue;
}
if !written.insert(column.clone()) {
return Err(async_graphql::Error::new(format!(
"\"{}\" is written twice in one update; a column may be \
changed by one operator at a time",
column
)));
}
let quoted = postrust_sql::escape_ident(column);
let placeholder = match *operator {
"_set" | "_inc" => {
write_expression(&column_types, column, value, &format!("${}", param_idx))
}
_ => format!("${}", param_idx),
};
let assignment = match *operator {
"_set" => format!("{} = {}", quoted, placeholder),
"_inc" => format!("{} = {} + {}", quoted, quoted, placeholder),
"_append" => format!("{} = {} || {}::jsonb", quoted, quoted, placeholder),
"_prepend" => format!("{} = {}::jsonb || {}", quoted, placeholder, quoted),
"_delete_key" => format!("{} = {} - {}::text", quoted, quoted, placeholder),
"_delete_elem" => format!("{} = {} - {}::int", quoted, quoted, placeholder),
"_delete_at_path" => {
let Some(steps) = value.as_array() else {
return Err(async_graphql::Error::new(format!(
"\"_delete_at_path\" on \"{}\" takes a list of keys",
column
)));
};
let path = sql_array(steps, "text[]", &mut param_idx, &mut set_values);
set_parts.push(format!("{} = {} #- {}", quoted, quoted, path));
continue;
}
other => {
return Err(async_graphql::Error::new(format!(
"unsupported update operator \"{}\"",
other
)))
}
};
if *operator == "_set"
&& matches!(
column_types.get(column).map(String::as_str),
Some("geometry") | Some("geography")
)
{
check_geojson(value)?;
}
set_parts.push(assignment);
set_values.push(match *operator {
"_set" => write_operand(&column_types, column, value),
_ => value.clone(),
});
param_idx += 1;
}
}
if set_parts.is_empty() {
return Ok(Vec::new());
}
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let scope = WhereScope::table(schema_name, table_name, type_name, names)
.under_alias(WRITTEN_ROW)
.with_resolution(cache, relationships);
let (where_sql, where_values) = build_where_clause(where_clause.as_ref(), param_idx, &scope)?;
if where_sql.is_empty() {
return Err(async_graphql::Error::new(format!(
"update on \"{}\" requires a `where` argument with at least one \
recognised condition; refusing to update every row",
table_name
)));
}
let (check_sql, check_values) = permission_check_sql(
gql_ctx,
names,
schema_name,
table_name,
crate::role::Verb::Update,
&scope,
param_idx + where_values.len(),
)?;
let sql = format!(
"UPDATE {}.{} AS {} SET {} {} RETURNING {}{}",
postrust_sql::escape_ident(schema_name),
postrust_sql::escape_ident(table_name),
postrust_sql::escape_ident(WRITTEN_ROW),
set_parts.join(", "),
where_sql,
row_json(&postrust_sql::escape_ident(WRITTEN_ROW), &column_types),
check_sql
);
trace!("Executing UPDATE SQL: {}", sql);
let mut query = sqlx::query(AssertSqlSafe(sql));
for val in &set_values {
query = bind_json_value(query, val);
}
for val in &where_values {
query = bind_json_value(query, val);
}
for val in &check_values {
query = bind_json_value(query, val);
}
let mut guard = write_tx(gql_ctx, pool).await?;
let conn = guard.as_mut().expect("write_tx opens one");
let rows = query.fetch_all(&mut **conn).await.map_err(database_error)?;
refuse_unchecked_rows(&rows, !check_sql.is_empty())?;
let updated: Vec<Value> = rows
.iter()
.filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
.map(json_to_value)
.collect();
Ok(updated)
}
fn permission_check_sql(
gql_ctx: &GraphQLContext,
names: &crate::names::NameOverrides,
schema_name: &str,
table_name: &str,
verb: crate::role::Verb,
scope: &WhereScope<'_>,
start_param_idx: usize,
) -> Result<(String, Vec<serde_json::Value>), async_graphql::Error> {
let check = crate::role::write_check(&gql_ctx.caller(), names, schema_name, table_name, verb)
.map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
let Some(check) = check else {
return Ok((String::new(), Vec::new()));
};
let mut values = Vec::new();
let mut param_idx = start_param_idx;
let mut alias_counter = 0usize;
let condition = build_condition(
&check,
scope,
&mut param_idx,
&mut values,
&mut alias_counter,
)?;
Ok(match condition {
None => (String::new(), Vec::new()),
Some(sql) => (
format!(
", ({}) AS {}",
sql,
postrust_sql::escape_ident(CHECK_COLUMN)
),
values,
),
})
}
const CHECK_COLUMN: &str = "pgrst_permission_check";
fn refuse_unchecked_rows(
rows: &[sqlx::postgres::PgRow],
checked: bool,
) -> Result<(), async_graphql::Error> {
if !checked {
return Ok(());
}
use sqlx::Row;
let all_passed = rows
.iter()
.all(|row| row.try_get::<bool, _>(CHECK_COLUMN).unwrap_or(false));
match all_passed {
true => Ok(()),
false => Err(coded_error("permission-error", crate::role::CHECK_FAILED)),
}
}
#[allow(clippy::too_many_arguments)] async fn execute_delete(
pool: &PgPool,
gql_ctx: &GraphQLContext,
schema_name: &str,
table_name: &str,
type_name: &str,
where_clause: Option<serde_json::Value>,
returning: Option<async_graphql::SelectionField<'_>>,
relationships: &HashMap<String, Vec<RelationshipField>>,
names: &crate::names::NameOverrides,
max_rows: Option<i64>,
) -> Result<Vec<Value>, async_graphql::Error> {
use sqlx::Row;
trace!("Delete mutation for {}", table_name);
let guard = gql_ctx
.schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
let column_types = cache
.get_table(&qi)
.map(|table| exposed_column_types(table, names))
.unwrap_or_default();
let scope = WhereScope::table(schema_name, table_name, type_name, names)
.under_alias(WRITTEN_ROW)
.with_resolution(cache, relationships);
let (where_sql, mut values) = build_where_clause(where_clause.as_ref(), 1, &scope)?;
if where_sql.is_empty() {
return Err(async_graphql::Error::new(format!(
"delete on \"{}\" requires a `where` argument with at least one \
recognised condition; refusing to delete every row",
table_name
)));
}
let mut projection = String::new();
let renamed = cache
.get_table(&qi)
.and_then(|table| rename_projection(table, "src", names));
if let Some(returning) = returning {
let mut param_idx = values.len() + 1;
let mut alias_counter = 0usize;
let embeds = build_embed_expressions(
&gql_ctx.caller(),
cache,
relationships,
type_name,
"src",
returning.selection_set(),
max_rows,
&mut alias_counter,
&mut param_idx,
&mut values,
names,
)?;
let computed = match cache.get_table(&qi) {
Some(table) => computed_projections(
table,
returning.selection_set(),
"src",
names,
cache,
&mut param_idx,
&mut values,
)?,
None => Vec::new(),
};
for expression in &computed {
projection.push_str(", ");
projection.push_str(expression);
}
for (name, expression) in &embeds {
projection.push_str(", ");
projection.push_str(expression);
projection.push_str(" AS ");
projection.push_str(&postrust_sql::escape_ident(name));
}
}
let written = postrust_sql::escape_ident(WRITTEN_ROW);
let sql = if projection.is_empty() && renamed.is_none() {
format!(
"DELETE FROM {}.{} AS {} {} RETURNING {}",
postrust_sql::escape_ident(schema_name),
postrust_sql::escape_ident(table_name),
written,
where_sql,
row_json(&written, &column_types)
)
} else {
format!(
"WITH pgrst_deleted AS (DELETE FROM {}.{} AS {} {} RETURNING {}.*) \
SELECT {} FROM (SELECT {}{} FROM pgrst_deleted AS src) AS pgrst_r",
postrust_sql::escape_ident(schema_name),
postrust_sql::escape_ident(table_name),
written,
where_sql,
written,
row_json("pgrst_r", &column_types),
renamed.unwrap_or_else(|| "src.*".to_string()),
projection
)
};
trace!("Executing DELETE SQL: {}", sql);
let mut query = sqlx::query(AssertSqlSafe(sql));
for val in &values {
query = bind_json_value(query, val);
}
let mut guard = write_tx(gql_ctx, pool).await?;
let conn = guard.as_mut().expect("write_tx opens one");
let rows = query.fetch_all(&mut **conn).await.map_err(database_error)?;
let deleted: Vec<Value> = rows
.iter()
.filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
.map(json_to_value)
.collect();
Ok(deleted)
}
fn permission_predicate(
caller: &crate::role::Caller<'_>,
names: &crate::names::NameOverrides,
schema: &str,
table: &str,
) -> Result<Option<serde_json::Value>, async_graphql::Error> {
permission_filter(caller, names, schema, table, crate::role::Verb::Select)
}
fn permission_filter(
caller: &crate::role::Caller<'_>,
names: &crate::names::NameOverrides,
schema: &str,
table: &str,
verb: crate::role::Verb,
) -> Result<Option<serde_json::Value>, async_graphql::Error> {
crate::role::row_filter(caller, names, schema, table, verb)
.map(|filter| filter.map(|predicate| serde_json::json!({ PERMISSION: predicate })))
.map_err(|fault| coded_error(fault.code(), fault.to_string()))
}
const PERMISSION: &str = "_permission";
fn permission_limit(
caller: &crate::role::Caller<'_>,
names: &crate::names::NameOverrides,
schema: &str,
table: &str,
) -> Option<i64> {
crate::role::read_limit(caller, names, schema, table)
}
fn and_predicates(
request: Option<serde_json::Value>,
permission: Option<serde_json::Value>,
) -> Option<serde_json::Value> {
match (request, permission) {
(None, None) => None,
(Some(only), None) | (None, Some(only)) => Some(only),
(Some(request), Some(permission)) => {
Some(serde_json::json!({ "_and": [request, permission] }))
}
}
}
fn build_where_clause(
where_value: Option<&serde_json::Value>,
start_param_idx: usize,
scope: &WhereScope<'_>,
) -> Result<(String, Vec<serde_json::Value>), async_graphql::Error> {
let mut values: Vec<serde_json::Value> = Vec::new();
let mut param_idx = start_param_idx;
let mut alias_counter = 0usize;
let condition = match where_value {
Some(value) => build_condition(
value,
scope,
&mut param_idx,
&mut values,
&mut alias_counter,
)?,
None => None,
};
Ok(match condition {
Some(sql) => (format!("WHERE {}", sql), values),
None => (String::new(), values),
})
}
pub struct WhereScope<'a> {
qualified: postrust_core::api_request::QualifiedIdentifier,
sql_ref: String,
row_ref: String,
type_name: String,
root_ref: String,
root_qualified: postrust_core::api_request::QualifiedIdentifier,
names: &'a crate::names::NameOverrides,
resolution: Option<WhereResolution<'a>>,
caller: Option<crate::role::Caller<'a>>,
}
struct WhereResolution<'a> {
cache: &'a SchemaCache,
relationships: &'a HashMap<String, Vec<RelationshipField>>,
}
impl<'a> WhereScope<'a> {
pub fn table(
schema: &str,
table: &str,
type_name: &str,
names: &'a crate::names::NameOverrides,
) -> Self {
Self {
qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
sql_ref: format!(
"{}.{}",
postrust_sql::escape_ident(schema),
postrust_sql::escape_ident(table)
),
row_ref: postrust_sql::escape_ident(table),
type_name: type_name.to_string(),
root_ref: format!(
"{}.{}",
postrust_sql::escape_ident(schema),
postrust_sql::escape_ident(table)
),
root_qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
names,
resolution: None,
caller: None,
}
}
fn under_alias(mut self, alias: &str) -> Self {
if self.root_ref == self.sql_ref {
self.root_ref = postrust_sql::escape_ident(alias);
}
self.sql_ref = postrust_sql::escape_ident(alias);
self.row_ref = postrust_sql::escape_ident(alias);
self
}
fn with_resolution(
mut self,
cache: &'a SchemaCache,
relationships: &'a HashMap<String, Vec<RelationshipField>>,
) -> Self {
self.resolution = Some(WhereResolution {
cache,
relationships,
});
self
}
#[allow(clippy::too_many_arguments)]
fn for_alias(
schema: &str,
table: &str,
alias: &str,
type_name: &str,
cache: &'a SchemaCache,
relationships: &'a HashMap<String, Vec<RelationshipField>>,
names: &'a crate::names::NameOverrides,
) -> Self {
Self {
qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
sql_ref: postrust_sql::escape_ident(alias),
row_ref: postrust_sql::escape_ident(alias),
type_name: type_name.to_string(),
root_ref: postrust_sql::escape_ident(alias),
root_qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
names,
resolution: Some(WhereResolution {
cache,
relationships,
}),
caller: None,
}
}
fn aliased(
schema: &str,
table: &str,
alias: &str,
type_name: &str,
from: &WhereScope<'a>,
) -> Self {
Self {
qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
sql_ref: postrust_sql::escape_ident(alias),
row_ref: postrust_sql::escape_ident(alias),
type_name: type_name.to_string(),
root_ref: from.root_ref.clone(),
root_qualified: from.root_qualified.clone(),
names: from.names,
resolution: from.resolution.as_ref().map(|r| WhereResolution {
cache: r.cache,
relationships: r.relationships,
}),
caller: from.caller,
}
}
pub fn for_caller(mut self, caller: crate::role::Caller<'a>) -> Self {
self.caller = Some(caller);
self
}
fn as_system(&self) -> WhereScope<'a> {
WhereScope {
qualified: self.qualified.clone(),
sql_ref: self.sql_ref.clone(),
row_ref: self.row_ref.clone(),
type_name: self.type_name.clone(),
root_ref: self.sql_ref.clone(),
root_qualified: self.qualified.clone(),
names: self.names,
resolution: self.resolution.as_ref().map(|r| WhereResolution {
cache: r.cache,
relationships: r.relationships,
}),
caller: None,
}
}
fn column(&self, name: &str) -> String {
let column = self.column_name(name);
let plain = format!("{}.{}", self.sql_ref, postrust_sql::escape_ident(column));
let Some(cache) = self.resolution.as_ref().map(|r| r.cache) else {
return plain;
};
let Some(table) = cache.get_table(&self.qualified) else {
return plain;
};
if table.get_column(column).is_some() {
return plain;
}
let function = self
.names
.computed_source(&table.schema, &table.name, name)
.unwrap_or(name);
match table.get_computed_column(function) {
Some(definition) => {
computed_call(definition, &format!("{}.*", self.row_ref), &self.row_ref)
}
None => plain,
}
}
fn column_name<'n>(&'n self, name: &'n str) -> &'n str {
self.names
.column_source(&self.qualified.schema, &self.qualified.name, name)
.unwrap_or(name)
}
fn column_type(&self, name: &str) -> Option<String> {
let cache = self.resolution.as_ref()?.cache;
let table = cache.get_table(&self.qualified)?;
table
.get_column(self.column_name(name))
.map(|c| c.nominal_type.clone())
}
fn other_column(
&self,
operand: &serde_json::Value,
op: &str,
) -> Result<String, async_graphql::Error> {
use serde_json::Value;
let refused = |what: &str| {
async_graphql::Error::new(format!(
"the \"{}\" comparison takes the name of a column, {}",
op, what
))
};
let (against, qualified, name) = match operand {
Value::String(name) => (&self.sql_ref, &self.qualified, name.as_str()),
Value::Array(path) => match path.as_slice() {
[Value::String(name)] => (&self.sql_ref, &self.qualified, name.as_str()),
[Value::String(root), Value::String(name)] if root == "$" => {
(&self.root_ref, &self.root_qualified, name.as_str())
}
_ => return Err(refused("or [\"$\", \"a column\"]")),
},
_ => return Err(refused("written as a string")),
};
let column = self
.names
.column_source(&qualified.schema, &qualified.name, name)
.unwrap_or(name);
Ok(format!(
"{}.{}",
against,
postrust_sql::escape_ident(column)
))
}
fn relationship(&self, name: &str) -> Option<std::borrow::Cow<'a, RelationshipField>> {
let resolution = self.resolution.as_ref()?;
let published = resolution
.relationships
.get(&self.type_name)
.and_then(|fields| fields.iter().find(|r| r.name == name));
if let Some(found) = published {
return Some(std::borrow::Cow::Borrowed(found));
}
let table = resolution.cache.get_table(&self.qualified)?;
if let Some(declared) = self
.names
.declared_relationships(&table.schema, &table.name)
{
if let Some(entry) = declared.iter().find(|entry| entry.name == name) {
let target = entry
.target(&table.schema)
.map(|(_, target)| target)
.unwrap_or_default();
if let Some(field) = crate::schema::mapped_relationship_field(
entry,
table,
resolution.cache,
&table.schema,
&target,
&target,
&target,
) {
return Some(std::borrow::Cow::Owned(field));
}
if let Some(key) = entry.using.as_deref() {
if let Some(found) = resolution
.cache
.get_relationships(&self.qualified, &table.schema)
.into_iter()
.flatten()
.find(|rel| {
crate::schema::relationship_keys(rel)
.iter()
.any(|found| found == key)
})
{
return Some(std::borrow::Cow::Owned(
crate::schema::relationship::RelationshipField::from_relationship_named(
found,
&found.foreign_table().name,
Some(&entry.name),
),
));
}
}
}
}
resolution
.cache
.get_relationships(&self.qualified, &table.schema)?
.iter()
.find_map(|rel| {
let given = crate::schema::relationship_keys(rel)
.iter()
.find_map(|key| self.names.relationship(&table.schema, &table.name, key));
let field = RelationshipField::from_relationship_named(
rel,
&rel.foreign_table().name,
given,
);
(field.name == name).then_some(std::borrow::Cow::Owned(field))
})
}
}
fn column_comparison(op: &str) -> Option<&'static str> {
Some(match op {
"_ceq" => "=",
"_cne" => "<>",
"_cgt" => ">",
"_cgte" => ">=",
"_clt" => "<",
"_clte" => "<=",
_ => return None,
})
}
fn build_condition(
value: &serde_json::Value,
scope: &WhereScope<'_>,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
alias_counter: &mut usize,
) -> Result<Option<String>, async_graphql::Error> {
let serde_json::Value::Object(map) = value else {
return Ok(None);
};
let mut conditions: Vec<String> = Vec::new();
for (key, val) in map {
match key.as_str() {
"_and" | "_or" => {
let members = val.as_array().ok_or_else(|| {
async_graphql::Error::new(format!("\"{}\" takes a list of expressions", key))
})?;
let mut parts = Vec::new();
for member in members {
if let Some(sql) =
build_condition(member, scope, param_idx, values, alias_counter)?
{
parts.push(sql);
}
}
if parts.is_empty() {
if key == "_or" {
conditions.push("false".to_string());
}
continue;
}
let joiner = if key == "_and" { " AND " } else { " OR " };
conditions.push(format!("({})", parts.join(joiner)));
}
"_not" => {
if let Some(sql) = build_condition(val, scope, param_idx, values, alias_counter)? {
conditions.push(format!("NOT ({})", sql));
}
}
PERMISSION => {
if let Some(sql) =
build_condition(val, &scope.as_system(), param_idx, values, alias_counter)?
{
conditions.push(sql);
}
}
"_exists" => {
conditions.push(table_exists_sql(
val,
scope,
param_idx,
values,
alias_counter,
)?);
}
name if scope.relationship(name).is_some() => {
let relationship = scope.relationship(name).expect("just checked");
conditions.push(exists_sql(
&relationship,
val,
scope,
param_idx,
values,
alias_counter,
)?);
}
name if scope.column_type(name).is_none()
&& name
.strip_suffix("_aggregate")
.and_then(|rel| scope.relationship(rel))
.is_some_and(|rel| rel.is_list) =>
{
let relationship = name
.strip_suffix("_aggregate")
.and_then(|rel| scope.relationship(rel))
.expect("just checked");
conditions.push(aggregate_predicate_sql(
&relationship,
val,
scope,
param_idx,
values,
alias_counter,
)?);
}
column => {
let quoted = scope.column(column);
match val {
serde_json::Value::Object(ops) => {
let column_type = scope.column_type(column);
for (op, operand) in ops {
if let Some(other) = column_comparison(op) {
conditions.push(format!(
"{} {} {}",
quoted,
other,
scope.other_column(operand, op)?
));
continue;
}
conditions.push(comparison_sql(
"ed,
column,
column_type.as_deref(),
op,
operand,
param_idx,
values,
)?);
}
}
other => {
conditions.push(format!("{} = ${}", quoted, param_idx));
values.push(other.clone());
*param_idx += 1;
}
}
}
}
}
Ok(if conditions.is_empty() {
None
} else if conditions.len() == 1 {
conditions.pop()
} else {
Some(conditions.join(" AND "))
})
}
fn exists_sql(
relationship: &RelationshipField,
child_expression: &serde_json::Value,
scope: &WhereScope<'_>,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
alias_counter: &mut usize,
) -> Result<String, async_graphql::Error> {
let cache = scope.resolution.as_ref().map(|r| r.cache).ok_or_else(|| {
async_graphql::Error::new(format!(
"filtering on the relationship \"{}\" is not available here",
relationship.name
))
})?;
let plan = postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
if plan.junction.is_some() || (plan.function.is_none() && plan.columns.is_empty()) {
return Err(async_graphql::Error::new(format!(
"filtering on \"{}\" is not supported: it is reached through a junction \
rather than by a key",
relationship.name
)));
}
*alias_counter += 1;
let alias = format!("pgrst_rel_{}", alias_counter);
let child_scope = WhereScope::aliased(
&plan.foreign_schema,
&plan.foreign_table,
&alias,
&relationship.target_type,
scope,
);
let (source, mut correlation) = match &plan.function {
Some(function) => (
format!(
"{}.{}({})",
postrust_sql::escape_ident(&function.schema),
postrust_sql::escape_ident(&function.name),
scope.row_ref
),
Vec::new(),
),
None => {
let mut columns = Vec::with_capacity(plan.columns.len());
for (parent_column, child_column) in &plan.columns {
columns.push(format!(
"{} = {}",
scope.column(parent_column),
child_scope.column(child_column)
));
}
(
format!(
"{}.{}",
postrust_sql::escape_ident(&plan.foreign_schema),
postrust_sql::escape_ident(&plan.foreign_table)
),
columns,
)
}
};
let child_condition = build_condition(
child_expression,
&child_scope,
param_idx,
values,
alias_counter,
)?;
if let Some(sql) = child_condition {
correlation.push(format!("({})", sql));
}
if let Some(caller) = &child_scope.caller {
let permission = permission_predicate(
caller,
child_scope.names,
&plan.foreign_schema,
&plan.foreign_table,
)?;
if let Some(predicate) = permission {
if let Some(sql) =
build_condition(&predicate, &child_scope, param_idx, values, alias_counter)?
{
correlation.push(format!("({})", sql));
}
}
}
if correlation.is_empty() {
correlation.push("true".to_string());
}
Ok(format!(
"EXISTS (SELECT 1 FROM {} AS {} WHERE {})",
source,
postrust_sql::escape_ident(&alias),
correlation.join(" AND ")
))
}
fn table_exists_sql(
spec: &serde_json::Value,
scope: &WhereScope<'_>,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
alias_counter: &mut usize,
) -> Result<String, async_graphql::Error> {
let Some((schema, table)) = crate::role::exists_target(spec, &scope.qualified.schema) else {
return Err(async_graphql::Error::new(
"\"_exists\" needs a \"_table\" to look in",
));
};
let qualified = postrust_core::api_request::QualifiedIdentifier::new(&schema, &table);
if scope
.resolution
.as_ref()
.is_some_and(|r| r.cache.get_table(&qualified).is_none())
{
return Err(async_graphql::Error::new(format!(
"\"_exists\" names a table this server does not have: \"{}.{}\"",
schema, table
)));
}
*alias_counter += 1;
let alias = format!("pgrst_exists_{}", alias_counter);
let type_name = scope
.names
.base_name(&schema, &table)
.unwrap_or(&table)
.to_string();
let mut inner = WhereScope::aliased(&schema, &table, &alias, &type_name, scope);
inner.caller = None;
let condition = match spec.get("_where") {
Some(predicate) => build_condition(predicate, &inner, param_idx, values, alias_counter)?,
None => None,
};
Ok(format!(
"EXISTS (SELECT 1 FROM {}.{} AS {} WHERE {})",
postrust_sql::escape_ident(&schema),
postrust_sql::escape_ident(&table),
postrust_sql::escape_ident(&alias),
condition.unwrap_or_else(|| "true".to_string())
))
}
fn aggregate_predicate_sql(
relationship: &RelationshipField,
expression: &serde_json::Value,
scope: &WhereScope<'_>,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
alias_counter: &mut usize,
) -> Result<String, async_graphql::Error> {
let serde_json::Value::Object(over) = expression else {
return Ok("true".to_string());
};
let cache = scope.resolution.as_ref().map(|r| r.cache).ok_or_else(|| {
async_graphql::Error::new(format!(
"filtering on the relationship \"{}\" is not available here",
relationship.name
))
})?;
let plan = postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
if plan.junction.is_some() || (plan.function.is_none() && plan.columns.is_empty()) {
return Err(async_graphql::Error::new(format!(
"aggregating over \"{}\" is not supported: it is reached through a junction \
rather than by a key",
relationship.name
)));
}
let mut conditions: Vec<String> = Vec::new();
for (function, spec) in over {
if spec.is_null() {
continue;
}
let serde_json::Value::Object(spec) = spec else {
continue;
};
let Some(predicate) = spec.get("predicate").filter(|p| !p.is_null()) else {
return Err(validation_error(format!(
"\"{}\" over \"{}\" needs a predicate",
function, relationship.name
)));
};
*alias_counter += 1;
let alias = format!("pgrst_agg_{}", alias_counter);
let child_scope = WhereScope::aliased(
&plan.foreign_schema,
&plan.foreign_table,
&alias,
&relationship.target_type,
scope,
);
let (source, mut correlation) = match &plan.function {
Some(routine) => (
format!(
"{}.{}({})",
postrust_sql::escape_ident(&routine.schema),
postrust_sql::escape_ident(&routine.name),
scope.row_ref
),
Vec::new(),
),
None => {
let mut columns = Vec::with_capacity(plan.columns.len());
for (parent_column, child_column) in &plan.columns {
columns.push(format!(
"{} = {}",
scope.column(parent_column),
child_scope.column(child_column)
));
}
(
format!(
"{}.{}",
postrust_sql::escape_ident(&plan.foreign_schema),
postrust_sql::escape_ident(&plan.foreign_table)
),
columns,
)
}
};
if let Some(filter) = spec.get("filter").filter(|f| !f.is_null()) {
if let Some(sql) =
build_condition(filter, &child_scope, param_idx, values, alias_counter)?
{
correlation.push(format!("({})", sql));
}
}
if correlation.is_empty() {
correlation.push("true".to_string());
}
let arguments = spec.get("arguments").filter(|a| !a.is_null());
let distinct = matches!(spec.get("distinct"), Some(serde_json::Value::Bool(true)));
let (aggregate, operand_type) = match function.as_str() {
"count" => {
let columns: Vec<String> = match arguments {
Some(serde_json::Value::Array(items)) => items
.iter()
.filter_map(|item| item.as_str())
.map(|column| child_scope.column(column))
.collect(),
Some(serde_json::Value::String(one)) => vec![child_scope.column(one)],
_ => Vec::new(),
};
let counted = match columns.len() {
0 => "*".to_string(),
1 => columns.into_iter().next().unwrap_or_default(),
_ => format!("({})", columns.join(", ")),
};
let counted = match distinct && counted != "*" {
true => format!("DISTINCT {}", counted),
false => counted,
};
(format!("count({})", counted), "integer")
}
"bool_and" | "bool_or" => {
let Some(column) = arguments.and_then(|a| a.as_str()) else {
return Err(validation_error(format!(
"\"{}\" over \"{}\" needs the column to fold",
function, relationship.name
)));
};
(
format!("{}({})", function, child_scope.column(column)),
"boolean",
)
}
other => {
return Err(validation_error(format!(
"\"{}\" is not an aggregate a predicate can be written over",
other
)))
}
};
let subselect = format!(
"(SELECT {} FROM {} AS {} WHERE {})",
aggregate,
source,
postrust_sql::escape_ident(&alias),
correlation.join(" AND ")
);
let serde_json::Value::Object(operators) = predicate else {
return Err(validation_error(format!(
"the predicate on \"{}\" is not a comparison",
function
)));
};
for (operator, operand) in operators {
conditions.push(comparison_sql(
&subselect,
function,
Some(operand_type),
operator,
operand,
param_idx,
values,
)?);
}
}
Ok(match conditions.is_empty() {
true => "true".to_string(),
false => format!("({})", conditions.join(" AND ")),
})
}
#[allow(clippy::too_many_arguments)]
fn postgis_sql(
quoted: &str,
column_type: Option<&str>,
function: &str,
op: &str,
operand: &serde_json::Value,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
) -> Result<String, async_graphql::Error> {
let is_geography = column_type == Some("geography");
let mut shape = |value: &serde_json::Value, param_idx: &mut usize| -> String {
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(value.clone());
match is_geography {
true => format!("ST_GeomFromGeoJSON({})::geography", placeholder),
false => format!("ST_GeomFromGeoJSON({})", placeholder),
}
};
match op {
"_st_d_within" | "_st_3d_d_within" => {
let serde_json::Value::Object(spec) = operand else {
return Err(async_graphql::Error::new(format!(
"\"{}\" takes {{distance, from}}",
op
)));
};
let Some(from) = spec.get("from") else {
return Err(async_graphql::Error::new(format!(
"\"{}\" needs a shape to measure from",
op
)));
};
let Some(distance) = spec.get("distance") else {
return Err(async_graphql::Error::new(format!(
"\"{}\" needs a distance",
op
)));
};
let from_sql = shape(from, param_idx);
let distance_sql = format!("${}", param_idx);
*param_idx += 1;
values.push(distance.clone());
let spheroid = match (is_geography, spec.get("use_spheroid")) {
(true, Some(use_spheroid)) if !use_spheroid.is_null() => {
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(use_spheroid.clone());
format!(", {}::boolean", placeholder)
}
_ => String::new(),
};
Ok(format!(
"{}({}, {}, {}::float8{})",
function, quoted, from_sql, distance_sql, spheroid
))
}
"_st_intersects_rast" => {
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(operand.clone());
Ok(format!("{}({}, {}::raster)", function, quoted, placeholder))
}
"_st_intersects_geom_nband" | "_st_intersects_nband_geom" => {
let serde_json::Value::Object(spec) = operand else {
return Err(async_graphql::Error::new(format!(
"\"{}\" takes an object",
op
)));
};
let Some(geometry) = spec.get("geommin") else {
return Err(async_graphql::Error::new(format!(
"\"{}\" needs a shape",
op
)));
};
let band = spec.get("nband").filter(|v| !v.is_null());
let geometry_sql = shape(geometry, param_idx);
Ok(match band {
None => format!("{}({}, {})", function, quoted, geometry_sql),
Some(band) => {
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(band.clone());
format!(
"{}({}, {}::int, {})",
function, quoted, placeholder, geometry_sql
)
}
})
}
_ => {
let other = shape(operand, param_idx);
Ok(format!("{}({}, {})", function, quoted, other))
}
}
}
fn sql_array(
items: &[serde_json::Value],
cast: &str,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
) -> String {
if items.is_empty() {
return format!("ARRAY[]::{}", cast);
}
let placeholders: Vec<String> = items
.iter()
.map(|item| {
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(item.clone());
placeholder
})
.collect();
format!("ARRAY[{}]::{}", placeholders.join(", "), cast)
}
#[allow(clippy::too_many_arguments)]
fn comparison_sql(
quoted: &str,
column: &str,
column_type: Option<&str>,
op: &str,
operand: &serde_json::Value,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
) -> Result<String, async_graphql::Error> {
if op == "_cast" {
let serde_json::Value::Object(casts) = operand else {
return Err(async_graphql::Error::new(
"_cast takes an object naming the type to compare as",
));
};
let mut conditions = Vec::new();
for (target, comparisons) in casts {
let pg_type = match target.as_str() {
"geometry" | "geography" => target.as_str(),
"String" => "text",
_ => {
return Err(async_graphql::Error::new(format!(
"cannot compare \"{}\" as \"{}\"",
column, target
)))
}
};
let serde_json::Value::Object(ops) = comparisons else {
continue;
};
let cast = format!("{}::{}", quoted, pg_type);
for (nested_op, nested_operand) in ops {
conditions.push(comparison_sql(
&cast,
column,
Some(pg_type),
nested_op,
nested_operand,
param_idx,
values,
)?);
}
}
return Ok(match conditions.len() {
0 => "true".to_string(),
1 => conditions.pop().expect("just counted"),
_ => format!("({})", conditions.join(" AND ")),
});
}
if let Some(function) = crate::input::bool_exp::postgis_function(op) {
return postgis_sql(
quoted,
column_type,
function,
op,
operand,
param_idx,
values,
);
}
if let Some((operator, cast)) = crate::input::bool_exp::ltree_operator(op) {
if cast.ends_with("[]") {
let items = operand.as_array().ok_or_else(|| {
async_graphql::Error::new(format!(
"the \"{}\" comparison on \"{}\" takes a list",
op, column
))
})?;
return Ok(format!(
"{} {} {}",
quoted,
operator,
sql_array(items, cast, param_idx, values)
));
}
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(operand.clone());
return Ok(format!("{} {} {}::{}", quoted, operator, placeholder, cast));
}
if let Some(function) = match op {
"_jsonb_path_exists" => Some("jsonb_path_exists"),
"_jsonb_path_match" => Some("jsonb_path_match"),
_ => None,
} {
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(operand.clone());
return Ok(format!(
"{}({}, {}::jsonpath)",
function, quoted, placeholder
));
}
let binary = match op {
"_eq" => Some("="),
"_neq" => Some("<>"),
"_gt" => Some(">"),
"_gte" => Some(">="),
"_lt" => Some("<"),
"_lte" => Some("<="),
"_like" => Some("LIKE"),
"_nlike" => Some("NOT LIKE"),
"_ilike" => Some("ILIKE"),
"_nilike" => Some("NOT ILIKE"),
"_similar" => Some("SIMILAR TO"),
"_nsimilar" => Some("NOT SIMILAR TO"),
"_regex" => Some("~"),
"_nregex" => Some("!~"),
"_iregex" => Some("~*"),
"_niregex" => Some("!~*"),
"_contains" => Some("@>"),
"_contained_in" => Some("<@"),
"_has_key" => Some("?"),
_ => None,
};
if let Some(sql_op) = binary {
let placeholder = format!("${}", param_idx);
*param_idx += 1;
values.push(match op {
"_contains" | "_contained_in" => match operand {
serde_json::Value::String(_)
| serde_json::Value::Number(_)
| serde_json::Value::Bool(_) => serde_json::Value::String(operand.to_string()),
other => other.clone(),
},
_ => operand.clone(),
});
let cast = match op {
"_contains" | "_contained_in" => "::jsonb".to_string(),
"_has_key" => "::text".to_string(),
_ if operand.is_null() => match column_type {
Some(pg_type) => format!("::{}", pg_type),
None => String::new(),
},
_ if operand.is_string() => match column_type {
Some(pg_type) => format!("::{}", pg_type),
None => String::new(),
},
_ => String::new(),
};
return Ok(format!("{} {} {}{}", quoted, sql_op, placeholder, cast));
}
match op {
"_is_null" => Ok(if operand.as_bool().unwrap_or(false) {
format!("{} IS NULL", quoted)
} else {
format!("{} IS NOT NULL", quoted)
}),
"_in" | "_nin" => {
let items = operand.as_array().ok_or_else(|| {
async_graphql::Error::new(format!(
"the \"{}\" comparison on \"{}\" requires a list of values",
op, column
))
})?;
if items.is_empty() {
return Ok(if op == "_in" { "false" } else { "true" }.to_string());
}
let mut placeholders = Vec::with_capacity(items.len());
for item in items {
let cast = match (item.is_null() || item.is_string(), column_type) {
(true, Some(pg_type)) => format!("::{}", pg_type),
_ => String::new(),
};
placeholders.push(format!("${}{}", param_idx, cast));
values.push(item.clone());
*param_idx += 1;
}
Ok(format!(
"{} {} ({})",
quoted,
if op == "_in" { "IN" } else { "NOT IN" },
placeholders.join(", ")
))
}
"_has_keys_any" | "_has_keys_all" => {
let items = operand.as_array().ok_or_else(|| {
async_graphql::Error::new(format!(
"the \"{}\" comparison on \"{}\" requires a list of keys",
op, column
))
})?;
Ok(format!(
"{} {} {}",
quoted,
if op == "_has_keys_any" { "?|" } else { "?&" },
sql_array(items, "text[]", param_idx, values)
))
}
other => Err(async_graphql::Error::new(format!(
"unsupported comparison \"{}\" on \"{}\"",
other, column
))),
}
}
#[allow(clippy::too_many_arguments)] async fn build_order_by_clause(
ctx: &ResolverContext<'_>,
schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
schema_name: &str,
table_name: &str,
type_name: &str,
relationships: &HashMap<String, Vec<RelationshipField>>,
reference: &str,
names: &crate::names::NameOverrides,
) -> Result<String, async_graphql::Error> {
let Ok(order_arg) = ctx.args.try_get("order_by") else {
return Ok(String::new());
};
let value = accessor_to_json(&order_arg);
let entries: Vec<&serde_json::Value> = match &value {
serde_json::Value::Array(items) => items.iter().collect(),
serde_json::Value::Object(_) => vec![&value],
serde_json::Value::Null => return Ok(String::new()),
_ => {
return Err(async_graphql::Error::new(
"order_by takes an object or a list of them",
))
}
};
if entries.is_empty() {
return Ok(String::new());
}
let guard = schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
let table = cache
.get_table(&qi)
.ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
let mut terms = Vec::new();
let mut alias_counter = 0usize;
for entry in entries {
order_terms_into(
entry,
cache,
relationships,
type_name,
table,
reference,
names,
&mut alias_counter,
&mut terms,
)?;
}
if terms.is_empty() {
return Ok(String::new());
}
Ok(format!(" ORDER BY {}", terms.join(", ")))
}
#[allow(clippy::too_many_arguments)]
fn order_terms_into(
entry: &serde_json::Value,
cache: &SchemaCache,
relationships: &HashMap<String, Vec<RelationshipField>>,
type_name: &str,
table: &postrust_core::schema_cache::Table,
reference: &str,
names: &crate::names::NameOverrides,
alias_counter: &mut usize,
terms: &mut Vec<String>,
) -> Result<(), async_graphql::Error> {
let serde_json::Value::Object(map) = entry else {
return Err(async_graphql::Error::new(
"each order_by entry is an object mapping a column to a direction",
));
};
let available: &[RelationshipField] = relationships
.get(type_name)
.map(|r| r.as_slice())
.unwrap_or(&[]);
for (key, value) in map {
if value.is_null() {
continue;
}
if let Some(name) = value.as_str() {
let sql = crate::input::order_by::direction_sql(name).ok_or_else(|| {
async_graphql::Error::new(format!(
"\"{}\" is not a sort direction; expected one of asc, desc, \
asc_nulls_first, asc_nulls_last, desc_nulls_first, desc_nulls_last",
name
))
})?;
let column = table_column_for(names, table, key);
if table.get_column(column).is_some() {
terms.push(format!(
"{}.{} {}",
reference,
postrust_sql::escape_ident(column),
sql
));
continue;
}
let function = names
.computed_source(&table.schema, &table.name, key)
.unwrap_or(key);
if let Some(definition) = table.get_computed_column(function) {
terms.push(format!(
"{} {}",
computed_call(definition, &format!("{}.*", reference), reference),
sql
));
continue;
}
return Err(async_graphql::Error::new(format!(
"cannot order by unknown column \"{}\" on \"{}\"",
key, table.name
)));
}
if let Some(rel) = key
.strip_suffix("_aggregate")
.and_then(|name| available.iter().find(|r| r.name == name && r.is_list))
{
aggregate_order_terms(value, rel, cache, reference, alias_counter, terms)?;
continue;
}
if let Some(rel) = available.iter().find(|r| r.name == *key && !r.is_list) {
let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
if plan.columns.is_empty() {
return Err(async_graphql::Error::new(format!(
"cannot order by \"{}\": it is not reached by a key",
key
)));
}
*alias_counter += 1;
let alias = format!("pgrst_ord_{}", alias_counter);
let quoted_alias = postrust_sql::escape_ident(&alias);
let correlation = plan
.columns
.iter()
.map(|(local, foreign)| {
format!(
"{} = {}.{}",
format_args!("{}.{}", reference, postrust_sql::escape_ident(local)),
quoted_alias,
postrust_sql::escape_ident(foreign)
)
})
.collect::<Vec<_>>()
.join(" AND ");
let target_qi = postrust_core::api_request::QualifiedIdentifier::new(
&plan.foreign_schema,
&plan.foreign_table,
);
let Some(target) = cache.get_table(&target_qi) else {
return Err(async_graphql::Error::new(format!(
"unknown table \"{}\"",
plan.foreign_table
)));
};
let mut nested = Vec::new();
order_terms_into(
value,
cache,
relationships,
&rel.target_type,
target,
"ed_alias,
names,
alias_counter,
&mut nested,
)?;
for term in nested {
let (expression, direction) = term
.rsplit_once(' ')
.map(|(e, d)| (e.to_string(), d.to_string()))
.unwrap_or((term.clone(), String::new()));
terms.push(format!(
"(SELECT {} FROM {}.{} AS {} WHERE {}) {}",
expression,
postrust_sql::escape_ident(&plan.foreign_schema),
postrust_sql::escape_ident(&plan.foreign_table),
quoted_alias,
correlation,
direction
));
}
continue;
}
return Err(async_graphql::Error::new(format!(
"cannot order by \"{}\" on \"{}\"",
key, table.name
)));
}
Ok(())
}
fn aggregate_order_terms(
value: &serde_json::Value,
rel: &RelationshipField,
cache: &SchemaCache,
reference: &str,
alias_counter: &mut usize,
terms: &mut Vec<String>,
) -> Result<(), async_graphql::Error> {
let serde_json::Value::Object(spec) = value else {
return Err(async_graphql::Error::new(
"ordering by an aggregate takes an object, such as {count: desc}",
));
};
let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
if plan.columns.is_empty() && plan.function.is_none() {
return Err(async_graphql::Error::new(format!(
"cannot order by an aggregate of \"{}\": it is not reached by a key",
rel.name
)));
}
if plan.function.is_some() && !rel.arguments.is_empty() {
return Err(async_graphql::Error::new(format!(
"cannot order by an aggregate of \"{}\": it takes arguments, and an \
ordering has nowhere to write them",
rel.name
)));
}
*alias_counter += 1;
let alias = postrust_sql::escape_ident(&format!("pgrst_ord_{}", alias_counter));
let (source, correlation) = match &plan.function {
Some(function) => {
let session = plan.row_argument.as_ref().and_then(|row_argument| {
computed_session_argument(cache, function, row_argument)
.map(|session| (row_argument.clone(), session))
});
let call = match session {
Some((row_argument, session)) => format!(
"{} => {}, {} => {}",
postrust_sql::escape_ident(&row_argument),
reference,
postrust_sql::escape_ident(&session),
SESSION_ARGUMENT
),
None => reference.to_string(),
};
(
format!(
"{}.{}({})",
postrust_sql::escape_ident(&function.schema),
postrust_sql::escape_ident(&function.name),
call
),
"true".to_string(),
)
}
None => (
format!(
"{}.{}",
postrust_sql::escape_ident(&plan.foreign_schema),
postrust_sql::escape_ident(&plan.foreign_table)
),
plan.columns
.iter()
.map(|(local, foreign)| {
format!(
"{}.{} = {}.{}",
reference,
postrust_sql::escape_ident(local),
alias,
postrust_sql::escape_ident(foreign)
)
})
.collect::<Vec<_>>()
.join(" AND "),
),
};
let mut wanted: Vec<(String, String)> = Vec::new();
for (function, argument) in spec {
if argument.is_null() {
continue;
}
if function == "count" {
let name = argument.as_str().unwrap_or_default();
let Some(direction) = crate::input::order_by::direction_sql(name) else {
return Err(async_graphql::Error::new(format!(
"\"{}\" is not a sort direction",
name
)));
};
wanted.push(("count(*)".to_string(), direction.to_string()));
continue;
}
let serde_json::Value::Object(columns) = argument else {
continue;
};
for (column, direction) in columns {
if direction.is_null() {
continue;
}
let name = direction.as_str().unwrap_or_default();
let Some(direction) = crate::input::order_by::direction_sql(name) else {
return Err(async_graphql::Error::new(format!(
"\"{}\" is not a sort direction",
name
)));
};
wanted.push((
format!(
"{}({}.{})",
function,
alias,
postrust_sql::escape_ident(column)
),
direction.to_string(),
));
}
}
for (expression, direction) in wanted {
terms.push(format!(
"(SELECT {} FROM {} AS {} WHERE {}) {}",
expression, source, alias, correlation, direction
));
}
Ok(())
}
fn distinct_on_clause(
distinct_on: &[String],
written: Option<&str>,
) -> Result<(String, Option<String>), async_graphql::Error> {
if distinct_on.is_empty() {
return Ok((String::new(), written.map(str::to_string)));
}
let terms: Vec<&str> = written
.map(|written| written.split(", ").collect())
.unwrap_or_default();
let column_of = |term: &str| {
let expression = term.split_whitespace().next().unwrap_or(term);
expression
.rsplit_once('.')
.map(|(_, name)| name)
.unwrap_or(expression)
.to_string()
};
let leading: Vec<String> = terms
.iter()
.take(distinct_on.len())
.map(|term| column_of(term))
.collect();
if !terms.is_empty()
&& (leading.len() != distinct_on.len()
|| distinct_on
.iter()
.any(|column| !leading.contains(&column_of(column))))
{
return Err(validation_error(
"\"distinct_on\" columns must match initial \"order_by\" columns",
));
}
let order = match terms.is_empty() {
true => Some(distinct_on.join(", ")),
false => written.map(str::to_string),
};
Ok((format!("DISTINCT ON ({}) ", distinct_on.join(", ")), order))
}
async fn build_distinct_on(
ctx: &ResolverContext<'_>,
schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
schema_name: &str,
table_name: &str,
names: &crate::names::NameOverrides,
) -> Result<Vec<String>, async_graphql::Error> {
let Ok(arg) = ctx.args.try_get("distinct_on") else {
return Ok(Vec::new());
};
let value = accessor_to_json(&arg);
let requested: Vec<String> = match &value {
serde_json::Value::Array(items) => items
.iter()
.filter_map(|i| i.as_str().map(|s| s.to_string()))
.collect(),
serde_json::Value::String(one) => vec![one.clone()],
_ => Vec::new(),
};
if requested.is_empty() {
return Ok(Vec::new());
}
let guard = schema_cache
.get()
.await
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let cache = guard
.as_ref()
.ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
let table = cache
.get_table(&qi)
.ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
let mut quoted = Vec::with_capacity(requested.len());
for field in requested {
let name = table_column_for(names, table, &field).to_string();
if table.get_column(&name).is_none() {
return Err(async_graphql::Error::new(format!(
"cannot take distinct on unknown column \"{}\" of \"{}\"",
field, table_name
)));
}
quoted.push(postrust_sql::escape_ident(&name));
}
Ok(quoted)
}
#[allow(clippy::too_many_arguments)] fn computed_projections<'a>(
table: &postrust_core::schema_cache::Table,
fields: impl IntoIterator<Item = SelectionField<'a>>,
row_reference: &str,
names: &crate::names::NameOverrides,
schema_cache: &SchemaCache,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
) -> Result<Vec<String>, async_graphql::Error> {
let mut projections = Vec::new();
for field in fields {
let name = field.name();
if table
.get_column(table_column_for(names, table, name))
.is_some()
{
continue;
}
let function = names
.computed_source(&table.schema, &table.name, name)
.unwrap_or(name);
let Some(computed) = table.get_computed_column(function) else {
continue;
};
let arguments = crate::schema::computed_caller_arguments(computed, schema_cache);
let given = match arguments.is_empty() {
true => None,
false => embed_arguments(field).remove("args"),
};
projections.push(format!(
"{} AS {}",
computed_column_call(
computed,
row_reference,
name,
&arguments,
given.as_ref(),
param_idx,
values,
)?,
postrust_sql::escape_ident(name)
));
}
Ok(projections)
}
fn computed_column_call(
computed: &postrust_core::schema_cache::ComputedColumn,
row: &str,
field_name: &str,
arguments: &[(String, String, bool)],
given: Option<&serde_json::Value>,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
) -> Result<String, async_graphql::Error> {
if arguments.is_empty() {
return Ok(computed_call(computed, row, row));
}
let Some(row_argument) = &computed.row_argument else {
return Ok(computed_call(computed, row, row));
};
let function = format!(
"{}.{}",
postrust_sql::escape_ident(&computed.function.schema),
postrust_sql::escape_ident(&computed.function.name)
);
let mut passed = vec![format!(
"{} => {}",
postrust_sql::escape_ident(row_argument),
row
)];
if let Some(session) = &computed.session_argument {
passed.push(format!(
"{} => {}",
postrust_sql::escape_ident(session),
SESSION_ARGUMENT
));
}
for (name, pg_type, required) in arguments {
match given.and_then(|given| given.get(name)) {
None if *required => {
return Err(async_graphql::Error::new(format!(
"{} needs the argument \"{}\"",
field_name, name
)))
}
None => continue,
Some(value) => {
passed.push(format!(
"{} => ${}::{}",
postrust_sql::escape_ident(name),
param_idx,
pg_type
));
*param_idx += 1;
values.push(value.clone());
}
}
}
Ok(format!("{}({})", function, passed.join(", ")))
}
fn count_expression(
field: async_graphql::SelectionField<'_>,
table: (&str, &str),
alias: Option<&str>,
names: &crate::names::NameOverrides,
) -> String {
let arguments = embed_arguments(field);
let requested = match arguments.get("columns") {
Some(serde_json::Value::Array(items)) => items
.iter()
.filter_map(|item| item.as_str())
.collect::<Vec<_>>(),
Some(serde_json::Value::String(one)) => vec![one.as_str()],
_ => Vec::new(),
};
if requested.is_empty() {
return "count(*)".to_string();
}
let columns: Vec<String> = requested
.into_iter()
.map(|column| {
let (schema, name) = table;
let source = column_for(names, schema, name, column);
match alias {
Some(alias) => format!(
"{}.{}",
postrust_sql::escape_ident(alias),
postrust_sql::escape_ident(source)
),
None => postrust_sql::escape_ident(source),
}
})
.collect();
let counted = match columns.len() {
1 => columns.into_iter().next().unwrap_or_default(),
_ => format!("({})", columns.join(", ")),
};
let distinct = matches!(
arguments.get("distinct"),
Some(serde_json::Value::Bool(true))
);
match distinct {
true => format!("count(DISTINCT {})", counted),
false => format!("count({})", counted),
}
}
#[allow(clippy::too_many_arguments)] fn nested_aggregate_select(
caller: &crate::role::Caller<'_>,
selection: async_graphql::SelectionField<'_>,
child_alias: &str,
schema_cache: &SchemaCache,
relationships: &HashMap<String, Vec<RelationshipField>>,
target_type: &str,
child_table: Option<&postrust_core::schema_cache::Table>,
max_rows: Option<i64>,
alias_counter: &mut usize,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
names: &crate::names::NameOverrides,
) -> Result<String, async_graphql::Error> {
use crate::schema::aggregate as agg;
let mut parts: Vec<String> = Vec::new();
for child in selection.selection_set() {
match child.name() {
"aggregate" => {
let of_table = child_table
.map(|table| (table.schema.as_str(), table.name.as_str()))
.unwrap_or(("", ""));
let mut counts: Vec<(String, String)> = Vec::new();
let mut wanted: BTreeMap<&str, Vec<String>> = BTreeMap::new();
for function in child.selection_set() {
if function.name() == "count" {
counts.push((
function.alias().unwrap_or("count").to_string(),
count_expression(function, of_table, Some(child_alias), names),
));
continue;
}
let sql_function = agg::NUMERIC_AGGREGATES
.iter()
.chain(agg::ORDERED_AGGREGATES.iter())
.find(|(name, _)| *name == function.name())
.map(|(name, _)| *name);
let Some(sql_function) = sql_function else {
continue;
};
let columns = wanted.entry(sql_function).or_default();
for column in function.selection_set() {
let name = column.name().to_string();
if name != "__typename" && !columns.contains(&name) {
columns.push(name);
}
}
}
let mut fields = match counts.iter().any(|(key, _)| key == "count") {
true => Vec::new(),
false => vec!["'count', count(*)".to_string()],
};
for (key, expression) in &counts {
fields.push(format!(
"{}, {}",
postrust_sql::quote_literal(key),
expression
));
}
for (sql_function, named) in &wanted {
let columns: Vec<String> = named
.iter()
.map(|column| {
let source = child_table
.map(|table| table_column_for(names, table, column))
.unwrap_or(column.as_str());
format!(
"{}, {}({}.{})",
postrust_sql::quote_literal(column),
sql_function,
postrust_sql::escape_ident(child_alias),
postrust_sql::escape_ident(source)
)
})
.collect();
fields.push(format!(
"'{}', json_build_object({})",
sql_function,
columns.join(", ")
));
}
parts.push(format!(
"json_build_object({}) AS {}",
fields.join(", "),
postrust_sql::escape_ident("aggregate")
));
}
"nodes" => {
let embeds = build_embed_expressions(
caller,
schema_cache,
relationships,
target_type,
child_alias,
child.selection_set(),
max_rows,
alias_counter,
param_idx,
values,
names,
)?;
let mut columns: Vec<String> = Vec::new();
for column in child.selection_set() {
let name = column.name();
if name == "__typename" || embeds.iter().any(|(field, _)| field == name) {
continue;
}
columns.push(format!(
"{}, {}",
postrust_sql::quote_literal(name),
column_expression(child_table, child_alias, name, names)
));
}
for (field, expression) in &embeds {
columns.push(format!(
"{}, {}",
postrust_sql::quote_literal(field),
expression
));
}
let row = if columns.is_empty() {
format!("row_to_json({})", postrust_sql::escape_ident(child_alias))
} else {
format!("json_build_object({})", columns.join(", "))
};
parts.push(format!(
"COALESCE(json_agg({}), '[]'::json) AS {}",
row,
postrust_sql::escape_ident("nodes")
));
}
_ => {}
}
}
if parts.is_empty() {
parts.push("count(*) AS pgrst_empty".to_string());
}
Ok(parts.join(", "))
}
fn column_expression(
table: Option<&postrust_core::schema_cache::Table>,
alias: &str,
name: &str,
names: &crate::names::NameOverrides,
) -> String {
let column = table
.map(|t| table_column_for(names, t, name))
.unwrap_or(name);
let plain = format!(
"{}.{}",
postrust_sql::escape_ident(alias),
postrust_sql::escape_ident(column)
);
let computed = table
.filter(|t| t.get_column(column).is_none())
.and_then(|t| {
let function = names
.computed_source(&t.schema, &t.name, name)
.unwrap_or(name);
t.get_computed_column(function)
});
match computed {
Some(definition) => computed_call(
definition,
&format!("{}.*", postrust_sql::escape_ident(alias)),
&postrust_sql::escape_ident(alias),
),
None => plain,
}
}
fn order_terms(
order: &serde_json::Value,
schema_cache: &SchemaCache,
schema_name: &str,
table_name: &str,
alias: &str,
names: &crate::names::NameOverrides,
) -> Result<Option<String>, async_graphql::Error> {
let entries: Vec<&serde_json::Value> = match order {
serde_json::Value::Array(items) => items.iter().collect(),
serde_json::Value::Object(_) => vec![order],
_ => return Ok(None),
};
let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
let table = schema_cache
.get_table(&qi)
.ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
let mut terms = Vec::new();
for entry in entries {
let serde_json::Value::Object(map) = entry else {
continue;
};
for (field, direction) in map {
if direction.is_null() {
continue;
}
let column = table_column_for(names, table, field);
if table.get_column(column).is_none() {
return Err(async_graphql::Error::new(format!(
"cannot order by unknown column \"{}\" on \"{}\"",
field, table_name
)));
}
let name = direction.as_str().unwrap_or_default();
let sql = crate::input::order_by::direction_sql(name).ok_or_else(|| {
async_graphql::Error::new(format!("\"{}\" is not a sort direction", name))
})?;
terms.push(format!(
"{}.{} {}",
postrust_sql::escape_ident(alias),
postrust_sql::escape_ident(column),
sql
));
}
}
Ok(if terms.is_empty() {
None
} else {
Some(terms.join(", "))
})
}
fn always_present(rel: &RelationshipField, table: &postrust_core::schema_cache::Table) -> bool {
use postrust_core::schema_cache::{Cardinality, Relationship};
let Relationship::ForeignKey { cardinality, .. } = &rel.relationship else {
return false;
};
match cardinality {
Cardinality::M2O { .. } => {}
Cardinality::O2O {
is_parent: false, ..
} => {}
_ => return false,
}
let columns = cardinality.columns();
!columns.is_empty()
&& columns.iter().all(|(local, _)| {
table
.get_column(local)
.is_some_and(|column| !column.nullable)
})
}
fn child_holds_the_key(relationship: &postrust_core::schema_cache::Relationship) -> bool {
use postrust_core::schema_cache::{Cardinality, Relationship};
match relationship {
Relationship::ForeignKey { cardinality, .. } => match cardinality {
Cardinality::M2O { .. } => false,
Cardinality::O2O { is_parent, .. } => *is_parent,
_ => true,
},
Relationship::Computed { .. } => true,
}
}
fn computed_session_argument(
cache: &SchemaCache,
function: &postrust_core::api_request::QualifiedIdentifier,
row_argument: &str,
) -> Option<String> {
let _ = row_argument;
cache
.routines
.get(function)
.into_iter()
.flatten()
.find(|routine| routine.name == function.name)?
.params
.iter()
.find(|param| {
param.name == "hasura_session" && matches!(param.param_type.as_str(), "json" | "jsonb")
})
.map(|param| param.name.clone())
}
fn computed_args_type_name(type_name: &str, field: &str) -> String {
format!("{}_{}_args", type_name, field)
}
fn computed_arguments(
relationship: &RelationshipField,
plan: &postrust_core::embed::EmbedPlan,
parent_row: &str,
given: Option<&serde_json::Value>,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
schema_cache: &SchemaCache,
) -> Result<Option<String>, async_graphql::Error> {
let (Some(function), Some(row_argument)) = (&plan.function, &plan.row_argument) else {
return Ok(None);
};
let session = schema_cache
.routines
.get(function)
.into_iter()
.flatten()
.find(|routine| routine.name == function.name)
.and_then(|routine| {
routine
.params
.iter()
.find(|param| {
param.name == "hasura_session"
&& matches!(param.param_type.as_str(), "json" | "jsonb")
})
.map(|param| param.name.clone())
});
if relationship.arguments.is_empty() && session.is_none() {
return Ok(None);
}
let mut passed = vec![format!(
"{} => {}",
postrust_sql::escape_ident(row_argument),
parent_row
)];
if let Some(session) = session {
passed.push(format!(
"{} => {}",
postrust_sql::escape_ident(&session),
SESSION_ARGUMENT
));
}
for (name, pg_type, required) in &relationship.arguments {
match given.and_then(|given| given.get(name)) {
None if *required => {
return Err(async_graphql::Error::new(format!(
"{} needs the argument \"{}\"",
relationship.name, name
)))
}
None => continue,
Some(value) => {
passed.push(format!(
"{} => ${}::{}",
postrust_sql::escape_ident(name),
param_idx,
pg_type
));
*param_idx += 1;
values.push(value.clone());
}
}
}
Ok(Some(passed.join(", ")))
}
fn embed_arguments(field: async_graphql::SelectionField<'_>) -> HashMap<String, serde_json::Value> {
field
.arguments()
.map(|args| {
args.into_iter()
.map(|(name, value)| {
(
name.to_string(),
plain_numbers(value.into_json().unwrap_or(serde_json::Value::Null)),
)
})
.collect()
})
.unwrap_or_default()
}
type Narrowing = (String, Option<String>, Option<String>, Option<i64>, i64);
#[allow(clippy::too_many_arguments)] fn embed_narrowing(
arguments: &HashMap<String, serde_json::Value>,
plan: &postrust_core::embed::EmbedPlan,
schema_cache: &SchemaCache,
relationships: &HashMap<String, Vec<RelationshipField>>,
target_type: &str,
child_alias: &str,
max_rows: Option<i64>,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
names: &crate::names::NameOverrides,
caller: &crate::role::Caller<'_>,
) -> Result<Narrowing, async_graphql::Error> {
let permission =
permission_predicate(caller, names, &plan.foreign_schema, &plan.foreign_table)?;
let requested = arguments.get("where").filter(|v| !v.is_null()).cloned();
let child_where = match and_predicates(requested, permission) {
Some(expression) => {
let child_scope = WhereScope::for_alias(
&plan.foreign_schema,
&plan.foreign_table,
child_alias,
target_type,
schema_cache,
relationships,
names,
)
.for_caller(*caller);
let mut nested_alias = 0usize;
build_condition(
&expression,
&child_scope,
param_idx,
values,
&mut nested_alias,
)?
}
None => None,
};
let child_order = match arguments.get("order_by") {
Some(order) if !order.is_null() => order_terms(
order,
schema_cache,
&plan.foreign_schema,
&plan.foreign_table,
child_alias,
names,
)?,
_ => None,
};
let ceiling = match (
max_rows,
crate::role::read_limit(caller, names, &plan.foreign_schema, &plan.foreign_table),
) {
(Some(configured), Some(granted)) => Some(configured.min(granted)),
(only, None) | (None, only) => only,
};
let child_limit = match arguments.get("limit").and_then(|v| v.as_i64()) {
Some(requested) => match ceiling {
Some(ceiling) => Some(requested.min(ceiling)),
None => Some(requested),
},
None => ceiling,
};
let child_offset = arguments
.get("offset")
.and_then(|v| v.as_i64())
.unwrap_or(0);
let requested = match arguments.get("distinct_on") {
Some(serde_json::Value::Array(items)) => items
.iter()
.filter_map(|item| item.as_str())
.collect::<Vec<_>>(),
Some(serde_json::Value::String(one)) => vec![one.as_str()],
_ => Vec::new(),
};
let mut distinct_on = Vec::with_capacity(requested.len());
if !requested.is_empty() {
let qi = postrust_core::api_request::QualifiedIdentifier::new(
&plan.foreign_schema,
&plan.foreign_table,
);
let table = schema_cache.get_table(&qi).ok_or_else(|| {
async_graphql::Error::new(format!("unknown table \"{}\"", plan.foreign_table))
})?;
for field in requested {
let column = table_column_for(names, table, field).to_string();
if table.get_column(&column).is_none() {
return Err(async_graphql::Error::new(format!(
"cannot take distinct on unknown column \"{}\" of \"{}\"",
field, plan.foreign_table
)));
}
distinct_on.push(format!(
"{}.{}",
postrust_sql::escape_ident(child_alias),
postrust_sql::escape_ident(&column)
));
}
}
let (distinct, child_order) = distinct_on_clause(&distinct_on, child_order.as_deref())?;
Ok((
distinct,
child_where,
child_order,
child_limit,
child_offset,
))
}
#[allow(clippy::too_many_arguments)] fn build_embed_expressions<'a>(
caller: &crate::role::Caller<'_>,
schema_cache: &SchemaCache,
relationships: &HashMap<String, Vec<RelationshipField>>,
type_name: &str,
parent_alias: &str,
fields: impl IntoIterator<Item = SelectionField<'a>>,
max_rows: Option<i64>,
alias_counter: &mut usize,
param_idx: &mut usize,
values: &mut Vec<serde_json::Value>,
names: &crate::names::NameOverrides,
) -> Result<Vec<(String, String)>, async_graphql::Error> {
let Some(available) = relationships.get(type_name) else {
return Ok(Vec::new());
};
let mut expressions = Vec::new();
for field in fields {
let aggregate_of = field
.name()
.strip_suffix("_aggregate")
.and_then(|name| available.iter().find(|r| r.name == name && r.is_list));
if let Some(rel) = aggregate_of {
let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, schema_cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
*alias_counter += 1;
let child_alias = format!("a{}", alias_counter);
let arguments = embed_arguments(field);
let (child_distinct, child_where, child_order, child_limit, child_offset) =
embed_narrowing(
&arguments,
&plan,
schema_cache,
relationships,
&rel.target_type,
&child_alias,
max_rows,
param_idx,
values,
names,
caller,
)?;
let child_table =
schema_cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
&plan.foreign_schema,
&plan.foreign_table,
));
let select = nested_aggregate_select(
caller,
field,
&child_alias,
schema_cache,
relationships,
&rel.target_type,
child_table,
max_rows,
alias_counter,
param_idx,
values,
names,
)?;
let call = computed_arguments(
rel,
&plan,
&postrust_sql::escape_ident(parent_alias),
arguments.get("args"),
param_idx,
values,
schema_cache,
)?;
let expression = plan
.aggregate_expression(
parent_alias,
&postrust_sql::escape_ident(parent_alias),
&child_alias,
&select,
child_limit,
child_offset,
child_where.as_deref(),
child_order.as_deref(),
call.as_deref(),
&child_distinct,
postrust_core::EmbedRendering::Verbatim,
)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
expressions.push((field.name().to_string(), expression));
continue;
}
let Some(rel) = available.iter().find(|r| r.name == field.name()) else {
continue;
};
let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, schema_cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
*alias_counter += 1;
let child_alias = format!("e{}", alias_counter);
let nested = build_embed_expressions(
caller,
schema_cache,
relationships,
&rel.target_type,
&child_alias,
field.selection_set(),
max_rows,
alias_counter,
param_idx,
values,
names,
)?;
let arguments = embed_arguments(field);
let (child_distinct, child_where, child_order, child_limit, child_offset) =
embed_narrowing(
&arguments,
&plan,
schema_cache,
relationships,
&rel.target_type,
&child_alias,
max_rows,
param_idx,
values,
names,
caller,
)?;
let child_relationships = relationships.get(&rel.target_type);
let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
&plan.foreign_schema,
&plan.foreign_table,
);
let child_table = schema_cache.get_table(&child_qi);
let mut parts: Vec<String> = Vec::new();
for sub in field.selection_set() {
let name = sub.name();
let is_relationship = child_relationships
.map(|rels| {
rels.iter().any(|r| {
r.name == name || (r.is_list && format!("{}_aggregate", r.name) == name)
})
})
.unwrap_or(false);
if is_relationship || name == "__typename" {
continue;
}
let column = child_table
.map(|t| table_column_for(names, t, name))
.unwrap_or(name);
let computed = child_table
.filter(|t| t.get_column(column).is_none())
.and_then(|t| {
let function = names
.computed_source(&t.schema, &t.name, name)
.unwrap_or(name);
t.get_computed_column(function)
});
match computed {
Some(definition) => parts.push(format!(
"{} AS {}",
computed_call(
definition,
&format!("{}.*", postrust_sql::escape_ident(&child_alias)),
&postrust_sql::escape_ident(&child_alias),
),
postrust_sql::escape_ident(name)
)),
None if column != name => parts.push(format!(
"{} AS {}",
postrust_sql::escape_ident(column),
postrust_sql::escape_ident(name)
)),
None => parts.push(postrust_sql::escape_ident(name)),
}
}
if parts.is_empty() && nested.is_empty() {
parts.push(
child_table
.and_then(|table| rename_projection(table, &child_alias, names))
.unwrap_or_else(|| format!("{}.*", postrust_sql::escape_ident(&child_alias))),
);
}
for (field_name, expression) in nested {
parts.push(format!(
"{} AS {}",
expression,
postrust_sql::escape_ident(&field_name)
));
}
let computed_call_arguments = computed_arguments(
rel,
&plan,
&postrust_sql::escape_ident(parent_alias),
arguments.get("args"),
param_idx,
values,
schema_cache,
)?;
let expression = plan
.embed_expression(
parent_alias,
&postrust_sql::escape_ident(parent_alias),
&child_alias,
&format!("{}{}", child_distinct, parts.join(", ")),
child_limit,
child_offset,
child_where.as_deref(),
child_order.as_deref(),
computed_call_arguments.as_deref(),
postrust_core::EmbedRendering::Verbatim,
)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
expressions.push((rel.name.clone(), expression));
}
Ok(expressions)
}
struct EmbedContext<'c> {
schema_cache: &'c SchemaCache,
relationships: &'c HashMap<String, Vec<RelationshipField>>,
max_rows: Option<i64>,
}
fn embed_relationships<'f>(
conn: &'f mut sqlx::PgConnection,
ctx: &'f EmbedContext<'f>,
type_name: &'f str,
selection: async_graphql::SelectionField<'f>,
rows: &'f mut [serde_json::Value],
) -> futures::future::BoxFuture<'f, Result<(), async_graphql::Error>> {
Box::pin(async move {
if rows.is_empty() {
return Ok(());
}
let Some(available) = ctx.relationships.get(type_name) else {
return Ok(());
};
for requested in selection.selection_set() {
let Some(rel) = available.iter().find(|r| r.name == requested.name()) else {
continue;
};
let plan =
postrust_core::embed::EmbedPlan::resolve(&rel.relationship, ctx.schema_cache)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let keys = postrust_core::embed::parent_keys(rows, &plan.local_column);
let mut child_columns: Vec<String> = Vec::new();
let mut project_everything = false;
let child_relationships = ctx.relationships.get(&rel.target_type);
for field in requested.selection_set() {
let name = field.name();
match child_relationships.and_then(|rels| rels.iter().find(|r| r.name == name)) {
Some(nested_rel) => {
match postrust_core::embed::EmbedPlan::resolve(
&nested_rel.relationship,
ctx.schema_cache,
) {
Ok(nested_plan) => child_columns.push(nested_plan.local_column),
Err(_) => project_everything = true,
}
}
None if name == "__typename" => {}
None => child_columns.push(name.to_string()),
}
}
if project_everything || child_columns.is_empty() {
child_columns.clear();
}
let mut grouped = if keys.is_empty() {
std::collections::HashMap::new()
} else {
let sql = plan
.children_grouped_sql(ctx.max_rows, &child_columns)
.map_err(|e| async_graphql::Error::new(e.to_string()))?;
let fetched = sqlx::query(AssertSqlSafe(sql))
.bind(&keys)
.fetch_all(&mut *conn)
.await
.map_err(database_error)?;
use sqlx::Row;
let pairs: Vec<(serde_json::Value, serde_json::Value)> = fetched
.into_iter()
.filter_map(|row| {
Some((
row.try_get::<serde_json::Value, _>(0).ok()?,
row.try_get::<serde_json::Value, _>(1).ok()?,
))
})
.collect();
postrust_core::embed::group_from_aggregated(pairs)
};
let has_deeper_embed = ctx
.relationships
.get(&rel.target_type)
.map(|rels| {
requested
.selection_set()
.any(|field| rels.iter().any(|r| r.name == field.name()))
})
.unwrap_or(false);
if has_deeper_embed {
let mut order: Vec<(String, usize)> = Vec::with_capacity(grouped.len());
let mut flat: Vec<serde_json::Value> = Vec::new();
for (key, children) in grouped.drain() {
order.push((key, children.len()));
flat.extend(children);
}
embed_relationships(&mut *conn, ctx, &rel.target_type, requested, &mut flat)
.await?;
let mut rest = flat.into_iter();
for (key, count) in order {
grouped.insert(key, rest.by_ref().take(count).collect());
}
}
for row in rows.iter_mut() {
postrust_core::embed::attach_to_parent(row, &rel.name, &plan, &grouped);
}
}
Ok(())
})
}
fn pk_where_from_args(
ctx: &ResolverContext<'_>,
schema_name: &str,
table_name: &str,
pk_columns: &[(String, String)],
names: &crate::names::NameOverrides,
) -> Result<serde_json::Value, async_graphql::Error> {
if pk_columns.is_empty() {
return Err(async_graphql::Error::new(format!(
"\"{}\" has no primary key, so it cannot be mutated by key",
table_name
)));
}
let from_object = ctx
.args
.try_get("pk_columns")
.ok()
.map(|v| accessor_to_json(&v));
let mut conditions = serde_json::Map::new();
for (col_name, _) in pk_columns {
let field = names
.column(schema_name, table_name, col_name)
.unwrap_or(col_name);
let value = match &from_object {
Some(serde_json::Value::Object(map)) => map.get(field).cloned().ok_or_else(|| {
async_graphql::Error::new(format!(
"pk_columns is missing the key column \"{}\"",
field
))
})?,
_ => {
let arg = ctx.args.try_get(field).map_err(|_| {
async_graphql::Error::new(format!(
"missing required primary key argument \"{}\"",
field
))
})?;
accessor_to_json(&arg)
}
};
conditions.insert(field.to_string(), serde_json::json!({ "_eq": value }));
}
Ok(serde_json::Value::Object(conditions))
}
fn pk_argument_type(pg_type: &str) -> String {
let rendered = crate::types::pg_type_to_graphql(pg_type).to_string();
if rendered.starts_with('[') {
"String".to_string()
} else {
rendered
}
}
fn exposed_pk_argument_type(
names: &crate::names::NameOverrides,
schema: &str,
table: &str,
column: &str,
pg_type: &str,
) -> String {
names
.column_type(schema, table, column)
.map(|given| given.to_string())
.unwrap_or_else(|| pk_argument_type(pg_type))
}
fn graphql_type_ref(type_str: &str) -> TypeRef {
let is_list = type_str.starts_with('[');
let is_nn = type_str.ends_with('!');
let inner = if is_list {
let stripped = type_str
.trim_end_matches('!') .trim_start_matches('[') .trim_end_matches(']'); stripped
} else {
type_str.trim_end_matches('!')
};
let inner_nn = inner.ends_with('!');
let base_type = inner.trim_end_matches('!');
if is_list {
if is_nn {
if inner_nn {
TypeRef::named_nn_list_nn(base_type)
} else {
TypeRef::named_list_nn(base_type)
}
} else if inner_nn {
TypeRef::named_nn_list(base_type)
} else {
TypeRef::named_list(base_type)
}
} else if is_nn {
TypeRef::named_nn(base_type)
} else {
TypeRef::named(base_type)
}
}
const NUMBER_AS_TEXT: &str = "$serde_json::private::Number";
fn number_kept_as_text(
map: &serde_json::Map<String, serde_json::Value>,
) -> Option<serde_json::Value> {
if map.len() != 1 {
return None;
}
let text = map.get(NUMBER_AS_TEXT)?.as_str()?;
serde_json::from_str::<serde_json::Value>(text)
.ok()
.filter(|value| value.is_number())
}
fn plain_numbers(value: serde_json::Value) -> serde_json::Value {
match value {
serde_json::Value::Object(map) => match number_kept_as_text(&map) {
Some(number) => number,
None => serde_json::Value::Object(
map.into_iter()
.map(|(k, v)| (k, plain_numbers(v)))
.collect(),
),
},
serde_json::Value::Array(items) => {
serde_json::Value::Array(items.into_iter().map(plain_numbers).collect())
}
other => other,
}
}
fn accessor_to_json(accessor: &ValueAccessor<'_>) -> serde_json::Value {
if accessor.is_null() {
serde_json::Value::Null
} else if let Ok(b) = accessor.boolean() {
serde_json::Value::Bool(b)
} else if let Ok(i) = accessor.i64() {
serde_json::Value::Number(i.into())
} else if let Ok(f) = accessor.f64() {
serde_json::Number::from_f64(f)
.map(serde_json::Value::Number)
.unwrap_or(serde_json::Value::Null)
} else if let Ok(s) = accessor.string() {
serde_json::Value::String(s.to_string())
} else if let Ok(name) = accessor.enum_name() {
serde_json::Value::String(name.to_string())
} else if let Ok(list) = accessor.list() {
serde_json::Value::Array(list.iter().map(|v| accessor_to_json(&v)).collect())
} else if let Ok(obj) = accessor.object() {
let map: serde_json::Map<String, serde_json::Value> = obj
.iter()
.map(|(k, v)| (k.to_string(), accessor_to_json(&v)))
.collect();
match number_kept_as_text(&map) {
Some(number) => number,
None => serde_json::Value::Object(map),
}
} else {
serde_json::Value::Null
}
}
fn value_to_json(value: &Value) -> serde_json::Value {
match value {
Value::Null => serde_json::Value::Null,
Value::Boolean(b) => serde_json::Value::Bool(*b),
Value::Number(n) => {
if let Some(i) = n.as_i64() {
serde_json::Value::Number(i.into())
} else if let Some(f) = n.as_f64() {
serde_json::Value::Number(serde_json::Number::from_f64(f).unwrap())
} else {
serde_json::Value::Null
}
}
Value::String(s) => serde_json::Value::String(s.clone()),
Value::List(arr) => serde_json::Value::Array(arr.iter().map(value_to_json).collect()),
Value::Object(obj) => {
let map: serde_json::Map<String, serde_json::Value> = obj
.iter()
.map(|(k, v)| (k.to_string(), value_to_json(v)))
.collect();
serde_json::Value::Object(map)
}
Value::Binary(b) => serde_json::Value::String(base64::Engine::encode(
&base64::engine::general_purpose::STANDARD,
b,
)),
Value::Enum(e) => serde_json::Value::String(e.to_string()),
}
}
fn json_to_value(json: serde_json::Value) -> Value {
match json {
serde_json::Value::Null => Value::Null,
serde_json::Value::Bool(b) => Value::Boolean(b),
serde_json::Value::Number(n) => {
if let Some(i) = n.as_i64() {
Value::Number(i.into())
} else if let Some(f) = n.as_f64() {
Value::Number(async_graphql::Number::from_f64(f).unwrap())
} else {
Value::Null
}
}
serde_json::Value::String(s) => Value::String(s),
serde_json::Value::Array(arr) => Value::List(arr.into_iter().map(json_to_value).collect()),
serde_json::Value::Object(obj) => {
let map: indexmap::IndexMap<async_graphql::Name, Value> = obj
.into_iter()
.map(|(k, v)| (async_graphql::Name::new(k), json_to_value(v)))
.collect();
Value::Object(map)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use indexmap::IndexMap;
use postrust_core::schema_cache::{Cardinality, Column, Relationship, Table};
use postrust_core::QualifiedIdentifier;
use std::collections::{HashMap, HashSet};
fn entity_key_column(pg_type: &str, explicit_id: bool) -> EntityKeyColumn {
EntityKeyColumn {
field_name: "id".to_string(),
column_name: "id".to_string(),
pg_type: pg_type.to_string(),
explicit_id,
}
}
#[test]
fn federation_integer_keys_reject_strings() {
let column = entity_key_column("int4", false);
assert_eq!(
normalize_entity_key_value("users", &column, &serde_json::json!(2)).unwrap(),
serde_json::json!(2)
);
let error = normalize_entity_key_value("users", &column, &serde_json::json!("2"))
.expect_err("a GraphQL Int cannot be a string");
assert_eq!(
error.message,
"entity representation for \"users\" has invalid key field \"id\": expected a value compatible with PostgreSQL type \"int4\", found a string"
);
}
#[test]
fn federation_bigint_keys_accept_exact_decimal_strings() {
let column = entity_key_column("int8", false);
assert_eq!(
normalize_entity_key_value("users", &column, &serde_json::json!(7)).unwrap(),
serde_json::json!(7)
);
assert_eq!(
normalize_entity_key_value("users", &column, &serde_json::json!("9007199254740993"))
.unwrap(),
serde_json::json!(9_007_199_254_740_993_i64)
);
assert!(normalize_entity_key_value(
"users",
&column,
&serde_json::json!("9223372036854775808")
)
.is_err());
assert!(normalize_entity_key_value("users", &column, &serde_json::json!("7.5")).is_err());
}
#[test]
fn federation_numeric_keys_accept_exact_decimal_strings() {
let column = entity_key_column("numeric", false);
assert_eq!(
normalize_entity_key_value("users", &column, &serde_json::json!(7.5)).unwrap(),
serde_json::json!("7.5")
);
assert_eq!(
normalize_entity_key_value(
"users",
&column,
&serde_json::json!("9007199254740993.123456789")
)
.unwrap(),
serde_json::json!("9007199254740993.123456789")
);
assert!(
normalize_entity_key_value("users", &column, &serde_json::json!("not-a-number"))
.is_err()
);
}
#[test]
fn federation_integer_backed_ids_have_one_normal_form() {
let column = entity_key_column("int4", true);
let number = normalize_entity_key_value("users", &column, &serde_json::json!(2)).unwrap();
let string = normalize_entity_key_value("users", &column, &serde_json::json!("2")).unwrap();
assert_eq!(number, serde_json::json!(2));
assert_eq!(string, number);
assert!(normalize_entity_key_value("users", &column, &serde_json::json!("two")).is_err());
}
#[test]
fn federation_text_backed_ids_normalize_integer_input_to_text() {
let column = entity_key_column("text", true);
let number = normalize_entity_key_value("users", &column, &serde_json::json!(2)).unwrap();
let string = normalize_entity_key_value("users", &column, &serde_json::json!("2")).unwrap();
assert_eq!(number, serde_json::json!("2"));
assert_eq!(string, number);
}
#[test]
fn federation_uuid_backed_ids_are_validated_and_canonicalized() {
let column = entity_key_column("uuid", true);
assert_eq!(
normalize_entity_key_value(
"users",
&column,
&serde_json::json!("550E8400-E29B-41D4-A716-446655440000")
)
.unwrap(),
serde_json::json!("550e8400-e29b-41d4-a716-446655440000")
);
assert!(
normalize_entity_key_value("users", &column, &serde_json::json!("not-a-uuid")).is_err()
);
}
fn create_test_table(name: &str) -> Table {
let mut columns = IndexMap::new();
columns.insert(
"id".into(),
Column {
name: "id".into(),
description: None,
nullable: false,
data_type: "integer".into(),
nominal_type: "int4".into(),
max_len: None,
default: Some("nextval('id_seq')".into()),
enum_values: vec![],
is_pk: true,
position: 1,
always_generated: false,
domain_type: None,
},
);
columns.insert(
"name".into(),
Column {
name: "name".into(),
description: None,
nullable: false,
data_type: "text".into(),
nominal_type: "text".into(),
max_len: None,
default: None,
enum_values: vec![],
is_pk: false,
position: 2,
always_generated: false,
domain_type: None,
},
);
Table {
schema: "public".into(),
name: name.into(),
description: None,
is_view: false,
insertable: true,
updatable: true,
deletable: true,
pk_cols: vec!["id".into()],
unique_constraints: Vec::new(),
columns,
computed_columns: Default::default(),
is_partitioned: false,
}
}
fn create_test_schema_cache() -> SchemaCache {
let mut tables = HashMap::new();
let users = create_test_table("users");
tables.insert(users.qualified_identifier(), users);
SchemaCache {
tables,
relationships: HashMap::new(),
routines: HashMap::new(),
timezones: HashSet::new(),
media_handlers: HashMap::new(),
pg_version: 150000,
representations: Default::default(),
}
}
fn add_test_table(cache: &mut SchemaCache, name: &str) {
let table = create_test_table(name);
cache.tables.insert(table.qualified_identifier(), table);
}
fn add_test_posts_relationship(cache: &mut SchemaCache) {
let users = QualifiedIdentifier::new("public", "users");
cache.relationships.insert(
(users.clone(), "public".into()),
vec![Relationship::ForeignKey {
table: users,
foreign_table: QualifiedIdentifier::new("public", "posts"),
is_self: false,
cardinality: Cardinality::O2M {
constraint: "posts_user_id_fkey".into(),
columns: vec![("id".into(), "user_id".into())],
},
table_is_view: false,
foreign_table_is_view: false,
constraint_name: "posts_user_id_fkey".into(),
}],
);
}
#[test]
fn test_graphql_type_ref_simple() {
let _type_ref = graphql_type_ref("String");
}
#[test]
fn test_graphql_type_ref_non_null() {
let _type_ref = graphql_type_ref("String!");
}
#[test]
fn test_graphql_type_ref_list() {
let _type_ref = graphql_type_ref("[String]");
}
#[test]
fn test_graphql_type_ref_list_non_null() {
let _type_ref = graphql_type_ref("[String!]!");
}
#[test]
fn test_value_to_json_null() {
let value = Value::Null;
let json = value_to_json(&value);
assert_eq!(json, serde_json::Value::Null);
}
#[test]
fn test_value_to_json_boolean() {
let value = Value::Boolean(true);
let json = value_to_json(&value);
assert_eq!(json, serde_json::Value::Bool(true));
}
#[test]
fn test_value_to_json_number() {
let value = Value::Number(42.into());
let json = value_to_json(&value);
assert_eq!(json, serde_json::json!(42));
}
#[test]
fn test_value_to_json_string() {
let value = Value::String("hello".to_string());
let json = value_to_json(&value);
assert_eq!(json, serde_json::Value::String("hello".to_string()));
}
#[test]
fn test_value_to_json_list() {
let value = Value::List(vec![Value::Number(1.into()), Value::Number(2.into())]);
let json = value_to_json(&value);
assert_eq!(json, serde_json::json!([1, 2]));
}
#[test]
fn test_json_to_value_null() {
let json = serde_json::Value::Null;
let value = json_to_value(json);
assert!(matches!(value, Value::Null));
}
#[test]
fn test_json_to_value_boolean() {
let json = serde_json::Value::Bool(false);
let value = json_to_value(json);
assert!(matches!(value, Value::Boolean(false)));
}
#[test]
fn test_json_to_value_number() {
let json = serde_json::json!(123);
let value = json_to_value(json);
assert!(matches!(value, Value::Number(_)));
}
#[test]
fn test_json_to_value_string() {
let json = serde_json::Value::String("test".to_string());
let value = json_to_value(json);
assert!(matches!(value, Value::String(_)));
}
#[test]
fn test_json_to_value_array() {
let json = serde_json::json!([1, 2, 3]);
let value = json_to_value(json);
assert!(matches!(value, Value::List(_)));
}
#[test]
fn test_json_to_value_object() {
let json = serde_json::json!({"key": "value"});
let value = json_to_value(json);
assert!(matches!(value, Value::Object(_)));
}
#[test]
fn test_build_dynamic_schema() {
let cache = create_test_schema_cache();
let config = SchemaConfig::default();
let generated = build_schema(&cache, &config);
let result = build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(Default::default()),
std::time::Duration::from_secs(30),
None,
);
if let Err(ref e) = result {
eprintln!("Schema build error: {:?}", e);
}
assert!(result.is_ok(), "Schema build failed: {:?}", result.err());
}
#[test]
fn a_column_type_override_reaches_fields_and_key_arguments() {
let cache = create_test_schema_cache();
let names = crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"column_types": {"id": "ID"}}}}"#,
)
.unwrap();
let config = SchemaConfig {
names: names.clone(),
..SchemaConfig::default()
};
let generated = build_schema(&cache, &config);
let schema = build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(names),
std::time::Duration::from_secs(30),
None,
)
.expect("schema with an ID override should build");
let sdl = schema.sdl();
let input_body = |name: &str| {
sdl.split_once(&format!("input {name} {{"))
.and_then(|(_, rest)| rest.split_once('}'))
.map(|(body, _)| body.to_string())
.unwrap_or_else(|| panic!("missing input {name}:\n{sdl}"))
};
assert!(sdl.contains("id: ID!"), "object and input fields:\n{sdl}");
assert!(
sdl.contains("users_by_pk(id: ID!): users"),
"query key argument:\n{sdl}"
);
assert!(
sdl.contains("delete_users_by_pk(id: ID!): users"),
"delete key argument:\n{sdl}"
);
assert!(
input_body("users_pk_columns_input").contains("id: ID!"),
"update key input:\n{sdl}"
);
assert!(
input_body("users_bool_exp").contains("id: ID_comparison_exp"),
"comparison input:\n{sdl}"
);
assert!(
input_body("users_insert_input").contains("id: ID"),
"write input:\n{sdl}"
);
}
#[test]
fn a_type_prefix_names_the_complete_dynamic_schema() {
let cache = create_test_schema_cache();
let config = SchemaConfig {
type_prefix: Some("test".into()),
..SchemaConfig::default()
};
let generated = build_schema(&cache, &config);
let schema = build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(Default::default()),
std::time::Duration::from_secs(30),
None,
)
.expect("a prefixed schema should build");
let sdl = schema.sdl();
assert!(sdl.contains("type test_users "), "object type:\n{sdl}");
assert!(sdl.contains("test_users("), "query root:\n{sdl}");
assert!(sdl.contains("test_users_by_pk("), "by-PK root:\n{sdl}");
assert!(sdl.contains("test_users_bool_exp"), "derived input:\n{sdl}");
assert!(
sdl.contains("test_users_aggregate"),
"aggregate type:\n{sdl}"
);
assert!(sdl.contains("test_insert_users("), "mutation root:\n{sdl}");
}
#[test]
fn federation_is_absent_by_default() {
let cache = create_test_schema_cache();
let generated = build_schema(&cache, &SchemaConfig::default());
let schema = build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(Default::default()),
std::time::Duration::from_secs(30),
None,
)
.expect("a non-federated schema should build");
let sdl = schema.sdl();
assert!(!sdl.contains("_Service"), "service type:\n{sdl}");
assert!(!sdl.contains("_service"), "service field:\n{sdl}");
assert!(!sdl.contains("@key"), "federation key:\n{sdl}");
}
#[test]
fn federation_entity_selections_respect_fragment_type_conditions() {
let document = async_graphql::parser::parse_query(
r#"
query {
_entities(representations: []) {
__typename
... on authors { id items { title } }
...ShopFields
... { common }
...EntityName
}
}
fragment ShopFields on shops { id items { sku } }
fragment EntityName on _Entity { __typename }
"#,
)
.expect("query parses");
let (_, operation) = document.operations.iter().next().expect("one operation");
let entity = operation
.node
.selection_set
.node
.items
.iter()
.find_map(|selection| match &selection.node {
Selection::Field(field) if field.node.name.node == "_entities" => Some(&field.node),
_ => None,
})
.expect("_entities field");
assert_eq!(
entity_field_applicability(&entity.selection_set.node, &document.fragments, "authors"),
vec![true, true, true, false, false, true, true]
);
assert_eq!(
entity_field_applicability(&entity.selection_set.node, &document.fragments, "shops"),
vec![true, false, false, true, true, true, true]
);
}
#[tokio::test]
async fn federation_entity_lists_allow_null_members() {
let entity =
Object::new("widgets").field(Field::new("id", TypeRef::named_nn(TypeRef::INT), |_| {
FieldFuture::new(async { Ok(Some(Value::from(2))) })
}));
let union = Union::new("_Entity").possible_type("widgets");
let query = Object::new("Query").field(Field::new(
"_entities",
TypeRef::named_list_nn("_Entity"),
|_| {
FieldFuture::new(async {
Ok(Some(FieldValue::list([
FieldValue::NULL,
FieldValue::NULL.with_type("widgets"),
])))
})
},
));
let schema = Schema::build("Query", None, None)
.extension(NullableFederationEntity)
.register(entity)
.register(union)
.register(query)
.finish()
.expect("schema builds");
let response = schema
.execute("{ _entities { ... on widgets { id } } }")
.await;
assert!(response.errors.is_empty(), "{:?}", response.errors);
assert_eq!(
response.data,
async_graphql::value!({ "_entities": [null, { "id": 2 }] })
);
}
#[test]
fn federation_metadata_is_byte_for_byte_inert_when_disabled() {
let cache = create_test_schema_cache();
let sdl_for = |config: SchemaConfig| {
let generated = build_schema(&cache, &config);
build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(config.names.clone()),
std::time::Duration::from_secs(30),
None,
)
.expect("schema should build")
.sdl()
};
let baseline = sdl_for(SchemaConfig {
type_prefix: Some("test".into()),
..SchemaConfig::default()
});
let with_disabled_federation_metadata = sdl_for(SchemaConfig {
enable_federation: false,
type_prefix: Some("test".into()),
names: crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"federation": {"shared": true}}}}"#,
)
.expect("metadata"),
..SchemaConfig::default()
});
assert_eq!(with_disabled_federation_metadata, baseline);
assert!(
with_disabled_federation_metadata.contains("type test_users "),
"prefixed object type remains when federation is disabled:\n{with_disabled_federation_metadata}"
);
assert!(
!with_disabled_federation_metadata.contains("_Service"),
"service type:\n{with_disabled_federation_metadata}"
);
assert!(
!with_disabled_federation_metadata.contains("@key"),
"federation key:\n{with_disabled_federation_metadata}"
);
}
#[test]
fn shared_entities_make_relationship_fields_shareable() {
let federation_sdl = |shared: bool| {
let mut cache = create_test_schema_cache();
add_test_table(&mut cache, "posts");
add_test_posts_relationship(&mut cache);
let names = match shared {
true => crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"federation": {"shared": true}}}}"#,
)
.expect("metadata"),
false => crate::names::NameOverrides::default(),
};
let config = SchemaConfig {
enable_federation: true,
type_prefix: Some("test".into()),
names: names.clone(),
..SchemaConfig::default()
};
let generated = build_schema(&cache, &config);
build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(names),
std::time::Duration::from_secs(30),
None,
)
.expect("federated schema should build")
.sdl_with_options(async_graphql::SDLExportOptions::new().federation())
};
let shared = federation_sdl(true);
assert!(
shared.contains("): [test_posts!]! @shareable"),
"shared relationship field:\n{shared}"
);
assert!(
shared.contains("): test_posts_aggregate! @shareable"),
"shared relationship aggregate field:\n{shared}"
);
let not_shared = federation_sdl(false);
assert!(
not_shared.contains("): [test_posts!]!"),
"non-shared relationship field:\n{not_shared}"
);
assert!(
!not_shared.contains("): [test_posts!]! @shareable"),
"non-shared relationship field must not be shareable:\n{not_shared}"
);
assert!(
not_shared.contains("): test_posts_aggregate!"),
"non-shared relationship aggregate field:\n{not_shared}"
);
assert!(
!not_shared.contains("): test_posts_aggregate! @shareable"),
"non-shared relationship aggregate must not be shareable:\n{not_shared}"
);
}
#[tokio::test]
async fn federation_exposes_service_sdl_and_entity_keys() {
let mut cache = create_test_schema_cache();
add_test_table(&mut cache, "posts");
let config = SchemaConfig {
enable_federation: true,
type_prefix: Some("test".into()),
names: crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"federation": {"shared": true}}}}"#,
)
.expect("metadata"),
..SchemaConfig::default()
};
let generated = build_schema(&cache, &config);
let schema = build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(Default::default()),
std::time::Duration::from_secs(30),
None,
)
.expect("a federated schema should build");
let sdl = schema.sdl();
let federation_sdl =
schema.sdl_with_options(async_graphql::SDLExportOptions::new().federation());
assert!(sdl.contains("_Service"), "service type:\n{sdl}");
assert!(sdl.contains("_service"), "service field:\n{sdl}");
assert!(
federation_sdl.contains("type users @key(fields: \"id\")"),
"users key:\n{federation_sdl}"
);
assert!(
federation_sdl.contains("@shareable"),
"shared field directive:\n{federation_sdl}"
);
assert!(sdl.contains("type users "), "shared object type:\n{sdl}");
assert!(
!sdl.contains("type test_users "),
"prefixed shared object type:\n{sdl}"
);
assert!(sdl.contains("test_users("), "query root:\n{sdl}");
assert!(sdl.contains("test_users_by_pk("), "by-PK root:\n{sdl}");
assert!(sdl.contains("test_users_bool_exp"), "derived input:\n{sdl}");
assert!(
!sdl.contains("input users_bool_exp "),
"unprefixed derived input:\n{sdl}"
);
assert!(
sdl.contains("test_users_aggregate"),
"aggregate type:\n{sdl}"
);
assert!(
!sdl.contains("type users_aggregate "),
"unprefixed aggregate type:\n{sdl}"
);
assert!(
sdl.contains("test_users_mutation_response"),
"mutation response type:\n{sdl}"
);
assert!(
!sdl.contains("type users_mutation_response "),
"unprefixed mutation response type:\n{sdl}"
);
assert!(
sdl.contains("type test_posts {"),
"non-shared object type remains prefixed:\n{sdl}"
);
assert!(
!sdl.contains("type posts {"),
"non-shared object type must not become unprefixed:\n{sdl}"
);
assert!(sdl.contains("test_posts("), "prefixed posts root:\n{sdl}");
assert!(
sdl.contains("test_posts_bool_exp"),
"prefixed posts bool_exp:\n{sdl}"
);
let response = schema.execute("{ _service { sdl } }").await;
assert!(
response.errors.is_empty(),
"service query errors: {:?}",
response.errors
);
let service = response.data.into_json().expect("service response JSON");
let service_sdl = service
.get("_service")
.and_then(|value| value.get("sdl"))
.and_then(serde_json::Value::as_str)
.expect("service SDL string");
assert!(
service_sdl.contains("type Query"),
"federation root type:\n{service_sdl}"
);
assert!(
service_sdl.contains("test_users("),
"query fields are on the federation root:\n{service_sdl}"
);
assert!(
service_sdl.contains("test_posts("),
"non-shared query fields are on the federation root:\n{service_sdl}"
);
assert!(
!service_sdl.contains("type query_root"),
"Apollo composition treats query_root as an ordinary object without a schema definition:\n{service_sdl}"
);
assert!(
service_sdl.contains("type users @key(fields: \"id\")"),
"{service_sdl}"
);
assert!(
service_sdl.contains("type test_posts @key(fields: \"id\")"),
"{service_sdl}"
);
}
#[test]
fn exists_is_a_subselect_over_another_table() {
let cache = create_test_schema_cache();
let names = crate::names::NameOverrides::default();
let relationships: HashMap<String, Vec<RelationshipField>> = HashMap::new();
let scope = WhereScope::table("public", "users", "users", &names)
.with_resolution(&cache, &relationships);
let mut param_idx = 1usize;
let mut values: Vec<serde_json::Value> = Vec::new();
let mut aliases = 0usize;
let sql = build_condition(
&serde_json::json!({
"_exists": {"_table": "users", "_where": {"name": {"_eq": "Ann"}}}
}),
&scope,
&mut param_idx,
&mut values,
&mut aliases,
)
.expect("compiles")
.expect("constrains something");
assert_eq!(
sql,
"EXISTS (SELECT 1 FROM \"public\".\"users\" AS \"pgrst_exists_1\" \
WHERE \"pgrst_exists_1\".\"name\" = $1::text)"
);
assert_eq!(values, vec![serde_json::json!("Ann")]);
}
#[test]
fn exists_names_a_table_the_server_has() {
let cache = create_test_schema_cache();
let names = crate::names::NameOverrides::default();
let relationships: HashMap<String, Vec<RelationshipField>> = HashMap::new();
let scope = WhereScope::table("public", "users", "users", &names)
.with_resolution(&cache, &relationships);
let error = build_condition(
&serde_json::json!({"_exists": {"_table": "nowhere", "_where": {}}}),
&scope,
&mut 1,
&mut Vec::new(),
&mut 0,
)
.expect_err("a table that is not there is an error");
assert!(
error.message.contains("public.nowhere"),
"{}",
error.message
);
}
#[test]
fn a_table_a_role_may_count_and_not_read_has_a_count_and_no_rows() {
let cache = create_test_schema_cache();
let names = crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"permissions":
{"counter": {"select": {"columns": [], "filter": {},
"allow_aggregations": true}}}}}}"#,
)
.unwrap();
let view = crate::role::cache_for_role(&cache, &names, "counter", false);
assert!(!view.tables.is_empty(), "the table is there to be counted");
let config = SchemaConfig {
names: names.clone(),
role: Some("counter".to_string()),
..SchemaConfig::default()
};
let generated = build_schema(&view, &config);
let schema = build_dynamic_schema(
&generated,
&view,
None,
None,
Arc::new(names),
std::time::Duration::from_secs(30),
Some("counter"),
)
.expect("a role that may only count still has a buildable schema");
let sdl = schema.sdl();
assert!(
sdl.contains("users_aggregate"),
"the count is there:\n{}",
sdl
);
assert!(sdl.contains("count: Int!"), "and it is a count:\n{}", sdl);
assert!(!sdl.contains("type users "), "no row type:\n{}", sdl);
assert!(
!sdl.contains("nodes"),
"and no rows beside the numbers:\n{}",
sdl
);
assert!(
!sdl.contains("users_select_column"),
"no column enum to order by or count over:\n{}",
sdl
);
assert!(!sdl.contains("users_max_fields"), "no max:\n{}", sdl);
}
#[test]
fn a_table_a_role_may_only_write_gets_the_write_and_no_type() {
let cache = create_test_schema_cache();
let names = crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"permissions":
{"writer": {"insert": {"columns": "*", "check": {}}}}}}}"#,
)
.unwrap();
let view = crate::role::cache_for_role(&cache, &names, "writer", false);
let config = SchemaConfig {
names: names.clone(),
role: Some("writer".to_string()),
..SchemaConfig::default()
};
let generated = build_schema(&view, &config);
let schema = build_dynamic_schema(
&generated,
&view,
None,
None,
Arc::new(names),
std::time::Duration::from_secs(30),
Some("writer"),
)
.expect("a role that may only write still has a buildable schema");
let sdl = schema.sdl();
assert!(
sdl.contains("insert_users("),
"the write is there:\n{}",
sdl
);
assert!(
sdl.contains("users_insert_input"),
"and its input:\n{}",
sdl
);
assert!(!sdl.contains("type users "), "no row type:\n{}", sdl);
assert!(!sdl.contains("insert_users_one"), "no insert_one:\n{}", sdl);
assert!(
!sdl.contains("returning"),
"and nothing to return:\n{}",
sdl
);
assert!(
sdl.contains("users_mutation_response"),
"the count is still answered:\n{}",
sdl
);
assert!(
!sdl.contains("_mutation_response_mutation_response"),
"and no type is named twice over:\n{}",
sdl
);
}
fn cache_with_a_mutation_function() -> SchemaCache {
use postrust_core::schema_cache::{FuncVolatility, RetType, Routine};
let mut cache = create_test_schema_cache();
let routine = Routine {
schema: "public".into(),
name: "add_to_score".into(),
description: None,
params: Vec::new(),
return_type: RetType::SetOf("users".into()),
returns_composite: true,
volatility: FuncVolatility::Volatile,
has_variadic: false,
isolation_level: None,
settings: Vec::new(),
is_procedure: false,
media_type: None,
media_base_type: None,
output_columns: Vec::new(),
};
cache
.routines
.insert(routine.qualified_identifier(), vec![routine]);
cache
}
#[test]
fn a_mutation_function_needs_the_role_to_be_named() {
let cache = cache_with_a_mutation_function();
let withheld = crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"permissions":
{"reader": {"select": {"columns": "*", "filter": {}}}}}},
"functions": {"public.add_to_score": {"exposed_as": "mutation"}}}"#,
)
.unwrap();
let granted = crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"permissions":
{"reader": {"select": {"columns": "*", "filter": {}}}}}},
"functions": {"public.add_to_score":
{"exposed_as": "mutation", "roles": ["reader"]}}}"#,
)
.unwrap();
let sdl_for = |names: &crate::names::NameOverrides| {
let view = crate::role::cache_for_role(&cache, names, "reader", false);
let config = SchemaConfig {
names: names.clone(),
role: Some("reader".to_string()),
..SchemaConfig::default()
};
let generated = build_schema(&view, &config);
build_dynamic_schema(
&generated,
&view,
None,
None,
Arc::new(names.clone()),
std::time::Duration::from_secs(30),
Some("reader"),
)
.expect("schema builds")
.sdl()
};
let without = sdl_for(&withheld);
assert!(
!without.contains("add_to_score"),
"not granted, so not there:\n{}",
without
);
assert!(
!without.contains("mutation: mutation_root"),
"and no mutation root at all:\n{}",
without
);
let with = sdl_for(&granted);
assert!(
with.contains("add_to_score"),
"named by a function permission, so it is:\n{}",
with
);
}
#[test]
fn a_role_the_document_never_names_is_granted_nothing() {
let cache = Arc::new(create_test_schema_cache());
let config = SchemaConfig {
names: crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"permissions":
{"reader": {"select": {"columns": "*", "filter": {}}}}}}}"#,
)
.unwrap(),
..SchemaConfig::default()
};
let empty = build_ungranted_schema(&cache, &config).expect("it builds");
let sdl = empty.sdl();
assert!(
sdl.contains("no_queries_available"),
"nothing to read:\n{}",
sdl
);
assert!(!sdl.contains("type users "), "and no table:\n{}", sdl);
assert!(build_ungranted_schema(&cache, &SchemaConfig::default()).is_none());
}
#[test]
fn a_column_a_role_may_set_without_seeing_is_in_the_input_alone() {
let cache = create_test_schema_cache();
let names = crate::names::NameOverrides::parse(
r#"{"tables": {"public.users": {"permissions":
{"user": {"select": {"columns": ["id"], "filter": {}},
"insert": {"columns": "*", "check": {}}}}}}}"#,
)
.unwrap();
let view = crate::role::cache_for_role(&cache, &names, "user", false);
let config = SchemaConfig {
names: names.clone(),
role: Some("user".to_string()),
..SchemaConfig::default()
};
let generated = build_schema(&view, &config);
let object = &generated.object_types["users"];
assert_eq!(
object
.fields
.iter()
.map(|f| f.name.as_str())
.collect::<Vec<_>>(),
vec!["id"],
"the type shows what may be read"
);
assert!(
object.writable_fields.iter().any(|f| f.name == "name"),
"and the input keeps what may be written"
);
let schema = build_dynamic_schema(
&generated,
&view,
None,
None,
Arc::new(names),
std::time::Duration::from_secs(30),
Some("user"),
)
.expect("schema builds");
let sdl = schema.sdl();
let input = sdl
.split("input users_insert_input")
.nth(1)
.and_then(|rest| rest.split('}').next())
.expect("the insert input is registered");
assert!(input.contains("name:"), "settable without being readable");
}
#[test]
fn a_generated_column_is_readable_and_not_writable() {
let mut cache = create_test_schema_cache();
cache
.tables
.get_mut(&postrust_core::api_request::QualifiedIdentifier::new(
"public", "users",
))
.expect("the fixture has users")
.columns
.get_mut("id")
.expect("with an id")
.always_generated = true;
let names = crate::names::NameOverrides::default();
let config = SchemaConfig {
names: names.clone(),
..SchemaConfig::default()
};
let generated = build_schema(&cache, &config);
let schema = build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(names),
std::time::Duration::from_secs(30),
None,
)
.expect("schema builds");
let sdl = schema.sdl();
let block = |header: &str| {
sdl.split(header)
.nth(1)
.and_then(|rest| rest.split('}').next())
.unwrap_or_else(|| panic!("{} is registered:\n{}", header, sdl))
.to_string()
};
assert!(
block("type users ").contains("id:"),
"still readable:\n{}",
sdl
);
for header in ["input users_insert_input", "input users_set_input"] {
assert!(
!block(header).contains("id:"),
"{} may not name it:\n{}",
header,
block(header)
);
}
assert!(
!sdl.contains("users_inc_input"),
"and nothing is left to increment:\n{}",
sdl
);
assert!(
block("input users_insert_input").contains("name:"),
"and the rest of the table is untouched",
);
}
#[test]
fn test_create_object_type() {
let table = create_test_table("users");
let obj = TableObjectType::from_table(&table);
let _gql_obj = create_object_type(&obj, &[], None);
}
#[test]
fn test_create_query_type() {
let cache = create_test_schema_cache();
let config = SchemaConfig::default();
let generated = build_schema(&cache, &config);
let _query = create_query_type(
&generated,
"query_root",
None,
Arc::new(HashMap::new()),
Arc::new(Default::default()),
);
}
#[test]
fn test_create_mutation_type() {
let cache = create_test_schema_cache();
let config = SchemaConfig::default();
let generated = build_schema(&cache, &config);
let _mutation = create_mutation_type(
&generated,
"mutation_root",
Arc::new(HashMap::new()),
Arc::new(HashMap::new()),
Arc::new(Default::default()),
None,
None,
);
}
#[test]
fn a_scalar_is_named_the_way_a_client_declares_it() {
use crate::types::{pg_type_to_graphql, GraphQLType};
assert_eq!(pg_type_to_graphql("jsonb").to_string(), "jsonb");
assert_eq!(pg_type_to_graphql("json").to_string(), "json");
assert_eq!(pg_type_to_graphql("int8").to_string(), "bigint");
assert_eq!(pg_type_to_graphql("numeric").to_string(), "numeric");
assert_eq!(pg_type_to_graphql("timestamptz").to_string(), "timestamptz");
assert_eq!(pg_type_to_graphql("timestamp").to_string(), "timestamp");
assert_eq!(pg_type_to_graphql("uuid").to_string(), "uuid");
assert_eq!(pg_type_to_graphql("geometry").to_string(), "geometry");
assert_eq!(
pg_type_to_graphql("colors_enum"),
GraphQLType::Custom("colors_enum".to_string())
);
assert_eq!(
leaf_scalar_name(&pg_type_to_graphql("_text")),
"String".to_string()
);
}
#[test]
fn every_table_gets_a_boolean_expression() {
let cache = create_test_schema_cache();
let config = SchemaConfig::default();
let generated = build_schema(&cache, &config);
let (inputs, _) = crate::input::bool_exp::build_inputs(
&generated.object_types,
&generated.relationship_fields,
);
let names: HashSet<String> = inputs.iter().map(|i| i.type_name().to_string()).collect();
for table in generated.object_types.keys() {
let expected = format!("{}_bool_exp", table);
assert!(
names.contains(&expected),
"no {} among {:?}",
expected,
names
);
}
assert!(names.contains("String_comparison_exp"));
assert!(names.contains("Int_comparison_exp"));
}
#[test]
fn a_schema_carrying_the_boolean_expressions_still_builds() {
let cache = create_test_schema_cache();
let config = SchemaConfig::default();
let generated = build_schema(&cache, &config);
let schema = build_dynamic_schema(
&generated,
&cache,
None,
None,
Arc::new(Default::default()),
std::time::Duration::from_secs(30),
None,
);
assert!(schema.is_ok(), "{:?}", schema.err());
let sdl = schema.unwrap().sdl();
assert!(
sdl.contains("_bool_exp"),
"no boolean expressions in:\n{}",
sdl
);
assert!(sdl.contains("_and"), "no _and in the generated expressions");
}
#[test]
fn test_build_schema_with_subscriptions() {
let cache = create_test_schema_cache();
let config = SchemaConfig {
enable_subscriptions: true,
..SchemaConfig::default()
};
let generated = build_schema(&cache, &config);
let sub_fields = generate_subscription_fields(&cache, &generated);
assert!(!sub_fields.is_empty(), "Should have subscription fields");
let result = build_dynamic_schema(
&generated,
&cache,
Some(&sub_fields),
None,
Arc::new(Default::default()),
std::time::Duration::from_secs(30),
None,
);
assert!(result.is_ok(), "Schema with subscriptions should build");
}
#[test]
fn test_subscription_field_generation() {
let cache = create_test_schema_cache();
let config = SchemaConfig::default();
let generated = build_schema(&cache, &config);
let fields = generate_subscription_fields(&cache, &generated);
assert_eq!(fields.len(), 1);
assert_eq!(fields[0].name, "users");
assert_eq!(fields[0].table_name, "users");
assert_eq!(fields[0].channel_name(), "postrust_public_users");
}
#[test]
fn test_create_subscription_type() {
let cache = create_test_schema_cache();
let config = SchemaConfig::default();
let generated = build_schema(&cache, &config);
let _subscription = create_subscription_type(
&generated,
"subscription_root",
None,
Arc::new(Default::default()),
Arc::new(Default::default()),
std::time::Duration::from_secs(30),
);
}
}
#[derive(Debug, PartialEq, Eq)]
enum PathStep {
Key(String),
Index(usize),
}
fn walk_json_path(value: &Value, path: &str) -> Result<Value, async_graphql::Error> {
let mut at = value;
for step in parse_json_path(path)? {
let next = match (&step, at) {
(PathStep::Key(key), Value::Object(map)) => map.get(key.as_str()),
(PathStep::Index(index), Value::List(items)) => items.get(*index),
_ => None,
};
match next {
Some(next) => at = next,
None => return Ok(Value::Null),
}
}
Ok(at.clone())
}
fn unclosed_bracket(path: &str) -> async_graphql::Error {
let message = format!("\"{}\" is not a JSON path: a bracket is never closed", path);
validation_error(&message)
}
fn parse_json_path(path: &str) -> Result<Vec<PathStep>, async_graphql::Error> {
let mut steps = Vec::new();
let mut chars = path.chars().peekable();
if chars.peek() == Some(&'$') {
chars.next();
}
while let Some(&c) = chars.peek() {
match c {
'.' => {
chars.next();
}
'[' => {
chars.next();
let quote = match chars.peek() {
Some(&q @ ('\'' | '"')) => {
chars.next();
Some(q)
}
_ => None,
};
let mut inner = String::new();
loop {
let Some(c) = chars.next() else {
return Err(unclosed_bracket(path));
};
match (quote, c) {
(Some(quote), c) if c == quote => break,
(None, ']') => break,
_ => inner.push(c),
}
}
if quote.is_some() {
match chars.next() {
Some(']') => {}
_ => return Err(unclosed_bracket(path)),
}
}
match (quote, inner.parse::<usize>()) {
(None, Ok(index)) => steps.push(PathStep::Index(index)),
_ => steps.push(PathStep::Key(inner)),
}
}
_ => {
let mut key = String::new();
while let Some(&c) = chars.peek() {
if matches!(c, '.' | '[') {
break;
}
key.push(c);
chars.next();
}
steps.push(PathStep::Key(key));
}
}
}
Ok(steps)
}
#[cfg(test)]
mod json_path_tests {
use super::*;
fn steps(path: &str) -> Vec<PathStep> {
parse_json_path(path).expect("the path parses")
}
fn key(name: &str) -> PathStep {
PathStep::Key(name.to_string())
}
#[test]
fn the_document_itself_has_no_steps() {
assert_eq!(steps("$"), vec![]);
assert_eq!(steps(""), vec![]);
}
#[test]
fn a_key_is_written_bare_or_after_a_dot() {
assert_eq!(steps("a"), vec![key("a")]);
assert_eq!(steps(".obj.c1"), vec![key("obj"), key("c1")]);
assert_eq!(steps("._underscore"), vec![key("_underscore")]);
}
#[test]
fn a_bracket_holds_an_index_or_a_quoted_key() {
assert_eq!(steps("arr[0]"), vec![key("arr"), PathStep::Index(0)]);
assert_eq!(steps("['!@#$%^']"), vec![key("!@#$%^")]);
assert_eq!(
steps("translations['hello world!']"),
vec![key("translations"), key("hello world!")]
);
}
#[test]
fn a_dot_before_a_bracket_changes_nothing() {
let expected = vec![key("objs"), PathStep::Index(0), key("\u{4f60}\u{597d}")];
for spelling in [
"objs[0]['\u{4f60}\u{597d}']",
"objs[0][\"\u{4f60}\u{597d}\"]",
"objs[0].[\"\u{4f60}\u{597d}\"]",
"objs.[0].[\"\u{4f60}\u{597d}\"]",
] {
assert_eq!(steps(spelling), expected, "{}", spelling);
}
}
#[test]
fn a_bracket_that_never_closes_is_refused() {
assert!(parse_json_path("arr[0").is_err());
assert!(parse_json_path("['key'").is_err());
}
#[test]
fn a_step_the_document_does_not_have_reads_null() {
let document = Value::from_json(serde_json::json!({
"obj": {"c1": "c2"},
"arr": [1, 2, 3],
}))
.expect("a document");
let at = |path| walk_json_path(&document, path).expect("the path parses");
assert_eq!(at(".obj.c1"), Value::String("c2".to_string()));
assert_eq!(at("arr[1]"), Value::Number(2.into()));
assert_eq!(at("arr[9]"), Value::Null);
assert_eq!(at(".obj.nothing"), Value::Null);
assert_eq!(at("obj[0]"), Value::Null);
assert_eq!(at("arr.first"), Value::Null);
}
}