diff --git a/CHANGELOG.md b/CHANGELOG.md index dd2728f2..3e24d542 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,14 @@ for historical context. ## Unreleased +Companion runtime/macros version: **0.63.0**. + +- Seed propagated-event cycle detection with the complete typed target identity and origin so cycles cannot re-emit the direct source. +- Resolve generated upward propagation using all declared relation from/to components, independent of either entity's primary key. +- Revalidate source/target tuple witnesses and current relation-field authority before subscriber delivery; reject fanout that exceeds the remaining budget. + +## 0.62.0 (unreleased) + Companion runtime/macros version: **0.62.0**. - Add typed runtime `orderBy` to generated keyset queries and repository forward/backward reads. Whitelist persisted scalar order fields, append the complete primary-key suffix, bind dynamic cursors to the exact order, authorize all raw cursor fields, and retain existing configured cursor bytes when `orderBy` is absent. diff --git a/Cargo.lock b/Cargo.lock index 1fb1ba62..9b997ace 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3113,7 +3113,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" dependencies = [ "agql-auth", "async-graphql", @@ -3217,7 +3217,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" dependencies = [ "convert_case 0.7.1", "proc-macro2", diff --git a/Cargo.toml b/Cargo.toml index 6a401469..10f2daee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,7 +40,7 @@ cynic-parser = { version = "=0.11.2", features = ["pretty"] } futures = "0.3" getrandom = "0.3" graphql-composition = "=0.12.2" -graphql-orm = { path = "crates/graphql-orm", version = "0.62.0", default-features = false } +graphql-orm = { path = "crates/graphql-orm", version = "0.63.0", default-features = false } graphql-orm-ai-tool-profiles = { path = "crates/graphql-orm-ai-tool-profiles", version = "0.16.2" } graphql-orm-backup = { path = "crates/graphql-orm-backup", version = "0.10.0", default-features = false } graphql-orm-operation-catalog = { path = "crates/graphql-orm-operation-catalog", version = "0.4.2" } diff --git a/MIGRATION.md b/MIGRATION.md index bbb783e1..0e4a0576 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -23,6 +23,27 @@ Each dynamic order appends missing primary-key columns in ascending order. NULL Dynamic cursors bind the entity, exact ordered columns, directions and null placement. A cursor for a different order or the configured default is rejected. Omit `orderBy` to retain the configured order and existing cursor bytes. Repository callers may use `keyset_page_ordered` or `keyset_connection_page_ordered`; existing methods stay unchanged. Every composite primary-key component participates in configured and runtime validation. No schema or data migration is required. +## 0.62.0 to 0.63.0: propagation relation tuples + +Upgrade runtime/macros together. `propagate_change = "up"` now resolves the +complete declared `from`/`to` arrays, including scalar-primary-key entities +related by a multi-column natural key. It no longer substitutes the child primary +key or assumes the target relation column is its primary key. Scalar relations +use the same mapped path. Every target must fit the remaining bounded fanout; +a larger fanout fails the mutation so it cannot silently notify a partial set. + +A generated witness retains only typed source keys, target identity and the +complete relation mapping. At each subscriber release, current source authority, +source key and relation-field authority, every tuple component and the target +are revalidated. A relation changed since commit suppresses stale delivery. +Post-commit, rollback, provenance-hop and event budgets remain enforced. The +legacy hidden scalar witness constructor remains compatible with existing hosts. + +This is a focused part of issue #191. Composite-primary-key Changed generation +and read-only trusted projection subscriptions remain separate unfinished work; +this change does not complete #191, #184 or draft #189. No data/schema migration +or downstream pin change is included. + ## 0.60.0 to 0.61.0: subscription primary-key selection Upgrade runtime/macros together. Generated single-key subscriptions gain an optional `key` argument, using the entity primary key scalar and configured argument casing. Existing operations remain valid. For example, `recordChanged(key: "…", filter: {actions: [UPDATED]})` watches one row. Key selection narrows notifications only: it never grants row, field, scope, assurance, deleted-preimage or propagated-source authority. Refresh schema/catalogue snapshots. Composite-key subscriptions remain unsupported by this change. No data or schema migration is required. diff --git a/crates/graphql-orm-macros/Cargo.toml b/crates/graphql-orm-macros/Cargo.toml index 6805f29b..424b73b7 100644 --- a/crates/graphql-orm-macros/Cargo.toml +++ b/crates/graphql-orm-macros/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" edition = "2024" authors = ["Toby Martin"] description = "Procedural macros for async-graphql and ORM-backed entities, relations, and CRUD operations." diff --git a/crates/graphql-orm-macros/README.md b/crates/graphql-orm-macros/README.md index 7e68b1ff..8a5c15e0 100644 --- a/crates/graphql-orm-macros/README.md +++ b/crates/graphql-orm-macros/README.md @@ -15,6 +15,10 @@ The scalar-only input, complete primary-key suffix, order-bound cursor and curre field authorization are described in the [migration guide](../../MIGRATION.md#0610-to-0620-runtime-keyset-ordering). +Version 0.63.0 resolves generated upward propagation using complete relation +from/to tuples and revalidates the current mapping at release. See the +[migration guide](../../MIGRATION.md#0620-to-0630-propagation-relation-tuples). + Version 0.61.0 adds an optional typed `key` argument to generated single-primary-key change subscriptions, preserving current authorization and action filtering. See the [migration guide](../../MIGRATION.md#0600-to-0610-subscription-primary-key-selection). diff --git a/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock b/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock index 6788d2c9..9785350a 100644 --- a/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock +++ b/crates/graphql-orm-macros/fixtures/cross-crate-relations/Cargo.lock @@ -1049,7 +1049,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -1076,7 +1076,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm-macros/fixtures/relationship-capability/Cargo.lock b/crates/graphql-orm-macros/fixtures/relationship-capability/Cargo.lock index 63591074..9f8bc3f9 100644 --- a/crates/graphql-orm-macros/fixtures/relationship-capability/Cargo.lock +++ b/crates/graphql-orm-macros/fixtures/relationship-capability/Cargo.lock @@ -1110,7 +1110,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -1152,7 +1152,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm-macros/src/entity.rs b/crates/graphql-orm-macros/src/entity.rs index 5a93a1d8..fee9dd8c 100644 --- a/crates/graphql-orm-macros/src/entity.rs +++ b/crates/graphql-orm-macros/src/entity.rs @@ -4765,6 +4765,7 @@ fn generate_entity_impl( PartialEq, )] #aggregate_graphql_attribute + #[allow(clippy::enum_variant_names)] pub enum #aggregate_field_name { #(#aggregate_field_variants),* } diff --git a/crates/graphql-orm-macros/src/operations.rs b/crates/graphql-orm-macros/src/operations.rs index 0a919ced..bc5f5896 100644 --- a/crates/graphql-orm-macros/src/operations.rs +++ b/crates/graphql-orm-macros/src/operations.rs @@ -722,8 +722,8 @@ fn classify_propagation_value_type(ty: &syn::Type) -> Option<(PropagationValueKi struct PropagatedRelation { graphql_name: String, - source_field: syn::Ident, - source_ty: syn::Type, + source_fields: Vec<(syn::Ident, syn::Type, FieldMetadata)>, + target_fields: Vec, target_type: syn::Ident, } @@ -1780,41 +1780,51 @@ fn generate_graphql_operations_unfiltered( let rust_name = field_name.to_string(); let graphql_name = graphql_field_name(&meta, &rust_name, graphql_rename_fields, serde_rename_all); - let source_column = meta - .relation_from - .clone() - .unwrap_or_else(|| pk_field.to_string()); - let source_field = fields - .iter() - .find(|candidate| { - candidate - .ident - .as_ref() - .is_some_and(|ident| ident == source_column.as_str()) - }) - .and_then(|candidate| { - candidate - .ident + let source_names = meta.relation_from_fields.clone().unwrap_or_else(|| { + vec![ + meta.relation_from .clone() - .map(|ident| (ident, candidate.ty.clone())) - }) - .ok_or_else(|| { - syn::Error::new_spanned( - field, - format!( - "Relation '{}' references unknown propagation source field '{}'", - rust_name, source_column - ), - ) - })?; + .unwrap_or_else(|| pk_field.to_string()), + ] + }); + let target_fields = meta + .relation_to_fields + .clone() + .unwrap_or_else(|| vec![meta.relation_to.clone().unwrap_or_else(|| "id".to_string())]); + if source_names.is_empty() || source_names.len() != target_fields.len() { + return Err(syn::Error::new_spanned( + field, + "propagate_change requires equal nonempty from/to tuples", + )); + } + let mut source_fields = Vec::new(); + for source in source_names { + let candidate = fields.iter().find(|candidate| candidate.ident.as_ref().is_some_and(|ident| ident == source.as_str())) + .ok_or_else(|| syn::Error::new_spanned(field, format!("Relation '{rust_name}' references unknown propagation source field '{source}'")))?; + let metadata = parse_field_metadata(candidate)?; + if metadata.is_relation + || metadata.skip_db + || classify_propagation_value_type(&candidate.ty).is_none() + { + return Err(syn::Error::new_spanned( + candidate, + "propagate_change source fields must be persisted scalar fields", + )); + } + source_fields.push(( + candidate.ident.clone().expect("named field"), + candidate.ty.clone(), + metadata, + )); + } let target_type = syn::Ident::new( meta.relation_target.as_deref().unwrap_or("Unknown"), field.span(), ); propagated_relations.push(PropagatedRelation { graphql_name, - source_field: source_field.0, - source_ty: source_field.1, + source_fields, + target_fields, target_type, }); } @@ -2162,6 +2172,11 @@ fn generate_graphql_operations_unfiltered( } }; let propagated_subscription_emitter = quote! { + #[doc(hidden)] + pub(crate) fn __gom_propagation_key(entity: &Self) -> #pk_type_ty { + entity.#pk_field.clone() + } + #[doc(hidden)] #[allow(clippy::too_many_arguments, unused_variables)] pub(crate) fn __gom_queue_propagated_changed_event( @@ -2187,87 +2202,66 @@ fn generate_graphql_operations_unfiltered( #propagated_witness_emission } }; - let propagation_blocks: Vec = propagated_relations - .iter() + // Direct and propagated events must share exactly the same typed target + // identity and origin encoding, so a cycle cannot re-emit its source. + let propagation_visited_seed = quote! { + let origin_target_id = ::graphql_orm::graphql::orm::composite_key_id( + &Self::__gom_subscription_key_values(entity), + ); + let origin_dedupe_key = ::graphql_orm::serde_json::to_string(&( + #entity_name_lit, origin_target_id, #entity_name_lit, &source_id, + )).map_err(|error| ::graphql_orm::sqlx::Error::Protocol(error.to_string()))?; + let mut visited = ::std::collections::HashSet::from([origin_dedupe_key]); + }; + let propagation_blocks: Vec = propagated_relations.iter() .map(|relation| -> syn::Result { let relation_name = &relation.graphql_name; - let source_field = &relation.source_field; - let source_field_name = source_field.to_string(); let target_type = &relation.target_type; - let (kind, is_option) = classify_propagation_value_type(&relation.source_ty).ok_or_else( - || { - syn::Error::new_spanned( - &relation.source_ty, - format!( - "Unsupported propagate_change source field type for relation '{}': expected String/uuid/int/float/bool (optionals allowed)", - relation_name - ), - ) - }, - )?; - - let source_binding = if is_option { - quote! { - let Some(parent_id) = entity.#source_field.as_ref() else { - return Ok(()); - }; - } - } else { - quote! { - let parent_id = &entity.#source_field; - } - }; - - let parent_key_string = match kind { - PropagationValueKind::String => quote! { parent_id.clone() }, - _ => quote! { parent_id.to_string() }, - }; - - let fetch_parent = quote! { - let parent_entity = <#target_type>::__gom_fetch_by_id_on(hook_ctx, parent_id).await?; - }; - + let target_fields = &relation.target_fields; + let mut value_tokens = Vec::new(); + let mut optional_checks = Vec::new(); + let mut source_names = Vec::new(); + for (field, ty, metadata) in &relation.source_fields { + source_names.push(field.to_string()); + if is_option_type(ty) { optional_checks.push(quote! { entity.#field.is_some() }); } + let mut metadata = metadata.clone(); + metadata.transform_write = None; + value_tokens.push(push_create_input_sql_value_tokens(backend, struct_name, + quote! { entity.#field.clone() }, ty, &metadata, quote! { ::graphql_orm::sqlx::Error })); + } Ok(quote! { - { + if true #(&& #optional_checks)* { if visited.len() >= ::graphql_orm::graphql::subscriptions::MAX_SUBSCRIPTION_PROPAGATION_EVENTS { return Ok(()); } - #source_binding - let parent_id_string = #parent_key_string; - let mut next_path = path.clone(); - next_path.push(#relation_name.to_string()); - let mut next_provenance = provenance.to_vec(); - next_provenance.push(::graphql_orm::graphql::subscriptions::SubscriptionEventSource::<#backend_marker>::new::( - Self::__gom_subscription_key_values(entity), #relation_name, #source_field_name, - )); - let dedupe_key = format!( - "{}:{}:{}:{}", - <#target_type as ::graphql_orm::graphql::orm::Entity>::entity_name(), - parent_id_string, - source_entity, - source_id - ); - if visited.insert(dedupe_key) { - #fetch_parent - <#target_type>::__gom_queue_propagated_changed_event( - hook_ctx, action, parent_id.clone(), parent_entity.as_ref(), source_entity, - source_id, next_path.clone(), next_provenance.clone(), - ); - if let Some(parent_entity) = parent_entity.as_ref() { - <#target_type>::__gom_propagate_changed_event( - hook_ctx, - action, - parent_entity, - source_entity, - source_id, - next_path, - visited, - &next_provenance, - ).await?; + let mut bind_values = Vec::new(); + #(#value_tokens)* + let parent_entities = ::graphql_orm::graphql::subscriptions::propagation_targets::<#target_type, #backend_marker>( + hook_ctx, &[#(#target_fields),*], &bind_values, + ::graphql_orm::graphql::subscriptions::MAX_SUBSCRIPTION_PROPAGATION_EVENTS - visited.len(), + ).await?; + for parent_entity in parent_entities { + let parent_id = <#target_type>::__gom_propagation_key(&parent_entity); + let parent_keys = <#target_type>::__gom_subscription_key_values(&parent_entity); + let parent_id_string = ::graphql_orm::graphql::orm::composite_key_id(&parent_keys); + let mut next_path = path.clone(); + next_path.push(#relation_name.to_string()); + let mut next_provenance = provenance.to_vec(); + next_provenance.push(::graphql_orm::graphql::subscriptions::SubscriptionEventSource::<#backend_marker>::new_relation::( + Self::__gom_subscription_key_values(entity), #relation_name, vec![#((#source_names, #target_fields)),*], + )); + let dedupe_key = ::graphql_orm::serde_json::to_string(&( + <#target_type as ::graphql_orm::graphql::orm::Entity>::entity_name(), + parent_id_string, source_entity, source_id, + )).map_err(|error| ::graphql_orm::sqlx::Error::Protocol(error.to_string()))?; + if visited.insert(dedupe_key) { + <#target_type>::__gom_queue_propagated_changed_event(hook_ctx, action, parent_id, + Some(&parent_entity), source_entity, source_id, next_path.clone(), next_provenance.clone()); + <#target_type>::__gom_propagate_changed_event(hook_ctx, action, &parent_entity, + source_entity, source_id, next_path, visited, &next_provenance).await?; } } } }) - }) - .collect::>>()?; + }).collect::>>()?; // Generate input fields (excluding primary key for create, all optional for update) let mut create_input_fields: Vec = Vec::new(); @@ -5390,9 +5384,7 @@ fn generate_graphql_operations_unfiltered( }; hook_ctx.queue_event(event.clone()); #queue_subscription_witness - let mut visited = ::std::collections::HashSet::from([format!( - "{}:{}:{}:{}", #entity_name_lit, source_id, #entity_name_lit, source_id - )]); + #propagation_visited_seed Self::__gom_propagate_changed_event( hook_ctx, action, entity, #entity_name_lit, &source_id, Vec::new(), &mut visited, &[], ).await @@ -8212,13 +8204,7 @@ fn generate_graphql_operations_unfiltered( }; hook_ctx.queue_event(event.clone()); #queue_subscription_witness - let mut visited = ::std::collections::HashSet::from([format!( - "{}:{}:{}:{}", - #entity_name_lit, - source_id, - #entity_name_lit, - source_id - )]); + #propagation_visited_seed Self::__gom_propagate_changed_event( hook_ctx, action, diff --git a/crates/graphql-orm/Cargo.toml b/crates/graphql-orm/Cargo.toml index 51ed68a3..cc9ec317 100644 --- a/crates/graphql-orm/Cargo.toml +++ b/crates/graphql-orm/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" edition = "2024" description = "Runtime support crate for graphql-orm-macros" license = "MIT" @@ -73,7 +73,7 @@ futures = "0.3" geo = { version = "0.33", optional = true, default-features = false } geo-types = { version = "0.7", optional = true } geojson = { version = "1", optional = true, default-features = true } -graphql-orm-macros = { path = "../graphql-orm-macros", version = "0.62.0", default-features = false } +graphql-orm-macros = { path = "../graphql-orm-macros", version = "0.63.0", default-features = false } graphql-orm-operation-catalog = { workspace = true } rust_decimal = { workspace = true } serde = { version = "1", features = ["derive"] } diff --git a/crates/graphql-orm/README.md b/crates/graphql-orm/README.md index aa142f0f..d683a8d0 100644 --- a/crates/graphql-orm/README.md +++ b/crates/graphql-orm/README.md @@ -15,6 +15,10 @@ The scalar-only input, complete primary-key suffix, order-bound cursor and curre field authorization are described in the [migration guide](../../MIGRATION.md#0610-to-0620-runtime-keyset-ordering). +Version 0.63.0 resolves generated upward propagation using complete relation +from/to tuples and revalidates the current mapping at release. See the +[migration guide](../../MIGRATION.md#0620-to-0630-propagation-relation-tuples). + Version 0.61.0 adds an optional typed `key` argument to generated single-primary-key change subscriptions, preserving current authorization and action filtering. See the [migration guide](../../MIGRATION.md#0600-to-0610-subscription-primary-key-selection). diff --git a/crates/graphql-orm/src/graphql/subscriptions.rs b/crates/graphql-orm/src/graphql/subscriptions.rs index 882fc249..afbed8a1 100644 --- a/crates/graphql-orm/src/graphql/subscriptions.rs +++ b/crates/graphql-orm/src/graphql/subscriptions.rs @@ -24,6 +24,15 @@ type SourceCheck = for<'a, 'ctx> fn( &'static str, ) -> BoxFuture<'a, async_graphql::Result<()>>; +type RelationSourceCheck = for<'a, 'ctx> fn( + &'a Database, + &'a Context<'ctx>, + &'a [SqlValue], + &'a [SqlValue], + &'static str, + &'a [(&'static str, &'static str)], +) -> BoxFuture<'a, async_graphql::Result<()>>; + /// Server-only source identity. It retains typed keys, never source snapshots. #[doc(hidden)] #[derive(Clone)] @@ -33,6 +42,8 @@ pub struct SubscriptionEventSource { keys: Vec, relation: &'static str, source_field: &'static str, + relation_fields: Vec<(&'static str, &'static str)>, + relation_check: Option>, check: SourceCheck, } impl std::fmt::Debug for SubscriptionEventSource { @@ -57,9 +68,28 @@ impl SubscriptionEventSource { keys, relation, source_field, + relation_fields: Vec::new(), + relation_check: None, check: check_source::, } } + + /// Bind an upward relation to its complete source/target column mapping. + /// Both types and field names come from generated server code, never clients. + pub fn new_relation( + keys: Vec, + relation: &'static str, + fields: Vec<(&'static str, &'static str)>, + ) -> Self + where + S: Entity + FromSqlRow + Clone + Send + Sync + 'static, + T: Entity + FromSqlRow + Clone + Send + Sync + 'static, + { + let mut source = Self::new::(keys, relation, ""); + source.relation_fields = fields; + source.relation_check = Some(check_relation_source::); + source + } } /// Trusted host emission envelope used by generated code. @@ -133,15 +163,27 @@ impl SubscriptionEvent { .sources .get(index + 1) .map_or(self.keys.as_slice(), |next| next.keys.as_slice()); - (source.check)( - db, - ctx, - &source.keys, - target_keys, - source.relation, - source.source_field, - ) - .await?; + if let Some(check) = source.relation_check { + check( + db, + ctx, + &source.keys, + target_keys, + source.relation, + &source.relation_fields, + ) + .await?; + } else { + (source.check)( + db, + ctx, + &source.keys, + target_keys, + source.relation, + source.source_field, + ) + .await?; + } } Ok(()) } @@ -170,7 +212,7 @@ async fn authorize_subscription_row_matching( ctx: &Context<'_>, keys: &[SqlValue], deleted_preimage: Option<&T>, - link: Option<(&str, &SqlValue)>, + link: Option<(String, Vec)>, ) -> async_graphql::Result> where B: OrmBackend, @@ -195,11 +237,8 @@ where // PRIMARY_KEYS are trusted backend-rendered column identities. query = query.where_clause(&format!("{column} = ?"), value.clone()); } - if let Some((column, value)) = link { - query = query.where_clause( - &format!("{} = ?", B::DIALECT.quote_identifier_path(column)), - value.clone(), - ); + if let Some((clause, values)) = link { + query = query.where_values(&clause, values); } query = query .with_read_visibility(&visibility)? @@ -325,7 +364,10 @@ where ctx, keys, None, - Some((column.name, &target_keys[0])), + Some(( + format!("{} = ?", B::DIALECT.quote_identifier_path(column.name)), + vec![target_keys[0].clone()], + )), ) .await? .ok_or_else(|| OrmPublicError::forbidden().into_graphql_error())?; @@ -335,6 +377,134 @@ where }) } +fn check_relation_source<'a, 'ctx, S, T, B>( + db: &'a Database, + ctx: &'a Context<'ctx>, + keys: &'a [SqlValue], + target_keys: &'a [SqlValue], + relation: &'static str, + fields: &'a [(&'static str, &'static str)], +) -> BoxFuture<'a, async_graphql::Result<()>> +where + B: OrmBackend, + S: Entity + FromSqlRow + Clone + Send + Sync + 'static, + T: Entity + FromSqlRow + Clone + Send + Sync + 'static, +{ + Box::pin(async move { + let denied = || OrmPublicError::forbidden().into_graphql_error(); + if fields.is_empty() + || fields.len() > S::columns().len() + || target_keys.len() != T::PRIMARY_KEYS.len() + || target_keys.is_empty() + { + return Err(denied()); + } + let mut predicates = T::PRIMARY_KEYS + .iter() + .map(|column| format!("__gom_propagation_target.{column} = ?")) + .collect::>(); + let mut source_fields = Vec::new(); + let mut target_fields = Vec::new(); + for (source, target) in fields { + let source = S::columns() + .iter() + .find(|column| column.rust_name == *source) + .ok_or_else(denied)?; + let target = T::columns() + .iter() + .find(|column| column.rust_name == *target) + .ok_or_else(denied)?; + predicates.push(format!( + "__gom_propagation_target.{} = {}.{}", + B::DIALECT.quote_identifier_path(target.name), + B::DIALECT.quote_identifier_path(S::TABLE_NAME), + B::DIALECT.quote_identifier_path(source.name) + )); + source_fields.push(source.api_name); + target_fields.push(target.api_name); + } + let clause = format!( + "EXISTS (SELECT 1 FROM {} AS __gom_propagation_target WHERE {})", + B::DIALECT.quote_identifier_path(T::TABLE_NAME), + predicates.join(" AND ") + ); + let row = authorize_subscription_row_matching::( + db, + ctx, + keys, + None, + Some((clause, target_keys.to_vec())), + ) + .await? + .ok_or_else(denied)?; + authorize_subscription_keys::(db, ctx, &row).await?; + for field in source_fields { + check_field::(db, ctx, &row, field).await?; + } + check_field::(db, ctx, &row, relation).await?; + let target = authorize_subscription_row::(db, ctx, target_keys, None) + .await? + .ok_or_else(denied)?; + authorize_subscription_keys::(db, ctx, &target).await?; + for field in target_fields { + check_field::(db, ctx, &target, field).await?; + } + Ok(()) + }) +} + +/// Read bounded propagation targets on the same pinned mutation transaction. +/// This grants no release authority: every hop is revalidated for each subscriber. +#[cfg(any(feature = "sqlite", feature = "postgres", feature = "mssql"))] +#[doc(hidden)] +pub async fn propagation_targets( + context: &mut crate::graphql::orm::MutationContext<'_, B>, + fields: &[&str], + values: &[SqlValue], + remaining: usize, +) -> crate::Result> +where + B: crate::graphql::orm::WriteBackend, + T: Entity + FromSqlRow + Clone + Send + Sync + 'static, +{ + if fields.is_empty() + || fields.len() != values.len() + || remaining == 0 + || remaining > MAX_SUBSCRIPTION_PROPAGATION_EVENTS + { + return Err(sqlx::Error::Protocol( + "invalid generated propagation tuple/budget".into(), + )); + } + let mut query = EntityQuery::::new(); + for (field, value) in fields.iter().zip(values) { + let column = T::columns() + .iter() + .find(|column| column.rust_name == *field) + .ok_or_else(|| { + sqlx::Error::Protocol("unknown generated propagation target column".into()) + })?; + query = query.where_clause( + &format!("{} = ?", B::DIALECT.quote_identifier_path(column.name)), + value.clone(), + ); + } + // One lookahead detects a fanout that cannot be represented completely. + let rows = query + .paginate(&crate::graphql::orm::PageInput { + limit: Some(remaining.min(MAX_SUBSCRIPTION_PROPAGATION_EVENTS) as i64 + 1), + offset: Some(0), + }) + .fetch_all_in_transaction(context) + .await?; + if rows.len() > remaining { + return Err(sqlx::Error::Protocol( + "generated propagation target budget exceeded".into(), + )); + } + Ok(rows) +} + fn unsupported_subscription_selection() -> async_graphql::Error { OrmPublicError::with_message(crate::graphql::errors::OrmErrorCode::InvalidInput, "Subscription payloads support persisted subscribable scalar fields only; refresh authorized queries for relations or computed fields").into_graphql_error() diff --git a/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock b/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock index 7a589e08..bcb87353 100644 --- a/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/backend-coexistence/Cargo.lock @@ -1368,7 +1368,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" dependencies = [ "agql-auth", "async-graphql", @@ -1442,7 +1442,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock b/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock index abf1533f..34a7db40 100644 --- a/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/federation-end-to-end/Cargo.lock @@ -2508,7 +2508,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -2532,7 +2532,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" dependencies = [ "convert_case 0.7.1", "proc-macro2", diff --git a/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock b/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock index 45504991..4c348bdd 100644 --- a/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/migration-api-compatibility/Cargo.lock @@ -1025,7 +1025,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" dependencies = [ "async-graphql", "chacha20poly1305", @@ -1052,7 +1052,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock b/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock index e9a5103d..26826cae 100644 --- a/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock +++ b/crates/graphql-orm/tests/fixtures/repository-aggregate-consumer/Cargo.lock @@ -1025,7 +1025,7 @@ dependencies = [ [[package]] name = "graphql-orm" -version = "0.62.0" +version = "0.63.0" dependencies = [ "async-graphql", "async-trait", @@ -1054,7 +1054,7 @@ dependencies = [ [[package]] name = "graphql-orm-macros" -version = "0.62.0" +version = "0.63.0" dependencies = [ "convert_case", "proc-macro2", diff --git a/crates/graphql-orm/tests/subscription_relation_tuples.rs b/crates/graphql-orm/tests/subscription_relation_tuples.rs new file mode 100644 index 00000000..d8956a15 --- /dev/null +++ b/crates/graphql-orm/tests/subscription_relation_tuples.rs @@ -0,0 +1,393 @@ +#![cfg(feature = "sqlite")] +use graphql_orm::async_graphql::{Request, Response}; +use graphql_orm::futures::{Stream, StreamExt, future::poll_fn}; +use graphql_orm::prelude::*; +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; +use std::task::Poll; +use tokio::time::{Duration, timeout}; + +#[derive(GraphQLEntity, GraphQLOperations, Clone, Debug, serde::Serialize, serde::Deserialize)] +#[graphql_entity(table = "tuple_parents", plural = "TupleParents")] +struct TupleParent { + #[primary_key] + #[sortable] + #[graphql_orm(auto_generated = false)] + id: String, + organization_id: String, + provider_account_id: String, + subscription_id: String, +} +#[derive(GraphQLEntity, GraphQLOperations, Clone, Debug, serde::Serialize, serde::Deserialize)] +#[graphql_entity(table = "tuple_children", plural = "TupleChildren")] +struct TupleChild { + #[primary_key] + #[sortable] + #[graphql_orm(auto_generated = false)] + cache_id: String, + organization_id: String, + provider_account_id: String, + subscription_id: String, + #[graphql(skip)] + #[relation(target = "TupleParent", from = ["organization_id", "provider_account_id", "subscription_id"], to = ["organization_id", "provider_account_id", "subscription_id"], propagate_change = "up", emit_fk = false)] + parent: Option, +} +schema_roots! { query_custom_ops: [], entities: [TupleParent, TupleChild], } +#[derive(Clone)] +struct Policy { + source: Arc, + target: Arc, + fields: Arc, + target_fields: Arc, +} +impl RowPolicy for Policy { + fn can_read_row<'a>( + &'a self, + _: Option<&'a async_graphql::Context<'_>>, + _: &'a Database, + entity: &'static str, + _: Option<&'static str>, + _: EntityAccessSurface, + _: &'a (dyn std::any::Any + Send + Sync), + ) -> graphql_orm::futures::future::BoxFuture<'a, async_graphql::Result> { + Box::pin(async move { + Ok(if entity == "TupleChild" { + self.source.load(Ordering::SeqCst) + } else { + self.target.load(Ordering::SeqCst) + }) + }) + } + fn can_write_row<'a>( + &'a self, + _: Option<&'a async_graphql::Context<'_>>, + _: &'a Database, + _: &'static str, + _: Option<&'static str>, + _: EntityAccessSurface, + _: &'a (dyn std::any::Any + Send + Sync), + ) -> graphql_orm::futures::future::BoxFuture<'a, async_graphql::Result> { + Box::pin(async { Ok(true) }) + } +} +impl FieldPolicy for Policy { + fn can_read_field<'a>( + &'a self, + _: &'a async_graphql::Context<'_>, + _: &'a Database, + entity: &'static str, + field: &'static str, + _: Option<&'static str>, + _: Option<&'a (dyn std::any::Any + Send + Sync)>, + ) -> graphql_orm::futures::future::BoxFuture<'a, async_graphql::Result> { + Box::pin(async move { + Ok((entity != "TupleChild" + || field != "organizationId" + || self.fields.load(Ordering::SeqCst)) + && (entity != "TupleParent" + || field != "providerAccountId" + || self.target_fields.load(Ordering::SeqCst))) + }) + } + fn can_write_field<'a>( + &'a self, + _: &'a async_graphql::Context<'_>, + _: &'a Database, + _: &'static str, + _: &'static str, + _: Option<&'static str>, + _: Option<&'a (dyn std::any::Any + Send + Sync)>, + _: Option<&'a (dyn std::any::Any + Send + Sync)>, + ) -> graphql_orm::futures::future::BoxFuture<'a, async_graphql::Result> { + Box::pin(async { Ok(true) }) + } +} +async fn setup() -> (Database, Policy) { + let mut db = Database::connect_sqlite("sqlite::memory:").await.unwrap(); + sqlx::query("CREATE TABLE tuple_parents (id TEXT PRIMARY KEY, organization_id TEXT NOT NULL, provider_account_id TEXT NOT NULL, subscription_id TEXT NOT NULL)").execute(db.pool()).await.unwrap(); + sqlx::query("CREATE TABLE tuple_children (cache_id TEXT PRIMARY KEY, organization_id TEXT NOT NULL, provider_account_id TEXT NOT NULL, subscription_id TEXT NOT NULL)").execute(db.pool()).await.unwrap(); + sqlx::query("INSERT INTO tuple_parents VALUES ('parent-a','org-a','account','shared'), ('parent-b','org-b','account','shared')").execute(db.pool()).await.unwrap(); + let policy = Policy { + source: Arc::new(AtomicBool::new(true)), + target: Arc::new(AtomicBool::new(true)), + fields: Arc::new(AtomicBool::new(true)), + target_fields: Arc::new(AtomicBool::new(true)), + }; + db.set_row_policy(policy.clone()); + db.set_field_policy(policy.clone()); + (db, policy) +} +fn input(id: &str) -> CreateTupleChildInput { + CreateTupleChildInput { + cache_id: id.into(), + organization_id: "org-a".into(), + provider_account_id: "account".into(), + subscription_id: "shared".into(), + } +} +async fn registered + Unpin>(stream: &mut S) { + poll_fn(|cx| match stream.poll_next_unpin(cx) { + Poll::Pending => Poll::Ready(()), + other => panic!("unexpected initial event {other:?}"), + }) + .await; +} +const QUERY: &str = + "subscription { tupleParentChanged { id sourceId path tupleParent { organizationId } } }"; + +#[tokio::test] +async fn complete_relation_tuple_finds_scalar_parent_without_using_child_primary_key() { + let (db, _) = setup().await; + let schema = schema_builder(db.clone()).finish(); + let mut stream = + Box::pin(schema.execute_stream(Request::new(QUERY).data(AuthSubject::new("tester")))); + registered(&mut stream).await; + TupleChild::insert(&db, input("unrelated-cache-pk")) + .await + .unwrap(); + let response = timeout(Duration::from_secs(2), stream.next()) + .await + .unwrap() + .unwrap(); + assert!(response.errors.is_empty(), "{:?}", response.errors); + let value = response.data.into_json().unwrap(); + assert_eq!(value["tupleParentChanged"]["id"], "parent-a"); + assert_eq!( + value["tupleParentChanged"]["sourceId"], + "unrelated-cache-pk" + ); + assert_eq!( + value["tupleParentChanged"]["tupleParent"]["organizationId"], + "org-a" + ); + assert!( + timeout(Duration::from_millis(100), stream.next()) + .await + .is_err(), + "must not notify colliding org-b parent" + ); +} +#[tokio::test] +async fn source_target_and_each_tuple_field_revoke_after_registration() { + for denied in ["source", "target", "field", "target_field"] { + let (db, policy) = setup().await; + let schema = schema_builder(db.clone()).finish(); + let mut stream = + Box::pin(schema.execute_stream(Request::new(QUERY).data(AuthSubject::new("tester")))); + registered(&mut stream).await; + match denied { + "source" => policy.source.store(false, Ordering::SeqCst), + "target" => policy.target.store(false, Ordering::SeqCst), + "target_field" => policy.target_fields.store(false, Ordering::SeqCst), + _ => policy.fields.store(false, Ordering::SeqCst), + } + TupleChild::insert(&db, input("denied-cache")) + .await + .unwrap(); + assert!( + timeout(Duration::from_millis(100), stream.next()) + .await + .is_err(), + "released denied {denied}" + ); + } +} +#[tokio::test] +async fn rolled_back_tuple_notification_never_reaches_generated_stream() { + let (db, _) = setup().await; + let schema = schema_builder(db.clone()).finish(); + let mut stream = + Box::pin(schema.execute_stream(Request::new(QUERY).data(AuthSubject::new("tester")))); + registered(&mut stream).await; + let result: Result<(), TransactionError> = db + .transaction(TransactionMode::Default, |tx| { + Box::pin(async move { + tx.insert::(input("rolled-back")).await?; + Err(OrmPublicError::forbidden()) + }) + }) + .await; + assert!(result.is_err()); + assert!( + timeout(Duration::from_millis(100), stream.next()) + .await + .is_err() + ); +} +#[tokio::test] +async fn moving_relation_after_commit_suppresses_stale_provenance() { + let (db, _) = setup().await; + let schema = schema_builder(db.clone()).finish(); + let mut stream = + Box::pin(schema.execute_stream(Request::new(QUERY).data(AuthSubject::new("tester")))); + registered(&mut stream).await; + TupleChild::insert(&db, input("moved-cache")).await.unwrap(); + sqlx::query("UPDATE tuple_children SET organization_id='org-b' WHERE cache_id='moved-cache'") + .execute(db.pool()) + .await + .unwrap(); + assert!( + timeout(Duration::from_millis(100), stream.next()) + .await + .is_err() + ); +} + +// Provider IDs and cache primary keys deliberately have different Rust types. +// Using the first relation field as the parent's key fails to compile here. +mod alternate_provider_keys { + use super::*; + + #[derive( + GraphQLEntity, GraphQLOperations, Clone, Debug, serde::Serialize, serde::Deserialize, + )] + #[graphql_entity(table = "alternate_organizations", plural = "AlternateOrganizations")] + struct AlternateOrganization { + #[primary_key] + #[sortable] + #[graphql_orm(auto_generated = false)] + cache_id: String, + provider_id: i32, + } + + #[derive( + GraphQLEntity, GraphQLOperations, Clone, Debug, serde::Serialize, serde::Deserialize, + )] + #[graphql_entity(table = "alternate_devices", plural = "AlternateDevices")] + struct AlternateDevice { + #[primary_key] + #[sortable] + #[graphql_orm(auto_generated = false)] + cache_id: String, + organization_id: i32, + #[graphql(skip)] + #[relation( + target = "AlternateOrganization", + from = "organization_id", + to = "provider_id", + propagate_change = "up", + emit_fk = false + )] + organization: Option, + } + + #[derive( + GraphQLEntity, GraphQLOperations, Clone, Debug, serde::Serialize, serde::Deserialize, + )] + #[graphql_entity(table = "provider_locations", plural = "ProviderLocations")] + struct ProviderLocation { + #[primary_key] + #[sortable] + #[graphql_orm(auto_generated = false)] + cache_id: String, + organization_id: i32, + provider_id: i32, + } + + #[derive( + GraphQLEntity, GraphQLOperations, Clone, Debug, serde::Serialize, serde::Deserialize, + )] + #[graphql_entity(table = "location_devices", plural = "LocationDevices")] + struct LocationDevice { + #[primary_key] + #[sortable] + #[graphql_orm(auto_generated = false)] + cache_id: String, + organization_id: i32, + location_id: i32, + #[graphql(skip)] + #[relation(target = "ProviderLocation", from = ["organization_id", "location_id"], to = ["organization_id", "provider_id"], propagate_change = "up", emit_fk = false)] + location: Option, + } + + schema_roots! { query_custom_ops: [], entities: [AlternateOrganization, AlternateDevice, ProviderLocation, LocationDevice], } + + #[tokio::test] + async fn scalar_alternate_i32_relation_resolves_string_cache_primary_key() { + let db = Database::connect_sqlite("sqlite::memory:").await.unwrap(); + sqlx::query("CREATE TABLE alternate_organizations (cache_id TEXT PRIMARY KEY, provider_id INTEGER NOT NULL UNIQUE)").execute(db.pool()).await.unwrap(); + sqlx::query("CREATE TABLE alternate_devices (cache_id TEXT PRIMARY KEY, organization_id INTEGER NOT NULL)").execute(db.pool()).await.unwrap(); + sqlx::query("INSERT INTO alternate_organizations VALUES ('org-cache-key',42)") + .execute(db.pool()) + .await + .unwrap(); + let schema = schema_builder(db.clone()).finish(); + let mut stream = Box::pin( + schema.execute_stream( + Request::new("subscription { alternateOrganizationChanged { id sourceId } }") + .data(AuthSubject::new("tester")), + ), + ); + registered(&mut stream).await; + AlternateDevice::insert( + &db, + CreateAlternateDeviceInput { + cache_id: "device-cache-key".into(), + organization_id: 42, + }, + ) + .await + .unwrap(); + let response = timeout(Duration::from_secs(2), stream.next()) + .await + .unwrap() + .unwrap(); + assert!(response.errors.is_empty(), "{:?}", response.errors); + let value = response.data.into_json().unwrap(); + assert_eq!(value["alternateOrganizationChanged"]["id"], "org-cache-key"); + assert_eq!( + value["alternateOrganizationChanged"]["sourceId"], + "device-cache-key" + ); + assert!( + timeout(Duration::from_millis(100), stream.next()) + .await + .is_err() + ); + } + + #[tokio::test] + async fn numeric_composite_alternate_relation_keeps_organization_in_binding() { + let db = Database::connect_sqlite("sqlite::memory:").await.unwrap(); + sqlx::query("CREATE TABLE provider_locations (cache_id TEXT PRIMARY KEY, organization_id INTEGER NOT NULL, provider_id INTEGER NOT NULL, UNIQUE(organization_id,provider_id))").execute(db.pool()).await.unwrap(); + sqlx::query("CREATE TABLE location_devices (cache_id TEXT PRIMARY KEY, organization_id INTEGER NOT NULL, location_id INTEGER NOT NULL)").execute(db.pool()).await.unwrap(); + sqlx::query("INSERT INTO provider_locations VALUES ('location-org-a',1,42), ('location-org-b',2,42)").execute(db.pool()).await.unwrap(); + let schema = schema_builder(db.clone()).finish(); + let mut stream = Box::pin( + schema.execute_stream( + Request::new("subscription { providerLocationChanged { id sourceId } }") + .data(AuthSubject::new("tester")), + ), + ); + registered(&mut stream).await; + LocationDevice::insert( + &db, + CreateLocationDeviceInput { + cache_id: "device-other-key".into(), + organization_id: 1, + location_id: 42, + }, + ) + .await + .unwrap(); + let response = timeout(Duration::from_secs(2), stream.next()) + .await + .unwrap() + .unwrap(); + assert!(response.errors.is_empty(), "{:?}", response.errors); + let value = response.data.into_json().unwrap(); + assert_eq!(value["providerLocationChanged"]["id"], "location-org-a"); + assert_eq!( + value["providerLocationChanged"]["sourceId"], + "device-other-key" + ); + assert!( + timeout(Duration::from_millis(100), stream.next()) + .await + .is_err(), + "must not release org-b's colliding provider ID" + ); + } +} diff --git a/docs/reference/workspace-packages.md b/docs/reference/workspace-packages.md index 494298ac..1365cf2d 100644 --- a/docs/reference/workspace-packages.md +++ b/docs/reference/workspace-packages.md @@ -18,11 +18,11 @@ changes. | Package | Version | Path | Default features | Direct internal dependencies | | --- | --- | --- | --- | --- | -| `graphql-orm` | `0.62.0` | `crates/graphql-orm` | `sqlite` | `graphql-orm-macros`, `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | +| `graphql-orm` | `0.63.0` | `crates/graphql-orm` | `sqlite` | `graphql-orm-macros`, `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | | `graphql-orm-ai` | `0.109.1` | `crates/graphql-orm-ai` | `sqlite` | `graphql-orm`, `graphql-orm` (dev-only), `graphql-orm-ai-tool-profiles`, `graphql-orm-storage` | | `graphql-orm-ai-tool-profiles` | `0.16.2` | `crates/graphql-orm-ai-tool-profiles` | none | `graphql-orm-operation-catalog`, `graphql-orm-router-protocol` (dev-only) | | `graphql-orm-backup` | `0.10.0` | `crates/graphql-orm-backup` | `local` | `graphql-orm` (optional), `graphql-orm-storage` | -| `graphql-orm-macros` | `0.62.0` | `crates/graphql-orm-macros` | `sqlite` | none | +| `graphql-orm-macros` | `0.63.0` | `crates/graphql-orm-macros` | `sqlite` | none | | `graphql-orm-operation-catalog` | `0.4.2` | `crates/graphql-orm-operation-catalog` | none | `graphql-orm-router-protocol` (optional) | | `graphql-orm-router` | `0.6.5` | `crates/graphql-orm-router` | none | `graphql-orm-router-protocol` | | `graphql-orm-router-protocol` | `0.2.1` | `crates/graphql-orm-router-protocol` | none | none |