liminal-server 0.25.4

use std::io::Write;
use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use crate::ServerError;
use crate::server::listener::{
    acquire_reserve_descriptor, loopback_interrupt_target, prepare_accepted_socket,
    shed_on_fd_exhaustion,
};

use super::checks::{SharedReadinessState, health_check, readiness_check};
use super::reissue::{
    OperatorCredentialReissueOutcome, OperatorCredentialReissueRefusal,
    OperatorCredentialReissueRequest, OperatorCredentialReissuer, SharedOperatorCredentialReissue,
};
use super::unloadable::{SharedUnloadableConversations, UnloadableConversationRecord};

use super::metrics_route;

const HEALTH_PATH: &str = "/health";
const READY_PATH: &str = "/ready";
const METRICS_PATH: &str = "/metrics";
const UNLOADABLE_PATH: &str = "/unloadable-conversations";
/// R18 amendment A7 (§0.18): the operator credential re-issue operation.
const REISSUE_PATH: &str = "/operator/credential-reissue";
/// DUR-021 R4: each shard's reclaim status, read from the store on each request.
const RECLAIM_PATH: &str = "/reclaim";
/// DUR-021 R4: the one operator act that drives a pass to rest.
const QUIESCE_PATH: &str = "/operator/reclaim-quiesce";
const APPLICATION_JSON: &str = "application/json";
const READ_BUFFER_BYTES: usize = 2048;

/// Handle for a running health endpoint server.
///
/// W4 leg 2 (§4.2): the health accept worker BLOCKS in `accept` (kernel-parked,
/// zero idle wakes) rather than spinning a non-blocking poll with a backoff
/// sleep. Shutdown wakes the blocked `accept` with an explicit self-connect
/// interrupt to `interrupt_target` (the bound address, loopback-normalised),
/// mirroring the leg-1 listeners. Shutdown also interrupts an in-flight request
/// read directly via `active_stream`, so a silent client parked in
/// `handle_connection`'s read cannot defer shutdown by its admitted deadline.
#[derive(Debug)]
pub struct HealthServerHandle {
    drafts: admin::DraftRoute,
    grant_stream: Arc<OnceLock<Arc<crate::auth_membership::grant_stream::StreamState>>>,
    lys_root: Arc<OnceLock<crate::auth_lys::root_presence_start::RootSlot>>,
    local_addr: SocketAddr,
    /// Loopback-normalised self-connect target used to interrupt the blocking
    /// `accept` at shutdown.
    interrupt_target: SocketAddr,
    shutdown: Arc<AtomicBool>,
    /// Slot holding a `try_clone` of the request stream the worker is currently
    /// reading, if any. The worker registers it under the lock (with a shutdown
    /// recheck) before blocking on the request read and clears it on completion;
    /// `stop_worker` takes it and `shutdown(Both)`s it to interrupt an in-flight
    /// blocking read directly (TOLD) rather than waiting out the read's admitted
    /// deadline. At most one stream exists at a time (the worker is serial), so
    /// one slot suffices.
    active_stream: Arc<Mutex<Option<TcpStream>>>,
    /// The operator read surface for refused conversation loads, shared with
    /// the worker. The health server binds before any participant handler
    /// exists, so this starts empty and the participant's record is published
    /// into it by [`Self::install_unloadable_record`] once built. Pull-only:
    /// nothing reads it until a request arrives, so it cannot wake the worker.
    unloadable: SharedUnloadableConversations,
    /// The operator WRITE surface for A7 credential re-issue, shared with the
    /// worker. Empty until the participant is built, exactly like `unloadable`
    /// above and for the same reason. Call-driven only: nothing reads it until
    /// a request arrives, so it cannot wake the worker.
    reissue: SharedOperatorCredentialReissue,
    /// DUR-021 R4: the spine store, installed once the services exist, read
    /// on each request to `/reclaim` and driven once by the quiesce route.
    reclaim: super::reclaim_route::SharedReclaimSpine,
    ephemeral_sweep: Arc<OnceLock<liminal::durability::EphemeralSweepCounts>>,
    listen_addresses: Arc<OnceLock<Vec<SocketAddr>>>,
    link_bridges: crate::link_bridge::BridgeRegistry,
    worker: Option<JoinHandle<Result<(), ServerError>>>,
    /// Count of `accept` calls issued by the worker (test observability for the
    /// zero-idle-wakes oracle: on a silent listener this stays at the single
    /// parked call). The worker always maintains the counter; only the host-side
    /// handle for reading it is test-scoped.
    #[cfg(test)]
    accept_attempts: Arc<AtomicU64>,
    /// Count of connections shed under fd exhaustion via the reserve descriptor
    /// (test observability for the shed helper reused from leg 1).
    #[cfg(test)]
    shed_count: Arc<AtomicU64>,
    /// Count of accepted requests the worker has entered (incremented just
    /// before it blocks reading the request). Test observability so a race can
    /// deterministically catch the worker mid-request rather than parked in
    /// `accept`.
    #[cfg(test)]
    requests_entered: Arc<AtomicU64>,
}

