fraiseql-server 2.16.0

//! Server extensions: relay pagination, Arrow Flight service, and observer runtime
//! initialization.

use std::sync::Arc;

#[cfg(feature = "arrow")]
use fraiseql_arrow::FraiseQLFlightService;
use fraiseql_core::{
    db::traits::{RelayDatabaseAdapter, Writer},
    runtime::{Executor, SubscriptionManager},
    schema::CompiledSchema,
};
#[cfg(feature = "observers")]
use tokio::sync::RwLock;
#[cfg(any(feature = "observers", feature = "mcp"))]
use tracing::info;
#[cfg(feature = "observers")]
use tracing::warn;

#[cfg(feature = "observers")]
use super::{ObserverRuntime, ObserverRuntimeConfig};
use super::{Result, Server, ServerConfig};

impl Server {
    /// Create a server with relay pagination support enabled.
    ///
    /// The adapter must implement [`RelayDatabaseAdapter`]. Currently, only
    /// `PostgresAdapter` and `CachedDatabaseAdapter<PostgresAdapter>` satisfy this bound.
    ///
    /// Relay queries issued against a server created with [`Server::new`] return a
    /// `Validation` error at runtime; those issued against a server created with this
    /// constructor succeed.
    ///
    /// # Arguments
    ///
    /// * `config` - Server configuration
    /// * `schema` - Compiled GraphQL schema
    /// * `adapter` - Database adapter (must implement `RelayDatabaseAdapter`)
    /// * `db_pool` - Database connection pool (optional, required for observers)
    ///
    /// # Errors
    ///
    /// Returns error if OIDC validator initialization fails.
    ///
    /// # Example
    ///
    /// ```text
    /// // Requires: running PostgreSQL database and compiled schema file.
    /// let adapter = Arc::new(PostgresAdapter::new(db_url).await?);
    /// let server = Server::with_relay_pagination(config, schema, adapter, None).await?;
    /// server.serve().await?;
    /// ```
    pub async fn with_relay_pagination<
        A: Writer + RelayDatabaseAdapter + Clone + Send + Sync + 'static,
    >(
        config: ServerConfig,
        schema: CompiledSchema,
        adapter: Arc<A>,
        db_pool: Option<sqlx::PgPool>,
    ) -> Result<Self> {
        // Same boot gates as `Server::new` — these must not drift by constructor (H16).
        // Refuse to boot if any field is marked for at-rest encryption (H12); the write
        // path does not encrypt, so the data would be stored in plaintext.
        crate::server::initialization::field_encryption_unsupported_check(&schema)?;
        // Build the runtime config from the compiled schema (validates format version,
        // reads the audit flag, applies the #421 page-size ceiling + change-log toggle).
        let executor_config = crate::server::initialization::executor_runtime_config(
            &schema, &config,
        )
        .map_err(|msg| {
            super::ServerError::ConfigError(format!("Incompatible compiled schema: {msg}"))
        })?;

        // Read every schema-derived subsystem through the one shared seam.
        let subsystems = Self::schema_subsystems(&schema, &config).await?;

        // The one shared cache seam — see `build_cached_adapter` (#889).
        let (cached, cache_config) = crate::server::initialization::build_cached_adapter(
            &schema,
            config.cache_enabled,
            adapter,
        )
        .await?;
        let executor = Arc::new(Executor::with_config_and_relay(
            schema.clone(),
            Arc::new(cached),
            executor_config,
        ));
        let subscription_manager = Arc::new(SubscriptionManager::new(Arc::new(schema)));

        // Boxed: `from_executor` constructs every subsystem, and its future is
        // large enough that inlining it tips each public constructor past
        // clippy's `large_futures` stack budget — and, more to the point, puts a
        // multi-KiB frame on every caller's stack.
        let mut server = Box::pin(Self::from_executor(
            config,
            executor,
            subscription_manager,
            subsystems,
            db_pool,
            #[cfg(feature = "arrow")]
            None,
        ))
        .await?;

        server.adapter_cache_enabled = cache_config.enabled;

        server.apply_compiled_config()
    }
}

