fraiseql_server/server/

builder.rs

//! Server constructors and builder methods.

use std::sync::Arc;

#[cfg(feature = "arrow")]
use fraiseql_arrow::FraiseQLFlightService;
use fraiseql_core::{
    cache::{CacheConfig, CachedDatabaseAdapter, QueryResultCache},
    db::traits::DatabaseAdapter,
    runtime::{Executor, RuntimeConfig, SubscriptionManager},
    schema::CompiledSchema,
    security::{AuthConfig, AuthMiddleware, OidcValidator},
};
use tracing::{info, warn};

use super::{RateLimiter, Result, Server, ServerConfig, ServerError};

/// Build an HS256 validator from the server config, if configured.
pub(super) fn build_hs256_auth(config: &ServerConfig) -> Result<Option<Arc<AuthMiddleware>>> {
    let Some(ref hs) = config.auth_hs256 else {
        return Ok(None);
    };
    // Reject incompatible config shapes *before* loading the secret, so a
    // dev environment with a real secret env-var but missing audience still
    // surfaces the audience error rather than silently booting.  Closes the
    // cross-service token-confusion gap for the HS256 path (#359).
    hs.validate()
        .map_err(|e| ServerError::ConfigError(format!("Failed to initialize HS256 auth: {e}")))?;
    let secret = hs
        .load_secret()
        .map_err(|e| ServerError::ConfigError(format!("Failed to initialize HS256 auth: {e}")))?;
    let mut auth_config = AuthConfig::with_hs256(&secret);
    if let Some(ref iss) = hs.issuer {
        auth_config = auth_config.with_issuer(iss);
    }
    if let Some(ref aud) = hs.audience {
        auth_config = auth_config.with_audience(aud);
    }
    info!(
        secret_env = %hs.secret_env,
        issuer = ?hs.issuer,
        audience = ?hs.audience,
        "Initializing HS256 authentication (local validation, no network)"
    );
    Ok(Some(Arc::new(AuthMiddleware::from_config(auth_config))))
}

