pipedash_core/infrastructure/database/
sqlite_backend.rs

1use std::collections::HashMap;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use sqlx::{
6    Row as SqlxRow,
7    SqlitePool,
8};
9use tokio::time::sleep;
10
11use crate::domain::{
12    DomainError,
13    DomainResult,
14    ProviderConfig,
15};
16use crate::infrastructure::config_backend::{
17    ConfigBackend,
18    ConfigExport,
19    StoredPermissions,
20};
21
22async fn retry_on_busy<F, Fut, T>(operation: F) -> DomainResult<T>
23where
24    F: Fn() -> Fut,
25    Fut: std::future::Future<Output = DomainResult<T>>,
26{
27    const MAX_RETRIES: u32 = 3; // Reduced from 10 to 3 to prevent 5-minute delays
28    const INITIAL_DELAY_MS: u64 = 10;
29
30    let mut attempt = 0;
31    loop {
32        match operation().await {
33            Ok(result) => {
34                if attempt > 0 {
35                    tracing::warn!(
36                        "SQLite operation succeeded after {} retries ({}ms total)",
37                        attempt,
38                        INITIAL_DELAY_MS * (2_u64.pow(attempt) - 1)
39                    );
40                }
41                return Ok(result);
42            }
43            Err(DomainError::DatabaseError(ref msg))
44                if (msg.contains("database is locked")
45                    || msg.contains("SQLITE_BUSY")
46                    || msg.contains("locked"))
47                    && attempt < MAX_RETRIES =>
48            {
49                attempt += 1;
50                let delay = INITIAL_DELAY_MS * 2_u64.pow(attempt - 1);
51
52                if attempt > 3 {
53                    tracing::warn!(
54                        "SQLite busy, retry {}/{} (waiting {}ms)",
55                        attempt,
56                        MAX_RETRIES,
57                        delay
58                    );
59                }
60
61                sleep(Duration::from_millis(delay)).await;
62            }
63            Err(e) => {
64                if attempt > 0 {
65                    tracing::error!("SQLite operation failed after {} retries: {}", attempt, e);
66                }
67                return Err(e);
68            }
69        }
70    }
71}
72
73pub struct SqliteConfigBackend {
74    pool: SqlitePool,
75}
76
77impl SqliteConfigBackend {
78    pub fn new(pool: SqlitePool) -> Self {
79        Self { pool }
80    }
81
82    pub fn pool(&self) -> &SqlitePool {
83        &self.pool
84    }
85}
86
87#[async_trait]
88impl ConfigBackend for SqliteConfigBackend {
89    async fn list_providers(&self) -> DomainResult<Vec<ProviderConfig>> {
90        retry_on_busy(|| async {
91            let rows = sqlx::query(
92                r#"SELECT id, name, provider_type, token_encrypted, config_json, refresh_interval, version FROM providers"#,
93            )
94            .fetch_all(&self.pool)
95            .await
96            .map_err(|e| DomainError::DatabaseError(format!("Failed to list providers: {}", e)))?;
97
98            let mut providers = Vec::new();
99            for row in rows {
100                let id: i64 = row.get("id");
101                let config_json: String = row.get("config_json");
102                let config: HashMap<String, String> = if config_json.trim().is_empty() {
103                    HashMap::new()
104                } else {
105                    serde_json::from_str(&config_json).map_err(|e| {
106                        DomainError::DatabaseError(format!(
107                            "Failed to parse config for {}: {}",
108                            id, e
109                        ))
110                    })?
111                };
112
113                let token_ref: String = row.get("token_encrypted");
114
115                providers.push(ProviderConfig {
116                    id: Some(id),
117                    name: row.get("name"),
118                    provider_type: row.get("provider_type"),
119                    config,
120                    token: token_ref,
121                    refresh_interval: row.get("refresh_interval"),
122                    version: Some(row.get("version")),
123                });
124            }
125
126            Ok(providers)
127        })
128        .await
129    }
130
131    async fn get_provider(&self, id: i64) -> DomainResult<Option<ProviderConfig>> {
132        retry_on_busy(|| async {
133            let row = sqlx::query(
134                r#"SELECT id, name, provider_type, token_encrypted, config_json, refresh_interval, version FROM providers WHERE id = ?"#,
135            )
136            .bind(id)
137            .fetch_optional(&self.pool)
138            .await
139            .map_err(|e| DomainError::DatabaseError(format!("Failed to get provider: {}", e)))?;
140
141            match row {
142                Some(row) => {
143                    let config_json: String = row.get("config_json");
144                    let config: HashMap<String, String> = if config_json.trim().is_empty() {
145                        HashMap::new()
146                    } else {
147                        serde_json::from_str(&config_json).map_err(|e| {
148                            DomainError::DatabaseError(format!("Failed to parse config: {}", e))
149                        })?
150                    };
151
152                    let token_ref: String = row.get("token_encrypted");
153
154                    Ok(Some(ProviderConfig {
155                        id: Some(id),
156                        name: row.get("name"),
157                        provider_type: row.get("provider_type"),
158                        config,
159                        token: token_ref,
160                        refresh_interval: row.get("refresh_interval"),
161                        version: Some(row.get("version")),
162                    }))
163                }
164                None => Ok(None),
165            }
166        })
167        .await
168    }
169
170    async fn create_provider(&self, config: &ProviderConfig) -> DomainResult<i64> {
171        retry_on_busy(|| async {
172            let existing = sqlx::query_scalar::<_, i64>("SELECT id FROM providers WHERE name = ?")
173                .bind(&config.name)
174                .fetch_optional(&self.pool)
175                .await
176                .map_err(|e| {
177                    DomainError::DatabaseError(format!("Failed to check existing: {}", e))
178                })?;
179
180            if existing.is_some() {
181                return Err(DomainError::InvalidConfig(format!(
182                    "Provider '{}' already exists",
183                    config.name
184                )));
185            }
186
187            let config_json = serde_json::to_string(&config.config)
188                .map_err(|e| DomainError::DatabaseError(format!("Failed to serialize config: {}", e)))?;
189
190            let result = sqlx::query(
191                r#"INSERT INTO providers (name, provider_type, token_encrypted, config_json, refresh_interval)
192                   VALUES (?, ?, ?, ?, ?)"#,
193            )
194            .bind(&config.name)
195            .bind(&config.provider_type)
196            .bind(&config.token)
197            .bind(&config_json)
198            .bind(config.refresh_interval)
199            .execute(&self.pool)
200            .await
201            .map_err(|e| DomainError::DatabaseError(format!("Failed to insert provider: {}", e)))?;
202
203            Ok(result.last_insert_rowid())
204        })
205        .await
206    }
207
208    async fn update_provider(&self, id: i64, config: &ProviderConfig) -> DomainResult<()> {
209        retry_on_busy(|| async {
210            let config_json = serde_json::to_string(&config.config)
211                .map_err(|e| DomainError::DatabaseError(format!("Failed to serialize config: {}", e)))?;
212
213            let result = sqlx::query(
214                r#"UPDATE providers
215                   SET name = ?, provider_type = ?, token_encrypted = ?, config_json = ?, refresh_interval = ?, updated_at = datetime('now')
216                   WHERE id = ?"#,
217            )
218            .bind(&config.name)
219            .bind(&config.provider_type)
220            .bind(&config.token)
221            .bind(&config_json)
222            .bind(config.refresh_interval)
223            .bind(id)
224            .execute(&self.pool)
225            .await
226            .map_err(|e| DomainError::DatabaseError(format!("Failed to update provider: {}", e)))?;
227
228            if result.rows_affected() == 0 {
229                return Err(DomainError::ProviderNotFound(format!("Provider {} not found", id)));
230            }
231
232            Ok(())
233        })
234        .await
235    }
236
237    async fn delete_provider(&self, id: i64) -> DomainResult<()> {
238        retry_on_busy(|| async {
239            let _ = sqlx::query("DELETE FROM pipelines_cache WHERE provider_id = ?")
240                .bind(id)
241                .execute(&self.pool)
242                .await;
243
244            let _ = sqlx::query("DELETE FROM provider_permissions WHERE provider_id = ?")
245                .bind(id)
246                .execute(&self.pool)
247                .await;
248
249            let _ = sqlx::query("DELETE FROM table_preferences WHERE provider_id = ?")
250                .bind(id)
251                .execute(&self.pool)
252                .await;
253
254            let result = sqlx::query("DELETE FROM providers WHERE id = ?")
255                .bind(id)
256                .execute(&self.pool)
257                .await
258                .map_err(|e| {
259                    DomainError::DatabaseError(format!("Failed to delete provider: {}", e))
260                })?;
261
262            if result.rows_affected() == 0 {
263                return Err(DomainError::ProviderNotFound(format!(
264                    "Provider {} not found",
265                    id
266                )));
267            }
268
269            Ok(())
270        })
271        .await
272    }
273
274    async fn get_table_preferences(
275        &self, provider_id: i64, table_id: &str,
276    ) -> DomainResult<Option<String>> {
277        retry_on_busy(|| async {
278            let result = sqlx::query_scalar::<_, String>(
279                "SELECT preferences_json FROM table_preferences WHERE provider_id = ? AND table_id = ?",
280            )
281            .bind(provider_id)
282            .bind(table_id)
283            .fetch_optional(&self.pool)
284            .await
285            .map_err(|e| DomainError::DatabaseError(format!("Failed to get preferences: {}", e)))?;
286
287            Ok(result)
288        })
289        .await
290    }
291
292    async fn set_table_preferences(
293        &self, provider_id: i64, table_id: &str, preferences_json: &str,
294    ) -> DomainResult<()> {
295        retry_on_busy(|| async {
296            sqlx::query(
297                r#"INSERT INTO table_preferences (provider_id, table_id, preferences_json, created_at, updated_at)
298                   VALUES (?, ?, ?, datetime('now'), datetime('now'))
299                   ON CONFLICT(provider_id, table_id) DO UPDATE SET
300                       preferences_json = excluded.preferences_json,
301                       updated_at = datetime('now')"#,
302            )
303            .bind(provider_id)
304            .bind(table_id)
305            .bind(preferences_json)
306            .execute(&self.pool)
307            .await
308            .map_err(|e| DomainError::DatabaseError(format!("Failed to save preferences: {}", e)))?;
309
310            Ok(())
311        })
312        .await
313    }
314
315    async fn store_permissions(
316        &self, provider_id: i64, permissions: &StoredPermissions,
317    ) -> DomainResult<()> {
318        let start = std::time::Instant::now();
319        let permissions_count = permissions.permissions.len();
320
321        retry_on_busy(|| async {
322            let checked_at_str = permissions.last_checked.to_rfc3339();
323
324            sqlx::query("DELETE FROM provider_permissions WHERE provider_id = ?")
325                .bind(provider_id)
326                .execute(&self.pool)
327                .await
328                .map_err(|e| {
329                    DomainError::DatabaseError(format!("Failed to clear permissions: {}", e))
330                })?;
331
332            if permissions.permissions.is_empty() {
333                return Ok(());
334            }
335
336            const BATCH_SIZE: usize = 100;
337
338            let perms_vec: Vec<_> = permissions.permissions.iter().collect();
339
340            for chunk in perms_vec.chunks(BATCH_SIZE) {
341                let values_clause = chunk
342                    .iter()
343                    .map(|_| "(?, ?, ?, ?)")
344                    .collect::<Vec<_>>()
345                    .join(", ");
346
347                let sql = format!(
348                    r#"INSERT INTO provider_permissions (provider_id, permission_name, granted, checked_at)
349                       VALUES {}"#,
350                    values_clause
351                );
352
353                let mut query = sqlx::query(&sql);
354                for (permission_name, granted) in chunk {
355                    query = query
356                        .bind(provider_id)
357                        .bind(permission_name)
358                        .bind(**granted)
359                        .bind(&checked_at_str);
360                }
361
362                query
363                    .execute(&self.pool)
364                    .await
365                    .map_err(|e| {
366                        DomainError::DatabaseError(format!("Failed to store permission: {}", e))
367                    })?;
368            }
369
370            Ok(())
371        })
372        .await?;
373
374        let elapsed = start.elapsed();
375        tracing::debug!(
376            provider_id = provider_id,
377            permissions_count = permissions_count,
378            elapsed_ms = elapsed.as_millis(),
379            "Stored provider permissions (batch insert)"
380        );
381
382        Ok(())
383    }
384
385    async fn get_permissions(&self, provider_id: i64) -> DomainResult<Option<StoredPermissions>> {
386        retry_on_busy(|| async {
387            let rows = sqlx::query(
388                r#"SELECT permission_name, granted, checked_at
389                   FROM provider_permissions
390                   WHERE provider_id = ?"#,
391            )
392            .bind(provider_id)
393            .fetch_all(&self.pool)
394            .await
395            .map_err(|e| DomainError::DatabaseError(format!("Failed to get permissions: {}", e)))?;
396
397            if rows.is_empty() {
398                return Ok(None);
399            }
400
401            let checked_at_str: String = rows[0].get("checked_at");
402            let last_checked = chrono::DateTime::parse_from_rfc3339(&checked_at_str)
403                .map_err(|e| {
404                    DomainError::DatabaseError(format!("Failed to parse timestamp: {}", e))
405                })?
406                .with_timezone(&chrono::Utc);
407
408            let mut permissions = HashMap::new();
409            for row in rows {
410                let permission_name: String = row.get("permission_name");
411                let granted: bool = row.get("granted");
412                permissions.insert(permission_name, granted);
413            }
414
415            Ok(Some(StoredPermissions {
416                permissions,
417                last_checked,
418            }))
419        })
420        .await
421    }
422
423    async fn export_all(&self) -> DomainResult<ConfigExport> {
424        let providers = self.list_providers().await?;
425
426        let pref_rows =
427            sqlx::query(r#"SELECT provider_id, table_id, preferences_json FROM table_preferences"#)
428                .fetch_all(&self.pool)
429                .await
430                .map_err(|e| {
431                    DomainError::DatabaseError(format!("Failed to export preferences: {}", e))
432                })?;
433
434        let mut table_preferences = HashMap::new();
435        for row in pref_rows {
436            let provider_id: i64 = row.get("provider_id");
437            let table_id: String = row.get("table_id");
438            let preferences_json: String = row.get("preferences_json");
439            let key = format!("{}:{}", provider_id, table_id);
440            table_preferences.insert(key, preferences_json);
441        }
442
443        let perm_rows = sqlx::query(
444            r#"SELECT provider_id, permission_name, granted, checked_at FROM provider_permissions"#,
445        )
446        .fetch_all(&self.pool)
447        .await
448        .map_err(|e| DomainError::DatabaseError(format!("Failed to export permissions: {}", e)))?;
449
450        let mut permissions: HashMap<i64, StoredPermissions> = HashMap::new();
451        for row in perm_rows {
452            let provider_id: i64 = row.get("provider_id");
453            let permission_name: String = row.get("permission_name");
454            let granted: bool = row.get("granted");
455            let checked_at_str: String = row.get("checked_at");
456
457            let last_checked = chrono::DateTime::parse_from_rfc3339(&checked_at_str)
458                .map(|dt| dt.with_timezone(&chrono::Utc))
459                .unwrap_or_else(|_| chrono::Utc::now());
460
461            let entry = permissions
462                .entry(provider_id)
463                .or_insert_with(|| StoredPermissions {
464                    permissions: HashMap::new(),
465                    last_checked,
466                });
467            entry.permissions.insert(permission_name, granted);
468        }
469
470        Ok(ConfigExport {
471            version: "1.0".to_string(),
472            providers,
473            table_preferences,
474            permissions,
475        })
476    }
477
478    async fn import_all(&self, data: &ConfigExport) -> DomainResult<HashMap<i64, i64>> {
479        let mut id_mapping: HashMap<i64, i64> = HashMap::new();
480
481        for provider in &data.providers {
482            let old_id = provider.id.unwrap_or(0);
483
484            let existing = sqlx::query_scalar::<_, i64>("SELECT id FROM providers WHERE name = ?")
485                .bind(&provider.name)
486                .fetch_optional(&self.pool)
487                .await
488                .map_err(|e| {
489                    DomainError::DatabaseError(format!("Failed to check existing: {}", e))
490                })?;
491
492            let new_id = if let Some(existing_id) = existing {
493                existing_id
494            } else {
495                self.create_provider(provider).await?
496            };
497
498            if old_id != 0 {
499                id_mapping.insert(old_id, new_id);
500            }
501        }
502
503        for (key, preferences_json) in &data.table_preferences {
504            let parts: Vec<&str> = key.splitn(2, ':').collect();
505            if parts.len() == 2 {
506                if let Ok(old_provider_id) = parts[0].parse::<i64>() {
507                    let table_id = parts[1];
508                    let new_provider_id = id_mapping
509                        .get(&old_provider_id)
510                        .copied()
511                        .unwrap_or(old_provider_id);
512                    if self.get_provider(new_provider_id).await?.is_some() {
513                        self.set_table_preferences(new_provider_id, table_id, preferences_json)
514                            .await?;
515                    }
516                }
517            }
518        }
519
520        for (old_provider_id, perms) in &data.permissions {
521            let new_provider_id = id_mapping
522                .get(old_provider_id)
523                .copied()
524                .unwrap_or(*old_provider_id);
525            if self.get_provider(new_provider_id).await?.is_some() {
526                self.store_permissions(new_provider_id, perms).await?;
527            }
528        }
529
530        Ok(id_mapping)
531    }
532}
533
534#[cfg(test)]
535mod tests {
536    use super::*;
537    use crate::infrastructure::database::init_database;
538
539    #[tokio::test]
540    async fn test_sqlite_backend_providers() {
541        let dir = tempfile::tempdir().unwrap();
542        let db_path = dir.path().join("test.db");
543        let pool = init_database(db_path).await.unwrap();
544        let backend = SqliteConfigBackend::new(pool);
545
546        let config = ProviderConfig {
547            id: None,
548            name: "Test Provider".to_string(),
549            provider_type: "github".to_string(),
550            config: HashMap::from([("owner".to_string(), "test".to_string())]),
551            token: String::new(),
552            refresh_interval: 30,
553            version: None,
554        };
555
556        let id = backend.create_provider(&config).await.unwrap();
557        assert!(id > 0);
558
559        let retrieved = backend.get_provider(id).await.unwrap().unwrap();
560        assert_eq!(retrieved.name, "Test Provider");
561        assert_eq!(retrieved.provider_type, "github");
562
563        let all = backend.list_providers().await.unwrap();
564        assert_eq!(all.len(), 1);
565
566        let updated_config = ProviderConfig {
567            id: Some(id),
568            name: "Updated Provider".to_string(),
569            provider_type: "github".to_string(),
570            config: HashMap::from([("owner".to_string(), "updated".to_string())]),
571            token: String::new(),
572            refresh_interval: 60,
573            version: None,
574        };
575        backend.update_provider(id, &updated_config).await.unwrap();
576
577        let retrieved = backend.get_provider(id).await.unwrap().unwrap();
578        assert_eq!(retrieved.name, "Updated Provider");
579        assert_eq!(retrieved.refresh_interval, 60);
580
581        backend.delete_provider(id).await.unwrap();
582        assert!(backend.get_provider(id).await.unwrap().is_none());
583    }
584
585    #[tokio::test]
586    async fn test_sqlite_backend_table_preferences() {
587        let dir = tempfile::tempdir().unwrap();
588        let db_path = dir.path().join("test.db");
589        let pool = init_database(db_path).await.unwrap();
590        let backend = SqliteConfigBackend::new(pool);
591
592        let config = ProviderConfig {
593            id: None,
594            name: "Prefs Provider".to_string(),
595            provider_type: "github".to_string(),
596            config: HashMap::new(),
597            token: String::new(),
598            refresh_interval: 30,
599            version: None,
600        };
601        let provider_id = backend.create_provider(&config).await.unwrap();
602
603        let prefs_json = r#"{"columns":["name","status"],"sort":"name"}"#;
604        backend
605            .set_table_preferences(provider_id, "pipelines", prefs_json)
606            .await
607            .unwrap();
608
609        let retrieved = backend
610            .get_table_preferences(provider_id, "pipelines")
611            .await
612            .unwrap();
613        assert_eq!(retrieved, Some(prefs_json.to_string()));
614
615        let missing = backend
616            .get_table_preferences(provider_id, "other")
617            .await
618            .unwrap();
619        assert!(missing.is_none());
620    }
621
622    #[tokio::test]
623    async fn test_sqlite_backend_permissions() {
624        let dir = tempfile::tempdir().unwrap();
625        let db_path = dir.path().join("test.db");
626        let pool = init_database(db_path).await.unwrap();
627        let backend = SqliteConfigBackend::new(pool);
628
629        let config = ProviderConfig {
630            id: None,
631            name: "Perms Provider".to_string(),
632            provider_type: "github".to_string(),
633            config: HashMap::new(),
634            token: String::new(),
635            refresh_interval: 30,
636            version: None,
637        };
638        let provider_id = backend.create_provider(&config).await.unwrap();
639
640        let mut perms_map = HashMap::new();
641        perms_map.insert("repo:read".to_string(), true);
642        perms_map.insert("repo:write".to_string(), false);
643        let permissions = StoredPermissions {
644            permissions: perms_map,
645            last_checked: chrono::Utc::now(),
646        };
647        backend
648            .store_permissions(provider_id, &permissions)
649            .await
650            .unwrap();
651
652        let retrieved = backend.get_permissions(provider_id).await.unwrap().unwrap();
653        assert_eq!(retrieved.permissions.get("repo:read"), Some(&true));
654        assert_eq!(retrieved.permissions.get("repo:write"), Some(&false));
655    }
656
657    #[tokio::test]
658    async fn test_sqlite_backend_export_import() {
659        let dir = tempfile::tempdir().unwrap();
660        let db_path = dir.path().join("test.db");
661        let pool = init_database(db_path).await.unwrap();
662        let backend = SqliteConfigBackend::new(pool);
663
664        let config = ProviderConfig {
665            id: None,
666            name: "Export Test".to_string(),
667            provider_type: "gitlab".to_string(),
668            config: HashMap::from([("host".to_string(), "gitlab.com".to_string())]),
669            token: String::new(),
670            refresh_interval: 45,
671            version: None,
672        };
673        backend.create_provider(&config).await.unwrap();
674
675        let export = backend.export_all().await.unwrap();
676        assert_eq!(export.providers.len(), 1);
677        assert_eq!(export.providers[0].name, "Export Test");
678
679        let dir2 = tempfile::tempdir().unwrap();
680        let db_path2 = dir2.path().join("test2.db");
681        let pool2 = init_database(db_path2).await.unwrap();
682        let backend2 = SqliteConfigBackend::new(pool2);
683
684        backend2.import_all(&export).await.unwrap();
685
686        let imported = backend2.list_providers().await.unwrap();
687        assert_eq!(imported.len(), 1);
688        assert_eq!(imported[0].name, "Export Test");
689    }
690}
pipedash_core/infrastructure/database/sqlite_backend.rs

pipedash_core/infrastructure/database/
sqlite_backend.rs
