oxpanel/oxspeak_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
Nell
2026-08-29 18:39:28 +02:00
parent 88ad3c389f
commit 9551e90678
17 changed files with 642 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/routes/gateway/mod.rs
+331 -205
View File
@@ -1,37 +1,40 @@
use crate::domain::events::channel::{
ChannelCreatedEvent, ChannelDeletedEvent, ChannelUpdatedEvent,
};
use crate::domain::events::emoji::{EmojiCreatedEvent, EmojiDeletedEvent, EmojiUpdatedEvent};
use crate::domain::events::message::{
MessageCreatedEvent, MessageDeletedEvent, MessageReactionAddedEvent,
MessageReactionRemovedEvent, MessageUpdatedEvent,
};
use crate::models::user::Model as User;
use crate::models::{category, server, server_user, user};
use crate::repositories::Repositories;
use crate::routes::category::mapper::category_model_to_category_response;
use crate::routes::channel::mapper::channel_model_to_channel_response;
use crate::routes::message::mapper::{
message_model_to_message_response_with_data,
message_model_to_message_response_with_reactions,
message_model_to_message_response_with_data, message_model_to_message_response_with_reactions,
reaction_model_to_response,
};
use crate::routes::server::mapper::server_model_to_server_response;
use crate::services::Services;
use crate::repositories::Repositories;
use axum::extract::ws::Message;
use event_bus::EventBus;
use events::GatewayEvent;
use parking_lot::RwLock;
use std::collections::HashMap;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QuerySelect};
use serde::Serialize;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use uuid::Uuid;
pub mod events;
pub mod handlers;
pub mod routes;
/// Couche de transport WebSocket. Elle ne connaît pas les permissions.
#[derive(Debug)]
pub struct GatewayManager {
pub clients: RwLock<HashMap<ConnectionKey, GatewayClient>>,
services: Arc<Services>,
repositories: Arc<Repositories>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -42,59 +45,119 @@ pub struct ConnectionKey {
#[derive(Debug, Clone)]
pub struct GatewayClient {
user: User,
user: crate::models::user::Model,
connection_id: Uuid,
pub sender: mpsc::UnboundedSender<Message>,
pub event_bus: Arc<EventBus>,
_event_handles: Vec<Arc<JoinHandle<()>>>,
}
impl GatewayManager {
pub fn new(services: Arc<Services>, repositories: Arc<Repositories>) -> Self {
pub fn new() -> Self {
Self {
clients: RwLock::new(HashMap::new()),
}
}
pub(crate) fn add_client(&self, client: GatewayClient) {
self.clients.write().insert(client.key(), client);
}
pub(crate) fn remove_client(&self, client: &GatewayClient) {
self.clients.write().remove(&client.key());
}
pub(crate) fn send_to_users<T: Serialize>(
&self,
users: impl IntoIterator<Item = Uuid>,
namespace: &'static str,
action: &'static str,
content: T,
) {
let Ok(json) = serde_json::to_string(&GatewayEvent {
namespace,
action,
content,
}) else {
return;
};
let users: HashSet<_> = users.into_iter().collect();
for (key, client) in self.clients.read().iter() {
if users.contains(&key.user_id) {
let _ = client.sender.send(Message::Text(json.clone().into()));
}
}
}
}
/// Résout les audiences puis délègue l'envoi au transport WebSocket.
pub struct RealtimeRouter {
gateway: Arc<GatewayManager>,
services: Arc<Services>,
repositories: Arc<Repositories>,
}
impl RealtimeRouter {
pub fn new(
gateway: Arc<GatewayManager>,
services: Arc<Services>,
repositories: Arc<Repositories>,
) -> Self {
Self {
gateway,
services,
repositories,
}
}
/// Démarre les routeurs centraux des événements de messages.
pub fn start(self: &Arc<Self>, event_bus: Arc<EventBus>) {
let manager = Arc::clone(self);
let router = Arc::clone(self);
event_bus.on_async::<MessageCreatedEvent, _, _>("message_created", move |event| {
let manager = Arc::clone(&manager);
let router = Arc::clone(&router);
async move {
let message_id = event.message.id;
let attachments = manager
let attachments = router
.repositories
.message
.attachments_for_messages(&[message_id])
.attachments_for_messages(&[event.message.id])
.await
.ok()
.and_then(|mut items| items.remove(&message_id))
.and_then(|mut items| items.remove(&event.message.id))
.unwrap_or_default();
manager.broadcast_message(
event.channel_id,
let users = router
.services
.realtime_registry
.users_for_channel(event.channel_id);
router.gateway.send_to_users(
users,
"Message",
"add",
message_model_to_message_response_with_data(event.message, event.server_id, Vec::new(), attachments),
message_model_to_message_response_with_data(
event.message,
event.server_id,
Vec::new(),
attachments,
),
);
}
});
let manager = Arc::clone(self);
let router = Arc::clone(self);
event_bus.on_async::<MessageUpdatedEvent, _, _>("message_updated", move |event| {
let manager = Arc::clone(&manager);
let router = Arc::clone(&router);
async move {
let message_id = event.message.id;
let reactions = manager
let reactions = router
.services
.message_reaction
.grouped_for_messages(&[message_id])
.grouped_for_messages(&[event.message.id])
.await
.ok()
.and_then(|mut groups| groups.remove(&message_id))
.and_then(|mut groups| groups.remove(&event.message.id))
.unwrap_or_default();
manager.broadcast_message(
event.channel_id,
let users = router
.services
.realtime_registry
.users_for_channel(event.channel_id);
router.gateway.send_to_users(
users,
"Message",
"update",
message_model_to_message_response_with_reactions(
event.message,
@@ -105,22 +168,33 @@ impl GatewayManager {
}
});
let manager = Arc::clone(self);
let router = Arc::clone(self);
event_bus.on_async::<MessageDeletedEvent, _, _>("message_deleted", move |event| {
let manager = Arc::clone(&manager);
let router = Arc::clone(&router);
async move {
manager.broadcast_message(event.channel_id, "remove", event.message.id);
let users = router
.services
.realtime_registry
.users_for_channel(event.channel_id);
router
.gateway
.send_to_users(users, "Message", "remove", event.message.id);
}
});
let manager = Arc::clone(self);
let router = Arc::clone(self);
event_bus.on_async::<MessageReactionAddedEvent, _, _>(
"message_reaction_added",
move |event| {
let manager = Arc::clone(&manager);
let router = Arc::clone(&router);
async move {
manager.broadcast_reaction(
event.channel_id,
let users = router
.services
.realtime_registry
.users_for_channel(event.channel_id);
router.gateway.send_to_users(
users,
"Reaction",
"add",
reaction_model_to_response(event.reaction),
);
@@ -128,99 +202,248 @@ impl GatewayManager {
},
);
let manager = Arc::clone(self);
let router = Arc::clone(self);
event_bus.on_async::<MessageReactionRemovedEvent, _, _>(
"message_reaction_removed",
move |event| {
let manager = Arc::clone(&manager);
let router = Arc::clone(&router);
async move {
manager.broadcast_reaction(
event.channel_id,
let users = router
.services
.realtime_registry
.users_for_channel(event.channel_id);
router.gateway.send_to_users(
users,
"Reaction",
"remove",
reaction_model_to_response(event.reaction),
);
}
},
);
}
pub(crate) fn add_client(&self, gateway_client: GatewayClient) {
let key = gateway_client.key();
self.clients.write().insert(key, gateway_client);
}
pub(crate) fn remove_client(&self, gateway_client: &GatewayClient) {
let key = gateway_client.key();
self.clients.write().remove(&key);
}
fn broadcast_message<T: serde::Serialize>(
&self,
channel_id: Uuid,
action: &'static str,
content: T,
) {
let event = GatewayEvent {
namespace: "Message",
action,
content,
};
let Ok(json) = serde_json::to_string(&event) else {
return;
};
let users = self
.services
.realtime_registry
.users_for_channel(channel_id);
let clients = self.clients.read();
for (key, client) in clients.iter() {
if users.contains(&key.user_id) {
let _ = client.sender.send(Message::Text(json.clone().into()));
let router = Arc::clone(self);
event_bus.on_async::<ChannelCreatedEvent, _, _>("channel_created", move |event| {
let router = Arc::clone(&router);
async move {
if let Err(error) = router
.services
.realtime_registry
.refresh_channel(&router.repositories, event.channel.id)
.await
{
tracing::error!(channel_id = %event.channel.id, ?error, "Unable to refresh channel audience");
return;
}
let users = router
.services
.realtime_registry
.users_for_channel(event.channel.id);
router.gateway.send_to_users(
users,
"Channel",
"add",
channel_model_to_channel_response(event.channel),
);
}
}
});
let router = Arc::clone(self);
event_bus.on_async::<ChannelUpdatedEvent, _, _>("channel_updated", move |event| {
let router = Arc::clone(&router);
async move {
let mut users = router
.services
.realtime_registry
.users_for_channel(event.channel.id);
users.extend(
router
.services
.realtime_registry
.users_for_channel(event.previous.id),
);
router.gateway.send_to_users(
users,
"Channel",
"update",
channel_model_to_channel_response(event.channel),
);
}
});
let router = Arc::clone(self);
event_bus.on_async::<ChannelDeletedEvent, _, _>("channel_deleted", move |event| {
let router = Arc::clone(&router);
async move {
let users = router
.services
.realtime_registry
.users_for_channel(event.channel.id);
router
.gateway
.send_to_users(users, "Channel", "remove", event.channel.id);
router
.services
.realtime_registry
.remove_channel(event.channel.id);
}
});
let router = Arc::clone(self);
event_bus.on_async::<category::Model, _, _>("category_created", move |category| {
let router = Arc::clone(&router);
async move {
let users = router.server_users(category.server_id).await;
router.gateway.send_to_users(
users,
"Category",
"add",
category_model_to_category_response(category),
);
}
});
let router = Arc::clone(self);
event_bus.on_async::<category::Model, _, _>("category_updated", move |category| {
let router = Arc::clone(&router);
async move {
let users = router.server_users(category.server_id).await;
router.gateway.send_to_users(
users,
"Category",
"update",
category_model_to_category_response(category),
);
}
});
let router = Arc::clone(self);
event_bus.on_async::<category::Model, _, _>("category_deleted", move |category| {
let router = Arc::clone(&router);
async move {
let users = router.server_users(category.server_id).await;
router
.gateway
.send_to_users(users, "Category", "remove", category.id);
}
});
let router = Arc::clone(self);
event_bus.on_async::<server::Model, _, _>("server_created", move |server| {
let router = Arc::clone(&router);
async move {
let users = router.server_users(server.id).await;
router.gateway.send_to_users(
users,
"Server",
"add",
server_model_to_server_response(server),
);
}
});
let router = Arc::clone(self);
event_bus.on_async::<server::Model, _, _>("server_updated", move |server| {
let router = Arc::clone(&router);
async move {
let users = router.server_users(server.id).await;
router.gateway.send_to_users(
users,
"Server",
"update",
server_model_to_server_response(server),
);
}
});
let router = Arc::clone(self);
event_bus.on_async::<(server::Model, Vec<Uuid>), _, _>(
"server_deleted",
move |(server, users)| {
let router = Arc::clone(&router);
async move {
router
.gateway
.send_to_users(users, "Server", "remove", server.id);
}
},
);
let router = Arc::clone(self);
event_bus.on_async::<EmojiCreatedEvent, _, _>("emoji_created", move |event| {
let router = Arc::clone(&router);
async move {
let users = router.emoji_users(event.emoji.server_id).await;
router.gateway.send_to_users(
users,
"Emoji",
"add",
crate::routes::emoji::mapper::response(event.emoji),
);
}
});
let router = Arc::clone(self);
event_bus.on_async::<EmojiUpdatedEvent, _, _>("emoji_updated", move |event| {
let router = Arc::clone(&router);
async move {
let mut users = router.emoji_users(event.emoji.server_id).await;
users.extend(router.emoji_users(event.previous.server_id).await);
router.gateway.send_to_users(
users,
"Emoji",
"update",
crate::routes::emoji::mapper::response(event.emoji),
);
}
});
let router = Arc::clone(self);
event_bus.on_async::<EmojiDeletedEvent, _, _>("emoji_deleted", move |event| {
let router = Arc::clone(&router);
async move {
let users = router.emoji_users(event.emoji.server_id).await;
router.gateway.send_to_users(
users,
"Emoji",
"remove",
crate::routes::emoji::mapper::response(event.emoji),
);
}
});
}
fn broadcast_reaction<T: serde::Serialize>(
&self,
channel_id: Uuid,
action: &'static str,
content: T,
) {
let event = GatewayEvent {
namespace: "Reaction",
action,
content,
};
let Ok(json) = serde_json::to_string(&event) else {
return;
};
async fn server_users(&self, server_id: Uuid) -> Vec<Uuid> {
server_user::Entity::find()
.filter(server_user::Column::ServerId.eq(server_id))
.select_only()
.column(server_user::Column::UserId)
.into_tuple::<Uuid>()
.all(&self.repositories.server.context.db)
.await
.unwrap_or_default()
}
let users = self
.services
.realtime_registry
.users_for_channel(channel_id);
let clients = self.clients.read();
for (key, client) in clients.iter() {
if users.contains(&key.user_id) {
let _ = client.sender.send(Message::Text(json.clone().into()));
}
async fn emoji_users(&self, server_id: Option<Uuid>) -> Vec<Uuid> {
match server_id {
Some(server_id) => self.server_users(server_id).await,
None => user::Entity::find()
.select_only()
.column(user::Column::Id)
.into_tuple::<Uuid>()
.all(&self.repositories.server.context.db)
.await
.unwrap_or_default(),
}
}
}
impl GatewayClient {
pub fn new(
user: User,
sender: mpsc::UnboundedSender<Message>,
event_bus: Arc<EventBus>,
) -> Self {
pub fn new(user: crate::models::user::Model, sender: mpsc::UnboundedSender<Message>) -> Self {
Self {
user,
connection_id: Uuid::new_v4(),
sender,
event_bus,
_event_handles: Vec::new(),
}
}
@@ -231,113 +454,16 @@ impl GatewayClient {
}
}
fn subscribe_event<T, F, R>(
&self,
event_name: &'static str,
namespace: &'static str,
action: &'static str,
mapper: F,
) -> Arc<JoinHandle<()>>
where
T: Clone + Send + Sync + 'static,
R: serde::Serialize + Send + 'static,
F: Fn(T) -> R + Send + Sync + 'static,
{
let sender = self.sender.clone();
Arc::new(
self.event_bus
.on_async::<T, _, _>(event_name, move |payload| {
let sender = sender.clone();
let content = mapper(payload);
async move {
let event = GatewayEvent {
namespace,
action,
content,
};
if let Ok(json) = serde_json::to_string(&event) {
let _ = sender.send(Message::Text(json.into()));
}
}
}),
)
}
pub fn subscribe_to_events(&mut self) {
let mut handles = Vec::new();
// Les messages sont routés par GatewayManager selon le channel_id.
handles.push(self.subscribe_event(
"channel_created",
"Channel",
"add",
channel_model_to_channel_response,
));
handles.push(self.subscribe_event(
"channel_updated",
"Channel",
"update",
channel_model_to_channel_response,
));
handles.push(self.subscribe_event("channel_deleted", "Channel", "remove", |id: Uuid| id));
handles.push(self.subscribe_event(
"category_created",
"Category",
"add",
category_model_to_category_response,
));
handles.push(self.subscribe_event(
"category_updated",
"Category",
"update",
category_model_to_category_response,
));
handles.push(self.subscribe_event("category_deleted", "Category", "remove", |id: Uuid| id));
handles.push(self.subscribe_event(
"server_created",
"Server",
"add",
server_model_to_server_response,
));
handles.push(self.subscribe_event(
"server_updated",
"Server",
"update",
server_model_to_server_response,
));
handles.push(self.subscribe_event("server_deleted", "Server", "remove", |id: Uuid| id));
self._event_handles = handles;
}
pub fn unsubscribe_all(&mut self) {
for handle in self._event_handles.drain(..) {
handle.abort();
}
}
async fn on_connect(&mut self) {
tracing::info!(user_id = %self.user.id, "Client connected");
self.subscribe_to_events();
}
async fn on_disconnect(&mut self) {
tracing::info!(user_id = %self.user.id, "Client disconnected");
self.unsubscribe_all();
}
async fn on_message(&self, message: Message) {
match message {
Message::Binary(_) => {}
Message::Text(content) => {
tracing::info!(user_id = %self.user.id, "Received text message: {}", content);
}
Message::Ping(_) => {}
Message::Pong(_) => {}
Message::Close(_) => {}
if let Message::Text(content) = message {
tracing::info!(user_id = %self.user.id, "Received text message: {}", content);
}
}
}