impl<A: DatabaseAdapter + Clone + Send + Sync + 'static> Server<CachedDatabaseAdapter<A>> {
    /// Create new server.
    ///
    /// Relay pagination queries will return a `Validation` error at runtime. Use
    /// [`Server::with_relay_pagination`] when the adapter implements
    /// [`RelayDatabaseAdapter`](fraiseql_core::db::traits::RelayDatabaseAdapter)
    /// and relay support is required.
    ///
    /// # Arguments
    ///
    /// * `config` - Server configuration
    /// * `schema` - Compiled GraphQL schema
    /// * `adapter` - Database adapter
    /// * `db_pool` — forwarded to the observer runtime; `None` when observers are disabled.
    ///
    /// # Errors
    ///
    /// Returns error if OIDC validator initialization fails (e.g., unable to
    /// fetch discovery document or JWKS).
    ///
    /// # Panics
    ///
    /// Panics if the `adapter` `Arc` has been cloned before calling this constructor
    /// (refcount > 1). The builder must have exclusive ownership to unwrap the adapter
    /// for `CachedDatabaseAdapter` construction.
    ///
    /// # Example
    ///
    /// ```text
    /// // Requires: running PostgreSQL database and compiled schema file.
    /// let config = ServerConfig::default();
    /// let schema = CompiledSchema::from_json(schema_json, false)?;
    /// let adapter = Arc::new(PostgresAdapter::new(db_url).await?);
    ///
    /// let server = Server::new(config, schema, adapter, None).await?;
    /// server.serve().await?;
    /// ```
    #[allow(clippy::cognitive_complexity)] // Reason: server construction with subsystem initialization (auth, rate-limit, observers, etc.)
    pub async fn new(
        config: ServerConfig,
        schema: CompiledSchema,
        adapter: Arc<A>,
        db_pool: Option<sqlx::PgPool>,
    ) -> Result<Self> {
        // Validate compiled schema format version before any further setup.
        // Warns for legacy schemas (no version field); rejects incompatible future versions.
        if schema.schema_format_version.is_none() {
            warn!(
                "Loaded schema has no schema_format_version (pre-v2.1 format). \
                 Re-compile with the current fraiseql-cli for version compatibility checking."
            );
        }
        schema.validate_format_version().map_err(|msg| {
            ServerError::ConfigError(format!("Incompatible compiled schema: {msg}"))
        })?;

        // Read security configs from compiled schema BEFORE schema is moved.
        #[cfg(feature = "federation")]
        let circuit_breaker = schema.federation.as_ref().and_then(
            crate::federation::circuit_breaker::FederationCircuitBreakerManager::from_config,
        );
        #[cfg(not(feature = "federation"))]
        let circuit_breaker: Option<()> = None;
        #[cfg(not(feature = "federation"))]
        let _ = &schema.federation;
        let error_sanitizer = Self::error_sanitizer_from_schema(&schema);
        #[cfg(feature = "auth")]
        let state_encryption = Self::state_encryption_from_schema(&schema)?;
        #[cfg(not(feature = "auth"))]
        let state_encryption: Option<
            std::sync::Arc<crate::auth::state_encryption::StateEncryptionService>,
        > = None;
        #[cfg(feature = "auth")]
        let pkce_store = Self::pkce_store_from_schema(&schema, state_encryption.as_ref()).await?;
        #[cfg(not(feature = "auth"))]
        let pkce_store: Option<std::sync::Arc<crate::auth::PkceStateStore>> = None;
        #[cfg(feature = "auth")]
        let oidc_server_client = Self::oidc_server_client_from_schema(&schema);
        #[cfg(not(feature = "auth"))]
        let oidc_server_client: Option<std::sync::Arc<crate::auth::OidcServerClient>> = None;
        let schema_rate_limiter = Self::rate_limiter_from_schema(&schema).await?;
        let api_key_authenticator = crate::api_key::api_key_authenticator_from_schema(&schema);
        if api_key_authenticator.is_some() {
            info!("API key authentication enabled");
        }
        let revocation_manager = crate::token_revocation::revocation_manager_from_schema(&schema)?;
        if revocation_manager.is_some() {
            info!("Token revocation enabled");
        }
        // Collect lifecycle task handles into a JoinSet that will be moved onto
        // the `Server` so graceful shutdown can await them.
        let mut tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
        let trusted_docs = Self::trusted_docs_from_schema(&schema, &mut tasks);

        // Validate cache + RLS safety at startup.
        // Cache isolation relies entirely on per-user WHERE clauses in the cache key.
        // Without RLS, users with the same query and variables share the same cached
        // response, which can leak data across tenants.
        if config.cache_enabled && !schema.has_rls_configured() {
            if schema.is_multi_tenant() {
                // Multi-tenant + cache + no RLS is a hard safety violation.
                return Err(ServerError::ConfigError(
                    "Cache is enabled in a multi-tenant schema but no Row-Level Security \
                     policies are declared. This would allow cross-tenant cache hits and \
                     data leakage. In fraiseql.toml, either disable caching with \
                     [cache] enabled = false, declare [security.rls] policies, or set \
                     [security] multi_tenant = false to acknowledge single-tenant mode."
                        .to_string(),
                ));
            }
            // Single-tenant with cache and no RLS: safe, but warn in case of misconfiguration.
            warn!(
                "Query-result caching is enabled but no Row-Level Security policies are \
                 declared in the compiled schema. This is safe for single-tenant deployments. \
                 For multi-tenant deployments, declare RLS policies and set \
                 `security.multi_tenant = true` in your schema."
            );
        }

        // Build cache from config.
        let cache_config = CacheConfig::from(config.cache_enabled);
        let cache = QueryResultCache::new(cache_config);

        // Log cache state before consuming config.
        if cache_config.enabled {
            tracing::info!(
                max_entries   = cache_config.max_entries,
                ttl_seconds   = cache_config.ttl_seconds,
                rls_enforcement = ?cache_config.rls_enforcement,
                "Query result cache: active"
            );
        } else {
            tracing::info!("Query result cache: disabled");
        }

        // Read subscription config from compiled schema (hooks, limits).
        let subscriptions_config = schema.subscriptions_config.clone();

        // Unwrap Arc: refcount is 1 here — adapter has not been cloned since being passed in.
        let inner = Arc::into_inner(adapter)
            .expect("CachedDatabaseAdapter wrapping requires exclusive Arc ownership at startup");
        let cached = CachedDatabaseAdapter::new(inner, cache, schema.content_hash())
            .with_ttl_overrides_from_schema(&schema)
            .with_rls(schema.has_rls_configured());

        // Read audit_logging_enabled from compiled schema's enterprise security config.
        // Path: security.additional["enterprise"]["audit_logging_enabled"]
        let audit_mutations = schema
            .security
            .as_ref()
            .and_then(|s| s.additional.get("enterprise"))
            .and_then(|e| e.get("audit_logging_enabled"))
            .and_then(|v| v.as_bool())
            .unwrap_or(false);
        if audit_mutations {
            info!("Mutation audit logging enabled (target: fraiseql::mutation_audit)");
        }
        // #421: top-level page-size ceiling — FRAISEQL_MAX_PAGE_SIZE overrides the
        // compiled `[validation] max_page_size`, which overrides the default (1000).
        let max_page_size = page_size_precedence(
            std::env::var("FRAISEQL_MAX_PAGE_SIZE").ok().as_deref(),
            schema.validation_config.as_ref().and_then(|v| v.max_page_size),
        );
        // Change-Spine outbox write toggle (default on): FRAISEQL_CHANGELOG_ENABLED
        // overrides the compiled `[changelog] write_enabled`. Composes (AND) with
        // each mutation's per-endpoint `changelog` flag in the runner.
        let changelog_enabled = std::env::var("FRAISEQL_CHANGELOG_ENABLED")
            .ok()
            .map(|v| {
                !matches!(v.trim().to_ascii_lowercase().as_str(), "false" | "0" | "no" | "off")
            })
            .or_else(|| schema.changelog.as_ref().map(|c| c.write_enabled))
            .unwrap_or(true);
        if !changelog_enabled {
            info!(
                "Change-log outbox write disabled (FRAISEQL_CHANGELOG_ENABLED / [changelog] write_enabled)"
            );
        }
        let executor_config = RuntimeConfig {
            audit_mutations,
            max_page_size,
            changelog_enabled,
            ..RuntimeConfig::default()
        };
        let executor =
            Arc::new(Executor::with_config(schema.clone(), Arc::new(cached), executor_config));
        let subscription_manager = Arc::new(SubscriptionManager::new(Arc::new(schema)));

        let mut server = Self::from_executor(
            config,
            executor,
            subscription_manager,
            circuit_breaker,
            error_sanitizer,
            state_encryption,
            pkce_store,
            oidc_server_client,
            schema_rate_limiter,
            api_key_authenticator,
            revocation_manager,
            trusted_docs,
            db_pool,
            tasks,
        )
        .await?;

        server.adapter_cache_enabled = cache_config.enabled;

        // Apply pool tuning config from ServerConfig (if present).
        if let Some(pt) = server.config.pool_tuning.clone() {
            if pt.enabled {
                server = server
                    .with_pool_tuning(pt)
                    .map_err(|e| ServerError::ConfigError(format!("pool_tuning: {e}")))?;
            }
        }

        // Initialize MCP config from compiled schema when the feature is compiled in.
        #[cfg(feature = "mcp")]
        if let Some(ref cfg) = server.executor.schema().mcp_config {
            if cfg.enabled {
                let tool_count =
                    crate::mcp::tools::schema_to_tools(server.executor.schema(), cfg).len();
                info!(
                    path = %cfg.path,
                    transport = %cfg.transport,
                    tools = tool_count,
                    "MCP server configured"
                );
                server.mcp_config = Some(cfg.clone());
            }
        }

        // Initialize APQ store when enabled.
        if server.config.apq_enabled {
            let apq_store: fraiseql_core::apq::ArcApqStorage =
                Arc::new(fraiseql_core::apq::InMemoryApqStorage::default());
            server.apq_store = Some(apq_store);
            info!("APQ (Automatic Persisted Queries) enabled — in-memory backend");
        }

        // Apply subscription lifecycle/limits from compiled schema.
        if let Some(ref subs) = subscriptions_config {
            if let Some(max) = subs.max_subscriptions_per_connection {
                server.max_subscriptions_per_connection = Some(max);
            }
            if let Some(lifecycle) = crate::subscriptions::WebhookLifecycle::from_config(subs) {
                server.subscription_lifecycle = Arc::new(lifecycle);
            }
        }

        Ok(server)
    }
}

