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

feat: search by message components

Browse Source 
Signed-off-by: Zomatree <me@zomatree.live>
This commit is contained in:
Zomatree
2026-03-20 15:24:56 +00:00
parent 1745826d2f
commit 19df5c5ef5
14 changed files with 144 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -79,12 +79,12 @@ pub struct AckPayload {
pub message_id: String,
}
#[derive(Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageDeletePayload {
pub message_id: String
}
#[derive(Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChannelDeletePayload {
pub channel_id: String
}

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

@@ -890,19 +890,27 @@ impl Message {
/// Append content to message
pub async fn append(
db: &Database,
amqp: Option<&AMQP>,
id: String,
channel: String,
append: AppendMessage,
) -> Result<()> {
db.append_message(&id, &append).await?;
if let Some(message) = db.append_message(&id, &append).await? {
if let Some(amqp) = amqp {
if let Err(e) = amqp.edit_message_search(message).await {
log::error!("Error pushing message to RabbitMQ: {e}");
capture_error(&e);
}
}
EventV1::MessageAppend {
id,
channel: channel.to_string(),
append: append.into(),
EventV1::MessageAppend {
id,
channel: channel.to_string(),
append: append.into(),
}
.p(channel)
.await;
}
.p(channel)
.await;
Ok(())
}

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

@@ -24,7 +24,7 @@ pub trait AbstractMessages: Sync + Send {
async fn update_message(&self, id: &str, message: &PartialMessage, remove: Vec<FieldsMessage>) -> Result<()>;
/// Append information to a given message
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()>;
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>>;
/// Add a new reaction to a message
async fn add_reaction(&self, id: &str, emoji: &str, user: &str) -> Result<()>;

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

@@ -1,6 +1,6 @@
use bson::{to_bson, Document};
use futures::try_join;
use mongodb::options::{FindOptions, ReadConcern};
use mongodb::options::{FindOptions, ReadConcern, ReturnDocument};
use revolt_models::v0::MessageSort;
use revolt_result::Result;
@@ -201,7 +201,7 @@ impl AbstractMessages for MongoDb {
}
/// Append information to a given message
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>> {
let mut query = doc! {};
if let Some(embeds) = &append.embeds {
@@ -219,18 +219,18 @@ impl AbstractMessages for MongoDb {
}
if query.is_empty() {
return Ok(());
return Ok(None);
}
self.col::<Document>(COL)
.update_one(
self.col::<Message>(COL)
.find_one_and_update(
doc! {
"_id": id
},
query,
)
.return_document(ReturnDocument::After)
.await
.map(|_| ())
.map_err(|_| create_database_error!("update_one", COL))
}

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

@@ -208,7 +208,7 @@ impl AbstractMessages for ReferenceDb {
}
/// Append information to a given message
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<()> {
async fn append_message(&self, id: &str, append: &AppendMessage) -> Result<Option<Message>> {
let mut messages = self.messages.lock().await;
if let Some(message_data) = messages.get_mut(id) {
if let Some(embeds) = &append.embeds {
@@ -219,9 +219,11 @@ impl AbstractMessages for ReferenceDb {
message_data.embeds = Some(embeds.clone());
}
}
}
Ok(())
Ok(Some(message_data.clone()))
} else {
Ok(None)
}
} else {
Err(create_error!(NotFound))
}

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

@@ -19,7 +19,7 @@ pub fn start_workers(db: Database, amqp: AMQP) {
for _ in 0..WORKER_COUNT {
task::spawn(ack::worker(db.clone(), amqp.clone()));
task::spawn(last_message_id::worker(db.clone()));
task::spawn(process_embeds::worker(db.clone()));
task::spawn(process_embeds::worker(db.clone(), amqp.clone()));
}
}

6
crates/core/database/src/tasks/process_embeds.rs
View File

@@ -1,4 +1,4 @@
use crate::{models::Message, AppendMessage, Database};
use crate::{AMQP, AppendMessage, Database, models::Message};
use futures::future::join_all;
use linkify::{LinkFinder, LinkKind};
@@ -41,7 +41,7 @@ pub async fn queue(channel: String, id: String, content: String) {
}
/// Start a new worker
pub async fn worker(db: Database) {
pub async fn worker(db: Database, amqp: AMQP) {
let semaphore = Arc::new(Semaphore::new(
config().await.api.workers.max_concurrent_connections,
));
@@ -49,6 +49,7 @@ pub async fn worker(db: Database) {
loop {
let task = Q.pop().await;
let db = db.clone();
let amqp = amqp.clone();
let semaphore = semaphore.clone();
spawn(async move {
@@ -64,6 +65,7 @@ pub async fn worker(db: Database) {
if let Ok(embeds) = embeds {
if let Err(err) = Message::append(
&db,
Some(&amqp),
task.id,
task.channel,
AppendMessage {

62
crates/core/search/src/client.rs
View File

@@ -10,12 +10,12 @@ use elasticsearch::{
indices::{IndicesCreateParts, IndicesDeleteParts},
};
use elasticsearch_dsl::{FieldSort, Query, Search, SearchResponse, Sort};
use revolt_database::Message;
use revolt_database::{Database, Message};
use serde_json::{Map, Value, json};
pub use elasticsearch;
use crate::SearchTerms;
use crate::{MessageComponent, MetadataFile, SearchTerms};
#[derive(Debug)]
pub enum Error {
@@ -87,12 +87,6 @@ impl ElasticsearchClient {
.indices()
.create(IndicesCreateParts::Index("messages"))
.body(json!({
// "settings": {
// "index": {
// "sort.field": "_id",
// "sort.order": ["asc", "desc"],
// },
// },
"mappings": {
"properties": {
"content": {"type": "text"},
@@ -105,9 +99,13 @@ impl ElasticsearchClient {
"embeds": {
"properties": {}
},
"attachments": { // TODO: images, videos, files
"attachments": {
"type": "nested",
"properties": {}
"properties": {
"metadata.type": {
"type": "keyword"
}
}
},
// TODO: links
}
@@ -152,15 +150,27 @@ impl ElasticsearchClient {
}
}
// TODO: component filtering
// Need to pass FileHash to this to extract metadata
if let Some(components) = terms.filters.components {
let mut components_query = Query::bool();
for component in components {
match component {
MessageComponent::Image => components_query = components_query.should(Query::term("attachments.metadata.type", "Image")),
MessageComponent::Video => components_query = components_query.should(Query::term("attachments.metadata.type", "Video")),
MessageComponent::File => components_query = components_query.should(Query::exists("attachments.file._id")),
MessageComponent::Embed => query = query.filter(Query::exists("embeds")),
};
}
query = query.filter(Query::nested("attachments", components_query));
}
let search = Search::new()
.query(query)
.stats(false)
.sort([Sort::FieldSort(
.sort(Sort::FieldSort(
FieldSort::new("_id".to_string()).order(terms.sort.unwrap_or_default().into()),
)]);
));
let response = self
.inner
@@ -184,7 +194,7 @@ impl ElasticsearchClient {
}
}
fn create_message_source(&self, message: Message) -> Value {
fn create_message_source(&self, _db: &Database, message: Message) -> Value {
let mut map = Map::new();
map.insert("channel".to_string(), Value::String(message.channel));
@@ -196,9 +206,21 @@ impl ElasticsearchClient {
}
if let Some(attachments) = message.attachments {
let mut files = Vec::new();
for attachment in attachments {
// TODO: swap this out for fetching the metadata from FileHash because of deprecation
// let metadata = attachment.as_hash(db).await.expect("Failed to fetch FileHash").metadata;
files.push(MetadataFile {
metadata: attachment.metadata.clone(),
file: attachment,
})
}
map.insert(
"attachments".to_string(),
serde_json::to_value(attachments).unwrap(),
serde_json::to_value(files).unwrap(),
);
}
@@ -240,9 +262,9 @@ impl ElasticsearchClient {
Value::Object(map)
}
pub async fn index_message(&self, message: Message) -> Result<(), Error> {
pub async fn index_message(&self, db: &Database, message: Message) -> Result<(), Error> {
let id = message.id.clone();
let source = self.create_message_source(message);
let source = self.create_message_source(db, message);
let exception = self
.inner
@@ -260,9 +282,9 @@ impl ElasticsearchClient {
}
}
pub async fn edit_message(&self, message: Message) -> Result<(), Error> {
pub async fn edit_message(&self, db: &Database, message: Message) -> Result<(), Error> {
let id = message.id.clone();
let source = self.create_message_source(message);
let source = self.create_message_source(db, message);
let exception = self
.inner

19
crates/core/search/src/types.rs
View File

@@ -1,15 +1,16 @@
use iso8601_timestamp::Timestamp;
use revolt_models::v0;
use serde::Serialize;
use serde::{Deserialize, Serialize};
use revolt_database::{File, Metadata};
#[derive(Copy, Clone, PartialEq, Eq, Serialize)]
#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize)]
pub enum AuthorType {
User,
// Bot,
Webhook,
}
#[derive(Copy, Clone, PartialEq, Eq)]
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum MessageComponent {
Image,
Video,
@@ -18,7 +19,7 @@ pub enum MessageComponent {
Embed,
}
#[derive(Clone, Default, PartialEq)]
#[derive(Debug, Clone, Default, PartialEq)]
pub struct SearchFilters {
pub content: Option<String>,
pub author: Option<Vec<String>>,
@@ -31,14 +32,14 @@ pub struct SearchFilters {
pub components: Option<Vec<MessageComponent>>,
}
#[derive(Clone, Copy, Default, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum SortOrder {
Asc,
#[default]
Desc,
}
#[derive(Clone, PartialEq)]
#[derive(Debug, Clone, PartialEq)]
pub struct SearchTerms {
pub channels: Vec<String>,
pub filters: SearchFilters,
@@ -47,6 +48,12 @@ pub struct SearchTerms {
pub sort: Option<SortOrder>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MetadataFile {
pub file: File,
pub metadata: Metadata
}
impl From<v0::AuthorType> for AuthorType {
fn from(value: v0::AuthorType) -> Self {
match value {