fraiseql-server 2.16.0

//! FraiseQL Server binary.
#![allow(clippy::print_stdout, clippy::print_stderr)] // Reason: server binary; pre-tracing startup banner and CLI errors go to stdout/stderr.

use std::{path::Path, sync::Arc};

use clap::Parser;
#[cfg(feature = "wire-backend")]
use fraiseql_core::db::FraiseWireAdapter;
#[cfg(not(feature = "wire-backend"))]
use fraiseql_core::db::postgres::PostgresAdapter;
use fraiseql_core::schema::CompiledSchema;
use fraiseql_server::{
    Cli, CompiledSchemaLoader, Server, ServerConfig,
    usage::{aggregator::global_aggregator, layer::MutationAuditLayer},
};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};

// ── Helper functions ──────────────────────────────────────────────────────

/// Load configuration from file or use defaults.
///
/// # Errors
///
/// Returns an error if the config file cannot be read or is not valid TOML.
fn load_config(config_path: Option<&str>) -> anyhow::Result<ServerConfig> {
    if let Some(path) = config_path {
        tracing::info!(path = %path, "Loading configuration from file");
        let config = ServerConfig::from_file(path)
            .map_err(|e| anyhow::anyhow!("failed to parse config file `{path}`: {e}"))?;
        #[cfg(feature = "observers")]
        check_observer_config_layout(&config, path)?;
        Ok(config)
    } else {
        tracing::info!("Using default server configuration");
        Ok(ServerConfig::default())
    }
}

/// Reject the pre-#342 flat `[observers]` server-tuning layout with a clear
/// migration message.
///
/// As of v2.5.0 the server's runtime tuning (`poll_interval_ms`, `batch_size`,
/// `channel_capacity`, `auto_reload`, `reload_interval_secs`, `pool`) moved from
/// `[observers]` to `[observers.runtime]` so the `[observers]` table can be
/// shared with the compiler schema (#342). A flat key here is captured as a
/// migration trap rather than silently ignored — boot fails loud, the #342
/// contract.
///
/// # Errors
///
/// Returns an error naming the misplaced keys when a pre-#342 server-tuning key
/// is found directly under `[observers]`.
#[cfg(feature = "observers")]
fn check_observer_config_layout(config: &ServerConfig, path: &str) -> anyhow::Result<()> {
    if let Some(observers) = &config.observers {
        let misplaced = observers.misplaced_runtime_keys();
        if let Some(first) = misplaced.first() {
            anyhow::bail!(
                "config file `{path}`: the [observers] key(s) `{}` moved to \
                 [observers.runtime] in v2.5.0. Move them under an \
                 `[observers.runtime]` table, e.g.\n\n    [observers.runtime]\n    {first} = ...\n",
                misplaced.join("`, `"),
            );
        }
    }
    Ok(())
}

/// Validate that schema file exists.
///
/// # Errors
///
/// Returns an error with a user-friendly message if the file does not exist.
fn validate_schema_path(path: &Path) -> anyhow::Result<()> {
    if !path.exists() {
        anyhow::bail!(
            "Schema file not found: {}. \
             Please compile schema first with: fraiseql-cli compile schema.json",
            path.display()
        );
    }
    Ok(())
}

/// Set up tracing subscriber with `RUST_LOG` env filter and optional OTLP export.
///
/// When `FRAISEQL_LOG_FORMAT=json` (case-insensitive), logs are emitted as
/// newline-delimited JSON — suitable for structured log aggregators such as
/// Datadog, Loki, or `CloudWatch`. Otherwise the default human-readable format
/// is used.
///
/// If an OTLP endpoint is configured (via `TracingConfig.otlp_endpoint` or the
/// `OTEL_EXPORTER_OTLP_ENDPOINT` environment variable), an `OpenTelemetry` span
/// exporter is added as an additional tracing layer.  When no endpoint is set,
/// no gRPC connection is attempted and there is zero overhead.
fn init_tracing(config: &ServerConfig, is_json: bool) {
    let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| "fraiseql_server=info,tower_http=info,axum=info".into());

    // Audit layer is always installed; it only records events with the
    // `fraiseql::mutation_audit` target and is otherwise a zero-cost no-op.
    let audit_layer = MutationAuditLayer::new(Arc::clone(global_aggregator()));

    if is_json {
        let subscriber = tracing_subscriber::registry()
            .with(env_filter)
            .with(audit_layer)
            .with(tracing_subscriber::fmt::layer().json());

        #[cfg(feature = "tracing-opentelemetry")]
        let subscriber = subscriber.with(build_otlp_layer(config));

        #[cfg(not(feature = "tracing-opentelemetry"))]
        let _ = config;

        subscriber.init();
    } else {
        let subscriber = tracing_subscriber::registry()
            .with(env_filter)
            .with(audit_layer)
            .with(tracing_subscriber::fmt::layer());

        #[cfg(feature = "tracing-opentelemetry")]
        let subscriber = subscriber.with(build_otlp_layer(config));

        #[cfg(not(feature = "tracing-opentelemetry"))]
        let _ = config;

        subscriber.init();
    }
}

