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

Merge remote-tracking branch 'origin/main' into feat/elasticsearch

Browse Source 
Signed-off-by: Zomatree <me@zomatree.live>
This commit is contained in:
Zomatree
2026-06-25 06:19:17 +01:00
parent f3a64200cf 23ba7d8fd8
commit 6a421dcd37
241 changed files with 118635 additions and 7140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
crates/daemons/crond/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-crond"
version = "0.12.1"
version = "0.13.7"
license = "AGPL-3.0-or-later"
authors = ["Paul Makles <me@insrt.uk>"]
edition = "2021"
@@ -11,13 +11,28 @@ publish = false
[dependencies]
# Utility
log = "0.4"
log = { workspace = true }
# Async
tokio = { version = "1" }
tokio = { workspace = true }
futures = { workspace = true }
# Redis
redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
# RabbitMQ
lapin = { workspace = true }
futures-lite = { workspace = true }
# Processing
serde_json = { workspace = true }
revolt_optional_struct = { workspace = true }
serde = { workspace = true }
iso8601-timestamp = { workspace = true, features = ["serde", "bson"] }
# Core
revolt-database = { version = "0.12.1", path = "../../core/database" }
revolt-result = { version = "0.12.1", path = "../../core/result" }
revolt-config = { version = "0.12.1", path = "../../core/config" }
revolt-files = { version = "0.12.1", path = "../../core/files" }
revolt-database = { workspace = true }
revolt-result = { workspace = true }
revolt-config = { workspace = true }
revolt-files = { workspace = true }
revolt-permissions = { workspace = true }

53
crates/daemons/crond/src/main.rs
View File

@@ -1,20 +1,51 @@
use revolt_config::configure;
use revolt_database::DatabaseInfo;
use std::{future::Future, panic::AssertUnwindSafe, time::Duration};
use futures::FutureExt;
use revolt_config::{capture_error, configure};
use revolt_database::{Database, DatabaseInfo, AMQP};
use revolt_result::Result;
use tasks::{file_deletion, prune_dangling_files, prune_members};
use tokio::try_join;
use tasks::*;
use tokio::{join, time::sleep};
pub mod tasks;
pub async fn cron_task_wrapper<Fut: Future<Output = Result<()>>>(
func: fn(Database, AMQP) -> Fut,
db: Database,
amqp: AMQP,
) {
loop {
let wrapper = AssertUnwindSafe(func(db.clone(), amqp.clone()));
match wrapper.catch_unwind().await {
Ok(Ok(())) => {
log::error!("cron unexpectedly finshed, Retrying after 60s");
}
Ok(Err(error)) => {
log::error!("cron task failed unexpectedly: {error:?}\nRetrying after 60s");
capture_error(&error);
}
_ => {
log::error!("cron task failed unexpectedly\nRetrying after 60s");
}
}
sleep(Duration::from_secs(60)).await;
}
}
#[tokio::main]
async fn main() -> Result<()> {
async fn main() {
configure!(crond);
let db = DatabaseInfo::Auto.connect().await.expect("database");
try_join!(
file_deletion::task(db.clone()),
prune_dangling_files::task(db.clone()),
prune_members::task(db.clone())
)
.map(|_| ())
let amqp = AMQP::new_auto().await;
join!(
cron_task_wrapper(file_deletion::task, db.clone(), amqp.clone()),
cron_task_wrapper(prune_dangling_files::task, db.clone(), amqp.clone()),
cron_task_wrapper(prune_members::task, db.clone(), amqp.clone()),
cron_task_wrapper(delete_accounts::task, db.clone(), amqp.clone()),
cron_task_wrapper(acks::task, db.clone(), amqp.clone()),
);
}

164
crates/daemons/crond/src/tasks/acks.rs Normal file
View File

@@ -0,0 +1,164 @@
use futures_lite::stream::StreamExt;
use lapin::{
options::*,
types::FieldTable,
uri::{AMQPAuthority, AMQPQueryString, AMQPUri, AMQPUserInfo},
ConnectionBuilder, ConnectionProperties, ExchangeKind,
};
use log::{debug, info};
use redis_kiss::{get_connection, AsyncCommands, Conn as RedisConnection};
use revolt_config::config;
use revolt_database::{events::rabbit::AckEventPayload, Database, AMQP};
use revolt_result::{Result, ToRevoltError};
use serde_json;
pub async fn task(db: Database, amqp: AMQP) -> Result<()> {
let config = config().await;
let mut redis = get_connection()
.await
.expect("Failed to get redis connection");
let uri = AMQPUri {
scheme: lapin::uri::AMQPScheme::AMQP,
authority: AMQPAuthority {
userinfo: AMQPUserInfo {
username: config.rabbit.username,
password: config.rabbit.password,
},
host: config.rabbit.host,
port: config.rabbit.port,
},
vhost: "/".to_string(),
query: AMQPQueryString::default(),
};
let connection = ConnectionBuilder::new()
.expect("Builder")
.with_uri(uri)
.with_properties(ConnectionProperties::default())
.connect()
.await
.expect("Failed to connect to rabbitmq");
let reader_channel = connection
.create_channel()
.await
.expect("Failed to create channel");
reader_channel
.exchange_declare(
config.rabbit.default_exchange.clone().into(),
ExchangeKind::Topic,
ExchangeDeclareOptions {
durable: true,
..Default::default()
},
FieldTable::default(),
)
.await
.expect("Failed to declare exchange");
reader_channel
.queue_declare(
config.rabbit.queues.acks.clone().into(),
QueueDeclareOptions {
durable: true,
..Default::default()
},
FieldTable::default(),
)
.await
.expect("Failed to bind queue");
reader_channel
.queue_bind(
config.rabbit.queues.acks.clone().into(),
config.rabbit.default_exchange.into(),
config.rabbit.queues.acks.clone().into(),
QueueBindOptions::default(),
FieldTable::default(),
)
.await
.expect("Failed to bind channel");
let mut consumer = reader_channel
.basic_consume(
config.rabbit.queues.acks.into(),
"crond-ack-consumer".into(),
BasicConsumeOptions::default(),
FieldTable::default(),
)
.await
.expect("Failed to create consumer");
while let Some(delivery) = consumer.next().await {
if let Ok(delivery) = delivery {
let payload = serde_json::from_slice::<AckEventPayload>(&delivery.data);
if let Ok(payload) = payload {
debug!("Received ack event: {payload:?}");
if let Err(e) = process_channel_ack(
&db,
&amqp,
payload.user_id,
payload.channel_id.unwrap(),
&mut redis,
)
.await
{
revolt_config::capture_error(&e);
_ = delivery.reject(BasicRejectOptions { requeue: false }).await;
} else {
_ = delivery.ack(BasicAckOptions { multiple: false }).await;
}
} else {
revolt_config::capture_message(
format!("Failed to decode ack data: {:?}", delivery.data).as_str(),
revolt_config::Level::Error,
);
}
}
}
Ok(())
}
#[allow(clippy::disallowed_methods)]
async fn process_channel_ack(
db: &Database,
amqp: &AMQP,
user: String,
channel: String,
redis: &mut RedisConnection,
) -> Result<()> {
let message_id: Option<String> = redis
.get_del(format!("acker:{user}+{channel}"))
.await
.to_internal_error()?;
if let Some(message_id) = message_id {
let unread = db.fetch_unread(&user, &channel).await?;
let updated = db.acknowledge_message(&channel, &user, &message_id).await?;
info!("Set new state for ack: {}:{}:{}", channel, user, message_id);
if let (Some(before), Some(after)) = (unread, updated) {
let before_mentions = before.mentions.unwrap_or_default().len();
let after_mentions = after.mentions.unwrap_or_default().len();
if after_mentions < before_mentions {
if let Err(err) = amqp
.ack_notification_message(user.to_string(), channel.to_string(), message_id)
.await
{
revolt_config::capture_error(&err);
}
};
}
Ok(())
} else {
Err(message_id.to_internal_error().expect_err("no err"))
}
}

23
crates/daemons/crond/src/tasks/delete_accounts.rs Normal file
View File

@@ -0,0 +1,23 @@
use std::time::Duration;
use revolt_database::Database;
use revolt_result::Result;
use tokio::time::sleep;
pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
loop {
let accounts = db.fetch_accounts_due_for_deletion().await?;
let count = accounts.len();
for mut account in accounts {
let mut user = db.fetch_user(&account.id).await?;
user.delete(&db).await?;
account.mark_deleted(&db).await?;
}
log::info!("Deleted {count} accounts.");
sleep(Duration::from_hours(1)).await
}
}

