fraiseql-server 2.11.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 contents = std::fs::read_to_string(path)?;
        let config: ServerConfig = toml::from_str(&contents)
            .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<CompiledSchema> {
    validate_schema_path(&config.schema_path)?;
    let schema_loader = CompiledSchemaLoader::new(&config.schema_path);
    let schema = schema_loader.load().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(&schema.naming_acronyms);
    tracing::info!("Compiled schema loaded successfully");
    Ok(schema)
}

/// 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>> {
    tracing::info!(
        pool_min_size = config.pool_min_size,
        pool_max_size = config.pool_max_size,
        pool_timeout_secs = config.pool_timeout_secs,
        "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),
        },
    )
    .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>> {
    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.
#[cfg(feature = "secrets")]
async fn build_secrets_manager()
-> anyhow::Result<Option<Arc<fraiseql_server::secrets_manager::SecretsManager>>> {
    if std::env::var("FRAISEQL_SECRETS_BACKEND").is_err() {
        tracing::debug!("Secrets manager disabled (set FRAISEQL_SECRETS_BACKEND to enable)");
        return Ok(None);
    }
    tracing::info!("Initializing secrets manager from environment configuration");
    let cfg = fraiseql_server::secrets_manager::SecretsBackendConfig::Env;
    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."
    );
}

/// Warn at startup when the config file declares an `[observers]` section but
/// this binary was built without the `observers` feature (#469).
///
/// Without the feature the `ServerConfig.observers` field does not exist, so the
/// section is parsed-and-discarded by serde with no error: the observer runtime
/// never starts and the `/api/observers` admin routes return 404, with no
/// signal as to why. Because the field is compiled out, we cannot inspect the
/// deserialized `ServerConfig`; instead we re-read the raw TOML and check for a
/// top-level `[observers]` table so the operator gets a clear, actionable
/// warning rather than silent inaction.
#[cfg(not(feature = "observers"))]
fn warn_observers_feature_missing(config_path: Option<&str>) {
    let Some(path) = config_path else { return };
    let Ok(contents) = std::fs::read_to_string(path) else {
        return;
    };
    let has_observers =
        toml::from_str::<toml::Table>(&contents).is_ok_and(|table| table.contains_key("observers"));
    if has_observers {
        tracing::warn!(
            path,
            "[observers] is configured but this binary was built without the `observers` \
             feature; the section is ignored — the observer runtime will not start and the \
             /api/observers admin routes will return 404. Rebuild with `--features observers` \
             (or `observers-nats` / `observers-enterprise`)."
        );
    }
}