/// Redact credentials from an endpoint URL before logging.
///
/// If the URL contains userinfo (`user:pass@host`), the credentials are replaced
/// with `[REDACTED]`.  Non-URL strings and parse failures are returned as-is with
/// no credential risk (they don't contain structured userinfo).
#[cfg(feature = "tracing-opentelemetry")]
fn redact_endpoint_credentials(endpoint: &str) -> String {
    match url::Url::parse(endpoint) {
        Ok(mut parsed) => {
            if !parsed.username().is_empty() || parsed.password().is_some() {
                // Reason: infallible for valid URLs — set_username / set_password
                // only fail on `cannot-be-a-base` URLs which have a scheme.
                let _ = parsed.set_username("[REDACTED]");
                let _ = parsed.set_password(None);
            }
            parsed.to_string()
        },
        Err(_) => endpoint.to_string(),
    }
}

/// Resolve the OTLP endpoint from config or environment, returning `None` if
/// neither is set (meaning OTLP export should be skipped entirely).
#[cfg(feature = "tracing-opentelemetry")]
fn resolve_otlp_endpoint(config: &ServerConfig) -> Option<String> {
    config
        .otlp_endpoint
        .clone()
        .or_else(|| std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok())
}

/// Build an optional `OpenTelemetry` tracing layer.
///
/// Returns `Some(layer)` when an OTLP endpoint is configured, `None` otherwise.
/// Failures during OTLP setup are logged to stderr (tracing is not yet initialized)
/// and result in `None` — the server continues without OTLP export.
#[cfg(feature = "tracing-opentelemetry")]
fn build_otlp_layer<S>(
    config: &ServerConfig,
) -> Option<tracing_opentelemetry::OpenTelemetryLayer<S, opentelemetry_sdk::trace::Tracer>>
where
    S: tracing::Subscriber + for<'span> tracing_subscriber::registry::LookupSpan<'span>,
{
    use opentelemetry::trace::TracerProvider as _;
    use opentelemetry_otlp::WithExportConfig;
    use opentelemetry_sdk::trace::SdkTracerProvider;

    let endpoint = resolve_otlp_endpoint(config)?;

    let exporter = opentelemetry_otlp::SpanExporter::builder()
        .with_http()
        .with_endpoint(&endpoint)
        .with_timeout(std::time::Duration::from_secs(config.otlp_export_timeout_secs))
        .build()
        .map_err(|e| {
            eprintln!(
                "Failed to build OTLP exporter for {}: {e}",
                redact_endpoint_credentials(&endpoint)
            );
        })
        .ok()?;

    let provider = SdkTracerProvider::builder()
        .with_batch_exporter(exporter)
        .with_resource(
            opentelemetry_sdk::Resource::builder()
                .with_service_name(config.tracing_service_name.clone())
                .build(),
        )
        .build();

    let tracer = provider.tracer("fraiseql");
    eprintln!(
        "OTLP tracing export enabled: endpoint={}, service_name={}",
        redact_endpoint_credentials(&endpoint),
        config.tracing_service_name
    );

    Some(tracing_opentelemetry::layer().with_tracer(tracer))
}

/// Load config from file/defaults, apply all CLI/env overrides, then validate.
///
/// # Errors
///
/// Returns an error if configuration loading fails (file I/O, parse errors) or
/// if the resulting configuration is invalid.
fn load_and_validate_config(cli: &Cli) -> anyhow::Result<ServerConfig> {
    let mut config = load_config(cli.server.config.as_deref())?;

    // Apply all CLI flag and env var overrides in one pass.
    cli.server.apply_to_config(&mut config);

    if let Err(e) = config.validate() {
        tracing::error!(error = %e, "Configuration validation failed");
        anyhow::bail!(e);
    }

    Ok(config)
}

/// Load and validate the compiled schema from the path in `config`.
async fn load_schema(config: &ServerConfig) -> anyhow::Result<LoadedSchema> {
    validate_schema_path(&config.schema_path)?;
    let schema_loader = CompiledSchemaLoader::new(&config.schema_path);
    // `load_extended` rather than `load`: the `functions` section is handed to the
    // server explicitly (#896) instead of being re-read from `schema_path` inside
    // `prepare_functions_runtime`, where it could name a different artifact than the
    // schema serving queries.
    let extended = schema_loader.load_extended().await?;
    // Install the schema's casing acronyms (on top of the built-in defaults) so the
    // runtime's `to_snake_case` JSONB-key resolution matches the compiled surface.
    fraiseql_core::utils::casing::set_runtime_acronyms(&extended.schema.naming_acronyms);
    tracing::info!("Compiled schema loaded successfully");
    Ok(LoadedSchema {
        schema: extended.schema,
        #[cfg(feature = "functions-runtime")]
        functions: extended.functions,
    })
}

