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`.
7 use rand::{rngs::ThreadRng, Rng};
8 use serde::{Deserialize, Serialize};
10 use std::collections::{HashMap, HashSet};
11 use std::str::FromStr;
12 use std::time::SystemTime;
14 use crate::api::comment::*;
15 use crate::api::community::*;
16 use crate::api::post::*;
17 use crate::api::site::*;
18 use crate::api::user::*;
22 /// Chat server sends this messages to session
24 pub struct WSMessage(pub String);
26 /// Message for chat server communications
28 /// New chat session is created
32 pub addr: Recipient<WSMessage>,
36 /// Session is disconnected
38 pub struct Disconnect {
43 /// Send message to specific room
45 pub struct ClientMessage {
46 /// Id of the client session
54 #[derive(Serialize, Deserialize)]
55 pub struct StandardMessage {
56 /// Id of the client session
62 impl actix::Message for StandardMessage {
67 pub struct RateLimitBucket {
68 last_checked: SystemTime,
72 pub struct SessionInfo {
73 pub addr: Recipient<WSMessage>,
77 /// `ChatServer` manages chat rooms and responsible for coordinating chat
78 /// session. implementation is super primitive
79 pub struct ChatServer {
80 sessions: HashMap<usize, SessionInfo>, // A map from generated random ID to session addr
81 rate_limits: HashMap<String, RateLimitBucket>,
82 rooms: HashMap<i32, HashSet<usize>>, // A map from room / post name to set of connectionIDs
86 impl Default for ChatServer {
87 fn default() -> ChatServer {
89 let rooms = HashMap::new();
92 sessions: HashMap::new(),
93 rate_limits: HashMap::new(),
95 rng: rand::thread_rng(),
101 /// Send message to all users in the room
102 fn send_room_message(&self, room: &i32, message: &str, skip_id: usize) {
103 if let Some(sessions) = self.rooms.get(room) {
106 if let Some(info) = self.sessions.get(id) {
107 let _ = info.addr.do_send(WSMessage(message.to_owned()));
114 fn join_room(&mut self, room_id: i32, id: usize) {
115 // remove session from all rooms
116 for (_n, sessions) in &mut self.rooms {
117 sessions.remove(&id);
120 // If the room doesn't exist yet
121 if self.rooms.get_mut(&room_id).is_none() {
122 self.rooms.insert(room_id, HashSet::new());
125 &self.rooms.get_mut(&room_id).unwrap().insert(id);
128 fn send_community_message(
133 ) -> Result<(), Error> {
134 use crate::db::post_view::*;
136 let conn = establish_connection();
137 let posts = PostView::list(
139 ListingType::Community,
153 self.send_room_message(&post.id, message, skip_id);
159 fn check_rate_limit_register(&mut self, id: usize) -> Result<(), Error> {
160 self.check_rate_limit_full(
162 Settings::get().rate_limit_register,
163 Settings::get().rate_limit_register_per_second,
167 fn check_rate_limit_post(&mut self, id: usize) -> Result<(), Error> {
168 self.check_rate_limit_full(
170 Settings::get().rate_limit_post,
171 Settings::get().rate_limit_post_per_second,
175 fn check_rate_limit_message(&mut self, id: usize) -> Result<(), Error> {
176 self.check_rate_limit_full(
178 Settings::get().rate_limit_message,
179 Settings::get().rate_limit_message_per_second,
183 fn check_rate_limit_full(&mut self, id: usize, rate: i32, per: i32) -> Result<(), Error> {
184 if let Some(info) = self.sessions.get(&id) {
185 if let Some(rate_limit) = self.rate_limits.get_mut(&info.ip) {
187 if rate_limit.allowance == -2f64 {
188 rate_limit.allowance = rate as f64;
191 let current = SystemTime::now();
192 let time_passed = current.duration_since(rate_limit.last_checked)?.as_secs() as f64;
193 rate_limit.last_checked = current;
194 rate_limit.allowance += time_passed * (rate as f64 / per as f64);
195 if rate_limit.allowance > rate as f64 {
196 rate_limit.allowance = rate as f64;
199 if rate_limit.allowance < 1.0 {
201 "Rate limited IP: {}, time_passed: {}, allowance: {}",
202 &info.ip, time_passed, rate_limit.allowance
205 op: "Rate Limit".to_string(),
206 message: format!("Too many requests. {} per {} seconds", rate, per),
209 rate_limit.allowance -= 1.0;
221 /// Make actor from `ChatServer`
222 impl Actor for ChatServer {
223 /// We are going to use simple Context, we just need ability to communicate
224 /// with other actors.
225 type Context = Context<Self>;
228 /// Handler for Connect message.
230 /// Register new session and assign unique id to this session
231 impl Handler<Connect> for ChatServer {
234 fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
235 // notify all users in same room
236 // self.send_room_message(&"Main".to_owned(), "Someone joined", 0);
238 // register session with random id
239 let id = self.rng.gen::<usize>();
240 println!("{} joined", &msg.ip);
242 self.sessions.insert(
246 ip: msg.ip.to_owned(),
250 if self.rate_limits.get(&msg.ip).is_none() {
251 self.rate_limits.insert(
254 last_checked: SystemTime::now(),
264 /// Handler for Disconnect message.
265 impl Handler<Disconnect> for ChatServer {
268 fn handle(&mut self, msg: Disconnect, _: &mut Context<Self>) {
269 // let mut rooms: Vec<i32> = Vec::new();
272 if self.sessions.remove(&msg.id).is_some() {
273 // remove session from all rooms
274 for (_id, sessions) in &mut self.rooms {
275 if sessions.remove(&msg.id) {
283 /// Handler for Message message.
284 impl Handler<StandardMessage> for ChatServer {
285 type Result = MessageResult<StandardMessage>;
287 fn handle(&mut self, msg: StandardMessage, _: &mut Context<Self>) -> Self::Result {
288 let msg_out = match parse_json_message(self, msg) {
290 Err(e) => e.to_string(),
293 MessageResult(msg_out)
297 fn parse_json_message(chat: &mut ChatServer, msg: StandardMessage) -> Result<String, Error> {
298 let json: Value = serde_json::from_str(&msg.msg)?;
299 let data = &json["data"].to_string();
300 let op = &json["op"].as_str().ok_or(APIError {
301 op: "Unknown op type".to_string(),
302 message: format!("Unknown op type"),
305 let user_operation: UserOperation = UserOperation::from_str(&op)?;
307 match user_operation {
308 UserOperation::Login => {
309 let login: Login = serde_json::from_str(data)?;
310 let res = Oper::new(user_operation, login).perform()?;
311 Ok(serde_json::to_string(&res)?)
313 UserOperation::Register => {
314 let register: Register = serde_json::from_str(data)?;
315 let res = Oper::new(user_operation, register).perform();
317 chat.check_rate_limit_register(msg.id)?;
319 Ok(serde_json::to_string(&res?)?)
321 UserOperation::GetUserDetails => {
322 let get_user_details: GetUserDetails = serde_json::from_str(data)?;
323 let res = Oper::new(user_operation, get_user_details).perform()?;
324 Ok(serde_json::to_string(&res)?)
326 UserOperation::SaveUserSettings => {
327 let save_user_settings: SaveUserSettings = serde_json::from_str(data)?;
328 let res = Oper::new(user_operation, save_user_settings).perform()?;
329 Ok(serde_json::to_string(&res)?)
331 UserOperation::AddAdmin => {
332 let add_admin: AddAdmin = serde_json::from_str(data)?;
333 let res = Oper::new(user_operation, add_admin).perform()?;
334 Ok(serde_json::to_string(&res)?)
336 UserOperation::BanUser => {
337 let ban_user: BanUser = serde_json::from_str(data)?;
338 let res = Oper::new(user_operation, ban_user).perform()?;
339 Ok(serde_json::to_string(&res)?)
341 UserOperation::GetReplies => {
342 let get_replies: GetReplies = serde_json::from_str(data)?;
343 let res = Oper::new(user_operation, get_replies).perform()?;
344 Ok(serde_json::to_string(&res)?)
346 UserOperation::GetUserMentions => {
347 let get_user_mentions: GetUserMentions = serde_json::from_str(data)?;
348 let res = Oper::new(user_operation, get_user_mentions).perform()?;
349 Ok(serde_json::to_string(&res)?)
351 UserOperation::EditUserMention => {
352 let edit_user_mention: EditUserMention = serde_json::from_str(data)?;
353 let res = Oper::new(user_operation, edit_user_mention).perform()?;
354 Ok(serde_json::to_string(&res)?)
356 UserOperation::MarkAllAsRead => {
357 let mark_all_as_read: MarkAllAsRead = serde_json::from_str(data)?;
358 let res = Oper::new(user_operation, mark_all_as_read).perform()?;
359 Ok(serde_json::to_string(&res)?)
361 UserOperation::GetCommunity => {
362 let get_community: GetCommunity = serde_json::from_str(data)?;
363 let res = Oper::new(user_operation, get_community).perform()?;
364 Ok(serde_json::to_string(&res)?)
366 UserOperation::ListCommunities => {
367 let list_communities: ListCommunities = serde_json::from_str(data)?;
368 let res = Oper::new(user_operation, list_communities).perform()?;
369 Ok(serde_json::to_string(&res)?)
371 UserOperation::CreateCommunity => {
372 chat.check_rate_limit_register(msg.id)?;
373 let create_community: CreateCommunity = serde_json::from_str(data)?;
374 let res = Oper::new(user_operation, create_community).perform()?;
375 Ok(serde_json::to_string(&res)?)
377 UserOperation::EditCommunity => {
378 let edit_community: EditCommunity = serde_json::from_str(data)?;
379 let res = Oper::new(user_operation, edit_community).perform()?;
380 let mut community_sent: CommunityResponse = res.clone();
381 community_sent.community.user_id = None;
382 community_sent.community.subscribed = None;
383 let community_sent_str = serde_json::to_string(&community_sent)?;
384 chat.send_community_message(&community_sent.community.id, &community_sent_str, msg.id)?;
385 Ok(serde_json::to_string(&res)?)
387 UserOperation::FollowCommunity => {
388 let follow_community: FollowCommunity = serde_json::from_str(data)?;
389 let res = Oper::new(user_operation, follow_community).perform()?;
390 Ok(serde_json::to_string(&res)?)
392 UserOperation::GetFollowedCommunities => {
393 let followed_communities: GetFollowedCommunities = serde_json::from_str(data)?;
394 let res = Oper::new(user_operation, followed_communities).perform()?;
395 Ok(serde_json::to_string(&res)?)
397 UserOperation::BanFromCommunity => {
398 let ban_from_community: BanFromCommunity = serde_json::from_str(data)?;
399 let community_id = ban_from_community.community_id;
400 let res = Oper::new(user_operation, ban_from_community).perform()?;
401 let res_str = serde_json::to_string(&res)?;
402 chat.send_community_message(&community_id, &res_str, msg.id)?;
405 UserOperation::AddModToCommunity => {
406 let mod_add_to_community: AddModToCommunity = serde_json::from_str(data)?;
407 let community_id = mod_add_to_community.community_id;
408 let res = Oper::new(user_operation, mod_add_to_community).perform()?;
409 let res_str = serde_json::to_string(&res)?;
410 chat.send_community_message(&community_id, &res_str, msg.id)?;
413 UserOperation::ListCategories => {
414 let list_categories: ListCategories = ListCategories;
415 let res = Oper::new(user_operation, list_categories).perform()?;
416 Ok(serde_json::to_string(&res)?)
418 UserOperation::CreatePost => {
419 chat.check_rate_limit_post(msg.id)?;
420 let create_post: CreatePost = serde_json::from_str(data)?;
421 let res = Oper::new(user_operation, create_post).perform()?;
422 Ok(serde_json::to_string(&res)?)
424 UserOperation::GetPost => {
425 let get_post: GetPost = serde_json::from_str(data)?;
426 chat.join_room(get_post.id, msg.id);
427 let res = Oper::new(user_operation, get_post).perform()?;
428 Ok(serde_json::to_string(&res)?)
430 UserOperation::GetPosts => {
431 let get_posts: GetPosts = serde_json::from_str(data)?;
432 let res = Oper::new(user_operation, get_posts).perform()?;
433 Ok(serde_json::to_string(&res)?)
435 UserOperation::CreatePostLike => {
436 chat.check_rate_limit_message(msg.id)?;
437 let create_post_like: CreatePostLike = serde_json::from_str(data)?;
438 let res = Oper::new(user_operation, create_post_like).perform()?;
439 Ok(serde_json::to_string(&res)?)
441 UserOperation::EditPost => {
442 let edit_post: EditPost = serde_json::from_str(data)?;
443 let res = Oper::new(user_operation, edit_post).perform()?;
444 let mut post_sent = res.clone();
445 post_sent.post.my_vote = None;
446 let post_sent_str = serde_json::to_string(&post_sent)?;
447 chat.send_room_message(&post_sent.post.id, &post_sent_str, msg.id);
448 Ok(serde_json::to_string(&res)?)
450 UserOperation::SavePost => {
451 let save_post: SavePost = serde_json::from_str(data)?;
452 let res = Oper::new(user_operation, save_post).perform()?;
453 Ok(serde_json::to_string(&res)?)
455 UserOperation::CreateComment => {
456 chat.check_rate_limit_message(msg.id)?;
457 let create_comment: CreateComment = serde_json::from_str(data)?;
458 let post_id = create_comment.post_id;
459 let res = Oper::new(user_operation, create_comment).perform()?;
460 let mut comment_sent = res.clone();
461 comment_sent.comment.my_vote = None;
462 comment_sent.comment.user_id = None;
463 let comment_sent_str = serde_json::to_string(&comment_sent)?;
464 chat.send_room_message(&post_id, &comment_sent_str, msg.id);
465 Ok(serde_json::to_string(&res)?)
467 UserOperation::EditComment => {
468 let edit_comment: EditComment = serde_json::from_str(data)?;
469 let post_id = edit_comment.post_id;
470 let res = Oper::new(user_operation, edit_comment).perform()?;
471 let mut comment_sent = res.clone();
472 comment_sent.comment.my_vote = None;
473 comment_sent.comment.user_id = None;
474 let comment_sent_str = serde_json::to_string(&comment_sent)?;
475 chat.send_room_message(&post_id, &comment_sent_str, msg.id);
476 Ok(serde_json::to_string(&res)?)
478 UserOperation::SaveComment => {
479 let save_comment: SaveComment = serde_json::from_str(data)?;
480 let res = Oper::new(user_operation, save_comment).perform()?;
481 Ok(serde_json::to_string(&res)?)
483 UserOperation::CreateCommentLike => {
484 chat.check_rate_limit_message(msg.id)?;
485 let create_comment_like: CreateCommentLike = serde_json::from_str(data)?;
486 let post_id = create_comment_like.post_id;
487 let res = Oper::new(user_operation, create_comment_like).perform()?;
488 let mut comment_sent = res.clone();
489 comment_sent.comment.my_vote = None;
490 comment_sent.comment.user_id = None;
491 let comment_sent_str = serde_json::to_string(&comment_sent)?;
492 chat.send_room_message(&post_id, &comment_sent_str, msg.id);
493 Ok(serde_json::to_string(&res)?)
495 UserOperation::GetModlog => {
496 let get_modlog: GetModlog = serde_json::from_str(data)?;
497 let res = Oper::new(user_operation, get_modlog).perform()?;
498 Ok(serde_json::to_string(&res)?)
500 UserOperation::CreateSite => {
501 let create_site: CreateSite = serde_json::from_str(data)?;
502 let res = Oper::new(user_operation, create_site).perform()?;
503 Ok(serde_json::to_string(&res)?)
505 UserOperation::EditSite => {
506 let edit_site: EditSite = serde_json::from_str(data)?;
507 let res = Oper::new(user_operation, edit_site).perform()?;
508 Ok(serde_json::to_string(&res)?)
510 UserOperation::GetSite => {
511 let online: usize = chat.sessions.len();
512 let get_site: GetSite = serde_json::from_str(data)?;
513 let mut res = Oper::new(user_operation, get_site).perform()?;
515 Ok(serde_json::to_string(&res)?)
517 UserOperation::Search => {
518 let search: Search = serde_json::from_str(data)?;
519 let res = Oper::new(user_operation, search).perform()?;
520 Ok(serde_json::to_string(&res)?)
522 UserOperation::TransferCommunity => {
523 let transfer_community: TransferCommunity = serde_json::from_str(data)?;
524 let res = Oper::new(user_operation, transfer_community).perform()?;
525 Ok(serde_json::to_string(&res)?)
527 UserOperation::TransferSite => {
528 let transfer_site: TransferSite = serde_json::from_str(data)?;
529 let res = Oper::new(user_operation, transfer_site).perform()?;
530 Ok(serde_json::to_string(&res)?)
532 UserOperation::DeleteAccount => {
533 let delete_account: DeleteAccount = serde_json::from_str(data)?;
534 let res = Oper::new(user_operation, delete_account).perform()?;
535 Ok(serde_json::to_string(&res)?)