LemmyNet/lemmy
2
0
Fork 0
mirror of https://github.com/LemmyNet/lemmy.git synced 2024-11-22 20:31:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Implement helper functions for websocket message sending

Browse source
This commit is contained in:
Felix Ableitner 2021-08-13 19:30:19 +02:00
parent bc6294a834
commit 2d346a1487
34 changed files with 501 additions and 795 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -1944,6 +1944,8 @@ dependencies = [
"lemmy_api_common",
"lemmy_db_queries",
"lemmy_db_schema",
"lemmy_db_views",
"lemmy_db_views_actor",
"lemmy_utils",
"log",
"rand 0.8.4",

31
crates/api/src/comment.rs
View file

@ -18,7 +18,7 @@ use lemmy_db_queries::{source::comment::Comment_, Likeable, Saveable};
use lemmy_db_schema::{source::comment::*, LocalUserId};
use lemmy_db_views::{comment_view::CommentView, local_user_view::LocalUserView};
use lemmy_utils::{ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendComment, LemmyContext, UserOperation};
use lemmy_websocket::{send::send_comment_ws_message, LemmyContext, UserOperation};
use std::convert::TryInto;
#[async_trait::async_trait(?Send)]
@ -206,26 +206,15 @@ impl Perform for CreateCommentLike {
.await?;
}
// Have to refetch the comment to get the current state
let comment_id = data.comment_id;
let person_id = local_user_view.person.id;
let liked_comment = blocking(context.pool(), move |conn| {
CommentView::read(conn, comment_id, Some(person_id))
})
.await??;
let res = CommentResponse {
comment_view: liked_comment,
recipient_ids,
form_id: None,
};
context.chat_server().do_send(SendComment {
op: UserOperation::CreateCommentLike,
comment: res.clone(),
send_comment_ws_message(
data.comment_id,
UserOperation::CreateCommentLike,
websocket_id,
});
Ok(res)
None,
Some(local_user_view.person.id),
recipient_ids,
context,
)
.await
}
}

69
crates/api/src/post.rs
View file

@ -24,7 +24,7 @@ use lemmy_db_queries::{source::post::Post_, Crud, Likeable, Saveable};
use lemmy_db_schema::source::{moderator::*, post::*};
use lemmy_db_views::post_view::PostView;
use lemmy_utils::{ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendPost, LemmyContext, UserOperation};
use lemmy_websocket::{send::send_post_ws_message, LemmyContext, UserOperation};
use std::convert::TryInto;
#[async_trait::async_trait(?Send)]
@ -96,23 +96,14 @@ impl Perform for CreatePostLike {
// Mark the post as read
mark_post_as_read(person_id, post_id, context.pool()).await?;
let post_id = data.post_id;
let person_id = local_user_view.person.id;
let post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, Some(person_id))
})
.await?
.map_err(|_| ApiError::err("couldnt_find_post"))?;
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op: UserOperation::CreatePostLike,
post: res.clone(),
send_post_ws_message(
data.post_id,
UserOperation::CreatePostLike,
websocket_id,
});
Ok(res)
Some(local_user_view.person.id),
context,
)
.await
}
}
@ -171,22 +162,14 @@ impl Perform for LockPost {
)
.await?;
// Refetch the post
let post_id = data.post_id;
let post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, Some(local_user_view.person.id))
})
.await??;
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op: UserOperation::LockPost,
post: res.clone(),
send_post_ws_message(
data.post_id,
UserOperation::LockPost,
websocket_id,
});
Ok(res)
Some(local_user_view.person.id),
context,
)
.await
}
}
@ -249,22 +232,14 @@ impl Perform for StickyPost {
)
.await?;
// Refetch the post
let post_id = data.post_id;
let post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, Some(local_user_view.person.id))
})
.await??;
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op: UserOperation::StickyPost,
post: res.clone(),
send_post_ws_message(
data.post_id,
UserOperation::StickyPost,
websocket_id,
});
Ok(res)
Some(local_user_view.person.id),
context,
)
.await
}
}

32
crates/api/src/private_message.rs
View file

@ -7,9 +7,8 @@ use lemmy_api_common::{
};
use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud};
use lemmy_db_schema::source::private_message::PrivateMessage;
use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::PrivateMessageView};
use lemmy_utils::{ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperation};
use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperation};
#[async_trait::async_trait(?Send)]
impl Perform for MarkPrivateMessageAsRead {
@ -43,32 +42,7 @@ impl Perform for MarkPrivateMessageAsRead {
.map_err(|_| ApiError::err("couldnt_update_private_message"))?;
// No need to send an apub update
let private_message_id = data.private_message_id;
let private_message_view = blocking(context.pool(), move |conn| {
PrivateMessageView::read(conn, private_message_id)
})
.await??;
let res = PrivateMessageResponse {
private_message_view,
};
// Send notifications to the local recipient, if one exists
let recipient_id = orig_private_message.recipient_id;
if let Ok(local_recipient) = blocking(context.pool(), move |conn| {
LocalUserView::read_person(conn, recipient_id)
})
.await?
{
let local_recipient_id = local_recipient.local_user.id;
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperation::MarkPrivateMessageAsRead,
response: res.clone(),
local_recipient_id,
websocket_id,
});
}
Ok(res)
let op = UserOperation::MarkPrivateMessageAsRead;
send_pm_ws_message(data.private_message_id, op, websocket_id, context).await
}
}

37
crates/api_crud/src/comment/create.rs
View file

