liminal-server 0.24.1

use std::net::SocketAddr;
use std::path::PathBuf;
use std::time::Duration;

use crate::ServerError;

/// Declarative configuration for the standalone liminal server wrapper.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ServerConfig {
    /// Socket address where the standalone server will listen for client traffic.
    pub listen_address: SocketAddr,
    /// Additional named wire addresses on an encrypted network, sharing admission.
    #[serde(default)]
    pub extra_listen_addresses: Vec<SocketAddr>,
    /// Socket address where the health endpoint server will listen for probes.
    pub health_listen_address: SocketAddr,
    /// IGNORED SINCE 0.14.3. Kept only so an existing config file still loads.
    ///
    /// This was the maximum time shutdown would wait for existing connections
    /// to hang up by themselves before closing them. Nothing reads it now.
    ///
    /// A drain exists to let in-flight requests finish. In this server every
    /// write is durable and flushed BEFORE it is acknowledged, so at the
    /// instant shutdown begins there is no in-flight request to finish and
    /// nothing for the wait to buy. What it did in practice, on a deployment
    /// whose connections are long-lived idle clients that never hang up on
    /// their own, was spend its whole budget on every restart and then close
    /// the connections anyway — 5 s per restart, measured on Tom Whiting's
    /// estate on 2026-09-14. Shutdown now stops accepting, closes every
    /// connection itself (an orderly close each peer reads as a shutdown), and
    /// flushes durable state, with no timer in that path.
    ///
    /// Any value parses, including zero. Setting it changes nothing.
    pub drain_timeout_ms: u64,
    /// Channel topology definitions declared by the operator.
    pub channels: Vec<ChannelDef>,
    /// Declarative routing rules that connect configured channels.
    pub routing_rules: Vec<RoutingRuleDef>,
    /// Optional filesystem location for durable server state.
    ///
    /// A fresh boot surfaces `ConfigValidation` unless this path's PARENT
    /// directory already exists: the haematite engine creates the store dir
    /// exactly one level below a pre-existing parent it can fence (never
    /// `create_dir_all` — deliberate upstream hardening). Create the parent;
    /// the engine fences into it.
    pub persistence_path: Option<PathBuf>,
    /// Optional beamr distribution cluster membership configuration.
    pub cluster: Option<ClusterConfig>,
    /// Optional connection authentication configuration.
    ///
    /// When present, every client `Connect` handshake must carry a matching
    /// `auth_token`; when absent the server is open (byte-identical to the
    /// pre-auth behaviour). Not an ACL system — a single shared bearer token.
    #[serde(default)]
    pub auth: Option<AuthConfig>,
    /// Service construction profile. Absent `[services]` (or an absent `profile`
    /// key within it) defaults to `"full"`, so existing deployments build exactly
    /// what they build today.
    #[serde(default)]
    pub services: ServicesConfig,
    /// Operational bounds (§5). Absent `[limits]` (or any absent key within it)
    /// defaults to the certifying-pair-signed numbers, so an operator who sets
    /// nothing still runs bounded — "unlimited-by-silence is no longer a legal
    /// state" (§5). Every value is a hard cap enforced by a typed refusal; a
    /// zero (or otherwise invalid) value is a config validation error, never a
    /// silent "unlimited".
    #[serde(default)]
    pub limits: LimitsConfig,
    /// Optional WebSocket transport acceptor (LP-WS-TRANSPORT R1).
    ///
    /// When present the server binds a sibling WebSocket listener carrying the
    /// canonical liminal wire protocol (one binary message per canonical frame)
    /// alongside the main TCP listener. When absent NO HTTP/WebSocket listener
    /// is started and the server behaves byte-identically to the pre-WebSocket
    /// build. Every field inside is a deployment decision; the origin allow-list
    /// FAILS CLOSED (an absent or empty list refuses every Origin-bearing
    /// upgrade) and the keepalive ping interval is disabled unless explicitly
    /// configured.
    #[serde(default)]
    pub websocket: Option<WebSocketConfig>,
    /// Participant lifecycle activation (LP gap closure, Part B).
    ///
    /// When present the server installs the production participant semantic
    /// handler and advertises the participant capability bit on every
    /// connection. Every field inside is REQUIRED and carries NO default:
    /// participant lifecycle values are deployment decisions, and an absent
    /// field is a typed startup error rather than an assumed number. When the
    /// section is absent the participant capability stays disabled and the
    /// server behaves byte-identically to the pre-activation build.
    #[serde(default)]
    pub participant: Option<ParticipantConfig>,
}

