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 domain::{
13 DomainError,
14 DomainResult,
15 FetchStatus,
16 Pipeline,
17 PipelineRun,
18 PipelineStatus,
19 Provider,
20 ProviderConfig,
21 ProviderSummary,
22 TriggerParams,
23};
24pub use event::{
25 CoreEvent,
26 EventBus,
27 NoOpEventBus,
28};
29use infrastructure::database::{
30 init_database,
31 MetricsRepository,
32 Repository,
33};
34pub use infrastructure::{
35 ConflictResolution,
36 EnvTokenStore,
37 LocalStorage,
38 MemoryTokenStore,
39 StorageBackend,
40 SyncConfig,
41 SyncDirection,
42 SyncManager,
43 SyncResult,
44 TokenStore,
45};
46
47pub struct CoreContext {
48 pub event_bus: Arc<dyn EventBus>,
49
50 pub token_store: Arc<dyn TokenStore>,
51
52 pub http_client_manager: Arc<infrastructure::HttpClientManager>,
53
54 pub provider_service: Arc<application::ProviderService>,
55
56 pub pipeline_service: Arc<application::PipelineService>,
57
58 pub metrics_service: Option<Arc<application::MetricsService>>,
59
60 pub refresh_manager: Arc<application::RefreshManager>,
61}
62
63impl CoreContext {
64 pub async fn new(
65 data_dir: &std::path::Path, event_bus: Arc<dyn EventBus>, token_store: Arc<dyn TokenStore>,
66 ) -> anyhow::Result<Self> {
67 std::fs::create_dir_all(data_dir)?;
68
69 let config_state =
70 infrastructure::ConfigState::initialize(data_dir, token_store.clone()).await?;
71
72 let config = config_state.get().await;
73 let db_path = config.db_path();
74 let metrics_enabled = config.general.metrics_enabled;
75
76 if let Some(parent) = db_path.parent() {
77 std::fs::create_dir_all(parent)?;
78 }
79
80 let pool = init_database(db_path).await?;
81
82 let config_backend = Arc::new(infrastructure::SqliteConfigBackend::new(pool.clone()))
83 as Arc<dyn infrastructure::ConfigBackend>;
84 let cache_pool = pool.clone();
85
86 let sync_result = config_state
87 .sync_providers_from_toml(config_backend.as_ref(), false)
88 .await?;
89 if !sync_result.added.is_empty()
90 || !sync_result.updated.is_empty()
91 || !sync_result.removed.is_empty()
92 {
93 tracing::info!(
94 added = sync_result.added.len(),
95 updated = sync_result.updated.len(),
96 removed = sync_result.removed.len(),
97 "Synced providers from TOML to database on startup"
98 );
99 }
100
101 let metrics_service = if metrics_enabled {
102 let metrics_repository = Arc::new(MetricsRepository::new(pool.clone()));
103 let service = Arc::new(application::MetricsService::new(metrics_repository));
104
105 match service.check_and_repair_corruption().await {
106 Ok(repaired) => {
107 if !repaired.is_empty() {
108 tracing::warn!(
109 count = repaired.len(),
110 pipelines = ?repaired,
111 "AUTO-REPAIRED corrupted metrics processing states on startup"
112 );
113 } else {
114 tracing::debug!("No corrupted metrics states found on startup");
115 }
116 }
117 Err(e) => {
118 tracing::error!(
119 error = %e,
120 "Failed to check/repair metrics corruption on startup - metrics may not work correctly"
121 );
122 }
123 }
124
125 Some(service)
126 } else {
127 None
128 };
129
130 let repository = Arc::new(Repository::new(
131 config_backend,
132 infrastructure::database::DatabasePool::Sqlite(cache_pool),
133 token_store.clone(),
134 ));
135
136 let http_client_manager = Arc::new(infrastructure::HttpClientManager::new()?);
137
138 let provider_service = Arc::new(application::ProviderService::new(
139 Arc::clone(&repository),
140 Arc::clone(&http_client_manager),
141 Arc::clone(&event_bus),
142 ));
143 let pipeline_service = Arc::new(application::PipelineService::new(
144 Arc::clone(&repository),
145 Arc::clone(&provider_service),
146 metrics_service.clone(),
147 Arc::clone(&event_bus),
148 ));
149 let refresh_manager = Arc::new(application::RefreshManager::new(
150 Arc::clone(&pipeline_service),
151 metrics_service.clone(),
152 Arc::clone(&event_bus),
153 ));
154
155 Ok(Self {
156 event_bus,
157 token_store,
158 http_client_manager,
159 provider_service,
160 pipeline_service,
161 metrics_service,
162 refresh_manager,
163 })
164 }
165
166 pub async fn with_storage_manager(
167 storage_manager: &infrastructure::StorageManager, event_bus: Arc<dyn EventBus>,
168 ) -> anyhow::Result<Self> {
169 let token_store = storage_manager.token_store().await;
170 let config_backend = storage_manager.config_backend();
171 let data_dir = storage_manager.config().data_dir();
172 let _storage_config = storage_manager.config();
173
174 let config_state =
175 infrastructure::ConfigState::initialize(&data_dir, token_store.clone()).await?;
176
177 let sync_result = config_state
178 .sync_providers_from_toml(config_backend.as_ref(), false)
179 .await?;
180 if !sync_result.added.is_empty()
181 || !sync_result.updated.is_empty()
182 || !sync_result.removed.is_empty()
183 {
184 tracing::info!(
185 added = sync_result.added.len(),
186 updated = sync_result.updated.len(),
187 removed = sync_result.removed.len(),
188 "Synced providers from TOML to database on startup"
189 );
190 }
191
192 let cache_pool = storage_manager.database_pool();
193
194 let repository = Arc::new(infrastructure::database::Repository::new(
195 config_backend.clone(),
196 cache_pool.clone(),
197 token_store.clone(),
198 ));
199
200 let http_client_manager = Arc::new(infrastructure::HttpClientManager::new()?);
201
202 let config = config_state.get().await;
203 let metrics_service = if config.general.metrics_enabled {
204 let metrics_repository = Arc::new(
205 infrastructure::database::MetricsRepository::new_from_pool(cache_pool),
206 );
207 let service = Arc::new(application::MetricsService::new(metrics_repository));
208
209 match service.check_and_repair_corruption().await {
210 Ok(repaired) => {
211 if !repaired.is_empty() {
212 tracing::warn!(
213 count = repaired.len(),
214 pipelines = ?repaired,
215 "AUTO-REPAIRED corrupted metrics processing states on startup"
216 );
217 }
218 }
219 Err(e) => {
220 tracing::error!(
221 error = %e,
222 "Failed to check/repair metrics corruption on startup"
223 );
224 }
225 }
226
227 Some(service)
228 } else {
229 None
230 };
231
232 let provider_service = Arc::new(application::ProviderService::new(
233 repository.clone(),
234 Arc::clone(&http_client_manager),
235 Arc::clone(&event_bus),
236 ));
237 let pipeline_service = Arc::new(application::PipelineService::new(
238 repository.clone(),
239 Arc::clone(&provider_service),
240 metrics_service.clone(),
241 Arc::clone(&event_bus),
242 ));
243 let refresh_manager = Arc::new(application::RefreshManager::new(
244 Arc::clone(&pipeline_service),
245 metrics_service.clone(),
246 Arc::clone(&event_bus),
247 ));
248
249 Ok(Self {
250 event_bus,
251 token_store,
252 http_client_manager,
253 provider_service,
254 pipeline_service,
255 metrics_service,
256 refresh_manager,
257 })
258 }
259
260 pub async fn start_background_tasks(&self) {
261 let provider_service = Arc::clone(&self.provider_service);
262 let refresh_manager = Arc::clone(&self.refresh_manager);
263
264 tokio::spawn(async move {
265 if let Err(e) = provider_service.load_all_providers().await {
266 tracing::warn!("Failed to load providers during startup: {}", e);
267 }
268
269 refresh_manager.start().await;
270 });
271 }
272
273 pub async fn shutdown(&self) {
274 self.refresh_manager.stop().await;
275 }
276
277 pub async fn warmup_token_store(&self) -> anyhow::Result<()> {
278 tracing::info!("Warming up token store (may take 30-60 seconds on first startup)...");
279
280 self.token_store
281 .warmup()
282 .await
283 .map_err(|e| anyhow::anyhow!("Failed to warm up token store: {}", e))?;
284
285 tracing::info!("Token store warmed up successfully - providers can now access tokens");
286 Ok(())
287 }
288}