mod config;
mod events;
mod handle;
mod serve;
pub mod testkit;
use std::sync::Arc;
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use koi_client::KoiClient;
pub use config::{DnsConfigBuilder, KoiConfig, ServiceMode};
pub use events::KoiEvent;
pub use handle::{
CertmeshHandle, DnsHandle, HealthHandle, KoiHandle, MdnsHandle, ProxyHandle,
DEFAULT_DISCOVER_WINDOW,
};
pub use koi_common::firewall::{FirewallPort, FirewallProtocol};
pub use koi_certmesh::PeerClient;
pub use koi_common::diagnosis::{CheckStatus, DiagnosisCheck, DiagnosisStatus, TrustDiagnosis};
pub use koi_common::peer::Peer;
pub use koi_common::posture::{Posture, PostureLevel};
pub use koi_common::sealed::{Confidentiality, Opened, Sealed};
pub use koi_common::types::ServiceRecord;
pub use koi_config::state::DnsEntry;
pub use koi_health::{HealthCheck, HealthSnapshot, ServiceCheckKind};
pub use koi_mdns::protocol::{RegisterPayload, RegistrationResult};
pub use koi_mdns::MdnsEvent;
pub use koi_proxy::ProxyEntry;
pub use serve::serve_adaptive;
pub use koi_crypto::vault::{Vault, VaultError};
pub use koi_runtime::{RuntimeBackendKind, RuntimeConfig};
pub type Result<T> = std::result::Result<T, KoiError>;
#[derive(Debug, thiserror::Error)]
pub enum KoiError {
#[error("capability disabled: {0}")]
DisabledCapability(&'static str),
#[error("not available in client (remote) mode: {0}")]
RemoteUnsupported(&'static str),
#[error("mdns error: {0}")]
Mdns(#[from] koi_mdns::MdnsError),
#[error("dns error: {0}")]
Dns(#[from] koi_dns::DnsError),
#[error("health error: {0}")]
Health(#[from] koi_health::HealthError),
#[error("proxy error: {0}")]
Proxy(#[from] koi_proxy::ProxyError),
#[error("certmesh error: {0}")]
Certmesh(#[from] koi_certmesh::CertmeshError),
#[error("runtime error: {0}")]
Runtime(#[from] koi_runtime::RuntimeError),
#[error("client error: {0}")]
Client(#[from] koi_client::ClientError),
#[error("io error: {0}")]
Io(#[from] std::io::Error),
#[error("insecure configuration: {0}")]
InsecureConfig(String),
}
impl From<koi_compose::cores::BuildCoresError> for KoiError {
fn from(e: koi_compose::cores::BuildCoresError) -> Self {
use koi_compose::cores::BuildCoresError as B;
match e {
B::Mdns(e) => KoiError::Mdns(e),
B::Dns(e) => KoiError::Dns(e),
B::Proxy(e) => KoiError::Proxy(e),
B::Health(e) => KoiError::Health(e),
B::CertmeshInit(s) => KoiError::Io(std::io::Error::other(s)),
}
}
}
pub struct Builder {
config: KoiConfig,
event_handler: Option<Arc<dyn Fn(KoiEvent) + Send + Sync>>,
extra_firewall_ports: Vec<koi_common::firewall::FirewallPort>,
}
impl Builder {
pub fn new() -> Self {
Self {
config: KoiConfig::default(),
event_handler: None,
extra_firewall_ports: Vec::new(),
}
}
pub fn data_dir(mut self, path: impl Into<std::path::PathBuf>) -> Self {
self.config.data_dir = Some(path.into());
self
}
pub fn service_endpoint(mut self, endpoint: impl Into<String>) -> Self {
self.config.service_endpoint = endpoint.into();
self
}
pub fn service_token(mut self, token: impl Into<String>) -> Self {
self.config.service_token = Some(token.into());
self
}
pub fn service_mode(mut self, mode: ServiceMode) -> Self {
self.config.service_mode = mode;
self
}
pub fn http(mut self, enabled: bool) -> Self {
self.config.http_enabled = enabled;
self
}
pub fn mdns(mut self, enabled: bool) -> Self {
self.config.mdns_enabled = enabled;
self
}
pub fn dns<F>(mut self, configure: F) -> Self
where
F: FnOnce(DnsConfigBuilder) -> DnsConfigBuilder,
{
let builder = DnsConfigBuilder::new(self.config.dns_config.clone());
self.config.dns_config = configure(builder).build();
self
}
pub fn dns_enabled(mut self, enabled: bool) -> Self {
self.config.dns_enabled = enabled;
self
}
pub fn dns_auto_start(mut self, enabled: bool) -> Self {
self.config.dns_auto_start = enabled;
self
}
pub fn health(mut self, enabled: bool) -> Self {
self.config.health_enabled = enabled;
self
}
pub fn health_auto_start(mut self, enabled: bool) -> Self {
self.config.health_auto_start = enabled;
self
}
pub fn certmesh(mut self, enabled: bool) -> Self {
self.config.certmesh_enabled = enabled;
self
}
pub fn proxy(mut self, enabled: bool) -> Self {
self.config.proxy_enabled = enabled;
self
}
pub fn proxy_auto_start(mut self, enabled: bool) -> Self {
self.config.proxy_auto_start = enabled;
self
}
pub fn udp(mut self, enabled: bool) -> Self {
self.config.udp_enabled = enabled;
self
}
pub fn runtime(mut self, kind: koi_runtime::RuntimeBackendKind) -> Self {
self.config.runtime_enabled = true;
self.config.runtime_backend = kind;
self
}
pub fn runtime_auto(mut self) -> Self {
self.config.runtime_enabled = true;
self.config.runtime_backend = koi_runtime::RuntimeBackendKind::Auto;
self
}
pub fn orchestrator(mut self, enabled: bool) -> Self {
self.config.orchestrator_enabled = enabled;
self
}
pub fn certmesh_background(mut self, enabled: bool) -> Self {
self.config.certmesh_background_enabled = enabled;
self
}
pub fn http_port(mut self, port: u16) -> Self {
self.config.http_port = port;
self
}
pub fn dashboard(mut self, enabled: bool) -> Self {
self.config.dashboard_enabled = enabled;
self
}
pub fn api_docs(mut self, enabled: bool) -> Self {
self.config.api_docs_enabled = enabled;
self
}
pub fn mdns_browser(mut self, enabled: bool) -> Self {
self.config.mdns_browser_enabled = enabled;
self
}
pub fn announce_http(mut self, enabled: bool) -> Self {
self.config.announce_http = enabled;
self
}
pub fn http_token(mut self, token: impl Into<String>) -> Self {
self.config.http_token = Some(token.into());
self
}
pub fn events<F>(mut self, handler: F) -> Self
where
F: Fn(KoiEvent) + Send + Sync + 'static,
{
self.event_handler = Some(Arc::new(handler));
self
}
pub fn extra_firewall_ports(mut self, ports: Vec<koi_common::firewall::FirewallPort>) -> Self {
self.extra_firewall_ports = ports;
self
}
pub fn ensure_firewall_rules(self, prefix: &str) -> Self {
let mut all_ports = self.config.firewall_ports();
all_ports.extend(self.extra_firewall_ports.iter().cloned());
let count = koi_common::firewall::ensure_firewall_rules(prefix, &all_ports);
if count > 0 {
tracing::info!(count, "Firewall rules ensured");
}
self
}
pub fn build(self) -> Result<KoiEmbedded> {
Ok(KoiEmbedded {
config: self.config,
event_handler: self.event_handler,
})
}
}
impl Default for Builder {
fn default() -> Self {
Self::new()
}
}
pub struct KoiEmbedded {
config: KoiConfig,
event_handler: Option<Arc<dyn Fn(KoiEvent) + Send + Sync>>,
}
impl KoiEmbedded {
pub async fn start(self) -> Result<KoiHandle> {
let cancel = CancellationToken::new();
let (event_tx, _) = broadcast::channel(256);
let mut tasks: Vec<JoinHandle<()>> = Vec::new();
if self.config.service_mode != ServiceMode::EmbeddedOnly {
let client = Arc::new(build_remote_client(&self.config));
match self.config.service_mode {
ServiceMode::ClientOnly => {
tokio::task::spawn_blocking({
let client = Arc::clone(&client);
move || client.health()
})
.await
.map_err(map_join_error)??;
return Ok(KoiHandle::new_remote(client, event_tx, cancel, tasks));
}
ServiceMode::Auto => {
let health = tokio::task::spawn_blocking({
let client = Arc::clone(&client);
move || client.health()
})
.await;
if matches!(health, Ok(Ok(()))) {
return Ok(KoiHandle::new_remote(client, event_tx, cancel, tasks));
}
}
ServiceMode::EmbeddedOnly => {}
}
}
if self.config.http_enabled && self.config.announce_http && self.config.http_token.is_none()
{
return Err(KoiError::InsecureConfig(
"announce_http exposes the embedded HTTP adapter on 0.0.0.0; call \
.http_token(..) to require x-koi-token, or drop announce_http to bind loopback"
.into(),
));
}
let cores = koi_compose::cores::build_cores(
&koi_compose::cores::CoreSpec {
no_mdns: !self.config.mdns_enabled,
no_certmesh: !self.config.certmesh_enabled,
no_dns: !self.config.dns_enabled,
no_health: !self.config.health_enabled,
no_proxy: !self.config.proxy_enabled,
no_udp: !self.config.udp_enabled,
no_runtime: !self.config.runtime_enabled,
data_dir: self.config.data_dir.clone(),
dns_config: self.config.dns_config.clone(),
runtime: self.config.runtime_backend.to_string(),
http_port: self.config.http_port,
dns_state_path: self
.config
.data_dir
.as_ref()
.map(|dir| dir.join("state").join("dns.json")),
proxy_data_dir: self.config.data_dir.clone(),
dns_auto_start: self.config.dns_auto_start,
health_auto_start: self.config.health_auto_start,
proxy_auto_start: self.config.proxy_auto_start,
spawn_orchestrator: self.config.orchestrator_enabled,
spawn_certmesh_loops: self.config.certmesh_background_enabled,
fail_fast: true,
},
&cancel,
&mut tasks,
)
.await?;
let koi_compose::cores::Cores {
mdns,
certmesh,
dns,
health,
proxy,
udp,
runtime,
mdns_snapshot: mdns_bridge,
} = cores;
let dashboard_state = if self.config.dashboard_enabled && self.config.http_enabled {
let started_at = std::time::Instant::now();
let snap_mdns = mdns.clone();
let snap_certmesh = certmesh.clone();
let snap_dns = dns.clone();
let snap_health = health.clone();
let snap_proxy = proxy.clone();
let snap_udp = udp.clone();
let snap_runtime = runtime.clone();
let snapshot_fn: koi_dashboard::dashboard::SnapshotFn = Arc::new(move || {
let m = snap_mdns.clone();
let cm = snap_certmesh.clone();
let d = snap_dns.clone();
let h = snap_health.clone();
let p = snap_proxy.clone();
let u = snap_udp.clone();
let rt = snap_runtime.clone();
Box::pin(async move { build_embedded_snapshot(m, cm, d, h, p, u, rt).await })
});
let (dash_event_tx, _) = broadcast::channel(256);
let ds = koi_dashboard::dashboard::DashboardState {
identity: koi_dashboard::dashboard::DashboardIdentity {
version: env!("CARGO_PKG_VERSION").to_string(),
platform: std::env::consts::OS.to_string(),
},
mode: "embedded",
snapshot_fn,
event_tx: dash_event_tx.clone(),
started_at,
};
tasks.push(koi_dashboard::forward::spawn_event_forwarder(
koi_dashboard::forward::ForwarderCores {
mdns: mdns.clone(),
certmesh: certmesh.clone(),
dns: dns.clone(),
health: health.clone(),
proxy: proxy.clone(),
runtime: runtime.clone(),
},
dash_event_tx,
cancel.clone(),
));
Some(ds)
} else {
None
};
let browser_state = if self.config.mdns_browser_enabled && self.config.http_enabled {
if let Some(ref mdns_core) = mdns {
Some(koi_dashboard::browser::build_state(
mdns_core.clone(),
cancel.clone(),
))
} else {
tracing::warn!("mdns_browser enabled but mDNS is disabled — skipping browser");
None
}
} else {
None
};
let mut http_addr: Option<std::net::SocketAddr> = None;
if self.config.http_enabled {
let http_cancel = cancel.clone();
let http_cores = koi_compose::cores::Cores {
mdns: mdns.clone(),
certmesh: certmesh.clone(),
dns: dns.clone(),
health: health.clone(),
proxy: proxy.clone(),
udp: udp.clone(),
runtime: runtime.clone(),
mdns_snapshot: mdns_bridge.clone(),
};
let exposed = self.config.announce_http;
let bind_ip = if exposed {
std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED)
} else {
std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)
};
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
let http_cfg = koi_serve::http::HttpConfig {
bind_ip,
port: self.config.http_port,
started_at: std::time::Instant::now(),
dashboard: dashboard_state,
browser: browser_state,
auth: self.config.http_token.clone(),
mdns_snapshot: mdns_bridge.clone(),
mcp_http: false,
admin_shutdown: false,
api_docs: self.config.api_docs_enabled,
daemon: false,
ready: Some(ready_tx),
};
tasks.push(tokio::spawn(async move {
if let Err(e) = koi_serve::http::start(http_cores, http_cfg, http_cancel).await {
tracing::error!(error = %e, "embedded HTTP adapter failed");
}
}));
http_addr = ready_rx.await.ok();
}
let announce_cores = koi_compose::cores::Cores {
mdns: mdns.clone(),
certmesh: certmesh.clone(),
dns: dns.clone(),
health: health.clone(),
proxy: proxy.clone(),
udp: udp.clone(),
runtime: runtime.clone(),
mdns_snapshot: mdns_bridge.clone(),
};
let announce_http_port = http_addr.map(|a| a.port()).unwrap_or(self.config.http_port);
koi_compose::self_announce::spawn(
&announce_cores,
koi_compose::self_announce::SelfAnnounceConfig {
http_port: announce_http_port,
dashboard_enabled: self.config.dashboard_enabled,
announce_http: self.config.announce_http
&& self.config.http_enabled
&& self.config.mdns_enabled,
announce_mcp: false,
dns_zone: self.config.dns_config.zone.clone(),
},
cancel.clone(),
&mut tasks,
);
if let Some(core) = &mdns {
spawn_event_mapper(
core.subscribe(),
map_mdns_event,
event_tx.clone(),
self.event_handler.clone(),
cancel.clone(),
&mut tasks,
);
}
if let Some(runtime) = &health {
spawn_event_mapper(
runtime.core().subscribe(),
|e| Some(map_health_event(e)),
event_tx.clone(),
self.event_handler.clone(),
cancel.clone(),
&mut tasks,
);
}
if let Some(runtime) = &dns {
spawn_event_mapper(
runtime.core().subscribe(),
|e| Some(map_dns_event(e)),
event_tx.clone(),
self.event_handler.clone(),
cancel.clone(),
&mut tasks,
);
}
if let Some(core) = &certmesh {
spawn_event_mapper(
core.subscribe(),
|e| Some(map_certmesh_event(e)),
event_tx.clone(),
self.event_handler.clone(),
cancel.clone(),
&mut tasks,
);
spawn_posture_watcher(
core.watch_posture(),
event_tx.clone(),
self.event_handler.clone(),
cancel.clone(),
&mut tasks,
);
}
if let Some(runtime_proxy) = &proxy {
spawn_event_mapper(
runtime_proxy.core().subscribe(),
|e| Some(map_proxy_event(e)),
event_tx.clone(),
self.event_handler.clone(),
cancel.clone(),
&mut tasks,
);
}
if let Some(runtime_core) = &runtime {
spawn_event_mapper(
runtime_core.subscribe(),
map_runtime_event,
event_tx.clone(),
self.event_handler.clone(),
cancel.clone(),
&mut tasks,
);
}
if self.config.orchestrator_enabled && runtime.is_none() {
tracing::warn!(
"orchestrator enabled but the runtime adapter is not — skipping orchestrator"
);
}
if self.config.certmesh_background_enabled {
if let Some(ref certmesh_core) = certmesh {
koi_compose::certmesh::spawn_enrollment_approval(
certmesh_core,
koi_compose::certmesh::deny_and_log_decider(),
&cancel,
&mut tasks,
)
.await;
} else {
tracing::warn!(
"certmesh_background enabled but certmesh is not — skipping certmesh loops"
);
}
}
Ok(KoiHandle::new_embedded(
mdns,
dns,
health,
certmesh,
proxy,
udp,
runtime,
http_addr,
self.config.data_dir.clone(),
event_tx,
cancel,
tasks,
))
}
}
fn build_remote_client(config: &KoiConfig) -> KoiClient {
if let Some(token) = &config.service_token {
return KoiClient::with_token(&config.service_endpoint, token);
}
if let Some(bc) = koi_config::breadcrumb::read_breadcrumb() {
if endpoints_match(&bc.endpoint, &config.service_endpoint) {
return KoiClient::with_token(&config.service_endpoint, &bc.token);
}
}
KoiClient::new(&config.service_endpoint)
}
fn endpoints_match(a: &str, b: &str) -> bool {
fn norm(s: &str) -> String {
s.trim_end_matches('/')
.to_ascii_lowercase()
.replace("localhost", "127.0.0.1")
}
norm(a) == norm(b)
}
fn map_mdns_event(event: MdnsEvent) -> Option<KoiEvent> {
match event {
MdnsEvent::Found(record) => Some(KoiEvent::MdnsFound(record)),
MdnsEvent::Resolved(record) => Some(KoiEvent::MdnsResolved(record)),
MdnsEvent::Removed { name, service_type } => {
Some(KoiEvent::MdnsRemoved { name, service_type })
}
}
}
fn map_health_event(event: koi_health::HealthEvent) -> KoiEvent {
match event {
koi_health::HealthEvent::StatusChanged { name, status } => {
KoiEvent::HealthChanged { name, status }
}
}
}
fn map_dns_event(event: koi_dns::DnsEvent) -> KoiEvent {
match event {
koi_dns::DnsEvent::EntryUpdated { name, ip } => KoiEvent::DnsEntryUpdated { name, ip },
koi_dns::DnsEvent::EntryRemoved { name } => KoiEvent::DnsEntryRemoved { name },
}
}
fn map_certmesh_event(event: koi_certmesh::CertmeshEvent) -> KoiEvent {
match event {
koi_certmesh::CertmeshEvent::MemberJoined {
hostname,
fingerprint,
} => KoiEvent::CertmeshMemberJoined {
hostname,
fingerprint,
},
koi_certmesh::CertmeshEvent::MemberRevoked { hostname } => {
KoiEvent::CertmeshMemberRevoked { hostname }
}
koi_certmesh::CertmeshEvent::Destroyed => KoiEvent::CertmeshDestroyed,
koi_certmesh::CertmeshEvent::CertRenewed { expires_at } => {
KoiEvent::CertRenewed { expires_at }
}
koi_certmesh::CertmeshEvent::CertExpiringSoon { days_left } => {
KoiEvent::CertExpiringSoon { days_left }
}
koi_certmesh::CertmeshEvent::CertRenewalFailed {
reason,
consecutive_failures,
} => KoiEvent::CertRenewalFailed {
reason,
consecutive_failures,
},
koi_certmesh::CertmeshEvent::BundleUpdated { self_revoked } => {
KoiEvent::BundleUpdated { self_revoked }
}
}
}
fn map_proxy_event(event: koi_proxy::ProxyEvent) -> KoiEvent {
match event {
koi_proxy::ProxyEvent::EntryUpdated { entry } => KoiEvent::ProxyEntryUpdated { entry },
koi_proxy::ProxyEvent::EntryRemoved { name } => KoiEvent::ProxyEntryRemoved { name },
}
}
fn map_runtime_event(event: koi_runtime::RuntimeEvent) -> Option<KoiEvent> {
match event {
koi_runtime::RuntimeEvent::Started(instance) => Some(KoiEvent::RuntimeInstanceStarted {
name: instance.name,
backend: instance.backend,
}),
koi_runtime::RuntimeEvent::Stopped { name, .. } => {
Some(KoiEvent::RuntimeInstanceStopped { name })
}
_ => None,
}
}
fn spawn_event_mapper<E, F>(
mut rx: broadcast::Receiver<E>,
map: F,
tx: broadcast::Sender<KoiEvent>,
handler: Option<Arc<dyn Fn(KoiEvent) + Send + Sync>>,
cancel: CancellationToken,
tasks: &mut Vec<JoinHandle<()>>,
) where
E: Clone + Send + 'static,
F: Fn(E) -> Option<KoiEvent> + Send + 'static,
{
tasks.push(tokio::spawn(async move {
loop {
tokio::select! {
_ = cancel.cancelled() => break,
msg = rx.recv() => {
let Ok(event) = msg else { continue; };
if let Some(mapped) = map(event) {
emit_event(&tx, handler.as_ref(), mapped);
}
}
}
}
}));
}
fn spawn_posture_watcher(
mut rx: tokio::sync::watch::Receiver<koi_common::posture::Posture>,
tx: broadcast::Sender<KoiEvent>,
handler: Option<Arc<dyn Fn(KoiEvent) + Send + Sync>>,
cancel: CancellationToken,
tasks: &mut Vec<JoinHandle<()>>,
) {
tasks.push(tokio::spawn(async move {
let mut last = *rx.borrow_and_update();
loop {
tokio::select! {
_ = cancel.cancelled() => break,
res = rx.changed() => {
if res.is_err() {
break; }
let to = *rx.borrow_and_update();
if to != last {
emit_event(&tx, handler.as_ref(), KoiEvent::PostureChanged { from: last, to });
last = to;
}
}
}
}
}));
}
fn emit_event(
tx: &broadcast::Sender<KoiEvent>,
handler: Option<&Arc<dyn Fn(KoiEvent) + Send + Sync>>,
event: KoiEvent,
) {
if let Some(handler) = handler {
handler(event.clone());
}
let _ = tx.send(event);
}
pub(crate) fn map_join_error(err: tokio::task::JoinError) -> KoiError {
KoiError::Io(std::io::Error::other(err.to_string()))
}
async fn build_embedded_snapshot(
mdns: Option<Arc<koi_mdns::MdnsCore>>,
certmesh: Option<Arc<koi_certmesh::CertmeshCore>>,
dns: Option<Arc<koi_dns::DnsRuntime>>,
health: Option<Arc<koi_health::HealthRuntime>>,
proxy: Option<Arc<koi_proxy::ProxyRuntime>>,
udp: Option<Arc<koi_udp::UdpRuntime>>,
runtime: Option<Arc<koi_runtime::RuntimeCore>>,
) -> serde_json::Value {
let cores = koi_compose::cores::Cores {
mdns,
certmesh,
dns,
health,
proxy,
udp,
runtime,
mdns_snapshot: None,
};
koi_compose::snapshot::build_dashboard_snapshot(&cores).await
}
#[cfg(test)]
mod tests {
use super::*;
use koi_common::types::ServiceRecord;
use std::collections::HashMap;
fn sample_record() -> ServiceRecord {
ServiceRecord {
name: "Test Service".to_string(),
service_type: "_http._tcp".to_string(),
host: Some("host.local".to_string()),
ip: Some("10.0.0.1".to_string()),
port: Some(8080),
txt: HashMap::new(),
}
}
#[test]
fn koi_error_disabled_capability_display() {
let err = KoiError::DisabledCapability("mdns");
assert_eq!(err.to_string(), "capability disabled: mdns");
}
#[test]
fn koi_error_io_from_impl() {
let io_err = std::io::Error::new(std::io::ErrorKind::NotFound, "file missing");
let err: KoiError = io_err.into();
assert!(matches!(err, KoiError::Io(_)));
assert!(err.to_string().contains("file missing"));
}
#[test]
fn koi_error_debug_does_not_panic() {
let err = KoiError::DisabledCapability("proxy");
let debug = format!("{err:?}");
assert!(debug.contains("DisabledCapability"));
}
#[tokio::test]
async fn init_certmesh_core_honors_custom_data_dir_end_to_end() {
let base = koi_common::test::ensure_data_dir("koi-embedded-datadir-tests");
let data_dir = base.join("custom-data");
let paths = koi_certmesh::CertmeshPaths::with_data_dir(data_dir.clone());
let fresh =
koi_compose::cores::init_certmesh_core(Some(&data_dir)).expect("uninitialized core");
assert_eq!(
fresh.paths().data_dir(),
data_dir.as_path(),
"uninitialized core must keep the injected data_dir"
);
koi_certmesh::ca::create_ca("test-pass-strong", &[7u8; 32], &paths)
.expect("create CA under injected dir");
let roster = koi_certmesh::roster::Roster::new(false, true, Some("ops".to_string()));
koi_certmesh::roster::save_roster(&roster, &paths.roster_path())
.expect("save roster under injected dir");
let reopened =
koi_compose::cores::init_certmesh_core(Some(&data_dir)).expect("locked core");
assert_eq!(reopened.paths().data_dir(), data_dir.as_path());
reopened
.unlock("test-pass-strong")
.await
.expect("unlock CA from the injected data_dir");
}
#[test]
fn map_mdns_found() {
let record = sample_record();
let event = koi_mdns::MdnsEvent::Found(record.clone());
let mapped = map_mdns_event(event);
assert!(mapped.is_some());
match mapped.unwrap() {
KoiEvent::MdnsFound(r) => assert_eq!(r.name, "Test Service"),
other => panic!("expected MdnsFound, got {other:?}"),
}
}
#[test]
fn map_mdns_resolved() {
let record = sample_record();
let event = koi_mdns::MdnsEvent::Resolved(record);
let mapped = map_mdns_event(event);
assert!(mapped.is_some());
match mapped.unwrap() {
KoiEvent::MdnsResolved(r) => {
assert_eq!(r.port, Some(8080));
assert_eq!(r.service_type, "_http._tcp");
}
other => panic!("expected MdnsResolved, got {other:?}"),
}
}
#[test]
fn map_mdns_removed() {
let event = koi_mdns::MdnsEvent::Removed {
name: "Gone Service".to_string(),
service_type: "_http._tcp".to_string(),
};
let mapped = map_mdns_event(event);
assert!(mapped.is_some());
match mapped.unwrap() {
KoiEvent::MdnsRemoved { name, service_type } => {
assert_eq!(name, "Gone Service");
assert_eq!(service_type, "_http._tcp");
}
other => panic!("expected MdnsRemoved, got {other:?}"),
}
}
#[test]
fn map_health_status_changed_up() {
let event = koi_health::HealthEvent::StatusChanged {
name: "api".to_string(),
status: koi_health::HealthStatus::Up,
};
let mapped = map_health_event(event);
match mapped {
KoiEvent::HealthChanged { name, status } => {
assert_eq!(name, "api");
assert!(matches!(status, koi_health::HealthStatus::Up));
}
other => panic!("expected HealthChanged, got {other:?}"),
}
}
#[test]
fn map_health_status_changed_down() {
let event = koi_health::HealthEvent::StatusChanged {
name: "db".to_string(),
status: koi_health::HealthStatus::Down,
};
let mapped = map_health_event(event);
match mapped {
KoiEvent::HealthChanged { name, status } => {
assert_eq!(name, "db");
assert!(matches!(status, koi_health::HealthStatus::Down));
}
other => panic!("expected HealthChanged, got {other:?}"),
}
}
#[test]
fn map_dns_entry_updated() {
let event = koi_dns::DnsEvent::EntryUpdated {
name: "grafana".to_string(),
ip: "10.0.0.5".to_string(),
};
let mapped = map_dns_event(event);
match mapped {
KoiEvent::DnsEntryUpdated { name, ip } => {
assert_eq!(name, "grafana");
assert_eq!(ip, "10.0.0.5");
}
other => panic!("expected DnsEntryUpdated, got {other:?}"),
}
}
#[test]
fn map_dns_entry_removed() {
let event = koi_dns::DnsEvent::EntryRemoved {
name: "old-host".to_string(),
};
let mapped = map_dns_event(event);
match mapped {
KoiEvent::DnsEntryRemoved { name } => {
assert_eq!(name, "old-host");
}
other => panic!("expected DnsEntryRemoved, got {other:?}"),
}
}
#[test]
fn map_certmesh_member_joined() {
let event = koi_certmesh::CertmeshEvent::MemberJoined {
hostname: "node-a".to_string(),
fingerprint: "sha256:abc".to_string(),
};
let mapped = map_certmesh_event(event);
match mapped {
KoiEvent::CertmeshMemberJoined {
hostname,
fingerprint,
} => {
assert_eq!(hostname, "node-a");
assert_eq!(fingerprint, "sha256:abc");
}
other => panic!("expected CertmeshMemberJoined, got {other:?}"),
}
}
#[test]
fn map_certmesh_member_revoked() {
let event = koi_certmesh::CertmeshEvent::MemberRevoked {
hostname: "node-b".to_string(),
};
let mapped = map_certmesh_event(event);
match mapped {
KoiEvent::CertmeshMemberRevoked { hostname } => {
assert_eq!(hostname, "node-b");
}
other => panic!("expected CertmeshMemberRevoked, got {other:?}"),
}
}
#[test]
fn map_certmesh_destroyed() {
let event = koi_certmesh::CertmeshEvent::Destroyed;
let mapped = map_certmesh_event(event);
assert!(matches!(mapped, KoiEvent::CertmeshDestroyed));
}
#[tokio::test]
async fn posture_watcher_emits_upgrade_and_degrade() {
use koi_common::posture::Posture;
let (tx_p, rx_p) = tokio::sync::watch::channel(Posture::OPEN);
let (ev_tx, mut ev_rx) = broadcast::channel(16);
let cancel = CancellationToken::new();
let mut tasks = Vec::new();
spawn_posture_watcher(rx_p, ev_tx, None, cancel.clone(), &mut tasks);
tokio::task::yield_now().await;
tx_p.send(Posture::new(true, false)).unwrap();
let ev = tokio::time::timeout(std::time::Duration::from_secs(1), ev_rx.recv())
.await
.expect("event arrives")
.expect("recv ok");
assert!(
matches!(ev, KoiEvent::PostureChanged { from, to } if !from.signed && to.signed),
"expected upgrade, got {ev:?}"
);
tx_p.send(Posture::OPEN).unwrap();
let ev = tokio::time::timeout(std::time::Duration::from_secs(1), ev_rx.recv())
.await
.expect("event arrives")
.expect("recv ok");
assert!(
matches!(ev, KoiEvent::PostureChanged { from, to } if from.signed && !to.signed),
"expected degrade, got {ev:?}"
);
cancel.cancel();
for t in tasks {
let _ = t.await;
}
}
#[test]
fn map_proxy_entry_updated() {
let entry = koi_proxy::ProxyEntry {
name: "web".to_string(),
listen_port: 443,
backend: "http://localhost:3000".to_string(),
allow_remote: true,
};
let event = koi_proxy::ProxyEvent::EntryUpdated {
entry: entry.clone(),
};
let mapped = map_proxy_event(event);
match mapped {
KoiEvent::ProxyEntryUpdated { entry } => {
assert_eq!(entry.name, "web");
assert_eq!(entry.listen_port, 443);
assert!(entry.allow_remote);
}
other => panic!("expected ProxyEntryUpdated, got {other:?}"),
}
}
#[test]
fn map_proxy_entry_removed() {
let event = koi_proxy::ProxyEvent::EntryRemoved {
name: "old-proxy".to_string(),
};
let mapped = map_proxy_event(event);
match mapped {
KoiEvent::ProxyEntryRemoved { name } => {
assert_eq!(name, "old-proxy");
}
other => panic!("expected ProxyEntryRemoved, got {other:?}"),
}
}
#[test]
fn map_join_error_produces_io_error() {
let io_err = std::io::Error::other("simulated join error");
let koi_err = KoiError::Io(io_err);
assert!(koi_err.to_string().contains("simulated join error"));
}
#[test]
fn builder_default_config() {
let builder = Builder::new();
let embedded = builder.build().expect("build should succeed");
assert!(embedded.config.mdns_enabled);
assert!(!embedded.config.http_enabled);
assert_eq!(embedded.config.http_port, 5641);
}
#[test]
fn builder_default_trait() {
let builder = Builder::default();
let embedded = builder.build().expect("build should succeed");
assert_eq!(embedded.config.service_endpoint, "http://127.0.0.1:5641");
}
#[test]
fn service_token_builder_sets_token() {
let embedded = Builder::new()
.service_token("secret-token")
.build()
.expect("build should succeed");
assert_eq!(
embedded.config.service_token.as_deref(),
Some("secret-token")
);
}
#[test]
fn endpoints_match_treats_localhost_as_loopback() {
assert!(endpoints_match(
"http://localhost:5641",
"http://127.0.0.1:5641"
));
assert!(endpoints_match(
"http://127.0.0.1:5641/",
"http://127.0.0.1:5641"
));
assert!(endpoints_match(
"HTTP://LOCALHOST:5641",
"http://127.0.0.1:5641"
));
}
#[test]
fn endpoints_match_rejects_different_hosts() {
assert!(!endpoints_match(
"http://127.0.0.1:5641",
"http://10.0.0.1:5641"
));
assert!(!endpoints_match(
"http://127.0.0.1:5641",
"http://127.0.0.1:9999"
));
}
#[test]
fn builder_fluent_overrides() {
let embedded = Builder::new()
.http(true)
.mdns(false)
.dns_enabled(false)
.health(true)
.certmesh(true)
.proxy(true)
.udp(true)
.http_port(9000)
.dashboard(true)
.api_docs(true)
.mdns_browser(true)
.announce_http(true)
.dns_auto_start(true)
.health_auto_start(true)
.proxy_auto_start(true)
.service_endpoint("http://10.0.0.1:8080")
.service_mode(ServiceMode::EmbeddedOnly)
.data_dir("/tmp/koi-test")
.build()
.expect("build should succeed");
assert!(embedded.config.http_enabled);
assert!(!embedded.config.mdns_enabled);
assert!(!embedded.config.dns_enabled);
assert!(embedded.config.health_enabled);
assert!(embedded.config.certmesh_enabled);
assert!(embedded.config.proxy_enabled);
assert!(embedded.config.udp_enabled);
assert_eq!(embedded.config.http_port, 9000);
assert!(embedded.config.dashboard_enabled);
assert!(embedded.config.api_docs_enabled);
assert!(embedded.config.mdns_browser_enabled);
assert!(embedded.config.announce_http);
assert!(embedded.config.dns_auto_start);
assert!(embedded.config.health_auto_start);
assert!(embedded.config.proxy_auto_start);
assert_eq!(embedded.config.service_endpoint, "http://10.0.0.1:8080");
assert_eq!(embedded.config.service_mode, ServiceMode::EmbeddedOnly);
assert_eq!(
embedded.config.data_dir,
Some(std::path::PathBuf::from("/tmp/koi-test"))
);
}
#[test]
fn orchestrator_and_certmesh_background_are_opt_in() {
let default_cfg = Builder::new().build().expect("build should succeed");
assert!(!default_cfg.config.orchestrator_enabled);
assert!(!default_cfg.config.certmesh_background_enabled);
let opted = Builder::new()
.runtime_auto()
.orchestrator(true)
.certmesh(true)
.certmesh_background(true)
.build()
.expect("build should succeed");
assert!(opted.config.orchestrator_enabled);
assert!(opted.config.certmesh_background_enabled);
}
#[test]
fn builder_dns_configure_closure() {
let embedded = Builder::new()
.dns(|b| b.port(5353).zone("home").local_ttl(120))
.build()
.expect("build should succeed");
assert_eq!(embedded.config.dns_config.port, 5353);
assert_eq!(embedded.config.dns_config.zone, "home");
assert_eq!(embedded.config.dns_config.local_ttl, 120);
}
#[test]
fn builder_event_handler() {
use std::sync::atomic::{AtomicBool, Ordering};
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
let embedded = Builder::new()
.events(move |_event| {
called_clone.store(true, Ordering::SeqCst);
})
.build()
.expect("build should succeed");
assert!(embedded.event_handler.is_some());
}
#[test]
fn builder_extra_firewall_ports() {
use koi_common::firewall::{FirewallPort, FirewallProtocol};
let extra = vec![FirewallPort::new("Custom", FirewallProtocol::Tcp, 12345)];
let _builder = Builder::new().extra_firewall_ports(extra);
}
#[test]
fn result_type_works_with_ok() {
let result: Result<i32> = Ok(42);
assert!(matches!(result, Ok(42)));
}
#[test]
fn result_type_works_with_err() {
let result: Result<i32> = Err(KoiError::DisabledCapability("test"));
assert!(result.is_err());
}
}