/// The compiled schema plus the platform sections the server is given explicitly.
struct LoadedSchema {
    schema:    CompiledSchema,
    /// The `functions` section, handed to `Server::with_functions_config` (#896).
    #[cfg(feature = "functions-runtime")]
    functions: Option<fraiseql_server::schema::loader::FunctionsConfig>,
}

/// Initialize security configuration from the compiled schema (auth feature only).
///
/// Without `[auth]` configured, this is a no-op and RBAC/admin endpoints are
/// unprotected by OIDC — use `admin_token` or network controls as defence-in-depth.
#[cfg(feature = "auth")]
fn init_security(schema: &CompiledSchema) -> anyhow::Result<()> {
    tracing::info!("Initializing security configuration from schema");
    let schema_json_str = schema.to_json().unwrap_or_else(|e| {
        tracing::warn!(error = %e, "Failed to serialize schema to JSON");
        "{}".to_string()
    });
    let security_config = fraiseql_server::auth::init_security_config(&schema_json_str)
        .unwrap_or_else(|e| {
            tracing::warn!(error = %e, "Failed to load security config from schema, using defaults");
            fraiseql_server::auth::init_default_security_config()
        });
    if let Err(e) = fraiseql_server::auth::validate_security_config(&security_config) {
        tracing::error!(error = %e, "Security configuration validation failed");
        anyhow::bail!(e);
    }
    fraiseql_server::auth::log_security_config(&security_config);
    Ok(())
}

#[cfg(not(feature = "auth"))]
fn init_security(_schema: &CompiledSchema) -> anyhow::Result<()> {
    Ok(())
}

/// Create the PostgreSQL adapter.
#[cfg(not(feature = "wire-backend"))]
async fn build_postgres_adapter(config: &ServerConfig) -> anyhow::Result<Arc<PostgresAdapter>> {
    // `[database_tls]` is consumed *here*, at the site that opens the connections it
    // describes. It used to be read only by a log line inside `serve()`, several hundred
    // lines after this pool had already been built with `NoTls` (#801).
    let tls = config.postgres_tls().map_err(|e| anyhow::anyhow!(e))?;
    tls.warn_if_unverified();

    tracing::info!(
        pool_min_size = config.pool_min_size,
        pool_max_size = config.pool_max_size,
        pool_timeout_secs = config.pool_timeout_secs,
        postgres_ssl_mode = %tls.effective_mode(),
        "Initializing PostgreSQL connection pool"
    );
    let adapter = PostgresAdapter::with_pool_config(
        &config.database_url,
        fraiseql_core::db::postgres::PoolPrewarmConfig {
            min_size: config.pool_min_size,
            max_size: config.pool_max_size,
            timeout_secs: Some(config.pool_timeout_secs),
            // Single-tenant server pool: relation resolution follows the server default.
            // Per-tenant schema isolation is installed by `create_tenant_executor`.
            search_path: None,
            tls,
            read_replicas: config.read_replicas(),
            max_streaming_reads: config.pool_max_streaming_reads,
            // `[server] vector_hnsw_iterative_scan` / `vector_ivfflat_iterative_scan`,
            // read through the one accessor so this pool and every per-tenant pool
            // agree (#1116).
            vector_scan: config.vector_scan(),
        },
    )
    .await?;
    tracing::info!("PostgreSQL adapter ready");
    Ok(Arc::new(adapter))
}

/// Create the FraiseQL Wire adapter (when the `wire-backend` feature is enabled).
#[cfg(feature = "wire-backend")]
async fn build_wire_adapter(config: &ServerConfig) -> anyhow::Result<Arc<FraiseWireAdapter>> {
    // The wire backend has no replica routing; silently ignoring configured
    // replicas would serve every read from the primary while the operator
    // believes reads are offloaded (#407).
    config.wire_backend_rejects_read_replicas().map_err(|e| anyhow::anyhow!(e))?;
    tracing::info!(
        database_url = %config.database_url,
        "Initializing FraiseQL Wire database adapter (low-memory streaming)"
    );
    let adapter = FraiseWireAdapter::new(&config.database_url);
    tracing::info!("FraiseQL Wire adapter initialized successfully");
    Ok(Arc::new(adapter))
}

/// Create a dedicated PostgreSQL pool for the observer runtime.
///
/// Observers require their own pool because the LISTEN/NOTIFY connection
/// occupies a persistent slot that must not be shared with request-serving
/// connections (request connections need to be available for concurrent queries).
///
/// The observer pool is configured independently via `[observers.runtime.pool]`
/// in `fraiseql.toml`. When absent, observer-specific defaults are used (smaller
/// than the application pool — observers need far fewer connections).
// Gated on `not(wire-backend)` to match its sole caller `run_postgres`: with
// `wire-backend` the wire dispatch path is compiled instead and never builds an
// observer pool, so an unconditional `observers` gate would leave this dead
// under `--all-features` (which enables both).
#[cfg(all(not(feature = "wire-backend"), feature = "observers"))]
async fn build_observer_pool(config: &ServerConfig) -> anyhow::Result<Option<sqlx::PgPool>> {
    use std::time::Duration;

    use sqlx::postgres::PgPoolOptions;

    let pool_cfg = config.observers.as_ref().map(|o| o.runtime.pool.clone()).unwrap_or_default();

    tracing::info!(
        min = pool_cfg.min_connections,
        max = pool_cfg.max_connections,
        timeout_secs = pool_cfg.acquire_timeout_secs,
        "Initializing observer PostgreSQL pool"
    );

    let pool = PgPoolOptions::new()
        .min_connections(pool_cfg.min_connections)
        .max_connections(pool_cfg.max_connections)
        .acquire_timeout(Duration::from_secs(pool_cfg.acquire_timeout_secs))
        .connect(&config.database_url)
        .await?;

    Ok(Some(pool))
}

