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

Merge remote-tracking branch 'origin/main' into chore/utoipa

Browse Source
This commit is contained in:
Zomatree
2025-11-17 23:11:21 +00:00
parent 43c94b68b0
commit d198796fde
155 changed files with 34536 additions and 846 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
crates/daemons/pushd/src/consumers/inbound/dm_call.rs Normal file
View File

@@ -0,0 +1,168 @@
use std::collections::HashMap;
use crate::consumers::inbound::internal::*;
use amqprs::{
channel::{BasicPublishArguments, Channel},
connection::Connection,
consumer::AsyncConsumer,
BasicProperties, Deliver,
};
use anyhow::Result;
use async_trait::async_trait;
use log::debug;
use revolt_database::{events::rabbit::*, Database};
pub struct DmCallConsumer {
#[allow(dead_code)]
db: Database,
authifier_db: authifier::Database,
conn: Option<Connection>,
channel: Option<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 {
db,
authifier_db,
conn: None,
channel: None,
}
}
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())?;
let payload = _p.payload;
debug!("Received dm call start/stop event");
let (revolt_database::Channel::DirectMessage { recipients, .. }
| revolt_database::Channel::Group { recipients, .. }) =
self.db.fetch_channel(&payload.channel_id).await?
else {
warn!(
"Discarding dm call start/stop event for non-dm/group channel {}",
payload.channel_id
);
return Ok(());
};
let call_recipients = if let Some(user_recipients) = _p.recipients {
user_recipients
.into_iter()
.filter(|user_id| recipients.contains(user_id) && user_id != &payload.initiator_id)
.collect()
} else {
recipients
.into_iter()
.filter(|user_id| user_id != &payload.initiator_id)
.collect::<Vec<_>>()
};
let config = revolt_config::config().await;
for user_id in call_recipients {
if let Ok(sessions) = self.authifier_db.find_sessions(&user_id).await {
for session in sessions {
if let Some(sub) = session.subscription {
let mut sendable = PayloadToService {
notification: PayloadKind::DmCallStartEnd(payload.clone()),
token: sub.auth,
user_id: session.user_id,
session_id: session.id,
extras: HashMap::new(),
};
let args: BasicPublishArguments;
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("p265dh".to_string(), sub.p256dh);
sendable
.extras
.insert("endpoint".to_string(), sub.endpoint.clone());
}
let payload = serde_json::to_string(&sendable)?;
publish_message(self, payload.into(), args).await;
}
}
}
}
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:?}");
}
}
}

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

@@ -1,4 +1,5 @@
pub mod ack;
pub mod dm_call;
pub mod fr_accepted;
pub mod fr_received;
pub mod generic;

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

