fraiseql_core/runtime/executor/

query.rs

//! Regular and relay query execution.

use std::sync::Arc;

use super::{Executor, null_masked_fields, resolve_inject_value};
use crate::{
    db::{
        CursorValue, ProjectionField, WhereClause, WhereOperator,
        projection_generator::PostgresProjectionGenerator, traits::DatabaseAdapter,
    },
    error::{FraiseQLError, Result},
    runtime::{JsonbStrategy, ResultProjector},
    schema::SqlProjectionHint,
    security::{RlsWhereClause, SecurityContext},
};

impl<A: DatabaseAdapter> Executor<A> {
    /// Execute a regular query with row-level security (RLS) filtering.
    ///
    /// This method:
    /// 1. Validates the user's security context (token expiration, etc.)
    /// 2. Evaluates RLS policies to determine what rows the user can access
    /// 3. Composes RLS filters with user-provided WHERE clauses
    /// 4. Passes the composed filter to the database adapter for SQL-level filtering
    ///
    /// RLS filtering happens at the database level, not in Rust, ensuring:
    /// - High performance (database can optimize filters)
    /// - Correct handling of pagination (LIMIT applied after RLS filtering)
    /// - Type-safe composition via `WhereClause` enum
    ///
    /// # Errors
    ///
    /// * [`FraiseQLError::Validation`] — security token expired, role check failed, or query not
    ///   found in schema.
    /// * [`FraiseQLError::Database`] — the database adapter returned an error.
    pub(super) async fn execute_regular_query_with_security(
        &self,
        query: &str,
        variables: Option<&serde_json::Value>,
        security_context: &SecurityContext,
    ) -> Result<String> {
        // 1. Validate security context (check expiration, etc.)
        if security_context.is_expired() {
            return Err(FraiseQLError::Validation {
                message: "Security token has expired".to_string(),
                path:    Some("request.authorization".to_string()),
            });
        }

        // 2. Match query to compiled template
        let query_match = self.matcher.match_query(query, variables)?;

        // 2b. Enforce requires_role — return "not found" (not "forbidden") to prevent enumeration
        if let Some(ref required_role) = query_match.query_def.requires_role {
            if !security_context.roles.iter().any(|r| r == required_role) {
                return Err(FraiseQLError::Validation {
                    message: format!("Query '{}' not found in schema", query_match.query_def.name),
                    path:    None,
                });
            }
        }

        // Route relay queries to dedicated handler with security context.
        if query_match.query_def.relay {
            return self.execute_relay_query(&query_match, variables, Some(security_context)).await;
        }

        // 3. Create execution plan
        let plan = self.planner.plan(&query_match)?;

        // 4. Evaluate RLS policy and build WHERE clause filter. The return type is
        //    Option<RlsWhereClause> — a compile-time proof that the clause passed through RLS
        //    evaluation.
        let rls_where_clause: Option<RlsWhereClause> =
            if let Some(ref rls_policy) = self.config.rls_policy {
                // Evaluate RLS policy with user's security context
                rls_policy.evaluate(security_context, &query_match.query_def.name)?
            } else {
                // No RLS policy configured, allow all access
                None
            };

        // 5. Get SQL source from query definition
        let sql_source =
            query_match
                .query_def
                .sql_source
                .as_ref()
                .ok_or_else(|| FraiseQLError::Validation {
                    message: "Query has no SQL source".to_string(),
                    path:    None,
                })?;

        // 6. Generate SQL projection hint for requested fields (optimization) Look up field types
        //    from schema to use the correct JSONB operator: `->` for objects/lists (preserves
        //    JSONB), `->>` for scalars (text).
        let projection_hint = if !plan.projection_fields.is_empty()
            && plan.jsonb_strategy == JsonbStrategy::Project
        {
            let type_def = self.schema.find_type(&query_match.query_def.return_type);
            let typed_fields: Vec<ProjectionField> = plan
                .projection_fields
                .iter()
                .map(|name| {
                    let is_composite = type_def
                        .and_then(|td| td.fields.iter().find(|f| f.name == name.as_str()))
                        .is_some_and(|f| !f.field_type.is_scalar());
                    ProjectionField {
                        name: name.clone(),
                        is_composite,
                    }
                })
                .collect();

            let generator = PostgresProjectionGenerator::new();
            let projection_sql = generator
                .generate_typed_projection_sql(&typed_fields)
                .unwrap_or_else(|_| "data".to_string());

            Some(SqlProjectionHint {
                database:                    self.adapter.database_type(),
                projection_template:         projection_sql,
                estimated_reduction_percent: compute_projection_reduction(
                    plan.projection_fields.len(),
                ),
            })
        } else {
            // Stream strategy: return full JSONB, no projection hint
            None
        };

        // 7. AND inject conditions onto the RLS WHERE clause. Inject conditions always come after
        //    RLS so they cannot bypass it.
        let combined_where: Option<WhereClause> = if query_match.query_def.inject_params.is_empty()
        {
            // Common path: unwrap RlsWhereClause into WhereClause for the adapter
            rls_where_clause.map(RlsWhereClause::into_where_clause)
        } else {
            let mut conditions: Vec<WhereClause> = query_match
                .query_def
                .inject_params
                .iter()
                .map(|(col, source)| {
                    let value = resolve_inject_value(col, source, security_context)?;
                    Ok(WhereClause::Field {
                        path: vec![col.clone()],
                        operator: WhereOperator::Eq,
                        value,
                    })
                })
                .collect::<Result<Vec<_>>>()?;

            if let Some(rls) = rls_where_clause {
                conditions.insert(0, rls.into_where_clause());
            }
            match conditions.len() {
                0 => None,
                1 => Some(conditions.remove(0)),
                _ => Some(WhereClause::And(conditions)),
            }
        };

        // 5b. Compose user-supplied WHERE from GraphQL arguments when has_where is enabled.
        //     Security conditions (RLS + inject) are always first so they cannot be bypassed.
        let combined_where: Option<WhereClause> = if query_match.query_def.auto_params.has_where {
            let user_where = query_match
                .arguments
                .get("where")
                .map(WhereClause::from_graphql_json)
                .transpose()?;
            match (combined_where, user_where) {
                (None, None) => None,
                (Some(sec), None) => Some(sec),
                (None, Some(user)) => Some(user),
                (Some(sec), Some(user)) => Some(WhereClause::And(vec![sec, user])),
            }
        } else {
            combined_where
        };

        // 8. Extract limit/offset from query arguments when auto_params are enabled
        let limit = if query_match.query_def.auto_params.has_limit {
            query_match
                .arguments
                .get("limit")
                .and_then(|v| v.as_u64())
                .and_then(|v| u32::try_from(v).ok())
        } else {
            None
        };

        let offset = if query_match.query_def.auto_params.has_offset {
            query_match
                .arguments
                .get("offset")
                .and_then(|v| v.as_u64())
                .and_then(|v| u32::try_from(v).ok())
        } else {
            None
        };

        // 8b. Extract order_by from query arguments when has_order_by is enabled
        let order_by_clauses = if query_match.query_def.auto_params.has_order_by {
            query_match
                .arguments
                .get("orderBy")
                .map(crate::db::OrderByClause::from_graphql_json)
                .transpose()?
        } else {
            None
        };

        // 9. Execute query with combined WHERE clause filter
        let results = self
            .adapter
            .execute_with_projection(
                sql_source,
                projection_hint.as_ref(),
                combined_where.as_ref(),
                limit,
                offset,
                order_by_clauses.as_deref(),
            )
            .await?;

        // 10. Apply field-level RBAC filtering (reject / mask / allow)
        let access = self.apply_field_rbac_filtering(
            &query_match.query_def.return_type,
            plan.projection_fields,
            security_context,
        )?;

        // 11. Project results — include both allowed and masked fields in projection
        let mut all_projection_fields = access.allowed;
        all_projection_fields.extend(access.masked.iter().cloned());
        let projector = ResultProjector::new(all_projection_fields);
        let mut projected =
            projector.project_results(&results, query_match.query_def.returns_list)?;

        // 11. Null out masked fields in the projected result
        if !access.masked.is_empty() {
            null_masked_fields(&mut projected, &access.masked);
        }

        // 12. Wrap in GraphQL data envelope
        let response =
            ResultProjector::wrap_in_data_envelope(projected, &query_match.query_def.name);

        // 13. Serialize to JSON string
        Ok(serde_json::to_string(&response)?)
    }