/// 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(any(
    feature = "mysql",
    feature = "sqlite",
    feature = "sqlserver",
    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(any(
    feature = "mysql",
    feature = "sqlite",
    feature = "sqlserver",
    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(any(
    feature = "mysql",
    feature = "sqlite",
    feature = "sqlserver",
    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 schema = load_schema(&config).await?;
    init_security(&schema)?;

    warn_files_not_wired(&config);
    #[cfg(not(feature = "observers"))]
    warn_observers_feature_missing(cli.server.config.as_deref());

    // 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, schema, &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,
    schema: CompiledSchema,
    cli: &Cli,
) -> anyhow::Result<()> {
    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?;
    let server = Server::new(config, schema, adapter, None).await?;
    // 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,
    schema: CompiledSchema,
    cli: &Cli,
) -> anyhow::Result<()> {
    use fraiseql_server::url_guard::{DatabaseScheme, parse_database_url};

    // Box::pin each arm: the per-scheme server-setup futures exceed clippy's
    // `large_futures` 16-KiB threshold once optional subsystems (observers, MCP,
    // multiple adapters) are enabled. Heap-allocating once at startup is fine.
    match parse_database_url(&config.database_url)? {
        DatabaseScheme::Postgres => Box::pin(run_postgres(config, schema, cli)).await,
        DatabaseScheme::MySql => Box::pin(run_mysql(config, schema, cli)).await,
        DatabaseScheme::Sqlite => Box::pin(run_sqlite(config, schema, cli)).await,
        DatabaseScheme::SqlServer => Box::pin(run_sqlserver(config, schema, 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,
    schema: CompiledSchema,
    cli: &Cli,
) -> anyhow::Result<()> {
    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 {
        let unbacked =
            fraiseql_server::sql_source_check::find_unbacked_sources(&schema, adapter.as_ref())
                .await?;
        if !unbacked.is_empty() {
            anyhow::bail!("{}", fraiseql_server::sql_source_check::format_unbacked(&unbacked));
        }
        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.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, which differs by build. The non-arrow PG path wraps the adapter in
    // `CachedDatabaseAdapter` (via `Server::new`/`with_relay_pagination`), while the
    // arrow path keeps the raw `PostgresAdapter` (via `Server::with_flight_service`,
    // which never caches). Both implement `FromPoolConfig`, so each branch below builds
    // its factory with the matching type. Capture the flag here, before `config` is
    // moved into the constructor.
    let tenancy_runtime_enabled = config.tenancy.runtime.enabled;

    // Arrow Flight path: only available with the `arrow` feature, only on PG.
    #[cfg(feature = "arrow")]
    {
        use fraiseql_server::arrow::create_flight_service;
        let flight_service = create_flight_service(adapter.clone());
        tracing::info!("Arrow Flight service initialized with real database adapter");
        let server =
            Server::with_flight_service(config, schema, adapter, db_pool, Some(flight_service))
                .await?;
        // Arrow path: the server holds the raw `PostgresAdapter`, so the tenant
        // factory must produce `PostgresAdapter` executors to match its adapter type.
        let tenant_factory = tenancy_runtime_enabled
            .then(fraiseql_server::tenancy::make_executor_factory::<PostgresAdapter>);
        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,
        };
        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?
        };
        // 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.
        let tenant_factory = tenancy_runtime_enabled.then(
            fraiseql_server::tenancy::make_executor_factory::<
                fraiseql_core::cache::CachedDatabaseAdapter<PostgresAdapter>,
            >,
        );
        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,
        };
        finish_server(server, cli, /* with_arrow = */ false).await
    }
}

/// MySQL entry point. Observer pool, Arrow Flight, and relay-aware
/// construction are not wired for non-PG adapters today; observers in
/// particular rely on PostgreSQL LISTEN/NOTIFY and have no MySQL equivalent.
#[cfg(all(not(feature = "wire-backend"), feature = "mysql"))]
async fn run_mysql(config: ServerConfig, schema: CompiledSchema, cli: &Cli) -> anyhow::Result<()> {
    warn_storage_requires_postgres(&config, "mysql");
    warn_tenant_provisioning_requires_postgres(&config, "mysql");
    warn_revocation_requires_postgres(&schema, "mysql");
    tracing::info!(
        pool_min_size = config.pool_min_size,
        pool_max_size = config.pool_max_size,
        "Initializing MySQL connection pool"
    );
    let adapter = Arc::new(
        fraiseql_core::db::mysql::MySqlAdapter::with_pool_config(
            &config.database_url,
            u32::try_from(config.pool_min_size).unwrap_or(u32::MAX),
            u32::try_from(config.pool_max_size).unwrap_or(u32::MAX),
        )
        .await?,
    );
    tracing::info!("MySQL adapter ready");
    let server = Server::new(config, schema, adapter, None).await?;
    finish_server(server, cli, /* with_arrow = */ false).await
}

#[cfg(all(not(feature = "wire-backend"), not(feature = "mysql")))]
async fn run_mysql(_: ServerConfig, _: CompiledSchema, _: &Cli) -> anyhow::Result<()> {
    anyhow::bail!(feature_off_message("mysql", "mysql"))
}

/// SQLite entry point — read-only.
#[cfg(all(not(feature = "wire-backend"), feature = "sqlite"))]
async fn run_sqlite(config: ServerConfig, schema: CompiledSchema, cli: &Cli) -> anyhow::Result<()> {
    warn_storage_requires_postgres(&config, "sqlite");
    warn_tenant_provisioning_requires_postgres(&config, "sqlite");
    warn_revocation_requires_postgres(&schema, "sqlite");
    fraiseql_server::url_guard::guard_sqlite_mutations(&schema)?;
    tracing::info!(
        pool_min_size = config.pool_min_size,
        pool_max_size = config.pool_max_size,
        "Initializing SQLite connection pool"
    );
    let adapter = Arc::new(
        fraiseql_core::db::sqlite::SqliteAdapter::with_pool_config(
            &config.database_url,
            u32::try_from(config.pool_min_size).unwrap_or(u32::MAX),
            u32::try_from(config.pool_max_size).unwrap_or(u32::MAX),
        )
        .await?,
    );
    tracing::info!("SQLite adapter ready (read-only)");
    let server = Server::new(config, schema, adapter, None).await?;
    finish_server(server, cli, /* with_arrow = */ false).await
}

#[cfg(all(not(feature = "wire-backend"), not(feature = "sqlite")))]
async fn run_sqlite(_: ServerConfig, _: CompiledSchema, _: &Cli) -> anyhow::Result<()> {
    anyhow::bail!(feature_off_message("sqlite", "sqlite"))
}

/// SQL Server entry point.
#[cfg(all(not(feature = "wire-backend"), feature = "sqlserver"))]
async fn run_sqlserver(
    config: ServerConfig,
    schema: CompiledSchema,
    cli: &Cli,
) -> anyhow::Result<()> {
    warn_storage_requires_postgres(&config, "sqlserver");
    warn_tenant_provisioning_requires_postgres(&config, "sqlserver");
    warn_revocation_requires_postgres(&schema, "sqlserver");
    tracing::info!(
        pool_min_size = config.pool_min_size,
        pool_max_size = config.pool_max_size,
        "Initializing SQL Server connection pool"
    );
    let adapter = Arc::new(
        fraiseql_core::db::sqlserver::SqlServerAdapter::with_pool_config(
            &config.database_url,
            u32::try_from(config.pool_min_size).unwrap_or(u32::MAX),
            u32::try_from(config.pool_max_size).unwrap_or(u32::MAX),
        )
        .await?,
    );
    tracing::info!("SQL Server adapter ready");
    let server = Server::new(config, schema, adapter, None).await?;
    finish_server(server, cli, /* with_arrow = */ false).await
}

#[cfg(all(not(feature = "wire-backend"), not(feature = "sqlserver")))]
async fn run_sqlserver(_: ServerConfig, _: CompiledSchema, _: &Cli) -> anyhow::Result<()> {
    anyhow::bail!(feature_off_message("sqlserver", "sqlserver"))
}

#[cfg(all(
    not(feature = "wire-backend"),
    any(
        not(feature = "mysql"),
        not(feature = "sqlite"),
        not(feature = "sqlserver")
    )
))]
fn feature_off_message(scheme: &str, feature: &str) -> String {
    format!(
        "fraiseql-server: {scheme}:// URL provided but the binary was built without the \
         `{feature}` Cargo feature. Rebuild with `cargo install fraiseql-server --features \
         {feature}` (or enable the feature in your downstream crate) and retry."
    )
}

/// Finalize startup for any constructed `Server<X>`: 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<X>(server: Server<X>, cli: &Cli, with_arrow: bool) -> anyhow::Result<()>
where
    X: fraiseql_core::db::DatabaseAdapter + Clone + Send + Sync + 'static,
{
    // 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");
        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"));
    }

    server.serve().await?;
    Ok(())
}