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