26
crates/daemons/crond/src/tasks/file_deletion.rs
View File

@@ -6,27 +6,25 @@ use revolt_files::delete_from_s3;
use revolt_result::Result;
use tokio::time::sleep;
pub async fn task(db: Database) -> Result<()> {
pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
loop {
let files = db.fetch_deleted_attachments().await?;
for file in files {
let count = db
.count_file_hash_references(file.hash.as_ref().expect("no `hash` present"))
.await?;
if let Some(hash) = &file.hash {
let count = db.count_file_hash_references(hash).await?;
// No other files reference this file on disk anymore
if count <= 1 {
let file_hash = db
.fetch_attachment_hash(file.hash.as_ref().expect("no `hash` present"))
.await?;
// No other files reference this file on disk anymore
if count <= 1 {
let file_hash = db.fetch_attachment_hash(hash).await?;
// Delete from S3
delete_from_s3(&file_hash.bucket_id, &file_hash.path).await?;
// Delete from S3
delete_from_s3(&file_hash.bucket_id, &file_hash.path).await?;
// Delete the hash
db.delete_attachment_hash(&file_hash.id).await?;
info!("Deleted file hash {}", file_hash.id);
// Delete the hash
db.delete_attachment_hash(&file_hash.id).await?;
info!("Deleted file hash {}", file_hash.id);
}
}
// Delete the file

3
crates/daemons/crond/src/tasks/mod.rs
View File

@@ -1,3 +1,6 @@
pub mod delete_accounts;
pub mod acks;
pub mod file_deletion;
pub mod prune_dangling_files;
pub mod prune_members;
pub mod prune_mfa_tickets;

2
crates/daemons/crond/src/tasks/prune_dangling_files.rs
View File

@@ -6,7 +6,7 @@ use tokio::time::sleep;
use log::info;
pub async fn task(db: Database) -> Result<()> {
pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
loop {
// This could just be a single database query
// ... but timestamps are inconsistently serialised

2
crates/daemons/crond/src/tasks/prune_members.rs
View File

@@ -5,7 +5,7 @@ use revolt_database::Database;
use revolt_result::Result;
use tokio::time::sleep;
pub async fn task(db: Database) -> Result<()> {
pub async fn task(db: Database, _: revolt_database::AMQP) -> Result<()> {
loop {
let success = db.remove_dangling_members().await;
if let Err(s) = success {

14
crates/daemons/crond/src/tasks/prune_mfa_tickets.rs Normal file
View File

@@ -0,0 +1,14 @@
use std::time::Duration;
use revolt_database::Database;
use revolt_result::Result;
use tokio::time::sleep;
pub async fn task(db: Database) -> Result<()> {
loop {
let count = db.delete_expired_tickets().await?;
log::info!("Pruned {count} expired MFA tickets");
sleep(Duration::from_mins(5)).await
}
}

59
crates/daemons/pushd/Cargo.toml
View File

@@ -1,47 +1,38 @@
[package]
name = "revolt-pushd"
version = "0.12.1"
version = "0.13.7"
edition = "2021"
license = "AGPL-3.0-or-later"
publish = false
[dependencies]
revolt-result = { version = "0.12.1", path = "../../core/result" }
revolt-config = { version = "0.12.1", path = "../../core/config", features = [
"report-macros",
"anyhow",
] }
revolt-database = { version = "0.12.1", path = "../../core/database" }
revolt-models = { version = "0.12.1", path = "../../core/models", features = [
"validator",
] }
revolt-presence = { version = "0.12.1", path = "../../core/presence", features = [
"redis-is-patched",
] }
revolt-parser = { version = "0.12.1", path = "../../core/parser" }
revolt-result = { workspace = true }
revolt-config = { workspace = true, features = ["report-macros", "anyhow"] }
revolt-database = { workspace = true }
revolt-models = { workspace = true, features = ["validator"] }
revolt-presence = { workspace = true, features = ["redis-is-patched"] }
revolt-parser = { workspace = true }
anyhow = { version = "1.0.98" }
anyhow = { workspace = true }
amqprs = { version = "1.7.0" }
fcm_v1 = "0.3.0"
web-push = "0.10.0"
isahc = { optional = true, version = "1.7", features = ["json"] }
revolt_a2 = { version = "0.10", default-features = false, features = ["ring"] }
redis-kiss = "0.1.4"
tokio = "1.39.2"
async-trait = "0.1.81"
ulid = "1.0.0"
lapin = { workspace = true }
fcm_v1 = { workspace = true }
web-push = { workspace = true }
isahc = { workspace = true, features = ["json"], optional = true }
revolt_a2 = { workspace = true, features = ["ring"] }
redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
tokio = { workspace = true }
async-trait = { workspace = true }
ulid = { workspace = true }
authifier = "1.0.16"
log = { workspace = true }
pretty_env_logger = { workspace = true }
log = "0.4.11"
pretty_env_logger = "0.4.0"
regex = "1.12.3"
regex = { workspace = true }
#serialization
serde_json = "1"
revolt_optional_struct = "0.2.0"
serde = { version = "1", features = ["derive"] }
iso8601-timestamp = { version = "0.2.10", features = ["serde", "bson"] }
base64 = "0.22.1"
serde_json = { workspace = true }
revolt_optional_struct = { workspace = true }
serde = { workspace = true }
iso8601-timestamp = { workspace = true, features = ["serde", "bson"] }
base64 = { workspace = true }

148
crates/daemons/pushd/src/consumers/inbound/ack.rs
View File

@@ -1,96 +1,66 @@
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use std::sync::Arc;
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
pub struct AckConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
connection: Arc<Connection>,
channel: Arc<Channel>,
}
impl Channeled for AckConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl AckConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> AckConsumer {
AckConsumer {
db,
authifier_db,
conn: None,
channel: None,
}
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for AckConsumer {
impl Consumer for AckConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
channel,
}
}
fn channel(&self) -> &Arc<Channel> {
&self.channel
}
/// This consumer processes all acks the platform receives, and sends relevant badge updates to apple platforms.
async fn consume(
&mut self,
channel: &Channel,
deliver: Deliver,
basic_properties: BasicProperties,
content: Vec<u8>,
) {
let content = String::from_utf8(content).unwrap();
let payload: AckPayload = serde_json::from_str(content.as_str()).unwrap();
async fn consume(&self, delivery: Delivery) -> Result<()> {
let payload: AckPayload = serde_json::from_slice(&delivery.data)?;
// Step 1: fetch unreads and don't continue if there's no unreads
#[allow(clippy::disallowed_methods)]
let unreads = self.db.fetch_unread_mentions(&payload.user_id).await;
// #[allow(clippy::disallowed_methods)]
debug!("Processing unreads for {:}", &payload.user_id);
if let Ok(u) = &unreads {
let unreads = if let Ok(u) = self.db.fetch_unread_mentions(&payload.user_id).await {
if u.is_empty() {
debug!(
"Discarding unread task (no mentions found) for {:}",
&payload.user_id
);
return;
}
} else {
return;
}
return Ok(());
};
if let Ok(sessions) = self.authifier_db.find_sessions(&payload.user_id).await {
u
} else {
return Ok(());
};
if let Ok(sessions) = self.db.fetch_sessions(&payload.user_id).await {
let config = revolt_config::config().await;
// Step 2: find any apple sessions, since we don't need to calculate this for anything else.
// If there's no apple sessions, we can return early
let apple_sessions: Vec<&authifier::models::Session> = sessions
.iter()
let mut apple_sessions = sessions
.into_iter()
.filter(|session| {
if let Some(sub) = &session.subscription {
sub.endpoint == "apn"
@@ -98,19 +68,19 @@ impl AsyncConsumer for AckConsumer {
false
}
})
.collect();
.peekable();
if apple_sessions.is_empty() {
if apple_sessions.peek().is_none() {
debug!(
"Discarding unread task (no apn sessions found) for {:}",
&payload.user_id
);
return;
return Ok(());
}
// Step 3: calculate the actual mention count, since we have to send it out
let mut mention_count = 0;
for u in &unreads.unwrap() {
for u in &unreads {
mention_count += u.mentions.as_ref().unwrap().len()
}
@@ -123,26 +93,22 @@ impl AsyncConsumer for AckConsumer {
token: session.subscription.as_ref().unwrap().auth.clone(),
extras: Default::default(),
};
let raw_service_payload = serde_json::to_string(&service_payload);
let payload = serde_json::to_string(&service_payload)?;
if let Ok(p) = raw_service_payload {
let args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
log::debug!(
"Publishing ack to apn session {}",
session.subscription.as_ref().unwrap().auth
);
log::debug!(
"Publishing ack to apn session {}",
session.subscription.as_ref().unwrap().auth
);
publish_message(self, p.into(), args).await;
} else {
log::warn!("Failed to serialize ack badge update payload!");
revolt_config::capture_error(&raw_service_payload.unwrap_err());
}
self.publish_message(
payload.as_bytes(),
&config.pushd.exchange,
&config.pushd.apn.queue,
)
.await?;
}
}
Ok(())
}
}

143
crates/daemons/pushd/src/consumers/inbound/dm_call.rs
View File

@@ -1,70 +1,41 @@
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
pub struct DmCallConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
connection: Arc<Connection>,
channel: Arc<Channel>,
}
impl Channeled for DmCallConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl DmCallConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> DmCallConsumer {
DmCallConsumer {
#[async_trait]
impl Consumer for DmCallConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
authifier_db,
conn: None,
channel: None,
connection,
channel,
}
}
async fn consume_event(
&mut self,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let _p: InternalDmCallPayload = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<Channel> {
&self.channel
}
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(&self, delivery: Delivery) -> Result<()> {
let _p: InternalDmCallPayload = serde_json::from_slice(&delivery.data)?;
let payload = _p.payload;
debug!("Received dm call start/stop event");
@@ -96,7 +67,7 @@ impl DmCallConsumer {
let config = revolt_config::config().await;
for user_id in call_recipients {
if let Ok(sessions) = self.authifier_db.find_sessions(&user_id).await {
if let Ok(sessions) = self.db.fetch_sessions(&user_id).await {
for session in sessions {
if let Some(sub) = session.subscription {
let mut sendable = PayloadToService {
@@ -107,36 +78,27 @@ impl DmCallConsumer {
extras: HashMap::new(),
};
let args: BasicPublishArguments;
let routing_key = match sub.endpoint.as_str() {
"apn" => &config.pushd.apn.queue,
"fcm" => &config.pushd.fcm.queue,
endpoint => {
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), endpoint.to_string());
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
&config.pushd.vapid.queue
}
};
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
self.publish_message(
payload.as_bytes(),
&config.pushd.exchange,
routing_key,
)
.await?;
}
}
}
@@ -145,24 +107,3 @@ impl DmCallConsumer {
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for DmCallConsumer {
/// 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);
warn!("Failed to process dm call start/stop event: {err:?}");
}
}
}

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

@@ -1,74 +1,45 @@
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
pub struct FRAcceptedConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
connection: Arc<Connection>,
channel: Arc<Channel>,
}
impl Channeled for FRAcceptedConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl FRAcceptedConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> FRAcceptedConsumer {
FRAcceptedConsumer {
#[async_trait]
impl Consumer for FRAcceptedConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
authifier_db,
conn: None,
channel: None,
connection,
channel,
}
}
async fn consume_event(
&mut self,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: FRAcceptedPayload = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<Channel> {
&self.channel
}
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(&self, delivery: Delivery) -> Result<()> {
let payload: FRAcceptedPayload = serde_json::from_slice(&delivery.data)?;
debug!("Received FR accept event");
if let Ok(sessions) = self.authifier_db.find_sessions(&payload.user).await {
if let Ok(sessions) = self.db.fetch_sessions(&payload.user).await {
let config = revolt_config::config().await;
for session in sessions {
if let Some(sub) = session.subscription {
@@ -80,36 +51,23 @@ impl FRAcceptedConsumer {
extras: HashMap::new(),
};
let args: BasicPublishArguments;
let routing_key = match sub.endpoint.as_str() {
"apn" => &config.pushd.apn.queue,
"fcm" => &config.pushd.fcm.queue,
endpoint => {
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), endpoint.to_string());
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
&config.pushd.vapid.queue
}
};
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
.await?;
}
}
}
@@ -117,24 +75,3 @@ impl FRAcceptedConsumer {
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:?}");
}
}
}

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

@@ -1,74 +1,45 @@
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
pub struct FRReceivedConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
connection: Arc<Connection>,
channel: Arc<Channel>,
}
impl Channeled for FRReceivedConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl FRReceivedConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> FRReceivedConsumer {
FRReceivedConsumer {
#[async_trait]
impl Consumer for FRReceivedConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
authifier_db,
conn: None,
channel: None,
connection,
channel,
}
}
async fn consume_event(
&mut self,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: FRReceivedPayload = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<Channel> {
&self.channel
}
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(&self, delivery: Delivery) -> Result<()> {
let payload: FRReceivedPayload = serde_json::from_slice(&delivery.data)?;
debug!("Received FR received event");
if let Ok(sessions) = self.authifier_db.find_sessions(&payload.user).await {
if let Ok(sessions) = self.db.fetch_sessions(&payload.user).await {
let config = revolt_config::config().await;
for session in sessions {
if let Some(sub) = session.subscription {
@@ -80,36 +51,23 @@ impl FRReceivedConsumer {
extras: HashMap::new(),
};
let args: BasicPublishArguments;
let routing_key = match sub.endpoint.as_str() {
"apn" => &config.pushd.apn.queue,
"fcm" => &config.pushd.fcm.queue,
endpoint => {
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), endpoint.to_string());
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
&config.pushd.vapid.queue
}
};
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
.await?;
}
}
}
@@ -117,24 +75,3 @@ impl FRReceivedConsumer {
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:?}");
}
}
}

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

@@ -1,76 +1,47 @@
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{events::rabbit::*, Database};
#[derive(Clone)]
#[allow(unused)]
pub struct GenericConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
connection: Arc<Connection>,
channel: Arc<Channel>,
}
impl Channeled for GenericConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl GenericConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> GenericConsumer {
GenericConsumer {
#[async_trait]
impl Consumer for GenericConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
authifier_db,
conn: None,
channel: None,
connection,
channel,
}
}
async fn consume_event(
&mut self,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: MessageSentPayload = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<Channel> {
&self.channel
}
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(&self, delivery: Delivery) -> Result<()> {
let payload: MessageSentPayload = serde_json::from_slice(&delivery.data)?;
debug!("Received message event on origin");
if let Ok(sessions) = self
.authifier_db
.find_sessions_with_subscription(&payload.users)
.db
.fetch_sessions_with_subscription(&payload.users)
.await
{
let config = revolt_config::config().await;
@@ -86,36 +57,23 @@ impl GenericConsumer {
extras: HashMap::new(),
};
let args: BasicPublishArguments;
let routing_key = match sub.endpoint.as_str() {
"apn" => &config.pushd.apn.queue,
"fcm" => &config.pushd.fcm.queue,
endpoint => {
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), endpoint.to_string());
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
&config.pushd.vapid.queue
}
};
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
.await?;
}
}
}
@@ -123,24 +81,3 @@ impl GenericConsumer {
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:?}");
}
}
}

53
crates/daemons/pushd/src/consumers/inbound/internal.rs
View File

@@ -1,53 +0,0 @@
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::{Connection, OpenConnectionArguments},
BasicProperties,
};
use log::{debug, warn};
pub(crate) trait Channeled {
#[allow(unused)]
fn get_connection(&self) -> Option<&Connection>;
fn get_channel(&self) -> Option<&Channel>;
fn set_connection(&mut self, conn: Connection);
fn set_channel(&mut self, channel: Channel);
}
pub(crate) async fn make_channel<T: Channeled>(consumer: &mut T) {
let config = revolt_config::config().await;
let args = OpenConnectionArguments::new(
&config.rabbit.host,
config.rabbit.port,
&config.rabbit.username,
&config.rabbit.password,
);
let conn = amqprs::connection::Connection::open(&args).await.unwrap();
let channel = conn.open_channel(None).await.unwrap();
consumer.set_connection(conn);
consumer.set_channel(channel);
}
pub(crate) async fn publish_message<T: Channeled>(
consumer: &mut T,
payload: Vec<u8>,
args: BasicPublishArguments,
) {
let routing_key = &args.routing_key.clone();
let mut channel = consumer.get_channel();
if channel.is_none() {
make_channel(consumer).await;
channel = consumer.get_channel();
}
if let Some(chnl) = channel {
chnl.basic_publish(BasicProperties::default(), payload.clone(), args.clone())
.await
.unwrap();
debug!("Sent message to queue for target {}", routing_key);
} else {
warn!("Failed to unwrap channel (including attempt to make a channel)!")
}
}

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

