use crate::error::Result as AdminResult;
use crate::metrics::collect_system_metrics;
use crate::realm::realm_to_proto;
use actrix_proto::NodeAdminService;
use actrix_proto::{
ConfigOverrideEntry as ProtoConfigOverrideEntry, ConfigType, CreateRealmRequest,
CreateRealmResponse, DeleteConfigOverrideRequest, DeleteConfigOverrideResponse,
DeleteRealmRequest, DeleteRealmResponse, GetConfigRequest, GetConfigResponse,
GetNodeInfoRequest, GetNodeInfoResponse, GetRealmRequest, GetRealmResponse,
ListConfigOverridesRequest, ListConfigOverridesResponse, ListRealmsRequest, ListRealmsResponse,
RealmInfo, ServiceStatus, SetConfigOverrideRequest, SetConfigOverrideResponse, ShutdownRequest,
ShutdownResponse, SystemMetrics, UpdateConfigRequest, UpdateConfigResponse, UpdateRealmRequest,
UpdateRealmResponse,
};
use chrono::Utc;
use platform::ServiceCollector;
use platform::config::ActrixConfig;
use platform::config::config_store::{ConfigOverride, ConfigOverrideStore};
use platform::config::registry;
use platform::config::resolver::{self, ResolvedField};
use platform::realm::{
DEFAULT_REALM_SECRET_PREVIOUS_GRACE_SECS, hash_realm_secret, rotate_realm_secret,
};
use platform::realm::{Realm, RealmStatus};
use serde::Serialize;
use serde_json::Value as JsonValue;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use std::collections::HashMap;
use std::future::Future;
use std::path::PathBuf;
use std::pin::Pin;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::RwLock;
use tonic::{Request, Response, Status};
type MetricsFuture = Pin<Box<dyn Future<Output = AdminResult<SystemMetrics>> + Send>>;
type MetricsProvider = Arc<dyn Fn() -> MetricsFuture + Send + Sync>;
type ShutdownFuture = Pin<Box<dyn Future<Output = AdminResult<()>> + Send>>;
type ShutdownHandler =
Arc<dyn Fn(bool, Option<i32>, Option<String>) -> ShutdownFuture + Send + Sync>;
type ReloadFuture = Pin<Box<dyn Future<Output = AdminResult<bool>> + Send>>;
type ReloadHandler = Arc<dyn Fn() -> ReloadFuture + Send + Sync>;
type GrpcResult<T> = std::result::Result<T, Status>;
#[derive(Hash, Eq, PartialEq, Clone)]
struct ConfigKey {
config_type: i32,
key: String,
}
#[derive(Clone)]
pub struct AdminApiService {
node_id: String,
name: String,
location_tag: String,
version: String,
config_store: Arc<RwLock<HashMap<ConfigKey, String>>>,
override_store: Option<Arc<ConfigOverrideStore>>,
metrics_provider: MetricsProvider,
shutdown_handler: Option<ShutdownHandler>,
reload_handler: Option<ReloadHandler>,
service_collector: ServiceCollector,
started_at: Instant,
running_config: Option<ActrixConfig>,
toml_content: Option<String>,
config_path: Option<PathBuf>,
}
impl AdminApiService {
pub fn new(
node_id: impl Into<String>,
name: impl Into<String>,
location_tag: impl Into<String>,
version: impl Into<String>,
service_collector: ServiceCollector,
) -> AdminResult<Self> {
Ok(Self {
node_id: node_id.into(),
name: name.into(),
location_tag: location_tag.into(),
version: version.into(),
config_store: Arc::new(RwLock::new(HashMap::new())),
override_store: None,
metrics_provider: Arc::new(|| Box::pin(async { collect_system_metrics().await })),
shutdown_handler: None,
reload_handler: None,
service_collector,
started_at: Instant::now(),
running_config: None,
toml_content: None,
config_path: None,
})
}
pub fn with_metrics_provider<F, Fut>(mut self, provider: F) -> Self
where
F: Fn() -> Fut + Send + Sync + 'static,
Fut: Future<Output = AdminResult<SystemMetrics>> + Send + 'static,
{
self.metrics_provider = Arc::new(move || {
let fut = provider();
Box::pin(fut)
});
self
}
pub fn with_shutdown_handler<F, Fut>(mut self, handler: F) -> Self
where
F: Fn(bool, Option<i32>, Option<String>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = AdminResult<()>> + Send + 'static,
{
self.shutdown_handler = Some(Arc::new(move |graceful, timeout, reason| {
let fut = handler(graceful, timeout, reason);
Box::pin(fut)
}));
self
}
pub fn with_override_store(mut self, store: Arc<ConfigOverrideStore>) -> Self {
self.override_store = Some(store);
self
}
pub fn with_reload_handler<F, Fut>(mut self, handler: F) -> Self
where
F: Fn() -> Fut + Send + Sync + 'static,
Fut: Future<Output = AdminResult<bool>> + Send + 'static,
{
self.reload_handler = Some(Arc::new(move || {
let fut = handler();
Box::pin(fut)
}));
self
}
pub fn with_running_config(mut self, config: ActrixConfig) -> Self {
self.running_config = Some(config);
self
}
pub fn with_toml_content(mut self, content: String) -> Self {
self.toml_content = Some(content);
self
}
pub fn with_config_path(mut self, path: PathBuf) -> Self {
self.config_path = Some(path);
self
}
pub fn override_store(&self) -> Option<&Arc<ConfigOverrideStore>> {
self.override_store.as_ref()
}
fn build_config_key(config_type: ConfigType, key: String) -> ConfigKey {
ConfigKey {
config_type: config_type as i32,
key,
}
}
async fn load_realm(&self, realm_id: u32) -> GrpcResult<Realm> {
let realm = Realm::get(realm_id)
.await
.map_err(|e| Status::internal(format!("Failed to load realm: {e}")))?;
realm.ok_or_else(|| Status::not_found(format!("Realm not found: {realm_id}")))
}
async fn collect_metrics(&self) -> GrpcResult<SystemMetrics> {
(self.metrics_provider)()
.await
.map_err(|e| Status::internal(format!("Failed to collect metrics: {e}")))
}
pub async fn service_statuses(&self) -> Vec<ServiceStatus> {
self.service_collector.all_statuses().await
}
async fn overrides_list(&self) -> Vec<ConfigOverride> {
if let Some(store) = &self.override_store {
store.list_all().await.unwrap_or_default()
} else {
Vec::new()
}
}
fn toml_content_ref(&self) -> &str {
self.toml_content.as_deref().unwrap_or("")
}
fn require_running_config(&self) -> GrpcResult<&ActrixConfig> {
self.running_config
.as_ref()
.ok_or_else(|| Status::failed_precondition("Running config not available"))
}
pub async fn resolve_node_name(&self) -> String {
let overrides = self.overrides_list().await;
let fields = resolver::resolve_for_service("platform", self.toml_content_ref(), &overrides);
fields
.iter()
.find(|f| f.key == "name")
.map(|f| f.effective_value.trim().to_string())
.filter(|n| !n.is_empty())
.unwrap_or_else(|| {
self.running_config
.as_ref()
.map(|c| c.name.clone())
.unwrap_or_else(|| self.name.clone())
})
}
pub async fn health_info(&self) -> HealthInfo {
let node = self.resolve_node_name().await;
HealthInfo {
status: "healthy".to_string(),
node,
version: self.version.clone(),
}
}
pub async fn list_overrides_direct(&self) -> GrpcResult<Vec<ConfigOverride>> {
let store = self
.override_store
.as_ref()
.ok_or_else(|| Status::failed_precondition("Override store not available"))?;
store
.list_all()
.await
.map_err(|e| Status::internal(format!("Failed to list overrides: {e}")))
}
pub async fn set_override_direct(&self, key: &str, value: &str, by: &str) -> GrpcResult<()> {
let store = self
.override_store
.as_ref()
.ok_or_else(|| Status::failed_precondition("Override store not available"))?;
store
.set(key, value, by)
.await
.map_err(|e| Status::invalid_argument(e.to_string()))
}
pub async fn delete_override_direct(&self, key: &str) -> GrpcResult<bool> {
let store = self
.override_store
.as_ref()
.ok_or_else(|| Status::failed_precondition("Override store not available"))?;
store
.delete(key)
.await
.map_err(|e| Status::internal(format!("Failed to delete override: {e}")))
}
pub async fn get_platform_detail_direct(&self) -> GrpcResult<PlatformDetail> {
let cfg = self.require_running_config()?;
let http_bind = cfg.bind.http.as_ref().map(|h| {
serde_json::json!({
"ip": h.ip, "port": h.port,
"domain_name": h.domain_name, "advertised_ip": h.advertised_ip,
"advertised_port": h.effective_advertised_port(),
"cert": h.cert, "key": h.key,
"is_tls": h.is_tls(),
})
});
let config = serde_json::json!({
"enable": cfg.enable,
"name": cfg.name,
"env": cfg.env,
"location_tag": cfg.location_tag,
"sqlite_path": cfg.sqlite_path.display().to_string(),
"bind": {
"http": http_bind,
"ice": {
"ip": cfg.bind.ice.ip.to_string(),
"port": cfg.bind.ice.port,
"advertised_ip": cfg.bind.ice.advertised_ip.to_string(),
"advertised_port": cfg.bind.ice.advertised_port,
},
},
"turn": {
"relay_port_range": cfg.turn.relay_port_range,
},
"recording": {
"service_name": cfg.recording.service_name,
},
"control": {
"head": cfg.control.head,
"admin_ui": {
"session_expiry_secs": cfg.control.admin_ui.session_expiry_secs,
},
},
});
let overrides = self.overrides_list().await;
let config_fields =
resolver::resolve_for_service("platform", self.toml_content_ref(), &overrides);
Ok(PlatformDetail {
config,
config_fields,
})
}
pub async fn get_service_detail_direct(&self, name: &str) -> GrpcResult<ServiceDetail> {
let type_id = service_name_to_type_id(name)
.ok_or_else(|| Status::not_found(format!("Unknown service: {name}")))?;
let cfg = self.require_running_config()?;
let enabled = match name {
"stun" => cfg.is_stun_enabled(),
"turn" => cfg.is_turn_enabled(),
"signaling" => cfg.is_signaling_enabled(),
"ais" => cfg.is_ais_enabled(),
"signer" => cfg.is_signer_enabled(),
_ => false,
};
let statuses = self.service_statuses().await;
let status = statuses.iter().find(|s| s.r#type == type_id).map(|s| {
serde_json::json!({
"name": s.name,
"type": s.r#type,
"is_healthy": s.is_healthy,
"active_connections": s.active_connections,
"total_requests": s.total_requests,
"failed_requests": s.failed_requests,
"average_latency_ms": s.average_latency_ms,
"url": s.url,
"port": s.port,
"domain": s.domain,
})
});
let config: Option<JsonValue> = match name {
"stun" => Some(serde_json::json!({
"bind_ip": cfg.bind.ice.ip,
"bind_port": cfg.bind.ice.port,
"advertised_ip": cfg.bind.ice.advertised_ip,
"advertised_port": cfg.bind.ice.advertised_port,
})),
"turn" => Some(serde_json::json!({
"bind_ip": cfg.bind.ice.ip,
"bind_port": cfg.bind.ice.port,
"advertised_ip": cfg.bind.ice.advertised_ip,
"advertised_port": cfg.bind.ice.advertised_port,
"relay_port_range": cfg.turn.relay_port_range,
"realm": cfg.turn.realm,
})),
"signaling" => cfg.services.signaling.as_ref().map(|s| {
serde_json::json!({
"ws_path": s.server.ws_path,
"rate_limit": {
"connection": {
"enabled": s.server.rate_limit.connection.enabled,
"per_minute": s.server.rate_limit.connection.per_minute,
"burst_size": s.server.rate_limit.connection.burst_size,
"max_concurrent_per_ip": s.server.rate_limit.connection.max_concurrent_per_ip,
},
"message": {
"enabled": s.server.rate_limit.message.enabled,
"per_second": s.server.rate_limit.message.per_second,
"burst_size": s.server.rate_limit.message.burst_size,
}
},
"dependencies": {
"signer": s.dependencies.signer.as_ref().map(|k| serde_json::json!({"endpoint": k.endpoint})),
"ais": s.dependencies.ais.as_ref().map(|a| serde_json::json!({"endpoint": a.endpoint})),
}
})
}),
"ais" => cfg.services.ais.as_ref().map(|a| {
serde_json::json!({
"token_ttl_secs": a.server.token_ttl_secs,
"signaling_heartbeat_interval_secs": a.server.signaling_heartbeat_interval_secs,
"dependencies": {
"signer": a.dependencies.signer.as_ref().map(|k| serde_json::json!({"endpoint": k.endpoint})),
}
})
}),
"signer" => cfg.services.signer.as_ref().map(|k| {
serde_json::json!({
"storage_backend": format!("{:?}", k.storage.backend),
"key_ttl_seconds": k.storage.key_ttl_seconds,
"tolerance_seconds": k.tolerance_seconds,
})
}),
_ => None,
};
let config_fields = if !name.is_empty() {
let overrides = self.overrides_list().await;
Some(resolver::resolve_for_service(
name,
self.toml_content_ref(),
&overrides,
))
} else {
None
};
Ok(ServiceDetail {
enabled,
status,
config,
config_fields,
})
}
pub fn get_registry_direct(&self) -> &'static [registry::ConfigFieldDef] {
registry::all_fields()
}
pub async fn get_config_file_direct(&self) -> GrpcResult<ConfigFileContent> {
let path = self
.config_path
.as_ref()
.ok_or_else(|| Status::failed_precondition("Config path not available"))?;
let path_display = path.display().to_string();
let content = tokio::fs::read_to_string(path)
.await
.map_err(|e| Status::internal(format!("Failed to read config file: {e}")))?;
Ok(ConfigFileContent {
content,
path: path_display,
})
}
pub async fn save_config_file_direct(&self, content: &str) -> GrpcResult<()> {
let path = self
.config_path
.as_ref()
.ok_or_else(|| Status::failed_precondition("Config path not available"))?;
let new_config = ActrixConfig::from_toml(content)
.map_err(|e| Status::invalid_argument(format!("Invalid TOML: {e}")))?;
if let Err(errors) = new_config.validate() {
let non_warnings: Vec<_> = errors
.iter()
.filter(|e| !e.starts_with("Warning:"))
.cloned()
.collect();
if !non_warnings.is_empty() {
return Err(Status::invalid_argument(format!(
"Validation failed: {}",
non_warnings.join("; ")
)));
}
}
tokio::fs::write(path, content.as_bytes())
.await
.map_err(|e| Status::internal(format!("Failed to write: {e}")))?;
platform::recording::info!("Config file saved via admin API");
Ok(())
}
pub async fn reload_direct(&self) -> GrpcResult<bool> {
platform::recording::info!("Reload requested via admin API, sending SIGHUP to self");
#[cfg(unix)]
{
let pid = std::process::id() as i32;
let ret = unsafe { nix::libc::kill(pid, nix::libc::SIGHUP) };
if ret == 0 {
Ok(true)
} else {
Err(Status::internal("Failed to send SIGHUP"))
}
}
#[cfg(not(unix))]
{
Err(Status::unimplemented(
"Reload not supported on this platform",
))
}
}
pub async fn get_signer_keys_direct(&self) -> GrpcResult<KsKeysResult> {
let cfg = self.require_running_config()?;
let db_path = cfg.sqlite_path.join("signer_keys.db");
if !db_path.exists() {
return Ok(KsKeysResult {
keys: vec![],
total_count: 0,
});
}
let url = format!("sqlite:{}?mode=ro", db_path.display());
let options = SqliteConnectOptions::from_str(&url)
.map_err(|e| Status::internal(format!("DB connect error: {e}")))?
.read_only(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
.await
.map_err(|e| Status::internal(format!("DB pool error: {e}")))?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs() as i64;
let rows = sqlx::query_as::<_, (i64, i32, i64, i64)>(
"SELECT key_id, length(public_key) as pk_size, created_at, expires_at FROM keys ORDER BY created_at DESC LIMIT 5",
)
.fetch_all(&pool)
.await
.map_err(|e| Status::internal(format!("Query error: {e}")))?;
let total_count = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM keys")
.fetch_one(&pool)
.await
.map(|r| r.0)
.unwrap_or(0);
let keys = rows
.iter()
.map(|(key_id, pk_size, created_at, expires_at)| {
let is_expired = *expires_at > 0 && *expires_at < now;
KeyInfo {
key_id: *key_id,
pk_size: *pk_size,
created_at: Some(*created_at),
fetched_at: None,
expires_at: *expires_at,
tolerance_seconds: None,
is_expired,
}
})
.collect();
Ok(KsKeysResult { keys, total_count })
}
pub async fn cleanup_signer_keys_direct(&self) -> GrpcResult<KsCleanupResult> {
let cfg = self.require_running_config()?;
let db_path = cfg.sqlite_path.join("signer_keys.db");
if !db_path.exists() {
return Ok(KsCleanupResult {
deleted: 0,
remaining: 0,
tolerance_seconds: 0,
});
}
let url = format!("sqlite:{}", db_path.display());
let options = SqliteConnectOptions::from_str(&url)
.map_err(|e| Status::internal(format!("DB connect error: {e}")))?;
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
.await
.map_err(|e| Status::internal(format!("DB pool error: {e}")))?;
let tolerance = cfg
.services
.signer
.as_ref()
.map(|k| k.tolerance_seconds)
.unwrap_or(3600);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs() as i64;
let cutoff = now - tolerance as i64;
let result = sqlx::query("DELETE FROM keys WHERE expires_at > 0 AND expires_at < ?")
.bind(cutoff)
.execute(&pool)
.await
.map_err(|e| Status::internal(format!("Cleanup error: {e}")))?;
let remaining = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM keys")
.fetch_one(&pool)
.await
.map(|r| r.0)
.unwrap_or(0);
Ok(KsCleanupResult {
deleted: result.rows_affected(),
remaining,
tolerance_seconds: tolerance,
})
}
pub async fn get_ais_keys_direct(&self) -> GrpcResult<Vec<KeyInfo>> {
let cfg = self.require_running_config()?;
let db_path = cfg.sqlite_path.join("ais_keys.db");
if !db_path.exists() {
return Ok(vec![]);
}
let url = format!("sqlite:{}?mode=ro", db_path.display());
let options = SqliteConnectOptions::from_str(&url)
.map_err(|e| Status::internal(format!("DB connect error: {e}")))?
.read_only(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
.await
.map_err(|e| Status::internal(format!("DB pool error: {e}")))?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs() as i64;
let row = sqlx::query_as::<_, (i64, i32, i64, i64, i64)>(
"SELECT key_id, length(public_key) as pk_size, fetched_at, expires_at, tolerance_seconds FROM current_key WHERE id = 1",
)
.fetch_optional(&pool)
.await
.map_err(|e| Status::internal(format!("Query error: {e}")))?;
match row {
Some((key_id, pk_size, fetched_at, expires_at, tolerance_seconds)) => {
let is_expired = expires_at > 0 && expires_at < now;
Ok(vec![KeyInfo {
key_id,
pk_size,
created_at: None,
fetched_at: Some(fetched_at),
expires_at,
tolerance_seconds: Some(tolerance_seconds),
is_expired,
}])
}
None => Ok(vec![]),
}
}
pub async fn node_info_direct(&self) -> GrpcResult<GetNodeInfoResponse> {
platform::recording::debug!("GetNodeInfo request received");
let uptime_secs = self.started_at.elapsed().as_secs() as i64;
let metrics = self.collect_metrics().await?;
let services = self.service_statuses().await;
Ok(GetNodeInfoResponse {
success: true,
error_message: None,
node_id: self.node_id.clone(),
name: self.name.clone(),
version: self.version.clone(),
location_tag: self.location_tag.clone(),
uptime_secs,
current_metrics: Some(metrics),
services,
})
}
pub async fn list_realms_direct(&self) -> GrpcResult<ListRealmsResponse> {
platform::recording::debug!("ListRealms request received");
let realms = Realm::get_all()
.await
.map_err(|e| Status::internal(format!("Failed to load realm list: {e}")))?;
let realms_info: Vec<RealmInfo> = realms.iter().map(realm_to_proto).collect();
let total_count = realms_info.len() as u32;
Ok(ListRealmsResponse {
success: true,
error_message: None,
realms: realms_info,
next_page_token: None,
total_count,
})
}
async fn create_local_realm_internal(
&self,
req: CreateRealmRequest,
expose_plain_secret: bool,
) -> GrpcResult<(CreateRealmResponse, Option<String>)> {
platform::recording::info!("Create local realm request received: name={}", req.name);
let plain_secret = platform::realm::secret::generate_realm_secret();
let secret_hash = hash_realm_secret(&plain_secret);
let mut realm = match Realm::create(req.name.clone(), secret_hash).await {
Ok(r) => r,
Err(err) => {
return Ok((
CreateRealmResponse {
success: false,
error_message: Some(format!("Failed to create realm: {err}")),
realm: None,
},
None,
));
}
};
realm.enabled = req.enabled;
if req.expires_at > 0 {
realm.expires_at = Some(req.expires_at);
}
if let Some(status) = req.status.as_deref() {
match parse_realm_status(status) {
Ok(status) => realm.status = status,
Err(err) => {
return Ok((
CreateRealmResponse {
success: false,
error_message: Some(err),
realm: None,
},
None,
));
}
}
}
if let Err(err) = realm.save().await {
let _ = Realm::delete(realm.id).await;
return Ok((
CreateRealmResponse {
success: false,
error_message: Some(format!("Failed to save realm settings: {err}")),
realm: None,
},
None,
));
}
platform::recording::info!("Realm created: id={}, name={}", realm.id, realm.name);
let realm_info = realm_to_proto(&realm);
Ok((
CreateRealmResponse {
success: true,
error_message: None,
realm: Some(realm_info),
},
if expose_plain_secret {
Some(plain_secret)
} else {
None
},
))
}
async fn create_managed_realm_internal(
&self,
req: CreateRealmRequest,
) -> GrpcResult<CreateRealmResponse> {
let Some(realm_id) = req.realm_id else {
return Ok(CreateRealmResponse {
success: false,
error_message: Some("realm_id is required for managed CreateRealm".to_string()),
realm: None,
});
};
let Some(secret_current) = req.secret_current_hash else {
return Ok(CreateRealmResponse {
success: false,
error_message: Some(
"secret_current_hash is required for managed CreateRealm".to_string(),
),
realm: None,
});
};
let status = match req.status.as_deref() {
Some(status) => match parse_realm_status(status) {
Ok(status) => status,
Err(err) => {
return Ok(CreateRealmResponse {
success: false,
error_message: Some(err),
realm: None,
});
}
},
None => RealmStatus::Active,
};
let secret_previous =
build_secret_previous(req.secret_previous_hash, req.secret_previous_valid_until);
platform::recording::info!(
"Create managed realm request received: id={}, name={}",
realm_id,
req.name
);
match Realm::upsert_managed(
realm_id,
req.name,
status,
req.enabled,
(req.expires_at > 0).then_some(req.expires_at),
secret_current,
secret_previous,
)
.await
{
Ok(realm) => Ok(CreateRealmResponse {
success: true,
error_message: None,
realm: Some(realm_to_proto(&realm)),
}),
Err(err) => Ok(CreateRealmResponse {
success: false,
error_message: Some(format!("Failed to upsert managed realm: {err}")),
realm: None,
}),
}
}
pub async fn create_realm_with_secret_direct(
&self,
req: CreateRealmRequest,
) -> GrpcResult<(CreateRealmResponse, Option<String>)> {
self.create_local_realm_internal(req, true).await
}
pub async fn create_realm_direct(
&self,
req: CreateRealmRequest,
) -> GrpcResult<CreateRealmResponse> {
self.create_managed_realm_internal(req).await
}
pub async fn get_realm_direct(&self, realm_id: u32) -> GrpcResult<GetRealmResponse> {
platform::recording::debug!("GetRealm request received: realm_id={}", realm_id);
match self.load_realm(realm_id).await {
Ok(realm) => Ok(GetRealmResponse {
success: true,
error_message: None,
realm: Some(realm_to_proto(&realm)),
}),
Err(status) if status.code() == tonic::Code::NotFound => Ok(GetRealmResponse {
success: false,
error_message: Some(status.message().to_string()),
realm: None,
}),
Err(e) => Err(e),
}
}
pub async fn update_realm_direct(
&self,
req: UpdateRealmRequest,
) -> GrpcResult<UpdateRealmResponse> {
let mut realm = match self.load_realm(req.realm_id).await {
Ok(r) => r,
Err(status) if status.code() == tonic::Code::NotFound => {
return Ok(UpdateRealmResponse {
success: false,
error_message: Some(status.message().to_string()),
realm: None,
});
}
Err(e) => return Err(e),
};
if let Some(name) = req.name {
realm.name = name;
}
if let Some(enabled) = req.enabled {
realm.enabled = enabled;
}
if let Some(status) = req.status.as_deref() {
match parse_realm_status(status) {
Ok(status) => realm.status = status,
Err(err) => {
return Ok(UpdateRealmResponse {
success: false,
error_message: Some(err),
realm: None,
});
}
}
}
if let Some(expires_at) = req.expires_at {
realm.expires_at = (expires_at > 0).then_some(expires_at);
}
if let Some(secret_current_hash) = req.secret_current_hash {
if secret_current_hash.trim().is_empty() {
return Ok(UpdateRealmResponse {
success: false,
error_message: Some("secret_current_hash must not be empty".to_string()),
realm: None,
});
}
realm.secret_current = secret_current_hash;
}
if req.secret_previous_hash.is_some() || req.secret_previous_valid_until.is_some() {
realm.secret_previous =
build_secret_previous(req.secret_previous_hash, req.secret_previous_valid_until);
}
if let Err(err) = realm.save().await {
return Ok(UpdateRealmResponse {
success: false,
error_message: Some(format!("Failed to update realm: {err}")),
realm: None,
});
}
Ok(UpdateRealmResponse {
success: true,
error_message: None,
realm: Some(realm_to_proto(&realm)),
})
}
pub async fn delete_realm_direct(&self, realm_id: u32) -> GrpcResult<DeleteRealmResponse> {
match Realm::soft_delete(realm_id).await {
Ok(true) => Ok(DeleteRealmResponse {
success: true,
error_message: None,
}),
Ok(false) => Ok(DeleteRealmResponse {
success: false,
error_message: Some("Realm not found".to_string()),
}),
Err(err) => Ok(DeleteRealmResponse {
success: false,
error_message: Some(format!("Failed to delete realm: {err}")),
}),
}
}
pub async fn delete_realm_hard_direct(&self, realm_id: u32) -> GrpcResult<DeleteRealmResponse> {
match Realm::delete(realm_id).await {
Ok(affected) if affected > 0 => Ok(DeleteRealmResponse {
success: true,
error_message: None,
}),
Ok(_) => Ok(DeleteRealmResponse {
success: false,
error_message: Some("Realm not found".to_string()),
}),
Err(err) => Ok(DeleteRealmResponse {
success: false,
error_message: Some(format!("Failed to delete realm: {err}")),
}),
}
}
pub async fn rotate_realm_secret_direct(
&self,
realm_id: u32,
) -> GrpcResult<RealmSecretRotationResult> {
let rotated = rotate_realm_secret(realm_id, Some(DEFAULT_REALM_SECRET_PREVIOUS_GRACE_SECS))
.await
.map_err(|e| Status::internal(format!("Failed to rotate realm secret: {e}")))?;
Ok(RealmSecretRotationResult {
realm_id,
realm_secret: rotated.new_secret,
previous_valid_until: rotated.previous_valid_until,
grace_seconds: DEFAULT_REALM_SECRET_PREVIOUS_GRACE_SECS,
})
}
pub async fn get_config_direct(
&self,
config_type: ConfigType,
config_key: String,
) -> GrpcResult<GetConfigResponse> {
let key = Self::build_config_key(config_type, config_key);
let store = self.config_store.read().await;
if let Some(value) = store.get(&key) {
Ok(GetConfigResponse {
success: true,
error_message: None,
config_value: Some(value.clone()),
})
} else {
Ok(GetConfigResponse {
success: false,
error_message: Some("Config not found".to_string()),
config_value: None,
})
}
}
pub async fn update_config_direct(
&self,
config_type: ConfigType,
config_key: String,
config_value: String,
) -> GrpcResult<UpdateConfigResponse> {
let key = Self::build_config_key(config_type, config_key);
let mut store = self.config_store.write().await;
let old_value = store.insert(key, config_value);
Ok(UpdateConfigResponse {
success: true,
error_message: None,
old_value,
})
}
pub async fn shutdown_direct(
&self,
graceful: bool,
timeout_secs: Option<i32>,
reason: Option<String>,
) -> GrpcResult<ShutdownResponse> {
if let Some(handler) = &self.shutdown_handler {
if let Err(e) = handler(graceful, timeout_secs, reason.clone()).await {
return Ok(ShutdownResponse {
accepted: false,
error_message: Some(format!("Shutdown handler failed: {e}")),
estimated_shutdown_time: None,
});
}
} else {
platform::recording::warn!("Shutdown requested but no handler registered");
}
let estimated = if graceful {
timeout_secs.map(|v| Utc::now().timestamp() + v as i64)
} else {
Some(Utc::now().timestamp())
};
Ok(ShutdownResponse {
accepted: true,
error_message: None,
estimated_shutdown_time: estimated,
})
}
}
#[derive(Debug, Clone, Serialize)]
pub struct HealthInfo {
pub status: String,
pub node: String,
pub version: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct PlatformDetail {
pub config: JsonValue,
pub config_fields: Vec<ResolvedField>,
}
#[derive(Debug, Clone, Serialize)]
pub struct ServiceDetail {
pub enabled: bool,
pub status: Option<JsonValue>,
pub config: Option<JsonValue>,
pub config_fields: Option<Vec<ResolvedField>>,
}
#[derive(Debug, Clone, Serialize)]
pub struct ConfigFileContent {
pub content: String,
pub path: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct KeyInfo {
pub key_id: i64,
pub pk_size: i32,
#[serde(skip_serializing_if = "Option::is_none")]
pub created_at: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub fetched_at: Option<i64>,
pub expires_at: i64,
#[serde(skip_serializing_if = "Option::is_none")]
pub tolerance_seconds: Option<i64>,
pub is_expired: bool,
}
#[derive(Debug, Clone, Serialize)]
pub struct RealmSecretRotationResult {
pub realm_id: u32,
pub realm_secret: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub previous_valid_until: Option<u64>,
pub grace_seconds: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct KsKeysResult {
pub keys: Vec<KeyInfo>,
pub total_count: i64,
}
#[derive(Debug, Clone, Serialize)]
pub struct KsCleanupResult {
pub deleted: u64,
pub remaining: i64,
pub tolerance_seconds: u64,
}
fn service_name_to_type_id(name: &str) -> Option<i32> {
match name {
"stun" => Some(1),
"turn" => Some(2),
"signaling" => Some(3),
"ais" => Some(4),
"signer" => Some(5),
_ => None,
}
}
fn parse_realm_status(value: &str) -> std::result::Result<RealmStatus, String> {
RealmStatus::from_str(value).map_err(|_| {
format!("Invalid realm status '{value}' (expected Active, Inactive, Suspended)")
})
}
fn build_secret_previous(hash: Option<String>, valid_until: Option<u64>) -> Option<(String, u64)> {
match (hash, valid_until) {
(Some(hash), Some(valid_until)) if !hash.trim().is_empty() && valid_until > 0 => {
Some((hash, valid_until))
}
_ => None,
}
}
#[tonic::async_trait]
impl NodeAdminService for AdminApiService {
async fn update_config(
&self,
request: Request<UpdateConfigRequest>,
) -> GrpcResult<Response<UpdateConfigResponse>> {
let req = request.into_inner();
self.update_config_direct(req.config_type(), req.config_key, req.config_value)
.await
.map(Response::new)
}
async fn get_config(
&self,
request: Request<GetConfigRequest>,
) -> GrpcResult<Response<GetConfigResponse>> {
let req = request.into_inner();
self.get_config_direct(req.config_type(), req.config_key)
.await
.map(Response::new)
}
async fn create_realm(
&self,
request: Request<CreateRealmRequest>,
) -> GrpcResult<Response<CreateRealmResponse>> {
self.create_realm_direct(request.into_inner())
.await
.map(Response::new)
}
async fn get_realm(
&self,
request: Request<GetRealmRequest>,
) -> GrpcResult<Response<GetRealmResponse>> {
let req = request.into_inner();
self.get_realm_direct(req.realm_id).await.map(Response::new)
}
async fn update_realm(
&self,
request: Request<UpdateRealmRequest>,
) -> GrpcResult<Response<UpdateRealmResponse>> {
self.update_realm_direct(request.into_inner())
.await
.map(Response::new)
}
async fn delete_realm(
&self,
request: Request<DeleteRealmRequest>,
) -> GrpcResult<Response<DeleteRealmResponse>> {
let req = request.into_inner();
self.delete_realm_direct(req.realm_id)
.await
.map(Response::new)
}
async fn list_realms(
&self,
request: Request<ListRealmsRequest>,
) -> GrpcResult<Response<ListRealmsResponse>> {
let _req = request.into_inner();
self.list_realms_direct().await.map(Response::new)
}
async fn get_node_info(
&self,
request: Request<GetNodeInfoRequest>,
) -> GrpcResult<Response<GetNodeInfoResponse>> {
let _req = request.into_inner();
self.node_info_direct().await.map(Response::new)
}
async fn shutdown(
&self,
request: Request<ShutdownRequest>,
) -> GrpcResult<Response<ShutdownResponse>> {
let req = request.into_inner();
self.shutdown_direct(req.graceful, req.timeout_secs, req.reason)
.await
.map(Response::new)
}
async fn list_config_overrides(
&self,
_request: Request<ListConfigOverridesRequest>,
) -> GrpcResult<Response<ListConfigOverridesResponse>> {
match self.list_overrides_direct().await {
Ok(overrides) => {
let entries = overrides
.into_iter()
.map(|o| ProtoConfigOverrideEntry {
key_path: o.key_path,
value: o.value,
updated_at: o.updated_at,
updated_by: o.updated_by,
})
.collect();
Ok(Response::new(ListConfigOverridesResponse {
success: true,
error_message: None,
overrides: entries,
}))
}
Err(e) => Ok(Response::new(ListConfigOverridesResponse {
success: false,
error_message: Some(e.message().to_string()),
overrides: vec![],
})),
}
}
async fn set_config_override(
&self,
request: Request<SetConfigOverrideRequest>,
) -> GrpcResult<Response<SetConfigOverrideResponse>> {
let req = request.into_inner();
let by = req.updated_by.as_deref().unwrap_or("admin");
match self.set_override_direct(&req.key, &req.value, by).await {
Ok(()) => Ok(Response::new(SetConfigOverrideResponse {
success: true,
error_message: None,
})),
Err(e) => Ok(Response::new(SetConfigOverrideResponse {
success: false,
error_message: Some(e.message().to_string()),
})),
}
}
async fn delete_config_override(
&self,
request: Request<DeleteConfigOverrideRequest>,
) -> GrpcResult<Response<DeleteConfigOverrideResponse>> {
let req = request.into_inner();
match self.delete_override_direct(&req.key).await {
Ok(deleted) => Ok(Response::new(DeleteConfigOverrideResponse {
success: true,
error_message: None,
deleted,
})),
Err(e) => Ok(Response::new(DeleteConfigOverrideResponse {
success: false,
error_message: Some(e.message().to_string()),
deleted: false,
})),
}
}
}