commit 6bfd52b91fdf812ed1408238828bd8fa88196790
parent 1b6dd39f5fa80a2b93025d18b9b0ebb3aba18e44
Author: MTRNord <mtrnord1@gmail.com>
Date: Thu, 23 Mar 2023 18:59:38 +0100
feat(matrix-indexer): Insert edges between events and rooms
Diffstat:
5 files changed, 113 insertions(+), 17 deletions(-)
diff --git a/crates/knowledge-search/src/main.rs b/crates/knowledge-search/src/main.rs
@@ -1,4 +1,31 @@
+#![deny(unsafe_code, clippy::unwrap_used)]
+#![warn(
+ clippy::cognitive_complexity,
+ clippy::branches_sharing_code,
+ clippy::imprecise_flops,
+ clippy::missing_const_for_fn,
+ clippy::mutex_integer,
+ clippy::path_buf_push_overwrite,
+ clippy::redundant_pub_crate,
+ clippy::pedantic,
+ clippy::dbg_macro,
+ clippy::todo,
+ clippy::fallible_impl_from,
+ clippy::filetype_is_file,
+ clippy::suboptimal_flops,
+ clippy::fn_to_numeric_cast_any,
+ clippy::if_then_some_else_none,
+ clippy::imprecise_flops,
+ clippy::lossy_float_literal,
+ clippy::panic_in_result_fn,
+ clippy::clone_on_ref_ptr
+)]
+#![allow(clippy::missing_panics_doc)]
+// I am lazy. Dont blame me!
+#![allow(missing_docs)]
+
mod algos;
+
fn main() {
println!("Hello, world!");
}
diff --git a/crates/matrix-indexer/src/indradb_utils.rs b/crates/matrix-indexer/src/indradb_utils.rs
@@ -59,7 +59,7 @@ impl BulkInserter {
self.requests.send(self.buf).await.unwrap();
}
self.requests.close();
- for worker in self.workers.into_iter() {
+ for worker in self.workers {
worker.await.unwrap();
}
self.client.sync().await.unwrap();
diff --git a/crates/matrix-indexer/src/main.rs b/crates/matrix-indexer/src/main.rs
@@ -1,8 +1,33 @@
+#![deny(unsafe_code, clippy::unwrap_used)]
+#![warn(
+ clippy::cognitive_complexity,
+ clippy::branches_sharing_code,
+ clippy::imprecise_flops,
+ clippy::missing_const_for_fn,
+ clippy::mutex_integer,
+ clippy::path_buf_push_overwrite,
+ clippy::redundant_pub_crate,
+ clippy::pedantic,
+ clippy::dbg_macro,
+ clippy::todo,
+ clippy::fallible_impl_from,
+ clippy::filetype_is_file,
+ clippy::suboptimal_flops,
+ clippy::fn_to_numeric_cast_any,
+ clippy::if_then_some_else_none,
+ clippy::imprecise_flops,
+ clippy::lossy_float_literal,
+ clippy::panic_in_result_fn,
+ clippy::clone_on_ref_ptr
+)]
+#![allow(clippy::missing_panics_doc)]
+// I am lazy. Dont blame me!
+#![allow(missing_docs)]
+
use color_eyre::Result;
use clap::Parser;
use matrix::IndexerBot;
-use tracing::info;
mod indradb_utils;
mod matrix;
@@ -19,8 +44,7 @@ async fn main() -> Result<()> {
let _args = Args::parse();
// TODO: config which rewrites itself to have the data after login
- let mut bot =
- IndexerBot::new(String::new(), String::new(), String::new()).await?;
+ let mut bot = IndexerBot::new(String::new(), String::new(), String::new()).await?;
bot.start_processing().await;
Ok(())
diff --git a/crates/matrix-indexer/src/matrix.rs b/crates/matrix-indexer/src/matrix.rs
@@ -24,6 +24,7 @@ struct Identifiers {
text_message_event_type: utils::indradb::Identifier,
notice_message_event_type: utils::indradb::Identifier,
event_id_type: utils::indradb::Identifier,
+ event_in_room_type: utils::indradb::Identifier,
}
pub struct IndexerBot {
@@ -52,6 +53,7 @@ impl IndexerBot {
let text_message_event_type = utils::indradb::Identifier::new("text_message_event")?;
let notice_message_event_type = utils::indradb::Identifier::new("notice_message_event")?;
let event_id_type = utils::indradb::Identifier::new("event_id")?;
+ let event_in_room_type = utils::indradb::Identifier::new("event_in_room")?;
indexer_client.index_property(room_name_type).await?;
indexer_client.index_property(room_topic_type).await?;
indexer_client.index_property(event_id_type).await?;
@@ -78,6 +80,7 @@ impl IndexerBot {
text_message_event_type,
notice_message_event_type,
event_id_type,
+ event_in_room_type,
},
))
}
@@ -95,7 +98,10 @@ impl IndexerBot {
let client_clone = client.clone();
tokio::spawn(async move {
let settings = SyncSettings::default();
- client_clone.sync(settings).await.unwrap();
+ client_clone
+ .sync(settings)
+ .await
+ .expect("Failed to start matrix sync");
});
Ok(IndexerBot {
@@ -127,7 +133,10 @@ impl IndexerBot {
let client_clone = client.clone();
tokio::spawn(async move {
let settings = SyncSettings::default();
- client_clone.sync(settings).await.unwrap();
+ client_clone
+ .sync(settings)
+ .await
+ .expect("Failed to start matrix sync");
});
Ok(IndexerBot {
@@ -138,6 +147,8 @@ impl IndexerBot {
})
}
+ // FIXME:_split into multiple functions
+ #[allow(clippy::too_many_lines)]
pub async fn start_processing(&mut self) {
let mut inserter = BulkInserter::new(self.indexer_client.clone());
@@ -204,10 +215,11 @@ impl IndexerBot {
}
}
}
- Ok(AnySyncTimelineEvent::MessageLike(_)) => {}
- Ok(AnySyncTimelineEvent::State(_)) => {}
+ Ok(
+ AnySyncTimelineEvent::MessageLike(_) | AnySyncTimelineEvent::State(_),
+ ) => {}
Err(e) => {
- error!("Error deserializing event: {}", e)
+ error!("Error deserializing event: {}", e);
}
}
}
@@ -277,6 +289,18 @@ impl IndexerBot {
inserter.push(event_property).await;
}
}
+
+ for (event_uuid, room_uuid) in &self.message_map.room_event_links {
+ inserter
+ .push(utils::indradb::BulkInsertItem::Edge(
+ utils::indradb::Edge::new(
+ *event_uuid,
+ self.identifiers.event_in_room_type,
+ *room_uuid,
+ ),
+ ))
+ .await;
+ }
inserter.sync().await.expect("Unable to sync indradb");
}
}
diff --git a/crates/utils/src/lib.rs b/crates/utils/src/lib.rs
@@ -1,13 +1,35 @@
-#[cfg(rocksdb)]
-use std::path::Path;
+#![deny(unsafe_code, clippy::unwrap_used)]
+#![warn(
+ clippy::cognitive_complexity,
+ clippy::branches_sharing_code,
+ clippy::imprecise_flops,
+ clippy::missing_const_for_fn,
+ clippy::mutex_integer,
+ clippy::path_buf_push_overwrite,
+ clippy::redundant_pub_crate,
+ clippy::pedantic,
+ clippy::dbg_macro,
+ clippy::todo,
+ clippy::fallible_impl_from,
+ clippy::filetype_is_file,
+ clippy::suboptimal_flops,
+ clippy::fn_to_numeric_cast_any,
+ clippy::if_then_some_else_none,
+ clippy::imprecise_flops,
+ clippy::lossy_float_literal,
+ clippy::panic_in_result_fn,
+ clippy::clone_on_ref_ptr
+)]
+#![allow(clippy::missing_panics_doc)]
+// I am lazy. Dont blame me!
+#![allow(missing_docs)]
pub use indradb;
pub use indradb_proto;
use tokio::time::{sleep, Duration};
pub async fn get_client() -> Result<indradb_proto::Client, indradb_proto::ClientError> {
- let mut client =
- indradb_proto::Client::new("grpc://127.0.0.1:27615".try_into().unwrap()).await?;
+ let mut client = indradb_proto::Client::new("grpc://127.0.0.1:27615".try_into()?).await?;
client.ping().await?;
Ok(client)
}
@@ -23,13 +45,12 @@ pub async fn get_client_retrying() -> Result<indradb_proto::Client, indradb_prot
last_err = Some(err);
if retry_count == 0 {
break;
- } else {
- sleep(Duration::from_secs(1)).await;
- retry_count -= 1;
}
+ sleep(Duration::from_secs(1)).await;
+ retry_count -= 1;
}
}
}
- Err(last_err.unwrap())
+ Err(last_err.expect("We didnt get an error even though connection failed"))
}