@@ -1,17 +1,13 @@
use std::{
collections::{HashMap, HashSet},
hash::RandomState,
sync::Arc,
};
use crate::{consumers::inbound::internal::*, utils};
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use crate::utils::{render_notification_content, Consumer};
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use revolt_database::{
events::rabbit::*, util::bulk_permissions::BulkDatabasePermissionQuery, Database, Member,
MessageFlagsValue,
@@ -19,58 +15,23 @@ use revolt_database::{
use revolt_models::v0::{MessageFlags, PushNotification};
use revolt_result::ToRevoltError;
#[derive(Clone)]
#[allow(unused)]
pub struct MassMessageConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
}
impl Channeled for MassMessageConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
connection: Arc<Connection>,
channel: Arc<Channel>,
}
impl MassMessageConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> MassMessageConsumer {
MassMessageConsumer {
db,
authifier_db,
conn: None,
channel: None,
}
}
async fn fire_notification_for_users(
&mut self,
&self,
push: &PushNotification,
users: &[String],
) -> Result<()> {
if let Ok(sessions) = self
.authifier_db
.find_sessions_with_subscription(users)
.db
.fetch_sessions_with_subscription(users)
.await
{
let config = revolt_config::config().await;
@@ -84,56 +45,56 @@ impl MassMessageConsumer {
extras: HashMap::new(),
};
let args: BasicPublishArguments;
let routing_key = match sub.endpoint.as_str() {
"apn" => &config.pushd.apn.queue,
"fcm" => &config.pushd.fcm.queue,
endpoint => {
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), endpoint.to_string());
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
&config.pushd.vapid.queue
}
};
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
.await?;
}
}
}
Ok(())
}
}
async fn consume_event(
&mut self,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
#[async_trait]
impl Consumer for MassMessageConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
connection,
channel,
}
}
fn channel(&self) -> &Arc<Channel> {
&self.channel
}
/// This consumer handles adding mentions for all the users affected by a mass mention ping, and then sends out push notifications.
async fn consume(&self, delivery: Delivery) -> Result<()> {
let mut payload: MassMessageSentPayload = serde_json::from_slice(&delivery.data)?;
let config = revolt_config::config().await;
let content = String::from_utf8(content)?;
let mut payload: MassMessageSentPayload = serde_json::from_str(content.as_str())?;
for push in payload.notifications.iter_mut() {
if let Ok(body) = utils::render_notification_content(push, &self.db)
if let Ok(body) = render_notification_content(push, &self.db)
.await
.to_internal_error()
{
@@ -280,24 +241,3 @@ impl MassMessageConsumer {
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:?}");
}
}
}

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

