HMC/handmade-revolt-backend
6
0
Fork 0
forked from jmug/stoatchat
Code Pull Requests Actions Packages Releases 2 Activity

Merge remote-tracking branch 'revoltchat/master' into webhooks

Browse Source
This commit is contained in:
Zomatree
2023-04-25 20:38:09 +01:00
parent 8c5de18991 1afb27a508
commit 175866ce00
90 changed files with 3907 additions and 634 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
crates/quark/Cargo.toml
View File

@@ -1,8 +1,8 @@
[package]
name = "revolt-quark"
version = "0.5.17"
license = "AGPL-3.0-or-later"
version = "0.5.19"
edition = "2021"
license = "AGPL-3.0-or-later"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -26,10 +26,10 @@ default = [ "test" ]
[dependencies]
# Serialisation
revolt_optional_struct = "0.2.0"
serde = { version = "1", features = ["derive"] }
validator = { version = "0.14", features = ["derive"] }
iso8601-timestamp = { version = "0.1.8", features = ["schema", "bson"] }
optional_struct = { git = "https://github.com/insertish/OptionalStruct", rev = "ee56427cee1f007839825d93d07fffd5a5e038c7" }
# Formats
bincode = "1.3.3"
@@ -42,7 +42,7 @@ revolt_okapi = "0.9.1"
revolt_rocket_okapi = { version = "0.9.1", features = [ "swagger" ] }
# Databases
redis-kiss = { version = "0.1.3" }
redis-kiss = { version = "0.1.4" }
mongodb = { optional = true, version = "2.1.0", default-features = false }
# Async
@@ -57,6 +57,7 @@ log = "0.4.14"
pretty_env_logger = "0.4.0"
# Util
rand = "0.8.5"
ulid = "0.5.0"
regex = "1.5.5"
nanoid = "0.4.0"
@@ -88,3 +89,7 @@ authifier = { version = "1.0.7", features = [ "async-std-runtime" ] }
# Sentry
sentry = "0.25.0"
# Core
revolt-result = { path = "../core/result", features = [ "serde", "schemas" ] }
revolt-presence = { path = "../core/presence", features = [ "redis-is-patched" ] }

34
crates/quark/examples/test.rs
View File

@@ -1,34 +0,0 @@
use revolt_quark::models::user::PartialUser;
use revolt_quark::presence::{
presence_create_session, presence_delete_session, presence_filter_online, presence_is_online,
};
use revolt_quark::*;
#[async_std::main]
async fn main() {
let db = DatabaseInfo::Dummy.connect().await.unwrap();
let sus = PartialUser {
username: Some("neat".into()),
..Default::default()
};
db.update_user("user id", &sus, vec![]).await.unwrap();
dbg!(presence_create_session("entry", 0).await);
dbg!(presence_is_online("entry").await);
dbg!(presence_filter_online(&["a".into(), "b".into(), "entry".into()]).await);
dbg!(presence_delete_session("entry", 0).await);
dbg!(presence_is_online("entry").await);
dbg!(presence_create_session("entry", 0).await);
dbg!(presence_create_session("entry", 0).await);
dbg!(presence_delete_session("entry", 0).await);
dbg!(presence_is_online("entry").await);
dbg!(presence_delete_session("entry", 1).await);
dbg!(presence_is_online("entry").await);
// __set_key("dietz", vec![0xFF]).await;
// dbg!(presence_filter_online(&["dietz".into(), "nuts".into()]).await);
}

12
crates/quark/src/events/impl.rs
View File