impl ServerConfig {
    /// Returns the configured drain timeout, which shutdown IGNORES since
    /// 0.14.3.
    ///
    /// Retained so an embedder that reads it keeps compiling, and so the value
    /// an operator wrote can still be named in the shutdown log line that
    /// reports it ignored. See [`ServerConfig::drain_timeout_ms`].
    #[must_use]
    pub const fn drain_timeout(&self) -> Duration {
        Duration::from_millis(self.drain_timeout_ms)
    }
}

/// Declarative channel definition loaded from server configuration.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ChannelDef {
    /// Unique channel name used by routing rules and operators.
    pub name: String,
    /// Filesystem path to a JSON Schema document that validates every message
    /// published to this channel.
    ///
    /// The path is resolved relative to the directory containing the config file
    /// (absolute paths are used verbatim). Config validation reads and parses the
    /// referenced document and stores the result in [`Self::loaded_schema`]; a
    /// missing file, an unreadable file, or a file that is not valid JSON is an
    /// accumulated validation error that stops startup.
    ///
    /// `None` means the channel has no schema: it keeps the permissive empty
    /// schema (`{}`) that accepts any JSON payload.
    #[serde(default)]
    pub schema_ref: Option<PathBuf>,
    /// Whether this channel requires durable persistence.
    pub durable: bool,
    /// Schema document loaded and parsed from [`Self::schema_ref`] during config
    /// validation. Populated only by [`crate::config::validate`]; a directly
    /// constructed [`ChannelDef`] that skips validation carries `None` here and is
    /// therefore built with the permissive empty schema regardless of
    /// [`Self::schema_ref`]. Never deserialized from the config file.
    #[serde(skip)]
    pub loaded_schema: Option<LoadedSchema>,
}

/// A channel's JSON Schema document as loaded from disk during config validation.
///
/// Carries both the parsed document (fed to the validation engine when the channel
/// is built) and the raw file bytes (hashed into the protocol schema id advertised
/// at subscribe time, so an SDK deriving ids from the same schema bytes converges).
#[derive(Debug, Clone)]
pub struct LoadedSchema {
    /// Raw bytes of the schema file, hashed to derive the protocol schema id.
    pub bytes: Vec<u8>,
    /// Parsed JSON Schema document, fed to the channel's validation engine.
    pub document: serde_json::Value,
}

/// Declarative routing rule definition loaded from server configuration.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RoutingRuleDef {
    /// Source channel name from which messages are routed.
    pub source_channel: String,
    /// Target channel name to which matching messages are routed.
    pub target_channel: String,
    /// Optional predicate expression that filters routed messages.
    pub predicate: Option<String>,
}

/// Public beamr distribution cookie, refused during configuration validation.
/// Mirrors beamr's own [`beamr::distribution::DEFAULT_COOKIE`].
pub const DEFAULT_COOKIE: &str = "beamr-cookie";

/// Beamr distribution for copies of one server on a trusted network.
///
/// Federation provides the network between servers; a cluster joins replicas
/// of the same server using an explicitly configured shared secret cookie.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterConfig {
    /// Unique node name advertised to the beamr distribution cluster.
    pub node_name: String,
    /// Socket address this node binds for inbound distribution links from peers.
    ///
    /// This is distinct from [`ServerConfig::listen_address`] (the client wire
    /// port): a clustered node listens on two ports — one for clients, one for
    /// peer distribution traffic.
    pub listen_address: SocketAddr,
    /// Seed node socket addresses used to join an existing cluster.
    pub seed_nodes: Vec<SocketAddr>,
    /// Required secret distribution handshake cookie. Every node in a cluster
    /// must explicitly set the same cookie or the OTP handshake is rejected.
    /// An omitted cookie fails deserialization; an empty cookie or the public
    /// [`DEFAULT_COOKIE`] fails configuration validation.
    pub cookie: String,
}

/// Connection authentication configuration.
///
/// A single shared bearer token compared (constant-time) against the `auth_token`
/// carried on every client `Connect` handshake. This is the table-stakes access
/// gate, not an ACL system: one token grants full access. Its absence (no `[auth]`
/// section) is open mode (ACCESS-010): the exchange listens on loopback only,
/// admits every connection whatever credential it presents, and asks Lys
/// nothing; `[auth]` makes it gated, and a gated exchange never falls back to
/// open.
#[derive(Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuthConfig {
    /// Shared secret token a client must present in its `Connect` handshake. Must
    /// be non-empty when the `[auth]` section is present (an empty token is a
    /// config validation error, since it would gate nothing).
    pub token: String,
    /// Optional registry-minted pass verifier, additive beside the bearer.
    #[serde(default)]
    pub pass: Option<PassConfig>,
    /// The Lys pass as a second credential beside the wire pass (ACCESS-001
    /// R2): when set, a `Connect` carrying a Lys pass record is verified
    /// offline against the issuer's published keys and decided per message.
    #[serde(default)]
    pub lys: Option<LysConfig>,
    /// Refuse every worker registration from a connection that proved no
    /// holder (ACCESS-004 R3): a bearer connection, or one before any pass.
    /// Off by default, so a worker that dials with the bearer, as aion's does,
    /// keeps registering; turning it on is the operator's act, and it needs
    /// `[auth.pass]`, without which no connection could prove a holder.
    #[serde(default)]
    pub require_verified_worker: bool,
}

