saddle-framework 0.3.5

//! Framework-owned process entry for the fixed ProfuseGW application.

use std::{
    any::Any,
    future::Future,
    net::SocketAddr,
    path::{Component, Path, PathBuf},
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use serde::{Deserialize, de::DeserializeOwned};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::{TcpListener, TcpStream},
};

use crate::{ErrorKind, Result, SaddleError};

const MAX_HEAD_BYTES: usize = 16 * 1024;
const IO_DEADLINE: Duration = Duration::from_secs(5);

/// Executable deployment configuration for one generated Saddle application.
///
/// The ordinary TOML file and its declared secret references are resolved and
/// frozen once, before any listener is published.
pub struct ProcessConfig<B = ()> {
    listen: SocketAddr,
    management_bind: SocketAddr,
    profusecontract_authority_template: saddle_boundary::ProfuseContractAuthorityTemplate,
    business: BusinessConfig<B>,
    database: Option<saddle_db::DatabaseStartupInjection>,
    logging: saddle_observability::FileLoggingConfig,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct FileConfig<B> {
    framework: FrameworkConfig,
    secrets: SecretConfig,
    #[serde(default)]
    database: Option<DatabaseFileConfig>,
    business: Option<B>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct FrameworkConfig {
    listen: String,
    management: ManagementFileConfig,
    profusecontract: ProfuseContractFileConfig,
    #[serde(default)]
    observability: ObservabilityFileConfig,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ManagementFileConfig {
    bind: String,
}

#[derive(Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct ObservabilityFileConfig {
    #[serde(default)]
    logging: LoggingFileConfig,
}

#[derive(Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct LoggingFileConfig {
    directory: Option<PathBuf>,
    #[serde(default)]
    rotation: LoggingRotation,
}

#[derive(Default, Deserialize)]
#[serde(rename_all = "lowercase")]
enum LoggingRotation {
    #[default]
    Daily,
    Hourly,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct ProfuseContractFileConfig {
    authority: String,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct SecretConfig {
    #[serde(default)]
    database_url_env: Option<String>,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct DatabaseFileConfig {
    mapping_dir: PathBuf,
}

/// Read-only application configuration owned and frozen by the framework.
pub struct BusinessConfig<B> {
    value: Arc<B>,
    directory: Arc<PathBuf>,
}

impl<B> Clone for BusinessConfig<B> {
    fn clone(&self) -> Self {
        Self {
            value: Arc::clone(&self.value),
            directory: Arc::clone(&self.directory),
        }
    }
}

impl<B> BusinessConfig<B> {
    /// Borrows the strongly typed application configuration.
    pub fn get(&self) -> &B {
        &self.value
    }

    /// Resolves a normalized relative application path against `saddle.toml`.
    pub fn resolve_path(&self, relative: impl AsRef<Path>) -> Result<PathBuf> {
        let relative = relative.as_ref();
        if relative.as_os_str().is_empty()
            || relative.is_absolute()
            || relative
                .components()
                .any(|component| !matches!(component, Component::Normal(_)))
        {
            return Err(process_error("saddle.process.business_path_invalid"));
        }
        Ok(self.directory.join(relative))
    }
}

impl BusinessConfig<()> {
    /// Creates the sole no-business-configuration value for a test harness.
    pub fn unit() -> Self {
        Self {
            value: Arc::new(()),
            directory: Arc::new(PathBuf::new()),
        }
    }
}

impl<B: DeserializeOwned + 'static> ProcessConfig<B> {
    /// Parses the sole supported process command line: `--config <path>`.
    pub fn from_args() -> Result<Self> {
        let mut args = std::env::args_os();
        let _program = args.next();
        match (args.next(), args.next(), args.next()) {
            (Some(flag), Some(path), None) if flag == "--config" => Self::load(path),
            _ => Err(process_error("saddle.process.arguments_invalid")),
        }
    }

    /// Loads and validates all process configuration before listener startup.
    pub fn load(path: impl AsRef<Path>) -> Result<Self> {
        let path = path.as_ref();
        let source =
            std::fs::read(path).map_err(|_| process_error("saddle.process.config_unavailable"))?;
        let source = std::str::from_utf8(&source)
            .map_err(|_| process_error("saddle.process.config_invalid"))?;
        let file: FileConfig<B> =
            toml::from_str(source).map_err(|_| process_error("saddle.process.config_invalid"))?;
        let directory = path
            .parent()
            .filter(|parent| !parent.as_os_str().is_empty())
            .unwrap_or_else(|| Path::new("."))
            .canonicalize()
            .map_err(|_| process_error("saddle.process.config_directory_invalid"))?;
        let database = match (file.secrets.database_url_env, file.database) {
            (None, None) => None,
            (Some(connection_environment), Some(database)) => Some(
                saddle_db::DatabaseStartupInjection::load(
                    &directory,
                    &connection_environment,
                    &database.mapping_dir,
                )
                .map_err(|_| process_error("saddle.process.database_config_invalid"))?,
            ),
            _ => return Err(process_error("saddle.process.database_config_incomplete")),
        };
        let business = match file.business {
            Some(business) => business,
            None => *(Box::new(()) as Box<dyn Any>)
                .downcast::<B>()
                .map_err(|_| process_error("saddle.process.business_config_missing"))?,
        };
        Ok(Self {
            listen: file
                .framework
                .listen
                .parse()
                .map_err(|_| process_error("saddle.process.listen_invalid"))?,
            management_bind: file
                .framework
                .management
                .bind
                .parse()
                .map_err(|_| process_error("saddle.process.management_bind_invalid"))?,
            profusecontract_authority_template:
                saddle_boundary::ProfuseContractAuthorityTemplate::new(
                    file.framework.profusecontract.authority,
                )
                .map_err(|_| process_error("saddle.process.authority_template_invalid"))?,
            business: BusinessConfig {
                value: Arc::new(business),
                directory: Arc::new(directory),
            },
            database,
            logging: saddle_observability::FileLoggingConfig::new(
                file.framework
                    .observability
                    .logging
                    .directory
                    .unwrap_or_else(|| PathBuf::from("./logs")),
                match file.framework.observability.logging.rotation {
                    LoggingRotation::Daily => saddle_observability::Rotation::Daily,
                    LoggingRotation::Hourly => saddle_observability::Rotation::Hourly,
                },
            ),
        })
    }
}

#[doc(hidden)]
pub fn run_profusegw_process<B, C, Connect, ConnectFuture, Dispatch, DispatchFuture>(
    config: ProcessConfig<B>,
    adapter: saddle_boundary::ingress::ProfuseGwListenerAdapter,
    connect: Connect,
    dispatch: Dispatch,
) -> Result<()>
where
    B: Send + Sync + 'static,
    C: Clone + Send + Sync + 'static,
    Connect: FnOnce(
            saddle_boundary::ProfuseContractAuthorityTemplate,
            Option<saddle_db::DatabaseStartupInjection>,
        ) -> ConnectFuture
        + Send
        + 'static,
    ConnectFuture: Future<Output = Result<C>> + Send + 'static,
    Dispatch: Fn(saddle_boundary::ingress::AcceptedIngress, C, BusinessConfig<B>) -> DispatchFuture
        + Send
        + Sync
        + 'static,
    DispatchFuture: Future<Output = Result<Vec<u8>>> + Send + 'static,
{
    saddle_runtime::Application::run_with(move || async move {
        let observer = saddle_observability::init_file(Default::default(), config.logging)
            .map_err(|_| process_error("saddle.process.observability_init_failed"))?
            .clone();
        let deployment =
            connect(config.profusecontract_authority_template, config.database).await?;
        let (listener, management_listener) =
            bind_process_listeners(config.listen, config.management_bind).await?;
        let mut application = saddle_runtime::Application::new();
        application.register(observer.clone())?;
        application.register(ManagementEntry::new(
            management_listener,
            application.health(),
        ))?;
        let lifecycle = application.request_lifecycle();
        application.register(ProcessEntry::new(
            listener,
            adapter,
            deployment,
            dispatch,
            config.business,
            lifecycle,
            observer,
        ))?;
        Ok(application)
    })
}

async fn bind_process_listeners(
    business: SocketAddr,
    management: SocketAddr,
) -> Result<(TcpListener, TcpListener)> {
    let business = TcpListener::bind(business)
        .await
        .map_err(|_| process_error("saddle.process.listener_bind_failed"))?;
    let management = TcpListener::bind(management)
        .await
        .map_err(|_| process_error("saddle.process.management_bind_failed"))?;
    Ok((business, management))
}

struct ManagementEntry {
    listener: Mutex<Option<TcpListener>>,
    health: saddle_runtime::ApplicationHealth,
    running: Mutex<Option<Running>>,
}

impl ManagementEntry {
    fn new(listener: TcpListener, health: saddle_runtime::ApplicationHealth) -> Self {
        Self {
            listener: Mutex::new(Some(listener)),
            health,
            running: Mutex::new(None),
        }
    }
}

impl saddle_runtime::ComponentLifecycle for ManagementEntry {
    fn name(&self) -> &'static str {
        "management-health-entry"
    }

    fn start(&self) -> saddle_runtime::LifecycleFuture<'_> {
        Box::pin(async move {
            let listener = self
                .listener
                .lock()
                .unwrap()
                .take()
                .ok_or_else(|| process_error("saddle.process.management_listener_consumed"))?;
            let stop = Arc::new(AtomicBool::new(false));
            let worker = tokio::spawn(run_management_entry(
                listener,
                self.health.clone(),
                Arc::clone(&stop),
            ));
            *self.running.lock().unwrap() = Some(Running { stop, worker });
            Ok(())
        })
    }

