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

refactor(pushd): add non-panic error handling to all queue consumers

Browse Source
This commit is contained in:
izzy
2025-05-13 10:41:03 +01:00
parent 01e0f9e558
commit 2aff76c369
14 changed files with 460 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
crates/daemons/pushd/src/consumers/inbound/fr_accepted.rs
View File

@@ -7,6 +7,7 @@ use amqprs::{
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use anyhow::Result;
use async_trait::async_trait;
use log::debug;
use revolt_database::{events::rabbit::*, Database};
@@ -54,21 +55,16 @@ impl FRAcceptedConsumer {
channel: None,
}
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for FRAcceptedConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
async fn consume_event(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) {
let content = String::from_utf8(content).unwrap();
let payload: FRAcceptedPayload = serde_json::from_str(content.as_str()).unwrap();
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: FRAcceptedPayload = serde_json::from_str(content.as_str())?;
debug!("Received FR accept event");
@@ -111,11 +107,34 @@ impl AsyncConsumer for FRAcceptedConsumer {
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable).unwrap();
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
}
}
}
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for FRAcceptedConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
if let Err(err) = self
.consume_event(channel, deliver, basic_properties, content)
.await
{
revolt_config::capture_anyhow(&err);
eprintln!("Failed to process friend request accepted event: {err:?}");
}
}
}

45
crates/daemons/pushd/src/consumers/inbound/fr_received.rs
View File

@@ -7,6 +7,7 @@ use amqprs::{
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use anyhow::Result;
use async_trait::async_trait;
use log::debug;
use revolt_database::{events::rabbit::*, Database};
@@ -54,21 +55,16 @@ impl FRReceivedConsumer {
channel: None,
}
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for FRReceivedConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
async fn consume_event(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) {
let content = String::from_utf8(content).unwrap();
let payload: FRReceivedPayload = serde_json::from_str(content.as_str()).unwrap();
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: FRReceivedPayload = serde_json::from_str(content.as_str())?;
debug!("Received FR received event");
@@ -111,11 +107,34 @@ impl AsyncConsumer for FRReceivedConsumer {
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable).unwrap();
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
}
}
}
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for FRReceivedConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
if let Err(err) = self
.consume_event(channel, deliver, basic_properties, content)
.await
{
revolt_config::capture_anyhow(&err);
eprintln!("Failed to process friend request received event: {err:?}");
}
}
}

45
crates/daemons/pushd/src/consumers/inbound/generic.rs
View File

@@ -7,6 +7,7 @@ use amqprs::{
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use anyhow::Result;
use async_trait::async_trait;
use log::debug;
use revolt_database::{events::rabbit::*, Database};
@@ -54,21 +55,16 @@ impl GenericConsumer {
channel: None,
}
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for GenericConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
async fn consume_event(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) {
let content = String::from_utf8(content).unwrap();
let payload: MessageSentPayload = serde_json::from_str(content.as_str()).unwrap();
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: MessageSentPayload = serde_json::from_str(content.as_str())?;
debug!("Received message event on origin");
@@ -117,11 +113,34 @@ impl AsyncConsumer for GenericConsumer {
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable).unwrap();
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
}
}
}
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for GenericConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
if let Err(err) = self
.consume_event(channel, deliver, basic_properties, content)
.await
{
revolt_config::capture_anyhow(&err);
eprintln!("Failed to process generic event: {err:?}");
}
}
}

75
crates/daemons/pushd/src/consumers/inbound/mass_mention.rs
View File

@@ -10,6 +10,7 @@ use amqprs::{
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use anyhow::Result;
use async_trait::async_trait;
use revolt_database::{
events::rabbit::*, util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member,
@@ -61,7 +62,11 @@ impl MassMessageConsumer {
}
}
async fn fire_notification_for_users(&mut self, push: &PushNotification, users: &[String]) {
async fn fire_notification_for_users(
&mut self,
push: &PushNotification,
users: &[String],
) -> Result<()> {
if let Ok(sessions) = self
.authifier_db
.find_sessions_with_subscription(users)
@@ -105,29 +110,26 @@ impl MassMessageConsumer {
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable).unwrap();
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
}
}
}
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for MassMessageConsumer {
/// This consumer handles adding mentions for all the users affected by a mass mention ping, and then sends out push notifications
async fn consume(
Ok(())
}
async fn consume_event(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) {
) -> Result<()> {
let config = revolt_config::config().await;
let content = String::from_utf8(content).unwrap();
let payload: MassMessageSentPayload = serde_json::from_str(content.as_str()).unwrap();
let content = String::from_utf8(content)?;
let payload: MassMessageSentPayload = serde_json::from_str(content.as_str())?;
debug!("Received mass message event");
@@ -159,8 +161,7 @@ impl AsyncConsumer for MassMessageConsumer {
let mut db_query = self
.db
.fetch_all_members_chunked(&payload.server_id)
.await
.expect("Failed to fetch members from database");
.await?;
let mut exhausted = false;
let ack_chnl = vec![push.channel.id().to_string()];
@@ -203,7 +204,8 @@ impl AsyncConsumer for MassMessageConsumer {
target_users, online_users
);
self.fire_notification_for_users(&push, &target_users).await;
self.fire_notification_for_users(&push, &target_users)
.await?;
if exhausted {
break;
@@ -211,19 +213,11 @@ impl AsyncConsumer for MassMessageConsumer {
}
} else if let Some(roles) = &push.message.role_mentions {
// role mentions
let _role_members = self
let mut role_members = self
.db
.fetch_all_members_with_roles_chunked(&payload.server_id, roles)
.await;
.await?;
debug!("role members: {:?}", _role_members);
if _role_members.is_err() {
revolt_config::capture_error(&_role_members.err().unwrap());
return;
}
let mut role_members = _role_members.unwrap();
let mut chunk = vec![];
let mut exhausted = false;
@@ -266,10 +260,33 @@ impl AsyncConsumer for MassMessageConsumer {
debug!("targets: {:?}", targets);
self.fire_notification_for_users(&push, &targets).await;
self.fire_notification_for_users(&push, &targets).await?;
}
}
}
}
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for MassMessageConsumer {
/// This consumer handles adding mentions for all the users affected by a mass mention ping, and then sends out push notifications
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
if let Err(err) = self
.consume_event(channel, deliver, basic_properties, content)
.await
{
revolt_config::capture_anyhow(&err);
eprintln!("Failed to process mass message event: {err:?}");
}
}
}

45
crates/daemons/pushd/src/consumers/inbound/message.rs
View File

@@ -7,6 +7,7 @@ use amqprs::{
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use anyhow::Result;
use async_trait::async_trait;
use log::debug;
use revolt_database::{events::rabbit::*, Database};
@@ -54,21 +55,16 @@ impl MessageConsumer {
channel: None,
}
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for MessageConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
async fn consume_event(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) {
let content = String::from_utf8(content).unwrap();
let payload: MessageSentPayload = serde_json::from_str(content.as_str()).unwrap();
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: MessageSentPayload = serde_json::from_str(content.as_str())?;
debug!("Received message event on origin");
@@ -117,11 +113,34 @@ impl AsyncConsumer for MessageConsumer {
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable).unwrap();
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
}
}
}
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for MessageConsumer {
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
if let Err(err) = self
.consume_event(channel, deliver, basic_properties, content)
.await
{
revolt_config::capture_anyhow(&err);
eprintln!("Failed to process message event: {err:?}");
}
}
}