@@ -1,76 +1,43 @@
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
use crate::{consumers::inbound::internal::*, utils};
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use crate::utils::{render_notification_content, Consumer};
use anyhow::Result;
use async_trait::async_trait;
use lapin::{message::Delivery, Channel, Connection};
use log::debug;
use revolt_database::{events::rabbit::*, Database};
use revolt_result::ToRevoltError;
#[derive(Clone)]
#[allow(unused)]
pub struct MessageConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<Channel>,
connection: Arc<Connection>,
channel: Arc<Channel>,
}
impl Channeled for MessageConsumer {
fn get_connection(&self) -> Option<&Connection> {
if self.conn.is_none() {
None
} else {
Some(self.conn.as_ref().unwrap())
}
}
fn get_channel(&self) -> Option<&Channel> {
if self.channel.is_none() {
None
} else {
Some(self.channel.as_ref().unwrap())
}
}
fn set_connection(&mut self, conn: Connection) {
self.conn = Some(conn);
}
fn set_channel(&mut self, channel: Channel) {
self.channel = Some(channel)
}
}
impl MessageConsumer {
pub fn new(db: Database, authifier_db: authifier::Database) -> MessageConsumer {
MessageConsumer {
#[async_trait]
impl Consumer for MessageConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self {
Self {
db,
authifier_db,
conn: None,
channel: None,
connection,
channel,
}
}
async fn consume_event(
&mut self,
_channel: &Channel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let mut payload: MessageSentPayload = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<Channel> {
&self.channel
}
if let Ok(body) = utils::render_notification_content(&payload.notification, &self.db)
.await
.to_internal_error()
{
/// This consumer handles delegating messages into their respective platform queues.
async fn consume(&self, delivery: Delivery) -> Result<()> {
let mut payload: MessageSentPayload = serde_json::from_slice(&delivery.data)?;
if let Ok(body) = render_notification_content(&payload.notification, &self.db).await {
payload.notification.raw_body = Some(payload.notification.body);
payload.notification.body = body;
}
@@ -78,8 +45,8 @@ impl MessageConsumer {
debug!("Received message event on origin");
if let Ok(sessions) = self
.authifier_db
.find_sessions_with_subscription(&payload.users)
.db
.fetch_sessions_with_subscription(&payload.users)
.await
{
let config = revolt_config::config().await;
@@ -95,36 +62,22 @@ impl MessageConsumer {
extras: HashMap::new(),
};
let args: BasicPublishArguments;
let routing_key = match sub.endpoint.as_str() {
"apn" => &config.pushd.apn.queue,
"fcm" => &config.pushd.fcm.queue,
endpoint => {
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), endpoint.to_string());
if sub.endpoint == "apn" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.apn.queue.as_str(),
)
.finish();
} else if sub.endpoint == "fcm" {
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.fcm.queue.as_str(),
)
.finish();
} else {
// web push (vapid)
args = BasicPublishArguments::new(
config.pushd.exchange.as_str(),
config.pushd.vapid.queue.as_str(),
)
.finish();
sendable.extras.insert("p256dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
&config.pushd.vapid.queue
}
};
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
self.publish_message(payload.as_bytes(), &config.pushd.exchange, routing_key)
.await?;
}
}
}
@@ -132,24 +85,3 @@ impl MessageConsumer {
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:?}");
}
}
}

1
crates/daemons/pushd/src/consumers/inbound/mod.rs
View File

@@ -3,6 +3,5 @@ pub mod dm_call;
pub mod fr_accepted;
pub mod fr_received;
pub mod generic;
mod internal;
pub mod mass_mention;
pub mod message;

156
crates/daemons/pushd/src/consumers/outbound/apn.rs
View File