    fn shutdown(&self) -> saddle_runtime::LifecycleFuture<'_> {
        Box::pin(async move {
            let running = self
                .running
                .lock()
                .unwrap()
                .take()
                .ok_or_else(|| process_error("saddle.process.management_entry_not_started"))?;
            running.stop.store(true, Ordering::Release);
            running
                .worker
                .await
                .map_err(|_| process_error("saddle.process.management_entry_failed"))
        })
    }
}

async fn run_management_entry(
    listener: TcpListener,
    health: saddle_runtime::ApplicationHealth,
    stop: Arc<AtomicBool>,
) {
    while !stop.load(Ordering::Acquire) {
        tokio::select! {
            accepted = listener.accept() => {
                let Ok((mut socket, _)) = accepted else { continue };
                serve_management(&mut socket, &health).await;
            }
            () = tokio::time::sleep(Duration::from_millis(1)) => {}
        }
    }
}

async fn serve_management(socket: &mut TcpStream, health: &saddle_runtime::ApplicationHealth) {
    let response = match tokio::time::timeout(IO_DEADLINE, read_management_request(socket)).await {
        Ok(Ok(request)) => management_response(&request, health.snapshot()),
        _ => status(400),
    };
    let _ = tokio::time::timeout(IO_DEADLINE, async {
        socket.write_all(&response).await?;
        socket.shutdown().await
    })
    .await;
}

