1use std::sync::Arc;
4
5#[cfg(feature = "arrow")]
6use fraiseql_arrow::FraiseQLFlightService;
7use fraiseql_core::{
8 cache::{CacheConfig, CachedDatabaseAdapter, QueryResultCache},
9 db::traits::DatabaseAdapter,
10 runtime::{Executor, RuntimeConfig, SubscriptionManager},
11 schema::CompiledSchema,
12 security::{AuthConfig, AuthMiddleware, OidcValidator},
13};
14use tracing::{info, warn};
15
16use super::{RateLimiter, Result, Server, ServerConfig, ServerError};
17
18pub(super) fn build_hs256_auth(config: &ServerConfig) -> Result<Option<Arc<AuthMiddleware>>> {
20 let Some(ref hs) = config.auth_hs256 else {
21 return Ok(None);
22 };
23 hs.validate()
28 .map_err(|e| ServerError::ConfigError(format!("Failed to initialize HS256 auth: {e}")))?;
29 let secret = hs
30 .load_secret()
31 .map_err(|e| ServerError::ConfigError(format!("Failed to initialize HS256 auth: {e}")))?;
32 let mut auth_config = AuthConfig::with_hs256(&secret);
33 if let Some(ref iss) = hs.issuer {
34 auth_config = auth_config.with_issuer(iss);
35 }
36 if let Some(ref aud) = hs.audience {
37 auth_config = auth_config.with_audience(aud);
38 }
39 info!(
40 secret_env = %hs.secret_env,
41 issuer = ?hs.issuer,
42 audience = ?hs.audience,
43 "Initializing HS256 authentication (local validation, no network)"
44 );
45 Ok(Some(Arc::new(AuthMiddleware::from_config(auth_config))))
46}
47
48impl<A: DatabaseAdapter + Clone + Send + Sync + 'static> Server<CachedDatabaseAdapter<A>> {
49 #[allow(clippy::cognitive_complexity)] pub async fn new(
87 config: ServerConfig,
88 schema: CompiledSchema,
89 adapter: Arc<A>,
90 db_pool: Option<sqlx::PgPool>,
91 ) -> Result<Self> {
92 if schema.schema_format_version.is_none() {
95 warn!(
96 "Loaded schema has no schema_format_version (pre-v2.1 format). \
97 Re-compile with the current fraiseql-cli for version compatibility checking."
98 );
99 }
100 schema.validate_format_version().map_err(|msg| {
101 ServerError::ConfigError(format!("Incompatible compiled schema: {msg}"))
102 })?;
103
104 #[cfg(feature = "federation")]
106 let circuit_breaker = schema.federation.as_ref().and_then(
107 crate::federation::circuit_breaker::FederationCircuitBreakerManager::from_config,
108 );
109 #[cfg(not(feature = "federation"))]
110 let circuit_breaker: Option<()> = None;
111 #[cfg(not(feature = "federation"))]
112 let _ = &schema.federation;
113 let error_sanitizer = Self::error_sanitizer_from_schema(&schema);
114 #[cfg(feature = "auth")]
115 let state_encryption = Self::state_encryption_from_schema(&schema)?;
116 #[cfg(not(feature = "auth"))]
117 let state_encryption: Option<
118 std::sync::Arc<crate::auth::state_encryption::StateEncryptionService>,
119 > = None;
120 #[cfg(feature = "auth")]
121 let pkce_store = Self::pkce_store_from_schema(&schema, state_encryption.as_ref()).await;
122 #[cfg(not(feature = "auth"))]
123 let pkce_store: Option<std::sync::Arc<crate::auth::PkceStateStore>> = None;
124 #[cfg(feature = "auth")]
125 let oidc_server_client = Self::oidc_server_client_from_schema(&schema);
126 #[cfg(not(feature = "auth"))]
127 let oidc_server_client: Option<std::sync::Arc<crate::auth::OidcServerClient>> = None;
128 let schema_rate_limiter = Self::rate_limiter_from_schema(&schema).await;
129 let api_key_authenticator = crate::api_key::api_key_authenticator_from_schema(&schema);
130 if api_key_authenticator.is_some() {
131 info!("API key authentication enabled");
132 }
133 let revocation_manager = crate::token_revocation::revocation_manager_from_schema(&schema);
134 if revocation_manager.is_some() {
135 info!("Token revocation enabled");
136 }
137 let mut tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
140 let trusted_docs = Self::trusted_docs_from_schema(&schema, &mut tasks);
141
142 if config.cache_enabled && !schema.has_rls_configured() {
147 if schema.is_multi_tenant() {
148 return Err(ServerError::ConfigError(
150 "Cache is enabled in a multi-tenant schema but no Row-Level Security \
151 policies are declared. This would allow cross-tenant cache hits and \
152 data leakage. In fraiseql.toml, either disable caching with \
153 [cache] enabled = false, declare [security.rls] policies, or set \
154 [security] multi_tenant = false to acknowledge single-tenant mode."
155 .to_string(),
156 ));
157 }
158 warn!(
160 "Query-result caching is enabled but no Row-Level Security policies are \
161 declared in the compiled schema. This is safe for single-tenant deployments. \
162 For multi-tenant deployments, declare RLS policies and set \
163 `security.multi_tenant = true` in your schema."
164 );
165 }
166
167 let cache_config = CacheConfig::from(config.cache_enabled);
169 let cache = QueryResultCache::new(cache_config);
170
171 if cache_config.enabled {
173 tracing::info!(
174 max_entries = cache_config.max_entries,
175 ttl_seconds = cache_config.ttl_seconds,
176 rls_enforcement = ?cache_config.rls_enforcement,
177 "Query result cache: active"
178 );
179 } else {
180 tracing::info!("Query result cache: disabled");
181 }
182
183 let subscriptions_config = schema.subscriptions_config.clone();
185
186 let inner = Arc::into_inner(adapter)
188 .expect("CachedDatabaseAdapter wrapping requires exclusive Arc ownership at startup");
189 let cached = CachedDatabaseAdapter::new(inner, cache, schema.content_hash())
190 .with_ttl_overrides_from_schema(&schema)
191 .with_rls(schema.has_rls_configured());
192
193 let audit_mutations = schema
196 .security
197 .as_ref()
198 .and_then(|s| s.additional.get("enterprise"))
199 .and_then(|e| e.get("audit_logging_enabled"))
200 .and_then(|v| v.as_bool())
201 .unwrap_or(false);
202 if audit_mutations {
203 info!("Mutation audit logging enabled (target: fraiseql::mutation_audit)");
204 }
205 let executor_config = RuntimeConfig {
206 audit_mutations,
207 ..RuntimeConfig::default()
208 };
209 let executor =
210 Arc::new(Executor::with_config(schema.clone(), Arc::new(cached), executor_config));
211 let subscription_manager = Arc::new(SubscriptionManager::new(Arc::new(schema)));
212
213 let mut server = Self::from_executor(
214 config,
215 executor,
216 subscription_manager,
217 circuit_breaker,
218 error_sanitizer,
219 state_encryption,
220 pkce_store,
221 oidc_server_client,
222 schema_rate_limiter,
223 api_key_authenticator,
224 revocation_manager,
225 trusted_docs,
226 db_pool,
227 tasks,
228 )
229 .await?;
230
231 server.adapter_cache_enabled = cache_config.enabled;
232
233 if let Some(pt) = server.config.pool_tuning.clone() {
235 if pt.enabled {
236 server = server
237 .with_pool_tuning(pt)
238 .map_err(|e| ServerError::ConfigError(format!("pool_tuning: {e}")))?;
239 }
240 }
241
242 #[cfg(feature = "mcp")]
244 if let Some(ref cfg) = server.executor.schema().mcp_config {
245 if cfg.enabled {
246 let tool_count =
247 crate::mcp::tools::schema_to_tools(server.executor.schema(), cfg).len();
248 info!(
249 path = %cfg.path,
250 transport = %cfg.transport,
251 tools = tool_count,
252 "MCP server configured"
253 );
254 server.mcp_config = Some(cfg.clone());
255 }
256 }
257
258 if server.config.apq_enabled {
260 let apq_store: fraiseql_core::apq::ArcApqStorage =
261 Arc::new(fraiseql_core::apq::InMemoryApqStorage::default());
262 server.apq_store = Some(apq_store);
263 info!("APQ (Automatic Persisted Queries) enabled — in-memory backend");
264 }
265
266 if let Some(ref subs) = subscriptions_config {
268 if let Some(max) = subs.max_subscriptions_per_connection {
269 server.max_subscriptions_per_connection = Some(max);
270 }
271 if let Some(lifecycle) = crate::subscriptions::WebhookLifecycle::from_config(subs) {
272 server.subscription_lifecycle = Arc::new(lifecycle);
273 }
274 }
275
276 Ok(server)
277 }
278}
279
280impl<A: DatabaseAdapter + Clone + Send + Sync + 'static> Server<A> {
281 #[allow(clippy::too_many_arguments)]
286 #[allow(clippy::cognitive_complexity)] pub(super) async fn from_executor(
290 config: ServerConfig,
291 executor: Arc<Executor<A>>,
292 subscription_manager: Arc<SubscriptionManager>,
293 #[cfg(feature = "federation")] circuit_breaker: Option<
294 Arc<crate::federation::circuit_breaker::FederationCircuitBreakerManager>,
295 >,
296 #[cfg(not(feature = "federation"))] _circuit_breaker: Option<()>,
297 error_sanitizer: Arc<crate::config::error_sanitization::ErrorSanitizer>,
298 state_encryption: Option<Arc<crate::auth::state_encryption::StateEncryptionService>>,
299 pkce_store: Option<Arc<crate::auth::PkceStateStore>>,
300 oidc_server_client: Option<Arc<crate::auth::OidcServerClient>>,
301 schema_rate_limiter: Option<Arc<RateLimiter>>,
302 api_key_authenticator: Option<Arc<crate::api_key::ApiKeyAuthenticator>>,
303 revocation_manager: Option<Arc<crate::token_revocation::TokenRevocationManager>>,
304 trusted_docs: Option<Arc<crate::trusted_documents::TrustedDocumentStore>>,
305 #[cfg_attr(
307 not(any(feature = "observers", feature = "auth")),
308 allow(unused_variables)
309 )]
310 db_pool: Option<sqlx::PgPool>,
311 mut tasks: tokio::task::JoinSet<()>,
312 ) -> Result<Self> {
313 let oidc_validator = if let Some(ref auth_config) = config.auth {
315 info!(
316 issuer = %auth_config.issuer,
317 "Initializing OIDC authentication"
318 );
319 let validator = OidcValidator::new(auth_config.clone())
320 .await
321 .map_err(|e| ServerError::ConfigError(format!("Failed to initialize OIDC: {e}")))?;
322 Some(Arc::new(validator))
323 } else {
324 None
325 };
326
327 let hs256_auth = build_hs256_auth(&config)?;
329
330 let rate_limiter = if let Some(rl) = schema_rate_limiter {
332 Some(rl)
333 } else if let Some(ref rate_config) = config.rate_limiting {
334 if rate_config.enabled {
335 info!(
336 rps_per_ip = rate_config.rps_per_ip,
337 rps_per_user = rate_config.rps_per_user,
338 "Initializing rate limiting from server config"
339 );
340 Some(Arc::new(RateLimiter::new(rate_config.clone())))
341 } else {
342 info!("Rate limiting disabled by configuration");
343 None
344 }
345 } else {
346 None
347 };
348
349 #[cfg(feature = "observers")]
351 let observer_runtime = Self::init_observer_runtime(&config, db_pool.as_ref()).await;
352
353 #[cfg(feature = "arrow")]
355 let flight_service = {
356 let mut service = FraiseQLFlightService::new();
357 if let Some(ref validator) = oidc_validator {
358 info!("Enabling OIDC authentication for Arrow Flight");
359 service.set_oidc_validator(validator.clone());
360 } else {
361 info!("Arrow Flight initialized without authentication (dev mode)");
362 }
363 Some(service)
364 };
365
366 #[cfg(feature = "auth")]
368 if pkce_store.is_some() && oidc_server_client.is_none() {
369 tracing::error!(
370 "pkce.enabled = true but [auth] is not configured or OIDC client init failed. \
371 Auth routes (/auth/start, /auth/callback) will NOT be mounted. \
372 Add [auth] with discovery_url, client_id, client_secret_env, and \
373 server_redirect_uri to fraiseql.toml and recompile the schema."
374 );
375 }
376
377 #[cfg(feature = "auth")]
379 Self::check_redis_requirement(pkce_store.as_ref())?;
380
381 #[cfg(feature = "auth")]
383 Self::spawn_pkce_cleanup(pkce_store.as_ref(), &mut tasks);
384
385 #[cfg(not(feature = "auth"))]
388 let _ = (state_encryption, pkce_store, oidc_server_client);
389 Ok(Self {
390 config,
391 executor,
392 subscription_manager,
393 subscription_lifecycle: Arc::new(crate::subscriptions::NoopLifecycle),
394 max_subscriptions_per_connection: None,
395 oidc_validator,
396 hs256_auth,
397 rate_limiter,
398 #[cfg(feature = "secrets")]
399 secrets_manager: None,
400 #[cfg(feature = "federation")]
401 circuit_breaker,
402 error_sanitizer,
403 #[cfg(feature = "auth")]
404 state_encryption,
405 #[cfg(feature = "auth")]
406 pkce_store,
407 #[cfg(feature = "auth")]
408 oidc_server_client,
409 #[cfg(feature = "auth")]
410 social_login: None,
411 #[cfg(feature = "auth")]
412 mfa_state: None,
413 #[cfg(feature = "auth")]
414 anon_signup_state: None,
415 api_key_authenticator,
416 revocation_manager,
417 apq_store: None,
418 trusted_docs,
419 #[cfg(feature = "observers")]
420 observer_runtime,
421 #[cfg(feature = "auth")]
422 enrichment_pool: db_pool.clone(),
423 #[cfg(feature = "observers")]
424 db_pool,
425 storage_state: None,
426 realtime_state: None,
427 #[cfg(feature = "arrow")]
428 flight_service,
429 #[cfg(feature = "mcp")]
430 mcp_config: None,
431 pool_tuning_config: None,
432 adapter_cache_enabled: false,
433 broadcast_manager: None,
434 presence_manager: None,
435 storage_backend: None,
436 storage_max_upload_bytes: 100 * 1024 * 1024, #[cfg(feature = "functions")]
438 function_store: None,
439 #[cfg(feature = "functions")]
440 function_runtime: None,
441 usage: Arc::clone(crate::usage::aggregator::global_aggregator()),
442 tasks,
443 })
444 }
445
446 #[cfg(feature = "auth")]
451 pub(super) fn spawn_pkce_cleanup(
452 pkce_store: Option<&Arc<crate::auth::PkceStateStore>>,
453 tasks: &mut tokio::task::JoinSet<()>,
454 ) {
455 use std::time::Duration;
456
457 use tokio::time::MissedTickBehavior;
458
459 if let Some(store) = pkce_store {
460 let store_clone = Arc::clone(store);
461 tasks.spawn(async move {
462 let mut ticker = tokio::time::interval(Duration::from_secs(300));
463 ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
464 loop {
465 ticker.tick().await;
466 store_clone.cleanup_expired().await;
467 }
468 });
469 }
470 }
471
472 #[must_use]
474 pub fn with_subscription_lifecycle(
475 mut self,
476 lifecycle: Arc<dyn crate::subscriptions::SubscriptionLifecycle>,
477 ) -> Self {
478 self.subscription_lifecycle = lifecycle;
479 self
480 }
481
482 #[must_use]
484 pub const fn with_max_subscriptions_per_connection(mut self, max: u32) -> Self {
485 self.max_subscriptions_per_connection = Some(max);
486 self
487 }
488
489 #[must_use]
495 pub fn with_realtime(mut self, state: crate::realtime::server::RealtimeState) -> Self {
496 self.realtime_state = Some(state);
497 self
498 }
499
500 #[must_use]
502 pub fn with_broadcast(mut self, config: crate::subscriptions::BroadcastConfig) -> Self {
503 self.broadcast_manager =
504 Some(Arc::new(crate::subscriptions::BroadcastManager::new(config)));
505 self
506 }
507
508 #[must_use]
510 pub fn with_presence(mut self, config: crate::subscriptions::PresenceConfig) -> Self {
511 self.presence_manager = Some(Arc::new(crate::subscriptions::PresenceManager::new(config)));
512 self
513 }
514
515 pub fn with_pool_tuning(
524 mut self,
525 config: crate::config::pool_tuning::PoolPressureMonitorConfig,
526 ) -> std::result::Result<Self, String> {
527 config.validate()?;
528 self.pool_tuning_config = Some(config);
529 Ok(self)
530 }
531
532 #[cfg(feature = "auth")]
546 #[must_use]
547 pub fn with_social_login(
548 mut self,
549 social_login: Arc<crate::auth::social::SocialLoginState>,
550 ) -> Self {
551 self.social_login = Some(social_login);
552 self
553 }
554
555 #[cfg(feature = "auth")]
561 #[must_use]
562 pub fn with_anon_signup(mut self, state: Arc<crate::auth::AnonSignupState>) -> Self {
563 self.anon_signup_state = Some(state);
564 self
565 }
566
567 #[cfg(feature = "auth")]
576 #[must_use]
577 pub fn with_mfa(mut self, mfa_state: Arc<crate::auth::MfaRouteState>) -> Self {
578 self.mfa_state = Some(mfa_state);
579 self
580 }
581
582 #[must_use]
591 pub fn with_storage(mut self, backend: Arc<dyn crate::storage::StorageBackend>) -> Self {
592 self.storage_backend = Some(backend);
593 self
594 }
595
596 #[must_use]
601 pub const fn with_storage_max_upload_bytes(mut self, bytes: usize) -> Self {
602 self.storage_max_upload_bytes = bytes;
603 self
604 }
605
606 #[cfg(feature = "functions")]
612 #[must_use]
613 pub fn with_functions(
614 mut self,
615 store: Arc<dyn fraiseql_functions::FunctionStore>,
616 runtime: Arc<dyn fraiseql_functions::runtime::SendFunctionRuntime>,
617 ) -> Self {
618 self.function_store = Some(store);
619 self.function_runtime = Some(runtime);
620 self
621 }
622
623 #[cfg(feature = "secrets")]
627 pub fn set_secrets_manager(&mut self, manager: Arc<crate::secrets_manager::SecretsManager>) {
628 self.secrets_manager = Some(manager);
629 info!("Secrets manager attached to server");
630 }
631
632 #[cfg(feature = "mcp")]
642 pub async fn serve_mcp_stdio(self) -> Result<()> {
643 use rmcp::ServiceExt;
644
645 let mcp_cfg = self.mcp_config.ok_or_else(|| {
646 ServerError::ConfigError(
647 "FRAISEQL_MCP_STDIO=1 but MCP is not configured. \
648 Add [mcp] enabled = true to fraiseql.toml and recompile the schema."
649 .into(),
650 )
651 })?;
652
653 let schema = Arc::new(self.executor.schema().clone());
654 let executor = self.executor.clone();
655
656 let service = crate::mcp::handler::FraiseQLMcpService::new(schema, executor, mcp_cfg)
657 .with_oidc_validator(self.oidc_validator.clone());
658
659 info!("MCP stdio transport starting — reading from stdin, writing to stdout");
660
661 let running = service
662 .serve((tokio::io::stdin(), tokio::io::stdout()))
663 .await
664 .map_err(|e| ServerError::ConfigError(format!("MCP stdio init failed: {e}")))?;
665
666 running
667 .waiting()
668 .await
669 .map_err(|e| ServerError::ConfigError(format!("MCP stdio error: {e}")))?;
670
671 Ok(())
672 }
673}