refactor: minimum implementation of revolt-models and revolt-database

crates/core/database/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "revolt-database"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[features]
+mongodb = [ "dep:mongodb" ]
+default = [ "mongodb" ]
+
+[dependencies]
+# Serialisation
+serde_json = "1"
+serde = { version = "1", features = ["derive"] }
+
+# Database
+mongodb = { optional = true, version = "2.1.0", default-features = false }
+
+# Async Language Features
+futures = "0.3.19"
+async-recursion = "1.0.4"

crates/core/database/src/drivers/dummy.rs

@@ -0,0 +1,4 @@
+database_derived!(
+    /// Dummy implementation
+    pub struct DummyDb {}
+);

crates/core/database/src/drivers/mod.rs

@@ -0,0 +1,5 @@
+mod dummy;
+mod mongodb;
+
+pub use self::dummy::*;
+pub use self::mongodb::*;

crates/core/database/src/drivers/mongodb.rs

@@ -0,0 +1,228 @@
+use std::collections::HashMap;
+use std::ops::Deref;
+
+use futures::StreamExt;
+use mongodb::bson::{doc, to_document, Document};
+use mongodb::error::Result;
+use mongodb::options::{FindOneOptions, FindOptions};
+use mongodb::results::{DeleteResult, InsertOneResult, UpdateResult};
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+
+database_derived!(
+    #[cfg(feature = "mongodb")]
+    /// MongoDB implementation
+    pub struct MongoDb(pub ::mongodb::Client);
+);
+
+impl Deref for MongoDb {
+    type Target = mongodb::Client;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+#[allow(dead_code)]
+impl MongoDb {
+    /// Get the Revolt database
+    pub fn db(&self) -> mongodb::Database {
+        self.database("revolt")
+    }
+
+    /// Get a collection by its name
+    pub fn col<T>(&self, collection: &str) -> mongodb::Collection<T> {
+        self.db().collection(collection)
+    }
+
+    /// Insert one document into a collection
+    async fn insert_one<T: Serialize>(
+        &self,
+        collection: &'static str,
+        document: T,
+    ) -> Result<InsertOneResult> {
+        self.col::<T>(collection).insert_one(document, None).await
+    }
+
+    /// Find multiple documents in a collection with options
+    async fn find_with_options<O, T: DeserializeOwned + Unpin + Send + Sync>(
+        &self,
+        collection: &'static str,
+        projection: Document,
+        options: O,
+    ) -> Result<Vec<T>>
+    where
+        O: Into<Option<FindOptions>>,
+    {
+        Ok(self
+            .col::<T>(collection)
+            .find(projection, options)
+            .await?
+            .filter_map(|s| async {
+                if cfg!(debug_assertions) {
+                    // Hard fail on invalid documents
+                    Some(s.unwrap())
+                } else {
+                    s.ok()
+                }
+            })
+            .collect::<Vec<T>>()
+            .await)
+    }
+
+    /// Find multiple documents in a collection
+    async fn find<T: DeserializeOwned + Unpin + Send + Sync>(
+        &self,
+        collection: &'static str,
+        projection: Document,
+    ) -> Result<Vec<T>> {
+        self.find_with_options(collection, projection, None).await
+    }
+
+    /// Find one document with options
+    async fn find_one_with_options<O, T: DeserializeOwned + Unpin + Send + Sync>(
+        &self,
+        collection: &'static str,
+        projection: Document,
+        options: O,
+    ) -> Result<Option<T>>
+    where
+        O: Into<Option<FindOneOptions>>,
+    {
+        self.col::<T>(collection)
+            .find_one(projection, options)
+            .await
+    }
+
+    /// Find one document
+    async fn find_one<T: DeserializeOwned + Unpin + Send + Sync>(
+        &self,
+        collection: &'static str,
+        projection: Document,
+    ) -> Result<Option<T>> {
+        self.find_one_with_options(collection, projection, None)
+            .await
+    }
+
+    /// Find one document by its ID
+    async fn find_one_by_id<T: DeserializeOwned + Unpin + Send + Sync>(
+        &self,
+        collection: &'static str,
+        id: &str,
+    ) -> Result<Option<T>> {
+        self.find_one(
+            collection,
+            doc! {
+                "_id": id
+            },
+        )
+        .await
+    }
+
+    /// Update one document given a projection, partial document, and list of paths to unset
+    async fn update_one<P, T: Serialize>(
+        &self,
+        collection: &'static str,
+        projection: Document,
+        partial: T,
+        remove: Vec<&dyn IntoDocumentPath>,
+        prefix: P,
+    ) -> Result<UpdateResult>
+    where
+        P: Into<Option<String>>,
+    {
+        let prefix = prefix.into();
+
+        let mut unset = doc! {};
+        for field in remove {
+            if let Some(path) = field.as_path() {
+                if let Some(prefix) = &prefix {
+                    unset.insert(prefix.to_owned() + path, 1_i32);
+                } else {
+                    unset.insert(path, 1_i32);
+                }
+            }
+        }
+
+        let query = doc! {
+            "$unset": unset,
+            "$set": if let Some(prefix) = &prefix {
+                to_document(&prefix_keys(&partial, prefix))
+            } else {
+                to_document(&partial)
+            }?
+        };
+
+        self.col::<Document>(collection)
+            .update_one(projection, query, None)
+            .await
+    }
+
+    /// Update one document given an ID, partial document, and list of paths to unset
+    async fn update_one_by_id<P, T: Serialize>(
+        &self,
+        collection: &'static str,
+        id: &str,
+        partial: T,
+        remove: Vec<&dyn IntoDocumentPath>,
+        prefix: P,
+    ) -> Result<UpdateResult>
+    where
+        P: Into<Option<String>>,
+    {
+        self.update_one(
+            collection,
+            doc! {
+                "_id": id
+            },
+            partial,
+            remove,
+            prefix,
+        )
+        .await
+    }
+
+    /// Delete one document by the given projection
+    async fn delete_one(
+        &self,
+        collection: &'static str,
+        projection: Document,
+    ) -> Result<DeleteResult> {
+        self.col::<Document>(collection)
+            .delete_one(projection, None)
+            .await
+    }
+
+    /// Delete one document by the given ID
+    async fn delete_one_by_id(&self, collection: &'static str, id: &str) -> Result<DeleteResult> {
+        self.delete_one(
+            collection,
+            doc! {
+                "_id": id
+            },
+        )
+        .await
+    }
+}
+
+/// Just a string ID struct
+#[derive(Deserialize)]
+pub struct DocumentId {
+    #[serde(rename = "_id")]
+    pub id: String,
+}
+
+pub trait IntoDocumentPath: Send + Sync {
+    /// Create JSON key path
+    fn as_path(&self) -> Option<&'static str>;
+}
+
+/// Prefix keys on an arbitrary object
+pub fn prefix_keys<T: Serialize>(t: &T, prefix: &str) -> HashMap<String, serde_json::Value> {
+    let v: String = serde_json::to_string(t).unwrap();
+    let v: HashMap<String, serde_json::Value> = serde_json::from_str(&v).unwrap();
+    v.into_iter()
+        .filter(|(_k, v)| !v.is_null())
+        .map(|(k, v)| (prefix.to_owned() + &k, v))
+        .collect()
+}

crates/core/database/src/lib.rs

@@ -0,0 +1,66 @@
+#[macro_use]
+extern crate serde;
+
+#[macro_use]
+extern crate async_recursion;
+
+macro_rules! database_derived {
+    ( $( $item:item )+ ) => {
+        $(
+            #[derive(Clone)]
+            $item
+        )+
+    };
+}
+
+#[cfg(feature = "mongodb")]
+pub use mongodb;
+
+mod drivers;
+pub use drivers::*;
+
+/// Database information to use to create a client
+pub enum DatabaseInfo {
+    /// Auto-detect the database in use
+    Auto,
+    /// Use the mock database
+    Dummy,
+    /// Connect to MongoDB
+    MongoDb(String),
+    /// Use existing MongoDB connection
+    MongoDbFromClient(::mongodb::Client),
+}
+
+/// Database
+#[derive(Clone)]
+pub enum Database {
+    /// Mock database
+    Dummy(DummyDb),
+    /// MongoDB database
+    MongoDb(MongoDb),
+}
+
+impl DatabaseInfo {
+    /// Create a database client from the given database information
+    #[async_recursion]
+    pub async fn connect(self) -> Result<Database, String> {
+        Ok(match self {
+            DatabaseInfo::Auto => {
+                if let Ok(uri) = std::env::var("MONGODB") {
+                    return DatabaseInfo::MongoDb(uri).connect().await;
+                }
+
+                DatabaseInfo::Dummy.connect().await?
+            }
+            DatabaseInfo::Dummy => Database::Dummy(DummyDb {}),
+            DatabaseInfo::MongoDb(uri) => {
+                let client = mongodb::Client::with_uri_str(uri)
+                    .await
+                    .map_err(|_| "Failed to init db connection.".to_string())?;
+
+                Database::MongoDb(MongoDb(client))
+            }
+            DatabaseInfo::MongoDbFromClient(client) => Database::MongoDb(MongoDb(client)),
+        })
+    }
+}

crates/core/models/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "revolt-models"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[features]
+serde = [ "dep:serde" ]
+async-std-runtime = [ "async-std" ]
+database = [ "revolt-database", "revolt-database/default", "authifier/database-mongodb" ]
+
+default = [ "serde", "async-std-runtime", "database" ]
+
+[dependencies]
+# Utility
+log = "*"
+
+# Serialisation
+serde = { version = "1", features = ["derive"], optional = true }
+
+# Async Language Features
+futures = "0.3.19"
+async-trait = "0.1.51"
+
+# Async
+async-std = { version = "1.8.0", features = ["attributes"], optional = true }
+
+# Peer dependencies
+revolt-database = { path = "../database", optional = true, default-features = false }
+
+# Authifier
+authifier = { version = "1.0", default-features = false }

crates/core/models/examples/migrate.rs

@@ -0,0 +1,8 @@
+use revolt_database::DatabaseInfo;
+use revolt_models::*;
+
+#[async_std::main]
+async fn main() {
+    let db = Database(DatabaseInfo::Auto.connect().await.unwrap());
+    db.migrate_database().await.unwrap();
+}

crates/core/models/src/admin_migrations/mod.rs

@@ -0,0 +1,9 @@
+mod model;
+
+pub use model::*;
+
+#[cfg(feature = "database")]
+mod ops;
+
+#[cfg(feature = "database")]
+pub use ops::*;

crates/core/models/src/admin_migrations/model.rs

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

crates/core/models/src/admin_migrations/ops/dummy.rs

@@ -0,0 +1,11 @@
+use revolt_database::DummyDb;
+
+use super::AbstractMigrations;
+
+#[async_trait]
+impl AbstractMigrations for DummyDb {
+    /// Migrate the database
+    async fn migrate_database(&self) -> Result<(), ()> {
+        Ok(())
+    }
+}

crates/core/models/src/admin_migrations/ops/mod.rs

@@ -0,0 +1,8 @@
+mod dummy;
+mod mongodb;
+
+#[async_trait]
+pub trait AbstractMigrations: Sync + Send {
+    /// Migrate the database
+    async fn migrate_database(&self) -> Result<(), ()>;
+}

crates/core/models/src/admin_migrations/ops/mongodb.rs

@@ -0,0 +1,27 @@
+use revolt_database::MongoDb;
+
+use super::AbstractMigrations;
+
+mod init;
+mod scripts;
+
+#[async_trait]
+impl AbstractMigrations for MongoDb {
+    /// Migrate the database
+    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(())
+    }
+}