#[cfg(feature = "arrow")]
impl Server {
    /// Create new server with pre-configured Arrow Flight service.
    ///
    /// Use this constructor when you want to provide a Flight service with a real database adapter.
    ///
    /// The GraphQL side is wrapped in a
    /// [`CachedDatabaseAdapter`](fraiseql_core::cache::CachedDatabaseAdapter) exactly as
    /// [`Server::new`] wraps it — `cache_enabled` means the same thing on this boot
    /// path as on every other (#889). The Flight service keeps its own handle to the
    /// raw adapter; the two do not share a cache, which is why `flight_upload_tables`
    /// and `cache_enabled` are mutually exclusive (see Errors).
    ///
    /// # Arguments
    ///
    /// * `config` - Server configuration
    /// * `schema` - Compiled GraphQL schema
    /// * `adapter` - Database adapter
    /// * `db_pool` - Database connection pool (optional, required for observers)
    /// * `flight_service` - Pre-configured Flight service (only available with arrow feature)
    ///
    /// # Errors
    ///
    /// Returns `ServerError::ConfigError` when `cache_enabled = true` is combined with a
    /// non-empty `flight_upload_tables`: a Flight `Upload` is a direct INSERT that never
    /// reaches the mutation runner, so nothing invalidates the result cache for the views
    /// over the uploaded table and GraphQL reads would serve pre-upload rows until the TTL
    /// expired. Also returns an error if OIDC validator initialization fails.
    pub async fn with_flight_service<A: Writer + Clone + Send + Sync + 'static>(
        config: ServerConfig,
        schema: CompiledSchema,
        adapter: Arc<A>,
        #[allow(unused_variables)]
        // Reason: used inside #[cfg(feature = "observers")] block; unused when feature is off
        db_pool: Option<sqlx::PgPool>,
        flight_service: Option<FraiseQLFlightService>,
    ) -> Result<Self> {
        // Same boot gates as `Server::new` — these must not drift by constructor (H16).
        // Refuse to boot on at-rest-encryption-marked fields (H12, plaintext write path).
        crate::server::initialization::field_encryption_unsupported_check(&schema)?;

        // The one combination the cache cannot be made honest for. Everything else on
        // this path caches exactly like `Server::new`; an allow-listed Upload writes
        // rows behind the cache's back, so refuse rather than serve stale reads.
        if config.cache_enabled && !config.flight_upload_tables.is_empty() {
            return Err(super::ServerError::ConfigError(
                "`cache_enabled = true` cannot be combined with a non-empty \
                 `flight_upload_tables`: a Flight Upload writes rows directly and does not \
                 pass the mutation pipeline, so it invalidates nothing and cached GraphQL \
                 reads would keep serving pre-upload rows until the TTL expired. Set \
                 `cache_enabled = false`, or leave `flight_upload_tables` empty to keep \
                 Upload disabled."
                    .to_string(),
            ));
        }

        // Build the runtime config from the compiled schema (validates format version,
        // reads the audit flag, applies the #421 page-size ceiling + change-log toggle).
        let executor_config = crate::server::initialization::executor_runtime_config(
            &schema, &config,
        )
        .map_err(|msg| {
            super::ServerError::ConfigError(format!("Incompatible compiled schema: {msg}"))
        })?;

        // Read every schema-derived subsystem through the one shared seam.
        let subsystems = Self::schema_subsystems(&schema, &config).await?;

        // The one shared cache seam — see `build_cached_adapter` (#889). This path
        // used to skip it entirely, which is what made `cache_enabled` inert here.
        let (cached, cache_config) = crate::server::initialization::build_cached_adapter(
            &schema,
            config.cache_enabled,
            adapter,
        )
        .await?;

        let executor =
            Arc::new(Executor::with_config(schema.clone(), Arc::new(cached), executor_config));
        let subscription_manager = Arc::new(SubscriptionManager::new(Arc::new(schema)));

        // Boxed: `from_executor` constructs every subsystem, and its future is
        // large enough that inlining it tips each public constructor past
        // clippy's `large_futures` stack budget — and, more to the point, puts a
        // multi-KiB frame on every caller's stack.
        let mut server = Box::pin(Self::from_executor(
            config,
            executor,
            subscription_manager,
            subsystems,
            db_pool,
            flight_service,
        ))
        .await?;

        server.adapter_cache_enabled = cache_config.enabled;

        server.apply_compiled_config()
    }
}

