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::Arc;
11
12pub use application::{
13 LogExport,
14 LogService,
15};
16pub use domain::{
17 DomainError,
18 DomainResult,
19 FetchStatus,
20 LogCacheStats,
21 LogDocumentMeta,
22 LogLinesPage,
23 LogSearchQuery,
24 LogSearchResult,
25 Pipeline,
26 PipelineRun,
27 PipelineStatus,
28 Provider,
29 ProviderConfig,
30 ProviderSummary,
31 RunLogDocument,
32 RunLogTree,
33 SignInStart,
34 SignInStatus,
35 TriggerParams,
36};
37pub use event::{
38 CoreEvent,
39 EventBus,
40 NoOpEventBus,
41};
42use infrastructure::database::{
43 init_database,
44 MetricsRepository,
45 Repository,
46};
47pub use infrastructure::{
48 ConflictResolution,
49 EnvTokenStore,
50 LocalStorage,
51 MemoryTokenStore,
52 StorageBackend,
53 SyncConfig,
54 SyncDirection,
55 SyncManager,
56 SyncResult,
57 TokenStore,
58};
59
60pub struct CoreContext {
61 pub event_bus: Arc<dyn EventBus>,
62
63 pub token_store: Arc<dyn TokenStore>,
64
65 pub http_client_manager: Arc<infrastructure::HttpClientManager>,
66
67 pub provider_service: Arc<application::ProviderService>,
68
69 pub pipeline_service: Arc<application::PipelineService>,
70
71 pub metrics_service: Option<Arc<application::MetricsService>>,
72
73 pub refresh_manager: Arc<application::RefreshManager>,
74
75 pub health: application::HealthHandle,
76
77 pub health_service: Arc<application::ProviderHealthService>,
78
79 pub config_state: Arc<infrastructure::ConfigState>,
80
81 pub log_service: Arc<application::LogService>,
82
83 pub pipeline_view_service: Arc<application::PipelineViewService>,
84
85 config_watcher: Arc<application::ConfigWatcher>,
86}
87
88impl CoreContext {
89 fn from_services(
90 event_bus: Arc<dyn EventBus>, token_store: Arc<dyn TokenStore>,
91 http_client_manager: Arc<infrastructure::HttpClientManager>,
92 metrics_service: Option<Arc<application::MetricsService>>,
93 config_state: Arc<infrastructure::ConfigState>, services: application::Services,
94 ) -> Self {
95 Self {
96 event_bus,
97 token_store,
98 http_client_manager,
99 config_watcher: Arc::new(application::ConfigWatcher::new(
100 Arc::clone(&services.provider_service),
101 Arc::clone(&config_state),
102 )),
103 provider_service: services.provider_service,
104 pipeline_service: services.pipeline_service,
105 metrics_service,
106 refresh_manager: services.refresh_manager,
107 health: services.health,
108 health_service: services.health_service,
109 log_service: services.log_service,
110 pipeline_view_service: services.pipeline_view_service,
111 config_state,
112 }
113 }
114
115 pub async fn new(
116 data_dir: &std::path::Path, event_bus: Arc<dyn EventBus>, token_store: Arc<dyn TokenStore>,
117 ) -> anyhow::Result<Self> {
118 std::fs::create_dir_all(data_dir)?;
119
120 let config_state =
121 infrastructure::ConfigState::initialize(data_dir, token_store.clone()).await?;
122
123 let config = config_state.get().await;
124 let db_path = config.db_path();
125 let metrics_enabled = config.general.metrics_enabled;
126
127 if let Some(parent) = db_path.parent() {
128 std::fs::create_dir_all(parent)?;
129 }
130
131 let pool = init_database(db_path).await?;
132
133 let config_backend = Arc::new(infrastructure::SqliteConfigBackend::new(pool.clone()))
134 as Arc<dyn infrastructure::ConfigBackend>;
135 let cache_pool = pool.clone();
136
137 let sync_result = config_state
138 .sync_providers_from_toml(config_backend.as_ref(), false)
139 .await?;
140 if !sync_result.added.is_empty()
141 || !sync_result.updated.is_empty()
142 || !sync_result.removed.is_empty()
143 {
144 tracing::info!(
145 added = sync_result.added.len(),
146 updated = sync_result.updated.len(),
147 removed = sync_result.removed.len(),
148 "Synced providers from TOML to database on startup"
149 );
150 }
151
152 let metrics_service = metrics_enabled.then(|| {
153 let metrics_repository = Arc::new(MetricsRepository::new(pool.clone()));
154 Arc::new(application::MetricsService::new(metrics_repository))
155 });
156
157 let repository = Arc::new(Repository::new(
158 config_backend,
159 infrastructure::database::DatabasePool::Sqlite(cache_pool),
160 token_store.clone(),
161 ));
162
163 let http_client_manager = http_client_manager(&config.general.proxy_url)?;
164
165 let log_store = infrastructure::run_logs::LogStore::open(
166 config.cache_dir().join("run-logs"),
167 infrastructure::run_logs::CACHE_BUDGET_BYTES,
168 )
169 .await?;
170 let services = application::Services::new(
171 repository,
172 Arc::clone(&http_client_manager),
173 metrics_service.clone(),
174 Arc::clone(&event_bus),
175 Arc::clone(&config_state),
176 log_store,
177 );
178
179 Ok(Self::from_services(
180 event_bus,
181 token_store,
182 http_client_manager,
183 metrics_service,
184 config_state,
185 services,
186 ))
187 }
188
189 pub async fn with_storage_manager(
190 storage_manager: &infrastructure::StorageManager, event_bus: Arc<dyn EventBus>,
191 ) -> anyhow::Result<Self> {
192 let token_store = storage_manager.token_store().await;
193 let config_backend = storage_manager.config_backend();
194 let data_dir = storage_manager.config().data_dir();
195 let _storage_config = storage_manager.config();
196
197 let config_state =
198 infrastructure::ConfigState::initialize(&data_dir, token_store.clone()).await?;
199
200 let sync_result = config_state
201 .sync_providers_from_toml(config_backend.as_ref(), false)
202 .await?;
203 if !sync_result.added.is_empty()
204 || !sync_result.updated.is_empty()
205 || !sync_result.removed.is_empty()
206 {
207 tracing::info!(
208 added = sync_result.added.len(),
209 updated = sync_result.updated.len(),
210 removed = sync_result.removed.len(),
211 "Synced providers from TOML to database on startup"
212 );
213 }
214
215 let cache_pool = storage_manager.database_pool();
216
217 let repository = Arc::new(infrastructure::database::Repository::new(
218 config_backend.clone(),
219 cache_pool.clone(),
220 token_store.clone(),
221 ));
222
223 let config = config_state.get().await;
224 let http_client_manager = http_client_manager(&config.general.proxy_url)?;
225 let metrics_service = config.general.metrics_enabled.then(|| {
226 let metrics_repository = Arc::new(
227 infrastructure::database::MetricsRepository::new_from_pool(cache_pool),
228 );
229 Arc::new(application::MetricsService::new(metrics_repository))
230 });
231
232 let log_store = infrastructure::run_logs::LogStore::open(
233 config.cache_dir().join("run-logs"),
234 infrastructure::run_logs::CACHE_BUDGET_BYTES,
235 )
236 .await?;
237 let services = application::Services::new(
238 repository,
239 Arc::clone(&http_client_manager),
240 metrics_service.clone(),
241 Arc::clone(&event_bus),
242 Arc::clone(&config_state),
243 log_store,
244 );
245
246 Ok(Self::from_services(
247 event_bus,
248 token_store,
249 http_client_manager,
250 metrics_service,
251 config_state,
252 services,
253 ))
254 }
255
256 pub async fn start_background_tasks(&self) {
257 let provider_service = Arc::clone(&self.provider_service);
258 let refresh_manager = Arc::clone(&self.refresh_manager);
259 let health_service = Arc::clone(&self.health_service);
260 let config_watcher = Arc::clone(&self.config_watcher);
261 let health = self.health.clone();
262
263 tokio::spawn(async move {
264 if let Err(e) = provider_service.load_all_providers().await {
265 tracing::warn!("Failed to load providers during startup: {}", e);
266 }
267
268 health_service.start().await;
269 health.send(application::HealthInput::ProvidersChanged {
270 changed: provider_service.health_inputs().await,
271 });
272 refresh_manager.start().await;
273 config_watcher.start().await;
274 });
275 }
276
277 pub async fn apply_config_file(&self) -> DomainResult<infrastructure::ConfigSyncResult> {
278 self.provider_service
279 .apply_config_file(&self.config_state)
280 .await
281 }
282
283 pub async fn factory_reset(&self) -> DomainResult<usize> {
284 let providers = self.provider_service.list_providers().await?;
285 let providers_removed = providers.len();
286 for provider in providers {
287 self.provider_service.remove_provider(provider.id).await?;
288 }
289
290 for key in self.config_state.load_providers()?.keys() {
291 self.config_state.remove_provider(key).await?;
292 }
293 for provider_id in self.token_store.get_all_tokens().await?.into_keys() {
294 self.token_store.delete_token(provider_id).await?;
295 }
296 self.pipeline_service.clear_all_caches_atomic().await?;
297 self.pipeline_view_service.reset().await?;
298 self.log_service.clear().await?;
299 let metrics = MetricsRepository::new_from_pool(
300 self.provider_service.repository().cache_pool().clone(),
301 );
302 metrics.flush_metrics(None, true).await?;
303 metrics.reset_all_processing_states().await?;
304 Ok(providers_removed)
305 }
306
307 pub async fn shutdown(&self) {
308 self.log_service.shutdown().await;
309 self.config_watcher.stop().await;
310 self.refresh_manager.stop().await;
311 self.health_service.stop().await;
312 }
313
314 pub async fn warmup_token_store(&self) -> anyhow::Result<()> {
315 tracing::info!("Warming up token store (may take 30-60 seconds on first startup)...");
316
317 self.token_store
318 .warmup()
319 .await
320 .map_err(|e| anyhow::anyhow!("Failed to warm up token store: {}", e))?;
321
322 tracing::info!("Token store warmed up successfully - providers can now access tokens");
323 Ok(())
324 }
325}
326
327fn http_client_manager(proxy_url: &str) -> DomainResult<Arc<infrastructure::HttpClientManager>> {
328 let manager = infrastructure::HttpClientManager::new()?;
329 if let Err(e) = manager.set_proxy(proxy_url) {
330 tracing::error!(error = %e, "Ignoring general.proxy_url from config.toml");
331 }
332
333 Ok(Arc::new(manager))
334}
335
336#[cfg(test)]
337mod reset_tests {
338 use std::collections::HashMap;
339
340 use super::*;
341 use crate::application::test_support::FailOnceTokenStore;
342 use crate::infrastructure::{
343 ConfigLoader,
344 ConfigState,
345 PipedashConfig,
346 ProviderFileConfig,
347 SqliteConfigBackend,
348 };
349
350 async fn context(dir: &std::path::Path, tokens: Arc<dyn TokenStore>) -> CoreContext {
351 let mut config = PipedashConfig::default();
352 config.storage.data_dir = dir.display().to_string();
353 config.general.metrics_enabled = false;
354 config.general.default_refresh_interval = 73;
355 let path = dir.join("config.toml");
356 std::fs::write(&path, ConfigLoader::to_toml(&config).unwrap()).unwrap();
357 let config_state = ConfigState::from_config(config, path, Arc::clone(&tokens));
358 let pool = init_database(dir.join("pipedash.db")).await.unwrap();
359 let repository = Arc::new(Repository::new(
360 Arc::new(SqliteConfigBackend::new(pool.clone())),
361 infrastructure::database::DatabasePool::Sqlite(pool),
362 Arc::clone(&tokens),
363 ));
364 let bus: Arc<dyn EventBus> = Arc::new(NoOpEventBus);
365 let http = http_client_manager("").unwrap();
366 let log_store = infrastructure::run_logs::LogStore::open(dir.join("run-logs"), u64::MAX)
367 .await
368 .unwrap();
369 let services = application::Services::new(
370 repository,
371 Arc::clone(&http),
372 None,
373 Arc::clone(&bus),
374 Arc::clone(&config_state),
375 log_store,
376 );
377 CoreContext::from_services(bus, tokens, http, None, config_state, services)
378 }
379
380 async fn define_provider(core: &CoreContext, key: &str) {
381 core.config_state
382 .add_provider(
383 key.into(),
384 ProviderFileConfig {
385 name: None,
386 provider_type: "jenkins".into(),
387 token: "storage:1".into(),
388 refresh_interval: 30,
389 config: HashMap::new(),
390 },
391 )
392 .await
393 .unwrap();
394 }
395
396 fn provider(key: &str) -> ProviderConfig {
397 ProviderConfig {
398 id: None,
399 name: key.into(),
400 provider_type: "jenkins".into(),
401 token: "saved-token".into(),
402 config: HashMap::new(),
403 refresh_interval: 30,
404 version: None,
405 }
406 }
407
408 #[tokio::test]
409 async fn reset_removes_config_only_providers_and_persisted_metrics_when_disabled() {
410 let dir = tempfile::tempdir().unwrap();
411 let tokens = Arc::new(FailOnceTokenStore::default());
412 let core = context(dir.path(), tokens.clone()).await;
413 define_provider(&core, "ci-a").await;
414 define_provider(&core, "config-only").await;
415 core.provider_service
416 .repository()
417 .add_provider(&provider("ci-a"))
418 .await
419 .unwrap();
420 tokens.store_token(999, "orphan-token").await.unwrap();
421 let pool = core
422 .provider_service
423 .repository()
424 .cache_pool()
425 .as_sqlite()
426 .unwrap();
427 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();
428 sqlx::query("INSERT INTO metrics_processing_state (pipeline_id) VALUES ('retained')")
429 .execute(pool)
430 .await
431 .unwrap();
432 std::fs::write(dir.path().join("export.toml"), "user export").unwrap();
433
434 assert_eq!(core.factory_reset().await.unwrap(), 1);
435
436 assert!(core
437 .provider_service
438 .list_providers()
439 .await
440 .unwrap()
441 .is_empty());
442 assert!(tokens.get_all_tokens().await.unwrap().is_empty());
443 core.config_state.reload().await.unwrap();
444 assert!(core.config_state.get().await.providers.is_empty());
445 assert_eq!(
446 core.config_state
447 .get()
448 .await
449 .general
450 .default_refresh_interval,
451 73
452 );
453 core.config_state
454 .sync_providers_from_toml(
455 core.provider_service.repository().config_backend().as_ref(),
456 false,
457 )
458 .await
459 .unwrap();
460 assert!(core
461 .provider_service
462 .list_providers()
463 .await
464 .unwrap()
465 .is_empty());
466 let metrics: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pipeline_metrics")
467 .fetch_one(pool)
468 .await
469 .unwrap();
470 let processing: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM metrics_processing_state")
471 .fetch_one(pool)
472 .await
473 .unwrap();
474 assert_eq!((metrics, processing), (0, 0));
475 assert_eq!(
476 std::fs::read_to_string(dir.path().join("export.toml")).unwrap(),
477 "user export"
478 );
479 assert_eq!(core.factory_reset().await.unwrap(), 0);
480 }
481
482 #[tokio::test]
483 async fn reset_stops_on_token_deletion_failure_and_can_be_retried() {
484 let dir = tempfile::tempdir().unwrap();
485 let tokens = Arc::new(FailOnceTokenStore::default());
486 let core = context(dir.path(), tokens.clone()).await;
487 define_provider(&core, "ci-a").await;
488 let id = core
489 .provider_service
490 .repository()
491 .add_provider(&provider("ci-a"))
492 .await
493 .unwrap();
494 tokens.fail_next_delete();
495
496 assert!(core.factory_reset().await.is_err());
497 assert!(core
498 .provider_service
499 .repository()
500 .get_provider(id)
501 .await
502 .is_ok());
503 assert!(core
504 .config_state
505 .load_providers()
506 .unwrap()
507 .contains_key("ci-a"));
508
509 assert_eq!(core.factory_reset().await.unwrap(), 1);
510 assert!(tokens.get_all_tokens().await.unwrap().is_empty());
511 }
512
513 #[tokio::test]
514 async fn reset_reports_cache_and_metrics_failures_and_supports_retry() {
515 for table in ["workflow_parameters_cache", "pipeline_metrics"] {
516 let dir = tempfile::tempdir().unwrap();
517 let core = context(dir.path(), Arc::new(MemoryTokenStore::new())).await;
518 let pool = core
519 .provider_service
520 .repository()
521 .cache_pool()
522 .as_sqlite()
523 .unwrap();
524 sqlx::query(sqlx::AssertSqlSafe(format!(
525 "ALTER TABLE {table} RENAME TO retained_data"
526 )))
527 .execute(pool)
528 .await
529 .unwrap();
530
531 assert!(core.factory_reset().await.is_err(), "{table}");
532
533 sqlx::query(sqlx::AssertSqlSafe(format!(
534 "ALTER TABLE retained_data RENAME TO {table}"
535 )))
536 .execute(pool)
537 .await
538 .unwrap();
539 assert_eq!(core.factory_reset().await.unwrap(), 0);
540 }
541 }
542}