    /// Execute a regular (non-aggregate, non-relay) GraphQL query.
    ///
    /// # Errors
    ///
    /// Returns [`FraiseQLError::Validation`] if the query does not match a compiled
    /// template or requires a security context that is not present.
    /// Returns [`FraiseQLError::Database`] if the SQL execution or result projection fails.
    pub(super) async fn execute_regular_query(
        &self,
        query: &str,
        variables: Option<&serde_json::Value>,
    ) -> Result<String> {
        // 1. Match query to compiled template
        let query_match = self.matcher.match_query(query, variables)?;

        // Guard: role-restricted queries are invisible to unauthenticated users
        if query_match.query_def.requires_role.is_some() {
            return Err(FraiseQLError::Validation {
                message: format!("Query '{}' not found in schema", query_match.query_def.name),
                path:    None,
            });
        }

        // Guard: queries with inject params require a security context.
        if !query_match.query_def.inject_params.is_empty() {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "Query '{}' has inject params but was called without a security context",
                    query_match.query_def.name
                ),
                path:    None,
            });
        }

        // Route relay queries to dedicated handler.
        if query_match.query_def.relay {
            return self.execute_relay_query(&query_match, variables, None).await;
        }

        // 2. Create execution plan
        let plan = self.planner.plan(&query_match)?;

        // 3. Execute SQL query
        let sql_source = query_match.query_def.sql_source.as_ref().ok_or_else(|| {
            crate::error::FraiseQLError::Validation {
                message: "Query has no SQL source".to_string(),
                path:    None,
            }
        })?;

        // 3a. Generate SQL projection hint for requested fields (optimization)
        // Strategy selection: Project (extract fields) vs Stream (return full JSONB)
        // This reduces payload by projecting only requested fields at the database level
        let projection_hint = if !plan.projection_fields.is_empty()
            && plan.jsonb_strategy == JsonbStrategy::Project
        {
            let generator = PostgresProjectionGenerator::new();
            let projection_sql = generator
                .generate_projection_sql(&plan.projection_fields)
                .unwrap_or_else(|_| "data".to_string());

            Some(SqlProjectionHint {
                database:                    self.adapter.database_type(),
                projection_template:         projection_sql,
                estimated_reduction_percent: compute_projection_reduction(
                    plan.projection_fields.len(),
                ),
            })
        } else {
            // Stream strategy: return full JSONB, no projection hint
            None
        };

        // 3b. Extract auto_params (limit, offset, where, order_by) from arguments
        let user_where: Option<WhereClause> = if query_match.query_def.auto_params.has_where {
            query_match
                .arguments
                .get("where")
                .map(WhereClause::from_graphql_json)
                .transpose()?
        } else {
            None
        };

        let limit = if query_match.query_def.auto_params.has_limit {
            query_match
                .arguments
                .get("limit")
                .and_then(|v| v.as_u64())
                .and_then(|v| u32::try_from(v).ok())
        } else {
            None
        };

        let offset = if query_match.query_def.auto_params.has_offset {
            query_match
                .arguments
                .get("offset")
                .and_then(|v| v.as_u64())
                .and_then(|v| u32::try_from(v).ok())
        } else {
            None
        };

        let order_by_clauses = if query_match.query_def.auto_params.has_order_by {
            query_match
                .arguments
                .get("orderBy")
                .map(crate::db::OrderByClause::from_graphql_json)
                .transpose()?
        } else {
            None
        };

        let results = self
            .adapter
            .execute_with_projection(
                sql_source,
                projection_hint.as_ref(),
                user_where.as_ref(),
                limit,
                offset,
                order_by_clauses.as_deref(),
            )
            .await?;

        // 4. Project results
        let projector = ResultProjector::new(plan.projection_fields);
        let projected = projector.project_results(&results, query_match.query_def.returns_list)?;

        // 5. Wrap in GraphQL data envelope
        let response =
            ResultProjector::wrap_in_data_envelope(projected, &query_match.query_def.name);

        // 6. Serialize to JSON string
        Ok(serde_json::to_string(&response)?)
    }

    /// Execute a pre-built `QueryMatch` directly, bypassing GraphQL string parsing.
    ///
    /// Used by the REST transport for embedded sub-queries and NDJSON streaming
    /// where the query parameters are already resolved from HTTP request parameters.
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::Validation` if the query has no SQL source.
    /// Returns `FraiseQLError::Database` if the adapter returns an error.
    pub async fn execute_query_direct(
        &self,
        query_match: &crate::runtime::matcher::QueryMatch,
        _variables: Option<&serde_json::Value>,
        security_context: Option<&SecurityContext>,
    ) -> Result<String> {
        // Evaluate RLS policy if present.
        let rls_where_clause: Option<RlsWhereClause> = if let (Some(ref rls_policy), Some(ctx)) =
            (&self.config.rls_policy, security_context)
        {
            rls_policy.evaluate(ctx, &query_match.query_def.name)?
        } else {
            None
        };

        // Get SQL source.
        let sql_source =
            query_match
                .query_def
                .sql_source
                .as_ref()
                .ok_or_else(|| FraiseQLError::Validation {
                    message: "Query has no SQL source".to_string(),
                    path:    None,
                })?;

        // Build execution plan.
        let plan = self.planner.plan(query_match)?;

        // Extract auto_params from arguments.
        let user_where: Option<WhereClause> = if query_match.query_def.auto_params.has_where {
            query_match
                .arguments
                .get("where")
                .map(WhereClause::from_graphql_json)
                .transpose()?
        } else {
            None
        };

        let limit = query_match
            .arguments
            .get("limit")
            .and_then(|v| v.as_u64())
            .and_then(|v| u32::try_from(v).ok());

        let offset = query_match
            .arguments
            .get("offset")
            .and_then(|v| v.as_u64())
            .and_then(|v| u32::try_from(v).ok());

        let order_by_clauses = query_match
            .arguments
            .get("orderBy")
            .map(crate::db::OrderByClause::from_graphql_json)
            .transpose()?;

        // Compose RLS and user WHERE clauses.
        let composed_where = match (&rls_where_clause, &user_where) {
            (Some(rls), Some(user)) => {
                Some(WhereClause::And(vec![rls.as_where_clause().clone(), user.clone()]))
            },
            (Some(rls), None) => Some(rls.as_where_clause().clone()),
            (None, Some(user)) => Some(user.clone()),
            (None, None) => None,
        };

        // Inject security-derived params.
        if !query_match.query_def.inject_params.is_empty() {
            if let Some(ctx) = security_context {
                for (param_name, source) in &query_match.query_def.inject_params {
                    let _value = resolve_inject_value(param_name, source, ctx)?;
                    // Injected params are applied at the SQL level via WHERE clauses,
                    // not via GraphQL variables, so no mutation of variables is needed here.
                }
            }
        }

        // Execute.
        let results = self
            .adapter
            .execute_with_projection(
                sql_source,
                None,
                composed_where.as_ref(),
                limit,
                offset,
                order_by_clauses.as_deref(),
            )
            .await?;

        // Project results.
        let projector = ResultProjector::new(plan.projection_fields);
        let projected = projector.project_results(&results, query_match.query_def.returns_list)?;

        // Wrap in GraphQL data envelope.
        let response =
            ResultProjector::wrap_in_data_envelope(projected, &query_match.query_def.name);

        Ok(serde_json::to_string(&response)?)
    }

    /// Execute a mutation with security context for REST transport.
    ///
    /// Delegates to the standard mutation execution path with RLS enforcement.
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::Database` if the adapter returns an error.
    /// Returns `FraiseQLError::Validation` if inject params require a missing security context.
    pub async fn execute_mutation_with_security(
        &self,
        mutation_name: &str,
        arguments: &serde_json::Value,
        security_context: Option<&crate::security::SecurityContext>,
    ) -> crate::error::Result<String> {
        // Build a synthetic GraphQL mutation query and delegate to execute()
        let args_str = if let Some(obj) = arguments.as_object() {
            obj.iter().map(|(k, v)| format!("{k}: {v}")).collect::<Vec<_>>().join(", ")
        } else {
            String::new()
        };
        let query = if args_str.is_empty() {
            format!("mutation {{ {mutation_name} {{ status entity_id message }} }}")
        } else {
            format!("mutation {{ {mutation_name}({args_str}) {{ status entity_id message }} }}")
        };

        if let Some(ctx) = security_context {
            self.execute_with_security(&query, None, ctx).await
        } else {
            self.execute(&query, None).await
        }
    }

    /// Execute a batch of mutations (for REST bulk insert).
    ///
    /// Executes each mutation individually and collects results into a `BulkResult`.
    ///
    /// # Errors
    ///
    /// Returns the first error encountered during batch execution.
    pub async fn execute_mutation_batch(
        &self,
        mutation_name: &str,
        items: &[serde_json::Value],
        security_context: Option<&crate::security::SecurityContext>,
    ) -> crate::error::Result<crate::runtime::BulkResult> {
        let mut entities = Vec::with_capacity(items.len());
        for item in items {
            let result = self
                .execute_mutation_with_security(mutation_name, item, security_context)
                .await?;
            entities.push(serde_json::Value::String(result));
        }
        Ok(crate::runtime::BulkResult {
            affected_rows: entities.len() as u64,
            entities:      Some(entities),
        })
    }

    /// Execute a bulk operation (collection-level PATCH/DELETE) by filter.
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::Database` if the adapter returns an error.
    pub async fn execute_bulk_by_filter(
        &self,
        query_match: &crate::runtime::matcher::QueryMatch,
        mutation_name: &str,
        body: Option<&serde_json::Value>,
        _id_field: &str,
        _max_affected: u64,
        security_context: Option<&SecurityContext>,
    ) -> crate::error::Result<crate::runtime::BulkResult> {
        // Execute the filter query to find matching rows.
        let result_str = self.execute_query_direct(query_match, None, security_context).await?;

        let args = body.cloned().unwrap_or(serde_json::json!({}));
        let result = self
            .execute_mutation_with_security(mutation_name, &args, security_context)
            .await?;

        let parsed: serde_json::Value =
            serde_json::from_str(&result_str).unwrap_or(serde_json::json!({}));
        let count = parsed
            .get("data")
            .and_then(|d| d.as_object())
            .and_then(|o| o.values().next())
            .and_then(|v| v.as_array())
            .map_or(1, |a| a.len() as u64);

        Ok(crate::runtime::BulkResult {
            affected_rows: count,
            entities:      Some(vec![serde_json::Value::String(result)]),
        })
    }

    /// Count the total number of rows matching the query's WHERE and RLS conditions.
    ///
    /// Issues a `SELECT COUNT(*) FROM {view} WHERE {conditions}` query, ignoring
    /// pagination (ORDER BY, LIMIT, OFFSET). Useful for REST `X-Total-Count` headers
    /// and `count=exact` query parameter support.
    ///
    /// # Arguments
    ///
    /// * `query_match` - Pre-built query match identifying the SQL source and filters
    /// * `variables` - Optional variables (unused for count, reserved for future use)
    /// * `security_context` - Optional authenticated user context for RLS and inject
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::Validation` if the query has no SQL source, or if
    /// inject params are required but no security context is provided.
    /// Returns `FraiseQLError::Database` if the adapter returns an error.
    pub async fn count_rows(
        &self,
        query_match: &crate::runtime::matcher::QueryMatch,
        _variables: Option<&serde_json::Value>,
        security_context: Option<&SecurityContext>,
    ) -> Result<u64> {
        // 1. Evaluate RLS policy
        let rls_where_clause: Option<RlsWhereClause> = if let (Some(ref rls_policy), Some(ctx)) =
            (&self.config.rls_policy, security_context)
        {
            rls_policy.evaluate(ctx, &query_match.query_def.name)?
        } else {
            None
        };

        // 2. Get SQL source
        let sql_source =
            query_match
                .query_def
                .sql_source
                .as_ref()
                .ok_or_else(|| FraiseQLError::Validation {
                    message: "Query has no SQL source".to_string(),
                    path:    None,
                })?;

        // 3. Build combined WHERE clause (RLS + inject)
        let combined_where: Option<WhereClause> = if query_match.query_def.inject_params.is_empty()
        {
            rls_where_clause.map(RlsWhereClause::into_where_clause)
        } else {
            let ctx = security_context.ok_or_else(|| FraiseQLError::Validation {
                message: format!(
                    "Query '{}' has inject params but no security context is available",
                    query_match.query_def.name
                ),
                path:    None,
            })?;
            let mut conditions: Vec<WhereClause> = query_match
                .query_def
                .inject_params
                .iter()
                .map(|(col, source)| {
                    let value = resolve_inject_value(col, source, ctx)?;
                    Ok(WhereClause::Field {
                        path: vec![col.clone()],
                        operator: WhereOperator::Eq,
                        value,
                    })
                })
                .collect::<Result<Vec<_>>>()?;

            if let Some(rls) = rls_where_clause {
                conditions.insert(0, rls.into_where_clause());
            }
            match conditions.len() {
                0 => None,
                1 => Some(conditions.remove(0)),
                _ => Some(WhereClause::And(conditions)),
            }
        };

        // 3b. Compose user-supplied WHERE when has_where is enabled (same as execute_from_match).
        let combined_where: Option<WhereClause> = if query_match.query_def.auto_params.has_where {
            let user_where = query_match
                .arguments
                .get("where")
                .map(WhereClause::from_graphql_json)
                .transpose()?;
            match (combined_where, user_where) {
                (None, None) => None,
                (Some(sec), None) => Some(sec),
                (None, Some(user)) => Some(user),
                (Some(sec), Some(user)) => Some(WhereClause::And(vec![sec, user])),
            }
        } else {
            combined_where
        };

        // 4. Execute COUNT query via adapter
        let rows = self
            .adapter
            .execute_where_query(sql_source, combined_where.as_ref(), None, None, None)
            .await?;

        // Return the row count
        #[allow(clippy::cast_possible_truncation)] // Reason: row count fits u64
        Ok(rows.len() as u64)
    }

    /// Execute a Relay connection query with cursor-based (keyset) pagination.
    ///
    /// Reads `first`, `after`, `last`, `before` from `variables`, fetches a page
    /// of rows using `pk_{type}` keyset ordering, and wraps the result in the
    /// Relay `XxxConnection` format:
    /// ```json
    /// {
    ///   "data": {
    ///     "users": {
    ///       "edges": [{ "cursor": "NDI=", "node": { "id": "...", ... } }],
    ///       "pageInfo": {
    ///         "hasNextPage": true, "hasPreviousPage": false,
    ///         "startCursor": "NDI=", "endCursor": "Mw=="
    ///       }
    ///     }
    ///   }
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`FraiseQLError::Validation`] if required pagination variables are
    /// missing or contain invalid cursor values.
    /// Returns [`FraiseQLError::Database`] if the SQL execution or result projection fails.
    pub(super) async fn execute_relay_query(
        &self,
        query_match: &crate::runtime::matcher::QueryMatch,
        variables: Option<&serde_json::Value>,
        security_context: Option<&SecurityContext>,
    ) -> Result<String> {
        use crate::{
            compiler::aggregation::OrderByClause,
            runtime::relay::{decode_edge_cursor, decode_uuid_cursor, encode_edge_cursor},
            schema::CursorType,
        };

        let query_def = &query_match.query_def;

        // Guard: queries with inject params require a security context.
        if !query_def.inject_params.is_empty() && security_context.is_none() {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "Query '{}' has inject params but was called without a security context",
                    query_def.name
                ),
                path:    None,
            });
        }

        let sql_source =
            query_def.sql_source.as_deref().ok_or_else(|| FraiseQLError::Validation {
                message: format!("Relay query '{}' has no sql_source configured", query_def.name),
                path:    None,
            })?;

        let cursor_column =
            query_def
                .relay_cursor_column
                .as_deref()
                .ok_or_else(|| FraiseQLError::Validation {
                    message: format!(
                        "Relay query '{}' has no relay_cursor_column derived",
                        query_def.name
                    ),
                    path:    None,
                })?;

        // Guard: relay pagination requires the executor to have been constructed
        // via `Executor::new_with_relay` with a `RelayDatabaseAdapter`.
        let relay = self.relay.as_ref().ok_or_else(|| FraiseQLError::Validation {
            message: format!(
                "Relay pagination is not supported by the {} adapter. \
                 Use a relay-capable adapter (e.g. PostgreSQL) and construct \
                 the executor with `Executor::new_with_relay`.",
                self.adapter.database_type()
            ),
            path:    None,
        })?;

        // --- RLS + inject_params evaluation (same logic as execute_from_match) ---
        // Evaluate RLS policy to generate security WHERE clause.
        let rls_where_clause: Option<RlsWhereClause> = if let (Some(ref rls_policy), Some(ctx)) =
            (&self.config.rls_policy, security_context)
        {
            rls_policy.evaluate(ctx, &query_def.name)?
        } else {
            None
        };

        // Resolve inject_params from JWT claims and compose with RLS.
        let security_where: Option<WhereClause> = if query_def.inject_params.is_empty() {
            rls_where_clause.map(RlsWhereClause::into_where_clause)
        } else {
            let ctx = security_context.ok_or_else(|| FraiseQLError::Validation {
                message: format!(
                    "Query '{}' has inject params but was called without a security context",
                    query_def.name
                ),
                path:    None,
            })?;
            let mut conditions: Vec<WhereClause> = query_def
                .inject_params
                .iter()
                .map(|(col, source)| {
                    let value = resolve_inject_value(col, source, ctx)?;
                    Ok(WhereClause::Field {
                        path: vec![col.clone()],
                        operator: WhereOperator::Eq,
                        value,
                    })
                })
                .collect::<Result<Vec<_>>>()?;

            if let Some(rls) = rls_where_clause {
                conditions.insert(0, rls.into_where_clause());
            }
            match conditions.len() {
                0 => None,
                1 => Some(conditions.remove(0)),
                _ => Some(WhereClause::And(conditions)),
            }
        };

        // Extract relay pagination arguments from variables.
        let vars = variables.and_then(|v| v.as_object());
        let first: Option<u32> = vars
            .and_then(|v| v.get("first"))
            .and_then(|v| v.as_u64())
            .map(|n| u32::try_from(n).unwrap_or(u32::MAX));
        let last: Option<u32> = vars
            .and_then(|v| v.get("last"))
            .and_then(|v| v.as_u64())
            .map(|n| u32::try_from(n).unwrap_or(u32::MAX));
        let after_cursor: Option<&str> = vars.and_then(|v| v.get("after")).and_then(|v| v.as_str());
        let before_cursor: Option<&str> =
            vars.and_then(|v| v.get("before")).and_then(|v| v.as_str());

        // Decode base64 cursors — type depends on relay_cursor_type.
        // If a cursor string is provided but fails to decode, return a validation
        // error immediately. Silently ignoring an invalid cursor would return a
        // full result set, violating the client's pagination intent.
        let (after_pk, before_pk) =
            match query_def.relay_cursor_type {
                CursorType::Int64 => {
                    let after = match after_cursor {
                        Some(s) => Some(decode_edge_cursor(s).map(CursorValue::Int64).ok_or_else(
                            || FraiseQLError::Validation {
                                message: format!("invalid relay cursor for `after`: {s:?}"),
                                path:    Some("after".to_string()),
                            },
                        )?),
                        None => None,
                    };
                    let before = match before_cursor {
                        Some(s) => Some(decode_edge_cursor(s).map(CursorValue::Int64).ok_or_else(
                            || FraiseQLError::Validation {
                                message: format!("invalid relay cursor for `before`: {s:?}"),
                                path:    Some("before".to_string()),
                            },
                        )?),
                        None => None,
                    };
                    (after, before)
                },
                CursorType::Uuid => {
                    let after = match after_cursor {
                        Some(s) => {
                            Some(decode_uuid_cursor(s).map(CursorValue::Uuid).ok_or_else(|| {
                                FraiseQLError::Validation {
                                    message: format!("invalid relay cursor for `after`: {s:?}"),
                                    path:    Some("after".to_string()),
                                }
                            })?)
                        },
                        None => None,
                    };
                    let before = match before_cursor {
                        Some(s) => {
                            Some(decode_uuid_cursor(s).map(CursorValue::Uuid).ok_or_else(|| {
                                FraiseQLError::Validation {
                                    message: format!("invalid relay cursor for `before`: {s:?}"),
                                    path:    Some("before".to_string()),
                                }
                            })?)
                        },
                        None => None,
                    };
                    (after, before)
                },
            };

        // Determine direction and limit.
        // Forward pagination takes priority; fallback to 20 if neither first/last given.
        let (forward, page_size) = if last.is_some() && first.is_none() {
            (false, last.unwrap_or(20))
        } else {
            (true, first.unwrap_or(20))
        };

        // Fetch page_size + 1 rows to detect hasNextPage/hasPreviousPage.
        let fetch_limit = page_size + 1;

        // Parse optional `where` filter from variables.
        let user_where_clause = if query_def.auto_params.has_where {
            vars.and_then(|v| v.get("where"))
                .map(WhereClause::from_graphql_json)
                .transpose()?
        } else {
            None
        };

        // Compose final WHERE: security (RLS + inject) AND user-supplied WHERE.
        // Security conditions always come first so they cannot be bypassed.
        let combined_where = match (security_where, user_where_clause) {
            (None, None) => None,
            (Some(sec), None) => Some(sec),
            (None, Some(user)) => Some(user),
            (Some(sec), Some(user)) => Some(WhereClause::And(vec![sec, user])),
        };

        // Parse optional `orderBy` from variables.
        let order_by = if query_def.auto_params.has_order_by {
            vars.and_then(|v| v.get("orderBy"))
                .map(OrderByClause::from_graphql_json)
                .transpose()?
        } else {
            None
        };

        // Detect whether the client selected `totalCount` inside the connection.
        // Named fragment spreads are already expanded by the matcher's FragmentResolver.
        // Inline fragments (`... on UserConnection { totalCount }`) remain as FieldSelection
        // entries with a name starting with "..." — we recurse one level into those.
        let include_total_count = query_match
            .selections
            .iter()
            .find(|sel| sel.name == query_def.name)
            .is_some_and(|connection_field| {
                selections_contain_field(&connection_field.nested_fields, "totalCount")
            });

        // Capture before the move into execute_relay_page.
        let had_after = after_pk.is_some();
        let had_before = before_pk.is_some();

        let result = relay
            .execute_relay_page(
                sql_source,
                cursor_column,
                after_pk,
                before_pk,
                fetch_limit,
                forward,
                combined_where.as_ref(),
                order_by.as_deref(),
                include_total_count,
            )
            .await?;

        // Detect whether there are more pages.
        let has_extra = result.rows.len() > page_size as usize;
        let rows: Vec<_> = result.rows.into_iter().take(page_size as usize).collect();

        let (has_next_page, has_previous_page) = if forward {
            (has_extra, had_after)
        } else {
            (had_before, has_extra)
        };

        // Build edges: each edge has { cursor, node }.
        let mut edges = Vec::with_capacity(rows.len());
        let mut start_cursor_str: Option<String> = None;
        let mut end_cursor_str: Option<String> = None;

        for (i, row) in rows.iter().enumerate() {
            let data = &row.data;

            let col_val = data.as_object().and_then(|obj| obj.get(cursor_column));

            let cursor_str = match query_def.relay_cursor_type {
                CursorType::Int64 => col_val
                    .and_then(|v| v.as_i64())
                    .map(encode_edge_cursor)
                    .ok_or_else(|| FraiseQLError::Validation {
                        message: format!(
                            "Relay query '{}': cursor column '{}' not found or not an integer in \
                             result JSONB. Ensure the view exposes this column inside the `data` object.",
                            query_def.name, cursor_column
                        ),
                        path: None,
                    })?,
                CursorType::Uuid => col_val
                    .and_then(|v| v.as_str())
                    .map(crate::runtime::relay::encode_uuid_cursor)
                    .ok_or_else(|| FraiseQLError::Validation {
                        message: format!(
                            "Relay query '{}': cursor column '{}' not found or not a string in \
                             result JSONB. Ensure the view exposes this column inside the `data` object.",
                            query_def.name, cursor_column
                        ),
                        path: None,
                    })?,
            };

            if i == 0 {
                start_cursor_str = Some(cursor_str.clone());
            }
            end_cursor_str = Some(cursor_str.clone());

            edges.push(serde_json::json!({
                "cursor": cursor_str,
                "node": data,
            }));
        }

        let page_info = serde_json::json!({
            "hasNextPage": has_next_page,
            "hasPreviousPage": has_previous_page,
            "startCursor": start_cursor_str,
            "endCursor": end_cursor_str,
        });

        let mut connection = serde_json::json!({
            "edges": edges,
            "pageInfo": page_info,
        });

        // Include totalCount when the client requested it and the adapter provided it.
        if include_total_count {
            if let Some(count) = result.total_count {
                connection["totalCount"] = serde_json::json!(count);
            } else {
                connection["totalCount"] = serde_json::Value::Null;
            }
        }

        let response = ResultProjector::wrap_in_data_envelope(connection, &query_def.name);
        Ok(serde_json::to_string(&response)?)
    }

    /// Execute a Relay global `node(id: ID!)` query.
    ///
    /// Decodes the opaque node ID (`base64("TypeName:uuid")`), locates the
    /// appropriate SQL view by searching the compiled schema for a query that
    /// returns that type, and fetches the matching row.
    ///
    /// Returns `{ "data": { "node": <object> } }` on success, or
    /// `{ "data": { "node": null } }` when the object is not found.
    ///
    /// # Errors
    ///
    /// Returns `FraiseQLError::Validation` when:
    /// - The `id` argument is missing or malformed
    /// - No SQL view is registered for the requested type
    pub(super) async fn execute_node_query(
        &self,
        query: &str,
        variables: Option<&serde_json::Value>,
    ) -> Result<String> {
        use crate::{
            db::{WhereClause, where_clause::WhereOperator},
            runtime::relay::decode_node_id,
        };

        // 1. Extract the raw opaque ID. Priority: $variables.id > inline literal in query text.
        let raw_id: String = if let Some(id_val) = variables
            .and_then(|v| v.as_object())
            .and_then(|obj| obj.get("id"))
            .and_then(|v| v.as_str())
        {
            id_val.to_string()
        } else {
            // Fall back to extracting inline literal, e.g. node(id: "NDI=")
            Self::extract_inline_node_id(query).ok_or_else(|| FraiseQLError::Validation {
                message: "node query: missing or unresolvable 'id' argument".to_string(),
                path:    Some("node.id".to_string()),
            })?
        };

        // 2. Decode base64("TypeName:uuid") → (type_name, uuid).
        let (type_name, uuid) =
            decode_node_id(&raw_id).ok_or_else(|| FraiseQLError::Validation {
                message: format!("node query: invalid node ID '{raw_id}'"),
                path:    Some("node.id".to_string()),
            })?;

        // 3. Find the SQL view for this type (O(1) index lookup built at startup).
        let sql_source: Arc<str> =
            self.node_type_index.get(&type_name).cloned().ok_or_else(|| {
                FraiseQLError::Validation {
                    message: format!("node query: no registered SQL view for type '{type_name}'"),
                    path:    Some("node.id".to_string()),
                }
            })?;

        // 4. Build WHERE clause: data->>'id' = uuid
        let where_clause = WhereClause::Field {
            path:     vec!["id".to_string()],
            operator: WhereOperator::Eq,
            value:    serde_json::Value::String(uuid),
        };

        // 5. Execute the query (limit 1).
        let rows = self
            .adapter
            .execute_where_query(&sql_source, Some(&where_clause), Some(1), None, None)
            .await?;

        // 6. Return the first matching row (or null).
        let node_value = rows.into_iter().next().map_or(serde_json::Value::Null, |row| row.data);

        let response = ResultProjector::wrap_in_data_envelope(node_value, "node");
        Ok(serde_json::to_string(&response)?)
    }
}