@ -20,14 +20,13 @@ use lemmy_apub::{
};
use lemmy_db_queries::{source::comment::Comment_, Crud, Likeable};
use lemmy_db_schema::source::comment::*;
use lemmy_db_views::comment_view::CommentView;
use lemmy_utils::{
utils::{remove_slurs, scrape_text_for_mentions},
ApiError,
ConnectionId,
LemmyError,
};
use lemmy_websocket::{messages::SendComment, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_comment_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for CreateComment {
@ -137,37 +136,25 @@ impl PerformCrud for CreateComment {
)
.await?;
let person_id = local_user_view.person.id;
let mut comment_view = blocking(context.pool(), move |conn| {
CommentView::read(conn, inserted_comment.id, Some(person_id))
})
.await??;
// If its a comment to yourself, mark it as read
let comment_id = comment_view.comment.id;
if local_user_view.person.id == comment_view.get_recipient_id() {
if local_user_view.person.id == inserted_comment.creator_id {
let comment_id = inserted_comment.id;
blocking(context.pool(), move |conn| {
Comment::update_read(conn, comment_id, true)
})
.await?
.map_err(|_| ApiError::err("couldnt_update_comment"))?;
comment_view.comment.read = true;
}
let mut res = CommentResponse {
comment_view,
recipient_ids,
form_id: data.form_id.to_owned(),
};
context.chat_server().do_send(SendComment {
op: UserOperationCrud::CreateComment,
comment: res.clone(),
send_comment_ws_message(
inserted_comment.id,
UserOperationCrud::CreateComment,
websocket_id,
});
res.recipient_ids = Vec::new(); // Necessary to avoid doubles
Ok(res)
data.form_id.to_owned(),
Some(local_user_view.person.id),
recipient_ids,
context,
)
.await
}
}

93
crates/api_crud/src/comment/delete.rs
View file

@ -9,11 +9,11 @@ use lemmy_api_common::{
send_local_notifs,
};
use lemmy_apub::activities::deletion::{send_apub_delete, send_apub_remove};
use lemmy_db_queries::{source::comment::Comment_, Crud, DeleteableOrRemoveable};
use lemmy_db_schema::source::{comment::*, community::Community, moderator::*};
use lemmy_db_queries::{source::comment::Comment_, Crud};
use lemmy_db_schema::source::{comment::*, community::Community, moderator::*, post::Post};
use lemmy_db_views::comment_view::CommentView;
use lemmy_utils::{ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendComment, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_comment_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for DeleteComment {
@ -67,45 +67,28 @@ impl PerformCrud for DeleteComment {
)
.await?;
// Refetch it
let comment_id = data.comment_id;
let person_id = local_user_view.person.id;
let mut comment_view = blocking(context.pool(), move |conn| {
CommentView::read(conn, comment_id, Some(person_id))
})
.await??;
// Blank out deleted or removed info
if deleted {
comment_view.comment = comment_view.comment.blank_out_deleted_or_removed_info();
}
// Build the recipients
let comment_view_2 = comment_view.clone();
let mentions = vec![];
let post_id = updated_comment.post_id;
let post = blocking(context.pool(), move |conn| Post::read(conn, post_id)).await??;
let recipient_ids = send_local_notifs(
mentions,
vec![],
updated_comment,
local_user_view.person.clone(),
comment_view_2.post,
post,
context.pool(),
false,
)
.await?;
let res = CommentResponse {
comment_view,
recipient_ids,
form_id: None, // TODO a comment delete might clear forms?
};
context.chat_server().do_send(SendComment {
op: UserOperationCrud::DeleteComment,
comment: res.clone(),
send_comment_ws_message(
data.comment_id,
UserOperationCrud::DeleteComment,
websocket_id,
});
Ok(res)
None, // TODO a comment delete might clear forms?
Some(local_user_view.person.id),
recipient_ids,
context,
)
.await
}
}
@ -177,45 +160,27 @@ impl PerformCrud for RemoveComment {
)
.await?;
// Refetch it
let comment_id = data.comment_id;
let person_id = local_user_view.person.id;
let mut comment_view = blocking(context.pool(), move |conn| {
CommentView::read(conn, comment_id, Some(person_id))
})
.await??;
// Blank out deleted or removed info
if removed {
comment_view.comment = comment_view.comment.blank_out_deleted_or_removed_info();
}
// Build the recipients
let comment_view_2 = comment_view.clone();
let mentions = vec![];
let post_id = updated_comment.post_id;
let post = blocking(context.pool(), move |conn| Post::read(conn, post_id)).await??;
let recipient_ids = send_local_notifs(
mentions,
vec![],
updated_comment,
local_user_view.person.clone(),
comment_view_2.post,
post,
context.pool(),
false,
)
.await?;
let res = CommentResponse {
comment_view,
recipient_ids,
form_id: None, // TODO maybe this might clear other forms
};
context.chat_server().do_send(SendComment {
op: UserOperationCrud::RemoveComment,
comment: res.clone(),
send_comment_ws_message(
data.comment_id,
UserOperationCrud::RemoveComment,
websocket_id,
});
Ok(res)
None, // TODO maybe this might clear other forms
Some(local_user_view.person.id),
recipient_ids,
context,
)
.await
}
}

37
crates/api_crud/src/comment/update.rs
View file

@ -11,7 +11,7 @@ use lemmy_apub::activities::{
comment::create_or_update::CreateOrUpdateComment,
CreateOrUpdateType,
};
use lemmy_db_queries::{source::comment::Comment_, DeleteableOrRemoveable};
use lemmy_db_queries::source::comment::Comment_;
use lemmy_db_schema::source::comment::*;
use lemmy_db_views::comment_view::CommentView;
use lemmy_utils::{
@ -20,7 +20,7 @@ use lemmy_utils::{
ConnectionId,
LemmyError,
};
use lemmy_websocket::{messages::SendComment, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_comment_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for EditComment {
@ -83,30 +83,15 @@ impl PerformCrud for EditComment {
)
.await?;
let comment_id = data.comment_id;
let person_id = local_user_view.person.id;
let mut comment_view = blocking(context.pool(), move |conn| {
CommentView::read(conn, comment_id, Some(person_id))
})
.await??;
// Blank out deleted or removed info
if comment_view.comment.deleted || comment_view.comment.removed {
comment_view.comment = comment_view.comment.blank_out_deleted_or_removed_info();
}
let res = CommentResponse {
comment_view,
recipient_ids,
form_id: data.form_id.to_owned(),
};
context.chat_server().do_send(SendComment {
op: UserOperationCrud::EditComment,
comment: res.clone(),
send_comment_ws_message(
data.comment_id,
UserOperationCrud::EditComment,
websocket_id,
});
Ok(res)
data.form_id.to_owned(),
None,
recipient_ids,
context,
)
.await
}
}

67
crates/api_crud/src/community/delete.rs
View file

@ -1,18 +1,15 @@
use crate::{community::send_community_websocket, PerformCrud};
use crate::PerformCrud;
use actix_web::web::Data;
use lemmy_api_common::{blocking, community::*, get_local_user_view_from_jwt, is_admin};
use lemmy_apub::activities::deletion::{send_apub_delete, send_apub_remove};
use lemmy_db_queries::{source::community::Community_, Crud, DeleteableOrRemoveable};
use lemmy_db_queries::{source::community::Community_, Crud};
use lemmy_db_schema::source::{
community::*,
moderator::{ModRemoveCommunity, ModRemoveCommunityForm},
};
use lemmy_db_views_actor::{
community_moderator_view::CommunityModeratorView,
community_view::CommunityView,
};
use lemmy_db_views_actor::community_moderator_view::CommunityModeratorView;
use lemmy_utils::{utils::naive_from_unix, ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_community_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for DeleteCommunity {
@ -57,28 +54,14 @@ impl PerformCrud for DeleteCommunity {
)
.await?;
let community_id = data.community_id;
let person_id = local_user_view.person.id;
let mut community_view = blocking(context.pool(), move |conn| {
CommunityView::read(conn, community_id, Some(person_id))
})
.await??;
// Blank out deleted or removed info
if deleted {
community_view.community = community_view.community.blank_out_deleted_or_removed_info();
}
let res = CommunityResponse { community_view };
send_community_websocket(
&res,
context,
websocket_id,
send_community_ws_message(
data.community_id,
UserOperationCrud::DeleteCommunity,
);
Ok(res)
websocket_id,
Some(local_user_view.person.id),
context,
)
.await
}
}
@ -131,27 +114,13 @@ impl PerformCrud for RemoveCommunity {
)
.await?;
let community_id = data.community_id;
let person_id = local_user_view.person.id;
let mut community_view = blocking(context.pool(), move |conn| {
CommunityView::read(conn, community_id, Some(person_id))
})
.await??;
// Blank out deleted or removed info
if removed {
community_view.community = community_view.community.blank_out_deleted_or_removed_info();
}
let res = CommunityResponse { community_view };
send_community_websocket(
&res,
context,
websocket_id,
send_community_ws_message(
data.community_id,
UserOperationCrud::RemoveCommunity,
);
Ok(res)
websocket_id,
Some(local_user_view.person.id),
context,
)
.await
}
}

