use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use async_trait::async_trait;
use chrono::{DateTime, Duration, Utc};
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use tracing::info;
use crate::error::KernelError;
use crate::health::HealthStatus;
use crate::process::Pid;
use crate::service::{ServiceType, SystemService};
#[non_exhaustive]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum ConfigValue {
Text(String),
Integer(i64),
Float(f64),
Boolean(bool),
Json(serde_json::Value),
}
impl ConfigValue {
pub fn to_json(&self) -> serde_json::Value {
match self {
ConfigValue::Text(s) => serde_json::Value::String(s.clone()),
ConfigValue::Integer(n) => serde_json::json!(n),
ConfigValue::Float(f) => serde_json::json!(f),
ConfigValue::Boolean(b) => serde_json::json!(b),
ConfigValue::Json(v) => v.clone(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConfigEntry {
pub key: String,
pub namespace: String,
pub value: ConfigValue,
pub updated_at: DateTime<Utc>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConfigChange {
pub namespace: String,
pub key: String,
pub old_value: Option<serde_json::Value>,
pub new_value: Option<serde_json::Value>,
pub changed_by: Pid,
pub timestamp: DateTime<Utc>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SecretRef {
pub namespace: String,
pub key: String,
pub expires_at: DateTime<Utc>,
pub scoped_to: Vec<Pid>,
}
pub struct ConfigService {
configs: DashMap<String, serde_json::Value>,
entries: DashMap<String, ConfigEntry>,
secrets: DashMap<String, Vec<u8>>,
secret_refs: DashMap<String, SecretRef>,
subscribers: DashMap<String, Vec<Arc<RwLock<Vec<ConfigChange>>>>>,
encryption_key: [u8; 32],
change_log: RwLock<Vec<ConfigChange>>,
set_count: AtomicU64,
}
impl ConfigService {
pub fn new(encryption_key: [u8; 32]) -> Self {
Self {
configs: DashMap::new(),
entries: DashMap::new(),
secrets: DashMap::new(),
secret_refs: DashMap::new(),
subscribers: DashMap::new(),
encryption_key,
change_log: RwLock::new(Vec::new()),
set_count: AtomicU64::new(0),
}
}
pub fn new_default() -> Self {
Self::new([0u8; 32])
}
pub fn set(
&self,
namespace: &str,
key: &str,
value: serde_json::Value,
changed_by: Pid,
) -> Result<(), KernelError> {
let config_key = format!("{namespace}/{key}");
let old_value = self.configs.get(&config_key).map(|v| v.value().clone());
self.configs.insert(config_key, value.clone());
let change = ConfigChange {
namespace: namespace.to_string(),
key: key.to_string(),
old_value,
new_value: Some(value),
changed_by,
timestamp: Utc::now(),
};
self.notify_subscribers(namespace, &change);
if let Ok(mut log) = self.change_log.write() {
log.push(change);
}
self.set_count.fetch_add(1, Ordering::Relaxed);
Ok(())
}
pub fn get(&self, namespace: &str, key: &str) -> Option<serde_json::Value> {
let config_key = format!("{namespace}/{key}");
self.configs.get(&config_key).map(|v| v.value().clone())
}
pub fn delete(
&self,
namespace: &str,
key: &str,
changed_by: Pid,
) -> Result<(), KernelError> {
let config_key = format!("{namespace}/{key}");
let old_value = self.configs.remove(&config_key).map(|(_, v)| v);
let change = ConfigChange {
namespace: namespace.to_string(),
key: key.to_string(),
old_value,
new_value: None,
changed_by,
timestamp: Utc::now(),
};
self.notify_subscribers(namespace, &change);
Ok(())
}
pub fn list_keys(&self, namespace: &str) -> Vec<String> {
let prefix = format!("{namespace}/");
self.configs
.iter()
.filter(|e| e.key().starts_with(&prefix))
.map(|e| e.key()[prefix.len()..].to_string())
.collect()
}
pub fn set_typed(
&self,
namespace: &str,
key: &str,
value: ConfigValue,
changed_by: Pid,
) -> Result<(), KernelError> {
let config_key = format!("{namespace}/{key}");
let json_value = value.to_json();
let old_value = self.configs.get(&config_key).map(|v| v.value().clone());
self.configs.insert(config_key.clone(), json_value.clone());
let entry = ConfigEntry {
key: key.to_string(),
namespace: namespace.to_string(),
value,
updated_at: Utc::now(),
};
self.entries.insert(config_key, entry);
let change = ConfigChange {
namespace: namespace.to_string(),
key: key.to_string(),
old_value,
new_value: Some(json_value),
changed_by,
timestamp: Utc::now(),
};
self.notify_subscribers(namespace, &change);
if let Ok(mut log) = self.change_log.write() {
log.push(change);
}
self.set_count.fetch_add(1, Ordering::Relaxed);
Ok(())
}
pub fn get_typed(&self, namespace: &str, key: &str) -> Option<ConfigEntry> {
let config_key = format!("{namespace}/{key}");
self.entries.get(&config_key).map(|e| e.value().clone())
}
pub fn list(&self, namespace: &str) -> Vec<ConfigEntry> {
let prefix = format!("{namespace}/");
self.entries
.iter()
.filter(|e| e.key().starts_with(&prefix))
.map(|e| e.value().clone())
.collect()
}
pub fn delete_typed(
&self,
namespace: &str,
key: &str,
changed_by: Pid,
) -> bool {
let config_key = format!("{namespace}/{key}");
let removed = self.entries.remove(&config_key).is_some();
let old_value = self.configs.remove(&config_key).map(|(_, v)| v);
let change = ConfigChange {
namespace: namespace.to_string(),
key: key.to_string(),
old_value,
new_value: None,
changed_by,
timestamp: Utc::now(),
};
self.notify_subscribers(namespace, &change);
removed
}
pub fn subscribe(&self, namespace: &str) -> Arc<RwLock<Vec<ConfigChange>>> {
let queue = Arc::new(RwLock::new(Vec::new()));
self.subscribers
.entry(namespace.to_string())
.or_default()
.push(queue.clone());
queue
}
fn notify_subscribers(&self, namespace: &str, change: &ConfigChange) {
if let Some(mut subs) = self.subscribers.get_mut(namespace) {
subs.retain(|queue| {
if let Ok(mut q) = queue.write() {
q.push(change.clone());
true
} else {
false }
});
}
}
pub fn set_secret(
&self,
namespace: &str,
key: &str,
value: &[u8],
scoped_to: Vec<Pid>,
) -> Result<(), KernelError> {
let secret_key = format!("{namespace}/{key}");
let encrypted = self.xor_encrypt(value);
self.secrets.insert(secret_key.clone(), encrypted);
let secret_ref = SecretRef {
namespace: namespace.to_string(),
key: key.to_string(),
expires_at: Utc::now() + Duration::hours(24),
scoped_to,
};
self.secret_refs.insert(secret_key, secret_ref);
Ok(())
}
pub fn get_secret(
&self,
namespace: &str,
key: &str,
requester_pid: Pid,
) -> Result<Vec<u8>, KernelError> {
let secret_key = format!("{namespace}/{key}");
let secret_ref = self
.secret_refs
.get(&secret_key)
.ok_or_else(|| KernelError::Service("secret not found".into()))?;
if !secret_ref.scoped_to.is_empty()
&& !secret_ref.scoped_to.contains(&requester_pid)
{
return Err(KernelError::CapabilityDenied {
pid: requester_pid,
action: "read_secret".into(),
reason: format!("PID {} not authorized for secret {secret_key}", requester_pid),
});
}
if Utc::now() > secret_ref.expires_at {
return Err(KernelError::Service("secret expired".into()));
}
let encrypted = self
.secrets
.get(&secret_key)
.ok_or_else(|| KernelError::Service("secret data missing".into()))?;
Ok(self.xor_decrypt(&encrypted))
}
fn xor_encrypt(&self, data: &[u8]) -> Vec<u8> {
data.iter()
.enumerate()
.map(|(i, b)| b ^ self.encryption_key[i % 32])
.collect()
}
fn xor_decrypt(&self, data: &[u8]) -> Vec<u8> {
self.xor_encrypt(data)
}
pub fn change_log(&self) -> Vec<ConfigChange> {
self.change_log.read().map(|l| l.clone()).unwrap_or_default()
}
pub fn set_count(&self) -> u64 {
self.set_count.load(Ordering::Relaxed)
}
}
#[async_trait]
impl SystemService for ConfigService {
fn name(&self) -> &str {
"config-service"
}
fn service_type(&self) -> ServiceType {
ServiceType::Core
}
async fn start(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
info!("config service started");
Ok(())
}
async fn stop(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
info!(
configs = self.configs.len(),
secrets = self.secrets.len(),
"config service stopped"
);
Ok(())
}
async fn health_check(&self) -> HealthStatus {
HealthStatus::Healthy
}
}
#[cfg(test)]
mod tests {
use super::*;
fn pid(n: u64) -> Pid {
n
}
#[test]
fn set_and_get_config() {
let svc = ConfigService::new_default();
svc.set("app", "timeout", serde_json::json!(30), pid(1)).unwrap();
let val = svc.get("app", "timeout").unwrap();
assert_eq!(val, serde_json::json!(30));
}
#[test]
fn get_nonexistent_returns_none() {
let svc = ConfigService::new_default();
assert!(svc.get("app", "missing").is_none());
}
#[test]
fn delete_config() {
let svc = ConfigService::new_default();
svc.set("app", "key", serde_json::json!("val"), pid(1)).unwrap();
svc.delete("app", "key", pid(1)).unwrap();
assert!(svc.get("app", "key").is_none());
}
#[test]
fn list_keys_in_namespace() {
let svc = ConfigService::new_default();
svc.set("ns", "a", serde_json::json!(1), pid(1)).unwrap();
svc.set("ns", "b", serde_json::json!(2), pid(1)).unwrap();
svc.set("other", "c", serde_json::json!(3), pid(1)).unwrap();
let mut keys = svc.list_keys("ns");
keys.sort();
assert_eq!(keys, vec!["a", "b"]);
}
#[test]
fn config_change_notification() {
let svc = ConfigService::new_default();
let sub = svc.subscribe("watch");
svc.set("watch", "flag", serde_json::json!(true), pid(1)).unwrap();
let changes = sub.read().unwrap();
assert_eq!(changes.len(), 1);
assert_eq!(changes[0].key, "flag");
assert_eq!(changes[0].new_value, Some(serde_json::json!(true)));
}
#[test]
fn config_change_includes_old_value() {
let svc = ConfigService::new_default();
let sub = svc.subscribe("ver");
svc.set("ver", "v", serde_json::json!(1), pid(1)).unwrap();
svc.set("ver", "v", serde_json::json!(2), pid(1)).unwrap();
let changes = sub.read().unwrap();
assert_eq!(changes.len(), 2);
assert_eq!(changes[1].old_value, Some(serde_json::json!(1)));
assert_eq!(changes[1].new_value, Some(serde_json::json!(2)));
}
#[test]
fn secret_set_and_get() {
let key = [0xAB; 32];
let svc = ConfigService::new(key);
svc.set_secret("creds", "api_key", b"secret123", vec![pid(1)])
.unwrap();
let val = svc.get_secret("creds", "api_key", pid(1)).unwrap();
assert_eq!(val, b"secret123");
}
#[test]
fn secret_encrypted_at_rest() {
let key = [0xAB; 32];
let svc = ConfigService::new(key);
svc.set_secret("creds", "pass", b"plaintext", vec![pid(1)])
.unwrap();
let stored = svc.secrets.get("creds/pass").unwrap();
assert_ne!(stored.as_slice(), b"plaintext");
}
#[test]
fn unauthorized_pid_cannot_read_secret() {
let svc = ConfigService::new_default();
svc.set_secret("creds", "key", b"val", vec![pid(1)]).unwrap();
let result = svc.get_secret("creds", "key", pid(99));
assert!(result.is_err());
let err = result.unwrap_err().to_string();
assert!(err.contains("denied") || err.contains("authorized"), "got: {err}");
}
#[test]
fn empty_scope_allows_any_pid() {
let svc = ConfigService::new_default();
svc.set_secret("open", "key", b"val", vec![]).unwrap();
let val = svc.get_secret("open", "key", pid(42)).unwrap();
assert_eq!(val, b"val");
}
#[test]
fn change_log_recorded() {
let svc = ConfigService::new_default();
svc.set("ns", "k", serde_json::json!("v"), pid(1)).unwrap();
let log = svc.change_log();
assert_eq!(log.len(), 1);
assert_eq!(log[0].namespace, "ns");
}
#[tokio::test]
async fn system_service_impl() {
let svc = ConfigService::new_default();
assert_eq!(svc.name(), "config-service");
assert_eq!(svc.service_type(), ServiceType::Core);
svc.start().await.unwrap();
assert_eq!(svc.health_check().await, HealthStatus::Healthy);
svc.stop().await.unwrap();
}
#[test]
fn typed_set_get_roundtrip() {
let svc = ConfigService::new_default();
svc.set_typed("app", "name", ConfigValue::Text("myapp".into()), pid(1))
.unwrap();
let entry = svc.get_typed("app", "name").unwrap();
assert_eq!(entry.value, ConfigValue::Text("myapp".into()));
assert_eq!(entry.namespace, "app");
assert_eq!(entry.key, "name");
}
#[test]
fn typed_get_nonexistent_returns_none() {
let svc = ConfigService::new_default();
assert!(svc.get_typed("app", "missing").is_none());
}
#[test]
fn typed_list_returns_all_in_namespace() {
let svc = ConfigService::new_default();
svc.set_typed("db", "host", ConfigValue::Text("localhost".into()), pid(1))
.unwrap();
svc.set_typed("db", "port", ConfigValue::Integer(5432), pid(1))
.unwrap();
svc.set_typed("cache", "ttl", ConfigValue::Integer(60), pid(1))
.unwrap();
let mut entries = svc.list("db");
entries.sort_by(|a, b| a.key.cmp(&b.key));
assert_eq!(entries.len(), 2);
assert_eq!(entries[0].key, "host");
assert_eq!(entries[1].key, "port");
}
#[test]
fn typed_delete_removes_entry() {
let svc = ConfigService::new_default();
svc.set_typed("ns", "k", ConfigValue::Boolean(true), pid(1)).unwrap();
assert!(svc.delete_typed("ns", "k", pid(1)));
assert!(svc.get_typed("ns", "k").is_none());
assert!(!svc.delete_typed("ns", "k", pid(1)));
}
#[test]
fn typed_subscribe_receives_change_notification() {
let svc = ConfigService::new_default();
let sub = svc.subscribe("typed-ns");
svc.set_typed("typed-ns", "flag", ConfigValue::Boolean(true), pid(1))
.unwrap();
let changes = sub.read().unwrap();
assert_eq!(changes.len(), 1);
assert_eq!(changes[0].key, "flag");
assert_eq!(changes[0].new_value, Some(serde_json::json!(true)));
}
#[test]
fn typed_set_updates_existing_value() {
let svc = ConfigService::new_default();
svc.set_typed("app", "level", ConfigValue::Integer(1), pid(1)).unwrap();
svc.set_typed("app", "level", ConfigValue::Integer(2), pid(1)).unwrap();
let entry = svc.get_typed("app", "level").unwrap();
assert_eq!(entry.value, ConfigValue::Integer(2));
}
#[test]
fn typed_namespace_isolation() {
let svc = ConfigService::new_default();
svc.set_typed("alpha", "key", ConfigValue::Text("a".into()), pid(1)).unwrap();
svc.set_typed("beta", "key", ConfigValue::Text("b".into()), pid(1)).unwrap();
let a = svc.get_typed("alpha", "key").unwrap();
let b = svc.get_typed("beta", "key").unwrap();
assert_eq!(a.value, ConfigValue::Text("a".into()));
assert_eq!(b.value, ConfigValue::Text("b".into()));
assert_eq!(svc.list("alpha").len(), 1);
assert_eq!(svc.list("beta").len(), 1);
assert_eq!(svc.list("gamma").len(), 0);
}
#[test]
fn typed_all_value_variants() {
let svc = ConfigService::new_default();
svc.set_typed("t", "text", ConfigValue::Text("hello".into()), pid(1)).unwrap();
svc.set_typed("t", "int", ConfigValue::Integer(42), pid(1)).unwrap();
svc.set_typed("t", "float", ConfigValue::Float(3.14), pid(1)).unwrap();
svc.set_typed("t", "bool", ConfigValue::Boolean(false), pid(1)).unwrap();
svc.set_typed("t", "json", ConfigValue::Json(serde_json::json!({"a": 1})), pid(1)).unwrap();
assert_eq!(svc.get_typed("t", "text").unwrap().value, ConfigValue::Text("hello".into()));
assert_eq!(svc.get_typed("t", "int").unwrap().value, ConfigValue::Integer(42));
assert_eq!(svc.get_typed("t", "float").unwrap().value, ConfigValue::Float(3.14));
assert_eq!(svc.get_typed("t", "bool").unwrap().value, ConfigValue::Boolean(false));
assert_eq!(
svc.get_typed("t", "json").unwrap().value,
ConfigValue::Json(serde_json::json!({"a": 1}))
);
}
}