fraiseql-server 2.16.0

//! Server lifecycle: serve, `serve_with_shutdown`, and `shutdown_signal`.

use std::net::SocketAddr;

use axum::serve::ListenerExt;
use tokio::net::TcpListener;
use tracing::{error, info, warn};

use super::{Result, Server, ServerError, TlsSetup};
#[cfg(feature = "observers")]
use crate::subscriptions::event_bridge::{EventBridge, EventBridgeConfig};

impl Server {
    /// Start server and listen for requests.
    ///
    /// Uses SIGUSR1-aware shutdown signal when a schema path is configured,
    /// enabling zero-downtime schema reloads via `kill -USR1 <pid>`.
    ///
    /// # Errors
    ///
    /// Returns error if server fails to bind or encounters runtime errors.
    pub async fn serve(self) -> Result<()> {
        self.serve_with_shutdown(Self::shutdown_signal()).await
    }

    /// Provision the persistent schemas the router is about to depend on.
    ///
    /// Everything here is async, must happen before the first request is served, and
    /// must **fail the boot** rather than leave an endpoint mounted over a table that
    /// does not exist. `build_router` is synchronous, so it cannot do any of it.
    ///
    /// Reached only through `provisioned_app_state`, which every serve entry point —
    /// [`Self::serve_with_shutdown`], [`Self::serve_on_listener`] and `serve_mcp_stdio`
    /// — obtains its state from. It used to be inlined in the first only, so the
    /// in-process test harness — which serves on a listener — reached the routes
    /// without ever running the DDL behind them. #748 (a parse error in the RBAC schema
    /// that aborted boot outright) was therefore invisible to every integration test in
    /// the workspace: they got a happily-running server that answered `500` on the RBAC
    /// routes. A boot step that only some entry points perform is the same class of
    /// defect as a guard only some call sites consult; #1332 was the stdio transport
    /// skipping it.
    ///
    /// **Scope is deliberately "what the mounted routes need in place", not "every
    /// startup step".** Subsystem startup that opens external connections — the
    /// observer runtime, Flight, the IMAP/cron/source pollers — stays in
    /// `serve_with_shutdown`, which `serve_on_listener` documents itself as skipping.
    /// The functions runtime moved here in #896, once it stopped needing a file on
    /// disk: every entry point must mount the same before-mutation hooks, or the
    /// in-process one greens a dispatch surface that never ran.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::ConfigError`] if any subsystem's schema cannot be
    /// created. Usage persistence is the one deliberate exception: it degrades to
    /// in-memory counters with a warning, because losing metering is not worth
    /// refusing to serve traffic.
    pub(super) async fn provision_persistent_schemas(&mut self) -> Result<()> {
        // Ensure RBAC schema exists before the router mounts RBAC endpoints.
        // Must run here (async context) rather than inside build_router() (sync).
        #[cfg(feature = "observers")]
        if let Some(ref db_pool) = self.db_pool {
            if self.config.admin_token.is_some() {
                let rbac_backend =
                    crate::api::rbac_management::db_backend::RbacDbBackend::new(db_pool.clone());
                rbac_backend.ensure_schema().await.map_err(|e| {
                    ServerError::ConfigError(format!("Failed to initialize RBAC schema: {e}"))
                })?;
            }
        }

        // Tenant admin credentials (#1089): the table the admin gate reads must exist before
        // the first request reaches a tenant-aware admin router.
        if let Some(tokens) = self.admin_tokens.as_ref() {
            tokens.ensure_schema().await.map_err(|e| {
                ServerError::ConfigError(format!("Failed to initialize admin-token schema: {e}"))
            })?;
        }

        // Ensure the API-key table exists before any request can resolve a
        // Postgres-stored key (#627). Fail loud: a configured postgres store
        // whose DDL cannot run must refuse to serve, not authenticate nothing.
        if let Some(store) = self.api_key_authenticator.as_ref().and_then(|a| a.postgres_store()) {
            store.ensure_schema().await.map_err(|e| {
                ServerError::ConfigError(format!("Failed to initialize API-key schema: {e}"))
            })?;
            info!("API-key schema ready (core.tb_api_key)");
        }

        // Ensure the SAML session + account tables exist before the router
        // mounts /auth/saml/acs (#381): the first verified assertion mints a
        // Postgres session and resolves an account — a missing table there
        // would 500 the login instead of failing the boot.
        #[cfg(feature = "auth-saml")]
        if self.saml_state.is_some() {
            // auth-saml implies auth, so the enrichment_pool field exists and
            // carries the same pool the constructor validated as present.
            if let Some(ref pool) = self.enrichment_pool {
                fraiseql_auth::PostgresSessionStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize SAML session schema: {e}"
                        ))
                    },
                )?;
                fraiseql_auth::PostgresAccountStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize SAML account schema: {e}"
                        ))
                    },
                )?;
                info!("SAML session + account schema ready");
            }
        }

        // SCIM provisioning schema (#946). Fail loud: a mounted provisioning surface whose
        // tables do not exist would 500 on the IdP's first call, which reads to an operator
        // as "FraiseQL's SCIM is broken" rather than "the boot skipped a migration".
        #[cfg(feature = "auth")]
        if let (Some(scim), Some(pool)) = (self.config.scim.as_ref(), self.enrichment_pool.as_ref())
        {
            if scim.enabled {
                fraiseql_auth::PostgresAccountStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize the SCIM identity schema: {e}"
                        ))
                    },
                )?;
                // Deactivation revokes through the session store, so `_system.sessions`
                // has to exist even on a deployment that mounts SCIM and no login route:
                // otherwise the revoke half of offboarding fails on a missing relation and
                // only the "block new sessions" half works.
                fraiseql_auth::PostgresSessionStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize the session schema SCIM revokes through: {e}"
                        ))
                    },
                )?;
                fraiseql_auth::scim::ScimStore::init(&fraiseql_auth::scim::PgScimStore::new(
                    pool.clone(),
                    None,
                ))
                .await
                .map_err(|e| {
                    ServerError::ConfigError(format!("Failed to initialize the SCIM schema: {e}"))
                })?;
                info!("SCIM schema ready (core.tb_scim_group, core.tb_scim_token)");
            }
        }

        // Keep stored IdPs current (#947). Writes through this server's own admin API
        // refresh synchronously, so the ticker exists to pick up another replica's changes
        // and to keep the certificate-expiry warning honest. Spawned here rather than at
        // construction so a built-but-never-served Server leaves no task behind.
        #[cfg(feature = "auth-saml")]
        if let (Some(saml), Some(cfg)) = (self.saml_state.as_ref(), self.config.saml.as_ref()) {
            if cfg.store_enabled {
                let registry = saml.registry().clone();
                let interval = std::time::Duration::from_secs(cfg.refresh_interval_secs);
                let warning_days = cfg.certificate_expiry_warning_days;
                self.tasks.spawn(registry.refresh_loop(interval, warning_days));
                info!(
                    refresh_interval_secs = cfg.refresh_interval_secs,
                    "SAML IdP store refresher started"
                );
            }
        }

        // Keep stored bucket policies current (#974), for the same reason the
        // IdP refresher above exists: a policy pushed at one replica applies
        // there immediately, and an access control that reached one replica out
        // of three is not an access control. This bounds how long the rest of
        // the deployment lags.
        if let Some(ref storage_state) = self.storage_state {
            let state = storage_state.clone();
            self.tasks.spawn(
                state.policy_refresh_loop(fraiseql_storage::DEFAULT_POLICY_REFRESH_INTERVAL),
            );
            info!(
                refresh_interval_secs = fraiseql_storage::DEFAULT_POLICY_REFRESH_INTERVAL.as_secs(),
                "Storage bucket policy refresher started"
            );
        }

        // Ensure the [auth.local] tables exist before the router mounts their
        // endpoints (#367). Every one of these would otherwise 500 on the first
        // real request against a fresh database; failing the boot is the loud
        // version of the same problem.
        #[cfg(feature = "auth")]
        if let Some(ref pool) = self.enrichment_pool {
            let mut provisioned: Vec<&str> = Vec::new();
            if self.otp_state.is_some() || self.mfa_state.is_some() {
                // Sessions and accounts back every local method.
                fraiseql_auth::PostgresSessionStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize [auth.local] session schema: {e}"
                        ))
                    },
                )?;
                fraiseql_auth::PostgresAccountStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize [auth.local] account schema: {e}"
                        ))
                    },
                )?;
            }
            if self.otp_state.is_some() {
                fraiseql_auth::PgOtpStore::new(pool.clone()).init().await.map_err(|e| {
                    ServerError::ConfigError(format!("Failed to initialize OTP schema: {e}"))
                })?;
                provisioned.push("otp");
            }
            if self.mfa_state.is_some() {
                fraiseql_auth::PgMfaStore::new(pool.clone()).init().await.map_err(|e| {
                    ServerError::ConfigError(format!("Failed to initialize MFA schema: {e}"))
                })?;
                provisioned.push("mfa");
            }
            if let Some(ref pw) = self.local_password_state {
                // `LocalPasswordAuthenticator::init` runs the identity, credential
                // and reset-token DDL, so it also covers sessions/accounts.
                pw.authenticator.init().await.map_err(|e| {
                    ServerError::ConfigError(format!(
                        "Failed to initialize local-password schema: {e}"
                    ))
                })?;
                fraiseql_auth::PostgresSessionStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize [auth.local] session schema: {e}"
                        ))
                    },
                )?;
                provisioned.push("password");
            }
            if self.anon_signup_state.is_some() {
                fraiseql_auth::PostgresSessionStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize [auth.local] session schema: {e}"
                        ))
                    },
                )?;
                provisioned.push("anonymous");
            }
            if !provisioned.is_empty() {
                info!(methods = ?provisioned, "[auth.local] schema ready");
            }
        }

        // Ensure the session + account tables exist before the router mounts
        // /auth/v1/callback (#368): a successful OAuth callback mints a
        // Postgres session and resolves an account through the account store —
        // a missing table there would 500 the first login instead of failing
        // the boot.
        #[cfg(feature = "auth")]
        if self.social_login.is_some() {
            if let Some(ref pool) = self.enrichment_pool {
                fraiseql_auth::PostgresSessionStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize social-login session schema: {e}"
                        ))
                    },
                )?;
                fraiseql_auth::PostgresAccountStore::new(pool.clone()).init().await.map_err(
                    |e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize social-login account schema: {e}"
                        ))
                    },
                )?;
                info!("Social-login session + account schema ready");
            }
        }

        // Ensure the inbound-ingestion tables exist before the router mounts
        // POST /webhooks/{provider}. Like RBAC, this must run here (async context)
        // rather than in the sync build_router(). The spine holds normalized
        // messages; the idempotency ledger backs the pipeline's atomic claim.
        #[cfg(feature = "inbound")]
        if let Some(ref db_pool) = self.db_pool {
            if !self.config.webhooks.is_empty() {
                crate::inbound::WebhookInboundState::init_spine(db_pool).await.map_err(|e| {
                    ServerError::ConfigError(format!(
                        "Failed to initialize inbound spine schema: {e}"
                    ))
                })?;
                fraiseql_webhooks::PostgresIdempotencyStore::new(db_pool.clone())
                    .init()
                    .await
                    .map_err(|e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize webhook idempotency schema: {e}"
                        ))
                    })?;
                info!(
                    routes = self.config.webhooks.len(),
                    "Inbound ingestion schema ready (spine + idempotency ledger)"
                );
            }
        }

        // Initialize usage persistence backend if configured.
        // Must run before build_router() so the aggregator is populated before
        // serving requests, but after the DB pool is available (async context).
        if let Some(ref usage_cfg) = self.config.usage.clone() {
            use std::time::Duration;

            use sqlx::postgres::PgPoolOptions;
            use tokio::time::MissedTickBehavior;

            use crate::usage::aggregator::{PostgresBackend, global_aggregator};

            match PgPoolOptions::new()
                .max_connections(2) // small dedicated pool — only used for periodic flushes
                .connect(&self.config.database_url)
                .await
            {
                Ok(pool) => {
                    match PostgresBackend::new(pool).await {
                        Ok(backend) => {
                            let backend = std::sync::Arc::new(backend);
                            // Upgrade global aggregator's backend from NoopBackend.
                            global_aggregator().set_backend(backend.clone());
                            // Restore persisted counters before serving requests.
                            // A failed load is NOT recoverable-by-continuing: the
                            // flush loop would then be writing to a store this
                            // process could not read. Revert to the in-memory
                            // NoopBackend and do not arm the loop at all, so
                            // billing counters are never touched by a process that
                            // does not know what they were (#861).
                            if let Err(e) = global_aggregator().load_from_backend().await {
                                tracing::error!(
                                    error = %e,
                                    "Usage persistence: startup load failed — persistence is \
                                     DISABLED for this process. Counters are in-memory only and \
                                     will not be written back. Restart once the store is \
                                     reachable."
                                );
                                global_aggregator().set_backend(std::sync::Arc::new(
                                    crate::usage::aggregator::NoopBackend,
                                ));
                            } else {
                                info!("Usage persistence: loaded counters from PostgreSQL");
                                // Spawn background flush task on the server's JoinSet
                                // so graceful shutdown can await its termination.
                                let flush_interval =
                                    Duration::from_secs(usage_cfg.flush_interval_secs);
                                let agg = std::sync::Arc::clone(global_aggregator());
                                self.tasks.spawn(async move {
                                    let mut ticker = tokio::time::interval(flush_interval);
                                    ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
                                    ticker.tick().await; // skip immediate first tick
                                    loop {
                                        ticker.tick().await;
                                        if let Err(e) = agg.flush_to_backend().await {
                                            warn!(error = %e, "Usage persistence: background flush failed");
                                        }
                                    }
                                });
                                info!(
                                    flush_interval_secs = usage_cfg.flush_interval_secs,
                                    "Usage persistence: PostgreSQL backend active"
                                );
                            }
                        },
                        Err(e) => {
                            warn!(
                                error = %e,
                                "Usage persistence: PostgresBackend initialization failed — \
                                 continuing with in-memory (NoopBackend)"
                            );
                        },
                    }
                },
                Err(e) => {
                    warn!(
                        error = %e,
                        "Usage persistence: failed to connect to PostgreSQL — \
                         continuing with in-memory (NoopBackend)"
                    );
                },
            }
        }

        // Prepare functions-runtime dispatch (load modules, register runtimes, attach
        // the `send_email` wiring) before the router is built, so `build_app_state`
        // mounts the before-mutation hooks and after:mutation functions actually fire.
        //
        // This step lived on `serve_with_shutdown` alone until #896, because it
        // re-read the compiled schema from `config.schema_path` and `serve_on_listener`
        // is driven with an in-memory schema and no file on disk. It now reads the
        // functions section the server was built with, so it belongs here with every
        // other provisioning step — and the in-process entry point stops silently
        // mounting no functions at all.
        #[cfg(feature = "functions-runtime")]
        self.prepare_functions_runtime().await?;

        Ok(())
    }

    /// Start server with a custom shutdown future.
    ///
    /// Enables programmatic shutdown (e.g., for `--watch` hot-reload) by accepting any
    /// future that resolves when the server should stop.
    ///
    /// # Errors
    ///
    /// Returns error if server fails to bind or encounters runtime errors.
    #[allow(clippy::cognitive_complexity)] // Reason: server lifecycle with TLS/non-TLS binding, signal handling, and graceful shutdown
    pub async fn serve_with_shutdown<F>(mut self, shutdown: F) -> Result<()>
    where
        F: std::future::Future<Output = ()> + Send + 'static,
    {
        let app_state = self.provisioned_app_state().await?;
        let app = self.build_router(&app_state);

        // Start the async-operation workers (#391) on the server's JoinSet, so
        // graceful shutdown aborts them (an aborted execution's row goes stale
        // and is reclaimed after the staleness threshold — the P19 recovery
        // path, exercised on every deploy).
        self.spawn_async_operation_workers(&app_state);

        // Start the poll-IMAP email workers.
        // Each configured `[mailbox.<name>.imap]` half runs a background poll loop
        // on the server's JoinSet, so graceful shutdown aborts them. The workers
        // reuse the durable inbound spine and the `after:ingest` dispatch path;
        // attachments stream into the legacy storage backend when one is
        // configured. Must run here (async, after `build_router` supplies the
        // function-dispatch hooks) rather than in the sync `build_router`.
        #[cfg(feature = "inbound-email")]
        if let Some(ref db_pool) = self.db_pool {
            if self.config.mailbox.values().any(|mailbox| mailbox.imap.is_some()) {
                use crate::inbound::{email, spine::PostgresInboundSpine};

                // The email path shares the inbound spine; create it here too in
                // case only `[mailbox.*.imap]` (no `[webhooks.*]`) is configured.
                // Both DDLs are idempotent.
                PostgresInboundSpine::new(db_pool.clone()).init().await.map_err(|e| {
                    ServerError::ConfigError(format!(
                        "Failed to initialize inbound spine schema: {e}"
                    ))
                })?;
                email::init_cursor_store(db_pool).await.map_err(|e| {
                    ServerError::ConfigError(format!(
                        "Failed to initialize inbound email cursor schema: {e}"
                    ))
                })?;

                // The correlator transitions send-status and suppression from
                // inbound signals; its tables are also created by the send path,
                // but a receive-only deployment needs them here too (idempotent).
                let tracker = std::sync::Arc::new(email::PgSendTracker::new(db_pool.clone()));
                tracker.init().await.map_err(|e| {
                    ServerError::ConfigError(format!(
                        "Failed to initialize send-tracking schema: {e}"
                    ))
                })?;
                let correlator =
                    std::sync::Arc::clone(&tracker) as std::sync::Arc<dyn email::SendCorrelator>;
                let address_hash_key = self.build_address_hash_key();

                // The attachment sink is the server's one configured storage
                // backend. It used to hang off a separate `with_storage` builder
                // method that no binary ever called, so `[mailbox]
                // attachment_bucket` was unreachable from the shipped server and
                // attachments were dropped with a warning.
                let sink = self.storage_state.as_ref().map(|state| {
                    std::sync::Arc::new(email::StorageAttachmentSink::new(state.backend.clone()))
                        as std::sync::Arc<
                            dyn fraiseql_functions::host::live::storage::StorageBackend,
                        >
                });
                // Opt-in Return-Path probe: verify the provider preserves
                // plus-addressing before trusting VERP correlation. Blocks startup
                // up to the probe window per eligible mailbox; off by default.
                if self.config.send.verp_probe_on_start {
                    email::run_startup_probes(&self.config.mailbox, |name| {
                        std::env::var(name).ok()
                    })
                    .await;
                }

                let hooks = app_state.before_mutation_hooks.clone();
                // #594: the after:ingest `fraiseql_query` bridge factory, built over the
                // app's hot-reloadable executor (same one the route handlers use); only
                // meaningful when function-dispatch hooks are present.
                let query_executor_factory = hooks.as_ref().map(|_| {
                    crate::routes::after_mutation::make_query_executor_factory(
                        app_state.executor.clone(),
                    )
                });
                let pollers = email::build_pollers(
                    &self.config.mailbox,
                    db_pool,
                    hooks.as_ref(),
                    query_executor_factory.as_ref(),
                    sink.as_ref(),
                    Some(&correlator),
                    address_hash_key.as_ref(),
                    self.config.send.challenge_suppress_after,
                    |name| std::env::var(name).ok(),
                );
                let started = pollers.len();
                for (poller, interval) in pollers {
                    self.tasks.spawn(async move { poller.poll_forever(interval).await });
                }
                info!(mailboxes = started, "poll-IMAP email sources started");
            }
        }

        // Start the scheduled-ingress source scheduler (#573): one poller per
        // enabled Model B (Deno) source in the compiled schema, each firing its
        // connector on its cron schedule under a single-firing lease with a host
        // bound to the source's durable cursor and its `run_as` executor. Runs on
        // the server's JoinSet so graceful shutdown drains it. Native pull sources
        // (poll-IMAP email) run via their own pollers above. Must run here (async,
        // after `build_router` supplies the function-dispatch hooks).
        #[cfg(feature = "sources")]
        if let Some(ref db_pool) = self.db_pool {
            let sources = app_state.executor().schema().sources.clone();
            if !sources.is_empty() {
                let sources_config = self.config.sources.clone().unwrap_or_default();
                if !crate::sources::sources_enabled(&sources_config) {
                    info!(
                        count = sources.len(),
                        "source scheduler disabled by config — sources not started"
                    );
                } else if let Some(hooks) = app_state.before_mutation_hooks.as_ref() {
                    // The shared durable source-cursor table (idempotent DDL).
                    fraiseql_observers::PostgresSourceCursorStore::new(db_pool.clone())
                        .init()
                        .await
                        .map_err(|e| {
                            ServerError::ConfigError(format!(
                                "Failed to initialize source cursor schema: {e}"
                            ))
                        })?;
                    let host_config = crate::sources::source_host_config(&sources_config);
                    let pollers = crate::sources::build_source_pollers(
                        &sources,
                        db_pool,
                        &app_state.executor,
                        hooks.as_ref(),
                        &host_config,
                        &fraiseql_functions::ResourceLimits::default(),
                        sources_config.log_payloads,
                    )?;
                    let started = pollers.len();
                    for poller in pollers {
                        self.tasks.spawn(async move { poller.run_forever().await });
                    }
                    info!(sources = started, "source scheduler started");
                } else if let Some(source) = sources.iter().find(|source| source.enabled) {
                    // Provisioning refuses this case first (#1399); a backstop, so an
                    // enabled source can never be left silently unscheduled.
                    return Err(ServerError::ConfigError(format!(
                        "source {:?} is enabled but no functions subsystem was built to run \
                         its connector {:?}",
                        source.name, source.function
                    )));
                }
            }
        }

        // Start the outbound CDC drains (#382): one worker per configured
        // `[[cdc_outbound.sinks]]`, draining the change-log outbox to its
        // broker on the server's JoinSet so graceful shutdown stops them.
        // Building them is fail-loud (no pool / unreachable broker / DDL
        // failure all refuse to boot) — a server that starts without its drain
        // is silent data loss for every downstream consumer.
        #[cfg(feature = "cdc-outbound")]
        {
            let drains = std::mem::take(&mut self.cdc_drains);
            if !drains.is_empty() {
                let started = drains.len();
                crate::cdc_outbound::spawn_all(drains, &mut self.tasks);
                info!(sinks = started, "cdc outbound drains started");
            }
        }

        // Start the subscription Kafka mirror (#1102), on the server's JoinSet so a
        // graceful shutdown stops it with everything else. Building it was fail-loud at
        // boot; there is nothing left to refuse here.
        #[cfg(feature = "subscription-kafka")]
        if let Some(mirror) = self.subscription_kafka.take() {
            crate::subscription_kafka::spawn(mirror, &self.subscription_manager, &mut self.tasks);
        }

        // Start the `cron:` function scheduler (#595): one leased poller per cron
        // function in the compiled schema, each firing on its schedule under a
        // single-firing advisory lease with the phase-02 `run_as` host. A cron
        // function is a scheduled source without a cursor. Runs on the server's
        // JoinSet so graceful shutdown drains it; requires a DB pool (the advisory
        // lease + `_fraiseql_cron_state`) and the function-dispatch hooks.
        #[cfg(feature = "functions-runtime")]
        if let Some(ref db_pool) = self.db_pool {
            if let Some(hooks) = app_state.before_mutation_hooks.as_ref() {
                if !hooks.trigger_registry.cron_triggers.is_empty() {
                    let cron_state = crate::cron::PgCronState::new(db_pool.clone());
                    cron_state.init().await.map_err(|e| {
                        ServerError::ConfigError(format!(
                            "Failed to initialize cron state schema: {e}"
                        ))
                    })?;
                    let host_config = crate::routes::after_mutation::host_context_config();
                    let pollers = crate::cron::build_cron_pollers(
                        db_pool,
                        &app_state.executor,
                        hooks.as_ref(),
                        &host_config,
                        &fraiseql_functions::ResourceLimits::default(),
                    )
                    .await
                    .map_err(|e| {
                        // Fail loud: a cross-restart fire guard whose state cannot
                        // be read back is not a guard (#796).
                        ServerError::ConfigError(format!(
                            "Failed to read back cron state (_fraiseql_cron_state): {e}"
                        ))
                    })?;
                    let started = pollers.len();
                    for poller in pollers {
                        self.tasks.spawn(async move { poller.run_forever().await });
                    }
                    info!(cron_functions = started, "cron scheduler started");
                }
            }
        }

        // Spawn SIGUSR1 schema reload handler when running on Unix.
        // The handler loops forever, reloading on each signal, until the
        // server process exits — tracked on the server's JoinSet so graceful
        // shutdown awaits its termination.
        #[cfg(unix)]
        if let Some(ref schema_path) = app_state.schema_path {
            let reload_state = app_state.clone();
            let reload_path = schema_path.clone();
            self.tasks.spawn(async move {
                let mut sigusr1 = match tokio::signal::unix::signal(
                    tokio::signal::unix::SignalKind::user_defined1(),
                ) {
                    Ok(s) => s,
                    Err(e) => {
                        warn!(error = %e, "Failed to install SIGUSR1 handler — schema hot-reload disabled");
                        return;
                    },
                };
                loop {
                    sigusr1.recv().await;
                    info!(
                        path = %reload_path.display(),
                        "Received SIGUSR1 — reloading schema"
                    );
                    match reload_state.reload_schema(&reload_path).await {
                        Ok(()) => {
                            let hash = reload_state.executor().schema().content_hash();
                            reload_state
                                .metrics
                                .schema_reloads_total
                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                            info!(schema_hash = %hash, "Schema reloaded successfully via SIGUSR1");
                        },
                        Err(e) => {
                            reload_state
                                .metrics
                                .schema_reload_errors_total
                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                            error!(
                                error = %e,
                                path = %reload_path.display(),
                                "Schema reload failed via SIGUSR1 — keeping previous schema"
                            );
                        },
                    }
                }
            });
            info!(
                path = %schema_path.display(),
                "SIGUSR1 schema reload handler installed"
            );
        }

        // Server-side TLS is unsupported; database TLS is applied where the pool is built.
        let tls_setup = TlsSetup::new(self.config.tls.clone());

        // Refuse to boot if server-side `[tls]` is enabled. FraiseQL does not terminate TLS
        // itself — it serves plaintext and expects a reverse proxy / load balancer / service
        // mesh to terminate TLS in front of it. Previously an enabled `[tls]` built a rustls
        // config that was silently discarded while the server kept serving plaintext and
        // logged `mtls_required = true` (M-tls-enforce); failing loud is honest.
        if tls_setup.is_tls_enabled() {
            return Err(ServerError::ConfigError(
                "[tls] (server-side TLS termination) is enabled but not supported: FraiseQL \
                 serves plaintext HTTP and expects TLS to be terminated by a reverse proxy, \
                 load balancer, or service mesh. Remove the [tls] section (or set its \
                 `enabled = false`) and terminate TLS in front of the server. Database \
                 connection TLS ([database_tls]) is unaffected."
                    .to_string(),
            ));
        }

        info!(
            bind_addr = %self.config.bind_addr,
            graphql_path = %self.config.graphql_path,
            tls_enabled = tls_setup.is_tls_enabled(),
            "Starting FraiseQL server"
        );

        // Start observer runtime if configured, wiring CDC events to EventBridge
        #[cfg(feature = "observers")]
        #[allow(unused_variables)]
        // Reason: _bridge_handle is kept alive to prevent task cancellation
        let _bridge_handle = {
            let mut handle: Option<tokio::task::JoinHandle<()>> = None;
            if let Some(ref runtime) = self.observer_runtime {
                info!("Starting observer runtime...");

                // Create EventBridge to forward CDC events to GraphQL subscriptions,
                // and — since #1309 — to the entity-event fan-out the REST
                // `/{resource}/stream` mount reads. The runtime feeds it from this
                // replica's own read of the changes (#1503), never from a per-request
                // subscription: `EventTransport::subscribe` is a competing consumer, so a
                // browser tab subscribing would have taken events away from dispatch.
                let mut bridge =
                    EventBridge::new(self.subscription_manager.clone(), EventBridgeConfig::new());
                if let Some(ref fanout) = self.entity_event_fanout {
                    bridge = bridge.with_entity_fanout(fanout.clone());
                }
                let sender = bridge.sender();

                let mut guard = runtime.write().await;
                guard.set_event_bridge_sender(sender);

                // #366: wire after:capture dispatch — externally-captured writes
                // (from the change-log reader) drive `after:capture` functions on
                // the phase-02 `run_as` host. The hook is a cheap no-op for
                // FraiseQL's own (non-captured) rows, so mediated writes never loop.
                #[cfg(feature = "functions-runtime")]
                if let Some(hooks) = app_state.before_mutation_hooks.as_ref() {
                    let hooks = std::sync::Arc::clone(hooks);
                    let factory = crate::routes::after_mutation::make_query_executor_factory(
                        app_state.executor.clone(),
                    );
                    guard.set_capture_dispatch(std::sync::Arc::new(move |event| {
                        if let Some((fn_event, cdc)) =
                            crate::routes::after_mutation::observer_event_to_capture(event)
                        {
                            let plans = crate::routes::after_mutation::plan_after_capture_dispatch(
                                &hooks,
                                &fn_event,
                                cdc.as_deref(),
                            );
                            if !plans.is_empty() {
                                crate::routes::after_mutation::spawn_after_capture(
                                    &hooks,
                                    plans,
                                    Some(factory.clone()),
                                );
                            }
                        }
                    }));
                }

                match guard.start().await {
                    Ok(()) => {
                        info!("Observer runtime started");
                        // Spawn EventBridge after observer runtime is running
                        handle = Some(bridge.spawn());
                        info!(
                            "EventBridge started — CDC events will be forwarded to subscriptions"
                        );
                    },
                    Err(e) => {
                        // A broker-backed transport (NATS) was an explicit operator
                        // choice; refusing to boot in production rather than silently
                        // coming up without it is the #350 dead-broker contract. The
                        // default PostgreSQL transport keeps the resilient
                        // log-and-continue behaviour (and development downgrades the
                        // NATS failure to the same warning).
                        if guard.transport_requires_broker()
                            && crate::ServerConfig::is_production_mode()
                        {
                            error!(
                                error = %e,
                                "Observer runtime failed to start on its configured \
                                 transport; refusing to boot (set FRAISEQL_ENV=development \
                                 to downgrade to a warning)"
                            );
                            return Err(e);
                        }
                        error!("Failed to start observer runtime: {}", e);
                        warn!("Server will continue without observers");
                    },
                }
                drop(guard);
            }
            handle
        };

        // Explicitly enable TCP_NODELAY (disable Nagle's algorithm) on every
        // accepted connection to minimise latency for small GraphQL responses.
        let listener = TcpListener::bind(self.config.bind_addr)
            .await
            .map_err(|e| ServerError::BindError(e.to_string()))?
            .tap_io(|tcp_stream| {
                if let Err(err) = tcp_stream.set_nodelay(true) {
                    warn!("failed to set TCP_NODELAY: {err:#}");
                }
            });

        // Warn if the process file descriptor limit is below the recommended minimum.
        // A low limit causes "too many open files" errors under load.
        #[cfg(target_os = "linux")]
        {
            if let Ok(limits) = std::fs::read_to_string("/proc/self/limits") {
                for line in limits.lines() {
                    if line.starts_with("Max open files") {
                        let parts: Vec<&str> = line.split_whitespace().collect();
                        if let Some(soft) = parts.get(3) {
                            if let Ok(n) = soft.parse::<u64>() {
                                if n < 65_536 {
                                    warn!(
                                        current_fd_limit = n,
                                        recommended = 65_536,
                                        "File descriptor limit is low; consider raising ulimit -n"
                                    );
                                }
                            }
                        }
                        break;
                    }
                }
            }
        }

        // Deliberately no "database TLS applied" line here. `serve()` does not build the
        // connection pool — it is handed one — so it is in no position to report what
        // transport security that pool negotiated. The line that used to be here read
        // `postgres_ssl_mode=verify-full … Database connection TLS configuration applied`
        // over a pool constructed with `NoTls`, which is the whole of #801. The pool now
        // reports its own mode from the site that builds it.

        info!("Server listening on http://{}", self.config.bind_addr);

        // Start both HTTP and gRPC servers concurrently if Arrow Flight is enabled
        #[cfg(feature = "arrow")]
        if let Some(mut flight_service) = self.flight_service.take() {
            let flight_addr = self.config.flight_bind_addr;

            // #954: attach the policy seam, never a bare executor. This is the
            // first point where both halves exist — `create_flight_service` builds
            // the service before there is any `AppState` to enforce policy from —
            // so the wiring lives here rather than in the constructor. Without it
            // Flight is the one transport that skips tenant resolution, the
            // suspended-tenant gate, per-tenant quotas and trusted documents.
            let app_state = app_state.clone();

            // #1349: the same wiring point, for the same reason. Flight builds its own
            // principal from a session token and, until now, dispatched it unresolved —
            // so `[identity.enrichment]`'s "every authenticated request resolves and
            // fail-closes" was true of six transports out of seven, and this one was
            // refused outright by the engine's backstop on any enrichment-declaring
            // deployment.
            #[cfg(feature = "auth")]
            if let Some(resolver) = app_state.identity_resolver.clone() {
                flight_service.set_identity_enricher(resolver);
                info!("Arrow Flight resolves enriched identity before dispatch (#1349)");
            }

            flight_service
                .set_executor(crate::arrow::policy_seam::policy_gated_executor(app_state));
            info!(
                "Arrow Flight GraphQL execution enabled through the policy seam (tenant \
                 dispatch, trusted documents, per-tenant quotas)"
            );

            info!("Arrow Flight server listening on grpc://{}", flight_addr);

            // Spawn Flight server in background, registered on the server's
            // JoinSet. The set's `shutdown` step abort-then-awaits the gRPC
            // server when the HTTP server exits.
            self.tasks.spawn(async move {
                if let Err(e) = tonic::transport::Server::builder()
                    .add_service(flight_service.into_server())
                    .serve(flight_addr)
                    .await
                {
                    error!(error = %e, "Arrow Flight server terminated with error");
                }
            });

            // Wrap the user-supplied shutdown future so we can also stop observer runtime
            #[cfg(feature = "observers")]
            let observer_runtime = self.observer_runtime.clone();
            let subscription_drain = std::sync::Arc::clone(&self.subscription_drain);

            let shutdown_with_cleanup = async move {
                shutdown.await;
                // #571: signal live subscription connections FIRST, so they can send
                // their Complete/close frames while the graceful-shutdown window is
                // still open.
                let _ = subscription_drain.send(true);
                #[cfg(feature = "observers")]
                if let Some(ref runtime) = observer_runtime {
                    info!("Shutting down observer runtime");
                    let mut guard = runtime.write().await;
                    if let Err(e) = guard.stop().await {
                        #[cfg(feature = "observers")]
                        error!("Error stopping runtime: {}", e);
                    } else {
                        info!("Runtime stopped cleanly");
                    }
                }
            };

            // Run HTTP server with graceful shutdown
            axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
                .with_graceful_shutdown(shutdown_with_cleanup)
                .await
                .map_err(|e| ServerError::IoError(std::io::Error::other(e)))?;

            // Abort and await every lifecycle task (Flight server, SIGUSR1
            // handler, PKCE cleanup, trusted-docs reload, usage flush, …).
            drain_lifecycle_tasks(self.tasks, self.config.shutdown_timeout_secs).await;
        }

        // HTTP-only server (when arrow feature not enabled)
        #[cfg(not(feature = "arrow"))]
        {
            // #571: signal live subscription connections as soon as shutdown fires,
            // so they can send their Complete/close frames while the
            // graceful-shutdown window is still open.
            let subscription_drain = std::sync::Arc::clone(&self.subscription_drain);
            let shutdown_with_drain = async move {
                shutdown.await;
                let _ = subscription_drain.send(true);
            };
            axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
                .with_graceful_shutdown(shutdown_with_drain)
                .await
                .map_err(|e| ServerError::IoError(std::io::Error::other(e)))?;

            let shutdown_timeout =
                std::time::Duration::from_secs(self.config.shutdown_timeout_secs);
            info!(
                timeout_secs = self.config.shutdown_timeout_secs,
                "HTTP server stopped, draining remaining work"
            );

            let drain = tokio::time::timeout(shutdown_timeout, async {
                #[cfg(feature = "observers")]
                if let Some(ref runtime) = self.observer_runtime {
                    let mut guard = runtime.write().await;
                    match guard.stop().await {
                        Ok(()) => info!("Observer runtime stopped cleanly"),
                        Err(e) => warn!("Observer runtime shutdown error: {e}"),
                    }
                }
            })
            .await;

            if drain.is_err() {
                warn!(
                    timeout_secs = self.config.shutdown_timeout_secs,
                    "Shutdown drain timed out; forcing exit"
                );
            } else {
                info!("Graceful shutdown complete");
            }

            // Abort and await every lifecycle task (SIGUSR1 handler, PKCE
            // cleanup, trusted-docs reload, usage flush, …).
            drain_lifecycle_tasks(self.tasks, self.config.shutdown_timeout_secs).await;
        }

        Ok(())
    }

    /// Start server on an externally created listener.
    ///
    /// Used in tests to discover the bound port before serving.
    /// Skips TLS, Flight, and observer startup — suitable for unit/integration tests only.
    ///
    /// It does **not** skip `provision_persistent_schemas`: the tables the
    /// mounted routes read from are created here exactly as they are under
    /// [`Self::serve_with_shutdown`], so a boot-time provisioning failure fails this
    /// entry point too. Skipping it was how #748 reached a release — the RBAC DDL did
    /// not parse, and every test in the workspace served the RBAC routes over tables
    /// that had never been created.
    ///
    /// # Errors
    ///
    /// Returns error if the server encounters a runtime error, or if
    /// `provision_persistent_schemas` cannot provision a configured subsystem.
    pub async fn serve_on_listener<F>(mut self, listener: TcpListener, shutdown: F) -> Result<()>
    where
        F: std::future::Future<Output = ()> + Send + 'static,
    {
        let app_state = self.provisioned_app_state().await?;
        let app = self.build_router(&app_state);
        // Same worker spawn as `serve_with_shutdown` — the two entry points must
        // not drift (#858's construction-path rule): a test harness that mounts
        // the routes but never executes submissions would green a dead surface.
        self.spawn_async_operation_workers(&app_state);
        // #571: drain live subscription connections here too, so the in-process
        // test harness exercises the same graceful teardown as production.
        let subscription_drain = std::sync::Arc::clone(&self.subscription_drain);
        let shutdown_with_drain = async move {
            shutdown.await;
            let _ = subscription_drain.send(true);
        };
        axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
            .with_graceful_shutdown(shutdown_with_drain)
            .await
            .map_err(|e| ServerError::IoError(std::io::Error::other(e)))?;
        // Abort and await any lifecycle tasks spawned during construction
        // (e.g. PKCE cleanup, trusted-docs reload) so the test path doesn't
        // leak background work into the next test.
        drain_lifecycle_tasks(self.tasks, self.config.shutdown_timeout_secs).await;
        Ok(())
    }

    /// Listen for shutdown signals (Ctrl+C or SIGTERM)
    pub async fn shutdown_signal() {
        use tokio::signal;

        let ctrl_c = async {
            match signal::ctrl_c().await {
                Ok(()) => {},
                Err(e) => {
                    warn!(error = %e, "Failed to install Ctrl+C handler");
                    std::future::pending::<()>().await;
                },
            }
        };

        #[cfg(unix)]
        let terminate = async {
            match signal::unix::signal(signal::unix::SignalKind::terminate()) {
                Ok(mut s) => {
                    s.recv().await;
                },
                Err(e) => {
                    warn!(error = %e, "Failed to install SIGTERM handler");
                    std::future::pending::<()>().await;
                },
            }
        };

        #[cfg(not(unix))]
        let terminate = std::future::pending::<()>();

        tokio::select! {
            () = ctrl_c => info!("Received Ctrl+C"),
            () = terminate => info!("Received SIGTERM"),
        }
    }
}

/// Abort every lifecycle task on the supplied [`tokio::task::JoinSet`] and await
/// the resulting `JoinError`s so the runtime is fully drained before
/// `serve_with_shutdown` returns.
///
/// Tasks are awaited under an outer timeout so a stuck task cannot prevent
/// process exit. A `JoinError::is_cancelled()` after `JoinSet::abort_all` is
/// the expected case — only unexpected panics are logged.
pub(super) async fn drain_lifecycle_tasks(
    mut tasks: tokio::task::JoinSet<()>,
    shutdown_timeout_secs: u64,
) {
    if tasks.is_empty() {
        return;
    }

    tasks.abort_all();
    let timeout = std::time::Duration::from_secs(shutdown_timeout_secs);
    let drained = tokio::time::timeout(timeout, async {
        while let Some(res) = tasks.join_next().await {
            if let Err(e) = res {
                if !e.is_cancelled() {
                    warn!(error = %e, "Lifecycle task terminated with a non-cancellation error");
                }
            }
        }
    })
    .await;
    if drained.is_err() {
        warn!(
            timeout_secs = shutdown_timeout_secs,
            "Lifecycle task drain timed out; some background tasks did not stop in time"
        );
    } else {
        info!("All lifecycle background tasks drained");
    }
}