23
crates/api_crud/src/community/mod.rs
View file

@ -1,27 +1,4 @@
use actix_web::web::Data;
use lemmy_api_common::community::CommunityResponse;
use lemmy_utils::ConnectionId;
use lemmy_websocket::{messages::SendCommunityRoomMessage, LemmyContext, UserOperationCrud};
mod create;
mod delete;
mod read;
mod update;
pub(in crate::community) fn send_community_websocket(
res: &CommunityResponse,
context: &Data<LemmyContext>,
websocket_id: Option<ConnectionId>,
op: UserOperationCrud,
) {
// Strip out the person id and subscribed when sending to others
let mut res_sent = res.clone();
res_sent.community_view.subscribed = false;
context.chat_server().do_send(SendCommunityRoomMessage {
op,
response: res_sent,
community_id: res.community_view.community.id,
websocket_id,
});
}

35
crates/api_crud/src/community/update.rs
View file

@ -1,4 +1,4 @@
use crate::{community::send_community_websocket, PerformCrud};
use crate::PerformCrud;
use actix_web::web::Data;
use lemmy_api_common::{
blocking,
@ -6,18 +6,15 @@ use lemmy_api_common::{
get_local_user_view_from_jwt,
};
use lemmy_apub::CommunityType;
use lemmy_db_queries::{diesel_option_overwrite_to_url, Crud, DeleteableOrRemoveable};
use lemmy_db_queries::{diesel_option_overwrite_to_url, Crud};
use lemmy_db_schema::{
naive_now,
source::community::{Community, CommunityForm},
PersonId,
};
use lemmy_db_views_actor::{
community_moderator_view::CommunityModeratorView,
community_view::CommunityView,
};
use lemmy_db_views_actor::community_moderator_view::CommunityModeratorView;
use lemmy_utils::{utils::check_slurs_opt, ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_community_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for EditCommunity {
@ -76,27 +73,7 @@ impl PerformCrud for EditCommunity {
.send_update(local_user_view.person.to_owned(), context)
.await?;
let community_id = data.community_id;
let person_id = local_user_view.person.id;
let mut community_view = blocking(context.pool(), move |conn| {
CommunityView::read(conn, community_id, Some(person_id))
})
.await??;
// Blank out deleted or removed info
if community_view.community.deleted || community_view.community.removed {
community_view.community = community_view.community.blank_out_deleted_or_removed_info();
}
let res = CommunityResponse { community_view };
send_community_websocket(
&res,
context,
websocket_id,
UserOperationCrud::EditCommunity,
);
Ok(res)
let op = UserOperationCrud::EditCommunity;
send_community_ws_message(data.community_id, op, websocket_id, None, context).await
}
}

26
crates/api_crud/src/post/create.rs
View file

@ -19,7 +19,6 @@ use lemmy_apub::{
};
use lemmy_db_queries::{source::post::Post_, Crud, Likeable};
use lemmy_db_schema::source::post::*;
use lemmy_db_views::post_view::PostView;
use lemmy_utils::{
request::fetch_iframely_and_pictrs_data,
utils::{check_slurs, check_slurs_opt, clean_url_params, is_valid_post_title},
@ -27,7 +26,7 @@ use lemmy_utils::{
ConnectionId,
LemmyError,
};
use lemmy_websocket::{messages::SendPost, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_post_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for CreatePost {
@ -129,22 +128,13 @@ impl PerformCrud for CreatePost {
)
.await?;
// Refetch the view
let inserted_post_id = inserted_post.id;
let post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, inserted_post_id, Some(local_user_view.person.id))
})
.await?
.map_err(|_| ApiError::err("couldnt_find_post"))?;
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op: UserOperationCrud::CreatePost,
post: res.clone(),
send_post_ws_message(
inserted_post.id,
UserOperationCrud::CreatePost,
websocket_id,
});
Ok(res)
Some(local_user_view.person.id),
context,
)
.await
}
}

59
crates/api_crud/src/post/delete.rs
View file

@ -8,11 +8,10 @@ use lemmy_api_common::{
post::*,
};
use lemmy_apub::activities::deletion::{send_apub_delete, send_apub_remove};
use lemmy_db_queries::{source::post::Post_, Crud, DeleteableOrRemoveable};
use lemmy_db_queries::{source::post::Post_, Crud};
use lemmy_db_schema::source::{community::Community, moderator::*, post::*};
use lemmy_db_views::post_view::PostView;
use lemmy_utils::{ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendPost, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_post_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for DeletePost {
@ -63,26 +62,14 @@ impl PerformCrud for DeletePost {
)
.await?;
// Refetch the post
let post_id = data.post_id;
let mut post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, Some(local_user_view.person.id))
})
.await??;
if deleted {
post_view.post = post_view.post.blank_out_deleted_or_removed_info();
}
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op: UserOperationCrud::DeletePost,
post: res.clone(),
send_post_ws_message(
data.post_id,
UserOperationCrud::DeletePost,
websocket_id,
});
Ok(res)
Some(local_user_view.person.id),
context,
)
.await
}
}
@ -151,27 +138,13 @@ impl PerformCrud for RemovePost {
)
.await?;
// Refetch the post
let post_id = data.post_id;
let person_id = local_user_view.person.id;
let mut post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, Some(person_id))
})
.await??;
// Blank out deleted or removed info
if removed {
post_view.post = post_view.post.blank_out_deleted_or_removed_info();
}
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op: UserOperationCrud::RemovePost,
post: res.clone(),
send_post_ws_message(
data.post_id,
UserOperationCrud::RemovePost,
websocket_id,
});
Ok(res)
Some(local_user_view.person.id),
context,
)
.await
}
}

31
crates/api_crud/src/post/update.rs
View file

@ -2,9 +2,8 @@ use crate::PerformCrud;
use actix_web::web::Data;
use lemmy_api_common::{blocking, check_community_ban, get_local_user_view_from_jwt, post::*};
use lemmy_apub::activities::{post::create_or_update::CreateOrUpdatePost, CreateOrUpdateType};
use lemmy_db_queries::{source::post::Post_, Crud, DeleteableOrRemoveable};
use lemmy_db_queries::{source::post::Post_, Crud};
use lemmy_db_schema::{naive_now, source::post::*};
use lemmy_db_views::post_view::PostView;
use lemmy_utils::{
request::fetch_iframely_and_pictrs_data,
utils::{check_slurs_opt, clean_url_params, is_valid_post_title},
@ -12,7 +11,7 @@ use lemmy_utils::{
ConnectionId,
LemmyError,
};
use lemmy_websocket::{messages::SendPost, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_post_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for EditPost {
@ -100,25 +99,13 @@ impl PerformCrud for EditPost {
)
.await?;
let post_id = data.post_id;
let mut post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, Some(local_user_view.person.id))
})
.await??;
// Blank out deleted info
if post_view.post.deleted || post_view.post.removed {
post_view.post = post_view.post.blank_out_deleted_or_removed_info();
}
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op: UserOperationCrud::EditPost,
post: res.clone(),
send_post_ws_message(
data.post_id,
UserOperationCrud::EditPost,
websocket_id,
});
Ok(res)
Some(local_user_view.person.id),
context,
)
.await
}
}

