1pub mod application;
2pub mod domain;
3pub mod event;
4pub mod infrastructure;
5pub mod logging;
6pub mod plugins;
7#[cfg(test)]
8mod test_env;
9
10use std::sync::{
11 Arc,
12 Mutex,
13 PoisonError,
14};
15
16use application::services::device_alerts;
17pub use application::{
18 LogExport,
19 LogService,
20};
21use domain::PairedDevice;
22pub use domain::{
23 DomainError,
24 DomainResult,
25 FetchStatus,
26 LogCacheStats,
27 LogDocumentMeta,
28 LogLinesPage,
29 LogSearchQuery,
30 LogSearchResult,
31 Pipeline,
32 PipelineRun,
33 PipelineStatus,
34 Provider,
35 ProviderConfig,
36 ProviderSummary,
37 RunLogDocument,
38 RunLogTree,
39 SignInStart,
40 SignInStatus,
41 TriggerParams,
42};
43pub use event::{
44 CoreEvent,
45 EventBus,
46 NoOpEventBus,
47};
48use infrastructure::database::{
49 init_database,
50 DeviceAlertRepository,
51 DevicePushRepository,
52 DeviceRepository,
53 MetricsRepository,
54 Repository,
55};
56pub use infrastructure::{
57 ConflictResolution,
58 EnvTokenStore,
59 LocalStorage,
60 MemoryTokenStore,
61 StorageBackend,
62 SyncConfig,
63 SyncDirection,
64 SyncManager,
65 SyncResult,
66 TokenStore,
67};
68use tokio_util::sync::CancellationToken;
69
70type AlertEvents = Arc<Mutex<Option<tokio::sync::mpsc::Receiver<CoreEvent>>>>;
71
72#[derive(Clone)]
73pub struct CoreContext {
74 pub event_bus: Arc<dyn EventBus>,
75
76 pub token_store: Arc<dyn TokenStore>,
77
78 pub http_client_manager: Arc<infrastructure::HttpClientManager>,
79
80 pub provider_service: Arc<application::ProviderService>,
81
82 pub pipeline_service: Arc<application::PipelineService>,
83
84 pub metrics_service: Option<Arc<application::MetricsService>>,
85
86 pub refresh_manager: Arc<application::RefreshManager>,
87
88 pub health: application::HealthHandle,
89
90 pub health_service: Arc<application::ProviderHealthService>,
91
92 pub config_state: Arc<infrastructure::ConfigState>,
93
94 pub log_service: Arc<application::LogService>,
95
96 pub pipeline_view_service: Arc<application::PipelineViewService>,
97
98 config_watcher: Arc<application::ConfigWatcher>,
99
100 alert_events: AlertEvents,
101
102 alert_stop: CancellationToken,
103}
104
105impl CoreContext {
106 fn from_services(
107 event_bus: Arc<dyn EventBus>, token_store: Arc<dyn TokenStore>,
108 http_client_manager: Arc<infrastructure::HttpClientManager>,
109 metrics_service: Option<Arc<application::MetricsService>>,
110 config_state: Arc<infrastructure::ConfigState>, services: application::Services,
111 alert_events: AlertEvents,
112 ) -> Self {
113 Self {
114 alert_events,
115 alert_stop: CancellationToken::new(),
116 event_bus,
117 token_store,
118 http_client_manager,
119 config_watcher: Arc::new(application::ConfigWatcher::new(
120 Arc::clone(&services.provider_service),
121 Arc::clone(&config_state),
122 )),
123 provider_service: services.provider_service,
124 pipeline_service: services.pipeline_service,
125 metrics_service,
126 refresh_manager: services.refresh_manager,
127 health: services.health,
128 health_service: services.health_service,
129 log_service: services.log_service,
130 pipeline_view_service: services.pipeline_view_service,
131 config_state,
132 }
133 }
134
135 pub async fn new(
136 data_dir: &std::path::Path, event_bus: Arc<dyn EventBus>, token_store: Arc<dyn TokenStore>,
137 ) -> anyhow::Result<Self> {
138 std::fs::create_dir_all(data_dir)?;
139
140 let config_state =
141 infrastructure::ConfigState::initialize(data_dir, token_store.clone()).await?;
142
143 let config = config_state.get().await;
144 let db_path = config.db_path();
145 let metrics_enabled = config.general.metrics_enabled;
146
147 if let Some(parent) = db_path.parent() {
148 std::fs::create_dir_all(parent)?;
149 }
150
151 let pool = init_database(db_path).await?;
152
153 let config_backend = Arc::new(infrastructure::SqliteConfigBackend::new(pool.clone()))
154 as Arc<dyn infrastructure::ConfigBackend>;
155 let cache_pool = pool.clone();
156
157 let sync_result = config_state
158 .sync_providers_from_toml(config_backend.as_ref(), false)
159 .await?;
160 if !sync_result.added.is_empty()
161 || !sync_result.updated.is_empty()
162 || !sync_result.removed.is_empty()
163 {
164 tracing::info!(
165 added = sync_result.added.len(),
166 updated = sync_result.updated.len(),
167 removed = sync_result.removed.len(),
168 "Synced providers from TOML to database on startup"
169 );
170 }
171
172 let metrics_service = metrics_enabled.then(|| {
173 let metrics_repository = Arc::new(MetricsRepository::new(pool.clone()));
174 Arc::new(application::MetricsService::new(metrics_repository))
175 });
176
177 let repository = Arc::new(Repository::new(
178 config_backend,
179 infrastructure::database::DatabasePool::Sqlite(cache_pool),
180 token_store.clone(),
181 ));
182
183 let http_client_manager = http_client_manager(&config.general.proxy_url)?;
184
185 let log_store = infrastructure::run_logs::LogStore::open(
186 config.cache_dir().join("run-logs"),
187 infrastructure::run_logs::CACHE_BUDGET_BYTES,
188 )
189 .await?;
190 let (event_bus, alert_events) = application::recording_bus(event_bus);
191 let services = application::Services::new(
192 repository,
193 Arc::clone(&http_client_manager),
194 metrics_service.clone(),
195 Arc::clone(&event_bus),
196 Arc::clone(&config_state),
197 log_store,
198 );
199
200 Ok(Self::from_services(
201 event_bus,
202 token_store,
203 http_client_manager,
204 metrics_service,
205 config_state,
206 services,
207 Arc::new(Mutex::new(Some(alert_events))),
208 ))
209 }
210
211 pub async fn with_storage_manager(
212 storage_manager: &infrastructure::StorageManager, event_bus: Arc<dyn EventBus>,
213 ) -> anyhow::Result<Self> {
214 let token_store = storage_manager.token_store().await;
215 let config_backend = storage_manager.config_backend();
216 let data_dir = storage_manager.config().data_dir();
217 let _storage_config = storage_manager.config();
218
219 let config_state =
220 infrastructure::ConfigState::initialize(&data_dir, token_store.clone()).await?;
221
222 let sync_result = config_state
223 .sync_providers_from_toml(config_backend.as_ref(), false)
224 .await?;
225 if !sync_result.added.is_empty()
226 || !sync_result.updated.is_empty()
227 || !sync_result.removed.is_empty()
228 {
229 tracing::info!(
230 added = sync_result.added.len(),
231 updated = sync_result.updated.len(),
232 removed = sync_result.removed.len(),
233 "Synced providers from TOML to database on startup"
234 );
235 }
236
237 let cache_pool = storage_manager.database_pool();
238
239 let repository = Arc::new(infrastructure::database::Repository::new(
240 config_backend.clone(),
241 cache_pool.clone(),
242 token_store.clone(),
243 ));
244
245 let config = config_state.get().await;
246 let http_client_manager = http_client_manager(&config.general.proxy_url)?;
247 let metrics_service = config.general.metrics_enabled.then(|| {
248 let metrics_repository = Arc::new(
249 infrastructure::database::MetricsRepository::new_from_pool(cache_pool),
250 );
251 Arc::new(application::MetricsService::new(metrics_repository))
252 });
253
254 let log_store = infrastructure::run_logs::LogStore::open(
255 config.cache_dir().join("run-logs"),
256 infrastructure::run_logs::CACHE_BUDGET_BYTES,
257 )
258 .await?;
259 let (event_bus, alert_events) = application::recording_bus(event_bus);
260 let services = application::Services::new(
261 repository,
262 Arc::clone(&http_client_manager),
263 metrics_service.clone(),
264 Arc::clone(&event_bus),
265 Arc::clone(&config_state),
266 log_store,
267 );
268
269 Ok(Self::from_services(
270 event_bus,
271 token_store,
272 http_client_manager,
273 metrics_service,
274 config_state,
275 services,
276 Arc::new(Mutex::new(Some(alert_events))),
277 ))
278 }
279
280 pub async fn start_background_tasks(&self) {
281 let provider_service = Arc::clone(&self.provider_service);
282 let refresh_manager = Arc::clone(&self.refresh_manager);
283 let health_service = Arc::clone(&self.health_service);
284 let config_watcher = Arc::clone(&self.config_watcher);
285 let health = self.health.clone();
286 let alert_events = self
287 .alert_events
288 .lock()
289 .unwrap_or_else(PoisonError::into_inner)
290 .take();
291 if let Some(alert_events) = alert_events {
292 let state = device_alerts::seed(&self.pipeline_service, &self.provider_service).await;
293 tokio::spawn(device_alerts::record(
294 alert_events,
295 state,
296 self.alert_stop.clone(),
297 self.device_alerts(),
298 self.devices(),
299 Arc::clone(&self.event_bus),
300 ));
301 }
302
303 tokio::spawn(async move {
304 if let Err(e) = provider_service.load_all_providers().await {
305 tracing::warn!("Failed to load providers during startup: {}", e);
306 }
307
308 health_service.start().await;
309 health.send(application::HealthInput::ProvidersChanged {
310 changed: provider_service.health_inputs().await,
311 });
312 refresh_manager.start().await;
313 config_watcher.start().await;
314 });
315 }
316
317 pub async fn apply_config_file(&self) -> DomainResult<infrastructure::ConfigSyncResult> {
318 self.provider_service
319 .apply_config_file(&self.config_state)
320 .await
321 }
322
323 pub async fn factory_reset(&self) -> DomainResult<usize> {
324 let providers = self.provider_service.list_providers().await?;
325 let providers_removed = providers.len();
326 for provider in providers {
327 self.provider_service.remove_provider(provider.id).await?;
328 }
329
330 for key in self.config_state.load_providers()?.keys() {
331 self.config_state.remove_provider(key).await?;
332 }
333 for provider_id in self.token_store.get_all_tokens().await?.into_keys() {
334 self.token_store.delete_token(provider_id).await?;
335 }
336 self.pipeline_service.clear_all_caches_atomic().await?;
337 self.pipeline_view_service.reset().await?;
338 self.log_service.clear().await?;
339 let metrics = MetricsRepository::new_from_pool(
340 self.provider_service.repository().cache_pool().clone(),
341 );
342 metrics.flush_metrics(None, true).await?;
343 metrics.reset_all_processing_states().await?;
344 self.device_alerts().delete_all().await?;
345 self.device_pushes().delete_all().await?;
346 Ok(providers_removed)
347 }
348
349 pub fn devices(&self) -> DeviceRepository {
350 DeviceRepository::new(self.provider_service.repository().cache_pool().clone())
351 }
352
353 pub fn device_alerts(&self) -> DeviceAlertRepository {
354 DeviceAlertRepository::new(self.provider_service.repository().cache_pool().clone())
355 }
356
357 pub fn device_pushes(&self) -> DevicePushRepository {
358 DevicePushRepository::new(self.provider_service.repository().cache_pool().clone())
359 }
360
361 pub async fn pair_device(
362 &self, name: &str, platform: &str,
363 ) -> DomainResult<(PairedDevice, String)> {
364 let (device, token) = PairedDevice::issue(name, platform);
365 self.devices().insert(&device).await?;
366 self.event_bus
367 .emit(CoreEvent::DevicePaired {
368 id: device.id.clone(),
369 name: device.name.clone(),
370 })
371 .await;
372
373 Ok((device, token))
374 }
375
376 pub async fn revoke_device(&self, id: &str) -> DomainResult<bool> {
377 self.device_pushes().delete(id).await?;
378 let removed = self.devices().delete(id).await?;
379 if removed {
380 self.event_bus
381 .emit(CoreEvent::DeviceRevoked { id: id.to_string() })
382 .await;
383 }
384
385 Ok(removed)
386 }
387
388 pub async fn shutdown(&self) {
389 self.alert_stop.cancel();
390 self.log_service.shutdown().await;
391 self.config_watcher.stop().await;
392 self.refresh_manager.stop().await;
393 self.health_service.stop().await;
394 }
395
396 pub async fn warmup_token_store(&self) -> anyhow::Result<()> {
397 tracing::info!("Warming up token store (may take 30-60 seconds on first startup)...");
398
399 self.token_store
400 .warmup()
401 .await
402 .map_err(|e| anyhow::anyhow!("Failed to warm up token store: {}", e))?;
403
404 tracing::info!("Token store warmed up successfully - providers can now access tokens");
405 Ok(())
406 }
407}
408
409fn http_client_manager(proxy_url: &str) -> DomainResult<Arc<infrastructure::HttpClientManager>> {
410 let manager = infrastructure::HttpClientManager::new()?;
411 if let Err(e) = manager.set_proxy(proxy_url) {
412 tracing::error!(error = %e, "Ignoring general.proxy_url from config.toml");
413 }
414
415 Ok(Arc::new(manager))
416}
417
418#[cfg(test)]
419mod reset_tests {
420 use std::collections::HashMap;
421
422 use super::*;
423 use crate::application::test_support::FailOnceTokenStore;
424 use crate::infrastructure::{
425 ConfigLoader,
426 ConfigState,
427 PipedashConfig,
428 ProviderFileConfig,
429 SqliteConfigBackend,
430 };
431
432 async fn context(dir: &std::path::Path, tokens: Arc<dyn TokenStore>) -> CoreContext {
433 let mut config = PipedashConfig::default();
434 config.storage.data_dir = dir.display().to_string();
435 config.general.metrics_enabled = false;
436 config.general.default_refresh_interval = 73;
437 let path = dir.join("config.toml");
438 std::fs::write(&path, ConfigLoader::to_toml(&config).unwrap()).unwrap();
439 let config_state = ConfigState::from_config(config, path, Arc::clone(&tokens));
440 let pool = init_database(dir.join("pipedash.db")).await.unwrap();
441 let repository = Arc::new(Repository::new(
442 Arc::new(SqliteConfigBackend::new(pool.clone())),
443 infrastructure::database::DatabasePool::Sqlite(pool),
444 Arc::clone(&tokens),
445 ));
446 let bus: Arc<dyn EventBus> = Arc::new(NoOpEventBus);
447 let http = http_client_manager("").unwrap();
448 let log_store = infrastructure::run_logs::LogStore::open(dir.join("run-logs"), u64::MAX)
449 .await
450 .unwrap();
451 let services = application::Services::new(
452 repository,
453 Arc::clone(&http),
454 None,
455 Arc::clone(&bus),
456 Arc::clone(&config_state),
457 log_store,
458 );
459 CoreContext::from_services(
460 bus,
461 tokens,
462 http,
463 None,
464 config_state,
465 services,
466 Arc::default(),
467 )
468 }
469
470 async fn define_provider(core: &CoreContext, key: &str) {
471 core.config_state
472 .add_provider(
473 key.into(),
474 ProviderFileConfig {
475 name: None,
476 provider_type: "jenkins".into(),
477 token: "storage:1".into(),
478 refresh_interval: 30,
479 config: HashMap::new(),
480 },
481 )
482 .await
483 .unwrap();
484 }
485
486 fn provider(key: &str) -> ProviderConfig {
487 ProviderConfig {
488 id: None,
489 name: key.into(),
490 provider_type: "jenkins".into(),
491 token: "saved-token".into(),
492 config: HashMap::new(),
493 refresh_interval: 30,
494 version: None,
495 }
496 }
497
498 #[tokio::test]
499 async fn reset_removes_config_only_providers_and_persisted_metrics_when_disabled() {
500 let dir = tempfile::tempdir().unwrap();
501 let tokens = Arc::new(FailOnceTokenStore::default());
502 let core = context(dir.path(), tokens.clone()).await;
503 define_provider(&core, "ci-a").await;
504 define_provider(&core, "config-only").await;
505 core.provider_service
506 .repository()
507 .add_provider(&provider("ci-a"))
508 .await
509 .unwrap();
510 tokens.store_token(999, "orphan-token").await.unwrap();
511 let pool = core
512 .provider_service
513 .repository()
514 .cache_pool()
515 .as_sqlite()
516 .unwrap();
517 sqlx::query("INSERT INTO pipeline_metrics (pipeline_id, timestamp, metric_type, value) VALUES ('retained', '2026-10-26T00:00:00Z', 'duration', 1)").execute(pool).await.unwrap();
518 sqlx::query("INSERT INTO metrics_processing_state (pipeline_id) VALUES ('retained')")
519 .execute(pool)
520 .await
521 .unwrap();
522 std::fs::write(dir.path().join("export.toml"), "user export").unwrap();
523
524 assert_eq!(core.factory_reset().await.unwrap(), 1);
525
526 assert!(core
527 .provider_service
528 .list_providers()
529 .await
530 .unwrap()
531 .is_empty());
532 assert!(tokens.get_all_tokens().await.unwrap().is_empty());
533 core.config_state.reload().await.unwrap();
534 assert!(core.config_state.get().await.providers.is_empty());
535 assert_eq!(
536 core.config_state
537 .get()
538 .await
539 .general
540 .default_refresh_interval,
541 73
542 );
543 core.config_state
544 .sync_providers_from_toml(
545 core.provider_service.repository().config_backend().as_ref(),
546 false,
547 )
548 .await
549 .unwrap();
550 assert!(core
551 .provider_service
552 .list_providers()
553 .await
554 .unwrap()
555 .is_empty());
556 let metrics: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pipeline_metrics")
557 .fetch_one(pool)
558 .await
559 .unwrap();
560 let processing: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM metrics_processing_state")
561 .fetch_one(pool)
562 .await
563 .unwrap();
564 assert_eq!((metrics, processing), (0, 0));
565 assert_eq!(
566 std::fs::read_to_string(dir.path().join("export.toml")).unwrap(),
567 "user export"
568 );
569 assert_eq!(core.factory_reset().await.unwrap(), 0);
570 }
571
572 #[tokio::test]
573 async fn reset_stops_on_token_deletion_failure_and_can_be_retried() {
574 let dir = tempfile::tempdir().unwrap();
575 let tokens = Arc::new(FailOnceTokenStore::default());
576 let core = context(dir.path(), tokens.clone()).await;
577 define_provider(&core, "ci-a").await;
578 let id = core
579 .provider_service
580 .repository()
581 .add_provider(&provider("ci-a"))
582 .await
583 .unwrap();
584 tokens.fail_next_delete();
585
586 assert!(core.factory_reset().await.is_err());
587 assert!(core
588 .provider_service
589 .repository()
590 .get_provider(id)
591 .await
592 .is_ok());
593 assert!(core
594 .config_state
595 .load_providers()
596 .unwrap()
597 .contains_key("ci-a"));
598
599 assert_eq!(core.factory_reset().await.unwrap(), 1);
600 assert!(tokens.get_all_tokens().await.unwrap().is_empty());
601 }
602
603 #[tokio::test]
604 async fn reset_reports_cache_and_metrics_failures_and_supports_retry() {
605 for table in ["workflow_parameters_cache", "pipeline_metrics"] {
606 let dir = tempfile::tempdir().unwrap();
607 let core = context(dir.path(), Arc::new(MemoryTokenStore::new())).await;
608 let pool = core
609 .provider_service
610 .repository()
611 .cache_pool()
612 .as_sqlite()
613 .unwrap();
614 sqlx::query(sqlx::AssertSqlSafe(format!(
615 "ALTER TABLE {table} RENAME TO retained_data"
616 )))
617 .execute(pool)
618 .await
619 .unwrap();
620
621 assert!(core.factory_reset().await.is_err(), "{table}");
622
623 sqlx::query(sqlx::AssertSqlSafe(format!(
624 "ALTER TABLE retained_data RENAME TO {table}"
625 )))
626 .execute(pool)
627 .await
628 .unwrap();
629 assert_eq!(core.factory_reset().await.unwrap(), 0);
630 }
631 }
632
633 #[tokio::test]
634 async fn revoking_a_device_removes_its_push_registration() {
635 let dir = tempfile::tempdir().unwrap();
636 let core = context(dir.path(), Arc::new(MemoryTokenStore::new())).await;
637 let (device, _) = core.pair_device("Pixel", "android").await.unwrap();
638 core.device_pushes().set(&device.id, "sub-1").await.unwrap();
639
640 assert!(core.revoke_device(&device.id).await.unwrap());
641
642 assert_eq!(core.device_pushes().get(&device.id).await.unwrap(), None);
643 }
644
645 #[tokio::test]
646 async fn factory_reset_clears_alerts_and_push_registrations() {
647 let dir = tempfile::tempdir().unwrap();
648 let core = context(dir.path(), Arc::new(MemoryTokenStore::new())).await;
649 core.device_alerts()
650 .insert(&domain::NewDeviceAlert {
651 key: "pipeline:p".into(),
652 kind: domain::AlertKind::Failed,
653 title: "t".into(),
654 body: "b".into(),
655 pipeline_id: None,
656 provider_id: None,
657 })
658 .await
659 .unwrap();
660 core.device_pushes().set("d1", "sub-1").await.unwrap();
661
662 core.factory_reset().await.unwrap();
663
664 assert!(core.device_alerts().list().await.unwrap().is_empty());
665 assert!(core.device_pushes().list().await.unwrap().is_empty());
666 }
667
668 #[tokio::test]
669 async fn shutdown_stops_the_alert_recorder() {
670 let dir = tempfile::tempdir().unwrap();
671 let core = context(dir.path(), Arc::new(MemoryTokenStore::new())).await;
672 assert!(!core.alert_stop.is_cancelled());
673
674 core.shutdown().await;
675
676 assert!(core.alert_stop.is_cancelled());
677 }
678}