1use std::path::{Path, PathBuf};
4use std::time::Duration;
5
6use rskit_errors::{AppError, AppResult, ErrorCode};
7use rskit_fs::{async_io, sync_io};
8
9use crate::FileSource;
10
11use super::super::{FileStore, StoredFile, UploadOptions, prefixed_key};
12use super::config::LocalStoreConfig;
13use super::path::{
14 canonicalize_confined, ensure_target_parent_confined, file_not_found_error,
15 file_not_found_error_with_cause, normalize_local_key, replace_with_temp, storage_temp_path,
16};
17
18pub struct LocalStore {
20 config: LocalStoreConfig,
21}
22
23impl LocalStore {
24 pub fn new(config: LocalStoreConfig) -> AppResult<Self> {
26 if config.auto_create {
27 sync_io::dir::create_all(&config.root_dir)?;
28 }
29 Self::validate_root_dir(&config.root_dir)?;
30 Ok(Self { config })
31 }
32
33 fn resolve_path(&self, key: &str) -> AppResult<PathBuf> {
34 let key = normalize_local_key(key)?;
35 self.resolve_normalized_path(&key)
36 }
37
38 fn validate_root_dir(root: &Path) -> AppResult<()> {
39 let metadata = std::fs::symlink_metadata(root).map_err(|error| {
40 if error.kind() == std::io::ErrorKind::NotFound {
41 AppError::new(
42 ErrorCode::NotFound,
43 format!("store root {} does not exist", root.display()),
44 )
45 } else {
46 AppError::new(
47 ErrorCode::Internal,
48 format!("failed to inspect store root '{}': {error}", root.display()),
49 )
50 .with_cause(error)
51 }
52 })?;
53 if metadata.file_type().is_symlink() {
54 return Err(AppError::new(
55 ErrorCode::InvalidInput,
56 format!("store root {} must not be a symlink", root.display()),
57 ));
58 }
59 if !metadata.is_dir() {
60 return Err(AppError::new(
61 ErrorCode::InvalidInput,
62 format!("store root {} must be a directory", root.display()),
63 ));
64 }
65 Ok(())
66 }
67
68 fn inspect_existing_file_error(key: &str, path: &Path, error: std::io::Error) -> AppError {
69 if error.kind() == std::io::ErrorKind::NotFound {
70 file_not_found_error(key).with_cause(error)
71 } else {
72 AppError::new(
73 ErrorCode::Internal,
74 format!(
75 "failed to inspect storage file '{}': {error}",
76 path.display()
77 ),
78 )
79 .with_cause(error)
80 }
81 }
82
83 fn resolve_normalized_path(&self, key: &str) -> AppResult<PathBuf> {
84 rskit_fs::safe_join(&self.config.root_dir, Path::new(key))
85 .map_err(|error| AppError::new(ErrorCode::InvalidInput, error.to_string()))
86 }
87
88 async fn confined_existing_file_path(&self, key: &str) -> AppResult<PathBuf> {
89 let path = self.resolve_path(key)?;
90 let metadata = tokio::fs::symlink_metadata(&path)
91 .await
92 .map_err(|error| Self::inspect_existing_file_error(key, &path, error))?;
93 if !metadata.is_file() || metadata.file_type().is_symlink() {
94 return Err(file_not_found_error(key));
95 }
96 let canonical = canonicalize_confined(&self.config.root_dir, &path).await?;
97 Ok(canonical)
98 }
99
100 async fn confined_existing_dir_path(&self, key: &str) -> AppResult<Option<PathBuf>> {
101 let path = self.resolve_path(key)?;
102 let metadata = match tokio::fs::symlink_metadata(&path).await {
103 Ok(metadata) => metadata,
104 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
105 Err(error) => {
106 return Err(AppError::new(
107 ErrorCode::Internal,
108 format!("failed to inspect directory '{}': {error}", path.display()),
109 )
110 .with_cause(error));
111 }
112 };
113 if metadata.file_type().is_symlink() {
114 return Err(AppError::new(
115 ErrorCode::InvalidInput,
116 format!(
117 "storage directory '{}' must not be a symlink",
118 path.display()
119 ),
120 ));
121 }
122 if !metadata.is_dir() {
123 return Ok(None);
124 }
125 canonicalize_confined(&self.config.root_dir, &path)
126 .await
127 .map(Some)
128 }
129
130 async fn confined_presigned_path(&self, key: &str) -> AppResult<PathBuf> {
131 let path = self.resolve_path(key)?;
132 match tokio::fs::symlink_metadata(&path).await {
133 Ok(metadata) => {
134 if metadata.file_type().is_symlink() || !metadata.is_file() {
135 return Err(file_not_found_error(key));
136 }
137 canonicalize_confined(&self.config.root_dir, &path).await
138 }
139 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
140 let parent = path.parent().ok_or_else(|| {
141 AppError::new(
142 ErrorCode::InvalidInput,
143 format!(
144 "storage presigned URL target '{}' has no parent directory",
145 path.display()
146 ),
147 )
148 })?;
149 ensure_target_parent_confined(&self.config.root_dir, &path).await?;
150 let parent = canonicalize_confined(&self.config.root_dir, parent).await?;
151 let filename = path.file_name().ok_or_else(|| {
152 AppError::new(
153 ErrorCode::InvalidInput,
154 format!(
155 "storage presigned URL target '{}' has no filename",
156 path.display()
157 ),
158 )
159 })?;
160 Ok(parent.join(filename))
161 }
162 Err(error) => Err(AppError::new(
163 ErrorCode::Internal,
164 format!(
165 "failed to inspect presigned URL target '{}': {error}",
166 path.display()
167 ),
168 )
169 .with_cause(error)),
170 }
171 }
172
173 async fn stream_to_target(
174 &self,
175 reader: &mut (dyn tokio::io::AsyncRead + Unpin + Send),
176 target: &Path,
177 ) -> AppResult<u64> {
178 ensure_target_parent_confined(&self.config.root_dir, target).await?;
179 let temp_path = storage_temp_path(target);
180 let mut temp = tokio::fs::OpenOptions::new()
181 .write(true)
182 .create_new(true)
183 .open(&temp_path)
184 .await
185 .map_err(|error| {
186 AppError::new(
187 ErrorCode::Internal,
188 format!(
189 "failed to create temporary file '{}': {error}",
190 temp_path.display()
191 ),
192 )
193 .with_cause(error)
194 })?;
195 let result = async {
196 let size = tokio::io::copy(reader, &mut temp).await.map_err(|error| {
197 AppError::new(
198 ErrorCode::Internal,
199 format!("failed to stream {}: {error}", target.display()),
200 )
201 .with_cause(error)
202 })?;
203 temp.sync_data().await.map_err(|error| {
204 AppError::new(
205 ErrorCode::Internal,
206 format!(
207 "failed to sync temporary file '{}': {error}",
208 temp_path.display()
209 ),
210 )
211 .with_cause(error)
212 })?;
213 drop(temp);
214 replace_with_temp(&temp_path, target).await?;
215 Ok(size)
216 }
217 .await;
218 if result.is_err() {
219 let _ = tokio::fs::remove_file(&temp_path).await;
220 }
221 result
222 }
223}
224
225#[async_trait::async_trait]
226impl FileStore for LocalStore {
227 async fn upload(
228 &self,
229 source: &FileSource,
230 key: &str,
231 options: UploadOptions,
232 ) -> AppResult<StoredFile> {
233 let key = normalize_local_key(key)?;
234 let target = self.resolve_normalized_path(&key)?;
235
236 let mut reader = source.reader().await?;
237 let size = self.stream_to_target(&mut reader, &target).await?;
238
239 Ok(StoredFile::new(key, size, options.content_type()).with_metadata(options.metadata))
240 }
241
242 async fn download(&self, key: &str) -> AppResult<FileSource> {
243 let path = self.confined_existing_file_path(key).await?;
244 Ok(FileSource::Path(path))
245 }
246
247 async fn delete(&self, key: &str) -> AppResult<()> {
248 let path = self.confined_existing_file_path(key).await?;
249 if async_io::file::remove_if_exists(&path).await? {
250 Ok(())
251 } else {
252 Err(file_not_found_error(key))
253 }
254 }
255
256 async fn exists(&self, key: &str) -> AppResult<bool> {
257 match self.confined_existing_file_path(key).await {
258 Ok(_) => Ok(true),
259 Err(error) if error.code() == ErrorCode::NotFound => Ok(false),
260 Err(error) => Err(error),
261 }
262 }
263
264 async fn head(&self, key: &str) -> AppResult<StoredFile> {
265 let key = normalize_local_key(key)?;
266 let path = self.confined_existing_file_path(&key).await?;
267 let meta = async_io::file::metadata(&path)
268 .await
269 .map_err(|error| file_not_found_error_with_cause(&key, error))?;
270 if !meta.is_file || meta.is_symlink {
271 return Err(file_not_found_error(&key));
272 }
273
274 let mime = crate::detect_mime(&FileSource::Path(path)).await?;
275
276 let stored_at = meta
277 .modified
278 .map(chrono::DateTime::<chrono::Utc>::from)
279 .unwrap_or_else(chrono::Utc::now);
280
281 Ok(StoredFile::new(key, meta.len, Some(&mime)).with_stored_at(stored_at))
282 }
283
284 async fn list(&self, prefix: &str, limit: Option<usize>) -> AppResult<Vec<StoredFile>> {
285 let Some(dir) = self.confined_existing_dir_path(prefix).await? else {
286 return Ok(Vec::new());
287 };
288 let mut results = Vec::new();
289
290 let mut entries = tokio::fs::read_dir(&dir).await.map_err(|error| {
291 AppError::new(
292 ErrorCode::Internal,
293 format!("failed to read directory '{}': {error}", dir.display()),
294 )
295 .with_cause(error)
296 })?;
297 while let Some(entry) = entries.next_entry().await.map_err(|error| {
298 AppError::new(
299 ErrorCode::Internal,
300 format!("failed to read directory entry: {error}"),
301 )
302 .with_cause(error)
303 })? {
304 if let Some(max) = limit
305 && results.len() >= max
306 {
307 break;
308 }
309
310 let path = entry.path();
311 let file_type = entry.file_type().await.map_err(|error| {
312 AppError::new(
313 ErrorCode::Internal,
314 format!(
315 "failed to inspect directory entry '{}': {error}",
316 path.display()
317 ),
318 )
319 .with_cause(error)
320 })?;
321 if file_type.is_file() {
322 let meta = async_io::file::metadata(&path).await?;
323 let filename = entry.file_name();
324 let key = prefixed_key(Some(prefix), filename.to_string_lossy().as_ref());
325 let stored_at = meta
326 .modified
327 .map(chrono::DateTime::<chrono::Utc>::from)
328 .unwrap_or_else(chrono::Utc::now);
329 results.push(StoredFile::new(key, meta.len, None).with_stored_at(stored_at));
330 }
331 }
332
333 Ok(results)
334 }
335
336 async fn presigned_url(&self, key: &str, _expires_in: Duration) -> AppResult<String> {
337 let path = self.confined_presigned_path(key).await?;
338 Ok(format!("file://{}", path.display()))
339 }
340
341 async fn copy(&self, from_key: &str, to_key: &str) -> AppResult<StoredFile> {
342 let from = self.confined_existing_file_path(from_key).await?;
343 let to = self.resolve_path(to_key)?;
344
345 let mut reader = tokio::fs::File::open(&from).await.map_err(|error| {
346 AppError::new(
347 ErrorCode::Internal,
348 format!("failed to open source file '{}': {error}", from.display()),
349 )
350 .with_cause(error)
351 })?;
352 self.stream_to_target(&mut reader, &to).await?;
353
354 self.head(to_key).await
355 }
356
357 async fn rename(&self, from_key: &str, to_key: &str) -> AppResult<StoredFile> {
358 let from = self.confined_existing_file_path(from_key).await?;
359 let to = self.resolve_path(to_key)?;
360 ensure_target_parent_confined(&self.config.root_dir, &to).await?;
361
362 async_io::file::rename(&from, &to)
363 .await
364 .map_err(|error| error.context(format!("failed to rename {from_key} to {to_key}")))?;
365
366 self.head(to_key).await
367 }
368}
369
370#[cfg(test)]
371mod focused_tests {
372 use super::*;
373 use tokio::io::{AsyncRead, ReadBuf};
374
375 struct FailingReader;
376
377 impl AsyncRead for FailingReader {
378 fn poll_read(
379 self: std::pin::Pin<&mut Self>,
380 _cx: &mut std::task::Context<'_>,
381 _buf: &mut ReadBuf<'_>,
382 ) -> std::task::Poll<std::io::Result<()>> {
383 std::task::Poll::Ready(Err(std::io::Error::other("read failed")))
384 }
385 }
386
387 fn store_at(root: &Path) -> LocalStore {
388 LocalStore::new(LocalStoreConfig {
389 root_dir: root.to_path_buf(),
390 auto_create: true,
391 })
392 .unwrap()
393 }
394
395 #[test]
396 fn inspect_existing_file_error_maps_not_found_and_internal_causes() {
397 let not_found = LocalStore::inspect_existing_file_error(
398 "missing",
399 Path::new("missing"),
400 std::io::Error::new(std::io::ErrorKind::NotFound, "gone"),
401 );
402 assert_eq!(not_found.code(), ErrorCode::NotFound);
403
404 let denied = LocalStore::inspect_existing_file_error(
405 "blocked",
406 Path::new("blocked"),
407 std::io::Error::new(std::io::ErrorKind::PermissionDenied, "denied"),
408 );
409 assert_eq!(denied.code(), ErrorCode::Internal);
410 assert!(
411 denied
412 .to_string()
413 .contains("failed to inspect storage file")
414 );
415 }
416
417 #[test]
418 fn new_rejects_missing_root_when_auto_create_is_disabled() {
419 let root = tempfile::tempdir().unwrap();
420 let missing = root.path().join("missing");
421 let err = match LocalStore::new(LocalStoreConfig {
422 root_dir: missing,
423 auto_create: false,
424 }) {
425 Ok(_) => panic!("missing root should be rejected"),
426 Err(err) => err,
427 };
428
429 assert_eq!(err.code(), ErrorCode::NotFound);
430 assert!(err.to_string().contains("store root"));
431 }
432
433 #[tokio::test]
434 async fn private_path_flows_cover_directory_presign_and_stream_cleanup_errors() {
435 let root = tempfile::tempdir().unwrap();
436 let store = store_at(root.path());
437 std::fs::create_dir(root.path().join("dir")).unwrap();
438
439 assert!(
440 store
441 .confined_existing_dir_path("missing")
442 .await
443 .unwrap()
444 .is_none()
445 );
446 assert!(
447 store
448 .confined_existing_dir_path("dir")
449 .await
450 .unwrap()
451 .is_some()
452 );
453 assert_eq!(
454 store
455 .confined_presigned_path("dir")
456 .await
457 .unwrap_err()
458 .code(),
459 ErrorCode::NotFound
460 );
461
462 let mut reader = FailingReader;
463 let target = root.path().join("failed.bin");
464 let err = store
465 .stream_to_target(&mut reader, &target)
466 .await
467 .unwrap_err();
468 assert_eq!(err.code(), ErrorCode::Internal);
469 assert!(err.to_string().contains("failed to stream"));
470 assert!(!target.exists());
471 }
472
473 #[tokio::test]
474 async fn upload_copy_and_rename_normalize_keys_and_preserve_metadata_paths() {
475 let root = tempfile::tempdir().unwrap();
476 let store = store_at(root.path());
477 let source = FileSource::from_bytes(bytes::Bytes::from_static(b"data"));
478 let metadata = [("owner".to_string(), "dataset".to_string())]
479 .into_iter()
480 .collect();
481
482 let stored = store
483 .upload(
484 &source,
485 "/nested/item.txt",
486 UploadOptions::new()
487 .with_content_type("text/plain")
488 .with_metadata(metadata),
489 )
490 .await
491 .unwrap();
492 assert_eq!(stored.key, "nested/item.txt");
493 assert_eq!(
494 stored.metadata.get("owner").map(String::as_str),
495 Some("dataset")
496 );
497
498 let copied = store.copy("nested/item.txt", "copy.txt").await.unwrap();
499 assert_eq!(copied.key, "copy.txt");
500 let renamed = store.rename("copy.txt", "renamed.txt").await.unwrap();
501 assert_eq!(renamed.key, "renamed.txt");
502 assert!(store.exists("renamed.txt").await.unwrap());
503 }
504
505 #[tokio::test]
506 async fn download_head_list_presign_progress_and_delete_cover_local_store_contract() {
507 let root = tempfile::tempdir().unwrap();
508 let store = store_at(root.path());
509 let source = FileSource::from_bytes(bytes::Bytes::from_static(b"hello"));
510
511 let uploaded = store
512 .upload(
513 &source,
514 "dir/hello.txt",
515 UploadOptions::new()
516 .with_content_type("text/plain")
517 .with_progress(std::sync::Arc::new(|_| {})),
518 )
519 .await
520 .unwrap();
521 assert_eq!(uploaded.size, 5);
522
523 let head = store.head("dir/hello.txt").await.unwrap();
524 assert_eq!(head.key, "dir/hello.txt");
525 assert_eq!(head.size, 5);
526 assert!(head.content_type.contains("text"));
527
528 let downloaded = store.download("dir/hello.txt").await.unwrap();
529 assert!(matches!(downloaded, FileSource::Path(_)));
530 let listed = store.list("dir", Some(1)).await.unwrap();
531 assert_eq!(listed.len(), 1);
532 assert_eq!(listed[0].key, "dir/hello.txt");
533 assert!(store.list("missing", None).await.unwrap().is_empty());
534
535 let existing_url = store
536 .presigned_url("dir/hello.txt", Duration::from_secs(60))
537 .await
538 .unwrap();
539 assert!(existing_url.starts_with("file://"));
540 let missing_url = store
541 .presigned_url("dir/new.txt", Duration::from_secs(60))
542 .await
543 .unwrap();
544 assert!(missing_url.ends_with("dir/new.txt"));
545
546 store.delete("dir/hello.txt").await.unwrap();
547 assert!(!store.exists("dir/hello.txt").await.unwrap());
548 assert_eq!(
549 store.delete("dir/hello.txt").await.unwrap_err().code(),
550 ErrorCode::NotFound
551 );
552 }
553
554 #[tokio::test]
555 async fn file_operations_reject_directory_keys_as_missing_files() {
556 let root = tempfile::tempdir().unwrap();
557 let store = store_at(root.path());
558 std::fs::create_dir_all(root.path().join("dir")).unwrap();
559
560 assert_eq!(
561 store.download("dir").await.unwrap_err().code(),
562 ErrorCode::NotFound
563 );
564 assert_eq!(
565 store.head("dir").await.unwrap_err().code(),
566 ErrorCode::NotFound
567 );
568 assert!(!store.exists("dir").await.unwrap());
569 }
570}