diff --git a/src/core/mod.rs b/src/core/mod.rs index 91cb7ca..4da4b43 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -100,7 +100,9 @@ impl App { let mut interval = tokio::time::interval(Duration::from_secs(3600)); loop { interval.tick().await; - if let Err(error) = crate::routes::attachment::handlers::cleanup_expired(&cleanup_state).await { + if let Err(error) = + crate::routes::attachment::handlers::cleanup_expired(&cleanup_state).await + { tracing::warn!(%error, "Attachment cleanup failed"); } } diff --git a/src/domain/dto/message.rs b/src/domain/dto/message.rs index 515beab..7e6b6dc 100644 --- a/src/domain/dto/message.rs +++ b/src/domain/dto/message.rs @@ -1,5 +1,5 @@ -use crate::domain::dto::reaction::ReactionGroupResponse; use crate::domain::dto::attachment::AttachmentResponse; +use crate::domain::dto::reaction::ReactionGroupResponse; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use utoipa::ToSchema; diff --git a/src/repositories/message.rs b/src/repositories/message.rs index 80b5199..3ec6996 100644 --- a/src/repositories/message.rs +++ b/src/repositories/message.rs @@ -1,6 +1,6 @@ use super::types::MessageFilter; -use crate::models::message; use crate::models::attachment; +use crate::models::message; use crate::repositories::{AnyResult, RepositoryContext}; use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect}; use std::collections::HashMap; @@ -46,7 +46,10 @@ impl MessageRepository { let mut grouped = HashMap::new(); for item in items { if let Some(message_id) = item.message_id { - grouped.entry(message_id).or_insert_with(Vec::new).push(item); + grouped + .entry(message_id) + .or_insert_with(Vec::new) + .push(item); } } Ok(grouped) diff --git a/src/routes/attachment/handlers.rs b/src/routes/attachment/handlers.rs index da9c3a2..bd01a68 100644 --- a/src/routes/attachment/handlers.rs +++ b/src/routes/attachment/handlers.rs @@ -23,8 +23,12 @@ pub async fn cleanup_expired(state: &AppState) -> Result<(), HTTPError> { .all(&state.db) .await?; for item in pending { - let _ = tokio::fs::remove_file(PathBuf::from(&state.config.media.root).join(&item.file_path)).await; - attachment::Entity::delete_by_id(item.id).exec(&state.db).await?; + let _ = + tokio::fs::remove_file(PathBuf::from(&state.config.media.root).join(&item.file_path)) + .await; + attachment::Entity::delete_by_id(item.id) + .exec(&state.db) + .await?; } media::cleanup_temporary_files( PathBuf::from(&state.config.media.root).as_path(), @@ -58,13 +62,22 @@ pub async fn create( { let name = field.name().unwrap_or_default().to_string(); if name == "channel_id" { - let value = field.text().await.map_err(|error| HTTPError::BadRequest(error.to_string()))?; + let value = field + .text() + .await + .map_err(|error| HTTPError::BadRequest(error.to_string()))?; channel_id = Some(value.parse::().map_err(HTTPError::UuidError)?); } else if name == "files" || name == "file" { - let channel = channel_id.ok_or_else(|| HTTPError::BadRequest("channel_id must precede files".into()))?; + let channel = channel_id + .ok_or_else(|| HTTPError::BadRequest("channel_id must precede files".into()))?; let filename = field.file_name().unwrap_or("unnamed").to_string(); - let mime_type = field.content_type().unwrap_or("application/octet-stream").to_string(); - if !can_access(&state, channel, user.id).await? { return Err(HTTPError::Forbidden); } + let mime_type = field + .content_type() + .unwrap_or("application/octet-stream") + .to_string(); + if !can_access(&state, channel, user.id).await? { + return Err(HTTPError::Forbidden); + } let id = Uuid::new_v4(); let mut output = PendingMediaFile::begin( PathBuf::from(&state.config.media.root).as_path(), @@ -75,35 +88,84 @@ pub async fn create( .await .map_err(|error| HTTPError::InternalServerError(error.to_string()))?; let mut size = 0_i64; - while let Some(chunk) = field.chunk().await.map_err(|error| HTTPError::BadRequest(error.to_string()))? { + while let Some(chunk) = field + .chunk() + .await + .map_err(|error| HTTPError::BadRequest(error.to_string()))? + { size += chunk.len() as i64; - output.write(&chunk).await.map_err(|error| HTTPError::InternalServerError(error.to_string()))?; + output + .write(&chunk) + .await + .map_err(|error| HTTPError::InternalServerError(error.to_string()))?; } - let file_path = output.finish().await.map_err(|error| HTTPError::InternalServerError(error.to_string()))?; + let file_path = output + .finish() + .await + .map_err(|error| HTTPError::InternalServerError(error.to_string()))?; let model = attachment::ActiveModel { - id: Set(id), message_id: Set(None), channel_id: Set(channel), user_id: Set(user.id), - filename: Set(filename), file_size: Set(size), mime_type: Set(mime_type), file_path: Set(file_path.clone()), created_at: Set(Utc::now()), + id: Set(id), + message_id: Set(None), + channel_id: Set(channel), + user_id: Set(user.id), + filename: Set(filename), + file_size: Set(size), + mime_type: Set(mime_type), + file_path: Set(file_path.clone()), + created_at: Set(Utc::now()), }; match model.insert(&state.db).await { Ok(item) => created.push(item), - Err(error) => { let _ = tokio::fs::remove_file(PathBuf::from(&state.config.media.root).join(file_path)).await; return Err(HTTPError::Database(error)); } + Err(error) => { + let _ = tokio::fs::remove_file( + PathBuf::from(&state.config.media.root).join(file_path), + ) + .await; + return Err(HTTPError::Database(error)); + } } } } - let channel_id = channel_id.ok_or_else(|| HTTPError::BadRequest("channel_id is required".into()))?; - if !can_access(&state, channel_id, user.id).await? { return Err(HTTPError::Forbidden); } - if created.is_empty() { return Err(HTTPError::BadRequest("at least one file is required".into())); } - Ok((StatusCode::CREATED, axum::Json(AttachmentUploadResponse { attachments: created.into_iter().map(mapper::to_response).collect() }))) + let channel_id = + channel_id.ok_or_else(|| HTTPError::BadRequest("channel_id is required".into()))?; + if !can_access(&state, channel_id, user.id).await? { + return Err(HTTPError::Forbidden); + } + if created.is_empty() { + return Err(HTTPError::BadRequest( + "at least one file is required".into(), + )); + } + Ok(( + StatusCode::CREATED, + axum::Json(AttachmentUploadResponse { + attachments: created.into_iter().map(mapper::to_response).collect(), + }), + )) } -pub async fn file(State(state): State, Path(id): Path) -> Result { - let item = attachment::Entity::find_by_id(id).one(&state.db).await?.ok_or(HTTPError::NotFound)?; - let bytes = tokio::fs::read(PathBuf::from(&state.config.media.root).join(&item.file_path)).await.map_err(|_| HTTPError::NotFound)?; +pub async fn file( + State(state): State, + Path(id): Path, +) -> Result { + let item = attachment::Entity::find_by_id(id) + .one(&state.db) + .await? + .ok_or(HTTPError::NotFound)?; + let bytes = tokio::fs::read(PathBuf::from(&state.config.media.root).join(&item.file_path)) + .await + .map_err(|_| HTTPError::NotFound)?; let mut response = Response::new(Body::from(bytes)); - if let Ok(value) = item.mime_type.parse() { response.headers_mut().insert(header::CONTENT_TYPE, value); } + if let Ok(value) = item.mime_type.parse() { + response.headers_mut().insert(header::CONTENT_TYPE, value); + } if !item.mime_type.starts_with("image/") { let safe_name = item.filename.replace(['\"', '\r', '\n'], "_"); - if let Ok(value) = format!("attachment; filename=\"{safe_name}\"").parse() { response.headers_mut().insert(header::CONTENT_DISPOSITION, value); } + if let Ok(value) = format!("attachment; filename=\"{safe_name}\"").parse() { + response + .headers_mut() + .insert(header::CONTENT_DISPOSITION, value); + } } Ok(response) } diff --git a/src/routes/attachment/routes.rs b/src/routes/attachment/routes.rs index 189dfe2..2233c3d 100644 --- a/src/routes/attachment/routes.rs +++ b/src/routes/attachment/routes.rs @@ -1,6 +1,10 @@ -use axum::{Router, routing::{get, post}, extract::DefaultBodyLimit}; -use crate::core::state::AppState; use super::handlers; +use crate::core::state::AppState; +use axum::{ + Router, + extract::DefaultBodyLimit, + routing::{get, post}, +}; pub fn secure_router() -> Router { Router::new() diff --git a/src/routes/channel/handlers.rs b/src/routes/channel/handlers.rs index fd1b78e..8ae8c50 100644 --- a/src/routes/channel/handlers.rs +++ b/src/routes/channel/handlers.rs @@ -13,14 +13,30 @@ use axum::{ extract::{Path, Query, State}, http::StatusCode, }; -use uuid::Uuid; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; +use uuid::Uuid; -async fn require_channel_member(state: &AppState, channel_id: Uuid, user_id: Uuid) -> Result<(), HTTPError> { - let channel = state.repositories.channel.get_by_id(channel_id).await?.ok_or(HTTPError::NotFound)?; +async fn require_channel_member( + state: &AppState, + channel_id: Uuid, + user_id: Uuid, +) -> Result<(), HTTPError> { + let channel = state + .repositories + .channel + .get_by_id(channel_id) + .await? + .ok_or(HTTPError::NotFound)?; if channel.channel_type == channel::ChannelType::DM - && channel_user::Entity::find().filter(channel_user::Column::ChannelId.eq(channel_id)).filter(channel_user::Column::UserId.eq(user_id)).one(&state.db).await?.is_none() - { return Err(HTTPError::Forbidden); } + && channel_user::Entity::find() + .filter(channel_user::Column::ChannelId.eq(channel_id)) + .filter(channel_user::Column::UserId.eq(user_id)) + .one(&state.db) + .await? + .is_none() + { + return Err(HTTPError::Forbidden); + } Ok(()) } diff --git a/src/routes/conversation/handlers.rs b/src/routes/conversation/handlers.rs index d4549cc..d0222c0 100644 --- a/src/routes/conversation/handlers.rs +++ b/src/routes/conversation/handlers.rs @@ -6,70 +6,178 @@ use crate::domain::dto::conversation::{ use crate::http::context::CurrentUser; use crate::http::error::HTTPError; use crate::models::{channel, channel_user, message, user}; -use axum::{Json, extract::{Path, State}}; +use axum::{ + Json, + extract::{Path, State}, +}; use chrono::Utc; -use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set, TransactionTrait}; +use sea_orm::{ + ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set, TransactionTrait, +}; use std::collections::HashSet; use uuid::Uuid; async fn member_ids(state: &AppState, channel_id: Uuid) -> Result, HTTPError> { Ok(channel_user::Entity::find() .filter(channel_user::Column::ChannelId.eq(channel_id)) - .all(&state.db).await?.into_iter().map(|m| m.user_id).collect()) + .all(&state.db) + .await? + .into_iter() + .map(|m| m.user_id) + .collect()) } -async fn response(state: &AppState, current_user_id: Uuid, channel: channel::Model) -> Result { +async fn response( + state: &AppState, + current_user_id: Uuid, + channel: channel::Model, +) -> Result { let ids = member_ids(state, channel.id).await?; - let users = user::Entity::find().filter(user::Column::Id.is_in(ids.clone())).all(&state.db).await?; - let mut participants: Vec<_> = users.into_iter().map(|u| ConversationParticipantResponse { id: u.id, username: u.username }).collect(); + let users = user::Entity::find() + .filter(user::Column::Id.is_in(ids.clone())) + .all(&state.db) + .await?; + let mut participants: Vec<_> = users + .into_iter() + .map(|u| ConversationParticipantResponse { + id: u.id, + username: u.username, + }) + .collect(); participants.sort_by(|a, b| a.username.to_lowercase().cmp(&b.username.to_lowercase())); - let title = participants.iter().filter(|p| p.id != current_user_id).map(|p| p.username.clone()).collect::>().join(", "); - let last_message = message::Entity::find().filter(message::Column::ChannelId.eq(channel.id)).order_by_desc(message::Column::Id).one(&state.db).await?.map(|m| m.content); - let unread_count = state.repositories.read_state.unread_counts(&[channel.id], current_user_id).await?.get(&channel.id).copied().unwrap_or(0); - Ok(ConversationResponse { id: channel.id, title: if title.is_empty() { "Discussion".into() } else { title }, participants, last_message, unread_count, created_at: channel.created_at, updated_at: channel.updated_at }) + let title = participants + .iter() + .filter(|p| p.id != current_user_id) + .map(|p| p.username.clone()) + .collect::>() + .join(", "); + let last_message = message::Entity::find() + .filter(message::Column::ChannelId.eq(channel.id)) + .order_by_desc(message::Column::Id) + .one(&state.db) + .await? + .map(|m| m.content); + let unread_count = state + .repositories + .read_state + .unread_counts(&[channel.id], current_user_id) + .await? + .get(&channel.id) + .copied() + .unwrap_or(0); + Ok(ConversationResponse { + id: channel.id, + title: if title.is_empty() { + "Discussion".into() + } else { + title + }, + participants, + last_message, + unread_count, + created_at: channel.created_at, + updated_at: channel.updated_at, + }) } async fn create_channel(state: &AppState, ids: &[Uuid]) -> Result { let txn = state.db.begin().await?; - let channel = channel::ActiveModel { server_id: Set(None), category_id: Set(None), channel_type: Set(channel::ChannelType::DM), name: Set(None), ..Default::default() }.insert(&txn).await?; + let channel = channel::ActiveModel { + server_id: Set(None), + category_id: Set(None), + channel_type: Set(channel::ChannelType::DM), + name: Set(None), + ..Default::default() + } + .insert(&txn) + .await?; for user_id in ids { - channel_user::ActiveModel { channel_id: Set(channel.id), user_id: Set(*user_id), role: Set("member".into()), joined_at: Set(Utc::now()), ..Default::default() }.insert(&txn).await?; + channel_user::ActiveModel { + channel_id: Set(channel.id), + user_id: Set(*user_id), + role: Set("member".into()), + joined_at: Set(Utc::now()), + ..Default::default() + } + .insert(&txn) + .await?; } txn.commit().await?; - state.services.realtime_registry.set_channel_users(channel.id, ids.iter().copied()); + state + .services + .realtime_registry + .set_channel_users(channel.id, ids.iter().copied()); state.event_bus.emit("channel_created", channel.clone()); Ok(channel) } async fn validate_ids(state: &AppState, ids: &[Uuid]) -> Result<(), HTTPError> { - if ids.is_empty() { return Err(HTTPError::BadRequest("At least one participant is required".into())); } - let found = user::Entity::find().filter(user::Column::Id.is_in(ids.to_vec())).all(&state.db).await?; - if found.len() != ids.iter().collect::>().len() { return Err(HTTPError::BadRequest("Unknown participant".into())); } + if ids.is_empty() { + return Err(HTTPError::BadRequest( + "At least one participant is required".into(), + )); + } + let found = user::Entity::find() + .filter(user::Column::Id.is_in(ids.to_vec())) + .all(&state.db) + .await?; + if found.len() != ids.iter().collect::>().len() { + return Err(HTTPError::BadRequest("Unknown participant".into())); + } Ok(()) } #[utoipa::path(get, path = "/conversations", responses((status = 200, body = [ConversationResponse])), tag = "Conversations", security(("bearerAuth" = [])))] -pub async fn list(user: CurrentUser, State(state): State) -> Result>, HTTPError> { - let memberships = channel_user::Entity::find().filter(channel_user::Column::UserId.eq(user.id)).all(&state.db).await?; +pub async fn list( + user: CurrentUser, + State(state): State, +) -> Result>, HTTPError> { + let memberships = channel_user::Entity::find() + .filter(channel_user::Column::UserId.eq(user.id)) + .all(&state.db) + .await?; let ids: Vec<_> = memberships.into_iter().map(|m| m.channel_id).collect(); - if ids.is_empty() { return Ok(Json(Vec::new())); } - let channels = channel::Entity::find().filter(channel::Column::Id.is_in(ids)).filter(channel::Column::ChannelType.eq(channel::ChannelType::DM)).order_by_desc(channel::Column::UpdatedAt).all(&state.db).await?; + if ids.is_empty() { + return Ok(Json(Vec::new())); + } + let channels = channel::Entity::find() + .filter(channel::Column::Id.is_in(ids)) + .filter(channel::Column::ChannelType.eq(channel::ChannelType::DM)) + .order_by_desc(channel::Column::UpdatedAt) + .all(&state.db) + .await?; let mut result = Vec::with_capacity(channels.len()); - for channel in channels { result.push(response(&state, user.id, channel).await?); } + for channel in channels { + result.push(response(&state, user.id, channel).await?); + } Ok(Json(result)) } #[utoipa::path(post, path = "/conversations", request_body = CreateConversationRequest, responses((status = 200, body = ConversationResponse)), tag = "Conversations", security(("bearerAuth" = [])))] -pub async fn create(user: CurrentUser, State(state): State, Json(payload): Json) -> Result, HTTPError> { +pub async fn create( + user: CurrentUser, + State(state): State, + Json(payload): Json, +) -> Result, HTTPError> { let mut ids = payload.user_ids; ids.push(user.id); - ids.sort_unstable(); ids.dedup(); + ids.sort_unstable(); + ids.dedup(); validate_ids(&state, &ids).await?; - let channels = channel_user::Entity::find().filter(channel_user::Column::UserId.eq(user.id)).all(&state.db).await?; + let channels = channel_user::Entity::find() + .filter(channel_user::Column::UserId.eq(user.id)) + .all(&state.db) + .await?; for membership in channels { - if let Some(channel) = channel::Entity::find_by_id(membership.channel_id).filter(channel::Column::ChannelType.eq(channel::ChannelType::DM)).one(&state.db).await? { + if let Some(channel) = channel::Entity::find_by_id(membership.channel_id) + .filter(channel::Column::ChannelType.eq(channel::ChannelType::DM)) + .one(&state.db) + .await? + { let existing: HashSet<_> = member_ids(&state, channel.id).await?.into_iter().collect(); - if existing == ids.iter().copied().collect() { return Ok(Json(response(&state, user.id, channel).await?)); } + if existing == ids.iter().copied().collect() { + return Ok(Json(response(&state, user.id, channel).await?)); + } } } let channel = create_channel(&state, &ids).await?; @@ -77,12 +185,25 @@ pub async fn create(user: CurrentUser, State(state): State, Json(paylo } #[utoipa::path(post, path = "/conversations/{id}/fork", request_body = ForkConversationRequest, params(("id" = Uuid, Path)), responses((status = 200, body = ConversationResponse)), tag = "Conversations", security(("bearerAuth" = [])))] -pub async fn fork(user: CurrentUser, State(state): State, Path(id): Path, Json(payload): Json) -> Result, HTTPError> { - let source = channel::Entity::find_by_id(id).one(&state.db).await?.ok_or(HTTPError::NotFound)?; - if source.channel_type != channel::ChannelType::DM || !member_ids(&state, id).await?.contains(&user.id) { return Err(HTTPError::Forbidden); } +pub async fn fork( + user: CurrentUser, + State(state): State, + Path(id): Path, + Json(payload): Json, +) -> Result, HTTPError> { + let source = channel::Entity::find_by_id(id) + .one(&state.db) + .await? + .ok_or(HTTPError::NotFound)?; + if source.channel_type != channel::ChannelType::DM + || !member_ids(&state, id).await?.contains(&user.id) + { + return Err(HTTPError::Forbidden); + } let mut ids = member_ids(&state, id).await?; ids.extend(payload.user_ids); - ids.sort_unstable(); ids.dedup(); + ids.sort_unstable(); + ids.dedup(); validate_ids(&state, &ids).await?; let channel = create_channel(&state, &ids).await?; Ok(Json(response(&state, user.id, channel).await?)) diff --git a/src/routes/conversation/routes.rs b/src/routes/conversation/routes.rs index 2d7d916..d5eb4ac 100644 --- a/src/routes/conversation/routes.rs +++ b/src/routes/conversation/routes.rs @@ -1,6 +1,9 @@ use super::handlers; use crate::core::state::AppState; -use axum::{routing::{get, post}, Router}; +use axum::{ + Router, + routing::{get, post}, +}; pub fn router() -> Router { Router::new() diff --git a/src/routes/gateway/routes.rs b/src/routes/gateway/routes.rs index c79ff59..8f05510 100644 --- a/src/routes/gateway/routes.rs +++ b/src/routes/gateway/routes.rs @@ -1,7 +1,7 @@ use super::handlers; use crate::core::AppState; -use axum::Router; use axum::routing::get; +use axum::Router; pub fn router() -> Router { Router::new().route("/gateway", get(handlers::ws_handler)) diff --git a/src/routes/message/handlers.rs b/src/routes/message/handlers.rs index afcd3e3..25f97d1 100644 --- a/src/routes/message/handlers.rs +++ b/src/routes/message/handlers.rs @@ -13,11 +13,18 @@ use axum::{ extract::{Path, Query, State}, http::StatusCode, }; -use uuid::Uuid; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; +use uuid::Uuid; -pub(crate) async fn can_access(state: &AppState, channel_id: Uuid, user_id: Uuid) -> Result { - let Some(channel) = channel::Entity::find_by_id(channel_id).one(&state.db).await? else { +pub(crate) async fn can_access( + state: &AppState, + channel_id: Uuid, + user_id: Uuid, +) -> Result { + let Some(channel) = channel::Entity::find_by_id(channel_id) + .one(&state.db) + .await? + else { return Ok(false); }; if channel.channel_type != channel::ChannelType::DM { @@ -26,7 +33,9 @@ pub(crate) async fn can_access(state: &AppState, channel_id: Uuid, user_id: Uuid Ok(channel_user::Entity::find() .filter(channel_user::Column::ChannelId.eq(channel_id)) .filter(channel_user::Column::UserId.eq(user_id)) - .one(&state.db).await?.is_some()) + .one(&state.db) + .await? + .is_some()) } /// Liste une fenêtre paginée de messages @@ -67,7 +76,11 @@ pub async fn get_all( .message_reaction .grouped_for_messages(&message_ids) .await?; - let mut attachments = state.repositories.message.attachments_for_messages(&message_ids).await?; + let mut attachments = state + .repositories + .message + .attachments_for_messages(&message_ids) + .await?; let oldest_id = page.messages.first().map(|message| message.id); let newest_id = page.messages.last().map(|message| message.id); @@ -124,11 +137,20 @@ pub async fn get_by_id( .await? .remove(&id) .unwrap_or_default(); - let attachments = state.repositories.message.attachments_for_messages(&[id]).await?.remove(&id).unwrap_or_default(); + let attachments = state + .repositories + .message + .attachments_for_messages(&[id]) + .await? + .remove(&id) + .unwrap_or_default(); - Ok(Json( - mapper::message_model_to_message_response_with_data(message, None, reactions, attachments), - )) + Ok(Json(mapper::message_model_to_message_response_with_data( + message, + None, + reactions, + attachments, + ))) } /// Crée un nouveau message @@ -163,7 +185,9 @@ pub async fn create( } if payload.content.trim().is_empty() && payload.file_ids.is_empty() { - return Err(HTTPError::BadRequest("content or at least one file is required".into())); + return Err(HTTPError::BadRequest( + "content or at least one file is required".into(), + )); } // Optionnel: vérifier reply_to_id @@ -181,10 +205,22 @@ pub async fn create( let message = state .services .message - .create_message_with_attachments(payload.channel_id, user.id, payload.content, payload.file_ids, payload.reply_to_id) + .create_message_with_attachments( + payload.channel_id, + user.id, + payload.content, + payload.file_ids, + payload.reply_to_id, + ) .await .map_err(|error| HTTPError::BadRequest(error.to_string()))?; - let attachments = state.repositories.message.attachments_for_messages(&[message.id]).await?.remove(&message.id).unwrap_or_default(); + let attachments = state + .repositories + .message + .attachments_for_messages(&[message.id]) + .await? + .remove(&message.id) + .unwrap_or_default(); Ok(( StatusCode::CREATED, Json(mapper::message_model_to_message_response_with_data( @@ -247,10 +283,19 @@ pub async fn update( .await? .remove(&id) .unwrap_or_default(); - let attachments = state.repositories.message.attachments_for_messages(&[id]).await?.remove(&id).unwrap_or_default(); - Ok(Json( - mapper::message_model_to_message_response_with_data(message, None, reactions, attachments), - )) + let attachments = state + .repositories + .message + .attachments_for_messages(&[id]) + .await? + .remove(&id) + .unwrap_or_default(); + Ok(Json(mapper::message_model_to_message_response_with_data( + message, + None, + reactions, + attachments, + ))) } #[utoipa::path( @@ -273,8 +318,15 @@ pub async fn add_reaction( Path(message_id): Path, Json(payload): Json, ) -> Result<(StatusCode, Json), HTTPError> { - let message = state.repositories.message.get_by_id(message_id).await?.ok_or(HTTPError::NotFound)?; - if !can_access(&state, message.channel_id, user.id).await? { return Err(HTTPError::Forbidden); } + let message = state + .repositories + .message + .get_by_id(message_id) + .await? + .ok_or(HTTPError::NotFound)?; + if !can_access(&state, message.channel_id, user.id).await? { + return Err(HTTPError::Forbidden); + } let (reaction, created) = state .services .message_reaction @@ -312,8 +364,15 @@ pub async fn remove_reaction( Path((message_id, emoji_id)): Path<(Uuid, Uuid)>, Query(query): Query, ) -> Result { - let message = state.repositories.message.get_by_id(message_id).await?.ok_or(HTTPError::NotFound)?; - if !can_access(&state, message.channel_id, user.id).await? { return Err(HTTPError::Forbidden); } + let message = state + .repositories + .message + .get_by_id(message_id) + .await? + .ok_or(HTTPError::NotFound)?; + if !can_access(&state, message.channel_id, user.id).await? { + return Err(HTTPError::Forbidden); + } state .services .message_reaction diff --git a/src/routes/message/mapper.rs b/src/routes/message/mapper.rs index 4bd7e42..eeda343 100644 --- a/src/routes/message/mapper.rs +++ b/src/routes/message/mapper.rs @@ -3,8 +3,8 @@ use crate::domain::dto::message::{ }; use crate::domain::dto::reaction::ReactionGroupResponse; use crate::domain::dto::reaction::ReactionResponse; -use crate::models::message; use crate::models::attachment; +use crate::models::message; use crate::models::message_reaction; use crate::repositories::types::MessageFilter; use chrono::Utc; @@ -38,7 +38,10 @@ pub fn message_model_to_message_response_with_data( updated_at: model.updated_at, reply_to_id: model.reply_to_id, reactions, - attachments: attachments.into_iter().map(crate::routes::attachment::mapper::to_response).collect(), + attachments: attachments + .into_iter() + .map(crate::routes::attachment::mapper::to_response) + .collect(), } } diff --git a/src/services/emoji.rs b/src/services/emoji.rs index 4bcc856..49a596d 100644 --- a/src/services/emoji.rs +++ b/src/services/emoji.rs @@ -4,10 +4,7 @@ use crate::services::ServicesContext; use crate::services::media::PendingMediaFile; use sea_orm::{ActiveModelTrait, Set, TransactionTrait}; use sha2::{Digest, Sha256}; -use std::{ - path::Path, - sync::Arc, -}; +use std::{path::Path, sync::Arc}; use tokio::fs; use uuid::Uuid; @@ -91,10 +88,12 @@ impl EmojiService { let mut pending = PendingMediaFile::begin(root, "emoji", id, extension) .await .map_err(|e| HTTPError::InternalServerError(e.to_string()))?; - pending.write(data) + pending + .write(data) .await .map_err(|e| HTTPError::InternalServerError(e.to_string()))?; - let relative = pending.finish() + let relative = pending + .finish() .await .map_err(|e| HTTPError::InternalServerError(e.to_string()))?; Ok((relative, Self::hash(data))) diff --git a/src/services/media.rs b/src/services/media.rs index 2b38b99..e38e06f 100644 --- a/src/services/media.rs +++ b/src/services/media.rs @@ -12,7 +12,12 @@ pub struct PendingMediaFile { } impl PendingMediaFile { - pub async fn begin(root: &Path, directory: &str, id: Uuid, extension: Option<&str>) -> std::io::Result { + pub async fn begin( + root: &Path, + directory: &str, + id: Uuid, + extension: Option<&str>, + ) -> std::io::Result { let relative_directory = PathBuf::from(directory); let directory_path = root.join(&relative_directory); fs::create_dir_all(&directory_path).await?; @@ -31,7 +36,10 @@ impl PendingMediaFile { file, temporary_path, final_path, - relative_final_path: relative_directory.join(final_name).to_string_lossy().into_owned(), + relative_final_path: relative_directory + .join(final_name) + .to_string_lossy() + .into_owned(), }) } @@ -59,8 +67,16 @@ pub fn extension_from_filename(filename: &str) -> Option { .iter() .find(|candidate| lower.ends_with(&format!(".{candidate}"))) .map(|candidate| (*candidate).to_string()) - .or_else(|| Path::new(name).extension().and_then(|value| value.to_str()).map(str::to_ascii_lowercase))?; - if extension.chars().all(|character| character.is_ascii_alphanumeric() || character == '.') { + .or_else(|| { + Path::new(name) + .extension() + .and_then(|value| value.to_str()) + .map(str::to_ascii_lowercase) + })?; + if extension + .chars() + .all(|character| character.is_ascii_alphanumeric() || character == '.') + { Some(extension) } else { None @@ -117,7 +133,10 @@ mod tests { #[test] fn preserves_compound_extensions() { - assert_eq!(extension_from_filename("archive.tar.gz").as_deref(), Some("tar.gz")); + assert_eq!( + extension_from_filename("archive.tar.gz").as_deref(), + Some("tar.gz") + ); } #[test] diff --git a/src/services/message.rs b/src/services/message.rs index 220a3a8..d7c5caf 100644 --- a/src/services/message.rs +++ b/src/services/message.rs @@ -4,7 +4,9 @@ use crate::domain::events::message::{ use crate::models::{attachment, channel, message}; use crate::services::ServicesContext; use event_bus::Scope; -use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set, TransactionTrait}; +use sea_orm::{ + ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set, TransactionTrait, +}; use std::sync::Arc; use uuid::Uuid; @@ -87,14 +89,23 @@ impl MessageService { .await? }; if files.len() != file_ids.len() - || files.iter().any(|file| file.channel_id != channel_id || file.user_id != author_id || file.message_id.is_some()) + || files.iter().any(|file| { + file.channel_id != channel_id + || file.user_id != author_id + || file.message_id.is_some() + }) { return Err(anyhow::anyhow!("Invalid or already used attachment")); } let msg = message::ActiveModel { - channel_id: Set(channel_id), user_id: Set(author_id), content: Set(content), - reply_to_id: Set(reply_to_id), ..Default::default() - }.insert(&txn).await?; + channel_id: Set(channel_id), + user_id: Set(author_id), + content: Set(content), + reply_to_id: Set(reply_to_id), + ..Default::default() + } + .insert(&txn) + .await?; for file in files { let mut active: attachment::ActiveModel = file.into(); active.message_id = Set(Some(msg.id)); @@ -102,13 +113,25 @@ impl MessageService { } txn.commit().await?; let server_id = channel::Entity::find_by_id(msg.channel_id) - .select_only().column(channel::Column::ServerId) - .into_tuple::>().one(db).await?.flatten(); + .select_only() + .column(channel::Column::ServerId) + .into_tuple::>() + .one(db) + .await? + .flatten(); let mut scopes = vec![Scope::uuid("channel", msg.channel_id)]; - if let Some(server_id) = server_id { scopes.push(Scope::uuid("server", server_id)); } - event_bus.emit_scoped("message_created", scopes, MessageCreatedEvent { - server_id, channel_id: msg.channel_id, message: msg.clone(), - }); + if let Some(server_id) = server_id { + scopes.push(Scope::uuid("server", server_id)); + } + event_bus.emit_scoped( + "message_created", + scopes, + MessageCreatedEvent { + server_id, + channel_id: msg.channel_id, + message: msg.clone(), + }, + ); Ok(msg) } diff --git a/src/services/mod.rs b/src/services/mod.rs index c0f0e2e..f0297cf 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -16,9 +16,9 @@ use std::sync::{Arc, OnceLock}; pub mod category; pub mod channel; pub mod emoji; +pub mod media; pub mod message; pub mod message_reaction; -pub mod media; mod permission; pub mod permission_sync; pub mod realtime_registry;