feat: voice chats v2 (#414)
* initial livekit support fix up code undo changes to compose file add back .env.example * move voice states to global * track current state * more stuff * fix redis key inconsistancy * split voice ops into its own library * feat(livekit): more permission handling * feat(livekit): push unfinished code * fix: include voice states in servercreate event feat: call started system message in dms * feat: implement missing permission syncs * fix: remove locked rocket version * fix: remove local testing values from config * chore: switch to ids for parameter * feat: make Channel::server return a reference * feat: support multiple voice nodes * fix: expose list of nodes * fix: respond with the url when joining a voice channel * feat: being moved between voice channels * fix: use existing node if someone is already in the voice channel * refactor: Remove VoiceChannel * chore: add pushd to debug image script * refactor: add error messages for Rabbit issues * docs: add some notes on overrides; fix defaults * fix: ensure limit is always at least 1 when sent to database * fix: change permission check for fetching channel webhooks * chore: mount API at /0.8/ as well * fix: ensure ratelimiter adjusts for version prefix * chore: bump version to 0.8.2 * chore: disable LTO because it likely overloads GitHub worker * fix: don't allow users to time themselves out closes #376 * fix: match new error type for status code * feat: init crond crate * feat: file deletion implementation * chore: fix version references * feat: add crond container definitions * chore: clarify cargo deny / tomls * feat: add company information to email footers * chore: bump version to 0.8.3 * fix: don't bump the lockfile version * fix: include `production` default value for config * chore: ignore python venv and dev script for user generation Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * test: Add tests for invite/fetch Signed-off-by: phazeschift <51278042+phazeschift@users.noreply.github.com> * test: add utilities for common setup Signed-off-by: phazeschift <51278042+phazeschift@users.noreply.github.com> * test: add delete server channel test Signed-off-by: phazeschift <51278042+phazeschift@users.noreply.github.com> * feat: allow to set an icon when creating a group Signed-off-by: Nils Ponsard <nilsponsard@gmail.com> * fix: allow message pinning and unpinning for DM-like channels Signed-off-by: Bradlee Barnes <69256931+StupidRepo@users.noreply.github.com> * fix: remove SavedMessages and Group due to given permissions Signed-off-by: Bradlee Barnes <69256931+StupidRepo@users.noreply.github.com> * feat: allow bots to manage emojis (#407) * Remove bot check on emoji create Signed-off-by: Builderb <builderbgamer@gmail.com> * Remove bot check on emoji delete Signed-off-by: Builderb <builderbgamer@gmail.com> --------- Signed-off-by: Builderb <builderbgamer@gmail.com> * feat: add option to send message with missing replies Signed-off-by: ShaksterNano <54268387+shaksternano@users.noreply.github.com> * chore: bump version to 0.8.4 chore: enable webhooks in test builds * ci: downgrade lockfile * feat: Add Mass Mentions to the backend (#394) * feat: create base of push daemon Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * Add outbound senders * Make web_push send to rabbit instead (temp stuff) * feat: stability and friend requests * make vapid fr stuff not suck * swap naming of queue * move pushd into daemons folder * fix cargo file for move into daemons folder * feat: probably working fcm push notifs * comment out fcm webpush stuff since the config keys dont exist * fix fcm, name queues according to their prod status and configure routing keys * add pushd to docker * mix: Remove old code, add stuff to pushd * fix: lockfile * feat: update rocket to 5.0.1 * fix: fix queues and ack bugs * Move rabbit messsage processing into ack queue * chore: update readme * chore: optimizations for ack database hits * pushd flowchart * misc: update flowchart * exit dependancy hell * add rocket_impl flag to authifier * make the tests file of delta actually compile * fix: don't silence every push message * fix: don't silence all messages * add debug logging for sending data to rabbit from message events * validate mentions at a server membership level * put back that import that was actually important * minor fix to lockfile * update delta authifier * feat: proper permissions for push notifications * add unit test for mention sanitization * remove local file dependancy on authifier * update ports to proper defaults * fixTM the node bindings * Theoretically configure docker releases for pushd Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * declare exchange in pushd and delta * fix createbuckets script Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: reference db implementation Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: remove finally redundant code Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: changes Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: other changes Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: make channel name return result Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * Add role mention parsing * feat: update to mongo 3.1, add member generator. * integrate mass mentions into pushd * patch redis-rs with updated versions * feat: chunk role mentions * move permission bits to 37/38 to avoid livekit conflict * change role mention format to <%id> * fix the lockfile from merge * fix: PR change requests * feat: add tests * fix: i am a dumbass * fix: tests, again --------- Signed-off-by: IAmTomahawkx <iamtomahawkx@gmail.com> * fix: show full branch name on github webhook messages * fix: dont leak invisible presence to others * fix: add back missing early adopter badge * fix: update branch from main * fix: update branch from main * Merge remote-tracking branch 'origin/main' into livekit * feat: private livekit nodes * fix: remove node from channel voice state * fix: get_voice_state has incorrect unique_key and allowed_sources should be lowercase * chore: bump livekit dep * feat: ability to force disconnect existing voice connection * chore: cleanup code * feat: - track call length - move voice and video limits to config - seperate VoiceInformation into model and db model - fix build scripts * fix: move call system message to daemon, check max participants when creating a token to avoid giving tokens but erroring when attempting to join, check if the channel actually supports voice * fix: dont set max participants in livekit * fix: remove VoiceChannel channel type fix: calculate user overwrites correctly fix: dont include personal info in livekit user metadata fix: revoke video permissons on denied publish fix: add video to default permissions * fix: update migration * feat: Send push notifications for dm call start/end * chore: cleanup * feat: allow users to specify who is notified when starting a call * feat: track join time * merge: branch 'main' into feat/livekit * fix: include voice-ingress in build * fix: dont presume channel has a node * chore: add error logging to internal errors * chore: add debug logging to voice ingress * refactor: seperate out disconnect logic * fix: return correct error if user is already in a voice channel * fix: dont break on user not still being in channel and force disconnecting * fix: add Speak and Listen to default permissions * fix: include channel ids in UserMoveVoiceChannel * chore: add temp sentry logging system * fix: handle track_muted and track_unmuted * fix: dont set notification recipients when empty vec is passed * fix: only send UserMoveVoiceChannel to the user * fix: preserve order of replies in message (#447) Signed-off-by: Aeledfyr <aeledfyr@gmail.com> * fix: prevent timing out members which have TimeoutMembers permission * chore: update ci * fix: temporarily disable call started system messages See #457 * fix: update code from review
This commit is contained in:
Angelo Kontaxis
GitHub
168
crates/daemons/pushd/src/consumers/inbound/dm_call.rs
Normal file
168
crates/daemons/pushd/src/consumers/inbound/dm_call.rs
Normal 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,4 +1,5 @@
|
||||
pub mod ack;
|
||||
pub mod dm_call;
|
||||
pub mod fr_accepted;
|
||||
pub mod fr_received;
|
||||
pub mod generic;
|
||||
|
||||
@@ -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 ¬ification.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 {
|
||||
|
||||
@@ -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 ¬ification.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.");
|
||||
}
|
||||
|
||||
@@ -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.");
|
||||
}
|
||||
|
||||
@@ -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
1
crates/daemons/voice-ingress/.gitignore
vendored
Normal file
@@ -0,0 +1 @@
|
||||
/target
|
||||
48
crates/daemons/voice-ingress/Cargo.toml
Normal file
48
crates/daemons/voice-ingress/Cargo.toml
Normal 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
11
crates/daemons/voice-ingress/Dockerfile
Normal 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
274
crates/daemons/voice-ingress/src/api.rs
Normal 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
28
crates/daemons/voice-ingress/src/guard.rs
Normal 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
37
crates/daemons/voice-ingress/src/main.rs
Normal 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(())
|
||||
}
|
||||
Reference in New Issue
Block a user