crates/core/models/src/admin_migrations/ops/mongodb/init.rs

@@ -0,0 +1,193 @@
+use super::scripts::LATEST_REVISION;
+
+use revolt_database::mongodb::bson::doc;
+use revolt_database::mongodb::options::CreateCollectionOptions;
+use revolt_database::MongoDb;
+
+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.");
+}

crates/core/models/src/admin_migrations/ops/mongodb/scripts.rs

@@ -0,0 +1,760 @@
+use std::{ops::BitXor, time::Duration};
+
+use futures::StreamExt;
+use revolt_database::{
+    mongodb::{
+        bson::{doc, from_bson, from_document, to_document, Bson, DateTime, Document},
+        options::FindOptions,
+    },
+    MongoDb,
+};
+use serde::{Deserialize, Serialize};
+
+#[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
+                (4000323584).bitxor(if has_send { 0 } else { (1 << 22) 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": (1 << 22) 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
+}

crates/core/models/src/lib.rs

@@ -0,0 +1,40 @@
+#[cfg(feature = "serde")]
+#[macro_use]
+extern crate serde;
+
+#[macro_use]
+extern crate async_trait;
+
+#[macro_use]
+extern crate log;
+
+macro_rules! auto_derived {
+    ( $( $item:item )+ ) => {
+        $(
+            #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+            #[derive(Debug, Clone)]
+            $item
+        )+
+    };
+}
+
+mod admin_migrations;
+
+pub use admin_migrations::*;
+
+pub struct Database(pub revolt_database::Database);
+
+pub trait AbstractDatabase: Sync + Send + admin_migrations::AbstractMigrations {}
+impl AbstractDatabase for revolt_database::DummyDb {}
+impl AbstractDatabase for revolt_database::MongoDb {}
+
+impl std::ops::Deref for Database {
+    type Target = dyn AbstractDatabase;
+
+    fn deref(&self) -> &Self::Target {
+        match &self.0 {
+            revolt_database::Database::Dummy(dummy) => dummy,
+            revolt_database::Database::MongoDb(mongo) => mongo,
+        }
+    }
+}