2 comment::CommentResponse,
6 messages::{CaptchaItem, StandardMessage, WsMessage},
7 serialize_websocket_message,
14 use actix::prelude::*;
15 use anyhow::Context as acontext;
16 use lemmy_db_schema::{
17 newtypes::{CommunityId, LocalUserId, PostId},
18 source::secret::Secret,
24 rate_limit::RateLimitCell,
25 settings::structs::Settings,
29 use rand::rngs::ThreadRng;
30 use reqwest_middleware::ClientWithMiddleware;
32 use serde_json::Value;
34 collections::{HashMap, HashSet},
38 use tokio::macros::support::Pin;
40 type MessageHandlerType = fn(
41 context: LemmyContext,
45 ) -> Pin<Box<dyn Future<Output = Result<String, LemmyError>> + '_>>;
47 type MessageHandlerCrudType = fn(
48 context: LemmyContext,
50 op: UserOperationCrud,
52 ) -> Pin<Box<dyn Future<Output = Result<String, LemmyError>> + '_>>;
54 type MessageHandlerApubType = fn(
55 context: LemmyContext,
57 op: UserOperationApub,
59 ) -> Pin<Box<dyn Future<Output = Result<String, LemmyError>> + '_>>;
61 /// `ChatServer` manages chat rooms and responsible for coordinating chat
63 pub struct ChatServer {
64 /// A map from generated random ID to session addr
65 pub sessions: HashMap<ConnectionId, SessionInfo>,
67 /// A map from post_id to set of connectionIDs
68 pub post_rooms: HashMap<PostId, HashSet<ConnectionId>>,
70 /// A map from community to set of connectionIDs
71 pub community_rooms: HashMap<CommunityId, HashSet<ConnectionId>>,
73 pub mod_rooms: HashMap<CommunityId, HashSet<ConnectionId>>,
75 /// A map from user id to its connection ID for joined users. Remember a user can have multiple
76 /// sessions (IE clients)
77 pub(super) user_rooms: HashMap<LocalUserId, HashSet<ConnectionId>>,
79 pub(super) rng: ThreadRng,
82 pub(super) pool: DbPool,
85 pub(super) settings: Settings,
88 pub(super) secret: Secret,
90 /// A list of the current captchas
91 pub(super) captchas: Vec<CaptchaItem>,
93 message_handler: MessageHandlerType,
94 message_handler_crud: MessageHandlerCrudType,
95 message_handler_apub: MessageHandlerApubType,
98 client: ClientWithMiddleware,
100 rate_limit_cell: RateLimitCell,
103 pub struct SessionInfo {
104 pub addr: Recipient<WsMessage>,
108 /// `ChatServer` is an actor. It maintains list of connection client session.
109 /// And manages available rooms. Peers send messages to other peers in same
110 /// room through `ChatServer`.
112 #![allow(clippy::too_many_arguments)]
115 message_handler: MessageHandlerType,
116 message_handler_crud: MessageHandlerCrudType,
117 message_handler_apub: MessageHandlerApubType,
118 client: ClientWithMiddleware,
121 rate_limit_cell: RateLimitCell,
124 sessions: HashMap::new(),
125 post_rooms: HashMap::new(),
126 community_rooms: HashMap::new(),
127 mod_rooms: HashMap::new(),
128 user_rooms: HashMap::new(),
129 rng: rand::thread_rng(),
131 captchas: Vec::new(),
133 message_handler_crud,
134 message_handler_apub,
142 pub fn join_community_room(
144 community_id: CommunityId,
146 ) -> Result<(), LemmyError> {
147 // remove session from all rooms
148 for sessions in self.community_rooms.values_mut() {
149 sessions.remove(&id);
152 // Also leave all post rooms
153 // This avoids double messages
154 for sessions in self.post_rooms.values_mut() {
155 sessions.remove(&id);
158 // If the room doesn't exist yet
159 if self.community_rooms.get_mut(&community_id).is_none() {
160 self.community_rooms.insert(community_id, HashSet::new());
165 .get_mut(&community_id)
166 .context(location_info!())?
171 pub fn join_mod_room(
173 community_id: CommunityId,
175 ) -> Result<(), LemmyError> {
176 // remove session from all rooms
177 for sessions in self.mod_rooms.values_mut() {
178 sessions.remove(&id);
181 // If the room doesn't exist yet
182 if self.mod_rooms.get_mut(&community_id).is_none() {
183 self.mod_rooms.insert(community_id, HashSet::new());
188 .get_mut(&community_id)
189 .context(location_info!())?
194 pub fn join_post_room(&mut self, post_id: PostId, id: ConnectionId) -> Result<(), LemmyError> {
195 // remove session from all rooms
196 for sessions in self.post_rooms.values_mut() {
197 sessions.remove(&id);
200 // Also leave all communities
201 // This avoids double messages
202 // TODO found a bug, whereby community messages like
203 // delete and remove aren't sent, because
204 // you left the community room
205 for sessions in self.community_rooms.values_mut() {
206 sessions.remove(&id);
209 // If the room doesn't exist yet
210 if self.post_rooms.get_mut(&post_id).is_none() {
211 self.post_rooms.insert(post_id, HashSet::new());
217 .context(location_info!())?
223 pub fn join_user_room(
225 user_id: LocalUserId,
227 ) -> Result<(), LemmyError> {
228 // remove session from all rooms
229 for sessions in self.user_rooms.values_mut() {
230 sessions.remove(&id);
233 // If the room doesn't exist yet
234 if self.user_rooms.get_mut(&user_id).is_none() {
235 self.user_rooms.insert(user_id, HashSet::new());
241 .context(location_info!())?
247 fn send_post_room_message<OP, Response>(
252 websocket_id: Option<ConnectionId>,
253 ) -> Result<(), LemmyError>
255 OP: OperationType + ToString,
258 let res_str = &serialize_websocket_message(op, response)?;
259 if let Some(sessions) = self.post_rooms.get(&post_id) {
261 if let Some(my_id) = websocket_id {
266 self.sendit(res_str, *id);
272 pub fn send_community_room_message<OP, Response>(
276 community_id: CommunityId,
277 websocket_id: Option<ConnectionId>,
278 ) -> Result<(), LemmyError>
280 OP: OperationType + ToString,
283 let res_str = &serialize_websocket_message(op, response)?;
284 if let Some(sessions) = self.community_rooms.get(&community_id) {
286 if let Some(my_id) = websocket_id {
291 self.sendit(res_str, *id);
297 pub fn send_mod_room_message<OP, Response>(
301 community_id: CommunityId,
302 websocket_id: Option<ConnectionId>,
303 ) -> Result<(), LemmyError>
305 OP: OperationType + ToString,
308 let res_str = &serialize_websocket_message(op, response)?;
309 if let Some(sessions) = self.mod_rooms.get(&community_id) {
311 if let Some(my_id) = websocket_id {
316 self.sendit(res_str, *id);
322 pub fn send_all_message<OP, Response>(
326 websocket_id: Option<ConnectionId>,
327 ) -> Result<(), LemmyError>
329 OP: OperationType + ToString,
332 let res_str = &serialize_websocket_message(op, response)?;
333 for id in self.sessions.keys() {
334 if let Some(my_id) = websocket_id {
339 self.sendit(res_str, *id);
344 pub fn send_user_room_message<OP, Response>(
348 recipient_id: LocalUserId,
349 websocket_id: Option<ConnectionId>,
350 ) -> Result<(), LemmyError>
352 OP: OperationType + ToString,
355 let res_str = &serialize_websocket_message(op, response)?;
356 if let Some(sessions) = self.user_rooms.get(&recipient_id) {
358 if let Some(my_id) = websocket_id {
363 self.sendit(res_str, *id);
369 pub fn send_comment<OP>(
372 comment: &CommentResponse,
373 websocket_id: Option<ConnectionId>,
374 ) -> Result<(), LemmyError>
376 OP: OperationType + ToString,
378 let mut comment_reply_sent = comment.clone();
380 // Strip out my specific user info
381 comment_reply_sent.comment_view.my_vote = None;
383 // Send it to the post room
384 let mut comment_post_sent = comment_reply_sent.clone();
385 // Remove the recipients here to separate mentions / user messages from post or community comments
386 comment_post_sent.recipient_ids = Vec::new();
387 self.send_post_room_message(
390 comment_post_sent.comment_view.post.id,
394 // Send it to the community too
395 self.send_community_room_message(
401 self.send_community_room_message(
404 comment.comment_view.community.id,
408 // Send it to the recipient(s) including the mentioned users
409 for recipient_id in &comment_reply_sent.recipient_ids {
410 self.send_user_room_message(
421 pub fn send_post<OP>(
424 post_res: &PostResponse,
425 websocket_id: Option<ConnectionId>,
426 ) -> Result<(), LemmyError>
428 OP: OperationType + ToString,
430 let community_id = post_res.post_view.community.id;
432 // Don't send my data with it
433 let mut post_sent = post_res.clone();
434 post_sent.post_view.my_vote = None;
436 // Send it to /c/all and that community
437 self.send_community_room_message(user_operation, &post_sent, CommunityId(0), websocket_id)?;
438 self.send_community_room_message(user_operation, &post_sent, community_id, websocket_id)?;
440 // Send it to the post room
441 self.send_post_room_message(
444 post_res.post_view.post.id,
451 fn sendit(&self, message: &str, id: ConnectionId) {
452 if let Some(info) = self.sessions.get(&id) {
453 info.addr.do_send(WsMessage(message.to_owned()));
457 pub(super) fn parse_json_message(
459 msg: StandardMessage,
460 ctx: &mut Context<Self>,
461 ) -> impl Future<Output = Result<String, LemmyError>> {
462 let ip: IpAddr = match self.sessions.get(&msg.id) {
463 Some(info) => info.ip.clone(),
464 None => IpAddr("blank_ip".to_string()),
467 let context = LemmyContext::create(
471 self.settings.clone(),
473 self.rate_limit_cell.clone(),
475 let message_handler_crud = self.message_handler_crud;
476 let message_handler = self.message_handler;
477 let message_handler_apub = self.message_handler_apub;
478 let rate_limiter = self.rate_limit_cell.clone();
480 let json: Value = serde_json::from_str(&msg.msg)?;
481 let data = &json["data"].to_string();
484 .ok_or_else(|| LemmyError::from_message("missing op"))?;
486 // check if api call passes the rate limit, and generate future for later execution
487 let (passed, fut) = if let Ok(user_operation_crud) = UserOperationCrud::from_str(op) {
488 let passed = match user_operation_crud {
489 UserOperationCrud::Register => rate_limiter.register().check(ip),
490 UserOperationCrud::CreatePost => rate_limiter.post().check(ip),
491 UserOperationCrud::CreateCommunity => rate_limiter.register().check(ip),
492 UserOperationCrud::CreateComment => rate_limiter.comment().check(ip),
493 _ => rate_limiter.message().check(ip),
495 let fut = (message_handler_crud)(context, msg.id, user_operation_crud, data);
497 } else if let Ok(user_operation) = UserOperation::from_str(op) {
498 let passed = match user_operation {
499 UserOperation::GetCaptcha => rate_limiter.post().check(ip),
500 _ => rate_limiter.message().check(ip),
502 let fut = (message_handler)(context, msg.id, user_operation, data);
505 let user_operation = UserOperationApub::from_str(op)?;
506 let passed = match user_operation {
507 UserOperationApub::Search => rate_limiter.search().check(ip),
508 _ => rate_limiter.message().check(ip),
510 let fut = (message_handler_apub)(context, msg.id, user_operation, data);
514 // if rate limit passed, execute api call future
518 // if rate limit was hit, respond with message
519 Err(LemmyError::from_message("rate_limit_error"))