@@ -47,6 +47,32 @@ impl<'a> PayloadLike for MessagePayload<'a> {
}
}
#[derive(Serialize, Debug)]
struct CallStartStopPayload<'a> {
aps: APS<'a>,
#[serde(skip_serializing)]
options: NotificationOptions<'a>,
#[serde(skip_serializing)]
device_token: &'a str,
initiator_id: &'a str,
#[serde(rename = "camelCase")]
channel_id: &'a str,
#[serde(rename = "camelCase")]
started_at: &'a str,
#[serde(rename = "camelCase")]
ended: bool,
}
impl<'a> PayloadLike for CallStartStopPayload<'a> {
fn get_device_token(&self) -> &'a str {
self.device_token
}
fn get_options(&self) -> &NotificationOptions {
&self.options
}
}
// region: consumer
pub struct ApnsOutboundConsumer {
@@ -63,10 +89,11 @@ impl ApnsOutboundConsumer {
// 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, .. } | Channel::VoiceChannel { name, .. } => {
Channel::TextChannel { name, .. } => {
format!("{} in #{}", notification.author, name)
}
_ => "Unknown".to_string(),
@@ -309,6 +336,7 @@ impl ApnsOutboundConsumer {
);
resp = self.client.send(apn_payload).await;
}
PayloadKind::BadgeUpdate(badge) => {
let apn_payload = Payload {
aps: APS {
@@ -323,6 +351,35 @@ impl ApnsOutboundConsumer {
debug!("Sending badge update for user: {:}", &payload.user_id);
resp = self.client.send(apn_payload).await;
}
PayloadKind::DmCallStartEnd(alert) => {
let started_at = alert.started_at.map_or(String::new(), |f| f.clone());
let apn_payload = CallStartStopPayload {
aps: APS {
alert: None,
badge: self.get_badge_count(&payload.user_id).await,
sound: None,
thread_id: None,
content_available: None,
category: None,
mutable_content: Some(1),
url_args: None,
},
device_token: &payload.token,
options: payload_options.clone(),
initiator_id: &alert.initiator_id,
channel_id: &alert.channel_id,
started_at: &started_at,
ended: alert.ended,
};
debug!(
"Sending call start/stop notification for user: {:}",
&payload.user_id
);
resp = self.client.send(apn_payload).await;
}
}
if let Err(err) = resp {

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

@@ -5,11 +5,12 @@ use amqprs::{channel::Channel as AmqpChannel, consumer::AsyncConsumer, BasicProp
use anyhow::{anyhow, bail, Result};
use async_trait::async_trait;
use fcm_v1::{
android::AndroidConfig,
android::{AndroidConfig, AndroidMessagePriority},
auth::{Authenticator, ServiceAccountKey},
message::{Message, Notification},
Client, Error as FcmError,
};
use revolt_config::config;
use revolt_database::{events::rabbit::*, Database};
use revolt_models::v0::{Channel, PushNotification};
use serde_json::Value;
@@ -27,10 +28,11 @@ impl FcmOutboundConsumer {
// 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, .. } | Channel::VoiceChannel { name, .. } => {
Channel::TextChannel { name, .. } => {
format!("{} in #{}", notification.author, name)
}
_ => "Unknown".to_string(),
@@ -169,6 +171,37 @@ impl FcmOutboundConsumer {
resp = self.client.send(&msg).await;
}
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 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()
}),
..Default::default()
};
resp = self.client.send(&msg).await;
}
PayloadKind::BadgeUpdate(_) => {
bail!("FCM cannot handle badge updates and they should not be sent here.");
}

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

@@ -8,7 +8,7 @@ use base64::{
engine::{self},
Engine as _,
};
use revolt_database::{events::rabbit::*, Database};
use revolt_database::{events::rabbit::*, util::format_display_name, Database};
use web_push::{
ContentEncoding, IsahcWebPushClient, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder,
WebPushClient, WebPushError, WebPushMessageBuilder,
@@ -107,6 +107,33 @@ impl VapidOutboundConsumer {
PayloadKind::MessageNotification(alert) => {
payload_body = 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()
{
format_display_name(&self.db, &alert.initiator_id, Some(server_id)).await
} else {
format_display_name(&self.db, &alert.initiator_id, None).await
}?;
let channel = self.db.fetch_channel(&alert.channel_id).await?;
let mut body = HashMap::new();
match channel {
revolt_database::Channel::DirectMessage { .. } => {
body.insert("body", format!("{} is calling you", initiator_name));
}
revolt_database::Channel::Group { name, .. } => {
body.insert(
"body",
format!("{} is calling your group, {}", initiator_name, name),
);
}
_ => bail!("Invalid DmCallStart/End channel type"),
}
payload_body = serde_json::to_string(&body)?;
}
PayloadKind::BadgeUpdate(_) => {
bail!("Vapid cannot handle badge updates and they should not be sent here.");
}

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

@@ -16,8 +16,9 @@ use tokio::sync::Notify;
mod consumers;
use consumers::{
inbound::{
ack::AckConsumer, fr_accepted::FRAcceptedConsumer, fr_received::FRReceivedConsumer,
generic::GenericConsumer, mass_mention::MassMessageConsumer, message::MessageConsumer,
ack::AckConsumer, dm_call::DmCallConsumer, fr_accepted::FRAcceptedConsumer,
fr_received::FRReceivedConsumer, generic::GenericConsumer,
mass_mention::MassMessageConsumer, message::MessageConsumer,
},
outbound::{apn::ApnsOutboundConsumer, fcm::FcmOutboundConsumer, vapid::VapidOutboundConsumer},
};
@@ -102,6 +103,7 @@ async fn main() {
.await,
);
// inbound: Mass Mentions
connections.push(
make_queue_and_consume(
&config,
@@ -113,6 +115,18 @@ async fn main() {
.await,
);
// inbound: Dm Calls
connections.push(
make_queue_and_consume(
&config,
&config.pushd.dm_call_queue,
config.pushd.get_dm_call_routing_key().as_str(),
None,
DmCallConsumer::new(db.clone(), authifier.clone()),
)
.await,
);
if !config.pushd.apn.pkcs8.is_empty() {
connections.push(
make_queue_and_consume(

1
crates/daemons/voice-ingress/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
/target

48
crates/daemons/voice-ingress/Cargo.toml Normal file
View File

@@ -0,0 +1,48 @@
[package]
name = "revolt-voice-ingress"
version = "0.7.1"
license = "AGPL-3.0-or-later"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
# util
log = "*"
sentry = "0.31.5"
lru = "0.7.6"
ulid = "0.5.0"
redis-kiss = "0.1.4"
chrono = "0.4.15"
# Serde
serde_json = "1.0.79"
rmp-serde = "1.0.0"
serde = "1.0.136"
# Http
rocket = { version = "0.5.0-rc.2", features = ["json"] }
rocket_empty = "0.1.1"
# Async
futures = "0.3.21"
async-std = { version = "1.8.0", features = [
"tokio1",
"tokio02",
"attributes",
] }
# 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" }
# 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" }

11
crates/daemons/voice-ingress/Dockerfile Normal file
View File

@@ -0,0 +1,11 @@
# Build Stage
FROM ghcr.io/stoatchat/base:latest AS builder
FROM debian:12 AS debian
# Bundle Stage
FROM gcr.io/distroless/cc-debian12:nonroot
COPY --from=builder /home/rust/src/target/release/revolt-voice-ingress ./
COPY --from=debian /usr/bin/uname /usr/bin/uname
USER nonroot
CMD ["./revolt-voice-ingress"]

274
crates/daemons/voice-ingress/src/api.rs Normal file
View File

@@ -0,0 +1,274 @@
use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver};
use livekit_protocol::TrackType;
use revolt_database::{
events::client::EventV1,
iso8601_timestamp::{Duration, Timestamp},
util::reference::Reference,
voice::{
create_voice_state, delete_voice_state,
get_user_moved_from_voice, get_user_moved_to_voice,
update_voice_state_tracks, VoiceClient,
},
Database, AMQP,
};
use revolt_result::{Result, ToRevoltError};
use rocket::{post, State};
use rocket_empty::EmptyResponse;
use crate::guard::AuthHeader;
#[post("/<node>", data = "<body>")]
pub async fn ingress(
db: &State<Database>,
voice_client: &State<VoiceClient>,
_amqp: &State<AMQP>,
node: &str,
auth_header: AuthHeader<'_>,
body: &str,
) -> Result<EmptyResponse> {
log::debug!("received event: {body:?}");
let config = revolt_config::config().await;
let node_info = config
.api
.livekit
.nodes
.get(node)
.to_internal_error()
.inspect_err(|_| {
log::error!("Unknown node {node}, make sure livekit has the correct node name set and matches `hosts.livekit` and `api.livekit.nodes` in the Revolt config.")
})?;
let webhook_receiver = WebhookReceiver::new(TokenVerifier::with_api_key(
&node_info.key,
&node_info.secret,
));
let event = webhook_receiver
.receive(body, &auth_header)
.to_internal_error()?;
let channel_id = event.room.as_ref().map(|r| &r.name);
let user_id = event.participant.as_ref().map(|r| &r.identity);
match event.event.as_str() {
// User joined a channel
"participant_joined" => {
let channel_id = channel_id.to_internal_error()?;
let user_id = user_id.to_internal_error()?;
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
let joined_at = Timestamp::UNIX_EPOCH
.checked_add(Duration::seconds(event.created_at))
.unwrap();
let voice_state =
create_voice_state(channel_id, channel.server(), user_id, joined_at).await?;
// Only publish one event when a user is moved from one channel to another.
if let Some(moved_from) = get_user_moved_to_voice(channel_id, user_id).await? {
EventV1::VoiceChannelMove {
user: user_id.to_string(),
from: moved_from,
to: channel_id.to_string(),
state: voice_state,
}
.p(channel_id.to_string())
.await;
} else {
EventV1::VoiceChannelJoin {
id: channel_id.to_string(),
state: voice_state,
}
.p(channel_id.to_string())
.await;
};
// TODO: fix `num_participants` being incorrect sometimes see (#457)
// First user who joined - send call started system message.
// if event.room.as_ref().unwrap().num_participants == 1 {
// let user = Reference::from_unchecked(user_id).as_user(db).await?;
// let message_id =
// Ulid::from_datetime(DateTime::from_timestamp_secs(event.created_at).unwrap())
// .to_string();
// let mut call_started_message = SystemMessage::CallStarted {
// by: user_id.to_string(),
// finished_at: None,
// }
// .into_message(channel.id().to_string());
// call_started_message.id = message_id;
// set_channel_call_started_system_message(channel.id(), &call_started_message.id)
// .await?;
// call_started_message
// .send(
// db,
// Some(amqp),
// v0::MessageAuthor::System {
// username: &user.username,
// avatar: user.avatar.as_ref().map(|file| file.id.as_ref()),
// },
// None,
// None,
// &channel,
// false,
// )
// .await?;
// let recipients = get_call_notification_recipients(&channel_id, &user_id).await?;
// let now = joined_at.format_short().to_string();
// if let Err(e) = amqp
// .dm_call_updated(&user.id, channel.id(), Some(&now), false, recipients)
// .await
// {
// revolt_config::capture_error(&e);
// }
// }
}
// User left a channel
"participant_left" => {
let channel_id = channel_id.to_internal_error()?;
let user_id = user_id.to_internal_error()?;
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
delete_voice_state(channel_id, channel.server(), user_id).await?;
// Dont send leave event when a user is moved
if get_user_moved_from_voice(channel_id, user_id)
.await?
.is_none()
{
EventV1::VoiceChannelLeave {
id: channel_id.clone(),
user: user_id.clone(),
}
.p(channel_id.clone())
.await;
};
// See above for why this is commented out
// // Update CallStarted system message if everyone has left with the end time
// let members = get_voice_channel_members(channel_id).await?;
// if members.is_none_or(|m| m.is_empty()) {
// // The channel is empty so send out an "end" message for ringing
// if let Err(e) = amqp
// .dm_call_updated(user_id, channel_id, None, true, None)
// .await
// {
// revolt_config::capture_internal_error!(&e);
// }
// if let Some(system_message_id) =
// take_channel_call_started_system_message(channel_id).await?
// {
// // Could have been deleted
// if let Ok(mut message) = Reference::from_unchecked(&system_message_id)
// .as_message(db)
// .await
// {
// if let Some(SystemMessage::CallStarted { finished_at, .. }) =
// &mut message.system
// {
// *finished_at = Some(Timestamp::now_utc());
// message
// .update(
// db,
// PartialMessage {
// system: message.system.clone(),
// ..Default::default()
// },
// Vec::new(),
// )
// .await?;
// } else {
// log::error!("Broken State: Call started message ID ({}) does not contain a CallStarted system message.", &message.id)
// }
// };
// };
// }
}
// Audio/video track was started/stopped/unmuted/muted
"track_published" | "track_unpublished" | "track_unmuted" | "track_muted" => {
let channel_id = channel_id.to_internal_error()?;
let user_id = user_id.to_internal_error()?;
let track = event.track.as_ref().to_internal_error()?;
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
let user = Reference::from_unchecked(user_id).as_user(db).await?;
let user_limits = user.limits().await;
// forbid any size which goes over the limit and also limit the aspect ratio to stop people from making too tall or too wide and bypassing the limit.
// TODO: figure out how to track audio stream quality
if event.event == "track_published" {
let mut disconnect = false;
if track.r#type == TrackType::Data as i32 {
log::debug!("User published data");
disconnect = true;
};
if track.r#type == TrackType::Video as i32 {
if user_limits.video_resolution[0] != 0
&& user_limits.video_resolution[1] != 0
&& track.width * track.height
> user_limits.video_resolution[0] * user_limits.video_resolution[1]
{
log::debug!("User published video with out of bounds resolution");
disconnect = true;
};
if user_limits.video_aspect_ratio[0] != user_limits.video_aspect_ratio[1]
&& !(user_limits.video_aspect_ratio[0]..=user_limits.video_aspect_ratio[1])
.contains(&(track.width as f32 / track.height as f32))
{
log::debug!("User published video with out of bounds aspect ratio");
disconnect = true;
};
};
if disconnect {
log::debug!("Removing user {user_id} from channel {channel_id} {event:?} due to forbidden track.");
let _ = voice_client.remove_user(node, user_id, channel_id).await;
delete_voice_state(channel_id, channel.server(), user_id).await?;
return Ok(EmptyResponse);
};
};
let partial = update_voice_state_tracks(
channel_id,
channel.server(),
user_id,
event.event == "track_published" || event.event == "track_unmuted", // to avoid duplicating this entire case twice
track.source,
)
.await?;
EventV1::UserVoiceStateUpdate {
id: user_id.clone(),
channel_id: channel_id.clone(),
data: partial,
}
.p(channel_id.clone())
.await;
}
_ => {}
};
Ok(EmptyResponse)
}

28
crates/daemons/voice-ingress/src/guard.rs Normal file
View File

@@ -0,0 +1,28 @@
use revolt_result::{create_error, Error};
use rocket::{
http::Status,
request::{FromRequest, Outcome},
Request,
};
pub struct AuthHeader<'a>(&'a str);
#[rocket::async_trait]
impl<'r> FromRequest<'r> for AuthHeader<'r> {
type Error = Error;
async fn from_request(request: &'r Request<'_>) -> Outcome<Self, Self::Error> {
match request.headers().get("Authorization").next() {
Some(token) => Outcome::Success(Self(token)),
None => Outcome::Error((Status::Unauthorized, create_error!(NotAuthenticated))),
}
}
}
impl std::ops::Deref for AuthHeader<'_> {
type Target = str;
fn deref(&self) -> &Self::Target {
self.0
}
}

37
crates/daemons/voice-ingress/src/main.rs Normal file
View File

@@ -0,0 +1,37 @@
use std::env;
use revolt_database::DatabaseInfo;
use revolt_database::{voice::VoiceClient, AMQP};
use revolt_result::Result;
use rocket::{build, routes, Config};
use std::net::Ipv4Addr;
mod api;
mod guard;
#[rocket::main]
async fn main() -> Result<(), rocket::Error> {
revolt_config::configure!(voice_ingress);
let amqp = AMQP::new_auto().await;
let database = DatabaseInfo::Auto.connect().await.unwrap();
let voice_client = VoiceClient::from_revolt_config().await;
let _rocket = build()
.manage(database)
.manage(voice_client)
.manage(amqp)
.mount("/", routes![api::ingress])
.configure(Config {
port: 8500,
address: Ipv4Addr::new(0, 0, 0, 0).into(),
..Default::default()
})
.ignite()
.await?
.launch()
.await?;
Ok(())
}