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

feat: use bulk indexing for existing messages

Browse Source 
Signed-off-by: Zomatree <me@zomatree.live>
This commit is contained in:
Zomatree
2026-03-21 06:59:56 +00:00
parent 3fd170a7de
commit fb487130c8
3 changed files with 58 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
crates/daemons/searchd/src/index.rs
View File

@@ -1,14 +1,36 @@
use revolt_config::capture_error;
use revolt_database::{AMQP, Database};
use revolt_database::Database;
use revolt_search::ElasticsearchClient;
pub async fn index_existing_messages(db: Database, client: ElasticsearchClient) {
log::info!("Starting bulk indexing.");
pub async fn index_existing_messages(db: Database, amqp: AMQP) {
let mut generator = db.fetch_all_messages().await.expect("Database query failed");
let mut generator = db
.fetch_all_messages()
.await
.expect("Database query failed");
let mut chunk = Vec::new();
while let Some(message) = generator.next().await {
if let Err(e) = amqp.new_message_search(message).await {
log::error!("Error pushing message to RabbitMQ: {e}");
chunk.push(message);
if chunk.len() >= 1000
&& let Err(e) = client
.bulk_index_messages(&db, std::mem::take(&mut chunk))
.await
{
log::error!("Error bulk indexing messages: {e}");
capture_error(&e);
}
}
}
if !chunk.is_empty()
&& let Err(e) = client.bulk_index_messages(&db, chunk).await
{
log::error!("Error bulk indexing messages: {e}");
capture_error(&e);
}
log::info!("Finished bulk indexing.")
}

5
crates/daemons/searchd/src/main.rs
View File

@@ -7,7 +7,7 @@ use amqprs::{
consumer::AsyncConsumer,
};
use revolt_config::{Settings, config, configure};
use revolt_database::{AMQP, DatabaseInfo};
use revolt_database::DatabaseInfo;
use revolt_search::ElasticsearchClient;
use tokio::{signal::ctrl_c, spawn};
@@ -88,8 +88,7 @@ async fn _main() {
let mut task = None;
if std::env::var("INDEX_ALL_MESSAGES").as_deref().is_ok_and(|v| v == "1") {
let amqp = AMQP::new_auto().await;
task = Some(spawn(index::index_existing_messages(db, amqp)));
task = Some(spawn(index::index_existing_messages(db, client.clone())));
}
ctrl_c().await.expect("Failed to wait for ctrl-c");