1 //! `ChatServer` is an actor. It maintains list of connection client session.
2 //! And manages available rooms. Peers send messages to other peers in same
3 //! room through `ChatServer`.
6 use diesel::r2d2::{ConnectionManager, Pool, PooledConnection};
7 use diesel::PgConnection;
9 use rand::{rngs::ThreadRng, Rng};
10 use serde::{Deserialize, Serialize};
11 use serde_json::Value;
12 use std::collections::{HashMap, HashSet};
13 use std::str::FromStr;
14 use std::time::SystemTime;
16 use crate::api::comment::*;
17 use crate::api::community::*;
18 use crate::api::post::*;
19 use crate::api::site::*;
20 use crate::api::user::*;
22 use crate::websocket::UserOperation;
25 type ConnectionId = usize;
27 type CommunityId = i32;
31 /// Chat server sends this messages to session
33 #[rtype(result = "()")]
34 pub struct WSMessage(pub String);
36 /// Message for chat server communications
38 /// New chat session is created
42 pub addr: Recipient<WSMessage>,
46 /// Session is disconnected
48 #[rtype(result = "()")]
49 pub struct Disconnect {
54 #[derive(Serialize, Deserialize, Message)]
56 pub struct StandardMessage {
57 /// Id of the client session
64 pub struct RateLimitBucket {
65 last_checked: SystemTime,
69 pub struct SessionInfo {
70 pub addr: Recipient<WSMessage>,
74 /// `ChatServer` manages chat rooms and responsible for coordinating chat
76 pub struct ChatServer {
77 /// A map from generated random ID to session addr
78 sessions: HashMap<ConnectionId, SessionInfo>,
80 /// A map from post_id to set of connectionIDs
81 post_rooms: HashMap<PostId, HashSet<ConnectionId>>,
83 /// A map from community to set of connectionIDs
84 community_rooms: HashMap<CommunityId, HashSet<ConnectionId>>,
86 /// A map from user id to its connection ID for joined users. Remember a user can have multiple
87 /// sessions (IE clients)
88 user_rooms: HashMap<UserId, HashSet<ConnectionId>>,
90 /// Rate limiting based on IP addr
91 rate_limits: HashMap<IPAddr, RateLimitBucket>,
94 db: Pool<ConnectionManager<PgConnection>>,
97 // TODO show online users for communities too
98 // TODO GetPosts is the community / front page join.
99 // What is sent: New posts, post edits, post removes, post likes, community edits, community mod adds. Notifs for new posts?
100 // GetPost is the PostJoin, LeavePost is the leave
101 // What is sent: New comments, comment edits, comment likes
102 // UserJoin is the user join, a disconnect should remove you from all the scopes
104 pub fn startup(db: Pool<ConnectionManager<PgConnection>>) -> ChatServer {
106 sessions: HashMap::new(),
107 rate_limits: HashMap::new(),
108 post_rooms: HashMap::new(),
109 community_rooms: HashMap::new(),
110 user_rooms: HashMap::new(),
111 rng: rand::thread_rng(),
116 fn join_community_room(&mut self, community_id: CommunityId, id: ConnectionId) {
117 // remove session from all rooms
118 for sessions in self.community_rooms.values_mut() {
119 sessions.remove(&id);
122 // If the room doesn't exist yet
123 if self.community_rooms.get_mut(&community_id).is_none() {
124 self.community_rooms.insert(community_id, HashSet::new());
129 .get_mut(&community_id)
134 fn join_post_room(&mut self, post_id: PostId, id: ConnectionId) {
135 // remove session from all rooms
136 for sessions in self.post_rooms.values_mut() {
137 sessions.remove(&id);
140 // If the room doesn't exist yet
141 if self.post_rooms.get_mut(&post_id).is_none() {
142 self.post_rooms.insert(post_id, HashSet::new());
145 self.post_rooms.get_mut(&post_id).unwrap().insert(id);
148 fn join_user_room(&mut self, user_id: UserId, id: ConnectionId) {
149 // remove session from all rooms
150 for sessions in self.user_rooms.values_mut() {
151 sessions.remove(&id);
154 // If the room doesn't exist yet
155 if self.user_rooms.get_mut(&user_id).is_none() {
156 self.user_rooms.insert(user_id, HashSet::new());
159 self.user_rooms.get_mut(&user_id).unwrap().insert(id);
162 fn send_post_room_message(&self, post_id: PostId, message: &str, skip_id: ConnectionId) {
163 if let Some(sessions) = self.post_rooms.get(&post_id) {
166 if let Some(info) = self.sessions.get(id) {
167 let _ = info.addr.do_send(WSMessage(message.to_owned()));
174 fn send_community_room_message(
176 community_id: CommunityId,
178 skip_id: ConnectionId,
180 if let Some(sessions) = self.community_rooms.get(&community_id) {
183 if let Some(info) = self.sessions.get(id) {
184 let _ = info.addr.do_send(WSMessage(message.to_owned()));
191 fn send_user_room_message(&self, user_id: UserId, message: &str, skip_id: ConnectionId) {
192 if let Some(sessions) = self.user_rooms.get(&user_id) {
195 if let Some(info) = self.sessions.get(id) {
196 let _ = info.addr.do_send(WSMessage(message.to_owned()));
203 fn send_all_message(&self, message: &str, skip_id: ConnectionId) {
204 for id in self.sessions.keys() {
206 if let Some(info) = self.sessions.get(id) {
207 let _ = info.addr.do_send(WSMessage(message.to_owned()));
213 fn check_rate_limit_register(&mut self, id: usize) -> Result<(), Error> {
214 self.check_rate_limit_full(
216 Settings::get().rate_limit.register,
217 Settings::get().rate_limit.register_per_second,
221 fn check_rate_limit_post(&mut self, id: usize) -> Result<(), Error> {
222 self.check_rate_limit_full(
224 Settings::get().rate_limit.post,
225 Settings::get().rate_limit.post_per_second,
229 fn check_rate_limit_message(&mut self, id: usize) -> Result<(), Error> {
230 self.check_rate_limit_full(
232 Settings::get().rate_limit.message,
233 Settings::get().rate_limit.message_per_second,
237 #[allow(clippy::float_cmp)]
238 fn check_rate_limit_full(&mut self, id: usize, rate: i32, per: i32) -> Result<(), Error> {
239 if let Some(info) = self.sessions.get(&id) {
240 if let Some(rate_limit) = self.rate_limits.get_mut(&info.ip) {
242 if rate_limit.allowance == -2f64 {
243 rate_limit.allowance = rate as f64;
246 let current = SystemTime::now();
247 let time_passed = current.duration_since(rate_limit.last_checked)?.as_secs() as f64;
248 rate_limit.last_checked = current;
249 rate_limit.allowance += time_passed * (rate as f64 / per as f64);
250 if rate_limit.allowance > rate as f64 {
251 rate_limit.allowance = rate as f64;
254 if rate_limit.allowance < 1.0 {
256 "Rate limited IP: {}, time_passed: {}, allowance: {}",
257 &info.ip, time_passed, rate_limit.allowance
261 message: format!("Too many requests. {} per {} seconds", rate, per),
266 rate_limit.allowance -= 1.0;
278 /// Make actor from `ChatServer`
279 impl Actor for ChatServer {
280 /// We are going to use simple Context, we just need ability to communicate
281 /// with other actors.
282 type Context = Context<Self>;
285 /// Handler for Connect message.
287 /// Register new session and assign unique id to this session
288 impl Handler<Connect> for ChatServer {
291 fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
292 // register session with random id
293 let id = self.rng.gen::<usize>();
294 println!("{} joined", &msg.ip);
296 self.sessions.insert(
300 ip: msg.ip.to_owned(),
304 if self.rate_limits.get(&msg.ip).is_none() {
305 self.rate_limits.insert(
308 last_checked: SystemTime::now(),
318 /// Handler for Disconnect message.
319 impl Handler<Disconnect> for ChatServer {
322 fn handle(&mut self, msg: Disconnect, _: &mut Context<Self>) {
323 // Remove connections from sessions and all 3 scopes
324 if self.sessions.remove(&msg.id).is_some() {
325 for sessions in self.user_rooms.values_mut() {
326 sessions.remove(&msg.id);
329 for sessions in self.post_rooms.values_mut() {
330 sessions.remove(&msg.id);
333 for sessions in self.community_rooms.values_mut() {
334 sessions.remove(&msg.id);
340 /// Handler for Message message.
341 impl Handler<StandardMessage> for ChatServer {
342 type Result = MessageResult<StandardMessage>;
344 fn handle(&mut self, msg: StandardMessage, _: &mut Context<Self>) -> Self::Result {
345 let msg_out = match parse_json_message(self, msg) {
347 Err(e) => e.to_string(),
350 println!("Message Sent: {}", msg_out);
351 MessageResult(msg_out)
356 struct WebsocketResponse<T> {
361 fn to_json_string<T>(op: &UserOperation, data: T) -> Result<String, Error>
365 let response = WebsocketResponse {
369 Ok(serde_json::to_string(&response)?)
372 fn do_user_operation<'a, Data, Response>(
375 conn: &PooledConnection<ConnectionManager<PgConnection>>,
376 ) -> Result<String, Error>
378 for<'de> Data: Deserialize<'de> + 'a,
380 Oper<Data>: Perform<Response>,
382 let parsed_data: Data = serde_json::from_str(data)?;
383 let res = Oper::new(parsed_data).perform(&conn)?;
384 to_json_string(&op, &res)
387 fn parse_json_message(chat: &mut ChatServer, msg: StandardMessage) -> Result<String, Error> {
388 let json: Value = serde_json::from_str(&msg.msg)?;
389 let data = &json["data"].to_string();
390 let op = &json["op"].as_str().ok_or(APIError {
391 message: "Unknown op type".to_string(),
394 let conn = chat.db.get()?;
396 let user_operation: UserOperation = UserOperation::from_str(&op)?;
398 // TODO: none of the chat messages are going to work if stuff is submitted via http api,
399 // need to move that handling elsewhere
400 match user_operation {
401 UserOperation::Login => do_user_operation::<Login, LoginResponse>(user_operation, data, &conn),
402 UserOperation::Register => {
403 chat.check_rate_limit_register(msg.id)?;
404 do_user_operation::<Register, LoginResponse>(user_operation, data, &conn)
406 UserOperation::GetUserDetails => {
407 do_user_operation::<GetUserDetails, GetUserDetailsResponse>(user_operation, data, &conn)
409 UserOperation::SaveUserSettings => {
410 do_user_operation::<SaveUserSettings, LoginResponse>(user_operation, data, &conn)
412 UserOperation::AddAdmin => {
413 let add_admin: AddAdmin = serde_json::from_str(data)?;
414 let res = Oper::new(add_admin).perform(&conn)?;
415 let res_str = to_json_string(&user_operation, &res)?;
416 chat.send_all_message(&res_str, msg.id);
419 UserOperation::BanUser => {
420 let ban_user: BanUser = serde_json::from_str(data)?;
421 let res = Oper::new(ban_user).perform(&conn)?;
422 let res_str = to_json_string(&user_operation, &res)?;
423 chat.send_all_message(&res_str, msg.id);
426 UserOperation::GetReplies => {
427 do_user_operation::<GetReplies, GetRepliesResponse>(user_operation, data, &conn)
429 UserOperation::GetUserMentions => {
430 do_user_operation::<GetUserMentions, GetUserMentionsResponse>(user_operation, data, &conn)
432 UserOperation::EditUserMention => {
433 do_user_operation::<EditUserMention, UserMentionResponse>(user_operation, data, &conn)
435 UserOperation::MarkAllAsRead => {
436 do_user_operation::<MarkAllAsRead, GetRepliesResponse>(user_operation, data, &conn)
438 UserOperation::GetCommunity => {
439 let get_community: GetCommunity = serde_json::from_str(data)?;
440 let mut res = Oper::new(get_community).perform(&conn)?;
441 let community_id = res.community.id;
443 chat.join_community_room(community_id, msg.id);
445 res.online = if let Some(community_users) = chat.community_rooms.get(&community_id) {
446 community_users.len()
451 to_json_string(&user_operation, &res)
453 UserOperation::ListCommunities => {
454 do_user_operation::<ListCommunities, ListCommunitiesResponse>(user_operation, data, &conn)
456 UserOperation::CreateCommunity => {
457 chat.check_rate_limit_register(msg.id)?;
458 do_user_operation::<CreateCommunity, CommunityResponse>(user_operation, data, &conn)
460 UserOperation::EditCommunity => {
461 let edit_community: EditCommunity = serde_json::from_str(data)?;
462 let res = Oper::new(edit_community).perform(&conn)?;
463 let mut community_sent: CommunityResponse = res.clone();
464 community_sent.community.user_id = None;
465 community_sent.community.subscribed = None;
466 let community_sent_str = to_json_string(&user_operation, &community_sent)?;
467 chat.send_community_room_message(community_sent.community.id, &community_sent_str, msg.id);
468 to_json_string(&user_operation, &res)
470 UserOperation::FollowCommunity => {
471 do_user_operation::<FollowCommunity, CommunityResponse>(user_operation, data, &conn)
473 UserOperation::GetFollowedCommunities => do_user_operation::<
474 GetFollowedCommunities,
475 GetFollowedCommunitiesResponse,
476 >(user_operation, data, &conn),
477 UserOperation::BanFromCommunity => {
478 let ban_from_community: BanFromCommunity = serde_json::from_str(data)?;
479 let community_id = ban_from_community.community_id;
480 let res = Oper::new(ban_from_community).perform(&conn)?;
481 let res_str = to_json_string(&user_operation, &res)?;
482 chat.send_community_room_message(community_id, &res_str, msg.id);
485 UserOperation::AddModToCommunity => {
486 let mod_add_to_community: AddModToCommunity = serde_json::from_str(data)?;
487 let community_id = mod_add_to_community.community_id;
488 let res = Oper::new(mod_add_to_community).perform(&conn)?;
489 let res_str = to_json_string(&user_operation, &res)?;
490 chat.send_community_room_message(community_id, &res_str, msg.id);
493 UserOperation::ListCategories => {
494 do_user_operation::<ListCategories, ListCategoriesResponse>(user_operation, data, &conn)
496 UserOperation::CreatePost => {
497 chat.check_rate_limit_post(msg.id)?;
498 let create_post: CreatePost = serde_json::from_str(data)?;
499 let community_id = create_post.community_id;
500 let res = Oper::new(create_post).perform(&conn)?;
501 let res_str = to_json_string(&user_operation, &res)?;
503 // Don't send my data with it
504 let mut post_sent = res.clone();
505 post_sent.post.my_vote = None;
506 post_sent.post.user_id = None;
507 let post_sent_str = to_json_string(&user_operation, &post_sent)?;
509 // Send it to /c/all and that community
510 chat.send_community_room_message(0, &post_sent_str, msg.id);
511 chat.send_community_room_message(community_id, &post_sent_str, msg.id);
515 UserOperation::GetPost => {
516 let get_post: GetPost = serde_json::from_str(data)?;
517 let post_id = get_post.id;
518 chat.join_post_room(post_id, msg.id);
519 let mut res = Oper::new(get_post).perform(&conn)?;
521 res.online = if let Some(post_users) = chat.post_rooms.get(&post_id) {
527 to_json_string(&user_operation, &res)
529 UserOperation::GetPosts => {
530 let get_posts: GetPosts = serde_json::from_str(data)?;
531 if get_posts.community_id.is_none() {
532 // 0 is the "all" community
533 chat.join_community_room(0, msg.id);
535 let res = Oper::new(get_posts).perform(&conn)?;
536 to_json_string(&user_operation, &res)
538 UserOperation::UserJoin => {
539 let user_join: UserJoin = serde_json::from_str(data)?;
540 let res = Oper::new(user_join).perform(&conn)?;
541 chat.join_user_room(res.user_id, msg.id);
542 to_json_string(&user_operation, &res)
544 UserOperation::CreatePostLike => {
545 chat.check_rate_limit_message(msg.id)?;
546 let create_post_like: CreatePostLike = serde_json::from_str(data)?;
547 let res = Oper::new(create_post_like).perform(&conn)?;
548 let community_id = res.post.community_id;
549 let res_str = to_json_string(&user_operation, &res)?;
551 // Don't send my data with it
552 let mut post_sent = res.clone();
553 post_sent.post.my_vote = None;
554 post_sent.post.user_id = None;
555 let post_sent_str = to_json_string(&user_operation, &post_sent)?;
557 // Send it to /c/all and that community
558 chat.send_community_room_message(0, &post_sent_str, msg.id);
559 chat.send_community_room_message(community_id, &post_sent_str, msg.id);
563 UserOperation::EditPost => {
564 let edit_post: EditPost = serde_json::from_str(data)?;
565 let res = Oper::new(edit_post).perform(&conn)?;
566 let mut post_sent = res.clone();
567 post_sent.post.my_vote = None;
568 post_sent.post.user_id = None;
569 let post_sent_str = to_json_string(&user_operation, &post_sent)?;
571 // Send it to /c/all and that community
572 chat.send_community_room_message(0, &post_sent_str, msg.id);
573 chat.send_community_room_message(post_sent.post.community_id, &post_sent_str, msg.id);
574 to_json_string(&user_operation, &res)
576 UserOperation::SavePost => {
577 do_user_operation::<SavePost, PostResponse>(user_operation, data, &conn)
579 UserOperation::CreateComment => {
580 chat.check_rate_limit_message(msg.id)?;
581 let create_comment: CreateComment = serde_json::from_str(data)?;
582 let post_id = create_comment.post_id;
583 let res = Oper::new(create_comment).perform(&conn)?;
584 let mut comment_sent = res.clone();
585 comment_sent.comment.my_vote = None;
586 comment_sent.comment.user_id = None;
587 let comment_sent_str = to_json_string(&user_operation, &comment_sent)?;
589 // Send it to the post room
590 chat.send_post_room_message(post_id, &comment_sent_str, msg.id);
592 // Send it to the recipient(s) including the mentioned users
593 for recipient_id in comment_sent.recipient_ids {
594 chat.send_user_room_message(recipient_id, &comment_sent_str, msg.id);
597 to_json_string(&user_operation, &res)
599 UserOperation::EditComment => {
600 let edit_comment: EditComment = serde_json::from_str(data)?;
601 let post_id = edit_comment.post_id;
602 let res = Oper::new(edit_comment).perform(&conn)?;
603 let mut comment_sent = res.clone();
604 comment_sent.comment.my_vote = None;
605 comment_sent.comment.user_id = None;
606 let comment_sent_str = to_json_string(&user_operation, &comment_sent)?;
608 chat.send_post_room_message(post_id, &comment_sent_str, msg.id);
610 // Send it to the recipient(s) including the mentioned users
611 for recipient_id in comment_sent.recipient_ids {
612 chat.send_user_room_message(recipient_id, &comment_sent_str, msg.id);
615 to_json_string(&user_operation, &res)
617 UserOperation::SaveComment => {
618 do_user_operation::<SaveComment, CommentResponse>(user_operation, data, &conn)
620 UserOperation::CreateCommentLike => {
621 chat.check_rate_limit_message(msg.id)?;
622 let create_comment_like: CreateCommentLike = serde_json::from_str(data)?;
623 let post_id = create_comment_like.post_id;
624 let res = Oper::new(create_comment_like).perform(&conn)?;
625 let mut comment_sent = res.clone();
626 comment_sent.comment.my_vote = None;
627 comment_sent.comment.user_id = None;
628 let comment_sent_str = to_json_string(&user_operation, &comment_sent)?;
630 chat.send_post_room_message(post_id, &comment_sent_str, msg.id);
632 // Send it to the recipient(s) including the mentioned users
633 for recipient_id in comment_sent.recipient_ids {
634 chat.send_user_room_message(recipient_id, &comment_sent_str, msg.id);
636 to_json_string(&user_operation, &res)
638 UserOperation::GetModlog => {
639 do_user_operation::<GetModlog, GetModlogResponse>(user_operation, data, &conn)
641 UserOperation::CreateSite => {
642 do_user_operation::<CreateSite, SiteResponse>(user_operation, data, &conn)
644 UserOperation::EditSite => {
645 let edit_site: EditSite = serde_json::from_str(data)?;
646 let res = Oper::new(edit_site).perform(&conn)?;
647 let res_str = to_json_string(&user_operation, &res)?;
648 chat.send_all_message(&res_str, msg.id);
651 UserOperation::GetSite => {
652 let get_site: GetSite = serde_json::from_str(data)?;
653 let mut res = Oper::new(get_site).perform(&conn)?;
654 res.online = chat.sessions.len();
655 to_json_string(&user_operation, &res)
657 UserOperation::Search => {
658 do_user_operation::<Search, SearchResponse>(user_operation, data, &conn)
660 UserOperation::TransferCommunity => {
661 do_user_operation::<TransferCommunity, GetCommunityResponse>(user_operation, data, &conn)
663 UserOperation::TransferSite => {
664 do_user_operation::<TransferSite, GetSiteResponse>(user_operation, data, &conn)
666 UserOperation::DeleteAccount => {
667 do_user_operation::<DeleteAccount, LoginResponse>(user_operation, data, &conn)
669 UserOperation::PasswordReset => {
670 do_user_operation::<PasswordReset, PasswordResetResponse>(user_operation, data, &conn)
672 UserOperation::PasswordChange => {
673 do_user_operation::<PasswordChange, LoginResponse>(user_operation, data, &conn)
675 UserOperation::CreatePrivateMessage => {
676 chat.check_rate_limit_message(msg.id)?;
677 let create_private_message: CreatePrivateMessage = serde_json::from_str(data)?;
678 let recipient_id = create_private_message.recipient_id;
679 let res = Oper::new(create_private_message).perform(&conn)?;
680 let res_str = to_json_string(&user_operation, &res)?;
682 chat.send_user_room_message(recipient_id, &res_str, msg.id);
685 UserOperation::EditPrivateMessage => {
686 do_user_operation::<EditPrivateMessage, PrivateMessageResponse>(user_operation, data, &conn)
688 UserOperation::GetPrivateMessages => {
689 do_user_operation::<GetPrivateMessages, PrivateMessagesResponse>(user_operation, data, &conn)