use super::*; impl Database { /// Clears all room participants in rooms attached to a stale server. pub async fn clear_stale_room_participants( &self, room_id: RoomId, new_server_id: ServerId, ) -> Result<TransactionGuard<RefreshedRoom>> { self.room_transaction(room_id, |tx| async move { let stale_participant_filter = Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::AnsweringConnectionId.is_not_null()) .add(room_participant::Column::AnsweringConnectionServerId.ne(new_server_id)); let stale_participant_user_ids = room_participant::Entity::find() .filter(stale_participant_filter.clone()) .all(&*tx) .await? .into_iter() .map(|participant| participant.user_id) .collect::<Vec<_>>(); // Delete participants who failed to reconnect and cancel their calls. let mut canceled_calls_to_user_ids = Vec::new(); room_participant::Entity::delete_many() .filter(stale_participant_filter) .exec(&*tx) .await?; let called_participants = room_participant::Entity::find() .filter( Condition::all() .add( room_participant::Column::CallingUserId .is_in(stale_participant_user_ids.iter().copied()), ) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .all(&*tx) .await?; room_participant::Entity::delete_many() .filter( room_participant::Column::Id .is_in(called_participants.iter().map(|participant| participant.id)), ) .exec(&*tx) .await?; canceled_calls_to_user_ids.extend( called_participants .into_iter() .map(|participant| participant.user_id), ); let (channel, room) = self.get_channel_room(room_id, &tx).await?; if channel.is_none() { // Delete the room if it becomes empty. if room.participants.is_empty() { project::Entity::delete_many() .filter(project::Column::RoomId.eq(room_id)) .exec(&*tx) .await?; room::Entity::delete_by_id(room_id).exec(&*tx).await?; } }; Ok(RefreshedRoom { room, channel, stale_participant_user_ids, canceled_calls_to_user_ids, }) }) .await } /// Returns the incoming calls for user with the given ID. pub async fn incoming_call_for_user( &self, user_id: UserId, ) -> Result<Option<proto::IncomingCall>> { self.transaction(|tx| async move { let pending_participant = room_participant::Entity::find() .filter( room_participant::Column::UserId .eq(user_id) .and(room_participant::Column::AnsweringConnectionId.is_null()), ) .one(&*tx) .await?; if let Some(pending_participant) = pending_participant { let room = self.get_room(pending_participant.room_id, &tx).await?; Ok(Self::build_incoming_call(&room, user_id)) } else { Ok(None) } }) .await } /// Creates a new room. pub async fn create_room( &self, user_id: UserId, connection: ConnectionId, livekit_room: &str, ) -> Result<proto::Room> { self.transaction(|tx| async move { let room = room::ActiveModel { live_kit_room: ActiveValue::set(livekit_room.into()), ..Default::default() } .insert(&*tx) .await?; room_participant::ActiveModel { room_id: ActiveValue::set(room.id), user_id: ActiveValue::set(user_id), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), calling_user_id: ActiveValue::set(user_id), calling_connection_id: ActiveValue::set(connection.id as i32), calling_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), participant_index: ActiveValue::set(Some(0)), role: ActiveValue::set(Some(ChannelRole::Admin)), id: ActiveValue::NotSet, location_kind: ActiveValue::NotSet, location_project_id: ActiveValue::NotSet, initial_project_id: ActiveValue::NotSet, } .insert(&*tx) .await?; let room = self.get_room(room.id, &tx).await?; Ok(room) }) .await } pub async fn call( &self, room_id: RoomId, calling_user_id: UserId, calling_connection: ConnectionId, called_user_id: UserId, initial_project_id: Option<ProjectId>, ) -> Result<TransactionGuard<(proto::Room, proto::IncomingCall)>> { self.room_transaction(room_id, |tx| async move { let caller = room_participant::Entity::find() .filter( room_participant::Column::UserId .eq(calling_user_id) .and(room_participant::Column::RoomId.eq(room_id)), ) .one(&*tx) .await? .ok_or_else(|| anyhow!("user is not in the room"))?; let called_user_role = match caller.role.unwrap_or(ChannelRole::Member) { ChannelRole::Admin | ChannelRole::Member => ChannelRole::Member, ChannelRole::Guest | ChannelRole::Talker => ChannelRole::Guest, ChannelRole::Banned => return Err(anyhow!("banned users cannot invite").into()), }; room_participant::ActiveModel { room_id: ActiveValue::set(room_id), user_id: ActiveValue::set(called_user_id), answering_connection_lost: ActiveValue::set(false), participant_index: ActiveValue::NotSet, calling_user_id: ActiveValue::set(calling_user_id), calling_connection_id: ActiveValue::set(calling_connection.id as i32), calling_connection_server_id: ActiveValue::set(Some(ServerId( calling_connection.owner_id as i32, ))), initial_project_id: ActiveValue::set(initial_project_id), role: ActiveValue::set(Some(called_user_role)), id: ActiveValue::NotSet, answering_connection_id: ActiveValue::NotSet, answering_connection_server_id: ActiveValue::NotSet, location_kind: ActiveValue::NotSet, location_project_id: ActiveValue::NotSet, } .insert(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; let incoming_call = Self::build_incoming_call(&room, called_user_id) .ok_or_else(|| anyhow!("failed to build incoming call"))?; Ok((room, incoming_call)) }) .await } pub async fn call_failed( &self, room_id: RoomId, called_user_id: UserId, ) -> Result<TransactionGuard<proto::Room>> { self.room_transaction(room_id, |tx| async move { room_participant::Entity::delete_many() .filter( room_participant::Column::RoomId .eq(room_id) .and(room_participant::Column::UserId.eq(called_user_id)), ) .exec(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; Ok(room) }) .await } pub async fn decline_call( &self, expected_room_id: Option<RoomId>, user_id: UserId, ) -> Result<Option<TransactionGuard<proto::Room>>> { self.optional_room_transaction(|tx| async move { let mut filter = Condition::all() .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_null()); if let Some(room_id) = expected_room_id { filter = filter.add(room_participant::Column::RoomId.eq(room_id)); } let participant = room_participant::Entity::find() .filter(filter) .one(&*tx) .await?; let participant = if let Some(participant) = participant { participant } else if expected_room_id.is_some() { return Err(anyhow!("could not find call to decline"))?; } else { return Ok(None); }; let room_id = participant.room_id; room_participant::Entity::delete(participant.into_active_model()) .exec(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; Ok(Some((room_id, room))) }) .await } pub async fn cancel_call( &self, room_id: RoomId, calling_connection: ConnectionId, called_user_id: UserId, ) -> Result<TransactionGuard<proto::Room>> { self.room_transaction(room_id, |tx| async move { let participant = room_participant::Entity::find() .filter( Condition::all() .add(room_participant::Column::UserId.eq(called_user_id)) .add(room_participant::Column::RoomId.eq(room_id)) .add( room_participant::Column::CallingConnectionId .eq(calling_connection.id as i32), ) .add( room_participant::Column::CallingConnectionServerId .eq(calling_connection.owner_id as i32), ) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .one(&*tx) .await? .ok_or_else(|| anyhow!("no call to cancel"))?; room_participant::Entity::delete(participant.into_active_model()) .exec(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; Ok(room) }) .await } pub async fn join_room( &self, room_id: RoomId, user_id: UserId, connection: ConnectionId, ) -> Result<TransactionGuard<JoinRoom>> { self.room_transaction(room_id, |tx| async move { #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)] enum QueryChannelId { ChannelId, } let channel_id: Option<ChannelId> = room::Entity::find() .select_only() .column(room::Column::ChannelId) .filter(room::Column::Id.eq(room_id)) .into_values::<_, QueryChannelId>() .one(&*tx) .await? .ok_or_else(|| anyhow!("no such room"))?; if channel_id.is_some() { Err(anyhow!("tried to join channel call directly"))? } let participant_index = self .get_next_participant_index_internal(room_id, &tx) .await?; let result = room_participant::Entity::update_many() .filter( Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .set(room_participant::ActiveModel { participant_index: ActiveValue::Set(Some(participant_index)), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), ..Default::default() }) .exec(&*tx) .await?; if result.rows_affected == 0 { Err(anyhow!("room does not exist or was already joined"))?; } let room = self.get_room(room_id, &tx).await?; Ok(JoinRoom { room, channel: None, }) }) .await } pub async fn stale_room_connection(&self, user_id: UserId) -> Result<Option<ConnectionId>> { self.transaction(|tx| async move { let participant = room_participant::Entity::find() .filter(room_participant::Column::UserId.eq(user_id)) .one(&*tx) .await?; Ok(participant.and_then(|p| p.answering_connection())) }) .await } async fn get_next_participant_index_internal( &self, room_id: RoomId, tx: &DatabaseTransaction, ) -> Result<i32> { #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)] enum QueryParticipantIndices { ParticipantIndex, } let existing_participant_indices: Vec<i32> = room_participant::Entity::find() .filter( room_participant::Column::RoomId .eq(room_id) .and(room_participant::Column::ParticipantIndex.is_not_null()), ) .select_only() .column(room_participant::Column::ParticipantIndex) .into_values::<_, QueryParticipantIndices>() .all(tx) .await?; let mut participant_index = 0; while existing_participant_indices.contains(&participant_index) { participant_index += 1; } Ok(participant_index) } /// Returns the channel ID for the given room, if it has one. pub async fn channel_id_for_room(&self, room_id: RoomId) -> Result<Option<ChannelId>> { self.transaction(|tx| async move { let room: Option<room::Model> = room::Entity::find() .filter(room::Column::Id.eq(room_id)) .one(&*tx) .await?; Ok(room.and_then(|room| room.channel_id)) }) .await } pub(crate) async fn join_channel_room_internal( &self, room_id: RoomId, user_id: UserId, connection: ConnectionId, role: ChannelRole, tx: &DatabaseTransaction, ) -> Result<JoinRoom> { let participant_index = self .get_next_participant_index_internal(room_id, tx) .await?; // If someone has been invited into the room, accept the invite instead of inserting let result = room_participant::Entity::update_many() .filter( Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .set(room_participant::ActiveModel { participant_index: ActiveValue::Set(Some(participant_index)), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), ..Default::default() }) .exec(tx) .await?; if result.rows_affected == 0 { room_participant::Entity::insert(room_participant::ActiveModel { room_id: ActiveValue::set(room_id), user_id: ActiveValue::set(user_id), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), calling_user_id: ActiveValue::set(user_id), calling_connection_id: ActiveValue::set(connection.id as i32), calling_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), participant_index: ActiveValue::Set(Some(participant_index)), role: ActiveValue::set(Some(role)), id: ActiveValue::NotSet, location_kind: ActiveValue::NotSet, location_project_id: ActiveValue::NotSet, initial_project_id: ActiveValue::NotSet, }) .exec(tx) .await?; } let (channel, room) = self.get_channel_room(room_id, tx).await?; let channel = channel.ok_or_else(|| anyhow!("no channel for room"))?; Ok(JoinRoom { room, channel: Some(channel), }) } pub async fn rejoin_room( &self, rejoin_room: proto::RejoinRoom, user_id: UserId, connection: ConnectionId, ) -> Result<TransactionGuard<RejoinedRoom>> { let room_id = RoomId::from_proto(rejoin_room.id); self.room_transaction(room_id, |tx| async { let tx = tx; let participant_update = room_participant::Entity::update_many() .filter( Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_not_null()), ) .set(room_participant::ActiveModel { answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), ..Default::default() }) .exec(&*tx) .await?; if participant_update.rows_affected == 0 { return Err(anyhow!("room does not exist or was already joined"))?; } let mut reshared_projects = Vec::new();Lines 1–500 of 1381. Load more
use super::*; impl Database { /// Clears all room participants in rooms attached to a stale server. pub async fn clear_stale_room_participants( &self, room_id: RoomId, new_server_id: ServerId, ) -> Result<TransactionGuard<RefreshedRoom>> { self.room_transaction(room_id, |tx| async move { let stale_participant_filter = Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::AnsweringConnectionId.is_not_null()) .add(room_participant::Column::AnsweringConnectionServerId.ne(new_server_id)); let stale_participant_user_ids = room_participant::Entity::find() .filter(stale_participant_filter.clone()) .all(&*tx) .await? .into_iter() .map(|participant| participant.user_id) .collect::<Vec<_>>(); // Delete participants who failed to reconnect and cancel their calls. let mut canceled_calls_to_user_ids = Vec::new(); room_participant::Entity::delete_many() .filter(stale_participant_filter) .exec(&*tx) .await?; let called_participants = room_participant::Entity::find() .filter( Condition::all() .add( room_participant::Column::CallingUserId .is_in(stale_participant_user_ids.iter().copied()), ) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .all(&*tx) .await?; room_participant::Entity::delete_many() .filter( room_participant::Column::Id .is_in(called_participants.iter().map(|participant| participant.id)), ) .exec(&*tx) .await?; canceled_calls_to_user_ids.extend( called_participants .into_iter() .map(|participant| participant.user_id), ); let (channel, room) = self.get_channel_room(room_id, &tx).await?; if channel.is_none() { // Delete the room if it becomes empty. if room.participants.is_empty() { project::Entity::delete_many() .filter(project::Column::RoomId.eq(room_id)) .exec(&*tx) .await?; room::Entity::delete_by_id(room_id).exec(&*tx).await?; } }; Ok(RefreshedRoom { room, channel, stale_participant_user_ids, canceled_calls_to_user_ids, }) }) .await } /// Returns the incoming calls for user with the given ID. pub async fn incoming_call_for_user( &self, user_id: UserId, ) -> Result<Option<proto::IncomingCall>> { self.transaction(|tx| async move { let pending_participant = room_participant::Entity::find() .filter( room_participant::Column::UserId .eq(user_id) .and(room_participant::Column::AnsweringConnectionId.is_null()), ) .one(&*tx) .await?; if let Some(pending_participant) = pending_participant { let room = self.get_room(pending_participant.room_id, &tx).await?; Ok(Self::build_incoming_call(&room, user_id)) } else { Ok(None) } }) .await } /// Creates a new room. pub async fn create_room( &self, user_id: UserId, connection: ConnectionId, livekit_room: &str, ) -> Result<proto::Room> { self.transaction(|tx| async move { let room = room::ActiveModel { live_kit_room: ActiveValue::set(livekit_room.into()), ..Default::default() } .insert(&*tx) .await?; room_participant::ActiveModel { room_id: ActiveValue::set(room.id), user_id: ActiveValue::set(user_id), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), calling_user_id: ActiveValue::set(user_id), calling_connection_id: ActiveValue::set(connection.id as i32), calling_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), participant_index: ActiveValue::set(Some(0)), role: ActiveValue::set(Some(ChannelRole::Admin)), id: ActiveValue::NotSet, location_kind: ActiveValue::NotSet, location_project_id: ActiveValue::NotSet, initial_project_id: ActiveValue::NotSet, } .insert(&*tx) .await?; let room = self.get_room(room.id, &tx).await?; Ok(room) }) .await } pub async fn call( &self, room_id: RoomId, calling_user_id: UserId, calling_connection: ConnectionId, called_user_id: UserId, initial_project_id: Option<ProjectId>, ) -> Result<TransactionGuard<(proto::Room, proto::IncomingCall)>> { self.room_transaction(room_id, |tx| async move { let caller = room_participant::Entity::find() .filter( room_participant::Column::UserId .eq(calling_user_id) .and(room_participant::Column::RoomId.eq(room_id)), ) .one(&*tx) .await? .ok_or_else(|| anyhow!("user is not in the room"))?; let called_user_role = match caller.role.unwrap_or(ChannelRole::Member) { ChannelRole::Admin | ChannelRole::Member => ChannelRole::Member, ChannelRole::Guest | ChannelRole::Talker => ChannelRole::Guest, ChannelRole::Banned => return Err(anyhow!("banned users cannot invite").into()), }; room_participant::ActiveModel { room_id: ActiveValue::set(room_id), user_id: ActiveValue::set(called_user_id), answering_connection_lost: ActiveValue::set(false), participant_index: ActiveValue::NotSet, calling_user_id: ActiveValue::set(calling_user_id), calling_connection_id: ActiveValue::set(calling_connection.id as i32), calling_connection_server_id: ActiveValue::set(Some(ServerId( calling_connection.owner_id as i32, ))), initial_project_id: ActiveValue::set(initial_project_id), role: ActiveValue::set(Some(called_user_role)), id: ActiveValue::NotSet, answering_connection_id: ActiveValue::NotSet, answering_connection_server_id: ActiveValue::NotSet, location_kind: ActiveValue::NotSet, location_project_id: ActiveValue::NotSet, } .insert(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; let incoming_call = Self::build_incoming_call(&room, called_user_id) .ok_or_else(|| anyhow!("failed to build incoming call"))?; Ok((room, incoming_call)) }) .await } pub async fn call_failed( &self, room_id: RoomId, called_user_id: UserId, ) -> Result<TransactionGuard<proto::Room>> { self.room_transaction(room_id, |tx| async move { room_participant::Entity::delete_many() .filter( room_participant::Column::RoomId .eq(room_id) .and(room_participant::Column::UserId.eq(called_user_id)), ) .exec(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; Ok(room) }) .await } pub async fn decline_call( &self, expected_room_id: Option<RoomId>, user_id: UserId, ) -> Result<Option<TransactionGuard<proto::Room>>> { self.optional_room_transaction(|tx| async move { let mut filter = Condition::all() .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_null()); if let Some(room_id) = expected_room_id { filter = filter.add(room_participant::Column::RoomId.eq(room_id)); } let participant = room_participant::Entity::find() .filter(filter) .one(&*tx) .await?; let participant = if let Some(participant) = participant { participant } else if expected_room_id.is_some() { return Err(anyhow!("could not find call to decline"))?; } else { return Ok(None); }; let room_id = participant.room_id; room_participant::Entity::delete(participant.into_active_model()) .exec(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; Ok(Some((room_id, room))) }) .await } pub async fn cancel_call( &self, room_id: RoomId, calling_connection: ConnectionId, called_user_id: UserId, ) -> Result<TransactionGuard<proto::Room>> { self.room_transaction(room_id, |tx| async move { let participant = room_participant::Entity::find() .filter( Condition::all() .add(room_participant::Column::UserId.eq(called_user_id)) .add(room_participant::Column::RoomId.eq(room_id)) .add( room_participant::Column::CallingConnectionId .eq(calling_connection.id as i32), ) .add( room_participant::Column::CallingConnectionServerId .eq(calling_connection.owner_id as i32), ) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .one(&*tx) .await? .ok_or_else(|| anyhow!("no call to cancel"))?; room_participant::Entity::delete(participant.into_active_model()) .exec(&*tx) .await?; let room = self.get_room(room_id, &tx).await?; Ok(room) }) .await } pub async fn join_room( &self, room_id: RoomId, user_id: UserId, connection: ConnectionId, ) -> Result<TransactionGuard<JoinRoom>> { self.room_transaction(room_id, |tx| async move { #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)] enum QueryChannelId { ChannelId, } let channel_id: Option<ChannelId> = room::Entity::find() .select_only() .column(room::Column::ChannelId) .filter(room::Column::Id.eq(room_id)) .into_values::<_, QueryChannelId>() .one(&*tx) .await? .ok_or_else(|| anyhow!("no such room"))?; if channel_id.is_some() { Err(anyhow!("tried to join channel call directly"))? } let participant_index = self .get_next_participant_index_internal(room_id, &tx) .await?; let result = room_participant::Entity::update_many() .filter( Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .set(room_participant::ActiveModel { participant_index: ActiveValue::Set(Some(participant_index)), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), ..Default::default() }) .exec(&*tx) .await?; if result.rows_affected == 0 { Err(anyhow!("room does not exist or was already joined"))?; } let room = self.get_room(room_id, &tx).await?; Ok(JoinRoom { room, channel: None, }) }) .await } pub async fn stale_room_connection(&self, user_id: UserId) -> Result<Option<ConnectionId>> { self.transaction(|tx| async move { let participant = room_participant::Entity::find() .filter(room_participant::Column::UserId.eq(user_id)) .one(&*tx) .await?; Ok(participant.and_then(|p| p.answering_connection())) }) .await } async fn get_next_participant_index_internal( &self, room_id: RoomId, tx: &DatabaseTransaction, ) -> Result<i32> { #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)] enum QueryParticipantIndices { ParticipantIndex, } let existing_participant_indices: Vec<i32> = room_participant::Entity::find() .filter( room_participant::Column::RoomId .eq(room_id) .and(room_participant::Column::ParticipantIndex.is_not_null()), ) .select_only() .column(room_participant::Column::ParticipantIndex) .into_values::<_, QueryParticipantIndices>() .all(tx) .await?; let mut participant_index = 0; while existing_participant_indices.contains(&participant_index) { participant_index += 1; } Ok(participant_index) } /// Returns the channel ID for the given room, if it has one. pub async fn channel_id_for_room(&self, room_id: RoomId) -> Result<Option<ChannelId>> { self.transaction(|tx| async move { let room: Option<room::Model> = room::Entity::find() .filter(room::Column::Id.eq(room_id)) .one(&*tx) .await?; Ok(room.and_then(|room| room.channel_id)) }) .await } pub(crate) async fn join_channel_room_internal( &self, room_id: RoomId, user_id: UserId, connection: ConnectionId, role: ChannelRole, tx: &DatabaseTransaction, ) -> Result<JoinRoom> { let participant_index = self .get_next_participant_index_internal(room_id, tx) .await?; // If someone has been invited into the room, accept the invite instead of inserting let result = room_participant::Entity::update_many() .filter( Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_null()), ) .set(room_participant::ActiveModel { participant_index: ActiveValue::Set(Some(participant_index)), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), ..Default::default() }) .exec(tx) .await?; if result.rows_affected == 0 { room_participant::Entity::insert(room_participant::ActiveModel { room_id: ActiveValue::set(room_id), user_id: ActiveValue::set(user_id), answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), calling_user_id: ActiveValue::set(user_id), calling_connection_id: ActiveValue::set(connection.id as i32), calling_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), participant_index: ActiveValue::Set(Some(participant_index)), role: ActiveValue::set(Some(role)), id: ActiveValue::NotSet, location_kind: ActiveValue::NotSet, location_project_id: ActiveValue::NotSet, initial_project_id: ActiveValue::NotSet, }) .exec(tx) .await?; } let (channel, room) = self.get_channel_room(room_id, tx).await?; let channel = channel.ok_or_else(|| anyhow!("no channel for room"))?; Ok(JoinRoom { room, channel: Some(channel), }) } pub async fn rejoin_room( &self, rejoin_room: proto::RejoinRoom, user_id: UserId, connection: ConnectionId, ) -> Result<TransactionGuard<RejoinedRoom>> { let room_id = RoomId::from_proto(rejoin_room.id); self.room_transaction(room_id, |tx| async { let tx = tx; let participant_update = room_participant::Entity::update_many() .filter( Condition::all() .add(room_participant::Column::RoomId.eq(room_id)) .add(room_participant::Column::UserId.eq(user_id)) .add(room_participant::Column::AnsweringConnectionId.is_not_null()), ) .set(room_participant::ActiveModel { answering_connection_id: ActiveValue::set(Some(connection.id as i32)), answering_connection_server_id: ActiveValue::set(Some(ServerId( connection.owner_id as i32, ))), answering_connection_lost: ActiveValue::set(false), ..Default::default() }) .exec(&*tx) .await?; if participant_update.rows_affected == 0 { return Err(anyhow!("room does not exist or was already joined"))?; } let mut reshared_projects = Vec::new();