jmug/stoatchat
1
0
Fork 1
Code Issues Packages Projects Releases Wiki Activity

feat: Add Mass Mentions to the backend (#394)

Browse Source 
* feat: create base of push daemon

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* Add outbound senders

* Make web_push send to rabbit instead (temp stuff)

* feat: stability and friend requests

* make vapid fr stuff not suck

* swap naming of queue

* move pushd into daemons folder

* fix cargo file for move into daemons folder

* feat: probably working fcm push notifs

* comment out fcm webpush stuff since the config keys dont exist

* fix fcm, name queues according to their prod status and configure routing keys

* add pushd to docker

* mix: Remove old code, add stuff to pushd

* fix: lockfile

* feat: update rocket to 5.0.1

* fix: fix queues and ack bugs

* Move rabbit messsage processing into ack queue

* chore: update readme

* chore: optimizations for ack database hits

* pushd flowchart

* misc: update flowchart

* exit dependancy hell

* add rocket_impl flag to authifier

* make the tests file of delta actually compile

* fix: don't silence every push message

* fix: don't silence all messages

* add debug logging for sending data to rabbit from message events

* validate mentions at a server membership level

* put back that import that was actually important

* minor fix to lockfile

* update delta authifier

* feat: proper permissions for push notifications

* add unit test for mention sanitization

* remove local file dependancy on authifier

* update ports to proper defaults

* fixTM the node bindings

* Theoretically configure docker releases for pushd

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* declare exchange in pushd and delta

* fix createbuckets script

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: reference db implementation

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: remove finally redundant code

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: changes

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: other changes

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* fix: make channel name return result

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>

* Add role mention parsing

* feat: update to mongo 3.1, add member generator.

* integrate mass mentions into pushd

* patch redis-rs with updated versions

* feat: chunk role mentions

* move permission bits to 37/38 to avoid livekit conflict

* change role mention format to <%id>

* fix the lockfile from merge

* fix: PR change requests

* feat: add tests

* fix: i am a dumbass

* fix: tests, again

---------

Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com>
This commit is contained in:
Tom
2025-04-05 10:45:04 -07:00
committed by Zomatree
parent a110a087be
commit 2a36f5912d
51 changed files with 3530 additions and 1522 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.config/nextest.toml
View File

@@ -1,2 +1,2 @@
[profile.default] [profile.default]
slow-timeout = { period = "3s", terminate-after = 2 } slow-timeout = { period = "5s", terminate-after = 10 }

2448
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

6
Cargo.toml
View File

@@ -11,11 +11,7 @@ members = [
] ]
[patch.crates-io] [patch.crates-io]
redis22 = { package = "redis", version = "0.22.3", git = "https://github.com/revoltchat/redis-rs", rev = "1a41faf356fd21aebba71cea7eb7eb2653e5f0ef" } redis23 = { package = "redis", version = "0.23.3", git = "https://github.com/revoltchat/redis-rs", rev = "523b2937367e17bd0073722bf6e23d06042cb4e4" }
redis23 = { package = "redis", version = "0.23.1", git = "https://github.com/revoltchat/redis-rs", rev = "f8ca28ab85da59d2ccde526b4d2fb390eff5a5f9" }
# authifier = { package = "authifier", version = "1.0.8", path = "../authifier/crates/authifier" }
# rocket_authifier = { package = "rocket_authifier", version = "1.0.8", path = "../authifier/crates/rocket_authifier" }
# rocket = { git = "https://github.com/rwf2/Rocket/", rev = "4dcd928" }
# I'm 99% sure this is overloading the GitHub worker # I'm 99% sure this is overloading the GitHub worker
# hence builds have been failing since, let's just # hence builds have been failing since, let's just

2
crates/bonfire/Cargo.toml
View File

@@ -36,7 +36,7 @@ async-std = { version = "1.8.0", features = [
] } ] }
# core # core
authifier = { version = "1.0.9" } authifier = { version = "1.0.10" }
revolt-result = { path = "../core/result" } revolt-result = { path = "../core/result" }
revolt-models = { path = "../core/models" } revolt-models = { path = "../core/models" }
revolt-config = { path = "../core/config" } revolt-config = { path = "../core/config" }

19
crates/core/config/Revolt.toml
View File

@@ -76,13 +76,19 @@ max_concurrent_connections = 50
# Usually they have to be, so that messages sent from one or the other get sent to everyone # Usually they have to be, so that messages sent from one or the other get sent to everyone
production = true production = true
# Changes how many users are processed in each chunk when resolving role/everyone mentions.
# Increasing this will resolve mentions faster, but will consume more memory while resolving.
mass_mention_chunk_size = 200
# none of these should need changing # none of these should need changing
exchange = "revolt.notifications" exchange = "revolt.notifications"
message_queue = "notifications.origin.message" message_queue = "notifications.origin.message"
fr_accepted_queue = "notifications.ingest.fr_accepted" # friend request accepted mass_mention_queue = "notifications.origin.mass_mention" # handles messages that contain role or everyone mentions
fr_received_queue = "notifications.ingest.fr_received" # friend request received fr_accepted_queue = "notifications.ingest.fr_accepted" # friend request accepted
generic_queue = "notifications.ingest.generic" # generic messages (title + body) fr_received_queue = "notifications.ingest.fr_received" # friend request received
ack_queue = "notifications.process.ack" # updates badges for apple devices generic_queue = "notifications.ingest.generic" # generic messages (title + body)
ack_queue = "notifications.process.ack" # updates badges for apple devices
[pushd.vapid] [pushd.vapid]
queue = "notifications.outbound.vapid" queue = "notifications.outbound.vapid"
@@ -181,6 +187,11 @@ default_bucket = "revolt-uploads"
[features] [features]
# Feature gate options # Feature gate options
webhooks_enabled = false webhooks_enabled = false
# Enable push notifications for mass pings (everyone, online, roles)
# When false this will still ping in-client but will not send notifications from pushd
mass_mentions_send_notifications = true
# Can role/everyone pings be used at all
mass_mentions_enabled = true
[features.limits] [features.limits]

11
crates/core/config/src/lib.rs
View File

