2 messages::{CaptchaItem, StandardMessage, WsMessage},
3 serialize_websocket_message,
10 use anyhow::Context as acontext;
11 use lemmy_api_common::{comment::CommentResponse, post::PostResponse};
12 use lemmy_db_schema::{
13 newtypes::{CommunityId, LocalUserId, PostId},
14 source::secret::Secret,
20 rate_limit::RateLimitCell,
21 settings::structs::Settings,
25 use rand::rngs::ThreadRng;
26 use reqwest_middleware::ClientWithMiddleware;
28 use serde_json::Value;
30 collections::{HashMap, HashSet},
34 use tokio::macros::support::Pin;
36 type MessageHandlerType = fn(
37 context: LemmyContext,
41 ) -> Pin<Box<dyn Future<Output = Result<String, LemmyError>> + '_>>;
43 type MessageHandlerCrudType = fn(
44 context: LemmyContext,
46 op: UserOperationCrud,
48 ) -> Pin<Box<dyn Future<Output = Result<String, LemmyError>> + '_>>;
50 /// `ChatServer` manages chat rooms and responsible for coordinating chat
52 pub struct ChatServer {
53 /// A map from generated random ID to session addr
54 pub sessions: HashMap<ConnectionId, SessionInfo>,
56 /// A map from post_id to set of connectionIDs
57 pub post_rooms: HashMap<PostId, HashSet<ConnectionId>>,
59 /// A map from community to set of connectionIDs
60 pub community_rooms: HashMap<CommunityId, HashSet<ConnectionId>>,
62 pub mod_rooms: HashMap<CommunityId, HashSet<ConnectionId>>,
64 /// A map from user id to its connection ID for joined users. Remember a user can have multiple
65 /// sessions (IE clients)
66 pub(super) user_rooms: HashMap<LocalUserId, HashSet<ConnectionId>>,
68 pub(super) rng: ThreadRng,
71 pub(super) pool: DbPool,
74 pub(super) settings: Settings,
77 pub(super) secret: Secret,
79 /// A list of the current captchas
80 pub(super) captchas: Vec<CaptchaItem>,
82 message_handler: MessageHandlerType,
83 message_handler_crud: MessageHandlerCrudType,
86 client: ClientWithMiddleware,
88 rate_limit_cell: RateLimitCell,
91 pub struct SessionInfo {
92 pub addr: Recipient<WsMessage>,
96 /// `ChatServer` is an actor. It maintains list of connection client session.
97 /// And manages available rooms. Peers send messages to other peers in same
98 /// room through `ChatServer`.
100 #![allow(clippy::too_many_arguments)]
103 message_handler: MessageHandlerType,
104 message_handler_crud: MessageHandlerCrudType,
105 client: ClientWithMiddleware,
108 rate_limit_cell: RateLimitCell,
111 sessions: HashMap::new(),
112 post_rooms: HashMap::new(),
113 community_rooms: HashMap::new(),
114 mod_rooms: HashMap::new(),
115 user_rooms: HashMap::new(),
116 rng: rand::thread_rng(),
118 captchas: Vec::new(),
120 message_handler_crud,
128 pub fn join_community_room(
130 community_id: CommunityId,
132 ) -> Result<(), LemmyError> {
133 // remove session from all rooms
134 for sessions in self.community_rooms.values_mut() {
135 sessions.remove(&id);
138 // Also leave all post rooms
139 // This avoids double messages
140 for sessions in self.post_rooms.values_mut() {
141 sessions.remove(&id);
144 // If the room doesn't exist yet
145 if self.community_rooms.get_mut(&community_id).is_none() {
146 self.community_rooms.insert(community_id, HashSet::new());
151 .get_mut(&community_id)
152 .context(location_info!())?
157 pub fn join_mod_room(
159 community_id: CommunityId,
161 ) -> Result<(), LemmyError> {
162 // remove session from all rooms
163 for sessions in self.mod_rooms.values_mut() {
164 sessions.remove(&id);
167 // If the room doesn't exist yet
168 if self.mod_rooms.get_mut(&community_id).is_none() {
169 self.mod_rooms.insert(community_id, HashSet::new());
174 .get_mut(&community_id)
175 .context(location_info!())?
180 pub fn join_post_room(&mut self, post_id: PostId, id: ConnectionId) -> Result<(), LemmyError> {
181 // remove session from all rooms
182 for sessions in self.post_rooms.values_mut() {
183 sessions.remove(&id);
186 // Also leave all communities
187 // This avoids double messages
188 // TODO found a bug, whereby community messages like
189 // delete and remove aren't sent, because
190 // you left the community room
191 for sessions in self.community_rooms.values_mut() {
192 sessions.remove(&id);
195 // If the room doesn't exist yet
196 if self.post_rooms.get_mut(&post_id).is_none() {
197 self.post_rooms.insert(post_id, HashSet::new());
203 .context(location_info!())?
209 pub fn join_user_room(
211 user_id: LocalUserId,
213 ) -> Result<(), LemmyError> {
214 // remove session from all rooms
215 for sessions in self.user_rooms.values_mut() {
216 sessions.remove(&id);
219 // If the room doesn't exist yet
220 if self.user_rooms.get_mut(&user_id).is_none() {
221 self.user_rooms.insert(user_id, HashSet::new());
227 .context(location_info!())?
233 fn send_post_room_message<OP, Response>(
238 websocket_id: Option<ConnectionId>,
239 ) -> Result<(), LemmyError>
241 OP: OperationType + ToString,
244 let res_str = &serialize_websocket_message(op, response)?;
245 if let Some(sessions) = self.post_rooms.get(&post_id) {
247 if let Some(my_id) = websocket_id {
252 self.sendit(res_str, *id);
258 pub fn send_community_room_message<OP, Response>(
262 community_id: CommunityId,
263 websocket_id: Option<ConnectionId>,
264 ) -> Result<(), LemmyError>
266 OP: OperationType + ToString,
269 let res_str = &serialize_websocket_message(op, response)?;
270 if let Some(sessions) = self.community_rooms.get(&community_id) {
272 if let Some(my_id) = websocket_id {
277 self.sendit(res_str, *id);
283 pub fn send_mod_room_message<OP, Response>(
287 community_id: CommunityId,
288 websocket_id: Option<ConnectionId>,
289 ) -> Result<(), LemmyError>
291 OP: OperationType + ToString,
294 let res_str = &serialize_websocket_message(op, response)?;
295 if let Some(sessions) = self.mod_rooms.get(&community_id) {
297 if let Some(my_id) = websocket_id {
302 self.sendit(res_str, *id);
308 pub fn send_all_message<OP, Response>(
312 websocket_id: Option<ConnectionId>,
313 ) -> Result<(), LemmyError>
315 OP: OperationType + ToString,
318 let res_str = &serialize_websocket_message(op, response)?;
319 for id in self.sessions.keys() {
320 if let Some(my_id) = websocket_id {
325 self.sendit(res_str, *id);
330 pub fn send_user_room_message<OP, Response>(
334 recipient_id: LocalUserId,
335 websocket_id: Option<ConnectionId>,
336 ) -> Result<(), LemmyError>
338 OP: OperationType + ToString,
341 let res_str = &serialize_websocket_message(op, response)?;
342 if let Some(sessions) = self.user_rooms.get(&recipient_id) {
344 if let Some(my_id) = websocket_id {
349 self.sendit(res_str, *id);
355 pub fn send_comment<OP>(
358 comment: &CommentResponse,
359 websocket_id: Option<ConnectionId>,
360 ) -> Result<(), LemmyError>
362 OP: OperationType + ToString,
364 let mut comment_reply_sent = comment.clone();
366 // Strip out my specific user info
367 comment_reply_sent.comment_view.my_vote = None;
369 // Send it to the post room
370 let mut comment_post_sent = comment_reply_sent.clone();
371 // Remove the recipients here to separate mentions / user messages from post or community comments
372 comment_post_sent.recipient_ids = Vec::new();
373 self.send_post_room_message(
376 comment_post_sent.comment_view.post.id,
380 // Send it to the community too
381 self.send_community_room_message(
387 self.send_community_room_message(
390 comment.comment_view.community.id,
394 // Send it to the recipient(s) including the mentioned users
395 for recipient_id in &comment_reply_sent.recipient_ids {
396 self.send_user_room_message(
407 pub fn send_post<OP>(
410 post_res: &PostResponse,
411 websocket_id: Option<ConnectionId>,
412 ) -> Result<(), LemmyError>
414 OP: OperationType + ToString,
416 let community_id = post_res.post_view.community.id;
418 // Don't send my data with it
419 let mut post_sent = post_res.clone();
420 post_sent.post_view.my_vote = None;
422 // Send it to /c/all and that community
423 self.send_community_room_message(user_operation, &post_sent, CommunityId(0), websocket_id)?;
424 self.send_community_room_message(user_operation, &post_sent, community_id, websocket_id)?;
426 // Send it to the post room
427 self.send_post_room_message(
430 post_res.post_view.post.id,
437 fn sendit(&self, message: &str, id: ConnectionId) {
438 if let Some(info) = self.sessions.get(&id) {
439 info.addr.do_send(WsMessage(message.to_owned()));
443 pub(super) fn parse_json_message(
445 msg: StandardMessage,
446 ctx: &mut Context<Self>,
447 ) -> impl Future<Output = Result<String, LemmyError>> {
448 let ip: IpAddr = match self.sessions.get(&msg.id) {
449 Some(info) => info.ip.clone(),
450 None => IpAddr("blank_ip".to_string()),
453 let context = LemmyContext {
454 pool: self.pool.clone(),
455 chat_server: ctx.address(),
456 client: self.client.clone(),
457 settings: self.settings.clone(),
458 secret: self.secret.clone(),
459 rate_limit_cell: self.rate_limit_cell.clone(),
461 let message_handler_crud = self.message_handler_crud;
462 let message_handler = self.message_handler;
463 let rate_limiter = self.rate_limit_cell.clone();
465 let json: Value = serde_json::from_str(&msg.msg)?;
466 let data = &json["data"].to_string();
469 .ok_or_else(|| LemmyError::from_message("missing op"))?;
471 // check if api call passes the rate limit, and generate future for later execution
472 let (passed, fut) = if let Ok(user_operation_crud) = UserOperationCrud::from_str(op) {
473 let passed = match user_operation_crud {
474 UserOperationCrud::Register => rate_limiter.register().check(ip),
475 UserOperationCrud::CreatePost => rate_limiter.post().check(ip),
476 UserOperationCrud::CreateCommunity => rate_limiter.register().check(ip),
477 UserOperationCrud::CreateComment => rate_limiter.comment().check(ip),
478 _ => rate_limiter.message().check(ip),
480 let fut = (message_handler_crud)(context, msg.id, user_operation_crud, data);
483 let user_operation = UserOperation::from_str(op)?;
484 let passed = match user_operation {
485 UserOperation::GetCaptcha => rate_limiter.post().check(ip),
486 UserOperation::Search => rate_limiter.search().check(ip),
487 _ => rate_limiter.message().check(ip),
489 let fut = (message_handler)(context, msg.id, user_operation, data);
493 // if rate limit passed, execute api call future
497 // if rate limit was hit, respond with message
498 Err(LemmyError::from_message("rate_limit_error"))