impl HealthServerHandle {
    /// Shares the owner's live cursor; a second owner is refused before serving.
    ///
    /// # Errors
    /// Names an attempted replacement of the process's installed grant reader.
    pub fn install_grant_stream(
        &self,
        state: Arc<crate::auth_membership::grant_stream::StreamState>,
    ) -> Result<(), ServerError> {
        self.grant_stream
            .set(state)
            .map_err(|_| ServerError::HealthEndpoint {
                message: "grant_stream_owner_already_installed".into(),
            })
    }
    /// Shares the network-root question (SRV-038), or why a server with Lys
    /// asks none (SRV-039 R2), so `/health` reports it.
    ///
    /// # Errors
    /// Names a second installation.
    pub fn install_lys_root(
        &self,
        root: crate::auth_lys::root_presence_start::RootSlot,
    ) -> Result<(), ServerError> {
        self.lys_root
            .set(root)
            .map_err(|_| ServerError::HealthEndpoint {
                message: "lys_root_already_installed".into(),
            })
    }

    /// Publishes all bound wire addresses after the listeners are opened.
    ///
    /// # Errors
    /// Refuses a second publication by name.
    pub fn install_listen_addresses(&self, addresses: Vec<SocketAddr>) -> Result<(), ServerError> {
        self.listen_addresses
            .set(addresses)
            .map_err(|_| ServerError::HealthEndpoint {
                message: "wire listen addresses are already installed".to_owned(),
            })
    }

    /// The live per-link counters shared with exporting bridge workers.
    #[must_use]
    pub fn link_bridges(&self) -> crate::link_bridge::BridgeRegistry {
        self.link_bridges.clone()
    }
    /// Returns the bound address for the health endpoint server.
    #[must_use]
    pub const fn local_addr(&self) -> SocketAddr {
        self.local_addr
    }

    /// Publishes a participant's unloadable-conversation record onto
    /// `GET /unloadable-conversations`.
    ///
    /// Server startup binds the health endpoint FIRST — liveness has to be
    /// answerable while the rest of the server is still being built — so the
    /// record cannot be a constructor argument. Until this is called the route
    /// answers `participant_installed: false`, which is a different answer from
    /// "nothing is refused" and is reported as such.
    ///
    /// Installing shares the handler's live record rather than copying it: a
    /// refusal recorded after this call is reported by the next scrape. No
    /// thread, timer, or notification is created here.
    /// Publishes the spine store onto `/reclaim` and the quiesce route (DUR-021 R4).
    pub fn install_spine(&self, spine: &Arc<haematite::EventStore>) {
        if !self.reclaim.install(spine) {
            tracing::warn!(
                "the spine was installed on the health endpoint already; the first stands"
            );
        }
    }

    pub fn install_unloadable_record(&self, record: UnloadableConversationRecord) {
        self.unloadable.install(record);
    }