35
crates/api_crud/src/private_message/create.rs
View file

@ -16,9 +16,9 @@ use lemmy_apub::{
};
use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud};
use lemmy_db_schema::source::private_message::{PrivateMessage, PrivateMessageForm};
use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::PrivateMessageView};
use lemmy_db_views::local_user_view::LocalUserView;
use lemmy_utils::{utils::remove_slurs, ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for CreatePrivateMessage {
@ -78,36 +78,27 @@ impl PerformCrud for CreatePrivateMessage {
)
.await?;
let private_message_view = blocking(context.pool(), move |conn| {
PrivateMessageView::read(conn, inserted_private_message.id)
})
.await??;
let res = send_pm_ws_message(
inserted_private_message.id,
UserOperationCrud::CreatePrivateMessage,
websocket_id,
context,
)
.await?;
let res = PrivateMessageResponse {
private_message_view,
};
// Send notifications to the local recipient, if one exists
// Send email to the local recipient, if one exists
if res.private_message_view.recipient.local {
let recipient_id = data.recipient_id;
if let Ok(local_recipient) = blocking(context.pool(), move |conn| {
let local_recipient = blocking(context.pool(), move |conn| {
LocalUserView::read_person(conn, recipient_id)
})
.await?
{
.await??;
send_email_to_user(
&local_recipient,
"Private Message from",
"Private Message",
&content_slurs_removed,
);
let local_recipient_id = local_recipient.local_user.id;
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperationCrud::CreatePrivateMessage,
response: res.clone(),
local_recipient_id,
websocket_id,
});
}
Ok(res)

39
crates/api_crud/src/private_message/delete.rs
View file

@ -11,9 +11,8 @@ use lemmy_apub::activities::private_message::{
};
use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud, DeleteableOrRemoveable};
use lemmy_db_schema::source::private_message::PrivateMessage;
use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::PrivateMessageView};
use lemmy_utils::{ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for DeletePrivateMessage {
@ -59,39 +58,7 @@ impl PerformCrud for DeletePrivateMessage {
.await?;
}
let private_message_id = data.private_message_id;
let mut private_message_view = blocking(context.pool(), move |conn| {
PrivateMessageView::read(conn, private_message_id)
})
.await??;
// Blank out deleted or removed info
if deleted {
private_message_view.private_message = private_message_view
.private_message
.blank_out_deleted_or_removed_info();
}
let res = PrivateMessageResponse {
private_message_view,
};
// Send notifications to the local recipient, if one exists
let recipient_id = orig_private_message.recipient_id;
if let Ok(local_recipient) = blocking(context.pool(), move |conn| {
LocalUserView::read_person(conn, recipient_id)
})
.await?
{
let local_recipient_id = local_recipient.local_user.id;
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperationCrud::DeletePrivateMessage,
response: res.clone(),
local_recipient_id,
websocket_id,
});
}
Ok(res)
let op = UserOperationCrud::DeletePrivateMessage;
send_pm_ws_message(data.private_message_id, op, websocket_id, context).await
}
}

41
crates/api_crud/src/private_message/update.rs
View file

@ -9,11 +9,10 @@ use lemmy_apub::activities::{
private_message::create_or_update::CreateOrUpdatePrivateMessage,
CreateOrUpdateType,
};
use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud, DeleteableOrRemoveable};
use lemmy_db_queries::{source::private_message::PrivateMessage_, Crud};
use lemmy_db_schema::source::private_message::PrivateMessage;
use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::PrivateMessageView};
use lemmy_utils::{utils::remove_slurs, ApiError, ConnectionId, LemmyError};
use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
#[async_trait::async_trait(?Send)]
impl PerformCrud for EditPrivateMessage {
@ -55,39 +54,7 @@ impl PerformCrud for EditPrivateMessage {
)
.await?;
let private_message_id = data.private_message_id;
let mut private_message_view = blocking(context.pool(), move |conn| {
PrivateMessageView::read(conn, private_message_id)
})
.await??;
// Blank out deleted or removed info
if private_message_view.private_message.deleted {
private_message_view.private_message = private_message_view
.private_message
.blank_out_deleted_or_removed_info();
}
let res = PrivateMessageResponse {
private_message_view,
};
// Send notifications to the local recipient, if one exists
let recipient_id = orig_private_message.recipient_id;
if let Ok(local_recipient) = blocking(context.pool(), move |conn| {
LocalUserView::read_person(conn, recipient_id)
})
.await?
{
let local_recipient_id = local_recipient.local_user.id;
context.chat_server().do_send(SendUserRoomMessage {
op: UserOperationCrud::EditPrivateMessage,
response: res.clone(),
local_recipient_id,
websocket_id,
});
}
Ok(res)
let op = UserOperationCrud::EditPrivateMessage;
send_pm_ws_message(data.private_message_id, op, websocket_id, context).await
}
}

10
crates/apub/src/activities/comment/create_or_update.rs
View file