impl Server {
    /// Initialize observer runtime from configuration.
    ///
    /// # Errors
    ///
    /// Returns `ServerError::ConfigError` when a non-Postgres observer transport
    /// is selected that this binary cannot run (feature not compiled in, or NATS
    /// without a URL) while in production mode (#350), or when the transport
    /// configuration is otherwise invalid. In development such a selection is
    /// downgraded to a warning and the runtime falls back to PostgreSQL.
    #[cfg(feature = "observers")]
    // Reason: one of the `init_*` family the boot sequence awaits uniformly; the
    // other members connect to their transports.
    #[allow(unknown_lints, clippy::unused_async_trait_impl)]
    pub(super) async fn init_observer_runtime(
        config: &ServerConfig,
        pool: Option<&sqlx::PgPool>,
    ) -> crate::Result<Option<Arc<RwLock<ObserverRuntime>>>> {
        // Check if enabled
        let observer_config = match &config.observers {
            Some(cfg) if cfg.enabled => cfg,
            _ => {
                info!("Observer runtime disabled");
                return Ok(None);
            },
        };

        let Some(pool) = pool else {
            warn!("No database pool provided for observers");
            return Ok(None);
        };

        info!("Initializing observer runtime");

        let runtime_config = Self::observer_runtime_config(observer_config, pool)?;
        let runtime = ObserverRuntime::new(runtime_config);
        Ok(Some(Arc::new(RwLock::new(runtime))))
    }

    /// Resolve `[observers.runtime]` plus its environment overrides into the
    /// runtime's configuration: the half of [`Self::init_observer_runtime`] that
    /// reads the environment and validates, without starting anything.
    ///
    /// # Errors
    ///
    /// See [`Self::init_observer_runtime`].
    #[cfg(feature = "observers")]
    pub(super) fn observer_runtime_config(
        observer_config: &crate::server_config::ObserverConfig,
        pool: &sqlx::PgPool,
    ) -> crate::Result<ObserverRuntimeConfig> {
        use fraiseql_observers::config::TransportKind;

        // Resolve the event transport from compiled config + env overrides, then
        // fail loud (#350) on a selection this binary cannot run before validating
        // the finer NATS/JetStream bounds — never a silent fallback to PostgreSQL.
        if let Some(name) =
            UNRUN_TRANSPORT_VARS.iter().find(|name| std::env::var_os(name).is_some())
        {
            return Err(crate::ServerError::ConfigError(format!(
                "{name} is set, but fraiseql-server always runs the observer executors and runs no \
                 PostgreSQL → NATS bridge for it to configure; unset it"
            )));
        }
        let mut transport = observer_config.runtime.transport.to_runtime().with_env_overrides();
        let compiled_in = cfg!(feature = "observers-nats");
        let nats_url_present = !transport.nats.url.is_empty();
        crate::server::initialization::observer_transport_check(
            transport.transport,
            compiled_in,
            nats_url_present,
            crate::ServerConfig::is_production_mode(),
        )?;

        // In production an unrunnable selection already returned above; the only
        // way past the guard with an unrunnable transport is development, where it
        // was downgraded to a warning — fall back to PostgreSQL so local boot works.
        let usable = match transport.transport {
            TransportKind::Postgres | TransportKind::InMemory => true,
            TransportKind::Nats => compiled_in && nats_url_present,
            _ => false,
        };
        if !usable {
            transport.transport = TransportKind::Postgres;
        }

        transport.validate().map_err(|e| {
            crate::ServerError::ConfigError(format!("invalid observer transport config: {e}"))
        })?;

        let runtime_config = ObserverRuntimeConfig::new(pool.clone())
            .with_poll_interval(observer_config.runtime.poll_interval_ms)
            .with_batch_size(observer_config.runtime.batch_size)
            .with_channel_capacity(observer_config.runtime.channel_capacity)
            .with_max_dlq_size(observer_config.runtime.max_dlq_size)
            .with_transport(transport)
            .with_email(observer_config.runtime.email.clone())
            .with_log_payloads(observer_config.runtime.log_payloads)
            // #985: the Redis backend that makes a `type = "cache"` action work.
            // Same `[observers.runtime.redis]` block dedup/result-cache use.
            .with_redis(observer_redis(observer_config.runtime.redis.as_ref())?);

        Ok(runtime_config)
    }
}

