3 serialize_websocket_message,
10 use anyhow::Context as acontext;
11 use background_jobs::QueueHandle;
13 r2d2::{ConnectionManager, Pool},
16 use lemmy_api_common::{comment::*, post::*};
17 use lemmy_db_schema::{source::secret::Secret, CommunityId, LocalUserId, PostId};
20 rate_limit::RateLimit,
21 settings::structs::Settings,
27 use rand::rngs::ThreadRng;
30 use serde_json::Value;
32 collections::{HashMap, HashSet},
36 use tokio::macros::support::Pin;
38 type MessageHandlerType = fn(
39 context: LemmyContext,
43 ) -> Pin<Box<dyn Future<Output = Result<String, LemmyError>> + '_>>;
45 type MessageHandlerCrudType = fn(
46 context: LemmyContext,
48 op: UserOperationCrud,
50 ) -> Pin<Box<dyn Future<Output = Result<String, LemmyError>> + '_>>;
52 /// `ChatServer` manages chat rooms and responsible for coordinating chat
54 pub struct ChatServer {
55 /// A map from generated random ID to session addr
56 pub sessions: HashMap<ConnectionId, SessionInfo>,
58 /// A map from post_id to set of connectionIDs
59 pub post_rooms: HashMap<PostId, HashSet<ConnectionId>>,
61 /// A map from community to set of connectionIDs
62 pub community_rooms: HashMap<CommunityId, HashSet<ConnectionId>>,
64 pub mod_rooms: HashMap<CommunityId, HashSet<ConnectionId>>,
66 /// A map from user id to its connection ID for joined users. Remember a user can have multiple
67 /// sessions (IE clients)
68 pub(super) user_rooms: HashMap<LocalUserId, HashSet<ConnectionId>>,
70 pub(super) rng: ThreadRng,
73 pub(super) pool: Pool<ConnectionManager<PgConnection>>,
76 pub(super) settings: Settings,
79 pub(super) secret: Secret,
81 /// Rate limiting based on rate type and IP addr
82 pub(super) rate_limiter: RateLimit,
84 /// A list of the current captchas
85 pub(super) captchas: Vec<CaptchaItem>,
87 message_handler: MessageHandlerType,
88 message_handler_crud: MessageHandlerCrudType,
93 activity_queue: QueueHandle,
96 pub struct SessionInfo {
97 pub addr: Recipient<WsMessage>,
101 /// `ChatServer` is an actor. It maintains list of connection client session.
102 /// And manages available rooms. Peers send messages to other peers in same
103 /// room through `ChatServer`.
105 #![allow(clippy::too_many_arguments)]
107 pool: Pool<ConnectionManager<PgConnection>>,
108 rate_limiter: RateLimit,
109 message_handler: MessageHandlerType,
110 message_handler_crud: MessageHandlerCrudType,
112 activity_queue: QueueHandle,
117 sessions: HashMap::new(),
118 post_rooms: HashMap::new(),
119 community_rooms: HashMap::new(),
120 mod_rooms: HashMap::new(),
121 user_rooms: HashMap::new(),
122 rng: rand::thread_rng(),
125 captchas: Vec::new(),
127 message_handler_crud,
135 pub fn join_community_room(
137 community_id: CommunityId,
139 ) -> Result<(), LemmyError> {
140 // remove session from all rooms
141 for sessions in self.community_rooms.values_mut() {
142 sessions.remove(&id);
145 // Also leave all post rooms
146 // This avoids double messages
147 for sessions in self.post_rooms.values_mut() {
148 sessions.remove(&id);
151 // If the room doesn't exist yet
152 if self.community_rooms.get_mut(&community_id).is_none() {
153 self.community_rooms.insert(community_id, HashSet::new());
158 .get_mut(&community_id)
159 .context(location_info!())?
164 pub fn join_mod_room(
166 community_id: CommunityId,
168 ) -> Result<(), LemmyError> {
169 // remove session from all rooms
170 for sessions in self.mod_rooms.values_mut() {
171 sessions.remove(&id);
174 // If the room doesn't exist yet
175 if self.mod_rooms.get_mut(&community_id).is_none() {
176 self.mod_rooms.insert(community_id, HashSet::new());
181 .get_mut(&community_id)
182 .context(location_info!())?
187 pub fn join_post_room(&mut self, post_id: PostId, id: ConnectionId) -> Result<(), LemmyError> {
188 // remove session from all rooms
189 for sessions in self.post_rooms.values_mut() {
190 sessions.remove(&id);
193 // Also leave all communities
194 // This avoids double messages
195 // TODO found a bug, whereby community messages like
196 // delete and remove aren't sent, because
197 // you left the community room
198 for sessions in self.community_rooms.values_mut() {
199 sessions.remove(&id);
202 // If the room doesn't exist yet
203 if self.post_rooms.get_mut(&post_id).is_none() {
204 self.post_rooms.insert(post_id, HashSet::new());
210 .context(location_info!())?
216 pub fn join_user_room(
218 user_id: LocalUserId,
220 ) -> Result<(), LemmyError> {
221 // remove session from all rooms
222 for sessions in self.user_rooms.values_mut() {
223 sessions.remove(&id);
226 // If the room doesn't exist yet
227 if self.user_rooms.get_mut(&user_id).is_none() {
228 self.user_rooms.insert(user_id, HashSet::new());
234 .context(location_info!())?
240 fn send_post_room_message<OP, Response>(
245 websocket_id: Option<ConnectionId>,
246 ) -> Result<(), LemmyError>
248 OP: OperationType + ToString,
251 let res_str = &serialize_websocket_message(op, response)?;
252 if let Some(sessions) = self.post_rooms.get(&post_id) {
254 if let Some(my_id) = websocket_id {
259 self.sendit(res_str, *id);
265 pub fn send_community_room_message<OP, Response>(
269 community_id: CommunityId,
270 websocket_id: Option<ConnectionId>,
271 ) -> Result<(), LemmyError>
273 OP: OperationType + ToString,
276 let res_str = &serialize_websocket_message(op, response)?;
277 if let Some(sessions) = self.community_rooms.get(&community_id) {
279 if let Some(my_id) = websocket_id {
284 self.sendit(res_str, *id);
290 pub fn send_mod_room_message<OP, Response>(
294 community_id: CommunityId,
295 websocket_id: Option<ConnectionId>,
296 ) -> Result<(), LemmyError>
298 OP: OperationType + ToString,
301 let res_str = &serialize_websocket_message(op, response)?;
302 if let Some(sessions) = self.mod_rooms.get(&community_id) {
304 if let Some(my_id) = websocket_id {
309 self.sendit(res_str, *id);
315 pub fn send_all_message<OP, Response>(
319 websocket_id: Option<ConnectionId>,
320 ) -> Result<(), LemmyError>
322 OP: OperationType + ToString,
325 let res_str = &serialize_websocket_message(op, response)?;
326 for id in self.sessions.keys() {
327 if let Some(my_id) = websocket_id {
332 self.sendit(res_str, *id);
337 pub fn send_user_room_message<OP, Response>(
341 recipient_id: LocalUserId,
342 websocket_id: Option<ConnectionId>,
343 ) -> Result<(), LemmyError>
345 OP: OperationType + ToString,
348 let res_str = &serialize_websocket_message(op, response)?;
349 if let Some(sessions) = self.user_rooms.get(&recipient_id) {
351 if let Some(my_id) = websocket_id {
356 self.sendit(res_str, *id);
362 pub fn send_comment<OP>(
365 comment: &CommentResponse,
366 websocket_id: Option<ConnectionId>,
367 ) -> Result<(), LemmyError>
369 OP: OperationType + ToString,
371 let mut comment_reply_sent = comment.clone();
373 // Strip out my specific user info
374 comment_reply_sent.comment_view.my_vote = None;
376 // Send it to the post room
377 let mut comment_post_sent = comment_reply_sent.clone();
378 // Remove the recipients here to separate mentions / user messages from post or community comments
379 comment_post_sent.recipient_ids = Vec::new();
380 self.send_post_room_message(
383 comment_post_sent.comment_view.post.id,
387 // Send it to the community too
388 self.send_community_room_message(
394 self.send_community_room_message(
397 comment.comment_view.community.id,
401 // Send it to the recipient(s) including the mentioned users
402 for recipient_id in &comment_reply_sent.recipient_ids {
403 self.send_user_room_message(
414 pub fn send_post<OP>(
417 post_res: &PostResponse,
418 websocket_id: Option<ConnectionId>,
419 ) -> Result<(), LemmyError>
421 OP: OperationType + ToString,
423 let community_id = post_res.post_view.community.id;
425 // Don't send my data with it
426 let mut post_sent = post_res.clone();
427 post_sent.post_view.my_vote = None;
429 // Send it to /c/all and that community
430 self.send_community_room_message(user_operation, &post_sent, CommunityId(0), websocket_id)?;
431 self.send_community_room_message(user_operation, &post_sent, community_id, websocket_id)?;
433 // Send it to the post room
434 self.send_post_room_message(
437 post_res.post_view.post.id,
444 fn sendit(&self, message: &str, id: ConnectionId) {
445 if let Some(info) = self.sessions.get(&id) {
446 let _ = info.addr.do_send(WsMessage(message.to_owned()));
450 pub(super) fn parse_json_message(
452 msg: StandardMessage,
453 ctx: &mut Context<Self>,
454 ) -> impl Future<Output = Result<String, LemmyError>> {
455 let rate_limiter = self.rate_limiter.clone();
457 let ip: IpAddr = match self.sessions.get(&msg.id) {
458 Some(info) => info.ip.to_owned(),
459 None => IpAddr("blank_ip".to_string()),
462 let context = LemmyContext {
463 pool: self.pool.clone(),
464 chat_server: ctx.address(),
465 client: self.client.to_owned(),
466 activity_queue: self.activity_queue.to_owned(),
467 settings: self.settings.to_owned(),
468 secret: self.secret.to_owned(),
470 let message_handler_crud = self.message_handler_crud;
471 let message_handler = self.message_handler;
473 let json: Value = serde_json::from_str(&msg.msg)?;
474 let data = &json["data"].to_string();
477 .ok_or_else(|| ApiError::err_plain("missing op"))?;
479 if let Ok(user_operation_crud) = UserOperationCrud::from_str(op) {
480 let fut = (message_handler_crud)(context, msg.id, user_operation_crud.clone(), data);
481 match user_operation_crud {
482 UserOperationCrud::Register => rate_limiter.register().wrap(ip, fut).await,
483 UserOperationCrud::CreatePost => rate_limiter.post().wrap(ip, fut).await,
484 UserOperationCrud::CreateCommunity => rate_limiter.register().wrap(ip, fut).await,
485 _ => rate_limiter.message().wrap(ip, fut).await,
488 let user_operation = UserOperation::from_str(op)?;
489 let fut = (message_handler)(context, msg.id, user_operation.clone(), data);
490 rate_limiter.message().wrap(ip, fut).await