@@ -201,7 +201,11 @@ pub struct Api {
pub struct Pushd { pub struct Pushd {
pub production: bool, pub production: bool,
pub exchange: String, pub exchange: String,
pub mass_mention_chunk_size: usize,
// Queues
pub message_queue: String, pub message_queue: String,
pub mass_mention_queue: String,
pub fr_accepted_queue: String, pub fr_accepted_queue: String,
pub fr_received_queue: String, pub fr_received_queue: String,
pub generic_queue: String, pub generic_queue: String,
@@ -228,6 +232,10 @@ impl Pushd {
self.get_routing_key(self.message_queue.clone()) self.get_routing_key(self.message_queue.clone())
} }
pub fn get_mass_mention_routing_key(&self) -> String {
self.get_routing_key(self.mass_mention_queue.clone())
}
pub fn get_fr_accepted_routing_key(&self) -> String { pub fn get_fr_accepted_routing_key(&self) -> String {
self.get_routing_key(self.fr_accepted_queue.clone()) self.get_routing_key(self.fr_accepted_queue.clone())
} }
@@ -328,6 +336,9 @@ impl Default for FeaturesAdvanced {
pub struct Features { pub struct Features {
pub limits: FeaturesLimitsCollection, pub limits: FeaturesLimitsCollection,
pub webhooks_enabled: bool, pub webhooks_enabled: bool,
pub mass_mentions_send_notifications: bool,
pub mass_mentions_enabled: bool,
#[serde(default)] #[serde(default)]
pub advanced: FeaturesAdvanced, pub advanced: FeaturesAdvanced,
} }

4
crates/core/database/Cargo.toml
View File

@@ -64,7 +64,7 @@ redis-kiss = { version = "0.1.4" }
# Database # Database
bson = { optional = true, version = "2.1.0" } bson = { optional = true, version = "2.1.0" }
mongodb = { optional = true, version = "2.1.0", default-features = false } mongodb = { optional = true, version = "3.1.0" }
# Database Migration # Database Migration
unicode-segmentation = "1.10.1" unicode-segmentation = "1.10.1"
@@ -96,7 +96,7 @@ web-push = "0.10.0"
revolt_a2 = { version = "0.10", default-features = false, features = ["ring"] } revolt_a2 = { version = "0.10", default-features = false, features = ["ring"] }
# Authifier # Authifier
authifier = { version = "1.0.9", features = ["rocket_impl"] } authifier = { version = "1.0.10", features = ["rocket_impl"] }
# RabbitMQ # RabbitMQ
amqprs = { version = "1.7.0" } amqprs = { version = "1.7.0" }

32
crates/core/database/src/amqp/amqp.rs
View File

@@ -170,6 +170,38 @@ impl AMQP {
.await .await
} }
pub async fn mass_mention_message_sent(
&self,
server_id: String,
payload: Vec<PushNotification>,
) -> Result<(), AMQPError> {
let config = revolt_config::config().await;
let payload = MassMessageSentPayload {
notifications: payload,
server_id,
};
let payload = to_string(&payload).unwrap();
let routing_key = config.pushd.get_mass_mention_routing_key();
debug!(
"Sending mass mention payload on channel {}: {}",
routing_key, payload
);
self.channel
.basic_publish(
BasicProperties::default()
.with_content_type("application/json")
.with_persistence(true)
.finish(),
payload.into(),
BasicPublishArguments::new(&config.pushd.exchange, routing_key.as_str()),
)
.await
}
pub async fn ack_message( pub async fn ack_message(
&self, &self,
user_id: String, user_id: String,

18
crates/core/database/src/drivers/mongodb.rs
View File

@@ -32,17 +32,17 @@ impl MongoDb {
} }
/// Get a collection by its name /// Get a collection by its name
pub fn col<T>(&self, collection: &str) -> mongodb::Collection<T> { pub fn col<T: Send + Sync>(&self, collection: &str) -> mongodb::Collection<T> {
self.db().collection(collection) self.db().collection(collection)
} }
/// Insert one document into a collection /// Insert one document into a collection
pub async fn insert_one<T: Serialize>( pub async fn insert_one<T: Serialize + Send + Sync>(
&self, &self,
collection: &'static str, collection: &'static str,
document: T, document: T,
) -> Result<InsertOneResult> { ) -> Result<InsertOneResult> {
self.col::<T>(collection).insert_one(document, None).await self.col::<T>(collection).insert_one(document).await
} }
/// Count documents by projection /// Count documents by projection
@@ -52,7 +52,7 @@ impl MongoDb {
projection: Document, projection: Document,
) -> Result<u64> { ) -> Result<u64> {
self.col::<Document>(collection) self.col::<Document>(collection)
.count_documents(projection, None) .count_documents(projection)
.await .await
} }
@@ -68,7 +68,8 @@ impl MongoDb {
{ {
Ok(self Ok(self
.col::<T>(collection) .col::<T>(collection)
.find(projection, options) .find(projection)
.with_options(options)
.await? .await?
.filter_map(|s| async { .filter_map(|s| async {
if cfg!(debug_assertions) { if cfg!(debug_assertions) {
@@ -102,7 +103,8 @@ impl MongoDb {
O: Into<Option<FindOneOptions>>, O: Into<Option<FindOneOptions>>,
{ {
self.col::<T>(collection) self.col::<T>(collection)
.find_one(projection, options) .find_one(projection)
.with_options(options)
.await .await
} }
@@ -166,7 +168,7 @@ impl MongoDb {
}; };
self.col::<Document>(collection) self.col::<Document>(collection)
.update_one(projection, query, None) .update_one(projection, query)
.await .await
} }
@@ -201,7 +203,7 @@ impl MongoDb {
projection: Document, projection: Document,
) -> Result<DeleteResult> { ) -> Result<DeleteResult> {
self.col::<Document>(collection) self.col::<Document>(collection)
.delete_one(projection, None) .delete_one(projection)
.await .await
} }

6
crates/core/database/src/events/rabbit.rs
View File

@@ -11,6 +11,12 @@ pub struct MessageSentPayload {
pub users: Vec<String>, pub users: Vec<String>,
} }
#[derive(Serialize, Deserialize)]
pub struct MassMessageSentPayload {
pub notifications: Vec<PushNotification>,
pub server_id: String,
}
#[derive(Serialize, Deserialize, Clone)] #[derive(Serialize, Deserialize, Clone)]
pub struct FRAcceptedPayload { pub struct FRAcceptedPayload {
pub accepted_user: User, pub accepted_user: User,

4
crates/core/database/src/models/admin_migrations/ops/mongodb.rs
View File

@@ -10,7 +10,7 @@ impl AbstractMigrations for MongoDb {
#[cfg(test)] #[cfg(test)]
/// Drop the database /// Drop the database
async fn drop_database(&self) { async fn drop_database(&self) {
self.db().drop(None).await.ok(); self.db().drop().await.ok();
} }
/// Migrate the database /// Migrate the database
@@ -18,7 +18,7 @@ impl AbstractMigrations for MongoDb {
info!("Migrating the database."); info!("Migrating the database.");
let list = self let list = self
.list_database_names(None, None) .list_database_names()
.await .await
.expect("Failed to fetch database names."); .expect("Failed to fetch database names.");

342
crates/core/database/src/models/admin_migrations/ops/mongodb/init.rs
View File

@@ -8,278 +8,254 @@ pub async fn create_database(db: &MongoDb) {
info!("Creating database."); info!("Creating database.");
let db = db.db(); let db = db.db();
db.create_collection("accounts", None) db.create_collection("accounts")
.await .await
.expect("Failed to create accounts collection."); .expect("Failed to create accounts collection.");
db.create_collection("users", None) db.create_collection("users")
.await .await
.expect("Failed to create users collection."); .expect("Failed to create users collection.");
db.create_collection("channels", None) db.create_collection("channels")
.await .await
.expect("Failed to create channels collection."); .expect("Failed to create channels collection.");
db.create_collection("messages", None) db.create_collection("messages")
.await .await
.expect("Failed to create messages collection."); .expect("Failed to create messages collection.");
db.create_collection("servers", None) db.create_collection("servers")
.await .await
.expect("Failed to create servers collection."); .expect("Failed to create servers collection.");
db.create_collection("server_members", None) db.create_collection("server_members")
.await .await
.expect("Failed to create server_members collection."); .expect("Failed to create server_members collection.");
db.create_collection("server_bans", None) db.create_collection("server_bans")
.await .await
.expect("Failed to create server_bans collection."); .expect("Failed to create server_bans collection.");
db.create_collection("channel_invites", None) db.create_collection("channel_invites")
.await .await
.expect("Failed to create channel_invites collection."); .expect("Failed to create channel_invites collection.");
db.create_collection("channel_unreads", None) db.create_collection("channel_unreads")
.await .await
.expect("Failed to create channel_unreads collection."); .expect("Failed to create channel_unreads collection.");
db.create_collection("channel_webhooks", None) db.create_collection("channel_webhooks")
.await .await
.expect("Failed to create channel_webhooks collection."); .expect("Failed to create channel_webhooks collection.");
db.create_collection("migrations", None) db.create_collection("migrations")
.await .await
.expect("Failed to create migrations collection."); .expect("Failed to create migrations collection.");
db.create_collection("attachments", None) db.create_collection("attachments")
.await .await
.expect("Failed to create attachments collection."); .expect("Failed to create attachments collection.");
db.create_collection("attachment_hashes", None) db.create_collection("attachment_hashes")
.await .await
.expect("Failed to create attachment_hashes collection."); .expect("Failed to create attachment_hashes collection.");
db.create_collection("user_settings", None) db.create_collection("user_settings")
.await .await
.expect("Failed to create user_settings collection."); .expect("Failed to create user_settings collection.");
db.create_collection("safety_reports", None) db.create_collection("safety_reports")
.await .await
.expect("Failed to create safety_reports collection."); .expect("Failed to create safety_reports collection.");
db.create_collection("safety_snapshots", None) db.create_collection("safety_snapshots")
.await .await
.expect("Failed to create safety_snapshots collection."); .expect("Failed to create safety_snapshots collection.");
db.create_collection("safety_strikes", None) db.create_collection("safety_strikes")
.await .await
.expect("Failed to create safety_strikes collection."); .expect("Failed to create safety_strikes collection.");
db.create_collection("bots", None) db.create_collection("bots")
.await .await
.expect("Failed to create bots collection."); .expect("Failed to create bots collection.");
db.create_collection("ratelimit_events", None) db.create_collection("ratelimit_events")
.await .await
.expect("Failed to create ratelimit_events collection."); .expect("Failed to create ratelimit_events collection.");
db.create_collection( db.create_collection("pubsub")
"pubsub", .with_options(
CreateCollectionOptions::builder() CreateCollectionOptions::builder()
.capped(true) .capped(true)
.size(1_000_000) .size(1_000_000)
.build(), .build(),
) )
.await .await
.expect("Failed to create pubsub collection."); .expect("Failed to create pubsub collection.");
db.run_command( db.run_command(doc! {
doc! { "createIndexes": "users",
"createIndexes": "users", "indexes": [
"indexes": [ {
{ "key": {
"key": { "username": 1_i32
"username": 1_i32
},
"name": "username",
"unique": false,
"collation": {
"locale": "en",
"strength": 2_i32
}
}, },
{ "name": "username",
"key": { "unique": false,
"username": 1_i32, "collation": {
"discriminator": 1_i32 "locale": "en",
}, "strength": 2_i32
"name": "username_discriminator",
"unique": true,
"collation": {
"locale": "en",
"strength": 2_i32
}
} }
] },
}, {
None, "key": {
) "username": 1_i32,
"discriminator": 1_i32
},
"name": "username_discriminator",
"unique": true,
"collation": {
"locale": "en",
"strength": 2_i32
}
}
]
})
.await .await
.expect("Failed to create username index."); .expect("Failed to create username index.");
db.run_command( db.run_command(doc! {
doc! { "createIndexes": "messages",
"createIndexes": "messages", "indexes": [
"indexes": [ {
{ "key": {
"key": { "content": "text"
"content": "text"
},
"name": "content"
}, },
{ "name": "content"
"key": { },
"channel": 1_i32, {
"_id": 1_i32 "key": {
}, "channel": 1_i32,
"name": "channel_id_compound" "_id": 1_i32
}, },
{ "name": "channel_id_compound"
"key": { },
"author": 1_i32 {
}, "key": {
"name": "author" "author": 1_i32
}, },
{ "name": "author"
"key": { },
"channel": 1_i32, {
"pinned": 1_i32 "key": {
}, "channel": 1_i32,
"name": "channel_pinned_compound" "pinned": 1_i32
}, },
] "name": "channel_pinned_compound"
}, },
None, ]
) })
.await .await
.expect("Failed to create message index."); .expect("Failed to create message index.");
db.run_command( db.run_command(doc! {
doc! { "createIndexes": "channel_unreads",
"createIndexes": "channel_unreads", "indexes": [
"indexes": [ {
{ "key": {
"key": { "_id.channel": 1_i32,
"_id.channel": 1_i32, "_id.user": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
}, },
{ "name": "compound_id"
"key": { },
"_id.user": 1_i32, {
}, "key": {
"name": "user_id" "_id.user": 1_i32,
} },
] "name": "user_id"
}, }
None, ]
) })
.await .await
.expect("Failed to create channel_unreads index."); .expect("Failed to create channel_unreads index.");
db.run_command( db.run_command(doc! {
doc! { "createIndexes": "server_members",
"createIndexes": "server_members", "indexes": [
"indexes": [ {
{ "key": {
"key": { "_id.server": 1_i32,
"_id.server": 1_i32, "_id.user": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
}, },
{ "name": "compound_id"
"key": { },
"_id.user": 1_i32, {
}, "key": {
"name": "user_id" "_id.user": 1_i32,
} },
] "name": "user_id"
}, }
None, ]
) })
.await .await
.expect("Failed to create server_members index."); .expect("Failed to create server_members index.");
db.run_command( db.run_command(doc! {
doc! { "createIndexes": "attachments",
"createIndexes": "attachments", "indexes": [
"indexes": [ {
{ "key": {
"key": { "hash": 1_i32
"hash": 1_i32
},
"name": "hash"
}, },
{ "name": "hash"
"key": { },
"used_for.id": 1_i32 {
}, "key": {
"name": "used_for_id" "used_for.id": 1_i32
} },
] "name": "used_for_id"
}, }
None, ]
) })
.await .await
.expect("Failed to create attachments index."); .expect("Failed to create attachments index.");
db.run_command( db.run_command(doc! {
doc! { "createIndexes": "attachment_hashes",
"createIndexes": "attachment_hashes", "indexes": [
"indexes": [ {
{ "key": {
"key": { "processed_hash": 1_i32
"processed_hash": 1_i32 },
}, "name": "processed_hash"
"name": "processed_hash" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create attachment_hashes index."); .expect("Failed to create attachment_hashes index.");
db.collection("migrations") db.collection("migrations")
.insert_one( .insert_one(doc! {
doc! { "_id": 0_i32,
"_id": 0_i32, "revision": LATEST_REVISION
"revision": LATEST_REVISION })
},
None,
)
.await .await
.expect("Failed to save migration info."); .expect("Failed to save migration info.");
db.run_command( db.run_command(doc! {
doc! { "createIndexes": "ratelimit_events",
"createIndexes": "ratelimit_events", "indexes": [
"indexes": [ {
{ "key": {
"key": { "_id": 1_i32,
"_id": 1_i32, "target_id": 1_i32,
"target_id": 1_i32, "event_type": 1_i32,
"event_type": 1_i32, },
}, "name": "compound_key"
"name": "compound_key" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create ratelimit_events index."); .expect("Failed to create ratelimit_events index.");

571
crates/core/database/src/models/admin_migrations/ops/mongodb/scripts.rs
View File

@@ -26,7 +26,7 @@ pub const LATEST_REVISION: i32 = 33;
pub async fn migrate_database(db: &MongoDb) { pub async fn migrate_database(db: &MongoDb) {
let migrations = db.col::<Document>("migrations"); let migrations = db.col::<Document>("migrations");
let data = migrations let data = migrations
.find_one(None, None) .find_one(doc! {})
.await .await
.expect("Failed to fetch migration data."); .expect("Failed to fetch migration data.");
@@ -46,7 +46,6 @@ pub async fn migrate_database(db: &MongoDb) {
"revision": revision "revision": revision
} }
}, },
None,
) )
.await .await
.expect("Failed to commit migration information."); .expect("Failed to commit migration information.");
@@ -74,7 +73,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.update_many( .update_many(
doc! { "attachment": { "$exists": 1_i32 } }, doc! { "attachment": { "$exists": 1_i32 } },
doc! { "$set": { "attachment.tag": "attachments", "attachment.size": 0_i32 } }, doc! { "$set": { "attachment.tag": "attachments", "attachment.size": 0_i32 } },
None,
) )
.await .await
.expect("Failed to update messages."); .expect("Failed to update messages.");
@@ -83,7 +81,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.update_many( .update_many(
doc! {}, doc! {},
doc! { "$set": { "tag": "attachments", "size": 0_i32 } }, doc! { "$set": { "tag": "attachments", "size": 0_i32 } },
None,
) )
.await .await
.expect("Failed to update attachments."); .expect("Failed to update attachments.");
@@ -93,7 +90,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 2 / 2021-05-08]: Add servers collection."); info!("Running migration [revision 2 / 2021-05-08]: Add servers collection.");
db.db() db.db()
.create_collection("servers", None) .create_collection("servers")
.await .await
.expect("Failed to create servers collection."); .expect("Failed to create servers collection.");
} }
@@ -103,12 +100,12 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
let messages = db.col::<Document>("messages"); let messages = db.col::<Document>("messages");
let mut cursor = messages let mut cursor = messages
.find( .find(doc! {
doc! { "attachment": {
"attachment": { "$exists": 1_i32
"$exists": 1_i32 }
} })
}, .with_options(
FindOptions::builder() FindOptions::builder()
.projection(doc! { .projection(doc! {
"_id": 1_i32, "_id": 1_i32,
@@ -128,19 +125,18 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.update_one( .update_one(
doc! { "_id": id }, doc! { "_id": id },
doc! { "$unset": { "attachment": 1_i32 }, "$set": { "attachments": attachments } }, doc! { "$unset": { "attachment": 1_i32 }, "$set": { "attachments": attachments } },
None,
) )
.await .await
.unwrap(); .unwrap();
} }
db.db() db.db()
.create_collection("channel_unreads", None) .create_collection("channel_unreads")
.await .await
.expect("Failed to create channel_unreads collection."); .expect("Failed to create channel_unreads collection.");
db.db() db.db()
.create_collection("user_settings", None) .create_collection("user_settings")
.await .await
.expect("Failed to create user_settings collection."); .expect("Failed to create user_settings collection.");
} }
@@ -149,17 +145,17 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 4 / 2021-06-01]: Add more server collections."); info!("Running migration [revision 4 / 2021-06-01]: Add more server collections.");
db.db() db.db()
.create_collection("server_members", None) .create_collection("server_members")
.await .await
.expect("Failed to create server_members collection."); .expect("Failed to create server_members collection.");
db.db() db.db()
.create_collection("server_bans", None) .create_collection("server_bans")
.await .await
.expect("Failed to create server_bans collection."); .expect("Failed to create server_bans collection.");
db.db() db.db()
.create_collection("channel_invites", None) .create_collection("channel_invites")
.await .await
.expect("Failed to create channel_invites collection."); .expect("Failed to create channel_invites collection.");
} }
@@ -182,7 +178,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
doc! { doc! {
"$set": to_document(&server).unwrap() "$set": to_document(&server).unwrap()
}, },
None,
) )
.await .await
.expect("Failed to migrate servers."); .expect("Failed to migrate servers.");
@@ -192,20 +187,17 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 6 / 2021-07-09]: Add message text index."); info!("Running migration [revision 6 / 2021-07-09]: Add message text index.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "messages",
"createIndexes": "messages", "indexes": [
"indexes": [ {
{ "key": {
"key": { "content": "text"
"content": "text" },
}, "name": "content"
"name": "content" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create message index."); .expect("Failed to create message index.");
} }
@@ -214,7 +206,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 7 / 2021-08-11]: Add message text index."); info!("Running migration [revision 7 / 2021-08-11]: Add message text index.");
db.db() db.db()
.create_collection("bots", None) .create_collection("bots")
.await .await
.expect("Failed to create bots collection."); .expect("Failed to create bots collection.");
} }
@@ -223,22 +215,15 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 8 / 2021-09-10]: Update to Authifier version 1."); info!("Running migration [revision 8 / 2021-09-10]: Update to Authifier version 1.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "dropIndexes": "accounts",
"dropIndexes": "accounts", "index": ["email", "email_normalised"]
"index": ["email", "email_normalised"] })
},
None,
)
.await .await
.expect("Failed to delete legacy account indexes."); .expect("Failed to delete legacy account indexes.");
let col = db.col::<Document>("sessions"); let col = db.col::<Document>("sessions");
let mut cursor = db let mut cursor = db.col::<Document>("accounts").find(doc! {}).await.unwrap();
.col::<Document>("accounts")
.find(doc! {}, None)
.await
.unwrap();
while let Some(doc) = cursor.next().await { while let Some(doc) = cursor.next().await {
if let Ok(account) = doc { if let Ok(account) = doc {
@@ -267,7 +252,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
doc.insert("subscription", sub); doc.insert("subscription", sub);
} }
col.insert_one(doc, None).await.ok(); col.insert_one(doc).await.ok();
} }
} else { } else {
info!("Account doesn't have any sessions!"); info!("Account doesn't have any sessions!");
@@ -288,7 +273,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
} }
} }
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -297,11 +281,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
if revision <= 9 { if revision <= 9 {
info!("Running migration [revision 9 / 2021-09-14]: Switch from last_message to last_message_id."); info!("Running migration [revision 9 / 2021-09-14]: Switch from last_message to last_message_id.");
let mut cursor = db let mut cursor = db.col::<Document>("channels").find(doc! {}).await.unwrap();
.col::<Document>("channels")
.find(doc! {}, None)
.await
.unwrap();
while let Some(doc) = cursor.next().await { while let Some(doc) = cursor.next().await {
if let Ok(channel) = doc { if let Ok(channel) = doc {
@@ -340,7 +320,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"last_message": 1_i32, "last_message": 1_i32,
} }
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -362,7 +341,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"nonce": 1_i32, "nonce": 1_i32,
} }
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -375,7 +353,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"nonce": 1_i32, "nonce": 1_i32,
} }
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -385,70 +362,61 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 11 / 2021-11-14]: Add indexes to database."); info!("Running migration [revision 11 / 2021-11-14]: Add indexes to database.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "messages",
"createIndexes": "messages", "indexes": [
"indexes": [ {
{ "key": {
"key": { "channel": 1_i32
"channel": 1_i32 },
}, "name": "channel"
"name": "channel" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create message index."); .expect("Failed to create message index.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "channel_unreads",
"createIndexes": "channel_unreads", "indexes": [
"indexes": [ {
{ "key": {
"key": { "_id.channel": 1_i32,
"_id.channel": 1_i32, "_id.user": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
}, },
{ "name": "compound_id"
"key": { },
"_id.user": 1_i32, {
}, "key": {
"name": "user_id" "_id.user": 1_i32,
} },
] "name": "user_id"
}, }
None, ]
) })
.await .await
.expect("Failed to create channel_unreads index."); .expect("Failed to create channel_unreads index.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "server_members",
"createIndexes": "server_members", "indexes": [
"indexes": [ {
{ "key": {
"key": { "_id.server": 1_i32,
"_id.server": 1_i32, "_id.user": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
}, },
{ "name": "compound_id"
"key": { },
"_id.user": 1_i32, {
}, "key": {
"name": "user_id" "_id.user": 1_i32,
} },
] "name": "user_id"
}, }
None, ]
) })
.await .await
.expect("Failed to create server_members index."); .expect("Failed to create server_members index.");
} }
@@ -457,21 +425,18 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 12 / 2021-11-21]: Add indexes to database."); info!("Running migration [revision 12 / 2021-11-21]: Add indexes to database.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "messages",
"createIndexes": "messages", "indexes": [
"indexes": [ {
{ "key": {
"key": { "channel": 1_i32,
"channel": 1_i32, "_id": 1_i32
"_id": 1_i32 },
}, "name": "channel_id_compound"
"name": "channel_id_compound" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create message index."); .expect("Failed to create message index.");
} }
@@ -485,7 +450,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
async_std::task::sleep(Duration::from_secs(10)).await; async_std::task::sleep(Duration::from_secs(10)).await;
let servers = db.col::<Document>("servers"); let servers = db.col::<Document>("servers");
let mut cursor = servers.find(doc! {}, None).await.unwrap(); let mut cursor = servers.find(doc! {}).await.unwrap();
while let Some(Ok(mut document)) = cursor.next().await { while let Some(Ok(mut document)) = cursor.next().await {
let id = document.get_str("_id").unwrap().to_string(); let id = document.get_str("_id").unwrap().to_string();
@@ -529,13 +494,13 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
} }
servers servers
.update_one(doc! { "_id": id }, doc! { "$set": update }, None) .update_one(doc! { "_id": id }, doc! { "$set": update })
.await .await
.unwrap(); .unwrap();
} }
let channels = db.col::<Document>("channels"); let channels = db.col::<Document>("channels");
let mut cursor = channels.find(doc! {}, None).await.unwrap(); let mut cursor = channels.find(doc! {}).await.unwrap();
while let Some(Ok(document)) = cursor.next().await { while let Some(Ok(document)) = cursor.next().await {
let id = document.get_str("_id").unwrap().to_string(); let id = document.get_str("_id").unwrap().to_string();
@@ -575,7 +540,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
} }
channels channels
.update_one(doc! { "_id": id }, update, None) .update_one(doc! { "_id": id }, update)
.await .await
.unwrap(); .unwrap();
} }
@@ -596,7 +561,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"content": "system" "content": "system"
} }
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -621,25 +585,22 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.unwrap(); .unwrap();
db.db() db.db()
.create_collection("emojis", None) .create_collection("emojis")
.await .await
.expect("Failed to create emojis collection."); .expect("Failed to create emojis collection.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "emojis",
"createIndexes": "emojis", "indexes": [
"indexes": [ {
{ "key": {
"key": { "parent.id": 1_i32,
"parent.id": 1_i32, },
}, "name": "parent_id"
"name": "parent_id" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create emoji parent index."); .expect("Failed to create emoji parent index.");
} }
@@ -652,10 +613,9 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
doc! {}, doc! {},
doc! { doc! {
"$set": { "$set": {
"joined_at": DateTime::now().to_rfc3339_string() "joined_at": DateTime::now().try_to_rfc3339_string().expect("Failed to convert the date to rfc3339")
} }
}, },
None,
) )
.await .await
.expect("Failed to update server members."); .expect("Failed to update server members.");
@@ -666,13 +626,10 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
if db if db
.db() .db()
.run_command( .run_command(doc! {
doc! { "dropIndexes": "messages",
"dropIndexes": "messages", "index": ["channel"]
"index": ["channel"] })
},
None,
)
.await .await
.is_err() .is_err()
{ {
@@ -680,20 +637,17 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
} }
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "messages",
"createIndexes": "messages", "indexes": [
"indexes": [ {
{ "key": {
"key": { "author": 1_i32,
"author": 1_i32, },
}, "name": "author"
"name": "author" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create messages author index."); .expect("Failed to create messages author index.");
} }
@@ -703,35 +657,26 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"Running migration [revision 19 / 27-02-2023]: Create report / snapshot collections." "Running migration [revision 19 / 27-02-2023]: Create report / snapshot collections."
); );
db.db() db.db().create_collection("safety_reports").await.unwrap();
.create_collection("safety_reports", None)
.await
.unwrap();
db.db() db.db().create_collection("safety_snapshots").await.unwrap();
.create_collection("safety_snapshots", None)
.await
.unwrap();
} }
if revision <= 20 { if revision <= 20 {
info!("Running migration [revision 20 / 28-02-2023]: Add index `snapshot.report_id`."); info!("Running migration [revision 20 / 28-02-2023]: Add index `snapshot.report_id`.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "safety_snapshots",
"createIndexes": "safety_snapshots", "indexes": [
"indexes": [ {
{ "key": {
"key": { "report_id": 1_i32
"report_id": 1_i32 },
}, "name": "report_id"
"name": "report_id" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create safety snapshot index."); .expect("Failed to create safety snapshot index.");
} }
@@ -739,10 +684,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
if revision <= 21 { if revision <= 21 {
info!("Running migration [revision 21 / 31-05-2023]: Add collection `safety_strikes`."); info!("Running migration [revision 21 / 31-05-2023]: Add collection `safety_strikes`.");
db.db() db.db().create_collection("safety_strikes").await.unwrap();
.create_collection("safety_strikes", None)
.await
.unwrap();
} }
if revision <= 22 { if revision <= 22 {
@@ -756,7 +698,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"moderator_id": "01EX2NCWQ0CHS3QJF0FEQS1GR4" "moderator_id": "01EX2NCWQ0CHS3QJF0FEQS1GR4"
} }
}, },
None,
) )
.await .await
.expect("Failed to update server members."); .expect("Failed to update server members.");
@@ -766,13 +707,10 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 23 / 10-06-2023]: Generate discriminators for users."); info!("Running migration [revision 23 / 10-06-2023]: Generate discriminators for users.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "dropIndexes": "users",
"dropIndexes": "users", "index": "username"
"index": "username" })
},
None,
)
.await .await
.expect("Failed to drop existing username index."); .expect("Failed to drop existing username index.");
@@ -787,7 +725,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
let users: Vec<UserInformation> = db let users: Vec<UserInformation> = db
.col::<UserInformation>("users") .col::<UserInformation>("users")
.find(doc! {}, None) .find(doc! {})
.await .await
.unwrap() .unwrap()
.map(|doc| doc.expect("id and username")) .map(|doc| doc.expect("id and username"))
@@ -839,7 +777,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"discriminator": discriminator "discriminator": discriminator
} }
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -887,7 +824,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"display_name": &info.username "display_name": &info.username
} }
}, },
None,
) )
.await .await
.unwrap(); .unwrap();
@@ -898,43 +834,37 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
if revision <= 24 { if revision <= 24 {
info!("Running migration [revision 24 / 09-06-2023]: Add collection `channel_webhooks` if not exists, update users index."); info!("Running migration [revision 24 / 09-06-2023]: Add collection `channel_webhooks` if not exists, update users index.");
db.db() db.db().create_collection("channel_webhooks").await.ok();
.create_collection("channel_webhooks", None)
.await
.ok();
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "users",
"createIndexes": "users", "indexes": [
"indexes": [ {
{ "key": {
"key": { "username": 1_i32
"username": 1_i32
},
"name": "username",
"unique": false,
"collation": {
"locale": "en",
"strength": 2_i32
}
}, },
{ "name": "username",
"key": { "unique": false,
"username": 1_i32, "collation": {
"discriminator": 1_i32 "locale": "en",
}, "strength": 2_i32
"name": "username_discriminator",
"unique": true,
"collation": {
"locale": "en",
"strength": 2_i32
}
} }
] },
}, {
None, "key": {
) "username": 1_i32,
"discriminator": 1_i32
},
"name": "username_discriminator",
"unique": true,
"collation": {
"locale": "en",
"strength": 2_i32
}
}
]
})
.await .await
.expect("Failed to create username index."); .expect("Failed to create username index.");
}; };
@@ -950,7 +880,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"permissions": *DEFAULT_WEBHOOK_PERMISSIONS as i64 "permissions": *DEFAULT_WEBHOOK_PERMISSIONS as i64
} }
}, },
None,
) )
.await .await
.expect("Failed to update webhooks."); .expect("Failed to update webhooks.");
@@ -959,28 +888,22 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
if revision <= 25 { if revision <= 25 {
info!("Running migration [revision 25 / 15-06-2023]: Add collection `ratelimit_events` with index."); info!("Running migration [revision 25 / 15-06-2023]: Add collection `ratelimit_events` with index.");
db.db() db.db().create_collection("ratelimit_events").await.ok();
.create_collection("ratelimit_events", None)
.await
.ok();
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "ratelimit_events",
"createIndexes": "ratelimit_events", "indexes": [
"indexes": [ {
{ "key": {
"key": { "_id": 1_i32,
"_id": 1_i32, "target_id": 1_i32,
"target_id": 1_i32, "event_type": 1_i32,
"event_type": 1_i32, },
}, "name": "compound_key"
"name": "compound_key" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create ratelimit_events index."); .expect("Failed to create ratelimit_events index.");
} }
@@ -1017,12 +940,9 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
let invites = db let invites = db
.db() .db()
.collection::<Outer>("channel_invites") .collection::<Outer>("channel_invites")
.find( .find(doc! {
doc! { "type": { "$exists": false }
"type": { "$exists": false } })
},
None,
)
.await .await
.expect("failed to find invites") .expect("failed to find invites")
.filter_map(|s| async { s.ok() }) .filter_map(|s| async { s.ok() })
@@ -1056,18 +976,15 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
if !invites.is_empty() { if !invites.is_empty() {
db.db() db.db()
.collection("channel_invites") .collection("channel_invites")
.insert_many(invites, None) .insert_many(invites)
.await .await
.expect("failed to insert corrected invite"); .expect("failed to insert corrected invite");
db.db() db.db()
.collection::<Outer>("channel_invites") .collection::<Outer>("channel_invites")
.delete_many( .delete_many(doc! {
doc! { "type": { "$exists": false }
"type": { "$exists": false } })
},
None,
)
.await .await
.expect("failed to find invites"); .expect("failed to find invites");
} }
@@ -1077,21 +994,18 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 27 / 21-07-2024]: create message pinned index."); info!("Running migration [revision 27 / 21-07-2024]: create message pinned index.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "messages",
"createIndexes": "messages", "indexes": [
"indexes": [ {
{ "key": {
"key": { "channel": 1_i32,
"channel": 1_i32, "pinned": 1_i32
"pinned": 1_i32 },
}, "name": "channel_pinned_compound"
"name": "channel_pinned_compound" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create message index."); .expect("Failed to create message index.");
} }
@@ -1099,44 +1013,35 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
if revision <= 28 { if revision <= 28 {
info!("Running migration [revision 28 / 10-09-2024]: Add support for new Autumn."); info!("Running migration [revision 28 / 10-09-2024]: Add support for new Autumn.");
db.db() db.db().create_collection("attachment_hashes").await.ok();
.create_collection("attachment_hashes", None)
.await
.ok();
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "attachments",
"createIndexes": "attachments", "indexes": [
"indexes": [ {
{ "key": {
"key": { "hash": 1_i32
"hash": 1_i32 },
}, "name": "hash"
"name": "hash" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create attachments index."); .expect("Failed to create attachments index.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "attachment_hashes",
"createIndexes": "attachment_hashes", "indexes": [
"indexes": [ {
{ "key": {
"key": { "processed_hash": 1_i32
"processed_hash": 1_i32 },
}, "name": "processed_hash"
"name": "processed_hash" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create attachment_hashes index."); .expect("Failed to create attachment_hashes index.");
} }
@@ -1147,20 +1052,17 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Running migration [revision 30 / 29-09-2024]: Add index for used_for.id to attachments."); info!("Running migration [revision 30 / 29-09-2024]: Add index for used_for.id to attachments.");
db.db() db.db()
.run_command( .run_command(doc! {
doc! { "createIndexes": "attachments",
"createIndexes": "attachments", "indexes": [
"indexes": [ {
{ "key": {
"key": { "used_for.id": 1_i32
"used_for.id": 1_i32 },
}, "name": "used_for_id"
"name": "used_for_id" }
} ]
] })
},
None,
)
.await .await
.expect("Failed to create attachments index."); .expect("Failed to create attachments index.");
} }
@@ -1177,7 +1079,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
let webhooks = db let webhooks = db
.db() .db()
.collection::<WebhookShell>("channel_webhooks") .collection::<WebhookShell>("channel_webhooks")
.find(doc! {}, None) .find(doc! {})
.await .await
.expect("webhooks") .expect("webhooks")
.filter_map(|s| async { s.ok() }) .filter_map(|s| async { s.ok() })
@@ -1209,7 +1111,6 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
"creator_id": creator_id "creator_id": creator_id
} }
}, },
None,
) )
.await .await
.expect("update webhook"); .expect("update webhook");
@@ -1220,7 +1121,7 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
}) => { }) => {
db.db() db.db()
.collection::<WebhookShell>("channel_webhooks") .collection::<WebhookShell>("channel_webhooks")
.delete_one(doc! { "_id": webhook._id }, None) .delete_one(doc! { "_id": webhook._id })
.await .await
.expect("failed to delete invalid webhook"); .expect("failed to delete invalid webhook");
} }

9
crates/core/database/src/models/channel_invites/ops/mongodb.rs
View File

@@ -24,12 +24,9 @@ impl AbstractChannelInvites for MongoDb {
async fn fetch_invites_for_server(&self, server_id: &str) -> Result<Vec<Invite>> { async fn fetch_invites_for_server(&self, server_id: &str) -> Result<Vec<Invite>> {
Ok(self Ok(self
.col::<Invite>(COL) .col::<Invite>(COL)
.find( .find(doc! {
doc! { "server": server_id,
"server": server_id, })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", COL))? .map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async { .filter_map(|s| async {

8
crates/core/database/src/models/channel_unreads/ops.rs
View File

@@ -26,6 +26,14 @@ pub trait AbstractChannelUnreads: Sync + Send {
message_ids: &[String], message_ids: &[String],
) -> Result<()>; ) -> Result<()>;
/// Add a mention.
async fn add_mention_to_many_unreads<'a>(
&self,
channel_id: &str,
user_ids: &[String],
message_ids: &[String],
) -> Result<()>;
/// Fetch all unreads with mentions for a user. /// Fetch all unreads with mentions for a user.
async fn fetch_unread_mentions(&self, user_id: &str) -> Result<Vec<ChannelUnread>>; async fn fetch_unread_mentions(&self, user_id: &str) -> Result<Vec<ChannelUnread>>;

47
crates/core/database/src/models/channel_unreads/ops/mongodb.rs
View File

@@ -37,6 +37,8 @@ impl AbstractChannelUnreads for MongoDb {
"last_id": message_id "last_id": message_id
} }
}, },
)
.with_options(
FindOneAndUpdateOptions::builder() FindOneAndUpdateOptions::builder()
.upsert(true) .upsert(true)
.return_document(ReturnDocument::After) .return_document(ReturnDocument::After)
@@ -51,15 +53,12 @@ impl AbstractChannelUnreads for MongoDb {
let current_time = Ulid::new().to_string(); let current_time = Ulid::new().to_string();
self.col::<Document>(COL) self.col::<Document>(COL)
.delete_many( .delete_many(doc! {
doc! { "_id.channel": {
"_id.channel": { "$in": channel_ids
"$in": channel_ids
},
"_id.user": user_id
}, },
None, "_id.user": user_id
) })
.await .await
.map_err(|_| create_database_error!("delete_many", COL))?; .map_err(|_| create_database_error!("delete_many", COL))?;
@@ -77,7 +76,6 @@ impl AbstractChannelUnreads for MongoDb {
} }
}) })
.collect::<Vec<Document>>(), .collect::<Vec<Document>>(),
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -104,13 +102,42 @@ impl AbstractChannelUnreads for MongoDb {
} }
} }
}, },
UpdateOptions::builder().upsert(true).build(),
) )
.with_options(UpdateOptions::builder().upsert(true).build())
.await .await
.map(|_| ()) .map(|_| ())
.map_err(|_| create_database_error!("update_one", COL)) .map_err(|_| create_database_error!("update_one", COL))
} }
/// Add a mention to multiple users.
async fn add_mention_to_many_unreads<'a>(
&self,
channel_id: &str,
user_ids: &[String],
message_ids: &[String],
) -> Result<()> {
self.col::<Document>(COL)
.update_many(
doc! {
"_id.channel": channel_id,
"_id.user": {
"$in": user_ids
},
},
doc! {
"$push": {
"mentions": {
"$each": message_ids
}
}
},
)
.with_options(UpdateOptions::builder().upsert(true).build())
.await
.map(|_| ())
.map_err(|_| create_database_error!("update_many", COL))
}
/// Fetch all channel unreads for a user. /// Fetch all channel unreads for a user.
async fn fetch_unreads(&self, user_id: &str) -> Result<Vec<ChannelUnread>> { async fn fetch_unreads(&self, user_id: &str) -> Result<Vec<ChannelUnread>> {
query!( query!(

32
crates/core/database/src/models/channel_unreads/ops/reference.rs
View File

@@ -78,6 +78,38 @@ impl AbstractChannelUnreads for ReferenceDb {
Ok(()) Ok(())
} }
/// Add a mention to multiple users.
async fn add_mention_to_many_unreads<'a>(
&self,
channel_id: &str,
user_ids: &[String],
message_ids: &[String],
) -> Result<()> {
let mut unreads = self.channel_unreads.lock().await;
for user_id in user_ids {
let key = ChannelCompositeKey {
channel: channel_id.to_string(),
user: user_id.to_string(),
};
if let Some(unread) = unreads.get_mut(&key) {
unread.mentions.replace(message_ids.to_vec());
} else {
unreads.insert(
key.clone(),
ChannelUnread {
id: key,
last_id: None,
mentions: Some(message_ids.to_vec()),
},
);
}
}
Ok(())
}
async fn fetch_unread_mentions(&self, user_id: &str) -> Result<Vec<ChannelUnread>> { async fn fetch_unread_mentions(&self, user_id: &str) -> Result<Vec<ChannelUnread>> {
let unreads = self.channel_unreads.lock().await; let unreads = self.channel_unreads.lock().await;
Ok(unreads Ok(unreads

9
crates/core/database/src/models/channel_webhooks/ops/mongodb.rs
View File

@@ -24,12 +24,9 @@ impl AbstractWebhooks for MongoDb {
async fn fetch_webhooks_for_channel(&self, channel_id: &str) -> Result<Vec<Webhook>> { async fn fetch_webhooks_for_channel(&self, channel_id: &str) -> Result<Vec<Webhook>> {
Ok(self Ok(self
.col::<Webhook>(COL) .col::<Webhook>(COL)
.find( .find(doc! {
doc! { "channel_id": channel_id,
"channel_id": channel_id, })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", COL))? .map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async { .filter_map(|s| async {

44
crates/core/database/src/models/channels/ops/mongodb.rs
View File

@@ -23,14 +23,11 @@ impl AbstractChannels for MongoDb {
async fn fetch_channels<'a>(&self, ids: &'a [String]) -> Result<Vec<Channel>> { async fn fetch_channels<'a>(&self, ids: &'a [String]) -> Result<Vec<Channel>> {
Ok(self Ok(self
.col::<Channel>(COL) .col::<Channel>(COL)
.find( .find(doc! {
doc! { "_id": {
"_id": { "$in": ids
"$in": ids }
} })
},
None,
)
.await .await
.map_err(|_| create_database_error!("fetch", "channels"))? .map_err(|_| create_database_error!("fetch", "channels"))?
.filter_map(|s| async { .filter_map(|s| async {
@@ -119,7 +116,6 @@ impl AbstractChannels for MongoDb {
"recipients": user "recipients": user
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -141,7 +137,6 @@ impl AbstractChannels for MongoDb {
"role_permissions.".to_owned() + role: permissions "role_permissions.".to_owned() + role: permissions
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -179,7 +174,6 @@ impl AbstractChannels for MongoDb {
"recipients": user "recipients": user
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -253,7 +247,6 @@ impl AbstractChannels for MongoDb {
"_id": server.id "_id": server.id
}, },
update, update,
None,
) )
.await .await
.map_err(|_| create_database_error!("update_one", "servers"))?; .map_err(|_| create_database_error!("update_one", "servers"))?;
@@ -274,23 +267,17 @@ impl MongoDb {
pub async fn delete_associated_channel_objects(&self, id: Bson) -> Result<()> { pub async fn delete_associated_channel_objects(&self, id: Bson) -> Result<()> {
// Delete all invites to these channels. // Delete all invites to these channels.
self.col::<Document>("channel_invites") self.col::<Document>("channel_invites")
.delete_many( .delete_many(doc! {
doc! { "channel": &id
"channel": &id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("delete_many", "channel_invites"))?; .map_err(|_| create_database_error!("delete_many", "channel_invites"))?;
// Delete unread message objects on channels. // Delete unread message objects on channels.
self.col::<Document>("channel_unreads") self.col::<Document>("channel_unreads")
.delete_many( .delete_many(doc! {
doc! { "_id.channel": &id
"_id.channel": &id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("delete_many", "channel_unreads")) .map_err(|_| create_database_error!("delete_many", "channel_unreads"))
.map(|_| ())?; .map(|_| ())?;
@@ -299,12 +286,9 @@ impl MongoDb {
// Delete all webhooks on this channel. // Delete all webhooks on this channel.
self.col::<Document>("webhooks") self.col::<Document>("webhooks")
.delete_many( .delete_many(doc! {
doc! { "channel": &id
"channel": &id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("delete_many", "webhooks")) .map_err(|_| create_database_error!("delete_many", "webhooks"))
.map(|_| ()) .map(|_| ())

1
crates/core/database/src/models/emojis/ops/mongodb.rs
View File

@@ -60,7 +60,6 @@ impl AbstractEmojis for MongoDb {
} }
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())

1
crates/core/database/src/models/file_hashes/ops/mongodb.rs
View File

@@ -42,7 +42,6 @@ impl AbstractAttachmentHashes for MongoDb {
"iv": nonce "iv": nonce
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())

5
crates/core/database/src/models/files/ops/mongodb.rs
View File

@@ -110,7 +110,6 @@ impl AbstractAttachments for MongoDb {
"uploader_id": uploader_id "uploader_id": uploader_id
} }
}, },
None,
) )
.await .await
.map_err(|_| create_database_error!("update_one", COL))?; .map_err(|_| create_database_error!("update_one", COL))?;
@@ -130,7 +129,6 @@ impl AbstractAttachments for MongoDb {
"reported": true "reported": true
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -149,7 +147,6 @@ impl AbstractAttachments for MongoDb {
"deleted": true "deleted": true
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -170,7 +167,6 @@ impl AbstractAttachments for MongoDb {
"deleted": true "deleted": true
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -193,7 +189,6 @@ impl MongoDb {
"deleted": true "deleted": true
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())

212
crates/core/database/src/models/messages/model.rs
View File

@@ -5,9 +5,10 @@ use iso8601_timestamp::Timestamp;
use revolt_config::{config, FeaturesLimits}; use revolt_config::{config, FeaturesLimits};
use revolt_models::v0::{ use revolt_models::v0::{
self, BulkMessageResponse, DataMessageSend, Embed, MessageAuthor, MessageFlags, MessageSort, self, BulkMessageResponse, DataMessageSend, Embed, MessageAuthor, MessageFlags, MessageSort,
MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text, RE_MENTION, MessageWebhook, PushNotification, ReplyIntent, SendableEmbed, Text, RE_BLOCKS, RE_MENTION,
RE_ROLE_MENTION,
}; };
use revolt_permissions::{ChannelPermission, PermissionValue}; use revolt_permissions::{calculate_channel_permissions, ChannelPermission, PermissionValue};
use revolt_result::{ErrorType, Result}; use revolt_result::{ErrorType, Result};
use ulid::Ulid; use ulid::Ulid;
use validator::Validate; use validator::Validate;
@@ -15,7 +16,10 @@ use validator::Validate;
use crate::{ use crate::{
events::client::EventV1, events::client::EventV1,
tasks::{self, ack::AckEvent}, tasks::{self, ack::AckEvent},
util::{bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey}, util::{
bulk_permissions::BulkDatabasePermissionQuery, idempotency::IdempotencyKey,
permissions::DatabasePermissionQuery,
},
Channel, Database, Emoji, File, User, AMQP, Channel, Database, Emoji, File, User, AMQP,
}; };
@@ -53,6 +57,9 @@ auto_derived_partial!(
/// Array of user ids mentioned in this message /// Array of user ids mentioned in this message
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
pub mentions: Option<Vec<String>>, pub mentions: Option<Vec<String>>,
/// Array of role ids mentioned in this message
#[serde(skip_serializing_if = "Option::is_none")]
pub role_mentions: Option<Vec<String>>,
/// Array of message ids this message is replying to /// Array of message ids this message is replying to
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
pub replies: Option<Vec<String>>, pub replies: Option<Vec<String>>,
@@ -71,7 +78,7 @@ auto_derived_partial!(
/// Bitfield of message flags /// Bitfield of message flags
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
pub flags: Option<i32>, pub flags: Option<u32>,
}, },
"PartialMessage" "PartialMessage"
); );
@@ -201,6 +208,30 @@ auto_derived!(
} }
); );
pub struct MessageFlagsValue(pub u32);
impl MessageFlagsValue {
pub fn has(&self, flag: MessageFlags) -> bool {
self.has_value(flag as u32)
}
pub fn has_value(&self, bit: u32) -> bool {
let mask = 1 << bit;
self.0 & mask == mask
}
pub fn set(&mut self, flag: MessageFlags, toggle: bool) -> &mut Self {
self.set_value(flag as u32, toggle)
}
pub fn set_value(&mut self, bit: u32, toggle: bool) -> &mut Self {
if toggle {
self.0 |= 1 << bit;
} else {
self.0 &= !(1 << bit);
}
self
}
}
#[allow(clippy::derivable_impls)] #[allow(clippy::derivable_impls)]
impl Default for Message { impl Default for Message {
fn default() -> Self { fn default() -> Self {
@@ -216,6 +247,7 @@ impl Default for Message {
edited: None, edited: None,
embeds: None, embeds: None,
mentions: None, mentions: None,
role_mentions: None,
replies: None, replies: None,
reactions: Default::default(), reactions: Default::default(),
interactions: Default::default(), interactions: Default::default(),
@@ -264,13 +296,43 @@ impl Message {
return Err(create_error!(EmptyMessage)); return Err(create_error!(EmptyMessage));
} }
// Ensure flags are either not set or have permissible values let allow_mass_mentions = allow_mentions && config.features.mass_mentions_enabled;
if let Some(flags) = &data.flags {
if flags != &0 && flags != &1 { let mut mentions_everyone = false;
let mut mentions_online = false;
let mut suppress_notifications = false;
if let Some(raw_flags) = &data.flags {
if raw_flags > &7 {
// quick path to failure: bigger than all the bits combined
return Err(create_error!(InvalidProperty)); return Err(create_error!(InvalidProperty));
} }
// First step of mass mention resolution
let flags = MessageFlagsValue(*raw_flags);
suppress_notifications = flags.has(MessageFlags::SuppressNotifications);
mentions_everyone = allow_mentions && flags.has(MessageFlags::MentionsEveryone);
mentions_online = allow_mentions && flags.has(MessageFlags::MentionsOnline);
// Not a bot, and attempting to set mention flags
if user.as_ref().is_some_and(|u| u.bot.as_ref().is_none())
&& (mentions_everyone || mentions_online)
{
return Err(create_error!(IsNotBot));
}
if mentions_everyone && mentions_online {
return Err(create_error!(InvalidFlagValue));
}
} }
let server_id = match channel {
Channel::TextChannel { ref server, .. } | Channel::VoiceChannel { ref server, .. } => {
Some(server.clone())
}
_ => None,
};
// Ensure restrict_reactions is not specified without reactions list // Ensure restrict_reactions is not specified without reactions list
if let Some(interactions) = &data.interactions { if let Some(interactions) = &data.interactions {
if interactions.restrict_reactions { if interactions.restrict_reactions {
@@ -304,19 +366,99 @@ impl Message {
.unwrap_or_default(), .unwrap_or_default(),
author: author_id, author: author_id,
webhook: webhook.map(|w| w.into()), webhook: webhook.map(|w| w.into()),
flags: data.flags.map(|v| v as i32), flags: data.flags,
..Default::default() ..Default::default()
}; };
// Parse mentions in message. // Parse mentions in message.
let mut mentions = HashSet::new(); let mut mentions = HashSet::new();
if allow_mentions { let mut role_mentions = HashSet::new();
if let Some(content) = &data.content { let mut role_model_mentions = vec![];
for capture in RE_MENTION.captures_iter(content) {
if let Some(mention) = capture.get(1) { if let Some(raw_content) = &data.content {
mentions.insert(mention.as_str().to_string()); // remove all codeblocks before running the mention regex
let modified_content = RE_BLOCKS.replace_all(raw_content, "");
for capture in RE_MENTION.captures_iter(&modified_content).flatten() {
if let Some(mention) = capture.get(1) {
mentions.insert(mention.as_str().to_string());
}
}
if allow_mass_mentions && server_id.is_some() {
// Second step of mass mention resolution.
for capture in RE_ROLE_MENTION.captures_iter(&modified_content).flatten() {
let raw = capture.get(0).expect("No capture?").as_str();
if raw == "@everyone" {
mentions_everyone = true;
// Can't break early here since we want to capture any role ids
} else if raw == "@online" {
mentions_online = true;
} else if let Some(role_match) = capture.get(1) {
role_mentions.insert(role_match.as_str().to_string());
} }
} }
if !role_mentions.is_empty() {
let server_data = db
.fetch_server(server_id.unwrap().as_str())
.await
.expect("Failed to fetch server");
role_mentions = role_mentions
.iter()
.filter(|role_id| server_data.roles.contains_key(*role_id))
.cloned()
.collect();
role_model_mentions.extend(
role_mentions
.iter()
.map(|role_id| server_data.roles.get(role_id).to_owned()),
);
}
}
}
// Validate the user can perform a mass mention
if !config.features.mass_mentions_enabled
&& (mentions_everyone || mentions_online || !role_mentions.is_empty())
{
mentions_everyone = false;
mentions_online = false;
role_mentions.clear();
} else if mentions_everyone || mentions_online || !role_mentions.is_empty() {
debug!(
"Mentioned everyone: {}, mentioned online: {}, mentioned roles: {:?}",
mentions_everyone, mentions_online, &role_mentions
);
if let Some(user) = match author {
MessageAuthor::User(user) => Some(Ok(user)),
MessageAuthor::System { .. } => Some(Err(())), // DISALLOWED
MessageAuthor::Webhook(..) => None, // Bypass check
} {
if user.is_err() {
return Err(create_error!(InvalidProperty));
}
let owned_user: User = user.unwrap().to_owned().into();
let mut query = DatabasePermissionQuery::new(db, &owned_user).channel(&channel);
let perms = calculate_channel_permissions(&mut query).await;
if (mentions_everyone || mentions_online)
&& !perms.has_channel_permission(ChannelPermission::MentionEveryone)
{
return Err(create_error!(MissingPermission {
permission: ChannelPermission::MentionEveryone.to_string()
}));
}
if !role_mentions.is_empty()
&& !perms.has_channel_permission(ChannelPermission::MentionRoles)
{
return Err(create_error!(MissingPermission {
permission: ChannelPermission::MentionRoles.to_string()
}));
}
} }
} }
@@ -366,6 +508,7 @@ impl Message {
let recipients_hash: HashSet<&String, RandomState> = let recipients_hash: HashSet<&String, RandomState> =
HashSet::from_iter(recipients); HashSet::from_iter(recipients);
mentions.retain(|m| recipients_hash.contains(m)); mentions.retain(|m| recipients_hash.contains(m));
role_mentions.clear();
} }
Channel::TextChannel { ref server, .. } Channel::TextChannel { ref server, .. }
| Channel::VoiceChannel { ref server, .. } => { | Channel::VoiceChannel { ref server, .. } => {
@@ -406,12 +549,27 @@ impl Message {
message.mentions.replace(mentions.into_iter().collect()); message.mentions.replace(mentions.into_iter().collect());
} }
if !role_mentions.is_empty() {
message
.role_mentions
.replace(role_mentions.into_iter().collect());
}
if !replies.is_empty() { if !replies.is_empty() {
message message
.replies .replies
.replace(replies.into_iter().collect::<Vec<String>>()); .replace(replies.into_iter().collect::<Vec<String>>());
} }
// Calculate final message flags
let mut flag_value = MessageFlagsValue(0);
flag_value
.set(MessageFlags::SuppressNotifications, suppress_notifications)
.set(MessageFlags::MentionsEveryone, mentions_everyone)
.set(MessageFlags::MentionsOnline, mentions_online);
message.flags = Some(flag_value.0);
// Add attachments to message. // Add attachments to message.
let mut attachments = vec![]; let mut attachments = vec![];
if data if data
@@ -490,7 +648,12 @@ impl Message {
tasks::ack::queue_message( tasks::ack::queue_message(
self.channel.to_string(), self.channel.to_string(),
AckEvent::ProcessMessage { AckEvent::ProcessMessage {
messages: vec![(None, self.clone(), mentions.clone(), true)], messages: vec![(
None,
self.clone(),
mentions.clone(),
self.has_suppressed_notifications(),
)],
}, },
) )
.await; .await;
@@ -534,7 +697,9 @@ impl Message {
) )
.await?; .await?;
if !self.has_suppressed_notifications() { if !self.has_suppressed_notifications()
&& (self.mentions.is_some() || self.contains_mass_push_mention())
{
// send Push notifications // send Push notifications
tasks::ack::queue_message( tasks::ack::queue_message(
self.channel.to_string(), self.channel.to_string(),
@@ -557,7 +722,7 @@ impl Message {
} }
_ => vec![], _ => vec![],
}, },
self.has_suppressed_notifications(), false, // branch already dictates this
)], )],
}, },
) )
@@ -594,13 +759,24 @@ impl Message {
/// Whether this message has suppressed notifications /// Whether this message has suppressed notifications
pub fn has_suppressed_notifications(&self) -> bool { pub fn has_suppressed_notifications(&self) -> bool {
if let Some(flags) = self.flags { if let Some(flags) = self.flags {
flags & MessageFlags::SuppressNotifications as i32 flags & MessageFlags::SuppressNotifications as u32
== MessageFlags::SuppressNotifications as i32 == MessageFlags::SuppressNotifications as u32
} else { } else {
false false
} }
} }
pub fn contains_mass_push_mention(&self) -> bool {
let ping = if let Some(flags) = self.flags {
let flags = MessageFlagsValue(flags);
flags.has(MessageFlags::MentionsEveryone)
} else {
false
};
ping || self.role_mentions.is_some()
}
/// Update message data /// Update message data
pub async fn update( pub async fn update(
&mut self, &mut self,

22
crates/core/database/src/models/messages/ops/mongodb.rs
View File

@@ -228,7 +228,6 @@ impl AbstractMessages for MongoDb {
"_id": id "_id": id
}, },
query, query,
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -247,7 +246,6 @@ impl AbstractMessages for MongoDb {
format!("reactions.{emoji}"): user format!("reactions.{emoji}"): user
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -266,7 +264,6 @@ impl AbstractMessages for MongoDb {
format!("reactions.{emoji}"): user format!("reactions.{emoji}"): user
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -285,7 +282,6 @@ impl AbstractMessages for MongoDb {
format!("reactions.{emoji}"): 1 format!("reactions.{emoji}"): 1
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -300,15 +296,12 @@ impl AbstractMessages for MongoDb {
/// Delete messages from a channel by their ids and corresponding channel id /// Delete messages from a channel by their ids and corresponding channel id
async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()> { async fn delete_messages(&self, channel: &str, ids: &[String]) -> Result<()> {
self.col::<Document>(COL) self.col::<Document>(COL)
.delete_many( .delete_many(doc! {
doc! { "channel": channel,
"channel": channel, "_id": {
"_id": { "$in": ids
"$in": ids }
} })
},
None,
)
.await .await
.map(|_| ()) .map(|_| ())
.map_err(|_| create_database_error!("delete_many", COL)) .map_err(|_| create_database_error!("delete_many", COL))
@@ -362,7 +355,6 @@ impl MongoDb {
"deleted": true "deleted": true
} }
}, },
None,
) )
.await .await
.map_err(|_| create_database_error!("update_many", "attachments"))?; .map_err(|_| create_database_error!("update_many", "attachments"))?;
@@ -370,7 +362,7 @@ impl MongoDb {
// And then delete said messages. // And then delete said messages.
self.col::<Document>(COL) self.col::<Document>(COL)
.delete_many(projection, None) .delete_many(projection)
.await .await
.map(|_| ()) .map(|_| ())
.map_err(|_| create_database_error!("delete_many", COL)) .map_err(|_| create_database_error!("delete_many", COL))

15
crates/core/database/src/models/ratelimit_events/ops/mongodb.rs
View File

@@ -23,16 +23,13 @@ impl AbstractRatelimitEvents for MongoDb {
count: usize, count: usize,
) -> Result<bool> { ) -> Result<bool> {
self.col::<RatelimitEvent>(COL) self.col::<RatelimitEvent>(COL)
.count_documents( .count_documents(doc! {
doc! { "_id": {
"_id": { "$gte": Ulid::from_datetime(SystemTime::now() - period).to_string()
"$gte": Ulid::from_datetime(SystemTime::now() - period).to_string()
},
"target_id": target_id,
"event_type": event_type.to_string()
}, },
None, "target_id": target_id,
) "event_type": event_type.to_string()
})
.await .await
.map(|c| c as usize >= count) .map(|c| c as usize >= count)
.map_err(|_| create_database_error!("count_documents", COL)) .map_err(|_| create_database_error!("count_documents", COL))

79
crates/core/database/src/models/server_members/ops.rs
View File

@@ -1,3 +1,4 @@
use ::mongodb::SessionCursor;
use revolt_result::Result; use revolt_result::Result;
use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember}; use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
@@ -5,6 +6,66 @@ use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
mod mongodb; mod mongodb;
mod reference; mod reference;
#[derive(Debug)]
pub enum ChunkedServerMembersGenerator {
#[cfg(feature = "mongodb")]
MongoDb {
session: ::mongodb::ClientSession,
cursor: Option<SessionCursor<Member>>,
},
Reference {
offset: i32,
data: Option<Vec<Member>>,
},
}
impl ChunkedServerMembersGenerator {
#[cfg(feature = "mongodb")]
pub fn new_mongo(session: ::mongodb::ClientSession, cursor: SessionCursor<Member>) -> Self {
ChunkedServerMembersGenerator::MongoDb {
session,
cursor: Some(cursor),
}
}
pub fn new_reference(data: Vec<Member>) -> Self {
ChunkedServerMembersGenerator::Reference {
offset: 0,
data: Some(data),
}
}
pub async fn next(&mut self) -> Option<Member> {
match self {
#[cfg(feature = "mongodb")]
ChunkedServerMembersGenerator::MongoDb { session, cursor } => {
if let Some(cursor) = cursor {
let value = cursor.next(session).await;
value.map(|val| val.expect("Failed to fetch the next member"))
} else {
warn!("Attempted to access a (MongoDb) server member generator without first setting a cursor");
None
}
}
ChunkedServerMembersGenerator::Reference { offset, data } => {
if let Some(data) = data {
if data.len() as i32 >= *offset {
None
} else {
let resp = &data[*offset as usize];
*offset += 1;
Some(resp.clone())
}
} else {
warn!("Attempted to access a (Reference) server member generator without first providing data");
None
}
}
}
}
}
#[async_trait] #[async_trait]
pub trait AbstractServerMembers: Sync + Send { pub trait AbstractServerMembers: Sync + Send {
/// Insert a new server member into the database /// Insert a new server member into the database
@@ -16,6 +77,24 @@ pub trait AbstractServerMembers: Sync + Send {
/// Fetch all members in a server /// Fetch all members in a server
async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>>; async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>>;
/// Fetch all members in a server as an iterator
async fn fetch_all_members_chunked(
&self,
server_id: &str,
) -> Result<ChunkedServerMembersGenerator>;
async fn fetch_all_members_with_roles(
&self,
server_id: &str,
roles: &[String],
) -> Result<Vec<Member>>;
async fn fetch_all_members_with_roles_chunked(
&self,
server_id: &str,
roles: &[String],
) -> Result<ChunkedServerMembersGenerator>;
/// Fetch all memberships for a user /// Fetch all memberships for a user
async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>>; async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>>;

142
crates/core/database/src/models/server_members/ops/mongodb.rs
View File

@@ -1,10 +1,11 @@
use futures::StreamExt; use futures::StreamExt;
use mongodb::options::ReadConcern;
use revolt_result::Result; use revolt_result::Result;
use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember}; use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
use crate::{IntoDocumentPath, MongoDb}; use crate::{IntoDocumentPath, MongoDb};
use super::AbstractServerMembers; use super::{AbstractServerMembers, ChunkedServerMembersGenerator};
static COL: &str = "server_members"; static COL: &str = "server_members";
@@ -33,12 +34,9 @@ impl AbstractServerMembers for MongoDb {
async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> { async fn fetch_all_members<'a>(&self, server_id: &str) -> Result<Vec<Member>> {
Ok(self Ok(self
.col::<Member>(COL) .col::<Member>(COL)
.find( .find(doc! {
doc! { "_id.server": server_id
"_id.server": server_id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", COL))? .map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async { .filter_map(|s| async {
@@ -52,16 +50,101 @@ impl AbstractServerMembers for MongoDb {
.await) .await)
} }
/// Fetch all members in a server as a generator.
/// Uses config key pushd.mass_mention_chunk_size as the batch size.
async fn fetch_all_members_chunked(
&self,
server_id: &str,
) -> Result<ChunkedServerMembersGenerator> {
let config = revolt_config::config().await;
let mut session = self
.start_session()
.await
.map_err(|_| create_database_error!("start_session", COL))?;
session
.start_transaction()
.read_concern(ReadConcern::snapshot())
.await
.map_err(|_| create_database_error!("start_transaction", COL))?;
let cursor = self
.col::<Member>(COL)
.find(doc! {
"_id.server": server_id
})
.session(&mut session)
.batch_size(config.pushd.mass_mention_chunk_size as u32)
.await
.map_err(|_| create_database_error!("find", COL))?;
Ok(ChunkedServerMembersGenerator::new_mongo(session, cursor))
}
async fn fetch_all_members_with_roles(
&self,
server_id: &str,
roles: &[String],
) -> Result<Vec<Member>> {
Ok(self
.col::<Member>(COL)
.find(doc! {
"_id.server": server_id,
"roles": {"$in": roles}
})
.await
.map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async {
if cfg!(debug_assertions) {
Some(s.unwrap())
} else {
s.ok()
}
})
.collect()
.await)
}
async fn fetch_all_members_with_roles_chunked(
&self,
server_id: &str,
roles: &[String],
) -> Result<ChunkedServerMembersGenerator> {
let config = revolt_config::config().await;
let mut session = self
.start_session()
.await
.map_err(|_| create_database_error!("start_session", COL))?;
session
.start_transaction()
.read_concern(ReadConcern::snapshot())
.await
.map_err(|_| create_database_error!("start_transaction", COL))?;
let cursor = self
.col::<Member>(COL)
.find(doc! {
"_id.server": server_id,
"roles": {"$in": roles}
})
.session(&mut session)
.batch_size(config.pushd.mass_mention_chunk_size as u32)
.await
.map_err(|_| create_database_error!("find", COL))?;
return Ok(ChunkedServerMembersGenerator::new_mongo(session, cursor));
}
/// Fetch all memberships for a user /// Fetch all memberships for a user
async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> { async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
Ok(self Ok(self
.col::<Member>(COL) .col::<Member>(COL)
.find( .find(doc! {
doc! { "_id.user": user_id
"_id.user": user_id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", COL))? .map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async { .filter_map(|s| async {
@@ -79,15 +162,12 @@ impl AbstractServerMembers for MongoDb {
async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> { async fn fetch_members<'a>(&self, server_id: &str, ids: &'a [String]) -> Result<Vec<Member>> {
Ok(self Ok(self
.col::<Member>(COL) .col::<Member>(COL)
.find( .find(doc! {
doc! { "_id.server": server_id,
"_id.server": server_id, "_id.user": {
"_id.user": { "$in": ids
"$in": ids }
} })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", COL))? .map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async { .filter_map(|s| async {
@@ -104,12 +184,9 @@ impl AbstractServerMembers for MongoDb {
/// Fetch member count of a server /// Fetch member count of a server
async fn fetch_member_count(&self, server_id: &str) -> Result<usize> { async fn fetch_member_count(&self, server_id: &str) -> Result<usize> {
self.col::<Member>(COL) self.col::<Member>(COL)
.count_documents( .count_documents(doc! {
doc! { "_id.server": server_id
"_id.server": server_id })
},
None,
)
.await .await
.map(|c| c as usize) .map(|c| c as usize)
.map_err(|_| create_database_error!("count_documents", COL)) .map_err(|_| create_database_error!("count_documents", COL))
@@ -118,12 +195,9 @@ impl AbstractServerMembers for MongoDb {
/// Fetch server count of a user /// Fetch server count of a user
async fn fetch_server_count(&self, user_id: &str) -> Result<usize> { async fn fetch_server_count(&self, user_id: &str) -> Result<usize> {
self.col::<Member>(COL) self.col::<Member>(COL)
.count_documents( .count_documents(doc! {
doc! { "_id.user": user_id
"_id.user": user_id })
},
None,
)
.await .await
.map(|c| c as usize) .map(|c| c as usize)
.map_err(|_| create_database_error!("count_documents", COL)) .map_err(|_| create_database_error!("count_documents", COL))

66
crates/core/database/src/models/server_members/ops/reference.rs
View File

@@ -3,7 +3,7 @@ use revolt_result::Result;
use crate::ReferenceDb; use crate::ReferenceDb;
use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember}; use crate::{FieldsMember, Member, MemberCompositeKey, PartialMember};
use super::AbstractServerMembers; use super::{AbstractServerMembers, ChunkedServerMembersGenerator};
#[async_trait] #[async_trait]
impl AbstractServerMembers for ReferenceDb { impl AbstractServerMembers for ReferenceDb {
@@ -40,6 +40,70 @@ impl AbstractServerMembers for ReferenceDb {
.collect()) .collect())
} }
/// Fetch all members in a server as an iterator
async fn fetch_all_members_chunked(
&self,
server_id: &str,
) -> Result<ChunkedServerMembersGenerator> {
let server_members = self.server_members.lock().await;
let members = server_members
.clone()
.into_values()
.filter(move |member| member.id.server == server_id)
.collect();
// this is inefficient as shit but its the reference db so its fine
Ok(ChunkedServerMembersGenerator::new_reference(members))
}
/// Fetch all members that have any of the roles given
async fn fetch_all_members_with_roles(
&self,
server_id: &str,
roles: &[String],
) -> Result<Vec<Member>> {
let server_members = self.server_members.lock().await;
Ok(server_members
.clone()
.into_values()
.filter(|member| {
member.id.server == server_id
&& !member
.roles
.iter()
.filter(|p| roles.contains(*p))
.collect::<Vec<&String>>()
.is_empty()
})
.collect())
}
async fn fetch_all_members_with_roles_chunked(
&self,
server_id: &str,
roles: &[String],
) -> Result<ChunkedServerMembersGenerator> {
let server_members = self.server_members.lock().await;
let resp = server_members
.clone()
.into_values()
.filter(|member| {
member.id.server == server_id
&& !member
.roles
.iter()
.filter(|p| roles.contains(*p))
.collect::<Vec<&String>>()
.is_empty()
})
.collect();
return Ok(ChunkedServerMembersGenerator::new_reference(resp));
}
/// Fetch all memberships for a user /// Fetch all memberships for a user
async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> { async fn fetch_all_memberships<'a>(&self, user_id: &str) -> Result<Vec<Member>> {
let server_members = self.server_members.lock().await; let server_members = self.server_members.lock().await;

45
crates/core/database/src/models/servers/ops/mongodb.rs
View File

@@ -25,14 +25,11 @@ impl AbstractServers for MongoDb {
async fn fetch_servers<'a>(&self, ids: &'a [String]) -> Result<Vec<Server>> { async fn fetch_servers<'a>(&self, ids: &'a [String]) -> Result<Vec<Server>> {
Ok(self Ok(self
.col::<Server>(COL) .col::<Server>(COL)
.find( .find(doc! {
doc! { "_id": {
"_id": { "$in": ids
"$in": ids }
} })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", "servers"))? .map_err(|_| create_database_error!("find", "servers"))?
.filter_map(|s| async { .filter_map(|s| async {
@@ -84,7 +81,6 @@ impl AbstractServers for MongoDb {
.map_err(|_| create_database_error!("to_document", "role"))? .map_err(|_| create_database_error!("to_document", "role"))?
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -125,7 +121,6 @@ impl AbstractServers for MongoDb {
"roles": &role_id "roles": &role_id
} }
}, },
None,
) )
.await .await
.map_err(|_| create_database_error!("update_many", "server_members"))?; .map_err(|_| create_database_error!("update_many", "server_members"))?;
@@ -140,7 +135,6 @@ impl AbstractServers for MongoDb {
"role_permissions.".to_owned() + role_id: 1_i32 "role_permissions.".to_owned() + role_id: 1_i32
} }
}, },
None,
) )
.await .await
.map_err(|_| create_database_error!("update_one", "channels"))?; .map_err(|_| create_database_error!("update_one", "channels"))?;
@@ -155,7 +149,6 @@ impl AbstractServers for MongoDb {
"roles.".to_owned() + role_id: 1_i32 "roles.".to_owned() + role_id: 1_i32
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -188,12 +181,9 @@ impl MongoDb {
// Find all channels // Find all channels
let channels: Vec<String> = self let channels: Vec<String> = self
.col::<Document>("channels") .col::<Document>("channels")
.find( .find(doc! {
doc! { "server": server_id
"server": server_id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", "channels"))? .map_err(|_| create_database_error!("find", "channels"))?
.filter_map(|s| async { .filter_map(|s| async {
@@ -225,19 +215,15 @@ impl MongoDb {
} }
} }
}, },
None,
) )
.await .await
.map_err(|_| create_database_error!("update_many", "emojis"))?; .map_err(|_| create_database_error!("update_many", "emojis"))?;
// Delete all channels. // Delete all channels.
self.col::<Document>("channels") self.col::<Document>("channels")
.delete_many( .delete_many(doc! {
doc! { "server": &server_id
"server": &server_id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("delete_many", "channels"))?; .map_err(|_| create_database_error!("delete_many", "channels"))?;
@@ -248,12 +234,9 @@ impl MongoDb {
// Delete members and bans. // Delete members and bans.
for with in &["server_members", "server_bans"] { for with in &["server_members", "server_bans"] {
self.col::<Document>(with) self.col::<Document>(with)
.delete_many( .delete_many(doc! {
doc! { "_id.server": &server_id
"_id.server": &server_id })
},
None,
)
.await .await
.map_err(|_| create_database_error!("delete_many", with))?; .map_err(|_| create_database_error!("delete_many", with))?;
} }

2
crates/core/database/src/models/user_settings/ops/mongodb.rs
View File

@@ -53,8 +53,8 @@ impl AbstractUserSettings for MongoDb {
doc! { doc! {
"$set": set "$set": set
}, },
UpdateOptions::builder().upsert(true).build(),
) )
.with_options(UpdateOptions::builder().upsert(true).build())
.await .await
.map(|_| ()) .map(|_| ())
.map_err(|_| create_database_error!("update_one", "user_settings")) .map_err(|_| create_database_error!("update_one", "user_settings"))

142
crates/core/database/src/models/users/ops/mongodb.rs
View File

@@ -49,12 +49,9 @@ impl AbstractUsers for MongoDb {
/// Fetch a session from the database by token /// Fetch a session from the database by token
async fn fetch_session_by_token(&self, token: &str) -> Result<Session> { async fn fetch_session_by_token(&self, token: &str) -> Result<Session> {
self.col::<Session>("sessions") self.col::<Session>("sessions")
.find_one( .find_one(doc! {
doc! { "token": token
"token": token })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find_one", "sessions"))? .map_err(|_| create_database_error!("find_one", "sessions"))?
.ok_or_else(|| create_error!(InvalidSession)) .ok_or_else(|| create_error!(InvalidSession))
@@ -64,14 +61,11 @@ impl AbstractUsers for MongoDb {
async fn fetch_users<'a>(&self, ids: &'a [String]) -> Result<Vec<User>> { async fn fetch_users<'a>(&self, ids: &'a [String]) -> Result<Vec<User>> {
Ok(self Ok(self
.col::<User>(COL) .col::<User>(COL)
.find( .find(doc! {
doc! { "_id": {
"_id": { "$in": ids
"$in": ids }
} })
},
None,
)
.await .await
.map_err(|_| create_database_error!("find", COL))? .map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async { .filter_map(|s| async {
@@ -94,10 +88,10 @@ impl AbstractUsers for MongoDb {
Ok(self Ok(self
.col::<UserDocument>(COL) .col::<UserDocument>(COL)
.find( .find(doc! {
doc! { "username": username
"username": username })
}, .with_options(
FindOptions::builder() FindOptions::builder()
.collation( .collation(
Collation::builder() Collation::builder()
@@ -122,15 +116,13 @@ impl AbstractUsers for MongoDb {
async fn fetch_mutual_user_ids(&self, user_a: &str, user_b: &str) -> Result<Vec<String>> { async fn fetch_mutual_user_ids(&self, user_a: &str, user_b: &str) -> Result<Vec<String>> {
Ok(self Ok(self
.col::<DocumentId>(COL) .col::<DocumentId>(COL)
.find( .find(doc! {
doc! { "$and": [
"$and": [ { "relations": { "$elemMatch": { "_id": &user_a, "status": "Friend" } } },
{ "relations": { "$elemMatch": { "_id": &user_a, "status": "Friend" } } }, { "relations": { "$elemMatch": { "_id": &user_b, "status": "Friend" } } }
{ "relations": { "$elemMatch": { "_id": &user_b, "status": "Friend" } } } ]
] })
}, .with_options(FindOptions::builder().projection(doc! { "_id": 1 }).build())
FindOptions::builder().projection(doc! { "_id": 1 }).build(),
)
.await .await
.map_err(|_| create_database_error!("find", COL))? .map_err(|_| create_database_error!("find", COL))?
.filter_map(|s| async { s.ok() }) .filter_map(|s| async { s.ok() })
@@ -143,17 +135,15 @@ impl AbstractUsers for MongoDb {
async fn fetch_mutual_channel_ids(&self, user_a: &str, user_b: &str) -> Result<Vec<String>> { async fn fetch_mutual_channel_ids(&self, user_a: &str, user_b: &str) -> Result<Vec<String>> {
Ok(self Ok(self
.col::<DocumentId>("channels") .col::<DocumentId>("channels")
.find( .find(doc! {
doc! { "channel_type": {
"channel_type": { "$in": ["Group", "DirectMessage"]
"$in": ["Group", "DirectMessage"]
},
"recipients": {
"$all": [ user_a, user_b ]
}
}, },
FindOptions::builder().projection(doc! { "_id": 1 }).build(), "recipients": {
) "$all": [ user_a, user_b ]
}
})
.with_options(FindOptions::builder().projection(doc! { "_id": 1 }).build())
.await .await
.map_err(|_| create_database_error!("find", "channels"))? .map_err(|_| create_database_error!("find", "channels"))?
.filter_map(|s| async { s.ok() }) .filter_map(|s| async { s.ok() })
@@ -166,49 +156,46 @@ impl AbstractUsers for MongoDb {
async fn fetch_mutual_server_ids(&self, user_a: &str, user_b: &str) -> Result<Vec<String>> { async fn fetch_mutual_server_ids(&self, user_a: &str, user_b: &str) -> Result<Vec<String>> {
Ok(self Ok(self
.col::<DocumentId>("server_members") .col::<DocumentId>("server_members")
.aggregate( .aggregate(vec![
vec![ doc! {
doc! { "$match": {
"$match": { "_id.user": user_a
"_id.user": user_a }
} },
}, doc! {
doc! { "$lookup": {
"$lookup": { "from": "server_members",
"from": "server_members", "as": "members",
"as": "members", "let": {
"let": { "server": "$_id.server"
"server": "$_id.server" },
}, "pipeline": [
"pipeline": [ {
{ "$match": {
"$match": { "$expr": {
"$expr": { "$and": [
"$and": [ { "$eq": [ "$_id.user", user_b ] },
{ "$eq": [ "$_id.user", user_b ] }, { "$eq": [ "$_id.server", "$$server" ] }
{ "$eq": [ "$_id.server", "$$server" ] } ]
]
}
} }
} }
]
}
},
doc! {
"$match": {
"members": {
"$size": 1_i32
} }
]
}
},
doc! {
"$match": {
"members": {
"$size": 1_i32
} }
}, }
doc! { },
"$project": { doc! {
"_id": "$_id.server" "$project": {
} "_id": "$_id.server"
}, }
], },
None, ])
)
.await .await
.map_err(|_| create_database_error!("aggregate", "server_members"))? .map_err(|_| create_database_error!("aggregate", "server_members"))?
.filter_map(|s| async { s.ok() }) .filter_map(|s| async { s.ok() })
@@ -284,7 +271,6 @@ impl AbstractUsers for MongoDb {
} }
} }
}], }],
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -305,7 +291,6 @@ impl AbstractUsers for MongoDb {
} }
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())
@@ -329,7 +314,6 @@ impl AbstractUsers for MongoDb {
"subscription": 1 "subscription": 1
} }
}, },
None,
) )
.await .await
.map(|_| ()) .map(|_| ())

64
crates/core/database/src/tasks/ack.rs
View File

@@ -14,6 +14,7 @@ use validator::HasLen;
use revolt_result::Result; use revolt_result::Result;
use super::DelayedTask; use super::DelayedTask;
use crate::Channel::{TextChannel, VoiceChannel};
/// Enumeration of possible events /// Enumeration of possible events
#[derive(Debug, Eq, PartialEq)] #[derive(Debug, Eq, PartialEq)]
@@ -120,17 +121,22 @@ pub async fn handle_ack_event(
); );
// find all the users we'll be notifying // find all the users we'll be notifying
messages messages.iter().for_each(|(_, _, recipents, _)| {
.iter() users.extend(recipents.iter());
.for_each(|(_, _, recipents, _)| users.extend(recipents.iter())); });
debug!("Found {} users to notify.", users.len()); debug!("Found {} users to notify.", users.len());
for user in users { for user in users {
let message_ids: Vec<String> = messages let message_ids: Vec<String> = messages
.iter() .iter()
.filter(|(_, _, recipients, _)| recipients.contains(user)) .filter_map(|(_, message, recipients, _)| {
.map(|(_, message, _, _)| message.id.clone()) if recipients.contains(user) {
Some(message.id.clone())
} else {
None
}
})
.collect(); .collect();
if !message_ids.is_empty() { if !message_ids.is_empty() {
@@ -140,13 +146,18 @@ pub async fn handle_ack_event(
debug!("Added {} mentions for user {}", message_ids.len(), &user); debug!("Added {} mentions for user {}", message_ids.len(), &user);
} }
for (push, _, recipients, silenced) in messages { let mut mass_mentions = vec![];
if *silenced || recipients.is_empty() || push.is_none() {
for (push, message, recipients, silenced) in messages {
if *silenced
|| push.is_none()
|| (recipients.is_empty() && !message.contains_mass_push_mention())
{
debug!( debug!(
"Rejecting push: silenced: {}, recipient count: {}, push exists: {:?}", "Rejecting push: silenced: {}, recipient count: {}, push exists: {:?}",
*silenced, *silenced,
recipients.length(), recipients.length(),
push push.is_some()
); );
continue; continue;
} }
@@ -162,6 +173,35 @@ pub async fn handle_ack_event(
{ {
revolt_config::capture_error(&err); revolt_config::capture_error(&err);
} }
if message.contains_mass_push_mention() {
mass_mentions.push(push.clone().unwrap());
}
}
if !mass_mentions.is_empty() {
debug!(
"Sending mass mention push event to AMQP; channel {}",
&mass_mentions[0].message.channel
);
let channel = db
.fetch_channel(&mass_mentions[0].message.channel)
.await
.expect("Failed to fetch channel from db");
match channel {
TextChannel { server, .. } | VoiceChannel { server, .. } => {
if let Err(err) =
amqp.mass_mention_message_sent(server, mass_mentions).await
{
revolt_config::capture_error(&err);
}
}
_ => {
panic!("Unknown channel type when sending mass mention event");
}
}
} }
} }
}; };
@@ -192,7 +232,7 @@ pub async fn worker(db: Database, amqp: AMQP) {
revolt_config::capture_error(&err); revolt_config::capture_error(&err);
error!("{err:?} for {event:?}. ({user:?}, {channel})"); error!("{err:?} for {event:?}. ({user:?}, {channel})");
} else { } else {
info!("User {user:?} ack in {channel} with {event:?}"); debug!("User {user:?} ack in {channel} with {event:?}");
} }
} }
} }
@@ -224,6 +264,12 @@ pub async fn worker(db: Database, amqp: AMQP) {
// add the new message to the list of messages to be processed. // add the new message to the list of messages to be processed.
existing.append(new_data); existing.append(new_data);
// if the message contains a mass mention, do not delay it any further.
if new_data[0].1.contains_mass_push_mention() {
task.run_immediately();
continue;
}
// put a cap on the amount of messages that can be queued, for particularly active channels // put a cap on the amount of messages that can be queued, for particularly active channels
if (existing.length() as u16) if (existing.length() as u16)
< revolt_config::config() < revolt_config::config()

10
crates/core/database/src/tasks/mod.rs
View File

@@ -26,6 +26,7 @@ pub fn start_workers(db: Database, amqp: AMQP) {
/// Task with additional information on when it should run /// Task with additional information on when it should run
pub struct DelayedTask<T> { pub struct DelayedTask<T> {
pub data: T, pub data: T,
run_now: bool,
last_updated: Instant, last_updated: Instant,
first_seen: Instant, first_seen: Instant,
} }
@@ -41,6 +42,7 @@ impl<T> DelayedTask<T> {
pub fn new(data: T) -> Self { pub fn new(data: T) -> Self {
DelayedTask { DelayedTask {
data, data,
run_now: false,
last_updated: Instant::now(), last_updated: Instant::now(),
first_seen: Instant::now(), first_seen: Instant::now(),
} }
@@ -51,9 +53,15 @@ impl<T> DelayedTask<T> {
self.last_updated = Instant::now() self.last_updated = Instant::now()
} }
/// Flag the task to run right away, regardless of the time
pub fn run_immediately(&mut self) {
self.run_now = true
}
/// Check if a task should run yet /// Check if a task should run yet
pub fn should_run(&self) -> bool { pub fn should_run(&self) -> bool {
self.first_seen.elapsed().as_secs() > EXPIRE_CONSTANT self.run_now
|| self.first_seen.elapsed().as_secs() > EXPIRE_CONSTANT
|| self.last_updated.elapsed().as_secs() > SAVE_CONSTANT || self.last_updated.elapsed().as_secs() > SAVE_CONSTANT
} }
} }

2
crates/core/database/src/util/bridge/v0.rs
View File

@@ -491,6 +491,7 @@ impl crate::Message {
edited: self.edited, edited: self.edited,
embeds: self.embeds, embeds: self.embeds,
mentions: self.mentions, mentions: self.mentions,
role_mentions: self.role_mentions,
replies: self.replies, replies: self.replies,
reactions: self.reactions, reactions: self.reactions,
interactions: self.interactions.into(), interactions: self.interactions.into(),
@@ -519,6 +520,7 @@ impl From<crate::PartialMessage> for PartialMessage {
edited: value.edited, edited: value.edited,
embeds: value.embeds, embeds: value.embeds,
mentions: value.mentions, mentions: value.mentions,
role_mentions: value.role_mentions,
replies: value.replies, replies: value.replies,
reactions: value.reactions, reactions: value.reactions,
interactions: value.interactions.map(Into::into), interactions: value.interactions.map(Into::into),

23
crates/core/database/src/util/bulk_permissions.rs
View File

@@ -94,9 +94,28 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
} }
} }
pub async fn from_channel_id(self, channel_id: String) -> BulkDatabasePermissionQuery<'z> {
let channel = self
.database
.fetch_channel(channel_id.as_str())
.await
.expect("Valid channel id");
drop(channel_id);
BulkDatabasePermissionQuery {
channel: Some(channel),
..self
}
}
pub fn members(self, members: &'z [Member]) -> BulkDatabasePermissionQuery { pub fn members(self, members: &'z [Member]) -> BulkDatabasePermissionQuery {
BulkDatabasePermissionQuery { BulkDatabasePermissionQuery {
members: Some(members.to_owned()), members: Some(members.to_owned()),
cached_member_perms: None,
users: None,
cached_members: None,
cached_users: None,
..self ..self
} }
} }
@@ -104,6 +123,10 @@ impl<'z> BulkDatabasePermissionQuery<'z> {
pub fn users(self, users: &'z [User]) -> BulkDatabasePermissionQuery { pub fn users(self, users: &'z [User]) -> BulkDatabasePermissionQuery {
BulkDatabasePermissionQuery { BulkDatabasePermissionQuery {
users: Some(users.to_owned()), users: Some(users.to_owned()),
cached_member_perms: None,
members: None,
cached_members: None,
cached_users: None,
..self ..self
} }
} }

4
crates/core/models/Cargo.toml
View File

@@ -24,9 +24,11 @@ revolt-config = { version = "0.8.4", path = "../config" }
revolt-permissions = { version = "0.8.4", path = "../permissions" } revolt-permissions = { version = "0.8.4", path = "../permissions" }
# Utility # Utility
regex = "1" regex = "1.11"
fancy-regex = "0.14"
indexmap = "1.9.3" indexmap = "1.9.3"
once_cell = "1.17.1" once_cell = "1.17.1"
num_enum = "0.6.1"
# Rocket # Rocket
rocket = { optional = true, version = "0.5.0-rc.2", default-features = false } rocket = { optional = true, version = "0.5.0-rc.2", default-features = false }

15
crates/core/models/src/v0/messages.rs
View File

@@ -1,8 +1,8 @@
use std::time::SystemTime; use std::time::SystemTime;
use fancy_regex::Regex;
use indexmap::{IndexMap, IndexSet}; use indexmap::{IndexMap, IndexSet};
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use regex::Regex;
use revolt_config::config; use revolt_config::config;
#[cfg(feature = "validator")] #[cfg(feature = "validator")]
@@ -18,6 +18,12 @@ use super::{Channel, Embed, File, Member, MessageWebhook, User, Webhook, RE_COLO
pub static RE_MENTION: Lazy<Regex> = pub static RE_MENTION: Lazy<Regex> =
Lazy::new(|| Regex::new(r"<@([0-9A-HJKMNP-TV-Z]{26})>").unwrap()); Lazy::new(|| Regex::new(r"<@([0-9A-HJKMNP-TV-Z]{26})>").unwrap());
pub static RE_ROLE_MENTION: Lazy<Regex> =
Lazy::new(|| Regex::new(r"<%([0-9A-HJKMNP-TV-Z]{26})>|@online|@everyone").unwrap());
pub static RE_BLOCKS: Lazy<Regex> =
Lazy::new(|| Regex::new(r"(?s)(?<!\\)(?:(```.+?(?<!\\)```)|`[^`\n\r]+?(?<!\\)`)").unwrap());
auto_derived_partial!( auto_derived_partial!(
/// Message /// Message
pub struct Message { pub struct Message {
@@ -58,6 +64,8 @@ auto_derived_partial!(
/// Array of user ids mentioned in this message /// Array of user ids mentioned in this message
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
pub mentions: Option<Vec<String>>, pub mentions: Option<Vec<String>>,
/// Array of role ids mentioned in this message
pub role_mentions: Option<Vec<String>>,
/// Array of message ids this message is replying to /// Array of message ids this message is replying to
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
pub replies: Option<Vec<String>>, pub replies: Option<Vec<String>>,
@@ -373,6 +381,11 @@ auto_derived!(
pub enum MessageFlags { pub enum MessageFlags {
/// Message will not send push / desktop notifications /// Message will not send push / desktop notifications
SuppressNotifications = 1, SuppressNotifications = 1,
/// Message will mention all users who can see the channel
MentionsEveryone = 2,
/// Message will mention all users who are online and can see the channel.
/// This cannot be true if MentionsEveryone is true
MentionsOnline = 3,
} }
/// Optional fields on message /// Optional fields on message

8
crates/core/permissions/src/models/channel.rs
View File

@@ -92,8 +92,14 @@ pub enum ChannelPermission {
/// Listen to other users /// Listen to other users
Listen = 1 << 36, Listen = 1 << 36,
// * Channel permissions two electric boogaloo
/// Mention everyone and online members
MentionEveryone = 1 << 37,
/// Mention roles
MentionRoles = 1 << 38,
// * Misc. permissions // * Misc. permissions
// % Bits 36 to 52: free area // % Bits 38 to 52: free area
// % Bits 53 to 64: do not use // % Bits 53 to 64: do not use
// * Grant all permissions // * Grant all permissions

5
crates/core/result/src/axum.rs
View File

@@ -1,5 +1,4 @@
use axum::{http::StatusCode, response::IntoResponse, Json}; use axum::{http::StatusCode, response::IntoResponse, Json};
use rocket::http::Status;
use crate::{Error, ErrorType}; use crate::{Error, ErrorType};
@@ -52,6 +51,7 @@ impl IntoResponse for Error {
ErrorType::ReachedMaximumBots => StatusCode::BAD_REQUEST, ErrorType::ReachedMaximumBots => StatusCode::BAD_REQUEST,
ErrorType::IsBot => StatusCode::BAD_REQUEST, ErrorType::IsBot => StatusCode::BAD_REQUEST,
ErrorType::IsNotBot => StatusCode::BAD_REQUEST,
ErrorType::BotIsPrivate => StatusCode::FORBIDDEN, ErrorType::BotIsPrivate => StatusCode::FORBIDDEN,
ErrorType::CannotReportYourself => StatusCode::BAD_REQUEST, ErrorType::CannotReportYourself => StatusCode::BAD_REQUEST,
@@ -80,6 +80,9 @@ impl IntoResponse for Error {
ErrorType::NotAVoiceChannel => StatusCode::BAD_REQUEST, ErrorType::NotAVoiceChannel => StatusCode::BAD_REQUEST,
ErrorType::AlreadyConnected => StatusCode::BAD_REQUEST, ErrorType::AlreadyConnected => StatusCode::BAD_REQUEST,
ErrorType::UnknownNode => StatusCode::BAD_REQUEST, ErrorType::UnknownNode => StatusCode::BAD_REQUEST,
ErrorType::InvalidFlagValue => StatusCode::BAD_REQUEST,
ErrorType::FeatureDisabled { .. } => StatusCode::BAD_REQUEST,
ErrorType::ProxyError => StatusCode::BAD_REQUEST, ErrorType::ProxyError => StatusCode::BAD_REQUEST,
ErrorType::FileTooSmall => StatusCode::UNPROCESSABLE_ENTITY, ErrorType::FileTooSmall => StatusCode::UNPROCESSABLE_ENTITY,
ErrorType::FileTooLarge { .. } => StatusCode::UNPROCESSABLE_ENTITY, ErrorType::FileTooLarge { .. } => StatusCode::UNPROCESSABLE_ENTITY,

7
crates/core/result/src/lib.rs
View File

@@ -122,6 +122,7 @@ pub enum ErrorType {
// ? Bot related errors // ? Bot related errors
ReachedMaximumBots, ReachedMaximumBots,
IsBot, IsBot,
IsNotBot,
BotIsPrivate, BotIsPrivate,
// ? User safety related errors // ? User safety related errors
@@ -149,6 +150,7 @@ pub enum ErrorType {
InvalidCredentials, InvalidCredentials,
InvalidProperty, InvalidProperty,
InvalidSession, InvalidSession,
InvalidFlagValue,
NotAuthenticated, NotAuthenticated,
DuplicateNonce, DuplicateNonce,
NotFound, NotFound,
@@ -175,6 +177,11 @@ pub enum ErrorType {
// ? Legacy errors // ? Legacy errors
VosoUnavailable, VosoUnavailable,
// ? Feature flag disabled in the config
FeatureDisabled {
feature: String,
},
} }
#[macro_export] #[macro_export]

4
crates/core/result/src/rocket.rs
View File

@@ -42,6 +42,7 @@ impl<'r> Responder<'r, 'static> for Error {
ErrorType::NotInGroup => Status::NotFound, ErrorType::NotInGroup => Status::NotFound,
ErrorType::AlreadyPinned => Status::BadRequest, ErrorType::AlreadyPinned => Status::BadRequest,
ErrorType::NotPinned => Status::BadRequest, ErrorType::NotPinned => Status::BadRequest,
ErrorType::InvalidFlagValue => Status::BadRequest,
ErrorType::UnknownServer => Status::NotFound, ErrorType::UnknownServer => Status::NotFound,
ErrorType::InvalidRole => Status::NotFound, ErrorType::InvalidRole => Status::NotFound,
@@ -57,6 +58,7 @@ impl<'r> Responder<'r, 'static> for Error {
ErrorType::ReachedMaximumBots => Status::BadRequest, ErrorType::ReachedMaximumBots => Status::BadRequest,
ErrorType::IsBot => Status::BadRequest, ErrorType::IsBot => Status::BadRequest,
ErrorType::IsNotBot => Status::BadRequest,
ErrorType::BotIsPrivate => Status::Forbidden, ErrorType::BotIsPrivate => Status::Forbidden,
ErrorType::CannotReportYourself => Status::BadRequest, ErrorType::CannotReportYourself => Status::BadRequest,
@@ -84,6 +86,8 @@ impl<'r> Responder<'r, 'static> for Error {
ErrorType::AlreadyConnected => Status::BadRequest, ErrorType::AlreadyConnected => Status::BadRequest,
ErrorType::NotConnected => Status::BadRequest, ErrorType::NotConnected => Status::BadRequest,
ErrorType::UnknownNode => Status::BadRequest, ErrorType::UnknownNode => Status::BadRequest,
ErrorType::FeatureDisabled { .. } => Status::BadRequest,
ErrorType::ProxyError => Status::BadRequest, ErrorType::ProxyError => Status::BadRequest,
ErrorType::FileTooSmall => Status::UnprocessableEntity, ErrorType::FileTooSmall => Status::UnprocessableEntity,
ErrorType::FileTooLarge { .. } => Status::UnprocessableEntity, ErrorType::FileTooLarge { .. } => Status::UnprocessableEntity,

6
crates/daemons/pushd/Cargo.toml
View File

@@ -10,6 +10,9 @@ revolt-database = { version = "0.8.4", path = "../../core/database" }
revolt-models = { version = "0.8.4", path = "../../core/models", features = [ revolt-models = { version = "0.8.4", path = "../../core/models", features = [
"validator", "validator",
] } ] }
revolt-presence = { version = "0.8.1", path = "../../core/presence", features = [
"redis-is-patched",
] }
amqprs = { version = "1.7.0" } amqprs = { version = "1.7.0" }
fcm_v1 = "0.3.0" fcm_v1 = "0.3.0"
@@ -20,9 +23,10 @@ tokio = "1.39.2"
async-trait = "0.1.81" async-trait = "0.1.81"
ulid = "1.0.0" ulid = "1.0.0"
authifier = "1.0.8" authifier = "1.0.10"
log = "0.4.11" log = "0.4.11"
pretty_env_logger = "0.4.0"
#serialization #serialization
serde_json = "1" serde_json = "1"

BIN
crates/daemons/pushd/Pushd Flowchart.graffle
View File

Binary file not shown.

10
crates/daemons/pushd/src/consumers/inbound/ack.rs
View File

@@ -71,8 +71,14 @@ impl AsyncConsumer for AckConsumer {
#[allow(clippy::disallowed_methods)] #[allow(clippy::disallowed_methods)]
let unreads = self.db.fetch_unread_mentions(&payload.user_id).await; let unreads = self.db.fetch_unread_mentions(&payload.user_id).await;
debug!("Processing unreads for {:}", &payload.user_id);
if let Ok(u) = &unreads { if let Ok(u) = &unreads {
if u.is_empty() { if u.is_empty() {
debug!(
"Discarding unread task (no mentions found) for {:}",
&payload.user_id
);
return; return;
} }
} else { } else {
@@ -95,6 +101,10 @@ impl AsyncConsumer for AckConsumer {
.collect(); .collect();
if apple_sessions.is_empty() { if apple_sessions.is_empty() {
debug!(
"Discarding unread task (no apn sessions found) for {:}",
&payload.user_id
);
return; return;
} }

275
crates/daemons/pushd/src/consumers/inbound/mass_mention.rs Normal file
View File

@@ -0,0 +1,275 @@
use std::{
collections::{HashMap, HashSet},
hash::RandomState,
};
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use async_trait::async_trait;
use revolt_database::{
events::rabbit::*, util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member,
MessageFlagsValue,
};
use revolt_models::v0::{MessageFlags, PushNotification};
pub struct MassMessageConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
}
impl Channeled for MassMessageConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl MassMessageConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> MassMessageConsumer {
MassMessageConsumer {
db,
authifier_db,
conn: None,
channel: None,
}
}
async fn fire_notification_for_users(&mut self, push: &PushNotification, users: &[String]) {
if let Ok(sessions) = self
.authifier_db
.find_sessions_with_subscription(users)
.await
{
let config = revolt_config::config().await;
for session in sessions {
if let Some(sub) = session.subscription {
let mut sendable = PayloadToService {
notification: PayloadKind::MessageNotification(push.clone()),
token: sub.auth,
user_id: session.user_id,
session_id: session.id,
extras: HashMap::new(),
};
let args: BasicPublishArguments;
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p265dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable).unwrap();
publish_message(self, payload.into(), args).await;
}
}
}
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for MassMessageConsumer {
/// This consumer handles adding mentions for all the users affected by a mass mention ping, and then sends out push notifications
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
let config = revolt_config::config().await;
let content = String::from_utf8(content).unwrap();
let payload: MassMessageSentPayload = serde_json::from_str(content.as_str()).unwrap();
debug!("Received mass message event");
// We should only ever receive clumped messages from a single channel, so it's safe to reuse this many times.
let mut query: Option<BulkDatabasePermissionQuery<'_>> = None;
let query_db = self.db.clone();
for push in payload.notifications {
if query.is_none() {
query = Some(
BulkDatabasePermissionQuery::from_server_id(&query_db, &payload.server_id)
.await
.from_channel_id(push.channel.id().to_string()) // wrong channel model, so fetch the right one
.await,
);
}
let existing_mentions: HashSet<String, RandomState> =
if let Some(ref mentions) = push.message.mentions {
HashSet::from_iter(mentions.iter().cloned())
} else {
HashSet::new()
};
// KNOWN QUIRK: if you mention @online and role(s), the offline members with the role(s) wont get pinged
if let Some(ref query) = query {
let flags = MessageFlagsValue(push.message.flags);
if flags.has(MessageFlags::MentionsEveryone) {
let mut db_query = self
.db
.fetch_all_members_chunked(&payload.server_id)
.await
.expect("Failed to fetch members from database");
let mut exhausted = false;
let ack_chnl = vec![push.channel.id().to_string()];
loop {
let mut chunk: Vec<Member> = vec![];
for _ in 0..config.pushd.mass_mention_chunk_size {
if let Some(member) = db_query.next().await {
chunk.push(member);
} else {
exhausted = true;
break;
}
}
let userids: Vec<String> =
chunk.iter().map(|member| member.id.user.clone()).collect();
debug!("Userids in chunk: {:?}", userids);
if let Err(err) = self
.db
.add_mention_to_many_unreads(push.channel.id(), &userids, &ack_chnl)
.await
{
revolt_config::capture_error(&err);
}
// ignore anyone in this list
let online_users = revolt_presence::filter_online(&userids).await;
let target_users: Vec<String> = userids
.iter()
.filter(|id| {
!online_users.contains(*id) && !existing_mentions.contains(*id)
})
.cloned()
.collect();
debug!(
"Userids after filter: {:?} (online: {:?}",
target_users, online_users
);
self.fire_notification_for_users(&push, &target_users).await;
if exhausted {
break;
}
}
} else if let Some(roles) = &push.message.role_mentions {
// role mentions
let _role_members = self
.db
.fetch_all_members_with_roles_chunked(&payload.server_id, roles)
.await;
debug!("role members: {:?}", _role_members);
if _role_members.is_err() {
revolt_config::capture_error(&_role_members.err().unwrap());
return;
}
let mut role_members = _role_members.unwrap();
let mut chunk = vec![];
let mut exhausted = false;
while !exhausted {
chunk.clear();
for _ in 0..config.pushd.mass_mention_chunk_size {
if let Some(member) = role_members.next().await {
chunk.push(member);
} else {
exhausted = true;
break;
}
}
let mut q = query.clone().members(&chunk);
let viewing_members: Vec<String> = q
.members_can_see_channel()
.await
.iter()
.filter_map(|(uid, viewable)| {
if *viewable && !existing_mentions.contains(uid) {
Some(uid.clone())
} else {
None
}
})
.collect();
debug!("viewing members: {:?}", viewing_members);
let online = revolt_presence::filter_online(&viewing_members).await;
debug!("online: {:?}", online);
let targets: Vec<String> = viewing_members
.iter()
.filter(|m| !online.contains(*m))
.cloned()
.collect();
debug!("targets: {:?}", targets);
self.fire_notification_for_users(&push, &targets).await;
}
}
}
}
}
}

1
crates/daemons/pushd/src/consumers/inbound/mod.rs
View File

@@ -3,4 +3,5 @@ pub mod fr_accepted;
pub mod fr_received; pub mod fr_received;
pub mod generic; pub mod generic;
mod internal; mod internal;
pub mod mass_mention;
pub mod message; pub mod message;

23
crates/daemons/pushd/src/consumers/outbound/apn.rs
View File

@@ -82,7 +82,7 @@ impl ApnsOutboundConsumer {
} }
} }
println!("Got badge count for APN: {}", mention_count); debug!("Got badge count for APN: {}", mention_count);
return Some(mention_count); return Some(mention_count);
} }
@@ -189,6 +189,10 @@ impl AsyncConsumer for ApnsOutboundConsumer {
data: BTreeMap::new(), data: BTreeMap::new(),
}; };
debug!(
"Sending friend request received for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await; resp = self.client.send(apn_payload).await;
} }
@@ -231,6 +235,10 @@ impl AsyncConsumer for ApnsOutboundConsumer {
data: BTreeMap::new(), data: BTreeMap::new(),
}; };
debug!(
"Sending friend request accept for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await; resp = self.client.send(apn_payload).await;
} }
PayloadKind::Generic(alert) => { PayloadKind::Generic(alert) => {
@@ -260,6 +268,10 @@ impl AsyncConsumer for ApnsOutboundConsumer {
data: BTreeMap::new(), data: BTreeMap::new(),
}; };
debug!(
"Sending generic notification for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await; resp = self.client.send(apn_payload).await;
} }
@@ -295,6 +307,10 @@ impl AsyncConsumer for ApnsOutboundConsumer {
channel_name: alert.channel.name().unwrap_or(&title), channel_name: alert.channel.name().unwrap_or(&title),
}; };
debug!(
"Sending message notification for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await; resp = self.client.send(apn_payload).await;
} }
PayloadKind::BadgeUpdate(badge) => { PayloadKind::BadgeUpdate(badge) => {
@@ -308,6 +324,7 @@ impl AsyncConsumer for ApnsOutboundConsumer {
data: BTreeMap::new(), data: BTreeMap::new(),
}; };
debug!("Sending badge update for user: {:}", &payload.user_id);
resp = self.client.send(apn_payload).await; resp = self.client.send(apn_payload).await;
} }
} }
@@ -322,6 +339,10 @@ impl AsyncConsumer for ApnsOutboundConsumer {
}), }),
.. ..
}) => { }) => {
info!(
"Removing APNS subscription id {:} (user: {:}) due to invalid token",
&payload.session_id, &payload.user_id
);
if let Err(err) = self if let Err(err) = self
.db .db
.remove_push_subscription_by_session_id(&payload.session_id) .remove_push_subscription_by_session_id(&payload.session_id)

26
crates/daemons/pushd/src/main.rs
View File

@@ -1,3 +1,6 @@
#[macro_use]
extern crate log;
use amqprs::{ use amqprs::{
channel::{ channel::{
BasicConsumeArguments, Channel, ExchangeDeclareArguments, QueueBindArguments, BasicConsumeArguments, Channel, ExchangeDeclareArguments, QueueBindArguments,
@@ -14,7 +17,7 @@ mod consumers;
use consumers::{ use consumers::{
inbound::{ inbound::{
ack::AckConsumer, fr_accepted::FRAcceptedConsumer, fr_received::FRReceivedConsumer, ack::AckConsumer, fr_accepted::FRAcceptedConsumer, fr_received::FRReceivedConsumer,
generic::GenericConsumer, message::MessageConsumer, generic::GenericConsumer, mass_mention::MassMessageConsumer, message::MessageConsumer,
}, },
outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer}, outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer},
}; };
@@ -22,6 +25,7 @@ use consumers::{
#[tokio::main(flavor = "multi_thread", worker_threads = 2)] #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
async fn main() { async fn main() {
let config = config().await; let config = config().await;
pretty_env_logger::init();
// Setup database // Setup database
let db = revolt_database::DatabaseInfo::Auto.connect().await.unwrap(); let db = revolt_database::DatabaseInfo::Auto.connect().await.unwrap();
@@ -96,6 +100,17 @@ async fn main() {
.await, .await,
); );
connections.push(
make_queue_and_consume(
&config,
&config.pushd.mass_mention_queue,
config.pushd.get_mass_mention_routing_key().as_str(),
None,
MassMessageConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
if !config.pushd.apn.pkcs8.is_empty() { if !config.pushd.apn.pkcs8.is_empty() {
connections.push( connections.push(
make_queue_and_consume( make_queue_and_consume(
@@ -223,11 +238,10 @@ where
.manual_ack(false) .manual_ack(false)
.finish(); .finish();
channel.basic_consume(consumer, args).await.unwrap(); let routing_key = channel.basic_consume(consumer, args).await.unwrap();
log::info!( info!(
"Consuming routing key {} as queue {}", "Consuming routing key {} as queue {}, tag {}",
routing_key, routing_key, queue_name, routing_key
queue_name
); );
(channel, connection) (channel, connection)
} }

4
crates/delta/Cargo.toml
View File

@@ -55,7 +55,7 @@ lettre = "0.10.0-alpha.4"
rocket = { version = "0.5.1", default-features = false, features = ["json"] } rocket = { version = "0.5.1", default-features = false, features = ["json"] }
rocket_cors = { git = "https://github.com/lawliet89/rocket_cors", rev = "072d90359b23e9b291df6b672c07c93de9c46011" } rocket_cors = { git = "https://github.com/lawliet89/rocket_cors", rev = "072d90359b23e9b291df6b672c07c93de9c46011" }
rocket_empty = { version = "0.1.1", features = ["schema"] } rocket_empty = { version = "0.1.1", features = ["schema"] }
rocket_authifier = { version = "1.0.9" } rocket_authifier = { version = "1.0.10" }
rocket_prometheus = "0.10.0-rc.3" rocket_prometheus = "0.10.0-rc.3"
# spec generation # spec generation
@@ -66,7 +66,7 @@ revolt_rocket_okapi = { version = "0.10.0", features = ["swagger"] }
amqprs = { version = "1.7.0" } amqprs = { version = "1.7.0" }
# core # core
authifier = "1.0.9" authifier = "1.0.10"
revolt-config = { path = "../core/config" } revolt-config = { path = "../core/config" }
revolt-database = { path = "../core/database", features = [ revolt-database = { path = "../core/database", features = [
"rocket-impl", "rocket-impl",

194
crates/delta/src/routes/channels/message_send.rs
View File

@@ -117,10 +117,11 @@ mod test {
use crate::{rocket, util::test::TestHarness}; use crate::{rocket, util::test::TestHarness};
use revolt_database::{ use revolt_database::{
util::{idempotency::IdempotencyKey, reference::Reference}, util::{idempotency::IdempotencyKey, reference::Reference},
Channel, Member, Message, PartialChannel, PartialMember, Role, Server, Channel, Member, Message, MessageFlagsValue, PartialChannel, PartialMember, Role, Server,
}; };
use revolt_models::v0::{self, DataCreateServerChannel}; use revolt_models::v0::{self, DataCreateServerChannel, MessageFlags};
use revolt_permissions::{ChannelPermission, OverrideField}; use revolt_permissions::{ChannelPermission, OverrideField};
use revolt_result::ErrorType;
#[rocket::async_test] #[rocket::async_test]
async fn message_mention_constraints() { async fn message_mention_constraints() {
@@ -486,4 +487,193 @@ mod test {
.await .await
.expect_err("Created message with missing reply and none fail"); .expect_err("Created message with missing reply and none fail");
} }
#[rocket::async_test]
async fn mass_mentions_test() {
let harness = TestHarness::new().await;
let (_, _, user) = harness.new_user().await;
let (_, _, other_user) = harness.new_user().await;
let (server, _) = harness.new_server(&user).await;
let channel = harness.new_channel(&server).await;
let (role_id, mut role) = harness
.new_role(
&server,
1,
Some(OverrideField {
a: (ChannelPermission::MentionEveryone as i64)
| (ChannelPermission::MentionRoles as i64),
d: 0,
}),
)
.await;
let (mut other_member, _) = Member::create(&harness.db, &server, &other_user, None)
.await
.expect("Failed to add test member");
// Send a message with an everyone and role mention.
// Should fail
let bad_message_with_mentions = Message::create_from_api(
&harness.db,
Some(&harness.amqp),
channel.clone(),
v0::DataMessageSend {
content: Some(format!("Mentioning @everyone and role <%{}>", &role_id)),
nonce: None,
attachments: None,
replies: None,
embeds: None,
masquerade: None,
interactions: None,
flags: None,
},
v0::MessageAuthor::User(
&other_user
.clone()
.into(&harness.db, Some(&other_user))
.await,
),
Some(
other_user
.clone()
.into(&harness.db, Some(&other_user))
.await,
),
Some(other_member.clone().into()),
other_user.limits().await,
IdempotencyKey::unchecked_from_string("1".to_string()),
false,
true,
)
.await
.expect_err("Should not have created message with everyone and role pings");
assert!(
matches!(
bad_message_with_mentions.error_type,
ErrorType::MissingPermission { .. }
),
"Intentional permissions error did not return MissingPermission"
);
// Send a mass mention inside a codeblock.
// Should be undetected and therefor pass
let message_with_codeblock = Message::create_from_api(
&harness.db,
Some(&harness.amqp),
channel.clone(),
v0::DataMessageSend {
content: Some(format!("Mentioning `@everyone` and role `<%{}>`", &role_id)),
nonce: None,
attachments: None,
replies: None,
embeds: None,
masquerade: None,
interactions: None,
flags: None,
},
v0::MessageAuthor::User(
&other_user
.clone()
.into(&harness.db, Some(&other_user))
.await,
),
Some(
other_user
.clone()
.into(&harness.db, Some(&other_user))
.await,
),
Some(other_member.clone().into()),
other_user.limits().await,
IdempotencyKey::unchecked_from_string("1".to_string()),
false,
true,
)
.await
.expect("Failed to create message with everyone and role pings in codeblocks");
assert!(
message_with_codeblock.flags.is_none()
|| !MessageFlagsValue(message_with_codeblock.flags.unwrap())
.has(MessageFlags::MentionsEveryone),
"Message flags mentions everyone when inside codeblock",
);
assert!(
message_with_codeblock.role_mentions.is_none(),
"Role mentions detected when inside codeblock"
);
other_member.roles.push(role_id.clone());
harness
.db
.update_member(
&other_member.id,
&PartialMember {
avatar: None,
id: None,
joined_at: None,
nickname: None,
roles: Some(vec![role_id.clone()]),
timeout: None,
},
vec![],
)
.await
.expect("Failed to add role to user");
// Send a message with an everyone and role mention.
// Should succeed
let message_with_mentions = Message::create_from_api(
&harness.db,
Some(&harness.amqp),
channel.clone(),
v0::DataMessageSend {
content: Some(format!("Mentioning @everyone and role <%{}>", &role_id)),
nonce: None,
attachments: None,
replies: None,
embeds: None,
masquerade: None,
interactions: None,
flags: None,
},
v0::MessageAuthor::User(
&other_user
.clone()
.into(&harness.db, Some(&other_user))
.await,
),
Some(
other_user
.clone()
.into(&harness.db, Some(&other_user))
.await,
),
Some(other_member.clone().into()),
other_user.limits().await,
IdempotencyKey::unchecked_from_string("1".to_string()),
false,
true,
)
.await
.expect("Failed to create message with everyone and role pings");
assert!(
message_with_mentions.flags.is_some(),
"Message flags is None",
);
assert!(
MessageFlagsValue(message_with_mentions.flags.unwrap())
.has(MessageFlags::MentionsEveryone),
"Message flags does not mention everyone. Flag value is {}",
message_with_mentions.flags.unwrap()
);
assert!(
message_with_mentions.role_mentions.is_some(),
"Message has no role mentions"
);
}
} }

25
crates/delta/src/util/test.rs
View File

@@ -5,11 +5,12 @@ use authifier::{
use futures::StreamExt; use futures::StreamExt;
use rand::Rng; use rand::Rng;
use redis_kiss::redis::aio::PubSub; use redis_kiss::redis::aio::PubSub;
use revolt_database::util::idempotency::IdempotencyKey;
use revolt_database::{ use revolt_database::{
events::client::EventV1, Channel, Database, Member, Message, Server, User, AMQP, events::client::EventV1, Channel, Database, Member, Message, Server, User, AMQP,
}; };
use revolt_database::{util::idempotency::IdempotencyKey, Role};
use revolt_models::v0; use revolt_models::v0;
use revolt_permissions::OverrideField;
use rocket::http::Header; use rocket::http::Header;
use rocket::local::asynchronous::{Client, LocalRequest, LocalResponse}; use rocket::local::asynchronous::{Client, LocalRequest, LocalResponse};
@@ -122,6 +123,28 @@ impl TestHarness {
.expect("Failed to create test server") .expect("Failed to create test server")
} }
pub async fn new_role(
&self,
server: &Server,
rank: i64,
overrides: Option<OverrideField>,
) -> (String, Role) {
let role = Role {
name: TestHarness::rand_string(),
permissions: overrides.unwrap_or(OverrideField { a: 0, d: 0 }),
rank,
colour: None,
hoist: false,
};
let id = role
.create(&self.db, &server.id)
.await
.expect("Failed to create test role");
(id, role)
}
pub async fn new_channel(&self, server: &Server) -> Channel { pub async fn new_channel(&self, server: &Server) -> Channel {
Channel::create_server_channel( Channel::create_server_channel(
&self.db, &self.db,