/// Estimate the payload reduction percentage from projecting N fields.
///
/// Uses a simple heuristic: each projected field saves proportional space
/// relative to a baseline of 20 typical JSONB fields per row. Clamped to
/// [10, 90] so the hint is never misleadingly extreme.
fn compute_projection_reduction(projected_field_count: usize) -> u32 {
    // Baseline: assume a typical type has 20 fields.
    const BASELINE_FIELD_COUNT: usize = 20;
    let requested = projected_field_count.min(BASELINE_FIELD_COUNT);
    let saved = BASELINE_FIELD_COUNT.saturating_sub(requested);
    // saved / BASELINE * 100, clamped to [10, 90]
    #[allow(clippy::cast_possible_truncation)] // Reason: result is in 0..=100, fits u32
    let percent = ((saved * 100) / BASELINE_FIELD_COUNT) as u32;
    percent.clamp(10, 90)
}

/// Return `true` if `field_name` appears in `selections`, including inside inline
/// fragment entries (`FieldSelection` whose name starts with `"..."`).
///
/// Named fragment spreads are already flattened by [`FragmentResolver`] before this
/// is called, so we only need to recurse one level into inline fragments.
fn selections_contain_field(
    selections: &[crate::graphql::FieldSelection],
    field_name: &str,
) -> bool {
    for sel in selections {
        if sel.name == field_name {
            return true;
        }
        // Inline fragment: name starts with "..." (e.g. "...on UserConnection")
        if sel.name.starts_with("...") && selections_contain_field(&sel.nested_fields, field_name) {
            return true;
        }
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::graphql::FieldSelection;

    // -------------------------------------------------------------------------
    // Helpers
    // -------------------------------------------------------------------------

    fn leaf(name: &str) -> FieldSelection {
        FieldSelection {
            name:          name.to_string(),
            alias:         None,
            arguments:     vec![],
            nested_fields: vec![],
            directives:    vec![],
        }
    }

    fn fragment(name: &str, nested: Vec<FieldSelection>) -> FieldSelection {
        FieldSelection {
            name:          name.to_string(),
            alias:         None,
            arguments:     vec![],
            nested_fields: nested,
            directives:    vec![],
        }
    }

    // =========================================================================
    // compute_projection_reduction
    // =========================================================================

    #[test]
    fn projection_reduction_zero_fields_is_clamped_to_90() {
        // 0 fields requested → saved = 20 → 100% → clamped to 90
        assert_eq!(compute_projection_reduction(0), 90);
    }

    #[test]
    fn projection_reduction_all_fields_is_clamped_to_10() {
        // 20 fields (= baseline) → saved = 0 → 0% → clamped to 10
        assert_eq!(compute_projection_reduction(20), 10);
    }

    #[test]
    fn projection_reduction_above_baseline_clamps_to_10() {
        // 50 fields > 20 baseline → same as 20 → clamped to 10
        assert_eq!(compute_projection_reduction(50), 10);
    }

    #[test]
    fn projection_reduction_10_fields_is_50_percent() {
        // 10 requested → saved = 10 → 10/20 * 100 = 50 → within [10, 90]
        assert_eq!(compute_projection_reduction(10), 50);
    }

    #[test]
    fn projection_reduction_1_field_is_high() {
        // 1 requested → saved = 19 → 95% → clamped to 90
        assert_eq!(compute_projection_reduction(1), 90);
    }

    #[test]
    fn projection_reduction_result_always_in_clamp_range() {
        for n in 0_usize..=30 {
            let r = compute_projection_reduction(n);
            assert!((10..=90).contains(&r), "out of [10,90] for n={n}: got {r}");
        }
    }

    // =========================================================================
    // selections_contain_field
    // =========================================================================

    #[test]
    fn empty_selections_returns_false() {
        assert!(!selections_contain_field(&[], "totalCount"));
    }

    #[test]
    fn direct_match_returns_true() {
        let sels = vec![leaf("edges"), leaf("totalCount"), leaf("pageInfo")];
        assert!(selections_contain_field(&sels, "totalCount"));
    }

    #[test]
    fn absent_field_returns_false() {
        let sels = vec![leaf("edges"), leaf("pageInfo")];
        assert!(!selections_contain_field(&sels, "totalCount"));
    }

    #[test]
    fn inline_fragment_nested_match_returns_true() {
        // "...on UserConnection" wrapping totalCount
        let inline = fragment("...on UserConnection", vec![leaf("totalCount"), leaf("edges")]);
        let sels = vec![inline];
        assert!(selections_contain_field(&sels, "totalCount"));
    }

    #[test]
    fn inline_fragment_does_not_spuriously_match_fragment_name() {
        // The fragment entry (name "...on Foo") only matches a field named exactly "...on Foo"
        // when searched directly; it should NOT match an unrelated field name.
        let inline = fragment("...on Foo", vec![leaf("id")]);
        let sels = vec![inline];
        assert!(!selections_contain_field(&sels, "totalCount"));
        // "id" is nested inside the fragment and should be found via recursion
        assert!(selections_contain_field(&sels, "id"));
    }

    #[test]
    fn field_not_in_fragment_returns_false() {
        let inline = fragment("...on UserConnection", vec![leaf("edges"), leaf("pageInfo")]);
        let sels = vec![inline];
        assert!(!selections_contain_field(&sels, "totalCount"));
    }

    #[test]
    fn non_fragment_nested_field_not_searched() {
        // Only entries whose name starts with "..." trigger recursion.
        // A plain field's nested_fields should NOT be recursed into.
        let nested_count = fragment("edges", vec![leaf("totalCount")]);
        let sels = vec![nested_count];
        // "edges" doesn't start with "..." — nested fields not searched
        assert!(!selections_contain_field(&sels, "totalCount"));
    }

    #[test]
    fn multiple_fragments_any_can_match() {
        let frag1 = fragment("...on TypeA", vec![leaf("id")]);
        let frag2 = fragment("...on TypeB", vec![leaf("totalCount")]);
        let sels = vec![frag1, frag2];
        assert!(selections_contain_field(&sels, "totalCount"));
        assert!(selections_contain_field(&sels, "id"));
        assert!(!selections_contain_field(&sels, "name"));
    }

    #[test]
    fn mixed_direct_and_fragment_selections() {
        let inline = fragment("...on Connection", vec![leaf("pageInfo")]);
        let sels = vec![leaf("edges"), inline, leaf("metadata")];
        assert!(selections_contain_field(&sels, "edges"));
        assert!(selections_contain_field(&sels, "pageInfo"));
        assert!(selections_contain_field(&sels, "metadata"));
        assert!(!selections_contain_field(&sels, "cursor"));
    }
}