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 pub health: application::HealthHandle,
63
64 pub health_service: Arc<application::ProviderHealthService>,
65
66 pub config_state: Arc<infrastructure::ConfigState>,
67}
68
69impl CoreContext {
70 fn from_services(
71 event_bus: Arc<dyn EventBus>, token_store: Arc<dyn TokenStore>,
72 http_client_manager: Arc<infrastructure::HttpClientManager>,
73 metrics_service: Option<Arc<application::MetricsService>>,
74 config_state: Arc<infrastructure::ConfigState>, services: application::Services,
75 ) -> Self {
76 Self {
77 event_bus,
78 token_store,
79 http_client_manager,
80 provider_service: services.provider_service,
81 pipeline_service: services.pipeline_service,
82 metrics_service,
83 refresh_manager: services.refresh_manager,
84 health: services.health,
85 health_service: services.health_service,
86 config_state,
87 }
88 }
89
90 pub async fn new(
91 data_dir: &std::path::Path, event_bus: Arc<dyn EventBus>, token_store: Arc<dyn TokenStore>,
92 ) -> anyhow::Result<Self> {
93 std::fs::create_dir_all(data_dir)?;
94
95 let config_state =
96 infrastructure::ConfigState::initialize(data_dir, token_store.clone()).await?;
97
98 let config = config_state.get().await;
99 let db_path = config.db_path();
100 let metrics_enabled = config.general.metrics_enabled;
101
102 if let Some(parent) = db_path.parent() {
103 std::fs::create_dir_all(parent)?;
104 }
105
106 let pool = init_database(db_path).await?;
107
108 let config_backend = Arc::new(infrastructure::SqliteConfigBackend::new(pool.clone()))
109 as Arc<dyn infrastructure::ConfigBackend>;
110 let cache_pool = pool.clone();
111
112 let sync_result = config_state
113 .sync_providers_from_toml(config_backend.as_ref(), false)
114 .await?;
115 if !sync_result.added.is_empty()
116 || !sync_result.updated.is_empty()
117 || !sync_result.removed.is_empty()
118 {
119 tracing::info!(
120 added = sync_result.added.len(),
121 updated = sync_result.updated.len(),
122 removed = sync_result.removed.len(),
123 "Synced providers from TOML to database on startup"
124 );
125 }
126
127 let metrics_service = metrics_enabled.then(|| {
128 let metrics_repository = Arc::new(MetricsRepository::new(pool.clone()));
129 Arc::new(application::MetricsService::new(metrics_repository))
130 });
131
132 let repository = Arc::new(Repository::new(
133 config_backend,
134 infrastructure::database::DatabasePool::Sqlite(cache_pool),
135 token_store.clone(),
136 ));
137
138 let http_client_manager = Arc::new(infrastructure::HttpClientManager::new()?);
139
140 let services = application::Services::new(
141 repository,
142 Arc::clone(&http_client_manager),
143 metrics_service.clone(),
144 Arc::clone(&event_bus),
145 Arc::clone(&config_state),
146 );
147
148 Ok(Self::from_services(
149 event_bus,
150 token_store,
151 http_client_manager,
152 metrics_service,
153 config_state,
154 services,
155 ))
156 }
157
158 pub async fn with_storage_manager(
159 storage_manager: &infrastructure::StorageManager, event_bus: Arc<dyn EventBus>,
160 ) -> anyhow::Result<Self> {
161 let token_store = storage_manager.token_store().await;
162 let config_backend = storage_manager.config_backend();
163 let data_dir = storage_manager.config().data_dir();
164 let _storage_config = storage_manager.config();
165
166 let config_state =
167 infrastructure::ConfigState::initialize(&data_dir, token_store.clone()).await?;
168
169 let sync_result = config_state
170 .sync_providers_from_toml(config_backend.as_ref(), false)
171 .await?;
172 if !sync_result.added.is_empty()
173 || !sync_result.updated.is_empty()
174 || !sync_result.removed.is_empty()
175 {
176 tracing::info!(
177 added = sync_result.added.len(),
178 updated = sync_result.updated.len(),
179 removed = sync_result.removed.len(),
180 "Synced providers from TOML to database on startup"
181 );
182 }
183
184 let cache_pool = storage_manager.database_pool();
185
186 let repository = Arc::new(infrastructure::database::Repository::new(
187 config_backend.clone(),
188 cache_pool.clone(),
189 token_store.clone(),
190 ));
191
192 let http_client_manager = Arc::new(infrastructure::HttpClientManager::new()?);
193
194 let config = config_state.get().await;
195 let metrics_service = config.general.metrics_enabled.then(|| {
196 let metrics_repository = Arc::new(
197 infrastructure::database::MetricsRepository::new_from_pool(cache_pool),
198 );
199 Arc::new(application::MetricsService::new(metrics_repository))
200 });
201
202 let services = application::Services::new(
203 repository,
204 Arc::clone(&http_client_manager),
205 metrics_service.clone(),
206 Arc::clone(&event_bus),
207 Arc::clone(&config_state),
208 );
209
210 Ok(Self::from_services(
211 event_bus,
212 token_store,
213 http_client_manager,
214 metrics_service,
215 config_state,
216 services,
217 ))
218 }
219
220 pub async fn start_background_tasks(&self) {
221 let provider_service = Arc::clone(&self.provider_service);
222 let refresh_manager = Arc::clone(&self.refresh_manager);
223 let health_service = Arc::clone(&self.health_service);
224 let health = self.health.clone();
225
226 tokio::spawn(async move {
227 if let Err(e) = provider_service.load_all_providers().await {
228 tracing::warn!("Failed to load providers during startup: {}", e);
229 }
230
231 health_service.start().await;
232 health.send(application::HealthInput::ProvidersChanged {
233 changed: provider_service.health_inputs().await,
234 });
235 refresh_manager.start().await;
236 });
237 }
238
239 pub async fn shutdown(&self) {
240 self.refresh_manager.stop().await;
241 self.health_service.stop().await;
242 }
243
244 pub async fn warmup_token_store(&self) -> anyhow::Result<()> {
245 tracing::info!("Warming up token store (may take 30-60 seconds on first startup)...");
246
247 self.token_store
248 .warmup()
249 .await
250 .map_err(|e| anyhow::anyhow!("Failed to warm up token store: {}", e))?;
251
252 tracing::info!("Token store warmed up successfully - providers can now access tokens");
253 Ok(())
254 }
255}