@ -1,6 +1,6 @@
use crate::{
activities::{
comment::{collect_non_local_mentions, get_notif_recipients, send_websocket_message},
comment::{collect_non_local_mentions, get_notif_recipients},
community::announce::AnnouncableActivities,
extract_community,
generate_activity_id,
@ -24,7 +24,7 @@ use lemmy_apub_lib::{
use lemmy_db_queries::Crud;
use lemmy_db_schema::source::{comment::Comment, community::Community, person::Person, post::Post};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_comment_ws_message, LemmyContext, UserOperationCrud};
use serde::{Deserialize, Serialize};
use url::Url;
@ -116,7 +116,11 @@ impl ActivityHandler for CreateOrUpdateComment {
CreateOrUpdateType::Create => UserOperationCrud::CreateComment,
CreateOrUpdateType::Update => UserOperationCrud::EditComment,
};
send_websocket_message(comment.id, recipients, notif_type, context).await
send_comment_ws_message(
comment.id, notif_type, None, None, None, recipients, context,
)
.await?;
Ok(())
}
fn common(&self) -> &ActivityCommonFields {

37
crates/apub/src/activities/comment/mod.rs
View file

@ -5,21 +5,19 @@ use activitystreams::{
};
use anyhow::anyhow;
use itertools::Itertools;
use lemmy_api_common::{blocking, comment::CommentResponse, send_local_notifs, WebFingerResponse};
use lemmy_api_common::{blocking, send_local_notifs, WebFingerResponse};
use lemmy_db_queries::{Crud, DbPool};
use lemmy_db_schema::{
source::{comment::Comment, community::Community, person::Person, post::Post},
CommentId,
LocalUserId,
};
use lemmy_db_views::comment_view::CommentView;
use lemmy_utils::{
request::{retry, RecvError},
settings::structs::Settings,
utils::{scrape_text_for_mentions, MentionData},
LemmyError,
};
use lemmy_websocket::{messages::SendComment, LemmyContext};
use lemmy_websocket::LemmyContext;
use log::debug;
use reqwest::Client;
use url::Url;
@ -45,37 +43,6 @@ async fn get_notif_recipients(
send_local_notifs(mentions, comment.clone(), actor, post, context.pool(), true).await
}
// TODO: in many call sites we are setting an empty vec for recipient_ids, we should get the actual
// recipient actors from somewhere
pub(crate) async fn send_websocket_message<
OP: ToString + Send + lemmy_websocket::OperationType + 'static,
>(
comment_id: CommentId,
recipient_ids: Vec<LocalUserId>,
op: OP,
context: &LemmyContext,
) -> Result<(), LemmyError> {
// Refetch the view
let comment_view = blocking(context.pool(), move |conn| {
CommentView::read(conn, comment_id, None)
})
.await??;
let res = CommentResponse {
comment_view,
recipient_ids,
form_id: None,
};
context.chat_server().do_send(SendComment {
op,
comment: res,
websocket_id: None,
});
Ok(())
}
pub struct MentionsAndAddresses {
pub ccs: Vec<Url>,
pub inboxes: Vec<Url>,

30
crates/apub/src/activities/community/mod.rs
View file

@ -1,10 +1,8 @@
use crate::{check_is_apub_id_valid, CommunityType};
use itertools::Itertools;
use lemmy_api_common::{blocking, community::CommunityResponse};
use lemmy_db_schema::{source::community::Community, CommunityId};
use lemmy_db_views_actor::community_view::CommunityView;
use lemmy_db_schema::source::community::Community;
use lemmy_utils::{settings::structs::Settings, LemmyError};
use lemmy_websocket::{messages::SendCommunityRoomMessage, LemmyContext};
use lemmy_websocket::LemmyContext;
use url::Url;
pub mod add_mod;
@ -13,30 +11,6 @@ pub mod block_user;
pub mod undo_block_user;
pub mod update;
pub(crate) async fn send_websocket_message<
OP: ToString + Send + lemmy_websocket::OperationType + 'static,
>(
community_id: CommunityId,
op: OP,
context: &LemmyContext,
) -> Result<(), LemmyError> {
let community_view = blocking(context.pool(), move |conn| {
CommunityView::read(conn, community_id, None)
})
.await??;
let res = CommunityResponse { community_view };
context.chat_server().do_send(SendCommunityRoomMessage {
op,
response: res,
community_id,
websocket_id: None,
});
Ok(())
}
async fn list_community_follower_inboxes(
community: &Community,
additional_inboxes: Vec<Url>,

16
crates/apub/src/activities/community/update.rs
View file

@ -1,10 +1,5 @@
use crate::{
activities::{
community::send_websocket_message,
verify_activity,
verify_mod_action,
verify_person_in_community,
},
activities::{verify_activity, verify_mod_action, verify_person_in_community},
objects::community::Group,
};
use activitystreams::activity::kind::UpdateType;
@ -13,7 +8,7 @@ use lemmy_apub_lib::{values::PublicUrl, ActivityCommonFields, ActivityHandler};
use lemmy_db_queries::{ApubObject, Crud};
use lemmy_db_schema::source::community::{Community, CommunityForm};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_community_ws_message, LemmyContext, UserOperationCrud};
use url::Url;
/// This activity is received from a remote community mod, and updates the description or other
@ -71,12 +66,15 @@ impl ActivityHandler for UpdateCommunity {
})
.await??;
send_websocket_message(
send_community_ws_message(
updated_community.id,
UserOperationCrud::EditCommunity,
None,
None,
context,
)
.await
.await?;
Ok(())
}
fn common(&self) -> &ActivityCommonFields {

81
crates/apub/src/activities/deletion/delete.rs
View file

@ -1,13 +1,13 @@
use crate::{
activities::{
comment::send_websocket_message as send_comment_message,
community::{
announce::AnnouncableActivities,
send_websocket_message as send_community_message,
community::announce::AnnouncableActivities,
deletion::{
receive_delete_action,
verify_delete_activity,
DeletableObjects,
WebsocketMessages,
},
deletion::{send_apub_delete, verify_delete_activity, DeletableObjects},
generate_activity_id,
post::send_websocket_message as send_post_message,
verify_activity,
},
activity_queue::send_to_community_new,
@ -38,7 +38,11 @@ use lemmy_db_schema::source::{
post::Post,
};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{
send::{send_comment_ws_message_simple, send_community_ws_message, send_post_ws_message},
LemmyContext,
UserOperationCrud,
};
use url::Url;
/// This is very confusing, because there are four distinct cases to handle:
@ -99,7 +103,19 @@ impl ActivityHandler for Delete {
)
.await
} else {
self.receive_delete_action(context, request_counter).await
receive_delete_action(
&self.object,
&self.common.actor,
WebsocketMessages {
community: UserOperationCrud::DeleteCommunity,
post: UserOperationCrud::DeletePost,
comment: UserOperationCrud::DeleteComment,
},
true,
context,
request_counter,
)
.await
}
}
@ -134,46 +150,6 @@ impl Delete {
let activity = AnnouncableActivities::Delete(delete);
send_to_community_new(activity, &id, actor, community, vec![], context).await
}
// TODO: would be nice if we could merge this and receive_delete_action() into a single method
// (or merge with receive_undo_delete_action() instead)
async fn receive_delete_action(
&self,
context: &LemmyContext,
request_counter: &mut i32,
) -> Result<(), LemmyError> {
use UserOperationCrud::*;
match DeletableObjects::read_from_db(&self.object, context).await? {
DeletableObjects::Community(community) => {
if community.local {
let mod_ =
get_or_fetch_and_upsert_person(&self.common.actor, context, request_counter).await?;
let object = community.actor_id();
send_apub_delete(&mod_, &community.clone(), object, true, context).await?;
}
let deleted_community = blocking(context.pool(), move |conn| {
Community::update_deleted(conn, community.id, true)
})
.await??;
send_community_message(deleted_community.id, DeleteCommunity, context).await
}
DeletableObjects::Post(post) => {
let deleted_post = blocking(context.pool(), move |conn| {
Post::update_deleted(conn, post.id, true)
})
.await??;
send_post_message(deleted_post.id, DeletePost, context).await
}
DeletableObjects::Comment(comment) => {
let deleted_comment = blocking(context.pool(), move |conn| {
Comment::update_deleted(conn, comment.id, true)
})
.await??;
send_comment_message(deleted_comment.id, vec![], DeleteComment, context).await
}
}
}
}
// TODO: reason is optional for compat with v0.11, make it mandatory after removing the migration
@ -207,7 +183,7 @@ pub(in crate::activities) async fn receive_remove_action(
})
.await??;
send_community_message(deleted_community.id, RemoveCommunity, context).await
send_community_ws_message(deleted_community.id, RemoveCommunity, None, None, context).await?;
}
DeletableObjects::Post(post) => {
let form = ModRemovePostForm {
@ -224,7 +200,8 @@ pub(in crate::activities) async fn receive_remove_action(
Post::update_removed(conn, post.id, true)
})
.await??;
send_post_message(removed_post.id, RemovePost, context).await
send_post_ws_message(removed_post.id, RemovePost, None, None, context).await?;
}
DeletableObjects::Comment(comment) => {
let form = ModRemoveCommentForm {
@ -241,7 +218,9 @@ pub(in crate::activities) async fn receive_remove_action(
Comment::update_removed(conn, comment.id, true)
})
.await??;
send_comment_message(removed_comment.id, vec![], RemoveComment, context).await
send_comment_ws_message_simple(removed_comment.id, RemoveComment, context).await?;
}
}
Ok(())
}

