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

split voice ops into its own library

Browse Source
This commit is contained in:
Zomatree
2024-05-13 14:09:08 +01:00
parent 675c7d6c0d
commit ffbc899792
17 changed files with 446 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
Cargo.lock generated
View File

@@ -4016,6 +4016,7 @@ dependencies = [
"revolt-permissions",
"revolt-presence",
"revolt-result",
"revolt-voice",
"rmp-serde",
"sentry",
"serde",
@@ -4118,6 +4119,7 @@ dependencies = [
"revolt-models",
"revolt-permissions",
"revolt-result",
"revolt-voice",
"revolt_rocket_okapi",
"rocket",
"rocket_authifier",
@@ -4188,6 +4190,24 @@ dependencies = [
"serde_json",
]
[[package]]
name = "revolt-voice"
version = "0.1.0"
dependencies = [
"async-std",
"futures",
"livekit-api",
"livekit-protocol",
"redis-kiss",
"revolt-config",
"revolt-database",
"revolt-models",
"revolt-permissions",
"revolt-result",
"serde",
"serde_json",
]
[[package]]
name = "revolt-voice-ingress"
version = "0.7.1"
@@ -4203,8 +4223,8 @@ dependencies = [
"revolt-database",
"revolt-models",
"revolt-permissions",
"revolt-presence",
"revolt-result",
"revolt-voice",
"rmp-serde",
"rocket",
"rocket_empty",

1
crates/bonfire/Cargo.toml
View File

@@ -40,6 +40,7 @@ revolt-config = { path = "../core/config" }
revolt-database = { path = "../core/database" }
revolt-permissions = { version = "0.7.1", path = "../core/permissions" }
revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
revolt-voice = { path = "../core/voice" }
# redis
fred = { version = "8.0.1", features = ["subscriber-client"] }

48
crates/bonfire/src/events/impl.rs
View File

@@ -7,9 +7,9 @@ use revolt_database::{
use revolt_models::v0;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_presence::filter_online;
use revolt_result::{create_error, Result};
use revolt_result::Result;
use revolt_voice::{delete_voice_state, get_voice_channel_members, get_voice_state};
use redis_kiss::{get_connection, AsyncCommands, Conn};
use super::state::{Cache, State};
@@ -127,9 +127,6 @@ impl State {
// Filter server channels by permission.
let channels = self.cache.filter_accessible_channels(db, channels).await;
let mut voice_states = Vec::new();
let mut conn = get_connection().await.unwrap();
// Append known user IDs from DMs.
for channel in &channels {
match channel {
@@ -202,8 +199,11 @@ impl State {
self.insert_subscription(channel.id().to_string());
}
// fetch voice states for all the channels we can see
let mut voice_states = Vec::new();
for channel in &channels {
if let Ok(Some(voice_state)) = self.fetch_voice_state(&mut conn, channel).await {
if let Ok(Some(voice_state)) = self.fetch_voice_state(channel).await {
voice_states.push(voice_state)
}
}
@@ -564,41 +564,21 @@ impl State {
async fn fetch_voice_state(
&self,
conn: &mut Conn,
channel: &Channel,
) -> Result<Option<v0::ChannelVoiceState>> {
let members = conn
.smembers::<_, Vec<String>>(format!("vc-members-{}", channel.id()))
.await
.map_err(|_| create_error!(InternalError))?;
let channel_or_server_id = channel.server().unwrap_or_else(|| channel.id());
let members = get_voice_channel_members(&channel.id()).await?;
if !members.is_empty() {
let mut participants = Vec::with_capacity(members.len());
for id in members {
let unique_key = format!("{channel_or_server_id}-{id}");
for user_id in members {
if let Some(voice_state) = get_voice_state(&channel.id(), channel.server().as_deref(), &user_id).await? {
participants.push(voice_state);
} else {
log::info!("Voice state not found but member in voice channel members, removing.");
let (can_publish, can_receive, screensharing, camera) = conn
.mget::<_, (bool, bool, bool, bool)>(&[
format!("can_publish-{unique_key}"),
format!("can_receive-{unique_key}"),
format!("screensharing-{unique_key}"),
format!("camera-{unique_key}"),
])
.await
.map_err(|_| create_error!(InternalError))?;
let voice_state = v0::UserVoiceState {
id,
can_receive,
can_publish,
screensharing,
camera,
};
participants.push(voice_state);
delete_voice_state(&channel.id(), channel.server().as_deref(), &user_id).await?;
}
}
Ok(Some(v0::ChannelVoiceState {

15
crates/bonfire/src/main.rs
View File

@@ -1,7 +1,9 @@
use std::env;
use std::{env, sync::Arc};
use async_std::net::TcpListener;
use revolt_presence::clear_region;
use once_cell::sync::OnceCell;
use revolt_voice::VoiceClient;
#[macro_use]
extern crate log;
@@ -12,6 +14,15 @@ pub mod events;
mod database;
mod websocket;
pub static VOICE_CLIENT: OnceCell<Arc<VoiceClient>> = OnceCell::new();
pub fn get_voice_client() -> Arc<VoiceClient> {
VOICE_CLIENT
.get()
.expect("get_voice_client called before set")
.clone()
}
#[async_std::main]
async fn main() {
// Configure requirements for Bonfire.
@@ -21,6 +32,8 @@ async fn main() {
// Clean up the current region information.
clear_region(None).await;
VOICE_CLIENT.set(Arc::new(VoiceClient::from_revolt_config().await)).unwrap();
// Setup a TCP listener to accept WebSocket connections on.
// By default, we bind to port 9000 on all interfaces.
let bind = env::var("HOST").unwrap_or_else(|_| "0.0.0.0:9000".into());

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

@@ -1,4 +1,4 @@
use std::collections::HashMap;
use std::{borrow::Cow, collections::HashMap};
use revolt_config::config;
use revolt_models::v0::{self, MessageAuthor, VoiceInformation};
@@ -438,6 +438,15 @@ impl Channel {
}
}
/// Gets this channel's voice information
pub fn voice(&self) -> Option<Cow<VoiceInformation>> {
match self {
Self::DirectMessage { .. } | Channel::VoiceChannel { .. } => Some(Cow::Owned(v0::VoiceInformation::default())),
Self::TextChannel { voice: Some(voice), .. } => Some(Cow::Borrowed(voice)),
_ => None
}
}
/// Set role permission on a channel
pub async fn set_role_permission(
&mut self,

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

@@ -131,12 +131,15 @@ auto_derived!(
pub roles: Option<Vec<String>>,
/// Timestamp this member is timed out until
pub timeout: Option<Timestamp>,
/// Fields to remove from channel object
#[cfg_attr(feature = "validator", validate(length(min = 1)))]
pub remove: Option<Vec<FieldsMember>>,
/// server-wide voice muted
pub can_publish: Option<bool>,
/// server-wide voice deafened
pub can_receive: Option<bool>,
/// voice channel to move to if already in a voice channel
pub voice_channel: Option<String>,
/// Fields to remove from channel object
#[cfg_attr(feature = "validator", validate(length(min = 1)))]
pub remove: Option<Vec<FieldsMember>>,
}
);

2
crates/core/permissions/src/models/mod.rs
View File

@@ -8,7 +8,7 @@ pub use server::*;
pub use user::*;
/// Holds a permission value to manipulate.
#[derive(Clone, Debug)]
#[derive(Copy, Clone, Debug)]
pub struct PermissionValue(u64);
impl PermissionValue {

4
crates/core/result/src/lib.rs
View File

@@ -134,7 +134,9 @@ pub enum ErrorType {
// ? Voice errors
LiveKitUnavailable,
AlreadyInVoiceChannel
AlreadyInVoiceChannel,
NotAVoiceChannel,
AlreadyConnected
}
#[macro_export]

2
crates/core/result/src/rocket.rs
View File

@@ -77,6 +77,8 @@ impl<'r> Responder<'r, 'static> for Error {
ErrorType::LiveKitUnavailable => Status::BadRequest,
ErrorType::AlreadyInVoiceChannel => Status::BadRequest,
ErrorType::NotAVoiceChannel => Status::BadRequest,
ErrorType::AlreadyConnected => Status::BadRequest
};
// Serialize the error data structure into JSON.

33
crates/core/voice/Cargo.toml Normal file
View File

@@ -0,0 +1,33 @@
[package]
name = "revolt-voice"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
# voice
livekit-api = "0.3.2"
livekit-protocol = "0.3.2"
# core
revolt-result = { path = "../result" }
revolt-models = { path = "../models" }
revolt-config = { path = "../config" }
revolt-database = { path = "../database" }
revolt-permissions = { version = "0.7.1", path = "../permissions" }
# async
futures = "0.3.21"
async-std = { version = "1.8.0", features = [
"tokio1",
"tokio02",
"attributes",
] }
# util
redis-kiss = "0.1.4"
# serde
serde_json = "1.0.79"
serde = "1.0.136"

266
crates/core/voice/src/lib.rs Normal file
View File

@@ -0,0 +1,266 @@
use livekit_api::{access_token::{AccessToken, VideoGrants}, services::room::{CreateRoomOptions, RoomClient}};
use livekit_protocol::Room;
use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands};
use revolt_database::{Channel, User};
use revolt_models::v0::{self, PartialUserVoiceState, UserVoiceState};
use revolt_permissions::{ChannelPermission, PermissionValue};
use revolt_result::{Result, ToRevoltError, create_error};
use std::time::Duration;
use revolt_config::config;
pub async fn raise_if_in_voice(user: &User, target: &str) -> Result<()> {
let mut conn = get_connection()
.await
.to_internal_error()?;
if user.bot.is_some()
// bots can be in as many voice channels as it wants so we just check if its already connected to the one its trying to connect to
&& conn.sismember(format!("vc-{}", &user.id), target)
.await
.to_internal_error()?
{
Err(create_error!(AlreadyConnected))
} else if conn.scard::<_, u32>(format!("vc-{}", &user.id)) // check if the current vc set is empty
.await
.to_internal_error()? > 0
{
Err(create_error!(AlreadyInVoiceChannel))
} else {
Ok(())
}
}
pub fn get_allowed_sources(permissions: PermissionValue) -> Vec<String> {
let mut allowed_sources = Vec::new();
if permissions.has(ChannelPermission::Speak as u64) {
allowed_sources.push("MICROPHONE".to_string())
};
if permissions.has(ChannelPermission::Video as u64) {
allowed_sources.extend([
"CAMERA".to_string(),
"SCREEN_SHARE".to_string(),
"SCREEN_SHARE_AUDIO".to_string()
]);
};
allowed_sources
}
pub async fn create_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str) -> Result<UserVoiceState> {
let unique_key = format!(
"{}-{}",
&user_id,
server_id.unwrap_or(channel_id)
);
let voice_state = UserVoiceState {
id: user_id.to_string(),
can_receive: true,
can_publish: false,
screensharing: false,
camera: false,
};
Pipeline::new()
.sadd(format!("vc-members-{channel_id}"), user_id)
.sadd(format!("vc-{user_id}"), channel_id)
.set(format!("can_publish-{unique_key}"), voice_state.can_publish)
.set(format!("can_receive-{unique_key}"), voice_state.can_receive)
.set(format!("screensharing-{unique_key}"), voice_state.screensharing)
.set(format!("camera-{unique_key}"), voice_state.camera)
.query_async(&mut get_connection()
.await
.to_internal_error()?
.into_inner())
.await
.to_internal_error()?;
Ok(voice_state)
}
pub async fn delete_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str) -> Result<()> {
let unique_key = format!(
"{}-{}",
&user_id,
server_id.unwrap_or(channel_id)
);
Pipeline::new()
.srem(format!("vc-members-{channel_id}"), user_id)
.srem(format!("vc-{user_id}"), channel_id)
.del(&[
format!("can_publish-{unique_key}"),
format!("can_receive-{unique_key}"),
format!("screensharing-{unique_key}"),
format!("camera-{unique_key}"),
])
.query_async(&mut get_connection()
.await
.to_internal_error()?
.into_inner())
.await
.to_internal_error()?;
Ok(())
}
pub async fn update_voice_state_tracks(channel_id: &str, server_id: Option<&str>, user_id: &str, added: bool, track: i32) -> Result<PartialUserVoiceState> {
let partial = match track {
/* TrackSource::Unknown */ 0 => PartialUserVoiceState::default(),
/* TrackSource::Camera */ 1 => {
PartialUserVoiceState {
camera: Some(added),
..Default::default()
}
}
/* TrackSource::Microphone */ 2 => {
PartialUserVoiceState {
can_publish: Some(added),
..Default::default()
}
}
/* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */ 3 | 4 => {
PartialUserVoiceState {
screensharing: Some(added),
..Default::default()
}
}
_ => unreachable!(),
};
update_voice_state(channel_id, server_id, user_id, &partial).await?;
Ok(partial)
}
pub async fn update_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str, partial: &PartialUserVoiceState) -> Result<()> {
let unique_key = format!(
"{}-{}",
&user_id,
server_id.unwrap_or(channel_id)
);
let mut pipeline = Pipeline::new();
if let Some(camera) = &partial.camera {
pipeline.set(format!("camera-{unique_key}"), camera);
};
if let Some(can_publish) = &partial.can_publish {
pipeline.set(format!("can_publish-{unique_key}"), can_publish);
}
if let Some(can_receive) = &partial.can_receive {
pipeline.set(format!("can_receive-{unique_key}"), can_receive);
}
if let Some(screensharing) = &partial.screensharing {
pipeline.set(format!("screensharing-{unique_key}"), screensharing);
}
pipeline.query_async(&mut get_connection()
.await
.to_internal_error()?
.into_inner()
)
.await
.to_internal_error()?;
Ok(())
}
pub async fn get_voice_channel_members(channel_id: &str) -> Result<Vec<String>> {
get_connection()
.await
.to_internal_error()?
.smembers::<_, Vec<String>>(format!("vc-members-{}", channel_id))
.await
.to_internal_error()
}
pub async fn get_voice_state(channel_id: &str, server_id: Option<&str>, user_id: &str) -> Result<Option<UserVoiceState>> {
let unique_key = format!("{}-{user_id}", server_id.unwrap_or(channel_id));
let (can_publish, can_receive, screensharing, camera) = get_connection()
.await
.to_internal_error()?
.mget::<_, (Option<bool>, Option<bool>, Option<bool>, Option<bool>)>(&[
format!("can_publish-{unique_key}"),
format!("can_receive-{unique_key}"),
format!("screensharing-{unique_key}"),
format!("camera-{unique_key}"),
])
.await
.to_internal_error()?;
match (can_publish, can_receive, screensharing, camera) {
(Some(can_publish), Some(can_receive), Some(screensharing), Some(camera)) => {
Ok(Some(v0::UserVoiceState {
id: user_id.to_string(),
can_receive,
can_publish,
screensharing,
camera,
}))
},
_ => Ok(None)
}
}
#[derive(Debug)]
pub struct VoiceClient {
rooms: RoomClient,
api_key: String,
api_secret: String
}
impl VoiceClient {
pub fn new(url: String, api_key: String, api_secret: String) -> Self {
Self {
rooms: RoomClient::with_api_key(&url, &api_key, &api_secret),
api_key,
api_secret
}
}
pub async fn from_revolt_config() -> Self {
let config = config().await;
Self::new(config.hosts.livekit, config.api.livekit.key, config.api.livekit.secret)
}
pub fn create_token(&self, user: &User, permissions: PermissionValue, channel: &Channel) -> Result<String> {
let allowed_sources = get_allowed_sources(permissions);
AccessToken::with_api_key(&self.api_key, &self.api_secret)
.with_name(&format!("{}#{}", user.username, user.discriminator))
.with_identity(&user.id)
.with_metadata(&serde_json::to_string(&user).to_internal_error()?)
.with_ttl(Duration::from_secs(10))
.with_grants(VideoGrants {
room_join: true,
can_publish_sources: allowed_sources,
room: channel.id().to_string(),
..Default::default()
})
.to_jwt()
.to_internal_error()
}
pub async fn create_room(&self, channel: &Channel) -> Result<Room> {
let voice = channel
.voice()
.ok_or_else(|| create_error!(NotAVoiceChannel))?;
self.rooms.create_room(&channel.id(), CreateRoomOptions {
max_participants: voice.max_users.unwrap_or(u32::MAX),
empty_timeout: 5 * 60, // 5 minutes
..Default::default()
})
.await
.map_err(|_| create_error!(InternalError))
}
}

1
crates/delta/Cargo.toml
View File

@@ -79,6 +79,7 @@ revolt-models = { path = "../core/models", features = [
] }
revolt-result = { path = "../core/result", features = ["rocket", "okapi"] }
revolt-permissions = { path = "../core/permissions", features = ["schemas"] }
revolt-voice = { path = "../core/voice" }
# voice
livekit-api = "0.3.2"

8
crates/delta/src/main.rs
View File

@@ -99,10 +99,8 @@ pub async fn web() -> Rocket<Build> {
)
.into();
log::info!("{:?}", &config.api.livekit);
let room_client = RoomClient::with_api_key(&config.api.livekit.url, &config.api.livekit.key, &config.api.livekit.secret);
// Voice handler
let voice_client = revolt_voice::VoiceClient::new(config.api.livekit.url.clone(), config.api.livekit.key.clone(), config.api.livekit.secret.clone());
// Configure Rocket
let rocket = rocket::build();
@@ -117,7 +115,7 @@ pub async fn web() -> Rocket<Build> {
.manage(authifier)
.manage(db)
.manage(cors.clone())
.manage(room_client)
.manage(voice_client)
.attach(util::ratelimiter::RatelimitFairing)
.attach(cors)
.configure(rocket::Config {

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

@@ -1,12 +1,9 @@
use std::borrow::Cow;
use revolt_config::config;
use revolt_models::v0;
use revolt_database::{util::{permissions::perms, reference::Reference}, Channel, Database, User};
use revolt_database::{util::{permissions::perms, reference::Reference}, Database, User};
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
use revolt_result::Result;
use revolt_voice::{VoiceClient, raise_if_in_voice};
use livekit_api::{access_token::{AccessToken, VideoGrants}, services::room::{CreateRoomOptions, RoomClient}};
use rocket::{serde::json::Json, State};
/// # Join Call
@@ -14,69 +11,20 @@ use rocket::{serde::json::Json, State};
/// Asks the voice server for a token to join the call.
#[openapi(tag = "Voice")]
#[post("/<target>/join_call")]
pub async fn call(db: &State<Database>, rooms: &State<RoomClient>, user: User, target: Reference) -> Result<Json<v0::CreateVoiceUserResponse>> {
pub async fn call(db: &State<Database>, voice: &State<VoiceClient>, user: User, target: Reference) -> Result<Json<v0::CreateVoiceUserResponse>> {
let channel = target.as_channel(db).await?;
raise_if_in_voice(&user, &channel.id()).await?;
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)?;
// TODO - decide if we should allow being in multiple voice channels for users
let token = voice.create_token(&user, current_permissions, &channel)?;
let room = voice.create_room(&channel).await?;
// if user.current_voice_channel(&channel.server().unwrap_or_else(|| channel.id()))
// .await?
// .is_some()
// {
// return Err(create_error!(AlreadyInVoiceChannel))
// }
let config = config().await;
if config.api.livekit.url.is_empty() {
return Err(create_error!(LiveKitUnavailable));
};
let voice = match &channel {
Channel::DirectMessage { .. } | Channel::VoiceChannel { .. } => Cow::Owned(v0::VoiceInformation::default()),
Channel::TextChannel { voice: Some(voice), .. } => Cow::Borrowed(voice),
_ => return Err(create_error!(CannotJoinCall))
};
let mut allowed_sources = Vec::new();
if current_permissions.has(ChannelPermission::Speak as u64) {
allowed_sources.push("MICROPHONE".to_string())
};
if current_permissions.has(ChannelPermission::Video as u64) {
allowed_sources.extend(["CAMERA".to_string(), "SCREEN_SHARE".to_string(), "SCREEN_SHARE_AUDIO".to_string()]);
};
let token = AccessToken::with_api_key(&config.api.livekit.key, &config.api.livekit.secret)
.with_name(&format!("{}#{}", user.username, user.discriminator))
.with_identity(&user.id)
.with_metadata(&serde_json::to_string(&user).map_err(|_| create_error!(InternalError))?)
.with_grants(VideoGrants {
room_join: true,
can_publish_sources: allowed_sources,
room: channel.id().to_string(),
..Default::default()
})
.to_jwt()
.inspect_err(|e| log::error!("{e:?}"))
.map_err(|_| create_error!(InternalError))?;
let room = rooms.create_room(&channel.id(), CreateRoomOptions {
max_participants: voice.max_users.unwrap_or(u32::MAX),
empty_timeout: 5 * 60, // 5 minutes
..Default::default()
})
.await
.inspect_err(|e| log::error!("{e:?}"))
.map_err(|_| create_error!(InternalError))?;
log::info!("created room {room:?}");
log::debug!("created room {}", room.name);
Ok(Json(v0::CreateVoiceUserResponse { token }))
}

55
crates/delta/src/routes/servers/member_edit.rs
View File

@@ -1,17 +1,17 @@
use std::collections::HashSet;
use futures::TryFutureExt;
use livekit_api::services::room::{RoomClient, UpdateParticipantOptions};
use livekit_protocol::ParticipantPermission;
use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands};
use revolt_database::{
util::{permissions::DatabasePermissionQuery, reference::Reference},
Database, File, PartialMember, User,
events::client::EventV1, util::{permissions::DatabasePermissionQuery, reference::Reference}, Database, File, PartialMember, User
};
use revolt_models::v0;
use revolt_models::v0::{self, PartialUserVoiceState};
use revolt_permissions::{calculate_server_permissions, ChannelPermission};
use revolt_result::{create_error, Result, ToRevoltError};
use rocket::{serde::json::Json, State};
use rocket::{form::validate::Contains, serde::json::Json, State};
use validator::Validate;
/// # Edit Member
@@ -99,6 +99,25 @@ pub async fn edit(
permissions.throw_if_lacking_channel_permission(ChannelPermission::DeafenMembers)?;
}
if let Some(new_channel) = &data.voice_channel {
permissions.throw_if_lacking_channel_permission(ChannelPermission::MoveMembers)?;
// ensure the channel we are moving them to is in the server and is a voice channel
let channel = Reference::from_unchecked(new_channel.clone())
.as_channel(db)
.await
.map_err(|_| create_error!(UnknownChannel))?;
if !channel.server().is_some_and(|v| v == member.id.server) {
Err(create_error!(UnknownChannel))?
}
if channel.voice().is_none() {
Err(create_error!(NotAVoiceChannel))?
}
}
// Resolve our ranking
let our_ranking = query.get_member_rank().unwrap_or(i64::MIN);
@@ -136,6 +155,7 @@ pub async fn edit(
remove,
can_publish,
can_receive,
voice_channel
} = data;
let mut partial = PartialMember {
@@ -171,13 +191,15 @@ pub async fn edit(
)
.await?;
if can_publish.is_some() || can_receive.is_some() {
if can_publish.is_some() || can_receive.is_some() || voice_channel.is_some() {
let mut conn = get_connection().await.to_internal_error()?;
let unique_key = format!("{}-{}", &member.id.user, &member.id.server);
// if we edit the member while they are in a voice channel we need to also update the perms
// otherwise it wont take place until they leave and rejoin
if let Some(channel) = conn
.get::<_, Option<String>>(format!("vc-{}-{}", &member.id.user, &member.id.server))
.get::<_, Option<String>>(format!("vc-{}", &unique_key))
.await
.to_internal_error()?
{
@@ -186,7 +208,7 @@ pub async fn edit(
if let Some(can_publish) = can_publish {
pipeline.set(
format!("can_publish-{}-{}", &channel, &member.id.user),
format!("can_publish-{}", unique_key),
can_publish,
);
@@ -196,13 +218,18 @@ pub async fn edit(
if let Some(can_receive) = can_receive {
pipeline.set(
format!("can_receive-{}-{}", &channel, &member.id.user),
format!("can_receive-{}", unique_key),
can_receive,
);
new_perms.can_subscribe = can_receive;
};
if let Some(new_channel) = voice_channel {
pipeline
.smove(format!("vc-members-{channel}"), format!("vc-members-{new_channel}"), &member.id.user);
};
pipeline
.query_async(&mut conn.into_inner())
.await
@@ -220,6 +247,18 @@ pub async fn edit(
)
.await
.to_internal_error()?;
EventV1::UserVoiceStateUpdate {
id: member.id.user.clone(),
channel_id: channel.clone(),
data: PartialUserVoiceState {
can_publish,
can_receive,
..Default::default()
}
}
.p(channel)
.await;
};
};

4
crates/voice-ingress/Cargo.toml
View File

@@ -36,8 +36,8 @@ revolt-result = { path = "../core/result" }
revolt-models = { path = "../core/models" }
revolt-config = { path = "../core/config" }
revolt-database = { path = "../core/database" }
revolt-permissions = { version = "0.7.1", path = "../core/permissions" }
revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
revolt-permissions = { path = "../core/permissions" }
revolt-voice = { path = "../core/voice" }
# voice
livekit-api = "0.3.2"

112
crates/voice-ingress/src/main.rs
View File

@@ -1,12 +1,11 @@
use std::env;
use redis_kiss::{get_connection, redis::Pipeline, AsyncCommands};
use livekit_protocol::WebhookEvent;
use revolt_database::{
events::client::EventV1, util::reference::Reference, Database, DatabaseInfo,
};
use revolt_models::v0::{PartialUserVoiceState, UserVoiceState};
use revolt_voice::{create_voice_state, delete_voice_state, update_voice_state_tracks, VoiceClient};
use rocket::{build, post, routes, serde::json::Json, Config, State};
use rocket_empty::EmptyResponse;
@@ -17,9 +16,11 @@ async fn main() -> Result<(), rocket::Error> {
revolt_config::configure!(voice_ingress);
let database = DatabaseInfo::Auto.connect().await.unwrap();
let voice_client = VoiceClient::from_revolt_config().await;
let _rocket = build()
.manage(database)
.manage(voice_client)
.mount("/", routes![ingress])
.configure(Config {
port: 8500,
@@ -34,13 +35,10 @@ async fn main() -> Result<(), rocket::Error> {
}
#[post("/", data = "<body>")]
async fn ingress(db: &State<Database>, body: Json<WebhookEvent>) -> Result<EmptyResponse> {
let mut conn = get_connection().await.to_internal_error()?;
async fn ingress(db: &State<Database>, voice_client: &State<VoiceClient>, body: Json<WebhookEvent>) -> Result<EmptyResponse> {
log::debug!("received event: {body:?}");
let channel_id = body.room.as_ref().map(|r| &r.name);
let user_id = body.participant.as_ref().map(|r| &r.identity);
match body.event.as_str() {
@@ -52,29 +50,7 @@ async fn ingress(db: &State<Database>, body: Json<WebhookEvent>) -> Result<Empty
.as_channel(db)
.await?;
let unique_key = format!(
"{}-{user_id}",
channel.server().unwrap_or_else(|| channel.id())
);
Pipeline::new()
.sadd(format!("vc-members-{channel_id}"), user_id)
.set(format!("vc-{unique_key}"), channel_id)
.set(format!("can_publish-{unique_key}"), true)
.set(format!("can_receive-{unique_key}"), true)
.set(format!("screensharing-{unique_key}"), false)
.set(format!("camera-{unique_key}"), false)
.query_async(&mut conn.into_inner())
.await
.to_internal_error()?;
let voice_state = UserVoiceState {
id: user_id.clone(),
can_receive: true,
can_publish: false,
screensharing: false,
camera: false,
};
let voice_state = create_voice_state(channel_id, channel.server().as_deref(), user_id).await?;
EventV1::VoiceChannelJoin {
id: channel_id.clone(),
@@ -91,24 +67,7 @@ async fn ingress(db: &State<Database>, body: Json<WebhookEvent>) -> Result<Empty
.as_channel(db)
.await?;
conn.srem::<_, _, u64>(format!("vc-members-{channel_id}"), user_id)
.await
.to_internal_error()?;
let unique_key = format!(
"{}-{user_id}",
channel.server().unwrap_or_else(|| channel.id())
);
conn.del::<_, Vec<u64>>(&[
format!("vc-{unique_key}"),
format!("can_publish-{unique_key}"),
format!("can_receive-{unique_key}"),
format!("screensharing-{unique_key}"),
format!("camera-{unique_key}"),
])
.await
.to_internal_error()?;
delete_voice_state(channel_id, channel.server().as_deref(), user_id).await?;
EventV1::VoiceChannelLeave {
id: channel_id.clone(),
@@ -118,66 +77,21 @@ async fn ingress(db: &State<Database>, body: Json<WebhookEvent>) -> Result<Empty
.await
}
"track_published" | "track_unpublished" => {
let value = body.event == "track_published"; // to avoid duplicating this entire case twice
let channel_id = channel_id.to_internal_error()?;
let user_id = user_id.to_internal_error()?;
let track = body.track.as_ref().to_internal_error()?;
let user = Reference::from_unchecked(user_id.clone())
.as_user(db)
.await?;
let channel = Reference::from_unchecked(channel_id.clone())
.as_channel(db)
.await?;
let unique_key = if user.bot.is_some() {
format!(
"{}-{user_id}",
channel.server().unwrap_or_else(|| channel.id())
)
} else {
user_id.to_string()
};
let partial = match track.source {
/* TrackSource::Unknown */ 0 => PartialUserVoiceState::default(),
/* TrackSource::Camera */
1 => {
conn.set::<_, _, String>(format!("camera-{unique_key}"), value)
.await
.to_internal_error()?;
PartialUserVoiceState {
camera: Some(value),
..Default::default()
}
}
/* TrackSource::Microphone */
2 => {
conn.set::<_, _, String>(format!("can_receive-{unique_key}"), value)
.await
.to_internal_error()?;
PartialUserVoiceState {
can_publish: Some(value),
..Default::default()
}
}
/* TrackSource::ScreenShare | TrackSource::ScreenShareAudio */
3 | 4 => {
conn.set::<_, _, String>(format!("screensharing-{unique_key}"), value)
.await
.to_internal_error()?;
PartialUserVoiceState {
screensharing: Some(value),
..Default::default()
}
}
_ => unreachable!(),
};
let partial = update_voice_state_tracks(
channel_id,
channel.server().as_deref(),
user_id,
body.event == "track_published", // to avoid duplicating this entire case twice
track.source
).await?;
EventV1::UserVoiceStateUpdate {
id: user_id.clone(),