diff --git a/backend/Cargo.lock b/backend/Cargo.lock index a94d377..cdb3d4a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -190,6 +190,7 @@ dependencies = [ "axum-test", "chrono", "dotenvy", + "futures-util", "lazy-regex", "opendal", "rand 0.10.1", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 8dd0d86..e65fa22 100644 --- a/backend/Cargo.toml +++ b/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" diff --git a/backend/src/model.rs b/backend/src/model.rs index ada7880..965f806 100644 --- a/backend/src/model.rs +++ b/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, - file_name: String, + mut file_byte_stream: impl Stream> + Unpin, + file_storage_key: &str, + ) -> Result { + 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 { - 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, - storage_path_name + file_size as i64, + file_storage_key ) .fetch_one(&self.pool) .await .unwrap(); - Ok(file) + Ok(file_record) } - pub async fn download_file(&self, file_id: i64) -> Result, LoftError> { + pub async fn download_file( + &self, + file_id: i64, + ) -> Result> + 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 { @@ -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 - .upload_file(vec![0u8; 10], "a.jpg".to_string()) + let file_name = "a.png"; + 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 - .upload_file(vec![0u8; 10], "a.jpg".to_string()) + let file_name = "a.png"; + 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; diff --git a/backend/src/web/routes_file.rs b/backend/src/web/routes_file.rs index ea10aa3..906f524 100644 --- a/backend/src/web/routes_file.rs +++ b/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 = None; - let mut bytes = None; + let mut file_storage_key = None; + let mut file_size = None; while let Some(field) = multipart.next_field().await.unwrap() { - match field.name().unwrap() { - // "file_type" => file_type = Some(field.text().await.unwrap().parse().unwrap()), - "file" => { - file_name = field.file_name().map(|s| s.to_string()); - bytes = Some(field.bytes().await.unwrap().to_vec()) - } - _ => (), + if field.name().unwrap() == "file" { + let name = field.file_name().map(|s| s.to_string()).unwrap_or_default(); + let key = format!("{}-{}", name, uuid::Uuid::new_v4()); + let size = file_repository.upload_file(field, &key).await?; + file_name = Some(name); + file_storage_key = Some(key); + file_size = Some(size); } } - if let (Some(bytes), Some(file_name)) = (bytes, file_name) { - let file = file_repository.upload_file(bytes, file_name).await?; - return Ok(Json(file)); + if let (Some(name), Some(key), Some(size)) = (file_name, file_storage_key, file_size) { + let file_record = file_repository + .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, Path(file_id): Path, ) -> Result { - let bytes = file_repository.download_file(file_id as i64).await?; - Ok(bytes) + let stream = file_repository.download_file(file_id as i64).await?; + Ok(Body::from_stream(stream)) } async fn delete_file(