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 *self.config.write().await = Arc::new(new_config);
185
186 let event = ConfigChangeEvent::Reloaded {
187 changed_keys: vec![], };
189
190 let _ = self.change_tx.send(event.clone());
191
192 tracing::info!("Config reloaded from {:?}", self.config_path);
193 Ok(event)
194 }
195
196 pub async fn validate(&self) -> ValidationResult {
197 self.config.read().await.validate()
198 }
199
200 pub async fn add_provider(&self, id: String, provider: ProviderFileConfig) -> DomainResult<()> {
201 if !provider.token.is_empty() {
202 TokenReference::parse(&provider.token)
203 .map_err(|e| DomainError::InvalidConfig(format!("Invalid token format: {}", e)))?;
204 }
205
206 let current_config = self.config.read().await;
207 let mut config = (**current_config).clone();
208 drop(current_config);
209
210 if config.providers.contains_key(&id) {
211 return Err(DomainError::InvalidConfig(format!(
212 "Provider '{}' already exists",
213 id
214 )));
215 }
216
217 config.providers.insert(id.clone(), provider);
218
219 self.persist_config(&config).await?;
220
221 let event = ConfigChangeEvent::ProviderAdded { name: id };
222 let _ = self.change_tx.send(event.clone());
223
224 Ok(())
225 }
226
227 pub async fn update_provider(
228 &self, id: &str, provider: ProviderFileConfig,
229 ) -> DomainResult<()> {
230 if !provider.token.is_empty() {
231 TokenReference::parse(&provider.token)
232 .map_err(|e| DomainError::InvalidConfig(format!("Invalid token format: {}", e)))?;
233 }
234
235 let current_config = self.config.read().await;
236 let mut config = (**current_config).clone();
237 drop(current_config);
238
239 if !config.providers.contains_key(id) {
240 return Err(DomainError::InvalidConfig(format!(
241 "Provider '{}' not found",
242 id
243 )));
244 }
245
246 config.providers.insert(id.to_string(), provider);
247
248 self.persist_config(&config).await?;
249
250 let event = ConfigChangeEvent::ProviderUpdated {
251 name: id.to_string(),
252 };
253 let _ = self.change_tx.send(event.clone());
254
255 Ok(())
256 }
257
258 pub async fn remove_provider(&self, id: &str) -> DomainResult<()> {
259 let current_config = self.config.read().await;
260 let mut config = (**current_config).clone();
261 drop(current_config);
262
263 if config.providers.shift_remove(id).is_none() {
264 return Err(DomainError::InvalidConfig(format!(
265 "Provider '{}' not found",
266 id
267 )));
268 }
269
270 self.persist_config(&config).await?;
271
272 let event = ConfigChangeEvent::ProviderRemoved {
273 name: id.to_string(),
274 };
275 let _ = self.change_tx.send(event.clone());
276
277 Ok(())
278 }
279
280 pub async fn update_general_settings(
281 &self, metrics_enabled: bool, default_refresh_interval: u32,
282 ) -> DomainResult<()> {
283 let current_config = self.config.read().await;
284 let mut config = (**current_config).clone();
285 drop(current_config);
286
287 config.general.metrics_enabled = metrics_enabled;
288 config.general.default_refresh_interval = default_refresh_interval;
289
290 self.persist_config(&config).await?;
291
292 let _ = self.change_tx.send(ConfigChangeEvent::ValueChanged {
293 key: "general".to_string(),
294 old_value: None,
295 new_value: "updated".to_string(),
296 });
297
298 Ok(())
299 }
300
301 pub async fn sync_providers_from_toml(
302 &self, config_backend: &dyn ConfigBackend, delete_orphans: bool,
303 ) -> DomainResult<SyncResult> {
304 let config = self.config.read().await;
305 self.sync_service
306 .sync_toml_to_db(
307 config.providers.clone(),
308 config_backend,
309 self.token_store.as_ref(),
310 delete_orphans,
311 )
312 .await
313 }
314
315 async fn persist_config(&self, config: &PipedashConfig) -> DomainResult<()> {
316 let existing_content = std::fs::read_to_string(&self.config_path).ok();
317
318 let new_content = if let Some(existing) = existing_content {
319 Self::update_toml_preserving_comments(&existing, config)?
320 } else {
321 ConfigLoader::to_toml(config).map_err(|e| {
322 DomainError::DatabaseError(format!("Failed to serialize config: {}", e))
323 })?
324 };
325
326 std::fs::write(&self.config_path, &new_content).map_err(|e| {
327 DomainError::DatabaseError(format!("Failed to write config file: {}", e))
328 })?;
329
330 *self.config.write().await = Arc::new(config.clone());
331
332 tracing::debug!("Config persisted to {:?}", self.config_path);
333 Ok(())
334 }
335
336 fn update_toml_preserving_comments(
337 existing: &str, config: &PipedashConfig,
338 ) -> DomainResult<String> {
339 use toml_edit::{
340 value,
341 DocumentMut,
342 Item,
343 };
344
345 let mut doc: DocumentMut = existing.parse().map_err(|e| {
346 DomainError::DatabaseError(format!("Failed to parse existing TOML: {}", e))
347 })?;
348
349 if let Some(general) = doc.get_mut("general") {
350 if let Some(table) = general.as_table_mut() {
351 table["metrics_enabled"] = value(config.general.metrics_enabled);
352 table["default_refresh_interval"] =
353 value(config.general.default_refresh_interval as i64);
354 }
355 }
356
357 if let Some(server) = doc.get_mut("server") {
358 if let Some(table) = server.as_table_mut() {
359 table["bind_addr"] = value(&config.server.bind_addr);
360 table["cors_allow_all"] = value(config.server.cors_allow_all);
361 }
362 }
363
364 doc.remove("providers");
365
366 if !config.providers.is_empty() {
367 let mut providers_table = toml_edit::Table::new();
368
369 for (id, provider) in &config.providers {
370 let mut provider_table = toml_edit::Table::new();
371
372 if let Some(name) = &provider.name {
373 provider_table["name"] = value(name);
374 }
375
376 provider_table["type"] = value(&provider.provider_type);
377 provider_table["token"] = value(&provider.token);
378 provider_table["refresh_interval"] = value(provider.refresh_interval as i64);
379
380 if !provider.config.is_empty() {
381 let mut config_table = toml_edit::Table::new();
382 for (k, v) in &provider.config {
383 config_table[k] = value(v);
384 }
385 provider_table["config"] = Item::Table(config_table);
386 }
387
388 providers_table[id] = Item::Table(provider_table);
389 }
390
391 doc["providers"] = Item::Table(providers_table);
392 }
393
394 Ok(doc.to_string())
395 }
396
397 pub async fn get_string(&self, key: ConfigKey) -> String {
398 if let Ok(val) = std::env::var(key.env_var_name()) {
399 return val;
400 }
401
402 let config = self.config.read().await;
403 match key {
404 ConfigKey::MetricsEnabled => config.general.metrics_enabled.to_string(),
405 ConfigKey::DefaultRefreshInterval => {
406 config.general.default_refresh_interval.to_string()
407 }
408 ConfigKey::BindAddr => config.server.bind_addr.clone(),
409 ConfigKey::CorsAllowAll => config.server.cors_allow_all.to_string(),
410 ConfigKey::DataDir => {
411 if config.storage.data_dir.is_empty() {
412 PipedashConfig::default_data_dir()
413 .to_string_lossy()
414 .to_string()
415 } else {
416 config.storage.data_dir.clone()
417 }
418 }
419 ConfigKey::StorageBackend => config.storage.backend.to_string(),
420 ConfigKey::PostgresConnectionString => {
421 config.storage.postgres.connection_string.clone()
422 }
423 }
424 }
425
426 pub async fn get_bool(&self, key: ConfigKey) -> bool {
427 let val = self.get_string(key).await;
428 matches!(val.to_lowercase().as_str(), "true" | "1" | "yes")
429 }
430
431 pub async fn get_u32(&self, key: ConfigKey) -> u32 {
432 self.get_string(key).await.parse().unwrap_or(0)
433 }
434}
435
436#[cfg(test)]
437mod tests {
438 use tempfile::TempDir;
439
440 use super::*;
441 use crate::{
442 infrastructure::MemoryTokenStore,
443 test_env::EnvGuard,
444 };
445
446 fn create_test_state(temp_dir: &TempDir, token_store: Arc<dyn TokenStore>) -> Arc<ConfigState> {
447 let config_path = temp_dir.path().join("config.toml");
448 let config = PipedashConfig::default();
449 ConfigState::from_config(config, config_path, token_store)
450 }
451
452 #[tokio::test]
453 async fn test_config_state_from_config() {
454 let temp_dir = TempDir::new().unwrap();
455 let token_store = Arc::new(MemoryTokenStore::new());
456
457 let state = create_test_state(&temp_dir, token_store);
458
459 let config = state.get().await;
460 assert!(config.general.metrics_enabled);
461 }
462
463 #[tokio::test]
464 async fn test_add_remove_provider() {
465 let temp_dir = TempDir::new().unwrap();
466 let token_store = Arc::new(MemoryTokenStore::new());
467
468 let state = create_test_state(&temp_dir, token_store);
469
470 let provider = ProviderFileConfig {
471 name: Some("Test GitHub".to_string()),
472 provider_type: "github".to_string(),
473 token: "${GITHUB_TOKEN:-}".to_string(),
474 refresh_interval: 30,
475 config: std::collections::HashMap::new(),
476 };
477
478 state
479 .add_provider("test-github".to_string(), provider)
480 .await
481 .unwrap();
482
483 let config = state.get().await;
484 assert_eq!(config.providers.len(), 1);
485 assert!(config.providers.contains_key("test-github"));
486
487 state.remove_provider("test-github").await.unwrap();
488
489 let config = state.get().await;
490 assert!(config.providers.is_empty());
491 }
492
493 #[tokio::test]
494 async fn test_reload_replaces_in_memory_config() {
495 let temp_dir = TempDir::new().unwrap();
496 let token_store = Arc::new(MemoryTokenStore::new());
497
498 let state = create_test_state(&temp_dir, token_store);
499
500 let mut on_disk = PipedashConfig::default();
501 on_disk.general.default_refresh_interval = 123;
502 std::fs::write(
503 temp_dir.path().join("config.toml"),
504 ConfigLoader::to_toml(&on_disk).unwrap(),
505 )
506 .unwrap();
507
508 state.reload().await.unwrap();
509
510 assert_eq!(state.get().await.general.default_refresh_interval, 123);
511 }
512
513 #[tokio::test]
514 async fn test_subscribe_to_changes() {
515 let temp_dir = TempDir::new().unwrap();
516 let token_store = Arc::new(MemoryTokenStore::new());
517
518 let state = create_test_state(&temp_dir, token_store);
519
520 let mut rx = state.subscribe();
521
522 let provider = ProviderFileConfig {
523 name: None,
524 provider_type: "github".to_string(),
525 token: String::new(),
526 refresh_interval: 30,
527 config: std::collections::HashMap::new(),
528 };
529
530 state
531 .add_provider("test".to_string(), provider)
532 .await
533 .unwrap();
534
535 let event = rx.try_recv().unwrap();
536 assert!(matches!(event, ConfigChangeEvent::ProviderAdded { .. }));
537 }
538
539 #[tokio::test]
540 async fn test_persist_and_reload() {
541 let temp_dir = TempDir::new().unwrap();
542 let token_store = Arc::new(MemoryTokenStore::new());
543 let config_path = temp_dir.path().join("config.toml");
544
545 let state = ConfigState::from_config(
546 PipedashConfig::default(),
547 config_path.clone(),
548 token_store.clone(),
549 );
550
551 let provider = ProviderFileConfig {
552 name: Some("Persist Test".to_string()),
553 provider_type: "gitlab".to_string(),
554 token: "${GITLAB_TOKEN:-}".to_string(),
555 refresh_interval: 60,
556 config: std::collections::HashMap::new(),
557 };
558
559 state
560 .add_provider("persist-test".to_string(), provider)
561 .await
562 .unwrap();
563
564 let loaded_config = ConfigLoader::load(&config_path).unwrap();
565 assert_eq!(loaded_config.providers.len(), 1);
566 assert!(loaded_config.providers.contains_key("persist-test"));
567 assert_eq!(loaded_config.providers["persist-test"].refresh_interval, 60);
568 }
569
570 #[tokio::test]
571 async fn test_get_config_values() {
572 let _env = EnvGuard::set(&[
573 ("PIPEDASH_METRICS_ENABLED", None),
574 ("PIPEDASH_DEFAULT_REFRESH_INTERVAL", None),
575 ]);
576
577 let temp_dir = TempDir::new().unwrap();
578 let token_store = Arc::new(MemoryTokenStore::new());
579
580 let state = create_test_state(&temp_dir, token_store);
581
582 assert!(state.get_bool(ConfigKey::MetricsEnabled).await);
583 assert_eq!(state.get_u32(ConfigKey::DefaultRefreshInterval).await, 30);
584 }
585
586 #[tokio::test]
587 async fn test_env_var_override() {
588 let _env = EnvGuard::set(&[("PIPEDASH_METRICS_ENABLED", Some("false"))]);
589
590 let temp_dir = TempDir::new().unwrap();
591 let token_store = Arc::new(MemoryTokenStore::new());
592
593 let state = create_test_state(&temp_dir, token_store);
594
595 assert!(!state.get_bool(ConfigKey::MetricsEnabled).await);
596 }
597}