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::*;
20 use crate::apub::puller::*;
23 /// Chat server sends this messages to session
25 pub struct WSMessage(pub String);
27 /// Message for chat server communications
29 /// New chat session is created
33 pub addr: Recipient<WSMessage>,
37 /// Session is disconnected
39 pub struct Disconnect {
44 /// Send message to specific room
46 pub struct ClientMessage {
47 /// Id of the client session
55 #[derive(Serialize, Deserialize)]
56 pub struct StandardMessage {
57 /// Id of the client session
63 impl actix::Message for StandardMessage {
68 pub struct RateLimitBucket {
69 last_checked: SystemTime,
73 pub struct SessionInfo {
74 pub addr: Recipient<WSMessage>,
78 /// `ChatServer` manages chat rooms and responsible for coordinating chat
79 /// session. implementation is super primitive
80 pub struct ChatServer {
81 sessions: HashMap<usize, SessionInfo>, // A map from generated random ID to session addr
82 rate_limits: HashMap<String, RateLimitBucket>,
83 rooms: HashMap<i32, HashSet<usize>>, // A map from room / post name to set of connectionIDs
87 impl Default for ChatServer {
88 fn default() -> ChatServer {
90 let rooms = HashMap::new();
93 sessions: HashMap::new(),
94 rate_limits: HashMap::new(),
96 rng: rand::thread_rng(),
102 /// Send message to all users in the room
103 fn send_room_message(&self, room: i32, message: &str, skip_id: usize) {
104 if let Some(sessions) = self.rooms.get(&room) {
107 if let Some(info) = self.sessions.get(id) {
108 let _ = info.addr.do_send(WSMessage(message.to_owned()));
115 fn join_room(&mut self, room_id: i32, id: usize) {
116 // remove session from all rooms
117 for sessions in self.rooms.values_mut() {
118 sessions.remove(&id);
121 // If the room doesn't exist yet
122 if self.rooms.get_mut(&room_id).is_none() {
123 self.rooms.insert(room_id, HashSet::new());
126 self.rooms.get_mut(&room_id).unwrap().insert(id);
129 fn send_community_message(
134 ) -> Result<(), Error> {
135 use crate::db::post_view::*;
137 let conn = establish_connection();
139 let posts = PostQueryBuilder::create(&conn)
140 .listing_type(ListingType::Community)
141 .sort(&SortType::New)
142 .for_community_id(community_id)
147 self.send_room_message(post.id, message, skip_id);
153 fn check_rate_limit_register(&mut self, id: usize) -> Result<(), Error> {
154 self.check_rate_limit_full(
156 Settings::get().rate_limit.register,
157 Settings::get().rate_limit.register_per_second,
161 fn check_rate_limit_post(&mut self, id: usize) -> Result<(), Error> {
162 self.check_rate_limit_full(
164 Settings::get().rate_limit.post,
165 Settings::get().rate_limit.post_per_second,
169 fn check_rate_limit_message(&mut self, id: usize) -> Result<(), Error> {
170 self.check_rate_limit_full(
172 Settings::get().rate_limit.message,
173 Settings::get().rate_limit.message_per_second,
177 #[allow(clippy::float_cmp)]
178 fn check_rate_limit_full(&mut self, id: usize, rate: i32, per: i32) -> Result<(), Error> {
179 if let Some(info) = self.sessions.get(&id) {
180 if let Some(rate_limit) = self.rate_limits.get_mut(&info.ip) {
182 if rate_limit.allowance == -2f64 {
183 rate_limit.allowance = rate as f64;
186 let current = SystemTime::now();
187 let time_passed = current.duration_since(rate_limit.last_checked)?.as_secs() as f64;
188 rate_limit.last_checked = current;
189 rate_limit.allowance += time_passed * (rate as f64 / per as f64);
190 if rate_limit.allowance > rate as f64 {
191 rate_limit.allowance = rate as f64;
194 if rate_limit.allowance < 1.0 {
196 "Rate limited IP: {}, time_passed: {}, allowance: {}",
197 &info.ip, time_passed, rate_limit.allowance
201 op: "Rate Limit".to_string(),
202 message: format!("Too many requests. {} per {} seconds", rate, per),
207 rate_limit.allowance -= 1.0;
219 /// Make actor from `ChatServer`
220 impl Actor for ChatServer {
221 /// We are going to use simple Context, we just need ability to communicate
222 /// with other actors.
223 type Context = Context<Self>;
226 /// Handler for Connect message.
228 /// Register new session and assign unique id to this session
229 impl Handler<Connect> for ChatServer {
232 fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
233 // notify all users in same room
234 // self.send_room_message(&"Main".to_owned(), "Someone joined", 0);
236 // register session with random id
237 let id = self.rng.gen::<usize>();
238 println!("{} joined", &msg.ip);
240 self.sessions.insert(
244 ip: msg.ip.to_owned(),
248 if self.rate_limits.get(&msg.ip).is_none() {
249 self.rate_limits.insert(
252 last_checked: SystemTime::now(),
262 /// Handler for Disconnect message.
263 impl Handler<Disconnect> for ChatServer {
266 fn handle(&mut self, msg: Disconnect, _: &mut Context<Self>) {
267 // let mut rooms: Vec<i32> = Vec::new();
270 if self.sessions.remove(&msg.id).is_some() {
271 // remove session from all rooms
272 for sessions in self.rooms.values_mut() {
273 if sessions.remove(&msg.id) {
281 /// Handler for Message message.
282 impl Handler<StandardMessage> for ChatServer {
283 type Result = MessageResult<StandardMessage>;
285 fn handle(&mut self, msg: StandardMessage, _: &mut Context<Self>) -> Self::Result {
286 let msg_out = match parse_json_message(self, msg) {
288 Err(e) => e.to_string(),
291 MessageResult(msg_out)
295 fn parse_json_message(chat: &mut ChatServer, msg: StandardMessage) -> Result<String, Error> {
296 let json: Value = serde_json::from_str(&msg.msg)?;
297 let data = &json["data"].to_string();
298 let op = &json["op"].as_str().ok_or(APIError {
299 op: "Unknown op type".to_string(),
300 message: "Unknown op type".to_string(),
303 let user_operation: UserOperation = UserOperation::from_str(&op)?;
305 match user_operation {
306 UserOperation::Login => {
307 let login: Login = serde_json::from_str(data)?;
308 let res = Oper::new(user_operation, login).perform()?;
309 Ok(serde_json::to_string(&res)?)
311 UserOperation::Register => {
312 let register: Register = serde_json::from_str(data)?;
313 let res = Oper::new(user_operation, register).perform();
315 chat.check_rate_limit_register(msg.id)?;
317 Ok(serde_json::to_string(&res?)?)
319 UserOperation::GetUserDetails => {
320 let get_user_details: GetUserDetails = serde_json::from_str(data)?;
321 let res = Oper::new(user_operation, get_user_details).perform()?;
322 Ok(serde_json::to_string(&res)?)
324 UserOperation::SaveUserSettings => {
325 let save_user_settings: SaveUserSettings = serde_json::from_str(data)?;
326 let res = Oper::new(user_operation, save_user_settings).perform()?;
327 Ok(serde_json::to_string(&res)?)
329 UserOperation::AddAdmin => {
330 let add_admin: AddAdmin = serde_json::from_str(data)?;
331 let res = Oper::new(user_operation, add_admin).perform()?;
332 Ok(serde_json::to_string(&res)?)
334 UserOperation::BanUser => {
335 let ban_user: BanUser = serde_json::from_str(data)?;
336 let res = Oper::new(user_operation, ban_user).perform()?;
337 Ok(serde_json::to_string(&res)?)
339 UserOperation::GetReplies => {
340 let get_replies: GetReplies = serde_json::from_str(data)?;
341 let res = Oper::new(user_operation, get_replies).perform()?;
342 Ok(serde_json::to_string(&res)?)
344 UserOperation::GetUserMentions => {
345 let get_user_mentions: GetUserMentions = serde_json::from_str(data)?;
346 let res = Oper::new(user_operation, get_user_mentions).perform()?;
347 Ok(serde_json::to_string(&res)?)
349 UserOperation::EditUserMention => {
350 let edit_user_mention: EditUserMention = serde_json::from_str(data)?;
351 let res = Oper::new(user_operation, edit_user_mention).perform()?;
352 Ok(serde_json::to_string(&res)?)
354 UserOperation::MarkAllAsRead => {
355 let mark_all_as_read: MarkAllAsRead = serde_json::from_str(data)?;
356 let res = Oper::new(user_operation, mark_all_as_read).perform()?;
357 Ok(serde_json::to_string(&res)?)
359 UserOperation::GetCommunity => {
360 let mut get_community: GetCommunity = serde_json::from_str(data)?;
361 if Settings::get().federation_enabled && get_community.name.is_some() {
362 let name = &get_community.name.unwrap();
363 let remote_community = if name.contains("@") {
364 // TODO: need to support sort, filter etc for remote communities
365 get_remote_community(name.to_owned())?
367 get_community.name = Some(name.replace("!", ""));
368 Oper::new(user_operation, get_community).perform()?
370 Ok(serde_json::to_string(&remote_community)?)
372 let res = Oper::new(user_operation, get_community).perform()?;
373 Ok(serde_json::to_string(&res)?)
376 UserOperation::ListCommunities => {
377 if Settings::get().federation_enabled {
378 let res = get_all_communities()?;
379 let val = ListCommunitiesResponse {
380 op: UserOperation::ListCommunities.to_string(),
383 Ok(serde_json::to_string(&val)?)
385 let list_communities: ListCommunities = serde_json::from_str(data)?;
386 let res = Oper::new(user_operation, list_communities).perform()?;
387 Ok(serde_json::to_string(&res)?)
390 UserOperation::CreateCommunity => {
391 chat.check_rate_limit_register(msg.id)?;
392 let create_community: CreateCommunity = serde_json::from_str(data)?;
393 let res = Oper::new(user_operation, create_community).perform()?;
394 Ok(serde_json::to_string(&res)?)
396 UserOperation::EditCommunity => {
397 let edit_community: EditCommunity = serde_json::from_str(data)?;
398 let res = Oper::new(user_operation, edit_community).perform()?;
399 let mut community_sent: CommunityResponse = res.clone();
400 community_sent.community.user_id = None;
401 community_sent.community.subscribed = None;
402 let community_sent_str = serde_json::to_string(&community_sent)?;
403 chat.send_community_message(community_sent.community.id, &community_sent_str, msg.id)?;
404 Ok(serde_json::to_string(&res)?)
406 UserOperation::FollowCommunity => {
407 let follow_community: FollowCommunity = serde_json::from_str(data)?;
408 let res = Oper::new(user_operation, follow_community).perform()?;
409 Ok(serde_json::to_string(&res)?)
411 UserOperation::GetFollowedCommunities => {
412 let followed_communities: GetFollowedCommunities = serde_json::from_str(data)?;
413 let res = Oper::new(user_operation, followed_communities).perform()?;
414 Ok(serde_json::to_string(&res)?)
416 UserOperation::BanFromCommunity => {
417 let ban_from_community: BanFromCommunity = serde_json::from_str(data)?;
418 let community_id = ban_from_community.community_id;
419 let res = Oper::new(user_operation, ban_from_community).perform()?;
420 let res_str = serde_json::to_string(&res)?;
421 chat.send_community_message(community_id, &res_str, msg.id)?;
424 UserOperation::AddModToCommunity => {
425 let mod_add_to_community: AddModToCommunity = serde_json::from_str(data)?;
426 let community_id = mod_add_to_community.community_id;
427 let res = Oper::new(user_operation, mod_add_to_community).perform()?;
428 let res_str = serde_json::to_string(&res)?;
429 chat.send_community_message(community_id, &res_str, msg.id)?;
432 UserOperation::ListCategories => {
433 let list_categories: ListCategories = ListCategories;
434 let res = Oper::new(user_operation, list_categories).perform()?;
435 Ok(serde_json::to_string(&res)?)
437 UserOperation::CreatePost => {
438 chat.check_rate_limit_post(msg.id)?;
439 let create_post: CreatePost = serde_json::from_str(data)?;
440 let res = Oper::new(user_operation, create_post).perform()?;
441 Ok(serde_json::to_string(&res)?)
443 UserOperation::GetPost => {
444 let get_post: GetPost = serde_json::from_str(data)?;
445 chat.join_room(get_post.id, msg.id);
446 let res = Oper::new(user_operation, get_post).perform()?;
447 Ok(serde_json::to_string(&res)?)
449 UserOperation::GetPosts => {
450 let get_posts: GetPosts = serde_json::from_str(data)?;
451 let res = Oper::new(user_operation, get_posts).perform()?;
452 Ok(serde_json::to_string(&res)?)
454 UserOperation::CreatePostLike => {
455 chat.check_rate_limit_message(msg.id)?;
456 let create_post_like: CreatePostLike = serde_json::from_str(data)?;
457 let res = Oper::new(user_operation, create_post_like).perform()?;
458 Ok(serde_json::to_string(&res)?)
460 UserOperation::EditPost => {
461 let edit_post: EditPost = serde_json::from_str(data)?;
462 let res = Oper::new(user_operation, edit_post).perform()?;
463 let mut post_sent = res.clone();
464 post_sent.post.my_vote = None;
465 let post_sent_str = serde_json::to_string(&post_sent)?;
466 chat.send_room_message(post_sent.post.id, &post_sent_str, msg.id);
467 Ok(serde_json::to_string(&res)?)
469 UserOperation::SavePost => {
470 let save_post: SavePost = serde_json::from_str(data)?;
471 let res = Oper::new(user_operation, save_post).perform()?;
472 Ok(serde_json::to_string(&res)?)
474 UserOperation::CreateComment => {
475 chat.check_rate_limit_message(msg.id)?;
476 let create_comment: CreateComment = serde_json::from_str(data)?;
477 let post_id = create_comment.post_id;
478 let res = Oper::new(user_operation, create_comment).perform()?;
479 let mut comment_sent = res.clone();
480 comment_sent.comment.my_vote = None;
481 comment_sent.comment.user_id = None;
482 let comment_sent_str = serde_json::to_string(&comment_sent)?;
483 chat.send_room_message(post_id, &comment_sent_str, msg.id);
484 Ok(serde_json::to_string(&res)?)
486 UserOperation::EditComment => {
487 let edit_comment: EditComment = serde_json::from_str(data)?;
488 let post_id = edit_comment.post_id;
489 let res = Oper::new(user_operation, edit_comment).perform()?;
490 let mut comment_sent = res.clone();
491 comment_sent.comment.my_vote = None;
492 comment_sent.comment.user_id = None;
493 let comment_sent_str = serde_json::to_string(&comment_sent)?;
494 chat.send_room_message(post_id, &comment_sent_str, msg.id);
495 Ok(serde_json::to_string(&res)?)
497 UserOperation::SaveComment => {
498 let save_comment: SaveComment = serde_json::from_str(data)?;
499 let res = Oper::new(user_operation, save_comment).perform()?;
500 Ok(serde_json::to_string(&res)?)
502 UserOperation::CreateCommentLike => {
503 chat.check_rate_limit_message(msg.id)?;
504 let create_comment_like: CreateCommentLike = serde_json::from_str(data)?;
505 let post_id = create_comment_like.post_id;
506 let res = Oper::new(user_operation, create_comment_like).perform()?;
507 let mut comment_sent = res.clone();
508 comment_sent.comment.my_vote = None;
509 comment_sent.comment.user_id = None;
510 let comment_sent_str = serde_json::to_string(&comment_sent)?;
511 chat.send_room_message(post_id, &comment_sent_str, msg.id);
512 Ok(serde_json::to_string(&res)?)
514 UserOperation::GetModlog => {
515 let get_modlog: GetModlog = serde_json::from_str(data)?;
516 let res = Oper::new(user_operation, get_modlog).perform()?;
517 Ok(serde_json::to_string(&res)?)
519 UserOperation::CreateSite => {
520 let create_site: CreateSite = serde_json::from_str(data)?;
521 let res = Oper::new(user_operation, create_site).perform()?;
522 Ok(serde_json::to_string(&res)?)
524 UserOperation::EditSite => {
525 let edit_site: EditSite = serde_json::from_str(data)?;
526 let res = Oper::new(user_operation, edit_site).perform()?;
527 Ok(serde_json::to_string(&res)?)
529 UserOperation::GetSite => {
530 let online: usize = chat.sessions.len();
531 let get_site: GetSite = serde_json::from_str(data)?;
532 let mut res = Oper::new(user_operation, get_site).perform()?;
534 Ok(serde_json::to_string(&res)?)
536 UserOperation::Search => {
537 let search: Search = serde_json::from_str(data)?;
538 let res = Oper::new(user_operation, search).perform()?;
539 Ok(serde_json::to_string(&res)?)
541 UserOperation::TransferCommunity => {
542 let transfer_community: TransferCommunity = serde_json::from_str(data)?;
543 let res = Oper::new(user_operation, transfer_community).perform()?;
544 Ok(serde_json::to_string(&res)?)
546 UserOperation::TransferSite => {
547 let transfer_site: TransferSite = serde_json::from_str(data)?;
548 let res = Oper::new(user_operation, transfer_site).perform()?;
549 Ok(serde_json::to_string(&res)?)
551 UserOperation::DeleteAccount => {
552 let delete_account: DeleteAccount = serde_json::from_str(data)?;
553 let res = Oper::new(user_operation, delete_account).perform()?;
554 Ok(serde_json::to_string(&res)?)
556 UserOperation::PasswordReset => {
557 let password_reset: PasswordReset = serde_json::from_str(data)?;
558 let res = Oper::new(user_operation, password_reset).perform()?;
559 Ok(serde_json::to_string(&res)?)
561 UserOperation::PasswordChange => {
562 let password_change: PasswordChange = serde_json::from_str(data)?;
563 let res = Oper::new(user_operation, password_change).perform()?;
564 Ok(serde_json::to_string(&res)?)