/// The token is a credential: its `Debug` names its length only
/// (ACCESS-004 R5).
impl std::fmt::Debug for AuthConfig {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("AuthConfig")
            .field("token", &crate::auth_pass::ByteLength(self.token.len()))
            .field("pass", &self.pass)
            .field("lys", &self.lys)
            .field("require_verified_worker", &self.require_verified_worker)
            .finish()
    }
}

/// Which of the two access modes a configuration names (ACCESS-010 R1).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AccessMode {
    /// No `[auth]`: loopback only, every connection admitted, nothing asked of Lys.
    Open,
    /// `[auth]` present: a bearer, a pass verifier or a Lys admitter decides.
    Gated,
}

impl std::fmt::Display for AccessMode {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(match self {
            Self::Open => "open",
            Self::Gated => "gated",
        })
    }
}

impl ServerConfig {
    /// The access mode: open with no `[auth]`, gated with one.
    #[must_use]
    pub const fn access_mode(&self) -> AccessMode {
        if self.auth.is_some() {
            AccessMode::Gated
        } else {
            AccessMode::Open
        }
    }
}

/// Registry pass verification configuration. Every field is required when present.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PassConfig {
    /// Registry Ed25519 verifying key as exactly 64 hexadecimal characters.
    pub registry_verifying_key: String,
    /// Maximum admitted server-clock skew in seconds.
    pub maximum_clock_skew_seconds: u64,
}

/// WebSocket transport acceptor configuration (`[websocket]`, LP-WS-TRANSPORT R1).
///
/// The sibling WebSocket route is an explicit opt-in: the section itself must be
/// present for any HTTP/WebSocket listener to start, and inside it the listen
/// address and the single exact upgrade path are required with no defaults.
///
/// The deployment TLS contract (tear ruling Q1) is raw `ws://` behind a named
/// TLS-terminating proxy that owns public `wss://` and certificates; liminal
/// grows no TLS stack. Origin validation nonetheless belongs to this acceptor:
/// [`Self::allowed_origins`] is the explicit allow-list checked on every
/// Origin-bearing upgrade, and there is NO default list — absent or empty
/// configuration fails closed for browser-origin upgrades while a native client
/// that sends no `Origin` header may still upgrade (F6).
///
/// OPERATOR NOTE — the same deployment contract covers the pre-upgrade window
/// (domain-owner ruling, 2026-07-18): the fronting proxy must ALSO enforce
/// pre-upgrade read timeouts, handshake concurrency limits, and connection
/// rate limits. Between TCP accept and a completed WebSocket upgrade this
/// listener does not count the socket against `[limits] max_connections` and
/// applies no read deadline of its own (only the fixed request-head size
/// bound), so a deployment that exposes this port without the named proxy is
/// out of contract on untrusted networks. A named handshake read-deadline
/// config plus an in-flight handshake cap derived from the configured
/// `max_connections` value is the ledgered post-demo hardening.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WebSocketConfig {
    /// Socket address the WebSocket acceptor binds. Required; distinct from the
    /// main wire listener, the health listener, and any cluster listener.
    pub listen_address: SocketAddr,
    /// The single exact HTTP request path that accepts WebSocket upgrades.
    /// Required; must start with `/`. Every other path — and every ordinary
    /// HTTP request — receives a small fixed non-success response and closes.
    pub path: String,
    /// Explicit browser-origin allow-list checked on every Origin-bearing
    /// upgrade (F6). Entries are compared byte-exact against the request's
    /// serialized `Origin` header value (RFC 6454 ASCII serialization, e.g.
    /// `https://app.example.com`). Absent or empty means NO browser origin is
    /// accepted (fail closed); native clients sending no `Origin` header are
    /// unaffected.
    #[serde(default)]
    pub allowed_origins: Vec<String>,
    /// CON-014 R2: the console served beside this exchange, admitted at every
    /// address the machine is reached by. An Origin-bearing upgrade is also
    /// admitted when the Origin is `http://host[:port]` or `https://host[:port]`,
    /// its port (the explicit one, else the scheme's own) is listed here, and
    /// its host is the host of the request's own `Host` header. A site at
    /// another host on the same port is refused as today. Absent or empty adds
    /// nothing.
    #[serde(default)]
    pub allowed_origin_ports: Vec<u16>,
    /// Q-A transport-liveness keepalive: the server-side WebSocket Ping
    /// interval in milliseconds. This is a precise LAW-1 carve-out — liveness
    /// pings never mint application events, never re-arm application state, and
    /// never serve as a source of truth; failure detection remains the socket's
    /// typed terminal events. The bound is one ping per interval per
    /// connection, so the idle cost is `interval x connection-count`. Absent
    /// means pings are DISABLED, accepting proxy-idle-disconnect churn as the
    /// documented consequence. A configured zero is a validation error.
    #[serde(default)]
    pub ping_interval_ms: Option<u64>,
}

