liminal-server 0.25.4

use crate::ServerError;
use crate::auth_pass::PassVerifier;
use crate::cluster::ClusterHandle;
use crate::config::file::load_config;
use crate::config::types::ServiceProfile;
use crate::health::{ReadinessState, SharedReadinessState, start_health_server};
use crate::server::connection::services::{LiminalConnectionServices, build_connection_services};
use crate::server::connection::{BoundWebSocket, WebSocketListener};
use crate::server::connection::{ConnectionSupervisor, LysInstall, start_context_watch};
use crate::server::listener::{BoundAddresses, ServerListener};
use crate::server::self_streams::owner::PublisherRuntime;
use crate::server::shutdown::{ShutdownHandle, register_signal_handlers};
use crate::services::production::ServiceBindings;

use std::path::Path;
use std::sync::Arc;

#[path = "runtime_links.rs"]
pub(in crate::server) mod runtime_links;
use runtime_links::ConfiguredLinks;
#[path = "link_udp_endpoint.rs"]
pub mod link_udp_endpoint;

#[cfg(feature = "test-support")]
pub use runtime_links::test_support as link_test_support;

fn configured_authentication(
    config: &crate::config::types::ServerConfig,
) -> Result<(Option<Vec<u8>>, Option<PassVerifier>), ServerError> {
    let auth_token = config
        .auth
        .as_ref()
        .map(|auth| auth.token.clone().into_bytes());
    let pass_verifier = config
        .auth
        .as_ref()
        .map(PassVerifier::from_auth)
        .transpose()?
        .flatten();
    Ok((auth_token, pass_verifier))
}

/// Rewrites every participant conversation log under the configured
/// persistence path into the current row grammar, offline.
///
/// The store is opened exclusively, so this refuses while a server has it
/// open. Every conversation is replayed once, rewritten into its next log
/// generation, proved to replay to the same state, and only then switched.
///
/// # Errors
///
/// Returns [`ServerError`] when the configuration names no persistence path
/// or no `[participant]` section, the store cannot be opened, or a
/// conversation refuses the migration (named in the error).
pub fn run_participant_log_migration(
    config_path: &Path,
) -> Result<crate::server::participant::ParticipantLogMigration, ServerError> {
    let config = load_config(config_path)?;
    let Some(persistence_path) = config.persistence_path.as_deref() else {
        return Err(ServerError::ConfigValidation {
            message: "the participant log migration needs persistence_path; an ephemeral store has no logs to migrate".to_owned(),
        });
    };
    let Some(participant) = config.participant.as_ref() else {
        return Err(ServerError::ConfigValidation {
            message: "the participant log migration needs the [participant] section".to_owned(),
        });
    };
    let store = crate::server::connection::services::open_persistent_store(persistence_path)?;
    Ok(crate::server::participant::migrate_participant_logs(
        &store,
        participant,
    )?)
}