61
crates/apub/src/activities/deletion/mod.rs
View file

@ -4,17 +4,25 @@ use crate::{
verify_mod_action,
verify_person_in_community,
},
fetcher::person::get_or_fetch_and_upsert_person,
ActorType,
};
use lemmy_api_common::blocking;
use lemmy_apub_lib::{verify_domains_match, ActivityCommonFields};
use lemmy_db_queries::ApubObject;
use lemmy_db_queries::{
source::{comment::Comment_, community::Community_, post::Post_},
ApubObject,
};
use lemmy_db_schema::{
source::{comment::Comment, community::Community, person::Person, post::Post},
DbUrl,
};
use lemmy_utils::LemmyError;
use lemmy_websocket::LemmyContext;
use lemmy_websocket::{
send::{send_comment_ws_message_simple, send_community_ws_message, send_post_ws_message},
LemmyContext,
UserOperationCrud,
};
use url::Url;
pub mod delete;
@ -150,3 +158,52 @@ async fn verify_delete_activity_post_or_comment(
}
Ok(())
}
struct WebsocketMessages {
community: UserOperationCrud,
post: UserOperationCrud,
comment: UserOperationCrud,
}
/// Write deletion or restoring of an object to the database, and send websocket message.
/// TODO: we should do something similar for receive_remove_action(), but its much more complicated
/// because of the mod log
async fn receive_delete_action(
object: &Url,
actor: &Url,
ws_messages: WebsocketMessages,
deleted: bool,
context: &LemmyContext,
request_counter: &mut i32,
) -> Result<(), LemmyError> {
match DeletableObjects::read_from_db(object, context).await? {
DeletableObjects::Community(community) => {
if community.local {
let mod_ = get_or_fetch_and_upsert_person(actor, context, request_counter).await?;
let object = community.actor_id();
send_apub_delete(&mod_, &community.clone(), object, true, context).await?;
}
let community = blocking(context.pool(), move |conn| {
Community::update_deleted(conn, community.id, deleted)
})
.await??;
send_community_ws_message(community.id, ws_messages.community, None, None, context).await?;
}
DeletableObjects::Post(post) => {
let deleted_post = blocking(context.pool(), move |conn| {
Post::update_deleted(conn, post.id, deleted)
})
.await??;
send_post_ws_message(deleted_post.id, ws_messages.post, None, None, context).await?;
}
DeletableObjects::Comment(comment) => {
let deleted_comment = blocking(context.pool(), move |conn| {
Comment::update_deleted(conn, comment.id, deleted)
})
.await??;
send_comment_ws_message_simple(deleted_comment.id, ws_messages.comment, context).await?;
}
}
Ok(())
}

86
crates/apub/src/activities/deletion/undo_delete.rs
View file

@ -1,18 +1,18 @@
use crate::{
activities::{
comment::send_websocket_message as send_comment_message,
community::{
announce::AnnouncableActivities,
send_websocket_message as send_community_message,
community::announce::AnnouncableActivities,
deletion::{
delete::Delete,
receive_delete_action,
verify_delete_activity,
DeletableObjects,
WebsocketMessages,
},
deletion::{delete::Delete, send_apub_delete, verify_delete_activity, DeletableObjects},
generate_activity_id,
post::send_websocket_message as send_post_message,
verify_activity,
},
activity_queue::send_to_community_new,
extensions::context::lemmy_context,
fetcher::person::get_or_fetch_and_upsert_person,
ActorType,
};
use activitystreams::activity::kind::{DeleteType, UndoType};
@ -22,7 +22,11 @@ use lemmy_apub_lib::{values::PublicUrl, ActivityCommonFields, ActivityHandler};
use lemmy_db_queries::source::{comment::Comment_, community::Community_, post::Post_};
use lemmy_db_schema::source::{comment::Comment, community::Community, person::Person, post::Post};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{
send::{send_comment_ws_message_simple, send_community_ws_message, send_post_ws_message},
LemmyContext,
UserOperationCrud,
};
use url::Url;
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
@ -66,8 +70,18 @@ impl ActivityHandler for UndoDelete {
if self.object.summary.is_some() {
UndoDelete::receive_undo_remove_action(&self.object.object, context).await
} else {
self
.receive_undo_delete_action(context, request_counter)
receive_delete_action(
&self.object.object,
&self.common.actor,
WebsocketMessages {
community: UserOperationCrud::EditCommunity,
post: UserOperationCrud::EditPost,
comment: UserOperationCrud::EditComment,
},
false,
context,
request_counter,
)
.await
}
}
@ -116,50 +130,6 @@ impl UndoDelete {
let activity = AnnouncableActivities::UndoDelete(undo);
send_to_community_new(activity, &id, actor, community, vec![], context).await
}
async fn receive_undo_delete_action(
&self,
context: &LemmyContext,
request_counter: &mut i32,
) -> Result<(), LemmyError> {
use UserOperationCrud::*;
let object = DeletableObjects::read_from_db(&self.object.object, context).await?;
match object {
DeletableObjects::Community(community) => {
if community.local {
let mod_ =
get_or_fetch_and_upsert_person(&self.common.actor, context, request_counter).await?;
send_apub_delete(
&mod_,
&community.clone(),
community.actor_id(),
false,
context,
)
.await?;
}
let deleted_community = blocking(context.pool(), move |conn| {
Community::update_deleted(conn, community.id, false)
})
.await??;
send_community_message(deleted_community.id, EditCommunity, context).await
}
DeletableObjects::Post(post) => {
let deleted_post = blocking(context.pool(), move |conn| {
Post::update_deleted(conn, post.id, false)
})
.await??;
send_post_message(deleted_post.id, EditPost, context).await
}
DeletableObjects::Comment(comment) => {
let deleted_comment = blocking(context.pool(), move |conn| {
Comment::update_deleted(conn, comment.id, false)
})
.await??;
send_comment_message(deleted_comment.id, vec![], EditComment, context).await
}
}
}
pub(in crate::activities) async fn receive_undo_remove_action(
object: &Url,
@ -175,23 +145,23 @@ impl UndoDelete {
Community::update_removed(conn, community.id, false)
})
.await??;
send_community_message(deleted_community.id, EditCommunity, context).await
send_community_ws_message(deleted_community.id, EditCommunity, None, None, context).await?;
}
DeletableObjects::Post(post) => {
let removed_post = blocking(context.pool(), move |conn| {
Post::update_removed(conn, post.id, false)
})
.await??;
send_post_message(removed_post.id, EditPost, context).await
send_post_ws_message(removed_post.id, EditPost, None, None, context).await?;
}
DeletableObjects::Comment(comment) => {
let removed_comment = blocking(context.pool(), move |conn| {
Comment::update_removed(conn, comment.id, false)
})
.await??;
send_comment_message(removed_comment.id, vec![], EditComment, context).await
send_comment_ws_message_simple(removed_comment.id, EditComment, context).await?;
}
}
Ok(())
}
}

6
crates/apub/src/activities/post/create_or_update.rs
View file

