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

feat: - track call length

Browse Source 
- move voice and video limits to config
- seperate VoiceInformation into model and db model
- fix build scripts
This commit is contained in:
Zomatree
2025-08-12 01:48:57 +01:00
parent a770d5822a
commit 860c47fbad
17 changed files with 532 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,201 @@
use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver};
use livekit_protocol::TrackType;
use revolt_database::{
events::client::EventV1,
iso8601_timestamp::Timestamp,
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,
},
Database, PartialMessage, SystemMessage,
};
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>,
node: &str,
voice_client: &State<VoiceClient>,
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.clone())
.as_channel(db)
.await?;
let voice_state = create_voice_state(channel_id, channel.server(), user_id).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.clone(),
from: moved_from,
to: channel_id.clone(),
state: voice_state,
}
.p(channel_id.clone())
.await;
} else {
EventV1::VoiceChannelJoin {
id: channel_id.clone(),
state: voice_state,
}
.p(channel_id.clone())
.await;
};
}
// 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.clone())
.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;
};
// 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()) {
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
"track_published" | "track_unpublished" => {
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.clone())
.as_channel(db)
.await?;
let user = Reference::from_unchecked(user_id.clone())
.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"
&& (track.r#type == TrackType::Data as i32
|| (track.r#type == TrackType::Video as i32
&& (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])
|| (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)))))
{
voice_client.remove_user(node, user_id, channel_id).await?;
delete_voice_state(channel_id, channel.server(), user_id).await?;
} else {
let partial = update_voice_state_tracks(
channel_id,
channel.server(),
user_id,
event.event == "track_published", // 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
}
}

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

@@ -1,15 +1,13 @@
use std::env;
use livekit_protocol::TrackType;
use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver};
use revolt_database::{
events::client::EventV1, util::reference::Reference, voice::{create_voice_state, delete_voice_state, get_user_moved_from_voice, update_voice_state_tracks, VoiceClient}, Database, DatabaseInfo
};
use rocket::{build, http::Status, post, request::{FromRequest, Outcome}, routes, Config, Request, State};
use rocket_empty::EmptyResponse;
use revolt_database::voice::VoiceClient;
use revolt_database::DatabaseInfo;
use revolt_result::Result;
use rocket::{build, routes, Config};
use std::net::Ipv4Addr;
use revolt_result::{create_error, Error, Result, ToRevoltError};
mod api;
mod guard;
#[rocket::main]
async fn main() -> Result<(), rocket::Error> {
@@ -20,7 +18,7 @@ async fn main() -> Result<(), rocket::Error> {
let _rocket = build()
.manage(database)
.manage(voice_client)
.mount("/", routes![ingress])
.mount("/", routes![api::ingress])
.configure(Config {
port: 8500,
address: Ipv4Addr::new(0, 0, 0, 0).into(),
@@ -33,131 +31,3 @@ async fn main() -> Result<(), rocket::Error> {
Ok(())
}
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
}
}
#[post("/<node>", data = "<body>")]
async fn ingress(db: &State<Database>, node: &str, voice_client: &State<VoiceClient>, 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."))?;
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() {
"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.clone())
.as_channel(db)
.await?;
let voice_state = create_voice_state(channel_id, channel.server(), user_id).await?;
if let Some(moved_from) = get_user_moved_from_voice(channel_id, user_id).await? {
EventV1::VoiceChannelMove {
user: user_id.clone(),
from: moved_from,
to: channel_id.clone(),
state: voice_state
}
.p(channel_id.clone())
.await;
} else {
EventV1::VoiceChannelJoin {
id: channel_id.clone(),
state: voice_state,
}
.p(channel_id.clone())
.await;
};
}
"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.clone())
.as_channel(db)
.await?;
delete_voice_state(channel_id, channel.server(), user_id).await?;
EventV1::VoiceChannelLeave {
id: channel_id.clone(),
user: user_id.clone(),
}
.p(channel_id.clone())
.await
}
"track_published" | "track_unpublished" => {
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.clone())
.as_channel(db)
.await?;
// remove the user if they try publish a video larger than 1080x720 or they publish data
// TODO: move to config
if event.event == "track_published"
&& (
// handle any size which goes over the limit of "1080x720" to stop people from making too tall or too wide and bypassing the limit
(track.r#type == TrackType::Video as i32 && (track.width * track.height) >= (1080 * 720))
| (track.r#type == TrackType::Data as i32)
)
{
voice_client.remove_user(node, user_id, channel_id).await?;
delete_voice_state(channel_id, channel.server(), user_id).await?;
}
let partial = update_voice_state_tracks(
channel_id,
channel.server(),
user_id,
event.event == "track_published", // 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)
}