#[derive(Debug, Eq, PartialEq)]
struct ManagementRequest {
    method: String,
    path: String,
}

async fn read_management_request(socket: &mut TcpStream) -> std::io::Result<ManagementRequest> {
    let mut bytes = Vec::with_capacity(1024);
    let head_end = loop {
        if bytes.len() >= MAX_HEAD_BYTES {
            return Err(std::io::ErrorKind::InvalidData.into());
        }
        let mut chunk = [0_u8; 1024];
        let read = socket.read(&mut chunk).await?;
        if read == 0 {
            return Err(std::io::ErrorKind::UnexpectedEof.into());
        }
        bytes.extend_from_slice(&chunk[..read]);
        if let Some(index) = bytes.windows(4).position(|value| value == b"\r\n\r\n") {
            break index + 4;
        }
    };
    if bytes.len() != head_end {
        return Err(std::io::ErrorKind::InvalidData.into());
    }
    let head =
        std::str::from_utf8(&bytes[..head_end - 4]).map_err(|_| std::io::ErrorKind::InvalidData)?;
    let mut lines = head.split("\r\n");
    let mut request_line = lines
        .next()
        .ok_or(std::io::ErrorKind::InvalidData)?
        .split_ascii_whitespace();
    let method = request_line.next().ok_or(std::io::ErrorKind::InvalidData)?;
    let path = request_line.next().ok_or(std::io::ErrorKind::InvalidData)?;
    if request_line.next() != Some("HTTP/1.1") || request_line.next().is_some() {
        return Err(std::io::ErrorKind::InvalidData.into());
    }
    let mut content_length_seen = false;
    for line in lines {
        let (name, value) = line
            .split_once(':')
            .ok_or(std::io::ErrorKind::InvalidData)?;
        if name.eq_ignore_ascii_case("content-length") {
            if content_length_seen || value.trim() != "0" {
                return Err(std::io::ErrorKind::InvalidData.into());
            }
            content_length_seen = true;
        }
        if name.eq_ignore_ascii_case("transfer-encoding") {
            return Err(std::io::ErrorKind::InvalidData.into());
        }
    }
    Ok(ManagementRequest {
        method: method.to_owned(),
        path: path.to_owned(),
    })
}

