GKaszewski/libertas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

separation of database (#1)

Browse Source 
Reviewed-on: #1
This commit is contained in:
Gabriel Kaszewski
2025-11-03 02:26:19 +00:00
parent 8b98df745c
commit 39ee8d52a4
21 changed files with 407 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
libertas_api/src/services/media_service.rs
View File

@@ -46,89 +46,24 @@ impl MediaServiceImpl {
#[async_trait]
impl MediaService for MediaServiceImpl {
async fn upload_media(&self, mut data: UploadMediaData<'_>) -> CoreResult<Media> {
let user = self
.user_repo
.find_by_id(data.owner_id)
.await?
.ok_or(CoreError::NotFound("User".to_string(), data.owner_id))?;
let (file_bytes, hash, file_size) = self.hash_and_buffer_stream(&mut data).await?;
let mut hasher = Sha256::new();
let mut file_bytes = Vec::new();
let owner_id = data.owner_id;
let filename = data.filename;
let mime_type = data.mime_type;
while let Some(chunk_result) = data.stream.next().await {
let chunk = chunk_result.map_err(|e| CoreError::Io(e))?;
hasher.update(&chunk);
file_bytes.extend_from_slice(&chunk);
}
let file_size = file_bytes.len() as i64;
if user.storage_used + file_size > user.storage_quota {
return Err(CoreError::Auth(format!(
"Storage quota exceeded. Used: {}, Quota: {}",
user.storage_used, user.storage_quota
)));
}
let hash = format!("{:x}", hasher.finalize());
if self.repo.find_by_hash(&hash).await?.is_some() {
return Err(CoreError::Duplicate(
"A file with this content already exists".to_string(),
));
}
let now = chrono::Utc::now();
let year = now.year().to_string();
let month = format!("{:02}", now.month());
let mut dest_path = PathBuf::from(&self.config.media_library_path);
dest_path.push(year.clone());
dest_path.push(month.clone());
fs::create_dir_all(&dest_path)
.await
.map_err(|e| CoreError::Io(e))?;
dest_path.push(&data.filename);
let storage_path_str = PathBuf::from(&year)
.join(&month)
.join(&data.filename)
.to_string_lossy()
.to_string();
let mut file = fs::File::create(&dest_path)
.await
.map_err(|e| CoreError::Io(e))?;
file.write_all(&file_bytes)
.await
.map_err(|e| CoreError::Io(e))?;
let media_model = Media {
id: Uuid::new_v4(),
owner_id: data.owner_id,
storage_path: storage_path_str,
original_filename: data.filename,
mime_type: data.mime_type,
hash,
created_at: now,
extracted_location: None,
width: None,
height: None,
};
self.repo.create(&media_model).await?;
self.user_repo
.update_storage_used(user.id, file_size)
self.check_upload_prerequisites(owner_id, file_size, &hash)
.await?;
let job_payload = json!({ "media_id": media_model.id });
self.nats_client
.publish("media.new".to_string(), job_payload.to_string().into())
.await
.map_err(|e| CoreError::Unknown(format!("Failed to publish NATS job: {}", e)))?;
let storage_path = self.persist_media_file(&file_bytes, &filename).await?;
Ok(media_model)
let media = self
.persist_media_metadata(owner_id, filename, mime_type, storage_path, hash, file_size)
.await?;
self.publish_new_media_job(media.id).await?;
Ok(media)
}
async fn get_media_details(&self, id: Uuid, user_id: Uuid) -> CoreResult<Media> {
@@ -235,3 +170,110 @@ impl MediaService for MediaServiceImpl {
Ok(())
}
}
impl MediaServiceImpl {
async fn hash_and_buffer_stream(
&self,
stream: &mut UploadMediaData<'_>,
) -> CoreResult<(Vec<u8>, String, i64)> {
let mut hasher = Sha256::new();
let mut file_bytes = Vec::new();
while let Some(chunk_result) = stream.stream.next().await {
let chunk = chunk_result.map_err(|e| CoreError::Io(e))?;
hasher.update(&chunk);
file_bytes.extend_from_slice(&chunk);
}
let file_size = file_bytes.len() as i64;
let hash = format!("{:x}", hasher.finalize());
Ok((file_bytes, hash, file_size))
}
async fn check_upload_prerequisites(
&self,
user_id: Uuid,
file_size: i64,
hash: &str,
) -> CoreResult<()> {
let user = self
.user_repo
.find_by_id(user_id)
.await?
.ok_or(CoreError::NotFound("User".to_string(), user_id))?;
if user.storage_used + file_size > user.storage_quota {
return Err(CoreError::Auth(format!(
"Storage quota exceeded. Used: {}, Quota: {}",
user.storage_used, user.storage_quota
)));
}
if self.repo.find_by_hash(hash).await?.is_some() {
return Err(CoreError::Duplicate(
"A file with this content already exists".to_string(),
));
}
Ok(())
}
async fn persist_media_file(&self, file_bytes: &[u8], filename: &str) -> CoreResult<String> {
let now = chrono::Utc::now();
let year = now.year().to_string();
let month = format!("{:02}", now.month());
let mut dest_path = PathBuf::from(&self.config.media_library_path);
dest_path.push(year.clone());
dest_path.push(month.clone());
fs::create_dir_all(&dest_path).await?;
dest_path.push(filename);
let storage_path_str = PathBuf::from(&year)
.join(&month)
.join(filename)
.to_string_lossy()
.to_string();
let mut file = fs::File::create(&dest_path).await?;
file.write_all(&file_bytes).await?;
Ok(storage_path_str)
}
async fn persist_media_metadata(
&self,
owner_id: Uuid,
filename: String,
mime_type: String,
storage_path: String,
hash: String,
file_size: i64,
) -> CoreResult<Media> {
let media_model = Media {
id: Uuid::new_v4(),
owner_id,
storage_path,
original_filename: filename,
mime_type,
hash,
created_at: chrono::Utc::now(),
extracted_location: None,
width: None,
height: None,
};
self.repo.create(&media_model).await?;
self.user_repo
.update_storage_used(owner_id, file_size)
.await?;
Ok(media_model)
}
async fn publish_new_media_job(&self, media_id: Uuid) -> CoreResult<()> {
let job_payload = json!({ "media_id": media_id });
self.nats_client
.publish("media.new".to_string(), job_payload.to_string().into())
.await
.map_err(|e| CoreError::Unknown(format!("Failed to publish NATS job: {}", e)))
}
}