1use std::path::PathBuf;
2use std::sync::Arc;
3
4use tokio::sync::{
5 broadcast,
6 RwLock,
7};
8
9use super::loader::{
10 ConfigLoader,
11 Platform,
12};
13use super::schema::{
14 ConfigKey,
15 PipedashConfig,
16 ProviderFileConfig,
17};
18use super::sync::{
19 ProviderSyncService,
20 SyncResult,
21};
22use super::token_ref::TokenReference;
23use super::validation::ValidationResult;
24use crate::domain::{
25 DomainError,
26 DomainResult,
27};
28use crate::infrastructure::{
29 ConfigBackend,
30 TokenStore,
31};
32
33#[derive(Debug, Clone)]
34pub enum ConfigChangeEvent {
35 Reloaded {
36 changed_keys: Vec<String>,
37 },
38 ValueChanged {
39 key: String,
40 old_value: Option<String>,
41 new_value: String,
42 },
43 ProviderAdded {
44 name: String,
45 },
46 ProviderUpdated {
47 name: String,
48 },
49 ProviderRemoved {
50 name: String,
51 },
52}
53
54impl ConfigChangeEvent {
55 pub fn summary(&self) -> String {
56 match self {
57 Self::Reloaded { changed_keys } => {
58 format!("Config reloaded, {} keys changed", changed_keys.len())
59 }
60 Self::ValueChanged { key, .. } => format!("Value changed: {}", key),
61 Self::ProviderAdded { name } => format!("Provider added: {}", name),
62 Self::ProviderUpdated { name } => format!("Provider updated: {}", name),
63 Self::ProviderRemoved { name } => format!("Provider removed: {}", name),
64 }
65 }
66}
67
68pub struct ConfigState {
69 config: RwLock<Arc<PipedashConfig>>,
70
71 config_path: PathBuf,
72
73 change_tx: broadcast::Sender<ConfigChangeEvent>,
74
75 token_store: Arc<dyn TokenStore>,
76
77 platform: Platform,
78
79 sync_service: ProviderSyncService,
80}
81
82impl ConfigState {
83 pub async fn initialize(
84 data_dir: &std::path::Path, token_store: Arc<dyn TokenStore>,
85 ) -> DomainResult<Arc<Self>> {
86 let platform = Platform::detect();
87
88 let config_path = ConfigLoader::discover_config_path();
89 let legacy_path = data_dir.join("storage_config.json");
90
91 if legacy_path.exists() && !config_path.exists() {
92 let migrated_config =
93 super::migration::ConfigMigrator::migrate_if_needed(data_dir).await?;
94 return Self::from_config_impl(migrated_config, config_path, token_store);
95 }
96
97 let config = ConfigLoader::load_or_create(&config_path, platform)
98 .map_err(|e| DomainError::InvalidConfig(format!("Failed to load config: {}", e)))?;
99
100 let validation = config.validate();
101 for error in &validation.errors {
102 tracing::error!("Config error: {}", error);
103 }
104 for warning in &validation.warnings {
105 tracing::warn!("Config warning: {}", warning);
106 }
107
108 let (change_tx, _) = broadcast::channel(16);
109
110 let state = Arc::new(Self {
111 config: RwLock::new(Arc::new(config)),
112 config_path,
113 change_tx,
114 token_store,
115 platform,
116 sync_service: ProviderSyncService::new(),
117 });
118
119 Ok(state)
120 }
121
122 pub fn from_config(
123 config: PipedashConfig, config_path: PathBuf, token_store: Arc<dyn TokenStore>,
124 ) -> Arc<Self> {
125 let (change_tx, _) = broadcast::channel(16);
126
127 Arc::new(Self {
128 config: RwLock::new(Arc::new(config)),
129 config_path,
130 change_tx,
131 token_store,
132 platform: Platform::detect(),
133 sync_service: ProviderSyncService::new(),
134 })
135 }
136
137 fn from_config_impl(
138 config: PipedashConfig, config_path: PathBuf, token_store: Arc<dyn TokenStore>,
139 ) -> DomainResult<Arc<Self>> {
140 Ok(Self::from_config(config, config_path, token_store))
141 }
142
143 pub async fn get(&self) -> Arc<PipedashConfig> {
144 self.config.read().await.clone()
145 }
146
147 pub fn config_path(&self) -> &PathBuf {
148 &self.config_path
149 }
150
151 pub fn token_store(&self) -> Arc<dyn TokenStore> {
152 Arc::clone(&self.token_store)
153 }
154
155 pub fn platform(&self) -> Platform {
156 self.platform
157 }
158
159 pub fn subscribe(&self) -> broadcast::Receiver<ConfigChangeEvent> {
160 self.change_tx.subscribe()
161 }
162
163 pub async fn resolve_token(
164 &self, token_ref: &TokenReference, provider_id: Option<i64>,
165 ) -> DomainResult<String> {
166 token_ref
167 .resolve(self.token_store.as_ref(), provider_id)
168 .await
169 .map_err(|e| DomainError::InvalidConfig(format!("Token resolution failed: {}", e)))
170 }
171
172 pub async fn reload(&self) -> DomainResult<ConfigChangeEvent> {
173 let new_config = ConfigLoader::load(&self.config_path)
174 .map_err(|e| DomainError::InvalidConfig(format!("Failed to reload config: {}", e)))?;
175
176 let validation = new_config.validate();
177 if !validation.errors.is_empty() {
178 return Err(DomainError::InvalidConfig(format!(
179 "Config validation failed: {:?}",
180 validation.errors
181 )));
182 }
183
184 let event = ConfigChangeEvent::Reloaded {
185 changed_keys: vec![], };
187
188 let _ = self.change_tx.send(event.clone());
189
190 tracing::info!("Config reloaded from {:?}", self.config_path);
191 Ok(event)
192 }
193
194 pub async fn validate(&self) -> ValidationResult {
195 self.config.read().await.validate()
196 }
197
198 pub async fn add_provider(&self, id: String, provider: ProviderFileConfig) -> DomainResult<()> {
199 if !provider.token.is_empty() {
200 TokenReference::parse(&provider.token)
201 .map_err(|e| DomainError::InvalidConfig(format!("Invalid token format: {}", e)))?;
202 }
203
204 let current_config = self.config.read().await;
205 let mut config = (**current_config).clone();
206 drop(current_config);
207
208 if config.providers.contains_key(&id) {
209 return Err(DomainError::InvalidConfig(format!(
210 "Provider '{}' already exists",
211 id
212 )));
213 }
214
215 config.providers.insert(id.clone(), provider);
216
217 self.persist_config(&config).await?;
218
219 let event = ConfigChangeEvent::ProviderAdded { name: id };
220 let _ = self.change_tx.send(event.clone());
221
222 Ok(())
223 }
224
225 pub async fn update_provider(
226 &self, id: &str, provider: ProviderFileConfig,
227 ) -> DomainResult<()> {
228 if !provider.token.is_empty() {
229 TokenReference::parse(&provider.token)
230 .map_err(|e| DomainError::InvalidConfig(format!("Invalid token format: {}", e)))?;
231 }
232
233 let current_config = self.config.read().await;
234 let mut config = (**current_config).clone();
235 drop(current_config);
236
237 if !config.providers.contains_key(id) {
238 return Err(DomainError::InvalidConfig(format!(
239 "Provider '{}' not found",
240 id
241 )));
242 }
243
244 config.providers.insert(id.to_string(), provider);
245
246 self.persist_config(&config).await?;
247
248 let event = ConfigChangeEvent::ProviderUpdated {
249 name: id.to_string(),
250 };
251 let _ = self.change_tx.send(event.clone());
252
253 Ok(())
254 }
255
256 pub async fn remove_provider(&self, id: &str) -> DomainResult<()> {
257 let current_config = self.config.read().await;
258 let mut config = (**current_config).clone();
259 drop(current_config);
260
261 if config.providers.shift_remove(id).is_none() {
262 return Err(DomainError::InvalidConfig(format!(
263 "Provider '{}' not found",
264 id
265 )));
266 }
267
268 self.persist_config(&config).await?;
269
270 let event = ConfigChangeEvent::ProviderRemoved {
271 name: id.to_string(),
272 };
273 let _ = self.change_tx.send(event.clone());
274
275 Ok(())
276 }
277
278 pub async fn update_general_settings(
279 &self, metrics_enabled: bool, default_refresh_interval: u32,
280 ) -> DomainResult<()> {
281 let current_config = self.config.read().await;
282 let mut config = (**current_config).clone();
283 drop(current_config);
284
285 config.general.metrics_enabled = metrics_enabled;
286 config.general.default_refresh_interval = default_refresh_interval;
287
288 self.persist_config(&config).await?;
289
290 let _ = self.change_tx.send(ConfigChangeEvent::ValueChanged {
291 key: "general".to_string(),
292 old_value: None,
293 new_value: "updated".to_string(),
294 });
295
296 Ok(())
297 }
298
299 pub async fn sync_providers_from_toml(
300 &self, config_backend: &dyn ConfigBackend, delete_orphans: bool,
301 ) -> DomainResult<SyncResult> {
302 let config = self.config.read().await;
303 self.sync_service
304 .sync_toml_to_db(
305 config.providers.clone(),
306 config_backend,
307 self.token_store.as_ref(),
308 delete_orphans,
309 )
310 .await
311 }
312
313 async fn persist_config(&self, config: &PipedashConfig) -> DomainResult<()> {
314 let existing_content = std::fs::read_to_string(&self.config_path).ok();
315
316 let new_content = if let Some(existing) = existing_content {
317 Self::update_toml_preserving_comments(&existing, config)?
318 } else {
319 ConfigLoader::to_toml(config).map_err(|e| {
320 DomainError::DatabaseError(format!("Failed to serialize config: {}", e))
321 })?
322 };
323
324 std::fs::write(&self.config_path, &new_content).map_err(|e| {
325 DomainError::DatabaseError(format!("Failed to write config file: {}", e))
326 })?;
327
328 tracing::debug!("Config persisted to {:?}", self.config_path);
329 Ok(())
330 }
331
332 fn update_toml_preserving_comments(
333 existing: &str, config: &PipedashConfig,
334 ) -> DomainResult<String> {
335 use toml_edit::{
336 value,
337 DocumentMut,
338 Item,
339 };
340
341 let mut doc: DocumentMut = existing.parse().map_err(|e| {
342 DomainError::DatabaseError(format!("Failed to parse existing TOML: {}", e))
343 })?;
344
345 if let Some(general) = doc.get_mut("general") {
346 if let Some(table) = general.as_table_mut() {
347 table["metrics_enabled"] = value(config.general.metrics_enabled);
348 table["default_refresh_interval"] =
349 value(config.general.default_refresh_interval as i64);
350 }
351 }
352
353 if let Some(server) = doc.get_mut("server") {
354 if let Some(table) = server.as_table_mut() {
355 table["bind_addr"] = value(&config.server.bind_addr);
356 table["cors_allow_all"] = value(config.server.cors_allow_all);
357 }
358 }
359
360 doc.remove("providers");
361
362 if !config.providers.is_empty() {
363 let mut providers_table = toml_edit::Table::new();
364
365 for (id, provider) in &config.providers {
366 let mut provider_table = toml_edit::Table::new();
367
368 if let Some(name) = &provider.name {
369 provider_table["name"] = value(name);
370 }
371
372 provider_table["type"] = value(&provider.provider_type);
373 provider_table["token"] = value(&provider.token);
374 provider_table["refresh_interval"] = value(provider.refresh_interval as i64);
375
376 if !provider.config.is_empty() {
377 let mut config_table = toml_edit::Table::new();
378 for (k, v) in &provider.config {
379 config_table[k] = value(v);
380 }
381 provider_table["config"] = Item::Table(config_table);
382 }
383
384 providers_table[id] = Item::Table(provider_table);
385 }
386
387 doc["providers"] = Item::Table(providers_table);
388 }
389
390 Ok(doc.to_string())
391 }
392
393 pub async fn get_string(&self, key: ConfigKey) -> String {
394 if let Ok(val) = std::env::var(key.env_var_name()) {
395 return val;
396 }
397
398 let config = self.config.read().await;
399 match key {
400 ConfigKey::MetricsEnabled => config.general.metrics_enabled.to_string(),
401 ConfigKey::DefaultRefreshInterval => {
402 config.general.default_refresh_interval.to_string()
403 }
404 ConfigKey::BindAddr => config.server.bind_addr.clone(),
405 ConfigKey::CorsAllowAll => config.server.cors_allow_all.to_string(),
406 ConfigKey::DataDir => {
407 if config.storage.data_dir.is_empty() {
408 PipedashConfig::default_data_dir()
409 .to_string_lossy()
410 .to_string()
411 } else {
412 config.storage.data_dir.clone()
413 }
414 }
415 ConfigKey::StorageBackend => config.storage.backend.to_string(),
416 ConfigKey::PostgresConnectionString => {
417 config.storage.postgres.connection_string.clone()
418 }
419 }
420 }
421
422 pub async fn get_bool(&self, key: ConfigKey) -> bool {
423 let val = self.get_string(key).await;
424 matches!(val.to_lowercase().as_str(), "true" | "1" | "yes")
425 }
426
427 pub async fn get_u32(&self, key: ConfigKey) -> u32 {
428 self.get_string(key).await.parse().unwrap_or(0)
429 }
430}
431
432#[cfg(test)]
433mod tests {
434 use tempfile::TempDir;
435
436 use super::*;
437 use crate::infrastructure::MemoryTokenStore;
438
439 fn create_test_state(temp_dir: &TempDir, token_store: Arc<dyn TokenStore>) -> Arc<ConfigState> {
440 let config_path = temp_dir.path().join("config.toml");
441 let config = PipedashConfig::default();
442 ConfigState::from_config(config, config_path, token_store)
443 }
444
445 #[tokio::test]
446 async fn test_config_state_from_config() {
447 let temp_dir = TempDir::new().unwrap();
448 let token_store = Arc::new(MemoryTokenStore::new());
449
450 let state = create_test_state(&temp_dir, token_store);
451
452 let config = state.get().await;
453 assert!(config.general.metrics_enabled);
454 }
455
456 #[tokio::test]
457 async fn test_add_remove_provider() {
458 let temp_dir = TempDir::new().unwrap();
459 let token_store = Arc::new(MemoryTokenStore::new());
460
461 let state = create_test_state(&temp_dir, token_store);
462
463 let provider = ProviderFileConfig {
464 name: Some("Test GitHub".to_string()),
465 provider_type: "github".to_string(),
466 token: "${GITHUB_TOKEN:-}".to_string(),
467 refresh_interval: 30,
468 config: std::collections::HashMap::new(),
469 };
470
471 state
472 .add_provider("test-github".to_string(), provider)
473 .await
474 .unwrap();
475
476 let config = state.get().await;
477 assert_eq!(config.providers.len(), 1);
478 assert!(config.providers.contains_key("test-github"));
479
480 state.remove_provider("test-github").await.unwrap();
481
482 let config = state.get().await;
483 assert!(config.providers.is_empty());
484 }
485
486 #[tokio::test]
487 async fn test_subscribe_to_changes() {
488 let temp_dir = TempDir::new().unwrap();
489 let token_store = Arc::new(MemoryTokenStore::new());
490
491 let state = create_test_state(&temp_dir, token_store);
492
493 let mut rx = state.subscribe();
494
495 let provider = ProviderFileConfig {
496 name: None,
497 provider_type: "github".to_string(),
498 token: String::new(),
499 refresh_interval: 30,
500 config: std::collections::HashMap::new(),
501 };
502
503 state
504 .add_provider("test".to_string(), provider)
505 .await
506 .unwrap();
507
508 let event = rx.try_recv().unwrap();
509 assert!(matches!(event, ConfigChangeEvent::ProviderAdded { .. }));
510 }
511
512 #[tokio::test]
513 async fn test_persist_and_reload() {
514 let temp_dir = TempDir::new().unwrap();
515 let token_store = Arc::new(MemoryTokenStore::new());
516 let config_path = temp_dir.path().join("config.toml");
517
518 let state = ConfigState::from_config(
519 PipedashConfig::default(),
520 config_path.clone(),
521 token_store.clone(),
522 );
523
524 let provider = ProviderFileConfig {
525 name: Some("Persist Test".to_string()),
526 provider_type: "gitlab".to_string(),
527 token: "${GITLAB_TOKEN:-}".to_string(),
528 refresh_interval: 60,
529 config: std::collections::HashMap::new(),
530 };
531
532 state
533 .add_provider("persist-test".to_string(), provider)
534 .await
535 .unwrap();
536
537 let loaded_config = ConfigLoader::load(&config_path).unwrap();
538 assert_eq!(loaded_config.providers.len(), 1);
539 assert!(loaded_config.providers.contains_key("persist-test"));
540 assert_eq!(loaded_config.providers["persist-test"].refresh_interval, 60);
541 }
542
543 #[tokio::test]
544 async fn test_get_config_values() {
545 let old_metrics = std::env::var("PIPEDASH_METRICS_ENABLED").ok();
546 let old_refresh = std::env::var("PIPEDASH_DEFAULT_REFRESH_INTERVAL").ok();
547 std::env::remove_var("PIPEDASH_METRICS_ENABLED");
548 std::env::remove_var("PIPEDASH_DEFAULT_REFRESH_INTERVAL");
549
550 let temp_dir = TempDir::new().unwrap();
551 let token_store = Arc::new(MemoryTokenStore::new());
552
553 let state = create_test_state(&temp_dir, token_store);
554
555 assert!(state.get_bool(ConfigKey::MetricsEnabled).await);
556 assert_eq!(state.get_u32(ConfigKey::DefaultRefreshInterval).await, 30);
557
558 if let Some(v) = old_metrics {
559 std::env::set_var("PIPEDASH_METRICS_ENABLED", v);
560 }
561 if let Some(v) = old_refresh {
562 std::env::set_var("PIPEDASH_DEFAULT_REFRESH_INTERVAL", v);
563 }
564 }
565
566 #[tokio::test]
567 async fn test_env_var_override() {
568 let old_value = std::env::var("PIPEDASH_METRICS_ENABLED").ok();
569 std::env::set_var("PIPEDASH_METRICS_ENABLED", "false");
570
571 let temp_dir = TempDir::new().unwrap();
572 let token_store = Arc::new(MemoryTokenStore::new());
573
574 let state = create_test_state(&temp_dir, token_store);
575
576 assert!(!state.get_bool(ConfigKey::MetricsEnabled).await);
577
578 match old_value {
579 Some(v) => std::env::set_var("PIPEDASH_METRICS_ENABLED", v),
580 None => std::env::remove_var("PIPEDASH_METRICS_ENABLED"),
581 }
582 }
583}