impl<A: DatabaseAdapter + Clone + Send + Sync + 'static> Server<A> {
    /// Shared initialization path used by both `new` and `with_relay_pagination`.
    ///
    /// Accepts a pre-built executor so that relay vs. non-relay constructors can supply
    /// the appropriate variant without duplicating auth/rate-limiter/observer setup.
    #[allow(clippy::too_many_arguments)]
    // Reason: internal constructor collects all pre-built subsystems; a builder struct would not
    // reduce call-site clarity
    #[allow(clippy::cognitive_complexity)] // Reason: internal constructor that assembles server from pre-built subsystems
    pub(super) async fn from_executor(
        config: ServerConfig,
        executor: Arc<Executor<A>>,
        subscription_manager: Arc<SubscriptionManager>,
        #[cfg(feature = "federation")] circuit_breaker: Option<
            Arc<crate::federation::circuit_breaker::FederationCircuitBreakerManager>,
        >,
        #[cfg(not(feature = "federation"))] _circuit_breaker: Option<()>,
        error_sanitizer: Arc<crate::config::error_sanitization::ErrorSanitizer>,
        state_encryption: Option<Arc<crate::auth::state_encryption::StateEncryptionService>>,
        pkce_store: Option<Arc<crate::auth::PkceStateStore>>,
        oidc_server_client: Option<Arc<crate::auth::OidcServerClient>>,
        schema_rate_limiter: Option<Arc<RateLimiter>>,
        api_key_authenticator: Option<Arc<crate::api_key::ApiKeyAuthenticator>>,
        revocation_manager: Option<Arc<crate::token_revocation::TokenRevocationManager>>,
        trusted_docs: Option<Arc<crate::trusted_documents::TrustedDocumentStore>>,
        // `db_pool` is forwarded to the observer runtime and/or auth enrichment.
        #[cfg_attr(
            not(any(feature = "observers", feature = "auth")),
            allow(unused_variables)
        )]
        db_pool: Option<sqlx::PgPool>,
        mut tasks: tokio::task::JoinSet<()>,
    ) -> Result<Self> {
        // Initialize OIDC validator if auth is configured
        let oidc_validator = if let Some(ref auth_config) = config.auth {
            info!(
                issuer = %auth_config.issuer,
                "Initializing OIDC authentication"
            );
            let validator = OidcValidator::new(auth_config.clone())
                .await
                .map_err(|e| ServerError::ConfigError(format!("Failed to initialize OIDC: {e}")))?;
            Some(Arc::new(validator))
        } else {
            None
        };

        // Initialize HS256 validator if configured (mutually exclusive with OIDC).
        let hs256_auth = build_hs256_auth(&config)?;

        // Initialize rate limiter: compiled schema config takes priority over server config.
        let rate_limiter = if let Some(rl) = schema_rate_limiter {
            Some(rl)
        } else if let Some(ref rate_config) = config.rate_limiting {
            if rate_config.enabled {
                info!(
                    rps_per_ip = rate_config.rps_per_ip,
                    rps_per_user = rate_config.rps_per_user,
                    "Initializing rate limiting from server config"
                );
                Some(Arc::new(RateLimiter::new(rate_config.clone())))
            } else {
                info!("Rate limiting disabled by configuration");
                None
            }
        } else {
            None
        };

        // Initialize observer runtime
        #[cfg(feature = "observers")]
        let observer_runtime = Self::init_observer_runtime(&config, db_pool.as_ref()).await?;

        // Initialize Flight service with OIDC authentication if configured
        #[cfg(feature = "arrow")]
        let flight_service = {
            let mut service = FraiseQLFlightService::new();
            if let Some(ref validator) = oidc_validator {
                info!("Enabling OIDC authentication for Arrow Flight");
                service.set_oidc_validator(validator.clone());
            } else {
                info!("Arrow Flight initialized without authentication (dev mode)");
            }
            Some(service)
        };

        // Warn if PKCE is configured but [auth] is missing (no OidcServerClient).
        #[cfg(feature = "auth")]
        if pkce_store.is_some() && oidc_server_client.is_none() {
            tracing::error!(
                "pkce.enabled = true but [auth] is not configured or OIDC client init failed. \
                 Auth routes (/auth/start, /auth/callback) will NOT be mounted. \
                 Add [auth] with discovery_url, client_id, client_secret_env, and \
                 server_redirect_uri to fraiseql.toml and recompile the schema."
            );
        }

        // Refuse to start if FRAISEQL_REQUIRE_REDIS is set and PKCE store is in-memory.
        #[cfg(feature = "auth")]
        Self::check_redis_requirement(pkce_store.as_ref())?;

        // Spawn background PKCE state cleanup task (every 5 minutes).
        #[cfg(feature = "auth")]
        Self::spawn_pkce_cleanup(pkce_store.as_ref(), &mut tasks);

        // Reason: state_encryption/pkce_store/oidc_server_client are only stored when
        //         feature = "auth" is enabled; without it they are legitimately unused.
        #[cfg(not(feature = "auth"))]
        let _ = (state_encryption, pkce_store, oidc_server_client);
        Ok(Self {
            config,
            executor,
            subscription_manager,
            subscription_lifecycle: Arc::new(crate::subscriptions::NoopLifecycle),
            max_subscriptions_per_connection: None,
            oidc_validator,
            hs256_auth,
            rate_limiter,
            #[cfg(feature = "secrets")]
            secrets_manager: None,
            #[cfg(feature = "federation")]
            circuit_breaker,
            error_sanitizer,
            #[cfg(feature = "auth")]
            state_encryption,
            #[cfg(feature = "auth")]
            pkce_store,
            #[cfg(feature = "auth")]
            oidc_server_client,
            #[cfg(feature = "auth")]
            social_login: None,
            #[cfg(feature = "auth")]
            mfa_state: None,
            #[cfg(feature = "auth")]
            anon_signup_state: None,
            api_key_authenticator,
            revocation_manager,
            apq_store: None,
            trusted_docs,
            #[cfg(feature = "observers")]
            observer_runtime,
            #[cfg(feature = "auth")]
            enrichment_pool: db_pool.clone(),
            #[cfg(feature = "observers")]
            db_pool,
            storage_state: None,
            realtime_state: None,
            tenant_executor_factory: None,
            #[cfg(feature = "arrow")]
            flight_service,
            #[cfg(feature = "mcp")]
            mcp_config: None,
            pool_tuning_config: None,
            adapter_cache_enabled: false,
            broadcast_manager: None,
            presence_manager: None,
            storage_backend: None,
            storage_max_upload_bytes: 100 * 1024 * 1024, // 100 MiB default
            #[cfg(feature = "functions")]
            function_store: None,
            #[cfg(feature = "functions")]
            function_runtime: None,
            usage: Arc::clone(crate::usage::aggregator::global_aggregator()),
            tasks,
        })
    }

    /// Spawn the periodic PKCE state-cleanup task into the server's [`JoinSet`].
    ///
    /// Cleanup runs every 5 minutes for the lifetime of the server. The handle
    /// is owned by `tasks` so graceful shutdown awaits its termination.
    #[cfg(feature = "auth")]
    pub(super) fn spawn_pkce_cleanup(
        pkce_store: Option<&Arc<crate::auth::PkceStateStore>>,
        tasks: &mut tokio::task::JoinSet<()>,
    ) {
        use std::time::Duration;

        use tokio::time::MissedTickBehavior;

        if let Some(store) = pkce_store {
            let store_clone = Arc::clone(store);
            tasks.spawn(async move {
                let mut ticker = tokio::time::interval(Duration::from_secs(300));
                ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
                loop {
                    ticker.tick().await;
                    store_clone.cleanup_expired().await;
                }
            });
        }
    }

    /// Set lifecycle hooks for `WebSocket` subscriptions.
    #[must_use]
    pub fn with_subscription_lifecycle(
        mut self,
        lifecycle: Arc<dyn crate::subscriptions::SubscriptionLifecycle>,
    ) -> Self {
        self.subscription_lifecycle = lifecycle;
        self
    }

    /// Set maximum subscriptions allowed per `WebSocket` connection.
    #[must_use]
    pub const fn with_max_subscriptions_per_connection(mut self, max: u32) -> Self {
        self.max_subscriptions_per_connection = Some(max);
        self
    }

    /// Attach the per-tenant executor factory used to provision tenants at
    /// runtime (`PUT /api/v1/admin/tenants/{key}`).
    ///
    /// Build it with
    /// [`make_executor_factory`](crate::tenancy::make_executor_factory) from the
    /// concrete adapter type (which must implement
    /// [`FromPoolConfig`](crate::tenancy::FromPoolConfig)). When the multi-tenant
    /// runtime is enabled (`[tenancy.runtime] enabled = true`), `build_app_state`
    /// installs it into `AppState`. Call this before `serve`.
    #[must_use]
    pub fn with_tenant_executor_factory(
        mut self,
        factory: crate::tenancy::TenantExecutorFactory<A>,
    ) -> Self {
        self.tenant_executor_factory = Some(factory);
        self
    }

    /// Attach a pre-built realtime `WebSocket` state to the server.
    ///
    /// When set, `build_base_router` will merge `realtime_router(state)` at
    /// `/realtime/v1`.  Call this after constructing the server but before
    /// calling `serve` or `serve_with_shutdown`.
    #[must_use]
    pub fn with_realtime(mut self, state: crate::realtime::server::RealtimeState) -> Self {
        self.realtime_state = Some(state);
        self
    }

    /// Enable ephemeral broadcast channels (`POST /realtime/v1/broadcast`).
    #[must_use]
    pub fn with_broadcast(mut self, config: crate::subscriptions::BroadcastConfig) -> Self {
        self.broadcast_manager =
            Some(Arc::new(crate::subscriptions::BroadcastManager::new(config)));
        self
    }

    /// Enable room-based presence tracking.
    #[must_use]
    pub fn with_presence(mut self, config: crate::subscriptions::PresenceConfig) -> Self {
        self.presence_manager = Some(Arc::new(crate::subscriptions::PresenceManager::new(config)));
        self
    }

    /// Enable adaptive connection pool sizing.
    ///
    /// When `config.enabled` is `true`, the server will spawn a background
    /// polling task that samples pool metrics and recommends or applies resizes.
    ///
    /// # Errors
    ///
    /// Returns an error string if the configuration fails validation (e.g. `min >= max`).
    pub fn with_pool_tuning(
        mut self,
        config: crate::config::pool_tuning::PoolPressureMonitorConfig,
    ) -> std::result::Result<Self, String> {
        config.validate()?;
        self.pool_tuning_config = Some(config);
        Ok(self)
    }

    /// Attach a unified social-login provider registry.
    ///
    /// When set, the server mounts `GET /auth/v1/authorize` which redirects
    /// users to the specified `OAuth` provider's authorization URL with a `CSRF`
    /// state token.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use fraiseql_auth::social::{SocialLoginState, SocialProviderRegistry};
    /// let state = Arc::new(SocialLoginState { ... });
    /// let server = server.with_social_login(state);
    /// ```
    #[cfg(feature = "auth")]
    #[must_use]
    pub fn with_social_login(
        mut self,
        social_login: Arc<crate::auth::social::SocialLoginState>,
    ) -> Self {
        self.social_login = Some(social_login);
        self
    }

    /// Attach anonymous signup state to mount `POST /auth/v1/signup`.
    ///
    /// When set, any client can obtain a guest session without credentials.
    /// The returned `user_id` carries an `anon_` prefix and the session lasts
    /// 7 days.  Signups are rate-limited per client IP.
    #[cfg(feature = "auth")]
    #[must_use]
    pub fn with_anon_signup(mut self, state: Arc<crate::auth::AnonSignupState>) -> Self {
        self.anon_signup_state = Some(state);
        self
    }

    /// Attach `TOTP` `MFA` state to mount the `/auth/v1/mfa/` endpoints.
    ///
    /// Mounts four routes:
    /// - `POST /auth/v1/mfa/enroll` — begin enrollment, returns `otpauth://` `URI`
    /// - `POST /auth/v1/mfa/confirm` — confirm enrollment with a live `TOTP` code
    /// - `POST /auth/v1/mfa/challenge` — issue a short-lived challenge token
    /// - `POST /auth/v1/mfa/verify` — verify code and issue session
    /// - `POST /auth/v1/mfa/unenroll` — remove `MFA` from an account
    #[cfg(feature = "auth")]
    #[must_use]
    pub fn with_mfa(mut self, mfa_state: Arc<crate::auth::MfaRouteState>) -> Self {
        self.mfa_state = Some(mfa_state);
        self
    }

    /// Attach an object storage backend and mount `/storage/v1/` routes.
    ///
    /// When set, the server mounts `GET`, `POST`, `DELETE /storage/v1/object/*key`
    /// and `GET /storage/v1/object/sign/*key` endpoints backed by the given backend.
    ///
    /// Use [`StorageConfig`](crate::config::StorageConfig) and
    /// [`create_backend`](crate::storage::create_backend) to construct the backend
    /// from a TOML configuration block.
    #[must_use]
    pub fn with_storage(mut self, backend: Arc<dyn crate::storage::StorageBackend>) -> Self {
        self.storage_backend = Some(backend);
        self
    }

    /// Override the maximum allowed upload size for storage endpoints.
    ///
    /// Defaults to 100 `MiB`. Uploads exceeding this size are rejected with HTTP 413
    /// before the body is forwarded to the storage backend.
    #[must_use]
    pub const fn with_storage_max_upload_bytes(mut self, bytes: usize) -> Self {
        self.storage_max_upload_bytes = bytes;
        self
    }

    /// Attach a pre-built [`StorageState`](fraiseql_storage::StorageState) and
    /// mount the bucket-scoped `/storage/v1/*` routes (object upload/download/
    /// delete, list, presign) with per-bucket access policy and RLS.
    ///
    /// This is the full storage path (metadata-backed, bucket-aware); it differs
    /// from [`with_storage`](Self::with_storage), which mounts the simpler legacy
    /// object router. Authentication is applied by `mount_storage_state`: a
    /// configured `storage_token` acts as an admin bearer and, when an OIDC
    /// validator is present, per-user tokens populate the request's
    /// `StorageUser` for RLS.
    #[must_use]
    pub fn with_storage_state(mut self, state: fraiseql_storage::StorageState) -> Self {
        self.storage_state = Some(state);
        self
    }

    /// Install a pre-built token-revocation manager, replacing whatever the generic
    /// construction path produced.
    ///
    /// Used by the PostgreSQL runtime path to install the Postgres-backed store
    /// (#357): `revocation_manager_from_schema` defers the `postgres` backend because
    /// it needs a database connection, so `main.rs` builds it with
    /// `build_postgres_revocation_manager` and installs it here.
    #[must_use]
    pub fn with_revocation_manager(
        mut self,
        manager: Arc<crate::token_revocation::TokenRevocationManager>,
    ) -> Self {
        self.revocation_manager = Some(manager);
        self
    }

    /// Attach a function deployment store and runtime, mounting `/functions/v1/` routes.
    ///
    /// When set, the server mounts `POST /functions/v1/{name}` which loads the
    /// function bytecode from `store`, executes it via `runtime`, and returns the
    /// JSON-encoded [`FunctionResult`](fraiseql_functions::FunctionResult).
    #[cfg(feature = "functions")]
    #[must_use]
    pub fn with_functions(
        mut self,
        store: Arc<dyn fraiseql_functions::FunctionStore>,
        runtime: Arc<dyn fraiseql_functions::runtime::SendFunctionRuntime>,
    ) -> Self {
        self.function_store = Some(store);
        self.function_runtime = Some(runtime);
        self
    }

    /// Set secrets manager for the server.
    ///
    /// This allows attaching a secrets manager after server creation for credential management.
    #[cfg(feature = "secrets")]
    pub fn set_secrets_manager(&mut self, manager: Arc<crate::secrets_manager::SecretsManager>) {
        self.secrets_manager = Some(manager);
        info!("Secrets manager attached to server");
    }

    /// Serve MCP over stdio (stdin/stdout) instead of HTTP.
    ///
    /// This is used when `FRAISEQL_MCP_STDIO=1` is set.  The server reads JSON-RPC
    /// messages from stdin and writes responses to stdout, following the MCP stdio
    /// transport specification.
    ///
    /// # Errors
    ///
    /// Returns an error if MCP is not configured or the stdio transport fails.
    #[cfg(feature = "mcp")]
    pub async fn serve_mcp_stdio(self) -> Result<()> {
        use rmcp::ServiceExt;

        let mcp_cfg = self.mcp_config.ok_or_else(|| {
            ServerError::ConfigError(
                "FRAISEQL_MCP_STDIO=1 but MCP is not configured. \
                 Add [mcp] enabled = true to fraiseql.toml and recompile the schema."
                    .into(),
            )
        })?;

        let schema = Arc::new(self.executor.schema().clone());
        let executor = self.executor.clone();

        let service = crate::mcp::handler::FraiseQLMcpService::new(schema, executor, mcp_cfg)
            .with_oidc_validator(self.oidc_validator.clone());

        info!("MCP stdio transport starting — reading from stdin, writing to stdout");

        let running = service
            .serve((tokio::io::stdin(), tokio::io::stdout()))
            .await
            .map_err(|e| ServerError::ConfigError(format!("MCP stdio init failed: {e}")))?;

        running
            .waiting()
            .await
            .map_err(|e| ServerError::ConfigError(format!("MCP stdio error: {e}")))?;

        Ok(())
    }
}

/// Resolve the top-level page-size ceiling (#421) from its precedence chain.
///
/// `env` is the raw `FRAISEQL_MAX_PAGE_SIZE` value when set (a positive integer,
/// or `"0"`/`"none"` to disable the ceiling). It overrides `compiled` (the
/// `[validation] max_page_size` from the compiled schema), which overrides the
/// runtime default (1000). Returns `None` only when explicitly disabled.
pub fn page_size_precedence(env: Option<&str>, compiled: Option<u32>) -> Option<u32> {
    if let Some(raw) = env {
        let trimmed = raw.trim();
        if trimmed.eq_ignore_ascii_case("none") || trimmed == "0" {
            return None;
        }
        if let Ok(n) = trimmed.parse::<u32>() {
            return Some(n);
        }
        // Unparseable env value: ignore it and fall through to compiled/default.
    }
    compiled.or(RuntimeConfig::default().max_page_size)
}