liminal-server 0.20.1

use std::fmt;
use std::path::Path;
use std::sync::Arc;
use std::time::Instant;

use crate::ServerError;
use crate::auth_pass::PassVerifier;
use crate::cluster::{self, ClusterHandle};
use crate::config::file::load_config;
use crate::config::types::{ClusterConfig, ServiceProfile};
use crate::health::{ReadinessState, SharedReadinessState, start_health_server};
use crate::server::connection::WebSocketListener;
use crate::server::connection::services::{
    ChannelCluster, LiminalConnectionServices, build_connection_services,
};
use crate::server::connection::{
    ConnectionSupervisor, LysInstall, configured_lys, lys_ask_runtime, start_context_watch,
};
use crate::server::listener::ServerListener;
use crate::server::self_streams::cost::ServerMetadata;
use crate::server::self_streams::owner::PublisherRuntime;
use crate::server::self_streams::sink::SelfStreams;
use crate::server::shutdown::{
    ShutdownHandle, register_signal_handlers, run_shutdown_sequence_before_flush,
};
use crate::services::production::ServiceBindings;

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();
    validate_config_path(config_path)?;

    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");
    // 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_ask_runtime(config.auth.as_ref())?;
    let lys = config
        .auth
        .as_ref()
        .map(|auth| {
            configured_lys(
                auth,
                lys_ask.as_ref().map(|runtime| runtime.handle().clone()),
            )
        })
        .transpose()?
        .flatten();
    stages.mark("lys");
    let mut context_watch = None;

    // D2: the service profile selects which connection-services stack is built.
    // Full mode is byte-for-byte the previous construction path (build services,
    // reach the shared channel cluster, start clustering when configured). The
    // worker front door constructs the connection supervisor over the
    // capability-scoped adapter and NOTHING else — no channel/conversation/haematite
    // services, and therefore no distribution cluster (config validation rejects a
    // `[cluster]` section under this profile, so none can be present here).
    let mut self_stream_publisher = None;
    let (connection_supervisor, cluster_handle) = match config.services.profile()? {
        ServiceProfile::Full => {
            let full = start_full_profile(
                &config,
                (auth_token, pass_verifier),
                (lys, lys_ask.as_ref()),
                &shutdown_handle,
                &health_server,
                &readiness,
                &mut stages,
            )?;
            context_watch = full.context_watch;
            self_stream_publisher = full.self_stream_publisher;
            (full.connection_supervisor, full.cluster_handle)
        }
        ServiceProfile::WorkerFrontDoor => {
            let services = build_connection_services(&config)?;
            // The front door admits and decides on the pass; it serves no
            // channel operation, so the service bindings have no publisher
            // here and are not built.
            let connection_supervisor = ConnectionSupervisor::with_lys_and_fatal_shutdown(
                services,
                auth_token,
                pass_verifier,
                lys,
                config.limits,
                shutdown_handle.clone(),
            )?;
            readiness.set_cluster_configured(false);
            stages.mark("supervisor");
            (connection_supervisor, None)
        }
    };

    // 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.
    readiness.track_admission(connection_supervisor.admission_readiness());

    let (mut listener, mut websocket_listener) =
        bind_listeners(&config, connection_supervisor, &readiness)?;
    stages.mark("listeners");

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

    tracing::info!(
        listen_address = %listener.local_addr(),
        health_listen_address = %health_server.local_addr(),
        start_stages = %stages,
        "liminal server started"
    );

    shutdown_handle.wait();
    readiness.set_listener_bound(false);

    // 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();
    let shutdown_result = run_shutdown_sequence_before_flush(
        &mut listener,
        websocket_listener.as_mut(),
        &supervisor,
        config.drain_timeout(),
        || finish_self_stream_publisher(&mut self_stream_publisher),
    );
    let participant_fatal = supervisor.participant_service_fatal();
    let publisher_shutdown = finish_self_stream_publisher(&mut self_stream_publisher);
    // The watch stops before the Lys ask runtime it reports through goes.
    drop(context_watch);
    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.
    raise_in_precedence(
        health_server.shutdown(),
        raise_in_precedence(publisher_shutdown, shutdown_result),
    )?;
    participant_fatal?.map_or(Ok(()), |fatal| {
        Err(ServerError::ParticipantServiceFatal { 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 {
    connection_supervisor: ConnectionSupervisor,
    cluster_handle: Option<ClusterHandle>,
    context_watch: Option<crate::services::context_watch::ContextWatch>,
    self_stream_publisher: Option<PublisherRuntime>,
}

/// 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(config)?);
    stages.mark("services");
    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. Boot has already recorded every conversation
    // it refused by the time `from_config` returns, so the first scrape
    // after this line sees the complete boot answer. 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 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 {
        connection_supervisor,
        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)
}

fn bind_listeners(
    config: &crate::config::types::ServerConfig,
    supervisor: ConnectionSupervisor,
    readiness: &SharedReadinessState,
) -> Result<(ServerListener, Option<WebSocketListener>), ServerError> {
    let listener = ServerListener::bind(config, supervisor)?;
    // LP-WS-TRANSPORT R1.1: the sibling WebSocket acceptor is explicit opt-in.
    // Absent `[websocket]` binds nothing — no HTTP surface exists at all — and
    // a bind failure fails startup BEFORE readiness reports the listeners
    // bound, exactly like the main listener.
    let websocket_listener = match config.websocket.as_ref() {
        Some(websocket_config) => Some(WebSocketListener::bind(
            websocket_config,
            listener.supervisor(),
        )?),
        None => None,
    };
    readiness.set_config_loaded(true);
    readiness.set_listener_bound(true);
    Ok((listener, websocket_listener))
}

/// The start path's stages, each as milliseconds since `run` began, so a
/// slow start says where its time went. Logged once, on the started line.
struct StartStages {
    began: Instant,
    marks: Vec<(&'static str, u128)>,
}

impl StartStages {
    fn new() -> Self {
        Self {
            began: Instant::now(),
            marks: Vec::with_capacity(8),
        }
    }

    /// The stage `name` has ended, now.
    fn mark(&mut self, name: &'static str) {
        self.marks.push((name, self.began.elapsed().as_millis()));
    }
}

/// Each stage with its own length and where it ended: `config 3 (3),
/// health 1 (4), ...` in milliseconds.
impl fmt::Display for StartStages {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut before = 0;
        for (index, (name, at)) in self.marks.iter().enumerate() {
            if index > 0 {
                formatter.write_str(", ")?;
            }
            write!(formatter, "{name} {} ({at})", at.saturating_sub(before))?;
            before = *at;
        }
        Ok(())
    }
}

fn validate_config_path(config_path: &Path) -> Result<(), ServerError> {
    if config_path.as_os_str().is_empty() {
        return Err(ServerError::ConfigLoad {
            message: "configuration path is empty".to_owned(),
        });
    }
    Ok(())
}

pub(in crate::server) fn start_self_stream_publisher(
    services: &Arc<LiminalConnectionServices>,
    supervisor: &ConnectionSupervisor,
    config: &crate::config::types::ServerConfig,
    shutdown: &ShutdownHandle,
) -> Result<Option<PublisherRuntime>, ServerError> {
    bind_self_streams(services, supervisor, config, shutdown)?
        .map(|streams| {
            PublisherRuntime::start(services.signals().wake(), streams, shutdown.clone())
        })
        .transpose()
}

fn finish_self_stream_publisher(
    publisher: &mut Option<PublisherRuntime>,
) -> Result<(), ServerError> {
    publisher.as_mut().map_or(Ok(()), PublisherRuntime::finish)
}

fn start_configured_cluster(
    channels: &ChannelCluster,
    config: Option<&ClusterConfig>,
    readiness: &SharedReadinessState,
) -> Result<Option<ClusterHandle>, ServerError> {
    readiness.set_cluster_configured(config.is_some());
    config
        .map(|config| start_cluster(channels, config, readiness))
        .transpose()
}

/// Binds the shared signals and incarnation-owned subjects before cluster startup.
pub(in crate::server) fn bind_self_streams(
    services: &Arc<LiminalConnectionServices>,
    supervisor: &ConnectionSupervisor,
    config: &crate::config::types::ServerConfig,
    shutdown: &ShutdownHandle,
) -> Result<Option<Arc<SelfStreams>>, ServerError> {
    let signals = services.signals();
    supervisor.install_signals(signals.clone())?;
    let Some(incarnation) = supervisor.current_server_incarnation()? else {
        crate::metrics::self_streams_disabled();
        tracing::warn!(
            "SELF_STREAMS_REQUIRE_PARTICIPANT: self streams disabled because no participant is configured"
        );
        // SRV-012 R4: the console streams ride the same publisher owner, so
        // without one no console subject is registered and a console
        // subscribe answers CHANNEL_NOT_REGISTERED.
        crate::metrics::console_streams_disabled();
        tracing::warn!(
            "CONSOLE_STREAMS_REQUIRE_PARTICIPANT: console streams disabled because no participant is configured"
        );
        return Ok(None);
    };
    services.install_self_stream_channels()?;
    // SRV-012 R7: this device's uptime runs from the incarnation now bound.
    services
        .console()
        .incarnation_started(std::time::Instant::now())
        .map_err(|error| ServerError::ListenerAccept {
            message: format!("console hub: {error}"),
        })?;
    let metadata = ServerMetadata::from_config(config, incarnation).map_err(|error| {
        ServerError::ConfigValidation {
            message: error.to_string(),
        }
    })?;
    let streams = Arc::new(SelfStreams::new(
        metadata,
        services,
        signals.wake(),
        shutdown.clone(),
        config.services.self_streams,
    ));
    signals.install(Arc::clone(&streams))?;
    services.install_console_channels()?;
    signals.enable_console()?;
    let wake = signals.wake();
    wake.install_source(services, &streams)?;
    let counts = services.store_counts();
    let names = ["durable", "ephemeral-dedup"];
    let configured = names
        .get(..counts.store_count())
        .ok_or_else(|| SelfStreams::failure("unexpected number of counted stores"))?;
    counts.install_observer(configured, wake).map_err(|error| {
        SelfStreams::failure(&format!("store count observer installation: {error}"))
    })?;
    Ok(Some(streams))
}

/// Raises the health endpoint's shutdown error ahead of the shutdown
/// sequence's, a precedence chosen deliberately: a health-shutdown failure takes
/// the exit code. The sequence error it masks is logged rather than dropped, so
/// only its exit code goes missing.
fn raise_in_precedence(
    health_shutdown: Result<(), ServerError>,
    shutdown_sequence: Result<(), ServerError>,
) -> Result<(), ServerError> {
    if let Err(error) = health_shutdown {
        if let Err(sequence_error) = &shutdown_sequence {
            tracing::error!(
                %sequence_error,
                "shutdown sequence failed; the health endpoint shutdown failure takes the exit code"
            );
        }
        return Err(error);
    }
    shutdown_sequence
}

/// Starts clustering on the shared channel supervisor's scheduler (SRV-005).
///
/// Installs the cluster `sync` as the supervisor's [`ClusterObserver`] so channel
/// subscribe/unsubscribe/publish events drive process-group membership and
/// cross-node fan-out.
///
/// On the success path this marks cluster membership as established on `readiness`
/// (G2) via [`cluster::start`]'s `on_established` hook, so a clustered server's
/// `/ready` endpoint transitions from 503 to 200 once the cluster stack is up.
/// Every early return here (missing resolver, listener bind failure, no reachable
/// seed) leaves the flag unset, so `/ready` stays 503.
fn start_cluster(
    channel_cluster: &ChannelCluster,
    cluster_config: &ClusterConfig,
    readiness: &SharedReadinessState,
) -> Result<ClusterHandle, ServerError> {
    let resolver = channel_cluster
        .resolver()
        .cloned()
        .ok_or_else(|| ServerError::ClusterJoin {
            message: "clustering configured but channel supervisor has no distribution resolver"
                .to_owned(),
        })?;
    let scheduler = channel_cluster.supervisor().scheduler();
    let observer = Arc::clone(channel_cluster.observer());
    let readiness = readiness.clone();
    cluster::start(
        &scheduler,
        resolver,
        cluster_config,
        move |sync| {
            // Peers become the console topology's devices and links (R7),
            // set before the initial view is applied.
            let peers = Arc::clone(&observer);
            if !sync.set_peer_listener(Arc::new(move |name, present| peers.peer(name, present))) {
                tracing::error!("cluster peer listener was already installed");
            }
            // The supervisor's one observer forwards to the cluster from now
            // on (SRV-012 R6); it was installed when the supervisor was built.
            if let Err(error) = observer.set_cluster(Arc::new(sync)) {
                tracing::error!(%error, "cluster observer was not installed");
            }
        },
        move || readiness.set_cluster_membership_established(true),
    )
}

fn install_ephemeral_sweep_report(
    services: &LiminalConnectionServices,
    health: &crate::health::HealthServerHandle,
) -> Result<(), ServerError> {
    let report = services
        .ephemeral_sweep_counts()
        .ok_or_else(|| ServerError::HealthEndpoint {
            message: "full services have no ephemeral root startup sweep report".to_owned(),
        })?;
    health.install_ephemeral_sweep_report(report)
}

#[cfg(test)]
mod tests {
    use std::net::SocketAddr;

    use super::{ChannelCluster, ClusterConfig, SharedReadinessState, StartStages, start_cluster};
    use crate::ServerError;
    use crate::health::{ClusterReadiness, ReadinessCondition, ReadinessState, readiness_check};
    use crate::server::connection::services::LiminalConnectionServices;

    /// The started line names each stage with its own length and where it
    /// ended, so a slow start says which stage took the time.
    #[test]
    fn start_stages_read_as_lengths_and_ends() {
        let stages = StartStages {
            began: std::time::Instant::now(),
            marks: vec![("config", 3), ("health", 4), ("services", 2_104)],
        };
        assert_eq!(
            stages.to_string(),
            "config 3 (3), health 1 (4), services 2100 (2104)"
        );
    }

    /// A channel cluster with NO distribution resolver — the shape produced when a
    /// server was built without a `[cluster]` section. `start_cluster` must reject
    /// it before touching `cluster::start`, so its `on_established` hook never runs.
    fn unclustered_channel_cluster() -> Result<ChannelCluster, ServerError> {
        Ok(LiminalConnectionServices::empty()?
            .channel_cluster()
            .clone())
    }

    fn clustered_but_unmet_readiness() -> SharedReadinessState {
        SharedReadinessState::new(ReadinessState::new(
            true,
            true,
            ClusterReadiness::Configured {
                membership_established: false,
            },
        ))
    }

    fn sample_cluster_config() -> Result<ClusterConfig, Box<dyn std::error::Error>> {
        let listen_address: SocketAddr = "127.0.0.1:0".parse()?;
        Ok(ClusterConfig {
            node_name: "node-under-test@127.0.0.1".to_owned(),
            listen_address,
            seed_nodes: Vec::new(),
            cookie: "runtime-test-cookie".to_owned(),
        })
    }

    #[test]
    fn failed_cluster_start_leaves_membership_unestablished()
    -> Result<(), Box<dyn std::error::Error>> {
        let readiness = clustered_but_unmet_readiness();
        let channel_cluster = unclustered_channel_cluster()?;
        let config = sample_cluster_config()?;

        // Missing-resolver failure path: start_cluster returns Err before the
        // established hook can fire.
        let result = start_cluster(&channel_cluster, &config, &readiness);
        assert!(
            result.is_err(),
            "start_cluster must fail without a distribution resolver"
        );

        // The readiness flag stays unset, so /ready still lists the unmet gate.
        let status = readiness_check(&readiness.snapshot());
        assert!(
            !status.ready,
            "readiness must remain not-ready after a failed start"
        );
        assert!(
            status
                .unmet_conditions
                .contains(&ReadinessCondition::ClusterMembershipEstablished),
            "cluster membership gate must stay unmet after a failed start"
        );

        Ok(())
    }
}