1use crate::context::GraphQLContext;
7use crate::error::GraphQLError;
8use crate::schema::object::TableObjectType;
9use crate::schema::relationship::RelationshipField;
10use crate::schema::{build_schema, GeneratedSchema, MutationType, SchemaConfig};
11use crate::subscription::{
12 generate_subscription_fields, NotifyBroker, SubscriptionField as SubField,
13};
14use async_graphql::dynamic::*;
15use async_graphql::Value;
16use async_graphql_axum::{GraphQLRequest, GraphQLResponse};
17use axum::extract::State;
18use axum::response::IntoResponse;
19use postrust_core::schema_cache::SchemaCache;
20use sqlx::PgPool;
21use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
22use std::sync::Arc;
23use tokio::sync::RwLock;
24use tracing::{debug, info, trace};
25
26pub struct GraphQLState {
28 pub pool: PgPool,
30 pub schema_cache: Arc<SchemaCache>,
32 pub generated_schema: GeneratedSchema,
34 pub schema: Schema,
39 pub role_schemas: HashMap<(String, bool), Schema>,
55 pub ungranted_schema: Option<Schema>,
68 pub config: SchemaConfig,
70 pub subscription_fields: Vec<SubField>,
72 pub broker: Arc<RwLock<Option<NotifyBroker>>>,
74}
75
76fn build_role_schemas(
87 schema_cache: &Arc<SchemaCache>,
88 config: &SchemaConfig,
89) -> Result<HashMap<(String, bool), Schema>, GraphQLError> {
90 if !config.names.has_permissions() {
91 return Ok(HashMap::new());
92 }
93
94 let mut schemas = HashMap::new();
95 for role in config.names.roles() {
96 if role == postrust_auth::hasura::ADMIN_ROLE {
100 continue;
101 }
102
103 let variants: &[bool] = match crate::role::has_backend_only(&config.names, role) {
107 true => &[false, true],
108 false => &[false],
109 };
110
111 for &backend in variants {
112 let view = Arc::new(crate::role::cache_for_role(
113 schema_cache,
114 &config.names,
115 role,
116 backend,
117 ));
118 let mut role_config = config.clone();
119 role_config.role = Some(role.to_string());
120
121 let generated = build_schema(&view, &role_config);
122 let built = build_dynamic_schema(
127 &generated,
128 &view,
129 None,
130 role_config.max_rows,
131 Arc::new(role_config.names.clone()),
132 role_config.subscription_refresh(),
133 Some(role),
134 );
135
136 match built {
137 Ok(schema) => {
138 schemas.insert((role.to_string(), backend), schema);
139 }
140 Err(e) => tracing::error!(
147 "GraphQL schema for role \"{}\" could not be built, \
148 so that role is refused: {}",
149 role,
150 e
151 ),
152 }
153 }
154 }
155
156 tracing::info!("GraphQL schemas built for {} roles", schemas.len());
157 Ok(schemas)
158}
159
160fn build_ungranted_schema(
167 schema_cache: &Arc<SchemaCache>,
168 config: &SchemaConfig,
169) -> Option<Schema> {
170 if !config.names.has_permissions() {
171 return None;
172 }
173 const NOBODY: &str = "\u{0}ungranted";
176 let view = Arc::new(crate::role::cache_for_role(
177 schema_cache,
178 &config.names,
179 NOBODY,
180 false,
181 ));
182 let mut role_config = config.clone();
183 role_config.role = Some(NOBODY.to_string());
184 let generated = build_schema(&view, &role_config);
185 build_dynamic_schema(
186 &generated,
187 &view,
188 None,
189 role_config.max_rows,
190 Arc::new(role_config.names.clone()),
191 role_config.subscription_refresh(),
192 Some(NOBODY),
193 )
194 .map_err(|e| {
195 tracing::error!(
196 "the schema for a role with nothing granted could not be built, so such a role is refused instead: {}",
197 e
198 )
199 })
200 .ok()
201}
202
203impl GraphQLState {
204 pub fn new(
206 pool: PgPool,
207 schema_cache: Arc<SchemaCache>,
208 config: SchemaConfig,
209 ) -> Result<Self, GraphQLError> {
210 let generated_schema = build_schema(&schema_cache, &config);
211 let subscription_fields = if config.enable_subscriptions {
212 generate_subscription_fields(&schema_cache, &generated_schema)
213 } else {
214 Vec::new()
215 };
216 let schema = build_dynamic_schema(
217 &generated_schema,
218 &schema_cache,
219 if config.enable_subscriptions {
220 Some(subscription_fields.as_slice())
221 } else {
222 None
223 },
224 config.max_rows,
225 Arc::new(config.names.clone()),
226 config.subscription_refresh(),
227 None,
228 )?;
229
230 let role_schemas = build_role_schemas(&schema_cache, &config)?;
231 let ungranted_schema = build_ungranted_schema(&schema_cache, &config);
232
233 Ok(Self {
234 pool: pool.clone(),
235 schema_cache,
236 generated_schema,
237 schema,
238 role_schemas,
239 ungranted_schema,
240 config,
241 subscription_fields,
242 broker: Arc::new(RwLock::new(None)),
243 })
244 }
245
246 pub fn schema_for(&self, role: Option<&str>, backend: bool) -> Option<&Schema> {
259 if !self.config.names.has_permissions() {
260 return Some(&self.schema);
261 }
262 match role {
263 None | Some(postrust_auth::hasura::ADMIN_ROLE) => Some(&self.schema),
264 Some(role) => self
265 .role_schemas
266 .get(&(role.to_string(), backend))
267 .or_else(|| self.role_schemas.get(&(role.to_string(), false)))
270 .or(self.ungranted_schema.as_ref()),
271 }
272 }
273
274 #[allow(clippy::needless_update)]
276 pub fn rebuild(&mut self) -> Result<(), GraphQLError> {
277 self.role_schemas = build_role_schemas(&self.schema_cache, &self.config)?;
278 self.ungranted_schema = build_ungranted_schema(&self.schema_cache, &self.config);
279 self.generated_schema = build_schema(&self.schema_cache, &self.config);
280 self.subscription_fields = if self.config.enable_subscriptions {
281 generate_subscription_fields(&self.schema_cache, &self.generated_schema)
282 } else {
283 Vec::new()
284 };
285 self.schema = build_dynamic_schema(
286 &self.generated_schema,
287 &self.schema_cache,
288 if self.config.enable_subscriptions {
289 Some(self.subscription_fields.as_slice())
290 } else {
291 None
292 },
293 self.config.max_rows,
294 Arc::new(self.config.names.clone()),
295 self.config.subscription_refresh(),
296 None,
297 )?;
298 Ok(())
299 }
300
301 pub async fn init_subscriptions(&self) -> Result<(), crate::subscription::BrokerError> {
305 if !self.config.enable_subscriptions {
306 return Ok(());
307 }
308
309 let broker = NotifyBroker::new(self.pool.clone());
310
311 let channels: Vec<String> = self
313 .subscription_fields
314 .iter()
315 .map(|f| f.channel_name())
316 .collect();
317
318 if !channels.is_empty() {
319 broker.start(channels).await?;
320 info!(
321 "Subscription broker started with {} channels",
322 self.subscription_fields.len()
323 );
324 }
325
326 let mut broker_guard = self.broker.write().await;
328 *broker_guard = Some(broker);
329
330 Ok(())
331 }
332
333 pub async fn stop_subscriptions(&self) {
335 let broker_guard = self.broker.read().await;
336 if let Some(broker) = broker_guard.as_ref() {
337 broker.stop().await;
338 }
339 }
340
341 pub async fn get_broker(&self) -> Option<Arc<RwLock<Option<NotifyBroker>>>> {
343 Some(Arc::clone(&self.broker))
344 }
345}
346
347pub async fn graphql_handler(
349 State(state): State<Arc<GraphQLState>>,
350 ctx: GraphQLContext,
351 req: GraphQLRequest,
352) -> GraphQLResponse {
353 let request = req
354 .into_inner()
355 .data(ctx)
356 .data(state.pool.clone())
357 .data(Arc::clone(&state.broker));
358 state.schema.execute(request).await.into()
359}
360
361pub async fn graphql_ws_handler(
366 State(state): State<Arc<GraphQLState>>,
367 protocol: async_graphql_axum::GraphQLProtocol,
368 ws: axum::extract::WebSocketUpgrade,
369) -> impl IntoResponse {
370 let schema = state.schema.clone();
371 let pool = state.pool.clone();
372 let broker = Arc::clone(&state.broker);
373
374 ws.protocols(["graphql-transport-ws", "graphql-ws"])
375 .on_upgrade(move |socket| async move {
376 let mut data = async_graphql::Data::default();
377 data.insert(pool);
378 data.insert(broker);
379
380 async_graphql_axum::GraphQLWebSocket::new(socket, schema, protocol)
381 .with_data(data)
382 .serve()
383 .await
384 })
385}
386
387pub async fn graphql_playground() -> impl axum::response::IntoResponse {
389 axum::response::Html(async_graphql::http::playground_source(
390 async_graphql::http::GraphQLPlaygroundConfig::new("/graphql")
391 .subscription_endpoint("/graphql/ws"),
392 ))
393}
394
395fn column_for<'a>(
402 names: &'a crate::names::NameOverrides,
403 schema: &str,
404 table: &str,
405 field: &'a str,
406) -> &'a str {
407 names.column_source(schema, table, field).unwrap_or(field)
408}
409
410fn table_column_for<'a>(
412 names: &'a crate::names::NameOverrides,
413 table: &postrust_core::schema_cache::Table,
414 field: &'a str,
415) -> &'a str {
416 column_for(names, &table.schema, &table.name, field)
417}
418
419fn rename_projection(
427 table: &postrust_core::schema_cache::Table,
428 alias: &str,
429 names: &crate::names::NameOverrides,
430) -> Option<String> {
431 if !names.renames_columns(&table.schema, &table.name) {
432 return None;
433 }
434 let parts: Vec<String> = table
435 .columns
436 .values()
437 .map(|column| {
438 let exposed = names
439 .column(&table.schema, &table.name, &column.name)
440 .unwrap_or(&column.name);
441 format!(
442 "{}.{} AS {}",
443 postrust_sql::escape_ident(alias),
444 postrust_sql::escape_ident(&column.name),
445 postrust_sql::escape_ident(exposed)
446 )
447 })
448 .collect();
449 match parts.is_empty() {
450 true => None,
451 false => Some(parts.join(", ")),
452 }
453}
454
455fn exposed_column_types(
461 table: &postrust_core::schema_cache::Table,
462 names: &crate::names::NameOverrides,
463) -> HashMap<String, String> {
464 table
465 .columns
466 .values()
467 .map(|column| {
468 let exposed = names
469 .column(&table.schema, &table.name, &column.name)
470 .unwrap_or(&column.name);
471 (exposed.to_string(), column.nominal_type.clone())
472 })
473 .collect()
474}
475
476fn write_type_ref(graphql_type: &crate::types::GraphQLType) -> TypeRef {
482 match graphql_type {
483 crate::types::GraphQLType::List(inner) => TypeRef::named_list(leaf_scalar_name(inner)),
484 other => TypeRef::named(leaf_scalar_name(other)),
485 }
486}
487
488fn leaf_scalar_name(graphql_type: &crate::types::GraphQLType) -> String {
490 match graphql_type {
491 crate::types::GraphQLType::List(inner) => leaf_scalar_name(inner),
492 other => other.to_string(),
493 }
494}
495
496fn build_dynamic_schema(
498 generated: &GeneratedSchema,
499 _schema_cache: &SchemaCache,
500 subscription_fields: Option<&[SubField]>,
501 max_rows: Option<i64>,
502 names: Arc<crate::names::NameOverrides>,
503 subscription_refresh: std::time::Duration,
504 role: Option<&str>,
509) -> Result<Schema, GraphQLError> {
510 let mut object_types: HashMap<String, Object> = HashMap::new();
512
513 for (type_name, obj) in &generated.object_types {
518 if obj.fields.is_empty() {
519 continue;
520 }
521 let relationships = generated
522 .relationship_fields
523 .get(type_name)
524 .map(|r| r.as_slice())
525 .unwrap_or(&[]);
526 let table_obj = create_object_type(obj, relationships);
527 object_types.insert(type_name.clone(), table_obj);
528 }
529
530 let countable: HashSet<&str> = generated
536 .query_fields
537 .iter()
538 .filter(|field| field.aggregates && !field.is_by_pk)
539 .map(|field| field.type_name.as_str())
540 .collect();
541 for (type_name, obj) in &generated.object_types {
542 if obj.fields.is_empty() && !countable.contains(type_name.as_str()) {
543 continue;
544 }
545 for aggregate_type in create_aggregate_types(type_name, obj) {
546 object_types.insert(aggregate_type.type_name().to_string(), aggregate_type);
547 }
548 }
549
550 let mutable: HashSet<&str> = generated
554 .mutation_fields
555 .iter()
556 .map(|f| {
557 f.return_type
565 .trim_matches(|c| c == '[' || c == ']' || c == '!')
566 .trim_end_matches("_mutation_response")
567 })
568 .collect();
569 for base_name in mutable {
570 let returning = generated
574 .object_types
575 .get(base_name)
576 .is_some_and(|object| !object.fields.is_empty());
577 object_types.insert(
578 mutation_response_type_name(base_name),
579 create_mutation_response_type(base_name, returning),
580 );
581 }
582
583 let relationships = Arc::new(generated.relationship_fields.clone());
586 let type_names: Arc<HashMap<(String, String), String>> = Arc::new(
592 generated
593 .object_types
594 .iter()
595 .map(|(type_name, object)| {
596 (
597 (object.table.schema.clone(), object.table.name.clone()),
598 type_name.clone(),
599 )
600 })
601 .collect(),
602 );
603 let query = add_function_fields(
604 create_query_type(
605 generated,
606 max_rows,
607 Arc::clone(&relationships),
608 Arc::clone(&names),
609 ),
610 generated,
611 false,
612 max_rows,
613 Arc::clone(&relationships),
614 Arc::clone(&names),
615 );
616
617 let mutation = if !generated.mutation_fields.is_empty()
619 || generated.function_fields.iter().any(|f| f.volatile)
620 {
621 Some(add_function_fields(
622 create_mutation_type(
623 generated,
624 Arc::clone(&relationships),
625 Arc::clone(&type_names),
626 Arc::clone(&names),
627 max_rows,
628 role,
629 ),
630 generated,
631 true,
632 max_rows,
633 Arc::clone(&relationships),
634 Arc::clone(&names),
635 ))
636 } else {
637 None
638 };
639
640 let subscription = subscription_fields.map(|_| {
642 create_subscription_type(
643 generated,
644 max_rows,
645 Arc::clone(&relationships),
646 Arc::clone(&names),
647 subscription_refresh,
648 )
649 });
650
651 let mut builder = Schema::build(
656 "query_root",
657 mutation.as_ref().map(|_| "mutation_root"),
658 subscription.as_ref().map(|_| "subscription_root"),
659 );
660
661 for (_, obj) in object_types {
663 builder = builder.register(obj);
664 }
665
666 builder = builder.register(query);
668
669 if let Some(mutation) = mutation {
671 builder = builder.register(mutation);
672 }
673
674 if let Some(subscription) = subscription {
676 builder = builder.register(subscription);
677 }
678
679 let (bool_exp_inputs, bool_exp_scalars) = crate::input::bool_exp::build_inputs(
686 &generated.object_types,
687 &generated.relationship_fields,
688 );
689 for input in bool_exp_inputs {
690 builder = builder.register(input);
691 }
692
693 let mut scalar_names: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
694 for object in generated.object_types.values() {
695 for field in object.fields.iter().chain(&object.writable_fields) {
701 scalar_names.insert(leaf_scalar_name(&field.graphql_type));
702 }
703 }
704 scalar_names.insert("JSON".to_string());
707 for function in &generated.function_fields {
710 for (_, pg_type, _) in &function.arguments {
711 scalar_names.insert(leaf_scalar_name(&crate::types::pg_type_to_graphql(pg_type)));
712 }
713 }
714 for object in generated.object_types.values() {
717 for field in &object.fields {
718 for (_, pg_type, _) in &field.arguments {
719 scalar_names.insert(leaf_scalar_name(&crate::types::pg_type_to_graphql(pg_type)));
720 }
721 }
722 }
723 for relationships in generated.relationship_fields.values() {
724 for relationship in relationships {
725 for (_, pg_type, _) in &relationship.arguments {
726 scalar_names.insert(leaf_scalar_name(&crate::types::pg_type_to_graphql(pg_type)));
727 }
728 }
729 }
730
731 scalar_names.extend(bool_exp_scalars);
737
738 for name in scalar_names {
739 if matches!(name.as_str(), "Int" | "Float" | "String" | "Boolean" | "ID") {
740 continue;
741 }
742 if generated.enum_types.contains_key(&name) {
746 continue;
747 }
748 builder = builder
749 .register(Scalar::new(&name).description(format!("The PostgreSQL `{}` type.", name)));
750 }
751
752 let mut key_inputs: HashMap<String, InputObject> = HashMap::new();
760 for field in &generated.mutation_fields {
761 if field.mutation_type != MutationType::UpdateByPk || field.pk_columns.is_empty() {
762 continue;
763 }
764 let base_name = field
765 .return_type
766 .trim_matches(|c| c == '[' || c == ']' || c == '!')
767 .trim_end_matches("_mutation_response");
768 let type_name = format!("{}_pk_columns_input", base_name);
769 let mut input = InputObject::new(&type_name)
770 .description(format!("The primary key of one {} row.", base_name));
771 for (column, pg_type) in &field.pk_columns {
772 input = input.field(InputValue::new(
773 names
774 .column(&field.schema_name, &field.table_name, column)
775 .unwrap_or(column),
776 TypeRef::named_nn(pk_argument_type(pg_type)),
777 ));
778 }
779 key_inputs.insert(type_name, input);
780 }
781 for (_, input) in key_inputs {
782 builder = builder.register(input);
783 }
784
785 let insertable_types: HashSet<&str> = generated
806 .object_types
807 .iter()
808 .filter(|(_, object)| crate::input::mutation::is_insertable(&object.table))
809 .map(|(type_name, _)| type_name.as_str())
810 .collect();
811
812 for (type_name, object) in &generated.object_types {
813 let table = &object.table;
814 let writable_named = |allowed: Option<&crate::names::ColumnSet>| {
822 object
823 .writable_fields
824 .iter()
825 .filter(|field| {
826 let column = table_column_for(&names, table, &field.name);
827 table.get_column(column).is_some()
828 && allowed.is_none_or(|set| set.allows(column))
829 })
830 .collect::<Vec<&crate::schema::object::GraphQLField>>()
831 };
832 let writable = writable_named(None);
833 let insert_fields = write_fields(object, &names, role, crate::role::Verb::Insert);
834 let update_fields = write_fields(object, &names, role, crate::role::Verb::Update);
835 if writable.is_empty() {
836 continue;
837 }
838
839 let relationships = generated
840 .relationship_fields
841 .get(type_name)
842 .map(|r| r.as_slice())
843 .unwrap_or(&[]);
844 let conflict_type = match table.unique_constraints.is_empty() {
845 true => None,
846 false => Some(format!("{}_on_conflict", type_name)),
847 };
848
849 if crate::input::mutation::is_insertable(table) {
850 let mut insert = InputObject::new(format!("{}_insert_input", type_name))
851 .description(format!("The columns of a new {} row.", type_name));
852 let mut taken: HashSet<&str> = HashSet::new();
853 for field in &insert_fields {
854 taken.insert(field.name.as_str());
858 insert = insert.field(InputValue::new(
859 &field.name,
860 write_type_ref(&field.graphql_type),
861 ));
862 }
863 for relationship in relationships {
867 if !insertable_types.contains(relationship.target_type.as_str()) {
868 continue;
869 }
870 if !taken.insert(relationship.name.as_str()) {
871 continue;
872 }
873 let suffix = match relationship.is_list {
874 true => "arr_rel_insert_input",
875 false => "obj_rel_insert_input",
876 };
877 insert = insert.field(InputValue::new(
878 &relationship.name,
879 TypeRef::named(format!("{}_{}", relationship.target_type, suffix)),
880 ));
881 }
882 builder = builder.register(insert);
883
884 let data_type = format!("{}_insert_input", type_name);
888 let mut object_rel = InputObject::new(format!("{}_obj_rel_insert_input", type_name))
889 .description(format!("One {} row written beside its parent.", type_name))
890 .field(InputValue::new("data", TypeRef::named_nn(&data_type)));
891 let mut array_rel = InputObject::new(format!("{}_arr_rel_insert_input", type_name))
892 .description(format!("{} rows written beside their parent.", type_name))
893 .field(InputValue::new(
894 "data",
895 TypeRef::named_nn_list_nn(&data_type),
896 ));
897 if let Some(conflict) = &conflict_type {
898 object_rel =
899 object_rel.field(InputValue::new("on_conflict", TypeRef::named(conflict)));
900 array_rel =
901 array_rel.field(InputValue::new("on_conflict", TypeRef::named(conflict)));
902 }
903 builder = builder.register(object_rel).register(array_rel);
904 }
905
906 if crate::input::mutation::is_updatable(table) {
907 let mut set = InputObject::new(format!("{}_set_input", type_name))
908 .description(format!("Columns of {} to replace.", type_name));
909 let mut numeric = InputObject::new(format!("{}_inc_input", type_name))
910 .description(format!("Columns of {} to add to.", type_name));
911 let mut any_numeric = false;
912 let mut any_jsonb = false;
913 for field in &update_fields {
914 if matches!(&field.graphql_type, crate::types::GraphQLType::Json) {
915 any_jsonb = true;
916 }
917 set = set.field(InputValue::new(
918 &field.name,
919 write_type_ref(&field.graphql_type),
920 ));
921 if crate::schema::aggregate::is_incrementable(&field.graphql_type) {
924 any_numeric = true;
925 numeric = numeric.field(InputValue::new(
926 &field.name,
927 TypeRef::named(leaf_scalar_name(&field.graphql_type)),
928 ));
929 }
930 }
931 builder = builder.register(set);
932 if any_numeric {
934 builder = builder.register(numeric);
935 }
936
937 let mut updates = InputObject::new(format!("{}_updates", type_name))
942 .description(format!(
943 "One update to {}: which rows, and what to write.",
944 type_name
945 ))
946 .field(InputValue::new(
947 "where",
948 TypeRef::named_nn(crate::input::bool_exp::bool_exp_type_name(type_name)),
949 ))
950 .field(InputValue::new(
951 "_set",
952 TypeRef::named(format!("{}_set_input", type_name)),
953 ));
954 if any_numeric {
955 updates = updates.field(InputValue::new(
956 "_inc",
957 TypeRef::named(format!("{}_inc_input", type_name)),
958 ));
959 }
960 for (operator, _) in JSONB_OPERATORS {
967 if !any_jsonb {
968 break;
969 }
970 updates = updates.field(InputValue::new(
971 *operator,
972 TypeRef::named(jsonb_operator_input(type_name, operator)),
973 ));
974 }
975 builder = builder.register(updates);
976 if any_jsonb {
977 for (operator, item) in JSONB_OPERATORS {
978 let mut input =
979 InputObject::new(jsonb_operator_input(type_name, operator)).description(
980 format!("Columns of {} to apply `{}` to.", type_name, operator),
981 );
982 for field in &update_fields {
983 if !matches!(&field.graphql_type, crate::types::GraphQLType::Json) {
984 continue;
985 }
986 input = input.field(InputValue::new(&field.name, item()));
987 }
988 builder = builder.register(input);
989 }
990 }
991 }
992 }
993
994 for function in &generated.function_fields {
997 if function.arguments.is_empty() {
998 continue;
999 }
1000 let mut args = InputObject::new(format!("{}_args", function.name))
1001 .description(format!("Arguments to {}.", function.name));
1002 for (name, pg_type, _) in &function.arguments {
1003 let scalar = crate::types::pg_type_to_graphql(pg_type).to_string();
1004 args = args.field(InputValue::new(name, TypeRef::named(scalar)));
1012 }
1013 builder = builder.register(args);
1014 }
1015
1016 for (type_name, relationships) in &generated.relationship_fields {
1021 for relationship in relationships {
1022 if relationship.arguments.is_empty() {
1023 continue;
1024 }
1025 let mut args = InputObject::new(computed_args_type_name(type_name, &relationship.name))
1026 .description(format!(
1027 "Arguments to {}, beside the row it is asked of.",
1028 relationship.name
1029 ));
1030 for (name, pg_type, _) in &relationship.arguments {
1031 let scalar = crate::types::pg_type_to_graphql(pg_type).to_string();
1032 args = args.field(InputValue::new(name, TypeRef::named(scalar)));
1034 }
1035 builder = builder.register(args);
1036 }
1037 }
1038
1039 for (type_name, object) in &generated.object_types {
1044 for field in &object.fields {
1045 if field.arguments.is_empty() {
1046 continue;
1047 }
1048 let mut args = InputObject::new(computed_args_type_name(type_name, &field.name))
1049 .description(format!(
1050 "Arguments to {}, beside the row it is asked of.",
1051 field.name
1052 ));
1053 for (name, pg_type, _) in &field.arguments {
1054 let scalar = crate::types::pg_type_to_graphql(pg_type).to_string();
1055 args = args.field(InputValue::new(name, TypeRef::named(scalar)));
1057 }
1058 builder = builder.register(args);
1059 }
1060 }
1061
1062 {
1066 use crate::input::bool_exp as be;
1067 let mut targets: BTreeSet<&str> = BTreeSet::new();
1068 for relationships in generated.relationship_fields.values() {
1069 for relationship in relationships {
1070 if relationship.is_list {
1071 targets.insert(relationship.target_type.as_str());
1072 }
1073 }
1074 }
1075 for target in targets {
1076 let Some(object) = generated.object_types.get(target) else {
1077 continue;
1078 };
1079 let booleans: Vec<&str> = object
1083 .fields
1084 .iter()
1085 .filter(|field| {
1086 object.table.get_column(&field.name).is_some()
1087 && matches!(field.graphql_type, crate::types::GraphQLType::Boolean)
1088 })
1089 .map(|field| field.name.as_str())
1090 .collect();
1091
1092 let mut over = InputObject::new(be::aggregate_bool_exp_type_name(target)).description(
1093 format!("A question about the whole set of related {} rows.", target),
1094 );
1095 for function in be::AGGREGATE_PREDICATES {
1096 if function != "count" && booleans.is_empty() {
1097 continue;
1098 }
1099 over = over.field(InputValue::new(
1100 function,
1101 TypeRef::named(be::aggregate_bool_exp_function(target, function)),
1102 ));
1103
1104 let (arguments, predicate) = match function {
1107 "count" => (
1108 TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
1109 target,
1110 )),
1111 be::comparison_type_name("Int"),
1112 ),
1113 _ => (
1114 TypeRef::named_nn(be::aggregate_bool_exp_columns(target, function)),
1115 be::comparison_type_name("Boolean"),
1116 ),
1117 };
1118 let mut input = InputObject::new(be::aggregate_bool_exp_function(target, function))
1119 .description(format!(
1120 "`{}` over the related {} rows, and what it has to be.",
1121 function, target
1122 ))
1123 .field(InputValue::new("arguments", arguments))
1124 .field(InputValue::new("distinct", TypeRef::named("Boolean")))
1125 .field(InputValue::new(
1126 "filter",
1127 TypeRef::named(be::bool_exp_type_name(target)),
1128 ))
1129 .field(InputValue::new("predicate", TypeRef::named_nn(predicate)));
1130 if function == "count" {
1131 input = input.description(format!(
1132 "How many related {} rows there have to be.",
1133 target
1134 ));
1135 }
1136 builder = builder.register(input);
1137
1138 if function != "count" {
1139 let mut columns = Enum::new(be::aggregate_bool_exp_columns(target, function))
1140 .description(format!("A boolean column of {}.", target));
1141 for name in &booleans {
1142 columns = columns.item(EnumItem::new(*name));
1143 }
1144 builder = builder.register(columns);
1145 }
1146 }
1147 builder = builder.register(over);
1148 }
1149 }
1150
1151 for (type_name, object) in &generated.object_types {
1155 let updatable_columns = write_fields(object, &names, role, crate::role::Verb::Update);
1161 if object.table.unique_constraints.is_empty() {
1162 continue;
1163 }
1164
1165 let mut constraints = Enum::new(format!("{}_constraint", type_name)).description(format!(
1166 "A uniqueness of {} that an insert may conflict with.",
1167 type_name
1168 ));
1169 for (name, columns) in &object.table.unique_constraints {
1170 constraints = constraints
1171 .item(EnumItem::new(name).description(format!("unique ({})", columns.join(", "))));
1172 }
1173
1174 let mut updatable = Enum::new(format!("{}_update_column", type_name))
1175 .description(format!("A column of {} an upsert may write.", type_name));
1176 for field in &updatable_columns {
1177 updatable = updatable.item(EnumItem::new(&field.name));
1178 }
1179 if updatable_columns.is_empty() {
1188 updatable = updatable.item(
1189 EnumItem::new(PLACEHOLDER_COLUMN)
1190 .description("No column may be written by this upsert."),
1191 );
1192 }
1193
1194 let on_conflict = InputObject::new(format!("{}_on_conflict", type_name))
1195 .description(format!(
1196 "What to do when an insert into {} conflicts.",
1197 type_name
1198 ))
1199 .field(InputValue::new(
1200 "constraint",
1201 TypeRef::named_nn(format!("{}_constraint", type_name)),
1202 ))
1203 .field(
1208 InputValue::new(
1209 "update_columns",
1210 TypeRef::named_nn_list_nn(format!("{}_update_column", type_name)),
1211 )
1212 .default_value(Value::List(Vec::new())),
1213 )
1214 .field(InputValue::new(
1215 "where",
1216 TypeRef::named(crate::input::bool_exp::bool_exp_type_name(type_name)),
1217 ));
1218
1219 builder = builder
1220 .register(constraints)
1221 .register(updatable)
1222 .register(on_conflict);
1223 }
1224
1225 for (type_name, members) in &generated.enum_types {
1227 let mut generated_enum =
1228 Enum::new(type_name).description(format!("The values {} allows.", type_name));
1229 for (value, comment) in members {
1230 let item = EnumItem::new(value);
1231 generated_enum = generated_enum.item(match comment {
1232 Some(description) => item.description(description),
1233 None => item,
1234 });
1235 }
1236 builder = builder.register(generated_enum);
1237 }
1238
1239 let (order_inputs, order_enums) = crate::input::order_by::build_inputs(
1242 &generated.object_types,
1243 &generated.relationship_fields,
1244 );
1245 for input in order_inputs {
1246 builder = builder.register(input);
1247 }
1248 for enum_type in order_enums {
1249 builder = builder.register(enum_type);
1250 }
1251
1252 builder
1253 .finish()
1254 .map_err(|e| GraphQLError::SchemaError(e.to_string()))
1255}
1256
1257const PLACEHOLDER_COLUMN: &str = "_PLACEHOLDER";
1264
1265pub fn mutation_response_type_name(base_name: &str) -> String {
1267 format!("{}_mutation_response", base_name)
1268}
1269
1270fn create_mutation_response_type(base_name: &str, returning: bool) -> Object {
1277 let response_name = mutation_response_type_name(base_name);
1278 let row_type = base_name.to_string();
1279
1280 let response = Object::new(&response_name)
1281 .description(format!("The rows {} changed, and how many.", base_name))
1282 .field(Field::new(
1283 "affected_rows",
1284 TypeRef::named_nn(TypeRef::INT),
1285 |ctx| {
1286 FieldFuture::new(async move {
1287 if let Some(Value::Object(map)) = ctx.parent_value.as_value() {
1288 if let Some(value) = map.get(&async_graphql::Name::new("affected_rows")) {
1289 return Ok(Some(FieldValue::value(value.clone())));
1290 }
1291 }
1292 Ok(Some(FieldValue::value(Value::from(0))))
1293 })
1294 },
1295 ));
1296 if !returning {
1300 return response;
1301 }
1302 response.field(Field::new(
1303 "returning",
1304 TypeRef::named_nn_list_nn(row_type),
1305 |ctx| {
1306 FieldFuture::new(async move {
1307 let rows = match ctx.parent_value.as_value() {
1308 Some(Value::Object(map)) => {
1309 match map.get(&async_graphql::Name::new("returning")) {
1310 Some(Value::List(items)) => items.clone(),
1311 _ => Vec::new(),
1312 }
1313 }
1314 _ => Vec::new(),
1315 };
1316 Ok(Some(FieldValue::list(
1317 rows.into_iter().map(FieldValue::value),
1318 )))
1319 })
1320 },
1321 ))
1322}
1323
1324fn create_aggregate_types(base_name: &str, object: &TableObjectType) -> Vec<Object> {
1330 use crate::schema::aggregate as agg;
1331
1332 let mut types = Vec::new();
1333
1334 let fields_type = agg::aggregate_fields_type_name(base_name);
1336 let has_rows = !object.fields.is_empty();
1340 let mut over = Object::new(agg::aggregate_type_name(base_name))
1341 .description(match has_rows {
1342 true => format!("Aggregates over {}, with the rows themselves.", base_name),
1343 false => format!("Aggregates over {}.", base_name),
1344 })
1345 .field(Field::new(
1346 "aggregate",
1347 TypeRef::named(&fields_type),
1348 |ctx| FieldFuture::new(async move { Ok(child_of(&ctx, "aggregate")) }),
1349 ));
1350 if has_rows {
1351 over = over.field(Field::new(
1352 "nodes",
1353 TypeRef::named_nn_list_nn(base_name.to_string()),
1354 |ctx| {
1355 FieldFuture::new(async move {
1356 let rows = match child_value(&ctx, "nodes") {
1357 Some(Value::List(items)) => items,
1358 _ => Vec::new(),
1359 };
1360 Ok(Some(FieldValue::list(
1361 rows.into_iter().map(FieldValue::value),
1362 )))
1363 })
1364 },
1365 ));
1366 }
1367 types.push(over);
1368
1369 let mut count = Field::new("count", TypeRef::named_nn(TypeRef::INT), |ctx| {
1371 let key = ctx.ctx.field().alias().unwrap_or("count").to_string();
1376 FieldFuture::new(async move {
1377 Ok(Some(FieldValue::value(
1378 child_value(&ctx, &key).unwrap_or(Value::from(0)),
1379 )))
1380 })
1381 });
1382 if has_rows {
1388 count = count
1389 .argument(InputValue::new(
1390 "columns",
1391 TypeRef::named_nn_list(agg_select_column(base_name)),
1392 ))
1393 .argument(InputValue::new("distinct", TypeRef::named("Boolean")));
1394 }
1395 let mut aggregate_fields = Object::new(&fields_type)
1396 .description(format!("Aggregate functions over {}.", base_name))
1397 .field(count);
1398
1399 for (function, returns, columns) in agg::functions_for(object) {
1400 let function_type = agg::function_fields_type_name(base_name, function);
1401 let mut per_column = Object::new(&function_type).description(format!(
1402 "`{}` of each {} column it applies to.",
1403 function, base_name
1404 ));
1405 for column in &columns {
1406 let column_name = column.clone();
1407 let type_name = agg::field_type_for(object, column, returns);
1408 per_column =
1409 per_column.field(Field::new(column, TypeRef::named(type_name), move |ctx| {
1410 let column_name = column_name.clone();
1411 FieldFuture::new(async move {
1412 Ok(child_value(&ctx, &column_name).map(FieldValue::value))
1413 })
1414 }));
1415 }
1416 types.push(per_column);
1417
1418 let owned = function.to_string();
1419 aggregate_fields = aggregate_fields.field(Field::new(
1420 function,
1421 TypeRef::named(&function_type),
1422 move |ctx| {
1423 let owned = owned.clone();
1424 FieldFuture::new(async move { Ok(child_of(&ctx, &owned)) })
1425 },
1426 ));
1427 }
1428
1429 types.push(aggregate_fields);
1430 types
1431}
1432
1433fn agg_select_column(base_name: &str) -> String {
1435 crate::input::order_by::select_column_type_name(base_name)
1436}
1437
1438fn child_value(ctx: &ResolverContext<'_>, key: &str) -> Option<Value> {
1440 match ctx.parent_value.as_value() {
1441 Some(Value::Object(map)) => map.get(&async_graphql::Name::new(key)).cloned(),
1442 _ => None,
1443 }
1444}
1445
1446fn child_of<'a>(ctx: &ResolverContext<'_>, key: &str) -> Option<FieldValue<'a>> {
1451 match child_value(ctx, key) {
1452 None | Some(Value::Null) => None,
1453 Some(value) => Some(FieldValue::value(value)),
1454 }
1455}
1456
1457fn coded_error(code: &'static str, message: impl Into<String>) -> async_graphql::Error {
1463 let mut error = async_graphql::Error::new(message);
1464 let mut extensions = async_graphql::ErrorExtensionValues::default();
1465 extensions.set("code", code);
1466 error.extensions = Some(extensions);
1467 error
1468}
1469
1470fn validation_error(message: impl Into<String>) -> async_graphql::Error {
1472 coded_error("validation-failed", message)
1473}
1474
1475fn database_error(error: sqlx::Error) -> async_graphql::Error {
1487 let sqlx::Error::Database(db) = &error else {
1488 return async_graphql::Error::new(error.to_string());
1489 };
1490 let described = match db.code().as_deref() {
1491 Some("23505") => "Uniqueness violation. ",
1494 Some("23502") => "Not-NULL violation. ",
1495 Some("23503") => "Foreign key violation. ",
1496 Some("23514") => "Check constraint violation. ",
1497 _ => "",
1501 };
1502 let error = async_graphql::Error::new(format!("{}{}", described, db.message()));
1503 let coded = match db.code().as_deref() {
1508 Some(code) if code.starts_with("23") => Some("constraint-violation"),
1509 Some("22025") => Some("bad-request"),
1513 Some(code) if code.starts_with("22") => Some("data-exception"),
1515 _ => None,
1516 };
1517 match coded {
1518 Some(code) => at_code(error, code),
1519 None => error,
1520 }
1521}
1522
1523fn at_code(mut error: async_graphql::Error, code: &'static str) -> async_graphql::Error {
1525 let mut extensions = error.extensions.take().unwrap_or_default();
1526 extensions.set("code", code);
1527 error.extensions = Some(extensions);
1528 error
1529}
1530
1531fn at_path(mut error: async_graphql::Error, path: &str) -> async_graphql::Error {
1540 let mut extensions = error.extensions.take().unwrap_or_default();
1541 extensions.set("path", path);
1542 error.extensions = Some(extensions);
1543 error
1544}
1545
1546fn create_object_type(obj: &TableObjectType, relationships: &[RelationshipField]) -> Object {
1547 let mut object = Object::new(&obj.name);
1548
1549 object = match obj.description() {
1553 Some("") => object,
1556 Some(desc) => object.description(desc),
1557 None => object.description(format!(
1558 "columns and relationships of \"{}\"",
1559 obj.table_name()
1560 )),
1561 };
1562
1563 let mut taken: HashSet<String> = obj.fields.iter().map(|f| f.name.clone()).collect();
1576
1577 let mut fields: Vec<(String, Field)> = Vec::new();
1581
1582 for field in &obj.fields {
1583 let field_name = field.name.clone();
1584 let field_type = graphql_type_ref(&field.type_string());
1585 let takes_path = matches!(&field.graphql_type, crate::types::GraphQLType::Json)
1591 || matches!(
1592 &field.graphql_type,
1593 crate::types::GraphQLType::Custom(name) if name == "json"
1594 );
1595
1596 let gql_field = Field::new(&field.name, field_type, move |ctx| {
1600 let field_name = field_name.clone();
1601 FieldFuture::new(async move {
1602 if let Some(Value::Object(map)) = ctx.parent_value.as_value() {
1604 let key = async_graphql::Name::new(&field_name);
1606 let path = match takes_path {
1607 true => ctx.args.get("path").and_then(|v| v.string().ok()),
1608 false => None,
1609 };
1610 match map.get(&key) {
1611 Some(Value::Null) => return Ok(None),
1616 Some(val) => {
1617 let val = match path {
1618 Some(path) => walk_json_path(val, path)?,
1619 None => val.clone(),
1620 };
1621 return match val {
1622 Value::Null => Ok(None),
1623 val => Ok(Some(FieldValue::value(val))),
1624 };
1625 }
1626 None => {}
1627 }
1628 }
1629
1630 Ok(None)
1632 })
1633 });
1634
1635 let gql_field = match takes_path {
1636 true => gql_field.argument(
1637 InputValue::new("path", TypeRef::named("String")).description("JSON select path"),
1638 ),
1639 false => gql_field,
1640 };
1641
1642 let gql_field = match field.arguments.is_empty() {
1646 true => gql_field,
1647 false => gql_field.argument(InputValue::new(
1648 "args",
1649 TypeRef::named_nn(computed_args_type_name(&obj.name, &field.name)),
1650 )),
1651 };
1652
1653 let gql_field = if let Some(desc) = &field.description {
1654 gql_field.description(desc)
1655 } else {
1656 gql_field
1657 };
1658
1659 fields.push((field.name.clone(), gql_field));
1660 }
1661
1662 for rel in relationships {
1666 if !taken.insert(rel.name.clone()) {
1667 tracing::warn!(
1668 "{}: not exposing the relationship to \"{}\" as `{}` -- the table \
1669 already has a column of that name",
1670 obj.name,
1671 rel.target_type,
1672 rel.name
1673 );
1674 continue;
1675 }
1676 let field_name = rel.name.clone();
1677 let field_type = if rel.is_list {
1678 TypeRef::named_nn_list_nn(&rel.target_type)
1679 } else if always_present(rel, &obj.table) {
1680 TypeRef::named_nn(&rel.target_type)
1681 } else {
1682 TypeRef::named(&rel.target_type)
1683 };
1684
1685 let is_list = rel.is_list;
1686 let mut gql_field = Field::new(&rel.name, field_type, move |ctx| {
1687 let field_name = field_name.clone();
1688 FieldFuture::new(async move {
1689 if let Some(Value::Object(map)) = ctx.parent_value.as_value() {
1690 let key = async_graphql::Name::new(&field_name);
1691 match map.get(&key) {
1692 Some(Value::Null) if !is_list => return Ok(None),
1697 Some(val) => return Ok(Some(FieldValue::value(val.clone()))),
1698 None => {}
1699 }
1700 }
1701 Ok(None)
1702 })
1703 });
1704
1705 if !rel.arguments.is_empty() {
1716 gql_field = gql_field.argument(InputValue::new(
1717 "args",
1718 TypeRef::named_nn(computed_args_type_name(&obj.name, &rel.name)),
1719 ));
1720 }
1721 if rel.is_list {
1722 gql_field = gql_field
1723 .argument(InputValue::new(
1724 "distinct_on",
1725 TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
1726 &rel.target_type,
1727 )),
1728 ))
1729 .argument(InputValue::new(
1730 "where",
1731 TypeRef::named(crate::input::bool_exp::bool_exp_type_name(&rel.target_type)),
1732 ))
1733 .argument(InputValue::new(
1734 "order_by",
1735 TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(
1736 &rel.target_type,
1737 )),
1738 ))
1739 .argument(InputValue::new("limit", TypeRef::named("Int")))
1740 .argument(InputValue::new("offset", TypeRef::named("Int")));
1741 }
1742
1743 let gql_field = if let Some(desc) = &rel.description {
1744 gql_field.description(desc)
1745 } else {
1746 gql_field
1747 };
1748
1749 fields.push((rel.name.clone(), gql_field));
1750
1751 if rel.is_list {
1756 let aggregate_field = format!("{}_aggregate", rel.name);
1757 if taken.insert(aggregate_field.clone()) {
1758 let key = aggregate_field.clone();
1759 fields.push((
1760 aggregate_field.clone(),
1761 Field::new(
1762 &aggregate_field,
1763 TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
1764 &rel.target_type,
1765 )),
1766 move |ctx| {
1767 let key = key.clone();
1768 FieldFuture::new(async move {
1769 Ok(Some(FieldValue::value(
1773 child_value(&ctx, &key).unwrap_or(Value::Null),
1774 )))
1775 })
1776 },
1777 )
1778 .argument(InputValue::new(
1783 "distinct_on",
1784 TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
1785 &rel.target_type,
1786 )),
1787 ))
1788 .argument(InputValue::new(
1789 "where",
1790 TypeRef::named(crate::input::bool_exp::bool_exp_type_name(
1791 &rel.target_type,
1792 )),
1793 ))
1794 .argument(InputValue::new(
1795 "order_by",
1796 TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(
1797 &rel.target_type,
1798 )),
1799 ))
1800 .argument(InputValue::new("limit", TypeRef::named("Int")))
1801 .argument(InputValue::new("offset", TypeRef::named("Int")))
1802 .description(format!("Aggregates over {}.", rel.name)),
1803 ));
1804 if !rel.arguments.is_empty() {
1807 let last = fields.len() - 1;
1808 let (name, field) = fields.remove(last);
1809 fields.push((
1810 name,
1811 field.argument(InputValue::new(
1812 "args",
1813 TypeRef::named_nn(computed_args_type_name(&obj.name, &rel.name)),
1814 )),
1815 ));
1816 }
1817 }
1818 }
1819 }
1820
1821 fields.sort_by(|(a, _), (b, _)| a.cmp(b));
1822 for (_, field) in fields {
1823 object = object.field(field);
1824 }
1825
1826 object
1827}
1828
1829trait RootField: Sized {
1836 fn with_argument(self, value: InputValue) -> Self;
1837}
1838
1839impl RootField for Field {
1840 fn with_argument(self, value: InputValue) -> Self {
1841 self.argument(value)
1842 }
1843}
1844
1845impl RootField for SubscriptionField {
1846 fn with_argument(self, value: InputValue) -> Self {
1847 self.argument(value)
1848 }
1849}
1850
1851fn with_row_arguments<F: RootField>(field: F, type_name: &str) -> F {
1854 with_row_arguments_named(field, type_name, true)
1855}
1856
1857fn with_row_arguments_named<F: RootField>(field: F, type_name: &str, columns: bool) -> F {
1865 let mut field = field;
1866 if columns {
1867 field = field.with_argument(InputValue::new(
1868 "distinct_on",
1869 TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(type_name)),
1870 ));
1871 }
1872 field = field
1873 .with_argument(InputValue::new("limit", TypeRef::named("Int")))
1874 .with_argument(InputValue::new("offset", TypeRef::named("Int")));
1875 if columns {
1876 field = field.with_argument(InputValue::new(
1877 "order_by",
1878 TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(type_name)),
1879 ));
1880 }
1881 field.with_argument(InputValue::new(
1882 "where",
1883 TypeRef::named(crate::input::bool_exp::bool_exp_type_name(type_name)),
1884 ))
1885}
1886
1887fn with_key_arguments<F: RootField>(
1890 field: F,
1891 schema_name: &str,
1892 table_name: &str,
1893 pk_columns: &[(String, String)],
1894 names: &crate::names::NameOverrides,
1895) -> F {
1896 let mut field = field;
1897 for (column, pg_type) in pk_columns {
1898 field = field.with_argument(InputValue::new(
1899 names
1900 .column(schema_name, table_name, column)
1901 .unwrap_or(column),
1902 TypeRef::named_nn(pk_argument_type(pg_type)),
1903 ));
1904 }
1905 field
1906}
1907
1908fn create_query_type(
1910 generated: &GeneratedSchema,
1911 max_rows: Option<i64>,
1912 relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
1913 names: Arc<crate::names::NameOverrides>,
1914) -> Object {
1915 let mut roots: Vec<(String, Field)> = Vec::new();
1920
1921 for field in &generated.query_fields {
1922 let table_name = field.table_name.clone();
1923 let schema_name = field.schema_name.clone();
1924 let type_name = field.type_name.clone();
1925 let is_by_pk = field.is_by_pk;
1926 let pk_columns = field.pk_columns.clone();
1927 let return_type = graphql_type_ref(&field.return_type);
1928
1929 let spec_type_name = type_name.clone();
1930 let spec = Arc::new(QueryFieldSpec {
1931 schema_name,
1932 table_name,
1933 type_name,
1934 is_by_pk,
1935 pk_columns: pk_columns.clone(),
1936 max_rows,
1937 relationships: Arc::clone(&relationships),
1938 names: Arc::clone(&names),
1939 call: None,
1940 });
1941
1942 let mut gql_field = Field::new(&field.name, return_type, move |ctx| {
1943 let spec = Arc::clone(&spec);
1944 FieldFuture::new(async move { resolve_query(&ctx, &spec).await })
1945 });
1946
1947 gql_field = match is_by_pk {
1949 false => with_row_arguments(gql_field, &spec_type_name),
1950 true => with_key_arguments(
1951 gql_field,
1952 &field.schema_name,
1953 &field.table_name,
1954 &pk_columns,
1955 names.as_ref(),
1956 ),
1957 };
1958
1959 if let Some(desc) = &field.description {
1960 gql_field = gql_field.description(desc);
1961 }
1962
1963 if field.rows {
1968 roots.push((field.name.clone(), gql_field));
1969 }
1970
1971 if !is_by_pk && field.aggregates {
1979 let agg_spec = Arc::new(AggregateSpec {
1980 schema_name: field.schema_name.clone(),
1981 table_name: field.table_name.clone(),
1982 type_name: field.type_name.clone(),
1983 max_rows,
1984 relationships: Arc::clone(&relationships),
1985 names: Arc::clone(&names),
1986 call: None,
1987 });
1988 let aggregate_field_name = field
1989 .aggregate_name
1990 .clone()
1991 .unwrap_or_else(|| crate::schema::aggregate::aggregate_type_name(&field.type_name));
1992 let aggregate_field_name_for_sorting = aggregate_field_name.clone();
1993 let mut agg_field = Field::new(
1994 aggregate_field_name,
1995 TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
1996 &field.type_name,
1997 )),
1998 move |ctx| {
1999 let agg_spec = Arc::clone(&agg_spec);
2000 FieldFuture::new(async move { resolve_aggregate(&ctx, &agg_spec).await })
2001 },
2002 );
2003 agg_field = with_row_arguments_named(agg_field, &spec_type_name, field.rows);
2004 agg_field = match field.aggregate_description.as_deref() {
2005 Some("") => agg_field,
2006 Some(given) => agg_field.description(given),
2007 None => agg_field.description(format!(
2008 "fetch aggregated fields from the table: \"{}\"",
2009 field.table_name
2010 )),
2011 };
2012 roots.push((aggregate_field_name_for_sorting, agg_field));
2013 }
2014 }
2015
2016 roots.sort_by(|(a, _), (b, _)| a.cmp(b));
2017 let mut query = Object::new("query_root");
2018 if roots.is_empty() {
2022 return query.field(
2023 Field::new("no_queries_available", TypeRef::named("String"), |_| {
2024 FieldFuture::new(async move { Ok(None::<Value>) })
2025 })
2026 .description("There are no queries available to the current role."),
2027 );
2028 }
2029 for (_, field) in roots {
2030 query = query.field(field);
2031 }
2032
2033 query
2034}
2035
2036type JsonbOperator = (&'static str, fn() -> TypeRef);
2038
2039const JSONB_OPERATORS: &[JsonbOperator] = &[
2046 ("_append", || TypeRef::named("jsonb")),
2047 ("_delete_at_path", || TypeRef::named_nn_list("String")),
2048 ("_delete_elem", || TypeRef::named("Int")),
2049 ("_delete_key", || TypeRef::named("String")),
2050 ("_prepend", || TypeRef::named("jsonb")),
2051];
2052
2053fn jsonb_operator_input(type_name: &str, operator: &str) -> String {
2055 format!("{}{}_input", type_name, operator)
2056}
2057
2058fn with_update_operators(
2065 field: Field,
2066 base_name: &str,
2067 has_numeric: bool,
2068 has_jsonb: bool,
2069) -> Field {
2070 let mut field = field;
2071 if has_jsonb {
2074 for (operator, _) in JSONB_OPERATORS {
2075 field = field.argument(InputValue::new(
2076 *operator,
2077 TypeRef::named(jsonb_operator_input(base_name, operator)),
2078 ));
2079 }
2080 }
2081 if has_numeric {
2082 field = field.argument(InputValue::new(
2083 "_inc",
2084 TypeRef::named(format!("{}_inc_input", base_name)),
2085 ));
2086 }
2087 field.argument(InputValue::new(
2088 "_set",
2089 TypeRef::named(format!("{}_set_input", base_name)),
2090 ))
2091}
2092
2093fn write_fields<'a>(
2103 object: &'a crate::schema::object::TableObjectType,
2104 names: &crate::names::NameOverrides,
2105 role: Option<&str>,
2106 verb: crate::role::Verb,
2107) -> Vec<&'a crate::schema::object::GraphQLField> {
2108 let allowed = role.and_then(|role| {
2109 names
2110 .permissions(&object.table.schema, &object.table.name, role)
2111 .and_then(|granted| granted.write_columns(verb))
2112 });
2113 object
2114 .writable_fields
2115 .iter()
2116 .filter(|field| {
2117 let column = table_column_for(names, &object.table, &field.name);
2118 object
2126 .table
2127 .get_column(column)
2128 .is_some_and(|c| !c.always_generated)
2129 && allowed.is_none_or(|set| set.allows(column))
2130 })
2131 .collect()
2132}
2133
2134fn update_inputs(
2136 object: &crate::schema::object::TableObjectType,
2137 names: &crate::names::NameOverrides,
2138 role: Option<&str>,
2139) -> (bool, bool) {
2140 let fields = write_fields(object, names, role, crate::role::Verb::Update);
2141 (
2142 fields
2143 .iter()
2144 .any(|f| crate::schema::aggregate::is_incrementable(&f.graphql_type)),
2145 fields
2146 .iter()
2147 .any(|f| matches!(&f.graphql_type, crate::types::GraphQLType::Json)),
2148 )
2149}
2150
2151fn create_mutation_type(
2152 generated: &GeneratedSchema,
2153 relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
2154 type_names: Arc<HashMap<(String, String), String>>,
2155 names: Arc<crate::names::NameOverrides>,
2156 max_rows: Option<i64>,
2157 role: Option<&str>,
2158) -> Object {
2159 let mut mutation = Object::new("mutation_root");
2160 let mut roots: Vec<(String, Field)> = Vec::new();
2162
2163 let mut has_numeric_column: HashSet<String> = HashSet::new();
2170 let mut has_jsonb_column: HashSet<String> = HashSet::new();
2171 for (type_name, object) in &generated.object_types {
2172 let (numeric, jsonb) = update_inputs(object, &names, role);
2173 if numeric {
2174 has_numeric_column.insert(type_name.clone());
2175 }
2176 if jsonb {
2177 has_jsonb_column.insert(type_name.clone());
2178 }
2179 }
2180
2181 let has_conflict_target: HashSet<String> = generated
2182 .object_types
2183 .iter()
2184 .filter(|(_, object)| !object.table.unique_constraints.is_empty())
2185 .map(|(type_name, _)| type_name.clone())
2186 .collect();
2187
2188 for field in &generated.mutation_fields {
2189 let table_name = field.table_name.clone();
2190 let schema_name = field.schema_name.clone();
2191 let mutation_type = field.mutation_type;
2192 let pk_columns = field.pk_columns.clone();
2193 let return_type = graphql_type_ref(&field.return_type);
2194
2195 let where_type = field
2199 .return_type
2200 .trim_matches(|c| c == '[' || c == ']' || c == '!')
2201 .trim_end_matches("_mutation_response")
2202 .to_string();
2203
2204 let resolver_pk_columns = pk_columns.clone();
2205 let field_relationships = Arc::clone(&relationships);
2206 let field_type_names = Arc::clone(&type_names);
2207 let field_names = Arc::clone(&names);
2208 let mut gql_field = Field::new(&field.name, return_type, move |ctx| {
2209 let table_name = table_name.clone();
2210 let schema_name = schema_name.clone();
2211 let pk_columns = resolver_pk_columns.clone();
2212 let relationships = Arc::clone(&field_relationships);
2213 let type_names = Arc::clone(&field_type_names);
2214 let names = Arc::clone(&field_names);
2215 FieldFuture::new(async move {
2216 resolve_mutation(
2217 &ctx,
2218 &schema_name,
2219 &table_name,
2220 mutation_type,
2221 &pk_columns,
2222 &relationships,
2223 &type_names,
2224 &names,
2225 max_rows,
2226 )
2227 .await
2228 })
2229 });
2230
2231 match mutation_type {
2237 MutationType::Insert | MutationType::InsertOne => {
2238 let insert_input = format!("{}_insert_input", where_type);
2239 gql_field = if mutation_type == MutationType::Insert {
2240 gql_field.argument(InputValue::new(
2241 "objects",
2242 TypeRef::named_nn_list_nn(&insert_input),
2243 ))
2244 } else {
2245 gql_field.argument(InputValue::new("object", TypeRef::named_nn(&insert_input)))
2248 };
2249 if has_conflict_target.contains(&where_type) {
2250 gql_field = gql_field.argument(InputValue::new(
2251 "on_conflict",
2252 TypeRef::named(format!("{}_on_conflict", where_type)),
2253 ));
2254 }
2255 }
2256 MutationType::UpdateByPk => {
2257 gql_field = gql_field.argument(InputValue::new(
2258 "pk_columns",
2259 TypeRef::named_nn(format!("{}_pk_columns_input", where_type)),
2260 ));
2261 gql_field = with_update_operators(
2262 gql_field,
2263 &where_type,
2264 has_numeric_column.contains(&where_type),
2265 has_jsonb_column.contains(&where_type),
2266 );
2267 }
2268 MutationType::Update => {
2269 gql_field = gql_field.argument(InputValue::new(
2270 "where",
2271 TypeRef::named_nn(crate::input::bool_exp::bool_exp_type_name(&where_type)),
2272 ));
2273 gql_field = with_update_operators(
2274 gql_field,
2275 &where_type,
2276 has_numeric_column.contains(&where_type),
2277 has_jsonb_column.contains(&where_type),
2278 );
2279 }
2280 MutationType::UpdateMany => {
2281 gql_field = gql_field.argument(InputValue::new(
2282 "updates",
2283 TypeRef::named_nn_list_nn(format!("{}_updates", where_type)),
2284 ));
2285 }
2286 MutationType::DeleteByPk => {
2287 for (col_name, pg_type) in &pk_columns {
2288 gql_field = gql_field.argument(InputValue::new(
2289 names
2290 .column(&field.schema_name, &field.table_name, col_name)
2291 .unwrap_or(col_name),
2292 TypeRef::named_nn(pk_argument_type(pg_type)),
2293 ));
2294 }
2295 }
2296 MutationType::Delete => {
2297 gql_field = gql_field.argument(InputValue::new(
2304 "where",
2305 TypeRef::named_nn(crate::input::bool_exp::bool_exp_type_name(&where_type)),
2306 ));
2307 }
2308 }
2309
2310 if let Some(desc) = &field.description {
2311 gql_field = gql_field.description(desc);
2312 }
2313
2314 roots.push((field.name.clone(), gql_field));
2315 }
2316
2317 roots.sort_by(|(a, _), (b, _)| a.cmp(b));
2318 for (_, field) in roots {
2319 mutation = mutation.field(field);
2320 }
2321
2322 mutation
2323}
2324
2325fn create_subscription_type(
2342 generated: &GeneratedSchema,
2343 max_rows: Option<i64>,
2344 relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
2345 names: Arc<crate::names::NameOverrides>,
2346 refresh: std::time::Duration,
2347) -> Subscription {
2348 let mut roots: Vec<(String, SubscriptionField)> = Vec::new();
2349
2350 for field in &generated.query_fields {
2351 let spec_type_name = field.type_name.clone();
2352 let pk_columns = field.pk_columns.clone();
2353 let is_by_pk = field.is_by_pk;
2354 let return_type = graphql_type_ref(&field.return_type);
2355 let watched = vec![crate::subscription::table_channel_name(
2356 &field.schema_name,
2357 &field.table_name,
2358 )];
2359
2360 let spec = Arc::new(QueryFieldSpec {
2361 schema_name: field.schema_name.clone(),
2362 table_name: field.table_name.clone(),
2363 type_name: field.type_name.clone(),
2364 is_by_pk,
2365 pk_columns: pk_columns.clone(),
2366 max_rows,
2367 relationships: Arc::clone(&relationships),
2368 names: Arc::clone(&names),
2369 call: None,
2370 });
2371
2372 let mut gql_field = SubscriptionField::new(&field.name, return_type, move |ctx| {
2373 let spec = Arc::clone(&spec);
2374 let watched = watched.clone();
2375 SubscriptionFieldFuture::new(async move {
2376 let wake = wake_on(&ctx, &watched, refresh).await;
2377 Ok(live(ctx, LiveSource::Rows(spec), wake))
2378 })
2379 });
2380
2381 gql_field = match is_by_pk {
2382 false => with_row_arguments(gql_field, &spec_type_name),
2383 true => with_key_arguments(
2384 gql_field,
2385 &field.schema_name,
2386 &field.table_name,
2387 &pk_columns,
2388 names.as_ref(),
2389 ),
2390 };
2391 if let Some(desc) = &field.description {
2392 gql_field = gql_field.description(desc);
2393 }
2394 if field.rows {
2396 roots.push((field.name.clone(), gql_field));
2397 }
2398
2399 if is_by_pk || !field.aggregates {
2402 continue;
2403 }
2404
2405 let agg_spec = Arc::new(AggregateSpec {
2407 schema_name: field.schema_name.clone(),
2408 table_name: field.table_name.clone(),
2409 type_name: field.type_name.clone(),
2410 max_rows,
2411 relationships: Arc::clone(&relationships),
2412 names: Arc::clone(&names),
2413 call: None,
2414 });
2415 let agg_watched = vec![crate::subscription::table_channel_name(
2416 &field.schema_name,
2417 &field.table_name,
2418 )];
2419 let agg_name = field
2420 .aggregate_name
2421 .clone()
2422 .unwrap_or_else(|| crate::schema::aggregate::aggregate_type_name(&field.type_name));
2423 let mut agg_field = SubscriptionField::new(
2424 agg_name.clone(),
2425 TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
2426 &field.type_name,
2427 )),
2428 move |ctx| {
2429 let agg_spec = Arc::clone(&agg_spec);
2430 let watched = agg_watched.clone();
2431 SubscriptionFieldFuture::new(async move {
2432 let wake = wake_on(&ctx, &watched, refresh).await;
2433 Ok(live(ctx, LiveSource::Aggregate(agg_spec), wake))
2434 })
2435 },
2436 );
2437 agg_field = with_row_arguments_named(agg_field, &spec_type_name, field.rows);
2438 agg_field = match field.aggregate_description.as_deref() {
2439 Some("") => agg_field,
2440 Some(given) => agg_field.description(given),
2441 None => agg_field.description(format!(
2442 "fetch aggregated fields from the table: \"{}\"",
2443 field.table_name
2444 )),
2445 };
2446 roots.push((agg_name, agg_field));
2447 }
2448
2449 roots.sort_by(|(a, _), (b, _)| a.cmp(b));
2450 let mut subscription = Subscription::new("subscription_root");
2451 if roots.is_empty() {
2454 return subscription.field(SubscriptionField::new(
2455 "no_subscriptions_available",
2456 TypeRef::named("String"),
2457 |_| {
2458 SubscriptionFieldFuture::new(async move {
2459 Ok(futures::stream::empty::<
2460 Result<FieldValue, async_graphql::Error>,
2461 >())
2462 })
2463 },
2464 ));
2465 }
2466 for (_, field) in roots {
2467 subscription = subscription.field(field);
2468 }
2469 subscription
2470}
2471
2472async fn wake_on(
2478 ctx: &ResolverContext<'_>,
2479 channels: &[String],
2480 refresh: std::time::Duration,
2481) -> futures::stream::BoxStream<'static, ()> {
2482 use futures::StreamExt;
2483
2484 let mut streams: Vec<futures::stream::BoxStream<'static, ()>> =
2485 Vec::with_capacity(channels.len() + 1);
2486 streams.push(
2489 tokio_stream::wrappers::IntervalStream::new({
2490 let mut interval = tokio::time::interval(refresh);
2491 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
2492 interval
2493 })
2494 .map(|_| ())
2495 .boxed(),
2496 );
2497
2498 if let Ok(broker_arc) = ctx.data::<Arc<RwLock<Option<NotifyBroker>>>>() {
2499 let guard = broker_arc.read().await;
2500 if let Some(broker) = guard.as_ref() {
2501 for channel in channels {
2502 streams.push(
2506 broker
2507 .subscribe_or_create(channel)
2508 .await
2509 .map(|_| ())
2510 .boxed(),
2511 );
2512 }
2513 }
2514 }
2515
2516 futures::stream::select_all(streams).boxed()
2517}
2518
2519enum LiveSource {
2525 Rows(Arc<QueryFieldSpec>),
2526 Aggregate(Arc<AggregateSpec>),
2527}
2528
2529async fn read_live<'a>(
2531 ctx: &ResolverContext<'a>,
2532 source: &LiveSource,
2533) -> Result<(serde_json::Value, Option<FieldValue<'a>>), async_graphql::Error> {
2534 match source {
2535 LiveSource::Rows(spec) => {
2536 let rows = query_rows(ctx, spec).await?;
2537 Ok((
2538 serde_json::Value::Array(rows.clone()),
2539 rows_as_field_value(rows, spec.is_by_pk),
2540 ))
2541 }
2542 LiveSource::Aggregate(spec) => {
2543 let value = aggregate_value(ctx, spec).await?;
2544 let seen = serde_json::to_value(&value).unwrap_or_default();
2545 Ok((seen, Some(FieldValue::value(value))))
2546 }
2547 }
2548}
2549
2550fn live<'a>(
2558 ctx: ResolverContext<'a>,
2559 source: LiveSource,
2560 wake: futures::stream::BoxStream<'static, ()>,
2561) -> impl futures::Stream<Item = Result<FieldValue<'a>, async_graphql::Error>> + Send + 'a {
2562 use futures::StreamExt;
2563
2564 struct Live<'a> {
2565 ctx: ResolverContext<'a>,
2566 source: LiveSource,
2567 wake: futures::stream::BoxStream<'static, ()>,
2568 sent: Option<serde_json::Value>,
2569 started: bool,
2570 }
2571
2572 futures::stream::unfold(
2573 Live {
2574 ctx,
2575 source,
2576 wake,
2577 sent: None,
2578 started: false,
2579 },
2580 |mut live: Live<'a>| async move {
2581 loop {
2582 if live.started {
2583 live.wake.next().await?;
2584 } else {
2585 live.started = true;
2586 }
2587 let read = read_live(&live.ctx, &live.source).await;
2588 match read {
2589 Err(e) => return Some((Err(e), live)),
2593 Ok((seen, value)) => {
2594 if live.sent.as_ref() == Some(&seen) {
2595 continue;
2596 }
2597 live.sent = Some(seen);
2598 let value = value.unwrap_or_else(|| FieldValue::value(Value::Null));
2599 return Some((Ok(value), live));
2600 }
2601 }
2602 }
2603 },
2604 )
2605 .take_while(|item| futures::future::ready(item.is_ok()))
2606}
2607
2608struct QueryFieldSpec {
2610 schema_name: String,
2611 table_name: String,
2612 type_name: String,
2613 is_by_pk: bool,
2614 pk_columns: Vec<(String, String)>,
2615 max_rows: Option<i64>,
2616 relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
2617 names: Arc<crate::names::NameOverrides>,
2618 call: Option<Arc<FunctionCall>>,
2621}
2622
2623fn add_function_fields(
2630 mut root: Object,
2631 generated: &GeneratedSchema,
2632 volatile: bool,
2633 max_rows: Option<i64>,
2634 relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
2635 names: Arc<crate::names::NameOverrides>,
2636) -> Object {
2637 let taken: HashSet<&str> = if volatile {
2642 generated
2643 .mutation_fields
2644 .iter()
2645 .map(|f| f.name.as_str())
2646 .collect()
2647 } else {
2648 generated
2649 .query_fields
2650 .iter()
2651 .map(|f| f.name.as_str())
2652 .collect()
2653 };
2654
2655 for function in &generated.function_fields {
2656 if function.volatile != volatile {
2657 continue;
2658 }
2659 if taken.contains(function.name.as_str()) {
2660 tracing::warn!(
2661 "not exposing {}.{}({}) -- a table already has a root field \
2662 called {}",
2663 function.schema_name,
2664 function.function_name,
2665 function.argument_signature(),
2666 function.name,
2667 );
2668 continue;
2669 }
2670 let spec = Arc::new(QueryFieldSpec {
2671 schema_name: function.returns_table.0.clone(),
2672 table_name: function.returns_table.1.clone(),
2673 type_name: function.returns.clone(),
2674 is_by_pk: false,
2675 pk_columns: Vec::new(),
2676 max_rows,
2677 relationships: Arc::clone(&relationships),
2678 names: Arc::clone(&names),
2679 call: Some(Arc::new(FunctionCall {
2680 schema: function.schema_name.clone(),
2681 name: function.function_name.clone(),
2682 arguments: function.arguments.clone(),
2683 session_argument: function.session_argument.clone(),
2684 })),
2685 });
2686
2687 let mut field = Field::new(
2688 &function.name,
2689 TypeRef::named_nn_list_nn(&function.returns),
2690 move |ctx| {
2691 let spec = Arc::clone(&spec);
2692 FieldFuture::new(async move { resolve_query(&ctx, &spec).await })
2693 },
2694 );
2695 if !function.arguments.is_empty() {
2696 field = field.argument(InputValue::new(
2697 "args",
2698 TypeRef::named_nn(format!("{}_args", function.name)),
2699 ));
2700 }
2701 field = field
2702 .argument(InputValue::new(
2703 "where",
2704 TypeRef::named(crate::input::bool_exp::bool_exp_type_name(
2705 &function.returns,
2706 )),
2707 ))
2708 .argument(InputValue::new(
2709 "order_by",
2710 TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(
2711 &function.returns,
2712 )),
2713 ))
2714 .argument(InputValue::new(
2715 "distinct_on",
2716 TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
2717 &function.returns,
2718 )),
2719 ))
2720 .argument(InputValue::new("limit", TypeRef::named("Int")))
2721 .argument(InputValue::new("offset", TypeRef::named("Int")));
2722 if let Some(description) = &function.description {
2723 field = field.description(description);
2724 }
2725 root = root.field(field);
2726
2727 let agg_spec = Arc::new(AggregateSpec {
2732 schema_name: function.returns_table.0.clone(),
2733 table_name: function.returns_table.1.clone(),
2734 type_name: function.returns.clone(),
2735 max_rows,
2736 relationships: Arc::clone(&relationships),
2737 names: Arc::clone(&names),
2738 call: Some(Arc::new(FunctionCall {
2739 schema: function.schema_name.clone(),
2740 name: function.function_name.clone(),
2741 arguments: function.arguments.clone(),
2742 session_argument: function.session_argument.clone(),
2743 })),
2744 });
2745 let mut aggregate = Field::new(
2746 format!("{}_aggregate", function.name),
2747 TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
2748 &function.returns,
2749 )),
2750 move |ctx| {
2751 let agg_spec = Arc::clone(&agg_spec);
2752 FieldFuture::new(async move { resolve_aggregate(&ctx, &agg_spec).await })
2753 },
2754 );
2755 if !function.arguments.is_empty() {
2756 aggregate = aggregate.argument(InputValue::new(
2757 "args",
2758 TypeRef::named_nn(format!("{}_args", function.name)),
2759 ));
2760 }
2761 aggregate = with_row_arguments(aggregate, &function.returns).description(format!(
2762 "fetch aggregated fields from the table: \"{}\"",
2763 function.returns_table.1
2764 ));
2765 root = root.field(aggregate);
2766 }
2767 root
2768}
2769
2770fn function_source(
2778 ctx: &ResolverContext<'_>,
2779 call: &FunctionCall,
2780 alias: &str,
2781 bound_values: &mut Vec<serde_json::Value>,
2782) -> Result<String, async_graphql::Error> {
2783 let given = ctx
2784 .args
2785 .try_get("args")
2786 .ok()
2787 .map(|v| accessor_to_json(&v))
2788 .unwrap_or_else(|| serde_json::json!({}));
2789 let mut passed = Vec::new();
2790 for (name, _, required) in &call.arguments {
2791 match given.get(name) {
2792 None if *required => {
2793 return Err(async_graphql::Error::new(format!(
2794 "{} needs the argument \"{}\"",
2795 call.name, name
2796 )))
2797 }
2798 None => continue,
2801 Some(value) => {
2802 let (_, pg_type, _) = call
2809 .arguments
2810 .iter()
2811 .find(|(argument, _, _)| argument == name)
2812 .expect("iterating the arguments");
2813 passed.push(format!(
2814 "{} => ${}::{}",
2815 postrust_sql::escape_ident(name),
2816 bound_values.len() + 1,
2817 pg_type
2818 ));
2819 bound_values.push(value.clone());
2820 }
2821 }
2822 }
2823 if let Some(session) = &call.session_argument {
2827 passed.push(format!(
2828 "{} => {}",
2829 postrust_sql::escape_ident(session),
2830 SESSION_ARGUMENT
2831 ));
2832 }
2833 Ok(format!(
2834 "{}.{}({}) AS {}",
2835 postrust_sql::escape_ident(&call.schema),
2836 postrust_sql::escape_ident(&call.name),
2837 passed.join(", "),
2838 postrust_sql::escape_ident(alias)
2839 ))
2840}
2841
2842struct FunctionCall {
2844 schema: String,
2845 name: String,
2846 arguments: Vec<(String, String, bool)>,
2847 session_argument: Option<String>,
2850}
2851
2852struct AggregateSpec {
2854 schema_name: String,
2855 table_name: String,
2856 type_name: String,
2857 max_rows: Option<i64>,
2858 relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
2859 names: Arc<crate::names::NameOverrides>,
2860 call: Option<Arc<FunctionCall>>,
2864}
2865
2866async fn resolve_aggregate<'a>(
2873 ctx: &ResolverContext<'a>,
2874 spec: &AggregateSpec,
2875) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
2876 Ok(Some(FieldValue::value(aggregate_value(ctx, spec).await?)))
2877}
2878
2879async fn aggregate_value(
2884 ctx: &ResolverContext<'_>,
2885 spec: &AggregateSpec,
2886) -> Result<Value, async_graphql::Error> {
2887 use crate::schema::aggregate as agg;
2888
2889 let pool = ctx.data::<PgPool>()?;
2890 let gql_ctx = ctx.data::<GraphQLContext>()?;
2891
2892 let mut bound_values: Vec<serde_json::Value> = Vec::new();
2893 let source = match &spec.call {
2897 None => format!(
2898 "{}.{}",
2899 postrust_sql::escape_ident(&spec.schema_name),
2900 postrust_sql::escape_ident(&spec.table_name)
2901 ),
2902 Some(call) => function_source(ctx, call, &spec.table_name, &mut bound_values)?,
2903 };
2904
2905 let mut where_sql = String::new();
2906 {
2907 let requested = ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v));
2911 let permission = permission_predicate(
2912 &gql_ctx.caller(),
2913 spec.names.as_ref(),
2914 &spec.schema_name,
2915 &spec.table_name,
2916 )?;
2917 if let Some(predicate) = and_predicates(requested, permission) {
2918 let guard = gql_ctx
2919 .schema_cache
2920 .get()
2921 .await
2922 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
2923 let cache = guard
2924 .as_ref()
2925 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
2926 let scope = match &spec.call {
2933 None => WhereScope::table(
2934 &spec.schema_name,
2935 &spec.table_name,
2936 &spec.type_name,
2937 spec.names.as_ref(),
2938 )
2939 .with_resolution(cache, spec.relationships.as_ref()),
2940 Some(_) => WhereScope::for_alias(
2941 &spec.schema_name,
2942 &spec.table_name,
2943 &spec.table_name,
2944 &spec.type_name,
2945 cache,
2946 spec.relationships.as_ref(),
2947 spec.names.as_ref(),
2948 ),
2949 }
2950 .for_caller(gql_ctx.caller());
2951 let (sql, values) =
2952 build_where_clause(Some(&predicate), bound_values.len() + 1, &scope)?;
2953 if !sql.is_empty() {
2954 where_sql = format!(" {}", sql);
2955 bound_values.extend(values);
2956 }
2957 }
2958 }
2959
2960 let order_sql = build_order_by_clause(
2961 ctx,
2962 &gql_ctx.schema_cache,
2963 &spec.schema_name,
2964 &spec.table_name,
2965 &spec.type_name,
2966 spec.relationships.as_ref(),
2967 &format!(
2968 "{}.{}",
2969 postrust_sql::escape_ident(&spec.schema_name),
2970 postrust_sql::escape_ident(&spec.table_name)
2971 ),
2972 spec.names.as_ref(),
2973 )
2974 .await?;
2975
2976 let requested_limit = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok());
2977 let offset = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok());
2978 let ceiling = match (
2979 spec.max_rows,
2980 permission_limit(
2981 &gql_ctx.caller(),
2982 spec.names.as_ref(),
2983 &spec.schema_name,
2984 &spec.table_name,
2985 ),
2986 ) {
2987 (Some(configured), Some(granted)) => Some(configured.min(granted)),
2988 (only, None) | (None, only) => only,
2989 };
2990 let limit = match (requested_limit, ceiling) {
2991 (Some(requested), Some(ceiling)) => Some(requested.min(ceiling)),
2992 (Some(requested), None) => Some(requested),
2993 (None, ceiling) => ceiling,
2994 };
2995
2996 let rows = format!("SELECT * FROM {}{}{}", source, where_sql, order_sql);
2997
2998 let mut inner = rows.clone();
3000 if let Some(limit) = limit {
3001 inner.push_str(&format!(" LIMIT {}", limit));
3002 }
3003 if let Some(offset) = offset {
3004 inner.push_str(&format!(" OFFSET {}", offset));
3005 }
3006
3007 let mut counted = rows;
3021 if let Some(limit) = requested_limit {
3022 counted.push_str(&format!(" LIMIT {}", limit));
3023 }
3024 if let Some(offset) = offset {
3025 counted.push_str(&format!(" OFFSET {}", offset));
3026 }
3027
3028 let mut wants_nodes = false;
3033 let mut wanted: BTreeMap<String, Vec<String>> = BTreeMap::new();
3034 let mut counts: Vec<(String, String)> = Vec::new();
3037 for selection in ctx.field().selection_set() {
3038 match selection.name() {
3039 "nodes" => wants_nodes = true,
3040 "aggregate" => {
3041 for function in selection.selection_set() {
3042 if function.name() == "count" {
3043 counts.push((
3044 function.alias().unwrap_or("count").to_string(),
3045 count_expression(
3046 function,
3047 (&spec.schema_name, &spec.table_name),
3048 None,
3049 spec.names.as_ref(),
3050 ),
3051 ));
3052 continue;
3053 }
3054 let columns = wanted.entry(function.name().to_string()).or_default();
3055 for column in function.selection_set() {
3056 let name = column.name().to_string();
3057 if !columns.contains(&name) {
3058 columns.push(name);
3059 }
3060 }
3061 }
3062 }
3063 _ => {}
3064 }
3065 }
3066 wanted.retain(|_, columns| !columns.is_empty());
3067
3068 let mut result = async_graphql::indexmap::IndexMap::new();
3069
3070 if !counts.is_empty() || !wanted.is_empty() {
3071 let mut parts = match counts.iter().any(|(key, _)| key == "count") {
3075 true => Vec::new(),
3076 false => vec!["'count', count(*)".to_string()],
3077 };
3078 for (key, expression) in &counts {
3079 parts.push(format!("'{}', {}", key.replace('\'', "''"), expression));
3080 }
3081 for (function, columns) in &wanted {
3082 let sql_function = agg::NUMERIC_AGGREGATES
3086 .iter()
3087 .chain(agg::ORDERED_AGGREGATES.iter())
3088 .find(|(name, _)| name == function)
3089 .map(|(name, _)| *name)
3090 .ok_or_else(|| {
3091 async_graphql::Error::new(format!("unknown aggregate \"{}\"", function))
3092 })?;
3093 let per_column: Vec<String> = columns
3094 .iter()
3095 .map(|field| {
3096 let source = spec
3097 .names
3098 .column_source(&spec.schema_name, &spec.table_name, field)
3099 .unwrap_or(field);
3100 format!(
3101 "'{}', {}({})",
3102 field.replace('\'', "''"),
3103 sql_function,
3104 postrust_sql::escape_ident(source)
3105 )
3106 })
3107 .collect();
3108 parts.push(format!(
3109 "'{}', json_build_object({})",
3110 sql_function,
3111 per_column.join(", ")
3112 ));
3113 }
3114
3115 let sql = format!(
3116 "SELECT json_build_object({}) FROM ({}) AS pgrst_agg",
3117 parts.join(", "),
3118 counted
3119 );
3120 let mut conn =
3121 begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?;
3122 let rows = execute_query_on(&mut conn, &sql, &bound_values).await?;
3123 conn.commit().await?;
3124 if let Some(first) = rows.into_iter().next() {
3125 if let Value::Object(map) = json_to_value(first) {
3126 result.insert(async_graphql::Name::new("aggregate"), Value::Object(map));
3127 }
3128 }
3129 }
3130
3131 if wants_nodes {
3132 let nodes = ctx
3138 .field()
3139 .selection_set()
3140 .find(|selection| selection.name() == "nodes")
3141 .ok_or_else(|| async_graphql::Error::new("nodes was asked for and then was not"))?;
3142
3143 let mut node_values = bound_values.clone();
3147 let (projection, row_column_types) = {
3148 let guard = gql_ctx
3149 .schema_cache
3150 .get()
3151 .await
3152 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3153 let cache = guard
3154 .as_ref()
3155 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3156 let qi = postrust_core::api_request::QualifiedIdentifier::new(
3157 &spec.schema_name,
3158 &spec.table_name,
3159 );
3160 let table = cache.get_table(&qi);
3161 let mut param_idx = node_values.len() + 1;
3162 let computed = match table {
3163 Some(table) => computed_projections(
3164 table,
3165 nodes,
3166 "src",
3167 spec.names.as_ref(),
3168 cache,
3169 &mut param_idx,
3170 &mut node_values,
3171 )?,
3172 None => Vec::new(),
3173 };
3174 let embeds = build_embed_expressions(
3175 &gql_ctx.caller(),
3176 cache,
3177 spec.relationships.as_ref(),
3178 &spec.type_name,
3179 "src",
3180 nodes,
3181 spec.max_rows,
3182 &mut 0,
3183 &mut param_idx,
3184 &mut node_values,
3185 spec.names.as_ref(),
3186 )?;
3187 let mut projection = table
3191 .and_then(|table| rename_projection(table, "src", spec.names.as_ref()))
3192 .unwrap_or_else(|| "src.*".to_string());
3193 for expression in &computed {
3194 projection.push_str(", ");
3195 projection.push_str(expression);
3196 }
3197 for (field_name, expression) in &embeds {
3198 projection.push_str(", ");
3199 projection.push_str(expression);
3200 projection.push_str(" AS ");
3201 projection.push_str(&postrust_sql::escape_ident(field_name));
3202 }
3203 (
3204 projection,
3205 table
3206 .map(|table| exposed_column_types(table, spec.names.as_ref()))
3207 .unwrap_or_default(),
3208 )
3209 };
3210
3211 let sql = format!(
3212 "SELECT {} FROM (SELECT {} FROM ({}) AS src) AS pgrst_nodes",
3213 row_json("pgrst_nodes", &row_column_types),
3214 projection,
3215 inner
3216 );
3217 let mut conn =
3218 begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?;
3219 let rows = execute_query_on(&mut conn, &sql, &node_values).await?;
3220 conn.commit().await?;
3221 result.insert(
3222 async_graphql::Name::new("nodes"),
3223 Value::List(rows.into_iter().map(json_to_value).collect()),
3224 );
3225 }
3226
3227 let _ = &spec.type_name;
3228 Ok(Value::Object(result))
3229}
3230
3231async fn resolve_query<'a>(
3233 ctx: &ResolverContext<'a>,
3234 spec: &QueryFieldSpec,
3235) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
3236 Ok(rows_as_field_value(
3237 query_rows(ctx, spec).await?,
3238 spec.is_by_pk,
3239 ))
3240}
3241
3242async fn query_rows(
3248 ctx: &ResolverContext<'_>,
3249 spec: &QueryFieldSpec,
3250) -> Result<Vec<serde_json::Value>, async_graphql::Error> {
3251 let schema_name = spec.schema_name.as_str();
3252 let table_name = spec.table_name.as_str();
3253 let type_name = spec.type_name.as_str();
3254 let is_by_pk = spec.is_by_pk;
3255 let pk_columns = spec.pk_columns.as_slice();
3256 let max_rows = spec.max_rows;
3257 let relationships = spec.relationships.as_ref();
3258
3259 let pool = ctx.data::<PgPool>()?;
3260 let gql_ctx = ctx.data::<GraphQLContext>()?;
3261
3262 debug!("Resolving query for table: {}", table_name);
3263
3264 let requested_limit: Option<i64> = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok());
3266
3267 let offset: Option<i64> = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok());
3268
3269 let limit: Option<i64> = if is_by_pk {
3274 Some(1)
3275 } else {
3276 let ceiling = match (
3280 max_rows,
3281 permission_limit(
3282 &gql_ctx.caller(),
3283 spec.names.as_ref(),
3284 schema_name,
3285 table_name,
3286 ),
3287 ) {
3288 (Some(configured), Some(granted)) => Some(configured.min(granted)),
3289 (only, None) | (None, only) => only,
3290 };
3291 match (requested_limit, ceiling) {
3292 (Some(requested), Some(ceiling)) => Some(requested.min(ceiling)),
3293 (Some(requested), None) => Some(requested),
3294 (None, ceiling) => ceiling,
3295 }
3296 };
3297
3298 let mut where_sql = String::new();
3306 let mut bound_values: Vec<serde_json::Value> = Vec::new();
3307
3308 let source = match &spec.call {
3312 None => format!(
3318 "{}.{} AS {}",
3319 postrust_sql::escape_ident(schema_name),
3320 postrust_sql::escape_ident(table_name),
3321 postrust_sql::escape_ident(READ_ROW)
3322 ),
3323 Some(call) => function_source(ctx, call, table_name, &mut bound_values)?,
3324 };
3325
3326 let source_ref = match &spec.call {
3329 None => postrust_sql::escape_ident(READ_ROW),
3330 Some(_) => postrust_sql::escape_ident(table_name),
3331 };
3332
3333 if is_by_pk {
3334 if pk_columns.is_empty() {
3335 return Err(async_graphql::Error::new(format!(
3336 "\"{}\" has no primary key, so it cannot be queried by key",
3337 table_name
3338 )));
3339 }
3340
3341 let mut conditions = Vec::with_capacity(pk_columns.len());
3342 for (idx, (col_name, pg_type)) in pk_columns.iter().enumerate() {
3343 let field = spec
3344 .names
3345 .column(schema_name, table_name, col_name)
3346 .unwrap_or(col_name);
3347 let value = ctx.args.try_get(field).map_err(|_| {
3348 async_graphql::Error::new(format!(
3349 "missing required primary key argument \"{}\"",
3350 field
3351 ))
3352 })?;
3353
3354 conditions.push(format!(
3355 "{} = ${}::{}",
3356 postrust_sql::escape_ident(col_name),
3357 idx + 1,
3358 pg_type
3359 ));
3360 bound_values.push(accessor_to_json(&value));
3361 }
3362 where_sql = format!(" WHERE {}", conditions.join(" AND "));
3363 }
3364
3365 {
3370 let requested = match is_by_pk {
3371 true => None,
3372 false => ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v)),
3373 };
3374 let permission = permission_predicate(
3375 &gql_ctx.caller(),
3376 spec.names.as_ref(),
3377 schema_name,
3378 table_name,
3379 )?;
3380
3381 if let Some(predicate) = and_predicates(requested, permission) {
3382 let guard = gql_ctx
3383 .schema_cache
3384 .get()
3385 .await
3386 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3387 let cache = guard
3388 .as_ref()
3389 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3390 let scope = match &spec.call {
3391 None => WhereScope::table(schema_name, table_name, type_name, spec.names.as_ref())
3392 .under_alias(READ_ROW)
3393 .with_resolution(cache, relationships)
3394 .for_caller(gql_ctx.caller()),
3395 Some(_) => WhereScope::for_alias(
3398 schema_name,
3399 table_name,
3400 table_name,
3401 type_name,
3402 cache,
3403 relationships,
3404 spec.names.as_ref(),
3405 )
3406 .for_caller(gql_ctx.caller()),
3407 };
3408 let (filter_sql, filter_values) =
3409 build_where_clause(Some(&predicate), bound_values.len() + 1, &scope)?;
3410 if !filter_sql.is_empty() {
3411 where_sql = match where_sql.is_empty() {
3414 true => format!(" {}", filter_sql),
3415 false => format!(
3416 "{} AND ({})",
3417 where_sql,
3418 filter_sql.trim_start_matches("WHERE ")
3419 ),
3420 };
3421 bound_values.extend(filter_values);
3422 }
3423 }
3424 }
3425
3426 let (order_sql, distinct_on) = if is_by_pk {
3429 (String::new(), Vec::new())
3430 } else {
3431 (
3432 build_order_by_clause(
3433 ctx,
3434 &gql_ctx.schema_cache,
3435 schema_name,
3436 table_name,
3437 type_name,
3438 relationships,
3439 &source_ref,
3440 spec.names.as_ref(),
3441 )
3442 .await?,
3443 build_distinct_on(
3444 ctx,
3445 &gql_ctx.schema_cache,
3446 schema_name,
3447 table_name,
3448 spec.names.as_ref(),
3449 )
3450 .await?,
3451 )
3452 };
3453
3454 let (distinct_sql, order_sql) = {
3462 let written = order_sql.strip_prefix(" ORDER BY ");
3463 let (prefix, order) = distinct_on_clause(&distinct_on, written).map_err(|error| {
3466 at_path(
3467 error,
3468 &format!(
3469 "$.selectionSet.{}.args",
3470 ctx.ctx.field().alias().unwrap_or(ctx.ctx.field().name())
3471 ),
3472 )
3473 })?;
3474 (
3475 prefix,
3476 match order {
3477 Some(order) => format!(" ORDER BY {}", order),
3478 None => String::new(),
3479 },
3480 )
3481 };
3482
3483 let (computed, row_column_types) = {
3486 let guard = gql_ctx
3487 .schema_cache
3488 .get()
3489 .await
3490 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3491 match guard.as_ref() {
3492 Some(cache) => {
3493 let qi =
3494 postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
3495 match cache.get_table(&qi) {
3496 Some(table) => {
3497 let mut param_idx = bound_values.len() + 1;
3498 (
3499 computed_projections(
3500 table,
3501 ctx.field(),
3502 "src",
3503 spec.names.as_ref(),
3504 cache,
3505 &mut param_idx,
3506 &mut bound_values,
3507 )?,
3508 exposed_column_types(table, spec.names.as_ref()),
3509 )
3510 }
3511 None => (Vec::new(), HashMap::new()),
3512 }
3513 }
3514 None => (Vec::new(), HashMap::new()),
3515 }
3516 };
3517 let mut inner = format!(
3521 "SELECT {}* FROM {}{}{}",
3522 distinct_sql, source, where_sql, order_sql
3523 );
3524
3525 if let Some(limit) = limit {
3526 inner.push_str(&format!(" LIMIT {}", limit));
3527 }
3528
3529 if let Some(offset) = offset {
3530 inner.push_str(&format!(" OFFSET {}", offset));
3531 }
3532
3533 let embed_expressions = {
3536 let guard = gql_ctx
3537 .schema_cache
3538 .get()
3539 .await
3540 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3541 match guard.as_ref() {
3542 Some(cache) => {
3543 let mut param_idx = bound_values.len() + 1;
3548 build_embed_expressions(
3549 &gql_ctx.caller(),
3550 cache,
3551 relationships,
3552 type_name,
3553 "src",
3554 ctx.field(),
3555 max_rows,
3556 &mut 0,
3557 &mut param_idx,
3558 &mut bound_values,
3559 spec.names.as_ref(),
3560 )?
3561 }
3562 None => Vec::new(),
3563 }
3564 };
3565
3566 let renamed = {
3574 let guard = gql_ctx
3575 .schema_cache
3576 .get()
3577 .await
3578 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3579 guard
3580 .as_ref()
3581 .and_then(|cache| {
3582 cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
3583 schema_name,
3584 table_name,
3585 ))
3586 })
3587 .and_then(|table| rename_projection(table, "src", spec.names.as_ref()))
3588 };
3589 let inner = if embed_expressions.is_empty() && computed.is_empty() && renamed.is_none() {
3590 inner
3591 } else {
3592 let mut projection = renamed.unwrap_or_else(|| "src.*".to_string());
3593 for expression in &computed {
3594 projection.push_str(", ");
3595 projection.push_str(expression);
3596 }
3597 for (field_name, expression) in &embed_expressions {
3598 projection.push_str(", ");
3599 projection.push_str(expression);
3600 projection.push_str(" AS ");
3601 projection.push_str(&postrust_sql::escape_ident(field_name));
3602 }
3603 format!("SELECT {} FROM ({}) AS src", projection, inner)
3604 };
3605
3606 let sql = format!(
3607 "SELECT {} FROM ({}) t",
3608 row_json("t", &row_column_types),
3609 inner
3610 );
3611
3612 let mut tx = begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?;
3616
3617 let mut result = execute_query_on(&mut tx, &sql, &bound_values).await?;
3619
3620 if embed_expressions.is_empty() {
3622 let guard = gql_ctx
3623 .schema_cache
3624 .get()
3625 .await
3626 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3627 let cache = guard
3628 .as_ref()
3629 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3630
3631 let embed_ctx = EmbedContext {
3632 schema_cache: cache,
3633 relationships,
3634 max_rows,
3635 };
3636
3637 embed_relationships(&mut tx, &embed_ctx, type_name, ctx.field(), &mut result).await?;
3638 }
3639
3640 tx.commit().await?;
3641
3642 Ok(result)
3643}
3644
3645fn rows_as_field_value<'a>(rows: Vec<serde_json::Value>, is_by_pk: bool) -> Option<FieldValue<'a>> {
3650 match is_by_pk {
3651 true => rows
3652 .into_iter()
3653 .next()
3654 .map(|v| FieldValue::value(json_to_value(v))),
3655 false => Some(FieldValue::list(
3656 rows.into_iter()
3657 .map(|v| FieldValue::value(json_to_value(v))),
3658 )),
3659 }
3660}
3661
3662#[allow(clippy::too_many_arguments)]
3664async fn resolve_mutation<'a>(
3665 ctx: &ResolverContext<'a>,
3666 schema_name: &str,
3667 table_name: &str,
3668 mutation_type: MutationType,
3669 pk_columns: &[(String, String)],
3670 relationships: &Arc<HashMap<String, Vec<RelationshipField>>>,
3671 type_names: &Arc<HashMap<(String, String), String>>,
3672 names: &Arc<crate::names::NameOverrides>,
3673 max_rows: Option<i64>,
3674) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
3675 let pool = ctx.data::<PgPool>()?;
3676 let gql_ctx = ctx.data::<GraphQLContext>()?;
3677
3678 debug!(
3679 "Resolving mutation for table: {} type: {:?}",
3680 table_name, mutation_type
3681 );
3682
3683 let type_name = type_names
3689 .get(&(schema_name.to_string(), table_name.to_string()))
3690 .cloned()
3691 .unwrap_or_else(|| table_name.to_string());
3692
3693 let by_key = matches!(
3694 mutation_type,
3695 MutationType::InsertOne | MutationType::UpdateByPk | MutationType::DeleteByPk
3696 );
3697 let returning = if by_key {
3698 Some(ctx.field())
3699 } else {
3700 ctx.field()
3701 .selection_set()
3702 .find(|selection| selection.name() == "returning")
3703 };
3704
3705 if mutation_type == MutationType::UpdateMany {
3710 let updates = ctx
3711 .args
3712 .try_get("updates")
3713 .ok()
3714 .map(|v| accessor_to_json(&v))
3715 .unwrap_or(serde_json::Value::Null);
3716 let serde_json::Value::Array(entries) = updates else {
3717 return Err(async_graphql::Error::new(
3718 "\"updates\" takes a list of {where, _set, …} objects",
3719 ));
3720 };
3721 let column_types = column_types_of(&gql_ctx.schema_cache, schema_name, table_name).await;
3722 let mut answers: Vec<FieldValue<'_>> = Vec::with_capacity(entries.len());
3723 for entry in entries {
3724 let serde_json::Value::Object(entry) = entry else {
3725 return Err(async_graphql::Error::new(
3726 "each update is an object of {where, _set, …}",
3727 ));
3728 };
3729 let operators: Vec<(&'static str, serde_json::Value)> = UPDATE_OPERATORS
3730 .iter()
3731 .filter_map(|name| entry.get(*name).map(|value| (*name, value.clone())))
3732 .filter(|(_, value)| !value.is_null())
3733 .collect();
3734 let rows = execute_update(
3735 pool,
3736 gql_ctx,
3737 schema_name,
3738 table_name,
3739 &type_name,
3740 operators,
3741 column_types.clone(),
3742 entry.get("where").cloned(),
3743 relationships.as_ref(),
3744 names.as_ref(),
3745 )
3746 .await?;
3747 let affected = rows.len();
3748 let rows = match returning {
3749 Some(returning) if !rows.is_empty() => {
3750 reread_returning(
3751 pool,
3752 gql_ctx,
3753 schema_name,
3754 table_name,
3755 &type_name,
3756 rows,
3757 returning,
3758 relationships.as_ref(),
3759 names.as_ref(),
3760 max_rows,
3761 )
3762 .await?
3763 }
3764 _ => rows,
3765 };
3766 answers.extend(mutation_result(rows, affected, false));
3767 }
3768 return Ok(Some(FieldValue::list(answers)));
3769 }
3770
3771 let (result, affected) = match mutation_type {
3772 MutationType::Insert | MutationType::InsertOne => {
3773 let objects = ctx
3774 .args
3775 .try_get("objects")
3776 .ok()
3777 .map(|v| accessor_to_json(&v))
3778 .or_else(|| {
3779 ctx.args
3782 .try_get("object")
3783 .ok()
3784 .map(|v| serde_json::Value::Array(vec![accessor_to_json(&v)]))
3785 })
3786 .unwrap_or_else(|| serde_json::Value::Array(vec![]));
3787
3788 let guard = gql_ctx
3789 .schema_cache
3790 .get()
3791 .await
3792 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3793 let cache = guard
3794 .as_ref()
3795 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3796 let here = format!(
3800 "$.selectionSet.{}",
3801 ctx.ctx.field().alias().unwrap_or(ctx.ctx.field().name())
3802 );
3803 let named = match mutation_type {
3806 MutationType::InsertOne => "object",
3807 _ => "objects",
3808 };
3809 let context = InsertContext {
3810 on_conflict: ctx
3811 .args
3812 .try_get("on_conflict")
3813 .ok()
3814 .map(|v| accessor_to_json(&v))
3815 .filter(|v| !v.is_null()),
3816 cache,
3817 relationships: relationships.as_ref(),
3818 type_names: type_names.as_ref(),
3819 names: names.as_ref(),
3820 caller: gql_ctx.caller(),
3821 row: String::new(),
3824 objects: format!("{}.args.{}", here, named),
3825 conflict: format!("{}.args.on_conflict", here),
3826 };
3827
3828 execute_insert(
3829 pool,
3830 gql_ctx,
3831 schema_name,
3832 table_name,
3833 objects,
3834 &context,
3835 mutation_type == MutationType::InsertOne,
3836 )
3837 .await?
3838 }
3839 MutationType::UpdateMany | MutationType::Update | MutationType::UpdateByPk => {
3842 let operators: Vec<(&'static str, serde_json::Value)> = UPDATE_OPERATORS
3846 .iter()
3847 .filter_map(|name| {
3848 ctx.args
3849 .try_get(name)
3850 .ok()
3851 .map(|v| (*name, accessor_to_json(&v)))
3852 })
3853 .filter(|(_, value)| !value.is_null())
3854 .collect();
3855
3856 let where_clause = if mutation_type == MutationType::UpdateByPk {
3857 Some(pk_where_from_args(
3858 ctx,
3859 schema_name,
3860 table_name,
3861 pk_columns,
3862 names.as_ref(),
3863 )?)
3864 } else {
3865 ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v))
3866 };
3867 let where_clause = and_predicates(
3871 where_clause,
3872 permission_filter(
3873 &gql_ctx.caller(),
3874 names.as_ref(),
3875 schema_name,
3876 table_name,
3877 crate::role::Verb::Update,
3878 )?,
3879 );
3880
3881 execute_update(
3882 pool,
3883 gql_ctx,
3884 schema_name,
3885 table_name,
3886 &type_name,
3887 operators,
3888 column_types_of(&gql_ctx.schema_cache, schema_name, table_name).await,
3889 where_clause,
3890 relationships.as_ref(),
3891 names.as_ref(),
3892 )
3893 .await
3894 .map(|rows| {
3895 let count = rows.len();
3896 (rows, count)
3897 })?
3898 }
3899 MutationType::Delete | MutationType::DeleteByPk => {
3900 let where_clause = if mutation_type == MutationType::DeleteByPk {
3901 Some(pk_where_from_args(
3902 ctx,
3903 schema_name,
3904 table_name,
3905 pk_columns,
3906 names.as_ref(),
3907 )?)
3908 } else {
3909 ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v))
3910 };
3911 let where_clause = and_predicates(
3912 where_clause,
3913 permission_filter(
3914 &gql_ctx.caller(),
3915 names.as_ref(),
3916 schema_name,
3917 table_name,
3918 crate::role::Verb::Delete,
3919 )?,
3920 );
3921
3922 execute_delete(
3923 pool,
3924 gql_ctx,
3925 schema_name,
3926 table_name,
3927 &type_name,
3928 where_clause,
3929 returning,
3930 relationships.as_ref(),
3931 names.as_ref(),
3932 max_rows,
3933 )
3934 .await
3935 .map(|rows| {
3936 let count = rows.len();
3937 (rows, count)
3938 })?
3939 }
3940 };
3941
3942 let is_delete = matches!(
3943 mutation_type,
3944 MutationType::Delete | MutationType::DeleteByPk
3945 );
3946 let result = match returning {
3947 Some(returning) if !result.is_empty() && !is_delete => {
3948 reread_returning(
3949 pool,
3950 gql_ctx,
3951 schema_name,
3952 table_name,
3953 &type_name,
3954 result,
3955 returning,
3956 relationships.as_ref(),
3957 names.as_ref(),
3958 max_rows,
3959 )
3960 .await?
3961 }
3962 _ => result,
3963 };
3964
3965 Ok(mutation_result(result, affected, by_key))
3966}
3967
3968async fn write_tx<'a>(
3976 gql_ctx: &'a GraphQLContext,
3977 pool: &PgPool,
3978) -> Result<
3979 tokio::sync::MutexGuard<'a, Option<sqlx::Transaction<'static, sqlx::Postgres>>>,
3980 async_graphql::Error,
3981> {
3982 let mut guard = gql_ctx.write.lock().await;
3983 if guard.is_none() {
3984 *guard = Some(begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?);
3985 }
3986 Ok(guard)
3987}
3988
3989async fn begin_with_session(
3995 pool: &PgPool,
3996 role: &str,
3997 settings: &[(String, String)],
3998) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, async_graphql::Error> {
3999 let mut tx = pool.begin().await?;
4000
4001 if !settings.is_empty() {
4011 let calls = (0..settings.len())
4012 .map(|i| format!("set_config(${}, ${}, true)", i * 2 + 1, i * 2 + 2))
4013 .collect::<Vec<_>>()
4014 .join(", ");
4015 let sql = format!("SELECT {calls}");
4016 let mut query = sqlx::query_scalar::<_, String>(&sql);
4017 for (name, value) in settings {
4018 query = query.bind(name).bind(value);
4019 }
4020 query.fetch_optional(&mut *tx).await?;
4021 }
4022
4023 sqlx::query(&format!(
4024 "SET LOCAL ROLE {}",
4025 postrust_sql::escape_ident(role)
4026 ))
4027 .execute(&mut *tx)
4028 .await?;
4029 Ok(tx)
4030}
4031
4032fn row_json(expr: &str, column_types: &HashMap<String, String>) -> String {
4043 let mut shapes: Vec<(&String, &String)> = column_types
4044 .iter()
4045 .filter(|(_, pg_type)| matches!(pg_type.as_str(), "geometry" | "geography"))
4046 .collect();
4047 if shapes.is_empty() {
4048 return format!("row_to_json({})", expr);
4049 }
4050 shapes.sort();
4051
4052 let overrides: Vec<String> = shapes
4053 .iter()
4054 .map(|(name, _)| {
4055 format!(
4056 "'{}', ST_AsGeoJSON({}.{}, 9, 4)::jsonb",
4062 name.replace('\'', "''"),
4063 expr,
4064 postrust_sql::escape_ident(name)
4065 )
4066 })
4067 .collect();
4068 format!(
4069 "(to_jsonb({}) || jsonb_build_object({}))",
4070 expr,
4071 overrides.join(", ")
4072 )
4073}
4074
4075async fn execute_query_on(
4078 conn: &mut sqlx::PgConnection,
4079 sql: &str,
4080 params: &[serde_json::Value],
4081) -> Result<Vec<serde_json::Value>, async_graphql::Error> {
4082 use sqlx::Row;
4083
4084 trace!("Executing SQL: {}", sql);
4085
4086 let mut query = sqlx::query(sql);
4088 for param in params {
4089 query = bind_json_value(query, param);
4090 }
4091 let rows = query.fetch_all(&mut *conn).await.map_err(database_error)?;
4092
4093 let results: Vec<serde_json::Value> = rows
4096 .into_iter()
4097 .filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
4098 .collect();
4099
4100 Ok(results)
4101}
4102
4103fn check_geojson(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
4116 fn refuse<T>(message: &str) -> Result<T, async_graphql::Error> {
4117 Err(coded_error("parse-failed", message))
4118 }
4119
4120 fn position(
4121 value: &serde_json::Value,
4122 ) -> Result<&Vec<serde_json::Value>, async_graphql::Error> {
4123 match value.as_array() {
4124 Some(items) if items.len() >= 2 && items.iter().all(|i| i.is_number()) => Ok(items),
4125 _ => refuse("A Position needs at least 2 elements"),
4126 }
4127 }
4128
4129 fn positions(
4130 value: Option<&serde_json::Value>,
4131 least: usize,
4132 what: &str,
4133 ) -> Result<Vec<Vec<serde_json::Value>>, async_graphql::Error> {
4134 let items = match value.and_then(|v| v.as_array()) {
4135 Some(items) => items,
4136 None => return refuse(&format!("A {} needs at least {} Positions", what, least)),
4137 };
4138 if items.len() < least {
4139 return refuse(&format!("A {} needs at least {} Positions", what, least));
4140 }
4141 items.iter().map(|item| position(item).cloned()).collect()
4142 }
4143
4144 fn ring(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
4145 let points = positions(Some(value), 4, "LinearRing")?;
4146 match (points.first(), points.last()) {
4147 (Some(first), Some(last)) if first == last => Ok(()),
4148 _ => refuse("the first and last locations have to be equal for a LinearRing"),
4149 }
4150 }
4151
4152 fn polygon(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
4153 let Some(rings) = value.as_array() else {
4154 return refuse("A LinearRing needs at least 4 Positions");
4155 };
4156 rings.iter().try_for_each(ring)
4157 }
4158
4159 let serde_json::Value::Object(map) = value else {
4160 return Ok(());
4161 };
4162 let Some(kind) = map.get("type").and_then(|v| v.as_str()) else {
4163 return Ok(());
4164 };
4165 let coordinates = map.get("coordinates");
4166 match kind {
4167 "Point" => {
4168 position(coordinates.unwrap_or(&serde_json::Value::Null))?;
4169 }
4170 "MultiPoint" => {
4171 positions(coordinates, 0, "MultiPoint")?;
4172 }
4173 "LineString" => {
4174 positions(coordinates, 2, "LineString")?;
4175 }
4176 "MultiLineString" => {
4177 for line in coordinates.and_then(|v| v.as_array()).into_iter().flatten() {
4178 positions(Some(line), 2, "LineString")?;
4179 }
4180 }
4181 "Polygon" => polygon(coordinates.unwrap_or(&serde_json::Value::Null))?,
4182 "MultiPolygon" => {
4183 for one in coordinates.and_then(|v| v.as_array()).into_iter().flatten() {
4184 polygon(one)?;
4185 }
4186 }
4187 "GeometryCollection" => {
4188 for geometry in map
4189 .get("geometries")
4190 .and_then(|v| v.as_array())
4191 .into_iter()
4192 .flatten()
4193 {
4194 check_geojson(geometry)?;
4195 }
4196 }
4197 other => return refuse(&format!("unexpected geometry type: {}", other)),
4198 }
4199 Ok(())
4200}
4201
4202fn computed_call(
4210 definition: &postrust_core::schema_cache::ComputedColumn,
4211 row: &str,
4212 row_by_name: &str,
4213) -> String {
4214 let function = format!(
4215 "{}.{}",
4216 postrust_sql::escape_ident(&definition.function.schema),
4217 postrust_sql::escape_ident(&definition.function.name)
4218 );
4219 match (&definition.session_argument, &definition.row_argument) {
4220 (Some(session), Some(row_argument)) => format!(
4221 "{}({} => {}, {} => {})",
4222 function,
4223 postrust_sql::escape_ident(row_argument),
4224 row_by_name,
4225 postrust_sql::escape_ident(session),
4226 SESSION_ARGUMENT
4227 ),
4228 _ => format!("{}({})", function, row),
4229 }
4230}
4231
4232const SESSION_ARGUMENT: &str = "coalesce(current_setting('hasura.session', true), '{}')::json";
4238
4239const UPDATE_OPERATORS: [&str; 7] = [
4242 "_set",
4243 "_inc",
4244 "_append",
4245 "_prepend",
4246 "_delete_key",
4247 "_delete_elem",
4248 "_delete_at_path",
4249];
4250
4251const READ_ROW: &str = "pgrst_src";
4257
4258const WRITTEN_ROW: &str = "pgrst_row";
4263
4264fn write_cast(column_types: &HashMap<String, String>, column: &str) -> String {
4272 match column_types.get(column) {
4273 Some(pg_type) => format!("::{}", pg_type),
4274 None => String::new(),
4275 }
4276}
4277
4278fn write_operand(
4285 column_types: &HashMap<String, String>,
4286 column: &str,
4287 value: &serde_json::Value,
4288) -> serde_json::Value {
4289 match column_types.get(column).map(String::as_str) {
4290 Some("json") | Some("jsonb") => match value {
4291 serde_json::Value::String(_)
4292 | serde_json::Value::Number(_)
4293 | serde_json::Value::Bool(_) => serde_json::Value::String(value.to_string()),
4294 other => other.clone(),
4295 },
4296 Some(pg_type) if is_array_type(pg_type) => match value {
4301 serde_json::Value::Array(_) => serde_json::Value::String(array_literal(value)),
4302 other => other.clone(),
4303 },
4304 _ => value.clone(),
4305 }
4306}
4307
4308fn is_array_type(pg_type: &str) -> bool {
4313 pg_type.ends_with("[]") || pg_type.starts_with('_')
4314}
4315
4316fn array_literal(value: &serde_json::Value) -> String {
4322 fn element(value: &serde_json::Value) -> String {
4323 match value {
4324 serde_json::Value::Null => "NULL".to_string(),
4325 serde_json::Value::Array(_) => array_literal(value),
4326 serde_json::Value::String(text) => {
4327 format!("\"{}\"", text.replace('\\', "\\\\").replace('"', "\\\""))
4328 }
4329 other => format!("\"{}\"", other.to_string().replace('"', "\\\"")),
4330 }
4331 }
4332 let items = match value {
4333 serde_json::Value::Array(items) => items,
4334 other => return element(other),
4335 };
4336 format!(
4337 "{{{}}}",
4338 items.iter().map(element).collect::<Vec<_>>().join(",")
4339 )
4340}
4341
4342fn write_expression(
4353 column_types: &HashMap<String, String>,
4354 column: &str,
4355 value: &serde_json::Value,
4356 placeholder: &str,
4357) -> String {
4358 let pg_type = column_types.get(column).map(String::as_str);
4359 if value.is_object() && matches!(pg_type, Some("geometry") | Some("geography")) {
4360 return match pg_type {
4361 Some("geography") => format!("ST_GeomFromGeoJSON({})::geography", placeholder),
4362 _ => format!("ST_GeomFromGeoJSON({})", placeholder),
4363 };
4364 }
4365 if pg_type == Some("money") && value.is_number() {
4371 return format!("{}::numeric::money", placeholder);
4372 }
4373 format!("{}{}", placeholder, write_cast(column_types, column))
4374}
4375
4376async fn column_types_of(
4378 schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
4379 schema_name: &str,
4380 table_name: &str,
4381) -> HashMap<String, String> {
4382 let Ok(guard) = schema_cache.get().await else {
4383 return HashMap::new();
4384 };
4385 let Some(cache) = guard.as_ref() else {
4386 return HashMap::new();
4387 };
4388 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
4389 match cache.get_table(&qi) {
4390 Some(table) => table
4391 .columns
4392 .values()
4393 .map(|c| (c.name.clone(), c.nominal_type.clone()))
4394 .collect(),
4395 None => HashMap::new(),
4396 }
4397}
4398
4399fn insert_row<'life>(
4417 conn: &'life mut sqlx::PgConnection,
4418 schema_name: &'life str,
4419 table_name: &'life str,
4420 object: serde_json::Map<String, serde_json::Value>,
4421 context: &'life InsertContext<'life>,
4422 written_count: &'life mut usize,
4423) -> std::pin::Pin<
4424 Box<
4425 dyn std::future::Future<Output = Result<serde_json::Value, async_graphql::Error>>
4426 + Send
4427 + 'life,
4428 >,
4429> {
4430 Box::pin(async move {
4431 use sqlx::Row;
4432
4433 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
4434 let table = context.cache.get_table(&qi).ok_or_else(|| {
4435 async_graphql::Error::new(format!("unknown table \"{}\"", table_name))
4436 })?;
4437 let column_types: HashMap<String, String> = table
4438 .columns
4439 .values()
4440 .map(|c| (c.name.clone(), c.nominal_type.clone()))
4441 .collect();
4442
4443 let type_name = context
4445 .type_names
4446 .get(&(schema_name.to_string(), table_name.to_string()));
4447 let relationships: &[RelationshipField] = type_name
4448 .and_then(|name| context.relationships.get(name))
4449 .map(|r| r.as_slice())
4450 .unwrap_or(&[]);
4451
4452 let mut columns = serde_json::Map::new();
4453 type Nested<'r> = (
4454 &'r RelationshipField,
4455 serde_json::Value,
4456 Option<serde_json::Value>,
4457 );
4458 let mut to_one: Vec<Nested> = Vec::new();
4459 let mut to_many: Vec<Nested> = Vec::new();
4460
4461 for (key, value) in object {
4462 match relationships.iter().find(|r| r.name == key) {
4463 None => {
4464 columns.insert(
4467 table_column_for(context.names, table, &key).to_string(),
4468 value,
4469 );
4470 }
4471 Some(_) if value.is_null() => {
4472 continue;
4476 }
4477 Some(relationship) => {
4478 let (data, conflict) = match &value {
4482 serde_json::Value::Object(map) => (
4483 map.get("data").cloned().unwrap_or(value.clone()),
4484 map.get("on_conflict").cloned().filter(|v| !v.is_null()),
4485 ),
4486 other => (other.clone(), None),
4487 };
4488 if child_holds_the_key(&relationship.relationship) {
4495 to_many.push((relationship, data, conflict));
4496 } else {
4497 to_one.push((relationship, data, conflict));
4498 }
4499 }
4500 }
4501 }
4502
4503 for (column, value) in &columns {
4506 if matches!(
4507 column_types.get(column).map(String::as_str),
4508 Some("geometry") | Some("geography")
4509 ) {
4510 check_geojson(value).map_err(|error| {
4513 let field = context
4514 .names
4515 .column(schema_name, table_name, column)
4516 .unwrap_or(column);
4517 at_path(error, &format!("{}.{}", context.row, field))
4518 })?;
4519 }
4520 }
4521
4522 for (relationship, data, conflict) in to_one {
4525 let plan =
4526 postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, context.cache)
4527 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
4528 let serde_json::Value::Object(child) = data else {
4529 return Err(async_graphql::Error::new(format!(
4530 "\"{}\" takes an object to insert",
4531 relationship.name
4532 )));
4533 };
4534 for (local, _) in &plan.columns {
4538 if columns.get(local).is_some_and(|v| !v.is_null()) {
4539 return Err(at_path(
4540 validation_error(format!(
4541 "cannot insert object relationship \"{}\" as \"{}\" column \
4542 values are already determined",
4543 relationship.name, local
4544 )),
4545 &format!("{}.{}", context.row, relationship.name),
4546 ));
4547 }
4548 }
4549 let under = format!("{}.{}", context.row, relationship.name);
4552 let nested = InsertContext {
4553 on_conflict: conflict,
4554 cache: context.cache,
4555 relationships: context.relationships,
4556 type_names: context.type_names,
4557 names: context.names,
4558 caller: context.caller,
4559 row: format!("{}.data", under),
4560 objects: format!("{}.data", under),
4561 conflict: format!("{}.on_conflict", under),
4562 };
4563 let written = insert_row(
4564 conn,
4565 &plan.foreign_schema,
4566 &plan.foreign_table,
4567 child,
4568 &nested,
4569 written_count,
4570 )
4571 .await?;
4572 for (local, foreign) in &plan.columns {
4573 if let Some(value) = written.get(foreign) {
4574 columns.insert(local.clone(), value.clone());
4575 }
4576 }
4577 }
4578
4579 for (column, value) in crate::role::presets(
4585 &context.caller,
4586 context.names,
4587 schema_name,
4588 table_name,
4589 crate::role::Verb::Insert,
4590 )
4591 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?
4592 {
4593 columns.insert(column, value);
4594 }
4595
4596 let column_names: Vec<&str> = columns.keys().map(|k| k.as_str()).collect();
4602 let mut conflict_values: Vec<serde_json::Value> = Vec::new();
4603 let conflict_sql = match &context.on_conflict {
4604 Some(serde_json::Value::Object(spec)) => {
4605 let constraint = spec.get("constraint").and_then(|v| v.as_str());
4606 let Some(constraint) = constraint else {
4607 return Err(async_graphql::Error::new(
4608 "on_conflict needs a constraint to resolve against",
4609 ));
4610 };
4611 if !table
4612 .unique_constraints
4613 .iter()
4614 .any(|(name, _)| name == constraint)
4615 {
4616 return Err(async_graphql::Error::new(format!(
4617 "\"{}\" is not a unique constraint of \"{}\"",
4618 constraint, table_name
4619 )));
4620 }
4621 let updates: Vec<&str> = spec
4622 .get("update_columns")
4623 .and_then(|v| v.as_array())
4624 .map(|items| items.iter().filter_map(|i| i.as_str()).collect())
4625 .unwrap_or_default();
4626 if updates.contains(&PLACEHOLDER_COLUMN) {
4630 return Err(at_path(
4631 coded_error("validation-failed", "erroneous column name"),
4632 &context.conflict,
4633 ));
4634 }
4635
4636 if updates.is_empty() {
4637 format!(
4638 " ON CONFLICT ON CONSTRAINT {} DO NOTHING",
4639 postrust_sql::escape_ident(constraint)
4640 )
4641 } else {
4642 let preset = crate::role::presets(
4648 &context.caller,
4649 context.names,
4650 schema_name,
4651 table_name,
4652 crate::role::Verb::Update,
4653 )
4654 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
4655
4656 let mut param_idx = column_names.len() + 1;
4660 let mut assignments: Vec<String> =
4661 Vec::with_capacity(preset.len() + updates.len());
4662 for (column, value) in &preset {
4663 assignments.push(format!(
4664 "{} = {}",
4665 postrust_sql::escape_ident(column),
4666 write_expression(
4667 &column_types,
4668 column,
4669 value,
4670 &format!("${}", param_idx)
4671 )
4672 ));
4673 conflict_values.push(write_operand(&column_types, column, value));
4674 param_idx += 1;
4675 }
4676 for field in &updates {
4677 let column = table_column_for(context.names, table, field);
4678 if preset.iter().any(|(written, _)| written == column) {
4679 continue;
4680 }
4681 assignments.push(format!(
4682 "{} = EXCLUDED.{}",
4683 postrust_sql::escape_ident(column),
4684 postrust_sql::escape_ident(column)
4685 ));
4686 }
4687 let granted = permission_filter(
4700 &context.caller,
4701 context.names,
4702 schema_name,
4703 table_name,
4704 crate::role::Verb::Update,
4705 )?;
4706 let requested = spec.get("where").filter(|f| !f.is_null()).cloned();
4707 let condition = match and_predicates(requested, granted) {
4708 Some(filter) => {
4709 let filter = &filter;
4710 let type_name = context
4711 .type_names
4712 .get(&(schema_name.to_string(), table_name.to_string()))
4713 .map(String::as_str)
4714 .unwrap_or(table_name);
4715 let scope = WhereScope::table(
4716 schema_name,
4717 table_name,
4718 type_name,
4719 context.names,
4720 )
4721 .under_alias(WRITTEN_ROW)
4722 .with_resolution(context.cache, context.relationships);
4723 let mut alias_counter = 0usize;
4724 build_condition(
4725 filter,
4726 &scope,
4727 &mut param_idx,
4728 &mut conflict_values,
4729 &mut alias_counter,
4730 )?
4731 .map(|sql| format!(" WHERE {}", sql))
4732 .unwrap_or_default()
4733 }
4734 None => String::new(),
4735 };
4736 format!(
4737 " ON CONFLICT ON CONSTRAINT {} DO UPDATE SET {}{}",
4738 postrust_sql::escape_ident(constraint),
4739 assignments.join(", "),
4740 condition
4741 )
4742 }
4743 }
4744 _ => String::new(),
4745 };
4746
4747 let qualified_table = postrust_sql::escape_ident(WRITTEN_ROW);
4753
4754 let check_scope = WhereScope::for_alias(
4759 schema_name,
4760 table_name,
4761 WRITTEN_ROW,
4762 context
4763 .type_names
4764 .get(&(schema_name.to_string(), table_name.to_string()))
4765 .map(String::as_str)
4766 .unwrap_or(table_name),
4767 context.cache,
4768 context.relationships,
4769 context.names,
4770 );
4771 let (check_sql, check_values) = {
4772 let check = crate::role::write_check(
4773 &context.caller,
4774 context.names,
4775 schema_name,
4776 table_name,
4777 crate::role::Verb::Insert,
4778 )
4779 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
4780 match check {
4781 None => (String::new(), Vec::new()),
4782 Some(check) => {
4783 let mut values = Vec::new();
4784 let mut param_idx = column_names.len() + conflict_values.len() + 1;
4785 let mut alias_counter = 0usize;
4786 match build_condition(
4787 &check,
4788 &check_scope,
4789 &mut param_idx,
4790 &mut values,
4791 &mut alias_counter,
4792 )? {
4793 None => (String::new(), Vec::new()),
4794 Some(sql) => (
4795 format!(
4796 ", ({}) AS {}",
4797 sql,
4798 postrust_sql::escape_ident(CHECK_COLUMN)
4799 ),
4800 values,
4801 ),
4802 }
4803 }
4804 }
4805 };
4806
4807 let names = column_names;
4808 let written = if names.is_empty() {
4809 let sql = format!(
4812 "INSERT INTO {}.{} AS {} DEFAULT VALUES{} RETURNING {}{}",
4813 postrust_sql::escape_ident(schema_name),
4814 postrust_sql::escape_ident(table_name),
4815 qualified_table,
4816 conflict_sql,
4817 row_json(&qualified_table, &column_types),
4818 check_sql
4819 );
4820 let mut query = sqlx::query(&sql);
4821 for value in &conflict_values {
4822 query = bind_json_value(query, value);
4823 }
4824 for value in &check_values {
4825 query = bind_json_value(query, value);
4826 }
4827 query
4828 .fetch_optional(&mut *conn)
4829 .await
4830 .map_err(|error| at_path(database_error(error), &context.objects))?
4833 } else {
4834 let placeholders: Vec<String> = names
4835 .iter()
4836 .enumerate()
4837 .map(|(i, column)| {
4838 let placeholder = format!("${}", i + 1);
4839 match columns.get(*column) {
4840 Some(value) => write_expression(&column_types, column, value, &placeholder),
4841 None => placeholder,
4842 }
4843 })
4844 .collect();
4845 let sql = format!(
4846 "INSERT INTO {}.{} AS {} ({}) VALUES ({}){} RETURNING {}{}",
4847 postrust_sql::escape_ident(schema_name),
4848 postrust_sql::escape_ident(table_name),
4849 qualified_table,
4850 names
4851 .iter()
4852 .map(|c| postrust_sql::escape_ident(c))
4853 .collect::<Vec<_>>()
4854 .join(", "),
4855 placeholders.join(", "),
4856 conflict_sql,
4857 row_json(&qualified_table, &column_types),
4858 check_sql
4859 );
4860 trace!("Executing INSERT SQL: {}", sql);
4861 let mut query = sqlx::query(&sql);
4862 for column in &names {
4863 if let Some(value) = columns.get(*column) {
4864 query = bind_json_value(query, &write_operand(&column_types, column, value));
4865 }
4866 }
4867 for value in &conflict_values {
4868 query = bind_json_value(query, value);
4869 }
4870 for value in &check_values {
4871 query = bind_json_value(query, value);
4872 }
4873 query
4874 .fetch_optional(&mut *conn)
4875 .await
4876 .map_err(|error| at_path(database_error(error), &context.objects))?
4879 };
4880
4881 let Some(written) = written else {
4884 return Ok(serde_json::Value::Null);
4885 };
4886 refuse_unchecked_rows(std::slice::from_ref(&written), !check_sql.is_empty())
4887 .map_err(|error| at_path(error, &context.objects))?;
4888 let row: serde_json::Value = written.try_get(0).unwrap_or(serde_json::Value::Null);
4889 *written_count += 1;
4890
4891 for (relationship, data, conflict) in to_many {
4893 let plan =
4894 postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, context.cache)
4895 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
4896 let children = match data {
4897 serde_json::Value::Array(items) => items,
4898 serde_json::Value::Object(map) => vec![serde_json::Value::Object(map)],
4899 serde_json::Value::Null => Vec::new(),
4900 _ => {
4901 return Err(async_graphql::Error::new(format!(
4902 "\"{}\" takes objects to insert",
4903 relationship.name
4904 )))
4905 }
4906 };
4907 for (child_index, child) in children.into_iter().enumerate() {
4908 let serde_json::Value::Object(mut child) = child else {
4909 continue;
4910 };
4911 for (local, foreign) in &plan.columns {
4912 if child.get(foreign).is_some_and(|v| !v.is_null()) {
4916 return Err(at_path(
4917 validation_error(format!(
4918 "cannot insert \"{}\" columns as their values are already \
4919 being determined by parent insert",
4920 foreign
4921 )),
4922 &format!(
4923 "{}.{}{}",
4924 context.row,
4925 relationship.name,
4926 match relationship.is_list {
4927 true => format!(".data[{}]", child_index),
4928 false => ".data".to_string(),
4929 }
4930 ),
4931 ));
4932 }
4933 if let Some(value) = row.get(local) {
4934 child.insert(foreign.clone(), value.clone());
4935 }
4936 }
4937 let at = match relationship.is_list {
4943 true => format!(".data[{}]", child_index),
4944 false => ".data".to_string(),
4945 };
4946 let under = format!("{}.{}", context.row, relationship.name);
4947 let nested = InsertContext {
4948 on_conflict: conflict.clone(),
4949 cache: context.cache,
4950 relationships: context.relationships,
4951 type_names: context.type_names,
4952 names: context.names,
4953 caller: context.caller,
4954 row: format!("{}{}", under, at),
4955 objects: format!("{}.data", under),
4956 conflict: format!("{}.on_conflict", under),
4957 };
4958 insert_row(
4959 conn,
4960 &plan.foreign_schema,
4961 &plan.foreign_table,
4962 child,
4963 &nested,
4964 written_count,
4965 )
4966 .await?;
4967 }
4968 }
4969
4970 Ok(row)
4971 })
4972}
4973
4974#[derive(Clone)]
4980struct InsertContext<'a> {
4981 on_conflict: Option<serde_json::Value>,
4984 cache: &'a SchemaCache,
4985 relationships: &'a HashMap<String, Vec<RelationshipField>>,
4986 type_names: &'a HashMap<(String, String), String>,
4990 names: &'a crate::names::NameOverrides,
4993 caller: crate::role::Caller<'a>,
4997 row: String,
5002 objects: String,
5006 conflict: String,
5008}
5009
5010#[allow(clippy::too_many_arguments)]
5023async fn reread_returning(
5024 pool: &PgPool,
5025 gql_ctx: &GraphQLContext,
5026 schema_name: &str,
5027 table_name: &str,
5028 type_name: &str,
5029 rows: Vec<Value>,
5030 returning: async_graphql::SelectionField<'_>,
5031 relationships: &HashMap<String, Vec<RelationshipField>>,
5032 names: &crate::names::NameOverrides,
5033 max_rows: Option<i64>,
5034) -> Result<Vec<Value>, async_graphql::Error> {
5035 let guard = gql_ctx
5036 .schema_cache
5037 .get()
5038 .await
5039 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
5040 let Some(cache) = guard.as_ref() else {
5041 return Ok(rows);
5042 };
5043 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
5044 let Some(table) = cache.get_table(&qi) else {
5045 return Ok(rows);
5046 };
5047 if table.pk_cols.is_empty() {
5048 return Ok(rows);
5050 }
5051
5052 let mut param_idx = 1usize;
5053 let mut values: Vec<serde_json::Value> = Vec::new();
5054 let mut alias_counter = 0usize;
5055 let embeds = build_embed_expressions(
5056 &gql_ctx.caller(),
5057 cache,
5058 relationships,
5059 type_name,
5060 "src",
5061 returning,
5062 max_rows,
5063 &mut alias_counter,
5064 &mut param_idx,
5065 &mut values,
5066 names,
5067 )?;
5068 let computed = computed_projections(
5069 table,
5070 returning,
5071 "src",
5072 names,
5073 cache,
5074 &mut param_idx,
5075 &mut values,
5076 )?;
5077 let renames = names.renames_columns(&table.schema, &table.name);
5081 if embeds.is_empty() && computed.is_empty() && !renames {
5082 return Ok(rows);
5083 }
5084
5085 let mut conditions = Vec::with_capacity(rows.len());
5087 for row in &rows {
5088 let mut parts = Vec::with_capacity(table.pk_cols.len());
5089 for column in &table.pk_cols {
5090 let Value::Object(map) = row else {
5091 return Ok(rows);
5092 };
5093 let Some(value) = map.get(&async_graphql::Name::new(column)) else {
5094 return Ok(rows);
5095 };
5096 parts.push(format!(
5097 "{} = ${}",
5098 postrust_sql::escape_ident(column),
5099 param_idx
5100 ));
5101 values.push(value_to_json(value));
5102 param_idx += 1;
5103 }
5104 conditions.push(format!("({})", parts.join(" AND ")));
5105 }
5106 if conditions.is_empty() {
5107 return Ok(rows);
5108 }
5109
5110 let mut projection =
5111 rename_projection(table, "src", names).unwrap_or_else(|| "src.*".to_string());
5112 for expression in &computed {
5113 projection.push_str(", ");
5114 projection.push_str(expression);
5115 }
5116 for (name, expression) in &embeds {
5117 projection.push_str(", ");
5118 projection.push_str(expression);
5119 projection.push_str(" AS ");
5120 projection.push_str(&postrust_sql::escape_ident(name));
5121 }
5122
5123 let sql = format!(
5124 "SELECT {} FROM (SELECT {} FROM (SELECT * FROM {}.{} WHERE {}) AS src) AS pgrst_r",
5125 row_json("pgrst_r", &exposed_column_types(table, names)),
5126 projection,
5127 postrust_sql::escape_ident(schema_name),
5128 postrust_sql::escape_ident(table_name),
5129 conditions.join(" OR ")
5130 );
5131
5132 let mut guard = write_tx(gql_ctx, pool).await?;
5135 let conn = guard.as_mut().expect("write_tx opens one");
5136 let reread = execute_query_on(conn, &sql, &values).await?;
5137 drop(guard);
5138
5139 if reread.is_empty() {
5140 return Ok(rows);
5141 }
5142 Ok(reread.into_iter().map(json_to_value).collect())
5143}
5144
5145async fn execute_insert(
5147 pool: &PgPool,
5148 gql_ctx: &GraphQLContext,
5149 schema_name: &str,
5150 table_name: &str,
5151 objects: serde_json::Value,
5152 context: &InsertContext<'_>,
5153 single: bool,
5157) -> Result<(Vec<Value>, usize), async_graphql::Error> {
5158 trace!("Insert mutation for {}: {:?}", table_name, objects);
5159
5160 let objects_array = match objects {
5162 serde_json::Value::Array(arr) => arr,
5163 serde_json::Value::Object(obj) => vec![serde_json::Value::Object(obj)],
5164 _ => {
5165 return Err(async_graphql::Error::new(
5166 "objects must be an array or object",
5167 ))
5168 }
5169 };
5170
5171 if objects_array.is_empty() {
5172 return Err(async_graphql::Error::new("objects cannot be empty"));
5173 }
5174
5175 let mut guard = write_tx(gql_ctx, pool).await?;
5176 let conn = guard.as_mut().expect("write_tx opens one");
5177
5178 let mut inserted: Vec<Value> = Vec::new();
5179 let mut written = 0usize;
5180
5181 for (index, object) in objects_array.into_iter().enumerate() {
5182 let serde_json::Value::Object(map) = object else {
5183 return Err(async_graphql::Error::new(
5184 "each object to insert is an object",
5185 ));
5186 };
5187 let context = InsertContext {
5190 row: match single {
5191 true => context.objects.clone(),
5192 false => format!("{}[{}]", context.objects, index),
5193 },
5194 ..context.clone()
5195 };
5196 let row = insert_row(conn, schema_name, table_name, map, &context, &mut written).await?;
5197 if !row.is_null() {
5201 inserted.push(json_to_value(row));
5202 }
5203 }
5204
5205 Ok((inserted, written))
5206}
5207
5208fn mutation_result<'a>(rows: Vec<Value>, affected: usize, by_key: bool) -> Option<FieldValue<'a>> {
5220 if by_key {
5221 return rows.into_iter().next().map(FieldValue::value);
5222 }
5223 let mut response = async_graphql::indexmap::IndexMap::new();
5224 response.insert(
5225 async_graphql::Name::new("affected_rows"),
5226 Value::from(affected),
5227 );
5228 response.insert(async_graphql::Name::new("returning"), Value::List(rows));
5229 Some(FieldValue::value(Value::Object(response)))
5230}
5231
5232fn bind_json_value<'q>(
5234 query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
5235 value: &serde_json::Value,
5236) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> {
5237 match value {
5238 serde_json::Value::Null => query.bind(None::<String>),
5239 serde_json::Value::Bool(b) => query.bind(*b),
5240 serde_json::Value::Number(n) => {
5241 if let Some(i) = n.as_i64() {
5242 query.bind(i)
5243 } else if let Some(f) = n.as_f64() {
5244 query.bind(f)
5245 } else {
5246 query.bind(n.to_string())
5247 }
5248 }
5249 serde_json::Value::String(s) => query.bind(s.clone()),
5250 _ => query.bind(value.to_string()),
5251 }
5252}
5253
5254#[allow(clippy::too_many_arguments)] async fn execute_update(
5257 pool: &PgPool,
5258 gql_ctx: &GraphQLContext,
5259 schema_name: &str,
5260 table_name: &str,
5261 type_name: &str,
5262 operators: Vec<(&'static str, serde_json::Value)>,
5263 column_types: HashMap<String, String>,
5264 where_clause: Option<serde_json::Value>,
5265 relationships: &HashMap<String, Vec<RelationshipField>>,
5266 names: &crate::names::NameOverrides,
5267) -> Result<Vec<Value>, async_graphql::Error> {
5268 use sqlx::Row;
5269
5270 trace!("Update mutation for {}: {:?}", table_name, operators);
5271
5272 let mut set_parts: Vec<String> = Vec::new();
5277 let mut set_values: Vec<serde_json::Value> = Vec::new();
5278 let mut param_idx = 1;
5279 let mut written: HashSet<String> = HashSet::new();
5280
5281 let preset = crate::role::presets(
5291 &gql_ctx.caller(),
5292 names,
5293 schema_name,
5294 table_name,
5295 crate::role::Verb::Update,
5296 )
5297 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
5298 let preset_columns: HashSet<&str> = preset.iter().map(|(name, _)| name.as_str()).collect();
5299 for (column, value) in &preset {
5300 let placeholder =
5301 write_expression(&column_types, column, value, &format!("${}", param_idx));
5302 set_parts.push(format!(
5303 "{} = {}",
5304 postrust_sql::escape_ident(column),
5305 placeholder
5306 ));
5307 set_values.push(write_operand(&column_types, column, value));
5308 param_idx += 1;
5309 }
5310
5311 for (operator, payload) in &operators {
5312 let serde_json::Value::Object(map) = payload else {
5313 return Err(async_graphql::Error::new(format!(
5314 "\"{}\" takes an object mapping columns to values",
5315 operator
5316 )));
5317 };
5318 for (field, value) in map {
5319 let column = &names
5320 .column_source(schema_name, table_name, field)
5321 .unwrap_or(field)
5322 .to_string();
5323 if preset_columns.contains(column.as_str()) {
5326 continue;
5327 }
5328 if !written.insert(column.clone()) {
5329 return Err(async_graphql::Error::new(format!(
5330 "\"{}\" is written twice in one update; a column may be \
5331 changed by one operator at a time",
5332 column
5333 )));
5334 }
5335 let quoted = postrust_sql::escape_ident(column);
5336 let placeholder = match *operator {
5345 "_set" | "_inc" => {
5346 write_expression(&column_types, column, value, &format!("${}", param_idx))
5347 }
5348 _ => format!("${}", param_idx),
5349 };
5350 let assignment = match *operator {
5353 "_set" => format!("{} = {}", quoted, placeholder),
5354 "_inc" => format!("{} = {} + {}", quoted, quoted, placeholder),
5355 "_append" => format!("{} = {} || {}::jsonb", quoted, quoted, placeholder),
5356 "_prepend" => format!("{} = {}::jsonb || {}", quoted, placeholder, quoted),
5357 "_delete_key" => format!("{} = {} - {}::text", quoted, quoted, placeholder),
5358 "_delete_elem" => format!("{} = {} - {}::int", quoted, quoted, placeholder),
5359 "_delete_at_path" => {
5365 let Some(steps) = value.as_array() else {
5366 return Err(async_graphql::Error::new(format!(
5367 "\"_delete_at_path\" on \"{}\" takes a list of keys",
5368 column
5369 )));
5370 };
5371 let path = sql_array(steps, "text[]", &mut param_idx, &mut set_values);
5372 set_parts.push(format!("{} = {} #- {}", quoted, quoted, path));
5373 continue;
5374 }
5375 other => {
5376 return Err(async_graphql::Error::new(format!(
5377 "unsupported update operator \"{}\"",
5378 other
5379 )))
5380 }
5381 };
5382 if *operator == "_set"
5383 && matches!(
5384 column_types.get(column).map(String::as_str),
5385 Some("geometry") | Some("geography")
5386 )
5387 {
5388 check_geojson(value)?;
5389 }
5390 set_parts.push(assignment);
5391 set_values.push(match *operator {
5392 "_set" => write_operand(&column_types, column, value),
5393 _ => value.clone(),
5394 });
5395 param_idx += 1;
5396 }
5397 }
5398
5399 if set_parts.is_empty() {
5404 return Ok(Vec::new());
5405 }
5406
5407 let guard = gql_ctx
5411 .schema_cache
5412 .get()
5413 .await
5414 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
5415 let cache = guard
5416 .as_ref()
5417 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
5418 let scope = WhereScope::table(schema_name, table_name, type_name, names)
5419 .under_alias(WRITTEN_ROW)
5420 .with_resolution(cache, relationships);
5421 let (where_sql, where_values) = build_where_clause(where_clause.as_ref(), param_idx, &scope)?;
5422
5423 if where_sql.is_empty() {
5426 return Err(async_graphql::Error::new(format!(
5427 "update on \"{}\" requires a `where` argument with at least one \
5428 recognised condition; refusing to update every row",
5429 table_name
5430 )));
5431 }
5432
5433 let (check_sql, check_values) = permission_check_sql(
5438 gql_ctx,
5439 names,
5440 schema_name,
5441 table_name,
5442 crate::role::Verb::Update,
5443 &scope,
5444 param_idx + where_values.len(),
5445 )?;
5446
5447 let sql = format!(
5448 "UPDATE {}.{} AS {} SET {} {} RETURNING {}{}",
5449 postrust_sql::escape_ident(schema_name),
5450 postrust_sql::escape_ident(table_name),
5451 postrust_sql::escape_ident(WRITTEN_ROW),
5452 set_parts.join(", "),
5453 where_sql,
5454 row_json(&postrust_sql::escape_ident(WRITTEN_ROW), &column_types),
5455 check_sql
5456 );
5457
5458 trace!("Executing UPDATE SQL: {}", sql);
5459
5460 let mut query = sqlx::query(&sql);
5462
5463 for val in &set_values {
5465 query = bind_json_value(query, val);
5466 }
5467
5468 for val in &where_values {
5470 query = bind_json_value(query, val);
5471 }
5472
5473 for val in &check_values {
5475 query = bind_json_value(query, val);
5476 }
5477
5478 let mut guard = write_tx(gql_ctx, pool).await?;
5479 let conn = guard.as_mut().expect("write_tx opens one");
5480 let rows = query.fetch_all(&mut **conn).await.map_err(database_error)?;
5481
5482 refuse_unchecked_rows(&rows, !check_sql.is_empty())?;
5483
5484 let updated: Vec<Value> = rows
5485 .iter()
5486 .filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
5487 .map(json_to_value)
5488 .collect();
5489
5490 Ok(updated)
5491}
5492
5493fn permission_check_sql(
5500 gql_ctx: &GraphQLContext,
5501 names: &crate::names::NameOverrides,
5502 schema_name: &str,
5503 table_name: &str,
5504 verb: crate::role::Verb,
5505 scope: &WhereScope<'_>,
5506 start_param_idx: usize,
5507) -> Result<(String, Vec<serde_json::Value>), async_graphql::Error> {
5508 let check = crate::role::write_check(&gql_ctx.caller(), names, schema_name, table_name, verb)
5509 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
5510 let Some(check) = check else {
5511 return Ok((String::new(), Vec::new()));
5512 };
5513
5514 let mut values = Vec::new();
5515 let mut param_idx = start_param_idx;
5516 let mut alias_counter = 0usize;
5517 let condition = build_condition(
5518 &check,
5519 scope,
5520 &mut param_idx,
5521 &mut values,
5522 &mut alias_counter,
5523 )?;
5524 Ok(match condition {
5525 None => (String::new(), Vec::new()),
5528 Some(sql) => (
5529 format!(
5530 ", ({}) AS {}",
5531 sql,
5532 postrust_sql::escape_ident(CHECK_COLUMN)
5533 ),
5534 values,
5535 ),
5536 })
5537}
5538
5539const CHECK_COLUMN: &str = "pgrst_permission_check";
5541
5542fn refuse_unchecked_rows(
5549 rows: &[sqlx::postgres::PgRow],
5550 checked: bool,
5551) -> Result<(), async_graphql::Error> {
5552 if !checked {
5553 return Ok(());
5554 }
5555 use sqlx::Row;
5556 let all_passed = rows
5557 .iter()
5558 .all(|row| row.try_get::<bool, _>(CHECK_COLUMN).unwrap_or(false));
5559 match all_passed {
5560 true => Ok(()),
5561 false => Err(coded_error("permission-error", crate::role::CHECK_FAILED)),
5562 }
5563}
5564
5565#[allow(clippy::too_many_arguments)] async fn execute_delete(
5568 pool: &PgPool,
5569 gql_ctx: &GraphQLContext,
5570 schema_name: &str,
5571 table_name: &str,
5572 type_name: &str,
5573 where_clause: Option<serde_json::Value>,
5574 returning: Option<async_graphql::SelectionField<'_>>,
5575 relationships: &HashMap<String, Vec<RelationshipField>>,
5576 names: &crate::names::NameOverrides,
5577 max_rows: Option<i64>,
5578) -> Result<Vec<Value>, async_graphql::Error> {
5579 use sqlx::Row;
5580
5581 trace!("Delete mutation for {}", table_name);
5582
5583 let guard = gql_ctx
5584 .schema_cache
5585 .get()
5586 .await
5587 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
5588 let cache = guard
5589 .as_ref()
5590 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
5591 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
5592 let column_types = cache
5593 .get_table(&qi)
5594 .map(|table| exposed_column_types(table, names))
5595 .unwrap_or_default();
5596
5597 let scope = WhereScope::table(schema_name, table_name, type_name, names)
5601 .under_alias(WRITTEN_ROW)
5602 .with_resolution(cache, relationships);
5603 let (where_sql, mut values) = build_where_clause(where_clause.as_ref(), 1, &scope)?;
5604
5605 if where_sql.is_empty() {
5608 return Err(async_graphql::Error::new(format!(
5609 "delete on \"{}\" requires a `where` argument with at least one \
5610 recognised condition; refusing to delete every row",
5611 table_name
5612 )));
5613 }
5614
5615 let mut projection = String::new();
5621 let renamed = cache
5622 .get_table(&qi)
5623 .and_then(|table| rename_projection(table, "src", names));
5624 if let Some(returning) = returning {
5625 let mut param_idx = values.len() + 1;
5626 let mut alias_counter = 0usize;
5627 let embeds = build_embed_expressions(
5628 &gql_ctx.caller(),
5629 cache,
5630 relationships,
5631 type_name,
5632 "src",
5633 returning,
5634 max_rows,
5635 &mut alias_counter,
5636 &mut param_idx,
5637 &mut values,
5638 names,
5639 )?;
5640 let computed = match cache.get_table(&qi) {
5641 Some(table) => computed_projections(
5642 table,
5643 returning,
5644 "src",
5645 names,
5646 cache,
5647 &mut param_idx,
5648 &mut values,
5649 )?,
5650 None => Vec::new(),
5651 };
5652 for expression in &computed {
5653 projection.push_str(", ");
5654 projection.push_str(expression);
5655 }
5656 for (name, expression) in &embeds {
5657 projection.push_str(", ");
5658 projection.push_str(expression);
5659 projection.push_str(" AS ");
5660 projection.push_str(&postrust_sql::escape_ident(name));
5661 }
5662 }
5663
5664 let written = postrust_sql::escape_ident(WRITTEN_ROW);
5665 let sql = if projection.is_empty() && renamed.is_none() {
5666 format!(
5667 "DELETE FROM {}.{} AS {} {} RETURNING {}",
5668 postrust_sql::escape_ident(schema_name),
5669 postrust_sql::escape_ident(table_name),
5670 written,
5671 where_sql,
5672 row_json(&written, &column_types)
5673 )
5674 } else {
5675 format!(
5676 "WITH pgrst_deleted AS (DELETE FROM {}.{} AS {} {} RETURNING {}.*) \
5677 SELECT {} FROM (SELECT {}{} FROM pgrst_deleted AS src) AS pgrst_r",
5678 postrust_sql::escape_ident(schema_name),
5679 postrust_sql::escape_ident(table_name),
5680 written,
5681 where_sql,
5682 written,
5683 row_json("pgrst_r", &column_types),
5684 renamed.unwrap_or_else(|| "src.*".to_string()),
5685 projection
5686 )
5687 };
5688
5689 trace!("Executing DELETE SQL: {}", sql);
5690
5691 let mut query = sqlx::query(&sql);
5692 for val in &values {
5693 query = bind_json_value(query, val);
5694 }
5695
5696 let mut guard = write_tx(gql_ctx, pool).await?;
5697 let conn = guard.as_mut().expect("write_tx opens one");
5698 let rows = query.fetch_all(&mut **conn).await.map_err(database_error)?;
5699
5700 let deleted: Vec<Value> = rows
5701 .iter()
5702 .filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
5703 .map(json_to_value)
5704 .collect();
5705
5706 Ok(deleted)
5707}
5708
5709fn permission_predicate(
5717 caller: &crate::role::Caller<'_>,
5718 names: &crate::names::NameOverrides,
5719 schema: &str,
5720 table: &str,
5721) -> Result<Option<serde_json::Value>, async_graphql::Error> {
5722 permission_filter(caller, names, schema, table, crate::role::Verb::Select)
5723}
5724
5725fn permission_filter(
5727 caller: &crate::role::Caller<'_>,
5728 names: &crate::names::NameOverrides,
5729 schema: &str,
5730 table: &str,
5731 verb: crate::role::Verb,
5732) -> Result<Option<serde_json::Value>, async_graphql::Error> {
5733 crate::role::row_filter(caller, names, schema, table, verb)
5734 .map(|filter| filter.map(|predicate| serde_json::json!({ PERMISSION: predicate })))
5735 .map_err(|fault| coded_error(fault.code(), fault.to_string()))
5736}
5737
5738const PERMISSION: &str = "_permission";
5759
5760fn permission_limit(
5766 caller: &crate::role::Caller<'_>,
5767 names: &crate::names::NameOverrides,
5768 schema: &str,
5769 table: &str,
5770) -> Option<i64> {
5771 crate::role::read_limit(caller, names, schema, table)
5772}
5773
5774fn and_predicates(
5781 request: Option<serde_json::Value>,
5782 permission: Option<serde_json::Value>,
5783) -> Option<serde_json::Value> {
5784 match (request, permission) {
5785 (None, None) => None,
5786 (Some(only), None) | (None, Some(only)) => Some(only),
5787 (Some(request), Some(permission)) => {
5788 Some(serde_json::json!({ "_and": [request, permission] }))
5789 }
5790 }
5791}
5792
5793fn build_where_clause(
5803 where_value: Option<&serde_json::Value>,
5804 start_param_idx: usize,
5805 scope: &WhereScope<'_>,
5806) -> Result<(String, Vec<serde_json::Value>), async_graphql::Error> {
5807 let mut values: Vec<serde_json::Value> = Vec::new();
5808 let mut param_idx = start_param_idx;
5809 let mut alias_counter = 0usize;
5810
5811 let condition = match where_value {
5812 Some(value) => build_condition(
5813 value,
5814 scope,
5815 &mut param_idx,
5816 &mut values,
5817 &mut alias_counter,
5818 )?,
5819 None => None,
5820 };
5821
5822 Ok(match condition {
5823 Some(sql) => (format!("WHERE {}", sql), values),
5824 None => (String::new(), values),
5825 })
5826}
5827
5828pub struct WhereScope<'a> {
5840 qualified: postrust_core::api_request::QualifiedIdentifier,
5842 sql_ref: String,
5844 row_ref: String,
5848 type_name: String,
5850 root_ref: String,
5858 root_qualified: postrust_core::api_request::QualifiedIdentifier,
5859 names: &'a crate::names::NameOverrides,
5862 resolution: Option<WhereResolution<'a>>,
5863 caller: Option<crate::role::Caller<'a>>,
5870}
5871
5872struct WhereResolution<'a> {
5874 cache: &'a SchemaCache,
5875 relationships: &'a HashMap<String, Vec<RelationshipField>>,
5876}
5877
5878impl<'a> WhereScope<'a> {
5879 pub fn table(
5881 schema: &str,
5882 table: &str,
5883 type_name: &str,
5884 names: &'a crate::names::NameOverrides,
5885 ) -> Self {
5886 Self {
5887 qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5888 sql_ref: format!(
5889 "{}.{}",
5890 postrust_sql::escape_ident(schema),
5891 postrust_sql::escape_ident(table)
5892 ),
5893 row_ref: postrust_sql::escape_ident(table),
5894 type_name: type_name.to_string(),
5895 root_ref: format!(
5896 "{}.{}",
5897 postrust_sql::escape_ident(schema),
5898 postrust_sql::escape_ident(table)
5899 ),
5900 root_qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5901 names,
5902 resolution: None,
5903 caller: None,
5904 }
5905 }
5906
5907 fn under_alias(mut self, alias: &str) -> Self {
5915 if self.root_ref == self.sql_ref {
5918 self.root_ref = postrust_sql::escape_ident(alias);
5919 }
5920 self.sql_ref = postrust_sql::escape_ident(alias);
5921 self.row_ref = postrust_sql::escape_ident(alias);
5922 self
5923 }
5924
5925 fn with_resolution(
5927 mut self,
5928 cache: &'a SchemaCache,
5929 relationships: &'a HashMap<String, Vec<RelationshipField>>,
5930 ) -> Self {
5931 self.resolution = Some(WhereResolution {
5932 cache,
5933 relationships,
5934 });
5935 self
5936 }
5937
5938 #[allow(clippy::too_many_arguments)]
5940 fn for_alias(
5941 schema: &str,
5942 table: &str,
5943 alias: &str,
5944 type_name: &str,
5945 cache: &'a SchemaCache,
5946 relationships: &'a HashMap<String, Vec<RelationshipField>>,
5947 names: &'a crate::names::NameOverrides,
5948 ) -> Self {
5949 Self {
5950 qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5951 sql_ref: postrust_sql::escape_ident(alias),
5952 row_ref: postrust_sql::escape_ident(alias),
5953 type_name: type_name.to_string(),
5954 root_ref: postrust_sql::escape_ident(alias),
5955 root_qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5956 names,
5957 resolution: Some(WhereResolution {
5958 cache,
5959 relationships,
5960 }),
5961 caller: None,
5962 }
5963 }
5964
5965 fn aliased(
5972 schema: &str,
5973 table: &str,
5974 alias: &str,
5975 type_name: &str,
5976 from: &WhereScope<'a>,
5977 ) -> Self {
5978 Self {
5979 qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5980 sql_ref: postrust_sql::escape_ident(alias),
5981 row_ref: postrust_sql::escape_ident(alias),
5982 type_name: type_name.to_string(),
5983 root_ref: from.root_ref.clone(),
5986 root_qualified: from.root_qualified.clone(),
5987 names: from.names,
5988 resolution: from.resolution.as_ref().map(|r| WhereResolution {
5989 cache: r.cache,
5990 relationships: r.relationships,
5991 }),
5992 caller: from.caller,
5996 }
5997 }
5998
5999 pub fn for_caller(mut self, caller: crate::role::Caller<'a>) -> Self {
6001 self.caller = Some(caller);
6002 self
6003 }
6004
6005 fn as_system(&self) -> WhereScope<'a> {
6008 WhereScope {
6009 qualified: self.qualified.clone(),
6010 sql_ref: self.sql_ref.clone(),
6011 row_ref: self.row_ref.clone(),
6012 type_name: self.type_name.clone(),
6013 root_ref: self.sql_ref.clone(),
6020 root_qualified: self.qualified.clone(),
6021 names: self.names,
6022 resolution: self.resolution.as_ref().map(|r| WhereResolution {
6023 cache: r.cache,
6024 relationships: r.relationships,
6025 }),
6026 caller: None,
6027 }
6028 }
6029
6030 fn column(&self, name: &str) -> String {
6038 let column = self.column_name(name);
6039 let plain = format!("{}.{}", self.sql_ref, postrust_sql::escape_ident(column));
6040 let Some(cache) = self.resolution.as_ref().map(|r| r.cache) else {
6041 return plain;
6042 };
6043 let Some(table) = cache.get_table(&self.qualified) else {
6044 return plain;
6045 };
6046 if table.get_column(column).is_some() {
6047 return plain;
6048 }
6049 let function = self
6050 .names
6051 .computed_source(&table.schema, &table.name, name)
6052 .unwrap_or(name);
6053 match table.get_computed_column(function) {
6054 Some(definition) => {
6055 computed_call(definition, &format!("{}.*", self.row_ref), &self.row_ref)
6056 }
6057 None => plain,
6058 }
6059 }
6060
6061 fn column_name<'n>(&'n self, name: &'n str) -> &'n str {
6063 self.names
6064 .column_source(&self.qualified.schema, &self.qualified.name, name)
6065 .unwrap_or(name)
6066 }
6067
6068 fn column_type(&self, name: &str) -> Option<String> {
6072 let cache = self.resolution.as_ref()?.cache;
6073 let table = cache.get_table(&self.qualified)?;
6074 table
6075 .get_column(self.column_name(name))
6076 .map(|c| c.nominal_type.clone())
6077 }
6078
6079 fn other_column(
6093 &self,
6094 operand: &serde_json::Value,
6095 op: &str,
6096 ) -> Result<String, async_graphql::Error> {
6097 use serde_json::Value;
6098
6099 let refused = |what: &str| {
6100 async_graphql::Error::new(format!(
6101 "the \"{}\" comparison takes the name of a column, {}",
6102 op, what
6103 ))
6104 };
6105 let (against, qualified, name) = match operand {
6106 Value::String(name) => (&self.sql_ref, &self.qualified, name.as_str()),
6107 Value::Array(path) => match path.as_slice() {
6108 [Value::String(name)] => (&self.sql_ref, &self.qualified, name.as_str()),
6109 [Value::String(root), Value::String(name)] if root == "$" => {
6110 (&self.root_ref, &self.root_qualified, name.as_str())
6111 }
6112 _ => return Err(refused("or [\"$\", \"a column\"]")),
6113 },
6114 _ => return Err(refused("written as a string")),
6115 };
6116 let column = self
6119 .names
6120 .column_source(&qualified.schema, &qualified.name, name)
6121 .unwrap_or(name);
6122 Ok(format!(
6123 "{}.{}",
6124 against,
6125 postrust_sql::escape_ident(column)
6126 ))
6127 }
6128
6129 fn relationship(&self, name: &str) -> Option<std::borrow::Cow<'a, RelationshipField>> {
6143 let resolution = self.resolution.as_ref()?;
6144 let published = resolution
6145 .relationships
6146 .get(&self.type_name)
6147 .and_then(|fields| fields.iter().find(|r| r.name == name));
6148 if let Some(found) = published {
6149 return Some(std::borrow::Cow::Borrowed(found));
6150 }
6151
6152 let table = resolution.cache.get_table(&self.qualified)?;
6153
6154 if let Some(declared) = self
6160 .names
6161 .declared_relationships(&table.schema, &table.name)
6162 {
6163 if let Some(entry) = declared.iter().find(|entry| entry.name == name) {
6164 if let Some(field) = crate::schema::mapped_relationship_field(
6167 entry,
6168 table,
6169 resolution.cache,
6170 &table.schema,
6171 &entry
6172 .target(&table.schema)
6173 .map(|(_, target)| target)
6174 .unwrap_or_default(),
6175 ) {
6176 return Some(std::borrow::Cow::Owned(field));
6177 }
6178 if let Some(key) = entry.using.as_deref() {
6182 if let Some(found) = resolution
6183 .cache
6184 .get_relationships(&self.qualified, &table.schema)
6185 .into_iter()
6186 .flatten()
6187 .find(|rel| {
6188 crate::schema::relationship_keys(rel)
6189 .iter()
6190 .any(|found| found == key)
6191 })
6192 {
6193 return Some(std::borrow::Cow::Owned(
6194 crate::schema::relationship::RelationshipField::from_relationship_named(
6195 found,
6196 &found.foreign_table().name,
6197 Some(&entry.name),
6198 ),
6199 ));
6200 }
6201 }
6202 }
6203 }
6204
6205 resolution
6206 .cache
6207 .get_relationships(&self.qualified, &table.schema)?
6208 .iter()
6209 .find_map(|rel| {
6210 let given = crate::schema::relationship_keys(rel)
6211 .iter()
6212 .find_map(|key| self.names.relationship(&table.schema, &table.name, key));
6213 let field = RelationshipField::from_relationship_named(
6217 rel,
6218 &rel.foreign_table().name,
6219 given,
6220 );
6221 (field.name == name).then_some(std::borrow::Cow::Owned(field))
6222 })
6223 }
6224}
6225
6226fn column_comparison(op: &str) -> Option<&'static str> {
6235 Some(match op {
6236 "_ceq" => "=",
6237 "_cne" => "<>",
6238 "_cgt" => ">",
6239 "_cgte" => ">=",
6240 "_clt" => "<",
6241 "_clte" => "<=",
6242 _ => return None,
6243 })
6244}
6245
6246fn build_condition(
6248 value: &serde_json::Value,
6249 scope: &WhereScope<'_>,
6250 param_idx: &mut usize,
6251 values: &mut Vec<serde_json::Value>,
6252 alias_counter: &mut usize,
6253) -> Result<Option<String>, async_graphql::Error> {
6254 let serde_json::Value::Object(map) = value else {
6255 return Ok(None);
6256 };
6257
6258 let mut conditions: Vec<String> = Vec::new();
6259
6260 for (key, val) in map {
6261 match key.as_str() {
6262 "_and" | "_or" => {
6266 let members = val.as_array().ok_or_else(|| {
6267 async_graphql::Error::new(format!("\"{}\" takes a list of expressions", key))
6268 })?;
6269 let mut parts = Vec::new();
6270 for member in members {
6271 if let Some(sql) =
6272 build_condition(member, scope, param_idx, values, alias_counter)?
6273 {
6274 parts.push(sql);
6275 }
6276 }
6277 if parts.is_empty() {
6278 if key == "_or" {
6279 conditions.push("false".to_string());
6280 }
6281 continue;
6282 }
6283 let joiner = if key == "_and" { " AND " } else { " OR " };
6284 conditions.push(format!("({})", parts.join(joiner)));
6285 }
6286 "_not" => {
6287 if let Some(sql) = build_condition(val, scope, param_idx, values, alias_counter)? {
6288 conditions.push(format!("NOT ({})", sql));
6289 }
6290 }
6291 PERMISSION => {
6294 if let Some(sql) =
6295 build_condition(val, &scope.as_system(), param_idx, values, alias_counter)?
6296 {
6297 conditions.push(sql);
6298 }
6299 }
6300 "_exists" => {
6304 conditions.push(table_exists_sql(
6305 val,
6306 scope,
6307 param_idx,
6308 values,
6309 alias_counter,
6310 )?);
6311 }
6312 name if scope.relationship(name).is_some() => {
6318 let relationship = scope.relationship(name).expect("just checked");
6319 conditions.push(exists_sql(
6320 &relationship,
6321 val,
6322 scope,
6323 param_idx,
6324 values,
6325 alias_counter,
6326 )?);
6327 }
6328 name if scope.column_type(name).is_none()
6333 && name
6334 .strip_suffix("_aggregate")
6335 .and_then(|rel| scope.relationship(rel))
6336 .is_some_and(|rel| rel.is_list) =>
6337 {
6338 let relationship = name
6339 .strip_suffix("_aggregate")
6340 .and_then(|rel| scope.relationship(rel))
6341 .expect("just checked");
6342 conditions.push(aggregate_predicate_sql(
6343 &relationship,
6344 val,
6345 scope,
6346 param_idx,
6347 values,
6348 alias_counter,
6349 )?);
6350 }
6351 column => {
6352 let quoted = scope.column(column);
6353 match val {
6354 serde_json::Value::Object(ops) => {
6355 let column_type = scope.column_type(column);
6356 for (op, operand) in ops {
6357 if let Some(other) = column_comparison(op) {
6363 conditions.push(format!(
6364 "{} {} {}",
6365 quoted,
6366 other,
6367 scope.other_column(operand, op)?
6368 ));
6369 continue;
6370 }
6371 conditions.push(comparison_sql(
6372 "ed,
6373 column,
6374 column_type.as_deref(),
6375 op,
6376 operand,
6377 param_idx,
6378 values,
6379 )?);
6380 }
6381 }
6382 other => {
6386 conditions.push(format!("{} = ${}", quoted, param_idx));
6387 values.push(other.clone());
6388 *param_idx += 1;
6389 }
6390 }
6391 }
6392 }
6393 }
6394
6395 Ok(if conditions.is_empty() {
6396 None
6397 } else if conditions.len() == 1 {
6398 conditions.pop()
6399 } else {
6400 Some(conditions.join(" AND "))
6401 })
6402}
6403
6404fn exists_sql(
6411 relationship: &RelationshipField,
6412 child_expression: &serde_json::Value,
6413 scope: &WhereScope<'_>,
6414 param_idx: &mut usize,
6415 values: &mut Vec<serde_json::Value>,
6416 alias_counter: &mut usize,
6417) -> Result<String, async_graphql::Error> {
6418 let cache = scope.resolution.as_ref().map(|r| r.cache).ok_or_else(|| {
6419 async_graphql::Error::new(format!(
6420 "filtering on the relationship \"{}\" is not available here",
6421 relationship.name
6422 ))
6423 })?;
6424
6425 let plan = postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, cache)
6426 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
6427
6428 if plan.junction.is_some() || (plan.function.is_none() && plan.columns.is_empty()) {
6429 return Err(async_graphql::Error::new(format!(
6430 "filtering on \"{}\" is not supported: it is reached through a junction \
6431 rather than by a key",
6432 relationship.name
6433 )));
6434 }
6435
6436 *alias_counter += 1;
6437 let alias = format!("pgrst_rel_{}", alias_counter);
6438 let child_scope = WhereScope::aliased(
6439 &plan.foreign_schema,
6440 &plan.foreign_table,
6441 &alias,
6442 &relationship.target_type,
6443 scope,
6444 );
6445
6446 let (source, mut correlation) = match &plan.function {
6449 Some(function) => (
6450 format!(
6451 "{}.{}({})",
6452 postrust_sql::escape_ident(&function.schema),
6453 postrust_sql::escape_ident(&function.name),
6454 scope.row_ref
6455 ),
6456 Vec::new(),
6457 ),
6458 None => {
6459 let mut columns = Vec::with_capacity(plan.columns.len());
6460 for (parent_column, child_column) in &plan.columns {
6461 columns.push(format!(
6462 "{} = {}",
6463 scope.column(parent_column),
6464 child_scope.column(child_column)
6465 ));
6466 }
6467 (
6468 format!(
6469 "{}.{}",
6470 postrust_sql::escape_ident(&plan.foreign_schema),
6471 postrust_sql::escape_ident(&plan.foreign_table)
6472 ),
6473 columns,
6474 )
6475 }
6476 };
6477
6478 let child_condition = build_condition(
6479 child_expression,
6480 &child_scope,
6481 param_idx,
6482 values,
6483 alias_counter,
6484 )?;
6485 if let Some(sql) = child_condition {
6486 correlation.push(format!("({})", sql));
6487 }
6488
6489 if let Some(caller) = &child_scope.caller {
6494 let permission = permission_predicate(
6501 caller,
6502 child_scope.names,
6503 &plan.foreign_schema,
6504 &plan.foreign_table,
6505 )?;
6506 if let Some(predicate) = permission {
6507 if let Some(sql) =
6508 build_condition(&predicate, &child_scope, param_idx, values, alias_counter)?
6509 {
6510 correlation.push(format!("({})", sql));
6511 }
6512 }
6513 }
6514 if correlation.is_empty() {
6515 correlation.push("true".to_string());
6516 }
6517
6518 Ok(format!(
6519 "EXISTS (SELECT 1 FROM {} AS {} WHERE {})",
6520 source,
6521 postrust_sql::escape_ident(&alias),
6522 correlation.join(" AND ")
6523 ))
6524}
6525
6526fn table_exists_sql(
6551 spec: &serde_json::Value,
6552 scope: &WhereScope<'_>,
6553 param_idx: &mut usize,
6554 values: &mut Vec<serde_json::Value>,
6555 alias_counter: &mut usize,
6556) -> Result<String, async_graphql::Error> {
6557 let Some((schema, table)) = crate::role::exists_target(spec, &scope.qualified.schema) else {
6558 return Err(async_graphql::Error::new(
6559 "\"_exists\" needs a \"_table\" to look in",
6560 ));
6561 };
6562
6563 let qualified = postrust_core::api_request::QualifiedIdentifier::new(&schema, &table);
6567 if scope
6568 .resolution
6569 .as_ref()
6570 .is_some_and(|r| r.cache.get_table(&qualified).is_none())
6571 {
6572 return Err(async_graphql::Error::new(format!(
6573 "\"_exists\" names a table this server does not have: \"{}.{}\"",
6574 schema, table
6575 )));
6576 }
6577
6578 *alias_counter += 1;
6579 let alias = format!("pgrst_exists_{}", alias_counter);
6580 let type_name = scope
6584 .names
6585 .base_name(&schema, &table)
6586 .unwrap_or(&table)
6587 .to_string();
6588 let mut inner = WhereScope::aliased(&schema, &table, &alias, &type_name, scope);
6589 inner.caller = None;
6592
6593 let condition = match spec.get("_where") {
6594 Some(predicate) => build_condition(predicate, &inner, param_idx, values, alias_counter)?,
6595 None => None,
6596 };
6597
6598 Ok(format!(
6599 "EXISTS (SELECT 1 FROM {}.{} AS {} WHERE {})",
6600 postrust_sql::escape_ident(&schema),
6601 postrust_sql::escape_ident(&table),
6602 postrust_sql::escape_ident(&alias),
6603 condition.unwrap_or_else(|| "true".to_string())
6604 ))
6605}
6606
6607fn aggregate_predicate_sql(
6622 relationship: &RelationshipField,
6623 expression: &serde_json::Value,
6624 scope: &WhereScope<'_>,
6625 param_idx: &mut usize,
6626 values: &mut Vec<serde_json::Value>,
6627 alias_counter: &mut usize,
6628) -> Result<String, async_graphql::Error> {
6629 let serde_json::Value::Object(over) = expression else {
6630 return Ok("true".to_string());
6631 };
6632
6633 let cache = scope.resolution.as_ref().map(|r| r.cache).ok_or_else(|| {
6634 async_graphql::Error::new(format!(
6635 "filtering on the relationship \"{}\" is not available here",
6636 relationship.name
6637 ))
6638 })?;
6639 let plan = postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, cache)
6640 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
6641 if plan.junction.is_some() || (plan.function.is_none() && plan.columns.is_empty()) {
6642 return Err(async_graphql::Error::new(format!(
6643 "aggregating over \"{}\" is not supported: it is reached through a junction \
6644 rather than by a key",
6645 relationship.name
6646 )));
6647 }
6648
6649 let mut conditions: Vec<String> = Vec::new();
6650 for (function, spec) in over {
6651 if spec.is_null() {
6652 continue;
6653 }
6654 let serde_json::Value::Object(spec) = spec else {
6655 continue;
6656 };
6657 let Some(predicate) = spec.get("predicate").filter(|p| !p.is_null()) else {
6658 return Err(validation_error(format!(
6659 "\"{}\" over \"{}\" needs a predicate",
6660 function, relationship.name
6661 )));
6662 };
6663
6664 *alias_counter += 1;
6665 let alias = format!("pgrst_agg_{}", alias_counter);
6666 let child_scope = WhereScope::aliased(
6667 &plan.foreign_schema,
6668 &plan.foreign_table,
6669 &alias,
6670 &relationship.target_type,
6671 scope,
6672 );
6673
6674 let (source, mut correlation) = match &plan.function {
6677 Some(routine) => (
6678 format!(
6679 "{}.{}({})",
6680 postrust_sql::escape_ident(&routine.schema),
6681 postrust_sql::escape_ident(&routine.name),
6682 scope.row_ref
6683 ),
6684 Vec::new(),
6685 ),
6686 None => {
6687 let mut columns = Vec::with_capacity(plan.columns.len());
6688 for (parent_column, child_column) in &plan.columns {
6689 columns.push(format!(
6690 "{} = {}",
6691 scope.column(parent_column),
6692 child_scope.column(child_column)
6693 ));
6694 }
6695 (
6696 format!(
6697 "{}.{}",
6698 postrust_sql::escape_ident(&plan.foreign_schema),
6699 postrust_sql::escape_ident(&plan.foreign_table)
6700 ),
6701 columns,
6702 )
6703 }
6704 };
6705 if let Some(filter) = spec.get("filter").filter(|f| !f.is_null()) {
6708 if let Some(sql) =
6709 build_condition(filter, &child_scope, param_idx, values, alias_counter)?
6710 {
6711 correlation.push(format!("({})", sql));
6712 }
6713 }
6714 if correlation.is_empty() {
6715 correlation.push("true".to_string());
6716 }
6717
6718 let arguments = spec.get("arguments").filter(|a| !a.is_null());
6719 let distinct = matches!(spec.get("distinct"), Some(serde_json::Value::Bool(true)));
6720 let (aggregate, operand_type) = match function.as_str() {
6721 "count" => {
6722 let columns: Vec<String> = match arguments {
6723 Some(serde_json::Value::Array(items)) => items
6724 .iter()
6725 .filter_map(|item| item.as_str())
6726 .map(|column| child_scope.column(column))
6727 .collect(),
6728 Some(serde_json::Value::String(one)) => vec![child_scope.column(one)],
6729 _ => Vec::new(),
6730 };
6731 let counted = match columns.len() {
6732 0 => "*".to_string(),
6733 1 => columns.into_iter().next().unwrap_or_default(),
6734 _ => format!("({})", columns.join(", ")),
6735 };
6736 let counted = match distinct && counted != "*" {
6737 true => format!("DISTINCT {}", counted),
6738 false => counted,
6739 };
6740 (format!("count({})", counted), "integer")
6741 }
6742 "bool_and" | "bool_or" => {
6743 let Some(column) = arguments.and_then(|a| a.as_str()) else {
6744 return Err(validation_error(format!(
6745 "\"{}\" over \"{}\" needs the column to fold",
6746 function, relationship.name
6747 )));
6748 };
6749 (
6750 format!("{}({})", function, child_scope.column(column)),
6751 "boolean",
6752 )
6753 }
6754 other => {
6755 return Err(validation_error(format!(
6756 "\"{}\" is not an aggregate a predicate can be written over",
6757 other
6758 )))
6759 }
6760 };
6761
6762 let subselect = format!(
6763 "(SELECT {} FROM {} AS {} WHERE {})",
6764 aggregate,
6765 source,
6766 postrust_sql::escape_ident(&alias),
6767 correlation.join(" AND ")
6768 );
6769 let serde_json::Value::Object(operators) = predicate else {
6770 return Err(validation_error(format!(
6771 "the predicate on \"{}\" is not a comparison",
6772 function
6773 )));
6774 };
6775 for (operator, operand) in operators {
6776 conditions.push(comparison_sql(
6777 &subselect,
6778 function,
6779 Some(operand_type),
6780 operator,
6781 operand,
6782 param_idx,
6783 values,
6784 )?);
6785 }
6786 }
6787
6788 Ok(match conditions.is_empty() {
6789 true => "true".to_string(),
6790 false => format!("({})", conditions.join(" AND ")),
6791 })
6792}
6793
6794#[allow(clippy::too_many_arguments)]
6802fn postgis_sql(
6803 quoted: &str,
6804 column_type: Option<&str>,
6805 function: &str,
6806 op: &str,
6807 operand: &serde_json::Value,
6808 param_idx: &mut usize,
6809 values: &mut Vec<serde_json::Value>,
6810) -> Result<String, async_graphql::Error> {
6811 let is_geography = column_type == Some("geography");
6812 let mut shape = |value: &serde_json::Value, param_idx: &mut usize| -> String {
6813 let placeholder = format!("${}", param_idx);
6814 *param_idx += 1;
6815 values.push(value.clone());
6816 match is_geography {
6817 true => format!("ST_GeomFromGeoJSON({})::geography", placeholder),
6818 false => format!("ST_GeomFromGeoJSON({})", placeholder),
6819 }
6820 };
6821
6822 match op {
6823 "_st_d_within" | "_st_3d_d_within" => {
6825 let serde_json::Value::Object(spec) = operand else {
6826 return Err(async_graphql::Error::new(format!(
6827 "\"{}\" takes {{distance, from}}",
6828 op
6829 )));
6830 };
6831 let Some(from) = spec.get("from") else {
6832 return Err(async_graphql::Error::new(format!(
6833 "\"{}\" needs a shape to measure from",
6834 op
6835 )));
6836 };
6837 let Some(distance) = spec.get("distance") else {
6838 return Err(async_graphql::Error::new(format!(
6839 "\"{}\" needs a distance",
6840 op
6841 )));
6842 };
6843 let from_sql = shape(from, param_idx);
6844 let distance_sql = format!("${}", param_idx);
6845 *param_idx += 1;
6846 values.push(distance.clone());
6847 let spheroid = match (is_geography, spec.get("use_spheroid")) {
6852 (true, Some(use_spheroid)) if !use_spheroid.is_null() => {
6853 let placeholder = format!("${}", param_idx);
6854 *param_idx += 1;
6855 values.push(use_spheroid.clone());
6856 format!(", {}::boolean", placeholder)
6857 }
6858 _ => String::new(),
6859 };
6860 Ok(format!(
6861 "{}({}, {}, {}::float8{})",
6862 function, quoted, from_sql, distance_sql, spheroid
6863 ))
6864 }
6865 "_st_intersects_rast" => {
6867 let placeholder = format!("${}", param_idx);
6868 *param_idx += 1;
6869 values.push(operand.clone());
6870 Ok(format!("{}({}, {}::raster)", function, quoted, placeholder))
6871 }
6872 "_st_intersects_geom_nband" | "_st_intersects_nband_geom" => {
6874 let serde_json::Value::Object(spec) = operand else {
6875 return Err(async_graphql::Error::new(format!(
6876 "\"{}\" takes an object",
6877 op
6878 )));
6879 };
6880 let Some(geometry) = spec.get("geommin") else {
6881 return Err(async_graphql::Error::new(format!(
6882 "\"{}\" needs a shape",
6883 op
6884 )));
6885 };
6886 let band = spec.get("nband").filter(|v| !v.is_null());
6887 let geometry_sql = shape(geometry, param_idx);
6888 Ok(match band {
6889 None => format!("{}({}, {})", function, quoted, geometry_sql),
6890 Some(band) => {
6891 let placeholder = format!("${}", param_idx);
6892 *param_idx += 1;
6893 values.push(band.clone());
6894 format!(
6898 "{}({}, {}::int, {})",
6899 function, quoted, placeholder, geometry_sql
6900 )
6901 }
6902 })
6903 }
6904 _ => {
6906 let other = shape(operand, param_idx);
6907 Ok(format!("{}({}, {})", function, quoted, other))
6908 }
6909 }
6910}
6911
6912fn sql_array(
6918 items: &[serde_json::Value],
6919 cast: &str,
6920 param_idx: &mut usize,
6921 values: &mut Vec<serde_json::Value>,
6922) -> String {
6923 if items.is_empty() {
6924 return format!("ARRAY[]::{}", cast);
6925 }
6926 let placeholders: Vec<String> = items
6927 .iter()
6928 .map(|item| {
6929 let placeholder = format!("${}", param_idx);
6930 *param_idx += 1;
6931 values.push(item.clone());
6932 placeholder
6933 })
6934 .collect();
6935 format!("ARRAY[{}]::{}", placeholders.join(", "), cast)
6936}
6937
6938#[allow(clippy::too_many_arguments)]
6940fn comparison_sql(
6941 quoted: &str,
6942 column: &str,
6943 column_type: Option<&str>,
6944 op: &str,
6945 operand: &serde_json::Value,
6946 param_idx: &mut usize,
6947 values: &mut Vec<serde_json::Value>,
6948) -> Result<String, async_graphql::Error> {
6949 if op == "_cast" {
6955 let serde_json::Value::Object(casts) = operand else {
6956 return Err(async_graphql::Error::new(
6957 "_cast takes an object naming the type to compare as",
6958 ));
6959 };
6960 let mut conditions = Vec::new();
6961 for (target, comparisons) in casts {
6962 let pg_type = match target.as_str() {
6966 "geometry" | "geography" => target.as_str(),
6967 "String" => "text",
6968 _ => {
6969 return Err(async_graphql::Error::new(format!(
6970 "cannot compare \"{}\" as \"{}\"",
6971 column, target
6972 )))
6973 }
6974 };
6975 let serde_json::Value::Object(ops) = comparisons else {
6976 continue;
6977 };
6978 let cast = format!("{}::{}", quoted, pg_type);
6979 for (nested_op, nested_operand) in ops {
6980 conditions.push(comparison_sql(
6981 &cast,
6982 column,
6983 Some(pg_type),
6984 nested_op,
6985 nested_operand,
6986 param_idx,
6987 values,
6988 )?);
6989 }
6990 }
6991 return Ok(match conditions.len() {
6992 0 => "true".to_string(),
6993 1 => conditions.pop().expect("just counted"),
6994 _ => format!("({})", conditions.join(" AND ")),
6995 });
6996 }
6997
6998 if let Some(function) = crate::input::bool_exp::postgis_function(op) {
7001 return postgis_sql(
7002 quoted,
7003 column_type,
7004 function,
7005 op,
7006 operand,
7007 param_idx,
7008 values,
7009 );
7010 }
7011
7012 if let Some((operator, cast)) = crate::input::bool_exp::ltree_operator(op) {
7016 if cast.ends_with("[]") {
7017 let items = operand.as_array().ok_or_else(|| {
7018 async_graphql::Error::new(format!(
7019 "the \"{}\" comparison on \"{}\" takes a list",
7020 op, column
7021 ))
7022 })?;
7023 return Ok(format!(
7024 "{} {} {}",
7025 quoted,
7026 operator,
7027 sql_array(items, cast, param_idx, values)
7028 ));
7029 }
7030 let placeholder = format!("${}", param_idx);
7031 *param_idx += 1;
7032 values.push(operand.clone());
7033 return Ok(format!("{} {} {}::{}", quoted, operator, placeholder, cast));
7034 }
7035
7036 if let Some(function) = match op {
7041 "_jsonb_path_exists" => Some("jsonb_path_exists"),
7042 "_jsonb_path_match" => Some("jsonb_path_match"),
7043 _ => None,
7044 } {
7045 let placeholder = format!("${}", param_idx);
7046 *param_idx += 1;
7047 values.push(operand.clone());
7048 return Ok(format!(
7049 "{}({}, {}::jsonpath)",
7050 function, quoted, placeholder
7051 ));
7052 }
7053
7054 let binary = match op {
7056 "_eq" => Some("="),
7057 "_neq" => Some("<>"),
7058 "_gt" => Some(">"),
7059 "_gte" => Some(">="),
7060 "_lt" => Some("<"),
7061 "_lte" => Some("<="),
7062 "_like" => Some("LIKE"),
7063 "_nlike" => Some("NOT LIKE"),
7064 "_ilike" => Some("ILIKE"),
7065 "_nilike" => Some("NOT ILIKE"),
7066 "_similar" => Some("SIMILAR TO"),
7067 "_nsimilar" => Some("NOT SIMILAR TO"),
7068 "_regex" => Some("~"),
7069 "_nregex" => Some("!~"),
7070 "_iregex" => Some("~*"),
7071 "_niregex" => Some("!~*"),
7072 "_contains" => Some("@>"),
7073 "_contained_in" => Some("<@"),
7074 "_has_key" => Some("?"),
7075 _ => None,
7076 };
7077
7078 if let Some(sql_op) = binary {
7079 let placeholder = format!("${}", param_idx);
7080 *param_idx += 1;
7081 values.push(match op {
7086 "_contains" | "_contained_in" => match operand {
7087 serde_json::Value::String(_)
7088 | serde_json::Value::Number(_)
7089 | serde_json::Value::Bool(_) => serde_json::Value::String(operand.to_string()),
7090 other => other.clone(),
7091 },
7092 _ => operand.clone(),
7093 });
7094 let cast = match op {
7106 "_contains" | "_contained_in" => "::jsonb".to_string(),
7107 "_has_key" => "::text".to_string(),
7108 _ if operand.is_null() => match column_type {
7117 Some(pg_type) => format!("::{}", pg_type),
7118 None => String::new(),
7119 },
7120 _ if operand.is_string() => match column_type {
7131 Some(pg_type) => format!("::{}", pg_type),
7132 None => String::new(),
7133 },
7134 _ => String::new(),
7135 };
7136 return Ok(format!("{} {} {}{}", quoted, sql_op, placeholder, cast));
7137 }
7138
7139 match op {
7140 "_is_null" => Ok(if operand.as_bool().unwrap_or(false) {
7141 format!("{} IS NULL", quoted)
7142 } else {
7143 format!("{} IS NOT NULL", quoted)
7144 }),
7145 "_in" | "_nin" => {
7146 let items = operand.as_array().ok_or_else(|| {
7147 async_graphql::Error::new(format!(
7148 "the \"{}\" comparison on \"{}\" requires a list of values",
7149 op, column
7150 ))
7151 })?;
7152 if items.is_empty() {
7153 return Ok(if op == "_in" { "false" } else { "true" }.to_string());
7156 }
7157 let mut placeholders = Vec::with_capacity(items.len());
7158 for item in items {
7159 let cast = match (item.is_null() || item.is_string(), column_type) {
7162 (true, Some(pg_type)) => format!("::{}", pg_type),
7163 _ => String::new(),
7164 };
7165 placeholders.push(format!("${}{}", param_idx, cast));
7166 values.push(item.clone());
7167 *param_idx += 1;
7168 }
7169 Ok(format!(
7170 "{} {} ({})",
7171 quoted,
7172 if op == "_in" { "IN" } else { "NOT IN" },
7173 placeholders.join(", ")
7174 ))
7175 }
7176 "_has_keys_any" | "_has_keys_all" => {
7177 let items = operand.as_array().ok_or_else(|| {
7178 async_graphql::Error::new(format!(
7179 "the \"{}\" comparison on \"{}\" requires a list of keys",
7180 op, column
7181 ))
7182 })?;
7183 Ok(format!(
7184 "{} {} {}",
7185 quoted,
7186 if op == "_has_keys_any" { "?|" } else { "?&" },
7187 sql_array(items, "text[]", param_idx, values)
7188 ))
7189 }
7190 other => Err(async_graphql::Error::new(format!(
7191 "unsupported comparison \"{}\" on \"{}\"",
7192 other, column
7193 ))),
7194 }
7195}
7196
7197#[allow(clippy::too_many_arguments)] async fn build_order_by_clause(
7207 ctx: &ResolverContext<'_>,
7208 schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
7209 schema_name: &str,
7210 table_name: &str,
7211 type_name: &str,
7212 relationships: &HashMap<String, Vec<RelationshipField>>,
7213 reference: &str,
7214 names: &crate::names::NameOverrides,
7215) -> Result<String, async_graphql::Error> {
7216 let Ok(order_arg) = ctx.args.try_get("order_by") else {
7217 return Ok(String::new());
7218 };
7219
7220 let value = accessor_to_json(&order_arg);
7221 let entries: Vec<&serde_json::Value> = match &value {
7224 serde_json::Value::Array(items) => items.iter().collect(),
7225 serde_json::Value::Object(_) => vec![&value],
7226 serde_json::Value::Null => return Ok(String::new()),
7227 _ => {
7228 return Err(async_graphql::Error::new(
7229 "order_by takes an object or a list of them",
7230 ))
7231 }
7232 };
7233 if entries.is_empty() {
7234 return Ok(String::new());
7235 }
7236
7237 let guard = schema_cache
7238 .get()
7239 .await
7240 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7241 let cache = guard
7242 .as_ref()
7243 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
7244 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
7245 let table = cache
7246 .get_table(&qi)
7247 .ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
7248
7249 let mut terms = Vec::new();
7250 let mut alias_counter = 0usize;
7251 for entry in entries {
7252 order_terms_into(
7253 entry,
7254 cache,
7255 relationships,
7256 type_name,
7257 table,
7258 reference,
7259 names,
7260 &mut alias_counter,
7261 &mut terms,
7262 )?;
7263 }
7264
7265 if terms.is_empty() {
7266 return Ok(String::new());
7267 }
7268 Ok(format!(" ORDER BY {}", terms.join(", ")))
7269}
7270
7271#[allow(clippy::too_many_arguments)]
7279fn order_terms_into(
7280 entry: &serde_json::Value,
7281 cache: &SchemaCache,
7282 relationships: &HashMap<String, Vec<RelationshipField>>,
7283 type_name: &str,
7284 table: &postrust_core::schema_cache::Table,
7285 reference: &str,
7286 names: &crate::names::NameOverrides,
7287 alias_counter: &mut usize,
7288 terms: &mut Vec<String>,
7289) -> Result<(), async_graphql::Error> {
7290 let serde_json::Value::Object(map) = entry else {
7291 return Err(async_graphql::Error::new(
7292 "each order_by entry is an object mapping a column to a direction",
7293 ));
7294 };
7295 let available: &[RelationshipField] = relationships
7296 .get(type_name)
7297 .map(|r| r.as_slice())
7298 .unwrap_or(&[]);
7299
7300 for (key, value) in map {
7301 if value.is_null() {
7307 continue;
7308 }
7309
7310 if let Some(name) = value.as_str() {
7312 let sql = crate::input::order_by::direction_sql(name).ok_or_else(|| {
7313 async_graphql::Error::new(format!(
7314 "\"{}\" is not a sort direction; expected one of asc, desc, \
7315 asc_nulls_first, asc_nulls_last, desc_nulls_first, desc_nulls_last",
7316 name
7317 ))
7318 })?;
7319 let column = table_column_for(names, table, key);
7320 if table.get_column(column).is_some() {
7321 terms.push(format!(
7322 "{}.{} {}",
7323 reference,
7324 postrust_sql::escape_ident(column),
7325 sql
7326 ));
7327 continue;
7328 }
7329 let function = names
7332 .computed_source(&table.schema, &table.name, key)
7333 .unwrap_or(key);
7334 if let Some(definition) = table.get_computed_column(function) {
7335 terms.push(format!(
7336 "{} {}",
7337 computed_call(definition, &format!("{}.*", reference), reference),
7338 sql
7339 ));
7340 continue;
7341 }
7342 return Err(async_graphql::Error::new(format!(
7343 "cannot order by unknown column \"{}\" on \"{}\"",
7344 key, table.name
7345 )));
7346 }
7347
7348 if let Some(rel) = key
7350 .strip_suffix("_aggregate")
7351 .and_then(|name| available.iter().find(|r| r.name == name && r.is_list))
7352 {
7353 aggregate_order_terms(value, rel, cache, reference, alias_counter, terms)?;
7354 continue;
7355 }
7356
7357 if let Some(rel) = available.iter().find(|r| r.name == *key && !r.is_list) {
7359 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, cache)
7360 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7361 if plan.columns.is_empty() {
7362 return Err(async_graphql::Error::new(format!(
7363 "cannot order by \"{}\": it is not reached by a key",
7364 key
7365 )));
7366 }
7367 *alias_counter += 1;
7368 let alias = format!("pgrst_ord_{}", alias_counter);
7369 let quoted_alias = postrust_sql::escape_ident(&alias);
7370 let correlation = plan
7371 .columns
7372 .iter()
7373 .map(|(local, foreign)| {
7374 format!(
7375 "{} = {}.{}",
7376 format_args!("{}.{}", reference, postrust_sql::escape_ident(local)),
7377 quoted_alias,
7378 postrust_sql::escape_ident(foreign)
7379 )
7380 })
7381 .collect::<Vec<_>>()
7382 .join(" AND ");
7383
7384 let target_qi = postrust_core::api_request::QualifiedIdentifier::new(
7385 &plan.foreign_schema,
7386 &plan.foreign_table,
7387 );
7388 let Some(target) = cache.get_table(&target_qi) else {
7389 return Err(async_graphql::Error::new(format!(
7390 "unknown table \"{}\"",
7391 plan.foreign_table
7392 )));
7393 };
7394
7395 let mut nested = Vec::new();
7398 order_terms_into(
7399 value,
7400 cache,
7401 relationships,
7402 &rel.target_type,
7403 target,
7404 "ed_alias,
7405 names,
7406 alias_counter,
7407 &mut nested,
7408 )?;
7409 for term in nested {
7410 let (expression, direction) = term
7411 .rsplit_once(' ')
7412 .map(|(e, d)| (e.to_string(), d.to_string()))
7413 .unwrap_or((term.clone(), String::new()));
7414 terms.push(format!(
7415 "(SELECT {} FROM {}.{} AS {} WHERE {}) {}",
7416 expression,
7417 postrust_sql::escape_ident(&plan.foreign_schema),
7418 postrust_sql::escape_ident(&plan.foreign_table),
7419 quoted_alias,
7420 correlation,
7421 direction
7422 ));
7423 }
7424 continue;
7425 }
7426
7427 return Err(async_graphql::Error::new(format!(
7428 "cannot order by \"{}\" on \"{}\"",
7429 key, table.name
7430 )));
7431 }
7432 Ok(())
7433}
7434
7435fn aggregate_order_terms(
7437 value: &serde_json::Value,
7438 rel: &RelationshipField,
7439 cache: &SchemaCache,
7440 reference: &str,
7441 alias_counter: &mut usize,
7442 terms: &mut Vec<String>,
7443) -> Result<(), async_graphql::Error> {
7444 let serde_json::Value::Object(spec) = value else {
7445 return Err(async_graphql::Error::new(
7446 "ordering by an aggregate takes an object, such as {count: desc}",
7447 ));
7448 };
7449 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, cache)
7450 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7451 if plan.columns.is_empty() && plan.function.is_none() {
7452 return Err(async_graphql::Error::new(format!(
7453 "cannot order by an aggregate of \"{}\": it is not reached by a key",
7454 rel.name
7455 )));
7456 }
7457 if plan.function.is_some() && !rel.arguments.is_empty() {
7460 return Err(async_graphql::Error::new(format!(
7461 "cannot order by an aggregate of \"{}\": it takes arguments, and an \
7462 ordering has nowhere to write them",
7463 rel.name
7464 )));
7465 }
7466
7467 *alias_counter += 1;
7468 let alias = postrust_sql::escape_ident(&format!("pgrst_ord_{}", alias_counter));
7469 let (source, correlation) = match &plan.function {
7472 Some(function) => {
7473 let session = plan.row_argument.as_ref().and_then(|row_argument| {
7474 computed_session_argument(cache, function, row_argument)
7475 .map(|session| (row_argument.clone(), session))
7476 });
7477 let call = match session {
7478 Some((row_argument, session)) => format!(
7479 "{} => {}, {} => {}",
7480 postrust_sql::escape_ident(&row_argument),
7481 reference,
7482 postrust_sql::escape_ident(&session),
7483 SESSION_ARGUMENT
7484 ),
7485 None => reference.to_string(),
7486 };
7487 (
7488 format!(
7489 "{}.{}({})",
7490 postrust_sql::escape_ident(&function.schema),
7491 postrust_sql::escape_ident(&function.name),
7492 call
7493 ),
7494 "true".to_string(),
7495 )
7496 }
7497 None => (
7498 format!(
7499 "{}.{}",
7500 postrust_sql::escape_ident(&plan.foreign_schema),
7501 postrust_sql::escape_ident(&plan.foreign_table)
7502 ),
7503 plan.columns
7504 .iter()
7505 .map(|(local, foreign)| {
7506 format!(
7507 "{}.{} = {}.{}",
7508 reference,
7509 postrust_sql::escape_ident(local),
7510 alias,
7511 postrust_sql::escape_ident(foreign)
7512 )
7513 })
7514 .collect::<Vec<_>>()
7515 .join(" AND "),
7516 ),
7517 };
7518
7519 let mut wanted: Vec<(String, String)> = Vec::new();
7521 for (function, argument) in spec {
7522 if argument.is_null() {
7525 continue;
7526 }
7527 if function == "count" {
7528 let name = argument.as_str().unwrap_or_default();
7529 let Some(direction) = crate::input::order_by::direction_sql(name) else {
7530 return Err(async_graphql::Error::new(format!(
7531 "\"{}\" is not a sort direction",
7532 name
7533 )));
7534 };
7535 wanted.push(("count(*)".to_string(), direction.to_string()));
7536 continue;
7537 }
7538 let serde_json::Value::Object(columns) = argument else {
7539 continue;
7540 };
7541 for (column, direction) in columns {
7542 if direction.is_null() {
7543 continue;
7544 }
7545 let name = direction.as_str().unwrap_or_default();
7546 let Some(direction) = crate::input::order_by::direction_sql(name) else {
7547 return Err(async_graphql::Error::new(format!(
7548 "\"{}\" is not a sort direction",
7549 name
7550 )));
7551 };
7552 wanted.push((
7555 format!(
7556 "{}({}.{})",
7557 function,
7558 alias,
7559 postrust_sql::escape_ident(column)
7560 ),
7561 direction.to_string(),
7562 ));
7563 }
7564 }
7565
7566 for (expression, direction) in wanted {
7567 terms.push(format!(
7568 "(SELECT {} FROM {} AS {} WHERE {}) {}",
7569 expression, source, alias, correlation, direction
7570 ));
7571 }
7572 Ok(())
7573}
7574
7575fn distinct_on_clause(
7589 distinct_on: &[String],
7590 written: Option<&str>,
7591) -> Result<(String, Option<String>), async_graphql::Error> {
7592 if distinct_on.is_empty() {
7593 return Ok((String::new(), written.map(str::to_string)));
7594 }
7595 let terms: Vec<&str> = written
7596 .map(|written| written.split(", ").collect())
7597 .unwrap_or_default();
7598 let column_of = |term: &str| {
7601 let expression = term.split_whitespace().next().unwrap_or(term);
7602 expression
7603 .rsplit_once('.')
7604 .map(|(_, name)| name)
7605 .unwrap_or(expression)
7606 .to_string()
7607 };
7608 let leading: Vec<String> = terms
7609 .iter()
7610 .take(distinct_on.len())
7611 .map(|term| column_of(term))
7612 .collect();
7613 if !terms.is_empty()
7614 && (leading.len() != distinct_on.len()
7615 || distinct_on
7616 .iter()
7617 .any(|column| !leading.contains(&column_of(column))))
7618 {
7619 return Err(validation_error(
7620 "\"distinct_on\" columns must match initial \"order_by\" columns",
7621 ));
7622 }
7623 let order = match terms.is_empty() {
7624 true => Some(distinct_on.join(", ")),
7625 false => written.map(str::to_string),
7626 };
7627 Ok((format!("DISTINCT ON ({}) ", distinct_on.join(", ")), order))
7628}
7629
7630async fn build_distinct_on(
7638 ctx: &ResolverContext<'_>,
7639 schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
7640 schema_name: &str,
7641 table_name: &str,
7642 names: &crate::names::NameOverrides,
7643) -> Result<Vec<String>, async_graphql::Error> {
7644 let Ok(arg) = ctx.args.try_get("distinct_on") else {
7645 return Ok(Vec::new());
7646 };
7647 let value = accessor_to_json(&arg);
7648 let requested: Vec<String> = match &value {
7649 serde_json::Value::Array(items) => items
7650 .iter()
7651 .filter_map(|i| i.as_str().map(|s| s.to_string()))
7652 .collect(),
7653 serde_json::Value::String(one) => vec![one.clone()],
7654 _ => Vec::new(),
7655 };
7656 if requested.is_empty() {
7657 return Ok(Vec::new());
7658 }
7659
7660 let guard = schema_cache
7661 .get()
7662 .await
7663 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7664 let cache = guard
7665 .as_ref()
7666 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
7667 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
7668 let table = cache
7669 .get_table(&qi)
7670 .ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
7671
7672 let mut quoted = Vec::with_capacity(requested.len());
7673 for field in requested {
7674 let name = table_column_for(names, table, &field).to_string();
7675 if table.get_column(&name).is_none() {
7676 return Err(async_graphql::Error::new(format!(
7677 "cannot take distinct on unknown column \"{}\" of \"{}\"",
7678 field, table_name
7679 )));
7680 }
7681 quoted.push(postrust_sql::escape_ident(&name));
7682 }
7683 Ok(quoted)
7684}
7685
7686#[allow(clippy::too_many_arguments)] fn computed_projections(
7700 table: &postrust_core::schema_cache::Table,
7701 selection: async_graphql::SelectionField<'_>,
7702 row_reference: &str,
7703 names: &crate::names::NameOverrides,
7704 schema_cache: &SchemaCache,
7705 param_idx: &mut usize,
7706 values: &mut Vec<serde_json::Value>,
7707) -> Result<Vec<String>, async_graphql::Error> {
7708 let mut projections = Vec::new();
7709 for field in selection.selection_set() {
7710 let name = field.name();
7711 if table
7714 .get_column(table_column_for(names, table, name))
7715 .is_some()
7716 {
7717 continue;
7718 }
7719 let function = names
7722 .computed_source(&table.schema, &table.name, name)
7723 .unwrap_or(name);
7724 let Some(computed) = table.get_computed_column(function) else {
7725 continue;
7726 };
7727 let arguments = crate::schema::computed_caller_arguments(computed, schema_cache);
7728 let given = match arguments.is_empty() {
7729 true => None,
7730 false => embed_arguments(field).remove("args"),
7731 };
7732 projections.push(format!(
7733 "{} AS {}",
7734 computed_column_call(
7735 computed,
7736 row_reference,
7737 name,
7738 &arguments,
7739 given.as_ref(),
7740 param_idx,
7741 values,
7742 )?,
7743 postrust_sql::escape_ident(name)
7744 ));
7745 }
7746 Ok(projections)
7747}
7748
7749fn computed_column_call(
7758 computed: &postrust_core::schema_cache::ComputedColumn,
7759 row: &str,
7760 field_name: &str,
7761 arguments: &[(String, String, bool)],
7762 given: Option<&serde_json::Value>,
7763 param_idx: &mut usize,
7764 values: &mut Vec<serde_json::Value>,
7765) -> Result<String, async_graphql::Error> {
7766 if arguments.is_empty() {
7767 return Ok(computed_call(computed, row, row));
7768 }
7769 let Some(row_argument) = &computed.row_argument else {
7770 return Ok(computed_call(computed, row, row));
7771 };
7772 let function = format!(
7773 "{}.{}",
7774 postrust_sql::escape_ident(&computed.function.schema),
7775 postrust_sql::escape_ident(&computed.function.name)
7776 );
7777 let mut passed = vec![format!(
7778 "{} => {}",
7779 postrust_sql::escape_ident(row_argument),
7780 row
7781 )];
7782 if let Some(session) = &computed.session_argument {
7783 passed.push(format!(
7784 "{} => {}",
7785 postrust_sql::escape_ident(session),
7786 SESSION_ARGUMENT
7787 ));
7788 }
7789 for (name, pg_type, required) in arguments {
7790 match given.and_then(|given| given.get(name)) {
7791 None if *required => {
7792 return Err(async_graphql::Error::new(format!(
7793 "{} needs the argument \"{}\"",
7794 field_name, name
7795 )))
7796 }
7797 None => continue,
7799 Some(value) => {
7800 passed.push(format!(
7801 "{} => ${}::{}",
7802 postrust_sql::escape_ident(name),
7803 param_idx,
7804 pg_type
7805 ));
7806 *param_idx += 1;
7807 values.push(value.clone());
7808 }
7809 }
7810 }
7811 Ok(format!("{}({})", function, passed.join(", ")))
7812}
7813
7814fn count_expression(
7826 field: async_graphql::SelectionField<'_>,
7827 table: (&str, &str),
7828 alias: Option<&str>,
7829 names: &crate::names::NameOverrides,
7830) -> String {
7831 let arguments = embed_arguments(field);
7832 let requested = match arguments.get("columns") {
7833 Some(serde_json::Value::Array(items)) => items
7834 .iter()
7835 .filter_map(|item| item.as_str())
7836 .collect::<Vec<_>>(),
7837 Some(serde_json::Value::String(one)) => vec![one.as_str()],
7838 _ => Vec::new(),
7839 };
7840 if requested.is_empty() {
7841 return "count(*)".to_string();
7842 }
7843 let columns: Vec<String> = requested
7844 .into_iter()
7845 .map(|column| {
7846 let (schema, name) = table;
7847 let source = column_for(names, schema, name, column);
7848 match alias {
7849 Some(alias) => format!(
7850 "{}.{}",
7851 postrust_sql::escape_ident(alias),
7852 postrust_sql::escape_ident(source)
7853 ),
7854 None => postrust_sql::escape_ident(source),
7855 }
7856 })
7857 .collect();
7858 let counted = match columns.len() {
7861 1 => columns.into_iter().next().unwrap_or_default(),
7862 _ => format!("({})", columns.join(", ")),
7863 };
7864 let distinct = matches!(
7865 arguments.get("distinct"),
7866 Some(serde_json::Value::Bool(true))
7867 );
7868 match distinct {
7869 true => format!("count(DISTINCT {})", counted),
7870 false => format!("count({})", counted),
7871 }
7872}
7873
7874#[allow(clippy::too_many_arguments)] fn nested_aggregate_select(
7882 caller: &crate::role::Caller<'_>,
7883 selection: async_graphql::SelectionField<'_>,
7884 child_alias: &str,
7885 schema_cache: &SchemaCache,
7886 relationships: &HashMap<String, Vec<RelationshipField>>,
7887 target_type: &str,
7888 child_table: Option<&postrust_core::schema_cache::Table>,
7889 max_rows: Option<i64>,
7890 alias_counter: &mut usize,
7891 param_idx: &mut usize,
7892 values: &mut Vec<serde_json::Value>,
7893 names: &crate::names::NameOverrides,
7894) -> Result<String, async_graphql::Error> {
7895 use crate::schema::aggregate as agg;
7896
7897 let mut parts: Vec<String> = Vec::new();
7898
7899 for child in selection.selection_set() {
7900 match child.name() {
7901 "aggregate" => {
7902 let of_table = child_table
7903 .map(|table| (table.schema.as_str(), table.name.as_str()))
7904 .unwrap_or(("", ""));
7905 let mut counts: Vec<(String, String)> = Vec::new();
7910 let mut wanted: BTreeMap<&str, Vec<String>> = BTreeMap::new();
7911 for function in child.selection_set() {
7912 if function.name() == "count" {
7913 counts.push((
7914 function.alias().unwrap_or("count").to_string(),
7915 count_expression(function, of_table, Some(child_alias), names),
7916 ));
7917 continue;
7918 }
7919 let sql_function = agg::NUMERIC_AGGREGATES
7920 .iter()
7921 .chain(agg::ORDERED_AGGREGATES.iter())
7922 .find(|(name, _)| *name == function.name())
7923 .map(|(name, _)| *name);
7924 let Some(sql_function) = sql_function else {
7925 continue;
7926 };
7927 let columns = wanted.entry(sql_function).or_default();
7928 for column in function.selection_set() {
7929 let name = column.name().to_string();
7930 if !columns.contains(&name) {
7931 columns.push(name);
7932 }
7933 }
7934 }
7935 let mut fields = match counts.iter().any(|(key, _)| key == "count") {
7938 true => Vec::new(),
7939 false => vec!["'count', count(*)".to_string()],
7940 };
7941 for (key, expression) in &counts {
7942 fields.push(format!("'{}', {}", key.replace('\'', "''"), expression));
7943 }
7944 for (sql_function, named) in &wanted {
7945 let columns: Vec<String> = named
7946 .iter()
7947 .map(|column| {
7948 let source = child_table
7949 .map(|table| table_column_for(names, table, column))
7950 .unwrap_or(column.as_str());
7951 format!(
7952 "'{}', {}({}.{})",
7953 column.replace('\'', "''"),
7954 sql_function,
7955 postrust_sql::escape_ident(child_alias),
7956 postrust_sql::escape_ident(source)
7957 )
7958 })
7959 .collect();
7960 if !columns.is_empty() {
7961 fields.push(format!(
7962 "'{}', json_build_object({})",
7963 sql_function,
7964 columns.join(", ")
7965 ));
7966 }
7967 }
7968 parts.push(format!(
7969 "json_build_object({}) AS {}",
7970 fields.join(", "),
7971 postrust_sql::escape_ident("aggregate")
7972 ));
7973 }
7974 "nodes" => {
7979 let embeds = build_embed_expressions(
7980 caller,
7981 schema_cache,
7982 relationships,
7983 target_type,
7984 child_alias,
7985 child,
7986 max_rows,
7987 alias_counter,
7988 param_idx,
7989 values,
7990 names,
7991 )?;
7992 let mut columns: Vec<String> = Vec::new();
7993 for column in child.selection_set() {
7994 let name = column.name();
7995 if name == "__typename" || embeds.iter().any(|(field, _)| field == name) {
7996 continue;
7997 }
7998 columns.push(format!(
7999 "'{}', {}",
8000 name.replace('\'', "''"),
8001 column_expression(child_table, child_alias, name, names)
8002 ));
8003 }
8004 for (field, expression) in &embeds {
8005 columns.push(format!("'{}', {}", field.replace('\'', "''"), expression));
8006 }
8007 let row = if columns.is_empty() {
8008 format!("row_to_json({})", postrust_sql::escape_ident(child_alias))
8009 } else {
8010 format!("json_build_object({})", columns.join(", "))
8011 };
8012 parts.push(format!(
8013 "COALESCE(json_agg({}), '[]'::json) AS {}",
8014 row,
8015 postrust_sql::escape_ident("nodes")
8016 ));
8017 }
8018 _ => {}
8019 }
8020 }
8021
8022 if parts.is_empty() {
8023 parts.push("count(*) AS pgrst_empty".to_string());
8026 }
8027 Ok(parts.join(", "))
8028}
8029
8030fn column_expression(
8035 table: Option<&postrust_core::schema_cache::Table>,
8036 alias: &str,
8037 name: &str,
8038 names: &crate::names::NameOverrides,
8039) -> String {
8040 let column = table
8041 .map(|t| table_column_for(names, t, name))
8042 .unwrap_or(name);
8043 let plain = format!(
8044 "{}.{}",
8045 postrust_sql::escape_ident(alias),
8046 postrust_sql::escape_ident(column)
8047 );
8048 let computed = table
8049 .filter(|t| t.get_column(column).is_none())
8050 .and_then(|t| {
8051 let function = names
8052 .computed_source(&t.schema, &t.name, name)
8053 .unwrap_or(name);
8054 t.get_computed_column(function)
8055 });
8056 match computed {
8057 Some(definition) => computed_call(
8058 definition,
8059 &format!("{}.*", postrust_sql::escape_ident(alias)),
8060 &postrust_sql::escape_ident(alias),
8061 ),
8062 None => plain,
8063 }
8064}
8065
8066fn order_terms(
8074 order: &serde_json::Value,
8075 schema_cache: &SchemaCache,
8076 schema_name: &str,
8077 table_name: &str,
8078 alias: &str,
8079 names: &crate::names::NameOverrides,
8080) -> Result<Option<String>, async_graphql::Error> {
8081 let entries: Vec<&serde_json::Value> = match order {
8082 serde_json::Value::Array(items) => items.iter().collect(),
8083 serde_json::Value::Object(_) => vec![order],
8084 _ => return Ok(None),
8085 };
8086
8087 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
8088 let table = schema_cache
8089 .get_table(&qi)
8090 .ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
8091
8092 let mut terms = Vec::new();
8093 for entry in entries {
8094 let serde_json::Value::Object(map) = entry else {
8095 continue;
8096 };
8097 for (field, direction) in map {
8098 if direction.is_null() {
8100 continue;
8101 }
8102 let column = table_column_for(names, table, field);
8103 if table.get_column(column).is_none() {
8104 return Err(async_graphql::Error::new(format!(
8105 "cannot order by unknown column \"{}\" on \"{}\"",
8106 field, table_name
8107 )));
8108 }
8109 let name = direction.as_str().unwrap_or_default();
8110 let sql = crate::input::order_by::direction_sql(name).ok_or_else(|| {
8111 async_graphql::Error::new(format!("\"{}\" is not a sort direction", name))
8112 })?;
8113 terms.push(format!(
8114 "{}.{} {}",
8115 postrust_sql::escape_ident(alias),
8116 postrust_sql::escape_ident(column),
8117 sql
8118 ));
8119 }
8120 }
8121
8122 Ok(if terms.is_empty() {
8123 None
8124 } else {
8125 Some(terms.join(", "))
8126 })
8127}
8128
8129fn always_present(rel: &RelationshipField, table: &postrust_core::schema_cache::Table) -> bool {
8141 use postrust_core::schema_cache::{Cardinality, Relationship};
8142 let Relationship::ForeignKey { cardinality, .. } = &rel.relationship else {
8143 return false;
8144 };
8145 match cardinality {
8146 Cardinality::M2O { .. } => {}
8147 Cardinality::O2O {
8148 is_parent: false, ..
8149 } => {}
8150 _ => return false,
8151 }
8152 let columns = cardinality.columns();
8153 !columns.is_empty()
8154 && columns.iter().all(|(local, _)| {
8155 table
8156 .get_column(local)
8157 .is_some_and(|column| !column.nullable)
8158 })
8159}
8160
8161fn child_holds_the_key(relationship: &postrust_core::schema_cache::Relationship) -> bool {
8169 use postrust_core::schema_cache::{Cardinality, Relationship};
8170 match relationship {
8171 Relationship::ForeignKey { cardinality, .. } => match cardinality {
8172 Cardinality::M2O { .. } => false,
8173 Cardinality::O2O { is_parent, .. } => *is_parent,
8174 _ => true,
8175 },
8176 Relationship::Computed { .. } => true,
8177 }
8178}
8179
8180fn computed_session_argument(
8182 cache: &SchemaCache,
8183 function: &postrust_core::api_request::QualifiedIdentifier,
8184 row_argument: &str,
8185) -> Option<String> {
8186 let _ = row_argument;
8187 cache
8188 .routines
8189 .get(function)
8190 .into_iter()
8191 .flatten()
8192 .find(|routine| routine.name == function.name)?
8193 .params
8194 .iter()
8195 .find(|param| {
8196 param.name == "hasura_session" && matches!(param.param_type.as_str(), "json" | "jsonb")
8197 })
8198 .map(|param| param.name.clone())
8199}
8200
8201fn computed_args_type_name(type_name: &str, field: &str) -> String {
8203 format!("{}_{}_args", type_name, field)
8204}
8205
8206fn computed_arguments(
8213 relationship: &RelationshipField,
8214 plan: &postrust_core::embed::EmbedPlan,
8215 parent_row: &str,
8216 given: Option<&serde_json::Value>,
8217 param_idx: &mut usize,
8218 values: &mut Vec<serde_json::Value>,
8219 schema_cache: &SchemaCache,
8220) -> Result<Option<String>, async_graphql::Error> {
8221 let (Some(function), Some(row_argument)) = (&plan.function, &plan.row_argument) else {
8222 return Ok(None);
8223 };
8224 let session = schema_cache
8225 .routines
8226 .get(function)
8227 .into_iter()
8228 .flatten()
8229 .find(|routine| routine.name == function.name)
8230 .and_then(|routine| {
8231 routine
8232 .params
8233 .iter()
8234 .find(|param| {
8235 param.name == "hasura_session"
8236 && matches!(param.param_type.as_str(), "json" | "jsonb")
8237 })
8238 .map(|param| param.name.clone())
8239 });
8240 if relationship.arguments.is_empty() && session.is_none() {
8241 return Ok(None);
8242 }
8243
8244 let mut passed = vec![format!(
8245 "{} => {}",
8246 postrust_sql::escape_ident(row_argument),
8247 parent_row
8248 )];
8249 if let Some(session) = session {
8250 passed.push(format!(
8251 "{} => {}",
8252 postrust_sql::escape_ident(&session),
8253 SESSION_ARGUMENT
8254 ));
8255 }
8256 for (name, pg_type, required) in &relationship.arguments {
8257 match given.and_then(|given| given.get(name)) {
8258 None if *required => {
8259 return Err(async_graphql::Error::new(format!(
8260 "{} needs the argument \"{}\"",
8261 relationship.name, name
8262 )))
8263 }
8264 None => continue,
8266 Some(value) => {
8267 passed.push(format!(
8268 "{} => ${}::{}",
8269 postrust_sql::escape_ident(name),
8270 param_idx,
8271 pg_type
8272 ));
8273 *param_idx += 1;
8274 values.push(value.clone());
8275 }
8276 }
8277 }
8278 Ok(Some(passed.join(", ")))
8279}
8280
8281fn embed_arguments(field: async_graphql::SelectionField<'_>) -> HashMap<String, serde_json::Value> {
8283 field
8284 .arguments()
8285 .map(|args| {
8286 args.into_iter()
8287 .map(|(name, value)| {
8288 (
8289 name.to_string(),
8290 plain_numbers(value.into_json().unwrap_or(serde_json::Value::Null)),
8291 )
8292 })
8293 .collect()
8294 })
8295 .unwrap_or_default()
8296}
8297
8298type Narrowing = (String, Option<String>, Option<String>, Option<i64>, i64);
8302
8303#[allow(clippy::too_many_arguments)] fn embed_narrowing(
8310 arguments: &HashMap<String, serde_json::Value>,
8311 plan: &postrust_core::embed::EmbedPlan,
8312 schema_cache: &SchemaCache,
8313 relationships: &HashMap<String, Vec<RelationshipField>>,
8314 target_type: &str,
8315 child_alias: &str,
8316 max_rows: Option<i64>,
8317 param_idx: &mut usize,
8318 values: &mut Vec<serde_json::Value>,
8319 names: &crate::names::NameOverrides,
8320 caller: &crate::role::Caller<'_>,
8321) -> Result<Narrowing, async_graphql::Error> {
8322 let permission =
8326 permission_predicate(caller, names, &plan.foreign_schema, &plan.foreign_table)?;
8327 let requested = arguments.get("where").filter(|v| !v.is_null()).cloned();
8328 let child_where = match and_predicates(requested, permission) {
8329 Some(expression) => {
8330 let child_scope = WhereScope::for_alias(
8331 &plan.foreign_schema,
8332 &plan.foreign_table,
8333 child_alias,
8334 target_type,
8335 schema_cache,
8336 relationships,
8337 names,
8338 )
8339 .for_caller(*caller);
8340 let mut nested_alias = 0usize;
8341 build_condition(
8342 &expression,
8343 &child_scope,
8344 param_idx,
8345 values,
8346 &mut nested_alias,
8347 )?
8348 }
8349 None => None,
8350 };
8351
8352 let child_order = match arguments.get("order_by") {
8353 Some(order) if !order.is_null() => order_terms(
8354 order,
8355 schema_cache,
8356 &plan.foreign_schema,
8357 &plan.foreign_table,
8358 child_alias,
8359 names,
8360 )?,
8361 _ => None,
8362 };
8363
8364 let ceiling = match (
8369 max_rows,
8370 crate::role::read_limit(caller, names, &plan.foreign_schema, &plan.foreign_table),
8371 ) {
8372 (Some(configured), Some(granted)) => Some(configured.min(granted)),
8373 (only, None) | (None, only) => only,
8374 };
8375 let child_limit = match arguments.get("limit").and_then(|v| v.as_i64()) {
8376 Some(requested) => match ceiling {
8377 Some(ceiling) => Some(requested.min(ceiling)),
8378 None => Some(requested),
8379 },
8380 None => ceiling,
8381 };
8382 let child_offset = arguments
8383 .get("offset")
8384 .and_then(|v| v.as_i64())
8385 .unwrap_or(0);
8386
8387 let requested = match arguments.get("distinct_on") {
8391 Some(serde_json::Value::Array(items)) => items
8392 .iter()
8393 .filter_map(|item| item.as_str())
8394 .collect::<Vec<_>>(),
8395 Some(serde_json::Value::String(one)) => vec![one.as_str()],
8396 _ => Vec::new(),
8397 };
8398 let mut distinct_on = Vec::with_capacity(requested.len());
8399 if !requested.is_empty() {
8400 let qi = postrust_core::api_request::QualifiedIdentifier::new(
8401 &plan.foreign_schema,
8402 &plan.foreign_table,
8403 );
8404 let table = schema_cache.get_table(&qi).ok_or_else(|| {
8405 async_graphql::Error::new(format!("unknown table \"{}\"", plan.foreign_table))
8406 })?;
8407 for field in requested {
8408 let column = table_column_for(names, table, field).to_string();
8409 if table.get_column(&column).is_none() {
8410 return Err(async_graphql::Error::new(format!(
8411 "cannot take distinct on unknown column \"{}\" of \"{}\"",
8412 field, plan.foreign_table
8413 )));
8414 }
8415 distinct_on.push(format!(
8416 "{}.{}",
8417 postrust_sql::escape_ident(child_alias),
8418 postrust_sql::escape_ident(&column)
8419 ));
8420 }
8421 }
8422 let (distinct, child_order) = distinct_on_clause(&distinct_on, child_order.as_deref())?;
8423
8424 Ok((
8425 distinct,
8426 child_where,
8427 child_order,
8428 child_limit,
8429 child_offset,
8430 ))
8431}
8432
8433#[allow(clippy::too_many_arguments)] fn build_embed_expressions(
8435 caller: &crate::role::Caller<'_>,
8436 schema_cache: &SchemaCache,
8437 relationships: &HashMap<String, Vec<RelationshipField>>,
8438 type_name: &str,
8439 parent_alias: &str,
8440 selection: async_graphql::SelectionField<'_>,
8441 max_rows: Option<i64>,
8442 alias_counter: &mut usize,
8443 param_idx: &mut usize,
8444 values: &mut Vec<serde_json::Value>,
8445 names: &crate::names::NameOverrides,
8446) -> Result<Vec<(String, String)>, async_graphql::Error> {
8447 let Some(available) = relationships.get(type_name) else {
8448 return Ok(Vec::new());
8449 };
8450
8451 let mut expressions = Vec::new();
8452
8453 for field in selection.selection_set() {
8454 let aggregate_of = field
8458 .name()
8459 .strip_suffix("_aggregate")
8460 .and_then(|name| available.iter().find(|r| r.name == name && r.is_list));
8461
8462 if let Some(rel) = aggregate_of {
8463 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, schema_cache)
8464 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8465 *alias_counter += 1;
8466 let child_alias = format!("a{}", alias_counter);
8467
8468 let arguments = embed_arguments(field);
8472 let (child_distinct, child_where, child_order, child_limit, child_offset) =
8473 embed_narrowing(
8474 &arguments,
8475 &plan,
8476 schema_cache,
8477 relationships,
8478 &rel.target_type,
8479 &child_alias,
8480 max_rows,
8481 param_idx,
8482 values,
8483 names,
8484 caller,
8485 )?;
8486
8487 let child_table =
8488 schema_cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
8489 &plan.foreign_schema,
8490 &plan.foreign_table,
8491 ));
8492 let select = nested_aggregate_select(
8493 caller,
8494 field,
8495 &child_alias,
8496 schema_cache,
8497 relationships,
8498 &rel.target_type,
8499 child_table,
8500 max_rows,
8501 alias_counter,
8502 param_idx,
8503 values,
8504 names,
8505 )?;
8506
8507 let call = computed_arguments(
8508 rel,
8509 &plan,
8510 &postrust_sql::escape_ident(parent_alias),
8511 arguments.get("args"),
8512 param_idx,
8513 values,
8514 schema_cache,
8515 )?;
8516 let expression = plan
8517 .aggregate_expression(
8518 parent_alias,
8519 &postrust_sql::escape_ident(parent_alias),
8520 &child_alias,
8521 &select,
8522 child_limit,
8523 child_offset,
8524 child_where.as_deref(),
8525 child_order.as_deref(),
8526 call.as_deref(),
8527 &child_distinct,
8528 postrust_core::EmbedRendering::Verbatim,
8532 )
8533 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8534 expressions.push((field.name().to_string(), expression));
8535 continue;
8536 }
8537
8538 let Some(rel) = available.iter().find(|r| r.name == field.name()) else {
8539 continue;
8540 };
8541
8542 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, schema_cache)
8543 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8544
8545 *alias_counter += 1;
8546 let child_alias = format!("e{}", alias_counter);
8547
8548 let nested = build_embed_expressions(
8549 caller,
8550 schema_cache,
8551 relationships,
8552 &rel.target_type,
8553 &child_alias,
8554 field,
8555 max_rows,
8556 alias_counter,
8557 param_idx,
8558 values,
8559 names,
8560 )?;
8561
8562 let arguments = embed_arguments(field);
8564 let (child_distinct, child_where, child_order, child_limit, child_offset) =
8565 embed_narrowing(
8566 &arguments,
8567 &plan,
8568 schema_cache,
8569 relationships,
8570 &rel.target_type,
8571 &child_alias,
8572 max_rows,
8573 param_idx,
8574 values,
8575 names,
8576 caller,
8577 )?;
8578
8579 let child_relationships = relationships.get(&rel.target_type);
8582 let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
8583 &plan.foreign_schema,
8584 &plan.foreign_table,
8585 );
8586 let child_table = schema_cache.get_table(&child_qi);
8587 let mut parts: Vec<String> = Vec::new();
8588 for sub in field.selection_set() {
8589 let name = sub.name();
8590 let is_relationship = child_relationships
8591 .map(|rels| {
8592 rels.iter().any(|r| {
8593 r.name == name || (r.is_list && format!("{}_aggregate", r.name) == name)
8594 })
8595 })
8596 .unwrap_or(false);
8597 if is_relationship {
8598 continue;
8599 }
8600 let column = child_table
8603 .map(|t| table_column_for(names, t, name))
8604 .unwrap_or(name);
8605 let computed = child_table
8606 .filter(|t| t.get_column(column).is_none())
8607 .and_then(|t| {
8608 let function = names
8609 .computed_source(&t.schema, &t.name, name)
8610 .unwrap_or(name);
8611 t.get_computed_column(function)
8612 });
8613 match computed {
8614 Some(definition) => parts.push(format!(
8615 "{} AS {}",
8616 computed_call(
8617 definition,
8618 &format!("{}.*", postrust_sql::escape_ident(&child_alias)),
8619 &postrust_sql::escape_ident(&child_alias),
8620 ),
8621 postrust_sql::escape_ident(name)
8622 )),
8623 None if column != name => parts.push(format!(
8626 "{} AS {}",
8627 postrust_sql::escape_ident(column),
8628 postrust_sql::escape_ident(name)
8629 )),
8630 None => parts.push(postrust_sql::escape_ident(name)),
8631 }
8632 }
8633 if parts.is_empty() && nested.is_empty() {
8634 parts.push(
8635 child_table
8636 .and_then(|table| rename_projection(table, &child_alias, names))
8637 .unwrap_or_else(|| format!("{}.*", postrust_sql::escape_ident(&child_alias))),
8638 );
8639 }
8640 for (field_name, expression) in nested {
8641 parts.push(format!(
8642 "{} AS {}",
8643 expression,
8644 postrust_sql::escape_ident(&field_name)
8645 ));
8646 }
8647
8648 let computed_call_arguments = computed_arguments(
8649 rel,
8650 &plan,
8651 &postrust_sql::escape_ident(parent_alias),
8652 arguments.get("args"),
8653 param_idx,
8654 values,
8655 schema_cache,
8656 )?;
8657 let expression = plan
8658 .embed_expression(
8659 parent_alias,
8660 &postrust_sql::escape_ident(parent_alias),
8664 &child_alias,
8665 &format!("{}{}", child_distinct, parts.join(", ")),
8666 child_limit,
8667 child_offset,
8668 child_where.as_deref(),
8669 child_order.as_deref(),
8670 computed_call_arguments.as_deref(),
8671 postrust_core::EmbedRendering::Verbatim,
8672 )
8673 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8674
8675 expressions.push((rel.name.clone(), expression));
8676 }
8677
8678 Ok(expressions)
8679}
8680
8681struct EmbedContext<'c> {
8687 schema_cache: &'c SchemaCache,
8688 relationships: &'c HashMap<String, Vec<RelationshipField>>,
8689 max_rows: Option<i64>,
8690}
8691
8692fn embed_relationships<'f>(
8698 conn: &'f mut sqlx::PgConnection,
8699 ctx: &'f EmbedContext<'f>,
8700 type_name: &'f str,
8701 selection: async_graphql::SelectionField<'f>,
8702 rows: &'f mut [serde_json::Value],
8703) -> futures::future::BoxFuture<'f, Result<(), async_graphql::Error>> {
8704 Box::pin(async move {
8705 if rows.is_empty() {
8706 return Ok(());
8707 }
8708
8709 let Some(available) = ctx.relationships.get(type_name) else {
8710 return Ok(());
8711 };
8712
8713 for requested in selection.selection_set() {
8714 let Some(rel) = available.iter().find(|r| r.name == requested.name()) else {
8715 continue;
8716 };
8717
8718 let plan =
8719 postrust_core::embed::EmbedPlan::resolve(&rel.relationship, ctx.schema_cache)
8720 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8721
8722 let keys = postrust_core::embed::parent_keys(rows, &plan.local_column);
8723
8724 let mut child_columns: Vec<String> = Vec::new();
8734 let mut project_everything = false;
8735 let child_relationships = ctx.relationships.get(&rel.target_type);
8736 for field in requested.selection_set() {
8737 let name = field.name();
8738 match child_relationships.and_then(|rels| rels.iter().find(|r| r.name == name)) {
8739 Some(nested_rel) => {
8740 match postrust_core::embed::EmbedPlan::resolve(
8741 &nested_rel.relationship,
8742 ctx.schema_cache,
8743 ) {
8744 Ok(nested_plan) => child_columns.push(nested_plan.local_column),
8745 Err(_) => project_everything = true,
8746 }
8747 }
8748 None => child_columns.push(name.to_string()),
8749 }
8750 }
8751 if project_everything || child_columns.is_empty() {
8752 child_columns.clear();
8753 }
8754
8755 let mut grouped = if keys.is_empty() {
8756 std::collections::HashMap::new()
8757 } else {
8758 let sql = plan
8759 .children_grouped_sql(ctx.max_rows, &child_columns)
8760 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8761
8762 let fetched = sqlx::query(&sql)
8763 .bind(&keys)
8764 .fetch_all(&mut *conn)
8765 .await
8766 .map_err(database_error)?;
8767
8768 use sqlx::Row;
8771 let pairs: Vec<(serde_json::Value, serde_json::Value)> = fetched
8772 .into_iter()
8773 .filter_map(|row| {
8774 Some((
8775 row.try_get::<serde_json::Value, _>(0).ok()?,
8776 row.try_get::<serde_json::Value, _>(1).ok()?,
8777 ))
8778 })
8779 .collect();
8780
8781 postrust_core::embed::group_from_aggregated(pairs)
8782 };
8783
8784 let has_deeper_embed = ctx
8789 .relationships
8790 .get(&rel.target_type)
8791 .map(|rels| {
8792 requested
8793 .selection_set()
8794 .any(|field| rels.iter().any(|r| r.name == field.name()))
8795 })
8796 .unwrap_or(false);
8797
8798 if has_deeper_embed {
8799 let mut order: Vec<(String, usize)> = Vec::with_capacity(grouped.len());
8800 let mut flat: Vec<serde_json::Value> = Vec::new();
8801 for (key, children) in grouped.drain() {
8802 order.push((key, children.len()));
8803 flat.extend(children);
8804 }
8805
8806 embed_relationships(&mut *conn, ctx, &rel.target_type, requested, &mut flat)
8807 .await?;
8808
8809 let mut rest = flat.into_iter();
8810 for (key, count) in order {
8811 grouped.insert(key, rest.by_ref().take(count).collect());
8812 }
8813 }
8814
8815 for row in rows.iter_mut() {
8816 postrust_core::embed::attach_to_parent(row, &rel.name, &plan, &grouped);
8817 }
8818 }
8819
8820 Ok(())
8821 })
8822}
8823
8824fn pk_where_from_args(
8829 ctx: &ResolverContext<'_>,
8830 schema_name: &str,
8831 table_name: &str,
8832 pk_columns: &[(String, String)],
8833 names: &crate::names::NameOverrides,
8834) -> Result<serde_json::Value, async_graphql::Error> {
8835 if pk_columns.is_empty() {
8836 return Err(async_graphql::Error::new(format!(
8837 "\"{}\" has no primary key, so it cannot be mutated by key",
8838 table_name
8839 )));
8840 }
8841
8842 let from_object = ctx
8846 .args
8847 .try_get("pk_columns")
8848 .ok()
8849 .map(|v| accessor_to_json(&v));
8850
8851 let mut conditions = serde_json::Map::new();
8852 for (col_name, _) in pk_columns {
8853 let field = names
8857 .column(schema_name, table_name, col_name)
8858 .unwrap_or(col_name);
8859 let value = match &from_object {
8860 Some(serde_json::Value::Object(map)) => map.get(field).cloned().ok_or_else(|| {
8861 async_graphql::Error::new(format!(
8862 "pk_columns is missing the key column \"{}\"",
8863 field
8864 ))
8865 })?,
8866 _ => {
8867 let arg = ctx.args.try_get(field).map_err(|_| {
8868 async_graphql::Error::new(format!(
8869 "missing required primary key argument \"{}\"",
8870 field
8871 ))
8872 })?;
8873 accessor_to_json(&arg)
8874 }
8875 };
8876 conditions.insert(field.to_string(), serde_json::json!({ "_eq": value }));
8877 }
8878
8879 Ok(serde_json::Value::Object(conditions))
8880}
8881
8882fn pk_argument_type(pg_type: &str) -> String {
8889 let rendered = crate::types::pg_type_to_graphql(pg_type).to_string();
8890 if rendered.starts_with('[') {
8891 "String".to_string()
8892 } else {
8893 rendered
8894 }
8895}
8896
8897fn graphql_type_ref(type_str: &str) -> TypeRef {
8899 let is_list = type_str.starts_with('[');
8901 let is_nn = type_str.ends_with('!');
8902
8903 let inner = if is_list {
8905 let stripped = type_str
8906 .trim_end_matches('!') .trim_start_matches('[') .trim_end_matches(']'); stripped
8910 } else {
8911 type_str.trim_end_matches('!')
8912 };
8913
8914 let inner_nn = inner.ends_with('!');
8915 let base_type = inner.trim_end_matches('!');
8916
8917 if is_list {
8918 if is_nn {
8919 if inner_nn {
8920 TypeRef::named_nn_list_nn(base_type)
8921 } else {
8922 TypeRef::named_list_nn(base_type)
8923 }
8924 } else if inner_nn {
8925 TypeRef::named_nn_list(base_type)
8926 } else {
8927 TypeRef::named_list(base_type)
8928 }
8929 } else if is_nn {
8930 TypeRef::named_nn(base_type)
8931 } else {
8932 TypeRef::named(base_type)
8933 }
8934}
8935
8936const NUMBER_AS_TEXT: &str = "$serde_json::private::Number";
8946
8947fn number_kept_as_text(
8950 map: &serde_json::Map<String, serde_json::Value>,
8951) -> Option<serde_json::Value> {
8952 if map.len() != 1 {
8953 return None;
8954 }
8955 let text = map.get(NUMBER_AS_TEXT)?.as_str()?;
8956 serde_json::from_str::<serde_json::Value>(text)
8957 .ok()
8958 .filter(|value| value.is_number())
8959}
8960
8961fn plain_numbers(value: serde_json::Value) -> serde_json::Value {
8963 match value {
8964 serde_json::Value::Object(map) => match number_kept_as_text(&map) {
8965 Some(number) => number,
8966 None => serde_json::Value::Object(
8967 map.into_iter()
8968 .map(|(k, v)| (k, plain_numbers(v)))
8969 .collect(),
8970 ),
8971 },
8972 serde_json::Value::Array(items) => {
8973 serde_json::Value::Array(items.into_iter().map(plain_numbers).collect())
8974 }
8975 other => other,
8976 }
8977}
8978
8979fn accessor_to_json(accessor: &ValueAccessor<'_>) -> serde_json::Value {
8981 if accessor.is_null() {
8983 serde_json::Value::Null
8984 } else if let Ok(b) = accessor.boolean() {
8985 serde_json::Value::Bool(b)
8986 } else if let Ok(i) = accessor.i64() {
8987 serde_json::Value::Number(i.into())
8988 } else if let Ok(f) = accessor.f64() {
8989 serde_json::Number::from_f64(f)
8990 .map(serde_json::Value::Number)
8991 .unwrap_or(serde_json::Value::Null)
8992 } else if let Ok(s) = accessor.string() {
8993 serde_json::Value::String(s.to_string())
8994 } else if let Ok(name) = accessor.enum_name() {
8995 serde_json::Value::String(name.to_string())
8998 } else if let Ok(list) = accessor.list() {
8999 serde_json::Value::Array(list.iter().map(|v| accessor_to_json(&v)).collect())
9000 } else if let Ok(obj) = accessor.object() {
9001 let map: serde_json::Map<String, serde_json::Value> = obj
9002 .iter()
9003 .map(|(k, v)| (k.to_string(), accessor_to_json(&v)))
9004 .collect();
9005 match number_kept_as_text(&map) {
9006 Some(number) => number,
9007 None => serde_json::Value::Object(map),
9008 }
9009 } else {
9010 serde_json::Value::Null
9011 }
9012}
9013
9014fn value_to_json(value: &Value) -> serde_json::Value {
9016 match value {
9017 Value::Null => serde_json::Value::Null,
9018 Value::Boolean(b) => serde_json::Value::Bool(*b),
9019 Value::Number(n) => {
9020 if let Some(i) = n.as_i64() {
9021 serde_json::Value::Number(i.into())
9022 } else if let Some(f) = n.as_f64() {
9023 serde_json::Value::Number(serde_json::Number::from_f64(f).unwrap())
9024 } else {
9025 serde_json::Value::Null
9026 }
9027 }
9028 Value::String(s) => serde_json::Value::String(s.clone()),
9029 Value::List(arr) => serde_json::Value::Array(arr.iter().map(value_to_json).collect()),
9030 Value::Object(obj) => {
9031 let map: serde_json::Map<String, serde_json::Value> = obj
9032 .iter()
9033 .map(|(k, v)| (k.to_string(), value_to_json(v)))
9034 .collect();
9035 serde_json::Value::Object(map)
9036 }
9037 Value::Binary(b) => serde_json::Value::String(base64::Engine::encode(
9038 &base64::engine::general_purpose::STANDARD,
9039 b,
9040 )),
9041 Value::Enum(e) => serde_json::Value::String(e.to_string()),
9042 }
9043}
9044
9045fn json_to_value(json: serde_json::Value) -> Value {
9047 match json {
9048 serde_json::Value::Null => Value::Null,
9049 serde_json::Value::Bool(b) => Value::Boolean(b),
9050 serde_json::Value::Number(n) => {
9051 if let Some(i) = n.as_i64() {
9052 Value::Number(i.into())
9053 } else if let Some(f) = n.as_f64() {
9054 Value::Number(async_graphql::Number::from_f64(f).unwrap())
9055 } else {
9056 Value::Null
9057 }
9058 }
9059 serde_json::Value::String(s) => Value::String(s),
9060 serde_json::Value::Array(arr) => Value::List(arr.into_iter().map(json_to_value).collect()),
9061 serde_json::Value::Object(obj) => {
9062 let map: indexmap::IndexMap<async_graphql::Name, Value> = obj
9063 .into_iter()
9064 .map(|(k, v)| (async_graphql::Name::new(k), json_to_value(v)))
9065 .collect();
9066 Value::Object(map)
9067 }
9068 }
9069}
9070
9071#[cfg(test)]
9081mod tests {
9082 use super::*;
9083 use indexmap::IndexMap;
9084 use postrust_core::schema_cache::{Column, Table};
9085 use std::collections::{HashMap, HashSet};
9086
9087 fn create_test_table(name: &str) -> Table {
9088 let mut columns = IndexMap::new();
9089 columns.insert(
9090 "id".into(),
9091 Column {
9092 name: "id".into(),
9093 description: None,
9094 nullable: false,
9095 data_type: "integer".into(),
9096 nominal_type: "int4".into(),
9097 max_len: None,
9098 default: Some("nextval('id_seq')".into()),
9099 enum_values: vec![],
9100 is_pk: true,
9101 position: 1,
9102 always_generated: false,
9103 domain_type: None,
9104 },
9105 );
9106 columns.insert(
9107 "name".into(),
9108 Column {
9109 name: "name".into(),
9110 description: None,
9111 nullable: false,
9112 data_type: "text".into(),
9113 nominal_type: "text".into(),
9114 max_len: None,
9115 default: None,
9116 enum_values: vec![],
9117 is_pk: false,
9118 position: 2,
9119 always_generated: false,
9120 domain_type: None,
9121 },
9122 );
9123
9124 Table {
9125 schema: "public".into(),
9126 name: name.into(),
9127 description: None,
9128 is_view: false,
9129 insertable: true,
9130 updatable: true,
9131 deletable: true,
9132 pk_cols: vec!["id".into()],
9133 unique_constraints: Vec::new(),
9134 columns,
9135 computed_columns: Default::default(),
9136 is_partitioned: false,
9137 }
9138 }
9139
9140 fn create_test_schema_cache() -> SchemaCache {
9141 let mut tables = HashMap::new();
9142 let users = create_test_table("users");
9143 tables.insert(users.qualified_identifier(), users);
9144
9145 SchemaCache {
9146 tables,
9147 relationships: HashMap::new(),
9148 routines: HashMap::new(),
9149 timezones: HashSet::new(),
9150 media_handlers: HashMap::new(),
9151 pg_version: 150000,
9152 representations: Default::default(),
9153 }
9154 }
9155
9156 #[test]
9161 fn test_graphql_type_ref_simple() {
9162 let _type_ref = graphql_type_ref("String");
9163 }
9165
9166 #[test]
9167 fn test_graphql_type_ref_non_null() {
9168 let _type_ref = graphql_type_ref("String!");
9169 }
9170
9171 #[test]
9172 fn test_graphql_type_ref_list() {
9173 let _type_ref = graphql_type_ref("[String]");
9174 }
9175
9176 #[test]
9177 fn test_graphql_type_ref_list_non_null() {
9178 let _type_ref = graphql_type_ref("[String!]!");
9179 }
9180
9181 #[test]
9186 fn test_value_to_json_null() {
9187 let value = Value::Null;
9188 let json = value_to_json(&value);
9189 assert_eq!(json, serde_json::Value::Null);
9190 }
9191
9192 #[test]
9193 fn test_value_to_json_boolean() {
9194 let value = Value::Boolean(true);
9195 let json = value_to_json(&value);
9196 assert_eq!(json, serde_json::Value::Bool(true));
9197 }
9198
9199 #[test]
9200 fn test_value_to_json_number() {
9201 let value = Value::Number(42.into());
9202 let json = value_to_json(&value);
9203 assert_eq!(json, serde_json::json!(42));
9204 }
9205
9206 #[test]
9207 fn test_value_to_json_string() {
9208 let value = Value::String("hello".to_string());
9209 let json = value_to_json(&value);
9210 assert_eq!(json, serde_json::Value::String("hello".to_string()));
9211 }
9212
9213 #[test]
9214 fn test_value_to_json_list() {
9215 let value = Value::List(vec![Value::Number(1.into()), Value::Number(2.into())]);
9216 let json = value_to_json(&value);
9217 assert_eq!(json, serde_json::json!([1, 2]));
9218 }
9219
9220 #[test]
9221 fn test_json_to_value_null() {
9222 let json = serde_json::Value::Null;
9223 let value = json_to_value(json);
9224 assert!(matches!(value, Value::Null));
9225 }
9226
9227 #[test]
9228 fn test_json_to_value_boolean() {
9229 let json = serde_json::Value::Bool(false);
9230 let value = json_to_value(json);
9231 assert!(matches!(value, Value::Boolean(false)));
9232 }
9233
9234 #[test]
9235 fn test_json_to_value_number() {
9236 let json = serde_json::json!(123);
9237 let value = json_to_value(json);
9238 assert!(matches!(value, Value::Number(_)));
9239 }
9240
9241 #[test]
9242 fn test_json_to_value_string() {
9243 let json = serde_json::Value::String("test".to_string());
9244 let value = json_to_value(json);
9245 assert!(matches!(value, Value::String(_)));
9246 }
9247
9248 #[test]
9249 fn test_json_to_value_array() {
9250 let json = serde_json::json!([1, 2, 3]);
9251 let value = json_to_value(json);
9252 assert!(matches!(value, Value::List(_)));
9253 }
9254
9255 #[test]
9256 fn test_json_to_value_object() {
9257 let json = serde_json::json!({"key": "value"});
9258 let value = json_to_value(json);
9259 assert!(matches!(value, Value::Object(_)));
9260 }
9261
9262 #[test]
9267 fn test_build_dynamic_schema() {
9268 let cache = create_test_schema_cache();
9269 let config = SchemaConfig::default();
9270 let generated = build_schema(&cache, &config);
9271
9272 let result = build_dynamic_schema(
9273 &generated,
9274 &cache,
9275 None,
9276 None,
9277 Arc::new(Default::default()),
9278 std::time::Duration::from_secs(30),
9279 None,
9280 );
9281 if let Err(ref e) = result {
9282 eprintln!("Schema build error: {:?}", e);
9283 }
9284 assert!(result.is_ok(), "Schema build failed: {:?}", result.err());
9285 }
9286
9287 #[test]
9290 fn exists_is_a_subselect_over_another_table() {
9291 let cache = create_test_schema_cache();
9292 let names = crate::names::NameOverrides::default();
9293 let relationships: HashMap<String, Vec<RelationshipField>> = HashMap::new();
9294 let scope = WhereScope::table("public", "users", "users", &names)
9295 .with_resolution(&cache, &relationships);
9296
9297 let mut param_idx = 1usize;
9298 let mut values: Vec<serde_json::Value> = Vec::new();
9299 let mut aliases = 0usize;
9300 let sql = build_condition(
9301 &serde_json::json!({
9302 "_exists": {"_table": "users", "_where": {"name": {"_eq": "Ann"}}}
9303 }),
9304 &scope,
9305 &mut param_idx,
9306 &mut values,
9307 &mut aliases,
9308 )
9309 .expect("compiles")
9310 .expect("constrains something");
9311
9312 assert_eq!(
9313 sql,
9314 "EXISTS (SELECT 1 FROM \"public\".\"users\" AS \"pgrst_exists_1\" \
9315 WHERE \"pgrst_exists_1\".\"name\" = $1::text)"
9316 );
9317 assert_eq!(values, vec![serde_json::json!("Ann")]);
9318 }
9319
9320 #[test]
9322 fn exists_names_a_table_the_server_has() {
9323 let cache = create_test_schema_cache();
9324 let names = crate::names::NameOverrides::default();
9325 let relationships: HashMap<String, Vec<RelationshipField>> = HashMap::new();
9326 let scope = WhereScope::table("public", "users", "users", &names)
9327 .with_resolution(&cache, &relationships);
9328 let error = build_condition(
9329 &serde_json::json!({"_exists": {"_table": "nowhere", "_where": {}}}),
9330 &scope,
9331 &mut 1,
9332 &mut Vec::new(),
9333 &mut 0,
9334 )
9335 .expect_err("a table that is not there is an error");
9336 assert!(
9337 error.message.contains("public.nowhere"),
9338 "{}",
9339 error.message
9340 );
9341 }
9342
9343 #[test]
9350 fn a_table_a_role_may_count_and_not_read_has_a_count_and_no_rows() {
9351 let cache = create_test_schema_cache();
9352 let names = crate::names::NameOverrides::parse(
9353 r#"{"tables": {"public.users": {"permissions":
9354 {"counter": {"select": {"columns": [], "filter": {},
9355 "allow_aggregations": true}}}}}}"#,
9356 )
9357 .unwrap();
9358 let view = crate::role::cache_for_role(&cache, &names, "counter", false);
9359 assert!(!view.tables.is_empty(), "the table is there to be counted");
9360
9361 let config = SchemaConfig {
9362 names: names.clone(),
9363 role: Some("counter".to_string()),
9364 ..SchemaConfig::default()
9365 };
9366 let generated = build_schema(&view, &config);
9367 let schema = build_dynamic_schema(
9368 &generated,
9369 &view,
9370 None,
9371 None,
9372 Arc::new(names),
9373 std::time::Duration::from_secs(30),
9374 Some("counter"),
9375 )
9376 .expect("a role that may only count still has a buildable schema");
9377 let sdl = schema.sdl();
9378
9379 assert!(
9380 sdl.contains("users_aggregate"),
9381 "the count is there:\n{}",
9382 sdl
9383 );
9384 assert!(sdl.contains("count: Int!"), "and it is a count:\n{}", sdl);
9385 assert!(!sdl.contains("type users "), "no row type:\n{}", sdl);
9387 assert!(
9388 !sdl.contains("nodes"),
9389 "and no rows beside the numbers:\n{}",
9390 sdl
9391 );
9392 assert!(
9393 !sdl.contains("users_select_column"),
9394 "no column enum to order by or count over:\n{}",
9395 sdl
9396 );
9397 assert!(!sdl.contains("users_max_fields"), "no max:\n{}", sdl);
9399 }
9400
9401 #[test]
9408 fn a_table_a_role_may_only_write_gets_the_write_and_no_type() {
9409 let cache = create_test_schema_cache();
9410 let names = crate::names::NameOverrides::parse(
9411 r#"{"tables": {"public.users": {"permissions":
9412 {"writer": {"insert": {"columns": "*", "check": {}}}}}}}"#,
9413 )
9414 .unwrap();
9415 let view = crate::role::cache_for_role(&cache, &names, "writer", false);
9416 let config = SchemaConfig {
9417 names: names.clone(),
9418 role: Some("writer".to_string()),
9419 ..SchemaConfig::default()
9420 };
9421 let generated = build_schema(&view, &config);
9422
9423 let schema = build_dynamic_schema(
9424 &generated,
9425 &view,
9426 None,
9427 None,
9428 Arc::new(names),
9429 std::time::Duration::from_secs(30),
9430 Some("writer"),
9431 )
9432 .expect("a role that may only write still has a buildable schema");
9433 let sdl = schema.sdl();
9434
9435 assert!(
9436 sdl.contains("insert_users("),
9437 "the write is there:\n{}",
9438 sdl
9439 );
9440 assert!(
9441 sdl.contains("users_insert_input"),
9442 "and its input:\n{}",
9443 sdl
9444 );
9445 assert!(!sdl.contains("type users "), "no row type:\n{}", sdl);
9447 assert!(!sdl.contains("insert_users_one"), "no insert_one:\n{}", sdl);
9448 assert!(
9449 !sdl.contains("returning"),
9450 "and nothing to return:\n{}",
9451 sdl
9452 );
9453 assert!(
9454 sdl.contains("users_mutation_response"),
9455 "the count is still answered:\n{}",
9456 sdl
9457 );
9458 assert!(
9459 !sdl.contains("_mutation_response_mutation_response"),
9460 "and no type is named twice over:\n{}",
9461 sdl
9462 );
9463 }
9464
9465 fn cache_with_a_mutation_function() -> SchemaCache {
9468 use postrust_core::schema_cache::{FuncVolatility, RetType, Routine};
9469
9470 let mut cache = create_test_schema_cache();
9471 let routine = Routine {
9472 schema: "public".into(),
9473 name: "add_to_score".into(),
9474 description: None,
9475 params: Vec::new(),
9476 return_type: RetType::SetOf("users".into()),
9477 returns_composite: true,
9478 volatility: FuncVolatility::Volatile,
9479 has_variadic: false,
9480 isolation_level: None,
9481 settings: Vec::new(),
9482 is_procedure: false,
9483 media_type: None,
9484 media_base_type: None,
9485 output_columns: Vec::new(),
9486 };
9487 cache
9488 .routines
9489 .insert(routine.qualified_identifier(), vec![routine]);
9490 cache
9491 }
9492
9493 #[test]
9498 fn a_mutation_function_needs_the_role_to_be_named() {
9499 let cache = cache_with_a_mutation_function();
9500 let withheld = crate::names::NameOverrides::parse(
9501 r#"{"tables": {"public.users": {"permissions":
9502 {"reader": {"select": {"columns": "*", "filter": {}}}}}},
9503 "functions": {"public.add_to_score": {"exposed_as": "mutation"}}}"#,
9504 )
9505 .unwrap();
9506 let granted = crate::names::NameOverrides::parse(
9507 r#"{"tables": {"public.users": {"permissions":
9508 {"reader": {"select": {"columns": "*", "filter": {}}}}}},
9509 "functions": {"public.add_to_score":
9510 {"exposed_as": "mutation", "roles": ["reader"]}}}"#,
9511 )
9512 .unwrap();
9513
9514 let sdl_for = |names: &crate::names::NameOverrides| {
9515 let view = crate::role::cache_for_role(&cache, names, "reader", false);
9516 let config = SchemaConfig {
9517 names: names.clone(),
9518 role: Some("reader".to_string()),
9519 ..SchemaConfig::default()
9520 };
9521 let generated = build_schema(&view, &config);
9522 build_dynamic_schema(
9523 &generated,
9524 &view,
9525 None,
9526 None,
9527 Arc::new(names.clone()),
9528 std::time::Duration::from_secs(30),
9529 Some("reader"),
9530 )
9531 .expect("schema builds")
9532 .sdl()
9533 };
9534
9535 let without = sdl_for(&withheld);
9536 assert!(
9537 !without.contains("add_to_score"),
9538 "not granted, so not there:\n{}",
9539 without
9540 );
9541 assert!(
9544 !without.contains("mutation: mutation_root"),
9545 "and no mutation root at all:\n{}",
9546 without
9547 );
9548
9549 let with = sdl_for(&granted);
9550 assert!(
9551 with.contains("add_to_score"),
9552 "named by a function permission, so it is:\n{}",
9553 with
9554 );
9555 }
9556
9557 #[test]
9559 fn a_role_the_document_never_names_is_granted_nothing() {
9560 let cache = Arc::new(create_test_schema_cache());
9561 let config = SchemaConfig {
9562 names: crate::names::NameOverrides::parse(
9563 r#"{"tables": {"public.users": {"permissions":
9564 {"reader": {"select": {"columns": "*", "filter": {}}}}}}}"#,
9565 )
9566 .unwrap(),
9567 ..SchemaConfig::default()
9568 };
9569 let empty = build_ungranted_schema(&cache, &config).expect("it builds");
9570 let sdl = empty.sdl();
9571 assert!(
9572 sdl.contains("no_queries_available"),
9573 "nothing to read:\n{}",
9574 sdl
9575 );
9576 assert!(!sdl.contains("type users "), "and no table:\n{}", sdl);
9577 assert!(build_ungranted_schema(&cache, &SchemaConfig::default()).is_none());
9579 }
9580
9581 #[test]
9584 fn a_column_a_role_may_set_without_seeing_is_in_the_input_alone() {
9585 let cache = create_test_schema_cache();
9586 let names = crate::names::NameOverrides::parse(
9587 r#"{"tables": {"public.users": {"permissions":
9588 {"user": {"select": {"columns": ["id"], "filter": {}},
9589 "insert": {"columns": "*", "check": {}}}}}}}"#,
9590 )
9591 .unwrap();
9592 let view = crate::role::cache_for_role(&cache, &names, "user", false);
9593 let config = SchemaConfig {
9594 names: names.clone(),
9595 role: Some("user".to_string()),
9596 ..SchemaConfig::default()
9597 };
9598 let generated = build_schema(&view, &config);
9599 let object = &generated.object_types["users"];
9600 assert_eq!(
9601 object
9602 .fields
9603 .iter()
9604 .map(|f| f.name.as_str())
9605 .collect::<Vec<_>>(),
9606 vec!["id"],
9607 "the type shows what may be read"
9608 );
9609 assert!(
9610 object.writable_fields.iter().any(|f| f.name == "name"),
9611 "and the input keeps what may be written"
9612 );
9613
9614 let schema = build_dynamic_schema(
9615 &generated,
9616 &view,
9617 None,
9618 None,
9619 Arc::new(names),
9620 std::time::Duration::from_secs(30),
9621 Some("user"),
9622 )
9623 .expect("schema builds");
9624 let sdl = schema.sdl();
9625 let input = sdl
9626 .split("input users_insert_input")
9627 .nth(1)
9628 .and_then(|rest| rest.split('}').next())
9629 .expect("the insert input is registered");
9630 assert!(input.contains("name:"), "settable without being readable");
9631 }
9632
9633 #[test]
9637 fn a_generated_column_is_readable_and_not_writable() {
9638 let mut cache = create_test_schema_cache();
9639 cache
9640 .tables
9641 .get_mut(&postrust_core::api_request::QualifiedIdentifier::new(
9642 "public", "users",
9643 ))
9644 .expect("the fixture has users")
9645 .columns
9646 .get_mut("id")
9647 .expect("with an id")
9648 .always_generated = true;
9649
9650 let names = crate::names::NameOverrides::default();
9651 let config = SchemaConfig {
9652 names: names.clone(),
9653 ..SchemaConfig::default()
9654 };
9655 let generated = build_schema(&cache, &config);
9656 let schema = build_dynamic_schema(
9657 &generated,
9658 &cache,
9659 None,
9660 None,
9661 Arc::new(names),
9662 std::time::Duration::from_secs(30),
9663 None,
9664 )
9665 .expect("schema builds");
9666 let sdl = schema.sdl();
9667
9668 let block = |header: &str| {
9669 sdl.split(header)
9670 .nth(1)
9671 .and_then(|rest| rest.split('}').next())
9672 .unwrap_or_else(|| panic!("{} is registered:\n{}", header, sdl))
9673 .to_string()
9674 };
9675 assert!(
9676 block("type users ").contains("id:"),
9677 "still readable:\n{}",
9678 sdl
9679 );
9680 for header in ["input users_insert_input", "input users_set_input"] {
9681 assert!(
9682 !block(header).contains("id:"),
9683 "{} may not name it:\n{}",
9684 header,
9685 block(header)
9686 );
9687 }
9688 assert!(
9691 !sdl.contains("users_inc_input"),
9692 "and nothing is left to increment:\n{}",
9693 sdl
9694 );
9695 assert!(
9696 block("input users_insert_input").contains("name:"),
9697 "and the rest of the table is untouched",
9698 );
9699 }
9700
9701 #[test]
9702 fn test_create_object_type() {
9703 let table = create_test_table("users");
9704 let obj = TableObjectType::from_table(&table);
9705 let _gql_obj = create_object_type(&obj, &[]);
9706 }
9707
9708 #[test]
9709 fn test_create_query_type() {
9710 let cache = create_test_schema_cache();
9711 let config = SchemaConfig::default();
9712 let generated = build_schema(&cache, &config);
9713
9714 let _query = create_query_type(
9715 &generated,
9716 None,
9717 Arc::new(HashMap::new()),
9718 Arc::new(Default::default()),
9719 );
9720 }
9721
9722 #[test]
9723 fn test_create_mutation_type() {
9724 let cache = create_test_schema_cache();
9725 let config = SchemaConfig::default();
9726 let generated = build_schema(&cache, &config);
9727
9728 let _mutation = create_mutation_type(
9729 &generated,
9730 Arc::new(HashMap::new()),
9731 Arc::new(HashMap::new()),
9732 Arc::new(Default::default()),
9733 None,
9734 None,
9735 );
9736 }
9737
9738 #[test]
9743 fn a_scalar_is_named_the_way_a_client_declares_it() {
9744 use crate::types::{pg_type_to_graphql, GraphQLType};
9745
9746 assert_eq!(pg_type_to_graphql("jsonb").to_string(), "jsonb");
9751 assert_eq!(pg_type_to_graphql("json").to_string(), "json");
9752 assert_eq!(pg_type_to_graphql("int8").to_string(), "bigint");
9753 assert_eq!(pg_type_to_graphql("numeric").to_string(), "numeric");
9754 assert_eq!(pg_type_to_graphql("timestamptz").to_string(), "timestamptz");
9755 assert_eq!(pg_type_to_graphql("timestamp").to_string(), "timestamp");
9756 assert_eq!(pg_type_to_graphql("uuid").to_string(), "uuid");
9757
9758 assert_eq!(pg_type_to_graphql("geometry").to_string(), "geometry");
9761 assert_eq!(
9762 pg_type_to_graphql("colors_enum"),
9763 GraphQLType::Custom("colors_enum".to_string())
9764 );
9765
9766 assert_eq!(
9768 leaf_scalar_name(&pg_type_to_graphql("_text")),
9769 "String".to_string()
9770 );
9771 }
9772
9773 #[test]
9778 fn every_table_gets_a_boolean_expression() {
9779 let cache = create_test_schema_cache();
9780 let config = SchemaConfig::default();
9781 let generated = build_schema(&cache, &config);
9782
9783 let (inputs, _) = crate::input::bool_exp::build_inputs(
9784 &generated.object_types,
9785 &generated.relationship_fields,
9786 );
9787 let names: HashSet<String> = inputs.iter().map(|i| i.type_name().to_string()).collect();
9788
9789 for table in generated.object_types.keys() {
9790 let expected = format!("{}_bool_exp", table);
9791 assert!(
9792 names.contains(&expected),
9793 "no {} among {:?}",
9794 expected,
9795 names
9796 );
9797 }
9798 assert!(names.contains("String_comparison_exp"));
9800 assert!(names.contains("Int_comparison_exp"));
9801 }
9802
9803 #[test]
9804 fn a_schema_carrying_the_boolean_expressions_still_builds() {
9805 let cache = create_test_schema_cache();
9806 let config = SchemaConfig::default();
9807 let generated = build_schema(&cache, &config);
9808
9809 let schema = build_dynamic_schema(
9810 &generated,
9811 &cache,
9812 None,
9813 None,
9814 Arc::new(Default::default()),
9815 std::time::Duration::from_secs(30),
9816 None,
9817 );
9818 assert!(schema.is_ok(), "{:?}", schema.err());
9819
9820 let sdl = schema.unwrap().sdl();
9821 assert!(
9822 sdl.contains("_bool_exp"),
9823 "no boolean expressions in:\n{}",
9824 sdl
9825 );
9826 assert!(sdl.contains("_and"), "no _and in the generated expressions");
9827 }
9828
9829 #[test]
9834 fn test_build_schema_with_subscriptions() {
9835 let cache = create_test_schema_cache();
9836 let config = SchemaConfig {
9837 enable_subscriptions: true,
9838 ..SchemaConfig::default()
9839 };
9840 let generated = build_schema(&cache, &config);
9841
9842 let sub_fields = generate_subscription_fields(&cache, &generated);
9844 assert!(!sub_fields.is_empty(), "Should have subscription fields");
9845
9846 let result = build_dynamic_schema(
9848 &generated,
9849 &cache,
9850 Some(&sub_fields),
9851 None,
9852 Arc::new(Default::default()),
9853 std::time::Duration::from_secs(30),
9854 None,
9855 );
9856 assert!(result.is_ok(), "Schema with subscriptions should build");
9857 }
9858
9859 #[test]
9860 fn test_subscription_field_generation() {
9861 let cache = create_test_schema_cache();
9862 let config = SchemaConfig::default();
9863 let generated = build_schema(&cache, &config);
9864
9865 let fields = generate_subscription_fields(&cache, &generated);
9866
9867 assert_eq!(fields.len(), 1);
9869 assert_eq!(fields[0].name, "users");
9870 assert_eq!(fields[0].table_name, "users");
9871 assert_eq!(fields[0].channel_name(), "postrust_public_users");
9872 }
9873
9874 #[test]
9875 fn test_create_subscription_type() {
9876 let cache = create_test_schema_cache();
9877 let config = SchemaConfig::default();
9878 let generated = build_schema(&cache, &config);
9879 let _subscription = create_subscription_type(
9882 &generated,
9883 None,
9884 Arc::new(Default::default()),
9885 Arc::new(Default::default()),
9886 std::time::Duration::from_secs(30),
9887 );
9888 }
9889}
9890
9891#[derive(Debug, PartialEq, Eq)]
9893enum PathStep {
9894 Key(String),
9895 Index(usize),
9896}
9897
9898fn walk_json_path(value: &Value, path: &str) -> Result<Value, async_graphql::Error> {
9905 let mut at = value;
9906 for step in parse_json_path(path)? {
9907 let next = match (&step, at) {
9908 (PathStep::Key(key), Value::Object(map)) => map.get(key.as_str()),
9909 (PathStep::Index(index), Value::List(items)) => items.get(*index),
9910 _ => None,
9913 };
9914 match next {
9915 Some(next) => at = next,
9916 None => return Ok(Value::Null),
9917 }
9918 }
9919 Ok(at.clone())
9920}
9921
9922fn unclosed_bracket(path: &str) -> async_graphql::Error {
9924 let message = format!("\"{}\" is not a JSON path: a bracket is never closed", path);
9925 validation_error(&message)
9926}
9927
9928fn parse_json_path(path: &str) -> Result<Vec<PathStep>, async_graphql::Error> {
9935 let mut steps = Vec::new();
9936 let mut chars = path.chars().peekable();
9937 if chars.peek() == Some(&'$') {
9939 chars.next();
9940 }
9941 while let Some(&c) = chars.peek() {
9942 match c {
9943 '.' => {
9945 chars.next();
9946 }
9947 '[' => {
9948 chars.next();
9949 let quote = match chars.peek() {
9950 Some(&q @ ('\'' | '"')) => {
9951 chars.next();
9952 Some(q)
9953 }
9954 _ => None,
9955 };
9956 let mut inner = String::new();
9957 loop {
9958 let Some(c) = chars.next() else {
9959 return Err(unclosed_bracket(path));
9960 };
9961 match (quote, c) {
9962 (Some(quote), c) if c == quote => break,
9963 (None, ']') => break,
9964 _ => inner.push(c),
9965 }
9966 }
9967 if quote.is_some() {
9969 match chars.next() {
9970 Some(']') => {}
9971 _ => return Err(unclosed_bracket(path)),
9972 }
9973 }
9974 match (quote, inner.parse::<usize>()) {
9977 (None, Ok(index)) => steps.push(PathStep::Index(index)),
9978 _ => steps.push(PathStep::Key(inner)),
9979 }
9980 }
9981 _ => {
9982 let mut key = String::new();
9983 while let Some(&c) = chars.peek() {
9984 if matches!(c, '.' | '[') {
9985 break;
9986 }
9987 key.push(c);
9988 chars.next();
9989 }
9990 steps.push(PathStep::Key(key));
9991 }
9992 }
9993 }
9994 Ok(steps)
9995}
9996
9997#[cfg(test)]
9998mod json_path_tests {
9999 use super::*;
10000
10001 fn steps(path: &str) -> Vec<PathStep> {
10002 parse_json_path(path).expect("the path parses")
10003 }
10004
10005 fn key(name: &str) -> PathStep {
10006 PathStep::Key(name.to_string())
10007 }
10008
10009 #[test]
10010 fn the_document_itself_has_no_steps() {
10011 assert_eq!(steps("$"), vec![]);
10012 assert_eq!(steps(""), vec![]);
10013 }
10014
10015 #[test]
10016 fn a_key_is_written_bare_or_after_a_dot() {
10017 assert_eq!(steps("a"), vec![key("a")]);
10018 assert_eq!(steps(".obj.c1"), vec![key("obj"), key("c1")]);
10019 assert_eq!(steps("._underscore"), vec![key("_underscore")]);
10020 }
10021
10022 #[test]
10023 fn a_bracket_holds_an_index_or_a_quoted_key() {
10024 assert_eq!(steps("arr[0]"), vec![key("arr"), PathStep::Index(0)]);
10025 assert_eq!(steps("['!@#$%^']"), vec![key("!@#$%^")]);
10026 assert_eq!(
10027 steps("translations['hello world!']"),
10028 vec![key("translations"), key("hello world!")]
10029 );
10030 }
10031
10032 #[test]
10034 fn a_dot_before_a_bracket_changes_nothing() {
10035 let expected = vec![key("objs"), PathStep::Index(0), key("\u{4f60}\u{597d}")];
10036 for spelling in [
10037 "objs[0]['\u{4f60}\u{597d}']",
10038 "objs[0][\"\u{4f60}\u{597d}\"]",
10039 "objs[0].[\"\u{4f60}\u{597d}\"]",
10040 "objs.[0].[\"\u{4f60}\u{597d}\"]",
10041 ] {
10042 assert_eq!(steps(spelling), expected, "{}", spelling);
10043 }
10044 }
10045
10046 #[test]
10047 fn a_bracket_that_never_closes_is_refused() {
10048 assert!(parse_json_path("arr[0").is_err());
10049 assert!(parse_json_path("['key'").is_err());
10050 }
10051
10052 #[test]
10053 fn a_step_the_document_does_not_have_reads_null() {
10054 let document = Value::from_json(serde_json::json!({
10055 "obj": {"c1": "c2"},
10056 "arr": [1, 2, 3],
10057 }))
10058 .expect("a document");
10059 let at = |path| walk_json_path(&document, path).expect("the path parses");
10060 assert_eq!(at(".obj.c1"), Value::String("c2".to_string()));
10061 assert_eq!(at("arr[1]"), Value::Number(2.into()));
10062 assert_eq!(at("arr[9]"), Value::Null);
10063 assert_eq!(at(".obj.nothing"), Value::Null);
10064 assert_eq!(at("obj[0]"), Value::Null);
10067 assert_eq!(at("arr.first"), Value::Null);
10068 }
10069}