#[cfg(all(not(feature = "wire-backend"), not(feature = "observers")))]
async fn build_observer_pool(_config: &ServerConfig) -> anyhow::Result<Option<sqlx::PgPool>> {
    Ok(None)
}

/// Initialize the secrets manager backend if `--secrets-backend` / `FRAISEQL_SECRETS_BACKEND` is
/// set.
///
/// `FRAISEQL_SECRETS_BACKEND` selects the backend: `env`, `file` or `vault`.
/// This used to accept **any** value and build the environment backend for all of
/// them — an operator who set `FRAISEQL_SECRETS_BACKEND=vault` got env-var
/// "secrets management" with a healthy startup log (#856's root cause). Unknown
/// values and missing per-backend settings now refuse to boot.
///
/// Backend settings, all read from the environment:
///
/// * `file` — `FRAISEQL_SECRETS_FILE_PATH` (directory of one-file-per-secret).
/// * `vault` — `VAULT_ADDR` plus either `VAULT_TOKEN` (static token) or `VAULT_ROLE_ID` +
///   `VAULT_SECRET_ID` (`AppRole`, with automatic token renewal); optional `VAULT_NAMESPACE`;
///   `VAULT_TLS_VERIFY=false` to accept invalid TLS certificates (refused in production).
#[cfg(feature = "secrets")]
async fn build_secrets_manager()
-> anyhow::Result<Option<Arc<fraiseql_server::secrets_manager::SecretsManager>>> {
    use fraiseql_server::secrets_manager::{SecretsBackendConfig, VaultAuth};

    let Ok(backend) = std::env::var("FRAISEQL_SECRETS_BACKEND") else {
        tracing::debug!("Secrets manager disabled (set FRAISEQL_SECRETS_BACKEND to enable)");
        return Ok(None);
    };

    let env_or_bail = |name: &str| -> anyhow::Result<String> {
        std::env::var(name).map_err(|_| {
            anyhow::anyhow!(
                "FRAISEQL_SECRETS_BACKEND={backend} requires {name} to be set; refusing to \
                 boot without it rather than running with a partially configured secrets \
                 backend"
            )
        })
    };

    let cfg = match backend.as_str() {
        "env" => SecretsBackendConfig::Env,
        "file" => SecretsBackendConfig::File {
            path: std::path::PathBuf::from(env_or_bail("FRAISEQL_SECRETS_FILE_PATH")?),
        },
        "vault" => {
            let addr = env_or_bail("VAULT_ADDR")?;
            let auth = match (std::env::var("VAULT_TOKEN"), std::env::var("VAULT_ROLE_ID")) {
                (Ok(token), _) => VaultAuth::Token(zeroize::Zeroizing::new(token)),
                (Err(_), Ok(role_id)) => VaultAuth::AppRole {
                    role_id,
                    secret_id: zeroize::Zeroizing::new(env_or_bail("VAULT_SECRET_ID")?),
                },
                (Err(_), Err(_)) => anyhow::bail!(
                    "FRAISEQL_SECRETS_BACKEND=vault requires either VAULT_TOKEN (static \
                     token) or VAULT_ROLE_ID + VAULT_SECRET_ID (AppRole login)"
                ),
            };
            SecretsBackendConfig::Vault {
                addr,
                auth,
                namespace: std::env::var("VAULT_NAMESPACE").ok(),
                // Anything but an explicit "false" verifies TLS; the factory refuses
                // `false` in production (#727).
                tls_verify: !std::env::var("VAULT_TLS_VERIFY")
                    .is_ok_and(|v| v.eq_ignore_ascii_case("false")),
            }
        },
        other => anyhow::bail!(
            "unknown FRAISEQL_SECRETS_BACKEND {other:?}; expected \"env\", \"file\" or \
             \"vault\". (Every value used to silently select the env backend, so a \
             deployment that asked for Vault ran on environment variables while its \
             startup log read healthy.)"
        ),
    };

    tracing::info!(backend = %backend, "Initializing secrets manager");
    match fraiseql_server::secrets_manager::create_secrets_manager(cfg).await {
        Ok(manager) => Ok(Some(manager)),
        Err(e) => {
            tracing::error!(error = %e, "Failed to initialize secrets manager");
            anyhow::bail!("Secrets manager initialization failed: {}", e)
        },
    }
}

