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

feat: track join time

Browse Source
This commit is contained in:
Zomatree
2025-09-20 05:26:58 +01:00
parent 938480ccec
commit 5b50532d15
19 changed files with 230 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,7 +12,6 @@ use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
use revolt_presence::filter_online;
use revolt_result::Result;
use super::state::{Cache, State};
/// Cache Manager
@@ -217,7 +216,14 @@ impl State {
// fetch voice states for all the channels we can see
let mut voice_states = Vec::new();
for channel in &channels {
for channel in channels.iter().filter(|c| {
matches!(
c,
Channel::DirectMessage { .. }
| Channel::Group { .. }
| Channel::TextChannel { voice: Some(_), .. }
)
}) {
if let Ok(Some(voice_state)) = get_channel_voice_state(channel).await {
voice_states.push(voice_state)
}

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

@@ -153,6 +153,7 @@ auto_derived!(
Description,
Icon,
DefaultPermissions,
Voice,
}
);
@@ -424,9 +425,7 @@ impl Channel {
/// Clone this channel's server id
pub fn server(&self) -> Option<&str> {
match self {
Channel::TextChannel { server, .. } => {
Some(server)
}
Channel::TextChannel { server, .. } => Some(server),
_ => None,
}
}
@@ -535,6 +534,12 @@ impl Channel {
}
_ => {}
},
FieldsChannel::Voice => match self {
Self::TextChannel { voice, .. } => {
voice.take();
}
_ => {}
},
}
}
@@ -767,6 +772,7 @@ impl IntoDocumentPath for FieldsChannel {
FieldsChannel::Description => "description",
FieldsChannel::Icon => "icon",
FieldsChannel::DefaultPermissions => "default_permissions",
FieldsChannel::Voice => "voice",
})
}
}

2
crates/core/database/src/util/bridge/v0.rs
View File

@@ -312,6 +312,7 @@ impl From<FieldsChannel> for crate::FieldsChannel {
FieldsChannel::Description => crate::FieldsChannel::Description,
FieldsChannel::Icon => crate::FieldsChannel::Icon,
FieldsChannel::DefaultPermissions => crate::FieldsChannel::DefaultPermissions,
FieldsChannel::Voice => crate::FieldsChannel::Voice,
}
}
}
@@ -322,6 +323,7 @@ impl From<crate::FieldsChannel> for FieldsChannel {
crate::FieldsChannel::Description => FieldsChannel::Description,
crate::FieldsChannel::Icon => FieldsChannel::Icon,
crate::FieldsChannel::DefaultPermissions => FieldsChannel::DefaultPermissions,
crate::FieldsChannel::Voice => FieldsChannel::Voice,
}
}
}

75
crates/core/database/src/voice/mod.rs
View File