    /// Publishes the participant authority onto
    /// `POST /operator/credential-reissue` (R18 amendment A7, §0.18).
    ///
    /// Same timing constraint as [`Self::install_unloadable_record`]: the
    /// health endpoint binds before the participant exists, so the authority
    /// arrives here once built. Until it does the route answers 503 and says
    /// no participant is installed, which is a different answer from an
    /// unknown identity and is reported as such. No thread, timer, or
    /// notification is created here.
    pub fn install_credential_reissuer(&self, reissuer: Arc<dyn OperatorCredentialReissuer>) {
        self.reissue.install(reissuer);
    }

    /// Installs the immutable startup sweep answer.
    ///
    /// # Errors
    /// Returns an endpoint error if startup attempts a second installation.
    pub fn install_ephemeral_sweep_report(
        &self,
        report: liminal::durability::EphemeralSweepCounts,
    ) -> Result<(), ServerError> {
        self.ephemeral_sweep
            .set(report)
            .map_err(|_| ServerError::HealthEndpoint {
                message: "ephemeral sweep report is already installed".to_owned(),
            })
    }

    /// Stops the health endpoint server and waits for its worker thread to exit.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::HealthEndpoint`] if the worker thread cannot be
    /// joined cleanly or if the server loop recorded a serving error.
    pub fn shutdown(mut self) -> Result<(), ServerError> {
        self.stop_worker()
    }

    fn stop_worker(&mut self) -> Result<(), ServerError> {
        self.shutdown.store(true, Ordering::SeqCst);
        let Some(worker) = self.worker.take() else {
            return Ok(());
        };
        // Interrupt an in-flight request read directly (TOLD): the flag store
        // above happens-before this lock acquire, so any stream the worker
        // registered before observing the flag is taken here and shut down,
        // waking its blocked read at once. `shutdown(Both)` on the clone reaches
        // the same underlying socket the worker is reading (a dup'd fd). If the
        // worker is instead parked in `accept`, the slot is empty and the
        // self-connect below wakes it. The guard is released (statement end)
        // before the socket call, so no lock is held across `shutdown`. A
        // poisoned slot is not trusted: the worker sheds every request once it
        // is poisoned, so no stream is registered to wake and none is taken.
        let in_flight = self
            .active_stream
            .lock()
            .map_or(None, |mut slot| slot.take());
        if let Some(stream) = in_flight
            && let Err(error) = stream.shutdown(Shutdown::Both)
        {
            tracing::debug!(%error, "health endpoint in-flight stream already closed at shutdown");
        }
        // Explicit cross-platform interrupt (mirrors the leg-1 listeners): a
        // single self-connect wakes a blocked `accept`. The worker sees the
        // shutdown flag it observed under the same ordering and sheds the woken
        // (spurious) socket rather than serving it — at most one spurious accept
        // per interrupt. If the worker already exited (a real accept then flag
        // check), the listener is gone and this connect fails fast; the join then
        // returns immediately. A failure while the worker is still parked in
        // `accept` is the other case, and there nothing else will wake it — the
        // join below blocks forever, so it is named rather than swallowed.
        match TcpStream::connect(self.interrupt_target) {
            Ok(waker) => drop(waker),
            Err(error) => tracing::warn!(
                interrupt_target = %self.interrupt_target,
                %error,
                "health endpoint accept interrupt could not connect; the accept \
                 worker will only stop if it has already exited"
            ),
        }

        worker.join().map_err(|_| ServerError::HealthEndpoint {
            message: "health endpoint worker thread terminated unexpectedly".to_owned(),
        })?
    }

    /// `accept` calls issued by the worker (test observability, oracle 8).
    #[cfg(test)]
    fn accept_attempts(&self) -> u64 {
        self.accept_attempts.load(Ordering::SeqCst)
    }

    /// Connections shed under fd exhaustion (test observability).
    #[cfg(test)]
    fn shed_count(&self) -> u64 {
        self.shed_count.load(Ordering::SeqCst)
    }