#[cfg(not(feature = "secrets"))]
async fn build_secrets_manager() -> anyhow::Result<Option<std::convert::Infallible>> {
    // Fail loud (M-secrets-backend-stub): if an operator sets
    // FRAISEQL_SECRETS_BACKEND but this binary was built without the `secrets`
    // feature, returning `Ok(None)` silently runs with NO secrets manager — the
    // operator believes secrets are managed when they are not. Refuse to boot.
    if std::env::var("FRAISEQL_SECRETS_BACKEND").is_ok() {
        anyhow::bail!(
            "FRAISEQL_SECRETS_BACKEND is set, but this binary was built without the `secrets` \
             feature, so no secrets backend can be initialized. Rebuild with \
             `--features secrets`, or unset FRAISEQL_SECRETS_BACKEND to run without a secrets \
             manager."
        );
    }
    Ok(None)
}

/// Warn at startup when `[files.<name>]` sections are configured: file-upload
/// routes are not yet wired into the binary, so the sections are parsed but
/// otherwise ignored. Emitting a warning keeps them from being silently dropped.
fn warn_files_not_wired(config: &ServerConfig) {
    if config.files.is_empty() {
        return;
    }
    let mut names: Vec<&str> = config.files.keys().map(String::as_str).collect();
    names.sort_unstable();
    tracing::warn!(
        sections = %names.join(", "),
        "[files.<name>] sections are configured but file-upload routes are not yet wired into \
         the fraiseql-server binary; these sections are ignored."
    );
}

// #469's warn-only handling of `[observers]`-without-the-feature is gone: since
// `ServerConfig` denies unknown fields (#839), a build without the feature refuses
// the config at parse time and `ServerConfig::from_file` names the missing build
// feature in the error.

/// Warn at startup when `[storage.<name>]` is configured for a database the
/// binary cannot mount storage on. Object storage is PostgreSQL-only because the
/// object-metadata repository requires a `sqlx::PgPool`.
#[cfg(feature = "wire-backend")]
fn warn_storage_requires_postgres(config: &ServerConfig, database: &str) {
    if config.storage.is_empty() {
        return;
    }
    tracing::warn!(
        database,
        "[storage.<name>] is configured but object storage via the binary is PostgreSQL-only \
         (the metadata repository requires PostgreSQL); storage routes will NOT be mounted."
    );
}

/// Warn at startup when `[security.token_revocation] backend = "postgres"` is set on
/// a database the binary cannot back it with. The Postgres revocation store requires
/// a `sqlx::PgPool`, so on non-PostgreSQL deployments the backend is unavailable and
/// token revocation will not be active (#357). Use `memory`/`redis`, or run on PostgreSQL.
#[cfg(feature = "wire-backend")]
fn warn_revocation_requires_postgres(schema: &CompiledSchema, database: &str) {
    if fraiseql_server::token_revocation::revocation_backend_is_postgres(schema) {
        tracing::warn!(
            database,
            "[security.token_revocation] backend = \"postgres\" but this binary is not running on \
             PostgreSQL; token revocation will NOT be active. Set backend = \"memory\" or \
             \"redis\", or deploy on PostgreSQL."
        );
    }
}

/// Warn at startup when the multi-tenant runtime is enabled on a database where
/// runtime tenant *provisioning* is unavailable. The registry and `X-Tenant-ID`
/// dispatch still work for pre-registered tenants, but `PUT /api/v1/admin/tenants`
/// needs a `FromPoolConfig` adapter, which only PostgreSQL provides today.
#[cfg(feature = "wire-backend")]
fn warn_tenant_provisioning_requires_postgres(config: &ServerConfig, database: &str) {
    if config.tenancy.runtime.enabled {
        tracing::warn!(
            database,
            "[tenancy.runtime] is enabled but runtime tenant provisioning \
             (PUT /api/v1/admin/tenants) is PostgreSQL-only; dispatch to \
             pre-registered tenants still works, but registration will fail."
        );
    }
}

// ── Entry point ───────────────────────────────────────────────────────────

