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

Merge branch 'master' into feat/embed_improvements

Browse Source
This commit is contained in:
Paul Makles
2021-09-14 20:54:21 +01:00
committed by GitHub
parent 6a29c7cf03 4b8205b75b
commit 536e335b25
37 changed files with 1049 additions and 882 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/database/entities/channel.rs
View File

@@ -13,14 +13,6 @@ use mongodb::{
use rocket::serde::json::Value;
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct LastMessage {
#[serde(rename = "_id")]
id: String,
author: String,
short: String,
}
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(tag = "channel_type")]
pub enum Channel {
@@ -36,7 +28,7 @@ pub enum Channel {
active: bool,
recipients: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
last_message: Option<LastMessage>,
last_message_id: Option<String>,
},
Group {
#[serde(rename = "_id")]
@@ -53,7 +45,7 @@ pub enum Channel {
#[serde(skip_serializing_if = "Option::is_none")]
icon: Option<File>,
#[serde(skip_serializing_if = "Option::is_none")]
last_message: Option<LastMessage>,
last_message_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
permissions: Option<i32>,
@@ -75,7 +67,7 @@ pub enum Channel {
#[serde(skip_serializing_if = "Option::is_none")]
icon: Option<File>,
#[serde(skip_serializing_if = "Option::is_none")]
last_message: Option<String>,
last_message_id: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
default_permissions: Option<i32>,

181
src/database/entities/message.rs
View File

@@ -8,15 +8,13 @@ use crate::{
use futures::StreamExt;
use mongodb::options::UpdateOptions;
use mongodb::{
bson::{doc, to_bson, DateTime},
options::FindOptions,
bson::{doc, to_bson, DateTime, Document},
};
use rauth::entities::{Model, Session};
use rocket::serde::json::Value;
use serde::{Deserialize, Serialize};
use ulid::Ulid;
use web_push::{
ContentEncoding, SubscriptionInfo, VapidSignatureBuilder, WebPushClient, WebPushMessageBuilder,
};
use web_push::{ContentEncoding, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder, WebPushClient, WebPushMessageBuilder};
use std::time::SystemTime;
#[derive(Serialize, Deserialize, Debug, Clone)]
@@ -173,7 +171,6 @@ impl Message {
nonce: None,
channel,
author,
content,
attachments: None,
edited: None,
@@ -198,76 +195,19 @@ impl Message {
let ss = self.clone();
let c_clone = channel.clone();
async_std::task::spawn(async move {
let mut set = if let Content::Text(text) = &ss.content {
doc! {
"last_message": {
"_id": ss.id.clone(),
"author": ss.author.clone(),
"short": text.chars().take(128).collect::<String>()
}
}
} else {
doc! {}
};
// ! MARK AS ACTIVE
// ! FIXME: temp code
let last_message_id = ss.id.clone();
let mut set = doc! { "last_message_id": last_message_id };
let channels = get_collection("channels");
match &c_clone {
Channel::DirectMessage { id, .. } => {
// ! MARK AS ACTIVE
set.insert("active", true);
channels
.update_one(
doc! { "_id": id },
doc! {
"$set": set
},
None,
)
.await
/*.map_err(|_| Error::DatabaseError {
operation: "update_one",
with: "channel",
})?;*/
.unwrap();
}
Channel::Group { id, .. } => {
if let Content::Text(_) = &ss.content {
channels
.update_one(
doc! { "_id": id },
doc! {
"$set": set
},
None,
)
.await
/*.map_err(|_| Error::DatabaseError {
operation: "update_one",
with: "channel",
})?;*/
.unwrap();
}
}
Channel::TextChannel { id, .. } => {
if let Content::Text(_) = &ss.content {
channels
.update_one(
doc! { "_id": id },
doc! {
"$set": {
"last_message": &ss.id
}
},
None,
)
.await
/*.map_err(|_| Error::DatabaseError {
operation: "update_one",
with: "channel",
})?;*/
.unwrap();
}
update_channels_last_message(&channels, id, &set).await;
},
Channel::Group { id, .. } | Channel::TextChannel { id, .. } => {
update_channels_last_message(&channels, id, &set).await;
}
_ => {}
}
@@ -335,60 +275,47 @@ impl Message {
}
// Fetch their corresponding sessions.
if let Ok(mut cursor) = get_collection("accounts")
.find(
doc! {
"_id": {
"$in": target_ids
},
"sessions.subscription": {
"$exists": true
}
},
FindOptions::builder()
.projection(doc! { "sessions": 1 })
.build(),
)
.await
{
let mut subscriptions = vec![];
while let Some(result) = cursor.next().await {
if let Ok(doc) = result {
if let Ok(sessions) = doc.get_array("sessions") {
for session in sessions {
if let Some(doc) = session.as_document() {
if let Ok(sub) = doc.get_document("subscription") {
let endpoint = sub.get_str("endpoint").unwrap().to_string();
let p256dh = sub.get_str("p256dh").unwrap().to_string();
let auth = sub.get_str("auth").unwrap().to_string();
subscriptions
.push(SubscriptionInfo::new(endpoint, p256dh, auth));
}
}
if target_ids.len() > 0 {
if let Ok(mut cursor) = Session::find(
&get_db(),
doc! {
"_id": {
"$in": target_ids
},
"subscription": {
"$exists": true
}
}
}
}
if subscriptions.len() > 0 {
},
None
)
.await {
let enc = serde_json::to_string(&PushNotification::new(self, &c_clone).await).unwrap();
let client = WebPushClient::new();
let key =
base64::decode_config(VAPID_PRIVATE_KEY.clone(), base64::URL_SAFE).unwrap();
for subscription in subscriptions {
let mut builder = WebPushMessageBuilder::new(&subscription).unwrap();
let sig_builder = VapidSignatureBuilder::from_pem(
std::io::Cursor::new(&key),
&subscription,
)
.unwrap();
let signature = sig_builder.build().unwrap();
builder.set_vapid_signature(signature);
builder.set_payload(ContentEncoding::AesGcm, enc.as_bytes());
let m = builder.build().unwrap();
client.send(m).await.ok();
while let Some(Ok(session)) = cursor.next().await {
if let Some(sub) = session.subscription {
let subscription = SubscriptionInfo {
endpoint: sub.endpoint,
keys: SubscriptionKeys {
auth: sub.auth,
p256dh: sub.p256dh
}
};
let mut builder = WebPushMessageBuilder::new(&subscription).unwrap();
let sig_builder = VapidSignatureBuilder::from_pem(
std::io::Cursor::new(&key),
&subscription,
)
.unwrap();
let signature = sig_builder.build().unwrap();
builder.set_vapid_signature(signature);
builder.set_payload(ContentEncoding::AesGcm, enc.as_bytes());
let m = builder.build().unwrap();
client.send(m).await.ok();
}
}
}
}
@@ -482,7 +409,7 @@ impl Message {
let attachment_ids: Vec<String> =
attachments.iter().map(|f| f.id.to_string()).collect();
get_collection("attachments")
.update_one(
.update_many(
doc! {
"_id": {
"$in": attachment_ids
@@ -497,7 +424,7 @@ impl Message {
)
.await
.map_err(|_| Error::DatabaseError {
operation: "update_one",
operation: "update_many",
with: "attachment",
})?;
}
@@ -505,3 +432,15 @@ impl Message {
Ok(())
}
}
async fn update_channels_last_message(channels: &Collection, channel_id: &String, set: &Document) {
channels
.update_one(
doc! { "_id": channel_id },
doc! { "$set": set },
None,
)
.await
.expect("Server should not run with no, or a corrupted db");
}

3
src/database/entities/microservice/january.rs
View File

@@ -47,6 +47,9 @@ pub enum Special {
None,
YouTube {
id: String,
#[serde(skip_serializing_if = "Option::is_none")]
timestamp: Option<String>,
},
Twitch {
content_type: TwitchType,

5
src/database/entities/server.rs
View File

@@ -92,7 +92,7 @@ pub struct Server {
pub name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
pub channels: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub categories: Option<Vec<Category>>,
@@ -108,6 +108,9 @@ pub struct Server {
#[serde(skip_serializing_if = "Option::is_none")]
pub banner: Option<File>,
#[serde(skip_serializing_if = "Option::is_none")]
pub flags: Option<i32>,
#[serde(skip_serializing_if = "if_false", default)]
pub nsfw: bool
}

7
src/database/entities/user.rs
View File

@@ -16,6 +16,7 @@ use crate::database::*;
use crate::notifications::websocket::is_online;
use crate::util::result::{Error, Result};
use crate::util::variables::EARLY_ADOPTER_BADGE;
use crate::util::variables::MAX_SERVER_COUNT;
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
pub enum RelationshipStatus {
@@ -323,4 +324,10 @@ impl User {
.collect::<Vec<Document>>()
.await)
}
/// Check if this user can acquire another server.
pub async fn can_acquire_server(id: &str) -> Result<bool> {
let server_ids = User::fetch_server_ids(&id).await?;
Ok(server_ids.len() < *MAX_SERVER_COUNT)
}
}

55
src/database/guards/user.rs
View File

@@ -1,7 +1,7 @@
use crate::database::*;
use mongodb::bson::{doc, from_document};
use rauth::auth::Session;
use rauth::entities::Session;
use rocket::http::Status;
use rocket::request::{self, FromRequest, Outcome, Request};
@@ -15,7 +15,7 @@ impl<'r> FromRequest<'r> for User {
.get("x-bot-token")
.next()
.map(|x| x.to_string());
if let Some(bot_token) = header_bot_token {
return if let Ok(result) = get_collection("bots")
.find_one(
@@ -40,7 +40,10 @@ impl<'r> FromRequest<'r> for User {
if let Some(doc) = result {
Outcome::Success(from_document(doc).unwrap())
} else {
Outcome::Failure((Status::Forbidden, rauth::util::Error::InvalidSession))
Outcome::Failure((
Status::Forbidden,
rauth::util::Error::InvalidSession,
))
}
} else {
Outcome::Failure((
@@ -62,32 +65,34 @@ impl<'r> FromRequest<'r> for User {
with: "bot",
},
))
}
};
} else {
let session: Session = request.guard::<Session>().await.unwrap();
if let Ok(result) = get_collection("users")
.find_one(
doc! {
"_id": &session.user_id
},
None,
)
.await
{
if let Some(doc) = result {
Outcome::Success(from_document(doc).unwrap())
if let Outcome::Success(session) = request.guard::<Session>().await {
if let Ok(result) = get_collection("users")
.find_one(
doc! {
"_id": &session.user_id
},
None,
)
.await
{
if let Some(doc) = result {
Outcome::Success(from_document(doc).unwrap())
} else {
Outcome::Failure((Status::Forbidden, rauth::util::Error::InvalidSession))
}
} else {
Outcome::Failure((Status::Forbidden, rauth::util::Error::InvalidSession))
Outcome::Failure((
Status::InternalServerError,
rauth::util::Error::DatabaseError {
operation: "find_one",
with: "user",
},
))
}
} else {
Outcome::Failure((
Status::InternalServerError,
rauth::util::Error::DatabaseError {
operation: "find_one",
with: "user",
},
))
Outcome::Failure((Status::Forbidden, rauth::util::Error::InvalidSession))
}
}
}

33
src/database/migrations/init.rs
View File

@@ -71,39 +71,6 @@ pub async fn create_database() {
.await
.expect("Failed to create pubsub collection.");
db.run_command(
doc! {
"createIndexes": "accounts",
"indexes": [
{
"key": {
"email": 1
},
"name": "email",
"unique": true,
"collation": {
"locale": "en",
"strength": 2
}
},
{
"key": {
"email_normalised": 1
},
"name": "email_normalised",
"unique": true,
"collation": {
"locale": "en",
"strength": 2
}
}
]
},
None,
)
.await
.expect("Failed to create account index.");
db.run_command(
doc! {
"createIndexes": "users",

5
src/database/migrations/mod.rs
View File

@@ -10,10 +10,13 @@ pub async fn run_migrations() {
.list_database_names(None, None)
.await
.expect("Failed to fetch database names.");
if list.iter().position(|x| x == "revolt").is_none() {
init::create_database().await;
} else {
scripts::migrate_database().await;
}
// panic!("https://pbs.twimg.com/media/EDTpB5JWwAUvyxd.jpg");
rauth::entities::sync_models(&super::get_db()).await;
}

136
src/database/migrations/scripts.rs
View File

@@ -2,7 +2,7 @@ use crate::database::{permissions, get_collection, get_db, PermissionTuple};
use futures::StreamExt;
use log::info;
use mongodb::{bson::{doc, from_document, to_document}, options::FindOptions};
use mongodb::{bson::{Document, doc, from_bson, from_document, to_document}, options::FindOptions};
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize)]
@@ -11,7 +11,7 @@ struct MigrationInfo {
revision: i32,
}
pub const LATEST_REVISION: i32 = 8;
pub const LATEST_REVISION: i32 = 10;
pub async fn migrate_database() {
let migrations = get_collection("migrations");
@@ -212,6 +212,138 @@ pub async fn run_migrations(revision: i32) -> i32 {
.expect("Failed to create bots collection.");
}
if revision <= 8 {
info!("Running migration [revision 8 / 2021-09-10]: Update to rAuth version 1.");
get_db()
.run_command(
doc! {
"dropIndexes": "accounts",
"index": ["email", "email_normalised"]
},
None,
)
.await
.expect("Failed to delete legacy account indexes.");
let col = get_collection("sessions");
let mut cursor = get_collection("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.clone(),
"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!");
}
}
}
get_collection("accounts")
.update_many(
doc! { },
doc! {
"$unset": {
"sessions": 1,
},
"$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 = get_collection("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);
get_collection("channels")
.update_one(
doc! {
"_id": channel_id
},
doc! {
"$set": {
"last_message_id": id
},
"$unset": {
"last_message": 1,
}
},
None
)
.await
.unwrap();
} else {
info!("{} has no last_message.", &channel_id);
}
}
}
}
// 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
}