    /// Accepted requests the worker has entered (test observability).
    #[cfg(test)]
    fn requests_entered(&self) -> u64 {
        self.requests_entered.load(Ordering::SeqCst)
    }
}

impl Drop for HealthServerHandle {
    fn drop(&mut self) {
        if let Err(error) = self.stop_worker() {
            tracing::debug!(%error, "health endpoint shutdown during drop failed");
        }
    }
}

/// Starts the health endpoint HTTP server on a distinct health bind address.
///
/// The returned server handle is independent from the main wire protocol
/// listener. Binding the health endpoint does not mark the main listener ready.
///
/// # Errors
///
/// Returns [`ServerError::HealthEndpoint`] when the health listener cannot bind
/// or cannot report its local address. The listener stays BLOCKING (W4 leg 2):
/// the accept worker kernel-parks in `accept` with zero idle wakes; shutdown
/// wakes it via the self-connect interrupt.
pub fn start_health_server(
    bind_address: SocketAddr,
    readiness: SharedReadinessState,
) -> Result<HealthServerHandle, ServerError> {
    let started_at = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_err(|error| ServerError::HealthEndpoint {
            message: format!("startup clock: {error}"),
        })?;
    let started_at =
        u64::try_from(started_at.as_millis()).map_err(|error| ServerError::HealthEndpoint {
            message: format!("startup clock: {error}"),
        })?;
    let listener =
        TcpListener::bind(bind_address).map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to bind health endpoint at {bind_address}: {error}"),
        })?;
    // The listener stays BLOCKING (W4 leg 2): the accept worker kernel-parks in
    // `accept` with zero idle wakes; shutdown wakes it via the self-connect
    // interrupt below.
    let local_addr = listener
        .local_addr()
        .map_err(|error| ServerError::HealthEndpoint {
            message: format!("failed to inspect health endpoint listener address: {error}"),
        })?;
    let interrupt_target = loopback_interrupt_target(local_addr);
    let shutdown = Arc::new(AtomicBool::new(false));
    let active_stream = Arc::new(Mutex::new(None));
    let unloadable = SharedUnloadableConversations::default();
    let reissue = SharedOperatorCredentialReissue::default();
    let reclaim = super::reclaim_route::SharedReclaimSpine::default();
    let ephemeral_sweep = Arc::new(OnceLock::new());
    let listen_addresses = Arc::new(OnceLock::new());
    let link_bridges = crate::link_bridge::BridgeRegistry::default();
    let grant_stream = Arc::new(OnceLock::new());
    let lys_root = Arc::new(OnceLock::new());
    let drafts = Arc::new(Mutex::new(None));
    let accept_attempts = Arc::new(AtomicU64::new(0));
    let shed_count = Arc::new(AtomicU64::new(0));
    let requests_entered = Arc::new(AtomicU64::new(0));
    let worker_shutdown = Arc::clone(&shutdown);
    let worker_stream = Arc::clone(&active_stream);
    let worker_attempts = Arc::clone(&accept_attempts);
    let worker_shed = Arc::clone(&shed_count);
    let worker_entered = Arc::clone(&requests_entered);
    let served = ServedState {
        started_at,
        drafts: drafts.clone(),
        grant_stream: Arc::clone(&grant_stream),
        lys_root: Arc::clone(&lys_root),
        readiness,
        unloadable: unloadable.clone(),
        reissue: reissue.clone(),
        reclaim: reclaim.clone(),
        ephemeral_sweep: Arc::clone(&ephemeral_sweep),
        listen_addresses: Arc::clone(&listen_addresses),
        link_bridges: link_bridges.clone(),
    };
    let worker = thread::spawn(move || {
        serve(
            &listener,
            &served,
            &worker_shutdown,
            &worker_stream,
            &worker_attempts,
            &worker_shed,
            &worker_entered,
        )
    });

    Ok(HealthServerHandle {
        drafts,
        grant_stream,
        lys_root,
        local_addr,
        interrupt_target,
        shutdown,
        active_stream,
        unloadable,
        reissue,
        reclaim,
        ephemeral_sweep,
        listen_addresses,
        link_bridges,
        worker: Some(worker),
        #[cfg(test)]
        accept_attempts,
        #[cfg(test)]
        shed_count,
        #[cfg(test)]
        requests_entered,
    })
}

