perf(backend): replace buffered upload/download with opendal Writer/Reader streams

backend/Cargo.lock

@@ -190,6 +190,7 @@ dependencies = [
  "axum-test",
  "chrono",
  "dotenvy",
+ "futures-util",
  "lazy-regex",
  "opendal",
  "rand 0.10.1",

backend/Cargo.toml

@@ -20,6 +20,7 @@ dotenvy = "0.15.7"
 opendal = { version = "0.56.0", features = ["tests", "services-fs"] }
 argon2 = "0.5.3"
 rand = "0.10.1"
+futures-util = "0.3.32"
 
 [dev-dependencies]
 axum-test = "20.0.0"

backend/src/model.rs

@@ -1,6 +1,8 @@
+use axum::{body::Bytes, extract::multipart::MultipartError};
+use futures_util::{Stream, StreamExt};
 use opendal::{Operator, layers::LoggingLayer, services};
 use serde::{Deserialize, Serialize};
-use sqlx::{PgPool, prelude::FromRow, types::uuid};
+use sqlx::{PgPool, prelude::FromRow};
 use std::fmt::Display;
 use tracing::info;
 
@@ -56,17 +58,29 @@ impl FileRepository {
 
     pub async fn upload_file(
         &self,
-        bytes: Vec<u8>,
+        mut file_byte_stream: impl Stream<Item = Result<Bytes, MultipartError>> + Unpin,
-        file_name: String,
+        file_storage_key: &str,
+    ) -> Result<usize, LoftError> {
+        let mut writer = self.op.writer(&file_storage_key).await.unwrap();
+        let mut total_size = 0;
+        while let Some(chunk) = file_byte_stream.next().await {
+            let chunk = chunk.unwrap();
+            total_size += chunk.len();
+            writer.write(chunk).await.unwrap();
+        }
+        // must
+        writer.close().await.unwrap();
+        Ok(total_size)
+    }
+
+    pub async fn create_file_record(
+        &self,
+        file_name: &str,
+        file_size: usize,
+        file_storage_key: &str,
     ) -> Result<FileRecord, LoftError> {
-        let storage_path_name = format!("{}-{}", file_name, uuid::Uuid::new_v4());
-        let bytes_length = bytes.len();
-
-        info!("Uploading file \"{}\"", file_name);
-        self.op.write(&storage_path_name, bytes).await.unwrap();
-
         info!("Saving metadata of file \"{}\" in file_records", file_name);
-        let file = sqlx::query_as!(
+        let file_record = sqlx::query_as!(
             FileRecord,
             r#"
             INSERT INTO file_records (name, file_type, size, storage_path)
@@ -75,26 +89,30 @@ impl FileRepository {
             "#,
             file_name,
             "TODO-file_type".to_string(),
-            bytes_length as i64,
+            file_size as i64,
-            storage_path_name
+            file_storage_key
         )
         .fetch_one(&self.pool)
         .await
         .unwrap();
 
-        Ok(file)
+        Ok(file_record)
     }
 
-    pub async fn download_file(&self, file_id: i64) -> Result<Vec<u8>, LoftError> {
+    pub async fn download_file(
+        &self,
+        file_id: i64,
+    ) -> Result<impl Stream<Item = std::io::Result<Bytes>> + use<>, LoftError> {
         info!(
             "Fetching metadata of file \"{}\" from file_records",
             file_id
         );
         let record = self.get_file(file_id).await?;
         info!("Downloading file \"{}\"", file_id);
-        let bytes = self.op.read(&record.storage_path).await.unwrap();
+        let reader = self.op.reader(&record.storage_path).await.unwrap();
+        let stream = reader.into_bytes_stream(0..).await.unwrap();
 
-        Ok(bytes.to_vec())
+        Ok(stream)
     }
 
     pub async fn get_file(&self, file_id: i64) -> Result<FileRecord, LoftError> {
@@ -292,7 +310,9 @@ impl UserRepository {
 
 #[cfg(test)]
 mod tests {
+    use futures_util::stream;
     use rand::RngExt;
+    use sqlx::types::uuid;
 
     use super::*;
 
@@ -317,11 +337,26 @@ mod tests {
         truncate_file_records(&file_repository.pool).await;
 
         file_repository
-            .upload_file(vec![0u8; 10], "a.jpg".to_string())
+            .upload_file(
+                stream::iter(vec![Ok(Bytes::from(vec![0u8; 10]))]),
+                "a.jpg-uuid",
+            )
             .await
             .unwrap();
         file_repository
-            .upload_file(vec![0u8; 10], "b.jpg".to_string())
+            .upload_file(
+                stream::iter(vec![Ok(Bytes::from(vec![0u8; 10]))]),
+                "b.jpg-uuid",
+            )
+            .await
+            .unwrap();
+
+        file_repository
+            .create_file_record("a.png", 2, "a.jpg-uuid")
+            .await
+            .unwrap();
+        file_repository
+            .create_file_record("b.png", 2, "b.jpg-uuid")
             .await
             .unwrap();
         let files = file_repository.list_files().await.unwrap();
@@ -336,13 +371,24 @@ mod tests {
         let file_repository = file_repository().await.unwrap();
         truncate_file_records(&file_repository.pool).await;
 
-        let uploaded = file_repository
+        let file_name = "a.png";
-            .upload_file(vec![0u8; 10], "a.jpg".to_string())
+        let file_storage_key = format!("{}-{}", file_name, uuid::Uuid::new_v4());
+        let file_size = file_repository
+            .upload_file(
+                stream::iter(vec![Ok(Bytes::from(vec![0u8; 10]))]),
+                &file_storage_key,
+            )
             .await
             .unwrap();
-        let downloaded = file_repository.download_file(uploaded.id).await.unwrap();
 
-        assert!(!downloaded.is_empty());
+        let file_record = file_repository
+            .create_file_record("a.jpg", file_size, &file_storage_key)
+            .await
+            .unwrap();
+
+        let downloaded = file_repository.download_file(file_record.id).await.unwrap();
+
+        // assert!();
         truncate_file_records(&file_repository.pool).await;
     }
 
@@ -364,14 +410,25 @@ mod tests {
         let file_repository = file_repository().await.unwrap();
         truncate_file_records(&file_repository.pool).await;
 
-        let uploaded = file_repository
+        let file_name = "a.png";
-            .upload_file(vec![0u8; 10], "a.jpg".to_string())
+        let file_storage_key = format!("{}-{}", file_name, uuid::Uuid::new_v4());
+        let file_size = file_repository
+            .upload_file(
+                stream::iter(vec![Ok(Bytes::from(vec![0u8; 10]))]),
+                &file_storage_key,
+            )
             .await
             .unwrap();
-        file_repository.delete_file(uploaded.id).await.unwrap();
+
+        let file_record = file_repository
+            .create_file_record("a.jpg", file_size, &file_storage_key)
+            .await
+            .unwrap();
+
+        file_repository.delete_file(file_record.id).await.unwrap();
 
         assert!(matches!(
-            file_repository.download_file(uploaded.id).await,
+            file_repository.download_file(file_record.id).await,
             Err(LoftError::FileIdNotFound)
         ));
         truncate_file_records(&file_repository.pool).await;

backend/src/web/routes_file.rs

@@ -1,9 +1,11 @@
 use axum::{
     Json, Router,
+    body::Body,
     extract::{Multipart, Path, State},
     response::IntoResponse,
     routing::get,
 };
+use sqlx::types::uuid;
 use tracing::info;
 
 use crate::{
@@ -26,23 +28,25 @@ async fn upload_file(
     info!("handler: upload_file");
 
     let mut file_name = None;
-    let mut file_type: Option<String> = None;
+    let mut file_storage_key = None;
-    let mut bytes = None;
+    let mut file_size = None;
 
     while let Some(field) = multipart.next_field().await.unwrap() {
-        match field.name().unwrap() {
+        if field.name().unwrap() == "file" {
-            // "file_type" => file_type = Some(field.text().await.unwrap().parse().unwrap()),
+            let name = field.file_name().map(|s| s.to_string()).unwrap_or_default();
-            "file" => {
+            let key = format!("{}-{}", name, uuid::Uuid::new_v4());
-                file_name = field.file_name().map(|s| s.to_string());
+            let size = file_repository.upload_file(field, &key).await?;
-                bytes = Some(field.bytes().await.unwrap().to_vec())
+            file_name = Some(name);
-            }
+            file_storage_key = Some(key);
-            _ => (),
+            file_size = Some(size);
         }
     }
 
-    if let (Some(bytes), Some(file_name)) = (bytes, file_name) {
+    if let (Some(name), Some(key), Some(size)) = (file_name, file_storage_key, file_size) {
-        let file = file_repository.upload_file(bytes, file_name).await?;
+        let file_record = file_repository
-        return Ok(Json(file));
+            .create_file_record(&name, size, &key)
+            .await?;
+        return Ok(Json(file_record));
     }
 
     Err(LoftError::UndefinedErrorType)
@@ -62,8 +66,8 @@ async fn download_file(
     State(file_repository): State<FileRepository>,
     Path(file_id): Path<u64>,
 ) -> Result<impl IntoResponse, LoftError> {
-    let bytes = file_repository.download_file(file_id as i64).await?;
+    let stream = file_repository.download_file(file_id as i64).await?;
-    Ok(bytes)
+    Ok(Body::from_stream(stream))
 }
 
 async fn delete_file(