fn management_response(
    request: &ManagementRequest,
    snapshot: saddle_runtime::ApplicationHealthSnapshot,
) -> Vec<u8> {
    if request.method != "GET" {
        return status(405);
    }
    match request.path.as_str() {
        "/live" => json_status(
            if snapshot.is_live() { 200 } else { 503 },
            "live",
            snapshot.is_live(),
        ),
        "/ready" => json_status(
            if snapshot.is_ready() { 200 } else { 503 },
            "ready",
            snapshot.is_ready(),
        ),
        _ => status(404),
    }
}

fn json_status(code: u16, field: &str, value: bool) -> Vec<u8> {
    let body = format!(r#"{{"{field}":{value}}}"#);
    format!(
        "HTTP/1.1 {code} Health\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .into_bytes()
}

struct Running {
    stop: Arc<AtomicBool>,
    worker: tokio::task::JoinHandle<()>,
}

struct ProcessEntry<B, C, Dispatch> {
    listener: Mutex<Option<TcpListener>>,
    adapter: Mutex<Option<saddle_boundary::ingress::ProfuseGwListenerAdapter>>,
    deployment: Mutex<Option<C>>,
    dispatch: Mutex<Option<Dispatch>>,
    business: BusinessConfig<B>,
    lifecycle: saddle_runtime::RequestLifecycle,
    observer: saddle_observability::Observer,
    running: Mutex<Option<Running>>,
}

impl<B, C, Dispatch> ProcessEntry<B, C, Dispatch> {
    fn new(
        listener: TcpListener,
        adapter: saddle_boundary::ingress::ProfuseGwListenerAdapter,
        deployment: C,
        dispatch: Dispatch,
        business: BusinessConfig<B>,
        lifecycle: saddle_runtime::RequestLifecycle,
        observer: saddle_observability::Observer,
    ) -> Self {
        Self {
            listener: Mutex::new(Some(listener)),
            adapter: Mutex::new(Some(adapter)),
            deployment: Mutex::new(Some(deployment)),
            dispatch: Mutex::new(Some(dispatch)),
            business,
            lifecycle,
            observer,
            running: Mutex::new(None),
        }
    }
}

impl<B, C, Dispatch, DispatchFuture> saddle_runtime::ComponentLifecycle
    for ProcessEntry<B, C, Dispatch>
where
    B: Send + Sync + 'static,
    C: Clone + Send + Sync + 'static,
    Dispatch: Fn(saddle_boundary::ingress::AcceptedIngress, C, BusinessConfig<B>) -> DispatchFuture
        + Send
        + Sync
        + 'static,
    DispatchFuture: Future<Output = Result<Vec<u8>>> + Send + 'static,
{
    fn name(&self) -> &'static str {
        "profusegw-process-entry"
    }

    fn start(&self) -> saddle_runtime::LifecycleFuture<'_> {
        Box::pin(async move {
            let listener = self
                .listener
                .lock()
                .unwrap()
                .take()
                .ok_or_else(|| process_error("saddle.process.listener_consumed"))?;
            let adapter = self
                .adapter
                .lock()
                .unwrap()
                .take()
                .ok_or_else(|| process_error("saddle.process.adapter_consumed"))?;
            let deployment = self
                .deployment
                .lock()
                .unwrap()
                .take()
                .ok_or_else(|| process_error("saddle.process.deployment_consumed"))?;
            let dispatch = self
                .dispatch
                .lock()
                .unwrap()
                .take()
                .ok_or_else(|| process_error("saddle.process.dispatch_consumed"))?;
            let stop = Arc::new(AtomicBool::new(false));
            let worker = tokio::spawn(run_entry(
                listener,
                adapter,
                deployment,
                dispatch,
                self.business.clone(),
                self.lifecycle.clone(),
                self.observer.clone(),
                Arc::clone(&stop),
            ));
            *self.running.lock().unwrap() = Some(Running { stop, worker });
            Ok(())
        })
    }

    fn shutdown(&self) -> saddle_runtime::LifecycleFuture<'_> {
        Box::pin(async move {
            let running = self
                .running
                .lock()
                .unwrap()
                .take()
                .ok_or_else(|| process_error("saddle.process.entry_not_started"))?;
            running.stop.store(true, Ordering::Release);
            running
                .worker
                .await
                .map_err(|_| process_error("saddle.process.entry_failed"))
        })
    }
}