@@ -7,11 +7,11 @@ use crate::{
user::{PartialUser, Presence, RelationshipStatus},
Channel, Member, User,
},
perms,
presence::presence_filter_online,
Database, Permission, Result,
perms, Database, Permission, Result,
};
use revolt_presence::filter_online;
use super::{
client::EventV1,
state::{Cache, State},
@@ -99,7 +99,7 @@ impl State {
let mut user = self.clone_user();
// Find all relationships to the user.
let mut user_ids: Vec<String> = user
let mut user_ids: HashSet<String> = user
.relations
.as_ref()
.map(|arr| arr.iter().map(|x| x.id.to_string()).collect())
@@ -128,14 +128,14 @@ impl State {
for channel in &channels {
match channel {
Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
user_ids.append(&mut recipients.clone());
user_ids.extend(&mut recipients.clone().into_iter());
}
_ => {}
}
}
// Fetch presence data for known users.
let online_ids = presence_filter_online(&user_ids).await;
let online_ids = filter_online(&user_ids.iter().cloned().collect::<Vec<String>>()).await;
user.online = Some(true);
// Fetch user data.

11
crates/quark/src/impl/dummy/admin/migrations.rs
View File

@@ -1,11 +0,0 @@
use crate::{AbstractMigrations, Result};
use super::super::DummyDb;
#[async_trait]
impl AbstractMigrations for DummyDb {
async fn migrate_database(&self) -> Result<()> {
info!("Migrating the database.");
Ok(())
}
}

1
crates/quark/src/impl/dummy/mod.rs
View File

@@ -1,7 +1,6 @@
use crate::AbstractDatabase;
pub mod admin {
pub mod migrations;
pub mod stats;
}

1
crates/quark/src/impl/generic/admin/migrations.rs
View File

@@ -1 +0,0 @@

10
crates/quark/src/impl/generic/channels/message.rs
View File

@@ -1,5 +1,6 @@
use std::collections::HashSet;
use revolt_presence::filter_online;
use serde_json::json;
use ulid::Ulid;
use validator::Validate;
@@ -14,7 +15,6 @@ use crate::{
Channel, Emoji, Message, User,
},
permissions::PermissionCalculator,
presence::presence_filter_online,
tasks::ack::AckEvent,
types::{
january::{Embed, Text},
@@ -79,7 +79,7 @@ impl Message {
Channel::DirectMessage { recipients, .. }
| Channel::Group { recipients, .. } => {
target_ids = (&recipients.iter().cloned().collect::<HashSet<String>>()
- &presence_filter_online(recipients).await)
- &filter_online(recipients).await)
.into_iter()
.collect::<Vec<String>>();
}
@@ -313,12 +313,12 @@ impl IntoUsers for Message {
impl IntoUsers for Vec<Message> {
fn get_user_ids(&self) -> Vec<String> {
let mut ids = vec![];
let mut ids = HashSet::new();
for message in self {
ids.append(&mut message.get_user_ids());
ids.extend(&mut message.get_user_ids().into_iter());
}
ids
ids.into_iter().collect()
}
}

4
crates/quark/src/impl/generic/mod.rs
View File

@@ -1,9 +1,5 @@
//! Database agnostic implementations.
pub mod admin {
pub mod migrations;
}
pub mod media {
pub mod attachment;
pub mod emoji;

4
crates/quark/src/impl/generic/users/user.rs
View File

@@ -4,11 +4,11 @@ use crate::models::user::{
};
use crate::permissions::defn::UserPerms;
use crate::permissions::r#impl::user::get_relationship;
use crate::presence::presence_filter_online;
use crate::{perms, Database, Error, Result};
use futures::try_join;
use impl_ops::impl_op_ex_commutative;
use revolt_presence::filter_online;
use std::ops;
impl_op_ex_commutative!(+ |a: &i32, b: &Badges| -> i32 { *a | *b as i32 });
@@ -98,7 +98,7 @@ impl User {
/// Fetch foreign users by a list of IDs
pub async fn fetch_foreign_users(db: &Database, user_ids: &[String]) -> Result<Vec<User>> {
let online_ids = presence_filter_online(user_ids).await;
let online_ids = filter_online(user_ids).await;
Ok(db
.fetch_users(user_ids)

26
crates/quark/src/impl/mongo/admin/migrations.rs
View File

@@ -1,26 +0,0 @@
use crate::{AbstractMigrations, Result};
use super::super::MongoDb;
mod init;
mod scripts;
#[async_trait]
impl AbstractMigrations for MongoDb {
async fn migrate_database(&self) -> Result<()> {
info!("Migrating the database.");
let list = self
.list_database_names(None, None)
.await
.expect("Failed to fetch database names.");
if list.iter().any(|x| x == "revolt") {
scripts::migrate_database(self).await;
} else {
init::create_database(self).await;
}
Ok(())
}
}

194
crates/quark/src/impl/mongo/admin/migrations/init.rs
View File

@@ -1,194 +0,0 @@
use crate::r#impl::mongo::MongoDb;
use super::scripts::LATEST_REVISION;
use mongodb::bson::doc;
use mongodb::options::CreateCollectionOptions;
pub async fn create_database(db: &MongoDb) {
info!("Creating database.");
let db = db.db();
db.create_collection("accounts", None)
.await
.expect("Failed to create accounts collection.");
db.create_collection("users", None)
.await
.expect("Failed to create users collection.");
db.create_collection("channels", None)
.await
.expect("Failed to create channels collection.");
db.create_collection("messages", None)
.await
.expect("Failed to create messages collection.");
db.create_collection("servers", None)
.await
.expect("Failed to create servers collection.");
db.create_collection("server_members", None)
.await
.expect("Failed to create server_members collection.");
db.create_collection("server_bans", None)
.await
.expect("Failed to create server_bans collection.");
db.create_collection("channel_invites", None)
.await
.expect("Failed to create channel_invites collection.");
db.create_collection("channel_unreads", None)
.await
.expect("Failed to create channel_unreads collection.");
db.create_collection("migrations", None)
.await
.expect("Failed to create migrations collection.");
db.create_collection("attachments", None)
.await
.expect("Failed to create attachments collection.");
db.create_collection("user_settings", None)
.await
.expect("Failed to create user_settings collection.");
db.create_collection("safety_reports", None)
.await
.expect("Failed to create safety_reports collection.");
db.create_collection("safety_snapshots", None)
.await
.expect("Failed to create safety_snapshots collection.");
db.create_collection("bots", None)
.await
.expect("Failed to create bots collection.");
db.create_collection(
"pubsub",
CreateCollectionOptions::builder()
.capped(true)
.size(1_000_000)
.build(),
)
.await
.expect("Failed to create pubsub collection.");
db.run_command(
doc! {
"createIndexes": "users",
"indexes": [
{
"key": {
"username": 1_i32
},
"name": "username",
"unique": true,
"collation": {
"locale": "en",
"strength": 2_i32
}
}
]
},
None,
)
.await
.expect("Failed to create username index.");
db.run_command(
doc! {
"createIndexes": "messages",
"indexes": [
{
"key": {
"content": "text"
},
"name": "content"
},
{
"key": {
"channel": 1_i32,
"_id": 1_i32
},
"name": "channel_id_compound"
},
{
"key": {
"author": 1_i32
},
"name": "author"
}
]
},
None,
)
.await
.expect("Failed to create message index.");
db.run_command(
doc! {
"createIndexes": "channel_unreads",
"indexes": [
{
"key": {
"_id.channel": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
},
{
"key": {
"_id.user": 1_i32,
},
"name": "user_id"
}
]
},
None,
)
.await
.expect("Failed to create channel_unreads index.");
db.run_command(
doc! {
"createIndexes": "server_members",
"indexes": [
{
"key": {
"_id.server": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
},
{
"key": {
"_id.user": 1_i32,
},
"name": "user_id"
}
]
},
None,
)
.await
.expect("Failed to create server_members index.");
db.collection("migrations")
.insert_one(
doc! {
"_id": 0_i32,
"revision": LATEST_REVISION
},
None,
)
.await
.expect("Failed to save migration info.");
info!("Created database.");
}

760
crates/quark/src/impl/mongo/admin/migrations/scripts.rs
View File

@@ -1,760 +0,0 @@
use std::{time::Duration, ops::BitXor};
use bson::{Bson, DateTime};
use futures::StreamExt;
use mongodb::{
bson::{doc, from_bson, from_document, to_document, Document},
options::FindOptions,
};
use serde::{Deserialize, Serialize};
use crate::{r#impl::mongo::MongoDb, Permission, DEFAULT_PERMISSION_SERVER};
#[derive(Serialize, Deserialize)]
struct MigrationInfo {
_id: i32,
revision: i32,
}
pub const LATEST_REVISION: i32 = 21;
pub async fn migrate_database(db: &MongoDb) {
let migrations = db.col::<Document>("migrations");
let data = migrations
.find_one(None, None)
.await
.expect("Failed to fetch migration data.");
if let Some(doc) = data {
let info: MigrationInfo =
from_document(doc).expect("Failed to read migration information.");
let revision = run_migrations(db, info.revision).await;
migrations
.update_one(
doc! {
"_id": info._id
},
doc! {
"$set": {
"revision": revision
}
},
None,
)
.await
.expect("Failed to commit migration information.");
info!("Migration complete. Currently at revision {}.", revision);
} else {
panic!("Database was configured incorrectly, possibly because initalization failed.")
}
}
pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
info!("Starting database migration.");
if revision <= 0 {
info!("Running migration [revision 0]: Test migration system.");
}
if revision <= 1 {
info!("Running migration [revision 1 / 2021-04-24]: Migrate to Autumn v1.0.0.");
let messages = db.col::<Document>("messages");
let attachments = db.col::<Document>("attachments");
messages
.update_many(
doc! { "attachment": { "$exists": 1_i32 } },
doc! { "$set": { "attachment.tag": "attachments", "attachment.size": 0_i32 } },
None,
)
.await
.expect("Failed to update messages.");
attachments
.update_many(
doc! {},
doc! { "$set": { "tag": "attachments", "size": 0_i32 } },
None,
)
.await
.expect("Failed to update attachments.");
}
if revision <= 2 {
info!("Running migration [revision 2 / 2021-05-08]: Add servers collection.");
db.db()
.create_collection("servers", None)
.await
.expect("Failed to create servers collection.");
}
if revision <= 3 {
info!("Running migration [revision 3 / 2021-05-25]: Support multiple file uploads, add channel_unreads and user_settings.");
let messages = db.col::<Document>("messages");
let mut cursor = messages
.find(
doc! {
"attachment": {
"$exists": 1_i32
}
},
FindOptions::builder()
.projection(doc! {
"_id": 1_i32,
"attachments": [ "$attachment" ]
})
.build(),
)
.await
.expect("Failed to fetch messages.");
while let Some(result) = cursor.next().await {
let doc = result.unwrap();
let id = doc.get_str("_id").unwrap();
let attachments = doc.get_array("attachments").unwrap();
messages
.update_one(
doc! { "_id": id },
doc! { "$unset": { "attachment": 1_i32 }, "$set": { "attachments": attachments } },
None,
)
.await
.unwrap();
}
db.db()
.create_collection("channel_unreads", None)
.await
.expect("Failed to create channel_unreads collection.");
db.db()
.create_collection("user_settings", None)
.await
.expect("Failed to create user_settings collection.");
}
if revision <= 4 {
info!("Running migration [revision 4 / 2021-06-01]: Add more server collections.");
db.db()
.create_collection("server_members", None)
.await
.expect("Failed to create server_members collection.");
db.db()
.create_collection("server_bans", None)
.await
.expect("Failed to create server_bans collection.");
db.db()
.create_collection("channel_invites", None)
.await
.expect("Failed to create channel_invites collection.");
}
if revision <= 5 {
info!("Running migration [revision 5 / 2021-06-26]: Add permissions.");
#[derive(Serialize)]
struct Server {
pub default_permissions: (i32, i32),
}
let server = Server {
default_permissions: (0_i32, 0_i32),
};
db.col::<Document>("servers")
.update_many(
doc! {},
doc! {
"$set": to_document(&server).unwrap()
},
None,
)
.await
.expect("Failed to migrate servers.");
}
if revision <= 6 {
info!("Running migration [revision 6 / 2021-07-09]: Add message text index.");
db.db()
.run_command(
doc! {
"createIndexes": "messages",
"indexes": [
{
"key": {
"content": "text"
},
"name": "content"
}
]
},
None,
)
.await
.expect("Failed to create message index.");
}
if revision <= 7 {
info!("Running migration [revision 7 / 2021-08-11]: Add message text index.");
db.db()
.create_collection("bots", None)
.await
.expect("Failed to create bots collection.");
}
if revision <= 8 {
info!("Running migration [revision 8 / 2021-09-10]: Update to Authifier version 1.");
db.db()
.run_command(
doc! {
"dropIndexes": "accounts",
"index": ["email", "email_normalised"]
},
None,
)
.await
.expect("Failed to delete legacy account indexes.");
let col = db.col::<Document>("sessions");
let mut cursor = db
.col::<Document>("accounts")
.find(doc! {}, None)
.await
.unwrap();
while let Some(doc) = cursor.next().await {
if let Ok(account) = doc {
let id = account.get_str("_id").unwrap();
if let Some(sessions) = account.get("sessions") {
#[derive(Deserialize)]
struct Session {
id: String,
token: String,
friendly_name: String,
subscription: Option<Document>,
}
let sessions = from_bson::<Vec<Session>>(sessions.clone()).unwrap();
for session in sessions {
info!("Converting session {} to new format.", &session.id);
let mut doc = doc! {
"_id": session.id,
"token": session.token,
"user_id": id,
"name": session.friendly_name,
};
if let Some(sub) = session.subscription {
doc.insert("subscription", sub);
}
col.insert_one(doc, None).await.ok();
}
} else {
info!("Account doesn't have any sessions!");
}
}
}
db.col::<Document>("accounts")
.update_many(
doc! {},
doc! {
"$unset": {
"sessions": 1_i32,
},
"$set": {
"mfa": {
"recovery_codes": []
}
}
},
None,
)
.await
.unwrap();
}
if revision <= 9 {
info!("Running migration [revision 9 / 2021-09-14]: Switch from last_message to last_message_id.");
let mut cursor = db
.col::<Document>("channels")
.find(doc! {}, None)
.await
.unwrap();
while let Some(doc) = cursor.next().await {
if let Ok(channel) = doc {
let channel_id = channel.get_str("_id").unwrap();
if let Some(last_message) = channel.get("last_message") {
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct Obj {
#[serde(rename = "_id")]
id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(untagged)]
pub enum LastMessage {
Obj(Obj),
Id(String),
}
let lm = from_bson::<LastMessage>(last_message.clone()).unwrap();
let id = match lm {
LastMessage::Obj(Obj { id }) => id,
LastMessage::Id(id) => id,
};
info!("Converting session {} to new format.", &channel_id);
db.col::<Document>("channels")
.update_one(
doc! {
"_id": channel_id
},
doc! {
"$set": {
"last_message_id": id
},
"$unset": {
"last_message": 1_i32,
}
},
None,
)
.await
.unwrap();
} else {
info!("{} has no last_message.", &channel_id);
}
}
}
}
if revision <= 10 {
info!("Running migration [revision 10 / 2021-11-01]: Remove nonce values on channels and servers.");
db.col::<Document>("servers")
.update_many(
doc! {},
doc! {
"$unset": {
"nonce": 1_i32,
}
},
None,
)
.await
.unwrap();
db.col::<Document>("channels")
.update_many(
doc! {},
doc! {
"$unset": {
"nonce": 1_i32,
}
},
None,
)
.await
.unwrap();
}
if revision <= 11 {
info!("Running migration [revision 11 / 2021-11-14]: Add indexes to database.");
db.db()
.run_command(
doc! {
"createIndexes": "messages",
"indexes": [
{
"key": {
"channel": 1_i32
},
"name": "channel"
}
]
},
None,
)
.await
.expect("Failed to create message index.");
db.db()
.run_command(
doc! {
"createIndexes": "channel_unreads",
"indexes": [
{
"key": {
"_id.channel": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
},
{
"key": {
"_id.user": 1_i32,
},
"name": "user_id"
}
]
},
None,
)
.await
.expect("Failed to create channel_unreads index.");
db.db()
.run_command(
doc! {
"createIndexes": "server_members",
"indexes": [
{
"key": {
"_id.server": 1_i32,
"_id.user": 1_i32,
},
"name": "compound_id"
},
{
"key": {
"_id.user": 1_i32,
},
"name": "user_id"
}
]
},
None,
)
.await
.expect("Failed to create server_members index.");
}
if revision <= 12 {
info!("Running migration [revision 12 / 2021-11-21]: Add indexes to database.");
db.db()
.run_command(
doc! {
"createIndexes": "messages",
"indexes": [
{
"key": {
"channel": 1_i32,
"_id": 1_i32
},
"name": "channel_id_compound"
}
]
},
None,
)
.await
.expect("Failed to create message index.");
}
if revision <= 13 {
info!("Running migration [revision 13 / 22-02-2022]: Wipe legacy permission values.");
warn!("This is a destructive operation and will wipe existing permission data (excl. defaults for SendMessage).");
warn!("Taking a backup is advised.");
warn!("Continuing in 10 seconds...");
async_std::task::sleep(Duration::from_secs(10)).await;
let servers = db.col::<Document>("servers");
let mut cursor = servers.find(doc! {}, None).await.unwrap();
while let Some(Ok(mut document)) = cursor.next().await {
let id = document.get_str("_id").unwrap().to_string();
info!("Updating server {id}");
let mut update = doc! {};
// Try to pluck channel permission SendMessage (0x2)
// Structure of default_permissions used to be [server, channel]
let has_send = document
.get_array("default_permissions")
.map(|x| {
x.get(1)
.map(|x| x.as_i32().map(|x| (x as u32 & 0x2) == 0x2))
})
.ok()
.flatten()
.flatten()
.unwrap_or_default();
update.insert(
"default_permissions",
// Remove Send Message permission if it wasn't originally granted
DEFAULT_PERMISSION_SERVER.bitxor(if has_send { 0 } else { Permission::SendMessage as u64}) as i64,
);
if let Some(Bson::Document(mut roles)) = document.remove("roles") {
for role in roles.keys().cloned().collect::<Vec<String>>() {
if let Some(Bson::Document(role)) = roles.get_mut(role) {
role.insert(
"permissions",
doc! {
"a": 0_i64,
"d": 0_i64,
},
);
}
}
update.insert("roles", roles);
}
servers
.update_one(doc! { "_id": id }, doc! { "$set": update }, None)
.await
.unwrap();
}
let channels = db.col::<Document>("channels");
let mut cursor = channels.find(doc! {}, None).await.unwrap();
while let Some(Ok(document)) = cursor.next().await {
let id = document.get_str("_id").unwrap().to_string();
info!("Updating channel {id}");
let mut unset = doc! {
"permissions": 1_i32,
"role_permissions": 1_i32,
};
// Try to pluck channel permission SendMessage (0x2)
let has_send = document
.get_i32("default_permissions")
.map(|x| (x as u32 & 0x2) == 0x2)
.unwrap_or(true);
if has_send {
// Let parent permissions fall through.
unset.insert("default_permissions", 1_i32);
}
let mut update = doc! {
"$unset": unset
};
if !has_send {
// Block send message permission.
update.insert(
"$set",
doc! {
"default_permissions": {
"a": 0_i64,
"d": Permission::SendMessage as i64
}
},
);
}
channels
.update_one(doc! { "_id": id }, update, None)
.await
.unwrap();
}
}
if revision <= 14 {
info!("Running migration [revision 14 / 21-04-2022]: Split content into content and system fields.");
db.col::<Document>("messages")
.update_many(
doc! {
"content": {
"$type": "object"
}
},
doc! {
"$rename": {
"content": "system"
}
},
None,
)
.await
.unwrap();
}
if revision <= 15 {
info!("Running migration [revision 15 / 04-06-2022]: Migrate Authifier to latest version.");
let db = authifier::Database::MongoDb(authifier::database::MongoDb(db.db()));
db.run_migration(authifier::Migration::M2022_06_03EnsureUpToSpec)
.await
.unwrap();
}
if revision <= 16 {
info!("Running migration [revision 16 / 07-07-2022]: Add `emojis` collection and Authifier migration.");
let authifier_db = authifier::Database::MongoDb(authifier::database::MongoDb(db.db()));
authifier_db
.run_migration(authifier::Migration::M2022_06_09AddIndexForDeletion)
.await
.unwrap();
db.db()
.create_collection("emojis", None)
.await
.expect("Failed to create emojis collection.");
db.db()
.run_command(
doc! {
"createIndexes": "emojis",
"indexes": [
{
"key": {
"parent.id": 1_i32,
},
"name": "parent_id"
}
]
},
None,
)
.await
.expect("Failed to create emoji parent index.");
}
if revision <= 17 {
info!("Running migration [revision 17 / 15-07-2022]: Initialise `joined_at` property on server members.");
db.col::<Document>("server_members")
.update_many(
doc! {},
doc! {
"$set": {
"joined_at": DateTime::now().to_rfc3339_string()
}
},
None,
)
.await
.expect("Failed to update server members.");
}
if revision <= 18 {
info!("Running migration [revision 18 / 27-02-2022]: Create author index on messages. Drop plain channel index if exists.");
if db
.db()
.run_command(
doc! {
"dropIndexes": "messages",
"index": ["channel"]
},
None,
)
.await
.is_err()
{
info!("Failed to drop `messages.channel` index but this is ok since that means it's probably gone.");
}
db.db()
.run_command(
doc! {
"createIndexes": "messages",
"indexes": [
{
"key": {
"author": 1_i32,
},
"name": "author"
}
]
},
None,
)
.await
.expect("Failed to create messages author index.");
}
if revision <= 19 {
info!("Running migration [revision 19 / 27-02-2023]: Create report / snapshot collections, migrate to new model if applicable.");
// TODO: make these fail once production is migrated
if db
.db()
.create_collection("safety_reports", None)
.await
.is_err()
{
info!("Failed to create safety_reports collection but this is expected in production.");
}
if db
.db()
.create_collection("safety_snapshots", None)
.await
.is_err()
{
info!(
"Failed to create safety_snapshots collection but this is expected in production."
);
}
db.col::<Document>("safety_reports")
.update_many(
doc! {},
doc! {
"$set": {
"status": "Created"
}
},
None,
)
.await
.unwrap();
}
if revision <= 20 {
info!("Running migration [revision 20 / 28-02-2023]: Add index `snapshot.report_id`.");
db.db()
.run_command(
doc! {
"createIndexes": "safety_snapshots",
"indexes": [
{
"key": {
"report_id": 1_i32
},
"name": "report_id"
}
]
},
None,
)
.await
.expect("Failed to create safety snapshot index.");
}
// Need to migrate fields on attachments, change `user_id`, `object_id`, etc to `parent`.
// Reminder to update LATEST_REVISION when adding new migrations.
LATEST_REVISION
}

1
crates/quark/src/impl/mongo/mod.rs
View File

@@ -12,7 +12,6 @@ use serde::{Deserialize, Serialize};
use crate::{util::manipulation::prefix_keys, AbstractDatabase, Error, Result};
pub mod admin {
pub mod migrations;
pub mod stats;
}

3
crates/quark/src/lib.rs
View File

@@ -9,7 +9,7 @@ extern crate log;
#[macro_use]
extern crate impl_ops;
#[macro_use]
extern crate optional_struct;
extern crate revolt_optional_struct;
#[macro_use]
extern crate bitfield;
#[macro_use]
@@ -22,7 +22,6 @@ pub use redis_kiss;
pub mod events;
pub mod r#impl;
pub mod models;
pub mod presence;
pub mod tasks;
pub mod types;
pub mod util;

11
crates/quark/src/models/admin/migrations.rs
View File

@@ -1,11 +0,0 @@
use serde::{Deserialize, Serialize};
/// Document representing migration information
#[derive(Serialize, Deserialize)]
pub struct MigrationInfo {
/// Unique Id
#[serde(rename = "_id")]
id: i32,
/// Current database revision
revision: i32,
}

2
crates/quark/src/models/mod.rs
View File

@@ -1,5 +1,4 @@
mod admin {
pub mod migrations;
pub mod simple;
pub mod stats;
}
@@ -48,7 +47,6 @@ pub use channel_invite::Invite;
pub use channel_unread::ChannelUnread;
pub use emoji::Emoji;
pub use message::Message;
pub use migrations::MigrationInfo;
pub use report::Report;
pub use server::Server;
pub use server_ban::ServerBan;

64
crates/quark/src/presence/entry.rs
View File

@@ -1,64 +0,0 @@
use std::env;
use serde::{Deserialize, Serialize};
use once_cell::sync::Lazy;
pub static REGION_ID: Lazy<u16> = Lazy::new(|| env::var("REGION_ID")
.unwrap_or_else(|_| "0".to_string())
.parse()
.unwrap());
pub static REGION_KEY: Lazy<String> = Lazy::new(|| format!("region{}", &*REGION_ID));
/// Compact presence information for a user
#[derive(Serialize, Deserialize, Debug)]
pub struct PresenceEntry {
/// Region this session exists in
///
/// We can have up to 65535 regions
pub region_id: u16,
/// Unique session ID
pub session_id: u8,
/// Known flags about session
pub flags: u8,
}
impl PresenceEntry {
/// Create a new presence entry from a given session ID and known flags
pub fn from(session_id: u8, flags: u8) -> Self {
Self {
region_id: *REGION_ID,
session_id,
flags,
}
}
}
pub trait PresenceOp {
/// Find next available session ID
fn find_next_id(&self) -> u8;
}
impl PresenceOp for Vec<PresenceEntry> {
fn find_next_id(&self) -> u8 {
// O(n^2) scan algorithm
// should be relatively fast at low numbers anyways
for i in 0..255 {
let mut found = false;
for entry in self {
if entry.session_id == i {
found = true;
break;
}
}
if !found {
return i;
}
}
255
}
}

162
crates/quark/src/presence/mod.rs
View File

@@ -1,162 +0,0 @@
use std::collections::HashSet;
use redis_kiss::{get_connection, AsyncCommands};
mod entry;
mod operations;
use entry::{PresenceEntry, PresenceOp};
use operations::{
__add_to_set_sessions, __delete_key_presence_entry, __get_key_presence_entry,
__get_set_sessions, __remove_from_set_sessions, __set_key_presence_entry,
};
use crate::presence::operations::__delete_set_sessions;
use self::entry::REGION_KEY;
/// Create a new presence session, returns the ID of this session
pub async fn presence_create_session(user_id: &str, flags: u8) -> (bool, u8) {
info!("Creating a presence session for {user_id} with flags {flags}");
// Try to find the presence entry for this user.
let mut conn = get_connection().await.unwrap();
let mut entry: Vec<PresenceEntry> = __get_key_presence_entry(&mut conn, user_id)
.await
.unwrap_or_default();
// Return whether this was the first session.
let was_empty = entry.is_empty();
info!("User ID {} just came online.", &user_id);
// Generate session ID and push new entry.
let session_id = entry.find_next_id();
entry.push(PresenceEntry::from(session_id, flags));
__set_key_presence_entry(&mut conn, user_id, entry).await;
// Add to region set in case of failure.
__add_to_set_sessions(&mut conn, &REGION_KEY, user_id, session_id).await;
(was_empty, session_id)
}
/// Delete existing presence session
pub async fn presence_delete_session(user_id: &str, session_id: u8) -> bool {
presence_delete_session_internal(user_id, session_id, false).await
}
/// Delete existing presence session (but also choose whether to skip region)
async fn presence_delete_session_internal(
user_id: &str,
session_id: u8,
skip_region: bool,
) -> bool {
info!("Deleting presence session for {user_id} with id {session_id}");
// Return whether this was the last session.
let mut is_empty = false;
// Only continue if we can actually find one.
let mut conn = get_connection().await.unwrap();
let entry: Option<Vec<PresenceEntry>> = __get_key_presence_entry(&mut conn, user_id).await;
if let Some(entry) = entry {
let entries = entry
.into_iter()
.filter(|x| x.session_id != session_id)
.collect::<Vec<PresenceEntry>>();
// If entry is empty, then just delete it.
if entries.is_empty() {
__delete_key_presence_entry(&mut conn, user_id).await;
is_empty = true;
} else {
__set_key_presence_entry(&mut conn, user_id, entries).await;
}
// Remove from region set.
if !skip_region {
__remove_from_set_sessions(&mut conn, &REGION_KEY, user_id, session_id).await;
}
}
if is_empty {
info!("User ID {} just went offline.", &user_id);
}
is_empty
}
/// Check whether a given user ID is online
pub async fn presence_is_online(user_id: &str) -> bool {
if let Ok(mut conn) = get_connection().await {
conn.exists(user_id).await.unwrap_or(false)
} else {
false
}
}
/// Check whether a set of users is online, returns a set of the online user IDs
pub async fn presence_filter_online(user_ids: &'_ [String]) -> HashSet<String> {
// Ignore empty list immediately, to save time.
let mut set = HashSet::new();
if user_ids.is_empty() {
return set;
}
// We need to handle a special case where only one is present
// as for some reason or another, Redis does not like us sending
// a list of just one ID to the server.
if user_ids.len() == 1 {
if presence_is_online(&user_ids[0]).await {
set.insert(user_ids[0].to_string());
}
return set;
}
// Otherwise, go ahead as normal.
if let Ok(mut conn) = get_connection().await {
let data: Vec<Option<Vec<u8>>> = conn.get(user_ids).await.unwrap_or_default();
if data.is_empty() {
return set;
}
// We filter known values to figure out who is online.
for i in 0..user_ids.len() {
if data[i].is_some() {
set.insert(user_ids[i].to_string());
}
}
}
set
}
/// Reset any stale presence data
pub async fn presence_clear_region(region_id: Option<&str>) {
let region_id = region_id.unwrap_or(&*REGION_KEY);
let mut conn = get_connection().await.expect("Redis connection");
let sessions = __get_set_sessions(&mut conn, region_id).await;
if !sessions.is_empty() {
info!(
"Cleaning up {} sessions, this may take a while...",
sessions.len()
);
// Iterate and delete each session, this will
// also send out any relevant events.
for session in sessions {
let parts = session.split(':').collect::<Vec<&str>>();
if let (Some(user_id), Some(session_id)) = (parts.first(), parts.get(1)) {
if let Ok(session_id) = session_id.parse() {
presence_delete_session_internal(user_id, session_id, true).await;
}
}
}
// Then clear the set in Redis.
__delete_set_sessions(&mut conn, region_id).await;
info!("Clean up complete.");
}
}

57
crates/quark/src/presence/operations.rs
View File

@@ -1,57 +0,0 @@
use redis_kiss::{AsyncCommands, Conn};
use super::entry::PresenceEntry;
/// Set presence entry by given ID
pub async fn __set_key_presence_entry(conn: &mut Conn, id: &str, data: Vec<PresenceEntry>) {
let _: Option<()> = conn.set(id, bincode::serialize(&data).unwrap()).await.ok();
}
/// Delete presence entry by given ID
pub async fn __delete_key_presence_entry(conn: &mut Conn, id: &str) {
let _: Option<()> = conn.del(id).await.ok();
}
/// Get presence entry by given ID
pub async fn __get_key_presence_entry(conn: &mut Conn, id: &str) -> Option<Vec<PresenceEntry>> {
conn.get::<_, Option<Vec<u8>>>(id)
.await
.unwrap()
.map(|entry| bincode::deserialize(&entry[..]).unwrap())
}
/// Add to region session set
pub async fn __add_to_set_sessions(
conn: &mut Conn,
region_id: &str,
user_id: &str,
session_id: u8,
) {
let _: Option<()> = conn
.sadd(region_id, format!("{user_id}:{session_id}"))
.await
.ok();
}
/// Remove from region session set
pub async fn __remove_from_set_sessions(
conn: &mut Conn,
region_id: &str,
user_id: &str,
session_id: u8,
) {
let _: Option<()> = conn
.srem(region_id, format!("{user_id}:{session_id}"))
.await
.ok();
}
/// Get region session set as list
pub async fn __get_set_sessions(conn: &mut Conn, region_id: &str) -> Vec<String> {
conn.smembers::<_, Vec<String>>(region_id).await.unwrap()
}
/// Delete region session set
pub async fn __delete_set_sessions(conn: &mut Conn, region_id: &str) {
conn.del::<_, ()>(region_id).await.unwrap();
}

6
crates/quark/src/traits/admin/migrations.rs
View File

@@ -1,6 +0,0 @@
use crate::Result;
#[async_trait]
pub trait AbstractMigrations: Sync + Send {
async fn migrate_database(&self) -> Result<()>;
}

3
crates/quark/src/traits/mod.rs
View File

@@ -1,5 +1,4 @@
mod admin {
pub mod migrations;
pub mod stats;
}
@@ -33,7 +32,6 @@ mod safety {
pub mod snapshot;
}
pub use admin::migrations::AbstractMigrations;
pub use admin::stats::AbstractStats;
pub use media::attachment::AbstractAttachment;
@@ -59,7 +57,6 @@ pub use safety::snapshot::AbstractSnapshot;
pub trait AbstractDatabase:
Sync
+ Send
+ AbstractMigrations
+ AbstractStats
+ AbstractAttachment
+ AbstractEmoji

2
crates/quark/src/util/log.rs
View File

@@ -14,7 +14,7 @@ pub fn setup_logging(release: &'static str) -> sentry::ClientInitGuard {
info!("Starting {release}");
sentry::init((
"https://62fd0e02c5354905b4e286757f4beb16@sentry.insert.moe/4",
"https://d1d2a6f15c6245a987c532bbbcb30a04@glitchtip.insert.moe/2",
sentry::ClientOptions {
release: Some(release.into()),
..Default::default()

8
crates/quark/src/util/ref.rs
View File

@@ -1,11 +1,13 @@
use futures::future::join;
use revolt_presence::is_online;
use rocket::request::FromParam;
use schemars::schema::{InstanceType, Schema, SchemaObject, SingleOrVec};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use crate::models::{Bot, Channel, Emoji, Invite, Member, Message, Server, ServerBan, User, Webhook, Report};
use crate::presence::presence_is_online;
use crate::models::{
Bot, Channel, Emoji, Invite, Member, Message, Report, Server, ServerBan, User, Webhook
};
use crate::{Database, Error, Result};
/// Reference to some object in the database
@@ -23,7 +25,7 @@ impl Ref {
/// Fetch user from Ref
pub async fn as_user(&self, db: &Database) -> Result<User> {
let (user, online) = join(db.fetch_user(&self.id), presence_is_online(&self.id)).await;
let (user, online) = join(db.fetch_user(&self.id), is_online(&self.id)).await;
let mut user = user?;
user.online = Some(online);
Ok(user)

22
crates/quark/src/util/result.rs
View File

@@ -19,6 +19,12 @@ pub enum Error {
/// This error was not labeled :(
LabelMe,
/// Core crate error
Core {
#[serde(flatten)]
error: revolt_result::Error,
},
// ? Onboarding related errors
AlreadyOnboarded,
@@ -39,13 +45,13 @@ pub enum Error {
CannotEditMessage,
CannotJoinCall,
TooManyAttachments {
max: usize
max: usize,
},
TooManyReplies {
max: usize
max: usize,
},
TooManyChannels {
max: usize
max: usize,
},
EmptyMessage,
PayloadTooLarge,
@@ -64,10 +70,10 @@ pub enum Error {
max: usize,
},
TooManyEmoji {
max: usize
max: usize,
},
TooManyRoles {
max: usize
max: usize,
},
// ? Bot related errors
@@ -135,6 +141,11 @@ impl Error {
error: validation_error,
})
}
/// Create a error from core error
pub fn from_core(error: revolt_result::Error) -> Error {
Error::Core { error }
}
}
/// Result type with custom Error
@@ -145,6 +156,7 @@ impl<'r> Responder<'r, 'static> for Error {
fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
let status = match self {
Error::LabelMe => Status::InternalServerError,
Error::Core { .. } => Status::InternalServerError,
Error::AlreadyOnboarded => Status::Forbidden,