/// Starts the server deployment wrapper for the supplied configuration path.
///
/// # Errors
///
/// Returns [`ServerError`] when a later server lifecycle phase fails.
pub fn run(config_path: &Path) -> Result<(), ServerError> {
    let mut stages = StartStages::new();
    let config = load_config(config_path)?;
    stages.mark("config");

    // Enable metrics for this process before the health server accepts scrapes,
    // so `/metrics` renders the server families. Standalone liminal library users
    // never call this, so the registry gate stays off for them.
    crate::metrics::init();

    let readiness = SharedReadinessState::new(ReadinessState::default());
    let health_server = start_health_server(config.health_listen_address, readiness.clone())?;
    stages.mark("health");
    let shutdown_handle = ShutdownHandle::new();
    let signal_registration = register_signal_handlers(shutdown_handle.clone())?;

    // The configured [auth] token must ride along here: these call sites build
    // services themselves (full mode reaches the shared channel cluster first;
    // the worker front door builds no cluster at all) and so cannot use
    // `from_config`, which is the only other place the token is wired.
    let (auth_token, pass_verifier) = configured_authentication(&config)?;
    stages.mark("auth");
    // SRV-041 R3 and SRV-044 R2: the wire addresses and the WebSocket
    // address bind before Lys is asked and the store opens; a client
    // connecting during the build waits in the listen backlog until
    // accepting starts below, and is never refused for it.
    let bound = crate::server::boot_steps::timed("listener_bind", || bind_wire(&config))?;
    // ACCESS-003 R1 FOUND: the production server built its supervisor by a
    // path that installed no Lys, so `[auth.lys]` never reached a connection,
    // and it started with no tokio runtime, so a live ask had nothing to run
    // on. Both are built here, once, and the Lys install takes the same path
    // for every profile.
    let (lys_ask, lys, mut machine_issue) = runtime_links::lys(&config)?;
    stages.mark("lys");
    let FullProfile {
        connection_supervisor,
        cluster_handle,
        mut links,
        services: typed_services,
        context_watch,
        mut self_stream_publisher,
        spine,
    } = start_profile(
        &config,
        (auth_token, pass_verifier),
        (lys, lys_ask.as_ref()),
        &shutdown_handle,
        &health_server,
        &readiness,
        &mut stages,
    )?;

    // P0 #56 R4: readiness now reports whether the server can ADMIT, not just
    // whether it finished starting. It is installed here rather than at
    // `SharedReadinessState::new` above because the health endpoint binds before
    // the supervisor exists — liveness has to be answerable while the rest of
    // the server is still being built — so the authority that owns the answer
    // is not available until now.
    // SRV-028: installed before any connection is admitted; its processes
    // stop when the guard drops.
    let (services, mut grant_stream) = bootstrap::lys_owners(
        &config,
        &connection_supervisor,
        lys_ask.as_ref(),
        typed_services.as_ref(),
        &health_server,
    )?;
    let (mut listener, mut websocket_listener) =
        crate::server::boot_steps::timed("accept_start", || {
            open_listeners(
                bound,
                connection_supervisor,
                &readiness,
                &mut stages,
                spine.as_ref(),
            )
        })?;

    // SRV-028 R5: the address and the live restart after the bind.
    health_server.install_listen_addresses(listener.local_addresses())?;
    services.listening(listener.local_addr())?;
    machine_issue.begin()?;
    links.start(&health_server.link_bridges(), &listener.supervisor())?;
    stages.mark("services_registry");

    tracing::debug!(
        config_path = %config_path.display(),
        listen_address = %config.listen_address,
        health_listen_address = %health_server.local_addr(),
        "liminal server configuration validated"
    );

    listening::report_started(&listener, &health_server, &stages);

    shutdown_handle.wait();
    readiness.set_listener_bound(false);
    // SRV-028: before any connection drains, so a worker leaving with its
    // connection keeps its service live for the next start.
    services.shutdown_begun();

    // Tear the cluster down before draining connections: stop accepting peer
    // links and halt the membership poll loop. Each node shuts down independently
    // (no cluster-wide coordinated shutdown — that boundary belongs to SRV-004).
    if let Some(mut cluster_handle) = cluster_handle {
        cluster_handle.shutdown();
    }

    let supervisor = listener.supervisor();
    health_server.release_drafts()?;
    drop(typed_services);
    let shutdown_result = runtime_links::shutdown(
        &supervisor,
        config.drain_timeout(),
        || crate::server::shutdown::stop_accepting(&mut listener, websocket_listener.as_mut()),
        (&mut links, || {
            let grants = crate::auth_membership::grant_stream::finish_configured(&mut grant_stream);
            let publisher = finish_self_stream_publisher(&mut self_stream_publisher);
            grants.and(publisher)
        }),
    );
    let participant_fatal = supervisor.participant_service_fatal();
    let start_failure = supervisor.incarnation_start_failure();
    let publisher_shutdown = finish_self_stream_publisher(&mut self_stream_publisher);
    // The watch and the pass issue stop before the Lys ask runtime they use goes.
    drop(context_watch);
    let issue_finish = machine_issue.finish();
    drop(lys_ask);
    drop(websocket_listener);
    drop(signal_registration);
    // The health endpoint is torn down before the shutdown-sequence result is
    // raised: the teardown has to happen either way.
    bootstrap::run_answer(
        start_failure,
        raise_in_precedence(
            health_server.shutdown(),
            raise_in_precedence(
                publisher_shutdown,
                raise_in_precedence(issue_finish, shutdown_result),
            ),
        ),
        participant_fatal,
    )
}

/// What the full profile starts: the connection supervisor over the
/// production services, and the cluster, context watch and self-stream
/// publisher beside it when configured.
struct FullProfile {
    services: Option<Arc<LiminalConnectionServices>>,
    links: runtime_links::ConfiguredLinks,
    connection_supervisor: ConnectionSupervisor,
    /// The spine store opened held (DUR-021), released after the listeners bind.
    spine: Option<Arc<haematite::EventStore>>,
    cluster_handle: Option<ClusterHandle>,
    context_watch: Option<crate::services::context_watch::ContextWatch>,
    self_stream_publisher: Option<PublisherRuntime>,
}

