use std::path::Path;
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use aetherdb::runtime::EngineRuntime;
use aetherdb::{DatabaseRef, Node, OpenOptions};
use tracing::info;
use crate::kwt_replay::kdl_build::{self, HOST_DATA_PRESET};
use crate::kwt_replay::topology::{host_mode_from_env, DeployMode, Topology};
pub struct HostConfig {
pub bind: String,
pub data_dir: String,
pub replicate_hex: String,
pub control_key_hex: Option<String>,
pub tls: Option<TlsPaths>,
}
pub struct TlsPaths {
pub cert: std::path::PathBuf,
pub key: std::path::PathBuf,
pub ca: std::path::PathBuf,
}
pub enum HostHandle {
Replicate(DatabaseRef),
Cluster(Arc<Node>),
}
impl HostHandle {
#[must_use]
pub fn mode_label(&self) -> &'static str {
match self {
Self::Replicate(_) => "replicate",
Self::Cluster(_) => "cluster",
}
}
#[must_use]
pub fn bind_endpoint(&self) -> Option<String> {
match self {
Self::Replicate(db) => db.bus().bind_endpoint().map(|e| e.to_string()),
Self::Cluster(node) => node.bus().bind_endpoint().map(|e| e.to_string()),
}
}
pub fn shutdown(&self) -> Result<(), String> {
match self {
Self::Replicate(db) => db.close().map_err(|e| e.to_string()),
Self::Cluster(node) => node.shutdown().map_err(|e| e.to_string()),
}
}
}
pub fn host_config_from_env() -> Result<HostConfig, String> {
host_config_from_parts(
std::env::var("AETHERDB_HOST_BIND").ok(),
std::env::var("AETHERDB_HOST_DATA_DIR").ok(),
std::env::var("AETHERDB_HOST_REPLICATE_KEY_HEX")
.ok()
.or_else(|| std::env::var("AETHERDB_HUB_REPLICATE_KEY_HEX").ok()),
std::env::var("AETHERDB_HOST_CONTROL_KEY_HEX")
.ok()
.or_else(|| std::env::var("RUSTIS_CONTROL_KWT_KEY").ok()),
)
}
pub fn host_config_from_parts(
bind: Option<String>,
data_dir: Option<String>,
replicate_hex: Option<String>,
control_key_hex: Option<String>,
) -> Result<HostConfig, String> {
let bind = bind
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| "tcp://127.0.0.1:5555".into());
let data_dir = data_dir
.filter(|s| !s.trim().is_empty())
.unwrap_or_else(|| "/data".into());
let replicate_hex = replicate_hex
.filter(|s| !s.trim().is_empty())
.ok_or_else(|| "replicate_key_hex required (64 hex chars)".to_string())?;
let control_key_hex = control_key_hex.filter(|s| !s.trim().is_empty());
let tls = tls_paths_from_env()?;
warn_if_insecure_wire_bind(&bind, &tls);
Ok(HostConfig {
bind,
data_dir,
replicate_hex: replicate_hex.trim().to_string(),
control_key_hex,
tls,
})
}
fn tls_paths_from_env() -> Result<Option<TlsPaths>, String> {
let cert = first_env_path(&[
"AETHERDB_HOST_TLS_CERT_PATH",
"AETHERDB_HUB_TLS_CERT_PATH",
])?;
let key = first_env_path(&[
"AETHERDB_HOST_TLS_KEY_PATH",
"AETHERDB_HUB_TLS_KEY_PATH",
])?;
let ca = first_env_path(&["AETHERDB_HOST_TLS_CA_PATH", "AETHERDB_HUB_TLS_CA_PATH"])?;
match (cert, key, ca) {
(Some(cert), Some(key), Some(ca)) => Ok(Some(TlsPaths { cert, key, ca })),
(None, None, None) => Ok(None),
_ => Err(
"AetherDB host TLS requires AETHERDB_HOST_TLS_{CERT,KEY,CA}_PATH (all three)"
.into(),
),
}
}
fn first_env_path(names: &[&str]) -> Result<Option<std::path::PathBuf>, String> {
for name in names {
if let Ok(value) = std::env::var(name) {
if value.trim().is_empty() {
return Err(format!("{name} is empty"));
}
return Ok(Some(Path::new(value.trim()).into()));
}
}
Ok(None)
}
pub fn open_host(cfg: &HostConfig, topology: &Topology) -> Result<HostHandle, String> {
aetherdb::decode_master_key_hex(cfg.replicate_hex.trim())
.map_err(|e| format!("invalid replicate key hex: {e}"))?;
if let Some(ref hex) = cfg.control_key_hex {
aetherdb::decode_master_key_hex(hex.trim())
.map_err(|e| format!("invalid control key hex: {e}"))?;
}
if topology.mode == DeployMode::Cluster {
if cfg.control_key_hex.is_none() {
return Err(
"control_key_hex required in cluster mode (AETHERDB_HOST_CONTROL_KEY_HEX)"
.into(),
);
}
if cfg.tls.is_none() && !cluster_tls_insecure_allowed() {
return Err(
"cluster data nodes require TLS cert/key/ca paths \
(or dev-only AETHERDB_HOST_TLS_INSECURE=1)"
.into(),
);
}
}
let tls_block = cfg.tls.as_ref().map_or(String::new(), |tls| {
kdl_build::tls_block_from_paths(&tls.cert, &tls.key, &tls.ca)
});
let kdl = kdl_build::host_data_kdl(
topology,
&cfg.bind,
cfg.replicate_hex.trim(),
&tls_block,
cfg.control_key_hex.as_deref(),
);
std::fs::create_dir_all(&cfg.data_dir).map_err(|e| format!("create {}: {e}", cfg.data_dir))?;
let runtime = EngineRuntime::with_defaults().map_err(|e| e.to_string())?;
let replicate_key = aetherdb::decode_master_key_hex(cfg.replicate_hex.trim())
.map_err(|e| e.to_string())?
.to_vec();
let handle = if topology.mode == DeployMode::Cluster {
open_cluster_node(cfg, topology, &kdl, &runtime, replicate_key)?
} else {
open_replicate_hub(cfg, &kdl, &runtime, replicate_key)?
};
thread::sleep(Duration::from_millis(100));
if let Some(endpoint) = handle.bind_endpoint() {
info!(
%endpoint,
mode = handle.mode_label(),
"AetherDB data plane listening (in-process)"
);
}
Ok(handle)
}
impl HostHandle {
pub fn primary_database(&self, id: &str, kdl: &str) -> Result<DatabaseRef, String> {
match self {
Self::Replicate(db) => Ok(db.clone()),
Self::Cluster(node) => node
.spawn_db_str(id, kdl, HOST_DATA_PRESET)
.map_err(|e| e.to_string()),
}
}
}
pub fn run_host() -> Result<(), String> {
let cfg = host_config_from_env()?;
let mut topology = Topology::from_env();
topology.mode = host_mode_from_env();
let tls_block = cfg.tls.as_ref().map_or(String::new(), |tls| {
kdl_build::tls_block_from_paths(&tls.cert, &tls.key, &tls.ca)
});
let kdl = kdl_build::host_data_kdl(
&topology,
&cfg.bind,
cfg.replicate_hex.trim(),
&tls_block,
cfg.control_key_hex.as_deref(),
);
let handle = open_host(&cfg, &topology)?;
let _ = handle.primary_database(
&format!("nautalid-node-{}", topology.node_id),
&kdl,
)?;
park_forever();
}
fn open_replicate_hub(
cfg: &HostConfig,
kdl: &str,
runtime: &Arc<EngineRuntime>,
replicate_key: Vec<u8>,
) -> Result<HostHandle, String> {
let hub = DatabaseRef::open_str_with_options(
kdl,
HOST_DATA_PRESET,
OpenOptions {
runtime: Some(Arc::clone(runtime)),
transport_bind: Some(cfg.bind.clone()),
replication_handler: true,
replicate_master_key: Some(replicate_key),
data_dir: Some(Path::new(&cfg.data_dir).join("data")),
..OpenOptions::default()
},
)
.map_err(|e| e.to_string())?;
Ok(HostHandle::Replicate(hub))
}
fn open_cluster_node(
cfg: &HostConfig,
_topology: &Topology,
kdl: &str,
_runtime: &Arc<EngineRuntime>,
replicate_key: Vec<u8>,
) -> Result<HostHandle, String> {
let preset = aetherdb::preset::load_named_preset(kdl, HOST_DATA_PRESET)
.map_err(|e| e.to_string())?;
let mut node_config = aetherdb::node_config_from_preset(&preset).map_err(|e| e.to_string())?;
node_config.replicate_master_key = Some(replicate_key);
if let Some(hex) = &cfg.control_key_hex {
node_config.control_kwt_master_key = Some(
aetherdb::decode_master_key_hex(hex.trim())
.map_err(|e| e.to_string())?
.to_vec(),
);
}
let node = Node::start(node_config).map_err(|e| e.to_string())?;
Ok(HostHandle::Cluster(node))
}
fn cluster_tls_insecure_allowed() -> bool {
crate::surfaces::env_enabled("AETHERDB_HOST_TLS_INSECURE")
}
fn warn_if_insecure_wire_bind(bind: &str, tls: &Option<TlsPaths>) {
if tls.is_some() || cluster_tls_insecure_allowed() {
return;
}
if wire_bind_is_non_loopback(bind) {
tracing::warn!(
bind,
"Aether wire bind is reachable off loopback without TLS; \
set AETHERDB_HOST_TLS_* paths, bind to 127.0.0.1, or use AETHERDB_HOST_TLS_INSECURE=1 for dev only"
);
}
}
fn wire_bind_is_non_loopback(bind: &str) -> bool {
let endpoint = bind.trim();
let host = endpoint
.strip_prefix("tcp://")
.and_then(|rest| rest.rsplit_once(':').map(|(h, _)| h))
.unwrap_or(endpoint);
let host = host.trim_matches(|c| c == '[' || c == ']');
!(matches!(host, "127.0.0.1" | "localhost" | "::1") || host.starts_with("127."))
}
#[cfg(test)]
mod tests {
use super::wire_bind_is_non_loopback;
#[test]
fn wire_bind_detects_non_loopback_hosts() {
assert!(!wire_bind_is_non_loopback("tcp://127.0.0.1:5555"));
assert!(!wire_bind_is_non_loopback("tcp://localhost:5555"));
assert!(!wire_bind_is_non_loopback("tcp://[::1]:5555"));
assert!(!wire_bind_is_non_loopback("tcp://127.0.0.2:5555"));
assert!(wire_bind_is_non_loopback("tcp://0.0.0.0:5555"));
assert!(wire_bind_is_non_loopback("tcp://192.168.1.10:5555"));
assert!(wire_bind_is_non_loopback("tcp://[::]:5555"));
}
}
fn park_forever() -> ! {
loop {
thread::park_timeout(Duration::from_secs(3600));
}
}