nautalid 0.1.0

//! Generic **ZeroMQ** REP transport and control-plane dispatch.
//!
//! Domain modules register on [`BusRegistry`] (feature **`filter-control`**).
//! Default endpoint: **`NAUTALID_BUS_IPC`** or [`DEFAULT_BUS_IPC`].

#[cfg(feature = "filter-control")]
mod auth;
pub mod response;

#[cfg(feature = "filter-control")]
mod registry;

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

use thiserror::Error;
use tracing::info;
use crate::zeromq::{RepSocket, Socket, SocketRecv, SocketSend, ZmqMessage};

use crate::readiness;
use crate::shutdown;

#[cfg(feature = "filter-control")]
pub use registry::{BusModule, BusRegistry};

pub use response::{err, modules, ok};

/// Default IPC socket path in the scratch image (parent **0700**, socket **0600**).
pub const DEFAULT_BUS_IPC: &str = "ipc:///run/nautalid/bus.sock";

/// Maximum control-plane request size (sum of ZMQ frame bytes).
pub const MAX_BUS_REQUEST_BYTES: usize = 64 * 1024;

#[derive(Debug, Error)]
pub enum BusError {
    #[error("zeromq: {0}")]
    Zmq(#[from] crate::zeromq::ZmqError),
    #[error("invalid bus endpoint `{0}`")]
    BadEndpoint(String),
    #[error("control-plane bus request exceeds {MAX_BUS_REQUEST_BYTES} bytes")]
    RequestTooLarge,
    #[error("control-plane bus failed to start")]
    StartFailed,
    #[error(transparent)]
    Io(#[from] std::io::Error),
}

/// REP socket endpoint from **`NAUTALID_BUS_IPC`** or [`DEFAULT_BUS_IPC`].
pub fn endpoint_from_env() -> String {
    std::env::var("NAUTALID_BUS_IPC")
        .ok()
        .filter(|s| !s.is_empty())
        .unwrap_or_else(|| DEFAULT_BUS_IPC.to_string())
}

fn bus_tcp_allowed() -> bool {
    matches!(
        std::env::var("NAUTALID_BUS_ALLOW_TCP").ok().as_deref(),
        Some("1") | Some("true") | Some("yes")
    )
}

/// Reject remote/network bus endpoints unless explicitly opted in.
pub fn validate_bus_endpoint(endpoint: &str) -> Result<(), BusError> {
    if endpoint.starts_with("ipc://") {
        return Ok(());
    }
    if endpoint.starts_with("tcp://") || endpoint.starts_with("inproc://") {
        if bus_tcp_allowed() {
            return Ok(());
        }
        return Err(BusError::BadEndpoint(format!(
            "{endpoint} (set NAUTALID_BUS_ALLOW_TCP=1 to allow non-IPC bus endpoints)"
        )));
    }
    Err(BusError::BadEndpoint(format!(
        "{endpoint} (only ipc:// is allowed by default)"
    )))
}

/// Validate bus endpoint before bind (IPC by default; TCP requires opt-in).
#[cfg(feature = "filter-control")]
pub fn ensure_control_plane_config() -> Result<(), BusError> {
    validate_bus_endpoint(&endpoint_from_env())?;
    Ok(())
}

/// Bind the control plane and wait until the REP socket is listening.
#[cfg(feature = "filter-control")]
pub async fn start_control_plane(state: &crate::state::AppState) -> Result<(), BusError> {
    ensure_control_plane_config()?;
    let registry = Arc::new(BusRegistry::from_state(state));
    let endpoint = endpoint_from_env();
    let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();

    tokio::spawn(async move {
        if let Err(e) = serve_rep(endpoint, move |req| registry.handle(req), Some(ready_tx)).await {
            readiness::set_bus_ready(false);
            tracing::error!(%e, "control-plane bus stopped");
        }
    });

    match tokio::time::timeout(std::time::Duration::from_secs(5), ready_rx).await {
        Ok(Ok(())) => Ok(()),
        Ok(Err(_)) => {
            readiness::set_bus_ready(false);
            Err(BusError::StartFailed)
        }
        Err(_) => {
            readiness::set_bus_ready(false);
            Err(BusError::StartFailed)
        }
    }
}

/// Serve **`handler(request)`** replies on a REP socket until process shutdown.
pub async fn serve_rep<F>(
    endpoint: String,
    handler: F,
    ready: Option<tokio::sync::oneshot::Sender<()>>,
) -> Result<(), BusError>
where
    F: Fn(&str) -> String + Send + Sync + 'static,
{
    validate_bus_endpoint(&endpoint)?;
    prepare_ipc_path(&endpoint)?;
    let mut socket = RepSocket::new();
    socket.bind(&endpoint).await?;
    restrict_ipc_socket_permissions(&endpoint)?;
    readiness::set_bus_ready(true);
    if let Some(tx) = ready {
        let _ = tx.send(());
    }
    info!(%endpoint, "nautalid bus listening (REP)");

    loop {
        tokio::select! {
            msg = socket.recv() => {
                let msg = msg?;
                let request = message_to_string(&msg)?;
                let body = handler(&request);
                socket.send(body.into()).await?;
            }
            () = shutdown::wait_for_shutdown() => {
                info!("nautalid bus shutting down");
                break;
            }
        }
    }
    readiness::set_bus_ready(false);
    Ok(())
}

fn message_to_string(msg: &ZmqMessage) -> Result<String, BusError> {
    let mut total = 0usize;
    let parts: Vec<String> = msg
        .iter()
        .map(|part| {
            total = total.saturating_add(part.len());
            if total > MAX_BUS_REQUEST_BYTES {
                return Err(BusError::RequestTooLarge);
            }
            Ok(String::from_utf8_lossy(part).into_owned())
        })
        .collect::<Result<Vec<_>, _>>()?;
    Ok(parts.join(""))
}

fn prepare_ipc_path(endpoint: &str) -> Result<(), BusError> {
    let Some(path_str) = endpoint.strip_prefix("ipc://") else {
        return Ok(());
    };
    let path = Path::new(path_str);
    if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
        let existed = parent.exists();
        std::fs::create_dir_all(parent)?;
        // Only chmod directories we create — never touch existing parents like /tmp.
        #[cfg(unix)]
        if !existed {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700))?;
        }
    }
    if path.exists() {
        std::fs::remove_file(path)?;
    }
    Ok(())
}

