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 for function in &generated.function_fields {
2638 if function.volatile != volatile {
2639 continue;
2640 }
2641 let spec = Arc::new(QueryFieldSpec {
2642 schema_name: function.returns_table.0.clone(),
2643 table_name: function.returns_table.1.clone(),
2644 type_name: function.returns.clone(),
2645 is_by_pk: false,
2646 pk_columns: Vec::new(),
2647 max_rows,
2648 relationships: Arc::clone(&relationships),
2649 names: Arc::clone(&names),
2650 call: Some(Arc::new(FunctionCall {
2651 schema: function.schema_name.clone(),
2652 name: function.function_name.clone(),
2653 arguments: function.arguments.clone(),
2654 session_argument: function.session_argument.clone(),
2655 })),
2656 });
2657
2658 let mut field = Field::new(
2659 &function.name,
2660 TypeRef::named_nn_list_nn(&function.returns),
2661 move |ctx| {
2662 let spec = Arc::clone(&spec);
2663 FieldFuture::new(async move { resolve_query(&ctx, &spec).await })
2664 },
2665 );
2666 if !function.arguments.is_empty() {
2667 field = field.argument(InputValue::new(
2668 "args",
2669 TypeRef::named_nn(format!("{}_args", function.name)),
2670 ));
2671 }
2672 field = field
2673 .argument(InputValue::new(
2674 "where",
2675 TypeRef::named(crate::input::bool_exp::bool_exp_type_name(
2676 &function.returns,
2677 )),
2678 ))
2679 .argument(InputValue::new(
2680 "order_by",
2681 TypeRef::named_nn_list(crate::input::order_by::order_by_type_name(
2682 &function.returns,
2683 )),
2684 ))
2685 .argument(InputValue::new(
2686 "distinct_on",
2687 TypeRef::named_nn_list(crate::input::order_by::select_column_type_name(
2688 &function.returns,
2689 )),
2690 ))
2691 .argument(InputValue::new("limit", TypeRef::named("Int")))
2692 .argument(InputValue::new("offset", TypeRef::named("Int")));
2693 if let Some(description) = &function.description {
2694 field = field.description(description);
2695 }
2696 root = root.field(field);
2697
2698 let agg_spec = Arc::new(AggregateSpec {
2703 schema_name: function.returns_table.0.clone(),
2704 table_name: function.returns_table.1.clone(),
2705 type_name: function.returns.clone(),
2706 max_rows,
2707 relationships: Arc::clone(&relationships),
2708 names: Arc::clone(&names),
2709 call: Some(Arc::new(FunctionCall {
2710 schema: function.schema_name.clone(),
2711 name: function.function_name.clone(),
2712 arguments: function.arguments.clone(),
2713 session_argument: function.session_argument.clone(),
2714 })),
2715 });
2716 let mut aggregate = Field::new(
2717 format!("{}_aggregate", function.name),
2718 TypeRef::named_nn(crate::schema::aggregate::aggregate_type_name(
2719 &function.returns,
2720 )),
2721 move |ctx| {
2722 let agg_spec = Arc::clone(&agg_spec);
2723 FieldFuture::new(async move { resolve_aggregate(&ctx, &agg_spec).await })
2724 },
2725 );
2726 if !function.arguments.is_empty() {
2727 aggregate = aggregate.argument(InputValue::new(
2728 "args",
2729 TypeRef::named_nn(format!("{}_args", function.name)),
2730 ));
2731 }
2732 aggregate = with_row_arguments(aggregate, &function.returns).description(format!(
2733 "fetch aggregated fields from the table: \"{}\"",
2734 function.returns_table.1
2735 ));
2736 root = root.field(aggregate);
2737 }
2738 root
2739}
2740
2741fn function_source(
2749 ctx: &ResolverContext<'_>,
2750 call: &FunctionCall,
2751 alias: &str,
2752 bound_values: &mut Vec<serde_json::Value>,
2753) -> Result<String, async_graphql::Error> {
2754 let given = ctx
2755 .args
2756 .try_get("args")
2757 .ok()
2758 .map(|v| accessor_to_json(&v))
2759 .unwrap_or_else(|| serde_json::json!({}));
2760 let mut passed = Vec::new();
2761 for (name, _, required) in &call.arguments {
2762 match given.get(name) {
2763 None if *required => {
2764 return Err(async_graphql::Error::new(format!(
2765 "{} needs the argument \"{}\"",
2766 call.name, name
2767 )))
2768 }
2769 None => continue,
2772 Some(value) => {
2773 let (_, pg_type, _) = call
2780 .arguments
2781 .iter()
2782 .find(|(argument, _, _)| argument == name)
2783 .expect("iterating the arguments");
2784 passed.push(format!(
2785 "{} => ${}::{}",
2786 postrust_sql::escape_ident(name),
2787 bound_values.len() + 1,
2788 pg_type
2789 ));
2790 bound_values.push(value.clone());
2791 }
2792 }
2793 }
2794 if let Some(session) = &call.session_argument {
2798 passed.push(format!(
2799 "{} => {}",
2800 postrust_sql::escape_ident(session),
2801 SESSION_ARGUMENT
2802 ));
2803 }
2804 Ok(format!(
2805 "{}.{}({}) AS {}",
2806 postrust_sql::escape_ident(&call.schema),
2807 postrust_sql::escape_ident(&call.name),
2808 passed.join(", "),
2809 postrust_sql::escape_ident(alias)
2810 ))
2811}
2812
2813struct FunctionCall {
2815 schema: String,
2816 name: String,
2817 arguments: Vec<(String, String, bool)>,
2818 session_argument: Option<String>,
2821}
2822
2823struct AggregateSpec {
2825 schema_name: String,
2826 table_name: String,
2827 type_name: String,
2828 max_rows: Option<i64>,
2829 relationships: Arc<HashMap<String, Vec<RelationshipField>>>,
2830 names: Arc<crate::names::NameOverrides>,
2831 call: Option<Arc<FunctionCall>>,
2835}
2836
2837async fn resolve_aggregate<'a>(
2844 ctx: &ResolverContext<'a>,
2845 spec: &AggregateSpec,
2846) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
2847 Ok(Some(FieldValue::value(aggregate_value(ctx, spec).await?)))
2848}
2849
2850async fn aggregate_value(
2855 ctx: &ResolverContext<'_>,
2856 spec: &AggregateSpec,
2857) -> Result<Value, async_graphql::Error> {
2858 use crate::schema::aggregate as agg;
2859
2860 let pool = ctx.data::<PgPool>()?;
2861 let gql_ctx = ctx.data::<GraphQLContext>()?;
2862
2863 let mut bound_values: Vec<serde_json::Value> = Vec::new();
2864 let source = match &spec.call {
2868 None => format!(
2869 "{}.{}",
2870 postrust_sql::escape_ident(&spec.schema_name),
2871 postrust_sql::escape_ident(&spec.table_name)
2872 ),
2873 Some(call) => function_source(ctx, call, &spec.table_name, &mut bound_values)?,
2874 };
2875
2876 let mut where_sql = String::new();
2877 {
2878 let requested = ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v));
2882 let permission = permission_predicate(
2883 &gql_ctx.caller(),
2884 spec.names.as_ref(),
2885 &spec.schema_name,
2886 &spec.table_name,
2887 )?;
2888 if let Some(predicate) = and_predicates(requested, permission) {
2889 let guard = gql_ctx
2890 .schema_cache
2891 .get()
2892 .await
2893 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
2894 let cache = guard
2895 .as_ref()
2896 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
2897 let scope = match &spec.call {
2904 None => WhereScope::table(
2905 &spec.schema_name,
2906 &spec.table_name,
2907 &spec.type_name,
2908 spec.names.as_ref(),
2909 )
2910 .with_resolution(cache, spec.relationships.as_ref()),
2911 Some(_) => WhereScope::for_alias(
2912 &spec.schema_name,
2913 &spec.table_name,
2914 &spec.table_name,
2915 &spec.type_name,
2916 cache,
2917 spec.relationships.as_ref(),
2918 spec.names.as_ref(),
2919 ),
2920 }
2921 .for_caller(gql_ctx.caller());
2922 let (sql, values) =
2923 build_where_clause(Some(&predicate), bound_values.len() + 1, &scope)?;
2924 if !sql.is_empty() {
2925 where_sql = format!(" {}", sql);
2926 bound_values.extend(values);
2927 }
2928 }
2929 }
2930
2931 let order_sql = build_order_by_clause(
2932 ctx,
2933 &gql_ctx.schema_cache,
2934 &spec.schema_name,
2935 &spec.table_name,
2936 &spec.type_name,
2937 spec.relationships.as_ref(),
2938 &format!(
2939 "{}.{}",
2940 postrust_sql::escape_ident(&spec.schema_name),
2941 postrust_sql::escape_ident(&spec.table_name)
2942 ),
2943 spec.names.as_ref(),
2944 )
2945 .await?;
2946
2947 let requested_limit = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok());
2948 let offset = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok());
2949 let ceiling = match (
2950 spec.max_rows,
2951 permission_limit(
2952 &gql_ctx.caller(),
2953 spec.names.as_ref(),
2954 &spec.schema_name,
2955 &spec.table_name,
2956 ),
2957 ) {
2958 (Some(configured), Some(granted)) => Some(configured.min(granted)),
2959 (only, None) | (None, only) => only,
2960 };
2961 let limit = match (requested_limit, ceiling) {
2962 (Some(requested), Some(ceiling)) => Some(requested.min(ceiling)),
2963 (Some(requested), None) => Some(requested),
2964 (None, ceiling) => ceiling,
2965 };
2966
2967 let rows = format!("SELECT * FROM {}{}{}", source, where_sql, order_sql);
2968
2969 let mut inner = rows.clone();
2971 if let Some(limit) = limit {
2972 inner.push_str(&format!(" LIMIT {}", limit));
2973 }
2974 if let Some(offset) = offset {
2975 inner.push_str(&format!(" OFFSET {}", offset));
2976 }
2977
2978 let mut counted = rows;
2992 if let Some(limit) = requested_limit {
2993 counted.push_str(&format!(" LIMIT {}", limit));
2994 }
2995 if let Some(offset) = offset {
2996 counted.push_str(&format!(" OFFSET {}", offset));
2997 }
2998
2999 let mut wants_nodes = false;
3004 let mut wanted: BTreeMap<String, Vec<String>> = BTreeMap::new();
3005 let mut counts: Vec<(String, String)> = Vec::new();
3008 for selection in ctx.field().selection_set() {
3009 match selection.name() {
3010 "nodes" => wants_nodes = true,
3011 "aggregate" => {
3012 for function in selection.selection_set() {
3013 if function.name() == "count" {
3014 counts.push((
3015 function.alias().unwrap_or("count").to_string(),
3016 count_expression(
3017 function,
3018 (&spec.schema_name, &spec.table_name),
3019 None,
3020 spec.names.as_ref(),
3021 ),
3022 ));
3023 continue;
3024 }
3025 let columns = wanted.entry(function.name().to_string()).or_default();
3026 for column in function.selection_set() {
3027 let name = column.name().to_string();
3028 if !columns.contains(&name) {
3029 columns.push(name);
3030 }
3031 }
3032 }
3033 }
3034 _ => {}
3035 }
3036 }
3037 wanted.retain(|_, columns| !columns.is_empty());
3038
3039 let mut result = async_graphql::indexmap::IndexMap::new();
3040
3041 if !counts.is_empty() || !wanted.is_empty() {
3042 let mut parts = match counts.iter().any(|(key, _)| key == "count") {
3046 true => Vec::new(),
3047 false => vec!["'count', count(*)".to_string()],
3048 };
3049 for (key, expression) in &counts {
3050 parts.push(format!("'{}', {}", key.replace('\'', "''"), expression));
3051 }
3052 for (function, columns) in &wanted {
3053 let sql_function = agg::NUMERIC_AGGREGATES
3057 .iter()
3058 .chain(agg::ORDERED_AGGREGATES.iter())
3059 .find(|(name, _)| name == function)
3060 .map(|(name, _)| *name)
3061 .ok_or_else(|| {
3062 async_graphql::Error::new(format!("unknown aggregate \"{}\"", function))
3063 })?;
3064 let per_column: Vec<String> = columns
3065 .iter()
3066 .map(|field| {
3067 let source = spec
3068 .names
3069 .column_source(&spec.schema_name, &spec.table_name, field)
3070 .unwrap_or(field);
3071 format!(
3072 "'{}', {}({})",
3073 field.replace('\'', "''"),
3074 sql_function,
3075 postrust_sql::escape_ident(source)
3076 )
3077 })
3078 .collect();
3079 parts.push(format!(
3080 "'{}', json_build_object({})",
3081 sql_function,
3082 per_column.join(", ")
3083 ));
3084 }
3085
3086 let sql = format!(
3087 "SELECT json_build_object({}) FROM ({}) AS pgrst_agg",
3088 parts.join(", "),
3089 counted
3090 );
3091 let mut conn =
3092 begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?;
3093 let rows = execute_query_on(&mut conn, &sql, &bound_values).await?;
3094 conn.commit().await?;
3095 if let Some(first) = rows.into_iter().next() {
3096 if let Value::Object(map) = json_to_value(first) {
3097 result.insert(async_graphql::Name::new("aggregate"), Value::Object(map));
3098 }
3099 }
3100 }
3101
3102 if wants_nodes {
3103 let nodes = ctx
3109 .field()
3110 .selection_set()
3111 .find(|selection| selection.name() == "nodes")
3112 .ok_or_else(|| async_graphql::Error::new("nodes was asked for and then was not"))?;
3113
3114 let mut node_values = bound_values.clone();
3118 let (projection, row_column_types) = {
3119 let guard = gql_ctx
3120 .schema_cache
3121 .get()
3122 .await
3123 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3124 let cache = guard
3125 .as_ref()
3126 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3127 let qi = postrust_core::api_request::QualifiedIdentifier::new(
3128 &spec.schema_name,
3129 &spec.table_name,
3130 );
3131 let table = cache.get_table(&qi);
3132 let mut param_idx = node_values.len() + 1;
3133 let computed = match table {
3134 Some(table) => computed_projections(
3135 table,
3136 nodes,
3137 "src",
3138 spec.names.as_ref(),
3139 cache,
3140 &mut param_idx,
3141 &mut node_values,
3142 )?,
3143 None => Vec::new(),
3144 };
3145 let embeds = build_embed_expressions(
3146 &gql_ctx.caller(),
3147 cache,
3148 spec.relationships.as_ref(),
3149 &spec.type_name,
3150 "src",
3151 nodes,
3152 spec.max_rows,
3153 &mut 0,
3154 &mut param_idx,
3155 &mut node_values,
3156 spec.names.as_ref(),
3157 )?;
3158 let mut projection = table
3162 .and_then(|table| rename_projection(table, "src", spec.names.as_ref()))
3163 .unwrap_or_else(|| "src.*".to_string());
3164 for expression in &computed {
3165 projection.push_str(", ");
3166 projection.push_str(expression);
3167 }
3168 for (field_name, expression) in &embeds {
3169 projection.push_str(", ");
3170 projection.push_str(expression);
3171 projection.push_str(" AS ");
3172 projection.push_str(&postrust_sql::escape_ident(field_name));
3173 }
3174 (
3175 projection,
3176 table
3177 .map(|table| exposed_column_types(table, spec.names.as_ref()))
3178 .unwrap_or_default(),
3179 )
3180 };
3181
3182 let sql = format!(
3183 "SELECT {} FROM (SELECT {} FROM ({}) AS src) AS pgrst_nodes",
3184 row_json("pgrst_nodes", &row_column_types),
3185 projection,
3186 inner
3187 );
3188 let mut conn =
3189 begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?;
3190 let rows = execute_query_on(&mut conn, &sql, &node_values).await?;
3191 conn.commit().await?;
3192 result.insert(
3193 async_graphql::Name::new("nodes"),
3194 Value::List(rows.into_iter().map(json_to_value).collect()),
3195 );
3196 }
3197
3198 let _ = &spec.type_name;
3199 Ok(Value::Object(result))
3200}
3201
3202async fn resolve_query<'a>(
3204 ctx: &ResolverContext<'a>,
3205 spec: &QueryFieldSpec,
3206) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
3207 Ok(rows_as_field_value(
3208 query_rows(ctx, spec).await?,
3209 spec.is_by_pk,
3210 ))
3211}
3212
3213async fn query_rows(
3219 ctx: &ResolverContext<'_>,
3220 spec: &QueryFieldSpec,
3221) -> Result<Vec<serde_json::Value>, async_graphql::Error> {
3222 let schema_name = spec.schema_name.as_str();
3223 let table_name = spec.table_name.as_str();
3224 let type_name = spec.type_name.as_str();
3225 let is_by_pk = spec.is_by_pk;
3226 let pk_columns = spec.pk_columns.as_slice();
3227 let max_rows = spec.max_rows;
3228 let relationships = spec.relationships.as_ref();
3229
3230 let pool = ctx.data::<PgPool>()?;
3231 let gql_ctx = ctx.data::<GraphQLContext>()?;
3232
3233 debug!("Resolving query for table: {}", table_name);
3234
3235 let requested_limit: Option<i64> = ctx.args.try_get("limit").ok().and_then(|v| v.i64().ok());
3237
3238 let offset: Option<i64> = ctx.args.try_get("offset").ok().and_then(|v| v.i64().ok());
3239
3240 let limit: Option<i64> = if is_by_pk {
3245 Some(1)
3246 } else {
3247 let ceiling = match (
3251 max_rows,
3252 permission_limit(
3253 &gql_ctx.caller(),
3254 spec.names.as_ref(),
3255 schema_name,
3256 table_name,
3257 ),
3258 ) {
3259 (Some(configured), Some(granted)) => Some(configured.min(granted)),
3260 (only, None) | (None, only) => only,
3261 };
3262 match (requested_limit, ceiling) {
3263 (Some(requested), Some(ceiling)) => Some(requested.min(ceiling)),
3264 (Some(requested), None) => Some(requested),
3265 (None, ceiling) => ceiling,
3266 }
3267 };
3268
3269 let mut where_sql = String::new();
3277 let mut bound_values: Vec<serde_json::Value> = Vec::new();
3278
3279 let source = match &spec.call {
3283 None => format!(
3289 "{}.{} AS {}",
3290 postrust_sql::escape_ident(schema_name),
3291 postrust_sql::escape_ident(table_name),
3292 postrust_sql::escape_ident(READ_ROW)
3293 ),
3294 Some(call) => function_source(ctx, call, table_name, &mut bound_values)?,
3295 };
3296
3297 let source_ref = match &spec.call {
3300 None => postrust_sql::escape_ident(READ_ROW),
3301 Some(_) => postrust_sql::escape_ident(table_name),
3302 };
3303
3304 if is_by_pk {
3305 if pk_columns.is_empty() {
3306 return Err(async_graphql::Error::new(format!(
3307 "\"{}\" has no primary key, so it cannot be queried by key",
3308 table_name
3309 )));
3310 }
3311
3312 let mut conditions = Vec::with_capacity(pk_columns.len());
3313 for (idx, (col_name, pg_type)) in pk_columns.iter().enumerate() {
3314 let field = spec
3315 .names
3316 .column(schema_name, table_name, col_name)
3317 .unwrap_or(col_name);
3318 let value = ctx.args.try_get(field).map_err(|_| {
3319 async_graphql::Error::new(format!(
3320 "missing required primary key argument \"{}\"",
3321 field
3322 ))
3323 })?;
3324
3325 conditions.push(format!(
3326 "{} = ${}::{}",
3327 postrust_sql::escape_ident(col_name),
3328 idx + 1,
3329 pg_type
3330 ));
3331 bound_values.push(accessor_to_json(&value));
3332 }
3333 where_sql = format!(" WHERE {}", conditions.join(" AND "));
3334 }
3335
3336 {
3341 let requested = match is_by_pk {
3342 true => None,
3343 false => ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v)),
3344 };
3345 let permission = permission_predicate(
3346 &gql_ctx.caller(),
3347 spec.names.as_ref(),
3348 schema_name,
3349 table_name,
3350 )?;
3351
3352 if let Some(predicate) = and_predicates(requested, permission) {
3353 let guard = gql_ctx
3354 .schema_cache
3355 .get()
3356 .await
3357 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3358 let cache = guard
3359 .as_ref()
3360 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3361 let scope = match &spec.call {
3362 None => WhereScope::table(schema_name, table_name, type_name, spec.names.as_ref())
3363 .under_alias(READ_ROW)
3364 .with_resolution(cache, relationships)
3365 .for_caller(gql_ctx.caller()),
3366 Some(_) => WhereScope::for_alias(
3369 schema_name,
3370 table_name,
3371 table_name,
3372 type_name,
3373 cache,
3374 relationships,
3375 spec.names.as_ref(),
3376 )
3377 .for_caller(gql_ctx.caller()),
3378 };
3379 let (filter_sql, filter_values) =
3380 build_where_clause(Some(&predicate), bound_values.len() + 1, &scope)?;
3381 if !filter_sql.is_empty() {
3382 where_sql = match where_sql.is_empty() {
3385 true => format!(" {}", filter_sql),
3386 false => format!(
3387 "{} AND ({})",
3388 where_sql,
3389 filter_sql.trim_start_matches("WHERE ")
3390 ),
3391 };
3392 bound_values.extend(filter_values);
3393 }
3394 }
3395 }
3396
3397 let (order_sql, distinct_on) = if is_by_pk {
3400 (String::new(), Vec::new())
3401 } else {
3402 (
3403 build_order_by_clause(
3404 ctx,
3405 &gql_ctx.schema_cache,
3406 schema_name,
3407 table_name,
3408 type_name,
3409 relationships,
3410 &source_ref,
3411 spec.names.as_ref(),
3412 )
3413 .await?,
3414 build_distinct_on(
3415 ctx,
3416 &gql_ctx.schema_cache,
3417 schema_name,
3418 table_name,
3419 spec.names.as_ref(),
3420 )
3421 .await?,
3422 )
3423 };
3424
3425 let (distinct_sql, order_sql) = {
3433 let written = order_sql.strip_prefix(" ORDER BY ");
3434 let (prefix, order) = distinct_on_clause(&distinct_on, written).map_err(|error| {
3437 at_path(
3438 error,
3439 &format!(
3440 "$.selectionSet.{}.args",
3441 ctx.ctx.field().alias().unwrap_or(ctx.ctx.field().name())
3442 ),
3443 )
3444 })?;
3445 (
3446 prefix,
3447 match order {
3448 Some(order) => format!(" ORDER BY {}", order),
3449 None => String::new(),
3450 },
3451 )
3452 };
3453
3454 let (computed, row_column_types) = {
3457 let guard = gql_ctx
3458 .schema_cache
3459 .get()
3460 .await
3461 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3462 match guard.as_ref() {
3463 Some(cache) => {
3464 let qi =
3465 postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
3466 match cache.get_table(&qi) {
3467 Some(table) => {
3468 let mut param_idx = bound_values.len() + 1;
3469 (
3470 computed_projections(
3471 table,
3472 ctx.field(),
3473 "src",
3474 spec.names.as_ref(),
3475 cache,
3476 &mut param_idx,
3477 &mut bound_values,
3478 )?,
3479 exposed_column_types(table, spec.names.as_ref()),
3480 )
3481 }
3482 None => (Vec::new(), HashMap::new()),
3483 }
3484 }
3485 None => (Vec::new(), HashMap::new()),
3486 }
3487 };
3488 let mut inner = format!(
3492 "SELECT {}* FROM {}{}{}",
3493 distinct_sql, source, where_sql, order_sql
3494 );
3495
3496 if let Some(limit) = limit {
3497 inner.push_str(&format!(" LIMIT {}", limit));
3498 }
3499
3500 if let Some(offset) = offset {
3501 inner.push_str(&format!(" OFFSET {}", offset));
3502 }
3503
3504 let embed_expressions = {
3507 let guard = gql_ctx
3508 .schema_cache
3509 .get()
3510 .await
3511 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3512 match guard.as_ref() {
3513 Some(cache) => {
3514 let mut param_idx = bound_values.len() + 1;
3519 build_embed_expressions(
3520 &gql_ctx.caller(),
3521 cache,
3522 relationships,
3523 type_name,
3524 "src",
3525 ctx.field(),
3526 max_rows,
3527 &mut 0,
3528 &mut param_idx,
3529 &mut bound_values,
3530 spec.names.as_ref(),
3531 )?
3532 }
3533 None => Vec::new(),
3534 }
3535 };
3536
3537 let renamed = {
3545 let guard = gql_ctx
3546 .schema_cache
3547 .get()
3548 .await
3549 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3550 guard
3551 .as_ref()
3552 .and_then(|cache| {
3553 cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
3554 schema_name,
3555 table_name,
3556 ))
3557 })
3558 .and_then(|table| rename_projection(table, "src", spec.names.as_ref()))
3559 };
3560 let inner = if embed_expressions.is_empty() && computed.is_empty() && renamed.is_none() {
3561 inner
3562 } else {
3563 let mut projection = renamed.unwrap_or_else(|| "src.*".to_string());
3564 for expression in &computed {
3565 projection.push_str(", ");
3566 projection.push_str(expression);
3567 }
3568 for (field_name, expression) in &embed_expressions {
3569 projection.push_str(", ");
3570 projection.push_str(expression);
3571 projection.push_str(" AS ");
3572 projection.push_str(&postrust_sql::escape_ident(field_name));
3573 }
3574 format!("SELECT {} FROM ({}) AS src", projection, inner)
3575 };
3576
3577 let sql = format!(
3578 "SELECT {} FROM ({}) t",
3579 row_json("t", &row_column_types),
3580 inner
3581 );
3582
3583 let mut tx = begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?;
3587
3588 let mut result = execute_query_on(&mut tx, &sql, &bound_values).await?;
3590
3591 if embed_expressions.is_empty() {
3593 let guard = gql_ctx
3594 .schema_cache
3595 .get()
3596 .await
3597 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3598 let cache = guard
3599 .as_ref()
3600 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3601
3602 let embed_ctx = EmbedContext {
3603 schema_cache: cache,
3604 relationships,
3605 max_rows,
3606 };
3607
3608 embed_relationships(&mut tx, &embed_ctx, type_name, ctx.field(), &mut result).await?;
3609 }
3610
3611 tx.commit().await?;
3612
3613 Ok(result)
3614}
3615
3616fn rows_as_field_value<'a>(rows: Vec<serde_json::Value>, is_by_pk: bool) -> Option<FieldValue<'a>> {
3621 match is_by_pk {
3622 true => rows
3623 .into_iter()
3624 .next()
3625 .map(|v| FieldValue::value(json_to_value(v))),
3626 false => Some(FieldValue::list(
3627 rows.into_iter()
3628 .map(|v| FieldValue::value(json_to_value(v))),
3629 )),
3630 }
3631}
3632
3633#[allow(clippy::too_many_arguments)]
3635async fn resolve_mutation<'a>(
3636 ctx: &ResolverContext<'a>,
3637 schema_name: &str,
3638 table_name: &str,
3639 mutation_type: MutationType,
3640 pk_columns: &[(String, String)],
3641 relationships: &Arc<HashMap<String, Vec<RelationshipField>>>,
3642 type_names: &Arc<HashMap<(String, String), String>>,
3643 names: &Arc<crate::names::NameOverrides>,
3644 max_rows: Option<i64>,
3645) -> Result<Option<FieldValue<'a>>, async_graphql::Error> {
3646 let pool = ctx.data::<PgPool>()?;
3647 let gql_ctx = ctx.data::<GraphQLContext>()?;
3648
3649 debug!(
3650 "Resolving mutation for table: {} type: {:?}",
3651 table_name, mutation_type
3652 );
3653
3654 let type_name = type_names
3660 .get(&(schema_name.to_string(), table_name.to_string()))
3661 .cloned()
3662 .unwrap_or_else(|| table_name.to_string());
3663
3664 let by_key = matches!(
3665 mutation_type,
3666 MutationType::InsertOne | MutationType::UpdateByPk | MutationType::DeleteByPk
3667 );
3668 let returning = if by_key {
3669 Some(ctx.field())
3670 } else {
3671 ctx.field()
3672 .selection_set()
3673 .find(|selection| selection.name() == "returning")
3674 };
3675
3676 if mutation_type == MutationType::UpdateMany {
3681 let updates = ctx
3682 .args
3683 .try_get("updates")
3684 .ok()
3685 .map(|v| accessor_to_json(&v))
3686 .unwrap_or(serde_json::Value::Null);
3687 let serde_json::Value::Array(entries) = updates else {
3688 return Err(async_graphql::Error::new(
3689 "\"updates\" takes a list of {where, _set, …} objects",
3690 ));
3691 };
3692 let column_types = column_types_of(&gql_ctx.schema_cache, schema_name, table_name).await;
3693 let mut answers: Vec<FieldValue<'_>> = Vec::with_capacity(entries.len());
3694 for entry in entries {
3695 let serde_json::Value::Object(entry) = entry else {
3696 return Err(async_graphql::Error::new(
3697 "each update is an object of {where, _set, …}",
3698 ));
3699 };
3700 let operators: Vec<(&'static str, serde_json::Value)> = UPDATE_OPERATORS
3701 .iter()
3702 .filter_map(|name| entry.get(*name).map(|value| (*name, value.clone())))
3703 .filter(|(_, value)| !value.is_null())
3704 .collect();
3705 let rows = execute_update(
3706 pool,
3707 gql_ctx,
3708 schema_name,
3709 table_name,
3710 &type_name,
3711 operators,
3712 column_types.clone(),
3713 entry.get("where").cloned(),
3714 relationships.as_ref(),
3715 names.as_ref(),
3716 )
3717 .await?;
3718 let affected = rows.len();
3719 let rows = match returning {
3720 Some(returning) if !rows.is_empty() => {
3721 reread_returning(
3722 pool,
3723 gql_ctx,
3724 schema_name,
3725 table_name,
3726 &type_name,
3727 rows,
3728 returning,
3729 relationships.as_ref(),
3730 names.as_ref(),
3731 max_rows,
3732 )
3733 .await?
3734 }
3735 _ => rows,
3736 };
3737 answers.extend(mutation_result(rows, affected, false));
3738 }
3739 return Ok(Some(FieldValue::list(answers)));
3740 }
3741
3742 let (result, affected) = match mutation_type {
3743 MutationType::Insert | MutationType::InsertOne => {
3744 let objects = ctx
3745 .args
3746 .try_get("objects")
3747 .ok()
3748 .map(|v| accessor_to_json(&v))
3749 .or_else(|| {
3750 ctx.args
3753 .try_get("object")
3754 .ok()
3755 .map(|v| serde_json::Value::Array(vec![accessor_to_json(&v)]))
3756 })
3757 .unwrap_or_else(|| serde_json::Value::Array(vec![]));
3758
3759 let guard = gql_ctx
3760 .schema_cache
3761 .get()
3762 .await
3763 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
3764 let cache = guard
3765 .as_ref()
3766 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
3767 let here = format!(
3771 "$.selectionSet.{}",
3772 ctx.ctx.field().alias().unwrap_or(ctx.ctx.field().name())
3773 );
3774 let named = match mutation_type {
3777 MutationType::InsertOne => "object",
3778 _ => "objects",
3779 };
3780 let context = InsertContext {
3781 on_conflict: ctx
3782 .args
3783 .try_get("on_conflict")
3784 .ok()
3785 .map(|v| accessor_to_json(&v))
3786 .filter(|v| !v.is_null()),
3787 cache,
3788 relationships: relationships.as_ref(),
3789 type_names: type_names.as_ref(),
3790 names: names.as_ref(),
3791 caller: gql_ctx.caller(),
3792 row: String::new(),
3795 objects: format!("{}.args.{}", here, named),
3796 conflict: format!("{}.args.on_conflict", here),
3797 };
3798
3799 execute_insert(
3800 pool,
3801 gql_ctx,
3802 schema_name,
3803 table_name,
3804 objects,
3805 &context,
3806 mutation_type == MutationType::InsertOne,
3807 )
3808 .await?
3809 }
3810 MutationType::UpdateMany | MutationType::Update | MutationType::UpdateByPk => {
3813 let operators: Vec<(&'static str, serde_json::Value)> = UPDATE_OPERATORS
3817 .iter()
3818 .filter_map(|name| {
3819 ctx.args
3820 .try_get(name)
3821 .ok()
3822 .map(|v| (*name, accessor_to_json(&v)))
3823 })
3824 .filter(|(_, value)| !value.is_null())
3825 .collect();
3826
3827 let where_clause = if mutation_type == MutationType::UpdateByPk {
3828 Some(pk_where_from_args(
3829 ctx,
3830 schema_name,
3831 table_name,
3832 pk_columns,
3833 names.as_ref(),
3834 )?)
3835 } else {
3836 ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v))
3837 };
3838 let where_clause = and_predicates(
3842 where_clause,
3843 permission_filter(
3844 &gql_ctx.caller(),
3845 names.as_ref(),
3846 schema_name,
3847 table_name,
3848 crate::role::Verb::Update,
3849 )?,
3850 );
3851
3852 execute_update(
3853 pool,
3854 gql_ctx,
3855 schema_name,
3856 table_name,
3857 &type_name,
3858 operators,
3859 column_types_of(&gql_ctx.schema_cache, schema_name, table_name).await,
3860 where_clause,
3861 relationships.as_ref(),
3862 names.as_ref(),
3863 )
3864 .await
3865 .map(|rows| {
3866 let count = rows.len();
3867 (rows, count)
3868 })?
3869 }
3870 MutationType::Delete | MutationType::DeleteByPk => {
3871 let where_clause = if mutation_type == MutationType::DeleteByPk {
3872 Some(pk_where_from_args(
3873 ctx,
3874 schema_name,
3875 table_name,
3876 pk_columns,
3877 names.as_ref(),
3878 )?)
3879 } else {
3880 ctx.args.try_get("where").ok().map(|v| accessor_to_json(&v))
3881 };
3882 let where_clause = and_predicates(
3883 where_clause,
3884 permission_filter(
3885 &gql_ctx.caller(),
3886 names.as_ref(),
3887 schema_name,
3888 table_name,
3889 crate::role::Verb::Delete,
3890 )?,
3891 );
3892
3893 execute_delete(
3894 pool,
3895 gql_ctx,
3896 schema_name,
3897 table_name,
3898 &type_name,
3899 where_clause,
3900 returning,
3901 relationships.as_ref(),
3902 names.as_ref(),
3903 max_rows,
3904 )
3905 .await
3906 .map(|rows| {
3907 let count = rows.len();
3908 (rows, count)
3909 })?
3910 }
3911 };
3912
3913 let is_delete = matches!(
3914 mutation_type,
3915 MutationType::Delete | MutationType::DeleteByPk
3916 );
3917 let result = match returning {
3918 Some(returning) if !result.is_empty() && !is_delete => {
3919 reread_returning(
3920 pool,
3921 gql_ctx,
3922 schema_name,
3923 table_name,
3924 &type_name,
3925 result,
3926 returning,
3927 relationships.as_ref(),
3928 names.as_ref(),
3929 max_rows,
3930 )
3931 .await?
3932 }
3933 _ => result,
3934 };
3935
3936 Ok(mutation_result(result, affected, by_key))
3937}
3938
3939async fn write_tx<'a>(
3947 gql_ctx: &'a GraphQLContext,
3948 pool: &PgPool,
3949) -> Result<
3950 tokio::sync::MutexGuard<'a, Option<sqlx::Transaction<'static, sqlx::Postgres>>>,
3951 async_graphql::Error,
3952> {
3953 let mut guard = gql_ctx.write.lock().await;
3954 if guard.is_none() {
3955 *guard = Some(begin_with_session(pool, gql_ctx.role(), &gql_ctx.session_settings()).await?);
3956 }
3957 Ok(guard)
3958}
3959
3960async fn begin_with_session(
3966 pool: &PgPool,
3967 role: &str,
3968 settings: &[(String, String)],
3969) -> Result<sqlx::Transaction<'static, sqlx::Postgres>, async_graphql::Error> {
3970 let mut tx = pool.begin().await?;
3971
3972 if !settings.is_empty() {
3982 let calls = (0..settings.len())
3983 .map(|i| format!("set_config(${}, ${}, true)", i * 2 + 1, i * 2 + 2))
3984 .collect::<Vec<_>>()
3985 .join(", ");
3986 let sql = format!("SELECT {calls}");
3987 let mut query = sqlx::query_scalar::<_, String>(&sql);
3988 for (name, value) in settings {
3989 query = query.bind(name).bind(value);
3990 }
3991 query.fetch_optional(&mut *tx).await?;
3992 }
3993
3994 sqlx::query(&format!(
3995 "SET LOCAL ROLE {}",
3996 postrust_sql::escape_ident(role)
3997 ))
3998 .execute(&mut *tx)
3999 .await?;
4000 Ok(tx)
4001}
4002
4003fn row_json(expr: &str, column_types: &HashMap<String, String>) -> String {
4014 let mut shapes: Vec<(&String, &String)> = column_types
4015 .iter()
4016 .filter(|(_, pg_type)| matches!(pg_type.as_str(), "geometry" | "geography"))
4017 .collect();
4018 if shapes.is_empty() {
4019 return format!("row_to_json({})", expr);
4020 }
4021 shapes.sort();
4022
4023 let overrides: Vec<String> = shapes
4024 .iter()
4025 .map(|(name, _)| {
4026 format!(
4027 "'{}', ST_AsGeoJSON({}.{}, 9, 4)::jsonb",
4033 name.replace('\'', "''"),
4034 expr,
4035 postrust_sql::escape_ident(name)
4036 )
4037 })
4038 .collect();
4039 format!(
4040 "(to_jsonb({}) || jsonb_build_object({}))",
4041 expr,
4042 overrides.join(", ")
4043 )
4044}
4045
4046async fn execute_query_on(
4049 conn: &mut sqlx::PgConnection,
4050 sql: &str,
4051 params: &[serde_json::Value],
4052) -> Result<Vec<serde_json::Value>, async_graphql::Error> {
4053 use sqlx::Row;
4054
4055 trace!("Executing SQL: {}", sql);
4056
4057 let mut query = sqlx::query(sql);
4059 for param in params {
4060 query = bind_json_value(query, param);
4061 }
4062 let rows = query.fetch_all(&mut *conn).await.map_err(database_error)?;
4063
4064 let results: Vec<serde_json::Value> = rows
4067 .into_iter()
4068 .filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
4069 .collect();
4070
4071 Ok(results)
4072}
4073
4074fn check_geojson(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
4087 fn refuse<T>(message: &str) -> Result<T, async_graphql::Error> {
4088 Err(coded_error("parse-failed", message))
4089 }
4090
4091 fn position(
4092 value: &serde_json::Value,
4093 ) -> Result<&Vec<serde_json::Value>, async_graphql::Error> {
4094 match value.as_array() {
4095 Some(items) if items.len() >= 2 && items.iter().all(|i| i.is_number()) => Ok(items),
4096 _ => refuse("A Position needs at least 2 elements"),
4097 }
4098 }
4099
4100 fn positions(
4101 value: Option<&serde_json::Value>,
4102 least: usize,
4103 what: &str,
4104 ) -> Result<Vec<Vec<serde_json::Value>>, async_graphql::Error> {
4105 let items = match value.and_then(|v| v.as_array()) {
4106 Some(items) => items,
4107 None => return refuse(&format!("A {} needs at least {} Positions", what, least)),
4108 };
4109 if items.len() < least {
4110 return refuse(&format!("A {} needs at least {} Positions", what, least));
4111 }
4112 items.iter().map(|item| position(item).cloned()).collect()
4113 }
4114
4115 fn ring(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
4116 let points = positions(Some(value), 4, "LinearRing")?;
4117 match (points.first(), points.last()) {
4118 (Some(first), Some(last)) if first == last => Ok(()),
4119 _ => refuse("the first and last locations have to be equal for a LinearRing"),
4120 }
4121 }
4122
4123 fn polygon(value: &serde_json::Value) -> Result<(), async_graphql::Error> {
4124 let Some(rings) = value.as_array() else {
4125 return refuse("A LinearRing needs at least 4 Positions");
4126 };
4127 rings.iter().try_for_each(ring)
4128 }
4129
4130 let serde_json::Value::Object(map) = value else {
4131 return Ok(());
4132 };
4133 let Some(kind) = map.get("type").and_then(|v| v.as_str()) else {
4134 return Ok(());
4135 };
4136 let coordinates = map.get("coordinates");
4137 match kind {
4138 "Point" => {
4139 position(coordinates.unwrap_or(&serde_json::Value::Null))?;
4140 }
4141 "MultiPoint" => {
4142 positions(coordinates, 0, "MultiPoint")?;
4143 }
4144 "LineString" => {
4145 positions(coordinates, 2, "LineString")?;
4146 }
4147 "MultiLineString" => {
4148 for line in coordinates.and_then(|v| v.as_array()).into_iter().flatten() {
4149 positions(Some(line), 2, "LineString")?;
4150 }
4151 }
4152 "Polygon" => polygon(coordinates.unwrap_or(&serde_json::Value::Null))?,
4153 "MultiPolygon" => {
4154 for one in coordinates.and_then(|v| v.as_array()).into_iter().flatten() {
4155 polygon(one)?;
4156 }
4157 }
4158 "GeometryCollection" => {
4159 for geometry in map
4160 .get("geometries")
4161 .and_then(|v| v.as_array())
4162 .into_iter()
4163 .flatten()
4164 {
4165 check_geojson(geometry)?;
4166 }
4167 }
4168 other => return refuse(&format!("unexpected geometry type: {}", other)),
4169 }
4170 Ok(())
4171}
4172
4173fn computed_call(
4181 definition: &postrust_core::schema_cache::ComputedColumn,
4182 row: &str,
4183 row_by_name: &str,
4184) -> String {
4185 let function = format!(
4186 "{}.{}",
4187 postrust_sql::escape_ident(&definition.function.schema),
4188 postrust_sql::escape_ident(&definition.function.name)
4189 );
4190 match (&definition.session_argument, &definition.row_argument) {
4191 (Some(session), Some(row_argument)) => format!(
4192 "{}({} => {}, {} => {})",
4193 function,
4194 postrust_sql::escape_ident(row_argument),
4195 row_by_name,
4196 postrust_sql::escape_ident(session),
4197 SESSION_ARGUMENT
4198 ),
4199 _ => format!("{}({})", function, row),
4200 }
4201}
4202
4203const SESSION_ARGUMENT: &str = "coalesce(current_setting('hasura.session', true), '{}')::json";
4209
4210const UPDATE_OPERATORS: [&str; 7] = [
4213 "_set",
4214 "_inc",
4215 "_append",
4216 "_prepend",
4217 "_delete_key",
4218 "_delete_elem",
4219 "_delete_at_path",
4220];
4221
4222const READ_ROW: &str = "pgrst_src";
4228
4229const WRITTEN_ROW: &str = "pgrst_row";
4234
4235fn write_cast(column_types: &HashMap<String, String>, column: &str) -> String {
4243 match column_types.get(column) {
4244 Some(pg_type) => format!("::{}", pg_type),
4245 None => String::new(),
4246 }
4247}
4248
4249fn write_operand(
4256 column_types: &HashMap<String, String>,
4257 column: &str,
4258 value: &serde_json::Value,
4259) -> serde_json::Value {
4260 match column_types.get(column).map(String::as_str) {
4261 Some("json") | Some("jsonb") => match value {
4262 serde_json::Value::String(_)
4263 | serde_json::Value::Number(_)
4264 | serde_json::Value::Bool(_) => serde_json::Value::String(value.to_string()),
4265 other => other.clone(),
4266 },
4267 Some(pg_type) if is_array_type(pg_type) => match value {
4272 serde_json::Value::Array(_) => serde_json::Value::String(array_literal(value)),
4273 other => other.clone(),
4274 },
4275 _ => value.clone(),
4276 }
4277}
4278
4279fn is_array_type(pg_type: &str) -> bool {
4284 pg_type.ends_with("[]") || pg_type.starts_with('_')
4285}
4286
4287fn array_literal(value: &serde_json::Value) -> String {
4293 fn element(value: &serde_json::Value) -> String {
4294 match value {
4295 serde_json::Value::Null => "NULL".to_string(),
4296 serde_json::Value::Array(_) => array_literal(value),
4297 serde_json::Value::String(text) => {
4298 format!("\"{}\"", text.replace('\\', "\\\\").replace('"', "\\\""))
4299 }
4300 other => format!("\"{}\"", other.to_string().replace('"', "\\\"")),
4301 }
4302 }
4303 let items = match value {
4304 serde_json::Value::Array(items) => items,
4305 other => return element(other),
4306 };
4307 format!(
4308 "{{{}}}",
4309 items.iter().map(element).collect::<Vec<_>>().join(",")
4310 )
4311}
4312
4313fn write_expression(
4324 column_types: &HashMap<String, String>,
4325 column: &str,
4326 value: &serde_json::Value,
4327 placeholder: &str,
4328) -> String {
4329 let pg_type = column_types.get(column).map(String::as_str);
4330 if value.is_object() && matches!(pg_type, Some("geometry") | Some("geography")) {
4331 return match pg_type {
4332 Some("geography") => format!("ST_GeomFromGeoJSON({})::geography", placeholder),
4333 _ => format!("ST_GeomFromGeoJSON({})", placeholder),
4334 };
4335 }
4336 if pg_type == Some("money") && value.is_number() {
4342 return format!("{}::numeric::money", placeholder);
4343 }
4344 format!("{}{}", placeholder, write_cast(column_types, column))
4345}
4346
4347async fn column_types_of(
4349 schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
4350 schema_name: &str,
4351 table_name: &str,
4352) -> HashMap<String, String> {
4353 let Ok(guard) = schema_cache.get().await else {
4354 return HashMap::new();
4355 };
4356 let Some(cache) = guard.as_ref() else {
4357 return HashMap::new();
4358 };
4359 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
4360 match cache.get_table(&qi) {
4361 Some(table) => table
4362 .columns
4363 .values()
4364 .map(|c| (c.name.clone(), c.nominal_type.clone()))
4365 .collect(),
4366 None => HashMap::new(),
4367 }
4368}
4369
4370fn insert_row<'life>(
4388 conn: &'life mut sqlx::PgConnection,
4389 schema_name: &'life str,
4390 table_name: &'life str,
4391 object: serde_json::Map<String, serde_json::Value>,
4392 context: &'life InsertContext<'life>,
4393 written_count: &'life mut usize,
4394) -> std::pin::Pin<
4395 Box<
4396 dyn std::future::Future<Output = Result<serde_json::Value, async_graphql::Error>>
4397 + Send
4398 + 'life,
4399 >,
4400> {
4401 Box::pin(async move {
4402 use sqlx::Row;
4403
4404 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
4405 let table = context.cache.get_table(&qi).ok_or_else(|| {
4406 async_graphql::Error::new(format!("unknown table \"{}\"", table_name))
4407 })?;
4408 let column_types: HashMap<String, String> = table
4409 .columns
4410 .values()
4411 .map(|c| (c.name.clone(), c.nominal_type.clone()))
4412 .collect();
4413
4414 let type_name = context
4416 .type_names
4417 .get(&(schema_name.to_string(), table_name.to_string()));
4418 let relationships: &[RelationshipField] = type_name
4419 .and_then(|name| context.relationships.get(name))
4420 .map(|r| r.as_slice())
4421 .unwrap_or(&[]);
4422
4423 let mut columns = serde_json::Map::new();
4424 type Nested<'r> = (
4425 &'r RelationshipField,
4426 serde_json::Value,
4427 Option<serde_json::Value>,
4428 );
4429 let mut to_one: Vec<Nested> = Vec::new();
4430 let mut to_many: Vec<Nested> = Vec::new();
4431
4432 for (key, value) in object {
4433 match relationships.iter().find(|r| r.name == key) {
4434 None => {
4435 columns.insert(
4438 table_column_for(context.names, table, &key).to_string(),
4439 value,
4440 );
4441 }
4442 Some(_) if value.is_null() => {
4443 continue;
4447 }
4448 Some(relationship) => {
4449 let (data, conflict) = match &value {
4453 serde_json::Value::Object(map) => (
4454 map.get("data").cloned().unwrap_or(value.clone()),
4455 map.get("on_conflict").cloned().filter(|v| !v.is_null()),
4456 ),
4457 other => (other.clone(), None),
4458 };
4459 if child_holds_the_key(&relationship.relationship) {
4466 to_many.push((relationship, data, conflict));
4467 } else {
4468 to_one.push((relationship, data, conflict));
4469 }
4470 }
4471 }
4472 }
4473
4474 for (column, value) in &columns {
4477 if matches!(
4478 column_types.get(column).map(String::as_str),
4479 Some("geometry") | Some("geography")
4480 ) {
4481 check_geojson(value).map_err(|error| {
4484 let field = context
4485 .names
4486 .column(schema_name, table_name, column)
4487 .unwrap_or(column);
4488 at_path(error, &format!("{}.{}", context.row, field))
4489 })?;
4490 }
4491 }
4492
4493 for (relationship, data, conflict) in to_one {
4496 let plan =
4497 postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, context.cache)
4498 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
4499 let serde_json::Value::Object(child) = data else {
4500 return Err(async_graphql::Error::new(format!(
4501 "\"{}\" takes an object to insert",
4502 relationship.name
4503 )));
4504 };
4505 for (local, _) in &plan.columns {
4509 if columns.get(local).is_some_and(|v| !v.is_null()) {
4510 return Err(at_path(
4511 validation_error(format!(
4512 "cannot insert object relationship \"{}\" as \"{}\" column \
4513 values are already determined",
4514 relationship.name, local
4515 )),
4516 &format!("{}.{}", context.row, relationship.name),
4517 ));
4518 }
4519 }
4520 let under = format!("{}.{}", context.row, relationship.name);
4523 let nested = InsertContext {
4524 on_conflict: conflict,
4525 cache: context.cache,
4526 relationships: context.relationships,
4527 type_names: context.type_names,
4528 names: context.names,
4529 caller: context.caller,
4530 row: format!("{}.data", under),
4531 objects: format!("{}.data", under),
4532 conflict: format!("{}.on_conflict", under),
4533 };
4534 let written = insert_row(
4535 conn,
4536 &plan.foreign_schema,
4537 &plan.foreign_table,
4538 child,
4539 &nested,
4540 written_count,
4541 )
4542 .await?;
4543 for (local, foreign) in &plan.columns {
4544 if let Some(value) = written.get(foreign) {
4545 columns.insert(local.clone(), value.clone());
4546 }
4547 }
4548 }
4549
4550 for (column, value) in crate::role::presets(
4556 &context.caller,
4557 context.names,
4558 schema_name,
4559 table_name,
4560 crate::role::Verb::Insert,
4561 )
4562 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?
4563 {
4564 columns.insert(column, value);
4565 }
4566
4567 let column_names: Vec<&str> = columns.keys().map(|k| k.as_str()).collect();
4573 let mut conflict_values: Vec<serde_json::Value> = Vec::new();
4574 let conflict_sql = match &context.on_conflict {
4575 Some(serde_json::Value::Object(spec)) => {
4576 let constraint = spec.get("constraint").and_then(|v| v.as_str());
4577 let Some(constraint) = constraint else {
4578 return Err(async_graphql::Error::new(
4579 "on_conflict needs a constraint to resolve against",
4580 ));
4581 };
4582 if !table
4583 .unique_constraints
4584 .iter()
4585 .any(|(name, _)| name == constraint)
4586 {
4587 return Err(async_graphql::Error::new(format!(
4588 "\"{}\" is not a unique constraint of \"{}\"",
4589 constraint, table_name
4590 )));
4591 }
4592 let updates: Vec<&str> = spec
4593 .get("update_columns")
4594 .and_then(|v| v.as_array())
4595 .map(|items| items.iter().filter_map(|i| i.as_str()).collect())
4596 .unwrap_or_default();
4597 if updates.contains(&PLACEHOLDER_COLUMN) {
4601 return Err(at_path(
4602 coded_error("validation-failed", "erroneous column name"),
4603 &context.conflict,
4604 ));
4605 }
4606
4607 if updates.is_empty() {
4608 format!(
4609 " ON CONFLICT ON CONSTRAINT {} DO NOTHING",
4610 postrust_sql::escape_ident(constraint)
4611 )
4612 } else {
4613 let preset = crate::role::presets(
4619 &context.caller,
4620 context.names,
4621 schema_name,
4622 table_name,
4623 crate::role::Verb::Update,
4624 )
4625 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
4626
4627 let mut param_idx = column_names.len() + 1;
4631 let mut assignments: Vec<String> =
4632 Vec::with_capacity(preset.len() + updates.len());
4633 for (column, value) in &preset {
4634 assignments.push(format!(
4635 "{} = {}",
4636 postrust_sql::escape_ident(column),
4637 write_expression(
4638 &column_types,
4639 column,
4640 value,
4641 &format!("${}", param_idx)
4642 )
4643 ));
4644 conflict_values.push(write_operand(&column_types, column, value));
4645 param_idx += 1;
4646 }
4647 for field in &updates {
4648 let column = table_column_for(context.names, table, field);
4649 if preset.iter().any(|(written, _)| written == column) {
4650 continue;
4651 }
4652 assignments.push(format!(
4653 "{} = EXCLUDED.{}",
4654 postrust_sql::escape_ident(column),
4655 postrust_sql::escape_ident(column)
4656 ));
4657 }
4658 let granted = permission_filter(
4671 &context.caller,
4672 context.names,
4673 schema_name,
4674 table_name,
4675 crate::role::Verb::Update,
4676 )?;
4677 let requested = spec.get("where").filter(|f| !f.is_null()).cloned();
4678 let condition = match and_predicates(requested, granted) {
4679 Some(filter) => {
4680 let filter = &filter;
4681 let type_name = context
4682 .type_names
4683 .get(&(schema_name.to_string(), table_name.to_string()))
4684 .map(String::as_str)
4685 .unwrap_or(table_name);
4686 let scope = WhereScope::table(
4687 schema_name,
4688 table_name,
4689 type_name,
4690 context.names,
4691 )
4692 .under_alias(WRITTEN_ROW)
4693 .with_resolution(context.cache, context.relationships);
4694 let mut alias_counter = 0usize;
4695 build_condition(
4696 filter,
4697 &scope,
4698 &mut param_idx,
4699 &mut conflict_values,
4700 &mut alias_counter,
4701 )?
4702 .map(|sql| format!(" WHERE {}", sql))
4703 .unwrap_or_default()
4704 }
4705 None => String::new(),
4706 };
4707 format!(
4708 " ON CONFLICT ON CONSTRAINT {} DO UPDATE SET {}{}",
4709 postrust_sql::escape_ident(constraint),
4710 assignments.join(", "),
4711 condition
4712 )
4713 }
4714 }
4715 _ => String::new(),
4716 };
4717
4718 let qualified_table = postrust_sql::escape_ident(WRITTEN_ROW);
4724
4725 let check_scope = WhereScope::for_alias(
4730 schema_name,
4731 table_name,
4732 WRITTEN_ROW,
4733 context
4734 .type_names
4735 .get(&(schema_name.to_string(), table_name.to_string()))
4736 .map(String::as_str)
4737 .unwrap_or(table_name),
4738 context.cache,
4739 context.relationships,
4740 context.names,
4741 );
4742 let (check_sql, check_values) = {
4743 let check = crate::role::write_check(
4744 &context.caller,
4745 context.names,
4746 schema_name,
4747 table_name,
4748 crate::role::Verb::Insert,
4749 )
4750 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
4751 match check {
4752 None => (String::new(), Vec::new()),
4753 Some(check) => {
4754 let mut values = Vec::new();
4755 let mut param_idx = column_names.len() + conflict_values.len() + 1;
4756 let mut alias_counter = 0usize;
4757 match build_condition(
4758 &check,
4759 &check_scope,
4760 &mut param_idx,
4761 &mut values,
4762 &mut alias_counter,
4763 )? {
4764 None => (String::new(), Vec::new()),
4765 Some(sql) => (
4766 format!(
4767 ", ({}) AS {}",
4768 sql,
4769 postrust_sql::escape_ident(CHECK_COLUMN)
4770 ),
4771 values,
4772 ),
4773 }
4774 }
4775 }
4776 };
4777
4778 let names = column_names;
4779 let written = if names.is_empty() {
4780 let sql = format!(
4783 "INSERT INTO {}.{} AS {} DEFAULT VALUES{} RETURNING {}{}",
4784 postrust_sql::escape_ident(schema_name),
4785 postrust_sql::escape_ident(table_name),
4786 qualified_table,
4787 conflict_sql,
4788 row_json(&qualified_table, &column_types),
4789 check_sql
4790 );
4791 let mut query = sqlx::query(&sql);
4792 for value in &conflict_values {
4793 query = bind_json_value(query, value);
4794 }
4795 for value in &check_values {
4796 query = bind_json_value(query, value);
4797 }
4798 query
4799 .fetch_optional(&mut *conn)
4800 .await
4801 .map_err(|error| at_path(database_error(error), &context.objects))?
4804 } else {
4805 let placeholders: Vec<String> = names
4806 .iter()
4807 .enumerate()
4808 .map(|(i, column)| {
4809 let placeholder = format!("${}", i + 1);
4810 match columns.get(*column) {
4811 Some(value) => write_expression(&column_types, column, value, &placeholder),
4812 None => placeholder,
4813 }
4814 })
4815 .collect();
4816 let sql = format!(
4817 "INSERT INTO {}.{} AS {} ({}) VALUES ({}){} RETURNING {}{}",
4818 postrust_sql::escape_ident(schema_name),
4819 postrust_sql::escape_ident(table_name),
4820 qualified_table,
4821 names
4822 .iter()
4823 .map(|c| postrust_sql::escape_ident(c))
4824 .collect::<Vec<_>>()
4825 .join(", "),
4826 placeholders.join(", "),
4827 conflict_sql,
4828 row_json(&qualified_table, &column_types),
4829 check_sql
4830 );
4831 trace!("Executing INSERT SQL: {}", sql);
4832 let mut query = sqlx::query(&sql);
4833 for column in &names {
4834 if let Some(value) = columns.get(*column) {
4835 query = bind_json_value(query, &write_operand(&column_types, column, value));
4836 }
4837 }
4838 for value in &conflict_values {
4839 query = bind_json_value(query, value);
4840 }
4841 for value in &check_values {
4842 query = bind_json_value(query, value);
4843 }
4844 query
4845 .fetch_optional(&mut *conn)
4846 .await
4847 .map_err(|error| at_path(database_error(error), &context.objects))?
4850 };
4851
4852 let Some(written) = written else {
4855 return Ok(serde_json::Value::Null);
4856 };
4857 refuse_unchecked_rows(std::slice::from_ref(&written), !check_sql.is_empty())
4858 .map_err(|error| at_path(error, &context.objects))?;
4859 let row: serde_json::Value = written.try_get(0).unwrap_or(serde_json::Value::Null);
4860 *written_count += 1;
4861
4862 for (relationship, data, conflict) in to_many {
4864 let plan =
4865 postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, context.cache)
4866 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
4867 let children = match data {
4868 serde_json::Value::Array(items) => items,
4869 serde_json::Value::Object(map) => vec![serde_json::Value::Object(map)],
4870 serde_json::Value::Null => Vec::new(),
4871 _ => {
4872 return Err(async_graphql::Error::new(format!(
4873 "\"{}\" takes objects to insert",
4874 relationship.name
4875 )))
4876 }
4877 };
4878 for (child_index, child) in children.into_iter().enumerate() {
4879 let serde_json::Value::Object(mut child) = child else {
4880 continue;
4881 };
4882 for (local, foreign) in &plan.columns {
4883 if child.get(foreign).is_some_and(|v| !v.is_null()) {
4887 return Err(at_path(
4888 validation_error(format!(
4889 "cannot insert \"{}\" columns as their values are already \
4890 being determined by parent insert",
4891 foreign
4892 )),
4893 &format!(
4894 "{}.{}{}",
4895 context.row,
4896 relationship.name,
4897 match relationship.is_list {
4898 true => format!(".data[{}]", child_index),
4899 false => ".data".to_string(),
4900 }
4901 ),
4902 ));
4903 }
4904 if let Some(value) = row.get(local) {
4905 child.insert(foreign.clone(), value.clone());
4906 }
4907 }
4908 let at = match relationship.is_list {
4914 true => format!(".data[{}]", child_index),
4915 false => ".data".to_string(),
4916 };
4917 let under = format!("{}.{}", context.row, relationship.name);
4918 let nested = InsertContext {
4919 on_conflict: conflict.clone(),
4920 cache: context.cache,
4921 relationships: context.relationships,
4922 type_names: context.type_names,
4923 names: context.names,
4924 caller: context.caller,
4925 row: format!("{}{}", under, at),
4926 objects: format!("{}.data", under),
4927 conflict: format!("{}.on_conflict", under),
4928 };
4929 insert_row(
4930 conn,
4931 &plan.foreign_schema,
4932 &plan.foreign_table,
4933 child,
4934 &nested,
4935 written_count,
4936 )
4937 .await?;
4938 }
4939 }
4940
4941 Ok(row)
4942 })
4943}
4944
4945#[derive(Clone)]
4951struct InsertContext<'a> {
4952 on_conflict: Option<serde_json::Value>,
4955 cache: &'a SchemaCache,
4956 relationships: &'a HashMap<String, Vec<RelationshipField>>,
4957 type_names: &'a HashMap<(String, String), String>,
4961 names: &'a crate::names::NameOverrides,
4964 caller: crate::role::Caller<'a>,
4968 row: String,
4973 objects: String,
4977 conflict: String,
4979}
4980
4981#[allow(clippy::too_many_arguments)]
4994async fn reread_returning(
4995 pool: &PgPool,
4996 gql_ctx: &GraphQLContext,
4997 schema_name: &str,
4998 table_name: &str,
4999 type_name: &str,
5000 rows: Vec<Value>,
5001 returning: async_graphql::SelectionField<'_>,
5002 relationships: &HashMap<String, Vec<RelationshipField>>,
5003 names: &crate::names::NameOverrides,
5004 max_rows: Option<i64>,
5005) -> Result<Vec<Value>, async_graphql::Error> {
5006 let guard = gql_ctx
5007 .schema_cache
5008 .get()
5009 .await
5010 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
5011 let Some(cache) = guard.as_ref() else {
5012 return Ok(rows);
5013 };
5014 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
5015 let Some(table) = cache.get_table(&qi) else {
5016 return Ok(rows);
5017 };
5018 if table.pk_cols.is_empty() {
5019 return Ok(rows);
5021 }
5022
5023 let mut param_idx = 1usize;
5024 let mut values: Vec<serde_json::Value> = Vec::new();
5025 let mut alias_counter = 0usize;
5026 let embeds = build_embed_expressions(
5027 &gql_ctx.caller(),
5028 cache,
5029 relationships,
5030 type_name,
5031 "src",
5032 returning,
5033 max_rows,
5034 &mut alias_counter,
5035 &mut param_idx,
5036 &mut values,
5037 names,
5038 )?;
5039 let computed = computed_projections(
5040 table,
5041 returning,
5042 "src",
5043 names,
5044 cache,
5045 &mut param_idx,
5046 &mut values,
5047 )?;
5048 let renames = names.renames_columns(&table.schema, &table.name);
5052 if embeds.is_empty() && computed.is_empty() && !renames {
5053 return Ok(rows);
5054 }
5055
5056 let mut conditions = Vec::with_capacity(rows.len());
5058 for row in &rows {
5059 let mut parts = Vec::with_capacity(table.pk_cols.len());
5060 for column in &table.pk_cols {
5061 let Value::Object(map) = row else {
5062 return Ok(rows);
5063 };
5064 let Some(value) = map.get(&async_graphql::Name::new(column)) else {
5065 return Ok(rows);
5066 };
5067 parts.push(format!(
5068 "{} = ${}",
5069 postrust_sql::escape_ident(column),
5070 param_idx
5071 ));
5072 values.push(value_to_json(value));
5073 param_idx += 1;
5074 }
5075 conditions.push(format!("({})", parts.join(" AND ")));
5076 }
5077 if conditions.is_empty() {
5078 return Ok(rows);
5079 }
5080
5081 let mut projection =
5082 rename_projection(table, "src", names).unwrap_or_else(|| "src.*".to_string());
5083 for expression in &computed {
5084 projection.push_str(", ");
5085 projection.push_str(expression);
5086 }
5087 for (name, expression) in &embeds {
5088 projection.push_str(", ");
5089 projection.push_str(expression);
5090 projection.push_str(" AS ");
5091 projection.push_str(&postrust_sql::escape_ident(name));
5092 }
5093
5094 let sql = format!(
5095 "SELECT {} FROM (SELECT {} FROM (SELECT * FROM {}.{} WHERE {}) AS src) AS pgrst_r",
5096 row_json("pgrst_r", &exposed_column_types(table, names)),
5097 projection,
5098 postrust_sql::escape_ident(schema_name),
5099 postrust_sql::escape_ident(table_name),
5100 conditions.join(" OR ")
5101 );
5102
5103 let mut guard = write_tx(gql_ctx, pool).await?;
5106 let conn = guard.as_mut().expect("write_tx opens one");
5107 let reread = execute_query_on(conn, &sql, &values).await?;
5108 drop(guard);
5109
5110 if reread.is_empty() {
5111 return Ok(rows);
5112 }
5113 Ok(reread.into_iter().map(json_to_value).collect())
5114}
5115
5116async fn execute_insert(
5118 pool: &PgPool,
5119 gql_ctx: &GraphQLContext,
5120 schema_name: &str,
5121 table_name: &str,
5122 objects: serde_json::Value,
5123 context: &InsertContext<'_>,
5124 single: bool,
5128) -> Result<(Vec<Value>, usize), async_graphql::Error> {
5129 trace!("Insert mutation for {}: {:?}", table_name, objects);
5130
5131 let objects_array = match objects {
5133 serde_json::Value::Array(arr) => arr,
5134 serde_json::Value::Object(obj) => vec![serde_json::Value::Object(obj)],
5135 _ => {
5136 return Err(async_graphql::Error::new(
5137 "objects must be an array or object",
5138 ))
5139 }
5140 };
5141
5142 if objects_array.is_empty() {
5143 return Err(async_graphql::Error::new("objects cannot be empty"));
5144 }
5145
5146 let mut guard = write_tx(gql_ctx, pool).await?;
5147 let conn = guard.as_mut().expect("write_tx opens one");
5148
5149 let mut inserted: Vec<Value> = Vec::new();
5150 let mut written = 0usize;
5151
5152 for (index, object) in objects_array.into_iter().enumerate() {
5153 let serde_json::Value::Object(map) = object else {
5154 return Err(async_graphql::Error::new(
5155 "each object to insert is an object",
5156 ));
5157 };
5158 let context = InsertContext {
5161 row: match single {
5162 true => context.objects.clone(),
5163 false => format!("{}[{}]", context.objects, index),
5164 },
5165 ..context.clone()
5166 };
5167 let row = insert_row(conn, schema_name, table_name, map, &context, &mut written).await?;
5168 if !row.is_null() {
5172 inserted.push(json_to_value(row));
5173 }
5174 }
5175
5176 Ok((inserted, written))
5177}
5178
5179fn mutation_result<'a>(rows: Vec<Value>, affected: usize, by_key: bool) -> Option<FieldValue<'a>> {
5191 if by_key {
5192 return rows.into_iter().next().map(FieldValue::value);
5193 }
5194 let mut response = async_graphql::indexmap::IndexMap::new();
5195 response.insert(
5196 async_graphql::Name::new("affected_rows"),
5197 Value::from(affected),
5198 );
5199 response.insert(async_graphql::Name::new("returning"), Value::List(rows));
5200 Some(FieldValue::value(Value::Object(response)))
5201}
5202
5203fn bind_json_value<'q>(
5205 query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
5206 value: &serde_json::Value,
5207) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> {
5208 match value {
5209 serde_json::Value::Null => query.bind(None::<String>),
5210 serde_json::Value::Bool(b) => query.bind(*b),
5211 serde_json::Value::Number(n) => {
5212 if let Some(i) = n.as_i64() {
5213 query.bind(i)
5214 } else if let Some(f) = n.as_f64() {
5215 query.bind(f)
5216 } else {
5217 query.bind(n.to_string())
5218 }
5219 }
5220 serde_json::Value::String(s) => query.bind(s.clone()),
5221 _ => query.bind(value.to_string()),
5222 }
5223}
5224
5225#[allow(clippy::too_many_arguments)] async fn execute_update(
5228 pool: &PgPool,
5229 gql_ctx: &GraphQLContext,
5230 schema_name: &str,
5231 table_name: &str,
5232 type_name: &str,
5233 operators: Vec<(&'static str, serde_json::Value)>,
5234 column_types: HashMap<String, String>,
5235 where_clause: Option<serde_json::Value>,
5236 relationships: &HashMap<String, Vec<RelationshipField>>,
5237 names: &crate::names::NameOverrides,
5238) -> Result<Vec<Value>, async_graphql::Error> {
5239 use sqlx::Row;
5240
5241 trace!("Update mutation for {}: {:?}", table_name, operators);
5242
5243 let mut set_parts: Vec<String> = Vec::new();
5248 let mut set_values: Vec<serde_json::Value> = Vec::new();
5249 let mut param_idx = 1;
5250 let mut written: HashSet<String> = HashSet::new();
5251
5252 let preset = crate::role::presets(
5262 &gql_ctx.caller(),
5263 names,
5264 schema_name,
5265 table_name,
5266 crate::role::Verb::Update,
5267 )
5268 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
5269 let preset_columns: HashSet<&str> = preset.iter().map(|(name, _)| name.as_str()).collect();
5270 for (column, value) in &preset {
5271 let placeholder =
5272 write_expression(&column_types, column, value, &format!("${}", param_idx));
5273 set_parts.push(format!(
5274 "{} = {}",
5275 postrust_sql::escape_ident(column),
5276 placeholder
5277 ));
5278 set_values.push(write_operand(&column_types, column, value));
5279 param_idx += 1;
5280 }
5281
5282 for (operator, payload) in &operators {
5283 let serde_json::Value::Object(map) = payload else {
5284 return Err(async_graphql::Error::new(format!(
5285 "\"{}\" takes an object mapping columns to values",
5286 operator
5287 )));
5288 };
5289 for (field, value) in map {
5290 let column = &names
5291 .column_source(schema_name, table_name, field)
5292 .unwrap_or(field)
5293 .to_string();
5294 if preset_columns.contains(column.as_str()) {
5297 continue;
5298 }
5299 if !written.insert(column.clone()) {
5300 return Err(async_graphql::Error::new(format!(
5301 "\"{}\" is written twice in one update; a column may be \
5302 changed by one operator at a time",
5303 column
5304 )));
5305 }
5306 let quoted = postrust_sql::escape_ident(column);
5307 let placeholder = match *operator {
5316 "_set" | "_inc" => {
5317 write_expression(&column_types, column, value, &format!("${}", param_idx))
5318 }
5319 _ => format!("${}", param_idx),
5320 };
5321 let assignment = match *operator {
5324 "_set" => format!("{} = {}", quoted, placeholder),
5325 "_inc" => format!("{} = {} + {}", quoted, quoted, placeholder),
5326 "_append" => format!("{} = {} || {}::jsonb", quoted, quoted, placeholder),
5327 "_prepend" => format!("{} = {}::jsonb || {}", quoted, placeholder, quoted),
5328 "_delete_key" => format!("{} = {} - {}::text", quoted, quoted, placeholder),
5329 "_delete_elem" => format!("{} = {} - {}::int", quoted, quoted, placeholder),
5330 "_delete_at_path" => {
5336 let Some(steps) = value.as_array() else {
5337 return Err(async_graphql::Error::new(format!(
5338 "\"_delete_at_path\" on \"{}\" takes a list of keys",
5339 column
5340 )));
5341 };
5342 let path = sql_array(steps, "text[]", &mut param_idx, &mut set_values);
5343 set_parts.push(format!("{} = {} #- {}", quoted, quoted, path));
5344 continue;
5345 }
5346 other => {
5347 return Err(async_graphql::Error::new(format!(
5348 "unsupported update operator \"{}\"",
5349 other
5350 )))
5351 }
5352 };
5353 if *operator == "_set"
5354 && matches!(
5355 column_types.get(column).map(String::as_str),
5356 Some("geometry") | Some("geography")
5357 )
5358 {
5359 check_geojson(value)?;
5360 }
5361 set_parts.push(assignment);
5362 set_values.push(match *operator {
5363 "_set" => write_operand(&column_types, column, value),
5364 _ => value.clone(),
5365 });
5366 param_idx += 1;
5367 }
5368 }
5369
5370 if set_parts.is_empty() {
5375 return Ok(Vec::new());
5376 }
5377
5378 let guard = gql_ctx
5382 .schema_cache
5383 .get()
5384 .await
5385 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
5386 let cache = guard
5387 .as_ref()
5388 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
5389 let scope = WhereScope::table(schema_name, table_name, type_name, names)
5390 .under_alias(WRITTEN_ROW)
5391 .with_resolution(cache, relationships);
5392 let (where_sql, where_values) = build_where_clause(where_clause.as_ref(), param_idx, &scope)?;
5393
5394 if where_sql.is_empty() {
5397 return Err(async_graphql::Error::new(format!(
5398 "update on \"{}\" requires a `where` argument with at least one \
5399 recognised condition; refusing to update every row",
5400 table_name
5401 )));
5402 }
5403
5404 let (check_sql, check_values) = permission_check_sql(
5409 gql_ctx,
5410 names,
5411 schema_name,
5412 table_name,
5413 crate::role::Verb::Update,
5414 &scope,
5415 param_idx + where_values.len(),
5416 )?;
5417
5418 let sql = format!(
5419 "UPDATE {}.{} AS {} SET {} {} RETURNING {}{}",
5420 postrust_sql::escape_ident(schema_name),
5421 postrust_sql::escape_ident(table_name),
5422 postrust_sql::escape_ident(WRITTEN_ROW),
5423 set_parts.join(", "),
5424 where_sql,
5425 row_json(&postrust_sql::escape_ident(WRITTEN_ROW), &column_types),
5426 check_sql
5427 );
5428
5429 trace!("Executing UPDATE SQL: {}", sql);
5430
5431 let mut query = sqlx::query(&sql);
5433
5434 for val in &set_values {
5436 query = bind_json_value(query, val);
5437 }
5438
5439 for val in &where_values {
5441 query = bind_json_value(query, val);
5442 }
5443
5444 for val in &check_values {
5446 query = bind_json_value(query, val);
5447 }
5448
5449 let mut guard = write_tx(gql_ctx, pool).await?;
5450 let conn = guard.as_mut().expect("write_tx opens one");
5451 let rows = query.fetch_all(&mut **conn).await.map_err(database_error)?;
5452
5453 refuse_unchecked_rows(&rows, !check_sql.is_empty())?;
5454
5455 let updated: Vec<Value> = rows
5456 .iter()
5457 .filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
5458 .map(json_to_value)
5459 .collect();
5460
5461 Ok(updated)
5462}
5463
5464fn permission_check_sql(
5471 gql_ctx: &GraphQLContext,
5472 names: &crate::names::NameOverrides,
5473 schema_name: &str,
5474 table_name: &str,
5475 verb: crate::role::Verb,
5476 scope: &WhereScope<'_>,
5477 start_param_idx: usize,
5478) -> Result<(String, Vec<serde_json::Value>), async_graphql::Error> {
5479 let check = crate::role::write_check(&gql_ctx.caller(), names, schema_name, table_name, verb)
5480 .map_err(|fault| coded_error(fault.code(), fault.to_string()))?;
5481 let Some(check) = check else {
5482 return Ok((String::new(), Vec::new()));
5483 };
5484
5485 let mut values = Vec::new();
5486 let mut param_idx = start_param_idx;
5487 let mut alias_counter = 0usize;
5488 let condition = build_condition(
5489 &check,
5490 scope,
5491 &mut param_idx,
5492 &mut values,
5493 &mut alias_counter,
5494 )?;
5495 Ok(match condition {
5496 None => (String::new(), Vec::new()),
5499 Some(sql) => (
5500 format!(
5501 ", ({}) AS {}",
5502 sql,
5503 postrust_sql::escape_ident(CHECK_COLUMN)
5504 ),
5505 values,
5506 ),
5507 })
5508}
5509
5510const CHECK_COLUMN: &str = "pgrst_permission_check";
5512
5513fn refuse_unchecked_rows(
5520 rows: &[sqlx::postgres::PgRow],
5521 checked: bool,
5522) -> Result<(), async_graphql::Error> {
5523 if !checked {
5524 return Ok(());
5525 }
5526 use sqlx::Row;
5527 let all_passed = rows
5528 .iter()
5529 .all(|row| row.try_get::<bool, _>(CHECK_COLUMN).unwrap_or(false));
5530 match all_passed {
5531 true => Ok(()),
5532 false => Err(coded_error("permission-error", crate::role::CHECK_FAILED)),
5533 }
5534}
5535
5536#[allow(clippy::too_many_arguments)] async fn execute_delete(
5539 pool: &PgPool,
5540 gql_ctx: &GraphQLContext,
5541 schema_name: &str,
5542 table_name: &str,
5543 type_name: &str,
5544 where_clause: Option<serde_json::Value>,
5545 returning: Option<async_graphql::SelectionField<'_>>,
5546 relationships: &HashMap<String, Vec<RelationshipField>>,
5547 names: &crate::names::NameOverrides,
5548 max_rows: Option<i64>,
5549) -> Result<Vec<Value>, async_graphql::Error> {
5550 use sqlx::Row;
5551
5552 trace!("Delete mutation for {}", table_name);
5553
5554 let guard = gql_ctx
5555 .schema_cache
5556 .get()
5557 .await
5558 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
5559 let cache = guard
5560 .as_ref()
5561 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
5562 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
5563 let column_types = cache
5564 .get_table(&qi)
5565 .map(|table| exposed_column_types(table, names))
5566 .unwrap_or_default();
5567
5568 let scope = WhereScope::table(schema_name, table_name, type_name, names)
5572 .under_alias(WRITTEN_ROW)
5573 .with_resolution(cache, relationships);
5574 let (where_sql, mut values) = build_where_clause(where_clause.as_ref(), 1, &scope)?;
5575
5576 if where_sql.is_empty() {
5579 return Err(async_graphql::Error::new(format!(
5580 "delete on \"{}\" requires a `where` argument with at least one \
5581 recognised condition; refusing to delete every row",
5582 table_name
5583 )));
5584 }
5585
5586 let mut projection = String::new();
5592 let renamed = cache
5593 .get_table(&qi)
5594 .and_then(|table| rename_projection(table, "src", names));
5595 if let Some(returning) = returning {
5596 let mut param_idx = values.len() + 1;
5597 let mut alias_counter = 0usize;
5598 let embeds = build_embed_expressions(
5599 &gql_ctx.caller(),
5600 cache,
5601 relationships,
5602 type_name,
5603 "src",
5604 returning,
5605 max_rows,
5606 &mut alias_counter,
5607 &mut param_idx,
5608 &mut values,
5609 names,
5610 )?;
5611 let computed = match cache.get_table(&qi) {
5612 Some(table) => computed_projections(
5613 table,
5614 returning,
5615 "src",
5616 names,
5617 cache,
5618 &mut param_idx,
5619 &mut values,
5620 )?,
5621 None => Vec::new(),
5622 };
5623 for expression in &computed {
5624 projection.push_str(", ");
5625 projection.push_str(expression);
5626 }
5627 for (name, expression) in &embeds {
5628 projection.push_str(", ");
5629 projection.push_str(expression);
5630 projection.push_str(" AS ");
5631 projection.push_str(&postrust_sql::escape_ident(name));
5632 }
5633 }
5634
5635 let written = postrust_sql::escape_ident(WRITTEN_ROW);
5636 let sql = if projection.is_empty() && renamed.is_none() {
5637 format!(
5638 "DELETE FROM {}.{} AS {} {} RETURNING {}",
5639 postrust_sql::escape_ident(schema_name),
5640 postrust_sql::escape_ident(table_name),
5641 written,
5642 where_sql,
5643 row_json(&written, &column_types)
5644 )
5645 } else {
5646 format!(
5647 "WITH pgrst_deleted AS (DELETE FROM {}.{} AS {} {} RETURNING {}.*) \
5648 SELECT {} FROM (SELECT {}{} FROM pgrst_deleted AS src) AS pgrst_r",
5649 postrust_sql::escape_ident(schema_name),
5650 postrust_sql::escape_ident(table_name),
5651 written,
5652 where_sql,
5653 written,
5654 row_json("pgrst_r", &column_types),
5655 renamed.unwrap_or_else(|| "src.*".to_string()),
5656 projection
5657 )
5658 };
5659
5660 trace!("Executing DELETE SQL: {}", sql);
5661
5662 let mut query = sqlx::query(&sql);
5663 for val in &values {
5664 query = bind_json_value(query, val);
5665 }
5666
5667 let mut guard = write_tx(gql_ctx, pool).await?;
5668 let conn = guard.as_mut().expect("write_tx opens one");
5669 let rows = query.fetch_all(&mut **conn).await.map_err(database_error)?;
5670
5671 let deleted: Vec<Value> = rows
5672 .iter()
5673 .filter_map(|row| row.try_get::<serde_json::Value, _>(0).ok())
5674 .map(json_to_value)
5675 .collect();
5676
5677 Ok(deleted)
5678}
5679
5680fn permission_predicate(
5688 caller: &crate::role::Caller<'_>,
5689 names: &crate::names::NameOverrides,
5690 schema: &str,
5691 table: &str,
5692) -> Result<Option<serde_json::Value>, async_graphql::Error> {
5693 permission_filter(caller, names, schema, table, crate::role::Verb::Select)
5694}
5695
5696fn permission_filter(
5698 caller: &crate::role::Caller<'_>,
5699 names: &crate::names::NameOverrides,
5700 schema: &str,
5701 table: &str,
5702 verb: crate::role::Verb,
5703) -> Result<Option<serde_json::Value>, async_graphql::Error> {
5704 crate::role::row_filter(caller, names, schema, table, verb)
5705 .map(|filter| filter.map(|predicate| serde_json::json!({ PERMISSION: predicate })))
5706 .map_err(|fault| coded_error(fault.code(), fault.to_string()))
5707}
5708
5709const PERMISSION: &str = "_permission";
5730
5731fn permission_limit(
5737 caller: &crate::role::Caller<'_>,
5738 names: &crate::names::NameOverrides,
5739 schema: &str,
5740 table: &str,
5741) -> Option<i64> {
5742 crate::role::read_limit(caller, names, schema, table)
5743}
5744
5745fn and_predicates(
5752 request: Option<serde_json::Value>,
5753 permission: Option<serde_json::Value>,
5754) -> Option<serde_json::Value> {
5755 match (request, permission) {
5756 (None, None) => None,
5757 (Some(only), None) | (None, Some(only)) => Some(only),
5758 (Some(request), Some(permission)) => {
5759 Some(serde_json::json!({ "_and": [request, permission] }))
5760 }
5761 }
5762}
5763
5764fn build_where_clause(
5774 where_value: Option<&serde_json::Value>,
5775 start_param_idx: usize,
5776 scope: &WhereScope<'_>,
5777) -> Result<(String, Vec<serde_json::Value>), async_graphql::Error> {
5778 let mut values: Vec<serde_json::Value> = Vec::new();
5779 let mut param_idx = start_param_idx;
5780 let mut alias_counter = 0usize;
5781
5782 let condition = match where_value {
5783 Some(value) => build_condition(
5784 value,
5785 scope,
5786 &mut param_idx,
5787 &mut values,
5788 &mut alias_counter,
5789 )?,
5790 None => None,
5791 };
5792
5793 Ok(match condition {
5794 Some(sql) => (format!("WHERE {}", sql), values),
5795 None => (String::new(), values),
5796 })
5797}
5798
5799pub struct WhereScope<'a> {
5811 qualified: postrust_core::api_request::QualifiedIdentifier,
5813 sql_ref: String,
5815 row_ref: String,
5819 type_name: String,
5821 root_ref: String,
5829 root_qualified: postrust_core::api_request::QualifiedIdentifier,
5830 names: &'a crate::names::NameOverrides,
5833 resolution: Option<WhereResolution<'a>>,
5834 caller: Option<crate::role::Caller<'a>>,
5841}
5842
5843struct WhereResolution<'a> {
5845 cache: &'a SchemaCache,
5846 relationships: &'a HashMap<String, Vec<RelationshipField>>,
5847}
5848
5849impl<'a> WhereScope<'a> {
5850 pub fn table(
5852 schema: &str,
5853 table: &str,
5854 type_name: &str,
5855 names: &'a crate::names::NameOverrides,
5856 ) -> Self {
5857 Self {
5858 qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5859 sql_ref: format!(
5860 "{}.{}",
5861 postrust_sql::escape_ident(schema),
5862 postrust_sql::escape_ident(table)
5863 ),
5864 row_ref: postrust_sql::escape_ident(table),
5865 type_name: type_name.to_string(),
5866 root_ref: format!(
5867 "{}.{}",
5868 postrust_sql::escape_ident(schema),
5869 postrust_sql::escape_ident(table)
5870 ),
5871 root_qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5872 names,
5873 resolution: None,
5874 caller: None,
5875 }
5876 }
5877
5878 fn under_alias(mut self, alias: &str) -> Self {
5886 if self.root_ref == self.sql_ref {
5889 self.root_ref = postrust_sql::escape_ident(alias);
5890 }
5891 self.sql_ref = postrust_sql::escape_ident(alias);
5892 self.row_ref = postrust_sql::escape_ident(alias);
5893 self
5894 }
5895
5896 fn with_resolution(
5898 mut self,
5899 cache: &'a SchemaCache,
5900 relationships: &'a HashMap<String, Vec<RelationshipField>>,
5901 ) -> Self {
5902 self.resolution = Some(WhereResolution {
5903 cache,
5904 relationships,
5905 });
5906 self
5907 }
5908
5909 #[allow(clippy::too_many_arguments)]
5911 fn for_alias(
5912 schema: &str,
5913 table: &str,
5914 alias: &str,
5915 type_name: &str,
5916 cache: &'a SchemaCache,
5917 relationships: &'a HashMap<String, Vec<RelationshipField>>,
5918 names: &'a crate::names::NameOverrides,
5919 ) -> Self {
5920 Self {
5921 qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5922 sql_ref: postrust_sql::escape_ident(alias),
5923 row_ref: postrust_sql::escape_ident(alias),
5924 type_name: type_name.to_string(),
5925 root_ref: postrust_sql::escape_ident(alias),
5926 root_qualified: postrust_core::api_request::QualifiedIdentifier::new(schema, table),
5927 names,
5928 resolution: Some(WhereResolution {
5929 cache,
5930 relationships,
5931 }),
5932 caller: None,
5933 }
5934 }
5935
5936 fn aliased(
5943 schema: &str,
5944 table: &str,
5945 alias: &str,
5946 type_name: &str,
5947 from: &WhereScope<'a>,
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: from.root_ref.clone(),
5957 root_qualified: from.root_qualified.clone(),
5958 names: from.names,
5959 resolution: from.resolution.as_ref().map(|r| WhereResolution {
5960 cache: r.cache,
5961 relationships: r.relationships,
5962 }),
5963 caller: from.caller,
5967 }
5968 }
5969
5970 pub fn for_caller(mut self, caller: crate::role::Caller<'a>) -> Self {
5972 self.caller = Some(caller);
5973 self
5974 }
5975
5976 fn as_system(&self) -> WhereScope<'a> {
5979 WhereScope {
5980 qualified: self.qualified.clone(),
5981 sql_ref: self.sql_ref.clone(),
5982 row_ref: self.row_ref.clone(),
5983 type_name: self.type_name.clone(),
5984 root_ref: self.sql_ref.clone(),
5991 root_qualified: self.qualified.clone(),
5992 names: self.names,
5993 resolution: self.resolution.as_ref().map(|r| WhereResolution {
5994 cache: r.cache,
5995 relationships: r.relationships,
5996 }),
5997 caller: None,
5998 }
5999 }
6000
6001 fn column(&self, name: &str) -> String {
6009 let column = self.column_name(name);
6010 let plain = format!("{}.{}", self.sql_ref, postrust_sql::escape_ident(column));
6011 let Some(cache) = self.resolution.as_ref().map(|r| r.cache) else {
6012 return plain;
6013 };
6014 let Some(table) = cache.get_table(&self.qualified) else {
6015 return plain;
6016 };
6017 if table.get_column(column).is_some() {
6018 return plain;
6019 }
6020 let function = self
6021 .names
6022 .computed_source(&table.schema, &table.name, name)
6023 .unwrap_or(name);
6024 match table.get_computed_column(function) {
6025 Some(definition) => {
6026 computed_call(definition, &format!("{}.*", self.row_ref), &self.row_ref)
6027 }
6028 None => plain,
6029 }
6030 }
6031
6032 fn column_name<'n>(&'n self, name: &'n str) -> &'n str {
6034 self.names
6035 .column_source(&self.qualified.schema, &self.qualified.name, name)
6036 .unwrap_or(name)
6037 }
6038
6039 fn column_type(&self, name: &str) -> Option<String> {
6043 let cache = self.resolution.as_ref()?.cache;
6044 let table = cache.get_table(&self.qualified)?;
6045 table
6046 .get_column(self.column_name(name))
6047 .map(|c| c.nominal_type.clone())
6048 }
6049
6050 fn other_column(
6064 &self,
6065 operand: &serde_json::Value,
6066 op: &str,
6067 ) -> Result<String, async_graphql::Error> {
6068 use serde_json::Value;
6069
6070 let refused = |what: &str| {
6071 async_graphql::Error::new(format!(
6072 "the \"{}\" comparison takes the name of a column, {}",
6073 op, what
6074 ))
6075 };
6076 let (against, qualified, name) = match operand {
6077 Value::String(name) => (&self.sql_ref, &self.qualified, name.as_str()),
6078 Value::Array(path) => match path.as_slice() {
6079 [Value::String(name)] => (&self.sql_ref, &self.qualified, name.as_str()),
6080 [Value::String(root), Value::String(name)] if root == "$" => {
6081 (&self.root_ref, &self.root_qualified, name.as_str())
6082 }
6083 _ => return Err(refused("or [\"$\", \"a column\"]")),
6084 },
6085 _ => return Err(refused("written as a string")),
6086 };
6087 let column = self
6090 .names
6091 .column_source(&qualified.schema, &qualified.name, name)
6092 .unwrap_or(name);
6093 Ok(format!(
6094 "{}.{}",
6095 against,
6096 postrust_sql::escape_ident(column)
6097 ))
6098 }
6099
6100 fn relationship(&self, name: &str) -> Option<std::borrow::Cow<'a, RelationshipField>> {
6114 let resolution = self.resolution.as_ref()?;
6115 let published = resolution
6116 .relationships
6117 .get(&self.type_name)
6118 .and_then(|fields| fields.iter().find(|r| r.name == name));
6119 if let Some(found) = published {
6120 return Some(std::borrow::Cow::Borrowed(found));
6121 }
6122
6123 let table = resolution.cache.get_table(&self.qualified)?;
6124
6125 if let Some(declared) = self
6131 .names
6132 .declared_relationships(&table.schema, &table.name)
6133 {
6134 if let Some(entry) = declared.iter().find(|entry| entry.name == name) {
6135 if let Some(field) = crate::schema::mapped_relationship_field(
6138 entry,
6139 table,
6140 resolution.cache,
6141 &table.schema,
6142 &entry
6143 .target(&table.schema)
6144 .map(|(_, target)| target)
6145 .unwrap_or_default(),
6146 ) {
6147 return Some(std::borrow::Cow::Owned(field));
6148 }
6149 if let Some(key) = entry.using.as_deref() {
6153 if let Some(found) = resolution
6154 .cache
6155 .get_relationships(&self.qualified, &table.schema)
6156 .into_iter()
6157 .flatten()
6158 .find(|rel| {
6159 crate::schema::relationship_keys(rel)
6160 .iter()
6161 .any(|found| found == key)
6162 })
6163 {
6164 return Some(std::borrow::Cow::Owned(
6165 crate::schema::relationship::RelationshipField::from_relationship_named(
6166 found,
6167 &found.foreign_table().name,
6168 Some(&entry.name),
6169 ),
6170 ));
6171 }
6172 }
6173 }
6174 }
6175
6176 resolution
6177 .cache
6178 .get_relationships(&self.qualified, &table.schema)?
6179 .iter()
6180 .find_map(|rel| {
6181 let given = crate::schema::relationship_keys(rel)
6182 .iter()
6183 .find_map(|key| self.names.relationship(&table.schema, &table.name, key));
6184 let field = RelationshipField::from_relationship_named(
6188 rel,
6189 &rel.foreign_table().name,
6190 given,
6191 );
6192 (field.name == name).then_some(std::borrow::Cow::Owned(field))
6193 })
6194 }
6195}
6196
6197fn column_comparison(op: &str) -> Option<&'static str> {
6206 Some(match op {
6207 "_ceq" => "=",
6208 "_cne" => "<>",
6209 "_cgt" => ">",
6210 "_cgte" => ">=",
6211 "_clt" => "<",
6212 "_clte" => "<=",
6213 _ => return None,
6214 })
6215}
6216
6217fn build_condition(
6219 value: &serde_json::Value,
6220 scope: &WhereScope<'_>,
6221 param_idx: &mut usize,
6222 values: &mut Vec<serde_json::Value>,
6223 alias_counter: &mut usize,
6224) -> Result<Option<String>, async_graphql::Error> {
6225 let serde_json::Value::Object(map) = value else {
6226 return Ok(None);
6227 };
6228
6229 let mut conditions: Vec<String> = Vec::new();
6230
6231 for (key, val) in map {
6232 match key.as_str() {
6233 "_and" | "_or" => {
6237 let members = val.as_array().ok_or_else(|| {
6238 async_graphql::Error::new(format!("\"{}\" takes a list of expressions", key))
6239 })?;
6240 let mut parts = Vec::new();
6241 for member in members {
6242 if let Some(sql) =
6243 build_condition(member, scope, param_idx, values, alias_counter)?
6244 {
6245 parts.push(sql);
6246 }
6247 }
6248 if parts.is_empty() {
6249 if key == "_or" {
6250 conditions.push("false".to_string());
6251 }
6252 continue;
6253 }
6254 let joiner = if key == "_and" { " AND " } else { " OR " };
6255 conditions.push(format!("({})", parts.join(joiner)));
6256 }
6257 "_not" => {
6258 if let Some(sql) = build_condition(val, scope, param_idx, values, alias_counter)? {
6259 conditions.push(format!("NOT ({})", sql));
6260 }
6261 }
6262 PERMISSION => {
6265 if let Some(sql) =
6266 build_condition(val, &scope.as_system(), param_idx, values, alias_counter)?
6267 {
6268 conditions.push(sql);
6269 }
6270 }
6271 "_exists" => {
6275 conditions.push(table_exists_sql(
6276 val,
6277 scope,
6278 param_idx,
6279 values,
6280 alias_counter,
6281 )?);
6282 }
6283 name if scope.relationship(name).is_some() => {
6289 let relationship = scope.relationship(name).expect("just checked");
6290 conditions.push(exists_sql(
6291 &relationship,
6292 val,
6293 scope,
6294 param_idx,
6295 values,
6296 alias_counter,
6297 )?);
6298 }
6299 name if scope.column_type(name).is_none()
6304 && name
6305 .strip_suffix("_aggregate")
6306 .and_then(|rel| scope.relationship(rel))
6307 .is_some_and(|rel| rel.is_list) =>
6308 {
6309 let relationship = name
6310 .strip_suffix("_aggregate")
6311 .and_then(|rel| scope.relationship(rel))
6312 .expect("just checked");
6313 conditions.push(aggregate_predicate_sql(
6314 &relationship,
6315 val,
6316 scope,
6317 param_idx,
6318 values,
6319 alias_counter,
6320 )?);
6321 }
6322 column => {
6323 let quoted = scope.column(column);
6324 match val {
6325 serde_json::Value::Object(ops) => {
6326 let column_type = scope.column_type(column);
6327 for (op, operand) in ops {
6328 if let Some(other) = column_comparison(op) {
6334 conditions.push(format!(
6335 "{} {} {}",
6336 quoted,
6337 other,
6338 scope.other_column(operand, op)?
6339 ));
6340 continue;
6341 }
6342 conditions.push(comparison_sql(
6343 "ed,
6344 column,
6345 column_type.as_deref(),
6346 op,
6347 operand,
6348 param_idx,
6349 values,
6350 )?);
6351 }
6352 }
6353 other => {
6357 conditions.push(format!("{} = ${}", quoted, param_idx));
6358 values.push(other.clone());
6359 *param_idx += 1;
6360 }
6361 }
6362 }
6363 }
6364 }
6365
6366 Ok(if conditions.is_empty() {
6367 None
6368 } else if conditions.len() == 1 {
6369 conditions.pop()
6370 } else {
6371 Some(conditions.join(" AND "))
6372 })
6373}
6374
6375fn exists_sql(
6382 relationship: &RelationshipField,
6383 child_expression: &serde_json::Value,
6384 scope: &WhereScope<'_>,
6385 param_idx: &mut usize,
6386 values: &mut Vec<serde_json::Value>,
6387 alias_counter: &mut usize,
6388) -> Result<String, async_graphql::Error> {
6389 let cache = scope.resolution.as_ref().map(|r| r.cache).ok_or_else(|| {
6390 async_graphql::Error::new(format!(
6391 "filtering on the relationship \"{}\" is not available here",
6392 relationship.name
6393 ))
6394 })?;
6395
6396 let plan = postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, cache)
6397 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
6398
6399 if plan.junction.is_some() || (plan.function.is_none() && plan.columns.is_empty()) {
6400 return Err(async_graphql::Error::new(format!(
6401 "filtering on \"{}\" is not supported: it is reached through a junction \
6402 rather than by a key",
6403 relationship.name
6404 )));
6405 }
6406
6407 *alias_counter += 1;
6408 let alias = format!("pgrst_rel_{}", alias_counter);
6409 let child_scope = WhereScope::aliased(
6410 &plan.foreign_schema,
6411 &plan.foreign_table,
6412 &alias,
6413 &relationship.target_type,
6414 scope,
6415 );
6416
6417 let (source, mut correlation) = match &plan.function {
6420 Some(function) => (
6421 format!(
6422 "{}.{}({})",
6423 postrust_sql::escape_ident(&function.schema),
6424 postrust_sql::escape_ident(&function.name),
6425 scope.row_ref
6426 ),
6427 Vec::new(),
6428 ),
6429 None => {
6430 let mut columns = Vec::with_capacity(plan.columns.len());
6431 for (parent_column, child_column) in &plan.columns {
6432 columns.push(format!(
6433 "{} = {}",
6434 scope.column(parent_column),
6435 child_scope.column(child_column)
6436 ));
6437 }
6438 (
6439 format!(
6440 "{}.{}",
6441 postrust_sql::escape_ident(&plan.foreign_schema),
6442 postrust_sql::escape_ident(&plan.foreign_table)
6443 ),
6444 columns,
6445 )
6446 }
6447 };
6448
6449 let child_condition = build_condition(
6450 child_expression,
6451 &child_scope,
6452 param_idx,
6453 values,
6454 alias_counter,
6455 )?;
6456 if let Some(sql) = child_condition {
6457 correlation.push(format!("({})", sql));
6458 }
6459
6460 if let Some(caller) = &child_scope.caller {
6465 let permission = permission_predicate(
6472 caller,
6473 child_scope.names,
6474 &plan.foreign_schema,
6475 &plan.foreign_table,
6476 )?;
6477 if let Some(predicate) = permission {
6478 if let Some(sql) =
6479 build_condition(&predicate, &child_scope, param_idx, values, alias_counter)?
6480 {
6481 correlation.push(format!("({})", sql));
6482 }
6483 }
6484 }
6485 if correlation.is_empty() {
6486 correlation.push("true".to_string());
6487 }
6488
6489 Ok(format!(
6490 "EXISTS (SELECT 1 FROM {} AS {} WHERE {})",
6491 source,
6492 postrust_sql::escape_ident(&alias),
6493 correlation.join(" AND ")
6494 ))
6495}
6496
6497fn table_exists_sql(
6522 spec: &serde_json::Value,
6523 scope: &WhereScope<'_>,
6524 param_idx: &mut usize,
6525 values: &mut Vec<serde_json::Value>,
6526 alias_counter: &mut usize,
6527) -> Result<String, async_graphql::Error> {
6528 let Some((schema, table)) = crate::role::exists_target(spec, &scope.qualified.schema) else {
6529 return Err(async_graphql::Error::new(
6530 "\"_exists\" needs a \"_table\" to look in",
6531 ));
6532 };
6533
6534 let qualified = postrust_core::api_request::QualifiedIdentifier::new(&schema, &table);
6538 if scope
6539 .resolution
6540 .as_ref()
6541 .is_some_and(|r| r.cache.get_table(&qualified).is_none())
6542 {
6543 return Err(async_graphql::Error::new(format!(
6544 "\"_exists\" names a table this server does not have: \"{}.{}\"",
6545 schema, table
6546 )));
6547 }
6548
6549 *alias_counter += 1;
6550 let alias = format!("pgrst_exists_{}", alias_counter);
6551 let type_name = scope
6555 .names
6556 .base_name(&schema, &table)
6557 .unwrap_or(&table)
6558 .to_string();
6559 let mut inner = WhereScope::aliased(&schema, &table, &alias, &type_name, scope);
6560 inner.caller = None;
6563
6564 let condition = match spec.get("_where") {
6565 Some(predicate) => build_condition(predicate, &inner, param_idx, values, alias_counter)?,
6566 None => None,
6567 };
6568
6569 Ok(format!(
6570 "EXISTS (SELECT 1 FROM {}.{} AS {} WHERE {})",
6571 postrust_sql::escape_ident(&schema),
6572 postrust_sql::escape_ident(&table),
6573 postrust_sql::escape_ident(&alias),
6574 condition.unwrap_or_else(|| "true".to_string())
6575 ))
6576}
6577
6578fn aggregate_predicate_sql(
6593 relationship: &RelationshipField,
6594 expression: &serde_json::Value,
6595 scope: &WhereScope<'_>,
6596 param_idx: &mut usize,
6597 values: &mut Vec<serde_json::Value>,
6598 alias_counter: &mut usize,
6599) -> Result<String, async_graphql::Error> {
6600 let serde_json::Value::Object(over) = expression else {
6601 return Ok("true".to_string());
6602 };
6603
6604 let cache = scope.resolution.as_ref().map(|r| r.cache).ok_or_else(|| {
6605 async_graphql::Error::new(format!(
6606 "filtering on the relationship \"{}\" is not available here",
6607 relationship.name
6608 ))
6609 })?;
6610 let plan = postrust_core::embed::EmbedPlan::resolve(&relationship.relationship, cache)
6611 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
6612 if plan.junction.is_some() || (plan.function.is_none() && plan.columns.is_empty()) {
6613 return Err(async_graphql::Error::new(format!(
6614 "aggregating over \"{}\" is not supported: it is reached through a junction \
6615 rather than by a key",
6616 relationship.name
6617 )));
6618 }
6619
6620 let mut conditions: Vec<String> = Vec::new();
6621 for (function, spec) in over {
6622 if spec.is_null() {
6623 continue;
6624 }
6625 let serde_json::Value::Object(spec) = spec else {
6626 continue;
6627 };
6628 let Some(predicate) = spec.get("predicate").filter(|p| !p.is_null()) else {
6629 return Err(validation_error(format!(
6630 "\"{}\" over \"{}\" needs a predicate",
6631 function, relationship.name
6632 )));
6633 };
6634
6635 *alias_counter += 1;
6636 let alias = format!("pgrst_agg_{}", alias_counter);
6637 let child_scope = WhereScope::aliased(
6638 &plan.foreign_schema,
6639 &plan.foreign_table,
6640 &alias,
6641 &relationship.target_type,
6642 scope,
6643 );
6644
6645 let (source, mut correlation) = match &plan.function {
6648 Some(routine) => (
6649 format!(
6650 "{}.{}({})",
6651 postrust_sql::escape_ident(&routine.schema),
6652 postrust_sql::escape_ident(&routine.name),
6653 scope.row_ref
6654 ),
6655 Vec::new(),
6656 ),
6657 None => {
6658 let mut columns = Vec::with_capacity(plan.columns.len());
6659 for (parent_column, child_column) in &plan.columns {
6660 columns.push(format!(
6661 "{} = {}",
6662 scope.column(parent_column),
6663 child_scope.column(child_column)
6664 ));
6665 }
6666 (
6667 format!(
6668 "{}.{}",
6669 postrust_sql::escape_ident(&plan.foreign_schema),
6670 postrust_sql::escape_ident(&plan.foreign_table)
6671 ),
6672 columns,
6673 )
6674 }
6675 };
6676 if let Some(filter) = spec.get("filter").filter(|f| !f.is_null()) {
6679 if let Some(sql) =
6680 build_condition(filter, &child_scope, param_idx, values, alias_counter)?
6681 {
6682 correlation.push(format!("({})", sql));
6683 }
6684 }
6685 if correlation.is_empty() {
6686 correlation.push("true".to_string());
6687 }
6688
6689 let arguments = spec.get("arguments").filter(|a| !a.is_null());
6690 let distinct = matches!(spec.get("distinct"), Some(serde_json::Value::Bool(true)));
6691 let (aggregate, operand_type) = match function.as_str() {
6692 "count" => {
6693 let columns: Vec<String> = match arguments {
6694 Some(serde_json::Value::Array(items)) => items
6695 .iter()
6696 .filter_map(|item| item.as_str())
6697 .map(|column| child_scope.column(column))
6698 .collect(),
6699 Some(serde_json::Value::String(one)) => vec![child_scope.column(one)],
6700 _ => Vec::new(),
6701 };
6702 let counted = match columns.len() {
6703 0 => "*".to_string(),
6704 1 => columns.into_iter().next().unwrap_or_default(),
6705 _ => format!("({})", columns.join(", ")),
6706 };
6707 let counted = match distinct && counted != "*" {
6708 true => format!("DISTINCT {}", counted),
6709 false => counted,
6710 };
6711 (format!("count({})", counted), "integer")
6712 }
6713 "bool_and" | "bool_or" => {
6714 let Some(column) = arguments.and_then(|a| a.as_str()) else {
6715 return Err(validation_error(format!(
6716 "\"{}\" over \"{}\" needs the column to fold",
6717 function, relationship.name
6718 )));
6719 };
6720 (
6721 format!("{}({})", function, child_scope.column(column)),
6722 "boolean",
6723 )
6724 }
6725 other => {
6726 return Err(validation_error(format!(
6727 "\"{}\" is not an aggregate a predicate can be written over",
6728 other
6729 )))
6730 }
6731 };
6732
6733 let subselect = format!(
6734 "(SELECT {} FROM {} AS {} WHERE {})",
6735 aggregate,
6736 source,
6737 postrust_sql::escape_ident(&alias),
6738 correlation.join(" AND ")
6739 );
6740 let serde_json::Value::Object(operators) = predicate else {
6741 return Err(validation_error(format!(
6742 "the predicate on \"{}\" is not a comparison",
6743 function
6744 )));
6745 };
6746 for (operator, operand) in operators {
6747 conditions.push(comparison_sql(
6748 &subselect,
6749 function,
6750 Some(operand_type),
6751 operator,
6752 operand,
6753 param_idx,
6754 values,
6755 )?);
6756 }
6757 }
6758
6759 Ok(match conditions.is_empty() {
6760 true => "true".to_string(),
6761 false => format!("({})", conditions.join(" AND ")),
6762 })
6763}
6764
6765#[allow(clippy::too_many_arguments)]
6773fn postgis_sql(
6774 quoted: &str,
6775 column_type: Option<&str>,
6776 function: &str,
6777 op: &str,
6778 operand: &serde_json::Value,
6779 param_idx: &mut usize,
6780 values: &mut Vec<serde_json::Value>,
6781) -> Result<String, async_graphql::Error> {
6782 let is_geography = column_type == Some("geography");
6783 let mut shape = |value: &serde_json::Value, param_idx: &mut usize| -> String {
6784 let placeholder = format!("${}", param_idx);
6785 *param_idx += 1;
6786 values.push(value.clone());
6787 match is_geography {
6788 true => format!("ST_GeomFromGeoJSON({})::geography", placeholder),
6789 false => format!("ST_GeomFromGeoJSON({})", placeholder),
6790 }
6791 };
6792
6793 match op {
6794 "_st_d_within" | "_st_3d_d_within" => {
6796 let serde_json::Value::Object(spec) = operand else {
6797 return Err(async_graphql::Error::new(format!(
6798 "\"{}\" takes {{distance, from}}",
6799 op
6800 )));
6801 };
6802 let Some(from) = spec.get("from") else {
6803 return Err(async_graphql::Error::new(format!(
6804 "\"{}\" needs a shape to measure from",
6805 op
6806 )));
6807 };
6808 let Some(distance) = spec.get("distance") else {
6809 return Err(async_graphql::Error::new(format!(
6810 "\"{}\" needs a distance",
6811 op
6812 )));
6813 };
6814 let from_sql = shape(from, param_idx);
6815 let distance_sql = format!("${}", param_idx);
6816 *param_idx += 1;
6817 values.push(distance.clone());
6818 let spheroid = match (is_geography, spec.get("use_spheroid")) {
6823 (true, Some(use_spheroid)) if !use_spheroid.is_null() => {
6824 let placeholder = format!("${}", param_idx);
6825 *param_idx += 1;
6826 values.push(use_spheroid.clone());
6827 format!(", {}::boolean", placeholder)
6828 }
6829 _ => String::new(),
6830 };
6831 Ok(format!(
6832 "{}({}, {}, {}::float8{})",
6833 function, quoted, from_sql, distance_sql, spheroid
6834 ))
6835 }
6836 "_st_intersects_rast" => {
6838 let placeholder = format!("${}", param_idx);
6839 *param_idx += 1;
6840 values.push(operand.clone());
6841 Ok(format!("{}({}, {}::raster)", function, quoted, placeholder))
6842 }
6843 "_st_intersects_geom_nband" | "_st_intersects_nband_geom" => {
6845 let serde_json::Value::Object(spec) = operand else {
6846 return Err(async_graphql::Error::new(format!(
6847 "\"{}\" takes an object",
6848 op
6849 )));
6850 };
6851 let Some(geometry) = spec.get("geommin") else {
6852 return Err(async_graphql::Error::new(format!(
6853 "\"{}\" needs a shape",
6854 op
6855 )));
6856 };
6857 let band = spec.get("nband").filter(|v| !v.is_null());
6858 let geometry_sql = shape(geometry, param_idx);
6859 Ok(match band {
6860 None => format!("{}({}, {})", function, quoted, geometry_sql),
6861 Some(band) => {
6862 let placeholder = format!("${}", param_idx);
6863 *param_idx += 1;
6864 values.push(band.clone());
6865 format!(
6869 "{}({}, {}::int, {})",
6870 function, quoted, placeholder, geometry_sql
6871 )
6872 }
6873 })
6874 }
6875 _ => {
6877 let other = shape(operand, param_idx);
6878 Ok(format!("{}({}, {})", function, quoted, other))
6879 }
6880 }
6881}
6882
6883fn sql_array(
6889 items: &[serde_json::Value],
6890 cast: &str,
6891 param_idx: &mut usize,
6892 values: &mut Vec<serde_json::Value>,
6893) -> String {
6894 if items.is_empty() {
6895 return format!("ARRAY[]::{}", cast);
6896 }
6897 let placeholders: Vec<String> = items
6898 .iter()
6899 .map(|item| {
6900 let placeholder = format!("${}", param_idx);
6901 *param_idx += 1;
6902 values.push(item.clone());
6903 placeholder
6904 })
6905 .collect();
6906 format!("ARRAY[{}]::{}", placeholders.join(", "), cast)
6907}
6908
6909#[allow(clippy::too_many_arguments)]
6911fn comparison_sql(
6912 quoted: &str,
6913 column: &str,
6914 column_type: Option<&str>,
6915 op: &str,
6916 operand: &serde_json::Value,
6917 param_idx: &mut usize,
6918 values: &mut Vec<serde_json::Value>,
6919) -> Result<String, async_graphql::Error> {
6920 if op == "_cast" {
6926 let serde_json::Value::Object(casts) = operand else {
6927 return Err(async_graphql::Error::new(
6928 "_cast takes an object naming the type to compare as",
6929 ));
6930 };
6931 let mut conditions = Vec::new();
6932 for (target, comparisons) in casts {
6933 let pg_type = match target.as_str() {
6937 "geometry" | "geography" => target.as_str(),
6938 "String" => "text",
6939 _ => {
6940 return Err(async_graphql::Error::new(format!(
6941 "cannot compare \"{}\" as \"{}\"",
6942 column, target
6943 )))
6944 }
6945 };
6946 let serde_json::Value::Object(ops) = comparisons else {
6947 continue;
6948 };
6949 let cast = format!("{}::{}", quoted, pg_type);
6950 for (nested_op, nested_operand) in ops {
6951 conditions.push(comparison_sql(
6952 &cast,
6953 column,
6954 Some(pg_type),
6955 nested_op,
6956 nested_operand,
6957 param_idx,
6958 values,
6959 )?);
6960 }
6961 }
6962 return Ok(match conditions.len() {
6963 0 => "true".to_string(),
6964 1 => conditions.pop().expect("just counted"),
6965 _ => format!("({})", conditions.join(" AND ")),
6966 });
6967 }
6968
6969 if let Some(function) = crate::input::bool_exp::postgis_function(op) {
6972 return postgis_sql(
6973 quoted,
6974 column_type,
6975 function,
6976 op,
6977 operand,
6978 param_idx,
6979 values,
6980 );
6981 }
6982
6983 if let Some((operator, cast)) = crate::input::bool_exp::ltree_operator(op) {
6987 if cast.ends_with("[]") {
6988 let items = operand.as_array().ok_or_else(|| {
6989 async_graphql::Error::new(format!(
6990 "the \"{}\" comparison on \"{}\" takes a list",
6991 op, column
6992 ))
6993 })?;
6994 return Ok(format!(
6995 "{} {} {}",
6996 quoted,
6997 operator,
6998 sql_array(items, cast, param_idx, values)
6999 ));
7000 }
7001 let placeholder = format!("${}", param_idx);
7002 *param_idx += 1;
7003 values.push(operand.clone());
7004 return Ok(format!("{} {} {}::{}", quoted, operator, placeholder, cast));
7005 }
7006
7007 if let Some(function) = match op {
7012 "_jsonb_path_exists" => Some("jsonb_path_exists"),
7013 "_jsonb_path_match" => Some("jsonb_path_match"),
7014 _ => None,
7015 } {
7016 let placeholder = format!("${}", param_idx);
7017 *param_idx += 1;
7018 values.push(operand.clone());
7019 return Ok(format!(
7020 "{}({}, {}::jsonpath)",
7021 function, quoted, placeholder
7022 ));
7023 }
7024
7025 let binary = match op {
7027 "_eq" => Some("="),
7028 "_neq" => Some("<>"),
7029 "_gt" => Some(">"),
7030 "_gte" => Some(">="),
7031 "_lt" => Some("<"),
7032 "_lte" => Some("<="),
7033 "_like" => Some("LIKE"),
7034 "_nlike" => Some("NOT LIKE"),
7035 "_ilike" => Some("ILIKE"),
7036 "_nilike" => Some("NOT ILIKE"),
7037 "_similar" => Some("SIMILAR TO"),
7038 "_nsimilar" => Some("NOT SIMILAR TO"),
7039 "_regex" => Some("~"),
7040 "_nregex" => Some("!~"),
7041 "_iregex" => Some("~*"),
7042 "_niregex" => Some("!~*"),
7043 "_contains" => Some("@>"),
7044 "_contained_in" => Some("<@"),
7045 "_has_key" => Some("?"),
7046 _ => None,
7047 };
7048
7049 if let Some(sql_op) = binary {
7050 let placeholder = format!("${}", param_idx);
7051 *param_idx += 1;
7052 values.push(match op {
7057 "_contains" | "_contained_in" => match operand {
7058 serde_json::Value::String(_)
7059 | serde_json::Value::Number(_)
7060 | serde_json::Value::Bool(_) => serde_json::Value::String(operand.to_string()),
7061 other => other.clone(),
7062 },
7063 _ => operand.clone(),
7064 });
7065 let cast = match op {
7077 "_contains" | "_contained_in" => "::jsonb".to_string(),
7078 "_has_key" => "::text".to_string(),
7079 _ if operand.is_null() => match column_type {
7088 Some(pg_type) => format!("::{}", pg_type),
7089 None => String::new(),
7090 },
7091 _ if operand.is_string() => match column_type {
7102 Some(pg_type) => format!("::{}", pg_type),
7103 None => String::new(),
7104 },
7105 _ => String::new(),
7106 };
7107 return Ok(format!("{} {} {}{}", quoted, sql_op, placeholder, cast));
7108 }
7109
7110 match op {
7111 "_is_null" => Ok(if operand.as_bool().unwrap_or(false) {
7112 format!("{} IS NULL", quoted)
7113 } else {
7114 format!("{} IS NOT NULL", quoted)
7115 }),
7116 "_in" | "_nin" => {
7117 let items = operand.as_array().ok_or_else(|| {
7118 async_graphql::Error::new(format!(
7119 "the \"{}\" comparison on \"{}\" requires a list of values",
7120 op, column
7121 ))
7122 })?;
7123 if items.is_empty() {
7124 return Ok(if op == "_in" { "false" } else { "true" }.to_string());
7127 }
7128 let mut placeholders = Vec::with_capacity(items.len());
7129 for item in items {
7130 let cast = match (item.is_null() || item.is_string(), column_type) {
7133 (true, Some(pg_type)) => format!("::{}", pg_type),
7134 _ => String::new(),
7135 };
7136 placeholders.push(format!("${}{}", param_idx, cast));
7137 values.push(item.clone());
7138 *param_idx += 1;
7139 }
7140 Ok(format!(
7141 "{} {} ({})",
7142 quoted,
7143 if op == "_in" { "IN" } else { "NOT IN" },
7144 placeholders.join(", ")
7145 ))
7146 }
7147 "_has_keys_any" | "_has_keys_all" => {
7148 let items = operand.as_array().ok_or_else(|| {
7149 async_graphql::Error::new(format!(
7150 "the \"{}\" comparison on \"{}\" requires a list of keys",
7151 op, column
7152 ))
7153 })?;
7154 Ok(format!(
7155 "{} {} {}",
7156 quoted,
7157 if op == "_has_keys_any" { "?|" } else { "?&" },
7158 sql_array(items, "text[]", param_idx, values)
7159 ))
7160 }
7161 other => Err(async_graphql::Error::new(format!(
7162 "unsupported comparison \"{}\" on \"{}\"",
7163 other, column
7164 ))),
7165 }
7166}
7167
7168#[allow(clippy::too_many_arguments)] async fn build_order_by_clause(
7178 ctx: &ResolverContext<'_>,
7179 schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
7180 schema_name: &str,
7181 table_name: &str,
7182 type_name: &str,
7183 relationships: &HashMap<String, Vec<RelationshipField>>,
7184 reference: &str,
7185 names: &crate::names::NameOverrides,
7186) -> Result<String, async_graphql::Error> {
7187 let Ok(order_arg) = ctx.args.try_get("order_by") else {
7188 return Ok(String::new());
7189 };
7190
7191 let value = accessor_to_json(&order_arg);
7192 let entries: Vec<&serde_json::Value> = match &value {
7195 serde_json::Value::Array(items) => items.iter().collect(),
7196 serde_json::Value::Object(_) => vec![&value],
7197 serde_json::Value::Null => return Ok(String::new()),
7198 _ => {
7199 return Err(async_graphql::Error::new(
7200 "order_by takes an object or a list of them",
7201 ))
7202 }
7203 };
7204 if entries.is_empty() {
7205 return Ok(String::new());
7206 }
7207
7208 let guard = schema_cache
7209 .get()
7210 .await
7211 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7212 let cache = guard
7213 .as_ref()
7214 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
7215 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
7216 let table = cache
7217 .get_table(&qi)
7218 .ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
7219
7220 let mut terms = Vec::new();
7221 let mut alias_counter = 0usize;
7222 for entry in entries {
7223 order_terms_into(
7224 entry,
7225 cache,
7226 relationships,
7227 type_name,
7228 table,
7229 reference,
7230 names,
7231 &mut alias_counter,
7232 &mut terms,
7233 )?;
7234 }
7235
7236 if terms.is_empty() {
7237 return Ok(String::new());
7238 }
7239 Ok(format!(" ORDER BY {}", terms.join(", ")))
7240}
7241
7242#[allow(clippy::too_many_arguments)]
7250fn order_terms_into(
7251 entry: &serde_json::Value,
7252 cache: &SchemaCache,
7253 relationships: &HashMap<String, Vec<RelationshipField>>,
7254 type_name: &str,
7255 table: &postrust_core::schema_cache::Table,
7256 reference: &str,
7257 names: &crate::names::NameOverrides,
7258 alias_counter: &mut usize,
7259 terms: &mut Vec<String>,
7260) -> Result<(), async_graphql::Error> {
7261 let serde_json::Value::Object(map) = entry else {
7262 return Err(async_graphql::Error::new(
7263 "each order_by entry is an object mapping a column to a direction",
7264 ));
7265 };
7266 let available: &[RelationshipField] = relationships
7267 .get(type_name)
7268 .map(|r| r.as_slice())
7269 .unwrap_or(&[]);
7270
7271 for (key, value) in map {
7272 if value.is_null() {
7278 continue;
7279 }
7280
7281 if let Some(name) = value.as_str() {
7283 let sql = crate::input::order_by::direction_sql(name).ok_or_else(|| {
7284 async_graphql::Error::new(format!(
7285 "\"{}\" is not a sort direction; expected one of asc, desc, \
7286 asc_nulls_first, asc_nulls_last, desc_nulls_first, desc_nulls_last",
7287 name
7288 ))
7289 })?;
7290 let column = table_column_for(names, table, key);
7291 if table.get_column(column).is_some() {
7292 terms.push(format!(
7293 "{}.{} {}",
7294 reference,
7295 postrust_sql::escape_ident(column),
7296 sql
7297 ));
7298 continue;
7299 }
7300 let function = names
7303 .computed_source(&table.schema, &table.name, key)
7304 .unwrap_or(key);
7305 if let Some(definition) = table.get_computed_column(function) {
7306 terms.push(format!(
7307 "{} {}",
7308 computed_call(definition, &format!("{}.*", reference), reference),
7309 sql
7310 ));
7311 continue;
7312 }
7313 return Err(async_graphql::Error::new(format!(
7314 "cannot order by unknown column \"{}\" on \"{}\"",
7315 key, table.name
7316 )));
7317 }
7318
7319 if let Some(rel) = key
7321 .strip_suffix("_aggregate")
7322 .and_then(|name| available.iter().find(|r| r.name == name && r.is_list))
7323 {
7324 aggregate_order_terms(value, rel, cache, reference, alias_counter, terms)?;
7325 continue;
7326 }
7327
7328 if let Some(rel) = available.iter().find(|r| r.name == *key && !r.is_list) {
7330 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, cache)
7331 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7332 if plan.columns.is_empty() {
7333 return Err(async_graphql::Error::new(format!(
7334 "cannot order by \"{}\": it is not reached by a key",
7335 key
7336 )));
7337 }
7338 *alias_counter += 1;
7339 let alias = format!("pgrst_ord_{}", alias_counter);
7340 let quoted_alias = postrust_sql::escape_ident(&alias);
7341 let correlation = plan
7342 .columns
7343 .iter()
7344 .map(|(local, foreign)| {
7345 format!(
7346 "{} = {}.{}",
7347 format_args!("{}.{}", reference, postrust_sql::escape_ident(local)),
7348 quoted_alias,
7349 postrust_sql::escape_ident(foreign)
7350 )
7351 })
7352 .collect::<Vec<_>>()
7353 .join(" AND ");
7354
7355 let target_qi = postrust_core::api_request::QualifiedIdentifier::new(
7356 &plan.foreign_schema,
7357 &plan.foreign_table,
7358 );
7359 let Some(target) = cache.get_table(&target_qi) else {
7360 return Err(async_graphql::Error::new(format!(
7361 "unknown table \"{}\"",
7362 plan.foreign_table
7363 )));
7364 };
7365
7366 let mut nested = Vec::new();
7369 order_terms_into(
7370 value,
7371 cache,
7372 relationships,
7373 &rel.target_type,
7374 target,
7375 "ed_alias,
7376 names,
7377 alias_counter,
7378 &mut nested,
7379 )?;
7380 for term in nested {
7381 let (expression, direction) = term
7382 .rsplit_once(' ')
7383 .map(|(e, d)| (e.to_string(), d.to_string()))
7384 .unwrap_or((term.clone(), String::new()));
7385 terms.push(format!(
7386 "(SELECT {} FROM {}.{} AS {} WHERE {}) {}",
7387 expression,
7388 postrust_sql::escape_ident(&plan.foreign_schema),
7389 postrust_sql::escape_ident(&plan.foreign_table),
7390 quoted_alias,
7391 correlation,
7392 direction
7393 ));
7394 }
7395 continue;
7396 }
7397
7398 return Err(async_graphql::Error::new(format!(
7399 "cannot order by \"{}\" on \"{}\"",
7400 key, table.name
7401 )));
7402 }
7403 Ok(())
7404}
7405
7406fn aggregate_order_terms(
7408 value: &serde_json::Value,
7409 rel: &RelationshipField,
7410 cache: &SchemaCache,
7411 reference: &str,
7412 alias_counter: &mut usize,
7413 terms: &mut Vec<String>,
7414) -> Result<(), async_graphql::Error> {
7415 let serde_json::Value::Object(spec) = value else {
7416 return Err(async_graphql::Error::new(
7417 "ordering by an aggregate takes an object, such as {count: desc}",
7418 ));
7419 };
7420 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, cache)
7421 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7422 if plan.columns.is_empty() && plan.function.is_none() {
7423 return Err(async_graphql::Error::new(format!(
7424 "cannot order by an aggregate of \"{}\": it is not reached by a key",
7425 rel.name
7426 )));
7427 }
7428 if plan.function.is_some() && !rel.arguments.is_empty() {
7431 return Err(async_graphql::Error::new(format!(
7432 "cannot order by an aggregate of \"{}\": it takes arguments, and an \
7433 ordering has nowhere to write them",
7434 rel.name
7435 )));
7436 }
7437
7438 *alias_counter += 1;
7439 let alias = postrust_sql::escape_ident(&format!("pgrst_ord_{}", alias_counter));
7440 let (source, correlation) = match &plan.function {
7443 Some(function) => {
7444 let session = plan.row_argument.as_ref().and_then(|row_argument| {
7445 computed_session_argument(cache, function, row_argument)
7446 .map(|session| (row_argument.clone(), session))
7447 });
7448 let call = match session {
7449 Some((row_argument, session)) => format!(
7450 "{} => {}, {} => {}",
7451 postrust_sql::escape_ident(&row_argument),
7452 reference,
7453 postrust_sql::escape_ident(&session),
7454 SESSION_ARGUMENT
7455 ),
7456 None => reference.to_string(),
7457 };
7458 (
7459 format!(
7460 "{}.{}({})",
7461 postrust_sql::escape_ident(&function.schema),
7462 postrust_sql::escape_ident(&function.name),
7463 call
7464 ),
7465 "true".to_string(),
7466 )
7467 }
7468 None => (
7469 format!(
7470 "{}.{}",
7471 postrust_sql::escape_ident(&plan.foreign_schema),
7472 postrust_sql::escape_ident(&plan.foreign_table)
7473 ),
7474 plan.columns
7475 .iter()
7476 .map(|(local, foreign)| {
7477 format!(
7478 "{}.{} = {}.{}",
7479 reference,
7480 postrust_sql::escape_ident(local),
7481 alias,
7482 postrust_sql::escape_ident(foreign)
7483 )
7484 })
7485 .collect::<Vec<_>>()
7486 .join(" AND "),
7487 ),
7488 };
7489
7490 let mut wanted: Vec<(String, String)> = Vec::new();
7492 for (function, argument) in spec {
7493 if argument.is_null() {
7496 continue;
7497 }
7498 if function == "count" {
7499 let name = argument.as_str().unwrap_or_default();
7500 let Some(direction) = crate::input::order_by::direction_sql(name) else {
7501 return Err(async_graphql::Error::new(format!(
7502 "\"{}\" is not a sort direction",
7503 name
7504 )));
7505 };
7506 wanted.push(("count(*)".to_string(), direction.to_string()));
7507 continue;
7508 }
7509 let serde_json::Value::Object(columns) = argument else {
7510 continue;
7511 };
7512 for (column, direction) in columns {
7513 if direction.is_null() {
7514 continue;
7515 }
7516 let name = direction.as_str().unwrap_or_default();
7517 let Some(direction) = crate::input::order_by::direction_sql(name) else {
7518 return Err(async_graphql::Error::new(format!(
7519 "\"{}\" is not a sort direction",
7520 name
7521 )));
7522 };
7523 wanted.push((
7526 format!(
7527 "{}({}.{})",
7528 function,
7529 alias,
7530 postrust_sql::escape_ident(column)
7531 ),
7532 direction.to_string(),
7533 ));
7534 }
7535 }
7536
7537 for (expression, direction) in wanted {
7538 terms.push(format!(
7539 "(SELECT {} FROM {} AS {} WHERE {}) {}",
7540 expression, source, alias, correlation, direction
7541 ));
7542 }
7543 Ok(())
7544}
7545
7546fn distinct_on_clause(
7560 distinct_on: &[String],
7561 written: Option<&str>,
7562) -> Result<(String, Option<String>), async_graphql::Error> {
7563 if distinct_on.is_empty() {
7564 return Ok((String::new(), written.map(str::to_string)));
7565 }
7566 let terms: Vec<&str> = written
7567 .map(|written| written.split(", ").collect())
7568 .unwrap_or_default();
7569 let column_of = |term: &str| {
7572 let expression = term.split_whitespace().next().unwrap_or(term);
7573 expression
7574 .rsplit_once('.')
7575 .map(|(_, name)| name)
7576 .unwrap_or(expression)
7577 .to_string()
7578 };
7579 let leading: Vec<String> = terms
7580 .iter()
7581 .take(distinct_on.len())
7582 .map(|term| column_of(term))
7583 .collect();
7584 if !terms.is_empty()
7585 && (leading.len() != distinct_on.len()
7586 || distinct_on
7587 .iter()
7588 .any(|column| !leading.contains(&column_of(column))))
7589 {
7590 return Err(validation_error(
7591 "\"distinct_on\" columns must match initial \"order_by\" columns",
7592 ));
7593 }
7594 let order = match terms.is_empty() {
7595 true => Some(distinct_on.join(", ")),
7596 false => written.map(str::to_string),
7597 };
7598 Ok((format!("DISTINCT ON ({}) ", distinct_on.join(", ")), order))
7599}
7600
7601async fn build_distinct_on(
7609 ctx: &ResolverContext<'_>,
7610 schema_cache: &postrust_core::schema_cache::SchemaCacheRef,
7611 schema_name: &str,
7612 table_name: &str,
7613 names: &crate::names::NameOverrides,
7614) -> Result<Vec<String>, async_graphql::Error> {
7615 let Ok(arg) = ctx.args.try_get("distinct_on") else {
7616 return Ok(Vec::new());
7617 };
7618 let value = accessor_to_json(&arg);
7619 let requested: Vec<String> = match &value {
7620 serde_json::Value::Array(items) => items
7621 .iter()
7622 .filter_map(|i| i.as_str().map(|s| s.to_string()))
7623 .collect(),
7624 serde_json::Value::String(one) => vec![one.clone()],
7625 _ => Vec::new(),
7626 };
7627 if requested.is_empty() {
7628 return Ok(Vec::new());
7629 }
7630
7631 let guard = schema_cache
7632 .get()
7633 .await
7634 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
7635 let cache = guard
7636 .as_ref()
7637 .ok_or_else(|| async_graphql::Error::new("schema cache is not loaded"))?;
7638 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
7639 let table = cache
7640 .get_table(&qi)
7641 .ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
7642
7643 let mut quoted = Vec::with_capacity(requested.len());
7644 for field in requested {
7645 let name = table_column_for(names, table, &field).to_string();
7646 if table.get_column(&name).is_none() {
7647 return Err(async_graphql::Error::new(format!(
7648 "cannot take distinct on unknown column \"{}\" of \"{}\"",
7649 field, table_name
7650 )));
7651 }
7652 quoted.push(postrust_sql::escape_ident(&name));
7653 }
7654 Ok(quoted)
7655}
7656
7657#[allow(clippy::too_many_arguments)] fn computed_projections(
7671 table: &postrust_core::schema_cache::Table,
7672 selection: async_graphql::SelectionField<'_>,
7673 row_reference: &str,
7674 names: &crate::names::NameOverrides,
7675 schema_cache: &SchemaCache,
7676 param_idx: &mut usize,
7677 values: &mut Vec<serde_json::Value>,
7678) -> Result<Vec<String>, async_graphql::Error> {
7679 let mut projections = Vec::new();
7680 for field in selection.selection_set() {
7681 let name = field.name();
7682 if table
7685 .get_column(table_column_for(names, table, name))
7686 .is_some()
7687 {
7688 continue;
7689 }
7690 let function = names
7693 .computed_source(&table.schema, &table.name, name)
7694 .unwrap_or(name);
7695 let Some(computed) = table.get_computed_column(function) else {
7696 continue;
7697 };
7698 let arguments = crate::schema::computed_caller_arguments(computed, schema_cache);
7699 let given = match arguments.is_empty() {
7700 true => None,
7701 false => embed_arguments(field).remove("args"),
7702 };
7703 projections.push(format!(
7704 "{} AS {}",
7705 computed_column_call(
7706 computed,
7707 row_reference,
7708 name,
7709 &arguments,
7710 given.as_ref(),
7711 param_idx,
7712 values,
7713 )?,
7714 postrust_sql::escape_ident(name)
7715 ));
7716 }
7717 Ok(projections)
7718}
7719
7720fn computed_column_call(
7729 computed: &postrust_core::schema_cache::ComputedColumn,
7730 row: &str,
7731 field_name: &str,
7732 arguments: &[(String, String, bool)],
7733 given: Option<&serde_json::Value>,
7734 param_idx: &mut usize,
7735 values: &mut Vec<serde_json::Value>,
7736) -> Result<String, async_graphql::Error> {
7737 if arguments.is_empty() {
7738 return Ok(computed_call(computed, row, row));
7739 }
7740 let Some(row_argument) = &computed.row_argument else {
7741 return Ok(computed_call(computed, row, row));
7742 };
7743 let function = format!(
7744 "{}.{}",
7745 postrust_sql::escape_ident(&computed.function.schema),
7746 postrust_sql::escape_ident(&computed.function.name)
7747 );
7748 let mut passed = vec![format!(
7749 "{} => {}",
7750 postrust_sql::escape_ident(row_argument),
7751 row
7752 )];
7753 if let Some(session) = &computed.session_argument {
7754 passed.push(format!(
7755 "{} => {}",
7756 postrust_sql::escape_ident(session),
7757 SESSION_ARGUMENT
7758 ));
7759 }
7760 for (name, pg_type, required) in arguments {
7761 match given.and_then(|given| given.get(name)) {
7762 None if *required => {
7763 return Err(async_graphql::Error::new(format!(
7764 "{} needs the argument \"{}\"",
7765 field_name, name
7766 )))
7767 }
7768 None => continue,
7770 Some(value) => {
7771 passed.push(format!(
7772 "{} => ${}::{}",
7773 postrust_sql::escape_ident(name),
7774 param_idx,
7775 pg_type
7776 ));
7777 *param_idx += 1;
7778 values.push(value.clone());
7779 }
7780 }
7781 }
7782 Ok(format!("{}({})", function, passed.join(", ")))
7783}
7784
7785fn count_expression(
7797 field: async_graphql::SelectionField<'_>,
7798 table: (&str, &str),
7799 alias: Option<&str>,
7800 names: &crate::names::NameOverrides,
7801) -> String {
7802 let arguments = embed_arguments(field);
7803 let requested = match arguments.get("columns") {
7804 Some(serde_json::Value::Array(items)) => items
7805 .iter()
7806 .filter_map(|item| item.as_str())
7807 .collect::<Vec<_>>(),
7808 Some(serde_json::Value::String(one)) => vec![one.as_str()],
7809 _ => Vec::new(),
7810 };
7811 if requested.is_empty() {
7812 return "count(*)".to_string();
7813 }
7814 let columns: Vec<String> = requested
7815 .into_iter()
7816 .map(|column| {
7817 let (schema, name) = table;
7818 let source = column_for(names, schema, name, column);
7819 match alias {
7820 Some(alias) => format!(
7821 "{}.{}",
7822 postrust_sql::escape_ident(alias),
7823 postrust_sql::escape_ident(source)
7824 ),
7825 None => postrust_sql::escape_ident(source),
7826 }
7827 })
7828 .collect();
7829 let counted = match columns.len() {
7832 1 => columns.into_iter().next().unwrap_or_default(),
7833 _ => format!("({})", columns.join(", ")),
7834 };
7835 let distinct = matches!(
7836 arguments.get("distinct"),
7837 Some(serde_json::Value::Bool(true))
7838 );
7839 match distinct {
7840 true => format!("count(DISTINCT {})", counted),
7841 false => format!("count({})", counted),
7842 }
7843}
7844
7845#[allow(clippy::too_many_arguments)] fn nested_aggregate_select(
7853 caller: &crate::role::Caller<'_>,
7854 selection: async_graphql::SelectionField<'_>,
7855 child_alias: &str,
7856 schema_cache: &SchemaCache,
7857 relationships: &HashMap<String, Vec<RelationshipField>>,
7858 target_type: &str,
7859 child_table: Option<&postrust_core::schema_cache::Table>,
7860 max_rows: Option<i64>,
7861 alias_counter: &mut usize,
7862 param_idx: &mut usize,
7863 values: &mut Vec<serde_json::Value>,
7864 names: &crate::names::NameOverrides,
7865) -> Result<String, async_graphql::Error> {
7866 use crate::schema::aggregate as agg;
7867
7868 let mut parts: Vec<String> = Vec::new();
7869
7870 for child in selection.selection_set() {
7871 match child.name() {
7872 "aggregate" => {
7873 let of_table = child_table
7874 .map(|table| (table.schema.as_str(), table.name.as_str()))
7875 .unwrap_or(("", ""));
7876 let mut counts: Vec<(String, String)> = Vec::new();
7881 let mut wanted: BTreeMap<&str, Vec<String>> = BTreeMap::new();
7882 for function in child.selection_set() {
7883 if function.name() == "count" {
7884 counts.push((
7885 function.alias().unwrap_or("count").to_string(),
7886 count_expression(function, of_table, Some(child_alias), names),
7887 ));
7888 continue;
7889 }
7890 let sql_function = agg::NUMERIC_AGGREGATES
7891 .iter()
7892 .chain(agg::ORDERED_AGGREGATES.iter())
7893 .find(|(name, _)| *name == function.name())
7894 .map(|(name, _)| *name);
7895 let Some(sql_function) = sql_function else {
7896 continue;
7897 };
7898 let columns = wanted.entry(sql_function).or_default();
7899 for column in function.selection_set() {
7900 let name = column.name().to_string();
7901 if !columns.contains(&name) {
7902 columns.push(name);
7903 }
7904 }
7905 }
7906 let mut fields = match counts.iter().any(|(key, _)| key == "count") {
7909 true => Vec::new(),
7910 false => vec!["'count', count(*)".to_string()],
7911 };
7912 for (key, expression) in &counts {
7913 fields.push(format!("'{}', {}", key.replace('\'', "''"), expression));
7914 }
7915 for (sql_function, named) in &wanted {
7916 let columns: Vec<String> = named
7917 .iter()
7918 .map(|column| {
7919 let source = child_table
7920 .map(|table| table_column_for(names, table, column))
7921 .unwrap_or(column.as_str());
7922 format!(
7923 "'{}', {}({}.{})",
7924 column.replace('\'', "''"),
7925 sql_function,
7926 postrust_sql::escape_ident(child_alias),
7927 postrust_sql::escape_ident(source)
7928 )
7929 })
7930 .collect();
7931 if !columns.is_empty() {
7932 fields.push(format!(
7933 "'{}', json_build_object({})",
7934 sql_function,
7935 columns.join(", ")
7936 ));
7937 }
7938 }
7939 parts.push(format!(
7940 "json_build_object({}) AS {}",
7941 fields.join(", "),
7942 postrust_sql::escape_ident("aggregate")
7943 ));
7944 }
7945 "nodes" => {
7950 let embeds = build_embed_expressions(
7951 caller,
7952 schema_cache,
7953 relationships,
7954 target_type,
7955 child_alias,
7956 child,
7957 max_rows,
7958 alias_counter,
7959 param_idx,
7960 values,
7961 names,
7962 )?;
7963 let mut columns: Vec<String> = Vec::new();
7964 for column in child.selection_set() {
7965 let name = column.name();
7966 if name == "__typename" || embeds.iter().any(|(field, _)| field == name) {
7967 continue;
7968 }
7969 columns.push(format!(
7970 "'{}', {}",
7971 name.replace('\'', "''"),
7972 column_expression(child_table, child_alias, name, names)
7973 ));
7974 }
7975 for (field, expression) in &embeds {
7976 columns.push(format!("'{}', {}", field.replace('\'', "''"), expression));
7977 }
7978 let row = if columns.is_empty() {
7979 format!("row_to_json({})", postrust_sql::escape_ident(child_alias))
7980 } else {
7981 format!("json_build_object({})", columns.join(", "))
7982 };
7983 parts.push(format!(
7984 "COALESCE(json_agg({}), '[]'::json) AS {}",
7985 row,
7986 postrust_sql::escape_ident("nodes")
7987 ));
7988 }
7989 _ => {}
7990 }
7991 }
7992
7993 if parts.is_empty() {
7994 parts.push("count(*) AS pgrst_empty".to_string());
7997 }
7998 Ok(parts.join(", "))
7999}
8000
8001fn column_expression(
8006 table: Option<&postrust_core::schema_cache::Table>,
8007 alias: &str,
8008 name: &str,
8009 names: &crate::names::NameOverrides,
8010) -> String {
8011 let column = table
8012 .map(|t| table_column_for(names, t, name))
8013 .unwrap_or(name);
8014 let plain = format!(
8015 "{}.{}",
8016 postrust_sql::escape_ident(alias),
8017 postrust_sql::escape_ident(column)
8018 );
8019 let computed = table
8020 .filter(|t| t.get_column(column).is_none())
8021 .and_then(|t| {
8022 let function = names
8023 .computed_source(&t.schema, &t.name, name)
8024 .unwrap_or(name);
8025 t.get_computed_column(function)
8026 });
8027 match computed {
8028 Some(definition) => computed_call(
8029 definition,
8030 &format!("{}.*", postrust_sql::escape_ident(alias)),
8031 &postrust_sql::escape_ident(alias),
8032 ),
8033 None => plain,
8034 }
8035}
8036
8037fn order_terms(
8045 order: &serde_json::Value,
8046 schema_cache: &SchemaCache,
8047 schema_name: &str,
8048 table_name: &str,
8049 alias: &str,
8050 names: &crate::names::NameOverrides,
8051) -> Result<Option<String>, async_graphql::Error> {
8052 let entries: Vec<&serde_json::Value> = match order {
8053 serde_json::Value::Array(items) => items.iter().collect(),
8054 serde_json::Value::Object(_) => vec![order],
8055 _ => return Ok(None),
8056 };
8057
8058 let qi = postrust_core::api_request::QualifiedIdentifier::new(schema_name, table_name);
8059 let table = schema_cache
8060 .get_table(&qi)
8061 .ok_or_else(|| async_graphql::Error::new(format!("unknown table \"{}\"", table_name)))?;
8062
8063 let mut terms = Vec::new();
8064 for entry in entries {
8065 let serde_json::Value::Object(map) = entry else {
8066 continue;
8067 };
8068 for (field, direction) in map {
8069 if direction.is_null() {
8071 continue;
8072 }
8073 let column = table_column_for(names, table, field);
8074 if table.get_column(column).is_none() {
8075 return Err(async_graphql::Error::new(format!(
8076 "cannot order by unknown column \"{}\" on \"{}\"",
8077 field, table_name
8078 )));
8079 }
8080 let name = direction.as_str().unwrap_or_default();
8081 let sql = crate::input::order_by::direction_sql(name).ok_or_else(|| {
8082 async_graphql::Error::new(format!("\"{}\" is not a sort direction", name))
8083 })?;
8084 terms.push(format!(
8085 "{}.{} {}",
8086 postrust_sql::escape_ident(alias),
8087 postrust_sql::escape_ident(column),
8088 sql
8089 ));
8090 }
8091 }
8092
8093 Ok(if terms.is_empty() {
8094 None
8095 } else {
8096 Some(terms.join(", "))
8097 })
8098}
8099
8100fn always_present(rel: &RelationshipField, table: &postrust_core::schema_cache::Table) -> bool {
8112 use postrust_core::schema_cache::{Cardinality, Relationship};
8113 let Relationship::ForeignKey { cardinality, .. } = &rel.relationship else {
8114 return false;
8115 };
8116 match cardinality {
8117 Cardinality::M2O { .. } => {}
8118 Cardinality::O2O {
8119 is_parent: false, ..
8120 } => {}
8121 _ => return false,
8122 }
8123 let columns = cardinality.columns();
8124 !columns.is_empty()
8125 && columns.iter().all(|(local, _)| {
8126 table
8127 .get_column(local)
8128 .is_some_and(|column| !column.nullable)
8129 })
8130}
8131
8132fn child_holds_the_key(relationship: &postrust_core::schema_cache::Relationship) -> bool {
8140 use postrust_core::schema_cache::{Cardinality, Relationship};
8141 match relationship {
8142 Relationship::ForeignKey { cardinality, .. } => match cardinality {
8143 Cardinality::M2O { .. } => false,
8144 Cardinality::O2O { is_parent, .. } => *is_parent,
8145 _ => true,
8146 },
8147 Relationship::Computed { .. } => true,
8148 }
8149}
8150
8151fn computed_session_argument(
8153 cache: &SchemaCache,
8154 function: &postrust_core::api_request::QualifiedIdentifier,
8155 row_argument: &str,
8156) -> Option<String> {
8157 let _ = row_argument;
8158 cache
8159 .routines
8160 .get(function)
8161 .into_iter()
8162 .flatten()
8163 .find(|routine| routine.name == function.name)?
8164 .params
8165 .iter()
8166 .find(|param| {
8167 param.name == "hasura_session" && matches!(param.param_type.as_str(), "json" | "jsonb")
8168 })
8169 .map(|param| param.name.clone())
8170}
8171
8172fn computed_args_type_name(type_name: &str, field: &str) -> String {
8174 format!("{}_{}_args", type_name, field)
8175}
8176
8177fn computed_arguments(
8184 relationship: &RelationshipField,
8185 plan: &postrust_core::embed::EmbedPlan,
8186 parent_row: &str,
8187 given: Option<&serde_json::Value>,
8188 param_idx: &mut usize,
8189 values: &mut Vec<serde_json::Value>,
8190 schema_cache: &SchemaCache,
8191) -> Result<Option<String>, async_graphql::Error> {
8192 let (Some(function), Some(row_argument)) = (&plan.function, &plan.row_argument) else {
8193 return Ok(None);
8194 };
8195 let session = schema_cache
8196 .routines
8197 .get(function)
8198 .into_iter()
8199 .flatten()
8200 .find(|routine| routine.name == function.name)
8201 .and_then(|routine| {
8202 routine
8203 .params
8204 .iter()
8205 .find(|param| {
8206 param.name == "hasura_session"
8207 && matches!(param.param_type.as_str(), "json" | "jsonb")
8208 })
8209 .map(|param| param.name.clone())
8210 });
8211 if relationship.arguments.is_empty() && session.is_none() {
8212 return Ok(None);
8213 }
8214
8215 let mut passed = vec![format!(
8216 "{} => {}",
8217 postrust_sql::escape_ident(row_argument),
8218 parent_row
8219 )];
8220 if let Some(session) = session {
8221 passed.push(format!(
8222 "{} => {}",
8223 postrust_sql::escape_ident(&session),
8224 SESSION_ARGUMENT
8225 ));
8226 }
8227 for (name, pg_type, required) in &relationship.arguments {
8228 match given.and_then(|given| given.get(name)) {
8229 None if *required => {
8230 return Err(async_graphql::Error::new(format!(
8231 "{} needs the argument \"{}\"",
8232 relationship.name, name
8233 )))
8234 }
8235 None => continue,
8237 Some(value) => {
8238 passed.push(format!(
8239 "{} => ${}::{}",
8240 postrust_sql::escape_ident(name),
8241 param_idx,
8242 pg_type
8243 ));
8244 *param_idx += 1;
8245 values.push(value.clone());
8246 }
8247 }
8248 }
8249 Ok(Some(passed.join(", ")))
8250}
8251
8252fn embed_arguments(field: async_graphql::SelectionField<'_>) -> HashMap<String, serde_json::Value> {
8254 field
8255 .arguments()
8256 .map(|args| {
8257 args.into_iter()
8258 .map(|(name, value)| {
8259 (
8260 name.to_string(),
8261 plain_numbers(value.into_json().unwrap_or(serde_json::Value::Null)),
8262 )
8263 })
8264 .collect()
8265 })
8266 .unwrap_or_default()
8267}
8268
8269type Narrowing = (String, Option<String>, Option<String>, Option<i64>, i64);
8273
8274#[allow(clippy::too_many_arguments)] fn embed_narrowing(
8281 arguments: &HashMap<String, serde_json::Value>,
8282 plan: &postrust_core::embed::EmbedPlan,
8283 schema_cache: &SchemaCache,
8284 relationships: &HashMap<String, Vec<RelationshipField>>,
8285 target_type: &str,
8286 child_alias: &str,
8287 max_rows: Option<i64>,
8288 param_idx: &mut usize,
8289 values: &mut Vec<serde_json::Value>,
8290 names: &crate::names::NameOverrides,
8291 caller: &crate::role::Caller<'_>,
8292) -> Result<Narrowing, async_graphql::Error> {
8293 let permission =
8297 permission_predicate(caller, names, &plan.foreign_schema, &plan.foreign_table)?;
8298 let requested = arguments.get("where").filter(|v| !v.is_null()).cloned();
8299 let child_where = match and_predicates(requested, permission) {
8300 Some(expression) => {
8301 let child_scope = WhereScope::for_alias(
8302 &plan.foreign_schema,
8303 &plan.foreign_table,
8304 child_alias,
8305 target_type,
8306 schema_cache,
8307 relationships,
8308 names,
8309 )
8310 .for_caller(*caller);
8311 let mut nested_alias = 0usize;
8312 build_condition(
8313 &expression,
8314 &child_scope,
8315 param_idx,
8316 values,
8317 &mut nested_alias,
8318 )?
8319 }
8320 None => None,
8321 };
8322
8323 let child_order = match arguments.get("order_by") {
8324 Some(order) if !order.is_null() => order_terms(
8325 order,
8326 schema_cache,
8327 &plan.foreign_schema,
8328 &plan.foreign_table,
8329 child_alias,
8330 names,
8331 )?,
8332 _ => None,
8333 };
8334
8335 let ceiling = match (
8340 max_rows,
8341 crate::role::read_limit(caller, names, &plan.foreign_schema, &plan.foreign_table),
8342 ) {
8343 (Some(configured), Some(granted)) => Some(configured.min(granted)),
8344 (only, None) | (None, only) => only,
8345 };
8346 let child_limit = match arguments.get("limit").and_then(|v| v.as_i64()) {
8347 Some(requested) => match ceiling {
8348 Some(ceiling) => Some(requested.min(ceiling)),
8349 None => Some(requested),
8350 },
8351 None => ceiling,
8352 };
8353 let child_offset = arguments
8354 .get("offset")
8355 .and_then(|v| v.as_i64())
8356 .unwrap_or(0);
8357
8358 let requested = match arguments.get("distinct_on") {
8362 Some(serde_json::Value::Array(items)) => items
8363 .iter()
8364 .filter_map(|item| item.as_str())
8365 .collect::<Vec<_>>(),
8366 Some(serde_json::Value::String(one)) => vec![one.as_str()],
8367 _ => Vec::new(),
8368 };
8369 let mut distinct_on = Vec::with_capacity(requested.len());
8370 if !requested.is_empty() {
8371 let qi = postrust_core::api_request::QualifiedIdentifier::new(
8372 &plan.foreign_schema,
8373 &plan.foreign_table,
8374 );
8375 let table = schema_cache.get_table(&qi).ok_or_else(|| {
8376 async_graphql::Error::new(format!("unknown table \"{}\"", plan.foreign_table))
8377 })?;
8378 for field in requested {
8379 let column = table_column_for(names, table, field).to_string();
8380 if table.get_column(&column).is_none() {
8381 return Err(async_graphql::Error::new(format!(
8382 "cannot take distinct on unknown column \"{}\" of \"{}\"",
8383 field, plan.foreign_table
8384 )));
8385 }
8386 distinct_on.push(format!(
8387 "{}.{}",
8388 postrust_sql::escape_ident(child_alias),
8389 postrust_sql::escape_ident(&column)
8390 ));
8391 }
8392 }
8393 let (distinct, child_order) = distinct_on_clause(&distinct_on, child_order.as_deref())?;
8394
8395 Ok((
8396 distinct,
8397 child_where,
8398 child_order,
8399 child_limit,
8400 child_offset,
8401 ))
8402}
8403
8404#[allow(clippy::too_many_arguments)] fn build_embed_expressions(
8406 caller: &crate::role::Caller<'_>,
8407 schema_cache: &SchemaCache,
8408 relationships: &HashMap<String, Vec<RelationshipField>>,
8409 type_name: &str,
8410 parent_alias: &str,
8411 selection: async_graphql::SelectionField<'_>,
8412 max_rows: Option<i64>,
8413 alias_counter: &mut usize,
8414 param_idx: &mut usize,
8415 values: &mut Vec<serde_json::Value>,
8416 names: &crate::names::NameOverrides,
8417) -> Result<Vec<(String, String)>, async_graphql::Error> {
8418 let Some(available) = relationships.get(type_name) else {
8419 return Ok(Vec::new());
8420 };
8421
8422 let mut expressions = Vec::new();
8423
8424 for field in selection.selection_set() {
8425 let aggregate_of = field
8429 .name()
8430 .strip_suffix("_aggregate")
8431 .and_then(|name| available.iter().find(|r| r.name == name && r.is_list));
8432
8433 if let Some(rel) = aggregate_of {
8434 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, schema_cache)
8435 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8436 *alias_counter += 1;
8437 let child_alias = format!("a{}", alias_counter);
8438
8439 let arguments = embed_arguments(field);
8443 let (child_distinct, child_where, child_order, child_limit, child_offset) =
8444 embed_narrowing(
8445 &arguments,
8446 &plan,
8447 schema_cache,
8448 relationships,
8449 &rel.target_type,
8450 &child_alias,
8451 max_rows,
8452 param_idx,
8453 values,
8454 names,
8455 caller,
8456 )?;
8457
8458 let child_table =
8459 schema_cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
8460 &plan.foreign_schema,
8461 &plan.foreign_table,
8462 ));
8463 let select = nested_aggregate_select(
8464 caller,
8465 field,
8466 &child_alias,
8467 schema_cache,
8468 relationships,
8469 &rel.target_type,
8470 child_table,
8471 max_rows,
8472 alias_counter,
8473 param_idx,
8474 values,
8475 names,
8476 )?;
8477
8478 let call = computed_arguments(
8479 rel,
8480 &plan,
8481 &postrust_sql::escape_ident(parent_alias),
8482 arguments.get("args"),
8483 param_idx,
8484 values,
8485 schema_cache,
8486 )?;
8487 let expression = plan
8488 .aggregate_expression(
8489 parent_alias,
8490 &postrust_sql::escape_ident(parent_alias),
8491 &child_alias,
8492 &select,
8493 child_limit,
8494 child_offset,
8495 child_where.as_deref(),
8496 child_order.as_deref(),
8497 call.as_deref(),
8498 &child_distinct,
8499 )
8500 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8501 expressions.push((field.name().to_string(), expression));
8502 continue;
8503 }
8504
8505 let Some(rel) = available.iter().find(|r| r.name == field.name()) else {
8506 continue;
8507 };
8508
8509 let plan = postrust_core::embed::EmbedPlan::resolve(&rel.relationship, schema_cache)
8510 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8511
8512 *alias_counter += 1;
8513 let child_alias = format!("e{}", alias_counter);
8514
8515 let nested = build_embed_expressions(
8516 caller,
8517 schema_cache,
8518 relationships,
8519 &rel.target_type,
8520 &child_alias,
8521 field,
8522 max_rows,
8523 alias_counter,
8524 param_idx,
8525 values,
8526 names,
8527 )?;
8528
8529 let arguments = embed_arguments(field);
8531 let (child_distinct, child_where, child_order, child_limit, child_offset) =
8532 embed_narrowing(
8533 &arguments,
8534 &plan,
8535 schema_cache,
8536 relationships,
8537 &rel.target_type,
8538 &child_alias,
8539 max_rows,
8540 param_idx,
8541 values,
8542 names,
8543 caller,
8544 )?;
8545
8546 let child_relationships = relationships.get(&rel.target_type);
8549 let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
8550 &plan.foreign_schema,
8551 &plan.foreign_table,
8552 );
8553 let child_table = schema_cache.get_table(&child_qi);
8554 let mut parts: Vec<String> = Vec::new();
8555 for sub in field.selection_set() {
8556 let name = sub.name();
8557 let is_relationship = child_relationships
8558 .map(|rels| {
8559 rels.iter().any(|r| {
8560 r.name == name || (r.is_list && format!("{}_aggregate", r.name) == name)
8561 })
8562 })
8563 .unwrap_or(false);
8564 if is_relationship {
8565 continue;
8566 }
8567 let column = child_table
8570 .map(|t| table_column_for(names, t, name))
8571 .unwrap_or(name);
8572 let computed = child_table
8573 .filter(|t| t.get_column(column).is_none())
8574 .and_then(|t| {
8575 let function = names
8576 .computed_source(&t.schema, &t.name, name)
8577 .unwrap_or(name);
8578 t.get_computed_column(function)
8579 });
8580 match computed {
8581 Some(definition) => parts.push(format!(
8582 "{} AS {}",
8583 computed_call(
8584 definition,
8585 &format!("{}.*", postrust_sql::escape_ident(&child_alias)),
8586 &postrust_sql::escape_ident(&child_alias),
8587 ),
8588 postrust_sql::escape_ident(name)
8589 )),
8590 None if column != name => parts.push(format!(
8593 "{} AS {}",
8594 postrust_sql::escape_ident(column),
8595 postrust_sql::escape_ident(name)
8596 )),
8597 None => parts.push(postrust_sql::escape_ident(name)),
8598 }
8599 }
8600 if parts.is_empty() && nested.is_empty() {
8601 parts.push(
8602 child_table
8603 .and_then(|table| rename_projection(table, &child_alias, names))
8604 .unwrap_or_else(|| format!("{}.*", postrust_sql::escape_ident(&child_alias))),
8605 );
8606 }
8607 for (field_name, expression) in nested {
8608 parts.push(format!(
8609 "{} AS {}",
8610 expression,
8611 postrust_sql::escape_ident(&field_name)
8612 ));
8613 }
8614
8615 let computed_call_arguments = computed_arguments(
8616 rel,
8617 &plan,
8618 &postrust_sql::escape_ident(parent_alias),
8619 arguments.get("args"),
8620 param_idx,
8621 values,
8622 schema_cache,
8623 )?;
8624 let expression = plan
8625 .embed_expression(
8626 parent_alias,
8627 &postrust_sql::escape_ident(parent_alias),
8631 &child_alias,
8632 &format!("{}{}", child_distinct, parts.join(", ")),
8633 child_limit,
8634 child_offset,
8635 child_where.as_deref(),
8636 child_order.as_deref(),
8637 computed_call_arguments.as_deref(),
8638 )
8639 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8640
8641 expressions.push((rel.name.clone(), expression));
8642 }
8643
8644 Ok(expressions)
8645}
8646
8647struct EmbedContext<'c> {
8653 schema_cache: &'c SchemaCache,
8654 relationships: &'c HashMap<String, Vec<RelationshipField>>,
8655 max_rows: Option<i64>,
8656}
8657
8658fn embed_relationships<'f>(
8664 conn: &'f mut sqlx::PgConnection,
8665 ctx: &'f EmbedContext<'f>,
8666 type_name: &'f str,
8667 selection: async_graphql::SelectionField<'f>,
8668 rows: &'f mut [serde_json::Value],
8669) -> futures::future::BoxFuture<'f, Result<(), async_graphql::Error>> {
8670 Box::pin(async move {
8671 if rows.is_empty() {
8672 return Ok(());
8673 }
8674
8675 let Some(available) = ctx.relationships.get(type_name) else {
8676 return Ok(());
8677 };
8678
8679 for requested in selection.selection_set() {
8680 let Some(rel) = available.iter().find(|r| r.name == requested.name()) else {
8681 continue;
8682 };
8683
8684 let plan =
8685 postrust_core::embed::EmbedPlan::resolve(&rel.relationship, ctx.schema_cache)
8686 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8687
8688 let keys = postrust_core::embed::parent_keys(rows, &plan.local_column);
8689
8690 let mut child_columns: Vec<String> = Vec::new();
8700 let mut project_everything = false;
8701 let child_relationships = ctx.relationships.get(&rel.target_type);
8702 for field in requested.selection_set() {
8703 let name = field.name();
8704 match child_relationships.and_then(|rels| rels.iter().find(|r| r.name == name)) {
8705 Some(nested_rel) => {
8706 match postrust_core::embed::EmbedPlan::resolve(
8707 &nested_rel.relationship,
8708 ctx.schema_cache,
8709 ) {
8710 Ok(nested_plan) => child_columns.push(nested_plan.local_column),
8711 Err(_) => project_everything = true,
8712 }
8713 }
8714 None => child_columns.push(name.to_string()),
8715 }
8716 }
8717 if project_everything || child_columns.is_empty() {
8718 child_columns.clear();
8719 }
8720
8721 let mut grouped = if keys.is_empty() {
8722 std::collections::HashMap::new()
8723 } else {
8724 let sql = plan
8725 .children_grouped_sql(ctx.max_rows, &child_columns)
8726 .map_err(|e| async_graphql::Error::new(e.to_string()))?;
8727
8728 let fetched = sqlx::query(&sql)
8729 .bind(&keys)
8730 .fetch_all(&mut *conn)
8731 .await
8732 .map_err(database_error)?;
8733
8734 use sqlx::Row;
8737 let pairs: Vec<(serde_json::Value, serde_json::Value)> = fetched
8738 .into_iter()
8739 .filter_map(|row| {
8740 Some((
8741 row.try_get::<serde_json::Value, _>(0).ok()?,
8742 row.try_get::<serde_json::Value, _>(1).ok()?,
8743 ))
8744 })
8745 .collect();
8746
8747 postrust_core::embed::group_from_aggregated(pairs)
8748 };
8749
8750 let has_deeper_embed = ctx
8755 .relationships
8756 .get(&rel.target_type)
8757 .map(|rels| {
8758 requested
8759 .selection_set()
8760 .any(|field| rels.iter().any(|r| r.name == field.name()))
8761 })
8762 .unwrap_or(false);
8763
8764 if has_deeper_embed {
8765 let mut order: Vec<(String, usize)> = Vec::with_capacity(grouped.len());
8766 let mut flat: Vec<serde_json::Value> = Vec::new();
8767 for (key, children) in grouped.drain() {
8768 order.push((key, children.len()));
8769 flat.extend(children);
8770 }
8771
8772 embed_relationships(&mut *conn, ctx, &rel.target_type, requested, &mut flat)
8773 .await?;
8774
8775 let mut rest = flat.into_iter();
8776 for (key, count) in order {
8777 grouped.insert(key, rest.by_ref().take(count).collect());
8778 }
8779 }
8780
8781 for row in rows.iter_mut() {
8782 postrust_core::embed::attach_to_parent(row, &rel.name, &plan, &grouped);
8783 }
8784 }
8785
8786 Ok(())
8787 })
8788}
8789
8790fn pk_where_from_args(
8795 ctx: &ResolverContext<'_>,
8796 schema_name: &str,
8797 table_name: &str,
8798 pk_columns: &[(String, String)],
8799 names: &crate::names::NameOverrides,
8800) -> Result<serde_json::Value, async_graphql::Error> {
8801 if pk_columns.is_empty() {
8802 return Err(async_graphql::Error::new(format!(
8803 "\"{}\" has no primary key, so it cannot be mutated by key",
8804 table_name
8805 )));
8806 }
8807
8808 let from_object = ctx
8812 .args
8813 .try_get("pk_columns")
8814 .ok()
8815 .map(|v| accessor_to_json(&v));
8816
8817 let mut conditions = serde_json::Map::new();
8818 for (col_name, _) in pk_columns {
8819 let field = names
8823 .column(schema_name, table_name, col_name)
8824 .unwrap_or(col_name);
8825 let value = match &from_object {
8826 Some(serde_json::Value::Object(map)) => map.get(field).cloned().ok_or_else(|| {
8827 async_graphql::Error::new(format!(
8828 "pk_columns is missing the key column \"{}\"",
8829 field
8830 ))
8831 })?,
8832 _ => {
8833 let arg = ctx.args.try_get(field).map_err(|_| {
8834 async_graphql::Error::new(format!(
8835 "missing required primary key argument \"{}\"",
8836 field
8837 ))
8838 })?;
8839 accessor_to_json(&arg)
8840 }
8841 };
8842 conditions.insert(field.to_string(), serde_json::json!({ "_eq": value }));
8843 }
8844
8845 Ok(serde_json::Value::Object(conditions))
8846}
8847
8848fn pk_argument_type(pg_type: &str) -> String {
8855 let rendered = crate::types::pg_type_to_graphql(pg_type).to_string();
8856 if rendered.starts_with('[') {
8857 "String".to_string()
8858 } else {
8859 rendered
8860 }
8861}
8862
8863fn graphql_type_ref(type_str: &str) -> TypeRef {
8865 let is_list = type_str.starts_with('[');
8867 let is_nn = type_str.ends_with('!');
8868
8869 let inner = if is_list {
8871 let stripped = type_str
8872 .trim_end_matches('!') .trim_start_matches('[') .trim_end_matches(']'); stripped
8876 } else {
8877 type_str.trim_end_matches('!')
8878 };
8879
8880 let inner_nn = inner.ends_with('!');
8881 let base_type = inner.trim_end_matches('!');
8882
8883 if is_list {
8884 if is_nn {
8885 if inner_nn {
8886 TypeRef::named_nn_list_nn(base_type)
8887 } else {
8888 TypeRef::named_list_nn(base_type)
8889 }
8890 } else if inner_nn {
8891 TypeRef::named_nn_list(base_type)
8892 } else {
8893 TypeRef::named_list(base_type)
8894 }
8895 } else if is_nn {
8896 TypeRef::named_nn(base_type)
8897 } else {
8898 TypeRef::named(base_type)
8899 }
8900}
8901
8902const NUMBER_AS_TEXT: &str = "$serde_json::private::Number";
8912
8913fn number_kept_as_text(
8916 map: &serde_json::Map<String, serde_json::Value>,
8917) -> Option<serde_json::Value> {
8918 if map.len() != 1 {
8919 return None;
8920 }
8921 let text = map.get(NUMBER_AS_TEXT)?.as_str()?;
8922 serde_json::from_str::<serde_json::Value>(text)
8923 .ok()
8924 .filter(|value| value.is_number())
8925}
8926
8927fn plain_numbers(value: serde_json::Value) -> serde_json::Value {
8929 match value {
8930 serde_json::Value::Object(map) => match number_kept_as_text(&map) {
8931 Some(number) => number,
8932 None => serde_json::Value::Object(
8933 map.into_iter()
8934 .map(|(k, v)| (k, plain_numbers(v)))
8935 .collect(),
8936 ),
8937 },
8938 serde_json::Value::Array(items) => {
8939 serde_json::Value::Array(items.into_iter().map(plain_numbers).collect())
8940 }
8941 other => other,
8942 }
8943}
8944
8945fn accessor_to_json(accessor: &ValueAccessor<'_>) -> serde_json::Value {
8947 if accessor.is_null() {
8949 serde_json::Value::Null
8950 } else if let Ok(b) = accessor.boolean() {
8951 serde_json::Value::Bool(b)
8952 } else if let Ok(i) = accessor.i64() {
8953 serde_json::Value::Number(i.into())
8954 } else if let Ok(f) = accessor.f64() {
8955 serde_json::Number::from_f64(f)
8956 .map(serde_json::Value::Number)
8957 .unwrap_or(serde_json::Value::Null)
8958 } else if let Ok(s) = accessor.string() {
8959 serde_json::Value::String(s.to_string())
8960 } else if let Ok(name) = accessor.enum_name() {
8961 serde_json::Value::String(name.to_string())
8964 } else if let Ok(list) = accessor.list() {
8965 serde_json::Value::Array(list.iter().map(|v| accessor_to_json(&v)).collect())
8966 } else if let Ok(obj) = accessor.object() {
8967 let map: serde_json::Map<String, serde_json::Value> = obj
8968 .iter()
8969 .map(|(k, v)| (k.to_string(), accessor_to_json(&v)))
8970 .collect();
8971 match number_kept_as_text(&map) {
8972 Some(number) => number,
8973 None => serde_json::Value::Object(map),
8974 }
8975 } else {
8976 serde_json::Value::Null
8977 }
8978}
8979
8980fn value_to_json(value: &Value) -> serde_json::Value {
8982 match value {
8983 Value::Null => serde_json::Value::Null,
8984 Value::Boolean(b) => serde_json::Value::Bool(*b),
8985 Value::Number(n) => {
8986 if let Some(i) = n.as_i64() {
8987 serde_json::Value::Number(i.into())
8988 } else if let Some(f) = n.as_f64() {
8989 serde_json::Value::Number(serde_json::Number::from_f64(f).unwrap())
8990 } else {
8991 serde_json::Value::Null
8992 }
8993 }
8994 Value::String(s) => serde_json::Value::String(s.clone()),
8995 Value::List(arr) => serde_json::Value::Array(arr.iter().map(value_to_json).collect()),
8996 Value::Object(obj) => {
8997 let map: serde_json::Map<String, serde_json::Value> = obj
8998 .iter()
8999 .map(|(k, v)| (k.to_string(), value_to_json(v)))
9000 .collect();
9001 serde_json::Value::Object(map)
9002 }
9003 Value::Binary(b) => serde_json::Value::String(base64::Engine::encode(
9004 &base64::engine::general_purpose::STANDARD,
9005 b,
9006 )),
9007 Value::Enum(e) => serde_json::Value::String(e.to_string()),
9008 }
9009}
9010
9011fn json_to_value(json: serde_json::Value) -> Value {
9013 match json {
9014 serde_json::Value::Null => Value::Null,
9015 serde_json::Value::Bool(b) => Value::Boolean(b),
9016 serde_json::Value::Number(n) => {
9017 if let Some(i) = n.as_i64() {
9018 Value::Number(i.into())
9019 } else if let Some(f) = n.as_f64() {
9020 Value::Number(async_graphql::Number::from_f64(f).unwrap())
9021 } else {
9022 Value::Null
9023 }
9024 }
9025 serde_json::Value::String(s) => Value::String(s),
9026 serde_json::Value::Array(arr) => Value::List(arr.into_iter().map(json_to_value).collect()),
9027 serde_json::Value::Object(obj) => {
9028 let map: indexmap::IndexMap<async_graphql::Name, Value> = obj
9029 .into_iter()
9030 .map(|(k, v)| (async_graphql::Name::new(k), json_to_value(v)))
9031 .collect();
9032 Value::Object(map)
9033 }
9034 }
9035}
9036
9037#[cfg(test)]
9047mod tests {
9048 use super::*;
9049 use indexmap::IndexMap;
9050 use postrust_core::schema_cache::{Column, Table};
9051 use std::collections::{HashMap, HashSet};
9052
9053 fn create_test_table(name: &str) -> Table {
9054 let mut columns = IndexMap::new();
9055 columns.insert(
9056 "id".into(),
9057 Column {
9058 name: "id".into(),
9059 description: None,
9060 nullable: false,
9061 data_type: "integer".into(),
9062 nominal_type: "int4".into(),
9063 max_len: None,
9064 default: Some("nextval('id_seq')".into()),
9065 enum_values: vec![],
9066 is_pk: true,
9067 position: 1,
9068 always_generated: false,
9069 domain_type: None,
9070 },
9071 );
9072 columns.insert(
9073 "name".into(),
9074 Column {
9075 name: "name".into(),
9076 description: None,
9077 nullable: false,
9078 data_type: "text".into(),
9079 nominal_type: "text".into(),
9080 max_len: None,
9081 default: None,
9082 enum_values: vec![],
9083 is_pk: false,
9084 position: 2,
9085 always_generated: false,
9086 domain_type: None,
9087 },
9088 );
9089
9090 Table {
9091 schema: "public".into(),
9092 name: name.into(),
9093 description: None,
9094 is_view: false,
9095 insertable: true,
9096 updatable: true,
9097 deletable: true,
9098 pk_cols: vec!["id".into()],
9099 unique_constraints: Vec::new(),
9100 columns,
9101 computed_columns: Default::default(),
9102 is_partitioned: false,
9103 }
9104 }
9105
9106 fn create_test_schema_cache() -> SchemaCache {
9107 let mut tables = HashMap::new();
9108 let users = create_test_table("users");
9109 tables.insert(users.qualified_identifier(), users);
9110
9111 SchemaCache {
9112 tables,
9113 relationships: HashMap::new(),
9114 routines: HashMap::new(),
9115 timezones: HashSet::new(),
9116 media_handlers: HashMap::new(),
9117 pg_version: 150000,
9118 representations: Default::default(),
9119 }
9120 }
9121
9122 #[test]
9127 fn test_graphql_type_ref_simple() {
9128 let _type_ref = graphql_type_ref("String");
9129 }
9131
9132 #[test]
9133 fn test_graphql_type_ref_non_null() {
9134 let _type_ref = graphql_type_ref("String!");
9135 }
9136
9137 #[test]
9138 fn test_graphql_type_ref_list() {
9139 let _type_ref = graphql_type_ref("[String]");
9140 }
9141
9142 #[test]
9143 fn test_graphql_type_ref_list_non_null() {
9144 let _type_ref = graphql_type_ref("[String!]!");
9145 }
9146
9147 #[test]
9152 fn test_value_to_json_null() {
9153 let value = Value::Null;
9154 let json = value_to_json(&value);
9155 assert_eq!(json, serde_json::Value::Null);
9156 }
9157
9158 #[test]
9159 fn test_value_to_json_boolean() {
9160 let value = Value::Boolean(true);
9161 let json = value_to_json(&value);
9162 assert_eq!(json, serde_json::Value::Bool(true));
9163 }
9164
9165 #[test]
9166 fn test_value_to_json_number() {
9167 let value = Value::Number(42.into());
9168 let json = value_to_json(&value);
9169 assert_eq!(json, serde_json::json!(42));
9170 }
9171
9172 #[test]
9173 fn test_value_to_json_string() {
9174 let value = Value::String("hello".to_string());
9175 let json = value_to_json(&value);
9176 assert_eq!(json, serde_json::Value::String("hello".to_string()));
9177 }
9178
9179 #[test]
9180 fn test_value_to_json_list() {
9181 let value = Value::List(vec![Value::Number(1.into()), Value::Number(2.into())]);
9182 let json = value_to_json(&value);
9183 assert_eq!(json, serde_json::json!([1, 2]));
9184 }
9185
9186 #[test]
9187 fn test_json_to_value_null() {
9188 let json = serde_json::Value::Null;
9189 let value = json_to_value(json);
9190 assert!(matches!(value, Value::Null));
9191 }
9192
9193 #[test]
9194 fn test_json_to_value_boolean() {
9195 let json = serde_json::Value::Bool(false);
9196 let value = json_to_value(json);
9197 assert!(matches!(value, Value::Boolean(false)));
9198 }
9199
9200 #[test]
9201 fn test_json_to_value_number() {
9202 let json = serde_json::json!(123);
9203 let value = json_to_value(json);
9204 assert!(matches!(value, Value::Number(_)));
9205 }
9206
9207 #[test]
9208 fn test_json_to_value_string() {
9209 let json = serde_json::Value::String("test".to_string());
9210 let value = json_to_value(json);
9211 assert!(matches!(value, Value::String(_)));
9212 }
9213
9214 #[test]
9215 fn test_json_to_value_array() {
9216 let json = serde_json::json!([1, 2, 3]);
9217 let value = json_to_value(json);
9218 assert!(matches!(value, Value::List(_)));
9219 }
9220
9221 #[test]
9222 fn test_json_to_value_object() {
9223 let json = serde_json::json!({"key": "value"});
9224 let value = json_to_value(json);
9225 assert!(matches!(value, Value::Object(_)));
9226 }
9227
9228 #[test]
9233 fn test_build_dynamic_schema() {
9234 let cache = create_test_schema_cache();
9235 let config = SchemaConfig::default();
9236 let generated = build_schema(&cache, &config);
9237
9238 let result = build_dynamic_schema(
9239 &generated,
9240 &cache,
9241 None,
9242 None,
9243 Arc::new(Default::default()),
9244 std::time::Duration::from_secs(30),
9245 None,
9246 );
9247 if let Err(ref e) = result {
9248 eprintln!("Schema build error: {:?}", e);
9249 }
9250 assert!(result.is_ok(), "Schema build failed: {:?}", result.err());
9251 }
9252
9253 #[test]
9256 fn exists_is_a_subselect_over_another_table() {
9257 let cache = create_test_schema_cache();
9258 let names = crate::names::NameOverrides::default();
9259 let relationships: HashMap<String, Vec<RelationshipField>> = HashMap::new();
9260 let scope = WhereScope::table("public", "users", "users", &names)
9261 .with_resolution(&cache, &relationships);
9262
9263 let mut param_idx = 1usize;
9264 let mut values: Vec<serde_json::Value> = Vec::new();
9265 let mut aliases = 0usize;
9266 let sql = build_condition(
9267 &serde_json::json!({
9268 "_exists": {"_table": "users", "_where": {"name": {"_eq": "Ann"}}}
9269 }),
9270 &scope,
9271 &mut param_idx,
9272 &mut values,
9273 &mut aliases,
9274 )
9275 .expect("compiles")
9276 .expect("constrains something");
9277
9278 assert_eq!(
9279 sql,
9280 "EXISTS (SELECT 1 FROM \"public\".\"users\" AS \"pgrst_exists_1\" \
9281 WHERE \"pgrst_exists_1\".\"name\" = $1::text)"
9282 );
9283 assert_eq!(values, vec![serde_json::json!("Ann")]);
9284 }
9285
9286 #[test]
9288 fn exists_names_a_table_the_server_has() {
9289 let cache = create_test_schema_cache();
9290 let names = crate::names::NameOverrides::default();
9291 let relationships: HashMap<String, Vec<RelationshipField>> = HashMap::new();
9292 let scope = WhereScope::table("public", "users", "users", &names)
9293 .with_resolution(&cache, &relationships);
9294 let error = build_condition(
9295 &serde_json::json!({"_exists": {"_table": "nowhere", "_where": {}}}),
9296 &scope,
9297 &mut 1,
9298 &mut Vec::new(),
9299 &mut 0,
9300 )
9301 .expect_err("a table that is not there is an error");
9302 assert!(
9303 error.message.contains("public.nowhere"),
9304 "{}",
9305 error.message
9306 );
9307 }
9308
9309 #[test]
9316 fn a_table_a_role_may_count_and_not_read_has_a_count_and_no_rows() {
9317 let cache = create_test_schema_cache();
9318 let names = crate::names::NameOverrides::parse(
9319 r#"{"tables": {"public.users": {"permissions":
9320 {"counter": {"select": {"columns": [], "filter": {},
9321 "allow_aggregations": true}}}}}}"#,
9322 )
9323 .unwrap();
9324 let view = crate::role::cache_for_role(&cache, &names, "counter", false);
9325 assert!(!view.tables.is_empty(), "the table is there to be counted");
9326
9327 let config = SchemaConfig {
9328 names: names.clone(),
9329 role: Some("counter".to_string()),
9330 ..SchemaConfig::default()
9331 };
9332 let generated = build_schema(&view, &config);
9333 let schema = build_dynamic_schema(
9334 &generated,
9335 &view,
9336 None,
9337 None,
9338 Arc::new(names),
9339 std::time::Duration::from_secs(30),
9340 Some("counter"),
9341 )
9342 .expect("a role that may only count still has a buildable schema");
9343 let sdl = schema.sdl();
9344
9345 assert!(
9346 sdl.contains("users_aggregate"),
9347 "the count is there:\n{}",
9348 sdl
9349 );
9350 assert!(sdl.contains("count: Int!"), "and it is a count:\n{}", sdl);
9351 assert!(!sdl.contains("type users "), "no row type:\n{}", sdl);
9353 assert!(
9354 !sdl.contains("nodes"),
9355 "and no rows beside the numbers:\n{}",
9356 sdl
9357 );
9358 assert!(
9359 !sdl.contains("users_select_column"),
9360 "no column enum to order by or count over:\n{}",
9361 sdl
9362 );
9363 assert!(!sdl.contains("users_max_fields"), "no max:\n{}", sdl);
9365 }
9366
9367 #[test]
9374 fn a_table_a_role_may_only_write_gets_the_write_and_no_type() {
9375 let cache = create_test_schema_cache();
9376 let names = crate::names::NameOverrides::parse(
9377 r#"{"tables": {"public.users": {"permissions":
9378 {"writer": {"insert": {"columns": "*", "check": {}}}}}}}"#,
9379 )
9380 .unwrap();
9381 let view = crate::role::cache_for_role(&cache, &names, "writer", false);
9382 let config = SchemaConfig {
9383 names: names.clone(),
9384 role: Some("writer".to_string()),
9385 ..SchemaConfig::default()
9386 };
9387 let generated = build_schema(&view, &config);
9388
9389 let schema = build_dynamic_schema(
9390 &generated,
9391 &view,
9392 None,
9393 None,
9394 Arc::new(names),
9395 std::time::Duration::from_secs(30),
9396 Some("writer"),
9397 )
9398 .expect("a role that may only write still has a buildable schema");
9399 let sdl = schema.sdl();
9400
9401 assert!(
9402 sdl.contains("insert_users("),
9403 "the write is there:\n{}",
9404 sdl
9405 );
9406 assert!(
9407 sdl.contains("users_insert_input"),
9408 "and its input:\n{}",
9409 sdl
9410 );
9411 assert!(!sdl.contains("type users "), "no row type:\n{}", sdl);
9413 assert!(!sdl.contains("insert_users_one"), "no insert_one:\n{}", sdl);
9414 assert!(
9415 !sdl.contains("returning"),
9416 "and nothing to return:\n{}",
9417 sdl
9418 );
9419 assert!(
9420 sdl.contains("users_mutation_response"),
9421 "the count is still answered:\n{}",
9422 sdl
9423 );
9424 assert!(
9425 !sdl.contains("_mutation_response_mutation_response"),
9426 "and no type is named twice over:\n{}",
9427 sdl
9428 );
9429 }
9430
9431 fn cache_with_a_mutation_function() -> SchemaCache {
9434 use postrust_core::schema_cache::{FuncVolatility, RetType, Routine};
9435
9436 let mut cache = create_test_schema_cache();
9437 let routine = Routine {
9438 schema: "public".into(),
9439 name: "add_to_score".into(),
9440 description: None,
9441 params: Vec::new(),
9442 return_type: RetType::SetOf("users".into()),
9443 returns_composite: true,
9444 volatility: FuncVolatility::Volatile,
9445 has_variadic: false,
9446 isolation_level: None,
9447 settings: Vec::new(),
9448 is_procedure: false,
9449 media_type: None,
9450 media_base_type: None,
9451 output_columns: Vec::new(),
9452 };
9453 cache
9454 .routines
9455 .insert(routine.qualified_identifier(), vec![routine]);
9456 cache
9457 }
9458
9459 #[test]
9464 fn a_mutation_function_needs_the_role_to_be_named() {
9465 let cache = cache_with_a_mutation_function();
9466 let withheld = crate::names::NameOverrides::parse(
9467 r#"{"tables": {"public.users": {"permissions":
9468 {"reader": {"select": {"columns": "*", "filter": {}}}}}},
9469 "functions": {"public.add_to_score": {"exposed_as": "mutation"}}}"#,
9470 )
9471 .unwrap();
9472 let granted = crate::names::NameOverrides::parse(
9473 r#"{"tables": {"public.users": {"permissions":
9474 {"reader": {"select": {"columns": "*", "filter": {}}}}}},
9475 "functions": {"public.add_to_score":
9476 {"exposed_as": "mutation", "roles": ["reader"]}}}"#,
9477 )
9478 .unwrap();
9479
9480 let sdl_for = |names: &crate::names::NameOverrides| {
9481 let view = crate::role::cache_for_role(&cache, names, "reader", false);
9482 let config = SchemaConfig {
9483 names: names.clone(),
9484 role: Some("reader".to_string()),
9485 ..SchemaConfig::default()
9486 };
9487 let generated = build_schema(&view, &config);
9488 build_dynamic_schema(
9489 &generated,
9490 &view,
9491 None,
9492 None,
9493 Arc::new(names.clone()),
9494 std::time::Duration::from_secs(30),
9495 Some("reader"),
9496 )
9497 .expect("schema builds")
9498 .sdl()
9499 };
9500
9501 let without = sdl_for(&withheld);
9502 assert!(
9503 !without.contains("add_to_score"),
9504 "not granted, so not there:\n{}",
9505 without
9506 );
9507 assert!(
9510 !without.contains("mutation: mutation_root"),
9511 "and no mutation root at all:\n{}",
9512 without
9513 );
9514
9515 let with = sdl_for(&granted);
9516 assert!(
9517 with.contains("add_to_score"),
9518 "named by a function permission, so it is:\n{}",
9519 with
9520 );
9521 }
9522
9523 #[test]
9525 fn a_role_the_document_never_names_is_granted_nothing() {
9526 let cache = Arc::new(create_test_schema_cache());
9527 let config = SchemaConfig {
9528 names: crate::names::NameOverrides::parse(
9529 r#"{"tables": {"public.users": {"permissions":
9530 {"reader": {"select": {"columns": "*", "filter": {}}}}}}}"#,
9531 )
9532 .unwrap(),
9533 ..SchemaConfig::default()
9534 };
9535 let empty = build_ungranted_schema(&cache, &config).expect("it builds");
9536 let sdl = empty.sdl();
9537 assert!(
9538 sdl.contains("no_queries_available"),
9539 "nothing to read:\n{}",
9540 sdl
9541 );
9542 assert!(!sdl.contains("type users "), "and no table:\n{}", sdl);
9543 assert!(build_ungranted_schema(&cache, &SchemaConfig::default()).is_none());
9545 }
9546
9547 #[test]
9550 fn a_column_a_role_may_set_without_seeing_is_in_the_input_alone() {
9551 let cache = create_test_schema_cache();
9552 let names = crate::names::NameOverrides::parse(
9553 r#"{"tables": {"public.users": {"permissions":
9554 {"user": {"select": {"columns": ["id"], "filter": {}},
9555 "insert": {"columns": "*", "check": {}}}}}}}"#,
9556 )
9557 .unwrap();
9558 let view = crate::role::cache_for_role(&cache, &names, "user", false);
9559 let config = SchemaConfig {
9560 names: names.clone(),
9561 role: Some("user".to_string()),
9562 ..SchemaConfig::default()
9563 };
9564 let generated = build_schema(&view, &config);
9565 let object = &generated.object_types["users"];
9566 assert_eq!(
9567 object
9568 .fields
9569 .iter()
9570 .map(|f| f.name.as_str())
9571 .collect::<Vec<_>>(),
9572 vec!["id"],
9573 "the type shows what may be read"
9574 );
9575 assert!(
9576 object.writable_fields.iter().any(|f| f.name == "name"),
9577 "and the input keeps what may be written"
9578 );
9579
9580 let schema = build_dynamic_schema(
9581 &generated,
9582 &view,
9583 None,
9584 None,
9585 Arc::new(names),
9586 std::time::Duration::from_secs(30),
9587 Some("user"),
9588 )
9589 .expect("schema builds");
9590 let sdl = schema.sdl();
9591 let input = sdl
9592 .split("input users_insert_input")
9593 .nth(1)
9594 .and_then(|rest| rest.split('}').next())
9595 .expect("the insert input is registered");
9596 assert!(input.contains("name:"), "settable without being readable");
9597 }
9598
9599 #[test]
9603 fn a_generated_column_is_readable_and_not_writable() {
9604 let mut cache = create_test_schema_cache();
9605 cache
9606 .tables
9607 .get_mut(&postrust_core::api_request::QualifiedIdentifier::new(
9608 "public", "users",
9609 ))
9610 .expect("the fixture has users")
9611 .columns
9612 .get_mut("id")
9613 .expect("with an id")
9614 .always_generated = true;
9615
9616 let names = crate::names::NameOverrides::default();
9617 let config = SchemaConfig {
9618 names: names.clone(),
9619 ..SchemaConfig::default()
9620 };
9621 let generated = build_schema(&cache, &config);
9622 let schema = build_dynamic_schema(
9623 &generated,
9624 &cache,
9625 None,
9626 None,
9627 Arc::new(names),
9628 std::time::Duration::from_secs(30),
9629 None,
9630 )
9631 .expect("schema builds");
9632 let sdl = schema.sdl();
9633
9634 let block = |header: &str| {
9635 sdl.split(header)
9636 .nth(1)
9637 .and_then(|rest| rest.split('}').next())
9638 .unwrap_or_else(|| panic!("{} is registered:\n{}", header, sdl))
9639 .to_string()
9640 };
9641 assert!(
9642 block("type users ").contains("id:"),
9643 "still readable:\n{}",
9644 sdl
9645 );
9646 for header in ["input users_insert_input", "input users_set_input"] {
9647 assert!(
9648 !block(header).contains("id:"),
9649 "{} may not name it:\n{}",
9650 header,
9651 block(header)
9652 );
9653 }
9654 assert!(
9657 !sdl.contains("users_inc_input"),
9658 "and nothing is left to increment:\n{}",
9659 sdl
9660 );
9661 assert!(
9662 block("input users_insert_input").contains("name:"),
9663 "and the rest of the table is untouched",
9664 );
9665 }
9666
9667 #[test]
9668 fn test_create_object_type() {
9669 let table = create_test_table("users");
9670 let obj = TableObjectType::from_table(&table);
9671 let _gql_obj = create_object_type(&obj, &[]);
9672 }
9673
9674 #[test]
9675 fn test_create_query_type() {
9676 let cache = create_test_schema_cache();
9677 let config = SchemaConfig::default();
9678 let generated = build_schema(&cache, &config);
9679
9680 let _query = create_query_type(
9681 &generated,
9682 None,
9683 Arc::new(HashMap::new()),
9684 Arc::new(Default::default()),
9685 );
9686 }
9687
9688 #[test]
9689 fn test_create_mutation_type() {
9690 let cache = create_test_schema_cache();
9691 let config = SchemaConfig::default();
9692 let generated = build_schema(&cache, &config);
9693
9694 let _mutation = create_mutation_type(
9695 &generated,
9696 Arc::new(HashMap::new()),
9697 Arc::new(HashMap::new()),
9698 Arc::new(Default::default()),
9699 None,
9700 None,
9701 );
9702 }
9703
9704 #[test]
9709 fn a_scalar_is_named_the_way_a_client_declares_it() {
9710 use crate::types::{pg_type_to_graphql, GraphQLType};
9711
9712 assert_eq!(pg_type_to_graphql("jsonb").to_string(), "jsonb");
9717 assert_eq!(pg_type_to_graphql("json").to_string(), "json");
9718 assert_eq!(pg_type_to_graphql("int8").to_string(), "bigint");
9719 assert_eq!(pg_type_to_graphql("numeric").to_string(), "numeric");
9720 assert_eq!(pg_type_to_graphql("timestamptz").to_string(), "timestamptz");
9721 assert_eq!(pg_type_to_graphql("timestamp").to_string(), "timestamp");
9722 assert_eq!(pg_type_to_graphql("uuid").to_string(), "uuid");
9723
9724 assert_eq!(pg_type_to_graphql("geometry").to_string(), "geometry");
9727 assert_eq!(
9728 pg_type_to_graphql("colors_enum"),
9729 GraphQLType::Custom("colors_enum".to_string())
9730 );
9731
9732 assert_eq!(
9734 leaf_scalar_name(&pg_type_to_graphql("_text")),
9735 "String".to_string()
9736 );
9737 }
9738
9739 #[test]
9744 fn every_table_gets_a_boolean_expression() {
9745 let cache = create_test_schema_cache();
9746 let config = SchemaConfig::default();
9747 let generated = build_schema(&cache, &config);
9748
9749 let (inputs, _) = crate::input::bool_exp::build_inputs(
9750 &generated.object_types,
9751 &generated.relationship_fields,
9752 );
9753 let names: HashSet<String> = inputs.iter().map(|i| i.type_name().to_string()).collect();
9754
9755 for table in generated.object_types.keys() {
9756 let expected = format!("{}_bool_exp", table);
9757 assert!(
9758 names.contains(&expected),
9759 "no {} among {:?}",
9760 expected,
9761 names
9762 );
9763 }
9764 assert!(names.contains("String_comparison_exp"));
9766 assert!(names.contains("Int_comparison_exp"));
9767 }
9768
9769 #[test]
9770 fn a_schema_carrying_the_boolean_expressions_still_builds() {
9771 let cache = create_test_schema_cache();
9772 let config = SchemaConfig::default();
9773 let generated = build_schema(&cache, &config);
9774
9775 let schema = build_dynamic_schema(
9776 &generated,
9777 &cache,
9778 None,
9779 None,
9780 Arc::new(Default::default()),
9781 std::time::Duration::from_secs(30),
9782 None,
9783 );
9784 assert!(schema.is_ok(), "{:?}", schema.err());
9785
9786 let sdl = schema.unwrap().sdl();
9787 assert!(
9788 sdl.contains("_bool_exp"),
9789 "no boolean expressions in:\n{}",
9790 sdl
9791 );
9792 assert!(sdl.contains("_and"), "no _and in the generated expressions");
9793 }
9794
9795 #[test]
9800 fn test_build_schema_with_subscriptions() {
9801 let cache = create_test_schema_cache();
9802 let config = SchemaConfig {
9803 enable_subscriptions: true,
9804 ..SchemaConfig::default()
9805 };
9806 let generated = build_schema(&cache, &config);
9807
9808 let sub_fields = generate_subscription_fields(&cache, &generated);
9810 assert!(!sub_fields.is_empty(), "Should have subscription fields");
9811
9812 let result = build_dynamic_schema(
9814 &generated,
9815 &cache,
9816 Some(&sub_fields),
9817 None,
9818 Arc::new(Default::default()),
9819 std::time::Duration::from_secs(30),
9820 None,
9821 );
9822 assert!(result.is_ok(), "Schema with subscriptions should build");
9823 }
9824
9825 #[test]
9826 fn test_subscription_field_generation() {
9827 let cache = create_test_schema_cache();
9828 let config = SchemaConfig::default();
9829 let generated = build_schema(&cache, &config);
9830
9831 let fields = generate_subscription_fields(&cache, &generated);
9832
9833 assert_eq!(fields.len(), 1);
9835 assert_eq!(fields[0].name, "users");
9836 assert_eq!(fields[0].table_name, "users");
9837 assert_eq!(fields[0].channel_name(), "postrust_public_users");
9838 }
9839
9840 #[test]
9841 fn test_create_subscription_type() {
9842 let cache = create_test_schema_cache();
9843 let config = SchemaConfig::default();
9844 let generated = build_schema(&cache, &config);
9845 let _subscription = create_subscription_type(
9848 &generated,
9849 None,
9850 Arc::new(Default::default()),
9851 Arc::new(Default::default()),
9852 std::time::Duration::from_secs(30),
9853 );
9854 }
9855}
9856
9857#[derive(Debug, PartialEq, Eq)]
9859enum PathStep {
9860 Key(String),
9861 Index(usize),
9862}
9863
9864fn walk_json_path(value: &Value, path: &str) -> Result<Value, async_graphql::Error> {
9871 let mut at = value;
9872 for step in parse_json_path(path)? {
9873 let next = match (&step, at) {
9874 (PathStep::Key(key), Value::Object(map)) => map.get(key.as_str()),
9875 (PathStep::Index(index), Value::List(items)) => items.get(*index),
9876 _ => None,
9879 };
9880 match next {
9881 Some(next) => at = next,
9882 None => return Ok(Value::Null),
9883 }
9884 }
9885 Ok(at.clone())
9886}
9887
9888fn unclosed_bracket(path: &str) -> async_graphql::Error {
9890 let message = format!("\"{}\" is not a JSON path: a bracket is never closed", path);
9891 validation_error(&message)
9892}
9893
9894fn parse_json_path(path: &str) -> Result<Vec<PathStep>, async_graphql::Error> {
9901 let mut steps = Vec::new();
9902 let mut chars = path.chars().peekable();
9903 if chars.peek() == Some(&'$') {
9905 chars.next();
9906 }
9907 while let Some(&c) = chars.peek() {
9908 match c {
9909 '.' => {
9911 chars.next();
9912 }
9913 '[' => {
9914 chars.next();
9915 let quote = match chars.peek() {
9916 Some(&q @ ('\'' | '"')) => {
9917 chars.next();
9918 Some(q)
9919 }
9920 _ => None,
9921 };
9922 let mut inner = String::new();
9923 loop {
9924 let Some(c) = chars.next() else {
9925 return Err(unclosed_bracket(path));
9926 };
9927 match (quote, c) {
9928 (Some(quote), c) if c == quote => break,
9929 (None, ']') => break,
9930 _ => inner.push(c),
9931 }
9932 }
9933 if quote.is_some() {
9935 match chars.next() {
9936 Some(']') => {}
9937 _ => return Err(unclosed_bracket(path)),
9938 }
9939 }
9940 match (quote, inner.parse::<usize>()) {
9943 (None, Ok(index)) => steps.push(PathStep::Index(index)),
9944 _ => steps.push(PathStep::Key(inner)),
9945 }
9946 }
9947 _ => {
9948 let mut key = String::new();
9949 while let Some(&c) = chars.peek() {
9950 if matches!(c, '.' | '[') {
9951 break;
9952 }
9953 key.push(c);
9954 chars.next();
9955 }
9956 steps.push(PathStep::Key(key));
9957 }
9958 }
9959 }
9960 Ok(steps)
9961}
9962
9963#[cfg(test)]
9964mod json_path_tests {
9965 use super::*;
9966
9967 fn steps(path: &str) -> Vec<PathStep> {
9968 parse_json_path(path).expect("the path parses")
9969 }
9970
9971 fn key(name: &str) -> PathStep {
9972 PathStep::Key(name.to_string())
9973 }
9974
9975 #[test]
9976 fn the_document_itself_has_no_steps() {
9977 assert_eq!(steps("$"), vec![]);
9978 assert_eq!(steps(""), vec![]);
9979 }
9980
9981 #[test]
9982 fn a_key_is_written_bare_or_after_a_dot() {
9983 assert_eq!(steps("a"), vec![key("a")]);
9984 assert_eq!(steps(".obj.c1"), vec![key("obj"), key("c1")]);
9985 assert_eq!(steps("._underscore"), vec![key("_underscore")]);
9986 }
9987
9988 #[test]
9989 fn a_bracket_holds_an_index_or_a_quoted_key() {
9990 assert_eq!(steps("arr[0]"), vec![key("arr"), PathStep::Index(0)]);
9991 assert_eq!(steps("['!@#$%^']"), vec![key("!@#$%^")]);
9992 assert_eq!(
9993 steps("translations['hello world!']"),
9994 vec![key("translations"), key("hello world!")]
9995 );
9996 }
9997
9998 #[test]
10000 fn a_dot_before_a_bracket_changes_nothing() {
10001 let expected = vec![key("objs"), PathStep::Index(0), key("\u{4f60}\u{597d}")];
10002 for spelling in [
10003 "objs[0]['\u{4f60}\u{597d}']",
10004 "objs[0][\"\u{4f60}\u{597d}\"]",
10005 "objs[0].[\"\u{4f60}\u{597d}\"]",
10006 "objs.[0].[\"\u{4f60}\u{597d}\"]",
10007 ] {
10008 assert_eq!(steps(spelling), expected, "{}", spelling);
10009 }
10010 }
10011
10012 #[test]
10013 fn a_bracket_that_never_closes_is_refused() {
10014 assert!(parse_json_path("arr[0").is_err());
10015 assert!(parse_json_path("['key'").is_err());
10016 }
10017
10018 #[test]
10019 fn a_step_the_document_does_not_have_reads_null() {
10020 let document = Value::from_json(serde_json::json!({
10021 "obj": {"c1": "c2"},
10022 "arr": [1, 2, 3],
10023 }))
10024 .expect("a document");
10025 let at = |path| walk_json_path(&document, path).expect("the path parses");
10026 assert_eq!(at(".obj.c1"), Value::String("c2".to_string()));
10027 assert_eq!(at("arr[1]"), Value::Number(2.into()));
10028 assert_eq!(at("arr[9]"), Value::Null);
10029 assert_eq!(at(".obj.nothing"), Value::Null);
10030 assert_eq!(at("obj[0]"), Value::Null);
10033 assert_eq!(at("arr.first"), Value::Null);
10034 }
10035}