async fn run_entry<B, C, Dispatch, DispatchFuture>(
    listener: TcpListener,
    adapter: saddle_boundary::ingress::ProfuseGwListenerAdapter,
    deployment: C,
    dispatch: Dispatch,
    business: BusinessConfig<B>,
    lifecycle: saddle_runtime::RequestLifecycle,
    observer: saddle_observability::Observer,
    stop: Arc<AtomicBool>,
) where
    C: Clone,
    Dispatch: Fn(saddle_boundary::ingress::AcceptedIngress, C, BusinessConfig<B>) -> DispatchFuture,
    DispatchFuture: Future<Output = Result<Vec<u8>>>,
{
    while !stop.load(Ordering::Acquire) {
        tokio::select! {
            accepted = listener.accept() => {
                let Ok((mut socket, _)) = accepted else { continue };
                let Ok(guard) = lifecycle.try_accept() else {
                    let _ = socket.write_all(&status(503)).await;
                    let _ = socket.shutdown().await;
                    continue;
                };
                serve_one(
                    socket,
                    &adapter,
                    deployment.clone(),
                    business.clone(),
                    &dispatch,
                    &observer,
                ).await;
                drop(guard);
            }
            () = tokio::time::sleep(Duration::from_millis(1)) => {}
        }
    }
}

async fn serve_one<B, C, Dispatch, DispatchFuture>(
    mut socket: TcpStream,
    adapter: &saddle_boundary::ingress::ProfuseGwListenerAdapter,
    deployment: C,
    business: BusinessConfig<B>,
    dispatch: &Dispatch,
    observer: &saddle_observability::Observer,
) where
    Dispatch: Fn(saddle_boundary::ingress::AcceptedIngress, C, BusinessConfig<B>) -> DispatchFuture,
    DispatchFuture: Future<Output = Result<Vec<u8>>>,
{
    let response = match tokio::time::timeout(IO_DEADLINE, read_request(&mut socket)).await {
        Ok(Ok(request)) => match adapter.accept(
            &request.method,
            &request.path,
            &request.content_type,
            request.identity,
            &request.body,
        ) {
            Ok(accepted) => {
                let call = observer.start_external_call_checked(
                    adapter.application(),
                    "profusegw",
                    "profusegw",
                    accepted.interface_id.clone(),
                    Some(&accepted.trace_id),
                );
                match call {
                    Ok((call, _)) => match dispatch(accepted, deployment, business).await {
                        Ok(response) => {
                            call.succeed();
                            response
                        }
                        Err(error) => {
                            call.fail(&error);
                            status(500)
                        }
                    },
                    Err(_) => status(500),
                }
            }
            Err(error) => status(error.http_status),
        },
        _ => status(400),
    };
    let _ = tokio::time::timeout(IO_DEADLINE, async {
        socket.write_all(&response).await?;
        socket.shutdown().await
    })
    .await;
}

struct ParsedRequest {
    method: String,
    path: String,
    content_type: String,
    identity: saddle_boundary::ingress::IngressIdentity,
    body: Vec<u8>,
}

