pipedash_core/infrastructure/storage/
mod.rs

1#[cfg(feature = "postgres")]
2pub mod postgres;
3
4use std::collections::HashMap;
5use std::path::PathBuf;
6
7use async_trait::async_trait;
8use chrono::{
9    DateTime,
10    Utc,
11};
12#[cfg(feature = "postgres")]
13pub use postgres::{
14    PostgresCacheConfig,
15    PostgresStorage,
16};
17use serde::{
18    Deserialize,
19    Serialize,
20};
21use tokio::fs;
22
23use crate::domain::{
24    DomainError,
25    DomainResult,
26};
27
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct ObjectMetadata {
30    pub key: String,
31    pub size: u64,
32    pub last_modified: DateTime<Utc>,
33    pub etag: Option<String>,
34    pub content_type: Option<String>,
35    #[serde(default)]
36    pub metadata: HashMap<String, String>,
37}
38
39#[async_trait]
40pub trait StorageBackend: Send + Sync {
41    fn backend_type(&self) -> &str;
42
43    async fn is_available(&self) -> bool;
44
45    async fn list(&self, prefix: Option<&str>) -> DomainResult<Vec<ObjectMetadata>>;
46
47    async fn get(&self, key: &str) -> DomainResult<Vec<u8>>;
48
49    async fn put(
50        &self, key: &str, data: &[u8], content_type: Option<&str>,
51    ) -> DomainResult<ObjectMetadata>;
52
53    async fn delete(&self, key: &str) -> DomainResult<()>;
54
55    async fn exists(&self, key: &str) -> DomainResult<bool>;
56
57    async fn head(&self, key: &str) -> DomainResult<Option<ObjectMetadata>>;
58}
59
60pub struct LocalStorage {
61    base_path: PathBuf,
62}
63
64impl LocalStorage {
65    pub fn new(base_path: PathBuf) -> Self {
66        Self { base_path }
67    }
68
69    pub fn default_location() -> DomainResult<Self> {
70        let base_path = std::env::var("PIPEDASH_DATA_DIR")
71            .map(PathBuf::from)
72            .or_else(|_| {
73                dirs::data_dir().map(|p| p.join("pipedash")).ok_or_else(|| {
74                    DomainError::InternalError("Could not determine data directory".to_string())
75                })
76            })?;
77
78        Ok(Self::new(base_path))
79    }
80
81    fn full_path(&self, key: &str) -> PathBuf {
82        self.base_path.join(key)
83    }
84
85    pub fn database_path(&self) -> PathBuf {
86        self.base_path.join("pipedash.db")
87    }
88
89    pub fn metrics_database_path(&self) -> PathBuf {
90        self.base_path.join("metrics.db")
91    }
92
93    pub fn config_dir(&self) -> PathBuf {
94        self.base_path.join("config")
95    }
96
97    pub fn cache_dir(&self) -> PathBuf {
98        self.base_path.join("cache")
99    }
100
101    pub fn base_path(&self) -> &PathBuf {
102        &self.base_path
103    }
104
105    pub async fn ensure_directories(&self) -> DomainResult<()> {
106        fs::create_dir_all(&self.base_path)
107            .await
108            .map_err(|e| DomainError::DatabaseError(format!("Failed to create data dir: {}", e)))?;
109
110        fs::create_dir_all(self.config_dir()).await.map_err(|e| {
111            DomainError::DatabaseError(format!("Failed to create config dir: {}", e))
112        })?;
113
114        fs::create_dir_all(self.cache_dir()).await.map_err(|e| {
115            DomainError::DatabaseError(format!("Failed to create cache dir: {}", e))
116        })?;
117
118        Ok(())
119    }
120}
121
122#[async_trait]
123impl StorageBackend for LocalStorage {
124    fn backend_type(&self) -> &str {
125        "local"
126    }
127
128    async fn is_available(&self) -> bool {
129        fs::metadata(&self.base_path).await.is_ok()
130    }
131
132    async fn list(&self, prefix: Option<&str>) -> DomainResult<Vec<ObjectMetadata>> {
133        let search_path = match prefix {
134            Some(p) => self.full_path(p),
135            None => self.base_path.clone(),
136        };
137
138        if !search_path.exists() {
139            return Ok(Vec::new());
140        }
141
142        let mut entries = fs::read_dir(&search_path)
143            .await
144            .map_err(|e| DomainError::DatabaseError(format!("Failed to read directory: {}", e)))?;
145
146        let mut results = Vec::new();
147
148        while let Some(entry) = entries
149            .next_entry()
150            .await
151            .map_err(|e| DomainError::DatabaseError(format!("Failed to read entry: {}", e)))?
152        {
153            let metadata = entry.metadata().await.map_err(|e| {
154                DomainError::DatabaseError(format!("Failed to read metadata: {}", e))
155            })?;
156
157            if metadata.is_file() {
158                let key = entry
159                    .path()
160                    .strip_prefix(&self.base_path)
161                    .unwrap_or(entry.path().as_path())
162                    .to_string_lossy()
163                    .to_string();
164
165                let last_modified = metadata
166                    .modified()
167                    .map(DateTime::<Utc>::from)
168                    .unwrap_or_else(|_| Utc::now());
169
170                results.push(ObjectMetadata {
171                    key,
172                    size: metadata.len(),
173                    last_modified,
174                    etag: None,
175                    content_type: None,
176                    metadata: HashMap::new(),
177                });
178            }
179        }
180
181        Ok(results)
182    }
183
184    async fn get(&self, key: &str) -> DomainResult<Vec<u8>> {
185        fs::read(self.full_path(key))
186            .await
187            .map_err(|e| DomainError::DatabaseError(format!("Failed to read file: {}", e)))
188    }
189
190    async fn put(
191        &self, key: &str, data: &[u8], _content_type: Option<&str>,
192    ) -> DomainResult<ObjectMetadata> {
193        let path = self.full_path(key);
194
195        if let Some(parent) = path.parent() {
196            fs::create_dir_all(parent)
197                .await
198                .map_err(|e| DomainError::DatabaseError(format!("Failed to create dir: {}", e)))?;
199        }
200
201        fs::write(&path, data)
202            .await
203            .map_err(|e| DomainError::DatabaseError(format!("Failed to write file: {}", e)))?;
204
205        let metadata = fs::metadata(&path)
206            .await
207            .map_err(|e| DomainError::DatabaseError(format!("Failed to read metadata: {}", e)))?;
208
209        let last_modified = metadata
210            .modified()
211            .map(DateTime::<Utc>::from)
212            .unwrap_or_else(|_| Utc::now());
213
214        Ok(ObjectMetadata {
215            key: key.to_string(),
216            size: metadata.len(),
217            last_modified,
218            etag: None,
219            content_type: None,
220            metadata: HashMap::new(),
221        })
222    }
223
224    async fn delete(&self, key: &str) -> DomainResult<()> {
225        let path = self.full_path(key);
226        if path.exists() {
227            fs::remove_file(&path)
228                .await
229                .map_err(|e| DomainError::DatabaseError(format!("Failed to delete file: {}", e)))?;
230        }
231        Ok(())
232    }
233
234    async fn exists(&self, key: &str) -> DomainResult<bool> {
235        Ok(self.full_path(key).exists())
236    }
237
238    async fn head(&self, key: &str) -> DomainResult<Option<ObjectMetadata>> {
239        let path = self.full_path(key);
240
241        match fs::metadata(&path).await {
242            Ok(metadata) => {
243                let last_modified = metadata
244                    .modified()
245                    .map(DateTime::<Utc>::from)
246                    .unwrap_or_else(|_| Utc::now());
247
248                Ok(Some(ObjectMetadata {
249                    key: key.to_string(),
250                    size: metadata.len(),
251                    last_modified,
252                    etag: None,
253                    content_type: None,
254                    metadata: HashMap::new(),
255                }))
256            }
257            Err(_) => Ok(None),
258        }
259    }
260}
261
262#[cfg(test)]
263mod tests {
264    use tempfile::TempDir;
265
266    use super::*;
267
268    #[tokio::test]
269    async fn test_local_storage() {
270        let temp_dir = TempDir::new().unwrap();
271        let storage = LocalStorage::new(temp_dir.path().to_path_buf());
272
273        let data = b"Hello, World!";
274        let meta = storage.put("test.txt", data, None).await.unwrap();
275        assert_eq!(meta.key, "test.txt");
276        assert_eq!(meta.size, data.len() as u64);
277
278        assert!(storage.exists("test.txt").await.unwrap());
279        assert!(!storage.exists("nonexistent.txt").await.unwrap());
280
281        let retrieved = storage.get("test.txt").await.unwrap();
282        assert_eq!(retrieved, data);
283
284        let head = storage.head("test.txt").await.unwrap();
285        assert!(head.is_some());
286        assert_eq!(head.unwrap().size, data.len() as u64);
287
288        let list = storage.list(None).await.unwrap();
289        assert_eq!(list.len(), 1);
290
291        storage.delete("test.txt").await.unwrap();
292        assert!(!storage.exists("test.txt").await.unwrap());
293    }
294}
pipedash_core/infrastructure/storage/mod.rs

pipedash_core/infrastructure/storage/
mod.rs