@ -3,7 +3,6 @@ use crate::{
community::announce::AnnouncableActivities,
extract_community,
generate_activity_id,
post::send_websocket_message,
verify_activity,
verify_mod_action,
verify_person_in_community,
@ -27,7 +26,7 @@ use lemmy_apub_lib::{
use lemmy_db_queries::Crud;
use lemmy_db_schema::source::{community::Community, person::Person, post::Post};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_post_ws_message, LemmyContext, UserOperationCrud};
use url::Url;
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
@ -126,7 +125,8 @@ impl ActivityHandler for CreateOrUpdatePost {
CreateOrUpdateType::Create => UserOperationCrud::CreatePost,
CreateOrUpdateType::Update => UserOperationCrud::EditPost,
};
send_websocket_message(post.id, notif_type, context).await
send_post_ws_message(post.id, notif_type, None, None, context).await?;
Ok(())
}
fn common(&self) -> &ActivityCommonFields {

29
crates/apub/src/activities/post/mod.rs
View file

@ -1,30 +1 @@
use lemmy_api_common::{blocking, post::PostResponse};
use lemmy_db_schema::PostId;
use lemmy_db_views::post_view::PostView;
use lemmy_utils::LemmyError;
use lemmy_websocket::{messages::SendPost, LemmyContext};
pub mod create_or_update;
pub(crate) async fn send_websocket_message<
OP: ToString + Send + lemmy_websocket::OperationType + 'static,
>(
post_id: PostId,
op: OP,
context: &LemmyContext,
) -> Result<(), LemmyError> {
let post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, None)
})
.await??;
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op,
post: res,
websocket_id: None,
});
Ok(())
}

12
crates/apub/src/activities/private_message/create_or_update.rs
View file

@ -1,11 +1,5 @@
use crate::{
activities::{
generate_activity_id,
private_message::send_websocket_message,
verify_activity,
verify_person,
CreateOrUpdateType,
},
activities::{generate_activity_id, verify_activity, verify_person, CreateOrUpdateType},
activity_queue::send_activity_new,
extensions::context::lemmy_context,
objects::{private_message::Note, FromApub, ToApub},
@ -16,7 +10,7 @@ use lemmy_apub_lib::{verify_domains_match, ActivityCommonFields, ActivityHandler
use lemmy_db_queries::Crud;
use lemmy_db_schema::source::{person::Person, private_message::PrivateMessage};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
use url::Url;
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
@ -83,7 +77,7 @@ impl ActivityHandler for CreateOrUpdatePrivateMessage {
CreateOrUpdateType::Create => UserOperationCrud::CreatePrivateMessage,
CreateOrUpdateType::Update => UserOperationCrud::EditPrivateMessage,
};
send_websocket_message(private_message.id, notif_type, context).await?;
send_pm_ws_message(private_message.id, notif_type, None, context).await?;
Ok(())
}

12
crates/apub/src/activities/private_message/delete.rs
View file

@ -1,10 +1,5 @@
use crate::{
activities::{
generate_activity_id,
private_message::send_websocket_message,
verify_activity,
verify_person,
},
activities::{generate_activity_id, verify_activity, verify_person},
activity_queue::send_activity_new,
extensions::context::lemmy_context,
ActorType,
@ -15,7 +10,7 @@ use lemmy_apub_lib::{verify_domains_match, ActivityCommonFields, ActivityHandler
use lemmy_db_queries::{source::private_message::PrivateMessage_, ApubObject, Crud};
use lemmy_db_schema::source::{person::Person, private_message::PrivateMessage};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
use url::Url;
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
@ -84,9 +79,10 @@ impl ActivityHandler for DeletePrivateMessage {
})
.await??;
send_websocket_message(
send_pm_ws_message(
deleted_private_message.id,
UserOperationCrud::DeletePrivateMessage,
None,
context,
)
.await?;

38
crates/apub/src/activities/private_message/mod.rs
View file

@ -1,41 +1,3 @@
use lemmy_api_common::{blocking, person::PrivateMessageResponse};
use lemmy_db_schema::PrivateMessageId;
use lemmy_db_views::{local_user_view::LocalUserView, private_message_view::PrivateMessageView};
use lemmy_utils::LemmyError;
use lemmy_websocket::{messages::SendUserRoomMessage, LemmyContext, UserOperationCrud};
pub mod create_or_update;
pub mod delete;
pub mod undo_delete;
async fn send_websocket_message(
private_message_id: PrivateMessageId,
op: UserOperationCrud,
context: &LemmyContext,
) -> Result<(), LemmyError> {
let message = blocking(context.pool(), move |conn| {
PrivateMessageView::read(conn, private_message_id)
})
.await??;
let res = PrivateMessageResponse {
private_message_view: message,
};
// Send notifications to the local recipient, if one exists
let recipient_id = res.private_message_view.recipient.id;
let local_recipient_id = blocking(context.pool(), move |conn| {
LocalUserView::read_person(conn, recipient_id)
})
.await??
.local_user
.id;
context.chat_server().do_send(SendUserRoomMessage {
op,
response: res,
local_recipient_id,
websocket_id: None,
});
Ok(())
}

7
crates/apub/src/activities/private_message/undo_delete.rs
View file

@ -1,7 +1,7 @@
use crate::{
activities::{
generate_activity_id,
private_message::{delete::DeletePrivateMessage, send_websocket_message},
private_message::delete::DeletePrivateMessage,
verify_activity,
verify_person,
},
@ -20,7 +20,7 @@ use lemmy_apub_lib::{
use lemmy_db_queries::{source::private_message::PrivateMessage_, ApubObject, Crud};
use lemmy_db_schema::source::{person::Person, private_message::PrivateMessage};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperationCrud};
use lemmy_websocket::{send::send_pm_ws_message, LemmyContext, UserOperationCrud};
use url::Url;
#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
@ -104,9 +104,10 @@ impl ActivityHandler for UndoDeletePrivateMessage {
})
.await??;
send_websocket_message(
send_pm_ws_message(
deleted_private_message.id,
UserOperationCrud::EditPrivateMessage,
None,
context,
)
.await?;

37
crates/apub/src/activities/voting/mod.rs
View file

@ -1,8 +1,4 @@
use crate::activities::{
comment::send_websocket_message as send_comment_message,
post::send_websocket_message as send_post_message,
voting::vote::VoteType,
};
use crate::activities::voting::vote::VoteType;
use lemmy_api_common::blocking;
use lemmy_db_queries::Likeable;
use lemmy_db_schema::source::{
@ -11,7 +7,11 @@ use lemmy_db_schema::source::{
post::{Post, PostLike, PostLikeForm},
};
use lemmy_utils::LemmyError;
use lemmy_websocket::{LemmyContext, UserOperation};
use lemmy_websocket::{
send::{send_comment_ws_message_simple, send_post_ws_message},
LemmyContext,
UserOperation,
};
pub mod undo_vote;
pub mod vote;
@ -36,13 +36,8 @@ async fn vote_comment(
})
.await??;
send_comment_message(
comment_id,
vec![],
UserOperation::CreateCommentLike,
context,
)
.await
send_comment_ws_message_simple(comment_id, UserOperation::CreateCommentLike, context).await?;
Ok(())
}
async fn vote_post(
@ -64,7 +59,8 @@ async fn vote_post(
})
.await??;
send_post_message(post.id, UserOperation::CreatePostLike, context).await
send_post_ws_message(post.id, UserOperation::CreatePostLike, None, None, context).await?;
Ok(())
}
async fn undo_vote_comment(
@ -79,13 +75,8 @@ async fn undo_vote_comment(
})
.await??;
send_comment_message(
comment.id,
vec![],
UserOperation::CreateCommentLike,
context,
)
.await
send_comment_ws_message_simple(comment_id, UserOperation::CreateCommentLike, context).await?;
Ok(())
}
async fn undo_vote_post(
@ -99,5 +90,7 @@ async fn undo_vote_post(
PostLike::remove(conn, person_id, post_id)
})
.await??;
send_post_message(post.id, UserOperation::CreatePostLike, context).await
send_post_ws_message(post_id, UserOperation::CreatePostLike, None, None, context).await?;
Ok(())
}

