use super::*;use rpc::{ proto::{channel_member::Kind, ChannelBufferVersion, VectorClockEntry}, ErrorCode, ErrorCodeExt,};use sea_orm::{DbBackend, TryGetableMany}; impl Database { #[cfg(test)] pub async fn all_channels(&self) -> Result<Vec<(ChannelId, String)>> { self.transaction(move |tx| async move { let mut channels = Vec::new(); let mut rows = channel::Entity::find().stream(&*tx).await?; while let Some(row) = rows.next().await { let row = row?; channels.push((row.id, row.name)); } Ok(channels) }) .await } #[cfg(test)] pub async fn create_root_channel(&self, name: &str, creator_id: UserId) -> Result<ChannelId> { Ok(self.create_channel(name, None, creator_id).await?.0.id) } #[cfg(test)] pub async fn create_sub_channel( &self, name: &str, parent: ChannelId, creator_id: UserId, ) -> Result<ChannelId> { Ok(self .create_channel(name, Some(parent), creator_id) .await? .0 .id) } /// Creates a new channel. pub async fn create_channel( &self, name: &str, parent_channel_id: Option<ChannelId>, admin_id: UserId, ) -> Result<(channel::Model, Option<channel_member::Model>)> { let name = Self::sanitize_channel_name(name)?; self.transaction(move |tx| async move { let mut parent = None; let mut membership = None; if let Some(parent_channel_id) = parent_channel_id { let parent_channel = self.get_channel_internal(parent_channel_id, &tx).await?; self.check_user_is_channel_admin(&parent_channel, admin_id, &tx) .await?; parent = Some(parent_channel); } let channel = channel::ActiveModel { id: ActiveValue::NotSet, name: ActiveValue::Set(name.to_string()), visibility: ActiveValue::Set(ChannelVisibility::Members), parent_path: ActiveValue::Set( parent .as_ref() .map_or(String::new(), |parent| parent.path()), ), requires_zed_cla: ActiveValue::NotSet, } .insert(&*tx) .await?; if parent.is_none() { membership = Some( channel_member::ActiveModel { id: ActiveValue::NotSet, channel_id: ActiveValue::Set(channel.id), user_id: ActiveValue::Set(admin_id), accepted: ActiveValue::Set(true), role: ActiveValue::Set(ChannelRole::Admin), } .insert(&*tx) .await?, ); } Ok((channel, membership)) }) .await } /// Adds a user to the specified channel. pub async fn join_channel( &self, channel_id: ChannelId, user_id: UserId, connection: ConnectionId, ) -> Result<(JoinRoom, Option<MembershipUpdated>, ChannelRole)> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; let mut role = self.channel_role_for_user(&channel, user_id, &tx).await?; let mut accept_invite_result = None; if role.is_none() { if let Some(invitation) = self .pending_invite_for_channel(&channel, user_id, &tx) .await? { // note, this may be a parent channel role = Some(invitation.role); channel_member::Entity::update(channel_member::ActiveModel { accepted: ActiveValue::Set(true), ..invitation.into_active_model() }) .exec(&*tx) .await?; accept_invite_result = Some( self.calculate_membership_updated(&channel, user_id, &tx) .await?, ); debug_assert!( self.channel_role_for_user(&channel, user_id, &tx).await? == role ); } else if channel.visibility == ChannelVisibility::Public { role = Some(ChannelRole::Guest); channel_member::Entity::insert(channel_member::ActiveModel { id: ActiveValue::NotSet, channel_id: ActiveValue::Set(channel.root_id()), user_id: ActiveValue::Set(user_id), accepted: ActiveValue::Set(true), role: ActiveValue::Set(ChannelRole::Guest), }) .exec(&*tx) .await?; accept_invite_result = Some( self.calculate_membership_updated(&channel, user_id, &tx) .await?, ); debug_assert!( self.channel_role_for_user(&channel, user_id, &tx).await? == role ); } } if role.is_none() || role == Some(ChannelRole::Banned) { Err(ErrorCode::Forbidden.anyhow())? } let role = role.unwrap(); let livekit_room = format!("channel-{}", nanoid::nanoid!(30)); let room_id = self .get_or_create_channel_room(channel_id, &livekit_room, &tx) .await?; self.join_channel_room_internal(room_id, user_id, connection, role, &tx) .await .map(|jr| (jr, accept_invite_result, role)) }) .await } /// Sets the visibility of the given channel. pub async fn set_channel_visibility( &self, channel_id: ChannelId, visibility: ChannelVisibility, admin_id: UserId, ) -> Result<channel::Model> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, admin_id, &tx) .await?; if visibility == ChannelVisibility::Public { if let Some(parent_id) = channel.parent_id() { let parent = self.get_channel_internal(parent_id, &tx).await?; if parent.visibility != ChannelVisibility::Public { Err(ErrorCode::BadPublicNesting .with_tag("direction", "parent") .anyhow())?; } } } else if visibility == ChannelVisibility::Members && self .get_channel_descendants_excluding_self([&channel], &tx) .await? .into_iter() .any(|channel| channel.visibility == ChannelVisibility::Public) { Err(ErrorCode::BadPublicNesting .with_tag("direction", "children") .anyhow())?; } let mut model = channel.into_active_model(); model.visibility = ActiveValue::Set(visibility); let channel = model.update(&*tx).await?; Ok(channel) }) .await } #[cfg(test)] pub async fn set_channel_requires_zed_cla( &self, channel_id: ChannelId, requires_zed_cla: bool, ) -> Result<()> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; let mut model = channel.into_active_model(); model.requires_zed_cla = ActiveValue::Set(requires_zed_cla); model.update(&*tx).await?; Ok(()) }) .await } /// Deletes the channel with the specified ID. pub async fn delete_channel( &self, channel_id: ChannelId, user_id: UserId, ) -> Result<(ChannelId, Vec<ChannelId>)> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, user_id, &tx) .await?; let channels_to_remove = self .get_channel_descendants_excluding_self([&channel], &tx) .await? .into_iter() .map(|channel| channel.id) .chain(Some(channel_id)) .collect::<Vec<_>>(); channel::Entity::delete_many() .filter(channel::Column::Id.is_in(channels_to_remove.iter().copied())) .exec(&*tx) .await?; Ok((channel.root_id(), channels_to_remove)) }) .await } /// Invites a user to a channel as a member. pub async fn invite_channel_member( &self, channel_id: ChannelId, invitee_id: UserId, inviter_id: UserId, role: ChannelRole, ) -> Result<InviteMemberResult> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, inviter_id, &tx) .await?; if !channel.is_root() { Err(ErrorCode::NotARootChannel.anyhow())? } channel_member::ActiveModel { id: ActiveValue::NotSet, channel_id: ActiveValue::Set(channel_id), user_id: ActiveValue::Set(invitee_id), accepted: ActiveValue::Set(false), role: ActiveValue::Set(role), } .insert(&*tx) .await?; let channel = Channel::from_model(channel); let notifications = self .create_notification( invitee_id, rpc::Notification::ChannelInvitation { channel_id: channel_id.to_proto(), channel_name: channel.name.clone(), inviter_id: inviter_id.to_proto(), }, true, &tx, ) .await? .into_iter() .collect(); Ok(InviteMemberResult { channel, notifications, }) }) .await } fn sanitize_channel_name(name: &str) -> Result<&str> { let new_name = name.trim().trim_start_matches('#'); if new_name.is_empty() { Err(anyhow!("channel name can't be blank"))?; } Ok(new_name) } /// Renames the specified channel. pub async fn rename_channel( &self, channel_id: ChannelId, admin_id: UserId, new_name: &str, ) -> Result<channel::Model> { self.transaction(move |tx| async move { let new_name = Self::sanitize_channel_name(new_name)?.to_string(); let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, admin_id, &tx) .await?; let mut model = channel.into_active_model(); model.name = ActiveValue::Set(new_name.clone()); let channel = model.update(&*tx).await?; Ok(channel) }) .await } /// accept or decline an invite to join a channel pub async fn respond_to_channel_invite( &self, channel_id: ChannelId, user_id: UserId, accept: bool, ) -> Result<RespondToChannelInvite> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; let membership_update = if accept { let rows_affected = channel_member::Entity::update_many() .set(channel_member::ActiveModel { accepted: ActiveValue::Set(accept), ..Default::default() }) .filter( channel_member::Column::ChannelId .eq(channel_id) .and(channel_member::Column::UserId.eq(user_id)) .and(channel_member::Column::Accepted.eq(false)), ) .exec(&*tx) .await? .rows_affected; if rows_affected == 0 { Err(anyhow!("no such invitation"))?; } Some( self.calculate_membership_updated(&channel, user_id, &tx) .await?, ) } else { let rows_affected = channel_member::Entity::delete_many() .filter( channel_member::Column::ChannelId .eq(channel_id) .and(channel_member::Column::UserId.eq(user_id)) .and(channel_member::Column::Accepted.eq(false)), ) .exec(&*tx) .await? .rows_affected; if rows_affected == 0 { Err(anyhow!("no such invitation"))?; } None }; Ok(RespondToChannelInvite { membership_update, notifications: self .mark_notification_as_read_with_response( user_id, &rpc::Notification::ChannelInvitation { channel_id: channel_id.to_proto(), channel_name: Default::default(), inviter_id: Default::default(), }, accept, &tx, ) .await? .into_iter() .collect(), }) }) .await } async fn calculate_membership_updated( &self, channel: &channel::Model, user_id: UserId, tx: &DatabaseTransaction, ) -> Result<MembershipUpdated> { let new_channels = self .get_user_channels(user_id, Some(channel), false, tx) .await?; let removed_channels = self .get_channel_descendants_excluding_self([channel], tx) .await? .into_iter() .map(|channel| channel.id) .chain([channel.id]) .filter(|channel_id| !new_channels.channels.iter().any(|c| c.id == *channel_id)) .collect::<Vec<_>>(); Ok(MembershipUpdated { channel_id: channel.id, new_channels, removed_channels, }) } /// Removes a channel member. pub async fn remove_channel_member( &self, channel_id: ChannelId, member_id: UserId, admin_id: UserId, ) -> Result<RemoveChannelMemberResult> { self.transaction(|tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; if member_id != admin_id { self.check_user_is_channel_admin(&channel, admin_id, &tx) .await?; } let result = channel_member::Entity::delete_many() .filter( channel_member::Column::ChannelId .eq(channel_id) .and(channel_member::Column::UserId.eq(member_id)), ) .exec(&*tx) .await?; if result.rows_affected == 0 { Err(anyhow!("no such member"))?; } Ok(RemoveChannelMemberResult { membership_update: self .calculate_membership_updated(&channel, member_id, &tx) .await?, notification_id: self .remove_notification( member_id, rpc::Notification::ChannelInvitation { channel_id: channel_id.to_proto(), channel_name: Default::default(), inviter_id: Default::default(), }, &tx, ) .await?, }) }) .await } /// Returns all channels for the user with the given ID. pub async fn get_channels_for_user(&self, user_id: UserId) -> Result<ChannelsForUser> { self.transaction(|tx| async move { self.get_user_channels(user_id, None, true, &tx).await }) .await } /// Returns all channels for the user with the given ID that are descendants /// of the specified ancestor channel. pub async fn get_user_channels( &self, user_id: UserId, ancestor_channel: Option<&channel::Model>, include_invites: bool, tx: &DatabaseTransaction, ) -> Result<ChannelsForUser> { let mut filter = channel_member::Column::UserId.eq(user_id);Lines 1–500 of 1001. Load more
use super::*; use rpc::{ proto::{channel_member::Kind, ChannelBufferVersion, VectorClockEntry}, ErrorCode, ErrorCodeExt, }; use sea_orm::{DbBackend, TryGetableMany}; impl Database { #[cfg(test)] pub async fn all_channels(&self) -> Result<Vec<(ChannelId, String)>> { self.transaction(move |tx| async move { let mut channels = Vec::new(); let mut rows = channel::Entity::find().stream(&*tx).await?; while let Some(row) = rows.next().await { let row = row?; channels.push((row.id, row.name)); } Ok(channels) }) .await } #[cfg(test)] pub async fn create_root_channel(&self, name: &str, creator_id: UserId) -> Result<ChannelId> { Ok(self.create_channel(name, None, creator_id).await?.0.id) } #[cfg(test)] pub async fn create_sub_channel( &self, name: &str, parent: ChannelId, creator_id: UserId, ) -> Result<ChannelId> { Ok(self .create_channel(name, Some(parent), creator_id) .await? .0 .id) } /// Creates a new channel. pub async fn create_channel( &self, name: &str, parent_channel_id: Option<ChannelId>, admin_id: UserId, ) -> Result<(channel::Model, Option<channel_member::Model>)> { let name = Self::sanitize_channel_name(name)?; self.transaction(move |tx| async move { let mut parent = None; let mut membership = None; if let Some(parent_channel_id) = parent_channel_id { let parent_channel = self.get_channel_internal(parent_channel_id, &tx).await?; self.check_user_is_channel_admin(&parent_channel, admin_id, &tx) .await?; parent = Some(parent_channel); } let channel = channel::ActiveModel { id: ActiveValue::NotSet, name: ActiveValue::Set(name.to_string()), visibility: ActiveValue::Set(ChannelVisibility::Members), parent_path: ActiveValue::Set( parent .as_ref() .map_or(String::new(), |parent| parent.path()), ), requires_zed_cla: ActiveValue::NotSet, } .insert(&*tx) .await?; if parent.is_none() { membership = Some( channel_member::ActiveModel { id: ActiveValue::NotSet, channel_id: ActiveValue::Set(channel.id), user_id: ActiveValue::Set(admin_id), accepted: ActiveValue::Set(true), role: ActiveValue::Set(ChannelRole::Admin), } .insert(&*tx) .await?, ); } Ok((channel, membership)) }) .await } /// Adds a user to the specified channel. pub async fn join_channel( &self, channel_id: ChannelId, user_id: UserId, connection: ConnectionId, ) -> Result<(JoinRoom, Option<MembershipUpdated>, ChannelRole)> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; let mut role = self.channel_role_for_user(&channel, user_id, &tx).await?; let mut accept_invite_result = None; if role.is_none() { if let Some(invitation) = self .pending_invite_for_channel(&channel, user_id, &tx) .await? { // note, this may be a parent channel role = Some(invitation.role); channel_member::Entity::update(channel_member::ActiveModel { accepted: ActiveValue::Set(true), ..invitation.into_active_model() }) .exec(&*tx) .await?; accept_invite_result = Some( self.calculate_membership_updated(&channel, user_id, &tx) .await?, ); debug_assert!( self.channel_role_for_user(&channel, user_id, &tx).await? == role ); } else if channel.visibility == ChannelVisibility::Public { role = Some(ChannelRole::Guest); channel_member::Entity::insert(channel_member::ActiveModel { id: ActiveValue::NotSet, channel_id: ActiveValue::Set(channel.root_id()), user_id: ActiveValue::Set(user_id), accepted: ActiveValue::Set(true), role: ActiveValue::Set(ChannelRole::Guest), }) .exec(&*tx) .await?; accept_invite_result = Some( self.calculate_membership_updated(&channel, user_id, &tx) .await?, ); debug_assert!( self.channel_role_for_user(&channel, user_id, &tx).await? == role ); } } if role.is_none() || role == Some(ChannelRole::Banned) { Err(ErrorCode::Forbidden.anyhow())? } let role = role.unwrap(); let livekit_room = format!("channel-{}", nanoid::nanoid!(30)); let room_id = self .get_or_create_channel_room(channel_id, &livekit_room, &tx) .await?; self.join_channel_room_internal(room_id, user_id, connection, role, &tx) .await .map(|jr| (jr, accept_invite_result, role)) }) .await } /// Sets the visibility of the given channel. pub async fn set_channel_visibility( &self, channel_id: ChannelId, visibility: ChannelVisibility, admin_id: UserId, ) -> Result<channel::Model> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, admin_id, &tx) .await?; if visibility == ChannelVisibility::Public { if let Some(parent_id) = channel.parent_id() { let parent = self.get_channel_internal(parent_id, &tx).await?; if parent.visibility != ChannelVisibility::Public { Err(ErrorCode::BadPublicNesting .with_tag("direction", "parent") .anyhow())?; } } } else if visibility == ChannelVisibility::Members && self .get_channel_descendants_excluding_self([&channel], &tx) .await? .into_iter() .any(|channel| channel.visibility == ChannelVisibility::Public) { Err(ErrorCode::BadPublicNesting .with_tag("direction", "children") .anyhow())?; } let mut model = channel.into_active_model(); model.visibility = ActiveValue::Set(visibility); let channel = model.update(&*tx).await?; Ok(channel) }) .await } #[cfg(test)] pub async fn set_channel_requires_zed_cla( &self, channel_id: ChannelId, requires_zed_cla: bool, ) -> Result<()> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; let mut model = channel.into_active_model(); model.requires_zed_cla = ActiveValue::Set(requires_zed_cla); model.update(&*tx).await?; Ok(()) }) .await } /// Deletes the channel with the specified ID. pub async fn delete_channel( &self, channel_id: ChannelId, user_id: UserId, ) -> Result<(ChannelId, Vec<ChannelId>)> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, user_id, &tx) .await?; let channels_to_remove = self .get_channel_descendants_excluding_self([&channel], &tx) .await? .into_iter() .map(|channel| channel.id) .chain(Some(channel_id)) .collect::<Vec<_>>(); channel::Entity::delete_many() .filter(channel::Column::Id.is_in(channels_to_remove.iter().copied())) .exec(&*tx) .await?; Ok((channel.root_id(), channels_to_remove)) }) .await } /// Invites a user to a channel as a member. pub async fn invite_channel_member( &self, channel_id: ChannelId, invitee_id: UserId, inviter_id: UserId, role: ChannelRole, ) -> Result<InviteMemberResult> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, inviter_id, &tx) .await?; if !channel.is_root() { Err(ErrorCode::NotARootChannel.anyhow())? } channel_member::ActiveModel { id: ActiveValue::NotSet, channel_id: ActiveValue::Set(channel_id), user_id: ActiveValue::Set(invitee_id), accepted: ActiveValue::Set(false), role: ActiveValue::Set(role), } .insert(&*tx) .await?; let channel = Channel::from_model(channel); let notifications = self .create_notification( invitee_id, rpc::Notification::ChannelInvitation { channel_id: channel_id.to_proto(), channel_name: channel.name.clone(), inviter_id: inviter_id.to_proto(), }, true, &tx, ) .await? .into_iter() .collect(); Ok(InviteMemberResult { channel, notifications, }) }) .await } fn sanitize_channel_name(name: &str) -> Result<&str> { let new_name = name.trim().trim_start_matches('#'); if new_name.is_empty() { Err(anyhow!("channel name can't be blank"))?; } Ok(new_name) } /// Renames the specified channel. pub async fn rename_channel( &self, channel_id: ChannelId, admin_id: UserId, new_name: &str, ) -> Result<channel::Model> { self.transaction(move |tx| async move { let new_name = Self::sanitize_channel_name(new_name)?.to_string(); let channel = self.get_channel_internal(channel_id, &tx).await?; self.check_user_is_channel_admin(&channel, admin_id, &tx) .await?; let mut model = channel.into_active_model(); model.name = ActiveValue::Set(new_name.clone()); let channel = model.update(&*tx).await?; Ok(channel) }) .await } /// accept or decline an invite to join a channel pub async fn respond_to_channel_invite( &self, channel_id: ChannelId, user_id: UserId, accept: bool, ) -> Result<RespondToChannelInvite> { self.transaction(move |tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; let membership_update = if accept { let rows_affected = channel_member::Entity::update_many() .set(channel_member::ActiveModel { accepted: ActiveValue::Set(accept), ..Default::default() }) .filter( channel_member::Column::ChannelId .eq(channel_id) .and(channel_member::Column::UserId.eq(user_id)) .and(channel_member::Column::Accepted.eq(false)), ) .exec(&*tx) .await? .rows_affected; if rows_affected == 0 { Err(anyhow!("no such invitation"))?; } Some( self.calculate_membership_updated(&channel, user_id, &tx) .await?, ) } else { let rows_affected = channel_member::Entity::delete_many() .filter( channel_member::Column::ChannelId .eq(channel_id) .and(channel_member::Column::UserId.eq(user_id)) .and(channel_member::Column::Accepted.eq(false)), ) .exec(&*tx) .await? .rows_affected; if rows_affected == 0 { Err(anyhow!("no such invitation"))?; } None }; Ok(RespondToChannelInvite { membership_update, notifications: self .mark_notification_as_read_with_response( user_id, &rpc::Notification::ChannelInvitation { channel_id: channel_id.to_proto(), channel_name: Default::default(), inviter_id: Default::default(), }, accept, &tx, ) .await? .into_iter() .collect(), }) }) .await } async fn calculate_membership_updated( &self, channel: &channel::Model, user_id: UserId, tx: &DatabaseTransaction, ) -> Result<MembershipUpdated> { let new_channels = self .get_user_channels(user_id, Some(channel), false, tx) .await?; let removed_channels = self .get_channel_descendants_excluding_self([channel], tx) .await? .into_iter() .map(|channel| channel.id) .chain([channel.id]) .filter(|channel_id| !new_channels.channels.iter().any(|c| c.id == *channel_id)) .collect::<Vec<_>>(); Ok(MembershipUpdated { channel_id: channel.id, new_channels, removed_channels, }) } /// Removes a channel member. pub async fn remove_channel_member( &self, channel_id: ChannelId, member_id: UserId, admin_id: UserId, ) -> Result<RemoveChannelMemberResult> { self.transaction(|tx| async move { let channel = self.get_channel_internal(channel_id, &tx).await?; if member_id != admin_id { self.check_user_is_channel_admin(&channel, admin_id, &tx) .await?; } let result = channel_member::Entity::delete_many() .filter( channel_member::Column::ChannelId .eq(channel_id) .and(channel_member::Column::UserId.eq(member_id)), ) .exec(&*tx) .await?; if result.rows_affected == 0 { Err(anyhow!("no such member"))?; } Ok(RemoveChannelMemberResult { membership_update: self .calculate_membership_updated(&channel, member_id, &tx) .await?, notification_id: self .remove_notification( member_id, rpc::Notification::ChannelInvitation { channel_id: channel_id.to_proto(), channel_name: Default::default(), inviter_id: Default::default(), }, &tx, ) .await?, }) }) .await } /// Returns all channels for the user with the given ID. pub async fn get_channels_for_user(&self, user_id: UserId) -> Result<ChannelsForUser> { self.transaction(|tx| async move { self.get_user_channels(user_id, None, true, &tx).await }) .await } /// Returns all channels for the user with the given ID that are descendants /// of the specified ancestor channel. pub async fn get_user_channels( &self, user_id: UserId, ancestor_channel: Option<&channel::Model>, include_invites: bool, tx: &DatabaseTransaction, ) -> Result<ChannelsForUser> { let mut filter = channel_member::Column::UserId.eq(user_id);