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

Merge branch 'master' into webhooks

Browse Source
This commit is contained in:
Angelo Kontaxis
2023-04-01 22:52:11 +01:00
committed by GitHub
parent 7b39f4e9d9 b7b70346b4
commit a60ae78176
71 changed files with 1173 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -57,6 +57,14 @@ pub async fn create_database(db: &MongoDb) {
.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.");
@@ -103,18 +111,18 @@ pub async fn create_database(db: &MongoDb) {
},
"name": "content"
},
{
"key": {
"channel": 1_i32
},
"name": "channel"
},
{
"key": {
"channel": 1_i32,
"_id": 1_i32
},
"name": "channel_id_compound"
},
{
"key": {
"author": 1_i32
},
"name": "author"
}
]
},

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

@@ -1,4 +1,4 @@
use std::time::Duration;
use std::{time::Duration, ops::BitXor};
use bson::{Bson, DateTime};
use futures::StreamExt;
@@ -16,7 +16,7 @@ struct MigrationInfo {
revision: i32,
}
pub const LATEST_REVISION: i32 = 18;
pub const LATEST_REVISION: i32 = 21;
pub async fn migrate_database(db: &MongoDb) {
let migrations = db.col::<Document>("migrations");
@@ -503,13 +503,8 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
update.insert(
"default_permissions",
(*DEFAULT_PERMISSION_SERVER
// Remove Send Message permission if it wasn't originally granted
^ (if has_send {
0
} else {
Permission::SendMessage as u64
})) as i64,
// 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") {
@@ -661,6 +656,103 @@ pub async fn run_migrations(db: &MongoDb, revision: i32) -> i32 {
.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.

90
crates/quark/src/impl/mongo/admin/stats.rs Normal file
View File

@@ -0,0 +1,90 @@
use std::collections::HashMap;
use bson::{from_document, Document};
use futures::StreamExt;
use crate::{
models::stats::{Index, Stats},
AbstractStats, Error, Result,
};
use super::super::MongoDb;
#[async_trait]
impl AbstractStats for MongoDb {
async fn generate_stats(&self) -> Result<Stats> {
let mut indices = HashMap::new();
let mut coll_stats = HashMap::new();
let collection_names =
self.db()
.list_collection_names(None)
.await
.map_err(|_| Error::DatabaseError {
operation: "list_collection_names",
with: "database",
})?;
for collection in collection_names {
indices.insert(
collection.to_string(),
self.col::<Document>(&collection)
.aggregate(
vec![doc! {
"$indexStats": { }
}],
None,
)
.await
.map_err(|_| Error::DatabaseError {
operation: "aggregate",
with: "col",
})?
.filter_map(|s| async { s.ok() })
.collect::<Vec<Document>>()
.await
.into_iter()
.filter_map(|doc| from_document(doc).ok())
.collect::<Vec<Index>>(),
);
coll_stats.insert(
collection.to_string(),
self.col::<Document>(&collection)
.aggregate(
vec![doc! {
"$collStats": {
"latencyStats": {
"histograms": true
},
"storageStats": {},
"count": {},
"queryExecStats": {}
}
}],
None,
)
.await
.map_err(|_| Error::DatabaseError {
operation: "aggregate",
with: "col",
})?
.filter_map(|s| async { s.ok() })
.collect::<Vec<Document>>()
.await
.into_iter()
.filter_map(|doc| from_document(doc).ok())
.next()
.ok_or(Error::DatabaseError {
operation: "next aggregation",
with: "col",
})?,
);
}
Ok(Stats {
indices,
coll_stats,
})
}
}

255
crates/quark/src/impl/mongo/channels/message.rs
View File

@@ -2,7 +2,9 @@ use bson::{to_bson, Document};
use futures::try_join;
use mongodb::options::FindOptions;
use crate::models::message::{AppendMessage, Message, MessageSort, PartialMessage};
use crate::models::message::{
AppendMessage, Message, MessageQuery, MessageSort, MessageTimePeriod, PartialMessage,
};
use crate::r#impl::mongo::DocumentId;
use crate::{AbstractMessage, Error, Result};
@@ -139,145 +141,138 @@ impl AbstractMessage for MongoDb {
.await
}
async fn fetch_messages(
&self,
channel: &str,
limit: Option<i64>,
before: Option<String>,
after: Option<String>,
sort: Option<MessageSort>,
nearby: Option<String>,
) -> Result<Vec<Message>> {
let limit = limit.unwrap_or(50);
Ok(if let Some(nearby) = nearby {
let (a, b) = try_join!(
self.find_with_options::<_, Message>(
COL,
async fn fetch_messages(&self, query: MessageQuery) -> Result<Vec<Message>> {
let mut filter = doc! {};
// 1. Apply message filters
if let Some(channel) = query.filter.channel {
filter.insert("channel", channel);
}
if let Some(author) = query.filter.author {
filter.insert("author", author);
}
let is_search_query = if let Some(query) = query.filter.query {
filter.insert(
"$text",
doc! {
"$search": query
},
);
true
} else {
false
};
// 2. Find query limit
let limit = query.limit.unwrap_or(50);
// 3. Apply message time period
match query.time_period {
MessageTimePeriod::Relative { nearby } => {
// 3.1. Prepare filters
let mut older_message_filter = filter.clone();
let mut newer_message_filter = filter;
older_message_filter.insert(
"_id",
doc! {
"channel": channel,
"_id": {
"$gte": &nearby
}
"$lt": &nearby
},
FindOptions::builder()
.limit(limit / 2 + 1)
.sort(doc! {
"_id": 1_i32
})
.build(),
),
self.find_with_options::<_, Message>(
COL,
);
newer_message_filter.insert(
"_id",
doc! {
"channel": channel,
"_id": {
"$lt": &nearby
}
"$gte": &nearby
},
);
// 3.2. Execute in both directions
let (a, b) = try_join!(
self.find_with_options::<_, Message>(
COL,
newer_message_filter,
FindOptions::builder()
.limit(limit / 2 + 1)
.sort(doc! {
"_id": 1_i32
})
.build(),
),
self.find_with_options::<_, Message>(
COL,
older_message_filter,
FindOptions::builder()
.limit(limit / 2)
.sort(doc! {
"_id": -1_i32
})
.build(),
)
)?;
Ok([a, b].concat())
}
MessageTimePeriod::Absolute {
before,
after,
sort,
} => {
// 3.1. Apply message ID filter
if let Some(doc) = match (before, after) {
(Some(before), Some(after)) => Some(doc! {
"$lt": before,
"$gt": after
}),
(Some(before), _) => Some(doc! {
"$lt": before
}),
(_, Some(after)) => Some(doc! {
"$gt": after
}),
_ => None,
} {
filter.insert("_id", doc);
}
// 3.2. Execute with given message sort
self.find_with_options(
COL,
filter,
FindOptions::builder()
.limit(limit / 2)
.sort(doc! {
"_id": -1_i32
.limit(limit)
.sort(match sort.unwrap_or(MessageSort::Latest) {
// Sort by relevance, fallback to latest
MessageSort::Relevance => {
if is_search_query {
doc! {
"score": {
"$meta": "textScore"
}
}
} else {
doc! {
"_id": -1_i32
}
}
}
// Sort by latest first
MessageSort::Latest => doc! {
"_id": -1_i32
},
// Sort by oldest first
MessageSort::Oldest => doc! {
"_id": 1_i32
},
})
.build(),
)
)?;
[a, b].concat()
} else {
let mut query = doc! { "channel": channel };
if let Some(before) = before {
query.insert("_id", doc! { "$lt": before });
.await
}
if let Some(after) = after {
query.insert("_id", doc! { "$gt": after });
}
let sort: i32 = if let MessageSort::Latest = sort.unwrap_or(MessageSort::Latest) {
-1
} else {
1
};
self.find_with_options::<_, Message>(
COL,
query,
FindOptions::builder()
.limit(limit)
.sort(doc! {
"_id": sort
})
.build(),
)
.await?
})
}
async fn search_messages(
&self,
channel: &str,
query: &str,
limit: Option<i64>,
before: Option<String>,
after: Option<String>,
sort: MessageSort,
) -> Result<Vec<Message>> {
let limit = limit.unwrap_or(50);
let mut filter = doc! {
"channel": channel,
"$text": {
"$search": query
}
};
if let Some(doc) = match (before, after) {
(Some(before), Some(after)) => Some(doc! {
"lt": before,
"gt": after
}),
(Some(before), _) => Some(doc! {
"lt": before
}),
(_, Some(after)) => Some(doc! {
"gt": after
}),
_ => None,
} {
filter.insert("_id", doc);
}
self.find_with_options(
COL,
filter,
FindOptions::builder()
.projection(if let MessageSort::Relevance = &sort {
doc! {
"score": {
"$meta": "textScore"
}
}
} else {
doc! {}
})
.limit(limit)
.sort(match &sort {
MessageSort::Relevance => doc! {
"score": {
"$meta": "textScore"
}
},
MessageSort::Latest => doc! {
"_id": -1_i32
},
MessageSort::Oldest => doc! {
"_id": 1_i32
},
})
.build(),
)
.await
}
/// Add a new reaction to a message

2
crates/quark/src/impl/mongo/media/attachment.rs
View File

@@ -37,7 +37,7 @@ impl AbstractAttachment for MongoDb {
parent_type: &str,
parent_id: &str,
) -> Result<File> {
let key = format!("{}_id", parent_type);
let key = format!("{parent_type}_id");
match self
.find_one::<File>(
COL,

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

@@ -13,6 +13,7 @@ use crate::{util::manipulation::prefix_keys, AbstractDatabase, Error, Result};
pub mod admin {
pub mod migrations;
pub mod stats;
}
pub mod media {
@@ -90,17 +91,25 @@ impl MongoDb {
where
O: Into<Option<FindOptions>>,
{
Ok(self
.col::<T>(collection)
.find(projection, options)
.await
.map_err(|_| Error::DatabaseError {
let result = self.col::<T>(collection).find(projection, options).await;
Ok(if cfg!(debug_assertions) {
result.unwrap()
} else {
result.map_err(|_| Error::DatabaseError {
operation: "find",
with: collection,
})?
.filter_map(|s| async { s.ok() })
.collect::<Vec<T>>()
.await)
}
.filter_map(|s| async {
if cfg!(debug_assertions) {
// Hard fail on invalid documents
Some(s.unwrap())
} else {
s.ok()
}
})
.collect::<Vec<T>>()
.await)
}
async fn find<T: DeserializeOwned + Unpin + Send + Sync>(

15
crates/quark/src/impl/mongo/safety/report.rs
View File

@@ -1,3 +1,4 @@
use crate::models::report::PartialReport;
use crate::models::Report;
use crate::{AbstractReport, Result};
@@ -10,4 +11,18 @@ impl AbstractReport for MongoDb {
async fn insert_report(&self, report: &Report) -> Result<()> {
self.insert_one(COL, report).await.map(|_| ())
}
async fn update_report(&self, id: &str, report: &PartialReport) -> Result<()> {
self.update_one_by_id(COL, id, report, vec![], None)
.await
.map(|_| ())
}
async fn fetch_report(&self, report_id: &str) -> Result<Report> {
self.find_one_by_id(COL, report_id).await
}
async fn fetch_reports(&self) -> Result<Vec<Report>> {
self.find(COL, doc! {}).await
}
}

10
crates/quark/src/impl/mongo/safety/snapshot.rs
View File

@@ -10,4 +10,14 @@ impl AbstractSnapshot for MongoDb {
async fn insert_snapshot(&self, snapshot: &Snapshot) -> Result<()> {
self.insert_one(COL, snapshot).await.map(|_| ())
}
async fn fetch_snapshot(&self, report_id: &str) -> Result<Snapshot> {
self.find_one(
COL,
doc! {
"report_id": report_id
},
)
.await
}
}

21
crates/quark/src/impl/mongo/users/user.rs
View File

@@ -1,6 +1,7 @@
use bson::Document;
use futures::StreamExt;
use mongodb::options::{Collation, CollationStrength, FindOneOptions, FindOptions};
use once_cell::sync::Lazy;
use crate::models::user::{FieldsUser, PartialUser, RelationshipStatus, User};
use crate::r#impl::mongo::IntoDocumentPath;
@@ -8,16 +9,14 @@ use crate::{AbstractUser, Error, Result};
use super::super::MongoDb;
lazy_static! {
static ref FIND_USERNAME_OPTIONS: FindOneOptions = FindOneOptions::builder()
.collation(
Collation::builder()
.locale("en")
.strength(CollationStrength::Secondary)
.build()
)
.build();
}
static FIND_USERNAME_OPTIONS: Lazy<FindOneOptions> = Lazy::new(|| FindOneOptions::builder()
.collation(
Collation::builder()
.locale("en")
.strength(CollationStrength::Secondary)
.build()
)
.build());
static COL: &str = "users";
@@ -265,7 +264,7 @@ impl AbstractUser for MongoDb {
[
{
"_id": target_id,
"status": format!("{:?}", relationship)
"status": format!("{relationship:?}")
}
]
]