HMC/handmade-revolt-backend
6
0
Fork 0
forked from jmug/stoatchat
Code Pull Requests Actions Packages Releases 2 Activity

fix: move call system message to daemon,

Browse Source 
check max participants when creating a token to avoid giving tokens but erroring when attempting to join,
check if the channel actually supports voice
This commit is contained in:
Zomatree
2025-08-18 04:58:06 +01:00
parent 860c47fbad
commit a36b138978
8 changed files with 151 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@@ -6773,6 +6773,7 @@ dependencies = [
name = "revolt-voice-ingress"
version = "0.7.1"
dependencies = [
"amqprs",
"async-std",
"futures",
"livekit-api",

32
crates/core/database/src/models/channels/model.rs
View File

@@ -107,7 +107,7 @@ auto_derived!(
/// Voice Information for when this channel is also a voice channel
#[serde(skip_serializing_if = "Option::is_none")]
voice: Option<VoiceInformation>
voice: Option<VoiceInformation>,
},
#[deprecated = "Use TextChannel { voice } instead"]
VoiceChannel {
@@ -146,7 +146,7 @@ auto_derived!(
pub struct VoiceInformation {
/// Maximium amount of users allowed in the voice channel at once
#[serde(skip_serializing_if = "Option::is_none")]
pub max_users: Option<u32>
pub max_users: Option<usize>,
}
);
@@ -233,7 +233,7 @@ impl Channel {
default_permissions: None,
role_permissions: HashMap::new(),
nsfw: data.nsfw.unwrap_or(false),
voice: data.voice.map(|voice| voice.into())
voice: data.voice.map(|voice| voice.into()),
},
v0::LegacyServerChannelType::Voice => Channel::TextChannel {
id: id.clone(),
@@ -245,7 +245,7 @@ impl Channel {
default_permissions: None,
role_permissions: HashMap::new(),
nsfw: data.nsfw.unwrap_or(false),
voice: Some(data.voice.unwrap_or_default().into())
voice: Some(data.voice.unwrap_or_default().into()),
},
};
@@ -454,17 +454,23 @@ impl Channel {
/// Clone this channel's server id
pub fn server(&self) -> Option<&str> {
match self {
Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => Some(server),
_ => None
Channel::TextChannel { server, .. } | Channel::VoiceChannel { server, .. } => {
Some(server)
}
_ => None,
}
}
/// Gets this channel's voice information
pub fn voice(&self) -> Option<Cow<VoiceInformation>> {
match self {
Self::DirectMessage { .. } | Channel::VoiceChannel { .. } => Some(Cow::Owned(VoiceInformation::default())),
Self::TextChannel { voice: Some(voice), .. } => Some(Cow::Borrowed(voice)),
_ => None
Self::DirectMessage { .. } | Channel::VoiceChannel { .. } => {
Some(Cow::Owned(VoiceInformation::default()))
}
Self::TextChannel {
voice: Some(voice), ..
} => Some(Cow::Borrowed(voice)),
_ => None,
}
}
@@ -539,15 +545,13 @@ impl Channel {
pub fn remove_field(&mut self, field: &FieldsChannel) {
match field {
FieldsChannel::Description => match self {
Self::Group { description, .. }
| Self::TextChannel { description, .. } => {
Self::Group { description, .. } | Self::TextChannel { description, .. } => {
description.take();
}
_ => {}
},
FieldsChannel::Icon => match self {
Self::Group { icon, .. }
| Self::TextChannel { icon, .. } => {
Self::Group { icon, .. } | Self::TextChannel { icon, .. } => {
icon.take();
}
_ => {}
@@ -651,7 +655,7 @@ impl Channel {
if let Some(v) = partial.voice {
voice.replace(v);
}
},
}
Self::VoiceChannel { .. } => {}
}
}

2
crates/core/database/src/voice/voice_client.rs
View File

@@ -103,7 +103,7 @@ impl VoiceClient {
.create_room(
channel.id(),
CreateRoomOptions {
max_participants: voice.max_users.unwrap_or(0),
max_participants: voice.max_users.unwrap_or(0) as u32,
empty_timeout: 5 * 60, // 5 minutes,
..Default::default()
},

9
crates/core/models/src/v0/channels.rs
View File

@@ -111,7 +111,7 @@ auto_derived!(
/// Voice Information for when this channel is also a voice channel
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
voice: Option<VoiceInformation>
voice: Option<VoiceInformation>,
},
#[deprecated = "Use TextChannel { voice } instead"]
VoiceChannel {
@@ -159,7 +159,7 @@ auto_derived!(
/// Maximium amount of users allowed in the voice channel at once
#[cfg_attr(feature = "validator", validate(range(min = 1)))]
#[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))]
pub max_users: Option<u32>
pub max_users: Option<usize>,
}
/// Partial representation of a channel
@@ -283,7 +283,7 @@ auto_derived!(
/// Voice Information for when this channel is also a voice channel
#[serde(skip_serializing_if = "Option::is_none")]
pub voice: Option<VoiceInformation>
pub voice: Option<VoiceInformation>,
}
/// New default permissions
@@ -334,9 +334,8 @@ auto_derived!(
/// Whether to force disconnect any other existing voice connections.
///
/// useful for disconnecting on another device and joining on a new.
pub force_disconnect: Option<bool>
pub force_disconnect: Option<bool>,
}
);
impl Channel {

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

@@ -14,16 +14,16 @@ lru = "0.7.6"
ulid = "0.5.0"
redis-kiss = "0.1.4"
# serde
# Serde
serde_json = "1.0.79"
rmp-serde = "1.0.0"
serde = "1.0.136"
# http
# Http
rocket = { version = "0.5.0-rc.2", features = ["json"] }
rocket_empty = "0.1.1"
# async
# Async
futures = "0.3.21"
async-std = { version = "1.8.0", features = [
"tokio1",
@@ -31,14 +31,17 @@ async-std = { version = "1.8.0", features = [
"attributes",
] }
# core
# Core
revolt-result = { path = "../../core/result" }
revolt-models = { path = "../../core/models" }
revolt-config = { path = "../../core/config" }
revolt-database = { path = "../../core/database" }
revolt-permissions = { path = "../../core/permissions" }
# voice
# 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" }

41
crates/daemons/voice-ingress/src/api.rs
View File

@@ -6,11 +6,12 @@ use revolt_database::{
util::reference::Reference,
voice::{
create_voice_state, delete_voice_state, get_user_moved_from_voice, get_user_moved_to_voice,
get_voice_channel_members, take_channel_call_started_system_message,
update_voice_state_tracks, VoiceClient,
get_voice_channel_members, set_channel_call_started_system_message,
take_channel_call_started_system_message, update_voice_state_tracks, VoiceClient,
},
Database, PartialMessage, SystemMessage,
Database, PartialMessage, SystemMessage, AMQP,
};
use revolt_models::v0;
use revolt_result::{Result, ToRevoltError};
use rocket::{post, State};
use rocket_empty::EmptyResponse;
@@ -20,8 +21,9 @@ use crate::guard::AuthHeader;
#[post("/<node>", data = "<body>")]
pub async fn ingress(
db: &State<Database>,
node: &str,
voice_client: &State<VoiceClient>,
amqp: &State<AMQP>,
node: &str,
auth_header: AuthHeader<'_>,
body: &str,
) -> Result<EmptyResponse> {
@@ -81,6 +83,37 @@ pub async fn ingress(
.p(channel_id.clone())
.await;
};
// 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.clone())
.as_user(db)
.await?;
let mut call_started_message = SystemMessage::CallStarted {
by: user_id.clone(),
finished_at: None,
}
.into_message(channel.id().to_string());
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?;
}
}
// User left a channel
"participant_left" => {

36
crates/daemons/voice-ingress/src/main.rs
View File

@@ -1,7 +1,12 @@
use std::env;
use revolt_database::voice::VoiceClient;
use amqprs::{
channel::ExchangeDeclareArguments,
connection::{Connection, OpenConnectionArguments},
};
use revolt_config::config;
use revolt_database::DatabaseInfo;
use revolt_database::{voice::VoiceClient, AMQP};
use revolt_result::Result;
use rocket::{build, routes, Config};
use std::net::Ipv4Addr;
@@ -13,11 +18,40 @@ mod guard;
async fn main() -> Result<(), rocket::Error> {
revolt_config::configure!(voice_ingress);
let config = config().await;
let database = DatabaseInfo::Auto.connect().await.unwrap();
let voice_client = VoiceClient::from_revolt_config().await;
let connection = Connection::open(&OpenConnectionArguments::new(
&config.rabbit.host,
config.rabbit.port,
&config.rabbit.username,
&config.rabbit.password,
))
.await
.expect("Failed to connect to RabbitMQ");
let channel = connection
.open_channel(None)
.await
.expect("Failed to open RabbitMQ channel");
channel
.exchange_declare(
ExchangeDeclareArguments::new(&config.pushd.exchange, "direct")
.durable(true)
.finish(),
)
.await
.expect("Failed to declare exchange");
let amqp = AMQP::new(connection, channel);
let _rocket = build()
.manage(database)
.manage(voice_client)
.manage(amqp)
.mount("/", routes![api::ingress])
.configure(Config {
port: 8500,

82
crates/delta/src/routes/channels/voice_join.rs
View File

@@ -1,6 +1,13 @@
use revolt_config::config;
use revolt_database::{
util::{permissions::perms, reference::Reference},
voice::{
delete_voice_state, get_channel_node, get_user_voice_channels, get_voice_channel_members,
raise_if_in_voice, VoiceClient,
},
Database, User,
};
use revolt_models::v0;
use revolt_database::{util::{permissions::perms, reference::Reference}, voice::{delete_voice_state, get_channel_node, get_user_voice_channels, raise_if_in_voice, set_channel_call_started_system_message, VoiceClient}, Channel, Database, SystemMessage, User, AMQP};
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
@@ -10,43 +17,59 @@ use rocket::{serde::json::Json, State};
///
/// Asks the voice server for a token to join the call.
#[openapi(tag = "Voice")]
#[post("/<target>/join_call", data="<data>")]
#[post("/<target>/join_call", data = "<data>")]
pub async fn call(
db: &State<Database>,
amqp: &State<AMQP>,
voice_client: &State<VoiceClient>,
user: User,
target: Reference,
data: Json<v0::DataJoinCall>
data: Json<v0::DataJoinCall>,
) -> Result<Json<v0::CreateVoiceUserResponse>> {
if !voice_client.is_enabled() {
return Err(create_error!(LiveKitUnavailable))
return Err(create_error!(LiveKitUnavailable));
}
let v0::DataJoinCall {
node,
force_disconnect
force_disconnect,
} = data.into_inner();
if user.bot.is_some() && force_disconnect == Some(true) {
return Err(create_error!(IsBot))
return Err(create_error!(IsBot));
}
let config = config().await;
let channel = target.as_channel(db).await?;
let Some(voice_info) = channel.voice() else {
return Err(create_error!(NotAVoiceChannel));
};
let mut permissions = perms(db, &user).channel(&channel);
let current_permissions = calculate_channel_permissions(&mut permissions).await;
current_permissions.throw_if_lacking_channel_permission(ChannelPermission::Connect)?;
if get_voice_channel_members(channel.id())
.await?
.zip(voice_info.max_users)
.is_some_and(|(ms, max_users)| ms.len() >= max_users)
&& !current_permissions.has(ChannelPermission::ManageChannel as u64)
{
return Err(create_error!(CannotJoinCall));
}
let existing_node = get_channel_node(channel.id()).await?;
let node = existing_node.or(node)
let node = existing_node
.or(node)
.ok_or_else(|| create_error!(UnknownNode))?;
let node_host = config.hosts.livekit.get(&node)
let config = config().await;
let node_host = config
.hosts
.livekit
.get(&node)
.ok_or_else(|| create_error!(UnknownNode))?
.clone();
@@ -56,39 +79,28 @@ pub async fn call(
for channel_id in get_user_voice_channels(&user.id).await? {
let node = get_channel_node(&channel_id).await?.unwrap();
let channel = Reference::from_unchecked(channel_id.clone()).as_channel(db).await?;
let channel = Reference::from_unchecked(channel_id.clone())
.as_channel(db)
.await?;
voice_client.remove_user(&node, &user.id, &channel_id).await?;
voice_client
.remove_user(&node, &user.id, &channel_id)
.await?;
delete_voice_state(&channel_id, channel.server(), &user.id).await?;
}
} else {
raise_if_in_voice(&user, channel.id()).await?;
}
let token = voice_client.create_token(&node, &user, current_permissions, &channel).await?;
let token = voice_client
.create_token(&node, &user, current_permissions, &channel)
.await?;
let room = voice_client.create_room(&node, &channel).await?;
log::debug!("Created room {}", room.name);
let mut call_started_message = SystemMessage::CallStarted {
by: user.id.clone(),
finished_at: None
}
.into_message(channel.id().to_string());
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?;
Ok(Json(v0::CreateVoiceUserResponse { token, url: node_host.clone() }))
Ok(Json(v0::CreateVoiceUserResponse {
token,
url: node_host.clone(),
}))
}