/// Entry point.
///
/// Initialization sequence:
/// 1. **CLI** — parse command-line flags and env var overrides via clap.
/// 2. **Config** — load `ServerConfig` from file (via `--config` / `FRAISEQL_CONFIG`) or defaults,
///    then apply CLI/env overrides for database URL, bind address, schema path, metrics, admin API,
///    introspection, and rate limiting.
/// 3. **Tracing** — set up `tracing_subscriber` with `RUST_LOG` env filter.
/// 4. **Schema** — validate the compiled schema file exists and load it.
/// 5. **Security** — (auth feature) initialize and validate security config from schema.
/// 6. **Database** — create the PostgreSQL or Wire database adapter.
/// 7. **Observers / Secrets** — optionally create sqlx pool for observers and initialize the
///    secrets manager backend.
/// 8. **Server** — construct `Server` (with optional Arrow Flight service), optionally attach
///    secrets manager, then call `serve()` (or `serve_mcp_stdio()`).
#[tokio::main(flavor = "multi_thread")]
async fn main() -> anyhow::Result<()> {
    let cli = Cli::parse();

    // Load config first so tracing can include the OTLP layer if configured.
    // Tracing calls in load_and_validate_config are silently discarded (no
    // subscriber yet); critical errors surface via the Result return.
    let config = load_and_validate_config(&cli)?;
    init_tracing(&config, cli.server.is_json_log_format());
    tracing::info!("FraiseQL Server v{}", env!("CARGO_PKG_VERSION"));
    tracing::info!(
        bind_addr = %config.bind_addr,
        database_url = %config.database_url,
        graphql_path = %config.graphql_path,
        health_path = %config.health_path,
        introspection_path = %config.introspection_path,
        metrics_enabled = config.metrics_enabled,
        "Server configuration loaded"
    );

    // Install the global `metrics`-facade recorder so transitive crates
    // (e.g. fraiseql-wire) have their facade emissions captured and exported via
    // the `/metrics` endpoint (audit H45). No-op unless the `metrics` feature is
    // built and metrics are enabled in config.
    #[cfg(feature = "metrics")]
    if config.metrics_enabled {
        fraiseql_server::metrics_recorder::install();
    }

    let loaded = load_schema(&config).await?;
    init_security(&loaded.schema)?;

    warn_files_not_wired(&config);
    // `warn_sources_feature_missing` used to live here (#573): a lean build logged a
    // warning and started no scheduler. Superseded by #1326 — `load_extended` now
    // REFUSES a schema declaring an enabled source this build cannot run, which happens
    // before this point, so the warning could only ever fire for an all-disabled
    // `sources` array. For that case its advice was wrong: the operator turned the
    // sources off, and "rebuild with `--features sources`" answers a question nobody
    // asked.

    // Box::pin: the per-scheme dispatch holds adapter init futures for all
    // enabled adapters, which combined exceeds clippy's `large_futures`
    // 16-KiB stack threshold. Heap-allocating once at startup is fine.
    Box::pin(dispatch_server(config, loaded, &cli)).await
}

/// Dispatch on the configured database URL scheme and run the matching
/// adapter-specific server entry point.
///
/// The `wire-backend` feature short-circuits the URL-scheme dispatch entirely
/// because `FraiseWireAdapter` accepts its own URL formats.
#[cfg(feature = "wire-backend")]
async fn dispatch_server(
    config: ServerConfig,
    loaded: LoadedSchema,
    cli: &Cli,
) -> anyhow::Result<()> {
    let LoadedSchema {
        schema,
        #[cfg(feature = "functions-runtime")]
        functions,
    } = loaded;
    warn_storage_requires_postgres(&config, "wire-backend");
    warn_tenant_provisioning_requires_postgres(&config, "wire-backend");
    warn_revocation_requires_postgres(&schema, "wire-backend");
    let adapter = build_wire_adapter(&config).await?;
    // `FraiseWireAdapter` is read-only — it implements `DatabaseAdapter` and
    // deliberately not `Writer` — so this server is built through the
    // constructor that says so rather than the one bounded on write capability.
    let server = Server::new_read_only(config, schema, adapter, None).await?;
    // #896: the functions section comes from the artifact we loaded, not from a
    // second read of `schema_path` inside the serve path.
    #[cfg(feature = "functions-runtime")]
    let server = server.with_functions_config(functions);
    // Box::pin: with the wire backend enabled the combined server/serve future
    // exceeds clippy's `large_futures` 16-KiB threshold. Heap-allocating once at
    // startup is fine.
    Box::pin(finish_server(server, cli, /* with_arrow = */ false)).await
}

#[cfg(not(feature = "wire-backend"))]
async fn dispatch_server(
    config: ServerConfig,
    loaded: LoadedSchema,
    cli: &Cli,
) -> anyhow::Result<()> {
    use fraiseql_server::url_guard::{DatabaseScheme, parse_database_url};

    // Box::pin: the server-setup future exceeds clippy's `large_futures`
    // 16-KiB threshold once optional subsystems (observers, MCP) are enabled.
    // Heap-allocating once at startup is fine.
    match parse_database_url(&config.database_url)? {
        DatabaseScheme::Postgres => Box::pin(run_postgres(config, loaded, cli)).await,
    }
}

