use std::sync::OnceLock;
use thiserror::Error;
use crate::config_manager;
use crate::db_manager;
use crate::handler_registry;
use crate::http_server;
use crate::logger;
use crate::plugin_manager;
use crate::types::KdcCallbacks;
#[derive(Error, Debug)]
pub enum ProcessError {
#[error("config error: {0}")]
ConfigError(#[from] config_manager::ConfigError),
#[error("logger error: {0}")]
LoggerError(String),
#[error("token verifier error: {0}")]
TokenVerifierError(String),
#[error("database error: {0}")]
DbError(String),
#[error("http error: {0}")]
HttpError(String),
#[error("runtime error: {0}")]
RuntimeError(String),
}
static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
fn get_runtime() -> Result<&'static tokio::runtime::Runtime, ProcessError> {
match RUNTIME.get() {
Some(rt) => Ok(rt),
None => {
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.map_err(|e| {
ProcessError::RuntimeError(format!(
"failed to create tokio runtime: {}",
e
))
})?;
if RUNTIME.set(rt).is_err() {
return Err(ProcessError::RuntimeError(
"runtime initialization race: another thread set the runtime first".to_string(),
));
}
RUNTIME.get().ok_or(ProcessError::RuntimeError(
"runtime not available after initialization".to_string(),
))
}
}
}
pub fn init() -> Result<(), ProcessError> {
config_manager::init_config_center()?;
logger::init_logger()
.map_err(|e| {
eprintln!("[kdc_agent] logger init failed: {}", e);
ProcessError::LoggerError(e.to_string())
})?;
crate::token_verifier::init_token_verifier()
.map_err(|e| ProcessError::TokenVerifierError(e.to_string()))?;
db_manager::init_db_manager()
.map_err(|e| ProcessError::DbError(format!("database init failed: {}", e)))?;
let rt = get_runtime()?;
rt.spawn(async {
if let Err(e) = http_server::start_http_server().await {
log::error!("HTTP server error: {}", e);
}
});
if let Err(e) = plugin_manager::load_plugins() {
log::error!("plugin load failed: {}", e);
}
let callbacks = KdcCallbacks {
kdclog: logger::kdc_log_callback,
kdc_register_request_handle: handler_registry::register_handler,
kdc_unregister_request_handle: handler_registry::unregister_handler,
kdc_db_execute: db_manager::kdc_db_execute_sql,
kdc_db_query: db_manager::kdc_db_query_sql,
kdc_db_free: db_manager::kdc_db_free,
kdc_db_sync: db_manager::kdc_db_sync,
};
if let Err(e) = plugin_manager::set_callbacks_for_all_plugins(callbacks) {
log::error!("plugin set callbacks failed: {}", e);
}
if let Err(e) = plugin_manager::init_all_plugins() {
log::error!("plugin init failed: {}", e);
}
log::info!("kdc_agent initialization complete");
Ok(())
}
pub fn start() -> Result<(), ProcessError> {
if let Err(e) = plugin_manager::start_all_plugins() {
log::error!("plugin start failed: {}", e);
}
get_runtime()?.block_on(async {
let mut sigterm = tokio::signal::unix::signal(
tokio::signal::unix::SignalKind::terminate(),
)
.map_err(|e| {
ProcessError::HttpError(format!("failed to register SIGTERM handler: {}", e))
})?;
sigterm.recv().await;
Ok::<(), ProcessError>(())
})?;
Ok(())
}
pub fn stop() -> Result<(), ProcessError> {
log::info!("kdc_agent stopping...");
http_server::stop_http_server()
.map_err(|e| ProcessError::HttpError(e.to_string()))?;
if let Err(e) = plugin_manager::stop_all_plugins() {
log::error!("plugin stop failed: {}", e);
}
if let Err(e) = plugin_manager::uninit_all_plugins() {
log::error!("plugin uninit failed: {}", e);
}
log::info!("kdc_agent stopped");
Ok(())
}
#[cfg_attr(coverage_nightly, coverage(off))]
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_process_error_config() {
let err = ProcessError::ConfigError(config_manager::ConfigError::AlreadyInitialized);
assert!(err.to_string().contains("ConfigCenter already initialized"));
}
#[test]
fn test_process_error_logger() {
let err = ProcessError::LoggerError("log fail".to_string());
assert!(err.to_string().contains("log fail"));
}
#[test]
fn test_process_error_token_verifier() {
let err = ProcessError::TokenVerifierError("token fail".to_string());
assert!(err.to_string().contains("token fail"));
}
#[test]
fn test_process_error_db() {
let err = ProcessError::DbError("db fail".to_string());
assert!(err.to_string().contains("db fail"));
}
#[test]
fn test_process_error_http() {
let err = ProcessError::HttpError("http fail".to_string());
assert!(err.to_string().contains("http fail"));
}
#[test]
fn test_process_error_runtime() {
let err = ProcessError::RuntimeError("runtime fail".to_string());
assert!(err.to_string().contains("runtime fail"));
}
#[test]
fn test_process_error_debug() {
let err = ProcessError::RuntimeError("test".to_string());
let debug = format!("{:?}", err);
assert!(debug.contains("RuntimeError"));
}
#[test]
fn test_init_no_config_file() {
let result = init();
assert!(result.is_err());
}
#[test]
fn test_stop_no_server() {
let result = stop();
assert!(result.is_err());
match result {
Err(ProcessError::HttpError(msg)) => {
assert!(msg.contains("server not started"));
}
other => panic!("expected HttpError, got {:?}", other),
}
}
#[test]
fn test_get_runtime_creates() {
let result = get_runtime();
assert!(result.is_ok());
}
#[test]
fn test_get_runtime_idempotent() {
let rt1 = get_runtime().unwrap() as *const _;
let rt2 = get_runtime().unwrap() as *const _;
assert_eq!(rt1, rt2, "should return the same singleton runtime");
}
#[test]
fn test_process_error_from_config_error() {
let config_err = config_manager::ConfigError::ValidationError("test".to_string());
let process_err: ProcessError = config_err.into();
assert!(process_err.to_string().contains("test"));
}
#[test]
fn test_process_error_debug_all_variants() {
assert!(format!("{:?}", ProcessError::ConfigError(config_manager::ConfigError::AlreadyInitialized)).contains("ConfigError"));
assert!(format!("{:?}", ProcessError::LoggerError("l".into())).contains("LoggerError"));
assert!(format!("{:?}", ProcessError::TokenVerifierError("t".into())).contains("TokenVerifierError"));
assert!(format!("{:?}", ProcessError::DbError("d".into())).contains("DbError"));
assert!(format!("{:?}", ProcessError::HttpError("h".into())).contains("HttpError"));
assert!(format!("{:?}", ProcessError::RuntimeError("r".into())).contains("RuntimeError"));
}
#[test]
fn test_init_config_file_missing() {
match init() {
Err(ProcessError::ConfigError(_)) | Err(ProcessError::LoggerError(_)) => {}
Err(ProcessError::TokenVerifierError(_)) => {}
Err(ProcessError::DbError(_)) => {}
Ok(()) => {}
other => panic!("unexpected result: {:?}", other),
}
}
#[test]
fn test_stop_returns_http_error() {
match stop() {
Err(ProcessError::HttpError(msg)) => {
assert!(msg.contains("server not started"));
}
Err(e) => panic!("expected HttpError, got: {:?}", e),
Ok(()) => {}
}
}
#[test]
fn test_get_runtime_spawn_task() {
let rt = get_runtime().unwrap();
let handle = rt.handle();
let result = handle.block_on(async { 42 });
assert_eq!(result, 42);
}
#[test]
fn test_process_error_from_io_error() {
let io_err = std::io::Error::new(std::io::ErrorKind::NotFound, "file missing");
let config_err = config_manager::ConfigError::IoError(io_err);
let process_err: ProcessError = config_err.into();
let msg = format!("{}", process_err);
assert!(msg.contains("file missing"));
}
#[test]
fn test_process_error_from_parse_error() {
let parse_err: serde_json::Error = serde_json::from_str::<serde_json::Value>("{bad}").unwrap_err();
let config_err = config_manager::ConfigError::ParseError(parse_err);
let process_err: ProcessError = config_err.into();
assert!(process_err.to_string().contains("config error"));
}
#[test]
fn test_init_logger_failure() {
fn mock_init_config_center() -> Result<(), config_manager::ConfigError> {
Ok(())
}
fn mock_init_logger() -> Result<(), crate::logger::LoggerError> {
Err(crate::logger::LoggerError::InitError("logger failed".to_string()))
}
let _m1 = mockrs::mock!(crate::config_manager::init_config_center, mock_init_config_center);
let _m2 = mockrs::mock!(crate::logger::init_logger, mock_init_logger);
let result = init();
match result {
Err(ProcessError::LoggerError(msg)) => {
assert!(msg.contains("logger failed"));
}
other => panic!("expected LoggerError, got {:?}", other),
}
}
#[test]
fn test_init_token_verifier_failure() {
fn mock_init_config_center() -> Result<(), config_manager::ConfigError> {
Ok(())
}
fn mock_init_logger() -> Result<(), crate::logger::LoggerError> {
Ok(())
}
fn mock_init_token_verifier() -> Result<(), crate::token_verifier::TokenVerifierError> {
Err(crate::token_verifier::TokenVerifierError::NotInitialized)
}
let _m1 = mockrs::mock!(crate::config_manager::init_config_center, mock_init_config_center);
let _m2 = mockrs::mock!(crate::logger::init_logger, mock_init_logger);
let _m3 = mockrs::mock!(crate::token_verifier::init_token_verifier, mock_init_token_verifier);
let result = init();
match result {
Err(ProcessError::TokenVerifierError(msg)) => {
assert!(msg.contains("RA public key not initialized"));
}
other => panic!("expected TokenVerifierError, got {:?}", other),
}
}
#[test]
fn test_init_db_manager_failure() {
fn mock_init_config_center() -> Result<(), config_manager::ConfigError> {
Ok(())
}
fn mock_init_logger() -> Result<(), crate::logger::LoggerError> {
Ok(())
}
fn mock_init_token_verifier() -> Result<(), crate::token_verifier::TokenVerifierError> {
Ok(())
}
fn mock_init_db_manager() -> Result<(), crate::db_manager::DbError> {
Err(crate::db_manager::DbError::NotInitialized)
}
let _m1 = mockrs::mock!(crate::config_manager::init_config_center, mock_init_config_center);
let _m2 = mockrs::mock!(crate::logger::init_logger, mock_init_logger);
let _m3 = mockrs::mock!(crate::token_verifier::init_token_verifier, mock_init_token_verifier);
let _m4 = mockrs::mock!(crate::db_manager::init_db_manager, mock_init_db_manager);
let result = init();
match result {
Err(ProcessError::DbError(msg)) => {
assert!(msg.contains("database init failed"));
}
other => panic!("expected DbError, got {:?}", other),
}
}
#[test]
fn test_init_load_plugins_failure() {
fn mock_init_config_center() -> Result<(), config_manager::ConfigError> {
Ok(())
}
fn mock_init_logger() -> Result<(), crate::logger::LoggerError> {
Ok(())
}
fn mock_init_token_verifier() -> Result<(), crate::token_verifier::TokenVerifierError> {
Ok(())
}
fn mock_init_db_manager() -> Result<(), crate::db_manager::DbError> {
Ok(())
}
fn mock_load_plugins() -> Result<(), crate::plugin_manager::PluginError> {
Err(crate::plugin_manager::PluginError::LoadFailed {
name: "test_plugin".to_string(),
reason: "load error".to_string(),
})
}
fn mock_set_callbacks(
_callbacks: crate::types::KdcCallbacks,
) -> Result<(), crate::plugin_manager::PluginError> {
Ok(())
}
fn mock_init_all_plugins() -> Result<(), crate::plugin_manager::PluginError> {
Ok(())
}
let _m1 = mockrs::mock!(crate::config_manager::init_config_center, mock_init_config_center);
let _m2 = mockrs::mock!(crate::logger::init_logger, mock_init_logger);
let _m3 = mockrs::mock!(crate::token_verifier::init_token_verifier, mock_init_token_verifier);
let _m4 = mockrs::mock!(crate::db_manager::init_db_manager, mock_init_db_manager);
let _m5 = mockrs::mock!(crate::plugin_manager::load_plugins, mock_load_plugins);
let _m6 = mockrs::mock!(crate::plugin_manager::set_callbacks_for_all_plugins, mock_set_callbacks);
let _m7 = mockrs::mock!(crate::plugin_manager::init_all_plugins, mock_init_all_plugins);
let result = init();
assert!(result.is_ok(), "init should return Ok when load_plugins fails, got {:?}", result);
}
#[test]
fn test_start_get_runtime_failure() {
fn mock_start_all_plugins() -> Result<(), crate::plugin_manager::PluginError> {
Ok(())
}
fn mock_get_runtime() -> Result<&'static tokio::runtime::Runtime, ProcessError> {
Err(ProcessError::RuntimeError("mock runtime failure".to_string()))
}
let _m1 = mockrs::mock!(
crate::plugin_manager::start_all_plugins,
mock_start_all_plugins
);
let _m2 = mockrs::mock!(get_runtime, mock_get_runtime);
let result = start();
assert!(result.is_err());
match result {
Err(ProcessError::RuntimeError(msg)) => {
assert!(msg.contains("mock runtime failure"));
}
other => panic!("expected RuntimeError, got {:?}", other),
}
}
#[test]
fn test_stop_with_plugin_errors() {
fn mock_stop_http_server() -> Result<(), crate::http_server::ServerError> {
Ok(())
}
fn mock_stop_all_plugins() -> Result<(), crate::plugin_manager::PluginError> {
Err(crate::plugin_manager::PluginError::LifecycleError {
name: "test_plugin".to_string(),
code: 1,
})
}
fn mock_uninit_all_plugins() -> Result<(), crate::plugin_manager::PluginError> {
Err(crate::plugin_manager::PluginError::LifecycleError {
name: "test_plugin".to_string(),
code: 2,
})
}
let _m1 = mockrs::mock!(
crate::http_server::stop_http_server,
mock_stop_http_server
);
let _m2 = mockrs::mock!(
crate::plugin_manager::stop_all_plugins,
mock_stop_all_plugins
);
let _m3 = mockrs::mock!(
crate::plugin_manager::uninit_all_plugins,
mock_uninit_all_plugins
);
let result = stop();
assert!(
result.is_ok(),
"stop should succeed even when plugins report errors"
);
}
#[test]
fn test_start_sigterm_completes() {
fn mock_start_all_plugins() -> Result<(), crate::plugin_manager::PluginError> {
Ok(())
}
let _m1 = mockrs::mock!(
crate::plugin_manager::start_all_plugins,
mock_start_all_plugins
);
let pid = std::process::id();
let handle = std::thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_millis(200));
let _ = std::process::Command::new("kill")
.args(["-s", "TERM", &pid.to_string()])
.output();
});
let result = start();
let _ = handle.join();
assert!(
result.is_ok(),
"start should return Ok after receiving SIGTERM, got {:?}",
result
);
}
}