fn start_profile(
    config: &crate::config::types::ServerConfig,
    (auth_token, pass_verifier): (Option<Vec<u8>>, Option<PassVerifier>),
    (lys, lys_ask): (Option<LysInstall>, Option<&tokio::runtime::Runtime>),
    shutdown: &ShutdownHandle,
    health: &crate::health::HealthServerHandle,
    readiness: &SharedReadinessState,
    stages: &mut StartStages,
) -> Result<FullProfile, ServerError> {
    match config.services.profile()? {
        ServiceProfile::Full => start_full_profile(
            config,
            (auth_token, pass_verifier),
            (lys, lys_ask),
            shutdown,
            health,
            readiness,
            stages,
        ),
        ServiceProfile::WorkerFrontDoor => {
            let services = build_connection_services(config)?;
            let connection_supervisor = ConnectionSupervisor::with_lys_and_fatal_shutdown(
                services,
                auth_token,
                pass_verifier,
                lys,
                config.limits,
                shutdown.clone(),
            )?;
            readiness.set_cluster_configured(false);
            stages.mark("supervisor");
            Ok(FullProfile {
                services: None,
                links: ConfiguredLinks::default(),
                connection_supervisor,
                spine: None,
                cluster_handle: None,
                context_watch: None,
                self_stream_publisher: None,
            })
        }
    }
}

/// Starts the full profile: the production connection services, the health
/// records they publish, the Lys install with its service bindings, the
/// connection supervisor, the context watch, the self-stream publisher and
/// clustering when configured.
fn start_full_profile(
    config: &crate::config::types::ServerConfig,
    (auth_token, pass_verifier): (Option<Vec<u8>>, Option<PassVerifier>),
    (lys, lys_ask): (Option<LysInstall>, Option<&tokio::runtime::Runtime>),
    shutdown_handle: &ShutdownHandle,
    health_server: &crate::health::HealthServerHandle,
    readiness: &SharedReadinessState,
    stages: &mut StartStages,
) -> Result<FullProfile, ServerError> {
    let services = Arc::new(LiminalConnectionServices::from_config_with_runtime(
        config,
        lys_ask.map(|runtime| runtime.handle().clone()),
    )?);
    stages.mark("services");
    // DUR-021 R4: the health endpoint reads the spine's reclaim status on each
    // request and offers the one operator route that quiesces it.
    let spine = services.durable_store().reclaim_spine();
    if let Some(spine) = &spine {
        health_server.install_spine(spine);
    }
    install_ephemeral_sweep_report(&services, health_server)?;
    // Publish the participant's refused-load record onto the health
    // endpoint. It happens HERE and not at `start_health_server`
    // because the endpoint binds before the participant exists —
    // liveness has to be answerable while the rest of the server is
    // still being built. Construction restores no conversation
    // (DUR-020): the record fills as conversations restore, on first
    // touch or in the warm, and is complete once the warm has finished.
    // The worker-front-
    // door profile configures no participant and installs nothing, and
    // the route reports that rather than an empty refusal set.
    if let Some(record) = services.unloadable_conversation_record() {
        health_server.install_unloadable_record(record);
    }
    // R18 amendment A7 (§0.18): the operator credential-re-issue
    // authority, published on the same line of reasoning and at the
    // same moment. A profile with no participant installs nothing and
    // the route says so.
    if let Some(reissuer) = services.credential_reissuer() {
        health_server.install_credential_reissuer(reissuer);
    }
    let channel_cluster = services.channel_cluster().clone();
    let signals_owner = Arc::clone(&services);
    let lys = lys
        .map(|install| configured_service_bindings(install, config, &services, lys_ask))
        .transpose()?;
    let bindings = lys.as_ref().and_then(|install| install.services.clone());
    let links = runtime_links::prepare(config, lys.as_ref(), &signals_owner)?;
    let connection_supervisor = ConnectionSupervisor::with_lys_and_fatal_shutdown(
        services,
        auth_token,
        pass_verifier,
        lys,
        config.limits,
        shutdown_handle.clone(),
    )?;
    stages.mark("supervisor");
    let mut context_watch = None;
    if let Some(bindings) = bindings.as_deref() {
        context_watch = start_context_watch(signals_owner.as_ref(), bindings, &config.limits)?;
        stages.mark("services");
    }

    let self_stream_publisher = start_self_stream_publisher(
        &signals_owner,
        &connection_supervisor,
        config,
        shutdown_handle,
    )?;
    stages.mark("incarnation");

    // SRV-005: start clustering on the channel-supervisor scheduler when a
    // [cluster] section is configured. The returned handle owns the inbound
    // distribution listener and the membership poll loop; it must outlive the
    // server and is torn down in the shutdown sequence below.
    let cluster_handle =
        start_configured_cluster(&channel_cluster, config.cluster.as_ref(), readiness)?;
    stages.mark("cluster");
    Ok(FullProfile {
        services: Some(signals_owner),
        links,
        connection_supervisor,
        spine,
        cluster_handle,
        context_watch,
        self_stream_publisher,
    })
}