fn restrict_ipc_socket_permissions(endpoint: &str) -> Result<(), BusError> {
    let Some(path_str) = endpoint.strip_prefix("ipc://") else {
        return Ok(());
    };
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(path_str, std::fs::Permissions::from_mode(0o600))?;
    }
    Ok(())
}

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

    #[test]
    fn prepare_ipc_existing_parent_skips_chmod() {
        let parent = std::env::temp_dir();
        let sock = parent.join(format!("nautalid-bus-{}.sock", std::process::id()));
        let endpoint = format!("ipc://{}", sock.display());
        prepare_ipc_path(&endpoint).expect("existing parent (e.g. /tmp) must not chmod");
        if sock.exists() {
            let _ = std::fs::remove_file(&sock);
        }
    }

    #[test]
    fn prepare_ipc_under_tmp_does_not_chmod_parent() {
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("bus.sock");
        let endpoint = format!("ipc://{}", sock.display());
        prepare_ipc_path(&endpoint).expect("prepare under writable temp parent");
    }

    #[test]
    fn rejects_tcp_by_default() {
        assert!(validate_bus_endpoint("tcp://0.0.0.0:5555").is_err());
    }

    #[test]
    fn allows_ipc() {
        assert!(validate_bus_endpoint(DEFAULT_BUS_IPC).is_ok());
    }
}