crab_vault_engine/
fs.rs

1use serde::de::DeserializeOwned;
2use std::path::{Path, PathBuf};
3use tokio::{
4    fs::{self, File},
5    io::{AsyncReadExt, AsyncWriteExt},
6};
7
8use crate::{
9    error::{EngineError, EngineResult},
10    {BucketMeta, DataEngine, MetaEngine, ObjectMeta},
11};
12
13pub struct FsDataEngine {
14    base_dir: PathBuf,
15}
16
17impl FsDataEngine {
18    fn path_of_object(&self, bucket_name: &str, object_name: &str) -> PathBuf {
19        self.base_dir.join(bucket_name).join(object_name)
20    }
21
22    fn path_of_bucket(&self, bucket_name: &str) -> PathBuf {
23        self.base_dir.join(bucket_name)
24    }
25}
26
27/// helper function，将 [IO Error](std::io::Error) 转换为 [`StorageError`]
28#[inline(always)]
29fn io_error<P: AsRef<Path> + ?Sized>(e: std::io::Error, path: &P) -> EngineError {
30    EngineError::Io {
31        error: e,
32        path: path.as_ref().to_string_lossy().to_string(),
33    }
34}
35
36impl DataEngine for FsDataEngine {
37    type Uri = Path;
38
39    fn new<P: AsRef<Path>>(base_dir: P) -> EngineResult<Self> {
40        let base_dir = base_dir.as_ref().to_path_buf();
41        std::fs::create_dir_all(&base_dir).map_err(|e| io_error(e, &base_dir))?;
42        Ok(Self { base_dir })
43    }
44
45    async fn create_bucket(&self, bucket_name: &str) -> EngineResult<()> {
46        let path = self.path_of_bucket(bucket_name);
47
48        fs::create_dir_all(&path)
49            .await
50            .map_err(|e| io_error(e, &path))?;
51
52        Ok(())
53    }
54
55    async fn delete_bucket(&self, bucket_name: &str) -> EngineResult<()> {
56        let path = self.path_of_bucket(bucket_name);
57
58        // 直接尝试删除目录
59        if let Err(e) = fs::remove_dir(&path).await {
60            if e.kind() == std::io::ErrorKind::DirectoryNotEmpty && path.is_dir() {
61                return Err(EngineError::BucketNotEmpty {
62                    bucket: bucket_name.to_string(),
63                });
64            } else if e.kind() == std::io::ErrorKind::NotFound {
65                return Ok(())
66            }
67            // 对于其他类型的 IO 错误，正常地返回
68            return Err(io_error(e, &path));
69        }
70
71        Ok(())
72    }
73
74    async fn create_object(
75        &self,
76        bucket_name: &str,
77        object_name: &str,
78        data: &[u8],
79    ) -> EngineResult<()> {
80        let path = self.path_of_object(bucket_name, object_name);
81
82        if let Some(parent) = path.parent()
83            && !parent.exists()
84        {
85            return Err(EngineError::BucketNotFound {
86                bucket: bucket_name.to_string(),
87            });
88        }
89
90        // 异步写入文件
91        let mut file = File::create(&path).await.map_err(|e| io_error(e, &path))?;
92        file.write_all(data).await.map_err(|e| io_error(e, &path))?;
93        file.flush().await.map_err(|e| io_error(e, &path))?;
94
95        Ok(())
96    }
97
98    async fn read_object(&self, bucket_name: &str, object_name: &str) -> EngineResult<Vec<u8>> {
99        let path = self.path_of_object(bucket_name, object_name);
100        let map_io_err = |e| io_error(e, &path);
101
102        // 直接尝试打开文件，并处理 NotFound 错误
103        let mut file = match File::open(&path).await {
104            Ok(file) => file,
105            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
106                return Err(EngineError::ObjectNotFound {
107                    bucket: bucket_name.to_string(),
108                    object: object_name.to_string(),
109                });
110            }
111            Err(e) => return Err(map_io_err(e)),
112        };
113
114        let mut contents = Vec::new();
115        file.read_to_end(&mut contents).await.map_err(map_io_err)?;
116
117        Ok(contents)
118    }
119
120    async fn delete_object(&self, bucket_name: &str, object_name: &str) -> EngineResult<()> {
121        let path = self.path_of_object(bucket_name, object_name);
122
123        match fs::remove_file(&path).await {
124            Ok(_) => Ok(()),
125            // 如果文件不存在，我们认为删除操作是成功的（幂等性）
126            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
127            Err(e) => Err(io_error(e, &path)),
128        }
129    }
130}
131
132pub struct FsMetaEngine {
133    base_dir: PathBuf,
134}
135
136impl FsMetaEngine {
137    // 优化的路径结构
138    fn bucket_meta_path(&self, bucket_name: &str) -> PathBuf {
139        self.base_dir
140            .join("buckets")
141            .join(format!("{}.json", bucket_name))
142    }
143
144    fn object_meta_path(&self, bucket_name: &str, object_name: &str) -> PathBuf {
145        self.base_dir
146            .join("objects")
147            .join(bucket_name)
148            .join(format!("{}.json", object_name))
149    }
150
151    // 获取对象元数据目录的路径
152    fn objects_dir_path(&self, bucket_name: &str) -> PathBuf {
153        self.base_dir.join("objects").join(bucket_name)
154    }
155
156    // 获取 bucket 元数据目录的路径
157    fn buckets_dir_path(&self) -> PathBuf {
158        self.base_dir.join("buckets")
159    }
160}
161
162/// 辅助函数，用于从目录中列出并反序列化所有JSON元数据文件。
163async fn list_meta_from_dir<T: DeserializeOwned>(dir_path: &Path) -> EngineResult<Vec<T>> {
164    // 如果目录不存在，这是一个正常情况，只返回一个空列表。
165    if !dir_path.exists() {
166        return Ok(Vec::new());
167    }
168
169    let mut entries = fs::read_dir(dir_path)
170        .await
171        .map_err(|e| io_error(e, dir_path))?;
172
173    let mut results = Vec::new();
174
175    while let Some(entry) = entries
176        .next_entry()
177        .await
178        .map_err(|e| io_error(e, dir_path))?
179    {
180        let path = entry.path();
181        if path.is_file() && path.extension().and_then(|s| s.to_str()) == Some("json") {
182            let data = fs::read_to_string(&path)
183                .await
184                .map_err(|e| io_error(e, &path))?;
185            // 如果单个文件损坏，我们可以选择跳过它或返回错误。这里我们选择失败。
186            let meta: T = serde_json::from_str(&data)?;
187            results.push(meta);
188        }
189    }
190
191    Ok(results)
192}
193
194impl MetaEngine for FsMetaEngine {
195    type Uri = Path;
196
197    fn new<P: AsRef<Path>>(base_dir: P) -> EngineResult<Self> {
198        let base_dir = base_dir.as_ref().to_path_buf();
199        // 在初始化时创建元数据根目录
200        std::fs::create_dir_all(&base_dir).map_err(|e| io_error(e, &base_dir))?;
201        Ok(Self { base_dir })
202    }
203
204    async fn create_object_meta(&self, meta: &ObjectMeta) -> EngineResult<()> {
205        let path = self.object_meta_path(&meta.bucket_name, &meta.object_name);
206
207        if let Some(parent) = path.parent() {
208            fs::create_dir_all(parent)
209                .await
210                .map_err(|e| io_error(e, parent))?;
211        }
212
213        let json = serde_json::to_string_pretty(meta)?;
214        fs::write(&path, json).await.map_err(|e| io_error(e, &path))
215    }
216
217    async fn read_object_meta(
218        &self,
219        bucket_name: &str,
220        object_name: &str,
221    ) -> EngineResult<ObjectMeta> {
222        let path = self.object_meta_path(bucket_name, object_name);
223
224        match fs::read_to_string(&path).await {
225            Ok(data) => Ok(serde_json::from_str(&data)?),
226            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
227                Err(EngineError::ObjectMetaNotFound {
228                    bucket: bucket_name.to_string(),
229                    object: object_name.to_string(),
230                })
231            }
232            Err(e) => Err(io_error(e, &path)),
233        }
234    }
235
236    async fn delete_object_meta(&self, bucket_name: &str, object_name: &str) -> EngineResult<()> {
237        let path = self.object_meta_path(bucket_name, object_name);
238
239        match fs::remove_file(&path).await {
240            Ok(_) => Ok(()),
241            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
242            Err(e) => Err(io_error(e, &path)),
243        }
244    }
245
246    async fn list_objects_meta(&self, bucket_name: &str) -> EngineResult<Vec<ObjectMeta>> {
247        let dir_path = self.objects_dir_path(bucket_name);
248        list_meta_from_dir(&dir_path).await
249    }
250
251    async fn touch_object(&self, bucket_name: &str, object_name: &str) -> EngineResult<()> {
252        let path = self.object_meta_path(bucket_name, object_name);
253
254        match fs::read_to_string(&path).await {
255            Ok(data) => {
256                let mut meta: ObjectMeta = serde_json::from_str(&data)?;
257                meta.updated_at = chrono::Utc::now();
258                fs::write(&path, serde_json::to_string_pretty(&meta)?)
259                    .await
260                    .map_err(|e| io_error(e, &path))
261            }
262            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
263                Err(EngineError::ObjectMetaNotFound {
264                    bucket: bucket_name.to_string(),
265                    object: object_name.to_string(),
266                })
267            }
268            Err(e) => Err(io_error(e, &path)),
269        }
270    }
271
272    async fn create_bucket_meta(&self, meta: &BucketMeta) -> EngineResult<()> {
273        let path = self.bucket_meta_path(&meta.name);
274
275        if let Some(parent) = path.parent() {
276            fs::create_dir_all(parent)
277                .await
278                .map_err(|e| io_error(e, parent))?;
279        }
280
281        let json = serde_json::to_string_pretty(meta)?;
282        fs::write(&path, json).await.map_err(|e| io_error(e, &path))
283    }
284
285    async fn read_bucket_meta(&self, name: &str) -> EngineResult<BucketMeta> {
286        let path = self.bucket_meta_path(name);
287
288        match fs::read_to_string(&path).await {
289            Ok(data) => Ok(serde_json::from_str(&data)?),
290            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
291                Err(EngineError::BucketMetaNotFound {
292                    bucket: name.to_string(),
293                })
294            }
295            Err(e) => Err(io_error(e, &path)),
296        }
297    }
298
299    async fn delete_bucket_meta(&self, name: &str) -> EngineResult<()> {
300        let path = self.bucket_meta_path(name);
301
302        match fs::remove_file(&path).await {
303            Ok(_) => Ok(()),
304            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
305            Err(e) => Err(io_error(e, &path)),
306        }?;
307
308        match fs::remove_dir(self.objects_dir_path(name)).await {
309            Ok(_) => Ok(()),
310            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
311            Err(e) => Err(io_error(e, &path)),
312        }?;
313
314        Ok(())
315    }
316
317    async fn touch_bucket(&self, bucket_name: &str) -> EngineResult<()> {
318        let path = self.bucket_meta_path(bucket_name);
319
320        match fs::read_to_string(&path).await {
321            Ok(data) => {
322                let mut meta: BucketMeta = serde_json::from_str(&data)?;
323                meta.updated_at = chrono::Utc::now();
324                fs::write(&path, serde_json::to_string_pretty(&meta)?)
325                    .await
326                    .map_err(|e| io_error(e, &path))
327            }
328            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
329                Err(EngineError::BucketMetaNotFound {
330                    bucket: bucket_name.to_string(),
331                })
332            }
333            Err(e) => Err(io_error(e, &path)),
334        }
335    }
336
337    async fn list_buckets_meta(&self) -> EngineResult<Vec<BucketMeta>> {
338        let dir_path = self.buckets_dir_path();
339        list_meta_from_dir(&dir_path).await
340    }
341}
crab_vault_engine/fs.rs

crab_vault_engine/
fs.rs
