This commit is contained in:
2026-08-29 14:05:43 +02:00
parent 5e97e9c223
commit 88ad3c389f
15 changed files with 423 additions and 109 deletions
+3 -1
View File
@@ -100,7 +100,9 @@ impl App {
let mut interval = tokio::time::interval(Duration::from_secs(3600)); let mut interval = tokio::time::interval(Duration::from_secs(3600));
loop { loop {
interval.tick().await; 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"); tracing::warn!(%error, "Attachment cleanup failed");
} }
} }
+1 -1
View File
@@ -1,5 +1,5 @@
use crate::domain::dto::reaction::ReactionGroupResponse;
use crate::domain::dto::attachment::AttachmentResponse; use crate::domain::dto::attachment::AttachmentResponse;
use crate::domain::dto::reaction::ReactionGroupResponse;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use utoipa::ToSchema; use utoipa::ToSchema;
+5 -2
View File
@@ -1,6 +1,6 @@
use super::types::MessageFilter; use super::types::MessageFilter;
use crate::models::message;
use crate::models::attachment; use crate::models::attachment;
use crate::models::message;
use crate::repositories::{AnyResult, RepositoryContext}; use crate::repositories::{AnyResult, RepositoryContext};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect}; use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use std::collections::HashMap; use std::collections::HashMap;
@@ -46,7 +46,10 @@ impl MessageRepository {
let mut grouped = HashMap::new(); let mut grouped = HashMap::new();
for item in items { for item in items {
if let Some(message_id) = item.message_id { 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) Ok(grouped)
+83 -21
View File
@@ -23,8 +23,12 @@ pub async fn cleanup_expired(state: &AppState) -> Result<(), HTTPError> {
.all(&state.db) .all(&state.db)
.await?; .await?;
for item in pending { for item in pending {
let _ = tokio::fs::remove_file(PathBuf::from(&state.config.media.root).join(&item.file_path)).await; let _ =
attachment::Entity::delete_by_id(item.id).exec(&state.db).await?; 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( media::cleanup_temporary_files(
PathBuf::from(&state.config.media.root).as_path(), 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(); let name = field.name().unwrap_or_default().to_string();
if name == "channel_id" { 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::<Uuid>().map_err(HTTPError::UuidError)?); channel_id = Some(value.parse::<Uuid>().map_err(HTTPError::UuidError)?);
} else if name == "files" || name == "file" { } 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 filename = field.file_name().unwrap_or("unnamed").to_string();
let mime_type = field.content_type().unwrap_or("application/octet-stream").to_string(); let mime_type = field
if !can_access(&state, channel, user.id).await? { return Err(HTTPError::Forbidden); } .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 id = Uuid::new_v4();
let mut output = PendingMediaFile::begin( let mut output = PendingMediaFile::begin(
PathBuf::from(&state.config.media.root).as_path(), PathBuf::from(&state.config.media.root).as_path(),
@@ -75,35 +88,84 @@ pub async fn create(
.await .await
.map_err(|error| HTTPError::InternalServerError(error.to_string()))?; .map_err(|error| HTTPError::InternalServerError(error.to_string()))?;
let mut size = 0_i64; 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; 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 { let model = attachment::ActiveModel {
id: Set(id), message_id: Set(None), channel_id: Set(channel), user_id: Set(user.id), id: Set(id),
filename: Set(filename), file_size: Set(size), mime_type: Set(mime_type), file_path: Set(file_path.clone()), created_at: Set(Utc::now()), 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 { match model.insert(&state.db).await {
Ok(item) => created.push(item), 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()))?; let channel_id =
if !can_access(&state, channel_id, user.id).await? { return Err(HTTPError::Forbidden); } channel_id.ok_or_else(|| HTTPError::BadRequest("channel_id is required".into()))?;
if created.is_empty() { return Err(HTTPError::BadRequest("at least one file is required".into())); } if !can_access(&state, channel_id, user.id).await? {
Ok((StatusCode::CREATED, axum::Json(AttachmentUploadResponse { attachments: created.into_iter().map(mapper::to_response).collect() }))) 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<AppState>, Path(id): Path<Uuid>) -> Result<Response, HTTPError> { pub async fn file(
let item = attachment::Entity::find_by_id(id).one(&state.db).await?.ok_or(HTTPError::NotFound)?; State(state): State<AppState>,
let bytes = tokio::fs::read(PathBuf::from(&state.config.media.root).join(&item.file_path)).await.map_err(|_| HTTPError::NotFound)?; Path(id): Path<Uuid>,
) -> Result<Response, HTTPError> {
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)); 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/") { if !item.mime_type.starts_with("image/") {
let safe_name = item.filename.replace(['\"', '\r', '\n'], "_"); 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) Ok(response)
} }
+6 -2
View File
@@ -1,6 +1,10 @@
use axum::{Router, routing::{get, post}, extract::DefaultBodyLimit};
use crate::core::state::AppState;
use super::handlers; use super::handlers;
use crate::core::state::AppState;
use axum::{
Router,
extract::DefaultBodyLimit,
routing::{get, post},
};
pub fn secure_router() -> Router<AppState> { pub fn secure_router() -> Router<AppState> {
Router::new() Router::new()
+21 -5
View File
@@ -13,14 +13,30 @@ use axum::{
extract::{Path, Query, State}, extract::{Path, Query, State},
http::StatusCode, http::StatusCode,
}; };
use uuid::Uuid;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
use uuid::Uuid;
async fn require_channel_member(state: &AppState, channel_id: Uuid, user_id: Uuid) -> Result<(), HTTPError> { async fn require_channel_member(
let channel = state.repositories.channel.get_by_id(channel_id).await?.ok_or(HTTPError::NotFound)?; 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 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() && channel_user::Entity::find()
{ return Err(HTTPError::Forbidden); } .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(()) Ok(())
} }
+151 -30
View File
@@ -6,70 +6,178 @@ use crate::domain::dto::conversation::{
use crate::http::context::CurrentUser; use crate::http::context::CurrentUser;
use crate::http::error::HTTPError; use crate::http::error::HTTPError;
use crate::models::{channel, channel_user, message, user}; use crate::models::{channel, channel_user, message, user};
use axum::{Json, extract::{Path, State}}; use axum::{
Json,
extract::{Path, State},
};
use chrono::Utc; 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 std::collections::HashSet;
use uuid::Uuid; use uuid::Uuid;
async fn member_ids(state: &AppState, channel_id: Uuid) -> Result<Vec<Uuid>, HTTPError> { async fn member_ids(state: &AppState, channel_id: Uuid) -> Result<Vec<Uuid>, HTTPError> {
Ok(channel_user::Entity::find() Ok(channel_user::Entity::find()
.filter(channel_user::Column::ChannelId.eq(channel_id)) .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<ConversationResponse, HTTPError> { async fn response(
state: &AppState,
current_user_id: Uuid,
channel: channel::Model,
) -> Result<ConversationResponse, HTTPError> {
let ids = member_ids(state, channel.id).await?; 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 users = user::Entity::find()
let mut participants: Vec<_> = users.into_iter().map(|u| ConversationParticipantResponse { id: u.id, username: u.username }).collect(); .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())); 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::<Vec<_>>().join(", "); let title = participants
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); .iter()
let unread_count = state.repositories.read_state.unread_counts(&[channel.id], current_user_id).await?.get(&channel.id).copied().unwrap_or(0); .filter(|p| p.id != current_user_id)
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 }) .map(|p| p.username.clone())
.collect::<Vec<_>>()
.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<channel::Model, HTTPError> { async fn create_channel(state: &AppState, ids: &[Uuid]) -> Result<channel::Model, HTTPError> {
let txn = state.db.begin().await?; 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 { 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?; 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()); state.event_bus.emit("channel_created", channel.clone());
Ok(channel) Ok(channel)
} }
async fn validate_ids(state: &AppState, ids: &[Uuid]) -> Result<(), HTTPError> { 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())); } if ids.is_empty() {
let found = user::Entity::find().filter(user::Column::Id.is_in(ids.to_vec())).all(&state.db).await?; return Err(HTTPError::BadRequest(
if found.len() != ids.iter().collect::<HashSet<_>>().len() { return Err(HTTPError::BadRequest("Unknown participant".into())); } "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::<HashSet<_>>().len() {
return Err(HTTPError::BadRequest("Unknown participant".into()));
}
Ok(()) Ok(())
} }
#[utoipa::path(get, path = "/conversations", responses((status = 200, body = [ConversationResponse])), tag = "Conversations", security(("bearerAuth" = [])))] #[utoipa::path(get, path = "/conversations", responses((status = 200, body = [ConversationResponse])), tag = "Conversations", security(("bearerAuth" = [])))]
pub async fn list(user: CurrentUser, State(state): State<AppState>) -> Result<Json<Vec<ConversationResponse>>, HTTPError> { pub async fn list(
let memberships = channel_user::Entity::find().filter(channel_user::Column::UserId.eq(user.id)).all(&state.db).await?; user: CurrentUser,
State(state): State<AppState>,
) -> Result<Json<Vec<ConversationResponse>>, 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(); let ids: Vec<_> = memberships.into_iter().map(|m| m.channel_id).collect();
if ids.is_empty() { return Ok(Json(Vec::new())); } if ids.is_empty() {
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?; 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()); 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)) Ok(Json(result))
} }
#[utoipa::path(post, path = "/conversations", request_body = CreateConversationRequest, responses((status = 200, body = ConversationResponse)), tag = "Conversations", security(("bearerAuth" = [])))] #[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<AppState>, Json(payload): Json<CreateConversationRequest>) -> Result<Json<ConversationResponse>, HTTPError> { pub async fn create(
user: CurrentUser,
State(state): State<AppState>,
Json(payload): Json<CreateConversationRequest>,
) -> Result<Json<ConversationResponse>, HTTPError> {
let mut ids = payload.user_ids; let mut ids = payload.user_ids;
ids.push(user.id); ids.push(user.id);
ids.sort_unstable(); ids.dedup(); ids.sort_unstable();
ids.dedup();
validate_ids(&state, &ids).await?; 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 { 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(); 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?; let channel = create_channel(&state, &ids).await?;
@@ -77,12 +185,25 @@ pub async fn create(user: CurrentUser, State(state): State<AppState>, 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" = [])))] #[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<AppState>, Path(id): Path<Uuid>, Json(payload): Json<ForkConversationRequest>) -> Result<Json<ConversationResponse>, HTTPError> { pub async fn fork(
let source = channel::Entity::find_by_id(id).one(&state.db).await?.ok_or(HTTPError::NotFound)?; user: CurrentUser,
if source.channel_type != channel::ChannelType::DM || !member_ids(&state, id).await?.contains(&user.id) { return Err(HTTPError::Forbidden); } State(state): State<AppState>,
Path(id): Path<Uuid>,
Json(payload): Json<ForkConversationRequest>,
) -> Result<Json<ConversationResponse>, 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?; let mut ids = member_ids(&state, id).await?;
ids.extend(payload.user_ids); ids.extend(payload.user_ids);
ids.sort_unstable(); ids.dedup(); ids.sort_unstable();
ids.dedup();
validate_ids(&state, &ids).await?; validate_ids(&state, &ids).await?;
let channel = create_channel(&state, &ids).await?; let channel = create_channel(&state, &ids).await?;
Ok(Json(response(&state, user.id, channel).await?)) Ok(Json(response(&state, user.id, channel).await?))
+4 -1
View File
@@ -1,6 +1,9 @@
use super::handlers; use super::handlers;
use crate::core::state::AppState; use crate::core::state::AppState;
use axum::{routing::{get, post}, Router}; use axum::{
Router,
routing::{get, post},
};
pub fn router() -> Router<AppState> { pub fn router() -> Router<AppState> {
Router::new() Router::new()
+1 -1
View File
@@ -1,7 +1,7 @@
use super::handlers; use super::handlers;
use crate::core::AppState; use crate::core::AppState;
use axum::Router;
use axum::routing::get; use axum::routing::get;
use axum::Router;
pub fn router() -> Router<AppState> { pub fn router() -> Router<AppState> {
Router::new().route("/gateway", get(handlers::ws_handler)) Router::new().route("/gateway", get(handlers::ws_handler))
+79 -20
View File
@@ -13,11 +13,18 @@ use axum::{
extract::{Path, Query, State}, extract::{Path, Query, State},
http::StatusCode, http::StatusCode,
}; };
use uuid::Uuid;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
use uuid::Uuid;
pub(crate) async fn can_access(state: &AppState, channel_id: Uuid, user_id: Uuid) -> Result<bool, HTTPError> { pub(crate) async fn can_access(
let Some(channel) = channel::Entity::find_by_id(channel_id).one(&state.db).await? else { state: &AppState,
channel_id: Uuid,
user_id: Uuid,
) -> Result<bool, HTTPError> {
let Some(channel) = channel::Entity::find_by_id(channel_id)
.one(&state.db)
.await?
else {
return Ok(false); return Ok(false);
}; };
if channel.channel_type != channel::ChannelType::DM { 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() Ok(channel_user::Entity::find()
.filter(channel_user::Column::ChannelId.eq(channel_id)) .filter(channel_user::Column::ChannelId.eq(channel_id))
.filter(channel_user::Column::UserId.eq(user_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 /// Liste une fenêtre paginée de messages
@@ -67,7 +76,11 @@ pub async fn get_all(
.message_reaction .message_reaction
.grouped_for_messages(&message_ids) .grouped_for_messages(&message_ids)
.await?; .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 oldest_id = page.messages.first().map(|message| message.id);
let newest_id = page.messages.last().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? .await?
.remove(&id) .remove(&id)
.unwrap_or_default(); .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( Ok(Json(mapper::message_model_to_message_response_with_data(
mapper::message_model_to_message_response_with_data(message, None, reactions, attachments), message,
)) None,
reactions,
attachments,
)))
} }
/// Crée un nouveau message /// Crée un nouveau message
@@ -163,7 +185,9 @@ pub async fn create(
} }
if payload.content.trim().is_empty() && payload.file_ids.is_empty() { 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 // Optionnel: vérifier reply_to_id
@@ -181,10 +205,22 @@ pub async fn create(
let message = state let message = state
.services .services
.message .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 .await
.map_err(|error| HTTPError::BadRequest(error.to_string()))?; .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(( Ok((
StatusCode::CREATED, StatusCode::CREATED,
Json(mapper::message_model_to_message_response_with_data( Json(mapper::message_model_to_message_response_with_data(
@@ -247,10 +283,19 @@ pub async fn update(
.await? .await?
.remove(&id) .remove(&id)
.unwrap_or_default(); .unwrap_or_default();
let attachments = state.repositories.message.attachments_for_messages(&[id]).await?.remove(&id).unwrap_or_default(); let attachments = state
Ok(Json( .repositories
mapper::message_model_to_message_response_with_data(message, None, reactions, attachments), .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( #[utoipa::path(
@@ -273,8 +318,15 @@ pub async fn add_reaction(
Path(message_id): Path<Uuid>, Path(message_id): Path<Uuid>,
Json(payload): Json<CreateReactionRequest>, Json(payload): Json<CreateReactionRequest>,
) -> Result<(StatusCode, Json<ReactionResponse>), HTTPError> { ) -> Result<(StatusCode, Json<ReactionResponse>), HTTPError> {
let message = state.repositories.message.get_by_id(message_id).await?.ok_or(HTTPError::NotFound)?; let message = state
if !can_access(&state, message.channel_id, user.id).await? { return Err(HTTPError::Forbidden); } .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 let (reaction, created) = state
.services .services
.message_reaction .message_reaction
@@ -312,8 +364,15 @@ pub async fn remove_reaction(
Path((message_id, emoji_id)): Path<(Uuid, Uuid)>, Path((message_id, emoji_id)): Path<(Uuid, Uuid)>,
Query(query): Query<DeleteReactionQuery>, Query(query): Query<DeleteReactionQuery>,
) -> Result<StatusCode, HTTPError> { ) -> Result<StatusCode, HTTPError> {
let message = state.repositories.message.get_by_id(message_id).await?.ok_or(HTTPError::NotFound)?; let message = state
if !can_access(&state, message.channel_id, user.id).await? { return Err(HTTPError::Forbidden); } .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 state
.services .services
.message_reaction .message_reaction
+5 -2
View File
@@ -3,8 +3,8 @@ use crate::domain::dto::message::{
}; };
use crate::domain::dto::reaction::ReactionGroupResponse; use crate::domain::dto::reaction::ReactionGroupResponse;
use crate::domain::dto::reaction::ReactionResponse; use crate::domain::dto::reaction::ReactionResponse;
use crate::models::message;
use crate::models::attachment; use crate::models::attachment;
use crate::models::message;
use crate::models::message_reaction; use crate::models::message_reaction;
use crate::repositories::types::MessageFilter; use crate::repositories::types::MessageFilter;
use chrono::Utc; use chrono::Utc;
@@ -38,7 +38,10 @@ pub fn message_model_to_message_response_with_data(
updated_at: model.updated_at, updated_at: model.updated_at,
reply_to_id: model.reply_to_id, reply_to_id: model.reply_to_id,
reactions, 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(),
} }
} }
+5 -6
View File
@@ -4,10 +4,7 @@ use crate::services::ServicesContext;
use crate::services::media::PendingMediaFile; use crate::services::media::PendingMediaFile;
use sea_orm::{ActiveModelTrait, Set, TransactionTrait}; use sea_orm::{ActiveModelTrait, Set, TransactionTrait};
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use std::{ use std::{path::Path, sync::Arc};
path::Path,
sync::Arc,
};
use tokio::fs; use tokio::fs;
use uuid::Uuid; use uuid::Uuid;
@@ -91,10 +88,12 @@ impl EmojiService {
let mut pending = PendingMediaFile::begin(root, "emoji", id, extension) let mut pending = PendingMediaFile::begin(root, "emoji", id, extension)
.await .await
.map_err(|e| HTTPError::InternalServerError(e.to_string()))?; .map_err(|e| HTTPError::InternalServerError(e.to_string()))?;
pending.write(data) pending
.write(data)
.await .await
.map_err(|e| HTTPError::InternalServerError(e.to_string()))?; .map_err(|e| HTTPError::InternalServerError(e.to_string()))?;
let relative = pending.finish() let relative = pending
.finish()
.await .await
.map_err(|e| HTTPError::InternalServerError(e.to_string()))?; .map_err(|e| HTTPError::InternalServerError(e.to_string()))?;
Ok((relative, Self::hash(data))) Ok((relative, Self::hash(data)))
+24 -5
View File
@@ -12,7 +12,12 @@ pub struct PendingMediaFile {
} }
impl PendingMediaFile { impl PendingMediaFile {
pub async fn begin(root: &Path, directory: &str, id: Uuid, extension: Option<&str>) -> std::io::Result<Self> { pub async fn begin(
root: &Path,
directory: &str,
id: Uuid,
extension: Option<&str>,
) -> std::io::Result<Self> {
let relative_directory = PathBuf::from(directory); let relative_directory = PathBuf::from(directory);
let directory_path = root.join(&relative_directory); let directory_path = root.join(&relative_directory);
fs::create_dir_all(&directory_path).await?; fs::create_dir_all(&directory_path).await?;
@@ -31,7 +36,10 @@ impl PendingMediaFile {
file, file,
temporary_path, temporary_path,
final_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<String> {
.iter() .iter()
.find(|candidate| lower.ends_with(&format!(".{candidate}"))) .find(|candidate| lower.ends_with(&format!(".{candidate}")))
.map(|candidate| (*candidate).to_string()) .map(|candidate| (*candidate).to_string())
.or_else(|| Path::new(name).extension().and_then(|value| value.to_str()).map(str::to_ascii_lowercase))?; .or_else(|| {
if extension.chars().all(|character| character.is_ascii_alphanumeric() || character == '.') { 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) Some(extension)
} else { } else {
None None
@@ -117,7 +133,10 @@ mod tests {
#[test] #[test]
fn preserves_compound_extensions() { 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] #[test]
+34 -11
View File
@@ -4,7 +4,9 @@ use crate::domain::events::message::{
use crate::models::{attachment, channel, message}; use crate::models::{attachment, channel, message};
use crate::services::ServicesContext; use crate::services::ServicesContext;
use event_bus::Scope; 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 std::sync::Arc;
use uuid::Uuid; use uuid::Uuid;
@@ -87,14 +89,23 @@ impl MessageService {
.await? .await?
}; };
if files.len() != file_ids.len() 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")); return Err(anyhow::anyhow!("Invalid or already used attachment"));
} }
let msg = message::ActiveModel { let msg = message::ActiveModel {
channel_id: Set(channel_id), user_id: Set(author_id), content: Set(content), channel_id: Set(channel_id),
reply_to_id: Set(reply_to_id), ..Default::default() user_id: Set(author_id),
}.insert(&txn).await?; content: Set(content),
reply_to_id: Set(reply_to_id),
..Default::default()
}
.insert(&txn)
.await?;
for file in files { for file in files {
let mut active: attachment::ActiveModel = file.into(); let mut active: attachment::ActiveModel = file.into();
active.message_id = Set(Some(msg.id)); active.message_id = Set(Some(msg.id));
@@ -102,13 +113,25 @@ impl MessageService {
} }
txn.commit().await?; txn.commit().await?;
let server_id = channel::Entity::find_by_id(msg.channel_id) let server_id = channel::Entity::find_by_id(msg.channel_id)
.select_only().column(channel::Column::ServerId) .select_only()
.into_tuple::<Option<Uuid>>().one(db).await?.flatten(); .column(channel::Column::ServerId)
.into_tuple::<Option<Uuid>>()
.one(db)
.await?
.flatten();
let mut scopes = vec![Scope::uuid("channel", msg.channel_id)]; let mut scopes = vec![Scope::uuid("channel", msg.channel_id)];
if let Some(server_id) = server_id { scopes.push(Scope::uuid("server", server_id)); } if let Some(server_id) = server_id {
event_bus.emit_scoped("message_created", scopes, MessageCreatedEvent { scopes.push(Scope::uuid("server", server_id));
server_id, channel_id: msg.channel_id, message: msg.clone(), }
}); event_bus.emit_scoped(
"message_created",
scopes,
MessageCreatedEvent {
server_id,
channel_id: msg.channel_id,
message: msg.clone(),
},
);
Ok(msg) Ok(msg)
} }
+1 -1
View File
@@ -16,9 +16,9 @@ use std::sync::{Arc, OnceLock};
pub mod category; pub mod category;
pub mod channel; pub mod channel;
pub mod emoji; pub mod emoji;
pub mod media;
pub mod message; pub mod message;
pub mod message_reaction; pub mod message_reaction;
pub mod media;
mod permission; mod permission;
pub mod permission_sync; pub mod permission_sync;
pub mod realtime_registry; pub mod realtime_registry;