2
crates/websocket/Cargo.toml
View file

@ -15,6 +15,8 @@ lemmy_utils = { version = "=0.11.3", path = "../utils" }
lemmy_api_common = { version = "=0.11.3", path = "../api_common" }
lemmy_db_queries = { version = "=0.11.3", path = "../db_queries" }
lemmy_db_schema = { version = "=0.11.3", path = "../db_schema" }
lemmy_db_views = { version = "=0.11.3", path = "../db_views" }
lemmy_db_views_actor = { version = "=0.11.3", path = "../db_views_actor" }
reqwest = { version = "0.11.4", features = ["json"] }
log = "0.4.14"
rand = "0.8.4"

1
crates/websocket/src/lib.rs
View file

@ -13,6 +13,7 @@ pub mod chat_server;
pub mod handlers;
pub mod messages;
pub mod routes;
pub mod send;
pub struct LemmyContext {
pub pool: DbPool,

4
crates/websocket/src/messages.rs
View file

@ -75,7 +75,7 @@ pub struct SendModRoomMessage<Response> {
#[derive(Message)]
#[rtype(result = "()")]
pub struct SendPost<OP: ToString> {
pub(crate) struct SendPost<OP: ToString> {
pub op: OP,
pub post: PostResponse,
pub websocket_id: Option<ConnectionId>,
@ -83,7 +83,7 @@ pub struct SendPost<OP: ToString> {
#[derive(Message)]
#[rtype(result = "()")]
pub struct SendComment<OP: ToString> {
pub(crate) struct SendComment<OP: ToString> {
pub op: OP,
pub comment: CommentResponse,
pub websocket_id: Option<ConnectionId>,

162
crates/websocket/src/send.rs Normal file
View file

@ -0,0 +1,162 @@
use crate::{
messages::{SendComment, SendCommunityRoomMessage, SendPost, SendUserRoomMessage},
LemmyContext,
OperationType,
};
use lemmy_api_common::{
blocking,
comment::CommentResponse,
community::CommunityResponse,
person::PrivateMessageResponse,
post::PostResponse,
};
use lemmy_db_queries::DeleteableOrRemoveable;
use lemmy_db_schema::{CommentId, CommunityId, LocalUserId, PersonId, PostId, PrivateMessageId};
use lemmy_db_views::{
comment_view::CommentView,
local_user_view::LocalUserView,
post_view::PostView,
private_message_view::PrivateMessageView,
};
use lemmy_db_views_actor::community_view::CommunityView;
use lemmy_utils::{ConnectionId, LemmyError};
pub async fn send_post_ws_message<OP: ToString + Send + OperationType + 'static>(
post_id: PostId,
op: OP,
websocket_id: Option<ConnectionId>,
person_id: Option<PersonId>,
context: &LemmyContext,
) -> Result<PostResponse, LemmyError> {
let mut post_view = blocking(context.pool(), move |conn| {
PostView::read(conn, post_id, person_id)
})
.await??;
if post_view.post.deleted || post_view.post.removed {
post_view.post = post_view.post.blank_out_deleted_or_removed_info();
}
let res = PostResponse { post_view };
context.chat_server().do_send(SendPost {
op,
post: res.clone(),
websocket_id,
});
Ok(res)
}
// TODO: in many call sites in apub crate, we are setting an empty vec for recipient_ids,
// we should get the actual recipient actors from somewhere
pub async fn send_comment_ws_message_simple<OP: ToString + Send + OperationType + 'static>(
comment_id: CommentId,
op: OP,
context: &LemmyContext,
) -> Result<CommentResponse, LemmyError> {
send_comment_ws_message(comment_id, op, None, None, None, vec![], context).await
}
pub async fn send_comment_ws_message<OP: ToString + Send + OperationType + 'static>(
comment_id: CommentId,
op: OP,
websocket_id: Option<ConnectionId>,
form_id: Option<String>,
person_id: Option<PersonId>,
recipient_ids: Vec<LocalUserId>,
context: &LemmyContext,
) -> Result<CommentResponse, LemmyError> {
let mut view = blocking(context.pool(), move |conn| {
CommentView::read(conn, comment_id, person_id)
})
.await??;
if view.comment.deleted || view.comment.removed {
view.comment = view.comment.blank_out_deleted_or_removed_info();
}
let res = CommentResponse {
comment_view: view,
recipient_ids,
form_id,
};
context.chat_server().do_send(SendComment {
op,
comment: res.clone(),
websocket_id,
});
Ok(res)
}
pub async fn send_community_ws_message<OP: ToString + Send + OperationType + 'static>(
community_id: CommunityId,
op: OP,
websocket_id: Option<ConnectionId>,
person_id: Option<PersonId>,
context: &LemmyContext,
) -> Result<CommunityResponse, LemmyError> {
let mut community_view = blocking(context.pool(), move |conn| {
CommunityView::read(conn, community_id, person_id)
})
.await??;
// Blank out deleted or removed info
if community_view.community.deleted || community_view.community.removed {
community_view.community = community_view.community.blank_out_deleted_or_removed_info();
}
let res = CommunityResponse { community_view };
// Strip out the person id and subscribed when sending to others
let mut res_mut = res.clone();
res_mut.community_view.subscribed = false;
context.chat_server().do_send(SendCommunityRoomMessage {
op,
response: res_mut,
community_id: res.community_view.community.id,
websocket_id,
});
Ok(res)
}
pub async fn send_pm_ws_message<OP: ToString + Send + OperationType + 'static>(
private_message_id: PrivateMessageId,
op: OP,
websocket_id: Option<ConnectionId>,
context: &LemmyContext,
) -> Result<PrivateMessageResponse, LemmyError> {
let mut view = blocking(context.pool(), move |conn| {
PrivateMessageView::read(conn, private_message_id)
})
.await??;
// Blank out deleted or removed info
if view.private_message.deleted {
view.private_message = view.private_message.blank_out_deleted_or_removed_info();
}
let res = PrivateMessageResponse {
private_message_view: view,
};
// Send notifications to the local recipient, if one exists
if res.private_message_view.recipient.local {
let recipient_id = res.private_message_view.recipient.id;
let local_recipient = blocking(context.pool(), move |conn| {
LocalUserView::read_person(conn, recipient_id)
})
.await??;
context.chat_server().do_send(SendUserRoomMessage {
op,
response: res.clone(),
local_recipient_id: local_recipient.local_user.id,
websocket_id,
});
}
Ok(res)
}