@@ -4,6 +4,7 @@ use crate::{
util::{permissions::DatabasePermissionQuery, reference::Reference},
Database, Server,
};
use iso8601_timestamp::{Duration, Timestamp};
use livekit_protocol::ParticipantPermission;
use redis_kiss::{get_connection as _get_connection, redis::Pipeline, AsyncCommands, Conn};
use revolt_config::FeaturesLimits;
@@ -148,10 +149,12 @@ pub async fn create_voice_state(
channel_id: &str,
server_id: Option<&str>,
user_id: &str,
joined_at: Timestamp,
) -> Result<UserVoiceState> {
let unique_key = format!("{}:{}", &user_id, server_id.unwrap_or(channel_id));
let voice_state = UserVoiceState {
joined_at,
id: user_id.to_string(),
is_receiving: true,
is_publishing: false,
@@ -163,6 +166,12 @@ pub async fn create_voice_state(
.sadd(format!("vc_members:{channel_id}"), user_id)
.sadd(format!("vc:{user_id}"), channel_id)
.set(&unique_key, channel_id)
.set(
format!("joined_at:{unique_key}"),
joined_at
.duration_since(Timestamp::UNIX_EPOCH)
.whole_milliseconds() as i64,
)
.set(
format!("is_publishing:{unique_key}"),
voice_state.is_publishing,
@@ -194,6 +203,7 @@ pub async fn delete_voice_state(
.srem(format!("vc_members:{channel_id}"), user_id)
.srem(format!("vc:{user_id}"), channel_id)
.del(&[
format!("joined_at:{unique_key}"),
format!("is_publishing:{unique_key}"),
format!("is_receiving:{unique_key}"),
format!("screensharing:{unique_key}"),
@@ -205,6 +215,35 @@ pub async fn delete_voice_state(
.to_internal_error()
}
pub async fn delete_channel_voice_state(
channel_id: &str,
server_id: Option<&str>,
user_ids: &[String],
) -> Result<()> {
let parent_id = server_id.unwrap_or(channel_id);
let mut pipeline = Pipeline::new();
pipeline.del(format!("vc_members:{channel_id}"));
for user_id in user_ids {
let unique_key = format!("{user_id}:{parent_id}");
pipeline.srem(format!("vc:{user_id}"), channel_id).del(&[
format!("joined_at:{unique_key}"),
format!("is_publishing:{unique_key}"),
format!("is_receiving:{unique_key}"),
format!("screensharing:{unique_key}"),
format!("camera:{unique_key}"),
unique_key.clone(),
]);
}
pipeline
.query_async(&mut get_connection().await?.into_inner())
.await
.to_internal_error()
}
pub async fn update_voice_state_tracks(
channel_id: &str,
server_id: Option<&str>,
@@ -285,9 +324,10 @@ pub async fn get_voice_state(
) -> Result<Option<UserVoiceState>> {
let unique_key = format!("{}:{}", user_id, server_id.unwrap_or(channel_id));
let (is_publishing, is_receiving, screensharing, camera) = get_connection()
let (joined_at, is_publishing, is_receiving, screensharing, camera) = get_connection()
.await?
.mget(&[
format!("joined_at:{unique_key}"),
format!("is_publishing:{unique_key}"),
format!("is_receiving:{unique_key}"),
format!("screensharing:{unique_key}"),
@@ -296,16 +336,29 @@ pub async fn get_voice_state(
.await
.to_internal_error()?;
match (is_publishing, is_receiving, screensharing, camera) {
(Some(is_publishing), Some(is_receiving), Some(screensharing), Some(camera)) => {
Ok(Some(v0::UserVoiceState {
id: user_id.to_string(),
is_receiving,
is_publishing,
screensharing,
camera,
}))
}
match (
joined_at,
is_publishing,
is_receiving,
screensharing,
camera,
) {
(
Some(joined_at),
Some(is_publishing),
Some(is_receiving),
Some(screensharing),
Some(camera),
) => Ok(Some(v0::UserVoiceState {
joined_at: Timestamp::UNIX_EPOCH
.checked_add(Duration::milliseconds(joined_at))
.unwrap(),
id: user_id.to_string(),
is_receiving,
is_publishing,
screensharing,
camera,
})),
_ => Ok(None),
}
}

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

@@ -1,4 +1,7 @@
use crate::{models::{Channel, User}, Database};
use crate::{
models::{Channel, User},
Database,
};
use livekit_api::{
access_token::{AccessToken, VideoGrants},
services::room::{CreateRoomOptions, RoomClient as InnerRoomClient, UpdateParticipantOptions},
@@ -76,7 +79,9 @@ impl VoiceClient {
AccessToken::with_api_key(&room.node.key, &room.node.secret)
.with_name(&format!("{}#{}", user.username, user.discriminator))
.with_identity(&user.id)
.with_metadata(&serde_json::to_string(&user.clone().into(db, None).await).to_internal_error()?)
.with_metadata(
&serde_json::to_string(&user.clone().into(db, None).await).to_internal_error()?,
)
.with_ttl(Duration::from_secs(10))
.with_grants(VideoGrants {
room_join: true,
@@ -139,4 +144,13 @@ impl VoiceClient {
.await
.to_internal_error()
}
pub async fn delete_room(&self, node: &str, channel_id: &str) -> Result<()> {
let room = self.get_node(node)?;
room.client
.delete_room(channel_id)
.await
.to_internal_error()
}
}

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

@@ -157,6 +157,7 @@ auto_derived!(
Description,
Icon,
DefaultPermissions,
Voice,
}
/// New webhook information

2
crates/core/models/src/v0/users.rs
View File

@@ -1,3 +1,4 @@
use iso8601_timestamp::Timestamp;
use once_cell::sync::Lazy;
use regex::Regex;
@@ -283,6 +284,7 @@ auto_derived_partial!(
/// Voice State information for a user
pub struct UserVoiceState {
pub id: String,
pub joined_at: Timestamp,
pub is_receiving: bool,
pub is_publishing: bool,
pub screensharing: bool,

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

@@ -3,7 +3,7 @@ use livekit_api::{access_token::TokenVerifier, webhooks::WebhookReceiver};
use livekit_protocol::TrackType;
use revolt_database::{
events::client::EventV1,
iso8601_timestamp::Timestamp,
iso8601_timestamp::{Duration, Timestamp},
util::reference::Reference,
voice::{
create_voice_state, delete_voice_state, get_call_notification_recipients,
@@ -64,7 +64,12 @@ pub async fn ingress(
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
let voice_state = create_voice_state(channel_id, channel.server(), user_id).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? {
@@ -89,16 +94,9 @@ pub async fn ingress(
if event.room.as_ref().unwrap().num_participants == 1 {
let user = Reference::from_unchecked(user_id).as_user(db).await?;
let started_at = Timestamp::now_utc();
let message_id = Ulid::from_datetime(
DateTime::from_timestamp_millis(
started_at
.duration_since(Timestamp::UNIX_EPOCH)
.whole_milliseconds() as i64,
)
.unwrap(),
)
.to_string();
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(),
@@ -127,7 +125,7 @@ pub async fn ingress(
.await?;
let recipients = get_call_notification_recipients(&channel_id, &user_id).await?;
let now = started_at.format_short().to_string();
let now = joined_at.format_short().to_string();
if let Err(e) = amqp
.dm_call_updated(&user.id, channel.id(), Some(&now), false, recipients)
@@ -165,7 +163,7 @@ pub async fn ingress(
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)
.dm_call_updated(user_id, channel_id, None, true, None)
.await
{
revolt_config::capture_internal_error!(&e);

16
crates/delta/src/routes/bots/delete.rs
View File

@@ -1,4 +1,4 @@
use revolt_database::{util::reference::Reference, Database, User};
use revolt_database::{util::reference::Reference, voice::{delete_voice_state, get_channel_node, get_user_voice_channels, VoiceClient}, Database, User};
use revolt_result::{create_error, Result};
use rocket::State;
use rocket_empty::EmptyResponse;
@@ -10,6 +10,7 @@ use rocket_empty::EmptyResponse;
#[delete("/<target>")]
pub async fn delete_bot(
db: &State<Database>,
voice_client: &State<VoiceClient>,
user: User,
target: Reference<'_>,
) -> Result<EmptyResponse> {
@@ -18,7 +19,18 @@ pub async fn delete_bot(
return Err(create_error!(NotFound));
}
bot.delete(db).await.map(|_| EmptyResponse)
bot.delete(db).await?;
for channel_id in get_user_voice_channels(&bot.id).await? {
let node = get_channel_node(&channel_id).await?.unwrap();
let channel = Reference::from_unchecked(&channel_id).as_channel(db).await?;
voice_client.remove_user(&node, &bot.id, &channel_id).await?;
delete_voice_state(&channel_id, channel.server(), &bot.id).await?;
}
Ok(EmptyResponse)
}
#[cfg(test)]

11
crates/delta/src/routes/channels/channel_delete.rs
View File

@@ -1,8 +1,7 @@
use revolt_database::{
util::{permissions::DatabasePermissionQuery, reference::Reference},
voice::{
delete_voice_state, get_channel_node, get_voice_channel_members, is_in_voice_channel,
VoiceClient,
delete_channel_voice_state, delete_voice_state, get_channel_node, get_voice_channel_members, is_in_voice_channel, VoiceClient
},
Channel, Database, PartialChannel, User, AMQP,
};
@@ -63,6 +62,7 @@ pub async fn delete(
voice_client
.remove_user(&node, &user.id, channel.id())
.await?;
delete_voice_state(channel.id(), None, &user.id).await?;
};
}
@@ -73,10 +73,9 @@ pub async fn delete(
if let Some(users) = get_voice_channel_members(channel.id()).await? {
let node = get_channel_node(channel.id()).await?.unwrap();
for user in users {
voice_client.remove_user(&node, &user, channel.id()).await?;
delete_voice_state(channel.id(), channel.server(), &user).await?;
}
voice_client.delete_room(&node, channel.id()).await?;
delete_channel_voice_state(channel.id(), channel.server(), &users).await?;
};
}
};

17
crates/delta/src/routes/channels/channel_edit.rs
View File

@@ -1,6 +1,5 @@
use revolt_database::{
util::{permissions::DatabasePermissionQuery, reference::Reference},
Channel, Database, File, PartialChannel, SystemMessage, User, AMQP,
util::{permissions::DatabasePermissionQuery, reference::Reference}, voice::{delete_channel_voice_state, get_channel_node, get_voice_channel_members, VoiceClient}, Channel, Database, File, PartialChannel, SystemMessage, User, AMQP
};
use revolt_models::v0;
use revolt_permissions::{calculate_channel_permissions, ChannelPermission};
@@ -15,6 +14,7 @@ use validator::Validate;
#[patch("/<target>", data = "<data>")]
pub async fn edit(
db: &State<Database>,
voice_client: &State<VoiceClient>,
amqp: &State<AMQP>,
user: User,
target: Reference<'_>,
@@ -214,6 +214,9 @@ pub async fn edit(
v0::FieldsChannel::Icon => {
icon.take();
}
v0::FieldsChannel::Voice => {
voice.take();
}
_ => {}
}
}
@@ -257,5 +260,15 @@ pub async fn edit(
)
.await?;
if channel.voice().is_none() {
if let Some(users) = get_voice_channel_members(channel.id()).await? {
let node = get_channel_node(channel.id()).await?.unwrap();
voice_client.delete_room(&node, channel.id()).await?;
delete_channel_voice_state(channel.id(), channel.server(), &users).await?;
};
}
Ok(Json(channel.into()))
}

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

@@ -87,6 +87,7 @@ pub async fn call(
voice_client
.remove_user(&node, &user.id, &channel_id)
.await?;
delete_voice_state(&channel_id, channel.server(), &user.id).await?;
}
} else {

2
crates/delta/src/routes/servers/ban_create.rs
View File

@@ -60,7 +60,9 @@ pub async fn ban(
// If the member is in a voice channel while banned kick them from the voice channel
if let Some(channel_id) = get_user_voice_channel_in_server(&user.id, &server.id).await? {
let node = get_channel_node(&channel_id).await?.unwrap();
voice_client.remove_user(&node, &user.id, &channel_id).await?;
delete_voice_state(&channel_id, Some(&server.id), &user.id).await?
}
}

14
crates/delta/src/routes/servers/member_remove.rs
View File

@@ -42,15 +42,17 @@ pub async fn kick(
return Err(create_error!(NotElevated));
}
member
.remove(db, &server, RemovalIntention::Kick, false)
.await?;
if let Some(channel_id) = get_user_voice_channel_in_server(&user.id, &server.id).await? {
let node = get_channel_node(&channel_id).await?.unwrap();
voice_client.remove_user(&node, &user.id, &channel_id).await?;
delete_voice_state(&channel_id, Some(&server.id), &user.id).await?;
}
member
.remove(db, &server, RemovalIntention::Kick, false)
.await
.map(|_| EmptyResponse)
delete_voice_state(&channel_id, Some(&server.id), &user.id).await?;
};
Ok(EmptyResponse)
}

60
crates/delta/src/routes/servers/permissions_set.rs
View File

@@ -1,5 +1,7 @@
use revolt_database::{
util::{permissions::DatabasePermissionQuery, reference::Reference}, voice::{sync_voice_permissions, VoiceClient}, Database, User
util::{permissions::DatabasePermissionQuery, reference::Reference},
voice::{sync_voice_permissions, VoiceClient},
Database, User,
};
use revolt_models::v0;
use revolt_permissions::{calculate_server_permissions, ChannelPermission, Override};
@@ -22,36 +24,38 @@ pub async fn set_role_permission(
let data = data.into_inner();
let mut server = target.as_server(db).await?;
if let Some((current_value, rank)) = server.roles.get(&role_id).map(|x| (x.permissions, x.rank))
{
let mut query = DatabasePermissionQuery::new(db, &user).server(&server);
let permissions = calculate_server_permissions(&mut query).await;
permissions.throw_if_lacking_channel_permission(ChannelPermission::ManagePermissions)?;
let (current_value, rank) = server
.roles
.get(&role_id)
.map(|x| (x.permissions, x.rank))
.ok_or_else(|| create_error!(NotFound))?;
// Prevent us from editing roles above us
if rank <= query.get_member_rank().unwrap_or(i64::MIN) {
return Err(create_error!(NotElevated));
}
let mut query = DatabasePermissionQuery::new(db, &user).server(&server);
let permissions = calculate_server_permissions(&mut query).await;
// Ensure we have access to grant these permissions forwards
let current_value: Override = current_value.into();
permissions
.throw_permission_override(current_value, &data.permissions)
.await?;
permissions.throw_if_lacking_channel_permission(ChannelPermission::ManagePermissions)?;
for channel_id in &server.channels {
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
sync_voice_permissions(db, voice_client, &channel, Some(&server), Some(&role_id)).await?;
};
server
.set_role_permission(db, &role_id, data.permissions.into())
.await?;
Ok(Json(server.into()))
} else {
Err(create_error!(NotFound))
// Prevent us from editing roles above us
if rank <= query.get_member_rank().unwrap_or(i64::MIN) {
return Err(create_error!(NotElevated));
}
// Ensure we have access to grant these permissions forwards
let current_value: Override = current_value.into();
permissions
.throw_permission_override(current_value, &data.permissions)
.await?;
server
.set_role_permission(db, &role_id, data.permissions.into())
.await?;
for channel_id in &server.channels {
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
sync_voice_permissions(db, voice_client, &channel, Some(&server), Some(&role_id)).await?;
};
Ok(Json(server.into()))
}

12
crates/delta/src/routes/servers/permissions_set_default.rs
View File

@@ -39,12 +39,6 @@ pub async fn set_default_server_permissions(
)
.await?;
for channel_id in &server.channels {
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
sync_voice_permissions(db, voice_client, &channel, Some(&server), None).await?;
};
server
.update(
db,
@@ -56,5 +50,11 @@ pub async fn set_default_server_permissions(
)
.await?;
for channel_id in &server.channels {
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
sync_voice_permissions(db, voice_client, &channel, Some(&server), None).await?;
};
Ok(Json(server.into()))
}

35
crates/delta/src/routes/servers/roles_delete.rs
View File

@@ -1,7 +1,7 @@
use revolt_database::{
util::{permissions::DatabasePermissionQuery, reference::Reference},
voice::{sync_voice_permissions, VoiceClient},
Database, User
Database, User,
};
use revolt_permissions::{calculate_server_permissions, ChannelPermission};
use revolt_result::{create_error, Result};
@@ -18,7 +18,7 @@ pub async fn delete(
user: User,
target: Reference<'_>,
role_id: String,
voice_client: &State<VoiceClient>
voice_client: &State<VoiceClient>,
) -> Result<EmptyResponse> {
let mut server = target.as_server(db).await?;
let mut query = DatabasePermissionQuery::new(db, &user).server(&server);
@@ -28,21 +28,22 @@ pub async fn delete(
let member_rank = query.get_member_rank().unwrap_or(i64::MIN);
if let Some(role) = server.roles.remove(&role_id) {
if role.rank <= member_rank {
return Err(create_error!(NotElevated));
}
let role = server
.roles
.remove(&role_id)
.ok_or_else(|| create_error!(NotFound))?;
for channel_id in &server.channels {
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
sync_voice_permissions(db, voice_client, &channel, Some(&server), Some(&role_id)).await?;
};
role.delete(db, &server.id, &role_id)
.await
.map(|_| EmptyResponse)
} else {
Err(create_error!(NotFound))
if role.rank <= member_rank {
return Err(create_error!(NotElevated));
}
role.delete(db, &server.id, &role_id).await?;
for channel_id in &server.channels {
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
sync_voice_permissions(db, voice_client, &channel, Some(&server), Some(&role_id)).await?;
}
Ok(EmptyResponse)
}

10
crates/delta/src/routes/servers/roles_edit_positions.rs
View File

@@ -1,6 +1,5 @@
use revolt_database::{
util::{permissions::DatabasePermissionQuery, reference::Reference},
Database, User,
util::{permissions::DatabasePermissionQuery, reference::Reference}, voice::{sync_voice_permissions, VoiceClient}, Database, User
};
use revolt_models::v0;
use revolt_permissions::{calculate_server_permissions, ChannelPermission};
@@ -14,6 +13,7 @@ use rocket::{serde::json::Json, State};
#[patch("/<target>/roles/ranks", data = "<data>")]
pub async fn edit_role_ranks(
db: &State<Database>,
voice_client: &State<VoiceClient>,
user: User,
target: Reference<'_>,
data: Json<v0::DataEditRoleRanks>,
@@ -70,6 +70,12 @@ pub async fn edit_role_ranks(
server.set_role_ordering(db, new_order).await?;
for channel_id in &server.channels {
let channel = Reference::from_unchecked(channel_id).as_channel(db).await?;
sync_voice_permissions(db, voice_client, &channel, Some(&server), None).await?;
};
Ok(Json(server.into()))
}

13
crates/delta/src/routes/servers/server_delete.rs
View File

@@ -1,6 +1,6 @@
use revolt_database::{
util::reference::Reference,
voice::{delete_voice_state, get_channel_node, get_user_voice_channel_in_server, get_voice_channel_members, VoiceClient},
voice::{delete_channel_voice_state, delete_voice_state, get_channel_node, get_user_voice_channel_in_server, get_voice_channel_members, VoiceClient},
Database, RemovalIntention, User,
};
use revolt_models::v0;
@@ -29,11 +29,10 @@ pub async fn delete(
if let Some(users) = get_voice_channel_members(channel_id).await? {
let node = get_channel_node(channel_id).await?.unwrap();
for user in users {
voice_client.remove_user(&node, &user, channel_id).await?;
delete_voice_state(channel_id, Some(&server.id), &user).await?;
}
}
voice_client.delete_room(&node, channel_id).await?;
delete_channel_voice_state(channel_id, Some(&server.id), &users).await?;
};
}
server.delete(db).await
@@ -41,7 +40,9 @@ pub async fn delete(
if let Some(channel_id) = get_user_voice_channel_in_server(&user.id, &server.id).await? {
if server.channels.iter().any(|c| c == &channel_id) {
let node = get_channel_node(&channel_id).await?.unwrap();
voice_client.remove_user(&node, &user.id, &channel_id).await?;
delete_voice_state(&channel_id, Some(&server.id), &user.id).await?;
}
};