/// Service construction profile selection (D2).
///
/// The `profile` value is carried as a raw string here rather than a typed enum so
/// an unrecognised value is a config *validation* error with a helpful message
/// (via [`Self::profile`]) rather than an opaque deserialization failure — matching
/// how every other semantic config check surfaces. Absent `profile` defaults to
/// `"full"`.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ServicesConfig {
    /// Construction profile: `"full"` (the default, unchanged behaviour) or
    /// `"worker-front-door"` (capability-scoped worker deployments).
    #[serde(default = "default_service_profile")]
    pub profile: String,
    /// Row bounds for the server-owned durable streams.
    #[serde(default)]
    pub self_streams: SelfStreamsConfig,
}

/// Retained row bounds for each server-owned durable stream.
#[derive(Debug, Clone, Copy, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SelfStreamsConfig {
    /// Maximum retained cost records.
    #[serde(default = "default_cost_rows")]
    pub cost_rows: u64,
    /// Maximum retained lifecycle and connection events.
    #[serde(default = "default_events_rows")]
    pub events_rows: u64,
}

impl Default for SelfStreamsConfig {
    fn default() -> Self {
        Self {
            cost_rows: default_cost_rows(),
            events_rows: default_events_rows(),
        }
    }
}

const fn default_cost_rows() -> u64 {
    10_000
}

const fn default_events_rows() -> u64 {
    100_000
}

impl Default for ServicesConfig {
    fn default() -> Self {
        Self {
            profile: default_service_profile(),
            self_streams: SelfStreamsConfig::default(),
        }
    }
}

impl ServicesConfig {
    /// Resolves the raw `profile` string into a typed [`ServiceProfile`].
    ///
    /// # Errors
    /// Returns [`ServerError::ConfigValidation`] when the value is not a recognised
    /// profile.
    pub fn profile(&self) -> Result<ServiceProfile, ServerError> {
        ServiceProfile::parse(&self.profile)
    }
}

fn default_service_profile() -> String {
    ServiceProfile::FULL.to_owned()
}

/// Which connection-services adapter the server constructs (D2).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ServiceProfile {
    /// Full channel/conversation/durability services — the default. Constructs the
    /// haematite store, channel supervisor, conversation supervisor, and dedup
    /// cache exactly as before.
    Full,
    /// Capability-scoped worker front door: the connection supervisor only, with no
    /// channel/conversation/haematite machinery. Backs worker registration,
    /// correlated push/reply, and notifier-consumed reserved publishes; ordinary
    /// channel and conversation frames are rejected with a typed error frame.
    WorkerFrontDoor,
}

impl ServiceProfile {
    /// Config value selecting the full-service profile.
    pub const FULL: &'static str = "full";
    /// Config value selecting the worker-front-door profile.
    pub const WORKER_FRONT_DOOR: &'static str = "worker-front-door";

    /// Parses a `[services] profile` value into a typed profile.
    ///
    /// # Errors
    /// Returns [`ServerError::ConfigValidation`] for any value other than
    /// [`Self::FULL`] or [`Self::WORKER_FRONT_DOOR`].
    pub fn parse(value: &str) -> Result<Self, ServerError> {
        match value {
            Self::FULL => Ok(Self::Full),
            Self::WORKER_FRONT_DOOR => Ok(Self::WorkerFrontDoor),
            other => Err(ServerError::ConfigValidation {
                message: format!(
                    "services.profile: unknown profile '{other}'; expected \"{}\" or \"{}\"",
                    Self::FULL,
                    Self::WORKER_FRONT_DOOR
                ),
            }),
        }
    }
}

mod limits;
mod lys;
mod participant;
pub use crate::auth_membership::grant_stream::GrantStreamConfig;
pub use limits::LimitsConfig;
pub use lys::*;
pub use participant::{ParticipantConfig, PruningConfig};