From 8f6e5be1d444f840375fdeb6cbba6a9650950117 Mon Sep 17 00:00:00 2001 From: Paul Date: Wed, 28 Jul 2021 13:40:03 +0100 Subject: [PATCH] Events: Distribute user updates to servers. Fix: Deleting a server now correctly deletes all associated objects. --- set_version.sh | 2 +- src/database/entities/channel.rs | 169 +++++++++++++++------------- src/database/entities/server.rs | 96 +++------------- src/database/entities/user.rs | 8 +- src/database/permissions/user.rs | 2 +- src/notifications/events.rs | 12 ++ src/notifications/payload.rs | 2 +- src/notifications/subscriptions.rs | 3 +- src/notifications/websocket.rs | 4 +- src/routes/sync/get_unreads.rs | 2 +- src/routes/users/change_username.rs | 2 +- src/routes/users/edit_user.rs | 39 ++----- src/version.rs | 2 +- 13 files changed, 142 insertions(+), 201 deletions(-) diff --git a/set_version.sh b/set_version.sh index f27347ed..0c444099 100755 --- a/set_version.sh +++ b/set_version.sh @@ -1,3 +1,3 @@ #!/bin/bash -export version=0.5.1-alpha.16 +export version=0.5.1-alpha.17 echo "pub const VERSION: &str = \"${version}\";" > src/version.rs diff --git a/src/database/entities/channel.rs b/src/database/entities/channel.rs index ead1df5f..aa17af67 100644 --- a/src/database/entities/channel.rs +++ b/src/database/entities/channel.rs @@ -4,6 +4,7 @@ use crate::database::*; use crate::notifications::events::ClientboundNotification; use crate::util::result::{Error, Result}; use futures::StreamExt; +use mongodb::bson::Bson; use mongodb::{ bson::{doc, to_document, Document}, options::FindOptions, @@ -150,11 +151,7 @@ impl Channel { Ok(()) } - pub async fn delete(&self) -> Result<()> { - let id = self.id(); - let messages = get_collection("messages"); - - // Delete any invites. + pub async fn delete_associated_objects(id: Bson) -> Result<()> { get_collection("channel_invites") .delete_many( doc! { @@ -163,87 +160,103 @@ impl Channel { None, ) .await + .map(|_| ()) .map_err(|_| Error::DatabaseError { operation: "delete_many", with: "channel_invites", + }) + } + + pub async fn delete_messages(id: Bson) -> Result<()> { + let messages = get_collection("messages"); + + // Delete any unreads. + get_collection("channel_unreads") + .delete_many( + doc! { + "_id.channel": &id + }, + None, + ) + .await + .map_err(|_| Error::DatabaseError { + operation: "delete_many", + with: "channel_unreads", })?; + // Check if there are any attachments we need to delete. + let message_ids = messages + .find( + doc! { + "channel": &id, + "attachment": { + "$exists": 1 + } + }, + FindOptions::builder().projection(doc! { "_id": 1 }).build(), + ) + .await + .map_err(|_| Error::DatabaseError { + operation: "fetch_many", + with: "messages", + })? + .filter_map(async move |s| s.ok()) + .collect::>() + .await + .into_iter() + .filter_map(|x| x.get_str("_id").ok().map(|x| x.to_string())) + .collect::>(); + + // If we found any, mark them as deleted. + if message_ids.len() > 0 { + get_collection("attachments") + .update_many( + doc! { + "message_id": { + "$in": message_ids + } + }, + doc! { + "$set": { + "deleted": true + } + }, + None, + ) + .await + .map_err(|_| Error::DatabaseError { + operation: "update_many", + with: "attachments", + })?; + } + + // And then delete said messages. + messages + .delete_many( + doc! { + "channel": id + }, + None, + ) + .await + .map(|_| ()) + .map_err(|_| Error::DatabaseError { + operation: "delete_many", + with: "messages", + }) + } + + pub async fn delete(&self) -> Result<()> { + let id = self.id(); + + // Delete any invites. + Channel::delete_associated_objects(Bson::String(id.to_string())).await?; + + // Delete messages. match &self { Channel::VoiceChannel { .. } => {}, _ => { - // Delete any unreads. - get_collection("channel_unreads") - .delete_many( - doc! { - "_id.channel": id - }, - None, - ) - .await - .map_err(|_| Error::DatabaseError { - operation: "delete_many", - with: "channel_unreads", - })?; - - // Check if there are any attachments we need to delete. - let message_ids = messages - .find( - doc! { - "channel": id, - "attachment": { - "$exists": 1 - } - }, - FindOptions::builder().projection(doc! { "_id": 1 }).build(), - ) - .await - .map_err(|_| Error::DatabaseError { - operation: "fetch_many", - with: "messages", - })? - .filter_map(async move |s| s.ok()) - .collect::>() - .await - .into_iter() - .filter_map(|x| x.get_str("_id").ok().map(|x| x.to_string())) - .collect::>(); - - // If we found any, mark them as deleted. - if message_ids.len() > 0 { - get_collection("attachments") - .update_many( - doc! { - "message_id": { - "$in": message_ids - } - }, - doc! { - "$set": { - "deleted": true - } - }, - None, - ) - .await - .map_err(|_| Error::DatabaseError { - operation: "update_many", - with: "attachments", - })?; - } - - // And then delete said messages. - messages - .delete_many( - doc! { - "channel": id - }, - None, - ) - .await - .map_err(|_| Error::DatabaseError { - operation: "delete_many", - with: "messages", - })?; + Channel::delete_messages(Bson::String(id.to_string())).await?; } } diff --git a/src/database/entities/server.rs b/src/database/entities/server.rs index 9bf95c21..5e8ff48d 100644 --- a/src/database/entities/server.rs +++ b/src/database/entities/server.rs @@ -4,11 +4,10 @@ use crate::database::*; use crate::notifications::events::ClientboundNotification; use crate::util::result::{Error, Result}; use futures::StreamExt; -use mongodb::bson::doc; +use mongodb::bson::{Bson, doc}; use mongodb::bson::from_document; use mongodb::bson::to_document; use mongodb::bson::Document; -use mongodb::options::FindOptions; use rocket_contrib::json::JsonValue; use serde::{Deserialize, Serialize}; @@ -133,58 +132,11 @@ impl Server { } pub async fn delete(&self) -> Result<()> { - let messages = get_collection("messages"); - // Check if there are any attachments we need to delete. - // ! FIXME: make this generic and merge with channel delete - // ! e.g. delete_channel(filter: doc!) - let message_ids = messages - .find( - doc! { - "server": &self.id, - "attachment": { - "$exists": 1 - } - }, - FindOptions::builder().projection(doc! { "_id": 1 }).build(), - ) - .await - .map_err(|_| Error::DatabaseError { - operation: "fetch_many", - with: "messages", - })? - .filter_map(async move |s| s.ok()) - .collect::>() - .await - .into_iter() - .filter_map(|x| x.get_str("_id").ok().map(|x| x.to_string())) - .collect::>(); + Channel::delete_messages(Bson::Document(doc! { "$in": &self.channels })).await?; - // If we found any, mark them as deleted. - if message_ids.len() > 0 { - get_collection("attachments") - .update_many( - doc! { - "message_id": { - "$in": message_ids - } - }, - doc! { - "$set": { - "deleted": true - } - }, - None, - ) - .await - .map_err(|_| Error::DatabaseError { - operation: "update_many", - with: "attachments", - })?; - } - - // And then delete said messages. - messages + // Delete all channels. + get_collection("channels") .delete_many( doc! { "server": &self.id @@ -194,27 +146,13 @@ impl Server { .await .map_err(|_| Error::DatabaseError { operation: "delete_many", - with: "messages", + with: "channels", })?; - // Delete all channels, members, bans and invites. - for with in &["channels", "channel_invites"] { - get_collection(with) - .delete_many( - doc! { - "server": &self.id - }, - None, - ) - .await - .map_err(|_| Error::DatabaseError { - operation: "delete_many", - with, - })?; - } - - // ! FIXME: delete any unreads + // Delete any associated objects, e.g. unreads and invites. + Channel::delete_associated_objects(Bson::Document(doc! { "$in": &self.channels })).await?; + // Delete members and bans. for with in &["server_members", "server_bans"] { get_collection(with) .delete_many( @@ -230,6 +168,16 @@ impl Server { })?; } + // Delete server icon / banner. + if let Some(attachment) = &self.icon { + attachment.delete().await?; + } + + if let Some(attachment) = &self.banner { + attachment.delete().await?; + } + + // Delete the server get_collection("servers") .delete_one( doc! { @@ -248,14 +196,6 @@ impl Server { } .publish(self.id.clone()); - if let Some(attachment) = &self.icon { - attachment.delete().await?; - } - - if let Some(attachment) = &self.banner { - attachment.delete().await?; - } - Ok(()) } diff --git a/src/database/entities/user.rs b/src/database/entities/user.rs index 50521624..520307c9 100644 --- a/src/database/entities/user.rs +++ b/src/database/entities/user.rs @@ -229,11 +229,11 @@ impl User { } /// Utility function to get all the server IDs the user is in. - pub async fn fetch_server_ids(&self) -> Result> { + pub async fn fetch_server_ids(id: &str) -> Result> { Ok(get_collection("server_members") .find( doc! { - "_id.user": &self.id + "_id.user": id }, None, ) @@ -256,11 +256,11 @@ impl User { } /// Utility function to fetch unread objects for user. - pub async fn fetch_unreads(&self) -> Result> { + pub async fn fetch_unreads(id: &str) -> Result> { Ok(get_collection("channel_unreads") .find( doc! { - "_id.user": &self.id + "_id.user": id }, None, ) diff --git a/src/database/permissions/user.rs b/src/database/permissions/user.rs index 3e0835dc..a37118e3 100644 --- a/src/database/permissions/user.rs +++ b/src/database/permissions/user.rs @@ -70,7 +70,7 @@ impl<'a> PermissionCalculator<'a> { } let check_server_overlap = async || { - let server_ids = self.perspective.fetch_server_ids().await?; + let server_ids = User::fetch_server_ids(&self.perspective.id).await?; Ok( get_collection("server_members") diff --git a/src/notifications/events.rs b/src/notifications/events.rs index a1f5b798..6d3a407b 100644 --- a/src/notifications/events.rs +++ b/src/notifications/events.rs @@ -172,6 +172,18 @@ impl ClientboundNotification { .ok(); }); } + + pub fn publish_as_user(self, user: String) { + self.clone().publish(user.clone()); + + async_std::task::spawn(async move { + if let Ok(server_ids) = User::fetch_server_ids(&user).await { + for server in server_ids { + self.clone().publish(server.clone()); + } + } + }); + } } pub async fn prehandle_hook(notification: &ClientboundNotification) -> Result<()> { diff --git a/src/notifications/payload.rs b/src/notifications/payload.rs index ac5c4539..d79a16cf 100644 --- a/src/notifications/payload.rs +++ b/src/notifications/payload.rs @@ -19,7 +19,7 @@ pub async fn generate_ready(mut user: User) -> Result { ); } - let server_ids = user.fetch_server_ids().await?; + let server_ids = User::fetch_server_ids(&user.id).await?; let mut cursor = get_collection("servers") .find( doc! { diff --git a/src/notifications/subscriptions.rs b/src/notifications/subscriptions.rs index 4535a893..ba56057b 100644 --- a/src/notifications/subscriptions.rs +++ b/src/notifications/subscriptions.rs @@ -17,8 +17,7 @@ pub async fn generate_subscriptions(user: &User) -> Result<(), String> { } } - let server_ids = user - .fetch_server_ids() + let server_ids = User::fetch_server_ids(&user.id) .await .map_err(|_| "Failed to fetch memberships.".to_string())?; diff --git a/src/notifications/websocket.rs b/src/notifications/websocket.rs index 29bf3b3b..1f077397 100644 --- a/src/notifications/websocket.rs +++ b/src/notifications/websocket.rs @@ -136,7 +136,7 @@ async fn accept(stream: TcpStream) { }), clear: None } - .publish(id); + .publish_as_user(id); } } Err(_) => { @@ -238,7 +238,7 @@ async fn accept(stream: TcpStream) { }), clear: None } - .publish(id); + .publish_as_user(id); } } diff --git a/src/routes/sync/get_unreads.rs b/src/routes/sync/get_unreads.rs index 02e709b8..ffd1af8a 100644 --- a/src/routes/sync/get_unreads.rs +++ b/src/routes/sync/get_unreads.rs @@ -6,5 +6,5 @@ use rocket_contrib::json::JsonValue; #[get("/unreads")] pub async fn req(user: User) -> Result { - Ok(json!(user.fetch_unreads().await?)) + Ok(json!(User::fetch_unreads(&user.id).await?)) } diff --git a/src/routes/users/change_username.rs b/src/routes/users/change_username.rs index 6dabeb8a..b25edc13 100644 --- a/src/routes/users/change_username.rs +++ b/src/routes/users/change_username.rs @@ -62,7 +62,7 @@ pub async fn req( }), clear: None, } - .publish(user.id.clone()); + .publish_as_user(user.id.clone()); Ok(()) } diff --git a/src/routes/users/edit_user.rs b/src/routes/users/edit_user.rs index 9c89b2d6..14a9c841 100644 --- a/src/routes/users/edit_user.rs +++ b/src/routes/users/edit_user.rs @@ -99,7 +99,7 @@ pub async fn req(user: User, data: Json, _ignore_id: String) -> Result<()> } let avatar = std::mem::replace(&mut data.avatar, None); - let attachment = if let Some(attachment_id) = avatar { + if let Some(attachment_id) = avatar { let attachment = File::find_and_use(&attachment_id, "avatars", "user", &user.id).await?; set.insert( "avatar", @@ -110,14 +110,11 @@ pub async fn req(user: User, data: Json, _ignore_id: String) -> Result<()> ); remove_avatar = true; - Some(attachment) - } else { - None - }; + } let mut operations = doc! {}; if set.len() > 0 { - operations.insert("$set", set); + operations.insert("$set", &set); } if unset.len() > 0 { @@ -134,32 +131,12 @@ pub async fn req(user: User, data: Json, _ignore_id: String) -> Result<()> })?; } - if let Some(status) = &data.status { - ClientboundNotification::UserUpdate { - id: user.id.clone(), - data: json!({ "status": status }), - clear: None, - } - .publish(user.id.clone()); - } - - if let Some(avatar) = attachment { - ClientboundNotification::UserUpdate { - id: user.id.clone(), - data: json!({ "avatar": avatar }), - clear: None, - } - .publish(user.id.clone()); - } - - if let Some(clear) = data.remove { - ClientboundNotification::UserUpdate { - id: user.id.clone(), - data: json!({}), - clear: Some(clear), - } - .publish(user.id.clone()); + ClientboundNotification::UserUpdate { + id: user.id.clone(), + data: json!(set), + clear: data.remove, } + .publish_as_user(user.id.clone()); if remove_avatar { if let Some(old_avatar) = user.avatar { diff --git a/src/version.rs b/src/version.rs index 4de1a812..ad2fd7ce 100644 --- a/src/version.rs +++ b/src/version.rs @@ -1 +1 @@ -pub const VERSION: &str = "0.5.1-alpha.16"; +pub const VERSION: &str = "0.5.1-alpha.17";