pipedash_core/infrastructure/config/
analysis.rs1use std::sync::Arc;
2
3use serde::Serialize;
4
5use super::{
6 PipedashConfig,
7 StorageBackend,
8 StorageManager,
9};
10use crate::domain::{
11 DomainError,
12 DomainResult,
13};
14use crate::event::EventBus;
15use crate::infrastructure::migration::{
16 MigrationOptions,
17 MigrationOrchestrator,
18 MigrationPlan,
19};
20use crate::infrastructure::token_store::TokenStore;
21
22#[derive(Debug, Clone, Serialize)]
23pub struct ConfigIssue {
24 pub field: String,
25 pub message: String,
26 pub code: String,
27}
28
29#[derive(Debug, Serialize)]
30pub struct ConnectionTestResult {
31 pub success: bool,
32 pub message: String,
33 #[serde(skip_serializing_if = "Option::is_none")]
34 pub latency_ms: Option<u64>,
35}
36
37#[derive(Debug, Serialize)]
38pub struct ConfigDataStats {
39 pub providers_count: usize,
40 pub tokens_count: usize,
41 pub cache_entries_count: usize,
42}
43
44#[derive(Debug, Serialize)]
45pub struct ConfigAnalysis {
46 pub valid: bool,
47 pub errors: Vec<ConfigIssue>,
48 pub warnings: Vec<ConfigIssue>,
49 #[serde(skip_serializing_if = "Option::is_none")]
50 pub migration_plan: Option<MigrationPlan>,
51 #[serde(skip_serializing_if = "Option::is_none")]
52 pub postgres_connection: Option<ConnectionTestResult>,
53 #[serde(skip_serializing_if = "Option::is_none")]
54 pub stats: Option<ConfigDataStats>,
55}
56
57pub async fn analyze_config(
58 new_content: &str, storage_manager: &StorageManager, event_bus: Option<Arc<dyn EventBus>>,
59 target_token_store: Option<Arc<dyn TokenStore>>,
60) -> DomainResult<ConfigAnalysis> {
61 let new_config: PipedashConfig = toml::from_str(new_content)
62 .map_err(|e| DomainError::InvalidConfig(format!("Invalid TOML syntax: {}", e)))?;
63
64 let validation = new_config.validate();
65
66 let errors = validation
67 .errors
68 .iter()
69 .map(|e| ConfigIssue {
70 field: e.field.clone(),
71 message: e.message.clone(),
72 code: e.code.to_string(),
73 })
74 .collect();
75
76 let warnings = validation
77 .warnings
78 .iter()
79 .map(|w| ConfigIssue {
80 field: w.field.clone(),
81 message: w.message.clone(),
82 code: w.code.to_string(),
83 })
84 .collect();
85
86 let current_config = storage_manager.config();
87
88 let (migration_plan, stats) = if backends_changed(current_config, &new_config) {
89 let orchestrator =
90 MigrationOrchestrator::from_manager(storage_manager, event_bus, target_token_store)
91 .await;
92 let plan = orchestrator.plan_migration(new_config.clone(), &MigrationOptions::default())?;
93
94 let providers = storage_manager.config_backend().list_providers().await?;
95 let tokens = storage_manager.token_store().await.get_all_tokens().await?;
96 let cache_entries = storage_manager.cache_backend().list(None).await?;
97
98 let stats = ConfigDataStats {
99 providers_count: providers.len(),
100 tokens_count: tokens.len(),
101 cache_entries_count: cache_entries.len(),
102 };
103
104 (Some(plan), Some(stats))
105 } else {
106 (None, None)
107 };
108
109 let postgres_connection = if should_test_postgres_connection(current_config, &new_config) {
110 Some(test_postgres_connection(&new_config.storage.postgres.connection_string).await)
111 } else {
112 None
113 };
114
115 Ok(ConfigAnalysis {
116 valid: validation.is_ok(),
117 errors,
118 warnings,
119 migration_plan,
120 postgres_connection,
121 stats,
122 })
123}
124
125fn backends_changed(old: &PipedashConfig, new: &PipedashConfig) -> bool {
126 old.storage.backend != new.storage.backend || old.data_dir() != new.data_dir()
127}
128
129fn should_test_postgres_connection(old: &PipedashConfig, new: &PipedashConfig) -> bool {
130 let uses_postgres = new.storage.backend == StorageBackend::Postgres;
131
132 let connection_changed =
133 old.storage.postgres.connection_string != new.storage.postgres.connection_string;
134
135 let switching_to_postgres = old.storage.backend != StorageBackend::Postgres
136 && new.storage.backend == StorageBackend::Postgres;
137
138 uses_postgres && (connection_changed || switching_to_postgres)
139}
140
141async fn test_postgres_connection(connection_string: &str) -> ConnectionTestResult {
142 #[cfg(feature = "postgres")]
143 {
144 use std::time::Instant;
145
146 use sqlx::postgres::PgPoolOptions;
147
148 if connection_string.is_empty() {
149 return ConnectionTestResult {
150 success: false,
151 message: "PostgreSQL connection string is empty".to_string(),
152 latency_ms: None,
153 };
154 }
155
156 let start = Instant::now();
157
158 let timeout_duration = std::time::Duration::from_secs(5);
159 let connect_result = tokio::time::timeout(timeout_duration, async {
160 let pool = PgPoolOptions::new()
161 .max_connections(1)
162 .acquire_timeout(std::time::Duration::from_secs(3))
163 .connect(connection_string)
164 .await?;
165
166 sqlx::query("SELECT 1").execute(&pool).await?;
167
168 Ok::<_, sqlx::Error>(pool)
169 })
170 .await;
171
172 match connect_result {
173 Ok(Ok(_pool)) => {
174 let latency = start.elapsed().as_millis() as u64;
175 ConnectionTestResult {
176 success: true,
177 message: "Successfully connected to PostgreSQL".to_string(),
178 latency_ms: Some(latency),
179 }
180 }
181 Ok(Err(e)) => ConnectionTestResult {
182 success: false,
183 message: format!("Failed to connect to PostgreSQL: {}", e),
184 latency_ms: None,
185 },
186 Err(_) => ConnectionTestResult {
187 success: false,
188 message: format!(
189 "Connection timeout after {} seconds",
190 timeout_duration.as_secs()
191 ),
192 latency_ms: None,
193 },
194 }
195 }
196
197 #[cfg(not(feature = "postgres"))]
198 {
199 let _ = connection_string;
200 ConnectionTestResult {
201 success: false,
202 message: "PostgreSQL feature not enabled. Compile with --features postgres".to_string(),
203 latency_ms: None,
204 }
205 }
206}
207
208#[cfg(test)]
209mod tests {
210 use tempfile::TempDir;
211
212 use super::*;
213 use crate::infrastructure::token_store::MemoryTokenStore;
214
215 async fn sqlite_manager(data_dir: &TempDir) -> StorageManager {
216 let mut config = PipedashConfig::default();
217 config.storage.data_dir = data_dir.path().to_string_lossy().to_string();
218
219 StorageManager::with_token_store(config, Arc::new(MemoryTokenStore::new()), true)
220 .await
221 .unwrap()
222 }
223
224 #[tokio::test]
225 async fn switching_sqlite_to_postgres_returns_a_migration_plan() {
226 let data_dir = TempDir::new().unwrap();
227 let manager = sqlite_manager(&data_dir).await;
228 let content = format!(
229 "[storage]\ndata_dir = {:?}\nbackend = \"postgres\"\n\n[storage.postgres]\nconnection_string = \"postgres://user:pass@127.0.0.1:1/pipedash\"\n",
230 data_dir.path().to_string_lossy()
231 );
232
233 let analysis = analyze_config(&content, &manager, None, None)
234 .await
235 .unwrap();
236
237 let plan = analysis.migration_plan.expect("migration plan");
238 assert_eq!(plan.to.storage.backend, StorageBackend::Postgres);
239 assert!(!plan.steps.is_empty());
240 assert!(analysis.stats.is_some());
241 assert!(
242 !analysis
243 .postgres_connection
244 .expect("connection test")
245 .success
246 );
247 }
248
249 #[tokio::test]
250 async fn unchanged_storage_returns_no_migration_plan() {
251 let data_dir = TempDir::new().unwrap();
252 let manager = sqlite_manager(&data_dir).await;
253 let content = format!(
254 "[storage]\ndata_dir = {:?}\nbackend = \"sqlite\"\n",
255 data_dir.path().to_string_lossy()
256 );
257
258 let analysis = analyze_config(&content, &manager, None, None)
259 .await
260 .unwrap();
261
262 assert!(analysis.valid);
263 assert!(analysis.migration_plan.is_none());
264 assert!(analysis.postgres_connection.is_none());
265 }
266
267 #[tokio::test]
268 async fn invalid_toml_is_an_invalid_config_error() {
269 let data_dir = TempDir::new().unwrap();
270 let manager = sqlite_manager(&data_dir).await;
271
272 let err = analyze_config("[storage", &manager, None, None)
273 .await
274 .unwrap_err();
275
276 assert!(matches!(err, DomainError::InvalidConfig(_)));
277 }
278}