ZIm/crates/collab/src/db/queries/channels.rs

use super::*;
use anyhow::Context as _;
use rpc::{
    ErrorCode, ErrorCodeExt,
    proto::{ChannelBufferVersion, VectorClockEntry, channel_member::Kind},
};
use sea_orm::{ActiveValue, 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 parent_path = parent
                .as_ref()
                .map_or(String::new(), |parent| parent.path());

            // Find the maximum channel_order among siblings to set the new channel at the end
            let max_order = if parent_path.is_empty() {
                0
            } else {
                max_order(&parent_path, &tx).await?
            };

            log::info!(
                "Creating channel '{}' with parent_path='{}', max_order={}, new_order={}",
                name,
                parent_path,
                max_order,
                max_order + 1
            );

            let channel = channel::ActiveModel {
                id: ActiveValue::NotSet,
                name: ActiveValue::Set(name.to_string()),
                visibility: ActiveValue::Set(ChannelVisibility::Members),
                parent_path: ActiveValue::Set(parent_path),
                requires_zed_cla: ActiveValue::NotSet,
                channel_order: ActiveValue::Set(max_order + 1),
            }
            .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);
        if !include_invites {
            filter = filter.and(channel_member::Column::Accepted.eq(true))
        }
        if let Some(ancestor) = ancestor_channel {
            filter = filter.and(channel_member::Column::ChannelId.eq(ancestor.root_id()));
        }

        let mut channels = Vec::<channel::Model>::new();
        let mut invited_channels = Vec::<Channel>::new();
        let mut channel_memberships = Vec::<channel_member::Model>::new();
        let mut rows = channel_member::Entity::find()
            .filter(filter)
            .inner_join(channel::Entity)
            .select_also(channel::Entity)
            .stream(tx)
            .await?;
        while let Some(row) = rows.next().await {
            if let (membership, Some(channel)) = row? {
                if membership.accepted {
                    channel_memberships.push(membership);
                    channels.push(channel);
                } else {
                    invited_channels.push(Channel::from_model(channel));
                }
            }
        }
        drop(rows);

        let mut descendants = self
            .get_channel_descendants_excluding_self(channels.iter(), tx)
            .await?;

        descendants.extend(channels);

        let roles_by_channel_id = channel_memberships
            .iter()
            .map(|membership| (membership.channel_id, membership.role))
            .collect::<HashMap<_, _>>();

        let channels: Vec<Channel> = descendants
            .into_iter()
            .filter_map(|channel| {
                let parent_role = roles_by_channel_id.get(&channel.root_id())?;
                if parent_role.can_see_channel(channel.visibility) {
                    Some(Channel::from_model(channel))
                } else {
                    None
                }
            })
            .collect();

        #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
        enum QueryUserIdsAndChannelIds {
            ChannelId,
            UserId,
        }

        let mut channel_participants: HashMap<ChannelId, Vec<UserId>> = HashMap::default();
        {
            let mut rows = room_participant::Entity::find()
                .inner_join(room::Entity)
                .filter(room::Column::ChannelId.is_in(channels.iter().map(|c| c.id)))
                .select_only()
                .column(room::Column::ChannelId)
                .column(room_participant::Column::UserId)
                .into_values::<_, QueryUserIdsAndChannelIds>()
                .stream(tx)
                .await?;
            while let Some(row) = rows.next().await {
                let row: (ChannelId, UserId) = row?;
                channel_participants.entry(row.0).or_default().push(row.1)
            }
        }

        let channel_ids = channels.iter().map(|c| c.id).collect::<Vec<_>>();

        let mut channel_ids_by_buffer_id = HashMap::default();
        let mut latest_buffer_versions: Vec<ChannelBufferVersion> = vec![];
        let mut rows = buffer::Entity::find()
            .filter(buffer::Column::ChannelId.is_in(channel_ids.iter().copied()))
            .stream(tx)
            .await?;
        while let Some(row) = rows.next().await {
            let row = row?;
            channel_ids_by_buffer_id.insert(row.id, row.channel_id);
            latest_buffer_versions.push(ChannelBufferVersion {
                channel_id: row.channel_id.0 as u64,
                epoch: row.latest_operation_epoch.unwrap_or_default() as u64,
                version: if let Some((latest_lamport_timestamp, latest_replica_id)) = row
                    .latest_operation_lamport_timestamp
                    .zip(row.latest_operation_replica_id)
                {
                    vec![VectorClockEntry {
                        timestamp: latest_lamport_timestamp as u32,
                        replica_id: latest_replica_id as u32,
                    }]
                } else {
                    vec![]
                },
            });
        }
        drop(rows);

        let latest_channel_messages = self.latest_channel_messages(&channel_ids, tx).await?;

        let observed_buffer_versions = self
            .observed_channel_buffer_changes(&channel_ids_by_buffer_id, user_id, tx)
            .await?;

        let observed_channel_messages = self
            .observed_channel_messages(&channel_ids, user_id, tx)
            .await?;

        Ok(ChannelsForUser {
            channel_memberships,
            channels,
            invited_channels,
            channel_participants,
            latest_buffer_versions,
            latest_channel_messages,
            observed_buffer_versions,
            observed_channel_messages,
        })
    }

    /// Sets the role for the specified channel member.
    pub async fn set_channel_member_role(
        &self,
        channel_id: ChannelId,
        admin_id: UserId,
        for_user: UserId,
        role: ChannelRole,
    ) -> Result<SetMemberRoleResult> {
        self.transaction(|tx| async move {
            let channel = self.get_channel_internal(channel_id, &tx).await?;
            self.check_user_is_channel_admin(&channel, admin_id, &tx)
                .await?;

            let membership = channel_member::Entity::find()
                .filter(
                    channel_member::Column::ChannelId
                        .eq(channel_id)
                        .and(channel_member::Column::UserId.eq(for_user)),
                )
                .one(&*tx)
                .await?
                .context("no such member")?;

            let mut update = membership.into_active_model();
            update.role = ActiveValue::Set(role);
            let updated = channel_member::Entity::update(update).exec(&*tx).await?;

            if updated.accepted {
                Ok(SetMemberRoleResult::MembershipUpdated(
                    self.calculate_membership_updated(&channel, for_user, &tx)
                        .await?,
                ))
            } else {
                Ok(SetMemberRoleResult::InviteUpdated(Channel::from_model(
                    channel,
                )))
            }
        })
        .await
    }

    /// Returns the details for the specified channel member.
    pub async fn get_channel_participant_details(
        &self,
        channel_id: ChannelId,
        filter: &str,
        limit: u64,
        user_id: UserId,
    ) -> Result<(Vec<proto::ChannelMember>, Vec<proto::User>)> {
        let members = self
            .transaction(move |tx| async move {
                let channel = self.get_channel_internal(channel_id, &tx).await?;
                self.check_user_is_channel_participant(&channel, user_id, &tx)
                    .await?;
                let mut query = channel_member::Entity::find()
                    .find_also_related(user::Entity)
                    .filter(channel_member::Column::ChannelId.eq(channel.root_id()));

                if cfg!(any(test, feature = "sqlite")) && self.pool.get_database_backend() == DbBackend::Sqlite {
                    query = query.filter(Expr::cust_with_values(
                        "UPPER(github_login) LIKE ?",
                        [Self::fuzzy_like_string(&filter.to_uppercase())],
                    ))
                } else {
                    query = query.filter(Expr::cust_with_values(
                        "github_login ILIKE $1",
                        [Self::fuzzy_like_string(filter)],
                    ))
                }
                let members = query.order_by(
                        Expr::cust(
                            "not role = 'admin', not role = 'member', not role = 'guest', not accepted, github_login",
                        ),
                        sea_orm::Order::Asc,
                    )
                    .limit(limit)
                    .all(&*tx)
                    .await?;

                Ok(members)
            })
            .await?;

        let mut users: Vec<proto::User> = Vec::with_capacity(members.len());

        let members = members
            .into_iter()
            .map(|(member, user)| {
                if let Some(user) = user {
                    users.push(proto::User {
                        id: user.id.to_proto(),
                        avatar_url: format!(
                            "https://github.com/{}.png?size=128",
                            user.github_login
                        ),
                        github_login: user.github_login,
                        name: user.name,
                        email: user.email_address,
                    })
                }
                proto::ChannelMember {
                    role: member.role.into(),
                    user_id: member.user_id.to_proto(),
                    kind: if member.accepted {
                        Kind::Member
                    } else {
                        Kind::Invitee
                    }
                    .into(),
                }
            })
            .collect();

        Ok((members, users))
    }

    /// Returns whether the given user is an admin in the specified channel.
    pub async fn check_user_is_channel_admin(
        &self,
        channel: &channel::Model,
        user_id: UserId,
        tx: &DatabaseTransaction,
    ) -> Result<ChannelRole> {
        let role = self.channel_role_for_user(channel, user_id, tx).await?;
        match role {
            Some(ChannelRole::Admin) => Ok(role.unwrap()),
            Some(ChannelRole::Member)
            | Some(ChannelRole::Talker)
            | Some(ChannelRole::Banned)
            | Some(ChannelRole::Guest)
            | None => Err(anyhow!(
                "user is not a channel admin or channel does not exist"
            ))?,
        }
    }

    /// Returns whether the given user is a member of the specified channel.
    pub async fn check_user_is_channel_member(
        &self,
        channel: &channel::Model,
        user_id: UserId,
        tx: &DatabaseTransaction,
    ) -> Result<ChannelRole> {
        let channel_role = self.channel_role_for_user(channel, user_id, tx).await?;
        match channel_role {
            Some(ChannelRole::Admin) | Some(ChannelRole::Member) => Ok(channel_role.unwrap()),
            Some(ChannelRole::Banned)
            | Some(ChannelRole::Guest)
            | Some(ChannelRole::Talker)
            | None => Err(anyhow!(
                "user is not a channel member or channel does not exist"
            ))?,
        }
    }

    /// Returns whether the given user is a participant in the specified channel.
    pub async fn check_user_is_channel_participant(
        &self,
        channel: &channel::Model,
        user_id: UserId,
        tx: &DatabaseTransaction,
    ) -> Result<ChannelRole> {
        let role = self.channel_role_for_user(channel, user_id, tx).await?;
        match role {
            Some(ChannelRole::Admin)
            | Some(ChannelRole::Member)
            | Some(ChannelRole::Guest)
            | Some(ChannelRole::Talker) => Ok(role.unwrap()),
            Some(ChannelRole::Banned) | None => Err(anyhow!(
                "user is not a channel participant or channel does not exist"
            ))?,
        }
    }

    /// Returns a user's pending invite for the given channel, if one exists.
    pub async fn pending_invite_for_channel(
        &self,
        channel: &channel::Model,
        user_id: UserId,
        tx: &DatabaseTransaction,
    ) -> Result<Option<channel_member::Model>> {
        let row = channel_member::Entity::find()
            .filter(channel_member::Column::ChannelId.eq(channel.root_id()))
            .filter(channel_member::Column::UserId.eq(user_id))
            .filter(channel_member::Column::Accepted.eq(false))
            .one(tx)
            .await?;

        Ok(row)
    }

    /// Returns the role for a user in the given channel.
    pub async fn channel_role_for_user(
        &self,
        channel: &channel::Model,
        user_id: UserId,
        tx: &DatabaseTransaction,
    ) -> Result<Option<ChannelRole>> {
        let membership = channel_member::Entity::find()
            .filter(
                channel_member::Column::ChannelId
                    .eq(channel.root_id())
                    .and(channel_member::Column::UserId.eq(user_id))
                    .and(channel_member::Column::Accepted.eq(true)),
            )
            .one(tx)
            .await?;

        let Some(membership) = membership else {
            return Ok(None);
        };

        if !membership.role.can_see_channel(channel.visibility) {
            return Ok(None);
        }

        Ok(Some(membership.role))
    }

    // Get the descendants of the given set if channels, ordered by their
    // path.
    pub(crate) async fn get_channel_descendants_excluding_self(
        &self,
        channels: impl IntoIterator<Item = &channel::Model>,
        tx: &DatabaseTransaction,
    ) -> Result<Vec<channel::Model>> {
        let mut filter = Condition::any();
        for channel in channels.into_iter() {
            filter = filter.add(channel::Column::ParentPath.like(channel.descendant_path_filter()));
        }

        if filter.is_empty() {
            return Ok(vec![]);
        }

        Ok(channel::Entity::find()
            .filter(filter)
            .order_by_asc(Expr::cust("parent_path || id || '/'"))
            .all(tx)
            .await?)
    }

    /// Returns the channel with the given ID.
    pub async fn get_channel(&self, channel_id: ChannelId, user_id: UserId) -> Result<Channel> {
        self.transaction(|tx| async move {
            let channel = self.get_channel_internal(channel_id, &tx).await?;
            self.check_user_is_channel_participant(&channel, user_id, &tx)
                .await?;

            Ok(Channel::from_model(channel))
        })
        .await
    }

    pub(crate) async fn get_channel_internal(
        &self,
        channel_id: ChannelId,
        tx: &DatabaseTransaction,
    ) -> Result<channel::Model> {
        Ok(channel::Entity::find_by_id(channel_id)
            .one(tx)
            .await?
            .ok_or_else(|| proto::ErrorCode::NoSuchChannel.anyhow())?)
    }

    pub(crate) async fn get_or_create_channel_room(
        &self,
        channel_id: ChannelId,
        livekit_room: &str,
        tx: &DatabaseTransaction,
    ) -> Result<RoomId> {
        let room = room::Entity::find()
            .filter(room::Column::ChannelId.eq(channel_id))
            .one(tx)
            .await?;

        let room_id = if let Some(room) = room {
            room.id
        } else {
            let result = room::Entity::insert(room::ActiveModel {
                channel_id: ActiveValue::Set(Some(channel_id)),
                live_kit_room: ActiveValue::Set(livekit_room.to_string()),
                ..Default::default()
            })
            .exec(tx)
            .await?;

            result.last_insert_id
        };

        Ok(room_id)
    }

    /// Move a channel from one parent to another
    pub async fn move_channel(
        &self,
        channel_id: ChannelId,
        new_parent_id: ChannelId,
        admin_id: UserId,
    ) -> Result<(ChannelId, Vec<Channel>)> {
        self.transaction(|tx| async move {
            let channel = self.get_channel_internal(channel_id, &tx).await?;
            self.check_user_is_channel_admin(&channel, admin_id, &tx)
                .await?;
            let new_parent = self.get_channel_internal(new_parent_id, &tx).await?;

            if new_parent.root_id() != channel.root_id() {
                Err(anyhow!(ErrorCode::WrongMoveTarget))?;
            }

            if new_parent
                .ancestors_including_self()
                .any(|id| id == channel.id)
            {
                Err(anyhow!(ErrorCode::CircularNesting))?;
            }

            if channel.visibility == ChannelVisibility::Public
                && new_parent.visibility != ChannelVisibility::Public
            {
                Err(anyhow!(ErrorCode::BadPublicNesting))?;
            }

            let root_id = channel.root_id();
            let new_parent_path = new_parent.path();
            let old_path = format!("{}{}/", channel.parent_path, channel.id);
            let new_path = format!("{}{}/", &new_parent_path, channel.id);
            let new_order = max_order(&new_parent_path, &tx).await? + 1;

            let mut model = channel.into_active_model();
            model.parent_path = ActiveValue::Set(new_parent.path());
            model.channel_order = ActiveValue::Set(new_order);
            let channel = model.update(&*tx).await?;

            let descendent_ids =
                ChannelId::find_by_statement::<QueryIds>(Statement::from_sql_and_values(
                    self.pool.get_database_backend(),
                    "
                    UPDATE channels SET parent_path = REPLACE(parent_path, $1, $2)
                    WHERE parent_path LIKE $3 || '%'
                    RETURNING id
                ",
                    [old_path.clone().into(), new_path.into(), old_path.into()],
                ))
                .all(&*tx)
                .await?;

            let all_moved_ids = Some(channel.id).into_iter().chain(descendent_ids);

            let channels = channel::Entity::find()
                .filter(channel::Column::Id.is_in(all_moved_ids))
                .all(&*tx)
                .await?
                .into_iter()
                .map(Channel::from_model)
                .collect::<Vec<_>>();

            Ok((root_id, channels))
        })
        .await
    }

    pub async fn reorder_channel(
        &self,
        channel_id: ChannelId,
        direction: proto::reorder_channel::Direction,
        user_id: UserId,
    ) -> Result<Vec<Channel>> {
        self.transaction(|tx| async move {
            let mut channel = self.get_channel_internal(channel_id, &tx).await?;

            if channel.is_root() {
                log::info!("Skipping reorder of root channel {}", channel.id,);
                return Ok(vec![]);
            }

            log::info!(
                "Reordering channel {} (parent_path: '{}', order: {})",
                channel.id,
                channel.parent_path,
                channel.channel_order
            );

            // Check if user is admin of the channel
            self.check_user_is_channel_admin(&channel, user_id, &tx)
                .await?;

            // Find the sibling channel to swap with
            let sibling_channel = match direction {
                proto::reorder_channel::Direction::Up => {
                    log::info!(
                        "Looking for sibling with parent_path='{}' and order < {}",
                        channel.parent_path,
                        channel.channel_order
                    );
                    // Find channel with highest order less than current
                    channel::Entity::find()
                        .filter(
                            channel::Column::ParentPath
                                .eq(&channel.parent_path)
                                .and(channel::Column::ChannelOrder.lt(channel.channel_order)),
                        )
                        .order_by_desc(channel::Column::ChannelOrder)
                        .one(&*tx)
                        .await?
                }
                proto::reorder_channel::Direction::Down => {
                    log::info!(
                        "Looking for sibling with parent_path='{}' and order > {}",
                        channel.parent_path,
                        channel.channel_order
                    );
                    // Find channel with lowest order greater than current
                    channel::Entity::find()
                        .filter(
                            channel::Column::ParentPath
                                .eq(&channel.parent_path)
                                .and(channel::Column::ChannelOrder.gt(channel.channel_order)),
                        )
                        .order_by_asc(channel::Column::ChannelOrder)
                        .one(&*tx)
                        .await?
                }
            };

            let mut sibling_channel = match sibling_channel {
                Some(sibling) => {
                    log::info!(
                        "Found sibling {} (parent_path: '{}', order: {})",
                        sibling.id,
                        sibling.parent_path,
                        sibling.channel_order
                    );
                    sibling
                }
                None => {
                    log::warn!("No sibling found to swap with");
                    // No sibling to swap with
                    return Ok(vec![]);
                }
            };

            let current_order = channel.channel_order;
            let sibling_order = sibling_channel.channel_order;

            channel::ActiveModel {
                id: ActiveValue::Unchanged(sibling_channel.id),
                channel_order: ActiveValue::Set(current_order),
                ..Default::default()
            }
            .update(&*tx)
            .await?;
            sibling_channel.channel_order = current_order;

            channel::ActiveModel {
                id: ActiveValue::Unchanged(channel.id),
                channel_order: ActiveValue::Set(sibling_order),
                ..Default::default()
            }
            .update(&*tx)
            .await?;
            channel.channel_order = sibling_order;

            log::info!(
                "Reorder complete. Swapped channels {} and {}",
                channel.id,
                sibling_channel.id
            );

            let swapped_channels = vec![
                Channel::from_model(channel),
                Channel::from_model(sibling_channel),
            ];

            Ok(swapped_channels)
        })
        .await
    }
}

async fn max_order(parent_path: &str, tx: &TransactionHandle) -> Result<i32> {
    let max_order = channel::Entity::find()
        .filter(channel::Column::ParentPath.eq(parent_path))
        .select_only()
        .column_as(channel::Column::ChannelOrder.max(), "max_order")
        .into_tuple::<Option<i32>>()
        .one(&**tx)
        .await?
        .flatten()
        .unwrap_or(0);

    Ok(max_order)
}

#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
enum QueryIds {
    Id,
}

#[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
enum QueryUserIds {
    UserId,
}