/// PostgreSQL entry point — preserves the existing observer-pool, Arrow
/// Flight, and relay-detection behaviour.
#[cfg(not(feature = "wire-backend"))]
async fn run_postgres(config: ServerConfig, loaded: LoadedSchema, cli: &Cli) -> anyhow::Result<()> {
    let LoadedSchema {
        schema,
        #[cfg(feature = "functions-runtime")]
        functions,
    } = loaded;
    let adapter = build_postgres_adapter(&config).await?;

    // #487: opt-in fail-fast existence check — every declared `sql_source` (query
    // view / mutation function) must be backed by the database. Default OFF (the
    // boot path is unchanged); when on, an unbacked source fails boot with a
    // precise list instead of surfacing as an opaque per-request 500 later. Runs
    // here, after the adapter exists, where schema + adapter coexist.
    if config.validate_sql_sources {
        use fraiseql_server::sql_source_check as check;
        let unbacked = check::find_unbacked_sources(&schema, adapter.as_ref()).await?;
        // #1426: present is not enough; the server's own role must be able to use it.
        let unusable = check::find_unusable_sources(&schema, adapter.as_ref()).await?;
        if !unbacked.is_empty() || !unusable.is_empty() {
            let role = check::connected_role(adapter.as_ref()).await?;
            anyhow::bail!("{}", check::format_source_problems(&unbacked, &unusable, &role));
        }
        tracing::info!(
            sources = schema.queries.len() + schema.mutations.len(),
            "sql_source validation passed: all declared sources are backed"
        );
    }

    let db_pool = build_observer_pool(&config).await?;

    // Wire `[storage.<name>]` into a mounted /storage/v1/* route group. Built
    // before the server is constructed (which moves `config`); a None result
    // means no storage was configured. PostgreSQL-only — see build_storage_state.
    let storage_state = fraiseql_server::server_config::build_storage_state(&config)
        .await
        .map_err(|e| anyhow::anyhow!("{e}"))?;
    if let Some(state) = &storage_state {
        tracing::info!(
            buckets = state.buckets.load().len(),
            "Object storage configured; mounting /storage/v1/*"
        );
    }

    // Postgres-backed token revocation (#357): revocation_manager_from_schema defers
    // the "postgres" backend because it needs a database connection, so build it here
    // (where the database URL is available) and install it on the server below.
    let pg_revocation_manager =
        fraiseql_server::token_revocation::build_postgres_revocation_manager(
            &config.database_url,
            &schema,
        )
        .await
        .map_err(|e| anyhow::anyhow!("{e}"))?;

    // Multi-tenant runtime provisioning (#330): the per-tenant executor factory is
    // built when `[tenancy.runtime] enabled = true`. Its adapter type MUST match the
    // server's — `CachedDatabaseAdapter<PostgresAdapter>` on every PG constructor
    // since #889 gave the arrow path the same result cache as its siblings. Capture
    // the flag here, before `config` is moved into the constructor.
    let tenancy_runtime_enabled = config.tenancy.runtime.enabled;

    // Captured before `config` is moved into the constructor, for the same reason.
    // Validated already by `build_postgres_adapter`, so this cannot be a new failure.
    let database_tls = config.postgres_tls().map_err(|e| anyhow::anyhow!(e))?;
    // Likewise: the vector-scan settings every tenant pool inherits (#1116), and
    // the read-replica routing policy (#957).
    let tenant_read_replica_policy = config.read_replica_policy();
    let tenant_vector_scan = config.vector_scan();

    // Arrow Flight path: only available with the `arrow` feature, only on PG.
    #[cfg(feature = "arrow")]
    {
        use fraiseql_server::arrow::create_flight_service;
        // #953: the Upload allow-list is an operator decision, read from
        // `flight_upload_tables`. Empty (the default) leaves Upload disabled.
        let flight_service = create_flight_service(adapter.clone(), &config.flight_upload_tables);
        // `OptimizedView` serves exactly the operator's `flight_views`, nothing else.
        fraiseql_server::arrow::register_flight_views(&flight_service, &config.flight_views).await;
        if config.flight_upload_tables.is_empty() {
            tracing::info!(
                "Arrow Flight service initialized with real database adapter (Upload disabled: \
                 flight_upload_tables is empty)"
            );
        } else {
            tracing::warn!(
                tables = ?config.flight_upload_tables,
                "Arrow Flight service initialized with real database adapter; Upload is ENABLED \
                 for these tables — clients may INSERT into them without passing the mutation \
                 pipeline's authorizer or field RBAC"
            );
        }
        let server =
            Server::with_flight_service(config, schema, adapter, db_pool, Some(flight_service))
                .await?;
        // #896: the functions section comes from the artifact we loaded, not from a
        // second read of `schema_path` inside the serve path.
        #[cfg(feature = "functions-runtime")]
        let server = server.with_functions_config(functions);
        // Every constructor wraps the adapter in `CachedDatabaseAdapter` (#889), so the
        // tenant factory must produce cached executors to match the server's adapter
        // type — the same expression the non-arrow branch uses.
        // Tenant pools inherit the server's `[database_tls]`, its read-replica
        // routing policy and its vector-scan settings; the registration request body
        // cannot influence any of them (see `make_executor_factory`). It does supply
        // its own replica URLs.
        let tenant_factory = tenancy_runtime_enabled.then(|| {
            fraiseql_server::tenancy::make_executor_factory::<
                fraiseql_core::cache::CachedDatabaseAdapter<PostgresAdapter>,
            >(
                database_tls.clone(), tenant_read_replica_policy.clone(), tenant_vector_scan
            )
        });
        let server = match storage_state {
            Some(state) => server.with_storage_state(state),
            None => server,
        };
        let server = match tenant_factory {
            Some(factory) => server.with_tenant_executor_factory(factory),
            None => server,
        };
        let server = match pg_revocation_manager {
            Some(manager) => server.with_revocation_manager(manager),
            None => server,
        };
        let server = enable_rest_writes(server);
        return finish_server(server, cli, /* with_arrow = */ true).await;
    }

    // Non-arrow PG path: pick the relay-capable constructor when the schema
    // declares relay queries (fraiseql/fraiseql#191).
    #[cfg(not(feature = "arrow"))]
    {
        let has_relay_queries = schema.queries.iter().any(|q| q.relay);
        let server = if has_relay_queries {
            Server::with_relay_pagination(config, schema, adapter, db_pool).await?
        } else {
            Server::new(config, schema, adapter, db_pool).await?
        };
        // #896: the functions section comes from the artifact we loaded, not from a
        // second read of `schema_path` inside the serve path.
        #[cfg(feature = "functions-runtime")]
        let server = server.with_functions_config(functions);
        // Non-arrow path: `Server::new`/`with_relay_pagination` wrap the adapter in
        // `CachedDatabaseAdapter`, so the tenant factory must produce cached executors
        // to match the server's adapter type.
        // Tenant pools inherit the server's `[database_tls]`, its read-replica
        // routing policy and its vector-scan settings; the registration request body
        // cannot influence any of them (see `make_executor_factory`). It does supply
        // its own replica URLs.
        let tenant_factory = tenancy_runtime_enabled.then(|| {
            fraiseql_server::tenancy::make_executor_factory::<
                fraiseql_core::cache::CachedDatabaseAdapter<PostgresAdapter>,
            >(
                database_tls.clone(), tenant_read_replica_policy.clone(), tenant_vector_scan
            )
        });
        let server = match storage_state {
            Some(state) => server.with_storage_state(state),
            None => server,
        };
        let server = match tenant_factory {
            Some(factory) => server.with_tenant_executor_factory(factory),
            None => server,
        };
        let server = match pg_revocation_manager {
            Some(manager) => server.with_revocation_manager(manager),
            None => server,
        };
        let server = enable_rest_writes(server);
        finish_server(server, cli, /* with_arrow = */ false).await
    }
}

