pipedash_core/infrastructure/database/
postgres_backend.rs1use std::collections::HashMap;
2
3use async_trait::async_trait;
4use sqlx::postgres::PgPool;
5use sqlx::Row;
6
7use crate::domain::{
8 DomainError,
9 DomainResult,
10 ProviderConfig,
11};
12use crate::infrastructure::config_backend::{
13 ConfigBackend,
14 ConfigExport,
15 StoredPermissions,
16};
17
18pub struct PostgresConfigBackend {
19 pool: PgPool,
20}
21
22impl PostgresConfigBackend {
23 pub async fn new(connection_string: &str) -> DomainResult<Self> {
24 let pool = crate::infrastructure::database::init_postgres_database(connection_string)
25 .await
26 .map_err(|e| {
27 DomainError::DatabaseError(format!("Failed to initialize PostgreSQL: {}", e))
28 })?;
29
30 Ok(Self { pool })
31 }
32
33 pub fn with_pool(pool: PgPool) -> Self {
34 Self { pool }
35 }
36
37 pub fn pool(&self) -> &PgPool {
38 &self.pool
39 }
40}
41
42#[async_trait]
43impl ConfigBackend for PostgresConfigBackend {
44 async fn list_providers(&self) -> DomainResult<Vec<ProviderConfig>> {
45 let rows = sqlx::query(
46 "SELECT id, name, provider_type, config_json, refresh_interval, version FROM providers",
47 )
48 .fetch_all(&self.pool)
49 .await
50 .map_err(|e| DomainError::DatabaseError(format!("Failed to list providers: {}", e)))?;
51
52 let mut providers = Vec::new();
53 for row in rows {
54 let id: i64 = row.get("id");
55 let config_json: String = row.get("config_json");
56 let config: HashMap<String, String> = if config_json.trim().is_empty() {
57 HashMap::new()
58 } else {
59 serde_json::from_str(&config_json).map_err(|e| {
60 DomainError::DatabaseError(format!("Failed to parse config for {}: {}", id, e))
61 })?
62 };
63
64 providers.push(ProviderConfig {
65 id: Some(id),
66 name: row.get("name"),
67 provider_type: row.get("provider_type"),
68 config,
69 token: String::new(), refresh_interval: row.get::<i32, _>("refresh_interval") as i64,
71 version: Some(row.get::<i32, _>("version") as i64),
72 });
73 }
74
75 Ok(providers)
76 }
77
78 async fn get_provider(&self, id: i64) -> DomainResult<Option<ProviderConfig>> {
79 let row = sqlx::query(
80 "SELECT id, name, provider_type, config_json, refresh_interval, version FROM providers WHERE id = $1",
81 )
82 .bind(id)
83 .fetch_optional(&self.pool)
84 .await
85 .map_err(|e| DomainError::DatabaseError(format!("Failed to get provider: {}", e)))?;
86
87 match row {
88 Some(row) => {
89 let config_json: String = row.get("config_json");
90 let config: HashMap<String, String> = if config_json.trim().is_empty() {
91 HashMap::new()
92 } else {
93 serde_json::from_str(&config_json).map_err(|e| {
94 DomainError::DatabaseError(format!("Failed to parse config: {}", e))
95 })?
96 };
97
98 Ok(Some(ProviderConfig {
99 id: Some(id),
100 name: row.get("name"),
101 provider_type: row.get("provider_type"),
102 config,
103 token: String::new(),
104 refresh_interval: row.get::<i32, _>("refresh_interval") as i64,
105 version: Some(row.get::<i32, _>("version") as i64),
106 }))
107 }
108 None => Ok(None),
109 }
110 }
111
112 async fn create_provider(&self, config: &ProviderConfig) -> DomainResult<i64> {
113 let existing: Option<i64> = sqlx::query_scalar("SELECT id FROM providers WHERE name = $1")
114 .bind(&config.name)
115 .fetch_optional(&self.pool)
116 .await
117 .map_err(|e| DomainError::DatabaseError(format!("Failed to check existing: {}", e)))?;
118
119 if existing.is_some() {
120 return Err(DomainError::InvalidConfig(format!(
121 "Provider '{}' already exists",
122 config.name
123 )));
124 }
125
126 let config_json = serde_json::to_string(&config.config).map_err(|e| {
127 DomainError::DatabaseError(format!("Failed to serialize config: {}", e))
128 })?;
129
130 let id: i64 = sqlx::query_scalar(
131 r#"
132 INSERT INTO providers (name, provider_type, token_encrypted, config_json, refresh_interval)
133 VALUES ($1, $2, '', $3, $4)
134 RETURNING id
135 "#,
136 )
137 .bind(&config.name)
138 .bind(&config.provider_type)
139 .bind(&config_json)
140 .bind(config.refresh_interval)
141 .fetch_one(&self.pool)
142 .await
143 .map_err(|e| DomainError::DatabaseError(format!("Failed to insert provider: {}", e)))?;
144
145 Ok(id)
146 }
147
148 async fn update_provider(&self, id: i64, config: &ProviderConfig) -> DomainResult<()> {
149 let config_json = serde_json::to_string(&config.config).map_err(|e| {
150 DomainError::DatabaseError(format!("Failed to serialize config: {}", e))
151 })?;
152
153 let result = sqlx::query(
154 r#"
155 UPDATE providers
156 SET name = $1, provider_type = $2, config_json = $3, refresh_interval = $4, updated_at = NOW()
157 WHERE id = $5
158 "#,
159 )
160 .bind(&config.name)
161 .bind(&config.provider_type)
162 .bind(&config_json)
163 .bind(config.refresh_interval)
164 .bind(id)
165 .execute(&self.pool)
166 .await
167 .map_err(|e| DomainError::DatabaseError(format!("Failed to update provider: {}", e)))?;
168
169 if result.rows_affected() == 0 {
170 return Err(DomainError::ProviderNotFound(format!(
171 "Provider {} not found",
172 id
173 )));
174 }
175
176 Ok(())
177 }
178
179 async fn delete_provider(&self, id: i64) -> DomainResult<()> {
180 let result = sqlx::query("DELETE FROM providers WHERE id = $1")
181 .bind(id)
182 .execute(&self.pool)
183 .await
184 .map_err(|e| DomainError::DatabaseError(format!("Failed to delete provider: {}", e)))?;
185
186 if result.rows_affected() == 0 {
187 return Err(DomainError::ProviderNotFound(format!(
188 "Provider {} not found",
189 id
190 )));
191 }
192
193 Ok(())
194 }
195
196 async fn get_table_preferences(
197 &self, provider_id: i64, table_id: &str,
198 ) -> DomainResult<Option<String>> {
199 let result: Option<String> = sqlx::query_scalar(
200 "SELECT preferences_json FROM table_preferences WHERE provider_id = $1 AND table_id = $2",
201 )
202 .bind(provider_id)
203 .bind(table_id)
204 .fetch_optional(&self.pool)
205 .await
206 .map_err(|e| DomainError::DatabaseError(format!("Failed to get preferences: {}", e)))?;
207
208 Ok(result)
209 }
210
211 async fn set_table_preferences(
212 &self, provider_id: i64, table_id: &str, preferences_json: &str,
213 ) -> DomainResult<()> {
214 sqlx::query(
215 r#"
216 INSERT INTO table_preferences (provider_id, table_id, preferences_json)
217 VALUES ($1, $2, $3)
218 ON CONFLICT (provider_id, table_id) DO UPDATE SET
219 preferences_json = EXCLUDED.preferences_json,
220 updated_at = NOW()
221 "#,
222 )
223 .bind(provider_id)
224 .bind(table_id)
225 .bind(preferences_json)
226 .execute(&self.pool)
227 .await
228 .map_err(|e| DomainError::DatabaseError(format!("Failed to save preferences: {}", e)))?;
229
230 Ok(())
231 }
232
233 async fn store_permissions(
234 &self, provider_id: i64, permissions: &StoredPermissions,
235 ) -> DomainResult<()> {
236 sqlx::query("DELETE FROM provider_permissions WHERE provider_id = $1")
237 .bind(provider_id)
238 .execute(&self.pool)
239 .await
240 .map_err(|e| {
241 DomainError::DatabaseError(format!("Failed to clear permissions: {}", e))
242 })?;
243
244 for (permission_name, granted) in &permissions.permissions {
245 sqlx::query(
246 r#"
247 INSERT INTO provider_permissions (provider_id, permission_name, granted, checked_at)
248 VALUES ($1, $2, $3, $4)
249 "#,
250 )
251 .bind(provider_id)
252 .bind(permission_name)
253 .bind(*granted)
254 .bind(permissions.last_checked)
255 .execute(&self.pool)
256 .await
257 .map_err(|e| {
258 DomainError::DatabaseError(format!("Failed to store permission: {}", e))
259 })?;
260 }
261
262 Ok(())
263 }
264
265 async fn get_permissions(&self, provider_id: i64) -> DomainResult<Option<StoredPermissions>> {
266 let rows = sqlx::query(
267 "SELECT permission_name, granted, checked_at FROM provider_permissions WHERE provider_id = $1",
268 )
269 .bind(provider_id)
270 .fetch_all(&self.pool)
271 .await
272 .map_err(|e| DomainError::DatabaseError(format!("Failed to get permissions: {}", e)))?;
273
274 if rows.is_empty() {
275 return Ok(None);
276 }
277
278 let last_checked: chrono::DateTime<chrono::Utc> = rows[0].get("checked_at");
279
280 let mut permissions = HashMap::new();
281 for row in rows {
282 let permission_name: String = row.get("permission_name");
283 let granted: bool = row.get("granted");
284 permissions.insert(permission_name, granted);
285 }
286
287 Ok(Some(StoredPermissions {
288 permissions,
289 last_checked,
290 }))
291 }
292
293 async fn export_all(&self) -> DomainResult<ConfigExport> {
294 let providers = self.list_providers().await?;
295
296 let pref_rows =
297 sqlx::query("SELECT provider_id, table_id, preferences_json FROM table_preferences")
298 .fetch_all(&self.pool)
299 .await
300 .map_err(|e| {
301 DomainError::DatabaseError(format!("Failed to export preferences: {}", e))
302 })?;
303
304 let mut table_preferences = HashMap::new();
305 for row in pref_rows {
306 let provider_id: i64 = row.get("provider_id");
307 let table_id: String = row.get("table_id");
308 let preferences_json: String = row.get("preferences_json");
309 let key = format!("{}:{}", provider_id, table_id);
310 table_preferences.insert(key, preferences_json);
311 }
312
313 let perm_rows = sqlx::query(
314 "SELECT provider_id, permission_name, granted, checked_at FROM provider_permissions",
315 )
316 .fetch_all(&self.pool)
317 .await
318 .map_err(|e| DomainError::DatabaseError(format!("Failed to export permissions: {}", e)))?;
319
320 let mut permissions: HashMap<i64, StoredPermissions> = HashMap::new();
321 for row in perm_rows {
322 let provider_id: i64 = row.get("provider_id");
323 let permission_name: String = row.get("permission_name");
324 let granted: bool = row.get("granted");
325 let last_checked: chrono::DateTime<chrono::Utc> = row.get("checked_at");
326
327 let entry = permissions
328 .entry(provider_id)
329 .or_insert_with(|| StoredPermissions {
330 permissions: HashMap::new(),
331 last_checked,
332 });
333 entry.permissions.insert(permission_name, granted);
334 }
335
336 Ok(ConfigExport {
337 version: "1.0".to_string(),
338 providers,
339 table_preferences,
340 permissions,
341 })
342 }
343
344 async fn import_all(&self, data: &ConfigExport) -> DomainResult<HashMap<i64, i64>> {
345 let mut id_mapping: HashMap<i64, i64> = HashMap::new();
346
347 for provider in &data.providers {
348 let old_id = provider.id.unwrap_or(0);
349
350 let existing: Option<i64> =
351 sqlx::query_scalar("SELECT id FROM providers WHERE name = $1")
352 .bind(&provider.name)
353 .fetch_optional(&self.pool)
354 .await
355 .map_err(|e| {
356 DomainError::DatabaseError(format!("Failed to check existing: {}", e))
357 })?;
358
359 let new_id = if let Some(existing_id) = existing {
360 existing_id
361 } else {
362 self.create_provider(provider).await?
363 };
364
365 if old_id != 0 {
366 id_mapping.insert(old_id, new_id);
367 }
368 }
369
370 for (key, preferences_json) in &data.table_preferences {
371 let parts: Vec<&str> = key.splitn(2, ':').collect();
372 if parts.len() == 2 {
373 if let Ok(old_provider_id) = parts[0].parse::<i64>() {
374 let table_id = parts[1];
375 let new_provider_id = id_mapping
376 .get(&old_provider_id)
377 .copied()
378 .unwrap_or(old_provider_id);
379 if self.get_provider(new_provider_id).await?.is_some() {
380 self.set_table_preferences(new_provider_id, table_id, preferences_json)
381 .await?;
382 }
383 }
384 }
385 }
386
387 for (old_provider_id, perms) in &data.permissions {
388 let new_provider_id = id_mapping
389 .get(old_provider_id)
390 .copied()
391 .unwrap_or(*old_provider_id);
392 if self.get_provider(new_provider_id).await?.is_some() {
393 self.store_permissions(new_provider_id, perms).await?;
394 }
395 }
396
397 Ok(id_mapping)
398 }
399}