/// Everything a request is answered from, carried as one value so the worker's
/// parameter list does not grow a slot per operator surface.
#[derive(Debug, Clone)]
struct ServedState {
    started_at: u64,
    drafts: admin::DraftRoute,
    grant_stream: Arc<OnceLock<Arc<crate::auth_membership::grant_stream::StreamState>>>,
    lys_root: Arc<OnceLock<crate::auth_lys::root_presence_start::RootSlot>>,
    readiness: SharedReadinessState,
    unloadable: SharedUnloadableConversations,
    reissue: SharedOperatorCredentialReissue,
    reclaim: super::reclaim_route::SharedReclaimSpine,
    ephemeral_sweep: Arc<OnceLock<liminal::durability::EphemeralSweepCounts>>,
    listen_addresses: Arc<OnceLock<Vec<SocketAddr>>>,
    link_bridges: crate::link_bridge::BridgeRegistry,
}

#[derive(serde::Serialize)]
struct HealthWithLinks {
    started_at: u64,
    grant_stream: Option<crate::auth_membership::grant_stream::StreamHealth>,
    /// SRV-038: `held`, `absent`, `unknown` or `not_asked` (with its cause, SRV-039); null with no Lys.
    lys_root: Option<&'static str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    lys_root_cause: Option<&'static str>,
    #[serde(flatten, serialize_with = "health_with_lys_asks")]
    health: super::checks::HealthStatus,
    links: std::collections::BTreeMap<String, crate::link_bridge::BridgeCounters>,
    listen_addresses: Vec<SocketAddr>,
}

fn health_with_lys_asks<S: serde::Serializer>(
    health: &super::checks::HealthStatus,
    serializer: S,
) -> Result<S::Ok, S::Error> {
    // Read at serialization so a client installed after the listener starts is
    // included without rebinding the grant-stream owner's health state.
    #[derive(serde::Serialize)]
    struct WithAsks<'a> {
        #[serde(flatten)]
        health: &'a super::checks::HealthStatus,
        lys_ask: crate::auth_lys::live::AskStatistics,
    }
    let lys_ask = crate::auth_lys::live::ask_statistics().map_err(serde::ser::Error::custom)?;
    serde::Serialize::serialize(&WithAsks { health, lys_ask }, serializer)
}

#[derive(serde::Serialize)]
struct ReadyWithStream {
    ready: bool,
    unmet_conditions: Vec<ReadyCondition>,
    grant_stream: Option<crate::auth_membership::grant_stream::StreamHealth>,
}

#[derive(serde::Serialize)]
#[serde(untagged)]
enum ReadyCondition {
    Existing(super::checks::ReadinessCondition),
    Stream(&'static str),
}

mod responses;
mod routes;
mod serve;
use responses::{StatusCode, empty_response, json_response, parse_request_line, response};
use routes::response_for_request;
use serve::serve;

#[cfg(test)]
#[path = "endpoint/tests.rs"]
mod tests;

#[cfg(test)]
mod lys_ask_health_tests {
    use super::*;

    #[test]
    fn health_reports_lys_ask_count_and_running_maximum() -> Result<(), Box<dyn std::error::Error>>
    {
        let health = HealthWithLinks {
            started_at: 1,
            grant_stream: None,
            lys_root: None,
            lys_root_cause: None,
            health: health_check(),
            links: std::collections::BTreeMap::new(),
            listen_addresses: Vec::new(),
        };
        let value = serde_json::to_value(health)?;
        assert!(value["lys_ask"]["count"].is_u64());
        assert!(value["lys_ask"]["max_elapsed_ms"].is_f64());
        assert_eq!(value["status"], "healthy");
        assert!(value["grant_stream"].is_null());
        Ok(())
    }
}

pub(crate) mod admin;