/// Transport variables that configure a bridge or a bridge-only node. The
/// server runs neither, so setting one is refused (#1496).
#[cfg(feature = "observers")]
const UNRUN_TRANSPORT_VARS: [&str; 6] = [
    "FRAISEQL_NATS_RUN_EXECUTORS",
    "FRAISEQL_NATS_ENABLE_BRIDGE",
    "FRAISEQL_BRIDGE_TRANSPORT_NAME",
    "FRAISEQL_BRIDGE_BATCH_SIZE",
    "FRAISEQL_BRIDGE_POLL_INTERVAL_SECS",
    "FRAISEQL_BRIDGE_NOTIFY_CHANNEL",
];

/// `FRAISEQL_REDIS_*` variables that configure the library's dedup and
/// result cache. The server runs neither, so setting one is refused.
#[cfg(feature = "observers")]
const UNUSED_REDIS_VARS: [&str; 3] = [
    "FRAISEQL_REDIS_POOL_SIZE",
    "FRAISEQL_REDIS_DEDUP_WINDOW_SECS",
    "FRAISEQL_REDIS_CACHE_TTL_SECS",
];

/// The `[observers.runtime.redis]` block with its `FRAISEQL_REDIS_*` overrides
/// applied, as the runtime's `RedisConfig`, validated (#1466).
///
/// A deployment that declares no block can supply one through
/// `FRAISEQL_REDIS_URL`: the URL is the one value without which there is no
/// Redis to configure, so the timeout variables alone leave the backend absent.
#[cfg(feature = "observers")]
fn observer_redis(
    declared: Option<&crate::server_config::ObserverRedisConfig>,
) -> crate::Result<Option<fraiseql_observers::config::RedisConfig>> {
    use crate::{ServerError::ConfigError, server_config::ObserverRedisConfig};

    if let Some(name) = UNUSED_REDIS_VARS.iter().find(|name| std::env::var_os(name).is_some()) {
        return Err(ConfigError(format!(
            "{name} is set, but fraiseql-server runs no Redis deduplication or result cache \
             for it to configure; unset it"
        )));
    }

    let env_url = std::env::var("FRAISEQL_REDIS_URL").ok();
    let Some(mut redis) = declared
        .cloned()
        .or_else(|| env_url.as_ref().map(|_| ObserverRedisConfig::default()))
    else {
        return Ok(None);
    };
    if let Some(url) = env_url {
        redis.url = url;
    }
    if let Some(secs) = env_secs("FRAISEQL_REDIS_CONNECT_TIMEOUT_SECS")? {
        redis.connect_timeout_secs = secs;
    }
    if let Some(secs) = env_secs("FRAISEQL_REDIS_COMMAND_TIMEOUT_SECS")? {
        redis.command_timeout_secs = secs;
    }

    let config = fraiseql_observers::config::RedisConfig {
        url: redis.url,
        connect_timeout_secs: redis.connect_timeout_secs,
        command_timeout_secs: redis.command_timeout_secs,
        ..fraiseql_observers::config::RedisConfig::default()
    };
    config.validate().map_err(|e| {
        ConfigError(format!(
            "invalid [observers.runtime.redis] (after FRAISEQL_REDIS_* overrides): {e}"
        ))
    })?;
    Ok(Some(config))
}

/// A whole-seconds environment variable; unset is `None`, unparseable an error.
#[cfg(feature = "observers")]
fn env_secs(name: &str) -> crate::Result<Option<u64>> {
    std::env::var(name).ok().map_or(Ok(None), |v| {
        v.parse().map(Some).map_err(|_| {
            crate::ServerError::ConfigError(format!(
                "{name}={v:?} is not a whole number of seconds"
            ))
        })
    })
}