async fn read_request(socket: &mut TcpStream) -> std::io::Result<ParsedRequest> {
    let mut bytes = Vec::with_capacity(MAX_HEAD_BYTES);
    let head_end = loop {
        if bytes.len() >= MAX_HEAD_BYTES {
            return Err(std::io::ErrorKind::InvalidData.into());
        }
        let mut chunk = [0_u8; 1024];
        let read = socket.read(&mut chunk).await?;
        if read == 0 {
            return Err(std::io::ErrorKind::UnexpectedEof.into());
        }
        bytes.extend_from_slice(&chunk[..read]);
        if let Some(index) = bytes.windows(4).position(|value| value == b"\r\n\r\n") {
            break index + 4;
        }
    };
    let head =
        std::str::from_utf8(&bytes[..head_end - 4]).map_err(|_| std::io::ErrorKind::InvalidData)?;
    let mut lines = head.split("\r\n");
    let mut request_line = lines
        .next()
        .ok_or(std::io::ErrorKind::InvalidData)?
        .split_ascii_whitespace();
    let method = request_line.next().ok_or(std::io::ErrorKind::InvalidData)?;
    let path = request_line.next().ok_or(std::io::ErrorKind::InvalidData)?;
    if request_line.next() != Some("HTTP/1.1") || request_line.next().is_some() {
        return Err(std::io::ErrorKind::InvalidData.into());
    }
    let mut content_type = None;
    let mut content_length = None;
    let mut request_id = None;
    let mut call_id = None;
    let mut deadline = None;
    for line in lines {
        let (name, value) = line
            .split_once(':')
            .ok_or(std::io::ErrorKind::InvalidData)?;
        let value = value.trim();
        match name.to_ascii_lowercase().as_str() {
            "content-type" => set_once(&mut content_type, value)?,
            "content-length" => set_once(&mut content_length, value)?,
            "x-request-id" => set_once(&mut request_id, value)?,
            "x-call-id" => set_once(&mut call_id, value)?,
            "x-deadline-unix-ms" => set_once(&mut deadline, value)?,
            _ => {}
        }
    }
    let length: usize = content_length
        .ok_or(std::io::ErrorKind::InvalidData)?
        .parse()
        .map_err(|_| std::io::ErrorKind::InvalidData)?;
    if length > saddle_boundary::ingress::MAX_BODY_BYTES {
        return Err(std::io::ErrorKind::InvalidData.into());
    }
    let method = method.to_owned();
    let path = path.to_owned();
    let content_type = content_type
        .ok_or(std::io::ErrorKind::InvalidData)?
        .to_owned();
    let request_id = request_id
        .ok_or(std::io::ErrorKind::InvalidData)?
        .to_owned();
    let call_id = call_id.ok_or(std::io::ErrorKind::InvalidData)?.to_owned();
    let deadline = deadline
        .ok_or(std::io::ErrorKind::InvalidData)?
        .parse()
        .map_err(|_| std::io::ErrorKind::InvalidData)?;
    while bytes.len() < head_end + length {
        let remaining = head_end + length - bytes.len();
        let mut chunk = vec![0_u8; remaining.min(8192)];
        let read = socket.read(&mut chunk).await?;
        if read == 0 {
            return Err(std::io::ErrorKind::UnexpectedEof.into());
        }
        bytes.extend_from_slice(&chunk[..read]);
    }
    if bytes.len() != head_end + length {
        return Err(std::io::ErrorKind::InvalidData.into());
    }
    let identity = saddle_boundary::ingress::IngressIdentity::new(request_id, call_id, deadline)
        .map_err(|_| std::io::ErrorKind::InvalidData)?;
    let body = bytes[head_end..].to_vec();
    Ok(ParsedRequest {
        method,
        path,
        content_type,
        identity,
        body,
    })
}

fn set_once<'a>(slot: &mut Option<&'a str>, value: &'a str) -> std::io::Result<()> {
    if slot.replace(value).is_some() {
        return Err(std::io::ErrorKind::InvalidData.into());
    }
    Ok(())
}

fn status(code: u16) -> Vec<u8> {
    format!("HTTP/1.1 {code} Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").into_bytes()
}

fn process_error(code: &'static str) -> SaddleError {
    SaddleError::new(
        ErrorKind::Infrastructure,
        code,
        "Saddle process startup failed",
    )
}

#[doc(hidden)]
pub fn process_contract_error() -> SaddleError {
    process_error("saddle.process.profusecontract_connect_failed")
}