/// Enable the REST write surface on a server whose adapter supports mutations.
///
/// One helper rather than a copy at each boot path, because "which backends serve REST
/// writes" is precisely the question that drifted out of existence in #865: `rest_router`
/// lost its only production caller while the served `OpenAPI` document went on
/// advertising every write path, so following the published contract earned a 405.
///
/// The read-only backend needs no entry here: this helper is reached only from the
/// non-`wire-backend` boot paths. It used to be bounded on the write capability as well,
/// described as the type system keeping writes off `FraiseWireAdapter` — a claim that
/// retired with the type parameter. The executor a wire-backend boot builds is
/// constructed read-only, so mounting the write routes over it would mount routes that
/// refuse, not routes that write.
#[cfg(all(not(feature = "wire-backend"), feature = "rest"))]
fn enable_rest_writes(server: Server) -> Server {
    server.with_rest_write_surface()
}

/// No-op counterpart for builds without the `rest` feature, where there is no REST
/// transport to mount at all.
#[cfg(all(not(feature = "wire-backend"), not(feature = "rest")))]
const fn enable_rest_writes(server: Server) -> Server {
    server
}

/// Finalize startup for any constructed `Server`: attach the secrets
/// manager if configured, dispatch to MCP-stdio mode if requested, otherwise
/// start the HTTP server.
#[cfg_attr(not(feature = "mcp"), allow(unused_variables))]
async fn finish_server(server: Server, cli: &Cli, with_arrow: bool) -> anyhow::Result<()> {
    // Attach secrets manager if configured.
    #[cfg(feature = "secrets")]
    let mut server = server;
    #[cfg(feature = "secrets")]
    if let Some(mgr) = build_secrets_manager().await? {
        server.set_secrets_manager(mgr);
    }
    #[cfg(not(feature = "secrets"))]
    let _ = build_secrets_manager().await?;

    // Serve MCP over stdio if requested, otherwise start HTTP server.
    #[cfg(feature = "mcp")]
    if cli.mcp_stdio.is_some() {
        tracing::info!("FraiseQL MCP stdio mode starting");
        // Boxed: the stdio path now builds the full `AppState` (#858), which puts
        // this future over the `clippy::large_futures` threshold.
        Box::pin(server.serve_mcp_stdio()).await?;
        return Ok(());
    }

    if with_arrow {
        tracing::info!(
            "FraiseQL Server {} starting (HTTP + Arrow Flight)",
            env!("CARGO_PKG_VERSION")
        );
    } else {
        tracing::info!("FraiseQL Server {} starting (HTTP only)", env!("CARGO_PKG_VERSION"));
    }

    // Boxed because `Server` is large and `serve()`'s future carries it (#1102 pushed it
    // past clippy::large_futures' 16 KiB threshold). One heap allocation, once per
    // process, on a future that lives for the process's whole run.
    Box::pin(server.serve()).await?;
    Ok(())
}