refactor(tvix/castore/import): move upload_blob_at_path into fs mod r/8046

This is only useful for when we have access to a filesystem, so it
shouldn't be in the root.

Change-Id: I9923aaed1aef9d3a1e8fad41f58821d51c2eb34b
Reviewed-on: https://cl.tvl.fyi/c/depot/+/11555
Autosubmit: flokli <flokli@flokli.de>
Reviewed-by: firefly <firefly@firefly.nu>
Tested-by: BuildkiteCI

1 files changed, 27 insertions, 2 deletions

diff --git a/tvix/castore/src/import/fs.rs b/tvix/castore/src/import/fs.rs
index 5e7061bec66d..13e4a78cb799 100644
--- a/tvix/castore/src/import/fs.rs
+++ b/tvix/castore/src/import/fs.rs
@@ -13,9 +13,9 @@ use walkdir::WalkDir;
 use crate::blobservice::BlobService;
 use crate::directoryservice::DirectoryService;
 use crate::proto::node::Node;
+use crate::B3Digest;
 
 use super::ingest_entries;
-use super::upload_blob_at_path;
 use super::Error;
 use super::IngestionEntry;
 
@@ -119,7 +119,7 @@ where
             .metadata()
             .map_err(|e| Error::UnableToStat(entry.path().to_path_buf(), e.into()))?;
 
-        let digest = upload_blob_at_path(blob_service, entry.path().to_path_buf()).await?;
+        let digest = upload_blob(blob_service, entry.path().to_path_buf()).await?;
 
         Ok(IngestionEntry::Regular {
             path,
@@ -133,3 +133,28 @@ where
         Ok(IngestionEntry::Unknown { path, file_type })
     }
 }
+
+/// Uploads the file at the provided [Path] the the [BlobService].
+#[instrument(skip(blob_service), fields(path), err)]
+async fn upload_blob<BS>(blob_service: BS, path: impl AsRef<Path>) -> Result<B3Digest, Error>
+where
+    BS: BlobService,
+{
+    let mut file = match tokio::fs::File::open(path.as_ref()).await {
+        Ok(file) => file,
+        Err(e) => return Err(Error::UnableToRead(path.as_ref().to_path_buf(), e)),
+    };
+
+    let mut writer = blob_service.open_write().await;
+
+    if let Err(e) = tokio::io::copy(&mut file, &mut writer).await {
+        return Err(Error::UnableToRead(path.as_ref().to_path_buf(), e));
+    };
+
+    let digest = writer
+        .close()
+        .await
+        .map_err(|e| Error::UnableToRead(path.as_ref().to_path_buf(), e))?;
+
+    Ok(digest)
+}