@@ -1,12 +1,13 @@
use std::{borrow::Cow, collections::BTreeMap, io::Cursor};
use std::{borrow::Cow, collections::BTreeMap, io::Cursor, sync::Arc};
use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
use anyhow::{anyhow, Result};
use crate::utils::Consumer;
use anyhow::Result;
use async_trait::async_trait;
use base64::{
engine::{self},
Engine as _,
};
use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
use revolt_a2::{
request::{
notification::{DefaultAlert, NotificationOptions},
@@ -42,7 +43,7 @@ impl<'a> PayloadLike for MessagePayload<'a> {
fn get_device_token(&self) -> &'a str {
self.device_token
}
fn get_options(&self) -> &NotificationOptions {
fn get_options(&self) -> &NotificationOptions<'a> {
&self.options
}
}
@@ -68,16 +69,19 @@ impl<'a> PayloadLike for CallStartStopPayload<'a> {
fn get_device_token(&self) -> &'a str {
self.device_token
}
fn get_options(&self) -> &NotificationOptions {
fn get_options(&self) -> &NotificationOptions<'a> {
&self.options
}
}
// region: consumer
#[derive(Clone)]
#[allow(unused)]
pub struct ApnsOutboundConsumer {
#[allow(dead_code)]
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
client: Client,
}
@@ -117,15 +121,20 @@ impl ApnsOutboundConsumer {
}
}
impl ApnsOutboundConsumer {
pub async fn new(db: Database) -> Result<ApnsOutboundConsumer, &'static str> {
#[async_trait]
impl Consumer for ApnsOutboundConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
) -> Self {
let config = revolt_config::config().await;
if config.pushd.apn.pkcs8.is_empty()
|| config.pushd.apn.key_id.is_empty()
|| config.pushd.apn.team_id.is_empty()
{
return Err("Missing APN keys.");
panic!("Missing APN keys.");
}
let endpoint = if config.pushd.apn.sandbox {
@@ -148,18 +157,20 @@ impl ApnsOutboundConsumer {
)
.expect("could not create APN client");
Ok(ApnsOutboundConsumer { db, client })
Self {
db,
connection,
channel,
client,
}
}
async fn consume_event(
&mut self,
_channel: &AmqpChannel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: PayloadToService = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<AMQPChannel> {
&self.channel
}
async fn consume(&self, delivery: Delivery) -> Result<()> {
let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
let payload_options = NotificationOptions {
apns_id: None,
@@ -170,20 +181,15 @@ impl ApnsOutboundConsumer {
apns_collapse_id: None,
};
let resp: Result<Response, Error>;
match payload.notification {
let resp = match payload.notification {
PayloadKind::FRReceived(alert) => {
let loc_args = vec![Cow::from(
alert
.from_user
.display_name
.or(Some(format!(
alert.from_user.display_name.clone().unwrap_or_else(|| {
format!(
"{}#{}",
alert.from_user.username, alert.from_user.discriminator
)))
.clone()
.ok_or_else(|| anyhow!("missing name"))?,
)
}),
)];
let apn_payload = Payload {
@@ -216,20 +222,17 @@ impl ApnsOutboundConsumer {
"Sending friend request received for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await;
self.client.send(apn_payload).await
}
PayloadKind::FRAccepted(alert) => {
let loc_args = vec![Cow::from(
alert
.accepted_user
.display_name
.or(Some(format!(
alert.accepted_user.display_name.clone().unwrap_or_else(|| {
format!(
"{}#{}",
alert.accepted_user.username, alert.accepted_user.discriminator
)))
.clone()
.ok_or_else(|| anyhow!("missing name"))?,
)
}),
)];
let apn_payload = Payload {
@@ -262,7 +265,7 @@ impl ApnsOutboundConsumer {
"Sending friend request accept for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await;
self.client.send(apn_payload).await
}
PayloadKind::Generic(alert) => {
let apn_payload = Payload {
@@ -295,7 +298,7 @@ impl ApnsOutboundConsumer {
"Sending generic notification for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await;
self.client.send(apn_payload).await
}
PayloadKind::MessageNotification(alert) => {
@@ -334,7 +337,7 @@ impl ApnsOutboundConsumer {
"Sending message notification for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await;
self.client.send(apn_payload).await
}
PayloadKind::BadgeUpdate(badge) => {
@@ -349,7 +352,7 @@ impl ApnsOutboundConsumer {
};
debug!("Sending badge update for user: {:}", &payload.user_id);
resp = self.client.send(apn_payload).await;
self.client.send(apn_payload).await
}
PayloadKind::DmCallStartEnd(alert) => {
@@ -378,58 +381,37 @@ impl ApnsOutboundConsumer {
"Sending call start/stop notification for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await;
self.client.send(apn_payload).await
}
}
};
if let Err(err) = resp {
match err {
Error::ResponseError(Response {
error:
Some(ErrorBody {
reason: ErrorReason::BadDeviceToken | ErrorReason::Unregistered,
..
}),
..
}) => {
info!(
"Removing APNS subscription id {:} (user: {:}) due to invalid token",
&payload.session_id, &payload.user_id
);
if let Err(err) = self
.db
.remove_push_subscription_by_session_id(&payload.session_id)
.await
{
revolt_config::capture_error(&err);
}
}
err => {
match resp {
Err(Error::ResponseError(Response {
error:
Some(ErrorBody {
reason: ErrorReason::BadDeviceToken | ErrorReason::Unregistered,
..
}),
..
})) => {
info!(
"Removing APNS subscription id {:} (user: {:}) due to invalid token",
&payload.session_id, &payload.user_id
);
if let Err(err) = self
.db
.remove_push_subscription_by_session_id(&payload.session_id)
.await
{
revolt_config::capture_error(&err);
}
}
}
resp => {
resp?;
}
};
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for ApnsOutboundConsumer {
async fn consume(
&mut self,
channel: &AmqpChannel,
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 APN event: {err:?}");
}
}
}

318
crates/daemons/pushd/src/consumers/outbound/fcm.rs
View File

@@ -1,51 +1,142 @@
use std::{collections::HashMap, time::Duration};
use std::{collections::HashMap, sync::Arc, time::Duration};
use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
use anyhow::{anyhow, bail, Result};
use crate::utils::Consumer;
use anyhow::{bail, Result};
use async_trait::async_trait;
use fcm_v1::{
android::{AndroidConfig, AndroidMessagePriority},
auth::{Authenticator, ServiceAccountKey},
message::{Message, Notification},
message::Message,
Client, Error as FcmError,
};
use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
use revolt_config::config;
use revolt_database::{events::rabbit::*, Database};
use revolt_models::v0::{Channel, PushNotification};
use serde_json::Value;
pub struct FcmOutboundConsumer {
db: Database,
client: Client,
/// Custom notification data
#[derive(Debug, Clone, PartialEq)]
pub enum NotificationData {
FRReceived {
id: String,
username: String,
},
FRAccepted {
id: String,
username: String,
},
Generic {
title: String,
body: String,
image: Option<String>,
},
Message {
message: String,
body: String,
image: String,
channel: String,
author: String,
author_name: String,
},
DmCallStartEnd {
initiator_id: String,
channel_id: String,
started_at: String,
ended: bool,
duration: usize,
},
}
impl FcmOutboundConsumer {
fn format_title(&self, notification: &PushNotification) -> String {
// ideally this changes depending on context
// in a server, it would look like "Sendername, #channelname in servername"
// in a group, it would look like "Sendername in groupname"
// in a dm it should just be "Sendername".
// not sure how feasible all those are given the PushNotification object as it currently stands.
#[allow(deprecated)]
match &notification.channel {
Channel::DirectMessage { .. } => notification.author.clone(),
Channel::Group { name, .. } => format!("{}, #{}", notification.author, name),
Channel::TextChannel { name, .. } => {
format!("{} in #{}", notification.author, name)
}
_ => "Unknown".to_string(),
impl NotificationData {
pub fn get_type(&self) -> &str {
match self {
NotificationData::FRReceived { .. } => "push.fr.receive",
NotificationData::FRAccepted { .. } => "push.fr.accept",
NotificationData::Generic { .. } => "push.generic",
NotificationData::Message { .. } => "push.message",
NotificationData::DmCallStartEnd { .. } => "push.dm.call",
}
}
pub fn into_payload(self) -> HashMap<String, Value> {
let mut data = HashMap::new();
data.insert(
"type".to_string(),
Value::String(self.get_type().to_string()),
);
match self {
NotificationData::FRReceived { id, username } => {
data.insert("id".to_string(), Value::String(id));
data.insert("username".to_string(), Value::String(username));
}
NotificationData::FRAccepted { id, username } => {
data.insert("id".to_string(), Value::String(id));
data.insert("username".to_string(), Value::String(username));
}
NotificationData::Generic { title, body, image } => {
data.insert("title".to_string(), Value::String(title));
data.insert("body".to_string(), Value::String(body));
if let Some(image) = image {
data.insert("image".to_string(), Value::String(image));
}
}
NotificationData::Message {
message,
body,
image,
channel,
author,
author_name,
} => {
data.insert("message".to_string(), Value::String(message));
data.insert("body".to_string(), Value::String(body));
data.insert("image".to_string(), Value::String(image));
data.insert("channel".to_string(), Value::String(channel));
data.insert("author".to_string(), Value::String(author));
data.insert("author_name".to_string(), Value::String(author_name));
}
NotificationData::DmCallStartEnd {
initiator_id,
channel_id,
started_at,
ended,
duration,
} => {
data.insert("initiator_id".to_string(), Value::String(initiator_id));
data.insert("channel_id".to_string(), Value::String(channel_id));
data.insert("started_at".to_string(), Value::String(started_at));
data.insert("ended".to_string(), Value::Bool(ended));
data.insert("duration".to_string(), Value::Number(duration.into()));
}
}
data
}
}
impl FcmOutboundConsumer {
pub async fn new(db: Database) -> Result<FcmOutboundConsumer, &'static str> {
#[derive(Clone)]
#[allow(unused)]
pub struct FcmOutboundConsumer {
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
client: Client,
}
#[async_trait]
impl Consumer for FcmOutboundConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
) -> Self {
let config = revolt_config::config().await;
Ok(FcmOutboundConsumer {
Self {
db,
connection,
channel,
client: Client::new(
Authenticator::service_account::<&str>(ServiceAccountKey {
key_type: Some(config.pushd.fcm.key_type),
@@ -65,45 +156,36 @@ impl FcmOutboundConsumer {
false,
Duration::from_secs(5),
),
})
}
}
async fn consume_event(
&mut self,
_channel: &AmqpChannel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: PayloadToService = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<AMQPChannel> {
&self.channel
}
async fn consume(&self, delivery: Delivery) -> Result<()> {
let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
#[allow(clippy::needless_late_init)]
let resp: Result<Message, FcmError>;
match payload.notification {
PayloadKind::FRReceived(alert) => {
let name = alert
.from_user
.display_name
.or(Some(format!(
let name = alert.from_user.display_name.clone().unwrap_or_else(|| {
format!(
"{}#{}",
alert.from_user.username, alert.from_user.discriminator
)))
.clone()
.ok_or_else(|| anyhow!("missing name"))?;
)
});
let mut data = HashMap::new();
data.insert(
"type".to_string(),
Value::String("push.fr.receive".to_string()),
);
data.insert("id".to_string(), Value::String(alert.from_user.id));
data.insert("username".to_string(), Value::String(name));
let data = NotificationData::FRReceived {
id: alert.from_user.id,
username: name,
};
let msg = Message {
token: Some(payload.token),
data: Some(data),
data: Some(data.into_payload()),
..Default::default()
};
@@ -111,40 +193,36 @@ impl FcmOutboundConsumer {
}
PayloadKind::FRAccepted(alert) => {
let name = alert
.accepted_user
.display_name
.or(Some(format!(
let name = alert.accepted_user.display_name.clone().unwrap_or_else(|| {
format!(
"{}#{}",
alert.accepted_user.username, alert.accepted_user.discriminator
)))
.clone()
.ok_or_else(|| anyhow!("missing name"))?;
)
});
let mut data: HashMap<String, Value> = HashMap::new();
data.insert(
"type".to_string(),
Value::String("push.fr.accept".to_string()),
);
data.insert("id".to_string(), Value::String(alert.accepted_user.id));
data.insert("username".to_string(), Value::String(name));
let data = NotificationData::FRAccepted {
id: alert.accepted_user.id,
username: name,
};
let msg = Message {
token: Some(payload.token),
data: Some(data),
data: Some(data.into_payload()),
..Default::default()
};
resp = self.client.send(&msg).await;
}
PayloadKind::Generic(alert) => {
let data = NotificationData::Generic {
title: alert.title,
body: alert.body,
image: alert.icon,
};
let msg = Message {
token: Some(payload.token),
notification: Some(Notification {
title: Some(alert.title),
body: Some(alert.body),
image: alert.icon,
}),
data: Some(data.into_payload()),
..Default::default()
};
@@ -152,19 +230,18 @@ impl FcmOutboundConsumer {
}
PayloadKind::MessageNotification(alert) => {
let title = self.format_title(&alert);
let data = NotificationData::Message {
message: alert.message.id,
body: alert.body,
image: alert.icon,
channel: alert.message.channel,
author: alert.message.author,
author_name: alert.author,
};
let msg = Message {
token: Some(payload.token),
notification: Some(Notification {
title: Some(title),
body: Some(alert.body),
image: Some(alert.icon),
}),
android: Some(AndroidConfig {
collapse_key: Some(alert.tag),
..Default::default()
}),
data: Some(data.into_payload()),
..Default::default()
};
@@ -172,30 +249,17 @@ impl FcmOutboundConsumer {
}
PayloadKind::DmCallStartEnd(alert) => {
let mut data: HashMap<String, Value> = HashMap::new();
data.insert(
"initiator_id".to_string(),
Value::String(alert.initiator_id),
);
data.insert("channel_id".to_string(), Value::String(alert.channel_id));
data.insert(
"started_at".to_string(),
Value::String(alert.started_at.unwrap_or_else(|| "".to_string())),
);
data.insert("ended".to_string(), Value::Bool(alert.ended));
let data = NotificationData::DmCallStartEnd {
initiator_id: alert.initiator_id,
channel_id: alert.channel_id,
started_at: alert.started_at.unwrap_or_else(|| "".to_string()),
ended: alert.ended,
duration: config().await.api.livekit.call_ring_duration,
};
let msg = Message {
token: Some(payload.token),
notification: None,
data: Some(data),
android: Some(AndroidConfig {
priority: Some(AndroidMessagePriority::High),
ttl: Some(format!(
"{}s",
config().await.api.livekit.call_ring_duration
)),
..Default::default()
}),
data: Some(data.into_payload()),
..Default::default()
};
@@ -207,43 +271,21 @@ impl FcmOutboundConsumer {
}
}
if let Err(err) = resp {
match err {
FcmError::Auth => {
if let Err(err) = self
.db
.remove_push_subscription_by_session_id(&payload.session_id)
.await
{
revolt_config::capture_error(&err);
}
}
err => {
match resp {
Err(FcmError::Auth) => {
if let Err(err) = self
.db
.remove_push_subscription_by_session_id(&payload.session_id)
.await
{
revolt_config::capture_error(&err);
}
}
}
res => {
res?;
}
};
Ok(())
}
}
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for FcmOutboundConsumer {
async fn consume(
&mut self,
channel: &AmqpChannel,
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 FCM event: {err:?}");
}
}
}

140
crates/daemons/pushd/src/consumers/outbound/vapid.rs
View File

@@ -1,6 +1,6 @@
use std::collections::HashMap;
use std::{collections::HashMap, sync::Arc};
use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProperties, Deliver};
use crate::utils::Consumer;
use anyhow::{anyhow, bail, Result};
use async_trait::async_trait;
@@ -8,46 +8,57 @@ use base64::{
engine::{self},
Engine as _,
};
use lapin::{message::Delivery, Channel as AMQPChannel, Connection};
use revolt_database::{events::rabbit::*, util::format_display_name, Database};
use web_push::{
ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder,
WebPushClient, WebPushError, WebPushMessageBuilder,
};
#[derive(Clone)]
#[allow(unused)]
pub struct VapidOutboundConsumer {
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
client: IsahcWebPushClient,
pkey: Vec<u8>,
pkey: Arc<Vec<u8>>,
}
impl VapidOutboundConsumer {
pub async fn new(db: Database) -> Result<VapidOutboundConsumer> {
#[async_trait]
impl Consumer for VapidOutboundConsumer {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<AMQPChannel>,
) -> Self {
let config = revolt_config::config().await;
if config.pushd.vapid.private_key.is_empty() | config.pushd.vapid.public_key.is_empty() {
bail!("no Vapid keys present");
if config.pushd.vapid.private_key.is_empty() || config.pushd.vapid.public_key.is_empty() {
panic!("no Vapid keys present");
}
let web_push_private_key = engine::general_purpose::URL_SAFE_NO_PAD
.decode(config.pushd.vapid.private_key)
.expect("valid `VAPID_PRIVATE_KEY`");
let web_push_private_key = Arc::new(
engine::general_purpose::URL_SAFE_NO_PAD
.decode(config.pushd.vapid.private_key)
.expect("valid `VAPID_PRIVATE_KEY`"),
);
Ok(VapidOutboundConsumer {
Self {
db,
connection,
channel,
client: IsahcWebPushClient::new().unwrap(),
pkey: web_push_private_key,
})
}
}
async fn consume_event(
&mut self,
_channel: &AmqpChannel,
_deliver: Deliver,
_basic_properties: BasicProperties,
content: Vec<u8>,
) -> Result<()> {
let content = String::from_utf8(content)?;
let payload: PayloadToService = serde_json::from_str(content.as_str())?;
fn channel(&self) -> &Arc<AMQPChannel> {
&self.channel
}
async fn consume(&self, delivery: Delivery) -> Result<()> {
let payload: PayloadToService = serde_json::from_slice(&delivery.data)?;
let subscription = SubscriptionInfo {
endpoint: payload
@@ -65,10 +76,7 @@ impl VapidOutboundConsumer {
},
};
#[allow(clippy::needless_late_init)]
let payload_body: String;
match payload.notification {
let payload_body = match payload.notification {
PayloadKind::FRReceived(alert) => {
let name = alert
.from_user
@@ -83,7 +91,7 @@ impl VapidOutboundConsumer {
let mut body = HashMap::new();
body.insert("body", format!("{} sent you a friend request", name));
payload_body = serde_json::to_string(&body)?;
serde_json::to_string(&body)?
}
PayloadKind::FRAccepted(alert) => {
let name = alert
@@ -99,14 +107,10 @@ impl VapidOutboundConsumer {
let mut body = HashMap::new();
body.insert("body", format!("{} accepted your friend request", name));
payload_body = serde_json::to_string(&body)?;
}
PayloadKind::Generic(alert) => {
payload_body = serde_json::to_string(&alert)?;
}
PayloadKind::MessageNotification(alert) => {
payload_body = serde_json::to_string(&alert)?;
serde_json::to_string(&body)?
}
PayloadKind::Generic(alert) => serde_json::to_string(&alert)?,
PayloadKind::MessageNotification(alert) => serde_json::to_string(&alert)?,
PayloadKind::DmCallStartEnd(alert) => {
let initiator_name = if let Some(server_id) =
self.db.fetch_channel(&alert.channel_id).await?.server()
@@ -132,59 +136,41 @@ impl VapidOutboundConsumer {
_ => bail!("Invalid DmCallStart/End channel type"),
}
payload_body = serde_json::to_string(&body)?;
serde_json::to_string(&body)?
}
PayloadKind::BadgeUpdate(_) => {
bail!("Vapid cannot handle badge updates and they should not be sent here.");
}
}
};
match VapidSignatureBuilder::from_pem(std::io::Cursor::new(&self.pkey), &subscription) {
Ok(sig_builder) => match sig_builder.build() {
Ok(signature) => {
let mut builder = WebPushMessageBuilder::new(&subscription);
builder.set_vapid_signature(signature);
let signature = VapidSignatureBuilder::from_pem(
std::io::Cursor::new(self.pkey.as_ref()),
&subscription,
)?
.build()?;
builder.set_payload(ContentEncoding::AesGcm, payload_body.as_bytes());
let mut builder = WebPushMessageBuilder::new(&subscription);
builder.set_vapid_signature(signature);
match builder.build() {
Ok(msg) => {
if let Err(err) = self.client.send(msg).await {
if err == WebPushError::Unauthorized {
self.db
.remove_push_subscription_by_session_id(&payload.session_id)
.await?;
}
}
builder.set_payload(ContentEncoding::AesGcm, payload_body.as_bytes());
Ok(())
}
Err(err) => Err(err.into()),
}
let msg = builder.build()?;
match self.client.send(msg).await {
Err(WebPushError::Unauthorized) => {
if let Err(err) = self
.db
.remove_push_subscription_by_session_id(&payload.session_id)
.await
{
revolt_config::capture_error(&err);
}
Err(err) => Err(err.into()),
},
Err(err) => Err(err.into()),
}
}
}
}
res => {
res?;
}
};
#[allow(unused_variables)]
#[async_trait]
impl AsyncConsumer for VapidOutboundConsumer {
async fn consume(
&mut self,
channel: &AmqpChannel,
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 Vapid event: {err:?}");
}
Ok(())
}
}

248
crates/daemons/pushd/src/main.rs
View File

@@ -1,17 +1,16 @@
#[macro_use]
extern crate log;
use amqprs::{
channel::{
BasicConsumeArguments, Channel, ExchangeDeclareArguments, QueueBindArguments,
QueueDeclareArguments,
},
connection::{Connection, OpenConnectionArguments},
consumer::AsyncConsumer,
FieldTable,
use std::sync::Arc;
use lapin::{
options::{BasicConsumeOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions},
types::{AMQPValue, FieldTable},
Channel, Connection, ConnectionProperties,
};
use revolt_config::{config, Settings};
use tokio::sync::Notify;
use revolt_database::Database;
use tokio::signal::ctrl_c;
mod consumers;
mod utils;
@@ -24,6 +23,8 @@ use consumers::{
outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer},
};
use crate::utils::{Consumer, Delegate};
#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
async fn main() {
// Configure logging and environment
@@ -31,19 +32,25 @@ async fn main() {
// Setup database
let db = revolt_database::DatabaseInfo::Auto.connect().await.unwrap();
let authifier: authifier::Database;
if let Some(client) = match &db {
revolt_database::Database::Reference(_) => None,
revolt_database::Database::MongoDb(mongo) => Some(mongo),
} {
authifier =
authifier::Database::MongoDb(authifier::database::MongoDb(client.database("revolt")));
} else {
panic!("Mongo is not in use, can't connect via authifier!")
}
let config = config().await;
let mut connections: Vec<(Channel, Connection)> = Vec::new();
let connection = Arc::new(
Connection::connect(
&format!(
"amqp://{}:{}@{}:{}",
&config.rabbit.username,
&config.rabbit.password,
&config.rabbit.host,
&config.rabbit.port,
),
ConnectionProperties::default(),
)
.await
.expect("Failed to connect to RabbitMQ"),
);
let mut channels = Vec::new();
// An explainer of how this works:
// The inbound connections are on separate routing keys, such that they only receive the proper payload
@@ -54,171 +61,167 @@ async fn main() {
// This'll require some interesting shimming if we need to add more events once this is in prod (different payloads between prod and test),
// but that sounds like a problem for future us.
let config = config().await;
// inbound: generic
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<GenericConsumer>(
&db,
&connection,
&config,
&config.pushd.generic_queue,
config.pushd.get_generic_routing_key().as_str(),
&config.pushd.get_generic_routing_key(),
None,
GenericConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
// inbound: messages
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<MessageConsumer>(
&db,
&connection,
&config,
&config.pushd.message_queue,
config.pushd.get_message_routing_key().as_str(),
&config.pushd.get_message_routing_key(),
None,
MessageConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
// inbound: FR received
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<FRReceivedConsumer>(
&db,
&connection,
&config,
&config.pushd.fr_received_queue,
config.pushd.get_fr_received_routing_key().as_str(),
&config.pushd.get_fr_received_routing_key(),
None,
FRReceivedConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
// inbound: FR accepted
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<FRAcceptedConsumer>(
&db,
&connection,
&config,
&config.pushd.fr_accepted_queue,
config.pushd.get_fr_accepted_routing_key().as_str(),
&config.pushd.get_fr_accepted_routing_key(),
None,
FRAcceptedConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
// inbound: Mass Mentions
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<MassMessageConsumer>(
&db,
&connection,
&config,
&config.pushd.mass_mention_queue,
config.pushd.get_mass_mention_routing_key().as_str(),
&config.pushd.get_mass_mention_routing_key(),
None,
MassMessageConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
// inbound: Dm Calls
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<DmCallConsumer>(
&db,
&connection,
&config,
&config.pushd.dm_call_queue,
config.pushd.get_dm_call_routing_key().as_str(),
&config.pushd.get_dm_call_routing_key(),
None,
DmCallConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
if !config.pushd.apn.pkcs8.is_empty() {
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<ApnsOutboundConsumer>(
&db,
&connection,
&config,
&config.pushd.apn.queue,
&config.pushd.apn.queue,
None,
ApnsOutboundConsumer::new(db.clone()).await.unwrap(),
)
.await,
);
let mut table = FieldTable::new();
table.insert("x-message-deduplication".try_into().unwrap(), "true".into());
let mut table = FieldTable::default();
table.insert("x-message-deduplication".into(), AMQPValue::Boolean(true));
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<AckConsumer>(
&db,
&connection,
&config,
&config.pushd.ack_queue,
&config.pushd.ack_queue,
Some(table),
AckConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
}
if !config.pushd.fcm.auth_uri.is_empty() {
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<FcmOutboundConsumer>(
&db,
&connection,
&config,
&config.pushd.fcm.queue,
&config.pushd.fcm.queue,
None,
FcmOutboundConsumer::new(db.clone()).await.unwrap(),
)
.await,
)
);
}
if !config.pushd.vapid.public_key.is_empty() {
connections.push(
make_queue_and_consume(
channels.push(
make_queue_and_consume::<VapidOutboundConsumer>(
&db,
&connection,
&config,
&config.pushd.vapid.queue,
&config.pushd.vapid.queue,
None,
VapidOutboundConsumer::new(db.clone()).await.unwrap(),
)
.await,
)
);
}
let guard = Notify::new();
guard.notified().await;
ctrl_c().await.unwrap();
for (channel, conn) in connections {
channel.close().await.expect("Unable to close channel");
conn.close().await.expect("Unable to close connection");
for channel in channels {
let _ = channel.close(0, "close".into()).await;
}
}
async fn make_queue_and_consume<F>(
db: &Database,
connection: &Arc<Connection>,
config: &Settings,
queue_name: &str,
routing_key: &str,
queue_args: Option<FieldTable>,
consumer: F,
) -> (Channel, Connection)
) -> Arc<Channel>
where
F: AsyncConsumer + Send + 'static,
F: Consumer,
{
let connection = Connection::open(&OpenConnectionArguments::new(
&config.rabbit.host,
config.rabbit.port,
&config.rabbit.username,
&config.rabbit.password,
))
.await
.unwrap();
let channel = connection.open_channel(None).await.unwrap();
let channel = Arc::new(connection.create_channel().await.unwrap());
channel
.exchange_declare(
ExchangeDeclareArguments::new(&config.pushd.exchange, "direct")
.durable(true)
.finish(),
config.pushd.exchange.clone().into(),
lapin::ExchangeKind::Direct,
ExchangeDeclareOptions {
durable: true,
..Default::default()
},
FieldTable::default(),
)
.await
.expect("Failed to declare pushd exchange");
.expect("Failed to declare exchange");
let mut queue_name = queue_name.to_string();
@@ -230,35 +233,58 @@ where
let queue_name = queue_name.as_str();
let mut args = QueueDeclareArguments::new(queue_name);
args.durable(true);
if let Some(arg) = queue_args {
args.arguments(arg);
}
let args = args.finish();
_ = channel.queue_declare(args).await.unwrap().unwrap();
let args = QueueDeclareOptions {
durable: true,
..Default::default()
};
channel
.queue_bind(QueueBindArguments::new(
queue_name,
&config.pushd.exchange,
routing_key,
))
.queue_declare(queue_name.into(), args, queue_args.unwrap_or_default())
.await
.unwrap();
channel
.queue_bind(
queue_name.into(),
config.pushd.exchange.clone().into(),
routing_key.into(),
QueueBindOptions::default(),
FieldTable::default(),
)
.await
.expect(
"This probably means the revolt.notifications exchange does not exist in rabbitmq!",
);
let args = BasicConsumeArguments::new(queue_name, "")
.manual_ack(false)
.finish();
let routing_key = channel.basic_consume(consumer, args).await.unwrap();
let consumer = channel
.basic_consume(
queue_name.into(),
"".into(),
BasicConsumeOptions {
no_ack: true,
..Default::default()
},
FieldTable::default(),
)
.await
.unwrap();
info!(
"Consuming routing key {} as queue {}, tag {}",
routing_key, queue_name, routing_key
routing_key,
queue_name,
consumer.tag()
);
(channel, connection)
let delegate = Delegate(
F::create(
db.clone(),
connection.clone(),
channel.clone(),
)
.await,
);
consumer.set_delegate(delegate);
channel
}

90
crates/daemons/pushd/src/utils/consumer.rs Normal file
View File

@@ -0,0 +1,90 @@
use std::{
future::{ready, Future},
pin::Pin,
sync::Arc,
};
use anyhow::Result;
use async_trait::async_trait;
use lapin::{
message::{Delivery, DeliveryResult},
options::BasicPublishOptions,
BasicProperties, Channel, Connection, ConsumerDelegate, Error as AMQPError,
};
use log::debug;
use revolt_database::Database;
#[async_trait]
pub trait Consumer: Clone + Send + Sync + 'static {
async fn create(
db: Database,
connection: Arc<Connection>,
channel: Arc<Channel>,
) -> Self;
fn channel(&self) -> &Arc<Channel>;
async fn consume(&self, delivery: Delivery) -> Result<()>;
async fn publish_message_with_options(
&self,
payload: &[u8],
exchange: &str,
routing_key: &str,
options: BasicPublishOptions,
properties: BasicProperties,
) -> Result<(), AMQPError> {
let channel = self.channel();
channel
.basic_publish(
exchange.into(),
routing_key.into(),
options,
payload,
properties,
)
.await?;
debug!("Sent message to queue for target {}", routing_key);
Ok(())
}
async fn publish_message(
&self,
payload: &[u8],
exchange: &str,
routing_key: &str,
) -> Result<(), AMQPError> {
self.publish_message_with_options(
payload,
exchange,
routing_key,
BasicPublishOptions::default(),
BasicProperties::default(),
)
.await
}
}
pub struct Delegate<C: Consumer>(pub C);
impl<C: Consumer> ConsumerDelegate for Delegate<C> {
fn on_new_delivery(
&self,
delivery: DeliveryResult,
) -> Pin<Box<dyn Future<Output = ()> + Send>> {
match delivery {
Ok(Some(delivery)) => {
let consumer = self.0.clone();
Box::pin(async move {
if let Err(e) = consumer.consume(delivery).await {
revolt_config::capture_anyhow(&e);
log::error!("{e:?}");
};
})
}
Ok(None) => Box::pin(ready(())),
Err(e) => Box::pin(async move { log::error!("Received bad delivery: {e:?}") }),
}
}
}

3
crates/daemons/pushd/src/utils/mod.rs
View File

@@ -1,2 +1,5 @@
mod renderer;
mod consumer;
pub use renderer::render_notification_content;
pub use consumer::{Consumer, Delegate};

50
crates/daemons/voice-ingress/Cargo.toml
View File

@@ -1,6 +1,6 @@
[package]
name = "revolt-voice-ingress"
version = "0.12.1"
version = "0.13.7"
license = "AGPL-3.0-or-later"
edition = "2021"
publish = false
@@ -9,41 +9,33 @@ publish = false
[dependencies]
# util
log = "*"
sentry = "0.31.5"
lru = "0.7.6"
ulid = "0.5.0"
redis-kiss = "0.1.4"
chrono = "0.4.15"
log = { workspace = true }
sentry = { workspace = true }
lru = { workspace = true }
ulid = { workspace = true }
redis-kiss = { workspace = true, default-features = false, features = ["tokio-runtime"] }
chrono = { workspace = true }
# Serde
serde_json = "1.0.79"
rmp-serde = "1.0.0"
serde = "1.0.136"
serde_json = { workspace = true }
rmp-serde = { workspace = true }
serde = { workspace = true }
# Http
rocket = { version = "0.5.0-rc.2", features = ["json"] }
rocket_empty = "0.1.1"
rocket = { workspace = true, features = ["json"] }
rocket_empty = { workspace = true }
# Async
futures = "0.3.21"
async-std = { version = "1.8.0", features = [
"tokio1",
"tokio02",
"attributes",
] }
futures = { workspace = true }
# Core
revolt-result = { path = "../../core/result" }
revolt-models = { path = "../../core/models" }
revolt-config = { path = "../../core/config" }
revolt-database = { path = "../../core/database", features = ["voice"] }
revolt-permissions = { path = "../../core/permissions" }
revolt-result = { workspace = true, features = ["rocket"] }
revolt-models = { workspace = true }
revolt-config = { workspace = true }
revolt-database = { workspace = true, features = ["voice"] }
revolt-permissions = { workspace = true }
# Voice
livekit-api = "0.4.4"
livekit-protocol = "0.4.0"
livekit-runtime = { version = "0.3.1", features = ["tokio"] }
# RabbitMQ
amqprs = { version = "1.7.0" }
livekit-api = { workspace = true }
livekit-protocol = { workspace = true }
livekit-runtime = { workspace = true, features = ["tokio"] }