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