Adding websocket notification system.

- HTTP and APUB clients can now send live updating messages to websocket
  clients
- Rate limiting now affects both HTTP and websockets
- Rate limiting / Websocket logic is now moved into the API Perform
  functions.
- TODO This broke getting current online users, but that will have to
  wait for the perform trait to be made async.
- Fixes #446

1 files changed, 137 insertions, 16 deletions

diff --git a/server/src/api/comment.rs b/server/src/api/comment.rs
index 8373a338..8e398c9a 100644
--- a/server/src/api/comment.rs
+++ b/server/src/api/comment.rs
@@ -1,9 +1,4 @@
 use super::*;
-use crate::send_email;
-use crate::settings::Settings;
-use diesel::PgConnection;
-use log::error;
-use std::str::FromStr;
 
 #[derive(Serialize, Deserialize)]
 pub struct CreateComment {
@@ -65,7 +60,12 @@ pub struct GetCommentsResponse {
 }
 
 impl Perform<CommentResponse> for Oper<CreateComment> {
-  fn perform(&self, conn: &PgConnection) -> Result<CommentResponse, Error> {
+  fn perform(
+    &self,
+    pool: Pool<ConnectionManager<PgConnection>>,
+    websocket_info: Option<WebsocketInfo>,
+    rate_limit_info: Option<RateLimitInfo>,
+  ) -> Result<CommentResponse, Error> {
     let data: &CreateComment = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -77,6 +77,15 @@ impl Perform<CommentResponse> for Oper<CreateComment> {
 
     let hostname = &format!("https://{}", Settings::get().hostname);
 
+    if let Some(rl) = rate_limit_info {
+      rl.rate_limiter
+        .lock()
+        .unwrap()
+        .check_rate_limit_message(&rl.ip, false)?;
+    }
+
+    let conn = pool.get()?;
+
     // Check for a community ban
     let post = Post::read(&conn, data.post_id)?;
     if CommunityUserBanView::get(&conn, user_id, post.community_id).is_ok() {
@@ -223,15 +232,34 @@ impl Perform<CommentResponse> for Oper<CreateComment> {
 
     let comment_view = CommentView::read(&conn, inserted_comment.id, Some(user_id))?;
 
-    Ok(CommentResponse {
+    let mut res = CommentResponse {
       comment: comment_view,
       recipient_ids,
-    })
+    };
+
+    if let Some(ws) = websocket_info {
+      ws.chatserver.do_send(SendComment {
+        op: UserOperation::CreateComment,
+        comment: res.clone(),
+        my_id: ws.id,
+      });
+
+      // strip out the recipient_ids, so that
+      // users don't get double notifs
+      res.recipient_ids = Vec::new();
+    }
+
+    Ok(res)
   }
 }
 
 impl Perform<CommentResponse> for Oper<EditComment> {
-  fn perform(&self, conn: &PgConnection) -> Result<CommentResponse, Error> {
+  fn perform(
+    &self,
+    pool: Pool<ConnectionManager<PgConnection>>,
+    websocket_info: Option<WebsocketInfo>,
+    rate_limit_info: Option<RateLimitInfo>,
+  ) -> Result<CommentResponse, Error> {
     let data: &EditComment = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -241,6 +269,15 @@ impl Perform<CommentResponse> for Oper<EditComment> {
 
     let user_id = claims.id;
 
+    if let Some(rl) = rate_limit_info {
+      rl.rate_limiter
+        .lock()
+        .unwrap()
+        .check_rate_limit_message(&rl.ip, false)?;
+    }
+
+    let conn = pool.get()?;
+
     let orig_comment = CommentView::read(&conn, data.edit_id, None)?;
 
     // You are allowed to mark the comment as read even if you're banned.
@@ -353,15 +390,34 @@ impl Perform<CommentResponse> for Oper<EditComment> {
 
     let comment_view = CommentView::read(&conn, data.edit_id, Some(user_id))?;
 
-    Ok(CommentResponse {
+    let mut res = CommentResponse {
       comment: comment_view,
       recipient_ids,
-    })
+    };
+
+    if let Some(ws) = websocket_info {
+      ws.chatserver.do_send(SendComment {
+        op: UserOperation::EditComment,
+        comment: res.clone(),
+        my_id: ws.id,
+      });
+
+      // strip out the recipient_ids, so that
+      // users don't get double notifs
+      res.recipient_ids = Vec::new();
+    }
+
+    Ok(res)
   }
 }
 
 impl Perform<CommentResponse> for Oper<SaveComment> {
-  fn perform(&self, conn: &PgConnection) -> Result<CommentResponse, Error> {
+  fn perform(
+    &self,
+    pool: Pool<ConnectionManager<PgConnection>>,
+    _websocket_info: Option<WebsocketInfo>,
+    rate_limit_info: Option<RateLimitInfo>,
+  ) -> Result<CommentResponse, Error> {
     let data: &SaveComment = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -376,6 +432,15 @@ impl Perform<CommentResponse> for Oper<SaveComment> {
       user_id,
     };
 
+    if let Some(rl) = rate_limit_info {
+      rl.rate_limiter
+        .lock()
+        .unwrap()
+        .check_rate_limit_message(&rl.ip, false)?;
+    }
+
+    let conn = pool.get()?;
+
     if data.save {
       match CommentSaved::save(&conn, &comment_saved_form) {
         Ok(comment) => comment,
@@ -398,7 +463,12 @@ impl Perform<CommentResponse> for Oper<SaveComment> {
 }
 
 impl Perform<CommentResponse> for Oper<CreateCommentLike> {
-  fn perform(&self, conn: &PgConnection) -> Result<CommentResponse, Error> {
+  fn perform(
+    &self,
+    pool: Pool<ConnectionManager<PgConnection>>,
+    websocket_info: Option<WebsocketInfo>,
+    rate_limit_info: Option<RateLimitInfo>,
+  ) -> Result<CommentResponse, Error> {
     let data: &CreateCommentLike = &self.data;
 
     let claims = match Claims::decode(&data.auth) {
@@ -410,6 +480,15 @@ impl Perform<CommentResponse> for Oper<CreateCommentLike> {
 
     let mut recipient_ids = Vec::new();
 
+    if let Some(rl) = rate_limit_info {
+      rl.rate_limiter
+        .lock()
+        .unwrap()
+        .check_rate_limit_message(&rl.ip, false)?;
+    }
+
+    let conn = pool.get()?;
+
     // Don't do a downvote if site has downvotes disabled
     if data.score == -1 {
       let site = SiteView::read(&conn)?;
@@ -467,15 +546,34 @@ impl Perform<CommentResponse> for Oper<CreateCommentLike> {
     // Have to refetch the comment to get the current state
     let liked_comment = CommentView::read(&conn, data.comment_id, Some(user_id))?;
 
-    Ok(CommentResponse {
+    let mut res = CommentResponse {
       comment: liked_comment,
       recipient_ids,
-    })
+    };
+
+    if let Some(ws) = websocket_info {
+      ws.chatserver.do_send(SendComment {
+        op: UserOperation::CreateCommentLike,
+        comment: res.clone(),
+        my_id: ws.id,
+      });
+
+      // strip out the recipient_ids, so that
+      // users don't get double notifs
+      res.recipient_ids = Vec::new();
+    }
+
+    Ok(res)
   }
 }
 
 impl Perform<GetCommentsResponse> for Oper<GetComments> {
-  fn perform(&self, conn: &PgConnection) -> Result<GetCommentsResponse, Error> {
+  fn perform(
+    &self,
+    pool: Pool<ConnectionManager<PgConnection>>,
+    websocket_info: Option<WebsocketInfo>,
+    rate_limit_info: Option<RateLimitInfo>,
+  ) -> Result<GetCommentsResponse, Error> {
     let data: &GetComments = &self.data;
 
     let user_claims: Option<Claims> = match &data.auth {
@@ -494,6 +592,15 @@ impl Perform<GetCommentsResponse> for Oper<GetComments> {
     let type_ = ListingType::from_str(&data.type_)?;
     let sort = SortType::from_str(&data.sort)?;
 
+    if let Some(rl) = rate_limit_info {
+      rl.rate_limiter
+        .lock()
+        .unwrap()
+        .check_rate_limit_message(&rl.ip, false)?;
+    }
+
+    let conn = pool.get()?;
+
     let comments = match CommentQueryBuilder::create(&conn)
       .listing_type(type_)
       .sort(&sort)
@@ -507,6 +614,20 @@ impl Perform<GetCommentsResponse> for Oper<GetComments> {
       Err(_e) => return Err(APIError::err("couldnt_get_comments").into()),
     };
 
+    if let Some(ws) = websocket_info {
+      // You don't need to join the specific community room, bc this is already handled by
+      // GetCommunity
+      if data.community_id.is_none() {
+        if let Some(id) = ws.id {
+          // 0 is the "all" community
+          ws.chatserver.do_send(JoinCommunityRoom {
+            community_id: 0,
+            id,
+          });
+        }
+      }
+    }
+
     Ok(GetCommentsResponse { comments })
   }
 }