/// Binds the wire listener and, when configured, its WebSocket sibling.
/// Readiness reports the listeners bound once both are.
/// Builds the ACCESS-003 service bindings `[auth.lys.services]` configures
/// into the Lys install: made as this machine, on the Lys ask runtime,
/// publishing through the server's own connection services.
fn configured_service_bindings(
    mut install: LysInstall,
    config: &crate::config::types::ServerConfig,
    services: &Arc<LiminalConnectionServices>,
    lys_ask: Option<&tokio::runtime::Runtime>,
) -> Result<LysInstall, ServerError> {
    let Some(lys) = config.auth.as_ref().and_then(|auth| auth.lys.as_ref()) else {
        return Ok(install);
    };
    let Some(runtime) = lys_ask else {
        return Err(ServerError::ConfigValidation {
            message: "auth.lys.services needs the Lys ask runtime; none was started".to_owned(),
        });
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |elapsed| elapsed.as_secs());
    install.services =
        ServiceBindings::configure(lys, Arc::clone(services), runtime.handle().clone(), now)
            .map_err(|error| ServerError::ConfigValidation {
                message: format!("auth.lys.services: {error}"),
            })?
            .map(Arc::new);
    Ok(install)
}

/// Binds the listeners and, once they are bound, starts the release of the
/// spine's held reclaim (DUR-021 R1 and R3) on its own thread, with what each
/// shard resumed on the log: the start does not wait for a shard to finish
/// its reclaim step to answer.
fn open_listeners(
    bound: BoundWire,
    supervisor: ConnectionSupervisor,
    readiness: &SharedReadinessState,
    stages: &mut StartStages,
    spine: Option<&Arc<haematite::EventStore>>,
) -> Result<(ServerListener, Option<WebSocketListener>), ServerError> {
    readiness.track_admission(supervisor.admission_readiness());
    let listeners = bind_listeners(bound, supervisor, readiness)?;
    stages.mark("listeners");
    crate::server::boot_steps::finished("time_to_listening", stages.began);
    if let Some(spine) = spine {
        let release =
            crate::server::connection::reclaim_placement::release_after_bind(spine, stages.began)?;
        listeners.0.supervisor().retain_reclaim_release(release)?;
    }
    Ok(listeners)
}

/// The wire addresses and, when `[websocket]` is configured, the WebSocket
/// address: bound and not yet accepting (SRV-044 R2).
#[derive(Debug)]
struct BoundWire {
    addresses: BoundAddresses,
    websocket: Option<BoundWebSocket>,
}

/// Binds every address the server listens on, in one step before Lys is
/// asked and the store opens. LP-WS-TRANSPORT R1.1: the WebSocket acceptor is
/// explicit opt-in; absent `[websocket]` binds nothing, no HTTP surface at all.
fn bind_wire(config: &crate::config::types::ServerConfig) -> Result<BoundWire, ServerError> {
    let addresses = BoundAddresses::bind(config)?;
    let websocket = config
        .websocket
        .as_ref()
        .map(BoundWebSocket::bind)
        .transpose()?;
    Ok(BoundWire {
        addresses,
        websocket,
    })
}

fn bind_listeners(
    bound: BoundWire,
    supervisor: ConnectionSupervisor,
    readiness: &SharedReadinessState,
) -> Result<(ServerListener, Option<WebSocketListener>), ServerError> {
    let listener = ServerListener::accept_on(bound.addresses, supervisor)?;
    let websocket_listener = bound
        .websocket
        .map(|websocket| WebSocketListener::accept_on(websocket, listener.supervisor()))
        .transpose()?;
    readiness.set_config_loaded(true);
    readiness.set_listener_bound(true);
    Ok((listener, websocket_listener))
}

#[cfg(test)]
#[path = "runtime/bind_wire_tests.rs"]
mod bind_wire_tests;

#[path = "runtime/start_stages.rs"]
mod start_stages;
use start_stages::StartStages;

mod bootstrap;
#[cfg(any(test, feature = "test-support"))]
pub(in crate::server) use bootstrap::bind_self_streams;
pub(in crate::server) use bootstrap::start_self_stream_publisher;
use bootstrap::{
    finish_self_stream_publisher, install_ephemeral_sweep_report, raise_in_precedence,
    start_configured_cluster,
};
#[cfg(test)]
#[path = "runtime/tests.rs"]
mod tests;

mod listening;