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::*;
21 const RATE_LIMIT_MESSAGE: i32 = 30;
22 const RATE_LIMIT_MESSAGES_PER_SECOND: i32 = 60;
23 const RATE_LIMIT_POST: i32 = 3;
24 const RATE_LIMIT_POSTS_PER_SECOND: i32 = 60 * 10;
25 const RATE_LIMIT_REGISTER: i32 = 1;
26 const RATE_LIMIT_REGISTER_PER_SECOND: i32 = 60 * 60;
28 /// Chat server sends this messages to session
30 pub struct WSMessage(pub String);
32 /// Message for chat server communications
34 /// New chat session is created
38 pub addr: Recipient<WSMessage>,
42 /// Session is disconnected
44 pub struct Disconnect {
49 /// Send message to specific room
51 pub struct ClientMessage {
52 /// Id of the client session
60 #[derive(Serialize, Deserialize)]
61 pub struct StandardMessage {
62 /// Id of the client session
68 impl actix::Message for StandardMessage {
73 pub struct RateLimitBucket {
74 last_checked: SystemTime,
78 pub struct SessionInfo {
79 pub addr: Recipient<WSMessage>,
83 /// `ChatServer` manages chat rooms and responsible for coordinating chat
84 /// session. implementation is super primitive
85 pub struct ChatServer {
86 sessions: HashMap<usize, SessionInfo>, // A map from generated random ID to session addr
87 rate_limits: HashMap<String, RateLimitBucket>,
88 rooms: HashMap<i32, HashSet<usize>>, // A map from room / post name to set of connectionIDs
92 impl Default for ChatServer {
93 fn default() -> ChatServer {
95 let rooms = HashMap::new();
98 sessions: HashMap::new(),
99 rate_limits: HashMap::new(),
101 rng: rand::thread_rng(),
107 /// Send message to all users in the room
108 fn send_room_message(&self, room: &i32, message: &str, skip_id: usize) {
109 if let Some(sessions) = self.rooms.get(room) {
112 if let Some(info) = self.sessions.get(id) {
113 let _ = info.addr.do_send(WSMessage(message.to_owned()));
120 fn join_room(&mut self, room_id: i32, id: usize) {
121 // remove session from all rooms
122 for (_n, sessions) in &mut self.rooms {
123 sessions.remove(&id);
126 // If the room doesn't exist yet
127 if self.rooms.get_mut(&room_id).is_none() {
128 self.rooms.insert(room_id, HashSet::new());
131 &self.rooms.get_mut(&room_id).unwrap().insert(id);
134 fn send_community_message(
139 ) -> Result<(), Error> {
140 use crate::db::post_view::*;
142 let conn = establish_connection();
143 let posts = PostView::list(
145 PostListingType::Community,
159 self.send_room_message(&post.id, message, skip_id);
165 fn check_rate_limit_register(&mut self, id: usize) -> Result<(), Error> {
166 self.check_rate_limit_full(id, RATE_LIMIT_REGISTER, RATE_LIMIT_REGISTER_PER_SECOND)
169 fn check_rate_limit_post(&mut self, id: usize) -> Result<(), Error> {
170 self.check_rate_limit_full(id, RATE_LIMIT_POST, RATE_LIMIT_POSTS_PER_SECOND)
173 fn check_rate_limit_message(&mut self, id: usize) -> Result<(), Error> {
174 self.check_rate_limit_full(id, RATE_LIMIT_MESSAGE, RATE_LIMIT_MESSAGES_PER_SECOND)
177 fn check_rate_limit_full(&mut self, id: usize, rate: i32, per: i32) -> Result<(), Error> {
178 if let Some(info) = self.sessions.get(&id) {
179 if let Some(rate_limit) = self.rate_limits.get_mut(&info.ip) {
181 if rate_limit.allowance == -2f64 {
182 rate_limit.allowance = rate as f64;
185 let current = SystemTime::now();
186 let time_passed = current.duration_since(rate_limit.last_checked)?.as_secs() as f64;
187 rate_limit.last_checked = current;
188 rate_limit.allowance += time_passed * (rate as f64 / per as f64);
189 if rate_limit.allowance > rate as f64 {
190 rate_limit.allowance = rate as f64;
193 if rate_limit.allowance < 1.0 {
195 "Rate limited IP: {}, time_passed: {}, allowance: {}",
196 &info.ip, time_passed, rate_limit.allowance
199 op: "Rate Limit".to_string(),
200 message: format!("Too many requests. {} per {} seconds", rate, per),
203 rate_limit.allowance -= 1.0;
215 /// Make actor from `ChatServer`
216 impl Actor for ChatServer {
217 /// We are going to use simple Context, we just need ability to communicate
218 /// with other actors.
219 type Context = Context<Self>;
222 /// Handler for Connect message.
224 /// Register new session and assign unique id to this session
225 impl Handler<Connect> for ChatServer {
228 fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
229 // notify all users in same room
230 // self.send_room_message(&"Main".to_owned(), "Someone joined", 0);
232 // register session with random id
233 let id = self.rng.gen::<usize>();
234 println!("{} joined", &msg.ip);
236 self.sessions.insert(
240 ip: msg.ip.to_owned(),
244 if self.rate_limits.get(&msg.ip).is_none() {
245 self.rate_limits.insert(
248 last_checked: SystemTime::now(),
258 /// Handler for Disconnect message.
259 impl Handler<Disconnect> for ChatServer {
262 fn handle(&mut self, msg: Disconnect, _: &mut Context<Self>) {
263 // let mut rooms: Vec<i32> = Vec::new();
266 if self.sessions.remove(&msg.id).is_some() {
267 // remove session from all rooms
268 for (_id, sessions) in &mut self.rooms {
269 if sessions.remove(&msg.id) {
277 /// Handler for Message message.
278 impl Handler<StandardMessage> for ChatServer {
279 type Result = MessageResult<StandardMessage>;
281 fn handle(&mut self, msg: StandardMessage, _: &mut Context<Self>) -> Self::Result {
282 let msg_out = match parse_json_message(self, msg) {
284 Err(e) => e.to_string(),
287 MessageResult(msg_out)
291 fn parse_json_message(chat: &mut ChatServer, msg: StandardMessage) -> Result<String, Error> {
292 let json: Value = serde_json::from_str(&msg.msg)?;
293 let data = &json["data"].to_string();
294 let op = &json["op"].as_str().ok_or(APIError {
295 op: "Unknown op type".to_string(),
296 message: format!("Unknown op type"),
299 let user_operation: UserOperation = UserOperation::from_str(&op)?;
301 match user_operation {
302 UserOperation::Login => {
303 let login: Login = serde_json::from_str(data)?;
304 let res = Oper::new(user_operation, login).perform()?;
305 Ok(serde_json::to_string(&res)?)
307 UserOperation::Register => {
308 chat.check_rate_limit_register(msg.id)?;
309 let register: Register = serde_json::from_str(data)?;
310 let res = Oper::new(user_operation, register).perform()?;
311 Ok(serde_json::to_string(&res)?)
313 UserOperation::GetUserDetails => {
314 let get_user_details: GetUserDetails = serde_json::from_str(data)?;
315 let res = Oper::new(user_operation, get_user_details).perform()?;
316 Ok(serde_json::to_string(&res)?)
318 UserOperation::SaveUserSettings => {
319 let save_user_settings: SaveUserSettings = serde_json::from_str(data)?;
320 let res = Oper::new(user_operation, save_user_settings).perform()?;
321 Ok(serde_json::to_string(&res)?)
323 UserOperation::AddAdmin => {
324 let add_admin: AddAdmin = serde_json::from_str(data)?;
325 let res = Oper::new(user_operation, add_admin).perform()?;
326 Ok(serde_json::to_string(&res)?)
328 UserOperation::BanUser => {
329 let ban_user: BanUser = serde_json::from_str(data)?;
330 let res = Oper::new(user_operation, ban_user).perform()?;
331 Ok(serde_json::to_string(&res)?)
333 UserOperation::GetReplies => {
334 let get_replies: GetReplies = serde_json::from_str(data)?;
335 let res = Oper::new(user_operation, get_replies).perform()?;
336 Ok(serde_json::to_string(&res)?)
338 UserOperation::MarkAllAsRead => {
339 let mark_all_as_read: MarkAllAsRead = serde_json::from_str(data)?;
340 let res = Oper::new(user_operation, mark_all_as_read).perform()?;
341 Ok(serde_json::to_string(&res)?)
343 UserOperation::GetCommunity => {
344 let get_community: GetCommunity = serde_json::from_str(data)?;
345 let res = Oper::new(user_operation, get_community).perform()?;
346 Ok(serde_json::to_string(&res)?)
348 UserOperation::ListCommunities => {
349 let list_communities: ListCommunities = serde_json::from_str(data)?;
350 let res = Oper::new(user_operation, list_communities).perform()?;
351 Ok(serde_json::to_string(&res)?)
353 UserOperation::CreateCommunity => {
354 chat.check_rate_limit_register(msg.id)?;
355 let create_community: CreateCommunity = serde_json::from_str(data)?;
356 let res = Oper::new(user_operation, create_community).perform()?;
357 Ok(serde_json::to_string(&res)?)
359 UserOperation::EditCommunity => {
360 let edit_community: EditCommunity = serde_json::from_str(data)?;
361 let res = Oper::new(user_operation, edit_community).perform()?;
362 let mut community_sent: CommunityResponse = res.clone();
363 community_sent.community.user_id = None;
364 community_sent.community.subscribed = None;
365 let community_sent_str = serde_json::to_string(&community_sent)?;
366 chat.send_community_message(&community_sent.community.id, &community_sent_str, msg.id)?;
367 Ok(serde_json::to_string(&res)?)
369 UserOperation::FollowCommunity => {
370 let follow_community: FollowCommunity = serde_json::from_str(data)?;
371 let res = Oper::new(user_operation, follow_community).perform()?;
372 Ok(serde_json::to_string(&res)?)
374 UserOperation::GetFollowedCommunities => {
375 let followed_communities: GetFollowedCommunities = serde_json::from_str(data)?;
376 let res = Oper::new(user_operation, followed_communities).perform()?;
377 Ok(serde_json::to_string(&res)?)
379 UserOperation::BanFromCommunity => {
380 let ban_from_community: BanFromCommunity = serde_json::from_str(data)?;
381 let community_id = ban_from_community.community_id;
382 let res = Oper::new(user_operation, ban_from_community).perform()?;
383 let res_str = serde_json::to_string(&res)?;
384 chat.send_community_message(&community_id, &res_str, msg.id)?;
387 UserOperation::AddModToCommunity => {
388 let mod_add_to_community: AddModToCommunity = serde_json::from_str(data)?;
389 let community_id = mod_add_to_community.community_id;
390 let res = Oper::new(user_operation, mod_add_to_community).perform()?;
391 let res_str = serde_json::to_string(&res)?;
392 chat.send_community_message(&community_id, &res_str, msg.id)?;
395 UserOperation::ListCategories => {
396 let list_categories: ListCategories = ListCategories;
397 let res = Oper::new(user_operation, list_categories).perform()?;
398 Ok(serde_json::to_string(&res)?)
400 UserOperation::CreatePost => {
401 chat.check_rate_limit_post(msg.id)?;
402 let create_post: CreatePost = serde_json::from_str(data)?;
403 let res = Oper::new(user_operation, create_post).perform()?;
404 Ok(serde_json::to_string(&res)?)
406 UserOperation::GetPost => {
407 let get_post: GetPost = serde_json::from_str(data)?;
408 chat.join_room(get_post.id, msg.id);
409 let res = Oper::new(user_operation, get_post).perform()?;
410 Ok(serde_json::to_string(&res)?)
412 UserOperation::GetPosts => {
413 let get_posts: GetPosts = serde_json::from_str(data)?;
414 let res = Oper::new(user_operation, get_posts).perform()?;
415 Ok(serde_json::to_string(&res)?)
417 UserOperation::CreatePostLike => {
418 chat.check_rate_limit_message(msg.id)?;
419 let create_post_like: CreatePostLike = serde_json::from_str(data)?;
420 let res = Oper::new(user_operation, create_post_like).perform()?;
421 Ok(serde_json::to_string(&res)?)
423 UserOperation::EditPost => {
424 let edit_post: EditPost = serde_json::from_str(data)?;
425 let res = Oper::new(user_operation, edit_post).perform()?;
426 let mut post_sent = res.clone();
427 post_sent.post.my_vote = None;
428 let post_sent_str = serde_json::to_string(&post_sent)?;
429 chat.send_room_message(&post_sent.post.id, &post_sent_str, msg.id);
430 Ok(serde_json::to_string(&res)?)
432 UserOperation::SavePost => {
433 let save_post: SavePost = serde_json::from_str(data)?;
434 let res = Oper::new(user_operation, save_post).perform()?;
435 Ok(serde_json::to_string(&res)?)
437 UserOperation::CreateComment => {
438 chat.check_rate_limit_message(msg.id)?;
439 let create_comment: CreateComment = serde_json::from_str(data)?;
440 let post_id = create_comment.post_id;
441 let res = Oper::new(user_operation, create_comment).perform()?;
442 let mut comment_sent = res.clone();
443 comment_sent.comment.my_vote = None;
444 comment_sent.comment.user_id = None;
445 let comment_sent_str = serde_json::to_string(&comment_sent)?;
446 chat.send_room_message(&post_id, &comment_sent_str, msg.id);
447 Ok(serde_json::to_string(&res)?)
449 UserOperation::EditComment => {
450 let edit_comment: EditComment = serde_json::from_str(data)?;
451 let post_id = edit_comment.post_id;
452 let res = Oper::new(user_operation, edit_comment).perform()?;
453 let mut comment_sent = res.clone();
454 comment_sent.comment.my_vote = None;
455 comment_sent.comment.user_id = None;
456 let comment_sent_str = serde_json::to_string(&comment_sent)?;
457 chat.send_room_message(&post_id, &comment_sent_str, msg.id);
458 Ok(serde_json::to_string(&res)?)
460 UserOperation::SaveComment => {
461 let save_comment: SaveComment = serde_json::from_str(data)?;
462 let res = Oper::new(user_operation, save_comment).perform()?;
463 Ok(serde_json::to_string(&res)?)
465 UserOperation::CreateCommentLike => {
466 chat.check_rate_limit_message(msg.id)?;
467 let create_comment_like: CreateCommentLike = serde_json::from_str(data)?;
468 let post_id = create_comment_like.post_id;
469 let res = Oper::new(user_operation, create_comment_like).perform()?;
470 let mut comment_sent = res.clone();
471 comment_sent.comment.my_vote = None;
472 comment_sent.comment.user_id = None;
473 let comment_sent_str = serde_json::to_string(&comment_sent)?;
474 chat.send_room_message(&post_id, &comment_sent_str, msg.id);
475 Ok(serde_json::to_string(&res)?)
477 UserOperation::GetModlog => {
478 let get_modlog: GetModlog = serde_json::from_str(data)?;
479 let res = Oper::new(user_operation, get_modlog).perform()?;
480 Ok(serde_json::to_string(&res)?)
482 UserOperation::CreateSite => {
483 let create_site: CreateSite = serde_json::from_str(data)?;
484 let res = Oper::new(user_operation, create_site).perform()?;
485 Ok(serde_json::to_string(&res)?)
487 UserOperation::EditSite => {
488 let edit_site: EditSite = serde_json::from_str(data)?;
489 let res = Oper::new(user_operation, edit_site).perform()?;
490 Ok(serde_json::to_string(&res)?)
492 UserOperation::GetSite => {
493 let get_site: GetSite = serde_json::from_str(data)?;
494 let res = Oper::new(user_operation, get_site).perform()?;
495 Ok(serde_json::to_string(&res)?)
497 UserOperation::Search => {
498 let search: Search = serde_json::from_str(data)?;
499 let res = Oper::new(user_operation, search).perform()?;
500 Ok(serde_json::to_string(&res)?)
502 UserOperation::TransferCommunity => {
503 let transfer_community: TransferCommunity = serde_json::from_str(data)?;
504 let res = Oper::new(user_operation, transfer_community).perform()?;
505 Ok(serde_json::to_string(&res)?)
507 UserOperation::TransferSite => {
508 let transfer_site: TransferSite = serde_json::from_str(data)?;
509 let res = Oper::new(user_operation, transfer_site).perform()?;
510 Ok(serde_json::to_string(&res)?)