#[doc(hidden)]
pub fn process_dispatch_error() -> SaddleError {
    process_error("saddle.process.dispatch_failed")
}

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

    #[derive(Deserialize)]
    struct TestBusiness {
        region: String,
    }

    #[test]
    fn unified_database_config_loads_once_and_fails_fast() {
        let root = std::env::temp_dir().join(format!(
            "saddle-alpha10-database-config-{}",
            std::process::id()
        ));
        let mappings = root.join("mappings");
        let config = root.join("saddle.toml");
        let unit_config = root.join("unit-saddle.toml");
        let legacy_config = root.join("legacy-saddle.toml");
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&mappings).unwrap();
        std::fs::write(
            &config,
            r#"[framework]
	listen = "127.0.0.1:39001"
	[framework.management]
	bind = "127.0.0.1:39002"
	[framework.profusecontract]
authority = "http://profusecontract.internal:50051"
[secrets]
databaseUrlEnv = "SADDLE_ALPHA10_DATABASE_TEST"
[database]
mappingDir = "mappings"
[business]
region = "cn-test"
"#,
        )
        .unwrap();
        std::fs::write(
            &unit_config,
            r#"[framework]
	listen = "127.0.0.1:39001"
	[framework.management]
	bind = "127.0.0.1:39002"
	[framework.profusecontract]
authority = "http://profusecontract-{zone}.internal:50051"
[secrets]
"#,
        )
        .unwrap();
        std::fs::write(
            &legacy_config,
            r#"[framework]
	listen = "127.0.0.1:39001"
	[framework.management]
	bind = "127.0.0.1:39002"
	profusecontractAuthorityTemplate = "http://profusecontract.internal:50051"
[secrets]
"#,
        )
        .unwrap();

        let executable = std::env::current_exe().unwrap();
        for mode in [
            "happy",
            "relative-happy",
            "old-key",
            "missing-secret",
            "bad-directory",
            "unit-business",
        ] {
            if mode == "bad-directory" {
                std::fs::remove_dir_all(&mappings).unwrap();
            }
            let mut child = std::process::Command::new(&executable);
            child
                .args([
                    "--ignored",
                    "--exact",
                    "process::tests::unified_database_config_child",
                ])
                .env("SADDLE_ALPHA10_PROCESS_MODE", mode)
                .env_remove("SADDLE_ALPHA10_DATABASE_TEST");
            match mode {
                "relative-happy" => {
                    child
                        .current_dir(&root)
                        .env("SADDLE_ALPHA10_PROCESS_CONFIG", "saddle.toml");
                }
                "unit-business" => {
                    child.env("SADDLE_ALPHA10_PROCESS_CONFIG", &unit_config);
                }
                "old-key" => {
                    child.env("SADDLE_ALPHA10_PROCESS_CONFIG", &legacy_config);
                }
                _ => {
                    child.env("SADDLE_ALPHA10_PROCESS_CONFIG", &config);
                }
            }
            if mode != "missing-secret" {
                child.env(
                    "SADDLE_ALPHA10_DATABASE_TEST",
                    "mysql://deployment-secret@127.0.0.1/database",
                );
            }
            assert!(
                child.status().unwrap().success(),
                "child mode {mode} failed"
            );
        }
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn logging_config_defaults_and_accepts_only_calendar_rotation() {
        let root =
            std::env::temp_dir().join(format!("saddle-logging-config-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&root).unwrap();
        let config = root.join("saddle.toml");
        let base = r#"[framework]
	listen = "127.0.0.1:39001"
	[framework.management]
	bind = "127.0.0.1:39002"
	[framework.profusecontract]
authority = "http://profusecontract.internal:50051"
[secrets]
"#;
        std::fs::write(&config, base).unwrap();
        let loaded = ProcessConfig::<()>::load(&config).unwrap();
        assert_eq!(loaded.logging.directory(), Path::new("./logs"));
        assert_eq!(
            loaded.logging.rotation(),
            saddle_observability::Rotation::Daily
        );

        std::fs::write(
            &config,
            format!(
                "{base}[framework.observability.logging]\ndirectory = \"/var/log/saddle\"\nrotation = \"hourly\"\n"
            ),
        )
        .unwrap();
        let loaded = ProcessConfig::<()>::load(&config).unwrap();
        assert_eq!(loaded.logging.directory(), Path::new("/var/log/saddle"));
        assert_eq!(
            loaded.logging.rotation(),
            saddle_observability::Rotation::Hourly
        );

        std::fs::write(
            &config,
            format!("{base}[framework.observability.logging]\nrotation = \"size\"\n"),
        )
        .unwrap();
        assert_eq!(
            ProcessConfig::<()>::load(&config).err().unwrap().code(),
            "saddle.process.config_invalid"
        );
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn management_config_is_required_and_invalid_bind_fails_before_startup() {
        let root =
            std::env::temp_dir().join(format!("saddle-management-config-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&root).unwrap();
        let config = root.join("saddle.toml");
        let write = |management: &str| {
            std::fs::write(
                &config,
                format!(
                    r#"[framework]
listen = "127.0.0.1:39001"
{management}
[framework.profusecontract]
authority = "http://profusecontract.internal:50051"
[secrets]
"#,
                ),
            )
            .unwrap();
        };

        write("");
        assert_eq!(
            ProcessConfig::<()>::load(&config).err().unwrap().code(),
            "saddle.process.config_invalid"
        );
        write("[framework.management]\nbind = \"not-a-socket\"");
        assert_eq!(
            ProcessConfig::<()>::load(&config).err().unwrap().code(),
            "saddle.process.management_bind_invalid"
        );
        write("[framework.management]\nbind = \"127.0.0.1:39002\"");
        assert_eq!(
            ProcessConfig::<()>::load(&config).unwrap().management_bind,
            "127.0.0.1:39002".parse::<SocketAddr>().unwrap()
        );
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn management_bind_collision_is_fail_fast_and_releases_business_bind() {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(async {
                let occupied = TcpListener::bind("127.0.0.1:0").await.unwrap();
                let occupied_address = occupied.local_addr().unwrap();
                let error =
                    bind_process_listeners("127.0.0.1:0".parse().unwrap(), occupied_address)
                        .await
                        .err()
                        .unwrap();
                assert_eq!(error.code(), "saddle.process.management_bind_failed");
            });
    }

    #[test]
    fn management_entry_exposes_only_fixed_read_only_health_projection() {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(async {
                let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
                let address = listener.local_addr().unwrap();
                let application = saddle_runtime::Application::new();
                let entry = ManagementEntry::new(listener, application.health());
                saddle_runtime::ComponentLifecycle::start(&entry)
                    .await
                    .unwrap();

                async fn request(address: SocketAddr, request: &[u8]) -> Vec<u8> {
                    let mut socket = TcpStream::connect(address).await.unwrap();
                    socket.write_all(request).await.unwrap();
                    let mut response = Vec::new();
                    socket.read_to_end(&mut response).await.unwrap();
                    response
                }

                let live = request(address, b"GET /live HTTP/1.1\r\nHost: localhost\r\n\r\n").await;
                assert!(live.starts_with(b"HTTP/1.1 200 Health\r\n"));
                assert!(live.ends_with(b"{\"live\":true}"));

                let ready =
                    request(address, b"GET /ready HTTP/1.1\r\nHost: localhost\r\n\r\n").await;
                assert!(ready.starts_with(b"HTTP/1.1 503 Health\r\n"));
                assert!(ready.ends_with(b"{\"ready\":false}"));

                let unknown =
                    request(address, b"GET /metrics HTTP/1.1\r\nHost: localhost\r\n\r\n").await;
                assert!(unknown.starts_with(b"HTTP/1.1 404 Error\r\n"));

                let mutation =
                    request(address, b"POST /ready HTTP/1.1\r\nHost: localhost\r\n\r\n").await;
                assert!(mutation.starts_with(b"HTTP/1.1 405 Error\r\n"));

                saddle_runtime::ComponentLifecycle::shutdown(&entry)
                    .await
                    .unwrap();
            });
    }

    #[test]
    #[ignore = "executed in isolated child processes by the parent test"]
    fn unified_database_config_child() {
        let config = PathBuf::from(std::env::var_os("SADDLE_ALPHA10_PROCESS_CONFIG").unwrap());
        let mode = std::env::var("SADDLE_ALPHA10_PROCESS_MODE").unwrap();
        if mode == "unit-business" {
            assert!(ProcessConfig::<()>::load(config).is_ok());
            return;
        }
        let loaded = ProcessConfig::<TestBusiness>::load(config);
        match mode.as_str() {
            "happy" | "relative-happy" => {
                let loaded = loaded.unwrap();
                assert_eq!(loaded.business.get().region, "cn-test");
                assert!(loaded.database.is_some());
            }
            "missing-secret" | "bad-directory" => assert_eq!(
                loaded.err().unwrap().code(),
                "saddle.process.database_config_invalid"
            ),
            "old-key" => assert_eq!(
                loaded.err().unwrap().code(),
                "saddle.process.config_invalid"
            ),
            _ => panic!("unknown child mode"),
        }
    }
}